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