Set proper restartLsn for all origens on start

kelvich · kelvich · commit bc9dc445d6ff · 2017-11-13T02:23:41.000+03:00
diff --git a/pglogical_receiver.c b/pglogical_receiver.c
@@ -217,15 +217,14 @@ pglogical_receiver_main(Datum main_arg)
 	MtmReplicationMode mode;
 
 	ByteBuffer buf;
-	RepOriginId origenId;
-	char* origenName;
 	/* Buffer for COPY data */
 	char	*copybuf = NULL;
 	int spill_file = -1;
 	StringInfoData spill_info;
 	char *slotName;
 	char* connString = psprintf("replication=database %s", Mtm->nodes[nodeId-1].con.connStr);
 	static PortalData fakePortal;
+	int i;
 
 	MtmBackgroundWorker = true;
 
@@ -258,16 +257,27 @@ pglogical_receiver_main(Datum main_arg)
 	ActivePortal->status = PORTAL_ACTIVE;
 	ActivePortal->sourceText = "";
 
-	/* Create origenid */
-	StartTransactionCommand();
-	origenName = psprintf(MULTIMASTER_SLOT_PATTERN, nodeId);
-	origenId = replorigen_by_name(origenName, true);
-	if (origenId == InvalidRepOriginId) {
-		origenId = replorigen_create(origenName);
+	/*
+	 * Set proper restartLsn for all origens
+	 */
+	MtmLock(LW_EXCLUSIVE);
+	for (i = 0; i < Mtm->nAllNodes; i++)
+	{
+		char	   *origenName;
+		RepOriginId origenId;
+
+		StartTransactionCommand();
+		origenName = psprintf(MULTIMASTER_SLOT_PATTERN, i + 1);
+		origenId = replorigen_by_name(origenName, true);
+		if (origenId == InvalidRepOriginId) {
+			origenId = replorigen_create(origenName);
+		}
+		CommitTransactionCommand();
+		if (Mtm->nodes[i].restartLSN == INVALID_LSN)
+			Mtm->nodes[i].restartLSN = replorigen_get_progress(origenId, true);
+		Mtm->nodes[i].origenId = origenId;
 	}
-	CommitTransactionCommand();
-	Mtm->nodes[nodeId-1].origenId = origenId;
-	Mtm->nodes[nodeId-1].restartLSN = INVALID_LSN;
+	MtmUnlock();
 
 	/* This is main loop of logical replication.
 	 * In case of errors we will try to reestablish connection.
@@ -277,7 +287,7 @@ pglogical_receiver_main(Datum main_arg)
 	{
 		int	 count;
 		ConnStatusType status;
-		lsn_t origenStartPos = Mtm->nodes[nodeId-1].restartLSN;
+		lsn_t origenStartPos;
 		int timeline;
 
 		/*
@@ -308,7 +318,7 @@ pglogical_receiver_main(Datum main_arg)
 		query = createPQExpBuffer();
 
 		/* Start logical replication at specified position */
-		origenStartPos = replorigen_get_progress(origenId, false);
+		origenStartPos = replorigen_get_progress(Mtm->nodes[nodeId-1].origenId, false);
 		if (origenStartPos == INVALID_LSN || Mtm->nodes[nodeId-1].manualRecovery) {
 			/*
 			 * We are just creating new replication slot.
@@ -337,7 +347,7 @@ pglogical_receiver_main(Datum main_arg)
 				MTM_LOG1("Advance restartLSN for node %d: from %llx to %llx (pglogical_receiver_main)", nodeId, Mtm->nodes[nodeId-1].restartLSN, origenStartPos);
 				Mtm->nodes[nodeId-1].restartLSN = origenStartPos;
 			}
-			MTM_LOG1("Restart logical receiver at position %llx with origen=%d from node %d", origenStartPos, origenId, nodeId);
+			MTM_LOG1("Restart logical receiver at position %llx from node %d", origenStartPos, nodeId);
 		}
 
 		MTM_LOG1("Start replication on slot %s from node %d at position %llx, mode %s, recovered lsn %llx",
