From f9934a6070d2300f912258aff2f19df5bfe039ce Mon Sep 17 00:00:00 2001 From: whoisaldo Date: Thu, 27 Aug 2026 12:25:45 -0400 Subject: [PATCH] Fix: one bad datagram could permanently freeze the video MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A media fragment carrying a stream_epoch higher than the host will ever reach latched that epoch on the receiver. Every real fragment afterwards compared as stale and was dropped, forever — the seq-restart path re-saved the poisoned epoch, and nothing can exceed u32::MAX, so the only escape was a full reconnect. Nothing noticed it happen. Stale-epoch drops return before the loss counters, so the HUD showed 0% loss, and the liveness watchdog keys on control-plane heartbeats, which keep flowing from a perfectly healthy host. The user sees a frozen picture with every indicator green. Reachable from a single spoofed datagram by anyone on the same L2 (the session id rides in cleartext in every media header), or non-adversarially from a middlebox mangling the four epoch bytes of a legitimate packet. Both receivers now re-sync: after 512 CONSECUTIVE stale-epoch drops the held epoch is treated as bogus and the assembler adopts whatever is actually arriving. Genuine stragglers from a previous run can't trip it, because the new run's fragments interleave and are accepted, which resets the streak — the count only climbs when nothing at all is getting through, which is exactly the stuck state. Recovery costs well under a second at streaming rates instead of never. Regression tests on both sides cover the poison case and prove interleaved stragglers stay dropped. Also: the iPad parsed STREAM_CONFIG and threw it away (`case .streamConfig: break`), so the host's immediate notify on an ABR step or codec switch was dead traffic and the Settings readout lagged up to a heartbeat behind. Wired through to the published config. Found by /code-review over the merged revamp; both findings verified against the tree before fixing. --- .../App/ConnectionManager.swift | 5 ++ .../Network/ControlChannel.swift | 11 ++- .../Network/FrameAssembler.swift | 27 +++++- .../FrameAssemblerTests.swift | 41 +++++++++ proto/src/reassembly.rs | 85 ++++++++++++++++++- 5 files changed, 163 insertions(+), 6 deletions(-) diff --git a/ios/EternalMonitor/App/ConnectionManager.swift b/ios/EternalMonitor/App/ConnectionManager.swift index dd22500..9c9c501 100644 --- a/ios/EternalMonitor/App/ConnectionManager.swift +++ b/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) diff --git a/ios/EternalMonitor/Network/ControlChannel.swift b/ios/EternalMonitor/Network/ControlChannel.swift index 8dd0cac..02b7f85 100644 --- a/ios/EternalMonitor/Network/ControlChannel.swift +++ b/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 } diff --git a/ios/EternalMonitor/Network/FrameAssembler.swift b/ios/EternalMonitor/Network/FrameAssembler.swift index dac76a5..465e083 100644 --- a/ios/EternalMonitor/Network/FrameAssembler.swift +++ b/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() } } diff --git a/ios/EternalMonitorTests/FrameAssemblerTests.swift b/ios/EternalMonitorTests/FrameAssemblerTests.swift index b1f96e9..83b2a01 100644 --- a/ios/EternalMonitorTests/FrameAssemblerTests.swift +++ b/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.., + /// 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();