0001-Detach-the-error-queue-before-stopping-parallel-appl.patch

From c188075f07e5dcfa53a77c92b25913af31622e54 Mon Sep 17 00:00:00 2001
From: sherlockcpp <sherlockcpp@foxmail.com>
Date: Sun, 30 Apr 2023 20:58:42 +0800
Subject: [PATCH 1/2] Detach the error queue before stopping parallel apply
 worker

Detach from the error_mq_handle for the parallel apply worker before
terminating it. This prevents the leader apply worker from receiving the
worker termination message and sending it to logs when the same is
already done by the parallel worker.
---
 .../replication/logical/applyparallelworker.c | 11 +---------
 src/backend/replication/logical/launcher.c    | 21 ++++++++++++++++---
 src/include/replication/worker_internal.h     |  2 +-
 3 files changed, 20 insertions(+), 14 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index ee7a18137f..c79765ce20 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -577,16 +577,7 @@ pa_free_worker(ParallelApplyWorkerInfo *winfo)
 		list_length(ParallelApplyWorkerPool) >
 		(max_parallel_apply_workers_per_subscription / 2))
 	{
-		int			slot_no;
-		uint16		generation;
-
-		SpinLockAcquire(&winfo->shared->mutex);
-		generation = winfo->shared->logicalrep_worker_generation;
-		slot_no = winfo->shared->logicalrep_worker_slot_no;
-		SpinLockRelease(&winfo->shared->mutex);
-
-		logicalrep_pa_worker_stop(slot_no, generation);
-
+		logicalrep_pa_worker_stop(winfo);
 		pa_free_worker_info(winfo);
 
 		return;
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index ceea126231..ae107a2184 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -609,19 +609,34 @@ logicalrep_worker_stop(Oid subid, Oid relid)
 }
 
 /*
- * Stop the logical replication parallel apply worker corresponding to the
- * input slot number.
+ * Stop the given logical replication parallel apply worker.
  *
  * Node that the function sends SIGINT instead of SIGTERM to the parallel apply
  * worker so that the worker exits cleanly.
  */
 void
-logicalrep_pa_worker_stop(int slot_no, uint16 generation)
+logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
 {
+	int		slot_no;
+	uint16	generation;
 	LogicalRepWorker *worker;
 
+	SpinLockAcquire(&winfo->shared->mutex);
+	generation = winfo->shared->logicalrep_worker_generation;
+	slot_no = winfo->shared->logicalrep_worker_slot_no;
+	SpinLockRelease(&winfo->shared->mutex);
+
 	Assert(slot_no >= 0 && slot_no < max_logical_replication_workers);
 
+	/*
+	 * Detach from the error_mq_handle for the parallel apply worker before
+	 * terminating it. This prevents the leader apply worker from receiving the
+	 * worker termination message and sending it to logs when the same is
+	 * already done by the parallel worker.
+	 */
+	shm_mq_detach(winfo->error_mq_handle);
+	winfo->error_mq_handle = NULL;
+
 	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
 	worker = &LogicalRepCtx->workers[slot_no];
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index b57eed052f..343e781896 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -235,7 +235,7 @@ extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
 									 Oid userid, Oid relid,
 									 dsm_handle subworker_dsm);
 extern void logicalrep_worker_stop(Oid subid, Oid relid);
-extern void logicalrep_pa_worker_stop(int slot_no, uint16 generation);
+extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
 extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
 
-- 
2.30.0.windows.2