Replay from Sequence
Replay events starting from a specific sequence number
Overview
Replaying from a sequence number lets a consumer resume an events-store subscription from an exact point in a channel's history, instead of re-reading everything or only catching new traffic. It's the checkpoint-recovery pattern: a worker persists the last sequence it processed, and after a crash or redeploy it reopens the subscription right there — no gap, no reprocessing everything that came before.
Sequence numbers are broker-assigned per channel, starting at 1 and increasing monotonically with every stored event; they never reset unless the channel is purged. Setting events_store_type=EventStoreStartPosition.StartAtSequence with events_store_sequence_value=3 tells the broker to begin delivery at that sequence inclusive, replaying stored events from that point, then transitioning the subscription to live delivery for anything published afterward.
Gotchas: the sequence value is inclusive, so events_store_sequence_value=3 still delivers event 3 — off by one and you'll reprocess or silently drop a message; you must track and persist the "last processed" sequence yourself, KubeMQ doesn't checkpoint it for you; and requesting a sequence past the current head isn't an error — you'll just get nothing until new events catch up to it.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Replay from sequence — subscribe starting from a specific sequence number."""
from __future__ import annotations
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncPubSubClient,
EventStoreMessage,
EventsStoreSubscription,
)
from kubemq.pubsub import EventStoreStartPosition
async def main() -> None:
async with AsyncPubSubClient(
address="localhost:50000",
client_id="python-events-store-replay-from-sequence-client",
) as client:
# Pre-populate with some messages
for i in range(5):
result = await client.send_event_store(
EventStoreMessage(
channel="python-events-store.replay-from-sequence",
body=f"Message-{i + 1}".encode(),
)
)
print(f"Sent message {i + 1}, result: {result}")
await asyncio.sleep(1)
# Subscribe starting from sequence 3 — will replay messages 3, 4, 5
token = AsyncCancellationToken()
async def subscriber() -> None:
async for event in client.subscribe_to_events_store(
subscription=EventsStoreSubscription(
channel="python-events-store.replay-from-sequence",
on_receive_event_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
events_store_type=EventStoreStartPosition.StartAtSequence,
events_store_sequence_value=3,
),
cancellation_token=token,
):
print(f"Received: {event.body.decode('utf-8')}")
task = asyncio.create_task(subscriber())
await asyncio.sleep(3)
token.cancel()
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
How It Works
- Five messages are pre-published and persisted with sequences 1–5; the subscription then starts at sequence 3, replaying only messages 3, 4, and 5.
EventStoreStartPosition.StartAtSequencecombined withevents_store_sequence_value=3tells the broker to begin replay from that exact sequence number.- Sequence numbers are broker-assigned and monotonically increasing per channel; they are reliable anchors for exactly-once replay.
- After replaying the history, the subscription transitions to live delivery, receiving any new events published to the channel.
Related
Was this page helpful?