trevon/yams
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1
yams/tests/unit/daemon/service_manager_catch2_test.cpp
Trevon 4f6534f65a
 fix(daemon): mark the vector DB ready once its first vectors commit (#348) 
readiness.vectorDbReady was set only at vector DB init, as "rows > 0".
A corpus that starts empty kept it false after every document was
embedded: status and doctor reported the vector DB not ready until a
restart, and SearchComponent already worked around it by probing the
live row count.

Set it when the embedding service reports committed vectors, once the
vector DB has been initialized; ServiceManager owns the state and
already handles that callback.

The retrieval bench waits up to 30 s for vector_db_ready before its
queries; every run hit the full wait. On grim (gcc Profiling,
search_product_clean_baseline, base is #346):

                         SciFact              NFCorpus
  setup total            56.4 -> 27.7 s       411.7 -> 382.1 s
  searchability ready    54.6 -> 25.3 s       409.8 -> 380.2 s
  guard waits            3/3 -> 0/3           3/3 -> 0/3
  MAP / MRR / nDCG       identical            within base run-to-run range
  latency p50 / p95      41.2/54.0 ->         29.8/59.3 ->
                         47.0/66.7 ms         29.3/59.0 ms

The SciFact queries now start right after ingest instead of after the
idle guard. In-window background zones stay under 0.1 s; the query
path is unchanged, and NFCorpus latency is flat.
2026-10-25 14:58:26 -06:00

2082 lines
89 KiB
C++
Raw Permalink Blame History

// Catch2 migration of service_manager_test.cpp
// Migration: yams-3s4 (daemon unit tests)
// Unit tests for ServiceManager component - construction, initialization, and service access
// pi-lens-ignore: fatal error
#include <catch2/catch_test_macros.hpp>
#include <sqlite3.h>
#include <nlohmann/json.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/post.hpp>
#include <boost/asio/use_future.hpp>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <exception>
#include <filesystem>
#include <fstream>
#include <future>
#include <memory>
#include <mutex>
#include <set>
#include <stdexcept>
#include <system_error>
#include <thread>
#ifdef _WIN32
#include <aclapi.h>
#include <windows.h>
#endif
#include "../../common/test_helpers_catch2.h"
#include "../../../include/yams/daemon/components/db_integrity_stamp.h"
#include "../../../src/daemon/components/service_manager/bootstrap_status.h"
#include <yams/compat/thread_stop_compat.h>
#include <yams/crypto/hasher.h>
#include <yams/daemon/components/ConfigResolver.h>
#include <yams/daemon/components/DaemonLifecycleFsm.h>
#include <yams/daemon/components/db_recovery.h>
#include <yams/daemon/components/InternalEventBus.h>
#include <yams/daemon/components/repair/repair_health_probe.h>
#include <yams/daemon/components/RepairService.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/components/StateComponent.h>
#include <yams/daemon/components/TuneAdvisor.h>
#include <yams/daemon/components/WorkCoordinator.h>
#include <yams/daemon/daemon.h>
#include <yams/daemon/shutdown_budget.h>
#include <yams/memory_sync/memory_sync_service.h>
#include <yams/memory_sync/writer_auth.h>
#include <yams/metadata/database.h>
namespace fs = std::filesystem;
using namespace yams;
using namespace yams::daemon;
namespace yams::daemon::test {
namespace {
TEST_CASE("Bootstrap status snapshot preserves readiness schema and ETA values",
"[daemon][service_manager][status][eta]") {
using service_manager::BootstrapStatusData;
const auto start = std::chrono::steady_clock::time_point{std::chrono::seconds{100}};
const auto now = start + std::chrono::seconds{2};
BootstrapStatusData data;
data.dataDir = "/tmp/yams-data";
data.startTime = start;
data.readiness.ipcServerReady = true;
data.readiness.contentStoreReady = true;
data.readiness.vectorDbInitAttempted = true;
data.readiness.vectorDbReady = true;
data.readiness.vectorDbDim = 768;
data.readiness.searchProgress = 100;
data.readiness.vectorIndexProgress = 50;
data.readiness.modelLoadProgress = 25;
data.freshness = service_manager::BootstrapFreshnessSnapshot{
.simeonLexicalConfigured = true,
.simeonLexicalReady = false,
.simeonLexicalBuilding = true,
.simeonFragmentGeometryReady = false,
};
data.databaseRecoveredAt = "2026-01-02T03:04:05Z";
data.databaseRecoveredFrom = "/tmp/yams.db.corrupt";
data.databasePhase = "migrating";
data.databasePhaseSince = now - std::chrono::milliseconds{750};
data.maintenancePhase = "recovery_cleanup";
data.maintenancePhaseSince = now - std::chrono::milliseconds{250};
data.storageWarning = "slow storage";
data.initDurationsMs = {{"vector_index", 7000}, {"database", 1500}, {"plugins", 500}};
const auto status = service_manager::buildBootstrapStatusSnapshot(data, now);
CHECK_FALSE(status.at("ready").get<bool>());
CHECK(status.at("overall") == "initializing");
const auto& readiness = status.at("readiness");
CHECK(readiness.at("ipc_server") == true);
CHECK(readiness.at("content_store") == true);
CHECK(readiness.at("database") == false);
CHECK(readiness.at("metadata_repo") == false);
CHECK(readiness.at("search_engine") == false);
CHECK(readiness.at("model_provider") == false);
CHECK(readiness.at("vector_index") == false);
CHECK(readiness.at("plugins") == false);
CHECK(readiness.at("vector_db_init_attempted") == true);
CHECK(readiness.at("vector_db_ready") == true);
CHECK(readiness.at("vector_db_dim") == 768);
CHECK(readiness.at("search_engine_lexical_enhancement_configured") == true);
CHECK(readiness.at("search_engine_lexical_enhancement_ready") == false);
CHECK(readiness.at("search_engine_lexical_enhancement_building") == true);
CHECK(readiness.at("search_engine_fragment_geometry_ready") == false);
CHECK(status.at("search_engine_lexical_enhancement_state") == "building");
CHECK(status.at("database_recovered_at") == "2026-01-02T03:04:05Z");
CHECK(status.at("database_recovered_from") == "/tmp/yams.db.corrupt");
CHECK(status.at("database_phase") == "migrating");
CHECK(status.at("database_phase_elapsed_ms") == 750);
CHECK(status.at("maintenance_phase") == "recovery_cleanup");
CHECK(status.at("maintenance_phase_elapsed_ms") == 250);
CHECK(status.at("storage_warning") == "slow storage");
CHECK(status.at("progress").at("search_engine") == 100);
CHECK(status.at("progress").at("vector_index") == 50);
CHECK(status.at("progress").at("model_provider") == 25);
CHECK(status.at("eta_seconds").at("vector_index") == 5);
CHECK(status.at("eta_seconds").at("model_provider") == 18);
CHECK_FALSE(status.at("eta_seconds").contains("content_store"));
CHECK(status.at("durations_ms").at("database") == 1500);
CHECK(status.at("top_slowest").at(0).at("name") == "vector_index");
CHECK(status.at("top_slowest").at(0).at("elapsed_ms") == 7000);
CHECK(status.at("uptime_seconds") == 2);
CHECK(status.at("data_dir") == "/tmp/yams-data");
}
TEST_CASE("Bootstrap status throttles only pre-ready publication",
"[daemon][service_manager][status][throttle]") {
const auto previous = std::chrono::steady_clock::time_point{std::chrono::milliseconds{1000}};
CHECK_FALSE(service_manager::shouldPublishBootstrapStatus(
false, previous, previous + std::chrono::milliseconds{249}));
CHECK(service_manager::shouldPublishBootstrapStatus(false, previous,
previous + std::chrono::milliseconds{250}));
CHECK(service_manager::shouldPublishBootstrapStatus(true, previous,
previous + std::chrono::milliseconds{1}));
CHECK(service_manager::shouldPublishBootstrapStatus(
false, std::chrono::steady_clock::time_point{}, previous));
}
TEST_CASE("Bootstrap status snapshot publishes completed readiness immediately",
"[daemon][service_manager][status][ready]") {
service_manager::BootstrapStatusData data;
data.readiness.ipcServerReady = true;
data.readiness.contentStoreReady = true;
data.readiness.databaseReady = true;
data.readiness.metadataRepoReady = true;
data.readiness.searchEngineReady = true;
data.readiness.modelProviderReady = true;
data.readiness.vectorIndexReady = true;
data.readiness.pluginsReady = true;
data.freshness = service_manager::BootstrapFreshnessSnapshot{};
const auto status = service_manager::buildBootstrapStatusSnapshot(data, data.startTime);
CHECK(status.at("ready") == true);
CHECK(status.at("overall") == "ready");
CHECK(status.at("eta_seconds").empty());
CHECK(status.at("search_engine_lexical_enhancement_state") == "disabled");
}
fs::path metadataDbPath(const DaemonConfig& config) {
return config.dataDir / "yams.db";
}
void seedMetadataDb(const fs::path& dbPath, const std::string& value = "seed") {
yams::metadata::Database db;
REQUIRE(db.open(dbPath.string(), yams::metadata::ConnectionMode::Create));
REQUIRE(
db.execute("CREATE TABLE IF NOT EXISTS startup_probe(id INTEGER PRIMARY KEY, value TEXT)"));
REQUIRE(db.execute("DELETE FROM startup_probe"));
REQUIRE(db.execute("INSERT INTO startup_probe(value) VALUES('" + value + "')"));
REQUIRE(db.execute("PRAGMA wal_checkpoint(TRUNCATE)"));
db.close();
REQUIRE(clearDbCleanShutdownSidecars(dbPath));
}
void writeCorruptDb(const fs::path& dbPath) {
std::ofstream out(dbPath, std::ios::binary | std::ios::trunc);
REQUIRE(out.good());
static constexpr char kSqliteHeader[] = "SQLite format 3";
out.write(kSqliteHeader, static_cast<std::streamsize>(sizeof(kSqliteHeader)));
const std::string garbage(4096, '\xff');
out.write(garbage.data(), static_cast<std::streamsize>(garbage.size()));
out.close();
}
std::optional<fs::path> findQuarantinedFile(const fs::path& dataDir) {
std::error_code ec;
if (!fs::exists(dataDir, ec)) {
return std::nullopt;
}
const std::string prefix = yams::daemon::corruptDbPrefix();
for (fs::directory_iterator it(dataDir, ec), end; it != end; it.increment(ec)) {
if (ec) {
ec.clear();
continue;
}
const auto name = it->path().filename().string();
if (name.starts_with(prefix) && !name.ends_with("-wal") && !name.ends_with("-shm")) {
return it->path();
}
}
return std::nullopt;
}
std::size_t startupProbeRowCount(const fs::path& dbPath) {
yams::metadata::Database db;
REQUIRE(db.open(dbPath.string(), yams::metadata::ConnectionMode::ReadWrite));
auto stmtR = db.prepare("SELECT COUNT(*) FROM startup_probe");
REQUIRE(stmtR);
auto& stmt = stmtR.value();
REQUIRE(stmt.step());
auto count = static_cast<std::size_t>(stmt.getInt(0));
db.close();
return count;
}
void seedDummyWalSidecar(const fs::path& dbPath, const std::string& payload = "stale-wal") {
std::ofstream out(dbPath.string() + "-wal", std::ios::binary | std::ios::trunc);
REQUIRE(out.good());
out.write(payload.data(), static_cast<std::streamsize>(payload.size()));
out.close();
REQUIRE(fs::exists(dbPath.string() + "-wal"));
}
bool walSidecarCleared(const fs::path& dbPath) {
std::error_code ec;
const fs::path walPath(dbPath.string() + "-wal");
if (!fs::exists(walPath, ec)) {
return true;
}
return fs::file_size(walPath, ec) == 0;
}
bool walSidecarStillMatchesPayload(const fs::path& dbPath, const std::string& payload) {
const fs::path walPath(dbPath.string() + "-wal");
std::error_code ec;
if (!fs::exists(walPath, ec) || fs::file_size(walPath, ec) != payload.size()) {
return false;
}
std::ifstream in(walPath, std::ios::binary);
std::string actual(payload.size(), '\0');
in.read(actual.data(), static_cast<std::streamsize>(actual.size()));
return in.good() && actual == payload;
}
class SessionWatchIndexingService final : public app::services::IIndexingService {
public:
Result<app::services::AddDirectoryResponse>
addDirectory(const app::services::AddDirectoryRequest& request) override {
requests.push_back(request);
if (failuresRemaining > 0) {
--failuresRemaining;
return Error{ErrorCode::IOError, "simulated indexing failure"};
}
app::services::AddDirectoryResponse response;
response.filesProcessed = request.includePatterns.size();
for (const auto& relative : request.includePatterns) {
app::services::IndexedFileResult result;
result.path = (fs::path(request.directoryPath) / relative).string();
result.success = !failingFiles.contains(relative);
if (result.success) {
++response.filesIndexed;
} else {
++response.filesFailed;
result.error = "simulated per-file failure";
}
response.results.push_back(std::move(result));
}
return response;
}
std::size_t failuresRemaining{0};
std::set<std::string> failingFiles;
std::vector<app::services::AddDirectoryRequest> requests;
};
class SessionWatchDocumentService final : public app::services::IDocumentService {
public:
Result<app::services::StoreDocumentResponse>
store(const app::services::StoreDocumentRequest&) override {
return ErrorCode::NotImplemented;
}
Result<app::services::RetrieveDocumentResponse>
retrieve(const app::services::RetrieveDocumentRequest&) override {
return ErrorCode::NotImplemented;
}
Result<app::services::CatDocumentResponse>
cat(const app::services::CatDocumentRequest&) override {
return ErrorCode::NotImplemented;
}
Result<app::services::ListDocumentsResponse>
list(const app::services::ListDocumentsRequest&) override {
return ErrorCode::NotImplemented;
}
Result<app::services::UpdateMetadataResponse>
updateMetadata(const app::services::UpdateMetadataRequest&) override {
return ErrorCode::NotImplemented;
}
Result<std::string> resolveNameToHash(const std::string&, bool) override {
return ErrorCode::NotImplemented;
}
Result<app::services::DeleteByNameResponse>
deleteByName(const app::services::DeleteByNameRequest& request) override {
deleteRequests.push_back(request);
if (failuresRemaining > 0) {
--failuresRemaining;
return Error{ErrorCode::IOError, "simulated deletion failure"};
}
app::services::DeleteByNameResponse response;
response.count = 1;
response.deleted.push_back(
{.name = request.name, .deleted = true, .errorCode = ErrorCode::Success});
return response;
}
Result<app::services::DeleteByNameResponse>
pruneVersions(const app::services::PruneVersionsRequest&) override {
return ErrorCode::NotImplemented;
}
std::size_t failuresRemaining{0};
std::vector<app::services::DeleteByNameRequest> deleteRequests;
};
void requireReadyDatabaseState(const StateComponent& state) {
CHECK(state.readiness.databaseReady.load());
std::lock_guard<std::mutex> lk(state.readiness.recoveryMutex);
CHECK((state.readiness.databasePhase == "ready"));
}
struct PluginHostLifecycleCounters {
std::atomic<std::size_t> listCalls{0};
std::atomic<std::size_t> unloadCalls{0};
};
class CountingAbiPluginHost final : public AbiPluginHost {
public:
explicit CountingAbiPluginHost(std::shared_ptr<PluginHostLifecycleCounters> counters)
: counters_(counters) {}
std::vector<PluginDescriptor> listLoaded() const override {
counters_->listCalls.fetch_add(1, std::memory_order_relaxed);
if (counters_->unloadCalls.load(std::memory_order_relaxed) == 0) {
return {{.name = "counted_plugin"}};
}
return {};
}
Result<void> unload(const std::string&) override {
counters_->unloadCalls.fetch_add(1, std::memory_order_relaxed);
return {};
}
private:
std::shared_ptr<PluginHostLifecycleCounters> counters_;
};
fs::path writeProtectedWriterManifest(const fs::path& directory, std::string_view nodeId,
std::string_view corpusId, std::uint64_t corpusEpoch) {
auto keys = memory_sync::generateWriterKeyPair();
REQUIRE(keys.has_value());
const auto privateKey = directory / "writer-private.pem";
const auto publicKey = directory / "writer-public.pem";
REQUIRE(
ServiceManager::__test_writeProtectedP2pPrivateKey(privateKey, keys.value().privateKeyPem)
.has_value());
REQUIRE(ServiceManager::__test_writeProtectedP2pPrivateKey(publicKey, keys.value().publicKeyPem)
.has_value());
const auto manifest = directory / "writers.json";
const nlohmann::json content{
{"schema_version", 1},
{"corpus_id", corpusId},
{"corpus_epoch", corpusEpoch},
{"local_key",
{{"writer_id", nodeId},
{"key_id", "local-v1"},
{"private_key_path", privateKey.string()}}},
{"trusted_writers", nlohmann::json::array({{{"writer_id", nodeId},
{"key_id", "local-v1"},
{"public_key_path", publicKey.string()}}})}};
REQUIRE(
ServiceManager::__test_writeProtectedP2pPrivateKey(manifest, content.dump()).has_value());
return manifest;
}
void grantUntrustedKeyAccess(const fs::path& keyPath, bool writable) {
#ifdef _WIN32
BYTE worldSid[SECURITY_MAX_SID_SIZE];
DWORD worldSidBytes = sizeof(worldSid);
REQUIRE(CreateWellKnownSid(WinWorldSid, nullptr, worldSid, &worldSidBytes));
PACL currentAcl = nullptr;
PSECURITY_DESCRIPTOR descriptor = nullptr;
REQUIRE(GetNamedSecurityInfoW(const_cast<wchar_t*>(keyPath.c_str()), SE_FILE_OBJECT,
DACL_SECURITY_INFORMATION, nullptr, nullptr, &currentAcl, nullptr,
&descriptor) == ERROR_SUCCESS);
EXPLICIT_ACCESSW access{};
access.grfAccessPermissions = writable ? GENERIC_WRITE : GENERIC_READ;
access.grfAccessMode = GRANT_ACCESS;
access.grfInheritance = NO_INHERITANCE;
BuildTrusteeWithSidW(&access.Trustee, worldSid);
PACL broadenedAcl = nullptr;
REQUIRE(SetEntriesInAclW(1, &access, currentAcl, &broadenedAcl) == ERROR_SUCCESS);
REQUIRE(SetNamedSecurityInfoW(const_cast<wchar_t*>(keyPath.c_str()), SE_FILE_OBJECT,
DACL_SECURITY_INFORMATION | PROTECTED_DACL_SECURITY_INFORMATION,
nullptr, nullptr, broadenedAcl, nullptr) == ERROR_SUCCESS);
LocalFree(broadenedAcl);
LocalFree(descriptor);
#else
fs::permissions(keyPath, writable ? fs::perms::group_write : fs::perms::group_read,
fs::perm_options::add);
#endif
}
} // namespace
// Test fixture for ServiceManager tests
struct ServiceManagerFixture {
yams::test::ScopedEnvVar disableWatcher_{"YAMS_DISABLE_SESSION_WATCHER", std::string("1")};
DaemonConfig config_;
StateComponent state_;
DaemonLifecycleFsm lifecycleFsm_;
fs::path testDir_;
ServiceManagerFixture() {
// Create isolated test directory
testDir_ =
fs::temp_directory_path() /
("sm_test_" + std::to_string(std::hash<std::thread::id>{}(std::this_thread::get_id())) +
"_" + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()));
fs::create_directories(testDir_);
// Setup basic config
config_.dataDir = testDir_ / "data";
config_.socketPath = testDir_ / "daemon.sock";
config_.pidFile = testDir_ / "daemon.pid";
config_.logFile = testDir_ / "daemon.log";
fs::create_directories(config_.dataDir);
}
~ServiceManagerFixture() {
// Cleanup test directory
if (fs::exists(testDir_)) {
std::error_code ec;
fs::remove_all(testDir_, ec);
}
}
};
struct ScopedCurrentPath {
fs::path original_;
explicit ScopedCurrentPath(const fs::path& target) : original_(fs::current_path()) {
fs::current_path(target);
}
~ScopedCurrentPath() {
std::error_code ec;
fs::current_path(original_, ec);
}
};
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager data-directory snapshots survive publication",
"[daemon][service_manager][data-dir-snapshot]") {
yams::test::SpdlogLevelGuard logGuard;
spdlog::set_level(spdlog::level::warn);
ServiceManager manager(config_, state_, lifecycleFsm_);
// Holding a const reference must extend the lifetime of an owning snapshot,
// not expose the mutable path that early metrics and storage fallback publish.
const auto& beforePublication = manager.getResolvedDataDir();
REQUIRE(beforePublication.empty());
const auto primary = testDir_ / "primary-data";
manager.testingSetResolvedDataDir(primary);
CHECK(beforePublication.empty());
const auto& primarySnapshot = manager.getResolvedDataDir();
REQUIRE(primarySnapshot == primary);
const auto fallback = testDir_ / "fallback-data";
manager.testingSetResolvedDataDir(fallback);
CHECK(primarySnapshot == primary);
CHECK(manager.getResolvedDataDir() == fallback);
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager snapshots remain whole during publication",
"[daemon][service_manager][data-dir-snapshot][concurrency]") {
yams::test::SpdlogLevelGuard logGuard;
spdlog::set_level(spdlog::level::warn);
ServiceManager manager(config_, state_, lifecycleFsm_);
// These are path values only: no filesystem entry with a long component is created.
const auto primary = testDir_ / std::string(512, 'a');
const auto fallback = testDir_ / std::string(512, 'b');
manager.testingSetResolvedDataDir(primary);
const auto& initialSnapshot = manager.getResolvedDataDir();
std::promise<void> begin;
const auto ready = begin.get_future().share();
yams::compat::jthread writer([&] {
ready.wait();
for (unsigned i = 0; i < 256; ++i) {
manager.testingSetResolvedDataDir(i % 2 == 0 ? primary : fallback);
}
});
begin.set_value();
unsigned invalidSnapshots = 0;
for (unsigned i = 0; i < 256; ++i) {
const auto snapshot = manager.getResolvedDataDir();
if (snapshot != primary && snapshot != fallback) {
++invalidSnapshots;
}
}
writer.join();
CHECK(invalidSnapshots == 0);
CHECK(initialSnapshot == primary);
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager construction succeeds",
"[daemon][service_manager]") {
REQUIRE_NOTHROW(ServiceManager(config_, state_, lifecycleFsm_));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager marks the vector DB ready once its first vectors commit",
"[daemon][components][service-manager][catch2][vector-readiness]") {
ServiceManager manager(config_, state_, lifecycleFsm_);
state_.readiness.vectorDbReady = false;
// Before the vector DB was opened, a commit callback proves nothing about it.
state_.readiness.vectorDbInitAttempted = false;
manager.testingMarkVectorDbReadyAfterCommit({"hash-a"});
CHECK_FALSE(state_.readiness.vectorDbReady.load());
// An initialized DB that started empty becomes ready on the first non-empty commit.
state_.readiness.vectorDbInitAttempted = true;
manager.testingMarkVectorDbReadyAfterCommit({});
CHECK_FALSE(state_.readiness.vectorDbReady.load());
manager.testingMarkVectorDbReadyAfterCommit({"hash-a", "hash-b"});
CHECK(state_.readiness.vectorDbReady.load());
}
TEST_CASE("Async initialization completion wait survives caller timeout",
"[daemon][service_manager][async-completion]") {
AsyncInitOrchestrator init;
REQUIRE(init.tryStart());
const auto token = init.getStopToken();
// A caller may time out before co_spawn installs its future.
CHECK_FALSE(init.waitForCompletion(std::chrono::milliseconds(0)));
CHECK_FALSE(token.stop_requested());
std::promise<void> completion;
init.setFuture(completion.get_future());
CHECK_FALSE(init.waitForCompletion(std::chrono::milliseconds(0)));
CHECK_FALSE(token.stop_requested());
// The future remains owned by the orchestrator, not the timed-out caller.
completion.set_value();
CHECK(init.waitForCompletion(std::chrono::milliseconds(0)));
CHECK_FALSE(token.stop_requested());
CHECK(init.requestStopAndWait(std::chrono::milliseconds(0)));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager getName returns correct component name",
"[daemon][service_manager]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
REQUIRE((std::string(sm.getName()) == "ServiceManager"));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager service accessors after construction",
"[daemon][service_manager]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
// PBI-057: Vector DB initialization is deferred to async phase
REQUIRE((sm.getVectorDatabase() == nullptr));
// Other services are initialized during initialize(), not in constructor
REQUIRE((sm.getContentStore() == nullptr));
REQUIRE((sm.getMetadataRepo() == nullptr));
REQUIRE((sm.getModelProvider() == nullptr));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager multiple construction is idempotent",
"[daemon][service_manager]") {
// Create two instances sequentially - should not throw
ServiceManager sm1(config_, state_, lifecycleFsm_);
ServiceManager sm2(config_, state_, lifecycleFsm_);
SUCCEED();
}
TEST_CASE_METHOD(ServiceManagerFixture,
"Concurrent embedded ServiceManager preserves active tuning lifecycle",
"[daemon][service_manager][config][tuning][lifecycle]") {
yams::test::ScopedEnvVar timeout{"YAMS_IPC_TIMEOUT_MS", std::nullopt};
TuneAdvisor::setIpcTimeoutMs(4321);
config_.tuning.tuneAdvisorOverridesResolved = true;
ServiceManager owner(config_, state_, lifecycleFsm_);
REQUIRE((owner.getRuntimeTuningStatus().at("ipc.timeout_ms") == "4321"));
auto directConfig = config_;
directConfig.tuning.tuneAdvisorOverridesResolved = false;
ServiceManager embedded(directConfig, state_, lifecycleFsm_);
CHECK((TuneAdvisor::ipcTimeoutMs() == 4321u));
CHECK((owner.getRuntimeTuningStatus().at("ipc.timeout_ms") == "4321"));
CHECK((embedded.getRuntimeTuningStatus().at("ipc.timeout_ms") == "4321"));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager construction with missing data directory",
"[daemon][service_manager]") {
fs::remove_all(config_.dataDir);
REQUIRE_NOTHROW(ServiceManager(config_, state_, lifecycleFsm_));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager initialize reports data directory creation failures",
"[daemon][service_manager][startup][error]") {
fs::remove_all(config_.dataDir);
{
std::ofstream blocker(config_.dataDir);
REQUIRE(blocker.good());
blocker << "not a directory";
}
ServiceManager sm(config_, state_, lifecycleFsm_);
const auto result = sm.initialize();
REQUIRE_FALSE(result);
CHECK((result.error().code == ErrorCode::IOError));
CHECK((result.error().message.find("Failed to create data directory") != std::string::npos));
CHECK((result.error().message.find(config_.dataDir.string()) != std::string::npos));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"Task record publication and reads validate immutable revision identity",
"[daemon][service_manager][task-record]") {
storage::BackendConfig backendConfig;
backendConfig.localPath = config_.dataDir / "task-record-store";
auto backend = std::make_unique<storage::FilesystemBackend>();
REQUIRE(backend->initialize(backendConfig).has_value());
ServiceManager manager(config_, state_, lifecycleFsm_);
manager.testingSetMemorySyncService(std::make_unique<memory_sync::MemorySyncService>(
std::move(backend), memory_sync::MemorySyncConfig{"A", 50}));
auto* sync = manager.testingMemorySyncService();
const std::string taskId = "123e4567-e89b-42d3-a456-426614174000";
nlohmann::json record = {
{"schema", "yams.task-record/v1"},
{"task_id", taskId},
{"kind", "claim"},
{"owner", "agent-a"},
{"source_uri", "repo:src/search"},
{"source_revision", "git:9883a438"},
{"recorded_at_ms", 2000},
{"source_timestamp_ms", 1000},
{"supersedes", nlohmann::json::array()},
{"ownership_semantics", "record_only"},
{"body", "Investigating retrieval"},
};
const auto value = record.dump();
const auto hash = crypto::SHA256Hasher::hash(std::as_bytes(std::span(value)));
const auto key = "task-record/" + taskId + "/" + hash;
const auto published = manager.publishMemorySync(key, value);
CAPTURE(backendConfig.localPath.string(), key.size());
INFO((published ? "task record published" : published.error().message));
REQUIRE(published.has_value());
REQUIRE(sync->syncOnce().has_value());
const auto hydrated = manager.readMemorySyncCached(key);
REQUIRE(hydrated.has_value());
CHECK(hydrated.value() == value);
record["body"] = "changed revision";
CHECK_FALSE(manager.publishMemorySync(key, record.dump()).has_value());
record["ownership_semantics"] = "acquired";
const auto acquired = record.dump();
const auto acquiredHash = crypto::SHA256Hasher::hash(std::as_bytes(std::span(acquired)));
CHECK_FALSE(manager.publishMemorySync("task-record/" + taskId + "/" + acquiredHash, acquired)
.has_value());
// A remote/lower-level writer cannot bypass the typed read contract.
const auto namespaced = memory_sync::userLogicalKey(key);
REQUIRE(namespaced.has_value());
REQUIRE(sync->publish(namespaced.value(), std::as_bytes(std::span(acquired))).has_value());
REQUIRE(sync->syncOnce().has_value());
CHECK_FALSE(manager.readMemorySyncCached(key).has_value());
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager rejects replication of a personal corpus",
"[daemon][service_manager][sharing][security][config]") {
config_.memorySync.enabled = true;
config_.memorySync.nodeId = "123e4567-e89b-42d3-a456-42661417400a";
config_.memorySync.corpusId = "personal-corpus";
config_.memorySync.corpusEpoch = 1;
config_.memorySync.path = (config_.dataDir / "forbidden-sync-store").string();
SECTION("direct") {
config_.memorySync.transport = "direct";
}
SECTION("shared store") {
config_.memorySync.transport = "shared-store";
}
ServiceManager manager(config_, state_, lifecycleFsm_);
REQUIRE(manager.initialize().has_value());
boost::asio::io_context io;
compat::stop_source stopSource;
auto future = boost::asio::co_spawn(
io, manager.initializeAsyncAwaitable(stopSource.get_token()), boost::asio::use_future);
io.run();
const auto initialized = future.get();
REQUIRE_FALSE(initialized.has_value());
CHECK(initialized.error().message.find("corpus_scope=shared") != std::string::npos);
CHECK_FALSE(fs::exists(config_.memorySync.path));
CHECK_FALSE(fs::exists(config_.dataDir / "p2p" / "op-store"));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager rejects direct P2P without usable writer authentication",
"[daemon][service_manager][p2p][security][config]") {
config_.memorySync.enabled = true;
config_.memorySync.corpusScope = yams::memory_sync::CorpusScope::Shared;
config_.memorySync.transport = "direct";
config_.memorySync.nodeId = "123e4567-e89b-42d3-a456-42661417400a";
config_.memorySync.corpusId = "missing-writer-auth";
config_.memorySync.corpusEpoch = 1;
ServiceManager manager(config_, state_, lifecycleFsm_);
REQUIRE(manager.initialize().has_value());
boost::asio::io_context io;
compat::stop_source stopSource;
auto future = boost::asio::co_spawn(
io, manager.initializeAsyncAwaitable(stopSource.get_token()), boost::asio::use_future);
io.run();
const auto initialized = future.get();
REQUIRE_FALSE(initialized.has_value());
CHECK(initialized.error().code == ErrorCode::InvalidArgument);
CHECK(initialized.error().message.find("writer_auth_required=true") != std::string::npos);
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager rejects provenance-unknown direct operation stores",
"[daemon][service_manager][p2p][security][migration]") {
const std::string nodeId = "123e4567-e89b-42d3-a456-42661417400a";
const std::string corpusId = "legacy-direct-store";
const auto manifest = writeProtectedWriterManifest(testDir_, nodeId, corpusId, 2);
const auto legacyObject = config_.dataDir / "p2p" / "op-store" / "legacy" / "record.json";
fs::create_directories(legacyObject.parent_path());
std::ofstream(legacyObject) << "unsigned-history";
config_.memorySync.enabled = true;
config_.memorySync.corpusScope = yams::memory_sync::CorpusScope::Shared;
config_.memorySync.transport = "direct";
config_.memorySync.nodeId = nodeId;
config_.memorySync.corpusId = corpusId;
config_.memorySync.corpusEpoch = 2;
config_.memorySync.writerAuthRequired = true;
config_.memorySync.writerAuthManifestPath = manifest.string();
ServiceManager manager(config_, state_, lifecycleFsm_);
REQUIRE(manager.initialize().has_value());
boost::asio::io_context io;
compat::stop_source stopSource;
auto future = boost::asio::co_spawn(
io, manager.initializeAsyncAwaitable(stopSource.get_token()), boost::asio::use_future);
io.run();
const auto initialized = future.get();
REQUIRE_FALSE(initialized.has_value());
CHECK(initialized.error().code == ErrorCode::InvalidState);
CHECK(initialized.error().message.find("provenance-unknown") != std::string::npos);
CHECK(fs::exists(legacyObject));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager rejects symbolic-link P2P identity keys",
"[daemon][service_manager][p2p][security]") {
#ifdef _WIN32
SKIP("symbolic-link permission semantics differ on Windows");
#else
auto generated = memory_sync::generateWriterKeyPair();
REQUIRE(generated.has_value());
const auto victim = testDir_ / "victim.pem";
{
std::ofstream output(victim, std::ios::binary | std::ios::trunc);
REQUIRE(output.good());
output << generated.value().privateKeyPem;
}
fs::permissions(victim, fs::perms::owner_read | fs::perms::owner_write,
fs::perm_options::replace);
const auto publicKey = testDir_ / "identity.pub.pem";
{
std::ofstream output(publicKey, std::ios::binary | std::ios::trunc);
REQUIRE(output.good());
output << generated.value().publicKeyPem;
}
// Trust files must not be group/other-writable; set it explicitly so the result does not
// depend on the process umask (002 on Debian-style user-private-group systems).
constexpr auto kTrustFilePerms = fs::perms::owner_read | fs::perms::owner_write |
fs::perms::group_read | fs::perms::others_read;
fs::permissions(publicKey, kTrustFilePerms, fs::perm_options::replace);
const auto manifest = testDir_ / "writers.json";
{
nlohmann::json content{
{"schema_version", 1},
{"corpus_id", "identity-security-test"},
{"corpus_epoch", 1},
{"local_key",
{{"writer_id", "123e4567-e89b-42d3-a456-42661417400a"},
{"key_id", "local-v1"},
{"private_key_path", victim.string()}}},
{"trusted_writers",
nlohmann::json::array({{{"writer_id", "123e4567-e89b-42d3-a456-42661417400a"},
{"key_id", "local-v1"},
{"public_key_path", publicKey.string()}}})}};
std::ofstream output(manifest, std::ios::binary | std::ios::trunc);
REQUIRE(output.good());
output << content.dump();
}
fs::permissions(manifest, kTrustFilePerms, fs::perm_options::replace);
const auto identity = testDir_ / "identity.pem";
fs::create_symlink(victim, identity);
config_.memorySync.enabled = true;
config_.memorySync.corpusScope = yams::memory_sync::CorpusScope::Shared;
config_.memorySync.transport = "direct";
config_.memorySync.nodeId = "123e4567-e89b-42d3-a456-42661417400a";
config_.memorySync.corpusId = "identity-security-test";
config_.memorySync.corpusEpoch = 1;
config_.memorySync.identityKeyPath = identity.string();
config_.memorySync.writerAuthRequired = true;
config_.memorySync.writerAuthManifestPath = manifest.string();
ServiceManager manager(config_, state_, lifecycleFsm_);
REQUIRE(manager.initialize().has_value());
boost::asio::io_context io;
compat::stop_source stopSource;
auto future = boost::asio::co_spawn(
io, manager.initializeAsyncAwaitable(stopSource.get_token()), boost::asio::use_future);
io.run();
const auto initialized = future.get();
REQUIRE_FALSE(initialized.has_value());
CHECK(initialized.error().code == ErrorCode::Unauthorized);
CHECK(initialized.error().message.find("symbolic link") != std::string::npos);
CHECK(fs::is_symlink(fs::symlink_status(identity)));
#endif
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager rejects broadly accessible direct writer private keys",
"[daemon][service_manager][p2p][security][platform]") {
const std::string nodeId = "123e4567-e89b-42d3-a456-42661417400a";
const std::string corpusId = "writer-key-security";
const auto manifest = writeProtectedWriterManifest(testDir_, nodeId, corpusId, 1);
const auto identity = testDir_ / "separate-identity.pem";
REQUIRE(ServiceManager::__test_loadOrCreateP2pPrivateKey(identity).has_value());
grantUntrustedKeyAccess(testDir_ / "writer-private.pem", false);
config_.memorySync.enabled = true;
config_.memorySync.corpusScope = yams::memory_sync::CorpusScope::Shared;
config_.memorySync.transport = "direct";
config_.memorySync.nodeId = nodeId;
config_.memorySync.corpusId = corpusId;
config_.memorySync.corpusEpoch = 1;
config_.memorySync.identityKeyPath = identity.string();
config_.memorySync.writerAuthRequired = true;
config_.memorySync.writerAuthManifestPath = manifest.string();
ServiceManager manager(config_, state_, lifecycleFsm_);
REQUIRE(manager.initialize().has_value());
boost::asio::io_context io;
compat::stop_source stopSource;
auto future = boost::asio::co_spawn(
io, manager.initializeAsyncAwaitable(stopSource.get_token()), boost::asio::use_future);
io.run();
const auto initialized = future.get();
REQUIRE_FALSE(initialized.has_value());
CHECK(initialized.error().code == ErrorCode::Unauthorized);
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager rejects writable direct writer trust files",
"[daemon][service_manager][p2p][security][platform]") {
const std::string nodeId = "123e4567-e89b-42d3-a456-42661417400a";
const std::string corpusId = "writer-trust-security";
const auto manifest = writeProtectedWriterManifest(testDir_, nodeId, corpusId, 1);
const auto identity = testDir_ / "trust-test-identity.pem";
REQUIRE(ServiceManager::__test_loadOrCreateP2pPrivateKey(identity).has_value());
SECTION("writable manifest") {
grantUntrustedKeyAccess(manifest, true);
}
SECTION("writable public key") {
grantUntrustedKeyAccess(testDir_ / "writer-public.pem", true);
}
config_.memorySync.enabled = true;
config_.memorySync.corpusScope = yams::memory_sync::CorpusScope::Shared;
config_.memorySync.transport = "direct";
config_.memorySync.nodeId = nodeId;
config_.memorySync.corpusId = corpusId;
config_.memorySync.corpusEpoch = 1;
config_.memorySync.identityKeyPath = identity.string();
config_.memorySync.writerAuthRequired = true;
config_.memorySync.writerAuthManifestPath = manifest.string();
ServiceManager manager(config_, state_, lifecycleFsm_);
REQUIRE(manager.initialize().has_value());
boost::asio::io_context io;
compat::stop_source stopSource;
auto future = boost::asio::co_spawn(
io, manager.initializeAsyncAwaitable(stopSource.get_token()), boost::asio::use_future);
io.run();
const auto initialized = future.get();
REQUIRE_FALSE(initialized.has_value());
CHECK(initialized.error().code == ErrorCode::Unauthorized);
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager creates owner-only P2P identity keys and rejects broad access",
"[daemon][service_manager][p2p][security][platform]") {
const auto identity = testDir_ / "generated-identity.pem";
auto created = ServiceManager::__test_loadOrCreateP2pPrivateKey(identity);
REQUIRE(created.has_value());
REQUIRE_FALSE(created.value().empty());
REQUIRE(ServiceManager::__test_loadOrCreateP2pPrivateKey(identity).has_value());
grantUntrustedKeyAccess(identity, false);
auto rejected = ServiceManager::__test_loadOrCreateP2pPrivateKey(identity);
REQUIRE_FALSE(rejected.has_value());
CHECK(rejected.error().code == ErrorCode::Unauthorized);
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager rejects inherited Windows P2P identity ACLs",
"[daemon][service_manager][p2p][security][platform]") {
#ifndef _WIN32
SKIP("Windows ACL-specific coverage");
#else
const auto identity = testDir_ / "inherited-identity.pem";
REQUIRE(ServiceManager::__test_loadOrCreateP2pPrivateKey(identity).has_value());
PACL currentAcl = nullptr;
PSECURITY_DESCRIPTOR descriptor = nullptr;
REQUIRE(GetNamedSecurityInfoW(const_cast<wchar_t*>(identity.c_str()), SE_FILE_OBJECT,
DACL_SECURITY_INFORMATION, nullptr, nullptr, &currentAcl, nullptr,
&descriptor) == ERROR_SUCCESS);
REQUIRE(SetNamedSecurityInfoW(const_cast<wchar_t*>(identity.c_str()), SE_FILE_OBJECT,
DACL_SECURITY_INFORMATION | UNPROTECTED_DACL_SECURITY_INFORMATION,
nullptr, nullptr, currentAcl, nullptr) == ERROR_SUCCESS);
LocalFree(descriptor);
auto rejected = ServiceManager::__test_loadOrCreateP2pPrivateKey(identity);
REQUIRE_FALSE(rejected.has_value());
CHECK(rejected.error().code == ErrorCode::Unauthorized);
#endif
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager pool configuration failures propagate through Result",
"[daemon][service_manager][startup][error]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
const auto result = sm.__test_initializeWithPoolConfiguration(
[] { throw std::runtime_error("forced pool configuration failure"); });
REQUIRE_FALSE(result);
CHECK((result.error().code == ErrorCode::InternalError));
CHECK((result.error().message.find("forced pool configuration failure") != std::string::npos));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager destructor handles cleanup",
"[daemon][service_manager]") {
auto sm = std::make_unique<ServiceManager>(config_, state_, lifecycleFsm_);
REQUIRE_NOTHROW(sm.reset());
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager delegates shared plugin host shutdown once",
"[daemon][service_manager][plugin][ownership][shutdown]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
auto counters = std::make_shared<PluginHostLifecycleCounters>();
sm.__test_setAbiHost(std::make_unique<CountingAbiPluginHost>(counters));
sm.shutdown();
CHECK((counters->listCalls.load(std::memory_order_relaxed) == 1));
CHECK((counters->unloadCalls.load(std::memory_order_relaxed) == 1));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager getConfig returns configuration",
"[daemon][service_manager]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
const auto& cfg = sm.getConfig();
REQUIRE((cfg.dataDir == config_.dataDir));
REQUIRE((cfg.socketPath == config_.socketPath));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager snapshots search rebuild compatibility policy once",
"[daemon][service_manager][config][search]") {
yams::test::ScopedEnvVar disableRebuilds{"YAMS_DISABLE_SEARCH_REBUILDS", std::string{"1"}};
ServiceManager sm(config_, state_, lifecycleFsm_);
const auto& policy = sm.getConfig().searchMaintenance;
REQUIRE(policy.automaticRebuildsEnabled.has_value());
CHECK_FALSE(*policy.automaticRebuildsEnabled);
CHECK((policy.automaticRebuildsSource ==
"compatibility-environment:YAMS_DISABLE_SEARCH_REBUILDS"));
REQUIRE((sm.getSearchComponent() != nullptr));
CHECK_FALSE(sm.getSearchComponent()->getConfig().automaticRebuildsEnabled);
disableRebuilds.set("0");
CHECK_FALSE(*sm.getConfig().searchMaintenance.automaticRebuildsEnabled);
}
TEST_CASE_METHOD(ServiceManagerFixture,
"Typed search rebuild policy outranks compatibility environment",
"[daemon][service_manager][config][search]") {
yams::test::ScopedEnvVar disableRebuilds{"YAMS_DISABLE_SEARCH_REBUILDS", std::string{"1"}};
config_.searchMaintenance.automaticRebuildsEnabled = true;
config_.searchMaintenance.automaticRebuildsSource = "typed:test";
ServiceManager sm(config_, state_, lifecycleFsm_);
const auto& policy = sm.getConfig().searchMaintenance;
REQUIRE(policy.automaticRebuildsEnabled.has_value());
CHECK(*policy.automaticRebuildsEnabled);
CHECK((policy.automaticRebuildsSource == "typed:test"));
REQUIRE((sm.getSearchComponent() != nullptr));
CHECK(sm.getSearchComponent()->getConfig().automaticRebuildsEnabled);
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager snapshots runtime tuning provenance across live refresh",
"[daemon][service_manager][config][tuning][snapshot]") {
yams::test::ScopedEnvVar timeout{"YAMS_IPC_TIMEOUT_MS", std::string{"4321"}};
ServiceManager sm(config_, state_, lifecycleFsm_);
const auto snapshot = sm.getTuningConfig();
REQUIRE((sm.getRuntimeTuningStatus().at("ipc.timeout_ms.source") ==
"compatibility-environment:YAMS_IPC_TIMEOUT_MS"));
timeout.unset();
sm.setTuningConfig(snapshot);
CHECK((sm.getRuntimeTuningStatus().at("ipc.timeout_ms.source") ==
"compatibility-environment:YAMS_IPC_TIMEOUT_MS"));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"Live tuning refresh reports preserved construction-time capacity",
"[daemon][service_manager][config][tuning][physical-state]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
config_.tuning.postIngestCapacity = 111;
config_.tuning.provenance["tuning.post_ingest_capacity"] = "config:tuning.post_ingest_capacity";
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
auto beforeInitialization = sm->getTuningConfig();
beforeInitialization.postIngestCapacity = 222;
beforeInitialization.provenance["tuning.post_ingest_capacity"] =
"config:tuning.post_ingest_capacity";
sm->setTuningConfig(beforeInitialization);
REQUIRE(sm->initialize());
sm->startAsyncInit();
const auto ready = sm->waitForServiceManagerTerminalState(30);
REQUIRE((ready.state == ServiceManagerState::Ready));
REQUIRE(sm->getPostIngestQueue() != nullptr);
CHECK((sm->getPostIngestQueue()->capacity() == 222u));
TuningConfig refreshed;
refreshed.tuneAdvisorOverridesResolved = true;
sm->setTuningConfig(refreshed);
const auto effective = sm->getTuningConfig();
CHECK((effective.postIngestCapacity == 222u));
CHECK((effective.provenance.at("tuning.post_ingest_capacity") ==
"runtime:construction-time-post-ingest-channel"));
const auto status = sm->getRuntimeTuningStatus();
CHECK((status.at("post_ingest.capacity") == "222"));
CHECK((status.at("post_ingest.capacity.source") ==
"runtime:construction-time-post-ingest-channel"));
sm->shutdown();
}
TEST_CASE_METHOD(ServiceManagerFixture,
"Runtime tuning status publishes values from one configured generation",
"[daemon][service_manager][config][tuning][snapshot]") {
config_.tuning.tuneAdvisorOverridesResolved = true;
TuneAdvisor::setIpcTimeoutMs(1001);
TuneAdvisor::setMemoryWarningThreshold(0.61);
ServiceManager sm(config_, state_, lifecycleFsm_);
std::atomic<bool> stop{false};
std::thread writer([&] {
bool alternate = false;
while (!stop.load(std::memory_order_acquire)) {
[[maybe_unused]] auto update = TuneAdvisor::beginConfiguredOverrideUpdate();
TuneAdvisor::setIpcTimeoutMs(alternate ? 1001 : 2002);
TuneAdvisor::setMemoryWarningThreshold(alternate ? 0.61 : 0.82);
alternate = !alternate;
}
});
for (int i = 0; i < 500; ++i) {
sm.setTuningConfig(sm.getTuningConfig());
const auto status = sm.getRuntimeTuningStatus();
const auto timeout = status.at("ipc.timeout_ms");
const auto warning = status.at("resource.memory_warning_threshold");
CHECK(((timeout == "1001" && warning == "0.610000") ||
(timeout == "2002" && warning == "0.820000")));
}
stop.store(true, std::memory_order_release);
writer.join();
}
TEST_CASE_METHOD(ServiceManagerFixture,
"Direct embedded ServiceManager starts a fresh tuning lifecycle",
"[daemon][service_manager][config][tuning][lifecycle]") {
yams::test::ScopedEnvVar timeout{"YAMS_IPC_TIMEOUT_MS", std::nullopt};
TuneAdvisor::setIpcTimeoutMs(4321);
REQUIRE_FALSE(config_.tuning.tuneAdvisorOverridesResolved);
ServiceManager sm(config_, state_, lifecycleFsm_);
CHECK(sm.getTuningConfig().tuneAdvisorOverridesResolved);
CHECK((sm.getRuntimeTuningStatus().at("ipc.timeout_ms") == "15000"));
}
TEST_CASE("ServiceManager session watcher honors only the test disable gate",
"[daemon][service_manager][session_watch]") {
CHECK(ServiceManager::__test_shouldStartSessionWatcher(""));
CHECK(ServiceManager::__test_shouldStartSessionWatcher("0"));
CHECK_FALSE(ServiceManager::__test_shouldStartSessionWatcher("true"));
}
TEST_CASE("ServiceManager session watcher uses configured intervals without busy polling",
"[daemon][service_manager][session_watch]") {
using namespace std::chrono_literals;
CHECK((ServiceManager::__test_sessionWatcherDelay(false, 0) == 2s));
CHECK((ServiceManager::__test_sessionWatcherDelay(true, 250) == 250ms));
CHECK((ServiceManager::__test_sessionWatcherDelay(true, 20) == 100ms));
}
TEST_CASE("ServiceManager session watcher refreshes every retrieval index",
"[daemon][service_manager][session_watch]") {
auto request = ServiceManager::__test_makeSessionWatchRequest("coding", "/workspace/yams",
{"src/search/search_engine.cpp"});
CHECK((request.directoryPath == "/workspace/yams"));
CHECK((request.includePatterns == std::vector<std::string>{"src/search/search_engine.cpp"}));
CHECK(request.recursive);
CHECK((request.sessionId == "coding"));
CHECK_FALSE(request.noEmbeddings);
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager session watcher retries failed indexing and deletion",
"[daemon][service_manager][session_watch]") {
ServiceManager serviceManager(config_, state_, lifecycleFsm_);
SessionWatchIndexingService indexingService;
SessionWatchDocumentService documentService;
const auto watchedDirectory = testDir_ / "watched";
const auto watchedFile = watchedDirectory / "changed.cpp";
fs::create_directories(watchedDirectory);
{
std::ofstream output(watchedFile);
REQUIRE(output.good());
output << "int changed = 1;\n";
}
indexingService.failuresRemaining = 1;
CHECK_FALSE(serviceManager.__test_scanSessionWatchDirectory(indexingService, &documentService,
"coding", watchedDirectory));
REQUIRE((indexingService.requests.size() == 1));
CHECK((indexingService.requests.front().includePatterns ==
std::vector<std::string>{"changed.cpp"}));
CHECK(serviceManager.__test_scanSessionWatchDirectory(indexingService, &documentService,
"coding", watchedDirectory));
CHECK((indexingService.requests.size() == 2));
CHECK(serviceManager.__test_scanSessionWatchDirectory(indexingService, &documentService,
"coding", watchedDirectory));
CHECK((indexingService.requests.size() == 2));
REQUIRE(fs::remove(watchedFile));
documentService.failuresRemaining = 1;
CHECK_FALSE(serviceManager.__test_scanSessionWatchDirectory(indexingService, &documentService,
"coding", watchedDirectory));
REQUIRE((documentService.deleteRequests.size() == 1));
CHECK((documentService.deleteRequests.front().name == watchedFile.string()));
CHECK(serviceManager.__test_scanSessionWatchDirectory(indexingService, &documentService,
"coding", watchedDirectory));
CHECK((documentService.deleteRequests.size() == 2));
CHECK(serviceManager.__test_scanSessionWatchDirectory(indexingService, &documentService,
"coding", watchedDirectory));
CHECK((documentService.deleteRequests.size() == 2));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager session watcher keeps progressing past a failing path",
"[daemon][service_manager][session_watch]") {
ServiceManager serviceManager(config_, state_, lifecycleFsm_);
SessionWatchIndexingService indexingService;
SessionWatchDocumentService documentService;
const auto watchedDirectory = testDir_ / "watched_progress";
fs::create_directories(watchedDirectory);
const auto writeFile = [&](const fs::path& path, std::string_view text) {
std::ofstream output(path);
REQUIRE(output.good());
output << text;
};
writeFile(watchedDirectory / "kept.cpp", "int kept = 1;\n");
writeFile(watchedDirectory / "gone.cpp", "int gone = 1;\n");
REQUIRE(serviceManager.__test_scanSessionWatchDirectory(indexingService, &documentService,
"coding", watchedDirectory));
REQUIRE((indexingService.requests.size() == 1));
SECTION("a removal that keeps failing does not re-index unrelated files") {
REQUIRE(fs::remove(watchedDirectory / "gone.cpp"));
documentService.failuresRemaining = 1000;
CHECK_FALSE(serviceManager.__test_scanSessionWatchDirectory(
indexingService, &documentService, "coding", watchedDirectory));
REQUIRE_FALSE(documentService.deleteRequests.empty());
// The path no longer exists, so every document stored at it is stale.
CHECK(documentService.deleteRequests.back().force);
writeFile(watchedDirectory / "added.cpp", "int added = 1;\n");
CHECK_FALSE(serviceManager.__test_scanSessionWatchDirectory(
indexingService, &documentService, "coding", watchedDirectory));
REQUIRE((indexingService.requests.size() == 2));
CHECK((indexingService.requests.back().includePatterns ==
std::vector<std::string>{"added.cpp"}));
// Before the fix the stale snapshot made added.cpp "changed" on every pass.
CHECK_FALSE(serviceManager.__test_scanSessionWatchDirectory(
indexingService, &documentService, "coding", watchedDirectory));
CHECK((indexingService.requests.size() == 2));
documentService.failuresRemaining = 0;
const auto deletesBefore = documentService.deleteRequests.size();
CHECK(serviceManager.__test_scanSessionWatchDirectory(indexingService, &documentService,
"coding", watchedDirectory));
CHECK((documentService.deleteRequests.size() == deletesBefore + 1));
CHECK(serviceManager.__test_scanSessionWatchDirectory(indexingService, &documentService,
"coding", watchedDirectory));
CHECK((documentService.deleteRequests.size() == deletesBefore + 1));
}
SECTION("only the file that failed to index is retried") {
writeFile(watchedDirectory / "good.cpp", "int good = 1;\n");
writeFile(watchedDirectory / "bad.cpp", "int bad = 1;\n");
indexingService.failingFiles = {"bad.cpp"};
CHECK_FALSE(serviceManager.__test_scanSessionWatchDirectory(
indexingService, &documentService, "coding", watchedDirectory));
REQUIRE((indexingService.requests.size() == 2));
indexingService.failingFiles.clear();
CHECK(serviceManager.__test_scanSessionWatchDirectory(indexingService, &documentService,
"coding", watchedDirectory));
REQUIRE((indexingService.requests.size() == 3));
CHECK((indexingService.requests.back().includePatterns ==
std::vector<std::string>{"bad.cpp"}));
}
}
TEST_CASE("ServiceManager topology readiness follows artifact freshness",
"[daemon][service_manager][topology]") {
yams::search::IndexFreshnessSnapshot snapshot;
snapshot.topologyStatusKnown = true;
snapshot.topologyEpoch = 42;
snapshot.topologyArtifactsFresh = true;
CHECK(ServiceManager::isTopologyRoutingReady(snapshot));
snapshot.topologyEpoch = 0;
CHECK_FALSE(ServiceManager::isTopologyRoutingReady(snapshot));
snapshot.topologyEpoch = 42;
snapshot.topologyArtifactsFresh = false;
CHECK_FALSE(ServiceManager::isTopologyRoutingReady(snapshot));
snapshot.topologyArtifactsFresh = true;
snapshot.topologyRebuildRunning = true;
CHECK_FALSE(ServiceManager::isTopologyRoutingReady(snapshot));
snapshot.topologyRebuildRunning = false;
snapshot.postIngestQueued = 1;
CHECK_FALSE(ServiceManager::isTopologyRoutingReady(snapshot));
snapshot.postIngestQueued = 0;
snapshot.topologyDirtyDocuments = 1;
CHECK_FALSE(ServiceManager::isTopologyRoutingReady(snapshot));
snapshot.topologyDirtyDocuments = 0;
snapshot.awaitingDrain = true;
snapshot.lexicalReady = false;
snapshot.kgReady = false;
CHECK(ServiceManager::isTopologyRoutingReady(snapshot));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager PostIngestQueue accessor before init",
"[daemon][service_manager]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
// May be null before initialization
auto piq = sm.getPostIngestQueue();
(void)piq;
SUCCEED();
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager tuning config getter doesn't crash",
"[daemon][service_manager]") {
config_.tuning.ingestStoreBatchSize = 19;
ServiceManager sm(config_, state_, lifecycleFsm_);
const auto& tuning = sm.getTuningConfig();
CHECK((tuning.ingestStoreBatchSize == 19));
CHECK((sm.getIngestStoreBatchSize() == 19));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager set tuning config doesn't crash",
"[daemon][service_manager]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
TuningConfig tc;
tc.postIngestCapacity = 1000;
tc.postIngestThreadsMin = 2;
tc.postIngestThreadsMax = 4;
tc.ingestStoreBatchSize = 23;
REQUIRE_NOTHROW(sm.setTuningConfig(tc));
CHECK((sm.getIngestStoreBatchSize() == 23));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager getWorkerQueueDepth doesn't crash",
"[daemon][service_manager]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
auto depth = sm.getWorkerQueueDepth();
(void)depth;
SUCCEED();
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager enqueuePostIngest doesn't crash",
"[daemon][service_manager]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
REQUIRE_NOTHROW(sm.enqueuePostIngest("test_hash", "text/plain"));
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager search engine snapshot doesn't crash",
"[daemon][service_manager]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
auto snapshot = sm.getSearchEngineFsmSnapshot();
(void)snapshot.buildReason;
(void)snapshot.vectorEnabled;
SUCCEED();
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager cached search engine access doesn't crash",
"[daemon][service_manager]") {
ServiceManager sm(config_, state_, lifecycleFsm_);
auto engine = sm.getCachedSearchEngine();
(void)engine; // May be null, that's OK
SUCCEED();
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager memory cleanup verification",
"[daemon][service_manager][.slow]") {
// Note: This test rapidly creates/destroys ServiceManagers which may be flaky
// on some platforms due to timing-sensitive cleanup
for (int i = 0; i < 3; ++i) {
auto sm = std::make_unique<ServiceManager>(config_, state_, lifecycleFsm_);
sm.reset();
}
SUCCEED();
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager restart creates fresh io_context (PBI-066-38)",
"[daemon][service_manager]") {
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
// First init
auto r1 = sm->initialize();
REQUIRE(r1);
// Stop services
sm->shutdown();
// Second init should succeed and not throw bad executor
auto r2 = sm->initialize();
REQUIRE(r2);
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager initializes WAL under configured data directory",
"[daemon][service_manager][wal]") {
const auto runRoot = testDir_ / "cwd";
fs::create_directories(runRoot);
ScopedCurrentPath cwdGuard(runRoot);
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
auto init = sm->initialize();
REQUIRE(init);
const auto expectedWalDir = config_.dataDir / "wal";
REQUIRE(fs::exists(expectedWalDir));
REQUIRE(fs::is_directory(expectedWalDir));
const auto misplacedWalDir = runRoot / "wal";
CHECK_FALSE(fs::exists(misplacedWalDir));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager async init starts only once per initialize cycle",
"[daemon][service_manager][startup][async_init]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
auto init = sm->initialize();
REQUIRE(init);
std::promise<void> firstBarrierPromise;
auto firstBarrierFuture = firstBarrierPromise.get_future();
std::atomic<bool> firstBarrierSet{false};
sm->startAsyncInit(&firstBarrierPromise, &firstBarrierSet);
REQUIRE((firstBarrierFuture.wait_for(std::chrono::seconds(2)) == std::future_status::ready));
CHECK(firstBarrierSet.load(std::memory_order_acquire));
std::promise<void> secondBarrierPromise;
auto secondBarrierFuture = secondBarrierPromise.get_future();
std::atomic<bool> secondBarrierSet{false};
sm->startAsyncInit(&secondBarrierPromise, &secondBarrierSet);
CHECK((secondBarrierFuture.wait_for(std::chrono::milliseconds(200)) ==
std::future_status::timeout));
CHECK_FALSE(secondBarrierSet.load(std::memory_order_acquire));
const auto snapshot = sm->waitForServiceManagerTerminalState(30);
REQUIRE((snapshot.state == ServiceManagerState::Ready));
requireReadyDatabaseState(state_);
sm->shutdown();
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager consumes clean shutdown proof and republishes it on shutdown",
"[daemon][service_manager][startup][integrity_stamp]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
const auto dbPath = metadataDbPath(config_);
seedMetadataDb(dbPath, "clean_restart");
const auto initialStamp = publishDbCleanShutdownStamp(dbPath);
const std::string initialStampInfo =
initialStamp ? "clean stamp published" : initialStamp.error().message;
INFO(initialStampInfo);
REQUIRE(initialStamp);
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
REQUIRE(sm->initialize());
sm->startAsyncInit();
const auto ready = sm->waitForServiceManagerTerminalState(30);
REQUIRE((ready.state == ServiceManagerState::Ready));
CHECK(state_.readiness.databaseIntegrityFastPath.load(std::memory_order_acquire));
sm->shutdown();
const auto nextStartup = consumeDbCleanShutdownStamp(dbPath);
CHECK(nextStartup.trustedCleanShutdown);
CHECK(nextStartup.invalidationPersisted);
}
namespace {
std::atomic<std::int64_t> gInFlightQueryDeadlineNs{0};
void keepRunningUntilDeadline(sqlite3_context* context, int, sqlite3_value**) {
const auto nowNs = std::chrono::duration_cast<std::chrono::nanoseconds>(
std::chrono::steady_clock::now().time_since_epoch())
.count();
sqlite3_result_int(context, nowNs < gInFlightQueryDeadlineNs.load() ? 1 : 0);
}
} // namespace
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager shutdown interrupts an in-flight metadata query and keeps the "
"clean-shutdown stamp",
"[daemon][service_manager][shutdown][integrity_stamp][cancellation]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
config_.enableAutoRepair = false;
const auto dbPath = metadataDbPath(config_);
seedMetadataDb(dbPath, "inflight_handler");
REQUIRE(publishDbCleanShutdownStamp(dbPath));
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
REQUIRE(sm->initialize());
sm->startAsyncInit();
const auto ready = sm->waitForServiceManagerTerminalState(30);
REQUIRE((ready.state == ServiceManagerState::Ready));
auto readPool = sm->getReadConnectionPool();
REQUIRE((readPool != nullptr));
// A request handler pinned on a WorkCoordinator worker inside one long SQLite statement, the
// shape seen when a `yams list` over a large corpus was in flight during `daemon restart`.
// The statement would run well past both Phase 5 join windows unless shutdown interrupts it.
constexpr auto kStatementBudget = std::chrono::seconds(20);
gInFlightQueryDeadlineNs.store(
std::chrono::duration_cast<std::chrono::nanoseconds>(
(std::chrono::steady_clock::now() + kStatementBudget).time_since_epoch())
.count());
std::promise<void> started;
auto startedFuture = started.get_future();
std::promise<bool> finished;
auto finishedFuture = finished.get_future();
boost::asio::post(sm->getWorkerExecutor(), [readPool, &started, &finished]() {
WorkCoordinator::JobScope job("test.inflight_metadata_query");
auto conn = readPool->acquire(std::chrono::seconds(5));
if (!conn) {
started.set_value();
finished.set_value(false);
return;
}
auto& db = **conn.value();
sqlite3_create_function(db.rawHandle(), "yams_test_keep_running", 0, SQLITE_UTF8, nullptr,
keepRunningUntilDeadline, nullptr, nullptr);
started.set_value();
const auto result = db.execute("WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x + 1 "
"FROM c WHERE yams_test_keep_running()) SELECT count(*) "
"FROM c");
finished.set_value(!result.has_value());
});
REQUIRE((startedFuture.wait_for(std::chrono::seconds(5)) == std::future_status::ready));
// Let the statement enter sqlite3_step() so shutdown interrupts a running query.
std::this_thread::sleep_for(std::chrono::milliseconds(100));
const auto shutdownStart = std::chrono::steady_clock::now();
sm->shutdown();
const auto shutdownElapsed = std::chrono::steady_clock::now() - shutdownStart;
REQUIRE((finishedFuture.wait_for(std::chrono::seconds(0)) == std::future_status::ready));
CHECK(finishedFuture.get());
CHECK((shutdownElapsed < shutdown_budget::kWorkCoordinatorJoinTimeout));
CHECK((sm->getServiceManagerFsmSnapshot().state == ServiceManagerState::Stopped));
const auto nextStartup = consumeDbCleanShutdownStamp(dbPath);
CHECK(nextStartup.trustedCleanShutdown);
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager preserves metadata when integrity validation cannot finish",
"[daemon][service_manager][startup][integrity][cancellation]") {
const auto dbPath = metadataDbPath(config_);
seedMetadataDb(dbPath, "must-survive");
auto database = std::make_shared<metadata::Database>();
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
SECTION("shutdown cancels validation without recovery") {
REQUIRE(database->open(dbPath.string(), metadata::ConnectionMode::ReadWrite));
sm->shutdown();
}
SECTION("an unavailable connection is not evidence of corruption") {
// The on-disk database is valid; the supplied connection is closed.
}
CHECK_FALSE(sm->testingEnsureDatabaseIntegrityOrRecover(dbPath, database));
database->close();
CHECK_FALSE(findQuarantinedFile(config_.dataDir).has_value());
CHECK(state_.readiness.databaseRecoveredFrom.empty());
CHECK(startupProbeRowCount(dbPath) == 1);
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager records recovery provenance for corrupt metadata DB startup",
"[daemon][service_manager][startup][recovery]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
const auto dbPath = metadataDbPath(config_);
writeCorruptDb(dbPath);
seedDummyWalSidecar(dbPath, "corrupt-sidecar");
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
auto init = sm->initialize();
REQUIRE(init);
sm->startAsyncInit();
const auto smSnap = sm->waitForServiceManagerTerminalState(30);
REQUIRE((smSnap.state == ServiceManagerState::Ready));
requireReadyDatabaseState(state_);
std::optional<fs::path> quarantinedPath;
{
std::lock_guard<std::mutex> lk(state_.readiness.recoveryMutex);
REQUIRE_FALSE(state_.readiness.databaseRecoveredFrom.empty());
quarantinedPath = findQuarantinedFile(config_.dataDir);
REQUIRE(quarantinedPath.has_value());
CHECK((state_.readiness.databaseRecoveredFrom == quarantinedPath->string()));
}
CHECK(fs::exists(*quarantinedPath));
CHECK_FALSE(walSidecarStillMatchesPayload(dbPath, "corrupt-sidecar"));
sm->shutdown();
CHECK(walSidecarCleared(dbPath));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager startup salvage keeps the corrupt DB for retention",
"[daemon][service_manager][startup][recovery][salvage]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
// A readable earlier metadata DB with more documents than the (not yet created) live DB.
const auto corruptPath = config_.dataDir / "yams.db.corrupt-20200101T000000Z";
{
metadata::Database db;
REQUIRE(db.open(corruptPath.string(), metadata::ConnectionMode::Create));
REQUIRE(db.execute("CREATE TABLE documents (id INTEGER PRIMARY KEY AUTOINCREMENT, "
"file_path TEXT NOT NULL, file_name TEXT NOT NULL, file_extension "
"TEXT, file_size INTEGER NOT NULL, sha256_hash TEXT UNIQUE NOT NULL, "
"mime_type TEXT, created_time INTEGER, modified_time INTEGER, "
"indexed_time INTEGER, content_extracted BOOLEAN DEFAULT 0, "
"extraction_status TEXT DEFAULT 'pending', extraction_error TEXT, "
"path_prefix TEXT, reverse_path TEXT, path_hash TEXT, parent_hash "
"TEXT, path_depth INTEGER DEFAULT 0, repair_status TEXT DEFAULT "
"'pending', repair_attempted_at INTEGER, repair_attempts INTEGER "
"DEFAULT 0)"));
for (int i = 0; i < 2; ++i) {
const auto n = std::to_string(i);
REQUIRE(db.execute("INSERT INTO documents (file_path, file_name, file_size, "
"sha256_hash) VALUES ('/salvage/" +
n + ".txt', '" + n + ".txt', 1, '" + std::string(63, 'a') + n +
"')"));
}
REQUIRE(db.execute("PRAGMA wal_checkpoint(TRUNCATE)"));
db.close();
}
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
REQUIRE(sm->initialize());
sm->startAsyncInit();
const auto smSnap = sm->waitForServiceManagerTerminalState(30);
REQUIRE((smSnap.state == ServiceManagerState::Ready));
requireReadyDatabaseState(state_);
CHECK(state_.readiness.databaseSalvaged.load(std::memory_order_acquire));
// Startup salvages but never deletes: removal is the repair service's job, after the
// retention window and only with evidence that every document reached the live DB.
CHECK(fs::exists(corruptPath));
sm->shutdown();
}
TEST_CASE_METHOD(ServiceManagerFixture, "ServiceManager clears stale metadata WAL during startup",
"[daemon][service_manager][wal][startup]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
const auto dbPath = metadataDbPath(config_);
seedMetadataDb(dbPath, "stale_wal_seed");
REQUIRE(publishDbCleanShutdownStamp(dbPath));
const std::string staleWalPayload = "stale-wal";
seedDummyWalSidecar(dbPath, staleWalPayload);
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
auto init = sm->initialize();
REQUIRE(init);
sm->startAsyncInit();
auto smSnap = sm->waitForServiceManagerTerminalState(30);
REQUIRE((smSnap.state == ServiceManagerState::Ready));
requireReadyDatabaseState(state_);
CHECK_FALSE(state_.readiness.databaseIntegrityFastPath.load(std::memory_order_acquire));
{
std::lock_guard<std::mutex> lk(state_.readiness.recoveryMutex);
CHECK(state_.readiness.databaseRecoveredFrom.empty());
}
CHECK_FALSE(walSidecarStillMatchesPayload(dbPath, staleWalPayload));
CHECK((startupProbeRowCount(dbPath) == 1U));
sm->shutdown();
CHECK(walSidecarCleared(dbPath));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager shutdown remains idempotent after async init reaches ready",
"[daemon][service_manager][shutdown][idempotent]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
auto init = sm->initialize();
REQUIRE(init);
sm->startAsyncInit();
const auto readySnap = sm->waitForServiceManagerTerminalState(30);
REQUIRE((readySnap.state == ServiceManagerState::Ready));
sm->shutdown();
const auto firstShutdown = sm->getServiceManagerFsmSnapshot();
CHECK((firstShutdown.state == ServiceManagerState::Stopped));
CHECK((sm->__test_getAbiHost() == nullptr));
REQUIRE_NOTHROW(sm->shutdown());
const auto secondShutdown = sm->getServiceManagerFsmSnapshot();
CHECK((secondShutdown.state == ServiceManagerState::Stopped));
CHECK((sm->getMetadataRepo() == nullptr));
CHECK((sm->getContentStore() == nullptr));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager shutdown waits for an in-flight worker task",
"[daemon][service_manager][shutdown][regression]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
config_.enableAutoRepair = false;
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
auto init = sm->initialize();
REQUIRE(init);
sm->startAsyncInit();
const auto readySnap = sm->waitForServiceManagerTerminalState(30);
REQUIRE((readySnap.state == ServiceManagerState::Ready));
struct SharedShutdownState {
std::mutex gateMutex;
std::condition_variable gateCv;
bool allowWorkerExit{false};
std::promise<void> workerEntered;
std::promise<void> workerFinished;
std::promise<void> shutdownFinished;
std::atomic<unsigned> completionSequence{0};
std::atomic<unsigned> workerCompletionOrder{0};
std::atomic<unsigned> shutdownCompletionOrder{0};
};
struct ShutdownCleanup {
explicit ShutdownCleanup(const std::shared_ptr<SharedShutdownState>& sharedState)
: state(sharedState) {}
void releaseWorker() noexcept {
{
std::lock_guard<std::mutex> lk(state->gateMutex);
state->allowWorkerExit = true;
}
state->gateCv.notify_all();
}
void join() noexcept {
if (shutdownThread.joinable()) {
shutdownThread.join();
}
}
~ShutdownCleanup() {
releaseWorker();
join();
}
std::shared_ptr<SharedShutdownState> state;
std::thread shutdownThread;
};
auto sharedState = std::make_shared<SharedShutdownState>();
auto workerEntered = sharedState->workerEntered.get_future();
auto workerFinished = sharedState->workerFinished.get_future();
auto shutdownFinished = sharedState->shutdownFinished.get_future();
ShutdownCleanup cleanup(sharedState);
boost::asio::post(sm->getWorkerExecutor(), [sharedState]() {
try {
sharedState->workerEntered.set_value();
std::unique_lock<std::mutex> lk(sharedState->gateMutex);
sharedState->gateCv.wait(lk, [sharedState] { return sharedState->allowWorkerExit; });
lk.unlock();
const auto order =
sharedState->completionSequence.fetch_add(1, std::memory_order_acq_rel) + 1;
sharedState->workerCompletionOrder.store(order, std::memory_order_release);
sharedState->workerFinished.set_value();
} catch (...) {
const auto error = std::current_exception();
try {
sharedState->workerEntered.set_exception(error);
} catch (...) {
}
try {
sharedState->workerFinished.set_exception(error);
} catch (...) {
}
}
});
REQUIRE((workerEntered.wait_for(std::chrono::seconds(5)) == std::future_status::ready));
REQUIRE_NOTHROW(workerEntered.get());
cleanup.shutdownThread = std::thread([sm, sharedState]() {
try {
sm->shutdown();
const auto order =
sharedState->completionSequence.fetch_add(1, std::memory_order_acq_rel) + 1;
sharedState->shutdownCompletionOrder.store(order, std::memory_order_release);
sharedState->shutdownFinished.set_value();
} catch (...) {
try {
sharedState->shutdownFinished.set_exception(std::current_exception());
} catch (...) {
}
}
});
// ServiceManagerFsm has no wait API for this non-terminal state. Poll the synchronized
// public snapshot with a deadline so the assertion observes shutdown entry, not elapsed time.
REQUIRE(yams::test::wait_for_condition(
std::chrono::seconds(5), std::chrono::milliseconds(5), [sm]() {
return sm->getServiceManagerFsmSnapshot().state == ServiceManagerState::ShuttingDown;
}));
CHECK((workerFinished.wait_for(std::chrono::seconds(0)) == std::future_status::timeout));
CHECK((shutdownFinished.wait_for(std::chrono::seconds(0)) == std::future_status::timeout));
cleanup.releaseWorker();
REQUIRE((workerFinished.wait_for(std::chrono::seconds(5)) == std::future_status::ready));
REQUIRE_NOTHROW(workerFinished.get());
REQUIRE((shutdownFinished.wait_for(std::chrono::seconds(30)) == std::future_status::ready));
REQUIRE_NOTHROW(shutdownFinished.get());
cleanup.join();
CHECK((sharedState->workerCompletionOrder.load(std::memory_order_acquire) <
sharedState->shutdownCompletionOrder.load(std::memory_order_acquire)));
CHECK((sm->getServiceManagerFsmSnapshot().state == ServiceManagerState::Stopped));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager shutdown remains bounded during auto-repair graph health probe",
"[daemon][service_manager][shutdown][auto_repair][regression]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
config_.enableAutoRepair = false;
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
auto init = sm->initialize();
REQUIRE(init);
sm->startAsyncInit();
const auto readySnap = sm->waitForServiceManagerTerminalState(30);
REQUIRE((readySnap.state == ServiceManagerState::Ready));
auto writePool = sm->getWriteConnectionPool();
REQUIRE((writePool != nullptr));
std::vector<std::unique_ptr<metadata::PooledConnection>> heldConnections;
heldConnections.reserve(32);
for (std::size_t i = 0; i < 128; ++i) {
auto conn = writePool->acquire(std::chrono::milliseconds(0));
if (!conn) {
break;
}
heldConnections.emplace_back();
std::swap(heldConnections.back(), conn.value());
}
REQUIRE_FALSE(heldConnections.empty());
REQUIRE_FALSE(writePool->acquire(std::chrono::milliseconds(0)).has_value());
std::atomic<bool> probeStarted{false};
std::atomic<bool> probeFinished{false};
std::atomic<bool> shutdownReturned{false};
boost::asio::post(sm->getWorkerExecutor(), [sm, &probeStarted, &probeFinished]() {
probeStarted.store(true, std::memory_order_release);
repair::RepairHealthProbe probe(sm->getMetadataRepo(), sm->getVectorDatabase(),
sm->getKgStore());
repair::RepairHealthOptions opts;
opts.checkFts5 = false;
opts.checkEmbeddings = false;
opts.checkGraph = true;
opts.scanDocuments = false;
(void)probe.probe(opts);
probeFinished.store(true, std::memory_order_release);
});
REQUIRE(yams::test::wait_for_condition(
std::chrono::seconds(5), std::chrono::milliseconds(5),
[&]() { return probeStarted.load(std::memory_order_acquire); }));
std::thread shutdownThread([&]() {
sm->shutdown();
shutdownReturned.store(true, std::memory_order_release);
});
REQUIRE(yams::test::wait_for_condition(
std::chrono::seconds(5), std::chrono::milliseconds(5), [&]() {
return shutdownReturned.load(std::memory_order_acquire) &&
probeFinished.load(std::memory_order_acquire);
}));
shutdownThread.join();
CHECK(shutdownReturned.load(std::memory_order_acquire));
CHECK(probeFinished.load(std::memory_order_acquire));
CHECK((sm->getServiceManagerFsmSnapshot().state == ServiceManagerState::Stopped));
heldConnections.clear();
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager shutdown cancels salvage auto-repair with full embed queue",
"[daemon][service_manager][shutdown][auto_repair][cancellation][regression]") {
yams::test::ScopedEnvVar disableVectors("YAMS_DISABLE_VECTORS",
std::optional<std::string>{"1"});
yams::test::ScopedEnvVar disableVectorDb("YAMS_DISABLE_VECTOR_DB",
std::optional<std::string>{"1"});
yams::test::ScopedEnvVar autoInitialDelay("YAMS_REPAIR_AUTO_INITIAL_DELAY_MIN",
std::optional<std::string>{"0"});
yams::test::ScopedEnvVar autoFastInterval("YAMS_REPAIR_AUTO_FAST_MIN",
std::optional<std::string>{"1"});
yams::test::ScopedEnvVar autoWarmInterval("YAMS_REPAIR_AUTO_WARM_HOURS",
std::optional<std::string>{"0"});
yams::test::ScopedEnvVar autoColdInterval("YAMS_REPAIR_AUTO_COLD_HOURS",
std::optional<std::string>{"0"});
yams::test::ScopedEnvVar embedChannelCapacity("YAMS_EMBED_CHANNEL_CAPACITY",
std::optional<std::string>{"256"});
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
config_.enableAutoRepair = true;
config_.autoRepairBatchSize = 1;
auto sm = std::make_shared<ServiceManager>(config_, state_, lifecycleFsm_);
auto init = sm->initialize();
REQUIRE(init);
sm->startAsyncInit();
const auto readySnap = sm->waitForServiceManagerTerminalState(30);
REQUIRE((readySnap.state == ServiceManagerState::Ready));
auto meta = sm->getMetadataRepo();
REQUIRE((meta != nullptr));
const auto now =
std::chrono::time_point_cast<std::chrono::seconds>(std::chrono::system_clock::now());
metadata::DocumentInfo doc{};
doc.fileName = "salvaged-auto-repair.txt";
doc.filePath = (config_.dataDir / doc.fileName).string();
doc.fileExtension = "txt";
doc.fileSize = 26;
doc.sha256Hash = std::string(64, 'a');
doc.mimeType = "text/plain";
doc.modifiedTime = now;
doc.indexedTime = now;
auto docId = meta->insertDocument(doc);
REQUIRE(docId.has_value());
metadata::DocumentContent content{};
content.documentId = docId.value();
content.contentText = "salvage auto repair content";
content.contentLength = static_cast<int64_t>(content.contentText.size());
content.extractionMethod = "test";
content.language = "en";
REQUIRE(meta->insertContent(content).has_value());
REQUIRE(meta->updateDocumentExtractionStatus(docId.value(), true,
metadata::ExtractionStatus::Success)
.has_value());
{
std::lock_guard<std::mutex> lk(state_.readiness.recoveryMutex);
state_.readiness.databaseRecoveredFrom = "salvaged-auto-repair";
}
auto embedChannel =
InternalEventBus::instance().get_or_create_channel<InternalEventBus::EmbedJob>("embed_jobs",
256);
REQUIRE((embedChannel != nullptr));
InternalEventBus::EmbedJob drained;
while (embedChannel->try_pop(drained)) {
}
InternalEventBus::EmbedJob filler;
filler.hashes = {std::string(64, 'f')};
filler.batchSize = 1;
filler.skipExisting = true;
std::size_t filled = 0;
while (embedChannel->try_push(filler)) {
++filled;
}
REQUIRE((filled > 0));
sm->startRepairService([]() -> size_t { return 0; });
auto repairService = sm->getRepairServiceShared();
REQUIRE((repairService != nullptr));
sm->startBackgroundTasks();
REQUIRE(yams::test::wait_for_condition(std::chrono::seconds(5), std::chrono::milliseconds(10),
[&]() { return repairService->isRepairInProgress(); }));
std::atomic<bool> shutdownReturned{false};
const auto shutdownStart = std::chrono::steady_clock::now();
std::thread shutdownThread([&]() {
sm->shutdown();
shutdownReturned.store(true, std::memory_order_release);
});
REQUIRE(yams::test::wait_for_condition(
std::chrono::seconds(3), std::chrono::milliseconds(10),
[&]() { return shutdownReturned.load(std::memory_order_acquire); }));
const auto shutdownElapsed = std::chrono::duration_cast<std::chrono::milliseconds>(
std::chrono::steady_clock::now() - shutdownStart);
shutdownThread.join();
CHECK((shutdownElapsed < std::chrono::seconds(3)));
CHECK((sm->getServiceManagerFsmSnapshot().state == ServiceManagerState::Stopped));
CHECK_FALSE(state_.stats.repairInProgress.load(std::memory_order_acquire));
}
TEST_CASE_METHOD(ServiceManagerFixture,
"ServiceManager restart cycles recover freshly seeded stale WAL",
"[daemon][service_manager][wal][restart]") {
config_.enableModelProvider = false;
config_.useMockModelProvider = false;
config_.autoLoadPlugins = false;
const auto dbPath = metadataDbPath(config_);
seedMetadataDb(dbPath, "restart_cycle_seed");
for (int cycle = 0; cycle < 3; ++cycle) {
StateComponent cycleState;
DaemonLifecycleFsm cycleLifecycle;
const std::string staleWalPayload = "wal-cycle-" + std::to_string(cycle);
seedDummyWalSidecar(dbPath, staleWalPayload);
auto sm = std::make_shared<ServiceManager>(config_, cycleState, cycleLifecycle);
auto init = sm->initialize();
REQUIRE(init);
sm->startAsyncInit();
auto smSnap = sm->waitForServiceManagerTerminalState(30);
REQUIRE((smSnap.state == ServiceManagerState::Ready));
INFO("cycle=" << cycle);
requireReadyDatabaseState(cycleState);
CHECK_FALSE(walSidecarStillMatchesPayload(dbPath, staleWalPayload));
CHECK((startupProbeRowCount(dbPath) == 1U));
sm->shutdown();
CHECK(walSidecarCleared(dbPath));
}
}
TEST_CASE("ServiceManager auto-vacuum requires material reclaimable pages or file tail",
"[daemon][service_manager][vacuum]") {
constexpr std::uint64_t kPageSize = 4096;
constexpr std::uint64_t kLargeDatabaseBytes = 13ULL * 1024 * 1024 * 1024;
CHECK_FALSE(
ServiceManager::__test_shouldAutoVacuum(kLargeDatabaseBytes, 3'405'384, 182, kPageSize));
CHECK_FALSE(
ServiceManager::__test_shouldAutoVacuum(256ULL * 1024 * 1024, 65'536, 20'000, kPageSize));
CHECK(ServiceManager::__test_shouldAutoVacuum(2ULL * 1024 * 1024 * 1024, 524'288, 131'072,
kPageSize));
CHECK(ServiceManager::__test_shouldAutoVacuum(600ULL * 1024 * 1024, 8, 0, kPageSize));
}
} // namespace yams::daemon::test
Reference in a new issue View git blame Copy permalink