/*-------------------------------------------------------------------------

 *

 * xlogrecovery.c

 *    Functions for WAL recovery, standby mode

 *

 * This source file contains functions controlling WAL recovery.

 * InitWalRecovery() initializes the system for crash or archive recovery,

 * or standby mode, depending on configuration options and the state of

 * the control file and possible backup label file.  PerformWalRecovery()

 * performs the actual WAL replay, calling the rmgr-specific redo routines.

 * FinishWalRecovery() performs end-of-recovery checks and cleanup actions,

 * and prepares information needed to initialize the WAL for writes.  In

 * addition to these three main functions, there are a bunch of functions

 * for interrogating recovery state and controlling the recovery process.

 *

 *

 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994, Regents of the University of California

 *

 * src/backend/access/transam/xlogrecovery.c

 *

 *-------------------------------------------------------------------------

 */

#include "postgres.h"

#include <ctype.h>

#include <math.h>

#include <time.h>

#include <sys/stat.h>

#include <sys/time.h>

#include <unistd.h>

#include "access/timeline.h"

#include "access/transam.h"

#include "access/xact.h"

#include "access/xlog_internal.h"

#include "access/xlogarchive.h"

#include "access/xlogprefetcher.h"

#include "access/xlogreader.h"

#include "access/xlogrecovery.h"

#include "access/xlogutils.h"

#include "backup/basebackup.h"

#include "catalog/pg_control.h"

#include "commands/tablespace.h"

#include "common/file_utils.h"

#include "miscadmin.h"

#include "pgstat.h"

#include "postmaster/bgwriter.h"

#include "postmaster/startup.h"

#include "replication/slot.h"

#include "replication/slotsync.h"

#include "replication/walreceiver.h"

#include "storage/fd.h"

#include "storage/ipc.h"

#include "storage/latch.h"

#include "storage/pmsignal.h"

#include "storage/procarray.h"

#include "storage/spin.h"

#include "utils/datetime.h"

#include "utils/fmgrprotos.h"

#include "utils/guc_hooks.h"

#include "utils/pgstat_internal.h"

#include "utils/pg_lsn.h"

#include "utils/ps_status.h"

#include "utils/pg_rusage.h"

/* Unsupported old recovery command file names (relative to $PGDATA) */

#define RECOVERY_COMMAND_FILE "recovery.conf"

#define RECOVERY_COMMAND_DONE "recovery.done"

/*

 * GUC support

 */

const struct config_enum_entry recovery_target_action_options[] = {

  {"pause", RECOVERY_TARGET_ACTION_PAUSE, false},

  {"promote", RECOVERY_TARGET_ACTION_PROMOTE, false},

  {"shutdown", RECOVERY_TARGET_ACTION_SHUTDOWN, false},

  {NULL, 0, false}

};

/* options formerly taken from recovery.conf for archive recovery */

char     *recoveryRestoreCommand = NULL;

char     *recoveryEndCommand = NULL;

char     *archiveCleanupCommand = NULL;

RecoveryTargetType recoveryTarget = RECOVERY_TARGET_UNSET;

bool    recoveryTargetInclusive = true;

int     recoveryTargetAction = RECOVERY_TARGET_ACTION_PAUSE;

TransactionId recoveryTargetXid;

char     *recovery_target_time_string;

TimestampTz recoveryTargetTime;

const char *recoveryTargetName;

XLogRecPtr  recoveryTargetLSN;

int     recovery_min_apply_delay = 0;

/* options formerly taken from recovery.conf for XLOG streaming */

char     *PrimaryConnInfo = NULL;

char     *PrimarySlotName = NULL;

bool    wal_receiver_create_temp_slot = false;

/*

 * recoveryTargetTimeLineGoal: what the user requested, if any

 *

 * recoveryTargetTLIRequested: numeric value of requested timeline, if constant

 *

 * recoveryTargetTLI: the currently understood target timeline; changes

 *

 * expectedTLEs: a list of TimeLineHistoryEntries for recoveryTargetTLI and

 * the timelines of its known parents, newest first (so recoveryTargetTLI is

 * always the first list member).  Only these TLIs are expected to be seen in

 * the WAL segments we read, and indeed only these TLIs will be considered as

 * candidate WAL files to open at all.

 *

 * curFileTLI: the TLI appearing in the name of the current input WAL file.

 * (This is not necessarily the same as the timeline from which we are

 * replaying WAL, which StartupXLOG calls replayTLI, because we could be

 * scanning data that was copied from an ancestor timeline when the current

 * file was created.)  During a sequential scan we do not allow this value

 * to decrease.

 */

RecoveryTargetTimeLineGoal recoveryTargetTimeLineGoal = RECOVERY_TARGET_TIMELINE_LATEST;

TimeLineID  recoveryTargetTLIRequested = 0;

TimeLineID  recoveryTargetTLI = 0;

static List *expectedTLEs;

static TimeLineID curFileTLI;

/*

 * When ArchiveRecoveryRequested is set, archive recovery was requested,

 * ie. signal files were present.  When InArchiveRecovery is set, we are

 * currently recovering using offline XLOG archives.  These variables are only

 * valid in the startup process.

 *

 * When ArchiveRecoveryRequested is true, but InArchiveRecovery is false, we're

 * currently performing crash recovery using only XLOG files in pg_wal, but

 * will switch to using offline XLOG archives as soon as we reach the end of

 * WAL in pg_wal.

 */

bool    ArchiveRecoveryRequested = false;

bool    InArchiveRecovery = false;

/*

 * When StandbyModeRequested is set, standby mode was requested, i.e.

 * standby.signal file was present.  When StandbyMode is set, we are currently

 * in standby mode.  These variables are only valid in the startup process.

 * They work similarly to ArchiveRecoveryRequested and InArchiveRecovery.

 */

static bool StandbyModeRequested = false;

bool    StandbyMode = false;

/* was a signal file present at startup? */

static bool standby_signal_file_found = false;

static bool recovery_signal_file_found = false;

/*

 * CheckPointLoc is the position of the checkpoint record that determines

 * where to start the replay.  It comes from the backup label file or the

 * control file.

 *

 * RedoStartLSN is the checkpoint's REDO location, also from the backup label

 * file or the control file.  In standby mode, XLOG streaming usually starts

 * from the position where an invalid record was found.  But if we fail to

 * read even the initial checkpoint record, we use the REDO location instead

 * of the checkpoint location as the start position of XLOG streaming.

 * Otherwise we would have to jump backwards to the REDO location after

 * reading the checkpoint record, because the REDO record can precede the

 * checkpoint record.

 */

static XLogRecPtr CheckPointLoc = InvalidXLogRecPtr;

static TimeLineID CheckPointTLI = 0;

static XLogRecPtr RedoStartLSN = InvalidXLogRecPtr;

static TimeLineID RedoStartTLI = 0;

/*

 * Local copy of SharedHotStandbyActive variable. False actually means "not

 * known, need to check the shared state".

 */

static bool LocalHotStandbyActive = false;

/*

 * Local copy of SharedPromoteIsTriggered variable. False actually means "not

 * known, need to check the shared state".

 */

static bool LocalPromoteIsTriggered = false;

/* Has the recovery code requested a walreceiver wakeup? */

static bool doRequestWalReceiverReply;

/* XLogReader object used to parse the WAL records */

static XLogReaderState *xlogreader = NULL;

/* XLogPrefetcher object used to consume WAL records with read-ahead */

static XLogPrefetcher *xlogprefetcher = NULL;

/* Parameters passed down from ReadRecord to the XLogPageRead callback. */

typedef struct XLogPageReadPrivate

{

  int     emode;

  bool    fetching_ckpt;  /* are we fetching a checkpoint record? */

  bool    randAccess;

  TimeLineID  replayTLI;

} XLogPageReadPrivate;

/* flag to tell XLogPageRead that we have started replaying */

static bool InRedo = false;

/*

 * Codes indicating where we got a WAL file from during recovery, or where

 * to attempt to get one.

 */

typedef enum

{

  XLOG_FROM_ANY = 0,      /* request to read WAL from any source */

  XLOG_FROM_ARCHIVE,      /* restored using restore_command */

  XLOG_FROM_PG_WAL,     /* existing file in pg_wal */

  XLOG_FROM_STREAM,     /* streamed from primary */

} XLogSource;

/* human-readable names for XLogSources, for debugging output */

static const char *const xlogSourceNames[] = {"any", "archive", "pg_wal", "stream"};

/*

 * readFile is -1 or a kernel FD for the log file segment that's currently

 * open for reading.  readSegNo identifies the segment.  readOff is the offset

 * of the page just read, readLen indicates how much of it has been read into

 * readBuf, and readSource indicates where we got the currently open file from.

 *

 * Note: we could use Reserve/ReleaseExternalFD to track consumption of this

 * FD too (like for openLogFile in xlog.c); but it doesn't currently seem

 * worthwhile, since the XLOG is not read by general-purpose sessions.

 */

static int  readFile = -1;

static XLogSegNo readSegNo = 0;

static uint32 readOff = 0;

static uint32 readLen = 0;

static XLogSource readSource = XLOG_FROM_ANY;

/*

 * Keeps track of which source we're currently reading from. This is

 * different from readSource in that this is always set, even when we don't

 * currently have a WAL file open. If lastSourceFailed is set, our last

 * attempt to read from currentSource failed, and we should try another source

 * next.

 *

 * pendingWalRcvRestart is set when a config change occurs that requires a

 * walreceiver restart.  This is only valid in XLOG_FROM_STREAM state.

 */

static XLogSource currentSource = XLOG_FROM_ANY;

static bool lastSourceFailed = false;

static bool pendingWalRcvRestart = false;

/*

 * These variables track when we last obtained some WAL data to process,

 * and where we got it from.  (XLogReceiptSource is initially the same as

 * readSource, but readSource gets reset to zero when we don't have data

 * to process right now.  It is also different from currentSource, which

 * also changes when we try to read from a source and fail, while

 * XLogReceiptSource tracks where we last successfully read some WAL.)

 */

static TimestampTz XLogReceiptTime = 0;

static XLogSource XLogReceiptSource = XLOG_FROM_ANY;

/* Local copy of WalRcv->flushedUpto */

static XLogRecPtr flushedUpto = 0;

static TimeLineID receiveTLI = 0;

/*

 * Copy of minRecoveryPoint and backupEndPoint from the control file.

 *

 * In order to reach consistency, we must replay the WAL up to

 * minRecoveryPoint.  If backupEndRequired is true, we must also reach

 * backupEndPoint, or if it's invalid, an end-of-backup record corresponding

 * to backupStartPoint.

 *

 * Note: In archive recovery, after consistency has been reached, the

 * functions in xlog.c will start updating minRecoveryPoint in the control

 * file.  But this copy of minRecoveryPoint variable reflects the value at the

 * beginning of recovery, and is *not* updated after consistency is reached.

 */

static XLogRecPtr minRecoveryPoint;

static TimeLineID minRecoveryPointTLI;

static XLogRecPtr backupStartPoint;

static XLogRecPtr backupEndPoint;

static bool backupEndRequired = false;

/*

 * Have we reached a consistent database state?  In crash recovery, we have

 * to replay all the WAL, so reachedConsistency is never set.  During archive

 * recovery, the database is consistent once minRecoveryPoint is reached.

 *

 * Consistent state means that the system is internally consistent, all

 * the WAL has been replayed up to a certain point, and importantly, there

 * is no trace of later actions on disk.

 *

 * This flag is used only by the startup process and postmaster. When

 * minRecoveryPoint is reached, the startup process sets it to true and

 * sends a PMSIGNAL_RECOVERY_CONSISTENT signal to the postmaster,

 * which then sets it to true upon receiving the signal.

 */

bool    reachedConsistency = false;

/* Buffers dedicated to consistency checks of size BLCKSZ */

static char *replay_image_masked = NULL;

static char *primary_image_masked = NULL;

/*

 * Shared-memory state for WAL recovery.

 */

typedef struct XLogRecoveryCtlData

{

  /*

   * SharedHotStandbyActive indicates if we allow hot standby queries to be

   * run.  Protected by info_lck.

   */

  bool    SharedHotStandbyActive;

  /*

   * SharedPromoteIsTriggered indicates if a standby promotion has been

   * triggered.  Protected by info_lck.

   */

  bool    SharedPromoteIsTriggered;

  /*

   * recoveryWakeupLatch is used to wake up the startup process to continue

   * WAL replay, if it is waiting for WAL to arrive or promotion to be

   * requested.

   *

   * Note that the startup process also uses another latch, its procLatch,

   * to wait for recovery conflict. If we get rid of recoveryWakeupLatch for

   * signaling the startup process in favor of using its procLatch, which

   * comports better with possible generic signal handlers using that latch.

   * But we should not do that because the startup process doesn't assume

   * that it's waken up by walreceiver process or SIGHUP signal handler

   * while it's waiting for recovery conflict. The separate latches,

   * recoveryWakeupLatch and procLatch, should be used for inter-process

   * communication for WAL replay and recovery conflict, respectively.

   */

  Latch   recoveryWakeupLatch;

  /*

   * Last record successfully replayed.

   */

  XLogRecPtr  lastReplayedReadRecPtr; /* start position */

  XLogRecPtr  lastReplayedEndRecPtr;  /* end+1 position */

  TimeLineID  lastReplayedTLI;  /* timeline */

  /*

   * When we're currently replaying a record, ie. in a redo function,

   * replayEndRecPtr points to the end+1 of the record being replayed,

   * otherwise it's equal to lastReplayedEndRecPtr.

   */

  XLogRecPtr  replayEndRecPtr;

  TimeLineID  replayEndTLI;

  /* timestamp of last COMMIT/ABORT record replayed (or being replayed) */

  TimestampTz recoveryLastXTime;

  /*

   * timestamp of when we started replaying the current chunk of WAL data,

   * only relevant for replication or archive recovery

   */

  TimestampTz currentChunkStartTime;

  /* Recovery pause state */

  RecoveryPauseState recoveryPauseState;

  ConditionVariable recoveryNotPausedCV;

  slock_t   info_lck;   /* locks shared variables shown above */

} XLogRecoveryCtlData;

static XLogRecoveryCtlData *XLogRecoveryCtl = NULL;

/*

 * abortedRecPtr is the start pointer of a broken record at end of WAL when

 * recovery completes; missingContrecPtr is the location of the first

 * contrecord that went missing.  See CreateOverwriteContrecordRecord for

 * details.

 */

static XLogRecPtr abortedRecPtr;

static XLogRecPtr missingContrecPtr;

/*

 * if recoveryStopsBefore/After returns true, it saves information of the stop

 * point here

 */

static TransactionId recoveryStopXid;

static TimestampTz recoveryStopTime;

static XLogRecPtr recoveryStopLSN;

static char recoveryStopName[MAXFNAMELEN];

static bool recoveryStopAfter;

/* prototypes for local functions */

static void ApplyWalRecord(XLogReaderState *xlogreader, XLogRecord *record, TimeLineID *replayTLI);

static void EnableStandbyMode(void);

static void readRecoverySignalFile(void);

static void validateRecoveryParameters(void);

static bool read_backup_label(XLogRecPtr *checkPointLoc,

                TimeLineID *backupLabelTLI,

                bool *backupEndRequired, bool *backupFromStandby);

static bool read_tablespace_map(List **tablespaces);

static void xlogrecovery_redo(XLogReaderState *record, TimeLineID replayTLI);

static void CheckRecoveryConsistency(void);

static void rm_redo_error_callback(void *arg);

#ifdef WAL_DEBUG

static void xlog_outrec(StringInfo buf, XLogReaderState *record);

#endif

static void xlog_block_info(StringInfo buf, XLogReaderState *record);

static void checkTimeLineSwitch(XLogRecPtr lsn, TimeLineID newTLI,

                TimeLineID prevTLI, TimeLineID replayTLI);

static bool getRecordTimestamp(XLogReaderState *record, TimestampTz *recordXtime);

static void verifyBackupPageConsistency(XLogReaderState *record);

static bool recoveryStopsBefore(XLogReaderState *record);

static bool recoveryStopsAfter(XLogReaderState *record);

static char *getRecoveryStopReason(void);

static void recoveryPausesHere(bool endOfRecovery);

static bool recoveryApplyDelay(XLogReaderState *record);

static void ConfirmRecoveryPaused(void);

static XLogRecord *ReadRecord(XLogPrefetcher *xlogprefetcher,

                int emode, bool fetching_ckpt,

                TimeLineID replayTLI);

static int  XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,

             int reqLen, XLogRecPtr targetRecPtr, char *readBuf);

static XLogPageReadResult WaitForWALToBecomeAvailable(XLogRecPtr RecPtr,

                            bool randAccess,

                            bool fetching_ckpt,

                            XLogRecPtr tliRecPtr,

                            TimeLineID replayTLI,

                            XLogRecPtr replayLSN,

                            bool nonblocking);

static int  emode_for_corrupt_record(int emode, XLogRecPtr RecPtr);

static XLogRecord *ReadCheckpointRecord(XLogPrefetcher *xlogprefetcher,

                    XLogRecPtr RecPtr, TimeLineID replayTLI);

static bool rescanLatestTimeLine(TimeLineID replayTLI, XLogRecPtr replayLSN);

static int  XLogFileRead(XLogSegNo segno, TimeLineID tli,

             XLogSource source, bool notfoundOk);

static int  XLogFileReadAnyTLI(XLogSegNo segno, XLogSource source);

static bool CheckForStandbyTrigger(void);

static void SetPromoteIsTriggered(void);

static bool HotStandbyActiveInReplay(void);

static void SetCurrentChunkStartTime(TimestampTz xtime);

static void SetLatestXTime(TimestampTz xtime);

/*

 * Initialization of shared memory for WAL recovery

 */

Size

XLogRecoveryShmemSize(void)

{

  Size    size;

  /* XLogRecoveryCtl */

  size = sizeof(XLogRecoveryCtlData);

  return size;

}

void

XLogRecoveryShmemInit(void)

{

  bool    found;

  XLogRecoveryCtl = (XLogRecoveryCtlData *)

    ShmemInitStruct("XLOG Recovery Ctl", XLogRecoveryShmemSize(), &found);

  if (found)

    return;

  memset(XLogRecoveryCtl, 0, sizeof(XLogRecoveryCtlData));

  SpinLockInit(&XLogRecoveryCtl->info_lck);

  InitSharedLatch(&XLogRecoveryCtl->recoveryWakeupLatch);

  ConditionVariableInit(&XLogRecoveryCtl->recoveryNotPausedCV);

}

/*

 * A thin wrapper to enable StandbyMode and do other preparatory work as

 * needed.

 */

static void

EnableStandbyMode(void)

{

  StandbyMode = true;

  /*

   * To avoid server log bloat, we don't report recovery progress in a

   * standby as it will always be in recovery unless promoted. We disable

   * startup progress timeout in standby mode to avoid calling

   * startup_progress_timeout_handler() unnecessarily.

   */

  disable_startup_progress_timeout();

}

/*

 * Prepare the system for WAL recovery, if needed.

 *

 * This is called by StartupXLOG() which coordinates the server startup

 * sequence.  This function analyzes the control file and the backup label

 * file, if any, and figures out whether we need to perform crash recovery or

 * archive recovery, and how far we need to replay the WAL to reach a

 * consistent state.

 *

 * This doesn't yet change the on-disk state, except for creating the symlinks

 * from table space map file if any, and for fetching WAL files needed to find

 * the checkpoint record.  On entry, the caller has already read the control

 * file into memory, and passes it as argument.  This function updates it to

 * reflect the recovery state, and the caller is expected to write it back to

 * disk does after initializing other subsystems, but before calling

 * PerformWalRecovery().

 *

 * This initializes some global variables like ArchiveRecoveryRequested, and

 * StandbyModeRequested and InRecovery.

 */

void

InitWalRecovery(ControlFileData *ControlFile, bool *wasShutdown_ptr,

        bool *haveBackupLabel_ptr, bool *haveTblspcMap_ptr)

{

  XLogPageReadPrivate *private;

  struct stat st;

  bool    wasShutdown;

  XLogRecord *record;

  DBState   dbstate_at_startup;

  bool    haveTblspcMap = false;

  bool    haveBackupLabel = false;

  CheckPoint  checkPoint;

  bool    backupFromStandby = false;

  dbstate_at_startup = ControlFile->state;

  /*

   * Initialize on the assumption we want to recover to the latest timeline

   * that's active according to pg_control.

   */

  if (ControlFile->minRecoveryPointTLI >

    ControlFile->checkPointCopy.ThisTimeLineID)

    recoveryTargetTLI = ControlFile->minRecoveryPointTLI;

  else

    recoveryTargetTLI = ControlFile->checkPointCopy.ThisTimeLineID;

  /*

   * Check for signal files, and if so set up state for offline recovery

   */

  readRecoverySignalFile();

  validateRecoveryParameters();

  /*

   * Take ownership of the wakeup latch if we're going to sleep during

   * recovery, if required.

   */

  if (ArchiveRecoveryRequested)

    OwnLatch(&XLogRecoveryCtl->recoveryWakeupLatch);

  /*

   * Set the WAL reading processor now, as it will be needed when reading

   * the checkpoint record required (backup_label or not).

   */

  private = palloc0(sizeof(XLogPageReadPrivate));

  xlogreader =

    XLogReaderAllocate(wal_segment_size, NULL,

               XL_ROUTINE(.page_read = &XLogPageRead,

                    .segment_open = NULL,

                    .segment_close = wal_segment_close),

               private);

  if (!xlogreader)

    ereport(ERROR,

        (errcode(ERRCODE_OUT_OF_MEMORY),

         errmsg("out of memory"),

         errdetail("Failed while allocating a WAL reading processor.")));

  xlogreader->system_identifier = ControlFile->system_identifier;

  /*

   * Set the WAL decode buffer size.  This limits how far ahead we can read

   * in the WAL.

   */

  XLogReaderSetDecodeBuffer(xlogreader, NULL, wal_decode_buffer_size);

  /* Create a WAL prefetcher. */

  xlogprefetcher = XLogPrefetcherAllocate(xlogreader);

  /*

   * Allocate two page buffers dedicated to WAL consistency checks.  We do

   * it this way, rather than just making static arrays, for two reasons:

   * (1) no need to waste the storage in most instantiations of the backend;

   * (2) a static char array isn't guaranteed to have any particular

   * alignment, whereas palloc() will provide MAXALIGN'd storage.

   */

  replay_image_masked = (char *) palloc(BLCKSZ);

  primary_image_masked = (char *) palloc(BLCKSZ);

  /*

   * Read the backup_label file.  We want to run this part of the recovery

   * process after checking for signal files and after performing validation

   * of the recovery parameters.

   */

  if (read_backup_label(&CheckPointLoc, &CheckPointTLI, &backupEndRequired,

              &backupFromStandby))

  {

    List     *tablespaces = NIL;

    /*

     * Archive recovery was requested, and thanks to the backup label

     * file, we know how far we need to replay to reach consistency. Enter

     * archive recovery directly.

     */

    InArchiveRecovery = true;

    if (StandbyModeRequested)

      EnableStandbyMode();

    /*

     * Omitting backup_label when creating a new replica, PITR node etc.

     * unfortunately is a common cause of corruption.  Logging that

     * backup_label was used makes it a bit easier to exclude that as the

     * cause of observed corruption.

     *

     * Do so before we try to read the checkpoint record (which can fail),

     * as otherwise it can be hard to understand why a checkpoint other

     * than ControlFile->checkPoint is used.

     */

    ereport(LOG,

        errmsg("starting backup recovery with redo LSN %X/%08X, checkpoint LSN %X/%08X, on timeline ID %u",

             LSN_FORMAT_ARGS(RedoStartLSN),

             LSN_FORMAT_ARGS(CheckPointLoc),

             CheckPointTLI));

    /*

     * When a backup_label file is present, we want to roll forward from

     * the checkpoint it identifies, rather than using pg_control.

     */

    record = ReadCheckpointRecord(xlogprefetcher, CheckPointLoc,

                    CheckPointTLI);

    if (record != NULL)

    {

      memcpy(&checkPoint, XLogRecGetData(xlogreader), sizeof(CheckPoint));

      wasShutdown = ((record->xl_info & ~XLR_INFO_MASK) == XLOG_CHECKPOINT_SHUTDOWN);

      ereport(DEBUG1,

          errmsg_internal("checkpoint record is at %X/%08X",

                  LSN_FORMAT_ARGS(CheckPointLoc)));

      InRecovery = true;  /* force recovery even if SHUTDOWNED */

      /*

       * Make sure that REDO location exists. This may not be the case

       * if there was a crash during an online backup, which left a

       * backup_label around that references a WAL segment that's

       * already been archived.

       */

      if (checkPoint.redo < CheckPointLoc)

      {

        XLogPrefetcherBeginRead(xlogprefetcher, checkPoint.redo);

        if (!ReadRecord(xlogprefetcher, LOG, false,

                checkPoint.ThisTimeLineID))

          ereport(FATAL,

              errmsg("could not find redo location %X/%08X referenced by checkpoint record at %X/%08X",

                   LSN_FORMAT_ARGS(checkPoint.redo), LSN_FORMAT_ARGS(CheckPointLoc)),

              errhint("If you are restoring from a backup, touch \"%s/recovery.signal\" or \"%s/standby.signal\" and add required recovery options.\n"

                  "If you are not restoring from a backup, try removing the file \"%s/backup_label\".\n"

                  "Be careful: removing \"%s/backup_label\" will result in a corrupt cluster if restoring from a backup.",

                  DataDir, DataDir, DataDir, DataDir));

      }

    }

    else

    {

      ereport(FATAL,

          errmsg("could not locate required checkpoint record at %X/%08X",

               LSN_FORMAT_ARGS(CheckPointLoc)),

          errhint("If you are restoring from a backup, touch \"%s/recovery.signal\" or \"%s/standby.signal\" and add required recovery options.\n"

              "If you are not restoring from a backup, try removing the file \"%s/backup_label\".\n"

              "Be careful: removing \"%s/backup_label\" will result in a corrupt cluster if restoring from a backup.",

              DataDir, DataDir, DataDir, DataDir));

      wasShutdown = false;  /* keep compiler quiet */

    }

    /* Read the tablespace_map file if present and create symlinks. */

    if (read_tablespace_map(&tablespaces))

    {

      ListCell   *lc;

      foreach(lc, tablespaces)

      {

        tablespaceinfo *ti = lfirst(lc);

        char     *linkloc;

        linkloc = psprintf("%s/%u", PG_TBLSPC_DIR, ti->oid);

        /*

         * Remove the existing symlink if any and Create the symlink

         * under PGDATA.

         */

        remove_tablespace_symlink(linkloc);

        if (symlink(ti->path, linkloc) < 0)

          ereport(ERROR,

              (errcode_for_file_access(),

               errmsg("could not create symbolic link \"%s\": %m",

                  linkloc)));

        pfree(ti->path);

        pfree(ti);

      }

      /* tell the caller to delete it later */

      haveTblspcMap = true;

    }

    /* tell the caller to delete it later */

    haveBackupLabel = true;

  }

  else

  {

    /* No backup_label file has been found if we are here. */

    /*

     * If tablespace_map file is present without backup_label file, there

     * is no use of such file.  There is no harm in retaining it, but it

     * is better to get rid of the map file so that we don't have any

     * redundant file in data directory and it will avoid any sort of

     * confusion.  It seems prudent though to just rename the file out of

     * the way rather than delete it completely, also we ignore any error

     * that occurs in rename operation as even if map file is present

     * without backup_label file, it is harmless.

     */

    if (stat(TABLESPACE_MAP, &st) == 0)

    {

      unlink(TABLESPACE_MAP_OLD);

      if (durable_rename(TABLESPACE_MAP, TABLESPACE_MAP_OLD, DEBUG1) == 0)

        ereport(LOG,

            (errmsg("ignoring file \"%s\" because no file \"%s\" exists",

                TABLESPACE_MAP, BACKUP_LABEL_FILE),

             errdetail("File \"%s\" was renamed to \"%s\".",

                   TABLESPACE_MAP, TABLESPACE_MAP_OLD)));

      else

        ereport(LOG,

            (errmsg("ignoring file \"%s\" because no file \"%s\" exists",

                TABLESPACE_MAP, BACKUP_LABEL_FILE),

             errdetail("Could not rename file \"%s\" to \"%s\": %m.",

                   TABLESPACE_MAP, TABLESPACE_MAP_OLD)));

    }

    /*

     * It's possible that archive recovery was requested, but we don't

     * know how far we need to replay the WAL before we reach consistency.

     * This can happen for example if a base backup is taken from a

     * running server using an atomic filesystem snapshot, without calling

     * pg_backup_start/stop. Or if you just kill a running primary server

     * and put it into archive recovery by creating a recovery signal

     * file.

     *

     * Our strategy in that case is to perform crash recovery first,

     * replaying all the WAL present in pg_wal, and only enter archive

     * recovery after that.

     *

     * But usually we already know how far we need to replay the WAL (up

     * to minRecoveryPoint, up to backupEndPoint, or until we see an

     * end-of-backup record), and we can enter archive recovery directly.

     */

    if (ArchiveRecoveryRequested &&

      (ControlFile->minRecoveryPoint != InvalidXLogRecPtr ||

       ControlFile->backupEndRequired ||

       ControlFile->backupEndPoint != InvalidXLogRecPtr ||

       ControlFile->state == DB_SHUTDOWNED))

    {

      InArchiveRecovery = true;

      if (StandbyModeRequested)

        EnableStandbyMode();

    }

    /*

     * For the same reason as when starting up with backup_label present,

     * emit a log message when we continue initializing from a base

     * backup.

     */

    if (!XLogRecPtrIsInvalid(ControlFile->backupStartPoint))

      ereport(LOG,

          errmsg("restarting backup recovery with redo LSN %X/%08X",

               LSN_FORMAT_ARGS(ControlFile->backupStartPoint)));

    /* Get the last valid checkpoint record. */

    CheckPointLoc = ControlFile->checkPoint;

    CheckPointTLI = ControlFile->checkPointCopy.ThisTimeLineID;

    RedoStartLSN = ControlFile->checkPointCopy.redo;

    RedoStartTLI = ControlFile->checkPointCopy.ThisTimeLineID;

    record = ReadCheckpointRecord(xlogprefetcher, CheckPointLoc,

                    CheckPointTLI);

    if (record != NULL)

    {

      ereport(DEBUG1,

          errmsg_internal("checkpoint record is at %X/%08X",

                  LSN_FORMAT_ARGS(CheckPointLoc)));

    }

    else

    {

      /*

       * We used to attempt to go back to a secondary checkpoint record

       * here, but only when not in standby mode. We now just fail if we

       * can't read the last checkpoint because this allows us to

       * simplify processing around checkpoints.

       */

      ereport(PANIC,

          errmsg("could not locate a valid checkpoint record at %X/%08X",

               LSN_FORMAT_ARGS(CheckPointLoc)));

    }

    memcpy(&checkPoint, XLogRecGetData(xlogreader), sizeof(CheckPoint));

    wasShutdown = ((record->xl_info & ~XLR_INFO_MASK) == XLOG_CHECKPOINT_SHUTDOWN);

  }

  if (ArchiveRecoveryRequested)

  {

    if (StandbyModeRequested)

      ereport(LOG,

          (errmsg("entering standby mode")));

    else if (recoveryTarget == RECOVERY_TARGET_XID)

      ereport(LOG,

          (errmsg("starting point-in-time recovery to XID %u",

              recoveryTargetXid)));

    else if (recoveryTarget == RECOVERY_TARGET_TIME)

      ereport(LOG,

          (errmsg("starting point-in-time recovery to %s",

              timestamptz_to_str(recoveryTargetTime))));

    else if (recoveryTarget == RECOVERY_TARGET_NAME)

      ereport(LOG,

          (errmsg("starting point-in-time recovery to \"%s\"",

              recoveryTargetName)));

    else if (recoveryTarget == RECOVERY_TARGET_LSN)

      ereport(LOG,

          errmsg("starting point-in-time recovery to WAL location (LSN) \"%X/%08X\"",

               LSN_FORMAT_ARGS(recoveryTargetLSN)));

    else if (recoveryTarget == RECOVERY_TARGET_IMMEDIATE)

      ereport(LOG,

          (errmsg("starting point-in-time recovery to earliest consistent point")));

    else

      ereport(LOG,

          (errmsg("starting archive recovery")));

  }

  /*

   * If the location of the checkpoint record is not on the expected

   * timeline in the history of the requested timeline, we cannot proceed:

   * the backup is not part of the history of the requested timeline.

   */

  Assert(expectedTLEs);   /* was initialized by reading checkpoint

                 * record */

  if (tliOfPointInHistory(CheckPointLoc, expectedTLEs) !=

    CheckPointTLI)

  {

    XLogRecPtr  switchpoint;

    /*

     * tliSwitchPoint will throw an error if the checkpoint's timeline is

     * not in expectedTLEs at all.

     */

    switchpoint = tliSwitchPoint(CheckPointTLI, expectedTLEs, NULL);

    ereport(FATAL,

        (errmsg("requested timeline %u is not a child of this server's history",

            recoveryTargetTLI),

    /* translator: %s is a backup_label file or a pg_control file */

         errdetail("Latest checkpoint in file \"%s\" is at %X/%08X on timeline %u, but in the history of the requested timeline, the server forked off from that timeline at %X/%08X.",

               haveBackupLabel ? "backup_label" : "pg_control",

               LSN_FORMAT_ARGS(CheckPointLoc),

               CheckPointTLI,

               LSN_FORMAT_ARGS(switchpoint))));

  }

  /*

   * The min recovery point should be part of the requested timeline's

   * history, too.

   */

  if (!XLogRecPtrIsInvalid(ControlFile->minRecoveryPoint) &&

    tliOfPointInHistory(ControlFile->minRecoveryPoint - 1, expectedTLEs) !=

    ControlFile->minRecoveryPointTLI)

    ereport(FATAL,

        errmsg("requested timeline %u does not contain minimum recovery point %X/%08X on timeline %u",

             recoveryTargetTLI,

             LSN_FORMAT_ARGS(ControlFile->minRecoveryPoint),

             ControlFile->minRecoveryPointTLI));

  ereport(DEBUG1,

      errmsg_internal("redo record is at %X/%08X; shutdown %s",

              LSN_FORMAT_ARGS(checkPoint.redo),

              wasShutdown ? "true" : "false"));

  ereport(DEBUG1,

      (errmsg_internal("next transaction ID: " UINT64_FORMAT "; next OID: %u",

               U64FromFullTransactionId(checkPoint.nextXid),

               checkPoint.nextOid)));

  ereport(DEBUG1,

      (errmsg_internal("next MultiXactId: %u; next MultiXactOffset: %u",

               checkPoint.nextMulti, checkPoint.nextMultiOffset)));

  ereport(DEBUG1,

      (errmsg_internal("oldest unfrozen transaction ID: %u, in database %u",

               checkPoint.oldestXid, checkPoint.oldestXidDB)));

  ereport(DEBUG1,

      (errmsg_internal("oldest MultiXactId: %u, in database %u",

               checkPoint.oldestMulti, checkPoint.oldestMultiDB)));

  ereport(DEBUG1,

      (errmsg_internal("commit timestamp Xid oldest/newest: %u/%u",

               checkPoint.oldestCommitTsXid,

               checkPoint.newestCommitTsXid)));

  if (!TransactionIdIsNormal(XidFromFullTransactionId(checkPoint.nextXid)))

    ereport(PANIC,

        (errmsg("invalid next transaction ID")));

  /* sanity check */

  if (checkPoint.redo > CheckPointLoc)

    ereport(PANIC,

        (errmsg("invalid redo in checkpoint record")));

  /*

   * Check whether we need to force recovery from WAL.  If it appears to

   * have been a clean shutdown and we did not have a recovery signal file,

   * then assume no recovery needed.

   */

  if (checkPoint.redo < CheckPointLoc)

  {

    if (wasShutdown)

      ereport(PANIC,

          (errmsg("invalid redo record in shutdown checkpoint")));

    InRecovery = true;

  }

  else if (ControlFile->state != DB_SHUTDOWNED)

    InRecovery = true;

  else if (ArchiveRecoveryRequested)

  {

    /* force recovery due to presence of recovery signal file */

    InRecovery = true;

  }

  /*

   * If recovery is needed, update our in-memory copy of pg_control to show

   * that we are recovering and to show the selected checkpoint as the place

   * we are starting from. We also mark pg_control with any minimum recovery

   * stop point obtained from a backup history file.

   *

   * We don't write the changes to disk yet, though. Only do that after

   * initializing various subsystems.

   */

  if (InRecovery)

  {

    if (InArchiveRecovery)

    {

      ControlFile->state = DB_IN_ARCHIVE_RECOVERY;

    }

    else

    {

      ereport(LOG,

          (errmsg("database system was not properly shut down; "

              "automatic recovery in progress")));

      if (recoveryTargetTLI > ControlFile->checkPointCopy.ThisTimeLineID)

        ereport(LOG,

            (errmsg("crash recovery starts in timeline %u "

                "and has target timeline %u",

                ControlFile->checkPointCopy.ThisTimeLineID,

                recoveryTargetTLI)));

      ControlFile->state = DB_IN_CRASH_RECOVERY;

    }

    ControlFile->checkPoint = CheckPointLoc;

    ControlFile->checkPointCopy = checkPoint;

    if (InArchiveRecovery)

    {

      /* initialize minRecoveryPoint if not set yet */

      if (ControlFile->minRecoveryPoint < checkPoint.redo)

      {

        ControlFile->minRecoveryPoint = checkPoint.redo;

        ControlFile->minRecoveryPointTLI = checkPoint.ThisTimeLineID;

      }

    }

    /*

     * Set backupStartPoint if we're starting recovery from a base backup.

     *

     * Also set backupEndPoint and use minRecoveryPoint as the backup end

     * location if we're starting recovery from a base backup which was

     * taken from a standby. In this case, the database system status in

     * pg_control must indicate that the database was already in recovery.

     * Usually that will be DB_IN_ARCHIVE_RECOVERY but also can be

     * DB_SHUTDOWNED_IN_RECOVERY if recovery previously was interrupted

     * before reaching this point; e.g. because restore_command or

     * primary_conninfo were faulty.

     *

     * Any other state indicates that the backup somehow became corrupted

     * and we can't sensibly continue with recovery.

     */

    if (haveBackupLabel)

    {

      ControlFile->backupStartPoint = checkPoint.redo;

      ControlFile->backupEndRequired = backupEndRequired;

      if (backupFromStandby)

      {

        if (dbstate_at_startup != DB_IN_ARCHIVE_RECOVERY &&

          dbstate_at_startup != DB_SHUTDOWNED_IN_RECOVERY)

          ereport(FATAL,

              (errmsg("backup_label contains data inconsistent with control file"),

               errhint("This means that the backup is corrupted and you will "

                   "have to use another backup for recovery.")));

        ControlFile->backupEndPoint = ControlFile->minRecoveryPoint;

      }

    }

  }

  /* remember these, so that we know when we have reached consistency */

  backupStartPoint = ControlFile->backupStartPoint;

  backupEndRequired = ControlFile->backupEndRequired;

  backupEndPoint = ControlFile->backupEndPoint;

  if (InArchiveRecovery)

  {

    minRecoveryPoint = ControlFile->minRecoveryPoint;

    minRecoveryPointTLI = ControlFile->minRecoveryPointTLI;