trevon/yams
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2
yams/tests/unit/daemon/repair_maintenance_catch2_test.cpp
trvon b5a5396931
 test(repair): run the vectors.db vacuum case with real vector tables 
CI runs the unit lanes with YAMS_SQLITE_VEC_SKIP_INIT=1, so the test's
VectorDatabase came up without vector tables and the post-VACUUM insert
failed on every platform. The case already clears the other vector
toggles; clear this one too.

Signed-off-by: trvon <me@trevon.dev>
2026-09-29 22:16:05 -06:00

409 lines
17 KiB
C++
Raw Permalink Blame History

// Idle maintenance run by RepairService: vectors.db VACUUM and session expiry.
#include <catch2/catch_test_macros.hpp>
#include "../../common/test_helpers_catch2.h"
#include <yams/app/services/session_service.hpp>
#include <yams/daemon/components/RepairService.h>
#include <yams/daemon/components/ResourceGovernor.h>
#include <yams/daemon/components/sqlite_vacuum.h>
#include <yams/metadata/connection_pool.h>
#include <yams/metadata/metadata_repository.h>
#include <yams/vector/vector_database.h>
#include <nlohmann/json.hpp>
#include <sqlite3.h>
#include <algorithm>
#include <chrono>
#include <cstdint>
#include <filesystem>
#include <fstream>
#include <memory>
#include <optional>
#include <string>
#include <vector>
using namespace yams::daemon;
namespace fs = std::filesystem;
namespace {
constexpr std::uint64_t kPageSize = 4096;
constexpr std::uint64_t kGiB = 1024ULL * 1024 * 1024;
// A policy small enough for a test file: any file with >= 64 KiB of free pages and a 10%
// free ratio qualifies.
SqliteVacuumPolicy smallFilePolicy() {
SqliteVacuumPolicy policy;
policy.minDatabaseBytes = 0;
policy.minReclaimableBytes = 64 * 1024;
policy.minReclaimableRatio = 0.10;
return policy;
}
void execOrFail(sqlite3* db, const char* sql) {
char* err = nullptr;
const int rc = sqlite3_exec(db, sql, nullptr, nullptr, &err);
INFO(sql << " -> " << (err ? err : "ok"));
sqlite3_free(err);
REQUIRE(rc == SQLITE_OK);
}
// Keep a few thousand live rows (so VACUUM has real copying to do) and add ~4 MiB of blobs
// that are then deleted, leaving most pages on the freelist.
void addReclaimableFreePages(const fs::path& path) {
sqlite3* db = nullptr;
REQUIRE(sqlite3_open_v2(path.string().c_str(), &db, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE,
nullptr) == SQLITE_OK);
execOrFail(db, "PRAGMA journal_mode=WAL");
execOrFail(db, "CREATE TABLE IF NOT EXISTS filler(id INTEGER PRIMARY KEY, payload BLOB)");
execOrFail(db, "WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < "
"1024) INSERT INTO filler(payload) SELECT zeroblob(4096) FROM n");
execOrFail(db, "CREATE TABLE IF NOT EXISTS live_rows(id INTEGER PRIMARY KEY, label TEXT)");
execOrFail(db, "WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < "
"4000) INSERT INTO live_rows(label) SELECT 'row-' || i FROM n");
execOrFail(db, "DELETE FROM filler");
execOrFail(db, "PRAGMA wal_checkpoint(TRUNCATE)");
sqlite3_close_v2(db);
}
std::uint64_t fileBytes(const fs::path& path) {
return static_cast<std::uint64_t>(fs::file_size(path));
}
using namespace std::chrono_literals;
std::int64_t epochSeconds(std::chrono::system_clock::time_point when) {
return std::chrono::duration_cast<std::chrono::seconds>(when.time_since_epoch()).count();
}
// Write a session file the way SessionService does, last touched `idle` ago.
void writeSession(const fs::path& dir, const std::string& name, std::chrono::hours idle,
nlohmann::json extra = nlohmann::json::object()) {
const auto touched = std::chrono::system_clock::now() - idle;
nlohmann::json j = {{"name", name},
{"state", "closed"},
{"createdTime", epochSeconds(touched)},
{"lastOpenedTime", epochSeconds(touched)},
{"lastClosedTime", 0},
{"selectors", nlohmann::json::array()},
{"materialized", nlohmann::json::array()}};
j.update(extra);
const auto path = dir / (name + ".json");
std::ofstream(path) << j.dump(2);
fs::last_write_time(path, fs::file_time_type::clock::now() - idle);
}
bool contains(const std::vector<std::string>& names, const std::string& name) {
return std::find(names.begin(), names.end(), name) != names.end();
}
// A session store shaped like the audited machine: per-run tool sessions plus a few that must
// survive expiry.
struct SessionStore {
yams::test::TempDirGuard tmp{"yams_session_expiry_"};
fs::path dir{tmp.path() / "sessions"};
fs::path watchedDir{tmp.path() / "project"};
SessionStore() {
fs::create_directories(dir);
fs::create_directories(watchedDir);
std::ofstream(dir / "index.json") << R"({"current": "active-now"})";
fs::last_write_time(dir / "index.json", fs::file_time_type::clock::now() - 24h * 90);
writeSession(dir, "opencode-1a2b", 24h * 40);
writeSession(dir, "mcp-grep-hot-7f", 24h * 45);
writeSession(dir, "active-now", 24h * 60); // current, however old
writeSession(dir, "used-yesterday", 24h);
writeSession(dir, "watching-project", 24h * 60,
{{"watch", {{"enabled", true}, {"interval_ms", 2000}}},
{"selectors", {{{"path", watchedDir.string()}}}}});
writeSession(dir, "watching-deleted-dir", 24h * 60,
{{"watch", {{"enabled", true}, {"interval_ms", 2000}}},
{"selectors", {{{"path", (tmp.path() / "gone").string()}}}}});
writeSession(dir, "owns-documents", 24h * 60);
std::ofstream(dir / "corrupt.json") << "{ not json";
fs::last_write_time(dir / "corrupt.json", fs::file_time_type::clock::now() - 24h * 90);
}
};
} // namespace
TEST_CASE("shouldVacuumSqlite keeps the yams.db thresholds",
"[daemon][repair][maintenance][vacuum][catch2]") {
const SqliteVacuumPolicy policy;
SECTION("large file with mostly free pages is worth it") {
// The audited install: vectors.db 39.7 GB with 48% free pages.
const std::uint64_t dbBytes = 397ULL * kGiB / 10;
const std::uint64_t pages = dbBytes / kPageSize;
CHECK(shouldVacuumSqlite(dbBytes, pages, pages * 48 / 100, kPageSize, policy));
}
SECTION("small files are never vacuumed") {
CHECK_FALSE(shouldVacuumSqlite(256ULL * 1024 * 1024, 65'536, 60'000, kPageSize, policy));
}
SECTION("a low free-page ratio is not worth a rewrite") {
const std::uint64_t pages = 8ULL * kGiB / kPageSize;
CHECK_FALSE(shouldVacuumSqlite(8ULL * kGiB, pages, pages / 20, kPageSize, policy));
}
SECTION("trailing bytes past the logical end count as reclaimable") {
CHECK(shouldVacuumSqlite(600ULL * 1024 * 1024, 8, 0, kPageSize, policy));
}
SECTION("inconsistent statistics are rejected") {
CHECK_FALSE(shouldVacuumSqlite(2ULL * kGiB, 10, 11, kPageSize, policy));
CHECK_FALSE(shouldVacuumSqlite(2ULL * kGiB, 0, 0, kPageSize, policy));
CHECK_FALSE(shouldVacuumSqlite(2ULL * kGiB, 10, 5, 0, policy));
}
SECTION("free disk must exceed the current file size") {
CHECK(hasSpaceForSqliteVacuum(10 * kGiB, 10 * kGiB + 1));
CHECK_FALSE(hasSpaceForSqliteVacuum(10 * kGiB, 10 * kGiB));
CHECK_FALSE(hasSpaceForSqliteVacuum(10 * kGiB, kGiB));
}
}
TEST_CASE("vacuumSqliteFileIfUseful reclaims free pages and yields to work",
"[daemon][repair][maintenance][vacuum][catch2]") {
yams::test::TempDirGuard tmp("yams_sqlite_vacuum_");
const auto dbPath = tmp.path() / "vectors.db";
addReclaimableFreePages(dbPath);
const auto before = fileBytes(dbPath);
REQUIRE(before > 2ULL * 1024 * 1024);
SECTION("the default policy leaves a small file alone") {
const auto outcome = vacuumSqliteFileIfUseful(dbPath, SqliteVacuumPolicy{}, {});
CHECK(outcome.status == SqliteVacuumStatus::NotNeeded);
CHECK(fileBytes(dbPath) == before);
}
SECTION("a qualifying file is rewritten and shrinks") {
const auto outcome = vacuumSqliteFileIfUseful(dbPath, smallFilePolicy(), {});
CHECK(outcome.status == SqliteVacuumStatus::Vacuumed);
CHECK(outcome.bytesBefore == before);
CHECK(outcome.bytesAfter < before / 4);
CHECK(fileBytes(dbPath) == outcome.bytesAfter);
}
SECTION("an abort request rolls the VACUUM back") {
const auto outcome =
vacuumSqliteFileIfUseful(dbPath, smallFilePolicy(), [] { return true; });
CHECK(outcome.status == SqliteVacuumStatus::Interrupted);
CHECK(fileBytes(dbPath) == before);
}
SECTION("a live writer makes it back off instead of waiting") {
sqlite3* writer = nullptr;
REQUIRE(sqlite3_open_v2(dbPath.string().c_str(), &writer, SQLITE_OPEN_READWRITE, nullptr) ==
SQLITE_OK);
execOrFail(writer, "BEGIN IMMEDIATE");
const auto outcome = vacuumSqliteFileIfUseful(dbPath, smallFilePolicy(), {});
execOrFail(writer, "ROLLBACK");
sqlite3_close_v2(writer);
CHECK(outcome.status == SqliteVacuumStatus::Busy);
CHECK(fileBytes(dbPath) == before);
}
SECTION("a missing file is never created") {
const auto missing = tmp.path() / "absent.db";
const auto outcome = vacuumSqliteFileIfUseful(missing, smallFilePolicy(), {});
CHECK(outcome.status == SqliteVacuumStatus::OpenFailed);
CHECK_FALSE(fs::exists(missing));
}
}
TEST_CASE("vectors.db VACUUM is admitted only while the daemon is idle",
"[daemon][repair][maintenance][vacuum][catch2]") {
RepairService::VectorVacuumSignals idle;
idle.maintenanceAllowed = true;
idle.pressure = ResourcePressureLevel::Normal;
CHECK(RepairService::vectorVacuumAdmitted(idle));
auto busy = idle;
busy.shuttingDown = true;
CHECK_FALSE(RepairService::vectorVacuumAdmitted(busy));
busy = idle;
busy.maintenanceAllowed = false; // a client is connected
CHECK_FALSE(RepairService::vectorVacuumAdmitted(busy));
busy = idle;
busy.pressure = ResourcePressureLevel::Warning;
CHECK_FALSE(RepairService::vectorVacuumAdmitted(busy));
busy = idle;
busy.embeddingQueued = 1;
CHECK_FALSE(RepairService::vectorVacuumAdmitted(busy));
busy = idle;
busy.embeddingInFlight = 1;
CHECK_FALSE(RepairService::vectorVacuumAdmitted(busy));
busy = idle;
busy.indexMutating = true; // rebuild or bulk load owns the vector connection
CHECK_FALSE(RepairService::vectorVacuumAdmitted(busy));
busy = idle;
busy.repairInProgress = true;
CHECK_FALSE(RepairService::vectorVacuumAdmitted(busy));
busy = idle;
busy.hostDeferred = true; // another process is keeping the host busy
CHECK_FALSE(RepairService::vectorVacuumAdmitted(busy));
}
TEST_CASE("RepairService maintenance vacuums the live vectors.db when idle",
"[daemon][repair][maintenance][vacuum][catch2]") {
yams::test::ScopedEnvVar vectorsOn("YAMS_DISABLE_VECTORS", std::nullopt);
yams::test::ScopedEnvVar vectorsOnAlias("YAMS_DISABLE_VECTOR_DB", std::nullopt);
yams::test::ScopedEnvVar vectorsOnSingular("YAMS_DISABLE_VECTOR", std::nullopt);
yams::test::ScopedEnvVar onDisk("YAMS_VDB_IN_MEMORY", std::nullopt);
// CI runs the unit lanes with sqlite-vec init skipped; this case needs real vector tables.
yams::test::ScopedEnvVar vecInit("YAMS_SQLITE_VEC_SKIP_INIT", std::nullopt);
ResourceGovernor::instance().testing_setPressureState(ResourcePressureLevel::Normal,
std::chrono::steady_clock::now());
yams::test::TempDirGuard tmp("yams_repair_vacuum_");
const auto dbPath = tmp.path() / "vectors.db";
yams::vector::VectorDatabaseConfig vcfg;
vcfg.database_path = dbPath.string();
vcfg.embedding_dim = 8;
auto vectorDb = std::make_shared<yams::vector::VectorDatabase>(vcfg);
REQUIRE(vectorDb->initialize());
addReclaimableFreePages(dbPath);
const auto before = fileBytes(dbPath);
std::size_t activeConnections = 0;
std::size_t queuedEmbeddings = 0;
RepairServiceContext ctx;
ctx.getVectorDatabase = [vectorDb] { return vectorDb; };
ctx.getEmbeddingQueuedJobs = [&queuedEmbeddings] { return queuedEmbeddings; };
ctx.getEmbeddingInFlightJobs = [] { return std::size_t{0}; };
RepairService::Config cfg;
cfg.enable = true;
cfg.dataDir = tmp.path();
RepairService service(ctx, nullptr, [&activeConnections] { return activeConnections; }, cfg);
service.testing_setVectorVacuumPolicy(smallFilePolicy());
SECTION("a connected client defers it") {
activeConnections = 1;
CHECK_FALSE(service.testing_runVectorVacuumMaintenance().has_value());
CHECK(fileBytes(dbPath) == before);
}
SECTION("queued embedding writes defer it") {
queuedEmbeddings = 3;
CHECK_FALSE(service.testing_runVectorVacuumMaintenance().has_value());
CHECK(fileBytes(dbPath) == before);
}
SECTION("an idle daemon reclaims the free pages and the database stays usable") {
const auto outcome = service.testing_runVectorVacuumMaintenance();
REQUIRE(outcome.has_value());
CHECK(outcome->status == SqliteVacuumStatus::Vacuumed);
CHECK(fileBytes(dbPath) < before / 4);
yams::vector::VectorRecord record;
record.chunk_id = "chunk-after-vacuum";
record.document_hash = std::string(64, 'a');
record.embedding = std::vector<float>(8, 0.5f);
record.content = "still writable";
CHECK(vectorDb->insertVector(record));
CHECK(vectorDb->getVectorCount() == 1);
}
SECTION("vectors disabled by configuration leaves the file alone") {
yams::test::ScopedEnvVar disabled("YAMS_DISABLE_VECTORS", std::string("1"));
CHECK_FALSE(service.testing_runVectorVacuumMaintenance().has_value());
CHECK(fileBytes(dbPath) == before);
}
vectorDb->close();
}
TEST_CASE("expireIdleSessions deletes only idle, unowned, unwatched sessions",
"[daemon][repair][maintenance][sessions][catch2]") {
SessionStore store;
yams::app::services::SessionExpiryOptions options;
options.maxIdle = 24h * 30;
options.hasSessionDocuments = [](const std::string& name) { return name == "owns-documents"; };
const auto result = yams::app::services::expireIdleSessions(store.dir, options);
CHECK(contains(result.expired, "opencode-1a2b"));
CHECK(contains(result.expired, "mcp-grep-hot-7f"));
CHECK(contains(result.expired, "watching-deleted-dir"));
CHECK(result.expired.size() == 3);
CHECK_FALSE(fs::exists(store.dir / "opencode-1a2b.json"));
CHECK_FALSE(fs::exists(store.dir / "mcp-grep-hot-7f.json"));
for (const char* kept : {"active-now", "used-yesterday", "watching-project", "owns-documents",
"corrupt", "index"}) {
INFO(kept);
CHECK(fs::exists(store.dir / (std::string(kept) + ".json")));
}
CHECK(result.kept == 5);
}
TEST_CASE("expireIdleSessions keeps everything when ownership cannot be checked",
"[daemon][repair][maintenance][sessions][catch2]") {
SessionStore store;
yams::app::services::SessionExpiryOptions options;
options.maxIdle = 24h * 30;
CHECK(yams::app::services::expireIdleSessions(store.dir, options).expired.empty());
options.hasSessionDocuments = [](const std::string&) { return false; };
options.maxIdle = std::chrono::system_clock::duration::zero(); // expiry disabled
CHECK(yams::app::services::expireIdleSessions(store.dir, options).expired.empty());
CHECK(fs::exists(store.dir / "opencode-1a2b.json"));
CHECK(yams::app::services::expireIdleSessions(store.tmp.path() / "missing", options)
.expired.empty());
}
TEST_CASE("RepairService maintenance expires idle sessions using document ownership",
"[daemon][repair][maintenance][sessions][catch2]") {
SessionStore store;
const auto dbPath = store.tmp.path() / "yams.db";
yams::metadata::ConnectionPoolConfig poolConfig;
poolConfig.minConnections = 1;
poolConfig.maxConnections = 2;
auto pool = std::make_unique<yams::metadata::ConnectionPool>(dbPath.string(), poolConfig);
REQUIRE(pool->initialize().has_value());
auto repo = std::make_shared<yams::metadata::MetadataRepository>(*pool);
yams::metadata::DocumentInfo doc;
doc.filePath = (store.tmp.path() / "note.md").string();
doc.fileName = "note.md";
doc.fileExtension = ".md";
doc.fileSize = 4;
doc.sha256Hash = std::string(64, 'b');
doc.mimeType = "text/markdown";
auto docId = repo->insertDocument(doc);
REQUIRE(docId.has_value());
REQUIRE(repo->setMetadata(docId.value(), "session_id",
yams::metadata::MetadataValue(std::string("owns-documents")))
.has_value());
std::size_t activeConnections = 0;
RepairServiceContext ctx;
ctx.getMetadataRepo = [repo] { return repo; };
RepairService::Config cfg;
cfg.enable = true;
cfg.dataDir = store.tmp.path();
cfg.sessionsDir = store.dir;
RepairService service(ctx, nullptr, [&activeConnections] { return activeConnections; }, cfg);
SECTION("a connected client defers it") {
activeConnections = 1;
CHECK_FALSE(service.testing_runSessionExpiryMaintenance().has_value());
CHECK(fs::exists(store.dir / "opencode-1a2b.json"));
}
SECTION("an idle daemon expires the tool sessions but not the one owning documents") {
const auto result = service.testing_runSessionExpiryMaintenance();
REQUIRE(result.has_value());
CHECK(contains(result->expired, "opencode-1a2b"));
CHECK(contains(result->expired, "mcp-grep-hot-7f"));
CHECK_FALSE(contains(result->expired, "owns-documents"));
CHECK(fs::exists(store.dir / "owns-documents.json"));
CHECK(fs::exists(store.dir / "active-now.json"));
}
repo.reset();
pool->shutdown();
}
Reference in a new issue View git blame Copy permalink