LCOV - code coverage report
Current view: top level - src/bin/pg_basebackup - receivelog.c (source / functions) Hit Total Coverage
Test: PostgreSQL 13beta1 Lines: 216 424 50.9 %
Date: 2020-05-29 01:06:25 Functions: 14 17 82.4 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * receivelog.c - receive WAL files using the streaming
       4             :  *                replication protocol.
       5             :  *
       6             :  * Author: Magnus Hagander <magnus@hagander.net>
       7             :  *
       8             :  * Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group
       9             :  *
      10             :  * IDENTIFICATION
      11             :  *        src/bin/pg_basebackup/receivelog.c
      12             :  *-------------------------------------------------------------------------
      13             :  */
      14             : 
      15             : #include "postgres_fe.h"
      16             : 
      17             : #include <sys/stat.h>
      18             : #include <unistd.h>
      19             : #ifdef HAVE_SYS_SELECT_H
      20             : #include <sys/select.h>
      21             : #endif
      22             : 
      23             : #include "access/xlog_internal.h"
      24             : #include "common/file_utils.h"
      25             : #include "common/logging.h"
      26             : #include "libpq-fe.h"
      27             : #include "receivelog.h"
      28             : #include "streamutil.h"
      29             : 
      30             : /* fd and filename for currently open WAL file */
      31             : static Walfile *walfile = NULL;
      32             : static char current_walfile_name[MAXPGPATH] = "";
      33             : static bool reportFlushPosition = false;
      34             : static XLogRecPtr lastFlushPosition = InvalidXLogRecPtr;
      35             : 
      36             : static bool still_sending = true;   /* feedback still needs to be sent? */
      37             : 
      38             : static PGresult *HandleCopyStream(PGconn *conn, StreamCtl *stream,
      39             :                                   XLogRecPtr *stoppos);
      40             : static int  CopyStreamPoll(PGconn *conn, long timeout_ms, pgsocket stop_socket);
      41             : static int  CopyStreamReceive(PGconn *conn, long timeout, pgsocket stop_socket,
      42             :                               char **buffer);
      43             : static bool ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf,
      44             :                                 int len, XLogRecPtr blockpos, TimestampTz *last_status);
      45             : static bool ProcessXLogDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
      46             :                                XLogRecPtr *blockpos);
      47             : static PGresult *HandleEndOfCopyStream(PGconn *conn, StreamCtl *stream, char *copybuf,
      48             :                                        XLogRecPtr blockpos, XLogRecPtr *stoppos);
      49             : static bool CheckCopyStreamStop(PGconn *conn, StreamCtl *stream, XLogRecPtr blockpos,
      50             :                                 XLogRecPtr *stoppos);
      51             : static long CalculateCopyStreamSleeptime(TimestampTz now, int standby_message_timeout,
      52             :                                          TimestampTz last_status);
      53             : 
      54             : static bool ReadEndOfStreamingResult(PGresult *res, XLogRecPtr *startpos,
      55             :                                      uint32 *timeline);
      56             : 
      57             : static bool
      58          10 : mark_file_as_archived(StreamCtl *stream, const char *fname)
      59             : {
      60             :     Walfile    *f;
      61             :     static char tmppath[MAXPGPATH];
      62             : 
      63          10 :     snprintf(tmppath, sizeof(tmppath), "archive_status/%s.done",
      64             :              fname);
      65             : 
      66          10 :     f = stream->walmethod->open_for_write(tmppath, NULL, 0);
      67          10 :     if (f == NULL)
      68             :     {
      69           0 :         pg_log_error("could not create archive status file \"%s\": %s",
      70             :                      tmppath, stream->walmethod->getlasterror());
      71           0 :         return false;
      72             :     }
      73             : 
      74          10 :     stream->walmethod->close(f, CLOSE_NORMAL);
      75             : 
      76          10 :     return true;
      77             : }
      78             : 
      79             : /*
      80             :  * Open a new WAL file in the specified directory.
      81             :  *
      82             :  * Returns true if OK; on failure, returns false after printing an error msg.
      83             :  * On success, 'walfile' is set to the FD for the file, and the base filename
      84             :  * (without partial_suffix) is stored in 'current_walfile_name'.
      85             :  *
      86             :  * The file will be padded to 16Mb with zeroes.
      87             :  */
      88             : static bool
      89         128 : open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
      90             : {
      91             :     Walfile    *f;
      92             :     char        fn[MAXPGPATH];
      93             :     ssize_t     size;
      94             :     XLogSegNo   segno;
      95             : 
      96         128 :     XLByteToSeg(startpoint, segno, WalSegSz);
      97         128 :     XLogFileName(current_walfile_name, stream->timeline, segno, WalSegSz);
      98             : 
      99         128 :     snprintf(fn, sizeof(fn), "%s%s", current_walfile_name,
     100         128 :              stream->partial_suffix ? stream->partial_suffix : "");
     101             : 
     102             :     /*
     103             :      * When streaming to files, if an existing file exists we verify that it's
     104             :      * either empty (just created), or a complete WalSegSz segment (in which
     105             :      * case it has been created and padded). Anything else indicates a corrupt
     106             :      * file.
     107             :      *
     108             :      * When streaming to tar, no file with this name will exist before, so we
     109             :      * never have to verify a size.
     110             :      */
     111         128 :     if (stream->walmethod->existsfile(fn))
     112             :     {
     113           0 :         size = stream->walmethod->get_file_size(fn);
     114           0 :         if (size < 0)
     115             :         {
     116           0 :             pg_log_error("could not get size of write-ahead log file \"%s\": %s",
     117             :                          fn, stream->walmethod->getlasterror());
     118           0 :             return false;
     119             :         }
     120           0 :         if (size == WalSegSz)
     121             :         {
     122             :             /* Already padded file. Open it for use */
     123           0 :             f = stream->walmethod->open_for_write(current_walfile_name, stream->partial_suffix, 0);
     124           0 :             if (f == NULL)
     125             :             {
     126           0 :                 pg_log_error("could not open existing write-ahead log file \"%s\": %s",
     127             :                              fn, stream->walmethod->getlasterror());
     128           0 :                 return false;
     129             :             }
     130             : 
     131             :             /* fsync file in case of a previous crash */
     132           0 :             if (stream->walmethod->sync(f) != 0)
     133             :             {
     134           0 :                 pg_log_fatal("could not fsync existing write-ahead log file \"%s\": %s",
     135             :                              fn, stream->walmethod->getlasterror());
     136           0 :                 stream->walmethod->close(f, CLOSE_UNLINK);
     137           0 :                 exit(1);
     138             :             }
     139             : 
     140           0 :             walfile = f;
     141           0 :             return true;
     142             :         }
     143           0 :         if (size != 0)
     144             :         {
     145             :             /* if write didn't set errno, assume problem is no disk space */
     146           0 :             if (errno == 0)
     147           0 :                 errno = ENOSPC;
     148           0 :             pg_log_error(ngettext("write-ahead log file \"%s\" has %d byte, should be 0 or %d",
     149             :                                   "write-ahead log file \"%s\" has %d bytes, should be 0 or %d",
     150             :                                   size),
     151             :                          fn, (int) size, WalSegSz);
     152           0 :             return false;
     153             :         }
     154             :         /* File existed and was empty, so fall through and open */
     155             :     }
     156             : 
     157             :     /* No file existed, so create one */
     158             : 
     159         256 :     f = stream->walmethod->open_for_write(current_walfile_name,
     160         128 :                                           stream->partial_suffix, WalSegSz);
     161         128 :     if (f == NULL)
     162             :     {
     163           0 :         pg_log_error("could not open write-ahead log file \"%s\": %s",
     164             :                      fn, stream->walmethod->getlasterror());
     165           0 :         return false;
     166             :     }
     167             : 
     168         128 :     walfile = f;
     169         128 :     return true;
     170             : }
     171             : 
     172             : /*
     173             :  * Close the current WAL file (if open), and rename it to the correct
     174             :  * filename if it's complete. On failure, prints an error message to stderr
     175             :  * and returns false, otherwise returns true.
     176             :  */
     177             : static bool
     178         136 : close_walfile(StreamCtl *stream, XLogRecPtr pos)
     179             : {
     180             :     off_t       currpos;
     181             :     int         r;
     182             : 
     183         136 :     if (walfile == NULL)
     184           8 :         return true;
     185             : 
     186         128 :     currpos = stream->walmethod->get_current_pos(walfile);
     187         128 :     if (currpos == -1)
     188             :     {
     189           0 :         pg_log_error("could not determine seek position in file \"%s\": %s",
     190             :                      current_walfile_name, stream->walmethod->getlasterror());
     191           0 :         stream->walmethod->close(walfile, CLOSE_UNLINK);
     192           0 :         walfile = NULL;
     193             : 
     194           0 :         return false;
     195             :     }
     196             : 
     197         128 :     if (stream->partial_suffix)
     198             :     {
     199           2 :         if (currpos == WalSegSz)
     200           0 :             r = stream->walmethod->close(walfile, CLOSE_NORMAL);
     201             :         else
     202             :         {
     203           2 :             pg_log_info("not renaming \"%s%s\", segment is not complete",
     204             :                         current_walfile_name, stream->partial_suffix);
     205           2 :             r = stream->walmethod->close(walfile, CLOSE_NO_RENAME);
     206             :         }
     207             :     }
     208             :     else
     209         126 :         r = stream->walmethod->close(walfile, CLOSE_NORMAL);
     210             : 
     211         128 :     walfile = NULL;
     212             : 
     213         128 :     if (r != 0)
     214             :     {
     215           0 :         pg_log_error("could not close file \"%s\": %s",
     216             :                      current_walfile_name, stream->walmethod->getlasterror());
     217           0 :         return false;
     218             :     }
     219             : 
     220             :     /*
     221             :      * Mark file as archived if requested by the caller - pg_basebackup needs
     222             :      * to do so as files can otherwise get archived again after promotion of a
     223             :      * new node. This is in line with walreceiver.c always doing a
     224             :      * XLogArchiveForceDone() after a complete segment.
     225             :      */
     226         128 :     if (currpos == WalSegSz && stream->mark_done)
     227             :     {
     228             :         /* writes error message if failed */
     229          10 :         if (!mark_file_as_archived(stream, current_walfile_name))
     230           0 :             return false;
     231             :     }
     232             : 
     233         128 :     lastFlushPosition = pos;
     234         128 :     return true;
     235             : }
     236             : 
     237             : 
     238             : /*
     239             :  * Check if a timeline history file exists.
     240             :  */
     241             : static bool
     242         130 : existsTimeLineHistoryFile(StreamCtl *stream)
     243             : {
     244             :     char        histfname[MAXFNAMELEN];
     245             : 
     246             :     /*
     247             :      * Timeline 1 never has a history file. We treat that as if it existed,
     248             :      * since we never need to stream it.
     249             :      */
     250         130 :     if (stream->timeline == 1)
     251         130 :         return true;
     252             : 
     253           0 :     TLHistoryFileName(histfname, stream->timeline);
     254             : 
     255           0 :     return stream->walmethod->existsfile(histfname);
     256             : }
     257             : 
     258             : static bool
     259           0 : writeTimeLineHistoryFile(StreamCtl *stream, char *filename, char *content)
     260             : {
     261           0 :     int         size = strlen(content);
     262             :     char        histfname[MAXFNAMELEN];
     263             :     Walfile    *f;
     264             : 
     265             :     /*
     266             :      * Check that the server's idea of how timeline history files should be
     267             :      * named matches ours.
     268             :      */
     269           0 :     TLHistoryFileName(histfname, stream->timeline);
     270           0 :     if (strcmp(histfname, filename) != 0)
     271             :     {
     272           0 :         pg_log_error("server reported unexpected history file name for timeline %u: %s",
     273             :                      stream->timeline, filename);
     274           0 :         return false;
     275             :     }
     276             : 
     277           0 :     f = stream->walmethod->open_for_write(histfname, ".tmp", 0);
     278           0 :     if (f == NULL)
     279             :     {
     280           0 :         pg_log_error("could not create timeline history file \"%s\": %s",
     281             :                      histfname, stream->walmethod->getlasterror());
     282           0 :         return false;
     283             :     }
     284             : 
     285           0 :     if ((int) stream->walmethod->write(f, content, size) != size)
     286             :     {
     287           0 :         pg_log_error("could not write timeline history file \"%s\": %s",
     288             :                      histfname, stream->walmethod->getlasterror());
     289             : 
     290             :         /*
     291             :          * If we fail to make the file, delete it to release disk space
     292             :          */
     293           0 :         stream->walmethod->close(f, CLOSE_UNLINK);
     294             : 
     295           0 :         return false;
     296             :     }
     297             : 
     298           0 :     if (stream->walmethod->close(f, CLOSE_NORMAL) != 0)
     299             :     {
     300           0 :         pg_log_error("could not close file \"%s\": %s",
     301             :                      histfname, stream->walmethod->getlasterror());
     302           0 :         return false;
     303             :     }
     304             : 
     305             :     /* Maintain archive_status, check close_walfile() for details. */
     306           0 :     if (stream->mark_done)
     307             :     {
     308             :         /* writes error message if failed */
     309           0 :         if (!mark_file_as_archived(stream, histfname))
     310           0 :             return false;
     311             :     }
     312             : 
     313           0 :     return true;
     314             : }
     315             : 
     316             : /*
     317             :  * Send a Standby Status Update message to server.
     318             :  */
     319             : static bool
     320         128 : sendFeedback(PGconn *conn, XLogRecPtr blockpos, TimestampTz now, bool replyRequested)
     321             : {
     322             :     char        replybuf[1 + 8 + 8 + 8 + 8 + 1];
     323         128 :     int         len = 0;
     324             : 
     325         128 :     replybuf[len] = 'r';
     326         128 :     len += 1;
     327         128 :     fe_sendint64(blockpos, &replybuf[len]); /* write */
     328         128 :     len += 8;
     329         128 :     if (reportFlushPosition)
     330         126 :         fe_sendint64(lastFlushPosition, &replybuf[len]);    /* flush */
     331             :     else
     332           2 :         fe_sendint64(InvalidXLogRecPtr, &replybuf[len]);    /* flush */
     333         128 :     len += 8;
     334         128 :     fe_sendint64(InvalidXLogRecPtr, &replybuf[len]);    /* apply */
     335         128 :     len += 8;
     336         128 :     fe_sendint64(now, &replybuf[len]);  /* sendTime */
     337         128 :     len += 8;
     338         128 :     replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */
     339         128 :     len += 1;
     340             : 
     341         128 :     if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
     342             :     {
     343           0 :         pg_log_error("could not send feedback packet: %s",
     344             :                      PQerrorMessage(conn));
     345           0 :         return false;
     346             :     }
     347             : 
     348         128 :     return true;
     349             : }
     350             : 
     351             : /*
     352             :  * Check that the server version we're connected to is supported by
     353             :  * ReceiveXlogStream().
     354             :  *
     355             :  * If it's not, an error message is printed to stderr, and false is returned.
     356             :  */
     357             : bool
     358         274 : CheckServerVersionForStreaming(PGconn *conn)
     359             : {
     360             :     int         minServerMajor,
     361             :                 maxServerMajor;
     362             :     int         serverMajor;
     363             : 
     364             :     /*
     365             :      * The message format used in streaming replication changed in 9.3, so we
     366             :      * cannot stream from older servers. And we don't support servers newer
     367             :      * than the client; it might work, but we don't know, so err on the safe
     368             :      * side.
     369             :      */
     370         274 :     minServerMajor = 903;
     371         274 :     maxServerMajor = PG_VERSION_NUM / 100;
     372         274 :     serverMajor = PQserverVersion(conn) / 100;
     373         274 :     if (serverMajor < minServerMajor)
     374             :     {
     375           0 :         const char *serverver = PQparameterStatus(conn, "server_version");
     376             : 
     377           0 :         pg_log_error("incompatible server version %s; client does not support streaming from server versions older than %s",
     378             :                      serverver ? serverver : "'unknown'",
     379             :                      "9.3");
     380           0 :         return false;
     381             :     }
     382         274 :     else if (serverMajor > maxServerMajor)
     383             :     {
     384           0 :         const char *serverver = PQparameterStatus(conn, "server_version");
     385             : 
     386           0 :         pg_log_error("incompatible server version %s; client does not support streaming from server versions newer than %s",
     387             :                      serverver ? serverver : "'unknown'",
     388             :                      PG_VERSION);
     389           0 :         return false;
     390             :     }
     391         274 :     return true;
     392             : }
     393             : 
     394             : /*
     395             :  * Receive a log stream starting at the specified position.
     396             :  *
     397             :  * Individual parameters are passed through the StreamCtl structure.
     398             :  *
     399             :  * If sysidentifier is specified, validate that both the system
     400             :  * identifier and the timeline matches the specified ones
     401             :  * (by sending an extra IDENTIFY_SYSTEM command)
     402             :  *
     403             :  * All received segments will be written to the directory
     404             :  * specified by basedir. This will also fetch any missing timeline history
     405             :  * files.
     406             :  *
     407             :  * The stream_stop callback will be called every time data
     408             :  * is received, and whenever a segment is completed. If it returns
     409             :  * true, the streaming will stop and the function
     410             :  * return. As long as it returns false, streaming will continue
     411             :  * indefinitely.
     412             :  *
     413             :  * If stream_stop() checks for external input, stop_socket should be set to
     414             :  * the FD it checks.  This will allow such input to be detected promptly
     415             :  * rather than after standby_message_timeout (which might be indefinite).
     416             :  * Note that signals will interrupt waits for input as well, but that is
     417             :  * race-y since a signal received while busy won't interrupt the wait.
     418             :  *
     419             :  * standby_message_timeout controls how often we send a message
     420             :  * back to the master letting it know our progress, in milliseconds.
     421             :  * Zero means no messages are sent.
     422             :  * This message will only contain the write location, and never
     423             :  * flush or replay.
     424             :  *
     425             :  * If 'partial_suffix' is not NULL, files are initially created with the
     426             :  * given suffix, and the suffix is removed once the file is finished. That
     427             :  * allows you to tell the difference between partial and completed files,
     428             :  * so that you can continue later where you left.
     429             :  *
     430             :  * If 'synchronous' is true, the received WAL is flushed as soon as written,
     431             :  * otherwise only when the WAL file is closed.
     432             :  *
     433             :  * Note: The WAL location *must* be at a log segment start!
     434             :  */
     435             : bool
     436         130 : ReceiveXlogStream(PGconn *conn, StreamCtl *stream)
     437             : {
     438             :     char        query[128];
     439             :     char        slotcmd[128];
     440             :     PGresult   *res;
     441             :     XLogRecPtr  stoppos;
     442             : 
     443             :     /*
     444             :      * The caller should've checked the server version already, but doesn't do
     445             :      * any harm to check it here too.
     446             :      */
     447         130 :     if (!CheckServerVersionForStreaming(conn))
     448           0 :         return false;
     449             : 
     450             :     /*
     451             :      * Decide whether we want to report the flush position. If we report the
     452             :      * flush position, the primary will know what WAL we'll possibly
     453             :      * re-request, and it can then remove older WAL safely. We must always do
     454             :      * that when we are using slots.
     455             :      *
     456             :      * Reporting the flush position makes one eligible as a synchronous
     457             :      * replica. People shouldn't include generic names in
     458             :      * synchronous_standby_names, but we've protected them against it so far,
     459             :      * so let's continue to do so unless specifically requested.
     460             :      */
     461         130 :     if (stream->replication_slot != NULL)
     462             :     {
     463         126 :         reportFlushPosition = true;
     464         126 :         sprintf(slotcmd, "SLOT \"%s\" ", stream->replication_slot);
     465             :     }
     466             :     else
     467             :     {
     468           4 :         if (stream->synchronous)
     469           2 :             reportFlushPosition = true;
     470             :         else
     471           2 :             reportFlushPosition = false;
     472           4 :         slotcmd[0] = 0;
     473             :     }
     474             : 
     475         130 :     if (stream->sysidentifier != NULL)
     476             :     {
     477             :         /* Validate system identifier hasn't changed */
     478         128 :         res = PQexec(conn, "IDENTIFY_SYSTEM");
     479         128 :         if (PQresultStatus(res) != PGRES_TUPLES_OK)
     480             :         {
     481           0 :             pg_log_error("could not send replication command \"%s\": %s",
     482             :                          "IDENTIFY_SYSTEM", PQerrorMessage(conn));
     483           0 :             PQclear(res);
     484           0 :             return false;
     485             :         }
     486         128 :         if (PQntuples(res) != 1 || PQnfields(res) < 3)
     487             :         {
     488           0 :             pg_log_error("could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields",
     489             :                          PQntuples(res), PQnfields(res), 1, 3);
     490           0 :             PQclear(res);
     491           0 :             return false;
     492             :         }
     493         128 :         if (strcmp(stream->sysidentifier, PQgetvalue(res, 0, 0)) != 0)
     494             :         {
     495           0 :             pg_log_error("system identifier does not match between base backup and streaming connection");
     496           0 :             PQclear(res);
     497           0 :             return false;
     498             :         }
     499         128 :         if (stream->timeline > atoi(PQgetvalue(res, 0, 1)))
     500             :         {
     501           0 :             pg_log_error("starting timeline %u is not present in the server",
     502             :                          stream->timeline);
     503           0 :             PQclear(res);
     504           0 :             return false;
     505             :         }
     506         128 :         PQclear(res);
     507             :     }
     508             : 
     509             :     /*
     510             :      * initialize flush position to starting point, it's the caller's
     511             :      * responsibility that that's sane.
     512             :      */
     513         130 :     lastFlushPosition = stream->startpos;
     514             : 
     515             :     while (1)
     516             :     {
     517             :         /*
     518             :          * Fetch the timeline history file for this timeline, if we don't have
     519             :          * it already. When streaming log to tar, this will always return
     520             :          * false, as we are never streaming into an existing file and
     521             :          * therefore there can be no pre-existing timeline history file.
     522             :          */
     523         130 :         if (!existsTimeLineHistoryFile(stream))
     524             :         {
     525           0 :             snprintf(query, sizeof(query), "TIMELINE_HISTORY %u", stream->timeline);
     526           0 :             res = PQexec(conn, query);
     527           0 :             if (PQresultStatus(res) != PGRES_TUPLES_OK)
     528             :             {
     529             :                 /* FIXME: we might send it ok, but get an error */
     530           0 :                 pg_log_error("could not send replication command \"%s\": %s",
     531             :                              "TIMELINE_HISTORY", PQresultErrorMessage(res));
     532           0 :                 PQclear(res);
     533           0 :                 return false;
     534             :             }
     535             : 
     536             :             /*
     537             :              * The response to TIMELINE_HISTORY is a single row result set
     538             :              * with two fields: filename and content
     539             :              */
     540           0 :             if (PQnfields(res) != 2 || PQntuples(res) != 1)
     541             :             {
     542           0 :                 pg_log_warning("unexpected response to TIMELINE_HISTORY command: got %d rows and %d fields, expected %d rows and %d fields",
     543             :                                PQntuples(res), PQnfields(res), 1, 2);
     544             :             }
     545             : 
     546             :             /* Write the history file to disk */
     547           0 :             writeTimeLineHistoryFile(stream,
     548             :                                      PQgetvalue(res, 0, 0),
     549             :                                      PQgetvalue(res, 0, 1));
     550             : 
     551           0 :             PQclear(res);
     552             :         }
     553             : 
     554             :         /*
     555             :          * Before we start streaming from the requested location, check if the
     556             :          * callback tells us to stop here.
     557             :          */
     558         130 :         if (stream->stream_stop(stream->startpos, stream->timeline, false))
     559           0 :             return true;
     560             : 
     561             :         /* Initiate the replication stream at specified location */
     562         390 :         snprintf(query, sizeof(query), "START_REPLICATION %s%X/%X TIMELINE %u",
     563             :                  slotcmd,
     564         130 :                  (uint32) (stream->startpos >> 32), (uint32) stream->startpos,
     565             :                  stream->timeline);
     566         130 :         res = PQexec(conn, query);
     567         130 :         if (PQresultStatus(res) != PGRES_COPY_BOTH)
     568             :         {
     569           2 :             pg_log_error("could not send replication command \"%s\": %s",
     570             :                          "START_REPLICATION", PQresultErrorMessage(res));
     571           2 :             PQclear(res);
     572           2 :             return false;
     573             :         }
     574         128 :         PQclear(res);
     575             : 
     576             :         /* Stream the WAL */
     577         128 :         res = HandleCopyStream(conn, stream, &stoppos);
     578         128 :         if (res == NULL)
     579           0 :             goto error;
     580             : 
     581             :         /*
     582             :          * Streaming finished.
     583             :          *
     584             :          * There are two possible reasons for that: a controlled shutdown, or
     585             :          * we reached the end of the current timeline. In case of
     586             :          * end-of-timeline, the server sends a result set after Copy has
     587             :          * finished, containing information about the next timeline. Read
     588             :          * that, and restart streaming from the next timeline. In case of
     589             :          * controlled shutdown, stop here.
     590             :          */
     591         128 :         if (PQresultStatus(res) == PGRES_TUPLES_OK)
     592             :         {
     593             :             /*
     594             :              * End-of-timeline. Read the next timeline's ID and starting
     595             :              * position. Usually, the starting position will match the end of
     596             :              * the previous timeline, but there are corner cases like if the
     597             :              * server had sent us half of a WAL record, when it was promoted.
     598             :              * The new timeline will begin at the end of the last complete
     599             :              * record in that case, overlapping the partial WAL record on the
     600             :              * old timeline.
     601             :              */
     602             :             uint32      newtimeline;
     603             :             bool        parsed;
     604             : 
     605           0 :             parsed = ReadEndOfStreamingResult(res, &stream->startpos, &newtimeline);
     606           0 :             PQclear(res);
     607           0 :             if (!parsed)
     608           0 :                 goto error;
     609             : 
     610             :             /* Sanity check the values the server gave us */
     611           0 :             if (newtimeline <= stream->timeline)
     612             :             {
     613           0 :                 pg_log_error("server reported unexpected next timeline %u, following timeline %u",
     614             :                              newtimeline, stream->timeline);
     615           0 :                 goto error;
     616             :             }
     617           0 :             if (stream->startpos > stoppos)
     618             :             {
     619           0 :                 pg_log_error("server stopped streaming timeline %u at %X/%X, but reported next timeline %u to begin at %X/%X",
     620             :                              stream->timeline, (uint32) (stoppos >> 32), (uint32) stoppos,
     621             :                              newtimeline, (uint32) (stream->startpos >> 32), (uint32) stream->startpos);
     622           0 :                 goto error;
     623             :             }
     624             : 
     625             :             /* Read the final result, which should be CommandComplete. */
     626           0 :             res = PQgetResult(conn);
     627           0 :             if (PQresultStatus(res) != PGRES_COMMAND_OK)
     628             :             {
     629           0 :                 pg_log_error("unexpected termination of replication stream: %s",
     630             :                              PQresultErrorMessage(res));
     631           0 :                 PQclear(res);
     632           0 :                 goto error;
     633             :             }
     634           0 :             PQclear(res);
     635             : 
     636             :             /*
     637             :              * Loop back to start streaming from the new timeline. Always
     638             :              * start streaming at the beginning of a segment.
     639             :              */
     640           0 :             stream->timeline = newtimeline;
     641           0 :             stream->startpos = stream->startpos -
     642           0 :                 XLogSegmentOffset(stream->startpos, WalSegSz);
     643           0 :             continue;
     644             :         }
     645         128 :         else if (PQresultStatus(res) == PGRES_COMMAND_OK)
     646             :         {
     647         128 :             PQclear(res);
     648             : 
     649             :             /*
     650             :              * End of replication (ie. controlled shut down of the server).
     651             :              *
     652             :              * Check if the callback thinks it's OK to stop here. If not,
     653             :              * complain.
     654             :              */
     655         128 :             if (stream->stream_stop(stoppos, stream->timeline, false))
     656         128 :                 return true;
     657             :             else
     658             :             {
     659           0 :                 pg_log_error("replication stream was terminated before stop point");
     660           0 :                 goto error;
     661             :             }
     662             :         }
     663             :         else
     664             :         {
     665             :             /* Server returned an error. */
     666           0 :             pg_log_error("unexpected termination of replication stream: %s",
     667             :                          PQresultErrorMessage(res));
     668           0 :             PQclear(res);
     669           0 :             goto error;
     670             :         }
     671             :     }
     672             : 
     673           0 : error:
     674           0 :     if (walfile != NULL && stream->walmethod->close(walfile, CLOSE_NO_RENAME) != 0)
     675           0 :         pg_log_error("could not close file \"%s\": %s",
     676             :                      current_walfile_name, stream->walmethod->getlasterror());
     677           0 :     walfile = NULL;
     678           0 :     return false;
     679             : }
     680             : 
     681             : /*
     682             :  * Helper function to parse the result set returned by server after streaming
     683             :  * has finished. On failure, prints an error to stderr and returns false.
     684             :  */
     685             : static bool
     686           0 : ReadEndOfStreamingResult(PGresult *res, XLogRecPtr *startpos, uint32 *timeline)
     687             : {
     688             :     uint32      startpos_xlogid,
     689             :                 startpos_xrecoff;
     690             : 
     691             :     /*----------
     692             :      * The result set consists of one row and two columns, e.g:
     693             :      *
     694             :      *  next_tli | next_tli_startpos
     695             :      * ----------+-------------------
     696             :      *         4 | 0/9949AE0
     697             :      *
     698             :      * next_tli is the timeline ID of the next timeline after the one that
     699             :      * just finished streaming. next_tli_startpos is the WAL location where
     700             :      * the server switched to it.
     701             :      *----------
     702             :      */
     703           0 :     if (PQnfields(res) < 2 || PQntuples(res) != 1)
     704             :     {
     705           0 :         pg_log_error("unexpected result set after end-of-timeline: got %d rows and %d fields, expected %d rows and %d fields",
     706             :                      PQntuples(res), PQnfields(res), 1, 2);
     707           0 :         return false;
     708             :     }
     709             : 
     710           0 :     *timeline = atoi(PQgetvalue(res, 0, 0));
     711           0 :     if (sscanf(PQgetvalue(res, 0, 1), "%X/%X", &startpos_xlogid,
     712             :                &startpos_xrecoff) != 2)
     713             :     {
     714           0 :         pg_log_error("could not parse next timeline's starting point \"%s\"",
     715             :                      PQgetvalue(res, 0, 1));
     716           0 :         return false;
     717             :     }
     718           0 :     *startpos = ((uint64) startpos_xlogid << 32) | startpos_xrecoff;
     719             : 
     720           0 :     return true;
     721             : }
     722             : 
     723             : /*
     724             :  * The main loop of ReceiveXlogStream. Handles the COPY stream after
     725             :  * initiating streaming with the START_REPLICATION command.
     726             :  *
     727             :  * If the COPY ends (not necessarily successfully) due a message from the
     728             :  * server, returns a PGresult and sets *stoppos to the last byte written.
     729             :  * On any other sort of error, returns NULL.
     730             :  */
     731             : static PGresult *
     732         128 : HandleCopyStream(PGconn *conn, StreamCtl *stream,
     733             :                  XLogRecPtr *stoppos)
     734             : {
     735         128 :     char       *copybuf = NULL;
     736         128 :     TimestampTz last_status = -1;
     737         128 :     XLogRecPtr  blockpos = stream->startpos;
     738             : 
     739         128 :     still_sending = true;
     740             : 
     741             :     while (1)
     742        1686 :     {
     743             :         int         r;
     744             :         TimestampTz now;
     745             :         long        sleeptime;
     746             : 
     747             :         /*
     748             :          * Check if we should continue streaming, or abort at this point.
     749             :          */
     750        1814 :         if (!CheckCopyStreamStop(conn, stream, blockpos, stoppos))
     751           0 :             goto error;
     752             : 
     753        1814 :         now = feGetCurrentTimestamp();
     754             : 
     755             :         /*
     756             :          * If synchronous option is true, issue sync command as soon as there
     757             :          * are WAL data which has not been flushed yet.
     758             :          */
     759        1814 :         if (stream->synchronous && lastFlushPosition < blockpos && walfile != NULL)
     760             :         {
     761           0 :             if (stream->walmethod->sync(walfile) != 0)
     762             :             {
     763           0 :                 pg_log_fatal("could not fsync file \"%s\": %s",
     764             :                              current_walfile_name, stream->walmethod->getlasterror());
     765           0 :                 exit(1);
     766             :             }
     767           0 :             lastFlushPosition = blockpos;
     768             : 
     769             :             /*
     770             :              * Send feedback so that the server sees the latest WAL locations
     771             :              * immediately.
     772             :              */
     773           0 :             if (!sendFeedback(conn, blockpos, now, false))
     774           0 :                 goto error;
     775           0 :             last_status = now;
     776             :         }
     777             : 
     778             :         /*
     779             :          * Potentially send a status message to the master
     780             :          */
     781        3542 :         if (still_sending && stream->standby_message_timeout > 0 &&
     782        1728 :             feTimestampDifferenceExceeds(last_status, now,
     783             :                                          stream->standby_message_timeout))
     784             :         {
     785             :             /* Time to send feedback! */
     786         128 :             if (!sendFeedback(conn, blockpos, now, false))
     787           0 :                 goto error;
     788         128 :             last_status = now;
     789             :         }
     790             : 
     791             :         /*
     792             :          * Calculate how long send/receive loops should sleep
     793             :          */
     794        1814 :         sleeptime = CalculateCopyStreamSleeptime(now, stream->standby_message_timeout,
     795             :                                                  last_status);
     796             : 
     797        1814 :         r = CopyStreamReceive(conn, sleeptime, stream->stop_socket, &copybuf);
     798        5490 :         while (r != 0)
     799             :         {
     800        3804 :             if (r == -1)
     801           0 :                 goto error;
     802        3804 :             if (r == -2)
     803             :             {
     804         128 :                 PGresult   *res = HandleEndOfCopyStream(conn, stream, copybuf, blockpos, stoppos);
     805             : 
     806         128 :                 if (res == NULL)
     807           0 :                     goto error;
     808             :                 else
     809         128 :                     return res;
     810             :             }
     811             : 
     812             :             /* Check the message type. */
     813        3676 :             if (copybuf[0] == 'k')
     814             :             {
     815           0 :                 if (!ProcessKeepaliveMsg(conn, stream, copybuf, r, blockpos,
     816             :                                          &last_status))
     817           0 :                     goto error;
     818             :             }
     819        3676 :             else if (copybuf[0] == 'w')
     820             :             {
     821        3676 :                 if (!ProcessXLogDataMsg(conn, stream, copybuf, r, &blockpos))
     822           0 :                     goto error;
     823             : 
     824             :                 /*
     825             :                  * Check if we should continue streaming, or abort at this
     826             :                  * point.
     827             :                  */
     828        3676 :                 if (!CheckCopyStreamStop(conn, stream, blockpos, stoppos))
     829           0 :                     goto error;
     830             :             }
     831             :             else
     832             :             {
     833           0 :                 pg_log_error("unrecognized streaming header: \"%c\"",
     834             :                              copybuf[0]);
     835           0 :                 goto error;
     836             :             }
     837             : 
     838             :             /*
     839             :              * Process the received data, and any subsequent data we can read
     840             :              * without blocking.
     841             :              */
     842        3676 :             r = CopyStreamReceive(conn, 0, stream->stop_socket, &copybuf);
     843             :         }
     844             :     }
     845             : 
     846           0 : error:
     847           0 :     if (copybuf != NULL)
     848           0 :         PQfreemem(copybuf);
     849           0 :     return NULL;
     850             : }
     851             : 
     852             : /*
     853             :  * Wait until we can read a CopyData message,
     854             :  * or timeout, or occurrence of a signal or input on the stop_socket.
     855             :  * (timeout_ms < 0 means wait indefinitely; 0 means don't wait.)
     856             :  *
     857             :  * Returns 1 if data has become available for reading, 0 if timed out
     858             :  * or interrupted by signal or stop_socket input, and -1 on an error.
     859             :  */
     860             : static int
     861        5302 : CopyStreamPoll(PGconn *conn, long timeout_ms, pgsocket stop_socket)
     862             : {
     863             :     int         ret;
     864             :     fd_set      input_mask;
     865             :     int         connsocket;
     866             :     int         maxfd;
     867             :     struct timeval timeout;
     868             :     struct timeval *timeoutptr;
     869             : 
     870        5302 :     connsocket = PQsocket(conn);
     871        5302 :     if (connsocket < 0)
     872             :     {
     873           0 :         pg_log_error("invalid socket: %s", PQerrorMessage(conn));
     874           0 :         return -1;
     875             :     }
     876             : 
     877        5302 :     FD_ZERO(&input_mask);
     878        5302 :     FD_SET(connsocket, &input_mask);
     879        5302 :     maxfd = connsocket;
     880        5302 :     if (stop_socket != PGINVALID_SOCKET)
     881             :     {
     882        5296 :         FD_SET(stop_socket, &input_mask);
     883        5296 :         maxfd = Max(maxfd, stop_socket);
     884             :     }
     885             : 
     886        5302 :     if (timeout_ms < 0)
     887          86 :         timeoutptr = NULL;
     888             :     else
     889             :     {
     890        5216 :         timeout.tv_sec = timeout_ms / 1000L;
     891        5216 :         timeout.tv_usec = (timeout_ms % 1000L) * 1000L;
     892        5216 :         timeoutptr = &timeout;
     893             :     }
     894             : 
     895        5302 :     ret = select(maxfd + 1, &input_mask, NULL, NULL, timeoutptr);
     896             : 
     897        5302 :     if (ret < 0)
     898             :     {
     899           0 :         if (errno == EINTR)
     900           0 :             return 0;           /* Got a signal, so not an error */
     901           0 :         pg_log_error("select() failed: %m");
     902           0 :         return -1;
     903             :     }
     904        5302 :     if (ret > 0 && FD_ISSET(connsocket, &input_mask))
     905        3784 :         return 1;               /* Got input on connection socket */
     906             : 
     907        1518 :     return 0;                   /* Got timeout or input on stop_socket */
     908             : }
     909             : 
     910             : /*
     911             :  * Receive CopyData message available from XLOG stream, blocking for
     912             :  * maximum of 'timeout' ms.
     913             :  *
     914             :  * If data was received, returns the length of the data. *buffer is set to
     915             :  * point to a buffer holding the received message. The buffer is only valid
     916             :  * until the next CopyStreamReceive call.
     917             :  *
     918             :  * Returns 0 if no data was available within timeout, or if wait was
     919             :  * interrupted by signal or stop_socket input.
     920             :  * -1 on error. -2 if the server ended the COPY.
     921             :  */
     922             : static int
     923        5490 : CopyStreamReceive(PGconn *conn, long timeout, pgsocket stop_socket,
     924             :                   char **buffer)
     925             : {
     926        5490 :     char       *copybuf = NULL;
     927             :     int         rawlen;
     928             : 
     929        5490 :     if (*buffer != NULL)
     930        3676 :         PQfreemem(*buffer);
     931        5490 :     *buffer = NULL;
     932             : 
     933             :     /* Try to receive a CopyData message */
     934        5490 :     rawlen = PQgetCopyData(conn, &copybuf, 1);
     935        5490 :     if (rawlen == 0)
     936             :     {
     937             :         int         ret;
     938             : 
     939             :         /*
     940             :          * No data available.  Wait for some to appear, but not longer than
     941             :          * the specified timeout, so that we can ping the server.  Also stop
     942             :          * waiting if input appears on stop_socket.
     943             :          */
     944        5302 :         ret = CopyStreamPoll(conn, timeout, stop_socket);
     945        5302 :         if (ret <= 0)
     946        1518 :             return ret;
     947             : 
     948             :         /* Now there is actually data on the socket */
     949        3784 :         if (PQconsumeInput(conn) == 0)
     950             :         {
     951           0 :             pg_log_error("could not receive data from WAL stream: %s",
     952             :                          PQerrorMessage(conn));
     953           0 :             return -1;
     954             :         }
     955             : 
     956             :         /* Now that we've consumed some input, try again */
     957        3784 :         rawlen = PQgetCopyData(conn, &copybuf, 1);
     958        3784 :         if (rawlen == 0)
     959         168 :             return 0;
     960             :     }
     961        3804 :     if (rawlen == -1)           /* end-of-streaming or error */
     962         128 :         return -2;
     963        3676 :     if (rawlen == -2)
     964             :     {
     965           0 :         pg_log_error("could not read COPY data: %s", PQerrorMessage(conn));
     966           0 :         return -1;
     967             :     }
     968             : 
     969             :     /* Return received messages to caller */
     970        3676 :     *buffer = copybuf;
     971        3676 :     return rawlen;
     972             : }
     973             : 
     974             : /*
     975             :  * Process the keepalive message.
     976             :  */
     977             : static bool
     978           0 : ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
     979             :                     XLogRecPtr blockpos, TimestampTz *last_status)
     980             : {
     981             :     int         pos;
     982             :     bool        replyRequested;
     983             :     TimestampTz now;
     984             : 
     985             :     /*
     986             :      * Parse the keepalive message, enclosed in the CopyData message. We just
     987             :      * check if the server requested a reply, and ignore the rest.
     988             :      */
     989           0 :     pos = 1;                    /* skip msgtype 'k' */
     990           0 :     pos += 8;                   /* skip walEnd */
     991           0 :     pos += 8;                   /* skip sendTime */
     992             : 
     993           0 :     if (len < pos + 1)
     994             :     {
     995           0 :         pg_log_error("streaming header too small: %d", len);
     996           0 :         return false;
     997             :     }
     998           0 :     replyRequested = copybuf[pos];
     999             : 
    1000             :     /* If the server requested an immediate reply, send one. */
    1001           0 :     if (replyRequested && still_sending)
    1002             :     {
    1003           0 :         if (reportFlushPosition && lastFlushPosition < blockpos &&
    1004           0 :             walfile != NULL)
    1005             :         {
    1006             :             /*
    1007             :              * If a valid flush location needs to be reported, flush the
    1008             :              * current WAL file so that the latest flush location is sent back
    1009             :              * to the server. This is necessary to see whether the last WAL
    1010             :              * data has been successfully replicated or not, at the normal
    1011             :              * shutdown of the server.
    1012             :              */
    1013           0 :             if (stream->walmethod->sync(walfile) != 0)
    1014             :             {
    1015           0 :                 pg_log_fatal("could not fsync file \"%s\": %s",
    1016             :                              current_walfile_name, stream->walmethod->getlasterror());
    1017           0 :                 exit(1);
    1018             :             }
    1019           0 :             lastFlushPosition = blockpos;
    1020             :         }
    1021             : 
    1022           0 :         now = feGetCurrentTimestamp();
    1023           0 :         if (!sendFeedback(conn, blockpos, now, false))
    1024           0 :             return false;
    1025           0 :         *last_status = now;
    1026             :     }
    1027             : 
    1028           0 :     return true;
    1029             : }
    1030             : 
    1031             : /*
    1032             :  * Process XLogData message.
    1033             :  */
    1034             : static bool
    1035        3676 : ProcessXLogDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
    1036             :                    XLogRecPtr *blockpos)
    1037             : {
    1038             :     int         xlogoff;
    1039             :     int         bytes_left;
    1040             :     int         bytes_written;
    1041             :     int         hdr_len;
    1042             : 
    1043             :     /*
    1044             :      * Once we've decided we don't want to receive any more, just ignore any
    1045             :      * subsequent XLogData messages.
    1046             :      */
    1047        3676 :     if (!(still_sending))
    1048         200 :         return true;
    1049             : 
    1050             :     /*
    1051             :      * Read the header of the XLogData message, enclosed in the CopyData
    1052             :      * message. We only need the WAL location field (dataStart), the rest of
    1053             :      * the header is ignored.
    1054             :      */
    1055        3476 :     hdr_len = 1;                /* msgtype 'w' */
    1056        3476 :     hdr_len += 8;               /* dataStart */
    1057        3476 :     hdr_len += 8;               /* walEnd */
    1058        3476 :     hdr_len += 8;               /* sendTime */
    1059        3476 :     if (len < hdr_len)
    1060             :     {
    1061           0 :         pg_log_error("streaming header too small: %d", len);
    1062           0 :         return false;
    1063             :     }
    1064        3476 :     *blockpos = fe_recvint64(&copybuf[1]);
    1065             : 
    1066             :     /* Extract WAL location for this block */
    1067        3476 :     xlogoff = XLogSegmentOffset(*blockpos, WalSegSz);
    1068             : 
    1069             :     /*
    1070             :      * Verify that the initial location in the stream matches where we think
    1071             :      * we are.
    1072             :      */
    1073        3476 :     if (walfile == NULL)
    1074             :     {
    1075             :         /* No file open yet */
    1076         128 :         if (xlogoff != 0)
    1077             :         {
    1078           0 :             pg_log_error("received write-ahead log record for offset %u with no file open",
    1079             :                          xlogoff);
    1080           0 :             return false;
    1081             :         }
    1082             :     }
    1083             :     else
    1084             :     {
    1085             :         /* More data in existing segment */
    1086        3348 :         if (stream->walmethod->get_current_pos(walfile) != xlogoff)
    1087             :         {
    1088           0 :             pg_log_error("got WAL data offset %08x, expected %08x",
    1089             :                          xlogoff, (int) stream->walmethod->get_current_pos(walfile));
    1090           0 :             return false;
    1091             :         }
    1092             :     }
    1093             : 
    1094        3476 :     bytes_left = len - hdr_len;
    1095        3476 :     bytes_written = 0;
    1096             : 
    1097        6950 :     while (bytes_left)
    1098             :     {
    1099             :         int         bytes_to_write;
    1100             : 
    1101             :         /*
    1102             :          * If crossing a WAL boundary, only write up until we reach wal
    1103             :          * segment size.
    1104             :          */
    1105        3476 :         if (xlogoff + bytes_left > WalSegSz)
    1106           0 :             bytes_to_write = WalSegSz - xlogoff;
    1107             :         else
    1108        3476 :             bytes_to_write = bytes_left;
    1109             : 
    1110        3476 :         if (walfile == NULL)
    1111             :         {
    1112         128 :             if (!open_walfile(stream, *blockpos))
    1113             :             {
    1114             :                 /* Error logged by open_walfile */
    1115           0 :                 return false;
    1116             :             }
    1117             :         }
    1118             : 
    1119        3476 :         if (stream->walmethod->write(walfile, copybuf + hdr_len + bytes_written,
    1120        3476 :                                      bytes_to_write) != bytes_to_write)
    1121             :         {
    1122           0 :             pg_log_error("could not write %u bytes to WAL file \"%s\": %s",
    1123             :                          bytes_to_write, current_walfile_name,
    1124             :                          stream->walmethod->getlasterror());
    1125           0 :             return false;
    1126             :         }
    1127             : 
    1128             :         /* Write was successful, advance our position */
    1129        3476 :         bytes_written += bytes_to_write;
    1130        3476 :         bytes_left -= bytes_to_write;
    1131        3476 :         *blockpos += bytes_to_write;
    1132        3476 :         xlogoff += bytes_to_write;
    1133             : 
    1134             :         /* Did we reach the end of a WAL segment? */
    1135        3476 :         if (XLogSegmentOffset(*blockpos, WalSegSz) == 0)
    1136             :         {
    1137          10 :             if (!close_walfile(stream, *blockpos))
    1138             :                 /* Error message written in close_walfile() */
    1139           0 :                 return false;
    1140             : 
    1141          10 :             xlogoff = 0;
    1142             : 
    1143          10 :             if (still_sending && stream->stream_stop(*blockpos, stream->timeline, true))
    1144             :             {
    1145           2 :                 if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
    1146             :                 {
    1147           0 :                     pg_log_error("could not send copy-end packet: %s",
    1148             :                                  PQerrorMessage(conn));
    1149           0 :                     return false;
    1150             :                 }
    1151           2 :                 still_sending = false;
    1152           2 :                 return true;    /* ignore the rest of this XLogData packet */
    1153             :             }
    1154             :         }
    1155             :     }
    1156             :     /* No more data left to write, receive next copy packet */
    1157             : 
    1158        3474 :     return true;
    1159             : }
    1160             : 
    1161             : /*
    1162             :  * Handle end of the copy stream.
    1163             :  */
    1164             : static PGresult *
    1165         128 : HandleEndOfCopyStream(PGconn *conn, StreamCtl *stream, char *copybuf,
    1166             :                       XLogRecPtr blockpos, XLogRecPtr *stoppos)
    1167             : {
    1168         128 :     PGresult   *res = PQgetResult(conn);
    1169             : 
    1170             :     /*
    1171             :      * The server closed its end of the copy stream.  If we haven't closed
    1172             :      * ours already, we need to do so now, unless the server threw an error,
    1173             :      * in which case we don't.
    1174             :      */
    1175         128 :     if (still_sending)
    1176             :     {
    1177           0 :         if (!close_walfile(stream, blockpos))
    1178             :         {
    1179             :             /* Error message written in close_walfile() */
    1180           0 :             PQclear(res);
    1181           0 :             return NULL;
    1182             :         }
    1183           0 :         if (PQresultStatus(res) == PGRES_COPY_IN)
    1184             :         {
    1185           0 :             if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
    1186             :             {
    1187           0 :                 pg_log_error("could not send copy-end packet: %s",
    1188             :                              PQerrorMessage(conn));
    1189           0 :                 PQclear(res);
    1190           0 :                 return NULL;
    1191             :             }
    1192           0 :             res = PQgetResult(conn);
    1193             :         }
    1194           0 :         still_sending = false;
    1195             :     }
    1196         128 :     if (copybuf != NULL)
    1197           0 :         PQfreemem(copybuf);
    1198         128 :     *stoppos = blockpos;
    1199         128 :     return res;
    1200             : }
    1201             : 
    1202             : /*
    1203             :  * Check if we should continue streaming, or abort at this point.
    1204             :  */
    1205             : static bool
    1206        5490 : CheckCopyStreamStop(PGconn *conn, StreamCtl *stream, XLogRecPtr blockpos,
    1207             :                     XLogRecPtr *stoppos)
    1208             : {
    1209        5490 :     if (still_sending && stream->stream_stop(blockpos, stream->timeline, false))
    1210             :     {
    1211         126 :         if (!close_walfile(stream, blockpos))
    1212             :         {
    1213             :             /* Potential error message is written by close_walfile */
    1214           0 :             return false;
    1215             :         }
    1216         126 :         if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
    1217             :         {
    1218           0 :             pg_log_error("could not send copy-end packet: %s",
    1219             :                          PQerrorMessage(conn));
    1220           0 :             return false;
    1221             :         }
    1222         126 :         still_sending = false;
    1223             :     }
    1224             : 
    1225        5490 :     return true;
    1226             : }
    1227             : 
    1228             : /*
    1229             :  * Calculate how long send/receive loops should sleep
    1230             :  */
    1231             : static long
    1232        1814 : CalculateCopyStreamSleeptime(TimestampTz now, int standby_message_timeout,
    1233             :                              TimestampTz last_status)
    1234             : {
    1235        1814 :     TimestampTz status_targettime = 0;
    1236             :     long        sleeptime;
    1237             : 
    1238        1814 :     if (standby_message_timeout && still_sending)
    1239        1728 :         status_targettime = last_status +
    1240        1728 :             (standby_message_timeout - 1) * ((int64) 1000);
    1241             : 
    1242        1814 :     if (status_targettime > 0)
    1243             :     {
    1244             :         long        secs;
    1245             :         int         usecs;
    1246             : 
    1247        1728 :         feTimestampDifference(now,
    1248             :                               status_targettime,
    1249             :                               &secs,
    1250             :                               &usecs);
    1251             :         /* Always sleep at least 1 sec */
    1252        1728 :         if (secs <= 0)
    1253             :         {
    1254           0 :             secs = 1;
    1255           0 :             usecs = 0;
    1256             :         }
    1257             : 
    1258        1728 :         sleeptime = secs * 1000 + usecs / 1000;
    1259             :     }
    1260             :     else
    1261          86 :         sleeptime = -1;
    1262             : 
    1263        1814 :     return sleeptime;
    1264             : }

Generated by: LCOV version 1.13