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..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 @@ -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 "-" 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 "-"; return the milliseconds + def msg_id_ms(msg_id) + msg_id.split('-')[0].to_i + 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 bd2fe62bbd..2976acc9bf 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/streaming_api.rb +++ b/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 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 f865a436ac..cd45c466b4 100644 --- a/openc3-cosmos-cmd-tlm-api/app/models/streaming_thread.rb +++ b/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 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..d70ad12d5b --- /dev/null +++ b/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 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 b5ecc03e57..1b4d424d17 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 @@ -143,6 +143,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 '#kill' do class FakeStreamingThread < StreamingThread def initialize(streaming_api, stop_delay) @@ -653,6 +670,23 @@ def start_fake_thread(stop_delay) @api.kill expect(@api.instance_variable_get('@realtime_thread')).to be_nil end + + it 'skips ahead and notifies when far behind realtime' do + # Messages read are 2 minutes older than the stream head so we're well past the max lag + @time = @start_time - 120 + 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) + 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 + @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..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 @@ -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,47 @@ 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 + } + // 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 = 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 + let message = `Streaming fell behind realtime, skipped ${skippedSeconds.toFixed(1)}s of data` + if (count > 1) { + 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, + 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