On Mon, Sep 7, 2026 at 12:15 PM Bharath Rupireddy <[email protected]> wrote: > 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.
Álvaro pointed me here from my thread: https://postgr.es/m/cam527d9a7fbxjn7a3w2vapq0gojm7tnudfmh1uwjitgfi2k...@mail.gmail.com My AI harness tested v4. Silent worker exits still hang three post-attachment waits. Mihail raised snapshot/replay liveness in December 2025. Snapshot repro on a test build configured with --enable-injection-points, with v4 plus only the attached patch's test support: create extension injection_points; create table t (i int primary key); insert into t select generate_series(1, 100); set statement_timeout = '2s'; select injection_points_attach( 'repack-worker-before-snapshot-export', 'injection_points', 'injection_exit', null); repack (concurrently) t; This times out after worker exit. The attached patch applies atop v4. As far as I can see, the current uncommitted patches cover overall 5 cases: - startup before queue attach: Bharath v4 - full error queue teardown: Bharath v4 - silent exit after attach, before initialization: attached patch - silent exit before snapshot export: attached patch - silent exit before replay export: attached patch Nik On Sun, Sep 13, 2026 at 11:09 AM Bharath Rupireddy <[email protected]> wrote: > > Hi, > > On Tue, Sep 8, 2026 at 12:17 AM Antonin Houska <[email protected]> wrote: > > > > Bharath Rupireddy <[email protected]> wrote: > > > > > On Fri, Sep 4, 2026 at 5:11 PM Masahiko Sawada <[email protected]> > > > wrote: > > > > > > 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. > > > > Another reason for two separate waits is that two separate event types make > > sense: WAIT_EVENT_BGWORKER_STARTUP and WAIT_EVENT_REPACK_WORKER_EXPORT. > > That's correct. > > > > Please find the attached v3 patch. > > > > Just two comments: > > Thanks for reviewing it. > > > * wait_for_repack_worker_to_attach() - as the decoding_worker variable is > > declared static, this function does not necessarily need the argument. > > Removed the function parameter. > > > * Regarding comment: when the following is reached, the error message queue > > has already been detached, so no implicit detaching should happen. Also, > > there are no "other shared memory queues". > > > > + /* > > + * 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; > > + } > > I borrowed it from parallel.c. I agree it can be simplified, and I > have done that. > > Please find the attached v4 patch. > > -- > Bharath Rupireddy > Amazon Web Services: https://aws.amazon.com
From d9349e69acb0541c5838fd2f8786b40617c7682c Mon Sep 17 00:00:00 2001 From: Nik Samokhvalov <[email protected]> Date: Sun, 13 Sep 2026 08:32:29 -0700 Subject: [PATCH] Detect REPACK worker exits after error queue attachment Queue attachment does not protect the subsequent initialization, snapshot, and replay-file waits from a worker that exits without reporting an error. Wait on the leader's latch and check worker status in those predicate loops. Read status before the shared predicate so completed work wins over worker exit. Preserve queued errors before reporting a generic failure. Test silent exits in all three phases and ordinary errors in both wait loops, with same-backend retry and checks of data and filenodes. This is incremental to Bharath Rupireddy's v4 worker startup/teardown patch; its attachment wait and teardown changes are retained. Reported-by: Mihail Nikalayeu <[email protected]> Discussion: https://postgr.es/m/CADzfLwXp4c-MJx7yVDxAGNNxPbX4o9dqyivxavtHvmUsdXYqBQ@mail.gmail.com Discussion: https://postgr.es/m/CALj2ACVAxA9HxvFe8HSspTJ-UO4Aoz=kuQdZBeLrod0gqUxH3g@mail.gmail.com --- src/backend/commands/repack.c | 133 +++++++++++------- src/backend/commands/repack_worker.c | 4 + src/test/modules/injection_points/Makefile | 1 + .../expected/repack_worker_exit.out | 100 +++++++++++++ .../injection_points/injection_points.c | 10 ++ src/test/modules/injection_points/meson.build | 1 + .../specs/repack_worker_exit.spec | 66 +++++++++ 7 files changed, 267 insertions(+), 48 deletions(-) create mode 100644 src/test/modules/injection_points/expected/repack_worker_exit.out create mode 100644 src/test/modules/injection_points/specs/repack_worker_exit.spec diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c index 93fddd9..0f62f1e 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -220,6 +220,8 @@ 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 void wait_for_repack_worker_file(DecodingWorkerShared *shared, + int expected_file); static Snapshot get_initial_snapshot(DecodingWorker *worker); static void ProcessRepackMessage(StringInfo msg); @@ -3114,29 +3116,8 @@ process_concurrent_changes(XLogRecPtr end_of_wal, ChangeContext *chgcxt, bool do shared->done = done; SpinLockRelease(&shared->mutex); - /* - * The worker needs to finish processing of the current WAL record. Even - * if it's idle, it'll need to close the output file. Thus we're likely to - * wait, so prepare for sleep. - */ - ConditionVariablePrepareToSleep(&shared->cv); - for (;;) - { - int last_exported; - - SpinLockAcquire(&shared->mutex); - last_exported = shared->last_exported; - SpinLockRelease(&shared->mutex); - - /* - * Has the worker exported the file we are waiting for? - */ - if (last_exported == chgcxt->cc_file_seq) - break; - - ConditionVariableSleep(&shared->cv, WAIT_EVENT_REPACK_WORKER_EXPORT); - } - ConditionVariableCancelSleep(); + /* Wait for the worker to finish and close the requested file. */ + wait_for_repack_worker_file(shared, chgcxt->cc_file_seq); /* Open the file. */ DecodingWorkerFileName(fname, shared->relid, chgcxt->cc_file_seq); @@ -3826,10 +3807,23 @@ start_repack_decoding_worker(Oid relid) * waiting for the caller's transaction to end. Therefore wait here until * the worker indicates that it has the logical decoding initialized. */ - ConditionVariablePrepareToSleep(&shared->cv); for (;;) { + BgwHandleStatus status; bool initialized; + pid_t pid; + + ResetLatch(MyLatch); + CHECK_FOR_INTERRUPTS(); + + /* See wait_for_repack_worker_file() for the wait protocol. */ + ConditionVariablePrepareToSleep(&shared->cv); + + /* + * Read status first, so a worker that initializes and then exits + * still satisfies the predicate below. + */ + status = GetBackgroundWorkerPid(decoding_worker->handle, &pid); SpinLockAcquire(&shared->mutex); initialized = shared->initialized; @@ -3838,7 +3832,21 @@ start_repack_decoding_worker(Oid relid) if (initialized) break; - ConditionVariableSleep(&shared->cv, WAIT_EVENT_REPACK_WORKER_EXPORT); + if (status == BGWH_STOPPED) + { + /* Report a queued worker error, if one is available. */ + ConditionVariableCancelSleep(); + ProcessRepackMessages(); + + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("REPACK worker failed to initialize"), + errhint("More details may be available in the server log."))); + } + + (void) WaitLatch(MyLatch, + WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, + -1, WAIT_EVENT_REPACK_WORKER_EXPORT); } ConditionVariableCancelSleep(); } @@ -3998,43 +4006,72 @@ stop_repack_decoding_worker_cb(int code, Datum arg) } /* - * Get the initial snapshot from the decoding worker. + * Wait for the decoding worker to export a requested file. + * + * Worker exit sets our latch without signaling the CV, so wait on the latch + * directly. Re-register on the CV before each predicate check, since a CV + * signal removes us from its wait list and WaitLatch() does not re-add us. */ -static Snapshot -get_initial_snapshot(DecodingWorker *worker) +static void +wait_for_repack_worker_file(DecodingWorkerShared *shared, int expected_file) { - DecodingWorkerShared *shared; - char fname[MAXPGPATH]; - BufFile *file; - Size snap_size; - char *snap_space; - Snapshot snapshot; - - shared = (DecodingWorkerShared *) dsm_segment_address(worker->seg); - - /* - * The worker needs to initialize the logical decoding, which usually - * takes some time. Therefore it makes sense to prepare for the sleep - * first. - */ - ConditionVariablePrepareToSleep(&shared->cv); for (;;) { + BgwHandleStatus status; int last_exported; + pid_t pid; + ResetLatch(MyLatch); + CHECK_FOR_INTERRUPTS(); + ConditionVariablePrepareToSleep(&shared->cv); + + /* + * Read status before the counter, so a worker that publishes its + * final file and then exits still satisfies the predicate below. + */ + status = GetBackgroundWorkerPid(decoding_worker->handle, &pid); SpinLockAcquire(&shared->mutex); last_exported = shared->last_exported; SpinLockRelease(&shared->mutex); - /* - * Has the worker exported the file we are waiting for? - */ - if (last_exported == WORKER_FILE_SNAPSHOT) + if (last_exported == expected_file) break; - ConditionVariableSleep(&shared->cv, WAIT_EVENT_REPACK_WORKER_EXPORT); + if (status == BGWH_STOPPED) + { + ConditionVariableCancelSleep(); + /* Preserve any error the worker queued before exiting. */ + ProcessRepackMessages(); + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("REPACK worker exited before exporting requested data"), + errhint("More details may be available in the server log."))); + } + + (void) WaitLatch(MyLatch, + WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, + -1, WAIT_EVENT_REPACK_WORKER_EXPORT); } ConditionVariableCancelSleep(); +} + +/* + * Get the initial snapshot from the decoding worker. + */ +static Snapshot +get_initial_snapshot(DecodingWorker *worker) +{ + DecodingWorkerShared *shared; + char fname[MAXPGPATH]; + BufFile *file; + Size snap_size; + char *snap_space; + Snapshot snapshot; + + shared = (DecodingWorkerShared *) dsm_segment_address(worker->seg); + + /* Wait for the worker to export the initial snapshot. */ + wait_for_repack_worker_file(shared, WORKER_FILE_SNAPSHOT); /* Read the snapshot from a file. */ DecodingWorkerFileName(fname, shared->relid, WORKER_FILE_SNAPSHOT); diff --git a/src/backend/commands/repack_worker.c b/src/backend/commands/repack_worker.c index b4ba9cf..3e34088 100644 --- a/src/backend/commands/repack_worker.c +++ b/src/backend/commands/repack_worker.c @@ -26,6 +26,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" @@ -100,6 +101,7 @@ RepackWorkerMain(Datum main_arg) pq_redirect_to_shm_mq(seg, mqh); 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 @@ -147,6 +149,7 @@ 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); /* @@ -487,6 +490,7 @@ decode_concurrent_changes(LogicalDecodingContext *ctx, /* * Close the file so we can make it available to the backend. */ + INJECTION_POINT("repack-worker-before-replay-export", NULL); BufFileClose(dstate->file); dstate->file = NULL; SpinLockAcquire(&shared->mutex); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 408a35c..81512df 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -19,6 +19,7 @@ ISOLATION = basic \ reindex_concurrently_deferred \ repack \ repack_decode \ + repack_worker_exit \ repack_temporal \ repack_temporal_multirange \ repack_toast \ diff --git a/src/test/modules/injection_points/expected/repack_worker_exit.out b/src/test/modules/injection_points/expected/repack_worker_exit.out new file mode 100644 index 0000000..df39844 --- /dev/null +++ b/src/test/modules/injection_points/expected/repack_worker_exit.out @@ -0,0 +1,100 @@ +Parsed test spec with 1 sessions + +starting permutation: repack init_error repack detach_init repack snapshot_error repack detach_snapshot repack detach_replay check_failure check_data repack check_success check_data +step repack: REPACK (CONCURRENTLY) repack_worker_exit; +ERROR: REPACK worker failed to initialize +step init_error: + SELECT injection_points_detach('repack-worker-after-error-queue-attach'); + SELECT injection_points_attach('repack-worker-after-error-queue-attach', 'error'); + +injection_points_detach +----------------------- + +(1 row) + +injection_points_attach +----------------------- + +(1 row) + +step repack: REPACK (CONCURRENTLY) repack_worker_exit; +ERROR: error triggered for injection point repack-worker-after-error-queue-attach +step detach_init: + SELECT injection_points_detach('repack-worker-after-error-queue-attach'); + +injection_points_detach +----------------------- + +(1 row) + +step repack: REPACK (CONCURRENTLY) repack_worker_exit; +ERROR: REPACK worker exited before exporting requested data +step snapshot_error: + SELECT injection_points_detach('repack-worker-before-snapshot-export'); + SELECT injection_points_attach('repack-worker-before-snapshot-export', 'error'); + +injection_points_detach +----------------------- + +(1 row) + +injection_points_attach +----------------------- + +(1 row) + +step repack: REPACK (CONCURRENTLY) repack_worker_exit; +ERROR: error triggered for injection point repack-worker-before-snapshot-export +step detach_snapshot: + SELECT injection_points_detach('repack-worker-before-snapshot-export'); + +injection_points_detach +----------------------- + +(1 row) + +step repack: REPACK (CONCURRENTLY) repack_worker_exit; +ERROR: REPACK worker exited before exporting requested data +step detach_replay: + SELECT injection_points_detach('repack-worker-before-replay-export'); + +injection_points_detach +----------------------- + +(1 row) + +step check_failure: + SELECT pg_relation_filenode('repack_worker_exit') = node AS failure_unchanged + FROM original_node; + +failure_unchanged +----------------- +t +(1 row) + +step check_data: SELECT * FROM repack_worker_exit ORDER BY i; +i|j +-+----- +1|one +2|two +3|three +(3 rows) + +step repack: REPACK (CONCURRENTLY) repack_worker_exit; +step check_success: + SELECT pg_relation_filenode('repack_worker_exit') <> node AS success_rewritten + FROM original_node; + +success_rewritten +----------------- +t +(1 row) + +step check_data: SELECT * FROM repack_worker_exit ORDER BY i; +i|j +-+----- +1|one +2|two +3|three +(3 rows) + diff --git a/src/test/modules/injection_points/injection_points.c b/src/test/modules/injection_points/injection_points.c index 1683498..d3a4647 100644 --- a/src/test/modules/injection_points/injection_points.c +++ b/src/test/modules/injection_points/injection_points.c @@ -80,6 +80,9 @@ 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); /* track if injection points attached in this process are linked to it */ static bool injection_point_local = false; @@ -222,6 +225,13 @@ injection_notice(const char *name, const void *private_data, void *arg) elog(NOTICE, "notice triggered for injection point %s", name); } +/* Exit without reporting an error. */ +void +injection_exit(const char *name, const void *private_data, void *arg) +{ + proc_exit(1); +} + /* * Error cleanup callback for injection point waits. */ diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index a7b40e0..1484630 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -48,6 +48,7 @@ tests += { 'reindex_concurrently_deferred', 'repack', 'repack_decode', + 'repack_worker_exit', 'repack_temporal', 'repack_temporal_multirange', 'repack_toast', diff --git a/src/test/modules/injection_points/specs/repack_worker_exit.spec b/src/test/modules/injection_points/specs/repack_worker_exit.spec new file mode 100644 index 0000000..ca44614 --- /dev/null +++ b/src/test/modules/injection_points/specs/repack_worker_exit.spec @@ -0,0 +1,66 @@ +# Exercise each post-attachment wait, then retry in the same backend. +setup +{ + CREATE EXTENSION injection_points; + SELECT injection_points_attach('repack-worker-after-error-queue-attach', + 'injection_points', 'injection_exit', NULL); + SELECT injection_points_attach('repack-worker-before-snapshot-export', + 'injection_points', 'injection_exit', NULL); + SELECT injection_points_attach('repack-worker-before-replay-export', + 'injection_points', 'injection_exit', NULL); + CREATE TABLE repack_worker_exit(i int PRIMARY KEY, j text); + INSERT INTO repack_worker_exit VALUES (1, 'one'), (2, 'two'), (3, 'three'); + CREATE TABLE original_node AS + SELECT pg_relation_filenode('repack_worker_exit') AS node; +} + +teardown +{ + DROP TABLE repack_worker_exit, original_node; + DROP EXTENSION injection_points; +} + +session s1 +step repack { REPACK (CONCURRENTLY) repack_worker_exit; } +step init_error +{ + SELECT injection_points_detach('repack-worker-after-error-queue-attach'); + SELECT injection_points_attach('repack-worker-after-error-queue-attach', 'error'); +} +step detach_init +{ + SELECT injection_points_detach('repack-worker-after-error-queue-attach'); +} +step snapshot_error +{ + SELECT injection_points_detach('repack-worker-before-snapshot-export'); + SELECT injection_points_attach('repack-worker-before-snapshot-export', 'error'); +} +step detach_snapshot +{ + SELECT injection_points_detach('repack-worker-before-snapshot-export'); +} +step detach_replay +{ + SELECT injection_points_detach('repack-worker-before-replay-export'); +} +step check_failure +{ + SELECT pg_relation_filenode('repack_worker_exit') = node AS failure_unchanged + FROM original_node; +} +step check_success +{ + SELECT pg_relation_filenode('repack_worker_exit') <> node AS success_rewritten + FROM original_node; +} +step check_data { SELECT * FROM repack_worker_exit ORDER BY i; } + +# A queued error must retain priority over the generic stopped-worker error. +permutation + repack init_error + repack detach_init + repack snapshot_error + repack detach_snapshot + repack detach_replay check_failure check_data + repack check_success check_data -- 2.50.1 (Apple Git-155)
