Add mmts control file

Author: knizhnik

multimaster.c

@@ -213,6 +213,7 @@ bool  MtmVolksWagenMode;
 TransactionId  MtmUtilityProcessedInXid;
 
 static char* MtmConnStrs;
+static char* MtmClusterName;
 static int   MtmQueueSize;
 static int   MtmWorkers;
 static int   MtmVacuumDelay;
@@ -1866,6 +1867,39 @@ static void MtmRaftableInitialize()
 	raftable_start(MtmNodeId - 1);
 }
 
+static void MtmCheckControlFile(void)
+{
+	char controlFilePath[MAXPGPATH];
+	char buf[MULTIMASTER_MAX_CTL_STR_SIZE];
+	FILE* f;
+	snprintf(controlFilePath, MAXPGPATH, "%s/global/mmts_control", DataDir);
+	f = fopen(controlFilePath, "r");
+	if (f != NULL && fgets(buf, sizeof buf, f)) {
+		char* sep = strchr(buf, ':');
+		if (sep == NULL) {
+			elog(FATAL, "File mmts_control doesn't contain cluster name");
+		}
+		*sep = '\0';
+		if (strcmp(buf, MtmClusterName) != 0) { 
+			elog(FATAL, "Database belongs to some other cluster %s rather than %s", buf, MtmClusterName);
+		}
+		if (sscanf(sep+1, "%d", &Mtm->donorNodeId) != 1) { 
+			elog(FATAL, "File mmts_control doesn't contain node id");
+		}
+		fclose(f);
+	} else { 
+		if (f != NULL) { 
+			fclose(f);
+		}
+		f = fopen(controlFilePath, "w");
+		if (f == NULL) { 
+			elog(FATAL, "Failed to create mmts_control file: %m");
+		}
+		Mtm->donorNodeId = -1;
+		fprintf(f, "%s:%d\n", MtmClusterName, Mtm->donorNodeId);
+		fclose(f);
+	}
+}
 
 static void MtmInitialize()
 {
@@ -1930,6 +1964,8 @@ static void MtmInitialize()
     MtmDoReplication = true;
 	TM = &MtmTM;
 	LWLockRelease(AddinShmemInitLock);
+
+	MtmCheckControlFile();
 }
 
 static void 
@@ -2471,6 +2507,19 @@ _PG_init(void)
 		NULL         /* GucShowHook show_hook */
 	);
 
+	DefineCustomStringVariable(
+		"multimaster.cluster_name",
+		"Name of the cluster",
+		NULL,
+		&MtmClusterName,
+		"mmts",
+		PGC_BACKEND, /* context */
+		0,           /* flags */
+		NULL,        /* GucStringCheckHook check_hook */
+		NULL,        /* GucStringAssignHook assign_hook */
+		NULL         /* GucShowHook show_hook */
+	);
+    
 	DefineCustomIntVariable(
 		"multimaster.node_id",
 		"Multimaster node ID",
@@ -2608,8 +2657,10 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 		MtmLock(LW_EXCLUSIVE);
 		if (Mtm->status == MTM_RECOVERY) { 
 			recovery = true;
-			if (Mtm->recoverySlot == 0 || Mtm->recoverySlot == nodeId) { 
-				/* Choose for recovery first available slot */
+			if ((Mtm->recoverySlot == 0 && (Mtm->donorNodeId < 0 || Mtm->donorNodeId == nodeId))
+				|| Mtm->recoverySlot == nodeId) 
+			{ 
+				/* Choose for recovery first available slot or slot of donor node (if any) */
 				elog(WARNING, "Process %d starts recovery from node %d", MyProcPid, nodeId);
 				Mtm->recoverySlot = nodeId;
 				Mtm->nReceivers = 0;
@@ -2697,6 +2748,8 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 {
 	ListCell *param;
 	bool recoveryCompleted = false;
+	XLogRecPtr recoveryStartPos = InvalidXLogRecPtr;
+
 	MtmIsRecoverySession = false;
 	Mtm->nodes[MtmReplicationNodeId-1].senderPid = MyProcPid;
 	Mtm->nodes[MtmReplicationNodeId-1].senderStartTime = MtmGetSystemTime();
@@ -2716,11 +2769,21 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 				elog(ERROR, "Replication mode is not specified");
 			}				
 			break;
+		} else if (strcmp("mtm_restart_pos", elem->defname) == 0) { 
+			if (elem->arg != NULL && strVal(elem->arg) != NULL) {
+				recoveryStartPos = intVal(elem->arg);
+			} else { 
+				elog(ERROR, "Restart position is not specified");
+			}
 		}
 	}
 	MtmLock(LW_EXCLUSIVE);
-	if (MtmIsRecoverySession) {
-		MTM_LOG1("%d: Node %d start recovery of node %d", MyProcPid, MtmNodeId, MtmReplicationNodeId);
+	if (MtmIsRecoverySession) {		
+		MTM_LOG1("%d: Node %d start recovery of node %d at position %lx", MyProcPid, MtmNodeId, MtmReplicationNodeId, recoveryStartPos);
+		Assert(MyReplicationSlot != NULL);
+		if (recoveryStartPos < MyReplicationSlot->data.restart_lsn) { 
+			elog(ERROR, "Specified recovery start position %lx is beyond restart lsn %lx", recoveryStartPos, MyReplicationSlot->data.restart_lsn);
+		}
 		if (!BIT_CHECK(Mtm->disabledNodeMask,  MtmReplicationNodeId-1)) {
 			MtmDisableNode(MtmReplicationNodeId);
 			MtmCheckQuorum();

multimaster.h

@@ -52,6 +52,7 @@
 #define MULTIMASTER_MAX_CONN_STR_SIZE   128
 #define MULTIMASTER_MAX_HOST_NAME_SIZE  64
 #define MULTIMASTER_MAX_LOCAL_TABLES    256
+#define MULTIMASTER_MAX_CTL_STR_SIZE    256
 #define MULTIMASTER_BROADCAST_SERVICE   "mtm_broadcast"
 #define MULTIMASTER_ADMIN               "mtm_admin"
 
@@ -241,6 +242,7 @@ typedef struct
 	int    nActiveTransactions;        /* Nunmber of active 2PC transactions */
 	int    nConfigChanges;             /* Number of cluster configuration changes */
 	int    recoveryCount;              /* Number of completed recoveries */
+	int    donorNodeId;               /* Cluster node from which this node was populated */
 	int64  timeShift;                  /* Local time correction */
 	csn_t  csn;                        /* Last obtained timestamp: used to provide unique acending CSNs based on system time */
 	csn_t  lastCsn;                    /* CSN of last committed transaction */

pglogical_receiver.c

@@ -285,7 +285,7 @@ pglogical_receiver_main(Datum main_arg)
 			timeline = Mtm->nodes[nodeId-1].timeline;
 			newTimeline = true;
 		}
-		/* My original assumption was that we can perfrom recovery only fromm existed slot, 
+		/* My original assumption was that we can perfrom recovery only from existed slot, 
 		 * but unfortunately looks like slots can "disapear" together with WAL-sender.
 		 * So let's try to recreate slot always. */
 		/* if (mode != REPLMODE_REPLICATION) */
@@ -325,7 +325,7 @@ pglogical_receiver_main(Datum main_arg)
 				 * Them are either empty, either new node is synchronized using base_backup.
 				 * So we assume that LSNs are the same for local and remote node
 				 */
-				originStartPos = Mtm->status == MTM_RECOVERY ? GetXLogInsertRecPtr() : InvalidXLogRecPtr;
+				originStartPos = Mtm->status == MTM_RECOVERY && Mtm->donorNodeId == nodeId ? GetXLogInsertRecPtr() : InvalidXLogRecPtr;
 				MTM_LOG1("Start logical receiver at position %lx from node %d", originStartPos, nodeId);
 			} else { 
 				originStartPos = replorigin_get_progress(originId, false);
@@ -335,13 +335,14 @@ pglogical_receiver_main(Datum main_arg)
 			CommitTransactionCommand();
 		}
 
-		appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %x/%x (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"forward_changesets\" '1', \"mtm_replication_mode\" '%s')",
+		appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %x/%x (\"startup_params_format\" '1', \"max_proto_version\" '%d',  \"min_proto_version\" '%d', \"forward_changesets\" '1', \"mtm_replication_mode\" '%s', \"mtm_restart_pos\" '%lx')",
 						  slotName,
 						  (uint32) (originStartPos >> 32),
 						  (uint32) originStartPos,
 						  MULTIMASTER_MAX_PROTO_VERSION,
 						  MULTIMASTER_MIN_PROTO_VERSION,
-						  MtmReplicationModeName[mode]
+						  MtmReplicationModeName[mode],
+						  originStartPos
 			);
 		res = PQexec(conn, query->data);
 		if (PQresultStatus(res) != PGRES_COPY_BOTH)