twophase_replay.v2.diff

diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index c4fd9ef..55352a6 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -45,8 +45,8 @@
  *		  fsynced
  *		* If COMMIT happens after checkpoint then backend reads state data from
  *		  files
- *		* In case of crash replay will move data from xlog to files, if that
- *		  hasn't happened before. XXX TODO - move to shmem in replay also
+ *
+ *		The same procedure happens during replication and crash recovery.
  *
  *-------------------------------------------------------------------------
  */
@@ -1252,8 +1252,6 @@ XlogReadTwoPhaseData(XLogRecPtr lsn, char **buf, int *len)
 	XLogReaderState *xlogreader;
 	char	   *errormsg;
 
-	Assert(!RecoveryInProgress());
-
 	xlogreader = XLogReaderAllocate(&read_local_xlog_page, NULL);
 	if (!xlogreader)
 		ereport(ERROR,
@@ -1296,12 +1294,30 @@ StandbyTransactionIdIsPrepared(TransactionId xid)
 	char	   *buf;
 	TwoPhaseFileHeader *hdr;
 	bool		result;
+	int			i;
 
 	Assert(TransactionIdIsValid(xid));
 
 	if (max_prepared_xacts <= 0)
 		return false;			/* nothing to do */
 
+	/*
+	 * At first check prepared tx that wasn't yet moved to disk.
+	 */
+	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
+	{
+		GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
+		PGXACT *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
+
+		if (TransactionIdEquals(pgxact->xid, xid))
+		{
+			LWLockRelease(TwoPhaseStateLock);
+			return true;
+		}
+	}
+	LWLockRelease(TwoPhaseStateLock);
+
 	/* Read and validate file */
 	buf = ReadTwoPhaseFile(xid, false);
 	if (buf == NULL)
@@ -1576,6 +1592,110 @@ RecreateTwoPhaseFile(TransactionId xid, void *content, int len)
 				 errmsg("could not close two-phase state file: %m")));
 }
 
+
+/*
+ * XlogRedoFinishPrepared()
+ *
+ * This function is called during replay when xlog reader faces 2pc commit or
+ * abort record. That function should clean up memory state that was created
+ * while replaying prepare xlog record.
+ * Actions are the same as in FinishPreparedTransaction() but without any
+ * writes to xlog and files (as it was already done).
+ */
+void
+XlogRedoFinishPrepared(TransactionId xid, bool isCommit)
+{
+	int			i;
+	char 	   *buf;
+	char	   *bufptr;
+	TwoPhaseFileHeader *hdr;
+	TransactionId latestXid;
+	TransactionId *children;
+
+	GlobalTransaction gxact;
+	PGPROC	   *proc;
+	PGXACT	   *pgxact;
+
+	/* During replay that lock isn't really necessary, but let's take it anyway */
+	LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
+	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
+	{
+		gxact = TwoPhaseState->prepXacts[i];
+		proc = &ProcGlobal->allProcs[gxact->pgprocno];
+		pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
+
+		if (TransactionIdEquals(xid, pgxact->xid))
+		{
+			gxact->locking_backend = MyBackendId;
+			MyLockedGxact = gxact;
+			break;
+		}
+	}
+	LWLockRelease(TwoPhaseStateLock);
+
+	/*
+	 * If requested xid isn't in numPrepXacts array that means that prepare
+	 * record was moved to files before our replay started. That's okay and we
+	 * have nothing to clean.
+	 */
+	if (i == TwoPhaseState->numPrepXacts)
+		return;
+
+	if (gxact->ondisk)
+		buf = ReadTwoPhaseFile(xid, true);
+	else
+		XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL);
+
+	/*
+	 * Disassemble the header area
+	 */
+	hdr = (TwoPhaseFileHeader *) buf;
+
+	Assert(TransactionIdEquals(hdr->xid, xid));
+
+	bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
+	children = (TransactionId *) bufptr;
+	bufptr += MAXALIGN(hdr->gidlen);
+	bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
+	bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
+	bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
+	bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
+
+	/*
+	 * Here we don't need to care about putting records to xlog or
+	 * deleting files, as it already done by process that have written
+	 * that xlog record. We need just to clean up memory state.
+	 */
+	latestXid = TransactionIdLatest(xid, hdr->nsubxacts, children);
+	ProcArrayRemove(proc, latestXid);
+	gxact->valid = false;
+
+	/*
+	 * 2REVIEWER: I assume that we can skip invalidation callbacks here,
+	 * as they were executed in xact_redo_commit().
+	 */
+
+	/* And release locks */
+	if (isCommit)
+		ProcessRecords(bufptr, xid, twophase_postcommit_callbacks);
+	else
+		ProcessRecords(bufptr, xid, twophase_postabort_callbacks);
+
+	PredicateLockTwoPhaseFinish(xid, true);
+	RemoveGXact(gxact);
+	MyLockedGxact = NULL;
+
+	/*
+	 * And now we can clean up any files we may have left.
+	 */
+	if (gxact->ondisk)
+		RemoveTwoPhaseFile(xid, true);
+
+	pfree(buf);
+}
+
+
+
 /*
  * CheckPointTwoPhase -- handle 2PC component of checkpointing.
  *
@@ -1690,7 +1810,48 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
 	TransactionId *xids = NULL;
 	int			nxids = 0;
 	int			allocsize = 0;
+	int			i;
 
+	/*
+	 * We need to check the PGXACT array for prepared transactions that doesn't
+	 * have any state file in case of a slave restart with the master being off.
+	 */
+	LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+	for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
+	{
+		GlobalTransaction gxact = TwoPhaseState->prepXacts[i];
+		PGXACT	   *pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
+
+		if (!gxact->valid)
+			continue;
+
+		if (TransactionIdPrecedes(pgxact->xid, result))
+			result = pgxact->xid;
+
+		if (xids_p)
+		{
+			if (nxids == allocsize)
+			{
+				if (nxids == 0)
+				{
+					allocsize = 10;
+					xids = palloc(allocsize * sizeof(TransactionId));
+				}
+				else
+				{
+					allocsize = allocsize * 2;
+					xids = repalloc(xids, allocsize * sizeof(TransactionId));
+				}
+			}
+			xids[nxids++] = pgxact->xid;
+		}
+	}
+	LWLockRelease(TwoPhaseStateLock);
+
+
+	/*
+	 * And now scan files in pg_twophase directory
+	 */
 	cldir = AllocateDir(TWOPHASE_DIR);
 	while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
 	{
@@ -1701,7 +1862,6 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
 			char	   *buf;
 			TwoPhaseFileHeader *hdr;
 			TransactionId *subxids;
-			int			i;
 
 			xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
 
@@ -1809,97 +1969,14 @@ PrescanPreparedTransactions(TransactionId **xids_p, int *nxids_p)
 }
 
 /*
- * StandbyRecoverPreparedTransactions
- *
- * Scan the pg_twophase directory and setup all the required information to
- * allow standby queries to treat prepared transactions as still active.
- * This is never called at the end of recovery - we use
- * RecoverPreparedTransactions() at that point.
- *
- * Currently we simply call SubTransSetParent() for any subxids of prepared
- * transactions. If overwriteOK is true, it's OK if some XIDs have already
- * been marked in pg_subtrans.
- */
-void
-StandbyRecoverPreparedTransactions(bool overwriteOK)
-{
-	DIR		   *cldir;
-	struct dirent *clde;
-
-	cldir = AllocateDir(TWOPHASE_DIR);
-	while ((clde = ReadDir(cldir, TWOPHASE_DIR)) != NULL)
-	{
-		if (strlen(clde->d_name) == 8 &&
-			strspn(clde->d_name, "0123456789ABCDEF") == 8)
-		{
-			TransactionId xid;
-			char	   *buf;
-			TwoPhaseFileHeader *hdr;
-			TransactionId *subxids;
-			int			i;
-
-			xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
-
-			/* Already processed? */
-			if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
-			{
-				ereport(WARNING,
-						(errmsg("removing stale two-phase state file \"%s\"",
-								clde->d_name)));
-				RemoveTwoPhaseFile(xid, true);
-				continue;
-			}
-
-			/* Read and validate file */
-			buf = ReadTwoPhaseFile(xid, true);
-			if (buf == NULL)
-			{
-				ereport(WARNING,
-					  (errmsg("removing corrupt two-phase state file \"%s\"",
-							  clde->d_name)));
-				RemoveTwoPhaseFile(xid, true);
-				continue;
-			}
-
-			/* Deconstruct header */
-			hdr = (TwoPhaseFileHeader *) buf;
-			if (!TransactionIdEquals(hdr->xid, xid))
-			{
-				ereport(WARNING,
-					  (errmsg("removing corrupt two-phase state file \"%s\"",
-							  clde->d_name)));
-				RemoveTwoPhaseFile(xid, true);
-				pfree(buf);
-				continue;
-			}
-
-			/*
-			 * Examine subtransaction XIDs ... they should all follow main
-			 * XID.
-			 */
-			subxids = (TransactionId *)
-				(buf + MAXALIGN(sizeof(TwoPhaseFileHeader)));
-			for (i = 0; i < hdr->nsubxacts; i++)
-			{
-				TransactionId subxid = subxids[i];
-
-				Assert(TransactionIdFollows(subxid, xid));
-				SubTransSetParent(xid, subxid, overwriteOK);
-			}
-		}
-	}
-	FreeDir(cldir);
-}
-
-/*
- * RecoverPreparedTransactions
+ * RecoverPreparedFromFiles
  *
  * Scan the pg_twophase directory and reload shared-memory state for each
  * prepared transaction (reacquire locks, etc).  This is run during database
  * startup.
  */
 void
-RecoverPreparedTransactions(void)
+RecoverPreparedFromFiles(bool forceOverwriteOK)
 {
 	char		dir[MAXPGPATH];
 	DIR		   *cldir;
@@ -1922,9 +1999,25 @@ RecoverPreparedTransactions(void)
 			GlobalTransaction gxact;
 			const char	*gid;
 			int			i;
+			PGXACT	   *pgxact;
 
 			xid = (TransactionId) strtoul(clde->d_name, NULL, 16);
 
+			/* Already recovered from WAL? */
+			LWLockAcquire(TwoPhaseStateLock, LW_SHARED);
+			for (i = 0; i < TwoPhaseState->numPrepXacts; i++)
+			{
+				gxact = TwoPhaseState->prepXacts[i];
+				pgxact = &ProcGlobal->allPgXact[gxact->pgprocno];
+
+				if (TransactionIdEquals(xid, pgxact->xid))
+				{
+					LWLockRelease(TwoPhaseStateLock);
+					goto next_file;
+				}
+			}
+			LWLockRelease(TwoPhaseStateLock);
+
 			/* Already processed? */
 			if (TransactionIdDidCommit(xid) || TransactionIdDidAbort(xid))
 			{
@@ -1971,6 +2064,12 @@ RecoverPreparedTransactions(void)
 				overwriteOK = true;
 
 			/*
+			 * Caller can also force overwriteOK.
+			 */
+			if (forceOverwriteOK)
+				overwriteOK = true;
+
+			/*
 			 * Reconstruct subtrans state for the transaction --- needed
 			 * because pg_subtrans is not preserved over a restart.  Note that
 			 * we are linking all the subtransactions directly to the
@@ -2012,10 +2111,108 @@ RecoverPreparedTransactions(void)
 
 			pfree(buf);
 		}
+
+next_file:
+		continue;
+
 	}
 	FreeDir(cldir);
 }
 
+
+/*
+ * RecoverPreparedFromXLOG
+ *
+ * To avoid creation of state files during replay we registering
+ * prepare xlog records in shared memory in the same way as it happens
+ * while not in recovery. If replay faces commit xlog record before
+ * checkpoint/restartpoint happens then we avoid using files at all.
+ *
+ * We need this behaviour because the speed of the 2PC replay on the replica
+ * should be at least the same as the 2PC transaction speed of the master.
+ */
+void
+RecoverPreparedFromXLOG(XLogReaderState *record)
+{
+	bool		overwriteOK = false;
+	TransactionId xid = XLogRecGetXid(record);
+	char *buf = (char *) XLogRecGetData(record);
+	char	   *bufptr;
+	const char *gid;
+	TwoPhaseFileHeader *hdr;
+	TransactionId *subxids;
+	GlobalTransaction gxact;
+	int			i;
+
+	/* Deconstruct header */
+	hdr = (TwoPhaseFileHeader *) buf;
+	Assert(TransactionIdEquals(hdr->xid, xid));
+	bufptr = buf + MAXALIGN(sizeof(TwoPhaseFileHeader));
+	gid = (const char *) bufptr;
+	bufptr += MAXALIGN(hdr->gidlen);
+	subxids = (TransactionId *) bufptr;
+	bufptr += MAXALIGN(hdr->nsubxacts * sizeof(TransactionId));
+	bufptr += MAXALIGN(hdr->ncommitrels * sizeof(RelFileNode));
+	bufptr += MAXALIGN(hdr->nabortrels * sizeof(RelFileNode));
+	bufptr += MAXALIGN(hdr->ninvalmsgs * sizeof(SharedInvalidationMessage));
+
+	/*
+	 * It's possible that SubTransSetParent has been set before, if
+	 * the prepared transaction generated xid assignment records. Test
+	 * here must match one used in AssignTransactionId().
+	 */
+	if (InHotStandby && (hdr->nsubxacts >= PGPROC_MAX_CACHED_SUBXIDS ||
+						 XLogLogicalInfoActive()))
+		overwriteOK = true;
+
+	/*
+	 * Reconstruct subtrans state for the transaction --- needed
+	 * because pg_subtrans is not preserved over a restart.  Note that
+	 * we are linking all the subtransactions directly to the
+	 * top-level XID; there may originally have been a more complex
+	 * hierarchy, but there's no need to restore that exactly.
+	 */
+	for (i = 0; i < hdr->nsubxacts; i++)
+		SubTransSetParent(subxids[i], xid, overwriteOK);
+
+	/*
+	 * Recreate its GXACT and dummy PGPROC
+	 *
+	 * MarkAsPreparing sets prepare_start_lsn to InvalidXLogRecPtr
+	 * so next checkpoint will skip that transaction.
+	 */
+	gxact = MarkAsPreparing(xid, gid,
+							hdr->prepared_at,
+							hdr->owner, hdr->database);
+	GXactLoadSubxactData(gxact, hdr->nsubxacts, subxids);
+	MarkAsPrepared(gxact);
+
+	gxact->prepare_start_lsn = record->ReadRecPtr;
+	gxact->prepare_end_lsn = record->EndRecPtr;
+
+	/*
+	 * Recover other state (notably locks) using resource managers
+	 */
+	ProcessRecords(bufptr, xid, twophase_recover_callbacks);
+
+	/*
+	 * Release locks held by the standby process after we process each
+	 * prepared transaction. As a result, we don't need too many
+	 * additional locks at any one time.
+	 */
+	if (InHotStandby)
+		StandbyReleaseLockTree(xid, hdr->nsubxacts, subxids);
+
+	/*
+	 * We're done with recovering this transaction. Clear
+	 * MyLockedGxact, like we do in PrepareTransaction() during normal
+	 * operation.
+	 */
+	PostPrepare_Twophase();
+}
+
+
+
 /*
  *	RecordTransactionCommitPrepared
  *
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 8a2cd45..a2a2f58 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -5573,7 +5573,7 @@ xact_redo(XLogReaderState *record)
 			Assert(TransactionIdIsValid(parsed.twophase_xid));
 			xact_redo_commit(&parsed, parsed.twophase_xid,
 							 record->EndRecPtr, XLogRecGetOrigin(record));
-			RemoveTwoPhaseFile(parsed.twophase_xid, false);
+			XlogRedoFinishPrepared(parsed.twophase_xid, true);
 		}
 	}
 	else if (info == XLOG_XACT_ABORT || info == XLOG_XACT_ABORT_PREPARED)
@@ -5593,14 +5593,12 @@ xact_redo(XLogReaderState *record)
 		{
 			Assert(TransactionIdIsValid(parsed.twophase_xid));
 			xact_redo_abort(&parsed, parsed.twophase_xid);
-			RemoveTwoPhaseFile(parsed.twophase_xid, false);
+			XlogRedoFinishPrepared(parsed.twophase_xid, false);
 		}
 	}
 	else if (info == XLOG_XACT_PREPARE)
 	{
-		/* the record contents are exactly the 2PC file */
-		RecreateTwoPhaseFile(XLogRecGetXid(record),
-						  XLogRecGetData(record), XLogRecGetDataLen(record));
+		RecoverPreparedFromXLOG(record);
 	}
 	else if (info == XLOG_XACT_ASSIGNMENT)
 	{
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 5b1c361..53e2ac3 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -6627,7 +6627,7 @@ StartupXLOG(void)
 
 				ProcArrayApplyRecoveryInfo(&running);
 
-				StandbyRecoverPreparedTransactions(false);
+				RecoverPreparedFromFiles(false);
 			}
 		}
 
@@ -7369,7 +7369,7 @@ StartupXLOG(void)
 	TrimMultiXact();
 
 	/* Reload shared-memory state for prepared transactions */
-	RecoverPreparedTransactions();
+	RecoverPreparedFromFiles(false);
 
 	/*
 	 * Shutdown the recovery environment. This must occur after
@@ -9283,7 +9283,7 @@ xlog_redo(XLogReaderState *record)
 
 			ProcArrayApplyRecoveryInfo(&running);
 
-			StandbyRecoverPreparedTransactions(true);
+			RecoverPreparedFromFiles(true);
 		}
 
 		/* ControlFile->checkPointCopy always tracks the latest ckpt XID */
diff --git a/src/include/access/twophase.h b/src/include/access/twophase.h
index b7ce0c6..416ef5e 100644
--- a/src/include/access/twophase.h
+++ b/src/include/access/twophase.h
@@ -17,6 +17,7 @@
 #include "access/xlogdefs.h"
 #include "datatype/timestamp.h"
 #include "storage/lock.h"
+#include "access/xlogreader.h"
 
 /*
  * GlobalTransactionData is defined in twophase.c; other places have no
@@ -46,8 +47,8 @@ extern bool StandbyTransactionIdIsPrepared(TransactionId xid);
 
 extern TransactionId PrescanPreparedTransactions(TransactionId **xids_p,
 							int *nxids_p);
-extern void StandbyRecoverPreparedTransactions(bool overwriteOK);
-extern void RecoverPreparedTransactions(void);
+extern void RecoverPreparedFromFiles(bool overwriteOK);
+extern void RecoverPreparedFromXLOG(XLogReaderState *record);
 
 extern void RecreateTwoPhaseFile(TransactionId xid, void *content, int len);
 extern void RemoveTwoPhaseFile(TransactionId xid, bool giveWarning);
@@ -56,4 +57,5 @@ extern void CheckPointTwoPhase(XLogRecPtr redo_horizon);
 
 extern void FinishPreparedTransaction(const char *gid, bool isCommit);
 
+extern void XlogRedoFinishPrepared(TransactionId xid, bool isCommit);
 #endif   /* TWOPHASE_H */
diff --git a/src/test/recovery/t/006_twophase.pl b/src/test/recovery/t/006_twophase.pl
new file mode 100644
index 0000000..82191d4
--- /dev/null
+++ b/src/test/recovery/t/006_twophase.pl
@@ -0,0 +1,226 @@
+# Checks for recovery_min_apply_delay
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More tests => 11;
+
+# Setup master node
+my $node_master = get_new_node("Candie");
+$node_master->init(allows_streaming => 1);
+$node_master->append_conf('postgresql.conf', qq(
+max_prepared_transactions = 10
+));
+$node_master->start;
+$node_master->backup('master_backup');
+$node_master->psql('postgres', "create table t(id int)");
+
+# Setup master node
+my $node_slave = get_new_node('Django');
+$node_slave->init_from_backup($node_master, 'master_backup', has_streaming => 1);
+$node_slave->start;
+
+# Switch to synchronous replication
+$node_master->append_conf('postgresql.conf', qq(
+synchronous_standby_names = '*'
+));
+$node_master->restart;
+
+my $psql_out = '';
+my $psql_rc = '';
+
+###############################################################################
+# Check that we can commit and abort tx after soft restart.
+# Here checkpoint happens before shutdown and no WAL replay will not occur
+# during start. So postgres should re-create memory state from files.
+###############################################################################
+
+$node_master->psql('postgres', "
+	begin;
+	insert into t values (42);
+	prepare transaction 'x';
+	begin;
+	insert into t values (142);
+	prepare transaction 'y';");
+$node_master->stop;
+$node_master->start;
+
+$psql_rc = $node_master->psql('postgres', "commit prepared 'x'");
+is($psql_rc, '0', 'Commit prepared tx after restart.');
+
+$psql_rc = $node_master->psql('postgres', "rollback prepared 'y'");
+is($psql_rc, '0', 'Rollback prepared tx after restart.');
+
+###############################################################################
+# Check that we can commit and abort after hard restart.
+# On startup WAL replay will re-create memory for global transactions that
+# happend after the last checkpoint.
+###############################################################################
+
+$node_master->psql('postgres', "
+	checkpoint;
+	begin;
+	insert into t values (42);
+	prepare transaction 'x';
+	begin;
+	insert into t values (142);
+	prepare transaction 'y';");
+$node_master->teardown_node;
+$node_master->start;
+
+$psql_rc = $node_master->psql('postgres', "commit prepared 'x'");
+is($psql_rc, '0', 'Commit prepared tx after teardown.');
+
+$psql_rc = $node_master->psql('postgres', "rollback prepared 'y'");
+is($psql_rc, '0', 'Rollback prepared tx after teardown.');
+
+###############################################################################
+# Check that we can replay several tx with same name.
+###############################################################################
+
+$node_master->psql('postgres', "
+	checkpoint;
+	begin;
+	insert into t values (42);
+	prepare transaction 'x';
+	commit prepared 'x';
+	begin;
+	insert into t values (42);
+	prepare transaction 'x';");
+$node_master->teardown_node;
+$node_master->start;
+
+$psql_rc = $node_master->psql('postgres', "commit prepared 'x'");
+is($psql_rc, '0', 'Replay several tx with same name.');
+
+###############################################################################
+# Check that WAL replay will cleanup it's memory state and release locks while
+# replaying commit.
+###############################################################################
+
+$node_master->psql('postgres', "
+	begin;
+	insert into t values (42);
+	prepare transaction 'x';
+	commit prepared 'x';");
+$node_master->teardown_node;
+$node_master->start;
+$psql_rc = $node_master->psql('postgres',"
+	begin;
+	insert into t values (42);
+	-- This prepare can fail due to 2pc identifier or locks conflicts if replay
+	-- didn't fully cleanup it's state on commit.
+	prepare transaction 'x';");
+is($psql_rc, '0', "Check that replay will cleanup it's memory state");
+
+$node_master->psql('postgres', "commit prepared 'x'");
+
+###############################################################################
+# Check that WAL replay will cleanup it's memory state on running slave.
+###############################################################################
+
+$node_master->psql('postgres', "
+	begin;
+	insert into t values (42);
+	prepare transaction 'x';
+	commit prepared 'x';
+	");
+$node_slave->psql('postgres', "select count(*) from pg_prepared_xacts;", stdout => \$psql_out);
+is($psql_out, '0', "Check that replay will cleanup it's memory state on running slave");
+
+###############################################################################
+# The same as in previous case, but let's force checkpoint on slave between
+# prepare and commit.
+###############################################################################
+
+$node_master->psql('postgres', "
+	begin;
+	insert into t values (42);
+	prepare transaction 'x';
+	");
+$node_slave->psql('postgres',"checkpoint;");
+$node_master->psql('postgres', "commit prepared 'x';");
+$node_slave->psql('postgres', "select count(*) from pg_prepared_xacts;", stdout => \$psql_out);
+is($psql_out, '0', "Check that replay will cleanup it's memory state on slave after checkpoint");
+
+###############################################################################
+# Check that we can commit transaction on promoted slave.
+###############################################################################
+
+$node_master->psql('postgres', "
+	begin;
+	insert into t values (42);
+	prepare transaction 'x';
+	");
+$node_master->teardown_node;
+$node_slave->promote;
+$node_slave->poll_query_until('postgres', "SELECT pg_is_in_recovery() <> true");
+
+$psql_rc = $node_slave->psql('postgres', "commit prepared 'x';");
+is($psql_rc, '0', "Restore prepared transaction on promoted slave.");
+
+# change roles
+($node_master, $node_slave) = ($node_slave, $node_master);
+$node_slave->enable_streaming($node_master);
+$node_slave->append_conf('recovery.conf', qq(
+recovery_target_timeline='latest'
+));
+$node_slave->start;
+
+###############################################################################
+# Check that we restore prepared xacts after slave soft restart while master is
+# down. Since slave knows that master is down it uses different code path on
+# start.
+###############################################################################
+
+$node_master->psql('postgres', "
+	begin;
+	insert into t values (42);
+	prepare transaction 'x';
+	");
+$node_master->stop;
+$node_slave->restart;
+$node_slave->promote;
+$node_slave->poll_query_until('postgres', "SELECT pg_is_in_recovery() <> true");
+
+$node_slave->psql('postgres',"select count(*) from pg_prepared_xacts", stdout => \$psql_out);
+is($psql_out, '1', "Restore prepared xacts after slave soft restart while master is down.");
+
+# restore state
+($node_master, $node_slave) = ($node_slave, $node_master);
+$node_slave->enable_streaming($node_master);
+$node_slave->append_conf('recovery.conf', qq(
+recovery_target_timeline='latest'
+));
+$node_slave->start;
+$node_master->psql('postgres',"commit prepared 'x'");
+
+###############################################################################
+# Check that we restore prepared xacts after slave hard restart while master is
+# down.
+###############################################################################
+
+$node_master->psql('postgres', "
+	begin;
+	insert into t values (242);
+	prepare transaction 'x';
+	");
+$node_master->stop;
+$node_slave->teardown_node;
+$node_slave->start;
+$node_slave->promote;
+$node_slave->poll_query_until('postgres', "SELECT pg_is_in_recovery() <> true");
+
+$node_slave->psql('postgres',"select count(*) from pg_prepared_xacts", stdout => \$psql_out);
+is($psql_out, '1', "Restore prepared xacts after slave hard restart while master is down.");
+
+# restore state
+($node_master, $node_slave) = ($node_slave, $node_master);
+$node_slave->enable_streaming($node_master);
+$node_slave->append_conf('recovery.conf', qq(
+recovery_target_timeline='latest'
+));
+$node_slave->start;
+$node_master->psql('postgres',"commit prepared 'x'");
+
+