feat(recovery): auto-readonly on non-trivial; refuse-to-start on failure

DatabaseService now consumes the RecoveryOutcome from recover() and:
- Throws on isFailure() -- caller (main.cpp) will exit with code 10
- Enters read-only mode on isNonTrivial() unless --force-readwrite is set
- Stores the outcome for GetReadOnlyStatus RPC

Adds isReadOnly/readOnlyReason/setReadOnly/recoveryOutcome/setForceReadwrite
public methods. read_only_ is atomic for lock-free reads on the hot path.

Also promotes PersistenceManager::Config to a member of DatabaseService::Config
(persistenceConfig) so the CLI parser / config loader in later tasks can set
recoveryMode without having to reach into setupComponents(). The simpler
top-level fields (walSyncIntervalMs, snapshotIntervalSec, compressionEnabled)
are still overlaid for backward compatibility.

Healthy deployments (TrivialSuccess) see no behavior change.

service/src/database_service.cpp

@@ -16,6 +16,26 @@ DatabaseService::~DatabaseService() {
     stop();
 }
 
+std::string DatabaseService::readOnlyReason() const {
+    std::lock_guard<std::mutex> lock(reason_mutex_);
+    return read_only_reason_;
+}
+
+void DatabaseService::setReadOnly(bool value, const std::string& reason) {
+    {
+        std::lock_guard<std::mutex> lock(reason_mutex_);
+        read_only_reason_ = value ? reason : "";
+    }
+    bool prev = read_only_.exchange(value, std::memory_order_acq_rel);
+    if (prev != value) {
+        if (value) {
+            spdlog::error("Database entered READ-ONLY mode: {}", reason);
+        } else {
+            spdlog::info("Database unlocked -- writes accepted");
+        }
+    }
+}
+
 bool DatabaseService::initialize() {
     spdlog::info("Initializing database service (node: {})", config_.nodeId);
 
@@ -39,7 +59,75 @@ bool DatabaseService::initialize() {
         }
 
         // Recover from persistence
-        persistence_->recover(*store_);
+        recovery_outcome_ = persistence_->recover(*store_);
+
+        if (recovery_outcome_.isFailure()) {
+            const std::string modeStr =
+                recoveryModeToString(config_.persistenceConfig.recoveryMode);
+            spdlog::error("");
+            spdlog::error("+------------------------------------------------------------------+");
+            spdlog::error("| RECOVERY REFUSED                                                 |");
+            spdlog::error("|                                                                  |");
+            spdlog::error("| Mode:              {}", modeStr);
+            spdlog::error("| Expected snapshot: {}", recovery_outcome_.expectedSnapshot.string());
+            spdlog::error("| Reason:            {}", recovery_outcome_.failureReason);
+            spdlog::error("|                                                                  |");
+            spdlog::error("| Snapshots available:  {}", recovery_outcome_.snapshotsAvailable);
+            spdlog::error("| Snapshots attempted:  {}", recovery_outcome_.snapshotsAttempted);
+            spdlog::error("|                                                                  |");
+            spdlog::error("| To escalate, restart with one of:                                |");
+            spdlog::error("|   --recovery-mode=snapshot_fallback   Try older snapshots        |");
+            spdlog::error("|   --recovery-mode=wal_only            Replay WAL only (slow)     |");
+            spdlog::error("|   --recovery-mode=best_effort         Try all of the above       |");
+            spdlog::error("|   --recovery-mode=force_empty         Start empty (LAST RESORT)  |");
+            spdlog::error("|                                                                  |");
+            spdlog::error("| Or set in config.json:  \"recovery\": {{ \"mode\": \"<mode>\" }}    |");
+            spdlog::error("|                                                                  |");
+            spdlog::error("| PRESERVE /var/lib/smartbotic-database/ BEFORE ESCALATING.        |");
+            spdlog::error("+------------------------------------------------------------------+");
+            throw std::runtime_error("recovery failed; refusing to start");
+        }
+
+        // Auto-readonly mode on non-trivial recovery
+        if (recovery_outcome_.isNonTrivial() && !force_readwrite_) {
+            std::string reason;
+            switch (recovery_outcome_.kind) {
+                case RecoveryOutcome::Kind::SnapshotFellBack:
+                    reason = "fell back to snapshot " +
+                             recovery_outcome_.snapshotUsed.filename().string() +
+                             " because " +
+                             recovery_outcome_.expectedSnapshot.filename().string() +
+                             " failed: " + recovery_outcome_.failureReason;
+                    break;
+                case RecoveryOutcome::Kind::WalOnlyReplay:
+                    reason = "WAL-only replay, no snapshot loaded (" +
+                             std::to_string(recovery_outcome_.walEntriesReplayed) +
+                             " entries)";
+                    break;
+                case RecoveryOutcome::Kind::ForcedEmpty:
+                    reason = "forced empty by operator (--recovery-mode=force_empty)";
+                    break;
+                default:
+                    reason = "non-trivial recovery";
+                    break;
+            }
+            reason += ". Run `smartbotic-db-cli unlock` to accept this state, "
+                      "or restart with --force-readwrite to bypass this check.";
+
+            setReadOnly(true, reason);
+
+            spdlog::error("");
+            spdlog::error("+------------------------------------------------------------------+");
+            spdlog::error("| [ERROR] Database booted in READ-ONLY mode after non-trivial      |");
+            spdlog::error("|         recovery.                                                 |");
+            spdlog::error("|                                                                  |");
+            spdlog::error("| Reason: {}", reason);
+            spdlog::error("|                                                                  |");
+            spdlog::error("| Writes will be REJECTED until you acknowledge this state:        |");
+            spdlog::error("|   smartbotic-db-cli unlock                 # live, no restart    |");
+            spdlog::error("|   smartbotic-database --force-readwrite    # on next restart     |");
+            spdlog::error("+------------------------------------------------------------------+");
+        }
 
         // Set replication sequence after recovery (WAL sequence is now known)
         replication_->setSequence(persistence_->currentWalSequence());
@@ -317,12 +405,18 @@ void DatabaseService::setupComponents() {
     config_manager_ = std::make_unique<CollectionConfigManager>(*store_);
     store_->setConfigManager(config_manager_.get());
 
-    // Create persistence manager
-    PersistenceManager::Config persistConfig;
+    // Create persistence manager. Start from any persistenceConfig values the
+    // caller (config loader / CLI parser) has already populated — including
+    // recoveryMode — and overlay the top-level convenience fields.
+    PersistenceManager::Config persistConfig = config_.persistenceConfig;
     persistConfig.dataDir = config_.dataDirectory;
     persistConfig.walSyncIntervalMs = config_.walSyncIntervalMs;
     persistConfig.snapshotIntervalSec = config_.snapshotIntervalSec;
     persistConfig.compressionEnabled = config_.compressionEnabled;
+    // Mirror the resolved recoveryMode back into our Config so downstream code
+    // (logging, RPC handlers) can inspect config_.persistenceConfig.recoveryMode
+    // without having to reach into the PersistenceManager.
+    config_.persistenceConfig = persistConfig;
     persistence_ = std::make_unique<PersistenceManager>(persistConfig);
 
     // Connect store callbacks to persistence - uses setPersistCallback for WAL logging

service/src/database_service.hpp

@@ -17,6 +17,7 @@
 #include <atomic>
 #include <filesystem>
 #include <memory>
+#include <mutex>
 #include <string>
 #include <thread>
 
@@ -56,6 +57,11 @@ public:
         uint32_t snapshotIntervalSec = 3600;
         bool compressionEnabled = true;
 
+        // Full persistence manager config (recovery mode, etc.).
+        // DatabaseService::setupComponents() overlays the simpler fields above
+        // onto this struct before constructing PersistenceManager.
+        PersistenceManager::Config persistenceConfig;
+
         // File storage settings
         uint64_t maxFileSizeMb = 500;
         std::vector<std::string> allowedFileTypes;
@@ -125,6 +131,21 @@ public:
      */
     ViewManager& viewManager() { return *view_manager_; }
 
+    /** Current read-only state (atomic, lock-free read). */
+    bool isReadOnly() const { return read_only_.load(std::memory_order_acquire); }
+
+    /** Human-readable reason the DB is read-only (empty when writable). */
+    std::string readOnlyReason() const;
+
+    /** Toggle read-only state. Called by SetReadOnly RPC and startup logic. */
+    void setReadOnly(bool value, const std::string& reason);
+
+    /** Recovery outcome from the last recover() call. Used by GetReadOnlyStatus RPC. */
+    const RecoveryOutcome& recoveryOutcome() const { return recovery_outcome_; }
+
+    /** Set the force-readwrite flag (from --force-readwrite CLI arg). */
+    void setForceReadwrite(bool value) { force_readwrite_ = value; }
+
     /**
      * Load configuration from a JSON file.
      */
@@ -159,6 +180,13 @@ private:
 
     // Migration runner
     std::unique_ptr<MigrationRunner> migrationRunner_;
+
+    // Read-only runtime state (v1.6.1 — auto-readonly on non-trivial recovery)
+    std::atomic<bool> read_only_{false};
+    mutable std::mutex reason_mutex_;
+    std::string read_only_reason_;
+    RecoveryOutcome recovery_outcome_;
+    bool force_readwrite_ = false;
 };
 
 } // namespace smartbotic::database