v26-0003-Allow-slot-sync-workers-to-wait-for-the-cascadin.patch

From d0b5ca20a08ec2d3d3ff741d20d49daa5cf3044c Mon Sep 17 00:00:00 2001
From: Shveta Malik <shveta.malik@gmail.com>
Date: Thu, 26 Oct 2023 10:28:47 +0530
Subject: [PATCH v26 3/3] Allow slot-sync workers to wait for the cascading
 standbys.

The GUC standby_slot_names is needed to be set on first standby
in order to allow it to wait for confirmation for cascading
standbys before updating logical 'synced' slots in slot-sync workers.
The intent is that the logical slots (synced ones) should not go ahead of
cascading standbys.

For the user created slots on first standby, we already have this wait
logic in place in logical walsender and in pg_logical_slot_get_changes_guts(),
but for synced slots (which can not be consumed yet), we need to make
sure that they are not going ahead of cascading standbys and that is
acheived by introducing the wait in slot-sync worker before we actually
update the slots.
---
 src/backend/replication/logical/slotsync.c | 110 +++++++++++++--
 src/backend/replication/walsender.c        | 155 +++++++++++----------
 src/include/replication/walsender.h        |   3 +
 3 files changed, 189 insertions(+), 79 deletions(-)

diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 4764fd4167..8206bfb2ce 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -76,6 +76,9 @@ typedef struct RemoteSlot
 #define WORKER_INACTIVITY_THRESHOLD_MS 1000L	/* 1 sec */
 
 static char *PrimaryConnInfoPreReload = NULL;
+static char *StandbySlotNamesPreReload = NULL;
+
+static bool ProcessSlotSyncInterrupts(WalReceiverConn *wrconn);
 
 /*
  * Wait for remote slot to pass locally reserved position.
@@ -460,6 +463,54 @@ drop_obsolete_slots(Oid *dbids, List *remote_slot_list)
 	}
 }
 
+/*
+ * Wait for cascading physical standbys corresponding to physical slots
+ * specified in standby_slot_names GUC to confirm receiving given lsn.
+ */
+static void
+slot_sync_wait_for_standby_confirmation(XLogRecPtr wait_for_lsn,
+										WalReceiverConn *wrconn)
+{
+	List	   *standby_slot_cpy;
+
+	/* Nothing to be done */
+	if (strcmp(standby_slot_names, "") == 0)
+		return;
+
+	standby_slot_cpy = list_copy(standby_slot_names_list);
+
+	for (;;)
+	{
+		bool		config_reloaded = false;
+
+		WaitForStandbyConfirmation(&standby_slot_cpy, wait_for_lsn);
+
+		/* Exit if done waiting for every slot. */
+		if (standby_slot_cpy == NIL)
+			break;
+
+		/* Process Interrupts if any */
+		config_reloaded = ProcessSlotSyncInterrupts(wrconn);
+
+		/*
+		 * Refresh the standby_slot_cpy if standby_slot_names_list got changed
+		 * after ConfigReload
+		 */
+		if (config_reloaded &&
+			strcmp(StandbySlotNamesPreReload, standby_slot_names) != 0)
+			standby_slot_cpy = list_copy(standby_slot_names_list);
+
+		/*
+		 * XXX: Is waiting for 5 second before retrying enough or more or
+		 * less?
+		 */
+		(void) WaitLatch(MyLatch,
+						 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+						 5000L,
+						 WAIT_EVENT_WAL_SENDER_WAIT_FOR_STANDBY_CONFIRMATION);
+	}
+}
+
 /*
  * Construct Slot Query
  *
@@ -652,6 +703,8 @@ synchronize_slots(dsa_area *dsa, WalReceiverConn *wrconn)
 	long		naptime = WORKER_DEFAULT_NAPTIME_MS;
 	Oid		   *dbids;
 	int			count = 0;
+	XLogRecPtr	max_confirmed_lsn = 0;
+	ListCell   *cell;
 
 	/* The primary_slot_name is not set yet or WALs not received yet */
 	if (!WalRcv ||
@@ -757,10 +810,11 @@ synchronize_slots(dsa_area *dsa, WalReceiverConn *wrconn)
 			count++;
 		}
 
-		/* Create list of remote slots to be used by drop_obsolete_slots */
+		/* Create list of remote slots */
 		remote_slot_list = lappend(remote_slot_list, remote_slot);
 
-		synchronize_one_slot(wrconn, remote_slot);
+		if (remote_slot->confirmed_lsn > max_confirmed_lsn)
+			max_confirmed_lsn = remote_slot->confirmed_lsn;
 
 		/*
 		 * Update naptime as required depending on slot activity. Check only
@@ -772,6 +826,25 @@ synchronize_slots(dsa_area *dsa, WalReceiverConn *wrconn)
 		ExecClearTuple(slot);
 	}
 
+	/*
+	 * If there are cascading standbys, wait for their confirmation before we
+	 * update synced logical slots locally.
+	 *
+	 * Instead of waiting on confirmation for lsn of each slot, let us wait
+	 * once for confirmation on max_confirmed_lsn. If that is confirmed by
+	 * each cascading standby, we are good to update all the slots.
+	 */
+	if (list_length(remote_slot_list))
+		slot_sync_wait_for_standby_confirmation(max_confirmed_lsn, wrconn);
+
+	/* Now sync the slots locally */
+	foreach(cell, remote_slot_list)
+	{
+		RemoteSlot *remote_slot = (RemoteSlot *) lfirst(cell);
+
+		synchronize_one_slot(wrconn, remote_slot);
+	}
+
 	/* Drop local slots that no longer need to be synced. */
 	drop_obsolete_slots(dbids, remote_slot_list);
 
@@ -822,6 +895,25 @@ reconnect_if_needed(WalReceiverConn *wrconn_prev)
 	return wrconn;
 }
 
+/*
+ * Save the current configurations related to slot-sync
+ *
+ * This function is invoked prior to each config-reload on receiving SIGHUP.
+ */
+static void
+save_current_configs()
+{
+	/* Free the previous allocations. */
+	if (PrimaryConnInfoPreReload)
+		pfree(PrimaryConnInfoPreReload);
+
+	if (StandbySlotNamesPreReload)
+		pfree(StandbySlotNamesPreReload);
+
+	PrimaryConnInfoPreReload = pstrdup(PrimaryConnInfo);
+	StandbySlotNamesPreReload = pstrdup(standby_slot_names);
+}
+
 /*
  * Interrupt handler for main loop of slot-sync worker.
  */
@@ -846,14 +938,14 @@ ProcessSlotSyncInterrupts(WalReceiverConn *wrconn)
 	{
 		ConfigReloadPending = false;
 
-		/* Free the previous allocation. */
-		if (PrimaryConnInfoPreReload)
-			pfree(PrimaryConnInfoPreReload);
-
-		/* Save the GUC primary_conninfo before reloading. */
-		PrimaryConnInfoPreReload = pstrdup(PrimaryConnInfo);
+		save_current_configs();
 
 		ProcessConfigFile(PGC_SIGHUP);
+
+		/* If standby_slot_names changed, recreate the standby_slot_names_list */
+		if (strcmp(StandbySlotNamesPreReload, standby_slot_names) != 0)
+			SlotSyncInitConfig();
+
 		reload_done = true;
 	}
 
@@ -921,6 +1013,8 @@ ReplSlotSyncWorkerMain(Datum main_arg)
 	/* Connect to primary node */
 	wrconn = remote_connect();
 
+	SlotSyncInitConfig();
+
 	/* Main wait loop. */
 	for (;;)
 	{
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 6dcda821ef..c9bbe9fa5e 100755
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1694,6 +1694,89 @@ WalSndSlotInList(char *slot_names, List *slot_names_list)
 	return inlist;
 }
 
+/*
+ * Helper function for WalSndWaitForStandbyConfirmation
+ *
+ * This is also used by slot-sync workers to wait for standbys'
+ * confirmation.
+ */
+void
+WaitForStandbyConfirmation(List **standby_slot_cpy, XLogRecPtr wait_for_lsn)
+{
+	ListCell   *l;
+
+	foreach(l, *standby_slot_cpy)
+	{
+		char	   *name = lfirst(l);
+		XLogRecPtr	restart_lsn = InvalidXLogRecPtr;
+		bool		invalidated = false;
+		char	   *warningfmt = NULL;
+		ReplicationSlot *slot;
+
+		slot = SearchNamedReplicationSlot(name, true);
+
+		if (slot && SlotIsPhysical(slot))
+		{
+			SpinLockAcquire(&slot->mutex);
+			restart_lsn = slot->data.restart_lsn;
+			invalidated = slot->data.invalidated != RS_INVAL_NONE;
+			SpinLockRelease(&slot->mutex);
+		}
+
+		/* Continue if the current slot hasn't caught up. */
+		if (!invalidated && !XLogRecPtrIsInvalid(restart_lsn) &&
+			restart_lsn < wait_for_lsn)
+		{
+			/* Log warning if no active_pid for this physical slot */
+			if (slot->active_pid == 0)
+			{
+				warningfmt = _("replication slot \"%s\" specified in parameter \"%s\" does not have active_pid");
+				ereport(WARNING, errmsg(warningfmt, name, "standby_slot_names"),
+						errdetail("Logical replication is waiting on the "
+								  "standby associated with \"%s\"", name),
+						errhint("Consider starting standby associated with "
+								"\"%s\" or amend standby_slot_names", name));
+			}
+
+			continue;
+		}
+
+		/*
+		 * It may happen that the slot specified in standby_slot_names GUC
+		 * value is dropped, so let's skip over it.
+		 */
+		else if (!slot)
+			warningfmt = _("replication slot \"%s\" specified in parameter \"%s\" does not exist, ignoring");
+
+		/*
+		 * If logical slot name is given in standby_slot_names, give WARNING
+		 * and skip it. Since it is harmless, so WARNING should be enough, no
+		 * need to error-out.
+		 */
+		else if (SlotIsLogical(slot))
+			warningfmt = _("cannot have logical replication slot \"%s\" in parameter \"%s\", ignoring");
+
+		/*
+		 * Specified physical slot may have been invalidated, so no point in
+		 * waiting for it.
+		 */
+		else if (XLogRecPtrIsInvalid(restart_lsn) || invalidated)
+			warningfmt = _("physical slot \"%s\" specified in parameter \"%s\" has been invalidated, ignoring");
+		else
+			Assert(restart_lsn >= wait_for_lsn);
+
+		/*
+		 * Reaching here indicates that either the slot has passed the
+		 * wait_for_lsn or there is an issue with the slot that requires a
+		 * warning to be reported.
+		 */
+		if (warningfmt)
+			ereport(WARNING, errmsg(warningfmt, name, "standby_slot_names"));
+
+		*standby_slot_cpy = foreach_delete_current(*standby_slot_cpy, l);
+	}
+}
+
 /*
  * Wait for physical standby to confirm receiving given lsn.
  *
@@ -1715,79 +1798,9 @@ WalSndWaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
 
 	for (;;)
 	{
-		ListCell   *l;
 		long		sleeptime = -1;
 
-		foreach(l, standby_slot_cpy)
-		{
-			char	   *name = lfirst(l);
-			XLogRecPtr	restart_lsn = InvalidXLogRecPtr;
-			bool		invalidated = false;
-			char	   *warningfmt = NULL;
-			ReplicationSlot *slot;
-
-			slot = SearchNamedReplicationSlot(name, true);
-
-			if (slot && SlotIsPhysical(slot))
-			{
-				SpinLockAcquire(&slot->mutex);
-				restart_lsn = slot->data.restart_lsn;
-				invalidated = slot->data.invalidated != RS_INVAL_NONE;
-				SpinLockRelease(&slot->mutex);
-			}
-
-			/* Continue if the current slot hasn't caught up. */
-			if (!invalidated && !XLogRecPtrIsInvalid(restart_lsn) &&
-				restart_lsn < wait_for_lsn)
-			{
-				/* Log warning if no active_pid for this physical slot */
-				if (slot->active_pid == 0)
-				{
-					warningfmt = _("replication slot \"%s\" specified in parameter \"%s\" does not have active_pid");
-					ereport(WARNING, errmsg(warningfmt, name, "standby_slot_names"),
-							errdetail("Logical replication is waiting on the "
-									  "standby associated with \"%s\"", name),
-							errhint("Consider starting standby associated with "
-									"\"%s\" or amend standby_slot_names", name));
-				}
-
-				continue;
-			}
-
-			/*
-			 * It may happen that the slot specified in standby_slot_names GUC
-			 * value is dropped, so let's skip over it.
-			 */
-			else if (!slot)
-				warningfmt = _("replication slot \"%s\" specified in parameter \"%s\" does not exist, ignoring");
-
-			/*
-			 * If logical slot name is given in standby_slot_names, give
-			 * WARNING and skip it. Since it is harmless, so WARNING should be
-			 * enough, no need to error-out.
-			 */
-			else if (SlotIsLogical(slot))
-				warningfmt = _("cannot have logical replication slot \"%s\" in parameter \"%s\", ignoring");
-
-			/*
-			 * Specified physical slot may have been invalidated, so no point
-			 * in waiting for it.
-			 */
-			else if (XLogRecPtrIsInvalid(restart_lsn) || invalidated)
-				warningfmt = _("physical slot \"%s\" specified in parameter \"%s\" has been invalidated, ignoring");
-			else
-				Assert(restart_lsn >= wait_for_lsn);
-
-			/*
-			 * Reaching here indicates that either the slot has passed the
-			 * wait_for_lsn or there is an issue with the slot that requires a
-			 * warning to be reported.
-			 */
-			if (warningfmt)
-				ereport(WARNING, errmsg(warningfmt, name, "standby_slot_names"));
-
-			standby_slot_cpy = foreach_delete_current(standby_slot_cpy, l);
-		}
+		WaitForStandbyConfirmation(&standby_slot_cpy, wait_for_lsn);
 
 		/* Exit if done waiting for every slot. */
 		if (standby_slot_cpy == NIL)
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index ecbd3526c5..320ab2f481 100644
--- a/src/include/replication/walsender.h
+++ b/src/include/replication/walsender.h
@@ -15,6 +15,7 @@
 #include <signal.h>
 
 #include "access/xlogdefs.h"
+#include "nodes/pg_list.h"
 
 /*
  * What to do with a snapshot in create replication slot command.
@@ -50,6 +51,8 @@ extern void WalSndWaitStopping(void);
 extern void HandleWalSndInitStopping(void);
 extern void WalSndRqstFileReload(void);
 extern void PhysicalConfirmReceivedLocation(XLogRecPtr lsn);
+extern void WaitForStandbyConfirmation(List **standby_slot_cpy,
+									   XLogRecPtr wait_for_lsn);
 extern void WalSndWaitForStandbyConfirmation(XLogRecPtr wait_for_lsn);
 
 /*
-- 
2.34.1