From 44162d3da40b3ba7f295da79efe8ab5507e632e1 Mon Sep 17 00:00:00 2001 From: Zhijie Hou Date: Mon, 1 Sep 2025 12:40:36 +0800 Subject: [PATCH v70 2/3] Resume retaining the information for conflict detection The patch allows the launcher to re-initialized invalidated slot, if at least one apply worker has confirmed that the retention duration is now within the max_retention_duration. --- doc/src/sgml/ref/create_subscription.sgml | 9 +- src/backend/commands/subscriptioncmds.c | 9 +- src/backend/replication/logical/launcher.c | 48 ++++- src/backend/replication/logical/worker.c | 206 ++++++++++++++++++--- src/include/replication/worker_internal.h | 6 + 5 files changed, 240 insertions(+), 38 deletions(-) diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml index fc314437311..ee22d0fb3ca 100644 --- a/doc/src/sgml/ref/create_subscription.sgml +++ b/doc/src/sgml/ref/create_subscription.sgml @@ -538,10 +538,11 @@ CREATE SUBSCRIPTION subscription_nameretain_dead_tuples is enabled, confirm that the retention duration has exceeded the max_retention_duration set within the corresponding - subscription. The retention will not be automatically resumed unless a - new subscription is created with retain_dead_tuples = - true, or the user manually re-enables - retain_dead_tuples. + subscription. The retention will be automatically resumed once at least + one apply worker confirms that the retention duration is within the + specified limit, or a new subscription is created with + retain_dead_tuples = true, or the user manually + re-enables retain_dead_tuples. Note that overall retention will not stop if other subscriptions that diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 82cf65fae73..cc622fce58f 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -854,7 +854,14 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt, pgstat_create_subscription(subid); - if (opts.enabled) + /* + * If the subscription is enabled, notify the launcher to start the apply + * worker. + * + * If the subscription has retain_dead_tuples enabled, notify the launcher + * to create or resume the conflict detection slot. + */ + if (opts.enabled || opts.retaindeadtuples) ApplyLauncherWakeupAtCommit(); ObjectAddressSet(myself, SubscriptionRelationId, subid); diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index add2e2e066c..226421e3dfa 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -101,7 +101,9 @@ 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 void compute_min_nonremovable_xid(LogicalRepWorker *worker, + bool can_update_xmin, + 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); @@ -468,6 +470,7 @@ retry: worker->oldest_nonremovable_xid = retain_dead_tuples ? MyReplicationSlot->data.xmin : InvalidTransactionId; + worker->wait_for_initial_xid = false; worker->last_lsn = InvalidXLogRecPtr; TIMESTAMP_NOBEGIN(worker->last_send_time); TIMESTAMP_NOBEGIN(worker->last_recv_time); @@ -1270,10 +1273,8 @@ ApplyLauncherMain(Datum main_arg) * required for conflict detection among all running apply * workers. */ - if (sub->retaindeadtuples && - sub->retentionactive && - can_update_xmin) - compute_min_nonremovable_xid(w, &xmin); + if (sub->retaindeadtuples && sub->retentionactive) + compute_min_nonremovable_xid(w, can_update_xmin, &xmin); /* worker is running already */ continue; @@ -1382,11 +1383,16 @@ 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. + * + * Additionally, if an apply worker has an invalid XID and is requesting to + * resume retention, assign the slot's xmin value to it. */ static void -compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin) +compute_min_nonremovable_xid(LogicalRepWorker *worker, bool can_update_xmin, + TransactionId *xmin) { TransactionId nonremovable_xid; + bool wait_for_xid; Assert(worker != NULL); @@ -1398,16 +1404,42 @@ compute_min_nonremovable_xid(LogicalRepWorker *worker, TransactionId *xmin) SpinLockAcquire(&worker->relmutex); nonremovable_xid = worker->oldest_nonremovable_xid; + wait_for_xid = worker->wait_for_initial_xid; SpinLockRelease(&worker->relmutex); /* - * Return if the apply worker has stopped retention concurrently. + * Assign slot.xmin to the apply worker's oldest_nonremovable_xid if + * requested. This ensures the apply worker continues to maintain the + * oldest_nonremovable_xid (see resume_conflict_info_retention). + */ + if (wait_for_xid) + { + nonremovable_xid = MyReplicationSlot->data.xmin; + + Assert(TransactionIdIsValid(nonremovable_xid)); + + SpinLockAcquire(&worker->relmutex); + worker->oldest_nonremovable_xid = nonremovable_xid; + SpinLockRelease(&worker->relmutex); + + /* Notify the apply worker to start the next cycle of management */ + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + logicalrep_worker_wakeup_ptr(worker); + LWLockRelease(LogicalRepWorkerLock); + } + + /* + * Return if the apply worker has stopped retention concurrently and has not + * yet resumed. * * Although this function is invoked only when retentionactive is true, * the apply worker might stop retention after the launcher fetches the * retentionactive flag. */ - if (!TransactionIdIsValid(nonremovable_xid)) + else if (!TransactionIdIsValid(nonremovable_xid)) + return; + + if (!can_update_xmin) return; if (!TransactionIdIsValid(*xmin) || diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index f5f9183bdaf..0f014f44db4 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -181,6 +181,19 @@ * pg_subscription.subretentionactive is updated to false within a new * transaction, and oldest_nonremovable_xid is set to InvalidTransactionId. * + * - RDT_RESUME_CONFLICT_INFO_RETENTION: + * This phase is required only when max_retention_duration is defined. We + * enter this phase if the retention was previously stopped, and the time + * required to advance the non-removable transaction ID in the + * RDT_WAIT_FOR_LOCAL_FLUSH phase has decreased to within acceptable limits + * (or if max_retention_duration is set to 0). During this phase, + * pg_subscription.subretentionactive is updated to true within a new + * transaction, and we wait for the launcher to initialize the + * oldest_nonremovable_xid before proceeding to RDT_GET_CANDIDATE_XID phase. + * Note that the state could transition to RDT_RESUME_CONFLICT_INFO_RETENTION + * at any phase if the retention has been stopped, but max_retention_duration + * is now set to 0. + * * The overall state progression is: GET_CANDIDATE_XID -> * REQUEST_PUBLISHER_STATUS -> WAIT_FOR_PUBLISHER_STATUS -> (loop to * REQUEST_PUBLISHER_STATUS till concurrent remote transactions end) -> @@ -382,7 +395,8 @@ typedef enum RDT_REQUEST_PUBLISHER_STATUS, RDT_WAIT_FOR_PUBLISHER_STATUS, RDT_WAIT_FOR_LOCAL_FLUSH, - RDT_STOP_CONFLICT_INFO_RETENTION + RDT_STOP_CONFLICT_INFO_RETENTION, + RDT_RESUME_CONFLICT_INFO_RETENTION, } RetainDeadTuplesPhase; /* @@ -569,6 +583,10 @@ static void wait_for_publisher_status(RetainDeadTuplesData *rdt_data, static void wait_for_local_flush(RetainDeadTuplesData *rdt_data); static bool should_stop_conflict_info_retention(RetainDeadTuplesData *rdt_data); static void stop_conflict_info_retention(RetainDeadTuplesData *rdt_data); +static bool should_resume_retention_immediately(RetainDeadTuplesData *rdt_data, + bool status_received); +static void resume_conflict_info_retention(RetainDeadTuplesData *rdt_data); +static void update_retention_status(bool active); static void reset_retention_data_fields(RetainDeadTuplesData *rdt_data); static void adjust_xid_advance_interval(RetainDeadTuplesData *rdt_data, bool new_xid_found); @@ -4339,6 +4357,13 @@ maybe_advance_nonremovable_xid(RetainDeadTuplesData *rdt_data, if (!can_advance_nonremovable_xid(rdt_data)) return; + /* + * Resume retention immediately if required. (See + * should_resume_retention_immediately() for details). + */ + if (should_resume_retention_immediately(rdt_data, status_received)) + rdt_data->phase = RDT_RESUME_CONFLICT_INFO_RETENTION; + process_rdt_phase_transition(rdt_data, status_received); } @@ -4361,10 +4386,6 @@ can_advance_nonremovable_xid(RetainDeadTuplesData *rdt_data) if (!MySubscription->retaindeadtuples) return false; - /* No need to advance if we have already stopped retaining */ - if (!MySubscription->retentionactive) - return false; - return true; } @@ -4393,6 +4414,9 @@ process_rdt_phase_transition(RetainDeadTuplesData *rdt_data, case RDT_STOP_CONFLICT_INFO_RETENTION: stop_conflict_info_retention(rdt_data); break; + case RDT_RESUME_CONFLICT_INFO_RETENTION: + resume_conflict_info_retention(rdt_data); + break; } } @@ -4649,6 +4673,18 @@ wait_for_local_flush(RetainDeadTuplesData *rdt_data) if (last_flushpos < rdt_data->remote_lsn) return; + /* + * If conflict info retention was previously stopped due to a timeout, and + * the time required to advance the non-removable transaction ID has now + * decreased to within acceptable limits, resume the rentention. + */ + if (!MySubscription->retentionactive) + { + rdt_data->phase = RDT_RESUME_CONFLICT_INFO_RETENTION; + process_rdt_phase_transition(rdt_data, false); + return; + } + /* * Reaching here means the remote WAL position has been received, and all * transactions up to that position on the publisher have been applied and @@ -4679,9 +4715,8 @@ wait_for_local_flush(RetainDeadTuplesData *rdt_data) * RDT_STOP_CONFLICT_INFO_RETENTION phase and return true. Otherwise, return * false. * - * Note: Retention won't be resumed automatically. The user must manually - * disable retain_dead_tuples and re-enable it after confirming that the - * replication slot maintained by the launcher has been dropped. + * The retention will resume automatically if the worker has confirmed that the + * retention duration is now within the max_retention_duration. */ static bool should_stop_conflict_info_retention(RetainDeadTuplesData *rdt_data) @@ -4712,10 +4747,16 @@ should_stop_conflict_info_retention(RetainDeadTuplesData *rdt_data) rdt_data->table_sync_wait_time)) return false; - rdt_data->phase = RDT_STOP_CONFLICT_INFO_RETENTION; + /* Stop retention if not yet */ + if (MySubscription->retentionactive) + { + rdt_data->phase = RDT_STOP_CONFLICT_INFO_RETENTION; - /* process the next phase */ - process_rdt_phase_transition(rdt_data, false); + /* process the next phase */ + process_rdt_phase_transition(rdt_data, false); + } + + reset_retention_data_fields(rdt_data); return true; } @@ -4725,6 +4766,131 @@ should_stop_conflict_info_retention(RetainDeadTuplesData *rdt_data) */ static void stop_conflict_info_retention(RetainDeadTuplesData *rdt_data) +{ + update_retention_status(false); + + SpinLockAcquire(&MyLogicalRepWorker->relmutex); + MyLogicalRepWorker->oldest_nonremovable_xid = InvalidTransactionId; + SpinLockRelease(&MyLogicalRepWorker->relmutex); + + ereport(LOG, + errmsg("logical replication worker for subscription \"%s\" has stopped retaining the information for detecting conflicts", + MySubscription->name), + errdetail("Retention of information used for conflict detection has exceeded max_retention_duration of %u ms.", + MySubscription->maxretention)); + + reset_retention_data_fields(rdt_data); + + /* process the next phase */ + process_rdt_phase_transition(rdt_data, false); +} + +/* + * Check whether retention should be resumed immediately if it has been + * previously stopped, but max_retention_duration is now set to 0. + */ +static bool +should_resume_retention_immediately(RetainDeadTuplesData *rdt_data, bool status_received) +{ + /* Return false if retention is already being resumed */ + if (rdt_data->phase == RDT_RESUME_CONFLICT_INFO_RETENTION) + return false; + + /* Return false if max_retention_duration is not 0 */ + if (MySubscription->maxretention) + return false; + + /* + * Do not resume when waiting for publisher status, as doing so may result + * in the message being processed after the data and phase have been + * reset, potentially causing it to be mistakenly identified as a new + * message. This could lead to the premature advancement of + * oldest_nonremovable_xid. + */ + if (rdt_data->phase == RDT_WAIT_FOR_PUBLISHER_STATUS && + !status_received) + return false; + + /* + * Resume retention if we are in the process of stopping or have already + * stopped retention. + */ + return rdt_data->phase == RDT_STOP_CONFLICT_INFO_RETENTION || + !MySubscription->retentionactive; +} + +/* + * Workhorse for the RDT_RESUME_CONFLICT_INFO_RETENTION phase. + */ +static void +resume_conflict_info_retention(RetainDeadTuplesData *rdt_data) +{ + TransactionId nonremovable_xid; + + /* Update the pg_subscription.retentionactive if not yet */ + if (!MySubscription->retentionactive) + { + update_retention_status(true); + + SpinLockAcquire(&MyLogicalRepWorker->relmutex); + MyLogicalRepWorker->wait_for_initial_xid = true; + SpinLockRelease(&MyLogicalRepWorker->relmutex); + + ereport(LOG, + errmsg("logical replication worker for subscription \"%s\" will resume retaining the information for detecting conflicts", + MySubscription->name), + MySubscription->maxretention + ? errdetail("Retention of information used for conflict detection is now within the max_retention_duration of %u ms.", + MySubscription->maxretention) + : errdetail("Retention of information used for conflict detection is now indefinite.")); + } + + SpinLockAcquire(&MyLogicalRepWorker->relmutex); + nonremovable_xid = MyLogicalRepWorker->oldest_nonremovable_xid; + SpinLockRelease(&MyLogicalRepWorker->relmutex); + + /* + * Return if the launcher has not initialized oldest_nonremovable_xid. + * + * It might seem feasible to directly check the conflict detection + * slot.xmin instead of relying on the launcher to assign the worker's + * oldest_nonremovable_xid; however, that could lead to a race condition + * where slot.xmin is set to InvalidTransactionId immediately after the + * check. In such cases, oldest_nonremovable_xid would no longer be + * protected by a replication slot and could become unreliable if a + * wraparound occurs. + */ + if (!TransactionIdIsValid(nonremovable_xid)) + return; + + SpinLockAcquire(&MyLogicalRepWorker->relmutex); + MyLogicalRepWorker->wait_for_initial_xid = false; + SpinLockRelease(&MyLogicalRepWorker->relmutex); + + /* + * Proceed to the next phase if either the launcher has initialized + * slot.xmin and assigned it to oldest_nonremovable_xid, or retention has + * not been stopped yet. The latter situation arises when transitioning + * from the RDT_STOP_CONFLICT_INFO_RETENTION phase but subretentionactive + * has not been updated due to the inability to start a new transaction + * (see stop_conflict_info_retention). + */ + Assert(MySubscription->retentionactive); + + reset_retention_data_fields(rdt_data); + + /* process the next phase */ + process_rdt_phase_transition(rdt_data, false); +} + +/* + * Update pg_subscription.subretentionactive to the given value within a new + * transaction. + * + * Skip the update if currently within an existing transaction. + */ +static void +update_retention_status(bool active) { /* * Do not update the catalog during an active transaction. The transaction @@ -4742,26 +4908,16 @@ stop_conflict_info_retention(RetainDeadTuplesData *rdt_data) */ PushActiveSnapshot(GetTransactionSnapshot()); - /* Set pg_subscription.subretentionactive to false */ - UpdateDeadTupleRetentionStatus(MySubscription->oid, false); + /* Update pg_subscription.subretentionactive */ + UpdateDeadTupleRetentionStatus(MySubscription->oid, active); PopActiveSnapshot(); CommitTransactionCommand(); - SpinLockAcquire(&MyLogicalRepWorker->relmutex); - MyLogicalRepWorker->oldest_nonremovable_xid = InvalidTransactionId; - SpinLockRelease(&MyLogicalRepWorker->relmutex); - - ereport(LOG, - errmsg("logical replication worker for subscription \"%s\" has stopped retaining the information for detecting conflicts", - MySubscription->name), - errdetail("Retention of information used for conflict detection has exceeded max_retention_duration of %u ms.", - MySubscription->maxretention)); - - /* Notify launcher to update the conflict slot */ + /* Notify launcher to update the xmin of the conflict slot */ ApplyLauncherWakeup(); - reset_retention_data_fields(rdt_data); + MySubscription->retentionactive = active; } /* diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 62ea1a00580..58f2dcc7f0b 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -100,6 +100,12 @@ typedef struct LogicalRepWorker */ TransactionId oldest_nonremovable_xid; + /* + * Indicates whether the apply worker is resuming retention and is waiting + * for the launcher to initialize oldest_nonremovable_xid. + */ + bool wait_for_initial_xid; + /* Stats. */ XLogRecPtr last_lsn; TimestampTz last_send_time; -- 2.51.0.windows.1