task-4: revert on_ack release to custody chain; fix drain_ack_buffer overflow

Revert on_ack release gating to release-on-any-ack (custody chain): release
custody on the first exact Ack from ANY peer the node dispatched the
observation to (relay or recipient), not just the recipient.

drain_ack_buffer send_n==0 bug: re-keep only the target addr's own pulled
entries instead of removing buffer.entries[0] (which dropped an unrelated
peer's oldest pending ack).

Rename the tines-sim addressed proptest to
addressed_always_held_by_carrier_or_recipient and reconcile its assertion to
the custody-chain holder invariant. Reword the misleading "Ids shrink
monotonically" comment. Add unit tests: ack_releases_custody_from_a_relay and
drain_ack_buffer_overflow_drops_only_target_peer.

Gate green: tines-core 106 lib tests, tines-sim suite, sh scripts/ci.sh.

tines-core/src/exchange.rs

@@ -414,7 +414,7 @@ impl<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
                     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<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
                 }
             }
             if send_n == 0 {
-                // Even a single id overflows `mtu`: drop the oldest for this
+                // Even a single id overflows `mtu`: drop this addr's oldest
-                // addr and let the next poll retry the rest rather than stall.
+                // pending ack and re-keep the rest for the next poll, so the
-                buffer.entries.remove(0);
+                // 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<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
     ) -> Option<Observation> {
         for id in ids.iter() {
             if was_sent(&self.bounce, *id, &addr) {
-                // Addressed: release custody for the local copy only when
+                // Addressed: release custody for the local copy on the first
-                // the Ack confirms delivery to the *intended* recipient. An
+                // exact Ack from ANY peer the node dispatched it to — a relay
-                // Ack from a peer that merely holds an epidemic-forwarded
+                // that merely holds an epidemic-forwarded copy, or the intended
-                // copy must not release custody — the copy would then be
+                // recipient alike. The custody chain tracks that a copy left us,
-                // ineligible to forward toward the recipient once the only
+                // so we stop forwarding this id; final delivery to the recipient
-                // contact to it has closed, so the observation could never be
+                // is best-effort, not guaranteed.
-                // delivered.
+                if store.get(id, now).is_some_and(|obs| obs.to.is_some()) {
-                let held_addressed = store.get(id, now).is_some_and(|obs| obs.to == Some(from));
-                if held_addressed {
                     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]

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
+//! * [`addressed_always_held_by_carrier_or_recipient`] — an addressed task
-//!   handed off hop-by-hop along a connected contact path reaches the
+//!   handed off hop-by-hop along a connected contact path is always held by
-//!   recipient. Addressed messages diverge by design (the recipient holds a
+//!   some node in the path (the recipient, or a carrier that relayed it): the
-//!   tasking the carriers no longer need), so this asserts only that the
+//!   custody chain releases on the first ack from *any* peer the node sent it
-//!   recipient eventually holds the id, never whole-graph convergence.
+//!   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),
+            held,
-            "addressed task from node {tasker} did not reach the max-id node {recipient_idx}"
+            "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)
         );
     }
 }