diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_channel.rb new file mode 100644 index 0000000000..c4039e9b90 --- /dev/null +++ b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_channel.rb @@ -0,0 +1,82 @@ +# 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. + +module ApplicationCable + # Shared lifecycle for channels that stream from a per-subscription + # broadcaster object. Including channels declare a key prefix with + # `broadcaster_prefix` and implement a private `create_broadcaster`. + # Broadcasters are tracked in BroadcasterRegistry so a subscription whose + # Disconnect never arrives is eventually reaped. + module BroadcasterChannel + extend ActiveSupport::Concern + + # Sent to a heartbeating client whose broadcaster no longer exists (reaped + # or lost), telling it to resubscribe. Handled inside the frontend Cable + # wrapper so it never reaches the channel's received callback. + RESUBSCRIBE_MESSAGE = { '__openc3_cable__' => 'resubscribe' }.freeze + + included do + # Identifies the broadcaster this subscription registered. Kept in + # AnyCable channel state because channel instances don't survive between + # the subscribe and unsubscribe RPCs. + state_attr_accessor :broadcaster_token + end + + class_methods do + def broadcaster_prefix(prefix = nil) + @broadcaster_prefix = prefix if prefix + @broadcaster_prefix + end + end + + def subscribed + # Defensive: if the auth before_subscribe callback rejected us, skip work. + return if subscription_rejected? + stream_from subscription_key + self.broadcaster_token = SecureRandom.uuid + # Clients that heartbeat say so when they subscribe so they can be reaped + # even if they die before sending their first heartbeat + BroadcasterRegistry.register(subscription_key, create_broadcaster(), + heartbeat: [true, 'true'].include?(params[:heartbeat]), token: broadcaster_token) + end + + def unsubscribed + # No token means subscribed was rejected and nothing was registered + token = broadcaster_token + return unless token + stop_stream_from subscription_key + # Only remove our own registration. A resubscribe on the same connection + # shares the key, and if its subscribe is handled before this unsubscribe + # the newer broadcaster must survive. + BroadcasterRegistry.unregister(subscription_key, token: token) + end + + def heartbeat(_data = nil) + unless BroadcasterRegistry.touch(subscription_key, heartbeat: true) + transmit(RESUBSCRIBE_MESSAGE) + end + end + + private + + def subscription_key + "#{self.class.broadcaster_prefix}_#{uuid}" + end + + # Any perform from the client also counts as activity + def broadcaster + BroadcasterRegistry.touch(subscription_key) + BroadcasterRegistry.get(subscription_key) + end + end +end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_registry.rb b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_registry.rb new file mode 100644 index 0000000000..dc8640c422 --- /dev/null +++ b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_registry.rb @@ -0,0 +1,182 @@ +# 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. + +module ApplicationCable + # Process-wide registry of the per-subscription broadcaster objects + # (StreamingApi, MessagesApi, *EventsApi) that the channels create. + # + # Entries are normally removed by the channel's `unsubscribed`. Under AnyCable + # that arrives as a separate Disconnect RPC from anycable-go, and if it is + # never delivered (anycable-go restarting or crashing, the gRPC stream being + # torn down) the broadcaster and its Redis-reading threads would live for the + # rest of the process. To bound that, clients send a periodic `heartbeat` + # perform and a background sweeper reaps any entry that has stopped + # heartbeating for longer than TTL. + # + # Only entries registered as heartbeating (the client said it heartbeats when + # it subscribed) or that have sent a heartbeat are reaped. Older clients (and + # the Ruby/Python scripting WebSocket APIs) never heartbeat, and a realtime + # stream legitimately goes a long time with no client-to-server traffic, so + # reaping them on idle time alone would kill healthy streams. + class BroadcasterRegistry + # Clients heartbeat every 60s, so this tolerates several missed heartbeats + # (e.g. background tab timer throttling) before reaping + DEFAULT_TTL = 300 + DEFAULT_SWEEP_INTERVAL = 60 + # A ttl shorter than the client heartbeat interval would reap healthy + # clients between heartbeats, so never go below two heartbeats + MIN_TTL = 120 + # Keeps a zero or invalid interval from turning the sweeper into a busy loop + MIN_SWEEP_INTERVAL = 1 + + Entry = Struct.new(:broadcaster, :last_activity, :heartbeat, :token) + + @entries = {} + @mutex = Mutex.new + @sweeper = nil + @clamp_warnings = {} + + class << self + def ttl + env_seconds('OPENC3_CABLE_SUBSCRIPTION_TTL', DEFAULT_TTL, MIN_TTL) + end + + def sweep_interval + env_seconds('OPENC3_CABLE_SWEEP_INTERVAL', DEFAULT_SWEEP_INTERVAL, MIN_SWEEP_INTERVAL) + end + + # heartbeat: true makes the entry reapable immediately rather than only + # after its first heartbeat, so a client that dies right after + # subscribing is still reaped. token identifies this registration so a + # late unregister from an older subscription can't remove a newer one. + def register(key, broadcaster, heartbeat: false, token: nil) + previous = nil + @mutex.synchronize do + previous = @entries[key]&.broadcaster + @entries[key] = Entry.new(broadcaster, now(), heartbeat, token) + end + # A duplicate subscribe for the same key must not leak the old one + kill(key, previous) if previous and !previous.equal?(broadcaster) + start_sweeper() + broadcaster + end + + def get(key) + @mutex.synchronize { @entries[key]&.broadcaster } + end + + def registered?(key) + @mutex.synchronize { @entries.key?(key) } + end + + # Record client activity. Returns false if the key is not registered, + # e.g. because it was already reaped. + def touch(key, heartbeat: false) + @mutex.synchronize do + entry = @entries[key] + return false unless entry + entry.last_activity = now() + entry.heartbeat = true if heartbeat + return true + end + end + + # When token is given only the registration made with that token is + # removed. Returns whether an entry was removed. + def unregister(key, token: nil) + entry = @mutex.synchronize do + current = @entries[key] + next nil unless current + next nil if token and current.token != token + @entries.delete(key) + end + kill(key, entry.broadcaster) if entry and entry.broadcaster + !entry.nil? + end + + # Remove and kill every heartbeating entry idle for longer than ttl. + # Returns the reaped keys. + def sweep(current_time = now()) + expired = [] + @mutex.synchronize do + cutoff = current_time - ttl() + @entries.each do |key, entry| + expired << [key, entry] if entry.heartbeat and entry.last_activity < cutoff + end + expired.each { |key, _| @entries.delete(key) } + end + # Kill outside the lock as it may wait on the broadcaster's threads + expired.each do |key, entry| + OpenC3::Logger.warn("Reaping cable subscription #{key}: no heartbeat in #{(current_time - entry.last_activity).round}s") + kill(key, entry.broadcaster) if entry.broadcaster + end + expired.map(&:first) + end + + def size + @mutex.synchronize { @entries.length } + end + + # For tests + def clear + @mutex.synchronize do + @entries.clear + @clamp_warnings.clear + end + end + + private + + def now + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + def env_seconds(name, default, minimum) + raw = ENV.fetch(name, nil) + return default.to_f if raw.nil? or raw.to_s.strip.empty? + value = Float(raw, exception: false) + return value if value and value >= minimum + fallback = value ? minimum.to_f : default.to_f + # Only warn once per value, ttl is read on every sweep + unless @clamp_warnings[name] == raw + @clamp_warnings[name] = raw + OpenC3::Logger.warn("#{name}=#{raw} is invalid or below the minimum of #{minimum}s, using #{fallback}s") + end + fallback + end + + def kill(key, broadcaster) + broadcaster.kill + rescue => e + OpenC3::Logger.error("Error killing cable subscription #{key}: #{e.formatted}") + end + + def start_sweeper + return if @sweeper&.alive? + @mutex.synchronize do + return if @sweeper&.alive? + @sweeper = Thread.new do + loop do + sleep(sweep_interval()) + begin + sweep() + rescue => e + OpenC3::Logger.error("Cable subscription sweeper error: #{e.formatted}") + end + end + end + end + end + end + end +end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/channel.rb index 00bb90f210..a346c3b08d 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/channel.rb @@ -19,6 +19,14 @@ class Channel < ActionCable::Channel::Base before_subscribe :authenticate_subscription! + # The frontend Cable sends a periodic heartbeat on every subscription. + # Accept it on every channel so ones that don't track liveness (see + # BroadcasterChannel) don't log it as an unknown action. + def heartbeat(_data = nil) + # Intentionally a no-op: channels without a broadcaster have no + # liveness to record. BroadcasterChannel overrides this. + end + private # Authenticate the subscription using either: diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/autonomic_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/autonomic_events_channel.rb index 9ec42dfdf1..a4b8899b2c 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/autonomic_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/autonomic_events_channel.rb @@ -16,23 +16,12 @@ # if purchased from OpenC3, Inc. class AutonomicEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'autonomic_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "autonomic_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = AutonomicEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "autonomic_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + AutonomicEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/calendar_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/calendar_events_channel.rb index 3000a8efd9..f6c23f6813 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/calendar_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/calendar_events_channel.rb @@ -16,23 +16,12 @@ # if purchased from OpenC3, Inc. class CalendarEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'calendar_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "calendar_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = CalendarEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "calendar_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + CalendarEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/config_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/config_events_channel.rb index 7e9b3ee399..4969b1c9db 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/config_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/config_events_channel.rb @@ -12,23 +12,12 @@ # if purchased from OpenC3, Inc. class ConfigEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'config_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "config_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = ConfigEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "config_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + ConfigEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/limits_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/limits_events_channel.rb index af9fce4651..7591fe4faf 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/limits_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/limits_events_channel.rb @@ -16,23 +16,12 @@ # if purchased from OpenC3, Inc. class LimitsEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'limits_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "limits_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = LimitsEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "limits_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + LimitsEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/messages_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/messages_channel.rb index 8578821f63..1c39be7339 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/messages_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/messages_channel.rb @@ -12,15 +12,13 @@ # if purchased from OpenC3, Inc. class MessagesChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'messages' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "messages_#{uuid}" - stream_from subscription_key + private - @@broadcasters[subscription_key] = MessagesApi.new( + def create_broadcaster + MessagesApi.new( subscription_key, params["history_count"], start_offset: params["start_offset"], @@ -31,14 +29,4 @@ def subscribed scope: scope ) end - - def unsubscribed - subscription_key = "messages_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end - end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/notebook_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/notebook_events_channel.rb index 17e13a0d40..b37db5471d 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/notebook_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/notebook_events_channel.rb @@ -7,23 +7,12 @@ # if purchased from OpenC3, Inc. class NotebookEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'notebook_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "notebook_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = NotebookEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "notebook_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + NotebookEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/queue_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/queue_events_channel.rb index 48ee9700b0..eaa01b9b9a 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/queue_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/queue_events_channel.rb @@ -12,23 +12,12 @@ # if purchased from OpenC3, Inc. class QueueEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'queue_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "queue_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = QueueEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "queue_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + QueueEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/streaming_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/streaming_channel.rb index 755c9c1e70..7e2e4969b5 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/streaming_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/streaming_channel.rb @@ -16,25 +16,8 @@ # if purchased from OpenC3, Inc. class StreamingChannel < ApplicationCable::Channel - @@broadcasters = {} - - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "streaming_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = StreamingApi.new(subscription_key, scope: scope) - end - - def unsubscribed - subscription_key = "streaming_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end - end + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'streaming' # data holds the following keys: # start_time - nsec_since_epoch - null for realtime @@ -42,7 +25,6 @@ def unsubscribed # items [Array of Item keys] ie ["DECOM__TLM__INST__ADCS__Q1__RAW"] # scope def add(data) - subscription_key = "streaming_#{uuid}" if validate_data(data) begin # Nil-guard the broadcaster: it only exists between `subscribed` (which @@ -51,7 +33,7 @@ def add(data) # raise NoMethodError here, get rescued below, and reject_subscription — # killing every panel on the connection. A no-op is the correct response; # the client re-subscribes and replays its adds. - @@broadcasters[subscription_key]&.add(data) + broadcaster()&.add(data) rescue OpenC3::AuthError, OpenC3::ForbiddenError transmit({ "error" => "unauthorized" }) reject() # Sets the rejected state on the connection @@ -68,13 +50,12 @@ def add(data) # items [Array of Item keys] ie ["DECOM__TLM__INST__ADCS__Q1__RAW"] # scope def remove(data) - subscription_key = "streaming_#{uuid}" if validate_data(data) begin # Nil-guard the broadcaster (see `add`): a `remove` that races a # rejected/torn-down subscription must be a harmless no-op, not a # NoMethodError that reject_subscription()s every panel on the connection. - @@broadcasters[subscription_key]&.remove(data) + broadcaster()&.remove(data) rescue OpenC3::AuthError, OpenC3::ForbiddenError transmit({ "error" => "unauthorized" }) reject() # Sets the rejected state on the connection @@ -89,6 +70,10 @@ def remove(data) private + def create_broadcaster + StreamingApi.new(subscription_key, scope: scope) + end + def validate_data(data) result = true unless data['scope'] diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/system_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/system_events_channel.rb index de699cb39e..70add0d41e 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/system_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/system_events_channel.rb @@ -12,23 +12,12 @@ # if purchased from OpenC3, Inc. class SystemEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'system_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "system_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = SystemEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "system_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + SystemEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/app/channels/timeline_events_channel.rb b/openc3-cosmos-cmd-tlm-api/app/channels/timeline_events_channel.rb index 6900b1c00c..4dfc04d39a 100644 --- a/openc3-cosmos-cmd-tlm-api/app/channels/timeline_events_channel.rb +++ b/openc3-cosmos-cmd-tlm-api/app/channels/timeline_events_channel.rb @@ -16,23 +16,12 @@ # if purchased from OpenC3, Inc. class TimelineEventsChannel < ApplicationCable::Channel - @@broadcasters = {} + include ApplicationCable::BroadcasterChannel + broadcaster_prefix 'timeline_events' - def subscribed - # Defensive: if the auth before_subscribe callback rejected us, skip work. - return if subscription_rejected? - subscription_key = "timeline_events_#{uuid}" - stream_from subscription_key - @@broadcasters[subscription_key] = TimelineEventsApi.new(subscription_key, params['history_count'], scope: scope) - end + private - def unsubscribed - subscription_key = "timeline_events_#{uuid}" - if @@broadcasters[subscription_key] - stop_stream_from subscription_key - @@broadcasters[subscription_key].kill - @@broadcasters[subscription_key] = nil - @@broadcasters.delete(subscription_key) - end + def create_broadcaster + TimelineEventsApi.new(subscription_key, params['history_count'], scope: scope) end end diff --git a/openc3-cosmos-cmd-tlm-api/spec/channels/broadcaster_registry_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/channels/broadcaster_registry_spec.rb new file mode 100644 index 0000000000..86691b0ed8 --- /dev/null +++ b/openc3-cosmos-cmd-tlm-api/spec/channels/broadcaster_registry_spec.rb @@ -0,0 +1,118 @@ +# 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 ApplicationCable::BroadcasterRegistry do + let(:registry) { described_class } + let(:broadcaster) { double("broadcaster", kill: nil) } + + def now + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + before(:each) do + registry.clear + allow(OpenC3::Logger).to receive(:warn) + allow(OpenC3::Logger).to receive(:error) + end + after(:each) { registry.clear } + + it "registers, gets and unregisters" do + registry.register('key', broadcaster) + expect(registry.get('key')).to equal(broadcaster) + expect(broadcaster).to receive(:kill) + expect(registry.unregister('key')).to be true + expect(registry.get('key')).to be_nil + expect(registry.unregister('key')).to be false + end + + it "touch returns false for unknown keys" do + expect(registry.touch('missing', heartbeat: true)).to be false + end + + it "keeps heartbeating entries alive while they heartbeat" do + registry.register('key', broadcaster) + registry.touch('key', heartbeat: true) + expect(registry.sweep(now + registry.ttl - 5)).to be_empty + expect(registry.registered?('key')).to be true + end + + it "reaps only heartbeating entries past the ttl" do + legacy = double("legacy", kill: nil) + registry.register('key', broadcaster) + registry.register('legacy', legacy) + registry.touch('key', heartbeat: true) + expect(broadcaster).to receive(:kill) + expect(legacy).not_to receive(:kill) + expect(registry.sweep(now + registry.ttl + 1)).to eql ['key'] + expect(registry.registered?('legacy')).to be true + end + + it "reaps an entry registered as heartbeating before its first heartbeat" do + registry.register('key', broadcaster, heartbeat: true) + expect(broadcaster).to receive(:kill) + expect(registry.sweep(now + registry.ttl + 1)).to eql ['key'] + end + + it "only unregisters the registration matching the token" do + newer = double("newer", kill: nil) + registry.register('key', broadcaster, token: 'old') + registry.register('key', newer, token: 'new') + expect(registry.unregister('key', token: 'old')).to be false + expect(registry.get('key')).to equal(newer) + expect(newer).not_to have_received(:kill) + expect(registry.unregister('key', token: 'new')).to be true + expect(newer).to have_received(:kill) + expect(registry.registered?('key')).to be false + end + + context "environment overrides" do + def stub_env(name, value) + allow(ENV).to receive(:fetch).and_call_original + allow(ENV).to receive(:fetch).with(name, anything).and_return(value) + end + + it "honors OPENC3_CABLE_SUBSCRIPTION_TTL" do + stub_env('OPENC3_CABLE_SUBSCRIPTION_TTL', '200') + registry.register('key', broadcaster) + registry.touch('key', heartbeat: true) + expect(registry.sweep(now + 201)).to eql ['key'] + end + + it "clamps a ttl below the minimum and warns once" do + stub_env('OPENC3_CABLE_SUBSCRIPTION_TTL', '10') + expect(OpenC3::Logger).to receive(:warn).with(/OPENC3_CABLE_SUBSCRIPTION_TTL=10/).once + expect(registry.ttl).to eql described_class::MIN_TTL.to_f + expect(registry.ttl).to eql described_class::MIN_TTL.to_f + end + + it "uses the default ttl when the value is not a number" do + stub_env('OPENC3_CABLE_SUBSCRIPTION_TTL', 'abc') + expect(registry.ttl).to eql described_class::DEFAULT_TTL.to_f + expect(OpenC3::Logger).to have_received(:warn).with(/OPENC3_CABLE_SUBSCRIPTION_TTL=abc/) + end + + it "clamps a zero sweep interval" do + stub_env('OPENC3_CABLE_SWEEP_INTERVAL', '0') + expect(registry.sweep_interval).to eql described_class::MIN_SWEEP_INTERVAL.to_f + end + end + + it "survives a broadcaster whose kill raises" do + allow(broadcaster).to receive(:kill).and_raise("boom") + registry.register('key', broadcaster) + registry.touch('key', heartbeat: true) + expect(registry.sweep(now + registry.ttl + 1)).to eql ['key'] + end +end diff --git a/openc3-cosmos-cmd-tlm-api/spec/channels/streaming_channel_spec.rb b/openc3-cosmos-cmd-tlm-api/spec/channels/streaming_channel_spec.rb index 02a666e337..9b03a78aef 100644 --- a/openc3-cosmos-cmd-tlm-api/spec/channels/streaming_channel_spec.rb +++ b/openc3-cosmos-cmd-tlm-api/spec/channels/streaming_channel_spec.rb @@ -80,7 +80,7 @@ def pubsub # client sends, so validate_data passes and we actually exercise the guard.) before do subscribe() - StreamingChannel.class_variable_get(:@@broadcasters)['streaming_12345'] = nil + ApplicationCable::BroadcasterRegistry.unregister('streaming_12345') end it "does not reject a remove" do @@ -93,4 +93,75 @@ def pubsub expect(subscription).not_to be_rejected end end + + context "broadcaster lifecycle" do + let(:registry) { ApplicationCable::BroadcasterRegistry } + + before(:each) { allow(OpenC3::Logger).to receive(:warn) } + + it "registers on subscribe and unregisters on unsubscribe" do + subscribe() + expect(registry.get('streaming_12345')).to be_a(StreamingApi) + unsubscribe() + expect(registry.registered?('streaming_12345')).to be false + end + + it "kills a previous broadcaster registered under the same key" do + subscribe() + first = registry.get('streaming_12345') + expect(first).to receive(:kill) + subscribe() + expect(registry.get('streaming_12345')).not_to equal(first) + end + + it "reaps a heartbeating subscription that stops heartbeating" do + subscribe() + subscription.heartbeat({}) + broadcaster = registry.get('streaming_12345') + expect(broadcaster).to receive(:kill) + now = Process.clock_gettime(Process::CLOCK_MONOTONIC) + expect(registry.sweep(now + registry.ttl + 1)).to eql ['streaming_12345'] + expect(registry.registered?('streaming_12345')).to be false + end + + it "does not reap a subscription that never heartbeats" do + subscribe() + now = Process.clock_gettime(Process::CLOCK_MONOTONIC) + expect(registry.sweep(now + registry.ttl + 1)).to be_empty + expect(registry.registered?('streaming_12345')).to be true + end + + it "reaps a subscription that said it heartbeats before its first heartbeat" do + subscribe(heartbeat: true) + expect(registry.get('streaming_12345')).to receive(:kill) + now = Process.clock_gettime(Process::CLOCK_MONOTONIC) + expect(registry.sweep(now + registry.ttl + 1)).to eql ['streaming_12345'] + end + + it "does not let a late unsubscribe remove a newer subscription's broadcaster" do + subscribe() + first = subscription + allow(registry.get('streaming_12345')).to receive(:kill) + subscribe() + second = registry.get('streaming_12345') + expect(second).not_to receive(:kill) + first.unsubscribe_from_channel + expect(registry.get('streaming_12345')).to equal(second) + end + + it "does nothing on unsubscribe when subscribe was rejected" do + subscribe() + subscription.broadcaster_token = nil + expect(registry.get('streaming_12345')).not_to receive(:kill) + unsubscribe() + expect(registry.registered?('streaming_12345')).to be true + end + + it "tells a heartbeating client to resubscribe once its broadcaster is gone" do + subscribe() + registry.unregister('streaming_12345') + subscription.heartbeat({}) + expect(transmissions.last).to eql({ '__openc3_cable__' => 'resubscribe' }) + end + end end diff --git a/openc3-cosmos-init/plugins/packages/openc3-js-common/src/services/cable.js b/openc3-cosmos-init/plugins/packages/openc3-js-common/src/services/cable.js index 4899bfb955..684bf54c1d 100644 --- a/openc3-cosmos-init/plugins/packages/openc3-js-common/src/services/cable.js +++ b/openc3-cosmos-init/plugins/packages/openc3-js-common/src/services/cable.js @@ -23,6 +23,19 @@ import { isAuthRequiredError } from './authGuard' // enough that a tool isn't visibly stuck. const RECOVERY_DELAY = 1000 +const DEFAULT_URL = '/openc3-api/cable' + +// How often each subscription tells the cmd-tlm-api it is still alive. The +// server reaps broadcasters that stop heartbeating (see +// ApplicationCable::BroadcasterRegistry), which bounds the lifetime of a +// subscription whose disconnect never reached it. +const HEARTBEAT_INTERVAL = 60000 + +// Sent by the server in reply to a heartbeat when it no longer has a +// broadcaster for this subscription (e.g. it was reaped), asking us to +// resubscribe so the channel's subscribed callback runs again. +const RESUBSCRIBE_KEY = '__openc3_cable__' + // Wraps an anycable subscription so one that dies unrecoverably is replaced // transparently, keeping the object the caller holds valid across the swap. // @@ -57,6 +70,7 @@ class ResilientSubscription { this._additionalOptions = additionalOptions this._subscription = null this._unsubscribed = false + this._heartbeat = null } get identifier() { @@ -86,6 +100,7 @@ class ResilientSubscription { unsubscribe() { this._unsubscribed = true + this._stopHeartbeat() this._cable._forget(this) return this._subscription?.unsubscribe() } @@ -100,10 +115,20 @@ class ResilientSubscription { { channel: this._channel, token: localStorage.openc3Token, + // Tells the server this subscription heartbeats, so it can be reaped + // even if it dies before the first heartbeat is sent + ...(this._cable._heartbeatInterval ? { heartbeat: true } : {}), ...this._additionalOptions, }, { ...this._callbacks, + received: (data) => { + if (data?.[RESUBSCRIBE_KEY] === 'resubscribe') { + this._cable._scheduleRecovery() + return + } + this._callbacks.received?.(data) + }, disconnected: (data) => { // We tore this subscription down ourselves (unsubscribe, or the // whole cable disconnecting because the component unmounted). The @@ -122,13 +147,45 @@ class ResilientSubscription { }, }, ) + this._startHeartbeat() + } + + _startHeartbeat() { + const interval = this._cable._heartbeatInterval + if (!interval || this._heartbeat) { + return + } + this._heartbeat = setInterval(() => { + if (this._unsubscribed) { + this._stopHeartbeat() + return + } + // Failures (e.g. while disconnected) are expected and harmless; the next + // heartbeat or the reconnect's resubscribe covers them + Promise.resolve() + .then(() => this._subscription?.perform('heartbeat', {})) + .catch(() => {}) + }, interval) + } + + _stopHeartbeat() { + if (this._heartbeat) { + clearInterval(this._heartbeat) + this._heartbeat = null + } } } export default class Cable { - constructor(url = '/openc3-api/cable') { + // Heartbeats default on only for the cmd-tlm-api cable, whose channels + // track subscription liveness. Other backends may not accept the action. + constructor( + url = DEFAULT_URL, + { heartbeatInterval = url === DEFAULT_URL ? HEARTBEAT_INTERVAL : 0 } = {}, + ) { this._cable = null this._url = url + this._heartbeatInterval = heartbeatInterval this._subscriptions = new Set() this._recovery = null } @@ -140,6 +197,7 @@ export default class Cable { // recovery, since we just cancelled one and the consumer is going away. this._subscriptions.forEach((subscription) => { subscription._unsubscribed = true + subscription._stopHeartbeat() }) this._subscriptions.clear() if (this._cable) { diff --git a/openc3/python/test/interfaces/protocols/test_template_protocol.py b/openc3/python/test/interfaces/protocols/test_template_protocol.py index 86e269b9dc..eb7f36120a 100644 --- a/openc3/python/test/interfaces/protocols/test_template_protocol.py +++ b/openc3/python/test/interfaces/protocols/test_template_protocol.py @@ -164,14 +164,17 @@ def test_waits_before_writing_during_the_initial_delay_period(self): self.interface.stream = TestTemplateProtocol.TemplateStream() self.interface.add_protocol(TemplateProtocol, ["0xABCD", "0xABCD", 0, 0.02], "READ_WRITE") packet = self._command_packet("SOUR'VOLT' , (self.)", VOLTAGE=1, CHANNEL=2) + # The initial delay is measured from connect, so start timing before + # connecting. Timing from after connect undercounts by however long + # connect took and can come in just under the delay on a slow runner + start = time.time() self.interface.connect() - write = time.time() self.interface.write(packet) # The write really blocks until the initial delay has elapsed. Only the # lower bound can be checked here because a loaded machine can take # arbitrarily longer; test_waits_no_longer_than_the_initial_delay_before_writing # covers the protocol not waiting longer than it was configured to - self.assertGreaterEqual(time.time() - write, 0.02) + self.assertGreaterEqual(time.time() - start, 0.02) self.assertEqual(TestTemplateProtocol.write_buffer, b"SOUR'VOLT' 1, (self.2)\xab\xcd") def test_works_without_a_response(self):