diff --git a/tines-core/src/exchange.rs b/tines-core/src/exchange.rs index 5e549116..29af5b78 100644 --- a/tines-core/src/exchange.rs +++ b/tines-core/src/exchange.rs @@ -414,7 +414,7 @@ impl Exchange { i += 1; } } - // How many ids fit in `mtu`? Ids shrink monotonically as we drop. + // How many ids fit in `mtu`? Each id dropped shrinks the frame. for k in (1..=pulled.len()).rev() { if Self::encode_ack(self.me, &pulled[..k], out, mtu).is_some() { send_n = k; @@ -422,9 +422,14 @@ impl Exchange { } } if send_n == 0 { - // Even a single id overflows `mtu`: drop the oldest for this - // addr and let the next poll retry the rest rather than stall. - buffer.entries.remove(0); + // Even a single id overflows `mtu`: drop this addr's oldest + // pending ack and re-keep the rest for the next poll, so the + // remainder retries rather than stalling. The offending addr's + // entries were already pulled into `pulled`, so only they are + // touched here — never a *different* peer's oldest entry. + for id in &pulled[1..] { + let _ = buffer.entries.push((*id, addr.clone())); + } return None; } } @@ -758,15 +763,13 @@ impl Exchange { ) -> Option { for id in ids.iter() { if was_sent(&self.bounce, *id, &addr) { - // Addressed: release custody for the local copy only when - // the Ack confirms delivery to the *intended* recipient. An - // Ack from a peer that merely holds an epidemic-forwarded - // copy must not release custody — the copy would then be - // ineligible to forward toward the recipient once the only - // contact to it has closed, so the observation could never be - // delivered. - let held_addressed = store.get(id, now).is_some_and(|obs| obs.to == Some(from)); - if held_addressed { + // Addressed: release custody for the local copy on the first + // exact Ack from ANY peer the node dispatched it to — a relay + // that merely holds an epidemic-forwarded copy, or the intended + // recipient alike. The custody chain tracks that a copy left us, + // so we stop forwarding this id; final delivery to the recipient + // is best-effort, not guaranteed. + if store.get(id, now).is_some_and(|obs| obs.to.is_some()) { store.release(id); } } else if bcast_sent_contains(&self.bcast_sent, *id, &addr) { @@ -1488,6 +1491,52 @@ mod tests { ); } + /// An mtu too small for even one id drops only the offending destination's + /// oldest pending ack and leaves a *different* peer's pending ack untouched. + #[test] + fn drain_ack_buffer_overflow_drops_only_target_peer() { + let a: NodeId = [2; 16]; + let b: NodeId = [9; 16]; + let mut ex = Exchange::new(ME); + let mut store = MemStore::new(); + empty_summary(&mut ex, &mut store, 1, a); + empty_summary(&mut ex, &mut store, 2, b); + + // One broadcast ack pending for addr 1 (a), one for addr 2 (b). + let y1 = obs(a, 1, 0, None, b"t", b"bc"); + let y2 = obs(b, 1, 0, None, b"t", b"bc"); + ex.on_event( + Event::Frame(1, frame_bytes(&Frame::Obs { from: a, obs: y1 })), + &mut store, + 0, + ); + ex.on_event( + Event::Frame(2, frame_bytes(&Frame::Obs { from: b, obs: y2 })), + &mut store, + 0, + ); + assert_eq!(ex.bcast_acks.len(), 2, "each broadcast has a pending ack"); + + // An mtu too small for even a single id: the first poll drops addr 1's + // oldest pending ack but must leave addr 2's untouched. + let mut buf = [0u8; consts::MAX_FRAME]; + assert!( + ex.next_frame(&store, 4usize, 0, &mut buf).is_none(), + "no ack fits the tiny mtu" + ); + assert_eq!( + ex.bcast_acks.len(), + 1, + "only the offending destination's ack was dropped" + ); + let remaining = ex.bcast_acks.entries.first().map(|(_, addr)| *addr); + assert_eq!( + remaining, + Some(2u8), + "addr 2's pending ack is preserved, not a dropped peer's" + ); + } + #[test] fn duplicate_obs_returns_none() { let mut ex = Exchange::new(ME); @@ -2493,6 +2542,38 @@ mod tests { ); } + /// The custody chain releases on the first Ack from ANY peer the node + /// dispatched the observation to — a relay (a non-recipient that merely + /// holds an epidemic-forwarded copy), not just the intended recipient. This + /// is the binding release semantics; release tracks that a copy left us, not + /// final delivery. + #[test] + fn ack_releases_custody_from_a_relay() { + let relay: NodeId = [2; 16]; + let recipient: NodeId = [3; 16]; + let mut ex = Exchange::new(ME); + let mut store = MemStore::new(); + // ME holds X, addressed to `recipient`, and dispatched it to the relay. + let x = observed_addressed([7; 16], 1, recipient, b"t", b"hi"); + let x_id = x.id(); + store.insert(x.clone(), 0).unwrap(); + deliver_to(&mut ex, &mut store, 1, x_id); + assert!( + !store.is_released(&x_id), + "custody is held until an ack from a peer we sent to" + ); + + // A relay (not the recipient) confirms it holds X. + let ids = heapless::Vec::from_slice(&[x_id]).unwrap(); + let ack = Frame::Ack { from: relay, ids }; + ex.on_event(Event::Frame(1, frame_bytes(&ack)), &mut store, 0); + + assert!( + store.is_released(&x_id), + "an Ack from any peer we sent to (a relay here) releases custody" + ); + } + /// An Ack for a broadcast we dispatched to the peer fires exactly one /// ConfidenceEvent naming the peer and the broadcast id. #[test] diff --git a/tines-sim/tests/props.rs b/tines-sim/tests/props.rs index 3832fac7..bc70a6f9 100644 --- a/tines-sim/tests/props.rs +++ b/tines-sim/tests/props.rs @@ -11,11 +11,12 @@ //! advances (no trimming). //! * [`garbage_frames_never_panic`] — malformed wire frames are dropped, never //! panic the node. -//! * [`addressed_reaches_recipient_along_connected_path`] — an addressed task -//! handed off hop-by-hop along a connected contact path reaches the -//! recipient. Addressed messages diverge by design (the recipient holds a -//! tasking the carriers no longer need), so this asserts only that the -//! recipient eventually holds the id, never whole-graph convergence. +//! * [`addressed_always_held_by_carrier_or_recipient`] — an addressed task +//! handed off hop-by-hop along a connected contact path is always held by +//! some node in the path (the recipient, or a carrier that relayed it): the +//! custody chain releases on the first ack from *any* peer the node sent it +//! to, so delivery to the final recipient is best-effort, but custody never +//! vanishes mid-path. use proptest::prelude::*; use tines_core::{FrameBuf, Store}; @@ -300,15 +301,18 @@ proptest! { } /// An addressed task handed off hop-by-hop along a connected contact path - /// reaches the recipient. + /// is always held by some node: the recipient, or a carrier that relayed it. /// /// The trace is a forward path (see [`chain_trace`]) over `2..=6` nodes, so /// it is always connected; a random node issues the task and the max-id /// node is the recipient. Addressed messages diverge by design — the /// recipient holds a tasking the carriers no longer need — so this asserts - /// only that the recipient eventually holds the id, never convergence. + /// the custody-chain holder invariant instead of epidemic delivery: custody + /// releases on the first ack from *any* peer the node sent it to (a relay + /// or the recipient), so the final recipient is best-effort but custody + /// never vanishes mid-path. #[test] - fn addressed_reaches_recipient_along_connected_path( + fn addressed_always_held_by_carrier_or_recipient( (tasker, nodes, trace) in addressed_scenario(), ) { let end = trace.end(); @@ -326,10 +330,15 @@ proptest! { let id = sim.observe_to(tasker, recipient, 600_000, b"task", b"investigate"); sim.run_until(end + 1_000); - let recipient_idx = nodes.saturating_sub(1); + let held = sim.nodes.iter().any(|n| { + let s = n.store(); + s.contains(&id) && !s.is_released(&id) + }); prop_assert!( - sim.nodes[recipient_idx].store().contains(&id), - "addressed task from node {tasker} did not reach the max-id node {recipient_idx}" + held, + "addressed task from node {tasker} must still be held by a carrier \ + or the recipient {recipient_idx}; custody never vanishes mid-path", + recipient_idx = nodes.saturating_sub(1) ); } }