do not remove prepared tx from running list while 2pc decoding. fixes #23

Author: kelvich

src/backend/replication/logical/decode.c

@@ -574,7 +574,7 @@ DecodeCommit(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	}
 
 	SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
-					   parsed->nsubxacts, parsed->subxacts);
+					   parsed->nsubxacts, parsed->subxacts, true);
 
 	/* ----
 	 * Check whether we are interested in this specific transaction, and tell
@@ -678,7 +678,7 @@ DecodePrepare(LogicalDecodingContext *ctx, XLogRecordBuffer *buf,
 	}
 
 	SnapBuildCommitTxn(ctx->snapshot_builder, buf->origptr, xid,
-					   parsed->nsubxacts, parsed->subxacts);
+					   parsed->nsubxacts, parsed->subxacts, false);
 
 	if (SnapBuildXactNeedsSkip(ctx->snapshot_builder, buf->origptr) ||
 		(parsed->dbId != InvalidOid && parsed->dbId != ctx->slot->data.database) ||

src/backend/replication/logical/snapbuild.c

@@ -950,7 +950,7 @@ SnapBuildAbortTxn(SnapBuild *builder, XLogRecPtr lsn,
  */
 void
 SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
-				   int nsubxacts, TransactionId *subxacts)
+				   int nsubxacts, TransactionId *subxacts, bool isCommit)
 {
 	int			nxact;
 
@@ -1026,7 +1026,8 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
 	 * Make sure toplevel txn is not tracked in running txn's anymore, switch
 	 * state to consistent if possible.
 	 */
-	SnapBuildEndTxn(builder, lsn, xid);
+	if (isCommit)
+		SnapBuildEndTxn(builder, lsn, xid);
 
 	if (forced_timetravel)
 	{

src/include/replication/snapbuild.h

@@ -70,7 +70,7 @@ extern bool SnapBuildXactNeedsSkip(SnapBuild *snapstate, XLogRecPtr ptr);
 
 extern void SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn,
 				   TransactionId xid, int nsubxacts,
-				   TransactionId *subxacts);
+				   TransactionId *subxacts, bool isCommit);
 extern void SnapBuildAbortTxn(SnapBuild *builder, XLogRecPtr lsn,
 				  TransactionId xid, int nsubxacts,
 				  TransactionId *subxacts);