Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * libpqwalreceiver.c
4 : *
5 : * This file contains the libpq-specific parts of walreceiver. It's
6 : * loaded as a dynamic module to avoid linking the main server binary with
7 : * libpq.
8 : *
9 : * Apart from walreceiver, the libpq-specific routines are now being used by
10 : * logical replication workers and slot synchronization.
11 : *
12 : * Portions Copyright (c) 2010-2025, PostgreSQL Global Development Group
13 : *
14 : *
15 : * IDENTIFICATION
16 : * src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
17 : *
18 : *-------------------------------------------------------------------------
19 : */
20 : #include "postgres.h"
21 :
22 : #include <unistd.h>
23 : #include <sys/time.h>
24 :
25 : #include "common/connect.h"
26 : #include "funcapi.h"
27 : #include "libpq-fe.h"
28 : #include "libpq/libpq-be-fe-helpers.h"
29 : #include "mb/pg_wchar.h"
30 : #include "miscadmin.h"
31 : #include "pgstat.h"
32 : #include "pqexpbuffer.h"
33 : #include "replication/walreceiver.h"
34 : #include "storage/latch.h"
35 : #include "utils/builtins.h"
36 : #include "utils/memutils.h"
37 : #include "utils/pg_lsn.h"
38 : #include "utils/tuplestore.h"
39 :
40 1718 : PG_MODULE_MAGIC_EXT(
41 : .name = "libpqwalreceiver",
42 : .version = PG_VERSION
43 : );
44 :
45 : struct WalReceiverConn
46 : {
47 : /* Current connection to the primary, if any */
48 : PGconn *streamConn;
49 : /* Used to remember if the connection is logical or physical */
50 : bool logical;
51 : /* Buffer for currently read records */
52 : char *recvBuf;
53 : };
54 :
55 : /* Prototypes for interface functions */
56 : static WalReceiverConn *libpqrcv_connect(const char *conninfo,
57 : bool replication, bool logical,
58 : bool must_use_password,
59 : const char *appname, char **err);
60 : static void libpqrcv_check_conninfo(const char *conninfo,
61 : bool must_use_password);
62 : static char *libpqrcv_get_conninfo(WalReceiverConn *conn);
63 : static void libpqrcv_get_senderinfo(WalReceiverConn *conn,
64 : char **sender_host, int *sender_port);
65 : static char *libpqrcv_identify_system(WalReceiverConn *conn,
66 : TimeLineID *primary_tli);
67 : static char *libpqrcv_get_dbname_from_conninfo(const char *connInfo);
68 : static int libpqrcv_server_version(WalReceiverConn *conn);
69 : static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
70 : TimeLineID tli, char **filename,
71 : char **content, int *len);
72 : static bool libpqrcv_startstreaming(WalReceiverConn *conn,
73 : const WalRcvStreamOptions *options);
74 : static void libpqrcv_endstreaming(WalReceiverConn *conn,
75 : TimeLineID *next_tli);
76 : static int libpqrcv_receive(WalReceiverConn *conn, char **buffer,
77 : pgsocket *wait_fd);
78 : static void libpqrcv_send(WalReceiverConn *conn, const char *buffer,
79 : int nbytes);
80 : static char *libpqrcv_create_slot(WalReceiverConn *conn,
81 : const char *slotname,
82 : bool temporary,
83 : bool two_phase,
84 : bool failover,
85 : CRSSnapshotAction snapshot_action,
86 : XLogRecPtr *lsn);
87 : static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
88 : const bool *failover, const bool *two_phase);
89 : static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
90 : static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
91 : const char *query,
92 : const int nRetTypes,
93 : const Oid *retTypes);
94 : static void libpqrcv_disconnect(WalReceiverConn *conn);
95 :
96 : static WalReceiverFunctionsType PQWalReceiverFunctions = {
97 : .walrcv_connect = libpqrcv_connect,
98 : .walrcv_check_conninfo = libpqrcv_check_conninfo,
99 : .walrcv_get_conninfo = libpqrcv_get_conninfo,
100 : .walrcv_get_senderinfo = libpqrcv_get_senderinfo,
101 : .walrcv_identify_system = libpqrcv_identify_system,
102 : .walrcv_server_version = libpqrcv_server_version,
103 : .walrcv_readtimelinehistoryfile = libpqrcv_readtimelinehistoryfile,
104 : .walrcv_startstreaming = libpqrcv_startstreaming,
105 : .walrcv_endstreaming = libpqrcv_endstreaming,
106 : .walrcv_receive = libpqrcv_receive,
107 : .walrcv_send = libpqrcv_send,
108 : .walrcv_create_slot = libpqrcv_create_slot,
109 : .walrcv_alter_slot = libpqrcv_alter_slot,
110 : .walrcv_get_dbname_from_conninfo = libpqrcv_get_dbname_from_conninfo,
111 : .walrcv_get_backend_pid = libpqrcv_get_backend_pid,
112 : .walrcv_exec = libpqrcv_exec,
113 : .walrcv_disconnect = libpqrcv_disconnect
114 : };
115 :
116 : /* Prototypes for private functions */
117 : static char *stringlist_to_identifierstr(PGconn *conn, List *strings);
118 :
119 : /*
120 : * Module initialization function
121 : */
122 : void
123 1718 : _PG_init(void)
124 : {
125 1718 : if (WalReceiverFunctions != NULL)
126 0 : elog(ERROR, "libpqwalreceiver already loaded");
127 1718 : WalReceiverFunctions = &PQWalReceiverFunctions;
128 1718 : }
129 :
130 : /*
131 : * Establish the connection to the primary server.
132 : *
133 : * This function can be used for both replication and regular connections.
134 : * If it is a replication connection, it could be either logical or physical
135 : * based on input argument 'logical'.
136 : *
137 : * If an error occurs, this function will normally return NULL and set *err
138 : * to a palloc'ed error message. However, if must_use_password is true and
139 : * the connection fails to use the password, this function will ereport(ERROR).
140 : * We do this because in that case the error includes a detail and a hint for
141 : * consistency with other parts of the system, and it's not worth adding the
142 : * machinery to pass all of those back to the caller just to cover this one
143 : * case.
144 : */
145 : static WalReceiverConn *
146 1714 : libpqrcv_connect(const char *conninfo, bool replication, bool logical,
147 : bool must_use_password, const char *appname, char **err)
148 : {
149 : WalReceiverConn *conn;
150 : const char *keys[6];
151 : const char *vals[6];
152 1714 : int i = 0;
153 :
154 : /*
155 : * Re-validate connection string. The validation already happened at DDL
156 : * time, but the subscription owner may have changed. If we don't recheck
157 : * with the correct must_use_password, it's possible that the connection
158 : * will obtain the password from a different source, such as PGPASSFILE or
159 : * PGPASSWORD.
160 : */
161 1714 : libpqrcv_check_conninfo(conninfo, must_use_password);
162 :
163 : /*
164 : * We use the expand_dbname parameter to process the connection string (or
165 : * URI), and pass some extra options.
166 : */
167 1694 : keys[i] = "dbname";
168 1694 : vals[i] = conninfo;
169 :
170 : /* We can not have logical without replication */
171 : Assert(replication || !logical);
172 :
173 1694 : if (replication)
174 : {
175 1670 : keys[++i] = "replication";
176 1670 : vals[i] = logical ? "database" : "true";
177 :
178 1670 : if (logical)
179 : {
180 : /* Tell the publisher to translate to our encoding */
181 1240 : keys[++i] = "client_encoding";
182 1240 : vals[i] = GetDatabaseEncodingName();
183 :
184 : /*
185 : * Force assorted GUC parameters to settings that ensure that the
186 : * publisher will output data values in a form that is unambiguous
187 : * to the subscriber. (We don't want to modify the subscriber's
188 : * GUC settings, since that might surprise user-defined code
189 : * running in the subscriber, such as triggers.) This should
190 : * match what pg_dump does.
191 : */
192 1240 : keys[++i] = "options";
193 1240 : vals[i] = "-c datestyle=ISO -c intervalstyle=postgres -c extra_float_digits=3";
194 : }
195 : else
196 : {
197 : /*
198 : * The database name is ignored by the server in replication mode,
199 : * but specify "replication" for .pgpass lookup.
200 : */
201 430 : keys[++i] = "dbname";
202 430 : vals[i] = "replication";
203 : }
204 : }
205 :
206 1694 : keys[++i] = "fallback_application_name";
207 1694 : vals[i] = appname;
208 :
209 1694 : keys[++i] = NULL;
210 1694 : vals[i] = NULL;
211 :
212 : Assert(i < lengthof(keys));
213 :
214 1694 : conn = palloc0(sizeof(WalReceiverConn));
215 1692 : conn->streamConn =
216 1694 : libpqsrv_connect_params(keys, vals,
217 : /* expand_dbname = */ true,
218 : WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
219 :
220 1692 : if (PQstatus(conn->streamConn) != CONNECTION_OK)
221 202 : goto bad_connection_errmsg;
222 :
223 1490 : if (must_use_password && !PQconnectionUsedPassword(conn->streamConn))
224 : {
225 0 : libpqsrv_disconnect(conn->streamConn);
226 0 : pfree(conn);
227 :
228 0 : ereport(ERROR,
229 : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
230 : errmsg("password is required"),
231 : errdetail("Non-superuser cannot connect if the server does not request a password."),
232 : errhint("Target server's authentication method must be changed, or set password_required=false in the subscription parameters.")));
233 : }
234 :
235 : /*
236 : * Set always-secure search path for the cases where the connection is
237 : * used to run SQL queries, so malicious users can't get control.
238 : */
239 1490 : if (!replication || logical)
240 : {
241 : PGresult *res;
242 :
243 1224 : res = libpqsrv_exec(conn->streamConn,
244 : ALWAYS_SECURE_SEARCH_PATH_SQL,
245 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
246 1224 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
247 : {
248 0 : PQclear(res);
249 0 : *err = psprintf(_("could not clear search path: %s"),
250 0 : pchomp(PQerrorMessage(conn->streamConn)));
251 0 : goto bad_connection;
252 : }
253 1224 : PQclear(res);
254 : }
255 :
256 1490 : conn->logical = logical;
257 :
258 1490 : return conn;
259 :
260 : /* error path, using libpq's error message */
261 202 : bad_connection_errmsg:
262 202 : *err = pchomp(PQerrorMessage(conn->streamConn));
263 :
264 : /* error path, error already set */
265 202 : bad_connection:
266 202 : libpqsrv_disconnect(conn->streamConn);
267 202 : pfree(conn);
268 202 : return NULL;
269 : }
270 :
271 : /*
272 : * Validate connection info string.
273 : *
274 : * If the connection string can't be parsed, this function will raise
275 : * an error. If must_use_password is true, the function raises an error
276 : * if no password is provided in the connection string. In any other case
277 : * it successfully completes.
278 : */
279 : static void
280 2074 : libpqrcv_check_conninfo(const char *conninfo, bool must_use_password)
281 : {
282 2074 : PQconninfoOption *opts = NULL;
283 : PQconninfoOption *opt;
284 2074 : char *err = NULL;
285 :
286 2074 : opts = PQconninfoParse(conninfo, &err);
287 2074 : if (opts == NULL)
288 : {
289 : /* The error string is malloc'd, so we must free it explicitly */
290 18 : char *errcopy = err ? pstrdup(err) : "out of memory";
291 :
292 18 : PQfreemem(err);
293 18 : ereport(ERROR,
294 : (errcode(ERRCODE_SYNTAX_ERROR),
295 : errmsg("invalid connection string syntax: %s", errcopy)));
296 : }
297 :
298 2056 : if (must_use_password)
299 : {
300 38 : bool uses_password = false;
301 :
302 1362 : for (opt = opts; opt->keyword != NULL; ++opt)
303 : {
304 : /* Ignore connection options that are not present. */
305 1336 : if (opt->val == NULL)
306 1248 : continue;
307 :
308 88 : if (strcmp(opt->keyword, "password") == 0 && opt->val[0] != '\0')
309 : {
310 12 : uses_password = true;
311 12 : break;
312 : }
313 : }
314 :
315 38 : if (!uses_password)
316 : {
317 : /* malloc'd, so we must free it explicitly */
318 26 : PQconninfoFree(opts);
319 :
320 26 : ereport(ERROR,
321 : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
322 : errmsg("password is required"),
323 : errdetail("Non-superusers must provide a password in the connection string.")));
324 : }
325 : }
326 :
327 2030 : PQconninfoFree(opts);
328 2030 : }
329 :
330 : /*
331 : * Return a user-displayable conninfo string. Any security-sensitive fields
332 : * are obfuscated.
333 : */
334 : static char *
335 266 : libpqrcv_get_conninfo(WalReceiverConn *conn)
336 : {
337 : PQconninfoOption *conn_opts;
338 : PQconninfoOption *conn_opt;
339 : PQExpBufferData buf;
340 : char *retval;
341 :
342 : Assert(conn->streamConn != NULL);
343 :
344 266 : initPQExpBuffer(&buf);
345 266 : conn_opts = PQconninfo(conn->streamConn);
346 :
347 266 : if (conn_opts == NULL)
348 0 : ereport(ERROR,
349 : (errcode(ERRCODE_OUT_OF_MEMORY),
350 : errmsg("could not parse connection string: %s",
351 : _("out of memory"))));
352 :
353 : /* build a clean connection string from pieces */
354 13566 : for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
355 : {
356 : bool obfuscate;
357 :
358 : /* Skip debug and empty options */
359 13300 : if (strchr(conn_opt->dispchar, 'D') ||
360 12236 : conn_opt->val == NULL ||
361 5076 : conn_opt->val[0] == '\0')
362 8490 : continue;
363 :
364 : /* Obfuscate security-sensitive options */
365 4810 : obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
366 :
367 9620 : appendPQExpBuffer(&buf, "%s%s=%s",
368 4810 : buf.len == 0 ? "" : " ",
369 : conn_opt->keyword,
370 : obfuscate ? "********" : conn_opt->val);
371 : }
372 :
373 266 : PQconninfoFree(conn_opts);
374 :
375 266 : retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
376 266 : termPQExpBuffer(&buf);
377 266 : return retval;
378 : }
379 :
380 : /*
381 : * Provides information of sender this WAL receiver is connected to.
382 : */
383 : static void
384 266 : libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host,
385 : int *sender_port)
386 : {
387 266 : char *ret = NULL;
388 :
389 266 : *sender_host = NULL;
390 266 : *sender_port = 0;
391 :
392 : Assert(conn->streamConn != NULL);
393 :
394 266 : ret = PQhost(conn->streamConn);
395 266 : if (ret && strlen(ret) != 0)
396 266 : *sender_host = pstrdup(ret);
397 :
398 266 : ret = PQport(conn->streamConn);
399 266 : if (ret && strlen(ret) != 0)
400 266 : *sender_port = atoi(ret);
401 266 : }
402 :
403 : /*
404 : * Check that primary's system identifier matches ours, and fetch the current
405 : * timeline ID of the primary.
406 : */
407 : static char *
408 638 : libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
409 : {
410 : PGresult *res;
411 : char *primary_sysid;
412 :
413 : /*
414 : * Get the system identifier and timeline ID as a DataRow message from the
415 : * primary server.
416 : */
417 638 : res = libpqsrv_exec(conn->streamConn,
418 : "IDENTIFY_SYSTEM",
419 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
420 638 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
421 : {
422 0 : PQclear(res);
423 0 : ereport(ERROR,
424 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
425 : errmsg("could not receive database system identifier and timeline ID from "
426 : "the primary server: %s",
427 : pchomp(PQerrorMessage(conn->streamConn)))));
428 : }
429 :
430 : /*
431 : * IDENTIFY_SYSTEM returns 3 columns in 9.3 and earlier, and 4 columns in
432 : * 9.4 and onwards.
433 : */
434 638 : if (PQnfields(res) < 3 || PQntuples(res) != 1)
435 : {
436 0 : int ntuples = PQntuples(res);
437 0 : int nfields = PQnfields(res);
438 :
439 0 : PQclear(res);
440 0 : ereport(ERROR,
441 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
442 : errmsg("invalid response from primary server"),
443 : errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
444 : ntuples, nfields, 1, 3)));
445 : }
446 638 : primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
447 638 : *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
448 638 : PQclear(res);
449 :
450 638 : return primary_sysid;
451 : }
452 :
453 : /*
454 : * Thin wrapper around libpq to obtain server version.
455 : */
456 : static int
457 1796 : libpqrcv_server_version(WalReceiverConn *conn)
458 : {
459 1796 : return PQserverVersion(conn->streamConn);
460 : }
461 :
462 : /*
463 : * Get database name from the primary server's conninfo.
464 : *
465 : * If dbname is not found in connInfo, return NULL value.
466 : */
467 : static char *
468 26 : libpqrcv_get_dbname_from_conninfo(const char *connInfo)
469 : {
470 : PQconninfoOption *opts;
471 26 : char *dbname = NULL;
472 26 : char *err = NULL;
473 :
474 26 : opts = PQconninfoParse(connInfo, &err);
475 26 : if (opts == NULL)
476 : {
477 : /* The error string is malloc'd, so we must free it explicitly */
478 0 : char *errcopy = err ? pstrdup(err) : "out of memory";
479 :
480 0 : PQfreemem(err);
481 0 : ereport(ERROR,
482 : (errcode(ERRCODE_SYNTAX_ERROR),
483 : errmsg("invalid connection string syntax: %s", errcopy)));
484 : }
485 :
486 1326 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
487 : {
488 : /*
489 : * If multiple dbnames are specified, then the last one will be
490 : * returned
491 : */
492 1300 : if (strcmp(opt->keyword, "dbname") == 0 && opt->val &&
493 24 : *opt->val)
494 : {
495 24 : if (dbname)
496 0 : pfree(dbname);
497 :
498 24 : dbname = pstrdup(opt->val);
499 : }
500 : }
501 :
502 26 : PQconninfoFree(opts);
503 26 : return dbname;
504 : }
505 :
506 : /*
507 : * Start streaming WAL data from given streaming options.
508 : *
509 : * Returns true if we switched successfully to copy-both mode. False
510 : * means the server received the command and executed it successfully, but
511 : * didn't switch to copy-mode. That means that there was no WAL on the
512 : * requested timeline and starting point, because the server switched to
513 : * another timeline at or before the requested starting point. On failure,
514 : * throws an ERROR.
515 : */
516 : static bool
517 1002 : libpqrcv_startstreaming(WalReceiverConn *conn,
518 : const WalRcvStreamOptions *options)
519 : {
520 : StringInfoData cmd;
521 : PGresult *res;
522 :
523 : Assert(options->logical == conn->logical);
524 : Assert(options->slotname || !options->logical);
525 :
526 1002 : initStringInfo(&cmd);
527 :
528 : /* Build the command. */
529 1002 : appendStringInfoString(&cmd, "START_REPLICATION");
530 1002 : if (options->slotname != NULL)
531 824 : appendStringInfo(&cmd, " SLOT \"%s\"",
532 824 : options->slotname);
533 :
534 1002 : if (options->logical)
535 736 : appendStringInfoString(&cmd, " LOGICAL");
536 :
537 1002 : appendStringInfo(&cmd, " %X/%X", LSN_FORMAT_ARGS(options->startpoint));
538 :
539 : /*
540 : * Additional options are different depending on if we are doing logical
541 : * or physical replication.
542 : */
543 1002 : if (options->logical)
544 : {
545 : char *pubnames_str;
546 : List *pubnames;
547 : char *pubnames_literal;
548 :
549 736 : appendStringInfoString(&cmd, " (");
550 :
551 736 : appendStringInfo(&cmd, "proto_version '%u'",
552 736 : options->proto.logical.proto_version);
553 :
554 736 : if (options->proto.logical.streaming_str)
555 720 : appendStringInfo(&cmd, ", streaming '%s'",
556 720 : options->proto.logical.streaming_str);
557 :
558 750 : if (options->proto.logical.twophase &&
559 14 : PQserverVersion(conn->streamConn) >= 150000)
560 14 : appendStringInfoString(&cmd, ", two_phase 'on'");
561 :
562 1472 : if (options->proto.logical.origin &&
563 736 : PQserverVersion(conn->streamConn) >= 160000)
564 736 : appendStringInfo(&cmd, ", origin '%s'",
565 736 : options->proto.logical.origin);
566 :
567 736 : pubnames = options->proto.logical.publication_names;
568 736 : pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
569 736 : if (!pubnames_str)
570 0 : ereport(ERROR,
571 : (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
572 : errmsg("could not start WAL streaming: %s",
573 : pchomp(PQerrorMessage(conn->streamConn)))));
574 736 : pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
575 : strlen(pubnames_str));
576 736 : if (!pubnames_literal)
577 0 : ereport(ERROR,
578 : (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
579 : errmsg("could not start WAL streaming: %s",
580 : pchomp(PQerrorMessage(conn->streamConn)))));
581 736 : appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
582 736 : PQfreemem(pubnames_literal);
583 736 : pfree(pubnames_str);
584 :
585 758 : if (options->proto.logical.binary &&
586 22 : PQserverVersion(conn->streamConn) >= 140000)
587 22 : appendStringInfoString(&cmd, ", binary 'true'");
588 :
589 736 : appendStringInfoChar(&cmd, ')');
590 : }
591 : else
592 266 : appendStringInfo(&cmd, " TIMELINE %u",
593 266 : options->proto.physical.startpointTLI);
594 :
595 : /* Start streaming. */
596 1002 : res = libpqsrv_exec(conn->streamConn,
597 1002 : cmd.data,
598 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
599 1002 : pfree(cmd.data);
600 :
601 1002 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
602 : {
603 0 : PQclear(res);
604 0 : return false;
605 : }
606 1002 : else if (PQresultStatus(res) != PGRES_COPY_BOTH)
607 : {
608 2 : PQclear(res);
609 2 : ereport(ERROR,
610 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
611 : errmsg("could not start WAL streaming: %s",
612 : pchomp(PQerrorMessage(conn->streamConn)))));
613 : }
614 1000 : PQclear(res);
615 1000 : return true;
616 : }
617 :
618 : /*
619 : * Stop streaming WAL data. Returns the next timeline's ID in *next_tli, as
620 : * reported by the server, or 0 if it did not report it.
621 : */
622 : static void
623 458 : libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
624 : {
625 : PGresult *res;
626 :
627 : /*
628 : * Send copy-end message. As in libpqsrv_exec, this could theoretically
629 : * block, but the risk seems small.
630 : */
631 846 : if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
632 388 : PQflush(conn->streamConn))
633 70 : ereport(ERROR,
634 : (errcode(ERRCODE_CONNECTION_FAILURE),
635 : errmsg("could not send end-of-streaming message to primary: %s",
636 : pchomp(PQerrorMessage(conn->streamConn)))));
637 :
638 388 : *next_tli = 0;
639 :
640 : /*
641 : * After COPY is finished, we should receive a result set indicating the
642 : * next timeline's ID, or just CommandComplete if the server was shut
643 : * down.
644 : *
645 : * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
646 : * also possible in case we aborted the copy in mid-stream.
647 : */
648 388 : res = libpqsrv_get_result(conn->streamConn,
649 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
650 388 : if (PQresultStatus(res) == PGRES_TUPLES_OK)
651 : {
652 : /*
653 : * Read the next timeline's ID. The server also sends the timeline's
654 : * starting point, but it is ignored.
655 : */
656 24 : if (PQnfields(res) < 2 || PQntuples(res) != 1)
657 0 : ereport(ERROR,
658 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
659 : errmsg("unexpected result set after end-of-streaming")));
660 24 : *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
661 24 : PQclear(res);
662 :
663 : /* the result set should be followed by CommandComplete */
664 24 : res = libpqsrv_get_result(conn->streamConn,
665 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
666 : }
667 364 : else if (PQresultStatus(res) == PGRES_COPY_OUT)
668 : {
669 364 : PQclear(res);
670 :
671 : /* End the copy */
672 364 : if (PQendcopy(conn->streamConn))
673 0 : ereport(ERROR,
674 : (errcode(ERRCODE_CONNECTION_FAILURE),
675 : errmsg("error while shutting down streaming COPY: %s",
676 : pchomp(PQerrorMessage(conn->streamConn)))));
677 :
678 : /* CommandComplete should follow */
679 364 : res = libpqsrv_get_result(conn->streamConn,
680 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
681 : }
682 :
683 388 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
684 0 : ereport(ERROR,
685 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
686 : errmsg("error reading result of streaming command: %s",
687 : pchomp(PQerrorMessage(conn->streamConn)))));
688 388 : PQclear(res);
689 :
690 : /* Verify that there are no more results */
691 388 : res = libpqsrv_get_result(conn->streamConn,
692 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
693 388 : if (res != NULL)
694 0 : ereport(ERROR,
695 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
696 : errmsg("unexpected result after CommandComplete: %s",
697 : pchomp(PQerrorMessage(conn->streamConn)))));
698 388 : }
699 :
700 : /*
701 : * Fetch the timeline history file for 'tli' from primary.
702 : */
703 : static void
704 22 : libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
705 : TimeLineID tli, char **filename,
706 : char **content, int *len)
707 : {
708 : PGresult *res;
709 : char cmd[64];
710 :
711 : Assert(!conn->logical);
712 :
713 : /*
714 : * Request the primary to send over the history file for given timeline.
715 : */
716 22 : snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
717 22 : res = libpqsrv_exec(conn->streamConn,
718 : cmd,
719 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
720 22 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
721 : {
722 0 : PQclear(res);
723 0 : ereport(ERROR,
724 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
725 : errmsg("could not receive timeline history file from "
726 : "the primary server: %s",
727 : pchomp(PQerrorMessage(conn->streamConn)))));
728 : }
729 22 : if (PQnfields(res) != 2 || PQntuples(res) != 1)
730 : {
731 0 : int ntuples = PQntuples(res);
732 0 : int nfields = PQnfields(res);
733 :
734 0 : PQclear(res);
735 0 : ereport(ERROR,
736 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
737 : errmsg("invalid response from primary server"),
738 : errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
739 : ntuples, nfields)));
740 : }
741 22 : *filename = pstrdup(PQgetvalue(res, 0, 0));
742 :
743 22 : *len = PQgetlength(res, 0, 1);
744 22 : *content = palloc(*len);
745 22 : memcpy(*content, PQgetvalue(res, 0, 1), *len);
746 22 : PQclear(res);
747 22 : }
748 :
749 : /*
750 : * Disconnect connection to primary, if any.
751 : */
752 : static void
753 1490 : libpqrcv_disconnect(WalReceiverConn *conn)
754 : {
755 1490 : libpqsrv_disconnect(conn->streamConn);
756 1490 : PQfreemem(conn->recvBuf);
757 1490 : pfree(conn);
758 1490 : }
759 :
760 : /*
761 : * Receive a message available from XLOG stream.
762 : *
763 : * Returns:
764 : *
765 : * If data was received, returns the length of the data. *buffer is set to
766 : * point to a buffer holding the received message. The buffer is only valid
767 : * until the next libpqrcv_* call.
768 : *
769 : * If no data was available immediately, returns 0, and *wait_fd is set to a
770 : * socket descriptor which can be waited on before trying again.
771 : *
772 : * -1 if the server ended the COPY.
773 : *
774 : * ereports on error.
775 : */
776 : static int
777 772922 : libpqrcv_receive(WalReceiverConn *conn, char **buffer,
778 : pgsocket *wait_fd)
779 : {
780 : int rawlen;
781 :
782 772922 : PQfreemem(conn->recvBuf);
783 772922 : conn->recvBuf = NULL;
784 :
785 : /* Try to receive a CopyData message */
786 772922 : rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
787 772922 : if (rawlen == 0)
788 : {
789 : /* Try consuming some data. */
790 504026 : if (PQconsumeInput(conn->streamConn) == 0)
791 70 : ereport(ERROR,
792 : (errcode(ERRCODE_CONNECTION_FAILURE),
793 : errmsg("could not receive data from WAL stream: %s",
794 : pchomp(PQerrorMessage(conn->streamConn)))));
795 :
796 : /* Now that we've consumed some input, try again */
797 503956 : rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
798 503956 : if (rawlen == 0)
799 : {
800 : /* Tell caller to try again when our socket is ready. */
801 184024 : *wait_fd = PQsocket(conn->streamConn);
802 184024 : return 0;
803 : }
804 : }
805 588828 : if (rawlen == -1) /* end-of-streaming or error */
806 : {
807 : PGresult *res;
808 :
809 476 : res = libpqsrv_get_result(conn->streamConn,
810 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
811 476 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
812 : {
813 446 : PQclear(res);
814 :
815 : /* Verify that there are no more results. */
816 446 : res = libpqsrv_get_result(conn->streamConn,
817 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
818 446 : if (res != NULL)
819 : {
820 70 : PQclear(res);
821 :
822 : /*
823 : * If the other side closed the connection orderly (otherwise
824 : * we'd seen an error, or PGRES_COPY_IN) don't report an error
825 : * here, but let callers deal with it.
826 : */
827 70 : if (PQstatus(conn->streamConn) == CONNECTION_BAD)
828 70 : return -1;
829 :
830 0 : ereport(ERROR,
831 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
832 : errmsg("unexpected result after CommandComplete: %s",
833 : PQerrorMessage(conn->streamConn))));
834 : }
835 :
836 376 : return -1;
837 : }
838 30 : else if (PQresultStatus(res) == PGRES_COPY_IN)
839 : {
840 24 : PQclear(res);
841 24 : return -1;
842 : }
843 : else
844 : {
845 6 : PQclear(res);
846 6 : ereport(ERROR,
847 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
848 : errmsg("could not receive data from WAL stream: %s",
849 : pchomp(PQerrorMessage(conn->streamConn)))));
850 : }
851 : }
852 588352 : if (rawlen < -1)
853 0 : ereport(ERROR,
854 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
855 : errmsg("could not receive data from WAL stream: %s",
856 : pchomp(PQerrorMessage(conn->streamConn)))));
857 :
858 : /* Return received messages to caller */
859 588352 : *buffer = conn->recvBuf;
860 588352 : return rawlen;
861 : }
862 :
863 : /*
864 : * Send a message to XLOG stream.
865 : *
866 : * ereports on error.
867 : */
868 : static void
869 193364 : libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
870 : {
871 386728 : if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
872 193364 : PQflush(conn->streamConn))
873 0 : ereport(ERROR,
874 : (errcode(ERRCODE_CONNECTION_FAILURE),
875 : errmsg("could not send data to WAL stream: %s",
876 : pchomp(PQerrorMessage(conn->streamConn)))));
877 193364 : }
878 :
879 : /*
880 : * Create new replication slot.
881 : * Returns the name of the exported snapshot for logical slot or NULL for
882 : * physical slot.
883 : */
884 : static char *
885 604 : libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
886 : bool temporary, bool two_phase, bool failover,
887 : CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
888 : {
889 : PGresult *res;
890 : StringInfoData cmd;
891 : char *snapshot;
892 : int use_new_options_syntax;
893 :
894 604 : use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
895 :
896 604 : initStringInfo(&cmd);
897 :
898 604 : appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
899 :
900 604 : if (temporary)
901 0 : appendStringInfoString(&cmd, " TEMPORARY");
902 :
903 604 : if (conn->logical)
904 : {
905 604 : appendStringInfoString(&cmd, " LOGICAL pgoutput ");
906 604 : if (use_new_options_syntax)
907 604 : appendStringInfoChar(&cmd, '(');
908 604 : if (two_phase)
909 : {
910 2 : appendStringInfoString(&cmd, "TWO_PHASE");
911 2 : if (use_new_options_syntax)
912 2 : appendStringInfoString(&cmd, ", ");
913 : else
914 0 : appendStringInfoChar(&cmd, ' ');
915 : }
916 :
917 604 : if (failover)
918 : {
919 12 : appendStringInfoString(&cmd, "FAILOVER");
920 12 : if (use_new_options_syntax)
921 12 : appendStringInfoString(&cmd, ", ");
922 : else
923 0 : appendStringInfoChar(&cmd, ' ');
924 : }
925 :
926 604 : if (use_new_options_syntax)
927 : {
928 604 : switch (snapshot_action)
929 : {
930 0 : case CRS_EXPORT_SNAPSHOT:
931 0 : appendStringInfoString(&cmd, "SNAPSHOT 'export'");
932 0 : break;
933 218 : case CRS_NOEXPORT_SNAPSHOT:
934 218 : appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
935 218 : break;
936 386 : case CRS_USE_SNAPSHOT:
937 386 : appendStringInfoString(&cmd, "SNAPSHOT 'use'");
938 386 : break;
939 : }
940 : }
941 : else
942 : {
943 0 : switch (snapshot_action)
944 : {
945 0 : case CRS_EXPORT_SNAPSHOT:
946 0 : appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
947 0 : break;
948 0 : case CRS_NOEXPORT_SNAPSHOT:
949 0 : appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
950 0 : break;
951 0 : case CRS_USE_SNAPSHOT:
952 0 : appendStringInfoString(&cmd, "USE_SNAPSHOT");
953 0 : break;
954 : }
955 : }
956 :
957 604 : if (use_new_options_syntax)
958 604 : appendStringInfoChar(&cmd, ')');
959 : }
960 : else
961 : {
962 0 : if (use_new_options_syntax)
963 0 : appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
964 : else
965 0 : appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
966 : }
967 :
968 604 : res = libpqsrv_exec(conn->streamConn,
969 604 : cmd.data,
970 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
971 604 : pfree(cmd.data);
972 :
973 604 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
974 : {
975 0 : PQclear(res);
976 0 : ereport(ERROR,
977 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
978 : errmsg("could not create replication slot \"%s\": %s",
979 : slotname, pchomp(PQerrorMessage(conn->streamConn)))));
980 : }
981 :
982 604 : if (lsn)
983 386 : *lsn = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid,
984 386 : CStringGetDatum(PQgetvalue(res, 0, 1))));
985 :
986 604 : if (!PQgetisnull(res, 0, 2))
987 0 : snapshot = pstrdup(PQgetvalue(res, 0, 2));
988 : else
989 604 : snapshot = NULL;
990 :
991 604 : PQclear(res);
992 :
993 604 : return snapshot;
994 : }
995 :
996 : /*
997 : * Change the definition of the replication slot.
998 : */
999 : static void
1000 8 : libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
1001 : const bool *failover, const bool *two_phase)
1002 : {
1003 : StringInfoData cmd;
1004 : PGresult *res;
1005 :
1006 8 : initStringInfo(&cmd);
1007 8 : appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
1008 : quote_identifier(slotname));
1009 :
1010 8 : if (failover)
1011 6 : appendStringInfo(&cmd, "FAILOVER %s",
1012 6 : *failover ? "true" : "false");
1013 :
1014 8 : if (failover && two_phase)
1015 0 : appendStringInfoString(&cmd, ", ");
1016 :
1017 8 : if (two_phase)
1018 2 : appendStringInfo(&cmd, "TWO_PHASE %s",
1019 2 : *two_phase ? "true" : "false");
1020 :
1021 8 : appendStringInfoString(&cmd, " );");
1022 :
1023 8 : res = libpqsrv_exec(conn->streamConn, cmd.data,
1024 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
1025 8 : pfree(cmd.data);
1026 :
1027 8 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1028 0 : ereport(ERROR,
1029 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
1030 : errmsg("could not alter replication slot \"%s\": %s",
1031 : slotname, pchomp(PQerrorMessage(conn->streamConn)))));
1032 :
1033 8 : PQclear(res);
1034 8 : }
1035 :
1036 : /*
1037 : * Return PID of remote backend process.
1038 : */
1039 : static pid_t
1040 0 : libpqrcv_get_backend_pid(WalReceiverConn *conn)
1041 : {
1042 0 : return PQbackendPID(conn->streamConn);
1043 : }
1044 :
1045 : /*
1046 : * Convert tuple query result to tuplestore.
1047 : */
1048 : static void
1049 2160 : libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
1050 : const int nRetTypes, const Oid *retTypes)
1051 : {
1052 : int tupn;
1053 : int coln;
1054 2160 : int nfields = PQnfields(pgres);
1055 : HeapTuple tuple;
1056 : AttInMetadata *attinmeta;
1057 : MemoryContext rowcontext;
1058 : MemoryContext oldcontext;
1059 :
1060 : /* Make sure we got expected number of fields. */
1061 2160 : if (nfields != nRetTypes)
1062 0 : ereport(ERROR,
1063 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
1064 : errmsg("invalid query response"),
1065 : errdetail("Expected %d fields, got %d fields.",
1066 : nRetTypes, nfields)));
1067 :
1068 2160 : walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
1069 :
1070 : /* Create tuple descriptor corresponding to expected result. */
1071 2160 : walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
1072 7600 : for (coln = 0; coln < nRetTypes; coln++)
1073 5440 : TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
1074 5440 : PQfname(pgres, coln), retTypes[coln], -1, 0);
1075 2160 : attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
1076 :
1077 : /* No point in doing more here if there were no tuples returned. */
1078 2160 : if (PQntuples(pgres) == 0)
1079 36 : return;
1080 :
1081 : /* Create temporary context for local allocations. */
1082 2124 : rowcontext = AllocSetContextCreate(CurrentMemoryContext,
1083 : "libpqrcv query result context",
1084 : ALLOCSET_DEFAULT_SIZES);
1085 :
1086 : /* Process returned rows. */
1087 4942 : for (tupn = 0; tupn < PQntuples(pgres); tupn++)
1088 : {
1089 : char *cstrs[MaxTupleAttributeNumber];
1090 :
1091 2818 : CHECK_FOR_INTERRUPTS();
1092 :
1093 : /* Do the allocations in temporary context. */
1094 2818 : oldcontext = MemoryContextSwitchTo(rowcontext);
1095 :
1096 : /*
1097 : * Fill cstrs with null-terminated strings of column values.
1098 : */
1099 11022 : for (coln = 0; coln < nfields; coln++)
1100 : {
1101 8204 : if (PQgetisnull(pgres, tupn, coln))
1102 1148 : cstrs[coln] = NULL;
1103 : else
1104 7056 : cstrs[coln] = PQgetvalue(pgres, tupn, coln);
1105 : }
1106 :
1107 : /* Convert row to a tuple, and add it to the tuplestore */
1108 2818 : tuple = BuildTupleFromCStrings(attinmeta, cstrs);
1109 2818 : tuplestore_puttuple(walres->tuplestore, tuple);
1110 :
1111 : /* Clean up */
1112 2818 : MemoryContextSwitchTo(oldcontext);
1113 2818 : MemoryContextReset(rowcontext);
1114 : }
1115 :
1116 2124 : MemoryContextDelete(rowcontext);
1117 : }
1118 :
1119 : /*
1120 : * Public interface for sending generic queries (and commands).
1121 : *
1122 : * This can only be called from process connected to database.
1123 : */
1124 : static WalRcvExecResult *
1125 3814 : libpqrcv_exec(WalReceiverConn *conn, const char *query,
1126 : const int nRetTypes, const Oid *retTypes)
1127 : {
1128 3814 : PGresult *pgres = NULL;
1129 3814 : WalRcvExecResult *walres = palloc0(sizeof(WalRcvExecResult));
1130 : char *diag_sqlstate;
1131 :
1132 3814 : if (MyDatabaseId == InvalidOid)
1133 0 : ereport(ERROR,
1134 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1135 : errmsg("the query interface requires a database connection")));
1136 :
1137 3814 : pgres = libpqsrv_exec(conn->streamConn,
1138 : query,
1139 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
1140 :
1141 3814 : switch (PQresultStatus(pgres))
1142 : {
1143 2160 : case PGRES_TUPLES_OK:
1144 : case PGRES_SINGLE_TUPLE:
1145 : case PGRES_TUPLES_CHUNK:
1146 2160 : walres->status = WALRCV_OK_TUPLES;
1147 2160 : libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
1148 2160 : break;
1149 :
1150 0 : case PGRES_COPY_IN:
1151 0 : walres->status = WALRCV_OK_COPY_IN;
1152 0 : break;
1153 :
1154 380 : case PGRES_COPY_OUT:
1155 380 : walres->status = WALRCV_OK_COPY_OUT;
1156 380 : break;
1157 :
1158 0 : case PGRES_COPY_BOTH:
1159 0 : walres->status = WALRCV_OK_COPY_BOTH;
1160 0 : break;
1161 :
1162 1270 : case PGRES_COMMAND_OK:
1163 1270 : walres->status = WALRCV_OK_COMMAND;
1164 1270 : break;
1165 :
1166 : /* Empty query is considered error. */
1167 0 : case PGRES_EMPTY_QUERY:
1168 0 : walres->status = WALRCV_ERROR;
1169 0 : walres->err = _("empty query");
1170 0 : break;
1171 :
1172 0 : case PGRES_PIPELINE_SYNC:
1173 : case PGRES_PIPELINE_ABORTED:
1174 0 : walres->status = WALRCV_ERROR;
1175 0 : walres->err = _("unexpected pipeline mode");
1176 0 : break;
1177 :
1178 4 : case PGRES_NONFATAL_ERROR:
1179 : case PGRES_FATAL_ERROR:
1180 : case PGRES_BAD_RESPONSE:
1181 4 : walres->status = WALRCV_ERROR;
1182 4 : walres->err = pchomp(PQerrorMessage(conn->streamConn));
1183 4 : diag_sqlstate = PQresultErrorField(pgres, PG_DIAG_SQLSTATE);
1184 4 : if (diag_sqlstate)
1185 4 : walres->sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
1186 : diag_sqlstate[1],
1187 : diag_sqlstate[2],
1188 : diag_sqlstate[3],
1189 : diag_sqlstate[4]);
1190 4 : break;
1191 : }
1192 :
1193 3814 : PQclear(pgres);
1194 :
1195 3814 : return walres;
1196 : }
1197 :
1198 : /*
1199 : * Given a List of strings, return it as single comma separated
1200 : * string, quoting identifiers as needed.
1201 : *
1202 : * This is essentially the reverse of SplitIdentifierString.
1203 : *
1204 : * The caller should free the result.
1205 : */
1206 : static char *
1207 736 : stringlist_to_identifierstr(PGconn *conn, List *strings)
1208 : {
1209 : ListCell *lc;
1210 : StringInfoData res;
1211 736 : bool first = true;
1212 :
1213 736 : initStringInfo(&res);
1214 :
1215 1964 : foreach(lc, strings)
1216 : {
1217 1228 : char *val = strVal(lfirst(lc));
1218 : char *val_escaped;
1219 :
1220 1228 : if (first)
1221 736 : first = false;
1222 : else
1223 492 : appendStringInfoChar(&res, ',');
1224 :
1225 1228 : val_escaped = PQescapeIdentifier(conn, val, strlen(val));
1226 1228 : if (!val_escaped)
1227 : {
1228 0 : free(res.data);
1229 0 : return NULL;
1230 : }
1231 1228 : appendStringInfoString(&res, val_escaped);
1232 1228 : PQfreemem(val_escaped);
1233 : }
1234 :
1235 736 : return res.data;
1236 : }
|
