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();