Make the tablesync worker's replication origen drop logic robust.

Amit Kapila · Amit Kapila · commit 88f488319bac · 2022-09-12T12:40:57.000+05:30
In commit f6c5edb, we started to drop the replication origen slots before tablesync worker exits to avoid consuming more slots than required. We were dropping the replication origen in the same transaction where we were marking the tablesync state as SYNCDONE. Now, if there is any error after we have dropped the origen but before we commit the containing transaction, the in-memory state of replication progress won't be rolled back. Due to this, after the restart, tablesync worker can start streaming from the wrong location and can apply the already processed transaction. To fix this, we need to opportunistically drop the origen after marking the tablesync state as SYNCDONE. Even, if the tablesync worker fails to remove the replication origen before exit, the apply worker ensures to clean it up afterward. Reported by Tom Lane as per buildfarm. Diagnosed-by: Masahiko Sawada Author: Hou Zhijie Reviewed-By: Masahiko Sawada, Amit Kapila Discussion: https://postgr.es/m/20220714115155.GA5439@depesz.com Discussion: https://postgr.es/m/CAD21AoAw0Oofi4kiDpJBOwpYyBBBkJj=sLUOn4Gd2GjUAKG-fw@mail.gmail.com
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
@@ -931,19 +931,22 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 				logicalrep_worker_stop(sub->oid, relid);
 
 				/*
-				 * For READY state and SYNCDONE state, we would have already
-				 * dropped the tablesync origen.
+				 * For READY state, we would have already dropped the
+				 * tablesync origen.
 				 */
-				if (state != SUBREL_STATE_READY && state != SUBREL_STATE_SYNCDONE)
+				if (state != SUBREL_STATE_READY)
 				{
 					char		origenname[NAMEDATALEN];
 
 					/*
 					 * Drop the tablesync's origen tracking if exists.
 					 *
 					 * 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.
+					 * tablesync worker, this can happen for the states before
+					 * SUBREL_STATE_FINISHEDCOPY. The tablesync worker or
+					 * 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.
 					 */
 					ReplicationOriginNameForTablesync(sub->oid, relid, origenname,
 													  sizeof(origenname));
@@ -1516,19 +1519,13 @@ 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.
 		 */
-		if (rstate->state != SUBREL_STATE_SYNCDONE)
-		{
-			ReplicationOriginNameForTablesync(subid, relid, origenname,
-											  sizeof(origenname));
-			replorigen_drop_by_name(origenname, true, false);
-		}
+		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
@@ -300,7 +300,6 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
 
 		/*
 		 * UpdateSubscriptionRelState must be called within a transaction.
-		 * That transaction will be ended within the finish_sync_worker().
 		 */
 		if (!IsTransactionState())
 			StartTransactionCommand();
@@ -310,30 +309,6 @@ 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.
@@ -343,7 +318,7 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
 		/*
 		 * Cleanup the tablesync slot.
 		 *
-		 * This has to be done after the data changes because otherwise if
+		 * 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.
 		 */
@@ -359,6 +334,49 @@ 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 origen 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.
+		 *
+		 * We need to do this after the table state is set to SYNCDONE.
+		 * Otherwise, if an error occurs while performing the database
+		 * operation, the worker will be restarted and the in-memory state of
+		 * replication progress (remote_lsn) won't be rolled-back which would
+		 * have been cleared before restart. So, the restarted worker will use
+		 * invalid replication progress state resulting in replay of
+		 * transactions that have already been applied.
+		 */
+		StartTransactionCommand();
+
+		ReplicationOriginNameForTablesync(MyLogicalRepWorker->subid,
+										  MyLogicalRepWorker->relid,
+										  origenname,
+										  sizeof(origenname));
+
+		/*
+		 * Resetting the origen session removes the ownership of the slot.
+		 * This is needed to allow the origen to be dropped.
+		 */
+		replorigen_session_reset();
+		replorigen_session_origen = InvalidRepOriginId;
+		replorigen_session_origen_lsn = InvalidXLogRecPtr;
+		replorigen_session_origen_timestamp = 0;
+
+		/*
+		 * Drop the tablesync's origen 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 origen
+		 * or the apply worker would have removed this origen. So passing
+		 * missing_ok = true.
+		 */
+		replorigen_drop_by_name(origenname, true, false);
+
 		finish_sync_worker();
 	}
 	else
@@ -466,6 +484,8 @@ 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)
@@ -475,7 +495,24 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
 				}
 
 				/*
-				 * Update the state to READY.
+				 * Remove the tablesync origen 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 origen or the tablesync worker would have
+				 * already removed this origen. We can't rely on tablesync
+				 * worker to remove the origen tracking as if there is any
+				 * error while dropping we won't restart it to drop the
+				 * origen. 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.
 				 */
 				UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
 										   rstate->relid, rstate->state,
