Separate catalog_xmin from xmin via ephemeral physical slot

When a standby connects without a replication slot and sends hot standby
feedback, lazily create an ephemeral physical slot so xmin and
catalog_xmin can be tracked separately and atomically (under the slot
mutex), instead of conflating both into the walsender's PGPROC xmin.

Add an error_if_full flag to ReplicationSlotCreate so the walsender can
degrade gracefully to the legacy min(xmin, catalog_xmin) behavior when the
slot pool is exhausted, rather than tearing down the connection.

Author: markrui

src/backend/commands/repack_worker.c

@@ -220,7 +220,7 @@ repack_setup_logical_decoding(Oid relid)
 	 */
 	snprintf(slotname, NAMEDATALEN, "pg_repack_%d", MyProcPid);
 	ReplicationSlotCreate(slotname, true, RS_TEMPORARY, false, true,
-						  false, false);
+						  false, false, true);
 	EnsureLogicalDecodingEnabled();
 
 	/*

src/backend/replication/logical/launcher.c

@@ -1576,7 +1576,7 @@ CreateConflictDetectionSlot(void)
 			errmsg("creating replication conflict detection slot"));
 
 	ReplicationSlotCreate(CONFLICT_DETECTION_SLOT, false, RS_PERSISTENT, false,
-						  false, false, false);
+						  false, false, false, true);
 
 	init_conflict_slot_xmin();
 }

src/backend/replication/logical/slotsync.c

@@ -899,7 +899,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid,
 							  remote_slot->two_phase,
 							  false,
 							  remote_slot->failover,
-							  true);
+							  true, true);
 
 		/* For shorter lines. */
 		slot = MyReplicationSlot;

src/backend/replication/slot.c

@@ -373,11 +373,17 @@ IsSlotForConflictCheck(const char *name)
  * failover: If enabled, allows the slot to be synced to standbys so
  *     that logical replication can be resumed after failover.
  * synced: True if the slot is synchronized from the primary server.
+ * error_if_full: If true, raise an error when no slot is free; if false,
+ *     return false instead so the caller can degrade gracefully.
+ *
+ * Returns true if a slot was created, false only when error_if_full is false
+ * and the slot pool is exhausted.
  */
-void
+bool
 ReplicationSlotCreate(const char *name, bool db_specific,
 					  ReplicationSlotPersistency persistency,
-					  bool two_phase, bool repack, bool failover, bool synced)
+					  bool two_phase, bool repack, bool failover, bool synced,
+					  bool error_if_full)
 {
 	ReplicationSlot *slot = NULL;
 	int			startpoint,
@@ -458,11 +464,18 @@ ReplicationSlotCreate(const char *name, bool db_specific,
 
 	/* If all slots are in use, we're out of luck. */
 	if (slot == NULL)
+	{
+		if (!error_if_full)
+		{
+			LWLockRelease(ReplicationSlotAllocationLock);
+			return false;
+		}
 		ereport(ERROR,
 				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
 				 errmsg("all replication slots are in use"),
 				 errhint("Free one or increase \"%s\".",
 						 repack ? "max_repack_replication_slots" : "max_replication_slots")));
+	}
 
 	/*
 	 * Since this slot is not in use, nobody should be looking at any part of
@@ -539,6 +552,8 @@ ReplicationSlotCreate(const char *name, bool db_specific,
 
 	/* Let everybody know we've modified this slot */
 	ConditionVariableBroadcast(&slot->active_cv);
+
+	return true;
 }
 
 /*

src/backend/replication/slotfuncs.c

@@ -53,7 +53,7 @@ create_physical_replication_slot(char *name, bool immediately_reserve,
 	/* acquire replication slot, this will check for conflicting names */
 	ReplicationSlotCreate(name, false,
 						  temporary ? RS_TEMPORARY : RS_PERSISTENT, false,
-						  false, false, false);
+						  false, false, false, true);
 
 	if (immediately_reserve)
 	{
@@ -146,7 +146,7 @@ create_logical_replication_slot(char *name, char *plugin,
 	 */
 	ReplicationSlotCreate(name, true,
 						  temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase,
-						  false, failover, false);
+						  false, failover, false, true);
 
 	/*
 	 * Ensure the logical decoding is enabled before initializing the logical

src/backend/replication/walsender.c

@@ -1286,7 +1286,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 	{
 		ReplicationSlotCreate(cmd->slotname, false,
 							  cmd->temporary ? RS_TEMPORARY : RS_PERSISTENT,
-							  false, false, false, false);
+							  false, false, false, false, true);
 
 		if (reserve_wal)
 		{
@@ -1317,7 +1317,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 		 */
 		ReplicationSlotCreate(cmd->slotname, true,
 							  cmd->temporary ? RS_TEMPORARY : RS_EPHEMERAL,
-							  two_phase, false, failover, false);
+							  two_phase, false, failover, false, true);
 
 		/*
 		 * Do options check early so that we can bail before calling the
@@ -2826,18 +2826,33 @@ ProcessStandbyHSFeedbackMessage(void)
 	 * obviously safe, and if we're moving it backwards, well, the data is at
 	 * risk already since a VACUUM could already have determined the horizon.)
 	 *
-	 * If we're using a replication slot we reserve the xmin via that,
-	 * otherwise via the walsender's PGPROC entry. We can only track the
-	 * catalog xmin separately when using a slot, so we store the least of the
-	 * two provided when not using a slot.
-	 *
-	 * XXX: It might make sense to generalize the ephemeral slot concept and
-	 * always use the slot mechanism to handle the feedback xmin.
+	 * If we're using a replication slot we reserve the xmin via that.  When the
+	 * standby connected without one, lazily create an ephemeral physical slot
+	 * here, so that we can still track xmin and catalog_xmin separately (and
+	 * atomically, under the slot mutex).  The ephemeral slot is dropped
+	 * automatically when this walsender exits.
 	 */
-	if (MyReplicationSlot != NULL)	/* XXX: persistency configurable? */
+	if (MyReplicationSlot == NULL)
+	{
+		char		slotname[NAMEDATALEN];
+
+		snprintf(slotname, sizeof(slotname), "pg_walsender_%d", MyProcPid);
+		ReplicationSlotCreate(slotname, false, RS_EPHEMERAL,
+							  false, false, false, false, false);
+	}
+
+	if (MyReplicationSlot != NULL)
 		PhysicalReplicationSlotNewXmin(feedbackXmin, feedbackCatalogXmin);
 	else
 	{
+		/*
+		 * The slot pool is exhausted, so we cannot track the two horizons
+		 * separately.  Degrade gracefully to the pre-existing behavior of
+		 * holding back both via the walsender's PGPROC entry, using the
+		 * older of the two values.  This loses the catalog/data separation
+		 * for this standby until a slot frees up, but never breaks the
+		 * connection.
+		 */
 		if (TransactionIdIsNormal(feedbackCatalogXmin)
 			&& TransactionIdPrecedes(feedbackCatalogXmin, feedbackXmin))
 			MyProc->xmin = feedbackCatalogXmin;

src/include/replication/slot.h

@@ -329,10 +329,10 @@ extern PGDLLIMPORT char *synchronized_standby_slots;
 extern PGDLLIMPORT int idle_replication_slot_timeout_secs;
 
 /* management of individual slots */
-extern void ReplicationSlotCreate(const char *name, bool db_specific,
+extern bool ReplicationSlotCreate(const char *name, bool db_specific,
 								  ReplicationSlotPersistency persistency,
 								  bool two_phase, bool repack, bool failover,
-								  bool synced);
+								  bool synced, bool error_if_full);
 extern void ReplicationSlotPersist(void);
 extern void ReplicationSlotDrop(const char *name, bool nowait);
 extern void ReplicationSlotDropAcquired(bool try_disable);