/*-------------------------------------------------------------------------
 *
 * walreceiver.c
 *
 * The WAL receiver process (walreceiver) is new as of Postgres 9.0. It
 * is the process in the standby server that takes charge of receiving
 * XLOG records from a primary server during streaming replication.
 *
 * When the startup process determines that it's time to start streaming,
 * it instructs postmaster to start walreceiver. Walreceiver first connects
 * to the primary server (it will be served by a walsender process
 * in the primary server), and then keeps receiving XLOG records and
 * writing them to the disk as long as the connection is alive. As XLOG
 * records are received and flushed to disk, it updates the
 * WalRcv->flushedUpto variable in shared memory, to inform the startup
 * process of how far it can proceed with XLOG replay.
 *
 * A WAL receiver cannot directly load GUC parameters used when establishing
 * its connection to the primary. Instead it relies on parameter values
 * that are passed down by the startup process when streaming is requested.
 * This applies, for example, to the replication slot and the connection
 * string to be used for the connection with the primary.
 *
 * If the primary server ends streaming, but doesn't disconnect, walreceiver
 * goes into "waiting" mode, and waits for the startup process to give new
 * instructions. The startup process will treat that the same as
 * disconnection, and will rescan the archive/pg_wal directory. But when the
 * startup process wants to try streaming replication again, it will just
 * nudge the existing walreceiver process that's waiting, instead of launching
 * a new one.
 *
 * Normal termination is by SIGTERM, which instructs the walreceiver to
 * exit(0). Emergency termination is by SIGQUIT; like any postmaster child
 * process, the walreceiver will simply abort and exit on SIGQUIT. A close
 * of the connection and a FATAL error are treated not as a crash but as
 * normal operation.
 *
 * This file contains the server-facing parts of walreceiver. The libpq-
 * specific parts are in the libpqwalreceiver module. It's loaded
 * dynamically to avoid linking the server with libpq.
 *
 * Portions Copyright (c) 2010-2025, PostgreSQL Global Development Group
 *
 *
 * IDENTIFICATION
 *   src/backend/replication/walreceiver.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <unistd.h>

#include "access/htup_details.h"
#include "access/timeline.h"
#include "access/transam.h"
#include "access/xlog_internal.h"
#include "access/xlogarchive.h"
#include "access/xlogrecovery.h"
#include "catalog/pg_authid.h"
#include "funcapi.h"
#include "libpq/pqformat.h"
#include "libpq/pqsignal.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/auxprocess.h"
#include "postmaster/interrupt.h"
#include "replication/walreceiver.h"
#include "replication/walsender.h"
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/procsignal.h"
#include "tcop/tcopprot.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/guc.h"
#include "utils/pg_lsn.h"
#include "utils/ps_status.h"
#include "utils/timestamp.h"


/*
 * GUC variables.  (Other variables that affect walreceiver are in xlog.c
 * because they're passed down from the startup process, for better
 * synchronization.)
 */
int   wal_receiver_status_interval;
int   wal_receiver_timeout;
bool  hot_standby_feedback;

/* libpqwalreceiver connection */
static WalReceiverConn *wrconn = NULL;
WalReceiverFunctionsType *WalReceiverFunctions = NULL;

/*
 * These variables are used similarly to openLogFile/SegNo,
 * but for walreceiver to write the XLOG. recvFileTLI is the TimeLineID
 * corresponding the filename of recvFile.
 */
static int recvFile = -1;
static TimeLineID recvFileTLI = 0;
static XLogSegNo recvSegNo = 0;

/*
 * LogstreamResult indicates the byte positions that we have already
 * written/fsynced.
 */
static struct
{
 XLogRecPtr Write;   /* last byte + 1 written out in the standby */
 XLogRecPtr Flush;   /* last byte + 1 flushed in the standby */
}   LogstreamResult;

/*
 * Reasons to wake up and perform periodic tasks.
 */
typedef enum WalRcvWakeupReason
{
 WALRCV_WAKEUP_TERMINATE,
 WALRCV_WAKEUP_PING,
 WALRCV_WAKEUP_REPLY,
 WALRCV_WAKEUP_HSFEEDBACK,
#define NUM_WALRCV_WAKEUPS (WALRCV_WAKEUP_HSFEEDBACK + 1)
} WalRcvWakeupReason;

/*
 * Wake up times for periodic tasks.
 */
static TimestampTz wakeup[NUM_WALRCV_WAKEUPS];

static StringInfoData reply_message;

/* Prototypes for private functions */
static void WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last);
static void WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI);
static void WalRcvDie(int code, Datum arg);
static void XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len,
         TimeLineID tli);
static void XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr,
       TimeLineID tli);
static void XLogWalRcvFlush(bool dying, TimeLineID tli);
static void XLogWalRcvClose(XLogRecPtr recptr, TimeLineID tli);
static void XLogWalRcvSendReply(bool force, bool requestReply);
static void XLogWalRcvSendHSFeedback(bool immed);
static void ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime);
static void WalRcvComputeNextWakeup(WalRcvWakeupReason reason, TimestampTz now);


/* Main entry point for walreceiver process */
void
WalReceiverMain(const void *startup_data, size_t startup_data_len)
{
 char  conninfo[MAXCONNINFO];
 char    *tmp_conninfo;
 char  slotname[NAMEDATALEN];
 bool  is_temp_slot;
 XLogRecPtr startpoint;
 TimeLineID startpointTLI;
 TimeLineID primaryTLI;
 bool  first_stream;
 WalRcvData *walrcv;
 TimestampTz now;
 char    *err;
 char    *sender_host = NULL;
 int   sender_port = 0;
 char    *appname;

 Assert(startup_data_len == 0);

 MyBackendType = B_WAL_RECEIVER;
 AuxiliaryProcessMainCommon();

 /*
  * WalRcv should be set up already (if we are a backend, we inherit this
  * by fork() or EXEC_BACKEND mechanism from the postmaster).
 */
 walrcv = WalRcv;
 Assert(walrcv != NULL);

 /*
  * Mark walreceiver as running in shared memory.
  *
  * Do this as early as possible, so that if we fail later on, we'll set
  * state to STOPPED. If we die before this, the startup process will keep
  * waiting for us to start up, until it times out.
 */
 SpinLockAcquire(&walrcv->mutex);
 Assert(walrcv->pid == 0);
 switch (walrcv->walRcvState)
 {
  case WALRCV_STOPPING:
   /* If we've already been requested to stop, don't start up. */
   walrcv->walRcvState = WALRCV_STOPPED;
   /* fall through */

  case WALRCV_STOPPED:
   SpinLockRelease(&walrcv->mutex);
   ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);
   proc_exit(1);
   break;

  case WALRCV_STARTING:
   /* The usual case */
   break;

  case WALRCV_WAITING:
  case WALRCV_STREAMING:
  case WALRCV_RESTARTING:
  default:
   /* Shouldn't happen */
   SpinLockRelease(&walrcv->mutex);
   elog(PANIC, "walreceiver still running according to shared memory state");
 }
 /* Advertise our PID so that the startup process can kill us */
 walrcv->pid = MyProcPid;
 walrcv->walRcvState = WALRCV_STREAMING;

 /* Fetch information required to start streaming */
 walrcv->ready_to_display = false;
 strlcpy(conninfo, walrcv->conninfo, MAXCONNINFO);
 strlcpy(slotname, walrcv->slotname, NAMEDATALEN);
 is_temp_slot = walrcv->is_temp_slot;
 startpoint = walrcv->receiveStart;
 startpointTLI = walrcv->receiveStartTLI;

 /*
  * At most one of is_temp_slot and slotname can be set; otherwise,
  * RequestXLogStreaming messed up.
 */
 Assert(!is_temp_slot || (slotname[0] == '\0'));

 /* Initialise to a sanish value */
 now = GetCurrentTimestamp();
 walrcv->lastMsgSendTime =
  walrcv->lastMsgReceiptTime = walrcv->latestWalEndTime = now;

 /* Report our proc number so that others can wake us up */
 walrcv->procno = MyProcNumber;

 SpinLockRelease(&walrcv->mutex);

 pg_atomic_write_u64(&WalRcv->writtenUpto, 0);

 /* Arrange to clean up at walreceiver exit */
 on_shmem_exit(WalRcvDie, PointerGetDatum(&startpointTLI));

 /* Properly accept or ignore signals the postmaster might send us */
 pqsignal(SIGHUP, SignalHandlerForConfigReload); /* set flag to read config
 * file */
 pqsignal(SIGINT, SIG_IGN);
 pqsignal(SIGTERM, die);  /* request shutdown */
 /* SIGQUIT handler was already set up by InitPostmasterChild */
 pqsignal(SIGALRM, SIG_IGN);
 pqsignal(SIGPIPE, SIG_IGN);
 pqsignal(SIGUSR1, procsignal_sigusr1_handler);
 pqsignal(SIGUSR2, SIG_IGN);

 /* Reset some signals that are accepted by postmaster but not here */
 pqsignal(SIGCHLD, SIG_DFL);

 /* Load the libpq-specific functions */
 load_file("libpqwalreceiver", false);
 if (WalReceiverFunctions == NULL)
  elog(ERROR, "libpqwalreceiver didn't initialize correctly");

 /* Unblock signals (they were blocked when the postmaster forked us) */
 sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);

 /* Establish the connection to the primary for XLOG streaming */
 appname = cluster_name[0] ? cluster_name : "walreceiver";
 wrconn = walrcv_connect(conninfo, true, false, false, appname, &err);
 if (!wrconn)
  ereport(ERROR,
    (errcode(ERRCODE_CONNECTION_FAILURE),
     errmsg("streaming replication receiver \"%s\" could not connect to the primary server: %s",
      appname, err)));

 /*
  * Save user-visible connection string.  This clobbers the original
  * conninfo, for security. Also save host and port of the sender server
  * this walreceiver is connected to.
 */
 tmp_conninfo = walrcv_get_conninfo(wrconn);
 walrcv_get_senderinfo(wrconn, &sender_host, &sender_port);
 SpinLockAcquire(&walrcv->mutex);
 memset(walrcv->conninfo, 0, MAXCONNINFO);
 if (tmp_conninfo)
  strlcpy(walrcv->conninfo, tmp_conninfo, MAXCONNINFO);

 memset(walrcv->sender_host, 0, NI_MAXHOST);
 if (sender_host)
  strlcpy(walrcv->sender_host, sender_host, NI_MAXHOST);

 walrcv->sender_port = sender_port;
 walrcv->ready_to_display = true;
 SpinLockRelease(&walrcv->mutex);

 if (tmp_conninfo)
  pfree(tmp_conninfo);

 if (sender_host)
  pfree(sender_host);

 first_stream = true;
 for (;;)
 {
  char    *primary_sysid;
  char  standby_sysid[32];
  WalRcvStreamOptions options;

  /*
   * Check that we're connected to a valid server using the
   * IDENTIFY_SYSTEM replication command.
 */
  primary_sysid = walrcv_identify_system(wrconn, &primaryTLI);

  snprintf(standby_sysid, sizeof(standby_sysid), UINT64_FORMAT,
     GetSystemIdentifier());
  if (strcmp(primary_sysid, standby_sysid) != 0)
  {
   ereport(ERROR,
     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
      errmsg("database system identifier differs between the primary and standby"),
      errdetail("The primary's identifier is %s, the standby's identifier is %s.",
          primary_sysid, standby_sysid)));
  }

  /*
   * Confirm that the current timeline of the primary is the same or
   * ahead of ours.
 */
  if (primaryTLI < startpointTLI)
   ereport(ERROR,
     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
      errmsg("highest timeline %u of the primary is behind recovery timeline %u",
       primaryTLI, startpointTLI)));

  /*
   * Get any missing history files. We do this always, even when we're
   * not interested in that timeline, so that if we're promoted to
   * become the primary later on, we don't select the same timeline that
   * was already used in the current primary. This isn't bullet-proof -
   * you'll need some external software to manage your cluster if you
   * need to ensure that a unique timeline id is chosen in every case,
   * but let's avoid the confusion of timeline id collisions where we
   * can.
 */
  WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);

  /*
   * Create temporary replication slot if requested, and update slot
   * name in shared memory.  (Note the slot name cannot already be set
   * in this case.)
 */
  if (is_temp_slot)
  {
   snprintf(slotname, sizeof(slotname),
      "pg_walreceiver_%lld",
      (long long int) walrcv_get_backend_pid(wrconn));

   walrcv_create_slot(wrconn, slotname, true, false, false, 0, NULL);

   SpinLockAcquire(&walrcv->mutex);
   strlcpy(walrcv->slotname, slotname, NAMEDATALEN);
   SpinLockRelease(&walrcv->mutex);
  }

  /*
   * Start streaming.
   *
   * We'll try to start at the requested starting point and timeline,
   * even if it's different from the server's latest timeline. In case
   * we've already reached the end of the old timeline, the server will
   * finish the streaming immediately, and we will go back to await
   * orders from the startup process. If recovery_target_timeline is
   * 'latest', the startup process will scan pg_wal and find the new
   * history file, bump recovery target timeline, and ask us to restart
   * on the new timeline.
 */
  options.logical = false;
  options.startpoint = startpoint;
  options.slotname = slotname[0] != '\0' ? slotname : NULL;
  options.proto.physical.startpointTLI = startpointTLI;
  if (walrcv_startstreaming(wrconn, &options))
  {
   if (first_stream)
    ereport(LOG,
      (errmsg("started streaming WAL from primary at %X/%X on timeline %u",
        LSN_FORMAT_ARGS(startpoint), startpointTLI)));
   else
    ereport(LOG,
      (errmsg("restarted WAL streaming at %X/%X on timeline %u",
        LSN_FORMAT_ARGS(startpoint), startpointTLI)));
   first_stream = false;

   /* Initialize LogstreamResult and buffers for processing messages */
   LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
   initStringInfo(&reply_message);

   /* Initialize nap wakeup times. */
   now = GetCurrentTimestamp();
   for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
    WalRcvComputeNextWakeup(i, now);

   /* Send initial reply/feedback messages. */
   XLogWalRcvSendReply(true, false);
   XLogWalRcvSendHSFeedback(true);

   /* Loop until end-of-streaming or error */
   for (;;)
   {
    char    *buf;
    int   len;
    bool  endofwal = false;
    pgsocket wait_fd = PGINVALID_SOCKET;
    int   rc;
    TimestampTz nextWakeup;
    long  nap;

    /*
     * Exit walreceiver if we're not in recovery. This should not
     * happen, but cross-check the status here.
 */
    if (!RecoveryInProgress())
     ereport(FATAL,
       (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
        errmsg("cannot continue WAL streaming, recovery has already ended")));

    /* Process any requests or signals received recently */
    CHECK_FOR_INTERRUPTS();

    if (ConfigReloadPending)
    {
     ConfigReloadPending = false;
     ProcessConfigFile(PGC_SIGHUP);
     /* recompute wakeup times */
     now = GetCurrentTimestamp();
     for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
      WalRcvComputeNextWakeup(i, now);
     XLogWalRcvSendHSFeedback(true);
    }

    /* See if we can read data immediately */
    len = walrcv_receive(wrconn, &buf, &wait_fd);
    if (len != 0)
    {
     /*
      * Process the received data, and any subsequent data we
      * can read without blocking.
 */
     for (;;)
     {
      if (len > 0)
      {
       /*
        * Something was received from primary, so adjust
        * the ping and terminate wakeup times.
 */
       now = GetCurrentTimestamp();
       WalRcvComputeNextWakeup(WALRCV_WAKEUP_TERMINATE,
             now);
       WalRcvComputeNextWakeup(WALRCV_WAKEUP_PING, now);
       XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1,
             startpointTLI);
      }
      else if (len == 0)
       break;
      else if (len < 0)
      {
       ereport(LOG,
         (errmsg("replication terminated by primary server"),
          errdetail("End of WAL reached on timeline %u at %X/%X.",
              startpointTLI,
              LSN_FORMAT_ARGS(LogstreamResult.Write))));
       endofwal = true;
       break;
      }
      len = walrcv_receive(wrconn, &buf, &wait_fd);
     }

     /* Let the primary know that we received some data. */
     XLogWalRcvSendReply(false, false);

     /*
      * If we've written some records, flush them to disk and
      * let the startup process and primary server know about
      * them.
 */
     XLogWalRcvFlush(false, startpointTLI);
    }

    /* Check if we need to exit the streaming loop. */
    if (endofwal)
     break;

    /* Find the soonest wakeup time, to limit our nap. */
    nextWakeup = TIMESTAMP_INFINITY;
    for (int i = 0; i < NUM_WALRCV_WAKEUPS; ++i)
     nextWakeup = Min(wakeup[i], nextWakeup);

    /* Calculate the nap time, clamping as necessary. */
    now = GetCurrentTimestamp();
    nap = TimestampDifferenceMilliseconds(now, nextWakeup);

    /*
     * Ideally we would reuse a WaitEventSet object repeatedly
     * here to avoid the overheads of WaitLatchOrSocket on epoll
     * systems, but we can't be sure that libpq (or any other
     * walreceiver implementation) has the same socket (even if
     * the fd is the same number, it may have been closed and
     * reopened since the last time).  In future, if there is a
     * function for removing sockets from WaitEventSet, then we
     * could add and remove just the socket each time, potentially
     * avoiding some system calls.
 */
    Assert(wait_fd != PGINVALID_SOCKET);
    rc = WaitLatchOrSocket(MyLatch,
            WL_EXIT_ON_PM_DEATH | WL_SOCKET_READABLE |
            WL_TIMEOUT | WL_LATCH_SET,
            wait_fd,
            nap,
            WAIT_EVENT_WAL_RECEIVER_MAIN);
    if (rc & WL_LATCH_SET)
    {
     ResetLatch(MyLatch);
     CHECK_FOR_INTERRUPTS();

     if (walrcv->force_reply)
     {
      /*
       * The recovery process has asked us to send apply
       * feedback now.  Make sure the flag is really set to
       * false in shared memory before sending the reply, so
       * we don't miss a new request for a reply.
 */
      walrcv->force_reply = false;
      pg_memory_barrier();
      XLogWalRcvSendReply(true, false);
     }
    }
    if (rc & WL_TIMEOUT)
    {
     /*
      * We didn't receive anything new. If we haven't heard
      * anything from the server for more than
      * wal_receiver_timeout / 2, ping the server. Also, if
      * it's been longer than wal_receiver_status_interval
      * since the last update we sent, send a status update to
      * the primary anyway, to report any progress in applying
      * WAL.
 */
     bool  requestReply = false;

     /*
      * Report pending statistics to the cumulative stats
      * system.  This location is useful for the report as it
      * is not within a tight loop in the WAL receiver, to
      * avoid bloating pgstats with requests, while also making
      * sure that the reports happen each time a status update
      * is sent.
 */
     pgstat_report_wal(false);

     /*
      * Check if time since last receive from primary has
      * reached the configured limit.
 */
     now = GetCurrentTimestamp();
     if (now >= wakeup[WALRCV_WAKEUP_TERMINATE])
      ereport(ERROR,
        (errcode(ERRCODE_CONNECTION_FAILURE),
         errmsg("terminating walreceiver due to timeout")));

     /*
      * If we didn't receive anything new for half of receiver
      * replication timeout, then ping the server.
 */
     if (now >= wakeup[WALRCV_WAKEUP_PING])
     {
      requestReply = true;
      wakeup[WALRCV_WAKEUP_PING] = TIMESTAMP_INFINITY;
     }

     XLogWalRcvSendReply(requestReply, requestReply);
     XLogWalRcvSendHSFeedback(false);
    }
   }

   /*
    * The backend finished streaming. Exit streaming COPY-mode from
    * our side, too.
 */
   walrcv_endstreaming(wrconn, &primaryTLI);

   /*
    * If the server had switched to a new timeline that we didn't
    * know about when we began streaming, fetch its timeline history
    * file now.
 */
   WalRcvFetchTimeLineHistoryFiles(startpointTLI, primaryTLI);
  }
  else
   ereport(LOG,
     (errmsg("primary server contains no more WAL on requested timeline %u",
       startpointTLI)));

  /*
   * End of WAL reached on the requested timeline. Close the last
   * segment, and await for new orders from the startup process.
 */
  if (recvFile >= 0)
  {
   char  xlogfname[MAXFNAMELEN];

   XLogWalRcvFlush(false, startpointTLI);
   XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
   if (close(recvFile) != 0)
    ereport(PANIC,
      (errcode_for_file_access(),
       errmsg("could not close WAL segment %s: %m",
        xlogfname)));

   /*
    * Create .done file forcibly to prevent the streamed segment from
    * being archived later.
 */
   if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
    XLogArchiveForceDone(xlogfname);
   else
    XLogArchiveNotify(xlogfname);
  }
  recvFile = -1;

  elog(DEBUG1, "walreceiver ended streaming and awaits new instructions");
  WalRcvWaitForStartPosition(&startpoint, &startpointTLI);
 }
 /* not reached */
}

/*
 * Wait for startup process to set receiveStart and receiveStartTLI.
 */
static void
WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
{
 WalRcvData *walrcv = WalRcv;
 int   state;

 SpinLockAcquire(&walrcv->mutex);
 state = walrcv->walRcvState;
 if (state != WALRCV_STREAMING)
 {
  SpinLockRelease(&walrcv->mutex);
  if (state == WALRCV_STOPPING)
   proc_exit(0);
  else
   elog(FATAL, "unexpected walreceiver state");
 }
 walrcv->walRcvState = WALRCV_WAITING;
 walrcv->receiveStart = InvalidXLogRecPtr;
 walrcv->receiveStartTLI = 0;
 SpinLockRelease(&walrcv->mutex);

 set_ps_display("idle");

 /*
  * nudge startup process to notice that we've stopped streaming and are
  * now waiting for instructions.
 */
 WakeupRecovery();
 for (;;)
 {
  ResetLatch(MyLatch);

  CHECK_FOR_INTERRUPTS();

  SpinLockAcquire(&walrcv->mutex);
  Assert(walrcv->walRcvState == WALRCV_RESTARTING ||
      walrcv->walRcvState == WALRCV_WAITING ||
      walrcv->walRcvState == WALRCV_STOPPING);
  if (walrcv->walRcvState == WALRCV_RESTARTING)
  {
   /*
    * No need to handle changes in primary_conninfo or
    * primary_slot_name here. Startup process will signal us to
    * terminate in case those change.
 */
   *startpoint = walrcv->receiveStart;
   *startpointTLI = walrcv->receiveStartTLI;
   walrcv->walRcvState = WALRCV_STREAMING;
   SpinLockRelease(&walrcv->mutex);
   break;
  }
  if (walrcv->walRcvState == WALRCV_STOPPING)
  {
   /*
    * We should've received SIGTERM if the startup process wants us
    * to die, but might as well check it here too.
 */
   SpinLockRelease(&walrcv->mutex);
   exit(1);
  }
  SpinLockRelease(&walrcv->mutex);

  (void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
       WAIT_EVENT_WAL_RECEIVER_WAIT_START);
 }

 if (update_process_title)
 {
  char  activitymsg[50];

  snprintf(activitymsg, sizeof(activitymsg), "restarting at %X/%X",
     LSN_FORMAT_ARGS(*startpoint));
  set_ps_display(activitymsg);
 }
}

/*
 * Fetch any missing timeline history files between 'first' and 'last'
 * (inclusive) from the server.
 */
static void
WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last)
{
 TimeLineID tli;

 for (tli = first; tli <= last; tli++)
 {
  /* there's no history file for timeline 1 */
  if (tli != 1 && !existsTimeLineHistory(tli))
  {
   char    *fname;
   char    *content;
   int   len;
   char  expectedfname[MAXFNAMELEN];

   ereport(LOG,
     (errmsg("fetching timeline history file for timeline %u from primary server",
       tli)));

   walrcv_readtimelinehistoryfile(wrconn, tli, &fname, &content, &len);

   /*
    * Check that the filename on the primary matches what we
    * calculated ourselves. This is just a sanity check, it should
    * always match.
 */
   TLHistoryFileName(expectedfname, tli);
   if (strcmp(fname, expectedfname) != 0)
    ereport(ERROR,
      (errcode(ERRCODE_PROTOCOL_VIOLATION),
       errmsg_internal("primary reported unexpected file name for timeline history file of timeline %u",
           tli)));

   /*
    * Write the file to pg_wal.
 */
   writeTimeLineHistoryFile(tli, content, len);

   /*
    * Mark the streamed history file as ready for archiving if
    * archive_mode is always.
 */
   if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
    XLogArchiveForceDone(fname);
   else
    XLogArchiveNotify(fname);

   pfree(fname);
   pfree(content);
  }
 }
}

/*
 * Mark us as STOPPED in shared memory at exit.
 */
static void
WalRcvDie(int code, Datum arg)
{
 WalRcvData *walrcv = WalRcv;
 TimeLineID *startpointTLI_p = (TimeLineID *) DatumGetPointer(arg);

 Assert(*startpointTLI_p != 0);

 /* Ensure that all WAL records received are flushed to disk */
 XLogWalRcvFlush(true, *startpointTLI_p);

 /* Mark ourselves inactive in shared memory */
 SpinLockAcquire(&walrcv->mutex);
 Assert(walrcv->walRcvState == WALRCV_STREAMING ||
     walrcv->walRcvState == WALRCV_RESTARTING ||
     walrcv->walRcvState == WALRCV_STARTING ||
     walrcv->walRcvState == WALRCV_WAITING ||
     walrcv->walRcvState == WALRCV_STOPPING);
 Assert(walrcv->pid == MyProcPid);
 walrcv->walRcvState = WALRCV_STOPPED;
 walrcv->pid = 0;
 walrcv->procno = INVALID_PROC_NUMBER;
 walrcv->ready_to_display = false;
 SpinLockRelease(&walrcv->mutex);

 ConditionVariableBroadcast(&walrcv->walRcvStoppedCV);

 /* Terminate the connection gracefully. */
 if (wrconn != NULL)
  walrcv_disconnect(wrconn);

 /* Wake up the startup process to notice promptly that we're gone */
 WakeupRecovery();
}

/*
 * Accept the message from XLOG stream, and process it.
 */
static void
XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len, TimeLineID tli)
{
 int   hdrlen;
 XLogRecPtr dataStart;
 XLogRecPtr walEnd;
 TimestampTz sendTime;
 bool  replyRequested;

 switch (type)
 {
  case 'w':    /* WAL records */
   {
    StringInfoData incoming_message;

    hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64);
    if (len < hdrlen)
     ereport(ERROR,
       (errcode(ERRCODE_PROTOCOL_VIOLATION),
        errmsg_internal("invalid WAL message received from primary")));

    /* initialize a StringInfo with the given buffer */
    initReadOnlyStringInfo(&incoming_message, buf, hdrlen);

    /* read the fields */
    dataStart = pq_getmsgint64(&incoming_message);
    walEnd = pq_getmsgint64(&incoming_message);
    sendTime = pq_getmsgint64(&incoming_message);
    ProcessWalSndrMessage(walEnd, sendTime);

    buf += hdrlen;
    len -= hdrlen;
    XLogWalRcvWrite(buf, len, dataStart, tli);
    break;
   }
  case 'k':    /* Keepalive */
   {
    StringInfoData incoming_message;

    hdrlen = sizeof(int64) + sizeof(int64) + sizeof(char);
    if (len != hdrlen)
     ereport(ERROR,
       (errcode(ERRCODE_PROTOCOL_VIOLATION),
        errmsg_internal("invalid keepalive message received from primary")));

    /* initialize a StringInfo with the given buffer */
    initReadOnlyStringInfo(&incoming_message, buf, hdrlen);

    /* read the fields */
    walEnd = pq_getmsgint64(&incoming_message);
    sendTime = pq_getmsgint64(&incoming_message);
    replyRequested = pq_getmsgbyte(&incoming_message);

    ProcessWalSndrMessage(walEnd, sendTime);

    /* If the primary requested a reply, send one immediately */
    if (replyRequested)
     XLogWalRcvSendReply(true, false);
    break;
   }
  default:
   ereport(ERROR,
     (errcode(ERRCODE_PROTOCOL_VIOLATION),
      errmsg_internal("invalid replication message type %d",
          type)));
 }
}

/*
 * Write XLOG data to disk.
 */
static void
XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr, TimeLineID tli)
{
 int   startoff;
 int   byteswritten;
 instr_time start;

 Assert(tli != 0);

 while (nbytes > 0)
 {
  int   segbytes;

  /* Close the current segment if it's completed */
  if (recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size))
   XLogWalRcvClose(recptr, tli);

  if (recvFile < 0)
  {
   /* Create/use new log file */
   XLByteToSeg(recptr, recvSegNo, wal_segment_size);
   recvFile = XLogFileInit(recvSegNo, tli);
   recvFileTLI = tli;
  }

  /* Calculate the start offset of the received logs */
  startoff = XLogSegmentOffset(recptr, wal_segment_size);

  if (startoff + nbytes > wal_segment_size)
   segbytes = wal_segment_size - startoff;
  else
   segbytes = nbytes;

  /* OK to write the logs */
  errno = 0;

  /*
   * Measure I/O timing to write WAL data, for pg_stat_io.
 */
  start = pgstat_prepare_io_time(track_wal_io_timing);

  pgstat_report_wait_start(WAIT_EVENT_WAL_WRITE);
  byteswritten = pg_pwrite(recvFile, buf, segbytes, (off_t) startoff);
  pgstat_report_wait_end();

  pgstat_count_io_op_time(IOOBJECT_WAL, IOCONTEXT_NORMAL,
        IOOP_WRITE, start, 1, byteswritten);

  if (byteswritten <= 0)
  {
   char  xlogfname[MAXFNAMELEN];
   int   save_errno;

   /* if write didn't set errno, assume no disk space */
   if (errno == 0)
    errno = ENOSPC;

   save_errno = errno;
   XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size);
   errno = save_errno;
   ereport(PANIC,
     (errcode_for_file_access(),
      errmsg("could not write to WAL segment %s "
       "at offset %d, length %lu: %m",
       xlogfname, startoff, (unsigned long) segbytes)));
  }

  /* Update state for write */
  recptr += byteswritten;

  nbytes -= byteswritten;
  buf += byteswritten;

  LogstreamResult.Write = recptr;
 }

 /* Update shared-memory status */
 pg_atomic_write_u64(&WalRcv->writtenUpto, LogstreamResult.Write);

 /*
  * Close the current segment if it's fully written up in the last cycle of
  * the loop, to create its archive notification file soon. Otherwise WAL
  * archiving of the segment will be delayed until any data in the next
  * segment is received and written.
 */
 if (recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size))
  XLogWalRcvClose(recptr, tli);
}

/*
 * Flush the log to disk.
 *
 * If we're in the midst of dying, it's unwise to do anything that might throw
 * an error, so we skip sending a reply in that case.
 */
static void
XLogWalRcvFlush(bool dying, TimeLineID tli)
{
 Assert(tli != 0);

 if (LogstreamResult.Flush < LogstreamResult.Write)
 {
  WalRcvData *walrcv = WalRcv;

  issue_xlog_fsync(recvFile, recvSegNo, tli);

  LogstreamResult.Flush = LogstreamResult.Write;

  /* Update shared-memory status */
  SpinLockAcquire(&walrcv->mutex);
  if (walrcv->flushedUpto < LogstreamResult.Flush)
  {
   walrcv->latestChunkStart = walrcv->flushedUpto;
   walrcv->flushedUpto = LogstreamResult.Flush;
   walrcv->receivedTLI = tli;
  }
  SpinLockRelease(&walrcv->mutex);

  /* Signal the startup process and walsender that new WAL has arrived */
  WakeupRecovery();
  if (AllowCascadeReplication())
   WalSndWakeup(true, false);

  /* Report XLOG streaming progress in PS display */
  if (update_process_title)
  {
   char  activitymsg[50];

   snprintf(activitymsg, sizeof(activitymsg), "streaming %X/%X",
      LSN_FORMAT_ARGS(LogstreamResult.Write));
   set_ps_display(activitymsg);
  }

  /* Also let the primary know that we made some progress */
  if (!dying)
  {
   XLogWalRcvSendReply(false, false);
   XLogWalRcvSendHSFeedback(false);
  }
 }
}

/*
 * Close the current segment.
 *
 * Flush the segment to disk before closing it. Otherwise we have to
 * reopen and fsync it later.
 *
 * Create an archive notification file since the segment is known completed.
 */
static void
XLogWalRcvClose(XLogRecPtr recptr, TimeLineID tli)
{
 char  xlogfname[MAXFNAMELEN];

 Assert(recvFile >= 0 && !XLByteInSeg(recptr, recvSegNo, wal_segment_size));
 Assert(tli != 0);

 /*
  * fsync() and close current file before we switch to next one. We would
  * otherwise have to reopen this file to fsync it later
 */
 XLogWalRcvFlush(false, tli);

 XLogFileName(xlogfname, recvFileTLI, recvSegNo, wal_segment_size);

 /*
  * XLOG segment files will be re-read by recovery in startup process soon,
  * so we don't advise the OS to release cache pages associated with the
  * file like XLogFileClose() does.
 */
 if (close(recvFile) != 0)
  ereport(PANIC,
    (errcode_for_file_access(),
     errmsg("could not close WAL segment %s: %m",
      xlogfname)));

 /*
  * Create .done file forcibly to prevent the streamed segment from being
  * archived later.
 */
 if (XLogArchiveMode != ARCHIVE_MODE_ALWAYS)
  XLogArchiveForceDone(xlogfname);
 else
  XLogArchiveNotify(xlogfname);

 recvFile = -1;
}

/*
 * Send reply message to primary, indicating our current WAL locations, oldest
 * xmin and the current time.
 *
 * If 'force' is not set, the message is only sent if enough time has
 * passed since last status update to reach wal_receiver_status_interval.
 * If wal_receiver_status_interval is disabled altogether and 'force' is
 * false, this is a no-op.
 *
 * If 'requestReply' is true, requests the server to reply immediately upon
 * receiving this message. This is used for heartbeats, when approaching
 * wal_receiver_timeout.
 */
static void
XLogWalRcvSendReply(bool force, bool requestReply)
{
 static XLogRecPtr writePtr = 0;
 static XLogRecPtr flushPtr = 0;
 XLogRecPtr applyPtr;
 TimestampTz now;

 /*
  * If the user doesn't want status to be reported to the primary, be sure
  * to exit before doing anything at all.
 */
 if (!force && wal_receiver_status_interval <= 0)
  return;

 /* Get current timestamp. */
 now = GetCurrentTimestamp();

 /*
  * We can compare the write and flush positions to the last message we
  * sent without taking any lock, but the apply position requires a spin
  * lock, so we don't check that unless something else has changed or 10
  * seconds have passed.  This means that the apply WAL location will
  * appear, from the primary's point of view, to lag slightly, but since
  * this is only for reporting purposes and only on idle systems, that's
  * probably OK.
 */
 if (!force
  && writePtr == LogstreamResult.Write
  && flushPtr == LogstreamResult.Flush
  && now < wakeup[WALRCV_WAKEUP_REPLY])
  return;

 /* Make sure we wake up when it's time to send another reply. */
 WalRcvComputeNextWakeup(WALRCV_WAKEUP_REPLY, now);

 /* Construct a new message */
 writePtr = LogstreamResult.Write;
 flushPtr = LogstreamResult.Flush;
 applyPtr = GetXLogReplayRecPtr(NULL);

 resetStringInfo(&reply_message);
 pq_sendbyte(&reply_message, 'r');
 pq_sendint64(&reply_message, writePtr);
 pq_sendint64(&reply_message, flushPtr);
 pq_sendint64(&reply_message, applyPtr);
 pq_sendint64(&reply_message, GetCurrentTimestamp());
 pq_sendbyte(&reply_message, requestReply ? 1 : 0);

 /* Send it */
 elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X%s",
   LSN_FORMAT_ARGS(writePtr),
   LSN_FORMAT_ARGS(flushPtr),
   LSN_FORMAT_ARGS(applyPtr),
   requestReply ? " (reply requested)" : "");

 walrcv_send(wrconn, reply_message.data, reply_message.len);
}

/*
 * Send hot standby feedback message to primary, plus the current time,
 * in case they don't have a watch.
 *
 * If the user disables feedback, send one final message to tell sender
 * to forget about the xmin on this standby. We also send this message
 * on first connect because a previous connection might have set xmin
 * on a replication slot. (If we're not using a slot it's harmless to
 * send a feedback message explicitly setting InvalidTransactionId).
 */
static void
XLogWalRcvSendHSFeedback(bool immed)
{
 TimestampTz now;
 FullTransactionId nextFullXid;
 TransactionId nextXid;
 uint32  xmin_epoch,
    catalog_xmin_epoch;
 TransactionId xmin,
    catalog_xmin;

 /* initially true so we always send at least one feedback message */
 static bool primary_has_standby_xmin = true;

 /*
  * If the user doesn't want status to be reported to the primary, be sure
  * to exit before doing anything at all.
 */
 if ((wal_receiver_status_interval <= 0 || !hot_standby_feedback) &&
  !primary_has_standby_xmin)
  return;

 /* Get current timestamp. */
 now = GetCurrentTimestamp();

 /* Send feedback at most once per wal_receiver_status_interval. */
 if (!immed && now < wakeup[WALRCV_WAKEUP_HSFEEDBACK])
  return;

 /* Make sure we wake up when it's time to send feedback again. */
 WalRcvComputeNextWakeup(WALRCV_WAKEUP_HSFEEDBACK, now);

 /*
  * If Hot Standby is not yet accepting connections there is nothing to
  * send. Check this after the interval has expired to reduce number of
  * calls.
  *
  * Bailing out here also ensures that we don't send feedback until we've
  * read our own replication slot state, so we don't tell the primary to
  * discard needed xmin or catalog_xmin from any slots that may exist on
  * this replica.
 */
 if (!HotStandbyActive())
  return;

 /*
  * Make the expensive call to get the oldest xmin once we are certain
  * everything else has been checked.
 */
 if (hot_standby_feedback)
 {
  GetReplicationHorizons(&xmin, &catalog_xmin);
 }
 else
 {
  xmin = InvalidTransactionId;
  catalog_xmin = InvalidTransactionId;
 }

 /*
  * Get epoch and adjust if nextXid and oldestXmin are different sides of
  * the epoch boundary.
 */
 nextFullXid = ReadNextFullTransactionId();
 nextXid = XidFromFullTransactionId(nextFullXid);
 xmin_epoch = EpochFromFullTransactionId(nextFullXid);
 catalog_xmin_epoch = xmin_epoch;
 if (nextXid < xmin)
  xmin_epoch--;
 if (nextXid < catalog_xmin)
  catalog_xmin_epoch--;

 elog(DEBUG2, "sending hot standby feedback xmin %u epoch %u catalog_xmin %u catalog_xmin_epoch %u",
   xmin, xmin_epoch, catalog_xmin, catalog_xmin_epoch);

 /* Construct the message and send it. */
 resetStringInfo(&reply_message);
 pq_sendbyte(&reply_message, 'h');
 pq_sendint64(&reply_message, GetCurrentTimestamp());
 pq_sendint32(&reply_message, xmin);
 pq_sendint32(&reply_message, xmin_epoch);
 pq_sendint32(&reply_message, catalog_xmin);
 pq_sendint32(&reply_message, catalog_xmin_epoch);
 walrcv_send(wrconn, reply_message.data, reply_message.len);
 if (TransactionIdIsValid(xmin) || TransactionIdIsValid(catalog_xmin))
  primary_has_standby_xmin = true;
 else
  primary_has_standby_xmin = false;
}

/*
 * Update shared memory status upon receiving a message from primary.
 *
 * 'walEnd' and 'sendTime' are the end-of-WAL and timestamp of the latest
 * message, reported by primary.
 */
static void
ProcessWalSndrMessage(XLogRecPtr walEnd, TimestampTz sendTime)
{
 WalRcvData *walrcv = WalRcv;
 TimestampTz lastMsgReceiptTime = GetCurrentTimestamp();

 /* Update shared-memory status */
 SpinLockAcquire(&walrcv->mutex);
 if (walrcv->latestWalEnd < walEnd)
  walrcv->latestWalEndTime = sendTime;
 walrcv->latestWalEnd = walEnd;
 walrcv->lastMsgSendTime = sendTime;
 walrcv->lastMsgReceiptTime = lastMsgReceiptTime;
 SpinLockRelease(&walrcv->mutex);

 if (message_level_is_interesting(DEBUG2))
 {
  char    *sendtime;
  char    *receipttime;
  int   applyDelay;

  /* Copy because timestamptz_to_str returns a static buffer */
  sendtime = pstrdup(timestamptz_to_str(sendTime));
  receipttime = pstrdup(timestamptz_to_str(lastMsgReceiptTime));
  applyDelay = GetReplicationApplyDelay();

  /* apply delay is not available */
  if (applyDelay == -1)
   elog(DEBUG2, "sendtime %s receipttime %s replication apply delay (N/A) transfer latency %d ms",
     sendtime,
     receipttime,
     GetReplicationTransferLatency());
  else
   elog(DEBUG2, "sendtime %s receipttime %s replication apply delay %d ms transfer latency %d ms",
     sendtime,
     receipttime,
     applyDelay,
     GetReplicationTransferLatency());

  pfree(sendtime);
  pfree(receipttime);
 }
}

/*
 * Compute the next wakeup time for a given wakeup reason.  Can be called to
 * initialize a wakeup time, to adjust it for the next wakeup, or to
 * reinitialize it when GUCs have changed.  We ask the caller to pass in the
 * value of "now" because this frequently avoids multiple calls of
 * GetCurrentTimestamp().  It had better be a reasonably up-to-date value
 * though.
 */
static void
WalRcvComputeNextWakeup(WalRcvWakeupReason reason, TimestampTz now)
{
 switch (reason)
 {
  case WALRCV_WAKEUP_TERMINATE:
   if (wal_receiver_timeout <= 0)
    wakeup[reason] = TIMESTAMP_INFINITY;
   else
    wakeup[reason] = TimestampTzPlusMilliseconds(now, wal_receiver_timeout);
   break;
  case WALRCV_WAKEUP_PING:
   if (wal_receiver_timeout <= 0)
    wakeup[reason] = TIMESTAMP_INFINITY;
   else
    wakeup[reason] = TimestampTzPlusMilliseconds(now, wal_receiver_timeout / 2);
   break;
  case WALRCV_WAKEUP_HSFEEDBACK:
   if (!hot_standby_feedback || wal_receiver_status_interval <= 0)
    wakeup[reason] = TIMESTAMP_INFINITY;
   else
    wakeup[reason] = TimestampTzPlusSeconds(now, wal_receiver_status_interval);
   break;
  case WALRCV_WAKEUP_REPLY:
   if (wal_receiver_status_interval <= 0)
    wakeup[reason] = TIMESTAMP_INFINITY;
   else
    wakeup[reason] = TimestampTzPlusSeconds(now, wal_receiver_status_interval);
   break;
   /* there's intentionally no default: here */
 }
}

/*
 * Wake up the walreceiver main loop.
 *
 * This is called by the startup process whenever interesting xlog records
 * are applied, so that walreceiver can check if it needs to send an apply
 * notification back to the primary which may be waiting in a COMMIT with
 * synchronous_commit = remote_apply.
 */
void
WalRcvForceReply(void)
{
 ProcNumber procno;

 WalRcv->force_reply = true;
 /* fetching the proc number is probably atomic, but don't rely on it */
 SpinLockAcquire(&WalRcv->mutex);
 procno = WalRcv->procno;
 SpinLockRelease(&WalRcv->mutex);
 if (procno != INVALID_PROC_NUMBER)
  SetLatch(&GetPGProcByNumber(procno)->procLatch);
}

/*
 * Return a string constant representing the state. This is used
 * in system functions and views, and should *not* be translated.
 */
static const char *
WalRcvGetStateString(WalRcvState state)
{
 switch (state)
 {
  case WALRCV_STOPPED:
   return "stopped";
  case WALRCV_STARTING:
   return "starting";
  case WALRCV_STREAMING:
   return "streaming";
  case WALRCV_WAITING:
   return "waiting";
  case WALRCV_RESTARTING:
   return "restarting";
  case WALRCV_STOPPING:
   return "stopping";
 }
 return "UNKNOWN";
}

/*
 * Returns activity of WAL receiver, including pid, state and xlog locations
 * received from the WAL sender of another server.
 */
Datum
pg_stat_get_wal_receiver(PG_FUNCTION_ARGS)
{
 TupleDesc tupdesc;
 Datum    *values;
 bool    *nulls;
 int   pid;
 bool  ready_to_display;
 WalRcvState state;
 XLogRecPtr receive_start_lsn;
 TimeLineID receive_start_tli;
 XLogRecPtr written_lsn;
 XLogRecPtr flushed_lsn;
 TimeLineID received_tli;
 TimestampTz last_send_time;
 TimestampTz last_receipt_time;
 XLogRecPtr latest_end_lsn;
 TimestampTz latest_end_time;
 char  sender_host[NI_MAXHOST];
 int   sender_port = 0;
 char  slotname[NAMEDATALEN];
 char  conninfo[MAXCONNINFO];

 /* Take a lock to ensure value consistency */
 SpinLockAcquire(&WalRcv->mutex);
 pid = (int) WalRcv->pid;
 ready_to_display = WalRcv->ready_to_display;
 state = WalRcv->walRcvState;
 receive_start_lsn = WalRcv->receiveStart;
 receive_start_tli = WalRcv->receiveStartTLI;
 flushed_lsn = WalRcv->flushedUpto;
 received_tli = WalRcv->receivedTLI;
 last_send_time = WalRcv->lastMsgSendTime;
 last_receipt_time = WalRcv->lastMsgReceiptTime;
 latest_end_lsn = WalRcv->latestWalEnd;
 latest_end_time = WalRcv->latestWalEndTime;
 strlcpy(slotname, WalRcv->slotname, sizeof(slotname));
 strlcpy(sender_host, WalRcv->sender_host, sizeof(sender_host));
 sender_port = WalRcv->sender_port;
 strlcpy(conninfo, WalRcv->conninfo, sizeof(conninfo));
 SpinLockRelease(&WalRcv->mutex);

 /*
  * No WAL receiver (or not ready yet), just return a tuple with NULL
  * values
 */
 if (pid == 0 || !ready_to_display)
  PG_RETURN_NULL();

 /*
  * Read "writtenUpto" without holding a spinlock.  Note that it may not be
  * consistent with the other shared variables of the WAL receiver
  * protected by a spinlock, but this should not be used for data integrity
  * checks.
 */
 written_lsn = pg_atomic_read_u64(&WalRcv->writtenUpto);

 /* determine result type */
 if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
  elog(ERROR, "return type must be a row type");

 values = palloc0(sizeof(Datum) * tupdesc->natts);
 nulls = palloc0(sizeof(bool) * tupdesc->natts);

 /* Fetch values */
 values[0] = Int32GetDatum(pid);

 if (!has_privs_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS))
 {
  /*
   * Only superusers and roles with privileges of pg_read_all_stats can
   * see details. Other users only get the pid value to know whether it
   * is a WAL receiver, but no details.
 */
  memset(&nulls[1], true, sizeof(bool) * (tupdesc->natts - 1));
 }
 else
 {
  values[1] = CStringGetTextDatum(WalRcvGetStateString(state));

  if (XLogRecPtrIsInvalid(receive_start_lsn))
   nulls[2] = true;
  else
   values[2] = LSNGetDatum(receive_start_lsn);
  values[3] = Int32GetDatum(receive_start_tli);
  if (XLogRecPtrIsInvalid(written_lsn))
   nulls[4] = true;
  else
   values[4] = LSNGetDatum(written_lsn);
  if (XLogRecPtrIsInvalid(flushed_lsn))
   nulls[5] = true;
  else
   values[5] = LSNGetDatum(flushed_lsn);
  values[6] = Int32GetDatum(received_tli);
  if (last_send_time == 0)
   nulls[7] = true;
  else
   values[7] = TimestampTzGetDatum(last_send_time);
  if (last_receipt_time == 0)
   nulls[8] = true;
  else
   values[8] = TimestampTzGetDatum(last_receipt_time);
  if (XLogRecPtrIsInvalid(latest_end_lsn))
   nulls[9] = true;
  else
   values[9] = LSNGetDatum(latest_end_lsn);
  if (latest_end_time == 0)
   nulls[10] = true;
  else
   values[10] = TimestampTzGetDatum(latest_end_time);
  if (*slotname == '\0')
   nulls[11] = true;
  else
   values[11] = CStringGetTextDatum(slotname);
  if (*sender_host == '\0')
   nulls[12] = true;
  else
   values[12] = CStringGetTextDatum(sender_host);
  if (sender_port == 0)
   nulls[13] = true;
  else
   values[13] = Int32GetDatum(sender_port);
  if (*conninfo == '\0')
   nulls[14] = true;
  else
   values[14] = CStringGetTextDatum(conninfo);
 }

 /* Returns the record as Datum */
 PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
}