safemesh_crdt/
transport.rs

1// SafeMesh — delta-state CRDT convergence, built on crdt-lean.
2// Copyright (C) 2026 Ben Cassie
3// SPDX-License-Identifier: Apache-2.0
4
5use crate::{Admission, Crdt, EventLog, Record, RecordId, Replica, VersionVector};
6use alloc::{collections::BTreeSet, vec::Vec};
7
8#[derive(Clone, Debug, PartialEq, Eq)]
9pub struct TransportEnvelope<D> {
10    pub from: u64,
11    pub to: u64,
12    pub records: Vec<Record<D>>,
13}
14
15#[derive(Clone, Copy, Debug, PartialEq, Eq)]
16pub enum TransportError {
17    NotSubscribed { peer: u64 },
18    Disconnected { from: u64, to: u64 },
19}
20
21impl core::fmt::Display for TransportError {
22    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
23        match self {
24            Self::NotSubscribed { peer } => write!(f, "peer {peer} is not subscribed"),
25            Self::Disconnected { from, to } => {
26                write!(f, "transport link from {from} to {to} is disconnected")
27            }
28        }
29    }
30}
31
32impl core::error::Error for TransportError {}
33
34/// Engineered transport coverage contract.
35///
36/// A transport adapter is responsible for peer subscription, connectivity
37/// checks, and moving record batches. It is tested infrastructure, not a
38/// proof that any real network delivers packets.
39pub trait TransportAdapter<D> {
40    fn subscribe(&mut self, peer: u64);
41    fn is_subscribed(&self, peer: u64) -> bool;
42    fn set_connected(&mut self, a: u64, b: u64, connected: bool);
43    fn is_connected(&self, a: u64, b: u64) -> bool;
44    fn send(&mut self, from: u64, to: u64, records: Vec<Record<D>>) -> Result<(), TransportError>;
45    fn drain(&mut self, peer: u64) -> Vec<TransportEnvelope<D>>;
46}
47
48/// Deterministic in-memory adapter for coverage-contract fault campaigns.
49///
50/// This adapter can drop, duplicate, reorder, partition, and heal links. It is
51/// useful for CI and demos; it is not a real radio/network adapter.
52/// Packets queued before a partition remain pending until the link heals;
53/// draining a peer delivers only packets whose links are currently connected.
54/// All links, including self-links, start connected and can be partitioned.
55#[derive(Clone, Debug, PartialEq, Eq)]
56pub struct InMemoryTransport<D> {
57    subscribed: BTreeSet<u64>,
58    disconnected: BTreeSet<(u64, u64)>,
59    queue: Vec<TransportEnvelope<D>>,
60    dropped: Vec<TransportEnvelope<D>>,
61    drop_next: bool,
62    duplicate_next: bool,
63}
64
65impl<D> InMemoryTransport<D> {
66    pub fn new() -> Self {
67        InMemoryTransport {
68            subscribed: BTreeSet::new(),
69            disconnected: BTreeSet::new(),
70            queue: Vec::new(),
71            dropped: Vec::new(),
72            drop_next: false,
73            duplicate_next: false,
74        }
75    }
76
77    pub fn drop_next_send(&mut self) {
78        self.drop_next = true;
79    }
80
81    pub fn duplicate_next_send(&mut self) {
82        self.duplicate_next = true;
83    }
84
85    pub fn reverse_pending_for(&mut self, peer: u64) {
86        let mut peer_items = Vec::new();
87        let mut retained = Vec::new();
88        for envelope in self.queue.drain(..) {
89            if envelope.to == peer {
90                peer_items.push(envelope);
91            } else {
92                retained.push(envelope);
93            }
94        }
95        peer_items.reverse();
96        retained.extend(peer_items);
97        self.queue = retained;
98    }
99
100    pub fn pending_len(&self) -> usize {
101        self.queue.len()
102    }
103
104    pub fn dropped_len(&self) -> usize {
105        self.dropped.len()
106    }
107
108    /// Returns the number of dropped envelopes and clears them.
109    pub fn drain_dropped(&mut self) -> usize {
110        let count = self.dropped_len();
111        self.clear_dropped();
112        count
113    }
114
115    /// Discards all dropped envelopes without changing pending deliveries or faults.
116    pub fn clear_dropped(&mut self) {
117        self.dropped.clear();
118    }
119
120    fn link(a: u64, b: u64) -> (u64, u64) {
121        if a <= b {
122            (a, b)
123        } else {
124            (b, a)
125        }
126    }
127}
128
129impl<D> Default for InMemoryTransport<D> {
130    fn default() -> Self {
131        Self::new()
132    }
133}
134
135impl<D: Clone> TransportAdapter<D> for InMemoryTransport<D> {
136    fn subscribe(&mut self, peer: u64) {
137        self.subscribed.insert(peer);
138    }
139
140    fn is_subscribed(&self, peer: u64) -> bool {
141        self.subscribed.contains(&peer)
142    }
143
144    fn set_connected(&mut self, a: u64, b: u64, connected: bool) {
145        let link = Self::link(a, b);
146        if connected {
147            self.disconnected.remove(&link);
148        } else {
149            self.disconnected.insert(link);
150        }
151    }
152
153    fn is_connected(&self, a: u64, b: u64) -> bool {
154        !self.disconnected.contains(&Self::link(a, b))
155    }
156
157    fn send(&mut self, from: u64, to: u64, records: Vec<Record<D>>) -> Result<(), TransportError> {
158        if !self.is_subscribed(from) {
159            return Err(TransportError::NotSubscribed { peer: from });
160        }
161        if !self.is_subscribed(to) {
162            return Err(TransportError::NotSubscribed { peer: to });
163        }
164        if !self.is_connected(from, to) {
165            return Err(TransportError::Disconnected { from, to });
166        }
167
168        let envelope = TransportEnvelope { from, to, records };
169        if self.drop_next {
170            self.drop_next = false;
171            self.dropped.push(envelope);
172            return Ok(());
173        }
174
175        self.queue.push(envelope.clone());
176        if self.duplicate_next {
177            self.duplicate_next = false;
178            self.queue.push(envelope);
179        }
180        Ok(())
181    }
182
183    fn drain(&mut self, peer: u64) -> Vec<TransportEnvelope<D>> {
184        let mut incoming = Vec::new();
185        let mut retained = Vec::new();
186        for envelope in self.queue.drain(..) {
187            let connected = !self
188                .disconnected
189                .contains(&Self::link(envelope.from, envelope.to));
190            if envelope.to == peer && connected {
191                incoming.push(envelope);
192            } else {
193                retained.push(envelope);
194            }
195        }
196        self.queue = retained;
197        incoming
198    }
199}
200
201/// Queue missing records. Admission happens at the receiver; use
202/// [`anti_entropy`] when both replicas are available to observe refusals.
203pub fn queue_anti_entropy<D, T>(
204    transport: &mut T,
205    from: u64,
206    to: u64,
207    local: &EventLog<D>,
208    remote_version: &VersionVector,
209) -> Result<(), TransportError>
210where
211    D: Clone,
212    T: TransportAdapter<D>,
213{
214    let records = local.since(remote_version);
215    if records.is_empty() {
216        Ok(())
217    } else {
218        transport.send(from, to, records)
219    }
220}
221
222/// Queue repair and admit the deliveries available for `to`, returning every
223/// record ID and verdict with its sending peer. In particular, Invalid is
224/// reported again on every retry, with its original reason. Refusals are not
225/// acknowledgements and are not cached: fixing the receiver permits the next
226/// ordinary sync to admit the record. Dropped or partitioned packets retain
227/// the adapter's usual delivery semantics.
228///
229/// All available envelopes for `to` are admitted, including previously queued
230/// deliveries from other peers. An empty result means no delivery, not an
231/// acknowledgement. Callers must inspect the verdicts, especially Invalid and
232/// Collision. This helper requires access to the receiving replica; network
233/// adapters must return equivalent admission feedback from the remote endpoint.
234/// A Collision also raises the receiver's collision alarm; send its
235/// [`Replica::collision_report_bytes`] back so the offerer raises it too.
236#[must_use = "inspect sync admissions for Invalid and Collision refusals"]
237pub fn anti_entropy<C, T>(
238    transport: &mut T,
239    from: u64,
240    to: u64,
241    local: &EventLog<C::Delta>,
242    remote: &mut Replica<C>,
243) -> Result<Vec<(u64, RecordId, Admission)>, TransportError>
244where
245    C: Crdt,
246    C::Delta: Clone + PartialEq,
247    T: TransportAdapter<C::Delta>,
248{
249    queue_anti_entropy(transport, from, to, local, remote.version())?;
250    let mut admissions = Vec::new();
251    for envelope in transport.drain(to) {
252        for record in envelope.records {
253            let id = record.id;
254            admissions.push((envelope.from, id, remote.admit(record)));
255        }
256    }
257    Ok(admissions)
258}
safemesh_crdt/transport.rs

safemesh_crdt/
transport.rs
