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

Reset conflict slot's xmin when a subscription (re)starts retention.

Browse Source 
Since the retain_dead_tuples feature uses a single cluster-wide
replication slot across all subscriptions, adding a new subscription on a
database with an older active transaction can require
conflict_detection_slot.xmin to move backward to avoid missing dead tuples
needed for update_deleted conflict detection.

Previously, if a newly created subscription had an older active
transaction than the slot's current xmin, the slot update occurred during
worker apply, leading to an assertion failure.

Fix this by having the launcher track the set of databases with active
tuple-retaining subscriptions. When a database joins this set (either via
a new subscription or when retention restarts), the launcher resets the
slot's xmin to the cluster-wide safe decoding horizon before starting any
workers. Because this horizon accounts for all active transactions across
the entire cluster, it serves as a safe seed without requiring retroactive
backward movement by individual apply workers.

In passing, update the retain_dead_tuples documentation to clarify two
operational boundary cases:
(a) Conflict data continues to accumulate even when a subscription is
disabled or its apply worker is stopped.
(b) track_commit_timestamp must be enabled for conflict detection to
function properly.

Reported-by: Nisha Moond <nisha.moond412@gmail.com>
Author: Zhijie Hou <houzj.fnst@fujitsu.com>
Author: Amit Kapila <amit.kapila16@gmail.com>
Reviewed-by: Amit Kapila <amit.kapila16@gmail.com>
Reviewed-by: Nisha Moond <nisha.moond412@gmail.com>
Reviewed-by: shveta malik <shveta.malik@gmail.com>
Reviewed-by: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Discussion: https://postgr.es/m/TY4PR01MB177182F547A62FC2666EC04EC94B72@TY4PR01MB17718.jpnprd01.prod.outlook.com
Backpatch-through: 19, where it was introduced
This commit is contained in:
Amit Kapila
2026-09-07 14:34:15 +05:30
parent 2ad8996420
commit 9186ffd633
3 changed files with 221 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
doc/src/sgml/ref/create_subscription.sgml
View File

@@ -493,7 +493,13 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
the subscription is disabled; thus, the information will accumulate
until the subscription is enabled. To prevent excessive accumulation,
it is recommended to disable <literal>retain_dead_tuples</literal>
if the subscription will be inactive for an extended period.
if the subscription will be inactive for an extended period. The
same applies while the subscription is enabled but its apply worker
is not running, for example because the worker repeatedly fails to
apply a change. The oldest information being retained can be seen as
the <structfield>xmin</structfield> of the
<quote><literal>pg_conflict_detection</literal></quote> slot in
<link linkend="view-pg-replication-slots">pg_replication_slots</link>.
</para>
<para>
@@ -544,6 +550,17 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
publisher. For the changes originating from different origins,
reliable conflict detection cannot be guaranteed.
</para>
<para>
Detection of <xref linkend="conflict-update-deleted"/> also requires
<link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
to be enabled on the subscriber, and a warning is issued if
<literal>retain_dead_tuples</literal> is enabled while it is not. If
<varname>track_commit_timestamp</varname> is disabled afterwards, the
information for conflict detection continues to be retained, but the
conflict is reported as <xref linkend="conflict-update-missing"/>
instead, so the retention no longer serves its purpose.
</para>
</listitem>
</varlistentry>
@@ -579,6 +596,18 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<literal>retain_dead_tuples</literal> is enabled and the apply
worker associated with the subscription is active.
</para>
<caution>
<para>
Note that the retention duration is not evaluated while the
subscription is disabled or its apply worker is not running. Thus,
the information retained for conflict detection will continue to
accumulate regardless of this setting until the subscription is
enabled or its apply worker resumes. To prevent excessive
accumulation, consider disabling
<literal>retain_dead_tuples</literal> if the subscription will be
inactive for an extended period.
</para>
</caution>
<warning>
<para>
Note that setting a non-zero value for this option could lead to

173
src/backend/replication/logical/launcher.c
View File

@@ -111,10 +111,12 @@ static int logicalrep_pa_worker_count(Oid subid);
static void logicalrep_launcher_attach_dshmem(void);
static void ApplyLauncherSetWorkerStartTime(Oid subid, TimestampTz start_time);
static TimestampTz ApplyLauncherGetWorkerStartTime(Oid subid);
static void compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin);
static bool collect_min_nonremovable_xid(Subscription *sub,
LogicalRepWorker *worker,
TransactionId *xmin);
static bool acquire_conflict_slot_if_exists(void);
static void update_conflict_slot_xmin(TransactionId new_xmin);
static void init_conflict_slot_xmin(void);
static void reset_conflict_slot_xmin_to_safe_horizon(void);
/*
@@ -1204,6 +1206,8 @@ ApplyLauncherWakeup(void)
void
ApplyLauncherMain(Datum main_arg)
{
List *retained_dbids;
ereport(DEBUG1,
(errmsg_internal("logical replication launcher started")));
@@ -1222,6 +1226,13 @@ ApplyLauncherMain(Datum main_arg)
*/
BackgroundWorkerInitializeConnection(NULL, NULL, 0);
/*
* Databases with actively-retaining subscriptions as of the previous
* cycle. Local memory only, so every such database counts as new after a
* launcher restart, resetting the slot's xmin once at startup.
*/
retained_dbids = NIL;
/*
* Acquire the conflict detection slot at startup to ensure it can be
* dropped if no longer needed after a restart.
@@ -1239,7 +1250,9 @@ ApplyLauncherMain(Datum main_arg)
long wait_time = DEFAULT_NAPTIME_PER_CYCLE;
bool can_update_xmin = true;
bool retain_dead_tuples = false;
bool reset_done = false;
TransactionId xmin = InvalidTransactionId;
List *current_dbids = NIL;
CHECK_FOR_INTERRUPTS();
@@ -1288,6 +1301,10 @@ ApplyLauncherMain(Datum main_arg)
if (sub->retentionactive)
{
/* Remember the retained databases for the next cycle. */
current_dbids = list_append_unique_oid(current_dbids,
sub->dbid);
/*
* Can't advance xmin of the slot unless all the
* subscriptions actively retaining dead tuples are
@@ -1301,11 +1318,25 @@ ApplyLauncherMain(Datum main_arg)
can_update_xmin &= sub->enabled;
/*
* Initialize the slot once the subscription activates
* retention.
* A retaining apply worker starts with the slot's xmin as
* its oldest_nonremovable_xid (see
* logicalrep_worker_launch()), so a database whose oldest
* active transaction ID precedes that xmin would trip the
* assertion in get_candidate_xid(). Reset the xmin before
* starting the worker, which also seeds it on the first
* cycle. The set of retained databases is keyed on
* subretentionactive, so a subscription resuming
* retention is also treated as a new arrival. One call
* per cycle is enough, as a single reset covers all such
* databases.
*/
if (!TransactionIdIsValid(MyReplicationSlot->data.xmin))
init_conflict_slot_xmin();
if (!reset_done &&
(!TransactionIdIsValid(MyReplicationSlot->data.xmin) ||
!list_member_oid(retained_dbids, sub->dbid)))
{
reset_conflict_slot_xmin_to_safe_horizon();
reset_done = true;
}
}
}
@@ -1316,37 +1347,18 @@ ApplyLauncherMain(Datum main_arg)
w = logicalrep_worker_find(WORKERTYPE_APPLY, sub->oid, InvalidOid,
false);
if (w != NULL)
{
/*
* Compute the minimum xmin required to protect dead tuples
* required for conflict detection among all running apply
* workers. This computation is performed while holding
* LogicalRepWorkerLock to prevent accessing invalid worker
* data, in scenarios where a worker might exit and reset its
* state concurrently.
*/
if (sub->retaindeadtuples &&
sub->retentionactive &&
can_update_xmin)
compute_min_nonremovable_xid(w, &xmin);
LWLockRelease(LogicalRepWorkerLock);
/* worker is running already */
continue;
}
/*
* Compute the minimum xmin required to protect dead tuples
* required for conflict detection among all running apply
* workers.
*/
can_update_xmin = can_update_xmin &&
collect_min_nonremovable_xid(sub, w, &xmin);
LWLockRelease(LogicalRepWorkerLock);
/*
* Can't advance xmin of the slot unless all the workers
* corresponding to subscriptions actively retaining dead tuples
* are running, disabling the further computation of the minimum
* nonremovable xid.
*/
if (sub->retaindeadtuples && sub->retentionactive)
can_update_xmin = false;
if (w != NULL)
continue; /* worker is running already */
/*
* If the worker is eligible to start now, launch it. Otherwise,
@@ -1414,6 +1426,14 @@ ApplyLauncherMain(Datum main_arg)
/* Switch back to original memory context. */
MemoryContextSwitchTo(oldctx);
/*
* Remember the current set of retained databases for the next cycle,
* in long-lived memory.
*/
list_free(retained_dbids);
retained_dbids = list_copy(current_dbids);
/* Clean the temporary memory. */
MemoryContextDelete(subctx);
@@ -1440,16 +1460,28 @@ ApplyLauncherMain(Datum main_arg)
}
/*
* Determine the minimum non-removable transaction ID across all apply workers
* for subscriptions that have retain_dead_tuples enabled. Store the result
* in *xmin.
* Determine the minimum non-removable transaction ID required for conflict
* detection, accumulating the contribution of the given subscription in
* *xmin. Subscriptions that are not actively retaining conflict information
* are ignored.
*
* Returns false if the slot's xmin cannot be advanced in this cycle, which is
* the case when the apply worker is not running (worker is NULL) or does not
* yet hold a valid oldest_nonremovable_xid.
*
* The caller must hold LogicalRepWorkerLock, to prevent accessing invalid
* worker data in scenarios where a worker might exit and reset its state
* concurrently.
*/
static void
compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin)
static bool
collect_min_nonremovable_xid(Subscription *sub, LogicalRepWorker *worker,
TransactionId *xmin)
{
TransactionId nonremovable_xid;
TransactionId nonremovable_xid = InvalidTransactionId;
Assert(worker != NULL);
/* Skip subscriptions that are not actively retaining dead tuples */
if (!sub->retaindeadtuples || !sub->retentionactive)
return true;
/*
* The replication slot for conflict detection must be created before the
@@ -1457,23 +1489,32 @@ compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin)
*/
Assert(MyReplicationSlot);
SpinLockAcquire(&worker->relmutex);
nonremovable_xid = worker->oldest_nonremovable_xid;
SpinLockRelease(&worker->relmutex);
if (worker)
{
SpinLockAcquire(&worker->relmutex);
nonremovable_xid = worker->oldest_nonremovable_xid;
SpinLockRelease(&worker->relmutex);
}
/*
* Return if the apply worker has stopped retention concurrently.
* Skip the xmin update if the apply worker's oldest_nonremovable_xid is
* invalid, or if the worker is not running (worker is NULL).
*
* Although this function is invoked only when retentionactive is true,
* the apply worker might stop retention after the launcher fetches the
* retentionactive flag.
* Although this function proceeds only when retentionactive is true,
* worker's oldest_nonremovable_xid being invalid can still happen when
* the worker has stopped retention after the launcher fetched
* retentionactive, or when the worker recently enabled retention but
* hasn't restarted yet and the XID is still invalid. In either case, it's
* unnecessary to update the slot's xmin in this cycle.
*/
if (!TransactionIdIsValid(nonremovable_xid))
return;
return false;
if (!TransactionIdIsValid(*xmin) ||
TransactionIdPrecedes(nonremovable_xid, *xmin))
*xmin = nonremovable_xid;
return true;
}
/*
@@ -1530,27 +1571,49 @@ update_conflict_slot_xmin(TransactionId new_xmin)
}
/*
* Initialize the xmin for the conflict detection slot.
* Reset the xmin of the conflict detection slot to the cluster-wide safe
* decoding horizon, which accounts for all running transactions and is
* therefore safe for an apply worker in any database. Called when the slot is
* created and when a database newly needs retention.
*
* The xmin is left unchanged if it is already at or before the horizon.
*/
static void
init_conflict_slot_xmin(void)
reset_conflict_slot_xmin_to_safe_horizon(void)
{
TransactionId xmin_horizon;
TransactionId old_xmin;
/* Replication slot must exist but shouldn't be initialized. */
Assert(MyReplicationSlot &&
!TransactionIdIsValid(MyReplicationSlot->data.xmin));
Assert(MyReplicationSlot);
LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
xmin_horizon = GetOldestSafeDecodingTransactionId(false);
old_xmin = MyReplicationSlot->data.xmin;
/*
* Nothing to do if the current xmin is valid and not newer than the
* horizon. The xmin must not be advanced here, as advancing is only safe
* once the apply workers confirm that all concurrent transactions have
* been applied. Regressing it only retains more than necessary, and the
* workers will advance it again in later cycles.
*/
if (TransactionIdIsValid(old_xmin) &&
TransactionIdPrecedesOrEquals(old_xmin, xmin_horizon))
{
LWLockRelease(ProcArrayLock);
LWLockRelease(ReplicationSlotControlLock);
return;
}
SpinLockAcquire(&MyReplicationSlot->mutex);
MyReplicationSlot->effective_xmin = xmin_horizon;
MyReplicationSlot->data.xmin = xmin_horizon;
SpinLockRelease(&MyReplicationSlot->mutex);
elog(DEBUG1, "reset conflict detection slot's xmin to %u", xmin_horizon);
ReplicationSlotsComputeRequiredXmin(true);
LWLockRelease(ProcArrayLock);
@@ -1578,7 +1641,7 @@ CreateConflictDetectionSlot(void)
ReplicationSlotCreate(CONFLICT_DETECTION_SLOT, false, RS_PERSISTENT, false,
false, false, false);
init_conflict_slot_xmin();
reset_conflict_slot_xmin_to_safe_horizon();
}
/*

73
src/test/subscription/t/035_conflicts.pl
View File

@@ -649,6 +649,79 @@ $result = $node_A->safe_psql('postgres',
);
is($result, qq(t), 'retention is active');
###############################################################################
# Check that the conflict detection slot's xmin is re-initialized when a
# database newly appears among the databases with retain_dead_tuples
# subscriptions.
#
# The slot's xmin is advanced according to the per-database horizons of the
# databases seen so far. Without re-initialization, a worker started in a
# newly retaining database whose oldest active transaction ID is older would
# be seeded with a value newer than its database's horizon.
###############################################################################
# Create a second database on node B, with the same table.
$node_B->safe_psql('postgres', "CREATE DATABASE dbb");
$node_B->safe_psql('dbb', "CREATE TABLE tab (a int PRIMARY KEY, b int)");
# Hold a transaction with an assigned transaction ID open in dbb, pinning its
# oldest active transaction ID.
my $dbb_session = $node_B->background_psql('dbb');
$dbb_session->query_until(
qr/starting_bg_psql/, q{
\echo starting_bg_psql
BEGIN;
SELECT txid_current();
});
# Push the transaction ID counter clearly past the pinned transaction ID and
# wait for the slot's xmin to advance past it. Only the apply worker in the
# postgres database drives the slot's xmin here, and postgres has no old
# transaction running.
$next_xid = $node_B->safe_psql('postgres', "SELECT txid_current() + 1");
ok( $node_B->poll_query_until(
'postgres',
"SELECT xmin::text::bigint >= $next_xid FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
),
"slot xmin advanced past the transaction ID pinned in dbb");
# Create the second retention subscription in dbb. The launcher must
# re-initialize the slot's xmin before launching dbb's apply worker.
my $subname_BA2 = 'tap_sub_b_a2';
$node_B->safe_psql('dbb',
"CREATE SUBSCRIPTION $subname_BA2
CONNECTION '$node_A_connstr application_name=$subname_BA2'
PUBLICATION tap_pub_A
WITH (retain_dead_tuples = true, origin = none)");
$node_B->wait_for_subscription_sync($node_A, $subname_BA2, 'dbb');
# The slot's xmin must regress to the horizon pinned in dbb.
ok( $node_B->poll_query_until(
'postgres',
"SELECT xmin::text::bigint < $next_xid FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
),
"slot xmin regressed to the horizon pinned in dbb");
# Once the pinned transaction commits, the xmin must be able to advance
# again.
$dbb_session->query_until(
qr/committed/, q{
COMMIT;
\echo committed
});
ok($dbb_session->quit, 'close pinned session');
$next_xid = $node_B->safe_psql('postgres', "SELECT txid_current() + 1");
ok( $node_B->poll_query_until(
'postgres',
"SELECT xmin::text::bigint >= $next_xid FROM pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
),
"slot xmin advances again after the pinned transaction commits");
# Clean up the second database.
$node_B->safe_psql('dbb', "DROP SUBSCRIPTION $subname_BA2");
$node_B->safe_psql('postgres', "DROP DATABASE dbb");
###############################################################################
# Check that the replication slot pg_conflict_detection is dropped after
# removing all the subscriptions.