trevon/yams
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2
yams/tests/unit/daemon/socket_proxy_request_activity_catch2_test.cpp
Trevon 2d162793cb
 fix(daemon): count active requests, not connections, for idle detection (#289) 
TuningManager treated any open non-health connection as activity, so a
client that kept its socket open held the daemon in busy mode and kept
background scaling up. RequestHandler now counts executing non-health
requests with an RAII guard, SocketServer wires the counter into
DaemonStats, and the idle/busy decisions read it.

Tests: request_handler_status_unary, socket_server_teardown, and two slow
real-service tests (tuning_registry_activity, socket_proxy_request_activity).

Signed-off-by: trvon <me@trevon.dev>
2026-10-14 08:32:54 -06:00

275 lines
12 KiB
C++
Raw Permalink Blame History

// SPDX-License-Identifier: GPL-3.0-or-later
#include "../../common/service_manager_test_helper.h"
#include "../../common/test_helpers_catch2.h"
#include <catch2/catch_test_macros.hpp>
#include <spdlog/spdlog.h>
#include <yams/daemon/components/IOCoordinator.h>
#include <yams/daemon/components/RequestDispatcher.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/components/StateComponent.h>
#include <yams/daemon/components/TuningSnapshot.h>
#include <yams/daemon/components/WorkCoordinator.h>
#include <yams/daemon/daemon.h>
#include <yams/daemon/daemon_lifecycle.h>
#include <yams/daemon/ipc/message_framing.h>
#include <yams/metadata/metadata_repository.h>
#include <boost/asio.hpp>
#include <array>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <future>
#include <memory>
#include <mutex>
#include <span>
#include <string>
#include <thread>
#include <variant>
#include <vector>
#ifndef _WIN32
#include <sys/un.h>
#include <yams/daemon/components/SocketServer.h>
using namespace yams::daemon;
using namespace std::chrono_literals;
namespace {
// Only the shutdown callback is controlled. Readiness is the real FSM/core-service state;
// this adapter never shuts down a host daemon or fabricates a Ready snapshot.
class ObservedLifecycle final : public IDaemonLifecycle {
public:
explicit ObservedLifecycle(DaemonLifecycleFsm& fsm) : fsm_(fsm) {}
LifecycleSnapshot getLifecycleSnapshot() const override { return fsm_.snapshot(); }
void setSubsystemDegraded(const std::string& name, bool degraded,
const std::string& reason) override {
fsm_.setSubsystemDegraded(name, degraded, reason);
}
void onDocumentRemoved(const std::string&) override {}
void requestShutdown(bool graceful, bool) override {
std::unique_lock lock(mutex_);
++calls_;
graceful_ = graceful;
cv_.notify_all();
cv_.wait(lock, [&] { return released_; });
}
bool waitForEntry() {
std::unique_lock lock(mutex_);
return cv_.wait_for(lock, 2s, [&] { return calls_ == 1; });
}
bool graceful() const {
std::lock_guard lock(mutex_);
return graceful_;
}
void release() {
std::lock_guard lock(mutex_);
released_ = true;
cv_.notify_all();
}
private:
DaemonLifecycleFsm& fsm_;
mutable std::mutex mutex_;
std::condition_variable cv_;
unsigned calls_{0};
bool graceful_{false};
bool released_{false};
};
struct ProxyActivityFixture {
struct LogGuard {
spdlog::level::level_enum previous{spdlog::get_level()};
LogGuard() { spdlog::set_level(spdlog::level::warn); }
~LogGuard() { spdlog::set_level(previous); }
} logGuard;
yams::test::TempDirGuard directory{"spa_"};
const std::shared_ptr<const TuningSnapshot> savedSnapshot{
TuningSnapshotRegistry::instance().get()};
DaemonConfig config;
StateComponent state;
DaemonLifecycleFsm lifecycle;
ObservedLifecycle callback{lifecycle};
std::shared_ptr<ServiceManager> services;
std::unique_ptr<RequestDispatcher> dispatcher;
std::unique_ptr<IOCoordinator> io;
std::unique_ptr<SocketServer> server;
// Buffers outlive clientIo, including canceled pending reads on failed REQUIREs.
std::vector<uint8_t> requestFrame;
std::array<uint8_t, MessageFramer::HEADER_SIZE> replyHeader{};
std::vector<uint8_t> replyFrame;
boost::asio::io_context clientIo;
boost::asio::local::stream_protocol::socket client{clientIo};
MessageFramer framer;
ProxyActivityFixture() {
config.dataDir = directory.path();
config.socketPath = directory.path() / "d.sock";
config.pidFile = directory.path() / "daemon.pid";
config.logFile = directory.path() / "daemon.log";
config.enableModelProvider = false;
config.modelProviderRequired = false;
config.useMockModelProvider = false;
config.autoLoadPlugins = false;
config.enableAutoRepair = false;
config.autoRebuildOnDimMismatch = false;
config.embeddedOneShot = true;
config.searchMaintenance.automaticRebuildsEnabled = false;
config.searchMaintenance.automaticRebuildsSource = "typed:test";
config.memorySync.enabled = false;
config.downloadPolicy.enable = false;
}
~ProxyActivityFixture() {
// The proxy path awaits unary dispatch inline on the connection's I/O context.
// Release before stop/join even if REQUIRE failed while that I/O thread was blocked.
callback.release();
closeClient();
teardownSocketServer(server, io);
// All proxy connection/request frames are gone before borrowed dispatch/executors.
dispatcher.reset();
if (services)
services->shutdown();
services.reset();
TuningSnapshotRegistry::instance().set(savedSnapshot);
}
void closeClient() {
boost::system::error_code ignored;
client.close(ignored);
}
template <typename Predicate> bool observe(Predicate predicate) {
// Public occupancy/activity atomics have no subscription. This tight bounded
// observation is a correctness deadline, not a latency/notification measurement.
const auto deadline = std::chrono::steady_clock::now() + 2s;
do {
if (predicate())
return true;
std::this_thread::sleep_for(10ms);
} while (std::chrono::steady_clock::now() < deadline);
return predicate();
}
template <typename T> T complete(std::future<T>& future) {
clientIo.run_for(2s);
REQUIRE(future.wait_for(0ms) == std::future_status::ready);
return future.get();
}
void initialize() {
// Initialization is outside the constructor: failure still invokes release/teardown.
services = std::make_shared<ServiceManager>(config, state, lifecycle);
REQUIRE(yams::test::initializeServiceManagerFully(services, 30s, false));
REQUIRE(state.readiness.contentStoreReady.load());
REQUIRE(state.readiness.metadataRepoReady.load());
auto repo = services->getMetadataRepo();
REQUIRE(repo != nullptr);
// Real repository query; unavailable FTS5 can legitimately return an empty set.
REQUIRE(repo->getFts5IndexedRowIdSet().has_value());
auto* coordinator = services->getWorkCoordinator();
REQUIRE(coordinator != nullptr);
REQUIRE(coordinator->isRunning());
REQUIRE(coordinator->getWorkerCount() >= 2);
dispatcher = std::make_unique<RequestDispatcher>(&callback, services.get(), &state);
IOCoordinator::Config ioConfig;
ioConfig.num_threads = 1;
io = std::make_unique<IOCoordinator>(ioConfig);
SocketServer::Config socketConfig;
socketConfig.socketPath = config.socketPath;
socketConfig.proxySocketPath = directory.path() / "p.sock";
socketConfig.connectionTimeout = 5s;
REQUIRE(socketConfig.proxySocketPath.string().size() < sizeof(sockaddr_un{}.sun_path));
server = std::make_unique<SocketServer>(socketConfig, io.get(), coordinator,
dispatcher.get(), &state);
const auto started = server->start();
REQUIRE(started);
io->start();
boost::system::error_code error;
client.connect(
boost::asio::local::stream_protocol::endpoint{socketConfig.proxySocketPath.string()},
error);
REQUIRE_FALSE(error);
REQUIRE(observe([&] {
return server->totalConnections() == 1 && server->activeConnections() == 1 &&
state.stats.activeConnections.load() == 1;
}));
REQUIRE(state.stats.healthCheckConnections.load() == 0);
REQUIRE(state.stats.activeNonHealthRequests.load() == 0);
}
void sendShutdown() {
Message message{};
message.requestId = 701;
message.timestamp = std::chrono::steady_clock::now();
message.payload = Request{ShutdownRequest{}};
const auto framed = framer.frame_message_into(message, requestFrame);
REQUIRE(framed);
clientIo.restart();
auto sent = boost::asio::async_write(client, boost::asio::buffer(requestFrame),
boost::asio::use_future);
REQUIRE(complete(sent) == requestFrame.size());
// Positive control: actual framing, acceptance, readiness and dispatch reached the
// downstream public callback. A provider/ABI/transport failure is not behavioral RED.
REQUIRE(callback.waitForEntry());
REQUIRE(callback.graceful());
}
void readSuccess() {
clientIo.restart();
auto headerRead = boost::asio::async_read(client, boost::asio::buffer(replyHeader),
boost::asio::use_future);
REQUIRE(complete(headerRead) == replyHeader.size());
const auto header = framer.parse_header(std::span<const uint8_t>{replyHeader});
REQUIRE(header);
REQUIRE_FALSE(header.value().is_chunked());
REQUIRE_FALSE(header.value().is_error());
REQUIRE(header.value().payload_size > 0);
REQUIRE(header.value().payload_size <= 4096);
replyFrame.assign(replyHeader.begin(), replyHeader.end());
replyFrame.resize(replyHeader.size() + header.value().payload_size);
clientIo.restart();
auto bodyRead =
boost::asio::async_read(client,
boost::asio::buffer(replyFrame.data() + replyHeader.size(),
header.value().payload_size),
boost::asio::use_future);
REQUIRE(complete(bodyRead) == header.value().payload_size);
const auto parsed = framer.parse_frame(replyFrame);
REQUIRE(parsed);
REQUIRE(parsed.value().requestId == 701);
const auto* response = std::get_if<Response>(&parsed.value().payload);
REQUIRE(response != nullptr);
REQUIRE(std::holds_alternative<SuccessResponse>(*response));
}
};
} // namespace
TEST_CASE_METHOD(ProxyActivityFixture,
"SocketServer proxy wires executing requests and balances completion/disconnect",
"[daemon][socket_server][proxy][activity][slow][regression]") {
initialize();
sendShutdown();
// No seeded activity: acceptance and the real handler must publish into this same State.
REQUIRE(state.stats.activeConnections.load() == 1);
REQUIRE(state.stats.healthCheckConnections.load() == 0);
REQUIRE(state.stats.activeNonHealthRequests.load() == 1);
SECTION("successful response finishes request while the connection stays accepted") {
callback.release();
readSuccess();
REQUIRE(observe([&] { return state.stats.activeNonHealthRequests.load() == 0; }));
CHECK(state.stats.activeConnections.load() == 1);
CHECK(server->activeConnections() == 1);
closeClient();
}
SECTION("disconnect during execution still releases the request guard") {
closeClient();
callback.release();
}
REQUIRE(observe([&] {
return server->activeConnections() == 0 && state.stats.activeConnections.load() == 0 &&
state.stats.activeNonHealthRequests.load() == 0 &&
state.stats.healthCheckConnections.load() == 0 &&
state.stats.ipcTasksPending.load() == 0 && state.stats.ipcTasksActive.load() == 0;
}));
CHECK(server->totalConnections() == 1);
// Proxy is deliberately serial: this does not validate detached main-socket mux lifetime,
// overlapping health requests, continuous tuning, cancellation latency or CPU idleness.
}
#endif
Reference in a new issue View git blame Copy permalink