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 1816 : 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 1816 : _PG_init(void)
124 : {
125 1816 : if (WalReceiverFunctions != NULL)
126 0 : elog(ERROR, "libpqwalreceiver already loaded");
127 1816 : WalReceiverFunctions = &PQWalReceiverFunctions;
128 1816 : }
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 1810 : 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 1810 : 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 1810 : 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 1784 : keys[i] = "dbname";
168 1784 : vals[i] = conninfo;
169 :
170 : /* We can not have logical without replication */
171 : Assert(replication || !logical);
172 :
173 1784 : if (replication)
174 : {
175 1760 : keys[++i] = "replication";
176 1760 : vals[i] = logical ? "database" : "true";
177 :
178 1760 : if (logical)
179 : {
180 : /* Tell the publisher to translate to our encoding */
181 1294 : keys[++i] = "client_encoding";
182 1294 : 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 1294 : keys[++i] = "options";
193 1294 : 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 466 : keys[++i] = "dbname";
202 466 : vals[i] = "replication";
203 : }
204 : }
205 :
206 1784 : keys[++i] = "fallback_application_name";
207 1784 : vals[i] = appname;
208 :
209 1784 : keys[++i] = NULL;
210 1784 : vals[i] = NULL;
211 :
212 : Assert(i < lengthof(keys));
213 :
214 1784 : conn = palloc0(sizeof(WalReceiverConn));
215 1782 : conn->streamConn =
216 1784 : libpqsrv_connect_params(keys, vals,
217 : /* expand_dbname = */ true,
218 : WAIT_EVENT_LIBPQWALRECEIVER_CONNECT);
219 :
220 1782 : if (PQstatus(conn->streamConn) != CONNECTION_OK)
221 238 : goto bad_connection_errmsg;
222 :
223 1544 : 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 1544 : PQsetNoticeReceiver(conn->streamConn, libpqsrv_notice_receiver,
236 : "received message via replication");
237 :
238 : /*
239 : * Set always-secure search path for the cases where the connection is
240 : * used to run SQL queries, so malicious users can't get control.
241 : */
242 1544 : if (!replication || logical)
243 : {
244 : PGresult *res;
245 :
246 1268 : res = libpqsrv_exec(conn->streamConn,
247 : ALWAYS_SECURE_SEARCH_PATH_SQL,
248 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
249 1268 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
250 : {
251 0 : PQclear(res);
252 0 : *err = psprintf(_("could not clear search path: %s"),
253 0 : pchomp(PQerrorMessage(conn->streamConn)));
254 0 : goto bad_connection;
255 : }
256 1268 : PQclear(res);
257 : }
258 :
259 1544 : conn->logical = logical;
260 :
261 1544 : return conn;
262 :
263 : /* error path, using libpq's error message */
264 238 : bad_connection_errmsg:
265 238 : *err = pchomp(PQerrorMessage(conn->streamConn));
266 :
267 : /* error path, error already set */
268 238 : bad_connection:
269 238 : libpqsrv_disconnect(conn->streamConn);
270 238 : pfree(conn);
271 238 : return NULL;
272 : }
273 :
274 : /*
275 : * Validate connection info string.
276 : *
277 : * If the connection string can't be parsed, this function will raise
278 : * an error. If must_use_password is true, the function raises an error
279 : * if no password is provided in the connection string. In any other case
280 : * it successfully completes.
281 : */
282 : static void
283 2182 : libpqrcv_check_conninfo(const char *conninfo, bool must_use_password)
284 : {
285 2182 : PQconninfoOption *opts = NULL;
286 : PQconninfoOption *opt;
287 2182 : char *err = NULL;
288 :
289 2182 : opts = PQconninfoParse(conninfo, &err);
290 2182 : if (opts == NULL)
291 : {
292 : /* The error string is malloc'd, so we must free it explicitly */
293 18 : char *errcopy = err ? pstrdup(err) : "out of memory";
294 :
295 18 : PQfreemem(err);
296 18 : ereport(ERROR,
297 : (errcode(ERRCODE_SYNTAX_ERROR),
298 : errmsg("invalid connection string syntax: %s", errcopy)));
299 : }
300 :
301 2164 : if (must_use_password)
302 : {
303 44 : bool uses_password = false;
304 :
305 1712 : for (opt = opts; opt->keyword != NULL; ++opt)
306 : {
307 : /* Ignore connection options that are not present. */
308 1680 : if (opt->val == NULL)
309 1574 : continue;
310 :
311 106 : if (strcmp(opt->keyword, "password") == 0 && opt->val[0] != '\0')
312 : {
313 12 : uses_password = true;
314 12 : break;
315 : }
316 : }
317 :
318 44 : if (!uses_password)
319 : {
320 : /* malloc'd, so we must free it explicitly */
321 32 : PQconninfoFree(opts);
322 :
323 32 : ereport(ERROR,
324 : (errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
325 : errmsg("password is required"),
326 : errdetail("Non-superusers must provide a password in the connection string.")));
327 : }
328 : }
329 :
330 2132 : PQconninfoFree(opts);
331 2132 : }
332 :
333 : /*
334 : * Return a user-displayable conninfo string. Any security-sensitive fields
335 : * are obfuscated.
336 : */
337 : static char *
338 276 : libpqrcv_get_conninfo(WalReceiverConn *conn)
339 : {
340 : PQconninfoOption *conn_opts;
341 : PQconninfoOption *conn_opt;
342 : PQExpBufferData buf;
343 : char *retval;
344 :
345 : Assert(conn->streamConn != NULL);
346 :
347 276 : initPQExpBuffer(&buf);
348 276 : conn_opts = PQconninfo(conn->streamConn);
349 :
350 276 : if (conn_opts == NULL)
351 0 : ereport(ERROR,
352 : (errcode(ERRCODE_OUT_OF_MEMORY),
353 : errmsg("could not parse connection string: %s",
354 : _("out of memory"))));
355 :
356 : /* build a clean connection string from pieces */
357 14352 : for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
358 : {
359 : bool obfuscate;
360 :
361 : /* Skip debug and empty options */
362 14076 : if (strchr(conn_opt->dispchar, 'D') ||
363 12972 : conn_opt->val == NULL ||
364 5266 : conn_opt->val[0] == '\0')
365 9086 : continue;
366 :
367 : /* Obfuscate security-sensitive options */
368 4990 : obfuscate = strchr(conn_opt->dispchar, '*') != NULL;
369 :
370 9980 : appendPQExpBuffer(&buf, "%s%s=%s",
371 4990 : buf.len == 0 ? "" : " ",
372 : conn_opt->keyword,
373 : obfuscate ? "********" : conn_opt->val);
374 : }
375 :
376 276 : PQconninfoFree(conn_opts);
377 :
378 276 : retval = PQExpBufferDataBroken(buf) ? NULL : pstrdup(buf.data);
379 276 : termPQExpBuffer(&buf);
380 276 : return retval;
381 : }
382 :
383 : /*
384 : * Provides information of sender this WAL receiver is connected to.
385 : */
386 : static void
387 276 : libpqrcv_get_senderinfo(WalReceiverConn *conn, char **sender_host,
388 : int *sender_port)
389 : {
390 276 : char *ret = NULL;
391 :
392 276 : *sender_host = NULL;
393 276 : *sender_port = 0;
394 :
395 : Assert(conn->streamConn != NULL);
396 :
397 276 : ret = PQhost(conn->streamConn);
398 276 : if (ret && strlen(ret) != 0)
399 276 : *sender_host = pstrdup(ret);
400 :
401 276 : ret = PQport(conn->streamConn);
402 276 : if (ret && strlen(ret) != 0)
403 276 : *sender_port = atoi(ret);
404 276 : }
405 :
406 : /*
407 : * Check that primary's system identifier matches ours, and fetch the current
408 : * timeline ID of the primary.
409 : */
410 : static char *
411 672 : libpqrcv_identify_system(WalReceiverConn *conn, TimeLineID *primary_tli)
412 : {
413 : PGresult *res;
414 : char *primary_sysid;
415 :
416 : /*
417 : * Get the system identifier and timeline ID as a DataRow message from the
418 : * primary server.
419 : */
420 672 : res = libpqsrv_exec(conn->streamConn,
421 : "IDENTIFY_SYSTEM",
422 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
423 672 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
424 0 : ereport(ERROR,
425 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
426 : errmsg("could not receive database system identifier and timeline ID from "
427 : "the primary server: %s",
428 : pchomp(PQerrorMessage(conn->streamConn)))));
429 :
430 : /*
431 : * IDENTIFY_SYSTEM returns 3 columns in 9.3 and earlier, and 4 columns in
432 : * 9.4 and onwards.
433 : */
434 672 : if (PQnfields(res) < 3 || PQntuples(res) != 1)
435 0 : ereport(ERROR,
436 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
437 : errmsg("invalid response from primary server"),
438 : errdetail("Could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields.",
439 : PQntuples(res), PQnfields(res), 1, 3)));
440 672 : primary_sysid = pstrdup(PQgetvalue(res, 0, 0));
441 672 : *primary_tli = pg_strtoint32(PQgetvalue(res, 0, 1));
442 672 : PQclear(res);
443 :
444 672 : return primary_sysid;
445 : }
446 :
447 : /*
448 : * Thin wrapper around libpq to obtain server version.
449 : */
450 : static int
451 1848 : libpqrcv_server_version(WalReceiverConn *conn)
452 : {
453 1848 : return PQserverVersion(conn->streamConn);
454 : }
455 :
456 : /*
457 : * Get database name from the primary server's conninfo.
458 : *
459 : * If dbname is not found in connInfo, return NULL value.
460 : */
461 : static char *
462 26 : libpqrcv_get_dbname_from_conninfo(const char *connInfo)
463 : {
464 : PQconninfoOption *opts;
465 26 : char *dbname = NULL;
466 26 : char *err = NULL;
467 :
468 26 : opts = PQconninfoParse(connInfo, &err);
469 26 : if (opts == NULL)
470 : {
471 : /* The error string is malloc'd, so we must free it explicitly */
472 0 : char *errcopy = err ? pstrdup(err) : "out of memory";
473 :
474 0 : PQfreemem(err);
475 0 : ereport(ERROR,
476 : (errcode(ERRCODE_SYNTAX_ERROR),
477 : errmsg("invalid connection string syntax: %s", errcopy)));
478 : }
479 :
480 1352 : for (PQconninfoOption *opt = opts; opt->keyword != NULL; ++opt)
481 : {
482 : /*
483 : * If multiple dbnames are specified, then the last one will be
484 : * returned
485 : */
486 1326 : if (strcmp(opt->keyword, "dbname") == 0 && opt->val &&
487 24 : *opt->val)
488 : {
489 24 : if (dbname)
490 0 : pfree(dbname);
491 :
492 24 : dbname = pstrdup(opt->val);
493 : }
494 : }
495 :
496 26 : PQconninfoFree(opts);
497 26 : return dbname;
498 : }
499 :
500 : /*
501 : * Start streaming WAL data from given streaming options.
502 : *
503 : * Returns true if we switched successfully to copy-both mode. False
504 : * means the server received the command and executed it successfully, but
505 : * didn't switch to copy-mode. That means that there was no WAL on the
506 : * requested timeline and starting point, because the server switched to
507 : * another timeline at or before the requested starting point. On failure,
508 : * throws an ERROR.
509 : */
510 : static bool
511 1038 : libpqrcv_startstreaming(WalReceiverConn *conn,
512 : const WalRcvStreamOptions *options)
513 : {
514 : StringInfoData cmd;
515 : PGresult *res;
516 :
517 : Assert(options->logical == conn->logical);
518 : Assert(options->slotname || !options->logical);
519 :
520 1038 : initStringInfo(&cmd);
521 :
522 : /* Build the command. */
523 1038 : appendStringInfoString(&cmd, "START_REPLICATION");
524 1038 : if (options->slotname != NULL)
525 850 : appendStringInfo(&cmd, " SLOT \"%s\"",
526 850 : options->slotname);
527 :
528 1038 : if (options->logical)
529 762 : appendStringInfoString(&cmd, " LOGICAL");
530 :
531 1038 : appendStringInfo(&cmd, " %X/%08X", LSN_FORMAT_ARGS(options->startpoint));
532 :
533 : /*
534 : * Additional options are different depending on if we are doing logical
535 : * or physical replication.
536 : */
537 1038 : if (options->logical)
538 : {
539 : char *pubnames_str;
540 : List *pubnames;
541 : char *pubnames_literal;
542 :
543 762 : appendStringInfoString(&cmd, " (");
544 :
545 762 : appendStringInfo(&cmd, "proto_version '%u'",
546 762 : options->proto.logical.proto_version);
547 :
548 762 : if (options->proto.logical.streaming_str)
549 746 : appendStringInfo(&cmd, ", streaming '%s'",
550 746 : options->proto.logical.streaming_str);
551 :
552 778 : if (options->proto.logical.twophase &&
553 16 : PQserverVersion(conn->streamConn) >= 150000)
554 16 : appendStringInfoString(&cmd, ", two_phase 'on'");
555 :
556 1524 : if (options->proto.logical.origin &&
557 762 : PQserverVersion(conn->streamConn) >= 160000)
558 762 : appendStringInfo(&cmd, ", origin '%s'",
559 762 : options->proto.logical.origin);
560 :
561 762 : pubnames = options->proto.logical.publication_names;
562 762 : pubnames_str = stringlist_to_identifierstr(conn->streamConn, pubnames);
563 762 : if (!pubnames_str)
564 0 : ereport(ERROR,
565 : (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
566 : errmsg("could not start WAL streaming: %s",
567 : pchomp(PQerrorMessage(conn->streamConn)))));
568 762 : pubnames_literal = PQescapeLiteral(conn->streamConn, pubnames_str,
569 : strlen(pubnames_str));
570 762 : if (!pubnames_literal)
571 0 : ereport(ERROR,
572 : (errcode(ERRCODE_OUT_OF_MEMORY), /* likely guess */
573 : errmsg("could not start WAL streaming: %s",
574 : pchomp(PQerrorMessage(conn->streamConn)))));
575 762 : appendStringInfo(&cmd, ", publication_names %s", pubnames_literal);
576 762 : PQfreemem(pubnames_literal);
577 762 : pfree(pubnames_str);
578 :
579 782 : if (options->proto.logical.binary &&
580 20 : PQserverVersion(conn->streamConn) >= 140000)
581 20 : appendStringInfoString(&cmd, ", binary 'true'");
582 :
583 762 : appendStringInfoChar(&cmd, ')');
584 : }
585 : else
586 276 : appendStringInfo(&cmd, " TIMELINE %u",
587 276 : options->proto.physical.startpointTLI);
588 :
589 : /* Start streaming. */
590 1038 : res = libpqsrv_exec(conn->streamConn,
591 1038 : cmd.data,
592 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
593 1038 : pfree(cmd.data);
594 :
595 1038 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
596 : {
597 0 : PQclear(res);
598 0 : return false;
599 : }
600 1038 : else if (PQresultStatus(res) != PGRES_COPY_BOTH)
601 2 : ereport(ERROR,
602 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
603 : errmsg("could not start WAL streaming: %s",
604 : pchomp(PQerrorMessage(conn->streamConn)))));
605 1036 : PQclear(res);
606 1036 : return true;
607 : }
608 :
609 : /*
610 : * Stop streaming WAL data. Returns the next timeline's ID in *next_tli, as
611 : * reported by the server, or 0 if it did not report it.
612 : */
613 : static void
614 458 : libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
615 : {
616 : PGresult *res;
617 :
618 : /*
619 : * Send copy-end message. As in libpqsrv_exec, this could theoretically
620 : * block, but the risk seems small.
621 : */
622 846 : if (PQputCopyEnd(conn->streamConn, NULL) <= 0 ||
623 388 : PQflush(conn->streamConn))
624 70 : ereport(ERROR,
625 : (errcode(ERRCODE_CONNECTION_FAILURE),
626 : errmsg("could not send end-of-streaming message to primary: %s",
627 : pchomp(PQerrorMessage(conn->streamConn)))));
628 :
629 388 : *next_tli = 0;
630 :
631 : /*
632 : * After COPY is finished, we should receive a result set indicating the
633 : * next timeline's ID, or just CommandComplete if the server was shut
634 : * down.
635 : *
636 : * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
637 : * also possible in case we aborted the copy in mid-stream.
638 : */
639 388 : res = libpqsrv_get_result(conn->streamConn,
640 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
641 388 : if (PQresultStatus(res) == PGRES_TUPLES_OK)
642 : {
643 : /*
644 : * Read the next timeline's ID. The server also sends the timeline's
645 : * starting point, but it is ignored.
646 : */
647 22 : if (PQnfields(res) < 2 || PQntuples(res) != 1)
648 0 : ereport(ERROR,
649 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
650 : errmsg("unexpected result set after end-of-streaming")));
651 22 : *next_tli = pg_strtoint32(PQgetvalue(res, 0, 0));
652 22 : PQclear(res);
653 :
654 : /* the result set should be followed by CommandComplete */
655 22 : res = libpqsrv_get_result(conn->streamConn,
656 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
657 : }
658 366 : else if (PQresultStatus(res) == PGRES_COPY_OUT)
659 : {
660 366 : PQclear(res);
661 :
662 : /* End the copy */
663 366 : if (PQendcopy(conn->streamConn))
664 0 : ereport(ERROR,
665 : (errcode(ERRCODE_CONNECTION_FAILURE),
666 : errmsg("error while shutting down streaming COPY: %s",
667 : pchomp(PQerrorMessage(conn->streamConn)))));
668 :
669 : /* CommandComplete should follow */
670 366 : res = libpqsrv_get_result(conn->streamConn,
671 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
672 : }
673 :
674 388 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
675 0 : ereport(ERROR,
676 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
677 : errmsg("error reading result of streaming command: %s",
678 : pchomp(PQerrorMessage(conn->streamConn)))));
679 388 : PQclear(res);
680 :
681 : /* Verify that there are no more results */
682 388 : res = libpqsrv_get_result(conn->streamConn,
683 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
684 388 : if (res != NULL)
685 0 : ereport(ERROR,
686 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
687 : errmsg("unexpected result after CommandComplete: %s",
688 : pchomp(PQerrorMessage(conn->streamConn)))));
689 388 : }
690 :
691 : /*
692 : * Fetch the timeline history file for 'tli' from primary.
693 : */
694 : static void
695 20 : libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
696 : TimeLineID tli, char **filename,
697 : char **content, int *len)
698 : {
699 : PGresult *res;
700 : char cmd[64];
701 :
702 : Assert(!conn->logical);
703 :
704 : /*
705 : * Request the primary to send over the history file for given timeline.
706 : */
707 20 : snprintf(cmd, sizeof(cmd), "TIMELINE_HISTORY %u", tli);
708 20 : res = libpqsrv_exec(conn->streamConn,
709 : cmd,
710 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
711 20 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
712 0 : ereport(ERROR,
713 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
714 : errmsg("could not receive timeline history file from "
715 : "the primary server: %s",
716 : pchomp(PQerrorMessage(conn->streamConn)))));
717 20 : if (PQnfields(res) != 2 || PQntuples(res) != 1)
718 0 : ereport(ERROR,
719 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
720 : errmsg("invalid response from primary server"),
721 : errdetail("Expected 1 tuple with 2 fields, got %d tuples with %d fields.",
722 : PQntuples(res), PQnfields(res))));
723 20 : *filename = pstrdup(PQgetvalue(res, 0, 0));
724 :
725 20 : *len = PQgetlength(res, 0, 1);
726 20 : *content = palloc(*len);
727 20 : memcpy(*content, PQgetvalue(res, 0, 1), *len);
728 20 : PQclear(res);
729 20 : }
730 :
731 : /*
732 : * Disconnect connection to primary, if any.
733 : */
734 : static void
735 1544 : libpqrcv_disconnect(WalReceiverConn *conn)
736 : {
737 1544 : libpqsrv_disconnect(conn->streamConn);
738 1544 : PQfreemem(conn->recvBuf);
739 1544 : pfree(conn);
740 1544 : }
741 :
742 : /*
743 : * Receive a message available from XLOG stream.
744 : *
745 : * Returns:
746 : *
747 : * If data was received, returns the length of the data. *buffer is set to
748 : * point to a buffer holding the received message. The buffer is only valid
749 : * until the next libpqrcv_* call.
750 : *
751 : * If no data was available immediately, returns 0, and *wait_fd is set to a
752 : * socket descriptor which can be waited on before trying again.
753 : *
754 : * -1 if the server ended the COPY.
755 : *
756 : * ereports on error.
757 : */
758 : static int
759 795774 : libpqrcv_receive(WalReceiverConn *conn, char **buffer,
760 : pgsocket *wait_fd)
761 : {
762 : int rawlen;
763 :
764 795774 : PQfreemem(conn->recvBuf);
765 795774 : conn->recvBuf = NULL;
766 :
767 : /* Try to receive a CopyData message */
768 795774 : rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
769 795774 : if (rawlen == 0)
770 : {
771 : /* Try consuming some data. */
772 540236 : if (PQconsumeInput(conn->streamConn) == 0)
773 76 : ereport(ERROR,
774 : (errcode(ERRCODE_CONNECTION_FAILURE),
775 : errmsg("could not receive data from WAL stream: %s",
776 : pchomp(PQerrorMessage(conn->streamConn)))));
777 :
778 : /* Now that we've consumed some input, try again */
779 540160 : rawlen = PQgetCopyData(conn->streamConn, &conn->recvBuf, 1);
780 540160 : if (rawlen == 0)
781 : {
782 : /* Tell caller to try again when our socket is ready. */
783 208234 : *wait_fd = PQsocket(conn->streamConn);
784 208234 : return 0;
785 : }
786 : }
787 587464 : if (rawlen == -1) /* end-of-streaming or error */
788 : {
789 : PGresult *res;
790 :
791 476 : res = libpqsrv_get_result(conn->streamConn,
792 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
793 476 : if (PQresultStatus(res) == PGRES_COMMAND_OK)
794 : {
795 448 : PQclear(res);
796 :
797 : /* Verify that there are no more results. */
798 448 : res = libpqsrv_get_result(conn->streamConn,
799 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
800 448 : if (res != NULL)
801 : {
802 70 : PQclear(res);
803 :
804 : /*
805 : * If the other side closed the connection orderly (otherwise
806 : * we'd seen an error, or PGRES_COPY_IN) don't report an error
807 : * here, but let callers deal with it.
808 : */
809 70 : if (PQstatus(conn->streamConn) == CONNECTION_BAD)
810 70 : return -1;
811 :
812 0 : ereport(ERROR,
813 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
814 : errmsg("unexpected result after CommandComplete: %s",
815 : PQerrorMessage(conn->streamConn))));
816 : }
817 :
818 378 : return -1;
819 : }
820 28 : else if (PQresultStatus(res) == PGRES_COPY_IN)
821 : {
822 22 : PQclear(res);
823 22 : return -1;
824 : }
825 : else
826 6 : ereport(ERROR,
827 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
828 : errmsg("could not receive data from WAL stream: %s",
829 : pchomp(PQerrorMessage(conn->streamConn)))));
830 : }
831 586988 : if (rawlen < -1)
832 0 : ereport(ERROR,
833 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
834 : errmsg("could not receive data from WAL stream: %s",
835 : pchomp(PQerrorMessage(conn->streamConn)))));
836 :
837 : /* Return received messages to caller */
838 586988 : *buffer = conn->recvBuf;
839 586988 : return rawlen;
840 : }
841 :
842 : /*
843 : * Send a message to XLOG stream.
844 : *
845 : * ereports on error.
846 : */
847 : static void
848 217958 : libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
849 : {
850 435916 : if (PQputCopyData(conn->streamConn, buffer, nbytes) <= 0 ||
851 217958 : PQflush(conn->streamConn))
852 0 : ereport(ERROR,
853 : (errcode(ERRCODE_CONNECTION_FAILURE),
854 : errmsg("could not send data to WAL stream: %s",
855 : pchomp(PQerrorMessage(conn->streamConn)))));
856 217958 : }
857 :
858 : /*
859 : * Create new replication slot.
860 : * Returns the name of the exported snapshot for logical slot or NULL for
861 : * physical slot.
862 : */
863 : static char *
864 612 : libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
865 : bool temporary, bool two_phase, bool failover,
866 : CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
867 : {
868 : PGresult *res;
869 : StringInfoData cmd;
870 : char *snapshot;
871 : int use_new_options_syntax;
872 :
873 612 : use_new_options_syntax = (PQserverVersion(conn->streamConn) >= 150000);
874 :
875 612 : initStringInfo(&cmd);
876 :
877 612 : appendStringInfo(&cmd, "CREATE_REPLICATION_SLOT \"%s\"", slotname);
878 :
879 612 : if (temporary)
880 0 : appendStringInfoString(&cmd, " TEMPORARY");
881 :
882 612 : if (conn->logical)
883 : {
884 612 : appendStringInfoString(&cmd, " LOGICAL pgoutput ");
885 612 : if (use_new_options_syntax)
886 612 : appendStringInfoChar(&cmd, '(');
887 612 : if (two_phase)
888 : {
889 2 : appendStringInfoString(&cmd, "TWO_PHASE");
890 2 : if (use_new_options_syntax)
891 2 : appendStringInfoString(&cmd, ", ");
892 : else
893 0 : appendStringInfoChar(&cmd, ' ');
894 : }
895 :
896 612 : if (failover)
897 : {
898 12 : appendStringInfoString(&cmd, "FAILOVER");
899 12 : if (use_new_options_syntax)
900 12 : appendStringInfoString(&cmd, ", ");
901 : else
902 0 : appendStringInfoChar(&cmd, ' ');
903 : }
904 :
905 612 : if (use_new_options_syntax)
906 : {
907 612 : switch (snapshot_action)
908 : {
909 0 : case CRS_EXPORT_SNAPSHOT:
910 0 : appendStringInfoString(&cmd, "SNAPSHOT 'export'");
911 0 : break;
912 224 : case CRS_NOEXPORT_SNAPSHOT:
913 224 : appendStringInfoString(&cmd, "SNAPSHOT 'nothing'");
914 224 : break;
915 388 : case CRS_USE_SNAPSHOT:
916 388 : appendStringInfoString(&cmd, "SNAPSHOT 'use'");
917 388 : break;
918 : }
919 : }
920 : else
921 : {
922 0 : switch (snapshot_action)
923 : {
924 0 : case CRS_EXPORT_SNAPSHOT:
925 0 : appendStringInfoString(&cmd, "EXPORT_SNAPSHOT");
926 0 : break;
927 0 : case CRS_NOEXPORT_SNAPSHOT:
928 0 : appendStringInfoString(&cmd, "NOEXPORT_SNAPSHOT");
929 0 : break;
930 0 : case CRS_USE_SNAPSHOT:
931 0 : appendStringInfoString(&cmd, "USE_SNAPSHOT");
932 0 : break;
933 : }
934 : }
935 :
936 612 : if (use_new_options_syntax)
937 612 : appendStringInfoChar(&cmd, ')');
938 : }
939 : else
940 : {
941 0 : if (use_new_options_syntax)
942 0 : appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
943 : else
944 0 : appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
945 : }
946 :
947 612 : res = libpqsrv_exec(conn->streamConn,
948 612 : cmd.data,
949 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
950 612 : pfree(cmd.data);
951 :
952 612 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
953 0 : ereport(ERROR,
954 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
955 : errmsg("could not create replication slot \"%s\": %s",
956 : slotname, pchomp(PQerrorMessage(conn->streamConn)))));
957 :
958 612 : if (lsn)
959 388 : *lsn = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid,
960 388 : CStringGetDatum(PQgetvalue(res, 0, 1))));
961 :
962 612 : if (!PQgetisnull(res, 0, 2))
963 0 : snapshot = pstrdup(PQgetvalue(res, 0, 2));
964 : else
965 612 : snapshot = NULL;
966 :
967 612 : PQclear(res);
968 :
969 612 : return snapshot;
970 : }
971 :
972 : /*
973 : * Change the definition of the replication slot.
974 : */
975 : static void
976 8 : libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
977 : const bool *failover, const bool *two_phase)
978 : {
979 : StringInfoData cmd;
980 : PGresult *res;
981 :
982 8 : initStringInfo(&cmd);
983 8 : appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( ",
984 : quote_identifier(slotname));
985 :
986 8 : if (failover)
987 6 : appendStringInfo(&cmd, "FAILOVER %s",
988 6 : *failover ? "true" : "false");
989 :
990 8 : if (failover && two_phase)
991 0 : appendStringInfoString(&cmd, ", ");
992 :
993 8 : if (two_phase)
994 2 : appendStringInfo(&cmd, "TWO_PHASE %s",
995 2 : *two_phase ? "true" : "false");
996 :
997 8 : appendStringInfoString(&cmd, " );");
998 :
999 8 : res = libpqsrv_exec(conn->streamConn, cmd.data,
1000 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
1001 8 : pfree(cmd.data);
1002 :
1003 8 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
1004 0 : ereport(ERROR,
1005 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
1006 : errmsg("could not alter replication slot \"%s\": %s",
1007 : slotname, pchomp(PQerrorMessage(conn->streamConn)))));
1008 :
1009 8 : PQclear(res);
1010 8 : }
1011 :
1012 : /*
1013 : * Return PID of remote backend process.
1014 : */
1015 : static pid_t
1016 0 : libpqrcv_get_backend_pid(WalReceiverConn *conn)
1017 : {
1018 0 : return PQbackendPID(conn->streamConn);
1019 : }
1020 :
1021 : /*
1022 : * Convert tuple query result to tuplestore.
1023 : */
1024 : static void
1025 2212 : libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
1026 : const int nRetTypes, const Oid *retTypes)
1027 : {
1028 : int tupn;
1029 : int coln;
1030 2212 : int nfields = PQnfields(pgres);
1031 : HeapTuple tuple;
1032 : AttInMetadata *attinmeta;
1033 : MemoryContext rowcontext;
1034 : MemoryContext oldcontext;
1035 :
1036 : /* Make sure we got expected number of fields. */
1037 2212 : if (nfields != nRetTypes)
1038 0 : ereport(ERROR,
1039 : (errcode(ERRCODE_PROTOCOL_VIOLATION),
1040 : errmsg("invalid query response"),
1041 : errdetail("Expected %d fields, got %d fields.",
1042 : nRetTypes, nfields)));
1043 :
1044 2212 : walres->tuplestore = tuplestore_begin_heap(true, false, work_mem);
1045 :
1046 : /* Create tuple descriptor corresponding to expected result. */
1047 2212 : walres->tupledesc = CreateTemplateTupleDesc(nRetTypes);
1048 7782 : for (coln = 0; coln < nRetTypes; coln++)
1049 5570 : TupleDescInitEntry(walres->tupledesc, (AttrNumber) coln + 1,
1050 5570 : PQfname(pgres, coln), retTypes[coln], -1, 0);
1051 2212 : attinmeta = TupleDescGetAttInMetadata(walres->tupledesc);
1052 :
1053 : /* No point in doing more here if there were no tuples returned. */
1054 2212 : if (PQntuples(pgres) == 0)
1055 46 : return;
1056 :
1057 : /* Create temporary context for local allocations. */
1058 2166 : rowcontext = AllocSetContextCreate(CurrentMemoryContext,
1059 : "libpqrcv query result context",
1060 : ALLOCSET_DEFAULT_SIZES);
1061 :
1062 : /* Process returned rows. */
1063 5034 : for (tupn = 0; tupn < PQntuples(pgres); tupn++)
1064 : {
1065 : char *cstrs[MaxTupleAttributeNumber];
1066 :
1067 2868 : CHECK_FOR_INTERRUPTS();
1068 :
1069 : /* Do the allocations in temporary context. */
1070 2868 : oldcontext = MemoryContextSwitchTo(rowcontext);
1071 :
1072 : /*
1073 : * Fill cstrs with null-terminated strings of column values.
1074 : */
1075 11258 : for (coln = 0; coln < nfields; coln++)
1076 : {
1077 8390 : if (PQgetisnull(pgres, tupn, coln))
1078 1174 : cstrs[coln] = NULL;
1079 : else
1080 7216 : cstrs[coln] = PQgetvalue(pgres, tupn, coln);
1081 : }
1082 :
1083 : /* Convert row to a tuple, and add it to the tuplestore */
1084 2868 : tuple = BuildTupleFromCStrings(attinmeta, cstrs);
1085 2868 : tuplestore_puttuple(walres->tuplestore, tuple);
1086 :
1087 : /* Clean up */
1088 2868 : MemoryContextSwitchTo(oldcontext);
1089 2868 : MemoryContextReset(rowcontext);
1090 : }
1091 :
1092 2166 : MemoryContextDelete(rowcontext);
1093 : }
1094 :
1095 : /*
1096 : * Public interface for sending generic queries (and commands).
1097 : *
1098 : * This can only be called from process connected to database.
1099 : */
1100 : static WalRcvExecResult *
1101 3878 : libpqrcv_exec(WalReceiverConn *conn, const char *query,
1102 : const int nRetTypes, const Oid *retTypes)
1103 : {
1104 3878 : PGresult *pgres = NULL;
1105 3878 : WalRcvExecResult *walres = palloc0(sizeof(WalRcvExecResult));
1106 : char *diag_sqlstate;
1107 :
1108 3878 : if (MyDatabaseId == InvalidOid)
1109 0 : ereport(ERROR,
1110 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1111 : errmsg("the query interface requires a database connection")));
1112 :
1113 3878 : pgres = libpqsrv_exec(conn->streamConn,
1114 : query,
1115 : WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
1116 :
1117 3878 : switch (PQresultStatus(pgres))
1118 : {
1119 2212 : case PGRES_TUPLES_OK:
1120 : case PGRES_SINGLE_TUPLE:
1121 : case PGRES_TUPLES_CHUNK:
1122 2212 : walres->status = WALRCV_OK_TUPLES;
1123 2212 : libpqrcv_processTuples(pgres, walres, nRetTypes, retTypes);
1124 2212 : break;
1125 :
1126 0 : case PGRES_COPY_IN:
1127 0 : walres->status = WALRCV_OK_COPY_IN;
1128 0 : break;
1129 :
1130 382 : case PGRES_COPY_OUT:
1131 382 : walres->status = WALRCV_OK_COPY_OUT;
1132 382 : break;
1133 :
1134 0 : case PGRES_COPY_BOTH:
1135 0 : walres->status = WALRCV_OK_COPY_BOTH;
1136 0 : break;
1137 :
1138 1280 : case PGRES_COMMAND_OK:
1139 1280 : walres->status = WALRCV_OK_COMMAND;
1140 1280 : break;
1141 :
1142 : /* Empty query is considered error. */
1143 0 : case PGRES_EMPTY_QUERY:
1144 0 : walres->status = WALRCV_ERROR;
1145 0 : walres->err = _("empty query");
1146 0 : break;
1147 :
1148 0 : case PGRES_PIPELINE_SYNC:
1149 : case PGRES_PIPELINE_ABORTED:
1150 0 : walres->status = WALRCV_ERROR;
1151 0 : walres->err = _("unexpected pipeline mode");
1152 0 : break;
1153 :
1154 4 : case PGRES_NONFATAL_ERROR:
1155 : case PGRES_FATAL_ERROR:
1156 : case PGRES_BAD_RESPONSE:
1157 4 : walres->status = WALRCV_ERROR;
1158 4 : walres->err = pchomp(PQerrorMessage(conn->streamConn));
1159 4 : diag_sqlstate = PQresultErrorField(pgres, PG_DIAG_SQLSTATE);
1160 4 : if (diag_sqlstate)
1161 4 : walres->sqlstate = MAKE_SQLSTATE(diag_sqlstate[0],
1162 : diag_sqlstate[1],
1163 : diag_sqlstate[2],
1164 : diag_sqlstate[3],
1165 : diag_sqlstate[4]);
1166 4 : break;
1167 : }
1168 :
1169 3878 : PQclear(pgres);
1170 :
1171 3878 : return walres;
1172 : }
1173 :
1174 : /*
1175 : * Given a List of strings, return it as single comma separated
1176 : * string, quoting identifiers as needed.
1177 : *
1178 : * This is essentially the reverse of SplitIdentifierString.
1179 : *
1180 : * The caller should free the result.
1181 : */
1182 : static char *
1183 762 : stringlist_to_identifierstr(PGconn *conn, List *strings)
1184 : {
1185 : ListCell *lc;
1186 : StringInfoData res;
1187 762 : bool first = true;
1188 :
1189 762 : initStringInfo(&res);
1190 :
1191 2016 : foreach(lc, strings)
1192 : {
1193 1254 : char *val = strVal(lfirst(lc));
1194 : char *val_escaped;
1195 :
1196 1254 : if (first)
1197 762 : first = false;
1198 : else
1199 492 : appendStringInfoChar(&res, ',');
1200 :
1201 1254 : val_escaped = PQescapeIdentifier(conn, val, strlen(val));
1202 1254 : if (!val_escaped)
1203 : {
1204 0 : free(res.data);
1205 0 : return NULL;
1206 : }
1207 1254 : appendStringInfoString(&res, val_escaped);
1208 1254 : PQfreemem(val_escaped);
1209 : }
1210 :
1211 762 : return res.data;
1212 : }
|
