trevon/yams
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2
yams/tests/unit/daemon/topology_manager_catch2_test.cpp
Trevon 3f3f3b613e
 fix(topology): full rebuild when louvain/kmeans need a wider region (#285) 
* test(topology): cover incremental rebuilds for louvain and kmeans

An incremental TopologyManager rebuild with the louvain or kmeans engine
publishes a snapshot that holds only the dirty documents. The store then
drops topology.cluster_id from every untouched document. Add a failing
case for both engines, and a regression case that the connected engine
still rebuilds only the seed's prior cluster.

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

* fix(topology): rebuild the corpus when a dirty region needs it

TopologyManager logged requiresWiderRebuild and then published the
engine's region update anyway. Louvain and KMeans have no incremental
path: their dirty region is the seed set and their update rebuilds only
those documents, so the published snapshot held just the dirty docs and
the store cleared the topology keys of every other document.

Honor the flag: re-extract the full corpus and call buildArtifacts, and
report the run as a full rebuild with one fallback. The connected engine
sets the flag only when a region exceeds its seed or document budget;
under budget its incremental path is unchanged. Over budget it used to
drop seeds past the cap, and now rebuilds the corpus.

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

---------

Signed-off-by: trvon <me@trevon.dev>
2026-10-16 09:43:40 -06:00

621 lines
26 KiB
C++
Raw Permalink Blame History

#include <catch2/catch_test_macros.hpp>
#include <catch2/generators/catch_generators.hpp>
#include <algorithm>
#include <cstdint>
#include <map>
#include <memory>
#include <optional>
#include <string>
#include <thread>
#include <unordered_set>
#include <vector>
#include <yams/daemon/components/TopologyManager.h>
#include <yams/metadata/connection_pool.h>
#include <yams/metadata/knowledge_graph_store.h>
#include <yams/metadata/metadata_repository.h>
#include <yams/topology/topology_metadata_store.h>
#include <yams/vector/vector_database.h>
#include "tests/common/test_helpers_catch2.h"
using namespace yams::daemon;
static TopologyManager::Dependencies nullDeps() {
return {
[] { return std::shared_ptr<yams::metadata::MetadataRepository>{}; },
[] { return std::shared_ptr<yams::metadata::KnowledgeGraphStore>{}; },
[] { return std::shared_ptr<yams::vector::VectorDatabase>{}; },
};
}
TEST_CASE("TopologyManager markDirty and hasDirtyHashes", "[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
CHECK_FALSE(mgr.hasDirtyHashes());
mgr.markDirty("hash1");
CHECK(mgr.hasDirtyHashes());
}
TEST_CASE("TopologyManager markDirty ignores empty strings", "[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
mgr.markDirty("");
CHECK_FALSE(mgr.hasDirtyHashes());
}
TEST_CASE("TopologyManager markDirtyBatch inserts multiple hashes",
"[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
mgr.markDirtyBatch({"a", "b", "", "c"});
auto overlay = mgr.getOverlayHashes(100);
CHECK(overlay.size() == 3);
}
TEST_CASE("TopologyManager getOverlayHashes respects limit", "[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
mgr.markDirtyBatch({"a", "b", "c", "d", "e"});
auto limited = mgr.getOverlayHashes(2);
CHECK(limited.size() == 2);
}
TEST_CASE("TopologyManager drainDirtyHashes clears state", "[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
mgr.markDirtyBatch({"x", "y", "z"});
auto drained = mgr.drainDirtyHashes();
CHECK(drained.size() == 3);
CHECK_FALSE(mgr.hasDirtyHashes());
}
TEST_CASE("TopologyManager restoreDirtyHashes re-inserts", "[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
mgr.markDirty("a");
auto drained = mgr.drainDirtyHashes();
CHECK_FALSE(mgr.hasDirtyHashes());
mgr.restoreDirtyHashes(drained);
CHECK(mgr.hasDirtyHashes());
CHECK(mgr.getOverlayHashes().size() == 1);
}
TEST_CASE("TopologyManager tryScheduleRebuild is one-shot", "[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
CHECK(mgr.tryScheduleRebuild());
CHECK_FALSE(mgr.tryScheduleRebuild());
mgr.clearScheduled();
CHECK(mgr.tryScheduleRebuild());
}
TEST_CASE("TopologyManager initial state", "[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
CHECK_FALSE(mgr.isRebuildInProgress());
CHECK(mgr.publishedEpoch() == 0);
auto telemetry = mgr.getTelemetrySnapshot();
CHECK_FALSE(telemetry.rebuildRunning);
CHECK(telemetry.rebuildsTotal == 0);
CHECK(telemetry.dirtyDocumentCount == 0);
}
TEST_CASE("TopologyManager roads are off until configured", "[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
CHECK(mgr.roadMaxPerCluster() == 0);
mgr.setRoadMaxPerCluster(4);
CHECK(mgr.roadMaxPerCluster() == 4);
mgr.setRoadMaxPerCluster(0);
CHECK(mgr.roadMaxPerCluster() == 0);
CHECK(mgr.roadOrder() == yams::topology::RoadOrderKind::None);
mgr.setRoadOrder(yams::topology::RoadOrderKind::SimilarityChain);
CHECK(mgr.roadOrder() == yams::topology::RoadOrderKind::SimilarityChain);
mgr.setRoadOrder(yams::topology::RoadOrderKind::None);
CHECK(mgr.roadOrder() == yams::topology::RoadOrderKind::None);
}
TEST_CASE("TopologyManager telemetry tracks dirty count", "[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
mgr.markDirtyBatch({"a", "b"});
auto telemetry = mgr.getTelemetrySnapshot();
CHECK(telemetry.dirtyDocumentCount == 2);
CHECK(telemetry.dirtySinceUnixMillis > 0);
}
TEST_CASE("TopologyManager rebuildArtifacts fails without deps",
"[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
auto result = mgr.rebuildArtifacts("test", false, {"hash1"}, "connected");
CHECK_FALSE(result.has_value());
auto telemetry = mgr.getTelemetrySnapshot();
CHECK(telemetry.rebuildsTotal == 1);
CHECK(telemetry.rebuildFailuresTotal == 1);
CHECK_FALSE(telemetry.rebuildRunning);
}
TEST_CASE("TopologyManager rebuildArtifacts concurrency guard",
"[daemon][topology-manager][catch2]") {
TopologyManager mgr(nullDeps());
std::atomic<bool> firstStarted{false};
std::atomic<bool> secondDone{false};
TopologyManager::Dependencies slowDeps;
slowDeps.getMetadataRepo = [&]() -> std::shared_ptr<yams::metadata::MetadataRepository> {
firstStarted.store(true);
while (!secondDone.load(std::memory_order_relaxed))
std::this_thread::yield();
return nullptr;
};
slowDeps.getKgStore = nullDeps().getKgStore;
slowDeps.getVectorDatabase = nullDeps().getVectorDatabase;
TopologyManager slowMgr(std::move(slowDeps));
std::thread t1([&] { slowMgr.rebuildArtifacts("first", false, {"h"}, "connected"); });
while (!firstStarted.load(std::memory_order_relaxed))
std::this_thread::yield();
auto result = slowMgr.rebuildArtifacts("second", false, {"h"}, "connected");
CHECK(result.has_value());
CHECK(result.value().skipped);
CHECK(result.value().issues.size() == 1);
secondDone.store(true);
t1.join();
}
namespace {
struct RebuildFixture {
yams::test::ScopedEnvVar enableVectors{"YAMS_DISABLE_VECTORS", std::string{"0"}};
yams::test::ScopedEnvVar initializeVectors{"YAMS_SQLITE_VEC_SKIP_INIT", std::string{"0"}};
yams::test::ScopedEnvVar vectorMemoryMode{"YAMS_VDB_IN_MEMORY", std::nullopt};
yams::test::TempDirGuard directory{"topology_rebuild_"};
std::unique_ptr<yams::metadata::ConnectionPool> pool;
std::shared_ptr<yams::metadata::MetadataRepository> repository;
std::shared_ptr<yams::metadata::KnowledgeGraphStore> kg;
std::shared_ptr<yams::vector::VectorDatabase> vectors;
RebuildFixture() {
yams::metadata::ConnectionPoolConfig poolConfig;
poolConfig.minConnections = 1;
poolConfig.maxConnections = 2;
pool = std::make_unique<yams::metadata::ConnectionPool>(
(directory.path() / "metadata.db").string(), poolConfig);
REQUIRE(pool->initialize().has_value());
repository = std::make_shared<yams::metadata::MetadataRepository>(*pool);
auto kgResult = yams::metadata::makeSqliteKnowledgeGraphStore(
*pool, yams::metadata::KnowledgeGraphStoreConfig{});
REQUIRE(kgResult.has_value());
kg = std::shared_ptr<yams::metadata::KnowledgeGraphStore>(std::move(kgResult.value()));
yams::vector::VectorDatabaseConfig vectorConfig;
vectorConfig.database_path = ":memory:";
vectorConfig.embedding_dim = 4;
vectorConfig.create_if_missing = true;
vectorConfig.use_in_memory = true;
vectors = std::make_shared<yams::vector::VectorDatabase>(vectorConfig);
REQUIRE(vectors->initialize());
}
std::int64_t addDocument(const std::string& hash,
std::vector<float> embedding = {1.0F, 0.0F, 0.0F, 0.0F}) {
yams::metadata::DocumentInfo document;
document.fileName = hash + ".md";
document.filePath = (directory.path() / document.fileName).string();
document.sha256Hash = hash;
document.mimeType = "text/plain";
document.contentExtracted = true;
document.extractionStatus = yams::metadata::ExtractionStatus::Success;
auto inserted = repository->insertDocument(document);
REQUIRE(inserted.has_value());
yams::metadata::KGNode node;
node.nodeKey = "doc:" + hash;
node.label = hash;
node.type = "document";
REQUIRE(kg->upsertNode(node).has_value());
yams::vector::VectorRecord vector;
vector.chunk_id = "doc-" + hash;
vector.document_hash = hash;
vector.embedding = std::move(embedding);
vector.model_id = "test-space-v1";
vector.level = yams::vector::EmbeddingLevel::DOCUMENT;
REQUIRE(vectors->insertVector(vector));
return inserted.value();
}
void linkNeighbors(const std::string& lhs, const std::string& rhs) {
auto lhsNode = kg->getNodeByKey("doc:" + lhs);
auto rhsNode = kg->getNodeByKey("doc:" + rhs);
REQUIRE(lhsNode.has_value());
REQUIRE(lhsNode.value().has_value());
REQUIRE(rhsNode.has_value());
REQUIRE(rhsNode.value().has_value());
yams::metadata::KGEdge forward;
forward.srcNodeId = lhsNode.value()->id;
forward.dstNodeId = rhsNode.value()->id;
forward.relation = "semantic_neighbor";
auto reverse = forward;
std::swap(reverse.srcNodeId, reverse.dstNodeId);
REQUIRE(kg->addEdgesUnique(std::vector{forward, reverse}).has_value());
}
TopologyManager::Dependencies dependencies() {
return {[this] { return repository; }, [this] { return kg; }, [this] { return vectors; }};
}
yams::topology::TopologyArtifactBatch latest() {
// A fresh store must observe the persisted pointer, not an earlier local cache.
yams::topology::MetadataKgTopologyArtifactStore store(repository, kg);
auto loaded = store.loadLatest();
REQUIRE(loaded.has_value());
REQUIRE(loaded.value().has_value());
return std::move(*loaded.value());
}
};
} // namespace
TEST_CASE("TopologyManager does not reuse deleted prior members", "[daemon][topology-recovery]") {
RebuildFixture fixture;
const auto removedId = fixture.addDocument("aaa");
fixture.addDocument("bbb");
fixture.addDocument("ccc");
// The removed document shares a cluster with a live member. Recovery must rebuild
// representatives and retain that live member, not merely discard the whole cluster.
auto aNode = fixture.kg->getNodeByKey("doc:aaa");
auto bNode = fixture.kg->getNodeByKey("doc:bbb");
REQUIRE(aNode.has_value());
REQUIRE(aNode.value().has_value());
REQUIRE(bNode.has_value());
REQUIRE(bNode.value().has_value());
yams::metadata::KGEdge forward;
forward.srcNodeId = aNode.value()->id;
forward.dstNodeId = bNode.value()->id;
forward.relation = "semantic_neighbor";
auto reverse = forward;
std::swap(reverse.srcNodeId, reverse.dstNodeId);
REQUIRE(fixture.kg->addEdgesUnique(std::vector{forward, reverse}).has_value());
TopologyManager manager(fixture.dependencies());
auto initial = manager.rebuildArtifacts("initial", false, {}, "connected");
REQUIRE(initial.has_value());
REQUIRE(initial.value().stored);
const auto previous = fixture.latest();
REQUIRE(previous.memberships.size() == 3);
REQUIRE(previous.clusters.size() == 2);
bool deleted = true;
bool dryRun = false;
std::vector<std::string> seeds{"ccc"};
SECTION("healthy incremental rebuild retains untouched live members") {
deleted = false;
}
SECTION("deleted member is outside the requested dirty region") {}
SECTION("the only dirty seed was deleted") {
seeds = {"aaa"};
}
SECTION("dry run detects stale reuse without publishing") {
dryRun = true;
}
if (deleted) {
REQUIRE(fixture.repository->deleteDocument(removedId).has_value());
// Stale vector/KG state must not resurrect a document absent from metadata.
REQUIRE_FALSE(fixture.vectors->getVectorsByDocument("aaa").empty());
}
// Work not included in this request must survive even if recovery needs a full rebuild.
manager.markDirty("new-pending");
auto result = manager.rebuildArtifacts("post_ingest_drain", dryRun, seeds, "connected");
INFO((result ? "rebuild succeeded" : result.error().message));
REQUIRE(result.has_value());
CHECK_FALSE(result.value().skipped);
CHECK(result.value().stored == !dryRun);
CHECK(result.value().fullRebuild == deleted);
CHECK(result.value().fallbackFullRebuilds == (deleted ? 1 : 0));
CHECK(result.value().membershipsBuilt == (deleted ? 2 : 3));
CHECK(manager.getOverlayHashes() == std::vector<std::string>{"new-pending"});
const auto current = fixture.latest();
CHECK(current.topologyEpoch == previous.topologyEpoch + (dryRun ? 0 : 1));
CHECK(manager.publishedEpoch() == current.topologyEpoch);
if (dryRun) {
CHECK(current.snapshotId == previous.snapshotId);
CHECK(current.memberships.size() == previous.memberships.size());
} else {
const std::unordered_set<std::string> expected =
deleted ? std::unordered_set<std::string>{"bbb", "ccc"}
: std::unordered_set<std::string>{"aaa", "bbb", "ccc"};
std::unordered_set<std::string> members;
for (const auto& membership : current.memberships) {
members.insert(membership.documentHash);
}
CHECK(members == expected);
CHECK(current.clusters.size() == 2);
for (const auto& cluster : current.clusters) {
REQUIRE(cluster.medoid.has_value());
CHECK(expected.contains(cluster.medoid->documentHash));
for (const auto& representative : cluster.routingRepresentatives) {
CHECK(expected.contains(representative.documentHash));
}
CHECK(cluster.memberCount == cluster.memberDocumentHashes.size());
for (const auto& hash : cluster.memberDocumentHashes) {
CHECK(expected.contains(hash));
}
}
}
}
TEST_CASE("TopologyManager incremental rebuilds honor the configured dirty-region expansion",
"[daemon][topology-manager][expansion]") {
// The mode had no path from config to the incremental rebuild, so every rebuild used the
// default. aaa and bbb are semantic neighbors in one prior cluster; a rebuild seeded from bbb
// pulls in that cluster's members only when the mode includes prior clusters.
RebuildFixture fixture;
fixture.addDocument("aaa");
fixture.addDocument("bbb");
fixture.addDocument("ccc");
auto aNode = fixture.kg->getNodeByKey("doc:aaa");
auto bNode = fixture.kg->getNodeByKey("doc:bbb");
REQUIRE(aNode.has_value());
REQUIRE(aNode.value().has_value());
REQUIRE(bNode.has_value());
REQUIRE(bNode.value().has_value());
yams::metadata::KGEdge forward;
forward.srcNodeId = aNode.value()->id;
forward.dstNodeId = bNode.value()->id;
forward.relation = "semantic_neighbor";
auto reverse = forward;
std::swap(reverse.srcNodeId, reverse.dstNodeId);
REQUIRE(fixture.kg->addEdgesUnique(std::vector{forward, reverse}).has_value());
TopologyManager manager(fixture.dependencies());
REQUIRE(manager.rebuildArtifacts("initial", false, {}, "connected").has_value());
auto expansionIssue = [&](const TopologyManager::RebuildStats& stats) {
for (const auto& issue : stats.issues) {
if (issue.starts_with("dirty-region expansion=")) {
return issue;
}
}
return std::string{};
};
SECTION("default includes the seed's prior cluster") {
auto result = manager.rebuildArtifacts("post_ingest_drain", true, {"bbb"}, "connected");
REQUIRE(result.has_value());
CHECK((expansionIssue(result.value()) == "dirty-region expansion=prior_cluster+neighbors"));
}
SECTION("neighbors_only stops at semantic neighbors") {
manager.setDirtyRegionExpansion(yams::topology::DirtyRegionExpansionMode::NeighborsOnly);
auto result = manager.rebuildArtifacts("post_ingest_drain", true, {"bbb"}, "connected");
REQUIRE(result.has_value());
CHECK((expansionIssue(result.value()) == "dirty-region expansion=neighbors"));
}
}
TEST_CASE("TopologyManager distinguishes an empty corpus from unready documents",
"[daemon][topology-recovery]") {
RebuildFixture fixture;
const auto removedId = fixture.addDocument("aaa");
const auto otherId = fixture.addDocument("bbb");
TopologyManager manager(fixture.dependencies());
auto initial = manager.rebuildArtifacts("initial", false, {}, "connected");
REQUIRE(initial.has_value());
REQUIRE(initial.value().stored);
const auto previous = fixture.latest();
REQUIRE(fixture.repository->deleteDocument(removedId).has_value());
bool emptyCorpus = false;
SECTION("all documents were deleted") {
REQUIRE(fixture.repository->deleteDocument(otherId).has_value());
emptyCorpus = true;
}
SECTION("a remaining document is temporarily missing embeddings") {
REQUIRE(fixture.vectors->deleteVectorsByDocument("bbb"));
}
manager.markDirty("new-pending");
auto result = manager.rebuildArtifacts("post_ingest_drain", false, {"aaa"}, "connected");
INFO((result ? "rebuild succeeded" : result.error().message));
REQUIRE(result.has_value());
CHECK(result.value().skipped == !emptyCorpus);
CHECK(result.value().stored == emptyCorpus);
CHECK(result.value().fullRebuild);
CHECK(result.value().fallbackFullRebuilds == 1);
CHECK(manager.getOverlayHashes() == std::vector<std::string>{"new-pending"});
const auto current = fixture.latest();
CHECK(manager.publishedEpoch() == current.topologyEpoch);
if (emptyCorpus) {
CHECK(current.topologyEpoch == previous.topologyEpoch + 1);
CHECK(current.memberships.empty());
CHECK(current.clusters.empty());
} else {
CHECK(result.value().documentsMissingEmbeddings == 1);
CHECK(current.snapshotId == previous.snapshotId);
CHECK(current.topologyEpoch == previous.topologyEpoch);
CHECK(current.memberships.size() == previous.memberships.size());
}
}
TEST_CASE("Topology publication rejects deletion after a successful membership check",
"[daemon][topology-recovery]") {
RebuildFixture fixture;
const auto liveId = fixture.addDocument("aaa");
const auto removedId = fixture.addDocument("bbb");
TopologyManager manager(fixture.dependencies());
auto initial = manager.rebuildArtifacts("initial", false, {}, "connected");
REQUIRE(initial.has_value());
REQUIRE(initial.value().stored);
const auto previous = fixture.latest();
auto existing = fixture.repository->getExistingDocumentHashes({"aaa", "bbb"});
REQUIRE(existing.has_value());
REQUIRE(existing.value().size() == 2);
REQUIRE(fixture.repository->deleteDocument(removedId).has_value());
auto candidate = previous;
candidate.snapshotId = "candidate-after-deletion";
++candidate.topologyEpoch;
std::sort(
candidate.memberships.begin(), candidate.memberships.end(),
[](const auto& left, const auto& right) { return left.documentHash < right.documentHash; });
yams::topology::MetadataKgTopologyArtifactStore store(fixture.repository, fixture.kg);
const auto result = store.storeBatch(candidate);
REQUIRE_FALSE(result.has_value());
CHECK(result.error().code == yams::ErrorCode::NotFound);
const auto current = fixture.latest();
CHECK(current.snapshotId == previous.snapshotId);
CHECK(current.topologyEpoch == previous.topologyEpoch);
CHECK(current.memberships.size() == previous.memberships.size());
const auto liveSnapshot = fixture.repository->getMetadata(liveId, "topology.snapshot_id");
REQUIRE(liveSnapshot.has_value());
REQUIRE(liveSnapshot.value().has_value());
CHECK(liveSnapshot.value()->asString() == previous.snapshotId);
}
namespace {
// Two well-separated groups: the aaa* documents share one embedding direction and are mutual
// semantic neighbors, as are the bbb* documents.
std::map<std::string, std::int64_t> addTwoGroupCorpus(RebuildFixture& fixture) {
const std::vector<std::string> groupA{"aaa1", "aaa2", "aaa3"};
const std::vector<std::string> groupB{"bbb1", "bbb2", "bbb3"};
std::map<std::string, std::int64_t> ids;
for (const auto& hash : groupA) {
ids[hash] = fixture.addDocument(hash, {1.0F, 0.0F, 0.0F, 0.0F});
}
for (const auto& hash : groupB) {
ids[hash] = fixture.addDocument(hash, {0.0F, 1.0F, 0.0F, 0.0F});
}
for (const auto* group : {&groupA, &groupB}) {
for (std::size_t i = 0; i < group->size(); ++i) {
for (std::size_t j = i + 1; j < group->size(); ++j) {
fixture.linkNeighbors((*group)[i], (*group)[j]);
}
}
}
return ids;
}
std::map<std::string, std::string>
membershipClusters(const yams::topology::TopologyArtifactBatch& batch) {
std::map<std::string, std::string> clusters;
for (const auto& membership : batch.memberships) {
clusters[membership.documentHash] = membership.clusterId;
}
return clusters;
}
std::map<std::string, std::vector<std::string>>
clusterMembers(const yams::topology::TopologyArtifactBatch& batch) {
std::map<std::string, std::vector<std::string>> members;
for (const auto& cluster : batch.clusters) {
auto hashes = cluster.memberDocumentHashes;
std::sort(hashes.begin(), hashes.end());
members[cluster.clusterId] = std::move(hashes);
}
return members;
}
bool hasIssue(const TopologyManager::RebuildStats& stats, const std::string& issue) {
return std::find(stats.issues.begin(), stats.issues.end(), issue) != stats.issues.end();
}
} // namespace
TEST_CASE("TopologyManager incremental rebuild keeps the whole corpus for engines without an "
"incremental path",
"[daemon][topology-manager][incremental]") {
// Louvain and KMeans have no incremental update: their dirty region is the seed set alone
// and requests a wider rebuild. Publishing a batch built from the seeds would replace the
// snapshot with one covering only the dirty documents and strip the topology keys of every
// untouched document.
const std::string algorithm = GENERATE(std::string{"louvain"}, std::string{"kmeans"});
INFO("algorithm=" << algorithm);
RebuildFixture fixture;
const auto ids = addTwoGroupCorpus(fixture);
TopologyManager manager(fixture.dependencies());
auto initial = manager.rebuildArtifacts("initial", false, {}, algorithm);
INFO((initial ? "initial rebuild succeeded" : initial.error().message));
REQUIRE(initial.has_value());
REQUIRE(initial.value().stored);
const auto previous = fixture.latest();
REQUIRE(previous.memberships.size() == ids.size());
const auto previousClusters = membershipClusters(previous);
const auto previousMembers = clusterMembers(previous);
REQUIRE(previousClusters.at("aaa1") != previousClusters.at("bbb1"));
auto result = manager.rebuildArtifacts("post_ingest_drain", false, {"aaa1"}, algorithm);
INFO((result ? "incremental rebuild succeeded" : result.error().message));
REQUIRE(result.has_value());
CHECK_FALSE(result.value().skipped);
CHECK(result.value().stored);
CHECK(result.value().fullRebuild);
CHECK(result.value().fallbackFullRebuilds >= 1);
CHECK(result.value().membershipsBuilt == ids.size());
const auto current = fixture.latest();
CHECK(current.topologyEpoch == previous.topologyEpoch + 1);
const auto currentClusters = membershipClusters(current);
CHECK(currentClusters.size() == ids.size());
for (const auto& [hash, id] : ids) {
INFO("document=" << hash);
auto clusterKey = fixture.repository->getMetadata(id, "topology.cluster_id");
REQUIRE(clusterKey.has_value());
CHECK(clusterKey.value().has_value());
CHECK(currentClusters.contains(hash));
if (clusterKey.value().has_value() && currentClusters.contains(hash)) {
CHECK(clusterKey.value()->asString() == currentClusters.at(hash));
}
}
// Clusters that do not contain the dirty document are unchanged.
const auto& dirtyCluster = previousClusters.at("aaa1");
const auto currentMembers = clusterMembers(current);
for (const auto& [clusterId, members] : previousMembers) {
if (clusterId == dirtyCluster) {
continue;
}
INFO("cluster=" << clusterId);
CHECK(currentMembers.contains(clusterId));
if (currentMembers.contains(clusterId)) {
CHECK(currentMembers.at(clusterId) == members);
}
}
}
TEST_CASE("TopologyManager connected engine keeps its incremental rebuild path",
"[daemon][topology-manager][incremental]") {
RebuildFixture fixture;
const auto ids = addTwoGroupCorpus(fixture);
TopologyManager manager(fixture.dependencies());
REQUIRE(manager.rebuildArtifacts("initial", false, {}, "connected").has_value());
const auto previous = fixture.latest();
REQUIRE(previous.memberships.size() == ids.size());
const auto previousMembers = clusterMembers(previous);
auto result = manager.rebuildArtifacts("post_ingest_drain", false, {"aaa1"}, "connected");
INFO((result ? "incremental rebuild succeeded" : result.error().message));
REQUIRE(result.has_value());
CHECK(result.value().stored);
CHECK_FALSE(result.value().fullRebuild);
CHECK(result.value().fallbackFullRebuilds == 0);
// Only the seed's prior cluster is re-extracted, not the whole corpus.
CHECK(result.value().documentsProcessed == 3);
CHECK(hasIssue(result.value(), "incremental topology rebuild expanded 1 seed docs to 3 docs"));
CHECK_FALSE(hasIssue(result.value(), "dirty-region requested wider rebuild fallback"));
const auto current = fixture.latest();
CHECK(current.memberships.size() == ids.size());
CHECK(clusterMembers(current) == previousMembers);
}
Reference in a new issue View git blame Copy permalink