Replay -> Galaxy Optimization

2026-11-14 00:13:42 +08:00 · 2026-10-14 22:10:26 -04:00
parent 302eeba70d
commit 9ba3e3667a
8 changed files with 101 additions and 15 deletions
@@ -137,6 +137,25 @@ assert.equal(controlClient.source, null, 'accepted seek removes the old video be
 controlClient.close(false)
 console.log('Replay: pending control acknowledgement and immediate seek media retirement passed')

+const bufferedVideo = { currentTime: 12.5, paused: false, pause() { this.paused = true },
+  removeAttribute() {}, load() { this.currentTime = 0 } }
+const bufferedClient = new ReplayClient({ clientId, video: () => bufferedVideo, unauthorized() {}, publish() {},
+  later: () => 1, cancelTimer() {}, revokeURL() {} })
+bufferedClient.active = true; bufferedClient.status = status; bufferedClient.ownedSession = sessionId
+bufferedClient.source = new Source()
+bufferedClient.buffer = { buffered: { length: 1, start: () => 12, end: () => 13.3 } }
+const bufferedSource = bufferedClient.source, bufferedGeneration = bufferedClient.mediaGeneration
+bufferedClient.receive({ ...status, revision: 4, state: 'buffering', frameReady: true, mediaReady: true })
+assert.equal(bufferedClient.source, bufferedSource, 'production buffering preserves valid buffered media')
+assert.equal(bufferedClient.mediaGeneration, bufferedGeneration)
+assert.equal(bufferedVideo.currentTime, 12.5)
+assert.equal(bufferedVideo.paused, false)
+bufferedClient.receive({ ...status, revision: 5, state: 'seeking', frameReady: false, mediaReady: false })
+assert.equal(bufferedClient.source, null, 'an accepted same-epoch seek still retires buffered media')
+assert.equal(bufferedVideo.currentTime, 0)
+bufferedClient.close(false)
+console.log('Replay: production buffering preserves playback and an accepted seek still invalidates it passed')
+
 controlClient.active = true; controlClient.startAttempted = true; controlClient.routeId = 'test-route'; controlClient.ownedSession = null
 controlClient.status = { ...status, revision: 5 }
 controlClient.receive({ ...status, revision: 6, routeId: 'test-route', owned: true })
@@ -33,6 +33,7 @@ class ReplayHTTPTest(unittest.TestCase):
    self.path_override = None
    self.evicted = False
    self.media_ready = True
+    self.replay_state = 'playing'

    def replay(payload):
      self.calls.append(payload)
@@ -52,7 +53,7 @@ class ReplayHTTPTest(unittest.TestCase):
        result = rows[0] if sequence == 0 or payload.get('count', 1) == 1 else {'fragments': rows}
        return {'ok': True, 'result': result}
      return {'ok': True, 'result': {'schemaVersion': 1, 'ownerId': OWNER, 'sessionId': SESSION,
-                                    'state': 'playing' if self.media_ready else 'seeking', 'mediaReady': self.media_ready,
+                                    'state': self.replay_state if self.media_ready else 'seeking', 'mediaReady': self.media_ready,
                                    'epoch': 1, 'offroad': True, 'owned': True, 'framePath': '/private/frame'}}

    self.server = make_server(port=0, owner=self.access, parked=lambda: True, replay_client=replay, replay_root=self.root)
@@ -137,6 +138,15 @@ class ReplayHTTPTest(unittest.TestCase):
    self.assertEqual(status, 503)
    self.assertNotIn(str(self.path_override).encode(), body)

+  def test_buffering_keeps_published_media_and_waits_for_missing_fragment(self):
+    self.replay_state = 'buffering'
+    cookie = self.login()
+    path = f'/api/replay/media?sessionId={SESSION}&clientId={CLIENT}&epoch=1&sequence='
+    status, body, _ = self.request(path + '1', cookie=cookie)
+    self.assertEqual(status, 200)
+    self.assertEqual(body, b''.join(f'fragment-{number}'.encode() for number in range(1, 9)))
+    self.assertEqual(self.request(path + '10', cookie=cookie)[0], 202)
+
  def test_logout_during_slow_media_withholds_bytes(self):
    cookie = self.login()
    self.release.clear()
@@ -395,7 +395,7 @@ export const ReplayViewer = {
    <div class="gx-recordings__player-head"><div><h3 id="gx-replay-title">Route replay</h3><p class="gx-note">{{ routeId }}<template v-if="startSegment !== null"> · from segment {{ startSegment }}</template> · Recorded driving view · Vehicle offroad only</p></div><GxIconButton label="Close replay" icon="bi-x-lg" @click="close" /></div>
    <div ref="player" class="gx-replay__player">
    <video ref="video" class="gx-replay__video" playsinline muted aria-label="Recorded route with driving interface" @waiting="client?.videoWaiting()" @error="client?.videoError()"></video>
-    <GxState class="gx-replay__loading" v-if="!replayError && (!replay || replayBuffering || !replayVideoReady && (playing || replayBusy || !replayDetail) || ['preparing', 'seeking', 'buffering'].includes(replay.state))" loading>{{ replay?.state === 'seeking' ? 'Seeking recorded route…' : replayBuffering ? 'Buffering replay…' : 'Loading replay…' }}</GxState>
+    <GxState class="gx-replay__loading" v-if="!replayError && (!replay || replayBuffering || !replayVideoReady && (playing || replayBusy || !replayDetail) || ['preparing', 'seeking'].includes(replay.state))" loading>{{ replay?.state === 'seeking' ? 'Seeking recorded route…' : replayBuffering ? 'Buffering replay…' : 'Loading replay…' }}</GxState>
    <GxIconButton v-if="playing" class="gx-replay__fullscreen" label="Fullscreen video" icon="bi-fullscreen" @click="fullscreen" />
    <div v-if="playing" class="gx-replay__media-controls">
      <label class="gx-replay__timeline"><span>{{ replayTime((seekPosition ?? replay.position)-replay.start) }} / {{ replayTime(replay.end-replay.start) }}</span><input type="range" aria-label="Replay route position" :min="replay.start" :max="replay.end" step="0.05" :value="seekPosition ?? replay.position" :disabled="controlsBusy" @input="seekPosition=Number($event.target.value)" @change="seek($event.target.value)" /></label>
@@ -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
@@ -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:
@@ -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):
@@ -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:
@@ -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