From d024e1dbb860234eee5dcfb08a4242b09112bf44 Mon Sep 17 00:00:00 2001 From: Andrei Lepikhov Date: Mon, 15 Jun 2026 15:34:32 +0200 Subject: [PATCH 1/3] test: regression guards for reference-node max in multi-leaf mtree diff MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Zaid re-reported ACE-189 on a 3-node cluster: with n3 holding the cluster max, every diff pair involving n3 was short by exactly one row (its largest), while n1/n2 was correct. Reproduced his exact scenario faithfully against the current branch — it now produces the correct 5/10/5 counts (id=2010 included), confirming bb94210/73c3ccd/f9a3961 already cover it. His earlier observation must have predated the fix in his checkout. These two guards lock that coverage in; both exercise a MULTI-LEAF tree (row counts above the cluster min block size of 1000), the case the existing single-leaf bidirectional tests never reached — the boundary where the last closed leaf [B, max] meets the open tail [max, NULL): * TestMerkleTreeReferenceMaxInMultiLeafTail — minimal 2-node form: the reference (n2, more rows) holds the cluster max; asserts the full {2001..2010} diff including the tail row. * TestMerkleTreeThreeNodeReferenceTail — faithful 3-node reproduction. The shared test_cluster.json only registers n1/n2, so this emits its own test_cluster_3node.json via the new writeClusterConfigJSON helper and points the task at it, leaving the rest of the suite (and its Nodes="all" behaviour) untouched. Asserts n1/n3=10, n2/n3=5, n1/n2=5 — Zaid's exact shape. Full mtree suite green with both added. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/integration/merkle_tree_test.go | 234 ++++++++++++++++++++++++++ 1 file changed, 234 insertions(+) diff --git a/tests/integration/merkle_tree_test.go b/tests/integration/merkle_tree_test.go index 901e86a..a9c2a84 100644 --- a/tests/integration/merkle_tree_test.go +++ b/tests/integration/merkle_tree_test.go @@ -13,6 +13,7 @@ package integration import ( "context" + "encoding/json" "fmt" "math/rand" "os" @@ -2055,3 +2056,236 @@ func extractDiffIDs(rows []types.OrderedMap) []int { sort.Ints(ids) return ids } + +// TestMerkleTreeReferenceMaxInMultiLeafTail reproduces the residual ACE-189 +// symptom Zaid reported: the reference node's single largest row — the +// range_start of the open-ended tail leaf in a MULTI-LEAF tree — is dropped +// from the diff even after the open-ended-tail fix (f9a3961). The earlier +// bidirectional tests use tiny row counts (single-leaf trees), so they never +// exercise the boundary where the last closed leaf [B, max] meets the open +// tail [max, NULL). +// +// Zaid saw it on a 3-node cluster (n3 held the cluster max; every pair +// involving n3 was short by one, n1/n2 was correct). The 3-node shape is +// incidental — the bug is "the reference node's own max is dropped" — so this +// reproduces it minimally on the 2-node test cluster, with the reference (n2, +// more rows) holding the cluster max. Row counts are above the cluster's +// min block size (1000) so the default block size still yields a multi-leaf +// tree, which is what exercises the last-closed-leaf/open-tail boundary. +// +// n1 = 1..2000, n2 = 1..2010 → n2 is the reference (3 leaves). +// Expected n1/n2 diff = {2001..2010} (10 rows). The bug drops id=2010 (n2's +// max), yielding 9 — the same off-by-the-last-row Zaid observed. +func TestMerkleTreeReferenceMaxInMultiLeafTail(t *testing.T) { + ctx := context.Background() + env := newSpockEnv() + + tableName := "mtree_ref_max_tail" + qualifiedTable := fmt.Sprintf("%s.%s", testSchema, tableName) + safeTable := pgx.Identifier{testSchema, tableName}.Sanitize() + + seedCount := map[*pgxpool.Pool]int{ + env.N1Pool: 2000, + env.N2Pool: 2010, // n2 holds the cluster max → reference node + } + + // No repset: each node must keep exactly what we seed (see the + // bidirectional test for why repair_mode alone is insufficient). + for _, pool := range env.pools() { + _, err := pool.Exec(ctx, + "CREATE TABLE IF NOT EXISTS "+safeTable+" (id INT PRIMARY KEY, name VARCHAR)") // nosemgrep + require.NoError(t, err, "create %s", qualifiedTable) + } + t.Cleanup(func() { + for _, pool := range env.pools() { + _, _ = pool.Exec(ctx, "DROP TABLE IF EXISTS "+safeTable+" CASCADE") // nosemgrep + } + files, _ := filepath.Glob("*_diffs-*.json") + for _, f := range files { + os.Remove(f) + } + }) + + for _, pool := range env.pools() { + env.withRepairMode(t, ctx, pool, func(conn *pgxpool.Conn) { + _, err := conn.Exec(ctx, "TRUNCATE TABLE "+safeTable) // nosemgrep + require.NoError(t, err, "truncate %s", qualifiedTable) + }) + _, err := pool.Exec(ctx, + "INSERT INTO "+safeTable+" (id, name) SELECT g, 'user_' || g FROM generate_series(1, $1) g", // nosemgrep + seedCount[pool]) + require.NoError(t, err, "seed table") + _, err = pool.Exec(ctx, "ANALYZE "+safeTable) // nosemgrep + require.NoError(t, err) + } + + mtreeTask := env.newMerkleTreeTask(t, qualifiedTable, + []string{env.ServiceN1, env.ServiceN2}) + require.NoError(t, mtreeTask.RunChecks(false)) + require.NoError(t, mtreeTask.MtreeInit()) + t.Cleanup(func() { + if err := mtreeTask.MtreeTeardown(); err != nil { + t.Logf("MtreeTeardown cleanup: %v", err) + } + }) + require.NoError(t, mtreeTask.BuildMtree()) + require.NoError(t, mtreeTask.DiffMtree()) + + nodeDiffs, ok := mtreeTask.DiffResult.NodeDiffs[env.pairKey()] + require.True(t, ok, "no diff for pair %s; result: %+v", + env.pairKey(), mtreeTask.DiffResult.NodeDiffs) + + expected := make([]int, 0, 10) + for i := 2001; i <= 2010; i++ { + expected = append(expected, i) + } + require.Equal(t, expected, extractDiffIDs(nodeDiffs.Rows[env.ServiceN2]), + "reference's max (id=2010) dropped from the open tail of a multi-leaf tree — ACE-189") +} + +// writeClusterConfigJSON emits a .json cluster config in the +// working directory listing the given nodes, so a task pointed at clusterName +// resolves all of them via utils.ReadClusterInfo. The shared test_cluster.json +// only registers n1/n2; this lets a single test opt into a 3-node cluster +// without changing that file (which would alter Nodes="all" suite-wide). +func writeClusterConfigJSON(t *testing.T, clusterName string, nodes []types.NodeGroup) { + t.Helper() + cfg := types.ClusterConfig{ + JSONVersion: "1.0", + ClusterName: clusterName, + LogLevel: "info", + UpdateDate: time.Now().Format(time.RFC3339), + } + cfg.PGEdge.PGVersion = 16 + cfg.PGEdge.AutoStart = "yes" + cfg.PGEdge.Spock = types.SpockConfig{SpockVersion: "4.0.10", AutoDDL: "yes"} + cfg.PGEdge.Databases = []types.Database{{DBName: dbName, DBUser: pgEdgeUser, DBPassword: pgEdgePassword}} + cfg.NodeGroups = nodes + + data, err := json.MarshalIndent(cfg, "", " ") + require.NoError(t, err) + path := clusterName + ".json" + require.NoError(t, os.WriteFile(path, data, 0644)) + t.Cleanup(func() { os.Remove(path) }) +} + +// TestMerkleTreeThreeNodeReferenceTail is the faithful 3-node reproduction of +// Zaid's ACE-189 report: n3 holds the cluster max, and every diff pair that +// involves n3 was short by exactly one row (its largest), while n1/n2 was +// correct. The shared test cluster only registers n1/n2, so this test emits +// its own 3-node cluster config and points the task at it. +// +// n1 = 1..2000, n2 = 1..2005, n3 = 1..2010 → n3 is the reference. +// +// Counts are above the cluster min block size (1000) so the tree is multi-leaf +// (the single-leaf case is already covered and behaves differently at the +// tail). Expected: n1/n3 = {2001..2010} (10), n2/n3 = {2006..2010} (5), +// n1/n2 = {2001..2005} (5) — Zaid's 10/5/5 ratio. The bug drops id=2010 from +// both n3 pairs (9 and 4). +func TestMerkleTreeThreeNodeReferenceTail(t *testing.T) { + ctx := context.Background() + env := newSpockEnv() + if env.N3Pool == nil { + t.Skip("requires a 3-node cluster") + } + + clusterName := "test_cluster_3node" + mkNode := func(name, host, port string) types.NodeGroup { + ng := types.NodeGroup{Name: name, IsActive: "yes", PublicIP: host, Port: port, Path: "/usr/local/bin"} + ng.SSH.OSUser = "pgedge" + return ng + } + writeClusterConfigJSON(t, clusterName, []types.NodeGroup{ + mkNode(env.ServiceN1, pgCluster.Node1Host, pgCluster.Node1Port), + mkNode(env.ServiceN2, pgCluster.Node2Host, pgCluster.Node2Port), + mkNode(env.ServiceN3, pgCluster.Node3Host, pgCluster.Node3Port), + }) + + tableName := "mtree_three_node_tail" + qualifiedTable := fmt.Sprintf("%s.%s", testSchema, tableName) + safeTable := pgx.Identifier{testSchema, tableName}.Sanitize() + + pools := map[string]*pgxpool.Pool{ + env.ServiceN1: env.N1Pool, + env.ServiceN2: env.N2Pool, + env.ServiceN3: env.N3Pool, + } + seedCount := map[string]int{ + env.ServiceN1: 2000, + env.ServiceN2: 2005, + env.ServiceN3: 2010, // cluster max → reference node + } + + // No repset: each node must keep exactly what we seed. + for _, pool := range pools { + _, err := pool.Exec(ctx, + "CREATE TABLE IF NOT EXISTS "+safeTable+" (id INT PRIMARY KEY, name VARCHAR)") // nosemgrep + require.NoError(t, err, "create %s", qualifiedTable) + } + t.Cleanup(func() { + for _, pool := range pools { + _, _ = pool.Exec(ctx, "DROP TABLE IF EXISTS "+safeTable+" CASCADE") // nosemgrep + } + files, _ := filepath.Glob("*_diffs-*.json") + for _, f := range files { + os.Remove(f) + } + }) + + for name, pool := range pools { + env.withRepairMode(t, ctx, pool, func(conn *pgxpool.Conn) { + _, err := conn.Exec(ctx, "TRUNCATE TABLE "+safeTable) // nosemgrep + require.NoError(t, err, "truncate %s on %s", qualifiedTable, name) + }) + _, err := pool.Exec(ctx, + "INSERT INTO "+safeTable+" (id, name) SELECT g, 'user_' || g FROM generate_series(1, $1) g", // nosemgrep + seedCount[name]) + require.NoError(t, err, "seed %s", name) + _, err = pool.Exec(ctx, "ANALYZE "+safeTable) // nosemgrep + require.NoError(t, err) + } + + mtreeTask := env.newMerkleTreeTask(t, qualifiedTable, + []string{env.ServiceN1, env.ServiceN2, env.ServiceN3}) + mtreeTask.ClusterName = clusterName + require.NoError(t, mtreeTask.RunChecks(false)) + require.NoError(t, mtreeTask.MtreeInit()) + t.Cleanup(func() { + if err := mtreeTask.MtreeTeardown(); err != nil { + t.Logf("MtreeTeardown cleanup: %v", err) + } + }) + require.NoError(t, mtreeTask.BuildMtree()) + require.NoError(t, mtreeTask.DiffMtree()) + + // Pair keys follow cluster-node order; look up under either ordering. + findPair := func(a, b string) (types.DiffByNodePair, bool) { + if d, ok := mtreeTask.DiffResult.NodeDiffs[a+"/"+b]; ok { + return d, true + } + d, ok := mtreeTask.DiffResult.NodeDiffs[b+"/"+a] + return d, ok + } + rangeIDs := func(lo, hi int) []int { + ids := make([]int, 0, hi-lo+1) + for i := lo; i <= hi; i++ { + ids = append(ids, i) + } + return ids + } + + d13, ok := findPair(env.ServiceN1, env.ServiceN3) + require.True(t, ok, "no diff for n1/n3; result: %+v", mtreeTask.DiffResult.NodeDiffs) + require.Equal(t, rangeIDs(2001, 2010), extractDiffIDs(d13.Rows[env.ServiceN3]), + "n1/n3: reference's max (2010) dropped from the open tail — ACE-189") + + d23, ok := findPair(env.ServiceN2, env.ServiceN3) + require.True(t, ok, "no diff for n2/n3; result: %+v", mtreeTask.DiffResult.NodeDiffs) + require.Equal(t, rangeIDs(2006, 2010), extractDiffIDs(d23.Rows[env.ServiceN3]), + "n2/n3: reference's max (2010) dropped from the open tail — ACE-189") + + d12, ok := findPair(env.ServiceN1, env.ServiceN2) + require.True(t, ok, "no diff for n1/n2; result: %+v", mtreeTask.DiffResult.NodeDiffs) + require.Equal(t, rangeIDs(2001, 2005), extractDiffIDs(d12.Rows[env.ServiceN2]), + "n1/n2: rows below the reference's max should be reported in full") +} From 187be93fde45f5d898112819649eb81b837ca6e3 Mon Sep 17 00:00:00 2001 From: Andrei Lepikhov Date: Fri, 19 Jun 2026 14:35:26 +0200 Subject: [PATCH 2/3] Enforce table-diff max_diff_rows per node pair MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit table-diff tracked max_diff_rows with a single task-wide counter (totalDiffRows), incremented by every differing row regardless of which node pair it belonged to and reset once per ExecuteTask. On a cluster with more than two nodes this turned the cap into one budget shared across all C(n,2) pairs: once the combined total reached the limit, every worker's stop-check fired and enumeration halted, leaving the report truncated and the reported differences split arbitrarily across pairs by whichever concurrent workers consumed the budget first. In the reported case (n1=1.1M rows, n2=n3=100k) the true divergence is ~1M rows on each of the two n1 pairs (n2/n3 identical). With the global cap of 1,000,000 only half was enumerated in one pass, split 510599 (n1/n3) + 489401 (n1/n2) = exactly the cap — requiring multiple diff+repair cycles to converge. A single pair (2-node cluster) was unaffected because "total" and "the pair" were the same thing; the latent assumption only surfaced for N>2. Make the limit per node pair: replace totalDiffRows with a sync.Map of pairKey -> *atomic.Int64 and route every limit check and increment through shouldStopPair(pairKey) / incrementPairDiffRowsLocked. The pre-diff initial-hash loop now uses only the error circuit-breaker (no rows are counted there, so the limit was always dead at that point). Each pair now enumerates up to max_diff_rows independently; a pair under the cap reports all its differences instead of being starved by another pair's divergence. Behaviour change worth noting: on N-node clusters the worst-case report size is now max_diff_rows x number_of_pairs rather than max_diff_rows total. A genuinely huge divergence therefore produces a larger report rather than silently truncating — the operator should resync such a node rather than repair row-by-row. (Docs follow separately.) Tests: * TestTableDiffMaxDiffRowsPerPair (new, 3-node) — n1 holds 1..20; n2/n3 each hold the same {1..5, 16..20} (10 middle rows missing, avoiding the open-tail boundary confound). With max_diff_rows=14 (> per-pair 10, < cross-pair total 20): before this fix the run stops at 14 total (n1/n3=10, n1/n2=4) and sets DiffRowLimitReached; after, both pairs report all 10 and the limit is not tripped. Uses its own test_cluster_3node.json via the new writeThreeNodeClusterConfig helper, leaving the shared 2-node cluster config untouched. * Existing 2-node MaxDiffRowsLimit tests (simple + composite PK) still pass — the per-pair cap collapses to the old behaviour for one pair. go build / go vet clean; table-diff simple/composite/until suites green. Author: Claude Opus 4.8 (1M context) --- internal/consistency/diff/table_diff.go | 98 +++++++++++++++---------- tests/integration/merkle_tree_test.go | 20 +++++ tests/integration/table_diff_test.go | 94 ++++++++++++++++++++++++ 3 files changed, 175 insertions(+), 37 deletions(-) diff --git a/internal/consistency/diff/table_diff.go b/internal/consistency/diff/table_diff.go index 100c8af..4717a88 100644 --- a/internal/consistency/diff/table_diff.go +++ b/internal/consistency/diff/table_diff.go @@ -114,7 +114,11 @@ type TableDiffTask struct { firstErrorMu sync.Mutex errorRecorded atomic.Bool - totalDiffRows atomic.Int64 + // pairDiffRows enforces max_diff_rows per node pair, keyed by pairKey -> + // *atomic.Int64. A single shared counter would make the cap a budget split + // across all C(n,2) pairs on clusters with more than two nodes, truncating + // the report and requiring multiple repair passes (ACE-191). + pairDiffRows sync.Map diffLimitTriggered atomic.Bool // diffSem limits how many recursive diff goroutines can run at the same time. @@ -163,36 +167,53 @@ func (t *TableDiffTask) hasError() bool { return t.errorRecorded.Load() } -// shouldStop returns true if diff workers should cease processing, either because -// the diff row limit was reached or because a node error has been recorded (circuit -// breaker). This prevents OOM when a node starts failing: without this check, -// goroutines would keep grinding through every remaining sub-range — each waiting -// up to 60 s for a timeout — accumulating error objects until the process is killed. -func (t *TableDiffTask) shouldStop() bool { - return t.shouldStopDueToLimit() || t.hasError() +// pairKeyFor returns the canonical (lexically ordered) "nodeA/nodeB" key used to +// bucket diffs and the per-pair row limit. +func pairKeyFor(node1, node2 string) string { + if strings.Compare(node1, node2) > 0 { + return node2 + "/" + node1 + } + return node1 + "/" + node2 +} + +// pairCounter returns the per-pair diff-row counter, creating it on first use. +func (t *TableDiffTask) pairCounter(pairKey string) *atomic.Int64 { + v, _ := t.pairDiffRows.LoadOrStore(pairKey, new(atomic.Int64)) + return v.(*atomic.Int64) } -func (t *TableDiffTask) shouldStopDueToLimit() bool { +// shouldStopPair returns true if enumeration for this node pair should cease, +// either because the pair reached max_diff_rows or because a node error has been +// recorded (circuit breaker that prevents OOM when a node starts failing). The +// row limit is per pair so one pair's divergence cannot exhaust another pair's +// budget on clusters with more than two nodes (ACE-191). +// +// The cap is best-effort, not exact: this gate is checked before the diffMutex +// is taken, so several concurrent comparisons for the same pair can each pass it +// just under the limit and then append their batches, leaving the pair a few +// rows over max_diff_rows. That is acceptable for a report-size bound and +// matches the prior global-counter behaviour. +func (t *TableDiffTask) shouldStopPair(pairKey string) bool { + if t.hasError() { + return true + } if t.MaxDiffRows <= 0 { return false } - if t.diffLimitTriggered.Load() { - return true - } - return t.totalDiffRows.Load() >= t.MaxDiffRows + return t.pairCounter(pairKey).Load() >= t.MaxDiffRows } // It's imperative for the caller to hold the diffMutex while calling this function -func (t *TableDiffTask) incrementDiffRowsLocked(delta int) bool { +func (t *TableDiffTask) incrementPairDiffRowsLocked(pairKey string, delta int) bool { if delta <= 0 || t.MaxDiffRows <= 0 { return false } - total := t.totalDiffRows.Add(int64(delta)) + total := t.pairCounter(pairKey).Add(int64(delta)) if total >= t.MaxDiffRows { t.DiffResult.Summary.DiffRowLimitReached = true if t.diffLimitTriggered.CompareAndSwap(false, true) { - logger.Warn("table-diff: detected %d differences which meets/exceeds max_diff_rows limit (%d); stopping early", total, t.MaxDiffRows) + logger.Warn("table-diff: %s reached max_diff_rows limit (%d); enumeration for this pair stops (other pairs continue)", pairKey, t.MaxDiffRows) } return true } @@ -1199,7 +1220,10 @@ func (t *TableDiffTask) ExecuteTask() (err error) { t.Task.TaskStatus = taskstore.StatusRunning t.Task.ClusterName = t.ClusterName - t.totalDiffRows.Store(0) + t.pairDiffRows.Range(func(k, _ any) bool { + t.pairDiffRows.Delete(k) + return true + }) t.diffLimitTriggered.Store(false) t.errorRecorded.Store(false) t.firstErrorMu.Lock() @@ -1536,7 +1560,10 @@ func (t *TableDiffTask) ExecuteTask() (err error) { go func() { defer initialHashWg.Done() for task := range hashTaskQueue { - if t.shouldStop() { + // Initial hashing runs before any rows are counted, so only the + // error circuit-breaker is relevant here; the per-pair row limit + // is enforced later in recursiveDiff. + if t.hasError() { bar.Increment() continue } @@ -1628,7 +1655,7 @@ func (t *TableDiffTask) ExecuteTask() (err error) { ) for _, task := range mismatchedTasks { - if t.shouldStop() { + if t.shouldStopPair(pairKeyFor(task.Node1Name, task.Node2Name)) { diffBar.Increment() continue } @@ -1952,12 +1979,14 @@ func (t *TableDiffTask) recursiveDiff( ) { defer wg.Done() - if t.shouldStop() { + node1Name := task.Node1Name + node2Name := task.Node2Name + pairKey := pairKeyFor(node1Name, node2Name) + + if t.shouldStopPair(pairKey) { return } - node1Name := task.Node1Name - node2Name := task.Node2Name currentRange := task.CurrentRange currentEstimatedBlockSize := task.CurrentEstimatedBlockSize @@ -1980,11 +2009,6 @@ func (t *TableDiffTask) recursiveDiff( return } - pairKey := node1Name + "/" + node2Name - if strings.Compare(node1Name, node2Name) > 0 { - pairKey = node2Name + "/" + node1Name - } - var currentDiffRowsForPair int limitReached := false if len(diffInfo.Node1OnlyRows) > 0 || len(diffInfo.Node2OnlyRows) > 0 || len(diffInfo.ModifiedRows) > 0 { @@ -2004,7 +2028,7 @@ func (t *TableDiffTask) recursiveDiff( } for _, row := range diffInfo.Node1OnlyRows { - if t.shouldStop() { + if t.shouldStopPair(pairKey) { limitReached = true break } @@ -2013,7 +2037,7 @@ func (t *TableDiffTask) recursiveDiff( rowAsOrderedMap := utils.MapToOrderedMap(rowWithMeta, t.Cols) t.DiffResult.NodeDiffs[pairKey].Rows[node1Name] = append(t.DiffResult.NodeDiffs[pairKey].Rows[node1Name], rowAsOrderedMap) currentDiffRowsForPair++ - if t.incrementDiffRowsLocked(1) { + if t.incrementPairDiffRowsLocked(pairKey, 1) { limitReached = true break } @@ -2021,7 +2045,7 @@ func (t *TableDiffTask) recursiveDiff( if !limitReached { for _, row := range diffInfo.Node2OnlyRows { - if t.shouldStop() { + if t.shouldStopPair(pairKey) { limitReached = true break } @@ -2030,7 +2054,7 @@ func (t *TableDiffTask) recursiveDiff( rowAsOrderedMap := utils.MapToOrderedMap(rowWithMeta, t.Cols) t.DiffResult.NodeDiffs[pairKey].Rows[node2Name] = append(t.DiffResult.NodeDiffs[pairKey].Rows[node2Name], rowAsOrderedMap) currentDiffRowsForPair++ - if t.incrementDiffRowsLocked(1) { + if t.incrementPairDiffRowsLocked(pairKey, 1) { limitReached = true break } @@ -2039,7 +2063,7 @@ func (t *TableDiffTask) recursiveDiff( if !limitReached { for _, modRow := range diffInfo.ModifiedRows { - if t.shouldStop() { + if t.shouldStopPair(pairKey) { limitReached = true break } @@ -2053,7 +2077,7 @@ func (t *TableDiffTask) recursiveDiff( node2DataAsOrderedMap := utils.MapToOrderedMap(node2DataWithMeta, t.Cols) t.DiffResult.NodeDiffs[pairKey].Rows[node2Name] = append(t.DiffResult.NodeDiffs[pairKey].Rows[node2Name], node2DataAsOrderedMap) currentDiffRowsForPair++ - if t.incrementDiffRowsLocked(1) { + if t.incrementPairDiffRowsLocked(pairKey, 1) { limitReached = true break } @@ -2066,14 +2090,14 @@ func (t *TableDiffTask) recursiveDiff( t.DiffResult.Summary.DiffRowsCount[pairKey] += currentDiffRowsForPair t.diffMutex.Unlock() - if limitReached || t.shouldStop() { + if limitReached || t.shouldStopPair(pairKey) { return } } return } - if t.shouldStop() { + if t.shouldStopPair(pairKey) { return } @@ -2112,7 +2136,7 @@ func (t *TableDiffTask) recursiveDiff( } for _, sr := range subRanges { - if t.shouldStop() { + if t.shouldStopPair(pairKey) { return } @@ -2156,7 +2180,7 @@ func (t *TableDiffTask) recursiveDiff( logger.Debug("%s Mismatch in sub-range %v-%v for %s (%s...) vs %s (%s...). Recursing.", utils.CrossMark, sr.Start, sr.End, node1Name, utils.SafeCut(res1.hash, 8), node2Name, utils.SafeCut(res2.hash, 8)) - if t.shouldStop() { + if t.shouldStopPair(pairKey) { return } diff --git a/tests/integration/merkle_tree_test.go b/tests/integration/merkle_tree_test.go index a9c2a84..7f97505 100644 --- a/tests/integration/merkle_tree_test.go +++ b/tests/integration/merkle_tree_test.go @@ -2169,6 +2169,26 @@ func writeClusterConfigJSON(t *testing.T, clusterName string, nodes []types.Node t.Cleanup(func() { os.Remove(path) }) } +// writeThreeNodeClusterConfig emits a 3-node cluster config (n1/n2/n3) for the +// isolated 3-node tests and returns the cluster name to point a task at. Keeps +// the types.NodeGroup construction in this file so other test files in the +// package can opt into a 3-node cluster without importing the config types. +func writeThreeNodeClusterConfig(t *testing.T, env *testEnv) string { + t.Helper() + clusterName := "test_cluster_3node" + mkNode := func(name, host, port string) types.NodeGroup { + ng := types.NodeGroup{Name: name, IsActive: "yes", PublicIP: host, Port: port, Path: "/usr/local/bin"} + ng.SSH.OSUser = "pgedge" + return ng + } + writeClusterConfigJSON(t, clusterName, []types.NodeGroup{ + mkNode(env.ServiceN1, pgCluster.Node1Host, pgCluster.Node1Port), + mkNode(env.ServiceN2, pgCluster.Node2Host, pgCluster.Node2Port), + mkNode(env.ServiceN3, pgCluster.Node3Host, pgCluster.Node3Port), + }) + return clusterName +} + // TestMerkleTreeThreeNodeReferenceTail is the faithful 3-node reproduction of // Zaid's ACE-189 report: n3 holds the cluster max, and every diff pair that // involves n3 was short by exactly one row (its largest), while n1/n2 was diff --git a/tests/integration/table_diff_test.go b/tests/integration/table_diff_test.go index d0bdcb8..8e8a131 100644 --- a/tests/integration/table_diff_test.go +++ b/tests/integration/table_diff_test.go @@ -21,6 +21,7 @@ import ( "testing" "time" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/pgedge/ace/internal/consistency/diff" "github.com/stretchr/testify/require" @@ -1260,6 +1261,99 @@ func testTableDiff_TableFilterNoRows(t *testing.T, env *testEnv) { require.Contains(t, err.Error(), "table filter produced no rows") } +// TestTableDiffMaxDiffRowsPerPair is the ACE-191 regression guard: max_diff_rows +// must bound EACH node pair independently, not act as one budget shared across +// all pairs. Bug: a single global counter (totalDiffRows) is incremented by +// every pair's rows, so on a 3-node cluster the cap is reached by the SUM across +// pairs and enumeration is truncated and split arbitrarily. +// +// Setup avoids the open-tail/boundary confound (ACE-189): n1 holds ids 1..20; +// n2 and n3 each hold the same {1..5, 16..20} (10 MIDDLE rows missing). So +// n1/n2 and n1/n3 each differ by exactly 10, n2/n3 match, and all nodes share +// min=1/max=20. With max_diff_rows=14 a per-pair cap reports all 10+10 and never +// trips the limit; the global-counter bug stops at 14 total (each pair < 10) and +// sets DiffRowLimitReached. +func TestTableDiffMaxDiffRowsPerPair(t *testing.T) { + ctx := context.Background() + env := newSpockEnv() + if env.N3Pool == nil { + t.Skip("requires a 3-node cluster") + } + + clusterName := writeThreeNodeClusterConfig(t, env) + + tableName := "tdiff_percap" + qualifiedTable := fmt.Sprintf("%s.%s", testSchema, tableName) + safeTable := pgx.Identifier{testSchema, tableName}.Sanitize() + + pools := map[string]*pgxpool.Pool{ + env.ServiceN1: env.N1Pool, + env.ServiceN2: env.N2Pool, + env.ServiceN3: env.N3Pool, + } + + for _, pool := range pools { + _, err := pool.Exec(ctx, + "CREATE TABLE IF NOT EXISTS "+safeTable+" (id INT PRIMARY KEY, name VARCHAR)") // nosemgrep + require.NoError(t, err, "create %s", qualifiedTable) + } + t.Cleanup(func() { + for _, pool := range pools { + _, _ = pool.Exec(ctx, "DROP TABLE IF EXISTS "+safeTable+" CASCADE") // nosemgrep + } + files, _ := filepath.Glob("*_diffs-*.json") + for _, f := range files { + os.Remove(f) + } + }) + + // n1: full 1..20. n2/n3: the same {1..5, 16..20} (missing the 10 middle rows). + seed := func(pool *pgxpool.Pool, where string) { + env.withRepairMode(t, ctx, pool, func(conn *pgxpool.Conn) { + _, err := conn.Exec(ctx, "TRUNCATE TABLE "+safeTable) // nosemgrep + require.NoError(t, err) + }) + q := "INSERT INTO " + safeTable + " (id, name) SELECT g, 'user_' || g FROM generate_series(1, 20) g" + if where != "" { + q += " WHERE " + where + } + _, err := pool.Exec(ctx, q) // nosemgrep + require.NoError(t, err) + _, err = pool.Exec(ctx, "ANALYZE "+safeTable) // nosemgrep + require.NoError(t, err) + } + seed(env.N1Pool, "") + seed(env.N2Pool, "g <= 5 OR g >= 16") + seed(env.N3Pool, "g <= 5 OR g >= 16") + + tdTask := env.newTableDiffTask(t, qualifiedTable, + []string{env.ServiceN1, env.ServiceN2, env.ServiceN3}) + tdTask.ClusterName = clusterName + tdTask.BlockSize = 10 + tdTask.CompareUnitSize = 1 + tdTask.MaxDiffRows = 14 // > per-pair (10), < total across pairs (20) + + require.NoError(t, tdTask.RunChecks(false)) + require.NoError(t, tdTask.ExecuteTask()) + + pk := func(a, b string) string { + if a < b { + return a + "/" + b + } + return b + "/" + a + } + counts := tdTask.DiffResult.Summary.DiffRowsCount + + require.False(t, tdTask.DiffResult.Summary.DiffRowLimitReached, + "ACE-191: per-pair divergence (10) is under max_diff_rows (14); the global-counter bug trips the limit on the cross-pair sum") + require.Equal(t, 10, counts[pk(env.ServiceN1, env.ServiceN2)], + "n1/n2 must report all 10 differing rows") + require.Equal(t, 10, counts[pk(env.ServiceN1, env.ServiceN3)], + "n1/n3 must report all 10 differing rows") + require.Zero(t, counts[pk(env.ServiceN2, env.ServiceN3)], + "n2/n3 are identical and must report no differences") +} + func testTableDiff_MaxDiffRowsLimit(t *testing.T, env *testEnv) { ctx := context.Background() tableName := "customers" From c9c57fecc4bba36e043c9fe1857d3ea4923fcef6 Mon Sep 17 00:00:00 2001 From: Andrei Lepikhov Date: Tue, 23 Jun 2026 11:00:39 +0200 Subject: [PATCH 3/3] Address the review Author: Claude Opus 4.8 (1M context) --- internal/consistency/diff/table_diff.go | 4 ++-- tests/integration/merkle_tree_test.go | 27 ++++++++++++------------- tests/integration/table_diff_test.go | 10 ++++----- 3 files changed, 20 insertions(+), 21 deletions(-) diff --git a/internal/consistency/diff/table_diff.go b/internal/consistency/diff/table_diff.go index 4717a88..3803fd5 100644 --- a/internal/consistency/diff/table_diff.go +++ b/internal/consistency/diff/table_diff.go @@ -117,7 +117,7 @@ type TableDiffTask struct { // pairDiffRows enforces max_diff_rows per node pair, keyed by pairKey -> // *atomic.Int64. A single shared counter would make the cap a budget split // across all C(n,2) pairs on clusters with more than two nodes, truncating - // the report and requiring multiple repair passes (ACE-191). + // the report and requiring multiple repair passes. pairDiffRows sync.Map diffLimitTriggered atomic.Bool @@ -186,7 +186,7 @@ func (t *TableDiffTask) pairCounter(pairKey string) *atomic.Int64 { // either because the pair reached max_diff_rows or because a node error has been // recorded (circuit breaker that prevents OOM when a node starts failing). The // row limit is per pair so one pair's divergence cannot exhaust another pair's -// budget on clusters with more than two nodes (ACE-191). +// budget on clusters with more than two nodes. // // The cap is best-effort, not exact: this gate is checked before the diffMutex // is taken, so several concurrent comparisons for the same pair can each pass it diff --git a/tests/integration/merkle_tree_test.go b/tests/integration/merkle_tree_test.go index 7f97505..5d5ab6d 100644 --- a/tests/integration/merkle_tree_test.go +++ b/tests/integration/merkle_tree_test.go @@ -2057,15 +2057,14 @@ func extractDiffIDs(rows []types.OrderedMap) []int { return ids } -// TestMerkleTreeReferenceMaxInMultiLeafTail reproduces the residual ACE-189 -// symptom Zaid reported: the reference node's single largest row — the +// TestMerkleTreeReferenceMaxInMultiLeafTail reproduces a residual symptom of +// the open-ended-tail diff bug: the reference node's single largest row — the // range_start of the open-ended tail leaf in a MULTI-LEAF tree — is dropped -// from the diff even after the open-ended-tail fix (f9a3961). The earlier -// bidirectional tests use tiny row counts (single-leaf trees), so they never -// exercise the boundary where the last closed leaf [B, max] meets the open -// tail [max, NULL). +// from the diff even after the open-ended-tail fix. The earlier bidirectional +// tests use tiny row counts (single-leaf trees), so they never exercise the +// boundary where the last closed leaf [B, max] meets the open tail [max, NULL). // -// Zaid saw it on a 3-node cluster (n3 held the cluster max; every pair +// It was observed on a 3-node cluster (n3 held the cluster max; every pair // involving n3 was short by one, n1/n2 was correct). The 3-node shape is // incidental — the bug is "the reference node's own max is dropped" — so this // reproduces it minimally on the 2-node test cluster, with the reference (n2, @@ -2075,7 +2074,7 @@ func extractDiffIDs(rows []types.OrderedMap) []int { // // n1 = 1..2000, n2 = 1..2010 → n2 is the reference (3 leaves). // Expected n1/n2 diff = {2001..2010} (10 rows). The bug drops id=2010 (n2's -// max), yielding 9 — the same off-by-the-last-row Zaid observed. +// max), yielding 9 — the same off-by-the-last-row symptom that was reported. func TestMerkleTreeReferenceMaxInMultiLeafTail(t *testing.T) { ctx := context.Background() env := newSpockEnv() @@ -2140,7 +2139,7 @@ func TestMerkleTreeReferenceMaxInMultiLeafTail(t *testing.T) { expected = append(expected, i) } require.Equal(t, expected, extractDiffIDs(nodeDiffs.Rows[env.ServiceN2]), - "reference's max (id=2010) dropped from the open tail of a multi-leaf tree — ACE-189") + "reference's max (id=2010) dropped from the open tail of a multi-leaf tree") } // writeClusterConfigJSON emits a .json cluster config in the @@ -2190,7 +2189,7 @@ func writeThreeNodeClusterConfig(t *testing.T, env *testEnv) string { } // TestMerkleTreeThreeNodeReferenceTail is the faithful 3-node reproduction of -// Zaid's ACE-189 report: n3 holds the cluster max, and every diff pair that +// the reported symptom: n3 holds the cluster max, and every diff pair that // involves n3 was short by exactly one row (its largest), while n1/n2 was // correct. The shared test cluster only registers n1/n2, so this test emits // its own 3-node cluster config and points the task at it. @@ -2200,8 +2199,8 @@ func writeThreeNodeClusterConfig(t *testing.T, env *testEnv) string { // Counts are above the cluster min block size (1000) so the tree is multi-leaf // (the single-leaf case is already covered and behaves differently at the // tail). Expected: n1/n3 = {2001..2010} (10), n2/n3 = {2006..2010} (5), -// n1/n2 = {2001..2005} (5) — Zaid's 10/5/5 ratio. The bug drops id=2010 from -// both n3 pairs (9 and 4). +// n1/n2 = {2001..2005} (5) — the reported 10/5/5 ratio. The bug drops id=2010 +// from both n3 pairs (9 and 4). func TestMerkleTreeThreeNodeReferenceTail(t *testing.T) { ctx := context.Background() env := newSpockEnv() @@ -2297,12 +2296,12 @@ func TestMerkleTreeThreeNodeReferenceTail(t *testing.T) { d13, ok := findPair(env.ServiceN1, env.ServiceN3) require.True(t, ok, "no diff for n1/n3; result: %+v", mtreeTask.DiffResult.NodeDiffs) require.Equal(t, rangeIDs(2001, 2010), extractDiffIDs(d13.Rows[env.ServiceN3]), - "n1/n3: reference's max (2010) dropped from the open tail — ACE-189") + "n1/n3: reference's max (2010) dropped from the open tail") d23, ok := findPair(env.ServiceN2, env.ServiceN3) require.True(t, ok, "no diff for n2/n3; result: %+v", mtreeTask.DiffResult.NodeDiffs) require.Equal(t, rangeIDs(2006, 2010), extractDiffIDs(d23.Rows[env.ServiceN3]), - "n2/n3: reference's max (2010) dropped from the open tail — ACE-189") + "n2/n3: reference's max (2010) dropped from the open tail") d12, ok := findPair(env.ServiceN1, env.ServiceN2) require.True(t, ok, "no diff for n1/n2; result: %+v", mtreeTask.DiffResult.NodeDiffs) diff --git a/tests/integration/table_diff_test.go b/tests/integration/table_diff_test.go index 8e8a131..06db3b3 100644 --- a/tests/integration/table_diff_test.go +++ b/tests/integration/table_diff_test.go @@ -1261,13 +1261,13 @@ func testTableDiff_TableFilterNoRows(t *testing.T, env *testEnv) { require.Contains(t, err.Error(), "table filter produced no rows") } -// TestTableDiffMaxDiffRowsPerPair is the ACE-191 regression guard: max_diff_rows -// must bound EACH node pair independently, not act as one budget shared across -// all pairs. Bug: a single global counter (totalDiffRows) is incremented by +// TestTableDiffMaxDiffRowsPerPair guards that max_diff_rows bounds EACH node +// pair independently, not as one budget shared across all pairs. The bug it +// guards against: a single global counter (totalDiffRows) is incremented by // every pair's rows, so on a 3-node cluster the cap is reached by the SUM across // pairs and enumeration is truncated and split arbitrarily. // -// Setup avoids the open-tail/boundary confound (ACE-189): n1 holds ids 1..20; +// Setup avoids the open-tail/boundary confound: n1 holds ids 1..20; // n2 and n3 each hold the same {1..5, 16..20} (10 MIDDLE rows missing). So // n1/n2 and n1/n3 each differ by exactly 10, n2/n3 match, and all nodes share // min=1/max=20. With max_diff_rows=14 a per-pair cap reports all 10+10 and never @@ -1345,7 +1345,7 @@ func TestTableDiffMaxDiffRowsPerPair(t *testing.T) { counts := tdTask.DiffResult.Summary.DiffRowsCount require.False(t, tdTask.DiffResult.Summary.DiffRowLimitReached, - "ACE-191: per-pair divergence (10) is under max_diff_rows (14); the global-counter bug trips the limit on the cross-pair sum") + "per-pair divergence (10) is under max_diff_rows (14); a global counter would trip the limit on the cross-pair sum") require.Equal(t, 10, counts[pk(env.ServiceN1, env.ServiceN2)], "n1/n2 must report all 10 differing rows") require.Equal(t, 10, counts[pk(env.ServiceN1, env.ServiceN3)],