fix last problem with lost PRECOMMIT confirmation

Author: kelvich

multimaster.c

@@ -1179,9 +1179,7 @@ void MtmPrecommitTransaction(char const* gid)
 				ts->status = TRANSACTION_STATUS_UNKNOWN;
 				ts->csn = MtmAssignCSN();
 				MtmAdjustSubtransactions(ts);
-				if (Mtm->status != MTM_RECOVERY) {
-					MtmSend2PCMessage(ts, MSG_PRECOMMITTED);
-				}
+				MtmSend2PCMessage(ts, MSG_PRECOMMITTED);
 				MtmUnlock();
 				Assert(replorigin_session_origin != InvalidRepOriginId);
 				if (!IsTransactionState()) {
@@ -1625,12 +1623,10 @@ void MtmSend2PCMessage(MtmTransState* ts, MtmMessageCode cmd)
 	memcpy(msg.gid, ts->gid, MULTIMASTER_MAX_GID_SIZE);
 
 	Assert(!MtmIsCoordinator(ts));	/* All broadcasts are now done through logical decoding */
-	if (!BIT_CHECK(Mtm->disabledNodeMask, ts->gtid.node-1)) {
-		MTM_LOG2("Send %s message to node %d xid=%d gid=%s", MtmMessageKindMnem[cmd], ts->gtid.node, ts->gtid.xid, ts->gid);
-		msg.node = ts->gtid.node;
-		msg.dxid = ts->gtid.xid;
-		MtmSendMessage(&msg);
-	}
+	MTM_TXTRACE(ts, "MtmSend2PCMessage sending %s message to node %d", MtmMessageKindMnem[cmd], ts->gtid.node);
+	msg.node = ts->gtid.node;
+	msg.dxid = ts->gtid.xid;
+	MtmSendMessage(&msg);
 }
 
 /*

multimaster.h

@@ -14,9 +14,6 @@
 #define DEBUG_LEVEL 0
 #endif
 
-#ifndef MTM_TRACE
-#define MTM_TRACE   0
-#endif
 
 #define MTM_TAG "[MTM] "
 #define MTM_ELOG(level,fmt,...) elog(level, MTM_TAG fmt, ## __VA_ARGS__)
@@ -44,12 +41,14 @@
 #define MTM_LOG4(fmt, ...) fprintf(stderr, fmt "\n", ## __VA_ARGS__)
 #endif
 
-// #if MTM_TRACE == 0
-// #define MTM_TXTRACE(tx, event)
-// #else
-#define MTM_TXTRACE(tx, event) \
-		fprintf(stderr, MTM_TAG "%s, %lld, %s, %d\n", tx->gid, (long long)MtmGetSystemTime(), event, MyProcPid)
-// #endif
+// #define MTM_TRACE 1
+
+#ifndef MTM_TRACE
+#define MTM_TXTRACE(tx, event, ...)
+#else
+#define MTM_TXTRACE(tx, event, ...) \
+		fprintf(stderr, MTM_TAG "%s, %lld, %u " event "\n", tx->gid, (long long)MtmGetSystemTime(), MyProcPid, ## __VA_ARGS__)
+#endif
 
 #define MULTIMASTER_NAME                 "multimaster"
 #define MULTIMASTER_SCHEMA_NAME          "mtm"
@@ -390,7 +389,6 @@ extern csn_t MtmDistributedTransactionSnapshot(TransactionId xid, int nodeId, no
 extern csn_t MtmAssignCSN(void);
 extern csn_t MtmSyncClock(csn_t csn);
 extern void  MtmJoinTransaction(GlobalTransactionId* gtid, csn_t snapshot, nodemask_t participantsMask);
-extern void  MtmReceiverStarted(int nodeId);
 extern MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shutdown);
 extern void  MtmExecute(void* work, int size);
 extern void  MtmExecutor(void* work, size_t size);

pglogical_apply.c

@@ -684,15 +684,15 @@ process_remote_commit(StringInfo in)
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			strncpy(gid, pq_getmsgstring(in), sizeof gid);
-			MTM_LOG2("%d: PGLOGICAL_PRECOMMIT_PREPARED %s", MyProcPid, gid);
+			MTM_LOG2("%d: PGLOGICAL_PRECOMMIT_PREPARED %s, (%llx,%llx,%llx)", MyProcPid, gid, commit_lsn, end_lsn, origin_lsn);
 			MtmBeginSession(origin_node);
 			MtmPrecommitTransaction(gid);
 			MtmEndSession(origin_node, true);
 			return;
 		}
 		case PGLOGICAL_COMMIT:
 		{
-			MTM_LOG2("%d: PGLOGICAL_COMMIT commit", MyProcPid);
+			MTM_LOG2("%d: PGLOGICAL_COMMIT %s, (%llx,%llx,%llx)", MyProcPid, gid, commit_lsn, end_lsn, origin_lsn);
 			if (IsTransactionState()) {
 				Assert(TransactionIdIsValid(MtmGetCurrentTransactionId()));
 				MtmBeginSession(origin_node);
@@ -705,12 +705,12 @@ process_remote_commit(StringInfo in)
 		{
 			Assert(IsTransactionState() && TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			strncpy(gid, pq_getmsgstring(in), sizeof gid);
+			MTM_LOG2("%d: PGLOGICAL_PREPARE %s, (%llx,%llx,%llx)", MyProcPid, gid, commit_lsn, end_lsn, origin_lsn);
 			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_IN_PROGRESS) == TRANSACTION_STATUS_ABORTED) { 
 				MTM_LOG1("Avoid prepare of previously aborted global transaction %s", gid);	
 				AbortCurrentTransaction();
 			} else { 				
 				/* prepare TBLOCK_INPROGRESS state for PrepareTransactionBlock() */
-				MTM_LOG2("PGLOGICAL_PREPARE commit: gid=%s", gid);
 				BeginTransactionBlock(false);
 				CommitTransactionCommand();
 				StartTransactionCommand();
@@ -745,7 +745,7 @@ process_remote_commit(StringInfo in)
 			 */
 			Assert(csn);
 			strncpy(gid, pq_getmsgstring(in), sizeof gid);
-			MTM_LOG2("PGLOGICAL_COMMIT_PREPARED commit: csn=%lld, gid=%s, lsn=%llx", csn, gid, end_lsn);
+			MTM_LOG2("%d: PGLOGICAL_COMMIT_PREPARED %s, (%llx,%llx,%llx)", MyProcPid, gid, commit_lsn, end_lsn, origin_lsn);
 			MtmResetTransaction();
 			StartTransactionCommand();
 			MtmBeginSession(origin_node);

pglogical_receiver.c

@@ -519,14 +519,6 @@ pglogical_receiver_main(Datum main_arg)
 				{
 					int msg_len = rc - hdr_len;
 					stmt = copybuf + hdr_len;
-					if (mode == REPLMODE_RECOVERED) {
-						/* Ingore all incompleted transactions from recovered node */
-						if (stmt[0] != 'B') {
-							output_written_lsn = Max(walEnd, output_written_lsn);
-							continue;
-						}
-						mode = REPLMODE_OPEN_EXISTED;
-					}
 					MTM_LOG3("Receive message %c from node %d", stmt[0], nodeId);
 					if (buf.used + msg_len + 1 >= MtmTransSpillThreshold*MB) {
 						if (spill_file < 0) {

state.c

@@ -112,7 +112,7 @@ MtmCheckState(void)
 		case MTM_RECOVERY:
 			if (!BIT_CHECK(Mtm->disabledNodeMask, MtmNodeId-1))
 			{
-				// BIT_SET(Mtm->originLockNodeMask, MtmNodeId-1); // kk trick
+				BIT_SET(Mtm->originLockNodeMask, MtmNodeId-1); // kk trick
 				MtmSetClusterStatus(MTM_RECOVERED);
 				return;
 			}
@@ -148,6 +148,7 @@ MtmStateProcessNeighborEvent(int node_id, MtmNeighborEvent ev)
 			break;
 
 		case MTM_NEIGHBOR_WAL_RECEIVER_START:
+			BIT_CLEAR(Mtm->originLockNodeMask, MtmNodeId-1);
 			if (!BIT_CHECK(Mtm->pglogicalReceiverMask, node_id - 1))
 				BIT_SET(Mtm->pglogicalReceiverMask, node_id - 1);
 			break;

t/004_recovery.pl

@@ -74,3 +74,7 @@
 diag("Sums: $sum0, $sum1, $sum2");
 is($sum2, $sum0, "Check that sum_2 == sum_0");
 is($sum2, $sum1, "Check that sum_2 == sum_1");
+
+$cluster->{nodes}->[0]->stop('fast');
+$cluster->{nodes}->[1]->stop('fast');
+$cluster->{nodes}->[2]->stop('fast');