Explorar el Código

feat(storage): move dual-write mirror under MemoryStore lock

Closes the race between MemoryStore visibility and LMDB visibility for
ALL concurrent readers (not just the writer's own ACK chain). The
applyDualWriteMirror call moves from the persist callback into
MemoryStore's private mirrorWriteToDocStore helper, invoked BEFORE
lock.unlock() at every write site (insert, update, upsert, remove,
patch, set add/remove, restore, expire).

MemoryStore now owns the lifecycle: setDocumentStoreMirror(ds, healthy,
drift) attaches the LMDB store + atomic flags; the persist callback in
database_service.cpp keeps WAL + replication + events but no longer
touches LMDB.

EXPIRE writes mirror as DELETE — the substrate has no expire concept,
but a TTL'd doc removed from MemoryStore must be removed from LMDB too
or the next Get from LMDB would return a ghost. WAL-side semantics
unchanged (EXPIRE still EXPIRE in the callback path).

Stress-validated against load_test_mixed (4 writers / 8 readers / 30s):
2213 writes/s, 5317 reads/s, 0 hard failures, 0 mirror failures, 0 LMDB
query failures. Throughput dropped ~6-8% vs out-of-lock mirror — the
LMDB commit now serializes per-collection. Acceptable for the race
elimination.

Full test suite (12/12) stays green.
fszontagh hace 2 meses
padre
commit
85e0158572
Se han modificado 3 ficheros con 109 adiciones y 4 borrados
  1. 14 4
      service/src/database_service.cpp
  2. 56 0
      service/src/memory_store.cpp
  3. 39 0
      service/src/memory_store.hpp

+ 14 - 4
service/src/database_service.cpp

@@ -660,6 +660,16 @@ void DatabaseService::setupComponents() {
     history_store_ = std::make_unique<HistoryStore>(histConfig);
     history_store_ = std::make_unique<HistoryStore>(histConfig);
     store_->setHistoryStore(history_store_.get());
     store_->setHistoryStore(history_store_.get());
 
 
+    // v2.0 Stage 4 — wire the LMDB mirror INTO MemoryStore. After this,
+    // every write that flows through store_ also commits to doc_store_
+    // BEFORE the per-collection lock releases. The persist callback no
+    // longer has its own mirror block; it does only WAL/replication/events.
+    if (doc_store_) {
+        store_->setDocumentStoreMirror(doc_store_.get(),
+                                       &mirror_healthy_,
+                                       &mirror_drift_count_);
+    }
+
     // Create persistence manager. Start from any persistenceConfig values the
     // Create persistence manager. Start from any persistenceConfig values the
     // caller (config loader / CLI parser) has already populated — including
     // caller (config loader / CLI parser) has already populated — including
     // recoveryMode — and overlay the top-level convenience fields.
     // recoveryMode — and overlay the top-level convenience fields.
@@ -700,10 +710,10 @@ void DatabaseService::setupComponents() {
             store_->recordWalSequence(collection, id, walSeq);
             store_->recordWalSequence(collection, id, walSeq);
         }
         }
 
 
-        // v2.0 dual-write mirror — see storage/dual_write_mirror.hpp.
-        smartbotic::db::storage::applyDualWriteMirror(
-            doc_store_.get(), mirror_healthy_, mirror_drift_count_,
-            collection, id, doc, eventType);
+        // v2.0 dual-write — moved INTO MemoryStore::mirrorWriteToDocStore so
+        // that it runs UNDER the per-collection write lock. That closes the
+        // race window where readers could see a doc in MemoryStore but not
+        // yet in LMDB. The callback now does only WAL + replication + events.
 
 
         // Queue for replication broadcast
         // Queue for replication broadcast
         if (replication_ && config_.replicationEnabled) {
         if (replication_ && config_.replicationEnabled) {

+ 56 - 0
service/src/memory_store.cpp

@@ -2,6 +2,8 @@
 
 
 #include "config/collection_config_manager.hpp"
 #include "config/collection_config_manager.hpp"
 #include "persistence/history_store.hpp"
 #include "persistence/history_store.hpp"
+#include "storage/document_store.hpp"
+#include "storage/dual_write_mirror.hpp"
 #include "storage/filter_eval.hpp"
 #include "storage/filter_eval.hpp"
 
 
 #if defined(__GLIBC__) && !defined(__APPLE__)
 #if defined(__GLIBC__) && !defined(__APPLE__)
@@ -291,6 +293,9 @@ std::string MemoryStore::insert(const std::string& collection, Document doc) {
     // Track memory increment
     // Track memory increment
     uint64_t docSize = estimateDocumentSize(doc);
     uint64_t docSize = estimateDocumentSize(doc);
 
 
+    // v2.0 dual-write under lock — see header comment on mirrorWriteToDocStore.
+    mirrorWriteToDocStore(collection, docId, doc, EventType::INSERT);
+
     lock.unlock();
     lock.unlock();
 
 
     // Update stats
     // Update stats
@@ -448,6 +453,9 @@ bool MemoryStore::update(const std::string& collection, const std::string& id, c
     // Track memory change (new size)
     // Track memory change (new size)
     uint64_t newSize = estimateDocumentSize(updated);
     uint64_t newSize = estimateDocumentSize(updated);
 
 
+    // v2.0 dual-write under lock
+    mirrorWriteToDocStore(collection, id, updated, EventType::UPDATE);
+
     lock.unlock();
     lock.unlock();
 
 
     // Update stats
     // Update stats
@@ -558,6 +566,10 @@ std::string MemoryStore::upsert(const std::string& collection, Document doc) {
     // Track new document size
     // Track new document size
     uint64_t newSize = estimateDocumentSize(doc);
     uint64_t newSize = estimateDocumentSize(doc);
 
 
+    // v2.0 dual-write under lock (upsert path — INSERT or UPDATE depending on isInsert)
+    mirrorWriteToDocStore(collection, docId, doc,
+                          isInsert ? EventType::INSERT : EventType::UPDATE);
+
     lock.unlock();
     lock.unlock();
 
 
     // Update stats
     // Update stats
@@ -616,6 +628,9 @@ bool MemoryStore::remove(const std::string& collection, const std::string& id) {
     removeVector(*coll, id);
     removeVector(*coll, id);
     coll->updatedAt = currentTimeMs();
     coll->updatedAt = currentTimeMs();
 
 
+    // v2.0 dual-write under lock (delete path)
+    mirrorWriteToDocStore(collection, id, std::nullopt, EventType::DELETE);
+
     lock.unlock();
     lock.unlock();
 
 
     // Update stats
     // Update stats
@@ -699,6 +714,9 @@ bool MemoryStore::updateIfVersion(const std::string& collection, const std::stri
     // Track memory change (new size)
     // Track memory change (new size)
     uint64_t newSize = estimateDocumentSize(updated);
     uint64_t newSize = estimateDocumentSize(updated);
 
 
+    // v2.0 dual-write under lock
+    mirrorWriteToDocStore(collection, id, updated, EventType::UPDATE);
+
     lock.unlock();
     lock.unlock();
 
 
     // Update stats
     // Update stats
@@ -782,6 +800,9 @@ uint64_t MemoryStore::patchDocument(const std::string& collection, const std::st
     // Capture the updated document for callbacks after releasing the lock
     // Capture the updated document for callbacks after releasing the lock
     Document updatedDoc = it->second;
     Document updatedDoc = it->second;
 
 
+    // v2.0 dual-write under lock
+    mirrorWriteToDocStore(collection, id, updatedDoc, EventType::UPDATE);
+
     lock.unlock();
     lock.unlock();
 
 
     // Update stats
     // Update stats
@@ -1041,6 +1062,9 @@ bool MemoryStore::setAdd(const std::string& collection, const std::string& setId
         coll->documents[setId] = doc;
         coll->documents[setId] = doc;
         coll->updatedAt = doc.updatedAt;
         coll->updatedAt = doc.updatedAt;
 
 
+        // v2.0 dual-write under lock
+        mirrorWriteToDocStore(collection, setId, doc, EventType::INSERT);
+
         lock.unlock();
         lock.unlock();
 
 
         {
         {
@@ -1080,6 +1104,9 @@ bool MemoryStore::setAdd(const std::string& collection, const std::string& setId
 
 
     Document updated = it->second;
     Document updated = it->second;
 
 
+    // v2.0 dual-write under lock (sadd-style: append member, update doc)
+    mirrorWriteToDocStore(collection, setId, updated, EventType::UPDATE);
+
     lock.unlock();
     lock.unlock();
 
 
     {
     {
@@ -1133,6 +1160,9 @@ bool MemoryStore::setRemove(const std::string& collection, const std::string& se
 
 
     Document updated = it->second;
     Document updated = it->second;
 
 
+    // v2.0 dual-write under lock (srem-style: trim members, update doc)
+    mirrorWriteToDocStore(collection, setId, updated, EventType::UPDATE);
+
     lock.unlock();
     lock.unlock();
 
 
     {
     {
@@ -1509,6 +1539,11 @@ uint64_t MemoryStore::expireDocuments() {
                 coll->documents.erase(docIt);
                 coll->documents.erase(docIt);
                 expired++;
                 expired++;
 
 
+                // v2.0 dual-write under lock — expire is semantically a delete
+                // for the substrate. Mirror it as DELETE so LMDB doesn't keep
+                // an orphan doc whose MemoryStore version has expired.
+                mirrorWriteToDocStore(collName, id, std::nullopt, EventType::DELETE);
+
                 // Emit callbacks (unlock first to avoid deadlock)
                 // Emit callbacks (unlock first to avoid deadlock)
                 collLock.unlock();
                 collLock.unlock();
 
 
@@ -1816,6 +1851,11 @@ uint64_t MemoryStore::restoreToVersion(const std::string& collection, const std:
 
 
     coll->updatedAt = restoredDoc.updatedAt;
     coll->updatedAt = restoredDoc.updatedAt;
 
 
+    // v2.0 dual-write under lock (restore path — INSERT if doc was previously
+    // deleted/missing, UPDATE if we replaced an existing doc).
+    mirrorWriteToDocStore(collection, id, restoredDoc,
+                          wasDeleted ? EventType::INSERT : EventType::UPDATE);
+
     lock.unlock();
     lock.unlock();
 
 
     // Update stats
     // Update stats
@@ -2200,6 +2240,22 @@ void MemoryStore::emitPersist(const std::string& collection, const std::string&
     }
     }
 }
 }
 
 
+void MemoryStore::setDocumentStoreMirror(smartbotic::db::storage::DocumentStore* ds,
+                                         std::atomic<bool>* healthy,
+                                         std::atomic<uint64_t>* drift) {
+    docStoreMirror_ = ds;
+    mirrorHealthy_ = healthy;
+    mirrorDriftCount_ = drift;
+}
+
+void MemoryStore::mirrorWriteToDocStore(const std::string& collection, const std::string& id,
+                                         const std::optional<Document>& doc, EventType eventType) {
+    if (!docStoreMirror_ || !mirrorHealthy_ || !mirrorDriftCount_) return;
+    smartbotic::db::storage::applyDualWriteMirror(
+        docStoreMirror_, *mirrorHealthy_, *mirrorDriftCount_,
+        collection, id, doc, eventType);
+}
+
 void MemoryStore::addToExpirationIndex(CollectionData& coll, const std::string& id, uint64_t expiresAt) {
 void MemoryStore::addToExpirationIndex(CollectionData& coll, const std::string& id, uint64_t expiresAt) {
     coll.expirationIndex[expiresAt].insert(id);
     coll.expirationIndex[expiresAt].insert(id);
 }
 }

+ 39 - 0
service/src/memory_store.hpp

@@ -16,6 +16,13 @@
 #include <unordered_set>
 #include <unordered_set>
 #include <vector>
 #include <vector>
 
 
+// v2.0 fwd decl — DocumentStore lives in smartbotic::db::storage (not
+// smartbotic::database). The mirror setter takes a pointer; full impl
+// is included only in memory_store.cpp where the mirror call site is.
+namespace smartbotic::db::storage {
+class DocumentStore;
+}
+
 namespace smartbotic::database {
 namespace smartbotic::database {
 
 
 class HistoryStore;  // v1.9.0 — fwd decl; disk-resident version history
 class HistoryStore;  // v1.9.0 — fwd decl; disk-resident version history
@@ -148,6 +155,23 @@ public:
      */
      */
     void setHistoryStore(HistoryStore* hs) { historyStore_ = hs; }
     void setHistoryStore(HistoryStore* hs) { historyStore_ = hs; }
 
 
+    /**
+     * v2.0 Stage 4 — attach the LMDB document-store mirror. Once set, every
+     * write path (insert / update / patch / remove / set / restore /
+     * expire) calls `applyDualWriteMirror` on the target store BEFORE
+     * releasing the per-collection lock. That closes the race where
+     * concurrent readers could observe a doc in MemoryStore but not yet in
+     * LMDB. Safe to leave unset (tests, pre-v2.0 paths) — the mirror call
+     * becomes a no-op.
+     *
+     * `healthy` flips to false on mirror failure (and ERROR is logged);
+     * `drift` accumulates failure count for observability. Both atomics are
+     * owned by the caller (DatabaseService) and must outlive this store.
+     */
+    void setDocumentStoreMirror(smartbotic::db::storage::DocumentStore* ds,
+                                std::atomic<bool>* healthy,
+                                std::atomic<uint64_t>* drift);
+
     // ===== Collection Management =====
     // ===== Collection Management =====
 
 
     /**
     /**
@@ -704,6 +728,14 @@ private:
     void emitPersist(const std::string& collection, const std::string& id,
     void emitPersist(const std::string& collection, const std::string& id,
                      const std::optional<Document>& doc, EventType eventType);
                      const std::optional<Document>& doc, EventType eventType);
 
 
+    // v2.0 Stage 4 — mirror a write into the LMDB DocumentStore. Called
+    // from within the per-collection write lock, BEFORE lock release, so
+    // any subsequent LMDB read by any thread observes the same state as
+    // a MemoryStore read. No-op when setDocumentStoreMirror() hasn't been
+    // called (legacy / test bootstrap path).
+    void mirrorWriteToDocStore(const std::string& collection, const std::string& id,
+                                const std::optional<Document>& doc, EventType eventType);
+
     // Update expiration index
     // Update expiration index
     void addToExpirationIndex(CollectionData& coll, const std::string& id, uint64_t expiresAt);
     void addToExpirationIndex(CollectionData& coll, const std::string& id, uint64_t expiresAt);
     void removeFromExpirationIndex(CollectionData& coll, const std::string& id, uint64_t expiresAt);
     void removeFromExpirationIndex(CollectionData& coll, const std::string& id, uint64_t expiresAt);
@@ -790,6 +822,13 @@ private:
     // history operations no-op (no history kept).
     // history operations no-op (no history kept).
     HistoryStore* historyStore_ = nullptr;
     HistoryStore* historyStore_ = nullptr;
 
 
+    // v2.0 Stage 4 — dual-write target. All three pointers set or all
+    // three null; callers own the lifetimes. When null, mirrorWriteToDocStore
+    // is a no-op (preserves test/bootstrap paths that don't wire LMDB).
+    smartbotic::db::storage::DocumentStore* docStoreMirror_ = nullptr;
+    std::atomic<bool>* mirrorHealthy_ = nullptr;
+    std::atomic<uint64_t>* mirrorDriftCount_ = nullptr;
+
     // Per-document last-write WAL sequence map. Populated by the persist
     // Per-document last-write WAL sequence map. Populated by the persist
     // callback after each WAL append and by recovery's WAL replay. Used
     // callback after each WAL append and by recovery's WAL replay. Used
     // by `evictDocument` to populate `EvictedDocumentStub.walSequence`
     // by `evictDocument` to populate `EvictedDocumentStub.walSequence`