trevon/yams
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2
yams/tests/unit/daemon/memory_sync_apply_catch2_test.cpp
Trevon 211db15606
 fix(daemon): bound semantic neighbor scores and report startup phase (#334) 
Semantic-neighbor maintenance aborted the daemon during startup:
std::clamp<float> asserted '!(__hi < __lo)' because the adaptive cosine
threshold could exceed 1.0 for a legal pair of identical vectors. The same
work was also admitted while the daemon was still initializing, so the
300-second starvation guard forced a pass before readiness.

+ semanticInverseNorm/semanticCosineSimilarity/semanticExplicitThreshold in the
  existing embed policy header, shared by the streamed and corpus branches;
  non-finite vectors or scores produce no neighbor and legal scores are bounded
  before ranking and threshold selection
+ startupMaintenanceEligible() requires ServiceManagerState::Ready, not
  stopping and idle, and is evaluated before
  ResourceGovernor::admitDeferrable(), so an elapsed deferral deadline can no
  longer manufacture missing readiness
+ StartupPhaseTracker (two atomics, no mutex) plus a warn-level init-waiter
  line reporting state, phase and phase_elapsed_ms instead of a debug-level
  state number; a 53-minute startup previously logged nothing actionable
+ regression through the public backfill and corpus-rebuild APIs using a 64-D
  vector whose old float inverse norm rounds its self-cosine above 1
- removed the duplicated inverse-norm and cosine lambdas from EmbeddingService
- renamed __test_setModelProvider/__test_setKgStore to testing_* because they
  used reserved identifiers, across 11 files
- re-anchored 13 pre-existing raw-environment allowlist entries to their new
  line numbers; no new exception was added

Startup is still blocked by corpus-sized memory-sync scans; this change stops
the abort and names the blocking phase, it does not make startup fast.

Signed-off-by: trvon <git@trevon.dev>
2026-10-22 16:48:49 -06:00

1009 lines
42 KiB
C++
Raw Permalink Blame History

// SPDX-License-Identifier: GPL-3.0-or-later
// Copyright 2026 YAMS Contributors
// Apply-cycle behavior of the daemon memory-sync path across several in-process nodes.
//
// Each node is a real ServiceManager with its own content store, knowledge graph, and memory-sync
// writer identity, all replicating through one shared-store backend. Cycles are driven explicitly
// (sync + apply + rate-limited outbound backfill), so convergence is asserted in a bounded number
// of rounds rather than with wall-clock polling.
// pi-lens-ignore: fatal error
#include <catch2/catch_test_macros.hpp>
#include <catch2/generators/catch_generators.hpp>
#include <chrono>
#include <cstddef>
#include <cstring>
#include <filesystem>
#include <memory>
#include <optional>
#include <span>
#include <string>
#include <string_view>
#include <system_error>
#include <thread>
#include <utility>
#include <vector>
#include "../../common/test_helpers_catch2.h"
#include <yams/api/content_store_builder.h>
#include <yams/crypto/hasher.h>
#include <yams/daemon/components/DaemonLifecycleFsm.h>
#include <yams/daemon/components/ServiceManager.h>
#include <yams/daemon/components/StateComponent.h>
#include <yams/daemon/components/VectorIndexCoordinator.h>
#include <yams/daemon/daemon.h>
#include <yams/memory_sync/memory_sync_service.h>
#include <yams/memory_sync/records.h>
#include <yams/metadata/connection_pool.h>
#include <yams/metadata/knowledge_graph_store.h>
#include <yams/metadata/topology_sync_adapter.h>
#include <yams/storage/storage_backend.h>
#include <yams/vector/vector_database.h>
namespace fs = std::filesystem;
using namespace yams;
using namespace yams::daemon;
namespace {
constexpr std::string_view kCorpus = "apply-deadlock-corpus";
constexpr std::string_view kRelation = "links_to";
std::vector<std::byte> bytes(std::string_view text) {
std::vector<std::byte> out(text.size());
std::memcpy(out.data(), text.data(), text.size());
return out;
}
std::string digest(std::span<const std::byte> data) {
crypto::SHA256Hasher hasher;
hasher.init();
hasher.update(data);
return hasher.finalize();
}
std::unique_ptr<storage::FilesystemBackend> makeBackend(const fs::path& path) {
storage::BackendConfig config;
config.type = "filesystem";
config.localPath = path;
auto backend = std::make_unique<storage::FilesystemBackend>();
REQUIRE(backend->initialize(config).has_value());
return backend;
}
struct TempRoot {
TempRoot() {
path = fs::temp_directory_path() /
("yams-memory-sync-apply-" +
std::to_string(std::hash<std::thread::id>{}(std::this_thread::get_id())) + "-" +
std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()));
fs::create_directories(path);
}
TempRoot(const TempRoot&) = delete;
TempRoot& operator=(const TempRoot&) = delete;
TempRoot(TempRoot&&) = delete;
TempRoot& operator=(TempRoot&&) = delete;
~TempRoot() {
std::error_code ec;
fs::remove_all(path, ec);
}
fs::path path;
};
/// One daemon-shaped replica: real ServiceManager apply/backfill code over local stores.
struct MeshNode {
/// Counted: the graph store runs over a connection pool this node owns, so a test can count
/// every connection the store takes (one per query).
enum class KgReads { Uncounted, Counted };
MeshNode(const fs::path& root, const std::string& nodeId, const fs::path& sharedStore,
KgReads kgReads = KgReads::Uncounted, memory_sync::MemorySyncLimits syncLimits = {},
std::unique_ptr<storage::IStorageBackend> syncBackend = {})
: name(nodeId) {
const auto home = root / nodeId;
fs::create_directories(home / "data");
config.dataDir = home / "data";
config.socketPath = home / "daemon.sock";
config.pidFile = home / "daemon.pid";
config.logFile = home / "daemon.log";
config.memorySync.enabled = true;
config.memorySync.nodeId = nodeId;
config.memorySync.corpusId = std::string(kCorpus);
config.memorySync.transport = "shared-store";
config.memorySync.backend = "filesystem";
manager = std::make_unique<ServiceManager>(config, state, lifecycleFsm);
auto content = api::ContentStoreBuilder::createDefault(home / "storage");
REQUIRE(content.has_value());
manager->__test_setContentStore(
std::shared_ptr<api::IContentStore>(std::move(content.value())));
const auto graphPath = (home / "kg.db").string();
auto graph = metadata::makeSqliteKnowledgeGraphStore(graphPath);
REQUIRE(graph.has_value());
if (kgReads == KgReads::Counted) {
// The path factory migrated the schema; reopen the store over a pool of our own.
graph.value().reset();
kgPool_ = std::make_unique<metadata::ConnectionPool>(graphPath);
REQUIRE(kgPool_->initialize().has_value());
graph = metadata::makeSqliteKnowledgeGraphStore(*kgPool_,
metadata::KnowledgeGraphStoreConfig{});
REQUIRE(graph.has_value());
}
kgStore = std::shared_ptr<metadata::KnowledgeGraphStore>(std::move(graph.value()));
manager->testing_setKgStore(kgStore);
if (!syncBackend) {
syncBackend = makeBackend(sharedStore);
}
auto sync = std::make_unique<memory_sync::MemorySyncService>(
std::move(syncBackend),
memory_sync::MemorySyncConfig{nodeId, 60'000, std::string(kCorpus), 1, syncLimits});
sync_ = sync.get();
manager->testingSetMemorySyncService(std::move(sync));
}
MeshNode(const MeshNode&) = delete;
MeshNode& operator=(const MeshNode&) = delete;
MeshNode(MeshNode&&) = delete;
MeshNode& operator=(MeshNode&&) = delete;
~MeshNode() {
if (sync_ != nullptr) {
sync_->stop();
}
manager.reset();
}
std::string sourceKey() const { return "entity:" + name + ":source"; }
std::string targetKey() const { return "entity:" + name + ":target"; }
/// Seed a local edge whose target node never reached the shared store: the edge and its
/// source were published, then publication stopped before the target (the backfill sweeps
/// node types in order, so an edge can precede its target node of a later type).
void seedEdgePublishedAheadOfTarget() {
const metadata::KGNode source{
.nodeKey = sourceKey(), .label = name + "-source", .type = "alpha"};
const metadata::KGNode target{
.nodeKey = targetKey(), .label = name + "-target", .type = "omega"};
const auto sourceId = kgStore->upsertNode(source);
const auto targetId = kgStore->upsertNode(target);
REQUIRE(sourceId.has_value());
REQUIRE(targetId.has_value());
const metadata::KGEdge edge{.srcNodeId = sourceId.value(),
.dstNodeId = targetId.value(),
.relation = std::string(kRelation),
.weight = 1.0F};
REQUIRE(kgStore->addEdge(edge).has_value());
metadata::TopologySyncAdapter publisher{*kgStore, *sync_};
REQUIRE(publisher.publishNode(source).has_value());
REQUIRE(publisher.publishEdge(source.nodeKey, edge, target.nodeKey).has_value());
}
/// One production-shaped cycle: inbound apply followed by the outbound backfill.
void runCycle() {
manager->testingExpireMemorySyncBackfillSchedule();
manager->testingApplyMemorySyncWinners();
}
bool hasEdge(const std::string& sourceNodeKey, const std::string& targetNodeKey) const {
const auto source = kgStore->getNodeByKey(sourceNodeKey);
const auto target = kgStore->getNodeByKey(targetNodeKey);
REQUIRE(source.has_value());
REQUIRE(target.has_value());
if (!source.value() || !target.value()) {
return false;
}
const auto edges = kgStore->getEdgesFrom(source.value()->id, std::string(kRelation));
REQUIRE(edges.has_value());
for (const auto& edge : edges.value()) {
if (edge.dstNodeId == target.value()->id) {
return true;
}
}
return false;
}
bool hasEdgeOf(const MeshNode& origin) const {
return hasEdge(origin.sourceKey(), origin.targetKey());
}
metadata::KGNode addNode(const std::string& key, const std::string& type) const {
metadata::KGNode node{.nodeKey = key, .label = key, .type = type};
const auto id = kgStore->upsertNode(node);
REQUIRE(id.has_value());
node.id = id.value();
return node;
}
void addEdge(const metadata::KGNode& source, const metadata::KGNode& target) const {
const metadata::KGEdge edge{.srcNodeId = source.id,
.dstNodeId = target.id,
.relation = std::string(kRelation),
.weight = 1.0F};
REQUIRE(kgStore->addEdge(edge).has_value());
}
/// What a topology rebuild does to a replaced cluster node: delete it, cascading its edges.
void deleteNode(const metadata::KGNode& node) const {
REQUIRE(kgStore->deleteNodeById(node.id).has_value());
}
bool hasNode(const std::string& key) const {
const auto node = kgStore->getNodeByKey(key);
REQUIRE(node.has_value());
return node.value().has_value();
}
/// Drive the outbound backfill alone, one bounded publish cycle per call.
void backfill(std::size_t cycles) {
for (std::size_t i = 0; i < cycles; ++i) {
manager->testingPublishMemorySyncBackfill();
}
}
std::uint64_t deferredTopology() const {
const auto status = manager->getMemorySyncStatus();
REQUIRE(status.has_value());
return status.value().apply.deferredIn(memory_sync::ApplyStage::Topology);
}
/// Connections the graph store has taken so far (KgReads::Counted only).
std::size_t kgStoreAcquisitions() const {
REQUIRE(kgPool_ != nullptr);
return kgPool_->getStats().totalAcquired;
}
std::string name;
yams::test::ScopedEnvVar disableWatcher{"YAMS_DISABLE_SESSION_WATCHER", std::string("1")};
DaemonConfig config;
StateComponent state;
DaemonLifecycleFsm lifecycleFsm;
std::unique_ptr<metadata::ConnectionPool> kgPool_;
std::shared_ptr<metadata::KnowledgeGraphStore> kgStore;
memory_sync::MemorySyncService* sync_{nullptr};
std::unique_ptr<ServiceManager> manager;
};
} // namespace
TEST_CASE("Memory sync nodes holding each other's missing prerequisites still converge",
"[daemon][memory-sync][apply][deadlock]") {
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode nodeA{root.path, "node-a", sharedStore};
MeshNode nodeB{root.path, "node-b", sharedStore};
// Each node holds an inbound edge whose target only the other node can publish.
nodeA.seedEdgePublishedAheadOfTarget();
nodeB.seedEdgePublishedAheadOfTarget();
// Two rounds suffice when publication is independent of inbound apply: round one publishes
// each missing target, round two applies it. Allow slack, but stay bounded.
constexpr std::size_t kMaxRounds = 4;
std::size_t rounds = 0;
bool converged = false;
while (!converged && rounds < kMaxRounds) {
nodeA.runCycle();
if (rounds == 0) {
// A ran first, before B published its target: B's edge is deferred, not an error.
const auto status = nodeA.manager->getMemorySyncStatus();
REQUIRE(status.has_value());
CHECK(status.value().apply.deferredIn(memory_sync::ApplyStage::Topology) == 1);
CHECK(status.value().apply.oldestDeferralAgeMs < 60'000);
CHECK(status.value().apply.applyFailedCycles == 0);
}
nodeB.runCycle();
++rounds;
converged = nodeA.hasEdgeOf(nodeB) && nodeB.hasEdgeOf(nodeA);
}
INFO("rounds=" << rounds);
CHECK(nodeA.hasEdgeOf(nodeB));
CHECK(nodeB.hasEdgeOf(nodeA));
CHECK(converged);
for (const auto* node : {&nodeA, &nodeB}) {
const auto status = node->manager->getMemorySyncStatus();
REQUIRE(status.has_value());
const auto& apply = status.value().apply;
INFO("node=" << node->name);
CHECK(apply.applyCycles == rounds);
CHECK(apply.applyFailedCycles == 0);
CHECK(apply.deferredTotal() == 0);
CHECK(apply.oldestDeferralAgeMs == 0);
CHECK(apply.publishSkippedCycles == 0);
CHECK(apply.publishFailedCycles == 0);
}
}
TEST_CASE("Memory sync content apply stores good blobs past a corrupt one",
"[daemon][memory-sync][apply][content]") {
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode writer{root.path, "writer", sharedStore};
MeshNode reader{root.path, "reader", sharedStore};
// Blobs apply in key order; put the corrupt one first so it cannot hide the good one.
const auto first = bytes("content-apply-first");
const auto second = bytes("content-apply-second");
const auto firstHash = digest(first);
const auto secondHash = digest(second);
const bool firstSortsFirst = firstHash < secondHash;
const auto& corruptHash = firstSortsFirst ? firstHash : secondHash;
const auto& goodHash = firstSortsFirst ? secondHash : firstHash;
const auto& goodBytes = firstSortsFirst ? second : first;
REQUIRE(
writer.sync_->publish("content-blob/" + corruptHash, bytes("not-the-bytes")).has_value());
REQUIRE(writer.sync_->publish("content-blob/" + goodHash, goodBytes).has_value());
reader.runCycle();
auto contentStore = reader.manager->getContentStore();
REQUIRE(contentStore != nullptr);
const auto good = contentStore->exists(goodHash);
REQUIRE(good.has_value());
CHECK(good.value());
const auto corrupt = contentStore->exists(corruptHash);
REQUIRE(corrupt.has_value());
CHECK_FALSE(corrupt.value());
// The corrupt blob is a genuine failure: it fails the stage and the cycle, visibly.
const auto status = reader.manager->getMemorySyncStatus();
REQUIRE(status.has_value());
const auto& apply = status.value().apply;
CHECK(apply.failuresIn(memory_sync::ApplyStage::Content) == 1);
CHECK(apply.applyCycles == 1);
CHECK(apply.applyFailedCycles == 1);
CHECK(apply.lastFailureStage == "content");
CHECK(apply.lastFailure.find("do not match the claimed hash") != std::string::npos);
}
namespace {
std::string topologyNodeKey(std::string_view nodeKey) {
return "topology-node/" + memory_sync::escapeRecordKeySegment(nodeKey);
}
std::string topologyEdgeKey(std::string_view sourceKey, std::string_view targetKey) {
memory_sync::TopologyEdgeRecord record;
record.sourceNodeKey = sourceKey;
record.relation = kRelation;
record.targetNodeKey = targetKey;
return "topology-edge/" + record.id();
}
/// A read-only view of what has been published to the shared store.
struct SharedStoreView {
explicit SharedStoreView(const fs::path& sharedStore)
: sync(makeBackend(sharedStore),
memory_sync::MemorySyncConfig{"observer", 60'000, std::string(kCorpus), 1}) {}
void refresh() { REQUIRE(sync.syncOnce().has_value()); }
bool published(const std::string& key) const { return sync.readCached(key).has_value(); }
bool tombstoned(const std::string& key) const { return sync.hasCommittedTombstone(key); }
memory_sync::MemorySyncService sync;
};
} // namespace
TEST_CASE("Memory sync topology backfill publishes every node despite deletes between items",
"[daemon][memory-sync][backfill][topology]") {
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore};
SharedStoreView view{sharedStore};
origin.manager->testingSetMemorySyncBackfillItemBudget(1);
std::vector<metadata::KGNode> previous;
for (int i = 0; i < 3; ++i) {
previous.push_back(origin.addNode("topology:cluster:old-" + std::to_string(i), "cluster"));
}
origin.backfill(2);
// A topology rebuild lands between two backfill items: it deletes every cluster node,
// including the ones the sweep already passed, and inserts the replacements.
for (const auto& node : previous) {
origin.deleteNode(node);
}
std::vector<std::string> current;
for (int i = 0; i < 3; ++i) {
current.push_back(
origin.addNode("topology:cluster:new-" + std::to_string(i), "cluster").nodeKey);
}
origin.backfill(16);
view.refresh();
for (const auto& key : current) {
INFO("node=" << key);
CHECK(view.published(topologyNodeKey(key)));
}
}
TEST_CASE("Memory sync topology backfill publishes an edge added after its node was swept",
"[daemon][memory-sync][backfill][topology]") {
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore};
MeshNode peer{root.path, "peer", sharedStore};
SharedStoreView view{sharedStore};
origin.manager->testingSetMemorySyncBackfillItemBudget(1);
const auto document = origin.addNode("doc:late-edge-source", "document");
const auto neighbor = origin.addNode("doc:late-edge-target", "document");
// Let the sweep pass both nodes (and finish the pass) before the edge exists.
origin.backfill(8);
view.refresh();
REQUIRE(view.published(topologyNodeKey(document.nodeKey)));
REQUIRE(view.published(topologyNodeKey(neighbor.nodeKey)));
origin.addEdge(document, neighbor);
origin.backfill(8);
view.refresh();
CHECK(view.published(topologyEdgeKey(document.nodeKey, neighbor.nodeKey)));
peer.runCycle();
CHECK(peer.hasEdge(document.nodeKey, neighbor.nodeKey));
CHECK(peer.deferredTopology() == 0);
}
TEST_CASE("Memory sync retracts a cluster node and its edges when a rebuild deletes them",
"[daemon][memory-sync][backfill][topology][retraction]") {
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore};
MeshNode peer{root.path, "peer", sharedStore};
SharedStoreView view{sharedStore};
const auto document = origin.addNode("doc:rebuild-member", "document");
const auto stale = origin.addNode("topology:cluster:stale", "topology_cluster");
origin.addEdge(document, stale);
origin.runCycle();
peer.runCycle();
REQUIRE(peer.hasEdge(document.nodeKey, stale.nodeKey));
// The rebuild replaces the cluster: the old node goes (cascading its member edge) and the
// document joins a new one.
origin.deleteNode(stale);
const auto fresh = origin.addNode("topology:cluster:fresh", "topology_cluster");
origin.addEdge(document, fresh);
for (int round = 0; round < 3; ++round) {
origin.runCycle();
peer.runCycle();
}
// The origin's own published records must not resurrect what its rebuild deleted.
CHECK_FALSE(origin.hasNode(stale.nodeKey));
view.refresh();
CHECK(view.tombstoned(topologyNodeKey(stale.nodeKey)));
CHECK(view.tombstoned(topologyEdgeKey(document.nodeKey, stale.nodeKey)));
CHECK_FALSE(peer.hasNode(stale.nodeKey));
CHECK_FALSE(peer.hasEdge(document.nodeKey, stale.nodeKey));
CHECK(peer.hasEdge(document.nodeKey, fresh.nodeKey));
CHECK(origin.deferredTopology() == 0);
CHECK(peer.deferredTopology() == 0);
}
TEST_CASE("Memory sync clears a deferral on an edge whose cluster node was deleted unpublished",
"[daemon][memory-sync][backfill][topology][retraction]") {
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore};
MeshNode peer{root.path, "peer", sharedStore};
// The state the production mesh was left in: the member edge and its document reached the
// shared store, but the rebuild deleted the cluster node before it was ever published.
const auto document = origin.addNode("doc:orphaned-member", "document");
const auto cluster = origin.addNode("topology:cluster:never-published", "topology_cluster");
origin.addEdge(document, cluster);
{
metadata::TopologySyncAdapter publisher{*origin.kgStore, *origin.sync_};
const auto edges = origin.kgStore->getEdgesFrom(document.id, std::string(kRelation));
REQUIRE(edges.has_value());
REQUIRE(edges.value().size() == 1);
REQUIRE(publisher.publishNode(document).has_value());
REQUIRE(publisher.publishEdge(document.nodeKey, edges.value().front(), cluster.nodeKey)
.has_value());
}
origin.deleteNode(cluster);
peer.runCycle();
REQUIRE(peer.deferredTopology() == 1);
for (int round = 0; round < 3; ++round) {
origin.runCycle();
peer.runCycle();
}
CHECK_FALSE(peer.hasEdge(document.nodeKey, cluster.nodeKey));
CHECK(origin.deferredTopology() == 0);
CHECK(peer.deferredTopology() == 0);
}
namespace {
// Pins the vector-store environment for one case. CI runs the unit lanes with vectors
// disabled (YAMS_DISABLE_VECTORS, YAMS_SQLITE_VEC_SKIP_INIT) and the vector DB forced in
// memory; VectorDatabase then skips creating its tables, so these cases must not inherit that.
struct VectorStoreEnv {
yams::test::ScopedEnvVar disable{"YAMS_DISABLE_VECTORS", std::nullopt};
yams::test::ScopedEnvVar disableSingular{"YAMS_DISABLE_VECTOR", std::nullopt};
yams::test::ScopedEnvVar disableDb{"YAMS_DISABLE_VECTOR_DB", std::nullopt};
yams::test::ScopedEnvVar skipVecInit{"YAMS_SQLITE_VEC_SKIP_INIT", std::nullopt};
yams::test::ScopedEnvVar inMemory{"YAMS_VDB_IN_MEMORY", std::nullopt};
};
std::shared_ptr<vector::VectorDatabase> attachVectorDb(MeshNode& node) {
vector::VectorDatabaseConfig config;
config.database_path = ":memory:";
config.embedding_dim = 4;
config.create_if_missing = true;
config.use_in_memory = true;
config.search_engine = vector::VectorSearchEngine::ExactScan;
auto database = std::make_shared<vector::VectorDatabase>(config);
REQUIRE(database->initializeChecked().has_value());
// Wire it as vector initialization does: the store, then the index coordinator.
node.manager->testingSetVectorDatabase(database);
if (auto coordinator = node.manager->getVectorIndexCoordinator()) {
coordinator->setVectorDatabase(database);
}
REQUIRE(node.manager->getVectorDatabase() == database);
return database;
}
vector::VectorRecord makeVector(std::string chunkId, char documentDigit, std::string model) {
vector::VectorRecord record;
record.chunk_id = std::move(chunkId);
record.document_hash = std::string(64, documentDigit);
record.model_id = std::move(model);
record.embedding = {1.0F, 0.0F, 0.0F, 0.0F};
record.embedding_dim = 4;
record.content = "vector backfill content";
return record;
}
} // namespace
TEST_CASE("Memory sync vector backfill publishes good vectors past an unpublishable one",
"[daemon][memory-sync][backfill][vector]") {
const VectorStoreEnv vectorEnv;
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore};
SharedStoreView view{sharedStore};
auto vectors = attachVectorDb(origin);
// The sweep walks (document_hash, chunk_id) order. The first row can never publish: it has
// no model identity, as rows written before model ids were recorded do not. The rows behind
// it are good and must still replicate.
REQUIRE(vectors->insertVectorChecked(makeVector("legacy-chunk", '1', "")).has_value());
REQUIRE(vectors->insertVectorChecked(makeVector("good-chunk-a", '2', "model")).has_value());
REQUIRE(vectors->insertVectorChecked(makeVector("good-chunk-b", '3', "model")).has_value());
origin.backfill(4);
view.refresh();
CHECK(view.published("embedding/model/good-chunk-a"));
CHECK(view.published("embedding/model/good-chunk-b"));
// A skipped record is not an outbound failure; it is logged once and passed.
const auto status = origin.manager->getMemorySyncStatus();
REQUIRE(status.has_value());
CHECK(status.value().apply.publishFailedCycles == 0);
}
TEST_CASE("Memory sync vector backfill progresses past a resident-limit rejection",
"[daemon][memory-sync][backfill][vector][limits]") {
const VectorStoreEnv vectorEnv;
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
memory_sync::MemorySyncLimits limits;
const bool cacheBound = GENERATE(false, true);
if (cacheBound) {
limits.maxCacheBytes = 1024;
} else {
limits.maxValueBytes = 1024;
}
CAPTURE(cacheBound);
MeshNode origin{root.path, "origin", sharedStore, MeshNode::KgReads::Uncounted, limits};
SharedStoreView view{sharedStore};
auto vectors = attachVectorDb(origin);
auto oversized = makeVector("oversized-chunk", '1', "model");
oversized.content.assign(2048, 'x');
REQUIRE(vectors->insertVectorChecked(oversized).has_value());
REQUIRE(vectors->insertVectorChecked(makeVector("later-chunk", '2', "model")).has_value());
SECTION("catch-up sweep") {}
SECTION("committed vector queue") {
origin.manager->notifyMemorySyncEmbeddingsCommitted(
{std::string(64, '1'), std::string(64, '2')});
}
// The first value is valid vector data, but cannot fit this service's typed limit.
// Retrying it must not keep every later vector behind the sweep cursor.
origin.backfill(2);
view.refresh();
CHECK_FALSE(view.published("embedding/model/oversized-chunk"));
CHECK(view.published("embedding/model/later-chunk"));
REQUIRE(vectors->getVector("oversized-chunk").has_value());
// The local record is retained. A corrected value changes the vector generation and
// must be reached by a later sweep rather than permanently remembered as skipped.
REQUIRE(vectors->deleteVectorsByDocumentChecked(std::string(64, '1')).has_value());
REQUIRE(vectors->insertVectorChecked(makeVector("oversized-chunk", '1', "model")).has_value());
origin.backfill(4);
view.refresh();
CHECK(view.published("embedding/model/oversized-chunk"));
}
TEST_CASE("Memory sync vector backfill retries a backend validation failure without new vectors",
"[daemon][memory-sync][backfill][vector][retry]") {
const VectorStoreEnv vectorEnv;
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
struct RejectingBackend final : storage::FilesystemBackend {
bool rejectBlobs{true};
Result<void> store(std::string_view key, std::span<const std::byte> value) override {
if (rejectBlobs && key.starts_with("blob/")) {
return Error{ErrorCode::InvalidArgument, "backend rejected the write"};
}
return FilesystemBackend::store(key, value);
}
};
auto backend = std::make_unique<RejectingBackend>();
storage::BackendConfig config;
config.localPath = sharedStore;
REQUIRE(backend->initialize(config).has_value());
auto* rejection = backend.get();
MeshNode origin{root.path, "origin", sharedStore, MeshNode::KgReads::Uncounted,
{}, std::move(backend)};
SharedStoreView view{sharedStore};
auto vectors = attachVectorDb(origin);
REQUIRE(vectors->insertVectorChecked(makeVector("retry-chunk", '1', "model")).has_value());
REQUIRE(vectors->insertVectorChecked(makeVector("later-chunk", '2', "model")).has_value());
origin.backfill(2);
view.refresh();
CHECK_FALSE(view.published("embedding/model/retry-chunk"));
const auto status = origin.manager->getMemorySyncStatus();
REQUIRE(status.has_value());
CHECK(status.value().apply.publishFailedCycles > 0);
// The same error code is not enough to classify a row as unpublishable. Once the
// backend recovers, the unchanged source generation must still be retried.
rejection->rejectBlobs = false;
origin.backfill(4);
view.refresh();
CHECK(view.published("embedding/model/retry-chunk"));
CHECK(view.published("embedding/model/later-chunk"));
}
TEST_CASE("Memory sync vector apply stores good embeddings past a bad one",
"[daemon][memory-sync][apply][vector]") {
const VectorStoreEnv vectorEnv;
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode writer{root.path, "writer", sharedStore};
MeshNode reader{root.path, "reader", sharedStore};
auto vectors = attachVectorDb(reader);
// Both embeddings belong to a document whose bytes replicate in the same cycle.
const auto document = bytes("vector-apply-document");
const auto documentHash = digest(document);
REQUIRE(writer.sync_->publish("content-blob/" + documentHash, document).has_value());
// Winners apply in key order; the bad record sorts first so it cannot hide the good one.
memory_sync::EmbeddingRecord bad;
bad.model = "model";
bad.chunkId = "a-bad-dimension";
bad.documentId = documentHash;
bad.dimensions = 4;
bad.values = {1.0F, 0.0F, 0.0F};
memory_sync::EmbeddingRecord good = bad;
good.chunkId = "b-good";
good.values = {0.0F, 1.0F, 0.0F, 0.0F};
REQUIRE(
writer.sync_->publish("embedding/model/a-bad-dimension", bytes(nlohmann::json(bad).dump()))
.has_value());
REQUIRE(writer.sync_->publish("embedding/model/b-good", bytes(nlohmann::json(good).dump()))
.has_value());
reader.runCycle();
CHECK(vectors->getVector("b-good").has_value());
CHECK_FALSE(vectors->getVector("a-bad-dimension").has_value());
// The bad record is a genuine failure: it fails the stage and the cycle, visibly, and the
// stage still scanned every record.
const auto status = reader.manager->getMemorySyncStatus();
REQUIRE(status.has_value());
const auto& apply = status.value().apply;
CHECK(apply.failuresIn(memory_sync::ApplyStage::Vector) == 1);
CHECK(apply.deferredIn(memory_sync::ApplyStage::Vector) == 0);
CHECK(apply.applyCycles == 1);
CHECK(apply.applyFailedCycles == 1);
CHECK(apply.lastFailureStage == "vector");
CHECK(apply.lastFailure.find("embedding/model/a-bad-dimension") != std::string::npos);
}
namespace {
std::string embeddingKey(std::string_view chunkId) {
return "embedding/model/" + std::string(chunkId);
}
std::vector<float> publishedValues(const SharedStoreView& view, std::string_view chunkId) {
const auto cached = view.sync.readCached(embeddingKey(chunkId));
REQUIRE(cached.has_value());
const std::string text(reinterpret_cast<const char*>(cached.value().data()),
cached.value().size());
return nlohmann::json::parse(text).get<memory_sync::EmbeddingRecord>().values;
}
/// Re-embed a one-chunk document: its vector is replaced by one with new values, as an
/// embedding job does when the document's content or model changes.
void reembed(vector::VectorDatabase& vectors, const std::string& chunkId, char documentDigit) {
REQUIRE(vectors.deleteVectorsByDocumentChecked(std::string(64, documentDigit)).has_value());
auto record = makeVector(chunkId, documentDigit, "model");
record.embedding = {0.0F, 0.0F, 1.0F, 0.0F};
REQUIRE(vectors.insertVectorChecked(record).has_value());
}
} // namespace
TEST_CASE("Memory sync vector backfill wraps to publish vectors behind its cursor",
"[daemon][memory-sync][backfill][vector]") {
const VectorStoreEnv vectorEnv;
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore};
SharedStoreView view{sharedStore};
auto vectors = attachVectorDb(origin);
origin.manager->testingSetMemorySyncBackfillItemBudget(1);
REQUIRE(vectors->insertVectorChecked(makeVector("swept-early", '2', "model")).has_value());
REQUIRE(vectors->insertVectorChecked(makeVector("swept-late", '6', "model")).has_value());
// Let the sweep publish both rows and reach the end of the table.
origin.backfill(4);
view.refresh();
REQUIRE(view.published(embeddingKey("swept-early")));
REQUIRE(view.published(embeddingKey("swept-late")));
// Both changes sort below the sweep's last position, (document '6...', "swept-late"), and
// nothing announces them: only a sweep that starts over can reach them.
REQUIRE(vectors->insertVectorChecked(makeVector("added-behind", '1', "model")).has_value());
reembed(*vectors, "swept-early", '2');
origin.backfill(8);
view.refresh();
CHECK(view.published(embeddingKey("added-behind")));
CHECK(publishedValues(view, "swept-early") == std::vector<float>{0.0F, 0.0F, 1.0F, 0.0F});
}
TEST_CASE("Memory sync publishes committed embeddings without waiting for the vector sweep",
"[daemon][memory-sync][backfill][vector]") {
const VectorStoreEnv vectorEnv;
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore};
SharedStoreView view{sharedStore};
auto vectors = attachVectorDb(origin);
origin.manager->testingSetMemorySyncBackfillItemBudget(1);
// A bulk document sorts first, so a sweep starting over spends many items before it could
// reach anything after it.
constexpr int kBulkChunks = 6;
for (int i = 0; i < kBulkChunks; ++i) {
REQUIRE(vectors->insertVectorChecked(makeVector("bulk-" + std::to_string(i), '1', "model"))
.has_value());
}
REQUIRE(vectors->insertVectorChecked(makeVector("reembedded", '5', "model")).has_value());
origin.backfill(kBulkChunks + 3);
view.refresh();
REQUIRE(view.published(embeddingKey("reembedded")));
// An embedding job commits a new document's vectors and re-embeds an existing one.
REQUIRE(vectors->insertVectorChecked(makeVector("committed-new", '3', "model")).has_value());
reembed(*vectors, "reembedded", '5');
origin.manager->notifyMemorySyncEmbeddingsCommitted(
{std::string(64, '3'), std::string(64, '5')});
// One item per committed vector: fewer than the sweep needs to get past the bulk document.
origin.backfill(2);
view.refresh();
CHECK(view.published(embeddingKey("committed-new")));
CHECK(publishedValues(view, "reembedded") == std::vector<float>{0.0F, 0.0F, 1.0F, 0.0F});
}
TEST_CASE("Memory sync topology backfill keeps progressing while the vector sweep wraps",
"[daemon][memory-sync][backfill][vector][topology]") {
const VectorStoreEnv vectorEnv;
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore};
SharedStoreView view{sharedStore};
auto vectors = attachVectorDb(origin);
origin.manager->testingSetMemorySyncBackfillItemBudget(1);
constexpr int kVectors = 8;
constexpr int kNodes = 4;
for (int i = 0; i < kVectors; ++i) {
REQUIRE(vectors->insertVectorChecked(makeVector("sweep-" + std::to_string(i), '1', "model"))
.has_value());
}
std::vector<std::string> nodes;
for (int i = 0; i < kNodes; ++i) {
nodes.push_back(origin.addNode("doc:fair-" + std::to_string(i), "document").nodeKey);
}
// The vector sweep has more items than these cycles can serve. They are fewer than the two
// sweeps need back to back, but enough for each to publish its share when neither can take
// every item.
origin.backfill(2 * kNodes + 2);
view.refresh();
for (const auto& key : nodes) {
INFO("node=" << key);
CHECK(view.published(topologyNodeKey(key)));
}
int vectorsPublished = 0;
for (int i = 0; i < kVectors; ++i) {
vectorsPublished += view.published(embeddingKey("sweep-" + std::to_string(i))) ? 1 : 0;
}
CHECK(vectorsPublished >= kNodes);
}
namespace {
/// Counts the items the outbound catch-up sweeps serve: one stage notification per node, edge,
/// vector, or retraction the sweeps publish or re-check.
struct SweepItemCounter {
explicit SweepItemCounter(MeshNode& node) {
node.manager->testingSetMemorySyncStageObserver([this](std::string_view stage) {
if (stage == "backfill.after_node" || stage == "backfill.after_edge" ||
stage == "backfill.after_vector" || stage == "backfill.after_retraction") {
++count;
}
});
}
std::size_t count{0};
};
/// The operation identity of the committed winner for `key`: it changes only when the record
/// is published again.
std::string winnerOperation(const SharedStoreView& view, const std::string& key) {
const auto winners = view.sync.committedWinnersAfter(key, "", 1);
REQUIRE(winners.size() == 1);
REQUIRE(winners.front().first == key);
return winners.front().second.operationId;
}
/// A small corpus whose first sweep pass completes within a few cycles: two documents, an edge
/// between them, and one vector.
struct IdleCorpus {
explicit IdleCorpus(MeshNode& node, vector::VectorDatabase& vectors)
: first(node.addNode("doc:idle-first", "document")),
second(node.addNode("doc:idle-second", "document")) {
node.addEdge(first, second);
REQUIRE(vectors.insertVectorChecked(makeVector("idle-chunk", '4', "model")).has_value());
}
/// Enough cycles for both sweeps to finish a full pass over this corpus.
void publishFirstPass(MeshNode& node, SharedStoreView& view) const {
node.backfill(8);
view.refresh();
REQUIRE(view.published(topologyNodeKey(first.nodeKey)));
REQUIRE(view.published(topologyNodeKey(second.nodeKey)));
REQUIRE(view.published(topologyEdgeKey(first.nodeKey, second.nodeKey)));
REQUIRE(view.published(embeddingKey("idle-chunk")));
}
/// Finish the first pass, then verify further cycles serve nothing.
void settle(MeshNode& node, SharedStoreView& view, SweepItemCounter& items) const {
publishFirstPass(node, view);
REQUIRE(items.count > 0);
const auto served = items.count;
node.backfill(2);
REQUIRE(items.count == served);
}
metadata::KGNode first;
metadata::KGNode second;
};
} // namespace
TEST_CASE("Memory sync catch-up sweeps stop once a pass finds the stores unchanged",
"[daemon][memory-sync][backfill][idle]") {
const VectorStoreEnv vectorEnv;
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore, MeshNode::KgReads::Counted};
SharedStoreView view{sharedStore};
auto vectors = attachVectorDb(origin);
SweepItemCounter items{origin};
const IdleCorpus corpus{origin, *vectors};
corpus.publishFirstPass(origin, view);
const auto itemsAfterPass = items.count;
const auto readsAfterPass = origin.kgStoreAcquisitions();
REQUIRE(itemsAfterPass > 0);
// Nothing changed: later cycles serve no sweep item and never touch the graph store.
origin.backfill(5);
CHECK(items.count == itemsAfterPass);
CHECK(origin.kgStoreAcquisitions() == readsAfterPass);
}
TEST_CASE("Memory sync catch-up sweeps resume for records written after a completed pass",
"[daemon][memory-sync][backfill][idle]") {
const VectorStoreEnv vectorEnv;
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore};
SharedStoreView view{sharedStore};
auto vectors = attachVectorDb(origin);
SweepItemCounter items{origin};
const IdleCorpus corpus{origin, *vectors};
corpus.settle(origin, view, items);
const auto firstOperation = winnerOperation(view, topologyNodeKey(corpus.first.nodeKey));
const auto chunkOperation = winnerOperation(view, embeddingKey("idle-chunk"));
// A node, an edge to it, and a vector land behind the idle cursors.
const auto third = origin.addNode("doc:idle-third", "document");
origin.addEdge(corpus.first, third);
REQUIRE(vectors->insertVectorChecked(makeVector("added-later", '2', "model")).has_value());
origin.backfill(8);
view.refresh();
CHECK(view.published(topologyNodeKey(third.nodeKey)));
CHECK(view.published(topologyEdgeKey(corpus.first.nodeKey, third.nodeKey)));
CHECK(view.published(embeddingKey("added-later")));
// Records that did not change are not published again.
CHECK(winnerOperation(view, topologyNodeKey(corpus.first.nodeKey)) == firstOperation);
CHECK(winnerOperation(view, embeddingKey("idle-chunk")) == chunkOperation);
}
TEST_CASE("Memory sync retracts records deleted after the sweeps went idle",
"[daemon][memory-sync][backfill][idle][retraction]") {
const VectorStoreEnv vectorEnv;
TempRoot root;
const auto sharedStore = root.path / "shared-store";
fs::create_directories(sharedStore);
MeshNode origin{root.path, "origin", sharedStore};
SharedStoreView view{sharedStore};
auto vectors = attachVectorDb(origin);
SweepItemCounter items{origin};
const IdleCorpus corpus{origin, *vectors};
corpus.settle(origin, view, items);
// Deleting the second node cascades the edge; both must be retracted by a new pass.
origin.deleteNode(corpus.second);
for (int round = 0; round < 3; ++round) {
origin.runCycle();
}
view.refresh();
CHECK(view.tombstoned(topologyNodeKey(corpus.second.nodeKey)));
CHECK(view.tombstoned(topologyEdgeKey(corpus.first.nodeKey, corpus.second.nodeKey)));
CHECK(view.published(topologyNodeKey(corpus.first.nodeKey)));
}
Reference in a new issue View git blame Copy permalink