diff --git a/compose.yaml b/compose.yaml index 6a841240d3..38d4f7e682 100644 --- a/compose.yaml +++ b/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 diff --git a/docs.openc3.com/docs/configuration/environment.md b/docs.openc3.com/docs/configuration/environment.md index 97c1f31ed9..4b2a6ed0bd 100644 --- a/docs.openc3.com/docs/configuration/environment.md +++ b/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. diff --git a/openc3/lib/openc3/logs/log_writer.rb b/openc3/lib/openc3/logs/log_writer.rb index 2a1ac32349..54a85f72d0 100644 --- a/openc3/lib/openc3/logs/log_writer.rb +++ b/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 diff --git a/openc3/lib/openc3/topics/topic.rb b/openc3/lib/openc3/topics/topic.rb index 7e395e7fce..99dd5edf21 100644 --- a/openc3/lib/openc3/topics/topic.rb +++ b/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,33 +131,51 @@ 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 @@ -144,6 +183,7 @@ def self.log_stream_min_age_seconds(target_name, cmd_or_tlm, scope:) 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) diff --git a/openc3/lib/openc3/utilities/store_autoload.rb b/openc3/lib/openc3/utilities/store_autoload.rb index 819f1186cb..0750bd3e36 100644 --- a/openc3/lib/openc3/utilities/store_autoload.rb +++ b/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) diff --git a/openc3/python/openc3/logs/log_writer.py b/openc3/python/openc3/logs/log_writer.py index 55385ffa88..e7bb50c7b9 100644 --- a/openc3/python/openc3/logs/log_writer.py +++ b/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() diff --git a/openc3/python/openc3/topics/topic.py b/openc3/python/openc3/topics/topic.py index 33661c53d0..7bd7b26e90 100644 --- a/openc3/python/openc3/topics/topic.py +++ b/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): diff --git a/openc3/python/openc3/utilities/store_implementation.py b/openc3/python/openc3/utilities/store_implementation.py index 6679033b50..36fd851ca4 100644 --- a/openc3/python/openc3/utilities/store_implementation.py +++ b/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) diff --git a/openc3/python/test/topics/test_topic.py b/openc3/python/test/topics/test_topic.py index 558b13e10c..a6a73f99d7 100644 --- a/openc3/python/test/topics/test_topic.py +++ b/openc3/python/test/topics/test_topic.py @@ -16,6 +16,7 @@ from openc3.models.target_model import TargetModel from openc3.topics.telemetry_topic import TelemetryTopic from openc3.topics.topic import Topic +from openc3.utilities.logger import Logger from test.test_helper import mock_redis @@ -43,6 +44,14 @@ def test_max_age_ms_is_cached(self): Topic.clear_stream_max_age_cache() self.assertEqual(Topic.stream_max_age_ms(), 120000) + def test_max_age_raised_to_minimum(self): + with ( + patch.dict(os.environ, {"OPENC3_STREAM_MAX_AGE_SECONDS": "0.5"}), + patch.object(Logger.instance(), "warn") as warn, + ): + self.assertEqual(Topic.stream_max_age_ms(), 120000) + self.assertIn("below the minimum", warn.call_args.args[0]) + @patch("openc3.topics.topic.time.time", return_value=1700000000.0) def test_minid_relative_to_now_for_auto_ids(self, _time): self.assertEqual(Topic.stream_safety_minid(), "1699999400000") @@ -95,6 +104,38 @@ def test_log_stream_minid_is_cached_per_stream(self): Topic.log_stream_safety_minid("INST", "CMD", "DEFAULT") self.assertEqual(get.call_count, 2) + def test_log_stream_min_age_raises_on_lookup_failure(self): + with ( + patch.object(TargetModel, "get", side_effect=RuntimeError("down")), + self.assertRaisesRegex(RuntimeError, "down"), + ): + Topic.log_stream_min_age_seconds("INST", "TLM", "DEFAULT") + + @patch("openc3.topics.topic.time.time", return_value=1700000000.0) + def test_log_stream_minid_not_cached_on_lookup_failure(self, _time): + model = {"tlm_log_cycle_time": 3600} + expected = str(1700000000000 - 7320000) + with ( + patch("openc3.topics.topic.time.monotonic", return_value=1000.0) as monotonic, + patch.object(Logger.instance(), "warn") as warn, + patch.object(TargetModel, "get", side_effect=[RuntimeError("down"), model]) as get, + ): + self.assertIsNone(Topic.log_stream_safety_minid("INST", "TLM", "DEFAULT")) + self.assertIn("Unable to determine log cycle time for INST TLM", warn.call_args.args[0]) + # Not retried until the retry interval passes + monotonic.return_value = 1000.0 + Topic.LOG_CYCLE_LOOKUP_RETRY_SECONDS - 1 + self.assertIsNone(Topic.log_stream_safety_minid("INST", "TLM", "DEFAULT")) + self.assertEqual(get.call_count, 1) + monotonic.return_value = 1000.0 + Topic.LOG_CYCLE_LOOKUP_RETRY_SECONDS + self.assertEqual(Topic.log_stream_safety_minid("INST", "TLM", "DEFAULT"), expected) + # Now cached + self.assertEqual(Topic.log_stream_safety_minid("INST", "TLM", "DEFAULT"), expected) + self.assertEqual(get.call_count, 2) + + def test_write_topic_rejects_maxlen_and_minid(self): + with self.assertRaisesRegex(ValueError, "both maxlen and minid"): + Topic.write_topic("TOPIC", {"a": 1}, "*", 10, minid="123") + def test_write_topic_trims_old_entries(self): Topic.write_topic("TOPIC", {"a": 1}, "1000-0") Topic.write_topic("TOPIC", {"a": 2}, "2000-0") diff --git a/openc3/spec/topics/topic_spec.rb b/openc3/spec/topics/topic_spec.rb index 9387a061a6..625def9fc9 100644 --- a/openc3/spec/topics/topic_spec.rb +++ b/openc3/spec/topics/topic_spec.rb @@ -52,6 +52,12 @@ module OpenC3 expect(Topic.stream_max_age_ms).to eql 120_000 end + it "raises small max ages to the minimum" do + ENV['OPENC3_STREAM_MAX_AGE_SECONDS'] = '0.5' + expect(Logger).to receive(:warn).with(/below the minimum/) + expect(Topic.stream_max_age_ms).to eql 120_000 + end + it "caps relative to now for auto generated ids" do allow(Process).to receive(:clock_gettime).and_call_original allow(Process).to receive(:clock_gettime).with(Process::CLOCK_REALTIME, :millisecond).and_return(1700000000000) @@ -95,6 +101,10 @@ module OpenC3 expect(Topic.log_stream_min_age_seconds('INST', :CMD, scope: 'DEFAULT')).to eql 2520 end + it "raises if the target lookup fails" do + allow(TargetModel).to receive(:get).and_raise(RuntimeError, 'down') + expect { Topic.log_stream_min_age_seconds('INST', :TLM, scope: 'DEFAULT') }.to raise_error(RuntimeError, 'down') + end end describe "self.log_stream_safety_minid" do @@ -119,6 +129,29 @@ module OpenC3 Topic.log_stream_safety_minid('INST', :CMD, scope: 'DEFAULT') Topic.log_stream_safety_minid('INST', :CMD, scope: 'DEFAULT') end + + it "doesn't cap or cache when the log cycle lookup fails and retries later" do + monotonic = 1000.0 + allow(Process).to receive(:clock_gettime).with(Process::CLOCK_MONOTONIC) { monotonic } + model = { 'tlm_log_cycle_time' => 3600 } + calls = 0 + allow(TargetModel).to receive(:get) do + calls += 1 + raise 'down' if calls == 1 + model + end + expect(Logger).to receive(:warn).with(/Unable to determine log cycle time for INST TLM/).once + expect(Topic.log_stream_safety_minid('INST', :TLM, scope: 'DEFAULT')).to be_nil + # Not retried until the retry interval passes + monotonic += Topic::LOG_CYCLE_LOOKUP_RETRY_SECONDS - 1 + expect(Topic.log_stream_safety_minid('INST', :TLM, scope: 'DEFAULT')).to be_nil + expect(calls).to eql 1 + monotonic += 1 + expect(Topic.log_stream_safety_minid('INST', :TLM, scope: 'DEFAULT')).to eql (1700000000000 - 7_320_000).to_s + # Now cached + expect(Topic.log_stream_safety_minid('INST', :TLM, scope: 'DEFAULT')).to eql (1700000000000 - 7_320_000).to_s + expect(calls).to eql 2 + end end describe "self.write_topic" do @@ -131,6 +164,10 @@ module OpenC3 expect_any_instance_of(MockRedis).to receive(:xadd).with('TOPIC', { 'a' => 1 }, id: '*', maxlen: 10, approximate: 'true').and_return('1-0') expect(Topic.write_topic('TOPIC', { 'a' => 1 }, '*', 10)).to eql '1-0' end + + it "raises if both maxlen and minid are given" do + expect { Topic.write_topic('TOPIC', { 'a' => 1 }, '*', 10, minid: '123') }.to raise_error(ArgumentError, /both maxlen and minid/) + end end describe "TelemetryTopic.write_packet" do