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") +}