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]