|
@@ -2,8 +2,82 @@
|
|
|
|
|
|
|
|
#include <spdlog/spdlog.h>
|
|
#include <spdlog/spdlog.h>
|
|
|
|
|
|
|
|
|
|
+#include <stdexcept>
|
|
|
|
|
+
|
|
|
namespace smartbotic::database {
|
|
namespace smartbotic::database {
|
|
|
|
|
|
|
|
|
|
+std::string recoveryModeToString(RecoveryMode mode) {
|
|
|
|
|
+ switch (mode) {
|
|
|
|
|
+ case RecoveryMode::Normal: return "normal";
|
|
|
|
|
+ case RecoveryMode::SnapshotFallback: return "snapshot_fallback";
|
|
|
|
|
+ case RecoveryMode::WalOnly: return "wal_only";
|
|
|
|
|
+ case RecoveryMode::BestEffort: return "best_effort";
|
|
|
|
|
+ case RecoveryMode::ForceEmpty: return "force_empty";
|
|
|
|
|
+ }
|
|
|
|
|
+ return "normal";
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+RecoveryMode recoveryModeFromString(const std::string& s) {
|
|
|
|
|
+ if (s == "normal") return RecoveryMode::Normal;
|
|
|
|
|
+ if (s == "snapshot_fallback") return RecoveryMode::SnapshotFallback;
|
|
|
|
|
+ if (s == "wal_only") return RecoveryMode::WalOnly;
|
|
|
|
|
+ if (s == "best_effort") return RecoveryMode::BestEffort;
|
|
|
|
|
+ if (s == "force_empty") return RecoveryMode::ForceEmpty;
|
|
|
|
|
+ throw std::invalid_argument("invalid recovery mode: '" + s +
|
|
|
|
|
+ "' (valid: normal, snapshot_fallback, wal_only, best_effort, force_empty)");
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+namespace {
|
|
|
|
|
+void applyWalEntry(MemoryStore& store, const WalEntry& entry) {
|
|
|
|
|
+ switch (entry.opType) {
|
|
|
|
|
+ case WalOpType::INSERT:
|
|
|
|
|
+ if (entry.data) {
|
|
|
|
|
+ Document doc = Document::fromJson(*entry.data);
|
|
|
|
|
+ store.loadDocument(entry.collection, doc);
|
|
|
|
|
+ }
|
|
|
|
|
+ break;
|
|
|
|
|
+ case WalOpType::UPDATE:
|
|
|
|
|
+ case WalOpType::UPSERT:
|
|
|
|
|
+ if (entry.data) {
|
|
|
|
|
+ Document doc = Document::fromJson(*entry.data);
|
|
|
|
|
+ store.loadDocumentWithHistory(entry.collection, doc);
|
|
|
|
|
+ }
|
|
|
|
|
+ break;
|
|
|
|
|
+ case WalOpType::DELETE:
|
|
|
|
|
+ store.remove(entry.collection, entry.documentId);
|
|
|
|
|
+ break;
|
|
|
|
|
+ case WalOpType::CREATE_COLLECTION:
|
|
|
|
|
+ if (entry.collectionOptions) {
|
|
|
|
|
+ store.createCollection(entry.collection, *entry.collectionOptions);
|
|
|
|
|
+ }
|
|
|
|
|
+ break;
|
|
|
|
|
+ case WalOpType::DROP_COLLECTION:
|
|
|
|
|
+ store.dropCollection(entry.collection);
|
|
|
|
|
+ break;
|
|
|
|
|
+ case WalOpType::SET_ADD:
|
|
|
|
|
+ if (entry.data && entry.data->contains("member")) {
|
|
|
|
|
+ store.setAdd(entry.collection, entry.documentId,
|
|
|
|
|
+ (*entry.data)["member"].get<std::string>());
|
|
|
|
|
+ }
|
|
|
|
|
+ break;
|
|
|
|
|
+ case WalOpType::SET_REMOVE:
|
|
|
|
|
+ if (entry.data && entry.data->contains("member")) {
|
|
|
|
|
+ store.setRemove(entry.collection, entry.documentId,
|
|
|
|
|
+ (*entry.data)["member"].get<std::string>());
|
|
|
|
|
+ }
|
|
|
|
|
+ break;
|
|
|
|
|
+ case WalOpType::VEC_PUT:
|
|
|
|
|
+ if (entry.vectorData.has_value()) {
|
|
|
|
|
+ store.loadVector(entry.collection, entry.documentId, *entry.vectorData);
|
|
|
|
|
+ }
|
|
|
|
|
+ break;
|
|
|
|
|
+ case WalOpType::VEC_DELETE:
|
|
|
|
|
+ // Handled via DELETE entry
|
|
|
|
|
+ break;
|
|
|
|
|
+ }
|
|
|
|
|
+}
|
|
|
|
|
+} // anonymous namespace
|
|
|
|
|
+
|
|
|
PersistenceManager::PersistenceManager(Config config)
|
|
PersistenceManager::PersistenceManager(Config config)
|
|
|
: config_(std::move(config)) {
|
|
: config_(std::move(config)) {
|
|
|
|
|
|
|
@@ -81,106 +155,79 @@ void PersistenceManager::stop() {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
RecoveryOutcome PersistenceManager::recover(MemoryStore& store) {
|
|
RecoveryOutcome PersistenceManager::recover(MemoryStore& store) {
|
|
|
- spdlog::info("Starting recovery...");
|
|
|
|
|
|
|
+ RecoveryOutcome outcome;
|
|
|
|
|
+ spdlog::info("Starting recovery (mode={}, auto_escalate={})",
|
|
|
|
|
+ recoveryModeToString(config_.recoveryMode), config_.autoEscalate);
|
|
|
|
|
|
|
|
- // Store reference for potential snapshot during recovery
|
|
|
|
|
{
|
|
{
|
|
|
std::lock_guard<std::mutex> lock(storeMutex_);
|
|
std::lock_guard<std::mutex> lock(storeMutex_);
|
|
|
currentStore_ = &store;
|
|
currentStore_ = &store;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+ auto snapshots = snapshotMgr_->listSnapshots();
|
|
|
|
|
+ outcome.snapshotsAvailable = snapshots.size();
|
|
|
|
|
+
|
|
|
|
|
+ // Fresh-install case: no snapshots at all
|
|
|
|
|
+ if (snapshots.empty()) {
|
|
|
|
|
+ if (!config_.allowEmptyOnFreshInstall) {
|
|
|
|
|
+ outcome.kind = RecoveryOutcome::Kind::Failed;
|
|
|
|
|
+ outcome.failureReason = "No snapshots found and allow_empty_on_fresh_install is false";
|
|
|
|
|
+ spdlog::error("Recovery refused: {}", outcome.failureReason);
|
|
|
|
|
+ return outcome;
|
|
|
|
|
+ }
|
|
|
|
|
+ if (!wal_->open()) {
|
|
|
|
|
+ outcome.kind = RecoveryOutcome::Kind::Failed;
|
|
|
|
|
+ outcome.failureReason = "Failed to open WAL for recovery";
|
|
|
|
|
+ spdlog::error("Recovery failed: {}", outcome.failureReason);
|
|
|
|
|
+ return outcome;
|
|
|
|
|
+ }
|
|
|
|
|
+ uint64_t replayed = wal_->replay(0, [&store](const WalEntry& entry) {
|
|
|
|
|
+ applyWalEntry(store, entry);
|
|
|
|
|
+ });
|
|
|
|
|
+ outcome.walEntriesReplayed = replayed;
|
|
|
|
|
+ wal_->close();
|
|
|
|
|
+ outcome.kind = (replayed == 0)
|
|
|
|
|
+ ? RecoveryOutcome::Kind::FreshInstall
|
|
|
|
|
+ : RecoveryOutcome::Kind::WalOnlyReplay;
|
|
|
|
|
+ spdlog::info("Fresh install or WAL-only recovery complete ({} WAL entries)", replayed);
|
|
|
|
|
+ return outcome;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ // TODO Task 5: mode-specific recovery logic
|
|
|
|
|
+ // For now, preserve pre-1.6.1 behavior: load latest snapshot, replay WAL
|
|
|
try {
|
|
try {
|
|
|
- // Load latest snapshot
|
|
|
|
|
- uint64_t snapshotSequence = snapshotMgr_->loadLatestSnapshot(store);
|
|
|
|
|
|
|
+ outcome.expectedSnapshot = snapshots.front();
|
|
|
|
|
+ uint64_t snapshotSequence = snapshotMgr_->loadSnapshot(snapshots.front(), store);
|
|
|
|
|
+ outcome.snapshotUsed = snapshots.front();
|
|
|
|
|
+ outcome.snapshotsAttempted = 1;
|
|
|
spdlog::info("Loaded snapshot at sequence {}", snapshotSequence);
|
|
spdlog::info("Loaded snapshot at sequence {}", snapshotSequence);
|
|
|
|
|
|
|
|
- // Open WAL for replay
|
|
|
|
|
if (!wal_->open()) {
|
|
if (!wal_->open()) {
|
|
|
- spdlog::error("Failed to open WAL for recovery");
|
|
|
|
|
- return RecoveryOutcome{.kind = RecoveryOutcome::Kind::Failed, .failureReason = "Failed to open WAL"};
|
|
|
|
|
|
|
+ outcome.kind = RecoveryOutcome::Kind::Failed;
|
|
|
|
|
+ outcome.failureReason = "Failed to open WAL after snapshot load";
|
|
|
|
|
+ return outcome;
|
|
|
}
|
|
}
|
|
|
-
|
|
|
|
|
- // Replay WAL entries after snapshot
|
|
|
|
|
- // INSERT uses loadDocument (no prior version to save).
|
|
|
|
|
- // UPDATE/UPSERT use loadDocumentWithHistory to preserve the prior
|
|
|
|
|
- // document state in version history during replay.
|
|
|
|
|
- // DELETE calls remove() which internally calls saveToHistory.
|
|
|
|
|
- uint64_t replayedCount = wal_->replay(snapshotSequence, [&store](const WalEntry& entry) {
|
|
|
|
|
- switch (entry.opType) {
|
|
|
|
|
- case WalOpType::INSERT:
|
|
|
|
|
- if (entry.data) {
|
|
|
|
|
- Document doc = Document::fromJson(*entry.data);
|
|
|
|
|
- store.loadDocument(entry.collection, doc);
|
|
|
|
|
- }
|
|
|
|
|
- break;
|
|
|
|
|
-
|
|
|
|
|
- case WalOpType::UPDATE:
|
|
|
|
|
- case WalOpType::UPSERT:
|
|
|
|
|
- if (entry.data) {
|
|
|
|
|
- Document doc = Document::fromJson(*entry.data);
|
|
|
|
|
- store.loadDocumentWithHistory(entry.collection, doc);
|
|
|
|
|
- }
|
|
|
|
|
- break;
|
|
|
|
|
-
|
|
|
|
|
- case WalOpType::DELETE:
|
|
|
|
|
- store.remove(entry.collection, entry.documentId);
|
|
|
|
|
- break;
|
|
|
|
|
-
|
|
|
|
|
- case WalOpType::CREATE_COLLECTION:
|
|
|
|
|
- if (entry.collectionOptions) {
|
|
|
|
|
- store.createCollection(entry.collection, *entry.collectionOptions);
|
|
|
|
|
- }
|
|
|
|
|
- break;
|
|
|
|
|
-
|
|
|
|
|
- case WalOpType::DROP_COLLECTION:
|
|
|
|
|
- store.dropCollection(entry.collection);
|
|
|
|
|
- break;
|
|
|
|
|
-
|
|
|
|
|
- case WalOpType::SET_ADD:
|
|
|
|
|
- if (entry.data && entry.data->contains("member")) {
|
|
|
|
|
- store.setAdd(entry.collection, entry.documentId,
|
|
|
|
|
- (*entry.data)["member"].get<std::string>());
|
|
|
|
|
- }
|
|
|
|
|
- break;
|
|
|
|
|
-
|
|
|
|
|
- case WalOpType::SET_REMOVE:
|
|
|
|
|
- if (entry.data && entry.data->contains("member")) {
|
|
|
|
|
- store.setRemove(entry.collection, entry.documentId,
|
|
|
|
|
- (*entry.data)["member"].get<std::string>());
|
|
|
|
|
- }
|
|
|
|
|
- break;
|
|
|
|
|
-
|
|
|
|
|
- case WalOpType::VEC_PUT:
|
|
|
|
|
- if (entry.vectorData.has_value()) {
|
|
|
|
|
- store.loadVector(entry.collection, entry.documentId, *entry.vectorData);
|
|
|
|
|
- }
|
|
|
|
|
- break;
|
|
|
|
|
-
|
|
|
|
|
- case WalOpType::VEC_DELETE:
|
|
|
|
|
- // Vector deletion is handled when the document is deleted
|
|
|
|
|
- // during normal recovery (DELETE entries). No separate action needed.
|
|
|
|
|
- break;
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ uint64_t replayed = wal_->replay(snapshotSequence, [&store](const WalEntry& entry) {
|
|
|
|
|
+ applyWalEntry(store, entry);
|
|
|
});
|
|
});
|
|
|
|
|
+ outcome.walEntriesReplayed = replayed;
|
|
|
|
|
+ spdlog::info("Replayed {} WAL entries", replayed);
|
|
|
|
|
|
|
|
- spdlog::info("Replayed {} WAL entries", replayedCount);
|
|
|
|
|
-
|
|
|
|
|
- // Update last snapshot sequence
|
|
|
|
|
{
|
|
{
|
|
|
std::lock_guard<std::mutex> lock(statsMutex_);
|
|
std::lock_guard<std::mutex> lock(statsMutex_);
|
|
|
lastSnapshotSequence_ = snapshotSequence;
|
|
lastSnapshotSequence_ = snapshotSequence;
|
|
|
}
|
|
}
|
|
|
|
|
+ wal_->close();
|
|
|
|
|
|
|
|
- wal_->close(); // Close for now, will be reopened by start()
|
|
|
|
|
-
|
|
|
|
|
|
|
+ outcome.kind = RecoveryOutcome::Kind::TrivialSuccess;
|
|
|
spdlog::info("Recovery complete, store has {} documents in {} collections",
|
|
spdlog::info("Recovery complete, store has {} documents in {} collections",
|
|
|
store.getStats().totalDocuments, store.getStats().totalCollections);
|
|
store.getStats().totalDocuments, store.getStats().totalCollections);
|
|
|
-
|
|
|
|
|
- return RecoveryOutcome{}; // Default: TrivialSuccess
|
|
|
|
|
-
|
|
|
|
|
|
|
+ return outcome;
|
|
|
} catch (const std::exception& e) {
|
|
} catch (const std::exception& e) {
|
|
|
|
|
+ outcome.kind = RecoveryOutcome::Kind::Failed;
|
|
|
|
|
+ outcome.failureReason = e.what();
|
|
|
spdlog::error("Recovery failed: {}", e.what());
|
|
spdlog::error("Recovery failed: {}", e.what());
|
|
|
- return RecoveryOutcome{.kind = RecoveryOutcome::Kind::Failed, .failureReason = e.what()};
|
|
|
|
|
|
|
+ return outcome;
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|