Abort transactions if configuration is changed since transaction was prepared

Author: knizhnik

contrib/mmts/arbiter.c

@@ -764,6 +764,7 @@ static void MtmSender(Datum arg)
 			}
 		}		
 		CHECK_FOR_INTERRUPTS();
+		MtmCheckHeartbeat();
 	}
 	elog(LOG, "Stop arbiter sender %d", MyProcPid);
 	proc_exit(1); /* force restart of this bgwroker */

contrib/mmts/multimaster--1.0.sql

@@ -27,7 +27,7 @@ CREATE FUNCTION mtm.get_snapshot() RETURNS bigint
 AS 'MODULE_PATHNAME','mtm_get_snapshot'
 LANGUAGE C;
 
-CREATE FUNCTION mtm.get_csn(integer xid) RETURNS bigint
+CREATE FUNCTION mtm.get_csn(xid integer) RETURNS bigint
 AS 'MODULE_PATHNAME','mtm_get_csn'
 LANGUAGE C;
 
@@ -51,7 +51,7 @@ CREATE FUNCTION mtm.get_trans_by_gid(git text) RETURNS mtm.trans_state
 AS 'MODULE_PATHNAME','mtm_get_trans_by_gid'
 LANGUAGE C;
 
-CREATE FUNCTION mtm.get_trans_by_xid(integer xid) RETURNS mtm.trans_state
+CREATE FUNCTION mtm.get_trans_by_xid(xid integer) RETURNS mtm.trans_state
 AS 'MODULE_PATHNAME','mtm_get_trans_by_xid'
 LANGUAGE C;
 

contrib/mmts/multimaster.c

@@ -1020,9 +1020,19 @@ Mtm2PCVoting(MtmCurrentTrans* x, MtmTransState* ts)
 	timestamp_t start = MtmGetSystemTime();
 	timestamp_t deadline = start + timeout;
 	timestamp_t now;
-	
+	nodemask_t  liveNodesMask = (((nodemask_t)1 << Mtm->nAllNodes) - 1) & ~Mtm->disabledNodeMask & ~((nodemask_t)1 << (MtmNodeId-1));
+
 	Assert(ts->csn > ts->snapshot);
 
+	if (ts->participantsMask != liveNodesMask) 
+	{ 
+		elog(WARNING, "Abort transaction %d (%s) because cluster configuration is changed from %lx to %lx since transaction start", 
+			 ts->xid, ts->gid, ts->participantsMask, liveNodesMask);
+		MtmAbortTransaction(ts);
+		x->status = TRANSACTION_STATUS_ABORTED;
+		return;
+	}
+
 	/* Wait votes from all nodes until: */
 	while (!MtmVotingCompleted(ts)
 		   && (ts->isPrepared || nConfigChanges == Mtm->nConfigChanges))
@@ -1534,11 +1544,13 @@ XidStatus MtmExchangeGlobalTransactionStatus(char const* gid, XidStatus new_stat
 		}
 		if (tm->state != NULL && old_status == TRANSACTION_STATUS_IN_PROGRESS) { 
 			/* Return UNKNOWN to mark that transaction was prepared */
-			MTM_LOG1("Change status of in-progress transaction %s to %s", gid, MtmTxtStatusMnem[new_status]);
+			if (new_status != TRANSACTION_STATUS_UNKNOWN) { 
+				MTM_LOG1("Change status of in-progress transaction %s to %s", gid, MtmTxnStatusMnem[new_status]);
+			}
 			old_status = TRANSACTION_STATUS_UNKNOWN;
 		}
 	} else { 
-		MTM_LOG1("Set status of unknown transaction %s to %s", gid, MtmTxtStatusMnem[new_status]);
+		MTM_LOG2("Set status of unknown transaction %s to %s", gid, MtmTxnStatusMnem[new_status]);
 		tm->state = NULL;
 		tm->status = new_status;
 	}