Fix 3-DC split-brain: monitor assigns wait_primary on replication stall (issue #997)

src/bin/pg_autoctl/cli_common.c

@@ -407,6 +407,14 @@ cli_common_keeper_getopts(int argc, char **argv,
 				break;
 			}
 
+			case 'G':
+			{
+				/* { "region", required_argument, NULL, 'G' } */
+				strlcpy(LocalOptionConfig.region, optarg, NAMEDATALEN);
+				log_trace("--region %s", LocalOptionConfig.region);
+				break;
+			}
+
 			case 'V':
 			{
 				/* keeper_cli_print_version prints version and exits. */

src/bin/pg_autoctl/cli_create_node.c

@@ -343,6 +343,7 @@ cli_create_postgres_getopts(int argc, char **argv)
 		{ "candidate-priority", required_argument, NULL, 'P' },
 		{ "replication-quorum", required_argument, NULL, 'r' },
 		{ "maximum-backup-rate", required_argument, NULL, 'R' },
+		{ "region", required_argument, NULL, 'G' },
 		{ "run", no_argument, NULL, 'x' },
 		{ "no-ssl", no_argument, NULL, 'N' },
 		{ "ssl-self-signed", no_argument, NULL, 's' },
@@ -356,7 +357,7 @@ cli_create_postgres_getopts(int argc, char **argv)
 
 	int optind =
 		cli_create_node_getopts(argc, argv, long_options,
-								"C:D:H:p:l:U:A:SLd:a:n:f:m:MI:W:w:RVvqhP:r:xsN",
+								"C:D:H:p:l:U:A:SLd:a:n:f:m:MI:W:w:RGVvqhP:r:xsN",
 								&options);
 
 	/* publish our option parsing in the global variable */
@@ -441,6 +442,7 @@ cli_create_coordinator_getopts(int argc, char **argv)
 		{ "citus-cluster", required_argument, NULL, 'Z' },
 		{ "candidate-priority", required_argument, NULL, 'P' },
 		{ "replication-quorum", required_argument, NULL, 'r' },
+		{ "region", required_argument, NULL, 'G' },
 		{ "run", no_argument, NULL, 'x' },
 		{ "no-ssl", no_argument, NULL, 'N' },
 		{ "ssl-self-signed", no_argument, NULL, 's' },
@@ -454,7 +456,7 @@ cli_create_coordinator_getopts(int argc, char **argv)
 
 	int optind =
 		cli_create_node_getopts(argc, argv, long_options,
-								"C:D:H:p:l:U:A:SLd:a:n:f:m:MRVvqhzZ:P:r:xsN",
+								"C:D:H:p:l:U:A:SLd:a:n:f:m:MRGVvqhzZ:P:r:xsN",
 								&options);
 
 	options.groupId = 0;
@@ -551,6 +553,7 @@ cli_create_worker_getopts(int argc, char **argv)
 		{ "citus-cluster", required_argument, NULL, 'Z' },
 		{ "candidate-priority", required_argument, NULL, 'P' },
 		{ "replication-quorum", required_argument, NULL, 'r' },
+		{ "region", required_argument, NULL, 'G' },
 		{ "run", no_argument, NULL, 'x' },
 		{ "no-ssl", no_argument, NULL, 'N' },
 		{ "ssl-self-signed", no_argument, NULL, 's' },
@@ -564,7 +567,7 @@ cli_create_worker_getopts(int argc, char **argv)
 
 	int optind =
 		cli_create_node_getopts(argc, argv, long_options,
-								"C:D:H:p:l:y:zZ:U:A:SLd:a:n:f:m:MRVvqhzP:r:xsN",
+								"C:D:H:p:l:y:zZ:U:A:SLd:a:n:f:m:MRGVvqhzP:r:xsN",
 								&options);
 
 	if (options.groupId == 0)

src/bin/pg_autoctl/defaults.h

@@ -19,7 +19,7 @@
 #define PG_AUTOCTL_VERSION GIT_VERSION
 
 /* version of the extension that we requite to talk to on the monitor */
-#define PG_AUTOCTL_EXTENSION_VERSION "2.2"
+#define PG_AUTOCTL_EXTENSION_VERSION "2.3"
 
 /* environment variable to use to make DEBUG facilities available */
 #define PG_AUTOCTL_DEBUG "PG_AUTOCTL_DEBUG"

src/bin/pg_autoctl/keeper.c

@@ -1670,6 +1670,7 @@ keeper_register_and_init(Keeper *keeper, NodeState initialState)
 								  config->pgSetup.settings.candidatePriority,
 								  config->pgSetup.settings.replicationQuorum,
 								  config->pgSetup.citusClusterName,
+								  config->region,
 								  &mayRetry,
 								  &assignedState))
 		{
@@ -1875,6 +1876,7 @@ keeper_register_again(Keeper *keeper)
 								  config->pgSetup.settings.candidatePriority,
 								  config->pgSetup.settings.replicationQuorum,
 								  DEFAULT_CITUS_CLUSTER_NAME,
+								  config->region,
 								  &mayRetry,
 								  &assignedState))
 		{

src/bin/pg_autoctl/keeper_config.c

@@ -61,6 +61,10 @@
 	make_strbuf_option("pg_autoctl", "nodekind", NULL, false, NAMEDATALEN, \
 					   config->nodeKind)
 
+#define OPTION_AUTOCTL_REGION(config) \
+	make_strbuf_option_default("pg_autoctl", "region", "region", \
+							   false, NAMEDATALEN, config->region, "")
+
 #define OPTION_POSTGRESQL_PGDATA(config) \
 	make_strbuf_option("postgresql", "pgdata", "pgdata", true, MAXPGPATH, \
 					   config->pgSetup.pgdata)
@@ -227,6 +231,7 @@
 		OPTION_AUTOCTL_HOSTNAME(config), \
 		OPTION_AUTOCTL_NODENAME(config), \
 		OPTION_AUTOCTL_NODEKIND(config), \
+		OPTION_AUTOCTL_REGION(config), \
 		OPTION_POSTGRESQL_PGDATA(config), \
 		OPTION_POSTGRESQL_PG_CTL(config), \
 		OPTION_POSTGRESQL_USERNAME(config), \

src/bin/pg_autoctl/keeper_config.h

@@ -46,6 +46,7 @@ typedef struct KeeperConfig
 	char name[_POSIX_HOST_NAME_MAX];
 	char hostname[_POSIX_HOST_NAME_MAX];
 	char nodeKind[NAMEDATALEN];
+	char region[NAMEDATALEN];
 
 	/* PostgreSQL setup */
 	PostgresSetup pgSetup;

src/bin/pg_autoctl/monitor.c

@@ -861,19 +861,20 @@ monitor_register_node(Monitor *monitor, char *formation,
 					  NodeState initialState,
 					  PgInstanceKind kind, int candidatePriority, bool quorum,
 					  char *citusClusterName,
+					  char *region,
 					  bool *mayRetry,
 					  MonitorAssignedState *assignedState)
 {
 	PGSQL *pgsql = &monitor->pgsql;
 	const char *sql =
 		"SELECT * FROM pgautofailover.register_node($1, $2, $3, $4, $5, $6, $7, "
-		"$8, $9::pgautofailover.replication_state, $10, $11, $12, $13)";
-	int paramCount = 13;
-	Oid paramTypes[13] = {
+		"$8, $9::pgautofailover.replication_state, $10, $11, $12, $13, $14)";
+	int paramCount = 14;
+	Oid paramTypes[14] = {
 		TEXTOID, TEXTOID, INT4OID, NAMEOID, TEXTOID, INT8OID,
-		INT8OID, INT4OID, TEXTOID, TEXTOID, INT4OID, BOOLOID, TEXTOID
+		INT8OID, INT4OID, TEXTOID, TEXTOID, INT4OID, BOOLOID, TEXTOID, TEXTOID
 	};
-	const char *paramValues[13];
+	const char *paramValues[14];
 	MonitorAssignedStateParseContext parseContext =
 	{ { 0 }, assignedState, false };
 	const char *nodeStateString = NodeStateToString(initialState);
@@ -899,6 +900,10 @@ monitor_register_node(Monitor *monitor, char *formation,
 		IS_EMPTY_STRING_BUFFER(citusClusterName)
 		? DEFAULT_CITUS_CLUSTER_NAME
 		: citusClusterName;
+	paramValues[13] =
+		IS_EMPTY_STRING_BUFFER(region)
+		? "default"
+		: region;
 
 	if (!pgsql_execute_with_params(pgsql, sql,
 								   paramCount, paramTypes, paramValues,
@@ -1950,7 +1955,7 @@ monitor_get_current_state(Monitor *monitor, char *formation, int group,
 				"         current_group_state, assigned_group_state, "
 				"         candidate_priority, replication_quorum, "
 				"         reported_tli, reported_lsn, health, nodecluster, "
-				"         healthlag, reportlag"
+				"         noderegion, healthlag, reportlag"
 				"    FROM pgautofailover.current_state($1) cs "
 				"    JOIN ("
 				"          select nodeid, "
@@ -1976,7 +1981,7 @@ monitor_get_current_state(Monitor *monitor, char *formation, int group,
 				"         current_group_state, assigned_group_state, "
 				"         candidate_priority, replication_quorum, "
 				"         reported_tli, reported_lsn, health, nodecluster, "
-				"         healthlag, reportlag"
+				"         noderegion, healthlag, reportlag"
 				"    FROM pgautofailover.current_state($1, $2) cs "
 				"    JOIN ("
 				"          select nodeid, "
@@ -2029,7 +2034,7 @@ parseCurrentNodeState(PGresult *result, int rowNumber,
 	int errors = 0;
 
 	/* we don't expect any of the column to be NULL */
-	for (colNumber = 0; colNumber < 16; colNumber++)
+	for (colNumber = 0; colNumber < 17; colNumber++)
 	{
 		if (PQgetisnull(result, rowNumber, 0))
 		{
@@ -2054,8 +2059,9 @@ parseCurrentNodeState(PGresult *result, int rowNumber,
 	 * 11 - OUT reported_lsn         pg_lsn,
 	 * 12 - OUT health               integer
 	 * 13 - OUT nodecluster          text
-	 * 14 -     healthlag            int (extract epoch from interval)
-	 * 15 -     reportlag            int (extract epoch from interval)
+	 * 14 - OUT noderegion           text
+	 * 15 -     healthlag            int (extract epoch from interval)
+	 * 16 -     reportlag            int (extract epoch from interval)
 	 *
 	 * We need the groupId to parse the formation kind into a nodeKind, so we
 	 * begin at column 1 and get back to column 0 later, after column 4.
@@ -2191,14 +2197,24 @@ parseCurrentNodeState(PGresult *result, int rowNumber,
 	}
 
 	value = PQgetvalue(result, rowNumber, 14);
+	length = strlcpy(nodeState->region, value, NAMEDATALEN);
+	if (length >= NAMEDATALEN)
+	{
+		log_error("Region \"%s\" returned by monitor is %d characters, "
+				  "the maximum supported by pg_autoctl is %d",
+				  value, length, NAMEDATALEN - 1);
+		++errors;
+	}
+
+	value = PQgetvalue(result, rowNumber, 15);
 
 	if (!stringToDouble(value, &(nodeState->healthLag)))
 	{
 		log_error("Invalid health lag \"%s\" returned by monitor", value);
 		++errors;
 	}
 
-	value = PQgetvalue(result, rowNumber, 15);
+	value = PQgetvalue(result, rowNumber, 16);
 
 	if (!stringToDouble(value, &(nodeState->reportLag)))
 	{
@@ -2231,10 +2247,10 @@ parseCurrentNodeStateArray(CurrentNodeStateArray *nodesArray, PGresult *result)
 		return false;
 	}
 
-	/* monitor_get_current_state selects 16 columns (0–15) */
-	if (PQnfields(result) != 16)
+	/* monitor_get_current_state selects 17 columns (0–16) */
+	if (PQnfields(result) != 17)
 	{
-		log_error("Query returned %d columns, expected 16", PQnfields(result));
+		log_error("Query returned %d columns, expected 17", PQnfields(result));
 		return false;
 	}
 

src/bin/pg_autoctl/monitor.h

@@ -140,6 +140,7 @@ bool monitor_register_node(Monitor *monitor,
 						   int candidatePriority,
 						   bool quorum,
 						   char *citusClusterName,
+						   char *region,
 						   bool *mayRetry,
 						   MonitorAssignedState *assignedState);
 bool monitor_node_active(Monitor *monitor,

src/bin/pg_autoctl/nodespec.c

@@ -114,6 +114,9 @@ nodespec_read(const char *path, NodeSpec *spec)
 		make_strbuf_option_default("settings", "replication_quorum", NULL, false,
 								   sizeof(replicationQuorumStr),
 								   replicationQuorumStr, "true"),
+		make_strbuf_option_default("settings", "region", NULL, false,
+								   sizeof(spec->region), spec->region,
+								   ""),
 
 		/* [options] — immutable, used only at create time */
 		make_strbuf_option_default("options", "ssl", NULL, false,
@@ -694,6 +697,13 @@ nodespec_create_argv(const NodeSpec *spec,
 			PUSH("false");
 		}
 
+		/* region label (omit when empty — monitor defaults to "default") */
+		if (!IS_EMPTY_STRING_BUFFER(spec->region))
+		{
+			PUSH("--region");
+			PUSH(spec->region);
+		}
+
 		/* Citus secondary/read-replica cluster settings */
 		if (spec->citusSecondary)
 		{

src/bin/pg_autoctl/nodespec.h

@@ -71,6 +71,7 @@ typedef struct NodeSpec
 	/* [settings]  — mutable; applied on SIGHUP / file change */
 	int candidate_priority;      /* 0-100, default 50 */
 	bool replication_quorum;     /* sync quorum participant, default true */
+	char region[NAMEDATALEN];    /* data-centre / availability zone label */
 
 	/* [options]    — immutable; used only at pg_autoctl create time */
 	char ssl[32];                /* self-signed | verify-ca | verify-full | off */

src/bin/pg_autoctl/nodestate_utils.h

@@ -25,6 +25,7 @@ typedef struct CurrentNodeState
 
 	char formation[NAMEDATALEN];
 	char citusClusterName[NAMEDATALEN];
+	char region[NAMEDATALEN];
 	int groupId;
 	PgInstanceKind pgKind;
 

src/bin/pg_autoctl/watch.c

@@ -898,6 +898,12 @@ compute_column_size(ColumnType type, NodeAddressHeaders *headers)
 			return headers->maxStateSize;
 		}
 
+		case COLUMN_TYPE_REGION:
+		{
+			/* "default" is 7 chars; most region names fit in 12 */
+			return 12;
+		}
+
 		default:
 		{
 			log_fatal("BUG: compute_column_size(%d)", type);
@@ -1085,6 +1091,12 @@ print_node_state(WatchContext *context, ColPolicy *policy,
 				break;
 			}
 
+			case COLUMN_TYPE_REGION:
+			{
+				mvprintw(r, cc, "%*s", len, nodeState->region);
+				break;
+			}
+
 			default:
 			{
 				log_fatal("BUG: print_node_state(%d)", cType);

src/bin/pg_autoctl/watch_colspecs.h

@@ -27,6 +27,7 @@ typedef enum
 	COLUMN_TYPE_CONN_REPORT_LAG,
 	COLUMN_TYPE_REPORTED_STATE,
 	COLUMN_TYPE_ASSIGNED_STATE,
+	COLUMN_TYPE_REGION,
 
 	COLUMN_TYPE_LAST
 } ColumnType;
@@ -38,7 +39,7 @@ typedef struct ColSpec
 	int len;
 } ColSpec;
 
-#define MAX_COL_SPECS 12
+#define MAX_COL_SPECS 14
 
 typedef struct ColPolicy
 {
@@ -134,6 +135,7 @@ ColPolicy ColumnPolicies[] = {
 		{
 			{ COLUMN_TYPE_NAME, "Name", 0 },
 			{ COLUMN_TYPE_ID, "Node", 0 },
+			{ COLUMN_TYPE_REGION, "Region", 0 },
 			{ COLUMN_TYPE_REPLICATION_QUORUM, "Quorum", 0 },
 			{ COLUMN_TYPE_CANDIDATE_PRIORITY, "Priority", 0 },
 			{ COLUMN_TYPE_CONN_HEALTH_LAG, "Check", 0 },
@@ -150,6 +152,7 @@ ColPolicy ColumnPolicies[] = {
 		{
 			{ COLUMN_TYPE_NAME, "Name", 0 },
 			{ COLUMN_TYPE_ID, "Node", 0 },
+			{ COLUMN_TYPE_REGION, "Region", 0 },
 			{ COLUMN_TYPE_REPLICATION_QUORUM, "Quorum", 0 },
 			{ COLUMN_TYPE_CANDIDATE_PRIORITY, "Priority", 0 },
 			{ COLUMN_TYPE_TLI_LSN, "TLI: LSN", 0 },
@@ -167,6 +170,7 @@ ColPolicy ColumnPolicies[] = {
 		{
 			{ COLUMN_TYPE_NAME, "Name", 0 },
 			{ COLUMN_TYPE_ID, "Node", 0 },
+			{ COLUMN_TYPE_REGION, "Region", 0 },
 			{ COLUMN_TYPE_REPLICATION_QUORUM, "Quorum", 0 },
 			{ COLUMN_TYPE_CANDIDATE_PRIORITY, "Priority", 0 },
 			{ COLUMN_TYPE_HOST_PORT, "Host:Port", 0 },
@@ -185,6 +189,7 @@ ColPolicy ColumnPolicies[] = {
 		{
 			{ COLUMN_TYPE_NAME, "Name", 0 },
 			{ COLUMN_TYPE_ID, "Node", 0 },
+			{ COLUMN_TYPE_REGION, "Region", 0 },
 			{ COLUMN_TYPE_REPLICATION_QUORUM, "Quorum", 0 },
 			{ COLUMN_TYPE_CANDIDATE_PRIORITY, "Priority", 0 },
 			{ COLUMN_TYPE_HOST_PORT, "Host:Port", 0 },

src/bin/pgaftest/cli_indent.c

@@ -450,6 +450,10 @@ print_node(FILE *out, const TestNode *n, int baseIndent)
 		sformat(numbuf, sizeof(numbuf), "%d", n->candidatePriority);
 		ADD("candidate-priority", numbuf);
 	}
+	if (n->region[0])
+	{
+		ADD("region", n->region);
+	}
 	if (!n->replicationQuorum)
 	{
 		ADD("replication-quorum", "false");

src/bin/pgaftest/compose_gen.c

@@ -1721,6 +1721,11 @@ compose_gen_write_node_ini(const TestCluster *cluster,
 			node->candidatePriority,
 			node->replicationQuorum ? "true" : "false");
 
+	if (node->region[0])
+	{
+		fformat(f, "region = %s\n", node->region);
+	}
+
 	/* Citus role/cluster settings live in their own [citus] section */
 	if (node->citusSecondary || node->citusClusterName[0])
 	{

src/bin/pgaftest/test_spec.h

@@ -73,6 +73,7 @@ typedef struct TestNode
 	char auth[32];               /* per-node auth override; "" = use cluster */
 	char replicationPassword[256]; /* replication.password written to node ini */
 	char monitorPassword[256];   /* pg_auto_failover.monitor_password written to node ini */
+	char region[64];             /* --region NAME; "" = omit (defaults to "default" on monitor) */
 
 	/* Extra Docker named volumes: volume <name> <containerPath> */
 	struct