diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index f837703..bf4e9d4 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4291,6 +4291,24 @@ ANY num_sync (
+ wal_receiver_start_condition (enum)
+
+ wal_receiver_start_condition configuration parameter
+
+
+
+
+ By default ("catchup" condition) the standby will connect to the primary after replaying all
+ WAL files available in the archive.
+ But in case of large recovery_min_apply_delay it may lead to master's WAL space overflow
+ because WAL receiver is not launched and so nobody is fetching changes from master.
+ Changine start condition to "consistency" will force starting WAL receiver once
+ consistency point is reached.
+
+
+
+
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 3b766e6..02e3fc2 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -802,6 +802,7 @@ static XLogSource readSource = 0; /* XLOG_FROM_* code */
*/
static XLogSource currentSource = 0; /* XLOG_FROM_* code */
static bool lastSourceFailed = false;
+static bool stopOnError = false;
typedef struct XLogPageReadPrivate
{
@@ -3992,6 +3993,49 @@ RemoveOldXlogFiles(XLogSegNo segno, XLogRecPtr RedoRecPtr, XLogRecPtr endptr)
}
/*
+ * Find latest WAL LSN
+ */
+static XLogRecPtr
+GetLastLSN(XLogRecPtr lsn)
+{
+ XLogReaderState *xlogreader;
+ char *errormsg;
+ XLogPageReadPrivate private;
+ MemSet(&private, 0, sizeof(XLogPageReadPrivate));
+
+ xlogreader = XLogReaderAllocate(wal_segment_size, NULL, &XLogPageRead, &private);
+
+ stopOnError = true;
+ while (XLogReadRecord(xlogreader, lsn, &errormsg) != NULL)
+ {
+ lsn = InvalidXLogRecPtr;
+ }
+ stopOnError = false;
+ lsn = xlogreader->EndRecPtr;
+ XLogReaderFree(xlogreader);
+
+ return lsn;
+}
+
+/*
+ * Launch WalReceiver starting from last LSN if not started yet.
+ */
+static void
+StartWalRcv(XLogRecPtr currLsn)
+{
+ if (!WalRcvStreaming() && PrimaryConnInfo && strcmp(PrimaryConnInfo, "") != 0)
+ {
+ XLogRecPtr lastLSN = GetLastLSN(currLsn);
+ if (lastLSN != InvalidXLogRecPtr)
+ {
+ curFileTLI = ThisTimeLineID;
+ RequestXLogStreaming(ThisTimeLineID, lastLSN, PrimaryConnInfo,
+ PrimarySlotName);
+ }
+ }
+}
+
+/*
* Remove WAL files that are not part of the given timeline's history.
*
* This is called during recovery, whenever we switch to follow a new
@@ -6013,6 +6057,13 @@ recoveryApplyDelay(XLogReaderState *record)
if (secs <= 0 && microsecs <= 0)
return false;
+ /*
+ * Start WAL receiver if not started yet, to avoid WALs overflow at primary node
+ * or large gap between primary and replica when apply delay is specified.
+ */
+ if (wal_receiver_start_condition == WAL_RCV_START_AT_CONSISTENCY)
+ StartWalRcv(record->EndRecPtr);
+
while (true)
{
ResetLatch(&XLogCtl->recoveryWakeupLatch);
@@ -11821,6 +11872,13 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
return false;
/*
+ * If WAL receiver was altery started because of apply delay,
+ * then restart it.
+ */
+ if (WalRcvStreaming())
+ ShutdownWalRcv();
+
+ /*
* If primary_conninfo is set, launch walreceiver to try
* to stream the missing WAL.
*
@@ -11996,6 +12054,9 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
if (readFile >= 0)
return true; /* success! */
+ if (stopOnError)
+ return false;
+
/*
* Nope, not found in archive or pg_wal.
*/
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index f54ae76..87e0bda 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -75,6 +75,7 @@
int wal_receiver_status_interval;
int wal_receiver_timeout;
bool hot_standby_feedback;
+int wal_receiver_start_condition;
/* libpqwalreceiver connection */
static WalReceiverConn *wrconn = NULL;
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 4b3769b..ce2a735 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -459,6 +459,14 @@ const struct config_enum_entry ssl_protocol_versions_info[] = {
{NULL, 0, false}
};
+const struct config_enum_entry wal_rcv_start_options[] = {
+ {"catchup", WAL_RCV_START_AT_CATCHUP, true},
+ {"consistency", WAL_RCV_START_AT_CONSISTENCY, true},
+ {NULL, 0, false}
+};
+
+
+
static struct config_enum_entry shared_memory_options[] = {
#ifndef WIN32
{"sysv", SHMEM_TYPE_SYSV, false},
@@ -4577,6 +4585,17 @@ static struct config_enum ConfigureNamesEnum[] =
NULL, NULL, NULL
},
+ {
+ {"wal_receiver_start_condition", PGC_POSTMASTER, REPLICATION_STANDBY,
+ gettext_noop("When to start WAL receiver."),
+ NULL,
+ },
+ &wal_receiver_start_condition,
+ WAL_RCV_START_AT_CATCHUP,
+ wal_rcv_start_options,
+ NULL, NULL, NULL
+ },
+
/* End-of-list marker */
{
{NULL, 0, 0, NULL, NULL}, NULL, 0, NULL, NULL, NULL, NULL
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index be02a76..c88fd4a 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -330,6 +330,7 @@
#wal_retrieve_retry_interval = 5s # time to wait before retrying to
# retrieve WAL after a failed attempt
#recovery_min_apply_delay = 0 # minimum delay for applying changes during recovery
+#wal_receiver_start_condition = 'catchup' # wal receiver start condition
# - Subscribers -
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index e12a934..d8bec2a 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -22,11 +22,17 @@
#include "pgtime.h"
#include "utils/tuplestore.h"
+typedef enum
+{
+ WAL_RCV_START_AT_CATCHUP, /* start a WAL receiver after replaying all WAL files */
+ WAL_RCV_START_AT_CONSISTENCY /* start a WAL receiver once consistency has been reached */
+} WalRcvStartCondition;
+
/* user-settable parameters */
extern int wal_receiver_status_interval;
extern int wal_receiver_timeout;
extern bool hot_standby_feedback;
-
+extern int wal_receiver_start_condition;
/*
* MAXCONNINFO: maximum size of a connection string.
*