Bound FLM replay memory

openpilot/starpilot/flm/learning_wire.py

@@ -4,6 +4,7 @@ from typing import TypedDict, cast
 
 
 from functools import lru_cache
+import struct
 
 from capnp.lib.capnp import _DynamicStructReader
 from openpilot.cereal import custom, messaging
@@ -85,6 +86,22 @@ def _write_effect_trace(trace):
     'appliedFullProfileSha256':trace.applied_full_profile_sha256}
 
 
+# Receiver histories retain up to 60,000 frames. Share exact immutable map
+# content instead of allocating five full30 maps for every unchanged frame.
+# Source timestamps and the per-frame transition are still validated below.
+@lru_cache(maxsize=128)
+def _effect_coordinate_rows(packed):
+  from openpilot.starpilot.flm.canonical_surface_adapter import NAMES
+  return tuple(zip(NAMES, struct.unpack('!30d', packed), strict=True))
+
+
+def _shared_effect_rows(rows):
+  from openpilot.starpilot.flm.canonical_surface_adapter import NAMES
+  if tuple(name for name, _ in rows) != NAMES:
+    return rows  # Preserve the original validator's rejection of foreign names.
+  return _effect_coordinate_rows(struct.pack('!30d', *(value for _, value in rows)))
+
+
 def _read_effect_trace(row):
   from openpilot.starpilot.flm.effect_trace import FullEffectTrace
   if row.version == 0:
@@ -94,7 +111,8 @@ def _read_effect_trace(row):
   envelope=_read_full_surface(row.envelope)
   if envelope is None:
     raise ValueError('effect trace original envelope')
-  values=[tuple((str(v.name),float(v.value)) for v in getattr(row,n)) for n in ('before','target','applied')]
+  values=[_shared_effect_rows(tuple((str(v.name),float(v.value)) for v in getattr(row,n)))
+          for n in ('before','target','applied')]
   return FullEffectTrace(1,envelope,int(row.controlMonoTime),int(row.previousMonoTime),bool(row.qualified),
     *cast(tuple[tuple[tuple[str, float], ...], tuple[tuple[str, float], ...], tuple[tuple[str, float], ...]], values),
     str(row.appliedFullProfileSha256)).validate()
@@ -303,6 +321,7 @@ def _write_full_surface(profile):
 
 def _read_full_surface(row):
   from openpilot.starpilot.flm.full_surface_profile import FullSurfaceEnvelope
+  from openpilot.starpilot.flm.effect_admission_cache import validated_envelope
   try:
     if row.version not in (1, 2) or len(row.priorCoordinates) != 30 or len(row.coordinates) != 30 or not 1 <= len(row.trialNames) <= 4:
       return None
@@ -310,7 +329,7 @@ def _read_full_surface(row):
     values.update(version=int(row.version), prior_coordinates=tuple((str(point.name), float(point.value)) for point in row.priorCoordinates),
                   coordinates=tuple((str(point.name), float(point.value)) for point in row.coordinates),
                   trial_names=tuple(str(name) for name in row.trialNames))
-    return FullSurfaceEnvelope(**cast(_FullSurfaceFields, values)).validate()
+    return validated_envelope(FullSurfaceEnvelope(**cast(_FullSurfaceFields, values)))
   except (AttributeError, KeyError, TypeError, ValueError, OverflowError):
     return None
 

openpilot/starpilot/flm/tests/test_learning_trace_retention.py

@@ -0,0 +1,110 @@
+"""Real Cereal trace retention shares immutable content, never per-frame clocks."""
+from dataclasses import replace
+import struct
+
+import pytest
+
+from openpilot.cereal import custom
+from openpilot.starpilot.flm import learning_wire as wire
+from openpilot.starpilot.flm.canonical_surface_adapter import NAMES, full_profile_sha256
+from openpilot.starpilot.flm.effect_trace import FullEffectTrace
+from openpilot.starpilot.flm.full_surface_profile import FullSurfaceEnvelope
+
+
+def trace(stamp=1_000_000_000, epoch='e' * 64):
+  envelope = FullSurfaceEnvelope.initial(cp_sha256='c' * 64,
+    controller_revision='selected-controller-v2:' + 'd' * 64, source_epoch=epoch,
+    reset_generation='0', selected_baseline_revision='a' * 64,
+    trial_names=('ff_gain_left',), persistent_source_epoch='9' * 64)
+  rows = envelope.coordinates
+  return FullEffectTrace(1, envelope, stamp, stamp - 10_000_000, True,
+                         rows, rows, rows, envelope.full_profile_sha256).validate()
+
+
+def decode(value):
+  payload = custom.LearningControlEvidence.new_message(effectTrace=wire._write_effect_trace(value)).to_bytes()
+  with custom.LearningControlEvidence.from_bytes(payload) as message:
+    return wire._read_effect_trace(message.effectTrace)
+
+
+def test_actual_cereal_retains_shared_content_and_distinct_original_clocks():
+  retained = [decode(trace(1_000_000_000 + i * 10_000_000)) for i in range(100)]
+  first, last = retained[0], retained[-1]
+  assert first is not last
+  assert first.envelope is last.envelope
+  for field in ('before', 'target', 'applied'):
+    assert getattr(first, field) is getattr(last, field)
+    assert tuple(name for name, _ in getattr(first, field)) == NAMES
+  assert [row.control_ns for row in retained] == [1_000_000_000 + i * 10_000_000 for i in range(100)]
+  assert all(row.previous_ns == row.control_ns - 10_000_000 for row in retained)
+  assert all(row.validate() == trace(row.control_ns) for row in retained)
+
+
+def test_actual_cereal_source_identity_never_aliases_a_different_envelope():
+  first, other = decode(trace()), decode(trace(epoch='f' * 64))
+  assert first.envelope is not other.envelope
+  assert first.envelope.source_epoch == 'e' * 64
+  assert other.envelope.source_epoch == 'f' * 64
+  assert first.applied is other.applied
+
+
+def test_actual_cereal_changed_coordinate_is_preserved():
+  base = trace()
+  rows = tuple((name, .001 if name == 'ff_gain_left' else value) for name, value in base.before)
+  # Withdrawing a nonzero prior effect must still produce the exact zero command.
+  changed = replace(base, before=rows, qualified=False).validate()
+  decoded = decode(changed)
+  assert decoded == changed
+  assert decoded.before is not decode(base).before
+  assert dict(decoded.before)['ff_gain_left'] == .001
+
+
+@pytest.mark.parametrize('change', ['version', 'names', 'bounds', 'nan', 'hash', 'context', 'manifest'])
+def test_actual_cereal_invalid_trace_is_still_rejected_after_cache_warmup(change):
+  base = trace()
+  decode(base)
+  data = wire._write_effect_trace(base)
+  if change == 'version':
+    data['version'] = 2
+  elif change == 'names':
+    data['before'][0]['name'] = 'foreign'
+  elif change == 'bounds':
+    data['before'].pop()
+  elif change == 'nan':
+    data['before'][0]['value'] = float('nan')
+  elif change == 'hash':
+    data['appliedFullProfileSha256'] = '0' * 64
+  elif change == 'context':
+    data['envelope']['contextKey'] = 'f' * 64
+  else:
+    data['envelope']['manifestSha256'] = '0' * 64
+  payload = custom.LearningControlEvidence.new_message(effectTrace=data).to_bytes()
+  with custom.LearningControlEvidence.from_bytes(payload) as message, pytest.raises(ValueError):
+    wire._read_effect_trace(message.effectTrace)
+
+
+def test_coordinate_cache_preserves_signed_zero_and_has_fixed_bound():
+  wire._effect_coordinate_rows.cache_clear()
+  positive = tuple((name, 0.) for name in NAMES)
+  negative = tuple((name, -0.) for name in NAMES)
+  first = wire._shared_effect_rows(positive)
+  second = wire._shared_effect_rows(negative)
+  assert first is not second
+  assert struct.pack('!d', second[0][1]) == struct.pack('!d', -0.)
+  for i in range(160):
+    rows = tuple((name, float(i)) for name in NAMES)
+    assert wire._shared_effect_rows(rows) == rows
+  assert wire._effect_coordinate_rows.cache_info().currsize == 128
+  assert wire._shared_effect_rows(positive) == positive
+  assert wire._shared_effect_rows(positive) is not first
+
+
+def test_actual_cereal_signed_zero_hash_is_not_normalized():
+  base = trace()
+  negative = tuple((name, -0.) for name in NAMES)
+  # Even numerically equal zero maps have a different canonical hash.
+  assert full_profile_sha256('a' * 64, negative) != base.applied_full_profile_sha256
+  changed = replace(base, applied=negative, applied_full_profile_sha256=full_profile_sha256('a' * 64, negative)).validate()
+  decoded = decode(changed)
+  assert struct.pack('!d', decoded.applied[0][1]) == struct.pack('!d', -0.)
+  assert decoded.applied_full_profile_sha256 == changed.applied_full_profile_sha256

openpilot/system/manager/process_config.py

@@ -161,7 +161,7 @@ procs = [
   NativeProcess("_pandad", "openpilot/selfdrive/pandad", ["./pandad"], always_run, enabled=False),
   PythonProcess("calibrationd", "openpilot.selfdrive.locationd.calibrationd", only_onroad),
   PythonProcess("torqued", "openpilot.selfdrive.locationd.torqued", only_onroad),
-  PythonProcess("steerlearnerd", "openpilot.starpilot.flm.steerlearnerd", only_onroad),
+  PythonProcess("steerlearnerd", "openpilot.starpilot.flm.steerlearnerd", only_onroad, restart_on_exit=True),
   PythonProcess("controlsd", "openpilot.selfdrive.controls.controlsd", and_(not_joystick, iscar)),
   PythonProcess("joystickd", "openpilot.tools.joystick.joystickd", or_(joystick, notcar)),
   PythonProcess("selfdrived", "openpilot.selfdrive.selfdrived.selfdrived", only_onroad),

openpilot/system/manager/test/test_process_recovery.py

@@ -119,6 +119,30 @@ class TestProcessRecovery(unittest.TestCase):
     self.assertIsNot(p.proc, old)
     self.assertTrue(old.joined)
 
+  def test_learner_recovers_with_backoff_only_while_onroad(self):
+    from openpilot.system.manager.process_config import managed_processes
+    configured = managed_processes["steerlearnerd"]
+    self.assertTrue(configured.restart_on_exit)
+    p = process.PythonProcess(configured.name, configured.module, configured.should_run,
+                              restart_on_exit=configured.restart_on_exit)
+    self.step(p)
+    old = p.proc
+    self.assertIsNotNone(old)
+    old.exitcode = 1
+    self.step(p)
+    self.assertIs(p.proc, old)
+    self.now += 1.01
+    self.step(p)
+    self.assertIsNot(p.proc, old)
+    self.assertTrue(old.joined)
+    p.proc.exitcode = 1
+    self.step(p)
+    self.step(p, started=False)
+    self.now += 60
+    self.step(p, started=False)
+    self.assertIsNone(p.proc)
+    self.assertEqual(p.restart_at, 0)
+
   def test_boot_racing_processes_restart(self):
     from openpilot.system.manager.process_config import procs
     restarting = {p.name for p in procs if getattr(p, "restart_on_exit", False)}

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "33dcbdddcb1da609d361b8e3ea7cd503dd7e49b944b46f05ff8c9ecede0f9da1", "schema_version": 1}
+{"protected_inventory_sha256": "2b69cabc7e00554545e4aac4d826a9f8b3add4ad0eba1fb5abaab576f6fd6185", "schema_version": 1}