Hello Mikhail, Amit, everyone,
The commitfest entry is tagged "Help - Stuck Rebasing", so I started
there. I ended up with a rebase, a negative control for the tests,
and one behaviour change that I did not expect and that I think is
worth deciding on before this moves forward.
Everything below was measured on master 09a579abaca, with clean builds
in fresh directories (--enable-cassert --enable-injection-points).
Scripts and logs are attached.
1. Why it stopped applying
--------------------------
Not your fault, and not a trivial conflict:
* v18-0001 attaches its injection point inside index_getnext_slot(),
and that function is gone. ddce1da5b1b ("Add slot-based table AM
index scan interface", Sep 15) removed index_getnext_tid(),
index_fetch_heap() and index_getnext_slot() from indexam.c.
* v18-0002 fails for a much smaller reason: index_beginscan() grew a
bool argument in dcd8cc1c852.
Attached v19-0001/0002 is your patch rebased, with your authorship
untouched. The injection point now lives in
heapam_index_getnext_slot() (heapam_indexscan.c), at the same point of
the scan as before: right after the index AM returned a TID and before
the heap fetch. The four TAP tests needed no changes at all, which is
a good sign for where the point was placed.
Two things I dropped while rebasing, both unrelated to the fix:
* v18-0002 contained a stray hunk in catalog/index.c that only turns
one space into two inside index_concurrently_swap().
* v18-0001 adds INJECTION_POINT("check_exclusion_or_unique_constraint
_no_conflict") in execIndexing.c, three lines below the equivalent
point master already has from bc32a12e0db ("check-exclusion-or-
unique-constraint-no-conflict", with dashes), and no test in the
patch uses the new name.
2. The tests do prove something (negative control)
--------------------------------------------------
I ran the three deterministic TAP tests against a build that has
0001 (tests + injection points) but NOT 0002 (the fix):
without the fix: 3 files, 11 assertions, 9 failed
with the fix: 3 files, 11 assertions, all passed
So the tests fail for the reason they claim to, and the rebase did not
quietly defeat them.
Full suite with the fix, same commit: make check, isolation, recovery
and the whole src/test/subscription set all pass.
3. What the MVCC scan gives up: the wait
-----------------------------------------
This is the part I would like your opinion on, and Amit's.
SnapshotDirty sees rows written by transactions that are still in
flight, and that is why the current code takes snap.xmin/snap.xmax and
calls XactLockTableWait(): it waits for the inserter and then decides.
A fresh MVCC snapshot cannot see that row, so there is nobody to wait
for, and the caller is told the row does not exist.
That is observable without any injection point. The subscriber starts
empty (copy_data = false), a local session has an INSERT of the same
key open, and only then does the UPDATE arrive from the publisher:
sub: BEGIN; INSERT INTO t VALUES (1, 'fromsub'); -- left open
pub: UPDATE t SET data = 'frompubnew' WHERE a = 1;
sub: COMMIT;
Measured 5 runs per tree, identical every time:
master, apply worker waiting on a lock: yes
master, conflict reported: none
master, final row: frompubnew
v19, apply worker waiting on a lock: no
v19, conflict reported: update_missing
v19, final row: fromsub
So in this case master applies the publisher's UPDATE and the patch
loses it. To be fair to master, it is not blindly applying: if the
local session does ROLLBACK instead of COMMIT, master waits, finds
nothing, and reports the row as missing, exactly like the patch does.
It waits and then decides correctly in both outcomes.
I am not claiming this is worse overall than the bug you are fixing -
the race you found is real and I reproduced it. But it is the same
symptom, a lost update on the subscriber, moved to a different case,
and it would be a shame to trade one for the other silently.
This looks like a concrete answer to the question you and Amit left
open in August about whether SnapshotDirty was giving any real
guarantee here. For an in-flight INSERT it gives one, and it is a
guarantee the conflict-detection code depends on.
4. A variant that keeps the wait
---------------------------------
Attached as a .txt (nocfbot-wait-for-inflight-inserter.diff), on top
of v19-0002. The idea is small: scan with the fresh MVCC snapshot as
your patch does, and only if that finds nothing, do one extra pass
with SnapshotDirty purely to find out whether someone is inserting the
row right now. If so, wait for that transaction and start over with a
new MVCC snapshot.
The tuple found by the dirty pass is never returned to the caller, so
it cannot bring back the concurrent-update race that 0002 fixes: the
worst case is that the dirty scan misses the inserter and we fall back
to exactly the behaviour of v19.
Measured on the same commit:
in-flight INSERT test, 5 runs: same result as master every time
(waits, applies, final row frompubnew)
ROLLBACK variant: same as master
039/040/041 (your tests): all pass
make check, isolation: pass
src/test/subscription: 43 files, 619 tests, pass
recovery: pass
I did not benchmark the extra pass. It only runs when the MVCC scan
found nothing, which for a healthy subscriber is the path that is
already about to log a conflict, so I would expect it not to matter,
but I have not measured it and I would rather say so than guess.
Take it or leave it - it is your patch and you may well prefer to
handle the in-flight case some other way, or to argue that losing the
wait is acceptable. I mainly wanted the trade-off to be visible.
5. The test I used
------------------
Attached as 090_insert_inflight.pl.txt. It is not proposed for
commit as it is: it prints its measurement by failing a comparison,
which is handy for a review and wrong for the tree. If you find the
case worth covering I am happy to turn it into a proper test.
Regards,
Manu
>From 9b22c6b3036e42d0565cf8d0abd6bb47db56f874 Mon Sep 17 00:00:00 2001
From: nkey <[email protected]>
Date: Tue, 22 Sep 2026 16:41:35 -0300
Subject: [PATCH v19 1/2] This patch introduces new injection points and TAP
tests to reproduce and verify conflict detection issues that arise during
SNAPSHOT_DIRTY index scans in logical replication.
Rebased on top of ddce1da5b1b ("Add slot-based table AM index scan interface"):
index_getnext_slot() is gone, so the injection point moved to
heapam_index_getnext_slot() in heapam_indexscan.c, at the same place in the
scan (right after the index AM returned a TID, before the heap fetch).
---
src/backend/access/heap/heapam_indexscan.c | 10 ++
src/backend/access/nbtree/README | 9 ++
src/backend/executor/execIndexing.c | 7 +-
src/backend/replication/logical/worker.c | 4 +
src/include/utils/snapshot.h | 14 ++
src/test/subscription/meson.build | 4 +
.../subscription/t/039_delete_missing_race.pl | 139 +++++++++++++++++
.../subscription/t/040_update_missing_race.pl | 141 +++++++++++++++++
.../t/041_update_missing_with_retain.pl | 143 ++++++++++++++++++
.../t/042_update_missing_simulation.pl | 125 +++++++++++++++
10 files changed, 595 insertions(+), 1 deletion(-)
create mode 100644 src/test/subscription/t/039_delete_missing_race.pl
create mode 100644 src/test/subscription/t/040_update_missing_race.pl
create mode 100644 src/test/subscription/t/041_update_missing_with_retain.pl
create mode 100644 src/test/subscription/t/042_update_missing_simulation.pl
diff --git a/src/backend/access/heap/heapam_indexscan.c
b/src/backend/access/heap/heapam_indexscan.c
index 0ae028ecd41..1ea3bc35adf 100644
--- a/src/backend/access/heap/heapam_indexscan.c
+++ b/src/backend/access/heap/heapam_indexscan.c
@@ -19,8 +19,11 @@
#include "access/relscan.h"
#include "access/tableam_indexscan.h"
#include "access/visibilitymap.h"
+#include "catalog/catalog.h"
#include "pgstat.h"
+#include "replication/logicalworker.h"
#include "storage/predicate.h"
+#include "utils/injection_point.h"
static bool heapam_index_plain_tuple_getnext_slot(IndexScanDesc scan,
@@ -346,6 +349,13 @@ heapam_index_getnext_slot(IndexScanDesc scan,
ScanDirection direction,
/* The scan's next TID was set in scan->xs_heaptid for us */
Assert(ItemPointerIsValid(&scan->xs_heaptid));
+#ifdef USE_INJECTION_POINTS
+ if (!IsCatalogRelation(scan->heapRelation) && IsLogicalWorker())
+ {
+
INJECTION_POINT("index_getnext_slot_before_fetch_apply_dirty", NULL);
+ }
+#endif
+
hscan = (IndexScanHeapData *) scan->xs_table_opaque;
if (!index_only)
diff --git a/src/backend/access/nbtree/README b/src/backend/access/nbtree/README
index cb921ca2ef6..c95b69ef26e 100644
--- a/src/backend/access/nbtree/README
+++ b/src/backend/access/nbtree/README
@@ -103,6 +103,15 @@ We also remember the left-link, and follow it when the
scan moves backwards
(though this requires extra handling to account for concurrent splits of
the left sibling; see detailed move-left algorithm below).
+Despite the described mechanics in place, inconsistent results may still occur
+during non-MVCC scans (SnapshotDirty and SnapshotSelf). This issue can occur
if a
+concurrent transaction deletes a tuple and inserts a new tuple with a new TID
in the
+same page or to the left/right (depending on scan direction) of current scan
position.
+If the scan has already visited the page and cached its content in the
+backend-local storage, it might skip the old tuple due to deletion and miss
the new
+tuple because the scan does not re-read the page. Note it affects not only
btree
+scan but also a heap scan.
+
In most cases we release our lock and pin on a page before attempting
to acquire pin and lock on the page we are moving to. In a few places
it is necessary to lock the next page before releasing the current one.
diff --git a/src/backend/executor/execIndexing.c
b/src/backend/executor/execIndexing.c
index d0a714a42f2..47283c96348 100644
--- a/src/backend/executor/execIndexing.c
+++ b/src/backend/executor/execIndexing.c
@@ -119,6 +119,7 @@
#include "utils/multirangetypes.h"
#include "utils/rangetypes.h"
#include "utils/snapmgr.h"
+#include "utils/injection_point.h"
/* waitMode argument to check_exclusion_or_unique_constraint() */
typedef enum
@@ -788,7 +789,9 @@ check_exclusion_or_unique_constraint(Relation heap,
Relation index,
/*
* Search the tuples that are in the index for any violations, including
* tuples that aren't visible yet.
- */
+ * Snapshot dirty may miss some tuples in the case of parallel updates,
+ * but it is acceptable here.
+ */
InitDirtySnapshot(DirtySnapshot);
for (i = 0; i < indnkeyatts; i++)
@@ -978,6 +981,8 @@ retry:
INJECTION_POINT("check-exclusion-or-unique-constraint-no-conflict", NULL);
#endif
+ if (!conflict)
+
INJECTION_POINT("check_exclusion_or_unique_constraint_no_conflict", NULL);
return !conflict;
}
diff --git a/src/backend/replication/logical/worker.c
b/src/backend/replication/logical/worker.c
index 7781bb1c168..95db69b4b66 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -290,6 +290,7 @@
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/guc.h"
+#include "utils/injection_point.h"
#include "utils/inval.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
@@ -3033,7 +3034,10 @@ apply_handle_update_internal(ApplyExecutionData *edata,
conflicttuple.origin != replorigin_xact_state.origin)
type = CT_UPDATE_DELETED;
else
+ {
+
INJECTION_POINT("apply_handle_update_internal_update_missing", NULL);
type = CT_UPDATE_MISSING;
+ }
/* Store the new tuple for conflict reporting */
slot_store_data(newslot, relmapentry, newtup);
diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h
index 9766aabcad4..2fc0058796a 100644
--- a/src/include/utils/snapshot.h
+++ b/src/include/utils/snapshot.h
@@ -53,6 +53,13 @@ typedef enum SnapshotType
* - previous commands of this transaction
* - changes made by the current command
*
+ * Note: such a snapshot may miss an existing logical tuple in case of
+ * parallel update.
+ * If a new version of a tuple is inserted into an already processed
page
+ * but the old one marked with committed xmax - snapshot will skip the
old
+ * one and never meet the new one during that scan - resulting in
skipping
+ * that tuple at all.
+ *
* Does _not_ include:
* - in-progress transactions (as of the current instant)
*
-------------------------------------------------------------------------
@@ -82,6 +89,13 @@ typedef enum SnapshotType
* transaction and committed/aborted xacts are concerned. However, it
* also includes the effects of other xacts still in progress.
*
+ * Note: such a snapshot may miss an existing logical tuple in case of
+ * parallel update.
+ * If a new version of a tuple is inserted into an already processed
page but the
+ * old one marked with committed/in-progress xmax - snapshot will skip
the old one
+ * and never meet the new one during that scan - resulting in skipping
that tuple
+ * at all.
+ *
* A special hack is that when a snapshot of this type is used to
* determine tuple visibility, the passed-in snapshot struct is used as
an
* output argument to return the xids of concurrent xacts that affected
diff --git a/src/test/subscription/meson.build
b/src/test/subscription/meson.build
index e71e95c6297..ae0b19b08c5 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -48,6 +48,10 @@ tests += {
't/036_sequences.pl',
't/037_except.pl',
't/038_walsnd_shutdown_timeout.pl',
+ 't/039_delete_missing_race.pl',
+ 't/040_update_missing_race.pl',
+ 't/041_update_missing_with_retain.pl',
+ 't/042_update_missing_simulation.pl',
't/100_bugs.pl',
],
},
diff --git a/src/test/subscription/t/039_delete_missing_race.pl
b/src/test/subscription/t/039_delete_missing_race.pl
new file mode 100644
index 00000000000..51dd351dc10
--- /dev/null
+++ b/src/test/subscription/t/039_delete_missing_race.pl
@@ -0,0 +1,139 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+# Test the conflict detection and resolution in logical replication
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+ plan skip_all => 'Injection points not supported by this build';
+}
+
+############################## Set it to 0 to make set success; TODO: delete
that for commit
+my $simulate_race_condition = 1;
+##############################
+
+###############################
+# Setup
+###############################
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+ qq(track_commit_timestamp = on));
+$node_publisher->start;
+
+
+# Create subscriber node with track_commit_timestamp enabled
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+ qq(track_commit_timestamp = on));
+$node_subscriber->start;
+
+
+# Check if the extension injection_points is available, as it may be
+# possible that this script is run with installcheck, where the module
+# would not be installed by default.
+if (!$node_subscriber->check_extension('injection_points'))
+{
+ plan skip_all => 'Extension injection_points not installed';
+}
+
+# Create table on publisher
+$node_publisher->safe_psql(
+ 'postgres',
+ "CREATE TABLE conf_tab(a int PRIMARY key, data text);");
+
+# Create similar table on subscriber with additional index to disable HOT
updates
+$node_subscriber->safe_psql(
+ 'postgres',
+ "CREATE TABLE conf_tab(a int PRIMARY key, data text);
+ CREATE INDEX data_index ON conf_tab(data);");
+
+# Set up extension to simulate race condition
+$node_subscriber->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub FOR TABLE conf_tab");
+
+# Insert row to be updated later
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO conf_tab(a, data) VALUES (1,'frompub')");
+
+# Create the subscription
+my $appname = 'tap_sub';
+$node_subscriber->safe_psql(
+ 'postgres',
+ "CREATE SUBSCRIPTION tap_sub
+ CONNECTION '$publisher_connstr application_name=$appname'
+ PUBLICATION tap_pub");
+
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+############################################
+# Race condition because of DirtySnapshot
+############################################
+
+my $psql_session_subscriber = $node_subscriber->background_psql('postgres');
+if ($simulate_race_condition)
+{
+ $node_subscriber->safe_psql('postgres',
+ "SELECT
injection_points_attach('index_getnext_slot_before_fetch_apply_dirty',
'wait')");
+}
+
+my $log_offset = -s $node_subscriber->logfile;
+
+# Delete tuple on publisher
+$node_publisher->safe_psql('postgres', "DELETE FROM conf_tab WHERE a=1;");
+
+if ($simulate_race_condition)
+{
+ # Wait apply worker to start the search for the tuple using index
+ $node_subscriber->wait_for_event('logical replication apply worker',
+ 'index_getnext_slot_before_fetch_apply_dirty');
+}
+
+# Updater tuple on subscriber
+$psql_session_subscriber->query_until(
+ qr/start/, qq[
+ \\echo start
+ UPDATE conf_tab SET data = 'fromsubnew' WHERE (a=1);
+]);
+
+
+if ($simulate_race_condition)
+{
+ # Wake up apply worker
+ $node_subscriber->safe_psql('postgres',"
+ SELECT
injection_points_detach('index_getnext_slot_before_fetch_apply_dirty');
+ SELECT
injection_points_wakeup('index_getnext_slot_before_fetch_apply_dirty');
+ ");
+}
+
+# Tuple was updated - so, we have conflict
+$node_subscriber->wait_for_log(
+ qr/conflict detected on relation \"public.conf_tab\"/,
+ $log_offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# But tuple should be deleted on subscriber any way
+is($node_subscriber->safe_psql('postgres', 'SELECT count(*) from conf_tab'),
0, 'record deleted on subscriber');
+
+ok(!$node_subscriber->log_contains(
+ qr/LOG: conflict detected on relation \"public.conf_tab\":
conflict=delete_missing/,
+ $log_offset), 'invalid conflict detected');
+
+ok($node_subscriber->log_contains(
+ qr/LOG: conflict detected on relation \"public.conf_tab\":
conflict=delete_origin_differs/,
+ $log_offset), 'correct conflict detected');
+
+done_testing();
diff --git a/src/test/subscription/t/040_update_missing_race.pl
b/src/test/subscription/t/040_update_missing_race.pl
new file mode 100644
index 00000000000..1e120f74bbd
--- /dev/null
+++ b/src/test/subscription/t/040_update_missing_race.pl
@@ -0,0 +1,141 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+# Test the conflict detection and resolution in logical replication
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+ plan skip_all => 'Injection points not supported by this build';
+}
+
+############################## Set it to 0 to make set success; TODO: delete
that for commit
+my $simulate_race_condition = 1;
+##############################
+
+###############################
+# Setup
+###############################
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+ qq(track_commit_timestamp = on));
+$node_publisher->start;
+
+
+# Create subscriber node with track_commit_timestamp enabled
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+ qq(track_commit_timestamp = on));
+$node_subscriber->start;
+
+
+# Check if the extension injection_points is available, as it may be
+# possible that this script is run with installcheck, where the module
+# would not be installed by default.
+if (!$node_subscriber->check_extension('injection_points'))
+{
+ plan skip_all => 'Extension injection_points not installed';
+}
+
+# Create table on publisher
+$node_publisher->safe_psql(
+ 'postgres',
+ "CREATE TABLE conf_tab(a int PRIMARY key, data text);");
+
+# Create similar table on subscriber with additional index to disable HOT
updates and additional column
+$node_subscriber->safe_psql(
+ 'postgres',
+ "CREATE TABLE conf_tab(a int PRIMARY key, data text, i int DEFAULT 0);
+ CREATE INDEX i_index ON conf_tab(i);");
+
+# Set up extension to simulate race condition
+$node_subscriber->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub FOR TABLE conf_tab");
+
+# Insert row to be updated later
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO conf_tab(a, data) VALUES (1,'frompub')");
+
+# Create the subscription
+my $appname = 'tap_sub';
+$node_subscriber->safe_psql(
+ 'postgres',
+ "CREATE SUBSCRIPTION tap_sub
+ CONNECTION '$publisher_connstr application_name=$appname'
+ PUBLICATION tap_pub");
+
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+############################################
+# Race condition because of DirtySnapshot
+############################################
+
+my $psql_session_subscriber = $node_subscriber->background_psql('postgres');
+if ($simulate_race_condition)
+{
+ $node_subscriber->safe_psql('postgres', "SELECT
injection_points_attach('index_getnext_slot_before_fetch_apply_dirty',
'wait')");
+}
+
+my $log_offset = -s $node_subscriber->logfile;
+
+# Update tuple on publisher
+$node_publisher->safe_psql('postgres',
+ "UPDATE conf_tab SET data = 'frompubnew' WHERE (a=1);");
+
+
+if ($simulate_race_condition)
+{
+ # Wait apply worker to start the search for the tuple using index
+ $node_subscriber->wait_for_event('logical replication apply worker',
'index_getnext_slot_before_fetch_apply_dirty');
+}
+
+# Update additional(!) column on the subscriber
+$psql_session_subscriber->query_until(
+ qr/start/, qq[
+ \\echo start
+ UPDATE conf_tab SET i = 1 WHERE (a=1);
+]);
+
+
+if ($simulate_race_condition)
+{
+ # Wake up apply worker
+ $node_subscriber->safe_psql('postgres',"
+ SELECT
injection_points_detach('index_getnext_slot_before_fetch_apply_dirty');
+ SELECT
injection_points_wakeup('index_getnext_slot_before_fetch_apply_dirty');
+ ");
+}
+
+# Tuple was updated - so, we have conflict
+$node_subscriber->wait_for_log(
+ qr/conflict detected on relation \"public.conf_tab\"/,
+ $log_offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# We need new column value be synced with subscriber
+is($node_subscriber->safe_psql('postgres', 'SELECT data from conf_tab WHERE a
= 1'), 'frompubnew', 'record updated on subscriber');
+# And additional column maintain updated value
+is($node_subscriber->safe_psql('postgres', 'SELECT i from conf_tab WHERE a =
1'), 1, 'column record updated on subscriber');
+
+ok(!$node_subscriber->log_contains(
+ qr/LOG: conflict detected on relation \"public.conf_tab\":
conflict=update_missing/,
+ $log_offset), 'invalid conflict detected');
+
+ok($node_subscriber->log_contains(
+ qr/LOG: conflict detected on relation \"public.conf_tab\":
conflict=update_origin_differs/,
+ $log_offset), 'correct conflict detected');
+
+done_testing();
diff --git a/src/test/subscription/t/041_update_missing_with_retain.pl
b/src/test/subscription/t/041_update_missing_with_retain.pl
new file mode 100644
index 00000000000..7b225d45f7f
--- /dev/null
+++ b/src/test/subscription/t/041_update_missing_with_retain.pl
@@ -0,0 +1,143 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+# Test the conflict detection and resolution in logical replication
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+ plan skip_all => 'Injection points not supported by this build';
+}
+
+############################## Set it to 0 to make set success; TODO: delete
that for commit
+my $simulate_race_condition = 1;
+##############################
+
+###############################
+# Setup
+###############################
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+ qq(track_commit_timestamp = on));
+$node_publisher->start;
+
+
+# Create subscriber node with track_commit_timestamp enabled
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+ qq(track_commit_timestamp = on));
+$node_subscriber->append_conf('postgresql.conf',
+ qq(wal_level = 'replica'));
+$node_subscriber->start;
+
+
+# Check if the extension injection_points is available, as it may be
+# possible that this script is run with installcheck, where the module
+# would not be installed by default.
+if (!$node_subscriber->check_extension('injection_points'))
+{
+ plan skip_all => 'Extension injection_points not installed';
+}
+
+# Create table on publisher
+$node_publisher->safe_psql(
+ 'postgres',
+ "CREATE TABLE conf_tab(a int PRIMARY key, data text);");
+
+# Create similar table on subscriber with additional index to disable HOT
updates and additional column
+$node_subscriber->safe_psql(
+ 'postgres',
+ "CREATE TABLE conf_tab(a int PRIMARY key, data text, i int DEFAULT 0);
+ CREATE INDEX i_index ON conf_tab(i);");
+
+# Set up extension to simulate race condition
+$node_subscriber->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub FOR TABLE conf_tab");
+
+# Insert row to be updated later
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO conf_tab(a, data) VALUES (1,'frompub')");
+
+# Create the subscription
+my $appname = 'tap_sub';
+$node_subscriber->safe_psql(
+ 'postgres',
+ "CREATE SUBSCRIPTION tap_sub
+ CONNECTION '$publisher_connstr application_name=$appname'
+ PUBLICATION tap_pub WITH (retain_dead_tuples = true)");
+
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+############################################
+# Race condition because of DirtySnapshot
+############################################
+
+my $psql_session_subscriber = $node_subscriber->background_psql('postgres');
+if ($simulate_race_condition)
+{
+ $node_subscriber->safe_psql('postgres', "SELECT
injection_points_attach('index_getnext_slot_before_fetch_apply_dirty',
'wait')");
+}
+
+my $log_offset = -s $node_subscriber->logfile;
+
+# Update tuple on publisher
+$node_publisher->safe_psql('postgres',
+ "UPDATE conf_tab SET data = 'frompubnew' WHERE (a=1);");
+
+
+if ($simulate_race_condition)
+{
+ # Wait apply worker to start the search for the tuple using index
+ $node_subscriber->wait_for_event('logical replication apply worker',
'index_getnext_slot_before_fetch_apply_dirty');
+}
+
+# Update additional(!) column on the subscriber
+$psql_session_subscriber->query_until(
+ qr/start/, qq[
+ \\echo start
+ UPDATE conf_tab SET i = 1 WHERE (a=1);
+]);
+
+
+if ($simulate_race_condition)
+{
+ # Wake up apply worker
+ $node_subscriber->safe_psql('postgres',"
+ SELECT
injection_points_detach('index_getnext_slot_before_fetch_apply_dirty');
+ SELECT
injection_points_wakeup('index_getnext_slot_before_fetch_apply_dirty');
+ ");
+}
+
+# Tuple was updated - so, we have conflict
+$node_subscriber->wait_for_log(
+ qr/conflict detected on relation \"public.conf_tab\"/,
+ $log_offset);
+
+$node_publisher->wait_for_catchup($appname);
+
+# We need new column value be synced with subscriber
+is($node_subscriber->safe_psql('postgres', 'SELECT data from conf_tab WHERE a
= 1'), 'frompubnew', 'record updated on subscriber');
+# And additional column maintain updated value
+is($node_subscriber->safe_psql('postgres', 'SELECT i from conf_tab WHERE a =
1'), 1, 'column record updated on subscriber');
+
+ok(!$node_subscriber->log_contains(
+ qr/LOG: conflict detected on relation \"public.conf_tab\":
conflict=update_deleted/,
+ $log_offset), 'invalid conflict detected');
+
+ok($node_subscriber->log_contains(
+ qr/LOG: conflict detected on relation \"public.conf_tab\":
conflict=update_origin_differs/,
+ $log_offset), 'correct conflict detected');
+
+done_testing();
diff --git a/src/test/subscription/t/042_update_missing_simulation.pl
b/src/test/subscription/t/042_update_missing_simulation.pl
new file mode 100644
index 00000000000..21fcd1ceb53
--- /dev/null
+++ b/src/test/subscription/t/042_update_missing_simulation.pl
@@ -0,0 +1,125 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+# Test the conflict detection and resolution in logical replication
+# Not intended to be committed because quite heavy
+# Here to demonstrate reproducibility with pgbench
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use IPC::Run qw(start finish);
+use Test::More;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+ plan skip_all => 'Injection points not supported by this build';
+}
+
+###############################
+# Setup
+###############################
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf('postgresql.conf',
+ qq(track_commit_timestamp = on));
+$node_publisher->start;
+
+# Create subscriber node with track_commit_timestamp enabled
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init;
+$node_subscriber->append_conf('postgresql.conf',
+ qq(track_commit_timestamp = on));
+$node_subscriber->start;
+
+# Check if the extension injection_points is available, as it may be
+# possible that this script is run with installcheck, where the module
+# would not be installed by default.
+if (!$node_subscriber->check_extension('injection_points'))
+{
+ plan skip_all => 'Extension injection_points not installed';
+}
+
+# Create table on publisher
+$node_publisher->safe_psql(
+ 'postgres',
+ "CREATE TABLE tbl(a int PRIMARY key, data_pub int);");
+
+# Create similar table on subscriber with additional index to disable HOT
updates
+$node_subscriber->safe_psql(
+ 'postgres',
+ "CREATE TABLE tbl(a int PRIMARY key, data_pub int, data_sub int default
0);
+ CREATE INDEX data_index ON tbl(data_pub);");
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub FOR TABLE tbl");
+
+# Create the subscription
+my $appname = 'tap_sub';
+$node_subscriber->safe_psql(
+ 'postgres',
+ "CREATE SUBSCRIPTION tap_sub
+ CONNECTION '$publisher_connstr application_name=$appname'
+ PUBLICATION tap_pub");
+
+my $num_rows = 10;
+my $num_updates = 10000;
+my $num_clients = 10;
+$node_publisher->safe_psql('postgres', "INSERT INTO tbl SELECT i, i * i FROM
generate_series(1,$num_rows) i");
+
+# Wait for initial table sync to finish
+$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
+
+# Prepare small pgbench scripts as files
+my $sub_sql = $node_subscriber->basedir . '/sub_update.sql';
+my $pub_sql = $node_publisher->basedir . '/pub_delete.sql';
+
+open my $fh1, '>', $sub_sql or die $!;
+print $fh1 "\\set num random(1,$num_rows)\nUPDATE tbl SET data_sub = data_sub
+ 1 WHERE a = :num;\n";
+close $fh1;
+
+open my $fh2, '>', $pub_sql or die $!;
+print $fh2 "\\set num random(1,$num_rows)\nUPDATE tbl SET data_pub = data_pub
+ 1 WHERE a = :num;\n";
+close $fh2;
+
+my @sub_cmd = (
+ 'pgbench',
+ '--no-vacuum', "--client=$num_clients", '--jobs=4', '--exit-on-abort',
"--transactions=$num_updates",
+ '-p', $node_subscriber->port, '-h', $node_subscriber->host, '-f',
$sub_sql, 'postgres'
+);
+
+my @pub_cmd = (
+ 'pgbench',
+ '--no-vacuum', "--client=$num_clients", '--jobs=4', '--exit-on-abort',
"--transactions=$num_updates",
+ '-p', $node_publisher->port, '-h', $node_publisher->host, '-f',
$pub_sql, 'postgres'
+);
+
+$node_subscriber->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
+# This should never happen
+$node_subscriber->safe_psql('postgres',
+ "SELECT
injection_points_attach('apply_handle_update_internal_update_missing',
'error')");
+my $log_offset = -s $node_subscriber->logfile;
+
+# Start both concurrently
+my ($sub_out, $sub_err, $pub_out, $pub_err) = ('', '', '', '');
+my $sub_h = start \@sub_cmd, '>', \$sub_out, '2>', \$sub_err;
+my $pub_h = start \@pub_cmd, '>', \$pub_out, '2>', \$pub_err;
+
+# Wait for completion
+finish $sub_h;
+finish $pub_h;
+
+like($sub_out, qr/actually processed/, 'subscriber pgbench completed');
+like($pub_out, qr/actually processed/, 'publisher pgbench completed');
+
+# Let subscription catch up, then check expectations
+$node_subscriber->wait_for_subscription_sync($node_publisher, 'tap_sub');
+
+ok(!$node_subscriber->log_contains(
+ qr/ERROR: error triggered for injection point
apply_handle_update_internal_update_missing/,
+ $log_offset), 'invalid conflict detected');
+
+done_testing();
--
2.55.0
>From f6d87ea9075fac207cb8eec69895f744e3486aad Mon Sep 17 00:00:00 2001
From: nkey <[email protected]>
Date: Tue, 22 Sep 2026 16:41:35 -0300
Subject: [PATCH v19 2/2] Fix logical replication conflict detection during
tuple lookup
SNAPSHOT_DIRTY scans could miss conflict detection with concurrent transactions
during logical replication.
Replace SNAPSHOT_DIRTY scan with the GetLatestSnapshot in
RelationFindReplTupleByIndex and RelationFindReplTupleSeq.
Rebased: index_beginscan() gained the new bool argument in dcd8cc1c852.
---
src/backend/executor/execReplication.c | 65 +++++++-------------------
1 file changed, 18 insertions(+), 47 deletions(-)
diff --git a/src/backend/executor/execReplication.c
b/src/backend/executor/execReplication.c
index fd9efd94737..94589cbdc88 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -187,8 +187,6 @@ RelationFindReplTupleByIndex(Relation rel, Oid idxoid,
ScanKeyData skey[INDEX_MAX_KEYS];
int skey_attoff;
IndexScanDesc scan;
- SnapshotData snap;
- TransactionId xwait;
Relation idxrel;
bool found;
TypeCacheEntry **eq = NULL;
@@ -199,18 +197,18 @@ RelationFindReplTupleByIndex(Relation rel, Oid idxoid,
isIdxSafeToSkipDuplicates = (GetRelationIdentityOrPK(rel) == idxoid);
- InitDirtySnapshot(snap);
-
/* Build scan key. */
skey_attoff = build_replindex_scan_key(skey, rel, idxrel, searchslot);
- /* Start an index scan. */
+ /* Start an index scan. SnapshotAny will be replaced below. */
scan = index_beginscan(rel, idxrel, false,
- &snap, NULL, skey_attoff, 0,
SO_NONE);
+ SnapshotAny, NULL,
skey_attoff, 0, SO_NONE);
retry:
found = false;
-
+ PushActiveSnapshot(GetLatestSnapshot());
+ /* Update the actual scan snapshot each retry */
+ scan->xs_snapshot = GetActiveSnapshot();
index_rescan(scan, skey, skey_attoff, NULL, 0);
/* Try to find the tuple */
@@ -231,19 +229,6 @@ retry:
ExecMaterializeSlot(outslot);
- xwait = TransactionIdIsValid(snap.xmin) ?
- snap.xmin : snap.xmax;
-
- /*
- * If the tuple is locked, wait for locking transaction to
finish and
- * retry.
- */
- if (TransactionIdIsValid(xwait))
- {
- XactLockTableWait(xwait, NULL, NULL, XLTW_None);
- goto retry;
- }
-
/* Found our tuple and it's not locked */
found = true;
break;
@@ -255,8 +240,6 @@ retry:
TM_FailureData tmfd;
TM_Result res;
- PushActiveSnapshot(GetLatestSnapshot());
-
res = table_tuple_lock(rel, &(outslot->tts_tid),
GetActiveSnapshot(),
outslot,
GetCurrentCommandId(false),
@@ -265,13 +248,15 @@ retry:
0 /* don't follow
updates */ ,
&tmfd);
- PopActiveSnapshot();
-
if (should_refetch_tuple(res, &tmfd))
+ {
+ PopActiveSnapshot();
goto retry;
+ }
}
index_endscan(scan);
+ PopActiveSnapshot();
/* Don't release lock until commit. */
index_close(idxrel, NoLock);
@@ -372,9 +357,7 @@ RelationFindReplTupleSeq(Relation rel, LockTupleMode
lockmode,
{
TupleTableSlot *scanslot;
TableScanDesc scan;
- SnapshotData snap;
TypeCacheEntry **eq;
- TransactionId xwait;
bool found;
TupleDesc desc PG_USED_FOR_ASSERTS_ONLY = RelationGetDescr(rel);
@@ -382,14 +365,15 @@ RelationFindReplTupleSeq(Relation rel, LockTupleMode
lockmode,
eq = palloc0_array(TypeCacheEntry *,
outslot->tts_tupleDescriptor->natts);
- /* Start a heap scan. */
- InitDirtySnapshot(snap);
- scan = table_beginscan(rel, &snap, 0, NULL,
- SO_NONE);
+ /* Start a heap scan. SnapshotAny will be replaced below. */
+ scan = table_beginscan(rel, SnapshotAny, 0, NULL, SO_NONE);
scanslot = table_slot_create(rel, NULL);
retry:
found = false;
+ PushActiveSnapshot(GetLatestSnapshot());
+ /* Update the actual scan snapshot each retry */
+ scan->rs_snapshot = GetActiveSnapshot();
table_rescan(scan, NULL);
@@ -402,19 +386,6 @@ retry:
found = true;
ExecCopySlot(outslot, scanslot);
- xwait = TransactionIdIsValid(snap.xmin) ?
- snap.xmin : snap.xmax;
-
- /*
- * If the tuple is locked, wait for locking transaction to
finish and
- * retry.
- */
- if (TransactionIdIsValid(xwait))
- {
- XactLockTableWait(xwait, NULL, NULL, XLTW_None);
- goto retry;
- }
-
/* Found our tuple and it's not locked */
break;
}
@@ -425,8 +396,6 @@ retry:
TM_FailureData tmfd;
TM_Result res;
- PushActiveSnapshot(GetLatestSnapshot());
-
res = table_tuple_lock(rel, &(outslot->tts_tid),
GetActiveSnapshot(),
outslot,
GetCurrentCommandId(false),
@@ -435,13 +404,15 @@ retry:
0 /* don't follow
updates */ ,
&tmfd);
- PopActiveSnapshot();
-
if (should_refetch_tuple(res, &tmfd))
+ {
+ PopActiveSnapshot();
goto retry;
+ }
}
table_endscan(scan);
+ PopActiveSnapshot();
ExecDropSingleTupleTableSlot(scanslot);
return found;
--
2.55.0
diff --git a/src/backend/executor/execReplication.c
b/src/backend/executor/execReplication.c
index 94589cbdc88..8871e4b2794 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -234,6 +234,51 @@ retry:
break;
}
+ /*
+ * Nothing is visible to our snapshot. Before we let the caller report
+ * the row as missing, check whether some transaction is inserting it
+ * right now: the SnapshotDirty scan this code used to do would have
+ * waited for that transaction, and conflict detection depends on that
+ * wait.
+ *
+ * This extra scan is only used to find a transaction to wait for. The
+ * tuple it returns is never handed back to the caller, so it cannot
+ * reintroduce the concurrent-update race that the MVCC scan above
fixes:
+ * after the wait we start over with a fresh MVCC snapshot.
+ */
+ if (!found)
+ {
+ SnapshotData snap;
+ TransactionId xwait = InvalidTransactionId;
+
+ InitDirtySnapshot(snap);
+ scan->xs_snapshot = &snap;
+ index_rescan(scan, skey, skey_attoff, NULL, 0);
+
+ while (table_index_getnext_slot(scan, ForwardScanDirection,
outslot))
+ {
+ if (!isIdxSafeToSkipDuplicates)
+ {
+ if (eq == NULL)
+ eq = palloc0_array(TypeCacheEntry *,
outslot->tts_tupleDescriptor->natts);
+
+ if (!tuples_equal(outslot, searchslot, eq,
NULL))
+ continue;
+ }
+
+ xwait = TransactionIdIsValid(snap.xmin) ?
+ snap.xmin : snap.xmax;
+ break;
+ }
+
+ if (TransactionIdIsValid(xwait))
+ {
+ PopActiveSnapshot();
+ XactLockTableWait(xwait, NULL, NULL, XLTW_None);
+ goto retry;
+ }
+ }
+
/* Found tuple, try to lock it in the lockmode. */
if (found)
{
@@ -390,6 +435,34 @@ retry:
break;
}
+ /* See the matching comment in RelationFindReplTupleByIndex(). */
+ if (!found)
+ {
+ SnapshotData snap;
+ TransactionId xwait = InvalidTransactionId;
+
+ InitDirtySnapshot(snap);
+ scan->rs_snapshot = &snap;
+ table_rescan(scan, NULL);
+
+ while (table_scan_getnextslot(scan, ForwardScanDirection,
scanslot))
+ {
+ if (!tuples_equal(scanslot, searchslot, eq, NULL))
+ continue;
+
+ xwait = TransactionIdIsValid(snap.xmin) ?
+ snap.xmin : snap.xmax;
+ break;
+ }
+
+ if (TransactionIdIsValid(xwait))
+ {
+ PopActiveSnapshot();
+ XactLockTableWait(xwait, NULL, NULL, XLTW_None);
+ goto retry;
+ }
+ }
+
/* Found tuple, try to lock it in the lockmode. */
if (found)
{
# Experimento propio (#5151): que se PIERDE al cambiar SnapshotDirty por un
# snapshot MVCC fresco en RelationFindReplTupleByIndex.
#
# SnapshotDirty ve las filas de transacciones EN CURSO, y por eso el codigo de
# master toma su xmin/xmax y hace XactLockTableWait: espera al que esta
# insertando y recien despues decide. Un scan MVCC no ve esa fila, asi que no
# hay a quien esperar.
#
# Escenario determinista, sin injection points: el INSERT ya esta en vuelo en
# el subscriber ANTES de que llegue el UPDATE del publisher.
#
# sub: BEGIN; INSERT (1,'fromsub'); <- abierta, sin commit
# pub: UPDATE t SET data='frompubnew' WHERE a=1;
# sub: COMMIT;
#
# master -> el apply worker espera al insertador y aplica el UPDATE encima.
# parche -> el apply worker no ve nada, reporta update_missing y sigue.
#
# No se juzga cual es "correcto": se mide que son DISTINTOS, porque el hilo
# discute exactamente si SnapshotDirty daba o no una garantia extra.
use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;
my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
$node_publisher->start;
my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init;
$node_subscriber->start;
$node_publisher->safe_psql('postgres',
"CREATE TABLE t (a int PRIMARY KEY, data text)");
$node_subscriber->safe_psql('postgres',
"CREATE TABLE t (a int PRIMARY KEY, data text)");
# La fila existe SOLO en el publisher: con copy_data=false el subscriber
# arranca vacio, asi que la unica version de la fila en el subscriber sera la
# que inserte la sesion local.
$node_publisher->safe_psql('postgres',
"INSERT INTO t VALUES (1, 'frompub')");
$node_publisher->safe_psql('postgres', "CREATE PUBLICATION pub FOR TABLE t");
my $connstr = $node_publisher->connstr . ' dbname=postgres';
my $appname = 'sub_inflight';
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION sub CONNECTION '$connstr application_name=$appname'
PUBLICATION pub WITH (copy_data = false)");
$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
is($node_subscriber->safe_psql('postgres', 'SELECT count(*) FROM t'),
'0', 'el subscriber arranca sin la fila');
# 1) INSERT en vuelo en el subscriber (transaccion abierta).
my $s = $node_subscriber->background_psql('postgres');
$s->query_until(
qr/inflight/, qq[
\\echo inflight
BEGIN;
INSERT INTO t VALUES (1, 'fromsub');
]);
my $log_offset = -s $node_subscriber->logfile;
# 2) El publisher actualiza la fila.
$node_publisher->safe_psql('postgres',
"UPDATE t SET data = 'frompubnew' WHERE a = 1");
# 3) Se le da tiempo al apply worker para que llegue al lookup y decida.
# En master se queda esperando al insertador; con el parche resuelve al tiro.
sleep 3;
my $quedo_esperando = $node_subscriber->safe_psql('postgres',
"SELECT count(*) FROM pg_stat_activity
WHERE backend_type = 'logical replication apply worker'
AND wait_event_type = 'Lock'");
my $reporto_missing = $node_subscriber->log_contains(
qr/conflict=update_missing/, $log_offset) ? 1 : 0;
# 4) Recien ahora termina la sesion local. Con ACCION=ROLLBACK se comprueba
# el otro desenlace: master espera y, si el insertador aborta, reporta
# update_missing igual que el parche (o sea, master no "siempre aplica":
# espera y decide segun lo que pase de verdad).
my $accion = $ENV{ACCION} // 'COMMIT';
$s->query_safe($accion);
$s->quit;
$node_publisher->wait_for_catchup($appname);
my $final = $node_subscriber->safe_psql('postgres',
"SELECT coalesce(max(data), '(sin fila)') FROM t WHERE a = 1");
note("apply worker esperando un lock: $quedo_esperando");
note("reporto update_missing antes del commit: $reporto_missing");
note("dato final en el subscriber: $final");
# Lo que queda registrado para el correo (no se juzga, se mide):
is("espera=$quedo_esperando missing=$reporto_missing final=$final",
$ENV{ESPERADO} // 'sin ESPERADO',
'resultado medido');
done_testing();
