qualia_core_db/p2p/
sync_ops.rs

1//! **libp2p op-transfer sync** (T3.1 p2p backend) — the wire protocol + reference relay that actually
2//! carries synchronisation operations peer-to-peer over the existing libp2p request-response swarm.
3//!
4//! Distinct from [`super::protocol`]'s `QualiaRequest::Sync`, which is only an *authorisation handshake*
5//! (hop-count / gatekeeper / target-shapes) and transfers no operations. This module adds the missing
6//! piece: **`Publish` / `PullSince` frames that move opaque, already-signed operation bytes**, plus a
7//! [`SyncOpRelay`] responder store. Operations travel as opaque `Vec<u8>` frames (the sync layer that
8//! owns the `SyncOperation` type serialises/deserialises them), so this stays a **dumb pipe**: it never
9//! validates or trusts — all trust remains in the consuming node's fail-closed inbox, exactly as the
10//! HTTP relay transport does. Encryption + auth come from libp2p's noise handshake on the connection.
11//!
12//! Native-only (libp2p). The swarm-driving `SyncTransport` bridge that maps the blocking
13//! publish/pull trait onto this protocol is the next step; this is the tested wire + store foundation.
14
15#![cfg(not(target_arch = "wasm32"))]
16
17use async_trait::async_trait;
18use libp2p::futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
19use libp2p::request_response::Codec;
20use libp2p::StreamProtocol;
21use serde::{de::DeserializeOwned, Deserialize, Serialize};
22use std::io;
23use std::sync::{Arc, Mutex};
24
25/// The request-response stream protocol id for op transfer (distinct from `/qualia/crdt-sync/1.0.0`).
26pub const SYNC_OP_PROTOCOL: &str = "/qualia/sync-ops/1.0.0";
27
28/// A hard cap on a single frame's encoded size, so a hostile peer cannot force an unbounded allocation.
29const MAX_FRAME_BYTES: usize = 64 * 1024 * 1024;
30
31/// A request in the op-transfer protocol. `op_frames` are opaque, already-signed operation bytes.
32#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
33pub enum SyncOpRequest {
34    /// Offer operations to the peer. Idempotent at the relay (dedup by frame content).
35    Publish { op_frames: Vec<Vec<u8>> },
36    /// Ask for operations the relay holds after `cursor` (`0` = from the start).
37    PullSince { cursor: u64 },
38}
39
40/// A response in the op-transfer protocol.
41#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
42pub enum SyncOpResponse {
43    /// How many *new* (non-duplicate) frames the publish added.
44    Published { accepted: u64 },
45    /// The frames after the requested cursor, plus the cursor to use next time.
46    Pulled {
47        op_frames: Vec<Vec<u8>>,
48        next_cursor: u64,
49    },
50}
51
52/// A reference op relay/store — append-only, dedup by frame content, cursor = position. The responder
53/// side of the protocol. A **dumb pipe**: it stores and serves opaque frames and trusts nothing.
54/// Cloning yields another handle onto the **same** store (so a node can both serve and be queried).
55#[derive(Clone, Default)]
56pub struct SyncOpRelay {
57    frames: Arc<Mutex<Vec<Vec<u8>>>>,
58}
59
60impl SyncOpRelay {
61    pub fn new() -> Self {
62        Self::default()
63    }
64
65    /// Add frames, skipping any already present (dedup by content). Returns the number newly added.
66    pub fn publish(&self, op_frames: &[Vec<u8>]) -> u64 {
67        let mut store = self.frames.lock().expect("relay lock");
68        let mut accepted = 0;
69        for f in op_frames {
70            if !store.iter().any(|e| e == f) {
71                store.push(f.clone());
72                accepted += 1;
73            }
74        }
75        accepted
76    }
77
78    /// The frames after `cursor` (in store order) plus the next cursor. Re-pulling is safe.
79    pub fn pull_since(&self, cursor: u64) -> (Vec<Vec<u8>>, u64) {
80        let store = self.frames.lock().expect("relay lock");
81        let start = (cursor as usize).min(store.len());
82        (store[start..].to_vec(), store.len() as u64)
83    }
84
85    /// Number of distinct frames held.
86    pub fn len(&self) -> usize {
87        self.frames.lock().map(|v| v.len()).unwrap_or(0)
88    }
89
90    pub fn is_empty(&self) -> bool {
91        self.len() == 0
92    }
93
94    /// Serve one request against the store — the responder's request→response handler.
95    pub fn handle(&self, req: SyncOpRequest) -> SyncOpResponse {
96        match req {
97            SyncOpRequest::Publish { op_frames } => SyncOpResponse::Published {
98                accepted: self.publish(&op_frames),
99            },
100            SyncOpRequest::PullSince { cursor } => {
101                let (op_frames, next_cursor) = self.pull_since(cursor);
102                SyncOpResponse::Pulled {
103                    op_frames,
104                    next_cursor,
105                }
106            }
107        }
108    }
109}
110
111/// The libp2p request-response wire codec for op transfer: length-prefixed plain-CBOR (ciborium). Mirrors
112/// [`super::protocol::QualiaSyncCodec`]'s framing (4-byte big-endian length + body), without the Q42
113/// CBOR-LD term-compaction (the payload is opaque operation bytes, so there is nothing to term-compact).
114#[derive(Clone, Default)]
115pub struct SyncOpCodec;
116
117async fn read_frame<T, V>(io: &mut T) -> io::Result<V>
118where
119    T: AsyncRead + Unpin + Send,
120    V: DeserializeOwned,
121{
122    let mut len_buf = [0u8; 4];
123    io.read_exact(&mut len_buf).await?;
124    let len = u32::from_be_bytes(len_buf) as usize;
125    if len > MAX_FRAME_BYTES {
126        return Err(io::Error::new(
127            io::ErrorKind::InvalidData,
128            "sync-op frame too large",
129        ));
130    }
131    let mut buf = vec![0u8; len];
132    io.read_exact(&mut buf).await?;
133    ciborium::from_reader(&buf[..])
134        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))
135}
136
137async fn write_frame<T, V>(io: &mut T, v: &V) -> io::Result<()>
138where
139    T: AsyncWrite + Unpin + Send,
140    V: Serialize,
141{
142    let mut buf = Vec::new();
143    ciborium::into_writer(v, &mut buf)
144        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?;
145    io.write_all(&(buf.len() as u32).to_be_bytes()).await?;
146    io.write_all(&buf).await?;
147    Ok(())
148}
149
150#[async_trait]
151impl Codec for SyncOpCodec {
152    type Protocol = StreamProtocol;
153    type Request = SyncOpRequest;
154    type Response = SyncOpResponse;
155
156    async fn read_request<T>(&mut self, _: &StreamProtocol, io: &mut T) -> io::Result<SyncOpRequest>
157    where
158        T: AsyncRead + Unpin + Send,
159    {
160        read_frame(io).await
161    }
162
163    async fn read_response<T>(
164        &mut self,
165        _: &StreamProtocol,
166        io: &mut T,
167    ) -> io::Result<SyncOpResponse>
168    where
169        T: AsyncRead + Unpin + Send,
170    {
171        read_frame(io).await
172    }
173
174    async fn write_request<T>(
175        &mut self,
176        _: &StreamProtocol,
177        io: &mut T,
178        req: SyncOpRequest,
179    ) -> io::Result<()>
180    where
181        T: AsyncWrite + Unpin + Send,
182    {
183        write_frame(io, &req).await
184    }
185
186    async fn write_response<T>(
187        &mut self,
188        _: &StreamProtocol,
189        io: &mut T,
190        res: SyncOpResponse,
191    ) -> io::Result<()>
192    where
193        T: AsyncWrite + Unpin + Send,
194    {
195        write_frame(io, &res).await
196    }
197}
198
199#[cfg(test)]
200mod tests {
201    use super::*;
202
203    #[test]
204    fn relay_dedups_publish_and_pulls_from_cursor() {
205        let relay = SyncOpRelay::new();
206        assert_eq!(relay.publish(&[b"a".to_vec(), b"b".to_vec()]), 2);
207        // Re-publishing 'a' is idempotent.
208        assert_eq!(relay.publish(&[b"a".to_vec()]), 0);
209        assert_eq!(relay.len(), 2);
210
211        let (frames, cursor) = relay.pull_since(0);
212        assert_eq!(frames, vec![b"a".to_vec(), b"b".to_vec()]);
213        assert_eq!(cursor, 2);
214        // Only what's new after the cursor.
215        assert_eq!(relay.publish(&[b"c".to_vec()]), 1);
216        let (fresh, next) = relay.pull_since(2);
217        assert_eq!(fresh, vec![b"c".to_vec()]);
218        assert_eq!(next, 3);
219    }
220
221    #[test]
222    fn handle_maps_requests_to_responses() {
223        let relay = SyncOpRelay::new();
224        assert_eq!(
225            relay.handle(SyncOpRequest::Publish {
226                op_frames: vec![b"x".to_vec()]
227            }),
228            SyncOpResponse::Published { accepted: 1 }
229        );
230        assert_eq!(
231            relay.handle(SyncOpRequest::PullSince { cursor: 0 }),
232            SyncOpResponse::Pulled {
233                op_frames: vec![b"x".to_vec()],
234                next_cursor: 1
235            }
236        );
237    }
238
239    /// The wire payload round-trips losslessly (the ciborium (de)serialization the codec applies before
240    /// the 4-byte length prefix). The framing is identical to the proven `QualiaSyncCodec`.
241    #[test]
242    fn wire_payload_roundtrips_losslessly() {
243        for req in [
244            SyncOpRequest::Publish {
245                op_frames: vec![b"op-1".to_vec(), b"op-2".to_vec()],
246            },
247            SyncOpRequest::PullSince { cursor: 42 },
248        ] {
249            let mut buf = Vec::new();
250            ciborium::into_writer(&req, &mut buf).unwrap();
251            let back: SyncOpRequest = ciborium::from_reader(&buf[..]).unwrap();
252            assert_eq!(back, req);
253        }
254        for res in [
255            SyncOpResponse::Published { accepted: 3 },
256            SyncOpResponse::Pulled {
257                op_frames: vec![b"op-1".to_vec()],
258                next_cursor: 7,
259            },
260        ] {
261            let mut buf = Vec::new();
262            ciborium::into_writer(&res, &mut buf).unwrap();
263            let back: SyncOpResponse = ciborium::from_reader(&buf[..]).unwrap();
264            assert_eq!(back, res);
265        }
266    }
267}
qualia_core_db/p2p/sync_ops.rs

qualia_core_db/p2p/
sync_ops.rs
