diff --git a/openpilot/starpilot/flm/learning_wire.py b/openpilot/starpilot/flm/learning_wire.py index 7059ae6739..02083193f1 100644 --- a/openpilot/starpilot/flm/learning_wire.py +++ b/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 diff --git a/openpilot/starpilot/flm/tests/test_learning_trace_retention.py b/openpilot/starpilot/flm/tests/test_learning_trace_retention.py new file mode 100644 index 0000000000..0144aa73b9 --- /dev/null +++ b/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 diff --git a/openpilot/system/manager/process_config.py b/openpilot/system/manager/process_config.py index 2b63036360..394e46ab18 100644 --- a/openpilot/system/manager/process_config.py +++ b/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), diff --git a/openpilot/system/manager/test/test_process_recovery.py b/openpilot/system/manager/test/test_process_recovery.py index aec54c0294..ac41e4f033 100644 --- a/openpilot/system/manager/test/test_process_recovery.py +++ b/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)} diff --git a/prebuilt.json b/prebuilt.json index 7d71ca2eec..65777dda67 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "33dcbdddcb1da609d361b8e3ea7cd503dd7e49b944b46f05ff8c9ecede0f9da1", "schema_version": 1} +{"protected_inventory_sha256": "2b69cabc7e00554545e4aac4d826a9f8b3add4ad0eba1fb5abaab576f6fd6185", "schema_version": 1}