Monitor: detect and recover from fast_forward candidate stuck with all WAL sources unhealthy (#1143)

* Fix fast_forward self-reference infinite loop

When a node is assigned the fast_forward goalstate it has not yet
reported that state back to the monitor, so reportedstate is still
report_lsn.  If the originally-selected upstream peer transitions
out of report_lsn before the fast_forward node calls
get_most_advanced_standby(), the fast_forward node becomes the only
remaining report_lsn node and is returned as its own upstream source.
The node then loops forever trying to fetch WAL from itself.

Fix across three layers:

SQL (get_most_advanced_standby): add caller_node_id bigint parameter
(default 0) and filter AND nodeid != $3 so the calling node can
never be returned as its own upstream.

monitor_get_most_advanced_standby: accept callerNodeId and a bool
*found output parameter.  A zero-row result is no longer an error:
it means the caller is already the most advanced node and the found
flag is set to false.

keeper_get_most_advanced_standby: pass the local node ID to the
monitor call; also skip self in the no-monitor path.  Propagates the
found flag to the caller.

fsm_fast_forward: when found is false (no valid upstream), skip the
WAL fetch and return true so the keeper can report its current state
to the monitor.  The monitor will then assign prepare_promotion on
the next node_active call, breaking the loop.

fsm_init_from_standby also uses keeper_get_most_advanced_standby;
there a not-found result is a genuine error (no source to clone from)
so it returns false with an explicit log message.

Fixes #1060

* Monitor: detect fast_forward candidate stuck with all WAL sources unhealthy

When a failover candidate is assigned fast_forward (its LSN lags behind a
peer standby that holds more WAL), it fetches the missing WAL before
promotion.  If the WAL-source node(s) die while the candidate is fetching,
the candidate gets stuck: it reports back report_lsn (failed fetch) while
its goal stays fast_forward.  IsBeingPromoted() holds the lock, so the
election cannot restart, and ProceedWithMSFailover does nothing useful
because CandidateNodeIsReadyToStreamWAL explicitly excludes fast_forward.

Fix (group_state_machine.c):
  Add WalSourceNodesAreAllUnhealthy() helper that scans the group for
  nodes in {report_lsn, report_lsn} (the WAL-source state) whose health
  is bad.  When this fires for a candidate in {report_lsn, fast_forward}
  inside ProceedGroupStateForMSFailover, the monitor acts based on the
  guard_data_loss GUC:

  - guard_data_loss=true (default): reset the candidate goal back to
    report_lsn and WARN the operator.  The election retries automatically
    if a source recovers.  Operators can unblock with:
      pg_autoctl perform failover --allow-data-loss

  - guard_data_loss=false: log and fall through.
    get_most_advanced_standby() now filters out unhealthy nodes when
    guard_data_loss=false (SQL change below), so fsm_fast_forward() finds
    no upstream, skips the WAL fetch, and reports fast_forward as current.
    The monitor then assigns prepare_promotion on the next call.

Fix (pgautofailover.sql):
  get_most_advanced_standby() adds:
    AND (current_setting('pgautofailover.guard_data_loss')::bool
         OR health > 0)
  When guard_data_loss=false, unhealthy nodes are excluded from WAL-source
  selection.  When guard_data_loss=true, the highest-LSN node is returned
  regardless of health (preserving the no-data-loss guarantee).

Tests:
  src/monitor/sql/fast_forward.sql  - regression test: bootstraps a 3-node
    formation, manually places the candidate in {report_lsn, fast_forward}
    with the WAL source unhealthy, then:
      Test A (guard_data_loss=true):  node_active returns report_lsn ✓
      Test B (guard_data_loss=false): node_active returns fast_forward ✓
    Also calls get_most_advanced_standby() directly to exercise both SQL
    filter paths.
  tests/tap/specs/fast_forward.pgaf - integration test: kills primary and
    WAL-source standby together, verifies node stays at report_lsn, then
    unblocks with --allow-data-loss and verifies full cluster recovery.
  src/monitor/Makefile and tests/tap/schedule updated accordingly.

Fixes #1060

Author: dimitri

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)