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();