From 74ff3a7a3d827e5c5ee24b885e3711b47a402a69 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Tue, 15 Sep 2026 06:25:59 +0000 Subject: [PATCH v5 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 | 240 ++++++++++++++++++ 4 files changed, 293 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..077c109072f --- /dev/null +++ b/src/test/modules/test_misc/t/100_repack_worker.pl @@ -0,0 +1,240 @@ +# 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. + +# 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'); + 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'); + 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'); + 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.47.3