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 start_position is set to KubeMQ::PubSub::EventStoreStartPosition::START_AT_TIME with start_position_value given a Unix epoch integer — 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 EventStoreStartPosition::START_AT_SEQUENCE instead if you need exact resumption.

Prerequisites

• KubeMQ server running on localhost:50000
• Ruby SDK installed (gem install kubemq)

Code

require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-events-store.replay-from-time'

begin
  client = KubeMQ::PubSubClient.new(address: address, client_id: 'es-time-example')
  puts "Connected to #{address}"

  3.times do |i|
    msg = KubeMQ::PubSub::EventStoreMessage.new(channel: channel, metadata: "old event #{i}", body: "data-#{i}")
    client.send_event_store(msg)
  end
  puts 'Pre-stored 3 events'

  replay_from = Time.now.to_i
  sleep 1

  3.times do |i|
    msg = KubeMQ::PubSub::EventStoreMessage.new(channel: channel, metadata: "new event #{i}", body: "data-#{i}")
    client.send_event_store(msg)
  end
  puts 'Stored 3 more events after timestamp'
  sleep 1

  cancel = KubeMQ::CancellationToken.new

  sub = KubeMQ::PubSub::EventsStoreSubscription.new(
    channel: channel,
    start_position: KubeMQ::PubSub::EventStoreStartPosition::START_AT_TIME,
    start_position_value: replay_from
  )
  client.subscribe_to_events_store(sub, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |event|
    puts "Received (from timestamp): #{event.metadata}"
  end
  puts "Subscribed with START_AT_TIME=#{replay_from}"

  sleep 3
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  cancel&.cancel
  client&.close
  puts 'Done'
end

How It Works

• START_AT_TIME replays events stored at or after the given Unix timestamp.
• Use Time.now.to_i to capture the current time as a Unix epoch timestamp.
• Only events stored after the timestamp are delivered — older events are skipped.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Replay from Sequence
• Start at Time Delta