yams/tests/unit/daemon/daemon_catch2_test.cpp
Trevon 0dd3985db4
refactor: simplify runtime paths and harden retrieval (#61)
* refactor(tuning): prune dormant runtime knobs

Remove unused tuning controls and adaptive search budget state. Extract testable ONNX slot reconciliation and search telemetry hydration seams, and correct the Apple Silicon sysctl value width.

* refactor(vector): retire dormant backends

Remove unused FAISS, model-infrastructure, VectorIndexManager, and TurboQuant paths while keeping Simeon PQ and float-vector storage.

Migrate unused V2.2 quantized schemas transactionally to V2.3. Databases containing non-default quantized formats or packed-code blobs fail closed with recovery guidance and no mutation.

* refactor(plugins): remove legacy SDK tooling

Remove the Ghidra and SDK submodules, their bespoke tests, and stale scan exclusions. Retain generic PluginProcess coverage against the repository mock fixture.

* build(vector): remove stale FAISS overrides

* refactor(ipc): remove dormant FSM snapshots

* refactor(ingest): unify bounded KG enqueue

* fix(plugins): harden allocation contracts

Keep ABI outputs null or fully initialized when allocations fail, and represent empty object payloads without relying on malloc(0).

* fix(storage): define empty object reads

Return a null buffer with zero length for successful empty-object reads instead of treating malloc(0) as an allocation contract.

* refactor(storage): trim through string views

* test: prune dormant harness helpers

Remove unused readiness and fixture helpers, and keep benchmark client-count doubling in a wide integer domain.

* chore(cpp): initialize analyzer-facing locals

* build(simeon): advance analyzer cleanup

* refactor(config): drop dead daemon client keys

* refactor(ipc): inject daemon client transport

Socket and in-process execution now share DaemonClient request dispatch, so CLI fallbacks do not need alternate service paths.

* refactor(retrieval): route CLI through daemon

Search and grep now use the same daemon handlers over socket or in-process transport. Live grep remains an explicit freshness inspection.

* refactor(graph): move CLI scope into daemon

Graph CLI requests now share daemon handlers across socket and in-process transports. CWD filtering occurs before pagination so scoped totals and pages stay correct.

* build: isolate test-only runtime controls

* refactor(metadata): decouple lock telemetry

* fix(services): return on request timeout

* fix(retrieval): enforce workspace scope

* fix(graph): scope nodes before pagination

* fix(grep): drop binary result lines

* fix(daemon): bound checkpoint wait

* fix(repair): use KG health checks

* fix(storage): always verify compressed CRC

* perf(vector): skip unchanged PQ snapshots

* build(zyp): exclude bench from default build

* build: protect Meson files from formatter

* refactor: remove unreachable production APIs

* refactor(daemon): remove dormant runtime state

* fix(vector): harden daemon provider calls

Timed-out coroutines must own their client and requests. Provider discovery must also fail instead of returning synthetic embeddings in production.

* refactor(storage): remove unusable adapters

* fix(cli): unify in-process add requests

* fix(tsan): preserve async worker progress

* test(cli): isolate daemon-dependent suites

* fix(repair): expose queue backoff progress
2026-07-31 14:16:45 -04:00

1222 lines
37 KiB
C++

// Catch2 migration of daemon_test.cpp
// Migration: yams-3s4 (daemon unit tests)
// Unit tests for YamsDaemon - creation, start/stop, signal handling, error paths
#include <catch2/catch_test_macros.hpp>
#include <atomic>
#include <chrono>
#include <filesystem>
#include <fstream>
#include <future>
#include <iomanip>
#include <sstream>
#include <string_view>
#include <thread>
#include <nlohmann/json.hpp>
using nlohmann::json;
#include <yams/compat/unistd.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/daemon.h>
#include <yams/daemon/daemon_lifecycle.h>
#include <yams/daemon/ipc/ipc_protocol.h>
#include <yams/daemon/ipc/proto_serializer.h>
#include <yams/metadata/document_metadata.h>
#include <yams/metadata/metadata_repository.h>
#include "tests/common/test_helpers_catch2.h"
#ifdef _WIN32
#include <fcntl.h>
#include <io.h>
#include <share.h>
using pid_t = int;
#else
#include <fcntl.h>
#include <signal.h>
#include <sys/file.h>
#include <sys/wait.h>
#endif
namespace fs = std::filesystem;
using namespace std::chrono_literals;
using namespace yams;
using namespace yams::daemon;
namespace yams::daemon::test {
// Test fixture for Daemon tests
struct DaemonFixture {
DaemonConfig config_;
std::unique_ptr<YamsDaemon> daemon_;
fs::path runtime_root_;
std::thread runLoopThread_;
DaemonFixture() {
#ifdef _WIN32
fs::path tmp = fs::temp_directory_path() / "yams_test";
#else
fs::path tmp{"/tmp"};
#endif
auto unique_suffix =
std::to_string(::getpid()) + "_" +
std::to_string(static_cast<unsigned long>(reinterpret_cast<uintptr_t>(this) & 0xffff));
runtime_root_ = tmp / ("ydtest_" + unique_suffix);
config_.dataDir = runtime_root_ / "data";
config_.socketPath = runtime_root_ / "sock";
config_.pidFile = runtime_root_ / "daemon.pid";
config_.logFile = runtime_root_ / "daemon.log";
// When ABI plugins are disabled, no model provider can be adopted,
// so disable the requirement to avoid init failure.
if (const char* v = std::getenv("YAMS_DISABLE_ABI_PLUGINS");
v && std::string_view(v) == "1") {
config_.enableModelProvider = false;
} else {
config_.enableModelProvider = true;
}
config_.autoLoadPlugins = true;
config_.modelPoolConfig.lazyLoading = true;
config_.modelPoolConfig.preloadModels.clear();
::setenv("YAMS_DB_OPEN_TIMEOUT_MS", "1500", 1);
::setenv("YAMS_DB_MIGRATE_TIMEOUT_MS", "2000", 1);
::setenv("YAMS_SEARCH_BUILD_TIMEOUT_MS", "1500", 1);
::setenv("YAMS_DISABLE_VECTORS", "1", 1);
::setenv("YAMS_DISABLE_SESSION_WATCHER", "1", 1);
std::error_code se;
fs::create_directories(config_.dataDir, se);
::setenv("YAMS_RUNTIME_DIR", runtime_root_.string().c_str(), 1);
::setenv("YAMS_SOCKET_PATH", config_.socketPath.string().c_str(), 1);
::setenv("YAMS_PID_FILE", config_.pidFile.string().c_str(), 1);
}
~DaemonFixture() {
stopRunLoop();
if (daemon_) {
daemon_->stop();
daemon_.reset();
}
cleanupDaemonFiles();
}
void startRunLoop() {
if (!daemon_ || runLoopThread_.joinable()) {
return;
}
runLoopThread_ = std::thread([this]() { daemon_->runLoop(); });
std::this_thread::sleep_for(50ms);
}
void stopRunLoop() {
if (daemon_ && daemon_->isRunning()) {
daemon_->requestStop();
}
if (runLoopThread_.joinable()) {
runLoopThread_.join();
}
}
void cleanupDaemonFiles() {
std::error_code ec;
if (!runtime_root_.empty()) {
fs::remove_all(runtime_root_, ec);
}
}
static bool isSocketPermissionDenied(const yams::Error& error) {
const std::string_view message{error.message};
return message.find("Operation not permitted") != std::string_view::npos ||
message.find("Permission denied") != std::string_view::npos;
}
bool waitForDaemonReady(std::chrono::milliseconds timeout = 5000ms) {
auto deadline = std::chrono::steady_clock::now() + timeout;
while (std::chrono::steady_clock::now() < deadline) {
const auto& state = daemon_->getState();
if (state.readiness.contentStoreReady.load()) {
return true;
}
std::this_thread::sleep_for(50ms);
}
return false;
}
static bool waitForCondition(std::chrono::milliseconds timeout,
const std::function<bool()>& predicate) {
auto deadline = std::chrono::steady_clock::now() + timeout;
while (std::chrono::steady_clock::now() < deadline) {
if (predicate()) {
return true;
}
std::this_thread::sleep_for(10ms);
}
return predicate();
}
};
#ifdef _WIN32
#define SKIP_ON_WINDOWS() SKIP("Daemon IPC tests unstable on Windows")
#else
#define SKIP_ON_WINDOWS() ((void)0)
#endif
TEST_CASE_METHOD(DaemonFixture, "Daemon creation and destruction", "[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
REQUIRE(daemon_ != nullptr);
REQUIRE_FALSE(daemon_->isRunning());
}
TEST_CASE_METHOD(DaemonFixture, "Daemon tuning reload preserves unspecified values",
"[daemon][tuning][reload]") {
SKIP_ON_WINDOWS();
const auto configPath = runtime_root_ / "config.toml";
{
std::ofstream out(configPath);
REQUIRE(out.is_open());
out << "[tuning]\n";
out << "target_cpu_percent = 321\n";
out << "post_ingest_capacity = 111\n";
out << "control_interval_ms = 222\n";
}
config_.configFilePath = configPath;
config_.tuning.postIngestThreadsMin = 5;
config_.tuning.holdMs = 777;
daemon_ = std::make_unique<YamsDaemon>(config_);
REQUIRE(daemon_ != nullptr);
daemon_->reloadTuningConfig();
CHECK(daemon_->config_.tuning.targetCpuPercent == 321u);
CHECK(daemon_->config_.tuning.postIngestCapacity == 111u);
CHECK(daemon_->config_.tuning.controlIntervalMs == 222u);
CHECK(daemon_->config_.tuning.postIngestThreadsMin == 5u);
CHECK(daemon_->config_.tuning.holdMs == 777u);
}
TEST_CASE_METHOD(DaemonFixture, "Lifecycle shutdown waits for owner-thread stop",
"[daemon][lifecycle][shutdown-request]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto startResult = daemon_->start();
if (!startResult && isSocketPermissionDenied(startResult.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(startResult);
startRunLoop();
REQUIRE(waitForCondition(
2s, [&] { return daemon_->runLoopStarted_.load(std::memory_order_acquire); }));
DaemonLifecycleAdapter lifecycle(daemon_.get());
lifecycle.requestShutdown(true, true);
REQUIRE(waitForCondition(2s, [&] {
return daemon_->stopRequested_.load(std::memory_order_acquire) &&
daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
}));
std::this_thread::sleep_for(100ms);
CHECK(daemon_->shutdownThreadActive_.load(std::memory_order_acquire));
stopRunLoop();
auto stopResult = daemon_->stop();
REQUIRE(stopResult);
REQUIRE(waitForCondition(
2s, [&] { return !daemon_->shutdownThreadActive_.load(std::memory_order_acquire); }));
daemon_->reapCompletedShutdownThread();
}
TEST_CASE_METHOD(DaemonFixture,
"Lifecycle shutdown request does not cancel an in-flight worker task",
"[daemon][lifecycle][shutdown-request][cancellation]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto startResult = daemon_->start();
if (!startResult && isSocketPermissionDenied(startResult.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(startResult);
startRunLoop();
REQUIRE(waitForCondition(
2s, [&] { return daemon_->runLoopStarted_.load(std::memory_order_acquire); }));
auto* serviceManager = daemon_->getServiceManager();
REQUIRE((serviceManager != nullptr));
std::mutex gateMutex;
std::condition_variable gateCv;
bool allowExit = false;
std::atomic<bool> workerEntered{false};
std::atomic<bool> workerFinished{false};
boost::asio::post(serviceManager->getWorkerExecutor(), [&]() {
{
std::lock_guard<std::mutex> lk(gateMutex);
workerEntered.store(true, std::memory_order_release);
}
gateCv.notify_all();
std::unique_lock<std::mutex> lk(gateMutex);
gateCv.wait(lk, [&] { return allowExit; });
workerFinished.store(true, std::memory_order_release);
});
REQUIRE(waitForCondition(2s, [&] { return workerEntered.load(std::memory_order_acquire); }));
DaemonLifecycleAdapter lifecycle(daemon_.get());
lifecycle.requestShutdown(true, true);
REQUIRE(waitForCondition(2s, [&] {
return daemon_->stopRequested_.load(std::memory_order_acquire) &&
daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
}));
std::this_thread::sleep_for(150ms);
CHECK(daemon_->shutdownThreadActive_.load(std::memory_order_acquire));
CHECK_FALSE(workerFinished.load(std::memory_order_acquire));
{
std::lock_guard<std::mutex> lk(gateMutex);
allowExit = true;
}
gateCv.notify_all();
REQUIRE(waitForCondition(2s, [&] { return workerFinished.load(std::memory_order_acquire); }));
stopRunLoop();
auto stopResult = daemon_->stop();
REQUIRE(stopResult);
REQUIRE(waitForCondition(
2s, [&] { return !daemon_->shutdownThreadActive_.load(std::memory_order_acquire); }));
daemon_->reapCompletedShutdownThread();
}
TEST_CASE_METHOD(DaemonFixture,
"Lifecycle shutdown completion stays pending while worker teardown is blocked",
"[daemon][lifecycle][shutdown-request][budget]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto startResult = daemon_->start();
if (!startResult && isSocketPermissionDenied(startResult.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(startResult);
startRunLoop();
REQUIRE(waitForCondition(
2s, [&] { return daemon_->runLoopStarted_.load(std::memory_order_acquire); }));
auto* serviceManager = daemon_->getServiceManager();
REQUIRE((serviceManager != nullptr));
std::mutex gateMutex;
std::condition_variable gateCv;
bool allowExit = false;
std::atomic<bool> workerEntered{false};
boost::asio::post(serviceManager->getWorkerExecutor(), [&]() {
{
std::lock_guard<std::mutex> lk(gateMutex);
workerEntered.store(true, std::memory_order_release);
}
gateCv.notify_all();
std::unique_lock<std::mutex> lk(gateMutex);
gateCv.wait(lk, [&] { return allowExit; });
});
REQUIRE(waitForCondition(2s, [&] { return workerEntered.load(std::memory_order_acquire); }));
DaemonLifecycleAdapter lifecycle(daemon_.get());
lifecycle.requestShutdown(true, true);
REQUIRE(waitForCondition(2s, [&] {
return daemon_->stopRequested_.load(std::memory_order_acquire) &&
daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
}));
CHECK_FALSE(daemon_->waitForStopCompletion(100ms));
std::this_thread::sleep_for(150ms);
CHECK(daemon_->shutdownThreadActive_.load(std::memory_order_acquire));
CHECK_FALSE(daemon_->waitForStopCompletion(100ms));
{
std::lock_guard<std::mutex> lk(gateMutex);
allowExit = true;
}
gateCv.notify_all();
stopRunLoop();
auto stopResult = daemon_->stop();
REQUIRE(stopResult);
REQUIRE(daemon_->waitForStopCompletion(2s));
daemon_->reapCompletedShutdownThread();
}
TEST_CASE_METHOD(DaemonFixture,
"Lifecycle shutdown timeout floor covers work coordinator join budget",
"[daemon][lifecycle][shutdown-request][budget]") {
SKIP_ON_WINDOWS();
yams::test::ScopedEnvVar shutdownBudget("YAMS_SHUTDOWN_FORCE_EXIT_MS", std::string("1000"));
daemon_ = std::make_unique<YamsDaemon>(config_);
auto startResult = daemon_->start();
if (!startResult && isSocketPermissionDenied(startResult.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(startResult);
startRunLoop();
REQUIRE(waitForCondition(
2s, [&] { return daemon_->runLoopStarted_.load(std::memory_order_acquire); }));
auto* serviceManager = daemon_->getServiceManager();
REQUIRE((serviceManager != nullptr));
std::mutex gateMutex;
std::condition_variable gateCv;
bool allowExit = false;
std::atomic<bool> workerEntered{false};
boost::asio::post(serviceManager->getWorkerExecutor(), [&]() {
{
std::lock_guard<std::mutex> lk(gateMutex);
workerEntered.store(true, std::memory_order_release);
}
gateCv.notify_all();
std::unique_lock<std::mutex> lk(gateMutex);
gateCv.wait(lk, [&] { return allowExit; });
});
REQUIRE(waitForCondition(2s, [&] { return workerEntered.load(std::memory_order_acquire); }));
DaemonLifecycleAdapter lifecycle(daemon_.get());
lifecycle.requestShutdown(true, true);
REQUIRE(waitForCondition(2s, [&] {
return daemon_->stopRequested_.load(std::memory_order_acquire) &&
daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
}));
std::this_thread::sleep_for(1200ms);
CHECK(daemon_->shutdownThreadActive_.load(std::memory_order_acquire));
CHECK_FALSE(daemon_->waitForStopCompletion(100ms));
{
std::lock_guard<std::mutex> lk(gateMutex);
allowExit = true;
}
gateCv.notify_all();
stopRunLoop();
auto stopResult = daemon_->stop();
REQUIRE(stopResult);
REQUIRE(daemon_->waitForStopCompletion(2s));
daemon_->reapCompletedShutdownThread();
}
TEST_CASE_METHOD(DaemonFixture, "Lifecycle shutdown falls back to direct stop without run loop",
"[daemon][lifecycle][shutdown-request][fallback]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto startResult = daemon_->start();
if (!startResult && isSocketPermissionDenied(startResult.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(startResult);
REQUIRE_FALSE(daemon_->runLoopStarted_.load(std::memory_order_acquire));
DaemonLifecycleAdapter lifecycle(daemon_.get());
lifecycle.requestShutdown(true, true);
REQUIRE(waitForCondition(2s, [&] {
return !daemon_->isRunning() &&
!daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
}));
CHECK(daemon_->getLifecycle().snapshot().state == LifecycleState::Stopped);
if (const auto* serviceManager = daemon_->getServiceManager()) {
CHECK(serviceManager->getServiceManagerFsmSnapshot().state == ServiceManagerState::Stopped);
}
daemon_->reapCompletedShutdownThread();
}
TEST_CASE_METHOD(DaemonFixture,
"Lifecycle shutdown avoids forced exit for foreground-managed daemon",
"[daemon][lifecycle][shutdown-request][foreground-managed]") {
SKIP_ON_WINDOWS();
yams::test::ScopedEnvVar foregroundManaged("YAMS_DAEMON_FOREGROUND", std::string("1"));
daemon_ = std::make_unique<YamsDaemon>(config_);
auto startResult = daemon_->start();
if (!startResult && isSocketPermissionDenied(startResult.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(startResult);
startRunLoop();
REQUIRE(waitForCondition(
2s, [&] { return daemon_->runLoopStarted_.load(std::memory_order_acquire); }));
DaemonLifecycleAdapter lifecycle(daemon_.get());
lifecycle.requestShutdown(true, false);
REQUIRE(waitForCondition(2s, [&] {
return daemon_->stopRequested_.load(std::memory_order_acquire) &&
daemon_->shutdownThreadActive_.load(std::memory_order_acquire);
}));
stopRunLoop();
auto stopResult = daemon_->stop();
REQUIRE(stopResult);
REQUIRE(waitForCondition(
2s, [&] { return !daemon_->shutdownThreadActive_.load(std::memory_order_acquire); }));
daemon_->reapCompletedShutdownThread();
REQUIRE_FALSE(daemon_->isRunning());
}
TEST_CASE_METHOD(DaemonFixture, "Daemon start and stop", "[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto startResult = daemon_->start();
if (!startResult && isSocketPermissionDenied(startResult.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(startResult);
REQUIRE(daemon_->isRunning());
REQUIRE(fs::exists(config_.pidFile));
startRunLoop();
REQUIRE(waitForDaemonReady());
stopRunLoop();
auto result = daemon_->stop();
REQUIRE(result);
REQUIRE_FALSE(daemon_->isRunning());
REQUIRE(daemon_->getLifecycle().snapshot().state == LifecycleState::Stopped);
if (const auto* serviceManager = daemon_->getServiceManager()) {
REQUIRE(serviceManager->getServiceManagerFsmSnapshot().state ==
ServiceManagerState::Stopped);
}
REQUIRE_FALSE(fs::exists(config_.pidFile));
}
TEST_CASE_METHOD(DaemonFixture, "Daemon single instance enforcement", "[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
auto daemon2 = std::make_unique<YamsDaemon>(config_);
auto result2 = daemon2->start();
REQUIRE_FALSE(result2);
REQUIRE(result2.error().code == ErrorCode::InvalidState);
daemon_->stop();
}
TEST_CASE_METHOD(DaemonFixture, "Daemon rejects pid file pointing to current process",
"[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
json payload = json::object();
payload["pid"] = static_cast<std::int64_t>(getpid());
payload["start_ns"] = 1;
payload["token"] = "test";
std::ofstream pidFile(config_.pidFile);
pidFile << payload.dump();
pidFile.close();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
REQUIRE_FALSE(result);
REQUIRE(result.error().code == ErrorCode::InvalidState);
}
TEST_CASE_METHOD(DaemonFixture, "Daemon refuses pid file pointing to unrelated running process",
"[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
#ifndef _WIN32
pid_t child = fork();
if (child < 0) {
SKIP("fork not available");
}
if (child == 0) {
for (;;) {
::pause();
}
}
json payload = json::object();
payload["pid"] = static_cast<std::int64_t>(child);
payload["start_ns"] = 1;
payload["token"] = "test";
std::ofstream pidFile(config_.pidFile);
pidFile << payload.dump();
pidFile.close();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
REQUIRE_FALSE(result);
REQUIRE(result.error().code == ErrorCode::InvalidState);
::kill(child, SIGKILL);
int status = 0;
(void)::waitpid(child, &status, 0);
#else
SKIP("POSIX process APIs not available on Windows");
#endif
}
TEST_CASE_METHOD(DaemonFixture, "Daemon does not terminate unverified data-dir lock holder",
"[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
#ifndef _WIN32
int readyPipe[2] = {-1, -1};
REQUIRE(::pipe(readyPipe) == 0);
const pid_t child = ::fork();
REQUIRE(child >= 0);
if (child == 0) {
::close(readyPipe[0]);
const auto lockPath = config_.dataDir / ".yams-lock";
const int lockFd = ::open(lockPath.c_str(), O_CREAT | O_RDWR, 0644);
if (lockFd < 0 || ::flock(lockFd, LOCK_EX) != 0) {
_exit(2);
}
const auto payload = json{{"pid", ::getpid()}, {"socket", "/tmp/not-a-daemon.sock"}}.dump();
(void)::ftruncate(lockFd, 0);
(void)::write(lockFd, payload.data(), payload.size());
const char ready = '1';
(void)::write(readyPipe[1], &ready, 1);
for (;;) {
::pause();
}
}
::close(readyPipe[1]);
struct ChildGuard {
pid_t pid;
int readyFd;
~ChildGuard() {
::close(readyFd);
(void)::kill(pid, SIGKILL);
int status = 0;
(void)::waitpid(pid, &status, 0);
}
} childGuard{child, readyPipe[0]};
char ready = 0;
REQUIRE(::read(readyPipe[0], &ready, 1) == 1);
REQUIRE(ready == '1');
daemon_ = std::make_unique<YamsDaemon>(config_);
const auto result = daemon_->start();
REQUIRE_FALSE(result);
CHECK(result.error().code == ErrorCode::InvalidState);
CHECK(::kill(child, 0) == 0);
#else
SKIP("POSIX process APIs not available on Windows");
#endif
}
TEST_CASE_METHOD(DaemonFixture, "Daemon restart", "[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
result = daemon_->stop();
REQUIRE(result);
result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted on restart");
}
REQUIRE(result);
REQUIRE(daemon_->isRunning());
daemon_->stop();
}
TEST_CASE_METHOD(DaemonFixture, "Daemon path resolution with empty config", "[daemon][config]") {
SKIP_ON_WINDOWS();
DaemonConfig autoConfig;
daemon_ = std::make_unique<YamsDaemon>(autoConfig);
REQUIRE(daemon_ != nullptr);
}
TEST_CASE_METHOD(DaemonFixture, "Daemon signal handling and PID file", "[daemon][signals]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
#ifndef _WIN32
std::ifstream pidFile(config_.pidFile);
std::string content;
std::getline(pidFile, content, '\0');
pidFile.close();
auto parsed = json::parse(content, nullptr, false);
REQUIRE_FALSE(parsed.is_discarded());
REQUIRE(parsed.value("pid", 0) == getpid());
#endif
daemon_->stop();
}
TEST_CASE_METHOD(DaemonFixture, "Daemon stale PID file cleanup", "[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
std::ofstream pidFile(config_.pidFile);
pidFile << "99999999\n";
pidFile.close();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
REQUIRE(daemon_->isRunning());
daemon_->stop();
}
TEST_CASE_METHOD(DaemonFixture, "Daemon invalid socket parent configuration", "[daemon][config]") {
SKIP_ON_WINDOWS();
const auto notADirectory = runtime_root_ / "socket-parent-file";
fs::create_directories(runtime_root_);
std::ofstream(notADirectory) << "not a directory\n";
config_.socketPath = notADirectory / "cannot-create.sock";
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
REQUIRE_FALSE(result);
}
TEST_CASE_METHOD(DaemonFixture,
"Daemon socket preflight does not clean unrelated stale PID artifacts",
"[daemon][config][lifecycle]") {
SKIP_ON_WINDOWS();
// Seed a stale PID artifact that this process does not own.
{
std::ofstream pidFile(config_.pidFile);
pidFile << "99999999\n";
}
REQUIRE(fs::exists(config_.pidFile));
// Force an AF_UNIX-overlong socket path so start fails in preflight, before lifecycle init.
config_.socketPath = runtime_root_ / std::string(140, 'x') / "daemon.sock";
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
REQUIRE_FALSE(result);
CHECK(result.error().message.find("Socket path too long") != std::string::npos);
// Preflight failure must not mutate stale PID artifacts we did not create.
CHECK(fs::exists(config_.pidFile));
}
TEST_CASE_METHOD(DaemonFixture, "Daemon concurrent start attempts", "[daemon][concurrency]") {
SKIP_ON_WINDOWS();
{
YamsDaemon probe(config_);
auto result = probe.start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
probe.stop();
}
std::atomic<int> successCount{0};
std::atomic<int> failCount{0};
std::once_flag startedFlag;
std::promise<void> startedPromise;
auto startedFuture = startedPromise.get_future().share();
std::promise<void> releasePromise;
auto releaseFuture = releasePromise.get_future().share();
std::promise<void> attemptGatePromise;
auto attemptGateFuture = attemptGatePromise.get_future().share();
std::mutex readyMutex;
std::condition_variable readyCv;
int readyCount = 0;
std::vector<std::thread> threads;
constexpr int kAttempts = 5;
threads.reserve(kAttempts);
for (int i = 0; i < kAttempts; ++i) {
threads.emplace_back([this, &successCount, &failCount, &startedFlag, startedFuture,
&startedPromise, releaseFuture, attemptGateFuture, &readyMutex,
&readyCv, &readyCount]() {
{
std::lock_guard<std::mutex> lock(readyMutex);
++readyCount;
}
readyCv.notify_one();
attemptGateFuture.wait();
YamsDaemon local(config_);
auto result = local.start();
if (result) {
successCount.fetch_add(1, std::memory_order_relaxed);
std::call_once(startedFlag, [&]() { startedPromise.set_value(); });
releaseFuture.wait();
local.stop();
} else {
startedFuture.wait();
failCount.fetch_add(1, std::memory_order_relaxed);
}
});
}
{
std::unique_lock<std::mutex> lock(readyMutex);
REQUIRE(readyCv.wait_for(lock, std::chrono::seconds(1),
[&] { return readyCount == kAttempts; }));
}
attemptGatePromise.set_value();
if (startedFuture.wait_for(std::chrono::seconds(1)) != std::future_status::ready) {
std::call_once(startedFlag, [&]() { startedPromise.set_value(); });
}
const auto failDeadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
while (failCount.load(std::memory_order_relaxed) < kAttempts - 1 &&
std::chrono::steady_clock::now() < failDeadline) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
}
releasePromise.set_value();
for (auto& t : threads) {
t.join();
}
REQUIRE(successCount.load() == 1);
REQUIRE(failCount.load() == kAttempts - 1);
}
TEST_CASE_METHOD(DaemonFixture, "Daemon stats tracking", "[daemon][stats]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
[[maybe_unused]] const auto& state = daemon_->getState();
daemon_->stop();
}
TEST_CASE_METHOD(DaemonFixture, "Daemon hybrid search smoke test", "[daemon][search]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
SearchRequest req;
req.query = "seed";
req.limit = 5;
req.fuzzy = false;
req.literalText = false;
req.similarity = 0.7;
Message msg;
msg.version = 1;
msg.requestId = 42;
msg.payload = Request{req};
auto enc = ProtoSerializer::encode_payload(msg);
REQUIRE(enc);
auto parsed = ProtoSerializer::decode_payload(enc.value());
REQUIRE(parsed);
REQUIRE(daemon_->isRunning());
daemon_->stop();
}
TEST_CASE_METHOD(DaemonFixture, "Daemon fallback search when hybrid unavailable",
"[daemon][search]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
SearchRequest req;
req.query = "fallback";
req.limit = 3;
req.fuzzy = true;
req.similarity = 0.6;
Message msg;
msg.version = 1;
msg.requestId = 7;
msg.payload = Request{req};
auto enc = ProtoSerializer::encode_payload(msg);
REQUIRE(enc);
auto parsed = ProtoSerializer::decode_payload(enc.value());
REQUIRE(parsed);
REQUIRE(daemon_->isRunning());
daemon_->stop();
}
TEST_CASE_METHOD(DaemonFixture, "StatusResponse serialization includes ready field",
"[daemon][serialization]") {
StatusResponse sr;
sr.running = true;
sr.ready = true;
sr.uptimeSeconds = 0;
sr.requestsProcessed = 0;
sr.activeConnections = 0;
sr.memoryUsageMb = 0.0;
sr.cpuUsagePercent = 0.0;
sr.version = "test";
Message msg;
msg.version = 1;
msg.requestId = 0;
msg.payload = Response{sr};
auto enc = ProtoSerializer::encode_payload(msg);
REQUIRE(enc);
auto des = ProtoSerializer::decode_payload(enc.value());
REQUIRE(des);
Message out = des.value();
auto respVariant = std::get<Response>(out.payload);
REQUIRE(std::holds_alternative<StatusResponse>(respVariant));
auto outSr = std::get<StatusResponse>(respVariant);
REQUIRE(outSr.ready == true);
REQUIRE(outSr.running == true);
REQUIRE(outSr.version == "test");
}
TEST_CASE_METHOD(DaemonFixture, "Daemon start fails when data directory path is a file",
"[daemon][config]") {
SKIP_ON_WINDOWS();
const auto notADirectory = runtime_root_ / "data-file";
fs::create_directories(runtime_root_);
std::ofstream(notADirectory) << "not a directory\n";
DaemonConfig bad = config_;
bad.dataDir = notADirectory;
daemon_ = std::make_unique<YamsDaemon>(bad);
auto result = daemon_->start();
REQUIRE_FALSE(result);
}
TEST_CASE_METHOD(DaemonFixture, "Daemon PID file cannot be created below a file path",
"[daemon][errors]") {
SKIP_ON_WINDOWS();
const auto notADirectory = runtime_root_ / "pid-parent-file";
fs::create_directories(runtime_root_);
std::ofstream(notADirectory) << "not a directory\n";
DaemonConfig bad_config = config_;
bad_config.pidFile = notADirectory / "yams.pid";
daemon_ = std::make_unique<YamsDaemon>(bad_config);
auto result = daemon_->start();
REQUIRE_FALSE(result);
if (!result) {
REQUIRE_FALSE(result.error().message.empty());
}
}
TEST_CASE_METHOD(DaemonFixture, "Daemon PID file cleaned up on start failure", "[daemon][errors]") {
SKIP_ON_WINDOWS();
const auto notADirectory = runtime_root_ / "data-file-for-pid-cleanup";
fs::create_directories(runtime_root_);
std::ofstream(notADirectory) << "not a directory\n";
DaemonConfig bad_config = config_;
bad_config.dataDir = notADirectory;
daemon_ = std::make_unique<YamsDaemon>(bad_config);
auto result = daemon_->start();
REQUIRE_FALSE(result);
REQUIRE_FALSE(fs::exists(bad_config.pidFile));
}
TEST_CASE_METHOD(DaemonFixture, "Daemon stop when not started is safe", "[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
REQUIRE_NOTHROW(daemon_->stop());
REQUIRE_NOTHROW(daemon_->stop());
}
TEST_CASE_METHOD(DaemonFixture, "Daemon double stop is idempotent", "[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
daemon_->stop();
REQUIRE_NOTHROW(daemon_->stop());
}
TEST_CASE_METHOD(DaemonFixture, "Daemon fails when socket parent is a file", "[daemon][config]") {
SKIP_ON_WINDOWS();
const auto notADirectory = runtime_root_ / "socket-parent-file-2";
fs::create_directories(runtime_root_);
std::ofstream(notADirectory) << "not a directory\n";
DaemonConfig bad_config = config_;
bad_config.socketPath = notADirectory / "subdir" / "yams.sock";
daemon_ = std::make_unique<YamsDaemon>(bad_config);
auto result = daemon_->start();
REQUIRE_FALSE(result);
}
TEST_CASE_METHOD(DaemonFixture, "Daemon socket cleanup after stop", "[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
REQUIRE(fs::exists(config_.socketPath));
daemon_->stop();
REQUIRE_FALSE(fs::exists(config_.socketPath));
}
TEST_CASE_METHOD(DaemonFixture, "Daemon restart with existing socket", "[daemon][lifecycle]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result1 = daemon_->start();
if (!result1 && isSocketPermissionDenied(result1.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result1);
daemon_->stop();
auto result2 = daemon_->start();
REQUIRE(result2);
daemon_->stop();
}
TEST_CASE_METHOD(DaemonFixture, "Daemon empty data directory handling", "[daemon][config]") {
SKIP_ON_WINDOWS();
DaemonConfig bad_config = config_;
bad_config.dataDir = "";
daemon_ = std::make_unique<YamsDaemon>(bad_config);
auto result = daemon_->start();
if (result) {
daemon_->stop();
} else {
REQUIRE_FALSE(result.error().message.empty());
}
}
TEST_CASE_METHOD(DaemonFixture, "Daemon status query on stopped daemon", "[daemon][state]") {
SKIP_ON_WINDOWS();
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
const auto& state_running = daemon_->getState();
REQUIRE(state_running.readiness.ipcServerReady.load());
daemon_->stop();
const auto& state_stopped = daemon_->getState();
REQUIRE_FALSE(state_stopped.readiness.ipcServerReady.load());
}
TEST_CASE_METHOD(DaemonFixture, "Daemon async init barrier prevents race condition",
"[daemon][concurrency]") {
SKIP_ON_WINDOWS();
if (const char* disable_vectors = std::getenv("YAMS_DISABLE_VECTORS")) {
if (std::string_view(disable_vectors) == "1") {
SKIP("Vector DB disabled via YAMS_DISABLE_VECTORS=1");
}
}
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
REQUIRE(daemon_->isRunning());
std::atomic<bool> runLoopEntered{false};
std::atomic<bool> runLoopExited{false};
std::thread runLoopThread([this, &runLoopEntered, &runLoopExited]() {
runLoopEntered.store(true);
daemon_->runLoop();
runLoopExited.store(true);
});
std::this_thread::sleep_for(100ms);
REQUIRE(runLoopEntered.load());
daemon_->requestStop();
if (runLoopThread.joinable()) {
runLoopThread.join();
}
REQUIRE(runLoopExited.load());
daemon_->stop();
}
TEST_CASE_METHOD(DaemonFixture, "Daemon async init barrier stress test",
"[daemon][concurrency][.slow]") {
SKIP_ON_WINDOWS();
constexpr int kIterations = 5;
for (int i = 0; i < kIterations; ++i) {
INFO("Iteration " << i);
daemon_ = std::make_unique<YamsDaemon>(config_);
auto result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
if (!result) {
std::this_thread::sleep_for(100ms);
result = daemon_->start();
if (!result && isSocketPermissionDenied(result.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(result);
}
REQUIRE(daemon_->isRunning());
std::atomic<bool> crashed{false};
std::thread runLoopThread([this, &crashed]() {
try {
daemon_->runLoop();
} catch (...) {
crashed.store(true);
}
});
std::this_thread::sleep_for(50ms);
daemon_->requestStop();
if (runLoopThread.joinable()) {
runLoopThread.join();
}
REQUIRE_FALSE(crashed.load());
daemon_->stop();
daemon_.reset();
std::this_thread::sleep_for(50ms);
}
}
} // namespace yams::daemon::test