diff --git a/server/src/config/mod.rs b/server/src/config/mod.rs index 29f43cc8..5ebba6a5 100644 --- a/server/src/config/mod.rs +++ b/server/src/config/mod.rs @@ -155,6 +155,12 @@ pub struct ProjectConfig { /// (disabled) — the on-demand `gc` MCP tool remains available regardless. #[serde(default)] pub gc_min_free_gb: u64, + /// Number of CRDT ops applied (locally or received from sync peers) + /// between periodic snapshot checkpoints (story 1249). Keeps the + /// replayed tail on startup bounded instead of growing forever between + /// the one-time post-replay snapshot and the next restart. Default: 5000. + #[serde(default = "default_snapshot_interval_ops")] + pub snapshot_interval_ops: usize, } /// Configuration for the filesystem watcher's sweep behaviour. @@ -255,6 +261,11 @@ fn default_max_mesh_peers() -> usize { 3 } +/// Default number of ops between periodic CRDT snapshot checkpoints (story 1249). +pub fn default_snapshot_interval_ops() -> usize { + 5000 +} + /// Configuration for the low-disk-space watchdog's free-space thresholds. /// /// Sleds check free space on the `/workspace` filesystem each tick and @@ -472,6 +483,7 @@ impl Default for ProjectConfig { status_push_enabled: default_status_push_enabled(), merge_failure_block_threshold: default_merge_failure_block_threshold(), gc_min_free_gb: 0, + snapshot_interval_ops: default_snapshot_interval_ops(), } } } @@ -564,6 +576,7 @@ impl ProjectConfig { status_push_enabled: default_status_push_enabled(), merge_failure_block_threshold: default_merge_failure_block_threshold(), gc_min_free_gb: 0, + snapshot_interval_ops: default_snapshot_interval_ops(), }; validate_agents(&config.agent)?; return Ok(config); @@ -607,6 +620,7 @@ impl ProjectConfig { status_push_enabled: default_status_push_enabled(), merge_failure_block_threshold: default_merge_failure_block_threshold(), gc_min_free_gb: 0, + snapshot_interval_ops: default_snapshot_interval_ops(), }; validate_agents(&config.agent)?; Ok(config) @@ -638,6 +652,7 @@ impl ProjectConfig { status_push_enabled: default_status_push_enabled(), merge_failure_block_threshold: default_merge_failure_block_threshold(), gc_min_free_gb: 0, + snapshot_interval_ops: default_snapshot_interval_ops(), }) } } diff --git a/server/src/config/tests.rs b/server/src/config/tests.rs index e1050ea8..a3816a8e 100644 --- a/server/src/config/tests.rs +++ b/server/src/config/tests.rs @@ -11,6 +11,21 @@ fn default_config_when_missing() { assert!(config.component.is_empty()); } +#[test] +fn snapshot_interval_ops_defaults_to_5000() { + let config = ProjectConfig::default(); + assert_eq!(config.snapshot_interval_ops, 5000); +} + +#[test] +fn snapshot_interval_ops_overridable() { + let toml_str = r#" +snapshot_interval_ops = 250 +"#; + let config = ProjectConfig::parse(toml_str).unwrap(); + assert_eq!(config.snapshot_interval_ops, 250); +} + #[test] fn parse_multi_agent_toml() { let toml_str = r#" diff --git a/server/src/crdt_state/mod.rs b/server/src/crdt_state/mod.rs index bd16b4b3..4aa9ba53 100644 --- a/server/src/crdt_state/mod.rs +++ b/server/src/crdt_state/mod.rs @@ -49,7 +49,7 @@ pub use read::{ read_all_items, read_item, tombstoned_ids, }; pub(crate) use state::flush_persistence; -pub use state::{init, subscribe}; +pub use state::{checkpoint_on_shutdown, init, subscribe}; pub use types::{ ActiveAgentCrdt, ActiveAgentView, AgentThrottleCrdt, AgentThrottleView, CrdtEvent, EpicId, EventLogEntryCrdt, GatewayConfigCrdt, GatewayProjectCrdt, GatewayProjectView, LlmSessionCrdt, diff --git a/server/src/crdt_state/state/init.rs b/server/src/crdt_state/state/init.rs index 5a73a115..136e6c1a 100644 --- a/server/src/crdt_state/state/init.rs +++ b/server/src/crdt_state/state/init.rs @@ -24,7 +24,7 @@ use super::indices::{ rebuild_index, rebuild_llm_session_index, rebuild_merge_job_index, rebuild_node_index, rebuild_test_job_index, rebuild_token_index, }; -use super::statics::{ALL_OPS, CRDT_EVENT_TX, PERSIST_PENDING, SYNC_TX, VECTOR_CLOCK}; +use super::statics::{self, ALL_OPS, CRDT_EVENT_TX, PERSIST_PENDING, SYNC_TX, VECTOR_CLOCK}; use super::{CRDT_STATE, CrdtState}; use crate::slog; @@ -41,8 +41,11 @@ pub(crate) enum PersistMsg { /// Opens the SQLite database, loads or creates a node keypair, replays any /// persisted ops to reconstruct state, and spawns a background persistence /// task. Safe to call only once; subsequent calls are no-ops. +/// +/// `snapshot_interval_ops` (story 1249) is the number of ops applied between +/// periodic snapshot checkpoints — see `ProjectConfig::snapshot_interval_ops`. #[allow(clippy::string_slice)] // op_id is hex::encode output (ASCII-only), &op_id[..12] is always valid -pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> { +pub async fn init(db_path: &Path, snapshot_interval_ops: usize) -> Result<(), sqlx::Error> { if CRDT_STATE.get().is_some() { return Ok(()); } @@ -52,6 +55,8 @@ pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> { .create_if_missing(true); let pool = SqlitePool::connect_with(options).await?; sqlx::migrate!("./migrations").run(&pool).await?; + let _ = statics::CRDT_POOL.set(pool.clone()); + let _ = statics::SNAPSHOT_INTERVAL_OPS.set(snapshot_interval_ops); // Load or create the node keypair. let keypair = load_or_create_keypair(&pool).await?; @@ -274,6 +279,8 @@ pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> { slog!("[crdt] Failed to persist op {}: {e}", &op_id[..12]); } PERSIST_PENDING.fetch_sub(1, Ordering::Relaxed); + + maybe_periodic_checkpoint().await; } PersistMsg::Flush(reply) => { // All ops queued before this message have already been processed. @@ -379,21 +386,27 @@ 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) { + let json = match serde_json::to_string(&crdt.doc) { Ok(j) => j, Err(e) => { slog!("[crdt] Failed to serialize snapshot: {e}"); return; } }; + save_snapshot_json(pool, &json, lamport_floor).await; +} + +/// Shared snapshot-writing body: back up the DB file and write `json` (an +/// already-serialized `PipelineDoc`) into the `crdt_snapshot` table. Used by +/// both the one-time post-replay snapshot ([`save_snapshot`]) and periodic / +/// shutdown checkpoints ([`checkpoint_now`]) added by story 1249. +async fn save_snapshot_json(pool: &SqlitePool, json: &str, at_seq: 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 json_len = json.len(); let now = chrono::Utc::now().to_rfc3339(); @@ -412,9 +425,9 @@ async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt, lamport_ "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(at_seq as i64) .bind(max_rowid) - .bind(&json) + .bind(json) .bind(&now) .execute(pool) .await; @@ -422,7 +435,7 @@ async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt, lamport_ match result { Ok(_) => slog!( "[crdt] Snapshot saved: at_seq={}, max_rowid={}, json={}B", - lamport_floor, + at_seq, max_rowid, json_len ), @@ -430,6 +443,69 @@ async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt, lamport_ } } +/// Pure threshold decision for periodic checkpoints (story 1249 AC2/AC3). +/// +/// Returns `true` only when at least one op has been applied since the last +/// checkpoint AND that count has reached `threshold`. The `ops_since_last > +/// 0` guard is what makes AC3 (skip when nothing changed) hold even for a +/// `threshold` of 0 or 1. +fn should_checkpoint(ops_since_last: usize, threshold: usize) -> bool { + ops_since_last > 0 && ops_since_last >= threshold +} + +/// Called after every op the persistence task writes. Triggers a checkpoint +/// once `OPS_SINCE_SNAPSHOT` reaches the configured `SNAPSHOT_INTERVAL_OPS` +/// (story 1249 AC1/AC2). +async fn maybe_periodic_checkpoint() { + let threshold = statics::SNAPSHOT_INTERVAL_OPS + .get() + .copied() + .unwrap_or(crate::config::default_snapshot_interval_ops()); + let ops_since_last = statics::OPS_SINCE_SNAPSHOT.load(Ordering::Relaxed); + if should_checkpoint(ops_since_last, threshold) { + checkpoint_now("periodic").await; + } +} + +/// Take a snapshot of the current CRDT state, unless no ops have been +/// applied since the last checkpoint (story 1249 AC3). Resets +/// `OPS_SINCE_SNAPSHOT` back to zero afterwards. `reason` is only used for +/// the log line (e.g. `"periodic"` or `"shutdown"`). +async fn checkpoint_now(reason: &str) { + if statics::OPS_SINCE_SNAPSHOT.load(Ordering::Relaxed) == 0 { + return; + } + let Some(pool) = statics::CRDT_POOL.get() else { + return; + }; + let Some(state_mutex) = super::get_crdt() else { + return; + }; + let json = { + let Ok(state) = state_mutex.lock() else { + return; + }; + match serde_json::to_string(&state.crdt.doc) { + Ok(j) => j, + Err(e) => { + slog!("[crdt] Failed to serialize {reason} checkpoint: {e}"); + return; + } + } + }; + let at_seq = statics::LAST_SEQ.load(Ordering::Relaxed); + save_snapshot_json(pool, &json, at_seq).await; + statics::OPS_SINCE_SNAPSHOT.store(0, Ordering::Relaxed); + slog!("[crdt] Checkpoint ({reason}) complete"); +} + +/// Take a final checkpoint on clean shutdown (story 1249 AC1), if any ops +/// have been applied since the last one. Safe to call even if the CRDT +/// layer was never initialised — it's a no-op in that case. +pub async fn checkpoint_on_shutdown() { + checkpoint_now("shutdown").await; +} + /// Build the abort message printed when a CRDT snapshot fails to deserialize /// into the current schema. `deserialize_error` is the `Display` text of the /// `serde_json::Error`, which names the missing/mismatched field; it is @@ -457,7 +533,25 @@ fn pool_path(pool: &SqlitePool) -> Option { #[cfg(test)] mod tests { - use super::snapshot_load_abort_message; + use super::{should_checkpoint, snapshot_load_abort_message}; + + /// AC3: a checkpoint must never fire when nothing has changed, no matter + /// how low the configured threshold is. + #[test] + fn should_checkpoint_false_when_no_ops_applied() { + assert!(!should_checkpoint(0, 5000)); + assert!(!should_checkpoint(0, 1)); + assert!(!should_checkpoint(0, 0)); + } + + /// AC2: the threshold is whatever N is configured to, not a hardcoded value. + #[test] + fn should_checkpoint_respects_configured_threshold() { + assert!(!should_checkpoint(4999, 5000)); + assert!(should_checkpoint(5000, 5000)); + assert!(should_checkpoint(5001, 5000)); + assert!(should_checkpoint(1, 1)); + } /// The abort message must name the failing field (via the serde error /// text), explain that the ops log is intact, and give the exact diff --git a/server/src/crdt_state/state/mod.rs b/server/src/crdt_state/state/mod.rs index af3672ba..5a0b10ce 100644 --- a/server/src/crdt_state/state/mod.rs +++ b/server/src/crdt_state/state/mod.rs @@ -26,8 +26,8 @@ mod tests; // ── Re-exports for crdt_state siblings ────────────────────────────── -pub use init::init; pub(crate) use init::{PersistMsg, flush_persistence}; +pub use init::{checkpoint_on_shutdown, init}; /// Subscribe to CRDT state-transition events. /// diff --git a/server/src/crdt_state/state/statics.rs b/server/src/crdt_state/state/statics.rs index 78e1026e..e908b722 100644 --- a/server/src/crdt_state/state/statics.rs +++ b/server/src/crdt_state/state/statics.rs @@ -10,10 +10,11 @@ //! tests do not share `ALL_OPS` — preventing one test's `apply_compaction` //! from pruning another test's freshly-written ops. -use std::sync::atomic::AtomicUsize; +use std::sync::atomic::{AtomicU64, AtomicUsize}; use std::sync::{Mutex, OnceLock}; use bft_json_crdt::json_crdt::SignedOp; +use sqlx::SqlitePool; use tokio::sync::broadcast; use super::super::VectorClock; @@ -46,6 +47,31 @@ pub(crate) static ALL_OPS: OnceLock>> = OnceLock::new(); /// re-parsing all ops when a peer requests `our_vector_clock()`. pub(crate) static VECTOR_CLOCK: OnceLock> = OnceLock::new(); +/// The CRDT SQLite pool, stashed here so periodic and shutdown checkpoints +/// (story 1249) can reuse it without threading it through every call site +/// that might trigger one. Set once, at the end of `init::init()`. +pub(crate) static CRDT_POOL: OnceLock = OnceLock::new(); + +/// Number of ops between periodic snapshot checkpoints (story 1249). +/// Set once, from `ProjectConfig::snapshot_interval_ops`, at the end of +/// `init::init()`. Falls back to `config::default_snapshot_interval_ops()` +/// (5000) if `init()` hasn't set it yet. +pub(crate) static SNAPSHOT_INTERVAL_OPS: OnceLock = OnceLock::new(); + +/// Count of ops applied (locally created or received from sync peers) since +/// the last snapshot checkpoint. Incremented in [`track_op`]; reset to zero +/// after a checkpoint is taken. A checkpoint is skipped when this is zero +/// (story 1249 AC3) so restarting the server or an idle period never writes +/// a redundant snapshot. +pub(crate) static OPS_SINCE_SNAPSHOT: AtomicUsize = AtomicUsize::new(0); + +/// Highest op sequence number seen since startup, updated in [`track_op`]. +/// Used as the informational `at_seq` value for periodic/shutdown +/// checkpoints (the replay-boundary correctness depends only on +/// `max_rowid`, computed fresh from SQLite at snapshot time — this is just +/// for accurate logging/debugging). +pub(crate) static LAST_SEQ: AtomicU64 = AtomicU64::new(0); + #[cfg(test)] thread_local! { /// Per-thread op journal for test isolation. Each test thread sees its @@ -107,4 +133,6 @@ pub(in crate::crdt_state) fn track_op(signed: &SignedOp, json: String) { let author_hex = hex::encode(&signed.author()); *clock.entry(author_hex).or_insert(0) += 1; } + OPS_SINCE_SNAPSHOT.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + LAST_SEQ.fetch_max(signed.inner.seq, std::sync::atomic::Ordering::Relaxed); } diff --git a/server/src/gateway/mod.rs b/server/src/gateway/mod.rs index 7d5e27bd..1eab3c56 100644 --- a/server/src/gateway/mod.rs +++ b/server/src/gateway/mod.rs @@ -94,7 +94,9 @@ pub async fn run(config_path: &Path, port: u16) -> Result<(), std::io::Error> { // Initialise the CRDT so gateway_config.active_project is persisted across restarts. let crdt_db = config_dir.join("gateway.db"); - if let Err(e) = crate::crdt_state::init(&crdt_db).await { + if let Err(e) = + crate::crdt_state::init(&crdt_db, crate::config::default_snapshot_interval_ops()).await + { crate::slog!( "[gateway] Warning: CRDT init failed ({e}); active-project selection will not persist" ); @@ -146,6 +148,13 @@ pub async fn run(config_path: &Path, port: u16) -> Result<(), std::io::Error> { .run(route) .await; + // Story 1249: take a final CRDT snapshot on clean shutdown (mirrors main.rs). + crate::crdt_state::checkpoint_on_shutdown().await; + + // Story 1249: take a final CRDT snapshot on clean shutdown so the next + // restart's replayed tail stays bounded. + crate::crdt_state::checkpoint_on_shutdown().await; + // Best-effort shutdown notification: signal the Matrix bot so it can post // "going offline" before the process exits. Mirror of main.rs:346. { diff --git a/server/src/main.rs b/server/src/main.rs index 6ef56bbb..23dd04ee 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -491,6 +491,11 @@ async fn main() -> Result<(), std::io::Error> { let result = Server::new(TcpListener::bind(&addr)).run(app).await; + // Story 1249: take a final CRDT snapshot on clean shutdown so the next + // restart's replayed tail stays bounded, even if fewer than + // `snapshot_interval_ops` ops have accumulated since the last periodic one. + crate::crdt_state::checkpoint_on_shutdown().await; + // ── Shutdown notifications (best-effort) ────────────────────────────────── startup::bots::notify_shutdown(&bot_ctxs).await; diff --git a/server/src/startup/project.rs b/server/src/startup/project.rs index 9274f9e6..ca68bc9a 100644 --- a/server/src/startup/project.rs +++ b/server/src/startup/project.rs @@ -338,7 +338,15 @@ pub(crate) async fn init_subsystems(app_state: &Arc, cwd: &Path, i let huskies_dir = db_path.parent().unwrap_or(db_path); migrate_json_stores_to_sqlite(huskies_dir).await; } - if let Err(e) = crdt_state::init(db_path).await { + let snapshot_interval_ops = app_state + .project_root + .lock() + .unwrap() + .as_ref() + .and_then(|root| config::ProjectConfig::load(root).ok()) + .map(|cfg| cfg.snapshot_interval_ops) + .unwrap_or_else(config::default_snapshot_interval_ops); + if let Err(e) = crdt_state::init(db_path, snapshot_interval_ops).await { crate::slog!("[crdt] Failed to initialise CRDT state layer: {e}"); } else { crdt_state::migrate_names_from_slugs(); diff --git a/server/src/worktree/cleanup.rs b/server/src/worktree/cleanup.rs index e5e56b69..f0248d40 100644 --- a/server/src/worktree/cleanup.rs +++ b/server/src/worktree/cleanup.rs @@ -233,6 +233,7 @@ mod tests { status_push_enabled: true, merge_failure_block_threshold: 3, gc_min_free_gb: 0, + snapshot_interval_ops: 5000, } } diff --git a/server/src/worktree/create.rs b/server/src/worktree/create.rs index bc4b7bfa..9ae1e4a9 100644 --- a/server/src/worktree/create.rs +++ b/server/src/worktree/create.rs @@ -272,6 +272,7 @@ mod tests { status_push_enabled: true, merge_failure_block_threshold: 3, gc_min_free_gb: 0, + snapshot_interval_ops: 5000, } } diff --git a/server/src/worktree/remove.rs b/server/src/worktree/remove.rs index e1aa0610..3cc01489 100644 --- a/server/src/worktree/remove.rs +++ b/server/src/worktree/remove.rs @@ -130,6 +130,7 @@ mod tests { status_push_enabled: true, merge_failure_block_threshold: 3, gc_min_free_gb: 0, + snapshot_interval_ops: 5000, } } diff --git a/server/src/worktree/sweep.rs b/server/src/worktree/sweep.rs index 17931bb0..31397695 100644 --- a/server/src/worktree/sweep.rs +++ b/server/src/worktree/sweep.rs @@ -154,6 +154,7 @@ mod tests { status_push_enabled: true, merge_failure_block_threshold: 3, gc_min_free_gb: 0, + snapshot_interval_ops: 5000, } }