From 9300fff8f9b5e49536812802815c22e581b7b203 Mon Sep 17 00:00:00 2001 From: Nik Samokhvalov Date: Sun, 13 Sep 2026 01:39:53 -0700 Subject: [PATCH] Avoid hanging if REPACK worker fails to initialize A registered dynamic background worker might fail to start or exit before initializing its shared state. The leader currently waits only on a condition variable, which does not return for plain latch notifications from the postmaster. Wait directly on the latch and inspect the background worker handle on every iteration. Preserve queued worker errors where available, and otherwise report a generic initialization failure. Add injection point coverage for failures before DSM attachment and after error queue attachment. --- src/backend/commands/repack.c | 41 +++++- src/backend/commands/repack_worker.c | 3 + src/test/modules/injection_points/Makefile | 1 + .../expected/repack_worker.out | 135 ++++++++++++++++++ src/test/modules/injection_points/meson.build | 1 + .../injection_points/specs/repack_worker.spec | 104 ++++++++++++++ 6 files changed, 283 insertions(+), 2 deletions(-) create mode 100644 src/test/modules/injection_points/expected/repack_worker.out create mode 100644 src/test/modules/injection_points/specs/repack_worker.spec diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c index 83168a4e6f3..7557228fec2 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -3730,10 +3730,33 @@ 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; + + /* + * Background worker state changes set our latch, but they don't signal + * the condition variable. Wait on the latch directly so that a worker + * which fails before initializing cannot leave us asleep forever. + */ + ResetLatch(MyLatch); + CHECK_FOR_INTERRUPTS(); + + /* + * ConditionVariableSignal() removes us from the wait list, while the + * direct latch wait below does not re-add us. Re-register on every + * iteration so we cannot miss an initialization signal. + */ + ConditionVariablePrepareToSleep(&shared->cv); + + /* + * Check the worker state before the shared flag. If the worker sets + * the flag and then exits, the flag still makes initialization + * successful even though the handle is already stopped. + */ + status = GetBackgroundWorkerPid(decoding_worker->handle, &pid); SpinLockAcquire(&shared->mutex); initialized = shared->initialized; @@ -3742,7 +3765,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(); } diff --git a/src/backend/commands/repack_worker.c b/src/backend/commands/repack_worker.c index b4ba9cfc67b..4c35719bc29 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" @@ -70,6 +71,7 @@ RepackWorkerMain(Datum main_arg) am_repack_worker = true; BackgroundWorkerUnblockSignals(); + INJECTION_POINT("repack-worker-before-dsm-attach", NULL); seg = dsm_attach(DatumGetUInt32(main_arg)); if (seg == NULL) @@ -100,6 +102,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 diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 408a35c3c21..0c6649ddab3 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 \ repack_temporal \ repack_temporal_multirange \ repack_toast \ diff --git a/src/test/modules/injection_points/expected/repack_worker.out b/src/test/modules/injection_points/expected/repack_worker.out new file mode 100644 index 00000000000..8370e0a724c --- /dev/null +++ b/src/test/modules/injection_points/expected/repack_worker.out @@ -0,0 +1,135 @@ +Parsed test spec with 2 sessions + +starting permutation: fail_before_attach repack_fail record_failure detach_before_attach repack_retry record_success check +step fail_before_attach: + SELECT injection_points_attach('repack-worker-before-dsm-attach', 'error'); + +injection_points_attach +----------------------- + +(1 row) + +step repack_fail: + REPACK (CONCURRENTLY) repack_worker_test + USING INDEX repack_worker_test_pkey; + +ERROR: REPACK worker failed to initialize +step record_failure: + INSERT INTO repack_worker_nodes + SELECT 'failed', relfilenode FROM pg_class + WHERE oid = 'repack_worker_test'::regclass; + +step detach_before_attach: + SELECT injection_points_detach('repack-worker-before-dsm-attach'); + +injection_points_detach +----------------------- + +(1 row) + +step repack_retry: + REPACK (CONCURRENTLY) repack_worker_test + USING INDEX repack_worker_test_pkey; + +step record_success: + INSERT INTO repack_worker_nodes + SELECT 'success', relfilenode FROM pg_class + WHERE oid = 'repack_worker_test'::regclass; + +step check: + SELECT count(*) AS differences + FROM + ( + (SELECT i, j FROM repack_worker_test + EXCEPT ALL + SELECT i, repeat(i::text, 100) FROM generate_series(1, 100) AS i) + UNION ALL + (SELECT i, repeat(i::text, 100) FROM generate_series(1, 100) AS i + EXCEPT ALL + SELECT i, j FROM repack_worker_test) + ) AS diff; + SELECT before.node = failed.node AS failure_unchanged, + before.node <> success.node AS success_rewritten + FROM repack_worker_nodes before, + repack_worker_nodes failed, + repack_worker_nodes success + WHERE before.phase = 'before' AND failed.phase = 'failed' + AND success.phase = 'success'; + +differences +----------- + 0 +(1 row) + +failure_unchanged|success_rewritten +-----------------+----------------- +t |t +(1 row) + + +starting permutation: fail_after_attach repack_fail record_failure detach_after_attach repack_retry record_success check +step fail_after_attach: + SELECT injection_points_attach('repack-worker-after-error-queue-attach', 'error'); + +injection_points_attach +----------------------- + +(1 row) + +step repack_fail: + REPACK (CONCURRENTLY) repack_worker_test + USING INDEX repack_worker_test_pkey; + +ERROR: error triggered for injection point repack-worker-after-error-queue-attach +step record_failure: + INSERT INTO repack_worker_nodes + SELECT 'failed', relfilenode FROM pg_class + WHERE oid = 'repack_worker_test'::regclass; + +step detach_after_attach: + SELECT injection_points_detach('repack-worker-after-error-queue-attach'); + +injection_points_detach +----------------------- + +(1 row) + +step repack_retry: + REPACK (CONCURRENTLY) repack_worker_test + USING INDEX repack_worker_test_pkey; + +step record_success: + INSERT INTO repack_worker_nodes + SELECT 'success', relfilenode FROM pg_class + WHERE oid = 'repack_worker_test'::regclass; + +step check: + SELECT count(*) AS differences + FROM + ( + (SELECT i, j FROM repack_worker_test + EXCEPT ALL + SELECT i, repeat(i::text, 100) FROM generate_series(1, 100) AS i) + UNION ALL + (SELECT i, repeat(i::text, 100) FROM generate_series(1, 100) AS i + EXCEPT ALL + SELECT i, j FROM repack_worker_test) + ) AS diff; + SELECT before.node = failed.node AS failure_unchanged, + before.node <> success.node AS success_rewritten + FROM repack_worker_nodes before, + repack_worker_nodes failed, + repack_worker_nodes success + WHERE before.phase = 'before' AND failed.phase = 'failed' + AND success.phase = 'success'; + +differences +----------- + 0 +(1 row) + +failure_unchanged|success_rewritten +-----------------+----------------- +t |t +(1 row) + diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index a7b40e084f6..6735c802873 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', 'repack_temporal', 'repack_temporal_multirange', 'repack_toast', diff --git a/src/test/modules/injection_points/specs/repack_worker.spec b/src/test/modules/injection_points/specs/repack_worker.spec new file mode 100644 index 00000000000..61fb0c111d8 --- /dev/null +++ b/src/test/modules/injection_points/specs/repack_worker.spec @@ -0,0 +1,104 @@ +# Test REPACK decoding worker startup failures. +setup +{ + CREATE EXTENSION injection_points; + + CREATE TABLE repack_worker_test(i int PRIMARY KEY, j text); + INSERT INTO repack_worker_test + SELECT i, repeat(i::text, 100) FROM generate_series(1, 100) AS i; + CREATE TABLE repack_worker_nodes(phase text, node oid); + INSERT INTO repack_worker_nodes + SELECT 'before', relfilenode FROM pg_class + WHERE oid = 'repack_worker_test'::regclass; +} + +teardown +{ + DROP TABLE repack_worker_test; + DROP TABLE repack_worker_nodes; + DROP EXTENSION injection_points; +} + +session s1 +step repack_fail +{ + REPACK (CONCURRENTLY) repack_worker_test + USING INDEX repack_worker_test_pkey; +} +step repack_retry +{ + REPACK (CONCURRENTLY) repack_worker_test + USING INDEX repack_worker_test_pkey; +} +step record_failure +{ + INSERT INTO repack_worker_nodes + SELECT 'failed', relfilenode FROM pg_class + WHERE oid = 'repack_worker_test'::regclass; +} +step record_success +{ + INSERT INTO repack_worker_nodes + SELECT 'success', relfilenode FROM pg_class + WHERE oid = 'repack_worker_test'::regclass; +} +step check +{ + SELECT count(*) AS differences + FROM + ( + (SELECT i, j FROM repack_worker_test + EXCEPT ALL + SELECT i, repeat(i::text, 100) FROM generate_series(1, 100) AS i) + UNION ALL + (SELECT i, repeat(i::text, 100) FROM generate_series(1, 100) AS i + EXCEPT ALL + SELECT i, j FROM repack_worker_test) + ) AS diff; + SELECT before.node = failed.node AS failure_unchanged, + before.node <> success.node AS success_rewritten + FROM repack_worker_nodes before, + repack_worker_nodes failed, + repack_worker_nodes success + WHERE before.phase = 'before' AND failed.phase = 'failed' + AND success.phase = 'success'; +} + +session s2 +step fail_before_attach +{ + SELECT injection_points_attach('repack-worker-before-dsm-attach', 'error'); +} +step fail_after_attach +{ + SELECT injection_points_attach('repack-worker-after-error-queue-attach', 'error'); +} +step detach_before_attach +{ + SELECT injection_points_detach('repack-worker-before-dsm-attach'); +} +step detach_after_attach +{ + SELECT injection_points_detach('repack-worker-after-error-queue-attach'); +} + +# A worker error before DSM and error-queue attachment cannot be propagated, +# so the leader reports a generic initialization failure and can retry. +permutation + fail_before_attach + repack_fail + record_failure + detach_before_attach + repack_retry + record_success + check + +# Once the error queue is attached, preserve the worker's original error. +permutation + fail_after_attach + repack_fail + record_failure + detach_after_attach + repack_retry + record_success + check -- 2.50.1 (Apple Git-155)