qualia_client_core/wellfair/
sync_transport.rs

1//! Sync **transport** (T3.1) — moves [`SyncOperation`]s between this node and a peer/relay.
2//!
3//! The transport is a **dumb pipe**: it neither validates nor trusts operations. All trust lives in
4//! the inbox's fail-closed [`validate_operation`](super::sync_protocol::validate_operation) —
5//! a hostile relay or peer can therefore only ever cause *rejections* at admission, never the
6//! admission of bad data. Convergence is guaranteed by
7//! [`merge_operations`](super::sync_protocol::merge_operations) (add-wins by operation id), so
8//! duplicate / reordered / replayed frames converge to the same validated set.
9//!
10//! Two backends:
11//! - [`InMemoryRelay`] — a shared, in-process op store (dedup by operation id). Clone it to get a
12//!   second handle onto the same relay; two nodes sharing one relay exchange ops. This is the
13//!   reference transport and the one the convergence / hostile-peer tests run over.
14//! - [`HttpRelayTransport`] (native) — a `reqwest::blocking` client that POSTs to `"{base}/sync/publish"`
15//!   and GETs `"{base}/sync/pull?since={cursor}"`. Its server counterpart is
16//!   [`super::sync_relay_server::SyncRelayServer`].
17
18use super::sync_protocol::SyncOperation;
19use std::sync::{Arc, Mutex};
20
21/// Moves operations to/from a peer or relay. A dumb pipe — validation is the inbox's job.
22pub trait SyncTransport {
23    /// Publish local operations to the relay/peer. Must be idempotent at the relay (dedup by
24    /// operation id), so re-publishing a queued op is harmless.
25    fn publish(&self, ops: &[SyncOperation]) -> Result<(), String>;
26
27    /// Pull operations the relay holds after cursor `since` (`0` = from the start). Returns them in
28    /// relay order. Re-pulling is safe: the inbox dedups by operation id on admission.
29    fn pull(&self, since: u64) -> Result<Vec<SyncOperation>, String>;
30}
31
32/// A shared in-memory relay — a dumb op store (append-only, dedup by operation id). Cloning yields
33/// another handle onto the **same** underlying store, so two nodes can rendezvous through one relay.
34#[derive(Clone, Default)]
35pub struct InMemoryRelay {
36    inner: Arc<Mutex<Vec<SyncOperation>>>,
37}
38
39impl InMemoryRelay {
40    pub fn new() -> Self {
41        Self::default()
42    }
43
44    /// Number of distinct operations the relay holds.
45    pub fn len(&self) -> usize {
46        self.inner.lock().map(|v| v.len()).unwrap_or(0)
47    }
48
49    pub fn is_empty(&self) -> bool {
50        self.len() == 0
51    }
52}
53
54impl SyncTransport for InMemoryRelay {
55    fn publish(&self, ops: &[SyncOperation]) -> Result<(), String> {
56        let mut store = self
57            .inner
58            .lock()
59            .map_err(|_| "relay lock poisoned".to_string())?;
60        for op in ops {
61            if !store.iter().any(|e| e.operation_id == op.operation_id) {
62                store.push(op.clone());
63            }
64        }
65        Ok(())
66    }
67
68    fn pull(&self, since: u64) -> Result<Vec<SyncOperation>, String> {
69        let store = self
70            .inner
71            .lock()
72            .map_err(|_| "relay lock poisoned".to_string())?;
73        let start = (since as usize).min(store.len());
74        Ok(store[start..].to_vec())
75    }
76}
77
78/// The wire body for `/sync/publish` and the response body for `/sync/pull`.
79#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
80pub struct SyncOpsBody {
81    pub ops: Vec<SyncOperation>,
82}
83
84/// An HTTP relay transport (native). POSTs a [`SyncOpsBody`] to `"{base}/sync/publish"` and GETs
85/// `"{base}/sync/pull?since={cursor}"`. Mirrors the crate's `reqwest::blocking` chat-relay pattern.
86#[cfg(not(target_arch = "wasm32"))]
87pub struct HttpRelayTransport {
88    base_url: String,
89    timeout: std::time::Duration,
90}
91
92#[cfg(not(target_arch = "wasm32"))]
93impl HttpRelayTransport {
94    pub fn new(base_url: impl Into<String>) -> Self {
95        Self {
96            base_url: base_url.into().trim_end_matches('/').to_string(),
97            timeout: std::time::Duration::from_secs(8),
98        }
99    }
100
101    fn client(&self) -> Result<reqwest::blocking::Client, String> {
102        reqwest::blocking::Client::builder()
103            .timeout(self.timeout)
104            .build()
105            .map_err(|e| e.to_string())
106    }
107}
108
109#[cfg(not(target_arch = "wasm32"))]
110impl SyncTransport for HttpRelayTransport {
111    fn publish(&self, ops: &[SyncOperation]) -> Result<(), String> {
112        let url = format!("{}/sync/publish", self.base_url);
113        let body = SyncOpsBody { ops: ops.to_vec() };
114        let resp = self
115            .client()?
116            .post(&url)
117            .json(&body)
118            .send()
119            .map_err(|e| e.to_string())?;
120        if !resp.status().is_success() {
121            return Err(format!("relay publish failed: HTTP {}", resp.status()));
122        }
123        Ok(())
124    }
125
126    fn pull(&self, since: u64) -> Result<Vec<SyncOperation>, String> {
127        let url = format!("{}/sync/pull?since={since}", self.base_url);
128        let resp = self.client()?.get(&url).send().map_err(|e| e.to_string())?;
129        if !resp.status().is_success() {
130            return Err(format!("relay pull failed: HTTP {}", resp.status()));
131        }
132        let parsed: SyncOpsBody = resp.json().map_err(|e| e.to_string())?;
133        Ok(parsed.ops)
134    }
135}
136
137/// A **libp2p** `SyncTransport` (native) — a noise-encrypted request-response pipe to a single peer or
138/// relay. It wraps the core-db [`BlockingSyncClient`](qualia_core_db::p2p::sync_node::BlockingSyncClient):
139/// each [`SyncOperation`] is serialized to a CBOR op frame on publish and decoded back on pull, so the
140/// wire carries only opaque signed-op bytes. A dumb pipe like the in-memory and HTTP backends — it never
141/// validates or trusts. Undecodable inbound frames are **skipped** (they could never be admitted anyway;
142/// the fail-closed inbox is the trust boundary), so a hostile peer injecting junk cannot break sync for
143/// well-formed operations.
144#[cfg(not(target_arch = "wasm32"))]
145pub struct Libp2pSyncTransport {
146    client: qualia_core_db::p2p::sync_node::BlockingSyncClient,
147}
148
149#[cfg(not(target_arch = "wasm32"))]
150impl Libp2pSyncTransport {
151    /// Connect to `peer_addr` (a libp2p multiaddr, e.g. `/ip4/1.2.3.4/tcp/4001`) identified by `peer_id`
152    /// (base58). The connection is established lazily on the first publish/pull.
153    pub fn connect(peer_id: &str, peer_addr: &str) -> Result<Self, String> {
154        let client = qualia_core_db::p2p::sync_node::BlockingSyncClient::connect(
155            qualia_core_db::p2p::sync_ops::SyncOpRelay::new(),
156            peer_id,
157            peer_addr,
158        )?;
159        Ok(Self { client })
160    }
161
162    /// This node's own libp2p peer id (base58) — so a peer can be told how to reach us back.
163    pub fn local_peer_id(&self) -> String {
164        self.client.local_peer_id().to_string()
165    }
166}
167
168/// Serialize an operation to its opaque CBOR wire frame.
169#[cfg(not(target_arch = "wasm32"))]
170fn op_to_frame(op: &SyncOperation) -> Result<Vec<u8>, String> {
171    let mut buf = Vec::new();
172    ciborium::into_writer(op, &mut buf).map_err(|e| format!("encode operation: {e}"))?;
173    Ok(buf)
174}
175
176#[cfg(not(target_arch = "wasm32"))]
177impl SyncTransport for Libp2pSyncTransport {
178    fn publish(&self, ops: &[SyncOperation]) -> Result<(), String> {
179        if ops.is_empty() {
180            return Ok(());
181        }
182        let frames = ops.iter().map(op_to_frame).collect::<Result<Vec<_>, _>>()?;
183        self.client.publish_frames(frames)?;
184        Ok(())
185    }
186
187    fn pull(&self, since: u64) -> Result<Vec<SyncOperation>, String> {
188        let (frames, _next_cursor) = self.client.pull_frames(since)?;
189        // Dumb pipe: decode what we can, skip junk (undecodable frames can never be valid operations,
190        // and the inbox would reject them regardless).
191        Ok(frames
192            .iter()
193            .filter_map(|f| ciborium::from_reader::<SyncOperation, _>(&f[..]).ok())
194            .collect())
195    }
196}
197
198#[cfg(test)]
199mod tests {
200    use super::*;
201    use crate::wellfair::sync_protocol::{AdmitOutcome, SyncInbox, SyncOperation};
202
203    fn signed(id: &str, kind: &str, summary: &str, lamport: u64) -> SyncOperation {
204        SyncOperation::new(
205            id,
206            format!("urn:wellfair:{kind}:{id}"),
207            kind,
208            "did:wf:remote",
209            "Restricted",
210            summary,
211            lamport,
212            1_700_000_000,
213        )
214        .with_signature("deadbeef")
215    }
216
217    #[test]
218    fn in_memory_relay_dedups_and_pulls_from_cursor() {
219        let relay = InMemoryRelay::new();
220        relay
221            .publish(&[signed("a", "ledger_entry", "1", 1)])
222            .unwrap();
223        relay
224            .publish(&[signed("b", "ledger_entry", "2", 2)])
225            .unwrap();
226        // Re-publishing 'a' is idempotent at the relay.
227        relay
228            .publish(&[signed("a", "ledger_entry", "1", 1)])
229            .unwrap();
230        assert_eq!(relay.len(), 2);
231
232        // Cursor: pull everything, then only what's new.
233        assert_eq!(relay.pull(0).unwrap().len(), 2);
234        assert_eq!(relay.pull(2).unwrap().len(), 0);
235        relay
236            .publish(&[signed("c", "ledger_entry", "3", 3)])
237            .unwrap();
238        let fresh = relay.pull(2).unwrap();
239        assert_eq!(fresh.len(), 1);
240        assert_eq!(fresh[0].operation_id, "c");
241    }
242
243    #[test]
244    fn two_nodes_converge_over_one_relay() {
245        // Node A and Node B each publish a disjoint op to a shared relay; after both pull and admit,
246        // their validated sets are identical (convergence).
247        let relay = InMemoryRelay::new();
248        let dir_a = tempfile::tempdir().unwrap();
249        let dir_b = tempfile::tempdir().unwrap();
250        let inbox_a = SyncInbox::open(dir_a.path()).unwrap();
251        let inbox_b = SyncInbox::open(dir_b.path()).unwrap();
252
253        let a_op = signed("a1", "contribution", "a", 1);
254        let b_op = signed("b1", "contribution", "b", 2);
255        // Each node admits its own op locally + publishes it.
256        inbox_a.admit(&a_op, 1).unwrap();
257        relay.publish(&[a_op.clone()]).unwrap();
258        inbox_b.admit(&b_op, 1).unwrap();
259        relay.publish(&[b_op.clone()]).unwrap();
260
261        // Each pulls the whole relay and admits.
262        for op in relay.pull(0).unwrap() {
263            inbox_a.admit(&op, 2).unwrap();
264            inbox_b.admit(&op, 2).unwrap();
265        }
266
267        let set_a = inbox_a.validated_operations().unwrap();
268        let set_b = inbox_b.validated_operations().unwrap();
269        assert_eq!(
270            set_a, set_b,
271            "nodes must converge to the same validated set"
272        );
273        assert_eq!(set_a.len(), 2);
274    }
275
276    #[test]
277    fn hostile_relay_ops_are_rejected_not_admitted() {
278        // A hostile peer publishes malformed ops through the relay. The inbox rejects every one
279        // fail-closed; the validated set stays empty.
280        let relay = InMemoryRelay::new();
281        let dir = tempfile::tempdir().unwrap();
282        let inbox = SyncInbox::open(dir.path()).unwrap();
283
284        // Unsigned.
285        let mut unsigned = signed("h1", "ledger_entry", "x", 1);
286        unsigned.signature = None;
287        // Tampered content hash.
288        let mut tampered = signed("h2", "ledger_entry", "orig", 1);
289        tampered.payload_summary = "changed".into();
290        // Wrong protocol version.
291        let mut bad_ver = signed("h3", "ledger_entry", "x", 1);
292        bad_ver.protocol_version = 99;
293        // Classified lane (must never traverse the ordinary inbox).
294        let mut classified = signed("h4", "sanctuary_note", "x", 1);
295        classified.sensitivity = "Classified".into();
296
297        relay
298            .publish(&[unsigned, tampered, bad_ver, classified])
299            .unwrap();
300
301        let mut rejected = 0;
302        for op in relay.pull(0).unwrap() {
303            if matches!(inbox.admit(&op, 5).unwrap(), AdmitOutcome::Rejected(_)) {
304                rejected += 1;
305            }
306        }
307        assert_eq!(rejected, 4, "all hostile ops must be rejected");
308        assert!(inbox.validated_operations().unwrap().is_empty());
309    }
310
311    #[test]
312    fn replayed_pull_is_idempotent() {
313        // Pulling and admitting the same relay contents twice never double-applies.
314        let relay = InMemoryRelay::new();
315        let dir = tempfile::tempdir().unwrap();
316        let inbox = SyncInbox::open(dir.path()).unwrap();
317        relay
318            .publish(&[signed("r1", "ledger_entry", "x", 1)])
319            .unwrap();
320
321        for _ in 0..3 {
322            for op in relay.pull(0).unwrap() {
323                inbox.admit(&op, 9).unwrap();
324            }
325        }
326        assert_eq!(inbox.validated_operations().unwrap().len(), 1);
327    }
328
329    #[test]
330    fn partition_then_rejoin_converges() {
331        // Two relays (a partition). Each node works its own side, then they exchange full contents
332        // (the rejoin) and both admit everything — converging.
333        let left = InMemoryRelay::new();
334        let right = InMemoryRelay::new();
335        left.publish(&[signed("L", "contribution", "l", 1)])
336            .unwrap();
337        right
338            .publish(&[signed("R", "contribution", "r", 2)])
339            .unwrap();
340
341        let dir_a = tempfile::tempdir().unwrap();
342        let dir_b = tempfile::tempdir().unwrap();
343        let inbox_a = SyncInbox::open(dir_a.path()).unwrap();
344        let inbox_b = SyncInbox::open(dir_b.path()).unwrap();
345
346        // Rejoin: merge both partitions' contents into one exchange set.
347        let mut all = left.pull(0).unwrap();
348        all.extend(right.pull(0).unwrap());
349        for op in &all {
350            inbox_a.admit(op, 1).unwrap();
351            inbox_b.admit(op, 1).unwrap();
352        }
353        assert_eq!(
354            inbox_a.validated_operations().unwrap(),
355            inbox_b.validated_operations().unwrap()
356        );
357        assert_eq!(inbox_a.validated_operations().unwrap().len(), 2);
358    }
359
360    /// **End-to-end over libp2p:** real signed `SyncOperation`s travel from the transport, through the
361    /// noise-encrypted request-response wire, into a responder's relay, are pulled back (surviving the
362    /// CBOR round-trip byte-for-byte), and admit into a fail-closed inbox as `Validated`. This proves the
363    /// last piece of T3.1 — the `SyncOperation`↔frame serialization + the blocking libp2p transport.
364    #[cfg(not(target_arch = "wasm32"))]
365    #[test]
366    fn libp2p_transport_round_trips_real_operations_and_inbox_admits() {
367        use qualia_core_db::p2p::sync_node::Libp2pSyncNode;
368        use qualia_core_db::p2p::sync_ops::SyncOpRelay;
369
370        // Responder A: listens and serves its relay on its own runtime; keep `a` + `rt_a` alive so A's
371        // event loop stays up for the whole test.
372        let rt_a = tokio::runtime::Builder::new_multi_thread()
373            .worker_threads(1)
374            .enable_all()
375            .build()
376            .unwrap();
377        let relay_a = SyncOpRelay::new();
378        let a = {
379            let _guard = rt_a.enter();
380            Libp2pSyncNode::spawn(relay_a.clone())
381        };
382        let a_addr = rt_a
383            .block_on(a.listen("/ip4/127.0.0.1/tcp/0"))
384            .expect("A listen");
385
386        // The libp2p SyncTransport (the spoke) dials A.
387        let transport = Libp2pSyncTransport::connect(&a.peer_id.to_string(), &a_addr.to_string())
388            .expect("connect");
389
390        // Publish two real signed operations, then pull them back.
391        let ops = vec![
392            signed("t1", "contribution", "alpha", 1),
393            signed("t2", "contribution", "beta", 2),
394        ];
395        transport.publish(&ops).expect("publish");
396        assert_eq!(relay_a.len(), 2, "A's relay holds both op frames");
397
398        let pulled = transport.pull(0).expect("pull");
399        assert_eq!(
400            pulled, ops,
401            "operations round-trip losslessly through CBOR + libp2p"
402        );
403
404        // The pulled operations admit into a fresh fail-closed inbox as Validated.
405        let dir = tempfile::tempdir().unwrap();
406        let inbox = SyncInbox::open(dir.path()).unwrap();
407        for op in &pulled {
408            assert_eq!(inbox.admit(op, 1).unwrap(), AdmitOutcome::Validated);
409        }
410        assert_eq!(inbox.validated_operations().unwrap().len(), 2);
411    }
412}
qualia_client_core/wellfair/sync_transport.rs

qualia_client_core/wellfair/
sync_transport.rs
