From 11836afcd68a17fe245a383c0d9dc9c8d62e3028 Mon Sep 17 00:00:00 2001 From: Andrei Lepikhov Date: Thu, 2 Jul 2026 13:16:31 +0200 Subject: [PATCH] fix(mtree): resolve phantom mismatches and heal stale leaves in table-diff After an insert replicated cleanly to a subscriber, mtree table-diff logged "Trees differ ... Found 1 mismatched blocks" yet concluded TABLES MATCH, and mtree update never cleared the stale leaf -- only a full mtree build did. Reproduced on a 2-node spock cluster; two drain defects combined to cause it: - The bounded drain's target was pg_current_wal_flush_lsn(). Spock applies transactions with asynchronous commit, so a row could be visible to readers while its commit record was still beyond the flush pointer: the drain stopped "caught up" without it and the subscriber's tree went stale behind data the user could already SELECT. The drain target (drainTargetLSN) is now captured by nudging a trivial committing transaction and reading the WAL insert position in the same statement: every commit visible at call time sits at or below the target, and the nudge's own commit record lands just above it, so the WAL writer flushes past the target within one wal_writer_delay in every synchronous_commit configuration -- the target is reachable by the walsender by construction. - All three stop points (data-path, keepalive prompt-stop, idle-timeout) could fire between a transaction's data messages and its commit, discarding the buffered changes while checkpointing at or past the commit -- skipping the transaction permanently, which is why only a rebuild recovered. A bounded drain now never stops while a transaction buffer is open (inTx). DiffMtree also degrades honestly and self-heals: when every mismatched block between a node pair holds no row differences, they are reported as stale tree hashes rather than left contradicting the TABLES MATCH verdict, and exactly those leaves are rehashed from live data (mark dirty -> rehash -> rebuild parents -> clear flags) so the phantom cannot recur on subsequent diffs. Healing recomputes honest hashes, so it can never mask real divergence; if roots still differ afterwards the diff warns that leaf ranges have diverged and recommends a rebuild. A zero diff count only proves staleness when the comparison actually completed, so pairs that lost comparison work items to errors are skipped with a re-run hint, --until runs never heal (their comparison deliberately excludes rows past the cutoff), and blocks dirtied concurrently by CDC keep their dirty state and counters for the next update. Refreshes are recorded per pair in the summary's stale_blocks_refreshed so reports distinguish a clean diff from a healed one. Integration-tested: a deterministic stale-leaf test asserts the diff reports zero row differences, heals the corrupted leaf back to its build-time hash, surfaces the refresh in the summary, clears its dirty flags, and runs clean on the next pass; the end-to-end replicated-insert scenario no longer reproduces the phantom in either the append or in-range variant. Co-Authored-By: Claude Fable 5 --- db/queries/queries.go | 21 +++ db/queries/templates.go | 11 ++ docs/commands/mtree/mtree-table-diff.md | 7 + docs/design/merkle.md | 2 +- internal/consistency/mtree/merkle.go | 168 +++++++++++++++++++-- internal/infra/cdc/listen.go | 93 ++++++++---- pkg/types/types.go | 4 + tests/integration/mtree_stale_leaf_test.go | 110 ++++++++++++++ 8 files changed, 374 insertions(+), 42 deletions(-) create mode 100644 tests/integration/mtree_stale_leaf_test.go diff --git a/db/queries/queries.go b/db/queries/queries.go index 3b20263..357410c 100644 --- a/db/queries/queries.go +++ b/db/queries/queries.go @@ -1417,6 +1417,27 @@ func UpdateLeafHashesBatch(ctx context.Context, db DBQuerier, mtreeTable string, } +// MarkLeavesDirtyByPositions flags the given leaf blocks for rehash. Used to +// refresh leaves whose stored hash is stale relative to the live table data +// (e.g. mismatches resolved as false positives during a diff). +func MarkLeavesDirtyByPositions(ctx context.Context, db DBQuerier, mtreeTable string, nodePositions []int64) error { + data := map[string]interface{}{ + "MtreeTable": mtreeTable, + } + + sql, err := RenderSQL(SQLTemplates.MarkLeavesDirtyByPositions, data) + if err != nil { + return fmt.Errorf("failed to render MarkLeavesDirtyByPositions SQL: %w", err) + } + + _, err = db.Exec(ctx, sql, nodePositions) + if err != nil { + return fmt.Errorf("query to mark leaves dirty for '%s' failed: %w", mtreeTable, err) + } + + return nil +} + func ClearDirtyFlags(ctx context.Context, db DBQuerier, mtreeTable string, nodePositions []int64) error { data := map[string]interface{}{ "MtreeTable": mtreeTable, diff --git a/db/queries/templates.go b/db/queries/templates.go index c4b54fa..4277ea7 100644 --- a/db/queries/templates.go +++ b/db/queries/templates.go @@ -64,6 +64,7 @@ type Templates struct { GetDirtyAndNewBlocks *template.Template ClearDirtyFlags *template.Template + MarkLeavesDirtyByPositions *template.Template BuildParentNodes *template.Template GetRootNode *template.Template GetNodeChildren *template.Template @@ -1034,6 +1035,16 @@ var SQLTemplates = Templates{ node_level = 0 AND node_position = ANY($1) `)), + MarkLeavesDirtyByPositions: template.Must(template.New("markLeavesDirtyByPositions").Parse(` + UPDATE + {{.MtreeTable}} + SET + dirty = true, + last_modified = current_timestamp + WHERE + node_level = 0 + AND node_position = ANY($1) + `)), BuildParentNodes: template.Must(template.New("buildParentNodes").Parse(` WITH pairs AS ( SELECT diff --git a/docs/commands/mtree/mtree-table-diff.md b/docs/commands/mtree/mtree-table-diff.md index 1cbe7b4..ea236e4 100644 --- a/docs/commands/mtree/mtree-table-diff.md +++ b/docs/commands/mtree/mtree-table-diff.md @@ -52,3 +52,10 @@ holding the node's slot, then re-run, if you need a guaranteed-current drain. warning is logged. This keeps a heavily diverged table from exhausting memory; lower the value in `ace.yaml` on memory-constrained hosts, and re-run after repairing to surface the remaining differences. +- When *all* mismatched blocks between a node pair turn out to hold no row + differences, the tree hashes were stale rather than the data divergent. The + diff reports this, refreshes those blocks from live data (recorded in the + summary's `stale_blocks_refreshed`), and the next diff runs clean. A pair + with any real differences, an incomplete comparison, or an `--until` filter + is not refreshed. If the trees still disagree after a refresh, the nodes' + leaf ranges have diverged; run `ace mtree build` to realign them. diff --git a/docs/design/merkle.md b/docs/design/merkle.md index 09869cc..38d3978 100644 --- a/docs/design/merkle.md +++ b/docs/design/merkle.md @@ -41,7 +41,7 @@ flowchart TB - **Initial build (full scan once)**: ACE partitions the table into PK-ordered blocks (like table-diff) and computes leaf hashes for every block on each node—this is the only full-table scan. Parent hashes are built upward using XOR to form the root. - **Change capture (pgoutput + replication slot)**: The table is added to a publication; pgoutput changes are streamed into a logical replication slot. ACE marks affected blocks as dirty based on PK ranges (no full re-scan). - **Update step**: `UpdateMtree` reads accumulated changes from the slot, splits or merges blocks if needed, recomputes hashes only for dirty/new leaves, rebuilds parent XORs, and clears dirty flags. Block size is recovered from metadata to stay consistent with the build. -- **Diff step**: `DiffMtree` first runs an update (to fold recent changes into the tree), then traverses trees across nodes. Matching hashes at a node mean “skip subtree”; mismatches recurse until leaves are reached, where row-level diffs are fetched if required. +- **Diff step**: `DiffMtree` first runs an update (to fold recent changes into the tree), then traverses trees across nodes. Matching hashes at a node mean “skip subtree”; mismatches recurse until leaves are reached, where row-level diffs are fetched if required. When every mismatched block for a node pair compares clean at row level, the diff refreshes those stale leaf hashes from live data (a write to the `ace_mtree_*` tables) so the phantom mismatch does not recur. ```mermaid flowchart TD diff --git a/internal/consistency/mtree/merkle.go b/internal/consistency/mtree/merkle.go index ee1fc9e..bd2c667 100644 --- a/internal/consistency/mtree/merkle.go +++ b/internal/consistency/mtree/merkle.go @@ -113,6 +113,11 @@ type MerkleTreeTask struct { // once. Both are guarded by diffMutex. diffRowCounts map[string]int64 diffLimitWarned bool + // pairCompareErrs records node pairs whose range comparison lost work items + // to errors. A zero diff count for such a pair means "not fully compared", + // not "no differences", so stale-block healing must not act on it. Guarded + // by diffMutex. + pairCompareErrs map[string]bool StartTime time.Time NodeOriginNames map[string]map[string]string @@ -280,6 +285,8 @@ func (m *MerkleTreeTask) compareRangesWorker(wg *sync.WaitGroup, jobs <-chan Com }() for work := range jobs { + nodePairKey := fmt.Sprintf("%s/%s", work.Node1["Name"], work.Node2["Name"]) + node1Name := work.Node1["Name"].(string) pool1, ok := pools[node1Name] if !ok { @@ -287,6 +294,7 @@ func (m *MerkleTreeTask) compareRangesWorker(wg *sync.WaitGroup, jobs <-chan Com pool1, err = auth.GetClusterNodeConnection(m.Ctx, work.Node1, m.connOpts()) if err != nil { logger.Error("worker failed to connect to %s: %v", node1Name, err) + m.recordPairCompareErr(nodePairKey) bar.Increment() continue } @@ -300,6 +308,7 @@ func (m *MerkleTreeTask) compareRangesWorker(wg *sync.WaitGroup, jobs <-chan Com pool2, err = auth.GetClusterNodeConnection(m.Ctx, work.Node2, m.connOpts()) if err != nil { logger.Error("worker failed to connect to %s: %v", node2Name, err) + m.recordPairCompareErr(nodePairKey) bar.Increment() continue } @@ -308,13 +317,25 @@ func (m *MerkleTreeTask) compareRangesWorker(wg *sync.WaitGroup, jobs <-chan Com err := m.processWorkItem(work, pool1, pool2) if err != nil { - nodePairKey := fmt.Sprintf("%s/%s", work.Node1["Name"], work.Node2["Name"]) logger.Error("failed to process work item for %s: %v", nodePairKey, err) + m.recordPairCompareErr(nodePairKey) } bar.Increment() } } +// recordPairCompareErr flags a node pair as incompletely compared: at least one +// of its range-comparison work items was lost to an error, so a zero diff +// count for the pair cannot be trusted. +func (m *MerkleTreeTask) recordPairCompareErr(pairKey string) { + m.diffMutex.Lock() + defer m.diffMutex.Unlock() + if m.pairCompareErrs == nil { + m.pairCompareErrs = make(map[string]bool) + } + m.pairCompareErrs[pairKey] = true +} + func (m *MerkleTreeTask) processWorkItem(work CompareRangesWorkItem, pool1, pool2 *pgxpool.Pool) error { logger.Debug("Processing work item with %d ranges for %s and %s", len(work.Ranges), work.Node1["Name"], work.Node2["Name"]) var whereClause string @@ -2150,11 +2171,18 @@ func (m *MerkleTreeTask) DiffMtree() (err error) { mtreeTableIdentifier := pgx.Identifier{m.aceSchema(), fmt.Sprintf("ace_mtree_%s_%s", m.Schema, m.Table)} mtreeTableName := mtreeTableIdentifier.Sanitize() - allNodePairBatches := make(map[string]struct { - node1 map[string]any - node2 map[string]any - batches [][2][]any - }) + // pairDiffWork collects, per node pair, the pkey ranges to row-compare plus + // the mismatched leaf positions and open pools, so blocks that turn out to + // hold no row differences (stale hashes) can be refreshed afterwards. + type pairDiffWork struct { + node1 map[string]any + node2 map[string]any + batches [][2][]any + positions []int64 + pool1 *pgxpool.Pool + pool2 *pgxpool.Pool + } + allNodePairBatches := make(map[string]*pairDiffWork) m.StartTime = time.Now() // Resolve the differing-row cap from the mtree config when the caller left @@ -2168,6 +2196,7 @@ func (m *MerkleTreeTask) DiffMtree() (err error) { } m.diffRowCounts = make(map[string]int64) m.diffLimitWarned = false + m.pairCompareErrs = make(map[string]bool) m.DiffResult = types.DiffOutput{ NodeDiffs: make(map[string]types.DiffByNodePair), Summary: types.DiffSummary{ @@ -2248,14 +2277,11 @@ func (m *MerkleTreeTask) DiffMtree() (err error) { nodePairKey := fmt.Sprintf("%s/%s", node1["Name"], node2["Name"]) entry, exists := allNodePairBatches[nodePairKey] if !exists { - entry = struct { - node1 map[string]any - node2 map[string]any - batches [][2][]any - }{node1: node1, node2: node2, batches: [][2][]any{}} + entry = &pairDiffWork{node1: node1, node2: node2, pool1: pool1, pool2: pool2} + allNodePairBatches[nodePairKey] = entry } entry.batches = append(entry.batches, batches...) - allNodePairBatches[nodePairKey] = entry + entry.positions = append(entry.positions, positions...) } @@ -2274,6 +2300,59 @@ func (m *MerkleTreeTask) DiffMtree() (err error) { if len(workItems) > 0 { m.CompareRanges(workItems) + + // Mismatched blocks whose row comparison found no differences had stale + // tree hashes, not data divergence (typically rows applied by + // replication that a past drain missed). Say so -- a bare "Found N + // mismatched blocks" followed by "TABLES MATCH" reads as a + // contradiction -- and rehash those leaves from live data so the + // phantom mismatch does not recur on every subsequent diff. A zero diff + // count only proves staleness when every work item for the pair was + // actually compared and no row filter narrowed the comparison, so pairs + // with lost work items -- and any run under --until, whose comparison + // deliberately excludes rows past the cutoff -- are left alone. + for pairKey, entry := range allNodePairBatches { + if len(entry.positions) == 0 || m.DiffResult.Summary.DiffRowsCount[pairKey] > 0 { + continue + } + if m.pairCompareErrs[pairKey] { + logger.Warn("comparison between %s and %s was incomplete (worker errors); "+ + "skipping stale-block refresh -- re-run the diff", + entry.node1["Name"], entry.node2["Name"]) + continue + } + if m.untilTime != nil { + continue + } + logger.Info("The mismatched block(s) between %s and %s contained no row differences; "+ + "the tree hashes were stale. Refreshing them from live data", + entry.node1["Name"], entry.node2["Name"]) + healed := true + for _, side := range []struct { + pool *pgxpool.Pool + name any + }{{entry.pool1, entry.node1["Name"]}, {entry.pool2, entry.node2["Name"]}} { + if err := m.refreshStaleLeaves(side.pool, entry.positions); err != nil { + healed = false + logger.Warn("failed to refresh stale blocks on %s (the diff result is unaffected; "+ + "the mismatch may reappear on the next diff): %v", side.name, err) + } + } + if healed { + if m.DiffResult.Summary.StaleBlocksRefreshed == nil { + m.DiffResult.Summary.StaleBlocksRefreshed = make(map[string]int) + } + m.DiffResult.Summary.StaleBlocksRefreshed[pairKey] = len(entry.positions) + r1, err1 := queries.GetRootNode(m.Ctx, entry.pool1, mtreeTableName) + r2, err2 := queries.GetRootNode(m.Ctx, entry.pool2, mtreeTableName) + if err1 == nil && err2 == nil && r1 != nil && r2 != nil && !bytes.Equal(r1.NodeHash, r2.NodeHash) { + logger.Warn("trees between %s and %s still differ after refreshing stale blocks; "+ + "their leaf ranges have likely diverged -- run 'ace mtree build' to realign them", + entry.node1["Name"], entry.node2["Name"]) + } + } + } + endTime := time.Now() m.DiffResult.Summary.EndTime = endTime.Format(time.RFC3339) m.DiffResult.Summary.TimeTaken = endTime.Sub(m.StartTime).String() @@ -2299,6 +2378,71 @@ func (m *MerkleTreeTask) DiffMtree() (err error) { return nil } +// refreshStaleLeaves rehashes exactly the given leaf blocks from live table +// data and rebuilds the parent chain, in one transaction on one node. Used +// when a diff resolves a leaf-hash mismatch as a false positive: the data +// matches, so only the stored hash is behind, and without a refresh the +// phantom mismatch would recur on every diff until a full rebuild. Blocks +// dirtied concurrently by CDC keep their dirty state and counters for the +// next update. +func (m *MerkleTreeTask) refreshStaleLeaves(pool *pgxpool.Pool, positions []int64) error { //nolint:gocyclo // linear mark->rehash->rebuild->clear pipeline; the count is per-step error returns, not branching logic + mtreeTableIdentifier := pgx.Identifier{m.aceSchema(), fmt.Sprintf("ace_mtree_%s_%s", m.Schema, m.Table)} + mtreeTableName := mtreeTableIdentifier.Sanitize() + + tx, err := pool.Begin(m.Ctx) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + defer tx.Rollback(m.Ctx) + + if err := queries.MarkLeavesDirtyByPositions(m.Ctx, tx, mtreeTableName, positions); err != nil { + return err + } + // Reuse the update pipeline: fetch the (now dirty) block ranges, rehash + // them from the live table, rebuild parents, clear the flags -- but only + // for the requested positions. GetDirtyAndNewBlocks also returns blocks a + // concurrent CDC consumer dirtied; rehash-and-clear on those would zero + // their counters without split evaluation and could clear a dirty flag for + // a change committed after the hash read. Leaving them dirty hands them to + // the next update, which owns that logic. + blocks, err := queries.GetDirtyAndNewBlocks(m.Ctx, tx, mtreeTableName, m.SimplePrimaryKey, m.Key) + if err != nil { + return err + } + requested := make(map[int64]struct{}, len(positions)) + for _, p := range positions { + requested[p] = struct{}{} + } + kept := blocks[:0] + for _, b := range blocks { + if _, ok := requested[b.NodePosition]; ok { + kept = append(kept, b) + } + } + blocks = kept + if len(blocks) == 0 { + return tx.Commit(m.Ctx) + } + numWorkers := int(float64(runtime.NumCPU()) * m.MaxCpuRatio) + if numWorkers < 1 { + numWorkers = 1 + } + if err := m.computeLeafHashes(pool, tx, blocks, numWorkers, "Refreshing stale blocks:"); err != nil { + return err + } + if err := m.buildParentNodes(tx); err != nil { + return err + } + affected := make([]int64, 0, len(blocks)) + for _, b := range blocks { + affected = append(affected, b.NodePosition) + } + if err := queries.ClearDirtyFlags(m.Ctx, tx, mtreeTableName, affected); err != nil { + return err + } + return tx.Commit(m.Ctx) +} + func (m *MerkleTreeTask) findMismatchedLeaves(pool1, pool2 *pgxpool.Pool, parentLevel int, parentPosition int64) (map[int64]bool, error) { mismatched := make(map[int64]bool) mtreeTableIdentifier := pgx.Identifier{m.aceSchema(), fmt.Sprintf("ace_mtree_%s_%s", m.Schema, m.Table)} diff --git a/internal/infra/cdc/listen.go b/internal/infra/cdc/listen.go index f2f881f..04bf6a9 100644 --- a/internal/infra/cdc/listen.go +++ b/internal/infra/cdc/listen.go @@ -73,9 +73,10 @@ func ListenForChanges(ctx context.Context, nodeInfo map[string]any) { } // UpdateFromCDC runs one bounded CDC drain for a node: it catches the node's -// Merkle tree up to the WAL flush position captured at call time, then returns. -// It returns an error rather than reporting success if it cannot reach that -// snapshot, so a stale tree is never silently treated as current (ACE-190). +// Merkle tree up to a WAL position that covers every commit visible at call +// time (see getDrainTargetLSN), then returns. It returns an error rather than +// reporting success if it cannot reach that snapshot, so a stale tree is never +// silently treated as current (ACE-190). func UpdateFromCDC(ctx context.Context, nodeInfo map[string]any) error { return processReplicationStream(ctx, nodeInfo, false) } @@ -95,7 +96,7 @@ var FlushBatchSize = 10_000 // the decoded changes to its Merkle tree. With continuous=true it streams // indefinitely, dispatching apply work to worker goroutines. With // continuous=false it performs a bounded catch-up: it must reach the call-time -// WAL flush snapshot (targetFlushLSN) before reporting success, applies changes +// snapshot (drainTargetLSN) before reporting success, applies changes // synchronously, and flushes them in memory-bounded sub-batches so a single // large transaction drains without buffering the whole transaction. func processReplicationStream(ctx context.Context, nodeInfo map[string]any, continuous bool) error { //nolint:gocyclo // CDC drain protocol state machine; interleaved keepalive/XLogData branching is inherent and splitting it would obscure the LSN/stop bookkeeping @@ -177,15 +178,15 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont } } - targetFlushLSN := pglogrepl.LSN(0) + drainTargetLSN := pglogrepl.LSN(0) if !continuous { - targetFlushLSN, err = getCurrentWalFlushLSN(ctx, pool) + drainTargetLSN, err = getDrainTargetLSN(ctx, pool) if err != nil { - logger.Error("failed to get current WAL flush LSN: %v", err) - return fmt.Errorf("failed to get current WAL flush LSN: %w", err) + logger.Error("failed to get drain target LSN: %v", err) + return fmt.Errorf("failed to get drain target LSN: %w", err) } - if targetFlushLSN == startLSN { + if drainTargetLSN == startLSN { logger.Info("CDC already up-to-date at LSN %s; exiting", startLSN) return nil } @@ -196,7 +197,7 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont var lastLSNVal atomic.Uint64 lastLSNVal.Store(uint64(lastLSN)) // Bounded-drain (non-continuous) safety state. ACE-190: a bounded drain must - // actually reach the call-time snapshot (targetFlushLSN) before it may report + // actually reach the call-time snapshot (drainTargetLSN) before it may report // success; otherwise UpdateMtree/DiffMtree would compare a stale tree and // silently miss real divergence. reachedTarget records whether we got there. reachedTarget := false @@ -261,6 +262,12 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont txChanges := make(map[uint32][]cdcMsg) var currentXID uint32 + // inTx is true between a BeginMessage and its CommitMessage. A bounded + // drain must never stop inside that window: the transaction's changes sit + // unapplied in txChanges, and stopping would discard them while the final + // standby update may confirm past the commit -- permanently skipping the + // transaction and leaving the Merkle tree stale. + inTx := false var wg sync.WaitGroup errCh := make(chan error, 1) stopStreaming := false @@ -422,20 +429,20 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont logger.Info("Replication stopping due to context cancellation") return nil } - if lastLSN < targetFlushLSN { + if lastLSN < drainTargetLSN { return fmt.Errorf("CDC drain reached LSN %s of target %s on slot %s but did not "+ "catch up within cdc_processing_timeout. Progress is durable: re-run "+ "'ace mtree update' to continue draining, or raise cdc_processing_timeout to "+ "drain in one pass (required when a single transaction is larger than the "+ - "timeout window): %w", lastLSN, targetFlushLSN, slotName, ctx.Err()) + "timeout window): %w", lastLSN, drainTargetLSN, slotName, ctx.Err()) } return ctx.Err() } if pgconn.Timeout(err) { if !continuous { - if lastLSN >= targetFlushLSN { + if lastLSN >= drainTargetLSN && !inTx { // Received every change up to the call-time snapshot. - logger.Info("Replication stream drained to target LSN %s.", targetFlushLSN) + logger.Info("Replication stream drained to target LSN %s.", drainTargetLSN) reachedTarget = true stopStreaming = true } @@ -478,7 +485,7 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont // ServerWALEnd on a logical keepalive is the walsender's decode // position (how far it has read this slot's WAL). When it reaches - // targetFlushLSN the decoder has processed every commit at or + // drainTargetLSN the decoder has processed every commit at or // before the target, and those commits' changes were streamed -- // and received, in LSN order -- ahead of this keepalive. So the // gap between lastLSN and the target holds no further tracked-table @@ -488,8 +495,12 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont // fires only after the decoder reaches the target; while it is // still decoding a large in-progress transaction, ServerWALEnd // stays below the target and we keep waiting -- see ACE-190.) - if !continuous && targetFlushLSN != 0 && pkm.ServerWALEnd >= targetFlushLSN { - logger.Info("Server caught up to target WAL flush LSN %s via keepalive; stopping replication stream", targetFlushLSN) + // The !inTx guard: a keepalive can interleave between a + // transaction's data messages and its commit message. Stopping + // there would discard the buffered changes yet checkpoint at the + // target, permanently skipping the transaction. + if !continuous && drainTargetLSN != 0 && pkm.ServerWALEnd >= drainTargetLSN && !inTx { + logger.Info("Server caught up to drain target LSN %s via keepalive; stopping replication stream", drainTargetLSN) // Advance the checkpoint to the target. We stopped on a keepalive // (not a data message), so lastLSN still sits at the start when no // tracked change reached the target; per the reasoning above the @@ -498,8 +509,8 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont // write below would persist the stale start LSN, the slot would // never advance, and every subsequent drain would re-scan the same // span (and the server would retain that WAL indefinitely). - if lastLSN < targetFlushLSN { - lastLSN = targetFlushLSN + if lastLSN < drainTargetLSN { + lastLSN = drainTargetLSN lastLSNVal.Store(uint64(lastLSN)) } reachedTarget = true @@ -523,6 +534,7 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont case *pglogrepl.BeginMessage: currentXID = logicalMsg.Xid txChanges[currentXID] = []cdcMsg{} + inTx = true case *pglogrepl.CommitMessage: if changes, ok := txChanges[currentXID]; ok { if len(changes) > 0 { @@ -531,6 +543,7 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont } delete(txChanges, currentXID) } + inTx = false case *pglogrepl.RelationMessage: relations[logicalMsg.RelationID] = logicalMsg case *pglogrepl.InsertMessage: @@ -593,8 +606,12 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont lastFlushTime = time.Now() } } - if !continuous && targetFlushLSN != 0 && lastLSN >= targetFlushLSN { - logger.Info("Reached target WAL flush LSN %s; stopping replication stream", targetFlushLSN) + // The !inTx guard: lastLSN advances on every data message, so it + // can cross the target between a transaction's changes and its + // commit. Keep reading until the commit lands -- it was already + // decoded and sent, so it is at most a few messages away. + if !continuous && drainTargetLSN != 0 && lastLSN >= drainTargetLSN && !inTx { + logger.Info("Reached drain target LSN %s; stopping replication stream", drainTargetLSN) reachedTarget = true stopStreaming = true } @@ -639,7 +656,7 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont // let UpdateMtree/DiffMtree compare a tree that never saw the backlog. if !continuous && processingErr == nil && !reachedTarget { processingErr = fmt.Errorf("CDC drain ended at LSN %s without reaching target %s on slot %s; "+ - "refusing to report a possibly-stale Merkle tree as current", lastLSN, targetFlushLSN, slotName) + "refusing to report a possibly-stale Merkle tree as current", lastLSN, drainTargetLSN, slotName) } if processingErr != nil { @@ -700,20 +717,38 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont return nil } -// getCurrentWalFlushLSN returns the node's current WAL flush LSN -- the snapshot -// position a bounded drain must reach before it is considered caught up. -func getCurrentWalFlushLSN(ctx context.Context, pool *pgxpool.Pool) (pglogrepl.LSN, error) { +// getDrainTargetLSN returns the snapshot position a bounded drain must reach +// before it is considered caught up. Two requirements shape it: +// +// - It must cover every commit visible to readers at call time. The flush +// LSN alone does not: transactions applied by Spock commit asynchronously +// (synchronous_commit=off), so a row can be visible while its commit +// record is still ahead of the flush pointer, and a flush-LSN target +// would let the drain stop before covering it. +// - It must be reachable by the walsender, whose send pointer is capped by +// *flushed* WAL. The raw insert LSN is not always reachable: the WAL +// writer flushes partial pages only up to the last async commit, so an +// insert-LSN target behind a non-commit tail record could stall the +// drain until the next checkpoint. +// +// Both are satisfied by nudging a trivial committing transaction and reading +// the insert LSN in the same statement: every previously visible commit sits +// at or below the captured position, and the nudge's own commit record lands +// just above it, guaranteeing the WAL writer flushes past the target within +// one wal_writer_delay in every synchronous_commit configuration. +func getDrainTargetLSN(ctx context.Context, pool *pgxpool.Pool) (pglogrepl.LSN, error) { var lsnStr string - err := pool.QueryRow(ctx, "SELECT pg_current_wal_flush_lsn()").Scan(&lsnStr) + var xid int64 + err := pool.QueryRow(ctx, "SELECT pg_current_wal_insert_lsn(), txid_current()").Scan(&lsnStr, &xid) if err != nil { - return 0, fmt.Errorf("failed to fetch pg_current_wal_flush_lsn: %w", err) + return 0, fmt.Errorf("failed to fetch drain target LSN: %w", err) } if lsnStr == "" { - return 0, fmt.Errorf("pg_current_wal_flush_lsn returned empty string") + return 0, fmt.Errorf("pg_current_wal_insert_lsn returned empty string") } lsn, err := pglogrepl.ParseLSN(lsnStr) if err != nil { - return 0, fmt.Errorf("failed to parse pg_current_wal_flush_lsn %s: %w", lsnStr, err) + return 0, fmt.Errorf("failed to parse drain target LSN %s: %w", lsnStr, err) } return lsn, nil } diff --git a/pkg/types/types.go b/pkg/types/types.go index c68826c..bdcf9d9 100644 --- a/pkg/types/types.go +++ b/pkg/types/types.go @@ -130,6 +130,10 @@ type DiffSummary struct { // the node's slot). The comparison for those nodes is best-effort and may // omit their most recent changes, so divergence can be under-reported. CDCSkippedNodes []string `json:"cdc_skipped_nodes,omitempty"` + // StaleBlocksRefreshed counts, per node pair, mismatched Merkle-tree blocks + // that row comparison resolved as false positives (stale hashes, no data + // differences) and that the diff refreshed from live data in place. + StaleBlocksRefreshed map[string]int `json:"stale_blocks_refreshed,omitempty"` } type KVPair struct { diff --git a/tests/integration/mtree_stale_leaf_test.go b/tests/integration/mtree_stale_leaf_test.go new file mode 100644 index 0000000..625f3e7 --- /dev/null +++ b/tests/integration/mtree_stale_leaf_test.go @@ -0,0 +1,110 @@ +// /////////////////////////////////////////////////////////////////////////// +// +// # ACE - Active Consistency Engine +// +// Copyright (C) 2023 - 2026, pgEdge (https://www.pgedge.com/) +// +// This software is released under the PostgreSQL License: +// https://opensource.org/license/postgresql +// +// /////////////////////////////////////////////////////////////////////////// + +package integration + +import ( + "context" + "fmt" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/pgedge/ace/pkg/config" + "github.com/stretchr/testify/require" +) + +// A stale leaf hash that row comparison resolves as a false positive is +// refreshed during the diff, so the "Found N mismatched blocks" phantom does +// not recur on the next diff. +func TestMerkleTreeStaleLeafHeals(t *testing.T) { + ctx := context.Background() + env := newSpockEnv() + + tableName := "mtree_stale_leaf" + qualified := fmt.Sprintf("%s.%s", testSchema, tableName) + safe := pgx.Identifier{testSchema, tableName}.Sanitize() + pools := []*pgxpool.Pool{env.N1Pool, env.N2Pool} + + for _, pool := range pools { + _, err := pool.Exec(ctx, "CREATE TABLE IF NOT EXISTS "+safe+" (id INT PRIMARY KEY, name VARCHAR)") // nosemgrep + require.NoError(t, err) + _, err = pool.Exec(ctx, "INSERT INTO "+safe+" (id, name) SELECT g, 'user_'||g FROM generate_series(1,1000) g") // nosemgrep + require.NoError(t, err) + // The build sizes its block ranges from planner estimates. + _, err = pool.Exec(ctx, "ANALYZE "+safe) // nosemgrep + require.NoError(t, err) + } + t.Cleanup(func() { + for _, pool := range pools { + _, _ = pool.Exec(ctx, "DROP TABLE IF EXISTS "+safe+" CASCADE") // nosemgrep + } + }) + + task := env.newMerkleTreeTask(t, qualified, []string{env.ServiceN1, env.ServiceN2}) + task.BlockSize = 100 + task.OverrideBlockSize = true + require.NoError(t, task.RunChecks(false)) + require.NoError(t, task.MtreeInit()) + t.Cleanup(func() { _ = task.MtreeTeardown() }) + require.NoError(t, task.BuildMtree()) + + mt := pgx.Identifier{config.Cfg.MTree.Schema, "ace_mtree_" + testSchema + "_" + tableName}.Sanitize() + rootHash := func(pool *pgxpool.Pool) (h string) { + require.NoError(t, pool.QueryRow(ctx, "SELECT encode(node_hash,'hex') FROM "+mt+ // nosemgrep + " WHERE node_level=(SELECT max(node_level) FROM "+mt+") AND node_position=0").Scan(&h)) + return + } + require.Equal(t, rootHash(env.N1Pool), rootHash(env.N2Pool), "sanity: trees match after build") + + var leafCount int + require.NoError(t, env.N2Pool.QueryRow(ctx, "SELECT count(*) FROM "+mt+" WHERE node_level = 0").Scan(&leafCount)) // nosemgrep + require.GreaterOrEqual(t, leafCount, 5, "sanity: need enough leaves to corrupt position 4") + + leafHash := func(pool *pgxpool.Pool, pos int) (h string) { + require.NoError(t, pool.QueryRow(ctx, "SELECT encode(node_hash,'hex') FROM "+mt+ // nosemgrep + " WHERE node_level = 0 AND node_position = $1", pos).Scan(&h)) + return + } + builtLeafHash := leafHash(env.N1Pool, 4) + + // Simulate a stale tree on n2: one leaf (and, so the divergence is visible + // from the root down, its ancestor levels) carries a hash that no longer + // reflects the identical live data -- the state a missed CDC drain leaves + // behind. + _, err := env.N2Pool.Exec(ctx, "UPDATE "+mt+" SET node_hash = decode('deadbeef','hex') WHERE node_level = 0 AND node_position = 4") // nosemgrep + require.NoError(t, err) + _, err = env.N2Pool.Exec(ctx, "UPDATE "+mt+" SET node_hash = decode('deadbeef','hex') WHERE node_level > 0") // nosemgrep + require.NoError(t, err) + require.NotEqual(t, rootHash(env.N1Pool), rootHash(env.N2Pool), "sanity: staleness visible at the root") + + // Diff #1 resolves the mismatch as a false positive: no row differences... + require.NoError(t, task.DiffMtree()) + total := 0 + for _, c := range task.DiffResult.Summary.DiffRowsCount { + total += c + } + require.Zero(t, total, "tables are identical; the mismatch is stale hashes only") + + // ...heals the stale leaf back to the build-time hash (not merely to some + // value both refresh passes agree on), reports the refresh in the summary, + // and leaves the trees agreeing without a rebuild or dirty leftovers. + require.Equal(t, builtLeafHash, leafHash(env.N2Pool, 4), "healed leaf must match the hash built from identical data") + require.Equal(t, rootHash(env.N1Pool), rootHash(env.N2Pool), "diff must refresh the stale blocks") + require.NotEmpty(t, task.DiffResult.Summary.StaleBlocksRefreshed, "refresh must be visible in the diff summary") + var dirtyLeft int + require.NoError(t, env.N2Pool.QueryRow(ctx, "SELECT count(*) FROM "+mt+" WHERE node_level = 0 AND dirty").Scan(&dirtyLeft)) // nosemgrep + require.Zero(t, dirtyLeft, "refresh must clear the dirty flags it set") + + // Diff #2 finds nothing to traverse: the phantom is gone. + require.NoError(t, task.DiffMtree()) + require.Empty(t, task.DiffResult.NodeDiffs, "no phantom mismatch on the next diff") +}