diff --git a/ace.yaml b/ace.yaml index a4572ac..44d8fb9 100644 --- a/ace.yaml +++ b/ace.yaml @@ -37,6 +37,13 @@ mtree: cdc_processing_timeout: 300 cdc_metadata_flush_seconds: 10 cdc_flush_batch_size: 10000 + # Escalate a table to a whole-tree rehash when a bounded drain sees more + # than max(adaptive_drain_min_changes, adaptive_drain_fraction * rows) + # UPDATE changes for it (inserts/deletes are always tracked individually, + # preserving block split/merge maintenance). Set adaptive_drain_fraction + # to -1 to disable. + adaptive_drain_fraction: 0.01 + adaptive_drain_min_changes: 1000 schema: "pgedge_ace" diff: diff --git a/docs/commands/mtree/index.md b/docs/commands/mtree/index.md index b2fad00..67f342b 100644 --- a/docs/commands/mtree/index.md +++ b/docs/commands/mtree/index.md @@ -60,6 +60,17 @@ divergence can be under-reported. For a diff that is guaranteed current to the present moment, ensure no `mtree listen` or other mtree operation is holding the node's slot, then re-run so the diff can perform its own bounded CDC drain. +Because the replication slot is shared, the bounded CDC drain that runs before a +diff or update processes the pending changes for **all** Merkle-tracked tables on +that node, not just the table named on the command line. For the other tables this +only records which of their tree blocks are now stale (marks them dirty); it never +recomputes their hashes. Each table's dirty blocks are rehashed when that table is +next the target of an `mtree table-diff`, `mtree update`, or `mtree listen` apply, +so a table-diff on one table may slightly speed up — never slow down or alter — +a later diff of another table. + +When a single bounded drain encounters a very large number of row updates for one table (by default more than ~1% of its rows), ACE marks that table's whole Merkle tree dirty instead of tracking each update individually, and the next tree update for that table recomputes its leaf hashes in bulk — much faster for heavily-updated tables, with no effect on diff correctness (inserts and deletes are always tracked individually, so block structure maintenance is unaffected). This applies to any tracked table whose updates appear in the drained stream, including tables other than the one being diffed. Keeping `mtree listen` running still gives the best interactive `table-diff` latency, since the tree is then maintained continuously. + ### Building Merkle Trees in Parallel (for Very Large Tables) If a table is extremely large (e.g., ~1B rows or ~1 TB), remote building the Merkle tree from a single ACE node can be slowed by network latency. You can parallelize the build (per node) to speed up the process. diff --git a/docs/configuration.md b/docs/configuration.md index 204cf2f..1090fcd 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -38,6 +38,8 @@ The [`ace.yaml` file](https://github.com/pgEdge/ace/blob/main/ace.yaml) defines | mtree → cdc --> cdc_processing_timeout | Wall-clock budget (s) for a CDC catch-up drain before it gives up. Progress is durable, so a timeout means "re-run or raise", not "rebuild". **Default: 300** | | mtree → cdc --> cdc_metadata_flush_seconds | How often (s) CDC metadata is flushed to disk. **Default: 10** | | mtree → cdc --> cdc_flush_batch_size | Peak un-committed CDC changes a bounded drain buffers before applying them to the tree and freeing memory. Only primary keys are buffered, so memory scales with key size, not row width. **Default: 10000** | +| mtree → cdc --> adaptive_drain_fraction | During a bounded CDC drain (`mtree update` / `mtree table-diff`), a table whose decoded UPDATE count exceeds `max(adaptive_drain_min_changes, adaptive_drain_fraction × table row estimate)` is escalated: instead of per-update block marking, all of its Merkle-tree blocks are marked dirty once and rehashed in bulk from live data on the following tree update. Inserts and deletes are always tracked individually regardless of escalation, so block split/merge maintenance is unaffected. This bounds drain cost for heavily-updated tables (a bulk rehash costs about as much as a tree build, while per-change marking grows linearly with the backlog). Set to `-1` to disable escalation. Escalation never affects correctness — block hashes are always recomputed from live table data. **Default: 0.01** | +| mtree → cdc --> adaptive_drain_min_changes | Floor for the escalation threshold, so small tables and modest change volumes always use the precise per-change path. **Default: 1000** | | mtree --> schema | Schema used for mtree metadata/objects. **Default: "pgedge_ace"** | | mtree → diff --> min_block_size | Minimum Merkle diff block size. **Default: 1000** | | mtree → diff --> block_size | Target Merkle diff block size. **Default: 100000** | diff --git a/internal/infra/cdc/escalate.go b/internal/infra/cdc/escalate.go new file mode 100644 index 0000000..c131557 --- /dev/null +++ b/internal/infra/cdc/escalate.go @@ -0,0 +1,102 @@ +// /////////////////////////////////////////////////////////////////////////// +// +// # 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 cdc + +// escalator decides, per table, when a bounded CDC drain should stop tracking +// individual changes and instead mark the table's whole Merkle tree dirty for +// one bulk rehash from live table data. Per-change dirty-marking costs a +// range-containment join per PK (UpdateMtreeCounters); once a table's change +// count reaches a meaningful fraction of its rows, a single bulk rehash +// (~the cost of a tree build) is far cheaper. Escalating is always safe: +// leaf hashes are recomputed from live data, so over-marking dirty blocks can +// cost redundant rehash work but can never miss a change. +// +// Not goroutine-safe: bounded drains decode the stream on a single goroutine, +// which is the only place this is used. +type escalator struct { + minChanges int64 + fraction float64 + rowEstimate func(schema, table string) int64 + counts map[string]int64 + thresholds map[string]int64 + escalated map[string]struct{} +} + +func newEscalator(minChanges int64, fraction float64, rowEstimate func(schema, table string) int64) *escalator { + return &escalator{ + minChanges: minChanges, + fraction: fraction, + rowEstimate: rowEstimate, + counts: make(map[string]int64), + thresholds: make(map[string]int64), + escalated: make(map[string]struct{}), + } +} + +func escKey(schema, table string) string { return schema + "." + table } + +// noteChange records one decoded change for schema.table. It returns true +// exactly while the table sits at/over its threshold and has not yet been +// escalated -- the caller performs the mark-all-dirty DB write and then calls +// markEscalated. The threshold is resolved lazily on the first change: +// max(minChanges, fraction*rowEstimate), falling back to minChanges when the +// row estimate is unavailable (<= 0). +func (e *escalator) noteChange(schema, table string) bool { + k := escKey(schema, table) + if _, ok := e.escalated[k]; ok { + return false + } + th, ok := e.thresholds[k] + if !ok { + th = e.minChanges + if rows := e.rowEstimate(schema, table); rows > 0 { + if f := int64(e.fraction * float64(rows)); f > th { + th = f + } + } + e.thresholds[k] = th + } + e.counts[k]++ + return e.counts[k] >= th +} + +func (e *escalator) isEscalated(schema, table string) bool { + _, ok := e.escalated[escKey(schema, table)] + return ok +} + +func (e *escalator) markEscalated(schema, table string) { + e.escalated[escKey(schema, table)] = struct{}{} +} + +// threshold returns the resolved threshold for logging; 0 if not yet resolved. +func (e *escalator) threshold(schema, table string) int64 { + return e.thresholds[escKey(schema, table)] +} + +// purgeTableChanges drops every buffered UPDATE for schema.table across all +// in-flight transactions. Called at escalation time: the table's tree is +// about to be fully marked dirty, so buffered per-PK UPDATE work is redundant +// -- dropping it both skips the expensive containment-join applies and frees +// buffer memory. INSERTs and DELETEs are kept: their per-block counters drive +// block split/merge maintenance, which mark-all-dirty does not cover. +func purgeTableChanges(txChanges map[uint32][]cdcMsg, schema, table string) { + for xid, msgs := range txChanges { + kept := msgs[:0] + for _, m := range msgs { + if m.schema != schema || m.table != table || m.operation != "UPDATE" { + kept = append(kept, m) + } + } + txChanges[xid] = kept + } +} diff --git a/internal/infra/cdc/escalate_test.go b/internal/infra/cdc/escalate_test.go new file mode 100644 index 0000000..c1cc962 --- /dev/null +++ b/internal/infra/cdc/escalate_test.go @@ -0,0 +1,95 @@ +// /////////////////////////////////////////////////////////////////////////// +// +// # 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 cdc + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func fixedEstimate(rows int64) func(schema, table string) int64 { + return func(schema, table string) int64 { return rows } +} + +func TestEscalatorThresholdIsFractionOfRows(t *testing.T) { + // 1% of 500k rows = 5000 > min 1000, so the fraction wins. + e := newEscalator(1000, 0.01, fixedEstimate(500000)) + for i := 0; i < 4999; i++ { + assert.False(t, e.noteChange("public", "t"), "change %d must not escalate", i) + } + assert.True(t, e.noteChange("public", "t"), "5000th change must cross the threshold") + assert.Equal(t, int64(5000), e.threshold("public", "t")) +} + +func TestEscalatorMinChangesFloor(t *testing.T) { + // 1% of 5k rows = 50 < min 1000, so the floor wins. + e := newEscalator(1000, 0.01, fixedEstimate(5000)) + assert.Equal(t, int64(0), e.counts["public.t"]) // untouched until first change + for i := 0; i < 999; i++ { + assert.False(t, e.noteChange("public", "t")) + } + assert.True(t, e.noteChange("public", "t")) + assert.Equal(t, int64(1000), e.threshold("public", "t")) +} + +func TestEscalatorUnknownRowEstimateUsesFloor(t *testing.T) { + // rowEstimate <= 0 (metadata missing/unreadable) -> threshold = minChanges. + e := newEscalator(1000, 0.01, fixedEstimate(0)) + assert.Equal(t, int64(0), e.threshold("public", "t")) // unresolved yet + e.noteChange("public", "t") + assert.Equal(t, int64(1000), e.threshold("public", "t")) +} + +func TestEscalatorStopsCountingOnceEscalated(t *testing.T) { + e := newEscalator(2, 0, fixedEstimate(0)) + assert.False(t, e.noteChange("public", "t")) + assert.True(t, e.noteChange("public", "t")) + // Caller does the DB work, then: + e.markEscalated("public", "t") + assert.True(t, e.isEscalated("public", "t")) + // Further changes never re-trigger. + assert.False(t, e.noteChange("public", "t")) + assert.False(t, e.noteChange("public", "t")) +} + +func TestEscalatorTablesAreIndependent(t *testing.T) { + e := newEscalator(2, 0, fixedEstimate(0)) + assert.False(t, e.noteChange("public", "a")) + assert.True(t, e.noteChange("public", "a")) + e.markEscalated("public", "a") + assert.False(t, e.isEscalated("public", "b")) + assert.False(t, e.noteChange("public", "b"), "table b count must start fresh") +} + +func TestPurgeTableChanges(t *testing.T) { + txChanges := map[uint32][]cdcMsg{ + 1: { + {operation: "INSERT", schema: "public", table: "flood"}, + {operation: "INSERT", schema: "public", table: "keep"}, + {operation: "UPDATE", schema: "public", table: "flood"}, + }, + 2: { + {operation: "DELETE", schema: "public", table: "flood"}, + {operation: "UPDATE", schema: "public", table: "flood"}, + }, + } + purgeTableChanges(txChanges, "public", "flood") + // Only flood's UPDATEs are dropped; its INSERT/DELETE survive (they feed + // block split/merge counters), and other tables are untouched. + assert.Len(t, txChanges[1], 2) + assert.Equal(t, "INSERT", txChanges[1][0].operation) + assert.Equal(t, "flood", txChanges[1][0].table) + assert.Equal(t, "keep", txChanges[1][1].table) + assert.Len(t, txChanges[2], 1) + assert.Equal(t, "DELETE", txChanges[2][0].operation) +} diff --git a/internal/infra/cdc/listen.go b/internal/infra/cdc/listen.go index 04bf6a9..9faa8cb 100644 --- a/internal/infra/cdc/listen.go +++ b/internal/infra/cdc/listen.go @@ -21,6 +21,7 @@ import ( "time" "github.com/jackc/pglogrepl" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgproto3" "github.com/jackc/pgx/v5/pgxpool" @@ -209,6 +210,35 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont if cfg.CDCFlushBatchSize > 0 { flushBatchSize = cfg.CDCFlushBatchSize } + // Adaptive drain (bounded mode only): once a table's decoded UPDATE count + // crosses max(adaptive_drain_min_changes, adaptive_drain_fraction * rows), + // per-update dirty-marking for it is replaced by one MarkAllLeavesDirty + + // bulk rehash on the next tree update. Inserts/deletes are exempt: their + // per-block counters drive split/merge maintenance, so they are always + // tracked individually. Measured crossover: per-PK applies cost + // ~60us/change while a full-tree rehash costs ~a tree build, so the + // break-even sits near 1% of table rows. Escalation can only over-mark + // dirty blocks (leaf hashes are recomputed from live data), never miss a + // change. A negative fraction disables escalation. + var esc *escalator + if !continuous && cfg.AdaptiveDrainFraction >= 0 { + fraction := cfg.AdaptiveDrainFraction + if fraction == 0 { + fraction = 0.01 + } + minChanges := int64(cfg.AdaptiveDrainMinChanges) + if minChanges <= 0 { + minChanges = 1000 + } + esc = newEscalator(minChanges, fraction, func(schema, table string) int64 { + rows, err := queries.GetRowCountEstimateFromMetadata(ctx, pool, schema, table) + if err != nil { + logger.Warn("adaptive drain: no row estimate for %s.%s (using min-changes floor): %v", schema, table, err) + return 0 + } + return rows + }) + } lastFlushTime := time.Now() var conn *pgconn.PgConn @@ -268,6 +298,58 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont // standby update may confirm past the commit -- permanently skipping the // transaction and leaving the Merkle tree stale. inTx := false + + // bufferChange records one decoded change for the bounded drain. + // Escalation applies to UPDATEs ONLY: an update never changes a row's + // block membership, so mark-all-dirty fully covers its effect on leaf + // hashes, and skipping its per-PK tracking is pure savings. INSERTs and + // DELETEs are always tracked per-PK regardless of escalation -- their + // per-block counters drive block split/merge maintenance, and discarding + // them would leave the tree structurally unbalanced until a full rebuild + // (regression: the MerkleTree Split* suites fail if inserts are dropped). + // Returns a terminal error only if the escalation's mark-all-dirty write + // fails: proceeding without either per-PK marks or the bulk mark would + // silently miss divergence. + bufferChange := func(op string, rel *pglogrepl.RelationMessage, tup *pglogrepl.TupleData) error { + if esc != nil && op == "UPDATE" { + if esc.isEscalated(rel.Namespace, rel.RelationName) { + return nil + } + if esc.noteChange(rel.Namespace, rel.RelationName) { + // Keepalive before the blocking write, like the sub-batch + // flush below, so a slow mark-all cannot trip wal_sender_timeout. + if conn != nil { + if err := pglogrepl.SendStandbyStatusUpdate(ctx, conn, pglogrepl.StandbyStatusUpdate{WALWritePosition: lastLSN}); err != nil { + logger.Warn("standby status update before adaptive-drain escalation failed: %v", err) + } else { + nextStandbyMessageDeadline = time.Now().Add(10 * time.Second) + } + } + // Sanitized: the mtree table is created quoted/case-preserved. + mtreeTable := pgx.Identifier{mtreeSchema, fmt.Sprintf("ace_mtree_%s_%s", rel.Namespace, rel.RelationName)}.Sanitize() + // processingCtx so the write survives the drain deadline, + // like every other apply here. + marked, err := queries.MarkAllLeavesDirty(processingCtx, pool, mtreeTable) + if err != nil { + return fmt.Errorf("adaptive drain: failed to mark all leaves dirty for %s: %w", mtreeTable, err) + } + esc.markEscalated(rel.Namespace, rel.RelationName) + purgeTableChanges(txChanges, rel.Namespace, rel.RelationName) + logger.Info("adaptive drain: %s.%s crossed %d buffered UPDATE changes; marked all %d blocks dirty and skipping its per-update tracking for the rest of this drain (inserts/deletes still tracked for block splits/merges)", + rel.Namespace, rel.RelationName, esc.threshold(rel.Namespace, rel.RelationName), marked) + return nil + } + } + txChanges[currentXID] = append(txChanges[currentXID], cdcMsg{ + operation: op, + schema: rel.Namespace, + table: rel.RelationName, + relation: rel, + pks: getPKs(rel, tup), + }) + return nil + } + var wg sync.WaitGroup errCh := make(chan error, 1) stopStreaming := false @@ -548,31 +630,22 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont relations[logicalMsg.RelationID] = logicalMsg case *pglogrepl.InsertMessage: rel := relations[logicalMsg.RelationID] - txChanges[currentXID] = append(txChanges[currentXID], cdcMsg{ - operation: "INSERT", - schema: rel.Namespace, - table: rel.RelationName, - relation: rel, - pks: getPKs(rel, logicalMsg.Tuple), - }) + if err := bufferChange("INSERT", rel, logicalMsg.Tuple); err != nil { + processingErr = err + stopStreaming = true + } case *pglogrepl.UpdateMessage: rel := relations[logicalMsg.RelationID] - txChanges[currentXID] = append(txChanges[currentXID], cdcMsg{ - operation: "UPDATE", - schema: rel.Namespace, - table: rel.RelationName, - relation: rel, - pks: getPKs(rel, logicalMsg.NewTuple), - }) + if err := bufferChange("UPDATE", rel, logicalMsg.NewTuple); err != nil { + processingErr = err + stopStreaming = true + } case *pglogrepl.DeleteMessage: rel := relations[logicalMsg.RelationID] - txChanges[currentXID] = append(txChanges[currentXID], cdcMsg{ - operation: "DELETE", - schema: rel.Namespace, - table: rel.RelationName, - relation: rel, - pks: getPKs(rel, logicalMsg.OldTuple), - }) + if err := bufferChange("DELETE", rel, logicalMsg.OldTuple); err != nil { + processingErr = err + stopStreaming = true + } } // Flush sub-batches of the in-flight transaction so memory stays diff --git a/pkg/config/config.go b/pkg/config/config.go index 6911d4b..5eb4673 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -64,6 +64,16 @@ type MTreeConfig struct { CDCProcessingTimeout int `yaml:"cdc_processing_timeout"` CDCMetadataFlushSec int `yaml:"cdc_metadata_flush_seconds"` CDCFlushBatchSize int `yaml:"cdc_flush_batch_size"` + // AdaptiveDrainFraction and AdaptiveDrainMinChanges tune when a + // bounded CDC drain escalates a flooded table's UPDATE tracking from + // per-change dirty-marking to one mark-all-dirty + bulk rehash + // (inserts/deletes are always tracked per-change: their per-block + // counters drive block split/merge maintenance). Threshold per + // table = max(min_changes, fraction * table_row_estimate), counting + // UPDATEs only. 0 means "use default" (0.01 / 1000); a negative + // fraction disables escalation entirely. + AdaptiveDrainFraction float64 `yaml:"adaptive_drain_fraction"` + AdaptiveDrainMinChanges int `yaml:"adaptive_drain_min_changes"` } `yaml:"cdc"` Schema string `yaml:"schema"` Diff struct { diff --git a/tests/integration/cdc_adaptive_drain_test.go b/tests/integration/cdc_adaptive_drain_test.go new file mode 100644 index 0000000..c207961 --- /dev/null +++ b/tests/integration/cdc_adaptive_drain_test.go @@ -0,0 +1,173 @@ +// /////////////////////////////////////////////////////////////////////////// +// +// # 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" + "strings" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/pgedge/ace/internal/consistency/mtree" + "github.com/pgedge/ace/internal/infra/cdc" + "github.com/pgedge/ace/pkg/config" + "github.com/stretchr/testify/require" +) + +// dirtyLeafCounts returns (dirty, total) leaf-block counts for a table's +// merkle tree on the given node. +func dirtyLeafCounts(t *testing.T, ctx context.Context, pool *pgxpool.Pool, tableName string) (int64, int64) { + t.Helper() + mtreeIdent := pgx.Identifier{ + config.Get().MTree.Schema, + fmt.Sprintf("ace_mtree_%s_%s", testSchema, tableName), + }.Sanitize() + var dirty, total int64 + err := pool.QueryRow(ctx, fmt.Sprintf( // nosemgrep + `SELECT count(*) FILTER (WHERE dirty), count(*) FROM %s WHERE node_level = 0`, mtreeIdent, + )).Scan(&dirty, &total) + require.NoError(t, err, "count dirty leaves for %s", tableName) + return dirty, total +} + +// TestAdaptiveDrainEscalation exercises the three regimes of the adaptive +// bounded drain in one pass over a shared slot: +// - flood table: changes >> threshold -> escalate: ALL leaves marked dirty +// - single table: one changed row -> exactly one leaf dirty (cheap path) +// - idle table: no changes -> zero leaves dirty (untouched) +// +// and then verifies end-to-end correctness: DiffMtree finds exactly the +// diverged rows for both changed tables. +func TestAdaptiveDrainEscalation(t *testing.T) { + ctx := context.Background() + env := newSpockEnv() + + tables := []struct { + name string + rows int + }{ + {"adaptive_flood", 50000}, + {"adaptive_single", 5000}, + {"adaptive_idle", 5000}, + } + + src := pgx.Identifier{testSchema, "customers_1M"}.Sanitize() + tasks := make(map[string]*mtree.MerkleTreeTask, len(tables)) + + t.Cleanup(func() { + for _, tbl := range tables { + if task := tasks[tbl.name]; task != nil { + if err := task.MtreeTeardown(); err != nil { + t.Logf("cleanup: MtreeTeardown(%s) failed: %v", tbl.name, err) + } + } + for _, pool := range []*pgxpool.Pool{env.N1Pool, env.N2Pool} { + ident := pgx.Identifier{testSchema, tbl.name}.Sanitize() + if _, err := pool.Exec(context.Background(), fmt.Sprintf(`DROP TABLE IF EXISTS %s CASCADE`, ident)); err != nil { // nosemgrep + t.Logf("cleanup: drop %s failed: %v", tbl.name, err) + } + } + } + }) + + // Seed identical data on both nodes (fresh tables are in no spock + // replication set, so per-node inserts stay local), then init once and + // build the merkle tree for each table. MtreeInit drops and recreates the + // SHARED publication and slot (SetupPublication/SetupReplicationSlot), so + // it must run exactly once; running it per table would unpublish the + // tables already built. BuildMtree adds each table to the publication and + // CDC metadata itself. + for i, tbl := range tables { + ident := pgx.Identifier{testSchema, tbl.name}.Sanitize() + for _, pool := range []*pgxpool.Pool{env.N1Pool, env.N2Pool} { + require.NoError(t, createTestTable(ctx, pool, testSchema, tbl.name)) + _, err := pool.Exec(ctx, fmt.Sprintf( // nosemgrep + `INSERT INTO %s SELECT * FROM %s WHERE index <= %d`, ident, src, tbl.rows)) + require.NoError(t, err, "seed %s", tbl.name) + // ANALYZE so BuildMtree's row estimate (and therefore the + // escalation threshold's metadata row count) is accurate. + _, err = pool.Exec(ctx, "ANALYZE "+ident) // nosemgrep + require.NoError(t, err) + } + task := env.newMerkleTreeTask(t, fmt.Sprintf("%s.%s", testSchema, tbl.name), + []string{env.ServiceN1, env.ServiceN2}) + // Pin the diff cap: MaxDiffRows==0 falls back to config, and a low + // configured cap would break the exact-count assertions below. + task.MaxDiffRows = int64(tbl.rows) + require.NoError(t, task.RunChecks(false), "RunChecks(%s)", tbl.name) + if i == 0 { + require.NoError(t, task.MtreeInit(), "MtreeInit (once, first table)") + } + require.NoError(t, task.BuildMtree(), "BuildMtree(%s)", tbl.name) + tasks[tbl.name] = task + } + + // Diverge on n2 only, under repair mode (captured by n2's CDC slot, not + // replicated to n1). Flood: 20k changed rows on a 50k-row table -- far + // over the escalation threshold max(1000, 0.01*50000=500) = 1000. + // Single: exactly one row. + env.withRepairMode(t, ctx, env.N2Pool, func(conn *pgxpool.Conn) { + floodIdent := pgx.Identifier{testSchema, "adaptive_flood"}.Sanitize() + _, err := conn.Exec(ctx, fmt.Sprintf( // nosemgrep + `UPDATE %s SET email = 'flood_' || index::text WHERE index <= 20000`, floodIdent)) + require.NoError(t, err, "flood divergence") + singleIdent := pgx.Identifier{testSchema, "adaptive_single"}.Sanitize() + _, err = conn.Exec(ctx, fmt.Sprintf( // nosemgrep + `UPDATE %s SET email = 'single_change' WHERE index = 42`, singleIdent)) + require.NoError(t, err, "single-row divergence") + }) + + // Drain n2's slot directly (no hash recompute) so the dirty flags are + // observable before UpdateMtree clears them. ClusterNodes[1] is n2. + require.NoError(t, cdc.UpdateFromCDC(ctx, pgCluster.ClusterNodes[1]), + "bounded CDC drain on n2") + + // Flood table escalated: every leaf dirty. + dirty, total := dirtyLeafCounts(t, ctx, env.N2Pool, "adaptive_flood") + require.Greater(t, total, int64(0)) + require.Equal(t, total, dirty, + "flood table must escalate to mark-all-dirty (dirty=%d of %d leaves)", dirty, total) + + // Single-change table stayed on the per-PK path: exactly one leaf dirty. + dirty, total = dirtyLeafCounts(t, ctx, env.N2Pool, "adaptive_single") + require.Greater(t, total, int64(1)) + require.Equal(t, int64(1), dirty, + "single-row table must have exactly one dirty leaf, got %d of %d", dirty, total) + + // Idle table untouched: zero leaves dirty. + dirty, _ = dirtyLeafCounts(t, ctx, env.N2Pool, "adaptive_idle") + require.Equal(t, int64(0), dirty, "idle table must have no dirty leaves") + + // End-to-end correctness: the escalated table's diff finds every diverged + // row, and the single-change table finds exactly one. + pairKey := env.ServiceN1 + "/" + env.ServiceN2 + if strings.Compare(env.ServiceN1, env.ServiceN2) > 0 { + pairKey = env.ServiceN2 + "/" + env.ServiceN1 + } + countDiffs := func(task *mtree.MerkleTreeTask) int { + nodeDiffs, ok := task.DiffResult.NodeDiffs[pairKey] + if !ok { + return 0 + } + return len(nodeDiffs.Rows[env.ServiceN1]) + } + + require.NoError(t, tasks["adaptive_flood"].DiffMtree(), "DiffMtree(flood)") + require.Equal(t, 20000, countDiffs(tasks["adaptive_flood"]), + "escalated table must report every diverged row") + + require.NoError(t, tasks["adaptive_single"].DiffMtree(), "DiffMtree(single)") + require.Equal(t, 1, countDiffs(tasks["adaptive_single"]), + "single-change table must report exactly one diverged row") +}