feat(cmd-tlm-api): skip ahead when realtime streaming falls behind

openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb

@@ -18,11 +18,148 @@
 require_relative 'streaming_thread'
 
 class RealtimeStreamingThread < StreamingThread
+  # How far behind realtime (in seconds) the stream may fall before we consider
+  # skipping ahead. Set OPENC3_STREAMING_MAX_LAG_SECONDS to 0 to disable.
+  DEFAULT_MAX_LAG_SECONDS = 10.0
+  # Beyond this multiple of the max lag we skip immediately, even if catching up,
+  # because Redis only retains a limited window and the data is about to be trimmed
+  SKIP_IMMEDIATELY_FACTOR = 6
+  # Minimum seconds between lag checks. Each check costs a Redis round trip per
+  # db_shard so don't do it every pass; lag thresholds are many seconds anyway.
+  LAG_CHECK_INTERVAL_SECONDS = 1.0
+
   def initialize(streaming_api, collection, scope: nil)
     super(streaming_api, collection, scope: scope)
+    @max_lag_seconds = max_lag_seconds()
+    # Monotonic time when the current "behind" grace window started (nil when caught up)
+    @behind_since = nil
+    # Lag measured at the start of the current grace window, compared against
+    # at the end of the window to decide whether we're catching up
+    @behind_lag = nil
+    # Monotonic time of the last lag check, used to rate limit the Redis queries
+    @last_lag_check = nil
   end
 
   def thread_body
     redis_thread_body()
   end
+
+  # Realtime displays want current values, so if we can't keep up we jump to the
+  # newest data rather than falling ever further behind. A stream that is over
+  # the threshold but catching up (e.g. just handed off from a historical query)
+  # gets a grace window of @max_lag_seconds, and each window in which the lag
+  # shrinks starts a new one. Only when the lag fails to shrink over a whole
+  # window (or gets extreme) do we skip.
+  #
+  # Lag is how far the newest msg_id we read trails the newest msg_id in the
+  # stream. Both are stamped by the same Redis server (XADD '*') so the result
+  # is immune to clock skew between this container and Redis.
+  def check_lag(newest_msg_ids, item_objects_by_topic, packet_objects_by_topic)
+    # Disabled, or nothing was read this pass so there is nothing to measure.
+    # An idle stream can't be behind: the blocking read would have returned data.
+    return if @max_lag_seconds <= 0 or newest_msg_ids.empty?
+
+    # Monotonic clock so wall clock adjustments (NTP, DST) can't affect the windows
+    now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    return if @last_lag_check and (now - @last_lag_check) < LAG_CHECK_INTERVAL_SECONDS
+    @last_lag_check = now
+
+    heads = stream_heads(newest_msg_ids.keys, item_objects_by_topic, packet_objects_by_topic)
+    topic_lags = {}
+    newest_msg_ids.each do |topic, msg_id|
+      head = heads[topic]
+      # Stream was deleted or trimmed to empty since we read it - can't measure
+      next unless head
+      # msg_ids are "<ms>-<seq>" so the difference of the ms parts is the lag
+      topic_lags[topic] = (msg_id_ms(head) - msg_id_ms(msg_id)) / 1000.0
+    end
+    # The stream as a whole is only as current as its slowest topic
+    lag = topic_lags.values.max || 0.0
+    if lag <= @max_lag_seconds
+      # Within tolerance so reset any grace window in progress
+      @behind_since = nil
+      @behind_lag = nil
+      return
+    end
+
+    # Moderately behind: apply the grace window logic. Extremely behind falls
+    # through to skip immediately since waiting risks reading trimmed data.
+    if lag < @max_lag_seconds * SKIP_IMMEDIATELY_FACTOR
+      if @behind_since.nil?
+        # Just fell behind - start the first grace window
+        @behind_since = now
+        @behind_lag = lag
+        return
+      end
+      # Still inside the current grace window - keep reading normally
+      return if (now - @behind_since) < @max_lag_seconds
+      if lag < @behind_lag
+        # Catching up - give it another window
+        @behind_since = now
+        @behind_lag = lag
+        return
+      end
+      # A whole window passed without the lag shrinking - fall through and skip
+    end
+
+    skip_ahead(lag, topic_lags, heads, item_objects_by_topic, packet_objects_by_topic)
+    # Skipping puts us at the head so start fresh on the next check
+    @behind_since = nil
+    @behind_lag = nil
+  end
+
+  # Only topics that are actually behind are advanced so up to date topics
+  # don't lose messages that arrived during this pass
+  def skip_ahead(lag, topic_lags, heads, item_objects_by_topic, packet_objects_by_topic)
+    OpenC3::Logger.warn("Realtime streaming fell #{lag.round(1)}s behind, skipping ahead to the latest data")
+    topic_lags.each do |topic, topic_lag|
+      next if topic_lag <= @max_lag_seconds
+      # Both item and packet subscriptions can share a topic so move them all
+      objects = (item_objects_by_topic[topic] || []) + (packet_objects_by_topic[topic] || [])
+      # The next XREAD returns entries after the offset, so setting it to the
+      # head msg_id resumes with the first message published after this check
+      objects.each { |object| object.offset = heads[topic] }
+    end
+    # Tell opted in clients so they can show the gap rather than silently dropping data
+    @streaming_api.transmit_notice('SKIPPED', "Streaming fell behind realtime, skipped #{lag.round(1)}s of data", skipped_seconds: lag.round(3))
+  end
+
+  # Only an explicit 0 disables lag handling. A malformed value (which to_f
+  # would silently turn into 0) falls back to the default instead.
+  def max_lag_seconds
+    raw = ENV.fetch('OPENC3_STREAMING_MAX_LAG_SECONDS', nil)
+    return DEFAULT_MAX_LAG_SECONDS if raw.nil? or raw.strip.empty?
+    value = Float(raw, exception: false)
+    return value if value
+    OpenC3::Logger.warn("Invalid OPENC3_STREAMING_MAX_LAG_SECONDS '#{raw}', using #{DEFAULT_MAX_LAG_SECONDS}")
+    DEFAULT_MAX_LAG_SECONDS
+  end
+
+  # Returns the newest msg_id of each topic using one pipelined round trip per db_shard
+  def stream_heads(topics, item_objects_by_topic, packet_objects_by_topic)
+    heads = {}
+    # Every object on a topic lives on the same db_shard so the first one is enough
+    topics_by_shard = topics.group_by do |topic|
+      # Removing the last item subscription leaves an empty array behind, so
+      # fall through to the packet subscriptions rather than defaulting to shard 0
+      objects = item_objects_by_topic[topic].presence || packet_objects_by_topic[topic]
+      objects&.first&.db_shard || 0
+    end
+    topics_by_shard.each do |db_shard, shard_topics|
+      store = OpenC3::EphemeralStore.instance(db_shard: db_shard)
+      # Pipeline so N topics cost a single round trip instead of N
+      results = store.redis_pool.pipelined do
+        # XREVRANGE with COUNT 1 returns just the newest entry
+        shard_topics.each { |topic| store.xrevrange(topic, count: 1) }
+      end
+      # Each result is [[msg_id, fields]] (or empty) so dig out the msg_id
+      shard_topics.zip(results) { |topic, result| heads[topic] = result&.dig(0, 0) }
+    end
+    heads
+  end
+
+  # Redis stream ids are "<milliseconds>-<sequence>"; return the milliseconds
+  def msg_id_ms(msg_id)
+    msg_id.split('-')[0].to_i
+  end
 end

openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb

@@ -38,6 +38,7 @@ def initialize(subscription_key, scope: nil)
     @mutex = Mutex.new
     @realtime_thread = nil
     @logged_threads = []
+    @notices = false
   end
 
   # Create new StreamingObjects using the data['items'] and add them to the StreamingObjectCollection
@@ -168,6 +169,8 @@ def expand_item_globs(data, scope:)
   # token: authorization token
   # start_time: 64-bit nanoseconds from unix epoch - If not present then realtime
   # end_time: 64-bit nanoseconds from unix epoch - If not present stream forever
+  # notices: true to also receive notice Hashes, e.g. { "__type" => "NOTICE", "notice" => "SKIPPED",
+  #   "message" => "...", "skipped_seconds" => 12.3 } when realtime streaming falls behind and skips ahead
   # items: [ [ MODE__CMDORTLM__TARGET__PACKET__ITEM__VALUETYPE__REDUCEDTYPE, item_key] ]
   #   MODE - RAW, DECOM, REDUCED_MINUTE, REDUCED_HOUR, or REDUCED_DAY
   #   CMDORTLM - CMD or TLM
@@ -195,6 +198,9 @@ def add(data)
       end_time = data["end_time"].to_i if data["end_time"]
       scope = data["scope"]
       token = data["token"]
+      # Clients opt in to out-of-band notices (a Hash rather than the usual
+      # Array of results) so older clients never see a message they can't parse
+      @notices = true if data["notices"]
 
       # Expand ALL wildcards in packets before building the collection
       expand_all_packets(data, scope: scope)
@@ -334,6 +340,14 @@ def transmit_results(results, force: false)
     end
   end
 
+  # Send an out-of-band notice (e.g. that data was skipped) to clients that opted in
+  def transmit_notice(notice, message, **extra)
+    return unless @notices
+    # Notices are a Hash while normal results are an Array, which is how clients
+    # tell them apart. Extra keys are stringified to match the JSON the client sees.
+    ActionCable.server.broadcast(@subscription_key, { "__type" => "NOTICE", "notice" => notice, "message" => message }.merge(extra.transform_keys(&:to_s)))
+  end
+
   # Returns if the calling thread should be canceled or not
   def handoff_to_realtime(collection)
     @mutex.synchronize do

openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb

@@ -102,9 +102,12 @@ def redis_thread_body
       # Read from each db_shard with proportionally shorter timeouts
       timeout_per_db_shard = [500 / [db_shard_groups.length, 1].max, 100].max
       any_result = false
+      newest_msg_ids = {} # topic => newest msg_id read this pass, used to measure lag
       db_shard_groups.each do |db_shard, group|
         break if @cancel_thread
         xread_result = OpenC3::Topic.read_topics(group[:topics], group[:offsets], timeout_per_db_shard, @max_batch_size, db_shard: db_shard) do |topic, msg_id, msg_hash, _|
+          # Recorded before the skip check so skipped messages still count as read
+          newest_msg_ids[topic] = msg_id
           if skip_message?(topic, msg_hash)
             # Still advance offsets past skipped messages, otherwise a run of them
             # longer than the read count would be re-read forever
@@ -165,11 +168,21 @@ def redis_thread_body
 
       # Check for completed objects by wall clock time if we got nothing
       check_for_completed_objects() unless any_result
+
+      # Done after processing so a skip only affects where the next pass reads from
+      check_lag(newest_msg_ids, item_objects_by_topic, packet_objects_by_topic) unless @cancel_thread
     else
       @cancel_thread = true
     end
   end
 
+  # Called after each pass through Redis with the newest msg_id read per topic.
+  # Subclasses that stream realtime data override this to react to falling behind.
+  def check_lag(newest_msg_ids, item_objects_by_topic, packet_objects_by_topic)
+    # Intentionally empty: historical (logged) streams replay old data on
+    # purpose so being behind realtime is expected and not acted upon
+  end
+
   # Returns true if a streamed message should not be delivered
   def skip_message?(_topic, msg_hash)
     # Ignore stored packets while realtime streaming

openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb

@@ -0,0 +1,154 @@
+# encoding: ascii-8bit
+
+# Copyright 2026 OpenC3, Inc.
+# All Rights Reserved.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+# See LICENSE.md for more details.
+#
+# This file may also be used under the terms of a commercial license
+# if purchased from OpenC3, Inc.
+
+require 'rails_helper'
+
+RSpec.describe RealtimeStreamingThread, type: :model do
+  let(:topic) { 'DEFAULT__DECOM__{INST}__HEALTH_STATUS' }
+  let(:other) { 'DEFAULT__DECOM__{INST}__ADCS' }
+  # Arbitrary Redis server time (ms) that msg_ids are stamped with
+  let(:base_ms) { 1_700_000_000_000 }
+
+  before(:each) do
+    @streaming_api = double('StreamingApi')
+    allow(@streaming_api).to receive(:transmit_notice)
+    @object = double('StreamingObject', db_shard: 0)
+    allow(@object).to receive(:offset=)
+    @other_object = double('OtherStreamingObject', db_shard: 0)
+    allow(@other_object).to receive(:offset=)
+    @thread = RealtimeStreamingThread.new(@streaming_api, StreamingObjectCollection.new, scope: 'DEFAULT')
+    allow(Process).to receive(:clock_gettime).and_call_original
+  end
+
+  def at(seconds)
+    allow(Process).to receive(:clock_gettime).with(Process::CLOCK_MONOTONIC).and_return(1000.0 + seconds)
+  end
+
+  def head_id(lag_seconds)
+    "#{base_ms + (lag_seconds * 1000).to_i}-0"
+  end
+
+  # Simulate a pass where the newest message read trails the stream head by lag_seconds
+  def check(lag_seconds, seconds)
+    at(seconds)
+    allow(@thread).to receive(:stream_heads).and_return({ topic => head_id(lag_seconds) })
+    @thread.check_lag({ topic => "#{base_ms}-0" }, { topic => [@object] }, {})
+  end
+
+  it "does nothing while within the max lag" do
+    check(1, 0)
+    check(9, 20)
+    expect(@streaming_api).to_not have_received(:transmit_notice)
+    expect(@object).to_not have_received(:offset=)
+  end
+
+  it "skips ahead to the stream head when the lag doesn't shrink over a window" do
+    check(15, 0)
+    expect(@streaming_api).to_not have_received(:transmit_notice)
+    check(16, 5)
+    expect(@streaming_api).to_not have_received(:transmit_notice)
+    check(20, 11)
+    expect(@object).to have_received(:offset=).with(head_id(20))
+    expect(@streaming_api).to have_received(:transmit_notice).with('SKIPPED', /skipped 20.0s/, skipped_seconds: be_within(0.01).of(20.0))
+  end
+
+  it "gives a stream that is catching up more time" do
+    check(40, 0)
+    check(30, 11)
+    check(20, 22)
+    check(11, 33)
+    expect(@streaming_api).to_not have_received(:transmit_notice)
+    check(5, 44)
+    check(15, 50)
+    check(15, 55)
+    expect(@streaming_api).to_not have_received(:transmit_notice)
+  end
+
+  it "skips immediately when far behind" do
+    check(61, 0)
+    expect(@streaming_api).to have_received(:transmit_notice).with('SKIPPED', anything, skipped_seconds: be_within(0.01).of(61.0))
+  end
+
+  it "only checks the lag once per interval" do
+    check(1, 0)
+    check(70, 0.5)
+    expect(@thread).to have_received(:stream_heads).once
+    expect(@streaming_api).to_not have_received(:transmit_notice)
+    check(70, 1.5)
+    expect(@streaming_api).to have_received(:transmit_notice)
+  end
+
+  it "uses the most behind topic and only advances topics that are behind" do
+    at(0)
+    allow(@thread).to receive(:stream_heads).and_return({ topic => head_id(1), other => head_id(70) })
+    @thread.check_lag({ topic => "#{base_ms}-0", other => "#{base_ms}-0" }, { topic => [@object] }, { other => [@other_object] })
+    expect(@other_object).to have_received(:offset=).with(head_id(70))
+    expect(@object).to_not have_received(:offset=)
+    expect(@streaming_api).to have_received(:transmit_notice).with('SKIPPED', anything, skipped_seconds: be_within(0.01).of(70.0))
+  end
+
+  it "can be disabled" do
+    stub_const('ENV', ENV.to_h.merge('OPENC3_STREAMING_MAX_LAG_SECONDS' => '0'))
+    @thread = RealtimeStreamingThread.new(@streaming_api, StreamingObjectCollection.new, scope: 'DEFAULT')
+    allow(@thread).to receive(:stream_heads)
+    check(1000, 0)
+    expect(@thread).to_not have_received(:stream_heads)
+    expect(@streaming_api).to_not have_received(:transmit_notice)
+  end
+
+  it "falls back to the default max lag when the setting is malformed" do
+    stub_const('ENV', ENV.to_h.merge('OPENC3_STREAMING_MAX_LAG_SECONDS' => 'ten'))
+    allow(OpenC3::Logger).to receive(:warn)
+    @thread = RealtimeStreamingThread.new(@streaming_api, StreamingObjectCollection.new, scope: 'DEFAULT')
+    expect(OpenC3::Logger).to have_received(:warn).with(/Invalid OPENC3_STREAMING_MAX_LAG_SECONDS 'ten'/)
+    check(61, 0)
+    expect(@streaming_api).to have_received(:transmit_notice).with('SKIPPED', anything, skipped_seconds: be_within(0.01).of(61.0))
+  end
+
+  it "reads the stream head from the packet subscription's shard when no item subscriptions remain" do
+    shard_object = double('ShardObject', db_shard: 2)
+    store = double('EphemeralStore')
+    pool = double('RedisPool')
+    allow(store).to receive(:redis_pool).and_return(pool)
+    allow(store).to receive(:xrevrange)
+    allow(pool).to receive(:pipelined) { |&block| block.call; [[['5000-0', {}]]] }
+    expect(OpenC3::EphemeralStore).to receive(:instance).with(db_shard: 2).and_return(store)
+    # An empty item array is what StreamingObjectCollection#remove leaves behind
+    heads = @thread.stream_heads([topic], { topic => [] }, { topic => [shard_object] })
+    expect(heads).to eql({ topic => '5000-0' })
+  end
+
+  context "with redis" do
+    before(:each) do
+      mock_redis()
+    end
+
+    it "reads the newest msg_id of each topic" do
+      OpenC3::Topic.write_topic(topic, { 'a' => '1' }, '1000-0')
+      OpenC3::Topic.write_topic(topic, { 'a' => '2' }, '2000-0')
+      OpenC3::Topic.write_topic(other, { 'a' => '3' }, '3000-0')
+      heads = @thread.stream_heads([topic, other, 'DEFAULT__DECOM__{INST}__EMPTY'], { topic => [@object] }, { other => [@object] })
+      expect(heads).to eql({ topic => '2000-0', other => '3000-0', 'DEFAULT__DECOM__{INST}__EMPTY' => nil })
+    end
+
+    it "is not fooled by clock skew between this host and redis" do
+      msg_id = OpenC3::Topic.write_topic(topic, { 'a' => '1' })
+      # This host's clock is far ahead of the Redis server that stamped msg_id
+      allow(Time).to receive(:now).and_return(Time.now + 120)
+      at(0)
+      @thread.check_lag({ topic => msg_id }, { topic => [@object] }, {})
+      expect(@streaming_api).to_not have_received(:transmit_notice)
+      expect(@object).to_not have_received(:offset=)
+    end
+  end
+end