fix(topics): harden stream safety trim config and lookup failures

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,7 +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. Raw streams always keep at least two log cycles plus one minute. Set on the operator. `0` or empty disables the cap |
+| `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

@@ -64,9 +64,9 @@ class LogWriter
 
     # Delay in seconds before trimming Redis streams
     # This is the primary trimmer of the TELEMETRY__ and COMMAND__ streams. Writers
-    # also apply a safety MINID cap (see Topic.stream_safety_minid) which assumes this
-    # value is mirrored by Topic::LOG_CLEANUP_DELAY_SECONDS.
-    CLEANUP_DELAY = 60
+    # 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/topic.rb

@@ -33,19 +33,30 @@ class Topic
     # 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).
-    # 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.
+    # 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
-    # Mirrors LogWriter::CLEANUP_DELAY and the TargetModel default log cycle time
+    # 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)
@@ -78,7 +89,17 @@ def self.stream_max_age_seconds
 
     # @return [Integer] Cached safety max age in milliseconds (0 means disabled)
     def self.stream_max_age_ms
-      @@stream_max_age_ms ||= [(stream_max_age_seconds * 1000).to_i, 0].max
+      @@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.
@@ -110,40 +131,59 @@ 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 = stream_max_age_ms
-        if max_age_ms > 0
-          min_age_ms = log_stream_min_age_seconds(target_name, cmd_or_tlm, scope: scope) * 1000
-          max_age_ms = min_age_ms if min_age_ms > max_age_ms
-        end
+        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
-      begin
-        require 'openc3/models/target_model'
-        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
-      rescue => e
-        Logger.warn("Unable to determine log cycle time for #{target_name}: #{e.message}")
+      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

@@ -271,6 +271,7 @@ def read_topics(topics, offsets = nil, timeout_ms = 1000, count = nil)
     # @return [String] the entry id
     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|
         if minid
           return redis.xadd(topic, msg_hash, id: id, minid: minid, approximate: approximate)

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/topics/topic.py

@@ -37,19 +37,30 @@ class Topic(metaclass=TopicMeta):
     # 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).
-    # 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.
+    # 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
-    # Mirrors LogWriter::CLEANUP_DELAY and the TargetModel default log cycle time
+    # 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 = None
     _log_stream_max_age_ms = {}
+    _log_cycle_lookup_retry_at = {}
 
     @classmethod
     def clear_topics(cls, topics, maxlen=0, db_shard=0):
@@ -98,7 +109,19 @@ def stream_max_age_seconds(cls) -> float:
     def stream_max_age_ms(cls) -> int:
         """Cached safety max age in milliseconds (0 means disabled)"""
         if cls._stream_max_age_ms is None:
-            cls._stream_max_age_ms = max(int(cls.stream_max_age_seconds() * 1000), 0)
+            seconds = cls.stream_max_age_seconds()
+            if seconds <= 0:
+                cls._stream_max_age_ms = 0
+            elif seconds < cls.STREAM_MAX_AGE_MIN_SECONDS:
+                from openc3.utilities.logger import Logger
+
+                Logger.warn(
+                    f"OPENC3_STREAM_MAX_AGE_SECONDS of {seconds} is below the minimum, "
+                    f"using {cls.STREAM_MAX_AGE_MIN_SECONDS}"
+                )
+                cls._stream_max_age_ms = cls.STREAM_MAX_AGE_MIN_SECONDS * 1000
+            else:
+                cls._stream_max_age_ms = int(seconds * 1000)
         return cls._stream_max_age_ms
 
     @classmethod
@@ -132,38 +155,62 @@ def log_stream_safety_minid(cls, target_name, cmd_or_tlm, scope):
         key = (scope, target_name, cmd_or_tlm)
         max_age_ms = cls._log_stream_max_age_ms.get(key)
         if max_age_ms is None:
-            max_age_ms = cls.stream_max_age_ms()
-            if max_age_ms > 0:
-                min_age_ms = cls.log_stream_min_age_seconds(target_name, cmd_or_tlm, scope) * 1000
-                max_age_ms = max(max_age_ms, min_age_ms)
+            max_age_ms = cls.log_stream_max_age_ms(target_name, cmd_or_tlm, scope)
+            if max_age_ms is None:
+                return None  # Log cycle time unknown so don't cap yet
             cls._log_stream_max_age_ms[key] = max_age_ms
         return cls.stream_safety_minid(max_age_ms=max_age_ms)
 
     @classmethod
-    def log_stream_min_age_seconds(cls, target_name, cmd_or_tlm, scope):
-        """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.
+    def log_stream_max_age_ms(cls, target_name, cmd_or_tlm, scope):
+        """Max age in milliseconds for a TELEMETRY__ / COMMAND__ stream (0 means disabled)
+        or None 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.
         """
-        cycle_time = cls.DEFAULT_LOG_CYCLE_TIME_SECONDS
+        max_age_ms = cls.stream_max_age_ms()
+        if max_age_ms <= 0:
+            return max_age_ms
+        key = (scope, target_name, cmd_or_tlm)
+        now = time.monotonic()
+        retry_at = cls._log_cycle_lookup_retry_at.get(key)
+        if retry_at is not None and now < retry_at:
+            return None
         try:
-            from openc3.models.target_model import TargetModel
-
-            model = TargetModel.get(name=target_name, scope=scope)
-            if model:
-                field = "cmd_log_cycle_time" if cmd_or_tlm == "CMD" else "tlm_log_cycle_time"
-                value = int(model.get(field) or 0)
-                if value > 0:
-                    cycle_time = value
+            min_age_ms = cls.log_stream_min_age_seconds(target_name, cmd_or_tlm, scope) * 1000
         except Exception as error:
             from openc3.utilities.logger import Logger
 
-            Logger.warn(f"Unable to determine log cycle time for {target_name}: {error}")
+            Logger.warn(
+                f"Unable to determine log cycle time for {target_name} {cmd_or_tlm}, stream age not capped: {error}"
+            )
+            cls._log_cycle_lookup_retry_at[key] = now + cls.LOG_CYCLE_LOOKUP_RETRY_SECONDS
+            return None
+        cls._log_cycle_lookup_retry_at.pop(key, None)
+        return max(max_age_ms, min_age_ms)
+
+    @classmethod
+    def log_stream_min_age_seconds(cls, target_name, cmd_or_tlm, scope):
+        """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.
+        Raises if the TargetModel lookup fails.
+        """
+        from openc3.models.target_model import TargetModel
+
+        cycle_time = cls.DEFAULT_LOG_CYCLE_TIME_SECONDS
+        model = TargetModel.get(name=target_name, scope=scope)
+        if model:
+            field = "cmd_log_cycle_time" if cmd_or_tlm == "CMD" else "tlm_log_cycle_time"
+            value = int(model.get(field) or 0)
+            if value > 0:
+                cycle_time = value
         return 2 * (cycle_time + cls.LOG_CLEANUP_DELAY_SECONDS)
 
     @classmethod
     def clear_stream_max_age_cache(cls):
         cls._stream_max_age_ms = None
         cls._log_stream_max_age_ms = {}
+        cls._log_cycle_lookup_retry_at = {}
 
     @classmethod
     def read_topics(cls, topics, offsets=None, timeout_ms=1000, count=None, db_shard=0):

openc3/python/openc3/utilities/store_implementation.py

@@ -286,6 +286,8 @@ def read_topics(self, topics, offsets=None, timeout_ms=1000, count=None):
     def write_topic(self, topic, msg_hash, id="*", maxlen=None, approximate=True, minid=None):
         if not id:
             id = "*"
+        if maxlen is not None and minid is not None:
+            raise ValueError("write_topic can't supply both maxlen and minid")
         with self.redis_pool.get() as redis:
             if minid:
                 return redis.xadd(topic, msg_hash, id=id, minid=minid, approximate=approximate)