From 8c7db9ef12b1cb1befc839a3cff1353bb5a35d92 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Rafael=20Chac=C3=B3n?= Date: Mon, 9 Nov 2026 09:57:04 -0500 Subject: [PATCH] Make SYNCDONE durable before slot drop A tablesync worker previously set SYNCDONE and dropped its replication slot on the publisher before it committed locally. A subscriber crash after the slot drop left the table in FINISHEDCOPY in pg_subscription_rel. The restarted worker then repeatedly failed with "replication slot ... does not exist". The tablesync worker now commits and flushes SYNCDONE before it drops the slot. This makes SYNCDONE durable across a subscriber crash. If a worker restarts in SYNCDONE, it removes any remaining slot and origin before it exits. It accepts a missing slot or origin during this cleanup. The replication origin tracks replication progress on the subscriber. The tablesync worker already removed this origin after it dropped the slot. The apply worker previously provided fallback cleanup before it marked the table READY. The patch removes that fallback. Once the apply worker reaches the synchronization position, it treats a remaining origin as unfinished cleanup and keeps the table in SYNCDONE. If no tablesync worker exists, it requests another worker under the existing retry and worker limits. The tablesync worker drops the slot before it removes the origin. The apply worker marks the table READY only after it reaches the synchronization position and the origin is absent. The apply worker now leaves subscription and catalog lock acquisition to UpdateSubscriptionRelState(). The function still acquires those locks when it updates the table state. DROP SUBSCRIPTION now also attempts to remove tablesync slots for SYNCDONE tables. ALTER SUBSCRIPTION ... REFRESH does the same for tables that it removes from the subscription. The apply worker no longer copies the tablesync worker's state or replication position from shared memory into its cached state. It accepts SYNCDONE only from the catalog, where the committed state survives a crash. --- src/backend/commands/subscriptioncmds.c | 45 +++---- src/backend/replication/logical/tablesync.c | 128 ++++++++++---------- 2 files changed, 79 insertions(+), 94 deletions(-) diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index e91a4f2d911..d59421898c6 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1303,10 +1303,10 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, * * It is possible that the origin is not yet created for * tablesync worker, this can happen for the states before - * SUBREL_STATE_DATASYNC. The tablesync worker or apply - * worker can also concurrently try to drop the origin and - * by this time the origin might be already removed. For - * these reasons, passing missing_ok = true. + * SUBREL_STATE_DATASYNC. The tablesync worker can also + * concurrently try to drop the origin and by this time the + * origin might be already removed. For these reasons, + * passing missing_ok = true. */ ReplicationOriginNameForLogicalRep(sub->oid, relid, originname, sizeof(originname)); @@ -1362,14 +1362,13 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, */ foreach_ptr(SubRemoveRels, sub_remove_rel, sub_remove_rels) { - if (sub_remove_rel->state != SUBREL_STATE_READY && - sub_remove_rel->state != SUBREL_STATE_SYNCDONE) + if (sub_remove_rel->state != SUBREL_STATE_READY) { char syncslotname[NAMEDATALEN] = {0}; /* - * For READY/SYNCDONE states we know the tablesync slot has - * already been dropped by the tablesync worker. + * For READY the tablesync worker already dropped the slot. + * SYNCDONE can still need cleanup after a worker failure. * * For other states, there is no certainty, maybe the slot * does not exist yet. Also, if we fail after removing some of @@ -2863,6 +2862,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) { SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); Oid relid = rstate->relid; + char syncslotname[NAMEDATALEN] = {0}; /* Only cleanup resources of tablesync workers */ if (!OidIsValid(relid)) @@ -2871,8 +2871,8 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) /* * Drop the tablesync slots associated with removed tables. * - * For SYNCDONE/READY states, the tablesync slot is known to have - * already been dropped by the tablesync worker. + * READY relations are not included in rstates. SYNCDONE can + * still have a slot awaiting cleanup by the tablesync worker. * * For other states, there is no certainty, maybe the slot does * not exist yet. Also, if we fail after removing some of the @@ -2880,14 +2880,9 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) * slots and fail. For these reasons, we allow missing_ok = true * for the drop. */ - if (rstate->state != SUBREL_STATE_SYNCDONE) - { - char syncslotname[NAMEDATALEN] = {0}; - - ReplicationSlotNameForTablesync(subid, relid, syncslotname, - sizeof(syncslotname)); - ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); - } + ReplicationSlotNameForTablesync(subid, relid, syncslotname, + sizeof(syncslotname)); + ReplicationSlotDropAtPubNode(wrconn, syncslotname, true); } list_free(rstates); @@ -3696,6 +3691,7 @@ ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err) { SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); Oid relid = rstate->relid; + char syncslotname[NAMEDATALEN] = {0}; /* Only cleanup resources of tablesync workers */ if (!OidIsValid(relid)) @@ -3705,15 +3701,10 @@ ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err) * Caller needs to ensure that relstate doesn't change underneath us. * See DropSubscription where we get the relstates. */ - if (rstate->state != SUBREL_STATE_SYNCDONE) - { - char syncslotname[NAMEDATALEN] = {0}; - - ReplicationSlotNameForTablesync(subid, relid, syncslotname, - sizeof(syncslotname)); - elog(WARNING, "could not drop tablesync replication slot \"%s\"", - syncslotname); - } + ReplicationSlotNameForTablesync(subid, relid, syncslotname, + sizeof(syncslotname)); + elog(WARNING, "could not drop tablesync replication slot \"%s\"", + syncslotname); } ereport(ERROR, diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 4015e861a64..eb63a11f25c 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -277,12 +277,17 @@ ProcessSyncingTablesForSync(XLogRecPtr current_lsn) */ walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli); + /* + * Dropping the slot cannot be rolled back, so SYNCDONE must be durable + * first. Otherwise a crash after the drop leaves FINISHEDCOPY with no + * slot to resume from. + */ + ForceSyncCommit(); + CommitTransactionCommand(); + pgstat_report_stat(false); + /* * Cleanup the tablesync slot. - * - * This has to be done after updating the state because otherwise if - * there is an error while doing the database operations we won't be - * able to rollback dropped slot. */ ReplicationSlotNameForTablesync(MyLogicalRepWorker->subid, MyLogicalRepWorker->relid, @@ -296,14 +301,11 @@ ProcessSyncingTablesForSync(XLogRecPtr current_lsn) */ ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, syncslotname, false); - CommitTransactionCommand(); - pgstat_report_stat(false); - /* * Start a new transaction to clean up the tablesync origin tracking. - * This transaction will be ended within the FinishSyncWorker(). Now, - * even, if we fail to remove this here, the apply worker will ensure - * to clean it up afterward. + * This transaction will be ended within the FinishSyncWorker(). If + * we fail before removing it, the origin tells the apply worker to + * restart this worker to finish the cleanup. * * We need to do this after the table state is set to SYNCDONE. * Otherwise, if an error occurs while performing the database @@ -331,9 +333,8 @@ ProcessSyncingTablesForSync(XLogRecPtr current_lsn) * Drop the tablesync's origin tracking if exists. * * There is a chance that the user is concurrently performing refresh - * for the subscription where we remove the table state and its origin - * or the apply worker would have removed this origin. So passing - * missing_ok = true. + * for the subscription where we remove the table state and its origin. + * So passing missing_ok = true. */ replorigin_drop_by_name(originname, true, false); @@ -376,7 +377,6 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn) ListCell *lc; bool started_tx; bool should_exit = false; - Relation rel = NULL; Assert(!IsTransactionState()); @@ -414,6 +414,7 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn) foreach(lc, table_states_not_ready) { SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); + bool cleanup_pending = false; if (!started_tx) { @@ -423,7 +424,26 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn) Assert(get_rel_relkind(rstate->relid) != RELKIND_SEQUENCE); - if (rstate->state == SUBREL_STATE_SYNCDONE) + /* + * The tablesync worker drops its origin last, so a remaining origin + * means that source slot cleanup may be unfinished. Only mark the + * table READY once no cleanup is pending. Otherwise, treat the table + * like one still being synchronized, so that a tablesync worker is + * started to finish the cleanup. + */ + if (rstate->state == SUBREL_STATE_SYNCDONE && + current_lsn >= rstate->lsn) + { + char originname[NAMEDATALEN]; + + ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid, + rstate->relid, + originname, + sizeof(originname)); + cleanup_pending = OidIsValid(replorigin_by_name(originname, true)); + } + + if (rstate->state == SUBREL_STATE_SYNCDONE && !cleanup_pending) { /* * Apply has caught up to the position where the table sync has @@ -432,44 +452,11 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn) */ if (current_lsn >= rstate->lsn) { - char originname[NAMEDATALEN]; - rstate->state = SUBREL_STATE_READY; rstate->lsn = current_lsn; - - /* - * Remove the tablesync origin tracking if exists. - * - * There is a chance that the user is concurrently performing - * refresh for the subscription where we remove the table - * state and its origin or the tablesync worker would have - * already removed this origin. We can't rely on tablesync - * worker to remove the origin tracking as if there is any - * error while dropping we won't restart it to drop the - * origin. So passing missing_ok = true. - * - * Lock the subscription and origin in the same order as we - * are doing during DDL commands to avoid deadlocks. See - * AlterSubscription_refresh. - */ - LockSharedObject(SubscriptionRelationId, MyLogicalRepWorker->subid, - 0, AccessShareLock); - - if (!rel) - rel = table_open(SubscriptionRelRelationId, RowExclusiveLock); - - ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid, - rstate->relid, - originname, - sizeof(originname)); - replorigin_drop_by_name(originname, true, false); - - /* - * Update the state to READY only after the origin cleanup. - */ UpdateSubscriptionRelState(MyLogicalRepWorker->subid, rstate->relid, rstate->state, - rstate->lsn, true); + rstate->lsn, false); } } else @@ -487,11 +474,15 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn) if (syncworker) { - /* Found one, update our copy of its state */ + char syncstate; + + /* + * Found one, read its state. Keep rstate as is: SYNCDONE is + * only taken from the catalog, where it is durable. + */ SpinLockAcquire(&syncworker->relmutex); - rstate->state = syncworker->relstate; - rstate->lsn = syncworker->relstate_lsn; - if (rstate->state == SUBREL_STATE_SYNCWAIT) + syncstate = syncworker->relstate; + if (syncstate == SUBREL_STATE_SYNCWAIT) { /* * Sync worker is waiting for apply. Tell sync worker it @@ -504,7 +495,7 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn) SpinLockRelease(&syncworker->relmutex); /* If we told worker to catch up, wait for it. */ - if (rstate->state == SUBREL_STATE_SYNCWAIT) + if (syncstate == SUBREL_STATE_SYNCWAIT) { /* Signal the sync worker, as it may be waiting for us. */ if (syncworker->proc) @@ -521,14 +512,7 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn) * This is required to avoid any undetected deadlocks * due to any existing lock as deadlock detector won't * be able to detect the waits on the latch. - * - * Also close any tables prior to the commit. */ - if (rel) - { - table_close(rel, NoLock); - rel = NULL; - } CommitTransactionCommand(); pgstat_report_stat(false); } @@ -572,11 +556,6 @@ ProcessSyncingTablesForApply(XLogRecPtr current_lsn) } } - /* Close table if opened */ - if (rel) - table_close(rel, NoLock); - - if (started_tx) { /* @@ -1291,7 +1270,6 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos) */ switch (relstate) { - case SUBREL_STATE_SYNCDONE: case SUBREL_STATE_READY: case SUBREL_STATE_UNKNOWN: FinishSyncWorker(); /* doesn't return */ @@ -1319,6 +1297,22 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos) errmsg("table synchronization worker for subscription \"%s\" could not connect to the publisher: %s", MySubscription->name, err))); + /* + * A previous worker failed after committing SYNCDONE, so only the source + * slot and the origin remain to be cleaned up. + */ + if (relstate == SUBREL_STATE_SYNCDONE) + { + ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, slotname, true); + + StartTransactionCommand(); + ReplicationOriginNameForLogicalRep(MySubscription->oid, + MyLogicalRepWorker->relid, + originname, sizeof(originname)); + replorigin_drop_by_name(originname, true, false); + FinishSyncWorker(); /* doesn't return */ + } + Assert(MyLogicalRepWorker->relstate == SUBREL_STATE_INIT || MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC || MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY); -- 2.54.0 (Apple Git-157)