replication_reserved_connections.patch

diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index d60026d..13caeef 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -2273,6 +2273,10 @@ InitWalSenderSlot(void)
 			walsnd->applyLag = -1;
 			walsnd->state = WALSNDSTATE_STARTUP;
 			walsnd->latch = &MyProc->procLatch;
+
+			/* increment the number of allocated wal sender slots */
+			pg_atomic_fetch_add_u32(&WalSndCtl->num_wal_senders, 1);
+
 			SpinLockRelease(&walsnd->mutex);
 			/* don't need the lock anymore */
 			MyWalSnd = (WalSnd *) walsnd;
@@ -2306,6 +2310,10 @@ WalSndKill(int code, Datum arg)
 	walsnd->latch = NULL;
 	/* Mark WalSnd struct as no longer being in use. */
 	walsnd->pid = 0;
+
+	/* decrement the number of allocated wal sender slots */
+	pg_atomic_fetch_sub_u32(&WalSndCtl->num_wal_senders, 1);
+
 	SpinLockRelease(&walsnd->mutex);
 }
 
@@ -3022,6 +3030,7 @@ WalSndShmemInit(void)
 	{
 		/* First time through, so initialize */
 		MemSet(WalSndCtl, 0, WalSndShmemSize());
+		pg_atomic_init_u32(&WalSndCtl->num_wal_senders, 0);
 
 		for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++)
 			SHMQueueInit(&(WalSndCtl->SyncRepQueue[i]));
@@ -3576,3 +3585,9 @@ LagTrackerRead(int head, XLogRecPtr lsn, TimestampTz now)
 	Assert(time != 0);
 	return now - time;
 }
+
+/* Return the amount of allocated wal_sender slots */
+uint32 NumWalSenders(void)
+{
+	return pg_atomic_read_u32(&WalSndCtl->num_wal_senders);
+}
diff --git a/src/backend/utils/init/postinit.c b/src/backend/utils/init/postinit.c
index 5ef6315..18392c1 100644
--- a/src/backend/utils/init/postinit.c
+++ b/src/backend/utils/init/postinit.c
@@ -789,17 +789,25 @@ InitPostgres(const char *in_dbname, Oid dboid, const char *username,
 	}
 
 	/*
-	 * The last few connection slots are reserved for superusers.  Although
-	 * replication connections currently require superuser privileges, we
-	 * don't allow them to consume the reserved slots, which are intended for
-	 * interactive use.
+	 * The last few connection slots are reserved for superusers and replication.
+	 * Superusers always have a priority over replication connections.
 	 */
-	if ((!am_superuser || am_walsender) &&
-		ReservedBackends > 0 &&
-		!HaveNFreeProcs(ReservedBackends))
-		ereport(FATAL,
-				(errcode(ERRCODE_TOO_MANY_CONNECTIONS),
-				 errmsg("remaining connection slots are reserved for non-replication superuser connections")));
+	if (am_walsender)
+	{
+		if (ReservedBackends > 0 && !HaveNFreeProcs(ReservedBackends))
+			ereport(FATAL,
+					(errcode(ERRCODE_TOO_MANY_CONNECTIONS),
+					 errmsg("remaining connection slots are reserved for non-replication superuser connections")));
+	}
+	else if (!am_superuser)
+	{
+		uint32	n = ReservedBackends + max_wal_senders - NumWalSenders();
+
+		if (n > 0 && !HaveNFreeProcs(n))
+			ereport(FATAL,
+					(errcode(ERRCODE_TOO_MANY_CONNECTIONS),
+					 errmsg("remaining connection slots are reserved for replication or superuser connections")));
+	}
 
 	/* Check replication permissions needed for walsender processes. */
 	if (am_walsender)
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index 45b72a7..9ebcb57 100644
--- a/src/include/replication/walsender.h
+++ b/src/include/replication/walsender.h
@@ -37,6 +37,7 @@ extern int	max_wal_senders;
 extern int	wal_sender_timeout;
 extern bool log_replication_commands;
 
+extern uint32 NumWalSenders(void);
 extern void InitWalSender(void);
 extern bool exec_replication_command(const char *query_string);
 extern void WalSndErrorCleanup(void);
diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h
index 4b90477..3ef8f4d 100644
--- a/src/include/replication/walsender_private.h
+++ b/src/include/replication/walsender_private.h
@@ -14,6 +14,7 @@
 
 #include "access/xlog.h"
 #include "nodes/nodes.h"
+#include "port/atomics.h"
 #include "replication/syncrep.h"
 #include "storage/latch.h"
 #include "storage/shmem.h"
@@ -101,6 +102,8 @@ typedef struct
 	 */
 	bool		sync_standbys_defined;
 
+	pg_atomic_uint32 num_wal_senders;
+
 	WalSnd		walsnds[FLEXIBLE_ARRAY_MEMBER];
 } WalSndCtlData;