Fix prepare/abort_prepared race condition

Author: knizhnik

arbiter.c

@@ -915,17 +915,15 @@ static void MtmTransReceiver(Datum arg)
 					} else { 
 						switch (msg->code) { 
 						  case MSG_PREPARE:
-							Assert(ts->status == TRANSACTION_STATUS_IN_PROGRESS);
-							ts->status = TRANSACTION_STATUS_UNKNOWN;
-							ts->csn = MtmAssignCSN();
-							MtmAdjustSubtransactions(ts);
-							MtmSendNotificationMessage(ts, MSG_PREPARED);
-#if 0							
+							if (ts->status == TRANSACTION_STATUS_IN_PROGRESS) {
+								ts->status = TRANSACTION_STATUS_UNKNOWN;
+								ts->csn = MtmAssignCSN();
+								MtmAdjustSubtransactions(ts);
+								MtmSendNotificationMessage(ts, MSG_PREPARED);
 							} else {
 								Assert(ts->status == TRANSACTION_STATUS_ABORTED);
 								MtmSendNotificationMessage(ts, MSG_ABORTED);
 							}
-#endif
 							break;
 						  default:
 							Assert(false);

multimaster.c

@@ -77,6 +77,8 @@ typedef struct {
 
 typedef struct {
 	char gid[MULTIMASTER_MAX_GID_SIZE];
+	bool abort;
+	XidStatus status;
 	MtmTransState* state;
 } MtmTransMap;
 
@@ -274,7 +276,7 @@ timestamp_t MtmGetSystemTime(void)
 {
     struct timeval tv;
     gettimeofday(&tv, NULL);
-    return (timestamp_t)tv.tv_sec*USECS_PER_SEC + tv.tv_usec + Mtm->timeShift;
+    return (timestamp_t)tv.tv_sec*USECS_PER_SEC + tv.tv_usec;
 }
 
 timestamp_t MtmGetCurrentTime(void)
@@ -721,11 +723,12 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 { 
 	MtmTransState* ts;
 	TransactionId* subxids;
-	
+
 	if (!x->isDistributed) {
 		return;
 	}
 
+
 	if (Mtm->inject2PCError == 1) { 
 		Mtm->inject2PCError = 0;
 		elog(ERROR, "ERROR INJECTION for transaction %d (%s)", x->xid, x->gid);
@@ -754,7 +757,6 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 	if (!x->isReplicated) { 
 		MtmCheckClusterLock();
 	}
-
 	ts = MtmCreateTransState(x);
 	/* 
 	 * Invalid CSN prevent replication of transaction by logical replication 
@@ -779,8 +781,7 @@ MtmPrePrepareTransaction(MtmCurrentTrans* x)
 	MtmAddSubtransactions(ts, subxids, ts->nSubxids);
 	MTM_LOG3("%d: MtmPrePrepareTransaction prepare commit of %d (gtid.xid=%d, gtid.node=%d, CSN=%ld)", 
 			 MyProcPid, x->xid, ts->gtid.xid, ts->gtid.node, ts->csn);
-	MtmUnlock();
-
+	MtmUnlock(); 
 }
 
 /*
@@ -818,7 +819,8 @@ MtmPostPrepareTransaction(MtmCurrentTrans* x)
 	Assert(ts != NULL);
 
 	if (!MtmIsCoordinator(ts) || Mtm->status == MTM_RECOVERY) {
-		MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, NULL);
+		bool found;
+		MtmTransMap* tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_ENTER, &found);
 		Assert(x->gid[0]);
 		tm->state = ts;	
 		ts->votingCompleted = true;
@@ -876,8 +878,8 @@ MtmAbortPreparedTransaction(MtmCurrentTrans* x)
 	if (x->status != TRANSACTION_STATUS_ABORTED) { 
 		MtmLock(LW_EXCLUSIVE);
 		tm = (MtmTransMap*)hash_search(MtmGid2State, x->gid, HASH_REMOVE, NULL);
-		Assert(tm != NULL);
-		MTM_LOG1("Abort prepared transaction %d with gid='%s' is already aborted", x->xid, x->gid);
+		Assert(tm != NULL && tm->state != NULL);
+		MTM_LOG1("%ld: Abort prepared transaction %d with gid='%s'", MtmGetSystemTime(), x->xid, x->gid);
 		MtmAbortTransaction(tm->state);
 		MtmUnlock();
 		x->status = TRANSACTION_STATUS_ABORTED;
@@ -1015,21 +1017,26 @@ XidStatus MtmGetCurrentTransactionStatus(void)
 	return MtmTx.status;
 }
 
-XidStatus MtmGetGlobalTransactionStatus(char const* gid)
+XidStatus MtmExchangeGlobalTransactionStatus(char const* gid, XidStatus new_status)
 {
-	XidStatus status;
 	MtmTransMap* tm;
+	bool found;
+	XidStatus old_status = TRANSACTION_STATUS_IN_PROGRESS;
 
 	Assert(gid[0]);
-	MtmLock(LW_SHARED);
-	tm = (MtmTransMap*)hash_search(MtmGid2State, gid, HASH_FIND, NULL);
-	if (tm != NULL) {
-		status = tm->state->status;
+	MtmLock(LW_EXCLUSIVE);
+	tm = (MtmTransMap*)hash_search(MtmGid2State, gid, HASH_ENTER, &found);
+	if (found) {
+		old_status = tm->status;
+		if (old_status != TRANSACTION_STATUS_ABORTED) { 
+			tm->status = new_status;
+		}
 	} else { 
-		status = TRANSACTION_STATUS_ABORTED;
+		tm->state = NULL;
+		tm->status = new_status;
 	}
 	MtmUnlock();
-	return status;
+	return old_status;
 }
 
 void  MtmSetCurrentTransactionCSN(csn_t csn)

multimaster.h

@@ -256,7 +256,7 @@ extern csn_t MtmGetTransactionCSN(TransactionId xid);
 extern void  MtmSetCurrentTransactionCSN(csn_t csn);
 extern TransactionId MtmGetCurrentTransactionId(void);
 extern XidStatus MtmGetCurrentTransactionStatus(void);
-extern XidStatus MtmGetGlobalTransactionStatus(char const* gid);
+extern XidStatus MtmExchangeGlobalTransactionStatus(char const* gid, XidStatus status);
 extern bool  MtmIsRecoveredNode(int nodeId);
 extern bool  MtmRefreshClusterStatus(bool nowait);
 extern void  MtmSwitchClusterMode(MtmNodeStatus mode);

pglogical_apply.c

@@ -508,7 +508,8 @@ process_remote_commit(StringInfo in)
 	uint8 		flags;
 	csn_t       csn;
 	const char *gid = NULL;	
-	XLogRecPtr end_lsn;
+	XLogRecPtr  end_lsn;
+
 	/* read flags */
 	flags = pq_getmsgbyte(in);
 	MtmReplicationNodeId = pq_getmsgbyte(in);
@@ -536,25 +537,38 @@ process_remote_commit(StringInfo in)
 		{
 			Assert(IsTransactionState() && TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
-			/* prepare TBLOCK_INPROGRESS state for PrepareTransactionBlock() */
-			MTM_LOG1("%d: PGLOGICAL_PREPARE commit: gid=%s", MyProcPid, gid);
-			BeginTransactionBlock();
-			CommitTransactionCommand();
-			StartTransactionCommand();
-			
-			MtmBeginSession();
-			/* PREPARE itself */
-			MtmSetCurrentTransactionGID(gid);
-			PrepareTransactionBlock(gid);
-			CommitTransactionCommand();
+			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_IN_PROGRESS) == TRANSACTION_STATUS_ABORTED) { 
+				MTM_LOG1("%ld: avoid prepare of previously aborted global transaction %s", MtmGetSystemTime(), gid);	
+				AbortCurrentTransaction();
+			} else { 				
+				/* prepare TBLOCK_INPROGRESS state for PrepareTransactionBlock() */
+				MTM_LOG1("%ld: PGLOGICAL_PREPARE commit: gid=%s", MtmGetSystemTime(), gid);
+				BeginTransactionBlock();
+				CommitTransactionCommand();
+				StartTransactionCommand();
+				
+				MtmBeginSession();
+				/* PREPARE itself */
+				MtmSetCurrentTransactionGID(gid);
+				PrepareTransactionBlock(gid);
+				CommitTransactionCommand();
+
+				if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_UNKNOWN) == TRANSACTION_STATUS_ABORTED) { 
+					MTM_LOG1("%ld: perform delayed rollback of prepared global transaction %s", MtmGetSystemTime(), gid);	
+					StartTransactionCommand();
+					MtmSetCurrentTransactionGID(gid);
+					FinishPreparedTransaction(gid, false);
+					CommitTransactionCommand();					
+				}	
+			}
 			break;
 		}
 		case PGLOGICAL_COMMIT_PREPARED:
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			csn = pq_getmsgint64(in); 
 			gid = pq_getmsgstring(in);
-			MTM_LOG1("%d: PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s", MyProcPid, csn, gid);
+			MTM_LOG1("%ld: PGLOGICAL_COMMIT_PREPARED commit: csn=%ld, gid=%s", MtmGetSystemTime(), csn, gid);
 			StartTransactionCommand();
 			MtmBeginSession();
 			MtmSetCurrentTransactionCSN(csn);
@@ -567,9 +581,9 @@ process_remote_commit(StringInfo in)
 		{
 			Assert(!TransactionIdIsValid(MtmGetCurrentTransactionId()));
 			gid = pq_getmsgstring(in);
-			MTM_LOG1("%d: PGLOGICAL_ABORT_PREPARED commit: gid=%s", MyProcPid, gid);
-			if (MtmGetGlobalTransactionStatus(gid) != TRANSACTION_STATUS_ABORTED) { 
-				MTM_LOG2("%d: PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", MyProcPid, gid);
+			MTM_LOG1("%ld: PGLOGICAL_ABORT_PREPARED commit: gid=%s",  MtmGetSystemTime(), gid);
+			if (MtmExchangeGlobalTransactionStatus(gid, TRANSACTION_STATUS_ABORTED) ==TRANSACTION_STATUS_UNKNOWN) { 
+				MTM_LOG1("%ld: PGLOGICAL_ABORT_PREPARED commit: gid=%s #2", MtmGetSystemTime(), gid);
 				StartTransactionCommand();
 				MtmSetCurrentTransactionGID(gid);
 				FinishPreparedTransaction(gid, false);

pglogical_proto.c

@@ -150,7 +150,8 @@ pglogical_write_commit(StringInfo out, PGLogicalOutputData *data,
 		 * INVALID_CSN means replicated transaction (transaction initiated by some other nodes).
 		 * We do not need to send such transactions unless we perform recovery
 		 */
-		if (csn == INVALID_CSN && !isRecovery) {
+		if (csn == INVALID_CSN && !isRecovery) 
+		{
 			return;
 		}
 		if (MtmRecoveryCaughtUp(MtmReplicationNodeId, txn->end_lsn)) {