database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2026-10-23 13:29:09 +03:00
Code Activity

Wait for transactions of an initial decoding snapshot to commit

Browse Source 
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 504fe10d1d, "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 <mihailnikalayeu@gmail.com>
Author: Antonin Houska <ah@cybertec.at>
Author: Rui Zhao <zhaorui126@gmail.com>
Backpatch-through: 14
Discussion: https://postgr.es/m/85833.1768840165@localhost
Discussion: https://postgr.es/m/CADzfLwU78as45To9a%3D-Qkr5jEg3tMxc5rUtdKy2MTv4r_SDGng%40mail.gmail.com
This commit is contained in:
Álvaro Herrera
2026-09-24 11:52:47 +02:00
parent 3ee0edb916
commit 46024c573b
9 changed files with 367 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/backend/access/transam/xact.c
View File

@@ -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
*/

39
src/backend/replication/logical/snapbuild.c
View File

@@ -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

1
src/test/modules/injection_points/Makefile
View File

@@ -18,6 +18,7 @@ ISOLATION = basic \
inplace \
reindex_concurrently_deferred \
repack \
repack_commit_race \
repack_decode \
repack_temporal \
repack_temporal_multirange \

64
src/test/modules/injection_points/expected/repack_commit_race.out Normal file
View File

@@ -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)

11
src/test/modules/injection_points/injection_points.c
View File

@@ -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

1
src/test/modules/injection_points/meson.build
View File

@@ -47,6 +47,7 @@ tests += {
'inplace',
'reindex_concurrently_deferred',
'repack',
'repack_commit_race',
'repack_decode',
'repack_temporal',
'repack_temporal_multirange',

95
src/test/modules/injection_points/specs/repack_commit_race.spec Normal file
View File

@@ -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

1
src/test/recovery/meson.build
View File

@@ -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',
],
},
}

148
src/test/recovery/t/057_snapshot_commit_race.pl Normal file
View File

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