|
@@ -133,6 +133,20 @@ bool DatabaseService::initialize() {
|
|
|
// Set replication sequence after recovery (WAL sequence is now known)
|
|
// Set replication sequence after recovery (WAL sequence is now known)
|
|
|
replication_->setSequence(persistence_->currentWalSequence());
|
|
replication_->setSequence(persistence_->currentWalSequence());
|
|
|
|
|
|
|
|
|
|
+ // v1.8.0 — start the persistence manager before loadFromStore /
|
|
|
|
|
+ // migrations so any system-collection writes those phases produce
|
|
|
|
|
+ // (view docs, _collection_meta entries, _migrations recordings)
|
|
|
|
|
+ // reach the WAL like normal mutations. Pre-v1.8, persistence_ was
|
|
|
|
|
+ // started later in start(), which silently dropped those writes
|
|
|
|
|
+ // (running_=false → logInsert no-op). That meant migrated views
|
|
|
|
|
+ // only "survived" because the runner re-created them every boot,
|
|
|
|
|
+ // and a manually-created view that landed during the racy startup
|
|
|
|
|
+ // window (before full readiness) could be lost.
|
|
|
|
|
+ if (!persistence_->start()) {
|
|
|
|
|
+ spdlog::error("Failed to start persistence manager before migrations");
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
// Load view definitions from the _views system collection (which is now
|
|
// Load view definitions from the _views system collection (which is now
|
|
|
// populated by the persistence recovery above).
|
|
// populated by the persistence recovery above).
|
|
|
view_manager_->loadFromStore();
|
|
view_manager_->loadFromStore();
|
|
@@ -221,6 +235,20 @@ void DatabaseService::stop() {
|
|
|
if (events_) {
|
|
if (events_) {
|
|
|
events_->stop();
|
|
events_->stop();
|
|
|
}
|
|
}
|
|
|
|
|
+ if (persistence_ && store_) {
|
|
|
|
|
+ // v1.8.0 — take a final snapshot on graceful shutdown so the next
|
|
|
|
|
+ // boot's recovery is "trivial" (snapshot-loaded) instead of "WAL-only
|
|
|
|
|
+ // replay" which trips auto-readonly mode. This makes ordinary
|
|
|
|
|
+ // restarts pass through cleanly without needing
|
|
|
|
|
+ // `smartbotic-db-cli unlock`.
|
|
|
|
|
+ try {
|
|
|
|
|
+ persistence_->forceSnapshot(*store_);
|
|
|
|
|
+ spdlog::info("Final snapshot taken on shutdown");
|
|
|
|
|
+ } catch (const std::exception& e) {
|
|
|
|
|
+ spdlog::warn("Final snapshot on shutdown failed: {} — recovery on next "
|
|
|
|
|
+ "boot may be WAL-only and trip auto-readonly mode", e.what());
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
if (persistence_) {
|
|
if (persistence_) {
|
|
|
persistence_->stop();
|
|
persistence_->stop();
|
|
|
}
|
|
}
|
|
@@ -475,6 +503,7 @@ void DatabaseService::setupComponents() {
|
|
|
persistConfig.walSyncIntervalMs = config_.walSyncIntervalMs;
|
|
persistConfig.walSyncIntervalMs = config_.walSyncIntervalMs;
|
|
|
persistConfig.snapshotIntervalSec = config_.snapshotIntervalSec;
|
|
persistConfig.snapshotIntervalSec = config_.snapshotIntervalSec;
|
|
|
persistConfig.compressionEnabled = config_.compressionEnabled;
|
|
persistConfig.compressionEnabled = config_.compressionEnabled;
|
|
|
|
|
+ persistConfig.nodeId = config_.nodeId;
|
|
|
// Mirror the resolved recoveryMode back into our Config so downstream code
|
|
// Mirror the resolved recoveryMode back into our Config so downstream code
|
|
|
// (logging, RPC handlers) can inspect config_.persistenceConfig.recoveryMode
|
|
// (logging, RPC handlers) can inspect config_.persistenceConfig.recoveryMode
|
|
|
// without having to reach into the PersistenceManager.
|
|
// without having to reach into the PersistenceManager.
|
|
@@ -484,13 +513,17 @@ void DatabaseService::setupComponents() {
|
|
|
// Connect store callbacks to persistence - uses setPersistCallback for WAL logging
|
|
// Connect store callbacks to persistence - uses setPersistCallback for WAL logging
|
|
|
store_->setPersistCallback([this](const std::string& collection, const std::string& id,
|
|
store_->setPersistCallback([this](const std::string& collection, const std::string& id,
|
|
|
const std::optional<Document>& doc, EventType eventType) {
|
|
const std::optional<Document>& doc, EventType eventType) {
|
|
|
- // Log to WAL based on event type
|
|
|
|
|
|
|
+ // Log to WAL based on event type. v1.7.4: capture the assigned WAL
|
|
|
|
|
+ // sequence and record it on the store so eviction stubs can carry
|
|
|
|
|
+ // a real seq (instead of doc.version, which was useless as a
|
|
|
|
|
+ // `fromSequence` hint and forced a full WAL scan on every fault).
|
|
|
|
|
+ uint64_t walSeq = 0;
|
|
|
switch (eventType) {
|
|
switch (eventType) {
|
|
|
case EventType::INSERT:
|
|
case EventType::INSERT:
|
|
|
- if (doc) persistence_->logInsert(collection, *doc);
|
|
|
|
|
|
|
+ if (doc) walSeq = persistence_->logInsert(collection, *doc);
|
|
|
break;
|
|
break;
|
|
|
case EventType::UPDATE:
|
|
case EventType::UPDATE:
|
|
|
- if (doc) persistence_->logUpdate(collection, *doc);
|
|
|
|
|
|
|
+ if (doc) walSeq = persistence_->logUpdate(collection, *doc);
|
|
|
break;
|
|
break;
|
|
|
case EventType::DELETE:
|
|
case EventType::DELETE:
|
|
|
persistence_->logDelete(collection, id);
|
|
persistence_->logDelete(collection, id);
|
|
@@ -498,12 +531,19 @@ void DatabaseService::setupComponents() {
|
|
|
default:
|
|
default:
|
|
|
break;
|
|
break;
|
|
|
}
|
|
}
|
|
|
|
|
+ if (walSeq > 0 && (eventType == EventType::INSERT || eventType == EventType::UPDATE)) {
|
|
|
|
|
+ store_->recordWalSequence(collection, id, walSeq);
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
// Queue for replication broadcast
|
|
// Queue for replication broadcast
|
|
|
if (replication_ && config_.replicationEnabled) {
|
|
if (replication_ && config_.replicationEnabled) {
|
|
|
databasepb::ReplicationEntry entry;
|
|
databasepb::ReplicationEntry entry;
|
|
|
entry.set_collection(collection);
|
|
entry.set_collection(collection);
|
|
|
entry.set_document_id(id);
|
|
entry.set_document_id(id);
|
|
|
|
|
+ // v1.8.0 — stamp our nodeId so peers can attribute the write to
|
|
|
|
|
+ // its actual origin. Pre-v1.8 broadcasts left this empty, which
|
|
|
|
|
+ // is now treated by receivers as "legacy / unknown origin".
|
|
|
|
|
+ entry.set_node_id(config_.nodeId);
|
|
|
entry.set_global_timestamp(
|
|
entry.set_global_timestamp(
|
|
|
std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
|
std::chrono::system_clock::now().time_since_epoch()
|
|
std::chrono::system_clock::now().time_since_epoch()
|
|
@@ -550,12 +590,20 @@ void DatabaseService::setupComponents() {
|
|
|
}
|
|
}
|
|
|
});
|
|
});
|
|
|
|
|
|
|
|
- // Set up document load callback for LRU eviction recovery
|
|
|
|
|
|
|
+ // Set up document load callback for LRU eviction recovery.
|
|
|
|
|
+ //
|
|
|
|
|
+ // v1.7.4: walSequence is now a real WAL sequence (set by the persist
|
|
|
|
|
+ // callback below at append time, not the per-doc version counter).
|
|
|
|
|
+ // Pass `walSequence - 1` as the exclusive `fromSequence` so replay
|
|
|
|
|
+ // returns entries with seq >= walSequence — the doc's last write is
|
|
|
|
|
+ // exactly at walSequence, so the very first hit is the one we want.
|
|
|
|
|
+ // For docs whose seq is unknown (0 — e.g. loaded from a snapshot
|
|
|
|
|
+ // that predates v1.7.4), fall back to the full-WAL scan from 0.
|
|
|
store_->setDocumentLoadCallback([this](const std::string& collection,
|
|
store_->setDocumentLoadCallback([this](const std::string& collection,
|
|
|
const std::string& id,
|
|
const std::string& id,
|
|
|
uint64_t walSequence) -> std::optional<Document> {
|
|
uint64_t walSequence) -> std::optional<Document> {
|
|
|
- // Load document from WAL via persistence manager
|
|
|
|
|
- return persistence_->loadDocument(collection, id, 0); // Search from beginning
|
|
|
|
|
|
|
+ uint64_t fromSequence = walSequence > 0 ? walSequence - 1 : 0;
|
|
|
|
|
+ return persistence_->loadDocument(collection, id, fromSequence);
|
|
|
});
|
|
});
|
|
|
|
|
|
|
|
// Create encryption manager
|
|
// Create encryption manager
|
|
@@ -605,6 +653,17 @@ void DatabaseService::setupComponents() {
|
|
|
|
|
|
|
|
void DatabaseService::applyReplicatedEntry(const databasepb::ReplicationEntry& entry) {
|
|
void DatabaseService::applyReplicatedEntry(const databasepb::ReplicationEntry& entry) {
|
|
|
try {
|
|
try {
|
|
|
|
|
+ // v1.8.0 — also append the replicated entry to the local WAL, tagged
|
|
|
|
|
+ // with the originating node's id. This is what makes follower-side
|
|
|
|
|
+ // recovery (eviction → WAL fallback, restart → WAL replay) preserve
|
|
|
|
|
+ // replicated documents. Echo amplification is prevented by:
|
|
|
|
|
+ // - GetEntriesSince emitting walEntry.nodeId (not local node id)
|
|
|
|
|
+ // - SyncProtocol skipping entries whose origin is the peer itself
|
|
|
|
|
+ // both implemented in the same v1.8.0 cycle.
|
|
|
|
|
+ const std::string origin = entry.node_id().empty()
|
|
|
|
|
+ ? config_.nodeId // legacy peer (pre-1.8) — best-guess local
|
|
|
|
|
+ : entry.node_id();
|
|
|
|
|
+
|
|
|
switch (entry.op()) {
|
|
switch (entry.op()) {
|
|
|
case databasepb::OP_INSERT:
|
|
case databasepb::OP_INSERT:
|
|
|
case databasepb::OP_UPDATE:
|
|
case databasepb::OP_UPDATE:
|
|
@@ -619,8 +678,20 @@ void DatabaseService::applyReplicatedEntry(const databasepb::ReplicationEntry& e
|
|
|
doc.id = entry.document_id();
|
|
doc.id = entry.document_id();
|
|
|
doc.nodeId = entry.node_id();
|
|
doc.nodeId = entry.node_id();
|
|
|
|
|
|
|
|
- // Use loadDocument to bypass normal callbacks (avoid re-replication)
|
|
|
|
|
|
|
+ // loadDocument bypasses the persist-and-broadcast callback to avoid
|
|
|
|
|
+ // re-replicating; we explicitly write to the WAL right after with
|
|
|
|
|
+ // the origin-aware overload so eviction/WAL recovery still works.
|
|
|
store_->loadDocument(entry.collection(), doc);
|
|
store_->loadDocument(entry.collection(), doc);
|
|
|
|
|
+
|
|
|
|
|
+ if (persistence_) {
|
|
|
|
|
+ uint64_t walSeq = (entry.op() == databasepb::OP_INSERT)
|
|
|
|
|
+ ? persistence_->logInsert(entry.collection(), doc, origin)
|
|
|
|
|
+ : persistence_->logUpdate(entry.collection(), doc, origin);
|
|
|
|
|
+ if (walSeq > 0) {
|
|
|
|
|
+ store_->recordWalSequence(entry.collection(), doc.id, walSeq);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
spdlog::trace("Applied replicated {} to {}/{} from {}",
|
|
spdlog::trace("Applied replicated {} to {}/{} from {}",
|
|
|
entry.op() == databasepb::OP_INSERT ? "insert" :
|
|
entry.op() == databasepb::OP_INSERT ? "insert" :
|
|
|
entry.op() == databasepb::OP_UPDATE ? "update" : "upsert",
|
|
entry.op() == databasepb::OP_UPDATE ? "update" : "upsert",
|
|
@@ -630,6 +701,9 @@ void DatabaseService::applyReplicatedEntry(const databasepb::ReplicationEntry& e
|
|
|
|
|
|
|
|
case databasepb::OP_DELETE: {
|
|
case databasepb::OP_DELETE: {
|
|
|
store_->remove(entry.collection(), entry.document_id());
|
|
store_->remove(entry.collection(), entry.document_id());
|
|
|
|
|
+ if (persistence_) {
|
|
|
|
|
+ persistence_->logDelete(entry.collection(), entry.document_id(), origin);
|
|
|
|
|
+ }
|
|
|
spdlog::trace("Applied replicated delete to {}/{} from {}",
|
|
spdlog::trace("Applied replicated delete to {}/{} from {}",
|
|
|
entry.collection(), entry.document_id(), entry.node_id());
|
|
entry.collection(), entry.document_id(), entry.node_id());
|
|
|
break;
|
|
break;
|
|
@@ -638,6 +712,9 @@ void DatabaseService::applyReplicatedEntry(const databasepb::ReplicationEntry& e
|
|
|
case databasepb::OP_CREATE_COLLECTION: {
|
|
case databasepb::OP_CREATE_COLLECTION: {
|
|
|
CollectionOptions options;
|
|
CollectionOptions options;
|
|
|
store_->createCollection(entry.collection(), options);
|
|
store_->createCollection(entry.collection(), options);
|
|
|
|
|
+ if (persistence_) {
|
|
|
|
|
+ persistence_->logCreateCollection(entry.collection(), options, origin);
|
|
|
|
|
+ }
|
|
|
spdlog::trace("Applied replicated create collection {} from {}",
|
|
spdlog::trace("Applied replicated create collection {} from {}",
|
|
|
entry.collection(), entry.node_id());
|
|
entry.collection(), entry.node_id());
|
|
|
break;
|
|
break;
|
|
@@ -645,6 +722,9 @@ void DatabaseService::applyReplicatedEntry(const databasepb::ReplicationEntry& e
|
|
|
|
|
|
|
|
case databasepb::OP_DROP_COLLECTION: {
|
|
case databasepb::OP_DROP_COLLECTION: {
|
|
|
store_->dropCollection(entry.collection());
|
|
store_->dropCollection(entry.collection());
|
|
|
|
|
+ if (persistence_) {
|
|
|
|
|
+ persistence_->logDropCollection(entry.collection(), origin);
|
|
|
|
|
+ }
|
|
|
spdlog::trace("Applied replicated drop collection {} from {}",
|
|
spdlog::trace("Applied replicated drop collection {} from {}",
|
|
|
entry.collection(), entry.node_id());
|
|
entry.collection(), entry.node_id());
|
|
|
break;
|
|
break;
|