|
|
@@ -0,0 +1,290 @@
|
|
|
+// v2.0 Stage 4 — dual-write mirror tests.
|
|
|
+//
|
|
|
+// Covers `applyDualWriteMirror` directly (unit) and the integration path
|
|
|
+// where it's installed as a MemoryStore::setPersistCallback (matching the
|
|
|
+// shape DatabaseService uses). Failure path uses a throwing DocumentStore
|
|
|
+// stub to verify health flag + drift counter behavior without needing to
|
|
|
+// induce a real LMDB error.
|
|
|
+
|
|
|
+#include <algorithm>
|
|
|
+#include <atomic>
|
|
|
+#include <cassert>
|
|
|
+#include <cstdlib>
|
|
|
+#include <filesystem>
|
|
|
+#include <iostream>
|
|
|
+#include <memory>
|
|
|
+#include <optional>
|
|
|
+#include <stdexcept>
|
|
|
+#include <string>
|
|
|
+#include <unistd.h>
|
|
|
+
|
|
|
+#include <nlohmann/json.hpp>
|
|
|
+
|
|
|
+#include "document.hpp"
|
|
|
+#include "memory_store.hpp"
|
|
|
+#include "storage/document_store.hpp"
|
|
|
+#include "storage/document_store_lmdb.hpp"
|
|
|
+#include "storage/dual_write_mirror.hpp"
|
|
|
+#include "storage/lmdb_env.hpp"
|
|
|
+
|
|
|
+namespace fs = std::filesystem;
|
|
|
+
|
|
|
+using smartbotic::database::Document;
|
|
|
+using smartbotic::database::EventType;
|
|
|
+using smartbotic::database::MemoryStore;
|
|
|
+using smartbotic::db::storage::applyDualWriteMirror;
|
|
|
+using smartbotic::db::storage::DocumentStore;
|
|
|
+using smartbotic::db::storage::LmdbDocumentStore;
|
|
|
+using smartbotic::db::storage::LmdbEnv;
|
|
|
+using smartbotic::db::storage::LmdbEnvOpts;
|
|
|
+using smartbotic::db::storage::ScanResult;
|
|
|
+
|
|
|
+namespace {
|
|
|
+
|
|
|
+int g_pass = 0;
|
|
|
+int g_fail = 0;
|
|
|
+
|
|
|
+void check(bool cond, const char* msg) {
|
|
|
+ if (cond) {
|
|
|
+ ++g_pass;
|
|
|
+ } else {
|
|
|
+ ++g_fail;
|
|
|
+ std::cerr << "FAIL: " << msg << "\n";
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+std::string make_tmpdir(const char* tag) {
|
|
|
+ static std::atomic<int> counter{0};
|
|
|
+ int n = counter.fetch_add(1);
|
|
|
+ std::string path = "/tmp/dual-write-mirror-test-" + std::to_string(::getpid()) +
|
|
|
+ "-" + std::to_string(n) + "-" + tag;
|
|
|
+ std::error_code ec;
|
|
|
+ fs::remove_all(path, ec);
|
|
|
+ return path;
|
|
|
+}
|
|
|
+
|
|
|
+struct TmpEnv {
|
|
|
+ std::string path;
|
|
|
+ LmdbEnv env;
|
|
|
+ explicit TmpEnv(const char* tag)
|
|
|
+ : path(make_tmpdir(tag)),
|
|
|
+ env(LmdbEnvOpts{path, 64ULL << 20, 256, 126, false}) {}
|
|
|
+ ~TmpEnv() {
|
|
|
+ std::error_code ec;
|
|
|
+ fs::remove_all(path, ec);
|
|
|
+ }
|
|
|
+ TmpEnv(const TmpEnv&) = delete;
|
|
|
+ TmpEnv& operator=(const TmpEnv&) = delete;
|
|
|
+};
|
|
|
+
|
|
|
+Document make_doc(const std::string& id,
|
|
|
+ const std::string& collection,
|
|
|
+ const nlohmann::json& data) {
|
|
|
+ Document d;
|
|
|
+ d.id = id;
|
|
|
+ d.collection = collection;
|
|
|
+ d.set_data(data);
|
|
|
+ d.createdAt = 1731628800000ULL;
|
|
|
+ d.updatedAt = 1731628800000ULL;
|
|
|
+ d.version = 1;
|
|
|
+ d.lastAccessedAt = 1731628800000ULL;
|
|
|
+ return d;
|
|
|
+}
|
|
|
+
|
|
|
+// DocumentStore stub whose put/del always throw — exercises the failure
|
|
|
+// path without forcing a real LMDB error.
|
|
|
+class ThrowingStore : public DocumentStore {
|
|
|
+public:
|
|
|
+ void put(std::string_view, std::string_view, const Document&) override {
|
|
|
+ throw std::runtime_error("simulated mirror failure");
|
|
|
+ }
|
|
|
+ std::optional<Document> get(std::string_view, std::string_view) override {
|
|
|
+ return std::nullopt;
|
|
|
+ }
|
|
|
+ bool del(std::string_view, std::string_view) override {
|
|
|
+ throw std::runtime_error("simulated mirror failure");
|
|
|
+ }
|
|
|
+ uint64_t count(std::string_view) override { return 0; }
|
|
|
+ ScanResult scan(std::string_view, const smartbotic::database::Query&) override {
|
|
|
+ return {};
|
|
|
+ }
|
|
|
+ std::vector<std::string> list_collections() override { return {}; }
|
|
|
+ bool drop_collection(std::string_view) override { return false; }
|
|
|
+};
|
|
|
+
|
|
|
+// ---------- Unit-level tests on applyDualWriteMirror ----------
|
|
|
+
|
|
|
+void test_unit_null_store_is_noop() {
|
|
|
+ std::atomic<bool> healthy{true};
|
|
|
+ std::atomic<uint64_t> drift{0};
|
|
|
+ Document d = make_doc("u1", "users", {{"name", "alice"}});
|
|
|
+ applyDualWriteMirror(nullptr, healthy, drift, "users", "u1", d, EventType::INSERT);
|
|
|
+ check(healthy.load(), "null doc_store: stays healthy");
|
|
|
+ check(drift.load() == 0, "null doc_store: drift unchanged");
|
|
|
+}
|
|
|
+
|
|
|
+void test_unit_system_collection_skipped() {
|
|
|
+ TmpEnv tmp("system-skip");
|
|
|
+ LmdbDocumentStore store(tmp.env);
|
|
|
+ std::atomic<bool> healthy{true};
|
|
|
+ std::atomic<uint64_t> drift{0};
|
|
|
+ Document d = make_doc("v1", "_views", {{"name", "v1"}});
|
|
|
+ applyDualWriteMirror(&store, healthy, drift, "_views", "v1", d, EventType::INSERT);
|
|
|
+
|
|
|
+ auto names = store.list_collections();
|
|
|
+ check(std::find(names.begin(), names.end(), "_views") == names.end(),
|
|
|
+ "system collection: not created in LMDB");
|
|
|
+ check(healthy.load(), "system collection: stays healthy");
|
|
|
+ check(drift.load() == 0, "system collection: drift unchanged");
|
|
|
+}
|
|
|
+
|
|
|
+void test_unit_insert_update_delete_round_trip() {
|
|
|
+ TmpEnv tmp("round-trip");
|
|
|
+ LmdbDocumentStore store(tmp.env);
|
|
|
+ std::atomic<bool> healthy{true};
|
|
|
+ std::atomic<uint64_t> drift{0};
|
|
|
+
|
|
|
+ Document d1 = make_doc("u1", "users", {{"name", "alice"}});
|
|
|
+ applyDualWriteMirror(&store, healthy, drift, "users", "u1", d1, EventType::INSERT);
|
|
|
+ auto got = store.get("users", "u1");
|
|
|
+ check(got.has_value(), "insert: doc landed in LMDB");
|
|
|
+ check(got->data().value("name", "") == "alice", "insert: data preserved");
|
|
|
+
|
|
|
+ Document d1b = make_doc("u1", "users", {{"name", "alice-updated"}});
|
|
|
+ applyDualWriteMirror(&store, healthy, drift, "users", "u1", d1b, EventType::UPDATE);
|
|
|
+ got = store.get("users", "u1");
|
|
|
+ check(got.has_value() && got->data().value("name", "") == "alice-updated",
|
|
|
+ "update: LMDB sees new value");
|
|
|
+
|
|
|
+ applyDualWriteMirror(&store, healthy, drift, "users", "u1", std::nullopt, EventType::DELETE);
|
|
|
+ check(!store.get("users", "u1").has_value(), "delete: LMDB doc gone");
|
|
|
+
|
|
|
+ check(healthy.load(), "round-trip: stays healthy");
|
|
|
+ check(drift.load() == 0, "round-trip: zero drift");
|
|
|
+}
|
|
|
+
|
|
|
+void test_unit_failure_marks_degraded() {
|
|
|
+ ThrowingStore store;
|
|
|
+ std::atomic<bool> healthy{true};
|
|
|
+ std::atomic<uint64_t> drift{0};
|
|
|
+
|
|
|
+ Document d = make_doc("u1", "users", {{"name", "alice"}});
|
|
|
+ applyDualWriteMirror(&store, healthy, drift, "users", "u1", d, EventType::INSERT);
|
|
|
+ check(!healthy.load(), "failure: mirror_healthy_ flipped to false");
|
|
|
+ check(drift.load() == 1, "failure: drift counter == 1");
|
|
|
+
|
|
|
+ applyDualWriteMirror(&store, healthy, drift, "users", "u1", std::nullopt, EventType::DELETE);
|
|
|
+ check(!healthy.load(), "failure(delete): stays degraded");
|
|
|
+ check(drift.load() == 2, "failure(delete): drift counter == 2");
|
|
|
+}
|
|
|
+
|
|
|
+void test_unit_insert_without_doc_is_noop() {
|
|
|
+ TmpEnv tmp("insert-no-doc");
|
|
|
+ LmdbDocumentStore store(tmp.env);
|
|
|
+ std::atomic<bool> healthy{true};
|
|
|
+ std::atomic<uint64_t> drift{0};
|
|
|
+
|
|
|
+ applyDualWriteMirror(&store, healthy, drift, "users", "u1", std::nullopt, EventType::INSERT);
|
|
|
+ check(!store.get("users", "u1").has_value(), "INSERT with nullopt doc: no LMDB write");
|
|
|
+ check(healthy.load() && drift.load() == 0, "INSERT with nullopt doc: clean");
|
|
|
+}
|
|
|
+
|
|
|
+// ---------- Integration via MemoryStore callback ----------
|
|
|
+//
|
|
|
+// Mirrors the DatabaseService wiring: install a callback on MemoryStore
|
|
|
+// that calls applyDualWriteMirror with the live doc_store_ + health flags,
|
|
|
+// then drive insert/update/delete through MemoryStore's public API and
|
|
|
+// verify LMDB sees each op.
|
|
|
+
|
|
|
+void test_integration_via_memory_store_callback() {
|
|
|
+ TmpEnv tmp("integration");
|
|
|
+ LmdbDocumentStore doc_store(tmp.env);
|
|
|
+ std::atomic<bool> healthy{true};
|
|
|
+ std::atomic<uint64_t> drift{0};
|
|
|
+
|
|
|
+ MemoryStore::Config cfg;
|
|
|
+ cfg.nodeId = "test-node";
|
|
|
+ cfg.maxMemoryBytes = 64ULL * 1024 * 1024;
|
|
|
+ MemoryStore mem(cfg);
|
|
|
+ mem.setPersistCallback([&](const std::string& coll, const std::string& id,
|
|
|
+ const std::optional<Document>& d, EventType ev) {
|
|
|
+ applyDualWriteMirror(&doc_store, healthy, drift, coll, id, d, ev);
|
|
|
+ });
|
|
|
+
|
|
|
+ Document d1 = make_doc("u1", "users", {{"name", "alice"}});
|
|
|
+ std::string inserted_id = mem.insert("users", d1);
|
|
|
+ check(inserted_id == "u1", "memstore insert: returns id");
|
|
|
+ check(doc_store.get("users", "u1").has_value(), "mem→lmdb: insert mirrored");
|
|
|
+
|
|
|
+ Document d1b = make_doc("u1", "users", {{"name", "alice-2"}});
|
|
|
+ d1b.version = 1;
|
|
|
+ bool updated = mem.update("users", "u1", d1b);
|
|
|
+ check(updated, "memstore update: success");
|
|
|
+ auto got = doc_store.get("users", "u1");
|
|
|
+ check(got.has_value() && got->data().value("name", "") == "alice-2",
|
|
|
+ "mem→lmdb: update mirrored");
|
|
|
+
|
|
|
+ bool removed = mem.remove("users", "u1");
|
|
|
+ check(removed, "memstore remove: success");
|
|
|
+ check(!doc_store.get("users", "u1").has_value(), "mem→lmdb: delete mirrored");
|
|
|
+
|
|
|
+ check(healthy.load() && drift.load() == 0, "integration: clean run");
|
|
|
+}
|
|
|
+
|
|
|
+// Simulates DatabaseService::backfillIntoDocStore — populate MemoryStore
|
|
|
+// without a persist callback (no live mirror), then walk listCollections +
|
|
|
+// getAllDocuments and feed each doc into applyDualWriteMirror with
|
|
|
+// EventType::INSERT. Verifies LMDB ends up with every user doc and no
|
|
|
+// system-collection docs.
|
|
|
+void test_backfill_iteration_pattern() {
|
|
|
+ TmpEnv tmp("backfill");
|
|
|
+ LmdbDocumentStore doc_store(tmp.env);
|
|
|
+ std::atomic<bool> healthy{true};
|
|
|
+ std::atomic<uint64_t> drift{0};
|
|
|
+
|
|
|
+ MemoryStore::Config cfg;
|
|
|
+ cfg.nodeId = "test-node";
|
|
|
+ cfg.maxMemoryBytes = 64ULL * 1024 * 1024;
|
|
|
+ MemoryStore mem(cfg);
|
|
|
+ // No persist callback — simulating "data loaded from snapshot/WAL".
|
|
|
+
|
|
|
+ mem.insert("users", make_doc("u1", "users", {{"name", "alice"}}));
|
|
|
+ mem.insert("users", make_doc("u2", "users", {{"name", "bob"}}));
|
|
|
+ mem.insert("orders", make_doc("o1", "orders", {{"total", 42}}));
|
|
|
+ // System collection write — should be excluded by the filter inside the
|
|
|
+ // helper, even when fed through the same iteration loop.
|
|
|
+ mem.insert("_views", make_doc("v1", "_views", {{"name", "v1"}}));
|
|
|
+
|
|
|
+ // Backfill loop — same shape as DatabaseService::backfillIntoDocStore.
|
|
|
+ for (const auto& coll : mem.listCollections()) {
|
|
|
+ if (coll.empty() || coll[0] == '_') continue;
|
|
|
+ for (const auto& doc : mem.getAllDocuments(coll)) {
|
|
|
+ std::optional<Document> opt(doc);
|
|
|
+ applyDualWriteMirror(&doc_store, healthy, drift,
|
|
|
+ coll, doc.id, opt, EventType::INSERT);
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ check(doc_store.get("users", "u1").has_value(), "backfill: u1 mirrored");
|
|
|
+ check(doc_store.get("users", "u2").has_value(), "backfill: u2 mirrored");
|
|
|
+ check(doc_store.get("orders", "o1").has_value(), "backfill: o1 mirrored");
|
|
|
+ auto names = doc_store.list_collections();
|
|
|
+ check(std::find(names.begin(), names.end(), "_views") == names.end(),
|
|
|
+ "backfill: _views NOT created in LMDB");
|
|
|
+ check(healthy.load() && drift.load() == 0, "backfill: clean");
|
|
|
+}
|
|
|
+
|
|
|
+} // namespace
|
|
|
+
|
|
|
+int main() {
|
|
|
+ test_unit_null_store_is_noop();
|
|
|
+ test_unit_system_collection_skipped();
|
|
|
+ test_unit_insert_update_delete_round_trip();
|
|
|
+ test_unit_failure_marks_degraded();
|
|
|
+ test_unit_insert_without_doc_is_noop();
|
|
|
+ test_integration_via_memory_store_callback();
|
|
|
+ test_backfill_iteration_pattern();
|
|
|
+
|
|
|
+ std::cout << "dual_write_mirror: " << g_pass << " passed, " << g_fail << " failed\n";
|
|
|
+ return g_fail == 0 ? 0 : 1;
|
|
|
+}
|
