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