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'"); + +