diff --git a/internal/infra/cdc/listen.go b/internal/infra/cdc/listen.go index 9faa8cb..ebf7412 100644 --- a/internal/infra/cdc/listen.go +++ b/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 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) + } +}