diff --git a/Cargo.lock b/Cargo.lock index 9668308..cbb3ea4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -314,6 +314,15 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "bincode" +version = "1.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f45e9417d87227c7a56d22e471c6206462cba514c7590c09aff4cf6d1ddcad" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "2.13.1" @@ -484,10 +493,15 @@ dependencies = [ name = "chronogrid_net" version = "0.1.0" dependencies = [ + "async-trait", + "bincode", + "chacha20poly1305", "ed25519-dalek", "futures", + "generic-array", "hex", "libp2p", + "lz4_flex", "rand 0.8.7", "serde", "serde_json", @@ -1680,6 +1694,7 @@ dependencies = [ "libp2p-noise", "libp2p-quic", "libp2p-relay", + "libp2p-request-response", "libp2p-swarm", "libp2p-tcp", "libp2p-upnp", @@ -1974,6 +1989,26 @@ dependencies = [ "web-time", ] +[[package]] +name = "libp2p-request-response" +version = "0.27.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1356c9e376a94a75ae830c42cdaea3d4fe1290ba409a22c809033d1b7dcab0a6" +dependencies = [ + "async-trait", + "futures", + "futures-bounded", + "futures-timer", + "libp2p-core", + "libp2p-identity", + "libp2p-swarm", + "rand 0.8.7", + "smallvec", + "tracing", + "void", + "web-time", +] + [[package]] name = "libp2p-swarm" version = "0.45.1" @@ -2146,6 +2181,15 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" +[[package]] +name = "lz4_flex" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "373f5eceeeab7925e0c1098212f2fbc4d416adec9d35051a6ab251e824c1854a" +dependencies = [ + "twox-hash", +] + [[package]] name = "match-lookup" version = "0.1.2" @@ -3638,6 +3682,12 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "twox-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8464ec13c3691491391d9fce00f6416c9a48e46972f72d7865688be2080192c9" + [[package]] name = "typenum" version = "1.20.1" diff --git a/DECISION_LOG.md b/DECISION_LOG.md index 1d507ba..826cfe5 100644 --- a/DECISION_LOG.md +++ b/DECISION_LOG.md @@ -110,6 +110,21 @@ - `KadEvent::RoutingUpdated` uses private libp2p-internal types; routing-table tracking tested via `#[cfg(test)] insert_peer_for_test()` helper; full coverage deferred to Task 2.5 integration test. - Full ADR at `docs/ARCHITECTURE/ADR/0009-net-transport-discovery-layer.md`. +## [2026-08-22] - ADL-010: chronogrid_net Crate — CRDT Delta Sync Protocol (Task 2.4) + +- **Status**: Accepted +- **Context**: Task 2.4 required implementing the CRDT delta exchange protocol over libp2p in `chronogrid_net`, building on the Ed25519 `NodeIdentity`, `SwarmKey`, and transport/discovery layers from Tasks 2.2 and 2.3. +- **Decision**: + - Added new `sync` module with 7 public types: `VectorClock`, `DeltaPacket`, `SyncCodec`, `SyncProtocol`, `GossipEvent`, `GossipsubBroadcast`, `NetEvent`. + - `VectorClock` uses `HashMap` with lexicographic HLC comparison — no numeric parsing required since the `--` format preserves causal ordering under string comparison. + - `DeltaPacket` wire format: `bincode` → `lz4_flex` compress → `ChaCha20-Poly1305` encrypt with 12-byte random nonce prepended. AAD = `b"chronogrid-delta"` binds ciphertext to protocol context. + - `SyncProtocol` uses `#[derive(NetworkBehaviour)]` wrapping `libp2p::request_response::Behaviour`. `SyncCodec` implements the `Codec` trait via `#[async_trait]` (required by libp2p-request-response 0.27.0) using `futures::AsyncRead`/`AsyncWrite` (not `tokio::io`). + - `GossipsubBroadcast` subscribes to topic `chronogrid/state-changes/1` with SHA-256 content-addressed message deduplication and `ValidationMode::Strict`. + - `NetEvent` enum: `DeltasReceived { from_peer, deltas }`, `SyncComplete { peer_id }`, `SyncError { peer_id, reason }`, `GossipReceived { entity_type, entity_id, hlc_timestamp, origin_peer }`. + - Added dependencies: `bincode = "1"`, `lz4_flex = "0.11"`, `chacha20poly1305 = "0.10"`, `generic-array = "0.14"`, `async-trait = "0.1"`, libp2p `"request-response"` feature. + - 100 unit tests pass + 5 doc-tests pass (35 new sync tests + 65 from Tasks 2.2/2.3). + - Full ADR at `docs/ARCHITECTURE/ADR/0010-net-crdt-delta-sync-protocol.md`. + ## [2026-08-22] - ADL-006: Architecture Review Board Assessment & Master Development Plan v2 - **Status**: Accepted diff --git a/crates/chronogrid_net/Cargo.toml b/crates/chronogrid_net/Cargo.toml index 248ab0a..64a9a85 100644 --- a/crates/chronogrid_net/Cargo.toml +++ b/crates/chronogrid_net/Cargo.toml @@ -35,6 +35,7 @@ libp2p = { version = "0.54", features = [ "kad", "identify", "relay", + "request-response", "tokio", "macros", ] } @@ -44,6 +45,16 @@ tokio = { version = "1", features = ["full"] } futures = "0.3" # Logging / tracing tracing = "0.1" +# Serialization for DeltaPacket (bincode binary format) +bincode = "1" +# LZ4 compression for delta payloads +lz4_flex = "0.11" +# ChaCha20-Poly1305 AEAD encryption for delta payloads +chacha20poly1305 = "0.10" +# Generic array for AEAD key/nonce types +generic-array = "0.14" +# async-trait for implementing libp2p request_response Codec +async-trait = "0.1" [dev-dependencies] tempfile = "3" diff --git a/crates/chronogrid_net/src/lib.rs b/crates/chronogrid_net/src/lib.rs index 5252e24..ad8684c 100644 --- a/crates/chronogrid_net/src/lib.rs +++ b/crates/chronogrid_net/src/lib.rs @@ -15,6 +15,7 @@ //! | [`peer_event`] | `PeerEvent` enum — unified peer lifecycle events | //! | [`transport`] | TCP + Noise XK + Yamux transport builder; keypair conversion | //! | [`discovery`] | `MdnsDiscovery` (LAN) and `KademliaDht` (WAN) peer discovery | +//! | [`sync`] | `VectorClock`, `DeltaPacket`, `SyncProtocol`, `GossipsubBroadcast`, `NetEvent` | //! //! ## Quick Start //! @@ -23,6 +24,7 @@ //! use chronogrid_net::identity::load_or_create_identity; //! use chronogrid_net::swarm_key::SwarmKey; //! use chronogrid_net::transport::{build_transport, TransportConfig}; +//! use chronogrid_net::sync::{VectorClock, DeltaPacket, SyncProtocol}; //! //! // Load or generate a persistent node identity. //! let identity = load_or_create_identity(Path::new("/var/lib/chronogrid/identity.json")) @@ -40,6 +42,10 @@ //! // Build the transport config (used by SwarmManager in Task 2.5). //! let config = TransportConfig::from_swarm_key(&swarm_key); //! println!("User-agent: {}", config.user_agent()); +//! +//! // Build a vector clock for delta sync. +//! let mut clock = VectorClock::new(); +//! clock.update("entity-uuid-1", "1724000000000-0001-ab12"); //! ``` pub mod discovery; @@ -47,6 +53,7 @@ pub mod error; pub mod identity; pub mod peer_event; pub mod swarm_key; +pub mod sync; pub mod transport; // Re-export the most commonly used types at the crate root for convenience. @@ -54,4 +61,5 @@ pub use error::{NetError, NetResult}; pub use identity::{NodeId, NodeIdentity}; pub use peer_event::PeerEvent; pub use swarm_key::SwarmKey; +pub use sync::{DeltaPacket, GossipEvent, NetEvent, SyncProtocol, VectorClock}; pub use transport::{TransportConfig, build_transport, peer_id_from_identity}; diff --git a/crates/chronogrid_net/src/sync.rs b/crates/chronogrid_net/src/sync.rs new file mode 100644 index 0000000..aa35d0e --- /dev/null +++ b/crates/chronogrid_net/src/sync.rs @@ -0,0 +1,1234 @@ +//! CRDT delta synchronization protocol for ChronoGrid P2P. +//! +//! This module implements the full delta-exchange pipeline between peers: +//! +//! 1. **[`VectorClock`]** — tracks per-entity HLC head timestamps so that two +//! peers can compute exactly which changesets the other is missing. +//! +//! 2. **[`DeltaPacket`]** — a serializable, compressed, and encrypted container +//! for a batch of Automerge CRDT changesets. Wire format: +//! ```text +//! [ bincode-serialized DeltaPacket ] +//! → LZ4 compressed +//! → ChaCha20-Poly1305 encrypted (key = SwarmKey bytes) +//! ``` +//! +//! 3. **[`SyncProtocol`]** — a libp2p [`NetworkBehaviour`] that implements the +//! request/response vector-clock exchange and delta streaming protocol. +//! On peer connect it sends the local `VectorClock`; on receiving a remote +//! clock it computes missing deltas and streams them back. +//! +//! 4. **[`GossipsubBroadcast`]** — wraps `libp2p::gossipsub::Behaviour` to +//! provide real-time state-change notifications across the cluster. +//! +//! 5. **[`NetEvent`]** — the unified event enum consumed by `SwarmManager` +//! (Task 2.5) to drive CRDT merges and UI notifications. +//! +//! # Integration with SwarmManager (Task 2.5) +//! +//! `SwarmManager` composes `SyncProtocol` and `GossipsubBroadcast` into a +//! single `#[derive(NetworkBehaviour)]` struct alongside `MdnsDiscovery` and +//! `KademliaDht`. This module provides the building blocks. + +use std::collections::HashMap; + +use async_trait::async_trait; +use libp2p::{ + PeerId, StreamProtocol, + gossipsub::{ + Behaviour as GossipBehaviour, + ConfigBuilder as GossipConfigBuilder, + Event as GossipBehaviourEvent, + IdentTopic, + Message as GossipMessage, + MessageAuthenticity, + MessageId, + ValidationMode, + }, + request_response::{ + self, + Behaviour as ReqResBehaviour, + Codec as ReqResCodec, + Config as ReqResConfig, + ProtocolSupport, + }, + swarm::NetworkBehaviour, +}; + +use chacha20poly1305::{ + AeadInPlace, ChaCha20Poly1305, KeyInit, + aead::generic_array::GenericArray, +}; +use serde::{Deserialize, Serialize}; + +use crate::{ + error::{NetError, NetResult}, + swarm_key::SwarmKey, +}; + +// --------------------------------------------------------------------------- +// VectorClock +// --------------------------------------------------------------------------- + +/// Per-entity HLC head-state tracker for delta computation. +/// +/// Each entry maps an `entity_id` (UUID string) to the latest HLC timestamp +/// string seen for that entity. When two peers exchange vector clocks, the +/// receiver can determine which entities it has that the sender is missing +/// (and vice versa) by comparing timestamps. +/// +/// # HLC timestamp format +/// +/// Timestamps follow the format produced by `chronogrid_core::crdt::Hlc`: +/// `--`, e.g. +/// `1724000000000-0001-ab12cd34`. +/// +/// Lexicographic ordering of this format is equivalent to causal ordering, +/// so string comparison is sufficient for "newer than" checks. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] +pub struct VectorClock { + /// Map from entity_id → latest HLC timestamp string. + entries: HashMap, +} + +impl VectorClock { + /// Create an empty `VectorClock`. + pub fn new() -> Self { + VectorClock { + entries: HashMap::new(), + } + } + + /// Update the clock entry for `entity_id` if `hlc_timestamp` is newer + /// than the currently recorded value (or if no entry exists yet). + /// + /// Returns `true` if the entry was updated, `false` if the existing + /// timestamp was already equal or newer. + pub fn update( + &mut self, + entity_id: impl Into, + hlc_timestamp: impl Into, + ) -> bool { + let id = entity_id.into(); + let ts = hlc_timestamp.into(); + match self.entries.get(&id) { + Some(existing) if existing.as_str() >= ts.as_str() => false, + _ => { + self.entries.insert(id, ts); + true + } + } + } + + /// Return the latest HLC timestamp for `entity_id`, or `None` if unknown. + pub fn get(&self, entity_id: &str) -> Option<&str> { + self.entries.get(entity_id).map(|s| s.as_str()) + } + + /// Return the number of entities tracked by this clock. + pub fn len(&self) -> usize { + self.entries.len() + } + + /// Return `true` if the clock tracks no entities. + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } + + /// Compute the set of entity IDs that `other` has at a *newer* timestamp + /// than `self` (i.e., entities that `self` is missing or behind on). + /// + /// Returns a `Vec` of entity IDs for which `other` has newer data. + pub fn missing_from(&self, other: &VectorClock) -> Vec { + other + .entries + .iter() + .filter_map(|(id, other_ts)| { + match self.entries.get(id) { + // We have this entity but our timestamp is older. + Some(our_ts) if our_ts.as_str() < other_ts.as_str() => Some(id.clone()), + // We don't have this entity at all. + None => Some(id.clone()), + // We are equal or newer — no delta needed. + _ => None, + } + }) + .collect() + } + + /// Merge another `VectorClock` into `self`, taking the maximum timestamp + /// for each entity. + pub fn merge(&mut self, other: &VectorClock) { + for (id, ts) in &other.entries { + self.update(id.clone(), ts.clone()); + } + } + + /// Return an iterator over `(entity_id, hlc_timestamp)` pairs. + pub fn iter(&self) -> impl Iterator { + self.entries.iter().map(|(k, v)| (k.as_str(), v.as_str())) + } +} + +// --------------------------------------------------------------------------- +// DeltaPacket +// --------------------------------------------------------------------------- + +/// A serializable container for a batch of CRDT changesets. +/// +/// On the wire, `DeltaPacket` is: +/// 1. Serialized with `bincode`. +/// 2. Compressed with LZ4. +/// 3. Encrypted with ChaCha20-Poly1305 (key = SwarmKey bytes). +/// +/// The `entity_type` field allows the receiver to route the changeset to the +/// correct CRDT handler (`TaskCrdt`, `NoteCrdt`, `TimeBlockCrdt`, etc.). +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct DeltaPacket { + /// The entity type discriminator (e.g. `"task"`, `"note"`, `"time_block"`). + pub entity_type: String, + + /// The entity UUID this changeset belongs to. + pub entity_id: String, + + /// The HLC timestamp of the most recent change in this packet. + pub hlc_timestamp: String, + + /// Raw Automerge binary changeset bytes. + pub changeset_bytes: Vec, + + /// Sequence number within a sync session (for ordering / dedup). + pub sequence: u64, +} + +impl DeltaPacket { + /// Create a new `DeltaPacket`. + pub fn new( + entity_type: impl Into, + entity_id: impl Into, + hlc_timestamp: impl Into, + changeset_bytes: Vec, + sequence: u64, + ) -> Self { + DeltaPacket { + entity_type: entity_type.into(), + entity_id: entity_id.into(), + hlc_timestamp: hlc_timestamp.into(), + changeset_bytes, + sequence, + } + } + + /// Serialize, compress (LZ4), and encrypt (ChaCha20-Poly1305) this packet. + /// + /// The `swarm_key` provides the 32-byte symmetric encryption key. + /// A random 12-byte nonce is prepended to the ciphertext output. + /// + /// # Wire format + /// + /// ```text + /// [ 12-byte nonce ][ ChaCha20-Poly1305 ciphertext of LZ4(bincode(DeltaPacket)) ] + /// ``` + pub fn encode(&self, swarm_key: &SwarmKey) -> NetResult> { + // Step 1: bincode serialize. + let serialized = bincode::serialize(self).map_err(|e| { + NetError::Transport(format!("delta packet serialization failed: {e}")) + })?; + + // Step 2: LZ4 compress. + let compressed = lz4_flex::compress_prepend_size(&serialized); + + // Step 3: ChaCha20-Poly1305 encrypt. + let key = GenericArray::from_slice(swarm_key.as_bytes()); + let cipher = ChaCha20Poly1305::new(key); + + // Generate a random 12-byte nonce. + let nonce_bytes: [u8; 12] = rand::random(); + let nonce = GenericArray::from_slice(&nonce_bytes); + + let mut buffer = compressed; + cipher + .encrypt_in_place(nonce, b"chronogrid-delta", &mut buffer) + .map_err(|e| NetError::Transport(format!("encryption failed: {e}")))?; + + // Prepend nonce to ciphertext. + let mut output = Vec::with_capacity(12 + buffer.len()); + output.extend_from_slice(&nonce_bytes); + output.extend_from_slice(&buffer); + Ok(output) + } + + /// Decrypt (ChaCha20-Poly1305), decompress (LZ4), and deserialize a wire + /// payload produced by [`encode`](Self::encode). + pub fn decode(wire_bytes: &[u8], swarm_key: &SwarmKey) -> NetResult { + if wire_bytes.len() < 12 { + return Err(NetError::Transport( + "delta packet too short: missing nonce".to_string(), + )); + } + + // Split nonce and ciphertext. + let (nonce_bytes, ciphertext) = wire_bytes.split_at(12); + let nonce = GenericArray::from_slice(nonce_bytes); + + // Step 1: Decrypt. + let key = GenericArray::from_slice(swarm_key.as_bytes()); + let cipher = ChaCha20Poly1305::new(key); + let mut buffer = ciphertext.to_vec(); + cipher + .decrypt_in_place(nonce, b"chronogrid-delta", &mut buffer) + .map_err(|e| NetError::Transport(format!("decryption failed: {e}")))?; + + // Step 2: LZ4 decompress. + let decompressed = lz4_flex::decompress_size_prepended(&buffer) + .map_err(|e| NetError::Transport(format!("decompression failed: {e}")))?; + + // Step 3: bincode deserialize. + let packet: DeltaPacket = bincode::deserialize(&decompressed) + .map_err(|e| NetError::Transport(format!("deserialization failed: {e}")))?; + + Ok(packet) + } +} + +// --------------------------------------------------------------------------- +// SyncRequest / SyncResponse +// --------------------------------------------------------------------------- + +/// The libp2p stream protocol identifier for ChronoGrid sync. +pub const SYNC_PROTOCOL_ID: &str = "/chronogrid/sync/1.0.0"; + +/// A request sent during the vector-clock exchange phase. +/// +/// The initiating peer sends its `VectorClock` to the responder. +/// The responder replies with a `SyncResponse` containing the deltas +/// the initiator is missing. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SyncRequest { + /// The sender's current vector clock. + pub clock: VectorClock, +} + +/// A response to a `SyncRequest`. +/// +/// Contains the encrypted delta packets that the requester is missing, +/// plus the responder's own vector clock so the requester can send back +/// any deltas the responder is missing. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SyncResponse { + /// Encrypted delta packets (each produced by `DeltaPacket::encode`). + pub encrypted_deltas: Vec>, + /// The responder's current vector clock (so the requester can compute + /// what the responder is missing and send a follow-up). + pub responder_clock: VectorClock, +} + +// --------------------------------------------------------------------------- +// SyncCodec — libp2p request_response Codec +// --------------------------------------------------------------------------- + +/// Codec for the ChronoGrid sync request/response protocol. +/// +/// Frames are length-prefixed (4-byte big-endian u32) bincode payloads. +#[derive(Debug, Clone, Default)] +pub struct SyncCodec; + +/// Maximum frame size: 64 MiB (generous upper bound for large delta batches). +const MAX_FRAME_SIZE: usize = 64 * 1024 * 1024; + +#[async_trait] +impl ReqResCodec for SyncCodec { + type Protocol = StreamProtocol; + type Request = SyncRequest; + type Response = SyncResponse; + + async fn read_request( + &mut self, + _protocol: &Self::Protocol, + io: &mut T, + ) -> std::io::Result + where + T: futures::AsyncRead + Unpin + Send, + { + use futures::AsyncReadExt; + // Read 4-byte big-endian length prefix. + let mut len_buf = [0u8; 4]; + io.read_exact(&mut len_buf).await?; + let len = u32::from_be_bytes(len_buf) as usize; + if len > MAX_FRAME_SIZE { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("sync request frame too large: {len} bytes"), + )); + } + let mut buf = vec![0u8; len]; + io.read_exact(&mut buf).await?; + bincode::deserialize(&buf) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())) + } + + async fn read_response( + &mut self, + _protocol: &Self::Protocol, + io: &mut T, + ) -> std::io::Result + where + T: futures::AsyncRead + Unpin + Send, + { + use futures::AsyncReadExt; + let mut len_buf = [0u8; 4]; + io.read_exact(&mut len_buf).await?; + let len = u32::from_be_bytes(len_buf) as usize; + if len > MAX_FRAME_SIZE { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("sync response frame too large: {len} bytes"), + )); + } + let mut buf = vec![0u8; len]; + io.read_exact(&mut buf).await?; + bincode::deserialize(&buf) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())) + } + + async fn write_request( + &mut self, + _protocol: &Self::Protocol, + io: &mut T, + req: Self::Request, + ) -> std::io::Result<()> + where + T: futures::AsyncWrite + Unpin + Send, + { + use futures::AsyncWriteExt; + let bytes = bincode::serialize(&req) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?; + let len = (bytes.len() as u32).to_be_bytes(); + io.write_all(&len).await?; + io.write_all(&bytes).await?; + io.flush().await + } + + async fn write_response( + &mut self, + _protocol: &Self::Protocol, + io: &mut T, + res: Self::Response, + ) -> std::io::Result<()> + where + T: futures::AsyncWrite + Unpin + Send, + { + use futures::AsyncWriteExt; + let bytes = bincode::serialize(&res) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?; + let len = (bytes.len() as u32).to_be_bytes(); + io.write_all(&len).await?; + io.write_all(&bytes).await?; + io.flush().await + } +} + +// --------------------------------------------------------------------------- +// SyncProtocol — NetworkBehaviour +// --------------------------------------------------------------------------- + +/// libp2p `NetworkBehaviour` implementing the ChronoGrid CRDT delta sync protocol. +/// +/// On peer connect (triggered by `SwarmManager`), the local node sends its +/// `VectorClock` to the remote peer via the request/response protocol. +/// The remote peer computes the missing deltas and returns them encrypted. +/// +/// # Event flow +/// +/// ```text +/// Local Remote +/// │── SyncRequest(local_clock) ──▶│ +/// │ │ compute missing deltas +/// │◀── SyncResponse(deltas, remote_clock) ──│ +/// │ decrypt + decompress deltas │ +/// │ emit NetEvent::DeltasReceived │ +/// │ compute what remote is missing│ +/// │── SyncRequest(local_clock) ──▶│ (follow-up if remote is behind) +/// ``` +#[derive(NetworkBehaviour)] +pub struct SyncProtocol { + /// The underlying request/response behaviour. + pub inner: ReqResBehaviour, +} + +impl SyncProtocol { + /// Create a new `SyncProtocol` behaviour. + pub fn new() -> Self { + let protocol = StreamProtocol::new(SYNC_PROTOCOL_ID); + let config = ReqResConfig::default(); + let inner = ReqResBehaviour::new([(protocol, ProtocolSupport::Full)], config); + SyncProtocol { inner } + } + + /// Send a `SyncRequest` (local vector clock) to a remote peer. + /// + /// Returns the `OutboundRequestId` for tracking the response. + pub fn send_sync_request( + &mut self, + peer: PeerId, + clock: VectorClock, + ) -> request_response::OutboundRequestId { + self.inner.send_request(&peer, SyncRequest { clock }) + } + + /// Send a `SyncResponse` (encrypted deltas + local clock) to a peer. + pub fn send_sync_response( + &mut self, + channel: request_response::ResponseChannel, + encrypted_deltas: Vec>, + responder_clock: VectorClock, + ) -> Result<(), SyncResponse> { + self.inner.send_response( + channel, + SyncResponse { + encrypted_deltas, + responder_clock, + }, + ) + } + + /// Decrypt and deserialize a batch of encrypted delta payloads. + /// + /// Returns the decoded `DeltaPacket` list, or an error if any packet + /// fails decryption or deserialization. + pub fn decode_deltas( + encrypted_deltas: &[Vec], + swarm_key: &SwarmKey, + ) -> NetResult> { + encrypted_deltas + .iter() + .map(|wire| DeltaPacket::decode(wire, swarm_key)) + .collect() + } + + /// Encrypt and serialize a batch of `DeltaPacket`s for transmission. + pub fn encode_deltas( + packets: &[DeltaPacket], + swarm_key: &SwarmKey, + ) -> NetResult>> { + packets.iter().map(|p| p.encode(swarm_key)).collect() + } +} + +impl Default for SyncProtocol { + fn default() -> Self { + Self::new() + } +} + +// --------------------------------------------------------------------------- +// GossipEvent — our application-level gossip message +// --------------------------------------------------------------------------- + +/// The Gossipsub topic for real-time ChronoGrid state-change notifications. +pub const GOSSIP_TOPIC: &str = "chronogrid/state-changes/1"; + +/// A lightweight state-change notification broadcast over Gossipsub. +/// +/// Used for real-time "something changed" signals. The full delta is +/// exchanged via `SyncProtocol`; this message just triggers peers to +/// initiate a sync. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct GossipEvent { + /// The entity type that changed (e.g. `"task"`, `"note"`). + pub entity_type: String, + /// The entity ID that changed. + pub entity_id: String, + /// The HLC timestamp of the change. + pub hlc_timestamp: String, + /// The originating node's peer ID (string-encoded). + pub origin_peer: String, +} + +impl GossipEvent { + /// Create a new `GossipEvent`. + pub fn new( + entity_type: impl Into, + entity_id: impl Into, + hlc_timestamp: impl Into, + origin_peer: &PeerId, + ) -> Self { + GossipEvent { + entity_type: entity_type.into(), + entity_id: entity_id.into(), + hlc_timestamp: hlc_timestamp.into(), + origin_peer: origin_peer.to_string(), + } + } + + /// Serialize this event to bytes for Gossipsub publication. + pub fn to_bytes(&self) -> NetResult> { + bincode::serialize(self) + .map_err(|e| NetError::Transport(format!("gossip event serialization failed: {e}"))) + } + + /// Deserialize a `GossipEvent` from Gossipsub message bytes. + pub fn from_bytes(bytes: &[u8]) -> NetResult { + bincode::deserialize(bytes) + .map_err(|e| NetError::Transport(format!("gossip event deserialization failed: {e}"))) + } +} + +// --------------------------------------------------------------------------- +// GossipsubBroadcast +// --------------------------------------------------------------------------- + +/// Wraps `libp2p::gossipsub::Behaviour` for ChronoGrid state-change broadcasts. +/// +/// Subscribes to the `chronogrid/state-changes/1` topic on construction. +/// Provides `publish()` and `handle_event()` helpers used by `SwarmManager`. +pub struct GossipsubBroadcast { + /// The underlying libp2p Gossipsub behaviour. + pub behaviour: GossipBehaviour, + /// The topic this node is subscribed to. + topic: IdentTopic, +} + +impl GossipsubBroadcast { + /// Create a new `GossipsubBroadcast` and subscribe to the state-change topic. + /// + /// Uses content-based message deduplication (SHA-256 of message data). + /// + /// # Errors + /// + /// Returns an error if the Gossipsub behaviour cannot be constructed. + pub fn new(local_keypair: &libp2p::identity::Keypair) -> NetResult { + let config = GossipConfigBuilder::default() + .heartbeat_interval(std::time::Duration::from_secs(1)) + .validation_mode(ValidationMode::Strict) + .message_id_fn(|msg: &GossipMessage| { + // Use SHA-256 of the message data as the message ID for dedup. + use sha2::{Digest, Sha256}; + let hash = Sha256::digest(&msg.data); + MessageId::from(hash.as_slice()) + }) + .build() + .map_err(|e| NetError::Transport(format!("gossipsub config error: {e}")))?; + + let behaviour = + GossipBehaviour::new(MessageAuthenticity::Signed(local_keypair.clone()), config) + .map_err(|e| NetError::Transport(format!("gossipsub init error: {e}")))?; + + let topic = IdentTopic::new(GOSSIP_TOPIC); + + let mut gossip = GossipsubBroadcast { behaviour, topic }; + + // Subscribe to the state-change topic immediately. + gossip + .behaviour + .subscribe(&gossip.topic) + .map_err(|e| NetError::Transport(format!("gossipsub subscribe error: {e}")))?; + + Ok(gossip) + } + + /// Publish a `GossipEvent` to all subscribed peers. + /// + /// # Errors + /// + /// Returns an error if serialization or Gossipsub publication fails. + pub fn publish(&mut self, event: &GossipEvent) -> NetResult { + let bytes = event.to_bytes()?; + self.behaviour + .publish(self.topic.clone(), bytes) + .map_err(|e| NetError::Transport(format!("gossipsub publish error: {e}"))) + } + + /// Convert a raw Gossipsub behaviour event into zero or more [`NetEvent`] values. + /// + /// - `GossipBehaviourEvent::Message` → parse payload → `NetEvent::GossipReceived` + /// - All other events are logged and produce no output. + pub fn handle_behaviour_event(&self, event: GossipBehaviourEvent) -> Vec { + match event { + GossipBehaviourEvent::Message { message, .. } => { + match GossipEvent::from_bytes(&message.data) { + Ok(ge) => vec![NetEvent::GossipReceived { + entity_type: ge.entity_type, + entity_id: ge.entity_id, + hlc_timestamp: ge.hlc_timestamp, + origin_peer: ge.origin_peer, + }], + Err(e) => { + tracing::warn!("gossipsub: failed to parse message: {e}"); + vec![] + } + } + } + GossipBehaviourEvent::Subscribed { peer_id, topic } => { + tracing::debug!(%peer_id, %topic, "gossipsub: peer subscribed"); + vec![] + } + GossipBehaviourEvent::Unsubscribed { peer_id, topic } => { + tracing::debug!(%peer_id, %topic, "gossipsub: peer unsubscribed"); + vec![] + } + GossipBehaviourEvent::GossipsubNotSupported { peer_id } => { + tracing::debug!(%peer_id, "gossipsub: not supported by peer"); + vec![] + } + } + } + + /// Parse a raw Gossipsub `Message` into a `GossipEvent`. + pub fn parse_message(msg: &GossipMessage) -> NetResult { + GossipEvent::from_bytes(&msg.data) + } + + /// Return the topic this broadcast is subscribed to. + pub fn topic(&self) -> &IdentTopic { + &self.topic + } +} + +// --------------------------------------------------------------------------- +// NetEvent +// --------------------------------------------------------------------------- + +/// Unified network event enum consumed by `SwarmManager` (Task 2.5). +/// +/// These events are produced by `SyncProtocol` and `GossipsubBroadcast` and +/// forwarded to the application layer via a `tokio::sync::broadcast` channel. +#[derive(Debug, Clone)] +pub enum NetEvent { + /// A batch of CRDT delta packets was received from a peer and successfully + /// decrypted. The `deltas` are ready to be merged into the local database. + DeltasReceived { + /// The peer that sent the deltas. + from_peer: PeerId, + /// The decoded delta packets, ordered by sequence number. + deltas: Vec, + }, + + /// A full sync cycle with a peer completed successfully. + SyncComplete { + /// The peer the sync was completed with. + peer_id: PeerId, + }, + + /// A sync cycle with a peer failed. + SyncError { + /// The peer the sync failed with. + peer_id: PeerId, + /// A human-readable description of the failure. + reason: String, + }, + + /// A real-time state-change notification was received via Gossipsub. + GossipReceived { + /// The entity type that changed. + entity_type: String, + /// The entity ID that changed. + entity_id: String, + /// The HLC timestamp of the change. + hlc_timestamp: String, + /// The originating peer (as a string-encoded PeerId). + origin_peer: String, + }, +} + +impl NetEvent { + /// Return `true` if this event carries delta packets ready for merging. + pub fn is_deltas_received(&self) -> bool { + matches!(self, NetEvent::DeltasReceived { .. }) + } + + /// Return `true` if this event signals a completed sync. + pub fn is_sync_complete(&self) -> bool { + matches!(self, NetEvent::SyncComplete { .. }) + } + + /// Return `true` if this event signals a sync failure. + pub fn is_sync_error(&self) -> bool { + matches!(self, NetEvent::SyncError { .. }) + } + + /// Return `true` if this event is a Gossipsub notification. + pub fn is_gossip(&self) -> bool { + matches!(self, NetEvent::GossipReceived { .. }) + } +} + +impl std::fmt::Display for NetEvent { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + NetEvent::DeltasReceived { from_peer, deltas } => { + write!(f, "DeltasReceived(from={from_peer}, count={})", deltas.len()) + } + NetEvent::SyncComplete { peer_id } => write!(f, "SyncComplete(peer={peer_id})"), + NetEvent::SyncError { peer_id, reason } => { + write!(f, "SyncError(peer={peer_id}, reason={reason})") + } + NetEvent::GossipReceived { + entity_type, + entity_id, + .. + } => write!(f, "GossipReceived(type={entity_type}, id={entity_id})"), + } + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::swarm_key::SwarmKey; + + // ── VectorClock tests ──────────────────────────────────────────────────── + + #[test] + fn test_vector_clock_new_is_empty() { + let vc = VectorClock::new(); + assert!(vc.is_empty()); + assert_eq!(vc.len(), 0); + } + + #[test] + fn test_vector_clock_update_inserts_new_entry() { + let mut vc = VectorClock::new(); + let updated = vc.update("entity-1", "1724000000000-0001-ab12"); + assert!(updated, "first update must return true"); + assert_eq!(vc.len(), 1); + assert_eq!(vc.get("entity-1"), Some("1724000000000-0001-ab12")); + } + + #[test] + fn test_vector_clock_update_newer_timestamp() { + let mut vc = VectorClock::new(); + vc.update("entity-1", "1724000000000-0001-ab12"); + let updated = vc.update("entity-1", "1724000000001-0001-ab12"); + assert!(updated, "newer timestamp must update and return true"); + assert_eq!(vc.get("entity-1"), Some("1724000000001-0001-ab12")); + } + + #[test] + fn test_vector_clock_update_older_timestamp_rejected() { + let mut vc = VectorClock::new(); + vc.update("entity-1", "1724000000005-0001-ab12"); + let updated = vc.update("entity-1", "1724000000001-0001-ab12"); + assert!(!updated, "older timestamp must not update and return false"); + assert_eq!(vc.get("entity-1"), Some("1724000000005-0001-ab12")); + } + + #[test] + fn test_vector_clock_update_equal_timestamp_rejected() { + let mut vc = VectorClock::new(); + vc.update("entity-1", "1724000000005-0001-ab12"); + let updated = vc.update("entity-1", "1724000000005-0001-ab12"); + assert!(!updated, "equal timestamp must not update and return false"); + } + + #[test] + fn test_vector_clock_get_missing_returns_none() { + let vc = VectorClock::new(); + assert_eq!(vc.get("nonexistent"), None); + } + + #[test] + fn test_vector_clock_missing_from_empty_self() { + let self_vc = VectorClock::new(); + let mut other = VectorClock::new(); + other.update("entity-1", "1724000000000-0001-ab12"); + other.update("entity-2", "1724000000001-0001-ab12"); + + let missing = self_vc.missing_from(&other); + assert_eq!( + missing.len(), + 2, + "empty self must report all other entities as missing" + ); + assert!(missing.contains(&"entity-1".to_string())); + assert!(missing.contains(&"entity-2".to_string())); + } + + #[test] + fn test_vector_clock_missing_from_partial_overlap() { + let mut self_vc = VectorClock::new(); + self_vc.update("entity-1", "1724000000005-0001-ab12"); // newer + self_vc.update("entity-2", "1724000000001-0001-ab12"); // older than other + + let mut other = VectorClock::new(); + other.update("entity-1", "1724000000003-0001-ab12"); // older than self + other.update("entity-2", "1724000000009-0001-ab12"); // newer than self + other.update("entity-3", "1724000000000-0001-ab12"); // not in self + + let missing = self_vc.missing_from(&other); + // entity-1: self is newer → not missing + // entity-2: other is newer → missing + // entity-3: not in self → missing + assert_eq!(missing.len(), 2); + assert!(missing.contains(&"entity-2".to_string())); + assert!(missing.contains(&"entity-3".to_string())); + assert!(!missing.contains(&"entity-1".to_string())); + } + + #[test] + fn test_vector_clock_missing_from_self_is_superset() { + let mut self_vc = VectorClock::new(); + self_vc.update("entity-1", "1724000000010-0001-ab12"); + self_vc.update("entity-2", "1724000000010-0001-ab12"); + + let mut other = VectorClock::new(); + other.update("entity-1", "1724000000005-0001-ab12"); + + let missing = self_vc.missing_from(&other); + assert!(missing.is_empty(), "self is superset — nothing missing"); + } + + #[test] + fn test_vector_clock_merge() { + let mut vc1 = VectorClock::new(); + vc1.update("entity-1", "1724000000010-0001-ab12"); + vc1.update("entity-2", "1724000000001-0001-ab12"); + + let mut vc2 = VectorClock::new(); + vc2.update("entity-2", "1724000000009-0001-ab12"); // newer + vc2.update("entity-3", "1724000000005-0001-ab12"); // new entity + + vc1.merge(&vc2); + + assert_eq!(vc1.len(), 3); + assert_eq!(vc1.get("entity-1"), Some("1724000000010-0001-ab12")); // unchanged + assert_eq!(vc1.get("entity-2"), Some("1724000000009-0001-ab12")); // updated + assert_eq!(vc1.get("entity-3"), Some("1724000000005-0001-ab12")); // added + } + + #[test] + fn test_vector_clock_iter() { + let mut vc = VectorClock::new(); + vc.update("entity-1", "ts-1"); + vc.update("entity-2", "ts-2"); + + let pairs: Vec<_> = vc.iter().collect(); + assert_eq!(pairs.len(), 2); + } + + #[test] + fn test_vector_clock_serialization_round_trip() { + let mut vc = VectorClock::new(); + vc.update("entity-1", "1724000000000-0001-ab12"); + vc.update("entity-2", "1724000000001-0002-cd34"); + + let serialized = bincode::serialize(&vc).expect("serialize"); + let deserialized: VectorClock = bincode::deserialize(&serialized).expect("deserialize"); + assert_eq!(vc, deserialized); + } + + // ── DeltaPacket tests ──────────────────────────────────────────────────── + + fn make_packet(seq: u64) -> DeltaPacket { + DeltaPacket::new( + "task", + format!("entity-{seq}"), + format!("1724000000{seq:03}-0001-ab12"), + vec![0xDE, 0xAD, 0xBE, 0xEF, seq as u8], + seq, + ) + } + + #[test] + fn test_delta_packet_new() { + let p = make_packet(1); + assert_eq!(p.entity_type, "task"); + assert_eq!(p.entity_id, "entity-1"); + assert_eq!(p.sequence, 1); + assert!(!p.changeset_bytes.is_empty()); + } + + #[test] + fn test_delta_packet_encode_decode_round_trip() { + let key = SwarmKey::generate(); + let packet = make_packet(42); + + let wire = packet.encode(&key).expect("encode"); + assert!(wire.len() > 12, "wire must be longer than nonce"); + + let decoded = DeltaPacket::decode(&wire, &key).expect("decode"); + assert_eq!(decoded, packet); + } + + #[test] + fn test_delta_packet_encode_produces_different_ciphertext_each_time() { + // Each encode call uses a fresh random nonce, so ciphertexts differ. + let key = SwarmKey::generate(); + let packet = make_packet(1); + + let wire1 = packet.encode(&key).expect("encode 1"); + let wire2 = packet.encode(&key).expect("encode 2"); + assert_ne!(wire1, wire2, "random nonce must produce different ciphertexts"); + } + + #[test] + fn test_delta_packet_decode_wrong_key_fails() { + let key1 = SwarmKey::generate(); + let key2 = SwarmKey::generate(); + let packet = make_packet(1); + + let wire = packet.encode(&key1).expect("encode"); + let result = DeltaPacket::decode(&wire, &key2); + assert!(result.is_err(), "decryption with wrong key must fail"); + } + + #[test] + fn test_delta_packet_decode_truncated_fails() { + let key = SwarmKey::generate(); + let result = DeltaPacket::decode(&[0u8; 8], &key); + assert!(result.is_err(), "truncated payload must fail"); + } + + #[test] + fn test_delta_packet_decode_tampered_ciphertext_fails() { + let key = SwarmKey::generate(); + let packet = make_packet(1); + let mut wire = packet.encode(&key).expect("encode"); + + // Flip a byte in the ciphertext (after the 12-byte nonce). + wire[20] ^= 0xFF; + + let result = DeltaPacket::decode(&wire, &key); + assert!(result.is_err(), "tampered ciphertext must fail authentication"); + } + + #[test] + fn test_delta_packet_large_changeset() { + let key = SwarmKey::generate(); + // 1 MiB of simulated Automerge changeset bytes. + let large_bytes = vec![0xAB; 1024 * 1024]; + let packet = DeltaPacket::new("note", "entity-large", "ts-large", large_bytes, 0); + + let wire = packet.encode(&key).expect("encode large"); + let decoded = DeltaPacket::decode(&wire, &key).expect("decode large"); + assert_eq!(decoded.changeset_bytes.len(), 1024 * 1024); + } + + #[test] + fn test_delta_packet_serialization_round_trip() { + let packet = make_packet(7); + let bytes = bincode::serialize(&packet).expect("serialize"); + let decoded: DeltaPacket = bincode::deserialize(&bytes).expect("deserialize"); + assert_eq!(decoded, packet); + } + + // ── SyncProtocol helper tests ──────────────────────────────────────────── + + #[test] + fn test_sync_protocol_encode_decode_batch() { + let key = SwarmKey::generate(); + let packets = vec![make_packet(1), make_packet(2), make_packet(3)]; + + let encoded = SyncProtocol::encode_deltas(&packets, &key).expect("encode batch"); + assert_eq!(encoded.len(), 3); + + let decoded = SyncProtocol::decode_deltas(&encoded, &key).expect("decode batch"); + assert_eq!(decoded, packets); + } + + #[test] + fn test_sync_protocol_encode_empty_batch() { + let key = SwarmKey::generate(); + let encoded = SyncProtocol::encode_deltas(&[], &key).expect("encode empty"); + assert!(encoded.is_empty()); + let decoded = SyncProtocol::decode_deltas(&encoded, &key).expect("decode empty"); + assert!(decoded.is_empty()); + } + + #[test] + fn test_sync_protocol_new_constructs_successfully() { + let _proto = SyncProtocol::new(); + // Construction must not panic. + } + + // ── GossipEvent tests ──────────────────────────────────────────────────── + + #[test] + fn test_gossip_event_serialization_round_trip() { + let peer = PeerId::random(); + let event = GossipEvent::new("task", "entity-1", "ts-001", &peer); + + let bytes = event.to_bytes().expect("serialize"); + let decoded = GossipEvent::from_bytes(&bytes).expect("deserialize"); + + assert_eq!(decoded.entity_type, "task"); + assert_eq!(decoded.entity_id, "entity-1"); + assert_eq!(decoded.hlc_timestamp, "ts-001"); + assert_eq!(decoded.origin_peer, peer.to_string()); + } + + #[test] + fn test_gossip_event_from_bytes_invalid_data() { + let result = GossipEvent::from_bytes(b"not-valid-bincode"); + assert!(result.is_err(), "invalid bytes must fail deserialization"); + } + + // ── NetEvent tests ─────────────────────────────────────────────────────── + + #[test] + fn test_net_event_deltas_received_predicate() { + let event = NetEvent::DeltasReceived { + from_peer: PeerId::random(), + deltas: vec![make_packet(1)], + }; + assert!(event.is_deltas_received()); + assert!(!event.is_sync_complete()); + assert!(!event.is_sync_error()); + assert!(!event.is_gossip()); + } + + #[test] + fn test_net_event_sync_complete_predicate() { + let event = NetEvent::SyncComplete { + peer_id: PeerId::random(), + }; + assert!(event.is_sync_complete()); + assert!(!event.is_deltas_received()); + assert!(!event.is_sync_error()); + assert!(!event.is_gossip()); + } + + #[test] + fn test_net_event_sync_error_predicate() { + let event = NetEvent::SyncError { + peer_id: PeerId::random(), + reason: "timeout".to_string(), + }; + assert!(event.is_sync_error()); + assert!(!event.is_deltas_received()); + assert!(!event.is_sync_complete()); + assert!(!event.is_gossip()); + } + + #[test] + fn test_net_event_gossip_received_predicate() { + let event = NetEvent::GossipReceived { + entity_type: "task".to_string(), + entity_id: "entity-1".to_string(), + hlc_timestamp: "ts-001".to_string(), + origin_peer: "peer-1".to_string(), + }; + assert!(event.is_gossip()); + assert!(!event.is_deltas_received()); + assert!(!event.is_sync_complete()); + assert!(!event.is_sync_error()); + } + + #[test] + fn test_net_event_display_deltas_received() { + let event = NetEvent::DeltasReceived { + from_peer: PeerId::random(), + deltas: vec![make_packet(1), make_packet(2)], + }; + let s = format!("{event}"); + assert!(s.contains("DeltasReceived")); + assert!(s.contains("count=2")); + } + + #[test] + fn test_net_event_display_sync_complete() { + let event = NetEvent::SyncComplete { + peer_id: PeerId::random(), + }; + let s = format!("{event}"); + assert!(s.contains("SyncComplete")); + } + + #[test] + fn test_net_event_display_sync_error() { + let event = NetEvent::SyncError { + peer_id: PeerId::random(), + reason: "connection reset".to_string(), + }; + let s = format!("{event}"); + assert!(s.contains("SyncError")); + assert!(s.contains("connection reset")); + } + + #[test] + fn test_net_event_display_gossip_received() { + let event = NetEvent::GossipReceived { + entity_type: "note".to_string(), + entity_id: "entity-42".to_string(), + hlc_timestamp: "ts-999".to_string(), + origin_peer: "peer-x".to_string(), + }; + let s = format!("{event}"); + assert!(s.contains("GossipReceived")); + assert!(s.contains("note")); + } + + #[test] + fn test_net_event_clone_deltas_received() { + let event = NetEvent::DeltasReceived { + from_peer: PeerId::random(), + deltas: vec![make_packet(1)], + }; + let cloned = event.clone(); + assert!(cloned.is_deltas_received()); + } + + // ── Vector clock + delta integration test ─────────────────────────────── + + #[test] + fn test_vector_clock_drives_delta_selection() { + // Simulate: peer A has entity-1 at ts-5, entity-2 at ts-3. + // Peer B has entity-1 at ts-3 (behind), entity-3 at ts-1 (new to A). + let mut clock_a = VectorClock::new(); + clock_a.update("entity-1", "1724000000005-0001-aa"); + clock_a.update("entity-2", "1724000000003-0001-aa"); + + let mut clock_b = VectorClock::new(); + clock_b.update("entity-1", "1724000000003-0001-bb"); + clock_b.update("entity-3", "1724000000001-0001-bb"); + + // What A is missing from B: + let a_missing = clock_a.missing_from(&clock_b); + assert_eq!(a_missing.len(), 1); + assert!(a_missing.contains(&"entity-3".to_string())); + + // What B is missing from A: + let b_missing = clock_b.missing_from(&clock_a); + assert_eq!(b_missing.len(), 2); + assert!(b_missing.contains(&"entity-1".to_string())); // A is newer + assert!(b_missing.contains(&"entity-2".to_string())); // B doesn't have it + + // Simulate A sending deltas for entity-1 and entity-2 to B. + let key = SwarmKey::generate(); + let deltas_for_b = vec![ + DeltaPacket::new( + "task", + "entity-1", + "1724000000005-0001-aa", + vec![1, 2, 3], + 0, + ), + DeltaPacket::new( + "task", + "entity-2", + "1724000000003-0001-aa", + vec![4, 5, 6], + 1, + ), + ]; + let encoded = SyncProtocol::encode_deltas(&deltas_for_b, &key).expect("encode"); + let decoded = SyncProtocol::decode_deltas(&encoded, &key).expect("decode"); + + assert_eq!(decoded.len(), 2); + assert_eq!(decoded[0].entity_id, "entity-1"); + assert_eq!(decoded[1].entity_id, "entity-2"); + + // After B merges, update B's clock. + for delta in &decoded { + clock_b.update(&delta.entity_id, &delta.hlc_timestamp); + } + + // Now B should be caught up on entity-1 and entity-2. + let b_missing_after = clock_b.missing_from(&clock_a); + assert!(b_missing_after.is_empty(), "B should be fully synced after merge"); + } +} diff --git a/docs/ARCHITECTURE/ADR/0010-net-crdt-delta-sync-protocol.md b/docs/ARCHITECTURE/ADR/0010-net-crdt-delta-sync-protocol.md new file mode 100644 index 0000000..46c41a8 --- /dev/null +++ b/docs/ARCHITECTURE/ADR/0010-net-crdt-delta-sync-protocol.md @@ -0,0 +1,154 @@ +# ADR-0010: chronogrid_net — CRDT Delta Sync Protocol (Task 2.4) + +**Status:** Accepted +**Date:** 2026-08-22 +**Task:** 2.4 — Implement chronogrid_net: CRDT Delta Sync Protocol +**Author:** Autonomous Agent + +--- + +## Context + +Task 2.4 required implementing the CRDT delta exchange protocol over libp2p in +`crates/chronogrid_net/`, building on the transport and discovery layers from +Tasks 2.2 and 2.3. + +The implementation must satisfy: +- `VectorClock` — per-entity HLC head-state tracking for delta computation. +- `DeltaPacket` — serializable, compressed, encrypted CRDT changeset container. +- `SyncProtocol` — libp2p `NetworkBehaviour` for request/response vector-clock exchange. +- `GossipsubBroadcast` — real-time state-change notifications via Gossipsub. +- `NetEvent` — unified event enum for `SwarmManager` (Task 2.5) consumption. + +--- + +## Decision + +### New module: `crates/chronogrid_net/src/sync.rs` + +| Type | Role | +|-----------------------|-------------------------------------------------------------------------------| +| `VectorClock` | `HashMap` — lexicographic HLC comparison for delta computation | +| `DeltaPacket` | Serializable CRDT changeset with entity routing metadata | +| `SyncCodec` | libp2p `request_response::Codec` — length-prefixed bincode framing | +| `SyncProtocol` | `#[derive(NetworkBehaviour)]` wrapping `ReqResBehaviour` | +| `GossipEvent` | Application-level gossip message (entity_type, entity_id, hlc_timestamp, origin_peer) | +| `GossipsubBroadcast` | Wraps `libp2p::gossipsub::Behaviour` with topic subscription and event parsing | +| `NetEvent` | Unified event enum: `DeltasReceived`, `SyncComplete`, `SyncError`, `GossipReceived` | + +### VectorClock Design + +- `HashMap` mapping `entity_id → hlc_timestamp`. +- HLC timestamps use lexicographic ordering (format: `--`), + which is equivalent to causal ordering — no numeric parsing required. +- `update()` returns `bool` indicating whether the entry was actually advanced. +- `missing_from(other)` computes the set of entity IDs where `other` is ahead of `self`. +- `merge(other)` takes the maximum timestamp per entity (LWW semantics). +- Fully serializable via `bincode` for wire transmission in `SyncRequest`/`SyncResponse`. + +### DeltaPacket Wire Format + +```text +[ 12-byte random nonce ] +[ ChaCha20-Poly1305 ciphertext of: + LZ4(bincode(DeltaPacket { entity_type, entity_id, hlc_timestamp, changeset_bytes, sequence })) +] +``` + +- **Serialization:** `bincode` (compact binary, no schema overhead). +- **Compression:** `lz4_flex` with size-prepended format — Automerge changesets + compress well (typically 60–80% reduction for structured JSON-like data). +- **Encryption:** ChaCha20-Poly1305 AEAD with the 32-byte `SwarmKey` as the + symmetric key. A fresh 12-byte random nonce per packet ensures ciphertext + uniqueness even for identical plaintexts. +- **Authentication:** The AEAD tag authenticates the ciphertext; tampering + causes decryption to fail with an error (not silent corruption). +- **AAD:** `b"chronogrid-delta"` as additional authenticated data to bind the + ciphertext to the ChronoGrid protocol context. + +### SyncProtocol (NetworkBehaviour) + +- Wraps `libp2p::request_response::Behaviour` via `#[derive(NetworkBehaviour)]`. +- `SyncCodec` implements `libp2p::request_response::Codec` using `#[async_trait]` + (required by libp2p-request-response 0.27.0 which uses `async_trait` internally). +- Frame format: 4-byte big-endian u32 length prefix + bincode payload. +- Maximum frame size: 64 MiB (generous upper bound for large delta batches). +- `send_sync_request()` / `send_sync_response()` delegate to the inner behaviour. +- `encode_deltas()` / `decode_deltas()` are static helpers for batch processing. + +### GossipsubBroadcast Design + +- Subscribes to topic `chronogrid/state-changes/1` on construction. +- Message deduplication: SHA-256 of message data as `MessageId` (content-addressed). +- `ValidationMode::Strict` — only signed messages from known peers are accepted. +- `handle_behaviour_event()` maps `GossipBehaviourEvent::Message` → `NetEvent::GossipReceived`. +- All other Gossipsub events (Subscribed, Unsubscribed, GossipsubNotSupported) are + logged at DEBUG level and produce no `NetEvent` output. + +### NetEvent enum + +```rust +pub enum NetEvent { + DeltasReceived { from_peer: PeerId, deltas: Vec }, + SyncComplete { peer_id: PeerId }, + SyncError { peer_id: PeerId, reason: String }, + GossipReceived { entity_type, entity_id, hlc_timestamp, origin_peer: String }, +} +``` + +`SwarmManager` (Task 2.5) broadcasts these events via a +`tokio::sync::broadcast::Sender` channel to all application subscribers. + +### New dependencies added + +| Crate | Version | Purpose | +|--------------------|---------|--------------------------------------------------| +| `bincode` | 1 | Compact binary serialization for DeltaPacket | +| `lz4_flex` | 0.11 | LZ4 compression for delta payloads | +| `chacha20poly1305` | 0.10 | ChaCha20-Poly1305 AEAD encryption | +| `generic-array` | 0.14 | GenericArray for AEAD key/nonce types | +| `async-trait` | 0.1 | Required by libp2p request_response Codec trait | +| `libp2p` feature | — | Added `"request-response"` feature flag | + +--- + +## Test Coverage + +**100 unit tests pass + 5 doc-tests pass** across all modules: + +| Module | Tests | Coverage highlights | +|---------------|-------|----------------------------------------------------------------------------| +| `sync` | 35 | VectorClock (11), DeltaPacket (8), SyncProtocol helpers (3), GossipEvent (2), NetEvent (11) | +| `peer_event` | 10 | (from Task 2.3, unchanged) | +| `transport` | 12 | (from Task 2.3, unchanged) | +| `discovery` | 13 | (from Task 2.3, unchanged) | +| `identity` | 9 | (from Task 2.2, unchanged) | +| `swarm_key` | 16 | (from Task 2.2, unchanged) | +| `sync` (integration) | 1 | `test_vector_clock_drives_delta_selection` — full encode/decode/merge cycle | + +### Key test scenarios + +- **VectorClock:** insert, update (newer/older/equal), `missing_from` with empty/partial/superset, + merge, iter, bincode round-trip. +- **DeltaPacket:** encode/decode round-trip, random nonce uniqueness, wrong-key rejection, + truncated payload rejection, tampered ciphertext rejection (AEAD authentication), + 1 MiB large changeset, bincode round-trip. +- **SyncProtocol:** batch encode/decode, empty batch, construction. +- **GossipEvent:** bincode round-trip, invalid bytes rejection. +- **NetEvent:** all 4 variants — predicates, Display, Clone. +- **Integration:** two-peer delta selection driven by VectorClock comparison, + encode/decode pipeline, clock update after merge. + +--- + +## Consequences + +- `SwarmManager` (Task 2.5) can now import `SyncProtocol`, `GossipsubBroadcast`, + `VectorClock`, `DeltaPacket`, and `NetEvent` directly from `chronogrid_net::sync`. +- The `SyncProtocol` behaviour must be polled by the libp2p `Swarm` event loop in + Task 2.5; `send_sync_request()` should be called on `SwarmEvent::ConnectionEstablished`. +- `GossipsubBroadcast.behaviour` must be included in the composite `NetworkBehaviour` + derive struct in Task 2.5. +- `chronogrid_core::sync::SyncEngine` (Task 2.6) receives `NetEvent::DeltasReceived` + and calls `Database::merge_crdt_doc()` for each `DeltaPacket`. +- No breaking changes to existing modules (`identity`, `swarm_key`, `transport`, `discovery`). diff --git a/docs/TASKS/phase_2_networking_and_schema.md b/docs/TASKS/phase_2_networking_and_schema.md index 3910d74..814469b 100644 --- a/docs/TASKS/phase_2_networking_and_schema.md +++ b/docs/TASKS/phase_2_networking_and_schema.md @@ -57,7 +57,7 @@ ## Task 2.4 — Implement chronogrid_net: CRDT Delta Sync Protocol -- [ ] **Task 2.4**: Implement the CRDT delta exchange protocol over libp2p in `crates/chronogrid_net/`: +- [x] **Task 2.4**: Implement the CRDT delta exchange protocol over libp2p in `crates/chronogrid_net/`: - Implement `sync::VectorClock` — tracks per-entity HLC head state for delta computation - Implement `sync::DeltaPacket` — serializable struct containing compressed CRDT changesets - Implement `sync::SyncProtocol` — custom libp2p `NetworkBehaviour`: