src/bin/pg_basebackup/pg_receivewal.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * pg_receivewal.c - receive streaming WAL data and write it
   4  *                                        to a local file.
   5  *
   6  * Author: Magnus Hagander <magnus@hagander.net>
   7  *
   8  * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
   9  *
  10  * IDENTIFICATION
  11  *                src/bin/pg_basebackup/pg_receivewal.c
  12  *-------------------------------------------------------------------------
  13  */
  14
  15 #include "postgres_fe.h"
  16
  17 #include <dirent.h>
  18 #include <limits.h>
  19 #include <signal.h>
  20 #include <sys/stat.h>
  21 #include <unistd.h>
  22
  23 #ifdef USE_LZ4
  24 #include <lz4frame.h>
  25 #endif
  26 #ifdef HAVE_LIBZ
  27 #include <zlib.h>
  28 #endif
  29
  30 #include "access/xlog_internal.h"
  31 #include "common/file_perm.h"
  32 #include "common/logging.h"
  33 #include "fe_utils/option_utils.h"
  34 #include "getopt_long.h"
  35 #include "libpq-fe.h"
  36 #include "receivelog.h"
  37 #include "streamutil.h"
  38
  39 /* Time to sleep between reconnection attempts */
  40 #define RECONNECT_SLEEP_TIME 5
  41
  42 /* Global options */
  43 static char *basedir = NULL;
  44 static int      verbose = 0;
  45 static int      compresslevel = 0;
  46 static int      noloop = 0;
  47 static int      standby_message_timeout = 10 * 1000;    /* 10 sec = default */
  48 static volatile sig_atomic_t time_to_stop = false;
  49 static bool do_create_slot = false;
  50 static bool slot_exists_ok = false;
  51 static bool do_drop_slot = false;
  52 static bool do_sync = true;
  53 static bool synchronous = false;
  54 static char *replication_slot = NULL;
  55 static pg_compress_algorithm compression_algorithm = PG_COMPRESSION_NONE;
  56 static XLogRecPtr endpos = InvalidXLogRecPtr;
  57
  58
  59 static void usage(void);
  60 static void parse_compress_options(char *option, char **algorithm,
  61                                                                    char **detail);
  62 static DIR *get_destination_dir(char *dest_folder);
  63 static void close_destination_dir(DIR *dest_dir, char *dest_folder);
  64 static XLogRecPtr FindStreamingStart(uint32 *tli);
  65 static void StreamLog(void);
  66 static bool stop_streaming(XLogRecPtr xlogpos, uint32 timeline,
  67                                                    bool segment_finished);
  68
  69 static void
  70 disconnect_atexit(void)
  71 {
  72         if (conn != NULL)
  73                 PQfinish(conn);
  74 }
  75
  76 static void
  77 usage(void)
  78 {
  79         printf(_("%s receives PostgreSQL streaming write-ahead logs.\n\n"),
  80                    progname);
  81         printf(_("Usage:\n"));
  82         printf(_("  %s [OPTION]...\n"), progname);
  83         printf(_("\nOptions:\n"));
  84         printf(_("  -D, --directory=DIR    receive write-ahead log files into this directory\n"));
  85         printf(_("  -E, --endpos=LSN       exit after receiving the specified LSN\n"));
  86         printf(_("      --if-not-exists    do not error if slot already exists when creating a slot\n"));
  87         printf(_("  -n, --no-loop          do not loop on connection lost\n"));
  88         printf(_("      --no-sync          do not wait for changes to be written safely to disk\n"));
  89         printf(_("  -s, --status-interval=SECS\n"
  90                          "                         time between status packets sent to server (default: %d)\n"), (standby_message_timeout / 1000));
  91         printf(_("  -S, --slot=SLOTNAME    replication slot to use\n"));
  92         printf(_("      --synchronous      flush write-ahead log immediately after writing\n"));
  93         printf(_("  -v, --verbose          output verbose messages\n"));
  94         printf(_("  -V, --version          output version information, then exit\n"));
  95         printf(_("  -Z, --compress=METHOD[:DETAIL]\n"
  96                          "                         compress as specified\n"));
  97         printf(_("  -?, --help             show this help, then exit\n"));
  98         printf(_("\nConnection options:\n"));
  99         printf(_("  -d, --dbname=CONNSTR   connection string\n"));
 100         printf(_("  -h, --host=HOSTNAME    database server host or socket directory\n"));
 101         printf(_("  -p, --port=PORT        database server port number\n"));
 102         printf(_("  -U, --username=NAME    connect as specified database user\n"));
 103         printf(_("  -w, --no-password      never prompt for password\n"));
 104         printf(_("  -W, --password         force password prompt (should happen automatically)\n"));
 105         printf(_("\nOptional actions:\n"));
 106         printf(_("      --create-slot      create a new replication slot (for the slot's name see --slot)\n"));
 107         printf(_("      --drop-slot        drop the replication slot (for the slot's name see --slot)\n"));
 108         printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
 109         printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
 110 }
 111
 112 /*
 113  * Basic parsing of a value specified for -Z/--compress
 114  *
 115  * The parsing consists of a METHOD:DETAIL string fed later on to a more
 116  * advanced routine in charge of proper validation checks.  This only extracts
 117  * METHOD and DETAIL.  If only an integer is found, the method is implied by
 118  * the value specified.
 119  */
 120 static void
 121 parse_compress_options(char *option, char **algorithm, char **detail)
 122 {
 123         char       *sep;
 124         char       *endp;
 125         long            result;
 126
 127         /*
 128          * Check whether the compression specification consists of a bare integer.
 129          *
 130          * For backward-compatibility, assume "none" if the integer found is zero
 131          * and "gzip" otherwise.
 132          */
 133         result = strtol(option, &endp, 10);
 134         if (*endp == '\0')
 135         {
 136                 if (result == 0)
 137                 {
 138                         *algorithm = pstrdup("none");
 139                         *detail = NULL;
 140                 }
 141                 else
 142                 {
 143                         *algorithm = pstrdup("gzip");
 144                         *detail = pstrdup(option);
 145                 }
 146                 return;
 147         }
 148
 149         /*
 150          * Check whether there is a compression detail following the algorithm
 151          * name.
 152          */
 153         sep = strchr(option, ':');
 154         if (sep == NULL)
 155         {
 156                 *algorithm = pstrdup(option);
 157                 *detail = NULL;
 158         }
 159         else
 160         {
 161                 char       *alg;
 162
 163                 alg = palloc((sep - option) + 1);
 164                 memcpy(alg, option, sep - option);
 165                 alg[sep - option] = '\0';
 166
 167                 *algorithm = alg;
 168                 *detail = pstrdup(sep + 1);
 169         }
 170 }
 171
 172 /*
 173  * Check if the filename looks like a WAL file, letting caller know if this
 174  * WAL segment is partial and/or compressed.
 175  */
 176 static bool
 177 is_xlogfilename(const char *filename, bool *ispartial,
 178                                 pg_compress_algorithm *wal_compression_algorithm)
 179 {
 180         size_t          fname_len = strlen(filename);
 181         size_t          xlog_pattern_len = strspn(filename, "0123456789ABCDEF");
 182
 183         /* File does not look like a WAL file */
 184         if (xlog_pattern_len != XLOG_FNAME_LEN)
 185                 return false;
 186
 187         /* File looks like a completed uncompressed WAL file */
 188         if (fname_len == XLOG_FNAME_LEN)
 189         {
 190                 *ispartial = false;
 191                 *wal_compression_algorithm = PG_COMPRESSION_NONE;
 192                 return true;
 193         }
 194
 195         /* File looks like a completed gzip-compressed WAL file */
 196         if (fname_len == XLOG_FNAME_LEN + strlen(".gz") &&
 197                 strcmp(filename + XLOG_FNAME_LEN, ".gz") == 0)
 198         {
 199                 *ispartial = false;
 200                 *wal_compression_algorithm = PG_COMPRESSION_GZIP;
 201                 return true;
 202         }
 203
 204         /* File looks like a completed LZ4-compressed WAL file */
 205         if (fname_len == XLOG_FNAME_LEN + strlen(".lz4") &&
 206                 strcmp(filename + XLOG_FNAME_LEN, ".lz4") == 0)
 207         {
 208                 *ispartial = false;
 209                 *wal_compression_algorithm = PG_COMPRESSION_LZ4;
 210                 return true;
 211         }
 212
 213         /* File looks like a partial uncompressed WAL file */
 214         if (fname_len == XLOG_FNAME_LEN + strlen(".partial") &&
 215                 strcmp(filename + XLOG_FNAME_LEN, ".partial") == 0)
 216         {
 217                 *ispartial = true;
 218                 *wal_compression_algorithm = PG_COMPRESSION_NONE;
 219                 return true;
 220         }
 221
 222         /* File looks like a partial gzip-compressed WAL file */
 223         if (fname_len == XLOG_FNAME_LEN + strlen(".gz.partial") &&
 224                 strcmp(filename + XLOG_FNAME_LEN, ".gz.partial") == 0)
 225         {
 226                 *ispartial = true;
 227                 *wal_compression_algorithm = PG_COMPRESSION_GZIP;
 228                 return true;
 229         }
 230
 231         /* File looks like a partial LZ4-compressed WAL file */
 232         if (fname_len == XLOG_FNAME_LEN + strlen(".lz4.partial") &&
 233                 strcmp(filename + XLOG_FNAME_LEN, ".lz4.partial") == 0)
 234         {
 235                 *ispartial = true;
 236                 *wal_compression_algorithm = PG_COMPRESSION_LZ4;
 237                 return true;
 238         }
 239
 240         /* File does not look like something we know */
 241         return false;
 242 }
 243
 244 static bool
 245 stop_streaming(XLogRecPtr xlogpos, uint32 timeline, bool segment_finished)
 246 {
 247         static uint32 prevtimeline = 0;
 248         static XLogRecPtr prevpos = InvalidXLogRecPtr;
 249
 250         /* we assume that we get called once at the end of each segment */
 251         if (verbose && segment_finished)
 252                 pg_log_info("finished segment at %X/%X (timeline %u)",
 253                                         LSN_FORMAT_ARGS(xlogpos),
 254                                         timeline);
 255
 256         if (!XLogRecPtrIsInvalid(endpos) && endpos < xlogpos)
 257         {
 258                 if (verbose)
 259                         pg_log_info("stopped log streaming at %X/%X (timeline %u)",
 260                                                 LSN_FORMAT_ARGS(xlogpos),
 261                                                 timeline);
 262                 time_to_stop = true;
 263                 return true;
 264         }
 265
 266         /*
 267          * Note that we report the previous, not current, position here. After a
 268          * timeline switch, xlogpos points to the beginning of the segment because
 269          * that's where we always begin streaming. Reporting the end of previous
 270          * timeline isn't totally accurate, because the next timeline can begin
 271          * slightly before the end of the WAL that we received on the previous
 272          * timeline, but it's close enough for reporting purposes.
 273          */
 274         if (verbose && prevtimeline != 0 && prevtimeline != timeline)
 275                 pg_log_info("switched to timeline %u at %X/%X",
 276                                         timeline,
 277                                         LSN_FORMAT_ARGS(prevpos));
 278
 279         prevtimeline = timeline;
 280         prevpos = xlogpos;
 281
 282         if (time_to_stop)
 283         {
 284                 if (verbose)
 285                         pg_log_info("received interrupt signal, exiting");
 286                 return true;
 287         }
 288         return false;
 289 }
 290
 291
 292 /*
 293  * Get destination directory.
 294  */
 295 static DIR *
 296 get_destination_dir(char *dest_folder)
 297 {
 298         DIR                *dir;
 299
 300         Assert(dest_folder != NULL);
 301         dir = opendir(dest_folder);
 302         if (dir == NULL)
 303                 pg_fatal("could not open directory \"%s\": %m", dest_folder);
 304
 305         return dir;
 306 }
 307
 308
 309 /*
 310  * Close existing directory.
 311  */
 312 static void
 313 close_destination_dir(DIR *dest_dir, char *dest_folder)
 314 {
 315         Assert(dest_dir != NULL && dest_folder != NULL);
 316         if (closedir(dest_dir))
 317                 pg_fatal("could not close directory \"%s\": %m", dest_folder);
 318 }
 319
 320
 321 /*
 322  * Determine starting location for streaming, based on any existing xlog
 323  * segments in the directory. We start at the end of the last one that is
 324  * complete (size matches wal segment size), on the timeline with highest ID.
 325  *
 326  * If there are no WAL files in the directory, returns InvalidXLogRecPtr.
 327  */
 328 static XLogRecPtr
 329 FindStreamingStart(uint32 *tli)
 330 {
 331         DIR                *dir;
 332         struct dirent *dirent;
 333         XLogSegNo       high_segno = 0;
 334         uint32          high_tli = 0;
 335         bool            high_ispartial = false;
 336
 337         dir = get_destination_dir(basedir);
 338
 339         while (errno = 0, (dirent = readdir(dir)) != NULL)
 340         {
 341                 uint32          tli;
 342                 XLogSegNo       segno;
 343                 pg_compress_algorithm wal_compression_algorithm;
 344                 bool            ispartial;
 345
 346                 if (!is_xlogfilename(dirent->d_name,
 347                                                          &ispartial, &wal_compression_algorithm))
 348                         continue;
 349
 350                 /*
 351                  * Looks like an xlog file. Parse its position.
 352                  */
 353                 XLogFromFileName(dirent->d_name, &tli, &segno, WalSegSz);
 354
 355                 /*
 356                  * Check that the segment has the right size, if it's supposed to be
 357                  * completed.  For non-compressed segments just check the on-disk size
 358                  * and see if it matches a completed segment.  For gzip-compressed
 359                  * segments, look at the last 4 bytes of the compressed file, which is
 360                  * where the uncompressed size is located for files with a size lower
 361                  * than 4GB, and then compare it to the size of a completed segment.
 362                  * The 4 last bytes correspond to the ISIZE member according to
 363                  * http://www.zlib.org/rfc-gzip.html.
 364                  *
 365                  * For LZ4-compressed segments, uncompress the file in a throw-away
 366                  * buffer keeping track of the uncompressed size, then compare it to
 367                  * the size of a completed segment.  Per its protocol, LZ4 does not
 368                  * store the uncompressed size of an object by default.  contentSize
 369                  * is one possible way to do that, but we need to rely on a method
 370                  * where WAL segments could have been compressed by a different source
 371                  * than pg_receivewal, like an archive_command with lz4.
 372                  */
 373                 if (!ispartial && wal_compression_algorithm == PG_COMPRESSION_NONE)
 374                 {
 375                         struct stat statbuf;
 376                         char            fullpath[MAXPGPATH * 2];
 377
 378                         snprintf(fullpath, sizeof(fullpath), "%s/%s", basedir, dirent->d_name);
 379                         if (stat(fullpath, &statbuf) != 0)
 380                                 pg_fatal("could not stat file \"%s\": %m", fullpath);
 381
 382                         if (statbuf.st_size != WalSegSz)
 383                         {
 384                                 pg_log_warning("segment file \"%s\" has incorrect size %lld, skipping",
 385                                                            dirent->d_name, (long long int) statbuf.st_size);
 386                                 continue;
 387                         }
 388                 }
 389                 else if (!ispartial && wal_compression_algorithm == PG_COMPRESSION_GZIP)
 390                 {
 391                         int                     fd;
 392                         char            buf[4];
 393                         int                     bytes_out;
 394                         char            fullpath[MAXPGPATH * 2];
 395                         int                     r;
 396
 397                         snprintf(fullpath, sizeof(fullpath), "%s/%s", basedir, dirent->d_name);
 398
 399                         fd = open(fullpath, O_RDONLY | PG_BINARY, 0);
 400                         if (fd < 0)
 401                                 pg_fatal("could not open compressed file \"%s\": %m",
 402                                                  fullpath);
 403                         if (lseek(fd, (off_t) (-4), SEEK_END) < 0)
 404                                 pg_fatal("could not seek in compressed file \"%s\": %m",
 405                                                  fullpath);
 406                         r = read(fd, (char *) buf, sizeof(buf));
 407                         if (r != sizeof(buf))
 408                         {
 409                                 if (r < 0)
 410                                         pg_fatal("could not read compressed file \"%s\": %m",
 411                                                          fullpath);
 412                                 else
 413                                         pg_fatal("could not read compressed file \"%s\": read %d of %zu",
 414                                                          fullpath, r, sizeof(buf));
 415                         }
 416
 417                         close(fd);
 418                         bytes_out = (buf[3] << 24) | (buf[2] << 16) |
 419                                 (buf[1] << 8) | buf[0];
 420
 421                         if (bytes_out != WalSegSz)
 422                         {
 423                                 pg_log_warning("compressed segment file \"%s\" has incorrect uncompressed size %d, skipping",
 424                                                            dirent->d_name, bytes_out);
 425                                 continue;
 426                         }
 427                 }
 428                 else if (!ispartial && wal_compression_algorithm == PG_COMPRESSION_LZ4)
 429                 {
 430 #ifdef USE_LZ4
 431 #define LZ4_CHUNK_SZ    64 * 1024       /* 64kB as maximum chunk size read */
 432                         int                     fd;
 433                         ssize_t         r;
 434                         size_t          uncompressed_size = 0;
 435                         char            fullpath[MAXPGPATH * 2];
 436                         char       *outbuf;
 437                         char       *readbuf;
 438                         LZ4F_decompressionContext_t ctx = NULL;
 439                         LZ4F_decompressOptions_t dec_opt;
 440                         LZ4F_errorCode_t status;
 441
 442                         memset(&dec_opt, 0, sizeof(dec_opt));
 443                         snprintf(fullpath, sizeof(fullpath), "%s/%s", basedir, dirent->d_name);
 444
 445                         fd = open(fullpath, O_RDONLY | PG_BINARY, 0);
 446                         if (fd < 0)
 447                                 pg_fatal("could not open file \"%s\": %m", fullpath);
 448
 449                         status = LZ4F_createDecompressionContext(&ctx, LZ4F_VERSION);
 450                         if (LZ4F_isError(status))
 451                                 pg_fatal("could not create LZ4 decompression context: %s",
 452                                                  LZ4F_getErrorName(status));
 453
 454                         outbuf = pg_malloc0(LZ4_CHUNK_SZ);
 455                         readbuf = pg_malloc0(LZ4_CHUNK_SZ);
 456                         do
 457                         {
 458                                 char       *readp;
 459                                 char       *readend;
 460
 461                                 r = read(fd, readbuf, LZ4_CHUNK_SZ);
 462                                 if (r < 0)
 463                                         pg_fatal("could not read file \"%s\": %m", fullpath);
 464
 465                                 /* Done reading the file */
 466                                 if (r == 0)
 467                                         break;
 468
 469                                 /* Process one chunk */
 470                                 readp = readbuf;
 471                                 readend = readbuf + r;
 472                                 while (readp < readend)
 473                                 {
 474                                         size_t          out_size = LZ4_CHUNK_SZ;
 475                                         size_t          read_size = readend - readp;
 476
 477                                         memset(outbuf, 0, LZ4_CHUNK_SZ);
 478                                         status = LZ4F_decompress(ctx, outbuf, &out_size,
 479                                                                                          readp, &read_size, &dec_opt);
 480                                         if (LZ4F_isError(status))
 481                                                 pg_fatal("could not decompress file \"%s\": %s",
 482                                                                  fullpath,
 483                                                                  LZ4F_getErrorName(status));
 484
 485                                         readp += read_size;
 486                                         uncompressed_size += out_size;
 487                                 }
 488
 489                                 /*
 490                                  * No need to continue reading the file when the
 491                                  * uncompressed_size exceeds WalSegSz, even if there are still
 492                                  * data left to read. However, if uncompressed_size is equal
 493                                  * to WalSegSz, it should verify that there is no more data to
 494                                  * read.
 495                                  */
 496                         } while (uncompressed_size <= WalSegSz && r > 0);
 497
 498                         close(fd);
 499                         pg_free(outbuf);
 500                         pg_free(readbuf);
 501
 502                         status = LZ4F_freeDecompressionContext(ctx);
 503                         if (LZ4F_isError(status))
 504                                 pg_fatal("could not free LZ4 decompression context: %s",
 505                                                  LZ4F_getErrorName(status));
 506
 507                         if (uncompressed_size != WalSegSz)
 508                         {
 509                                 pg_log_warning("compressed segment file \"%s\" has incorrect uncompressed size %zu, skipping",
 510                                                            dirent->d_name, uncompressed_size);
 511                                 continue;
 512                         }
 513 #else
 514                         pg_log_error("could not check file \"%s\"",
 515                                                  dirent->d_name);
 516                         pg_log_error_detail("This build does not support compression with %s.",
 517                                                                 "LZ4");
 518                         exit(1);
 519 #endif
 520                 }
 521
 522                 /* Looks like a valid segment. Remember that we saw it. */
 523                 if ((segno > high_segno) ||
 524                         (segno == high_segno && tli > high_tli) ||
 525                         (segno == high_segno && tli == high_tli && high_ispartial && !ispartial))
 526                 {
 527                         high_segno = segno;
 528                         high_tli = tli;
 529                         high_ispartial = ispartial;
 530                 }
 531         }
 532
 533         if (errno)
 534                 pg_fatal("could not read directory \"%s\": %m", basedir);
 535
 536         close_destination_dir(dir, basedir);
 537
 538         if (high_segno > 0)
 539         {
 540                 XLogRecPtr      high_ptr;
 541
 542                 /*
 543                  * Move the starting pointer to the start of the next segment, if the
 544                  * highest one we saw was completed. Otherwise start streaming from
 545                  * the beginning of the .partial segment.
 546                  */
 547                 if (!high_ispartial)
 548                         high_segno++;
 549
 550                 XLogSegNoOffsetToRecPtr(high_segno, 0, WalSegSz, high_ptr);
 551
 552                 *tli = high_tli;
 553                 return high_ptr;
 554         }
 555         else
 556                 return InvalidXLogRecPtr;
 557 }
 558
 559 /*
 560  * Start the log streaming
 561  */
 562 static void
 563 StreamLog(void)
 564 {
 565         XLogRecPtr      serverpos;
 566         TimeLineID      servertli;
 567         StreamCtl       stream = {0};
 568         char       *sysidentifier;
 569
 570         /*
 571          * Connect in replication mode to the server
 572          */
 573         if (conn == NULL)
 574                 conn = GetConnection();
 575         if (!conn)
 576                 /* Error message already written in GetConnection() */
 577                 return;
 578
 579         if (!CheckServerVersionForStreaming(conn))
 580         {
 581                 /*
 582                  * Error message already written in CheckServerVersionForStreaming().
 583                  * There's no hope of recovering from a version mismatch, so don't
 584                  * retry.
 585                  */
 586                 exit(1);
 587         }
 588
 589         /*
 590          * Identify server, obtaining start LSN position and current timeline ID
 591          * at the same time, necessary if not valid data can be found in the
 592          * existing output directory.
 593          */
 594         if (!RunIdentifySystem(conn, &sysidentifier, &servertli, &serverpos, NULL))
 595                 exit(1);
 596
 597         /*
 598          * Figure out where to start streaming.  First scan the local directory.
 599          */
 600         stream.startpos = FindStreamingStart(&stream.timeline);
 601         if (stream.startpos == InvalidXLogRecPtr)
 602         {
 603                 /*
 604                  * Try to get the starting point from the slot if any.  This is
 605                  * supported in PostgreSQL 15 and newer.
 606                  */
 607                 if (replication_slot != NULL &&
 608                         PQserverVersion(conn) >= 150000)
 609                 {
 610                         if (!GetSlotInformation(conn, replication_slot, &stream.startpos,
 611                                                                         &stream.timeline))
 612                         {
 613                                 /* Error is logged by GetSlotInformation() */
 614                                 return;
 615                         }
 616                 }
 617
 618                 /*
 619                  * If it the starting point is still not known, use the current WAL
 620                  * flush value as last resort.
 621                  */
 622                 if (stream.startpos == InvalidXLogRecPtr)
 623                 {
 624                         stream.startpos = serverpos;
 625                         stream.timeline = servertli;
 626                 }
 627         }
 628
 629         Assert(stream.startpos != InvalidXLogRecPtr &&
 630                    stream.timeline != 0);
 631
 632         /*
 633          * Always start streaming at the beginning of a segment
 634          */
 635         stream.startpos -= XLogSegmentOffset(stream.startpos, WalSegSz);
 636
 637         /*
 638          * Start the replication
 639          */
 640         if (verbose)
 641                 pg_log_info("starting log streaming at %X/%X (timeline %u)",
 642                                         LSN_FORMAT_ARGS(stream.startpos),
 643                                         stream.timeline);
 644
 645         stream.stream_stop = stop_streaming;
 646         stream.stop_socket = PGINVALID_SOCKET;
 647         stream.standby_message_timeout = standby_message_timeout;
 648         stream.synchronous = synchronous;
 649         stream.do_sync = do_sync;
 650         stream.mark_done = false;
 651         stream.walmethod = CreateWalDirectoryMethod(basedir,
 652                                                                                                 compression_algorithm,
 653                                                                                                 compresslevel,
 654                                                                                                 stream.do_sync);
 655         stream.partial_suffix = ".partial";
 656         stream.replication_slot = replication_slot;
 657         stream.sysidentifier = sysidentifier;
 658
 659         ReceiveXlogStream(conn, &stream);
 660
 661         if (!stream.walmethod->ops->finish(stream.walmethod))
 662         {
 663                 pg_log_info("could not finish writing WAL files: %m");
 664                 return;
 665         }
 666
 667         PQfinish(conn);
 668         conn = NULL;
 669
 670         stream.walmethod->ops->free(stream.walmethod);
 671 }
 672
 673 /*
 674  * When SIGINT/SIGTERM are caught, just tell the system to exit at the next
 675  * possible moment.
 676  */
 677 #ifndef WIN32
 678
 679 static void
 680 sigexit_handler(SIGNAL_ARGS)
 681 {
 682         time_to_stop = true;
 683 }
 684 #endif
 685
 686 int
 687 main(int argc, char **argv)
 688 {
 689         static struct option long_options[] = {
 690                 {"help", no_argument, NULL, '?'},
 691                 {"version", no_argument, NULL, 'V'},
 692                 {"directory", required_argument, NULL, 'D'},
 693                 {"dbname", required_argument, NULL, 'd'},
 694                 {"endpos", required_argument, NULL, 'E'},
 695                 {"host", required_argument, NULL, 'h'},
 696                 {"port", required_argument, NULL, 'p'},
 697                 {"username", required_argument, NULL, 'U'},
 698                 {"no-loop", no_argument, NULL, 'n'},
 699                 {"no-password", no_argument, NULL, 'w'},
 700                 {"password", no_argument, NULL, 'W'},
 701                 {"status-interval", required_argument, NULL, 's'},
 702                 {"slot", required_argument, NULL, 'S'},
 703                 {"verbose", no_argument, NULL, 'v'},
 704                 {"compress", required_argument, NULL, 'Z'},
 705 /* action */
 706                 {"create-slot", no_argument, NULL, 1},
 707                 {"drop-slot", no_argument, NULL, 2},
 708                 {"if-not-exists", no_argument, NULL, 3},
 709                 {"synchronous", no_argument, NULL, 4},
 710                 {"no-sync", no_argument, NULL, 5},
 711                 {NULL, 0, NULL, 0}
 712         };
 713
 714         int                     c;
 715         int                     option_index;
 716         char       *db_name;
 717         uint32          hi,
 718                                 lo;
 719         pg_compress_specification compression_spec;
 720         char       *compression_detail = NULL;
 721         char       *compression_algorithm_str = "none";
 722         char       *error_detail = NULL;
 723
 724         pg_logging_init(argv[0]);
 725         progname = get_progname(argv[0]);
 726         set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_basebackup"));
 727
 728         if (argc > 1)
 729         {
 730                 if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
 731                 {
 732                         usage();
 733                         exit(0);
 734                 }
 735                 else if (strcmp(argv[1], "-V") == 0 ||
 736                                  strcmp(argv[1], "--version") == 0)
 737                 {
 738                         puts("pg_receivewal (PostgreSQL) " PG_VERSION);
 739                         exit(0);
 740                 }
 741         }
 742
 743         while ((c = getopt_long(argc, argv, "D:d:E:h:p:U:s:S:nwWvZ:",
 744                                                         long_options, &option_index)) != -1)
 745         {
 746                 switch (c)
 747                 {
 748                         case 'D':
 749                                 basedir = pg_strdup(optarg);
 750                                 break;
 751                         case 'd':
 752                                 connection_string = pg_strdup(optarg);
 753                                 break;
 754                         case 'h':
 755                                 dbhost = pg_strdup(optarg);
 756                                 break;
 757                         case 'p':
 758                                 dbport = pg_strdup(optarg);
 759                                 break;
 760                         case 'U':
 761                                 dbuser = pg_strdup(optarg);
 762                                 break;
 763                         case 'w':
 764                                 dbgetpassword = -1;
 765                                 break;
 766                         case 'W':
 767                                 dbgetpassword = 1;
 768                                 break;
 769                         case 's':
 770                                 if (!option_parse_int(optarg, "-s/--status-interval", 0,
 771                                                                           INT_MAX / 1000,
 772                                                                           &standby_message_timeout))
 773                                         exit(1);
 774                                 standby_message_timeout *= 1000;
 775                                 break;
 776                         case 'S':
 777                                 replication_slot = pg_strdup(optarg);
 778                                 break;
 779                         case 'E':
 780                                 if (sscanf(optarg, "%X/%X", &hi, &lo) != 2)
 781                                         pg_fatal("could not parse end position \"%s\"", optarg);
 782                                 endpos = ((uint64) hi) << 32 | lo;
 783                                 break;
 784                         case 'n':
 785                                 noloop = 1;
 786                                 break;
 787                         case 'v':
 788                                 verbose++;
 789                                 break;
 790                         case 'Z':
 791                                 parse_compress_options(optarg, &compression_algorithm_str,
 792                                                                            &compression_detail);
 793                                 break;
 794 /* action */
 795                         case 1:
 796                                 do_create_slot = true;
 797                                 break;
 798                         case 2:
 799                                 do_drop_slot = true;
 800                                 break;
 801                         case 3:
 802                                 slot_exists_ok = true;
 803                                 break;
 804                         case 4:
 805                                 synchronous = true;
 806                                 break;
 807                         case 5:
 808                                 do_sync = false;
 809                                 break;
 810                         default:
 811                                 /* getopt_long already emitted a complaint */
 812                                 pg_log_error_hint("Try \"%s --help\" for more information.", progname);
 813                                 exit(1);
 814                 }
 815         }
 816
 817         /*
 818          * Any non-option arguments?
 819          */
 820         if (optind < argc)
 821         {
 822                 pg_log_error("too many command-line arguments (first is \"%s\")",
 823                                          argv[optind]);
 824                 pg_log_error_hint("Try \"%s --help\" for more information.", progname);
 825                 exit(1);
 826         }
 827
 828         if (do_drop_slot && do_create_slot)
 829         {
 830                 pg_log_error("cannot use --create-slot together with --drop-slot");
 831                 pg_log_error_hint("Try \"%s --help\" for more information.", progname);
 832                 exit(1);
 833         }
 834
 835         if (replication_slot == NULL && (do_drop_slot || do_create_slot))
 836         {
 837                 /* translator: second %s is an option name */
 838                 pg_log_error("%s needs a slot to be specified using --slot",
 839                                          do_drop_slot ? "--drop-slot" : "--create-slot");
 840                 pg_log_error_hint("Try \"%s --help\" for more information.", progname);
 841                 exit(1);
 842         }
 843
 844         if (synchronous && !do_sync)
 845         {
 846                 pg_log_error("cannot use --synchronous together with --no-sync");
 847                 pg_log_error_hint("Try \"%s --help\" for more information.", progname);
 848                 exit(1);
 849         }
 850
 851         /*
 852          * Required arguments
 853          */
 854         if (basedir == NULL && !do_drop_slot && !do_create_slot)
 855         {
 856                 pg_log_error("no target directory specified");
 857                 pg_log_error_hint("Try \"%s --help\" for more information.", progname);
 858                 exit(1);
 859         }
 860
 861         /*
 862          * Compression options
 863          */
 864         if (!parse_compress_algorithm(compression_algorithm_str,
 865                                                                   &compression_algorithm))
 866                 pg_fatal("unrecognized compression algorithm \"%s\"",
 867                                  compression_algorithm_str);
 868
 869         parse_compress_specification(compression_algorithm, compression_detail,
 870                                                                  &compression_spec);
 871         error_detail = validate_compress_specification(&compression_spec);
 872         if (error_detail != NULL)
 873                 pg_fatal("invalid compression specification: %s",
 874                                  error_detail);
 875
 876         /* Extract the compression level */
 877         compresslevel = compression_spec.level;
 878
 879         if (compression_algorithm == PG_COMPRESSION_ZSTD)
 880                 pg_fatal("compression with %s is not yet supported", "ZSTD");
 881
 882         /*
 883          * Check existence of destination folder.
 884          */
 885         if (!do_drop_slot && !do_create_slot)
 886         {
 887                 DIR                *dir = get_destination_dir(basedir);
 888
 889                 close_destination_dir(dir, basedir);
 890         }
 891
 892         /*
 893          * Obtain a connection before doing anything.
 894          */
 895         conn = GetConnection();
 896         if (!conn)
 897                 /* error message already written in GetConnection() */
 898                 exit(1);
 899         atexit(disconnect_atexit);
 900
 901         /*
 902          * Trap signals.  (Don't do this until after the initial password prompt,
 903          * if one is needed, in GetConnection.)
 904          */
 905 #ifndef WIN32
 906         pqsignal(SIGINT, sigexit_handler);
 907         pqsignal(SIGTERM, sigexit_handler);
 908 #endif
 909
 910         /*
 911          * Run IDENTIFY_SYSTEM to make sure we've successfully have established a
 912          * replication connection and haven't connected using a database specific
 913          * connection.
 914          */
 915         if (!RunIdentifySystem(conn, NULL, NULL, NULL, &db_name))
 916                 exit(1);
 917
 918         /*
 919          * Check that there is a database associated with connection, none should
 920          * be defined in this context.
 921          */
 922         if (db_name)
 923                 pg_fatal("replication connection using slot \"%s\" is unexpectedly database specific",
 924                                  replication_slot);
 925
 926         /*
 927          * Set umask so that directories/files are created with the same
 928          * permissions as directories/files in the source data directory.
 929          *
 930          * pg_mode_mask is set to owner-only by default and then updated in
 931          * GetConnection() where we get the mode from the server-side with
 932          * RetrieveDataDirCreatePerm() and then call SetDataDirectoryCreatePerm().
 933          */
 934         umask(pg_mode_mask);
 935
 936         /*
 937          * Drop a replication slot.
 938          */
 939         if (do_drop_slot)
 940         {
 941                 if (verbose)
 942                         pg_log_info("dropping replication slot \"%s\"", replication_slot);
 943
 944                 if (!DropReplicationSlot(conn, replication_slot))
 945                         exit(1);
 946                 exit(0);
 947         }
 948
 949         /* Create a replication slot */
 950         if (do_create_slot)
 951         {
 952                 if (verbose)
 953                         pg_log_info("creating replication slot \"%s\"", replication_slot);
 954
 955                 if (!CreateReplicationSlot(conn, replication_slot, NULL, false, true, false,
 956                                                                    slot_exists_ok, false))
 957                         exit(1);
 958                 exit(0);
 959         }
 960
 961         /* determine remote server's xlog segment size */
 962         if (!RetrieveWalSegSize(conn))
 963                 exit(1);
 964
 965         /*
 966          * Don't close the connection here so that subsequent StreamLog() can
 967          * reuse it.
 968          */
 969
 970         while (true)
 971         {
 972                 StreamLog();
 973                 if (time_to_stop)
 974                 {
 975                         /*
 976                          * We've been Ctrl-C'ed or end of streaming position has been
 977                          * willingly reached, so exit without an error code.
 978                          */
 979                         exit(0);
 980                 }
 981                 else if (noloop)
 982                         pg_fatal("disconnected");
 983                 else
 984                 {
 985                         /* translator: check source for value for %d */
 986                         pg_log_info("disconnected; waiting %d seconds to try again",
 987                                                 RECONNECT_SLEEP_TIME);
 988                         pg_usleep(RECONNECT_SLEEP_TIME * 1000000);
 989                 }
 990         }
 991 }