From 12000be25f8281230706ab69b1767487843b05f4 Mon Sep 17 00:00:00 2001 From: Ryan Melton Date: Sun, 27 Sep 2026 18:32:04 -0600 Subject: [PATCH 1/5] feat(cmd-tlm-api): skip ahead when realtime streaming falls behind A realtime stream that can't keep up would fall further and further behind, holding an ever growing backlog. Now when the newest data read is more than OPENC3_STREAMING_MAX_LAG_SECONDS (default 10s) old and not catching up, the stream jumps to the latest data and sends an opt-in SKIPPED notice, which the graph reports in its errors. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../app/models/realtime_streaming_thread.rb | 60 +++++++++++++ .../app/models/streaming_api.rb | 12 +++ .../app/models/streaming_thread.rb | 9 ++ .../models/realtime_streaming_thread_spec.rb | 90 +++++++++++++++++++ .../spec/models/streaming_api_spec.rb | 32 +++++++ .../src/components/Graph.vue | 42 +++++++++ 6 files changed, 245 insertions(+) create mode 100644 openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb diff --git a/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb b/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb index efdbf9bcb6..8a67c43bab 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb @@ -18,11 +18,71 @@ 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 + def initialize(streaming_api, collection, scope: nil) super(streaming_api, collection, scope: scope) + @max_lag_seconds = ENV.fetch('OPENC3_STREAMING_MAX_LAG_SECONDS', DEFAULT_MAX_LAG_SECONDS).to_f + @behind_since = nil + @behind_lag = 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. + def check_lag(newest_msg_ids, item_objects_by_topic, packet_objects_by_topic) + return if @max_lag_seconds <= 0 or newest_msg_ids.empty? + + now = Time.now.to_f + newest_ms = newest_msg_ids.values.map { |msg_id| msg_id.split('-')[0].to_i } + lag = now - (newest_ms.min / 1000.0) + if lag <= @max_lag_seconds + @behind_since = nil + @behind_lag = nil + return + end + + if lag < @max_lag_seconds * SKIP_IMMEDIATELY_FACTOR + if @behind_since.nil? + @behind_since = now + @behind_lag = lag + return + end + 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 + end + + skip_ahead(lag, item_objects_by_topic, packet_objects_by_topic) + @behind_since = nil + @behind_lag = nil + end + + def skip_ahead(lag, 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") + (item_objects_by_topic.keys | packet_objects_by_topic.keys).each do |topic| + objects = (item_objects_by_topic[topic] || []) + (packet_objects_by_topic[topic] || []) + next if objects.empty? + last_offset = OpenC3::Topic.get_last_offset(topic, db_shard: objects[0].db_shard) + next if last_offset == "0-0" + objects.each { |object| object.offset = last_offset } + end + @streaming_api.transmit_notice('SKIPPED', "Streaming fell behind realtime, skipped #{lag.round(1)}s of data", skipped_seconds: lag.round(3)) + end end diff --git a/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb b/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb index 7b059e50ee..7257e79793 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb @@ -36,6 +36,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 @@ -167,6 +168,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 @@ -194,6 +197,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) @@ -333,6 +339,12 @@ 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 + 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 diff --git a/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb b/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb index 01ec7b7d7e..d2b1e8269d 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb @@ -92,9 +92,11 @@ 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, db_shard: db_shard) do |topic, msg_id, msg_hash, _| + newest_msg_ids[topic] = msg_id stored = OpenC3::ConfigParser.handle_true_false(msg_hash["stored"]) next if stored # Ignore stored packets while realtime streaming @@ -151,11 +153,18 @@ def redis_thread_body # Check for completed objects by wall clock time if we got nothing check_for_completed_objects() unless any_result + + 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) + end + def handle_message(msg_hash, objects) first_object = objects[0] time = msg_hash['time'].to_i diff --git a/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb new file mode 100644 index 0000000000..a0c17eac07 --- /dev/null +++ b/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb @@ -0,0 +1,90 @@ +# 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' } + + 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=) + @thread = RealtimeStreamingThread.new(@streaming_api, StreamingObjectCollection.new, scope: 'DEFAULT') + allow(OpenC3::Topic).to receive(:get_last_offset).and_return('99999999999999-0') + end + + def msg_id(seconds_ago, now) + "#{((now - seconds_ago) * 1000).to_i}-0" + end + + def check(seconds_ago, now) + allow(Time).to receive(:now).and_return(Time.at(now)) + @thread.check_lag({ topic => msg_id(seconds_ago, now) }, { topic => [@object] }, {}) + end + + it "does nothing while within the max lag" do + now = Time.now.to_f + check(1, now) + check(9, now + 20) + expect(@streaming_api).to_not have_received(:transmit_notice) + expect(@object).to_not have_received(:offset=) + end + + it "skips ahead when the lag doesn't shrink over a window" do + now = Time.now.to_f + check(15, now) + expect(@streaming_api).to_not have_received(:transmit_notice) + check(16, now + 5) + expect(@streaming_api).to_not have_received(:transmit_notice) + check(20, now + 11) + expect(@object).to have_received(:offset=).with('99999999999999-0') + 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 + now = Time.now.to_f + check(40, now) + check(30, now + 11) + check(20, now + 22) + check(11, now + 33) + expect(@streaming_api).to_not have_received(:transmit_notice) + check(5, now + 44) + check(15, now + 50) + check(15, now + 55) + expect(@streaming_api).to_not have_received(:transmit_notice) + end + + it "skips immediately when far behind" do + now = Time.now.to_f + check(61, now) + expect(@streaming_api).to have_received(:transmit_notice).with('SKIPPED', anything, skipped_seconds: be_within(0.01).of(61.0)) + end + + it "uses the most behind topic" do + now = Time.now.to_f + allow(Time).to receive(:now).and_return(Time.at(now)) + other = 'DEFAULT__DECOM__{INST}__ADCS' + @thread.check_lag({ topic => msg_id(1, now), other => msg_id(70, now) }, { topic => [@object] }, { other => [@object] }) + expect(OpenC3::Topic).to have_received(:get_last_offset).twice + expect(@streaming_api).to have_received(:transmit_notice) + 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') + check(1000, Time.now.to_f) + expect(@streaming_api).to_not have_received(:transmit_notice) + end +end diff --git a/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb index c89f6767e2..b686fb12e5 100644 --- a/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb +++ b/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb @@ -139,6 +139,23 @@ def response.is_truncated expect(@api.instance_variable_get('@logged_threads')).to be_empty end + describe '#transmit_notice' do + it 'only sends notices to clients that opted in' do + @api.transmit_notice('SKIPPED', 'skipped', skipped_seconds: 1.5) + expect(@messages).to be_empty + + @api.instance_variable_set('@notices', true) + @api.transmit_notice('SKIPPED', 'skipped', skipped_seconds: 1.5) + expect(@messages).to eq([{ "__type" => "NOTICE", "notice" => "SKIPPED", "message" => "skipped", "skipped_seconds" => 1.5 }]) + end + + it 'opts in via the add request' do + allow(StreamingObject).to receive(:new).and_raise(OpenC3::ForbiddenError) + @api.add({ 'scope' => 'DEFAULT', 'notices' => true, 'end_time' => 0, 'packets' => ['DECOM__TLM__INST__PARAMS__CONVERTED'] }) + expect(@api.instance_variable_get('@notices')).to be true + end + end + describe '#expand_all_packets' do before(:each) do allow(OpenC3::TargetModel).to receive(:names).with(scope: 'DEFAULT').and_return(['INST', 'SYSTEM', 'EMPTY', 'UNKNOWN']) @@ -596,6 +613,21 @@ def response.is_truncated @api.kill expect(@api.instance_variable_get('@realtime_thread')).to be_nil end + + it 'skips ahead and notifies when far behind realtime' do + # Messages were written 2 minutes ago so we're well past the max lag + @time = @start_time - 120 + allow(OpenC3::Topic).to receive(:get_last_offset).and_return("#{Time.now.to_i * 1000}-0") + @send_count = 1 + data['notices'] = true + @api.add(data) + sleep 0.35 # Allow the thread to run + notice = @messages.find { |message| message.is_a?(Hash) } + expect(notice['notice']).to eq('SKIPPED') + expect(notice['skipped_seconds']).to be > 60 + expect(OpenC3::Topic).to have_received(:get_last_offset).at_least(:once) + @api.kill + end end context 'logging plus realtime' do diff --git a/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue b/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue index 40f2e750ca..31528b8ace 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue +++ b/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue @@ -2048,6 +2048,8 @@ export default { items: items, start_time: theStartTime, end_time: this.graphEndDateTime, + // Ask to be told when realtime streaming skips data + notices: true, }) }) // An AuthRequiredError means we have no token and are being @@ -2141,6 +2143,11 @@ export default { }, received: function (data) { this.cable.recordPing() + // Data always arrives as an Array, notices we opted into are an Object + if (!Array.isArray(data)) { + this.receivedNotice(data) + return + } // Buffer incoming data and schedule processing via requestAnimationFrame // to prevent blocking the main thread during historical data floods this.pendingData.push(data) @@ -2150,6 +2157,41 @@ export default { ) } }, + receivedNotice: function (notice) { + if (notice.__type !== 'NOTICE' || notice.notice !== 'SKIPPED') { + return + } + // Coalesce repeated skips into one entry so a graph that can't keep up + // doesn't bury every other error, and only re-report it to an embedding + // screen (which keeps its own list) once a minute + const now = new Date().getTime() + const existing = this.errors.find((error) => error.type === 'skipped') + const skippedSeconds = + (existing?.skippedSeconds || 0) + notice.skipped_seconds + const count = (existing?.count || 0) + 1 + let message = `Streaming fell behind realtime, skipped ${skippedSeconds.toFixed(1)}s of data` + if (count > 1) { + message += ` (${count} times)` + } + if (existing) { + existing.skippedSeconds = skippedSeconds + existing.count = count + existing.message = message + existing.time = now + if (now - existing.emitted > 60000) { + existing.emitted = now + this.$emit('error', existing) + } + } else { + this.addError({ + type: 'skipped', + message: message, + skippedSeconds: skippedSeconds, + count: count, + emitted: now, + }) + } + }, // Drains the pendingData buffer in a time-boxed loop. Each call is // triggered by a single requestAnimationFrame scheduled in received(). // When processing exceeds MAX_PROCESSING_MS we schedule another frame From 6ebdf31cfceba3329be0b721f998d549a50de9c6 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 12 Oct 2026 23:39:15 +0000 Subject: [PATCH 2/5] fix(review): apply Claude review fixes (turn 1) - openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb:165 - SonarQube S1186 flagged the empty base check_lag method; added a comment inside it explaining that it is empty on purpose because logged/historical streams are expected to be behind realtime - openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue:2167 - SonarQube S7759: replaced new Date().getTime() with Date.now() in receivedNotice AI-Review-Bot: true AI-Review-Run: 38322971611 --- openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb | 2 ++ .../plugins/packages/openc3-vue-common/src/components/Graph.vue | 2 +- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb b/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb index d2b1e8269d..4481b00e4e 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb @@ -163,6 +163,8 @@ def redis_thread_body # 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 def handle_message(msg_hash, objects) diff --git a/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue b/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue index 31528b8ace..d595fd95c3 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue +++ b/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue @@ -2164,7 +2164,7 @@ export default { // Coalesce repeated skips into one entry so a graph that can't keep up // doesn't bury every other error, and only re-report it to an embedding // screen (which keeps its own list) once a minute - const now = new Date().getTime() + const now = Date.now() const existing = this.errors.find((error) => error.type === 'skipped') const skippedSeconds = (existing?.skippedSeconds || 0) + notice.skipped_seconds From ea4a544baee30cf95d2d556ffdf13417e7f2b81d Mon Sep 17 00:00:00 2001 From: Jason Thomas Date: Thu, 15 Oct 2026 10:13:53 -0600 Subject: [PATCH 3/5] fix(cmd-tlm-api): measure streaming lag against the stream head Lag compared the API host clock to Redis-stamped msg_ids, so clock skew between hosts counted as lag and caused constant skipping. Measure lag as the gap between the newest msg_id read and the stream's newest msg_id, which both come from the Redis clock. Only advance topics that are behind. Co-Authored-By: Claude Opus 5.5 --- .../app/models/realtime_streaming_thread.rb | 58 ++++++++-- .../models/realtime_streaming_thread_spec.rb | 106 ++++++++++++------ .../spec/models/streaming_api_spec.rb | 8 +- 3 files changed, 127 insertions(+), 45 deletions(-) diff --git a/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb b/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb index 8a67c43bab..764f0706ee 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb @@ -24,12 +24,16 @@ class RealtimeStreamingThread < StreamingThread # 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 = ENV.fetch('OPENC3_STREAMING_MAX_LAG_SECONDS', DEFAULT_MAX_LAG_SECONDS).to_f @behind_since = nil @behind_lag = nil + @last_lag_check = nil end def thread_body @@ -42,12 +46,25 @@ def thread_body # 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) return if @max_lag_seconds <= 0 or newest_msg_ids.empty? - now = Time.now.to_f - newest_ms = newest_msg_ids.values.map { |msg_id| msg_id.split('-')[0].to_i } - lag = now - (newest_ms.min / 1000.0) + 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] + next unless head + topic_lags[topic] = (msg_id_ms(head) - msg_id_ms(msg_id)) / 1000.0 + end + lag = topic_lags.values.max || 0.0 if lag <= @max_lag_seconds @behind_since = nil @behind_lag = nil @@ -69,20 +86,41 @@ def check_lag(newest_msg_ids, item_objects_by_topic, packet_objects_by_topic) end end - skip_ahead(lag, item_objects_by_topic, packet_objects_by_topic) + skip_ahead(lag, topic_lags, heads, item_objects_by_topic, packet_objects_by_topic) @behind_since = nil @behind_lag = nil end - def skip_ahead(lag, item_objects_by_topic, packet_objects_by_topic) + # 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") - (item_objects_by_topic.keys | packet_objects_by_topic.keys).each do |topic| + topic_lags.each do |topic, topic_lag| + next if topic_lag <= @max_lag_seconds objects = (item_objects_by_topic[topic] || []) + (packet_objects_by_topic[topic] || []) - next if objects.empty? - last_offset = OpenC3::Topic.get_last_offset(topic, db_shard: objects[0].db_shard) - next if last_offset == "0-0" - objects.each { |object| object.offset = last_offset } + objects.each { |object| object.offset = heads[topic] } end @streaming_api.transmit_notice('SKIPPED', "Streaming fell behind realtime, skipped #{lag.round(1)}s of data", skipped_seconds: lag.round(3)) 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 = {} + topics_by_shard = topics.group_by do |topic| + objects = item_objects_by_topic[topic] || 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) + results = store.redis_pool.pipelined do + shard_topics.each { |topic| store.xrevrange(topic, count: 1) } + end + shard_topics.zip(results) { |topic, result| heads[topic] = result&.dig(0, 0) } + end + heads + end + + def msg_id_ms(msg_id) + msg_id.split('-')[0].to_i + end end diff --git a/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb index a0c17eac07..055669509e 100644 --- a/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb +++ b/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb @@ -15,76 +15,118 @@ 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(OpenC3::Topic).to receive(:get_last_offset).and_return('99999999999999-0') + allow(Process).to receive(:clock_gettime).and_call_original end - def msg_id(seconds_ago, now) - "#{((now - seconds_ago) * 1000).to_i}-0" + def at(seconds) + allow(Process).to receive(:clock_gettime).with(Process::CLOCK_MONOTONIC).and_return(1000.0 + seconds) end - def check(seconds_ago, now) - allow(Time).to receive(:now).and_return(Time.at(now)) - @thread.check_lag({ topic => msg_id(seconds_ago, now) }, { topic => [@object] }, {}) + 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 - now = Time.now.to_f - check(1, now) - check(9, now + 20) + 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 when the lag doesn't shrink over a window" do - now = Time.now.to_f - check(15, now) + 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, now + 5) + check(16, 5) expect(@streaming_api).to_not have_received(:transmit_notice) - check(20, now + 11) - expect(@object).to have_received(:offset=).with('99999999999999-0') + 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 - now = Time.now.to_f - check(40, now) - check(30, now + 11) - check(20, now + 22) - check(11, now + 33) + check(40, 0) + check(30, 11) + check(20, 22) + check(11, 33) expect(@streaming_api).to_not have_received(:transmit_notice) - check(5, now + 44) - check(15, now + 50) - check(15, now + 55) + 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 - now = Time.now.to_f - check(61, now) + 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 "uses the most behind topic" do - now = Time.now.to_f - allow(Time).to receive(:now).and_return(Time.at(now)) - other = 'DEFAULT__DECOM__{INST}__ADCS' - @thread.check_lag({ topic => msg_id(1, now), other => msg_id(70, now) }, { topic => [@object] }, { other => [@object] }) - expect(OpenC3::Topic).to have_received(:get_last_offset).twice + 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') - check(1000, Time.now.to_f) + 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 + + 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 diff --git a/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb index b686fb12e5..b05375f1ad 100644 --- a/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb +++ b/openc3-cosmos-cmd-tlm-api/spec/models/streaming_api_spec.rb @@ -615,9 +615,12 @@ def response.is_truncated end it 'skips ahead and notifies when far behind realtime' do - # Messages were written 2 minutes ago so we're well past the max lag + # Messages read are 2 minutes older than the stream head so we're well past the max lag @time = @start_time - 120 - allow(OpenC3::Topic).to receive(:get_last_offset).and_return("#{Time.now.to_i * 1000}-0") + head = "#{Time.now.to_i * 1000}-0" + allow_any_instance_of(RealtimeStreamingThread).to receive(:stream_heads) do |_thread, topics| + topics.to_h { |topic| [topic, head] } + end @send_count = 1 data['notices'] = true @api.add(data) @@ -625,7 +628,6 @@ def response.is_truncated notice = @messages.find { |message| message.is_a?(Hash) } expect(notice['notice']).to eq('SKIPPED') expect(notice['skipped_seconds']).to be > 60 - expect(OpenC3::Topic).to have_received(:get_last_offset).at_least(:once) @api.kill end end From 44a4ace150d0de25df09596c9181ccf5d8835b14 Mon Sep 17 00:00:00 2001 From: Ryan Melton Date: Thu, 15 Oct 2026 13:05:27 -0600 Subject: [PATCH 4/5] Add more commenting --- .../app/models/realtime_streaming_thread.rb | 26 +++++++++++++++++++ .../app/models/streaming_api.rb | 2 ++ .../app/models/streaming_thread.rb | 2 ++ .../src/components/Graph.vue | 6 +++++ 4 files changed, 36 insertions(+) diff --git a/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb b/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb index 764f0706ee..2b259937c4 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb @@ -31,8 +31,12 @@ class RealtimeStreamingThread < StreamingThread def initialize(streaming_api, collection, scope: nil) super(streaming_api, collection, scope: scope) @max_lag_seconds = ENV.fetch('OPENC3_STREAMING_MAX_LAG_SECONDS', DEFAULT_MAX_LAG_SECONDS).to_f + # 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 @@ -51,8 +55,11 @@ def thread_body # 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 @@ -61,22 +68,30 @@ def check_lag(newest_msg_ids, 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 "-" 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 @@ -84,9 +99,11 @@ def check_lag(newest_msg_ids, item_objects_by_topic, packet_objects_by_topic) @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 @@ -97,29 +114,38 @@ def skip_ahead(lag, topic_lags, heads, item_objects_by_topic, packet_objects_by_ 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 # 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| objects = item_objects_by_topic[topic] || 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 "-"; return the milliseconds def msg_id_ms(msg_id) msg_id.split('-')[0].to_i end diff --git a/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb b/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb index 7257e79793..cca214a4aa 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb @@ -342,6 +342,8 @@ def transmit_results(results, force: false) # 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 diff --git a/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb b/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb index 4481b00e4e..e75ace6145 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb @@ -96,6 +96,7 @@ def redis_thread_body 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, db_shard: db_shard) do |topic, msg_id, msg_hash, _| + # Recorded before the stored check so skipped stored packets still count as read newest_msg_ids[topic] = msg_id stored = OpenC3::ConfigParser.handle_true_false(msg_hash["stored"]) next if stored # Ignore stored packets while realtime streaming @@ -154,6 +155,7 @@ 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 diff --git a/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue b/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue index d595fd95c3..91b97fbc09 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue +++ b/openc3-cosmos-init/plugins/packages/openc3-vue-common/src/components/Graph.vue @@ -2158,6 +2158,8 @@ export default { } }, receivedNotice: function (notice) { + // Ignore anything we don't understand so new notice types added on the + // server don't break older clients if (notice.__type !== 'NOTICE' || notice.notice !== 'SKIPPED') { return } @@ -2166,6 +2168,7 @@ export default { // screen (which keeps its own list) once a minute const now = Date.now() const existing = this.errors.find((error) => error.type === 'skipped') + // Accumulate the totals across every skip since the entry was created const skippedSeconds = (existing?.skippedSeconds || 0) + notice.skipped_seconds const count = (existing?.count || 0) + 1 @@ -2174,15 +2177,18 @@ export default { message += ` (${count} times)` } if (existing) { + // Update in place so the errors list shows the latest totals existing.skippedSeconds = skippedSeconds existing.count = count existing.message = message existing.time = now + // Throttle emits to the parent, which would otherwise add a new entry per skip if (now - existing.emitted > 60000) { existing.emitted = now this.$emit('error', existing) } } else { + // First skip: addError stamps the time and emits to the parent this.addError({ type: 'skipped', message: message, From c2b3eac44338966a8b0c2d0ffbf7a1c8ddcbf3e1 Mon Sep 17 00:00:00 2001 From: Ryan Melton Date: Sun, 25 Oct 2026 16:26:28 -0600 Subject: [PATCH 5/5] fix(cmd-tlm-api): validate max lag setting and use the right shard for stream heads A malformed OPENC3_STREAMING_MAX_LAG_SECONDS became 0 via to_f and silently disabled skip-ahead; it now warns and uses the default. stream_heads picked an empty item array left by remove, so topics on a nonzero shard were queried on shard 0 and never skipped ahead. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../app/models/realtime_streaming_thread.rb | 17 ++++++++++++-- .../models/realtime_streaming_thread_spec.rb | 22 +++++++++++++++++++ 2 files changed, 37 insertions(+), 2 deletions(-) diff --git a/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb b/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb index 2b259937c4..c48e1cf607 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb +++ b/openc3-cosmos-cmd-tlm-api/app/models/realtime_streaming_thread.rb @@ -30,7 +30,7 @@ class RealtimeStreamingThread < StreamingThread def initialize(streaming_api, collection, scope: nil) super(streaming_api, collection, scope: scope) - @max_lag_seconds = ENV.fetch('OPENC3_STREAMING_MAX_LAG_SECONDS', DEFAULT_MAX_LAG_SECONDS).to_f + @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 @@ -124,12 +124,25 @@ def skip_ahead(lag, topic_lags, heads, item_objects_by_topic, packet_objects_by_ @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| - objects = item_objects_by_topic[topic] || packet_objects_by_topic[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| diff --git a/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb index 055669509e..d70ad12d5b 100644 --- a/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb +++ b/openc3-cosmos-cmd-tlm-api/spec/models/realtime_streaming_thread_spec.rb @@ -106,6 +106,28 @@ def check(lag_seconds, seconds) 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()