Monitor: detect and recover from fast_forward candidate stuck with all WAL sources unhealthy

src/bin/pg_autoctl/fsm_transition.c

@@ -1320,17 +1320,31 @@ fsm_fast_forward(Keeper *keeper)
 	ReplicationSource *upstream = &(postgres->replicationSource);
 
 	NodeAddress upstreamNode = { 0 };
+	bool found = false;
 
 	char slotName[MAXCONNINFO] = { 0 };
 
-	/* get the primary node to follow */
-	if (!keeper_get_most_advanced_standby(keeper, &upstreamNode))
+	/* get the most advanced peer standby to fetch missing WAL from */
+	if (!keeper_get_most_advanced_standby(keeper, &upstreamNode, &found))
 	{
 		log_error("Failed to fast forward from the most advanced standby node, "
 				  "see above for details");
 		return false;
 	}
 
+	/*
+	 * When no other report_lsn peer exists (because the intended upstream
+	 * already transitioned away), this node is now the most advanced.
+	 * Skip the WAL fetch — the monitor will assign prepare_promotion on the
+	 * next node_active call.
+	 */
+	if (!found)
+	{
+		log_info("No upstream standby found for fast_forward; "
+				 "skipping WAL fetch and proceeding to promotion");
+		return true;
+	}
+
 	/*
 	 * Postgres 10 does not have pg_replication_slot_advance(), so we don't
 	 * support replication slots on standby nodes there.
@@ -1471,15 +1485,23 @@ fsm_init_from_standby(Keeper *keeper)
 	LocalPostgresServer *postgres = &(keeper->postgres);
 
 	NodeAddress upstreamNode = { 0 };
+	bool found = false;
 
 	/* get the primary node to follow */
-	if (!keeper_get_most_advanced_standby(keeper, &upstreamNode))
+	if (!keeper_get_most_advanced_standby(keeper, &upstreamNode, &found))
 	{
 		log_error("Failed to initialise from the most advanced standby node, "
 				  "see above for details");
 		return false;
 	}
 
+	if (!found)
+	{
+		log_error("No standby node found to initialise from; "
+				  "cannot proceed without an upstream source");
+		return false;
+	}
+
 	if (!standby_init_replication_source(postgres,
 										 &upstreamNode,
 										 PG_AUTOCTL_REPLICA_USERNAME,

src/bin/pg_autoctl/keeper.c

@@ -3109,10 +3109,12 @@ keeper_get_primary(Keeper *keeper, NodeAddress *primaryNode)
  * the keeper->otherNodes array.
  */
 bool
-keeper_get_most_advanced_standby(Keeper *keeper, NodeAddress *upstreamNode)
+keeper_get_most_advanced_standby(Keeper *keeper, NodeAddress *upstreamNode,
+								 bool *found)
 {
 	KeeperConfig *config = &(keeper->config);
 	int groupId = keeper->state.current_group;
+	int64_t localNodeId = keeper->state.current_node_id;
 
 	if (!config->monitorDisabled)
 	{
@@ -3121,7 +3123,9 @@ keeper_get_most_advanced_standby(Keeper *keeper, NodeAddress *upstreamNode)
 		if (!monitor_get_most_advanced_standby(monitor,
 											   config->formation,
 											   groupId,
-											   upstreamNode))
+											   localNodeId,
+											   upstreamNode,
+											   found))
 		{
 			log_error("Failed to get the most advanced standby node "
 					  "from the monitor, see above for details");
@@ -3140,6 +3144,12 @@ keeper_get_most_advanced_standby(Keeper *keeper, NodeAddress *upstreamNode)
 			NodeAddress *node = &(keeper->otherNodes.nodes[i]);
 			uint64_t nodeLSN = 0;
 
+			/* skip self to avoid fetching WAL from ourselves */
+			if (node->nodeId == localNodeId)
+			{
+				continue;
+			}
+
 			if (!parseLSN(node->lsn, &nodeLSN))
 			{
 				log_error("Failed to parse node %" PRId64
@@ -3158,14 +3168,14 @@ keeper_get_most_advanced_standby(Keeper *keeper, NodeAddress *upstreamNode)
 
 		if (mostAdvandedStandbyNode == NULL)
 		{
-			log_error("Failed to get the most avdanced standby node "
-					  "from the current list of other nodes, "
-					  "refresh the list with the command: "
-					  "pg_autoctl do fsm nodes set");
-			return false;
+			log_info("No other standby found in local node list; "
+					 "node %" PRId64 " is the most advanced", localNodeId);
+			*found = false;
+			return true;
 		}
 
 		*upstreamNode = *mostAdvandedStandbyNode;
+		*found = true;
 		return true;
 	}
 

src/bin/pg_autoctl/keeper.h

@@ -120,7 +120,8 @@ bool keeper_refresh_citus_remove_dropped_nodes(Keeper *keeper,
 
 bool keeper_read_nodes_from_file(Keeper *keeper, NodeAddressArray *nodesArray);
 bool keeper_get_primary(Keeper *keeper, NodeAddress *primaryNode);
-bool keeper_get_most_advanced_standby(Keeper *keeper, NodeAddress *primaryNode);
+bool keeper_get_most_advanced_standby(Keeper *keeper, NodeAddress *primaryNode,
+									  bool *found);
 
 
 bool keeper_pg_autoctl_get_version_from_disk(Keeper *keeper,

src/bin/pg_autoctl/monitor.c

@@ -751,23 +751,26 @@ monitor_get_coordinator(Monitor *monitor, char *formation,
 bool
 monitor_get_most_advanced_standby(Monitor *monitor,
 								  char *formation, int groupId,
-								  NodeAddress *node)
+								  int64_t callerNodeId,
+								  NodeAddress *node, bool *found)
 {
 	PGSQL *pgsql = &monitor->pgsql;
 	const char *sql =
-		"SELECT * FROM pgautofailover.get_most_advanced_standby($1, $2)";
-	int paramCount = 2;
-	Oid paramTypes[2] = { TEXTOID, INT4OID };
-	const char *paramValues[2];
+		"SELECT * FROM pgautofailover.get_most_advanced_standby($1, $2, $3)";
+	int paramCount = 3;
+	Oid paramTypes[3] = { TEXTOID, INT4OID, INT8OID };
+	const char *paramValues[3];
 
-	/* we expect a single entry */
+	/* we expect zero or one entry */
 	NodeAddressArray nodeArray = { 0 };
 	NodeAddressArrayParseContext parseContext = { { 0 }, &nodeArray, false };
 
 	IntString groupIdString = intToString(groupId);
+	IntString callerNodeIdString = intToString(callerNodeId);
 
 	paramValues[0] = formation;
 	paramValues[1] = groupIdString.strValue;
+	paramValues[2] = callerNodeIdString.strValue;
 
 	if (!pgsql_execute_with_params(pgsql, sql,
 								   paramCount, paramTypes, paramValues,
@@ -781,7 +784,7 @@ monitor_get_most_advanced_standby(Monitor *monitor,
 		return false;
 	}
 
-	if (!parseContext.parsedOK || nodeArray.count != 1)
+	if (!parseContext.parsedOK)
 	{
 		log_error(
 			"Failed to get the most advanced standby node from the monitor "
@@ -792,6 +795,15 @@ monitor_get_most_advanced_standby(Monitor *monitor,
 		return false;
 	}
 
+	/* zero rows: no other report_lsn peer exists; caller is most advanced */
+	if (nodeArray.count == 0)
+	{
+		log_info("No other standby is reporting its LSN; "
+				 "node %" PRId64 " is the most advanced", callerNodeId);
+		*found = false;
+		return true;
+	}
+
 	/* copy the node we retrieved in the expected place */
 	node->nodeId = nodeArray.nodes[0].nodeId;
 	strlcpy(node->name, nodeArray.nodes[0].name, _POSIX_HOST_NAME_MAX);
@@ -803,6 +815,7 @@ monitor_get_most_advanced_standby(Monitor *monitor,
 	log_debug("The most advanced standby node is node " NODE_FORMAT,
 			  node->nodeId, node->name, node->host, node->port);
 
+	*found = true;
 	return true;
 }
 

src/bin/pg_autoctl/monitor.h

@@ -124,7 +124,8 @@ bool monitor_get_coordinator(Monitor *monitor, char *formation,
 							 CoordinatorNodeAddress *coordinatorNodeAddress);
 bool monitor_get_most_advanced_standby(Monitor *monitor,
 									   char *formation, int groupId,
-									   NodeAddress *node);
+									   int64_t callerNodeId,
+									   NodeAddress *node, bool *found);
 bool monitor_register_node(Monitor *monitor,
 						   char *formation,
 						   char *name,

src/monitor/Makefile

@@ -14,7 +14,7 @@ MODULE_big = $(EXTENSION)
 OBJS = $(patsubst ${SRC_DIR}%.c,%.o,$(wildcard ${SRC_DIR}*.c))
 PG_CPPFLAGS = -std=c99 -Wall -Werror -Wno-unused-parameter -Iinclude -I$(libpq_srcdir) -g
 SHLIB_LINK = $(libpq)
-REGRESS = create_extension monitor workers node_active_protocol guard_data_loss dummy_update drop_extension upgrade
+REGRESS = create_extension monitor workers node_active_protocol guard_data_loss fast_forward dummy_update drop_extension upgrade
 
 PG_CONFIG ?= pg_config
 PGXS = $(shell $(PG_CONFIG) --pgxs)