Fix setting origin for local commits

Author: knizhnik

arbiter.c

@@ -314,7 +314,7 @@ static void MtmSetSocketOptions(int sd)
 static void MtmCheckResponse(MtmArbiterMessage* resp)
 {
 	if (BIT_CHECK(resp->disabledNodeMask, MtmNodeId-1) && !BIT_CHECK(Mtm->disabledNodeMask, resp->node-1)) { 
-		elog(WARNING, "Node %d thinks that I was dead, while I am %s", resp->node, MtmNodeStatusMnem[Mtm->status]);
+		elog(WARNING, "Node %d thinks that I was dead, while I am %s (message %s)", resp->node, MtmNodeStatusMnem[Mtm->status], messageKindText[resp->code]);
 		if (Mtm->status != MTM_RECOVERY) { 
 			BIT_SET(Mtm->disabledNodeMask, MtmNodeId-1);
 			MtmSwitchClusterMode(MTM_RECOVERY);
@@ -918,7 +918,7 @@ static void MtmReceiver(Datum arg)
 								if (msg->status == TRANSACTION_STATUS_IN_PROGRESS || msg->status == TRANSACTION_STATUS_ABORTED) {
 									elog(LOG, "Abort transaction %s because it is in state %d at node %d",
 										 msg->gid, ts->status, node);
-									MtmFinishPreparedTransaction(node, ts, false);
+									MtmFinishPreparedTransaction(ts, false);
 								} 
 								else if (msg->status == TRANSACTION_STATUS_COMMITTED ||  msg->status == TRANSACTION_STATUS_UNKNOWN)
 								{ 
@@ -928,7 +928,7 @@ static void MtmReceiver(Datum arg)
 									}
 									if ((ts->participantsMask & ~Mtm->disabledNodeMask & ~ts->votedMask) == 0) {
 										elog(LOG, "Commit transaction %s because it is prepared at all live nodes", msg->gid);		
-										MtmFinishPreparedTransaction(node, ts, true);
+										MtmFinishPreparedTransaction(ts, true);
 									}
 								} else {
 									elog(LOG, "Receive response %d for transaction %s for node %d, votedMask=%llx, participantsMask=%llx",

multimaster.c

@@ -1103,7 +1103,7 @@ MtmAbortPreparedTransaction(MtmCurrentTrans* x)
 static void 
 MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 {
-	MTM_LOG1("%d: End transaction %d, prepared=%d, replicated=%d, distributed=%d, 2pc=%d, gid=%s -> %s", 
+	MTM_LOG2("%d: End transaction %d, prepared=%d, replicated=%d, distributed=%d, 2pc=%d, gid=%s -> %s", 
 			 MyProcPid, x->xid, x->isPrepared, x->isReplicated, x->isDistributed, x->isTwoPhase, x->gid, commit ? "commit" : "abort");
 	if (x->status != TRANSACTION_STATUS_ABORTED && x->isDistributed && (x->isPrepared || x->isReplicated) && !x->isTwoPhase) {
 		MtmTransState* ts = NULL;
@@ -1121,7 +1121,7 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 		}
 		if (ts != NULL) { 
 			if (*ts->gid)  
-				MTM_LOG1("TRANSLOG: %s transaction %s status %d", (commit ? "commit" : "rollback"), ts->gid, ts->status);
+				MTM_LOG2("TRANSLOG: %s transaction %s status %d", (commit ? "commit" : "rollback"), ts->gid, ts->status);
 			if (commit) {
 				if (!(ts->status == TRANSACTION_STATUS_UNKNOWN 
 					  || (ts->status == TRANSACTION_STATUS_IN_PROGRESS && Mtm->status == MTM_RECOVERY)))  
@@ -1176,6 +1176,9 @@ MtmEndTransaction(MtmCurrentTrans* x, bool commit)
 					Mtm->nActiveTransactions -= 1;
 				}
 				MtmTransactionListAppend(ts);
+				if (*x->gid) { 
+					LogLogicalMessage("A", x->gid, strlen(x->gid) + 1, false);
+				}
 			}
 			MtmSend2PCMessage(ts, MSG_ABORTED); /* send notification to coordinator */
 		} else if (x->status == TRANSACTION_STATUS_ABORTED && x->isReplicated && !x->isPrepared) {
@@ -1229,7 +1232,7 @@ void MtmSend2PCMessage(MtmTransState* ts, MtmMessageCode cmd)
 				MtmSendMessage(&msg);
 			}
 		}
-	} else {
+	} else if (!BIT_CHECK(Mtm->disabledNodeMask, ts->gtid.node-1)) {
 		msg.node = ts->gtid.node;
 		msg.dxid = ts->gtid.xid;
 		MtmSendMessage(&msg);
@@ -1435,7 +1438,7 @@ static void MtmPollStatusOfPreparedTransactions(int disabledNodeId)
 			Assert(ts->gid[0]);
 			if (ts->status == TRANSACTION_STATUS_IN_PROGRESS) { 
 				elog(LOG, "Abort transaction %s because its coordinator is disabled and it is not prepared at node %d", ts->gid, MtmNodeId);
-				MtmFinishPreparedTransaction(disabledNodeId, ts, false);
+				MtmFinishPreparedTransaction(ts, false);
 			} else {
 				MTM_LOG1("Poll state of transaction %d (%s)", ts->xid, ts->gid);				
 				MtmBroadcastPollMessage(ts);
@@ -1458,7 +1461,9 @@ static void MtmDisableNode(int nodeId)
 	if (nodeId != MtmNodeId) { 
 		Mtm->nLiveNodes -= 1;
 	}
+	MtmUnlock();
 	MtmPollStatusOfPreparedTransactions(nodeId);
+	MtmLock(LW_EXCLUSIVE);
 } 
 
 static void MtmEnableNode(int nodeId)
@@ -2779,34 +2784,41 @@ void MtmReleaseRecoverySlot(int nodeId)
 	}
 }
 
-void MtmFinishPreparedTransaction(int nodeId, MtmTransState* ts, bool commit)
+void MtmRollbackPreparedTransaction(char const* gid)
 {
+	MTM_LOG1("Abort prepared transaction %s", gid);
+	if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED) == TRANSACTION_STATUS_UNKNOWN) { 
+		MTM_LOG1("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
+		MtmResetTransaction();
+		StartTransactionCommand();
+		MtmBeginSession(MtmReplicationNodeId);
+		MtmSetCurrentTransactionGID(gid);
+		FinishPreparedTransaction(gid, false);
+		CommitTransactionCommand();
+		MtmEndSession(MtmReplicationNodeId, true);
+	}
+}	
+
+
+void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit)
+{
+	if (Mtm->nodes[MtmNodeId-1].originId == InvalidRepOriginId) { 
+		/* This dummy origin is used for local commits/aborts which should not be replicated */
+		Mtm->nodes[MtmNodeId-1].originId = replorigin_create(psprintf(MULTIMASTER_SLOT_PATTERN, MtmNodeId));
+	}
 	Assert(ts->votingCompleted);
 	Assert(!IsTransactionState());
 	MtmResetTransaction();
 	StartTransactionCommand();
-	MtmBeginSession(nodeId);
+	MtmBeginSession(MtmNodeId);
 	MtmSetCurrentTransactionCSN(ts->csn);
 	MtmSetCurrentTransactionGID(ts->gid);
 	FinishPreparedTransaction(ts->gid, commit);
 	CommitTransactionCommand();
-	MtmEndSession(nodeId, true);
+	MtmEndSession(MtmNodeId, true);
 	Assert(ts->status == commit ? TRANSACTION_STATUS_COMMITTED : TRANSACTION_STATUS_ABORTED);
 }
 
-#if 0
-static void MtmFinishAllPreparedTransactions(void)
-{
-	MtmTransState* ts;
-	for (ts = Mtm->transListHead; ts != NULL; ts = ts->next) {
-		if (ts->status != TRANSACTION_STATUS_COMMITTED && ts->status != TRANSACTION_STATUS_ABORTED) {
-			MtmFinishPreparedTransaction(MtmReplicationNodeId, ts, false);
-		}
-	}
-}
-#endif
-
-
 /* 
  * Determine when and how we should open replication slot.
  * Druing recovery we need to open only one replication slot from which node should receive all transactions.
@@ -2840,11 +2852,6 @@ MtmReplicationMode MtmGetReplicationMode(int nodeId, sig_atomic_t volatile* shut
 					Mtm->nodes[i].restartLsn = InvalidXLogRecPtr;
 				}
 				MtmUnlock();
-#if 0
-				MtmBeginSession(MtmReplicationNodeId);
-				FinishAllPreparedTransactions(false);
-				MtmEndSession(MtmReplicationNodeId, true);				
-#endif
 				return REPLMODE_RECOVERY;
 			}
 		}

multimaster.h

@@ -360,6 +360,6 @@ extern void MtmReleaseRecoverySlot(int nodeId);
 extern PGconn *PQconnectdb_safe(const char *conninfo);
 extern void MtmBeginSession(int nodeId);
 extern void MtmEndSession(int nodeId, bool unlock);
-extern void MtmFinishPreparedTransaction(int nodeId, MtmTransState* ts, bool commit);
-
+extern void MtmFinishPreparedTransaction(MtmTransState* ts, bool commit);
+extern void MtmRollbackPreparedTransaction(char const* gid);
 #endif

pglogical_apply.c

@@ -425,6 +425,12 @@ process_remote_message(StringInfo s)
 			}
 			break;
 		}
+	    case 'A':
+		{
+		    MtmRollbackPreparedTransaction(messageBody);
+			standalone = true;
+			break;
+		}
 		case 'L':
 		{
 			MTM_LOG3("%ld: Process deadlock message with size %d from %d", MtmGetSystemTime(), messageSize, MtmReplicationNodeId);
@@ -590,8 +596,8 @@ process_remote_commit(StringInfo in)
 	MtmReplicationNodeId = pq_getmsgbyte(in);
 
 	/* read fields */
-	replorigin_session_origin_lsn = pq_getmsgint64(in); /* commit_lsn */
-	end_lsn = pq_getmsgint64(in); /* end_lsn */
+	pq_getmsgint64(in); /* commit_lsn */
+	replorigin_session_origin_lsn = end_lsn = pq_getmsgint64(in); /* end_lsn */
 	replorigin_session_origin_timestamp = pq_getmsgint64(in); /* commit_time */
 
 	origin_node = pq_getmsgbyte(in);
@@ -609,6 +615,7 @@ process_remote_commit(StringInfo in)
 				Assert(TransactionIdIsValid(MtmGetCurrentTransactionId()));
 				MtmBeginSession(MtmReplicationNodeId);
 				CommitTransactionCommand();
+				MtmEndSession(MtmReplicationNodeId, true);
 			}
 			break;
 		}
@@ -639,6 +646,7 @@ process_remote_commit(StringInfo in)
 					FinishPreparedTransaction(gid, false);
 					CommitTransactionCommand();					
 				}	
+				MtmEndSession(MtmReplicationNodeId, true);
 			}
 			break;
 		}
@@ -655,34 +663,19 @@ process_remote_commit(StringInfo in)
 			MtmSetCurrentTransactionGID(gid);
 			FinishPreparedTransaction(gid, true);
 			CommitTransactionCommand();
+			MtmEndSession(MtmReplicationNodeId, true);
 			break;
 		}
 		case PGLOGICAL_ABORT_PREPARED:
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
-			MTM_LOG1("PGLOGICAL_ABORT_PREPARED commit: gid=%s",  gid);
-			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED) == TRANSACTION_STATUS_UNKNOWN) { 
-				MTM_LOG1("PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", gid);
-				MtmResetTransaction();
-				StartTransactionCommand();
-				MtmBeginSession(MtmReplicationNodeId);
-				MtmSetCurrentTransactionGID(gid);
-				FinishPreparedTransaction(gid, false);
-				CommitTransactionCommand();
-			}
+			MtmRollbackPreparedTransaction(gid);
 			break;
 		}
 		default:
 			Assert(false);
 	}
-#if 0 /* Do ont need to advance slot position here: it will be done by transaction commit */
-	if (replorigin_session_origin != InvalidRepOriginId) { 
-		replorigin_advance(replorigin_session_origin, end_lsn,
-						   XactLastCommitEnd, false, false);
-	}
-#endif
-	MtmEndSession(MtmReplicationNodeId, true);
 	MtmUpdateLsnMapping(MtmReplicationNodeId, end_lsn);
 	if (flags & PGLOGICAL_CAUGHT_UP) {
 		MtmRecoveryCompleted();

pglogical_proto.c

@@ -133,29 +133,29 @@ static void
 pglogical_write_message(StringInfo out,
 						const char *prefix, Size sz, const char *message)
 {
-	if (*prefix == 'L')
-	{
-		MTM_LOG1("Send deadlock message to node %d", MtmReplicationNodeId);
-	}
-	else if (*prefix == 'D')
-	{
-		if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN)
-		{
+	switch (*prefix) { 
+	  case 'L':
+		if (MtmIsRecoveredNode(MtmReplicationNodeId)) { 			
+			return;
+		} else { 
+			MTM_LOG1("Send deadlock message to node %d", MtmReplicationNodeId);
+		}
+		break;
+	  case 'D':
+		if (MtmTransactionSnapshot(MtmCurrentXid) == INVALID_CSN)	{
 			MTM_LOG2("%d: pglogical_write_message filtered", MyProcPid);
 			return;
 		}
 		DDLInProress = true;
-	}
-	else if (*prefix == 'E')
-	{
+		break;
+	  case 'E':
 		DDLInProress = false;
 		/*
 		 * we use End message only as indicator of DDL transaction finish,
 		 * so no need to send that to replicas.
 		 */
 		return;
 	}
-
 	pq_sendbyte(out, 'M');
 	pq_sendbyte(out, *prefix);
 	pq_sendint(out, sz, 4);