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 passes start_at: {:start_at_time, unix_seconds} — 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 {:start_at_sequence, n} instead if you need exact resumption.

Prerequisites

• KubeMQ server running on localhost:50000
• Elixir SDK installed ({:kubemq, "~> 1.0"} in mix.exs)

Code

channel = "elixir-events-store.replay-from-time"
{:ok, client} = KubeMQ.Client.start_link(address: "localhost:50000", client_id: "elixir-es-time")

for i <- 1..3 do
  {:ok, _} = KubeMQ.Client.send_event_store(client,
    KubeMQ.EventStore.new(channel: channel, body: "Old event #{i}"))
end

cutoff = System.system_time(:second)
Process.sleep(1_000)

for i <- 1..3 do
  {:ok, _} = KubeMQ.Client.send_event_store(client,
    KubeMQ.EventStore.new(channel: channel, body: "New event #{i}"))
end

IO.puts("Subscribing from timestamp #{cutoff}...")
parent = self()

{:ok, sub} =
  KubeMQ.Client.subscribe_to_events_store(client, channel,
    start_at: {:start_at_time, cutoff},
    on_event: fn event ->
      IO.puts("Replayed: #{event.body}")
      send(parent, :replayed)
    end
  )

for _ <- 1..3 do
  receive do
    :replayed -> :ok
  after
    5_000 -> :timeout
  end
end

IO.puts("Replayed events from cutoff time")
KubeMQ.Subscription.cancel(sub)
KubeMQ.Client.close(client)

How It Works

• {:start_at_time, unix_seconds} replays events stored after the given Unix timestamp
• Events stored before the cutoff are skipped
• Useful for recovering from a known checkpoint timestamp

Related

• Replay from Sequence
• Start at Time Delta