diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index da89c9dd037..36ace635595 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -493,7 +493,13 @@ CREATE SUBSCRIPTION subscription_nameretain_dead_tuples - 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 xmin of the + pg_conflict_detection slot in + pg_replication_slots. @@ -544,6 +550,17 @@ CREATE SUBSCRIPTION subscription_name + + + Detection of also requires + track_commit_timestamp + to be enabled on the subscriber, and a warning is issued if + retain_dead_tuples is enabled while it is not. If + track_commit_timestamp is disabled afterwards, the + information for conflict detection continues to be retained, but the + conflict is reported as + instead, so the retention no longer serves its purpose. + @@ -579,6 +596,18 @@ CREATE SUBSCRIPTION subscription_nameretain_dead_tuples is enabled and the apply worker associated with the subscription is active. + + + 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 + retain_dead_tuples if the subscription will be + inactive for an extended period. + + Note that setting a non-zero value for this option could lead to diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 313e31ff2e3..c328ad53ecc 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -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(); } /* diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl index f23fe6af2a5..ff9f4c9b760 100644 --- a/src/test/subscription/t/035_conflicts.pl +++ b/src/test/subscription/t/035_conflicts.pl @@ -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.