Add general check function "check_replication_slots_available()"

Make the code previously only used by "standby follow" generally
available - we'll want to use this from "node rejoin" as well.

While we're at it, when reporting failure due to lack of free
replication slots, report the current value of "max_replication_slots".

dbutils.c

@@ -4398,7 +4398,7 @@ get_slot_record(PGconn *conn, char *slot_name, t_replication_slot *record)
 
 
 int
-get_free_replication_slot_count(PGconn *conn)
+get_free_replication_slot_count(PGconn *conn, int *max_replication_slots)
 {
 	PQExpBufferData query;
 	PGresult   *res = NULL;
@@ -4408,7 +4408,10 @@ get_free_replication_slot_count(PGconn *conn)
 
 	appendPQExpBufferStr(&query,
 						 " SELECT pg_catalog.current_setting('max_replication_slots')::INT - "
-						 "        pg_catalog.count(*) AS free_slots"
+						 "          pg_catalog.count(*) "
+						 "          AS free_slots, "
+						 "        pg_catalog.current_setting('max_replication_slots')::INT "
+						 "          AS max_replication_slots "
 						 "   FROM pg_catalog.pg_replication_slots s"
 						 "  WHERE s.slot_type = 'physical'");
 
@@ -4419,15 +4422,17 @@ get_free_replication_slot_count(PGconn *conn)
 		log_db_error(conn, query.data,
 					 _("get_free_replication_slot_count(): unable to execute replication slot query"));
 
-		free_slots = -1;
+		free_slots = UNKNOWN_VALUE;
 	}
 	else if (PQntuples(res) == 0)
 	{
-		free_slots = -1;
+		free_slots = UNKNOWN_VALUE;
 	}
 	else
 	{
 		free_slots = atoi(PQgetvalue(res, 0, 0));
+		if (max_replication_slots != NULL)
+			*max_replication_slots = atoi(PQgetvalue(res, 0, 1));
 	}
 
 	termPQExpBuffer(&query);

dbutils.h

@@ -523,7 +523,7 @@ bool		drop_replication_slot_sql(PGconn *conn, char *slot_name);
 bool		drop_replication_slot_replprot(PGconn *repl_conn, char *slot_name);
 
 RecordStatus get_slot_record(PGconn *conn, char *slot_name, t_replication_slot *record);
-int			get_free_replication_slot_count(PGconn *conn);
+int			get_free_replication_slot_count(PGconn *conn, int *max_replication_slots);
 int			get_inactive_replication_slots(PGconn *conn, KeyValueList *list);
 
 /* tablespace functions */

repmgr-action-standby.c

@@ -2824,31 +2824,14 @@ do_standby_follow(void)
 
 	if (config_file_options.use_replication_slots)
 	{
-		int free_slots = get_free_replication_slot_count(follow_target_conn);
+		bool slots_available = check_replication_slots_available(follow_target_node_id,
-		if (free_slots < 0)
+																 follow_target_conn);
+		if (slots_available == false)
 		{
-			log_error(_("unable to determine number of free replication slots on node %i"),
-					  follow_target_node_id);
 			PQfinish(follow_target_conn);
 			PQfinish(local_conn);
 			exit(ERR_FOLLOW_FAIL);
 		}
-
-		if (free_slots == 0)
-		{
-			log_error(_("no free replication slots available on node %i"), follow_target_node_id);
-			log_hint(_("consider increasing \"max_replication_slots\""));
-			PQfinish(follow_target_conn);
-			PQfinish(local_conn);
-			exit(ERR_FOLLOW_FAIL);
-		}
-		else if (runtime_options.dry_run == true)
-		{
-			log_info(_("replication slots in use, %i free slots on node %i"),
-					 follow_target_node_id,
-					 free_slots);
-		}
-
 	}
 
 	/* XXX check this is not current upstream anyway */

repmgr-client-global.h

@@ -269,6 +269,7 @@ extern bool can_use_pg_rewind(PGconn *conn, const char *data_directory, PQExpBuf
 extern bool create_replication_slot(PGconn *conn, char *slot_name, t_node_info *upstream_node_record, PQExpBufferData *error_msg);
 extern bool drop_replication_slot_if_exists(PGconn *conn, int node_id, char *slot_name);
 
+extern bool check_replication_slots_available(int node_id, PGconn* conn);
 extern bool check_node_can_attach(TimeLineID local_tli, XLogRecPtr local_xlogpos, PGconn *follow_target_conn, t_node_info *follow_target_node_record, bool is_rejoin);
 extern void check_shared_library(PGconn *conn);
 extern bool is_repmgrd_running(PGconn *conn);

repmgr-client.c

@@ -3833,6 +3833,39 @@ _determine_replication_slot_user(PGconn *conn, t_node_info *upstream_node_record
 	}
 }
 
+
+bool
+check_replication_slots_available(int node_id, PGconn* conn)
+{
+	int max_replication_slots = UNKNOWN_VALUE;
+	int free_slots = get_free_replication_slot_count(conn, &max_replication_slots);
+
+	if (free_slots < 0)
+	{
+		log_error(_("unable to determine number of free replication slots on node %i"),
+				  node_id);
+		return false;
+	}
+
+	if (free_slots == 0)
+	{
+		log_error(_("no free replication slots available on node %i"),
+				  node_id);
+		log_hint(_("consider increasing \"max_replication_slots\" (current value: %i)"),
+				 max_replication_slots);
+		return false;
+	}
+	else if (runtime_options.dry_run == true)
+	{
+		log_info(_("replication slots in use, %i free slots on node %i"),
+				 node_id,
+				 free_slots);
+	}
+
+	return true;
+}
+
+
 /*
  * Here we'll perform some timeline sanity checks to ensure the follow target
  * can actually be followed.

repmgr.h

@@ -86,6 +86,7 @@
 #define UNKNOWN_SYSTEM_IDENTIFIER 0
 #define UNKNOWN_PID			-1
 #define UNKNOWN_REPLICATION_LAG	-1
+#define UNKNOWN_VALUE		-1
 
 #define NODE_NOT_FOUND		-1
 #define NO_UPSTREAM_NODE	-1