From d7b8f5f07d81a714bbe37ab1f1986aa594735fac Mon Sep 17 00:00:00 2001
From: Amit Kapila <akapila@postgresql.org>
Date: Mon, 27 Jul 2026 09:47:10 +0530
Subject: [PATCH v64_0002_houzj] Detect parallel apply worker exit via its
 worker slot.

Instead of the worker setting PARALLEL_TRANS_ERROR before aborting, the
leader now waits in pa_wait_for_xact_finish() for the worker to either report
its error or exit, detecting exit via its LogicalRepWorker slot. This drops
the worker-side flag and its spinlock from the error path.
---
 .../replication/logical/applyparallelworker.c | 50 ++++---------------
 src/backend/replication/logical/launcher.c    | 29 +++++++++++
 src/include/replication/worker_internal.h     |  4 +-
 3 files changed, 39 insertions(+), 44 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 55a0ee4831a..ef218117ff3 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -1018,16 +1018,6 @@ ParallelApplyWorkerMain(Datum main_arg)
 		FlushErrorState();
 		error_context_stack = NULL;
 
-		/*
-		 * Tell the leader we failed and are about to report the error and log
-		 * the conflict.  This must be set before AbortOutOfAnyTransaction()
-		 * below releases the transaction lock that the leader waits on in
-		 * pa_wait_for_xact_finish(); otherwise the leader would see a
-		 * non-finished state, assume the connection was lost, and tear this
-		 * worker down while it is still writing the conflict log tuple.
-		 */
-		pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_ERROR);
-
 		AbortOutOfAnyTransaction();
 
 		/*
@@ -1342,6 +1332,11 @@ pa_wait_for_xact_state(ParallelApplyWorkerInfo *winfo,
 
 		/* An interrupt may have occurred while we were waiting. */
 		CHECK_FOR_INTERRUPTS();
+
+		if (!logicalrep_pa_worker_running(winfo))
+			ereport(ERROR,
+					(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+					 errmsg("lost connection to the logical replication parallel apply worker")));
 	}
 }
 
@@ -1368,38 +1363,11 @@ pa_wait_for_xact_finish(ParallelApplyWorkerInfo *winfo)
 	pa_unlock_transaction(winfo->shared->xid, AccessShareLock);
 
 	/*
-	 * Check if the state becomes PARALLEL_TRANS_FINISHED in case the parallel
-	 * apply worker failed while applying changes causing the lock to be
-	 * released.
+	 * Wait for the transaction state to reach PARALLEL_TRANS_FINISHED. The wait
+	 * function handles the case where the parallel apply worker errors out
+	 * before updating the state.
 	 */
-	if (pa_get_xact_state(winfo->shared) != PARALLEL_TRANS_FINISHED)
-	{
-		/*
-		 * If the worker signalled that it errored (PARALLEL_TRANS_ERROR), it
-		 * is logging the conflict and will report the actual error via the
-		 * error queue before exiting.  Wait for that rather than reporting a
-		 * generic lost connection: CHECK_FOR_INTERRUPTS() drives
-		 * ProcessParallelApplyMessages(), which raises the real error on the
-		 * worker's ErrorResponse (or "lost connection" if the worker died
-		 * without reporting).  Waiting here also keeps the worker alive long
-		 * enough to finish writing the conflict log tuple.
-		 */
-		while (pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_ERROR)
-		{
-			CHECK_FOR_INTERRUPTS();
-
-			(void) WaitLatch(MyLatch,
-							 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
-							 10L,
-							 WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE);
-
-			ResetLatch(MyLatch);
-		}
-
-		ereport(ERROR,
-				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-				 errmsg("lost connection to the logical replication parallel apply worker")));
-	}
+	pa_wait_for_xact_state(winfo, PARALLEL_TRANS_FINISHED);
 }
 
 /*
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 313e31ff2e3..71fb3321d3e 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -725,6 +725,35 @@ logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
 	LWLockRelease(LogicalRepWorkerLock);
 }
 
+/*
+ * Is the given parallel apply worker still running?
+ *
+ * The generation guards against the slot having been reused by a different
+ * worker after ours exited.
+ */
+bool
+logicalrep_pa_worker_running(ParallelApplyWorkerInfo *winfo)
+{
+	int			slot_no;
+	uint16		generation;
+	LogicalRepWorker *worker;
+	bool		running;
+
+	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);
+
+	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+	worker = &LogicalRepCtx->workers[slot_no];
+	running = (worker->generation == generation && worker->proc != NULL);
+	LWLockRelease(LogicalRepWorkerLock);
+
+	return running;
+}
+
 /*
  * Wake up (using latch) any logical replication worker that matches the
  * specified worker type, subscription id, and relation id.
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 394f4c6265e..1a74a366a44 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -121,9 +121,6 @@ typedef enum ParallelTransState
 	PARALLEL_TRANS_UNKNOWN,
 	PARALLEL_TRANS_STARTED,
 	PARALLEL_TRANS_FINISHED,
-	PARALLEL_TRANS_ERROR,		/* worker failed; it will report the error
-								 * (and log the conflict, if any) before
-								 * exiting */
 } ParallelTransState;
 
 /*
@@ -276,6 +273,7 @@ extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
 extern void logicalrep_worker_stop(LogicalRepWorkerType wtype, Oid subid,
 								   Oid relid);
 extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
+extern bool logicalrep_pa_worker_running(ParallelApplyWorkerInfo *winfo);
 extern void logicalrep_worker_wakeup(LogicalRepWorkerType wtype, Oid subid,
 									 Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
-- 
2.43.0

