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