#include "database_service.hpp" #include "json_parse.hpp" #include "project_addressing.hpp" #include "storage/document_store.hpp" #include "storage/document_store_lmdb.hpp" #include "storage/dual_write_mirror.hpp" #include "auth/auth_interceptor.hpp" #include "auth/principal.hpp" #include "tls/cert_generator.hpp" #include #include #include #include #include "storage/lmdb_env.hpp" #include "storage/migrate_v1_to_v2.hpp" #include "storage/project_store.hpp" #include "storage/subdb_placement.hpp" #include #include #include #include #ifdef HAVE_SYSTEMD #include #endif #if defined(__GLIBC__) && !defined(__APPLE__) #include #endif namespace smartbotic::database { DatabaseService::DatabaseService(Config config) : config_(std::move(config)) { } DatabaseService::~DatabaseService() { stop(); } std::string DatabaseService::readOnlyReason() const { std::lock_guard lock(reason_mutex_); return read_only_reason_; } void DatabaseService::setReadOnly(bool value, const std::string& reason) { { std::lock_guard lock(reason_mutex_); read_only_reason_ = value ? reason : ""; } bool prev = read_only_.exchange(value, std::memory_order_acq_rel); if (prev != value) { if (value) { spdlog::error("Database entered READ-ONLY mode: {}", reason); } else { spdlog::info("Database unlocked -- writes accepted"); } } } bool DatabaseService::initialize() { spdlog::info("Initializing database service (node: {})", config_.nodeId); try { // Create data directory if needed std::error_code ec; std::filesystem::create_directories(config_.dataDirectory, ec); if (ec) { spdlog::error("Failed to create data directory: {}", ec.message()); return false; } setupComponents(); // Initialize encryption if (config_.encryptionEnabled) { if (!encryption_->initialize()) { spdlog::error("Failed to initialize encryption"); return false; } } // Recover from persistence recovery_outcome_ = persistence_->recover(*store_); #if defined(__GLIBC__) && !defined(__APPLE__) // v1.9.3 — release freelist pages accumulated during recovery. v1.9.1 // added trim at end of loadSnapshot, but the WAL replay phase that // runs afterward (`persistence_->recover`'s `replayWal` step) also // burns through GBs of small Document JSON allocations whose pages // sit on the per-thread freelist with no subsequent allocation to // shake them loose. On Zoe (BUG-memory-leak-zoe.md update 15:05) // this was 5.7 GB at 23 min uptime — fully reclaimable via // malloc_trim, just nothing called it. Paired with the periodic // every-5-minute trim in MemoryStore::logMemoryCheck, so any // post-boot bloat that escapes this trim gets cleaned up shortly. ::malloc_trim(0); #endif if (recovery_outcome_.isFailure()) { const std::string modeStr = recoveryModeToString(config_.persistenceConfig.recoveryMode); spdlog::error(""); spdlog::error("+------------------------------------------------------------------+"); spdlog::error("| RECOVERY REFUSED |"); spdlog::error("| |"); spdlog::error("| Mode: {}", modeStr); spdlog::error("| Expected snapshot: {}", recovery_outcome_.expectedSnapshot.string()); spdlog::error("| Reason: {}", recovery_outcome_.failureReason); spdlog::error("| |"); spdlog::error("| Snapshots available: {}", recovery_outcome_.snapshotsAvailable); spdlog::error("| Snapshots attempted: {}", recovery_outcome_.snapshotsAttempted); spdlog::error("| |"); spdlog::error("| To escalate, restart with one of: |"); spdlog::error("| --recovery-mode=snapshot_fallback Try older snapshots |"); spdlog::error("| --recovery-mode=wal_only Replay WAL only (slow) |"); spdlog::error("| --recovery-mode=best_effort Try all of the above |"); spdlog::error("| --recovery-mode=force_empty Start empty (LAST RESORT) |"); spdlog::error("| |"); spdlog::error("| Or set in config.json: \"recovery\": {{ \"mode\": \"\" }} |"); spdlog::error("| |"); spdlog::error("| PRESERVE /var/lib/smartbotic-database/ BEFORE ESCALATING. |"); spdlog::error("+------------------------------------------------------------------+"); throw std::runtime_error("recovery failed; refusing to start"); } // Auto-readonly mode on non-trivial recovery if (recovery_outcome_.isNonTrivial() && !force_readwrite_) { std::string reason; switch (recovery_outcome_.kind) { case RecoveryOutcome::Kind::SnapshotFellBack: reason = "fell back to snapshot " + recovery_outcome_.snapshotUsed.filename().string() + " because " + recovery_outcome_.expectedSnapshot.filename().string() + " failed: " + recovery_outcome_.failureReason; break; case RecoveryOutcome::Kind::WalOnlyReplay: reason = "WAL-only replay, no snapshot loaded (" + std::to_string(recovery_outcome_.walEntriesReplayed) + " entries)"; break; case RecoveryOutcome::Kind::ForcedEmpty: reason = "forced empty by operator (--recovery-mode=force_empty)"; break; default: reason = "non-trivial recovery"; break; } reason += ". Run `smartbotic-db-cli unlock` to accept this state, " "or restart with --force-readwrite to bypass this check."; setReadOnly(true, reason); spdlog::error(""); spdlog::error("+------------------------------------------------------------------+"); spdlog::error("| [ERROR] Database booted in READ-ONLY mode after non-trivial |"); spdlog::error("| recovery. |"); spdlog::error("| |"); spdlog::error("| Reason: {}", reason); spdlog::error("| |"); spdlog::error("| Writes will be REJECTED until you acknowledge this state: |"); spdlog::error("| smartbotic-db-cli unlock # live, no restart |"); spdlog::error("| smartbotic-database --force-readwrite # on next restart |"); spdlog::error("+------------------------------------------------------------------+"); } // Set replication sequence after recovery (WAL sequence is now known) 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 // populated by the persistence recovery above). view_manager_->loadFromStore(); // Load per-collection configs from the _collection_meta system collection. // Must happen AFTER persistence recovery and BEFORE migrations so any // migration-created documents are stamped with the correct precision. config_manager_->loadFromStore(); policy_manager_->loadFromStore(); // Run migrations if enabled if (config_.migrations.enabled && !config_.migrations.directory.empty()) { if (!runMigrations()) { if (config_.migrations.failOnError) { spdlog::error("Failed to run migrations"); return false; } spdlog::warn("Some migrations failed, continuing anyway"); } } // v2.0 Stage 4 — synchronous backfill into doc_store_ before // serving. Without this, only post-boot writes land in LMDB and // the read-flip downstream would see an empty mirror. backfillIntoDocStore(); // v2.4.4 — placement audit. Reports documents sitting in a sub-db // other than the one they declare, which is the signature of a // misbound MDB_dbi (see storage/subdb_identity.hpp). Read-only and // advisory: it never blocks startup, because a misplacement is a // data-location problem an operator repairs offline with // `smartbotic-db-cli reconcile-subdbs`, not a reason to refuse // service on the other 99% of the dataset. auditSubdbPlacement(); // v2.6.0 — stamp `project: "default"` onto file records written before // files were namespaced. Idempotent, and advisory: a file-metadata // stamp is not a reason to refuse service, so failures warn only. if (files_) { try { const uint32_t stamped = files_->stampMissingProjects(); if (stamped > 0) { spdlog::info("v2.6.0 file migration: stamped {} legacy file " "record(s) with project 'default'", stamped); } else { spdlog::debug("v2.6.0 file migration: no legacy file records"); } } catch (const std::exception& e) { spdlog::warn("v2.6.0 file migration failed: {}", e.what()); } } spdlog::info("Database service initialized successfully"); return true; } catch (const std::exception& e) { spdlog::error("Failed to initialize database service: {}", e.what()); return false; } } 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 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() { if (!config_.migrations.enabled) { return true; } MigrationRunner::Config migrationConfig; migrationConfig.directory = config_.migrations.directory; migrationConfig.autoApply = config_.migrations.autoApply; migrationConfig.failOnError = config_.migrations.failOnError; migrationRunner_ = std::make_unique(*store_, *view_manager_, migrationConfig); return migrationRunner_->runMigrations(); } void DatabaseService::auditSubdbPlacement() { if (!projects_) return; for (const auto& name : projects_->listOpen()) { auto h = projects_->getHandle(name); if (!h.env) continue; try { // audit_env(), NOT audit(). The path-taking overload opens a // second MDB_env on the same file; closing it would drop the // POSIX record locks this process already holds for `h.env` // (POSIX locks are per-process, and closing any fd on a file // releases all of them), leaving every later mdb_txn_begin // failing EINVAL. Borrow the open handle instead. const auto rep = smartbotic::db::storage::audit_env(h.env->raw(), name); if (rep.misplaced.empty()) { spdlog::debug("placement audit: project '{}' consistent ({} rows)", name, rep.rows_scanned); continue; } // Summarise per (physical -> declared home) pair; one line per // affected document would be unreadable at scale. std::map pairs; for (const auto& m : rep.misplaced) { ++pairs[m.physical_subdb + " -> " + m.home_subdb]; } spdlog::error("placement audit: project '{}' has {} document(s) in the " "wrong sub-db. Reads served from LMDB will not find them " "under their own collection. Repair offline with: " "smartbotic-db-cli reconcile-subdbs --env {} --project {} --apply", name, rep.misplaced.size(), h.env->path(), name); for (const auto& [pair, count] : pairs) { spdlog::error("placement audit: {} document(s) {}", count, pair); } } catch (const std::exception& e) { // Advisory only — a failed audit must not take the service down. spdlog::warn("placement audit: project '{}' could not be audited: {}", name, e.what()); } } } bool DatabaseService::backfillIntoDocStore() { if (!projects_) { // Registry open failed earlier; nothing to back-fill into. 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()); } } const auto collections = store_->listCollections(); uint64_t total_docs = 0; uint64_t total_failures = 0; auto t0 = std::chrono::steady_clock::now(); 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) { std::optional opt_doc(doc); const uint64_t drift_before = mirror_drift_count_.load(std::memory_order_relaxed); smartbotic::db::storage::applyDualWriteMirror( 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) { ++total_failures; } ++total_docs; if (total_docs % 10000 == 0) { sd_notify(0, "EXTEND_TIMEOUT_USEC=600000000"); const std::string status = "STATUS=v2.3 backfill: " + std::to_string(total_docs) + " docs mirrored"; sd_notify(0, status.c_str()); spdlog::info("v2.3 backfill: {} docs mirrored ({} failures so far)", total_docs, total_failures); } } } const auto elapsed = std::chrono::duration_cast( std::chrono::steady_clock::now() - t0).count(); if (total_failures == 0) { // 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 { 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); } return true; } void DatabaseService::start() { if (running_.exchange(true)) { return; } spdlog::info("Starting database service with {} listener(s)", config_.listeners.size()); // Start components store_->start(); persistence_->start(); events_->start(); files_->start(); if (config_.replicationEnabled) { replication_->start(); // Set initial local collections for discovery replication_->setLocalCollections(store_->listCollections()); } // Start gRPC server startGrpcServer(); spdlog::info("Database service started"); } void DatabaseService::stop() { if (!running_.exchange(false)) { return; } spdlog::info("Stopping database service..."); // Stop gRPC server first stopGrpcServer(); // Stop components in reverse order if (replication_) { replication_->stop(); } if (files_) { files_->stop(); } if (events_) { 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`. // // v1.8.2 — extend systemd's stop watchdog before doing it. On Zoe // (1.85M docs / ~5 GB tracked) the serialize takes ~70 s and the // default TimeoutStopSec=30s SIGKILLed the process mid-write, peak // RSS hit 11 GB (live state + uncompressed serialize buffer), and // the new snapshot never made it to disk anyway. EXTEND_TIMEOUT_USEC // tells systemd we're working — same protocol the recovery path // uses during startup. Paired with TimeoutStopSec=300 in the unit // file as the static cap. #ifdef HAVE_SYSTEMD sd_notify(0, "EXTEND_TIMEOUT_USEC=600000000"); sd_notify(0, "STATUS=Writing final snapshot"); #endif 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_) { persistence_->stop(); } if (store_) { store_->stop(); } spdlog::info("Database service stopped"); } void DatabaseService::wait() { if (serverThread_.joinable()) { serverThread_.join(); } } void DatabaseService::signalStop() { stopRequested_ = true; stop(); } nlohmann::json DatabaseService::getStats() const { nlohmann::json stats; if (store_) { auto storeStats = store_->getStats(); stats["documents"] = storeStats.totalDocuments; stats["collections"] = storeStats.totalCollections; stats["memory_bytes"] = storeStats.estimatedMemoryBytes; stats["inserts"] = storeStats.insertCount; stats["updates"] = storeStats.updateCount; stats["deletes"] = storeStats.deleteCount; stats["queries"] = storeStats.queryCount; } if (persistence_) { auto persistStats = persistence_->getStats(); stats["wal_sequence"] = persistStats.walSequence; stats["wal_size_bytes"] = persistStats.walSizeBytes; stats["snapshot_count"] = persistStats.snapshotCount; stats["last_snapshot_sequence"] = persistStats.lastSnapshotSequence; } if (events_) { stats["subscriptions"] = events_->subscriptionCount(); } return stats; } DatabaseService::Config DatabaseService::loadConfig(const std::filesystem::path& configPath) { std::ifstream file(configPath); if (!file) { throw std::runtime_error("Failed to open config file: " + configPath.string()); } nlohmann::json json; file >> json; return parseConfig(json); } DatabaseService::Config DatabaseService::parseConfig(const nlohmann::json& json) { Config config; // Check for both "storage" and "database" keys for backward compatibility const nlohmann::json* dbConfig = nullptr; if (json.contains("database")) { dbConfig = &json["database"]; } else if (json.contains("storage")) { dbConfig = &json["storage"]; } if (dbConfig) { const auto& db = *dbConfig; config.nodeId = db.value("node_id", config.nodeId); config.bindAddress = db.value("bind_address", config.bindAddress); config.rpcPort = db.value("rpc_port", config.rpcPort); // v2.4 — parse the per-listener fleet. If "listeners" is set, // it overrides the legacy bind_address+rpc_port. Otherwise we // synthesise one listener below using those legacy fields. if (db.contains("listeners") && db["listeners"].is_array()) { for (const auto& l : db["listeners"]) { ListenerConfig lc; lc.bind = l.value("bind", lc.bind); lc.port = l.value("port", lc.port); if (l.contains("tls") && l["tls"].is_object()) { const auto& t = l["tls"]; lc.tls.enabled = t.value("enabled", false); lc.tls.cert_path = t.value("cert_path", std::string{}); lc.tls.key_path = t.value("key_path", std::string{}); lc.tls.auto_self_signed_if_missing = t.value("auto_self_signed_if_missing", true); } if (l.contains("auth") && l["auth"].is_object()) { const auto& a = l["auth"]; lc.auth.required = a.value("required", false); if (a.contains("keys") && a["keys"].is_array()) { for (const auto& k : a["keys"]) { // v2.7.0 — a key is either a bare token (v2.4-v2.6 // shape) or {"name","key"}. The name becomes the // principal that policy is written against; a bare // token maps to the reserved `unnamed`, so old // configs keep authenticating but cannot be told // apart in a policy until they are named. if (k.is_string()) { lc.auth.keys.push_back( {smartbotic::database::auth::kPrincipalUnnamed, k.get()}); } else if (k.is_object()) { const std::string name = k.value("name", std::string{}); const std::string key = k.value("key", std::string{}); if (key.empty()) { spdlog::warn("auth: listener {}:{} has a key entry " "with no `key` value - skipped", lc.bind, lc.port); continue; } lc.auth.keys.push_back( {name.empty() ? std::string(smartbotic::database::auth::kPrincipalUnnamed) : name, key}); } } } } config.listeners.push_back(std::move(lc)); } } // Expand environment variables in data directory std::string dataDir = db.value("data_directory", ""); if (dataDir.find("${HOME}") != std::string::npos) { const char* home = std::getenv("HOME"); if (home) { size_t pos = dataDir.find("${HOME}"); dataDir.replace(pos, 7, home); } } config.dataDirectory = dataDir; // Migrations settings if (db.contains("migrations")) { auto& migrations = db["migrations"]; config.migrations.enabled = migrations.value("enabled", config.migrations.enabled); config.migrations.autoApply = migrations.value("auto_apply", config.migrations.autoApply); config.migrations.failOnError = migrations.value("fail_on_error", config.migrations.failOnError); std::string migDir = migrations.value("directory", ""); if (migDir.find("${HOME}") != std::string::npos) { const char* home = std::getenv("HOME"); if (home) { size_t pos = migDir.find("${HOME}"); migDir.replace(pos, 7, home); } } config.migrations.directory = migDir; } // Memory eviction settings if (db.contains("memory")) { auto& memory = db["memory"]; // v2.0 Stage 8 — deprecation log. These knobs control MemoryStore // eviction, which is still active for the MemoryStore-side mirror // but does NOT bound RSS in v2.0 (LMDB mmap is the dominant RSS // contributor). The substrate-level equivalent is the LMDB env // mapsize and OS page cache. v2.1 will rename `max_memory_mb` to // `buffer_pool_size_mb` per the Phase C plan. for (const char* deprecated : {"max_memory_mb", "eviction_threshold_percent", "eviction_target_percent", "eviction_check_interval_ms", "eviction_chunk_size", "eviction_chunk_pause_ms", "max_eviction_passes_per_trigger", "hot_write_floor_ms", "memory_priority"}) { if (memory.contains(deprecated)) { spdlog::warn("v2.0 deprecation: storage.memory.{} is deprecated " "and will be removed in v2.1. v2.0 RSS is bounded " "by the LMDB env mapsize + OS page cache, not by " "MemoryStore eviction. Setting still applied to " "the MemoryStore mirror for back-compat.", deprecated); } } config.maxMemoryMb = memory.value("max_memory_mb", config.maxMemoryMb); config.evictionThresholdPercent = memory.value("eviction_threshold_percent", config.evictionThresholdPercent); config.evictionTargetPercent = memory.value("eviction_target_percent", config.evictionTargetPercent); config.evictionCheckIntervalMs = memory.value("eviction_check_interval_ms", config.evictionCheckIntervalMs); // NEW v1.7.0 eviction tuning (schema-only in T2; runtime use lands in T3/T4) config.evictionChunkSize = memory.value("eviction_chunk_size", config.evictionChunkSize); config.evictionChunkPauseMs = memory.value("eviction_chunk_pause_ms", config.evictionChunkPauseMs); config.maxEvictionPassesPerTrigger = memory.value("max_eviction_passes_per_trigger", config.maxEvictionPassesPerTrigger); config.hotWriteFloorMs = memory.value("hot_write_floor_ms", config.hotWriteFloorMs); config.memorySoftPercent = memory.value("memory_soft_percent", config.memorySoftPercent); config.memoryHardPercent = memory.value("memory_hard_percent", config.memoryHardPercent); config.memoryEmergencyPercent = memory.value("memory_emergency_percent", config.memoryEmergencyPercent); // v1.7.0 T10 — eviction burst event threshold config.evictionBurstThreshold = memory.value("eviction_burst_threshold", config.evictionBurstThreshold); // v2.4.3 — eviction drain cap (0 disables) config.evictionMaxEpisodePercent = memory.value("eviction_max_episode_percent", config.evictionMaxEpisodePercent); } // Persistence settings if (db.contains("persistence")) { auto& persistence = db["persistence"]; config.walSyncIntervalMs = persistence.value("wal_sync_interval_ms", config.walSyncIntervalMs); config.snapshotIntervalSec = persistence.value("snapshot_interval_sec", config.snapshotIntervalSec); config.compressionEnabled = persistence.value("compression", "lz4") != "none"; // Snapshot durability (NEW in v1.6.1) if (persistence.contains("snapshots")) { const auto& snap = persistence["snapshots"]; config.persistenceConfig.validateAfterWrite = snap.value("validate_after_write", config.persistenceConfig.validateAfterWrite); config.persistenceConfig.cleanupOnlyIfVerified = snap.value("cleanup_only_if_verified", config.persistenceConfig.cleanupOnlyIfVerified); } // Recovery (NEW in v1.6.1) if (persistence.contains("recovery")) { const auto& rec = persistence["recovery"]; std::string modeStr = rec.value("mode", std::string("normal")); try { config.persistenceConfig.recoveryMode = recoveryModeFromString(modeStr); } catch (const std::exception& e) { spdlog::warn("Invalid recovery.mode '{}', defaulting to 'normal': {}", modeStr, e.what()); config.persistenceConfig.recoveryMode = RecoveryMode::Normal; } config.persistenceConfig.autoEscalate = rec.value("auto_escalate", config.persistenceConfig.autoEscalate); config.persistenceConfig.allowEmptyOnFreshInstall = rec.value("allow_empty_on_fresh_install", config.persistenceConfig.allowEmptyOnFreshInstall); } } // File settings if (db.contains("files")) { auto& files = db["files"]; config.maxFileSizeMb = files.value("max_file_size_mb", config.maxFileSizeMb); config.fileCleanupIntervalSec = files.value("cleanup_orphans_interval_sec", config.fileCleanupIntervalSec); if (files.contains("allowed_types")) { for (const auto& type : files["allowed_types"]) { config.allowedFileTypes.push_back(type.get()); } } } // Encryption settings if (db.contains("encryption")) { auto& encryption = db["encryption"]; config.encryptionEnabled = encryption.value("enabled", config.encryptionEnabled); config.autoGenerateKey = encryption.value("auto_generate_key", config.autoGenerateKey); std::string keyFile = encryption.value("key_file", ""); if (keyFile.find("${HOME}") != std::string::npos) { const char* home = std::getenv("HOME"); if (home) { size_t pos = keyFile.find("${HOME}"); keyFile.replace(pos, 7, home); } } config.keyFilePath = keyFile; } // Replication settings if (db.contains("replication")) { auto& replication = db["replication"]; config.replicationEnabled = replication.value("enabled", config.replicationEnabled); config.conflictResolution = replication.value("conflict_resolution", config.conflictResolution); if (replication.contains("peers")) { for (const auto& peer : replication["peers"]) { config.peerAddresses.push_back(peer.get()); } } } // gRPC settings (v1.6.2 — concurrency cap + configurable message sizes) if (db.contains("grpc")) { const auto& grpc = db["grpc"]; config.grpc.maxReceiveMessageSizeMb = grpc.value("max_receive_message_size_mb", config.grpc.maxReceiveMessageSizeMb); config.grpc.maxSendMessageSizeMb = grpc.value("max_send_message_size_mb", config.grpc.maxSendMessageSizeMb); config.grpc.resourceQuotaMemoryMb = grpc.value("resource_quota_memory_mb", config.grpc.resourceQuotaMemoryMb); config.grpc.maxConcurrentSubscribeStreams = grpc.value("max_concurrent_subscribe_streams", config.grpc.maxConcurrentSubscribeStreams); config.grpc.maxConcurrentFileStreams = grpc.value("max_concurrent_file_streams", config.grpc.maxConcurrentFileStreams); } } // v2.4 back-compat shim. If no listeners[] was provided, synthesize // one from the legacy bind_address + rpc_port fields. This is what // every existing v2.3 deployment will hit on first v2.4 boot — // their configs don't know about listeners[] yet, and they get // exactly the same listener they had before. if (config.listeners.empty()) { ListenerConfig legacy; legacy.bind = config.bindAddress; legacy.port = config.rpcPort; legacy.tls.enabled = false; legacy.auth.required = false; config.listeners.push_back(std::move(legacy)); } return config; } void DatabaseService::setupComponents() { // v2.3 multi-project storage substrate. // // Each project lives at `/projects//env/`. The default // project always exists and is what v2.0-v2.2 clients (which don't // address projects) implicitly use. // // Pre-2.3 installs have their data at the legacy `/env/` // path. migrateLegacyEnvToDefaultProject() detects that layout and // atomically moves it under `projects/default/env/` before the // registry opens. migrateLegacyEnvToDefaultProject(); try { constexpr size_t kMapSize = 4ULL << 30; // 4 GiB per project; LMDB grows lazily. constexpr uint32_t kMaxDbs = 1024; projects_ = std::make_unique( 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 MemoryStore::Config storeConfig; storeConfig.nodeId = config_.nodeId; storeConfig.maxMemoryBytes = config_.maxMemoryMb * 1024ULL * 1024ULL; storeConfig.evictionThresholdPercent = config_.evictionThresholdPercent; storeConfig.evictionTargetPercent = config_.evictionTargetPercent; storeConfig.evictionCheckIntervalMs = config_.evictionCheckIntervalMs; // NEW v1.7.0 eviction tuning (wired in T2; consumed in T3/T4). storeConfig.evictionChunkSize = config_.evictionChunkSize; storeConfig.evictionChunkPauseMs = config_.evictionChunkPauseMs; storeConfig.maxEvictionPassesPerTrigger = config_.maxEvictionPassesPerTrigger; storeConfig.hotWriteFloorMs = config_.hotWriteFloorMs; storeConfig.memorySoftPercent = config_.memorySoftPercent; storeConfig.memoryHardPercent = config_.memoryHardPercent; storeConfig.memoryEmergencyPercent = config_.memoryEmergencyPercent; storeConfig.evictionBurstThreshold = config_.evictionBurstThreshold; storeConfig.evictionMaxEpisodePercent = config_.evictionMaxEpisodePercent; store_ = std::make_unique(storeConfig); // Create view manager (cache loaded in initialize() after persistence recovery) view_manager_ = std::make_unique(*store_); // Create per-collection config manager and attach it to the store so the // write paths route document timestamp stamps through it. config_manager_ = std::make_unique(*store_); // v2.8.0 — access policy. Constructed here; its cache is loaded after // recovery alongside the view and collection-config caches. policy_manager_ = std::make_unique(*store_); store_->setConfigManager(config_manager_.get()); // v1.9.0 — disk-resident version history. Replaces the in-heap // `CollectionData::versionHistory` deques that were the dominant // source of the Zoe untracked-RSS gap. One file per collection at // `/history/.hlog`. Lifecycle: constructed // here, wired into the store immediately so snapshot deserializer // can migrate pre-v1.9 in-memory history blocks straight to disk. HistoryStore::Config histConfig; histConfig.dataDir = config_.dataDirectory; history_store_ = std::make_unique(histConfig); store_->setHistoryStore(history_store_.get()); // 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 // caller (config loader / CLI parser) has already populated — including // recoveryMode — and overlay the top-level convenience fields. PersistenceManager::Config persistConfig = config_.persistenceConfig; persistConfig.dataDir = config_.dataDirectory; persistConfig.walSyncIntervalMs = config_.walSyncIntervalMs; persistConfig.snapshotIntervalSec = config_.snapshotIntervalSec; persistConfig.compressionEnabled = config_.compressionEnabled; persistConfig.nodeId = config_.nodeId; // Mirror the resolved recoveryMode back into our Config so downstream code // (logging, RPC handlers) can inspect config_.persistenceConfig.recoveryMode // without having to reach into the PersistenceManager. config_.persistenceConfig = persistConfig; persistence_ = std::make_unique(persistConfig); // Connect store callbacks to persistence - uses setPersistCallback for WAL logging store_->setPersistCallback([this](const std::string& collection, const std::string& id, const std::optional& doc, EventType eventType) { // 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) { case EventType::INSERT: if (doc) walSeq = persistence_->logInsert(collection, *doc); break; case EventType::UPDATE: if (doc) walSeq = persistence_->logUpdate(collection, *doc); break; case EventType::DELETE: persistence_->logDelete(collection, id); break; default: break; } if (walSeq > 0 && (eventType == EventType::INSERT || eventType == EventType::UPDATE)) { store_->recordWalSequence(collection, id, walSeq); } // 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 if (replication_ && config_.replicationEnabled) { databasepb::ReplicationEntry entry; entry.set_collection(collection); 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( std::chrono::duration_cast( std::chrono::system_clock::now().time_since_epoch() ).count()); switch (eventType) { case EventType::INSERT: entry.set_op(databasepb::OP_INSERT); // Send the full Document envelope (id, collection, data, // version, timestamps, encryption state). The follower's // applyReplicatedEntry uses Document::fromJson() which // expects this envelope — sending only doc->data silently // dropped every user field on the other side. if (doc) entry.set_data(doc->toJson().dump()); break; case EventType::UPDATE: entry.set_op(databasepb::OP_UPDATE); if (doc) entry.set_data(doc->toJson().dump()); break; case EventType::DELETE: entry.set_op(databasepb::OP_DELETE); break; default: break; } replication_->queueForReplication(entry); } // Publish event if (events_) { DatabaseEvent event; event.type = eventType; event.collection = collection; event.documentId = id; event.timestamp = std::chrono::duration_cast( std::chrono::system_clock::now().time_since_epoch() ).count(); event.nodeId = config_.nodeId; if (doc) { event.data = doc->data(); } events_->publish(event); } }); // 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, const std::string& id, uint64_t walSequence) -> std::optional { uint64_t fromSequence = walSequence > 0 ? walSequence - 1 : 0; return persistence_->loadDocument(collection, id, fromSequence); }); // Create encryption manager EncryptionManager::Config encryptConfig; encryptConfig.enabled = config_.encryptionEnabled; encryptConfig.keyFilePath = config_.keyFilePath; encryptConfig.autoGenerateKey = config_.autoGenerateKey; encryption_ = std::make_unique(encryptConfig); // Create event manager EventManager::Config eventConfig; eventConfig.nodeId = config_.nodeId; events_ = std::make_unique(eventConfig); // Create file manager FileManager::Config fileConfig; fileConfig.filesDir = config_.dataDirectory / "files"; fileConfig.maxFileSizeMb = config_.maxFileSizeMb; fileConfig.allowedMimeTypes = config_.allowedFileTypes; fileConfig.cleanupIntervalSec = config_.fileCleanupIntervalSec; files_ = std::make_unique(fileConfig); // Create replication manager ReplicationManager::Config replConfig; replConfig.nodeId = config_.nodeId; replConfig.peerAddresses = config_.peerAddresses; replConfig.conflictResolution = config_.conflictResolution; replication_ = std::make_unique(replConfig); // Wire replication entry apply callback replication_->setEntryApplyCallback([this](const databasepb::ReplicationEntry& entry) { applyReplicatedEntry(entry); }); // Create gRPC implementations storageImpl_ = std::make_unique( *this, *store_, *persistence_, *events_, *files_, *encryption_, *view_manager_, *config_manager_, *policy_manager_ ); // v1.6.2 — wire the streaming-RPC concurrency limits from GrpcConfig. storageImpl_->setStreamLimits( config_.grpc.maxConcurrentSubscribeStreams, config_.grpc.maxConcurrentFileStreams); replicationImpl_ = std::make_unique( *store_, *replication_, *persistence_ ); } void DatabaseService::applyReplicatedEntry(const databasepb::ReplicationEntry& entry) { 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()) { case databasepb::OP_INSERT: case databasepb::OP_UPDATE: case databasepb::OP_UPSERT: { if (entry.data().empty()) { spdlog::warn("Replication entry has no data for op {}", static_cast(entry.op())); return; } auto json = smartbotic::db::parse_to_nlohmann(entry.data()); Document doc = Document::fromJson(json); doc.id = entry.document_id(); doc.nodeId = entry.node_id(); // 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); // v2.3.1 — also mirror to LMDB. loadDocument bypasses the // MemoryStore persist callback that normally drives the // dual-write mirror; in v2.0-v2.2 the mirror was a side // channel and reads still hit MemoryStore, so the bypass // was harmless. In v2.3 reads are LMDB-first, so a // follower that only filled MemoryStore would return // empty Find/Get for every replicated doc. Route the // entry through the same project-aware mirror the write // handlers use. if (projects_) { auto pc = smartbotic::database::parseProjectCollection( entry.collection()); if (auto* ds = projects_->getOrCreate(pc.project)) { std::optional opt_doc(doc); smartbotic::db::storage::applyDualWriteMirror( ds, mirror_healthy_, mirror_drift_count_, pc.collection, doc.id, opt_doc, EventType::INSERT); } } 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 {}", entry.op() == databasepb::OP_INSERT ? "insert" : entry.op() == databasepb::OP_UPDATE ? "update" : "upsert", entry.collection(), entry.document_id(), entry.node_id()); break; } case databasepb::OP_DELETE: { store_->remove(entry.collection(), entry.document_id()); if (persistence_) { persistence_->logDelete(entry.collection(), entry.document_id(), origin); } spdlog::trace("Applied replicated delete to {}/{} from {}", entry.collection(), entry.document_id(), entry.node_id()); break; } case databasepb::OP_CREATE_COLLECTION: { CollectionOptions options; store_->createCollection(entry.collection(), options); if (persistence_) { persistence_->logCreateCollection(entry.collection(), options, origin); } spdlog::trace("Applied replicated create collection {} from {}", entry.collection(), entry.node_id()); break; } case databasepb::OP_DROP_COLLECTION: { store_->dropCollection(entry.collection()); if (persistence_) { persistence_->logDropCollection(entry.collection(), origin); } spdlog::trace("Applied replicated drop collection {} from {}", entry.collection(), entry.node_id()); break; } default: spdlog::warn("Unknown replication operation type: {}", static_cast(entry.op())); break; } } catch (const std::exception& e) { spdlog::error("Failed to apply replicated entry: {}", e.what()); } } void DatabaseService::startGrpcServer() { if (config_.listeners.empty()) { // Should never hit — parseConfig's back-compat shim always // synthesises at least one. Defend anyway. throw std::runtime_error("startGrpcServer: no listeners configured"); } const int64_t mb = 1024 * 1024; spdlog::info("gRPC config: recv={}MB send={}MB quota={}MB subs={} files={}", config_.grpc.maxReceiveMessageSizeMb, config_.grpc.maxSendMessageSizeMb, config_.grpc.resourceQuotaMemoryMb, config_.grpc.maxConcurrentSubscribeStreams, config_.grpc.maxConcurrentFileStreams); grpcServers_.reserve(config_.listeners.size()); for (const auto& listener : config_.listeners) { const std::string addr = listener.bind + ":" + std::to_string(listener.port); grpc::ServerBuilder builder; // v2.4 — choose credentials per listener. Plaintext for // tls.enabled=false; SslServerCredentials with the operator's // cert+key (or an auto-generated self-signed pair) for true. std::shared_ptr creds; if (listener.tls.enabled) { std::filesystem::path cert_path = listener.tls.cert_path; std::filesystem::path key_path = listener.tls.key_path; // Resolve cert+key. If either is missing/unreadable AND // auto_self_signed_if_missing is set, generate a pair. const bool cert_ok = !cert_path.empty() && std::filesystem::exists(cert_path); const bool key_ok = !key_path.empty() && std::filesystem::exists(key_path); if (!cert_ok || !key_ok) { if (!listener.tls.auto_self_signed_if_missing) { throw std::runtime_error( "TLS listener " + addr + ": cert or key missing and auto_self_signed_if_missing=false"); } const auto out_dir = config_.dataDirectory / "tls"; auto gen = smartbotic::database::tls_util::generateSelfSignedCert( out_dir, listener.bind); cert_path = gen.cert_path; key_path = gen.key_path; spdlog::warn( "TLS listener {}: no operator cert at '{}' (or key at '{}'); " "auto-generated self-signed cert at '{}' (key '{}'). " "OK for dev. For production, drop a real cert + key at " "the configured cert_path/key_path.", addr, listener.tls.cert_path.string(), listener.tls.key_path.string(), cert_path.string(), key_path.string()); } auto slurp = [](const std::filesystem::path& p) { std::ifstream f(p); if (!f) throw std::runtime_error("read " + p.string()); std::stringstream ss; ss << f.rdbuf(); return ss.str(); }; grpc::SslServerCredentialsOptions sslOpts; grpc::SslServerCredentialsOptions::PemKeyCertPair kp; kp.private_key = slurp(key_path); kp.cert_chain = slurp(cert_path); sslOpts.pem_key_cert_pairs.push_back(std::move(kp)); creds = grpc::SslServerCredentials(sslOpts); } else { creds = grpc::InsecureServerCredentials(); } // v2.4 Stage D — attach the bearer-token auth processor when // the listener requires it. The processor validates the // `authorization: Bearer ` metadata against the listener's // configured keys list. gRPC rejects with UNAUTHENTICATED // (no per-handler code needed). if (listener.auth.required) { if (listener.auth.keys.empty()) { throw std::runtime_error( "Listener " + addr + ": auth.required=true but auth.keys is empty"); } // v2.7.0 — a key name is a principal, so it must be unambiguous. // Refuse at startup rather than resolving a policy against a name // that means two different callers. { std::set seen; for (const auto& k : listener.auth.keys) { if (k.name != smartbotic::database::auth::kPrincipalUnnamed && smartbotic::database::auth::isReservedPrincipal(k.name)) { throw std::runtime_error( "Listener " + addr + ": auth key name '" + k.name + "' is reserved. `anonymous` denotes an unauthenticated " "caller and `unnamed` denotes a legacy bare key; a real " "key must not be able to impersonate either in a policy."); } if (!seen.insert(k.name).second && k.name != smartbotic::database::auth::kPrincipalUnnamed) { throw std::runtime_error( "Listener " + addr + ": duplicate auth key name '" + k.name + "'. A principal must identify one caller, " "otherwise a policy written against it is ambiguous."); } } } if (!listener.tls.enabled) { spdlog::warn( "Listener {}: auth.required=true but tls.enabled=false — " "the bearer token is sent in cleartext over the wire. " "Consider enabling TLS for any non-loopback listener.", addr); } std::vector nk; nk.reserve(listener.auth.keys.size()); for (const auto& k : listener.auth.keys) nk.push_back({k.name, k.key}); creds->SetAuthMetadataProcessor( std::make_shared( std::move(nk))); } builder.AddListeningPort(addr, creds); builder.RegisterService(storageImpl_.get()); builder.RegisterService(replicationImpl_.get()); builder.SetMaxReceiveMessageSize( static_cast(config_.grpc.maxReceiveMessageSizeMb * mb)); builder.SetMaxSendMessageSize( static_cast(config_.grpc.maxSendMessageSizeMb * mb)); grpc::ResourceQuota quota("smartbotic-db-" + addr); quota.Resize(config_.grpc.resourceQuotaMemoryMb * mb); builder.SetResourceQuota(quota); auto server = builder.BuildAndStart(); if (!server) { throw std::runtime_error("Failed to start gRPC server on " + addr); } // v2.4 Stage B log line — listener identity printed so operators // know exactly which interface is exposed with which policy. // Stage C/D will extend with TLS + auth status. spdlog::info("gRPC listener up: {} (tls={}, auth_required={})", addr, listener.tls.enabled ? "true" : "false", listener.auth.required ? "true" : "false"); grpcServers_.push_back(std::move(server)); } } void DatabaseService::stopGrpcServer() { for (auto& server : grpcServers_) { if (server) server->Shutdown(); } grpcServers_.clear(); } } // namespace smartbotic::database