-
{{ replay?.state === 'seeking' ? 'Seeking recorded route…' : replayBuffering ? 'Buffering replay…' : 'Loading replay…' }}
+
{{ replay?.state === 'seeking' ? 'Seeking recorded route…' : replayBuffering ? 'Buffering replay…' : 'Loading replay…' }}
diff --git a/openpilot/starpilot/replay/owner.py b/openpilot/starpilot/replay/owner.py
index 2d8565ef70..7ce18eba1d 100644
--- a/openpilot/starpilot/replay/owner.py
+++ b/openpilot/starpilot/replay/owner.py
@@ -393,6 +393,8 @@ class ReplayOwner:
type(value.get('rate')) not in (int, float) or value.get('rate') not in (.5, 1, 2) or type(value.get('paused')) is not bool or
type(value.get('epoch')) is not int or not 0 <= value['epoch'] < 2**63):
raise Rejected('worker_failed')
+ applied = value.get('lastRequestId')
+ seeking = self.status['state'] == 'seeking' and self.pending is not None and applied != self.pending
self.status.update({key: value[key] for key in ('state', 'position', 'start', 'end', 'rate', 'paused', 'epoch')})
self.status['frameReady'] = value.get('frameReady') is True
self.status['mediaReady'] = value.get('mediaReady') is True
@@ -421,7 +423,9 @@ class ReplayOwner:
raise Rejected('worker_failed')
self.status['media'] = dict(media)
self.status['codec'] = media['codec']
- applied = value.get('lastRequestId')
+ # Queued pre-seek status cannot restore media before the new frame is acknowledged.
+ if seeking and value['state'] != 'failed':
+ self.status.update(state='seeking', frameReady=False, mediaReady=False)
if applied == self.pending and applied is not None:
self.status['lastRequestId'] = applied
self.pending = None
diff --git a/openpilot/starpilot/replay/render.py b/openpilot/starpilot/replay/render.py
index 0270b70833..f4e11abd78 100644
--- a/openpilot/starpilot/replay/render.py
+++ b/openpilot/starpilot/replay/render.py
@@ -626,6 +626,7 @@ def run(session_dir: Path, route_id: str, data_dir: Path, target: str, profile:
last_emit = last_draw = 0.0
published_at = 0.0
published_epoch = 0
+ clock_epoch = 0
from openpilot.starpilot.replay.media import FrameTimeline
timeline = FrameTimeline()
pending_frame = None
@@ -641,6 +642,8 @@ def run(session_dir: Path, route_id: str, data_dir: Path, target: str, profile:
if status:
core.wall_minus_mono_ns = parse_wall_offset(status.get('wallMinusMonoNs'))
sample = core.update()
+ if sample is not None and sample.epoch is not None:
+ clock_epoch = max(clock_epoch, sample.epoch)
now = time.monotonic()
valid = bool(status and sample is not None and sample.valid and sample.now_ns is not None and sample.epoch == status.get('epoch'))
ready_frames = []
@@ -705,11 +708,14 @@ def run(session_dir: Path, route_id: str, data_dir: Path, target: str, profile:
# Keep the frame cadence after delivery jitter, without accumulating a catch-up queue.
last_draw = max(last_draw + interval, now - interval)
if status and now - last_emit >= .1:
- # Readiness describes the validated published frame, not the next camera/message pair arriving separately.
- ready = published_epoch == status.get('epoch') and valid and (status['paused'] or now - published_at <= .5)
+ # A production stall does not invalidate already published media in the same epoch.
+ ready = bool(published_epoch and published_epoch == status.get('epoch') and
+ status['state'] in ('playing', 'paused', 'buffering', 'ended') and
+ clock_epoch == published_epoch)
+ fresh = ready and valid and (status['paused'] or now - published_at <= .5)
value = {key: status.get(key) for key in ('event', 'state', 'position', 'start', 'end', 'rate', 'paused', 'epoch', 'lastRequestId', 'error')}
- if status['state'] not in ('failed', 'ended') and not ready:
- value['state'] = 'buffering' if published_epoch == status.get('epoch') else 'seeking' if published_epoch else 'preparing'
+ if status['state'] not in ('failed', 'ended') and not fresh:
+ value['state'] = 'buffering' if ready else 'seeking' if published_epoch else 'preparing'
if engine.seek_request == value.get('lastRequestId') and not ready:
value['lastRequestId'] = acknowledged
if 'seekRanges' in status:
diff --git a/openpilot/starpilot/replay/tests/test_owner.py b/openpilot/starpilot/replay/tests/test_owner.py
index 9851d9e436..a9785eb35d 100644
--- a/openpilot/starpilot/replay/tests/test_owner.py
+++ b/openpilot/starpilot/replay/tests/test_owner.py
@@ -373,6 +373,28 @@ def test_strict_revision_numbers_and_request_identity(owner):
value.dispatch(dict(control(value, 'pause', 'paused', False, '7' * 32), expectedRevision=accepted['revision'] - 1))
+def test_accepted_seek_cannot_restore_media_from_unacknowledged_worker_status(owner, tmp_path):
+ value, _, _, _ = owner
+ value.client_id = CLIENT
+ value.status.update(sessionId='c' * 32, target='web', state='playing', revision=1, epoch=1, end=120.,
+ frameReady=True, mediaReady=True)
+ published = {**value.status, 'event': 'status', 'error': None, 'lastRequestId': None}
+ command = control(value, 'seek', 'position', 30.)
+ accepted = value.dispatch(command)
+ assert accepted['state'] == 'seeking' and not accepted['mediaReady']
+ for status in (published, {**published, 'epoch': 2, 'state': 'buffering', 'frameReady': False, 'mediaReady': False}):
+ value._observe(status, tmp_path)
+ assert value.status['state'] == 'seeking'
+ assert not value.status['frameReady'] and not value.status['mediaReady']
+ assert value.pending == command['requestId']
+ with pytest.raises(Rejected, match='epoch changed'):
+ value.dispatch({'op': 'media', 'sessionId': accepted['sessionId'], 'clientId': CLIENT,
+ 'epoch': value.status['epoch'], 'sequence': 0})
+ value._observe({**published, 'epoch': 2, 'position': 30., 'lastRequestId': command['requestId']}, tmp_path)
+ assert value.status['state'] == 'playing' and value.status['mediaReady'] and value.status['frameReady']
+ assert value.pending is None
+
+
@pytest.mark.parametrize('ranges', [None, [[0]], [[0, True]], [[0, float('nan')]], [[20, 10]],
[[0, 101]], [[0, 20], [10, 30]]])
def test_worker_seek_ranges_reject_malformed_or_overlapping_bounds(owner, tmp_path, ranges):
diff --git a/openpilot/starpilot/replay/tests/test_render.py b/openpilot/starpilot/replay/tests/test_render.py
index 1bb402988c..8d2afb74f7 100644
--- a/openpilot/starpilot/replay/tests/test_render.py
+++ b/openpilot/starpilot/replay/tests/test_render.py
@@ -176,7 +176,7 @@ def test_engine_starts_before_idle_renderer_placement_and_closes_if_renderer_fai
engine.close.assert_called_once()
-def test_published_readiness_survives_delivery_jitter_but_expires_and_rejects_old_epochs(tmp_path, monkeypatch):
+def test_published_readiness_survives_buffering_but_rejects_seeks_and_old_epochs(tmp_path, monkeypatch):
session = tmp_path / ('a' * 32)
session.mkdir(mode=0o700)
for key in ('STARPILOT_PROJECTION_READ_ONLY', 'BIG', 'SCALE'):
@@ -190,7 +190,7 @@ def test_published_readiness_survives_delivery_jitter_but_expires_and_rejects_ol
monkeypatch.setattr('openpilot.starpilot.replay.catalog.ReplayCatalog', lambda path: catalog)
publisher = Mock()
monkeypatch.setattr(render, 'ReplayFramePublisher', lambda *args: publisher)
- times, step = (1, 1.01, 1.11, 1.61, 10, 10.2, 10.4), [0]
+ times, step = (1, 1.01, 1.11, 1.31, 1.61, 10, 10.2, 10.4, 10.6, 10.8), [0]
monkeypatch.setattr(render.time, 'monotonic', lambda: times[step[0]])
monkeypatch.setattr(render.time, 'monotonic_ns', lambda: round(times[step[0]] * 1e9))
engine = Mock(error=None, status={'state': 'playing', 'paused': False, 'epoch': 1, 'position': 0,
@@ -204,7 +204,8 @@ def test_published_readiness_survives_delivery_jitter_but_expires_and_rejects_ol
core.ui.replay_needs_reprime = False
available = dict.fromkeys(('modelV2', 'extrinsicsCalibration', 'longitudinalPlan'), True)
core.ui.sm = SimpleNamespace(seen=available, valid=available, alive=available)
- core.update.side_effect = lambda: SimpleNamespace(valid=step[0] != 6, epoch=engine.status['epoch'], now_ns=100, boot_ns=100)
+ core.update.side_effect = lambda: None if step[0] == 7 else SimpleNamespace(
+ valid=step[0] not in (3, 6, 8), epoch=2 if step[0] == 6 else engine.status['epoch'], now_ns=100, boot_ns=100)
core.finish.side_effect = lambda: nullcontext((b'pixels', (1, 0, 1_010_000_000)) if step[0] == 2 else None)
core._primary_matches.return_value = core._optional_placeholder_ready.return_value = False
attempts = []
@@ -218,17 +219,20 @@ def test_published_readiness_survives_delivery_jitter_but_expires_and_rejects_ol
engine.error = 'test complete'
return
step[0] += 1
- if step[0] == 4:
- engine.status.update(paused=True, state='paused')
if step[0] == 5:
- engine.status['epoch'] = 2
+ engine.status.update(paused=True, state='paused')
+ if step[0] == 8:
+ engine.status['state'] = 'seeking'
+ if step[0] == 9:
+ engine.status.update(epoch=2, state='paused')
monkeypatch.setattr(render.time, 'sleep', next_loop)
states = []
with pytest.raises(RuntimeError, match='test complete'):
render.run(session, 'recorded-route', tmp_path, 'native', 'large', lambda value: states.append(value))
assert attempts[:2] == [1, 1.01] # A failed draw retries before the 50 ms frame interval.
assert [(value['state'], value['frameReady']) for value in states] == [
- ('preparing', False), ('playing', True), ('buffering', False), ('paused', True), ('seeking', False), ('seeking', False)]
+ ('preparing', False), ('playing', True), ('buffering', True), ('buffering', True), ('paused', True),
+ ('seeking', False), ('seeking', False), ('seeking', False), ('seeking', False)]
publisher.publish.assert_called_once_with(b'pixels', epoch=1, position=0, captured_ns=1_010_000_000)
@@ -532,7 +536,7 @@ def test_web_delivery_times_packets_and_flushes_pause_end_and_stall(tmp_path, mo
core.readback.pending = core.ui.replay_needs_reprime = False
available = dict.fromkeys(('modelV2', 'extrinsicsCalibration', 'longitudinalPlan'), True)
core.ui.sm = SimpleNamespace(seen=available, valid=available, alive=available)
- core.update.side_effect = lambda: SimpleNamespace(valid=True, epoch=engine.status['epoch'], now_ns=100, boot_ns=100)
+ core.update.side_effect = lambda: SimpleNamespace(valid=step[0] != 4, epoch=engine.status['epoch'], now_ns=100, boot_ns=100)
delivered = {0, 1, 2, 5, 6, 7, 9, 11, 12}
core.finish.side_effect = lambda: nullcontext((b'pixels', (engine.status['epoch'], 0, round(times[step[0]] * 1e9)))
if step[0] in delivered else None)
@@ -571,6 +575,9 @@ def test_web_delivery_times_packets_and_flushes_pause_end_and_stall(tmp_path, mo
assert [(call.kwargs['pts_us'], call.kwargs['duration_us']) for call in muxes[1].frame.call_args_list] == [
(0, 50000), (50000, 50000), (100000, 60000), (160000, 50000)]
assert states[0]['mediaReady'] is False
+ buffering = [state for state in states if state['state'] == 'buffering']
+ assert len(buffering) == 2 # An invalid clock and a stale publication do not invalidate existing media.
+ assert all(state['frameReady'] and state['mediaReady'] and state['media'] for state in buffering)
assert any(state['epoch'] == 2 and state['state'] == 'paused' and state['mediaReady'] for state in states)
assert states[-1]['state'] == 'ended' and states[-1]['mediaReady']
for mux in muxes:
diff --git a/openpilot/starpilot/ui/tests/test_replay_settings.py b/openpilot/starpilot/ui/tests/test_replay_settings.py
index bd43ad63de..26b7b438cf 100644
--- a/openpilot/starpilot/ui/tests/test_replay_settings.py
+++ b/openpilot/starpilot/ui/tests/test_replay_settings.py
@@ -226,6 +226,24 @@ def prepared_playback(request):
owner.executor.shutdown(wait=True)
+def test_buffering_preserves_published_native_frame(prepared_playback):
+ owner, backend, frames, _, factory = prepared_playback
+ frame = NS(session_id=owner.own_session, epoch=1, frame_id=1)
+ frames.append(frame)
+ owner.next_frame = 0
+ pump(owner, lambda: owner.frame is frame)
+ generation = owner.reader_generation
+ backend.status.update(state='buffering', paused=False)
+ owner.next_status = 0
+ pump(owner, lambda: owner.status['state'] == 'buffering')
+ assert owner.frame is frame and owner.reader_generation == generation
+ assert factory.call_count == 1
+ backend.status['offroad'] = False
+ owner.next_status = 0
+ pump(owner, lambda: not owner.available)
+ assert owner.frame is None
+
+
def test_autoplay_waits_one_second_after_first_valid_frame_and_runs_only_once(prepared_playback):
owner, backend, frames, clock, factory = prepared_playback
clock.now = 10