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.
- Clarify duplicate handling for priority syntax in the
  synchronized_standby_slots documentation.
- Simplify caught-up comments and clarify standby confirmation
  wait comments to match the final control flow.

Tests and docs:

- Add coverage for FIRST behavior and shorthand N (...) behavior.
- Add plain-list disambiguation with first-prefixed slot names.
- Add FIRST duplicate-entry recovery coverage to show duplicates
  do not create extra priority positions.
- Update docs for FIRST and shorthand priority syntax semantics.
- Clarify that duplicate slot names are ignored in priority-based
  forms and preserve first-occurrence order.

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>
Reviewed-by: Shlok Kyal <shlok.kyal.oss@gmail.com>

Author: ashucoek

doc/src/sgml/config.sgml

@@ -5274,6 +5274,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>
@@ -5285,9 +5286,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
@@ -5324,19 +5340,26 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
          duplicate entries are ignored and only the first occurrence is used.
          The semantics of <varname>synchronized_standby_slots</varname> are
          therefore based on the unique set of listed slot names, preserving the
-         original order of first occurrence. This means that
-         <literal>ANY 2 (slot1, slot1, slot2, slot3)</literal> is treated the
-         same as <literal>ANY 2 (slot1, slot2, slot3)</literal>, and a plain
-         list such as <literal>(slot1, slot1, slot2)</literal> is treated the
-         same as <literal>(slot1, slot2)</literal>. In particular,
+         original order of first occurrence. This means that, in
+         priority-based forms, duplicates do not create additional priority
+         positions: for example,
+         <literal>FIRST 2 (slot1, slot1, slot2, slot3)</literal> is treated the
+         same as <literal>FIRST 2 (slot1, slot2, slot3)</literal>.
+         Likewise, <literal>ANY 2 (slot1, slot1, slot2, slot3)</literal> is
+         treated the same as <literal>ANY 2 (slot1, slot2, slot3)</literal>,
+         and a plain list such as <literal>(slot1, slot1, slot2)</literal>
+         is treated the same as <literal>(slot1, slot2)</literal>. In particular,
          <replaceable class="parameter">num_sync</replaceable> must not exceed
          the number of unique listed slots. Such a configuration results in an
          error to prevent indefinite waits in WAL sender processes due to a
          misconfigured <varname>synchronized_standby_slots</varname> setting.
         </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

src/backend/replication/slot.c

@@ -3070,6 +3070,8 @@ CompactSyncRepConfigMemberNames(SyncRepConfigData *config)
  *
  *   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".
@@ -3110,14 +3112,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_errcode(ERRCODE_INVALID_PARAMETER_VALUE);
@@ -3339,6 +3333,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.
@@ -3389,11 +3389,14 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 	 * 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.
-	 * Duplicate configured slot names do not appear here because the check hook
-	 * compacts them out of the parsed configuration.
+	 * 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. Duplicate configured slot names do not
+	 * appear here because the check hook compacts them out of the parsed
+	 * configuration.
 	 */
 	required = synchronized_standby_slots_config->num_sync;
 	wait_for_all = (synchronized_standby_slots_config->syncrep_method == SYNC_REP_DEFAULT);
@@ -3476,16 +3479,19 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 			 * If a slot is inactive and lagging, report it as inactive. If it
 			 * is active and lagging, report it as lagging.
 			 *
-			 * In ALL mode: must wait for it. In ANY N (quorum) mode: skip and
-			 * use another slot.
+			 * 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;
 			slot_states[num_slot_states].state =
 				inactive ? SS_SLOT_INACTIVE_LAGGING : SS_SLOT_ACTIVE_LAGGING;
 			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;
 		}
@@ -3498,7 +3504,7 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 
 		caught_up_slot_num++;
 
-		/* Stop processing if the required number of slots have caught up. */
+		/* Stop once the required number of slots have caught up. */
 		if (caught_up_slot_num >= required)
 			break;
 
@@ -3513,8 +3519,8 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
 	 * problem states and return false.
 	 *
 	 * We only emit messages when the requirement is not met to avoid
-	 * misleading messages in quorum mode where other slots may have satisfied
-	 * the condition despite some slots having issues.
+	 * misleading messages in quorum/priority mode where other slots may have
+	 * satisfied the condition despite some slots having issues.
 	 */
 	if (caught_up_slot_num < required)
 	{