From d06f5b5410d854051ca418bd73f3ba3881179885 Mon Sep 17 00:00:00 2001 From: Huskies Agent Date: Tue, 21 Jul 2026 14:37:27 +0000 Subject: [PATCH] huskies: merge 1241 bug CRDT snapshot has no schema migration; a failed load silently starts empty --- server/src/crdt_state/state/init.rs | 62 ++++++++++++++++++++++++++--- 1 file changed, 57 insertions(+), 5 deletions(-) diff --git a/server/src/crdt_state/state/init.rs b/server/src/crdt_state/state/init.rs index 1a787e32..5a73a115 100644 --- a/server/src/crdt_state/state/init.rs +++ b/server/src/crdt_state/state/init.rs @@ -73,15 +73,19 @@ pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> { ); 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 doc: PipelineDoc = match serde_json::from_str(&state_json) { + Ok(doc) => doc, + Err(e) => { + eprintln!("\n{}\n", snapshot_load_abort_message(&e.to_string())); + std::process::exit(1); + } + }; let mut crdt = BaseCrdt::::new(&kp); crdt.doc = doc; - Ok::<_, String>(crdt) + crdt }) .await - .map_err(|e| sqlx::Error::Protocol(format!("snapshot restore panicked: {e}")))? - .map_err(sqlx::Error::Protocol)?; + .map_err(|e| sqlx::Error::Protocol(format!("snapshot restore panicked: {e}")))?; // Replay only ops that arrived after the snapshot. let tail_rows: Vec<(String,)> = @@ -426,6 +430,23 @@ async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt, lamport_ } } +/// 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 +/// included verbatim so the operator can see exactly what changed. +fn snapshot_load_abort_message(deserialize_error: &str) -> String { + format!( + "error: failed to load the CRDT snapshot (crdt_snapshot row):\n \ + {deserialize_error}\n\n\ + This snapshot was written by a binary with a different schema and is \ + missing a field the current binary expects.\n\ + No data has been lost: the ops log (crdt_ops) is intact and is the \ + durable source of truth — the snapshot is only a fast-path replay cache.\n\ + To recover: delete the crdt_snapshot row (e.g. `DELETE FROM crdt_snapshot \ + WHERE id = 1;`) so init.rs rebuilds state from crdt_ops on the next start." + ) +} + /// Extract the filesystem path from a SqlitePool's connect options. fn pool_path(pool: &SqlitePool) -> Option { use sqlx::ConnectOptions; @@ -433,3 +454,34 @@ fn pool_path(pool: &SqlitePool) -> Option { let filename = opts.get_filename(); filename.to_str().map(|s| s.to_string()) } + +#[cfg(test)] +mod tests { + use super::snapshot_load_abort_message; + + /// The abort message must name the failing field (via the serde error + /// text), explain that the ops log is intact, and give the exact + /// recovery step — this is what the operator reads at 3am. + #[test] + fn snapshot_load_abort_message_names_field_and_explains_recovery() { + let err = "missing field `gateway_config` at line 1 column 4821"; + let msg = snapshot_load_abort_message(err); + + assert!( + msg.contains("missing field `gateway_config`"), + "message must name the failing field verbatim: {msg}" + ); + assert!( + msg.contains("different schema"), + "message must explain the schema-mismatch cause: {msg}" + ); + assert!( + msg.contains("crdt_ops") && msg.contains("intact"), + "message must state the ops log is intact and no data is lost: {msg}" + ); + assert!( + msg.contains("DELETE FROM crdt_snapshot"), + "message must give the exact recovery step: {msg}" + ); + } +}