From aa4383f352a59610f874fb10450ee26a9cc9383c Mon Sep 17 00:00:00 2001
From: Ryan Melton <ryan@openc3.com>
Date: Sun, 27 Sep 2026 18:29:15 -0600
Subject: [PATCH 1/4] fix(cmd-tlm-api): reap cable subscriptions whose
 disconnect never arrives

Channel broadcasters were tracked in per-class @@broadcasters hashes pruned
only by unsubscribed. If anycable-go never delivered the Disconnect RPC, the
broadcaster and its Redis-reading threads lived for the rest of the process.
Track them in a shared registry, have the frontend Cable heartbeat each
subscription, and reap heartbeating subscriptions that go silent. Clients
that never heartbeat are left alone so existing script clients keep working.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../application_cable/broadcaster_channel.rb  |  67 ++++++++
 .../application_cable/broadcaster_registry.rb | 147 ++++++++++++++++++
 .../app/channels/application_cable/channel.rb |   6 +
 .../app/channels/autonomic_events_channel.rb  |  21 +--
 .../app/channels/calendar_events_channel.rb   |  21 +--
 .../app/channels/config_events_channel.rb     |  21 +--
 .../app/channels/limits_events_channel.rb     |  21 +--
 .../app/channels/messages_channel.rb          |  22 +--
 .../app/channels/notebook_events_channel.rb   |  21 +--
 .../app/channels/queue_events_channel.rb      |  21 +--
 .../app/channels/streaming_channel.rb         |  31 +---
 .../app/channels/system_events_channel.rb     |  21 +--
 .../app/channels/timeline_events_channel.rb   |  21 +--
 .../channels/broadcaster_registry_spec.rb     |  76 +++++++++
 .../spec/channels/streaming_channel_spec.rb   |  47 +++++-
 .../openc3-js-common/src/services/cable.js    |  57 ++++++-
 16 files changed, 451 insertions(+), 170 deletions(-)
 create mode 100644 openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_channel.rb
 create mode 100644 openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_registry.rb
 create mode 100644 openc3-cosmos-cmd-tlm-api/spec/channels/broadcaster_registry_spec.rb

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..8805df6df3
--- /dev/null
+++ b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_channel.rb
@@ -0,0 +1,67 @@
+# 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
+
+    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
+      BroadcasterRegistry.register(subscription_key, create_broadcaster())
+    end
+
+    def unsubscribed
+      if BroadcasterRegistry.registered?(subscription_key)
+        stop_stream_from subscription_key
+        BroadcasterRegistry.unregister(subscription_key)
+      end
+    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..9584352c0e
--- /dev/null
+++ b/openc3-cosmos-cmd-tlm-api/app/channels/application_cable/broadcaster_registry.rb
@@ -0,0 +1,147 @@
+# 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 that have sent at least one 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
+
+    Entry = Struct.new(:broadcaster, :last_activity, :heartbeat)
+
+    @entries = {}
+    @mutex = Mutex.new
+    @sweeper = nil
+
+    class << self
+      def ttl
+        ENV.fetch('OPENC3_CABLE_SUBSCRIPTION_TTL', DEFAULT_TTL).to_f
+      end
+
+      def sweep_interval
+        ENV.fetch('OPENC3_CABLE_SWEEP_INTERVAL', DEFAULT_SWEEP_INTERVAL).to_f
+      end
+
+      def register(key, broadcaster)
+        previous = nil
+        @mutex.synchronize do
+          previous = @entries[key]&.broadcaster
+          @entries[key] = Entry.new(broadcaster, now(), false)
+        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
+
+      def unregister(key)
+        entry = @mutex.synchronize { @entries.delete(key) }
+        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 { @entries.clear }
+      end
+
+      private
+
+      def now
+        Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      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..8910941446 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,12 @@ 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)
+    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..9b5b89622e
--- /dev/null
+++ b/openc3-cosmos-cmd-tlm-api/spec/channels/broadcaster_registry_spec.rb
@@ -0,0 +1,76 @@
+# 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 "honors OPENC3_CABLE_SUBSCRIPTION_TTL" do
+    allow(ENV).to receive(:fetch).and_call_original
+    allow(ENV).to receive(:fetch).with('OPENC3_CABLE_SUBSCRIPTION_TTL', anything).and_return('10')
+    registry.register('key', broadcaster)
+    registry.touch('key', heartbeat: true)
+    expect(registry.sweep(now + 11)).to eql ['key']
+  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..b76c53c0c3 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,49 @@ 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 "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 fc1ec780f7..fdb3f471c4 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() {
@@ -73,6 +87,7 @@ class ResilientSubscription {
 
   unsubscribe() {
     this._unsubscribed = true
+    this._stopHeartbeat()
     this._cable._forget(this)
     return this._subscription?.unsubscribe()
   }
@@ -91,6 +106,13 @@ class ResilientSubscription {
       },
       {
         ...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
@@ -109,13 +131,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
   }
@@ -127,6 +181,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) {

From 994463fbb55d49c98317a6ee2e248af776abfd4a Mon Sep 17 00:00:00 2001
From: "github-actions[bot]"
 <41898282+github-actions[bot]@users.noreply.github.com>
Date: Mon, 12 Oct 2026 02:35:29 +0000
Subject: [PATCH 2/4] fix(review): apply Claude review fixes (turn 1)

- openc3-cosmos-cmd-tlm-api/app/channels/application_cable/channel.rb:25 - SonarQube rubydre:S1186 flagged the empty `heartbeat` method. I added a nested comment explaining why it is an intentional no-op (BroadcasterChannel overrides it), which should also bring new_code_smells_severity back under the gate.

AI-Review-Bot: true
AI-Review-Run: 38196203780
---
 .../app/channels/application_cable/channel.rb                   | 2 ++
 1 file changed, 2 insertions(+)

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 8910941446..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
@@ -23,6 +23,8 @@ class Channel < ActionCable::Channel::Base
     # 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

From 3c26160fc523a8c098e7bd4a208ac63552c9a026 Mon Sep 17 00:00:00 2001
From: Ryan Melton <ryan@openc3.com>
Date: Sun, 25 Oct 2026 16:12:25 -0600
Subject: [PATCH 3/4] fix(cmd-tlm-api): clamp cable ttl and reap subscriptions
 before first heartbeat

A ttl below the heartbeat interval reaped healthy clients, and a zero sweep
interval busy-looped. Clients now declare heartbeat support on subscribe so
ones that die early are reaped, and unsubscribe only removes its own
registration so a late unsubscribe can't kill a newer broadcaster.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../application_cable/broadcaster_channel.rb  | 25 ++++++--
 .../application_cable/broadcaster_registry.rb | 59 +++++++++++++++----
 .../channels/broadcaster_registry_spec.rb     | 54 +++++++++++++++--
 .../spec/channels/streaming_channel_spec.rb   | 26 ++++++++
 .../openc3-js-common/src/services/cable.js    |  3 +
 5 files changed, 144 insertions(+), 23 deletions(-)

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
index 8805df6df3..c4039e9b90 100644
--- 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
@@ -25,6 +25,13 @@ module BroadcasterChannel
     # 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
@@ -36,14 +43,22 @@ def subscribed
       # Defensive: if the auth before_subscribe callback rejected us, skip work.
       return if subscription_rejected?
       stream_from subscription_key
-      BroadcasterRegistry.register(subscription_key, create_broadcaster())
+      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
-      if BroadcasterRegistry.registered?(subscription_key)
-        stop_stream_from subscription_key
-        BroadcasterRegistry.unregister(subscription_key)
-      end
+      # 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)
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
index 9584352c0e..dc8640c422 100644
--- 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
@@ -23,36 +23,47 @@ module ApplicationCable
   # perform and a background sweeper reaps any entry that has stopped
   # heartbeating for longer than TTL.
   #
-  # Only entries that have sent at least one 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.
+  # 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)
+    Entry = Struct.new(:broadcaster, :last_activity, :heartbeat, :token)
 
     @entries = {}
     @mutex = Mutex.new
     @sweeper = nil
+    @clamp_warnings = {}
 
     class << self
       def ttl
-        ENV.fetch('OPENC3_CABLE_SUBSCRIPTION_TTL', DEFAULT_TTL).to_f
+        env_seconds('OPENC3_CABLE_SUBSCRIPTION_TTL', DEFAULT_TTL, MIN_TTL)
       end
 
       def sweep_interval
-        ENV.fetch('OPENC3_CABLE_SWEEP_INTERVAL', DEFAULT_SWEEP_INTERVAL).to_f
+        env_seconds('OPENC3_CABLE_SWEEP_INTERVAL', DEFAULT_SWEEP_INTERVAL, MIN_SWEEP_INTERVAL)
       end
 
-      def register(key, broadcaster)
+      # 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(), false)
+          @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)
@@ -80,8 +91,15 @@ def touch(key, heartbeat: false)
         end
       end
 
-      def unregister(key)
-        entry = @mutex.synchronize { @entries.delete(key) }
+      # 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
@@ -111,7 +129,10 @@ def size
 
       # For tests
       def clear
-        @mutex.synchronize { @entries.clear }
+        @mutex.synchronize do
+          @entries.clear
+          @clamp_warnings.clear
+        end
       end
 
       private
@@ -120,6 +141,20 @@ 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
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
index 9b5b89622e..86691b0ed8 100644
--- a/openc3-cosmos-cmd-tlm-api/spec/channels/broadcaster_registry_spec.rb
+++ b/openc3-cosmos-cmd-tlm-api/spec/channels/broadcaster_registry_spec.rb
@@ -59,12 +59,54 @@ def now
     expect(registry.registered?('legacy')).to be true
   end
 
-  it "honors OPENC3_CABLE_SUBSCRIPTION_TTL" do
-    allow(ENV).to receive(:fetch).and_call_original
-    allow(ENV).to receive(:fetch).with('OPENC3_CABLE_SUBSCRIPTION_TTL', anything).and_return('10')
-    registry.register('key', broadcaster)
-    registry.touch('key', heartbeat: true)
-    expect(registry.sweep(now + 11)).to eql ['key']
+  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
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 b76c53c0c3..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
@@ -131,6 +131,32 @@ def pubsub
       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')
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 38c22f8837..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
@@ -115,6 +115,9 @@ 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,
       },
       {

From fc29ea1a8841b99b4cd557416b0dff0df77d1563 Mon Sep 17 00:00:00 2001
From: Ryan Melton <ryan@openc3.com>
Date: Sun, 25 Oct 2026 19:57:57 -0600
Subject: [PATCH 4/4] test(python): time template protocol initial delay from
 connect

The initial read delay is measured from connect, but the test started its
timer after connect returned, so the measured wait undercounted by connect's
own duration and failed on a slow runner (0.0196s < 0.02s).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
---
 .../test/interfaces/protocols/test_template_protocol.py    | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

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' <VOLTAGE>, (self.<CHANNEL>)", 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):