Hello, everyone! Manu reviewed the fix in the original thread [1] and pointed out a behavior change. I'm replying here since this thread is about the logical replication side.
The case is this: a local transaction inserts a row and remains open, while an UPDATE for the same key arrives from the publisher. On master, the dirty scan may see the uncommitted row, wait for the inserter, and then apply the UPDATE after it commits. With the patch, the row isn't visible, so we report update_missing immediately. That behavior on master is racy, though. It only waits if the insert has already made it into the index by the time the scan reaches it. If the INSERT happens a little later, master reports update_missing as well. In either case there is no committed row at the time the replicated change is applied, which is different from the bug this patch fixes, where the row exists throughout. I've updated the commit message to make that distinction clear. Manu also suggested doing another dirty scan when nothing is found, to preserve the wait in more cases. That works, but it only makes the race window smaller and reintroduces a dirty scan, so I left it out of a fix intended to be back-patchable. [1]: https://postgr.es/m/[email protected] Best regards, Mikhail.
From 752e6da93c51e2f1fd68685a0d63deeda11aee91 Mon Sep 17 00:00:00 2001 From: Mikhail Nikalayeu <[email protected]> Date: Mon, 5 Oct 2026 22:48:29 +0200 Subject: [PATCH v20] Find tuples to update or delete during apply using an MVCC snapshot RelationFindReplTupleByIndex() and RelationFindReplTupleSeq() used a dirty snapshot. This can miss a concurrently updated row if the old version is already dead and the new version is placed behind the scan position. Apply then treats the row as missing and skips the change. Use a fresh MVCC snapshot for each search attempt instead. Concurrent updates and deletes of a tuple found this way are still handled by table_tuple_lock(). This changes the behavior for rows inserted by an in-progress local transaction. A dirty snapshot could see such a row and wait for the inserter, while an MVCC snapshot reports it missing immediately. However, the old behavior was already timing-dependent: the wait only happened if the insert had reached the index before the scan, while an insert arriving later was reported missing. So this does not remove a guarantee that previously existed. Add a test that pauses apply during the search and updates the row locally. Also document this limitation of dirty snapshots in snapshot.h and the nbtree README. Reviewed-by: Amit Kapila <[email protected]> Reviewed-by: Manuel Reyes Bravo <[email protected]> Discussion: https://postgr.es/m/CANtu0oiktqQ2pwExoXqDpByXNCJa-KE5vQRodTRnmFHN_+qwHg@mail.gmail.com Discussion: https://postgr.es/m/CADzfLwXZVmbo11tFS_G2i+6TfFVwHU4VUUSeoqb+8UQfuoJs8A@mail.gmail.com --- src/backend/access/heap/heapam_indexscan.c | 10 ++ src/backend/access/nbtree/README | 11 ++ src/backend/executor/execIndexing.c | 2 + src/backend/executor/execReplication.c | 72 ++------ src/backend/replication/logical/worker.c | 3 + src/include/utils/snapshot.h | 14 ++ src/test/subscription/meson.build | 2 + .../t/039_apply_search_concurrent_update.pl | 156 ++++++++++++++++++ .../t/040_update_missing_pgbench_demo.pl | 128 ++++++++++++++ 9 files changed, 343 insertions(+), 55 deletions(-) create mode 100644 src/test/subscription/t/039_apply_search_concurrent_update.pl create mode 100644 src/test/subscription/t/040_update_missing_pgbench_demo.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..0d74387b6e0 100644 --- a/src/backend/access/nbtree/README +++ b/src/backend/access/nbtree/README @@ -103,6 +103,17 @@ 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 the 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. A heap scan with such a snapshot can +lose a tuple in the same way, if its new version is placed on a heap page +the scan has already left. + 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..8938b95b218 100644 --- a/src/backend/executor/execIndexing.c +++ b/src/backend/executor/execIndexing.c @@ -788,6 +788,8 @@ 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); diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c index 18da4fe08f6..00cdc4988ae 100644 --- a/src/backend/executor/execReplication.c +++ b/src/backend/executor/execReplication.c @@ -192,8 +192,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; @@ -201,18 +199,15 @@ RelationFindReplTupleByIndex(Relation rel, Oid idxoid, /* Open the index. */ idxrel = index_open(idxoid, RowExclusiveLock); - InitDirtySnapshot(snap); - /* Build scan key. */ skey_attoff = build_replindex_scan_key(skey, rel, idxrel, searchslot); - /* Start an index scan. */ - scan = index_beginscan(rel, idxrel, false, - &snap, NULL, skey_attoff, 0, SO_NONE); - retry: found = false; + PushActiveSnapshot(GetLatestSnapshot()); + scan = index_beginscan(rel, idxrel, false, + GetActiveSnapshot(), NULL, skey_attoff, 0, SO_NONE); index_rescan(scan, skey, skey_attoff, NULL, 0); /* Try to find the tuple */ @@ -233,32 +228,18 @@ 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; } + index_endscan(scan); + /* Found tuple, try to lock it in the lockmode. */ if (found) { TM_FailureData tmfd; TM_Result res; - PushActiveSnapshot(GetLatestSnapshot()); - res = table_tuple_lock(rel, &(outslot->tts_tid), GetActiveSnapshot(), outslot, GetCurrentCommandId(false), @@ -267,13 +248,14 @@ 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); @@ -374,26 +356,20 @@ 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); Assert(equalTupleDescs(desc, outslot->tts_tupleDescriptor)); eq = palloc0_array(TypeCacheEntry *, outslot->tts_tupleDescriptor->natts); - - /* Start a heap scan. */ - InitDirtySnapshot(snap); - scan = table_beginscan(rel, &snap, 0, NULL, - SO_NONE); scanslot = table_slot_create(rel, NULL); retry: found = false; + PushActiveSnapshot(GetLatestSnapshot()); - table_rescan(scan, NULL); + scan = table_beginscan(rel, GetActiveSnapshot(), 0, NULL, SO_NONE); /* Try to find the tuple */ while (table_scan_getnextslot(scan, ForwardScanDirection, scanslot)) @@ -403,32 +379,17 @@ 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; } + table_endscan(scan); + /* Found tuple, try to lock it in the lockmode. */ if (found) { TM_FailureData tmfd; TM_Result res; - PushActiveSnapshot(GetLatestSnapshot()); - res = table_tuple_lock(rel, &(outslot->tts_tid), GetActiveSnapshot(), outslot, GetCurrentCommandId(false), @@ -437,13 +398,14 @@ 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; diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 44d735cdc55..7a93280a63d 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -3029,7 +3029,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..7cb9b9d6b20 100644 --- a/src/test/subscription/meson.build +++ b/src/test/subscription/meson.build @@ -48,6 +48,8 @@ tests += { 't/036_sequences.pl', 't/037_except.pl', 't/038_walsnd_shutdown_timeout.pl', + 't/039_apply_search_concurrent_update.pl', + 't/040_update_missing_pgbench_demo.pl', 't/100_bugs.pl', ], }, diff --git a/src/test/subscription/t/039_apply_search_concurrent_update.pl b/src/test/subscription/t/039_apply_search_concurrent_update.pl new file mode 100644 index 00000000000..bbe1d2e14f3 --- /dev/null +++ b/src/test/subscription/t/039_apply_search_concurrent_update.pl @@ -0,0 +1,156 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Apply searching for a row that is updated concurrently on the subscriber. +# +# The apply worker looks up the row to update or delete through the replica +# identity index. A non-HOT update of that row on the subscriber gives it a +# new index entry. If that update commits after the index has returned the +# old version's TID but before apply fetches it, the old version is dead by +# then and the new entry may lie where the scan has already been. A search +# that sees the row in neither version takes it for missing and skips the +# change, although the row existed all along. Apply must find the row and +# report the actual conflict: a row modified by another origin. + +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'; +} + +my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +$node_publisher->append_conf('postgresql.conf', + 'track_commit_timestamp = on'); +$node_publisher->start; + +# track_commit_timestamp is needed to detect rows modified by another +# origin, and wal_level = replica for retain_dead_tuples. +my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$node_subscriber->init; +$node_subscriber->append_conf( + 'postgresql.conf', qq( +track_commit_timestamp = on +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'; +} +$node_subscriber->safe_psql('postgres', 'CREATE EXTENSION injection_points'); + +# The subscriber's table has a column of its own, indexed so that updating +# it is never HOT. +$node_publisher->safe_psql('postgres', + 'CREATE TABLE conf_tab (a int PRIMARY KEY, data text)'); +$node_subscriber->safe_psql( + 'postgres', q[ + CREATE TABLE conf_tab (a int PRIMARY KEY, data text, i int DEFAULT 0); + CREATE INDEX i_index ON conf_tab (i); +]); + +$node_publisher->safe_psql( + 'postgres', q[ + INSERT INTO conf_tab VALUES (1, 'frompub'), (2, 'frompub'), (3, 'frompub'); + CREATE PUBLICATION tap_pub FOR TABLE conf_tab; +]); + +my $appname = 'tap_sub'; +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub" +); +$node_subscriber->wait_for_subscription_sync($node_publisher, $appname); + +# Run $pub_sql on the publisher. Hold apply after the index has returned +# the row's TID and before the heap fetch, run $sub_sql on the subscriber +# meanwhile, then let apply go on. Return the conflicts apply reported. +sub apply_during_local_update +{ + my ($pub_sql, $sub_sql) = @_; + my $point = 'index_getnext_slot_before_fetch_apply_dirty'; + my $log_offset = -s $node_subscriber->logfile; + + $node_subscriber->safe_psql('postgres', + "SELECT injection_points_attach('$point', 'wait')"); + $node_publisher->safe_psql('postgres', $pub_sql); + $node_subscriber->wait_for_event('logical replication apply worker', + $point); + + $node_subscriber->safe_psql('postgres', $sub_sql); + + $node_subscriber->safe_psql( + 'postgres', qq[ + SELECT injection_points_detach('$point'); + SELECT injection_points_wakeup('$point'); + ]); + + $node_subscriber->wait_for_log( + qr/conflict detected on relation "public.conf_tab"/, $log_offset); + $node_publisher->wait_for_catchup($appname); + + my $log = slurp_file($node_subscriber->logfile, $log_offset); + return join(' ', + $log =~ + /conflict detected on relation "public\.conf_tab": conflict=(\w+)/g + ); +} + +# DELETE on the publisher, not delete_missing. +is( apply_during_local_update( + 'DELETE FROM conf_tab WHERE a = 1', + 'UPDATE conf_tab SET i = 1 WHERE a = 1'), + 'delete_origin_differs', + 'DELETE: conflict with the locally updated row'); +is( $node_subscriber->safe_psql( + 'postgres', 'SELECT count(*) FROM conf_tab WHERE a = 1'), + '0', + 'DELETE: row deleted on subscriber'); + +# UPDATE on the publisher, not update_missing. The local update touches a +# column the publisher does not have, so both updates must survive. +is( apply_during_local_update( + "UPDATE conf_tab SET data = 'frompubnew' WHERE a = 2", + 'UPDATE conf_tab SET i = 1 WHERE a = 2'), + 'update_origin_differs', + 'UPDATE: conflict with the locally updated row'); +is( $node_subscriber->safe_psql( + 'postgres', 'SELECT data, i FROM conf_tab WHERE a = 2'), + 'frompubnew|1', + 'UPDATE: both updates applied'); + +# The same with retain_dead_tuples, where a row taken for missing is looked +# for among the dead ones and found there: not update_deleted. +$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION tap_sub DISABLE'); +$node_subscriber->poll_query_until('postgres', + "SELECT count(*) = 0 FROM pg_stat_activity WHERE backend_type = 'logical replication apply worker'" +); +$node_subscriber->safe_psql('postgres', + 'ALTER SUBSCRIPTION tap_sub SET (retain_dead_tuples = true)'); +$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION tap_sub ENABLE'); +ok( $node_subscriber->poll_query_until( + 'postgres', + "SELECT xmin IS NOT NULL FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'" + ), + 'dead tuples are retained'); + +is( apply_during_local_update( + "UPDATE conf_tab SET data = 'frompubnew' WHERE a = 3", + 'UPDATE conf_tab SET i = 1 WHERE a = 3'), + 'update_origin_differs', + 'UPDATE with retain_dead_tuples: conflict with the locally updated row'); +is( $node_subscriber->safe_psql( + 'postgres', 'SELECT data, i FROM conf_tab WHERE a = 3'), + 'frompubnew|1', + 'UPDATE with retain_dead_tuples: both updates applied'); + +done_testing(); diff --git a/src/test/subscription/t/040_update_missing_pgbench_demo.pl b/src/test/subscription/t/040_update_missing_pgbench_demo.pl new file mode 100644 index 00000000000..2fbbea313e6 --- /dev/null +++ b/src/test/subscription/t/040_update_missing_pgbench_demo.pl @@ -0,0 +1,128 @@ +# Copyright (c) 2025, PostgreSQL Global Development Group + +# NOT INTENDED FOR COMMIT. A demonstration, too heavy for the regular +# suite, that the race covered by 039_apply_search_concurrent_update.pl +# happens without any injection point holding the apply worker: pgbench +# updates the same rows on the publisher and on the subscriber at once, +# and the injection point apply_handle_update_internal_update_missing +# turns any update_missing reported by apply into an error. +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.43.0
