Drop replication origen slots before tablesync worker exits.

Amit Kapila · Amit Kapila · commit f6c5edb8abca · 2022-08-30T08:51:41.000+05:30
Currently, the replication origen tracking of the tablesync worker is dropped by the apply worker. So, there will be a small lag between the tablesync worker exit and its origen tracking got removed. In the meantime, new tablesync workers can be launched and will try to set up a new origen tracking. This can lead the system to reach max configured limit (max_replication_slots) even if the user has configured the max limit considering the number of tablesync workers required in the system. We decided not to back-patch as this can occur in very narrow circumstances and users have to option to increase the configured limit by increasing max_replication_slots. Reported-by: Hubert Depesz Lubaczewski Author: Ajin Cherian Reviwed-by: Masahiko Sawada, Peter Smith, Hou Zhijie, Amit Kapila Discussion: https://postgr.es/m/20220714115155.GA5439@depesz.com
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
@@ -919,22 +919,19 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 				logicalrep_worker_stop(sub->oid, relid);
 
 				/*
-				 * For READY state, we would have already dropped the
-				 * tablesync origen.
+				 * For READY state and SYNCDONE state, we would have already
+				 * dropped the tablesync origen.
 				 */
-				if (state != SUBREL_STATE_READY)
+				if (state != SUBREL_STATE_READY && state != SUBREL_STATE_SYNCDONE)
 				{
 					char		origenname[NAMEDATALEN];
 
 					/*
 					 * Drop the tablesync's origen tracking if exists.
 					 *
 					 * It is possible that the origen is not yet created for
-					 * tablesync worker, this can happen for the states before
-					 * SUBREL_STATE_FINISHEDCOPY. The apply worker can also
-					 * concurrently try to drop the origen and by this time
-					 * the origen might be already removed. For these reasons,
-					 * passing missing_ok = true.
+					 * tablesync worker so passing missing_ok = true. This can
+					 * happen for the states before SUBREL_STATE_FINISHEDCOPY.
 					 */
 					ReplicationOriginNameForTablesync(sub->oid, relid, origenname,
 													  sizeof(origenname));
@@ -1507,13 +1504,19 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 		/*
 		 * Drop the tablesync's origen tracking if exists.
 		 *
+		 * For SYNCDONE/READY states, the tablesync origen tracking is known
+		 * to have already been dropped by the tablesync worker.
+		 *
 		 * It is possible that the origen is not yet created for tablesync
 		 * worker so passing missing_ok = true. This can happen for the states
 		 * before SUBREL_STATE_FINISHEDCOPY.
 		 */
-		ReplicationOriginNameForTablesync(subid, relid, origenname,
-										  sizeof(origenname));
-		replorigen_drop_by_name(origenname, true, false);
+		if (rstate->state != SUBREL_STATE_SYNCDONE)
+		{
+			ReplicationOriginNameForTablesync(subid, relid, origenname,
+											  sizeof(origenname));
+			replorigen_drop_by_name(origenname, true, false);
+		}
 	}
 
 	/* Clean up dependencies */
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
@@ -291,6 +291,7 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
 	{
 		TimeLineID	tli;
 		char		syncslotname[NAMEDATALEN] = {0};
+		char		origenname[NAMEDATALEN] = {0};
 
 		MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCDONE;
 		MyLogicalRepWorker->relstate_lsn = current_lsn;
@@ -309,6 +310,30 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
 								   MyLogicalRepWorker->relstate,
 								   MyLogicalRepWorker->relstate_lsn);
 
+		/*
+		 * Cleanup the tablesync origen tracking.
+		 *
+		 * Resetting the origen session removes the ownership of the slot.
+		 * This is needed to allow the origen to be dropped.
+		 */
+		ReplicationOriginNameForTablesync(MyLogicalRepWorker->subid,
+										  MyLogicalRepWorker->relid,
+										  origenname,
+										  sizeof(origenname));
+		replorigen_session_reset();
+		replorigen_session_origen = InvalidRepOriginId;
+		replorigen_session_origen_lsn = InvalidXLogRecPtr;
+		replorigen_session_origen_timestamp = 0;
+
+		/*
+		 * We expect that origen must be present. The concurrent operations
+		 * that remove origen like a refresh for the subscription take an
+		 * access exclusive lock on pg_subscription which prevent the previous
+		 * operation to update the rel state to SUBREL_STATE_SYNCDONE to
+		 * succeed.
+		 */
+		replorigen_drop_by_name(origenname, false, false);
+
 		/*
 		 * End streaming so that LogRepWorkerWalRcvConn can be used to drop
 		 * the slot.
@@ -318,7 +343,7 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
 		/*
 		 * Cleanup the tablesync slot.
 		 *
-		 * This has to be done after updating the state because otherwise if
+		 * This has to be done after the data changes because otherwise if
 		 * there is an error while doing the database operations we won't be
 		 * able to rollback dropped slot.
 		 */
@@ -441,8 +466,6 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 			 */
 			if (current_lsn >= rstate->lsn)
 			{
-				char		origenname[NAMEDATALEN];
-
 				rstate->state = SUBREL_STATE_READY;
 				rstate->lsn = current_lsn;
 				if (!started_tx)
@@ -452,26 +475,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 				}
 
 				/*
-				 * Remove the tablesync origen tracking if exists.
-				 *
-				 * The normal case origen drop is done here instead of in the
-				 * process_syncing_tables_for_sync function because we don't
-				 * allow to drop the origen till the process owning the origen
-				 * is alive.
-				 *
-				 * There is a chance that the user is concurrently performing
-				 * refresh for the subscription where we remove the table
-				 * state and its origen and by this time the origen might be
-				 * already removed. So passing missing_ok = true.
-				 */
-				ReplicationOriginNameForTablesync(MyLogicalRepWorker->subid,
-												  rstate->relid,
-												  origenname,
-												  sizeof(origenname));
-				replorigen_drop_by_name(origenname, true, false);
-
-				/*
-				 * Update the state to READY only after the origen cleanup.
+				 * Update the state to READY.
 				 */
 				UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
 										   rstate->relid, rstate->state,
