Node recovery

Author: knizhnik

contrib/mmts/arbiter.c

@@ -348,10 +348,9 @@ static int MtmConnectSocket(char const* host, int port, int max_attempts)
 			/* Some node considered that I am dead, so switch to recovery mode */
 			if (BIT_CHECK(resp.disabledNodeMask, MtmNodeId-1)) { 
 				elog(WARNING, "Node %d think that I am dead", resp.node);
+				BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
 				MtmSwitchClusterMode(MTM_RECOVERY);
 			}
-			/* Combine disable masks from all node. Is it actually correct or we should better check availability of nodes ourselves? */
-			Mtm->disabledNodeMask |= resp.disabledNodeMask;
 			return sd;
 		}
     }
@@ -377,7 +376,7 @@ static void MtmOpenConnections()
 	}
 	if (Mtm->nNodes < MtmNodes/2+1) { /* no quorum */
 		elog(WARNING, "Node is out of quorum: only %d nodes from %d are accssible", Mtm->nNodes, MtmNodes);
-		Mtm->status = MTM_OFFLINE;
+		Mtm->status = MTM_IN_MINORITY;
 	} else if (Mtm->status == MTM_INITIALIZATION) { 
 		MtmSwitchClusterMode(MTM_CONNECTED);
 	}
@@ -431,6 +430,7 @@ static void MtmAcceptOneConnection()
 			resp.dxid = HANDSHAKE_MAGIC;
 			resp.sxid = ShmemVariableCache->nextXid;
 			resp.csn  = MtmGetCurrentTime();
+			resp.node = MtmNodeId;
 			MtmUpdateNodeConnectionInfo(&Mtm->nodes[req.hdr.node-1].con, req.connStr);
 			if (!MtmWriteSocket(fd, &resp, sizeof resp)) { 
 				elog(WARNING, "Arbiter failed to write response for handshake message to node %d", resp.node);

contrib/mmts/multimaster.c

@@ -156,7 +156,8 @@ char const* const MtmNodeStatusMnem[] =
 	"Offline", 
 	"Connected",
 	"Online",
-	"Recovery"
+	"Recovery",
+	"InMinor"
 };
 
 bool  MtmDoReplication;
@@ -633,10 +634,10 @@ MtmBeginTransaction(MtmCurrentTrans* x)
         x->isDistributed = MtmIsUserTransaction();
 		x->isPrepared = false;
 		x->isTransactionBlock = IsTransactionBlock();
-		if (x->isDistributed && Mtm->status != MTM_ONLINE) { 
+		/* Application name can be cahnged usnig PGAPPNAME environment variable */
+		if (x->isDistributed && Mtm->status != MTM_ONLINE && strcmp(application_name, MULTIMASTER_ADMIN) != 0) { 
 			/* reject all user's transactions at offline cluster */
 			MtmUnlock();			
-			Assert(Mtm->status == MTM_ONLINE);
 			elog(ERROR, "Multimaster node is not online: current status %s", MtmNodeStatusMnem[Mtm->status]);
 		}
 		x->containsDML = false;
@@ -983,11 +984,14 @@ bool MtmIsRecoveredNode(int nodeId)
 			 * We have to maintain two bitmasks: one is marking wal sender, another - correspondent nodes. 
 			 * Is there some better way to establish mapping between nodes ad WAL-seconder?
 			 */
+			elog(WARNING,"Node %d is catching up", nodeId);
 			MtmLock(LW_EXCLUSIVE);
 			BIT_SET(Mtm->nodeLockerMask, nodeId-1);
 			BIT_SET(Mtm->walSenderLockerMask, MyWalSnd - WalSndCtl->walsnds);
 			Mtm->nLockers += 1;
 			MtmUnlock();
+		} else { 
+			MTM_INFO("Continue recovery of node %d, slot position %lx, WAL position %lx, lockers %d\n", nodeId, MyWalSnd->sentPtr, GetXLogInsertRecPtr(), Mtm->nLockers);			
 		}
 		return true;
 	}
@@ -1024,7 +1028,7 @@ MtmCheckClusterLock()
 						break;
 					} else { 
 						/* recovered replica catched up with master */
-						elog(WARNING, "WAL-sender %d complete receovery", i);
+						elog(WARNING, "WAL-sender %d complete recovery", i);
 						BIT_CLEAR(Mtm->walSenderLockerMask, i);
 					}
 				}
@@ -1610,8 +1614,9 @@ void MtmReceiverStarted(int nodeId)
 	if (!BIT_CHECK(Mtm->pglogicalNodeMask, nodeId-1)) { 
 		BIT_SET(Mtm->pglogicalNodeMask, nodeId-1);
 		if (++Mtm->nReceivers == Mtm->nNodes-1) {
-			Assert(Mtm->status == MTM_CONNECTED);
-			MtmSwitchClusterMode(MTM_ONLINE);
+			if (Mtm->status == MTM_CONNECTED) { 
+				MtmSwitchClusterMode(MTM_ONLINE);
+			}
 		}
      }
 	SpinLockRelease(&Mtm->spinlock);	
@@ -1624,19 +1629,28 @@ void MtmReceiverStarted(int nodeId)
  */
 MtmSlotMode MtmReceiverSlotMode(int nodeId)
 {
+	bool recovery = false;
 	while (Mtm->status != MTM_CONNECTED && Mtm->status != MTM_ONLINE) { 		
+		MTM_INFO("%d: receiver slot mode %s\n", MyProcPid, MtmNodeStatusMnem[Mtm->status]);
 		if (Mtm->status == MTM_RECOVERY) { 
+			recovery = true;
 			if (Mtm->recoverySlot == 0 || Mtm->recoverySlot == nodeId) { 
 				/* Choose for recovery first available slot */
+				elog(WARNING, "Start recovery from node %d", nodeId);
 				Mtm->recoverySlot = nodeId;
 				return SLOT_OPEN_EXISTED;
 			}
 		}
 		/* delay opening of other slots until recovery is completed */
 		MtmSleep(STATUS_POLL_DELAY);
 	}
+	if (recovery) { 
+		elog(WARNING, "Recreate replication slot for node %d after end of recovery", nodeId);
+	} else { 
+		MTM_INFO("%d: Reuse replication slot for node %d\n", MyProcPid, nodeId);
+	}
 	/* After recovery completion we need to drop all other slots to avoid receive of redundant data */
-	return Mtm->recoverySlot ? SLOT_CREATE_NEW : SLOT_OPEN_ALWAYS;
+	return recovery ? SLOT_CREATE_NEW : SLOT_OPEN_ALWAYS;
 }
 
 static bool MtmIsBroadcast() 
@@ -1692,7 +1706,11 @@ MtmReplicationShutdownHook(struct PGLogicalShutdownHookArgs* args)
 static bool 
 MtmReplicationTxnFilterHook(struct PGLogicalTxnFilterArgs* args)
 {
-	return args->origin_id == InvalidRepOriginId || MtmIsRecoveredNode(MtmReplicationNodeId);
+	bool res = Mtm->status != MTM_RECOVERY
+		&& (args->origin_id == InvalidRepOriginId 
+			|| MtmIsRecoveredNode(MtmReplicationNodeId));
+	MTM_TRACE("%d: MtmReplicationTxnFilterHook->%d\n", MyProcPid, res);
+	return res;
 }
 
 void MtmSetupReplicationHooks(struct PGLogicalHooks* hooks)

contrib/mmts/multimaster.h

@@ -27,6 +27,7 @@
 #define MULTIMASTER_MAX_CONN_STR_SIZE   128
 #define MULTIMASTER_MAX_HOST_NAME_SIZE  64
 #define MULTIMASTER_BROADCAST_SERVICE   "mtm_broadcast"
+#define MULTIMASTER_ADMIN               "mtm_admin"
 
 #define USEC 1000000
 

contrib/mmts/pglogical_proto.c

@@ -103,7 +103,7 @@ pglogical_write_begin(StringInfo out, PGLogicalOutputData *data,
 {
 	bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
 	csn_t csn = MtmTransactionSnapshot(txn->xid);
-	MTM_TRACE("pglogical_write_begin %d CSN=%ld\n", txn->xid, csn);
+	MTM_INFO("%d: pglogical_write_begin %d CSN=%ld\n", MyProcPid, txn->xid, csn);
 
 	if (csn == INVALID_CSN && !isRecovery) { 
 		MtmIsFilteredTxn = true;

src/backend/replication/slot.c

@@ -248,10 +248,11 @@ ReplicationSlotCreate(const char *name, bool db_specific,
 	{
 		ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
-		if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
+		if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0) {
 			ereport(ERROR,
 					(errcode(ERRCODE_DUPLICATE_OBJECT),
 					 errmsg("replication slot \"%s\" already exists", name)));
+		}
 		if (!s->in_use && slot == NULL)
 			slot = s;
 	}