Hi,
On Fri, Sep 4, 2026 at 5:11 PM Masahiko Sawada <[email protected]>
wrote:
>
> I've reviewed the patch and I have two questions:
Thanks for reviewing it.
> + /*
> + * Associate the worker's handle with the error queue, just as if it
had
> + * been passed to shm_mq_attach(); we passed NULL there because the
worke
> + * did not exist yet. This lets ProcessRepackMessages() notice the
worker
> + * is gone instead of blocking on the queue.
> + */
> + shm_mq_set_handle(decoding_worker->error_mqh,
decoding_worker->handle);
> +
>
> The second sentence starting with "This lets ..." is unclear to me.
> ProcessRepackMessage() receives the message with nowait, no?
Yes, the read is nowait, and I get that the comment is misleading. All we
do here is connect the backend running concurrent repack (the receiver)
with the decoding worker (the sender) by saving the worker's handle to the
error message queue. We could not pass that handle to shm_mq_attach() above
because the sender did not exist at that point. The handle only matters
until the sender attaches to the queue. Until then, the receiver uses the
handle to check whether the sender is still alive, and reports detached if
the sender is already gone, instead of blocking on the queue. parallel.c
does the same for its workers.
> ---
> + if (status == BGWH_STOPPED)
> + ereport(ERROR,
> + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
> + errmsg("REPACK decoding worker failed to start"),
> + errhint("More details may be available in the
> server log."));
> + if (status == BGWH_POSTMASTER_DIED)
>
> parallel.c handles BGWH_STOPPED differently; it checks that the worker
> stopped without attaching to the error queue. IIUC if the worker
> stopped after attaching to the error queue, an error message should
> arrive at the leader and the leader could handle it in the next CFI.
> Is there any reason why start_repack_decoding_worker() handles it
> differently?
Right. There are two cases. The worker started and exited after attaching
to the error message queue, or it started and exited before attaching to
it. In the second case, throwing the generic error is correct, and in the
first the next CFI captures the worker's error. The attached v3 handles
both. Please have a look.
After thinking more about this, I simplified the handling. There are two
things the backend needs to wait for. First, for the worker to come up and
attach to the error message queue. Second, for the worker to set up the
logical decoding machinery before it waits for snapshot export. The first
wait catches fork failures and worker startup issues. I used similar logic
to what parallel.c uses for this. The second wait catches failures that
happen after the worker starts up and attaches to the error queue but
before it finishes setting up the logical decoding machinery. I kept the
shared memory initialized flag with the CV wait as-is for this.
Although the initialized flag wait may seem redundant with the snapshot
export wait in get_initial_snapshot(), I would still keep it because it
ensures the worker has fully set up the decoding before the backend
proceeds.
Dividing this into two separate waits (waiting for the worker to come up
and attach to the error message queue, and then waiting for it to finish
setup) makes the logic simpler, lets us reuse most of parallel.c's code, is
easier to reason about, and fixes the hang issue without letting the
backend reach the snapshot export wait with the worker not fully ready.
Please find the attached v3 patch.
--
Bharath Rupireddy
Amazon Web Services: https://aws.amazon.com
From acac900fe6ab2b56b989f9243b9673b618bde1f5 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Mon, 7 Sep 2026 12:14:16 +0000
Subject: [PATCH v3] Fix hang and deadlock in concurrent REPACK worker
handling.
REPACK CONCURRENTLY starts a background worker that decodes the
changes made to a table while it is being repacked, and the
backend running the command coordinates with that worker through
shared memory and an error message queue.
If the worker fails to start, for example when fork() fails, the
backend can wait for it forever. While waiting for the worker to
finish its setup, the backend sleeps on a CV, which only wakes on
its own signal. It therefore ignores the SIGUSR1 that the
postmaster sends via bgw_notify_pid when the worker fails to start
or exits, so the backend never notices and keeps waiting while
holding its lock on the table.
Fix this by waiting for the worker to attach to the error message
queue before waiting for it to initialize decoding. The backend
waits on its process latch and checks the worker's status on each
wakeup. If the worker stopped without attaching, the backend
reports a startup failure. If it attached but reported an error
before exiting, that error is surfaced through the error message
queue. Only after the worker has started does the backend wait on
the CV for decoding to be initialized, so a worker that never
starts no longer causes a hang.
A second problem can occur when the backend stops the worker. It
waits for the worker to exit before detaching from the error
message queue. If the worker is blocked writing into a full
queue, it waits for the backend to read from it, while the
backend waits for the worker to exit, so neither makes progress.
Fix this too by terminating the worker and detaching from the
error message queue before waiting for the worker to exit.
Detaching before the wait lets the blocked write fail so that the
worker can exit. On a failed worker registration the backend now
also forgets the error message queue, so teardown does not try to
terminate a worker that was never started.
Both fixes closely follow parallel.c, which coordinates its
workers through an error message queue in the same way.
These are less likely to hit in practice, but a user can trigger
these through SQL, so backpatch to 19, where REPACK CONCURRENTLY
was introduced.
While here, report the startup wait as BGWORKER_STARTUP instead
of REPACK_WORKER_EXPORT, which describes waiting for an output
file and is misleading for a worker that has not started up yet.
Reported-by: Nathan Bossart <[email protected]>
Reported-by: Bharath Rupireddy <[email protected]>
Author: Bharath Rupireddy <[email protected]>
Reviewed-by: Antonin Houska <[email protected]>
Reviewed-by: Masahiko Sawada <[email protected]>
Discussion: https://postgr.es/m/CALj2ACVAxA9HxvFe8HSspTJ-UO4Aoz%3DkuQdZBeLrod0gqUxH3g%40mail.gmail.com
Discussion: https://postgr.es/m/apBpOVZOyqrakEr_@nathan
Backpatch-through: 19
---
src/backend/commands/repack.c | 170 +++++++++++++++++++++++++++++-----
1 file changed, 149 insertions(+), 21 deletions(-)
diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c
index 477c86b2ba6..eb8780e1582 100644
--- a/src/backend/commands/repack.c
+++ b/src/backend/commands/repack.c
@@ -216,6 +216,7 @@ static Oid determine_clustered_index(Relation rel, bool usingindex,
const char *indexname);
static void start_repack_decoding_worker(Oid relid);
+static void wait_for_repack_worker_to_attach(DecodingWorker *worker);
static void stop_repack_decoding_worker(void);
static void stop_repack_decoding_worker_cb(int code, Datum arg);
static Snapshot get_initial_snapshot(DecodingWorker *worker);
@@ -3673,10 +3674,34 @@ start_repack_decoding_worker(Oid relid)
bgw.bgw_notify_pid = MyProcPid;
if (!RegisterDynamicBackgroundWorker(&bgw, &decoding_worker->handle))
+ {
+ /*
+ * We couldn't register the worker, so forget the error message queue
+ * we set up for it. Otherwise stopping the worker later would try to
+ * terminate a worker that was never registered.
+ */
+ decoding_worker->handle = NULL;
+ shm_mq_detach(decoding_worker->error_mqh);
+ decoding_worker->error_mqh = NULL;
+
ereport(ERROR,
errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
errmsg("out of background worker slots"),
errhint("You might need to increase \"%s\".", "max_worker_processes"));
+ }
+
+ /*
+ * Now that the worker is registered, connect the error message queue to
+ * it.
+ */
+ shm_mq_set_handle(decoding_worker->error_mqh, decoding_worker->handle);
+
+ /*
+ * Make sure the worker has started before we wait for it to initialize
+ * decoding below, so that the failure-to-start case does not hang
+ * forever.
+ */
+ wait_for_repack_worker_to_attach(decoding_worker);
/*
* The decoding setup must be done before the caller can have XID assigned
@@ -3701,6 +3726,89 @@ start_repack_decoding_worker(Oid relid)
ConditionVariableCancelSleep();
}
+/*
+ * Wait for the decoding worker to start up, and throw an error if it fails
+ * to do so.
+ *
+ * This is similar to WaitForParallelWorkersToAttach(). The only reliable way
+ * to tell a worker that failed to start (fork failure, or an exit before it
+ * attached) from one that is merely slow is to check whether it became the
+ * sender on the error message queue. If it stopped without attaching, nothing
+ * was queued and we report the generic failure ourselves. If it attached, any
+ * error it reported is in the queue and is thrown when we process pending
+ * messages, either here or later while we wait for it to initialize decoding.
+ */
+static void
+wait_for_repack_worker_to_attach(DecodingWorker *worker)
+{
+ bool worker_attached = false;
+
+ for (;;)
+ {
+ BgwHandleStatus status;
+ shm_mq *mq;
+ int rc;
+ pid_t pid;
+
+ /*
+ * This will process any repack messages that are pending and it may
+ * also throw an error propagated from a worker.
+ */
+ CHECK_FOR_INTERRUPTS();
+
+ /* If the worker is known to have attached, we're done. */
+ if (worker_attached)
+ break;
+
+ /* If error_mqh is NULL, the worker has already exited cleanly. */
+ if (worker->error_mqh == NULL)
+ {
+ worker_attached = true;
+ continue;
+ }
+
+ status = GetBackgroundWorkerPid(worker->handle, &pid);
+ if (status == BGWH_STARTED)
+ {
+ /* Has the worker attached to the error message queue? */
+ mq = shm_mq_get_queue(worker->error_mqh);
+ if (shm_mq_get_sender(mq) != NULL)
+ worker_attached = true;
+ }
+ else if (status == BGWH_STOPPED)
+ {
+ /*
+ * If the worker stopped without attaching to the error message
+ * queue, throw an error. Otherwise it attached and reported an
+ * error before exiting, so mark it attached and let the next
+ * attempt to process pending messages, here or later while the
+ * initial snapshot is set up, throw that error.
+ */
+ mq = shm_mq_get_queue(worker->error_mqh);
+ if (shm_mq_get_sender(mq) == NULL)
+ ereport(ERROR,
+ errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("REPACK decoding worker failed to start"),
+ errhint("More details may be available in the server log."));
+
+ worker_attached = true;
+ }
+ else
+ {
+ /*
+ * Worker not yet started, so we must wait. The postmaster will
+ * notify us via bgw_notify_pid if its state changes.
+ */
+ rc = WaitLatch(MyLatch,
+ WL_LATCH_SET | WL_EXIT_ON_PM_DEATH,
+ -1, WAIT_EVENT_BGWORKER_STARTUP);
+
+ if (rc & WL_LATCH_SET)
+ ResetLatch(MyLatch);
+ }
+ }
+}
+
/*
* Stop the decoding worker and cleanup the related resources.
*
@@ -3714,13 +3822,47 @@ stop_repack_decoding_worker(void)
if (decoding_worker == NULL)
return;
- /* Terminate the worker process, if one is running. */
+ /* Terminate the worker and forget its error message queue. */
+ if (decoding_worker->error_mqh != NULL)
+ {
+ /*
+ * The error message queue is attached before the worker is
+ * registered.
+ */
+ if (decoding_worker->handle != NULL)
+ TerminateBackgroundWorker(decoding_worker->handle);
+
+ shm_mq_detach(decoding_worker->error_mqh);
+ decoding_worker->error_mqh = NULL;
+ }
+
+ /*
+ * Cancel any sleep on the condition variable before detaching the shared
+ * memory segment, because the CV lives in that segment. Otherwise later
+ * cleanup would touch freed memory.
+ */
+ ConditionVariableCancelSleep();
+
+ /*
+ * If we have allocated a shared memory segment, detach it. This will
+ * implicitly detach the error message queue, and any other shared memory
+ * queues, stored there.
+ */
+ if (decoding_worker->seg != NULL)
+ {
+ dsm_detach(decoding_worker->seg);
+ decoding_worker->seg = NULL;
+ }
+
+ /*
+ * We can't finish the REPACK command until the worker has exited. This
+ * means, in particular, that we can't respond to interrupts at this
+ * stage.
+ */
if (decoding_worker->handle != NULL)
{
BgwHandleStatus status;
- TerminateBackgroundWorker(decoding_worker->handle);
- /* The worker should really exit before the REPACK command does. */
HOLD_INTERRUPTS();
status = WaitForBackgroundWorkerShutdown(decoding_worker->handle);
RESUME_INTERRUPTS();
@@ -3731,22 +3873,6 @@ stop_repack_decoding_worker(void)
errmsg("postmaster exited during REPACK command"));
}
- /*
- * Now detach from our shared memory segment. In error cases there might
- * still be messages from the worker in the queue, which ProcessInterrupts
- * would try to read; this is pointless (and causes an assertion failure),
- * so set the global pointer to NULL to have ProcessRepackMessages ignore
- * them.
- *
- * We must also cancel the current sleep, if one is still set up. This is
- * critical because the CV lives in the DSM that we're about to detach, so
- * if we omit it, later automatic cleanup tries to clear freed memory.
- */
- if (decoding_worker->error_mqh != NULL)
- shm_mq_detach(decoding_worker->error_mqh);
- ConditionVariableCancelSleep();
- if (decoding_worker->seg != NULL)
- dsm_detach(decoding_worker->seg);
pfree(decoding_worker);
decoding_worker = NULL;
}
@@ -3851,9 +3977,11 @@ ProcessRepackMessages(void)
/*
* Nothing to do if we haven't launched the worker yet or have already
- * terminated it.
+ * terminated it. Stopping the worker detaches the error message queue
+ * before clearing decoding_worker, so also bail out once error_mqh is
+ * gone.
*/
- if (decoding_worker == NULL)
+ if (decoding_worker == NULL || decoding_worker->error_mqh == NULL)
return;
/*
--
2.47.3
