|
@@ -1,10 +1,12 @@
|
|
|
#include "database_service.hpp"
|
|
#include "database_service.hpp"
|
|
|
#include "json_parse.hpp"
|
|
#include "json_parse.hpp"
|
|
|
|
|
+#include "project_addressing.hpp"
|
|
|
#include "storage/document_store.hpp"
|
|
#include "storage/document_store.hpp"
|
|
|
#include "storage/document_store_lmdb.hpp"
|
|
#include "storage/document_store_lmdb.hpp"
|
|
|
#include "storage/dual_write_mirror.hpp"
|
|
#include "storage/dual_write_mirror.hpp"
|
|
|
#include "storage/lmdb_env.hpp"
|
|
#include "storage/lmdb_env.hpp"
|
|
|
#include "storage/migrate_v1_to_v2.hpp"
|
|
#include "storage/migrate_v1_to_v2.hpp"
|
|
|
|
|
+#include "storage/project_store.hpp"
|
|
|
|
|
|
|
|
#include <grpcpp/grpcpp.h>
|
|
#include <grpcpp/grpcpp.h>
|
|
|
#include <grpcpp/resource_quota.h>
|
|
#include <grpcpp/resource_quota.h>
|
|
@@ -209,6 +211,72 @@ bool DatabaseService::initialize() {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+void DatabaseService::migrateLegacyEnvToDefaultProject() {
|
|
|
|
|
+ const auto legacy = config_.dataDirectory / "env";
|
|
|
|
|
+ const auto target = config_.dataDirectory / "projects" /
|
|
|
|
|
+ smartbotic::database::kDefaultProject / "env";
|
|
|
|
|
+
|
|
|
|
|
+ std::error_code ec;
|
|
|
|
|
+ const bool legacy_exists = std::filesystem::exists(legacy, ec);
|
|
|
|
|
+ const bool target_exists = std::filesystem::exists(target, ec);
|
|
|
|
|
+
|
|
|
|
|
+ if (!legacy_exists) return; // fresh install or already migrated
|
|
|
|
|
+
|
|
|
|
|
+ if (target_exists) {
|
|
|
|
|
+ spdlog::error("v2.3 storage: both legacy '{}' and new '{}' exist — "
|
|
|
|
|
+ "refusing to start. Inspect manually; the safe move is "
|
|
|
|
|
+ "to either delete the legacy dir (if you confirm it's a "
|
|
|
|
|
+ "leftover) or stop and contact ops.",
|
|
|
|
|
+ legacy.string(), target.string());
|
|
|
|
|
+ throw std::runtime_error("v2.3 storage migration: ambiguous layout");
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ std::filesystem::create_directories(target.parent_path(), ec);
|
|
|
|
|
+ if (ec) {
|
|
|
|
|
+ throw std::runtime_error("v2.3 storage migration: cannot create '"
|
|
|
|
|
+ + target.parent_path().string() + "': " + ec.message());
|
|
|
|
|
+ }
|
|
|
|
|
+ std::filesystem::rename(legacy, target, ec);
|
|
|
|
|
+ if (ec) {
|
|
|
|
|
+ throw std::runtime_error("v2.3 storage migration: rename '"
|
|
|
|
|
+ + legacy.string() + "' -> '"
|
|
|
|
|
+ + target.string() + "' failed: " + ec.message());
|
|
|
|
|
+ }
|
|
|
|
|
+ spdlog::info("v2.3 storage: migrated legacy env '{}' -> '{}' (default project)",
|
|
|
|
|
+ legacy.string(), target.string());
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+smartbotic::db::storage::DocumentStore* DatabaseService::docStore() noexcept {
|
|
|
|
|
+ return docStore(smartbotic::database::kDefaultProject);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+smartbotic::db::storage::DocumentStore*
|
|
|
|
|
+DatabaseService::docStore(std::string_view project) noexcept {
|
|
|
|
|
+ if (!projects_) return nullptr;
|
|
|
|
|
+ return projects_->get(project);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+std::vector<std::string> DatabaseService::listProjects() const {
|
|
|
|
|
+ if (!projects_) return {};
|
|
|
|
|
+ return projects_->listOnDisk();
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool DatabaseService::createProject(const std::string& name, std::string& error) {
|
|
|
|
|
+ if (!projects_) {
|
|
|
|
|
+ error = "project registry not initialized";
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+ return projects_->create(name, error);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
|
|
+bool DatabaseService::dropProject(const std::string& name, std::string& error) {
|
|
|
|
|
+ if (!projects_) {
|
|
|
|
|
+ error = "project registry not initialized";
|
|
|
|
|
+ return false;
|
|
|
|
|
+ }
|
|
|
|
|
+ return projects_->drop(name, error);
|
|
|
|
|
+}
|
|
|
|
|
+
|
|
|
bool DatabaseService::runMigrations() {
|
|
bool DatabaseService::runMigrations() {
|
|
|
if (!config_.migrations.enabled) {
|
|
if (!config_.migrations.enabled) {
|
|
|
return true;
|
|
return true;
|
|
@@ -224,23 +292,27 @@ bool DatabaseService::runMigrations() {
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
bool DatabaseService::backfillIntoDocStore() {
|
|
bool DatabaseService::backfillIntoDocStore() {
|
|
|
- if (!doc_store_ || !env_) {
|
|
|
|
|
- // Env open failed earlier; nothing to backfill into. mirror_healthy_
|
|
|
|
|
- // stays true (default) — there's no mirror to be unhealthy.
|
|
|
|
|
|
|
+ if (!projects_) {
|
|
|
|
|
+ // Registry open failed earlier; nothing to back-fill into.
|
|
|
return true;
|
|
return true;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
- // Skip when migrate_v1_to_v2 already populated the env. That tool
|
|
|
|
|
- // writes _meta.schema_version=2 atomically on success.
|
|
|
|
|
- try {
|
|
|
|
|
- if (smartbotic::db::storage::migration_complete(*env_)) {
|
|
|
|
|
- spdlog::info("v2.0 backfill: env already migrated (schema_version=2), skipping");
|
|
|
|
|
- return true;
|
|
|
|
|
|
|
+ // v2.3 — MemoryStore stores collection keys in qualified form
|
|
|
|
|
+ // ("project:collection"). For the default project (back-compat
|
|
|
|
|
+ // path) the marker check operates on the default env. Skip when
|
|
|
|
|
+ // already migrated.
|
|
|
|
|
+ auto handle = projects_->getHandle(smartbotic::database::kDefaultProject);
|
|
|
|
|
+ if (handle.env) {
|
|
|
|
|
+ try {
|
|
|
|
|
+ if (smartbotic::db::storage::migration_complete(*handle.env)) {
|
|
|
|
|
+ spdlog::info("v2.3 backfill: default project already migrated, skipping");
|
|
|
|
|
+ return true;
|
|
|
|
|
+ }
|
|
|
|
|
+ } catch (const std::exception& e) {
|
|
|
|
|
+ spdlog::warn("v2.3 backfill: migration_complete probe failed: {} — "
|
|
|
|
|
+ "treating default env as fresh and proceeding",
|
|
|
|
|
+ e.what());
|
|
|
}
|
|
}
|
|
|
- } catch (const std::exception& e) {
|
|
|
|
|
- spdlog::warn("v2.0 backfill: migration_complete probe failed: {} — "
|
|
|
|
|
- "treating env as fresh and proceeding with backfill",
|
|
|
|
|
- e.what());
|
|
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
const auto collections = store_->listCollections();
|
|
const auto collections = store_->listCollections();
|
|
@@ -248,25 +320,39 @@ bool DatabaseService::backfillIntoDocStore() {
|
|
|
uint64_t total_failures = 0;
|
|
uint64_t total_failures = 0;
|
|
|
auto t0 = std::chrono::steady_clock::now();
|
|
auto t0 = std::chrono::steady_clock::now();
|
|
|
|
|
|
|
|
- for (const auto& collection : collections) {
|
|
|
|
|
- if (collection.empty() || collection[0] == '_') continue;
|
|
|
|
|
- const auto docs = store_->getAllDocuments(collection);
|
|
|
|
|
|
|
+ for (const auto& qualified : collections) {
|
|
|
|
|
+ if (qualified.empty() || qualified[0] == '_') continue;
|
|
|
|
|
+ // Each qualified key parses to (project, collection); route the
|
|
|
|
|
+ // mirror write to the right project's env.
|
|
|
|
|
+ smartbotic::database::ProjectCollection pc;
|
|
|
|
|
+ try {
|
|
|
|
|
+ pc = smartbotic::database::parseProjectCollection(qualified);
|
|
|
|
|
+ } catch (const std::exception&) {
|
|
|
|
|
+ // Legacy unqualified key — treat as default project.
|
|
|
|
|
+ pc = {std::string(smartbotic::database::kDefaultProject), qualified};
|
|
|
|
|
+ }
|
|
|
|
|
+ auto* ds = projects_->getOrCreate(pc.project);
|
|
|
|
|
+ if (!ds) {
|
|
|
|
|
+ ++total_failures;
|
|
|
|
|
+ continue;
|
|
|
|
|
+ }
|
|
|
|
|
+ const auto docs = store_->getAllDocuments(qualified);
|
|
|
for (const auto& doc : docs) {
|
|
for (const auto& doc : docs) {
|
|
|
std::optional<Document> opt_doc(doc);
|
|
std::optional<Document> opt_doc(doc);
|
|
|
const uint64_t drift_before = mirror_drift_count_.load(std::memory_order_relaxed);
|
|
const uint64_t drift_before = mirror_drift_count_.load(std::memory_order_relaxed);
|
|
|
smartbotic::db::storage::applyDualWriteMirror(
|
|
smartbotic::db::storage::applyDualWriteMirror(
|
|
|
- doc_store_.get(), mirror_healthy_, mirror_drift_count_,
|
|
|
|
|
- collection, doc.id, opt_doc, EventType::INSERT);
|
|
|
|
|
|
|
+ ds, mirror_healthy_, mirror_drift_count_,
|
|
|
|
|
+ pc.collection, doc.id, opt_doc, EventType::INSERT);
|
|
|
if (mirror_drift_count_.load(std::memory_order_relaxed) > drift_before) {
|
|
if (mirror_drift_count_.load(std::memory_order_relaxed) > drift_before) {
|
|
|
++total_failures;
|
|
++total_failures;
|
|
|
}
|
|
}
|
|
|
++total_docs;
|
|
++total_docs;
|
|
|
if (total_docs % 10000 == 0) {
|
|
if (total_docs % 10000 == 0) {
|
|
|
sd_notify(0, "EXTEND_TIMEOUT_USEC=600000000");
|
|
sd_notify(0, "EXTEND_TIMEOUT_USEC=600000000");
|
|
|
- const std::string status = "STATUS=v2.0 backfill: " +
|
|
|
|
|
|
|
+ const std::string status = "STATUS=v2.3 backfill: " +
|
|
|
std::to_string(total_docs) + " docs mirrored";
|
|
std::to_string(total_docs) + " docs mirrored";
|
|
|
sd_notify(0, status.c_str());
|
|
sd_notify(0, status.c_str());
|
|
|
- spdlog::info("v2.0 backfill: {} docs mirrored ({} failures so far)",
|
|
|
|
|
|
|
+ spdlog::info("v2.3 backfill: {} docs mirrored ({} failures so far)",
|
|
|
total_docs, total_failures);
|
|
total_docs, total_failures);
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
@@ -275,25 +361,23 @@ bool DatabaseService::backfillIntoDocStore() {
|
|
|
const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
const auto elapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
|
|
|
std::chrono::steady_clock::now() - t0).count();
|
|
std::chrono::steady_clock::now() - t0).count();
|
|
|
if (total_failures == 0) {
|
|
if (total_failures == 0) {
|
|
|
- // v2.2.1 — stamp _meta.schema_version=2 so subsequent boots
|
|
|
|
|
- // short-circuit at migration_complete() and skip the re-backfill.
|
|
|
|
|
- // Pre-2.2.1 the marker was only set by the offline migrate_v1_to_v2
|
|
|
|
|
- // tool, so every restart re-walked the entire MemoryStore.
|
|
|
|
|
- try {
|
|
|
|
|
- smartbotic::db::storage::mark_migration_complete(*env_);
|
|
|
|
|
- spdlog::info("v2.0 backfill: complete — {} docs across {} collections "
|
|
|
|
|
- "in {} ms (schema_version=2 marker set)",
|
|
|
|
|
- total_docs, collections.size(), elapsed);
|
|
|
|
|
- } catch (const std::exception& e) {
|
|
|
|
|
- spdlog::warn("v2.0 backfill: completed {} docs in {} ms but the "
|
|
|
|
|
- "schema_version marker write failed: {} — next boot "
|
|
|
|
|
- "will re-backfill",
|
|
|
|
|
- total_docs, elapsed, e.what());
|
|
|
|
|
|
|
+ // Stamp schema_version=2 on every opened project env. Subsequent
|
|
|
|
|
+ // boots short-circuit at migration_complete().
|
|
|
|
|
+ for (const auto& name : projects_->listOpen()) {
|
|
|
|
|
+ auto h = projects_->getHandle(name);
|
|
|
|
|
+ if (!h.env) continue;
|
|
|
|
|
+ try {
|
|
|
|
|
+ smartbotic::db::storage::mark_migration_complete(*h.env);
|
|
|
|
|
+ } catch (const std::exception& e) {
|
|
|
|
|
+ spdlog::warn("v2.3 backfill: schema_version marker write failed for "
|
|
|
|
|
+ "project '{}': {}", name, e.what());
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
|
|
+ spdlog::info("v2.3 backfill: complete — {} docs across {} collections in {} ms",
|
|
|
|
|
+ total_docs, collections.size(), elapsed);
|
|
|
} else {
|
|
} else {
|
|
|
- spdlog::error("v2.0 backfill: completed with {} failures out of {} docs "
|
|
|
|
|
- "in {} ms — mirror_healthy_=false, reads must NOT flip to LMDB "
|
|
|
|
|
- "and schema_version marker NOT set (next boot retries)",
|
|
|
|
|
|
|
+ spdlog::error("v2.3 backfill: completed with {} failures out of {} docs "
|
|
|
|
|
+ "in {} ms — mirror_healthy_=false, reads must NOT flip to LMDB",
|
|
|
total_failures, total_docs, elapsed);
|
|
total_failures, total_docs, elapsed);
|
|
|
}
|
|
}
|
|
|
return true;
|
|
return true;
|
|
@@ -617,50 +701,40 @@ DatabaseService::Config DatabaseService::parseConfig(const nlohmann::json& json)
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
void DatabaseService::setupComponents() {
|
|
void DatabaseService::setupComponents() {
|
|
|
- // v2.0 storage substrate (Stage 4 scaffold).
|
|
|
|
|
|
|
+ // v2.3 multi-project storage substrate.
|
|
|
//
|
|
//
|
|
|
- // Open the LMDB environment at <dataDir>/env/ — same path the v1.x
|
|
|
|
|
- // auto-migration tool in main.cpp writes to. By the time setupComponents
|
|
|
|
|
- // runs, either:
|
|
|
|
|
- // (a) the env already exists and contains migrated v2.0 data, or
|
|
|
|
|
- // (b) this is a fresh deployment and the env is empty.
|
|
|
|
|
- // Both cases are fine: LmdbDocumentStore opens sub-dbs lazily on first
|
|
|
|
|
- // put/read, so an empty env is a valid steady state.
|
|
|
|
|
|
|
+ // Each project lives at `<dataDir>/projects/<name>/env/`. The default
|
|
|
|
|
+ // project always exists and is what v2.0-v2.2 clients (which don't
|
|
|
|
|
+ // address projects) implicitly use.
|
|
|
//
|
|
//
|
|
|
- // map_size_bytes / max_dbs match the values main.cpp uses for migration
|
|
|
|
|
- // so that opening the same env from both call sites is consistent.
|
|
|
|
|
- //
|
|
|
|
|
- // Failure to open the env is currently logged-and-tolerated: Stage 4
|
|
|
|
|
- // does not yet route any handlers through doc_store_, so a missing or
|
|
|
|
|
- // unopenable env should not kill the service. Stage 5+ may upgrade this
|
|
|
|
|
- // to a hard failure once handlers depend on the store.
|
|
|
|
|
- {
|
|
|
|
|
- smartbotic::db::storage::LmdbEnvOpts envOpts;
|
|
|
|
|
- envOpts.path = (config_.dataDirectory / "env").string();
|
|
|
|
|
- envOpts.map_size_bytes = 4ULL << 30; // 4 GiB initial; LMDB grows lazily.
|
|
|
|
|
- envOpts.max_dbs = 1024;
|
|
|
|
|
|
|
+ // Pre-2.3 installs have their data at the legacy `<dataDir>/env/`
|
|
|
|
|
+ // path. migrateLegacyEnvToDefaultProject() detects that layout and
|
|
|
|
|
+ // atomically moves it under `projects/default/env/` before the
|
|
|
|
|
+ // registry opens.
|
|
|
|
|
+ migrateLegacyEnvToDefaultProject();
|
|
|
|
|
|
|
|
- std::error_code ec;
|
|
|
|
|
- std::filesystem::create_directories(envOpts.path, ec);
|
|
|
|
|
- if (ec) {
|
|
|
|
|
- spdlog::warn("v2.0 storage: failed to create env directory '{}': {} -- "
|
|
|
|
|
- "doc_store_ will be left null; Stage 4 still routes through "
|
|
|
|
|
- "MemoryStore so the service continues to boot",
|
|
|
|
|
- envOpts.path, ec.message());
|
|
|
|
|
- } else {
|
|
|
|
|
- try {
|
|
|
|
|
- env_ = std::make_unique<smartbotic::db::storage::LmdbEnv>(envOpts);
|
|
|
|
|
- doc_store_ = std::make_unique<smartbotic::db::storage::LmdbDocumentStore>(*env_);
|
|
|
|
|
- spdlog::info("v2.0 storage: LMDB env opened at {} (mapsize={}MB, max_dbs={})",
|
|
|
|
|
- envOpts.path, envOpts.map_size_bytes >> 20, envOpts.max_dbs);
|
|
|
|
|
- } catch (const std::exception& e) {
|
|
|
|
|
- spdlog::warn("v2.0 storage: failed to open LMDB env at '{}': {} -- "
|
|
|
|
|
- "doc_store_ will be left null; service continues with v1.x",
|
|
|
|
|
- envOpts.path, e.what());
|
|
|
|
|
- env_.reset();
|
|
|
|
|
- doc_store_.reset();
|
|
|
|
|
- }
|
|
|
|
|
- }
|
|
|
|
|
|
|
+ try {
|
|
|
|
|
+ constexpr size_t kMapSize = 4ULL << 30; // 4 GiB per project; LMDB grows lazily.
|
|
|
|
|
+ constexpr uint32_t kMaxDbs = 1024;
|
|
|
|
|
+ projects_ = std::make_unique<smartbotic::db::storage::ProjectStoreRegistry>(
|
|
|
|
|
+ config_.dataDirectory / "projects", kMapSize, kMaxDbs);
|
|
|
|
|
+ const auto opened = projects_->openExisting();
|
|
|
|
|
+ spdlog::info("v2.3 storage: opened {} project env(s): [{}]",
|
|
|
|
|
+ opened.size(),
|
|
|
|
|
+ [&opened]() {
|
|
|
|
|
+ std::string s;
|
|
|
|
|
+ for (size_t i = 0; i < opened.size(); ++i) {
|
|
|
|
|
+ if (i) s += ", ";
|
|
|
|
|
+ s += opened[i];
|
|
|
|
|
+ }
|
|
|
|
|
+ return s;
|
|
|
|
|
+ }());
|
|
|
|
|
+ } catch (const std::exception& e) {
|
|
|
|
|
+ spdlog::error("v2.3 storage: failed to open project registry at '{}': {} -- "
|
|
|
|
|
+ "service cannot start without LMDB; bailing",
|
|
|
|
|
+ (config_.dataDirectory / "projects").string(), e.what());
|
|
|
|
|
+ projects_.reset();
|
|
|
|
|
+ throw;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// Create memory store with eviction config
|
|
// Create memory store with eviction config
|
|
@@ -700,14 +774,17 @@ 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_);
|
|
|
|
|
|
|
+ // v2.3 — wire the LMDB mirror INTO MemoryStore with a project
|
|
|
|
|
+ // resolver. Every write that flows through store_ commits to the
|
|
|
|
|
+ // right per-project LMDB env BEFORE the per-collection lock releases.
|
|
|
|
|
+ if (projects_) {
|
|
|
|
|
+ auto* registry = projects_.get();
|
|
|
|
|
+ store_->setDocumentStoreMirror(
|
|
|
|
|
+ [registry](std::string_view project) -> smartbotic::db::storage::DocumentStore* {
|
|
|
|
|
+ return registry->getOrCreate(project);
|
|
|
|
|
+ },
|
|
|
|
|
+ &mirror_healthy_,
|
|
|
|
|
+ &mirror_drift_count_);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// Create persistence manager. Start from any persistenceConfig values the
|
|
// Create persistence manager. Start from any persistenceConfig values the
|