feat(topics): cap age of target streams with an XADD MINID safety trim

compose.yaml

@@ -395,6 +395,7 @@ services:
       - OPENC3_LOG_STDERR
       - OPENC3_SECRET_FILE_DIR
       - OPENC3_LOG_LEVEL
+      - OPENC3_STREAM_MAX_AGE_SECONDS
     extra_hosts:
       - host.docker.internal:host-gateway
 

docs.openc3.com/docs/configuration/environment.md

@@ -124,6 +124,7 @@ Set in `.env`, or overridden per service in `compose.override.yaml`:
 | `OPENC3_LOGS_BUCKET`, `OPENC3_TOOLS_BUCKET`, `OPENC3_CONFIG_BUCKET` | `logs`, `tools`, `config` | Bucket names |
 | `OPENC3_REDIS_HOSTNAME`, `OPENC3_REDIS_PORT` | `openc3-redis`, `6379` | Redis / Valkey connection |
 | `OPENC3_CLOUD` | `local` | Cloud provider, for bucket and secret handling |
+| `OPENC3_STREAM_MAX_AGE_SECONDS` | `600` | Safety cap on the age of data kept in the Redis / Valkey telemetry and command streams, which are normally trimmed by the log and TSDB microservices. Values below `120` are raised to `120`. Raw streams always keep at least two log cycles plus two minutes. Read by the operator. `0` or empty disables the cap. The cap is measured with each host's clock, so keep hosts time synchronized (NTP) |
 | `RUBYGEMS_URL`, `PYPI_URL`, `NPM_URL`, `MAVEN_URL` | public mirrors | Package sources used at *build* time. To change the URLs COSMOS uses at *run* time, set the `rubygems_url` / `pypi_url` Admin settings |
 
 Per-service runtime flags - `OPENC3_NO_*`, `OPENC3_FORCE_INSTALL`, `OPENC3_ALLOW_HTTP`, `OPENC3_DEFAULT_QUEUE`, `OPENC3_AUTH_RATE_LIMIT_*`, `OPENC3_LANGUAGE`, `OPENC3_LOG_STDERR` - are documented in `compose.override.yaml` under the service each one applies to, because a flag only takes effect on the container that reads it. See [Install and runtime flags](compose.md#install-and-runtime-flags) for how their values are interpreted.

openc3/lib/openc3/logs/log_writer.rb

@@ -63,7 +63,10 @@ class LogWriter
     CYCLE_TIME_INTERVAL = 10
 
     # Delay in seconds before trimming Redis streams
-    CLEANUP_DELAY = 60
+    # This is the primary trimmer of the TELEMETRY__ and COMMAND__ streams. Writers
+    # also apply a safety MINID cap (see Topic.stream_safety_minid) which depends on this
+    # value, so it is defined in Topic.
+    CLEANUP_DELAY = Topic::LOG_CLEANUP_DELAY_SECONDS
 
     # Mutex protecting class variables
     @@mutex = Mutex.new

openc3/lib/openc3/topics/command_decom_topic.rb

@@ -58,7 +58,7 @@ def self.write_packet(packet, scope:)
       msg_hash['json_data'] = JSON.generate(json_hash.as_json, allow_nan: true)
       msg_hash['extra'] = JSON.generate(packet.extra.as_json, allow_nan: true) if packet.extra
       db_shard = Store.db_shard_for_target(packet.target_name, scope: scope)
-      EphemeralStoreQueued.instance(db_shard: db_shard).write_topic(topic, msg_hash)
+      EphemeralStoreQueued.instance(db_shard: db_shard).write_topic(topic, msg_hash, minid: Topic.stream_safety_minid)
     end
 
     def self.get_cmd_item(target_name, packet_name, param_name, type: :FORMATTED, scope: $openc3_scope)

openc3/lib/openc3/topics/command_topic.rb

@@ -35,7 +35,8 @@ def self.write_packet(packet, scope:)
                    buffer: packet.buffer(false) }
       msg_hash[:extra] = JSON.generate(packet.extra.as_json, allow_nan: true) if packet.extra
       db_shard = Store.db_shard_for_target(packet.target_name, scope: scope)
-      EphemeralStoreQueued.instance(db_shard: db_shard).write_topic(topic, msg_hash)
+      minid = Topic.log_stream_safety_minid(packet.target_name, :CMD, scope: scope)
+      EphemeralStoreQueued.instance(db_shard: db_shard).write_topic(topic, msg_hash, minid: minid)
     end
 
     # @param command [Hash] Command hash structure read to be written to a topic

openc3/lib/openc3/topics/telemetry_decom_topic.rb

@@ -45,7 +45,7 @@ def self.write_packet(packet, id: nil, include_limits_states: true, scope:)
         }
         msg_hash[:extra] = JSON.generate(packet.extra.as_json, allow_nan: true) if packet.extra
         db_shard = Store.db_shard_for_target(packet.target_name, scope: scope)
-        Topic.write_topic("#{scope}__DECOM__{#{packet.target_name}}__#{packet.packet_name}", msg_hash, id, db_shard: db_shard)
+        Topic.write_topic("#{scope}__DECOM__{#{packet.target_name}}__#{packet.packet_name}", msg_hash, id, minid: Topic.stream_safety_minid(id), db_shard: db_shard)
 
         unless packet.stored
           # Also update the current value table with the latest decommutated data

openc3/lib/openc3/topics/telemetry_topic.rb

@@ -33,10 +33,11 @@ def self.write_packet(packet, queued: false, scope:)
       msg_hash[:extra] = JSON.generate(packet.extra.as_json, allow_nan: true) if packet.extra
       topic = "#{scope}__TELEMETRY__{#{packet.target_name}}__#{packet.packet_name}"
       db_shard = Store.db_shard_for_target(packet.target_name, scope: scope)
+      minid = Topic.log_stream_safety_minid(packet.target_name, :TLM, scope: scope)
       if queued
-        EphemeralStoreQueued.instance(db_shard: db_shard).write_topic(topic, msg_hash)
+        EphemeralStoreQueued.instance(db_shard: db_shard).write_topic(topic, msg_hash, minid: minid)
       else
-        Topic.write_topic(topic, msg_hash, db_shard: db_shard)
+        Topic.write_topic(topic, msg_hash, minid: minid, db_shard: db_shard)
       end
     end
   end

openc3/lib/openc3/topics/topic.rb

@@ -16,9 +16,48 @@
 # if purchased from OpenC3, Inc.
 
 require 'openc3/utilities/store'
+require 'openc3/utilities/logger'
 
 module OpenC3
   class Topic
+    # Stream trimming contract for the high rate target streams
+    # (TELEMETRY__, COMMAND__, DECOM__, DECOMCMD__):
+    #
+    # The streams are primarily trimmed by their consumers once the data is safely
+    # persisted elsewhere:
+    #   * TELEMETRY__ / COMMAND__ are trimmed by the log microservice (LogWriter) once a
+    #     log file has been moved to the bucket plus LogWriter::CLEANUP_DELAY.
+    #   * DECOM__ / DECOMCMD__ are trimmed by the TSDB microservice which keeps
+    #     TsdbMicroservice::TRIM_KEEP_MS of data.
+    #
+    # If a trimmer is absent (logging disabled, no TSDB) or falls far behind, the streams
+    # would grow without bound. As a safety net every write to these streams also passes
+    # an approximate XADD MINID so no entry older than the safety max age is retained.
+    # The max age is OPENC3_STREAM_MAX_AGE_SECONDS (default 600s, 0 or empty disables) and is
+    # never less than STREAM_MAX_AGE_MIN_SECONDS. For TELEMETRY__ / COMMAND__ the max age is
+    # never less than two log cycles plus the cleanup delay, because historical streaming reads
+    # data not yet in the bucket from the stream.
+    #
+    # When the minid is calculated from now (auto generated ids) it uses this host's clock while
+    # Redis generates the ids from its own clock. A writer clock ahead of Redis by more than the
+    # max age would trim the entire stream (including the entry being added) so hosts must be
+    # time synchronized (NTP).
+    STREAM_MAX_AGE_DEFAULT_SECONDS = 600
+    # Twice TsdbMicroservice::TRIM_KEEP_MS. Also protects against small clock offsets.
+    STREAM_MAX_AGE_MIN_SECONDS = 120
+    # LogWriter::CLEANUP_DELAY is defined from this
+    LOG_CLEANUP_DELAY_SECONDS = 60
+    # TargetModel default log cycle time
+    DEFAULT_LOG_CYCLE_TIME_SECONDS = 600
+    # How long to wait before retrying a failed log cycle time lookup
+    LOG_CYCLE_LOOKUP_RETRY_SECONDS = 60
+    # These are calculated on the write path of every packet so they are cached per process.
+    # The environment doesn't change while a microservice runs and the log cycle time only
+    # changes when the plugin (and therefore the microservices) are reinstalled.
+    @@stream_max_age_ms = nil
+    @@log_stream_max_age_ms = {}
+    @@log_cycle_lookup_retry_at = {}
+
     # Delegate all unknown class methods to EphemeralStore db_shard 0 (system-level topics)
     def self.method_missing(message, *args, **kwargs, &block)
       EphemeralStore.public_send(message, *args, **kwargs, &block)
@@ -37,8 +76,114 @@ def self.get_cnt(topic, db_shard: 0)
     # DB_Shard-aware topic methods for target-specific streams.
     # These explicitly route to the correct EphemeralStore db_shard.
 
-    def self.write_topic(topic, msg_hash, id = '*', maxlen = nil, approximate = 'true', db_shard: 0)
-      EphemeralStore.instance(db_shard: db_shard).write_topic(topic, msg_hash, id, maxlen, approximate)
+    def self.write_topic(topic, msg_hash, id = '*', maxlen = nil, approximate = 'true', minid: nil, db_shard: 0)
+      EphemeralStore.instance(db_shard: db_shard).write_topic(topic, msg_hash, id, maxlen, approximate, minid: minid)
+    end
+
+    # @return [Float] Configured safety max age in seconds (<= 0 means disabled)
+    def self.stream_max_age_seconds
+      value = ENV.fetch('OPENC3_STREAM_MAX_AGE_SECONDS', nil)
+      return STREAM_MAX_AGE_DEFAULT_SECONDS if value.nil?
+      return value.to_f
+    end
+
+    # @return [Integer] Cached safety max age in milliseconds (0 means disabled)
+    def self.stream_max_age_ms
+      @@stream_max_age_ms ||= begin
+        seconds = stream_max_age_seconds
+        if seconds <= 0
+          0
+        elsif seconds < STREAM_MAX_AGE_MIN_SECONDS
+          Logger.warn("OPENC3_STREAM_MAX_AGE_SECONDS of #{seconds} is below the minimum, using #{STREAM_MAX_AGE_MIN_SECONDS}")
+          STREAM_MAX_AGE_MIN_SECONDS * 1000
+        else
+          (seconds * 1000).to_i
+        end
+      end
+    end
+
+    # Calculates the MINID to pass to XADD to cap the age of entries in a target stream.
+    # The cap is anchored to the id being written (or now for auto generated ids) so an
+    # explicitly id'd entry (e.g. decom reusing the raw packet id) is never trimmed by its own add.
+    #
+    # @param id [String] The id being written ('*' or nil for auto generation)
+    # @param max_age_ms [Integer] Max age to retain (<= 0 means disabled)
+    # @return [String|nil] The minid or nil if the safety cap is disabled
+    def self.stream_safety_minid(id = '*', max_age_ms: stream_max_age_ms)
+      return nil if max_age_ms <= 0
+      if id.nil? or id == '*'
+        base_ms = Process.clock_gettime(Process::CLOCK_REALTIME, :millisecond)
+      else
+        base_ms = id.to_s.to_i # Stops at the '-' sequence separator
+      end
+      minid_ms = base_ms - max_age_ms
+      return nil if minid_ms <= 0
+      return minid_ms.to_s
+    end
+
+    # Calculates the MINID for a TELEMETRY__ / COMMAND__ stream write which also never
+    # trims younger than log_stream_min_age_seconds. The resulting max age is cached per stream.
+    #
+    # @param target_name [String] Target name
+    # @param cmd_or_tlm [Symbol] :CMD or :TLM
+    # @return [String|nil] The minid or nil if the safety cap is disabled
+    def self.log_stream_safety_minid(target_name, cmd_or_tlm, scope:)
+      target_cache = ((@@log_stream_max_age_ms[scope] ||= {})[target_name] ||= {})
+      max_age_ms = target_cache[cmd_or_tlm]
+      unless max_age_ms
+        max_age_ms = log_stream_max_age_ms(target_name, cmd_or_tlm, scope: scope)
+        return nil unless max_age_ms # Log cycle time unknown so don't cap yet
+        target_cache[cmd_or_tlm] = max_age_ms
+      end
+      return stream_safety_minid(max_age_ms: max_age_ms)
+    end
+
+    # @param target_name [String] Target name
+    # @param cmd_or_tlm [Symbol] :CMD or :TLM
+    # @return [Integer|nil] Max age in milliseconds for a TELEMETRY__ / COMMAND__ stream (0 means
+    #   disabled) or nil if the log cycle time couldn't be determined. Guessing the log cycle time
+    #   could trim data historical streaming still needs, so the stream isn't capped until a
+    #   lookup (retried every LOG_CYCLE_LOOKUP_RETRY_SECONDS) succeeds.
+    def self.log_stream_max_age_ms(target_name, cmd_or_tlm, scope:)
+      max_age_ms = stream_max_age_ms
+      return max_age_ms if max_age_ms <= 0
+      key = [scope, target_name, cmd_or_tlm]
+      now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      retry_at = @@log_cycle_lookup_retry_at[key]
+      return nil if retry_at and now < retry_at
+      begin
+        min_age_ms = log_stream_min_age_seconds(target_name, cmd_or_tlm, scope: scope) * 1000
+      rescue => e
+        Logger.warn("Unable to determine log cycle time for #{target_name} #{cmd_or_tlm}, stream age not capped: #{e.message}")
+        @@log_cycle_lookup_retry_at[key] = now + LOG_CYCLE_LOOKUP_RETRY_SECONDS
+        return nil
+      end
+      @@log_cycle_lookup_retry_at.delete(key)
+      return [max_age_ms, min_age_ms].max
+    end
+
+    # Minimum age to retain in a TELEMETRY__ / COMMAND__ stream so the log microservice
+    # and historical streaming always have the data not yet moved to the bucket.
+    #
+    # @param target_name [String] Target name
+    # @param cmd_or_tlm [Symbol] :CMD or :TLM
+    # @return [Integer] Minimum age in seconds
+    # @raise If the TargetModel lookup fails
+    def self.log_stream_min_age_seconds(target_name, cmd_or_tlm, scope:)
+      require 'openc3/models/target_model'
+      cycle_time = DEFAULT_LOG_CYCLE_TIME_SECONDS
+      model = TargetModel.get(name: target_name, scope: scope)
+      if model
+        value = model[cmd_or_tlm == :CMD ? 'cmd_log_cycle_time' : 'tlm_log_cycle_time'].to_i
+        cycle_time = value if value > 0
+      end
+      return 2 * (cycle_time + LOG_CLEANUP_DELAY_SECONDS)
+    end
+
+    def self.clear_stream_max_age_cache
+      @@stream_max_age_ms = nil
+      @@log_stream_max_age_ms = {}
+      @@log_cycle_lookup_retry_at = {}
     end
 
     def self.read_topics(topics, offsets = nil, timeout_ms = 1000, count = nil, db_shard: 0, &block)

openc3/lib/openc3/utilities/store_autoload.rb

@@ -264,13 +264,20 @@ def read_topics(topics, offsets = nil, timeout_ms = 1000, count = nil)
     # @option opts [String]  :id          the entry id, default value is `*`, it means auto generation,
     #   if `nil` id is passed it will be changed to `*`
     # @option opts [Integer] :maxlen      max length of entries, default value is `nil`, it means will grow forever
-    # @option opts [String] :approximate whether to add `~` modifier of maxlen or not, default value is 'true'
+    # @option opts [String] :approximate whether to add `~` modifier of maxlen or minid or not, default value is 'true'
+    # @option opts [String] :minid       trim entries with ids lower than this as part of the add (XADD MINID).
+    #   Cannot be combined with maxlen. See Topic.stream_safety_minid for the safety cap applied to target streams.
     #
     # @return [String] the entry id
-    def write_topic(topic, msg_hash, id = '*', maxlen = nil, approximate = 'true')
+    def write_topic(topic, msg_hash, id = '*', maxlen = nil, approximate = 'true', minid: nil)
       id = '*' if id.nil?
+      raise ArgumentError, "write_topic can't supply both maxlen and minid" if maxlen and minid
       @redis_pool.with do |redis|
-        return redis.xadd(topic, msg_hash, id: id, maxlen: maxlen, approximate: approximate)
+        if minid
+          return redis.xadd(topic, msg_hash, id: id, minid: minid, approximate: approximate)
+        else
+          return redis.xadd(topic, msg_hash, id: id, maxlen: maxlen, approximate: approximate)
+        end
       end
     end
 

openc3/python/openc3/logs/log_writer.py

@@ -32,7 +32,10 @@ class LogWriter:
     # granularity.
     CYCLE_TIME_INTERVAL = 10
     # Delay in seconds before trimming Redis streams
-    CLEANUP_DELAY = 60
+    # This is the primary trimmer of the TELEMETRY__ and COMMAND__ streams. Writers
+    # also apply a safety MINID cap (see Topic.stream_safety_minid) which depends on this
+    # value, so it is defined in Topic.
+    CLEANUP_DELAY = Topic.LOG_CLEANUP_DELAY_SECONDS
 
     # Mutex protecting class variables
     mutex = threading.Lock()

openc3/python/openc3/microservices/tsdb_microservice.py

@@ -29,6 +29,9 @@
 
 
 class TsdbMicroservice(Microservice):
+    # This is the primary trimmer of the DECOM__ and DECOMCMD__ streams. Writers also
+    # apply a safety MINID cap (see Topic.stream_safety_minid, default 10 minutes) so the
+    # streams stay bounded if this microservice is not running or falls behind.
     TRIM_KEEP_MS = 60000  # 1 minute
     DEFAULT_FLUSH_PERIOD_S = 5.0  # 5 seconds
 

openc3/python/openc3/topics/command_decom_topic.py

@@ -61,7 +61,7 @@ def write_packet(cls, packet, scope):
         if packet.extra:
             msg_hash["extra"] = json.dumps(packet.extra, cls=JsonEncoder)
         db_shard = Store.db_shard_for_target(packet.target_name, scope=scope)
-        EphemeralStoreQueued.instance(db_shard=db_shard).write_topic(topic, msg_hash)
+        EphemeralStoreQueued.instance(db_shard=db_shard).write_topic(topic, msg_hash, minid=Topic.stream_safety_minid())
 
     @classmethod
     def get_cmd_item(cls, target_name, packet_name, param_name, type="FORMATTED", scope=OPENC3_SCOPE):

openc3/python/openc3/topics/command_topic.py

@@ -38,7 +38,8 @@ def write_packet(cls, packet, scope):
         if packet.extra:
             msg_hash["extra"] = json.dumps(packet.extra)
         db_shard = Store.db_shard_for_target(packet.target_name, scope=scope)
-        EphemeralStoreQueued.instance(db_shard=db_shard).write_topic(topic, msg_hash)
+        minid = Topic.log_stream_safety_minid(packet.target_name, "CMD", scope)
+        EphemeralStoreQueued.instance(db_shard=db_shard).write_topic(topic, msg_hash, minid=minid)
 
     @classmethod
     def send_command(cls, command, timeout, scope, obfuscated_items=None):

openc3/python/openc3/topics/telemetry_decom_topic.py

@@ -50,6 +50,7 @@ def write_packet(cls, packet, id=None, include_limits_states=True, scope=None):
             msg_hash,
             id,
             db_shard=db_shard,
+            minid=Topic.stream_safety_minid(id),
         )
 
         if not packet.stored:

openc3/python/openc3/topics/telemetry_topic.py

@@ -32,14 +32,17 @@ def write_packet(cls, packet, scope, queued=False):
         if packet.extra:
             msg_hash["extra"] = json.dumps(packet.extra)
         db_shard = Store.db_shard_for_target(packet.target_name, scope=scope)
+        minid = Topic.log_stream_safety_minid(packet.target_name, "TLM", scope)
         if queued:
             EphemeralStoreQueued.instance(db_shard=db_shard).write_topic(
                 f"{scope}__TELEMETRY__{{{packet.target_name}}}__{packet.packet_name}",
                 msg_hash,
+                minid=minid,
             )
         else:
             Topic.write_topic(
                 f"{scope}__TELEMETRY__{{{packet.target_name}}}__{packet.packet_name}",
                 msg_hash,
                 db_shard=db_shard,
+                minid=minid,
             )