more price handling of recovery mode

Author: knizhnik

contrib/mmts/multimaster.c

@@ -1882,12 +1882,23 @@ static void
 MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 {
 	ListCell *param;
+	bool recoveryCompleted = false;
 	MtmIsRecoverySession = false;
 	foreach(param, args->in_params)
 	{
 		DefElem    *elem = lfirst(param);
 		if (strcmp("mtm_replication_mode", elem->defname) == 0) { 
-			MtmIsRecoverySession = elem->arg != NULL && strVal(elem->arg) != NULL && strcmp(strVal(elem->arg), "recovery") == 0;
+			if (elem->arg != NULL && strVal(elem->arg) != NULL) { 
+				if (strcmp(strVal(elem->arg), "recovery") == 0) { 
+					MtmIsRecoverySession = true;
+				} else if (strcmp(strVal(elem->arg), "recovered") == 0) { 
+					recoveryCompleted = true;
+				} else if (strcmp(strVal(elem->arg), "normal") != 0) { 
+					elog(ERROR, "Illegal recovery mode %s", strVal(elem->arg));
+				}
+			} else { 
+				elog(ERROR, "Replication mode is not specified");
+			}				
 			break;
 		}
 	}
@@ -1900,10 +1911,14 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 			MtmCheckQuorum();
 		}
 	} else if (BIT_CHECK(Mtm->disabledNodeMask,  MtmReplicationNodeId-1)) {
-		elog(WARNING, "Node %d consider that recovery of node %d is completed: start normal replication", MtmNodeId, MtmReplicationNodeId); 
-		BIT_CLEAR(Mtm->disabledNodeMask,  MtmReplicationNodeId-1);
-		Mtm->nNodes += 1;
-		MtmCheckQuorum();
+		if (recoveryCompleted) { 
+			elog(WARNING, "Node %d consider that recovery of node %d is completed: start normal replication", MtmNodeId, MtmReplicationNodeId); 
+			BIT_CLEAR(Mtm->disabledNodeMask,  MtmReplicationNodeId-1);
+			Mtm->nNodes += 1;
+			MtmCheckQuorum();
+		} else {
+			elog(ERROR, "Disabled node %d tries to reconnect without recovery", MtmReplicationNodeId); 
+		}
 	} else {
 		elog(NOTICE, "Node %d start logical replication to node %d in normal mode", MtmNodeId, MtmReplicationNodeId); 
 	}

contrib/mmts/pglogical_receiver.c

@@ -196,6 +196,13 @@ feTimestampDifference(int64 start_time, int64 stop_time,
 	}
 }
 
+static char const* const MtmReplicationModeName[] = 
+{
+	"recovered", /* SLOT_CREATE_NEW: recovery of node is completed so drop old slot and restart replication from the current position in WAL */
+	"recovery",  /* SLOT_OPEN_EXISTED: perform recorvery of the node by applying all data from theslot from specified point */
+	"normal"     /* SLOT_OPEN_ALWAYS: normal mode: use existeed slot or create new one and start receiving data from it from the specified position */
+};
+
 static void
 pglogical_receiver_main(Datum main_arg)
 {
@@ -298,7 +305,7 @@ pglogical_receiver_main(Datum main_arg)
 					  (uint32) originStartPos,
 					  MULTIMASTER_MAX_PROTO_VERSION,
 					  MULTIMASTER_MIN_PROTO_VERSION,
-					  mode == SLOT_OPEN_EXISTED ? "recovery" : "normal"					  
+					  MtmReplicationModeName[mode]
 		);
 	res = PQexec(conn, query->data);
 	if (PQresultStatus(res) != PGRES_COPY_BOTH)