Node recovery

Author: knizhnik

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);

multimaster.c

@@ -155,7 +155,8 @@ char const* const MtmNodeStatusMnem[] =
 	"Offline", 
 	"Connected",
 	"Online",
-	"Recovery"
+	"Recovery",
+	"InMinor"
 };
 
 bool  MtmDoReplication;
@@ -631,10 +632,11 @@ MtmBeginTransaction(MtmCurrentTrans* x)
         x->isReplicated = false;
         x->isDistributed = MtmIsUserTransaction();
 		x->isPrepared = false;
-		if (x->isDistributed && Mtm->status != MTM_ONLINE) { 
+		x->isTransactionBlock = IsTransactionBlock();
+		/* 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;
@@ -981,11 +983,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;
 	}
@@ -1022,7 +1027,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);
 					}
 				}
@@ -1608,8 +1613,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);	
@@ -1622,19 +1628,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() 
@@ -1690,7 +1705,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)

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
 

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;