Fix: one bad datagram could permanently freeze the video

ios/EternalMonitor/App/ConnectionManager.swift

@@ -327,6 +327,11 @@ final class ConnectionManager: ObservableObject {
                 self?.hostStreamConfig = heartbeat.streamConfig
             }
         }
+        channel.onStreamConfig = { [weak self] config in
+            Task { @MainActor in
+                self?.hostStreamConfig = config
+            }
+        }
 
         receiver.onControlDatagram = { [weak channel] data in
             channel?.handleControl(data)

ios/EternalMonitor/Network/ControlChannel.swift

@@ -24,6 +24,8 @@ final class ControlChannel {
     var onHandshakeTimeout: (() -> Void)?
     var onHelloAttempt: ((Int, Int) -> Void)?
     var onHeartbeat: ((Heartbeat) -> Void)?
+    /// Out-of-band stream-parameter change (bitrate, codec, resolution).
+    var onStreamConfig: ((StreamConfig) -> Void)?
     var onBye: ((ByeReason) -> Void)?
     var onDiagnostic: ((String) -> Void)?
     /// Snapshot provider for periodic receiver reports.
@@ -165,8 +167,13 @@ final class ControlChannel {
             clockSnapshot.withLock {
                 $0 = ClockSnapshot(offsetUs: estimator.offsetUs, rttUs: estimator.rttUs)
             }
-        case .streamConfig:
-            break
+        case .streamConfig(let config):
+            // The host sends this the moment the stream changes (an ABR step,
+            // a codec switch) so the UI doesn't wait up to a heartbeat to
+            // catch up. Dropping it made the host's immediate notify dead
+            // traffic.
+            guard header.sessionId == sessionId, sessionId != 0 else { return }
+            onStreamConfig?(config)
         default:
             break
         }

ios/EternalMonitor/Network/FrameAssembler.swift

@@ -50,6 +50,18 @@ final class FrameAssembler {
     /// dropped forever — the app appears frozen until it's force-quit.
     private static let streamRestartGap: UInt32 = 256
 
+    /// Consecutive stale-epoch drops, and the count after which the epoch we are
+    /// holding is treated as bogus and re-synced to whatever is arriving.
+    ///
+    /// One corrupted or spoofed fragment carrying a high epoch would otherwise
+    /// latch an epoch the host will never reach, and every real fragment would
+    /// be dropped for the rest of the session. Nothing would notice: control
+    /// heartbeats keep flowing, so the liveness watchdog stays happy while the
+    /// video is frozen. Genuine stragglers from a previous run can't trip this,
+    /// because the new run's fragments interleave and reset the streak.
+    private var staleEpochStreak: UInt32 = 0
+    private static let epochResyncThreshold: UInt32 = 512
+
     struct PendingFrame {
         let fragmentCount: UInt16
         let isKeyframe: Bool
@@ -84,7 +96,19 @@ final class FrameAssembler {
                 reset()
                 currentEpoch = epoch
             } else if epoch < current {
-                return
+                staleEpochStreak += 1
+                guard staleEpochStreak >= Self.epochResyncThreshold else { return }
+                // Nothing has been accepted across a long run of drops, so the
+                // epoch we are holding can't be the live one. Re-sync to the
+                // stream that is actually arriving rather than stay frozen.
+                onDiagnostic?(
+                    "Epoch \(current) never resumed after \(staleEpochStreak) dropped fragments"
+                        + " — re-syncing to epoch \(epoch)"
+                )
+                reset()
+                currentEpoch = epoch
+            } else {
+                staleEpochStreak = 0
             }
         } else {
             currentEpoch = epoch
@@ -192,6 +216,7 @@ final class FrameAssembler {
         latestCompletedSeq = 0
         cleanupCounter = 0
         currentEpoch = nil
+        staleEpochStreak = 0
         counters.withLock { $0 = Counters() }
     }
 

ios/EternalMonitorTests/FrameAssemblerTests.swift

@@ -57,6 +57,47 @@ final class FrameAssemblerTests: XCTestCase {
         XCTAssertEqual(completed.count, 2, "stale-epoch fragment must be dropped")
     }
 
+    func testOneBogusEpochCannotStrandTheStream() {
+        add(seq: 10, index: 0, count: 1, epoch: 5, byte: 0xA)
+        XCTAssertEqual(completed.count, 1)
+
+        // One corrupted or spoofed fragment claiming the maximum epoch.
+        add(seq: 11, index: 0, count: 1, epoch: .max, byte: 0xFF)
+        let afterPoison = completed.count
+
+        // The real stream is now "stale" against an epoch the host can never
+        // reach. It must not stay that way for the rest of the session.
+        for i in 0..<UInt32(600) {
+            add(seq: 100 + i, index: 0, count: 1, epoch: 5, byte: 0xB)
+        }
+        XCTAssertGreaterThan(
+            completed.count, afterPoison,
+            "the assembler must re-sync to the live stream instead of freezing forever"
+        )
+
+        // And it keeps flowing afterwards.
+        let beforeTail = completed.count
+        add(seq: 900, index: 0, count: 1, epoch: 5, byte: 0xC)
+        XCTAssertEqual(completed.count, beforeTail + 1)
+        XCTAssertEqual(completed.last, Data([0xC]))
+    }
+
+    func testInterleavedStragglersNeverTripTheResync() {
+        add(seq: 1, index: 0, count: 1, epoch: 7, byte: 0xA)
+
+        // A real restart: epoch 8 is live while epoch-7 stragglers keep
+        // arriving. Far more stale drops than the resync threshold, but the
+        // accepted fragments in between must keep resetting the streak.
+        for i in 0..<UInt32(1200) {
+            add(seq: 1000 + i, index: 0, count: 1, epoch: 8, byte: 0xB)
+            add(seq: 500 + i, index: 0, count: 1, epoch: 7, byte: 0xC)
+        }
+        XCTAssertFalse(
+            completed.contains(Data([0xC])),
+            "old-run stragglers must never be accepted while the new run is live"
+        )
+    }
+
     func testCompletionEvictsOlderPartials() {
         add(seq: 1, index: 0, count: 2, byte: 0xA)
         add(seq: 2, index: 0, count: 1, byte: 0xB)

proto/src/reassembly.rs

@@ -20,6 +20,16 @@ pub const STREAM_RESTART_GAP: u32 = 256;
 /// Partial frames older than this are evicted during periodic cleanup.
 pub const STALE_FRAME_TIMEOUT: Duration = Duration::from_millis(100);
 
+/// After this many CONSECUTIVE stale-epoch drops, the current epoch is treated
+/// as bogus and re-synced to whatever is actually arriving.
+///
+/// Without this, one corrupted or spoofed fragment carrying a high epoch
+/// latches an epoch the sender will never reach, and every real fragment is
+/// dropped forever — video freezes while the control channel stays healthy, so
+/// nothing notices. Genuine stragglers from a previous run can't trip it: the
+/// new run's fragments interleave and are accepted, which resets the streak.
+pub const EPOCH_RESYNC_THRESHOLD: u32 = 512;
+
 /// Cleanup runs every this-many fragments.
 const CLEANUP_INTERVAL: u32 = 100;
 
@@ -72,6 +82,8 @@ pub struct Reassembler {
     latest_completed_seq: u32,
     cleanup_counter: u32,
     current_epoch: Option<u32>,
+    /// Consecutive stale-epoch drops; see [`EPOCH_RESYNC_THRESHOLD`].
+    stale_epoch_streak: u32,
     counters: ReassemblyCounters,
 }
 
@@ -112,12 +124,25 @@ impl Reassembler {
             Some(current) if epoch > current => {
                 self.reset_internal();
                 self.current_epoch = Some(epoch);
+                self.stale_epoch_streak = 0;
             }
             Some(current) if epoch < current => {
-                return AddOutcome::Dropped(DropReason::StaleEpoch);
+                self.stale_epoch_streak += 1;
+                if self.stale_epoch_streak < EPOCH_RESYNC_THRESHOLD {
+                    return AddOutcome::Dropped(DropReason::StaleEpoch);
+                }
+                // Nothing has been accepted for a long run of drops, so the
+                // epoch we are holding can't be the live one. Re-sync to the
+                // stream that is actually arriving.
+                self.reset_internal();
+                self.current_epoch = Some(epoch);
+                self.stale_epoch_streak = 0;
+            }
+            Some(_) => self.stale_epoch_streak = 0,
+            None => {
+                self.current_epoch = Some(epoch);
+                self.stale_epoch_streak = 0;
             }
-            Some(_) => {}
-            None => self.current_epoch = Some(epoch),
         }
 
         if self.latest_completed_seq > 0 {
@@ -216,6 +241,7 @@ impl Reassembler {
     pub fn reset(&mut self) {
         self.reset_internal();
         self.current_epoch = None;
+        self.stale_epoch_streak = 0;
     }
 
     fn reset_internal(&mut self) {
@@ -331,6 +357,59 @@ mod tests {
         );
     }
 
+    #[test]
+    fn one_bogus_epoch_cannot_strand_the_stream() {
+        let now = Instant::now();
+        let mut r = Reassembler::new();
+        feed(&mut r, 10, 0, 1, 5, 0xA, now);
+
+        // One corrupted/spoofed fragment claiming the maximum epoch.
+        feed(&mut r, 11, 0, 1, u32::MAX, 0xFF, now);
+
+        // The real stream is now "stale" against an epoch it can never reach.
+        for i in 0..(EPOCH_RESYNC_THRESHOLD - 1) {
+            assert_eq!(
+                feed(&mut r, 100 + i, 0, 1, 5, 0xB, now),
+                AddOutcome::Dropped(DropReason::StaleEpoch),
+                "fragment {i} should still be dropped before the resync threshold"
+            );
+        }
+
+        // Crossing the threshold re-syncs to the stream that is really there.
+        assert_eq!(
+            feed(&mut r, 900, 0, 1, 5, 0xC, now),
+            AddOutcome::Completed(vec![0xC]),
+            "the receiver must recover instead of staying bricked forever"
+        );
+        // And it keeps flowing afterwards.
+        assert_eq!(
+            feed(&mut r, 901, 0, 1, 5, 0xD, now),
+            AddOutcome::Completed(vec![0xD])
+        );
+    }
+
+    #[test]
+    fn interleaved_stragglers_never_trip_the_resync() {
+        let now = Instant::now();
+        let mut r = Reassembler::new();
+        feed(&mut r, 1, 0, 1, 7, 0xA, now);
+
+        // A real restart: epoch 8 is live, epoch 7 stragglers keep arriving
+        // alongside it. Far more stale drops than the threshold, but the
+        // accepted fragments in between must keep resetting the streak.
+        for i in 0..(EPOCH_RESYNC_THRESHOLD * 2) {
+            assert_eq!(
+                feed(&mut r, 1000 + i, 0, 1, 8, 0xB, now),
+                AddOutcome::Completed(vec![0xB])
+            );
+            assert_eq!(
+                feed(&mut r, 500 + i, 0, 1, 7, 0xC, now),
+                AddOutcome::Dropped(DropReason::StaleEpoch),
+                "old-run straggler {i} must stay dropped"
+            );
+        }
+    }
+
     #[test]
     fn completion_evicts_older_partials_and_counts_them_dropped() {
         let now = Instant::now();