trevon/yams
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1
yams/tests/unit/daemon/daemon_catch2_test.cpp
Trevon 5eb01f4048
 test(memory-sync): stop the sync service off the apply lock in stage hooks (#352) 
* test(memory-sync): stop the sync service off the apply lock in stage hooks

Two integration tests stop the sync service from a stage observer to
simulate a stop during apply or backfill. The observer runs under
ServiceManager's apply lock, and the blocking stop() then takes the
service's callback lock, the reverse of the worker's order (callback
lock, then apply lock in applyMemorySyncWinners). TSan on a gcc Debug
build reports a lock-order inversion in daemon_memory_sync_apply_stop and
daemon_memory_sync_backfill_stop. stop() also joins the worker, which can
be blocked on that apply lock, so the test can hang.

Production shutdown calls stop() without the apply lock, so this is a
test hazard only. StopFromStage runs stop() on a helper thread, returns
once stopRequested() is visible, and joins after the stage call returns.

Signed-off-by: trvon <me@trevon.dev>

* fix(cli): keep the memory-sync status request alive until the call runs (#353)

fetchMemorySyncStatus passed a temporary MemorySyncRequest to
DaemonClient::call(), which is a coroutine taking the request by
reference, and returned the awaitable from the provider lambda.
runDaemonClient awaits it only after the lambda returns, so the call
read a destroyed request.

Under TSan on a gcc Debug build, `yams daemon status -d` aborted while
copying the request into the Request variant: "requested allocation size
0x7ffed4d77a51" (a stack address read as the key length). That failed
integration_smoke. ASan does not flag it without stack-use-after-return
detection.

The provider is now a coroutine lambda that owns the request. The other
providers that return call() awaitables capture their request by value
and keep the closure alive while awaiting (awaitResultSync), so they are
not affected.

Signed-off-by: trvon <me@trevon.dev>

* fix(extraction): return plugin output as a copy taken under its lock (#354)

PluginProcess::read_stdout() and read_stderr() returned a span into the
output buffer after releasing the buffer mutex. The reader thread keeps
appending to that buffer, and a reallocation frees the storage the span
points to. JsonRpcClient::read_next_response() scans that span for a
newline, so a plugin that writes while a response is being parsed reads
freed memory.

TSan on a gcc Debug build reports the race in plugin_process
(read_stdout_loop's insert against the test's unlocked read). Both
accessors now return std::vector<std::byte>, copied under the lock.
Callers use auto or pass the result as a span, so they are unchanged.

Signed-off-by: trvon <me@trevon.dev>

* fix(client): stop the write loop once close() has reset the socket (#355)

close() resets AsioConnection::socket on the strand. A write loop that
still had frames queued, or a new async_write_frame() after close, went
on to async_write(*socket, ...) through the null unique_ptr. Debug builds
with _GLIBCXX_ASSERTIONS abort there; other builds bind a reference to
null.

Seen on a gcc Debug TSan build: client_timeout_recovery aborted at
asio_connection.cpp:233, and Catch2's fatal-signal handler then
allocated inside the SIGABRT handler (TSan "signal-unsafe call") and
hung until the 60 s test timeout.

The write loop now ends with NetworkError when the socket is gone or
closed, mirroring the read loop, and the timeout path checks the socket
before closing it.

Signed-off-by: trvon <me@trevon.dev>

* test(tsan): apply the suppressions file to every test (#356)

tsan.suppressions reached only the CLI, smoke, search and services
suites, the ones whose test() env sets TSAN_OPTIONS. Every other test
ran unsuppressed. A default 'tsan' test setup now sets TSAN_OPTIONS for
all tests in a TSan build. Tests that already set it use the same value.

Two narrow rules cover false positives that a gcc TSan build reports
and clang does not:
- boost::asio::detail::ref_count_down: fetch_sub(release) then a
  standalone acquire fence, which gcc TSan does not model
  (PostIngestQueue, VectorIndexCoordinator, WorkCoordinator reports).
- walIndexTryHdr / walIndexWriteHdr: SQLite's lock-free WAL-index header
  protocol. SQLITE_NO_TSAN is clang-only (metadata commit, symspell
  commit, KG path node reports).

Signed-off-by: trvon <me@trevon.dev>

* fix(daemon): run ServiceManager teardown even when the shared FSM is stopped (#357)

ServiceManagerFsm keeps its state in tinyfsm's process-wide storage.
ServiceManager::shutdown() returned early when that state was already
ShuttingDown or Stopped, so another manager's shutdown in the same
process cancelled this one's teardown. The common case: a second daemon
that loses the instance lock destroys its never-started ServiceManager,
whose destructor runs shutdown() and drives the shared FSM to Stopped.
The running daemon's stop() then skipped every shutdown phase. Its
WorkCoordinator threads and background coroutines kept running and kept
the ServiceManager alive.

gcc TSan reported the result in daemon_catch2's concurrent-start test: a
worker still inside the FTS5 consumer read TuningSnapshotRegistry after
exit() had destroyed it (heap-use-after-free).

shutdownInvoked_ is the per-instance guard. The shared state now only
decides whether to dispatch ShutdownEvent, and completeShutdown()
already accepts a Stopped state. A new test starts a daemon, loses a
second start against it, stops the first, and checks that its
ServiceManager is released.

Signed-off-by: trvon <me@trevon.dev>

* fix(daemon): leave the repair hysteresis timestamps to the run loop (#358)

stop() cleared repairBusySince_ and repairReadySince_ from the caller's
thread while the runLoop thread updates them in updateRepairHysteresis().
TSan on a gcc Debug build reports the race in
lifecycle_single_instance_release and grep_service_expectations
(daemon.cpp:1141/1142 against runLoop at 914).

Only runLoop uses these fields, and start() resets them before the loop
thread is created, so stop() no longer touches them.

Signed-off-by: trvon <me@trevon.dev>

* test(sync-indexing): scale the indexing waits with the test timeout scale (#359)

sync_indexing_catch2 waits a fixed 3 s for a freshly added document to
show up in grep, with a 1.5 s per-call timeout and a 2 s timing check.
Under gcc TSan everything runs about 10x slower, so "Sync add then
immediate grep" failed REQUIRE(found) at 3 s.

The test now gets _test_timeout_cpp_arg like the smoke tests and
multiplies those three bounds by YAMS_TEST_TIMEOUT_SCALE (1 in normal
builds, 10 under TSan).

Signed-off-by: trvon <me@trevon.dev>

* fix(build): keep the S3 plugin mapped in Linux TSan builds (#360)

storage_submodule passed every test (3167 assertions) under gcc TSan and
then crashed in exit(). The S3 plugin links its own static OpenSSL,
whose init registers atexit(OPENSSL_cleanup) from inside the plugin.
glibc ties that handler to the DSO and runs it at dlclose, but TSan's
atexit interceptor keeps it until process exit and calls into the
unmapped plugin (gdb: at_exit_callback_installed_at -> unmapped pc in
OPENSSL_cleanup of libyams_object_storage_s3.so).

Linux TSan builds now link the plugin with -z nodelete. Other builds are
unchanged.

Signed-off-by: trvon <me@trevon.dev>

* test(cli): scale the serve_delegation per-call timeout with the build (#361)

serve_delegation copies the yams CLI into a temp dir and runs it with a
fixed 10 s timeout. In an ASan build the binary is over 500 MB, and on
macOS the first exec of a freshly copied executable waits for the system
policy scan, so under load a call can exceed 10 s. The pre-push ASan
lane failed this way twice with no product change.

The script now takes --timeout-scale, and meson passes the build's test
timeout scale: 1 by default, 3 in the ASan lane, 10 under TSan. With
scale 3 the suite passes in 50 s at load average 24.

Signed-off-by: trvon <me@trevon.dev>

* test(daemon): scale the socket probe budgets in sanitizer builds (#362)

"unix_socket_io_permitted: real probe completes within budget" requires
the first probe in under 100 ms and the cached call in under 50 us. On a
gcc ASan build running tests in parallel, the first probe missed 100 ms
while the code under test was unchanged.

The budgets are now 10x under ASan or TSan, detected at compile time,
and unchanged in normal builds.

Signed-off-by: trvon <me@trevon.dev>

* build(tsan): use atomic coverage counters in TSan + coverage builds (#363)

gcov counters are plain increments. A gcc Debug build with
-Db_sanitize=thread -Db_coverage=true reported every counter that two
threads touch as a data race (`Location is global '__gcov0...'`), 32
reports in the first 43 tests and none in YAMS code.

When b_coverage is on in a TSan build, the project now adds
-fprofile-update=atomic for C and C++. Other builds are unchanged.

Signed-off-by: trvon <me@trevon.dev>

---------

Signed-off-by: trvon <me@trevon.dev>
2026-10-25 20:54:41 -06:00

1517 lines
52 KiB
C++
Raw Permalink Blame History

// 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 <cerrno>
#include <chrono>
#include <cstdio>
#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/config/config_helpers.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_;
std::vector<yams::test::ScopedEnvVar> environment_;
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 auto value = yams::config::getenv_nonempty("YAMS_DISABLE_ABI_PLUGINS");
value && *value == "1") {
config_.enableModelProvider = false;
} else {
config_.enableModelProvider = true;
}
config_.autoLoadPlugins = true;
config_.modelPoolConfig.lazyLoading = true;
config_.modelPoolConfig.preloadModels.clear();
environment_.emplace_back("YAMS_DB_OPEN_TIMEOUT_MS", std::string{"1500"});
environment_.emplace_back("YAMS_DB_MIGRATE_TIMEOUT_MS", std::string{"2000"});
environment_.emplace_back("YAMS_SEARCH_BUILD_TIMEOUT_MS", std::string{"1500"});
environment_.emplace_back("YAMS_DISABLE_VECTORS", std::string{"1"});
environment_.emplace_back("YAMS_DISABLE_SESSION_WATCHER", std::string{"1"});
std::error_code se;
fs::create_directories(config_.dataDir, se);
environment_.emplace_back("YAMS_RUNTIME_DIR", runtime_root_.string());
environment_.emplace_back("YAMS_SOCKET_PATH", config_.socketPath.string());
environment_.emplace_back("YAMS_PID_FILE", config_.pidFile.string());
}
~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 starts repair service while provider-degraded",
"[daemon][lifecycle][repair]") {
SKIP_ON_WINDOWS();
yams::test::ScopedEnvVar skipModelLoading("YAMS_SKIP_MODEL_LOADING", "1");
yams::test::ScopedEnvVar embeddingBackend("YAMS_EMBED_BACKEND", "daemon");
config_.enableModelProvider = true;
config_.modelProviderRequired = false;
config_.autoLoadPlugins = false;
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(5s, [&] {
return !daemon_->getState().readiness.modelProviderReady.load(std::memory_order_acquire);
}));
REQUIRE(waitForCondition(
2s, [&] { return daemon_->getServiceManager()->getRepairServiceShared() != nullptr; }));
}
TEST_CASE_METHOD(DaemonFixture, "Daemon restores compatibility path environment",
"[daemon][lifecycle][environment]") {
yams::test::ScopedEnvVar storage("YAMS_STORAGE", "/before/storage");
yams::test::ScopedEnvVar data("YAMS_DATA_DIR", "/before/data");
yams::test::ScopedEnvVar config("YAMS_CONFIG", "/before/config.toml");
const auto compatibilityConfig = runtime_root_ / "before-compatibility.toml";
std::ofstream(compatibilityConfig) << "[daemon]\nmode = \"old\"\n";
yams::test::ScopedEnvVar configCompatibility("YAMS_CONFIG_PATH", compatibilityConfig.string());
yams::test::ScopedEnvVar socket("YAMS_DAEMON_SOCKET", "/before/socket.sock");
yams::test::ScopedEnvVar inDaemon("YAMS_IN_DAEMON", "before");
config_.configFilePath = runtime_root_ / "config.toml";
daemon_ = std::make_unique<YamsDaemon>(config_);
CHECK((yams::config::getenv_copy("YAMS_STORAGE") == config_.dataDir.string()));
CHECK((yams::config::getenv_copy("YAMS_DATA_DIR") == config_.dataDir.string()));
CHECK((yams::config::getenv_copy("YAMS_CONFIG") == config_.configFilePath.string()));
CHECK((yams::config::getenv_copy("YAMS_CONFIG_PATH") == config_.configFilePath.string()));
CHECK((yams::config::get_config_path() == config_.configFilePath));
CHECK((yams::config::getenv_copy("YAMS_DAEMON_SOCKET") == config_.socketPath.string()));
CHECK((yams::config::getenv_copy("YAMS_IN_DAEMON") == "1"));
daemon_.reset();
CHECK((yams::config::getenv_copy("YAMS_STORAGE") == "/before/storage"));
CHECK((yams::config::getenv_copy("YAMS_DATA_DIR") == "/before/data"));
CHECK((yams::config::getenv_copy("YAMS_CONFIG") == "/before/config.toml"));
CHECK((yams::config::getenv_copy("YAMS_CONFIG_PATH") == compatibilityConfig.string()));
CHECK((yams::config::getenv_copy("YAMS_DAEMON_SOCKET") == "/before/socket.sock"));
CHECK((yams::config::getenv_copy("YAMS_IN_DAEMON") == "before"));
}
TEST_CASE_METHOD(DaemonFixture, "Daemon environment lease preserves newer writers",
"[daemon][lifecycle][environment][ownership]") {
yams::test::ScopedEnvVar socket{"YAMS_DAEMON_SOCKET", std::string{"/before/socket.sock"}};
daemon_ = std::make_unique<YamsDaemon>(config_);
REQUIRE((yams::config::getenv_copy("YAMS_DAEMON_SOCKET") == config_.socketPath.string()));
socket.set("/newer/owner.sock");
daemon_.reset();
CHECK((yams::config::getenv_copy("YAMS_DAEMON_SOCKET") == "/newer/owner.sock"));
}
TEST_CASE_METHOD(DaemonFixture, "Daemon environment lease detects same-value newer writers",
"[daemon][lifecycle][environment][ownership]") {
yams::test::ScopedEnvVar socket{"YAMS_DAEMON_SOCKET", std::string{"/before/socket.sock"}};
daemon_ = std::make_unique<YamsDaemon>(config_);
const auto installed = config_.socketPath.string();
REQUIRE((yams::config::getenv_copy("YAMS_DAEMON_SOCKET") == installed));
socket.set(installed);
daemon_.reset();
CHECK((yams::config::getenv_copy("YAMS_DAEMON_SOCKET") == installed));
}
TEST_CASE_METHOD(DaemonFixture, "Nested daemons restore every runtime environment lease",
"[daemon][lifecycle][environment][ownership][nested]") {
std::vector<yams::test::ScopedEnvVar> original;
original.reserve(6);
original.emplace_back("YAMS_IN_DAEMON", std::string{"before-daemon"});
original.emplace_back("YAMS_STORAGE", std::string{"/before/storage"});
original.emplace_back("YAMS_DATA_DIR", std::string{"/before/data"});
original.emplace_back("YAMS_CONFIG", std::string{"/before/config"});
original.emplace_back("YAMS_CONFIG_PATH", std::string{"/before/config-path"});
original.emplace_back("YAMS_DAEMON_SOCKET", std::string{"/before/socket"});
auto outerConfig = config_;
outerConfig.dataDir = runtime_root_ / "outer-data";
outerConfig.configFilePath = runtime_root_ / "outer.toml";
outerConfig.socketPath = runtime_root_ / "outer.sock";
outerConfig.pidFile = runtime_root_ / "outer.pid";
auto innerConfig = config_;
innerConfig.dataDir = runtime_root_ / "inner-data";
innerConfig.configFilePath = runtime_root_ / "inner.toml";
innerConfig.socketPath = runtime_root_ / "inner.sock";
innerConfig.pidFile = runtime_root_ / "inner.pid";
const auto checkInstalled = [](const DaemonConfig& expected) {
CHECK((yams::config::getenv_copy("YAMS_IN_DAEMON") == "1"));
CHECK((yams::config::getenv_copy("YAMS_STORAGE") == expected.dataDir.string()));
CHECK((yams::config::getenv_copy("YAMS_DATA_DIR") == expected.dataDir.string()));
CHECK((yams::config::getenv_copy("YAMS_CONFIG") == expected.configFilePath.string()));
CHECK((yams::config::getenv_copy("YAMS_CONFIG_PATH") == expected.configFilePath.string()));
CHECK((yams::config::getenv_copy("YAMS_DAEMON_SOCKET") == expected.socketPath.string()));
};
const auto checkOriginal = [] {
CHECK((yams::config::getenv_copy("YAMS_IN_DAEMON") == "before-daemon"));
CHECK((yams::config::getenv_copy("YAMS_STORAGE") == "/before/storage"));
CHECK((yams::config::getenv_copy("YAMS_DATA_DIR") == "/before/data"));
CHECK((yams::config::getenv_copy("YAMS_CONFIG") == "/before/config"));
CHECK((yams::config::getenv_copy("YAMS_CONFIG_PATH") == "/before/config-path"));
CHECK((yams::config::getenv_copy("YAMS_DAEMON_SOCKET") == "/before/socket"));
};
SECTION("LIFO destruction resumes the outer daemon") {
auto outer = std::make_unique<YamsDaemon>(outerConfig);
checkInstalled(outerConfig);
auto inner = std::make_unique<YamsDaemon>(innerConfig);
checkInstalled(innerConfig);
inner.reset();
checkInstalled(outerConfig);
outer.reset();
checkOriginal();
}
SECTION("non-LIFO destruction rebases the inner daemon onto the original state") {
auto outer = std::make_unique<YamsDaemon>(outerConfig);
auto inner = std::make_unique<YamsDaemon>(innerConfig);
checkInstalled(innerConfig);
outer.reset();
checkInstalled(innerConfig);
inner.reset();
checkOriginal();
}
}
TEST_CASE_METHOD(DaemonFixture, "Daemon construction failure releases acquired environment leases",
"[daemon][lifecycle][environment][ownership][failure]") {
std::vector<yams::test::ScopedEnvVar> original;
original.reserve(6);
original.emplace_back("YAMS_IN_DAEMON", std::string{"before-daemon"});
original.emplace_back("YAMS_STORAGE", std::string{"/before/storage"});
original.emplace_back("YAMS_DATA_DIR", std::string{"/before/data"});
original.emplace_back("YAMS_CONFIG", std::string{"/before/config"});
original.emplace_back("YAMS_CONFIG_PATH", std::string{"/before/config-path"});
original.emplace_back("YAMS_DAEMON_SOCKET", std::string{"/before/socket"});
config_.configFilePath = runtime_root_ / "failure.toml";
yams::config::testing_fail_owned_environment_lease_after(2);
try {
auto unexpected = std::make_unique<YamsDaemon>(config_);
FAIL("YamsDaemon construction unexpectedly succeeded");
} catch (const std::runtime_error& error) {
CHECK(std::string_view{error.what()}.starts_with(
"YamsDaemon failed to lease environment key YAMS_DATA_DIR"));
}
CHECK((yams::config::getenv_copy("YAMS_IN_DAEMON") == "before-daemon"));
CHECK((yams::config::getenv_copy("YAMS_STORAGE") == "/before/storage"));
CHECK((yams::config::getenv_copy("YAMS_DATA_DIR") == "/before/data"));
CHECK((yams::config::getenv_copy("YAMS_CONFIG") == "/before/config"));
CHECK((yams::config::getenv_copy("YAMS_CONFIG_PATH") == "/before/config-path"));
CHECK((yams::config::getenv_copy("YAMS_DAEMON_SOCKET") == "/before/socket"));
}
TEST_CASE_METHOD(DaemonFixture, "Daemon tuning reload revokes removed overrides coherently",
"[daemon][tuning][reload]") {
SKIP_ON_WINDOWS();
yams::test::ScopedEnvVar ipcCompatibility{"YAMS_IPC_TIMEOUT_MS", std::nullopt};
yams::test::ScopedEnvVar admissionCompatibility{"YAMS_ADMISSION_CONTROL", std::nullopt};
yams::test::ScopedEnvVar memoryCompatibility{"YAMS_MEMORY_WARNING_PCT", std::nullopt};
yams::test::ScopedEnvVar postIngestCompatibility{"YAMS_POST_INGEST_RPC_QUEUE_MAX",
std::nullopt};
const auto configPath = runtime_root_ / "config.toml";
const auto writeOverrides = [&](bool enabled) {
std::ofstream out(configPath, std::ios::trunc);
REQUIRE(out.is_open());
if (!enabled) {
out << "# runtime tuning overrides removed\n";
return;
}
out << "[tuning]\n";
out << "target_cpu_percent = 321\n";
out << "[tuning.ipc]\n";
out << "timeout_ms = 4321\n";
out << "[tuning.resource]\n";
out << "admission_control = false\n";
out << "memory_warning_threshold = 0.91\n";
out << "[tuning.post_ingest]\n";
out << "rpc_queue_max = 333\n";
};
writeOverrides(true);
config_.configFilePath = configPath;
daemon_ = std::make_unique<YamsDaemon>(config_);
REQUIRE(daemon_ != nullptr);
auto activeTuning = daemon_->serviceManager_->getTuningConfig();
activeTuning.topologyAlgorithm = "exact";
daemon_->serviceManager_->setTuningConfig(activeTuning);
daemon_->reloadTuningConfig();
CHECK((daemon_->config_.tuning.targetCpuPercent == 321u));
CHECK((TuneAdvisor::ipcTimeoutMs() == 4321u));
CHECK_FALSE(TuneAdvisor::enableAdmissionControl());
CHECK((TuneAdvisor::memoryWarningThreshold() == 0.91));
CHECK((TuneAdvisor::postIngestRpcQueueMax() == 333u));
CHECK((daemon_->config_.tuning.provenance.at("tuning.ipc.timeout_ms") ==
"config:tuning.ipc.timeout_ms"));
writeOverrides(false);
const auto versionBeforeRemoval = TuneAdvisor::configuredOverridesVersion();
daemon_->reloadTuningConfig();
CHECK((TuneAdvisor::configuredOverridesVersion() == versionBeforeRemoval + 2));
CHECK((daemon_->config_.tuning.targetCpuPercent == 200u));
CHECK((daemon_->config_.tuning.topologyAlgorithm == "exact"));
CHECK(daemon_->config_.tuning.provenance.empty());
CHECK((TuneAdvisor::ipcTimeoutMs() == 15000u));
CHECK(TuneAdvisor::enableAdmissionControl());
CHECK((TuneAdvisor::memoryWarningThreshold() == 0.75));
CHECK((TuneAdvisor::postIngestRpcQueueMax() == 256u));
const auto status = daemon_->serviceManager_->getRuntimeTuningStatus();
CHECK((status.at("ipc.timeout_ms") == "15000"));
CHECK((status.at("ipc.timeout_ms.source") == "default"));
CHECK((status.at("resource.admission_control") == "1"));
CHECK((status.at("resource.admission_control.source") == "default"));
CHECK((status.at("resource.memory_warning_threshold") == "0.750000"));
CHECK((status.at("resource.memory_warning_threshold.source") == "default"));
CHECK((status.at("post_ingest.rpc_queue_max") == "256"));
CHECK((status.at("post_ingest.rpc_queue_max.source") == "default"));
}
TEST_CASE_METHOD(DaemonFixture, "Daemon tuning reload rejects multiple active lifecycles",
"[daemon][tuning][reload][lifecycle]") {
SKIP_ON_WINDOWS();
yams::test::ScopedEnvVar ipcCompatibility{"YAMS_IPC_TIMEOUT_MS", std::nullopt};
const auto configPath = runtime_root_ / "config.toml";
config_.configFilePath = configPath;
daemon_ = std::make_unique<YamsDaemon>(config_);
REQUIRE(daemon_ != nullptr);
REQUIRE((TuneAdvisor::ipcTimeoutMs() == 15000u));
auto nestedConfig = config_;
nestedConfig.dataDir = runtime_root_ / "nested-data";
nestedConfig.socketPath = runtime_root_ / "nested.sock";
nestedConfig.pidFile = runtime_root_ / "nested.pid";
nestedConfig.logFile = runtime_root_ / "nested.log";
YamsDaemon nested(nestedConfig);
{
std::ofstream out(configPath, std::ios::trunc);
REQUIRE(out.is_open());
out << "[tuning.ipc]\n";
out << "timeout_ms = 4321\n";
}
const auto versionBefore = TuneAdvisor::configuredOverridesVersion();
daemon_->reloadTuningConfig();
CHECK((TuneAdvisor::configuredOverridesVersion() == versionBefore));
CHECK((TuneAdvisor::ipcTimeoutMs() == 15000u));
CHECK((daemon_->config_.tuning.provenance.count("tuning.ipc.timeout_ms") == 0));
CHECK((daemon_->serviceManager_->getRuntimeTuningStatus().at("ipc.timeout_ms") == "15000"));
}
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
// Pre-compute fork-safe inputs in the parent: the child must not allocate
// (a fork in a multithreaded process can leave the malloc lock held by a
// vanished thread), so build the lock path here and format the payload with
// snprintf into a stack buffer in the child.
const std::string lockPathStr = (config_.dataDir / ".yams-lock").string();
int readyPipe[2] = {-1, -1};
REQUIRE(::pipe(readyPipe) == 0);
const pid_t child = ::fork();
REQUIRE(child >= 0);
if (child == 0) {
::close(readyPipe[0]);
const int lockFd = ::open(lockPathStr.c_str(), O_CREAT | O_RDWR, 0644);
if (lockFd < 0 || ::flock(lockFd, LOCK_EX) != 0) {
const char failed = '2';
(void)::write(readyPipe[1], &failed, 1);
_exit(2);
}
char payload[128];
const int payloadLen = ::snprintf(payload, sizeof(payload),
"{\"pid\":%d,\"socket\":\"/tmp/not-a-daemon.sock\"}",
static_cast<int>(::getpid()));
if (payloadLen > 0 && static_cast<size_t>(payloadLen) < sizeof(payload)) {
(void)::ftruncate(lockFd, 0);
(void)::write(lockFd, payload, static_cast<size_t>(payloadLen));
}
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;
ssize_t readyRead = -1;
do {
readyRead = ::read(readyPipe[0], &ready, 1);
} while (readyRead < 0 && errno == EINTR);
INFO("lock holder child status byte=" << static_cast<int>(ready) << " read=" << readyRead);
REQUIRE(readyRead == 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 stop shuts down its services after a failed second start",
"[daemon][lifecycle][concurrency]") {
SKIP_ON_WINDOWS();
std::weak_ptr<ServiceManager> ownerServices;
{
YamsDaemon owner(config_);
auto started = owner.start();
if (!started && isSocketPermissionDenied(started.error())) {
SKIP("UNIX domain sockets not permitted");
}
REQUIRE(started);
ownerServices = owner.getServiceManager()->weak_from_this();
{
// The second daemon loses the instance lock. Destroying its never-started
// ServiceManager runs shutdown(), which drives the process-wide service FSM to
// Stopped. The owner must still run its own shutdown afterwards.
YamsDaemon loser(config_);
CHECK_FALSE(loser.start());
}
REQUIRE(owner.stop());
}
// A completed shutdown releases every service-graph reference, so the daemon's destructor
// drops the last one. A skipped shutdown leaves background coroutines holding it.
CHECK(ownerServices.expired());
}
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
Reference in a new issue View git blame Copy permalink