Allow logical failover slots to wait on synchronous replication

From 805e5b7e35913f53b9caac996aaf0b62e937f274 Mon Sep 17 00:00:00 2001
From: John Hsu <johnhyvr@gmail.com>
Date: Tue, 4 Jun 2024 20:27:47 +0000
Subject: [PATCH] Wait on synchronous replication by default for logical
 failover slots

If synchronous replication is enabled, this patch
allows logical subscribers to wait for changes to
be replicated to synchronous replicas before
consuming the changes.

In the event that 'synchronized_standby_slots' is set, it
will take precedence.
---
 src/backend/replication/slot.c                | 271 ++++++++++--------
 src/backend/replication/syncrep.c             |  27 +-
 src/include/replication/syncrep.h             |   3 +
 .../t/040_standby_failover_slots_sync.pl      | 135 +++++++++
 4 files changed, 321 insertions(+), 115 deletions(-)

diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index baf9b89dc4..fafb6c7873 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -49,6 +49,7 @@
 #include "postmaster/interrupt.h"
 #include "replication/slotsync.h"
 #include "replication/slot.h"
+#include "replication/syncrep.h"
 #include "replication/walsender_private.h"
 #include "storage/fd.h"
 #include "storage/ipc.h"
@@ -2570,8 +2571,8 @@ SlotExistsInSyncStandbySlots(const char *slot_name)
 }
 
 /*
- * Return true if the slots specified in synchronized_standby_slots have caught up to
- * the given WAL location, false otherwise.
+ * Return true if the slots specified in synchronized_standby_slots or synchronous
+ * replication have caught up to the given WAL location, false otherwise.
  *
  * The elevel parameter specifies the error level used for logging messages
  * related to slots that do not exist, are invalidated, or are inactive.
@@ -2585,9 +2586,9 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 
 	/*
 	 * Don't need to wait for the standbys to catch up if there is no value in
-	 * synchronized_standby_slots.
+	 * synchronized_standby_slots or synchronous replication is not configured.
 	 */
-	if (synchronized_standby_slots_config == NULL)
+	if (synchronized_standby_slots_config == NULL && !SyncRepConfigured())
 		return true;
 
 	/*
@@ -2598,144 +2599,190 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 		return true;
 
 	/*
-	 * Don't need to wait for the standbys to catch up if they are already
-	 * beyond the specified WAL location.
-	 */
-	if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) &&
-		ss_oldest_flush_lsn >= wait_for_lsn)
-		return true;
-
-	/*
-	 * To prevent concurrent slot dropping and creation while filtering the
-	 * slots, take the ReplicationSlotControlLock outside of the loop.
+	 * In the event that synchronized_standby_slots and synchronous replication is
+	 * configured, have the former take precedence.
 	 */
-	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
-
-	name = synchronized_standby_slots_config->slot_names;
-	for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
+	if (synchronized_standby_slots_config != NULL)
 	{
-		XLogRecPtr	restart_lsn;
-		bool		invalidated;
-		bool		inactive;
-		ReplicationSlot *slot;
+		/*
+		* Don't need to wait for the standbys to catch up if they are already
+		* beyond the specified WAL location.
+		*/
+		if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) &&
+			ss_oldest_flush_lsn >= wait_for_lsn)
+			return true;
 
-		slot = SearchNamedReplicationSlot(name, false);
+		/*
+		* To prevent concurrent slot dropping and creation while filtering the
+		* slots, take the ReplicationSlotControlLock outside of the loop.
+		*/
+		LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
-		if (!slot)
+		name = synchronized_standby_slots_config->slot_names;
+		for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
 		{
-			/*
-			 * If a slot name provided in synchronized_standby_slots does not
-			 * exist, report a message and exit the loop. A user can specify a
-			 * slot name that does not exist just before the server startup.
-			 * The GUC check_hook(validate_sync_standby_slots) cannot validate
-			 * such a slot during startup as the ReplicationSlotCtl shared
-			 * memory is not initialized at that time. It is also possible for
-			 * a user to drop the slot in synchronized_standby_slots
-			 * afterwards.
-			 */
-			ereport(elevel,
-					errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					errmsg("replication slot \"%s\" specified in parameter %s does not exist",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting on the standby associated with \"%s\".",
-							  name),
-					errhint("Consider creating the slot \"%s\" or amend parameter %s.",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			XLogRecPtr	restart_lsn;
+			bool		invalidated;
+			bool		inactive;
+			ReplicationSlot *slot;
 
-		if (SlotIsLogical(slot))
-		{
-			/*
-			 * If a logical slot name is provided in
-			 * synchronized_standby_slots, report a message and exit the loop.
-			 * Similar to the non-existent case, a user can specify a logical
-			 * slot name in synchronized_standby_slots before the server
-			 * startup, or drop an existing physical slot and recreate a
-			 * logical slot with the same name.
-			 */
-			ereport(elevel,
-					errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					errmsg("cannot have logical replication slot \"%s\" in parameter %s",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting for correction on \"%s\".",
-							  name),
-					errhint("Consider removing logical slot \"%s\" from parameter %s.",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			slot = SearchNamedReplicationSlot(name, false);
 
-		SpinLockAcquire(&slot->mutex);
-		restart_lsn = slot->data.restart_lsn;
-		invalidated = slot->data.invalidated != RS_INVAL_NONE;
-		inactive = slot->active_pid == 0;
-		SpinLockRelease(&slot->mutex);
+			if (!slot)
+			{
+				/*
+				* If a slot name provided in synchronized_standby_slots does not
+				* exist, report a message and exit the loop. A user can specify a
+				* slot name that does not exist just before the server startup.
+				* The GUC check_hook(validate_sync_standby_slots) cannot validate
+				* such a slot during startup as the ReplicationSlotCtl shared
+				* memory is not initialized at that time. It is also possible for
+				* a user to drop the slot in synchronized_standby_slots
+				* afterwards.
+				*/
+				ereport(elevel,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("replication slot \"%s\" specified in parameter %s does not exist",
+							name, "synchronized_standby_slots"),
+						errdetail("Logical replication is waiting on the standby associated with \"%s\".",
+								name),
+						errhint("Consider creating the slot \"%s\" or amend parameter %s.",
+								name, "synchronized_standby_slots"));
+				break;
+			}
 
-		if (invalidated)
-		{
-			/* Specified physical slot has been invalidated */
-			ereport(elevel,
-					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-					errmsg("physical slot \"%s\" specified in parameter %s has been invalidated",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting on the standby associated with \"%s\".",
-							  name),
-					errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			if (SlotIsLogical(slot))
+			{
+				/*
+				* If a logical slot name is provided in
+				* synchronized_standby_slots, report a message and exit the loop.
+				* Similar to the non-existent case, a user can specify a logical
+				* slot name in synchronized_standby_slots before the server
+				* startup, or drop an existing physical slot and recreate a
+				* logical slot with the same name.
+				*/
+				ereport(elevel,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("cannot have logical replication slot \"%s\" in parameter %s",
+							name, "synchronized_standby_slots"),
+						errdetail("Logical replication is waiting for correction on \"%s\".",
+								name),
+						errhint("Consider removing logical slot \"%s\" from parameter %s.",
+								name, "synchronized_standby_slots"));
+				break;
+			}
 
-		if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
-		{
-			/* Log a message if no active_pid for this physical slot */
-			if (inactive)
+			SpinLockAcquire(&slot->mutex);
+			restart_lsn = slot->data.restart_lsn;
+			invalidated = slot->data.invalidated != RS_INVAL_NONE;
+			inactive = slot->active_pid == 0;
+			SpinLockRelease(&slot->mutex);
+
+			if (invalidated)
+			{
+				/* Specified physical slot has been invalidated */
 				ereport(elevel,
 						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-						errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid",
-							   name, "synchronized_standby_slots"),
+						errmsg("physical slot \"%s\" specified in parameter %s has been invalidated",
+							name, "synchronized_standby_slots"),
 						errdetail("Logical replication is waiting on the standby associated with \"%s\".",
-								  name),
-						errhint("Consider starting standby associated with \"%s\" or amend parameter %s.",
+								name),
+						errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.",
 								name, "synchronized_standby_slots"));
+				break;
+			}
 
-			/* Continue if the current slot hasn't caught up. */
-			break;
+			if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
+			{
+				/* Log a message if no active_pid for this physical slot */
+				if (inactive)
+					ereport(elevel,
+							errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+							errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid",
+								name, "synchronized_standby_slots"),
+							errdetail("Logical replication is waiting on the standby associated with \"%s\".",
+									name),
+							errhint("Consider starting standby associated with \"%s\" or amend parameter %s.",
+									name, "synchronized_standby_slots"));
+
+				/* Continue if the current slot hasn't caught up. */
+				break;
+			}
+
+			Assert(restart_lsn >= wait_for_lsn);
+
+			if (XLogRecPtrIsInvalid(min_restart_lsn) ||
+				min_restart_lsn > restart_lsn)
+				min_restart_lsn = restart_lsn;
+
+			caught_up_slot_num++;
+
+			name += strlen(name) + 1;
 		}
 
-		Assert(restart_lsn >= wait_for_lsn);
+		LWLockRelease(ReplicationSlotControlLock);
+
+		/*
+		* Return false if not all the standbys have caught up to the specified
+		* WAL location.
+		*/
+		if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
+			return false;
 
-		if (XLogRecPtrIsInvalid(min_restart_lsn) ||
-			min_restart_lsn > restart_lsn)
-			min_restart_lsn = restart_lsn;
+		/* The ss_oldest_flush_lsn must not retreat. */
+		Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) ||
+			min_restart_lsn >= ss_oldest_flush_lsn);
 
-		caught_up_slot_num++;
+		ss_oldest_flush_lsn = min_restart_lsn;
 
-		name += strlen(name) + 1;
+		return true;
 	}
+	else
+	{
+		volatile WalSndCtlData *walsndctl = WalSndCtl;
+		static XLogRecPtr	lsn[NUM_SYNC_REP_WAIT_MODE];
+		static bool initialized = false;
+		/* Initialize value in case SIGHUP changing to SYNC_REP_NO_WAIT */
+		int mode = SyncRepWaitMode;
+		int i;
+
+		if (!initialized)
+		{
+			for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
+			{
+				lsn[i] = InvalidXLogRecPtr;
+			}
+		}
 
-	LWLockRelease(ReplicationSlotControlLock);
+		if (mode == SYNC_REP_NO_WAIT)
+			return true;
 
-	/*
-	 * Return false if not all the standbys have caught up to the specified
-	 * WAL location.
-	 */
-	if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
-		return false;
+		if (lsn[mode] >= wait_for_lsn)
+			return true;
+
+		LWLockAcquire(SyncRepLock, LW_SHARED);
 
-	/* The ss_oldest_flush_lsn must not retreat. */
-	Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) ||
-		   min_restart_lsn >= ss_oldest_flush_lsn);
+		/* Cache values to reduce contention on lock */
+		for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
+		{
+			lsn[i] = walsndctl->lsn[i];
+		}
 
-	ss_oldest_flush_lsn = min_restart_lsn;
+		LWLockRelease(SyncRepLock);
 
-	return true;
+		if (lsn[mode] >= wait_for_lsn)
+			return true;
+
+		return false;
+	}
 }
 
 /*
  * Wait for physical standbys to confirm receiving the given lsn.
  *
  * Used by logical decoding SQL functions. It waits for physical standbys
- * corresponding to the physical slots specified in the synchronized_standby_slots GUC.
+ * corresponding to the physical slots specified in the synchronized_standby_slots GUC,
+ * or synchronous replication.
  */
 void
 WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
@@ -2745,7 +2792,7 @@ WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
 	 * slot is not a logical failover slot, or there is no value in
 	 * synchronized_standby_slots.
 	 */
-	if (!MyReplicationSlot->data.failover || !synchronized_standby_slots_config)
+	if (!MyReplicationSlot->data.failover || !(synchronized_standby_slots_config || SyncRepConfigured()))
 		return;
 
 	ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv);
diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c
index fa5988c824..69c061d8dc 100644
--- a/src/backend/replication/syncrep.c
+++ b/src/backend/replication/syncrep.c
@@ -78,6 +78,7 @@
 #include "common/int.h"
 #include "miscadmin.h"
 #include "pgstat.h"
+#include "replication/slot.h"
 #include "replication/syncrep.h"
 #include "replication/walsender.h"
 #include "replication/walsender_private.h"
@@ -95,7 +96,7 @@ char	   *SyncRepStandbyNames;
 static bool announce_next_takeover = true;
 
 SyncRepConfigData *SyncRepConfig = NULL;
-static int	SyncRepWaitMode = SYNC_REP_NO_WAIT;
+int	SyncRepWaitMode = SYNC_REP_NO_WAIT;
 
 static void SyncRepQueueInsert(int mode);
 static void SyncRepCancelWait(void);
@@ -124,6 +125,8 @@ static int	cmp_lsn(const void *a, const void *b);
 static bool SyncRepQueueIsOrderedByLSN(int mode);
 #endif
 
+bool SyncRepConfigured(void);
+
 /*
  * ===========================================================
  * Synchronous Replication functions for normal user backends
@@ -169,8 +172,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit)
 	 * described in SyncRepUpdateSyncStandbysDefined(). On the other hand, if
 	 * it's false, the lock is not necessary because we don't touch the queue.
 	 */
-	if (!SyncRepRequested() ||
-		!((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
+	if (!SyncRepConfigured())
 		return;
 
 	/* Cap the level for anything other than commit to remote flush only. */
@@ -523,6 +525,15 @@ SyncRepReleaseWaiters(void)
 
 	LWLockRelease(SyncRepLock);
 
+	/*
+	 * If synchronized_standby_slots is set, the respective walsender's
+	 * will be responsible for broadcasting the value.
+	 */
+	if (strcmp(synchronized_standby_slots, "") == 0)
+	{
+		ConditionVariableBroadcast(&WalSndCtl->wal_confirm_rcv_cv);
+	}
+
 	elog(DEBUG3, "released %d procs up to write %X/%X, %d procs up to flush %X/%X, %d procs up to apply %X/%X",
 		 numwrite, LSN_FORMAT_ARGS(writePtr),
 		 numflush, LSN_FORMAT_ARGS(flushPtr),
@@ -1069,3 +1080,13 @@ assign_synchronous_commit(int newval, void *extra)
 			break;
 	}
 }
+
+bool
+SyncRepConfigured()
+{
+	if (!SyncRepRequested() ||
+		!((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
+		return false;
+
+	return true;
+}
diff --git a/src/include/replication/syncrep.h b/src/include/replication/syncrep.h
index ea439e6da6..ec22bc72df 100644
--- a/src/include/replication/syncrep.h
+++ b/src/include/replication/syncrep.h
@@ -82,6 +82,7 @@ extern PGDLLIMPORT char *SyncRepStandbyNames;
 
 /* called by user backend */
 extern void SyncRepWaitForLSN(XLogRecPtr lsn, bool commit);
+extern bool SyncRepConfigured();
 
 /* called at backend exit */
 extern void SyncRepCleanupAtProcExit(void);
@@ -96,6 +97,8 @@ extern int	SyncRepGetCandidateStandbys(SyncRepStandbyData **standbys);
 /* called by checkpointer */
 extern void SyncRepUpdateSyncStandbysDefined(void);
 
+extern int	SyncRepWaitMode;
+
 /*
  * Internal functions for parsing synchronous_standby_names grammar,
  * in syncrep_gram.y and syncrep_scanner.l
diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl
index 2c51cfc3c8..00130d92dc 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -710,6 +710,141 @@ $result = $subscriber1->safe_psql('postgres',
 is($result, 't',
 	"subscriber1 gets data from primary after standby1 acknowledges changes");
 
+##################################################
+# Test that logical failover replication slots wait for the specified
+# synchronous replicas to receive the changes first. It uses the
+# following set up:
+#
+# 				(synchronous physical standbys)
+#				| ----> standby1 (application_name = standby1)
+#				| ----> standby2 (application_name = standby2)
+# primary -----	|
+#				(logical replication)
+#				| ----> subscriber1 (failover = true, slot_name = lsub1_slot)
+#				| ----> subscriber2 (failover = false, slot_name = lsub2_slot)
+#
+# synchronous_commit = 'on'
+# synchronous_standby_names = 'ANY 2 (standby1, standby2)'
+#
+# The setup is configured in such a way that the logical slot of subscriber1 is
+# enabled for failover, and thus the subscriber1 will wait for the changes to have
+# been synchronously replicated before receiving the decoded changes.
+##################################################
+
+$primary->safe_psql('postgres', "TRUNCATE tab_int;");
+# Setup synchronous replication
+$primary->append_conf(
+	'postgresql.conf', qq(
+		synchronous_commit = 'on'
+		synchronous_standby_names = 'ANY 2 (standby1, standby2)'
+));
+
+$primary->reload;
+
+$standby1->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby1'");
+$standby1->reload;
+
+$standby2->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby2'");
+$standby2->reload;
+
+# Check that synchronous replication is setup properly
+$standby2->stop;
+
+# Create some data on the primary
+$primary_row_count = 10;
+
+my $sync_back_q = $primary->background_psql(
+	'postgres',
+	on_error_stop => 0,
+	timeout => $PostgreSQL::Test::Utils::timeout_default);
+
+$sync_back_q->query_until(qr/insert_blocked_on_sync_rep/, q(
+    \echo insert_blocked_on_sync_rep
+   INSERT INTO tab_int SELECT generate_series(1, 10);
+));
+
+$result = $primary->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+
+is($result, 'f', "primary row count is not updated due to synchronous replication");
+
+# Verify the standby specified in synchronized_standby_slots (sb1_slot aka standby1)
+# catches up with the primary.
+$primary->wait_for_replay_catchup($standby1);
+$result = $standby1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't', "standby1 gets data from primary");
+
+# Validate that synchronized_standby_slots takes precedence over waiting for
+# changes to have been synchronous replicated.
+# Since the slot specified (sb1_slot) has received the changes, primary can send
+# the decoded changes to the subscription enabled for failover
+# (i.e. regress_mysub1).
+$primary->wait_for_catchup('regress_mysub1');
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber1 gets data from primary after standby1 acknowledges changes");
+
+$standby2->start;
+$primary->wait_for_replay_catchup($standby2);
+
+# Unset synchronized_standby_slots to test synchronous replication
+# blocks primary from sending logical decoded changes to failover slots until
+# changes have been synchronously replicated.
+$primary->append_conf(
+	'postgresql.conf', qq(
+	synchronized_standby_slots = ''));
+$primary->reload;
+
+$primary_row_count = 20;
+$standby2->stop;
+
+$sync_back_q->query_until(
+	qr/insert_blocked_on_additional_sync_rep/, q(
+	\echo insert_blocked_on_additional_sync_rep
+	INSERT INTO tab_int SELECT generate_series(11, 20);
+));
+
+# Since $standby2 has not received the changes, validate that subscriber1 (failover = true)
+# has not received the decoded changes, but subscriber2 (failover = false) has.
+$primary->wait_for_catchup('regress_mysub2');
+
+$result = $subscriber2->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber2 gets data from primary even if the changes have not been synchronously acknowledged.");
+
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 'f',
+	"subscriber1 does not get data from primary since changes have not been synchronously acknowledged.");
+
+# Start standby2 to allow the changes to be acknowledged by all the synchronous standbys.
+$standby2->start;
+$primary->wait_for_replay_catchup($standby2);
+$primary->wait_for_catchup('regress_mysub1');
+
+# Now that the changes have been replicated to all synchronous nodes,
+# primary can send the decoded changes to the subscription enabled for failover
+# (i.e. regress_mysub1). While the standby was down, regress_mysub1 didn't
+# receive any data from the primary. i.e. the primary didn't allow it to go
+# ahead of requirements of synchronous commit.
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber1 gets data from primary since changes have been syhcronously acknowledged.");
+# No longer need to run the background session.
+$sync_back_q->quit;
+
+# Reset synchronized_standby_slots and synchronous commit for below test cases
+$primary->append_conf(
+	'postgresql.conf', qq(
+synchronized_standby_slots = 'sb1_slot'
+synchronous_standby_names = ''
+));
+$primary->reload;
+
 ##################################################
 # Verify that when using pg_logical_slot_get_changes to consume changes from a
 # logical failover slot, it will also wait for the slots specified in
-- 
2.40.1

From 627a468a1402fde3632c8d409a32a4fceaf8cfcc Mon Sep 17 00:00:00 2001
From: John Hsu <johnhyvr@gmail.com>
Date: Mon, 26 Aug 2024 18:23:20 +0000
Subject: [PATCH] Wait on synchronous replication by default for logical
 failover slots

If synchronous replication is enabled, this patch
allows logical subscribers with failover_slots enabled
to wait for changes to be replicated to synchronous replicas
before consuming the changes.
---
 src/backend/replication/slot.c                | 289 +++++++++++-------
 src/backend/replication/syncrep.c             |  27 +-
 .../utils/activity/wait_event_names.txt       |   1 +
 src/include/replication/syncrep.h             |   3 +
 src/include/replication/walsender_private.h   |   6 +
 src/include/storage/lwlocklist.h              |   1 +
 .../t/040_standby_failover_slots_sync.pl      | 135 ++++++++
 7 files changed, 346 insertions(+), 116 deletions(-)

diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index c290339af5..22b0772eee 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -49,6 +49,7 @@
 #include "postmaster/interrupt.h"
 #include "replication/slotsync.h"
 #include "replication/slot.h"
+#include "replication/syncrep.h"
 #include "replication/walsender_private.h"
 #include "storage/fd.h"
 #include "storage/ipc.h"
@@ -2591,8 +2592,8 @@ SlotExistsInSyncStandbySlots(const char *slot_name)
 }
 
 /*
- * Return true if the slots specified in synchronized_standby_slots have caught up to
- * the given WAL location, false otherwise.
+ * Return true if the slots specified in synchronized_standby_slots or synchronous
+ * replication have caught up to the given WAL location, false otherwise.
  *
  * The elevel parameter specifies the error level used for logging messages
  * related to slots that do not exist, are invalidated, or are inactive.
@@ -2606,9 +2607,9 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 
 	/*
 	 * Don't need to wait for the standbys to catch up if there is no value in
-	 * synchronized_standby_slots.
+	 * synchronized_standby_slots or synchronous replication is not configured.
 	 */
-	if (synchronized_standby_slots_config == NULL)
+	if (synchronized_standby_slots_config == NULL && !SyncRepConfigured())
 		return true;
 
 	/*
@@ -2619,144 +2620,206 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 		return true;
 
 	/*
-	 * Don't need to wait for the standbys to catch up if they are already
-	 * beyond the specified WAL location.
+	 * In the event that synchronized_standby_slots and synchronous replication is
+	 * configured, have the former take precedence.
 	 */
-	if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) &&
-		ss_oldest_flush_lsn >= wait_for_lsn)
-		return true;
-
-	/*
-	 * To prevent concurrent slot dropping and creation while filtering the
-	 * slots, take the ReplicationSlotControlLock outside of the loop.
-	 */
-	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
-
-	name = synchronized_standby_slots_config->slot_names;
-	for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
+	if (synchronized_standby_slots_config != NULL)
 	{
-		XLogRecPtr	restart_lsn;
-		bool		invalidated;
-		bool		inactive;
-		ReplicationSlot *slot;
-
-		slot = SearchNamedReplicationSlot(name, false);
+		/*
+		 * Don't need to wait for the standbys to catch up if they are already
+		 * beyond the specified WAL location.
+		 */
+		if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) &&
+			ss_oldest_flush_lsn >= wait_for_lsn)
+			return true;
 
-		if (!slot)
+		/*
+		 * To prevent concurrent slot dropping and creation while filtering the
+		 * slots, take the ReplicationSlotControlLock outside of the loop.
+		 */
+		LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+		name = synchronized_standby_slots_config->slot_names;
+		for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
 		{
-			/*
-			 * If a slot name provided in synchronized_standby_slots does not
-			 * exist, report a message and exit the loop. A user can specify a
-			 * slot name that does not exist just before the server startup.
-			 * The GUC check_hook(validate_sync_standby_slots) cannot validate
-			 * such a slot during startup as the ReplicationSlotCtl shared
-			 * memory is not initialized at that time. It is also possible for
-			 * a user to drop the slot in synchronized_standby_slots
-			 * afterwards.
-			 */
-			ereport(elevel,
-					errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					errmsg("replication slot \"%s\" specified in parameter %s does not exist",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting on the standby associated with \"%s\".",
-							  name),
-					errhint("Consider creating the slot \"%s\" or amend parameter %s.",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			XLogRecPtr	restart_lsn;
+			bool		invalidated;
+			bool		inactive;
+			ReplicationSlot *slot;
 
-		if (SlotIsLogical(slot))
-		{
-			/*
-			 * If a logical slot name is provided in
-			 * synchronized_standby_slots, report a message and exit the loop.
-			 * Similar to the non-existent case, a user can specify a logical
-			 * slot name in synchronized_standby_slots before the server
-			 * startup, or drop an existing physical slot and recreate a
-			 * logical slot with the same name.
-			 */
-			ereport(elevel,
-					errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					errmsg("cannot have logical replication slot \"%s\" in parameter %s",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting for correction on \"%s\".",
-							  name),
-					errhint("Consider removing logical slot \"%s\" from parameter %s.",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			slot = SearchNamedReplicationSlot(name, false);
 
-		SpinLockAcquire(&slot->mutex);
-		restart_lsn = slot->data.restart_lsn;
-		invalidated = slot->data.invalidated != RS_INVAL_NONE;
-		inactive = slot->active_pid == 0;
-		SpinLockRelease(&slot->mutex);
+			if (!slot)
+			{
+				/*
+				 * If a slot name provided in synchronized_standby_slots does not
+				 * exist, report a message and exit the loop. A user can specify a
+				 * slot name that does not exist just before the server startup.
+				 * The GUC check_hook(validate_sync_standby_slots) cannot validate
+				 * such a slot during startup as the ReplicationSlotCtl shared
+				 * memory is not initialized at that time. It is also possible for
+				 * a user to drop the slot in synchronized_standby_slots
+				 * afterwards.
+				 */
+				ereport(elevel,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("replication slot \"%s\" specified in parameter %s does not exist",
+							name, "synchronized_standby_slots"),
+						errdetail("Logical replication is waiting on the standby associated with \"%s\".",
+								name),
+						errhint("Consider creating the slot \"%s\" or amend parameter %s.",
+								name, "synchronized_standby_slots"));
+				break;
+			}
 
-		if (invalidated)
-		{
-			/* Specified physical slot has been invalidated */
-			ereport(elevel,
-					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-					errmsg("physical slot \"%s\" specified in parameter %s has been invalidated",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting on the standby associated with \"%s\".",
-							  name),
-					errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			if (SlotIsLogical(slot))
+			{
+				/*
+				 * If a logical slot name is provided in
+				 * synchronized_standby_slots, report a message and exit the loop.
+				 * Similar to the non-existent case, a user can specify a logical
+				 * slot name in synchronized_standby_slots before the server
+				 * startup, or drop an existing physical slot and recreate a
+				 * logical slot with the same name.
+				 */
+				ereport(elevel,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("cannot have logical replication slot \"%s\" in parameter %s",
+							name, "synchronized_standby_slots"),
+						errdetail("Logical replication is waiting for correction on \"%s\".",
+								name),
+						errhint("Consider removing logical slot \"%s\" from parameter %s.",
+								name, "synchronized_standby_slots"));
+				break;
+			}
 
-		if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
-		{
-			/* Log a message if no active_pid for this physical slot */
-			if (inactive)
+			SpinLockAcquire(&slot->mutex);
+			restart_lsn = slot->data.restart_lsn;
+			invalidated = slot->data.invalidated != RS_INVAL_NONE;
+			inactive = slot->active_pid == 0;
+			SpinLockRelease(&slot->mutex);
+
+			if (invalidated)
+			{
+				/* Specified physical slot has been invalidated */
 				ereport(elevel,
 						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-						errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid",
-							   name, "synchronized_standby_slots"),
+						errmsg("physical slot \"%s\" specified in parameter %s has been invalidated",
+							name, "synchronized_standby_slots"),
 						errdetail("Logical replication is waiting on the standby associated with \"%s\".",
-								  name),
-						errhint("Consider starting standby associated with \"%s\" or amend parameter %s.",
+								name),
+						errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.",
 								name, "synchronized_standby_slots"));
+				break;
+			}
 
-			/* Continue if the current slot hasn't caught up. */
-			break;
+			if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
+			{
+				/* Log a message if no active_pid for this physical slot */
+				if (inactive)
+					ereport(elevel,
+							errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+							errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid",
+								name, "synchronized_standby_slots"),
+							errdetail("Logical replication is waiting on the standby associated with \"%s\".",
+									name),
+							errhint("Consider starting standby associated with \"%s\" or amend parameter %s.",
+									name, "synchronized_standby_slots"));
+
+				/* Continue if the current slot hasn't caught up. */
+				break;
+			}
+
+			Assert(restart_lsn >= wait_for_lsn);
+
+			if (XLogRecPtrIsInvalid(min_restart_lsn) ||
+				min_restart_lsn > restart_lsn)
+				min_restart_lsn = restart_lsn;
+
+			caught_up_slot_num++;
+
+			name += strlen(name) + 1;
 		}
 
-		Assert(restart_lsn >= wait_for_lsn);
+		LWLockRelease(ReplicationSlotControlLock);
+
+		/*
+		 * Return false if not all the standbys have caught up to the specified
+		 * WAL location.
+		 */
+		if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
+			return false;
 
-		if (XLogRecPtrIsInvalid(min_restart_lsn) ||
-			min_restart_lsn > restart_lsn)
-			min_restart_lsn = restart_lsn;
+		/* The ss_oldest_flush_lsn must not retreat. */
+		Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) ||
+			min_restart_lsn >= ss_oldest_flush_lsn);
 
-		caught_up_slot_num++;
+		ss_oldest_flush_lsn = min_restart_lsn;
 
-		name += strlen(name) + 1;
+		return true;
 	}
+	else
+	{
+		volatile WalSndCtlData *walsndctl = WalSndCtl;
+		static XLogRecPtr	lsn[NUM_SYNC_REP_WAIT_MODE]; /* Cache LSNs */
+		static bool initialized = false;
+		/* Initialize value in case SIGHUP changing to SYNC_REP_NO_WAIT */
+		int mode = SyncRepWaitMode;
+		int i;
+
+		if (!initialized)
+		{
+			for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
+			{
+				lsn[i] = InvalidXLogRecPtr;
+			}
+		}
 
-	LWLockRelease(ReplicationSlotControlLock);
+		if (mode == SYNC_REP_NO_WAIT)
+			return true;
 
-	/*
-	 * Return false if not all the standbys have caught up to the specified
-	 * WAL location.
-	 */
-	if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
-		return false;
+		if (lsn[mode] >= wait_for_lsn)
+			return true;
+
+		/* Check shared cache first to see if values updated */
+		LWLockAcquire(SyncRepLogicalCacheLock, LW_SHARED);
+		if (walsndctl->cached_lsn[mode] >= wait_for_lsn)
+		{
+			memcpy(lsn, (XLogRecPtr *) walsndctl->cached_lsn, sizeof(lsn));
+			LWLockRelease(SyncRepLogicalCacheLock);
+			return true;
+		}
 
-	/* The ss_oldest_flush_lsn must not retreat. */
-	Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) ||
-		   min_restart_lsn >= ss_oldest_flush_lsn);
+		LWLockRelease(SyncRepLogicalCacheLock);
 
-	ss_oldest_flush_lsn = min_restart_lsn;
+		LWLockAcquire(SyncRepLogicalCacheLock, LW_EXCLUSIVE);
+		/* Another walsender could have updated shared cache prior to obtaining lock */
+		if (walsndctl->cached_lsn[mode] >= wait_for_lsn)
+		{
+			memcpy(lsn, (XLogRecPtr *) walsndctl->cached_lsn, sizeof(lsn));
+			LWLockRelease(SyncRepLogicalCacheLock);
+			return true;
+		}
 
-	return true;
+		/* Cache values to reduce contention */
+		LWLockAcquire(SyncRepLock, LW_SHARED);
+		memcpy((XLogRecPtr *) walsndctl->cached_lsn, (XLogRecPtr *) walsndctl->lsn, sizeof(lsn));
+		LWLockRelease(SyncRepLock);
+		memcpy(lsn, (XLogRecPtr *) walsndctl->lsn, sizeof(lsn));
+		LWLockRelease(SyncRepLogicalCacheLock);
+
+		if (lsn[mode] >= wait_for_lsn)
+			return true;
+
+		return false;
+	}
 }
 
 /*
  * Wait for physical standbys to confirm receiving the given lsn.
  *
  * Used by logical decoding SQL functions. It waits for physical standbys
- * corresponding to the physical slots specified in the synchronized_standby_slots GUC.
+ * corresponding to the physical slots specified in the synchronized_standby_slots GUC,
+ * or synchronous replication.
  */
 void
 WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
@@ -2766,7 +2829,7 @@ WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
 	 * slot is not a logical failover slot, or there is no value in
 	 * synchronized_standby_slots.
 	 */
-	if (!MyReplicationSlot->data.failover || !synchronized_standby_slots_config)
+	if (!MyReplicationSlot->data.failover || !(synchronized_standby_slots_config || SyncRepConfigured()))
 		return;
 
 	ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv);
diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c
index fa5988c824..69c061d8dc 100644
--- a/src/backend/replication/syncrep.c
+++ b/src/backend/replication/syncrep.c
@@ -78,6 +78,7 @@
 #include "common/int.h"
 #include "miscadmin.h"
 #include "pgstat.h"
+#include "replication/slot.h"
 #include "replication/syncrep.h"
 #include "replication/walsender.h"
 #include "replication/walsender_private.h"
@@ -95,7 +96,7 @@ char	   *SyncRepStandbyNames;
 static bool announce_next_takeover = true;
 
 SyncRepConfigData *SyncRepConfig = NULL;
-static int	SyncRepWaitMode = SYNC_REP_NO_WAIT;
+int	SyncRepWaitMode = SYNC_REP_NO_WAIT;
 
 static void SyncRepQueueInsert(int mode);
 static void SyncRepCancelWait(void);
@@ -124,6 +125,8 @@ static int	cmp_lsn(const void *a, const void *b);
 static bool SyncRepQueueIsOrderedByLSN(int mode);
 #endif
 
+bool SyncRepConfigured(void);
+
 /*
  * ===========================================================
  * Synchronous Replication functions for normal user backends
@@ -169,8 +172,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit)
 	 * described in SyncRepUpdateSyncStandbysDefined(). On the other hand, if
 	 * it's false, the lock is not necessary because we don't touch the queue.
 	 */
-	if (!SyncRepRequested() ||
-		!((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
+	if (!SyncRepConfigured())
 		return;
 
 	/* Cap the level for anything other than commit to remote flush only. */
@@ -523,6 +525,15 @@ SyncRepReleaseWaiters(void)
 
 	LWLockRelease(SyncRepLock);
 
+	/*
+	 * If synchronized_standby_slots is set, the respective walsender's
+	 * will be responsible for broadcasting the value.
+	 */
+	if (strcmp(synchronized_standby_slots, "") == 0)
+	{
+		ConditionVariableBroadcast(&WalSndCtl->wal_confirm_rcv_cv);
+	}
+
 	elog(DEBUG3, "released %d procs up to write %X/%X, %d procs up to flush %X/%X, %d procs up to apply %X/%X",
 		 numwrite, LSN_FORMAT_ARGS(writePtr),
 		 numflush, LSN_FORMAT_ARGS(flushPtr),
@@ -1069,3 +1080,13 @@ assign_synchronous_commit(int newval, void *extra)
 			break;
 	}
 }
+
+bool
+SyncRepConfigured()
+{
+	if (!SyncRepRequested() ||
+		!((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
+		return false;
+
+	return true;
+}
diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt
index 8efb4044d6..c40884c44e 100644
--- a/src/backend/utils/activity/wait_event_names.txt
+++ b/src/backend/utils/activity/wait_event_names.txt
@@ -347,6 +347,7 @@ DSMRegistry	"Waiting to read or update the dynamic shared memory registry."
 InjectionPoint	"Waiting to read or update information related to injection points."
 SerialControl	"Waiting to read or update shared <filename>pg_serial</filename> state."
 WaitLSN	"Waiting to read or update shared Wait-for-LSN state."
+SyncRepLogicalCache	"Waiting to read or update information about the tatc of synchronous replication LSN cache."
 
 #
 # END OF PREDEFINED LWLOCKS (DO NOT CHANGE THIS LINE)
diff --git a/src/include/replication/syncrep.h b/src/include/replication/syncrep.h
index ea439e6da6..ec22bc72df 100644
--- a/src/include/replication/syncrep.h
+++ b/src/include/replication/syncrep.h
@@ -82,6 +82,7 @@ extern PGDLLIMPORT char *SyncRepStandbyNames;
 
 /* called by user backend */
 extern void SyncRepWaitForLSN(XLogRecPtr lsn, bool commit);
+extern bool SyncRepConfigured();
 
 /* called at backend exit */
 extern void SyncRepCleanupAtProcExit(void);
@@ -96,6 +97,8 @@ extern int	SyncRepGetCandidateStandbys(SyncRepStandbyData **standbys);
 /* called by checkpointer */
 extern void SyncRepUpdateSyncStandbysDefined(void);
 
+extern int	SyncRepWaitMode;
+
 /*
  * Internal functions for parsing synchronous_standby_names grammar,
  * in syncrep_gram.y and syncrep_scanner.l
diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h
index cf32ac2488..878492d3ae 100644
--- a/src/include/replication/walsender_private.h
+++ b/src/include/replication/walsender_private.h
@@ -102,6 +102,12 @@ typedef struct
 	 */
 	XLogRecPtr	lsn[NUM_SYNC_REP_WAIT_MODE];
 
+	/*
+	 * Cache of lsn[NUM_SYNC_REP_WAIT_MODE] above when SyncRep is enabled. Checked by logical walsenders
+	 * first to reduce contention on SyncRepLock. Protected by SyncRepLogicalCacheLock.
+	 */
+	XLogRecPtr cached_lsn[NUM_SYNC_REP_WAIT_MODE];
+
 	/*
 	 * Are any sync standbys defined?  Waiting backends can't reload the
 	 * config file safely, so checkpointer updates this value as needed.
diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h
index 88dc79b2bd..8ea84e5ec0 100644
--- a/src/include/storage/lwlocklist.h
+++ b/src/include/storage/lwlocklist.h
@@ -84,3 +84,4 @@ PG_LWLOCK(50, DSMRegistry)
 PG_LWLOCK(51, InjectionPoint)
 PG_LWLOCK(52, SerialControl)
 PG_LWLOCK(53, WaitLSN)
+PG_LWLOCK(54, SyncRepLogicalCache)
\ No newline at end of file
diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl
index 2c51cfc3c8..00130d92dc 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -710,6 +710,141 @@ $result = $subscriber1->safe_psql('postgres',
 is($result, 't',
 	"subscriber1 gets data from primary after standby1 acknowledges changes");
 
+##################################################
+# Test that logical failover replication slots wait for the specified
+# synchronous replicas to receive the changes first. It uses the
+# following set up:
+#
+# 				(synchronous physical standbys)
+#				| ----> standby1 (application_name = standby1)
+#				| ----> standby2 (application_name = standby2)
+# primary -----	|
+#				(logical replication)
+#				| ----> subscriber1 (failover = true, slot_name = lsub1_slot)
+#				| ----> subscriber2 (failover = false, slot_name = lsub2_slot)
+#
+# synchronous_commit = 'on'
+# synchronous_standby_names = 'ANY 2 (standby1, standby2)'
+#
+# The setup is configured in such a way that the logical slot of subscriber1 is
+# enabled for failover, and thus the subscriber1 will wait for the changes to have
+# been synchronously replicated before receiving the decoded changes.
+##################################################
+
+$primary->safe_psql('postgres', "TRUNCATE tab_int;");
+# Setup synchronous replication
+$primary->append_conf(
+	'postgresql.conf', qq(
+		synchronous_commit = 'on'
+		synchronous_standby_names = 'ANY 2 (standby1, standby2)'
+));
+
+$primary->reload;
+
+$standby1->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby1'");
+$standby1->reload;
+
+$standby2->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby2'");
+$standby2->reload;
+
+# Check that synchronous replication is setup properly
+$standby2->stop;
+
+# Create some data on the primary
+$primary_row_count = 10;
+
+my $sync_back_q = $primary->background_psql(
+	'postgres',
+	on_error_stop => 0,
+	timeout => $PostgreSQL::Test::Utils::timeout_default);
+
+$sync_back_q->query_until(qr/insert_blocked_on_sync_rep/, q(
+    \echo insert_blocked_on_sync_rep
+   INSERT INTO tab_int SELECT generate_series(1, 10);
+));
+
+$result = $primary->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+
+is($result, 'f', "primary row count is not updated due to synchronous replication");
+
+# Verify the standby specified in synchronized_standby_slots (sb1_slot aka standby1)
+# catches up with the primary.
+$primary->wait_for_replay_catchup($standby1);
+$result = $standby1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't', "standby1 gets data from primary");
+
+# Validate that synchronized_standby_slots takes precedence over waiting for
+# changes to have been synchronous replicated.
+# Since the slot specified (sb1_slot) has received the changes, primary can send
+# the decoded changes to the subscription enabled for failover
+# (i.e. regress_mysub1).
+$primary->wait_for_catchup('regress_mysub1');
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber1 gets data from primary after standby1 acknowledges changes");
+
+$standby2->start;
+$primary->wait_for_replay_catchup($standby2);
+
+# Unset synchronized_standby_slots to test synchronous replication
+# blocks primary from sending logical decoded changes to failover slots until
+# changes have been synchronously replicated.
+$primary->append_conf(
+	'postgresql.conf', qq(
+	synchronized_standby_slots = ''));
+$primary->reload;
+
+$primary_row_count = 20;
+$standby2->stop;
+
+$sync_back_q->query_until(
+	qr/insert_blocked_on_additional_sync_rep/, q(
+	\echo insert_blocked_on_additional_sync_rep
+	INSERT INTO tab_int SELECT generate_series(11, 20);
+));
+
+# Since $standby2 has not received the changes, validate that subscriber1 (failover = true)
+# has not received the decoded changes, but subscriber2 (failover = false) has.
+$primary->wait_for_catchup('regress_mysub2');
+
+$result = $subscriber2->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber2 gets data from primary even if the changes have not been synchronously acknowledged.");
+
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 'f',
+	"subscriber1 does not get data from primary since changes have not been synchronously acknowledged.");
+
+# Start standby2 to allow the changes to be acknowledged by all the synchronous standbys.
+$standby2->start;
+$primary->wait_for_replay_catchup($standby2);
+$primary->wait_for_catchup('regress_mysub1');
+
+# Now that the changes have been replicated to all synchronous nodes,
+# primary can send the decoded changes to the subscription enabled for failover
+# (i.e. regress_mysub1). While the standby was down, regress_mysub1 didn't
+# receive any data from the primary. i.e. the primary didn't allow it to go
+# ahead of requirements of synchronous commit.
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber1 gets data from primary since changes have been syhcronously acknowledged.");
+# No longer need to run the background session.
+$sync_back_q->quit;
+
+# Reset synchronized_standby_slots and synchronous commit for below test cases
+$primary->append_conf(
+	'postgresql.conf', qq(
+synchronized_standby_slots = 'sb1_slot'
+synchronous_standby_names = ''
+));
+$primary->reload;
+
 ##################################################
 # Verify that when using pg_logical_slot_get_changes to consume changes from a
 # logical failover slot, it will also wait for the slots specified in
-- 
2.40.1

From 603e01d634a30de0a0103f657010e4fce953390e Mon Sep 17 00:00:00 2001
From: John Hsu <johnhyvr@gmail.com>
Date: Mon, 26 Aug 2024 18:23:20 +0000
Subject: [PATCH] Wait on synchronous replication by default for logical
 failover slots

If synchronous replication is enabled, this patch
allows logical subscribers with failover_slots enabled
to wait for changes to be replicated to synchronous replicas
before consuming the changes.
---
 src/backend/replication/slot.c                | 266 ++++++++++--------
 src/backend/replication/syncrep.c             |  27 +-
 src/include/replication/syncrep.h             |   3 +
 src/include/storage/lwlocklist.h              |   2 +-
 .../t/040_standby_failover_slots_sync.pl      | 135 +++++++++
 5 files changed, 316 insertions(+), 117 deletions(-)

diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index c290339af5..460358d259 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -49,6 +49,7 @@
 #include "postmaster/interrupt.h"
 #include "replication/slotsync.h"
 #include "replication/slot.h"
+#include "replication/syncrep.h"
 #include "replication/walsender_private.h"
 #include "storage/fd.h"
 #include "storage/ipc.h"
@@ -2591,8 +2592,8 @@ SlotExistsInSyncStandbySlots(const char *slot_name)
 }
 
 /*
- * Return true if the slots specified in synchronized_standby_slots have caught up to
- * the given WAL location, false otherwise.
+ * Return true if the slots specified in synchronized_standby_slots or synchronous
+ * replication have caught up to the given WAL location, false otherwise.
  *
  * The elevel parameter specifies the error level used for logging messages
  * related to slots that do not exist, are invalidated, or are inactive.
@@ -2606,9 +2607,9 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 
 	/*
 	 * Don't need to wait for the standbys to catch up if there is no value in
-	 * synchronized_standby_slots.
+	 * synchronized_standby_slots or synchronous replication is not configured.
 	 */
-	if (synchronized_standby_slots_config == NULL)
+	if (synchronized_standby_slots_config == NULL && !SyncRepConfigured())
 		return true;
 
 	/*
@@ -2619,144 +2620,183 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 		return true;
 
 	/*
-	 * Don't need to wait for the standbys to catch up if they are already
-	 * beyond the specified WAL location.
-	 */
-	if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) &&
-		ss_oldest_flush_lsn >= wait_for_lsn)
-		return true;
-
-	/*
-	 * To prevent concurrent slot dropping and creation while filtering the
-	 * slots, take the ReplicationSlotControlLock outside of the loop.
+	 * In the event that synchronized_standby_slots and synchronous replication is
+	 * configured, have the former take precedence.
 	 */
-	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
-
-	name = synchronized_standby_slots_config->slot_names;
-	for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
+	if (synchronized_standby_slots_config != NULL)
 	{
-		XLogRecPtr	restart_lsn;
-		bool		invalidated;
-		bool		inactive;
-		ReplicationSlot *slot;
-
-		slot = SearchNamedReplicationSlot(name, false);
+		/*
+		 * Don't need to wait for the standbys to catch up if they are already
+		 * beyond the specified WAL location.
+		 */
+		if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) &&
+			ss_oldest_flush_lsn >= wait_for_lsn)
+			return true;
 
-		if (!slot)
+		/*
+		 * To prevent concurrent slot dropping and creation while filtering the
+		 * slots, take the ReplicationSlotControlLock outside of the loop.
+		 */
+		LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+		name = synchronized_standby_slots_config->slot_names;
+		for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
 		{
-			/*
-			 * If a slot name provided in synchronized_standby_slots does not
-			 * exist, report a message and exit the loop. A user can specify a
-			 * slot name that does not exist just before the server startup.
-			 * The GUC check_hook(validate_sync_standby_slots) cannot validate
-			 * such a slot during startup as the ReplicationSlotCtl shared
-			 * memory is not initialized at that time. It is also possible for
-			 * a user to drop the slot in synchronized_standby_slots
-			 * afterwards.
-			 */
-			ereport(elevel,
-					errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					errmsg("replication slot \"%s\" specified in parameter %s does not exist",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting on the standby associated with \"%s\".",
-							  name),
-					errhint("Consider creating the slot \"%s\" or amend parameter %s.",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			XLogRecPtr	restart_lsn;
+			bool		invalidated;
+			bool		inactive;
+			ReplicationSlot *slot;
 
-		if (SlotIsLogical(slot))
-		{
-			/*
-			 * If a logical slot name is provided in
-			 * synchronized_standby_slots, report a message and exit the loop.
-			 * Similar to the non-existent case, a user can specify a logical
-			 * slot name in synchronized_standby_slots before the server
-			 * startup, or drop an existing physical slot and recreate a
-			 * logical slot with the same name.
-			 */
-			ereport(elevel,
-					errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					errmsg("cannot have logical replication slot \"%s\" in parameter %s",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting for correction on \"%s\".",
-							  name),
-					errhint("Consider removing logical slot \"%s\" from parameter %s.",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			slot = SearchNamedReplicationSlot(name, false);
 
-		SpinLockAcquire(&slot->mutex);
-		restart_lsn = slot->data.restart_lsn;
-		invalidated = slot->data.invalidated != RS_INVAL_NONE;
-		inactive = slot->active_pid == 0;
-		SpinLockRelease(&slot->mutex);
+			if (!slot)
+			{
+				/*
+				 * If a slot name provided in synchronized_standby_slots does not
+				 * exist, report a message and exit the loop. A user can specify a
+				 * slot name that does not exist just before the server startup.
+				 * The GUC check_hook(validate_sync_standby_slots) cannot validate
+				 * such a slot during startup as the ReplicationSlotCtl shared
+				 * memory is not initialized at that time. It is also possible for
+				 * a user to drop the slot in synchronized_standby_slots
+				 * afterwards.
+				 */
+				ereport(elevel,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("replication slot \"%s\" specified in parameter %s does not exist",
+							name, "synchronized_standby_slots"),
+						errdetail("Logical replication is waiting on the standby associated with \"%s\".",
+								name),
+						errhint("Consider creating the slot \"%s\" or amend parameter %s.",
+								name, "synchronized_standby_slots"));
+				break;
+			}
 
-		if (invalidated)
-		{
-			/* Specified physical slot has been invalidated */
-			ereport(elevel,
-					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-					errmsg("physical slot \"%s\" specified in parameter %s has been invalidated",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting on the standby associated with \"%s\".",
-							  name),
-					errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			if (SlotIsLogical(slot))
+			{
+				/*
+				 * If a logical slot name is provided in
+				 * synchronized_standby_slots, report a message and exit the loop.
+				 * Similar to the non-existent case, a user can specify a logical
+				 * slot name in synchronized_standby_slots before the server
+				 * startup, or drop an existing physical slot and recreate a
+				 * logical slot with the same name.
+				 */
+				ereport(elevel,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("cannot have logical replication slot \"%s\" in parameter %s",
+							name, "synchronized_standby_slots"),
+						errdetail("Logical replication is waiting for correction on \"%s\".",
+								name),
+						errhint("Consider removing logical slot \"%s\" from parameter %s.",
+								name, "synchronized_standby_slots"));
+				break;
+			}
 
-		if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
-		{
-			/* Log a message if no active_pid for this physical slot */
-			if (inactive)
+			SpinLockAcquire(&slot->mutex);
+			restart_lsn = slot->data.restart_lsn;
+			invalidated = slot->data.invalidated != RS_INVAL_NONE;
+			inactive = slot->active_pid == 0;
+			SpinLockRelease(&slot->mutex);
+
+			if (invalidated)
+			{
+				/* Specified physical slot has been invalidated */
 				ereport(elevel,
 						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-						errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid",
-							   name, "synchronized_standby_slots"),
+						errmsg("physical slot \"%s\" specified in parameter %s has been invalidated",
+							name, "synchronized_standby_slots"),
 						errdetail("Logical replication is waiting on the standby associated with \"%s\".",
-								  name),
-						errhint("Consider starting standby associated with \"%s\" or amend parameter %s.",
+								name),
+						errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.",
 								name, "synchronized_standby_slots"));
+				break;
+			}
 
-			/* Continue if the current slot hasn't caught up. */
-			break;
+			if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
+			{
+				/* Log a message if no active_pid for this physical slot */
+				if (inactive)
+					ereport(elevel,
+							errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+							errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid",
+								name, "synchronized_standby_slots"),
+							errdetail("Logical replication is waiting on the standby associated with \"%s\".",
+									name),
+							errhint("Consider starting standby associated with \"%s\" or amend parameter %s.",
+									name, "synchronized_standby_slots"));
+
+				/* Continue if the current slot hasn't caught up. */
+				break;
+			}
+
+			Assert(restart_lsn >= wait_for_lsn);
+
+			if (XLogRecPtrIsInvalid(min_restart_lsn) ||
+				min_restart_lsn > restart_lsn)
+				min_restart_lsn = restart_lsn;
+
+			caught_up_slot_num++;
+
+			name += strlen(name) + 1;
 		}
 
-		Assert(restart_lsn >= wait_for_lsn);
+		LWLockRelease(ReplicationSlotControlLock);
+
+		/*
+		 * Return false if not all the standbys have caught up to the specified
+		 * WAL location.
+		 */
+		if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
+			return false;
 
-		if (XLogRecPtrIsInvalid(min_restart_lsn) ||
-			min_restart_lsn > restart_lsn)
-			min_restart_lsn = restart_lsn;
+		/* The ss_oldest_flush_lsn must not retreat. */
+		Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) ||
+			min_restart_lsn >= ss_oldest_flush_lsn);
 
-		caught_up_slot_num++;
+		ss_oldest_flush_lsn = min_restart_lsn;
 
-		name += strlen(name) + 1;
+		return true;
 	}
+	else
+	{
+		volatile WalSndCtlData *walsndctl = WalSndCtl;
+		static XLogRecPtr	lsn[NUM_SYNC_REP_WAIT_MODE]; /* Cache LSNs */
+		static bool initialized = false;
+		/* Initialize value in case SIGHUP changing to SYNC_REP_NO_WAIT */
+		int mode = SyncRepWaitMode;
+		int i;
+
+		if (!initialized)
+		{
+			for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
+			{
+				lsn[i] = InvalidXLogRecPtr;
+			}
+		}
 
-	LWLockRelease(ReplicationSlotControlLock);
+		if (mode == SYNC_REP_NO_WAIT)
+			return true;
 
-	/*
-	 * Return false if not all the standbys have caught up to the specified
-	 * WAL location.
-	 */
-	if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
-		return false;
+		if (lsn[mode] >= wait_for_lsn)
+			return true;
 
-	/* The ss_oldest_flush_lsn must not retreat. */
-	Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) ||
-		   min_restart_lsn >= ss_oldest_flush_lsn);
+		LWLockAcquire(SyncRepLock, LW_SHARED);
+		memcpy(lsn, (XLogRecPtr *) walsndctl->lsn, sizeof(lsn));
+		LWLockRelease(SyncRepLock);
 
-	ss_oldest_flush_lsn = min_restart_lsn;
+		if (lsn[mode] >= wait_for_lsn)
+			return true;
 
-	return true;
+		return false;
+	}
 }
 
 /*
  * Wait for physical standbys to confirm receiving the given lsn.
  *
  * Used by logical decoding SQL functions. It waits for physical standbys
- * corresponding to the physical slots specified in the synchronized_standby_slots GUC.
+ * corresponding to the physical slots specified in the synchronized_standby_slots GUC,
+ * or synchronous replication.
  */
 void
 WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
@@ -2766,7 +2806,7 @@ WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
 	 * slot is not a logical failover slot, or there is no value in
 	 * synchronized_standby_slots.
 	 */
-	if (!MyReplicationSlot->data.failover || !synchronized_standby_slots_config)
+	if (!MyReplicationSlot->data.failover || !(synchronized_standby_slots_config || SyncRepConfigured()))
 		return;
 
 	ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv);
diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c
index fa5988c824..69c061d8dc 100644
--- a/src/backend/replication/syncrep.c
+++ b/src/backend/replication/syncrep.c
@@ -78,6 +78,7 @@
 #include "common/int.h"
 #include "miscadmin.h"
 #include "pgstat.h"
+#include "replication/slot.h"
 #include "replication/syncrep.h"
 #include "replication/walsender.h"
 #include "replication/walsender_private.h"
@@ -95,7 +96,7 @@ char	   *SyncRepStandbyNames;
 static bool announce_next_takeover = true;
 
 SyncRepConfigData *SyncRepConfig = NULL;
-static int	SyncRepWaitMode = SYNC_REP_NO_WAIT;
+int	SyncRepWaitMode = SYNC_REP_NO_WAIT;
 
 static void SyncRepQueueInsert(int mode);
 static void SyncRepCancelWait(void);
@@ -124,6 +125,8 @@ static int	cmp_lsn(const void *a, const void *b);
 static bool SyncRepQueueIsOrderedByLSN(int mode);
 #endif
 
+bool SyncRepConfigured(void);
+
 /*
  * ===========================================================
  * Synchronous Replication functions for normal user backends
@@ -169,8 +172,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit)
 	 * described in SyncRepUpdateSyncStandbysDefined(). On the other hand, if
 	 * it's false, the lock is not necessary because we don't touch the queue.
 	 */
-	if (!SyncRepRequested() ||
-		!((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
+	if (!SyncRepConfigured())
 		return;
 
 	/* Cap the level for anything other than commit to remote flush only. */
@@ -523,6 +525,15 @@ SyncRepReleaseWaiters(void)
 
 	LWLockRelease(SyncRepLock);
 
+	/*
+	 * If synchronized_standby_slots is set, the respective walsender's
+	 * will be responsible for broadcasting the value.
+	 */
+	if (strcmp(synchronized_standby_slots, "") == 0)
+	{
+		ConditionVariableBroadcast(&WalSndCtl->wal_confirm_rcv_cv);
+	}
+
 	elog(DEBUG3, "released %d procs up to write %X/%X, %d procs up to flush %X/%X, %d procs up to apply %X/%X",
 		 numwrite, LSN_FORMAT_ARGS(writePtr),
 		 numflush, LSN_FORMAT_ARGS(flushPtr),
@@ -1069,3 +1080,13 @@ assign_synchronous_commit(int newval, void *extra)
 			break;
 	}
 }
+
+bool
+SyncRepConfigured()
+{
+	if (!SyncRepRequested() ||
+		!((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
+		return false;
+
+	return true;
+}
diff --git a/src/include/replication/syncrep.h b/src/include/replication/syncrep.h
index ea439e6da6..ec22bc72df 100644
--- a/src/include/replication/syncrep.h
+++ b/src/include/replication/syncrep.h
@@ -82,6 +82,7 @@ extern PGDLLIMPORT char *SyncRepStandbyNames;
 
 /* called by user backend */
 extern void SyncRepWaitForLSN(XLogRecPtr lsn, bool commit);
+extern bool SyncRepConfigured();
 
 /* called at backend exit */
 extern void SyncRepCleanupAtProcExit(void);
@@ -96,6 +97,8 @@ extern int	SyncRepGetCandidateStandbys(SyncRepStandbyData **standbys);
 /* called by checkpointer */
 extern void SyncRepUpdateSyncStandbysDefined(void);
 
+extern int	SyncRepWaitMode;
+
 /*
  * Internal functions for parsing synchronous_standby_names grammar,
  * in syncrep_gram.y and syncrep_scanner.l
diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h
index 88dc79b2bd..e44bc3a07b 100644
--- a/src/include/storage/lwlocklist.h
+++ b/src/include/storage/lwlocklist.h
@@ -83,4 +83,4 @@ PG_LWLOCK(49, WALSummarizer)
 PG_LWLOCK(50, DSMRegistry)
 PG_LWLOCK(51, InjectionPoint)
 PG_LWLOCK(52, SerialControl)
-PG_LWLOCK(53, WaitLSN)
+PG_LWLOCK(53, WaitLSN)
\ No newline at end of file
diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl
index 2c51cfc3c8..00130d92dc 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -710,6 +710,141 @@ $result = $subscriber1->safe_psql('postgres',
 is($result, 't',
 	"subscriber1 gets data from primary after standby1 acknowledges changes");
 
+##################################################
+# Test that logical failover replication slots wait for the specified
+# synchronous replicas to receive the changes first. It uses the
+# following set up:
+#
+# 				(synchronous physical standbys)
+#				| ----> standby1 (application_name = standby1)
+#				| ----> standby2 (application_name = standby2)
+# primary -----	|
+#				(logical replication)
+#				| ----> subscriber1 (failover = true, slot_name = lsub1_slot)
+#				| ----> subscriber2 (failover = false, slot_name = lsub2_slot)
+#
+# synchronous_commit = 'on'
+# synchronous_standby_names = 'ANY 2 (standby1, standby2)'
+#
+# The setup is configured in such a way that the logical slot of subscriber1 is
+# enabled for failover, and thus the subscriber1 will wait for the changes to have
+# been synchronously replicated before receiving the decoded changes.
+##################################################
+
+$primary->safe_psql('postgres', "TRUNCATE tab_int;");
+# Setup synchronous replication
+$primary->append_conf(
+	'postgresql.conf', qq(
+		synchronous_commit = 'on'
+		synchronous_standby_names = 'ANY 2 (standby1, standby2)'
+));
+
+$primary->reload;
+
+$standby1->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby1'");
+$standby1->reload;
+
+$standby2->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby2'");
+$standby2->reload;
+
+# Check that synchronous replication is setup properly
+$standby2->stop;
+
+# Create some data on the primary
+$primary_row_count = 10;
+
+my $sync_back_q = $primary->background_psql(
+	'postgres',
+	on_error_stop => 0,
+	timeout => $PostgreSQL::Test::Utils::timeout_default);
+
+$sync_back_q->query_until(qr/insert_blocked_on_sync_rep/, q(
+    \echo insert_blocked_on_sync_rep
+   INSERT INTO tab_int SELECT generate_series(1, 10);
+));
+
+$result = $primary->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+
+is($result, 'f', "primary row count is not updated due to synchronous replication");
+
+# Verify the standby specified in synchronized_standby_slots (sb1_slot aka standby1)
+# catches up with the primary.
+$primary->wait_for_replay_catchup($standby1);
+$result = $standby1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't', "standby1 gets data from primary");
+
+# Validate that synchronized_standby_slots takes precedence over waiting for
+# changes to have been synchronous replicated.
+# Since the slot specified (sb1_slot) has received the changes, primary can send
+# the decoded changes to the subscription enabled for failover
+# (i.e. regress_mysub1).
+$primary->wait_for_catchup('regress_mysub1');
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber1 gets data from primary after standby1 acknowledges changes");
+
+$standby2->start;
+$primary->wait_for_replay_catchup($standby2);
+
+# Unset synchronized_standby_slots to test synchronous replication
+# blocks primary from sending logical decoded changes to failover slots until
+# changes have been synchronously replicated.
+$primary->append_conf(
+	'postgresql.conf', qq(
+	synchronized_standby_slots = ''));
+$primary->reload;
+
+$primary_row_count = 20;
+$standby2->stop;
+
+$sync_back_q->query_until(
+	qr/insert_blocked_on_additional_sync_rep/, q(
+	\echo insert_blocked_on_additional_sync_rep
+	INSERT INTO tab_int SELECT generate_series(11, 20);
+));
+
+# Since $standby2 has not received the changes, validate that subscriber1 (failover = true)
+# has not received the decoded changes, but subscriber2 (failover = false) has.
+$primary->wait_for_catchup('regress_mysub2');
+
+$result = $subscriber2->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber2 gets data from primary even if the changes have not been synchronously acknowledged.");
+
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 'f',
+	"subscriber1 does not get data from primary since changes have not been synchronously acknowledged.");
+
+# Start standby2 to allow the changes to be acknowledged by all the synchronous standbys.
+$standby2->start;
+$primary->wait_for_replay_catchup($standby2);
+$primary->wait_for_catchup('regress_mysub1');
+
+# Now that the changes have been replicated to all synchronous nodes,
+# primary can send the decoded changes to the subscription enabled for failover
+# (i.e. regress_mysub1). While the standby was down, regress_mysub1 didn't
+# receive any data from the primary. i.e. the primary didn't allow it to go
+# ahead of requirements of synchronous commit.
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber1 gets data from primary since changes have been syhcronously acknowledged.");
+# No longer need to run the background session.
+$sync_back_q->quit;
+
+# Reset synchronized_standby_slots and synchronous commit for below test cases
+$primary->append_conf(
+	'postgresql.conf', qq(
+synchronized_standby_slots = 'sb1_slot'
+synchronous_standby_names = ''
+));
+$primary->reload;
+
 ##################################################
 # Verify that when using pg_logical_slot_get_changes to consume changes from a
 # logical failover slot, it will also wait for the slots specified in
-- 
2.40.1

From 366303c7ff79e4bea2c7c6eb491b2ce641d90b72 Mon Sep 17 00:00:00 2001
From: John Hsu <johnhyvr@gmail.com>
Date: Mon, 26 Aug 2024 18:23:20 +0000
Subject: [PATCH] Wait on synchronous replication by default for logical
 failover slots

If synchronous replication is enabled, this patch
allows logical subscribers with failover_slots enabled
to wait for changes to be replicated to synchronous replicas
before consuming the changes.
---
 src/backend/replication/slot.c                | 259 ++++++++++--------
 src/backend/replication/syncrep.c             |  27 +-
 src/include/replication/syncrep.h             |   3 +
 .../t/040_standby_failover_slots_sync.pl      | 135 +++++++++
 4 files changed, 311 insertions(+), 113 deletions(-)

diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 0a03776156..0708e46468 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -49,6 +49,7 @@
 #include "postmaster/interrupt.h"
 #include "replication/slotsync.h"
 #include "replication/slot.h"
+#include "replication/syncrep.h"
 #include "replication/walsender_private.h"
 #include "storage/fd.h"
 #include "storage/ipc.h"
@@ -2591,8 +2592,8 @@ SlotExistsInSyncStandbySlots(const char *slot_name)
 }
 
 /*
- * Return true if the slots specified in synchronized_standby_slots have caught up to
- * the given WAL location, false otherwise.
+ * Return true if the slots specified in synchronized_standby_slots or synchronous
+ * replication have caught up to the given WAL location, false otherwise.
  *
  * The elevel parameter specifies the error level used for logging messages
  * related to slots that do not exist, are invalidated, or are inactive.
@@ -2606,9 +2607,9 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 
 	/*
 	 * Don't need to wait for the standbys to catch up if there is no value in
-	 * synchronized_standby_slots.
+	 * synchronized_standby_slots or synchronous replication is not configured.
 	 */
-	if (synchronized_standby_slots_config == NULL)
+	if (synchronized_standby_slots_config == NULL && !SyncRepConfigured())
 		return true;
 
 	/*
@@ -2619,144 +2620,182 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 		return true;
 
 	/*
-	 * Don't need to wait for the standbys to catch up if they are already
-	 * beyond the specified WAL location.
-	 */
-	if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) &&
-		ss_oldest_flush_lsn >= wait_for_lsn)
-		return true;
-
-	/*
-	 * To prevent concurrent slot dropping and creation while filtering the
-	 * slots, take the ReplicationSlotControlLock outside of the loop.
+	 * In the event that synchronized_standby_slots and synchronous replication is
+	 * configured, have the former take precedence.
 	 */
-	LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
-
-	name = synchronized_standby_slots_config->slot_names;
-	for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
+	if (synchronized_standby_slots_config != NULL)
 	{
-		XLogRecPtr	restart_lsn;
-		bool		invalidated;
-		bool		inactive;
-		ReplicationSlot *slot;
+		/*
+		 * Don't need to wait for the standbys to catch up if they are already
+		 * beyond the specified WAL location.
+		 */
+		if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) &&
+			ss_oldest_flush_lsn >= wait_for_lsn)
+			return true;
 
-		slot = SearchNamedReplicationSlot(name, false);
+		/*
+		 * To prevent concurrent slot dropping and creation while filtering the
+		 * slots, take the ReplicationSlotControlLock outside of the loop.
+		 */
+		LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
-		if (!slot)
+		name = synchronized_standby_slots_config->slot_names;
+		for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
 		{
-			/*
-			 * If a slot name provided in synchronized_standby_slots does not
-			 * exist, report a message and exit the loop. A user can specify a
-			 * slot name that does not exist just before the server startup.
-			 * The GUC check_hook(validate_sync_standby_slots) cannot validate
-			 * such a slot during startup as the ReplicationSlotCtl shared
-			 * memory is not initialized at that time. It is also possible for
-			 * a user to drop the slot in synchronized_standby_slots
-			 * afterwards.
-			 */
-			ereport(elevel,
-					errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					errmsg("replication slot \"%s\" specified in parameter \"%s\" does not exist",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
-							  name),
-					errhint("Create the replication slot \"%s\" or amend parameter \"%s\".",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			XLogRecPtr	restart_lsn;
+			bool		invalidated;
+			bool		inactive;
+			ReplicationSlot	*slot;
 
-		if (SlotIsLogical(slot))
-		{
-			/*
-			 * If a logical slot name is provided in
-			 * synchronized_standby_slots, report a message and exit the loop.
-			 * Similar to the non-existent case, a user can specify a logical
-			 * slot name in synchronized_standby_slots before the server
-			 * startup, or drop an existing physical slot and recreate a
-			 * logical slot with the same name.
-			 */
-			ereport(elevel,
-					errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-					errmsg("cannot specify logical replication slot \"%s\" in parameter \"%s\"",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting for correction on replication slot \"%s\".",
-							  name),
-					errhint("Remove the logical replication slot \"%s\" from parameter \"%s\".",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			slot = SearchNamedReplicationSlot(name, false);
 
-		SpinLockAcquire(&slot->mutex);
-		restart_lsn = slot->data.restart_lsn;
-		invalidated = slot->data.invalidated != RS_INVAL_NONE;
-		inactive = slot->active_pid == 0;
-		SpinLockRelease(&slot->mutex);
+			if (!slot)
+			{
+				/*
+				 * If a slot name provided in synchronized_standby_slots does not
+				 * exist, report a message and exit the loop. A user can specify a
+				 * slot name that does not exist just before the server startup.
+				 * The GUC check_hook(validate_sync_standby_slots) cannot validate
+				 * such a slot during startup as the ReplicationSlotCtl shared
+				 * memory is not initialized at that time. It is also possible for
+				 * a user to drop the slot in synchronized_standby_slots
+				 * afterwards.
+				 */
+				ereport(elevel,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("replication slot \"%s\" specified in parameter \"%s\" does not exist",
+							   name, "synchronized_standby_slots"),
+						errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
+								  name),
+						errhint("Create the replication slot \"%s\" or amend parameter \"%s\".",
+								name, "synchronized_standby_slots"));
+				break;
+			}
 
-		if (invalidated)
-		{
-			/* Specified physical slot has been invalidated */
-			ereport(elevel,
-					errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-					errmsg("physical replication slot \"%s\" specified in parameter \"%s\" has been invalidated",
-						   name, "synchronized_standby_slots"),
-					errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
-							  name),
-					errhint("Drop and recreate the replication slot \"%s\", or amend parameter \"%s\".",
-							name, "synchronized_standby_slots"));
-			break;
-		}
+			if (SlotIsLogical(slot))
+			{
+				/*
+				 * If a logical slot name is provided in
+				 * synchronized_standby_slots, report a message and exit the loop.
+				 * Similar to the non-existent case, a user can specify a logical
+				 * slot name in synchronized_standby_slots before the server
+				 * startup, or drop an existing physical slot and recreate a
+				 * logical slot with the same name.
+				 */
+				ereport(elevel,
+						errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+						errmsg("cannot specify logical replication slot \"%s\" in parameter \"%s\"",
+							   name, "synchronized_standby_slots"),
+						errdetail("Logical replication is waiting for correction on replication slot \"%s\".",
+								  name),
+						errhint("Remove the logical replication slot \"%s\" from parameter \"%s\".",
+								name, "synchronized_standby_slots"));
+				break;
+			}
 
-		if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
-		{
-			/* Log a message if no active_pid for this physical slot */
-			if (inactive)
+			SpinLockAcquire(&slot->mutex);
+			restart_lsn = slot->data.restart_lsn;
+			invalidated = slot->data.invalidated != RS_INVAL_NONE;
+			inactive = slot->active_pid == 0;
+			SpinLockRelease(&slot->mutex);
+
+			if (invalidated)
+			{
+				/* Specified physical slot has been invalidated */
 				ereport(elevel,
 						errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-						errmsg("replication slot \"%s\" specified in parameter \"%s\" does not have active_pid",
+						errmsg("physical replication slot \"%s\" specified in parameter \"%s\" has been invalidated",
 							   name, "synchronized_standby_slots"),
 						errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
 								  name),
-						errhint("Start the standby associated with the replication slot \"%s\", or amend parameter \"%s\".",
+						errhint("Drop and recreate the replication slot \"%s\", or amend parameter \"%s\".",
 								name, "synchronized_standby_slots"));
+				break;
+			}
 
-			/* Continue if the current slot hasn't caught up. */
-			break;
+			if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
+			{
+				/* Log a message if no active_pid for this physical slot */
+				if (inactive)
+					ereport(elevel,
+							errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+							errmsg("replication slot \"%s\" specified in parameter \"%s\" does not have active_pid",
+								   name, "synchronized_standby_slots"),
+							errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
+									  name),
+							errhint("Start the standby associated with the replication slot \"%s\", or amend parameter \"%s\".",
+									name, "synchronized_standby_slots"));
+
+				/* Continue if the current slot hasn't caught up. */
+				break;
+			}
+
+			Assert(restart_lsn >= wait_for_lsn);
+
+			if (XLogRecPtrIsInvalid(min_restart_lsn) ||
+				min_restart_lsn > restart_lsn)
+				min_restart_lsn = restart_lsn;
+
+			caught_up_slot_num++;
+
+			name += strlen(name) + 1;
 		}
 
-		Assert(restart_lsn >= wait_for_lsn);
+		LWLockRelease(ReplicationSlotControlLock);
 
-		if (XLogRecPtrIsInvalid(min_restart_lsn) ||
-			min_restart_lsn > restart_lsn)
-			min_restart_lsn = restart_lsn;
+		/*
+		 * Return false if not all the standbys have caught up to the specified
+		 * WAL location.
+		 */
+		if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
+			return false;
 
-		caught_up_slot_num++;
+		/* The ss_oldest_flush_lsn must not retreat. */
+		Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) ||
+			   min_restart_lsn >= ss_oldest_flush_lsn);
 
-		name += strlen(name) + 1;
+		ss_oldest_flush_lsn = min_restart_lsn;
+
+		return true;
 	}
+	else
+	{
+		volatile WalSndCtlData *walsndctl = WalSndCtl;
+		static XLogRecPtr	lsn[NUM_SYNC_REP_WAIT_MODE]; /* Cache LSNs */
+		static bool initialized = false;
+		int i;
 
-	LWLockRelease(ReplicationSlotControlLock);
+		if (!initialized)
+		{
+			for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
+			{
+				lsn[i] = InvalidXLogRecPtr;
+			}
+			initialized = true;
+		}
 
-	/*
-	 * Return false if not all the standbys have caught up to the specified
-	 * WAL location.
-	 */
-	if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
-		return false;
+		Assert(SyncRepWaitMode >= 0);
 
-	/* The ss_oldest_flush_lsn must not retreat. */
-	Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) ||
-		   min_restart_lsn >= ss_oldest_flush_lsn);
+		if (lsn[SyncRepWaitMode] >= wait_for_lsn)
+			return true;
 
-	ss_oldest_flush_lsn = min_restart_lsn;
+		LWLockAcquire(SyncRepLock, LW_SHARED);
+		memcpy(lsn, (XLogRecPtr *) walsndctl->lsn, sizeof(lsn));
+		LWLockRelease(SyncRepLock);
 
-	return true;
+		if (lsn[SyncRepWaitMode] >= wait_for_lsn)
+			return true;
+
+		return false;
+	}
 }
 
 /*
  * Wait for physical standbys to confirm receiving the given lsn.
  *
  * Used by logical decoding SQL functions. It waits for physical standbys
- * corresponding to the physical slots specified in the synchronized_standby_slots GUC.
+ * corresponding to the physical slots specified in the synchronized_standby_slots GUC,
+ * or synchronous replication.
  */
 void
 WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
@@ -2766,7 +2805,7 @@ WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
 	 * slot is not a logical failover slot, or there is no value in
 	 * synchronized_standby_slots.
 	 */
-	if (!MyReplicationSlot->data.failover || !synchronized_standby_slots_config)
+	if (!MyReplicationSlot->data.failover || !(synchronized_standby_slots_config || SyncRepConfigured()))
 		return;
 
 	ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv);
diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c
index fa5988c824..69c061d8dc 100644
--- a/src/backend/replication/syncrep.c
+++ b/src/backend/replication/syncrep.c
@@ -78,6 +78,7 @@
 #include "common/int.h"
 #include "miscadmin.h"
 #include "pgstat.h"
+#include "replication/slot.h"
 #include "replication/syncrep.h"
 #include "replication/walsender.h"
 #include "replication/walsender_private.h"
@@ -95,7 +96,7 @@ char	   *SyncRepStandbyNames;
 static bool announce_next_takeover = true;
 
 SyncRepConfigData *SyncRepConfig = NULL;
-static int	SyncRepWaitMode = SYNC_REP_NO_WAIT;
+int	SyncRepWaitMode = SYNC_REP_NO_WAIT;
 
 static void SyncRepQueueInsert(int mode);
 static void SyncRepCancelWait(void);
@@ -124,6 +125,8 @@ static int	cmp_lsn(const void *a, const void *b);
 static bool SyncRepQueueIsOrderedByLSN(int mode);
 #endif
 
+bool SyncRepConfigured(void);
+
 /*
  * ===========================================================
  * Synchronous Replication functions for normal user backends
@@ -169,8 +172,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit)
 	 * described in SyncRepUpdateSyncStandbysDefined(). On the other hand, if
 	 * it's false, the lock is not necessary because we don't touch the queue.
 	 */
-	if (!SyncRepRequested() ||
-		!((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
+	if (!SyncRepConfigured())
 		return;
 
 	/* Cap the level for anything other than commit to remote flush only. */
@@ -523,6 +525,15 @@ SyncRepReleaseWaiters(void)
 
 	LWLockRelease(SyncRepLock);
 
+	/*
+	 * If synchronized_standby_slots is set, the respective walsender's
+	 * will be responsible for broadcasting the value.
+	 */
+	if (strcmp(synchronized_standby_slots, "") == 0)
+	{
+		ConditionVariableBroadcast(&WalSndCtl->wal_confirm_rcv_cv);
+	}
+
 	elog(DEBUG3, "released %d procs up to write %X/%X, %d procs up to flush %X/%X, %d procs up to apply %X/%X",
 		 numwrite, LSN_FORMAT_ARGS(writePtr),
 		 numflush, LSN_FORMAT_ARGS(flushPtr),
@@ -1069,3 +1080,13 @@ assign_synchronous_commit(int newval, void *extra)
 			break;
 	}
 }
+
+bool
+SyncRepConfigured()
+{
+	if (!SyncRepRequested() ||
+		!((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
+		return false;
+
+	return true;
+}
diff --git a/src/include/replication/syncrep.h b/src/include/replication/syncrep.h
index ea439e6da6..ec22bc72df 100644
--- a/src/include/replication/syncrep.h
+++ b/src/include/replication/syncrep.h
@@ -82,6 +82,7 @@ extern PGDLLIMPORT char *SyncRepStandbyNames;
 
 /* called by user backend */
 extern void SyncRepWaitForLSN(XLogRecPtr lsn, bool commit);
+extern bool SyncRepConfigured();
 
 /* called at backend exit */
 extern void SyncRepCleanupAtProcExit(void);
@@ -96,6 +97,8 @@ extern int	SyncRepGetCandidateStandbys(SyncRepStandbyData **standbys);
 /* called by checkpointer */
 extern void SyncRepUpdateSyncStandbysDefined(void);
 
+extern int	SyncRepWaitMode;
+
 /*
  * Internal functions for parsing synchronous_standby_names grammar,
  * in syncrep_gram.y and syncrep_scanner.l
diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl
index a474c626d9..5a69837f98 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -710,6 +710,141 @@ $result = $subscriber1->safe_psql('postgres',
 is($result, 't',
 	"subscriber1 gets data from primary after standby1 acknowledges changes");
 
+##################################################
+# Test that logical failover replication slots wait for the specified
+# synchronous replicas to receive the changes first. It uses the
+# following set up:
+#
+# 				(synchronous physical standbys)
+#				| ----> standby1 (application_name = standby1)
+#				| ----> standby2 (application_name = standby2)
+# primary -----	|
+#				(logical replication)
+#				| ----> subscriber1 (failover = true, slot_name = lsub1_slot)
+#				| ----> subscriber2 (failover = false, slot_name = lsub2_slot)
+#
+# synchronous_commit = 'on'
+# synchronous_standby_names = 'ANY 2 (standby1, standby2)'
+#
+# The setup is configured in such a way that the logical slot of subscriber1 is
+# enabled for failover, and thus the subscriber1 will wait for the changes to have
+# been synchronously replicated before receiving the decoded changes.
+##################################################
+
+$primary->safe_psql('postgres', "TRUNCATE tab_int;");
+# Setup synchronous replication
+$primary->append_conf(
+	'postgresql.conf', qq(
+		synchronous_commit = 'on'
+		synchronous_standby_names = 'ANY 2 (standby1, standby2)'
+));
+
+$primary->reload;
+
+$standby1->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby1'");
+$standby1->reload;
+
+$standby2->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby2'");
+$standby2->reload;
+
+# Check that synchronous replication is setup properly
+$standby2->stop;
+
+# Create some data on the primary
+$primary_row_count = 10;
+
+my $sync_back_q = $primary->background_psql(
+	'postgres',
+	on_error_stop => 0,
+	timeout => $PostgreSQL::Test::Utils::timeout_default);
+
+$sync_back_q->query_until(qr/insert_blocked_on_sync_rep/, q(
+    \echo insert_blocked_on_sync_rep
+   INSERT INTO tab_int SELECT generate_series(1, 10);
+));
+
+$result = $primary->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+
+is($result, 'f', "primary row count is not updated due to synchronous replication");
+
+# Verify the standby specified in synchronized_standby_slots (sb1_slot aka standby1)
+# catches up with the primary.
+$primary->wait_for_replay_catchup($standby1);
+$result = $standby1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't', "standby1 gets data from primary");
+
+# Validate that synchronized_standby_slots takes precedence over waiting for
+# changes to have been synchronous replicated.
+# Since the slot specified (sb1_slot) has received the changes, primary can send
+# the decoded changes to the subscription enabled for failover
+# (i.e. regress_mysub1).
+$primary->wait_for_catchup('regress_mysub1');
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber1 gets data from primary after standby1 acknowledges changes");
+
+$standby2->start;
+$primary->wait_for_replay_catchup($standby2);
+
+# Unset synchronized_standby_slots to test synchronous replication
+# blocks primary from sending logical decoded changes to failover slots until
+# changes have been synchronously replicated.
+$primary->append_conf(
+	'postgresql.conf', qq(
+	synchronized_standby_slots = ''));
+$primary->reload;
+
+$primary_row_count = 20;
+$standby2->stop;
+
+$sync_back_q->query_until(
+	qr/insert_blocked_on_additional_sync_rep/, q(
+	\echo insert_blocked_on_additional_sync_rep
+	INSERT INTO tab_int SELECT generate_series(11, 20);
+));
+
+# Since $standby2 has not received the changes, validate that subscriber1 (failover = true)
+# has not received the decoded changes, but subscriber2 (failover = false) has.
+$primary->wait_for_catchup('regress_mysub2');
+
+$result = $subscriber2->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber2 gets data from primary even if the changes have not been synchronously acknowledged.");
+
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 'f',
+	"subscriber1 does not get data from primary since changes have not been synchronously acknowledged.");
+
+# Start standby2 to allow the changes to be acknowledged by all the synchronous standbys.
+$standby2->start;
+$primary->wait_for_replay_catchup($standby2);
+$primary->wait_for_catchup('regress_mysub1');
+
+# Now that the changes have been replicated to all synchronous nodes,
+# primary can send the decoded changes to the subscription enabled for failover
+# (i.e. regress_mysub1). While the standby was down, regress_mysub1 didn't
+# receive any data from the primary. i.e. the primary didn't allow it to go
+# ahead of requirements of synchronous commit.
+$result = $subscriber1->safe_psql('postgres',
+	"SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't',
+	"subscriber1 gets data from primary since changes have been syhcronously acknowledged.");
+# No longer need to run the background session.
+$sync_back_q->quit;
+
+# Reset synchronized_standby_slots and synchronous commit for below test cases
+$primary->append_conf(
+	'postgresql.conf', qq(
+synchronized_standby_slots = 'sb1_slot'
+synchronous_standby_names = ''
+));
+$primary->reload;
+
 ##################################################
 # Verify that when using pg_logical_slot_get_changes to consume changes from a
 # logical failover slot, it will also wait for the slots specified in
-- 
2.40.1