Avoid restart of server to perform recovery

Author: knizhnik

Dockerfile

@@ -31,7 +31,7 @@ USER postgres
 ENV CFLAGS -O0
 WORKDIR /pg
 
-ENV REBUILD 5
+ENV REBUILD 6
 
 RUN cd /pg && \
 	git clone https://github.com/postgrespro/postgres_cluster.git --depth 1 && \

arbiter.c

@@ -195,7 +195,7 @@ static void MtmRegisterSocket(int fd, int node)
     ev.events = EPOLLIN;
     ev.data.u32 = node;        
     if (epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &ev) < 0) {
-        elog(ERROR, "Arbiter failed to add socket to epoll set: %d", errno);
+        elog(LOG, "Arbiter failed to add socket to epoll set: %d", errno);
     } 
 #else
     FD_SET(fd, &inset);    
@@ -209,7 +209,7 @@ static void MtmUnregisterSocket(int fd)
 {
 #if USE_EPOLL
     if (epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, NULL) < 0) { 
-		elog(ERROR, "Arbiter failed to unregister socket from epoll set: %d", errno);
+		elog(LOG, "Arbiter failed to unregister socket from epoll set: %d", errno);
     } 
 #else
 	FD_CLR(fd, &inset); 
@@ -266,12 +266,17 @@ 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);
-	if (rc <= 0) { 
-		Assert(errno != EINTR); /* should not happen in non-blocking call */
-		return -1;
+	int rc = MtmWaitSocket(sd, false, MtmHeartbeatSendTimeout);
+	if (rc == 1) { 
+		int rc = recv(sd, buf, buf_size, 0);
+		if (rc <= 0) { 
+			Assert(errno != EINTR); /* should not happen in non-blocking call */
+			return -1;
+		}
+		return rc;
+	} else { 
+		return 0;
 	}
-	return rc;
 }
 
 
@@ -325,7 +330,16 @@ static void MtmSetSocketOptions(int sd)
 #endif
 }
 
-
+static void MtmCheckResponse(MtmArbiterMessage* resp)
+{
+	if (BIT_CHECK(resp->disabledNodeMask, MtmNodeId-1) && !BIT_CHECK(Mtm->disabledNodeMask, resp->node-1)) { 
+		elog(WARNING, "Node %d thinks that I was dead, while I am %s", resp->node, MtmNodeStatusMnem[Mtm->status]);
+		if (Mtm->status != MTM_RECOVERY) { 
+			BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
+			MtmSwitchClusterMode(MTM_RECOVERY);
+		}
+	}
+}
 
 static void MtmScheduleHeartbeat()
 {
@@ -347,7 +361,8 @@ static void MtmSendHeartbeat()
 
 	for (i = 0; i < Mtm->nAllNodes; i++)
 	{
-		if (sockets[i] >= 0 && sockets[i] != busy_socket && !BIT_CHECK(Mtm->disabledNodeMask|Mtm->reconnectMask, i))
+		if (i+1 != MtmNodeId && sockets[i] != busy_socket 
+			&& ((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);
@@ -382,7 +397,8 @@ static int MtmConnectSocket(int node, int port, int timeout)
 	sock_inet.sin_port = htons(port);
 
 	if (!MtmResolveHostByName(host, addrs, &n_addrs)) {
-		elog(ERROR, "Arbiter failed to resolve host '%s' by name", host);
+		elog(LOG, "Arbiter failed to resolve host '%s' by name", host);
+		return -1;
 	}
 
   Retry:
@@ -391,11 +407,13 @@ static int MtmConnectSocket(int node, int port, int timeout)
 
 		sd = socket(AF_INET, SOCK_STREAM, 0);
 		if (sd < 0) {
-			elog(ERROR, "Arbiter failed to create socket: %d", errno);
+			elog(LOG, "Arbiter failed to create socket: %d", errno);
+			return -1;
 		}
 		rc = fcntl(sd, F_SETFL, O_NONBLOCK);
 		if (rc < 0) {
-			elog(ERROR, "Arbiter failed to switch socket to non-blocking mode: %d", errno);
+			elog(LOG, "Arbiter failed to switch socket to non-blocking mode: %d", errno);
+			return -1;
 		}
 		busy_socket = sd;
 		for (i = 0; i < n_addrs; ++i) {
@@ -463,14 +481,7 @@ static int MtmConnectSocket(int node, int port, int timeout)
 	}
 
 	MtmLock(LW_EXCLUSIVE);
-
-	/* Some node considered that I am dead, so switch to recovery mode */
-	if (BIT_CHECK(resp.disabledNodeMask, MtmNodeId-1)) { 
-		elog(WARNING, "Node %d thinks that I was dead", resp.node);
-		BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
-		MtmRollbackAllPreparedTransactions();
-		MtmSwitchClusterMode(MTM_RECOVERY);
-	}
+	MtmCheckResponse(&resp);
 	MtmUnlock();
 
 	return sd;
@@ -493,7 +504,7 @@ static void MtmOpenConnections()
 			char const* arbiterPortStr = strstr(Mtm->nodes[i].con.connStr, "arbiterport=");
 			if (arbiterPortStr != NULL) {
 				if (sscanf(arbiterPortStr+12, "%d", &arbiterPort) != 1) {
-					elog(ERROR, "Invalid arbiter port: %s", arbiterPortStr+12);
+					elog(ERROR, "Invalid arbiter port: %s", arbiterPortStr+12);					
 				}
 			} else { 
 				arbiterPort = MtmArbiterPort + i + 1;
@@ -518,11 +529,13 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 	while (true) {
 		if (sockets[node] >= 0 && BIT_CHECK(Mtm->reconnectMask, node)) {
 			elog(WARNING, "Arbiter is forced to reconnect to node %d", node+1); 
+			close(sockets[node]);
+			sockets[node] = -1;
+		}
+		if (BIT_CHECK(Mtm->reconnectMask, node)) {
 			MtmLock(LW_EXCLUSIVE);		
 			BIT_CLEAR(Mtm->reconnectMask, node);
 			MtmUnlock();
-			close(sockets[node]);
-			sockets[node] = -1;
 		}
 		if (sockets[node] < 0 || !MtmWriteSocket(sockets[node], buf, size)) { 
 			if (sockets[node] >= 0) { 
@@ -535,9 +548,6 @@ static bool MtmSendToNode(int node, void const* buf, int size)
 				MtmOnNodeDisconnect(node+1);
 				return false;
 			}
-			MtmLock(LW_EXCLUSIVE);		
-			BIT_CLEAR(Mtm->reconnectMask, node);
-			MtmUnlock();
 			MTM_LOG3("Arbiter restablished connection with node %d", node+1);
 		} else { 
 			return true;
@@ -563,14 +573,23 @@ static void MtmAcceptOneConnection()
 	} else { 	
 		MtmHandshakeMessage req;
 		MtmArbiterMessage resp;		
-		int rc = MtmReadSocket(fd, &req, sizeof req);
+		int rc = fcntl(fd, F_SETFL, O_NONBLOCK);
+		if (rc < 0) {
+			elog(ERROR, "Arbiter failed to switch socket to non-blocking mode: %d", errno);
+		}
+		rc = MtmReadSocket(fd, &req, sizeof req);
 		if (rc < sizeof(req)) { 
 			elog(WARNING, "Arbiter failed to handshake socket: %d, errno=%d", rc, errno);
 		} 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);
+
+			MtmLock(LW_EXCLUSIVE);
+			MtmCheckResponse(&req.hdr);
+			MtmUnlock();
+
 			resp.code = MSG_STATUS;
 			resp.disabledNodeMask = Mtm->disabledNodeMask;
 			resp.dxid = HANDSHAKE_MAGIC;
@@ -726,6 +745,7 @@ static void MtmTransSender(Datum arg)
 		CHECK_FOR_INTERRUPTS();
 	}
 	elog(LOG, "Stop arbiter sender %d", MyProcPid);
+	proc_exit(1); /* force restart of this bgwroker */
 }
 
 
@@ -863,9 +883,7 @@ static void MtmTransReceiver(Datum arg)
 						elog(WARNING, "Ignore response for unexisted transaction %d from node %d", msg->dxid, msg->node);
 						continue;
 					}
-					if (BIT_CHECK(msg->disabledNodeMask, MtmNodeId-1) && Mtm->status != MTM_RECOVERY) { 
-						elog(PANIC, "Node %d thinks that I was dead: perform hara-kiri not to be a zombie", msg->node);
-					}
+					MtmCheckResponse(msg);
 
 					if (MtmIsCoordinator(ts)) {
 						switch (msg->code) { 
@@ -975,5 +993,6 @@ static void MtmTransReceiver(Datum arg)
 			MtmRefreshClusterStatus(false);
 		}
 	}
+	proc_exit(1); /* force restart of this bgwroker */
 }
 

docker/docker-entrypoint.sh

@@ -25,6 +25,7 @@ cat >> "$PGDATA/postgresql.conf" <<-EOF
 	max_worker_processes = 100
 	wal_level = logical
 	fsync = off	
+    log_line_prefix = '%t: '
 	max_wal_senders = 10
 	wal_sender_timeout = 0
 	max_replication_slots = 10

multimaster.c

@@ -913,6 +913,9 @@ MtmAbortPreparedTransaction(MtmCurrentTrans* x)
 {
 	MtmTransMap* tm;
 
+	if (Mtm->status == MTM_RECOVERY) { 
+		return;
+	}
 	if (x->status != TRANSACTION_STATUS_ABORTED) { 
 		MtmLock(LW_EXCLUSIVE);
 		tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_REMOVE, NULL);
@@ -947,7 +950,9 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 		}
 		if (ts != NULL) { 
 			if (commit) {
-				Assert(ts->status == TRANSACTION_STATUS_UNKNOWN);
+				/* Assert(ts->status == TRANSACTION_STATUS_UNKNOWN); */
+				Assert(ts->status == TRANSACTION_STATUS_UNKNOWN 
+					   || (ts->status == TRANSACTION_STATUS_IN_PROGRESS && Mtm->status == MTM_RECOVERY)); /* ??? Why there is commit without prepare */
 				if (x->csn > ts->csn) {
 					ts->csn = x->csn;
 					MtmSyncClock(ts->csn);
@@ -1013,6 +1018,15 @@ void MtmSendNotificationMessage(MtmTransState* ts, MtmMessageCode cmd)
 	}
 }
 
+
+static void MtmStartRecovery()
+{
+	MtmLock(LW_EXCLUSIVE);
+	BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
+	MtmSwitchClusterMode(MTM_RECOVERY);
+	MtmUnlock();
+}
+
 void MtmJoinTransaction(GlobalTransactionId* gtid, csn_t globalSnapshot)
 {
 	MtmTx.gtid = *gtid;
@@ -1035,19 +1049,15 @@ void MtmJoinTransaction(GlobalTransactionId* gtid, csn_t globalSnapshot)
 	if (!TransactionIdIsValid(gtid->xid)) { 
 		/* In case of recovery InvalidTransactionId is passed */
 		if (Mtm->status != MTM_RECOVERY) { 
-			elog(PANIC, "Node %d tries to recover node %d which is in %s mode", gtid->node, MtmNodeId,  MtmNodeStatusMnem[Mtm->status]);
+			elog(WARNING, "Node %d tries to recover node %d which is in %s mode", gtid->node, MtmNodeId,  MtmNodeStatusMnem[Mtm->status]);
+			MtmStartRecovery();
 		}
 	} else if (Mtm->status == MTM_RECOVERY) { 
 		/* When recovery is completed we get normal transaction ID and switch to normal mode */
 		MtmRecoveryCompleted();
 	}
 }
 
-void MtmRollbackAllPreparedTransactions(void)
-{
-}
-
-
 void  MtmSetCurrentTransactionGID(char const* gid)
 {
 	MTM_LOG3("Set current transaction xid=%d GID %s", MtmTx.xid, gid);
@@ -1440,6 +1450,8 @@ bool MtmRefreshClusterStatus(bool nowait)
 			}
 		}
 #endif
+		Mtm->reconnectMask |= clique & Mtm->disabledNodeMask; /* new enabled nodes mask */		
+
 		if (disabled) { 
 			MtmCheckQuorum();
 		}
@@ -1470,8 +1482,8 @@ bool MtmRefreshClusterStatus(bool nowait)
 				MtmSwitchClusterMode(MTM_OFFLINE);
 			}
 		} else if (Mtm->status == MTM_OFFLINE) {
-			/* Should we somehow restart logical receivers? */ 
-			MtmSwitchClusterMode(MTM_RECOVERY);
+			/* Should we somehow restart logical receivers? */ 			
+			MtmStartRecovery();
 		}
 	} else { 
 		MTM_LOG1("Clique %lx has no quorum", (long) clique);
@@ -1500,7 +1512,13 @@ void MtmOnNodeDisconnect(int nodeId)
 { 
 	MtmTransState *ts;
 
-	if (Mtm->nodes[nodeId-1].lastStatusChangeTime + MSEC_TO_USEC(MtmNodeDisableDelay) > MtmGetSystemTime()) { 
+	if (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1))
+	{
+		/* Node is already disabled */
+		return;
+	}
+	if (Mtm->nodes[nodeId-1].lastStatusChangeTime + MSEC_TO_USEC(MtmNodeDisableDelay) > MtmGetSystemTime()) 
+	{ 
 		/* Avoid false detection of node failure and prevent node status blinking */
 		return;
 	}

multimaster.h

@@ -119,7 +119,7 @@ typedef enum
 {
 	SLOT_CREATE_NEW,   /* create new slot (drop existed) */
 	SLOT_OPEN_EXISTED, /* open existed slot */
-	SLOT_OPEN_ALWAYS,  /* open existed slot or create new if noty exists */
+	SLOT_OPEN_ALWAYS,  /* open existed slot or create new if not exists */
 } MtmSlotMode;
 
 typedef struct
@@ -279,7 +279,6 @@ extern void  MtmUpdateLsnMapping(int nodeId, XLogRecPtr endLsn);
 extern XLogRecPtr MtmGetFlushPosition(int nodeId);
 extern void MtmWatchdog(void);
 extern void MtmCheckHeartbeat(void);
-extern void MtmRollbackAllPreparedTransactions(void);
 
 
 

pglogical_apply.c

@@ -582,7 +582,7 @@ process_remote_commit(StringInfo in)
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
 			MTM_LOG2("%ld: PGLOGICAL_ABORT_PREPARED commit: gid=%s",  MtmGetSystemTime(), gid);
-			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED) ==TRANSACTION_STATUS_UNKNOWN) { 
+			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED) == TRANSACTION_STATUS_UNKNOWN) { 
 				MTM_LOG1("%ld: PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", MtmGetSystemTime(), gid);
 				StartTransactionCommand();
 				MtmSetCurrentTransactionGID(gid);