569 lines
20 KiB
C++
569 lines
20 KiB
C++
// Daemon components test suite (Catch2)
|
|
// Consolidates: SocketServer lifecycle, socket utils, resource pool
|
|
// Covers: IPC infrastructure, socket path resolution, resource lifecycle management
|
|
|
|
#include <catch2/catch_test_macros.hpp>
|
|
#include <catch2/generators/catch_generators.hpp>
|
|
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <cstdlib>
|
|
#include <filesystem>
|
|
#include <fstream>
|
|
#include <iostream>
|
|
#include <map>
|
|
#include <optional>
|
|
#include <set>
|
|
#include <thread>
|
|
#include "../../common/test_helpers_catch2.h"
|
|
#include <yams/compat/unistd.h>
|
|
|
|
#include <yams/daemon/components/IOCoordinator.h>
|
|
#include <yams/daemon/components/ResourceGovernor.h>
|
|
#include <yams/daemon/components/SocketServer.h>
|
|
#include <yams/daemon/components/StateComponent.h>
|
|
#include <yams/daemon/components/WorkCoordinator.h>
|
|
#include <yams/daemon/ipc/socket_utils.h>
|
|
#include <yams/daemon/resource/resource_pool.h>
|
|
|
|
#ifdef _WIN32
|
|
inline int geteuid() {
|
|
return 1000;
|
|
}
|
|
inline int getuid() {
|
|
return 1000;
|
|
}
|
|
#endif
|
|
|
|
using namespace yams::daemon;
|
|
using namespace std::chrono_literals;
|
|
namespace fs = std::filesystem;
|
|
|
|
#ifndef _WIN32
|
|
#include <sys/un.h>
|
|
#endif
|
|
|
|
// =============================================================================
|
|
// Test Helpers
|
|
// =============================================================================
|
|
|
|
namespace {
|
|
fs::path makeTempRuntimeDir(const std::string& name) {
|
|
auto base = fs::temp_directory_path();
|
|
#ifndef _WIN32
|
|
constexpr std::size_t maxUnixPath = sizeof(sockaddr_un::sun_path) - 1;
|
|
auto candidate = base / "yams-component-tests" / name;
|
|
auto candidateWithSocket = candidate / "ipc.sock";
|
|
std::cerr << "candidate path: " << candidateWithSocket
|
|
<< " len=" << candidateWithSocket.native().size() << " limit=" << maxUnixPath << "\n";
|
|
if (candidateWithSocket.native().size() >= maxUnixPath) {
|
|
base = fs::path("/tmp");
|
|
candidate = base / "yams-component-tests" / name;
|
|
candidateWithSocket = candidate / "ipc.sock";
|
|
std::cerr << "fallback path: " << candidateWithSocket
|
|
<< " len=" << candidateWithSocket.native().size() << "\n";
|
|
}
|
|
#endif
|
|
auto dir = base / "yams-component-tests" / name;
|
|
std::error_code ec;
|
|
fs::create_directories(dir, ec);
|
|
return dir;
|
|
}
|
|
|
|
std::string randomSuffix() {
|
|
return std::to_string(std::chrono::steady_clock::now().time_since_epoch().count());
|
|
}
|
|
|
|
template <typename ResultLike> bool isPermissionDenied(const ResultLike& result) {
|
|
if (result) {
|
|
return false;
|
|
}
|
|
std::string_view message{result.error().message};
|
|
return message.find("Operation not permitted") != std::string_view::npos ||
|
|
message.find("Permission denied") != std::string_view::npos;
|
|
}
|
|
|
|
// Test resource for pool testing
|
|
class TestResource {
|
|
public:
|
|
explicit TestResource(int id) : id_(id), valid_(true) { creationCount_++; }
|
|
|
|
~TestResource() { destructionCount_++; }
|
|
|
|
int getId() const { return id_; }
|
|
bool isValid() const { return valid_; }
|
|
void invalidate() { valid_ = false; }
|
|
|
|
void doWork() const {
|
|
workCount_++;
|
|
std::this_thread::sleep_for(1ms);
|
|
}
|
|
|
|
static void resetCounters() {
|
|
creationCount_ = 0;
|
|
destructionCount_ = 0;
|
|
workCount_ = 0;
|
|
}
|
|
|
|
static int getCreationCount() { return creationCount_; }
|
|
static int getDestructionCount() { return destructionCount_; }
|
|
static int getWorkCount() { return workCount_; }
|
|
|
|
private:
|
|
int id_;
|
|
bool valid_;
|
|
static std::atomic<int> creationCount_;
|
|
static std::atomic<int> destructionCount_;
|
|
static std::atomic<int> workCount_;
|
|
};
|
|
|
|
std::atomic<int> TestResource::creationCount_{0};
|
|
std::atomic<int> TestResource::destructionCount_{0};
|
|
std::atomic<int> TestResource::workCount_{0};
|
|
|
|
using yams::test::ScopedEnvVar;
|
|
} // namespace
|
|
|
|
// =============================================================================
|
|
// SocketServer Lifecycle Tests
|
|
// =============================================================================
|
|
|
|
TEST_CASE("SocketServer: Lifecycle management", "[daemon][components][socket]") {
|
|
auto runtimeDir = makeTempRuntimeDir("socket-lifecycle-" + randomSuffix());
|
|
auto socketPath = runtimeDir / "ipc.sock";
|
|
|
|
SocketServer::Config config;
|
|
config.socketPath = socketPath;
|
|
config.connectionTimeout = 1500ms;
|
|
|
|
StateComponent state;
|
|
state.stats.startTime = std::chrono::steady_clock::now();
|
|
|
|
SECTION("Start and stop clears stopping flag") {
|
|
WorkCoordinator coordinator;
|
|
coordinator.start(2);
|
|
IOCoordinator io;
|
|
io.start();
|
|
SocketServer server(config, &io, &coordinator, nullptr, &state);
|
|
|
|
auto first = server.start();
|
|
if (!first) {
|
|
UNSCOPED_INFO(first.error().message);
|
|
}
|
|
if (!first && isPermissionDenied(first)) {
|
|
SKIP("UNIX domain sockets not permitted on this system");
|
|
}
|
|
REQUIRE(first);
|
|
|
|
auto stopped = server.stop();
|
|
REQUIRE(stopped);
|
|
|
|
// Second start should work (stopping flag cleared)
|
|
auto second = server.start();
|
|
if (!second) {
|
|
UNSCOPED_INFO(second.error().message);
|
|
}
|
|
if (!second && isPermissionDenied(second)) {
|
|
SKIP("UNIX domain sockets not permitted on this system");
|
|
}
|
|
REQUIRE(second);
|
|
|
|
REQUIRE(server.stop());
|
|
|
|
io.stop();
|
|
io.join();
|
|
coordinator.stop();
|
|
coordinator.join();
|
|
}
|
|
|
|
SECTION("Multiple start attempts are safe") {
|
|
WorkCoordinator coordinator;
|
|
coordinator.start(2);
|
|
IOCoordinator io;
|
|
io.start();
|
|
SocketServer server(config, &io, &coordinator, nullptr, &state);
|
|
|
|
auto first = server.start();
|
|
if (!first) {
|
|
UNSCOPED_INFO(first.error().message);
|
|
}
|
|
if (!first && isPermissionDenied(first)) {
|
|
SKIP("UNIX domain sockets not permitted on this system");
|
|
}
|
|
REQUIRE(first);
|
|
|
|
// Second start while running should fail gracefully
|
|
auto second = server.start();
|
|
REQUIRE(!second); // Already running
|
|
|
|
REQUIRE(server.stop());
|
|
|
|
io.stop();
|
|
io.join();
|
|
coordinator.stop();
|
|
coordinator.join();
|
|
}
|
|
|
|
// Cleanup
|
|
std::error_code ec;
|
|
fs::remove_all(runtimeDir, ec);
|
|
}
|
|
|
|
TEST_CASE("SocketServer: main-socket admission ignores proxy connection load",
|
|
"[daemon][components][socket][admission]") {
|
|
SocketServer::Config config;
|
|
config.socketPath = "/tmp/yams-socket-admission-test.sock";
|
|
config.maxConnections = 4;
|
|
config.proxyMaxConnections = 64;
|
|
|
|
StateComponent state;
|
|
SocketServer server(config, nullptr, nullptr, nullptr, &state);
|
|
REQUIRE(server.resizeConnectionSlots(config.maxConnections));
|
|
|
|
AddDocumentRequest addRequest;
|
|
addRequest.path = "src";
|
|
addRequest.recursive = true;
|
|
|
|
SECTION("Main-socket write admission keys off main connections, not total connections") {
|
|
server.testing_setConnectionCounts(/*mainActive=*/0, /*proxyActive=*/10);
|
|
|
|
CHECK(server.activeConnections() == 10);
|
|
CHECK(server.proxyActiveConnections() == 10);
|
|
CHECK(server.testing_mainSocketAdmissionVerdict(addRequest) ==
|
|
SocketAdmissionVerdict::admit);
|
|
CHECK_FALSE(server.testing_mainSocketEmergencyGuardRejects());
|
|
CHECK(server.getSlotUtilization() == 0.0);
|
|
server.testing_setConnectionCounts(0, 0);
|
|
}
|
|
|
|
SECTION("Main-socket pressure still rejects once main connections cross the limit bands") {
|
|
server.testing_setConnectionCounts(/*mainActive=*/10, /*proxyActive=*/0);
|
|
|
|
CHECK(server.testing_mainSocketAdmissionVerdict(addRequest) ==
|
|
SocketAdmissionVerdict::reject);
|
|
server.testing_setConnectionCounts(0, 0);
|
|
}
|
|
|
|
SECTION("Control requests remain bypassed on the main socket") {
|
|
server.testing_setConnectionCounts(/*mainActive=*/10, /*proxyActive=*/20);
|
|
|
|
CHECK(server.testing_mainSocketAdmissionVerdict(StatusRequest{}) ==
|
|
SocketAdmissionVerdict::bypass);
|
|
server.testing_setConnectionCounts(0, 0);
|
|
}
|
|
|
|
SECTION("Emergency guard is driven only by main active connections") {
|
|
const auto hardLimit = AdmissionPolicy::emergencySessionLimit(config.maxConnections);
|
|
|
|
server.testing_setConnectionCounts(/*mainActive=*/hardLimit - 1, /*proxyActive=*/hardLimit);
|
|
CHECK_FALSE(server.testing_mainSocketEmergencyGuardRejects());
|
|
|
|
server.testing_setConnectionCounts(/*mainActive=*/hardLimit, /*proxyActive=*/0);
|
|
CHECK(server.testing_mainSocketEmergencyGuardRejects());
|
|
server.testing_setConnectionCounts(0, 0);
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// Socket Path Resolution Tests
|
|
// =============================================================================
|
|
|
|
TEST_CASE("Socket path resolution: Priority ordering", "[daemon][components][socket][utils]") {
|
|
auto tempDir = makeTempRuntimeDir("socket-utils-" + randomSuffix());
|
|
auto configBase = tempDir / "config";
|
|
auto configDir = configBase / "yams";
|
|
fs::create_directories(configDir);
|
|
|
|
// Save environment
|
|
ScopedEnvVar yamsSockGuard("YAMS_DAEMON_SOCKET");
|
|
ScopedEnvVar xdgRuntimeGuard("XDG_RUNTIME_DIR");
|
|
ScopedEnvVar xdgConfigGuard("XDG_CONFIG_HOME");
|
|
ScopedEnvVar homeGuard("HOME");
|
|
|
|
SECTION("Environment variable has highest priority") {
|
|
auto testPath = "/tmp/custom-daemon.sock";
|
|
setenv("YAMS_DAEMON_SOCKET", testPath, 1);
|
|
|
|
auto result = socket_utils::resolve_socket_path();
|
|
REQUIRE(result.string() == testPath);
|
|
|
|
auto resultConfigFirst = socket_utils::resolve_socket_path_config_first();
|
|
REQUIRE(resultConfigFirst.string() == testPath);
|
|
}
|
|
|
|
#ifndef _WIN32
|
|
SECTION("XDG_RUNTIME_DIR used when env var not set") {
|
|
if (::geteuid() == 0) {
|
|
// Root runs default to /var/run per resolve_socket_path().
|
|
// This test specifically targets non-root behavior.
|
|
SKIP("XDG_RUNTIME_DIR precedence test is for non-root users");
|
|
}
|
|
unsetenv("YAMS_DAEMON_SOCKET");
|
|
|
|
auto xdgDir = tempDir / "runtime";
|
|
fs::create_directories(xdgDir);
|
|
setenv("XDG_RUNTIME_DIR", xdgDir.string().c_str(), 1);
|
|
|
|
auto result = socket_utils::resolve_socket_path();
|
|
REQUIRE(result.parent_path() == xdgDir);
|
|
REQUIRE(result.filename() == "yams-daemon.sock");
|
|
}
|
|
|
|
SECTION("/tmp fallback for non-root when XDG not available") {
|
|
unsetenv("YAMS_DAEMON_SOCKET");
|
|
unsetenv("XDG_RUNTIME_DIR");
|
|
|
|
if (::geteuid() != 0) {
|
|
auto result = socket_utils::resolve_socket_path();
|
|
REQUIRE(result.parent_path() == "/tmp");
|
|
// Socket name includes UID for non-root users
|
|
auto uid = std::to_string(::getuid());
|
|
REQUIRE(result.filename() == "yams-daemon-" + uid + ".sock");
|
|
}
|
|
}
|
|
#endif // !_WIN32
|
|
|
|
// Cleanup
|
|
std::error_code ec;
|
|
fs::remove_all(tempDir, ec);
|
|
}
|
|
|
|
TEST_CASE("Socket path resolution: Config file reading", "[daemon][components][socket][utils]") {
|
|
auto tempDir = makeTempRuntimeDir("socket-config-" + randomSuffix());
|
|
auto configBase = tempDir / "config";
|
|
auto configDir = configBase / "yams";
|
|
fs::create_directories(configDir);
|
|
|
|
auto configFile = configDir / "config.toml";
|
|
|
|
ScopedEnvVar xdgConfigGuard("XDG_CONFIG_HOME");
|
|
ScopedEnvVar yamsSockGuard("YAMS_DAEMON_SOCKET");
|
|
|
|
#ifndef _WIN32
|
|
SECTION("Config file socket_path is used") {
|
|
std::ofstream out(configFile);
|
|
out << "[daemon]\n";
|
|
out << "socket_path = \"/opt/yams/daemon.sock\"\n";
|
|
out.close();
|
|
|
|
setenv("XDG_CONFIG_HOME", configBase.string().c_str(), 1);
|
|
unsetenv("YAMS_DAEMON_SOCKET");
|
|
|
|
auto result = socket_utils::resolve_socket_path_config_first();
|
|
REQUIRE(result.string() == "/opt/yams/daemon.sock");
|
|
}
|
|
#endif // !_WIN32
|
|
|
|
SECTION("Invalid config falls back to default") {
|
|
std::ofstream out(configFile);
|
|
out << "invalid toml content ][[\n";
|
|
out.close();
|
|
|
|
setenv("XDG_CONFIG_HOME", configBase.string().c_str(), 1);
|
|
unsetenv("YAMS_DAEMON_SOCKET");
|
|
|
|
// Should fall back to default resolution (doesn't crash)
|
|
auto result = socket_utils::resolve_socket_path_config_first();
|
|
REQUIRE(!result.empty());
|
|
}
|
|
|
|
// Cleanup
|
|
std::error_code ec;
|
|
fs::remove_all(tempDir, ec);
|
|
}
|
|
|
|
TEST_CASE("Socket path resolution: proxy path derived from main socket",
|
|
"[daemon][components][socket][utils]") {
|
|
const fs::path mainSocket{"/tmp/yams-daemon.sock"};
|
|
const auto proxySocket = socket_utils::derive_proxy_socket_path(mainSocket);
|
|
REQUIRE(proxySocket == fs::path{"/tmp/yams-daemon.proxy.sock"});
|
|
|
|
const fs::path nestedSocket{"/run/user/1000/custom.sock"};
|
|
const auto nestedProxy = socket_utils::derive_proxy_socket_path(nestedSocket);
|
|
REQUIRE(nestedProxy == fs::path{"/run/user/1000/custom.proxy.sock"});
|
|
}
|
|
|
|
// =============================================================================
|
|
// Resource Pool Tests
|
|
// =============================================================================
|
|
|
|
TEST_CASE("ResourcePool: Basic lifecycle", "[daemon][components][pool]") {
|
|
TestResource::resetCounters();
|
|
std::atomic<int> resourceIdCounter{0};
|
|
|
|
PoolConfig<TestResource> config;
|
|
config.minSize = 3;
|
|
config.maxSize = 5;
|
|
config.maxIdle = 3;
|
|
config.idleTimeout = 1s;
|
|
config.acquisitionTimeout = 1s;
|
|
config.validateOnAcquire = true;
|
|
config.preCreateResources = true;
|
|
|
|
SECTION("Pool pre-creates minimum resources") {
|
|
using namespace yams;
|
|
auto pool = std::make_unique<ResourcePool<TestResource>>(
|
|
config,
|
|
[&resourceIdCounter](const std::string&) -> Result<std::shared_ptr<TestResource>> {
|
|
return Result<std::shared_ptr<TestResource>>(
|
|
std::make_shared<TestResource>(++resourceIdCounter));
|
|
},
|
|
[](const TestResource& r) { return r.isValid(); });
|
|
|
|
std::this_thread::sleep_for(50ms); // Allow async creation
|
|
REQUIRE(TestResource::getCreationCount() >= static_cast<int>(config.minSize));
|
|
}
|
|
|
|
TestResource::resetCounters();
|
|
resourceIdCounter = 0;
|
|
|
|
SECTION("Resource acquisition and release") {
|
|
using namespace yams;
|
|
auto pool = std::make_unique<ResourcePool<TestResource>>(
|
|
config,
|
|
[&resourceIdCounter](const std::string&) -> Result<std::shared_ptr<TestResource>> {
|
|
return Result<std::shared_ptr<TestResource>>(
|
|
std::make_shared<TestResource>(++resourceIdCounter));
|
|
},
|
|
[](const TestResource& r) { return r.isValid(); });
|
|
|
|
std::this_thread::sleep_for(50ms);
|
|
|
|
// Acquire resource (no ID parameter, timeout is optional)
|
|
auto handleResult = pool->acquire();
|
|
REQUIRE(handleResult.has_value());
|
|
auto& handle = handleResult.value();
|
|
REQUIRE(handle.isValid());
|
|
REQUIRE(handle->isValid());
|
|
|
|
// Release by destroying handle (handle goes out of scope)
|
|
}
|
|
|
|
TestResource::resetCounters();
|
|
resourceIdCounter = 0;
|
|
|
|
SECTION("Invalid resources are replaced") {
|
|
using namespace yams;
|
|
auto pool = std::make_unique<ResourcePool<TestResource>>(
|
|
config,
|
|
[&resourceIdCounter](const std::string&) -> Result<std::shared_ptr<TestResource>> {
|
|
return Result<std::shared_ptr<TestResource>>(
|
|
std::make_shared<TestResource>(++resourceIdCounter));
|
|
},
|
|
[](const TestResource& r) { return r.isValid(); });
|
|
|
|
std::this_thread::sleep_for(50ms);
|
|
|
|
// Acquire and invalidate
|
|
{
|
|
auto handleResult = pool->acquire();
|
|
REQUIRE(handleResult.has_value());
|
|
auto& handle = handleResult.value();
|
|
handle->invalidate();
|
|
// handle goes out of scope and releases
|
|
}
|
|
|
|
// Next acquisition should get valid resource (invalid one discarded)
|
|
auto handle2Result = pool->acquire();
|
|
REQUIRE(handle2Result.has_value());
|
|
auto& handle2 = handle2Result.value();
|
|
REQUIRE(handle2->isValid());
|
|
}
|
|
}
|
|
|
|
TEST_CASE("ResourcePool: Concurrency", "[daemon][components][pool][concurrent]") {
|
|
TestResource::resetCounters();
|
|
std::atomic<int> resourceIdCounter{0};
|
|
|
|
PoolConfig<TestResource> config;
|
|
config.minSize = 2;
|
|
config.maxSize = 10;
|
|
config.maxIdle = 5;
|
|
config.idleTimeout = std::chrono::seconds(1);
|
|
config.acquisitionTimeout = std::chrono::seconds(1);
|
|
config.validateOnAcquire = true;
|
|
config.preCreateResources = true;
|
|
|
|
using namespace yams;
|
|
auto pool = std::make_unique<ResourcePool<TestResource>>(
|
|
config,
|
|
[&resourceIdCounter](const std::string&) -> Result<std::shared_ptr<TestResource>> {
|
|
return Result<std::shared_ptr<TestResource>>(
|
|
std::make_shared<TestResource>(++resourceIdCounter));
|
|
},
|
|
[](const TestResource& r) { return r.isValid(); });
|
|
|
|
std::this_thread::sleep_for(50ms);
|
|
|
|
SECTION("Concurrent acquisition and release") {
|
|
std::vector<std::thread> threads;
|
|
std::atomic<int> successCount{0};
|
|
|
|
for (int i = 0; i < 20; ++i) {
|
|
threads.emplace_back([&pool, &successCount]() {
|
|
auto handleResult = pool->acquire();
|
|
if (handleResult.has_value() && handleResult.value().isValid()) {
|
|
auto& handle = handleResult.value();
|
|
handle->doWork();
|
|
successCount++;
|
|
}
|
|
});
|
|
}
|
|
|
|
for (auto& t : threads) {
|
|
t.join();
|
|
}
|
|
|
|
// All threads should successfully acquire resources
|
|
REQUIRE(successCount == 20);
|
|
REQUIRE(TestResource::getWorkCount() == 20);
|
|
}
|
|
}
|
|
|
|
#include <yams/daemon/client/asio_connection_pool.h>
|
|
#include <yams/daemon/client/global_io_context.h>
|
|
|
|
TEST_CASE("Daemon components: ResourceGovernor backpressure policy helpers",
|
|
"[daemon][components][governor]") {
|
|
auto& governor = ResourceGovernor::instance();
|
|
|
|
constexpr std::uint32_t currentSlots = 16;
|
|
constexpr std::uint32_t minSlots = 4;
|
|
constexpr std::uint32_t maxSlots = 64;
|
|
constexpr std::uint32_t scaleStep = 8;
|
|
|
|
// Normalize singleton policy state to avoid cross-test carryover.
|
|
governor.reportDbLockContention(0, 1);
|
|
(void)governor.recommendConnectionSlotTarget(currentSlots, currentSlots / 2, minSlots, maxSlots,
|
|
scaleStep, 1);
|
|
|
|
auto unchanged = governor.recommendConnectionSlotTarget(currentSlots, currentSlots, minSlots,
|
|
maxSlots, scaleStep, 3);
|
|
CHECK(unchanged == currentSlots);
|
|
|
|
auto shrunk =
|
|
governor.recommendConnectionSlotTarget(currentSlots, 1, minSlots, maxSlots, scaleStep, 1);
|
|
CHECK(shrunk <= currentSlots);
|
|
CHECK(shrunk >= minSlots);
|
|
|
|
governor.reportDbLockContention(100, 10);
|
|
CHECK(governor.capKgConcurrencyForDbContention(8) <= 2);
|
|
CHECK(governor.capEmbedConcurrencyForDbContention(8) <= 1);
|
|
|
|
governor.reportDbLockContention(0, 10);
|
|
CHECK(governor.capKgConcurrencyForDbContention(8) == 8);
|
|
CHECK(governor.capEmbedConcurrencyForDbContention(4) == 4);
|
|
|
|
const auto retryBaseline = governor.recommendRetryAfterMs(0, 100, 0, 1024, 1, 100, 0, 100, 500);
|
|
const auto retryOverloaded = governor.recommendRetryAfterMs(
|
|
200, 100, 4 * 1024 * 1024, 1024 * 1024, 200, 50, 100, 100, 500);
|
|
CHECK(retryOverloaded >= retryBaseline);
|
|
CHECK(retryOverloaded <= 5000);
|
|
|
|
const auto pauseIdle = governor.recommendBackpressureReadPauseMs(100);
|
|
CHECK(pauseIdle >= 1);
|
|
CHECK(pauseIdle <= 5000);
|
|
|
|
// Leave singleton policy state neutral for subsequent tests.
|
|
governor.reportDbLockContention(0, 1);
|
|
}
|