From 46024c573bcb10b323874651702581cc312f4425 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=81lvaro=20Herrera?= Date: Thu, 24 Sep 2026 11:52:47 +0200 Subject: [PATCH] Wait for transactions of an initial decoding snapshot to commit SnapBuildInitialSnapshot() converts the snapshot builder's list of committed transactions into a regular MVCC snapshot, which is then used with HeapTupleSatisfiesMVCC(). However, a snapshot produced that way and used as an MVCC snapshot can potentially cause data corruption, if it captures a transaction after it writes its commit WAL record but before it has updated its CLOG entry: it will be incorrectly used to set hint bits as if that transaction had aborted. Fix by having SnapBuildInitialSnapshot() wait until such transactions have removed themselves from procarray, which guarantees correct visibility. Other uses of SnapBuildBuildSnapshot only use the snapshot as historical, so they don't need the waits, but add commentary in that function to alert possible future callers. While at this, backpatch commit 504fe10d1d26, "Fix misplaced comment in snapbuild.c." The comment it moved was nearby the place we modify. This bug was diagnosed by stress-testing of REPACK (CONCURRENTLY) in branch 19, so add a test case that tickles it using that feature in that branch and master. However, the bug exists in all branches, and in branch 18 we can add a test with regular logical decoding using injection points that tickles it. Unfortunately that test cannot be backpatched to earlier branches for lack of facilities. Reported-by: Mihail Nikalayeu Author: Antonin Houska Author: Rui Zhao Backpatch-through: 14 Discussion: https://postgr.es/m/85833.1768840165@localhost Discussion: https://postgr.es/m/CADzfLwU78as45To9a%3D-Qkr5jEg3tMxc5rUtdKy2MTv4r_SDGng%40mail.gmail.com --- src/backend/access/transam/xact.c | 10 ++ src/backend/replication/logical/snapbuild.c | 39 +++++ src/test/modules/injection_points/Makefile | 1 + .../expected/repack_commit_race.out | 64 ++++++++ .../injection_points/injection_points.c | 11 +- src/test/modules/injection_points/meson.build | 1 + .../specs/repack_commit_race.spec | 95 +++++++++++ src/test/recovery/meson.build | 1 + .../recovery/t/057_snapshot_commit_race.pl | 148 ++++++++++++++++++ 9 files changed, 367 insertions(+), 3 deletions(-) create mode 100644 src/test/modules/injection_points/expected/repack_commit_race.out create mode 100644 src/test/modules/injection_points/specs/repack_commit_race.spec create mode 100644 src/test/recovery/t/057_snapshot_commit_race.pl diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index ebb010853cf..7b67db514ec 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -65,6 +65,7 @@ #include "utils/builtins.h" #include "utils/combocid.h" #include "utils/guc.h" +#include "utils/injection_point.h" #include "utils/inval.h" #include "utils/memutils.h" #include "utils/relmapper.h" @@ -1377,6 +1378,9 @@ RecordTransactionCommit(void) &RelcacheInitFileInval); wrote_xlog = (XactLastRecEnd != 0); + /* Load the injection point before entering the critical section */ + INJECTION_POINT_LOAD("commit-before-clog-update"); + /* * If we haven't been assigned an XID yet, we neither can, nor do we want * to write a COMMIT record. @@ -1543,6 +1547,12 @@ RecordTransactionCommit(void) { XLogFlush(XactLastRecEnd); + /* + * The commit record is on disk, but not in CLOG yet. A test can stop + * here to see what others make of the transaction meanwhile. + */ + INJECTION_POINT_CACHED("commit-before-clog-update", NULL); + /* * Now we may update the CLOG, if we wrote a COMMIT record above */ diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c index de491ea0c4b..69caf5b02e7 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -359,6 +359,12 @@ SnapBuildSnapDecRefcount(Snapshot snap) * In-progress transactions with catalog access are *not* allowed to modify * these snapshots; they have to copy them and fill in appropriate ->curcid * and ->subxip/subxcnt values. + * + * Note that some of the transactions in the returned snapshot->xip might not + * have finished committing yet (we saw their commit WAL records, but we don't + * know if they've removed themselves from procarray). Callers that want to + * use the returned snapshot as an MVCC one must wait for this to happen; see + * SnapBuildInitialSnapshot(). */ static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder) @@ -470,6 +476,39 @@ SnapBuildInitialSnapshot(SnapBuild *builder) snap = SnapBuildBuildSnapshot(builder); + /* + * Transactions appear in snap->xip as soon as their commit WAL records + * are decoded, but that doesn't imply they can be considered committed in + * a regular MVCC snapshot: that happens only when they remove themselves + * from procarray. Wait for this to happen; otherwise somebody using the + * snapshot might set hint bits incorrectly. + * + * This is not needed during recovery, because the decoded commit record + * has been replayed already. + */ + if (!RecoveryInProgress()) + { + RunningTransactions running; + + running = GetRunningTransactionData(); + LWLockRelease(ProcArrayLock); + LWLockRelease(XidGenLock); + + /* + * Note we walk the array up to running->xcnt only, omitting this wait + * for subtransactions: the subxacts are covered by their top-level + * transaction already, there's no need for an additional wait. + */ + for (int i = 0; i < running->xcnt; i++) + { + TransactionId running_xid = running->xids[i]; + + if (bsearch(&running_xid, snap->xip, snap->xcnt, + sizeof(TransactionId), xidComparator) != NULL) + XactLockTableWait(running_xid, NULL, NULL, XLTW_None); + } + } + /* * Building an initial snapshot is expensive and an unenforced xmin * horizon would have bad consequences, therefore always double-check that diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 9d8b4b3540c..1c680abf7dd 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -18,6 +18,7 @@ ISOLATION = basic \ inplace \ reindex_concurrently_deferred \ repack \ + repack_commit_race \ repack_decode \ repack_temporal \ repack_temporal_multirange \ diff --git a/src/test/modules/injection_points/expected/repack_commit_race.out b/src/test/modules/injection_points/expected/repack_commit_race.out new file mode 100644 index 00000000000..4d0286029a5 --- /dev/null +++ b/src/test/modules/injection_points/expected/repack_commit_race.out @@ -0,0 +1,64 @@ +Parsed test spec with 5 sessions + +starting permutation: s2_begin s1_repack s3_begin s2_rollback s4_changes s3_rollback s5_wakeup s1_check +injection_points_attach +----------------------- + +(1 row) + +step s2_begin: + BEGIN; + SELECT pg_current_xact_id() IS NOT NULL; + +?column? +-------- +t +(1 row) + +step s1_repack: + REPACK (CONCURRENTLY) repack_race; + +step s3_begin: + BEGIN; + SELECT pg_current_xact_id() IS NOT NULL; + +?column? +-------- +t +(1 row) + +step s2_rollback: + ROLLBACK; + +step s4_changes: + INSERT INTO repack_race(i, j) VALUES (3, 3); + UPDATE repack_race SET j = j + 1 WHERE i = 1; + DELETE FROM repack_race WHERE i = 2; + +step s3_rollback: + ROLLBACK; + +step s5_wakeup: + SELECT injection_points_wakeup('commit-before-clog-update'); + +injection_points_wakeup +----------------------- + +(1 row) + +step s1_repack: <... completed> +step s4_changes: <... completed> +step s1_check: + SELECT i, j FROM repack_race ORDER BY i; + +i|j +-+- +1|2 +3|3 +(2 rows) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/injection_points.c b/src/test/modules/injection_points/injection_points.c index 66d8158d0c2..5e1bbc2f5c9 100644 --- a/src/test/modules/injection_points/injection_points.c +++ b/src/test/modules/injection_points/injection_points.c @@ -246,12 +246,17 @@ injection_wait(const char *name, const void *private_data, void *arg) char *argstr = arg; int delay_us = 0; - if (inj_state == NULL) - injection_init_shmem(); - + /* + * Check the condition before attaching to the shared state: attaching + * allocates memory, which a process that is not meant to wait here must + * not do if the injection point is in a critical section. + */ if (!injection_point_allowed(condition, argstr)) return; + if (inj_state == NULL) + injection_init_shmem(); + /* * Use the injection point name for this custom wait event. Note that * this custom wait event name is not released, but we don't care much for diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 80a09f34d78..45117b6ffec 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -47,6 +47,7 @@ tests += { 'inplace', 'reindex_concurrently_deferred', 'repack', + 'repack_commit_race', 'repack_decode', 'repack_temporal', 'repack_temporal_multirange', diff --git a/src/test/modules/injection_points/specs/repack_commit_race.spec b/src/test/modules/injection_points/specs/repack_commit_race.spec new file mode 100644 index 00000000000..9926a45839e --- /dev/null +++ b/src/test/modules/injection_points/specs/repack_commit_race.spec @@ -0,0 +1,95 @@ +# Test that the logical decoding initial snapshot build, used for +# REPACK (CONCURRENTLY), waits for transactions to remove themselves +# from ProcArray. Transactions are initially considered committed when +# their commit WAL records are decoded, but that's not sufficient +# for MVCC correctness. +setup +{ + CREATE EXTENSION injection_points; + + CREATE TABLE repack_race(i int PRIMARY KEY, j int); + INSERT INTO repack_race(i, j) VALUES (1, 1), (2, 2); +} + +teardown +{ + DROP TABLE repack_race; + DROP EXTENSION injection_points; +} + +session s1 +step s1_repack +{ + REPACK (CONCURRENTLY) repack_race; +} +step s1_check +{ + SELECT i, j FROM repack_race ORDER BY i; +} + +# s2 and s3 keep a transaction with an XID open, so that the snapshot builder +# has to go through its BUILDING_SNAPSHOT and FULL_SNAPSHOT states instead of +# becoming consistent right away. +session s2 +step s2_begin +{ + BEGIN; + SELECT pg_current_xact_id() IS NOT NULL; +} +step s2_rollback +{ + ROLLBACK; +} + +session s3 +step s3_begin +{ + BEGIN; + SELECT pg_current_xact_id() IS NOT NULL; +} +step s3_rollback +{ + ROLLBACK; +} + +# s4 changes the table and stops after writing its commit record, before +# updating CLOG. +session s4 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('commit-before-clog-update', 'wait'); +} +step s4_changes +{ + INSERT INTO repack_race(i, j) VALUES (3, 3); + UPDATE repack_race SET j = j + 1 WHERE i = 1; + DELETE FROM repack_race WHERE i = 2; +} +teardown +{ + SELECT injection_points_detach('commit-before-clog-update'); +} + +session s5 +step s5_wakeup +{ + SELECT injection_points_wakeup('commit-before-clog-update'); +} + +# The snapshot builder waits for s2, then for s3. While it waits for s3, s4 +# writes its commit record: the builder will count s4 as committed and start +# decoding after it, but CLOG does not know about s4 yet. REPACK must not use +# its snapshot before s4 has finished committing. +# +# s4 cannot finish before s5 wakes it up, and s1 cannot finish before s4 does; +# the marker on s4_changes keeps the reporting order stable. +permutation + s2_begin + s1_repack + s3_begin + s2_rollback + s4_changes(s1_repack) + s3_rollback + s5_wakeup + s1_check diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index 72113c5ac6e..ebb12dd8766 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -65,6 +65,7 @@ tests += { 't/054_unlogged_sequence_promotion.pl', 't/055_cascade_reconnect.pl', 't/056_standby_snapshot_export.pl', + 't/057_snapshot_commit_race.pl', ], }, } diff --git a/src/test/recovery/t/057_snapshot_commit_race.pl b/src/test/recovery/t/057_snapshot_commit_race.pl new file mode 100644 index 00000000000..c15a4266aae --- /dev/null +++ b/src/test/recovery/t/057_snapshot_commit_race.pl @@ -0,0 +1,148 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# The snapshot of CREATE_REPLICATION_SLOT ... USE_SNAPSHOT must not be handed +# out while a transaction it takes as committed is still between writing its +# commit record and removing itself from procarray. Checks that the slot +# creation waits for such a transaction, and that neither the slot's snapshot +# nor a later session loses the transaction's changes. +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; +use Time::HiRes qw(usleep); + +plan skip_all => 'Injection points not supported by this build' + unless $ENV{enable_injection_points} eq 'yes'; + +my $node = PostgreSQL::Test::Cluster->new('primary'); +$node->init(allows_streaming => 'logical'); +$node->start; + +if (!$node->check_extension('injection_points')) +{ + plan skip_all => 'Extension injection_points not installed'; +} + +$node->safe_psql('postgres', q(CREATE EXTENSION injection_points)); +$node->safe_psql( + 'postgres', + q(CREATE TABLE tab(i int PRIMARY KEY, j int); + INSERT INTO tab VALUES (1, 1), (2, 2))); + +# Wait until the given backend waits on the lock of the given transaction. +sub wait_for_xact_lock_wait +{ + my ($pid, $xid, $what) = @_; + + $node->poll_query_until('postgres', + "SELECT count(*) > 0 FROM pg_locks WHERE pid = $pid AND locktype = 'transactionid' AND transactionid = '$xid' AND NOT granted" + ) or die "$what did not wait on the lock of transaction $xid"; +} + +my $s2 = $node->background_psql('postgres'); +my $s3 = $node->background_psql('postgres'); +my $s4 = $node->background_psql('postgres'); +my $walsender = $node->background_psql('postgres', replication => 'database'); + +my $walsender_pid = $walsender->query_safe('SELECT pg_backend_pid()'); +my $s4_pid = $s4->query_safe('SELECT pg_backend_pid()'); + +# s2 and s3 hold transactions with an XID, so that the snapshot builder goes +# through its BUILDING_SNAPSHOT and FULL_SNAPSHOT states rather than becoming +# consistent right away. The walsender waits for s2, then for s3. +my $s2_xid = $s2->query_safe('BEGIN; SELECT pg_current_xact_id()'); + +$walsender->query_until( + qr/started/, q(\echo started +BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ; +CREATE_REPLICATION_SLOT slot_race TEMPORARY LOGICAL test_decoding USE_SNAPSHOT; +)); +wait_for_xact_lock_wait($walsender_pid, $s2_xid, 'walsender'); + +my $s3_xid = $s3->query_safe('BEGIN; SELECT pg_current_xact_id()'); +$s2->query_safe('ROLLBACK'); +wait_for_xact_lock_wait($walsender_pid, $s3_xid, 'walsender'); + +# While the walsender waits for s3, s4 changes the table and stops after +# writing its commit record, before updating CLOG. +$s4->query_safe( + q(SELECT injection_points_set_local(); + SELECT injection_points_attach('commit-before-clog-update', 'wait'))); +$s4->query_until( + qr/started/, q(\echo started +BEGIN; +INSERT INTO tab VALUES (3, 3); +UPDATE tab SET j = j + 1 WHERE i = 1; +DELETE FROM tab WHERE i = 2; +COMMIT; +)); +$node->poll_query_until('postgres', + "SELECT wait_event = 'commit-before-clog-update' FROM pg_stat_activity WHERE pid = $s4_pid" +) or die "s4 did not reach the injection point"; +my $s4_xid = $node->safe_psql('postgres', + "SELECT backend_xid FROM pg_stat_activity WHERE pid = $s4_pid"); + +# Now the walsender decodes s4's commit record and reaches a consistent +# state. It must wait for s4 rather than build the snapshot. +$s3->query_safe('ROLLBACK'); +my $state; +for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++) +{ + $state = $node->safe_psql( + 'postgres', + "SELECT CASE WHEN a.state = 'idle in transaction' THEN 'slot created' + WHEN l.pid IS NOT NULL THEN 'waiting for s4' END + FROM pg_stat_activity a + LEFT JOIN pg_locks l ON l.pid = a.pid AND l.locktype = 'transactionid' + AND l.transactionid = '$s4_xid' AND NOT l.granted + WHERE a.pid = $walsender_pid"); + last if $state ne ''; + usleep(100_000); +} +is( $state, + 'waiting for s4', + 'slot creation waits for the transaction that has not updated CLOG'); + +# If the slot got created without waiting, use its snapshot right away: +# the scan takes s4 as aborted and sets hint bits accordingly, which is +# what the last two checks then report. +if ($state eq 'slot created') +{ + $walsender->query_until(qr/test_decoding/, ''); + diag("rows seen through the slot's snapshot before s4 updated CLOG: " + . $walsender->query_safe('SELECT i, j FROM tab ORDER BY i')); +} + +$node->safe_psql('postgres', + "SELECT injection_points_wakeup('commit-before-clog-update')"); +$s4->quit; + +$node->poll_query_until('postgres', + "SELECT state = 'idle in transaction' FROM pg_stat_activity WHERE pid = $walsender_pid" +) or die "slot creation did not finish"; +# drain the result of CREATE_REPLICATION_SLOT +$walsender->query_until(qr/test_decoding/, '') + if $state ne 'slot created'; + +is($walsender->query_safe('SELECT i, j FROM tab ORDER BY i'), + "1|2\n3|3", "the slot's snapshot sees the transaction's changes"); +is($node->safe_psql('postgres', 'SELECT i, j FROM tab ORDER BY i'), + "1|2\n3|3", "a new session sees the transaction's changes"); + +is( $walsender->query_safe( + q(WITH s AS (SELECT pg_current_snapshot() AS snap) + SELECT count(*) FROM s, LATERAL pg_snapshot_xip(snap) AS x(xid) + WHERE xid < pg_snapshot_xmin(snap) OR xid >= pg_snapshot_xmax(snap)) + ), + '0', + 'snapshot XIDs are all within xmin and xmax'); + +$walsender->query_safe('ROLLBACK'); +$walsender->quit; +$s2->quit; +$s3->quit; +$node->stop; + +done_testing();