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 is set to EventsStoreSubscription::StartAtTime(since) with a SystemTime value — 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 time 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 EventsStoreSubscription::StartAtSequence instead if you need exact resumption.

Prerequisites

• KubeMQ server running on localhost:50000
• Rust SDK installed (cargo add kubemq)

Code

use kubemq::prelude::*;
use kubemq::EventsStoreSubscription;
use std::time::{Duration, SystemTime};

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    let channel = "rust-events-store.replay-from-time";
    let one_hour_ago = SystemTime::now() - Duration::from_secs(3600);

    let sub = client.subscribe_to_events_store(
        channel, "",
        EventsStoreSubscription::StartAtTime(one_hour_ago),
        |event| Box::pin(async move {
            println!("Seq {}: {}", event.sequence, String::from_utf8_lossy(&event.body));
        }),
        None,
    ).await?;

    tokio::time::sleep(Duration::from_secs(5)).await;
    sub.unsubscribe().await;
    client.close().await?;
    Ok(())
}

How It Works

• StartAtTime accepts a SystemTime value; the SDK converts to Unix nanoseconds for the server.
• Only events stored at or after the specified time are delivered.
• Review timeouts, channel names, and client IDs before running against shared environments.
• Run the program while the server from the prerequisites is available.

Related

• Events Store overview
• Rust SDK Reference
• Start at Time Delta