trevon/yams
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2
yams/tests/unit/daemon/service_manager_catch2_test.cpp
Trevon 59d11a1504
 fix(daemon): synchronize metrics completion and data-root snapshots (#327) 
Publish owning data-directory snapshots under a mutex and route bootstrap
and fallback publications through the same synchronization boundary.

Keep polling completion locked through its final notification so shutdown
cannot destroy the condition variable while the notifier still uses it.
Exercise both publication windows with deterministic event-based tests and
use the repository thread-compatibility wrapper.

Preserve Failed in status rendering instead of reporting initialization.
Reanchor the existing metrics environment-read entry; add no exceptions.

This does not close the separate queued-collector and embedding-borrower
lifetime audits or establish universal bounded callback shutdown.

Signed-off-by: trvon <git@trevon.dev>
2026-10-22 07:12:22 -06:00

2063 lines
88 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("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