From d7b8f5f07d81a714bbe37ab1f1986aa594735fac Mon Sep 17 00:00:00 2001 From: Amit Kapila 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