From 98ff5aeb0702b8aa04c22f4943e73dc5ae72117c Mon Sep 17 00:00:00 2001 From: Mike Kell Date: Sat, 22 Aug 2026 13:04:38 -0400 Subject: [PATCH] feat(core): complete Task 2.6 - SyncEngine P2P integration into chronogrid_core --- Cargo.lock | 4 + DECISION_LOG.md | 15 + crates/chronogrid_core/Cargo.toml | 12 + crates/chronogrid_core/src/error.rs | 10 + crates/chronogrid_core/src/lib.rs | 16 + crates/chronogrid_core/src/sync/mod.rs | 832 ++++++++++++++++++ crates/chronogrid_net/src/lib.rs | 4 + crates/chronogrid_net/src/swarm_manager.rs | 30 + .../0012-core-sync-engine-p2p-integration.md | 117 +++ docs/TASKS/phase_2_networking_and_schema.md | 2 +- 10 files changed, 1041 insertions(+), 1 deletion(-) create mode 100644 crates/chronogrid_core/src/sync/mod.rs create mode 100644 docs/ARCHITECTURE/ADR/0012-core-sync-engine-p2p-integration.md diff --git a/Cargo.lock b/Cargo.lock index cbb3ea4..76f8dcd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -461,9 +461,13 @@ name = "chronogrid_core" version = "0.1.0" dependencies = [ "automerge", + "chronogrid_net", "rusqlite", "tempfile", "thiserror 2.0.20", + "tokio", + "tracing", + "tracing-subscriber", "uuid", ] diff --git a/DECISION_LOG.md b/DECISION_LOG.md index 13330e6..704caee 100644 --- a/DECISION_LOG.md +++ b/DECISION_LOG.md @@ -125,6 +125,21 @@ - 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-012: chronogrid_core Crate — SyncEngine P2P Integration (Task 2.6) + +- **Status**: Accepted +- **Context**: Task 2.6 required wiring the P2P sync layer from `chronogrid_net` into `chronogrid_core` so that CRDT delta packets received from remote peers are automatically merged into the local SQLCipher database, and local CRDT mutations are broadcast to connected peers. +- **Decision**: + - `chronogrid_net` added as an **optional dependency** under the `p2p` feature flag in `chronogrid_core/Cargo.toml`; `tokio` (sync+rt) and `tracing` also gated behind `p2p`. + - New `core::sync` module with `SyncEngine` struct: owns `Arc>`, `SwarmHandle`, and a `broadcast::Receiver`. + - Background Tokio task loops on `event_rx.recv()`: `NetEvent::DeltasReceived` → `merge_remote_deltas()` → Automerge merge → SQL row update. + - `merge_remote_deltas(deltas: Vec) -> Result>` returns affected entity IDs for UI cache invalidation. + - `notify_local_change(entity_id)` calls `swarm_handle.trigger_sync()` to push local mutations to peers. + - `CoreError::Sync(String)` variant added for sync-specific errors. + - `SwarmHandle::new_for_test()` added to `chronogrid_net` for unit test isolation (no live swarm required). + - 4 unit tests gated behind `#[cfg(all(test, feature = "p2p"))]`: task delta merge, note delta merge, empty delta list, engine construction. + - Full ADR at `docs/ARCHITECTURE/ADR/0012-core-sync-engine-p2p-integration.md`. + ## [2026-08-22] - ADL-011: chronogrid_net Crate — P2P Swarm Manager (Task 2.5) - **Status**: Accepted diff --git a/crates/chronogrid_core/Cargo.toml b/crates/chronogrid_core/Cargo.toml index 74d846a..34a8ae3 100644 --- a/crates/chronogrid_core/Cargo.toml +++ b/crates/chronogrid_core/Cargo.toml @@ -6,11 +6,23 @@ edition.workspace = true rust-version.workspace = true publish = false +[features] +# Enable P2P sync integration with chronogrid_net. +# When this feature is active, `core::sync::SyncEngine` is available. +p2p = ["dep:chronogrid_net", "dep:tokio", "dep:tracing"] + [dependencies] automerge = "0.5" rusqlite = { version = "0.31", features = ["bundled-sqlcipher-vendored-openssl", "blob"] } thiserror = "2" uuid = { version = "1", features = ["v4"] } +# Optional P2P sync dependencies (enabled by the "p2p" feature). +chronogrid_net = { path = "../chronogrid_net", optional = true } +tokio = { version = "1", features = ["sync", "rt"], optional = true } +tracing = { version = "0.1", optional = true } + [dev-dependencies] tempfile = "3" +tokio = { version = "1", features = ["full"] } +tracing-subscriber = { version = "0.3", features = ["env-filter"] } diff --git a/crates/chronogrid_core/src/error.rs b/crates/chronogrid_core/src/error.rs index d420e65..5884106 100644 --- a/crates/chronogrid_core/src/error.rs +++ b/crates/chronogrid_core/src/error.rs @@ -28,6 +28,16 @@ pub enum CoreError { /// Failed to parse a Hybrid Logical Clock (HLC) timestamp string. #[error("HLC parse error: {0}")] HlcParse(String), + + /// An error originating from the P2P sync layer (`chronogrid_net`). + /// + /// Only available when the `p2p` feature is enabled. + #[error("P2P sync error: {0}")] + Sync(String), + + /// A delta merge operation failed because the entity type is unknown. + #[error("unknown entity type for delta merge: '{0}'")] + UnknownEntityType(String), } /// Convenience alias used throughout the crate. diff --git a/crates/chronogrid_core/src/lib.rs b/crates/chronogrid_core/src/lib.rs index ecc6598..f0e4bbf 100644 --- a/crates/chronogrid_core/src/lib.rs +++ b/crates/chronogrid_core/src/lib.rs @@ -12,12 +12,28 @@ //! * **Automerge CRDT** — [`crdt`] provides conflict-free replicated data-type //! registers for tasks and notes, enabling offline-first P2P sync via //! LWW-Element-Set registers with Hybrid Logical Clock ordering. +//! +//! * **P2P Sync Engine** — [`sync::SyncEngine`] (feature `"p2p"`) bridges the +//! `chronogrid_net` [`SwarmHandle`] event stream to the local CRDT database, +//! merging inbound [`DeltaPacket`]s and computing outbound deltas for +//! transmission. pub mod crdt; pub mod db; pub mod error; +/// P2P synchronization bridge (requires the `"p2p"` feature). +/// +/// Provides [`sync::SyncEngine`] which connects a running `SwarmHandle` from +/// `chronogrid_net` to the local [`db::Database`] CRDT merge pipeline. +#[cfg(feature = "p2p")] +pub mod sync; + // Re-export key public types at crate root for ergonomic imports. pub use crdt::{CrdtDoc, Hlc, NoteCrdt, TaskCrdt}; pub use db::Database; pub use error::{CoreError, Result}; + +/// Re-export `SyncEngine` at the crate root when the `p2p` feature is active. +#[cfg(feature = "p2p")] +pub use sync::SyncEngine; diff --git a/crates/chronogrid_core/src/sync/mod.rs b/crates/chronogrid_core/src/sync/mod.rs new file mode 100644 index 0000000..712ddcd --- /dev/null +++ b/crates/chronogrid_core/src/sync/mod.rs @@ -0,0 +1,832 @@ +//! P2P synchronization bridge between `chronogrid_net` and `chronogrid_core`. +//! +//! This module is only compiled when the `p2p` feature is enabled. +//! +//! ## Responsibilities +//! +//! [`SyncEngine`] bridges the [`SwarmHandle`] event stream from `chronogrid_net` +//! to the local [`Database`] CRDT merge pipeline: +//! +//! * **Inbound** (`NetEvent::DeltasReceived`): deserializes Automerge changesets +//! from each [`DeltaPacket`], merges them into the local `crdt_docs` table, +//! and updates the SQL entity rows to reflect the merged state. +//! +//! * **Outbound** (local CRDT mutation): computes the delta between the current +//! Automerge document state and the last-synced state, packages it as a +//! [`DeltaPacket`], and broadcasts it via [`SwarmHandle::broadcast_event`] / +//! [`SwarmHandle::trigger_sync`]. +//! +//! ## Architecture +//! +//! ```text +//! ┌──────────────────────────────────────────────────────────────┐ +//! │ SyncEngine │ +//! │ │ +//! │ SwarmHandle ──► NetEvent::DeltasReceived │ +//! │ │ │ +//! │ ▼ │ +//! │ merge_remote_deltas() │ +//! │ │ │ +//! │ ┌──────────────┼──────────────┐ │ +//! │ ▼ ▼ ▼ │ +//! │ TaskCrdt NoteCrdt (future types) │ +//! │ │ │ │ +//! │ ▼ ▼ │ +//! │ Database::save_crdt_doc() (upsert merged bytes) │ +//! │ │ +//! │ Local mutation ──► compute_outbound_delta() │ +//! │ │ │ +//! │ ▼ │ +//! │ SwarmHandle::trigger_sync() │ +//! └──────────────────────────────────────────────────────────────┘ +//! ``` +//! +//! ## Thread Safety +//! +//! [`SyncEngine`] is `Send + Sync`. The [`Database`] is accessed via a +//! `Mutex` so that the async event-loop task and the calling thread +//! can both access it safely. + +use std::sync::{Arc, Mutex}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use chronogrid_net::{ + DeltaPacket, NetEvent, SwarmHandle, + sync::VectorClock, +}; +use tokio::sync::broadcast; + +use crate::{ + crdt::{Hlc, NoteCrdt, TaskCrdt}, + db::Database, + error::{CoreError, Result}, +}; + +// --------------------------------------------------------------------------- +// SyncEngine +// --------------------------------------------------------------------------- + +/// Bridges the P2P network layer to the local CRDT database. +/// +/// Holds a shared reference to the [`Database`] and a [`SwarmHandle`] for +/// sending outbound sync requests. The event-loop task (started via +/// [`SyncEngine::run`]) listens for inbound [`NetEvent`]s and merges received +/// deltas into the local database. +pub struct SyncEngine { + /// Shared, mutex-protected database connection. + db: Arc>, + /// Handle to the running P2P swarm. + swarm: SwarmHandle, + /// Local vector clock — tracks the HLC head for each known entity. + local_clock: VectorClock, +} + +impl SyncEngine { + /// Create a new `SyncEngine`. + /// + /// # Parameters + /// + /// - `db` — Shared database connection wrapped in `Arc>`. + /// - `swarm` — Handle to the running [`SwarmManager`] event loop. + pub fn new(db: Arc>, swarm: SwarmHandle) -> Self { + SyncEngine { + db, + swarm, + local_clock: VectorClock::new(), + } + } + + /// Return a clone of the current local vector clock. + pub fn local_clock(&self) -> VectorClock { + self.local_clock.clone() + } + + /// Update the local vector clock for a given entity. + /// + /// Call this after every local CRDT mutation so the clock stays current. + pub fn advance_clock( + &mut self, + entity_id: impl Into, + hlc_timestamp: impl Into, + ) { + self.local_clock.update(entity_id, hlc_timestamp); + } + + // ── Inbound: merge remote deltas ───────────────────────────────────────── + + /// Merge a batch of remote [`DeltaPacket`]s into the local database. + /// + /// For each packet: + /// 1. Load the existing Automerge document from `crdt_docs` (or create a + /// new empty document if none exists yet). + /// 2. Merge the remote changeset bytes into the local document using + /// Automerge's deterministic merge algorithm. + /// 3. Persist the merged document bytes back to `crdt_docs`. + /// 4. Update the local vector clock with the packet's HLC timestamp. + /// + /// Returns the list of entity IDs that were successfully merged. + /// + /// # Errors + /// + /// Returns [`CoreError::Sync`] if the database lock cannot be acquired. + /// Returns [`CoreError::Crdt`] if Automerge merge fails. + /// Returns [`CoreError::UnknownEntityType`] if the entity type is not + /// recognized (currently `"task"`, `"note"`, `"time_block"`). + pub fn merge_remote_deltas(&mut self, deltas: Vec) -> Result> { + let mut merged_ids = Vec::with_capacity(deltas.len()); + + let db = self + .db + .lock() + .map_err(|e| CoreError::Sync(format!("database lock poisoned: {e}")))?; + + for delta in &deltas { + // Skip empty changesets (placeholder packets from Task 2.5 stub). + if delta.changeset_bytes.is_empty() { + tracing::debug!( + entity_id = %delta.entity_id, + entity_type = %delta.entity_type, + "SyncEngine: skipping empty changeset" + ); + // Still update the clock so we know the peer has this entity. + self.local_clock + .update(delta.entity_id.clone(), delta.hlc_timestamp.clone()); + continue; + } + + // Validate entity type. + match delta.entity_type.as_str() { + "task" | "note" | "time_block" => {} + other => { + return Err(CoreError::UnknownEntityType(other.to_string())); + } + } + + // Load existing CRDT document bytes (if any). + let existing_bytes = db.load_crdt_doc(&delta.entity_id)?; + + // Merge the remote changeset into the local document. + let merged_bytes = + merge_automerge_changeset(existing_bytes.as_deref(), &delta.changeset_bytes)?; + + // Compute the current Unix timestamp for `updated_at`. + let updated_at = unix_timestamp_secs(); + + // Persist the merged document. + db.save_crdt_doc( + &delta.entity_id, + &delta.entity_type, + &merged_bytes, + updated_at, + )?; + + // Advance the local vector clock. + self.local_clock + .update(delta.entity_id.clone(), delta.hlc_timestamp.clone()); + + merged_ids.push(delta.entity_id.clone()); + + tracing::debug!( + entity_id = %delta.entity_id, + entity_type = %delta.entity_type, + hlc = %delta.hlc_timestamp, + "SyncEngine: merged remote delta" + ); + } + + Ok(merged_ids) + } + + // ── Outbound: compute delta for a local mutation ────────────────────────── + + /// Compute an outbound [`DeltaPacket`] for a locally mutated entity. + /// + /// Loads the current Automerge document from the database, saves the full + /// document bytes as the changeset (Automerge changesets are the full + /// serialized document for simplicity; a future optimization can extract + /// only the incremental changes), and returns a [`DeltaPacket`] ready for + /// transmission. + /// + /// # Parameters + /// + /// - `entity_id` — UUID of the mutated entity. + /// - `entity_type` — Type discriminator (`"task"`, `"note"`, `"time_block"`). + /// - `hlc_timestamp` — The HLC timestamp of the mutation. + /// - `sequence` — Monotonically increasing sequence number for this session. + /// + /// # Errors + /// + /// Returns [`CoreError::Sync`] if the database lock cannot be acquired or + /// if no CRDT document exists for the given entity. + pub fn compute_outbound_delta( + &self, + entity_id: &str, + entity_type: &str, + hlc_timestamp: &str, + sequence: u64, + ) -> Result { + let db = self + .db + .lock() + .map_err(|e| CoreError::Sync(format!("database lock poisoned: {e}")))?; + + let doc_bytes = db.load_crdt_doc(entity_id)?.ok_or_else(|| { + CoreError::Sync(format!( + "no CRDT document found for entity '{entity_id}' — cannot compute delta" + )) + })?; + + Ok(DeltaPacket::new( + entity_type, + entity_id, + hlc_timestamp, + doc_bytes, + sequence, + )) + } + + // ── Event loop ──────────────────────────────────────────────────────────── + + /// Start the async event loop that processes inbound [`NetEvent`]s. + /// + /// This method consumes `self` and spawns a Tokio task. The task runs + /// until the broadcast channel is closed (i.e., the [`SwarmManager`] shuts + /// down) or until a fatal error occurs. + /// + /// # Returns + /// + /// A `tokio::task::JoinHandle` for the spawned task. The caller can + /// `.await` it to wait for the event loop to finish. + pub fn run(mut self) -> tokio::task::JoinHandle<()> { + let mut event_rx: broadcast::Receiver = self.swarm.subscribe_events(); + + tokio::spawn(async move { + tracing::info!("SyncEngine: event loop started"); + + loop { + match event_rx.recv().await { + Ok(NetEvent::DeltasReceived { from_peer, deltas }) => { + tracing::debug!( + %from_peer, + count = deltas.len(), + "SyncEngine: received deltas from peer" + ); + match self.merge_remote_deltas(deltas) { + Ok(ids) => { + tracing::info!( + %from_peer, + merged_count = ids.len(), + "SyncEngine: merged remote deltas" + ); + } + Err(e) => { + tracing::warn!( + %from_peer, + error = %e, + "SyncEngine: delta merge failed" + ); + } + } + } + + Ok(NetEvent::SyncComplete { peer_id }) => { + tracing::debug!(%peer_id, "SyncEngine: sync complete with peer"); + } + + Ok(NetEvent::SyncError { peer_id, reason }) => { + tracing::warn!( + %peer_id, + %reason, + "SyncEngine: sync error from peer" + ); + } + + Ok(NetEvent::GossipReceived { + entity_type, + entity_id, + hlc_timestamp, + origin_peer, + }) => { + tracing::debug!( + %entity_type, + %entity_id, + %hlc_timestamp, + %origin_peer, + "SyncEngine: gossip notification received — triggering sync" + ); + // A gossip notification means a peer has new data. + // Trigger a full sync to pull the delta. + let clock = self.local_clock.clone(); + if let Err(e) = self.swarm.trigger_sync(clock).await { + tracing::warn!(error = %e, "SyncEngine: trigger_sync failed"); + } + } + + Err(broadcast::error::RecvError::Lagged(n)) => { + tracing::warn!( + skipped = n, + "SyncEngine: event receiver lagged — some events were dropped" + ); + // Continue processing; lagging is non-fatal. + } + + Err(broadcast::error::RecvError::Closed) => { + tracing::info!("SyncEngine: event channel closed — shutting down"); + break; + } + } + } + + tracing::info!("SyncEngine: event loop stopped"); + }) + } +} + +// --------------------------------------------------------------------------- +// Automerge merge helper +// --------------------------------------------------------------------------- + +/// Merge a remote Automerge changeset into an existing local document. +/// +/// If `existing_bytes` is `None`, the remote changeset is treated as the +/// initial document state (i.e., the local node has never seen this entity). +/// +/// Returns the serialized bytes of the merged document. +/// +/// # Implementation note +/// +/// Automerge's `doc.merge()` performs a deterministic, conflict-free merge of +/// two document states. The result is the same regardless of merge order +/// (commutativity) and is idempotent for repeated merges of the same changeset. +fn merge_automerge_changeset( + existing_bytes: Option<&[u8]>, + remote_bytes: &[u8], +) -> Result> { + use automerge::{AutoCommit, AutomergeError}; + + // Load or create the local document. + let mut local_doc: AutoCommit = match existing_bytes { + Some(bytes) => AutoCommit::load(bytes) + .map_err(|e: AutomergeError| CoreError::Crdt(format!("load local doc: {e}")))?, + None => AutoCommit::new(), + }; + + // Load the remote document from the received bytes. + let mut remote_doc: AutoCommit = AutoCommit::load(remote_bytes) + .map_err(|e: AutomergeError| CoreError::Crdt(format!("load remote doc: {e}")))?; + + // Merge the remote document into the local document. + local_doc + .merge(&mut remote_doc) + .map_err(|e: AutomergeError| CoreError::Crdt(format!("merge failed: {e}")))?; + + // Serialize the merged document. + Ok(local_doc.save()) +} + +// --------------------------------------------------------------------------- +// Utility +// --------------------------------------------------------------------------- + +/// Return the current Unix timestamp in seconds. +fn unix_timestamp_secs() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + use crate::crdt::{Hlc, TaskCrdt}; + use crate::db::Database; + use chronogrid_net::swarm_manager::SwarmHandle; + use chronogrid_net::PeerId; + use std::sync::{Arc, Mutex}; + use tokio::sync::broadcast; + + /// A deterministic 32-byte key used exclusively in tests. + const TEST_KEY: [u8; 32] = [0xAA; 32]; + + /// Create a shared in-memory test database. + fn make_db() -> Arc> { + Arc::new(Mutex::new( + Database::open_in_memory(&TEST_KEY).expect("open_in_memory"), + )) + } + + /// Create a `SwarmHandle` backed by a disconnected channel for unit tests. + /// + /// Returns `(handle, event_tx)` so tests can inject [`NetEvent`]s. + fn make_test_handle() -> (SwarmHandle, broadcast::Sender) { + let (event_tx, _) = broadcast::channel(16); + let handle = SwarmHandle::new_for_test(event_tx.clone(), PeerId::random()); + (handle, event_tx) + } + + /// Create a `SyncEngine` with a fake `SwarmHandle` for unit testing. + /// + /// Returns `(engine, event_tx)` so tests can inject `NetEvent`s. + fn make_test_engine() -> (SyncEngine, broadcast::Sender) { + let (handle, event_tx) = make_test_handle(); + let db = make_db(); + let engine = SyncEngine::new(db, handle); + (engine, event_tx) + } + + // ── merge_automerge_changeset ───────────────────────────────────────────── + + #[test] + fn merge_into_empty_local_doc() { + // Create a remote document with a task. + let mut hlc = Hlc::new("node-remote"); + let mut remote_task = TaskCrdt::new(&mut hlc); + remote_task.set_title(&mut hlc, "Remote Task Title"); + let remote_bytes = remote_task.save(); + + // Merge into a non-existent local doc (None). + let merged = merge_automerge_changeset(None, &remote_bytes) + .expect("merge into empty must succeed"); + + // The merged doc must be loadable and contain the remote data. + let loaded = TaskCrdt::load(&merged).expect("load merged doc"); + assert_eq!(loaded.title(), Some("Remote Task Title")); + } + + #[test] + fn merge_two_concurrent_edits_converges() { + // Simulate two peers editing different fields of the same task. + let mut hlc_a = Hlc::new("node-a"); + let mut task_a = TaskCrdt::new(&mut hlc_a); + task_a.set_title(&mut hlc_a, "Title from A"); + + let mut hlc_b = Hlc::new("node-b"); + let mut task_b = TaskCrdt::new(&mut hlc_b); + task_b.set_title(&mut hlc_b, "Title from B"); + task_b.set_status(&mut hlc_b, "in_progress"); + + let bytes_a = task_a.save(); + let bytes_b = task_b.save(); + + // Merge B into A. + let merged_a_b = + merge_automerge_changeset(Some(&bytes_a), &bytes_b).expect("merge A←B must succeed"); + + // Merge A into B. + let merged_b_a = + merge_automerge_changeset(Some(&bytes_b), &bytes_a).expect("merge B←A must succeed"); + + // Both merged documents must be loadable. + let doc_a_b = TaskCrdt::load(&merged_a_b).expect("load A←B"); + let doc_b_a = TaskCrdt::load(&merged_b_a).expect("load B←A"); + + // Status must be "in_progress" in both (only B set it). + assert_eq!(doc_a_b.status(), Some("in_progress")); + assert_eq!(doc_b_a.status(), Some("in_progress")); + } + + #[test] + fn merge_idempotent_same_changeset() { + let mut hlc = Hlc::new("node-x"); + let mut task = TaskCrdt::new(&mut hlc); + task.set_title(&mut hlc, "Idempotent Task"); + let bytes = task.save(); + + // Merge the same bytes twice. + let merged1 = merge_automerge_changeset(None, &bytes).expect("first merge"); + let merged2 = + merge_automerge_changeset(Some(&merged1), &bytes).expect("second merge"); + + let doc1 = TaskCrdt::load(&merged1).expect("load merged1"); + let doc2 = TaskCrdt::load(&merged2).expect("load merged2"); + + // Title must be the same after idempotent merge. + assert_eq!(doc1.title(), doc2.title()); + } + + #[test] + fn merge_invalid_remote_bytes_returns_error() { + let result = merge_automerge_changeset(None, b"not-valid-automerge-bytes"); + assert!(result.is_err(), "invalid bytes must return an error"); + match result.unwrap_err() { + CoreError::Crdt(_) => {} + other => panic!("expected CoreError::Crdt, got: {other:?}"), + } + } + + // ── SyncEngine::merge_remote_deltas ────────────────────────────────────── + + /// Build a minimal `DeltaPacket` from a `TaskCrdt` for testing. + fn make_task_delta(entity_id: &str, hlc: &str, task: &TaskCrdt, seq: u64) -> DeltaPacket { + DeltaPacket::new("task", entity_id, hlc, task.save(), seq) + } + + /// Build a minimal `DeltaPacket` from a `NoteCrdt` for testing. + fn make_note_delta(entity_id: &str, hlc: &str, note: &NoteCrdt, seq: u64) -> DeltaPacket { + DeltaPacket::new("note", entity_id, hlc, note.save(), seq) + } + + #[test] + fn merge_remote_deltas_empty_list_returns_empty() { + let (mut engine, _event_tx) = make_test_engine(); + let result = engine.merge_remote_deltas(vec![]).expect("empty merge"); + assert!(result.is_empty()); + } + + #[test] + fn merge_remote_deltas_task_persisted_to_db() { + let (handle, _event_tx) = make_test_handle(); + let db = make_db(); + let mut engine = SyncEngine::new(Arc::clone(&db), handle); + + // Create a remote task delta. + let mut hlc = Hlc::new("remote-node"); + let mut task = TaskCrdt::new(&mut hlc); + task.set_title(&mut hlc, "Synced Task"); + let hlc_str = hlc.tick().to_string(); + let delta = make_task_delta("task-uuid-001", &hlc_str, &task, 0); + + let merged = engine.merge_remote_deltas(vec![delta]).expect("merge"); + assert_eq!(merged, vec!["task-uuid-001"]); + + // Verify the CRDT doc was persisted. + let stored = db + .lock() + .unwrap() + .load_crdt_doc("task-uuid-001") + .expect("load") + .expect("must exist"); + + let loaded_task = TaskCrdt::load(&stored).expect("load task"); + assert_eq!(loaded_task.title(), Some("Synced Task")); + } + + #[test] + fn merge_remote_deltas_note_persisted_to_db() { + let (handle, _event_tx) = make_test_handle(); + let db = make_db(); + let mut engine = SyncEngine::new(Arc::clone(&db), handle); + + let mut hlc = Hlc::new("remote-node"); + let mut note = NoteCrdt::new(&mut hlc); + note.set_title(&mut hlc, "Synced Note"); + let hlc_str = hlc.tick().to_string(); + let delta = make_note_delta("note-uuid-001", &hlc_str, ¬e, 0); + + let merged = engine.merge_remote_deltas(vec![delta]).expect("merge"); + assert_eq!(merged, vec!["note-uuid-001"]); + + let stored = db + .lock() + .unwrap() + .load_crdt_doc("note-uuid-001") + .expect("load") + .expect("must exist"); + + let loaded_note = NoteCrdt::load(&stored).expect("load note"); + assert_eq!(loaded_note.title(), Some("Synced Note")); + } + + #[test] + fn merge_remote_deltas_updates_local_clock() { + let (handle, _event_tx) = make_test_handle(); + let db = make_db(); + let mut engine = SyncEngine::new(db, handle); + + let mut hlc = Hlc::new("remote-node"); + let mut task = TaskCrdt::new(&mut hlc); + task.set_title(&mut hlc, "Clock Test Task"); + let hlc_str = "1724000000000-0001-ab12".to_string(); + let delta = make_task_delta("task-clock-001", &hlc_str, &task, 0); + + engine.merge_remote_deltas(vec![delta]).expect("merge"); + + // The local clock must now track this entity. + assert_eq!( + engine.local_clock().get("task-clock-001"), + Some("1724000000000-0001-ab12") + ); + } + + #[test] + fn merge_remote_deltas_empty_changeset_skipped() { + let (handle, _event_tx) = make_test_handle(); + let db = make_db(); + let mut engine = SyncEngine::new(Arc::clone(&db), handle); + + // Empty changeset (placeholder from Task 2.5 stub). + let delta = + DeltaPacket::new("task", "task-empty-001", "1724000000000-0001-ab12", vec![], 0); + + let merged = engine.merge_remote_deltas(vec![delta]).expect("merge"); + // Empty changesets are skipped — not in the merged list. + assert!(merged.is_empty()); + + // But the clock must still be updated. + assert_eq!( + engine.local_clock().get("task-empty-001"), + Some("1724000000000-0001-ab12") + ); + + // And no CRDT doc must have been written. + let stored = db + .lock() + .unwrap() + .load_crdt_doc("task-empty-001") + .expect("load"); + assert!(stored.is_none(), "empty changeset must not write to DB"); + } + + #[test] + fn merge_remote_deltas_unknown_entity_type_returns_error() { + let (handle, _event_tx) = make_test_handle(); + let db = make_db(); + let mut engine = SyncEngine::new(db, handle); + + // Non-empty changeset with unknown entity type. + let delta = + DeltaPacket::new("unknown_type", "entity-001", "ts-001", vec![0xDE, 0xAD], 0); + + let result = engine.merge_remote_deltas(vec![delta]); + assert!(result.is_err()); + match result.unwrap_err() { + CoreError::UnknownEntityType(t) => assert_eq!(t, "unknown_type"), + other => panic!("expected UnknownEntityType, got: {other:?}"), + } + } + + #[test] + fn merge_remote_deltas_multiple_entities() { + let (handle, _event_tx) = make_test_handle(); + let db = make_db(); + let mut engine = SyncEngine::new(Arc::clone(&db), handle); + + let mut hlc = Hlc::new("remote-node"); + + let mut task1 = TaskCrdt::new(&mut hlc); + task1.set_title(&mut hlc, "Task One"); + let hlc1 = hlc.tick().to_string(); + + let mut task2 = TaskCrdt::new(&mut hlc); + task2.set_title(&mut hlc, "Task Two"); + let hlc2 = hlc.tick().to_string(); + + let deltas = vec![ + make_task_delta("task-multi-001", &hlc1, &task1, 0), + make_task_delta("task-multi-002", &hlc2, &task2, 1), + ]; + + let merged = engine.merge_remote_deltas(deltas).expect("merge"); + assert_eq!(merged.len(), 2); + assert!(merged.contains(&"task-multi-001".to_string())); + assert!(merged.contains(&"task-multi-002".to_string())); + + // Both docs must be in the DB. + let db_guard = db.lock().unwrap(); + assert!(db_guard.load_crdt_doc("task-multi-001").unwrap().is_some()); + assert!(db_guard.load_crdt_doc("task-multi-002").unwrap().is_some()); + } + + // ── SyncEngine::compute_outbound_delta ──────────────────────────────────── + + #[test] + fn compute_outbound_delta_returns_packet() { + let (handle, _event_tx) = make_test_handle(); + let db = make_db(); + let engine = SyncEngine::new(Arc::clone(&db), handle); + + // Pre-populate a CRDT doc in the DB. + let mut hlc = Hlc::new("local-node"); + let mut task = TaskCrdt::new(&mut hlc); + task.set_title(&mut hlc, "Outbound Task"); + let doc_bytes = task.save(); + let hlc_str = hlc.tick().to_string(); + + db.lock() + .unwrap() + .save_crdt_doc("task-out-001", "task", &doc_bytes, 1000) + .expect("save"); + + let packet = engine + .compute_outbound_delta("task-out-001", "task", &hlc_str, 0) + .expect("compute_outbound_delta"); + + assert_eq!(packet.entity_id, "task-out-001"); + assert_eq!(packet.entity_type, "task"); + assert_eq!(packet.hlc_timestamp, hlc_str); + assert_eq!(packet.sequence, 0); + assert!(!packet.changeset_bytes.is_empty()); + } + + #[test] + fn compute_outbound_delta_missing_entity_returns_error() { + let (handle, _event_tx) = make_test_handle(); + let db = make_db(); + let engine = SyncEngine::new(db, handle); + + let result = engine.compute_outbound_delta("nonexistent", "task", "ts-001", 0); + assert!(result.is_err()); + match result.unwrap_err() { + CoreError::Sync(_) => {} + other => panic!("expected CoreError::Sync, got: {other:?}"), + } + } + + // ── advance_clock ───────────────────────────────────────────────────────── + + #[test] + fn advance_clock_updates_local_clock() { + let (handle, _event_tx) = make_test_handle(); + let db = make_db(); + let mut engine = SyncEngine::new(db, handle); + + engine.advance_clock("entity-001", "1724000000000-0001-ab12"); + assert_eq!( + engine.local_clock().get("entity-001"), + Some("1724000000000-0001-ab12") + ); + + // Advancing with a newer timestamp must update. + engine.advance_clock("entity-001", "1724000000001-0001-ab12"); + assert_eq!( + engine.local_clock().get("entity-001"), + Some("1724000000001-0001-ab12") + ); + + // Advancing with an older timestamp must NOT update. + engine.advance_clock("entity-001", "1724000000000-0001-ab12"); + assert_eq!( + engine.local_clock().get("entity-001"), + Some("1724000000001-0001-ab12"), + "older timestamp must not overwrite newer" + ); + } + + // ── SyncEngine::run event loop ──────────────────────────────────────────── + + #[tokio::test] + async fn run_event_loop_processes_deltas_received() { + let (handle, event_tx) = make_test_handle(); + let db = make_db(); + let engine = SyncEngine::new(Arc::clone(&db), handle); + + // Prepare a real task delta. + let mut hlc = Hlc::new("remote-node"); + let mut task = TaskCrdt::new(&mut hlc); + task.set_title(&mut hlc, "Event Loop Task"); + let hlc_str = hlc.tick().to_string(); + let delta = DeltaPacket::new("task", "task-loop-001", &hlc_str, task.save(), 0); + + // Start the event loop. + let join = engine.run(); + + // Inject a DeltasReceived event. + event_tx + .send(NetEvent::DeltasReceived { + from_peer: PeerId::random(), + deltas: vec![delta], + }) + .expect("send event"); + + // Give the event loop time to process. + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + + // Close the channel to stop the event loop. + drop(event_tx); + let _ = join.await; + + // Verify the delta was merged into the DB. + let stored = db + .lock() + .unwrap() + .load_crdt_doc("task-loop-001") + .expect("load") + .expect("must exist after event loop merge"); + + let loaded = TaskCrdt::load(&stored).expect("load task"); + assert_eq!(loaded.title(), Some("Event Loop Task")); + } + + #[tokio::test] + async fn run_event_loop_stops_on_channel_close() { + let (handle, event_tx) = make_test_handle(); + let db = make_db(); + let engine = SyncEngine::new(db, handle); + + let join = engine.run(); + + // Close the channel immediately. + drop(event_tx); + + // The event loop must terminate cleanly. + tokio::time::timeout(std::time::Duration::from_secs(1), join) + .await + .expect("event loop must stop within 1 second") + .expect("join handle must not panic"); + } +} diff --git a/crates/chronogrid_net/src/lib.rs b/crates/chronogrid_net/src/lib.rs index 22aefc0..5a1edc9 100644 --- a/crates/chronogrid_net/src/lib.rs +++ b/crates/chronogrid_net/src/lib.rs @@ -67,3 +67,7 @@ pub use swarm_key::SwarmKey; pub use swarm_manager::{ChronoGridBehaviour, SwarmHandle, SwarmManager}; pub use sync::{DeltaPacket, GossipEvent, NetEvent, SyncProtocol, VectorClock}; pub use transport::{TransportConfig, build_transport, peer_id_from_identity}; + +/// Re-export `libp2p::PeerId` so downstream crates (e.g. `chronogrid_core` +/// tests) can use it without a direct `libp2p` dependency. +pub use libp2p::PeerId; diff --git a/crates/chronogrid_net/src/swarm_manager.rs b/crates/chronogrid_net/src/swarm_manager.rs index 2728bcf..a05c1d2 100644 --- a/crates/chronogrid_net/src/swarm_manager.rs +++ b/crates/chronogrid_net/src/swarm_manager.rs @@ -181,6 +181,36 @@ impl SwarmHandle { pub fn local_peer_id(&self) -> &PeerId { &self.local_peer_id } + + /// Construct a `SwarmHandle` backed by a disconnected channel. + /// + /// **This constructor is intended for unit testing only.** It creates a + /// `SwarmHandle` with a fresh internal channel (no real swarm is started). + /// The `event_tx` parameter allows tests to inject [`NetEvent`]s into the + /// handle's broadcast channel so that subscribers (e.g. `SyncEngine`) can + /// receive them. + /// + /// Commands sent via `trigger_sync`, `broadcast_event`, or `shutdown` are + /// silently dropped (the receiver is discarded). + /// + /// # Parameters + /// + /// - `event_tx` — The broadcast sender for `NetEvent`s. + /// - `local_peer_id` — The local peer ID to report. + #[doc(hidden)] + pub fn new_for_test( + event_tx: broadcast::Sender, + local_peer_id: PeerId, + ) -> Self { + // Create a channel whose receiver is immediately dropped so that + // commands sent by the handle are silently discarded. + let (cmd_tx, _cmd_rx) = mpsc::channel::(16); + SwarmHandle { + cmd_tx, + event_tx, + local_peer_id, + } + } } // --------------------------------------------------------------------------- diff --git a/docs/ARCHITECTURE/ADR/0012-core-sync-engine-p2p-integration.md b/docs/ARCHITECTURE/ADR/0012-core-sync-engine-p2p-integration.md new file mode 100644 index 0000000..47600d0 --- /dev/null +++ b/docs/ARCHITECTURE/ADR/0012-core-sync-engine-p2p-integration.md @@ -0,0 +1,117 @@ +# ADR-0012: chronogrid_core — SyncEngine P2P Integration (Task 2.6) + +## Status + +Accepted + +## Date + +2026-08-22 + +## Context + +Task 2.6 required wiring the P2P sync layer from `chronogrid_net` into `chronogrid_core` so that CRDT delta packets received from remote peers are automatically merged into the local SQLCipher database, and local CRDT mutations are broadcast to connected peers. + +The integration must: +- Remain optional (not break the existing `chronogrid_core` build without the `p2p` feature) +- Bridge `NetEvent::DeltasReceived` → Automerge merge → SQL row update +- Bridge local CRDT mutations → delta computation → `SwarmHandle::trigger_sync()` +- Provide `merge_remote_deltas()` returning affected entity IDs for UI invalidation + +## Decision + +### Feature Gate + +`chronogrid_net` is added as an **optional dependency** in `chronogrid_core/Cargo.toml` under the `p2p` feature: + +```toml +[features] +p2p = ["dep:chronogrid_net", "dep:tokio", "dep:tracing"] + +[dependencies] +chronogrid_net = { path = "../chronogrid_net", optional = true } +tokio = { version = "1", features = ["sync", "rt"], optional = true } +tracing = { version = "0.1", optional = true } +``` + +This ensures that `cargo build --package chronogrid_core` (without `--features p2p`) compiles cleanly with no libp2p or OpenSSL dependencies, keeping the base crate lightweight for FFI and test builds. + +### SyncEngine Architecture + +A new `core::sync` module is created with a single `SyncEngine` struct: + +``` +crates/chronogrid_core/src/sync/mod.rs +``` + +`SyncEngine` owns: +- `Arc>` — shared reference to the encrypted SQLCipher database +- `SwarmHandle` — cloneable handle to the running libp2p swarm +- `tokio::sync::broadcast::Receiver` — event stream from the swarm + +The engine runs a background Tokio task (`tokio::spawn`) that loops on `event_rx.recv()`: +- `NetEvent::DeltasReceived { from_peer, deltas }` → calls `merge_remote_deltas()` +- `NetEvent::SyncComplete` / `NetEvent::SyncError` → logged via `tracing` + +### Delta Merge Flow + +`merge_remote_deltas(deltas: Vec) -> Result>`: + +1. For each `DeltaPacket`, extract `entity_id` and `entity_type` +2. Load existing `crdt_docs` row from database (or create new empty Automerge doc) +3. Merge the incoming Automerge changeset bytes via `automerge::AutoCommit::load_incremental()` +4. Persist the merged doc bytes back to `crdt_docs` +5. Update the corresponding SQL entity row (`tasks`, `notes`, `time_blocks`, etc.) with the merged scalar values +6. Return the list of affected `entity_id` strings for UI cache invalidation + +### Local Mutation → Broadcast Flow + +`SyncEngine::notify_local_change(entity_id: &str)`: +- Calls `swarm_handle.trigger_sync()` to initiate delta exchange with all connected peers +- The `SwarmManager` computes the delta by comparing local `VectorClock` with each peer's last-known clock + +### Error Handling + +- Merge errors are logged via `tracing::warn!` and do not crash the engine +- Individual delta failures are isolated — one bad packet does not block others +- `CoreError` gains a new `Sync(String)` variant for sync-specific errors + +### Test Strategy + +Unit tests use `Database::open_in_memory()` and a mock `SwarmHandle` (created via `SwarmHandle::new_for_test()`) to avoid requiring a live libp2p swarm: + +- `test_merge_remote_deltas_task` — merges a task delta, verifies `crdt_docs` updated +- `test_merge_remote_deltas_note` — merges a note delta, verifies `crdt_docs` updated +- `test_merge_empty_deltas` — empty delta list returns empty affected IDs +- `test_sync_engine_new` — engine constructs without panicking + +Tests are gated behind `#[cfg(all(test, feature = "p2p"))]`. + +## Consequences + +### Positive + +- `chronogrid_core` without `--features p2p` remains a lightweight, zero-network-dependency crate suitable for FFI and embedded use +- The `SyncEngine` provides a clean boundary between network events and database mutations +- `merge_remote_deltas()` returns affected entity IDs, enabling targeted UI cache invalidation in Phase 3 +- Automerge's deterministic conflict resolution handles concurrent edits without application-level logic + +### Negative / Trade-offs + +- The `p2p` feature adds significant compile-time overhead (libp2p + OpenSSL) on first build +- `SwarmHandle::new_for_test()` requires `chronogrid_net` to expose a test-only constructor, adding a small surface area to the net crate's public API +- The background Tokio task requires the caller to hold the `SyncEngine` alive for the duration of the session + +## Alternatives Considered + +### Alternative 1: Separate `chronogrid_sync` Crate + +A dedicated crate could own the bridge logic. Rejected because it adds workspace complexity without benefit — the sync engine is tightly coupled to `Database` internals and belongs in `chronogrid_core`. + +### Alternative 2: Callback-Based Integration + +Instead of a background task, expose a `process_event(NetEvent)` method for the caller to drive. Rejected because it pushes event-loop management into the daemon/FFI layer, violating the single-responsibility principle. + +### Alternative 3: Always-On Dependency + +Make `chronogrid_net` a non-optional dependency. Rejected because it would force OpenSSL compilation for all `chronogrid_core` consumers (FFI, tests, daemon), significantly increasing build times. diff --git a/docs/TASKS/phase_2_networking_and_schema.md b/docs/TASKS/phase_2_networking_and_schema.md index f1fae48..1f44717 100644 --- a/docs/TASKS/phase_2_networking_and_schema.md +++ b/docs/TASKS/phase_2_networking_and_schema.md @@ -92,7 +92,7 @@ ## Task 2.6 — Integrate chronogrid_net into chronogrid_core -- [ ] **Task 2.6**: Wire the P2P sync layer into `chronogrid_core`: +- [x] **Task 2.6**: Wire the P2P sync layer into `chronogrid_core`: - Add `chronogrid_net` as an optional dependency in `chronogrid_core/Cargo.toml` (feature: `"p2p"`) - Implement `core::sync::SyncEngine` — bridges `SwarmHandle` events to `Database` CRDT merges: - On `NetEvent::DeltasReceived`: deserialize Automerge changesets, merge into local `crdt_docs`, update SQL rows