PostgreSQL Source Code  git master
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-2023, 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 #include <sys/socket.h>
17 
18 #include "access/xlog.h"
19 #include "access/xlogdefs.h"
20 #include "pgtime.h"
21 #include "port/atomics.h"
23 #include "replication/walsender.h"
25 #include "storage/latch.h"
26 #include "storage/spin.h"
27 #include "utils/tuplestore.h"
28 
29 /* user-settable parameters */
33 
34 /*
35  * MAXCONNINFO: maximum size of a connection string.
36  *
37  * XXX: Should this move to pg_config_manual.h?
38  */
39 #define MAXCONNINFO 1024
40 
41 /* Can we allow the standby to accept replication connection from another standby? */
42 #define AllowCascadeReplication() (EnableHotStandby && max_wal_senders > 0)
43 
44 /*
45  * Values for WalRcv->walRcvState.
46  */
47 typedef enum
48 {
49  WALRCV_STOPPED, /* stopped and mustn't start up again */
50  WALRCV_STARTING, /* launched, but the process hasn't
51  * initialized yet */
52  WALRCV_STREAMING, /* walreceiver is streaming */
53  WALRCV_WAITING, /* stopped streaming, waiting for orders */
54  WALRCV_RESTARTING, /* asked to restart streaming */
55  WALRCV_STOPPING /* requested to stop, but still running */
57 
58 /* Shared memory area for management of walreceiver process */
59 typedef struct
60 {
61  /*
62  * PID of currently active walreceiver process, its current state and
63  * start time (actually, the time at which it was requested to be
64  * started).
65  */
66  pid_t pid;
70 
71  /*
72  * receiveStart and receiveStartTLI indicate the first byte position and
73  * timeline that will be received. When startup process starts the
74  * walreceiver, it sets these to the point where it wants the streaming to
75  * begin.
76  */
79 
80  /*
81  * flushedUpto-1 is the last byte position that has already been received,
82  * and receivedTLI is the timeline it came from. At the first startup of
83  * walreceiver, these are set to receiveStart and receiveStartTLI. After
84  * that, walreceiver updates these whenever it flushes the received WAL to
85  * disk.
86  */
89 
90  /*
91  * latestChunkStart is the starting byte position of the current "batch"
92  * of received WAL. It's actually the same as the previous value of
93  * flushedUpto before the last flush to disk. Startup process can use
94  * this to detect whether it's keeping up or not.
95  */
97 
98  /*
99  * Time of send and receive of any message received.
100  */
103 
104  /*
105  * Latest reported end of WAL on the sender
106  */
109 
110  /*
111  * connection string; initially set to connect to the primary, and later
112  * clobbered to hide security-sensitive fields.
113  */
114  char conninfo[MAXCONNINFO];
115 
116  /*
117  * Host name (this can be a host name, an IP address, or a directory path)
118  * and port number of the active replication connection.
119  */
120  char sender_host[NI_MAXHOST];
122 
123  /*
124  * replication slot name; is also used for walreceiver to connect with the
125  * primary
126  */
127  char slotname[NAMEDATALEN];
128 
129  /*
130  * If it's a temporary replication slot, it needs to be recreated when
131  * connecting.
132  */
134 
135  /* set true once conninfo is ready to display (obfuscated pwds etc) */
137 
138  /*
139  * Latch used by startup process to wake up walreceiver after telling it
140  * where to start streaming (after setting receiveStart and
141  * receiveStartTLI), and also to tell it to send apply feedback to the
142  * primary whenever specially marked commit records are applied. This is
143  * normally mapped to procLatch when walreceiver is running.
144  */
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 
167 typedef 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 
194 struct WalReceiverConn;
195 typedef struct WalReceiverConn WalReceiverConn;
196 
197 /*
198  * Status of walreceiver query execution.
199  *
200  * We only define statuses that are currently used.
201  */
202 typedef 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  */
218 typedef struct WalRcvExecResult
219 {
221  int sqlstate;
222  char *err;
226 
227 /* WAL receiver - libpqwalreceiver hooks */
228 
229 /*
230  * walrcv_connect_fn
231  *
232  * Establish connection to a cluster. 'logical' is true if the
233  * connection is logical, and false if the connection is physical.
234  * 'appname' is a name associated to the connection, to use for example
235  * with fallback_application_name or application_name. Returns the
236  * details about the connection established, as defined by
237  * WalReceiverConn for each WAL receiver module. On error, NULL is
238  * returned with 'err' including the error generated.
239  */
240 typedef WalReceiverConn *(*walrcv_connect_fn) (const char *conninfo,
241  bool logical,
242  bool must_use_password,
243  const char *appname,
244  char **err);
245 
246 /*
247  * walrcv_check_conninfo_fn
248  *
249  * Parse and validate the connection string given as of 'conninfo'.
250  */
251 typedef void (*walrcv_check_conninfo_fn) (const char *conninfo,
252  bool must_use_password);
253 
254 /*
255  * walrcv_get_conninfo_fn
256  *
257  * Returns a user-displayable conninfo string. Note that any
258  * security-sensitive fields should be obfuscated.
259  */
260 typedef char *(*walrcv_get_conninfo_fn) (WalReceiverConn *conn);
261 
262 /*
263  * walrcv_get_senderinfo_fn
264  *
265  * Provide information of the WAL sender this WAL receiver is connected
266  * to, as of 'sender_host' for the host of the sender and 'sender_port'
267  * for its port.
268  */
270  char **sender_host,
271  int *sender_port);
272 
273 /*
274  * walrcv_identify_system_fn
275  *
276  * Run IDENTIFY_SYSTEM on the cluster connected to and validate the
277  * identity of the cluster. Returns the system ID of the cluster
278  * connected to. 'primary_tli' is the timeline ID of the sender.
279  */
280 typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn,
281  TimeLineID *primary_tli);
282 
283 /*
284  * walrcv_server_version_fn
285  *
286  * Returns the version number of the cluster connected to.
287  */
289 
290 /*
291  * walrcv_readtimelinehistoryfile_fn
292  *
293  * Fetch from cluster the timeline history file for timeline 'tli'.
294  * Returns the name of the timeline history file as of 'filename', its
295  * contents as of 'content' and its 'size'.
296  */
298  TimeLineID tli,
299  char **filename,
300  char **content,
301  int *size);
302 
303 /*
304  * walrcv_startstreaming_fn
305  *
306  * Start streaming WAL data from given streaming options. Returns true
307  * if the connection has switched successfully to copy-both mode and false
308  * if the server received the command and executed it successfully, but
309  * didn't switch to copy-mode.
310  */
313 
314 /*
315  * walrcv_endstreaming_fn
316  *
317  * Stop streaming of WAL data. Returns the next timeline ID of the cluster
318  * connected to in 'next_tli', or 0 if there was no report.
319  */
321  TimeLineID *next_tli);
322 
323 /*
324  * walrcv_receive_fn
325  *
326  * Receive a message available from the WAL stream. 'buffer' is a pointer
327  * to a buffer holding the message received. Returns the length of the data,
328  * 0 if no data is available yet ('wait_fd' is a socket descriptor which can
329  * be waited on before a retry), and -1 if the cluster ended the COPY.
330  */
332  char **buffer,
333  pgsocket *wait_fd);
334 
335 /*
336  * walrcv_send_fn
337  *
338  * Send a message of size 'nbytes' to the WAL stream with 'buffer' as
339  * contents.
340  */
342  const char *buffer,
343  int nbytes);
344 
345 /*
346  * walrcv_create_slot_fn
347  *
348  * Create a new replication slot named 'slotname'. 'temporary' defines
349  * if the slot is temporary. 'snapshot_action' defines the behavior wanted
350  * for an exported snapshot (see replication protocol for more details).
351  * 'lsn' includes the LSN position at which the created slot became
352  * consistent. Returns the name of the exported snapshot for a logical
353  * slot, or NULL for a physical slot.
354  */
355 typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
356  const char *slotname,
357  bool temporary,
358  bool two_phase,
359  CRSSnapshotAction snapshot_action,
360  XLogRecPtr *lsn);
361 
362 /*
363  * walrcv_get_backend_pid_fn
364  *
365  * Returns the PID of the remote backend process.
366  */
368 
369 /*
370  * walrcv_exec_fn
371  *
372  * Send generic queries (and commands) to the remote cluster. 'nRetTypes'
373  * is the expected number of returned attributes, and 'retTypes' an array
374  * including their type OIDs. Returns the status of the execution and
375  * tuples if any.
376  */
377 typedef WalRcvExecResult *(*walrcv_exec_fn) (WalReceiverConn *conn,
378  const char *query,
379  const int nRetTypes,
380  const Oid *retTypes);
381 
382 /*
383  * walrcv_disconnect_fn
384  *
385  * Disconnect with the cluster.
386  */
388 
390 {
407 
409 
410 #define walrcv_connect(conninfo, logical, must_use_password, appname, err) \
411  WalReceiverFunctions->walrcv_connect(conninfo, logical, must_use_password, appname, err)
412 #define walrcv_check_conninfo(conninfo, must_use_password) \
413  WalReceiverFunctions->walrcv_check_conninfo(conninfo, must_use_password)
414 #define walrcv_get_conninfo(conn) \
415  WalReceiverFunctions->walrcv_get_conninfo(conn)
416 #define walrcv_get_senderinfo(conn, sender_host, sender_port) \
417  WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)
418 #define walrcv_identify_system(conn, primary_tli) \
419  WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)
420 #define walrcv_server_version(conn) \
421  WalReceiverFunctions->walrcv_server_version(conn)
422 #define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \
423  WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)
424 #define walrcv_startstreaming(conn, options) \
425  WalReceiverFunctions->walrcv_startstreaming(conn, options)
426 #define walrcv_endstreaming(conn, next_tli) \
427  WalReceiverFunctions->walrcv_endstreaming(conn, next_tli)
428 #define walrcv_receive(conn, buffer, wait_fd) \
429  WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
430 #define walrcv_send(conn, buffer, nbytes) \
431  WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
432 #define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn) \
433  WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)
434 #define walrcv_get_backend_pid(conn) \
435  WalReceiverFunctions->walrcv_get_backend_pid(conn)
436 #define walrcv_exec(conn, exec, nRetTypes, retTypes) \
437  WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)
438 #define walrcv_disconnect(conn) \
439  WalReceiverFunctions->walrcv_disconnect(conn)
440 
441 static inline void
443 {
444  if (!walres)
445  return;
446 
447  if (walres->err)
448  pfree(walres->err);
449 
450  if (walres->tuplestore)
451  tuplestore_end(walres->tuplestore);
452 
453  if (walres->tupledesc)
454  FreeTupleDesc(walres->tupledesc);
455 
456  pfree(walres);
457 }
458 
459 /* prototypes for functions in walreceiver.c */
460 extern void WalReceiverMain(void) pg_attribute_noreturn();
461 extern void ProcessWalRcvInterrupts(void);
462 extern void WalRcvForceReply(void);
463 
464 /* prototypes for functions in walreceiverfuncs.c */
465 extern Size WalRcvShmemSize(void);
466 extern void WalRcvShmemInit(void);
467 extern void ShutdownWalRcv(void);
468 extern bool WalRcvStreaming(void);
469 extern bool WalRcvRunning(void);
470 extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr,
471  const char *conninfo, const char *slotname,
472  bool create_temp_slot);
474 extern XLogRecPtr GetWalRcvWriteRecPtr(void);
475 extern int GetReplicationApplyDelay(void);
476 extern int GetReplicationTransferLatency(void);
477 
478 #endif /* _WALRECEIVER_H */
unsigned int uint32
Definition: c.h:495
#define PGDLLIMPORT
Definition: c.h:1326
#define pg_attribute_noreturn()
Definition: c.h:206
unsigned char bool
Definition: c.h:445
size_t Size
Definition: c.h:594
int64 TimestampTz
Definition: timestamp.h:39
void err(int eval, const char *fmt,...)
Definition: err.c:43
void pfree(void *pointer)
Definition: mcxt.c:1456
#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 slock_t
Definition: s_lock.h:754
PGconn * conn
Definition: streamutil.c:54
Definition: latch.h:111
Definition: pg_list.h:54
TimestampTz lastMsgReceiptTime
Definition: walreceiver.h:102
XLogRecPtr latestWalEnd
Definition: walreceiver.h:107
TimeLineID receiveStartTLI
Definition: walreceiver.h:78
TimeLineID receivedTLI
Definition: walreceiver.h:88
Latch * latch
Definition: walreceiver.h:145
pid_t pid
Definition: walreceiver.h:66
XLogRecPtr latestChunkStart
Definition: walreceiver.h:96
XLogRecPtr receiveStart
Definition: walreceiver.h:77
XLogRecPtr flushedUpto
Definition: walreceiver.h:87
sig_atomic_t force_reply
Definition: walreceiver.h:162
ConditionVariable walRcvStoppedCV
Definition: walreceiver.h:68
bool is_temp_slot
Definition: walreceiver.h:133
pg_atomic_uint64 writtenUpto
Definition: walreceiver.h:155
pg_time_t startTime
Definition: walreceiver.h:69
TimestampTz lastMsgSendTime
Definition: walreceiver.h:101
WalRcvState walRcvState
Definition: walreceiver.h:67
TimestampTz latestWalEndTime
Definition: walreceiver.h:108
bool ready_to_display
Definition: walreceiver.h:136
int sender_port
Definition: walreceiver.h:121
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:395
walrcv_receive_fn walrcv_receive
Definition: walreceiver.h:400
walrcv_server_version_fn walrcv_server_version
Definition: walreceiver.h:396
walrcv_disconnect_fn walrcv_disconnect
Definition: walreceiver.h:405
walrcv_get_backend_pid_fn walrcv_get_backend_pid
Definition: walreceiver.h:403
walrcv_connect_fn walrcv_connect
Definition: walreceiver.h:391
walrcv_get_senderinfo_fn walrcv_get_senderinfo
Definition: walreceiver.h:394
walrcv_check_conninfo_fn walrcv_check_conninfo
Definition: walreceiver.h:392
walrcv_send_fn walrcv_send
Definition: walreceiver.h:401
walrcv_get_conninfo_fn walrcv_get_conninfo
Definition: walreceiver.h:393
walrcv_exec_fn walrcv_exec
Definition: walreceiver.h:404
walrcv_startstreaming_fn walrcv_startstreaming
Definition: walreceiver.h:398
walrcv_create_slot_fn walrcv_create_slot
Definition: walreceiver.h:402
walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile
Definition: walreceiver.h:397
walrcv_endstreaming_fn walrcv_endstreaming
Definition: walreceiver.h:399
void FreeTupleDesc(TupleDesc tupdesc)
Definition: tupdesc.c:309
void tuplestore_end(Tuplestorestate *state)
Definition: tuplestore.c:453
WalRcvExecResult *(* walrcv_exec_fn)(WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
Definition: walreceiver.h:377
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:166
XLogRecPtr GetWalRcvFlushRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI)
void(* walrcv_endstreaming_fn)(WalReceiverConn *conn, TimeLineID *next_tli)
Definition: walreceiver.h:320
void WalReceiverMain(void) pg_attribute_noreturn()
Definition: walreceiver.c:187
char *(* walrcv_create_slot_fn)(WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
Definition: walreceiver.h:355
bool WalRcvStreaming(void)
struct WalRcvExecResult WalRcvExecResult
void(* walrcv_send_fn)(WalReceiverConn *conn, const char *buffer, int nbytes)
Definition: walreceiver.h:341
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:90
char *(* walrcv_get_conninfo_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:260
WalReceiverConn *(* walrcv_connect_fn)(const char *conninfo, bool logical, bool must_use_password, const char *appname, char **err)
Definition: walreceiver.h:240
char *(* walrcv_identify_system_fn)(WalReceiverConn *conn, TimeLineID *primary_tli)
Definition: walreceiver.h:280
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
pid_t(* walrcv_get_backend_pid_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:367
XLogRecPtr GetWalRcvWriteRecPtr(void)
#define MAXCONNINFO
Definition: walreceiver.h:39
void ShutdownWalRcv(void)
bool(* walrcv_startstreaming_fn)(WalReceiverConn *conn, const WalRcvStreamOptions *options)
Definition: walreceiver.h:311
int(* walrcv_receive_fn)(WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
Definition: walreceiver.h:331
void(* walrcv_readtimelinehistoryfile_fn)(WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *size)
Definition: walreceiver.h:297
PGDLLIMPORT int wal_receiver_timeout
Definition: walreceiver.c:91
static void walrcv_clear_result(WalRcvExecResult *walres)
Definition: walreceiver.h:442
void(* walrcv_check_conninfo_fn)(const char *conninfo, bool must_use_password)
Definition: walreceiver.h:251
PGDLLIMPORT WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:96
void(* walrcv_disconnect_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:387
struct WalReceiverFunctionsType WalReceiverFunctionsType
bool WalRcvRunning(void)
WalRcvState
Definition: walreceiver.h:48
@ WALRCV_STARTING
Definition: walreceiver.h:50
@ WALRCV_STOPPED
Definition: walreceiver.h:49
@ WALRCV_RESTARTING
Definition: walreceiver.h:54
@ WALRCV_STREAMING
Definition: walreceiver.h:52
@ WALRCV_WAITING
Definition: walreceiver.h:53
@ WALRCV_STOPPING
Definition: walreceiver.h:55
void(* walrcv_get_senderinfo_fn)(WalReceiverConn *conn, char **sender_host, int *sender_port)
Definition: walreceiver.h:269
int GetReplicationApplyDelay(void)
PGDLLIMPORT WalRcvData * WalRcv
void WalRcvShmemInit(void)
Size WalRcvShmemSize(void)
PGDLLIMPORT bool hot_standby_feedback
Definition: walreceiver.c:92
int(* walrcv_server_version_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:288
void WalRcvForceReply(void)
Definition: walreceiver.c:1353
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