Do not swich ourselves to offline

Author: knizhnik

multimaster.c

@@ -1316,13 +1316,15 @@ static void MtmPollStatusOfPreparedTransactions(int disabledNodeId)
 
 static void MtmDisableNode(int nodeId)
 {
+	timestamp_t now = MtmGetSystemTime();
+	elog(WARNING, "Disable node %d at xlog position %lx, last status change time %d msec ago", nodeId, GetXLogInsertRecPtr(), 
+		 (int)USEC_TO_MSEC(now - Mtm->nodes[nodeId-1].lastStatusChangeTime));
 	BIT_SET(Mtm->disabledNodeMask, nodeId-1);
-	Mtm->nodes[nodeId-1].lastStatusChangeTime = MtmGetSystemTime();
+	Mtm->nodes[nodeId-1].lastStatusChangeTime = now;
 	Mtm->nodes[nodeId-1].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
 	if (nodeId != MtmNodeId) { 
 		Mtm->nLiveNodes -= 1;
 	}
-	elog(WARNING, "Disable node %d at xlog position %lx", nodeId, GetXLogInsertRecPtr());
 	MtmPollStatusOfPreparedTransactions(nodeId);
 } 
 
@@ -1345,8 +1347,8 @@ void MtmRecoveryCompleted(void)
 			 MtmNodeId, Mtm->disabledNodeMask, Mtm->reconnectMask, Mtm->nLiveNodes);
 	MtmLock(LW_EXCLUSIVE);
 	Mtm->recoverySlot = 0;
-	Mtm->nodes[MtmNodeId-1].lastStatusChangeTime = MtmGetSystemTime();
 	BIT_CLEAR(Mtm->disabledNodeMask, MtmNodeId-1);
+	Mtm->nodes[MtmNodeId-1].lastStatusChangeTime = MtmGetSystemTime();
 	for (i = 0; i < Mtm->nAllNodes; i++) { 
 		Mtm->nodes[i].lastHeartbeat = 0; /* defuse watchdog until first heartbeat is received */
 	}
@@ -1600,10 +1602,11 @@ bool MtmRefreshClusterStatus(bool nowait, int testNodeId)
 		if (disabled) { 
 			timestamp_t now = MtmGetSystemTime();
 			for (i = 0, mask = disabled; mask != 0; i++, mask >>= 1) {
-				if (mask & 1) { 
-					if (Mtm->nodes[i].lastStatusChangeTime + MSEC_TO_USEC(MtmNodeDisableDelay) < now) {
-						MtmDisableNode(i+1);
-					}
+				if (i+1 != MtmNodeId 
+					&& (mask & 1) != 0
+					&& Mtm->nodes[i].lastStatusChangeTime + MSEC_TO_USEC(MtmNodeDisableDelay) < now)
+				{
+					MtmDisableNode(i+1);
 				}
 			}
 		}		
@@ -1615,15 +1618,16 @@ bool MtmRefreshClusterStatus(bool nowait, int testNodeId)
 
 		if (disabled|enabled) { 
 			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.nхode, ts->xid, ts->status, ts->gtid.xid);
-			if (MtmIsCoordinator(ts)) { 
-				if (!ts->votingCompleted && disabled != 0 && ts->status != TRANSACTION_STATUS_ABORTED) {
-					MtmAbortTransaction(ts);
-					MtmWakeUpBackend(ts);
+			
+			/* 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.nхode, ts->xid, ts->status, ts->gtid.xid);
+				if (MtmIsCoordinator(ts)) { 
+					if (!ts->votingCompleted && disabled != 0 && ts->status != TRANSACTION_STATUS_ABORTED) {
+						MtmAbortTransaction(ts);
+						MtmWakeUpBackend(ts);
+					}
 				}
 			}
 		}
@@ -2242,7 +2246,7 @@ _PG_init(void)
 		"Minimal amount of time (msec) between node status change",
 		"This delay is used to avoid false detection of node failure and to prevent blinking of node status node",
 		&MtmNodeDisableDelay,
-		1000,
+		2000,
 		1,
 		INT_MAX,
 		PGC_BACKEND,

pglogical_apply.c

@@ -341,7 +341,7 @@ process_remote_begin(StringInfo s)
 	Assert(gtid.node > 0);
 
 	MTM_LOG2("REMOTE begin node=%d xid=%d snapshot=%ld", gtid.node, gtid.xid, snapshot);
-#if 0
+#if 1
 	if (BIT_CHECK(Mtm->disabledNodeMask, gtid.node-1)) { 
 		elog(WARNING, "Ignore transaction %d from disabled node %d", gtid.xid, gtid.node);
 		MtmResetTransaction();		
@@ -633,9 +633,9 @@ process_remote_commit(StringInfo in)
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
-			MTM_LOG2("PGLOGICAL_ABORT_PREPARED commit: gid=%s",  gid);
+			MTM_LOG1("PGLOGICAL_ABORT_PREPARED commit: gid=%s",  gid);
 			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED) == TRANSACTION_STATUS_UNKNOWN) { 
-				MTM_LOG2("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
+				MTM_LOG1("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
 				StartTransactionCommand();
 				MtmBeginSession();
 				MtmSetCurrentTransactionGID(gid);

pglogical_proto.c

@@ -161,14 +161,22 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 			return;
 		}
 	} else { 
-		//csn_t csn = MtmTransactionSnapshot(txn->xid);
+		csn_t csn = MtmTransactionSnapshot(txn->xid);
 		bool isRecovery = MtmIsRecoveredNode(MtmReplicationNodeId);
 
+
 		if (!isRecovery && txn->origin_id != InvalidRepOriginId) 
 		{
+			if (flags == PGLOGICAL_ABORT_PREPARED) { 
+				MTM_LOG1("Skip ABORT_PREPARED for transaction %s to node %d", txn->gid, MtmReplicationNodeId);
+			}
 			Assert(MtmTransactionRecords == 0);
 			return;
 		}
+		if (flags == PGLOGICAL_ABORT_PREPARED) { 
+			MTM_LOG1("Send ABORT_PREPARED for transaction %d (%s) end_lsn=%lx to node %d, isRecovery=%d, txn->origin_id=%d, csn=%ld", 
+					 txn->xid, txn->gid, txn->end_lsn, MtmReplicationNodeId, isRecovery, txn->origin_id, csn);
+		}
 		if (MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn)) { 
 			MTM_LOG1("wal-sender complete recovery of node %d at LSN(commit %lx, end %lx, log %lx) in transaction %s event %d", MtmReplicationNodeId, commit_lsn, txn->end_lsn, GetXLogInsertRecPtr(), txn->gid, flags);
 			flags |= PGLOGICAL_CAUGHT_UP;