0003-Add-FIRST-N-and-N-.-priority-syntax-to-synchronized_.patch

From d0755c482b0197b4e0fa874ffddb9ec36c1cf9ba Mon Sep 17 00:00:00 2001
From: Ashutosh Sharma <ashu.coek88@gmail.com>
Date: Wed, 13 May 2026 11:39:44 +0000
Subject: [PATCH 3/3] Add FIRST N and N (...) priority syntax to
 synchronized_standby_slots

Extend synchronized_standby_slots to support explicit priority
forms aligned with synchronous_standby_names.

- FIRST N (slot1, slot2, ...)
- N (slot1, slot2, ...) as shorthand for FIRST N

Implementation details:

- Use the SYNC_REP_DEFAULT parser distinction from the earlier
  refactor so plain-list syntax remains separate from priority
  syntax.
- Extend StandbySlotsHaveCaughtup() priority handling.
- Select slots in list order.
- Skip missing, logical, invalidated, and inactive lagging slots.
- Wait for active lagging higher-priority slots.

Tests and docs:

- Add coverage for FIRST behavior and shorthand N (...) behavior.
- Add plain-list disambiguation with first-prefixed slot names.
- Update docs for FIRST and shorthand priority syntax semantics.

Author: Satya Narlapuram <satyanarlapuram@gmail.com>
Author: Ashutosh Sharma <ashu.coek88@gmail.com>
Reviewed-by: Shveta Malik <shveta.malik@gmail.com>
Reviewed-by: Ajin Cherian <itsajin@gmail.com>
Reviewed-by: Hou, Zhijie <houzj.fnst@fujitsu.com>
Reviewed-by: Dilip Kumar <dilipbalaut@gmail.com>
Reviewed-by: Surya Poondla <suryapoondla4@gmail.com>
Reviewed-by: Japin Li <japinli@hotmail.com>
---
 doc/src/sgml/config.sgml                      |  31 ++-
 src/backend/replication/slot.c                |  41 ++--
 .../053_synchronized_standby_slots_quorum.pl  | 226 ++++++++++++++++--
 3 files changed, 262 insertions(+), 36 deletions(-)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index f0f5c89a069..a581fa50137 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -5194,6 +5194,7 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
         sender processes must wait on before delivering decoded changes. This
         parameter uses the following syntax:
 <synopsis>
+  [FIRST] <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="parameter">slot_name</replaceable> [, ...] )
 ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="parameter">slot_name</replaceable> [, ...] )
 <replaceable class="parameter">slot_name</replaceable> [, ...]
 </synopsis>
@@ -5205,9 +5206,24 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
         <replaceable class="parameter">num_sync</replaceable>
         must be an integer value greater than zero and must not exceed the
         number of listed slots.
-        Other forms supported by
-        <xref linkend="guc-synchronous-standby-names"/>, such as priority
-        syntax, are not supported.
+        </para>
+        <para>
+         The keyword <literal>FIRST</literal>, coupled with
+         <replaceable class="parameter">num_sync</replaceable>, specifies
+         priority-based semantics. Logical decoding will wait for the first
+         <replaceable class="parameter">num_sync</replaceable> available
+         physical slots in priority order (the order they appear in the list).
+         Missing, logical, or invalidated slots are skipped. Inactive slots are
+         skipped only while they are lagging. However, if a slot exists and is
+         valid and active but has not yet caught up, the system will wait for it
+         rather than skipping to lower-priority slots. If, after skipping
+         unusable slots, fewer than
+         <replaceable class="parameter">num_sync</replaceable> usable slots
+         remain, logical decoding waits until enough slots become usable and
+         caught up, or until the configuration is changed. The keyword
+         <literal>FIRST</literal> is optional in this form, so
+         <literal>2 (slot1, slot2, slot3)</literal> and
+         <literal>FIRST 2 (slot1, slot2, slot3)</literal> are equivalent.
        </para>
        <para>
         A plain comma-separated list without a keyword specifies that
@@ -5239,9 +5255,12 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
         (<literal>synchronous_standby_names = 'ANY ...'</literal>), so that
         logical decoding availability matches the commit durability guarantee.
        </para>
-      <para>
-       <literal>ANY</literal> is case-insensitive.
-      </para>
+        <para>
+         <literal>FIRST</literal> and <literal>ANY</literal> are case-insensitive.
+         If these keywords are used as the name of a replication slot,
+         the <replaceable class="parameter">slot_name</replaceable> must
+         be double-quoted.
+        </para>
        <para>
         The use of <varname>synchronized_standby_slots</varname> guarantees
         that logical replication
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 9a460c992d9..a4dda6aca9e 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -3005,6 +3005,8 @@ GetSlotInvalidationCauseName(ReplicationSlotInvalidationCause cause)
  *
  *   slot1, slot2                   -- wait for ALL listed slots
  *   ANY N (slot1, slot2, ...)      -- wait for any N-of-M (quorum)
+ *   FIRST N (slot1, slot2, ...)    -- wait for first N in priority order
+ *   N (slot1, slot2, ...)          -- shorthand for FIRST N
  *
  * Note: Simple list syntax is interpreted as "wait for ALL" for this GUC,
  * unlike synchronous_standby_names where it means "FIRST 1".
@@ -3044,14 +3046,6 @@ check_synchronized_standby_slots(char **newval, void **extra, GucSource source)
 			return false;
 		}
 
-		if (syncrep_parse_result->syncrep_method == SYNC_REP_PRIORITY)
-		{
-			GUC_check_errcode(ERRCODE_INVALID_PARAMETER_VALUE);
-			GUC_check_errmsg("priority syntax is not supported for parameter \"%s\"",
-							 "synchronized_standby_slots");
-			return false;
-		}
-
 		if (syncrep_parse_result->num_sync <= 0)
 		{
 			GUC_check_errmsg("number of synchronized standby slots (%d) must be greater than zero",
@@ -3266,6 +3260,12 @@ ReportUnavailableSyncStandbySlots(SyncStandbySlotsStateInfo *slot_states,
  *   Simple list (e.g., "slot1, slot2"):
  *     ALL slots must have caught up. Returns false otherwise.
  *
+ *   FIRST N (e.g., "FIRST 2 (slot1, slot2, slot3)"):
+ *     Wait for the first N eligible slots in priority order. Skips missing,
+ *     invalid, logical, and inactive-lagging slots to find N eligible slots.
+ *     If an active slot is lagging, waits for it (does not skip to lower
+ *     priority slots).
+ *
  *   ANY N (e.g., "ANY 2 (slot1, slot2, slot3)"):
  *     Wait for any N eligible slots. Skips missing, invalid, logical, and
  *     lagging slots (inactive or active) to find N slots that have caught up.
@@ -3316,12 +3316,16 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 	 * on the first slot that is missing/invalid/logical, or the first slot
 	 * that is lagging (inactive or active).
 	 *
-	 * wait_for_all = false means we select N from M candidates (ANY N syntax).
-	 * In this mode, slots already caught up are counted even if inactive, and
-	 * lagging slots are skipped until enough slots have caught up.
+	 * wait_for_all = false means we select N from M candidates (FIRST N or
+	 * ANY N syntax). In this mode, slots already caught up are counted even if
+	 * inactive. In FIRST N mode, we skip missing/invalid/logical slots and
+	 * lagging inactive slots, but wait for an active lagging slot with higher
+	 * priority. In ANY N mode, we skip lagging slots (inactive or active) to
+	 * find any N that have caught up.
 	 */
 	required = synchronized_standby_slots_config->num_sync;
-	wait_for_all = (required == synchronized_standby_slots_config->nslotnames);
+	wait_for_all =
+		(synchronized_standby_slots_config->syncrep_method == SYNC_REP_DEFAULT);
 
 	/*
 	 * Allocate array to track slot states. Size it to the total number of
@@ -3402,6 +3406,8 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 			 * If it is active and lagging, report it as lagging.
 			 *
 			 * In ALL mode: must wait for it.
+			 * In FIRST N (priority) mode: lagging active slots block, while
+			 * inactive slots can be skipped.
 			 * In ANY N (quorum) mode: skip and use another slot.
 			 */
 			slot_states[num_slot_states].slot_name = name;
@@ -3411,7 +3417,9 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 				slot_states[num_slot_states].restart_lsn = restart_lsn;
 			num_slot_states++;
 
-			if (wait_for_all)
+			if (wait_for_all ||
+				(!inactive &&
+				 synchronized_standby_slots_config->syncrep_method == SYNC_REP_PRIORITY))
 				break;
 			goto next_slot;
 		}
@@ -3424,7 +3432,12 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 
 		caught_up_slot_num++;
 
-		/* Stop processing if the required number of slots have caught up. */
+		/*
+		 * Stop processing if the required number of slots have caught up. In
+		 * priority mode (FIRST N), this ensures we select the first N slots in
+		 * sequential order. In quorum mode (ANY N), we still process in order for
+		 * efficiency, stopping once we find any N.
+		 */
 		if (caught_up_slot_num >= required)
 			break;
 
diff --git a/src/test/recovery/t/053_synchronized_standby_slots_quorum.pl b/src/test/recovery/t/053_synchronized_standby_slots_quorum.pl
index 1279b540b8b..1cf8899cb9a 100644
--- a/src/test/recovery/t/053_synchronized_standby_slots_quorum.pl
+++ b/src/test/recovery/t/053_synchronized_standby_slots_quorum.pl
@@ -4,6 +4,7 @@
 # Test synchronized_standby_slots with different syntax modes:
 # - Plain list (ALL mode): slot1, slot2
 # - ANY N (quorum mode): ANY N (slot1, slot2, ...)
+# - FIRST N (priority mode): FIRST N (slot1, slot2, ...)
 #
 # Setup: a 3-node cluster with one primary, two physical standbys, and a
 # logical decoding client using a failover-enabled slot.
@@ -25,6 +26,12 @@
 #    - Skips missing/invalid/logical slots and lagging slots (inactive or active)
 #      to find N caught-up slots
 #
+# C) FIRST N (sb1_slot, sb2_slot) (priority mode)
+#    - Selects first N slots in priority order (list order)
+#    - Skips missing/invalid/logical slots and inactive lagging slots,
+#      but waits for active lagging slots
+#    - FIRST 1 works with one slot down (unlike plain list)
+
 use strict;
 use warnings FATAL => 'all';
 use PostgreSQL::Test::Cluster;
@@ -211,16 +218,168 @@ is($decoded_bc, '1',
 	'plain list: works when all standbys are up');
 
 ##################################################
-# PART D: ANY 2 waits on an active lagging slot
+# PART D: Verify FIRST N priority semantics
+##################################################
+
+# FIRST N should:
+# 1. Select first N slots in priority order (list order)
+# 2. Skip missing/invalid/logical slots and inactive lagging slots to find
+#    N caught-up slots
+# 3. Wait for active lagging slots (not skip to lower priority)
+
+# Test FIRST 2 (sb1_slot, sb2_slot) with both up; should wait for both.
+$primary->adjust_conf('postgresql.conf', 'synchronized_standby_slots',
+	"'FIRST 2 (sb1_slot, sb2_slot)'");
+$primary->reload;
+
+$primary->safe_psql('postgres',
+	"SELECT pg_logical_emit_message(true, 'qtest', 'first_2_both_up');"
+);
+$primary->wait_for_replay_catchup($standby1);
+$primary->wait_for_replay_catchup($standby2);
+
+my $decoded_e2 = $primary->safe_psql('postgres',
+	q{SELECT count(*) FROM pg_logical_slot_get_changes('logical_failover', NULL, NULL)
+	  WHERE data LIKE '%first_2_both_up%';});
+is($decoded_e2, '1',
+	'FIRST 2: decoding works when all required slots are up');
+
+# Test FIRST 1 (sb1_slot, sb2_slot) with sb1_slot unavailable.
+$standby1->stop;
+
+$primary->adjust_conf('postgresql.conf', 'synchronized_standby_slots',
+	"'FIRST 1 (sb1_slot, sb2_slot)'");
+$primary->reload;
+
+$primary->safe_psql('postgres',
+	"SELECT pg_logical_emit_message(true, 'qtest', 'first_1_skip_unavailable');"
+);
+$primary->wait_for_replay_catchup($standby2);
+
+# FIRST 1 should skip sb1_slot (unavailable) and use sb2_slot.
+my $decoded_e1 = $primary->safe_psql('postgres',
+	q{SELECT count(*) FROM pg_logical_slot_get_changes('logical_failover', NULL, NULL)
+	  WHERE data LIKE '%first_1_skip_unavailable%';});
+is($decoded_e1, '1',
+	'FIRST 1: skips unavailable first slot, uses second slot');
+
+# Test shorthand priority syntax: N (...) means FIRST N (...).
+$primary->adjust_conf('postgresql.conf', 'synchronized_standby_slots',
+	"'1 (sb1_slot, sb2_slot)'");
+$primary->reload;
+
+$primary->safe_psql('postgres',
+	"SELECT pg_logical_emit_message(true, 'qtest', 'num_1_shorthand_priority');"
+);
+$primary->wait_for_replay_catchup($standby2);
+
+my $decoded_num1 = $primary->safe_psql('postgres',
+	q{SELECT count(*) FROM pg_logical_slot_get_changes('logical_failover', NULL, NULL)
+	  WHERE data LIKE '%num_1_shorthand_priority%';});
+is($decoded_num1, '1',
+	'1 (...): shorthand priority syntax behaves like FIRST 1');
+
+##################################################
+# PART E: FIRST 1 and ANY 2 wait on an active lagging slot
 ##################################################
 
-# Stop standby1 so sb1_slot can be controlled by a raw replication connection
-# that keeps the slot active while lagging.
+# Bring standby1 back so sb1_slot is active and caught up.
+$standby1->start;
+$primary->wait_for_replay_catchup($standby1);
+
+# To test the active-but-lagging slot path deterministically, we open a raw
+# replication connection to sb1_slot starting from a deliberately old LSN.
+# psql in replication mode never sends Standby Status Update messages, so
+# the walsender keeps sb1_slot's active_pid set but restart_lsn never
+# advances.
+
+# Stop standby1 so its walsender releases sb1_slot, allowing our replication
+# connection below to acquire it.
 $standby1->stop;
 
+# Capture a safely old LSN to stream from, before the test WAL record.
 my $old_lsn = $primary->safe_psql('postgres',
 	"SELECT pg_current_wal_lsn();");
 
+# FIRST 1 must wait for the highest-priority slot when it is active but lagging.
+$primary->adjust_conf('postgresql.conf', 'synchronized_standby_slots',
+	"'FIRST 1 (sb1_slot, sb2_slot)'");
+$primary->reload;
+
+my $first_lag_lsn = $primary->safe_psql('postgres',
+	"SELECT pg_logical_emit_message(true, 'qtest', 'first_1_lagging_blocks');"
+);
+$primary->wait_for_replay_catchup($standby2);
+
+# Open a raw replication connection to sb1_slot starting from $old_lsn.
+# This activates the slot (active_pid IS NOT NULL) while keeping restart_lsn
+# frozen below $first_lag_lsn for the lifetime of the connection.
+my $repl_first = $primary->background_psql(
+	'postgres',
+	replication => 'database',
+	on_error_stop => 0,
+	timeout => $PostgreSQL::Test::Utils::timeout_default);
+
+$repl_first->query_until(
+	qr/^$/,
+	"START_REPLICATION SLOT sb1_slot PHYSICAL $old_lsn;\n");
+
+# Wait until sb1_slot shows active_pid, confirming the walsender is live.
+$primary->poll_query_until('postgres', q{
+	SELECT active_pid IS NOT NULL
+	FROM pg_replication_slots
+	WHERE slot_name = 'sb1_slot'
+}) or die "replication connection did not activate sb1_slot";
+
+# sb1_slot is now active and its restart_lsn is behind $first_lag_lsn.
+# Start logical decoding in the background; it must block.
+my $bg_first = $primary->background_psql(
+	'postgres',
+	on_error_stop => 0,
+	timeout => $PostgreSQL::Test::Utils::timeout_default);
+
+$bg_first->query_until(
+	qr/decode_start/, q(
+   \echo decode_start
+   SELECT pg_logical_slot_peek_changes('logical_failover', NULL, NULL);
+));
+
+ok( $primary->poll_query_until(
+		'postgres', q{
+SELECT EXISTS (
+	SELECT 1
+	FROM pg_stat_activity
+	WHERE wait_event = 'WaitForStandbyConfirmation'
+	  AND query LIKE '%pg_logical_slot_peek_changes(''logical_failover''%'
+);
+}),
+	'FIRST 1: decoding waits for active lagging higher-priority slot');
+
+$primary->adjust_conf('postgresql.conf', 'synchronized_standby_slots', "''");
+$primary->reload;
+$bg_first->quit;
+$repl_first->quit;
+
+# Ensure the previous replication connection has fully released sb1_slot
+# before reusing it in the next subtest.
+$primary->poll_query_until('postgres', q{
+	SELECT active_pid IS NULL
+	FROM pg_replication_slots
+	WHERE slot_name = 'sb1_slot'
+}) or die "replication connection did not release sb1_slot";
+
+# Consume the change so the slot is clean for the next test.
+$primary->safe_psql('postgres',
+	q{SELECT pg_logical_slot_get_changes('logical_failover', NULL, NULL);});
+
+# ANY 2 must also wait when only one of two required slots has caught up.
+# Reuse the same technique: open a raw replication connection to sb1_slot
+# from $old_lsn so it is active but its restart_lsn stays behind the target.
+
+# Capture another old LSN baseline before the next test WAL record.
+$old_lsn = $primary->safe_psql('postgres',
+	"SELECT pg_current_wal_lsn();");
+
 $primary->adjust_conf('postgresql.conf', 'synchronized_standby_slots',
 	"'ANY 2 (sb1_slot, sb2_slot)'");
 $primary->reload;
@@ -283,7 +442,54 @@ $primary->wait_for_replay_catchup($standby1);
 
 
 ##################################################
-# PART E: Verify GUC validation rejects bad values
+# PART F: Plain list with first-prefixed slot name still means ALL mode
+##################################################
+
+# Create a slot name starting with "first_" for parser disambiguation checks.
+$primary->safe_psql('postgres',
+	"SELECT pg_create_physical_replication_slot('first_slot');");
+
+# If simple-list syntax starts with a slot name like "first_slot", it must
+# still be treated as ALL mode (not as explicit FIRST N syntax).
+$primary->adjust_conf('postgresql.conf', 'synchronized_standby_slots',
+	"'first_slot, sb2_slot'");
+$primary->reload;
+
+$primary->safe_psql('postgres',
+	"SELECT pg_logical_emit_message(true, 'qtest', 'first_prefix_all_mode_blocks');"
+);
+$primary->wait_for_replay_catchup($standby2);
+
+$log_offset = -s $primary->logfile;
+
+$bg = $primary->background_psql(
+	'postgres',
+	on_error_stop => 0,
+	timeout => $PostgreSQL::Test::Utils::timeout_default);
+
+$bg->query_until(
+	qr/decode_start/, q(
+   \echo decode_start
+   SELECT pg_logical_slot_peek_changes('logical_failover', NULL, NULL);
+));
+
+# Plain list must require all listed slots; first_slot is intentionally inactive.
+$primary->wait_for_log(
+	qr/replication slot \"first_slot\" specified in parameter "synchronized_standby_slots" does not have active_pid/,
+	$log_offset);
+
+pass('plain list with first-prefixed slot name blocks in ALL mode');
+
+$primary->adjust_conf('postgresql.conf', 'synchronized_standby_slots', "''");
+$primary->reload;
+$bg->quit;
+
+# Consume the change for the next test.
+$primary->safe_psql('postgres',
+	q{SELECT pg_logical_slot_get_changes('logical_failover', NULL, NULL);});
+
+##################################################
+# PART G: Verify GUC validation rejects bad values
 ##################################################
 
 my ($result, $stdout, $stderr);
@@ -300,18 +506,6 @@ like($stderr, qr/ERROR/,
 like($stderr, qr/ERROR/,
 	'GUC rejects malformed ANY syntax');
 
-# Priority syntax is not supported by synchronized_standby_slots yet
-($result, $stdout, $stderr) = $primary->psql('postgres',
-	"ALTER SYSTEM SET synchronized_standby_slots = 'FIRST 1 (sb1_slot, sb2_slot)';");
-like($stderr, qr/priority syntax is not supported/,
-	'GUC rejects FIRST syntax');
-
-# Legacy priority syntax is not supported by synchronized_standby_slots yet
-($result, $stdout, $stderr) = $primary->psql('postgres',
-	"ALTER SYSTEM SET synchronized_standby_slots = '1 (sb1_slot, sb2_slot)';");
-like($stderr, qr/priority syntax is not supported/,
-	'GUC rejects legacy priority syntax');
-
 # Invalid slot name
 ($result, $stdout, $stderr) = $primary->psql('postgres',
 	"ALTER SYSTEM SET synchronized_standby_slots = 'ANY 1 (INVALID_UPPER)';");
-- 
2.43.0