Replay from Time

Overview

Replaying from a timestamp lets a consumer recover a window of history without knowing exact sequence numbers — you reach for it after a deploy, an outage, or any gap where you know roughly when you went dark but not where you left off in the stream. It turns an Events Store channel into a rewindable log: resubscribe with a point in time and the broker replays every event stored at or after it, then hands off to live delivery.

The subscription's StartPosition is set to EventStoreStartPosition.StartAtTimeDelta (a relative offset in seconds via StartTimeDeltaSeconds) or StartAtTime (an absolute DateTime) — the broker compares this against the storage timestamp it assigned to each event, not any timestamp embedded in the payload. Because it's wall-clock based, the window is approximate rather than exact: pass a value far enough back to be safe.

Gotchas: clock skew between your subscriber's clock and the server's matters — favor a generous buffer over a precise cutoff. Storage timestamps reflect when the broker persisted the event, not when the producer created it, so under load the two can drift. And unlike sequence-based replay, a time-based start position has no way to guarantee "no gaps, no duplicates" across a network hiccup — use EventStoreStartPosition.StartAtSequence instead if you need exact resumption.

Prerequisites

• KubeMQ server running on localhost:50000
• C# SDK installed (dotnet add package KubeMQ.SDK.CSharp)

Code

// KubeMQ .NET SDK — Events Store: Replay From Time
//
// This example demonstrates replaying events from a specific timestamp.
// Also shows the StartAtTimeDelta option for relative time offsets.
//
// Prerequisites:
//   - KubeMQ server running on localhost:50000
//   - dotnet run

using KubeMQ.Sdk.Client;
using KubeMQ.Sdk.EventsStore;
using System.Text;

await using var client = new KubeMQClient(new KubeMQClientOptions
{
    ClientId = "csharp-eventsstore-replay-from-time-client",
});
await client.ConnectAsync();

Console.WriteLine("Connected to KubeMQ server");

// Publish some events
for (var i = 1; i <= 5; i++)
{
    await client.SendEventStoreAsync(new EventStoreMessage
    {
        Channel = "csharp-eventsstore.replay-from-time",
        Body = Encoding.UTF8.GetBytes($"Event #{i}")
    });
}

Console.WriteLine("Published 5 events. Replaying from last 60 seconds...");

// Subscribe using time delta — replay events from the last 60 seconds
var cts = new CancellationTokenSource();
var subscribeTask = Task.Run(async () =>
{
    await foreach (var msg in client.SubscribeToEventsStoreAsync(
        new EventStoreSubscription
        {
            Channel = "csharp-eventsstore.replay-from-time",
            StartPosition = EventStoreStartPosition.StartAtTimeDelta,
            StartTimeDeltaSeconds = 60
        }, cts.Token))
    {
        Console.WriteLine($"[Seq={msg.Sequence}] {Encoding.UTF8.GetString(msg.Body.Span)}");
    }
});

await Task.Delay(3000);
cts.Cancel();

Console.WriteLine("Done.");

How It Works

• StartAtTimeDelta with StartTimeDeltaSeconds = 60 tells the broker to replay all events stored in the last 60 seconds, then continue with new events.
• This is a relative offset computed at subscription time — useful for catching up after a short outage without knowing exact sequence numbers.
• For an absolute timestamp, use StartAtTime with a DateTime value instead of StartAtTimeDelta.
• The 3-second delay before cancellation gives the replay time to drain; increase it if your channel accumulated many events in the window.

Related

• Pattern overview
• C# SDK Reference
• Persistent Pub/Sub
• Cancel Subscription