PostgreSQL Source Code git master
All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
walreceiver.h
Go to the documentation of this file.
1/*-------------------------------------------------------------------------
2 *
3 * walreceiver.h
4 * Exports from replication/walreceiverfuncs.c.
5 *
6 * Portions Copyright (c) 2010-2024, PostgreSQL Global Development Group
7 *
8 * src/include/replication/walreceiver.h
9 *
10 *-------------------------------------------------------------------------
11 */
12#ifndef _WALRECEIVER_H
13#define _WALRECEIVER_H
14
15#include <netdb.h>
16
17#include "access/xlog.h"
18#include "access/xlogdefs.h"
19#include "pgtime.h"
20#include "port/atomics.h"
24#include "storage/spin.h"
25#include "utils/tuplestore.h"
26
27/* user-settable parameters */
31
32/*
33 * MAXCONNINFO: maximum size of a connection string.
34 *
35 * XXX: Should this move to pg_config_manual.h?
36 */
37#define MAXCONNINFO 1024
38
39/* Can we allow the standby to accept replication connection from another standby? */
40#define AllowCascadeReplication() (EnableHotStandby && max_wal_senders > 0)
41
42/*
43 * Values for WalRcv->walRcvState.
44 */
45typedef enum
46{
47 WALRCV_STOPPED, /* stopped and mustn't start up again */
48 WALRCV_STARTING, /* launched, but the process hasn't
49 * initialized yet */
50 WALRCV_STREAMING, /* walreceiver is streaming */
51 WALRCV_WAITING, /* stopped streaming, waiting for orders */
52 WALRCV_RESTARTING, /* asked to restart streaming */
53 WALRCV_STOPPING, /* requested to stop, but still running */
55
56/* Shared memory area for management of walreceiver process */
57typedef struct
58{
59 /*
60 * Currently active walreceiver process's proc number and PID.
61 *
62 * The startup process uses the proc number to wake it up after telling it
63 * where to start streaming (after setting receiveStart and
64 * receiveStartTLI), and also to tell it to send apply feedback to the
65 * primary whenever specially marked commit records are applied.
66 */
68 pid_t pid;
69
70 /* Its current state */
73
74 /*
75 * Its start time (actually, the time at which it was requested to be
76 * started).
77 */
79
80 /*
81 * receiveStart and receiveStartTLI indicate the first byte position and
82 * timeline that will be received. When startup process starts the
83 * walreceiver, it sets these to the point where it wants the streaming to
84 * begin.
85 */
88
89 /*
90 * flushedUpto-1 is the last byte position that has already been received,
91 * and receivedTLI is the timeline it came from. At the first startup of
92 * walreceiver, these are set to receiveStart and receiveStartTLI. After
93 * that, walreceiver updates these whenever it flushes the received WAL to
94 * disk.
95 */
98
99 /*
100 * latestChunkStart is the starting byte position of the current "batch"
101 * of received WAL. It's actually the same as the previous value of
102 * flushedUpto before the last flush to disk. Startup process can use
103 * this to detect whether it's keeping up or not.
104 */
106
107 /*
108 * Time of send and receive of any message received.
109 */
112
113 /*
114 * Latest reported end of WAL on the sender
115 */
118
119 /*
120 * connection string; initially set to connect to the primary, and later
121 * clobbered to hide security-sensitive fields.
122 */
123 char conninfo[MAXCONNINFO];
124
125 /*
126 * Host name (this can be a host name, an IP address, or a directory path)
127 * and port number of the active replication connection.
128 */
129 char sender_host[NI_MAXHOST];
131
132 /*
133 * replication slot name; is also used for walreceiver to connect with the
134 * primary
135 */
136 char slotname[NAMEDATALEN];
137
138 /*
139 * If it's a temporary replication slot, it needs to be recreated when
140 * connecting.
141 */
143
144 /* set true once conninfo is ready to display (obfuscated pwds etc) */
146
147 slock_t mutex; /* locks shared variables shown above */
148
149 /*
150 * Like flushedUpto, but advanced after writing and before flushing,
151 * without the need to acquire the spin lock. Data can be read by another
152 * process up to this point, but shouldn't be used for data integrity
153 * purposes.
154 */
156
157 /*
158 * force walreceiver reply? This doesn't need to be locked; memory
159 * barriers for ordering are sufficient. But we do need atomic fetch and
160 * store semantics, so use sig_atomic_t.
161 */
162 sig_atomic_t force_reply; /* used as a bool */
163} WalRcvData;
164
166
167typedef struct
168{
169 bool logical; /* True if this is logical replication stream,
170 * false if physical stream. */
171 char *slotname; /* Name of the replication slot or NULL. */
172 XLogRecPtr startpoint; /* LSN of starting point. */
173
174 union
175 {
176 struct
177 {
178 TimeLineID startpointTLI; /* Starting timeline */
179 } physical;
180 struct
181 {
182 uint32 proto_version; /* Logical protocol version */
183 List *publication_names; /* String list of publications */
184 bool binary; /* Ask publisher to use binary */
185 char *streaming_str; /* Streaming of large transactions */
186 bool twophase; /* Streaming of two-phase transactions at
187 * prepare time */
188 char *origin; /* Only publish data originating from the
189 * specified origin */
190 } logical;
191 } proto;
193
194struct WalReceiverConn;
196
197/*
198 * Status of walreceiver query execution.
199 *
200 * We only define statuses that are currently used.
201 */
202typedef enum
203{
204 WALRCV_ERROR, /* There was error when executing the query. */
205 WALRCV_OK_COMMAND, /* Query executed utility or replication
206 * command. */
207 WALRCV_OK_TUPLES, /* Query returned tuples. */
208 WALRCV_OK_COPY_IN, /* Query started COPY FROM. */
209 WALRCV_OK_COPY_OUT, /* Query started COPY TO. */
210 WALRCV_OK_COPY_BOTH, /* Query started COPY BOTH replication
211 * protocol. */
213
214/*
215 * Return value for walrcv_exec, returns the status of the execution and
216 * tuples if any.
217 */
218typedef struct WalRcvExecResult
219{
222 char *err;
226
227/* WAL receiver - libpqwalreceiver hooks */
228
229/*
230 * walrcv_connect_fn
231 *
232 * Establish connection to a cluster. 'replication' is true if the
233 * connection is a replication connection, and false if it is a
234 * regular connection. If it is a replication connection, it could
235 * be either logical or physical based on input argument 'logical'.
236 * 'appname' is a name associated to the connection, to use for example
237 * with fallback_application_name or application_name. Returns the
238 * details about the connection established, as defined by
239 * WalReceiverConn for each WAL receiver module. On error, NULL is
240 * returned with 'err' including the error generated.
241 */
242typedef WalReceiverConn *(*walrcv_connect_fn) (const char *conninfo,
243 bool replication,
244 bool logical,
245 bool must_use_password,
246 const char *appname,
247 char **err);
248
249/*
250 * walrcv_check_conninfo_fn
251 *
252 * Parse and validate the connection string given as of 'conninfo'.
253 */
254typedef void (*walrcv_check_conninfo_fn) (const char *conninfo,
255 bool must_use_password);
256
257/*
258 * walrcv_get_conninfo_fn
259 *
260 * Returns a user-displayable conninfo string. Note that any
261 * security-sensitive fields should be obfuscated.
262 */
263typedef char *(*walrcv_get_conninfo_fn) (WalReceiverConn *conn);
264
265/*
266 * walrcv_get_senderinfo_fn
267 *
268 * Provide information of the WAL sender this WAL receiver is connected
269 * to, as of 'sender_host' for the host of the sender and 'sender_port'
270 * for its port.
271 */
273 char **sender_host,
274 int *sender_port);
275
276/*
277 * walrcv_identify_system_fn
278 *
279 * Run IDENTIFY_SYSTEM on the cluster connected to and validate the
280 * identity of the cluster. Returns the system ID of the cluster
281 * connected to. 'primary_tli' is the timeline ID of the sender.
282 */
283typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn,
284 TimeLineID *primary_tli);
285
286/*
287 * walrcv_get_dbname_from_conninfo_fn
288 *
289 * Returns the database name from the primary_conninfo
290 */
291typedef char *(*walrcv_get_dbname_from_conninfo_fn) (const char *conninfo);
292
293/*
294 * walrcv_server_version_fn
295 *
296 * Returns the version number of the cluster connected to.
297 */
299
300/*
301 * walrcv_readtimelinehistoryfile_fn
302 *
303 * Fetch from cluster the timeline history file for timeline 'tli'.
304 * Returns the name of the timeline history file as of 'filename', its
305 * contents as of 'content' and its 'size'.
306 */
308 TimeLineID tli,
309 char **filename,
310 char **content,
311 int *size);
312
313/*
314 * walrcv_startstreaming_fn
315 *
316 * Start streaming WAL data from given streaming options. Returns true
317 * if the connection has switched successfully to copy-both mode and false
318 * if the server received the command and executed it successfully, but
319 * didn't switch to copy-mode.
320 */
323
324/*
325 * walrcv_endstreaming_fn
326 *
327 * Stop streaming of WAL data. Returns the next timeline ID of the cluster
328 * connected to in 'next_tli', or 0 if there was no report.
329 */
331 TimeLineID *next_tli);
332
333/*
334 * walrcv_receive_fn
335 *
336 * Receive a message available from the WAL stream. 'buffer' is a pointer
337 * to a buffer holding the message received. Returns the length of the data,
338 * 0 if no data is available yet ('wait_fd' is a socket descriptor which can
339 * be waited on before a retry), and -1 if the cluster ended the COPY.
340 */
342 char **buffer,
343 pgsocket *wait_fd);
344
345/*
346 * walrcv_send_fn
347 *
348 * Send a message of size 'nbytes' to the WAL stream with 'buffer' as
349 * contents.
350 */
352 const char *buffer,
353 int nbytes);
354
355/*
356 * walrcv_create_slot_fn
357 *
358 * Create a new replication slot named 'slotname'. 'temporary' defines
359 * if the slot is temporary. 'snapshot_action' defines the behavior wanted
360 * for an exported snapshot (see replication protocol for more details).
361 * 'lsn' includes the LSN position at which the created slot became
362 * consistent. Returns the name of the exported snapshot for a logical
363 * slot, or NULL for a physical slot.
364 */
365typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
366 const char *slotname,
367 bool temporary,
368 bool two_phase,
369 bool failover,
370 CRSSnapshotAction snapshot_action,
371 XLogRecPtr *lsn);
372
373/*
374 * walrcv_alter_slot_fn
375 *
376 * Change the definition of a replication slot. Currently, it supports
377 * changing the failover and two_phase properties of the slot.
378 */
380 const char *slotname,
381 const bool *failover,
382 const bool *two_phase);
383
384
385/*
386 * walrcv_get_backend_pid_fn
387 *
388 * Returns the PID of the remote backend process.
389 */
391
392/*
393 * walrcv_exec_fn
394 *
395 * Send generic queries (and commands) to the remote cluster. 'nRetTypes'
396 * is the expected number of returned attributes, and 'retTypes' an array
397 * including their type OIDs. Returns the status of the execution and
398 * tuples if any.
399 */
400typedef WalRcvExecResult *(*walrcv_exec_fn) (WalReceiverConn *conn,
401 const char *query,
402 const int nRetTypes,
403 const Oid *retTypes);
404
405/*
406 * walrcv_disconnect_fn
407 *
408 * Disconnect with the cluster.
409 */
411
413{
432
434
435#define walrcv_connect(conninfo, replication, logical, must_use_password, appname, err) \
436 WalReceiverFunctions->walrcv_connect(conninfo, replication, logical, must_use_password, appname, err)
437#define walrcv_check_conninfo(conninfo, must_use_password) \
438 WalReceiverFunctions->walrcv_check_conninfo(conninfo, must_use_password)
439#define walrcv_get_conninfo(conn) \
440 WalReceiverFunctions->walrcv_get_conninfo(conn)
441#define walrcv_get_senderinfo(conn, sender_host, sender_port) \
442 WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)
443#define walrcv_identify_system(conn, primary_tli) \
444 WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)
445#define walrcv_get_dbname_from_conninfo(conninfo) \
446 WalReceiverFunctions->walrcv_get_dbname_from_conninfo(conninfo)
447#define walrcv_server_version(conn) \
448 WalReceiverFunctions->walrcv_server_version(conn)
449#define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \
450 WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)
451#define walrcv_startstreaming(conn, options) \
452 WalReceiverFunctions->walrcv_startstreaming(conn, options)
453#define walrcv_endstreaming(conn, next_tli) \
454 WalReceiverFunctions->walrcv_endstreaming(conn, next_tli)
455#define walrcv_receive(conn, buffer, wait_fd) \
456 WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
457#define walrcv_send(conn, buffer, nbytes) \
458 WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
459#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \
460 WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
461#define walrcv_alter_slot(conn, slotname, failover, two_phase) \
462 WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover, two_phase)
463#define walrcv_get_backend_pid(conn) \
464 WalReceiverFunctions->walrcv_get_backend_pid(conn)
465#define walrcv_exec(conn, exec, nRetTypes, retTypes) \
466 WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)
467#define walrcv_disconnect(conn) \
468 WalReceiverFunctions->walrcv_disconnect(conn)
469
470static inline void
472{
473 if (!walres)
474 return;
475
476 if (walres->err)
477 pfree(walres->err);
478
479 if (walres->tuplestore)
480 tuplestore_end(walres->tuplestore);
481
482 if (walres->tupledesc)
483 FreeTupleDesc(walres->tupledesc);
484
485 pfree(walres);
486}
487
488/* prototypes for functions in walreceiver.c */
489extern void WalReceiverMain(char *startup_data, size_t startup_data_len) pg_attribute_noreturn();
490extern void ProcessWalRcvInterrupts(void);
491extern void WalRcvForceReply(void);
492
493/* prototypes for functions in walreceiverfuncs.c */
494extern Size WalRcvShmemSize(void);
495extern void WalRcvShmemInit(void);
496extern void ShutdownWalRcv(void);
497extern bool WalRcvStreaming(void);
498extern bool WalRcvRunning(void);
499extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr,
500 const char *conninfo, const char *slotname,
501 bool create_temp_slot);
504extern int GetReplicationApplyDelay(void);
505extern int GetReplicationTransferLatency(void);
506
507#endif /* _WALRECEIVER_H */
#define PGDLLIMPORT
Definition: c.h:1274
#define pg_attribute_noreturn()
Definition: c.h:236
uint32_t uint32
Definition: c.h:485
size_t Size
Definition: c.h:559
int64 TimestampTz
Definition: timestamp.h:39
void err(int eval, const char *fmt,...)
Definition: err.c:43
void pfree(void *pointer)
Definition: mcxt.c:1521
#define NAMEDATALEN
static char * filename
Definition: pg_dumpall.c:119
static bool two_phase
static char ** options
int64 pg_time_t
Definition: pgtime.h:23
int pgsocket
Definition: port.h:29
unsigned int Oid
Definition: postgres_ext.h:31
int ProcNumber
Definition: procnumber.h:24
static pg_noinline void Size size
Definition: slab.c:607
PGconn * conn
Definition: streamutil.c:53
Definition: pg_list.h:54
TimestampTz lastMsgReceiptTime
Definition: walreceiver.h:111
XLogRecPtr latestWalEnd
Definition: walreceiver.h:116
TimeLineID receiveStartTLI
Definition: walreceiver.h:87
TimeLineID receivedTLI
Definition: walreceiver.h:97
pid_t pid
Definition: walreceiver.h:68
XLogRecPtr latestChunkStart
Definition: walreceiver.h:105
XLogRecPtr receiveStart
Definition: walreceiver.h:86
XLogRecPtr flushedUpto
Definition: walreceiver.h:96
sig_atomic_t force_reply
Definition: walreceiver.h:162
ProcNumber procno
Definition: walreceiver.h:67
ConditionVariable walRcvStoppedCV
Definition: walreceiver.h:72
bool is_temp_slot
Definition: walreceiver.h:142
pg_atomic_uint64 writtenUpto
Definition: walreceiver.h:155
pg_time_t startTime
Definition: walreceiver.h:78
TimestampTz lastMsgSendTime
Definition: walreceiver.h:110
WalRcvState walRcvState
Definition: walreceiver.h:71
TimestampTz latestWalEndTime
Definition: walreceiver.h:117
bool ready_to_display
Definition: walreceiver.h:145
int sender_port
Definition: walreceiver.h:130
slock_t mutex
Definition: walreceiver.h:147
Tuplestorestate * tuplestore
Definition: walreceiver.h:223
TupleDesc tupledesc
Definition: walreceiver.h:224
WalRcvExecStatus status
Definition: walreceiver.h:220
XLogRecPtr startpoint
Definition: walreceiver.h:172
TimeLineID startpointTLI
Definition: walreceiver.h:178
walrcv_identify_system_fn walrcv_identify_system
Definition: walreceiver.h:418
walrcv_receive_fn walrcv_receive
Definition: walreceiver.h:424
walrcv_server_version_fn walrcv_server_version
Definition: walreceiver.h:420
walrcv_disconnect_fn walrcv_disconnect
Definition: walreceiver.h:430
walrcv_get_backend_pid_fn walrcv_get_backend_pid
Definition: walreceiver.h:428
walrcv_connect_fn walrcv_connect
Definition: walreceiver.h:414
walrcv_get_dbname_from_conninfo_fn walrcv_get_dbname_from_conninfo
Definition: walreceiver.h:419
walrcv_get_senderinfo_fn walrcv_get_senderinfo
Definition: walreceiver.h:417
walrcv_check_conninfo_fn walrcv_check_conninfo
Definition: walreceiver.h:415
walrcv_send_fn walrcv_send
Definition: walreceiver.h:425
walrcv_get_conninfo_fn walrcv_get_conninfo
Definition: walreceiver.h:416
walrcv_exec_fn walrcv_exec
Definition: walreceiver.h:429
walrcv_startstreaming_fn walrcv_startstreaming
Definition: walreceiver.h:422
walrcv_create_slot_fn walrcv_create_slot
Definition: walreceiver.h:426
walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile
Definition: walreceiver.h:421
walrcv_alter_slot_fn walrcv_alter_slot
Definition: walreceiver.h:427
walrcv_endstreaming_fn walrcv_endstreaming
Definition: walreceiver.h:423
void FreeTupleDesc(TupleDesc tupdesc)
Definition: tupdesc.c:478
void tuplestore_end(Tuplestorestate *state)
Definition: tuplestore.c:492
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:162
XLogRecPtr GetWalRcvFlushRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI)
void(* walrcv_endstreaming_fn)(WalReceiverConn *conn, TimeLineID *next_tli)
Definition: walreceiver.h:330
bool WalRcvStreaming(void)
struct WalRcvExecResult WalRcvExecResult
char *(* walrcv_get_dbname_from_conninfo_fn)(const char *conninfo)
Definition: walreceiver.h:291
void(* walrcv_send_fn)(WalReceiverConn *conn, const char *buffer, int nbytes)
Definition: walreceiver.h:351
void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo, const char *slotname, bool create_temp_slot)
PGDLLIMPORT int wal_receiver_status_interval
Definition: walreceiver.c:87
char *(* walrcv_create_slot_fn)(WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, bool failover, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
Definition: walreceiver.h:365
WalRcvExecStatus
Definition: walreceiver.h:203
@ WALRCV_OK_COPY_IN
Definition: walreceiver.h:208
@ WALRCV_OK_COMMAND
Definition: walreceiver.h:205
@ WALRCV_ERROR
Definition: walreceiver.h:204
@ WALRCV_OK_TUPLES
Definition: walreceiver.h:207
@ WALRCV_OK_COPY_OUT
Definition: walreceiver.h:209
@ WALRCV_OK_COPY_BOTH
Definition: walreceiver.h:210
WalReceiverConn *(* walrcv_connect_fn)(const char *conninfo, bool replication, bool logical, bool must_use_password, const char *appname, char **err)
Definition: walreceiver.h:242
pid_t(* walrcv_get_backend_pid_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:390
void WalReceiverMain(char *startup_data, size_t startup_data_len) pg_attribute_noreturn()
Definition: walreceiver.c:183
XLogRecPtr GetWalRcvWriteRecPtr(void)
#define MAXCONNINFO
Definition: walreceiver.h:37
void ShutdownWalRcv(void)
bool(* walrcv_startstreaming_fn)(WalReceiverConn *conn, const WalRcvStreamOptions *options)
Definition: walreceiver.h:321
int(* walrcv_receive_fn)(WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
Definition: walreceiver.h:341
void(* walrcv_readtimelinehistoryfile_fn)(WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *size)
Definition: walreceiver.h:307
WalRcvExecResult *(* walrcv_exec_fn)(WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
Definition: walreceiver.h:400
PGDLLIMPORT int wal_receiver_timeout
Definition: walreceiver.c:88
static void walrcv_clear_result(WalRcvExecResult *walres)
Definition: walreceiver.h:471
void(* walrcv_check_conninfo_fn)(const char *conninfo, bool must_use_password)
Definition: walreceiver.h:254
PGDLLIMPORT WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:93
void(* walrcv_disconnect_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:410
struct WalReceiverFunctionsType WalReceiverFunctionsType
bool WalRcvRunning(void)
WalRcvState
Definition: walreceiver.h:46
@ WALRCV_STARTING
Definition: walreceiver.h:48
@ WALRCV_STOPPED
Definition: walreceiver.h:47
@ WALRCV_RESTARTING
Definition: walreceiver.h:52
@ WALRCV_STREAMING
Definition: walreceiver.h:50
@ WALRCV_WAITING
Definition: walreceiver.h:51
@ WALRCV_STOPPING
Definition: walreceiver.h:53
char *(* walrcv_identify_system_fn)(WalReceiverConn *conn, TimeLineID *primary_tli)
Definition: walreceiver.h:283
void(* walrcv_get_senderinfo_fn)(WalReceiverConn *conn, char **sender_host, int *sender_port)
Definition: walreceiver.h:272
int GetReplicationApplyDelay(void)
PGDLLIMPORT WalRcvData * WalRcv
void WalRcvShmemInit(void)
char *(* walrcv_get_conninfo_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:263
Size WalRcvShmemSize(void)
void(* walrcv_alter_slot_fn)(WalReceiverConn *conn, const char *slotname, const bool *failover, const bool *two_phase)
Definition: walreceiver.h:379
PGDLLIMPORT bool hot_standby_feedback
Definition: walreceiver.c:89
int(* walrcv_server_version_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:298
void WalRcvForceReply(void)
Definition: walreceiver.c:1359
int GetReplicationTransferLatency(void)
CRSSnapshotAction
Definition: walsender.h:21
uint64 XLogRecPtr
Definition: xlogdefs.h:21
uint32 TimeLineID
Definition: xlogdefs.h:59
static TimeLineID receiveTLI
Definition: xlogrecovery.c:263