From 253bc59af350169a6d4f2654a953319a203b5c56 Mon Sep 17 00:00:00 2001 From: Kirill Reshke Date: Mon, 19 Oct 2026 18:54:48 +0500 Subject: [PATCH v1] Allow physical WAL senders to restore missing segments Add wal_sender_restore to let a physical replication sender fetch missing WAL with restore_command on either a primary or a standby. Restore one segment per sender into a private temporary directory, without exposing it to recovery, archiving, or checkpoint recycling. Remove the directory on close or error, and remove leftovers during crash recovery. Retain the post-read recycling check for ordinary local WAL. Delay requests arriving within one second of a failed restore, including after reconnecting, before attempting the restore once. Keep existing timeline bounds and replication slot invalidation rules. Restore commands still run synchronously, so slow commands can delay shutdown or cause the receiver to time out. Add recovery tests for primary and cascading senders, concurrent restores, the failure delay, error cleanup, crash cleanup, and the transition back to live WAL. Thanks to Fujii Masao for the earlier proposal and to Alexander Kukushkin for pointing out the cost of repeated failed restores. Author: Japin Li Author: Kirill Reshke Author: Andrey Borodin Discussion: https://www.postgresql.org/message-id/ME0P300MB0445C80039D6534452A87539B657A@ME0P300MB0445.AUSP300.PROD.OUTLOOK.COM --- doc/src/sgml/config.sgml | 70 ++++- src/backend/access/transam/xlog.c | 6 +- src/backend/access/transam/xlogarchive.c | 17 +- src/backend/replication/walsender.c | 189 +++++++++++++- .../utils/activity/wait_event_names.txt | 1 + src/backend/utils/misc/guc_parameters.dat | 6 + src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/replication/walsender.h | 1 + src/include/replication/walsender_private.h | 4 + src/test/recovery/meson.build | 1 + src/test/recovery/t/061_walsender_archive.pl | 243 ++++++++++++++++++ src/test/recovery/t/walsender_restore | 27 ++ 12 files changed, 551 insertions(+), 15 deletions(-) create mode 100644 src/test/recovery/t/061_walsender_archive.pl create mode 100644 src/test/recovery/t/walsender_restore diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index f7db2995c3e..04160f2d53e 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4260,9 +4260,10 @@ include_dir 'conf.d' - This section describes the settings that apply only for the duration of - the recovery. They must be reset for any subsequent recovery you wish to - perform. + This section describes the settings that apply for the duration of + recovery. They must be reset for any subsequent recovery you wish to + perform. The restore_command setting is also used + outside recovery when is enabled. @@ -4325,6 +4326,13 @@ include_dir 'conf.d' Write %% to embed an actual % character. + + When is enabled, this command + also supplies missing WAL to physical replication clients, including + outside recovery. See that setting for the requirements on the + command in this case. + + It is important for the command to return a zero exit status only if it succeeds. The command will be asked for file @@ -5020,6 +5028,62 @@ HINT: If it is safe for all REPLICATION users to use this library as an output + + wal_sender_restore (boolean) + + wal_sender_restore configuration parameter + + + + + Allows physical replication senders to use + when a requested WAL segment is + missing from pg_wal. This applies to both primary + and standby servers. The default is off. + This parameter can only be set in the postgresql.conf + file or on the server command line. + + + Each sender restores one segment at a time into a private temporary + directory in pg_wal. These copies are not used + by recovery or archived again, and checkpoints do not recycle them. + They are removed when no longer needed or when the sender exits. + Crash recovery removes leftover directories. The restore command + must not recreate the destination directory or leave background + processes writing to the destination after the command exits. + It must also remove any temporary files it creates in that directory + other than the requested destination file. + The command must support concurrent invocations by different senders + and, on a standby, by recovery. + The %r substitution is an all-zero filename, so + the command must not remove archived WAL on behalf of the sender. + + + Requests arriving within one second of an archive restore failure + wait for the remainder of that interval before invoking the restore + command. The failure timestamp is shared across senders, so this + delay also applies after reconnecting. Each request attempts the + restore only once, and senders can restore files concurrently. + Requests for other archived segments are also delayed, but reads + from pg_wal and restoration by recovery are + unaffected. + + + The restore command runs synchronously. While it runs, the sender + cannot send keepalive messages or process replication feedback. + A slow command can therefore cause the receiver to time out and + can delay server shutdown. The command should bound its execution + time appropriately. + + + This does not change which timeline or WAL positions may be streamed, + enable logical decoding from archived WAL, or make an invalidated + replication slot usable again. If the segment cannot be restored, + the replication command fails as it does when this setting is disabled. + + + + wal_sender_timeout (integer) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 8636ad9ab2c..b9460a20852 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -3963,7 +3963,11 @@ RemoveTempXlogFiles(void) continue; snprintf(path, MAXPGPATH, XLOGDIR "/%s", xlde->d_name); - unlink(path); + /* Walsenders restore archived WAL into private temporary directories. */ + if (get_dirent_type(path, xlde, false, LOG) == PGFILETYPE_DIR) + rmtree(path, true); + else + unlink(path); elog(DEBUG2, "removed temporary WAL segment \"%s\"", path); } FreeDir(xldir); diff --git a/src/backend/access/transam/xlogarchive.c b/src/backend/access/transam/xlogarchive.c index 0d9ffc99726..08209c872c5 100644 --- a/src/backend/access/transam/xlogarchive.c +++ b/src/backend/access/transam/xlogarchive.c @@ -49,7 +49,8 @@ * * When 'cleanupEnabled' is false, refrain from deleting any old WAL segments * in the archive. This is used when fetching the initial checkpoint record, - * when we are not yet sure how far back we need the WAL. + * when we are not yet sure how far back we need the WAL, and by walsenders. + * A walsender must supply a private recovername and clean it up afterwards. */ bool RestoreArchivedFile(char *path, const char *xlogfname, @@ -67,10 +68,12 @@ RestoreArchivedFile(char *path, const char *xlogfname, /* * Ignore restore_command when not in archive recovery (meaning we are in - * crash recovery). + * crash recovery), except when serving archived WAL to a replication + * client. */ - if (!ArchiveRecoveryRequested) + if (!ArchiveRecoveryRequested && !am_walsender) goto not_available; + Assert(!am_walsender || !cleanupEnabled); /* In standby mode, restore_command might not be supplied */ if (recoveryRestoreCommand == NULL || strcmp(recoveryRestoreCommand, "") == 0) @@ -170,17 +173,21 @@ RestoreArchivedFile(char *path, const char *xlogfname, * it is best to put any additional logic before or after the * PreRestoreCommand()/PostRestoreCommand() section. */ - PreRestoreCommand(); + if (AmStartupProcess()) + PreRestoreCommand(); /* * Copy xlog from archival storage to XLOGDIR */ rc = system(xlogRestoreCmd); - PostRestoreCommand(); + if (AmStartupProcess()) + PostRestoreCommand(); pgstat_report_wait_end(); pfree(xlogRestoreCmd); + if (!AmStartupProcess()) + CHECK_FOR_INTERRUPTS(); if (rc == 0) { diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 9f21c7c8dfe..fad4b285e3d 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -56,6 +56,7 @@ #include "access/twophase.h" #include "access/xact.h" #include "access/xlog_internal.h" +#include "access/xlogarchive.h" #include "access/xlogreader.h" #include "access/xlogrecovery.h" #include "access/xlogutils.h" @@ -94,6 +95,7 @@ #include "utils/acl.h" #include "utils/builtins.h" #include "utils/guc.h" +#include "utils/injection_point.h" #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/pg_lsn.h" @@ -148,6 +150,7 @@ int wal_sender_shutdown_timeout = -1; /* maximum time to wait during * replication */ bool log_replication_commands = false; +bool wal_sender_restore = false; /* * State for WalSndWakeupRequest @@ -161,6 +164,14 @@ bool wake_wal_senders = false; */ static XLogReaderState *xlogreader = NULL; +/* Private archive copy for physical streaming, never installed as local WAL. */ +static char restoreDir[MAXPGPATH]; +static int restoredFile = -1; +static bool restoreInProgress = false; + +/* Minimum delay after a failed archive restore, in milliseconds. */ +#define WALSND_RESTORE_DELAY 1000 + /* * If the UPLOAD_MANIFEST command is used to provide a backup manifest in * preparation for an incremental backup, uploaded_manifest will be point @@ -323,6 +334,9 @@ static bool TransactionIdInRecentPast(TransactionId xid, uint32 epoch); static void WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p); +static void WalSndSegmentClose(XLogReaderState *state); +static bool WalSndRestoreSegment(XLogReaderState *state, const char *xlogfname); +static void WalSndRestoreCleanup(int code, Datum arg); /* Initialize walsender process before entering the main command loop */ @@ -333,6 +347,7 @@ InitWalSender(void) /* Create a per-walsender data structure in shared memory */ InitWalSenderSlot(); + on_shmem_exit(WalSndRestoreCleanup, 0); /* need resource owner for e.g. basebackups */ CreateAuxProcessResourceOwner(); @@ -382,7 +397,10 @@ WalSndErrorCleanup(void) pgaio_error_cleanup(); if (xlogreader != NULL && xlogreader->seg.ws_file >= 0) - wal_segment_close(xlogreader); + WalSndSegmentClose(xlogreader); + + /* The restore command can fail before a segment has been opened. */ + WalSndRestoreCleanup(0, 0); xlogreader = NULL; logical_decoding_ctx = NULL; @@ -870,7 +888,7 @@ StartReplication(StartReplicationCmd *cmd) xlogreader = XLogReaderAllocate(wal_segment_size, NULL, XL_ROUTINE(.segment_open = WalSndSegmentOpen, - .segment_close = wal_segment_close), + .segment_close = WalSndSegmentClose), NULL); if (!xlogreader) @@ -3371,6 +3389,13 @@ WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo, int save_errno = errno; XLogFileName(xlogfname, *tli_p, nextSegNo, wal_segment_size); + + /* Only physical streaming may fall back to the archive. */ + if (wal_sender_restore && + state->routine.segment_close == WalSndSegmentClose && + WalSndRestoreSegment(state, xlogfname)) + return; + errno = save_errno; ereport(ERROR, (errcode_for_file_access(), @@ -3384,6 +3409,144 @@ WalSndSegmentOpen(XLogReaderState *state, XLogSegNo nextSegNo, path))); } +/* Remove a private archive copy on normal close, ERROR, or process exit. */ +static void +WalSndRestoreCleanup(int code, Datum arg) +{ + /* Also handle failures that leave RestoreArchivedFile via ERROR or FATAL. */ + if (restoreInProgress) + { + TimestampTz now = GetCurrentTimestamp(); + + SpinLockAcquire(&WalSndCtl->restore_mutex); + WalSndCtl->lastRestoreFailure = now; + SpinLockRelease(&WalSndCtl->restore_mutex); + restoreInProgress = false; + } + + if (restoredFile >= 0) + { + close(restoredFile); + restoredFile = -1; + } + if (restoreDir[0] != '\0') + { + char path[MAXPGPATH]; + + /* + * Do not use rmtree(): this can run from a memory context reset + * callback, where its allocations could raise another error. + */ + snprintf(path, sizeof(path), "%s/RECOVERYXLOG", restoreDir); + if (unlink(path) != 0 && errno != ENOENT) + ereport(WARNING, + (errcode_for_file_access(), + errmsg("could not remove file \"%s\": %m", path))); + if (rmdir(restoreDir) != 0 && errno != ENOENT) + ereport(WARNING, + (errcode_for_file_access(), + errmsg("could not remove directory \"%s\": %m", restoreDir))); + restoreDir[0] = '\0'; + } +} + +/* XLogReaderRoutine->segment_close callback for physical streaming. */ +static void +WalSndSegmentClose(XLogReaderState *state) +{ + if (restoredFile >= 0) + { + Assert(state->seg.ws_file == restoredFile); + state->seg.ws_file = -1; + WalSndRestoreCleanup(0, 0); + } + else + wal_segment_close(state); +} + +/* + * Restore one segment without exposing it to checkpoint recycling, archiving, + * recovery, or other walsenders. RemoveTempXlogFiles removes the directory + * after a crash. Using a private directory also prevents a surviving restore + * command from recreating its destination after crash cleanup, provided the + * command does not itself create the destination directory. + */ +static bool +WalSndRestoreSegment(XLogReaderState *state, const char *xlogfname) +{ + char dir[MAXPGPATH]; + char recovername[MAXPGPATH]; + char path[MAXPGPATH]; + TimestampTz delayUntil; + + if (recoveryRestoreCommand == NULL || recoveryRestoreCommand[0] == '\0') + return false; + + Assert(restoredFile < 0 && restoreDir[0] == '\0'); + Assert(!restoreInProgress); + + /* + * Delay requests arriving within one second of a failed restore, even + * after reconnecting. Take a snapshot of the deadline so failures in + * other senders cannot keep extending this request's wait. We attempt the + * restore once after the delay, without serializing senders. + */ + SpinLockAcquire(&WalSndCtl->restore_mutex); + delayUntil = WalSndCtl->lastRestoreFailure; + SpinLockRelease(&WalSndCtl->restore_mutex); + delayUntil = TimestampTzPlusMilliseconds(delayUntil, WALSND_RESTORE_DELAY); + for (;;) + { + long delay; + + CHECK_FOR_INTERRUPTS(); + WalSndCheckShutdownTimeout(); + + delay = TimestampDifferenceMilliseconds(GetCurrentTimestamp(), delayUntil); + /* Do not prolong the wait after a backwards clock adjustment. */ + if (delay == 0 || delay > WALSND_RESTORE_DELAY) + break; + + (void) WaitLatch(MyLatch, + WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, + delay, WAIT_EVENT_WAL_SENDER_RESTORE_DELAY); + ResetLatch(MyLatch); + } + + restoreInProgress = true; + snprintf(recovername, sizeof(recovername), + "xlogtemp.%d." INT64_FORMAT ".walsender", MyProcPid, + MyStartTimestamp); + snprintf(dir, sizeof(dir), XLOGDIR "/%s", recovername); + if (MakePGDirectory(dir) < 0) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not create directory \"%s\": %m", dir))); + strlcpy(restoreDir, dir, sizeof(restoreDir)); + strlcat(recovername, "/RECOVERYXLOG", sizeof(recovername)); + + /* WalSndErrorCleanup and the exit callback handle failures from here on. */ + INJECTION_POINT("walsender-before-archive-restore", NULL); + if (!RestoreArchivedFile(path, xlogfname, recovername, + wal_segment_size, false)) + { + WalSndRestoreCleanup(0, 0); + return false; + } + + restoredFile = BasicOpenFile(path, O_RDONLY | PG_BINARY); + if (restoredFile < 0) + ereport(ERROR, + (errcode_for_file_access(), + errmsg("could not open file \"%s\": %m", path))); + state->seg.ws_file = restoredFile; + XLogReaderRegisterResetCallback(state); + + restoreInProgress = false; + INJECTION_POINT("walsender-after-archive-restore", NULL); + return true; +} + /* * Send out the WAL in its normal physical/stored form. * @@ -3546,7 +3709,7 @@ XLogSendPhysical(void) { /* close the current file. */ if (xlogreader->seg.ws_file >= 0) - wal_segment_close(xlogreader); + WalSndSegmentClose(xlogreader); /* Send CopyDone */ pq_putmessage_noblock(PqMsg_CopyDone, NULL, 0); @@ -3583,6 +3746,13 @@ XLogSendPhysical(void) endptr = startptr; endptr += MAX_SEND_SIZE; + /* + * Keep each read in one segment so its local/archive origin is + * unambiguous. + */ + endptr = Min(endptr, startptr + wal_segment_size - + XLogSegmentOffset(startptr, wal_segment_size)); + /* if we went beyond SendRqstPtr, back off */ if (SendRqstPtr <= endptr) { @@ -3640,7 +3810,8 @@ retry: /* See logical_read_xlog_page(). */ XLByteToSeg(startptr, segno, xlogreader->segcxt.ws_segsize); - CheckXLogRemoved(segno, xlogreader->seg.ws_tli); + if (nbytes == 0 || restoredFile < 0) + CheckXLogRemoved(segno, xlogreader->seg.ws_tli); /* * During recovery, the currently-open WAL file might be replaced with the @@ -3658,9 +3829,9 @@ retry: walsnd->needreload = false; SpinLockRelease(&walsnd->mutex); - if (reload && xlogreader->seg.ws_file >= 0) + if (reload && xlogreader->seg.ws_file >= 0 && restoredFile < 0) { - wal_segment_close(xlogreader); + WalSndSegmentClose(xlogreader); goto retry; } @@ -3669,6 +3840,10 @@ retry: output_message.len += nbytes; output_message.data[output_message.len] = '\0'; + /* Do not retain a complete archive segment while waiting for more WAL. */ + if (restoredFile >= 0 && XLogSegmentOffset(endptr, wal_segment_size) == 0) + WalSndSegmentClose(xlogreader); + /* * Fill the send timestamp last, so that it is taken as late as possible. */ @@ -4054,6 +4229,8 @@ WalSndShmemRequest(void *arg) static void WalSndShmemInit(void *arg) { + SpinLockInit(&WalSndCtl->restore_mutex); + for (int i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++) dlist_init(&(WalSndCtl->SyncRepQueue[i])); diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index 3d366fd1114..ff62f2ce133 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -188,6 +188,7 @@ REGISTER_SYNC_REQUEST "Waiting while sending synchronization requests to the che SPIN_DELAY "Waiting while acquiring a contended spinlock." VACUUM_DELAY "Waiting in a cost-based vacuum delay point." VACUUM_TRUNCATE "Waiting to acquire an exclusive lock to truncate off any empty pages at the end of a table vacuumed." +WAL_SENDER_RESTORE_DELAY "Waiting before restoring WAL after an archive restore failure." WAL_SUMMARIZER_ERROR "Waiting after a WAL summarizer error." ABI_compatibility: diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index 380679a76c0..66ab99a6f36 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -3573,6 +3573,12 @@ check_hook => 'check_wal_segment_size', }, +{ name => 'wal_sender_restore', type => 'bool', context => 'PGC_SIGHUP', group => 'REPLICATION_SENDING', + short_desc => 'Allows physical WAL senders to restore missing WAL from the archive.', + variable => 'wal_sender_restore', + boot_val => 'false', +}, + { name => 'wal_sender_shutdown_timeout', type => 'int', context => 'PGC_USERSET', group => 'REPLICATION_SENDING', short_desc => 'Sets the maximum time the server waits during shutdown for all WAL data to be replicated to the receiver.', long_desc => '-1 means wait indefinitely. 0 means do not wait.', diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 01f98de8a9f..8cc9671eb00 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -363,6 +363,7 @@ #max_slot_wal_keep_size = -1 # in megabytes; -1 disables #idle_replication_slot_timeout = 0 # in seconds; 0 disables #wal_sender_timeout = 60s # in milliseconds; 0 disables +#wal_sender_restore = off # fetch missing WAL using restore_command #wal_sender_shutdown_timeout = -1 # in milliseconds # -1 disables the timeout and waits for catch-up # 0 does not wait for catch-up diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h index 386cedfc7aa..de88a71d65c 100644 --- a/src/include/replication/walsender.h +++ b/src/include/replication/walsender.h @@ -35,6 +35,7 @@ extern PGDLLIMPORT int max_wal_senders; extern PGDLLIMPORT int wal_sender_timeout; extern PGDLLIMPORT int wal_sender_shutdown_timeout; extern PGDLLIMPORT bool log_replication_commands; +extern PGDLLIMPORT bool wal_sender_restore; extern void InitWalSender(void); extern bool exec_replication_command(const char *cmd_string); diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h index b0c80deeb24..966332812ac 100644 --- a/src/include/replication/walsender_private.h +++ b/src/include/replication/walsender_private.h @@ -113,6 +113,10 @@ typedef struct */ ConditionVariable wal_confirm_rcv_cv; + /* Time of last archive restore failure, protected by restore_mutex. */ + slock_t restore_mutex; + TimestampTz lastRestoreFailure; + WalSnd walsnds[FLEXIBLE_ARRAY_MEMBER]; } WalSndCtlData; diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index b6ac9dd07a7..dbf88060151 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -69,6 +69,7 @@ tests += { 't/058_shutdown_crash_restart.pl', 't/059_remote_apply_status_interval.pl', 't/060_multixact_crashed_updater.pl', + 't/061_walsender_archive.pl', ], }, } diff --git a/src/test/recovery/t/061_walsender_archive.pl b/src/test/recovery/t/061_walsender_archive.pl new file mode 100644 index 00000000000..2a85ee181e7 --- /dev/null +++ b/src/test/recovery/t/061_walsender_archive.pl @@ -0,0 +1,243 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# A physical sender can fetch old WAL without putting it back in the normal +# WAL namespace. Exercise primary and cascading senders, and cleanup after +# a failed restore, normal streaming, and crash restart. +use strict; +use warnings FATAL => 'all'; + +use FindBin; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +my $helper = "$FindBin::RealBin/walsender_restore"; +my $have_injection_points = $ENV{enable_injection_points} eq 'yes'; + +# This is a directory count, not a count of normally named WAL segments. +my $scratch_count = q{SELECT count(*) FROM pg_ls_dir('pg_wal') AS f + WHERE f LIKE 'xlogtemp.%.walsender'}; + +for my $cascading (0, 1) +{ + my $label = $cascading ? 'cascading' : 'primary'; + my $primary = PostgreSQL::Test::Cluster->new("${label}_primary"); + $primary->init( + allows_streaming => 1, + has_archiving => 1, + extra => ['--wal-segsize=1']); + $primary->append_conf( + 'postgresql.conf', qq( +min_wal_size = '2MB' +max_wal_size = '4MB' +wal_keep_size = 0 +checkpoint_timeout = '1h' +)); + $primary->start; + $primary->safe_psql('postgres', 'CREATE TABLE archive_test (i int)'); + $primary->backup('base'); + + my $sender = $primary; + if ($cascading) + { + $sender = PostgreSQL::Test::Cluster->new('cascading_sender'); + $sender->init_from_backup($primary, 'base', has_streaming => 1); + $sender->start; + } + + # Two receivers also exercise independent scratch copies in the primary + # case when injection points are available. + my @receivers; + for my $i (1 .. 2) + { + my $receiver = PostgreSQL::Test::Cluster->new("${label}_receiver_$i"); + $receiver->init_from_backup($primary, 'base', has_streaming => 1); + $receiver->enable_streaming($sender) if $cascading; + $receiver->append_conf( + 'postgresql.conf', qq( +restore_command = '' +wal_retrieve_retry_interval = '100ms' +)); + $receiver->start; + push @receivers, $receiver; + } + $primary->safe_psql('postgres', 'INSERT INTO archive_test VALUES (1)'); + my $initial_lsn = $primary->lsn('insert'); + $primary->wait_for_catchup($sender, 'replay', $initial_lsn) if $cascading; + for my $receiver (@receivers) + { + $sender->wait_for_catchup($receiver, 'replay', $initial_lsn); + $receiver->stop; + } + my $old_segment = $primary->safe_psql('postgres', + "SELECT pg_walfile_name('$initial_lsn')"); + + # Produce and archive enough WAL for checkpoint/restartpoint cleanup to + # remove the segment at which the receivers stopped. + $primary->safe_psql('postgres', 'INSERT INTO archive_test VALUES (2)'); + $primary->advance_wal(10); + $primary->safe_psql('postgres', 'CHECKPOINT'); + my $target_lsn = $primary->lsn('insert'); + my $archive = $primary->archive_dir; + $primary->poll_query_until('postgres', + "SELECT last_archived_wal > '$old_segment' FROM pg_stat_archiver") + or die 'archive did not advance'; + $primary->safe_psql('postgres', 'CHECKPOINT'); + + if ($cascading) + { + $primary->wait_for_catchup($sender, 'replay', + $primary->lsn('insert')); + $sender->safe_psql('postgres', 'CHECKPOINT'); + } + ok( !-e $sender->data_dir . "/pg_wal/$old_segment", + "$label: required WAL really left pg_wal"); + ok(-f "$archive/$old_segment", "$label: required WAL is archived"); + + my $restore_command = qq{"$^X" "$helper" "$archive" "%f" "%p" "%r"}; + $restore_command =~ s/\\/\\\\/g; + $restore_command =~ s/'/''/g; + $sender->append_conf('postgresql.conf', + "restore_command = '$restore_command'"); + $sender->reload; + + # Configuring restore_command alone does not change existing behavior. + my $log_offset = -s $sender->logfile; + $receivers[0]->start; + $sender->wait_for_log( + qr/requested WAL segment .* has already been removed/, $log_offset); + $receivers[0]->stop; + is($sender->safe_psql('postgres', $scratch_count), + '0', "$label: disabled fallback creates no scratch copies"); + + # Failed size checks must leave no private directories. The receiver + # reconnects automatically, but its next restore must still be delayed. + append_to_file("$archive/short", ''); + $sender->append_conf('postgresql.conf', 'wal_sender_restore = on'); + $sender->reload; + $log_offset = -s $sender->logfile; + $receivers[0]->start; + $sender->wait_for_log( + qr/archive file [^\n]* has wrong size.*archive file [^\n]* has wrong size/s, + $log_offset); + $receivers[0]->stop; + my @attempts = split /\n/, slurp_file("$archive/short_attempts"); + cmp_ok($attempts[1] - $attempts[0], + '>=', 1, "$label: failed restores are delayed across reconnections"); + unlink("$archive/short") or die "unlink short: $!"; + $sender->poll_query_until('postgres', $scratch_count, '0') + or die 'failed restore left a scratch directory'; + + if ($have_injection_points && !$cascading) + { + $sender->safe_psql('postgres', 'CREATE EXTENSION injection_points'); + + # A previous failure must not serialize later restores. Hold both + # senders before restore_command, so neither can succeed first. + my $restore_point = 'walsender-before-archive-restore'; + my $restore_waiters = qq{SELECT count(*) FROM pg_stat_activity + WHERE wait_event = '$restore_point'}; + $sender->safe_psql('postgres', + "SELECT injection_points_attach('$restore_point', 'wait')"); + $_->start for @receivers; + $sender->poll_query_until('postgres', $restore_waiters, '2') + or die 'senders did not reach archive restore point'; + is($sender->safe_psql('postgres', $scratch_count), + '2', 'restores after a failure can proceed concurrently'); + + $_->stop for @receivers; + $sender->safe_psql( + 'postgres', + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity + WHERE wait_event = '$restore_point'"); + $sender->poll_query_until('postgres', $scratch_count, '0') + or die 'terminated restores left scratch directories'; + $sender->safe_psql('postgres', + "SELECT injection_points_detach('$restore_point')"); + + # ERROR can occur with only a scratch directory, or after opening the + # restored segment. Neither case should prevent a later restore. + for my $point ( + 'walsender-before-archive-restore', + 'walsender-after-archive-restore') + { + $sender->safe_psql('postgres', + "SELECT injection_points_attach('$point', 'error')"); + $log_offset = -s $sender->logfile; + $receivers[0]->start; + $sender->wait_for_log( + qr/error triggered for injection point \Q$point\E/, + $log_offset); + $receivers[0]->stop; + $sender->poll_query_until('postgres', $scratch_count, '0') + or die "$point left a scratch directory"; + $sender->safe_psql('postgres', + "SELECT injection_points_detach('$point')"); + } + + $sender->safe_psql('postgres', + "SELECT injection_points_attach('walsender-after-archive-restore', 'wait')" + ); + } + + $_->start for @receivers; + if ($have_injection_points && !$cascading) + { + $sender->poll_query_until( + 'postgres', + q{SELECT count(*) = 2 FROM pg_stat_activity + WHERE wait_event = 'walsender-after-archive-restore'} + ) or die 'senders did not reach archive wait point'; + is($sender->safe_psql('postgres', $scratch_count), + '2', 'two senders have separate archive copies'); + $sender->safe_psql('postgres', 'CHECKPOINT'); + is($sender->safe_psql('postgres', $scratch_count), + '2', 'checkpoint leaves private archive copies alone'); + + # Leave real restored files behind with an immediate shutdown, then + # verify startup removes them before either receiver reconnects. + $_->stop for @receivers; + $sender->stop('immediate'); + opendir(my $wal_dir, $sender->data_dir . '/pg_wal') + or die "open WAL directory: $!"; + my @leftovers = + grep { /^xlogtemp\..*\.walsender$/ } readdir($wal_dir); + closedir($wal_dir); + is(scalar @leftovers, 2, 'crash leaves the two private directories'); + $sender->start; + is($sender->safe_psql('postgres', $scratch_count), + '0', 'crash restart removes private archive directories'); + $_->start for @receivers; + } + + for my $receiver (@receivers) + { + $sender->wait_for_catchup($receiver, 'replay', $target_lsn); + is( $receiver->safe_psql( + 'postgres', 'SELECT sum(i) FROM archive_test'), + '3', + "$label: receiver replayed WAL delivered from archive"); + } + ok(!-e $sender->data_dir . "/pg_wal/$old_segment", + "$label: restored WAL was not installed under its original name"); + $sender->poll_query_until('postgres', $scratch_count, '0') + or die 'completed archive reads left scratch directories'; + + # Verify that the same connections continue with ordinary live WAL. + $primary->safe_psql('postgres', 'INSERT INTO archive_test VALUES (3)'); + my $live_lsn = $primary->lsn('insert'); + $primary->wait_for_catchup($sender, 'replay', $live_lsn) if $cascading; + for my $receiver (@receivers) + { + $sender->wait_for_catchup($receiver, 'replay', $live_lsn); + is( $receiver->safe_psql( + 'postgres', 'SELECT sum(i) FROM archive_test'), + '6', + "$label: receiver continues with live WAL"); + $receiver->stop; + } + $sender->stop if $cascading; + $primary->stop; +} + +done_testing(); diff --git a/src/test/recovery/t/walsender_restore b/src/test/recovery/t/walsender_restore new file mode 100644 index 00000000000..11ea118205e --- /dev/null +++ b/src/test/recovery/t/walsender_restore @@ -0,0 +1,27 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# A restore_command for 061_walsender_archive.pl. Never create the destination +# directory: the sender owns its lifetime, including crash cleanup. +use strict; +use warnings FATAL => 'all'; +use File::Copy; +use Time::HiRes qw(time); + +my ($archive, $name, $destination, $restart) = @ARGV; +my $for_sender = $destination =~ /xlogtemp\./; +die "unexpected archive cleanup request: $restart" + if $for_sender && $restart ne '0' x 24; + +if ($for_sender && -e "$archive/short") +{ + open(my $file, '>', $destination) or die "open $destination: $!"; + print $file 'short'; + close($file) or die "close $destination: $!"; + open(my $attempts, '>>', "$archive/short_attempts") + or die "open short_attempts: $!"; + printf $attempts "%.6f\n", time; + close($attempts) or die "close short_attempts: $!"; + exit 0; +} + +exit(copy("$archive/$name", $destination) ? 0 : 1); -- That's all, folks. May the source be with you.