buggy

Author: kelvich

arbiter.c

@@ -999,7 +999,7 @@ static void MtmReceiver(Datum arg)
 					  default:
 						break;
 					}
-					if (BIT_CHECK(msg->disabledNodeMask, node-1)) {
+					if (BIT_CHECK(msg->disabledNodeMask, node-1) || BIT_CHECK(Mtm->disabledNodeMask, node-1)) {
 						MTM_ELOG(WARNING, "Ignore message from dead node %d\n", node);
 						continue;
 					}

multimaster.c

@@ -225,6 +225,7 @@ char const* const MtmNodeStatusMnem[] =
 	"Recovery",
 	"Recovered",
 	"InMinor",
+	"OutOfClique",
 	"OutOfService"
 };
 
@@ -373,6 +374,7 @@ void MtmLock(LWLockMode mode)
 		if (mode == LW_EXCLUSIVE) {
 			Assert(MtmLockCount == 0);
 			Mtm->lastLockHolder = MyProcPid;
+			Assert(MyProcPid);
 			MtmLockCount = 1;
 		}
 	}
@@ -1155,7 +1157,6 @@ bool MtmWatchdog(timestamp_t now)
 				MTM_LOG1("[STATE] Node %i: Disconnect due to heartbeat timeout (%d msec)",
 					 i+1, (int)USEC_TO_MSEC(now - Mtm->nodes[i].lastHeartbeat));
 				MtmOnNodeDisconnect(i+1);
-				MtmDisableNode(i+1);
 				allAlive = false;
 			}
 		}
@@ -1176,6 +1177,7 @@ void MtmPrecommitTransaction(char const* gid)
 			MTM_ELOG(WARNING, "MtmPrecommitTransaction: transaction '%s' is not found", gid);
 		} else {
 			MtmTransState* ts = tm->state;
+			// Assert(ts != NULL);
 			if (ts == NULL) {
 				MTM_ELOG(WARNING, "MtmPrecommitTransaction: transaction '%s' is not yet prepared, status %s", gid, MtmTxnStatusMnem[tm->status]);
 				MtmUnlock();
@@ -1501,6 +1503,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 				if (!(ts->status == TRANSACTION_STATUS_UNKNOWN
 					  || (ts->status == TRANSACTION_STATUS_IN_PROGRESS && Mtm->status == MTM_RECOVERY)))
 				{
+					MtmUnlock();
 					MTM_ELOG(ERROR, "Attempt to commit %s transaction %s (%llu)",
 						 MtmTxnStatusMnem[ts->status], ts->gid, (long64)ts->xid);
 				}
@@ -2026,15 +2029,19 @@ static int64 MtmGetSlotLag(int nodeId)
  */
 bool MtmIsRecoveredNode(int nodeId)
 {
-	if (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) {
-		if (!MtmIsRecoverySession) {
-			MTM_ELOG(ERROR, "Node %d is marked as disabled but is not in recovery mode", nodeId);
-		}
-		return true;
-	} else {
-		MtmIsRecoverySession = false; /* recovery is completed */
-		return false;
-	}
+	if (!BIT_CHECK(Mtm->disabledNodeMask, nodeId-1))
+		Assert(!MtmIsRecoverySession);
+
+	return BIT_CHECK(Mtm->disabledNodeMask, nodeId-1) && MtmIsRecoverySession;
+	// if (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) {
+	// 	if (!MtmIsRecoverySession) {
+	// 		MTM_ELOG(WARNING, "Node %d is marked as disabled but is not in recovery mode", nodeId);
+	// 	}
+	// 	return true;
+	// } else {
+	// 	MtmIsRecoverySession = false; /* recovery is completed */
+	// 	return false;
+	// }
 }
 
 /*
@@ -2060,7 +2067,7 @@ void MtmCheckRecoveryCaughtUp(int nodeId, lsn_t slotLSN)
 			 */
 			MTM_LOG1("Node %d is almost caught-up: slot position %llx, WAL position %llx, active transactions %d",
 				 nodeId, slotLSN, walLSN, Mtm->nActiveTransactions);
-			BIT_SET(Mtm->originLockNodeMask, nodeId-1);
+			BIT_SET(Mtm->originLockNodeMask, nodeId-1); // XXXX: log that
 		} else {
 			MTM_LOG2("Continue recovery of node %d, slot position %llx, WAL position %llx,"
 					 " WAL sender position %llx, lockers %llx, active transactions %d", nodeId, slotLSN,
@@ -2082,6 +2089,7 @@ bool MtmRecoveryCaughtUp(int nodeId, lsn_t walEndPtr)
 	if (MtmIsRecoveredNode(nodeId) && Mtm->nActiveTransactions == 0) {
 		MtmStateProcessNeighborEvent(nodeId, MTM_NEIGHBOR_RECOVERY_CAUGHTUP);
 		caughtUp = true;
+		MtmIsRecoverySession = false;
 	}
 	MtmUnlock();
 	return caughtUp;
@@ -2099,6 +2107,7 @@ MtmLockCluster(void)
 	}
 	MtmLock(LW_EXCLUSIVE);
 	if (BIT_CHECK(Mtm->originLockNodeMask, MtmNodeId-1)) {
+		MtmUnlock();
 		elog(ERROR, "There is already pending exclusive lock");
 	}
 	BIT_SET(Mtm->originLockNodeMask, MtmNodeId-1);
@@ -2351,6 +2360,7 @@ static void MtmInitialize()
 		Mtm->nLiveNodes = 0; //MtmNodes;
 		Mtm->nAllNodes = MtmNodes;
 		Mtm->disabledNodeMask = 7; //XXXX
+		Mtm->clique = 7; // XXXX
 		Mtm->stalledNodeMask = 0;
 		Mtm->stoppedNodeMask = 0;
 		Mtm->deadNodeMask = 0;
@@ -2383,7 +2393,7 @@ static void MtmInitialize()
 		for (i = 0; i < MtmNodes; i++) {
 			Mtm->nodes[i].oldestSnapshot = 0;
 			Mtm->nodes[i].disabledNodeMask = 0;
-			Mtm->nodes[i].connectivityMask = 7;
+			Mtm->nodes[i].connectivityMask = 7; // XXXX
 			Mtm->nodes[i].lockGraphUsed = 0;
 			Mtm->nodes[i].lockGraphAllocated = 0;
 			Mtm->nodes[i].lockGraphData = NULL;
@@ -3308,17 +3318,56 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 		Mtm->preparedTransactionsLoaded = true;
 	}
 
-	while (BIT_CHECK(Mtm->disabledNodeMask, MtmNodeId - 1) ||
-			BIT_CHECK(SELF_CONNECTIVITY_MASK, nodeId - 1))
+	// while (BIT_CHECK(Mtm->disabledNodeMask, MtmNodeId - 1) ||
+	// 		 BIT_CHECK(SELF_CONNECTIVITY_MASK, nodeId - 1) ||
+	// 		!BIT_CHECK(Mtm->clique, nodeId - 1) ||
+	// 		!BIT_CHECK(Mtm->clique, MtmNodeId - 1) )
+	// {
+	// 	if (*shutdown)
+	// 	{
+	// 		MtmUnlock();
+	// 		return REPLMODE_EXIT;
+	// 	}
+
+	// 	if ((Mtm->recoverySlot == 0 || Mtm->recoverySlot == nodeId) &&
+	// 		(!BIT_CHECK(SELF_CONNECTIVITY_MASK, nodeId - 1)))
+	// 	{
+	// 		/* Lock on us */
+	// 		Mtm->recoverySlot = nodeId;
+	// 		MtmPollStatusOfPreparedTransactions();
+	// 		MtmUnlock();
+	// 		return REPLMODE_RECOVERY;
+	// 	}
+
+	// 	MTM_LOG1("Replication to node %d is pending: recovery node=%d, donor node=%d, connectivity mask=%llx, disabled mask=%llx",
+	// 			nodeId, Mtm->recoverySlot, Mtm->donorNodeId, SELF_CONNECTIVITY_MASK, Mtm->disabledNodeMask);
+
+	// 	MtmUnlock();
+	// 	/* delay opening of other slots until recovery is completed */
+	// 	MtmSleep(STATUS_POLL_DELAY);
+	// 	MtmLock(LW_EXCLUSIVE);
+	// }
+
+	// MtmUnlock();
+
+	// return REPLMODE_RECOVERED;
+
+	/* Await until node is connected and both receiver and sender are in clique */
+	while (BIT_CHECK(SELF_CONNECTIVITY_MASK, nodeId - 1) ||
+			!BIT_CHECK(Mtm->clique, nodeId - 1) ||
+			!BIT_CHECK(Mtm->clique, MtmNodeId - 1) )
 	{
+		MtmUnlock();
 		if (*shutdown)
-		{
-			MtmUnlock();
 			return REPLMODE_EXIT;
-		}
+		MtmSleep(STATUS_POLL_DELAY);
+		MtmLock(LW_EXCLUSIVE);
+	}
 
-		if ((Mtm->recoverySlot == 0 || Mtm->recoverySlot == nodeId) &&
-			(!BIT_CHECK(SELF_CONNECTIVITY_MASK, nodeId - 1)))
+	if (BIT_CHECK(Mtm->disabledNodeMask, MtmNodeId - 1))
+	{
+		/* Ok, then start recovery by luckiest walreceiver */
+		if (Mtm->recoverySlot == 0 || Mtm->recoverySlot == nodeId)
 		{
 			/* Lock on us */
 			Mtm->recoverySlot = nodeId;
@@ -3327,87 +3376,19 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 			return REPLMODE_RECOVERY;
 		}
 
-		MTM_LOG1("Replication to node %d is pending: recovery node=%d, donor node=%d, connectivity mask=%llx, disabled mask=%llx",
-				nodeId, Mtm->recoverySlot, Mtm->donorNodeId, SELF_CONNECTIVITY_MASK, Mtm->disabledNodeMask);
-
-		MtmUnlock();
-		/* delay opening of other slots until recovery is completed */
-		MtmSleep(STATUS_POLL_DELAY);
-		MtmLock(LW_EXCLUSIVE);
+		/* And force less lucky walreceivers wait until recovery is completed */
+		while (BIT_CHECK(Mtm->disabledNodeMask, MtmNodeId - 1))
+		{
+			MtmUnlock();
+			if (*shutdown)
+				return REPLMODE_EXIT;
+			MtmSleep(STATUS_POLL_DELAY);
+			MtmLock(LW_EXCLUSIVE);
+		}
 	}
 
 	MtmUnlock();
-
 	return REPLMODE_RECOVERED;
-
-
-
-
-
-	// while ((Mtm->status != MTM_CONNECTED && Mtm->status != MTM_RECOVERED && Mtm->status != MTM_ONLINE)
-	// 	   || BIT_CHECK(Mtm->disabledNodeMask, nodeId-1))
-	// // while (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1))
-	// {
-	// 	if (*shutdown)
-	// 	{
-	// 		MtmUnlock();
-	// 		return REPLMODE_EXIT;
-	// 	}
-	// 	// /* We are not interested in receiving any deteriorated logical messages from recovered node, so recreate slot */
-	// 	// if (BIT_CHECK(Mtm->disabledNodeMask, nodeId-1)) {
-	// 	// 	mode = REPLMODE_CREATE_NEW;
-	// 	// }
-	// 	// MTM_LOG2("%d: receiver slot mode %s", MyProcPid, MtmNodeStatusMnem[Mtm->status]);
-
-	// 	if (Mtm->status == MTM_RECOVERY) {
-	// 		mode = REPLMODE_RECOVERED;
-	// 		/* Choose node for recovery if
-	// 		 * 1. It is not chosen yet or the same node was chosen before
-	// 		 * 2. It is donor node or there is no donor node
-	// 		 * 3. Connections with all other live nodes were established
-	// 		 */
-	// 		if ((Mtm->recoverySlot == 0 || Mtm->recoverySlot == nodeId)
-	// 			&& (Mtm->donorNodeId == MtmNodeId || Mtm->donorNodeId == nodeId)
-	// 			&& (SELF_CONNECTIVITY_MASK & ~Mtm->disabledNodeMask) == 0)
-	// 		{
-	// 			/* Choose for recovery first available slot or slot of donor node (if any) */
-	// 			if (Mtm->nAllNodes >= 3) {
-	// 				MTM_ELOG(WARNING, "Process %d starts recovery from node %d restartLSNs={%llx, %llx, %llx}",
-	// 					 MyProcPid, nodeId, Mtm->nodes[0].restartLSN, Mtm->nodes[1].restartLSN, Mtm->nodes[2].restartLSN);
-	// 			} else {
-	// 				MTM_ELOG(WARNING, "Process %d starts recovery from node %d", MyProcPid, nodeId);
-	// 			}
-	// 			Mtm->recoverySlot = nodeId;
-	// 			// Mtm->nReceivers = 0;
-	// 			// Mtm->nSenders = 0;
-	// 			// Mtm->recoveryCount += 1;
-	// 			// Mtm->pglogicalReceiverMask = 0;
-	// 			// Mtm->pglogicalSenderMask = 0;
-	// 			MtmPollStatusOfPreparedTransactions();
-	// 			MtmUnlock();
-	// 			return REPLMODE_RECOVERY;
-	// 		}
-	// 	}
-	// 	MTM_LOG1("Replication to node %d is pending: recovery node=%d, donor node=%d, connectivity mask=%llx, disabled mask=%llx",
-	// 			 nodeId, Mtm->recoverySlot, Mtm->donorNodeId, SELF_CONNECTIVITY_MASK, Mtm->disabledNodeMask);
-	// 	MtmUnlock();
-	// 	/* delay opening of other slots until recovery is completed */
-	// 	MtmSleep(STATUS_POLL_DELAY);
-	// 	MtmLock(LW_EXCLUSIVE);
-	// }
-	// if (Mtm->status == MTM_RECOVERED) {
-	// 	mode = REPLMODE_RECOVERED;
-	// }
-	// // if (mode == REPLMODE_RECOVERED) {
-	// // 	MTM_LOG1("%d: Restart replication from node %d after end of recovery", MyProcPid, nodeId);
-	// // } else if (mode == REPLMODE_CREATE_NEW) {
-	// // 	MTM_LOG1("%d: Start replication from recovered node %d", MyProcPid, nodeId);
-	// // } else {
-	// // 	MTM_LOG1("%d: Continue replication from node %d", MyProcPid, nodeId);
-	// // }
-	// BIT_SET(Mtm->reconnectMask, nodeId-1); /* arbiter should try to reestablish connection with this node */
-	// MtmUnlock();
-	// return mode;
 }
 
 static bool MtmIsBroadcast()

multimaster.h

@@ -108,7 +108,7 @@ typedef enum
 /* Identifier of global transaction */
 typedef struct
 {
-	int node;          /* Zero based index of node initiating transaction */
+	int node;          /* One based id of node initiating transaction */
 	TransactionId xid; /* Transaction ID at this node */
 } GlobalTransactionId;
 
@@ -137,6 +137,7 @@ typedef enum
 	MTM_RECOVERY,       /* Node is in recovery process */
 	MTM_RECOVERED,      /* Node is recovered by is not yet switched to ONLINE because not all sender/receivers are restarted */
 	MTM_IN_MINORITY,    /* Node is out of quorum */
+	MTM_OUT_OF_CLIQUE,  /* Node is out of cluster by clique detector */
 	MTM_OUT_OF_SERVICE  /* Node is not available to to critical, non-recoverable error */
 } MtmNodeStatus;
 
@@ -288,6 +289,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 clique;                 /* Bitmask of nodes that are connected and we allowed to connect/send wal/receive wal with them */
 	nodemask_t deadNodeMask;           /* Bitmask of nodes considered as dead by referee */
 	nodemask_t recoveredNodeMask;      /* Bitmask of nodes recoverd after been reported as dead by referee */
 	nodemask_t stalledNodeMask;        /* Bitmask of stalled nodes (node with dropped replication slot which makes it not possible automatic recovery of such node) */
@@ -417,7 +419,6 @@ extern void  MtmUpdateNodeConnectionInfo(MtmConnectionInfo* conn, char const* co
 extern void  MtmSetupReplicationHooks(struct PGLogicalHooks* hooks);
 extern bool  MtmRecoveryCaughtUp(int nodeId, lsn_t walEndPtr);
 extern void  MtmCheckRecoveryCaughtUp(int nodeId, lsn_t slotLSN);
-extern void  MtmRecoveryCompleted(void);
 extern void  MtmMakeTableLocal(char const* schema, char const* name);
 extern void  MtmHandleApplyError(void);
 extern void  MtmUpdateLsnMapping(int nodeId, lsn_t endLsn);

pglogical_apply.c

@@ -658,6 +658,7 @@ process_remote_commit(StringInfo in)
 	csn_t       csn;
 	lsn_t       end_lsn;
 	lsn_t       origin_lsn;
+	lsn_t       commit_lsn;
 	int         origin_node;
 	char        gid[MULTIMASTER_MAX_GID_SIZE];
 
@@ -668,7 +669,7 @@ process_remote_commit(StringInfo in)
 	MtmReplicationNodeId = pq_getmsgbyte(in);
 
 	/* read fields */
-	pq_getmsgint64(in); /* commit_lsn */
+	commit_lsn = pq_getmsgint64(in); /* commit_lsn */
 	end_lsn = pq_getmsgint64(in); /* end_lsn */
 	replorigin_session_origin_timestamp = pq_getmsgint64(in); /* commit_time */
 
@@ -692,7 +693,7 @@ process_remote_commit(StringInfo in)
 		}
 		case PGLOGICAL_COMMIT:
 		{
-			MTM_LOG2("%d: PGLOGICAL_COMMIT %s, (%llx,%llx,%llx)", MyProcPid, gid, commit_lsn, end_lsn, origin_lsn);
+			MTM_LOG1("%d: PGLOGICAL_COMMIT %s, (%llx,%llx,%llx)", MyProcPid, gid, commit_lsn, end_lsn, origin_lsn);
 			if (IsTransactionState()) {
 				Assert(TransactionIdIsValid(MtmGetCurrentTransactionId()));
 				MtmBeginSession(origin_node);
@@ -969,10 +970,10 @@ process_remote_update(StringInfo s, Relation rel)
 		{
 			StringInfoData o;
 			initStringInfo(&o);
-			tuple_to_stringinfo(&o, RelationGetDescr(rel), oldslot->tts_tuple);
+			tuple_to_stringinfo(&o, RelationGetDescr(rel), oldslot->tts_tuple, false);
 			appendStringInfo(&o, " to");
-			tuple_to_stringinfo(&o, RelationGetDescr(rel), remote_tuple);
-			MTM_LOG1(DEBUG1, "UPDATE:%s", o.data);
+			tuple_to_stringinfo(&o, RelationGetDescr(rel), remote_tuple, false);
+			MTM_LOG1("%lu: UPDATE: %s", GetCurrentTransactionId(), o.data);
 			resetStringInfo(&o);
 		}
 #endif
@@ -1191,7 +1192,6 @@ void MtmExecutor(void* work, size_t size)
 			}
 			case 'Z':
 			{
-				// MtmRecoveryCompleted();
 				MtmStateProcessEvent(MTM_RECOVERY_FINISH2);
 				inside_transaction = false;
 				break;