On 2026-Sep-21, Alvaro Herrera wrote: > Yeah, maybe this approach isn't great after all. We could turn that > around and search for each loop around the snap->xmin..snap->xmax loop > that is found in snap->xip in the running->xids array. That reduces the > number of times we go through XactLockTableWait() to only running > transactions (same as in Rui's original patch [1]). However, the > running->xids array is not sorted, so we would have to qsort() it, or do > a plain array walk for each element.
BTW just to make it clear what I'm talking about, attached is the patch for this approach, which I'm now thinking to throw away in favor of Rui's earlier version. (The test contains the change Rui suggested yesterday, and it's passing for me with this patch.) I didn't touch the commit message. -- Álvaro Herrera 48°01'N 7°57'E — https://www.EnterpriseDB.com/ "Las cosas son buenas o malas segun las hace nuestra opinión" (Lisias)
>From 91f4c1fda343562f555fc473f925966b97dc7b12 Mon Sep 17 00:00:00 2001 From: Rui Zhao <[email protected]> Date: Sun, 13 Sep 2026 00:00:27 +0800 Subject: [PATCH v6 1/3] Wait for the transactions of an initial decoding snapshot to finish SnapBuildInitialSnapshot() converts the snapshot builder's list of committed transactions into a regular MVCC snapshot, which is then used with HeapTupleSatisfiesMVCC(). That function consults CLOG about the transactions the snapshot takes as not running, so each of them has to have finished committing before the snapshot is handed out: the commit record is written first, CLOG is updated afterwards, and the transaction stays in the procarray until after that. Read the set of running transactions once, and wait on the transaction lock of those that are in the snapshot's list, as SnapBuildWaitSnapshot() does in the same code path; the others have left the procarray and so have updated CLOG. Historic snapshots built by SnapBuildBuildSnapshot() need no such wait: they rely on the xip array for transactions between xmin and xmax, and consult CLOG only for transactions below xmin, which had left the procarray when the xl_running_xacts record that set xmin was written. --- src/backend/replication/logical/snapbuild.c | 33 +++++++++++++++++++++ src/backend/storage/ipc/procarray.c | 4 +-- 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c index de491ea0c4b..5c51c65bce4 100644 --- a/src/backend/replication/logical/snapbuild.c +++ b/src/backend/replication/logical/snapbuild.c @@ -448,6 +448,7 @@ SnapBuildInitialSnapshot(SnapBuild *builder) TransactionId safeXid; TransactionId *newxip; int newxcnt = 0; + RunningTransactions running = NULL; Assert(XactIsoLevel == XACT_REPEATABLE_READ); Assert(builder->building_full_snapshot); @@ -493,6 +494,17 @@ SnapBuildInitialSnapshot(SnapBuild *builder) /* allocate in transaction context */ newxip = palloc_array(TransactionId, GetMaxSnapshotXidCount()); + /* + * Avoid excessive traffic through TransactionIdIsInProgress() below by + * acquiring the list of running transactions once. + */ + if (!RecoveryInProgress()) + { + running = GetRunningTransactionData(); + LWLockRelease(XidGenLock); + LWLockRelease(ProcArrayLock); + } + /* * snapbuild.c builds transactions in an "inverted" manner, which means it * stores committed transactions in ->xip, not ones in progress. Build a @@ -520,6 +532,27 @@ SnapBuildInitialSnapshot(SnapBuild *builder) newxip[newxcnt++] = xid; } + /* + * If a transaction is in the snapshot and reported as committed, we + * don't yet know for certain that the transaction was removed from + * procarray as opposed to merely got its WAL commit record written. + * For correctness reasons (involving hint-bit setting) we must not + * allow transactions in the latter state be reported as committed, so + * wait for them to end. + */ + if (!RecoveryInProgress() && test != NULL) + { + /* XXX We could qsort() and bsearch() this array ... */ + for (int i = 0; i < running->xcnt; i++) + { + if (xid == running->xids[i]) + { + XactLockTableWait(xid, NULL, NULL, XLTW_None); + break; + } + } + } + TransactionIdAdvance(xid); } diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index b7e03134ed8..fd678bea3b6 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -2663,8 +2663,8 @@ GetRunningTransactionData(void) * the lock, so we can't look at numProcs. Likewise, we allocate much * more subxip storage than is probably needed. * - * Should only be allocated in bgwriter, since only ever executed during - * checkpoints. + * This is only called during checkpoint and during initial logical + * decoding snapshot build, so the impact is limited. */ if (CurrentRunningXacts->xids == NULL) { -- 2.47.3
>From e5a4b42698e064fa7d0afb1e3346d619fdbaccd1 Mon Sep 17 00:00:00 2001 From: Rui Zhao <[email protected]> Date: Sun, 13 Sep 2026 00:59:09 +0800 Subject: [PATCH v6 2/3] Test the initial decoding snapshot against a commit that is not in CLOG yet The snapshot builder counts a transaction as committed once it has decoded its commit record, but the transaction updates CLOG only after writing that record. An initial snapshot built in between and converted to a regular MVCC snapshot makes HeapTupleSatisfiesMVCC() consult CLOG about a transaction it takes as not running, and the transaction comes out as aborted. Add an injection point between the flush of the commit record and the CLOG update, and an isolation test that stops a transaction there while REPACK (CONCURRENTLY) builds its snapshot. Without a fix the repacked table lacks the changes of that transaction. --- src/backend/access/transam/xact.c | 10 ++ 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 | 96 +++++++++++++++++++ 6 files changed, 180 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 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/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; + <waiting ...> +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; + <waiting ...> +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..a961d1c001d --- /dev/null +++ b/src/test/modules/injection_points/specs/repack_commit_race.spec @@ -0,0 +1,96 @@ +# REPACK (CONCURRENTLY) takes its initial snapshot from the logical decoding +# snapshot builder, which counts a transaction as committed as soon as it has +# decoded its commit record. The transaction itself may still be between +# writing that record and updating CLOG. The snapshot must not be used before +# the transaction has finished committing, or the copy of the table takes it +# as aborted and its changes are lost: decoding starts after its commit record. +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 -- 2.47.3
>From b445a316b9b1d34018d63af4f3ab8b1c38d2ff64 Mon Sep 17 00:00:00 2001 From: Rui Zhao <[email protected]> Date: Sat, 12 Sep 2026 02:26:17 +0800 Subject: [PATCH v6 3/3] Test slot creation with USE_SNAPSHOT against a commit that is not in CLOG yet Same scenario as the REPACK (CONCURRENTLY) isolation test, through the replication protocol: CREATE_REPLICATION_SLOT ... USE_SNAPSHOT has to wait for a transaction whose commit record it has decoded but which has not updated CLOG yet, and the snapshot and later sessions must see that transaction's changes. --- .../recovery/t/057_snapshot_commit_race.pl | 149 ++++++++++++++++++ 1 file changed, 149 insertions(+) create mode 100644 src/test/recovery/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..f2e8c7b5e88 --- /dev/null +++ b/src/test/recovery/t/057_snapshot_commit_race.pl @@ -0,0 +1,149 @@ +# 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 updating CLOG. 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); + +if ($ENV{enable_injection_points} ne 'yes') +{ + plan skip_all => 'Injection points not supported by this build'; +} + +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(); -- 2.47.3
