huskies: merge 1241 bug CRDT snapshot has no schema migration; a failed load silently starts empty

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::<PipelineDoc>::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<PipelineDoc>, 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<String> {
     use sqlx::ConnectOptions;
@@ -433,3 +454,34 @@ fn pool_path(pool: &SqlitePool) -> Option<String> {
     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}"
+        );
+    }
+}