Do not drop slot on end of recovery and try to explicitly specify restart position for it

Author: knizhnik

multimaster.c

@@ -1801,6 +1801,8 @@ static void MtmInitialize()
 			Mtm->nodes[i].con = MtmConnections[i];
 			Mtm->nodes[i].flushPos = 0;
 			Mtm->nodes[i].lastHeartbeat = 0;
+			Mtm->nodes[i].restartLsn = 0;
+			Mtm->nodes[i].originId = InvalidRepOriginId;
 		}
 		PGSemaphoreCreate(&Mtm->votingSemaphore);
 		PGSemaphoreReset(&Mtm->votingSemaphore);

multimaster.h

@@ -146,6 +146,7 @@ typedef struct
 	XLogRecPtr  flushPos;
 	csn_t       oldestSnapshot; /* Oldest snapshot used by active transactions at this node */	
 	XLogRecPtr  restartLsn;
+	RepOriginId originId;
 } MtmNodeInfo;
 
 typedef struct MtmTransState

pglogical_apply.c

@@ -539,11 +539,13 @@ MtmEndSession(void)
 static void
 process_remote_commit(StringInfo in)
 {
+	int         i;
 	uint8 		flags;
 	csn_t       csn;
 	const char *gid = NULL;	
 	XLogRecPtr  end_lsn;
 	XLogRecPtr  origin_lsn;
+	RepOriginId originId;
 	int         n_records;
 	/* read flags */
 	flags = pq_getmsgbyte(in);
@@ -558,9 +560,21 @@ process_remote_commit(StringInfo in)
 	replorigin_session_origin_lsn = pq_getmsgint64(in); /* commit_lsn */
 	end_lsn = pq_getmsgint64(in); /* end_lsn */
 	replorigin_session_origin_timestamp = pq_getmsgint64(in); /* commit_time */
-	origin_lsn = pq_getmsgint64(in); 
-	Mtm->nodes[MtmReplicationNodeId-1].restartLsn = origin_lsn;
-	
+
+	originId = (RepOriginId)pq_getmsgint(in, 2);
+	origin_lsn = pq_getmsgint64(in);
+
+	if (originId != InvalidRepOriginId) { 
+		for (i = 0; i < Mtm->nAllNodes; i++) { 
+			if (Mtm->nodes[i].originId == originId) { 
+				Mtm->nodes[i].restartLsn = origin_lsn;
+				break;
+			} 
+		}
+		if (i == Mtm->nAllNodes) { 
+			elog(WARNING, "Failed to map origin %d", originId);
+		}
+	}
 	Assert(replorigin_session_origin == InvalidRepOriginId);
 
 	switch(PGLOGICAL_XACT_EVENT(flags))

pglogical_proto.c

@@ -192,6 +192,8 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
     pq_sendint64(out, commit_lsn);
     pq_sendint64(out, txn->end_lsn);
     pq_sendint64(out, txn->commit_time);
+
+	pq_sendint(out, txn->origin_id, 2);
 	pq_sendint64(out, txn->origin_lsn);
 
 	if (txn->xact_action == XLOG_XACT_COMMIT_PREPARED) { 

pglogical_receiver.c

@@ -272,7 +272,7 @@ pglogical_receiver_main(Datum main_arg)
 		}
 
 		query = createPQExpBuffer();
-#if 1 /* Do we need to recreate slot ? */		
+#if 0 /* Do we need to recreate slot ? */		
 		if (mode == REPLMODE_RECOVERED) { /* recreate slot */
 			appendPQExpBuffer(query, "DROP_REPLICATION_SLOT \"%s\"", slotName);
 			res = PQexec(conn, query->data);
@@ -305,7 +305,7 @@ pglogical_receiver_main(Datum main_arg)
 
 		/* Start logical replication at specified position */
 		if (mode == REPLMODE_RECOVERED) {
-			originStartPos = Mtm->nodes[nodeId].restartLsn;
+			originStartPos = Mtm->nodes[nodeId-1].restartLsn;
 		} 
 		if (originStartPos == 0) { 
 			StartTransactionCommand();
@@ -325,6 +325,7 @@ pglogical_receiver_main(Datum main_arg)
 				originStartPos = replorigin_get_progress(originId, false);
 				MTM_LOG1("Restart logical receiver at position %lx with origin=%d from node %d", originStartPos, originId, nodeId);
 			}
+			Mtm->nodes[nodeId-1].originId = originId;
 			CommitTransactionCommand();
 		}