diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 86d31c46599..029990d9fce 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -63,7 +63,8 @@ static char *libpqrcv_get_conninfo(WalReceiverConn *conn); static void libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host, int *sender_port); static char *libpqrcv_identify_system(WalReceiverConn *conn, - TimeLineID *primary_tli); + TimeLineID *primary_tli, + XLogRecPtr *server_lsn); static char *libpqrcv_get_dbname_from_conninfo(const char *connInfo); static char *libpqrcv_get_option_from_conninfo(const char *connInfo, const char *keyword); @@ -421,7 +422,8 @@ libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host, * timeline ID of the primary. */ static char * -libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli) +libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli, + XLogRecPtr *server_lsn) { PGresult *res; char *primary_sysid; @@ -452,6 +454,21 @@ libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli) PQntuples(res), PQnfields(res), 1, 3))); primary_sysid = pstrdup(PQgetvalue(res, 0, 0)); *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1)); + + /* Column 2 is the server's current WAL flush position */ + if (server_lsn) + { + uint32 hi, + lo; + + if (sscanf(PQgetvalue(res, 0, 2), "%X/%X", &hi, &lo) != 2) + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("could not parse WAL location \"%s\"", + PQgetvalue(res, 0, 2)))); + *server_lsn = ((uint64) hi) << 32 | lo; + } + PQclear(res); return primary_sysid; diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 0bd19074010..1ca19c1a7a8 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -5732,7 +5732,7 @@ run_apply_worker(void) * We don't really use the output identify_system for anything but it does * some initializations on the upstream so let's still call it. */ - (void) walrcv_identify_system(LogRepWorkerWalRcvConn, &startpointTLI); + (void) walrcv_identify_system(LogRepWorkerWalRcvConn, &startpointTLI, NULL); set_apply_error_context_origin(originname); diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index 429a1b2d96d..61dc6a5588b 100644 --- a/src/backend/replication/walreceiver.c +++ b/src/backend/replication/walreceiver.c @@ -54,6 +54,7 @@ #include "access/htup_details.h" #include "access/timeline.h" #include "access/transam.h" +#include "access/xlog.h" #include "access/xlog_internal.h" #include "access/xlogarchive.h" #include "access/xlogrecovery.h" @@ -161,6 +162,8 @@ WalReceiverMain(const void *startup_data, size_t startup_data_len) TimeLineID startpointTLI; TimeLineID primaryTLI; bool first_stream; + bool upstream_catchup_logged = false; + TimestampTz upstream_catchup_deadline = 0; WalRcvData *walrcv; TimestampTz now; char *err; @@ -318,13 +321,15 @@ WalReceiverMain(const void *startup_data, size_t startup_data_len) { char *primary_sysid; char standby_sysid[32]; + XLogRecPtr primaryFlushPtr; WalRcvStreamOptions options; /* * Check that we're connected to a valid server using the * IDENTIFY_SYSTEM replication command. */ - primary_sysid = walrcv_identify_system(wrconn, &primaryTLI); + primary_sysid = walrcv_identify_system(wrconn, &primaryTLI, + &primaryFlushPtr); snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT, GetSystemIdentifier()); @@ -348,6 +353,71 @@ WalReceiverMain(const void *startup_data, size_t startup_data_len) errmsg("highest timeline %u of the primary is behind recovery timeline %u", primaryTLI, startpointTLI))); + /* + * If our requested startpoint is ahead of the upstream server's + * current WAL flush position, we cannot start streaming yet. (We say + * "upstream" here and not "primary" because this condition can only + * happen on a cascading standby.) This can happen when such a + * cascading standby has advanced past the upstream via archive + * recovery but the intermediate standby has not caught up with that + * yet. In this case, wait for the upstream to catch up before + * attempting START_REPLICATION, because that would fail with + * "requested starting point is ahead of the WAL flush position". + * + * We only perform this check when we're on the same timeline as the + * primary; when timelines differ, let START_REPLICATION handle the + * timeline negotiation. + * + * We also only wait if the gap is within one WAL segment. This is + * the expected case because archive recovery processes whole segment + * files: the cascade's next read position lands at the start of the + * following segment while the upstream's flush position is still + * inside the just-replayed one, producing at most a sub-segment gap. + * A larger gap means the upstream is genuinely behind, so we let + * START_REPLICATION fail normally and allow the startup process to + * fall back to other WAL sources. + * + * Honor wal_receiver_timeout so the walreceiver doesn't wait + * indefinitely: if the upstream hasn't caught up within the timeout, + * exit and let the startup process retry normally. + */ + if (startpointTLI == primaryTLI && + startpoint > primaryFlushPtr && + startpoint - primaryFlushPtr <= wal_segment_size) + { + /* Set deadline on first iteration */ + if (!upstream_catchup_logged && wal_receiver_timeout > 0) + upstream_catchup_deadline = + TimestampTzPlusMilliseconds(GetCurrentTimestamp(), + wal_receiver_timeout); + + ereport(upstream_catchup_logged ? DEBUG1 : LOG, + errmsg("walreceiver requested start point %X/%08X on timeline %u is ahead of the upstream server's flush position %X/%08X, waiting", + LSN_FORMAT_ARGS(startpoint), startpointTLI, + LSN_FORMAT_ARGS(primaryFlushPtr))); + upstream_catchup_logged = true; + + (void) WaitLatch(MyLatch, + WL_EXIT_ON_PM_DEATH | WL_TIMEOUT | WL_LATCH_SET, + wal_retrieve_retry_interval, + WAIT_EVENT_WAL_RECEIVER_UPSTREAM_CATCHUP); + ResetLatch(MyLatch); + + if (upstream_catchup_deadline > 0 && + GetCurrentTimestamp() >= upstream_catchup_deadline) + ereport(ERROR, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("terminating walreceiver due to timeout while waiting for upstream to catch up"))); + + CHECK_FOR_INTERRUPTS(); + continue; + } + else + { + upstream_catchup_logged = false; + upstream_catchup_deadline = 0; + } + /* * Get any missing history files. We do this always, even when we're * not interested in that timeline, so that if we're promoted to diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index 1016502d042..256b3a3c02e 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -163,6 +163,7 @@ WAIT_FOR_WAL_FLUSH "Waiting for WAL flush to reach a target LSN on a primary or WAIT_FOR_WAL_REPLAY "Waiting for WAL replay to reach a target LSN on a standby." WAIT_FOR_WAL_WRITE "Waiting for WAL write to reach a target LSN on a standby." WAL_RECEIVER_EXIT "Waiting for the WAL receiver to exit." +WAL_RECEIVER_UPSTREAM_CATCHUP "Waiting for upstream server WAL flush position to catch up to requested start point." WAL_RECEIVER_WAIT_START "Waiting for startup process to send initial data for streaming replication." WAL_SUMMARY_READY "Waiting for a new WAL summary to be generated." XACT_GROUP_UPDATE "Waiting for the group leader to update transaction status at transaction end." diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index 760364e3587..5cfe90f9989 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -280,9 +280,12 @@ typedef void (*walrcv_get_senderinfo_fn) (WalReceiverConn *conn, * Run IDENTIFY_SYSTEM on the cluster connected to and validate the * identity of the cluster. Returns the system ID of the cluster * connected to. 'primary_tli' is the timeline ID of the sender. + * If 'server_lsn' is not NULL, it is set to the current WAL flush + * position of the sender. */ typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn, - TimeLineID *primary_tli); + TimeLineID *primary_tli, + XLogRecPtr *server_lsn); /* * walrcv_get_dbname_from_conninfo_fn @@ -441,8 +444,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions; WalReceiverFunctions->walrcv_get_conninfo(conn) #define walrcv_get_senderinfo(conn, sender_host, sender_port) \ WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port) -#define walrcv_identify_system(conn, primary_tli) \ - WalReceiverFunctions->walrcv_identify_system(conn, primary_tli) +#define walrcv_identify_system(conn, primary_tli, server_lsn) \ + WalReceiverFunctions->walrcv_identify_system(conn, primary_tli, server_lsn) #define walrcv_get_dbname_from_conninfo(conninfo) \ WalReceiverFunctions->walrcv_get_dbname_from_conninfo(conninfo) #define walrcv_server_version(conn) \ diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index ad0d85f4189..39ec8c4946d 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -63,6 +63,7 @@ tests += { 't/052_checkpoint_segment_missing.pl', 't/053_standby_login_event_trigger.pl', 't/054_unlogged_sequence_promotion.pl', + 't/055_cascade_reconnect.pl', ], }, } diff --git a/src/test/recovery/t/055_cascade_reconnect.pl b/src/test/recovery/t/055_cascade_reconnect.pl new file mode 100644 index 00000000000..051a3292533 --- /dev/null +++ b/src/test/recovery/t/055_cascade_reconnect.pl @@ -0,0 +1,148 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Test that a cascading standby can reconnect to its upstream standby after +# advancing past the upstream's WAL flush position via archive recovery. +# +# Setup: praline -> samurai -> stubble +# stubble has both streaming (from samurai) and restore_command +# (from praline's archive). +# +# When samurai's walreceiver is stopped and stubble falls back to +# archive recovery, stubble may advance its recovery position past +# samurai's replay position. Previously, stubble's walreceiver +# would fail with "requested starting point is ahead of the WAL flush +# position" when reconnecting to samurai. + +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# Initialize praline with archiving +my $praline = PostgreSQL::Test::Cluster->new('praline'); +$praline->init(allows_streaming => 1, has_archiving => 1); +$praline->append_conf( + 'postgresql.conf', qq( +wal_keep_size = 128MB +checkpoint_timeout = 1h +)); +$praline->start; + +# Take backup and create samurai (streaming from praline, no archive) +my $backup_name = 'my_backup'; +$praline->backup($backup_name); + +my $samurai = PostgreSQL::Test::Cluster->new('samurai'); +$samurai->init_from_backup($praline, $backup_name, has_streaming => 1); +$samurai->start; + +# Wait for samurai to start streaming +$praline->wait_for_catchup($samurai); + +# Take backup from samurai and create stubble +# stubble streams from samurai AND restores from praline's archive +$samurai->backup($backup_name); + +my $stubble = PostgreSQL::Test::Cluster->new('stubble'); +$stubble->init_from_backup($samurai, $backup_name, has_streaming => 1); +$stubble->enable_restoring($praline); +$stubble->start; + +# Generate initial data and wait for full cascade replication +$praline->safe_psql('postgres', + "CREATE TABLE test_tab AS SELECT generate_series(1, 1000) AS id"); +$praline->wait_for_replay_catchup($samurai); +$samurai->wait_for_replay_catchup($stubble, $praline); + +my $result = $stubble->safe_psql('postgres', "SELECT count(*) FROM test_tab"); +is($result, '1000', 'initial data replicated to stubble'); + +# Disconnect samurai from praline by clearing primary_conninfo. +# This stops samurai's walreceiver, so samurai can no longer receive +# new WAL. Its GetStandbyFlushRecPtr() will return only replayPtr. +$samurai->append_conf('postgresql.conf', "primary_conninfo = ''"); +$samurai->reload; + +# Wait for samurai's walreceiver to stop +$samurai->poll_query_until('postgres', + "SELECT NOT EXISTS (SELECT 1 FROM pg_stat_wal_receiver)") + or die "Timed out waiting for samurai walreceiver to stop"; + +# Stop stubble cleanly. We'll restart it after generating new WAL +# so it enters the recovery state machine fresh and tries archive first. +$stubble->stop; + +# Force a checkpoint now so that no background checkpoint can generate +# extra WAL during the INSERT below and push it across a segment boundary. +# Combined with checkpoint_timeout = 1h this ensures the new WAL fits +# within a single segment, keeping the gap within wal_segment_size. +$praline->safe_psql('postgres', "CHECKPOINT"); + +# Generate more WAL on praline +$praline->safe_psql('postgres', + "INSERT INTO test_tab SELECT generate_series(1001, 2000)"); + +# Force WAL switch and wait for archiving to complete, so that +# stubble can find the new WAL in the archive when it starts. +my $walfile = $praline->safe_psql('postgres', + "SELECT pg_walfile_name(pg_current_wal_lsn())"); +$praline->safe_psql('postgres', "SELECT pg_switch_wal()"); +$praline->poll_query_until('postgres', + "SELECT '$walfile' <= last_archived_wal FROM pg_stat_archiver") + or die "Timed out waiting for WAL archiving"; + +# Rotate stubble's log so we can check just the new log output +$stubble->rotate_logfile; +my $stubble_log_offset = -s $stubble->logfile; + +# Start stubble. It will: +# 1. Read new WAL from praline's archive (XLOG_FROM_ARCHIVE) +# 2. Advance RecPtr past samurai's replay position +# 3. Try streaming from samurai (XLOG_FROM_STREAM) +# 4. detect that upstream is behind via +# IDENTIFY_SYSTEM and wait instead of failing +$stubble->start; + +# Wait for stubble to replay the new data from archive +$stubble->poll_query_until('postgres', + "SELECT count(*) >= 2000 FROM test_tab") + or die "Timed out waiting for stubble to replay archived WAL"; + +$result = $stubble->safe_psql('postgres', "SELECT count(*) FROM test_tab"); +is($result, '2000', 'stubble replayed new data from archive'); + +# Wait for walreceiver to hit the upstream-catchup wait event, proving we +# exercised the START_REPLICATION-ahead-of-upstream path. +$stubble->wait_for_event('walreceiver', 'WalReceiverUpstreamCatchup'); + +# Verify no errors occurred in stubble. +my $stubble_loglines = + PostgreSQL::Test::Utils::slurp_file($stubble->logfile, $stubble_log_offset); +ok( $stubble_loglines !~ m/ERROR/, 'no errors in stubble log'); + +# Now restore samurai's streaming from praline so it can catch up +$samurai->enable_streaming($praline); +$samurai->reload; + +# Wait for samurai to catch up with praline +$praline->wait_for_replay_catchup($samurai); + +# stubble's walreceiver should eventually connect to samurai and +# resume streaming (once samurai has caught up past stubble's position) +$samurai->poll_query_until('postgres', + "SELECT EXISTS (SELECT 1 FROM pg_stat_replication)") + or die "Timed out waiting for stubble to reconnect to samurai"; + +# Verify end-to-end cascade streaming works with new data +$praline->safe_psql('postgres', + "INSERT INTO test_tab SELECT generate_series(2001, 3000)"); +$praline->wait_for_replay_catchup($samurai); +$samurai->wait_for_replay_catchup($stubble, $praline); + +$result = $stubble->safe_psql('postgres', "SELECT count(*) FROM test_tab"); +is($result, '3000', + 'cascade streaming resumes normally after upstream catches up'); + +done_testing();