feat(storage): migration safety nets (failure, restart, idempotence)

Adds the safety-net layer over the v1.x → v2.0 migration tool:

1. migration_in_progress flag in the _meta sub-db, set at the start of
   every migration run and cleared atomically in the same write txn as
   the completion marker. Killed mid-flight, the flag stays set so the
   next boot's auto-detect can log a clear retry warning.
2. Idempotent retry — LMDB put-overwrite semantics make a re-run over
   partially-migrated state correct; the migration logs an explicit
   "previous attempt did not finish — retrying" warning when it sees
   the in-progress flag without a completion marker.
3. Operator-actionable error messages. format_error() embeds:
     - The collection that was being migrated when the failure hit
       (so operators know how far through they got).
     - The underlying cause (e.g. snapshot parse error, LMDB write).
     - The v1.9.5 backup path /var/backups/smartbotic-database/
       pre-upgrade-<ts>-from-1.x/ so rollback is unambiguous.

Tests added (test_migrate_v1_to_v2):
  - Crash-injection: callback throws after 1000 docs out of 2000;
    next run sees in_progress flag, retries, completes with all
    2000 docs intact.
  - in_progress flag is absent after a clean successful migration
    (verifying the atomic completion txn deletes the key).
  - Failure error string contains both the backup path and the
    word "Rollback" — so operators searching logs can find the
    actionable bit at a glance.

Phase C Stage 3 Task 3.4.

service/src/storage/migrate_v1_to_v2.cpp

@@ -56,8 +56,13 @@ namespace {
 constexpr const char* kMetaSubDb         = "_meta";
 constexpr const char* kKeySchemaVersion  = "schema_version";
 constexpr const char* kKeyMigratedAt     = "migration_completed_at";
+constexpr const char* kKeyInProgress     = "migration_in_progress";
 constexpr const char* kSchemaVersionV2   = "2";
 
+// Backup path advertised in error messages (v1.9.5 backup-before-upgrade
+// hook stages the v1.x data dir here, per debian/preinst).
+constexpr const char* kBackupRoot = "/var/backups/smartbotic-database/";
+
 // -------------------------------------------------------------------------
 // LMDB helpers — local subset.
 // -------------------------------------------------------------------------
@@ -96,6 +101,26 @@ std::optional<std::string> read_meta_key(LmdbEnv& env, const char* key) {
     return std::string(static_cast<const char*>(v.mv_data), v.mv_size);
 }
 
+// Set the migration_in_progress flag in a short write txn. Called at the
+// start of every migration run so that a crash mid-flight leaves the flag
+// set; the next attempt clears it as part of mark_complete()'s atomic
+// completion txn.
+void mark_in_progress(LmdbEnv& env) {
+    WriteTxn wtxn(env);
+    MDB_dbi dbi = 0;
+    mdb_check(mdb_dbi_open(wtxn.raw(), kMetaSubDb, MDB_CREATE, &dbi),
+              "dbi_open (_meta create)");
+    std::string val = "1";
+    MDB_val k = to_val(std::string_view(kKeyInProgress));
+    MDB_val v = to_val(val);
+    mdb_check(mdb_put(wtxn.raw(), dbi, &k, &v, 0), "put (_meta in_progress)");
+    wtxn.commit();
+}
+
+// Atomic completion: writes the schema_version + completed_at markers AND
+// deletes the in_progress flag in a single write txn. Either all three
+// changes land or none of them do; on any failure before this txn the
+// in_progress flag stays set and the next boot sees it.
 void mark_complete(LmdbEnv& env) {
     WriteTxn wtxn(env);
     MDB_dbi dbi = 0;
@@ -119,9 +144,35 @@ void mark_complete(LmdbEnv& env) {
         mdb_check(mdb_put(wtxn.raw(), dbi, &k, &v, 0),
                   "put (_meta migration_completed_at)");
     }
+    // Clear in_progress flag — delete the key outright so a future
+    // `migration_in_progress()` returns false cleanly.
+    {
+        MDB_val k = to_val(std::string_view(kKeyInProgress));
+        int rc = mdb_del(wtxn.raw(), dbi, &k, nullptr);
+        if (rc != MDB_SUCCESS && rc != MDB_NOTFOUND) {
+            throw_mdb(rc, "_meta del in_progress");
+        }
+    }
     wtxn.commit();
 }
 
+// Build an operator-actionable error message. Includes the collection
+// that failed (if known), the underlying cause, and the v1.9.5 backup
+// path so operators don't have to grep `dpkg` to find their rollback.
+std::string format_error(const std::string& where_collection,
+                         const std::string& cause) {
+    std::string msg = "v2.0 migration failed";
+    if (!where_collection.empty()) {
+        msg += " (collection '" + where_collection + "')";
+    }
+    msg += ": ";
+    msg += cause;
+    msg += ". Rollback: stop smartbotic-database and restore from ";
+    msg += kBackupRoot;
+    msg += "pre-upgrade-<ts>-from-1.x/.";
+    return msg;
+}
+
 // -------------------------------------------------------------------------
 // v1.x recovery driver — drives SnapshotManager + WriteAheadLog directly
 // rather than instantiating a PersistenceManager (which spins up background
@@ -305,6 +356,16 @@ bool migration_complete(LmdbEnv& env) {
     }
 }
 
+bool migration_in_progress(LmdbEnv& env) {
+    try {
+        auto v = read_meta_key(env, kKeyInProgress);
+        return v.has_value() && !v->empty();
+    } catch (const std::exception& e) {
+        spdlog::warn("migration_in_progress: read failed: {}", e.what());
+        return false;
+    }
+}
+
 AutoMigrateDecision should_run_migration(const std::string& v1_data_dir,
                                           LmdbEnv& env,
                                           const AutoMigrateOptions& opts) {
@@ -363,6 +424,7 @@ MigrationResult migrate_v1_to_v2(
     std::function<void(const MigrationProgress&)> progress_cb) {
 
     MigrationResult result;
+    std::string current_collection;  // tracked so format_error() can name it
     try {
         // 1. Pre-flight: clean install case.
         fs::path snap_dir = fs::path(v1_data_dir) / "snapshots";
@@ -386,7 +448,19 @@ MigrationResult migrate_v1_to_v2(
             return result;
         }
 
-        // 3. Drive v1.x recovery into a temporary MemoryStore.
+        // 3. In-progress detection (Task 3.4). If the previous attempt
+        //    died mid-flight, proceed as a retry — LMDB put-overwrite
+        //    semantics make this safe.
+        if (migration_in_progress(env)) {
+            spdlog::warn("migration: previous attempt did not finish (in_progress "
+                         "flag set) — retrying. Partial state will be overwritten.");
+        }
+
+        // 4. Mark in-progress so a crash before mark_complete() leaves a
+        //    detectable breadcrumb.
+        mark_in_progress(env);
+
+        // 5. Drive v1.x recovery into a temporary MemoryStore.
         smartbotic::database::MemoryStore::Config storeCfg;
         // 8 GB ceiling — migration is bounded by source size and the
         // ceiling is just to keep the in-RAM mirror from triggering
@@ -419,6 +493,7 @@ MigrationResult migrate_v1_to_v2(
         // 5. Copy phase. ONE PUT PER DOC — latency-tolerant.
         uint64_t docs_migrated = 0;
         for (const auto& collection : collections) {
+            current_collection = collection;
             auto docs = mstore.getAllDocuments(collection);
 
             // Some v1.x docs ship without `collection` set; round-trip
@@ -465,8 +540,11 @@ MigrationResult migrate_v1_to_v2(
         }
 
         result.docs_migrated = docs_migrated;
+        current_collection.clear();
 
-        // 6. Final marker write.
+        // 6. Final marker write. Single write txn atomically sets the
+        //    schema_version=2 + completed_at markers and clears the
+        //    in_progress flag.
         mark_complete(env);
 
         if (progress_cb) {
@@ -486,12 +564,15 @@ MigrationResult migrate_v1_to_v2(
         return result;
     } catch (const std::exception& e) {
         result.success = false;
-        result.error = std::string("v2.0 migration failed: ") + e.what();
+        // format_error embeds the failed-collection name (if known), the
+        // underlying cause, and the v1.9.5 backup path for rollback. The
+        // in_progress flag stays set — next boot's auto-detect retries.
+        result.error = format_error(current_collection, e.what());
         spdlog::error("migration: FAILED — {}", result.error);
         return result;
     } catch (...) {
         result.success = false;
-        result.error = "v2.0 migration failed: unknown exception";
+        result.error = format_error(current_collection, "unknown exception");
         spdlog::error("migration: FAILED — {}", result.error);
         return result;
     }

service/src/storage/migrate_v1_to_v2.hpp

@@ -72,6 +72,12 @@ MigrationResult migrate_v1_to_v2(
 // any thread.
 bool migration_complete(LmdbEnv& env);
 
+// True if env's _meta sub-db has an in-progress marker but no completion
+// marker — i.e. the previous migration attempt died mid-flight. Callers
+// should re-run migrate_v1_to_v2() to make progress (it overwrites
+// partial state cleanly via LMDB put-overwrite semantics).
+bool migration_in_progress(LmdbEnv& env);
+
 // -------------------------------------------------------------------------
 // Task 3.2 — auto-detect-on-boot helper.
 // -------------------------------------------------------------------------

tests/test_migrate_v1_to_v2.cpp

@@ -51,6 +51,7 @@ using smartbotic::db::storage::LmdbEnv;
 using smartbotic::db::storage::LmdbEnvOpts;
 using smartbotic::db::storage::migrate_v1_to_v2;
 using smartbotic::db::storage::migration_complete;
+using smartbotic::db::storage::migration_in_progress;
 using smartbotic::db::storage::MigrationProgress;
 using smartbotic::db::storage::MigrationResult;
 using smartbotic::db::storage::should_run_migration;
@@ -656,6 +657,112 @@ void test_migrate_non_vector_collection_no_sub_db() {
     std::cout << "PASS: non-vector collection has no vector sub-db\n";
 }
 
+// -------------------------------------------------------------------------
+// Task 3.4: Safety nets + crash-injection
+// -------------------------------------------------------------------------
+
+// Inject a callback that throws after N docs to simulate crash mid-migration.
+class CrashAfterN {
+public:
+    explicit CrashAfterN(uint64_t n) : trigger_at_(n) {}
+    void operator()(const MigrationProgress& p) {
+        if (p.docs_migrated >= trigger_at_ && !triggered_) {
+            triggered_ = true;
+            throw std::runtime_error("injected crash at " +
+                                     std::to_string(p.docs_migrated));
+        }
+    }
+private:
+    uint64_t trigger_at_;
+    bool triggered_ = false;
+};
+
+void test_kill_during_migration_then_restart() {
+    auto dir = make_tmpdir("crash");
+    {
+        V1Builder b(dir);
+        b.createCollection("c");
+        for (int i = 0; i < 2000; i++) {
+            b.insert("c", makeDoc("d" + std::to_string(i), {{"i", i}}));
+        }
+        b.writeSnapshot();
+    }
+
+    TmpEnv env_dir("crash-env");
+    LmdbDocumentStore target(env_dir.env);
+
+    // First attempt — throws after 1000 docs.
+    CrashAfterN crasher(1000);
+    auto r1 = migrate_v1_to_v2(dir, target, env_dir.env,
+        [&crasher](const MigrationProgress& p) { crasher(p); });
+    check(!r1.success, "first attempt fails (injected crash)");
+    check(!migration_complete(env_dir.env), "marker NOT set after crash");
+    check(migration_in_progress(env_dir.env), "in_progress flag set after crash");
+
+    // Second attempt — no crasher, should run to completion and overwrite
+    // the partial state cleanly via LMDB put-overwrite semantics.
+    auto r2 = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(r2.success, "second attempt succeeds");
+    check(migration_complete(env_dir.env), "marker set after success");
+    check(!migration_in_progress(env_dir.env),
+          "in_progress flag cleared on success");
+    check(target.count("c") == 2000, "all 2000 docs present after retry");
+
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: crash during migration + restart\n";
+}
+
+void test_migration_in_progress_flag_clears_on_success() {
+    auto dir = make_tmpdir("inprog-clear");
+    {
+        V1Builder b(dir);
+        b.createCollection("c");
+        b.insert("c", makeDoc("d1", {{"v", 1}}));
+        b.writeSnapshot();
+    }
+    TmpEnv env_dir("inprog-clear-env");
+    LmdbDocumentStore target(env_dir.env);
+    auto r = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(r.success, "migration succeeds");
+    check(!migration_in_progress(env_dir.env),
+          "in_progress flag absent after successful migration");
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: in_progress flag clears on success\n";
+}
+
+void test_migration_failure_includes_backup_path_in_error() {
+    auto dir = make_tmpdir("err-msg");
+    {
+        V1Builder b(dir);
+        b.createCollection("c");
+        b.insert("c", makeDoc("d1", {{"v", 1}}));
+        b.writeSnapshot();
+    }
+    // Corrupt the snapshot so the load step fails.
+    for (auto& e : fs::directory_iterator(fs::path(dir) / "snapshots")) {
+        if (e.is_regular_file()) {
+            uintmax_t sz = fs::file_size(e.path());
+            fs::resize_file(e.path(), sz / 2);
+            break;
+        }
+    }
+
+    TmpEnv env_dir("err-msg-env");
+    LmdbDocumentStore target(env_dir.env);
+    auto r = migrate_v1_to_v2(dir, target, env_dir.env);
+    check(!r.success, "corrupt snapshot fails");
+    check(r.error.find("/var/backups/smartbotic-database/") != std::string::npos,
+          "error message includes backup path");
+    check(r.error.find("Rollback") != std::string::npos,
+          "error message mentions rollback");
+
+    std::error_code ec;
+    fs::remove_all(dir, ec);
+    std::cout << "PASS: failure error includes backup path\n";
+}
+
 // -------------------------------------------------------------------------
 // main()
 // -------------------------------------------------------------------------
@@ -683,6 +790,11 @@ int main() {
     test_migrate_collection_with_partial_vectors();
     test_migrate_non_vector_collection_no_sub_db();
 
+    // Task 3.4: Safety nets + crash-injection
+    test_kill_during_migration_then_restart();
+    test_migration_in_progress_flag_clears_on_success();
+    test_migration_failure_includes_backup_path_in_error();
+
     std::cout << "\nAll migrate_v1_to_v2 tests passed.\n";
     return 0;
 }