Do not delete slots upon reconnect

Author: kelvich

contrib/mmts/multimaster.c

@@ -772,6 +772,11 @@ MtmAdjustOldestXid(TransactionId xid)
 			Mtm->transListHead = prev;
 		}
 	}
+
+	if (!MyReplicationSlot) {
+		MtmCheckSlots();
+	}
+
 	return xid;
 }
 
@@ -1551,9 +1556,6 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 	MtmUnlock();
 
 	MtmResetTransaction();
-	if (!MyReplicationSlot) {
-		MtmCheckSlots();
-	}
 	if (MtmClusterLocked) {
 		MtmUnlockCluster();
 	}
@@ -2728,6 +2730,7 @@ static void MtmInitialize()
 			Mtm->nodes[i].originId = InvalidRepOriginId;
 			Mtm->nodes[i].timeline = 0;
 			Mtm->nodes[i].nHeartbeats = 0;
+			Mtm->nodes[i].slotDeleted = false;
 		}
 		Mtm->nodes[MtmNodeId-1].originId = DoNotReplicateId;
 		/* All transaction originated from the current node should be ignored during recovery */
@@ -3791,8 +3794,8 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 				if (Mtm->nodes[MtmReplicationNodeId-1].restartLSN < recoveredLSN) {
 					MTM_LOG1("Advance restartLSN for node %d from %llx to %llx (MtmReplicationStartupHook)",
 							 MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, recoveredLSN);
-					Assert(Mtm->nodes[MtmReplicationNodeId-1].restartLSN == INVALID_LSN
-						   || recoveredLSN < Mtm->nodes[MtmReplicationNodeId-1].restartLSN + MtmMaxRecoveryLag);
+					// Assert(Mtm->nodes[MtmReplicationNodeId-1].restartLSN == INVALID_LSN
+					// 	   || recoveredLSN < Mtm->nodes[MtmReplicationNodeId-1].restartLSN + MtmMaxRecoveryLag);
 					Mtm->nodes[MtmReplicationNodeId-1].restartLSN = recoveredLSN;
 				}
 			} else {

contrib/mmts/multimaster.h

@@ -235,6 +235,9 @@ typedef struct
 	int         lockGraphAllocated;
 	int         lockGraphUsed;
 	uint64      nHeartbeats;
+	bool		slotDeleted;			/* Signalizes that node is already deleted our slot and
+										 * recovery from that node isn't possible.
+										 */
 } MtmNodeInfo;
 
 typedef struct MtmL2List

contrib/mmts/pglogical_receiver.c

@@ -299,23 +299,16 @@ pglogical_receiver_main(Datum main_arg)
 		}
 
 		query = createPQExpBuffer();
-		if ((mode == REPLMODE_OPEN_EXISTED && timeline != Mtm->nodes[nodeId-1].timeline)
-			|| mode == REPLMODE_CREATE_NEW)
-		{ /* recreate slot */
-			timestamp_t start = MtmGetSystemTime();
-			appendPQExpBuffer(query, "DROP_REPLICATION_SLOT \"%s\"", slotName);
-			res = PQexec(conn, query->data);
-			elog(LOG, "Drop replication slot %s: %ld milliseconds", slotName, (long)USEC_TO_MSEC(MtmGetSystemTime() - start));
-			PQclear(res);
-			resetPQExpBuffer(query);
-			timeline = Mtm->nodes[nodeId-1].timeline;
-		}
-		/* 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) */
-		{
-			timestamp_t start = MtmGetSystemTime();
+
+		/* Start logical replication at specified position */
+		originStartPos = replorigin_get_progress(originId, false);
+		if (originStartPos == INVALID_LSN) {
+			/*
+			 * We are just creating new replication slot.
+			 * It is assumed that state of local and remote nodes is the same at this moment.
+			 * They are either empty, either new node is synchronized using base_backup.
+			 * So we assume that LSNs are the same for local and remote node
+			 */
 			appendPQExpBuffer(query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL \"%s\"", slotName, MULTIMASTER_NAME);
 			res = PQexec(conn, query->data);
 			if (PQresultStatus(res) != PGRES_TUPLES_OK)
@@ -329,30 +322,14 @@ pglogical_receiver_main(Datum main_arg)
 					goto OnError;
 				}
 			}
-			elog(LOG, "Recreate replication slot %s: %ld milliseconds", slotName, (long)USEC_TO_MSEC(MtmGetSystemTime() - start));
 			PQclear(res);
 			resetPQExpBuffer(query);
-		}
-
-		/* Start logical replication at specified position */
-		if (originStartPos == INVALID_LSN) {
-			originStartPos = replorigin_get_progress(originId, false);
-			if (originStartPos == INVALID_LSN) {
-				/*
-				 * We are just creating new replication slot.
-				 * It is assumed that state of local and remote nodes is the same at this moment.
-				 * 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 = INVALID_LSN;
-				MTM_LOG1("Start logical receiver at position %llx from node %d", originStartPos, nodeId);
-			} else {
-				if (Mtm->nodes[nodeId-1].restartLSN < originStartPos) {
-					MTM_LOG1("Advance restartLSN for node %d: from %llx to %llx (pglogical_receiver_main)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
-					Mtm->nodes[nodeId-1].restartLSN = originStartPos;
-				}
-				MTM_LOG1("Restart logical receiver at position %llx with origin=%d from node %d", originStartPos, originId, nodeId);
+		} else {
+			if (Mtm->nodes[nodeId-1].restartLSN < originStartPos) {
+				MTM_LOG1("Advance restartLSN for node %d: from %llx to %llx (pglogical_receiver_main)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
+				Mtm->nodes[nodeId-1].restartLSN = originStartPos;
 			}
+			MTM_LOG1("Restart logical receiver at position %llx with origin=%d from node %d", originStartPos, originId, nodeId);
 		}
 
 		MTM_LOG1("Start replication on slot %s from node %d at position %llx, mode %s, recovered lsn %llx",
@@ -371,10 +348,21 @@ pglogical_receiver_main(Datum main_arg)
 		res = PQexec(conn, query->data);
 		if (PQresultStatus(res) != PGRES_COPY_BOTH)
 		{
-			PQclear(res);
-			ereport(WARNING, (MTM_ERRMSG("%s: Could not start logical replication",
-								 worker_proc)));
-			goto OnError;
+			int i, n_deleted_slots = 0;
+
+			elog(WARNING, "Can't find slot on node%d. Shutting down receiver.", nodeId);
+			Mtm->nodes[nodeId-1].slotDeleted = true;
+			for (i = 0; i < Mtm->nAllNodes; i++)
+			{
+				if (Mtm->nodes[i].slotDeleted)
+					n_deleted_slots++;
+			}
+			if (n_deleted_slots == Mtm->nAllNodes - 1)
+			{
+				elog(WARNING, "All neighbour nopes have no replication slot for us. Exiting.");
+				kill(PostmasterPid, SIGTERM);
+			}
+			proc_exit(1);
 		}
 		PQclear(res);
 		resetPQExpBuffer(query);

contrib/mmts/t/004_recovery.pl

@@ -14,18 +14,18 @@
 $cluster->pgbench(0, ('-i', -s => '10') );
 
 # kill node while neighbour is under load
-my $pgb_handle = $cluster->pgbench_async(0, ('-N', -T => '10') );
+my $pgb_handle = $cluster->pgbench_async(1, ('-N', -T => '10') );
 sleep(5);
 $cluster->{nodes}->[2]->stop('fast');
 $cluster->pgbench_await($pgb_handle);
 
 # start node while neighbour is under load
-$pgb_handle = $cluster->pgbench_async(0, ('-N', -T => '10') );
-sleep(5);
+$pgb_handle = $cluster->pgbench_async(0, ('-N', -T => '50') );
+sleep(10);
 $cluster->{nodes}->[2]->start;
 $cluster->pgbench_await($pgb_handle);
 
-# give it 10s to recover
+# give it extra 10s to recover
 sleep(10);
 
 # check data identity