From b6cfb97d3d473c2b43522a8f4ca98dc83cb33107 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Tue, 15 Sep 2026 06:25:58 +0000 Subject: [PATCH v5 2/4] Detect premature exit of the REPACK decoding worker. Previously, the backend running REPACK CONCURRENTLY took the decoding worker's error message queue going away as the normal end of the worker's work, because the worker exits as soon as it is done. It therefore could not tell that end from a worker that went away early, and it has no other way of noticing one: it waits for the worker in condition variable sleeps that wake only on the worker's own signal. As a result, a worker that exits before it initializes decoding, before it exports the initial snapshot, or before it exports the concurrent changes leaves the backend waiting forever for shared state that nobody will set, while REPACK holds its lock on the table. A worker that fails while the error message queue is full ends up the same way, because its error report blocks in the queue and is lost, so the failure only reaches the server log. Fix this by having the worker say when it is done. The worker now sends a Terminate message as its last act, and the backend forgets the error message queue when it receives that message. The queue going away without that message then means the worker is gone with the work unfinished, and the backend reports it instead of taking it for the normal end. The condition variable loops need no change, because the next interrupt check after the worker's signal throws the error. For this to be reliable the worker has to detach from the shared memory segment before it signals the backend. The other way round, the backend can read the queue while the worker still looks attached, and then nothing makes it read again. Backpatch to 19, where REPACK CONCURRENTLY was introduced. Reported-by: Nikolay Samokhvalov Author: Bharath Rupireddy Discussion: https://postgr.es/m/CALj2ACVAxA9HxvFe8HSspTJ-UO4Aoz%3DkuQdZBeLrod0gqUxH3g%40mail.gmail.com Discussion: https://postgr.es/m/CAM527d-bUOdoZezwXuhpjjwm-cB6q_m_YmJyVusgTPKSkohGJA%40mail.gmail.com Backpatch-through: 19 --- src/backend/commands/repack.c | 37 +++++++++++++++++++++++----- src/backend/commands/repack_worker.c | 29 +++++++++++++++------- 2 files changed, 51 insertions(+), 15 deletions(-) diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c index 068c6222e3d..9c51b945a8a 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -4113,10 +4113,12 @@ ProcessRepackMessages(void) RepackMessagePending = false; /* - * Read as many messages as we can from the worker, but stop when no more - * messages can be read from the worker without blocking. + * Read as many messages as we can from the worker, but stop when either + * (1) the worker's error message queue goes away, which can happen if we + * receive a Terminate message from the worker; or (2) no more messages + * can be read from the worker without blocking. */ - while (true) + while (decoding_worker->error_mqh != NULL) { shm_mq_result res; Size nbytes; @@ -4138,12 +4140,22 @@ ProcessRepackMessages(void) else { /* - * The decoding worker is special in that it exits as soon as it - * has its work done. Thus the DETACHED result code is fine. + * The worker detaches the error message queue when it exits, and + * a worker that finished its work told us so with a Terminate + * message. So the queue going away without that message means the + * worker is gone with the work unfinished, and we must report it + * here. Otherwise the REPACK command would wait forever for a + * worker that will never answer. + * + * The worker may well have failed with an error of its own that + * never reached us, so point to the server log for details. */ Assert(res == SHM_MQ_DETACHED); - break; + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("lost connection to REPACK decoding worker"), + errhint("More details may be available in the server log.")); } } @@ -4195,6 +4207,19 @@ ProcessRepackMessage(StringInfo msg) break; } + case PqMsg_Terminate: + { + /* + * The worker sends this once it has finished its work and is + * about to exit, so stop watching its error message queue. + * The queue going away then no longer means that the worker + * is gone with the work unfinished. + */ + shm_mq_detach(decoding_worker->error_mqh); + decoding_worker->error_mqh = NULL; + break; + } + default: { elog(ERROR, "unrecognized message type received from decoding worker: %c (message length %d bytes)", diff --git a/src/backend/commands/repack_worker.c b/src/backend/commands/repack_worker.c index bf2bc2dca13..bc30346861e 100644 --- a/src/backend/commands/repack_worker.c +++ b/src/backend/commands/repack_worker.c @@ -21,6 +21,7 @@ #include "access/xlogwait.h" #include "commands/repack.h" #include "commands/repack_internal.h" +#include "libpq/libpq.h" #include "libpq/pqmq.h" #include "replication/snapbuild.h" #include "storage/ipc.h" @@ -44,8 +45,9 @@ static bool am_repack_worker = false; /* The WAL segment being decoded. */ static XLogSegNo repack_current_segment = 0; -/* Our DSM segment, for shutting down */ -static dsm_segment *worker_dsm_segment = NULL; +/* Backend that launched us, for shutting down */ +static pid_t repack_backend_pid; +static ProcNumber repack_backend_proc_number; /* * Keep track of the table we're processing, to skip logical decoding of data @@ -76,12 +78,13 @@ RepackWorkerMain(Datum main_arg) ereport(ERROR, errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("could not map dynamic shared memory segment")); - worker_dsm_segment = seg; shared = (DecodingWorkerShared *) dsm_segment_address(seg); /* Arrange to signal the leader if we exit. */ - before_shmem_exit(RepackWorkerShutdown, PointerGetDatum(shared)); + repack_backend_pid = shared->backend_pid; + repack_backend_proc_number = shared->backend_proc_number; + before_shmem_exit(RepackWorkerShutdown, PointerGetDatum(seg)); /* * Join locking group - see the comments around the call of @@ -167,6 +170,9 @@ RepackWorkerMain(Datum main_arg) /* Cleanup. */ repack_cleanup_logical_decoding(decoding_ctx); CommitTransactionCommand(); + + /* Report success, so that our exit does not look like a failure. */ + pq_putmessage(PqMsg_Terminate, NULL, 0); } /* @@ -175,13 +181,18 @@ RepackWorkerMain(Datum main_arg) static void RepackWorkerShutdown(int code, Datum arg) { - DecodingWorkerShared *shared = (DecodingWorkerShared *) DatumGetPointer(arg); + /* + * Detach from the shared memory segment before we signal the backend. + * Detaching also detaches the error message queue, and the backend learns + * that we are gone by reading that queue when it handles our signal. If + * we signaled first, the backend could read the queue while it still + * looks attached, and nothing would make it read again. + */ + dsm_detach((dsm_segment *) DatumGetPointer(arg)); - SendProcSignal(shared->backend_pid, + SendProcSignal(repack_backend_pid, PROCSIG_REPACK_MESSAGE, - shared->backend_proc_number); - - dsm_detach(worker_dsm_segment); + repack_backend_proc_number); } bool -- 2.47.3