yams/tests/integration/daemon/test_daemon_harness.h

538 lines
24 KiB
C++

// Lightweight RAII harness to start/stop a YamsDaemon for integration tests
#pragma once
#include <spdlog/sinks/stdout_color_sinks.h>
#include <spdlog/spdlog.h>
#include <cstdio>
#include <cstdlib>
#include <filesystem>
#include <functional>
#include <map>
#include <optional>
#include <random>
#include <string>
#include <thread>
#include <vector>
#include "test_async_helpers.h"
#include <yams/compat/unistd.h>
#include <yams/daemon/client/asio_connection_pool.h>
#include <yams/daemon/client/daemon_client.h>
#include <yams/daemon/client/global_io_context.h>
#include <yams/daemon/daemon.h>
#include <boost/asio/io_context.hpp>
#include <boost/asio/local/stream_protocol.hpp>
// Windows daemon IPC tests are currently unstable due to socket shutdown race conditions
// The daemon's connection handler coroutines crash during cleanup when sockets are forcibly closed
// See: docs/developer/windows-daemon-ipc-plan.md
#ifdef _WIN32
#define SKIP_DAEMON_TEST_ON_WINDOWS() \
SKIP("Daemon IPC tests unstable on Windows - see windows-daemon-ipc-plan.md")
#else
#define SKIP_DAEMON_TEST_ON_WINDOWS() ((void)0)
#endif
namespace yams::test {
// Options struct must be defined outside DaemonHarness to avoid Clang issue with
// default member initializers in nested structs used as default function arguments.
// See: https://bugs.llvm.org/show_bug.cgi?id=36684
struct DaemonHarnessOptions {
bool enableModelProvider = true;
bool useMockModelProvider = true;
bool autoLoadPlugins = false;
bool pluginDirStrict = false;
bool enableAutoRepair = true;
bool isolateState = false;
bool skipSocketVerificationOnReady = false;
bool configureModelPool = false;
bool modelPoolLazyLoading = true;
std::vector<std::string> preloadModels = {};
std::optional<std::filesystem::path> pluginDir = std::nullopt;
// Per-plugin configuration: plugin name -> JSON config string
std::map<std::string, std::string> pluginConfigs = {};
// Override the data directory instead of creating a temp one.
// When set, the harness uses this existing path (e.g. a real corpus)
// and skips cleanup of the data dir on destruction.
std::optional<std::filesystem::path> dataDir = std::nullopt;
// Optional fixture content to seed `dataDir` with. When both are set
// and `dataDir` exists but is empty (or doesn't exist), the harness
// recursively copies the fixture's contents into `dataDir` before
// starting the daemon. If `dataDir` is non-empty, no copy is performed
// (lets subsequent spawns reuse the seeded scratch dir, preserving
// tuner state / topology artifacts across runs). Useful for benches
// that want a writable scratch seeded from a frozen fixture.
std::optional<std::filesystem::path> cloneFixtureFrom = std::nullopt;
// Optional explicit config.toml path for daemon startup.
// When set, the harness points daemon config resolution at this file.
std::optional<std::filesystem::path> configPath = std::nullopt;
// Additional trusted plugin search paths (appended to DaemonConfig::trustedPluginPaths)
std::vector<std::filesystem::path> trustedPluginPaths = {};
// Optional test hook to mutate the daemon config before construction.
std::function<void(yams::daemon::DaemonConfig&)> configureDaemon;
};
class DaemonHarness {
public:
using Options = DaemonHarnessOptions;
explicit DaemonHarness(Options options = Options()) : options_(std::move(options)) {
namespace fs = std::filesystem;
// Ensure mock provider and disable ABI plugins for stability
// Create temp working dir
auto id = random_id();
root_ = fs::temp_directory_path() / (std::string("yams_it_") + id);
fs::create_directories(root_);
// Use caller-provided dataDir if set (e.g. existing corpus), else create temp
if (options_.dataDir) {
data_ = *options_.dataDir;
externalDataDir_ = true; // skip cleanup of external data
fs::create_directories(data_);
} else {
data_ = root_ / "data";
fs::create_directories(data_);
}
// Seed scratch from a fixture when requested AND the scratch is empty.
// Subsequent spawns reuse the existing content (preserves tuner state,
// topology artifacts, etc. across reps).
if (options_.cloneFixtureFrom) {
std::error_code ec;
const auto& src = *options_.cloneFixtureFrom;
bool dataDirEmpty = true;
if (fs::exists(data_)) {
for (auto it = fs::directory_iterator(data_, ec);
!ec && it != fs::directory_iterator(); ++it) {
dataDirEmpty = false;
break;
}
}
if (dataDirEmpty && fs::exists(src, ec)) {
fs::copy(src, data_,
fs::copy_options::recursive | fs::copy_options::copy_symlinks |
fs::copy_options::skip_existing,
ec);
if (ec) {
spdlog::warn("[DaemonHarness] cloneFixtureFrom copy failed: {} ({} -> {})",
ec.message(), src.string(), data_.string());
} else {
spdlog::info("[DaemonHarness] seeded {} from fixture {}", data_.string(),
src.string());
}
}
}
// Use platform-appropriate temp path for socket
// On Windows, AF_UNIX sockets work but need a valid Windows path
// On Unix, use /tmp for short paths to avoid AF_UNIX length limits
#ifdef _WIN32
sock_ = fs::temp_directory_path() / ("daemon_" + id + ".sock");
#else
sock_ = std::filesystem::path("/tmp") / ("daemon_" + id + ".sock");
#endif
pid_ = root_ / ("daemon_" + id + ".pid");
log_ = root_ / ("daemon_" + id + ".log");
// NOTE: Do NOT create daemon instance in constructor!
// Creating YamsDaemon initializes ServiceManager which spawns worker threads.
// If start() is never called but destructor runs, thread cleanup can crash.
// Defer daemon creation until start() is explicitly called.
}
~DaemonHarness() {
// Explicitly stop daemon before cleanup to avoid crashes
stop();
cleanup();
}
// Pre-runLoop callback type: called after daemon->start() succeeds but before runLoop thread
// starts. Use this to set signal check hooks or external shutdown flags BEFORE the runLoop
// begins reading them, avoiding race conditions.
using PreRunLoopCallback = std::function<void(yams::daemon::YamsDaemon*)>;
bool start(std::chrono::milliseconds timeout = std::chrono::seconds(10),
PreRunLoopCallback preRunLoopCallback = nullptr) {
ensureDefaultLogger();
// Always create new daemon instance on start()
// (daemon cannot be restarted after stop() - must create new instance)
spdlog::info("[DaemonHarness] Creating new daemon instance...");
if (options_.isolateState) {
originalCwd_ = std::filesystem::current_path();
std::filesystem::current_path(root_);
originalXdgState_ = std::getenv("XDG_STATE_HOME") ? std::getenv("XDG_STATE_HOME") : "";
setenv("XDG_STATE_HOME", (root_ / "state").string().c_str(), 1);
std::filesystem::create_directories(root_ / "state" / "yams" / "sessions");
isolateStateActive_ = true;
}
yams::daemon::DaemonConfig cfg;
if (options_.configPath) {
previousYamsConfig_ = std::getenv("YAMS_CONFIG") ? std::getenv("YAMS_CONFIG") : "";
hadYamsConfig_ = std::getenv("YAMS_CONFIG") != nullptr;
setenv("YAMS_CONFIG", options_.configPath->string().c_str(), 1);
configEnvOwned_ = true;
cfg.configFilePath = *options_.configPath;
}
cfg.dataDir = data_;
cfg.socketPath = sock_;
cfg.pidFile = pid_;
cfg.logFile = log_;
cfg.enableModelProvider = options_.enableModelProvider;
cfg.autoLoadPlugins = options_.autoLoadPlugins;
cfg.pluginDirStrict = options_.pluginDirStrict;
cfg.useMockModelProvider = options_.useMockModelProvider;
cfg.enableAutoRepair = options_.enableAutoRepair;
if (options_.pluginDir) {
cfg.pluginDir = *options_.pluginDir;
}
if (options_.configureModelPool) {
cfg.modelPoolConfig.lazyLoading = options_.modelPoolLazyLoading;
cfg.modelPoolConfig.preloadModels = options_.preloadModels;
}
if (options_.configureDaemon) {
options_.configureDaemon(cfg);
}
// Pass plugin-specific configurations
cfg.pluginConfigs = options_.pluginConfigs;
// Pass trusted plugin paths
cfg.trustedPluginPaths = options_.trustedPluginPaths;
daemon_ = std::make_unique<yams::daemon::YamsDaemon>(cfg);
spdlog::info("[DaemonHarness] Daemon instance created");
spdlog::info("[DaemonHarness] Calling daemon_->start()...");
auto s = daemon_->start();
if (!s) {
spdlog::error("[DaemonHarness] daemon_->start() failed!");
return false;
}
spdlog::info("[DaemonHarness] daemon_->start() succeeded, starting runLoop thread...");
// Call pre-runLoop callback if provided (for setting up signal hooks, etc.)
// This runs BEFORE the runLoop thread starts, avoiding race conditions.
if (preRunLoopCallback) {
spdlog::info("[DaemonHarness] Invoking pre-runLoop callback...");
preRunLoopCallback(daemon_.get());
}
#ifdef YAMS_TESTING
if (!preRunLoopCallback && !options_.enableAutoRepair) {
daemon_->testingStartAsyncInitWithoutRunLoop();
spdlog::info("[DaemonHarness] Async init triggered without runLoop thread, polling for "
"daemon Ready state...");
} else
#endif
{
// Start runLoop in background thread - this triggers async initialization
// Without runLoop(), ServiceManager::startAsyncInit() is never called
runLoopThread_ = std::thread([this]() {
daemon_->runLoop();
auto stopResult = daemon_->stop();
if (!stopResult && daemon_->isRunning()) {
spdlog::warn("[DaemonHarness] Background stop after runLoop returned error: {}",
stopResult.error().message);
}
});
// Give runLoop time to enter and trigger async init
std::this_thread::sleep_for(std::chrono::milliseconds(50));
spdlog::info(
"[DaemonHarness] runLoop thread started, polling for daemon Ready state...");
}
// Poll for daemon to reach a usable lifecycle state in lifecycle FSM.
// Ready or Degraded both indicate initialization completed enough for IPC tests.
auto deadline = std::chrono::steady_clock::now() + timeout;
bool lifecycleReady = false;
while (std::chrono::steady_clock::now() < deadline) {
std::this_thread::sleep_for(std::chrono::milliseconds(100));
// Check lifecycle state first
auto lifecycle = daemon_->getLifecycle().snapshot();
if (lifecycle.state == yams::daemon::LifecycleState::Ready ||
lifecycle.state == yams::daemon::LifecycleState::Degraded) {
lifecycleReady = true;
spdlog::info("[DaemonHarness] Daemon lifecycle reached usable state: {}",
static_cast<int>(lifecycle.state));
break;
} else if (lifecycle.state == yams::daemon::LifecycleState::Failed) {
spdlog::error("[DaemonHarness] Daemon lifecycle reached Failed state: {}",
lifecycle.lastError);
return false;
}
}
if (!lifecycleReady) {
auto lifecycle = daemon_->getLifecycle().snapshot();
spdlog::error(
"[DaemonHarness] Timeout waiting for Ready state after {}ms (current state: {})",
timeout.count(), static_cast<int>(lifecycle.state));
return false;
}
// Verify socket connectivity as final sanity check
// Use simple socket connection instead of DaemonClient to avoid polluting
// the shared connection pool (DaemonClient destructor shuts down shared pools)
if (options_.skipSocketVerificationOnReady) {
spdlog::info(
"[DaemonHarness] Skipping socket verification after Ready (test override)");
return true;
}
spdlog::info("[DaemonHarness] Verifying socket connectivity at: {}", sock_.string());
bool socketOk = verifySocketConnectivity(sock_, std::chrono::seconds(5));
if (socketOk) {
spdlog::info("[DaemonHarness] Socket connection verified, daemon fully ready at: {}",
sock_.string());
return true;
}
spdlog::error("[DaemonHarness] Daemon Ready but socket connection failed at: {}",
sock_.string());
return false;
}
void stop() {
if (std::getenv("YAMS_TRACE_HARNESS_LIFETIME")) {
std::fprintf(stderr, "[DaemonHarness::stop] enter daemon=%p runLoopJoinable=%d\n",
static_cast<void*>(daemon_.get()), runLoopThread_.joinable() ? 1 : 0);
std::fflush(stderr);
}
if (daemon_) {
spdlog::info("[DaemonHarness] Stopping daemon (running={})...", daemon_->isRunning());
if (runLoopThread_.joinable()) {
spdlog::info("[DaemonHarness] Requesting runLoop stop before join...");
daemon_->requestStop();
spdlog::info("[DaemonHarness] Joining runLoop thread...");
runLoopThread_.join();
spdlog::info("[DaemonHarness] runLoop thread joined");
}
// Wait for daemon to fully stop - the background runLoop thread mirrors
// daemon_main by calling stop() after runLoop() returns on remote shutdown.
if (daemon_->isRunning() ||
!daemon_->waitForStopCompletion(std::chrono::milliseconds(0))) {
auto stopResult = daemon_->stop();
if (!stopResult) {
spdlog::warn("[DaemonHarness] Daemon stop returned error: {}",
stopResult.error().message);
}
}
// Wait for daemon to report it's no longer running
int isRunningRetries = 0;
while (daemon_->isRunning() && isRunningRetries < 50) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
isRunningRetries++;
}
if (isRunningRetries > 0) {
spdlog::info("[DaemonHarness] Waited {}ms for isRunning() to become false",
isRunningRetries * 10);
}
spdlog::info("[DaemonHarness] Daemon stopped (running={}), resetting instance...",
daemon_->isRunning());
// Reset GlobalIOContext to clean up threads and io_context state
// This is critical for test isolation - without it, threads accumulate across test
// cases Temporarily unset YAMS_TESTING to allow reset() to actually work
const char* yams_testing_env = std::getenv("YAMS_TESTING");
const char* yams_safe_env = std::getenv("YAMS_TEST_SAFE_SINGLE_INSTANCE");
std::string yams_testing = yams_testing_env ? yams_testing_env : "";
std::string yams_safe = yams_safe_env ? yams_safe_env : "";
if (yams_testing_env) {
unsetenv("YAMS_TESTING");
}
if (yams_safe_env) {
unsetenv("YAMS_TEST_SAFE_SINGLE_INSTANCE");
}
// Reset the global client IO state BEFORE restarting threads.
// This closes ConnectionRegistry sockets and clears pools while the
// existing io_context is still alive, preventing stale client FDs
// from accumulating across harness cycles.
yams::daemon::GlobalIOContext::reset();
spdlog::info("[DaemonHarness] GlobalIOContext reset complete");
// Windows needs additional time before GlobalIOContext::reset()
// Windows thread cleanup is slower than Unix
#ifdef _WIN32
std::this_thread::sleep_for(std::chrono::milliseconds(500));
// Skip GlobalIOContext operations on Windows for now - causes SIGSEGV
// The io_context threads will be cleaned up when the process exits
spdlog::info("[DaemonHarness] Skipping GlobalIOContext::restart() on Windows");
#else
spdlog::info("[DaemonHarness] GlobalIOContext restart deferred");
#endif
// Restore environment variables
if (yams_testing_env) {
setenv("YAMS_TESTING", yams_testing.c_str(), 1);
}
if (yams_safe_env) {
setenv("YAMS_TEST_SAFE_SINGLE_INSTANCE", yams_safe.c_str(), 1);
}
if (isolateStateActive_) {
std::error_code ec;
if (!originalCwd_.empty()) {
std::filesystem::current_path(originalCwd_, ec);
}
if (originalXdgState_.empty()) {
unsetenv("XDG_STATE_HOME");
} else {
setenv("XDG_STATE_HOME", originalXdgState_.c_str(), 1);
}
isolateStateActive_ = false;
}
if (configEnvOwned_) {
if (hadYamsConfig_) {
setenv("YAMS_CONFIG", previousYamsConfig_.c_str(), 1);
} else {
unsetenv("YAMS_CONFIG");
}
configEnvOwned_ = false;
hadYamsConfig_ = false;
previousYamsConfig_.clear();
}
// Reset daemon so it can be recreated on next start()
daemon_.reset();
if (std::getenv("YAMS_TRACE_HARNESS_LIFETIME")) {
std::fprintf(stderr, "[DaemonHarness::stop] exit daemon=%p runLoopJoinable=%d\n",
static_cast<void*>(daemon_.get()), runLoopThread_.joinable() ? 1 : 0);
std::fflush(stderr);
}
// CRITICAL: Allow OS to fully release thread resources (macOS needs this)
// Each daemon creates ~48 threads (32 SocketServer + 16 ServiceManager).
// macOS has strict per-process thread creation rate limits and needs time
// to reclaim thread resources before next daemon starts creating threads.
// Testing shows: 250ms → crashes at daemon #6, 500ms allows ~10 daemons, 750ms more
// conservative. Increased to 750ms for improved stability on resource-constrained CI
// runners.
std::this_thread::sleep_for(std::chrono::milliseconds(
750)); // Wait for socket file to be removed by daemon shutdown
int socketRetries = 0;
while (std::filesystem::exists(sock_) && socketRetries < 50) {
std::this_thread::sleep_for(std::chrono::milliseconds(20));
socketRetries++;
}
if (socketRetries > 0) {
spdlog::info("[DaemonHarness] Waited {}ms for socket file removal",
socketRetries * 20);
}
// Verify socket is truly gone
if (std::filesystem::exists(sock_)) {
spdlog::warn("[DaemonHarness] Socket file still exists after {}ms",
socketRetries * 20);
}
// Additional brief wait to ensure OS fully releases socket
std::this_thread::sleep_for(std::chrono::milliseconds(100));
spdlog::info(
"[DaemonHarness] Stop complete (isRunning check: {}ms, socket cleanup: {}ms)",
isRunningRetries * 10, socketRetries * 20);
}
}
const std::filesystem::path& socketPath() const { return sock_; }
const std::filesystem::path& dataDir() const { return data_; }
const std::filesystem::path& rootDir() const { return root_; }
yams::daemon::YamsDaemon* daemon() const { return daemon_.get(); }
private:
// Install a default spdlog logger once per process if none exists, so plugins
// and harness code that call spdlog::info(...) don't dereference a null default
// logger. Mirrors daemon_main's init order (set_default_logger before plugin load).
static void ensureDefaultLogger() {
static const bool once = [] {
if (!spdlog::default_logger()) {
try {
auto sink = std::make_shared<spdlog::sinks::stdout_color_sink_mt>();
auto logger = std::make_shared<spdlog::logger>("yams-test", sink);
spdlog::set_default_logger(logger);
} catch (...) {
}
}
return true;
}();
(void)once;
}
static std::string random_id() {
static const char* cs = "abcdefghijklmnopqrstuvwxyz0123456789";
thread_local std::mt19937_64 rng{std::random_device{}()};
std::uniform_int_distribution<size_t> dist(0, 35);
std::string out;
out.reserve(8);
for (int i = 0; i < 8; ++i)
out.push_back(cs[dist(rng)]);
return out;
}
// Simple socket connectivity check that doesn't use the shared connection pool
// This avoids DaemonClient's destructor from shutting down the pool before
// the actual test fixture's client can use it
static bool verifySocketConnectivity(const std::filesystem::path& socketPath,
std::chrono::milliseconds timeout) {
auto deadline = std::chrono::steady_clock::now() + timeout;
while (std::chrono::steady_clock::now() < deadline) {
try {
// Create fresh socket for each attempt (can't reuse after failed connect)
boost::asio::io_context io;
boost::asio::local::stream_protocol::socket sock(io);
boost::system::error_code ec;
sock.connect(boost::asio::local::stream_protocol::endpoint(socketPath.string()),
ec);
if (!ec) {
sock.close();
return true;
}
} catch (const std::exception& e) {
spdlog::debug("[DaemonHarness] Socket verification attempt failed: {}", e.what());
}
std::this_thread::sleep_for(std::chrono::milliseconds(50));
}
spdlog::error("[DaemonHarness] Socket verification timed out after {}ms", timeout.count());
return false;
}
void cleanup() {
namespace fs = std::filesystem;
std::error_code ec;
if (externalDataDir_) {
// Only remove temp root (socket, pid, log) — preserve external data dir
if (!root_.empty()) {
// Remove everything in root_ except data_ (which is external)
for (auto& entry : fs::directory_iterator(root_, ec)) {
fs::remove_all(entry.path(), ec);
}
fs::remove(root_, ec);
}
} else {
if (!root_.empty())
fs::remove_all(root_, ec);
}
}
std::unique_ptr<yams::daemon::YamsDaemon> daemon_;
std::thread runLoopThread_; // Background thread for runLoop
std::filesystem::path root_, data_, sock_, pid_, log_;
std::filesystem::path originalCwd_;
std::string originalXdgState_;
std::string previousYamsConfig_;
bool isolateStateActive_ = false;
bool externalDataDir_ = false;
bool configEnvOwned_ = false;
bool hadYamsConfig_ = false;
Options options_;
};
} // namespace yams::test