v45-0003-Allow-slot-sync-worker-to-wait-for-the-cascading.patch

From 82c236f4be380161b187b9ce8b6a6713be2b5b55 Mon Sep 17 00:00:00 2001
From: Shveta Malik <shveta.malik@gmail.com>
Date: Mon, 11 Dec 2023 12:28:46 +0530
Subject: [PATCH v45 3/3] Allow slot-sync worker 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 worker.
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 | 89 ++++++++++++++++++++--
 src/backend/replication/walsender.c        |  4 +-
 src/include/replication/walsender.h        |  5 ++
 3 files changed, 88 insertions(+), 10 deletions(-)

diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 0a8903ca92..48d4379714 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -112,7 +112,9 @@ static TimestampTz last_update_time;
  */
 #define WORKER_INACTIVITY_THRESHOLD_MS 10000L	/* 10 sec */
 
-static void ProcessSlotSyncInterrupts(WalReceiverConn *wrconn);
+static void ProcessSlotSyncInterrupts(WalReceiverConn *wrconn,
+									  List **standby_slots);
+
 
 /*
  * Wait for remote slot to pass locally reserved position.
@@ -171,7 +173,7 @@ wait_for_primary_slot_catchup(WalReceiverConn *wrconn, RemoteSlot *remote_slot,
 		CHECK_FOR_INTERRUPTS();
 
 		/* Handle any termination request if any */
-		ProcessSlotSyncInterrupts(wrconn);
+		ProcessSlotSyncInterrupts(wrconn, NULL /* standby_slots */ );
 
 		res = walrcv_exec(wrconn, cmd.data, WAIT_OUTPUT_COLUMN_COUNT, slotRow);
 
@@ -455,6 +457,52 @@ drop_obsolete_slots(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
+wait_for_standby_confirmation(XLogRecPtr wait_for_lsn,
+							  WalReceiverConn *wrconn)
+{
+	List	   *standby_slots;
+
+	/* Nothing to be done */
+	if (strcmp(standby_slot_names, "") == 0)
+		return;
+
+	standby_slots = GetStandbySlotList(true);
+
+	for (;;)
+	{
+		int			rc;
+
+		WalSndFilterStandbySlots(wait_for_lsn, &standby_slots);
+
+		/* Exit if done waiting for every slot. */
+		if (standby_slots == NIL)
+			break;
+
+		/*
+		 * This will reload configuration and will refresh the standby_slots
+		 * as well provided standby_slot_names GUC is changed by the user.
+		 */
+		ProcessSlotSyncInterrupts(wrconn, &standby_slots);
+
+		/*
+		 * XXX: Is waiting for 5 second before retrying enough or more or
+		 * less?
+		 */
+		rc = WaitLatch(MyLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+					   5000L,
+					   WAIT_EVENT_WAL_SENDER_WAIT_FOR_STANDBY_CONFIRMATION);
+
+		if (rc & WL_LATCH_SET)
+			ResetLatch(MyLatch);
+	}
+}
+
 /*
  * Synchronize single slot to given position.
  *
@@ -714,6 +762,7 @@ synchronize_slots(WalReceiverConn *wrconn)
 	MemoryContext oldctx = CurrentMemoryContext;
 	ListCell   *lc;
 	bool		slot_updated = false;
+	XLogRecPtr	max_confirmed_lsn = 0;
 
 	/*
 	 * The primary_slot_name is not set yet or WALs not received yet.
@@ -820,6 +869,9 @@ synchronize_slots(WalReceiverConn *wrconn)
 		/* Create list of remote slots */
 		remote_slot_list = lappend(remote_slot_list, remote_slot);
 
+		if (remote_slot->confirmed_lsn > max_confirmed_lsn)
+			max_confirmed_lsn = remote_slot->confirmed_lsn;
+
 		ExecClearTuple(tupslot);
 	}
 
@@ -830,6 +882,17 @@ synchronize_slots(WalReceiverConn *wrconn)
 	 */
 	drop_obsolete_slots(remote_slot_list);
 
+	/*
+	 * 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 (remote_slot_list)
+		wait_for_standby_confirmation(max_confirmed_lsn, wrconn);
+
 	/* Now sync the slots locally */
 	foreach(lc, remote_slot_list)
 	{
@@ -1001,7 +1064,7 @@ validate_slotsync_parameters(char **dbname)
  * worker will exit if the check fails.
  */
 static void
-slotsync_reread_config(WalReceiverConn *wrconn)
+slotsync_reread_config(WalReceiverConn *wrconn, List **standby_slots)
 {
 	char	   *old_primary_conninfo = pstrdup(PrimaryConnInfo);
 	char	   *old_primary_slot_name = pstrdup(PrimarySlotName);
@@ -1014,8 +1077,14 @@ slotsync_reread_config(WalReceiverConn *wrconn)
 	old_dbname = walrcv_get_dbname_from_conninfo(PrimaryConnInfo);
 	Assert(old_dbname);
 
-	ConfigReloadPending = false;
-	ProcessConfigFile(PGC_SIGHUP);
+	/*
+	 * Reload configs and recreate the standby_slot_names_list if GUC
+	 * standby_slot_names changed.
+	 */
+	if (standby_slots)
+		WalSndRereadConfigAndReInitSlotList(standby_slots);
+	else
+		ProcessConfigFile(PGC_SIGHUP);
 
 	conninfoChanged = strcmp(old_primary_conninfo, PrimaryConnInfo) != 0;
 	slotnameChanged = strcmp(old_primary_slot_name, PrimarySlotName) != 0;
@@ -1051,7 +1120,8 @@ slotsync_reread_config(WalReceiverConn *wrconn)
  * Interrupt handler for main loop of slot sync worker.
  */
 static void
-ProcessSlotSyncInterrupts(WalReceiverConn *wrconn)
+ProcessSlotSyncInterrupts(WalReceiverConn *wrconn,
+						  List **standby_slots)
 {
 	CHECK_FOR_INTERRUPTS();
 
@@ -1066,7 +1136,10 @@ ProcessSlotSyncInterrupts(WalReceiverConn *wrconn)
 	}
 
 	if (ConfigReloadPending)
-		slotsync_reread_config(wrconn);
+	{
+		ConfigReloadPending = false;
+		slotsync_reread_config(wrconn, standby_slots);
+	}
 }
 
 /*
@@ -1142,7 +1215,7 @@ ReplSlotSyncWorkerMain(Datum main_arg)
 		TimestampTz now;
 		bool		slot_updated;
 
-		ProcessSlotSyncInterrupts(wrconn);
+		ProcessSlotSyncInterrupts(wrconn, NULL /* standby_slots */ );
 
 		slot_updated = synchronize_slots(wrconn);
 
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 6b65c3c7dc..5ea47430a7 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1609,7 +1609,7 @@ PhysicalWakeupLogicalWalSnd(void)
  * Reload the config file and reinitialize the standby slot list if the GUC
  * standby_slot_names has changed.
  */
-static void
+void
 WalSndRereadConfigAndReInitSlotList(List **standby_slots)
 {
 	char	   *pre_standby_slot_names = pstrdup(standby_slot_names);
@@ -1634,7 +1634,7 @@ WalSndRereadConfigAndReInitSlotList(List **standby_slots)
  * it removes slots that have been invalidated, dropped, or converted to
  * logical slots.
  */
-static void
+void
 WalSndFilterStandbySlots(XLogRecPtr wait_for_lsn, List **standby_slots)
 {
 	ListCell   *lc;
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index ecd212c7b6..07b2c719ae 100644
--- a/src/include/replication/walsender.h
+++ b/src/include/replication/walsender.h
@@ -13,6 +13,7 @@
 #define _WALSENDER_H
 
 #include "access/xlogdefs.h"
+#include "nodes/pg_list.h"
 
 /*
  * What to do with a snapshot in create replication slot command.
@@ -48,7 +49,11 @@ extern void WalSndWaitStopping(void);
 extern void HandleWalSndInitStopping(void);
 extern void WalSndRqstFileReload(void);
 extern void PhysicalWakeupLogicalWalSnd(void);
+extern void WalSndFilterStandbySlots(XLogRecPtr wait_for_lsn,
+									 List **standby_slots);
 extern void WalSndWaitForStandbyConfirmation(XLogRecPtr wait_for_lsn);
+extern List *WalSndGetStandbySlots(void);
+extern void WalSndRereadConfigAndReInitSlotList(List **standby_slots);
 
 /*
  * Remember that we want to wakeup walsenders later
-- 
2.34.1