Recovery fixes

Author: knizhnik

arbiter.c

@@ -247,14 +247,13 @@ static bool MtmWriteSocket(int sd, void const* buf, int size)
     while (size != 0) {
 		int rc = MtmWaitSocket(sd, true, MtmHeartbeatSendTimeout);
 		if (rc == 1) { 
-			int n = send(sd, src, size, 0);
-			if (n < 0) {
-				Assert(errno != EINTR); /* should not happen in non-blocking call */
+			while ((rc = send(sd, src, size, 0)) < 0 && errno == EINTR);			
+			if (rc < 0) {
 				busy_socket = -1;
 				return false;
 			}
-			size -= n;
-			src += n;
+			size -= rc;
+			src += rc;
 		} else if (rc < 0) { 
 			busy_socket = -1;
 			return false;
@@ -266,15 +265,12 @@ static bool MtmWriteSocket(int sd, void const* buf, int size)
 
 static int MtmReadSocket(int sd, void* buf, int buf_size)
 {
-	int rc = recv(sd, buf, buf_size, 0);
+	int rc;
+	while ((rc = recv(sd, buf, buf_size, 0)) < 0 && errno == EINTR);			
 	if (rc < 0 && errno == EAGAIN) { 
 		rc = MtmWaitSocket(sd, false, MtmHeartbeatSendTimeout);
 		if (rc == 1) { 
-			rc = recv(sd, buf, buf_size, 0);
-			if (rc < 0) { 
-				Assert(errno != EINTR); /* should not happen in non-blocking call */
-				return -1;
-			}
+			while ((rc = recv(sd, buf, buf_size, 0)) < 0 && errno == EINTR);			
 		} else { 
 			return 0;
 		}
@@ -370,12 +366,13 @@ static void MtmSendHeartbeat()
 	for (i = 0; i < Mtm->nAllNodes; i++)
 	{
 		if (i+1 != MtmNodeId && sockets[i] != busy_socket 
-			&& ((sockets[i] >= 0 && !BIT_CHECK(Mtm->disabledNodeMask, i)) || BIT_CHECK(Mtm->reconnectMask, i)))
+			&& (Mtm->status != MTM_ONLINE 
+				|| (sockets[i] >= 0 && !BIT_CHECK(Mtm->disabledNodeMask, i) && !BIT_CHECK(Mtm->reconnectMask, i))))
 		{ 
 			if (!MtmSendToNode(i, &msg, sizeof(msg))) { 
 				elog(LOG, "Arbiter failed to send heartbeat to node %d", i+1);
 			} else {
-				MTM_LOG1("Send heartbeat to node %d with timestamp %ld", i+1, now);    
+				MTM_LOG2("Send heartbeat to node %d with timestamp %ld", i+1, now);    
 			}
 		}
 	}
@@ -593,8 +590,9 @@ static void MtmAcceptOneConnection()
 		} else if (req.hdr.code != MSG_HANDSHAKE && req.hdr.dxid != HANDSHAKE_MAGIC) { 
 			elog(WARNING, "Arbiter get unexpected handshake message %d", req.hdr.code);
 			close(fd);
-		} else{ 			
-			Assert(req.hdr.node > 0 && req.hdr.node <= Mtm->nAllNodes && req.hdr.node != MtmNodeId);
+		} else { 
+			int node = req.hdr.node-1;
+			Assert(node >= 0 && node < Mtm->nAllNodes && node+1 != MtmNodeId);
 
 			MtmLock(LW_EXCLUSIVE);
 			MtmCheckResponse(&req.hdr);
@@ -606,15 +604,18 @@ static void MtmAcceptOneConnection()
 			resp.sxid = ShmemVariableCache->nextXid;
 			resp.csn  = MtmGetCurrentTime();
 			resp.node = MtmNodeId;
-			MtmUpdateNodeConnectionInfo(&Mtm->nodes[req.hdr.node-1].con, req.connStr);
+			MtmUpdateNodeConnectionInfo(&Mtm->nodes[node].con, req.connStr);
 			if (!MtmWriteSocket(fd, &resp, sizeof resp)) { 
-				elog(WARNING, "Arbiter failed to write response for handshake message to node %d", resp.node);
+				elog(WARNING, "Arbiter failed to write response for handshake message to node %d", node+1);
 				close(fd);
 			} else { 
-				MTM_LOG1("Arbiter established connection with node %d", req.hdr.node); 
-				MtmRegisterSocket(fd, req.hdr.node-1);
-				sockets[req.hdr.node-1] = fd;
-				MtmOnNodeConnect(req.hdr.node);
+				MTM_LOG1("Arbiter established connection with node %d", node+1); 
+				if (sockets[node] >= 0) { 
+					MtmUnregisterSocket(sockets[node]);
+				}
+				sockets[node] = fd;
+				MtmRegisterSocket(fd, node);
+				MtmOnNodeConnect(node+1);
 			}
 		}
 	}
@@ -889,7 +890,7 @@ static void MtmTransReceiver(Datum arg)
 					Mtm->nodes[msg->node-1].lastHeartbeat = MtmGetSystemTime();
 
 					if (msg->code == MSG_HEARTBEAT) {
-						MTM_LOG1("Receive HEARTBEAT from node %d with timestamp %ld delay %ld", 
+						MTM_LOG2("Receive HEARTBEAT from node %d with timestamp %ld delay %ld", 
 								 msg->node, msg->csn, USEC_TO_MSEC(MtmGetSystemTime() - msg->csn)); 
 						continue;
 					}
@@ -1002,21 +1003,23 @@ static void MtmTransReceiver(Datum arg)
 				}
 			}
 		}
-		now = MtmGetSystemTime();
-		if (now > lastHeartbeatCheck + MSEC_TO_USEC(MtmHeartbeatRecvTimeout)) { 
-			if (!MtmWatchdog(stopPolling)) { 
-				for (i = 0; i < nNodes; i++) { 
-					if (Mtm->nodes[i].lastHeartbeat != 0 && sockets[i] >= 0) {
-						MTM_LOG1("Last hearbeat from node %d received %ld microseconds ago", i+1, now - Mtm->nodes[i].lastHeartbeat);
+		if (Mtm->status != MTM_RECOVERY) { 
+			now = MtmGetSystemTime();
+			if (now > lastHeartbeatCheck + MSEC_TO_USEC(MtmHeartbeatRecvTimeout)) { 
+				if (!MtmWatchdog(stopPolling)) { 
+					for (i = 0; i < nNodes; i++) { 
+						if (Mtm->nodes[i].lastHeartbeat != 0 && sockets[i] >= 0) {
+							MTM_LOG1("Last hearbeat from node %d received %ld microseconds ago", i+1, now - Mtm->nodes[i].lastHeartbeat);
+						}
 					}
+					MTM_LOG1("epoll started %ld and finished %ld microseconds ago", now - startPolling, now - stopPolling);
 				}
-				MTM_LOG1("epoll started %ld and finished %ld microseconds ago", now - startPolling, now - stopPolling);
+				lastHeartbeatCheck = now;
+			}
+			if (n == 0 && Mtm->disabledNodeMask != 0) { 
+				/* If timeout is expired and there are disabled nodes, then recheck cluster's state */
+				MtmRefreshClusterStatus(false);
 			}
-			lastHeartbeatCheck = now;
-		}
-		if (n == 0 && Mtm->disabledNodeMask != 0) { 
-			/* If timeout is expired and there are disabled nodes, then recheck cluster's state */
-			MtmRefreshClusterStatus(false);
 		}
 	}
 	proc_exit(1); /* force restart of this bgwroker */

multimaster.c

@@ -742,7 +742,6 @@ MtmCreateTransState(MtmCurrentTrans* x)
 		/* I am coordinator of transaction */
 		ts->gtid.xid = x->xid;
 		ts->gtid.node = MtmNodeId;
-		//ts->gid[0] = '\0';
 		strcpy(ts->gid, x->gid);
 	}
 	return ts;
@@ -1186,26 +1185,33 @@ static void MtmDisableNode(int nodeId)
 	BIT_SET(Mtm->disabledNodeMask, nodeId-1);
 	Mtm->nodes[nodeId-1].lastStatusChangeTime = MtmGetSystemTime();
 	Mtm->nodes[nodeId-1].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
-	Mtm->nLiveNodes -= 1;			
-}
+	if (nodeId != MtmNodeId) { 
+		Mtm->nLiveNodes -= 1;
+	}
+	elog(WARNING, "Disable node %d at xlog position %lx", nodeId, GetXLogInsertRecPtr());
+} 
 
 static void MtmEnableNode(int nodeId)
 { 
 	BIT_CLEAR(Mtm->disabledNodeMask, nodeId-1);
 	BIT_CLEAR(Mtm->reconnectMask, nodeId-1);
 	Mtm->nodes[nodeId-1].lastStatusChangeTime = MtmGetSystemTime();
 	Mtm->nodes[nodeId-1].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
-	Mtm->nLiveNodes += 1;			
+	if (nodeId != MtmNodeId) { 
+		Mtm->nLiveNodes += 1;			
+	}
+	elog(WARNING, "Enable node %d at xlog position %lx", nodeId, GetXLogInsertRecPtr());
 }
 
 void MtmRecoveryCompleted(void)
 {
-	MTM_LOG1("Recovery of node %d is completed", MtmNodeId);
+	MTM_LOG1("Recovery of node %d is completed, disabled mask=%lx, reconnect mask=%ld, live nodes=%d", 
+			 MtmNodeId, Mtm->disabledNodeMask, Mtm->reconnectMask, Mtm->nLiveNodes);
 	MtmLock(LW_EXCLUSIVE);
 	Mtm->recoverySlot = 0;
-	BIT_CLEAR(Mtm->disabledNodeMask, MtmNodeId-1);
 	Mtm->nodes[MtmNodeId-1].lastStatusChangeTime = MtmGetSystemTime();
-	/* Mode will be changed to online once all locagical reciever are connected */
+	BIT_CLEAR(Mtm->disabledNodeMask, MtmNodeId-1);
+	/* Mode will be changed to online once all logical reciever are connected */
 	MtmSwitchClusterMode(MTM_CONNECTED);
 	MtmUnlock();
 }
@@ -1464,16 +1470,16 @@ bool MtmRefreshClusterStatus(bool nowait)
 				MtmEnableNode(i+1);
 			}
 		}
-#endif
 		Mtm->reconnectMask |= clique & Mtm->disabledNodeMask; /* new enabled nodes mask */		
+#endif
 
 		if (disabled) { 
 			MtmCheckQuorum();
 		}
 		/* Interrupt voting for active transaction and abort them */
 		for (ts = Mtm->transListHead; ts != NULL; ts = ts->next) { 
 			MTM_LOG3("Active transaction gid='%s', coordinator=%d, xid=%d, status=%d, gtid.xid=%d",
-					 ts->gid, ts->gtid.node, ts->xid, ts->status, ts->gtid.xid);
+					 ts->gid, ts->gtid.nхode, ts->xid, ts->status, ts->gtid.xid);
 			if (MtmIsCoordinator(ts)) { 
 				if (!ts->votingCompleted && disabled != 0 && ts->status != TRANSACTION_STATUS_ABORTED) {
 					MtmAbortTransaction(ts);
@@ -1728,6 +1734,7 @@ static void MtmInitialize()
 		Mtm->transCount = 0;
 		Mtm->gcCount = 0;
 		Mtm->nConfigChanges = 0;
+		Mtm->recoveryCount = 0;
 		Mtm->localTablesHashLoaded = false;
 		Mtm->inject2PCError = 0;
 		for (i = 0; i < MtmNodes; i++) {
@@ -2271,6 +2278,9 @@ void MtmReceiverStarted(int nodeId)
 	MtmLock(LW_EXCLUSIVE);
 	if (!BIT_CHECK(Mtm->pglogicalNodeMask, nodeId-1)) { 
 		BIT_SET(Mtm->pglogicalNodeMask, nodeId-1);
+		if (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) {
+			MtmEnableNode(nodeId);
+		}
 		if (++Mtm->nReceivers == Mtm->nLiveNodes-1) {
 			if (Mtm->status == MTM_CONNECTED) { 
 				MtmSwitchClusterMode(MTM_ONLINE);
@@ -2296,6 +2306,9 @@ MtmSlotMode MtmReceiverSlotMode(int nodeId)
 				/* Choose for recovery first available slot */
 				MTM_LOG1("Start recovery from node %d", nodeId);
 				Mtm->recoverySlot = nodeId;
+				Mtm->nReceivers = 0;
+				Mtm->recoveryCount += 1;
+				Mtm->pglogicalNodeMask = 0;
 				FinishAllPreparedTransactions(false);
 				return SLOT_OPEN_EXISTED;
 			}

multimaster.h

@@ -174,7 +174,7 @@ typedef struct
 	LWLockPadded *locks;               /* multimaster lock tranche */
 	TransactionId oldestXid;           /* XID of oldest transaction visible by any active transaction (local or global) */
 	nodemask_t disabledNodeMask;       /* bitmask of disabled nodes */
-	nodemask_t connectivityMask;       /* bitmask of dicconnected nodes */
+	nodemask_t connectivityMask;       /* bitmask of disconnected nodes */
 	nodemask_t pglogicalNodeMask;      /* bitmask of started pglogic receivers */
 	nodemask_t walSenderLockerMask;    /* Mask of WAL-senders IDs locking the cluster */
 	nodemask_t nodeLockerMask;         /* Mask of node IDs which WAL-senders are locking the cluster */
@@ -188,6 +188,7 @@ typedef struct
 	int    nLockers;                   /* Number of lockers */
 	int    nActiveTransactions;        /* Nunmber of active 2PC transactions */
 	int    nConfigChanges;             /* Number of cluster configuration changes */
+	int    recoveryCount;              /* Number of completed recoveries */
 	int64  timeShift;                  /* Local time correction */
 	csn_t  csn;                        /* Last obtained timestamp: used to provide unique acending CSNs based on system time */
 	csn_t  lastCsn;                    /* CSN of last committed transaction */

pglogical_apply.c

@@ -492,7 +492,7 @@ static void
 MtmEndSession(void)
 {
 	if (replorigin_session_origin != InvalidRepOriginId) { 
-		MTM_LOG3("%d: Begin reset replorigin session: %d", MyProcPid, replorigin_session_origin);
+		MTM_LOG2("%d: Begin reset replorigin session for node %d: %d, progress %lx", MyProcPid, MtmReplicationNodeId, replorigin_session_origin, replorigin_session_get_progress(false));
 		replorigin_session_origin = InvalidRepOriginId;
 		replorigin_session_reset();
 		if (unlock) { 
@@ -568,7 +568,7 @@ process_remote_commit(StringInfo in)
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			csn = pq_getmsgint64(in); 
 			gid = pq_getmsgstring(in);
-			MTM_LOG2("PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s", csn, gid);
+			MTM_LOG2("PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s, lsn=%ld", csn, gid, end_lsn);
 			StartTransactionCommand();
 			MtmBeginSession();
 			MtmSetCurrentTransactionCSN(csn);
@@ -585,6 +585,7 @@ process_remote_commit(StringInfo in)
 			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED) == TRANSACTION_STATUS_UNKNOWN) { 
 				MTM_LOG1("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
 				StartTransactionCommand();
+				MtmBeginSession();
 				MtmSetCurrentTransactionGID(gid);
 				FinishPreparedTransaction(gid, false);
 				CommitTransactionCommand();
@@ -594,6 +595,12 @@ process_remote_commit(StringInfo in)
 		default:
 			Assert(false);
 	}
+#if 0 /* Do ont need to advance slot position here: it will be done by transaction commit */
+	if (replorigin_session_origin != InvalidRepOriginId) { 
+		replorigin_advance(replorigin_session_origin, end_lsn,
+						   XactLastCommitEnd, false, false);
+	}
+#endif
 	MtmEndSession(true);
 	MtmUpdateLsnMapping(MtmReplicationNodeId, end_lsn);
 	if (flags & PGLOGICAL_CAUGHT_UP) {
@@ -936,6 +943,11 @@ void MtmExecutor(int id, void* work, size_t size)
         while (true) { 
             char action = pq_getmsgbyte(&s);
             MTM_LOG3("%d: REMOTE process action %c", MyProcPid, action);
+#if 0
+			if (Mtm->status == MTM_RECOVERY) { 
+				MTM_LOG1("Replay action %c[%x]",   action, s.data[s.cursor]);
+			}
+#endif
             switch (action) {
                 /* BEGIN */
             case 'B':

pglogical_proto.c

@@ -424,7 +424,8 @@ decide_datum_transfer(Form_pg_attribute att, Form_pg_type typclass,
 PGLogicalProtoAPI *
 pglogical_init_api(PGLogicalProtoType typ)
 {
-    PGLogicalProtoAPI* res = palloc0(sizeof(PGLogicalProtoAPI));
+    PGLogicalProtoAPI* res = malloc(sizeof(PGLogicalProtoAPI));
+	MemSet(res, 0, sizeof(PGLogicalProtoAPI));
 	sscanf(MyReplicationSlot->data.name.data, MULTIMASTER_SLOT_PATTERN, &MtmReplicationNodeId);
 	MTM_LOG1("%d: PRGLOGICAL init API for slot %s node %d", MyProcPid, MyReplicationSlot->data.name.data, MtmReplicationNodeId);
     res->write_rel = pglogical_write_rel;

pglogical_receiver.c

@@ -244,12 +244,15 @@ pglogical_receiver_main(Datum main_arg)
 	 */
 	while (!got_sigterm)
 	{ 
+		int  count;
+
 		/* 
 		 * Determine when and how we should open replication slot.
 		 * Druing recovery we need to open only one replication slot from which node should receive all transactions.
 		 * Slots at other nodes should be removed 
 		 */
 		mode = MtmReceiverSlotMode(nodeId);	
+		count = Mtm->recoveryCount;
 
 		/* Establish connection to remote server */
 		conn = PQconnectdb(connString);
@@ -303,7 +306,7 @@ pglogical_receiver_main(Datum main_arg)
 			MTM_LOG1("Start logical receiver at position %lx from node %d", originStartPos, nodeId);
 		} else { 
 			originStartPos = replorigin_get_progress(originId, false);
-			MTM_LOG1("Restart logical receiver at position %lx from node %d", originStartPos, nodeId);
+			MTM_LOG1("Restart logical receiver at position %lx with origin=%d from node %d", originStartPos, originId, nodeId);
 		}
 		CommitTransactionCommand();
 
@@ -359,7 +362,12 @@ pglogical_receiver_main(Datum main_arg)
 
 			if (Mtm->status == MTM_OFFLINE || (Mtm->status == MTM_RECOVERY && Mtm->recoverySlot != nodeId)) 
 			{
-				ereport(LOG, (errmsg("%s: suspending WAL receiver because node was switched to %s mode", worker_proc, MtmNodeStatusMnem[Mtm->status])));
+				ereport(LOG, (errmsg("%s: restart WAL receiver because node was switched to %s mode", worker_proc, MtmNodeStatusMnem[Mtm->status])));
+				break;
+			}
+			if (count != Mtm->recoveryCount) { 
+				
+				ereport(LOG, (errmsg("%s: restart WAL receiver because node was recovered", worker_proc)));
 				break;
 			}