Hi Bharath, I reviewed v5 patch by patch. v6 is attached
0001 It works. I made fork() really fail with a low RLIMIT_NPROC. On master, REPACK (CONCURRENTLY) waits until statement_timeout. With 0001 it fails at once with "REPACK decoding worker failed to start". One problem. stop_repack_decoding_worker() now calls dsm_detach() before it waits for the worker to exit. If the worker has mapped the segment but not yet called SharedFileSetAttach(), the detach destroys the file set, and the worker fails with ERROR: could not attach to a SharedFileSet that is already destroyed I hit this 12 times in 1200 cancelled REPACKs. Only shm_mq_detach() has to come before the wait to fix the deadlock, so v6-0001 moves dsm_detach() back after the wait, as on master. That gives 0 in 1200. 0002 Looks good. It follows parallel.c, and detaching before signalling is needed, since with the old order the backend can read the queue too early and never look again. With 0002 in, polling the worker status in the wait loops, as Nikolay's patch does, is not needed. No false alarms in 100 REPACK (CONCURRENTLY) runs with concurrent updates and the owner's client_min_messages at debug1 and debug5, so the worker often blocks on a full queue. pg_terminate_backend() on the worker gives the backend the worker's own error at once. 0003 Please keep it. I checked that it catches each fix. Without the attach wait, without the lost connection error, or with the queue detached after the wait again, the test no longer passes. One problem. The REPACK calls have no timeout, so a regression hangs the test instead of failing it. Under meson it prints nothing until the test is killed, and under make check nothing stops it. v6-0003 passes timeout_default to those calls, so it now fails with "psql timed out" at the right line. The rest of the test already waits with timeout_default in poll_query_until(), pump_until() and background_psql(), so this adds no new risk on slow machines. 0004 Looks fine. It matches what parallel.c does. Thanks, Shihao
From fe951ed8e53892a9ae545e1d65bb04477e57af70 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Tue, 8 Sep 2026 16:43:36 +0000 Subject: [PATCH v6 1/4] Fix hang and deadlock in concurrent REPACK worker handling. REPACK CONCURRENTLY starts a background worker to decode the changes made to the table while it is being repacked, and the backend running the command coordinates with that worker through shared memory and an error message queue. Previously, the backend went straight to waiting on a condition variable for the worker to set up decoding. Nothing in that wait notices a worker that never arrived. The postmaster does signal the backend when the worker fails to start, for example when fork() fails, but the signal only wakes the sleep, which resumes without raising anything. The backend therefore waits forever 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 worker's status tells a worker that is gone from one that is merely slow, and whether it ever became the sender on the queue tells whether it left an error behind. A worker that stopped without attaching gets the generic startup failure, and one that attached and reported an error before exiting has that error come through the queue. The backend can also deadlock with the worker while stopping it, because it waits for the worker to exit before detaching from the error message queue. A worker blocked writing into a full queue normally escapes once its send loop acts on the termination, but not if it holds interrupts where it blocks, and the backend holds interrupts of its own across the exit wait. Neither side can be canceled. Fix this by detaching from the error message queue before waiting for the worker to exit, so that the blocked send fails with SHM_MQ_DETACHED and the worker leaves without having to act on an interrupt at all. Teardown only terminates a worker it has a handle for, so a worker that was never registered needs nothing undone. Only the queue is detached early. The shared memory segment is still detached after the worker has exited, because the worker attaches to the shared file set in that segment only after it has connected to the database. Detaching the segment before that would destroy the file set, and a worker that is still starting up would then fail with "could not attach to a SharedFileSet that is already destroyed". Backpatch to 19, where REPACK CONCURRENTLY was introduced. 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]> Reviewed-by: Shihao Zhong <[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 | 152 ++++++++++++++++++++++++++++++---- 1 file changed, 135 insertions(+), 17 deletions(-) diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c index 3e972f2cc14..4d19dc06b66 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -217,6 +217,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(void); 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); @@ -3796,6 +3797,19 @@ start_repack_decoding_worker(Oid relid) 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(); + /* * The decoding setup must be done before the caller can have XID assigned * for any reason, otherwise the worker might end up in a deadlock, @@ -3819,6 +3833,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(void) +{ + 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 (decoding_worker->error_mqh == NULL) + { + worker_attached = true; + continue; + } + + status = GetBackgroundWorkerPid(decoding_worker->handle, &pid); + if (status == BGWH_STARTED) + { + /* Has the worker attached to the error message queue? */ + mq = shm_mq_get_queue(decoding_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(decoding_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. * @@ -3832,13 +3929,36 @@ 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(); + + /* + * 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(); @@ -3850,21 +3970,17 @@ stop_repack_decoding_worker(void) } /* - * 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. + * Detach from the shared memory segment only now that the worker is gone. + * The worker attaches to the shared file set well after it maps the + * segment, so detaching any earlier can destroy the file set under a + * worker that is still starting up. */ - if (decoding_worker->error_mqh != NULL) - shm_mq_detach(decoding_worker->error_mqh); - ConditionVariableCancelSleep(); if (decoding_worker->seg != NULL) + { dsm_detach(decoding_worker->seg); + decoding_worker->seg = NULL; + } + pfree(decoding_worker); decoding_worker = NULL; } @@ -3969,9 +4085,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.37.1 (Apple Git-137.1)
From 5a7491c52e72afa0a001d340b159797a36538a86 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Tue, 15 Sep 2026 06:25:58 +0000 Subject: [PATCH v6 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 <[email protected]> Author: Bharath Rupireddy <[email protected]> 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 4d19dc06b66..489c74b5d64 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -4119,10 +4119,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; @@ -4144,12 +4146,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.")); } } @@ -4201,6 +4213,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.37.1 (Apple Git-137.1)
From 346ddaadcfce9462253502273596c0a8d120496d Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Tue, 15 Sep 2026 06:25:59 +0000 Subject: [PATCH v6 3/4] Add tests for a REPACK decoding worker that goes away. Commits 1bddaf2de14 and e067c4ea39f taught the backend running REPACK CONCURRENTLY to notice a decoding worker that went away with the work unfinished, instead of waiting for it forever or taking that for the normal end of the worker's work, but nothing in the tree exercises that. The worker exits on its own as soon as it is done, so the only way to reach those cases is to make it leave while the backend is still waiting for it. This adds injection points to the worker: one before it attaches to the error message queue, one at each of the three places where it has something to tell the backend, one right after the initial snapshot is exported, and one in its exit callback, before it detaches from the queue. It also adds two callbacks to the injection_points module: injection_exit(), which calls proc_exit() so that the worker leaves without reporting anything, and injection_notice_oversized(), which emits a notice larger than the error message queue so that the worker blocks writing into it. The test makes the worker exit before it attaches to the queue, where nothing can tell the backend about it, and checks that the backend reports the failure to start. It then makes the worker exit silently at each of the three points and checks that the backend reports the lost connection, makes it report an error at each of those points and checks that the error reaches the backend through the queue, and kills it while it blocks writing into a full queue that nobody reads. The deadlock that commit 1bddaf2de14 fixes needs a worker that holds interrupts where it blocks writing into the queue, which no SQL level failure arranges on its own. The test gets there through the worker's exit path: proc_exit() holds interrupts and clears the pending die and cancel flags, so anything the worker sends from its exit callback goes out uninterruptibly. Cancelling the backend while the worker waits at an injection point sends the backend into teardown with the worker still alive, and the oversized notice the worker emits on its way out then blocks in the queue that only the backend could have drained. Finally, the test checks that a failed REPACK leaves the table alone, and that an undisturbed one rewrites it and keeps its data. --- src/backend/commands/repack_worker.c | 12 + .../injection_points/injection_points.c | 40 +++ src/test/modules/test_misc/meson.build | 1 + .../modules/test_misc/t/100_repack_worker.pl | 250 ++++++++++++++++++ 4 files changed, 303 insertions(+) create mode 100644 src/test/modules/test_misc/t/100_repack_worker.pl diff --git a/src/backend/commands/repack_worker.c b/src/backend/commands/repack_worker.c index bc30346861e..a03c0f326d1 100644 --- a/src/backend/commands/repack_worker.c +++ b/src/backend/commands/repack_worker.c @@ -27,6 +27,7 @@ #include "storage/ipc.h" #include "storage/proc.h" #include "tcop/tcopprot.h" +#include "utils/injection_point.h" #include "utils/memutils.h" #define PGREPACK_PLUGIN "pgrepack" @@ -81,6 +82,9 @@ RepackWorkerMain(Datum main_arg) shared = (DecodingWorkerShared *) dsm_segment_address(seg); + /* Leaving here means leaving silently: no error queue, no signal yet. */ + INJECTION_POINT("repack-worker-before-error-queue-attach", NULL); + /* Arrange to signal the leader if we exit. */ repack_backend_pid = shared->backend_pid; repack_backend_proc_number = shared->backend_proc_number; @@ -104,6 +108,8 @@ RepackWorkerMain(Datum main_arg) pq_set_parallel_leader(shared->backend_pid, shared->backend_proc_number); + INJECTION_POINT("repack-worker-after-error-queue-attach", NULL); + /* * Connect to the database, skipping the connection authorization checks * as parallel workers do. Note that we run as the owner of the table @@ -150,7 +156,9 @@ RepackWorkerMain(Datum main_arg) /* Build the initial snapshot and export it. */ snapshot = SnapBuildInitialSnapshot(decoding_ctx->snapshot_builder); + INJECTION_POINT("repack-worker-before-snapshot-export", NULL); export_initial_snapshot(snapshot, shared); + INJECTION_POINT("repack-worker-after-snapshot-export", NULL); /* * Only historic snapshots should be used now. Do not let us restrict the @@ -181,6 +189,9 @@ RepackWorkerMain(Datum main_arg) static void RepackWorkerShutdown(int code, Datum arg) { + /* Anything we send from here on goes out with interrupts held. */ + INJECTION_POINT("repack-worker-before-exit", NULL); + /* * Detach from the shared memory segment before we signal the backend. * Detaching also detaches the error message queue, and the backend learns @@ -498,6 +509,7 @@ decode_concurrent_changes(LogicalDecodingContext *ctx, /* * Close the file so we can make it available to the backend. */ + INJECTION_POINT("repack-worker-before-changes-export", NULL); BufFileClose(dstate->file); dstate->file = NULL; SpinLockAcquire(&shared->mutex); diff --git a/src/test/modules/injection_points/injection_points.c b/src/test/modules/injection_points/injection_points.c index 66d8158d0c2..52afab59b51 100644 --- a/src/test/modules/injection_points/injection_points.c +++ b/src/test/modules/injection_points/injection_points.c @@ -41,6 +41,13 @@ PG_MODULE_MAGIC; #define INJ_MAX_WAIT 8 #define INJ_NAME_MAXLEN 64 +/* + * Length of the notice emitted by injection_notice_oversized(), chosen to + * exceed the size of any shared message queue that a background worker uses + * to talk to the process that launched it. + */ +#define INJ_OVERSIZED_NOTICE_LEN (256 * 1024) + /* Thresholds of waits */ #define INJ_WAIT_INITIAL_US 10 /* 10us */ #define INJ_WAIT_MAX_US 100000 /* 100ms */ @@ -81,6 +88,12 @@ extern PGDLLEXPORT void injection_notice(const char *name, extern PGDLLEXPORT void injection_wait(const char *name, const void *private_data, void *arg); +extern PGDLLEXPORT void injection_exit(const char *name, + const void *private_data, + void *arg); +extern PGDLLEXPORT void injection_notice_oversized(const char *name, + const void *private_data, + void *arg); /* track if injection points attached in this process are linked to it */ static bool injection_point_local = false; @@ -222,6 +235,33 @@ injection_notice(const char *name, const void *private_data, void *arg) elog(NOTICE, "notice triggered for injection point %s", name); } +/* + * Exit the process without reporting anything, the way a process that calls + * proc_exit() directly does. + */ +void +injection_exit(const char *name, const void *private_data, void *arg) +{ + proc_exit(1); +} + +/* + * Emit a notice too large to fit into the queue a background worker sends its + * messages through, so that the worker blocks until the process that launched + * it reads from that queue. + */ +void +injection_notice_oversized(const char *name, const void *private_data, + void *arg) +{ + char *message = palloc(INJ_OVERSIZED_NOTICE_LEN + 1); + + memset(message, 'x', INJ_OVERSIZED_NOTICE_LEN); + message[INJ_OVERSIZED_NOTICE_LEN] = '\0'; + + elog(NOTICE, "%s", message); +} + /* * Error cleanup callback for injection point waits. */ diff --git a/src/test/modules/test_misc/meson.build b/src/test/modules/test_misc/meson.build index 5d81f5b13be..1b3f248737d 100644 --- a/src/test/modules/test_misc/meson.build +++ b/src/test/modules/test_misc/meson.build @@ -24,6 +24,7 @@ tests += { 't/013_temp_obj_multisession.pl', 't/014_log_statement_max_length.pl', 't/015_temp_schema_exit_deferrable.pl', + 't/100_repack_worker.pl', ], # The injection points are cluster-wide, so disable installcheck 'runningcheck': false, diff --git a/src/test/modules/test_misc/t/100_repack_worker.pl b/src/test/modules/test_misc/t/100_repack_worker.pl new file mode 100644 index 00000000000..29c0b1cf3f0 --- /dev/null +++ b/src/test/modules/test_misc/t/100_repack_worker.pl @@ -0,0 +1,250 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Test that REPACK (CONCURRENTLY) does not wait forever for a decoding worker +# that goes away. The worker can leave before it can report anything at all, +# leave after reporting an error, leave without a word, or be killed while it +# blocks writing into a full error message queue, either while decoding or on +# its way out. + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +plan skip_all => 'Injection points not supported by this build' + unless $ENV{enable_injection_points} eq 'yes'; + +my $node = PostgreSQL::Test::Cluster->new('node'); +# REPACK (CONCURRENTLY) decodes WAL, so it needs more than wal_level=minimal. +$node->init(allows_streaming => 1); +$node->start; + +plan skip_all => 'Extension injection_points not installed' + unless $node->check_extension('injection_points'); + +$node->safe_psql( + 'postgres', qq[ + CREATE EXTENSION injection_points; + CREATE TABLE tbl (i int PRIMARY KEY, j text); + INSERT INTO tbl SELECT g, 'row ' || g FROM generate_series(1, 100) g; +]); + +my $filenode = + $node->safe_psql('postgres', "SELECT pg_relation_filenode('tbl')"); + +# Note that none of the injection points below is local to the session that +# attaches it, because they all have to fire in the decoding worker. +# +# Each REPACK below is given a timeout, so that a backend that waits forever +# for the worker, which is the bug being tested, fails the test instead of +# hanging it. + +# A worker that leaves before it attaches to the error message queue leaves +# without signalling the backend, so the queue cannot report it. The backend +# has to notice it while waiting for it to start. +{ + my $point = 'repack-worker-before-error-queue-attach'; + + $node->safe_psql( + 'postgres', qq[ + SELECT injection_points_attach('$point', 'injection_points', + 'injection_exit', NULL); + ]); + + my ($ret, $stdout, $stderr) = $node->psql( + 'postgres', + 'REPACK (CONCURRENTLY) tbl', + timeout => $PostgreSQL::Test::Utils::timeout_default); + isnt($ret, 0, "REPACK fails when the worker exits at $point"); + like( + $stderr, + qr/REPACK decoding worker failed to start/, + "REPACK reports the worker that never attached to the queue"); + + $node->safe_psql('postgres', "SELECT injection_points_detach('$point')"); +} + +# These are the three places where the worker has something to tell the +# backend, and thus the three places where the backend has something to wait +# for: the decoding setup, the initial snapshot, and the file of concurrent +# changes. +my @points = ( + 'repack-worker-after-error-queue-attach', + 'repack-worker-before-snapshot-export', + 'repack-worker-before-changes-export'); + +# A worker that leaves without a word. The backend only learns about it from +# the error message queue going away. +foreach my $point (@points) +{ + $node->safe_psql( + 'postgres', qq[ + SELECT injection_points_attach('$point', 'injection_points', + 'injection_exit', NULL); + ]); + + my ($ret, $stdout, $stderr) = $node->psql( + 'postgres', + 'REPACK (CONCURRENTLY) tbl', + timeout => $PostgreSQL::Test::Utils::timeout_default); + isnt($ret, 0, "REPACK fails when the worker exits at $point"); + like( + $stderr, + qr/lost connection to REPACK decoding worker/, + "REPACK reports the worker that exited at $point"); + + $node->safe_psql('postgres', "SELECT injection_points_detach('$point')"); +} + +# A worker that reports an error before it leaves. The error reaches the +# backend through the queue, so the backend reports that one instead. +foreach my $point (@points) +{ + $node->safe_psql('postgres', + "SELECT injection_points_attach('$point', 'error')"); + + my ($ret, $stdout, $stderr) = $node->psql( + 'postgres', + 'REPACK (CONCURRENTLY) tbl', + timeout => $PostgreSQL::Test::Utils::timeout_default); + isnt($ret, 0, "REPACK fails when the worker errors out at $point"); + like( + $stderr, + qr/error triggered for injection point $point/, + "REPACK reports the error of the worker at $point"); + + $node->safe_psql('postgres', "SELECT injection_points_detach('$point')"); +} + +is($node->safe_psql('postgres', "SELECT pg_relation_filenode('tbl')"), + $filenode, 'a failed REPACK leaves the table alone'); + +# A worker killed while it waits for the backend to read from a full error +# message queue. The error it reports on the way out cannot reach the backend, +# because the queue it would go through is the one that is already full. +SKIP: +{ + skip 'this test requires SIGSTOP', 1 if $windows_os; + + # Hold the worker until we have stopped the backend, and then make it send + # a message far larger than the error message queue. + $node->safe_psql( + 'postgres', qq[ + SELECT injection_points_attach('repack-worker-before-snapshot-export', + 'wait'); + SELECT injection_points_attach('repack-worker-after-snapshot-export', + 'injection_points', + 'injection_notice_oversized', NULL); + ]); + + my $psql = $node->background_psql('postgres', on_error_stop => 0); + my $backend_pid = $psql->query('SELECT pg_backend_pid()'); + + $psql->{stdin} .= "REPACK (CONCURRENTLY) tbl;\n"; + $psql->{run}->pump_nb(); + + # Once the worker is held, the backend is waiting for it and has not yet + # read anything from the queue. + $node->poll_query_until( + 'postgres', qq[ + SELECT count(*) = 1 FROM pg_stat_activity + WHERE backend_type = 'REPACK decoding worker' + AND wait_event = 'repack-worker-before-snapshot-export' + ]) or die "timed out while waiting for the decoding worker to start"; + + my $worker_pid = $node->safe_psql( + 'postgres', qq[ + SELECT pid FROM pg_stat_activity + WHERE backend_type = 'REPACK decoding worker' + ]); + + # Stop the backend, so that nothing reads from the queue any more, and let + # the worker fill it. + kill 'STOP', $backend_pid; + $node->safe_psql( + 'postgres', qq[ + SELECT injection_points_wakeup('repack-worker-before-snapshot-export'); + ]); + $node->poll_query_until( + 'postgres', qq[ + SELECT count(*) = 1 FROM pg_stat_activity + WHERE pid = $worker_pid AND wait_event = 'MessageQueuePutMessage' + ]) or die "timed out while waiting for the error message queue to fill up"; + + # Kill the worker while it waits, then let the backend run again. + $node->safe_psql('postgres', "SELECT pg_terminate_backend($worker_pid)"); + kill 'CONT', $backend_pid; + + ok( pump_until( + $psql->{run}, $psql->{timeout}, + \$psql->{stderr}, + qr/lost connection to REPACK decoding worker/), + 'REPACK reports the worker killed while the queue was full'); + + $psql->quit; + + $node->safe_psql( + 'postgres', qq[ + SELECT injection_points_detach('repack-worker-before-snapshot-export'); + SELECT injection_points_detach('repack-worker-after-snapshot-export'); + ]); +} + +# A worker blocked writing into a full error message queue on its way out, where +# proc_exit() holds interrupts and nothing can make it give up. The backend has +# to stop watching the queue before it waits for the worker, or the two deadlock. +{ + # Hold the worker where the backend is waiting for it, and make it emit a + # message far larger than the error message queue as it exits. + $node->safe_psql( + 'postgres', qq[ + SELECT injection_points_attach('repack-worker-before-snapshot-export', + 'wait'); + SELECT injection_points_attach('repack-worker-before-exit', + 'injection_points', + 'injection_notice_oversized', NULL); + ]); + + my $psql = $node->background_psql('postgres', on_error_stop => 0); + my $backend_pid = $psql->query('SELECT pg_backend_pid()'); + + $psql->{stdin} .= "REPACK (CONCURRENTLY) tbl;\n"; + $psql->{run}->pump_nb(); + + # While the worker is held, only it can export the snapshot the backend + # waits for, so the backend is asleep and reads nothing from the queue. + $node->poll_query_until( + 'postgres', qq[ + SELECT count(*) = 1 FROM pg_stat_activity + WHERE backend_type = 'REPACK decoding worker' + AND wait_event = 'repack-worker-before-snapshot-export' + ]) or die "timed out while waiting for the decoding worker to start"; + + # Cancel the backend, which sends it into teardown with the worker alive. + $node->safe_psql('postgres', "SELECT pg_cancel_backend($backend_pid)"); + + ok( pump_until( + $psql->{run}, $psql->{timeout}, + \$psql->{stderr}, + qr/canceling statement due to user request/), + 'REPACK stops the worker that blocks on the queue while exiting'); + + $psql->quit; + + $node->safe_psql( + 'postgres', qq[ + SELECT injection_points_detach('repack-worker-before-snapshot-export'); + SELECT injection_points_detach('repack-worker-before-exit'); + ]); +} + +# Nothing above left the table or the session in a state that keeps REPACK from +# working. +$node->safe_psql('postgres', 'REPACK (CONCURRENTLY) tbl'); +isnt($node->safe_psql('postgres', "SELECT pg_relation_filenode('tbl')"), + $filenode, 'REPACK rewrites the table once the worker is left alone'); +is($node->safe_psql('postgres', 'SELECT count(*), sum(i) FROM tbl'), + '100|5050', 'REPACK keeps the table data'); + +done_testing(); -- 2.37.1 (Apple Git-137.1)
v6-0004-Clear-the-REPACK-message-flag-when-there-is-nothi.patch
Description: Binary data
