Ordering & Replay

Imagine a deli counter that hands out numbered tickets. Everyone is served in the order they arrived — ticket 41 before 42 before 43 — and the numbers never repeat or skip. Now imagine the deli also kept a logbook of every ticket it ever served. A clerk arriving for the late shift could open the book, find where the morning clerk left off, and pick up from exactly that ticket — no customer served twice, none missed.

Those two ideas — serving in a fixed order and rewinding the log to any point — are ordering and replay. They are what separate a fleeting stream of notifications from a durable, rebuildable record of what happened.

Ordering — the idea

Ordering is the guarantee about what sequence consumers observe messages in. The strongest common form is FIFO (first-in, first-out): messages come out in exactly the order they went in. A queue gives you this naturally — like a single-file line, the message enqueued first is delivered first.

A FIFO queue delivers messages in the exact order they were enqueued.

Global FIFO across an entire channel is simple but limits throughput — only one consumer can safely process at a time without reordering. In practice most systems offer per-key ordering instead: messages that share a partition key (an order ID, a user ID) stay strictly ordered relative to each other, while unrelated keys flow in parallel. You get order where it matters and parallelism everywhere else.

Replay — the idea

A real-time channel is a PA announcement: hear it now or miss it forever. A replayable channel is a recording. To replay, the system has to do two things:

1. Number every message with a monotonically increasing sequence number (also called an offset) — a stable address for each message in the log.
2. Persist the log so messages survive after delivery, and let a consumer say "start me at offset N" instead of always "start me at the newest."

Every message is numbered and written to a persistent log; a late subscriber rewinds to any offset and re-reads history, while a live subscriber follows the tail.

Because the log keeps the sequence intact, replay and ordering reinforce each other: re-reading from offset 42 always returns 42, 43, 44… in the same order, every time. That determinism is what makes a log trustworthy as a system of record.

The event-sourcing idea

If the log is the source of truth, you do not need to store the current state of anything — you can rebuild it by replaying the events that produced it. This is event sourcing: instead of saving "account balance = $80," you save the sequence of facts (Deposited $100, Withdrew $20) and replay them to compute the balance on demand. A new service, a rebuilt cache, or a bug fix that needs to reprocess history all start the same way: replay from the beginning.

Precise definition

• Ordering — a delivery guarantee that consumers observe messages in a defined sequence. FIFO orders an entire channel; per-key ordering orders only messages sharing a partition key, allowing parallel processing across keys.
• Sequence number / offset — a monotonically increasing integer assigned to each message as it is persisted, giving every message a stable, addressable position in the log.
• Replay — re-reading messages from a persisted log starting at a chosen start position (a sequence number, a timestamp, the first message, or the last), rather than receiving only messages published after subscribing.
• Event sourcing — modeling state as the ordered log of events that produced it, and reconstructing current state by replaying that log from the start.

Trade-offs

In KubeMQ

The snippet below subscribes to a persisted channel from sequence 3 — replaying every stored event at or after that offset, then streaming new ones as they arrive. Swapping StartAtSequence for StartFromFirst replays the entire history; StartAtTimeDelta replays a recent time window.

sub, err := client.SubscribeToEventsStore(ctx, "orders.events", "",
    kubemq.StartAtSequence(3),
    kubemq.WithOnEvent(func(event *kubemq.Event) {
        fmt.Printf("[replay] seq=%d body=%s\n",
            event.Sequence, string(event.Body))
    }),
    kubemq.WithOnError(func(err error) { log.Println(err) }),
)

client.subscribe_to_events_store(
    subscription=EventsStoreSubscription(
        channel="orders.events",
        start_position=EventStoreStartPosition.StartAtSequence,
        start_position_value=3,
        on_receive_event_callback=lambda e: print(
            f"[replay] seq={e.sequence} body={e.body.decode('utf-8')}"
        ),
        on_error_callback=lambda e: print(f"Error: {e}"),
    ),
    cancel=CancellationToken(),
)

client.subscribeToEventsStore({
  channel: 'orders.events',
  startPosition: EventStoreStartPosition.StartAtSequence,
  startPositionValue: 3,
  onEvent: (msg) =>
    console.log(`[replay] seq=${msg.sequence} body=${new TextDecoder().decode(msg.body)}`),
  onError: (err) => console.error(err.message),
});

client.subscribeToEventsStore(EventsStoreSubscription.builder()
    .channel("orders.events")
    .startPosition(EventStoreStartPosition.StartAtSequence)
    .startPositionValue(3)
    .onReceiveEventCallback(event ->
        System.out.printf("[replay] seq=%d body=%s%n",
            event.getSequence(), new String(event.getBody())))
    .onErrorCallback(err -> System.err.println(err.getMessage()))
    .build());

await foreach (var msg in client.SubscribeToEventsStoreAsync(
    new EventsStoreSubscription
    {
        Channel = "orders.events",
        StartPosition = EventStoreStartPosition.StartAtSequence,
        StartPositionValue = 3,
    }))
{
    Console.WriteLine($"[replay] seq={msg.Sequence} body={Encoding.UTF8.GetString(msg.Body.Span)}");
}

client.subscribeToEventsStore {
    channel = "orders.events"
    startPosition = StartPosition.StartAtSequence
    startPositionValue = 3
}.collect { msg ->
    println("[replay] seq=${msg.sequence} body=${String(msg.body)}")
}

client->SubscribeToEventsStore("orders.events", "",
    kubemq::StartPosition::StartAtSequence, 3,
    [](const kubemq::EventStoreReceived& msg) {
        std::cout << "[replay] seq=" << msg.sequence()
                  << " body=" << msg.body() << std::endl;
    },
    [](const std::string& err) { std::cerr << err << std::endl; });

use kubemq::prelude::*;
use kubemq::EventsStoreSubscription;

let sub = client
    .subscribe_to_events_store(
        "orders.events",
        "",
        EventsStoreSubscription::StartAtSequence(3),
        |event| {
            Box::pin(async move {
                println!(
                    "[replay] seq={} body={}",
                    event.sequence,
                    String::from_utf8_lossy(&event.body)
                );
            })
        },
        None,
    )
    .await?;

sub = KubeMQ::PubSub::EventsStoreSubscription.new(
  channel: "orders.events",
  start_position: KubeMQ::PubSub::EventStoreStartPosition::START_AT_SEQUENCE,
  start_position_value: 3
)
client.subscribe_to_events_store(sub, cancellation_token: cancel, on_error: lambda { |e|
  puts "Error: #{e.message}"
}) do |event|
  puts "[replay] seq=#{event.sequence} body=#{event.body}"
end

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events_store(client, "orders.events",
    start_at: {:start_at_sequence, 3},
    on_event: fn event ->
      IO.puts("[replay] seq #{event.sequence}: #{event.body}")
    end
  )

How KubeMQ does this →