LCOV - code coverage report
Current view: top level - src/bin/pg_basebackup - pg_recvlogical.c (source / functions) Hit Total Coverage
Test: PostgreSQL 18beta1 Lines: 351 485 72.4 %
Date: 2025-06-06 14:17:25 Functions: 9 10 90.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * pg_recvlogical.c - receive data from a logical decoding slot in a streaming
       4             :  *                    fashion and write it to a local file.
       5             :  *
       6             :  * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
       7             :  *
       8             :  * IDENTIFICATION
       9             :  *        src/bin/pg_basebackup/pg_recvlogical.c
      10             :  *-------------------------------------------------------------------------
      11             :  */
      12             : 
      13             : #include "postgres_fe.h"
      14             : 
      15             : #include <dirent.h>
      16             : #include <limits.h>
      17             : #include <sys/select.h>
      18             : #include <sys/stat.h>
      19             : #include <unistd.h>
      20             : 
      21             : #include "common/file_perm.h"
      22             : #include "common/logging.h"
      23             : #include "fe_utils/option_utils.h"
      24             : #include "getopt_long.h"
      25             : #include "libpq-fe.h"
      26             : #include "libpq/pqsignal.h"
      27             : #include "pqexpbuffer.h"
      28             : #include "streamutil.h"
      29             : 
      30             : /* Time to sleep between reconnection attempts */
      31             : #define RECONNECT_SLEEP_TIME 5
      32             : 
      33             : typedef enum
      34             : {
      35             :     STREAM_STOP_NONE,
      36             :     STREAM_STOP_END_OF_WAL,
      37             :     STREAM_STOP_KEEPALIVE,
      38             :     STREAM_STOP_SIGNAL
      39             : } StreamStopReason;
      40             : 
      41             : /* Global Options */
      42             : static char *outfile = NULL;
      43             : static int  verbose = 0;
      44             : static bool two_phase = false;
      45             : static bool failover = false;
      46             : static int  noloop = 0;
      47             : static int  standby_message_timeout = 10 * 1000;    /* 10 sec = default */
      48             : static int  fsync_interval = 10 * 1000; /* 10 sec = default */
      49             : static XLogRecPtr startpos = InvalidXLogRecPtr;
      50             : static XLogRecPtr endpos = InvalidXLogRecPtr;
      51             : static bool do_create_slot = false;
      52             : static bool slot_exists_ok = false;
      53             : static bool do_start_slot = false;
      54             : static bool do_drop_slot = false;
      55             : static char *replication_slot = NULL;
      56             : 
      57             : /* filled pairwise with option, value. value may be NULL */
      58             : static char **options;
      59             : static size_t noptions = 0;
      60             : static const char *plugin = "test_decoding";
      61             : 
      62             : /* Global State */
      63             : static int  outfd = -1;
      64             : static volatile sig_atomic_t time_to_abort = false;
      65             : static volatile sig_atomic_t stop_reason = STREAM_STOP_NONE;
      66             : static volatile sig_atomic_t output_reopen = false;
      67             : static bool output_isfile;
      68             : static TimestampTz output_last_fsync = -1;
      69             : static bool output_needs_fsync = false;
      70             : static XLogRecPtr output_written_lsn = InvalidXLogRecPtr;
      71             : static XLogRecPtr output_fsync_lsn = InvalidXLogRecPtr;
      72             : 
      73             : static void usage(void);
      74             : static void StreamLogicalLog(void);
      75             : static bool flushAndSendFeedback(PGconn *conn, TimestampTz *now);
      76             : static void prepareToTerminate(PGconn *conn, XLogRecPtr endpos,
      77             :                                StreamStopReason reason,
      78             :                                XLogRecPtr lsn);
      79             : 
      80             : static void
      81           2 : usage(void)
      82             : {
      83           2 :     printf(_("%s controls PostgreSQL logical decoding streams.\n\n"),
      84             :            progname);
      85           2 :     printf(_("Usage:\n"));
      86           2 :     printf(_("  %s [OPTION]...\n"), progname);
      87           2 :     printf(_("\nAction to be performed:\n"));
      88           2 :     printf(_("      --create-slot      create a new replication slot (for the slot's name see --slot)\n"));
      89           2 :     printf(_("      --drop-slot        drop the replication slot (for the slot's name see --slot)\n"));
      90           2 :     printf(_("      --start            start streaming in a replication slot (for the slot's name see --slot)\n"));
      91           2 :     printf(_("\nOptions:\n"));
      92           2 :     printf(_("  -E, --endpos=LSN       exit after receiving the specified LSN\n"));
      93           2 :     printf(_("      --failover         enable replication slot synchronization to standby servers when\n"
      94             :              "                         creating a slot\n"));
      95           2 :     printf(_("  -f, --file=FILE        receive log into this file, - for stdout\n"));
      96           2 :     printf(_("  -F  --fsync-interval=SECS\n"
      97             :              "                         time between fsyncs to the output file (default: %d)\n"), (fsync_interval / 1000));
      98           2 :     printf(_("      --if-not-exists    do not error if slot already exists when creating a slot\n"));
      99           2 :     printf(_("  -I, --startpos=LSN     where in an existing slot should the streaming start\n"));
     100           2 :     printf(_("  -n, --no-loop          do not loop on connection lost\n"));
     101           2 :     printf(_("  -o, --option=NAME[=VALUE]\n"
     102             :              "                         pass option NAME with optional value VALUE to the\n"
     103             :              "                         output plugin\n"));
     104           2 :     printf(_("  -P, --plugin=PLUGIN    use output plugin PLUGIN (default: %s)\n"), plugin);
     105           2 :     printf(_("  -s, --status-interval=SECS\n"
     106             :              "                         time between status packets sent to server (default: %d)\n"), (standby_message_timeout / 1000));
     107           2 :     printf(_("  -S, --slot=SLOTNAME    name of the logical replication slot\n"));
     108           2 :     printf(_("  -t, --two-phase        enable decoding of prepared transactions when creating a slot\n"));
     109           2 :     printf(_("  -v, --verbose          output verbose messages\n"));
     110           2 :     printf(_("  -V, --version          output version information, then exit\n"));
     111           2 :     printf(_("  -?, --help             show this help, then exit\n"));
     112           2 :     printf(_("\nConnection options:\n"));
     113           2 :     printf(_("  -d, --dbname=DBNAME    database to connect to\n"));
     114           2 :     printf(_("  -h, --host=HOSTNAME    database server host or socket directory\n"));
     115           2 :     printf(_("  -p, --port=PORT        database server port number\n"));
     116           2 :     printf(_("  -U, --username=NAME    connect as specified database user\n"));
     117           2 :     printf(_("  -w, --no-password      never prompt for password\n"));
     118           2 :     printf(_("  -W, --password         force password prompt (should happen automatically)\n"));
     119           2 :     printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
     120           2 :     printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
     121           2 : }
     122             : 
     123             : /*
     124             :  * Send a Standby Status Update message to server.
     125             :  */
     126             : static bool
     127          50 : sendFeedback(PGconn *conn, TimestampTz now, bool force, bool replyRequested)
     128             : {
     129             :     static XLogRecPtr last_written_lsn = InvalidXLogRecPtr;
     130             :     static XLogRecPtr last_fsync_lsn = InvalidXLogRecPtr;
     131             : 
     132             :     char        replybuf[1 + 8 + 8 + 8 + 8 + 1];
     133          50 :     int         len = 0;
     134             : 
     135             :     /*
     136             :      * we normally don't want to send superfluous feedback, but if it's
     137             :      * because of a timeout we need to, otherwise wal_sender_timeout will kill
     138             :      * us.
     139             :      */
     140          50 :     if (!force &&
     141           0 :         last_written_lsn == output_written_lsn &&
     142           0 :         last_fsync_lsn == output_fsync_lsn)
     143           0 :         return true;
     144             : 
     145          50 :     if (verbose)
     146           0 :         pg_log_info("confirming write up to %X/%X, flush to %X/%X (slot %s)",
     147             :                     LSN_FORMAT_ARGS(output_written_lsn),
     148             :                     LSN_FORMAT_ARGS(output_fsync_lsn),
     149             :                     replication_slot);
     150             : 
     151          50 :     replybuf[len] = 'r';
     152          50 :     len += 1;
     153          50 :     fe_sendint64(output_written_lsn, &replybuf[len]);   /* write */
     154          50 :     len += 8;
     155          50 :     fe_sendint64(output_fsync_lsn, &replybuf[len]); /* flush */
     156          50 :     len += 8;
     157          50 :     fe_sendint64(InvalidXLogRecPtr, &replybuf[len]);    /* apply */
     158          50 :     len += 8;
     159          50 :     fe_sendint64(now, &replybuf[len]);  /* sendTime */
     160          50 :     len += 8;
     161          50 :     replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */
     162          50 :     len += 1;
     163             : 
     164          50 :     startpos = output_written_lsn;
     165          50 :     last_written_lsn = output_written_lsn;
     166          50 :     last_fsync_lsn = output_fsync_lsn;
     167             : 
     168          50 :     if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
     169             :     {
     170           0 :         pg_log_error("could not send feedback packet: %s",
     171             :                      PQerrorMessage(conn));
     172           0 :         return false;
     173             :     }
     174             : 
     175          50 :     return true;
     176             : }
     177             : 
     178             : static void
     179         100 : disconnect_atexit(void)
     180             : {
     181         100 :     if (conn != NULL)
     182          54 :         PQfinish(conn);
     183         100 : }
     184             : 
     185             : static bool
     186          50 : OutputFsync(TimestampTz now)
     187             : {
     188          50 :     output_last_fsync = now;
     189             : 
     190          50 :     output_fsync_lsn = output_written_lsn;
     191             : 
     192          50 :     if (fsync_interval <= 0)
     193           0 :         return true;
     194             : 
     195          50 :     if (!output_needs_fsync)
     196          38 :         return true;
     197             : 
     198          12 :     output_needs_fsync = false;
     199             : 
     200             :     /* can only fsync if it's a regular file */
     201          12 :     if (!output_isfile)
     202           8 :         return true;
     203             : 
     204           4 :     if (fsync(outfd) != 0)
     205           0 :         pg_fatal("could not fsync file \"%s\": %m", outfile);
     206             : 
     207           4 :     return true;
     208             : }
     209             : 
     210             : /*
     211             :  * Start the log streaming
     212             :  */
     213             : static void
     214          46 : StreamLogicalLog(void)
     215             : {
     216             :     PGresult   *res;
     217          46 :     char       *copybuf = NULL;
     218          46 :     TimestampTz last_status = -1;
     219             :     int         i;
     220             :     PQExpBuffer query;
     221             :     XLogRecPtr  cur_record_lsn;
     222             : 
     223          46 :     output_written_lsn = InvalidXLogRecPtr;
     224          46 :     output_fsync_lsn = InvalidXLogRecPtr;
     225          46 :     cur_record_lsn = InvalidXLogRecPtr;
     226             : 
     227             :     /*
     228             :      * Connect in replication mode to the server
     229             :      */
     230          46 :     if (!conn)
     231           0 :         conn = GetConnection();
     232          46 :     if (!conn)
     233             :         /* Error message already written in GetConnection() */
     234           0 :         return;
     235             : 
     236             :     /*
     237             :      * Start the replication
     238             :      */
     239          46 :     if (verbose)
     240           0 :         pg_log_info("starting log streaming at %X/%X (slot %s)",
     241             :                     LSN_FORMAT_ARGS(startpos),
     242             :                     replication_slot);
     243             : 
     244             :     /* Initiate the replication stream at specified location */
     245          46 :     query = createPQExpBuffer();
     246          46 :     appendPQExpBuffer(query, "START_REPLICATION SLOT \"%s\" LOGICAL %X/%X",
     247          46 :                       replication_slot, LSN_FORMAT_ARGS(startpos));
     248             : 
     249             :     /* print options if there are any */
     250          46 :     if (noptions)
     251          40 :         appendPQExpBufferStr(query, " (");
     252             : 
     253         126 :     for (i = 0; i < noptions; i++)
     254             :     {
     255             :         /* separator */
     256          80 :         if (i > 0)
     257          40 :             appendPQExpBufferStr(query, ", ");
     258             : 
     259             :         /* write option name */
     260          80 :         appendPQExpBuffer(query, "\"%s\"", options[(i * 2)]);
     261             : 
     262             :         /* write option value if specified */
     263          80 :         if (options[(i * 2) + 1] != NULL)
     264          80 :             appendPQExpBuffer(query, " '%s'", options[(i * 2) + 1]);
     265             :     }
     266             : 
     267          46 :     if (noptions)
     268          40 :         appendPQExpBufferChar(query, ')');
     269             : 
     270          46 :     res = PQexec(conn, query->data);
     271          46 :     if (PQresultStatus(res) != PGRES_COPY_BOTH)
     272             :     {
     273          12 :         pg_log_error("could not send replication command \"%s\": %s",
     274             :                      query->data, PQresultErrorMessage(res));
     275          12 :         PQclear(res);
     276          12 :         goto error;
     277             :     }
     278          34 :     PQclear(res);
     279          34 :     resetPQExpBuffer(query);
     280             : 
     281          34 :     if (verbose)
     282           0 :         pg_log_info("streaming initiated");
     283             : 
     284         944 :     while (!time_to_abort)
     285             :     {
     286             :         int         r;
     287             :         int         bytes_left;
     288             :         int         bytes_written;
     289             :         TimestampTz now;
     290             :         int         hdr_len;
     291             : 
     292         942 :         cur_record_lsn = InvalidXLogRecPtr;
     293             : 
     294         942 :         if (copybuf != NULL)
     295             :         {
     296         636 :             PQfreemem(copybuf);
     297         636 :             copybuf = NULL;
     298             :         }
     299             : 
     300             :         /*
     301             :          * Potentially send a status message to the primary.
     302             :          */
     303         942 :         now = feGetCurrentTimestamp();
     304             : 
     305        1850 :         if (outfd != -1 &&
     306         908 :             feTimestampDifferenceExceeds(output_last_fsync, now,
     307             :                                          fsync_interval))
     308             :         {
     309          34 :             if (!OutputFsync(now))
     310           4 :                 goto error;
     311             :         }
     312             : 
     313        1884 :         if (standby_message_timeout > 0 &&
     314         942 :             feTimestampDifferenceExceeds(last_status, now,
     315             :                                          standby_message_timeout))
     316             :         {
     317             :             /* Time to send feedback! */
     318          34 :             if (!sendFeedback(conn, now, true, false))
     319           0 :                 goto error;
     320             : 
     321          34 :             last_status = now;
     322             :         }
     323             : 
     324             :         /* got SIGHUP, close output file */
     325         942 :         if (outfd != -1 && output_reopen && strcmp(outfile, "-") != 0)
     326             :         {
     327           0 :             now = feGetCurrentTimestamp();
     328           0 :             if (!OutputFsync(now))
     329           0 :                 goto error;
     330           0 :             close(outfd);
     331           0 :             outfd = -1;
     332             :         }
     333         942 :         output_reopen = false;
     334             : 
     335             :         /* open the output file, if not open yet */
     336         942 :         if (outfd == -1)
     337             :         {
     338             :             struct stat statbuf;
     339             : 
     340          34 :             if (strcmp(outfile, "-") == 0)
     341          34 :                 outfd = fileno(stdout);
     342             :             else
     343           0 :                 outfd = open(outfile, O_CREAT | O_APPEND | O_WRONLY | PG_BINARY,
     344             :                              S_IRUSR | S_IWUSR);
     345          34 :             if (outfd == -1)
     346             :             {
     347           0 :                 pg_log_error("could not open log file \"%s\": %m", outfile);
     348           0 :                 goto error;
     349             :             }
     350             : 
     351          34 :             if (fstat(outfd, &statbuf) != 0)
     352             :             {
     353           0 :                 pg_log_error("could not stat file \"%s\": %m", outfile);
     354           0 :                 goto error;
     355             :             }
     356             : 
     357          34 :             output_isfile = S_ISREG(statbuf.st_mode) && !isatty(outfd);
     358             :         }
     359             : 
     360         942 :         r = PQgetCopyData(conn, &copybuf, 1);
     361         942 :         if (r == 0)
     362         272 :         {
     363             :             /*
     364             :              * In async mode, and no data available. We block on reading but
     365             :              * not more than the specified timeout, so that we can send a
     366             :              * response back to the client.
     367             :              */
     368             :             fd_set      input_mask;
     369         278 :             TimestampTz message_target = 0;
     370         278 :             TimestampTz fsync_target = 0;
     371             :             struct timeval timeout;
     372         278 :             struct timeval *timeoutptr = NULL;
     373             : 
     374         278 :             if (PQsocket(conn) < 0)
     375             :             {
     376           0 :                 pg_log_error("invalid socket: %s", PQerrorMessage(conn));
     377           4 :                 goto error;
     378             :             }
     379             : 
     380        4726 :             FD_ZERO(&input_mask);
     381         278 :             FD_SET(PQsocket(conn), &input_mask);
     382             : 
     383             :             /* Compute when we need to wakeup to send a keepalive message. */
     384         278 :             if (standby_message_timeout)
     385         278 :                 message_target = last_status + (standby_message_timeout - 1) *
     386             :                     ((int64) 1000);
     387             : 
     388             :             /* Compute when we need to wakeup to fsync the output file. */
     389         278 :             if (fsync_interval > 0 && output_needs_fsync)
     390         108 :                 fsync_target = output_last_fsync + (fsync_interval - 1) *
     391             :                     ((int64) 1000);
     392             : 
     393             :             /* Now compute when to wakeup. */
     394         278 :             if (message_target > 0 || fsync_target > 0)
     395             :             {
     396             :                 TimestampTz targettime;
     397             :                 long        secs;
     398             :                 int         usecs;
     399             : 
     400         278 :                 targettime = message_target;
     401             : 
     402         278 :                 if (fsync_target > 0 && fsync_target < targettime)
     403           0 :                     targettime = fsync_target;
     404             : 
     405         278 :                 feTimestampDifference(now,
     406             :                                       targettime,
     407             :                                       &secs,
     408             :                                       &usecs);
     409         278 :                 if (secs <= 0)
     410           0 :                     timeout.tv_sec = 1; /* Always sleep at least 1 sec */
     411             :                 else
     412         278 :                     timeout.tv_sec = secs;
     413         278 :                 timeout.tv_usec = usecs;
     414         278 :                 timeoutptr = &timeout;
     415             :             }
     416             : 
     417         278 :             r = select(PQsocket(conn) + 1, &input_mask, NULL, NULL, timeoutptr);
     418         278 :             if (r == 0 || (r < 0 && errno == EINTR))
     419             :             {
     420             :                 /*
     421             :                  * Got a timeout or signal. Continue the loop and either
     422             :                  * deliver a status packet to the server or just go back into
     423             :                  * blocking.
     424             :                  */
     425         274 :                 continue;
     426             :             }
     427         276 :             else if (r < 0)
     428             :             {
     429           0 :                 pg_log_error("%s() failed: %m", "select");
     430           0 :                 goto error;
     431             :             }
     432             : 
     433             :             /* Else there is actually data on the socket */
     434         276 :             if (PQconsumeInput(conn) == 0)
     435             :             {
     436           4 :                 pg_log_error("could not receive data from WAL stream: %s",
     437             :                              PQerrorMessage(conn));
     438           4 :                 goto error;
     439             :             }
     440         272 :             continue;
     441             :         }
     442             : 
     443             :         /* End of copy stream */
     444         664 :         if (r == -1)
     445          28 :             break;
     446             : 
     447             :         /* Failure while reading the copy stream */
     448         650 :         if (r == -2)
     449             :         {
     450           0 :             pg_log_error("could not read COPY data: %s",
     451             :                          PQerrorMessage(conn));
     452           0 :             goto error;
     453             :         }
     454             : 
     455             :         /* Check the message type. */
     456         650 :         if (copybuf[0] == 'k')
     457         464 :         {
     458             :             int         pos;
     459             :             bool        replyRequested;
     460             :             XLogRecPtr  walEnd;
     461         468 :             bool        endposReached = false;
     462             : 
     463             :             /*
     464             :              * Parse the keepalive message, enclosed in the CopyData message.
     465             :              * We just check if the server requested a reply, and ignore the
     466             :              * rest.
     467             :              */
     468         468 :             pos = 1;            /* skip msgtype 'k' */
     469         468 :             walEnd = fe_recvint64(&copybuf[pos]);
     470         468 :             output_written_lsn = Max(walEnd, output_written_lsn);
     471             : 
     472         468 :             pos += 8;           /* read walEnd */
     473             : 
     474         468 :             pos += 8;           /* skip sendTime */
     475             : 
     476         468 :             if (r < pos + 1)
     477             :             {
     478           0 :                 pg_log_error("streaming header too small: %d", r);
     479           0 :                 goto error;
     480             :             }
     481         468 :             replyRequested = copybuf[pos];
     482             : 
     483         468 :             if (endpos != InvalidXLogRecPtr && walEnd >= endpos)
     484             :             {
     485             :                 /*
     486             :                  * If there's nothing to read on the socket until a keepalive
     487             :                  * we know that the server has nothing to send us; and if
     488             :                  * walEnd has passed endpos, we know nothing else can have
     489             :                  * committed before endpos.  So we can bail out now.
     490             :                  */
     491           4 :                 endposReached = true;
     492             :             }
     493             : 
     494             :             /* Send a reply, if necessary */
     495         468 :             if (replyRequested || endposReached)
     496             :             {
     497           6 :                 if (!flushAndSendFeedback(conn, &now))
     498           0 :                     goto error;
     499           6 :                 last_status = now;
     500             :             }
     501             : 
     502         468 :             if (endposReached)
     503             :             {
     504           4 :                 stop_reason = STREAM_STOP_KEEPALIVE;
     505           4 :                 time_to_abort = true;
     506           4 :                 break;
     507             :             }
     508             : 
     509         464 :             continue;
     510             :         }
     511         182 :         else if (copybuf[0] != 'w')
     512             :         {
     513           0 :             pg_log_error("unrecognized streaming header: \"%c\"",
     514             :                          copybuf[0]);
     515           0 :             goto error;
     516             :         }
     517             : 
     518             :         /*
     519             :          * Read the header of the XLogData message, enclosed in the CopyData
     520             :          * message. We only need the WAL location field (dataStart), the rest
     521             :          * of the header is ignored.
     522             :          */
     523         182 :         hdr_len = 1;            /* msgtype 'w' */
     524         182 :         hdr_len += 8;           /* dataStart */
     525         182 :         hdr_len += 8;           /* walEnd */
     526         182 :         hdr_len += 8;           /* sendTime */
     527         182 :         if (r < hdr_len + 1)
     528             :         {
     529           0 :             pg_log_error("streaming header too small: %d", r);
     530           0 :             goto error;
     531             :         }
     532             : 
     533             :         /* Extract WAL location for this block */
     534         182 :         cur_record_lsn = fe_recvint64(&copybuf[1]);
     535             : 
     536         182 :         if (endpos != InvalidXLogRecPtr && cur_record_lsn > endpos)
     537             :         {
     538             :             /*
     539             :              * We've read past our endpoint, so prepare to go away being
     540             :              * cautious about what happens to our output data.
     541             :              */
     542           0 :             if (!flushAndSendFeedback(conn, &now))
     543           0 :                 goto error;
     544           0 :             stop_reason = STREAM_STOP_END_OF_WAL;
     545           0 :             time_to_abort = true;
     546           0 :             break;
     547             :         }
     548             : 
     549         182 :         output_written_lsn = Max(cur_record_lsn, output_written_lsn);
     550             : 
     551         182 :         bytes_left = r - hdr_len;
     552         182 :         bytes_written = 0;
     553             : 
     554             :         /* signal that a fsync is needed */
     555         182 :         output_needs_fsync = true;
     556             : 
     557         364 :         while (bytes_left)
     558             :         {
     559             :             int         ret;
     560             : 
     561         364 :             ret = write(outfd,
     562         182 :                         copybuf + hdr_len + bytes_written,
     563             :                         bytes_left);
     564             : 
     565         182 :             if (ret < 0)
     566             :             {
     567           0 :                 pg_log_error("could not write %d bytes to log file \"%s\": %m",
     568             :                              bytes_left, outfile);
     569           0 :                 goto error;
     570             :             }
     571             : 
     572             :             /* Write was successful, advance our position */
     573         182 :             bytes_written += ret;
     574         182 :             bytes_left -= ret;
     575             :         }
     576             : 
     577         182 :         if (write(outfd, "\n", 1) != 1)
     578             :         {
     579           0 :             pg_log_error("could not write %d bytes to log file \"%s\": %m",
     580             :                          1, outfile);
     581           0 :             goto error;
     582             :         }
     583             : 
     584         182 :         if (endpos != InvalidXLogRecPtr && cur_record_lsn == endpos)
     585             :         {
     586             :             /* endpos was exactly the record we just processed, we're done */
     587          10 :             if (!flushAndSendFeedback(conn, &now))
     588           0 :                 goto error;
     589          10 :             stop_reason = STREAM_STOP_END_OF_WAL;
     590          10 :             time_to_abort = true;
     591          10 :             break;
     592             :         }
     593             :     }
     594             : 
     595             :     /* Clean up connection state if stream has been aborted */
     596          30 :     if (time_to_abort)
     597          16 :         prepareToTerminate(conn, endpos, stop_reason, cur_record_lsn);
     598             : 
     599          30 :     res = PQgetResult(conn);
     600          30 :     if (PQresultStatus(res) == PGRES_COPY_OUT)
     601             :     {
     602          16 :         PQclear(res);
     603             : 
     604             :         /*
     605             :          * We're doing a client-initiated clean exit and have sent CopyDone to
     606             :          * the server. Drain any messages, so we don't miss a last-minute
     607             :          * ErrorResponse. The walsender stops generating XLogData records once
     608             :          * it sees CopyDone, so expect this to finish quickly. After CopyDone,
     609             :          * it's too late for sendFeedback(), even if this were to take a long
     610             :          * time. Hence, use synchronous-mode PQgetCopyData().
     611             :          */
     612             :         while (1)
     613           4 :         {
     614             :             int         r;
     615             : 
     616          20 :             if (copybuf != NULL)
     617             :             {
     618          18 :                 PQfreemem(copybuf);
     619          18 :                 copybuf = NULL;
     620             :             }
     621          20 :             r = PQgetCopyData(conn, &copybuf, 0);
     622          20 :             if (r == -1)
     623          16 :                 break;
     624           4 :             if (r == -2)
     625             :             {
     626           0 :                 pg_log_error("could not read COPY data: %s",
     627             :                              PQerrorMessage(conn));
     628           0 :                 time_to_abort = false;  /* unclean exit */
     629           0 :                 goto error;
     630             :             }
     631             :         }
     632             : 
     633          16 :         res = PQgetResult(conn);
     634             :     }
     635          30 :     if (PQresultStatus(res) != PGRES_COMMAND_OK)
     636             :     {
     637          12 :         pg_log_error("unexpected termination of replication stream: %s",
     638             :                      PQresultErrorMessage(res));
     639          12 :         PQclear(res);
     640          12 :         goto error;
     641             :     }
     642          18 :     PQclear(res);
     643             : 
     644          18 :     if (outfd != -1 && strcmp(outfile, "-") != 0)
     645             :     {
     646           0 :         TimestampTz t = feGetCurrentTimestamp();
     647             : 
     648             :         /* no need to jump to error on failure here, we're finishing anyway */
     649           0 :         OutputFsync(t);
     650             : 
     651           0 :         if (close(outfd) != 0)
     652           0 :             pg_log_error("could not close file \"%s\": %m", outfile);
     653             :     }
     654          18 :     outfd = -1;
     655          46 : error:
     656          46 :     if (copybuf != NULL)
     657             :     {
     658           0 :         PQfreemem(copybuf);
     659           0 :         copybuf = NULL;
     660             :     }
     661          46 :     destroyPQExpBuffer(query);
     662          46 :     PQfinish(conn);
     663          46 :     conn = NULL;
     664             : }
     665             : 
     666             : /*
     667             :  * Unfortunately we can't do sensible signal handling on windows...
     668             :  */
     669             : #ifndef WIN32
     670             : 
     671             : /*
     672             :  * When SIGINT/SIGTERM are caught, just tell the system to exit at the next
     673             :  * possible moment.
     674             :  */
     675             : static void
     676           2 : sigexit_handler(SIGNAL_ARGS)
     677             : {
     678           2 :     stop_reason = STREAM_STOP_SIGNAL;
     679           2 :     time_to_abort = true;
     680           2 : }
     681             : 
     682             : /*
     683             :  * Trigger the output file to be reopened.
     684             :  */
     685             : static void
     686           0 : sighup_handler(SIGNAL_ARGS)
     687             : {
     688           0 :     output_reopen = true;
     689           0 : }
     690             : #endif
     691             : 
     692             : 
     693             : int
     694         116 : main(int argc, char **argv)
     695             : {
     696             :     static struct option long_options[] = {
     697             : /* general options */
     698             :         {"file", required_argument, NULL, 'f'},
     699             :         {"fsync-interval", required_argument, NULL, 'F'},
     700             :         {"no-loop", no_argument, NULL, 'n'},
     701             :         {"failover", no_argument, NULL, 5},
     702             :         {"verbose", no_argument, NULL, 'v'},
     703             :         {"two-phase", no_argument, NULL, 't'},
     704             :         {"version", no_argument, NULL, 'V'},
     705             :         {"help", no_argument, NULL, '?'},
     706             : /* connection options */
     707             :         {"dbname", required_argument, NULL, 'd'},
     708             :         {"host", required_argument, NULL, 'h'},
     709             :         {"port", required_argument, NULL, 'p'},
     710             :         {"username", required_argument, NULL, 'U'},
     711             :         {"no-password", no_argument, NULL, 'w'},
     712             :         {"password", no_argument, NULL, 'W'},
     713             : /* replication options */
     714             :         {"startpos", required_argument, NULL, 'I'},
     715             :         {"endpos", required_argument, NULL, 'E'},
     716             :         {"option", required_argument, NULL, 'o'},
     717             :         {"plugin", required_argument, NULL, 'P'},
     718             :         {"status-interval", required_argument, NULL, 's'},
     719             :         {"slot", required_argument, NULL, 'S'},
     720             : /* action */
     721             :         {"create-slot", no_argument, NULL, 1},
     722             :         {"start", no_argument, NULL, 2},
     723             :         {"drop-slot", no_argument, NULL, 3},
     724             :         {"if-not-exists", no_argument, NULL, 4},
     725             :         {NULL, 0, NULL, 0}
     726             :     };
     727             :     int         c;
     728             :     int         option_index;
     729             :     uint32      hi,
     730             :                 lo;
     731             :     char       *db_name;
     732             : 
     733         116 :     pg_logging_init(argv[0]);
     734         116 :     progname = get_progname(argv[0]);
     735         116 :     set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_basebackup"));
     736             : 
     737         116 :     if (argc > 1)
     738             :     {
     739         114 :         if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
     740             :         {
     741           2 :             usage();
     742           2 :             exit(0);
     743             :         }
     744         112 :         else if (strcmp(argv[1], "-V") == 0 ||
     745         112 :                  strcmp(argv[1], "--version") == 0)
     746             :         {
     747           2 :             puts("pg_recvlogical (PostgreSQL) " PG_VERSION);
     748           2 :             exit(0);
     749             :         }
     750             :     }
     751             : 
     752         668 :     while ((c = getopt_long(argc, argv, "E:f:F:ntvd:h:p:U:wWI:o:P:s:S:",
     753         668 :                             long_options, &option_index)) != -1)
     754             :     {
     755         558 :         switch (c)
     756             :         {
     757             : /* general options */
     758          48 :             case 'f':
     759          48 :                 outfile = pg_strdup(optarg);
     760          48 :                 break;
     761           0 :             case 'F':
     762           0 :                 if (!option_parse_int(optarg, "-F/--fsync-interval", 0,
     763             :                                       INT_MAX / 1000,
     764             :                                       &fsync_interval))
     765           0 :                     exit(1);
     766           0 :                 fsync_interval *= 1000;
     767           0 :                 break;
     768          46 :             case 'n':
     769          46 :                 noloop = 1;
     770          46 :                 break;
     771           4 :             case 't':
     772           4 :                 two_phase = true;
     773           4 :                 break;
     774           0 :             case 'v':
     775           0 :                 verbose++;
     776           0 :                 break;
     777           2 :             case 5:
     778           2 :                 failover = true;
     779           2 :                 break;
     780             : /* connection options */
     781         104 :             case 'd':
     782         104 :                 dbname = pg_strdup(optarg);
     783         104 :                 break;
     784           0 :             case 'h':
     785           0 :                 dbhost = pg_strdup(optarg);
     786           0 :                 break;
     787           0 :             case 'p':
     788           0 :                 dbport = pg_strdup(optarg);
     789           0 :                 break;
     790           0 :             case 'U':
     791           0 :                 dbuser = pg_strdup(optarg);
     792           0 :                 break;
     793           0 :             case 'w':
     794           0 :                 dbgetpassword = -1;
     795           0 :                 break;
     796           0 :             case 'W':
     797           0 :                 dbgetpassword = 1;
     798           0 :                 break;
     799             : /* replication options */
     800           0 :             case 'I':
     801           0 :                 if (sscanf(optarg, "%X/%X", &hi, &lo) != 2)
     802           0 :                     pg_fatal("could not parse start position \"%s\"", optarg);
     803           0 :                 startpos = ((uint64) hi) << 32 | lo;
     804           0 :                 break;
     805          16 :             case 'E':
     806          16 :                 if (sscanf(optarg, "%X/%X", &hi, &lo) != 2)
     807           0 :                     pg_fatal("could not parse end position \"%s\"", optarg);
     808          16 :                 endpos = ((uint64) hi) << 32 | lo;
     809          16 :                 break;
     810          80 :             case 'o':
     811             :                 {
     812          80 :                     char       *data = pg_strdup(optarg);
     813          80 :                     char       *val = strchr(data, '=');
     814             : 
     815          80 :                     if (val != NULL)
     816             :                     {
     817             :                         /* remove =; separate data from val */
     818          80 :                         *val = '\0';
     819          80 :                         val++;
     820             :                     }
     821             : 
     822          80 :                     noptions += 1;
     823          80 :                     options = pg_realloc(options, sizeof(char *) * noptions * 2);
     824             : 
     825          80 :                     options[(noptions - 1) * 2] = data;
     826          80 :                     options[(noptions - 1) * 2 + 1] = val;
     827             :                 }
     828             : 
     829          80 :                 break;
     830          44 :             case 'P':
     831          44 :                 plugin = pg_strdup(optarg);
     832          44 :                 break;
     833           0 :             case 's':
     834           0 :                 if (!option_parse_int(optarg, "-s/--status-interval", 0,
     835             :                                       INT_MAX / 1000,
     836             :                                       &standby_message_timeout))
     837           0 :                     exit(1);
     838           0 :                 standby_message_timeout *= 1000;
     839           0 :                 break;
     840         108 :             case 'S':
     841         108 :                 replication_slot = pg_strdup(optarg);
     842         108 :                 break;
     843             : /* action */
     844          50 :             case 1:
     845          50 :                 do_create_slot = true;
     846          50 :                 break;
     847          50 :             case 2:
     848          50 :                 do_start_slot = true;
     849          50 :                 break;
     850           4 :             case 3:
     851           4 :                 do_drop_slot = true;
     852           4 :                 break;
     853           0 :             case 4:
     854           0 :                 slot_exists_ok = true;
     855           0 :                 break;
     856             : 
     857           2 :             default:
     858             :                 /* getopt_long already emitted a complaint */
     859           2 :                 pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     860           2 :                 exit(1);
     861             :         }
     862             :     }
     863             : 
     864             :     /*
     865             :      * Any non-option arguments?
     866             :      */
     867         110 :     if (optind < argc)
     868             :     {
     869           0 :         pg_log_error("too many command-line arguments (first is \"%s\")",
     870             :                      argv[optind]);
     871           0 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     872           0 :         exit(1);
     873             :     }
     874             : 
     875             :     /*
     876             :      * Required arguments
     877             :      */
     878         110 :     if (replication_slot == NULL)
     879             :     {
     880           2 :         pg_log_error("no slot specified");
     881           2 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     882           2 :         exit(1);
     883             :     }
     884             : 
     885         108 :     if (do_start_slot && outfile == NULL)
     886             :     {
     887           2 :         pg_log_error("no target file specified");
     888           2 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     889           2 :         exit(1);
     890             :     }
     891             : 
     892         106 :     if (!do_drop_slot && dbname == NULL)
     893             :     {
     894           2 :         pg_log_error("no database specified");
     895           2 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     896           2 :         exit(1);
     897             :     }
     898             : 
     899         104 :     if (!do_drop_slot && !do_create_slot && !do_start_slot)
     900             :     {
     901           2 :         pg_log_error("at least one action needs to be specified");
     902           2 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     903           2 :         exit(1);
     904             :     }
     905             : 
     906         102 :     if (do_drop_slot && (do_create_slot || do_start_slot))
     907             :     {
     908           0 :         pg_log_error("cannot use --create-slot or --start together with --drop-slot");
     909           0 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     910           0 :         exit(1);
     911             :     }
     912             : 
     913         102 :     if (startpos != InvalidXLogRecPtr && (do_create_slot || do_drop_slot))
     914             :     {
     915           0 :         pg_log_error("cannot use --create-slot or --drop-slot together with --startpos");
     916           0 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     917           0 :         exit(1);
     918             :     }
     919             : 
     920         102 :     if (endpos != InvalidXLogRecPtr && !do_start_slot)
     921             :     {
     922           0 :         pg_log_error("--endpos may only be specified with --start");
     923           0 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     924           0 :         exit(1);
     925             :     }
     926             : 
     927         102 :     if (!do_create_slot)
     928             :     {
     929          52 :         if (two_phase)
     930             :         {
     931           2 :             pg_log_error("--two-phase may only be specified with --create-slot");
     932           2 :             pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     933           2 :             exit(1);
     934             :         }
     935             : 
     936          50 :         if (failover)
     937             :         {
     938           0 :             pg_log_error("--failover may only be specified with --create-slot");
     939           0 :             pg_log_error_hint("Try \"%s --help\" for more information.", progname);
     940           0 :             exit(1);
     941             :         }
     942             :     }
     943             : 
     944             :     /*
     945             :      * Obtain a connection to server.  Notably, if we need a password, we want
     946             :      * to collect it from the user immediately.
     947             :      */
     948         100 :     conn = GetConnection();
     949         100 :     if (!conn)
     950             :         /* Error message already written in GetConnection() */
     951           0 :         exit(1);
     952         100 :     atexit(disconnect_atexit);
     953             : 
     954             :     /*
     955             :      * Trap signals.  (Don't do this until after the initial password prompt,
     956             :      * if one is needed, in GetConnection.)
     957             :      */
     958             : #ifndef WIN32
     959         100 :     pqsignal(SIGINT, sigexit_handler);
     960         100 :     pqsignal(SIGTERM, sigexit_handler);
     961         100 :     pqsignal(SIGHUP, sighup_handler);
     962             : #endif
     963             : 
     964             :     /*
     965             :      * Run IDENTIFY_SYSTEM to check the connection type for each action.
     966             :      * --create-slot and --start actions require a database-specific
     967             :      * replication connection because they handle logical replication slots.
     968             :      * --drop-slot can remove replication slots from any replication
     969             :      * connection without this restriction.
     970             :      */
     971         100 :     if (!RunIdentifySystem(conn, NULL, NULL, NULL, &db_name))
     972           0 :         exit(1);
     973             : 
     974         100 :     if (!do_drop_slot && db_name == NULL)
     975           0 :         pg_fatal("could not establish database-specific replication connection");
     976             : 
     977             :     /*
     978             :      * Set umask so that directories/files are created with the same
     979             :      * permissions as directories/files in the source data directory.
     980             :      *
     981             :      * pg_mode_mask is set to owner-only by default and then updated in
     982             :      * GetConnection() where we get the mode from the server-side with
     983             :      * RetrieveDataDirCreatePerm() and then call SetDataDirectoryCreatePerm().
     984             :      */
     985         100 :     umask(pg_mode_mask);
     986             : 
     987             :     /* Drop a replication slot. */
     988         100 :     if (do_drop_slot)
     989             :     {
     990           4 :         if (verbose)
     991           0 :             pg_log_info("dropping replication slot \"%s\"", replication_slot);
     992             : 
     993           4 :         if (!DropReplicationSlot(conn, replication_slot))
     994           0 :             exit(1);
     995             :     }
     996             : 
     997             :     /* Create a replication slot. */
     998         100 :     if (do_create_slot)
     999             :     {
    1000          50 :         if (verbose)
    1001           0 :             pg_log_info("creating replication slot \"%s\"", replication_slot);
    1002             : 
    1003          50 :         if (!CreateReplicationSlot(conn, replication_slot, plugin, false,
    1004             :                                    false, false, slot_exists_ok, two_phase,
    1005             :                                    failover))
    1006           0 :             exit(1);
    1007          50 :         startpos = InvalidXLogRecPtr;
    1008             :     }
    1009             : 
    1010         100 :     if (!do_start_slot)
    1011          54 :         exit(0);
    1012             : 
    1013             :     /* Stream loop */
    1014             :     while (true)
    1015             :     {
    1016          46 :         StreamLogicalLog();
    1017          46 :         if (time_to_abort)
    1018             :         {
    1019             :             /*
    1020             :              * We've been Ctrl-C'ed or reached an exit limit condition. That's
    1021             :              * not an error, so exit without an errorcode.
    1022             :              */
    1023          16 :             exit(0);
    1024             :         }
    1025          30 :         else if (noloop)
    1026          30 :             pg_fatal("disconnected");
    1027             :         else
    1028             :         {
    1029             :             /* translator: check source for value for %d */
    1030           0 :             pg_log_info("disconnected; waiting %d seconds to try again",
    1031             :                         RECONNECT_SLEEP_TIME);
    1032           0 :             pg_usleep(RECONNECT_SLEEP_TIME * 1000000);
    1033             :         }
    1034             :     }
    1035             : }
    1036             : 
    1037             : /*
    1038             :  * Fsync our output data, and send a feedback message to the server.  Returns
    1039             :  * true if successful, false otherwise.
    1040             :  *
    1041             :  * If successful, *now is updated to the current timestamp just before sending
    1042             :  * feedback.
    1043             :  */
    1044             : static bool
    1045          16 : flushAndSendFeedback(PGconn *conn, TimestampTz *now)
    1046             : {
    1047             :     /* flush data to disk, so that we send a recent flush pointer */
    1048          16 :     if (!OutputFsync(*now))
    1049           0 :         return false;
    1050          16 :     *now = feGetCurrentTimestamp();
    1051          16 :     if (!sendFeedback(conn, *now, true, false))
    1052           0 :         return false;
    1053             : 
    1054          16 :     return true;
    1055             : }
    1056             : 
    1057             : /*
    1058             :  * Try to inform the server about our upcoming demise, but don't wait around or
    1059             :  * retry on failure.
    1060             :  */
    1061             : static void
    1062          16 : prepareToTerminate(PGconn *conn, XLogRecPtr endpos, StreamStopReason reason,
    1063             :                    XLogRecPtr lsn)
    1064             : {
    1065          16 :     (void) PQputCopyEnd(conn, NULL);
    1066          16 :     (void) PQflush(conn);
    1067             : 
    1068          16 :     if (verbose)
    1069             :     {
    1070           0 :         switch (reason)
    1071             :         {
    1072           0 :             case STREAM_STOP_SIGNAL:
    1073           0 :                 pg_log_info("received interrupt signal, exiting");
    1074           0 :                 break;
    1075           0 :             case STREAM_STOP_KEEPALIVE:
    1076           0 :                 pg_log_info("end position %X/%X reached by keepalive",
    1077             :                             LSN_FORMAT_ARGS(endpos));
    1078           0 :                 break;
    1079           0 :             case STREAM_STOP_END_OF_WAL:
    1080             :                 Assert(!XLogRecPtrIsInvalid(lsn));
    1081           0 :                 pg_log_info("end position %X/%X reached by WAL record at %X/%X",
    1082             :                             LSN_FORMAT_ARGS(endpos), LSN_FORMAT_ARGS(lsn));
    1083           0 :                 break;
    1084           0 :             case STREAM_STOP_NONE:
    1085             :                 Assert(false);
    1086           0 :                 break;
    1087             :         }
    1088             :     }
    1089          16 : }

Generated by: LCOV version 1.16