Compare commits
No commits in common. "9d3bd8811ff456f3a1d5fc7ba41bd61ef91dbf66" and "7a8736b2973cbd0d85ed5ef65a4791200bb20ed8" have entirely different histories.
9d3bd8811f
...
7a8736b297
|
|
@ -461,13 +461,9 @@ name = "chronogrid_core"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"automerge",
|
"automerge",
|
||||||
"chronogrid_net",
|
|
||||||
"rusqlite",
|
"rusqlite",
|
||||||
"tempfile",
|
"tempfile",
|
||||||
"thiserror 2.0.20",
|
"thiserror 2.0.20",
|
||||||
"tokio",
|
|
||||||
"tracing",
|
|
||||||
"tracing-subscriber",
|
|
||||||
"uuid",
|
"uuid",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -125,21 +125,6 @@
|
||||||
- 100 unit tests pass + 5 doc-tests pass (35 new sync tests + 65 from Tasks 2.2/2.3).
|
- 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`.
|
- 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<Mutex<Database>>`, `SwarmHandle`, and a `broadcast::Receiver<NetEvent>`.
|
|
||||||
- Background Tokio task loops on `event_rx.recv()`: `NetEvent::DeltasReceived` → `merge_remote_deltas()` → Automerge merge → SQL row update.
|
|
||||||
- `merge_remote_deltas(deltas: Vec<DeltaPacket>) -> Result<Vec<String>>` 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)
|
## [2026-08-22] - ADL-011: chronogrid_net Crate — P2P Swarm Manager (Task 2.5)
|
||||||
|
|
||||||
- **Status**: Accepted
|
- **Status**: Accepted
|
||||||
|
|
|
||||||
|
|
@ -6,23 +6,11 @@ edition.workspace = true
|
||||||
rust-version.workspace = true
|
rust-version.workspace = true
|
||||||
publish = false
|
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]
|
[dependencies]
|
||||||
automerge = "0.5"
|
automerge = "0.5"
|
||||||
rusqlite = { version = "0.31", features = ["bundled-sqlcipher-vendored-openssl", "blob"] }
|
rusqlite = { version = "0.31", features = ["bundled-sqlcipher-vendored-openssl", "blob"] }
|
||||||
thiserror = "2"
|
thiserror = "2"
|
||||||
uuid = { version = "1", features = ["v4"] }
|
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]
|
[dev-dependencies]
|
||||||
tempfile = "3"
|
tempfile = "3"
|
||||||
tokio = { version = "1", features = ["full"] }
|
|
||||||
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
|
|
||||||
|
|
|
||||||
|
|
@ -28,16 +28,6 @@ pub enum CoreError {
|
||||||
/// Failed to parse a Hybrid Logical Clock (HLC) timestamp string.
|
/// Failed to parse a Hybrid Logical Clock (HLC) timestamp string.
|
||||||
#[error("HLC parse error: {0}")]
|
#[error("HLC parse error: {0}")]
|
||||||
HlcParse(String),
|
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.
|
/// Convenience alias used throughout the crate.
|
||||||
|
|
|
||||||
|
|
@ -12,28 +12,12 @@
|
||||||
//! * **Automerge CRDT** — [`crdt`] provides conflict-free replicated data-type
|
//! * **Automerge CRDT** — [`crdt`] provides conflict-free replicated data-type
|
||||||
//! registers for tasks and notes, enabling offline-first P2P sync via
|
//! registers for tasks and notes, enabling offline-first P2P sync via
|
||||||
//! LWW-Element-Set registers with Hybrid Logical Clock ordering.
|
//! 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 crdt;
|
||||||
pub mod db;
|
pub mod db;
|
||||||
pub mod error;
|
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.
|
// Re-export key public types at crate root for ergonomic imports.
|
||||||
pub use crdt::{CrdtDoc, Hlc, NoteCrdt, TaskCrdt};
|
pub use crdt::{CrdtDoc, Hlc, NoteCrdt, TaskCrdt};
|
||||||
pub use db::Database;
|
pub use db::Database;
|
||||||
pub use error::{CoreError, Result};
|
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;
|
|
||||||
|
|
|
||||||
|
|
@ -1,832 +0,0 @@
|
||||||
//! 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<Database>` 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<Mutex<Database>>,
|
|
||||||
/// 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<Mutex<_>>`.
|
|
||||||
/// - `swarm` — Handle to the running [`SwarmManager`] event loop.
|
|
||||||
pub fn new(db: Arc<Mutex<Database>>, 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<String>,
|
|
||||||
hlc_timestamp: impl Into<String>,
|
|
||||||
) {
|
|
||||||
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<DeltaPacket>) -> Result<Vec<String>> {
|
|
||||||
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<DeltaPacket> {
|
|
||||||
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<NetEvent> = 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<Vec<u8>> {
|
|
||||||
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<Mutex<Database>> {
|
|
||||||
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<NetEvent>) {
|
|
||||||
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<NetEvent>) {
|
|
||||||
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");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -67,7 +67,3 @@ pub use swarm_key::SwarmKey;
|
||||||
pub use swarm_manager::{ChronoGridBehaviour, SwarmHandle, SwarmManager};
|
pub use swarm_manager::{ChronoGridBehaviour, SwarmHandle, SwarmManager};
|
||||||
pub use sync::{DeltaPacket, GossipEvent, NetEvent, SyncProtocol, VectorClock};
|
pub use sync::{DeltaPacket, GossipEvent, NetEvent, SyncProtocol, VectorClock};
|
||||||
pub use transport::{TransportConfig, build_transport, peer_id_from_identity};
|
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;
|
|
||||||
|
|
|
||||||
|
|
@ -181,36 +181,6 @@ impl SwarmHandle {
|
||||||
pub fn local_peer_id(&self) -> &PeerId {
|
pub fn local_peer_id(&self) -> &PeerId {
|
||||||
&self.local_peer_id
|
&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<NetEvent>,
|
|
||||||
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::<SwarmCommand>(16);
|
|
||||||
SwarmHandle {
|
|
||||||
cmd_tx,
|
|
||||||
event_tx,
|
|
||||||
local_peer_id,
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|
|
||||||
|
|
@ -1,117 +0,0 @@
|
||||||
# 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<Mutex<Database>>` — shared reference to the encrypted SQLCipher database
|
|
||||||
- `SwarmHandle` — cloneable handle to the running libp2p swarm
|
|
||||||
- `tokio::sync::broadcast::Receiver<NetEvent>` — 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<DeltaPacket>) -> Result<Vec<String>>`:
|
|
||||||
|
|
||||||
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.
|
|
||||||
|
|
@ -92,7 +92,7 @@
|
||||||
|
|
||||||
## Task 2.6 — Integrate chronogrid_net into chronogrid_core
|
## Task 2.6 — Integrate chronogrid_net into chronogrid_core
|
||||||
|
|
||||||
- [x] **Task 2.6**: Wire the P2P sync layer into `chronogrid_core`:
|
- [ ] **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"`)
|
- 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:
|
- 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
|
- On `NetEvent::DeltasReceived`: deserialize Automerge changesets, merge into local `crdt_docs`, update SQL rows
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue