Added CRDT snapshotting

.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

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<T>
 where
     T: CrdtNode,
@@ -32,6 +36,7 @@ where
     pub ops: Vec<Op<T>>,
     /// 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<OpId, Vec<Op<T>>>,
     /// The sequence number of this node
     our_seq: SequenceNumber,

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<T>
 where
     T: CrdtNode,

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
+);

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::<PipelineDoc>::new(&keypair);
 
-    // Replay persisted ops to reconstruct state.
+    // Try to load a snapshot first — if one exists, we can skip replaying
-    let rows: Vec<(String,)> = sqlx::query_as("SELECT op_json FROM crdt_ops ORDER BY rowid ASC")
+    // the bulk of the op log and only replay ops that arrived after the
-        .fetch_all(&pool)
+    // snapshot was taken.
-        .await?;
+    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 crdt, all_ops_vec, vector_clock, lamport_floor) =
-    let mut vector_clock = VectorClock::new();
+        if let Some((snap_seq, snap_max_rowid, state_json)) = snapshot_row {
-    let mut lamport_floor: u64 = 0;
+            slog!(
-    for (op_json,) in &rows {
+                "[crdt] Loading snapshot (at_seq={}, max_rowid={})",
-        if let Ok(signed_op) = serde_json::from_str::<SignedOp>(op_json) {
+                snap_seq,
-            let author_hex = hex::encode(&signed_op.author());
+                snap_max_rowid
-            *vector_clock.entry(author_hex).or_insert(0) += 1;
+            );
-            lamport_floor = lamport_floor.max(signed_op.inner.seq);
+            let kp = keypair.clone();
-            crdt.apply(signed_op);
+            let restore_result = tokio::task::spawn_blocking(move || {
-            all_ops_vec.push(op_json.clone());
+                let doc: PipelineDoc = serde_json::from_str(&state_json)
+                    .map_err(|e| format!("snapshot deserialize failed: {e}"))?;
+                let mut crdt = BaseCrdt::<PipelineDoc>::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::<SignedOp>(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::<PipelineDoc>::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::<SignedOp>(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<String> = 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<String> = 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={}",
+        "[crdt] Initialised: {} items indexed, {} nodes indexed, lamport_floor={}",
-        rows.len(),
         index.len(),
         node_index.len(),
         lamport_floor,
@@ -265,3 +370,66 @@ async fn load_or_create_keypair(pool: &SqlitePool) -> Result<Ed25519KeyPair, sql
 
     Ok(kp)
 }
+
+/// Serialize the current CRDT document state into the `crdt_snapshot` table
+/// and back up the database file.  Errors are logged but not propagated —
+/// a failed snapshot just means the next restart will do a full replay.
+async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt<PipelineDoc>, 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<String> {
+    use sqlx::ConnectOptions;
+    let opts = pool.connect_options();
+    let filename = opts.get_filename();
+    filename.to_str().map(|s| s.to_string())
+}

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<String>,
@@ -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<PipelineItemCrdt>,
     pub nodes: ListCrdt<NodePresenceCrdt>,
@@ -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<String>,
@@ -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<String>,
     pub stage: LwwRegisterCrdt<String>,
@@ -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<String>,
@@ -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<String>,
@@ -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<String>,
@@ -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<String>,
@@ -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<String>,
@@ -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<String>,
@@ -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<String>,