Age Owner Branch data TLA 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
2798 peter@eisentraut.org 69 :CBC 9 : disconnect_atexit(void)
70 : : {
71 [ + + ]: 9 : if (conn != NULL)
72 : 3 : PQfinish(conn);
73 : 9 : }
74 : :
75 : : static void
5419 magnus@hagander.net 76 : 1 : usage(void)
77 : : {
3394 peter_e@gmx.net 78 : 1 : printf(_("%s receives PostgreSQL streaming write-ahead logs.\n\n"),
79 : : progname);
5419 magnus@hagander.net 80 : 1 : printf(_("Usage:\n"));
81 : 1 : printf(_(" %s [OPTION]...\n"), progname);
5189 peter_e@gmx.net 82 : 1 : printf(_("\nOptions:\n"));
3394 83 : 1 : printf(_(" -D, --directory=DIR receive write-ahead log files into this directory\n"));
3272 84 : 1 : printf(_(" -E, --endpos=LSN exit after receiving the specified LSN\n"));
3998 85 : 1 : printf(_(" --if-not-exists do not error if slot already exists when creating a slot\n"));
5140 alvherre@alvh.no-ip. 86 : 1 : printf(_(" -n, --no-loop do not loop on connection lost\n"));
3224 rhaas@postgresql.org 87 : 1 : printf(_(" --no-sync do not wait for changes to be written safely to disk\n"));
4337 peter_e@gmx.net 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"));
3394 91 : 1 : printf(_(" --synchronous flush write-ahead log immediately after writing\n"));
5140 alvherre@alvh.no-ip. 92 : 1 : printf(_(" -v, --verbose output verbose messages\n"));
93 : 1 : printf(_(" -V, --version output version information, then exit\n"));
1597 michael@paquier.xyz 94 : 1 : printf(_(" -Z, --compress=METHOD[:DETAIL]\n"
95 : : " compress as specified\n"));
5140 alvherre@alvh.no-ip. 96 : 1 : printf(_(" -?, --help show this help, then exit\n"));
5419 magnus@hagander.net 97 : 1 : printf(_("\nConnection options:\n"));
4931 heikki.linnakangas@i 98 : 1 : printf(_(" -d, --dbname=CONNSTR connection string\n"));
5140 alvherre@alvh.no-ip. 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"));
4343 andres@anarazel.de 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"));
2372 peter@eisentraut.org 107 : 1 : printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
108 : 1 : printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
5419 magnus@hagander.net 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
1757 michael@paquier.xyz 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 : : {
1757 michael@paquier.xyz 130 :UBC 0 : *ispartial = false;
1598 131 : 0 : *wal_compression_algorithm = PG_COMPRESSION_NONE;
1757 132 : 0 : return true;
133 : : }
134 : :
135 : : /* File looks like a completed gzip-compressed WAL file */
1757 michael@paquier.xyz 136 [ + + ]:CBC 9 : if (fname_len == XLOG_FNAME_LEN + strlen(".gz") &&
137 [ + - ]: 2 : strcmp(filename + XLOG_FNAME_LEN, ".gz") == 0)
138 : : {
139 : 2 : *ispartial = false;
1598 140 : 2 : *wal_compression_algorithm = PG_COMPRESSION_GZIP;
1757 141 : 2 : return true;
142 : : }
143 : :
144 : : /* File looks like a completed LZ4-compressed WAL file */
1756 145 [ + + ]: 7 : if (fname_len == XLOG_FNAME_LEN + strlen(".lz4") &&
146 [ + - ]: 1 : strcmp(filename + XLOG_FNAME_LEN, ".lz4") == 0)
147 : : {
148 : 1 : *ispartial = false;
1598 149 : 1 : *wal_compression_algorithm = PG_COMPRESSION_LZ4;
1756 150 : 1 : return true;
151 : : }
152 : :
153 : : /* File looks like a partial uncompressed WAL file */
1757 154 [ + + ]: 6 : if (fname_len == XLOG_FNAME_LEN + strlen(".partial") &&
155 [ + - ]: 3 : strcmp(filename + XLOG_FNAME_LEN, ".partial") == 0)
156 : : {
157 : 3 : *ispartial = true;
1598 158 : 3 : *wal_compression_algorithm = PG_COMPRESSION_NONE;
1757 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;
1598 167 : 2 : *wal_compression_algorithm = PG_COMPRESSION_GZIP;
1757 168 : 2 : return true;
169 : : }
170 : :
171 : : /* File looks like a partial LZ4-compressed WAL file */
1756 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;
1598 176 : 1 : *wal_compression_algorithm = PG_COMPRESSION_LZ4;
1756 177 : 1 : return true;
178 : : }
179 : :
180 : : /* File does not look like something we know */
1757 michael@paquier.xyz 181 :UBC 0 : return false;
182 : : }
183 : :
184 : : static bool
4970 heikki.linnakangas@i 185 :CBC 128 : 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 */
5207 magnus@hagander.net 191 [ + - + + ]: 128 : if (verbose && segment_finished)
416 alvherre@kurilemu.de 192 : 6 : pg_log_info("finished segment at %X/%08X (timeline %u)",
193 : : LSN_FORMAT_ARGS(xlogpos),
194 : : timeline);
195 : :
294 196 [ + - + + ]: 128 : if (XLogRecPtrIsValid(endpos) && endpos < xlogpos)
197 : : {
3272 peter_e@gmx.net 198 [ + - ]: 12 : if (verbose)
416 alvherre@kurilemu.de 199 : 12 : pg_log_info("stopped log streaming at %X/%08X (timeline %u)",
200 : : LSN_FORMAT_ARGS(xlogpos),
201 : : timeline);
3272 peter_e@gmx.net 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 : : */
3299 214 [ + - + + : 116 : if (verbose && prevtimeline != 0 && prevtimeline != timeline)
+ + ]
416 alvherre@kurilemu.de 215 : 1 : pg_log_info("switched to timeline %u at %X/%08X",
216 : : timeline,
217 : : LSN_FORMAT_ARGS(prevpos));
218 : :
4970 heikki.linnakangas@i 219 : 116 : prevtimeline = timeline;
220 : 116 : prevpos = xlogpos;
221 : :
3272 peter_e@gmx.net 222 [ - + ]: 116 : if (time_to_stop)
223 : : {
3299 peter_e@gmx.net 224 [ # # ]:UBC 0 : if (verbose)
2705 peter@eisentraut.org 225 : 0 : pg_log_info("received interrupt signal, exiting");
5419 magnus@hagander.net 226 : 0 : return true;
227 : : }
5419 magnus@hagander.net 228 :CBC 116 : return false;
229 : : }
230 : :
231 : :
232 : : /*
233 : : * Get destination directory.
234 : : */
235 : : static DIR *
4343 andres@anarazel.de 236 : 14 : get_destination_dir(char *dest_folder)
237 : : {
238 : : DIR *dir;
239 : :
240 [ - + ]: 14 : Assert(dest_folder != NULL);
241 : 14 : dir = opendir(dest_folder);
242 [ - + ]: 14 : if (dir == NULL)
1602 tgl@sss.pgh.pa.us 243 :UBC 0 : pg_fatal("could not open directory \"%s\": %m", dest_folder);
244 : :
4343 andres@anarazel.de 245 :CBC 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 [ + - + - ]: 14 : Assert(dest_dir != NULL && dest_folder != NULL);
256 [ - + ]: 14 : if (closedir(dest_dir))
1602 tgl@sss.pgh.pa.us 257 :UBC 0 : pg_fatal("could not close directory \"%s\": %m", dest_folder);
4343 andres@anarazel.de 258 :CBC 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
4970 heikki.linnakangas@i 269 : 7 : FindStreamingStart(uint32 *tli)
270 : : {
271 : : DIR *dir;
272 : : struct dirent *dirent;
5177 273 : 7 : XLogSegNo high_segno = 0;
4970 274 : 7 : uint32 high_tli = 0;
4721 275 : 7 : bool high_ispartial = false;
276 : :
4343 andres@anarazel.de 277 : 7 : dir = get_destination_dir(basedir);
278 : :
4542 bruce@momjian.us 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 : :
1757 michael@paquier.xyz 286 [ + + ]: 23 : if (!is_xlogfilename(dirent->d_name,
287 : : &ispartial, &wal_compression_algorithm))
5419 magnus@hagander.net 288 : 14 : continue;
289 : :
290 : : /*
291 : : * Looks like an xlog file. Parse its position.
292 : : */
3264 andres@anarazel.de 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 : : */
1598 michael@paquier.xyz 313 [ + + - + ]: 9 : if (!ispartial && wal_compression_algorithm == PG_COMPRESSION_NONE)
5419 magnus@hagander.net 314 :UBC 0 : {
315 : : struct stat statbuf;
316 : : char fullpath[MAXPGPATH * 2];
317 : :
4721 heikki.linnakangas@i 318 : 0 : snprintf(fullpath, sizeof(fullpath), "%s/%s", basedir, dirent->d_name);
319 [ # # ]: 0 : if (stat(fullpath, &statbuf) != 0)
1602 tgl@sss.pgh.pa.us 320 : 0 : pg_fatal("could not stat file \"%s\": %m", fullpath);
321 : :
3264 andres@anarazel.de 322 [ # # ]: 0 : if (statbuf.st_size != WalSegSz)
323 : : {
2163 peter@eisentraut.org 324 : 0 : pg_log_warning("segment file \"%s\" has incorrect size %lld, skipping",
325 : : dirent->d_name, (long long int) statbuf.st_size);
5419 magnus@hagander.net 326 : 0 : continue;
327 : : }
328 : : }
1598 michael@paquier.xyz 329 [ + + + + ]:CBC 9 : else if (!ispartial && wal_compression_algorithm == PG_COMPRESSION_GZIP)
3509 magnus@hagander.net 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 : :
2904 michael@paquier.xyz 339 : 2 : fd = open(fullpath, O_RDONLY | PG_BINARY, 0);
3509 magnus@hagander.net 340 [ - + ]: 2 : if (fd < 0)
1602 tgl@sss.pgh.pa.us 341 :UBC 0 : pg_fatal("could not open compressed file \"%s\": %m",
342 : : fullpath);
3389 bruce@momjian.us 343 [ - + ]:CBC 2 : if (lseek(fd, (off_t) (-4), SEEK_END) < 0)
1602 tgl@sss.pgh.pa.us 344 :UBC 0 : pg_fatal("could not seek in compressed file \"%s\": %m",
345 : : fullpath);
553 peter@eisentraut.org 346 :CBC 2 : r = read(fd, buf, sizeof(buf));
2962 michael@paquier.xyz 347 [ - + ]: 2 : if (r != sizeof(buf))
348 : : {
2962 michael@paquier.xyz 349 [ # # ]:UBC 0 : if (r < 0)
1602 tgl@sss.pgh.pa.us 350 : 0 : pg_fatal("could not read compressed file \"%s\": %m",
351 : : fullpath);
352 : : else
43 peter@eisentraut.org 353 :UNC 0 : pg_fatal("could not read compressed file \"%s\": read %zd of %zu",
354 : : fullpath, r, sizeof(buf));
355 : : }
356 : :
3509 magnus@hagander.net 357 :CBC 2 : close(fd);
358 : 2 : bytes_out = (buf[3] << 24) | (buf[2] << 16) |
3389 bruce@momjian.us 359 : 2 : (buf[1] << 8) | buf[0];
360 : :
3264 andres@anarazel.de 361 [ - + ]: 2 : if (bytes_out != WalSegSz)
362 : : {
2705 peter@eisentraut.org 363 :UBC 0 : pg_log_warning("compressed segment file \"%s\" has incorrect uncompressed size %d, skipping",
364 : : dirent->d_name, bytes_out);
3509 magnus@hagander.net 365 : 0 : continue;
366 : : }
367 : : }
1598 michael@paquier.xyz 368 [ + + + - ]:CBC 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;
1756 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)
1602 tgl@sss.pgh.pa.us 387 :UBC 0 : pg_fatal("could not open file \"%s\": %m", fullpath);
388 : :
1756 michael@paquier.xyz 389 :CBC 1 : status = LZ4F_createDecompressionContext(&ctx, LZ4F_VERSION);
390 [ - + ]: 1 : if (LZ4F_isError(status))
1602 tgl@sss.pgh.pa.us 391 :UBC 0 : pg_fatal("could not create LZ4 decompression context: %s",
392 : : LZ4F_getErrorName(status));
393 : :
1756 michael@paquier.xyz 394 :CBC 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)
1602 tgl@sss.pgh.pa.us 403 :UBC 0 : pg_fatal("could not read file \"%s\": %m", fullpath);
404 : :
405 : : /* Done reading the file */
1756 michael@paquier.xyz 406 [ + + ]:CBC 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))
1602 tgl@sss.pgh.pa.us 421 :UBC 0 : pg_fatal("could not decompress file \"%s\": %s",
422 : : fullpath,
423 : : LZ4F_getErrorName(status));
424 : :
1756 michael@paquier.xyz 425 :CBC 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))
1602 tgl@sss.pgh.pa.us 444 :UBC 0 : pg_fatal("could not free LZ4 decompression context: %s",
445 : : LZ4F_getErrorName(status));
446 : :
1756 michael@paquier.xyz 447 [ - + ]:CBC 1 : if (uncompressed_size != WalSegSz)
448 : : {
1753 tgl@sss.pgh.pa.us 449 :UBC 0 : pg_log_warning("compressed segment file \"%s\" has incorrect uncompressed size %zu, skipping",
450 : : dirent->d_name, uncompressed_size);
1756 michael@paquier.xyz 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. */
4721 heikki.linnakangas@i 461 [ + + ]:CBC 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 : :
4542 bruce@momjian.us 471 [ - + ]: 7 : if (errno)
1602 tgl@sss.pgh.pa.us 472 :UBC 0 : pg_fatal("could not read directory \"%s\": %m", basedir);
473 : :
4343 andres@anarazel.de 474 :CBC 7 : close_destination_dir(dir, basedir);
475 : :
5177 heikki.linnakangas@i 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 : : */
4721 485 [ - + ]: 3 : if (!high_ispartial)
4721 heikki.linnakangas@i 486 :UBC 0 : high_segno++;
487 : :
2971 alvherre@alvh.no-ip. 488 :CBC 3 : XLogSegNoOffsetToRecPtr(high_segno, 0, WalSegSz, high_ptr);
489 : :
4970 heikki.linnakangas@i 490 : 3 : *tli = high_tli;
5419 magnus@hagander.net 491 : 3 : return high_ptr;
492 : : }
493 : : else
4970 heikki.linnakangas@i 494 : 4 : return InvalidXLogRecPtr;
495 : : }
496 : :
497 : : /*
498 : : * Start the log streaming
499 : : */
500 : : static void
5419 magnus@hagander.net 501 : 7 : StreamLog(void)
502 : : {
503 : : XLogRecPtr serverpos;
504 : : TimeLineID servertli;
1503 peter@eisentraut.org 505 : 7 : StreamCtl stream = {0};
506 : : char *sysidentifier;
507 : :
508 : : /*
509 : : * Connect in replication mode to the server
510 : : */
4331 fujii@postgresql.org 511 [ - + ]: 7 : if (conn == NULL)
4331 fujii@postgresql.org 512 :UBC 0 : conn = GetConnection();
5205 magnus@hagander.net 513 [ - + ]:CBC 7 : if (!conn)
514 : : /* Error message already written in GetConnection() */
515 : 1 : return;
516 : :
4906 heikki.linnakangas@i 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 : : */
2798 peter@eisentraut.org 524 :UBC 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 : : */
1804 michael@paquier.xyz 532 [ - + ]:CBC 7 : if (!RunIdentifySystem(conn, &sysidentifier, &servertli, &serverpos, NULL))
2798 peter@eisentraut.org 533 :UBC 0 : exit(1);
534 : :
535 : : /*
536 : : * Figure out where to start streaming. First scan the local directory.
537 : : */
3821 magnus@hagander.net 538 :CBC 7 : stream.startpos = FindStreamingStart(&stream.timeline);
294 alvherre@kurilemu.de 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 : : */
1766 michael@paquier.xyz 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 : : */
294 alvherre@kurilemu.de 560 [ + + ]: 3 : if (!XLogRecPtrIsValid(stream.startpos))
561 : : {
1766 michael@paquier.xyz 562 : 1 : stream.startpos = serverpos;
563 : 1 : stream.timeline = servertli;
564 : : }
565 : : }
566 : :
294 alvherre@kurilemu.de 567 [ + - + - ]: 6 : Assert(XLogRecPtrIsValid(stream.startpos) &&
568 : : stream.timeline != 0);
569 : :
570 : : /*
571 : : * Always start streaming at the beginning of a segment
572 : : */
3264 andres@anarazel.de 573 : 6 : stream.startpos -= XLogSegmentOffset(stream.startpos, WalSegSz);
574 : :
575 : : /*
576 : : * Start the replication
577 : : */
5419 magnus@hagander.net 578 [ + - ]: 6 : if (verbose)
416 alvherre@kurilemu.de 579 : 6 : pg_log_info("starting log streaming at %X/%08X (timeline %u)",
580 : : LSN_FORMAT_ARGS(stream.startpos),
581 : : stream.timeline);
582 : :
3821 magnus@hagander.net 583 : 6 : stream.stream_stop = stop_streaming;
3409 tgl@sss.pgh.pa.us 584 : 6 : stream.stop_socket = PGINVALID_SOCKET;
3821 magnus@hagander.net 585 : 6 : stream.standby_message_timeout = standby_message_timeout;
586 : 6 : stream.synchronous = synchronous;
3224 rhaas@postgresql.org 587 : 6 : stream.do_sync = do_sync;
3821 magnus@hagander.net 588 : 6 : stream.mark_done = false;
1757 michael@paquier.xyz 589 : 12 : stream.walmethod = CreateWalDirectoryMethod(basedir,
590 : : compression_algorithm,
591 : : compresslevel,
3509 magnus@hagander.net 592 : 6 : stream.do_sync);
3821 593 : 6 : stream.partial_suffix = ".partial";
3510 594 : 6 : stream.replication_slot = replication_slot;
1804 michael@paquier.xyz 595 : 6 : stream.sysidentifier = sysidentifier;
596 : :
3821 magnus@hagander.net 597 : 6 : ReceiveXlogStream(conn, &stream);
598 : :
1438 rhaas@postgresql.org 599 [ - + ]: 6 : if (!stream.walmethod->ops->finish(stream.walmethod))
600 : : {
2705 peter@eisentraut.org 601 :UBC 0 : pg_log_info("could not finish writing WAL files: %m");
3595 magnus@hagander.net 602 : 0 : return;
603 : : }
604 : :
5411 magnus@hagander.net 605 :CBC 6 : PQfinish(conn);
2798 peter@eisentraut.org 606 : 6 : conn = NULL;
607 : :
1438 rhaas@postgresql.org 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
1443 tgl@sss.pgh.pa.us 618 :UBC 0 : sigexit_handler(SIGNAL_ARGS)
619 : : {
3272 peter_e@gmx.net 620 : 0 : time_to_stop = true;
5419 magnus@hagander.net 621 : 0 : }
622 : : #endif
623 : :
624 : : int
5419 magnus@hagander.net 625 :CBC 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;
1597 michael@paquier.xyz 656 : 19 : char *compression_detail = NULL;
657 : 19 : char *compression_algorithm_str = "none";
658 : 19 : char *error_detail = NULL;
659 : :
2705 peter@eisentraut.org 660 : 19 : pg_logging_init(argv[0]);
5419 magnus@hagander.net 661 : 19 : progname = get_progname(argv[0]);
3895 alvherre@alvh.no-ip. 662 : 19 : set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_basebackup"));
663 : :
5419 magnus@hagander.net 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 : : }
5140 alvherre@alvh.no-ip. 671 [ + - ]: 17 : else if (strcmp(argv[1], "-V") == 0 ||
672 [ + + ]: 17 : strcmp(argv[1], "--version") == 0)
673 : : {
3486 rhaas@postgresql.org 674 : 1 : puts("pg_receivewal (PostgreSQL) " PG_VERSION);
5419 magnus@hagander.net 675 : 1 : exit(0);
676 : : }
677 : : }
678 : :
1354 peter@eisentraut.org 679 : 68 : while ((c = getopt_long(argc, argv, "d:D:E:h:np:s:S:U:vwWZ:",
5419 magnus@hagander.net 680 [ + + ]: 68 : long_options, &option_index)) != -1)
681 : : {
682 [ - + + - : 54 : switch (c)
+ - - + -
+ - - + +
+ - + +
+ ]
683 : : {
1354 peter@eisentraut.org 684 :UBC 0 : case 'd':
685 : 0 : connection_string = pg_strdup(optarg);
686 : 0 : break;
5419 magnus@hagander.net 687 :CBC 11 : case 'D':
5077 tgl@sss.pgh.pa.us 688 : 11 : basedir = pg_strdup(optarg);
5419 magnus@hagander.net 689 : 11 : break;
1354 peter@eisentraut.org 690 : 9 : case 'E':
15 fujii@postgresql.org 691 [ + + ]:GNC 9 : if (!pg_parse_lsn(optarg, &endpos))
1354 peter@eisentraut.org 692 :GBC 2 : pg_fatal("could not parse end position \"%s\"", optarg);
4931 heikki.linnakangas@i 693 :CBC 7 : break;
5419 magnus@hagander.net 694 :UBC 0 : case 'h':
5077 tgl@sss.pgh.pa.us 695 : 0 : dbhost = pg_strdup(optarg);
5419 magnus@hagander.net 696 : 0 : break;
1354 peter@eisentraut.org 697 :CBC 7 : case 'n':
698 : 7 : noloop = true;
699 : 7 : break;
5419 magnus@hagander.net 700 :UBC 0 : case 'p':
5077 tgl@sss.pgh.pa.us 701 : 0 : dbport = pg_strdup(optarg);
5419 magnus@hagander.net 702 : 0 : break;
703 : 0 : case 's':
1860 michael@paquier.xyz 704 [ # # ]: 0 : if (!option_parse_int(optarg, "-s/--status-interval", 0,
705 : : INT_MAX / 1000,
706 : : &standby_message_timeout))
5419 magnus@hagander.net 707 : 0 : exit(1);
1860 michael@paquier.xyz 708 : 0 : standby_message_timeout *= 1000;
5419 magnus@hagander.net 709 : 0 : break;
4591 rhaas@postgresql.org 710 :CBC 5 : case 'S':
711 : 5 : replication_slot = pg_strdup(optarg);
712 : 5 : break;
1354 peter@eisentraut.org 713 :UBC 0 : case 'U':
714 : 0 : dbuser = pg_strdup(optarg);
5205 magnus@hagander.net 715 : 0 : break;
5419 magnus@hagander.net 716 :CBC 7 : case 'v':
717 : 7 : verbose++;
718 : 7 : break;
1354 peter@eisentraut.org 719 :UBC 0 : case 'w':
720 : 0 : dbgetpassword = -1;
721 : 0 : break;
722 : 0 : case 'W':
723 : 0 : dbgetpassword = 1;
724 : 0 : break;
3509 magnus@hagander.net 725 :CBC 3 : case 'Z':
1597 michael@paquier.xyz 726 : 3 : parse_compress_options(optarg, &compression_algorithm_str,
727 : : &compression_detail);
3509 magnus@hagander.net 728 : 3 : break;
4343 andres@anarazel.de 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;
4300 fujii@postgresql.org 735 :UBC 0 : case 3:
4064 andres@anarazel.de 736 : 0 : slot_exists_ok = true;
737 : 0 : break;
4064 andres@anarazel.de 738 :CBC 2 : case 4:
4300 fujii@postgresql.org 739 : 2 : synchronous = true;
740 : 2 : break;
3224 rhaas@postgresql.org 741 : 4 : case 5:
742 : 4 : do_sync = false;
743 : 4 : break;
5419 magnus@hagander.net 744 : 1 : default:
745 : : /* getopt_long already emitted a complaint */
1602 tgl@sss.pgh.pa.us 746 : 1 : pg_log_error_hint("Try \"%s --help\" for more information.", progname);
5419 magnus@hagander.net 747 : 1 : exit(1);
748 : : }
749 : : }
750 : :
751 : : /*
752 : : * Any non-option arguments?
753 : : */
754 [ - + ]: 14 : if (optind < argc)
755 : : {
2705 peter@eisentraut.org 756 :UBC 0 : pg_log_error("too many command-line arguments (first is \"%s\")",
757 : : argv[optind]);
1602 tgl@sss.pgh.pa.us 758 : 0 : pg_log_error_hint("Try \"%s --help\" for more information.", progname);
5419 magnus@hagander.net 759 : 0 : exit(1);
760 : : }
761 : :
4119 peter_e@gmx.net 762 [ + + + + ]:CBC 14 : if (do_drop_slot && do_create_slot)
763 : : {
2705 peter@eisentraut.org 764 : 1 : pg_log_error("cannot use --create-slot together with --drop-slot");
1602 tgl@sss.pgh.pa.us 765 : 1 : pg_log_error_hint("Try \"%s --help\" for more information.", progname);
4343 andres@anarazel.de 766 : 1 : exit(1);
767 : : }
768 : :
4119 peter_e@gmx.net 769 [ + + + - : 13 : if (replication_slot == NULL && (do_drop_slot || do_create_slot))
+ + ]
770 : : {
771 : : /* translator: %s is an option name */
2705 peter@eisentraut.org 772 [ - + ]: 1 : pg_log_error("%s needs a slot to be specified using --slot",
773 : : do_drop_slot ? "--drop-slot" : "--create-slot");
1602 tgl@sss.pgh.pa.us 774 : 1 : pg_log_error_hint("Try \"%s --help\" for more information.", progname);
4343 andres@anarazel.de 775 : 1 : exit(1);
776 : : }
777 : :
3224 rhaas@postgresql.org 778 [ + + + + ]: 12 : if (synchronous && !do_sync)
779 : : {
2705 peter@eisentraut.org 780 : 1 : pg_log_error("cannot use --synchronous together with --no-sync");
1602 tgl@sss.pgh.pa.us 781 : 1 : pg_log_error_hint("Try \"%s --help\" for more information.", progname);
3224 rhaas@postgresql.org 782 : 1 : exit(1);
783 : : }
784 : :
785 : : /*
786 : : * Required arguments
787 : : */
4035 andres@anarazel.de 788 [ + + + + : 11 : if (basedir == NULL && !do_drop_slot && !do_create_slot)
+ + ]
789 : : {
2705 peter@eisentraut.org 790 : 1 : pg_log_error("no target directory specified");
1602 tgl@sss.pgh.pa.us 791 : 1 : pg_log_error_hint("Try \"%s --help\" for more information.", progname);
5419 magnus@hagander.net 792 : 1 : exit(1);
793 : : }
794 : :
795 : : /*
796 : : * Compression options
797 : : */
1597 michael@paquier.xyz 798 [ - + ]: 10 : if (!parse_compress_algorithm(compression_algorithm_str,
799 : : &compression_algorithm))
1433 peter@eisentraut.org 800 :UBC 0 : pg_fatal("unrecognized compression algorithm: \"%s\"",
801 : : compression_algorithm_str);
802 : :
1597 michael@paquier.xyz 803 :CBC 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 */
1443 811 : 9 : compresslevel = compression_spec.level;
812 : :
813 [ - + ]: 9 : if (compression_algorithm == PG_COMPRESSION_ZSTD)
1443 michael@paquier.xyz 814 :UBC 0 : pg_fatal("compression with %s is not yet supported", "ZSTD");
815 : :
816 : : /*
817 : : * Check existence of destination folder.
818 : : */
4035 andres@anarazel.de 819 [ + + + + ]:CBC 9 : if (!do_drop_slot && !do_create_slot)
820 : : {
4114 bruce@momjian.us 821 : 7 : DIR *dir = get_destination_dir(basedir);
822 : :
4343 andres@anarazel.de 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() */
4343 andres@anarazel.de 832 :UBC 0 : exit(1);
2798 peter@eisentraut.org 833 :CBC 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
1443 dgustafsson@postgres 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 : : */
4343 andres@anarazel.de 849 [ - + ]: 9 : if (!RunIdentifySystem(conn, NULL, NULL, NULL, &db_name))
2798 peter@eisentraut.org 850 :UBC 0 : exit(1);
851 : :
852 : : /*
853 : : * Check that there is a database associated with connection, none should
854 : : * be defined in this context.
855 : : */
4343 andres@anarazel.de 856 [ - + ]:CBC 9 : if (db_name)
1602 tgl@sss.pgh.pa.us 857 :UBC 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 : : */
1804 michael@paquier.xyz 868 :CBC 9 : umask(pg_mode_mask);
869 : :
870 : : /*
871 : : * Drop a replication slot.
872 : : */
4343 andres@anarazel.de 873 [ + + ]: 9 : if (do_drop_slot)
874 : : {
875 [ - + ]: 1 : if (verbose)
2654 tgl@sss.pgh.pa.us 876 :UBC 0 : pg_log_info("dropping replication slot \"%s\"", replication_slot);
877 : :
4343 andres@anarazel.de 878 [ - + ]:CBC 1 : if (!DropReplicationSlot(conn, replication_slot))
2798 peter@eisentraut.org 879 :UBC 0 : exit(1);
2798 peter@eisentraut.org 880 :CBC 1 : exit(0);
881 : : }
882 : :
883 : : /* Create a replication slot */
4343 andres@anarazel.de 884 [ + + ]: 8 : if (do_create_slot)
885 : : {
886 [ - + ]: 1 : if (verbose)
2705 peter@eisentraut.org 887 :UBC 0 : pg_log_info("creating replication slot \"%s\"", replication_slot);
888 : :
3257 peter_e@gmx.net 889 [ - + ]:CBC 1 : if (!CreateReplicationSlot(conn, replication_slot, NULL, false, true, false,
890 : : slot_exists_ok, false, false))
2798 peter@eisentraut.org 891 :UBC 0 : exit(1);
2798 peter@eisentraut.org 892 :CBC 1 : exit(0);
893 : : }
894 : :
895 : : /* determine remote server's xlog segment size */
1804 michael@paquier.xyz 896 [ - + ]: 7 : if (!RetrieveWalSegSize(conn))
1804 michael@paquier.xyz 897 :UBC 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 : : {
5205 magnus@hagander.net 906 :CBC 7 : StreamLog();
3272 peter_e@gmx.net 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 : : */
5205 magnus@hagander.net 913 : 6 : exit(0);
914 : : }
915 [ + - ]: 1 : else if (noloop)
1602 tgl@sss.pgh.pa.us 916 : 1 : pg_fatal("disconnected");
917 : : else
918 : : {
919 : : /* translator: check source for value for %d */
2705 peter@eisentraut.org 920 :UBC 0 : pg_log_info("disconnected; waiting %d seconds to try again",
921 : : RECONNECT_SLEEP_TIME);
5205 magnus@hagander.net 922 : 0 : pg_usleep(RECONNECT_SLEEP_TIME * 1000000);
923 : : }
924 : : }
925 : : }
|
