40 #define RECONNECT_SLEEP_TIME 5
59 static void usage(
void);
67 bool segment_finished);
79 printf(
_(
"%s receives PostgreSQL streaming write-ahead logs.\n\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"
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);
133 result = strtol(
option, &endp, 10);
153 sep = strchr(
option,
':');
180 size_t fname_len = strlen(
filename);
181 size_t xlog_pattern_len = strspn(
filename,
"0123456789ABCDEF");
247 static uint32 prevtimeline = 0;
251 if (
verbose && segment_finished)
252 pg_log_info(
"finished segment at %X/%X (timeline %u)",
259 pg_log_info(
"stopped log streaming at %X/%X (timeline %u)",
274 if (
verbose && prevtimeline != 0 && prevtimeline != timeline)
279 prevtimeline = timeline;
300 Assert(dest_folder != NULL);
303 pg_fatal(
"could not open directory \"%s\": %m", dest_folder);
315 Assert(dest_dir != NULL && dest_folder != NULL);
317 pg_fatal(
"could not close directory \"%s\": %m", dest_folder);
335 bool high_ispartial =
false;
347 &ispartial, &wal_compression_algorithm))
379 if (
stat(fullpath, &statbuf) != 0)
380 pg_fatal(
"could not stat file \"%s\": %m", fullpath);
384 pg_log_warning(
"segment file \"%s\" has incorrect size %lld, skipping",
401 pg_fatal(
"could not open compressed file \"%s\": %m",
403 if (lseek(
fd, (off_t) (-4), SEEK_END) < 0)
404 pg_fatal(
"could not seek in compressed file \"%s\": %m",
407 if (r !=
sizeof(
buf))
410 pg_fatal(
"could not read compressed file \"%s\": %m",
413 pg_fatal(
"could not read compressed file \"%s\": read %d of %zu",
414 fullpath, r,
sizeof(
buf));
418 bytes_out = (
buf[3] << 24) | (
buf[2] << 16) |
423 pg_log_warning(
"compressed segment file \"%s\" has incorrect uncompressed size %d, skipping",
431 #define LZ4_CHUNK_SZ 64 * 1024
434 size_t uncompressed_size = 0;
438 LZ4F_decompressionContext_t ctx = NULL;
439 LZ4F_decompressOptions_t dec_opt;
442 memset(&dec_opt, 0,
sizeof(dec_opt));
447 pg_fatal(
"could not open file \"%s\": %m", fullpath);
449 status = LZ4F_createDecompressionContext(&ctx, LZ4F_VERSION);
451 pg_fatal(
"could not create LZ4 decompression context: %s",
452 LZ4F_getErrorName(
status));
461 r =
read(
fd, readbuf, LZ4_CHUNK_SZ);
463 pg_fatal(
"could not read file \"%s\": %m", fullpath);
471 readend = readbuf + r;
472 while (readp < readend)
474 size_t out_size = LZ4_CHUNK_SZ;
475 size_t read_size = readend - readp;
477 memset(outbuf, 0, LZ4_CHUNK_SZ);
478 status = LZ4F_decompress(ctx, outbuf, &out_size,
479 readp, &read_size, &dec_opt);
481 pg_fatal(
"could not decompress file \"%s\": %s",
483 LZ4F_getErrorName(
status));
486 uncompressed_size += out_size;
496 }
while (uncompressed_size <= WalSegSz && r > 0);
502 status = LZ4F_freeDecompressionContext(ctx);
504 pg_fatal(
"could not free LZ4 decompression context: %s",
505 LZ4F_getErrorName(
status));
509 pg_log_warning(
"compressed segment file \"%s\" has incorrect uncompressed size %zu, skipping",
523 if ((segno > high_segno) ||
524 (segno == high_segno && tli > high_tli) ||
525 (segno == high_segno && tli == high_tli && high_ispartial && !ispartial))
529 high_ispartial = ispartial;
570 MemSet(&stream, 0,
sizeof(stream));
643 pg_log_info(
"starting log streaming at %X/%X (timeline %u)",
665 pg_log_info(
"could not finish writing WAL files: %m");
693 static struct option long_options[] = {
724 char *compression_detail = NULL;
725 char *compression_algorithm_str =
"none";
726 char *error_detail = NULL;
734 if (strcmp(argv[1],
"--help") == 0 || strcmp(argv[1],
"-?") == 0)
739 else if (strcmp(argv[1],
"-V") == 0 ||
740 strcmp(argv[1],
"--version") == 0)
742 puts(
"pg_receivewal (PostgreSQL) " PG_VERSION);
747 while ((
c =
getopt_long(argc, argv,
"D:d:E:h:p:U:s:S:nwWvZ:",
748 long_options, &option_index)) != -1)
784 if (sscanf(
optarg,
"%X/%X", &hi, &lo) != 2)
786 endpos = ((uint64) hi) << 32 | lo;
796 &compression_detail);
826 pg_log_error(
"too many command-line arguments (first is \"%s\")",
834 pg_log_error(
"cannot use --create-slot together with --drop-slot");
842 pg_log_error(
"%s needs a slot to be specified using --slot",
850 pg_log_error(
"cannot use --synchronous together with --no-sync");
870 pg_fatal(
"unrecognized compression algorithm \"%s\"",
871 compression_algorithm_str);
876 if (error_detail != NULL)
877 pg_fatal(
"invalid compression specification: %s",
893 pg_log_info(
"no value specified for --compress, switching to default");
897 pg_fatal(
"this build does not support compression with %s",
903 pg_fatal(
"this build does not support compression with %s",
908 pg_fatal(
"compression with %s is not yet supported",
"ZSTD");
953 pg_fatal(
"replication connection using slot \"%s\" is unexpectedly database specific",
1016 pg_log_info(
"disconnected; waiting %d seconds to try again",
#define PG_TEXTDOMAIN(domain)
#define MemSet(start, val, len)
void set_pglocale_pgservice(const char *argv0, const char *app)
bool parse_compress_algorithm(char *name, pg_compress_algorithm *algorithm)
void parse_compress_specification(pg_compress_algorithm algorithm, char *specification, pg_compress_specification *result)
char * validate_compress_specification(pg_compress_specification *spec)
#define PG_COMPRESSION_OPTION_LEVEL
PGconn * GetConnection(UserMapping *user, bool will_prep_stmt, PgFdwConnState **state)
struct dirent * readdir(DIR *)
DIR * opendir(const char *)
int PQserverVersion(const PGconn *conn)
void PQfinish(PGconn *conn)
void * pg_malloc0(size_t size)
char * pg_strdup(const char *in)
int getopt_long(int argc, char *const argv[], const char *optstring, const struct option *longopts, int *longindex)
#define required_argument
Assert(fmt[strlen(fmt) - 1] !='\n')
void pg_logging_init(const char *argv0)
#define pg_log_error(...)
#define pg_log_error_hint(...)
#define pg_log_error_detail(...)
char * pstrdup(const char *in)
bool option_parse_int(const char *optarg, const char *optname, int min_range, int max_range, int *result)
#define Z_DEFAULT_COMPRESSION
PGDLLIMPORT char * optarg
static void close_destination_dir(DIR *dest_dir, char *dest_folder)
static volatile bool time_to_stop
static pg_compress_algorithm compression_algorithm
static void sigint_handler(int signum)
int main(int argc, char **argv)
static bool slot_exists_ok
static DIR * get_destination_dir(char *dest_folder)
static char * replication_slot
static void parse_compress_options(char *option, char **algorithm, char **detail)
static bool stop_streaming(XLogRecPtr segendpos, uint32 timeline, bool segment_finished)
static XLogRecPtr FindStreamingStart(uint32 *tli)
#define RECONNECT_SLEEP_TIME
static bool do_create_slot
static void disconnect_atexit(void)
static int standby_message_timeout
static bool is_xlogfilename(const char *filename, bool *ispartial, pg_compress_algorithm *wal_compression_algorithm)
static void StreamLog(void)
static void static void status(const char *fmt,...) pg_attribute_printf(1
#define pg_log_warning(...)
const char * get_progname(const char *argv0)
static int fd(const char *x, int i)
bool ReceiveXlogStream(PGconn *conn, StreamCtl *stream)
bool CheckServerVersionForStreaming(PGconn *conn)
void pg_usleep(long microsec)
pqsigfunc pqsignal(int signum, pqsigfunc handler)
bool RetrieveWalSegSize(PGconn *conn)
bool GetSlotInformation(PGconn *conn, const char *slot_name, XLogRecPtr *restart_lsn, TimeLineID *restart_tli)
bool RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli, XLogRecPtr *startpos, char **db_name)
stream_stop_callback stream_stop
int standby_message_timeout
WalWriteMethod * walmethod
WalWriteMethod * CreateWalDirectoryMethod(const char *basedir, pg_compress_algorithm compression_algorithm, int compression_level, bool sync)
void FreeWalDirectoryMethod(void)
static void CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
static void DropReplicationSlot(DropReplicationSlotCmd *cmd)
#define XLogSegmentOffset(xlogptr, wal_segsz_bytes)
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define XLogFromFileName(fname, tli, logSegNo, wal_segsz_bytes)
#define LSN_FORMAT_ARGS(lsn)
#define XLogRecPtrIsInvalid(r)
#define InvalidXLogRecPtr