return back filtering and status polling

Author: kelvich

arbiter.c

@@ -366,6 +366,10 @@ static void MtmSendHeartbeat()
 				 * true when disabled node connects to a major node which
 				 * is online. So just send it allways. --sk
 				 */
+				/*
+				 * Also recovered node need full visibitily matrix to build proper
+				 * clique. So hearbeats should be sent disregarding disabledNodeMask.
+				 */
 				// && (Mtm->status != MTM_ONLINE
 				// 	|| sockets[i] >= 0
 				// 	|| !BIT_CHECK(Mtm->disabledNodeMask, i)
@@ -1000,16 +1004,10 @@ static void MtmReceiver(Datum arg)
 									if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
 										MTM_ELOG(LOG, "Commit transaction %s because it is prepared at all live nodes", msg->gid);		
 
-										ts->status = msg->status;
-										MtmWakeUpBackend(ts);
-										// replorigin_session_origin = DoNotReplicateId;
+										replorigin_session_origin = DoNotReplicateId;
 										TXFINISH("%s COMMIT, MSG_POLL_STATUS", msg->gid);
-										// MtmUnlock();
-										// LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
-										// MtmFinishPreparedTransaction(ts, true);
-										// LWLockRelease(TwoPhaseStateLock);
-										// MtmLock(LW_EXCLUSIVE);
-										// replorigin_session_origin = InvalidRepOriginId;
+										MtmFinishPreparedTransaction(ts, true);
+										replorigin_session_origin = InvalidRepOriginId;
 									} else { 
 										MTM_LOG1("Receive response for transaction %s -> %s, participants=%llx, voted=%llx", 
 												 msg->gid, MtmTxnStatusMnem[msg->status], ts->participantsMask, ts->votedMask);		

multimaster.c

@@ -1750,7 +1750,7 @@ static void	MtmLoadPreparedTransactions(void)
 			MtmTransactionListAppend(ts);
 			tm->status = ts->status;
 			tm->state = ts;
-			// MtmBroadcastPollMessage(ts);
+			MtmBroadcastPollMessage(ts);
 		}
 	}
 	MTM_LOG1("Recover %d prepared transactions", n);
@@ -3404,7 +3404,7 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 		{
 			/* Lock on us */
 			Mtm->recoverySlot = nodeId;
-			// MtmPollStatusOfPreparedTransactions();
+			MtmPollStatusOfPreparedTransactions();
 			MtmUnlock();
 			return REPLMODE_RECOVERY;
 		}
@@ -3563,13 +3563,13 @@ MtmReplicationStartupHook(struct PGLogicalStartupHookArgs* args)
 				ulong64 recoveredLSN;
 				sscanf(strVal(elem->arg), "%llx", &recoveredLSN);
 				MTM_LOG1("Recovered position of node %d is %llx", MtmReplicationNodeId, recoveredLSN);
-				// if (Mtm->nodes[MtmReplicationNodeId-1].restartLSN < recoveredLSN) {
-				// 	MTM_LOG1("Advance restartLSN for node %d from %llx to %llx (MtmReplicationStartupHook)",
-				// 			 MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, recoveredLSN);
-				// 	// Assert(Mtm->nodes[MtmReplicationNodeId-1].restartLSN == INVALID_LSN
-				// 	// 	   || recoveredLSN < Mtm->nodes[MtmReplicationNodeId-1].restartLSN + MtmMaxRecoveryLag);
-				// 	Mtm->nodes[MtmReplicationNodeId-1].restartLSN = recoveredLSN;
-				// }
+				if (Mtm->nodes[MtmReplicationNodeId-1].restartLSN < recoveredLSN) {
+					MTM_LOG1("Advance restartLSN for node %d from %llx to %llx (MtmReplicationStartupHook)",
+							 MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, recoveredLSN);
+					// Assert(Mtm->nodes[MtmReplicationNodeId-1].restartLSN == INVALID_LSN
+					// 	   || recoveredLSN < Mtm->nodes[MtmReplicationNodeId-1].restartLSN + MtmMaxRecoveryLag);
+					Mtm->nodes[MtmReplicationNodeId-1].restartLSN = recoveredLSN;
+				}
 			} else {
 				MTM_ELOG(ERROR, "Recovered position is not specified");
 			}
@@ -3786,11 +3786,7 @@ bool MtmFilterTransaction(char* record, int size)
 	  default:
 		break;
 	}
-
-	return false;
-
 	restart_lsn = origin_node == MtmReplicationNodeId ? end_lsn : origin_lsn;
-
 	if (Mtm->nodes[origin_node-1].restartLSN < restart_lsn) {
 		MTM_LOG2("[restartlsn] node %d: %llx -> %llx (MtmFilterTransaction)", MtmReplicationNodeId, Mtm->nodes[MtmReplicationNodeId-1].restartLSN, restart_lsn);
 		if (event != PGLOGICAL_PREPARE) {

multimaster.h

@@ -41,7 +41,7 @@
 #define MTM_LOG4(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
 #endif
 
-// #define MTM_TXFINISH 1
+#define MTM_TXFINISH 1
 
 #ifndef MTM_TXFINISH
 #define TXFINISH(fmt, ...)

pglogical_apply.c

@@ -463,21 +463,20 @@ process_remote_message(StringInfo s)
 			 * restartLSN without locks
 			 */
 			if (origin_node == MtmReplicationNodeId) { 
-				// Assert(msg->origin_lsn == INVALID_LSN);
-				// msg->origin_lsn = MtmSenderWalEnd;
+				Assert(msg->origin_lsn == INVALID_LSN);
+				msg->origin_lsn = MtmSenderWalEnd;
+			}
+			if (Mtm->nodes[origin_node-1].restartLSN < msg->origin_lsn) { 
+				MTM_LOG1("Receive logical abort message for transaction %s from node %d: %llx < %llx", msg->gid, origin_node, Mtm->nodes[origin_node-1].restartLSN, msg->origin_lsn);
+				Mtm->nodes[origin_node-1].restartLSN = msg->origin_lsn;
+				replorigin_session_origin_lsn = msg->origin_lsn; 				
+				MtmRollbackPreparedTransaction(origin_node, msg->gid);
+			} else { 
+				if (msg->origin_lsn != INVALID_LSN) { 
+					MTM_LOG1("Ignore rollback of transaction %s from node %d because it's LSN %llx <= %llx", 
+							 msg->gid, origin_node, msg->origin_lsn, Mtm->nodes[origin_node-1].restartLSN);
+				}
 			}
-			MtmRollbackPreparedTransaction(origin_node, msg->gid);
-			// if (Mtm->nodes[origin_node-1].restartLSN < msg->origin_lsn) { 
-			// 	MTM_LOG1("Receive logical abort message for transaction %s from node %d: %llx < %llx", msg->gid, origin_node, Mtm->nodes[origin_node-1].restartLSN, msg->origin_lsn);
-			// 	Mtm->nodes[origin_node-1].restartLSN = msg->origin_lsn;
-			// 	replorigin_session_origin_lsn = msg->origin_lsn; 				
-			// 	MtmRollbackPreparedTransaction(origin_node, msg->gid);
-			// } else { 
-			// 	if (msg->origin_lsn != INVALID_LSN) { 
-			// 		MTM_LOG1("Ignore rollback of transaction %s from node %d because it's LSN %llx <= %llx", 
-			// 				 msg->gid, origin_node, msg->origin_lsn, Mtm->nodes[origin_node-1].restartLSN);
-			// 	}
-			// }
 			standalone = true;
 			break;
 		}

pglogical_receiver.c

@@ -349,10 +349,10 @@ pglogical_receiver_main(Datum main_arg)
 			resetPQExpBuffer(query);
 			Mtm->nodes[nodeId-1].manualRecovery = false;
 		} else {
-			// if (Mtm->nodes[nodeId-1].restartLSN < originStartPos) {
-			// 	MTM_LOG1("Advance restartLSN for node %d: from %llx to %llx (pglogical_receiver_main)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
-			// 	Mtm->nodes[nodeId-1].restartLSN = originStartPos;
-			// }
+			if (Mtm->nodes[nodeId-1].restartLSN < originStartPos) {
+				MTM_LOG1("Advance restartLSN for node %d: from %llx to %llx (pglogical_receiver_main)", nodeId, Mtm->nodes[nodeId-1].restartLSN, originStartPos);
+				Mtm->nodes[nodeId-1].restartLSN = originStartPos;
+			}
 			MTM_LOG1("Restart logical receiver at position %llx from node %d", originStartPos, nodeId);
 		}
 

state.c

@@ -287,9 +287,9 @@ void MtmDisableNode(int nodeId)
 
 	if (Mtm->status == MTM_ONLINE) {
 		/* Make decision about prepared transaction status only in quorum */
-		//MtmLock(LW_EXCLUSIVE);
-		//MtmPollStatusOfPreparedTransactionsForDisabledNode(nodeId, false);
-		//MtmUnlock();
+		MtmLock(LW_EXCLUSIVE);
+		MtmPollStatusOfPreparedTransactionsForDisabledNode(nodeId, false);
+		MtmUnlock();
 	}
 }
 

tests2/docker-entrypoint.sh

@@ -67,7 +67,7 @@ if [ "$1" = 'postgres' ]; then
 			multimaster.heartbeat_send_timeout = 250
 			multimaster.max_recovery_lag = 1GB
 			multimaster.min_recovery_lag = 10kB
-			multimaster.preserve_commit_order = off
+			multimaster.preserve_commit_order = on
 		EOF
 
 		if [ -n "$NODE_ID" ]; then

tests2/lib/bank_client.py

@@ -289,7 +289,7 @@ def run(self):
         self.loop = asyncio.get_event_loop()
 
         for i, _ in enumerate(self.dsns):
-            for j in range(15):
+            for j in range(3):
                 asyncio.ensure_future(self.exec_tx(self.transfer_tx, i, 'transfer', j))
             asyncio.ensure_future(self.exec_tx(self.total_tx, i, 'sumtotal', 0))