fix(cdc): mark all leaves dirty when a TRUNCATE is decoded

internal/infra/cdc/listen.go

@@ -350,6 +350,48 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont
 		return nil
 	}
 
+	// markTableTruncated reacts to a decoded TRUNCATE for one relation.
+	// pgoutput publishes a truncate as a single message carrying only relation
+	// OIDs -- no per-row DELETEs -- so per-PK tracking is impossible and the
+	// only sound reaction is the adaptive drain's bulk path: mark every leaf
+	// dirty so the next tree update rehashes the table from live (now empty)
+	// data. Dropping the message instead would leave the tree at its
+	// pre-truncate state and table-diff would report a truncated node as in
+	// sync. Buffered UPDATEs for the table are purged as at escalation time
+	// (redundant once every leaf is dirty); INSERT/DELETE tracking continues
+	// for split/merge counter maintenance. Block merges from the truncate
+	// itself lag until the next build, which costs rehash work, never
+	// correctness. A failure to mark is terminal for the same reason as in
+	// bufferChange: proceeding would silently miss the divergence.
+	markTableTruncated := func(rel *pglogrepl.RelationMessage) error {
+		// Keepalive before the blocking write, like the escalation path, 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 truncate mark-all 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("failed to mark all leaves dirty for truncated table %s.%s: %w", rel.Namespace, rel.RelationName, err)
+		}
+		purgeTableChanges(txChanges, rel.Namespace, rel.RelationName)
+		// All leaves are now dirty, which is the escalated state, so record it:
+		// later UPDATEs on a re-populated table skip per-PK tracking rather than
+		// re-running the threshold count and a redundant mark-all.
+		if esc != nil {
+			esc.markEscalated(rel.Namespace, rel.RelationName)
+		}
+		logger.Info("TRUNCATE decoded for %s.%s: marked all %d blocks dirty for rehash on next tree update", rel.Namespace, rel.RelationName, marked)
+		return nil
+	}
+
 	var wg sync.WaitGroup
 	errCh := make(chan error, 1)
 	stopStreaming := false
@@ -646,6 +688,27 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont
 						processingErr = err
 						stopStreaming = true
 					}
+				case *pglogrepl.TruncateMessage:
+					// The publication is created with the default publish list,
+					// which includes truncate, so the server does send these. A
+					// cascaded truncate lists every affected published relation
+					// here. pgoutput sends the RelationMessages ahead of this
+					// message, so the lookups should always hit; treat a miss as
+					// terminal rather than skipping a table whose tree would then
+					// silently go stale.
+					for _, relID := range logicalMsg.RelationIDs {
+						rel, ok := relations[relID]
+						if !ok {
+							processingErr = fmt.Errorf("TRUNCATE references unknown relation OID %d; cannot mark its Merkle tree dirty", relID)
+							stopStreaming = true
+							break
+						}
+						if err := markTableTruncated(rel); err != nil {
+							processingErr = err
+							stopStreaming = true
+							break
+						}
+					}
 				}
 
 				// Flush sub-batches of the in-flight transaction so memory stays

tests/integration/mtree_truncate_test.go

@@ -0,0 +1,131 @@
+// ///////////////////////////////////////////////////////////////////////////
+//
+// # 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"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/stretchr/testify/require"
+)
+
+// TestMerkleTreeDiffAfterTruncate reproduces a field report: after an initial
+// build+diff, TRUNCATE on one node followed by another mtree table-diff
+// reported the nodes as in sync. pgoutput publishes TRUNCATE as a single
+// TruncateMessage carrying only relation OIDs — no per-row DELETEs — so a CDC
+// drain that only handles Insert/Update/Delete messages dirties no leaves,
+// leaves the truncated node's tree at its pre-truncate state, and the diff
+// compares two identical stale trees. The drain must react to the truncate by
+// marking every leaf of the table dirty so the next update rehashes from the
+// (now empty) live table.
+func TestMerkleTreeDiffAfterTruncate(t *testing.T) {
+	ctx := context.Background()
+	env := newSpockEnv()
+	tableName := "mtree_truncate_diff"
+	qualifiedTable := fmt.Sprintf("%s.%s", testSchema, tableName)
+	safeTable := pgx.Identifier{testSchema, tableName}.Sanitize()
+
+	const rowCount = 20
+	seedIdenticalTruncateTable(t, ctx, env, safeTable, qualifiedTable, rowCount)
+
+	mtreeTask := env.newMerkleTreeTask(t, qualifiedTable,
+		[]string{env.ServiceN1, env.ServiceN2})
+	require.NoError(t, mtreeTask.RunChecks(false))
+	require.NoError(t, mtreeTask.MtreeInit())
+	t.Cleanup(func() {
+		if err := mtreeTask.MtreeTeardown(); err != nil {
+			t.Logf("MtreeTeardown cleanup: %v", err)
+		}
+	})
+	require.NoError(t, mtreeTask.BuildMtree())
+	require.NoError(t, mtreeTask.DiffMtree())
+
+	pair := env.pairKey()
+	if nodeDiffs, ok := mtreeTask.DiffResult.NodeDiffs[pair]; ok {
+		require.Empty(t, extractDiffIDs(nodeDiffs.Rows[env.ServiceN1]),
+			"nodes seeded identically; first diff must be clean")
+		require.Empty(t, extractDiffIDs(nodeDiffs.Rows[env.ServiceN2]),
+			"nodes seeded identically; first diff must be clean")
+	}
+
+	// The divergence under test: TRUNCATE on n2 only. repair_mode keeps
+	// Spock from replicating it to n1 even if the table were picked up.
+	env.withRepairMode(t, ctx, env.N2Pool, func(conn *pgxpool.Conn) {
+		_, err := conn.Exec(ctx, "TRUNCATE TABLE "+safeTable) // nosemgrep
+		require.NoError(t, err, "truncate %s on n2", qualifiedTable)
+	})
+
+	// DiffMtree drains CDC (via UpdateMtree) before comparing, mirroring the
+	// CLI's `ace mtree table-diff`. All rows now exist only on n1.
+	require.NoError(t, mtreeTask.DiffMtree())
+
+	nodeDiffs, ok := mtreeTask.DiffResult.NodeDiffs[pair]
+	require.True(t, ok,
+		"diff after TRUNCATE on n2 reported the nodes as identical; "+
+			"the CDC drain dropped the TruncateMessage. Full result: %+v",
+		mtreeTask.DiffResult)
+
+	gotN1 := extractDiffIDs(nodeDiffs.Rows[env.ServiceN1])
+	require.Len(t, gotN1, rowCount,
+		"every seeded row exists only on n1 after n2's truncate")
+	require.Empty(t, extractDiffIDs(nodeDiffs.Rows[env.ServiceN2]),
+		"n2 is empty after truncate; it can contribute no rows")
+}
+
+// seedIdenticalTruncateTable creates safeTable on every node, empties it, and
+// seeds rowCount identical rows so a first mtree diff is clean. It registers
+// table + diff-file cleanup on t. Conventions mirror
+// TestMerkleTreeBidirectionalDiff: IF NOT EXISTS in case Spock DDL replication
+// propagates the first CREATE, and the table stays outside any repset so writes
+// to one node cannot leak to the other.
+func seedIdenticalTruncateTable(t *testing.T, ctx context.Context, env *testEnv, safeTable, qualifiedTable string, rowCount int) {
+	t.Helper()
+	for _, pool := range env.pools() {
+		_, err := pool.Exec(ctx, // nosemgrep
+			"CREATE TABLE IF NOT EXISTS "+safeTable+" (id INT PRIMARY KEY, payload TEXT)") // nosemgrep
+		require.NoError(t, err, "create %s", qualifiedTable)
+	}
+	t.Cleanup(func() {
+		for _, pool := range env.pools() {
+			_, _ = pool.Exec(ctx, "DROP TABLE IF EXISTS "+safeTable+" CASCADE") // nosemgrep
+		}
+		files, _ := filepath.Glob("*_diffs-*.json")
+		for _, f := range files {
+			os.Remove(f)
+		}
+	})
+
+	for _, pool := range env.pools() {
+		env.withRepairMode(t, ctx, pool, func(conn *pgxpool.Conn) {
+			_, err := conn.Exec(ctx, "TRUNCATE TABLE "+safeTable) // nosemgrep
+			require.NoError(t, err, "pre-test truncate %s", qualifiedTable)
+		})
+	}
+
+	for _, pool := range env.pools() {
+		for id := 1; id <= rowCount; id++ {
+			_, err := pool.Exec(ctx, // nosemgrep
+				"INSERT INTO "+safeTable+" (id, payload) VALUES ($1, $2)", // nosemgrep
+				id, fmt.Sprintf("row-%d", id))
+			require.NoError(t, err, "seed id=%d", id)
+		}
+	}
+	for _, pool := range env.pools() {
+		_, err := pool.Exec(ctx, "ANALYZE "+safeTable) // nosemgrep
+		require.NoError(t, err)
+	}
+}