From 2738de495901a2374604b003ccb36586e4d5166d Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Mon, 19 Oct 2026 12:35:17 -0700 Subject: [PATCH] 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 | 157 ++++++++++++++++++++++++++++++++++--- tines-sim/src/capped.rs | 83 ++++++++++++++++++++ tines-sim/src/lib.rs | 2 + tines-sim/src/sim.rs | 21 ++++- tines-sim/tests/props.rs | 50 +++++++++++- 5 files changed, 299 insertions(+), 14 deletions(-) create mode 100644 tines-sim/src/capped.rs diff --git a/tines-core/src/exchange.rs b/tines-core/src/exchange.rs index 29af5b78..d12e272c 100644 --- a/tines-core/src/exchange.rs +++ b/tines-core/src/exchange.rs @@ -802,15 +802,21 @@ impl Exchange { 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 { + 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 { + 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]; diff --git a/tines-sim/src/capped.rs b/tines-sim/src/capped.rs new file mode 100644 index 00000000..249ef76d --- /dev/null +++ b/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 { + 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 { + 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) + } +} diff --git a/tines-sim/src/lib.rs b/tines-sim/src/lib.rs index b8832e68..5909d19c 100644 --- a/tines-sim/src/lib.rs +++ b/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; diff --git a/tines-sim/src/sim.rs b/tines-sim/src/sim.rs index cca940f0..8d6f39f3 100644 --- a/tines-sim/src/sim.rs +++ b/tines-sim/src/sim.rs @@ -19,15 +19,30 @@ use tines_core::{MemStore, Node, NodeId, ObservationId, Store}; pub type SimNode = Node; /// 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 +where + S: Store, +{ pub world: Rc>, - pub nodes: Vec, + pub nodes: Vec>, 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 Sim { + /// A run over `nodes` nodes, each given a fresh `S::default()` store. + pub fn with_store(cfg: WorldConfig, trace: ContactTrace) -> Sim { 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), )); } diff --git a/tines-sim/tests/props.rs b/tines-sim/tests/props.rs index bc70a6f9..22fefb69 100644 --- a/tines-sim/tests/props.rs +++ b/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::::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) + ); + } }