Commit stalled precommited transactions in case of single node alive

Author: kelvich

multimaster.c

@@ -1958,9 +1958,11 @@ void MtmHandleApplyError(void)
  * Actually, if node is precommitted (state == UNKNOWN) at any of nodes, then is is prepared at all nodes and so can be committed.
  * But if coordinator of transaction is crashed, we made a decision about transaction commit only if transaction is precommitted at ALL live nodes.
  * The reason is that we want to avoid extra polling to obtain maximum CSN from all nodes to assign it to committed transaction.
- * Called only from MtmDisableNode in critical section.
+ * Called only from MtmDisableNode and in major mode.
+ *
+ * commitPrecommited is used when nnodes=2 and we are switching to major/referee mode.
  */
-void MtmPollStatusOfPreparedTransactionsForDisabledNode(int disabledNodeId)
+void MtmPollStatusOfPreparedTransactionsForDisabledNode(int disabledNodeId, bool commitPrecommited)
 {
 	MtmTransState *ts;
 	for (ts = Mtm->transListHead; ts != NULL; ts = ts->next) {
@@ -1974,8 +1976,15 @@ void MtmPollStatusOfPreparedTransactionsForDisabledNode(int disabledNodeId)
 				MTM_ELOG(LOG, "Abort transaction %s because its coordinator is disabled and it is not prepared at node %d", ts->gid, MtmNodeId);
 				MtmFinishPreparedTransaction(ts, false);
 			} else {
-				MTM_LOG1("Poll state of transaction %s (%llu)", ts->gid, (long64)ts->xid);
-				MtmBroadcastPollMessage(ts);
+				if (commitPrecommited)
+				{
+					MtmFinishPreparedTransaction(ts, true);
+				}
+				else
+				{
+					MTM_LOG1("Poll state of transaction %s (%llu)", ts->gid, (long64)ts->xid);
+					MtmBroadcastPollMessage(ts);
+				}
 			}
 		} else {
 			MTM_LOG2("Skip transaction %s (%llu) with status %s gtid.node=%d gtid.xid=%llu votedMask=%llx",

multimaster.h

@@ -445,7 +445,7 @@ extern void MtmReleaseLocks(void);
 extern void MtmInitMessage(MtmArbiterMessage* msg, MtmMessageCode code);
 extern void MtmSetSnapshot(csn_t snapshot);
 extern void MtmRefereeInitialize(void);
-extern void MtmPollStatusOfPreparedTransactionsForDisabledNode(int disabledNodeId);
+extern void MtmPollStatusOfPreparedTransactionsForDisabledNode(int disabledNodeId, bool commitPrecommited);
 extern int MtmGetNumberOfVotingNodes(void);
 
 #endif

state.c

@@ -71,8 +71,7 @@ MtmSetClusterStatus(MtmNodeStatus status)
 		Mtm->recoverySlot = 0;
 		Mtm->pglogicalReceiverMask = 0;
 		Mtm->pglogicalSenderMask = 0;
-		// XXXX: better to enable, but not now. It requires better testing
-		// Mtm->recoveryCount++; /* this will restart replication connection */
+		Mtm->recoveryCount++; /* this will restart replication connection */
 	}
 
 	Mtm->status = status;
@@ -270,7 +269,7 @@ void MtmDisableNode(int nodeId)
 	if (Mtm->status == MTM_ONLINE) {
 		/* Make decision about prepared transaction status only in quorum */
 		MtmLock(LW_EXCLUSIVE);
-		MtmPollStatusOfPreparedTransactionsForDisabledNode(nodeId);
+		MtmPollStatusOfPreparedTransactionsForDisabledNode(nodeId, false);
 		MtmUnlock();
 	}
 }
@@ -411,6 +410,13 @@ MtmRefreshClusterStatus()
 							winner_node_id);
 				Mtm->refereeGrant = true;
 				MtmLock(LW_EXCLUSIVE);
+				if (countZeroBits(SELF_CONNECTIVITY_MASK, Mtm->nAllNodes) == 1)
+				{
+					// XXXX: that is valid for two nodes. Better idea is to parametrize MtmPollStatus*
+					// functions.
+					int neighbor_node_id = MtmNodeId == 1 ? 2 : 1;
+					MtmPollStatusOfPreparedTransactionsForDisabledNode(neighbor_node_id, true);
+				}
 				MtmEnableNode(MtmNodeId);
 				MtmCheckState();
 				MtmUnlock();

tests2/lib/bank_client.py

@@ -255,9 +255,9 @@ def total_tx(self, conn, cur, agg, conn_i):
         total = yield from cur.fetchone()
         if total[0] != self.total:
             agg.isolation += 1
+            print(datetime.datetime.utcnow(), 'Isolation error, total ', self.total, ' -> ', total[0], ', node ', conn_i+1)
             self.total = total[0]
             print(self.oops)
-            print(datetime.datetime.utcnow(), 'Isolation error, total = ', total, ', node ', conn_i+1)
             # yield from cur.execute('select * from mtm.get_nodes_state()')
             # nodes_state = yield from cur.fetchall()
             # for i, col in enumerate(self.nodes_state_fields):