fix(core): ack only observations the store kept

Exchange::accept enqueued a pending ack whenever
!(duplicate && addressed && released) held, even when the backend
store.insert returned Err (store full / backend error). On Err the
released flag computed to false, so an ack was still enqueued: a
receiver that did not keep the message acked it. For an addressed
offer that releases the sender's custody for a message nobody holds;
for a broadcast it fires a false confidence event at the sender.

Gate the ack enqueue on res.is_ok(): enqueue only when the store
actually retained the observation. The existing released-holder
exception still applies among kept copies. No bounce is recorded on
Err (already the case).

Adds a unit test exercising a capped store (StoreError::Full) for
both a broadcast and an addressed offer, and a tines-sim property
that drives custody with CappedStore-backed nodes and asserts the
"at least one holder" invariant still holds when stores fill.

tines-core/src/exchange.rs

@@ -802,15 +802,21 @@ impl<A: Clone + PartialEq + core::fmt::Debug> Exchange<A> {
         let addressed = obs.to.is_some();
         let returned = obs.clone();
         let res = store.insert(obs, now);
-        // Enqueue a pending ack for what we received: a newly stored
-        // observation, any duplicate, and even an un-stored id (its id is
-        // validated by the receiver). The single exception is a duplicate
-        // *addressed* offer whose local copy has been released — a released
-        // holder must not ack — so that one is held back.
-        let duplicate = !matches!(res, Ok(true));
-        let released = res.is_ok() && store.is_released(&id);
-        if !(duplicate && addressed && released) {
-            self.enqueue_pending_ack(id, addr.clone(), addressed);
+        // Enqueue a pending ack only for what the store actually kept: a newly
+        // stored observation, or a duplicate whose local copy is still held. A
+        // store that rejected the observation (full, backend error) holds
+        // nothing, so it must not ack — for an addressed offer that would
+        // release the sender's custody for a message nobody holds, and for a
+        // broadcast it would fire a false confidence event at the sender. The
+        // single exception among the kept copies is a duplicate *addressed*
+        // offer whose local copy has been released — a released holder must
+        // not ack.
+        if res.is_ok() {
+            let duplicate = !matches!(res, Ok(true));
+            let released = store.is_released(&id);
+            if !(duplicate && addressed && released) {
+                self.enqueue_pending_ack(id, addr.clone(), addressed);
+            }
         }
         if addressed && res.is_ok() {
             record_bounce(&mut self.bounce, id, addr, BounceKind::Received);
@@ -1127,7 +1133,7 @@ fn insert_pos(heap: &Vec<(ObservationId, u64), { consts::SEND_QUEUE_CAP }>, age:
 #[cfg(all(test, feature = "alloc"))]
 mod tests {
     use super::*;
-    use crate::{MemStore, Observation, Payload, SendError, Topic};
+    use crate::{MemStore, Observation, Payload, SendError, Store, StoreError, Topic, TrimPolicy};
     use std::vec::Vec;
 
     const ME: NodeId = [42; 16];
@@ -1372,6 +1378,137 @@ mod tests {
         );
     }
 
+    /// A `Store` that delegates to a [`MemStore`] but rejects `insert` with
+    /// `StoreError::Full` once it has held `cap` observations — a stand-in for
+    /// a real backend that refuses when full. `MemStore` itself never fills.
+    struct FullStore {
+        cap: usize,
+        inner: MemStore,
+    }
+
+    impl FullStore {
+        fn new(cap: usize) -> Self {
+            Self {
+                cap,
+                inner: MemStore::new(),
+            }
+        }
+    }
+
+    impl Store for FullStore {
+        fn insert(&mut self, obs: Observation, now: Tick) -> Result<bool, StoreError> {
+            if self.inner.len() >= self.cap {
+                return Err(StoreError::Full);
+            }
+            self.inner.insert(obs, now)
+        }
+
+        fn contains(&self, id: &ObservationId) -> bool {
+            self.inner.contains(id)
+        }
+
+        fn get(&self, id: &ObservationId, now: Tick) -> Option<Observation> {
+            self.inner.get(id, now)
+        }
+
+        fn len(&self) -> usize {
+            self.inner.len()
+        }
+
+        fn for_each_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+            self.inner.for_each_id(f)
+        }
+
+        fn for_each_broadcast_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+            self.inner.for_each_broadcast_id(f)
+        }
+
+        fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+            self.inner.for_each_addressed_id(f)
+        }
+
+        fn release(&mut self, id: &ObservationId) {
+            self.inner.release(id)
+        }
+
+        fn is_released(&self, id: &ObservationId) -> bool {
+            self.inner.is_released(id)
+        }
+
+        fn for_each_topic(&self, prefix: &[u8], now: Tick, f: &mut dyn FnMut(&Observation)) {
+            self.inner.for_each_topic(prefix, now, f)
+        }
+
+        fn trim(&mut self, policy: &TrimPolicy, now: Tick) -> usize {
+            self.inner.trim(policy, now)
+        }
+    }
+
+    /// A store that rejected a receive (full, or backend error) must not ack,
+    /// whether the offer was addressed or a broadcast. Acking would release
+    /// the sender's custody for a message nobody holds (addressed) or fire a
+    /// false confidence event (broadcast) — custody lost for a message no one
+    /// keeps. Regression test for the full-store custody bug.
+    #[test]
+    fn store_full_receive_is_not_acked() {
+        let a: NodeId = [2; 16];
+        let mut ex = Exchange::new(ME);
+        // A store that never keeps anything it is asked to insert.
+        let mut store = FullStore::new(0);
+        // Make peer `a` known and appearing to lack everything.
+        let bits = heapless::Vec::from_slice(&[0u8; consts::SUMMARY_BYTES]).unwrap();
+        let sum = Frame::Summary {
+            from: a,
+            salt: 0,
+            bits,
+        };
+        ex.on_event(Event::Frame(1, frame_bytes(&sum)), &mut store, 0);
+
+        // A broadcast from A, received into a store that rejects it.
+        let b = obs(a, 1, 0, None, b"t", b"bc");
+        let b_id = b.id();
+        ex.on_event(
+            Event::Frame(1, frame_bytes(&Frame::Obs { from: a, obs: b })),
+            &mut store,
+            0,
+        );
+
+        // An addressed offer from A to ME, received into the same store.
+        let x = observed_addressed(a, 2, ME, b"t", b"hi");
+        let x_id = x.id();
+        ex.on_event(
+            Event::Frame(1, frame_bytes(&Frame::Obs { from: a, obs: x })),
+            &mut store,
+            0,
+        );
+
+        // Nothing was kept, so nothing is acked: no pending ack queued and no
+        // Ack frame carries either id.
+        assert!(
+            ex.acks.entries.is_empty(),
+            "a store-full addressed receive must not be acked"
+        );
+        assert!(
+            ex.bcast_acks.entries.is_empty(),
+            "a store-full broadcast receive must not be acked"
+        );
+        for _ in 0..consts::MAX_FRAME {
+            let mut buf = [0u8; consts::MAX_FRAME];
+            if let Some((_, n)) = ex.next_frame(&store, consts::MAX_FRAME, 0, &mut buf) {
+                if let Ok(Frame::Ack { ids, .. }) = Frame::decode(&buf[..n]) {
+                    assert!(
+                        !ids.contains(&b_id) && !ids.contains(&x_id),
+                        "a store-full receive must not ack id {b_id:?} or {x_id:?}"
+                    );
+                }
+            }
+        }
+        assert!(
+            ex.acks.entries.is_empty(),
+            "the store-full acks never left the buffer"
+        );
+    }
+
     #[test]
     fn ack_fits_mtu() {
         let a: NodeId = [2; 16];

tines-sim/src/capped.rs

@@ -0,0 +1,83 @@
+//! A [`Store`] that delegates to a [`MemStore`] but rejects `insert` with
+//! [`StoreError::Full`] once it has held `capacity` observations.
+//!
+//! `MemStore` is unbounded, so it never fills; this wrapper lets the simulator
+//! exercise the full-store custody path (F1): a receiver whose store is full
+//! must not ack a received observation, or custody is released for a message
+//! nobody holds.
+
+use tines_core::{MemStore, Observation, ObservationId, Store, StoreError, Tick, TrimPolicy};
+
+/// A capacity-bounded [`Store`]: delegates to a [`MemStore`] but returns
+/// `StoreError::Full` on `insert` once `capacity` observations are held.
+pub struct CappedStore {
+    capacity: usize,
+    inner: MemStore,
+}
+
+impl CappedStore {
+    /// A capped store holding at most `capacity` observations.
+    pub fn new(capacity: usize) -> Self {
+        Self {
+            capacity,
+            inner: MemStore::new(),
+        }
+    }
+}
+
+impl Default for CappedStore {
+    fn default() -> Self {
+        // Small enough that a node filling up on received observations rejects
+        // later inserts, which is what drives the full-store ack path.
+        Self::new(8)
+    }
+}
+
+impl Store for CappedStore {
+    fn insert(&mut self, obs: Observation, now: Tick) -> Result<bool, StoreError> {
+        if self.inner.len() >= self.capacity {
+            return Err(StoreError::Full);
+        }
+        self.inner.insert(obs, now)
+    }
+
+    fn contains(&self, id: &ObservationId) -> bool {
+        self.inner.contains(id)
+    }
+
+    fn get(&self, id: &ObservationId, now: Tick) -> Option<Observation> {
+        self.inner.get(id, now)
+    }
+
+    fn len(&self) -> usize {
+        self.inner.len()
+    }
+
+    fn for_each_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+        self.inner.for_each_id(f)
+    }
+
+    fn for_each_broadcast_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+        self.inner.for_each_broadcast_id(f)
+    }
+
+    fn for_each_addressed_id(&self, f: &mut dyn FnMut(&ObservationId)) {
+        self.inner.for_each_addressed_id(f)
+    }
+
+    fn release(&mut self, id: &ObservationId) {
+        self.inner.release(id)
+    }
+
+    fn is_released(&self, id: &ObservationId) -> bool {
+        self.inner.is_released(id)
+    }
+
+    fn for_each_topic(&self, prefix: &[u8], now: Tick, f: &mut dyn FnMut(&Observation)) {
+        self.inner.for_each_topic(prefix, now, f)
+    }
+
+    fn trim(&mut self, policy: &TrimPolicy, now: Tick) -> usize {
+        self.inner.trim(policy, now)
+    }
+}

tines-sim/src/lib.rs

@@ -9,11 +9,13 @@
 
 #![deny(warnings)]
 
+pub mod capped;
 pub mod sim;
 pub mod trace;
 pub mod transport;
 pub mod world;
 
+pub use capped::CappedStore;
 pub use sim::{Sim, SimNode};
 pub use trace::{Contact, ContactTrace};
 pub use transport::SimTransport;

tines-sim/src/sim.rs

@@ -19,15 +19,30 @@ use tines_core::{MemStore, Node, NodeId, ObservationId, Store};
 pub type SimNode = Node<MemStore, SimTransport>;
 
 /// A run of the simulation, owned by the test.
-pub struct Sim {
+///
+/// Generic over the store backend `S` so tests can drive custody with a store
+/// that fills (see [`CappedStore`](crate::CappedStore)); the default is the
+/// unbounded [`MemStore`].
+pub struct Sim<S = MemStore>
+where
+    S: Store,
+{
     pub world: Rc<RefCell<SimWorld>>,
-    pub nodes: Vec<SimNode>,
+    pub nodes: Vec<Node<S, SimTransport>>,
     pub t: u64,
     pub step_ms: u64,
 }
 
 impl Sim {
+    /// A run over `nodes` nodes, each with its own unbounded [`MemStore`].
     pub fn new(cfg: WorldConfig, trace: ContactTrace) -> Sim {
+        Sim::with_store(cfg, trace)
+    }
+}
+
+impl<S: Store + Default> Sim<S> {
+    /// A run over `nodes` nodes, each given a fresh `S::default()` store.
+    pub fn with_store(cfg: WorldConfig, trace: ContactTrace) -> Sim<S> {
         let n = cfg.nodes;
         let world = SimWorld::new(cfg, trace);
         let mut nodes = Vec::with_capacity(n);
@@ -35,7 +50,7 @@ impl Sim {
             nodes.push(Node::new(
                 [i as u8; 16],
                 0,
-                MemStore::new(),
+                S::default(),
                 SimTransport::new(world.clone(), i),
             ));
         }

tines-sim/tests/props.rs

@@ -20,7 +20,7 @@
 
 use proptest::prelude::*;
 use tines_core::{FrameBuf, Store};
-use tines_sim::{Contact, ContactTrace, Sim, WorldConfig};
+use tines_sim::{CappedStore, Contact, ContactTrace, Sim, WorldConfig};
 
 /// A fixed set of short observation topics the tests reuse so the generated
 /// observations round-trip cleanly through the core (`MAX_TOPIC` is 64 bytes).
@@ -341,4 +341,52 @@ proptest! {
             recipient_idx = nodes.saturating_sub(1)
         );
     }
+
+    /// Like the holder-invariant test above, but every node is backed by a
+    /// [`CappedStore`] whose `insert` returns `StoreError::Full` once the store
+    /// has held `8` observations. This drives F1's full-store path: once a
+    /// receiver's store is full it must not ack what it could not keep, or
+    /// custody for the addressed task is released for a message nobody holds.
+    /// Six of each node's own broadcasts are seeded in (all land, since 6 < 8)
+    /// and the run's gossiped broadcasts tip each node past capacity, so many
+    /// nodes reject inserts and the "at least one holder" invariant must still
+    /// hold.
+    #[test]
+    fn addressed_always_held_by_carrier_or_recipient_with_full_store(
+        (tasker, nodes, trace) in addressed_scenario(),
+    ) {
+        let end = trace.end();
+        let cfg = WorldConfig {
+            nodes,
+            mtu: 512,
+            loss: 0.0,
+            dup: 0.05,
+            seed: 1,
+        };
+        let mut sim = Sim::<CappedStore>::with_store(cfg, trace);
+        // Seed 6 of each node's own broadcasts: 6 < CappedStore's default cap
+        // (8), so every seed lands and the run's received broadcasts push each
+        // node into the full-store ack path.
+        for node in 0..nodes {
+            for seq in 0..6u64 {
+                let _ = sim.observe(node, b"temp", &seq.to_le_bytes());
+            }
+        }
+        // The max-id node is the recipient.
+        let recipient = [nodes.saturating_sub(1) as u8; 16];
+        let id = sim.observe_to(tasker, recipient, 600_000, b"task", b"investigate");
+        sim.run_until(end + 1_000);
+
+        let held = sim.nodes.iter().any(|n| {
+            let s = n.store();
+            s.contains(&id) && !s.is_released(&id)
+        });
+        prop_assert!(
+            held,
+            "addressed task from node {tasker} must still be held by a carrier \
+             or the recipient {recipient_idx}; custody never vanishes mid-path \
+             even when a node's store is full",
+            recipient_idx = nodes.saturating_sub(1)
+        );
+    }
 }