fix prepare error propagation to publisher

Author: kelvich

src/include/multimaster.h

@@ -89,14 +89,16 @@ typedef uint64 nodemask_t;
 typedef struct
 {
 	int			node;			/* One based id of node initiating transaction */
-	TransactionId xid;			/* Transaction ID at this node */
+	TransactionId xid;			/* Transaction ID at origin node */
+	TransactionId my_xid;			/* Transaction ID at our node */
 } GlobalTransactionId;
 
 typedef struct
 {
-	bool		contains_ddl;
+	bool		contains_temp_ddl;
+	bool		contains_persistent_ddl;
 	bool		contains_dml;	/* transaction contains DML statements */
-	bool		accessed_temp;
+	// bool		accessed_temp;
 	bool		explicit_twophase;	/* user level 2PC */
 	bool		distributed;
 	pgid_t		gid;			/* global transaction identifier (only in case

src/pglogical_apply.c

@@ -479,8 +479,6 @@ create_rel_estate(Relation rel)
 static bool
 process_remote_begin(StringInfo s, GlobalTransactionId *gtid)
 {
-	TransactionId xid;
-
 	gtid->node = pq_getmsgint(s, 4);
 	gtid->xid = pq_getmsgint64(s);
 
@@ -496,8 +494,8 @@ process_remote_begin(StringInfo s, GlobalTransactionId *gtid)
 	StartTransactionCommand();
 
 	/* ddd */
-	xid = GetCurrentTransactionId();
-	MtmDeadlockDetectorAddXact(xid, gtid);
+	gtid->my_xid = GetCurrentTransactionId();
+	MtmDeadlockDetectorAddXact(gtid->my_xid, gtid);
 
 	// AcceptInvalidationMessages();
 	// if (!receiver_mtm_cfg_valid)
@@ -955,18 +953,17 @@ process_remote_commit(StringInfo in, GlobalTransactionId *current_gtid, MtmRecei
 			res = PrepareTransactionBlock(gid);
 			mtm_log(MtmTxFinish, "TXFINISH: %s prepared", gid);
 
+			CommitTransactionCommand();
+
 			/*
-			 * Clean this before CommitTransactionCommand(). Otherwise we'll
-			 * have a time window where exception handler will try to call
-			 * MtmDeadlockDetectorRemoveXact() with InvalidTransactionId,
-			 * because there is already no active transaction.
+			 * Clean this after CommitTransactionCommand(), as it may throw an
+			 * error that we should propagate to the originating node.
 			 */
 			current_gtid->node = 0;
 			current_gtid->xid = InvalidTransactionId;
+			current_gtid->my_xid = InvalidTransactionId;
 			MtmDeadlockDetectorRemoveXact(xid);
 
-			CommitTransactionCommand();
-
 			if (receiver_ctx->parallel_allowed)
 			{
 				TransactionId origin_xid = MtmGidParseXid(gid);
@@ -1614,7 +1611,7 @@ MtmExecutor(void* work, size_t size, MtmReceiverContext *receiver_ctx)
 		/* handle only prepare errors here */
 		if (TransactionIdIsValid(current_gtid.xid))
 		{
-			MtmDeadlockDetectorRemoveXact(GetCurrentTransactionIdIfAny());
+			MtmDeadlockDetectorRemoveXact(current_gtid.my_xid);
 			if (receiver_ctx->parallel_allowed)
 				mtm_send_reply(current_gtid.xid, current_gtid.node, MSG_ABORTED);
 		}