trevon/yams
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2
yams/tests/unit/cli/daemon_status_render_catch2_test.cpp
Trevon e0d15dc680
 fix(p2p): stop quarantining honest writers for concurrent-session races (#245) 
* feat(p2p): log durable writer quarantines and outbound session failures

A node that durably quarantined a peer writer left no trace of why: the
three p2p quarantine sites returned an error to their session, and the
reconnect loop dropped every outbound session error without logging it,
so a mesh that stopped receiving one writer showed only a quarantined
record count. Outbound sessions are half of all sessions in a full mesh.

Each p2p quarantine site now logs the writer, the failed check and the
local and frozen counters, and a failed outbound session logs its peer
and error, matching the existing inbound failure log.

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

* fix(p2p): stop quarantining honest writers for concurrent-session races

Each pair of direct peers normally runs two sessions at once, one
initiated from each side. Both freeze a window of the peer's writer at
handshake, and whichever applies first advances the receiver's durable
prefix under the other. The second session then compared its frozen
window with that live prefix and treated the difference as forged
history: validateHistoryExtension demanded the window start right after
the live commitment ("history extension is not contiguous"), and an
empty window demanded the live counter equal the frozen frontier. Both
durably quarantined an honest writer, after which the node dropped its
documents and refused every later session with it. A 5-node mesh run
logged exactly this (local=30 frontier=32 staged=4: a window frozen at
28 arriving after another session had applied 29-30), and the mesh
never converged.

The receiver now validates a window against durable history instead of
a snapshot. Operations at or below the advanced prefix are replays that
applyDeltas leaves untouched (a different record under a known operation
id is still a fork). The rest must chain from the durable commitment
exactly to the authenticated frontier, and an empty window must match
the durable commitment at its frontier. validateHistoryExtension returns
a typed WriterHistoryViolation for writer faults (fork, forged frontier,
gap, unauthenticated record) and an Error for local or transient
failures, and only a violation quarantines; an Error aborts the session
for a retry on the next cycle.

The security property is unchanged: nothing is applied unless a hash
chain links durable history, itself verified against the writer's
authenticated commitments, to the frontier the writer authenticated in
this handshake. Overlapping replays change no state, and forks, forged
frontiers, gaps and unauthenticated records still quarantine, now also
when they overlap an advanced prefix (new tests cover each).

The pre- and post-exchange history checks get the same treatment. A
live prefix between the proven handshake prefix and the frozen frontier
is consistent, since delta exchange verifies the rest; one beyond the
frontier is compared with the durable commitment at the frontier rather
than rejected as a rollback.

The handshake also clamps the live local-writer window resolver to the
frozen snapshot, so concurrent local writes no longer fail the session
with "local writer-window resolver violated negotiated bounds".

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

* feat(p2p): report outbound sessions and durable writer quarantines

`yams p2p status` showed inbound session failures but not outbound
ones, and only a `quarantined` record count that mixes durable writer
quarantine with transient rejections (a record whose causal dependency
from a third writer has not arrived yet). An operator could not tell a
mesh that was retrying from one that had permanently cut off a writer.

Status now reports:

- outbound_sessions / outbound_failures and the last outbound failure
  with its peer, counted for both explicit connects and reconnects.
  Failed sessions in either direction are aborted and retried.
- quarantined_writers: writers durably quarantined by this node, read
  from the replication checkpoint, so it survives restarts. Each
  quarantine is also logged with its reason.

The fields travel through MemorySyncStatus, MemorySyncResponse and
protobuf fields 38-42 to the text and JSON status and to `yams daemon
status -d`, following the apply-health fields.

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

* fix(memory-sync): verify replayed writer counters against durable history

validateHistoryExtension skipped every staged operation at or below the
writer's durable prefix as a replay without looking at its record hash,
leaving fork detection to applyDeltas. applyDeltas recognizes a fork only
by an operation id it still remembers, and that memory is lost on restart
and bounded by maxTrackedIdentities. A different, internally valid record
for an already-applied counter therefore passed as a harmless replay
instead of being quarantined as the fork it is, whether the window only
replayed old counters or rode the fork along an honest extension.

Before skipping a replayed counter, compare its record hash with the
durable history entry committed at that counter (rebuilding the entry like
computeHistoryCommitmentAt when it is missing). A mismatch is a
WriterHistoryViolation, so the p2p session durably quarantines the writer;
an identical replay is still accepted as a no-op. A history entry that
cannot be read stays a local Error and only aborts the session.

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

* fix(memory-sync): log every durable writer quarantine where it happens

Durable writer quarantine was logged only by its two p2p callers. The
duplicate-operation fork path inside applyDeltas and the shared-store scan
path quarantined silently, so the documented guarantee that every durable
quarantine logs its reason and counters did not hold, and an operator saw
a quarantined_writers count with nothing in the log to explain it.

quarantineWriter now takes the reason and is the one place that logs a new
quarantine: writer, reason, source node, and the writer's local history and
version counters. It logs nothing when the writer was already quarantined,
so a repeated request is not reported twice. The p2p callers drop their own
warnings and fold their session context (frontier, staged count, prefix
match) into the reason instead. memory_sync now declares its spdlog
dependency explicitly.

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

* fix(cli): show outbound p2p sessions in daemon status

The daemon already reports outbound sessions, failures, and the last
outbound failure, and `yams p2p status` prints them, but `yams daemon
status -d` only gained the quarantined-writers row. An outbound direction
that keeps failing is aborted and retried, never quarantined, so the
detailed status was the place an operator would miss it.

The Memory Sync section now shows an Outbound sessions row (started and
failed counts, warning when any failed) for direct-backend nodes, and a
Last outbound failure row with the peer, the error, and its age.

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

---------

Signed-off-by: trvon <me@trevon.dev>
2026-10-01 12:18:47 -06:00

433 lines
17 KiB
C++
Raw Permalink Blame History

// Characterization tests for the `yams daemon status` renderers.
//
// The renderers are pure StatusResponse -> text, so a synthetic response drives every branch
// the CLI used to reach only through a live daemon RPC.
#include <catch2/catch_test_macros.hpp>
#include <yams/cli/daemon_status_render.h>
#include <yams/cli/ui_helpers.hpp>
#include <yams/common/fs_utils.h>
#include <yams/daemon/ipc/ipc_protocol.h>
#include <yams/daemon/metric_keys.h>
#include <filesystem>
#include <sstream>
#include <string>
#include <string_view>
namespace {
using yams::cli::DaemonStatusRenderContext;
using yams::daemon::MemorySyncResponse;
using yams::daemon::StatusResponse;
struct PlainColors {
yams::cli::ui::ColorMode saved{yams::cli::ui::color_mode()};
PlainColors() { yams::cli::ui::set_colors_enabled_override(false); }
~PlainColors() { yams::cli::ui::set_color_mode(saved); }
};
StatusResponse readyDaemon() {
StatusResponse s{};
s.running = true;
s.ready = true;
s.version = "9.9.9";
s.uptimeSeconds = 125;
s.requestsProcessed = 42;
s.lifecycleState = "ready";
s.memoryUsageMb = 512;
s.embeddingAvailable = true;
s.embeddingModel = "all-MiniLM-L6-v2";
s.contentStoreRoot = "/srv/yams/storage";
s.vectorDbInitAttempted = true;
s.vectorDbReady = true;
s.readinessStates[std::string(yams::daemon::readiness::kDatabase)] = true;
s.requestCounts["repair_running"] = 1;
return s;
}
std::string renderBrief(const StatusResponse& s, const DaemonStatusRenderContext& ctx) {
std::ostringstream os;
yams::cli::renderDaemonStatusBrief(s, ctx, os);
return os.str();
}
std::string renderDetailed(const StatusResponse& s, const DaemonStatusRenderContext& ctx) {
std::ostringstream os;
yams::cli::renderDaemonStatusDetailed(s, ctx, os);
return os.str();
}
std::string lineContaining(const std::string& text, std::string_view needle) {
std::istringstream lines(text);
for (std::string line; std::getline(lines, line);) {
if (line.find(needle) != std::string::npos) {
return line;
}
}
return {};
}
} // namespace
TEST_CASE("primary status is flushed before optional collection", "[cli][status-order]") {
PlainColors plain;
for (bool detailed : {false, true}) {
struct Buffer : std::stringbuf {
bool flushed = false;
int sync() override {
flushed = true;
return std::stringbuf::sync();
}
} buffer;
std::ostream os(&buffer);
bool called = false;
yams::cli::renderDaemonStatusWithOptionalSection(readyDaemon(), {}, detailed, os,
[&](std::ostream& out) {
called = true;
CHECK_FALSE(buffer.str().empty());
CHECK(buffer.flushed);
out << "optional-status";
});
CHECK(called);
CHECK(buffer.str().find("optional-status") != std::string::npos);
}
}
TEST_CASE("daemon status brief renders the overview from a synthetic response",
"[cli][daemon][status][catch2]") {
PlainColors plain;
DaemonStatusRenderContext ctx;
ctx.configuredDataDir = "/srv/yams";
const std::string out = renderBrief(readyDaemon(), ctx);
CHECK(out.find("YAMS Daemon") != std::string::npos);
CHECK(out.find("Ready") != std::string::npos);
CHECK(out.find("v9.9.9") != std::string::npos);
CHECK(out.find("42 requests") != std::string::npos);
CHECK(out.find("all-MiniLM-L6-v2") != std::string::npos);
CHECK(out.find("Running") != std::string::npos);
CHECK(out.find("Data dir warning") == std::string::npos);
CHECK(out.find("Waiting on:") == std::string::npos);
CHECK(out.find("Error:") == std::string::npos);
CHECK(out.find("Memory Sync") == std::string::npos);
CHECK(out.find("yams daemon status -d") != std::string::npos);
}
TEST_CASE("daemon status names a running integrity check instead of 'Opening'",
"[cli][daemon][status][catch2]") {
PlainColors plain;
StatusResponse s = readyDaemon();
s.ready = false;
s.lifecycleState = "initializing";
s.readinessStates[std::string(yams::daemon::readiness::kDatabase)] = false;
s.databasePhase = std::string(yams::daemon::dbphase::kCheckingIntegrity);
s.databasePhaseElapsedMs = 671000;
s.metadataDbPath = "/srv/yams/yams.db";
const std::string out = renderBrief(s, {});
INFO(out);
CHECK(out.find("Checking integrity") != std::string::npos);
CHECK(out.find("11m") != std::string::npos);
CHECK(out.find("full scan after unclean stop") != std::string::npos);
CHECK(out.find("Opening") == std::string::npos);
}
TEST_CASE("daemon status shows the vector index as pending while vectors initialize",
"[cli][daemon][status][catch2]") {
PlainColors plain;
StatusResponse s = readyDaemon();
s.ready = false;
s.lifecycleState = "initializing";
s.vectorDbInitAttempted = false;
s.vectorDbReady = false;
s.readinessStates[std::string(yams::daemon::readiness::kDatabase)] = false;
s.databasePhase = std::string(yams::daemon::dbphase::kCheckingIntegrity);
s.databasePhaseElapsedMs = 90000;
const std::string brief = renderBrief(s, {});
INFO(brief);
CHECK(lineContaining(brief, "Vector Index").find("pending") != std::string::npos);
CHECK(brief.find("by configuration") == std::string::npos);
const std::string detailed = renderDetailed(s, {});
INFO(detailed);
CHECK(lineContaining(detailed, "Vector DB").find("Pending") != std::string::npos);
CHECK(detailed.find("by configuration") == std::string::npos);
}
TEST_CASE("daemon status says 'disabled by configuration' only when the daemon reports it",
"[cli][daemon][status][catch2]") {
PlainColors plain;
StatusResponse s = readyDaemon();
s.vectorDbInitAttempted = false;
s.vectorDbReady = false;
s.readinessStates[std::string(yams::daemon::readiness::kVectorDbDisabled)] = true;
const std::string brief = renderBrief(s, {});
INFO(brief);
const auto line = lineContaining(brief, "Vector Index");
CHECK(line.find("disabled") != std::string::npos);
CHECK(line.find("by configuration") != std::string::npos);
CHECK(line.find("pending") == std::string::npos);
const std::string detailed = renderDetailed(s, {});
INFO(detailed);
CHECK(lineContaining(detailed, "Vector DB").find("by configuration") != std::string::npos);
}
TEST_CASE("lexical containment rejects incomparable roots", "[cli][daemon][status][catch2]") {
using yams::common::isLexicallyContained;
CHECK(isLexicallyContained("/tmp", "/tmp"));
CHECK(isLexicallyContained("/tmp/child", "/tmp"));
CHECK_FALSE(isLexicallyContained("relative", "/tmp"));
CHECK_FALSE(isLexicallyContained("/tmp", "relative"));
CHECK_FALSE(isLexicallyContained("/tmp-other", "/tmp"));
CHECK_FALSE(isLexicallyContained("/tmp/../outside", "/tmp"));
CHECK_FALSE(isLexicallyContained({}, "/tmp"));
CHECK_FALSE(isLexicallyContained("/tmp", {}));
#ifdef _WIN32
CHECK_FALSE(isLexicallyContained("D:/data", "C:/temp"));
CHECK_FALSE(isLexicallyContained("C:relative", "C:/temp"));
CHECK_FALSE(isLexicallyContained("C:/temp", "C:relative"));
CHECK(isLexicallyContained("C:/temp/child", "C:/temp"));
#endif
}
TEST_CASE("daemon status brief surfaces blockers, errors, and data-dir drift",
"[cli][daemon][status][catch2]") {
PlainColors plain;
StatusResponse s = readyDaemon();
s.ready = false;
s.lifecycleState = "initializing";
s.lastError = "vector index rebuild failed";
s.readinessStates["content_store"] = false;
s.initProgress["content_store"] = 40;
DaemonStatusRenderContext ctx;
ctx.configuredDataDir = "/elsewhere/yams";
const std::string out = renderBrief(s, ctx);
CHECK(out.find("Initializing") != std::string::npos);
CHECK(out.find("Waiting on: Content Store (40%)") != std::string::npos);
CHECK(out.find("Error: vector index rebuild failed") != std::string::npos);
CHECK(out.find("Daemon data dir differs from current CLI/config") != std::string::npos);
SECTION("an explicit --socket makes the data-dir difference expected") {
ctx.explicitCustomSocket = true;
const std::string quiet = renderBrief(s, ctx);
CHECK(quiet.find("Daemon data dir differs") == std::string::npos);
}
}
TEST_CASE("daemon status brief warns when the daemon serves a temporary data directory",
"[cli][daemon][status][catch2]") {
PlainColors plain;
StatusResponse s = readyDaemon();
s.contentStoreRoot =
(std::filesystem::temp_directory_path() / "yams-ephemeral" / "storage").string();
DaemonStatusRenderContext ctx;
ctx.configuredDataDir = std::filesystem::temp_directory_path() / "yams-ephemeral";
const std::string out = renderBrief(s, ctx);
CHECK(out.find("Daemon is serving a temporary data directory") != std::string::npos);
CHECK(out.find("differs from current CLI/config") == std::string::npos);
}
TEST_CASE("daemon status names the active repair operation from the shared code table",
"[cli][daemon][status][catch2]") {
PlainColors plain;
StatusResponse s = readyDaemon();
s.requestCounts[std::string(yams::daemon::metrics::kRepairCurrentOperationCode)] =
yams::daemon::metrics::repairOperationCodeForName("block_refs");
s.requestCounts[std::string(yams::daemon::metrics::kRepairCurrentOperationElapsedMs)] = 65000;
DaemonStatusRenderContext ctx;
const std::string brief = renderBrief(s, ctx);
CHECK(brief.find("Running · block refs") != std::string::npos);
const std::string detailed = renderDetailed(s, ctx);
CHECK(detailed.find("block refs") != std::string::npos);
}
TEST_CASE("daemon status detailed renders every section from a synthetic response",
"[cli][daemon][status][catch2]") {
PlainColors plain;
StatusResponse s = readyDaemon();
StatusResponse::ProviderInfo provider;
provider.name = "onnx";
provider.ready = true;
provider.modelsLoaded = 2;
provider.isProvider = true;
s.providers.push_back(provider);
StatusResponse::ModelInfo model;
model.name = "all-MiniLM-L6-v2";
model.type = "onnx";
model.memoryMb = 128;
model.requestCount = 4;
s.models.push_back(model);
StatusResponse::PluginSkipInfo skipped;
skipped.path = "/plugins/legacy_plugin.so";
skipped.reason = "abi mismatch";
s.skippedPlugins.push_back(skipped);
DaemonStatusRenderContext ctx;
ctx.configuredDataDir = "/srv/yams";
const std::string out = renderDetailed(s, ctx);
for (const char* header : {"YAMS Daemon — Detailed", "Resources", "Transport", "Search",
"Post-Ingest Pipeline", "Internal Processing", "Repair Service",
"Storage & Embeddings", "Providers", "Skipped Plugins", "Models"}) {
INFO(header);
CHECK(out.find(header) != std::string::npos);
}
CHECK(out.find("/plugins/legacy_plugin.so") != std::string::npos);
CHECK(out.find("abi mismatch") != std::string::npos);
CHECK(out.find("128 MB · 4 req") != std::string::npos);
CHECK(out.find("2 models · active") != std::string::npos);
CHECK(out.find("Components Not Ready") == std::string::npos);
CHECK(out.find("Data Directory Warnings") == std::string::npos);
}
TEST_CASE("daemon status shows host pressure and deferred background work",
"[cli][daemon][status][host_pressure][catch2]") {
PlainColors plain;
SECTION("busy host names what is being deferred") {
StatusResponse s = readyDaemon();
s.hostPressureSource = "psi";
s.hostPressureElevated = true;
s.hostCpuPressurePct = 72.5;
s.hostIoPressurePct = 3.0;
s.hostMemoryPressurePct = 0.0;
s.hostLoadPerCpu = 1.9;
s.deferredBackgroundWork = "topology,repair";
for (const auto& out : {renderBrief(s, {}), renderDetailed(s, {})}) {
INFO(out);
CHECK(out.find("Host Load") != std::string::npos);
CHECK(out.find("Busy") != std::string::npos);
CHECK(out.find("cpu 72%") != std::string::npos);
CHECK(out.find("io 3%") != std::string::npos);
CHECK(out.find("deferring topology, repair") != std::string::npos);
}
}
SECTION("calm load-average host shows the per-CPU load") {
StatusResponse s = readyDaemon();
s.hostPressureSource = "loadavg";
s.hostLoadPerCpu = 0.42;
const std::string out = renderBrief(s, {});
INFO(out);
CHECK(out.find("Host Load") != std::string::npos);
CHECK(out.find("Calm") != std::string::npos);
CHECK(out.find("load 0.42/cpu") != std::string::npos);
CHECK(out.find("deferring") == std::string::npos);
}
SECTION("no host signal hides the row") {
const std::string out = renderBrief(readyDaemon(), {});
CHECK(out.find("Host Load") == std::string::npos);
}
}
TEST_CASE("memory sync section renders only when the loop has started",
"[cli][daemon][status][catch2]") {
PlainColors plain;
SECTION("no response") {
std::ostringstream os;
yams::cli::renderMemorySyncSection(nullptr, os);
CHECK(os.str().empty());
}
MemorySyncResponse m;
m.backend = "direct";
m.mode = "writer";
m.corpusId = "corpus-1";
m.corpusEpoch = 3;
m.nodeId = "node-a";
m.trustMode = "authenticated-writers";
m.peerCount = 2;
m.records = 10;
m.quarantinedRecords = 1;
m.successfulCycles = 5;
m.failedCycles = 1;
m.lastSuccessAgeMs = 1000;
SECTION("not started") {
m.started = false;
std::ostringstream os;
yams::cli::renderMemorySyncSection(&m, os);
CHECK(os.str().empty());
}
SECTION("started") {
m.started = true;
std::ostringstream os;
yams::cli::renderMemorySyncSection(&m, os);
const std::string out = os.str();
CHECK(out.find("Memory Sync") != std::string::npos);
CHECK(out.find("corpus-1 (epoch 3)") != std::string::npos);
CHECK(out.find("authenticated-writers") != std::string::npos);
CHECK(out.find("yams p2p peers for details") != std::string::npos);
CHECK(out.find("check daemon log for reasons") != std::string::npos);
CHECK(out.find("5 ok · 1 failed") != std::string::npos);
CHECK(out.find("0 ok · 0 with stage failures") != std::string::npos);
CHECK(out.find("0 records") != std::string::npos);
CHECK(out.find("Last apply error") == std::string::npos);
CHECK(out.find("Publish") == std::string::npos);
}
SECTION("started with a stuck apply path") {
m.started = true;
m.applyCycles = 1268;
m.applyFailedCycles = 1268;
m.deferredTopology = 2;
m.oldestDeferralAgeMs = 11 * 60 * 1000;
m.applyFailuresTopology = 1268;
m.lastApplyFailureStage = "topology";
m.lastApplyFailure = "edge identity mismatch";
m.publishSkippedCycles = 3;
std::ostringstream os;
yams::cli::renderMemorySyncSection(&m, os);
const std::string out = os.str();
CHECK(out.find("0 ok · 1268 with stage failures") != std::string::npos);
CHECK(out.find("2 records (oldest") != std::string::npos);
CHECK(out.find("prerequisite not replicated") != std::string::npos);
CHECK(out.find("topology: edge identity mismatch") != std::string::npos);
CHECK(out.find("3 skipped · 0 failed") != std::string::npos);
}
SECTION("started with failing outbound sessions and a quarantined writer") {
m.started = true;
m.quarantinedWriters = 1;
m.outboundSessions = 12;
m.outboundFailures = 3;
m.lastOutboundFailure = "node-b: connection refused";
m.lastOutboundFailureAgeMs = 90'000;
std::ostringstream os;
yams::cli::renderMemorySyncSection(&m, os);
const std::string out = os.str();
CHECK(out.find("Quarantined writers") != std::string::npos);
CHECK(out.find("durable; check daemon log for the reason") != std::string::npos);
CHECK(out.find("Outbound sessions") != std::string::npos);
CHECK(out.find("12 started · 3 failed") != std::string::npos);
CHECK(out.find("Last outbound failure") != std::string::npos);
CHECK(out.find("node-b: connection refused") != std::string::npos);
CHECK(out.find("1m 30s ago") != std::string::npos);
}
SECTION("started with healthy outbound sessions") {
m.started = true;
m.outboundSessions = 4;
std::ostringstream os;
yams::cli::renderMemorySyncSection(&m, os);
const std::string out = os.str();
CHECK(out.find("4 started · 0 failed") != std::string::npos);
CHECK(out.find("Last outbound failure") == std::string::npos);
}
}
Reference in a new issue View git blame Copy permalink