trevon/yams
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1
yams/tests/unit/daemon/memory_sync_apply_catch2_test.cpp
Trevon d59db71d86
 perf(memory-sync): skip inbound apply when replicated inputs are unchanged (#344) 
* perf(profiling): attribute metadata pool waits and memory-sync stages

MetadataPool::acquire and memory_sync.cycle were the largest self-time
zones in release-prep profiles, but neither said where the time went.

- connection_pool.cpp: zone the caller tag, the pool mutex, the wait
  for a free connection and the SELECT 1 validation probe.
- memory_sync.cpp: zone the metadata, vector and topology apply stages
  (content blobs and delete drain already had zones).

On grim (gcc Profiling) the split shows acquire itself at 0.4 us and
validation at 0.6 us; SciFact's 15.2 s in acquire is 962 waits for the
single-connection write pool during ingest, none in the query window.
An idle 3-node mesh spends ~800 ms per ~6 s cycle in apply_topology,
~370 ms in apply_vectors and ~104 ms in apply_metadata.

* perf(memory-sync): skip inbound apply when replicated inputs are unchanged

An idle mesh node spent about a quarter of a core re-applying the same
winners every ~6 s cycle: each pass re-read and re-parsed every record
and re-checked the local stores (topology ~1 s, vectors ~360 ms,
metadata ~100 ms per cycle on a 1.5k-record corpus).

Skip the inbound stages when an adapter pass would read the same
records as the last complete, clean pass:

- WinnerView::identity(): a process-unique store instance plus the
  winner generation, which advances on every select, discard and ban.
  No identity while the store is building or unhealthy.
- MemorySyncService::AdapterInputs: that identity plus the overlay
  rows. Equal rows compare without lookups; otherwise only rows an
  adapter can see (keys the winner store lacks) count, so resident
  cache churn does not read as a change.
- A pass is clean when every stage scans completely with no deferral,
  failure or admission deferral. Unknown inputs never skip.
- A full pass still runs every 5 minutes, so a replicated record
  deleted or changed locally is repaired within that bound.
- A skipped cycle still counts as an apply cycle and still runs the
  outbound backfill.
- Integration tests that observe apply stages per call disable the
  skip through a lock-free test hook.

Paired A/B on grim (gcc Profiling, 3 nodes x 2 cpu / 2 GiB, records
plateaued, concurrent 600 s idle sample):

  idle CPU per node   base 24.4 / 25.0 / 23.4 %   change 3.3 / 3.9 / 4.4 %
  300 s profile, n2   full passes 46 -> 1; apply_inputs 0.55 ms/cycle
                      VectorDB::getVector 11,040 -> 240
2026-10-25 14:58:25 -06:00

1138 lines
48 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)));
}
namespace {
/// Publish `sourceKey -> targetKey` from `writer` with both endpoints, as its backfill would.
void publishEdgeWithEndpoints(MeshNode& writer, const std::string& sourceKey,
const std::string& targetKey) {
const auto source = writer.addNode(sourceKey, "alpha");
const auto target = writer.addNode(targetKey, "omega");
writer.addEdge(source, target);
const metadata::KGEdge edge{.srcNodeId = source.id,
.dstNodeId = target.id,
.relation = std::string(kRelation),
.weight = 1.0F};
metadata::TopologySyncAdapter publisher{*writer.kgStore, *writer.sync_};
REQUIRE(publisher.publishEdgeWithEndpoints(source, edge, target).has_value());
}
/// Apply until `reader` holds the edge, then once more so a clean pass is on record.
void applyUntilEdge(MeshNode& reader, const std::string& sourceKey, const std::string& targetKey) {
for (int round = 0; round < 4 && !reader.hasEdge(sourceKey, targetKey); ++round) {
reader.manager->testingApplyMemorySyncWinners();
}
REQUIRE(reader.hasEdge(sourceKey, targetKey));
reader.manager->testingApplyMemorySyncWinners();
}
std::uint64_t applyCycles(const MeshNode& node) {
const auto status = node.manager->getMemorySyncStatus();
REQUIRE(status.has_value());
return status.value().apply.applyCycles;
}
} // namespace
TEST_CASE("Memory sync apply skips cycles whose replicated inputs have not changed",
"[daemon][memory-sync][apply][idle]") {
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, MeshNode::KgReads::Counted};
publishEdgeWithEndpoints(writer, "entity:idle:source", "entity:idle:target");
applyUntilEdge(reader, "entity:idle:source", "entity:idle:target");
const auto skipped = reader.manager->testingMemorySyncApplySkippedCycles();
const auto reads = reader.kgStoreAcquisitions();
const auto cycles = applyCycles(reader);
for (int round = 0; round < 4; ++round) {
reader.manager->testingApplyMemorySyncWinners();
}
// Unchanged inputs: the inbound stages never run and never touch the graph store, but each
// cycle still counts as a healthy apply cycle.
CHECK(reader.manager->testingMemorySyncApplySkippedCycles() == skipped + 4);
CHECK(reader.kgStoreAcquisitions() == reads);
CHECK(applyCycles(reader) == cycles + 4);
CHECK(reader.hasEdge("entity:idle:source", "entity:idle:target"));
}
TEST_CASE("Memory sync apply runs again as soon as a new winner arrives",
"[daemon][memory-sync][apply][idle]") {
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};
publishEdgeWithEndpoints(writer, "entity:first:source", "entity:first:target");
applyUntilEdge(reader, "entity:first:source", "entity:first:target");
reader.manager->testingApplyMemorySyncWinners();
const auto skipped = reader.manager->testingMemorySyncApplySkippedCycles();
REQUIRE(skipped > 0);
publishEdgeWithEndpoints(writer, "entity:second:source", "entity:second:target");
reader.manager->testingApplyMemorySyncWinners();
CHECK(reader.manager->testingMemorySyncApplySkippedCycles() == skipped);
CHECK(reader.hasEdge("entity:second:source", "entity:second:target"));
CHECK(reader.hasEdge("entity:first:source", "entity:first:target"));
}
TEST_CASE("Memory sync full apply repairs a replicated record deleted locally",
"[daemon][memory-sync][apply][idle]") {
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};
publishEdgeWithEndpoints(writer, "entity:drift:source", "entity:drift:target");
applyUntilEdge(reader, "entity:drift:source", "entity:drift:target");
// Local drift: the replicated source node (and, by cascade, its edge) disappears locally
// while its winner stays in the replicated index.
const auto local = reader.kgStore->getNodeByKey("entity:drift:source");
REQUIRE(local.has_value());
REQUIRE(local.value().has_value());
reader.deleteNode(*local.value());
// Before the full-pass deadline the cycle skips, so the drift stays.
reader.manager->testingApplyMemorySyncWinners();
CHECK_FALSE(reader.hasNode("entity:drift:source"));
// The next full pass restores it.
reader.manager->testingExpireMemorySyncFullApply();
reader.manager->testingApplyMemorySyncWinners();
CHECK(reader.hasNode("entity:drift:source"));
CHECK(reader.hasEdge("entity:drift:source", "entity:drift:target"));
}
TEST_CASE("Memory sync apply never skips a cycle that deferred records",
"[daemon][memory-sync][apply][idle]") {
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, MeshNode::KgReads::Counted};
// The writer's edge reaches the shared store ahead of its target node.
writer.seedEdgePublishedAheadOfTarget();
reader.manager->testingApplyMemorySyncWinners();
REQUIRE(reader.deferredTopology() == 1);
const auto reads = reader.kgStoreAcquisitions();
reader.manager->testingApplyMemorySyncWinners();
reader.manager->testingApplyMemorySyncWinners();
// A deferral waits on a local prerequisite, so every cycle keeps retrying it.
CHECK(reader.manager->testingMemorySyncApplySkippedCycles() == 0);
CHECK(reader.kgStoreAcquisitions() > reads);
CHECK(reader.deferredTopology() == 1);
}
Reference in a new issue View git blame Copy permalink