diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index e151cc8..8216579 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1073,14 +1073,13 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, */ for (off = 0; off < remove_rel_len; off++) { - if (sub_remove_rels[off].state != SUBREL_STATE_READY && - sub_remove_rels[off].state != SUBREL_STATE_SYNCDONE) + if (sub_remove_rels[off].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 @@ -1823,8 +1822,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 @@ -1832,7 +1831,6 @@ 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}; @@ -2299,7 +2297,6 @@ 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}; diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index b6bea66..d047838 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -325,12 +325,17 @@ process_syncing_tables_for_sync(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, @@ -344,14 +349,11 @@ process_syncing_tables_for_sync(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 finish_sync_worker(). - * Now, even, if we fail to remove this here, the apply worker will - * ensure to clean it up afterward. + * 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 @@ -381,9 +383,8 @@ process_syncing_tables_for_sync(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); @@ -464,8 +465,53 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) foreach(lc, table_states_not_ready) { SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); + bool cleanup_pending = false; + + /* + * The tablesync worker drops its origin last, so a remaining origin + * means that source slot cleanup may be unfinished. + */ + if (rstate->state == SUBREL_STATE_SYNCDONE && + current_lsn >= rstate->lsn) + { + char originname[NAMEDATALEN]; + + if (!started_tx) + { + StartTransactionCommand(); + started_tx = true; + } + + /* + * Lock the subscription and relation states 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); + + /* + * REFRESH may have removed or re-added this table while our cached + * state was stale. The locks keep it from changing again. + */ + if (GetSubscriptionRelState(MyLogicalRepWorker->subid, + rstate->relid, + &rstate->lsn) != SUBREL_STATE_SYNCDONE) + { + table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD; + continue; + } + + ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid, + rstate->relid, + originname, + sizeof(originname)); + cleanup_pending = OidIsValid(replorigin_by_name(originname, true)); + } - if (rstate->state == SUBREL_STATE_SYNCDONE) + if (rstate->state == SUBREL_STATE_SYNCDONE && !cleanup_pending) { /* * Apply has caught up to the position where the table sync has @@ -474,45 +520,8 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) */ if (current_lsn >= rstate->lsn) { - char originname[NAMEDATALEN]; - rstate->state = SUBREL_STATE_READY; rstate->lsn = current_lsn; - if (!started_tx) - { - StartTransactionCommand(); - started_tx = true; - } - - /* - * 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. - */ UpdateSubscriptionRelStateEx(MyLogicalRepWorker->subid, rstate->relid, rstate->state, rstate->lsn, true); @@ -532,10 +541,17 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) if (syncworker) { - /* Found one, update our copy of its state */ + /* + * Found one, update our copy of its state. 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_SYNCDONE && + syncworker->relstate != SUBREL_STATE_SYNCDONE) + { + rstate->state = syncworker->relstate; + rstate->lsn = syncworker->relstate_lsn; + } if (rstate->state == SUBREL_STATE_SYNCWAIT) { /* @@ -1344,7 +1360,6 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos) */ switch (relstate) { - case SUBREL_STATE_SYNCDONE: case SUBREL_STATE_READY: case SUBREL_STATE_UNKNOWN: finish_sync_worker(); /* doesn't return */ @@ -1371,6 +1386,22 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos) (errcode(ERRCODE_CONNECTION_FAILURE), errmsg("could not connect to the publisher: %s", 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); + finish_sync_worker(); /* doesn't return */ + } + Assert(MyLogicalRepWorker->relstate == SUBREL_STATE_INIT || MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC || MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY);