diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c index 7b6795c9b59..cf738d42c23 100644 --- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c +++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c @@ -466,6 +466,20 @@ libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli) } 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 */ + { + 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)))); + WalRcvIdentifySystemLsn = ((uint64) hi) << 32 | lo; + } + PQclear(res); return primary_sysid; diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index feff7094351..2da915aa3ce 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" @@ -95,6 +96,12 @@ bool hot_standby_feedback; static WalReceiverConn *wrconn = NULL; WalReceiverFunctionsType *WalReceiverFunctions = NULL; +/* + * Server's WAL flush position from the last IDENTIFY_SYSTEM call. + * Written by libpqwalreceiver, read by walreceiver main loop. + */ +XLogRecPtr WalRcvIdentifySystemLsn = InvalidXLogRecPtr; + /* * These variables are used similarly to openLogFile/SegNo, * but for walreceiver to write the XLOG. recvFileTLI is the TimeLineID @@ -194,6 +201,8 @@ WalReceiverMain(void) TimeLineID startpointTLI; TimeLineID primaryTLI; bool first_stream; + bool upstream_catchup_logged = false; + TimestampTz upstream_catchup_deadline = 0; WalRcvData *walrcv = WalRcv; TimestampTz now; char *err; @@ -339,8 +348,10 @@ WalReceiverMain(void) /* * Check that we're connected to a valid server using the - * IDENTIFY_SYSTEM replication command. + * IDENTIFY_SYSTEM replication command. Reset the global LSN + * first so we don't act on a stale value if the call fails. */ + WalRcvIdentifySystemLsn = InvalidXLogRecPtr; primary_sysid = walrcv_identify_system(wrconn, &primaryTLI); snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT, @@ -364,6 +375,72 @@ WalReceiverMain(void) 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 && + XLogRecPtrIsValid(WalRcvIdentifySystemLsn) && + startpoint > WalRcvIdentifySystemLsn && + startpoint - WalRcvIdentifySystemLsn <= 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(WalRcvIdentifySystemLsn))); + 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.c b/src/backend/utils/activity/wait_event.c index 05b71f9bc2a..8cd96ed74db 100644 --- a/src/backend/utils/activity/wait_event.c +++ b/src/backend/utils/activity/wait_event.c @@ -469,6 +469,9 @@ pgstat_get_wait_ipc(WaitEventIPC w) case WAIT_EVENT_XACT_GROUP_UPDATE: event_name = "XactGroupUpdate"; break; + case WAIT_EVENT_WAL_RECEIVER_UPSTREAM_CATCHUP: + event_name = "WalReceiverUpstreamCatchup"; + break; /* no default case, so that compiler will warn */ } diff --git a/src/backend/utils/adt/formatting.c b/src/backend/utils/adt/formatting.c index 7a043719a3d..a216a962196 100644 --- a/src/backend/utils/adt/formatting.c +++ b/src/backend/utils/adt/formatting.c @@ -2751,10 +2751,18 @@ DCH_to_char(FormatNode *node, bool is_interval, TmToChar *in, char *out, Oid col INVALID_FOR_INTERVAL; if (tmtcTzn(in)) { - /* We assume here that timezone names aren't localized */ + /* + * We assume here that timezone abbreviations aren't + * localized, so ASCII-only downcasing is sufficient. + */ char *p = asc_tolower_z(tmtcTzn(in)); - strcpy(s, p); + if (strlen(p) <= n->key->len * DCH_MAX_ITEM_SIZ) + strcpy(s, p); + else + ereport(ERROR, + (errcode(ERRCODE_DATETIME_VALUE_OUT_OF_RANGE), + errmsg("time zone format value too long"))); pfree(p); s += strlen(s); } @@ -2763,7 +2771,14 @@ DCH_to_char(FormatNode *node, bool is_interval, TmToChar *in, char *out, Oid col INVALID_FOR_INTERVAL; if (tmtcTzn(in)) { - strcpy(s, tmtcTzn(in)); + const char *p = tmtcTzn(in); + + if (strlen(p) <= n->key->len * DCH_MAX_ITEM_SIZ) + strcpy(s, p); + else + ereport(ERROR, + (errcode(ERRCODE_DATETIME_VALUE_OUT_OF_RANGE), + errmsg("time zone format value too long"))); s += strlen(s); } break; diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h index 281626fa6f5..62067468f41 100644 --- a/src/include/replication/walreceiver.h +++ b/src/include/replication/walreceiver.h @@ -164,6 +164,13 @@ typedef struct extern PGDLLIMPORT WalRcvData *WalRcv; +/* + * Server's WAL flush position as reported by the last IDENTIFY_SYSTEM call. + * Set by walrcv_identify_system(), used by the walreceiver to avoid + * requesting streaming from a point ahead of the upstream's flush position. + */ +extern PGDLLIMPORT XLogRecPtr WalRcvIdentifySystemLsn; + typedef struct { bool logical; /* True if this is logical replication stream, diff --git a/src/include/utils/wait_event.h b/src/include/utils/wait_event.h index 2adc5df2e79..13a69410128 100644 --- a/src/include/utils/wait_event.h +++ b/src/include/utils/wait_event.h @@ -131,7 +131,8 @@ typedef enum WAIT_EVENT_SYNC_REP, WAIT_EVENT_WAL_RECEIVER_EXIT, WAIT_EVENT_WAL_RECEIVER_WAIT_START, - WAIT_EVENT_XACT_GROUP_UPDATE + WAIT_EVENT_XACT_GROUP_UPDATE, + WAIT_EVENT_WAL_RECEIVER_UPSTREAM_CATCHUP } WaitEventIPC; /* ---------- diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index 5453e1aa6fa..b5f462f5653 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -48,6 +48,7 @@ tests += { 't/043_no_contrecord_switch.pl', 't/045_archive_restartpoint.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..b6fe05b063d --- /dev/null +++ b/src/test/recovery/t/055_cascade_reconnect.pl @@ -0,0 +1,154 @@ +# 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. +ok( $stubble->poll_query_until( + 'postgres', + qq[SELECT count(*) > 0 FROM pg_stat_activity + WHERE backend_type = 'walreceiver' AND wait_event = 'WalReceiverUpstreamCatchup' + ]), + "walreceiver has caught up" +) or die qq(Timed out while waiting for walreceiver to catch up); + +# 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();