From 872fe72fb579fbc00307651a90f7077390345815 Mon Sep 17 00:00:00 2001 From: Mason Sharp Date: Fri, 10 Jul 2026 13:27:55 -0700 Subject: [PATCH 1/2] fix(cdc): mark all leaves dirty when a TRUNCATE is decoded The CDC drain's message switch only handled Insert/Update/Delete, so the TruncateMessage pgoutput publishes for a TRUNCATE (the publication is created with the default publish list, which includes truncate) was silently dropped. No leaves were dirtied, the truncated node's Merkle tree kept its pre-truncate hashes, and mtree table-diff reported the nodes as in sync. A truncate carries only relation OIDs -- no per-row changes -- so per-PK tracking is impossible. React with the adaptive drain's bulk path instead: mark every leaf of each truncated relation dirty (cascades list all published relations) so the next tree update rehashes from the live, now-empty table. Buffered UPDATEs for the table are purged as at escalation time; a failed mark is terminal so a stale tree is never reported current. Applies to both bounded drains (table-diff/update) and continuous listen, which share the decode switch. The integration test reproduces the field report: identical nodes, clean diff, TRUNCATE on n2, then a second diff must surface every row instead of "Merkle trees are identical". --- internal/infra/cdc/listen.go | 57 ++++++++++ tests/integration/mtree_truncate_test.go | 131 +++++++++++++++++++++++ 2 files changed, 188 insertions(+) create mode 100644 tests/integration/mtree_truncate_test.go diff --git a/internal/infra/cdc/listen.go b/internal/infra/cdc/listen.go index 9faa8cb..c978b3f 100644 --- a/internal/infra/cdc/listen.go +++ b/internal/infra/cdc/listen.go @@ -350,6 +350,42 @@ 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) + 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 +682,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 diff --git a/tests/integration/mtree_truncate_test.go b/tests/integration/mtree_truncate_test.go new file mode 100644 index 0000000..5abfd45 --- /dev/null +++ b/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) + } +} From 49da06a1304dde55a4515fa76e4626e5ebe3e933 Mon Sep 17 00:00:00 2001 From: Mason Sharp Date: Mon, 13 Jul 2026 07:16:49 -0700 Subject: [PATCH 2/2] fix(cdc): mark truncated table escalated to skip redundant UPDATE tracking markTableTruncated already leaves every leaf dirty, which is exactly the adaptive drain's escalated state. Record it in the escalator so UPDATEs landing on a re-populated table later in the same bounded drain skip per-PK tracking instead of re-running the threshold count and a redundant mark-all-dirty. Guarded by esc != nil (nil in continuous mode); INSERT/DELETE split-merge tracking is unaffected. --- internal/infra/cdc/listen.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/internal/infra/cdc/listen.go b/internal/infra/cdc/listen.go index c978b3f..ebf7412 100644 --- a/internal/infra/cdc/listen.go +++ b/internal/infra/cdc/listen.go @@ -382,6 +382,12 @@ func processReplicationStream(ctx context.Context, nodeInfo map[string]any, cont 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 }