From 0937d6b92c2acc6b81cb0d8549e2d498000fa429 Mon Sep 17 00:00:00 2001 From: whoisaldo Date: Thu, 27 Aug 2026 13:31:02 -0400 Subject: [PATCH] Close the review's protocol-parity and input-correctness findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Rust reassembler is a hand-written mirror of the Swift one and names that file as its specification, but it had drifted in three ways that all made the host-side tests validate a receiver nobody runs. A fragment whose count disagreed with the live frame's rebuilt that frame, discarding everything already buffered for it; Swift keeps the first-seen count and ignores the odd fragment, so one late duplicate could throw away nine tenths of a keyframe. A replayed fragment overwrote good bytes and counted as another receipt. And frags_lost counted the fragments that DID arrive rather than the ones that did not, so a frame missing one of ten reported nine lost — the end-to-end test fed the ABR controller roughly nine times the real loss and passed, while a real iPad barely moved it. All three now match Swift, with tests for each; the test that locked in the rebuild behavior is replaced by one that proves progress survives. Control messages were authorized by source IP alone: the parsed header, which carries the session id, was discarded. An IP is shared by every process on the device and by everything behind the same NAT, and is trivially spoofed on the local link, so a stray INPUT_EVENT clicked the mouse, a stray RECEIVER_REPORT steered the bitrate, and a late BYE from a superseded session tore down the one that replaced it. Every message except HELLO2 now has to carry the id the handshake minted. The client already did this in the other direction. The client's BYE never went out. sendBye enqueued three copies on the control queue and the caller cancelled the socket on the next line, so even the zero-delay copy was still waiting its turn. The host saw nothing, streamed at a dead peer for the full liveness timeout, held the virtual display up for the same three seconds, and answered an immediate reconnect with "busy". The first copy is now sent synchronously. A drag that ended on a letterbox bar released at the middle of the desktop, dropping whatever was being dragged there, because an out-of-video release has no coordinate and fell back to screen centre. It now releases at the last point that was inside the video. Finally, the input deduper remembered exactly one edge id, but the client sends each press and release twice on purpose and Wi-Fi reorders freely: began(N), ended(N+1), began(N) walked the single slot and let the replayed press through as a second click. It remembers the last eight. --- host/src/input/mod.rs | 44 +++++- host/src/transport/mod.rs | 8 +- host/src/transport/session.rs | 139 ++++++++++++++---- ios/EternalMonitor/Input/TouchRelay.swift | 22 ++- .../Network/ControlChannel.swift | 23 ++- ios/EternalMonitorTests/TouchRelayTests.swift | 15 ++ proto/src/reassembly.rs | 122 +++++++++++---- 7 files changed, 297 insertions(+), 76 deletions(-) diff --git a/host/src/input/mod.rs b/host/src/input/mod.rs index 3cc8a53..2ce76bf 100644 --- a/host/src/input/mod.rs +++ b/host/src/input/mod.rs @@ -110,11 +110,19 @@ pub fn scroll_to_wheel(delta_px: i16) -> i32 { i32::from(delta_px) * 3 } -/// Drops duplicate began/ended events (the client sends edges twice for loss -/// tolerance) while letting distinct events through in order. +/// Number of recent edge ids remembered. The client sends each press/release +/// twice for loss tolerance, and Wi-Fi aggregation reorders freely, so the two +/// copies of one edge routinely arrive with other edges between them. A single +/// remembered id could not survive that: began(N), ended(N+1), began(N) walked +/// the slot N -> N+1 -> N and let the replayed press through, injecting a +/// double click where the user tapped once. +const RECENT_EDGE_IDS: usize = 8; + +/// Drops duplicate began/ended events while letting distinct events through. #[derive(Debug, Default)] pub struct EventDeduper { - last_edge_id: Option, + recent: [Option; RECENT_EDGE_IDS], + next: usize, } impl EventDeduper { @@ -122,12 +130,12 @@ impl EventDeduper { pub fn accept(&mut self, event: &InputEvent) -> bool { match event.phase { PHASE_BEGAN | PHASE_ENDED | PHASE_CANCELLED => { - if self.last_edge_id == Some(event.event_id) { - false - } else { - self.last_edge_id = Some(event.event_id); - true + if self.recent.contains(&Some(event.event_id)) { + return false; } + self.recent[self.next] = Some(event.event_id); + self.next = (self.next + 1) % RECENT_EDGE_IDS; + true } // Moves are idempotent-ish; duplicates are harmless. _ => true, @@ -388,6 +396,26 @@ mod tests { assert_eq!(recorder::take().len(), 4); } + #[test] + fn reordered_duplicate_edges_still_dedupe() { + // The wire sends each edge twice; the network may interleave them. + // began(7), ended(8), began(7), ended(8) must inject one press and + // one release, not two of each. + let mut dedupe = EventDeduper::default(); + let mut began = event(KIND_TOUCH, PHASE_BEGAN, 0, 0); + began.event_id = 7; + let mut ended = event(KIND_TOUCH, PHASE_ENDED, 0, 0); + ended.event_id = 8; + + assert!(dedupe.accept(&began)); + assert!(dedupe.accept(&ended)); + assert!( + !dedupe.accept(&began), + "the replayed press must not click again" + ); + assert!(!dedupe.accept(&ended)); + } + #[test] fn edge_duplicates_are_dropped_moves_pass() { let mut dedupe = EventDeduper::default(); diff --git a/host/src/transport/mod.rs b/host/src/transport/mod.rs index 66cf0d9..0f71957 100644 --- a/host/src/transport/mod.rs +++ b/host/src/transport/mod.rs @@ -150,9 +150,13 @@ pub async fn start_sender( match classify(datagram) { Classified::Control(_) => { match eternal_wire::v2::control::parse_control(datagram) { - Ok((_, message)) => { + Ok((header, message)) => { let mut actions = shared.session.lock().handle_control( - src, message, &config, Instant::now(), + src, + header.session_id, + message, + &config, + Instant::now(), ); if let Some((session_id, event)) = actions.input.take() { if let Some(geometry) = *shared.capture_geometry.lock() diff --git a/host/src/transport/session.rs b/host/src/transport/session.rs index e486d39..c563722 100644 --- a/host/src/transport/session.rs +++ b/host/src/transport/session.rs @@ -132,14 +132,31 @@ impl Session { } } - /// Feed one inbound control message. `now` is injected for testability. + /// Feed one inbound control message. `header_session_id` is the id the + /// datagram claims; every message except HELLO2 must match the live + /// session's. `now` is injected for testability. pub fn handle_control( &mut self, source: SocketAddr, + header_session_id: u32, message: ControlMessage, config: &impl ConfigSource, now: Instant, ) -> Actions { + // HELLO2 is the only message that legitimately carries no session id + // (it is asking for one). Everything else is authenticated by the id + // the handshake minted, not just by source IP: an IP is shared by + // every process on the device and by anything behind the same NAT, and + // is trivially spoofed on the local link. Without this, a stray + // INPUT_EVENT moved the mouse and clicked, a stray RECEIVER_REPORT + // steered the bitrate, and a late BYE from a superseded session tore + // down the one that replaced it. + if !matches!(message, ControlMessage::Hello2(_)) { + match self.active.as_ref() { + Some(session) if session.session_id == header_session_id => {} + _ => return Actions::default(), + } + } match message { ControlMessage::Hello2(hello) => self.handle_hello(source, hello, config, now), ControlMessage::Bye(reason) => self.handle_bye(source, reason), @@ -152,6 +169,19 @@ impl Session { } } + /// Test helper: send as the currently-connected client. + #[cfg(test)] + fn handle_control_authed( + &mut self, + source: SocketAddr, + message: ControlMessage, + config: &impl ConfigSource, + now: Instant, + ) -> Actions { + let id = self.session_id().unwrap_or(0); + self.handle_control(source, id, message, config, now) + } + fn handle_input(&mut self, source: SocketAddr, event: InputEvent, now: Instant) -> Actions { let mut actions = Actions::default(); let Some(session) = self.active.as_mut() else { @@ -510,7 +540,7 @@ mod tests { let now = Instant::now(); let peer = addr([10, 0, 0, 5], 50000); - let actions = session.handle_control(peer, hello(1, 50000), &TestConfig, now); + let actions = session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now); assert_eq!(actions.replies.len(), 1); let ack = parse_ack(&actions.replies[0].1); @@ -535,10 +565,10 @@ mod tests { let now = Instant::now(); let peer = addr([10, 0, 0, 5], 50000); - let first = session.handle_control(peer, hello(1, 50000), &TestConfig, now); + let first = session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now); let first_id = parse_ack(&first.replies[0].1).session_id; - let dup = session.handle_control(peer, hello(1, 50000), &TestConfig, now); + let dup = session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now); assert_eq!(parse_ack(&dup.replies[0].1).session_id, first_id); assert!(dup.new_target.is_none(), "retransmit must not re-target"); assert!(!dup.force_idr); @@ -549,12 +579,12 @@ mod tests { let mut session = Session::new(1234); let now = Instant::now(); let peer = addr([10, 0, 0, 5], 50000); - let first = session.handle_control(peer, hello(1, 50000), &TestConfig, now); + let first = session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now); let first_id = parse_ack(&first.replies[0].1).session_id; // App relaunched: new ephemeral source port, new nonce. let relaunched = addr([10, 0, 0, 5], 50101); - let second = session.handle_control(relaunched, hello(2, 50101), &TestConfig, now); + let second = session.handle_control_authed(relaunched, hello(2, 50101), &TestConfig, now); let second_id = parse_ack(&second.replies[0].1).session_id; assert_ne!( @@ -569,7 +599,7 @@ mod tests { fn different_ip_is_rejected_busy_while_active() { let mut session = Session::new(1234); let now = Instant::now(); - session.handle_control( + session.handle_control_authed( addr([10, 0, 0, 5], 50000), hello(1, 50000), &TestConfig, @@ -577,7 +607,7 @@ mod tests { ); let intruder = addr([10, 0, 0, 9], 40000); - let actions = session.handle_control(intruder, hello(9, 40000), &TestConfig, now); + let actions = session.handle_control_authed(intruder, hello(9, 40000), &TestConfig, now); let ack = parse_ack(&actions.replies[0].1); assert_eq!(ack.status, HelloStatus::Busy); assert_eq!(ack.session_id, 0); @@ -604,7 +634,7 @@ mod tests { refresh_hz: 60, device_name: String::new(), }); - let actions = session.handle_control(peer, msg, &TestConfig, now); + let actions = session.handle_control_authed(peer, msg, &TestConfig, now); assert_eq!( parse_ack(&actions.replies[0].1).status, HelloStatus::VersionUnsupported @@ -617,9 +647,9 @@ mod tests { let mut session = Session::new(1234); let now = Instant::now(); let peer = addr([10, 0, 0, 5], 50000); - session.handle_control(peer, hello(1, 50000), &TestConfig, now); + session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now); - let bye = session.handle_control( + let bye = session.handle_control_authed( peer, ControlMessage::Bye(ByeReason::UserDisconnect), &TestConfig, @@ -629,7 +659,7 @@ mod tests { assert!(!session.is_active()); // Re-establish, then let liveness lapse. - session.handle_control(peer, hello(2, 50000), &TestConfig, now); + session.handle_control_authed(peer, hello(2, 50000), &TestConfig, now); let expired = session.tick(&TestConfig, false, now + LIVENESS_TIMEOUT); assert!(expired.client_lost); assert!(!session.is_active()); @@ -640,12 +670,12 @@ mod tests { let mut session = Session::new(1234); let now = Instant::now(); let peer = addr([10, 0, 0, 5], 50000); - session.handle_control(peer, hello(1, 50000), &TestConfig, now); + session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now); let later = now + Duration::from_millis(2500); let mut report = ReceiverReport::default(); report.frames_complete = 99; - session.handle_control( + session.handle_control_authed( peer, ControlMessage::ReceiverReport(report), &TestConfig, @@ -667,7 +697,7 @@ mod tests { let mut session = Session::new(1234); let now = Instant::now(); let peer = addr([10, 0, 0, 5], 50000); - session.handle_control(peer, hello(1, 50000), &TestConfig, now); + session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now); let request = ControlMessage::KeyframeRequest(KeyframeRequest { stream_epoch: 7, @@ -675,10 +705,10 @@ mod tests { reason: KeyframeReason::GapLoss, }); - let first = session.handle_control(peer, request.clone(), &TestConfig, now); + let first = session.handle_control_authed(peer, request.clone(), &TestConfig, now); assert!(first.force_idr); - let spammed = session.handle_control( + let spammed = session.handle_control_authed( peer, request.clone(), &TestConfig, @@ -689,8 +719,12 @@ mod tests { "requests inside the window are coalesced" ); - let granted_again = - session.handle_control(peer, request, &TestConfig, now + Duration::from_millis(600)); + let granted_again = session.handle_control_authed( + peer, + request, + &TestConfig, + now + Duration::from_millis(600), + ); assert!(granted_again.force_idr); } @@ -719,8 +753,8 @@ mod tests { let peer = addr([10, 0, 0, 5], 50000); // View-only session: input is dropped. - session.handle_control(peer, hello(1, 50000), &TestConfig, now); - let dropped = session.handle_control(peer, input_event(1), &TestConfig, now); + session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now); + let dropped = session.handle_control_authed(peer, input_event(1), &TestConfig, now); assert!( dropped.input.is_none(), "view-only sessions must not inject" @@ -732,17 +766,68 @@ mod tests { _ => unreachable!(), }; wants.feature_caps = FEATURE_WANTS_INPUT; - session.handle_control(peer, ControlMessage::Hello2(wants), &TestConfig, now); + session.handle_control_authed(peer, ControlMessage::Hello2(wants), &TestConfig, now); - let relayed = session.handle_control(peer, input_event(2), &TestConfig, now); + let relayed = session.handle_control_authed(peer, input_event(2), &TestConfig, now); assert!(relayed.input.is_some()); // A different IP can't inject into this session. let intruder = addr([10, 0, 0, 9], 40000); - let foreign = session.handle_control(intruder, input_event(3), &TestConfig, now); + let foreign = session.handle_control_authed(intruder, input_event(3), &TestConfig, now); assert!(foreign.input.is_none()); } + #[test] + fn control_messages_need_the_negotiated_session_id() { + let mut session = Session::new(1234); + let now = Instant::now(); + let peer = addr([10, 0, 0, 5], 50000); + let mut wants = match hello(1, 50000) { + ControlMessage::Hello2(h) => h, + _ => unreachable!(), + }; + wants.feature_caps = FEATURE_WANTS_INPUT; + session.handle_control(peer, 0, ControlMessage::Hello2(wants), &TestConfig, now); + let live = session.session_id().unwrap(); + + // Same source address, wrong session id: another process on that + // device, or a spoofed packet on the local link. It must not move the + // mouse, steer the bitrate, or end the session. + let wrong = live.wrapping_add(1); + assert!(session + .handle_control(peer, wrong, input_event(1), &TestConfig, now) + .input + .is_none()); + assert!(session + .handle_control( + peer, + wrong, + ControlMessage::ReceiverReport(ReceiverReport::default()), + &TestConfig, + now + ) + .report + .is_none()); + let bye = session.handle_control( + peer, + wrong, + ControlMessage::Bye(ByeReason::UserDisconnect), + &TestConfig, + now, + ); + assert!( + !bye.client_lost, + "a foreign BYE must not tear the session down" + ); + assert!(session.is_active()); + + // The real client still works. + assert!(session + .handle_control(peer, live, input_event(2), &TestConfig, now) + .input + .is_some()); + } + #[test] fn input_extends_liveness() { let mut session = Session::new(1234); @@ -753,11 +838,11 @@ mod tests { _ => unreachable!(), }; wants.feature_caps = FEATURE_WANTS_INPUT; - session.handle_control(peer, ControlMessage::Hello2(wants), &TestConfig, now); + session.handle_control_authed(peer, ControlMessage::Hello2(wants), &TestConfig, now); // A drag mid-window keeps the session alive past the original deadline. let later = now + Duration::from_millis(2500); - session.handle_control(peer, input_event(1), &TestConfig, later); + session.handle_control_authed(peer, input_event(1), &TestConfig, later); let ticked = session.tick(&TestConfig, false, now + Duration::from_millis(4000)); assert!(!ticked.client_lost); } @@ -769,7 +854,7 @@ mod tests { assert!(session.tick(&TestConfig, true, now).replies.is_empty()); let peer = addr([10, 0, 0, 5], 50000); - session.handle_control(peer, hello(1, 50000), &TestConfig, now); + session.handle_control_authed(peer, hello(1, 50000), &TestConfig, now); let ticked = session.tick(&TestConfig, true, now + Duration::from_millis(100)); assert_eq!(ticked.replies.len(), 1); let (_, message) = eternal_wire::v2::control::parse_control(&ticked.replies[0].1).unwrap(); diff --git a/ios/EternalMonitor/Input/TouchRelay.swift b/ios/EternalMonitor/Input/TouchRelay.swift index ccc7949..b4fa5c1 100644 --- a/ios/EternalMonitor/Input/TouchRelay.swift +++ b/ios/EternalMonitor/Input/TouchRelay.swift @@ -100,6 +100,11 @@ struct TouchRelayMachine { var videoPixelSize: CGSize = .zero private var mode: Mode = .idle + /// The last touch position that fell inside the video. Releases outside it + /// (a drag that ends on a letterbox bar, which is the default layout on a + /// 4:3 iPad showing a 16:9 desktop) reuse this instead of jumping to the + /// middle of the desktop and dropping whatever was being dragged there. + private var lastInsidePoint: Point? private var touchCount = 0 private var nextEventId: UInt32 = 0 private var lastMoveSentUs: UInt64 = 0 @@ -135,6 +140,7 @@ struct TouchRelayMachine { mutating func touchBegan( at point: Point?, isPencil: Bool, timeUs: UInt64 ) -> [Output] { + if let point { lastInsidePoint = point } touchCount += 1 switch (mode, touchCount) { case (.idle, 1): @@ -151,14 +157,14 @@ struct TouchRelayMachine { return [] case (.pending, 2): // Second finger before commit: this is a scroll, not a click. - mode = .scrolling(lastCentroid: point ?? Point(x: 32767, y: 32767)) + mode = .scrolling(lastCentroid: point ?? lastInsidePoint ?? Point(x: 32767, y: 32767)) scrollRemainder = .zero return [] case (.leftDown(let isPencil), 2) where !isPencil: // Finger drag joined by a second finger: release, then scroll. let release = edge(Phase.ended, kind: Kind.touch, buttons: 1, - at: point ?? Point(x: 32767, y: 32767), timeUs: timeUs) - mode = .scrolling(lastCentroid: point ?? Point(x: 32767, y: 32767)) + at: point ?? lastInsidePoint ?? Point(x: 32767, y: 32767), timeUs: timeUs) + mode = .scrolling(lastCentroid: point ?? lastInsidePoint ?? Point(x: 32767, y: 32767)) scrollRemainder = .zero return release case (.scrolling, _): @@ -174,6 +180,7 @@ struct TouchRelayMachine { mutating func touchMoved( to point: Point?, centroid: Point?, isPencil: Bool, force: CGFloat, timeUs: UInt64 ) -> [Output] { + if let point { lastInsidePoint = point } switch mode { case .pending(let start, _): guard let point else { return [] } @@ -225,6 +232,7 @@ struct TouchRelayMachine { } mutating func touchEnded(at point: Point?, cancelled: Bool, timeUs: UInt64) -> [Output] { + if let point { lastInsidePoint = point } touchCount = max(0, touchCount - 1) let releasePhase = cancelled ? Phase.cancelled : Phase.ended switch mode { @@ -238,11 +246,11 @@ struct TouchRelayMachine { mode = .idle let kind = isPencil ? Kind.pencil : Kind.touch return edge(releasePhase, kind: kind, buttons: 1, - at: point ?? Point(x: 32767, y: 32767), timeUs: timeUs) + at: point ?? lastInsidePoint ?? Point(x: 32767, y: 32767), timeUs: timeUs) case .rightDown: mode = .idle return edge(releasePhase, kind: Kind.touch, buttons: 0b10, - at: point ?? Point(x: 32767, y: 32767), timeUs: timeUs) + at: point ?? lastInsidePoint ?? Point(x: 32767, y: 32767), timeUs: timeUs) case .scrolling: if touchCount == 0 { mode = .idle } return [] @@ -275,10 +283,10 @@ struct TouchRelayMachine { switch mode { case .leftDown(let isPencil): outputs = edge(Phase.cancelled, kind: isPencil ? Kind.pencil : Kind.touch, - buttons: 1, at: Point(x: 32767, y: 32767), timeUs: timeUs) + buttons: 1, at: lastInsidePoint ?? Point(x: 32767, y: 32767), timeUs: timeUs) case .rightDown: outputs = edge(Phase.cancelled, kind: Kind.touch, buttons: 0b10, - at: Point(x: 32767, y: 32767), timeUs: timeUs) + at: lastInsidePoint ?? Point(x: 32767, y: 32767), timeUs: timeUs) case .pending, .scrolling, .idle, .suppressed: break } diff --git a/ios/EternalMonitor/Network/ControlChannel.swift b/ios/EternalMonitor/Network/ControlChannel.swift index 02b7f85..754399a 100644 --- a/ios/EternalMonitor/Network/ControlChannel.swift +++ b/ios/EternalMonitor/Network/ControlChannel.swift @@ -239,13 +239,26 @@ final class ControlChannel { } /// Fire-and-forget goodbye, sent a few times for loss tolerance. + /// Say goodbye so the host stops streaming at once instead of waiting out + /// its liveness timeout. + /// + /// The first copy goes out SYNCHRONOUSLY. Callers tear the socket down on + /// the very next line, so anything merely enqueued here was still sitting + /// on the queue when the connection was cancelled — the host saw no BYE, + /// kept sending into a dead peer for the full three seconds, held the + /// virtual display up for the same three seconds, and answered an + /// immediate reconnect with "busy". func sendBye(_ reason: ByeReason) { - queue.async { [self] in + queue.sync { [self] in guard sessionId != 0 else { return } - for delay in [0, 50, 100] { - queue.asyncAfter(deadline: .now() + .milliseconds(delay)) { [weak self] in - self?.sendMessage(.bye(reason)) - } + sendMessage(.bye(reason)) + } + // Two more for loss tolerance; these are best-effort and become no-ops + // once the socket is gone. + for delay in [50, 100] { + queue.asyncAfter(deadline: .now() + .milliseconds(delay)) { [weak self] in + guard let self, self.sessionId != 0 else { return } + self.sendMessage(.bye(reason)) } } } diff --git a/ios/EternalMonitorTests/TouchRelayTests.swift b/ios/EternalMonitorTests/TouchRelayTests.swift index d2a244f..78d7c54 100644 --- a/ios/EternalMonitorTests/TouchRelayTests.swift +++ b/ios/EternalMonitorTests/TouchRelayTests.swift @@ -121,6 +121,21 @@ final class TouchRelayMachineTests: XCTestCase { XCTAssertEqual(outputs[0].phase, TouchRelayMachine.Phase.began) } + func testReleaseOutsideTheVideoUsesTheLastPointInside() { + // 16:9 desktop on a 4:3 iPad means letterbox bars: a drag that ends + // with the finger on a bar has no valid coordinate. Releasing at the + // screen centre there drops the dragged window in the middle of the + // desktop. + _ = machine.touchBegan(at: P(x: 40000, y: 40000), isPencil: false, timeUs: t) + _ = machine.touchMoved( + to: P(x: 50000, y: 50000), centroid: nil, isPencil: false, force: 0, timeUs: t + 20_000 + ) + let release = sends(machine.touchEnded(at: nil, cancelled: false, timeUs: t + 40_000)) + XCTAssertEqual(release.first?.xNorm, 50000) + XCTAssertEqual(release.first?.yNorm, 50000) + XCTAssertNotEqual(release.first?.xNorm, 32767, "must not fall back to screen centre") + } + func testSmallJitterStaysATap() { let start = P(x: 10000, y: 10000) _ = machine.touchBegan(at: start, isPencil: false, timeUs: t) diff --git a/proto/src/reassembly.rs b/proto/src/reassembly.rs index 3a3b3ec..f24b1e2 100644 --- a/proto/src/reassembly.rs +++ b/proto/src/reassembly.rs @@ -45,6 +45,10 @@ pub enum DropReason { ZeroCount, /// fragment_index >= fragment_count. IndexOutOfRange, + /// Fragment count disagrees with the live frame's; first-seen count wins. + CountMismatch, + /// This index of this frame was already stored. + DuplicateFragment, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -64,6 +68,16 @@ struct PendingFrame { created_at: Instant, } +impl PendingFrame { + /// Fragments that never arrived. THIS is what loss means: counting the + /// ones that did arrive (as this mirror used to) reported a frame missing + /// one fragment out of ten as nine lost, so the host ABR saw roughly nine + /// times the real loss and stepped down on a healthy link. + fn missing_fragments(&self) -> u64 { + u64::from(self.fragment_count).saturating_sub(self.fragments.len() as u64) + } +} + /// Cumulative counters for loss accounting (feeds receiver reports in v2). #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] pub struct ReassemblyCounters { @@ -169,32 +183,43 @@ impl Reassembler { return AddOutcome::Dropped(DropReason::IndexOutOfRange); } - self.counters.frags_received += 1; - - let recreate = match self.pending.get(&seq) { - Some(frame) => frame.fragment_count != count, - None => true, - }; - if recreate { - if let Some(old) = self.pending.remove(&seq) { - // Fragment count changed mid-frame: restart that frame. - self.counters.frags_lost += old.fragments.len() as u64; + match self.pending.get(&seq) { + // Conflicting metadata for a live frame: the FIRST-seen count wins. + // Rebuilding the frame here (as this mirror used to) let one late + // duplicate or spoofed datagram throw away every fragment already + // buffered for it. + Some(frame) if frame.fragment_count != count => { + return AddOutcome::Dropped(DropReason::CountMismatch); + } + Some(_) => {} + None => { + self.pending.insert( + seq, + PendingFrame { + fragment_count: count, + fragments: HashMap::new(), + created_at: now, + }, + ); } - self.pending.insert( - seq, - PendingFrame { - fragment_count: count, - fragments: HashMap::new(), - created_at: now, - }, - ); } let frame = self .pending .get_mut(&seq) - .expect("pending frame just inserted"); + .expect("pending frame present or just inserted"); + // First write wins, and only a first write counts as received: a + // replayed fragment must not overwrite good bytes or inflate the + // receive count the loss math is derived from. + if frame.fragments.contains_key(&index) { + return AddOutcome::Dropped(DropReason::DuplicateFragment); + } frame.fragments.insert(index, payload.to_vec()); + self.counters.frags_received += 1; + let frame = self + .pending + .get_mut(&seq) + .expect("pending frame present or just inserted"); let mut outcome = AddOutcome::Stored; if frame.fragments.len() == usize::from(frame.fragment_count) { @@ -219,7 +244,7 @@ impl Reassembler { let keep = pending_seq > seq; if !keep { dropped_frames += 1; - dropped_frags += frame.fragments.len() as u64; + dropped_frags += frame.missing_fragments(); } keep }); @@ -247,7 +272,7 @@ impl Reassembler { fn reset_internal(&mut self) { for (_, frame) in self.pending.drain() { self.counters.frames_dropped += 1; - self.counters.frags_lost += frame.fragments.len() as u64; + self.counters.frags_lost += frame.missing_fragments(); } self.latest_completed_seq = 0; self.cleanup_counter = 0; @@ -259,7 +284,7 @@ impl Reassembler { let fresh = now.duration_since(frame.created_at) < STALE_FRAME_TIMEOUT; if !fresh { counters.frames_dropped += 1; - counters.frags_lost += frame.fragments.len() as u64; + counters.frags_lost += frame.missing_fragments(); } fresh }); @@ -422,6 +447,7 @@ mod tests { ); assert_eq!(r.pending_frames(), 0); assert_eq!(r.counters().frames_dropped, 1); + // One fragment of the two-fragment frame never arrived. assert_eq!(r.counters().frags_lost, 1); // The evicted frame's late fragment is now stale. @@ -432,16 +458,58 @@ mod tests { } #[test] - fn fragment_count_change_restarts_that_frame() { + fn first_seen_fragment_count_wins_and_progress_survives() { let now = Instant::now(); let mut r = Reassembler::new(); feed(&mut r, 1, 0, 3, 1, 0xA, now); - // Same seq arrives claiming 2 fragments: previous progress discarded. - feed(&mut r, 1, 0, 2, 1, 0xB, now); + + // A fragment claiming a different count for a live frame is ignored; + // it must not discard what has already been buffered. assert_eq!( - feed(&mut r, 1, 1, 2, 1, 0xC, now), - AddOutcome::Completed(vec![0xB, 0xC]) + feed(&mut r, 1, 1, 2, 1, 0xB, now), + AddOutcome::Dropped(DropReason::CountMismatch) + ); + + assert_eq!(feed(&mut r, 1, 1, 3, 1, 0xB, now), AddOutcome::Stored); + assert_eq!( + feed(&mut r, 1, 2, 3, 1, 0xC, now), + AddOutcome::Completed(vec![0xA, 0xB, 0xC]), + "the original fragments must still be there" + ); + } + + #[test] + fn replayed_fragment_neither_overwrites_nor_counts_twice() { + let now = Instant::now(); + let mut r = Reassembler::new(); + feed(&mut r, 1, 0, 2, 1, 0xA, now); + assert_eq!( + feed(&mut r, 1, 0, 2, 1, 0xFF, now), + AddOutcome::Dropped(DropReason::DuplicateFragment) + ); + assert_eq!(r.counters().frags_received, 1, "a replay is not a receipt"); + assert_eq!( + feed(&mut r, 1, 1, 2, 1, 0xB, now), + AddOutcome::Completed(vec![0xA, 0xB]), + "the first write must win" + ); + } + + #[test] + fn loss_counts_the_fragments_that_never_arrived() { + let now = Instant::now(); + let mut r = Reassembler::new(); + // Frame 1 gets 9 of 10 fragments, then a newer frame completes and + // evicts it: exactly ONE fragment was lost, not nine. + for index in 0..9u16 { + feed(&mut r, 1, index, 10, 1, 0xA, now); + } + assert_eq!( + feed(&mut r, 2, 0, 1, 1, 0xB, now), + AddOutcome::Completed(vec![0xB]) ); + assert_eq!(r.counters().frames_dropped, 1); + assert_eq!(r.counters().frags_lost, 1); } #[test]