From bf9123ca2d8ee5c580ed10d1a51766d6ff4af923 Mon Sep 17 00:00:00 2001 From: Alexander Korotkov Date: Wed, 11 Mar 2026 15:46:44 +0200 Subject: [PATCH v9] Introduce a new 'wal_sender_shutdown_timeout' GUC Previously, during shutdown, walsenders always waited until all pending data was replicated to receivers. This ensures sender and receiver stay in sync after shutdown, which is important for physical replication switchovers, but it can significantly delay shutdown. For example, in logical replication, if apply workers are blocked on locks, walsenders may wait until those locks are released, preventing shutdown from completing for a long time. This commit introduces a new GUC, wal_sender_shutdown_timeout, which specifies the maximum time a walsender waits during shutdown for all pending data to be replicated. When set, shutdown completes once all data is replicated or the timeout expires. A value of -1 (the default) disables the timeout. This can reduce shutdown time when replication is slow or stalled. However, if the timeout is reached, the sender and receiver may be left out of sync, which can be problematic for physical replication switchovers. Discussion: https://postgr.es/m/TYAPR01MB586668E50FC2447AD7F92491F5E89%40TYAPR01MB5866.jpnprd01.prod.outlook.com Author: Andrey Silitskiy Co-authored-by: Hayato Kuroda Reviewed-by: Ashutosh Bapat Reviewed-by: Kyotaro Horiguchi Reviewed-by: Amit Kapila Reviewed-by: Dilip Kumar Reviewed-by: Masahiko Sawada Reviewed-by: Andres Freund Reviewed-by: Takamichi Osumi Reviewed-by: Peter Smith Reviewed-by: Greg Sabino Mullane Reviewed-by: Vitaly Davydov Reviewed-by: Fujii Masao Reviewed-by: Ronan Dunklau Reviewed-by: Michael Paquier Reviewed-by: Japin Li --- doc/src/sgml/config.sgml | 32 +++ doc/src/sgml/high-availability.sgml | 9 +- src/backend/replication/walsender.c | 145 ++++++++++- src/backend/utils/misc/guc_parameters.dat | 10 + src/backend/utils/misc/postgresql.conf.sample | 4 + src/include/replication/walsender.h | 1 + src/test/subscription/meson.build | 1 + .../t/038_walsnd_shutdown_timeout.pl | 237 ++++++++++++++++++ 8 files changed, 433 insertions(+), 6 deletions(-) create mode 100644 src/test/subscription/t/038_walsnd_shutdown_timeout.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 229f41353eb..f0f2310e315 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4791,6 +4791,38 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows + + wal_sender_shutdown_timeout (integer) + + wal_sender_shutdown_timeout configuration parameter + + + + + Specifies the maximum period of time the walsender process will wait + for a successful flush of WAL data by the receiver after receipt of + a shutdown request. If this value is specified without units, it is + taken as milliseconds. A value of -1 (the default) disables the + timeout mechanism. This parameter can be set per replication connection. + + + When replication is in use, the sending server normally waits until + all WAL data has been transferred to the receiver before completing + shutdown. This helps keep the sender and receiver in sync after + shutdown, which is especially important for physical replication + switchovers. However, it can delay server shutdown. + + + If this parameter is set, the server stops waiting and completes + shutdown when the timeout expires. This can shorten shutdown time, for + example, when replication is slow on high-latency networks or when a + logical replication apply worker is blocked waiting for locks. + However, in this case the sender and receiver may be out of sync after + shutdown. + + + + track_commit_timestamp (boolean) diff --git a/doc/src/sgml/high-availability.sgml b/doc/src/sgml/high-availability.sgml index c3f269e0364..d03118d9aa4 100644 --- a/doc/src/sgml/high-availability.sgml +++ b/doc/src/sgml/high-availability.sgml @@ -1190,10 +1190,11 @@ primary_slot_name = 'node_a_slot' - Users will stop waiting if a fast shutdown is requested. However, as - when using asynchronous replication, the server will not fully - shutdown until all outstanding WAL records are transferred to the currently - connected standby servers. + Users will stop waiting if a fast shutdown is requested. However, when + using replication, the server will not fully shutdown until all + outstanding WAL records are transferred to the currently connected standby + servers, or (if set) expires, regardless of + whether replication is synchronous or asynchronous. diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index 66507e9c2dd..7f2faba7e79 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -35,6 +35,8 @@ * checkpoint finishes, the postmaster sends us SIGUSR2. This instructs * walsender to send any outstanding WAL, including the shutdown checkpoint * record, wait for it to be replicated to the standby, and then exit. + * This waiting time can be limited by the wal_sender_shutdown_timeout + * parameter. * * * Portions Copyright (c) 2010-2026, PostgreSQL Global Development Group @@ -131,6 +133,11 @@ int max_wal_senders = 10; /* the maximum number of concurrent * walsenders */ int wal_sender_timeout = 60 * 1000; /* maximum time to send one WAL * data message */ + +int wal_sender_shutdown_timeout = -1; /* maximum time to wait for + * flush by receiver after + * shutdown request */ + bool log_replication_commands = false; /* @@ -190,6 +197,11 @@ static TimestampTz last_reply_timestamp = 0; /* Have we sent a heartbeat message asking for reply, since last reply? */ static bool waiting_for_ping_response = false; +/* + * Timestamp of receipt of shutdown request by walsender. + */ +static TimestampTz shutdown_request_timestamp = 0; + /* * While streaming WAL in Copy mode, streamingDoneSending is set to true * after we have sent CopyDone. We should not send any more CopyData messages @@ -263,6 +275,7 @@ static void WalSndKill(int code, Datum arg); pg_noreturn static void WalSndShutdown(void); static void XLogSendPhysical(void); static void XLogSendLogical(void); +pg_noreturn static void WalSndDoneImmediate(void); static void WalSndDone(WalSndSendDataCallback send_data); static void IdentifySystem(void); static void UploadManifest(void); @@ -282,6 +295,7 @@ static void ProcessPendingWrites(void); static void WalSndKeepalive(bool requestReply, XLogRecPtr writePtr); static void WalSndKeepaliveIfNecessary(void); static void WalSndCheckTimeOut(void); +static void WalSndCheckShutdownTimeout(void); static long WalSndComputeSleeptime(TimestampTz now); static void WalSndWait(uint32 socket_events, long timeout, uint32 wait_event); static void WalSndPrepareWrite(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid, bool last_write); @@ -1660,6 +1674,12 @@ ProcessPendingWrites(void) /* die if timeout was reached */ WalSndCheckTimeOut(); + /* If wal_sender_shutdown_timeout is expired, exit the process + * Call this before WalSndComputeSleeptime() so the timeout is + * considered when computing sleep time. + */ + WalSndCheckShutdownTimeout(); + /* Send keepalive if the time has come */ WalSndKeepaliveIfNecessary(); @@ -1975,6 +1995,18 @@ WalSndWaitForWal(XLogRecPtr loc) /* die if timeout was reached */ WalSndCheckTimeOut(); + /* + * If a shutdown is in progress and wal_sender_shutdown_timeout has + * expired, terminate the walsender immediately rather than waiting + * indefinitely for the standby to catch up. This is needed because + * when synchronized_standby_slots is configured and the physical + * standby is not responding, the logical walsender can get stuck in + * this loop waiting for NeedToWaitForStandbys() to return false. + * Call this before WalSndComputeSleeptime() so the timeout is + * considered when computing sleep time. + */ + WalSndCheckShutdownTimeout(); + /* Send keepalive if the time has come */ WalSndKeepaliveIfNecessary(); @@ -2834,16 +2866,18 @@ ProcessStandbyPSRequestMessage(void) * If wal_sender_timeout is enabled we want to wake up in time to send * keepalives and to abort the connection if wal_sender_timeout has been * reached. + * + * If wal_sender_shutdown_timeout is enabled, during shutdown, we want to + * wake up in time to exit when it expires. */ static long WalSndComputeSleeptime(TimestampTz now) { + TimestampTz wakeup_time; long sleeptime = 10000; /* 10 s */ if (wal_sender_timeout > 0 && last_reply_timestamp > 0) { - TimestampTz wakeup_time; - /* * At the latest stop sleeping once wal_sender_timeout has been * reached. @@ -2864,6 +2898,20 @@ WalSndComputeSleeptime(TimestampTz now) sleeptime = TimestampDifferenceMilliseconds(now, wakeup_time); } + if (shutdown_request_timestamp != 0 && wal_sender_shutdown_timeout > 0) + { + long shutdown_sleeptime; + + wakeup_time = TimestampTzPlusMilliseconds(shutdown_request_timestamp, + wal_sender_shutdown_timeout); + + shutdown_sleeptime = TimestampDifferenceMilliseconds(now, wakeup_time); + + /* Choose the earliest wakeup. */ + if (shutdown_sleeptime < sleeptime) + sleeptime = shutdown_sleeptime; + } + return sleeptime; } @@ -2905,6 +2953,48 @@ WalSndCheckTimeOut(void) } } +/* + * Check whether the walsender process should terminate due to the expiration + * of wal_sender_shutdown_timeout after the receipt of a shutdown request. + */ +static void +WalSndCheckShutdownTimeout(void) +{ + TimestampTz now; + + /* Do nothing if shutdown has not been requested yet */ + if (!(got_STOPPING || got_SIGUSR2)) + return; + + /* Terminate immediately if wal_sender_shutdown_timeout is set to 0. */ + if (wal_sender_shutdown_timeout == 0) + WalSndDoneImmediate(); + + now = GetCurrentTimestamp(); + + /* + * Record the shutdown request timestamp even if + * wal_sender_shutdown_timeout is disabled (-1), since the setting may + * change during shutdown and the timestamp will be needed in that case. + */ + if (shutdown_request_timestamp == 0) + { + shutdown_request_timestamp = now; + return; + } + + /* Do not check timeout if wal_sender_shutdown_timeout is disabled. */ + if (wal_sender_shutdown_timeout == -1) + return; + + /* Terminate immediately if the timeout expires */ + if (TimestampDifferenceExceeds(shutdown_request_timestamp, now, + wal_sender_shutdown_timeout)) + { + WalSndDoneImmediate(); + } +} + /* Main loop of walsender process that streams the WAL over Copy messages. */ static void WalSndLoop(WalSndSendDataCallback send_data) @@ -2959,6 +3049,14 @@ WalSndLoop(WalSndSendDataCallback send_data) if (pq_flush_if_writable() != 0) WalSndShutdown(); + /* + * Check for wal_sender_shutdown_timeout. If timeout is expired, we do + * not wait for successful sending of all data to the receiver. Call + * this before WalSndComputeSleeptime() so the timeout is considered + * when computing sleep time. + */ + WalSndCheckShutdownTimeout(); + /* If nothing remains to be sent right now ... */ if (WalSndCaughtUp && !pq_is_send_pending()) { @@ -3607,6 +3705,49 @@ XLogSendLogical(void) } } +/* + * Forced shutdown of walsender if wal_sender_shutdown_timeout has expired. + */ +static void +WalSndDoneImmediate(void) +{ + WalSndState state = MyWalSnd->state; + + if (state == WALSNDSTATE_CATCHUP || + state == WALSNDSTATE_STREAMING || + state == WALSNDSTATE_STOPPING) + { + QueryCompletion qc; + + /* Try to inform receiver that XLOG streaming is done */ + SetQueryCompletion(&qc, CMDTAG_COPY, 0); + EndCommand(&qc, DestRemote, false); + + /* + * Note that the output buffer may be full during the forced shutdown of + * walsender. If pq_flush() is called at that time, the walsender process + * will be stuck. Therefore, call pq_flush_if_writable() instead. + * Successful reception of the done message with the walsender forced into + * a shutdown is not guaranteed. + */ + pq_flush_if_writable(); + } + + /* + * Prevent ereport from attempting to send any more messages to the + * standby. Otherwise, it can cause the process to get stuck if the output + * buffers are full. + */ + if (whereToSendOutput == DestRemote) + whereToSendOutput = DestNone; + + ereport(WARNING, + (errmsg("terminating walsender due to wal_sender_shutdown_timeout"), + errdetail("Walsender may have terminated before all WAL data was replicated to the receiver"))); + + proc_exit(0); +} + /* * Shutdown if the sender is caught up. * diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat index 0a862693fcd..e06847b081d 100644 --- a/src/backend/utils/misc/guc_parameters.dat +++ b/src/backend/utils/misc/guc_parameters.dat @@ -3503,6 +3503,16 @@ check_hook => 'check_wal_segment_size', }, +{ name => 'wal_sender_shutdown_timeout', type => 'int', context => 'PGC_USERSET', group => 'REPLICATION_SENDING', + short_desc => 'Sets the maximum time to wait for receiver to flush WAL data after shutdown request.', + long_desc => '-1 disables timeout; 0 means immediate termination of walsender', + flags => 'GUC_UNIT_MS', + variable => 'wal_sender_shutdown_timeout', + boot_val => '-1', + min => '-1', + max => 'INT_MAX', +}, + { name => 'wal_sender_timeout', type => 'int', context => 'PGC_USERSET', group => 'REPLICATION_SENDING', short_desc => 'Sets the maximum time to wait for WAL replication.', flags => 'GUC_UNIT_MS', diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index cf15597385b..c5b2e425a7f 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -352,6 +352,10 @@ #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_shutdown_timeout = -1 # max time to wait for receiver to flush data + # after receipt of shutdown request; in milliseconds + # -1 disables (means waiting for complete flush) + # 0 means immediate termination of walsender #track_commit_timestamp = off # collect timestamp of transaction commit # (change requires restart) diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h index a4df3b8e0ae..999876b7699 100644 --- a/src/include/replication/walsender.h +++ b/src/include/replication/walsender.h @@ -33,6 +33,7 @@ extern PGDLLIMPORT bool wake_wal_senders; /* user-settable parameters */ 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 void InitWalSender(void); diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build index f4a9cf5057f..e71e95c6297 100644 --- a/src/test/subscription/meson.build +++ b/src/test/subscription/meson.build @@ -47,6 +47,7 @@ tests += { 't/035_conflicts.pl', 't/036_sequences.pl', 't/037_except.pl', + 't/038_walsnd_shutdown_timeout.pl', 't/100_bugs.pl', ], }, diff --git a/src/test/subscription/t/038_walsnd_shutdown_timeout.pl b/src/test/subscription/t/038_walsnd_shutdown_timeout.pl new file mode 100644 index 00000000000..22d0087cd30 --- /dev/null +++ b/src/test/subscription/t/038_walsnd_shutdown_timeout.pl @@ -0,0 +1,237 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Checks that the publisher is able to shut down without +# waiting for sending of all pending data to the subscriber +# with wal_sender_shutdown_timeout set + +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +sub test_shutdown_with_empty_buffers +{ + my ($publisher, $subscriber, $bpgsql, $desc) = @_; + + # start transaction on subscriber to hold locks + $bpgsql->query_safe("BEGIN; INSERT INTO pub_test VALUES (0);"); + + # run concurrent transaction on publisher and commit + $publisher->safe_psql('postgres', + 'BEGIN; INSERT INTO pub_test VALUES (0); COMMIT;'); + + my $log_offset = -s $publisher->logfile; + + # test publisher shutdown + $publisher->stop('fast'); + pass($desc); + + ok( $publisher->log_contains( + qr/WARNING: .* terminating walsender due to wal_sender_shutdown_timeout/, + $log_offset), + "warning was emitted for: $desc"); + + return; +} + +sub test_shutdown_with_full_buffers +{ + my ($publisher, $subscriber, $bpgsql, $desc) = @_; + + # lock table to make apply_worker hang + $bpgsql->query_safe("BEGIN; LOCK TABLE pub_test IN EXCLUSIVE MODE;"); + + my $last_sent_lsn = $publisher->safe_psql('postgres', + "select sent_lsn from pg_stat_replication where application_name = 'sub_all';" + ); + my $cur_sent_lsn; + + # generate big amount of wal records for locked table + $publisher->safe_psql('postgres', + 'BEGIN; INSERT INTO pub_test SELECT i from generate_series(1, 20000) s(i); COMMIT;' + ); + + # wait for walsender to fill output buffers + my $max_attempts = $PostgreSQL::Test::Utils::timeout_default; + while ($max_attempts-- >= 0) + { + sleep 1; + + $cur_sent_lsn = $publisher->safe_psql('postgres', + "select sent_lsn from pg_stat_replication where application_name = 'sub_all';" + ); + + my $diff = $publisher->safe_psql( + 'postgres', qq( + SELECT pg_wal_lsn_diff('$cur_sent_lsn', '$last_sent_lsn'); + )); + + last if $diff == 0; + + $last_sent_lsn = $cur_sent_lsn; + } + + my $log_offset = -s $publisher->logfile; + + # test publisher shutdown + $publisher->stop('fast'); + pass($desc); + + ok( $publisher->log_contains( + qr/WARNING: .* terminating walsender due to wal_sender_shutdown_timeout/, + $log_offset), + "warning was emitted for: $desc"); + + return; +} + +sub test_shutdown_with_standby +{ + my ($publisher, $subscriber, $bpgsql, $desc) = @_; + + $publisher->backup('publisher_backup',backup_options => [ + '--create-slot', '--slot', 'standbyslot', '-d', 'dbname=postgres', '--write-recovery-conf']); + + $publisher->append_conf('postgresql.conf', q{ + synchronized_standby_slots = 'standbyslot' + }); + + $publisher->reload(); + + # create standby + my $standby = PostgreSQL::Test::Cluster->new('standby'); + $standby->init_from_backup($publisher, 'publisher_backup'); + $standby->append_conf('postgresql.conf', q{ + sync_replication_slots = on + hot_standby_feedback = on + }); + + $standby->start(); + + # lock table on subscriber to make apply_worker hang + $bpgsql->query_safe("BEGIN; LOCK TABLE pub_test IN EXCLUSIVE MODE;"); + + $publisher->safe_psql('postgres', + 'BEGIN; INSERT INTO pub_test VALUES (1); COMMIT;'); + + # send SIGSTOP to walreceiver on standby + my $receiverpid = $standby->safe_psql('postgres', + "SELECT pid FROM pg_stat_wal_receiver LIMIT 1"); + like($receiverpid, qr/^[0-9]+$/, "have walreceiver pid $receiverpid"); + kill 'STOP', int($receiverpid); + + $publisher->safe_psql('postgres', + 'BEGIN; INSERT INTO pub_test VALUES (2); COMMIT;'); + + my $log_offset = -s $publisher->logfile; + + # test publisher shutdown + $publisher->stop('fast'); + pass($desc); + + return; +} + +sub cleanup_after_test_case +{ + my ($publisher, $bpgsql) = @_; + + $bpgsql->query_safe("ABORT;"); + + $publisher->start(); + $publisher->wait_for_catchup('sub_all'); +} + +# ============================================================================= +# Setup publisher and subscriber + +# create publisher +my $publisher = PostgreSQL::Test::Cluster->new('publisher'); +$publisher->init(allows_streaming => 'logical'); +# set wal_sender_shutdown_timeout GUC parameter +$publisher->append_conf( + 'postgresql.conf', + "wal_sender_timeout = 1h + wal_sender_shutdown_timeout = 10ms"); +$publisher->start(); + +# create subscriber +my $subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$subscriber->init(); +$subscriber->append_conf('postgresql.conf', + "wal_receiver_status_interval = 1"); +$subscriber->start(); + +# create publication for test table +$publisher->safe_psql( + 'postgres', q{ + CREATE TABLE pub_test (id int PRIMARY KEY); + CREATE PUBLICATION pub_all FOR TABLE pub_test; +}); + +# create matching table on subscriber +$subscriber->safe_psql( + 'postgres', q{ + CREATE TABLE pub_test (id int PRIMARY KEY); +}); + +# form connection string to publisher +my $pub_connstr = $publisher->connstr; + +# create the subscription on subscriber +$subscriber->safe_psql( + 'postgres', qq{ + CREATE SUBSCRIPTION sub_all + CONNECTION '$pub_connstr' + PUBLICATION pub_all + WITH (failover = 'on'); +}); + +# wait for initial sync to finish +$subscriber->wait_for_subscription_sync($publisher, 'sub_all'); + +# create background psql session +my $bpgsql = $subscriber->background_psql('postgres', on_error_stop => 0); + +# ============================================================================= + +# ============================================================================= +# Testcase: Shutdown of publisher when output buffers are not full +# (wal_sender_shutdown_timeout = 10ms) + +test_shutdown_with_empty_buffers($publisher, $subscriber, $bpgsql, + 'successful shutdown of publisher when output buffers are not full (wal_sender_shutdown_timeout = 10ms)'); + +# ============================================================================= + +cleanup_after_test_case($publisher, $bpgsql); + +# ============================================================================= +# Testcase: Shutdown of publisher with full output buffers +# (wal_sender_shutdown_timeout = 10ms) + +test_shutdown_with_full_buffers($publisher, $subscriber, $bpgsql, + 'successful shutdown of publisher with full output buffers (wal_sender_shutdown_timeout = 10ms)'); + +# ============================================================================= + +cleanup_after_test_case($publisher, $bpgsql); + +$publisher->safe_psql('postgres', q{ + TRUNCATE pub_test; +}); + +$publisher->wait_for_catchup('sub_all'); + +# ============================================================================= +# Testcase: Shutdown of publisher with standby with slot-sync and subscriber +# (wal_sender_shutdown_timeout = 10ms) + +test_shutdown_with_standby($publisher, $subscriber, $bpgsql, + 'successful shutdown of publisher with slot-sync standby (wal_sender_shutdown_timeout = 10ms)'); + +# ============================================================================= + +done_testing(); -- 2.34.1