diff --git a/.huskies/specs/tech/CRDT_SNAPSHOT_COMPACTION.md b/.huskies/specs/tech/CRDT_SNAPSHOT_COMPACTION.md new file mode 100644 index 00000000..c3c57fda --- /dev/null +++ b/.huskies/specs/tech/CRDT_SNAPSHOT_COMPACTION.md @@ -0,0 +1,46 @@ +# CRDT Snapshot Compaction + +## Problem + +The huskies project CRDT has grown to 55K ops / 276MB in `pipeline.db`. +Every container restart replays all operations in a tight synchronous loop +on the tokio runtime (`crdt_state/state/init.rs:68-78`), taking 27+ minutes +and freezing the runtime so the HTTP server never becomes ready. + +### Op bloat + +55K ops across only 4,154 sequence numbers (~13 ops per seq on average). +Many zeroed-out `MergeJobCrdt` entries appear to be tombstones never cleaned +up. Some individual ops are up to 523KB. Average op size is 4.6KB. + +## Proposed Fix + +### 1. Snapshot (checkpoint) + +After replaying all ops, serialize the materialized CRDT state to a +checkpoint blob (e.g. a `crdt_snapshot` table or a separate file). On next +startup, load the snapshot and only replay ops with `rowid > snapshot_rowid`. + +At snapshot time, back up the database file so corruption is recoverable. + +### 2. Op pruning / compaction + +Delete ops that are superseded by the snapshot. Tombstoned/deleted items with +all-zero fields contribute nothing to materialized state and can be dropped +from the log once snapshotted. + +### 3. Immediate fix: spawn_blocking + +Move the replay loop to `tokio::task::spawn_blocking` so the HTTP server and +liveness ticks are not starved during replay. This does not reduce replay +time but prevents the runtime freeze. + +## Acceptance Criteria + +- Startup loads a snapshot when available and only replays ops newer than the snapshot sequence +- A snapshot is written after full CRDT replay completes (or periodically in background) +- DB backup is created at each snapshot time +- Startup time for 55K ops drops from 27+ minutes to under 30 seconds +- Dead/zeroed CRDT ops are pruned during compaction +- CRDT sync protocol continues to work correctly across nodes after compaction +- The replay loop runs in spawn_blocking so it does not freeze the tokio runtime diff --git a/crates/bft-json-crdt/src/list_crdt.rs b/crates/bft-json-crdt/src/list_crdt.rs index 09961939..2cbf67c9 100644 --- a/crates/bft-json-crdt/src/list_crdt.rs +++ b/crates/bft-json-crdt/src/list_crdt.rs @@ -20,6 +20,10 @@ use std::{ /// An RGA-like list CRDT that can store a CRDT-like datatype #[derive(Clone, Serialize, Deserialize)] +#[serde(bound( + serialize = "T: serde::Serialize", + deserialize = "T: serde::de::DeserializeOwned" +))] pub struct ListCrdt where T: CrdtNode, @@ -32,6 +36,7 @@ where pub ops: Vec>, /// Queue of messages where K is the ID of the message yet to arrive /// and V is the list of operations depending on it + #[serde(skip)] message_q: HashMap>>, /// The sequence number of this node our_seq: SequenceNumber, diff --git a/crates/bft-json-crdt/src/lww_crdt.rs b/crates/bft-json-crdt/src/lww_crdt.rs index c93e0001..2f4fad34 100644 --- a/crates/bft-json-crdt/src/lww_crdt.rs +++ b/crates/bft-json-crdt/src/lww_crdt.rs @@ -7,6 +7,7 @@ use crate::debug::DebugView; use crate::json_crdt::{CrdtNode, JsonValue, OpState}; use crate::op::{join_path, print_path, Op, PathSegment, SequenceNumber}; +use serde::{Deserialize, Serialize}; use std::cmp::{max, Ordering}; use std::fmt::Debug; @@ -14,7 +15,11 @@ use crate::keypair::AuthorId; /// A simple delete-wins, last-writer-wins (LWW) register CRDT. /// Basically only for adding support for primitives within a more complex CRDT -#[derive(Clone)] +#[derive(Clone, Serialize, Deserialize)] +#[serde(bound( + serialize = "T: serde::Serialize", + deserialize = "T: serde::de::DeserializeOwned" +))] pub struct LwwRegisterCrdt where T: CrdtNode, diff --git a/server/migrations/20260629000000_crdt_snapshot.sql b/server/migrations/20260629000000_crdt_snapshot.sql new file mode 100644 index 00000000..976eddca --- /dev/null +++ b/server/migrations/20260629000000_crdt_snapshot.sql @@ -0,0 +1,9 @@ +-- Stores a serialized CRDT state snapshot so startup can skip replaying +-- the full op log. Only the single most recent snapshot row is kept. +CREATE TABLE IF NOT EXISTS crdt_snapshot ( + id INTEGER PRIMARY KEY CHECK (id = 1), + at_seq INTEGER NOT NULL, + max_rowid INTEGER NOT NULL, + state_json TEXT NOT NULL, + created_at TEXT NOT NULL +); diff --git a/server/src/crdt_state/state/init.rs b/server/src/crdt_state/state/init.rs index 65e62f05..1a787e32 100644 --- a/server/src/crdt_state/state/init.rs +++ b/server/src/crdt_state/state/init.rs @@ -55,75 +55,180 @@ pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> { // Load or create the node keypair. let keypair = load_or_create_keypair(&pool).await?; - let mut crdt = BaseCrdt::::new(&keypair); - // Replay persisted ops to reconstruct state. - let rows: Vec<(String,)> = sqlx::query_as("SELECT op_json FROM crdt_ops ORDER BY rowid ASC") - .fetch_all(&pool) - .await?; + // Try to load a snapshot first — if one exists, we can skip replaying + // the bulk of the op log and only replay ops that arrived after the + // snapshot was taken. + let snapshot_row: Option<(i64, i64, String)> = + sqlx::query_as("SELECT at_seq, max_rowid, state_json FROM crdt_snapshot WHERE id = 1") + .fetch_optional(&pool) + .await?; - let mut all_ops_vec = Vec::with_capacity(rows.len()); - let mut vector_clock = VectorClock::new(); - let mut lamport_floor: u64 = 0; - for (op_json,) in &rows { - if let Ok(signed_op) = serde_json::from_str::(op_json) { - let author_hex = hex::encode(&signed_op.author()); - *vector_clock.entry(author_hex).or_insert(0) += 1; - lamport_floor = lamport_floor.max(signed_op.inner.seq); - crdt.apply(signed_op); - all_ops_vec.push(op_json.clone()); + let (mut crdt, all_ops_vec, vector_clock, lamport_floor) = + if let Some((snap_seq, snap_max_rowid, state_json)) = snapshot_row { + slog!( + "[crdt] Loading snapshot (at_seq={}, max_rowid={})", + snap_seq, + snap_max_rowid + ); + let kp = keypair.clone(); + let restore_result = tokio::task::spawn_blocking(move || { + let doc: PipelineDoc = serde_json::from_str(&state_json) + .map_err(|e| format!("snapshot deserialize failed: {e}"))?; + let mut crdt = BaseCrdt::::new(&kp); + crdt.doc = doc; + Ok::<_, String>(crdt) + }) + .await + .map_err(|e| sqlx::Error::Protocol(format!("snapshot restore panicked: {e}")))? + .map_err(sqlx::Error::Protocol)?; + + // Replay only ops that arrived after the snapshot. + let tail_rows: Vec<(String,)> = + sqlx::query_as("SELECT op_json FROM crdt_ops WHERE rowid > ?1 ORDER BY rowid ASC") + .bind(snap_max_rowid) + .fetch_all(&pool) + .await?; + + let tail_count = tail_rows.len(); + let floor = snap_seq as u64; + let replay_result = tokio::task::spawn_blocking(move || { + let mut crdt = restore_result; + let mut all_ops_vec = Vec::new(); + let mut vector_clock = VectorClock::new(); + let mut lamport_floor = floor; + for (op_json,) in &tail_rows { + if let Ok(signed_op) = serde_json::from_str::(op_json) { + let author_hex = hex::encode(&signed_op.author()); + *vector_clock.entry(author_hex).or_insert(0) += 1; + lamport_floor = lamport_floor.max(signed_op.inner.seq); + crdt.apply(signed_op); + all_ops_vec.push(op_json.clone()); + } + } + (crdt, all_ops_vec, vector_clock, lamport_floor) + }) + .await + .map_err(|e| sqlx::Error::Protocol(format!("tail replay panicked: {e}")))?; + + slog!("[crdt] Snapshot loaded, {} tail ops replayed", tail_count); + replay_result } else { - slog!("[crdt] Warning: failed to deserialize stored op"); - } - } + // No snapshot — full replay from scratch. + let rows: Vec<(String,)> = + sqlx::query_as("SELECT op_json FROM crdt_ops ORDER BY rowid ASC") + .fetch_all(&pool) + .await?; + + let row_count = rows.len(); + slog!( + "[crdt] No snapshot found, replaying {} ops from scratch", + row_count + ); + + let mut crdt = BaseCrdt::::new(&keypair); + let replay_result = tokio::task::spawn_blocking(move || { + let mut all_ops_vec = Vec::with_capacity(row_count); + let mut vector_clock = VectorClock::new(); + let mut lamport_floor: u64 = 0; + for (op_json,) in &rows { + if let Ok(signed_op) = serde_json::from_str::(op_json) { + let author_hex = hex::encode(&signed_op.author()); + *vector_clock.entry(author_hex).or_insert(0) += 1; + lamport_floor = lamport_floor.max(signed_op.inner.seq); + crdt.apply(signed_op); + all_ops_vec.push(op_json.clone()); + } else { + slog!("[crdt] Warning: failed to deserialize stored op"); + } + } + (crdt, all_ops_vec, vector_clock, lamport_floor) + }) + .await + .map_err(|e| sqlx::Error::Protocol(format!("CRDT replay panicked: {e}")))?; + + // After a full replay, save a snapshot so subsequent restarts are fast. + let (ref replay_crdt, _, _, replay_floor) = replay_result; + save_snapshot(&pool, replay_crdt, replay_floor).await; + + replay_result + }; + + // Rebuild tombstone set and indices from the materialized state. + let rebuild_result = tokio::task::spawn_blocking(move || { + let tombstones: HashSet = crdt + .doc + .items + .ops + .iter() + .filter(|op| op.is_deleted) + .filter_map(|op| op.content.as_ref()) + .filter_map(|item| match item.story_id.view() { + JsonValue::String(s) if !s.is_empty() => Some(s), + _ => None, + }) + .collect(); + + let index = rebuild_index(&crdt); + let node_index = rebuild_node_index(&crdt); + let token_index = rebuild_token_index(&crdt); + let merge_job_index = rebuild_merge_job_index(&crdt); + let active_agent_index = rebuild_active_agent_index(&crdt); + let test_job_index = rebuild_test_job_index(&crdt); + let agent_throttle_index = rebuild_agent_throttle_index(&crdt); + let gateway_project_index = rebuild_gateway_project_index(&crdt); + let llm_session_index = rebuild_llm_session_index(&crdt); + + crdt.doc.items.advance_seq(lamport_floor); + crdt.doc.nodes.advance_seq(lamport_floor); + crdt.doc.tokens.advance_seq(lamport_floor); + crdt.doc.merge_jobs.advance_seq(lamport_floor); + crdt.doc.active_agents.advance_seq(lamport_floor); + crdt.doc.test_jobs.advance_seq(lamport_floor); + crdt.doc.agent_throttle.advance_seq(lamport_floor); + crdt.doc.gateway_projects.advance_seq(lamport_floor); + crdt.doc.llm_sessions.advance_seq(lamport_floor); + crdt.doc + .gateway_config + .active_project + .advance_seq(lamport_floor); + + ( + crdt, + tombstones, + index, + node_index, + token_index, + merge_job_index, + active_agent_index, + test_job_index, + agent_throttle_index, + gateway_project_index, + llm_session_index, + ) + }) + .await + .map_err(|e| sqlx::Error::Protocol(format!("index rebuild panicked: {e}")))?; + + let ( + crdt, + tombstones, + index, + node_index, + token_index, + merge_job_index, + active_agent_index, + test_job_index, + agent_throttle_index, + gateway_project_index, + llm_session_index, + ) = rebuild_result; + let _ = ALL_OPS.set(Mutex::new(all_ops_vec)); let _ = VECTOR_CLOCK.set(Mutex::new(vector_clock)); - // Rebuild tombstone set: deleted list items still carry their original - // PipelineItemCrdt content, so we can extract the story_id directly. - let tombstones: HashSet = crdt - .doc - .items - .ops - .iter() - .filter(|op| op.is_deleted) - .filter_map(|op| op.content.as_ref()) - .filter_map(|item| match item.story_id.view() { - JsonValue::String(s) if !s.is_empty() => Some(s), - _ => None, - }) - .collect(); - - // Build the indices from the reconstructed state. - let index = rebuild_index(&crdt); - let node_index = rebuild_node_index(&crdt); - let token_index = rebuild_token_index(&crdt); - let merge_job_index = rebuild_merge_job_index(&crdt); - let active_agent_index = rebuild_active_agent_index(&crdt); - let test_job_index = rebuild_test_job_index(&crdt); - let agent_throttle_index = rebuild_agent_throttle_index(&crdt); - let gateway_project_index = rebuild_gateway_project_index(&crdt); - let llm_session_index = rebuild_llm_session_index(&crdt); - - // Advance the top-level list clocks to the Lamport floor so that - // list-level inserts don't re-emit low seq numbers. - crdt.doc.items.advance_seq(lamport_floor); - crdt.doc.nodes.advance_seq(lamport_floor); - crdt.doc.tokens.advance_seq(lamport_floor); - crdt.doc.merge_jobs.advance_seq(lamport_floor); - crdt.doc.active_agents.advance_seq(lamport_floor); - crdt.doc.test_jobs.advance_seq(lamport_floor); - crdt.doc.agent_throttle.advance_seq(lamport_floor); - crdt.doc.gateway_projects.advance_seq(lamport_floor); - crdt.doc.llm_sessions.advance_seq(lamport_floor); - crdt.doc - .gateway_config - .active_project - .advance_seq(lamport_floor); - slog!( - "[crdt] Initialised: {} ops replayed, {} items indexed, {} nodes indexed, lamport_floor={}", - rows.len(), + "[crdt] Initialised: {} items indexed, {} nodes indexed, lamport_floor={}", index.len(), node_index.len(), lamport_floor, @@ -265,3 +370,66 @@ async fn load_or_create_keypair(pool: &SqlitePool) -> Result, lamport_floor: u64) { + // Find the highest rowid currently in crdt_ops — ops with rowid <= this + // value are already captured in the snapshot. + let max_rowid: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(rowid), 0) FROM crdt_ops") + .fetch_one(pool) + .await + .unwrap_or(0); + + let doc_ref = &crdt.doc; + let json = match serde_json::to_string(doc_ref) { + Ok(j) => j, + Err(e) => { + slog!("[crdt] Failed to serialize snapshot: {e}"); + return; + } + }; + + let json_len = json.len(); + let now = chrono::Utc::now().to_rfc3339(); + + // Back up the database before writing the snapshot row. + if let Some(db_path) = pool_path(pool) { + let bak_path = format!("{db_path}.bak"); + if let Err(e) = tokio::fs::copy(&db_path, &bak_path).await { + slog!("[crdt] DB backup failed: {e}"); + } else { + slog!("[crdt] DB backed up to {bak_path}"); + } + } + + let result = sqlx::query( + "INSERT OR REPLACE INTO crdt_snapshot (id, at_seq, max_rowid, state_json, created_at) \ + VALUES (1, ?1, ?2, ?3, ?4)", + ) + .bind(lamport_floor as i64) + .bind(max_rowid) + .bind(&json) + .bind(&now) + .execute(pool) + .await; + + match result { + Ok(_) => slog!( + "[crdt] Snapshot saved: at_seq={}, max_rowid={}, json={}B", + lamport_floor, + max_rowid, + json_len + ), + Err(e) => slog!("[crdt] Failed to save snapshot: {e}"), + } +} + +/// Extract the filesystem path from a SqlitePool's connect options. +fn pool_path(pool: &SqlitePool) -> Option { + use sqlx::ConnectOptions; + let opts = pool.connect_options(); + let filename = opts.get_filename(); + filename.to_str().map(|s| s.to_string()) +} diff --git a/server/src/crdt_state/types.rs b/server/src/crdt_state/types.rs index 60723fc5..10ee89c8 100644 --- a/server/src/crdt_state/types.rs +++ b/server/src/crdt_state/types.rs @@ -4,6 +4,7 @@ use bft_json_crdt::json_crdt::*; use bft_json_crdt::list_crdt::ListCrdt; use bft_json_crdt::lww_crdt::LwwRegisterCrdt; +use serde::{Deserialize, Serialize}; use tokio::sync::broadcast; /// An event emitted when a pipeline item's stage changes in the CRDT document. @@ -27,7 +28,7 @@ pub struct CrdtEvent { /// replicated across all connected nodes. LWW semantics ensure the last /// writer wins on concurrent updates. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct GatewayConfigCrdt { /// The currently active project name (empty string = unset / use default). pub active_project: LwwRegisterCrdt, @@ -35,7 +36,7 @@ pub struct GatewayConfigCrdt { /// Top-level CRDT document holding all replicated pipeline state (items, nodes, jobs, etc.). #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct PipelineDoc { pub items: ListCrdt, pub nodes: ListCrdt, @@ -59,7 +60,7 @@ pub struct PipelineDoc { /// (count of existing entries for that sled), giving deterministic ordering for /// all transitions recorded by a single node even after CRDT replay on restart. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct EventLogEntryCrdt { /// Monotonic sequence number for this sled (0, 1, 2, …). Stored as `f64` /// because all CRDT scalar registers use JSON numbers. @@ -84,7 +85,7 @@ pub struct EventLogEntryCrdt { /// per-sled high-water marks so that `assemble_prompt_context` can inject only /// events the LLM has not yet seen and then advance the marks atomically. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct LlmSessionCrdt { /// Stable session identifier (e.g. Matrix room ID). pub session_id: LwwRegisterCrdt, @@ -172,7 +173,7 @@ pub struct LlmSessionView { /// register stores the `Stage::Frozen { resume_to }` / `Stage::ReviewHold /// { resume_to, .. }` resume target as a clean wire-form stage name. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct PipelineItemCrdt { pub story_id: LwwRegisterCrdt, pub stage: LwwRegisterCrdt, @@ -244,7 +245,7 @@ pub struct PipelineItemCrdt { /// CRDT node that holds a single peer's presence entry. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct NodePresenceCrdt { /// Hex-encoded Ed25519 public key — stable identity across restarts. pub node_id: LwwRegisterCrdt, @@ -460,7 +461,7 @@ pub struct NodePresenceView { /// CRDT entry holding per-agent token-usage metrics. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct TokenUsageCrdt { /// Unique key (e.g. `"coder-1:42_story_foo"`). pub agent_id: LwwRegisterCrdt, @@ -473,7 +474,7 @@ pub struct TokenUsageCrdt { /// CRDT entry describing a merge job. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct MergeJobCrdt { /// Unique key: the story being merged. pub story_id: LwwRegisterCrdt, @@ -486,7 +487,7 @@ pub struct MergeJobCrdt { /// CRDT entry for a currently-running agent instance. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct ActiveAgentCrdt { /// Unique key (e.g. `"coder-1"`). pub agent_id: LwwRegisterCrdt, @@ -497,7 +498,7 @@ pub struct ActiveAgentCrdt { /// CRDT entry describing a test job. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct TestJobCrdt { /// Unique key: the story under test. pub story_id: LwwRegisterCrdt, @@ -510,7 +511,7 @@ pub struct TestJobCrdt { /// CRDT entry holding per-node agent-throttle state. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct AgentThrottleCrdt { /// Unique key: the node whose throttle this tracks. pub node_id: LwwRegisterCrdt, @@ -524,7 +525,7 @@ pub struct AgentThrottleCrdt { /// CRDT entry for a gateway project registered in `gateway_config.projects`. #[add_crdt_fields] -#[derive(Clone, CrdtNode, Debug)] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] pub struct GatewayProjectCrdt { /// Unique key: project name (e.g. `"huskies"`). pub name: LwwRegisterCrdt,