feat(mtree): adaptively escalate flooded tables to bulk rehash

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:

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.

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** |

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

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

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