Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : * slotsync.c
3 : : * Functionality for synchronizing slots to a standby server from the
4 : : * primary server.
5 : : *
6 : : * Copyright (c) 2024-2025, PostgreSQL Global Development Group
7 : : *
8 : : * IDENTIFICATION
9 : : * src/backend/replication/logical/slotsync.c
10 : : *
11 : : * This file contains the code for slot synchronization on a physical standby
12 : : * to fetch logical failover slots information from the primary server, create
13 : : * the slots on the standby and synchronize them periodically.
14 : : *
15 : : * Slot synchronization can be performed either automatically by enabling slot
16 : : * sync worker or manually by calling SQL function pg_sync_replication_slots().
17 : : *
18 : : * If the WAL corresponding to the remote's restart_lsn is not available on the
19 : : * physical standby or the remote's catalog_xmin precedes the oldest xid for
20 : : * which it is guaranteed that rows wouldn't have been removed then we cannot
21 : : * create the local standby slot because that would mean moving the local slot
22 : : * backward and decoding won't be possible via such a slot. In this case, the
23 : : * slot will be marked as RS_TEMPORARY. Once the primary server catches up,
24 : : * the slot will be marked as RS_PERSISTENT (which means sync-ready) after
25 : : * which slot sync worker can perform the sync periodically or user can call
26 : : * pg_sync_replication_slots() periodically to perform the syncs.
27 : : *
28 : : * If synchronized slots fail to build a consistent snapshot from the
29 : : * restart_lsn before reaching confirmed_flush_lsn, they would become
30 : : * unreliable after promotion due to potential data loss from changes
31 : : * before reaching a consistent point. This can happen because the slots can
32 : : * be synced at some random time and we may not reach the consistent point
33 : : * at the same WAL location as the primary. So, we mark such slots as
34 : : * RS_TEMPORARY. Once the decoding from corresponding LSNs can reach a
35 : : * consistent point, they will be marked as RS_PERSISTENT.
36 : : *
37 : : * The slot sync worker waits for some time before the next synchronization,
38 : : * with the duration varying based on whether any slots were updated during
39 : : * the last cycle. Refer to the comments above wait_for_slot_activity() for
40 : : * more details.
41 : : *
42 : : * If the SQL function pg_sync_replication_slots() is used to sync the slots,
43 : : * and if the slots are not ready to be synced and are marked as RS_TEMPORARY
44 : : * because of any of the reasons mentioned above, then the SQL function also
45 : : * waits and retries until the slots are marked as RS_PERSISTENT (which means
46 : : * sync-ready). Refer to the comments in SyncReplicationSlots() for more
47 : : * details.
48 : : *
49 : : * Any standby synchronized slots will be dropped if they no longer need
50 : : * to be synchronized. See comment atop drop_local_obsolete_slots() for more
51 : : * details.
52 : : *---------------------------------------------------------------------------
53 : : */
54 : :
55 : : #include "postgres.h"
56 : :
57 : : #include <time.h>
58 : :
59 : : #include "access/xlog_internal.h"
60 : : #include "access/xlogrecovery.h"
61 : : #include "catalog/pg_database.h"
62 : : #include "libpq/pqsignal.h"
63 : : #include "pgstat.h"
64 : : #include "postmaster/interrupt.h"
65 : : #include "replication/logical.h"
66 : : #include "replication/slotsync.h"
67 : : #include "replication/snapbuild.h"
68 : : #include "storage/ipc.h"
69 : : #include "storage/lmgr.h"
70 : : #include "storage/proc.h"
71 : : #include "storage/procarray.h"
72 : : #include "tcop/tcopprot.h"
73 : : #include "utils/builtins.h"
74 : : #include "utils/memutils.h"
75 : : #include "utils/pg_lsn.h"
76 : : #include "utils/ps_status.h"
77 : : #include "utils/timeout.h"
78 : :
79 : : /*
80 : : * Struct for sharing information to control slot synchronization.
81 : : *
82 : : * The 'pid' is either the slot sync worker's pid or the backend's pid running
83 : : * the SQL function pg_sync_replication_slots(). When the startup process sets
84 : : * 'stopSignaled' during promotion, it uses this 'pid' to wake up the currently
85 : : * synchronizing process so that the process can immediately stop its
86 : : * synchronizing work on seeing 'stopSignaled' set.
87 : : * Setting 'stopSignaled' is also used to handle the race condition when the
88 : : * postmaster has not noticed the promotion yet and thus may end up restarting
89 : : * the slot sync worker. If 'stopSignaled' is set, the worker will exit in such a
90 : : * case. The SQL function pg_sync_replication_slots() will also error out if
91 : : * this flag is set. Note that we don't need to reset this variable as after
92 : : * promotion the slot sync worker won't be restarted because the pmState
93 : : * changes to PM_RUN from PM_HOT_STANDBY and we don't support demoting
94 : : * primary without restarting the server. See LaunchMissingBackgroundProcesses.
95 : : *
96 : : * The 'syncing' flag is needed to prevent concurrent slot syncs to avoid slot
97 : : * overwrites.
98 : : *
99 : : * The 'last_start_time' is needed by postmaster to start the slot sync worker
100 : : * once per SLOTSYNC_RESTART_INTERVAL_SEC. In cases where an immediate restart
101 : : * is expected (e.g., slot sync GUCs change), slot sync worker will reset
102 : : * last_start_time before exiting, so that postmaster can start the worker
103 : : * without waiting for SLOTSYNC_RESTART_INTERVAL_SEC.
104 : : */
105 : : typedef struct SlotSyncCtxStruct
106 : : {
107 : : pid_t pid;
108 : : bool stopSignaled;
109 : : bool syncing;
110 : : time_t last_start_time;
111 : : slock_t mutex;
112 : : } SlotSyncCtxStruct;
113 : :
114 : : static SlotSyncCtxStruct *SlotSyncCtx = NULL;
115 : :
116 : : /* GUC variable */
117 : : bool sync_replication_slots = false;
118 : :
119 : : /*
120 : : * The sleep time (ms) between slot-sync cycles varies dynamically
121 : : * (within a MIN/MAX range) according to slot activity. See
122 : : * wait_for_slot_activity() for details.
123 : : */
124 : : #define MIN_SLOTSYNC_WORKER_NAPTIME_MS 200
125 : : #define MAX_SLOTSYNC_WORKER_NAPTIME_MS 30000 /* 30s */
126 : :
127 : : static long sleep_ms = MIN_SLOTSYNC_WORKER_NAPTIME_MS;
128 : :
129 : : /* The restart interval for slot sync work used by postmaster */
130 : : #define SLOTSYNC_RESTART_INTERVAL_SEC 10
131 : :
132 : : /*
133 : : * Flag to tell if we are syncing replication slots. Unlike the 'syncing' flag
134 : : * in SlotSyncCtxStruct, this flag is true only if the current process is
135 : : * performing slot synchronization.
136 : : */
137 : : static bool syncing_slots = false;
138 : :
139 : : /*
140 : : * Structure to hold information fetched from the primary server about a logical
141 : : * replication slot.
142 : : */
143 : : typedef struct RemoteSlot
144 : : {
145 : : char *name;
146 : : char *plugin;
147 : : char *database;
148 : : bool two_phase;
149 : : bool failover;
150 : : XLogRecPtr restart_lsn;
151 : : XLogRecPtr confirmed_lsn;
152 : : XLogRecPtr two_phase_at;
153 : : TransactionId catalog_xmin;
154 : :
155 : : /* RS_INVAL_NONE if valid, or the reason of invalidation */
156 : : ReplicationSlotInvalidationCause invalidated;
157 : : } RemoteSlot;
158 : :
159 : : static void slotsync_failure_callback(int code, Datum arg);
160 : : static void update_synced_slots_inactive_since(void);
161 : :
162 : : /*
163 : : * Update slot sync skip stats. This function requires the caller to acquire
164 : : * the slot.
165 : : */
166 : : static void
19 akapila@postgresql.o 167 :GNC 45 : update_slotsync_skip_stats(SlotSyncSkipReason skip_reason)
168 : : {
169 : : ReplicationSlot *slot;
170 : :
171 [ - + ]: 45 : Assert(MyReplicationSlot);
172 : :
173 : 45 : slot = MyReplicationSlot;
174 : :
175 : : /*
176 : : * Update the slot sync related stats in pg_stat_replication_slot when a
177 : : * slot sync is skipped
178 : : */
179 [ + + ]: 45 : if (skip_reason != SS_SKIP_NONE)
180 : 2 : pgstat_report_replslotsync(slot);
181 : :
182 : : /* Update the slot sync skip reason */
183 [ + + ]: 45 : if (slot->slotsync_skip_reason != skip_reason)
184 : : {
185 [ - + ]: 2 : SpinLockAcquire(&slot->mutex);
186 : 2 : slot->slotsync_skip_reason = skip_reason;
187 : 2 : SpinLockRelease(&slot->mutex);
188 : : }
189 : 45 : }
190 : :
191 : : /*
192 : : * If necessary, update the local synced slot's metadata based on the data
193 : : * from the remote slot.
194 : : *
195 : : * If no update was needed (the data of the remote slot is the same as the
196 : : * local slot) return false, otherwise true.
197 : : *
198 : : * *found_consistent_snapshot will be true iff the remote slot's LSN or xmin is
199 : : * modified, and decoding from the corresponding LSN's can reach a
200 : : * consistent snapshot.
201 : : *
202 : : * *remote_slot_precedes will be true if the remote slot's LSN or xmin
203 : : * precedes locally reserved position.
204 : : */
205 : : static bool
623 akapila@postgresql.o 206 :CBC 45 : update_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid,
207 : : bool *found_consistent_snapshot,
208 : : bool *remote_slot_precedes)
209 : : {
672 210 : 45 : ReplicationSlot *slot = MyReplicationSlot;
614 211 : 45 : bool updated_xmin_or_lsn = false;
212 : 45 : bool updated_config = false;
19 akapila@postgresql.o 213 :GNC 45 : SlotSyncSkipReason skip_reason = SS_SKIP_NONE;
214 : :
672 akapila@postgresql.o 215 [ - + ]:CBC 45 : Assert(slot->data.invalidated == RS_INVAL_NONE);
216 : :
623 217 [ + + ]: 45 : if (found_consistent_snapshot)
218 : 9 : *found_consistent_snapshot = false;
219 : :
614 220 [ + + ]: 45 : if (remote_slot_precedes)
221 : 9 : *remote_slot_precedes = false;
222 : :
223 : : /*
224 : : * Don't overwrite if we already have a newer catalog_xmin and
225 : : * restart_lsn.
226 : : */
227 [ + - + + ]: 90 : if (remote_slot->restart_lsn < slot->data.restart_lsn ||
228 : 45 : TransactionIdPrecedes(remote_slot->catalog_xmin,
229 : : slot->data.catalog_xmin))
230 : : {
231 : : /* Update slot sync skip stats */
19 akapila@postgresql.o 232 :GNC 2 : update_slotsync_skip_stats(SS_SKIP_WAL_OR_ROWS_REMOVED);
233 : :
234 : : /*
235 : : * This can happen in following situations:
236 : : *
237 : : * If the slot is temporary, it means either the initial WAL location
238 : : * reserved for the local slot is ahead of the remote slot's
239 : : * restart_lsn or the initial xmin_horizon computed for the local slot
240 : : * is ahead of the remote slot.
241 : : *
242 : : * If the slot is persistent, both restart_lsn and catalog_xmin of the
243 : : * synced slot could still be ahead of the remote slot. Since we use
244 : : * slot advance functionality to keep snapbuild/slot updated, it is
245 : : * possible that the restart_lsn and catalog_xmin are advanced to a
246 : : * later position than it has on the primary. This can happen when
247 : : * slot advancing machinery finds running xacts record after reaching
248 : : * the consistent state at a later point than the primary where it
249 : : * serializes the snapshot and updates the restart_lsn.
250 : : *
251 : : * We LOG the message if the slot is temporary as it can help the user
252 : : * to understand why the slot is not sync-ready. In the case of a
253 : : * persistent slot, it would be a more common case and won't directly
254 : : * impact the users, so we used DEBUG1 level to log the message.
255 : : */
614 akapila@postgresql.o 256 [ + - + - ]:GBC 2 : ereport(slot->data.persistency == RS_TEMPORARY ? LOG : DEBUG1,
257 : : errmsg("could not synchronize replication slot \"%s\"",
258 : : remote_slot->name),
259 : : errdetail("Synchronization could lead to data loss, because the remote slot needs WAL at LSN %X/%08X and catalog xmin %u, but the standby has LSN %X/%08X and catalog xmin %u.",
260 : : LSN_FORMAT_ARGS(remote_slot->restart_lsn),
261 : : remote_slot->catalog_xmin,
262 : : LSN_FORMAT_ARGS(slot->data.restart_lsn),
263 : : slot->data.catalog_xmin));
264 : :
265 [ + - ]: 2 : if (remote_slot_precedes)
266 : 2 : *remote_slot_precedes = true;
267 : :
268 : : /*
269 : : * Skip updating the configuration. This is required to avoid syncing
270 : : * two_phase_at without syncing confirmed_lsn. Otherwise, the prepared
271 : : * transaction between old confirmed_lsn and two_phase_at will
272 : : * unexpectedly get decoded and sent to the downstream after
273 : : * promotion. See comments in ReorderBufferFinishPrepared.
274 : : */
232 275 : 2 : return false;
276 : : }
277 : :
278 : : /*
279 : : * Attempt to sync LSNs and xmins only if remote slot is ahead of local
280 : : * slot.
281 : : */
232 akapila@postgresql.o 282 [ + + ]:CBC 43 : if (remote_slot->confirmed_lsn > slot->data.confirmed_flush ||
283 [ + + - + ]: 61 : remote_slot->restart_lsn > slot->data.restart_lsn ||
284 : 30 : TransactionIdFollows(remote_slot->catalog_xmin,
285 : : slot->data.catalog_xmin))
286 : : {
287 : : /*
288 : : * We can't directly copy the remote slot's LSN or xmin unless there
289 : : * exists a consistent snapshot at that point. Otherwise, after
290 : : * promotion, the slots may not reach a consistent point before the
291 : : * confirmed_flush_lsn which can lead to a data loss. To avoid data
292 : : * loss, we let slot machinery advance the slot which ensures that
293 : : * snapbuilder/slot statuses are updated properly.
294 : : */
623 295 [ + + ]: 13 : if (SnapBuildSnapshotExists(remote_slot->restart_lsn))
296 : : {
297 : : /*
298 : : * Update the slot info directly if there is a serialized snapshot
299 : : * at the restart_lsn, as the slot can quickly reach consistency
300 : : * at restart_lsn by restoring the snapshot.
301 : : */
302 [ - + ]: 3 : SpinLockAcquire(&slot->mutex);
303 : 3 : slot->data.restart_lsn = remote_slot->restart_lsn;
304 : 3 : slot->data.confirmed_flush = remote_slot->confirmed_lsn;
305 : 3 : slot->data.catalog_xmin = remote_slot->catalog_xmin;
306 : 3 : SpinLockRelease(&slot->mutex);
307 : :
308 [ + + ]: 3 : if (found_consistent_snapshot)
623 akapila@postgresql.o 309 :GBC 1 : *found_consistent_snapshot = true;
310 : : }
311 : : else
312 : : {
623 akapila@postgresql.o 313 :CBC 10 : LogicalSlotAdvanceAndCheckSnapState(remote_slot->confirmed_lsn,
314 : : found_consistent_snapshot);
315 : :
316 : : /* Sanity check */
614 317 [ - + ]: 10 : if (slot->data.confirmed_flush != remote_slot->confirmed_lsn)
614 akapila@postgresql.o 318 [ # # ]:UBC 0 : ereport(ERROR,
319 : : errmsg_internal("synchronized confirmed_flush for slot \"%s\" differs from remote slot",
320 : : remote_slot->name),
321 : : errdetail_internal("Remote slot has LSN %X/%08X but local slot has LSN %X/%08X.",
322 : : LSN_FORMAT_ARGS(remote_slot->confirmed_lsn),
323 : : LSN_FORMAT_ARGS(slot->data.confirmed_flush)));
324 : :
325 : : /*
326 : : * If we can't reach a consistent snapshot, the slot won't be
327 : : * persisted. See update_and_persist_local_synced_slot().
328 : : */
22 akapila@postgresql.o 329 [ + + - + ]:GNC 10 : if (found_consistent_snapshot && !(*found_consistent_snapshot))
19 akapila@postgresql.o 330 :UNC 0 : skip_reason = SS_SKIP_NO_CONSISTENT_SNAPSHOT;
331 : : }
332 : :
614 akapila@postgresql.o 333 :CBC 13 : updated_xmin_or_lsn = true;
334 : : }
335 : :
336 : : /* Update slot sync skip stats */
19 akapila@postgresql.o 337 :GNC 43 : update_slotsync_skip_stats(skip_reason);
338 : :
623 akapila@postgresql.o 339 [ + - ]:CBC 43 : if (remote_dbid != slot->data.database ||
340 [ + + ]: 43 : remote_slot->two_phase != slot->data.two_phase ||
341 [ + - ]: 42 : remote_slot->failover != slot->data.failover ||
258 342 [ + - ]: 42 : strcmp(remote_slot->plugin, NameStr(slot->data.plugin)) != 0 ||
343 [ - + ]: 42 : remote_slot->two_phase_at != slot->data.two_phase_at)
344 : : {
345 : : NameData plugin_name;
346 : :
347 : : /* Avoid expensive operations while holding a spinlock. */
623 348 : 1 : namestrcpy(&plugin_name, remote_slot->plugin);
349 : :
350 [ - + ]: 1 : SpinLockAcquire(&slot->mutex);
351 : 1 : slot->data.plugin = plugin_name;
352 : 1 : slot->data.database = remote_dbid;
353 : 1 : slot->data.two_phase = remote_slot->two_phase;
258 354 : 1 : slot->data.two_phase_at = remote_slot->two_phase_at;
623 355 : 1 : slot->data.failover = remote_slot->failover;
356 : 1 : SpinLockRelease(&slot->mutex);
357 : :
614 358 : 1 : updated_config = true;
359 : :
360 : : /*
361 : : * Ensure that there is no risk of sending prepared transactions
362 : : * unexpectedly after the promotion.
363 : : */
232 364 [ - + ]: 1 : Assert(slot->data.two_phase_at <= slot->data.confirmed_flush);
365 : : }
366 : :
367 : : /*
368 : : * We have to write the changed xmin to disk *before* we change the
369 : : * in-memory value, otherwise after a crash we wouldn't know that some
370 : : * catalog tuples might have been removed already.
371 : : */
614 372 [ + + + + ]: 43 : if (updated_config || updated_xmin_or_lsn)
373 : : {
374 : 14 : ReplicationSlotMarkDirty();
375 : 14 : ReplicationSlotSave();
376 : : }
377 : :
378 : : /*
379 : : * Now the new xmin is safely on disk, we can let the global value
380 : : * advance. We do not take ProcArrayLock or similar since we only advance
381 : : * xmin here and there's not much harm done by a concurrent computation
382 : : * missing that.
383 : : */
384 [ + + ]: 43 : if (updated_xmin_or_lsn)
385 : : {
386 [ - + ]: 13 : SpinLockAcquire(&slot->mutex);
387 : 13 : slot->effective_catalog_xmin = remote_slot->catalog_xmin;
388 : 13 : SpinLockRelease(&slot->mutex);
389 : :
390 : 13 : ReplicationSlotsComputeRequiredXmin(false);
391 : 13 : ReplicationSlotsComputeRequiredLSN();
392 : : }
393 : :
394 [ + + + + ]: 43 : return updated_config || updated_xmin_or_lsn;
395 : : }
396 : :
397 : : /*
398 : : * Get the list of local logical slots that are synchronized from the
399 : : * primary server.
400 : : */
401 : : static List *
672 402 : 25 : get_local_synced_slots(void)
403 : : {
404 : 25 : List *local_slots = NIL;
405 : :
406 : 25 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
407 : :
408 [ + + ]: 275 : for (int i = 0; i < max_replication_slots; i++)
409 : : {
410 : 250 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
411 : :
412 : : /* Check if it is a synchronized slot */
413 [ + + + + ]: 250 : if (s->in_use && s->data.synced)
414 : : {
415 [ - + ]: 41 : Assert(SlotIsLogical(s));
416 : 41 : local_slots = lappend(local_slots, s);
417 : : }
418 : : }
419 : :
420 : 25 : LWLockRelease(ReplicationSlotControlLock);
421 : :
422 : 25 : return local_slots;
423 : : }
424 : :
425 : : /*
426 : : * Helper function to check if local_slot is required to be retained.
427 : : *
428 : : * Return false either if local_slot does not exist in the remote_slots list
429 : : * or is invalidated while the corresponding remote slot is still valid,
430 : : * otherwise true.
431 : : */
432 : : static bool
433 : 41 : local_sync_slot_required(ReplicationSlot *local_slot, List *remote_slots)
434 : : {
435 : 41 : bool remote_exists = false;
436 : 41 : bool locally_invalidated = false;
437 : :
438 [ + - + + : 102 : foreach_ptr(RemoteSlot, remote_slot, remote_slots)
+ + ]
439 : : {
440 [ + + ]: 59 : if (strcmp(remote_slot->name, NameStr(local_slot->data.name)) == 0)
441 : : {
442 : 39 : remote_exists = true;
443 : :
444 : : /*
445 : : * If remote slot is not invalidated but local slot is marked as
446 : : * invalidated, then set locally_invalidated flag.
447 : : */
448 [ - + ]: 39 : SpinLockAcquire(&local_slot->mutex);
449 : 39 : locally_invalidated =
450 [ + - ]: 78 : (remote_slot->invalidated == RS_INVAL_NONE) &&
451 [ + + ]: 39 : (local_slot->data.invalidated != RS_INVAL_NONE);
452 : 39 : SpinLockRelease(&local_slot->mutex);
453 : :
454 : 39 : break;
455 : : }
456 : : }
457 : :
458 [ + + + + ]: 41 : return (remote_exists && !locally_invalidated);
459 : : }
460 : :
461 : : /*
462 : : * Drop local obsolete slots.
463 : : *
464 : : * Drop the local slots that no longer need to be synced i.e. these either do
465 : : * not exist on the primary or are no longer enabled for failover.
466 : : *
467 : : * Additionally, drop any slots that are valid on the primary but got
468 : : * invalidated on the standby. This situation may occur due to the following
469 : : * reasons:
470 : : * - The 'max_slot_wal_keep_size' on the standby is insufficient to retain WAL
471 : : * records from the restart_lsn of the slot.
472 : : * - 'primary_slot_name' is temporarily reset to null and the physical slot is
473 : : * removed.
474 : : * These dropped slots will get recreated in next sync-cycle and it is okay to
475 : : * drop and recreate such slots as long as these are not consumable on the
476 : : * standby (which is the case currently).
477 : : *
478 : : * Note: Change of 'wal_level' on the primary server to a level lower than
479 : : * logical may also result in slot invalidation and removal on the standby.
480 : : * This is because such 'wal_level' change is only possible if the logical
481 : : * slots are removed on the primary server, so it's expected to see the
482 : : * slots being invalidated and removed on the standby too (and re-created
483 : : * if they are re-created on the primary server).
484 : : */
485 : : static void
486 : 25 : drop_local_obsolete_slots(List *remote_slot_list)
487 : : {
488 : 25 : List *local_slots = get_local_synced_slots();
489 : :
490 [ + + + + : 91 : foreach_ptr(ReplicationSlot, local_slot, local_slots)
+ + ]
491 : : {
492 : : /* Drop the local slot if it is not required to be retained. */
493 [ + + ]: 41 : if (!local_sync_slot_required(local_slot, remote_slot_list))
494 : : {
495 : : bool synced_slot;
496 : :
497 : : /*
498 : : * Use shared lock to prevent a conflict with
499 : : * ReplicationSlotsDropDBSlots(), trying to drop the same slot
500 : : * during a drop-database operation.
501 : : */
502 : 3 : LockSharedObject(DatabaseRelationId, local_slot->data.database,
503 : : 0, AccessShareLock);
504 : :
505 : : /*
506 : : * In the small window between getting the slot to drop and
507 : : * locking the database, there is a possibility of a parallel
508 : : * database drop by the startup process and the creation of a new
509 : : * slot by the user. This new user-created slot may end up using
510 : : * the same shared memory as that of 'local_slot'. Thus check if
511 : : * local_slot is still the synced one before performing actual
512 : : * drop.
513 : : */
514 [ - + ]: 3 : SpinLockAcquire(&local_slot->mutex);
515 [ + - + - ]: 3 : synced_slot = local_slot->in_use && local_slot->data.synced;
516 : 3 : SpinLockRelease(&local_slot->mutex);
517 : :
518 [ + - ]: 3 : if (synced_slot)
519 : : {
320 520 : 3 : ReplicationSlotAcquire(NameStr(local_slot->data.name), true, false);
672 521 : 3 : ReplicationSlotDropAcquired();
522 : : }
523 : :
524 : 3 : UnlockSharedObject(DatabaseRelationId, local_slot->data.database,
525 : : 0, AccessShareLock);
526 : :
527 [ + - ]: 3 : ereport(LOG,
528 : : errmsg("dropped replication slot \"%s\" of database with OID %u",
529 : : NameStr(local_slot->data.name),
530 : : local_slot->data.database));
531 : : }
532 : : }
533 : 25 : }
534 : :
535 : : /*
536 : : * Reserve WAL for the currently active local slot using the specified WAL
537 : : * location (restart_lsn).
538 : : *
539 : : * If the given WAL location has been removed, reserve WAL using the oldest
540 : : * existing WAL segment.
541 : : */
542 : : static void
543 : 7 : reserve_wal_for_local_slot(XLogRecPtr restart_lsn)
544 : : {
545 : : XLogSegNo oldest_segno;
546 : : XLogSegNo segno;
547 : 7 : ReplicationSlot *slot = MyReplicationSlot;
548 : :
549 [ - + ]: 7 : Assert(slot != NULL);
41 alvherre@kurilemu.de 550 [ + - ]:GNC 7 : Assert(!XLogRecPtrIsValid(slot->data.restart_lsn));
551 : :
552 : : while (true)
553 : : {
672 akapila@postgresql.o 554 [ - + ]:CBC 7 : SpinLockAcquire(&slot->mutex);
555 : 7 : slot->data.restart_lsn = restart_lsn;
556 : 7 : SpinLockRelease(&slot->mutex);
557 : :
558 : : /* Prevent WAL removal as fast as possible */
559 : 7 : ReplicationSlotsComputeRequiredLSN();
560 : :
561 : 7 : XLByteToSeg(slot->data.restart_lsn, segno, wal_segment_size);
562 : :
563 : : /*
564 : : * Find the oldest existing WAL segment file.
565 : : *
566 : : * Normally, we can determine it by using the last removed segment
567 : : * number. However, if no WAL segment files have been removed by a
568 : : * checkpoint since startup, we need to search for the oldest segment
569 : : * file from the current timeline existing in XLOGDIR.
570 : : *
571 : : * XXX: Currently, we are searching for the oldest segment in the
572 : : * current timeline as there is less chance of the slot's restart_lsn
573 : : * from being some prior timeline, and even if it happens, in the
574 : : * worst case, we will wait to sync till the slot's restart_lsn moved
575 : : * to the current timeline.
576 : : */
577 : 7 : oldest_segno = XLogGetLastRemovedSegno() + 1;
578 : :
579 [ + + ]: 7 : if (oldest_segno == 1)
580 : : {
581 : : TimeLineID cur_timeline;
582 : :
583 : 5 : GetWalRcvFlushRecPtr(NULL, &cur_timeline);
584 : 5 : oldest_segno = XLogGetOldestSegno(cur_timeline);
585 : : }
586 : :
670 587 [ + + ]: 7 : elog(DEBUG1, "segno: " UINT64_FORMAT " of purposed restart_lsn for the synced slot, oldest_segno: " UINT64_FORMAT " available",
588 : : segno, oldest_segno);
589 : :
590 : : /*
591 : : * If all required WAL is still there, great, otherwise retry. The
592 : : * slot should prevent further removal of WAL, unless there's a
593 : : * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
594 : : * the new restart_lsn above, so normally we should never need to loop
595 : : * more than twice.
596 : : */
672 597 [ + - ]: 7 : if (segno >= oldest_segno)
598 : 7 : break;
599 : :
600 : : /* Retry using the location of the oldest wal segment */
672 akapila@postgresql.o 601 :UBC 0 : XLogSegNoOffsetToRecPtr(oldest_segno, 0, wal_segment_size, restart_lsn);
602 : : }
672 akapila@postgresql.o 603 :CBC 7 : }
604 : :
605 : : /*
606 : : * If the remote restart_lsn and catalog_xmin have caught up with the
607 : : * local ones, then update the LSNs and persist the local synced slot for
608 : : * future synchronization; otherwise, do nothing.
609 : : *
610 : : * *slot_persistence_pending is set to true if any of the slots fail to
611 : : * persist.
612 : : *
613 : : * Return true if the slot is marked as RS_PERSISTENT (sync-ready), otherwise
614 : : * false.
615 : : */
616 : : static bool
2 akapila@postgresql.o 617 :GNC 9 : update_and_persist_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid,
618 : : bool *slot_persistence_pending)
619 : : {
672 akapila@postgresql.o 620 :CBC 9 : ReplicationSlot *slot = MyReplicationSlot;
623 621 : 9 : bool found_consistent_snapshot = false;
614 622 : 9 : bool remote_slot_precedes = false;
623 : :
624 : : /* Slotsync skip stats are handled in function update_local_synced_slot() */
625 : 9 : (void) update_local_synced_slot(remote_slot, remote_dbid,
626 : : &found_consistent_snapshot,
627 : : &remote_slot_precedes);
628 : :
629 : : /*
630 : : * Check if the primary server has caught up. Refer to the comment atop
631 : : * the file for details on this check.
632 : : */
633 [ + + ]: 9 : if (remote_slot_precedes)
634 : : {
635 : : /*
636 : : * The remote slot didn't catch up to locally reserved position.
637 : : *
638 : : * We do not drop the slot because the restart_lsn can be ahead of the
639 : : * current location when recreating the slot in the next cycle. It may
640 : : * take more time to create such a slot. Therefore, we keep this slot
641 : : * and attempt the synchronization in the next cycle.
642 : : *
643 : : * We also update the slot_persistence_pending parameter, so the SQL
644 : : * function can retry.
645 : : */
2 akapila@postgresql.o 646 [ + - ]:GNC 2 : if (slot_persistence_pending)
647 : 2 : *slot_persistence_pending = true;
648 : :
664 akapila@postgresql.o 649 :GBC 2 : return false;
650 : : }
651 : :
652 : : /*
653 : : * Don't persist the slot if it cannot reach the consistent point from the
654 : : * restart_lsn. See comments atop this file.
655 : : */
623 akapila@postgresql.o 656 [ - + ]:CBC 7 : if (!found_consistent_snapshot)
657 : : {
623 akapila@postgresql.o 658 [ # # ]:UBC 0 : ereport(LOG,
659 : : errmsg("could not synchronize replication slot \"%s\"", remote_slot->name),
660 : : errdetail("Synchronization could lead to data loss, because the standby could not build a consistent snapshot to decode WALs at LSN %X/%08X.",
661 : : LSN_FORMAT_ARGS(slot->data.restart_lsn)));
662 : :
663 : : /* Set this, so that SQL function can retry */
2 akapila@postgresql.o 664 [ # # ]:UNC 0 : if (slot_persistence_pending)
665 : 0 : *slot_persistence_pending = true;
666 : :
623 akapila@postgresql.o 667 :UBC 0 : return false;
668 : : }
669 : :
672 akapila@postgresql.o 670 :CBC 7 : ReplicationSlotPersist();
671 : :
672 [ + - ]: 7 : ereport(LOG,
673 : : errmsg("newly created replication slot \"%s\" is sync-ready now",
674 : : remote_slot->name));
675 : :
664 676 : 7 : return true;
677 : : }
678 : :
679 : : /*
680 : : * Synchronize a single slot to the given position.
681 : : *
682 : : * This creates a new slot if there is no existing one and updates the
683 : : * metadata of the slot as per the data received from the primary server.
684 : : *
685 : : * The slot is created as a temporary slot and stays in the same state until the
686 : : * remote_slot catches up with locally reserved position and local slot is
687 : : * updated. The slot is then persisted and is considered as sync-ready for
688 : : * periodic syncs.
689 : : *
690 : : * *slot_persistence_pending is set to true if any of the slots fail to
691 : : * persist.
692 : : *
693 : : * Returns TRUE if the local slot is updated.
694 : : */
695 : : static bool
2 akapila@postgresql.o 696 :GNC 45 : synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid,
697 : : bool *slot_persistence_pending)
698 : : {
699 : : ReplicationSlot *slot;
22 700 : 45 : XLogRecPtr latestFlushPtr = GetStandbyFlushRecPtr(NULL);
664 akapila@postgresql.o 701 :CBC 45 : bool slot_updated = false;
702 : :
703 : : /* Search for the named slot */
672 704 [ + + ]: 45 : if ((slot = SearchNamedReplicationSlot(remote_slot->name, true)))
705 : : {
706 : : bool synced;
707 : :
708 [ - + ]: 38 : SpinLockAcquire(&slot->mutex);
709 : 38 : synced = slot->data.synced;
710 : 38 : SpinLockRelease(&slot->mutex);
711 : :
712 : : /* User-created slot with the same name exists, raise ERROR. */
713 [ - + ]: 38 : if (!synced)
672 akapila@postgresql.o 714 [ # # ]:UBC 0 : ereport(ERROR,
715 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
716 : : errmsg("exiting from slot synchronization because same"
717 : : " name slot \"%s\" already exists on the standby",
718 : : remote_slot->name));
719 : :
720 : : /*
721 : : * The slot has been synchronized before.
722 : : *
723 : : * It is important to acquire the slot here before checking
724 : : * invalidation. If we don't acquire the slot first, there could be a
725 : : * race condition that the local slot could be invalidated just after
726 : : * checking the 'invalidated' flag here and we could end up
727 : : * overwriting 'invalidated' flag to remote_slot's value. See
728 : : * InvalidatePossiblyObsoleteSlot() where it invalidates slot directly
729 : : * if the slot is not acquired by other processes.
730 : : *
731 : : * XXX: If it ever turns out that slot acquire/release is costly for
732 : : * cases when none of the slot properties is changed then we can do a
733 : : * pre-check to ensure that at least one of the slot properties is
734 : : * changed before acquiring the slot.
735 : : */
320 akapila@postgresql.o 736 :CBC 38 : ReplicationSlotAcquire(remote_slot->name, true, false);
737 : :
672 738 [ - + ]: 38 : Assert(slot == MyReplicationSlot);
739 : :
740 : : /*
741 : : * Copy the invalidation cause from remote only if local slot is not
742 : : * invalidated locally, we don't want to overwrite existing one.
743 : : */
744 [ + - ]: 38 : if (slot->data.invalidated == RS_INVAL_NONE &&
745 [ - + ]: 38 : remote_slot->invalidated != RS_INVAL_NONE)
746 : : {
672 akapila@postgresql.o 747 [ # # ]:UBC 0 : SpinLockAcquire(&slot->mutex);
748 : 0 : slot->data.invalidated = remote_slot->invalidated;
749 : 0 : SpinLockRelease(&slot->mutex);
750 : :
751 : : /* Make sure the invalidated state persists across server restart */
752 : 0 : ReplicationSlotMarkDirty();
753 : 0 : ReplicationSlotSave();
754 : :
664 755 : 0 : slot_updated = true;
756 : : }
757 : :
758 : : /* Skip the sync of an invalidated slot */
672 akapila@postgresql.o 759 [ - + ]:CBC 38 : if (slot->data.invalidated != RS_INVAL_NONE)
760 : : {
19 akapila@postgresql.o 761 :UNC 0 : update_slotsync_skip_stats(SS_SKIP_INVALID);
762 : :
672 763 : 0 : ReplicationSlotRelease();
664 764 : 0 : return slot_updated;
765 : : }
766 : :
767 : : /*
768 : : * Make sure that concerned WAL is received and flushed before syncing
769 : : * slot to target lsn received from the primary server.
770 : : *
771 : : * Report statistics only after the slot has been acquired, ensuring
772 : : * it cannot be dropped during the reporting process.
773 : : */
22 akapila@postgresql.o 774 [ - + ]:GNC 38 : if (remote_slot->confirmed_lsn > latestFlushPtr)
775 : : {
19 akapila@postgresql.o 776 :UNC 0 : update_slotsync_skip_stats(SS_SKIP_WAL_NOT_FLUSHED);
777 : :
778 : : /*
779 : : * Can get here only if GUC 'synchronized_standby_slots' on the
780 : : * primary server was not configured correctly.
781 : : */
22 782 [ # # # # ]: 0 : ereport(AmLogicalSlotSyncWorkerProcess() ? LOG : ERROR,
783 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
784 : : errmsg("skipping slot synchronization because the received slot sync"
785 : : " LSN %X/%08X for slot \"%s\" is ahead of the standby position %X/%08X",
786 : : LSN_FORMAT_ARGS(remote_slot->confirmed_lsn),
787 : : remote_slot->name,
788 : : LSN_FORMAT_ARGS(latestFlushPtr)));
789 : :
22 akapila@postgresql.o 790 :UBC 0 : ReplicationSlotRelease();
791 : :
792 : 0 : return slot_updated;
793 : : }
794 : :
795 : : /* Slot not ready yet, let's attempt to make it sync-ready now. */
672 akapila@postgresql.o 796 [ + + ]:CBC 38 : if (slot->data.persistency == RS_TEMPORARY)
797 : : {
664 akapila@postgresql.o 798 :GBC 2 : slot_updated = update_and_persist_local_synced_slot(remote_slot,
799 : : remote_dbid,
800 : : slot_persistence_pending);
801 : : }
802 : :
803 : : /* Slot ready for sync, so sync it. */
804 : : else
805 : : {
806 : : /*
807 : : * Sanity check: As long as the invalidations are handled
808 : : * appropriately as above, this should never happen.
809 : : *
810 : : * We don't need to check restart_lsn here. See the comments in
811 : : * update_local_synced_slot() for details.
812 : : */
614 akapila@postgresql.o 813 [ - + ]:CBC 36 : if (remote_slot->confirmed_lsn < slot->data.confirmed_flush)
614 akapila@postgresql.o 814 [ # # ]:UBC 0 : ereport(ERROR,
815 : : errmsg_internal("cannot synchronize local slot \"%s\"",
816 : : remote_slot->name),
817 : : errdetail_internal("Local slot's start streaming location LSN(%X/%08X) is ahead of remote slot's LSN(%X/%08X).",
818 : : LSN_FORMAT_ARGS(slot->data.confirmed_flush),
819 : : LSN_FORMAT_ARGS(remote_slot->confirmed_lsn)));
820 : :
614 akapila@postgresql.o 821 :CBC 36 : slot_updated = update_local_synced_slot(remote_slot, remote_dbid,
822 : : NULL, NULL);
823 : : }
824 : : }
825 : : /* Otherwise create the slot first. */
826 : : else
827 : : {
828 : : NameData plugin_name;
672 829 : 7 : TransactionId xmin_horizon = InvalidTransactionId;
830 : :
831 : : /* Skip creating the local slot if remote_slot is invalidated already */
832 [ - + ]: 7 : if (remote_slot->invalidated != RS_INVAL_NONE)
664 akapila@postgresql.o 833 :UBC 0 : return false;
834 : :
835 : : /*
836 : : * We create temporary slots instead of ephemeral slots here because
837 : : * we want the slots to survive after releasing them. This is done to
838 : : * avoid dropping and re-creating the slots in each synchronization
839 : : * cycle if the restart_lsn or catalog_xmin of the remote slot has not
840 : : * caught up.
841 : : */
672 akapila@postgresql.o 842 :CBC 7 : ReplicationSlotCreate(remote_slot->name, true, RS_TEMPORARY,
843 : 7 : remote_slot->two_phase,
844 : 7 : remote_slot->failover,
845 : : true);
846 : :
847 : : /* For shorter lines. */
848 : 7 : slot = MyReplicationSlot;
849 : :
850 : : /* Avoid expensive operations while holding a spinlock. */
851 : 7 : namestrcpy(&plugin_name, remote_slot->plugin);
852 : :
853 [ - + ]: 7 : SpinLockAcquire(&slot->mutex);
854 : 7 : slot->data.database = remote_dbid;
855 : 7 : slot->data.plugin = plugin_name;
856 : 7 : SpinLockRelease(&slot->mutex);
857 : :
858 : 7 : reserve_wal_for_local_slot(remote_slot->restart_lsn);
859 : :
860 : 7 : LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
861 : 7 : xmin_horizon = GetOldestSafeDecodingTransactionId(true);
862 [ - + ]: 7 : SpinLockAcquire(&slot->mutex);
863 : 7 : slot->effective_catalog_xmin = xmin_horizon;
864 : 7 : slot->data.catalog_xmin = xmin_horizon;
865 : 7 : SpinLockRelease(&slot->mutex);
866 : 7 : ReplicationSlotsComputeRequiredXmin(true);
867 : 7 : LWLockRelease(ProcArrayLock);
868 : :
869 : : /*
870 : : * Make sure that concerned WAL is received and flushed before syncing
871 : : * slot to target lsn received from the primary server.
872 : : *
873 : : * Report statistics only after the slot has been acquired, ensuring
874 : : * it cannot be dropped during the reporting process.
875 : : */
22 akapila@postgresql.o 876 [ - + ]:GNC 7 : if (remote_slot->confirmed_lsn > latestFlushPtr)
877 : : {
19 akapila@postgresql.o 878 :UNC 0 : update_slotsync_skip_stats(SS_SKIP_WAL_NOT_FLUSHED);
879 : :
880 : : /*
881 : : * Can get here only if GUC 'synchronized_standby_slots' on the
882 : : * primary server was not configured correctly.
883 : : */
22 884 [ # # # # ]: 0 : ereport(AmLogicalSlotSyncWorkerProcess() ? LOG : ERROR,
885 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
886 : : errmsg("skipping slot synchronization because the received slot sync"
887 : : " LSN %X/%08X for slot \"%s\" is ahead of the standby position %X/%08X",
888 : : LSN_FORMAT_ARGS(remote_slot->confirmed_lsn),
889 : : remote_slot->name,
890 : : LSN_FORMAT_ARGS(latestFlushPtr)));
891 : :
892 : 0 : ReplicationSlotRelease();
893 : :
894 : 0 : return false;
895 : : }
896 : :
2 akapila@postgresql.o 897 :GNC 7 : update_and_persist_local_synced_slot(remote_slot, remote_dbid,
898 : : slot_persistence_pending);
899 : :
664 akapila@postgresql.o 900 :CBC 7 : slot_updated = true;
901 : : }
902 : :
672 903 : 45 : ReplicationSlotRelease();
904 : :
664 905 : 45 : return slot_updated;
906 : : }
907 : :
908 : : /*
909 : : * Fetch remote slots.
910 : : *
911 : : * If slot_names is NIL, fetches all failover logical slots from the
912 : : * primary server, otherwise fetches only the ones with names in slot_names.
913 : : *
914 : : * Returns a list of remote slot information structures, or NIL if none
915 : : * are found.
916 : : */
917 : : static List *
2 akapila@postgresql.o 918 :GNC 25 : fetch_remote_slots(WalReceiverConn *wrconn, List *slot_names)
919 : : {
920 : : #define SLOTSYNC_COLUMN_COUNT 10
672 akapila@postgresql.o 921 :CBC 25 : Oid slotRow[SLOTSYNC_COLUMN_COUNT] = {TEXTOID, TEXTOID, LSNOID,
922 : : LSNOID, XIDOID, BOOLOID, LSNOID, BOOLOID, TEXTOID, TEXTOID};
923 : :
924 : : WalRcvExecResult *res;
925 : : TupleTableSlot *tupslot;
926 : 25 : List *remote_slot_list = NIL;
927 : : StringInfoData query;
928 : :
2 akapila@postgresql.o 929 :GNC 25 : initStringInfo(&query);
930 : 25 : appendStringInfoString(&query,
931 : : "SELECT slot_name, plugin, confirmed_flush_lsn,"
932 : : " restart_lsn, catalog_xmin, two_phase,"
933 : : " two_phase_at, failover,"
934 : : " database, invalidation_reason"
935 : : " FROM pg_catalog.pg_replication_slots"
936 : : " WHERE failover and NOT temporary");
937 : :
938 [ + + ]: 25 : if (slot_names != NIL)
939 : : {
940 : 2 : bool first_slot = true;
941 : :
942 : : /*
943 : : * Construct the query to fetch only the specified slots
944 : : */
945 : 2 : appendStringInfoString(&query, " AND slot_name IN (");
946 : :
947 [ + - + + : 6 : foreach_ptr(char, slot_name, slot_names)
+ + ]
948 : : {
949 [ - + ]: 2 : if (!first_slot)
2 akapila@postgresql.o 950 :UNC 0 : appendStringInfoString(&query, ", ");
951 : :
2 akapila@postgresql.o 952 :GNC 2 : appendStringInfo(&query, "%s", quote_literal_cstr(slot_name));
953 : 2 : first_slot = false;
954 : : }
955 : 2 : appendStringInfoChar(&query, ')');
956 : : }
957 : :
958 : : /* Execute the query */
959 : 25 : res = walrcv_exec(wrconn, query.data, SLOTSYNC_COLUMN_COUNT, slotRow);
960 : 25 : pfree(query.data);
672 akapila@postgresql.o 961 [ - + ]:CBC 25 : if (res->status != WALRCV_OK_TUPLES)
672 akapila@postgresql.o 962 [ # # ]:UBC 0 : ereport(ERROR,
963 : : errmsg("could not fetch failover logical slots info from the primary server: %s",
964 : : res->err));
965 : :
672 akapila@postgresql.o 966 :CBC 25 : tupslot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
967 [ + + ]: 70 : while (tuplestore_gettupleslot(res->tuplestore, true, false, tupslot))
968 : : {
969 : : bool isnull;
7 michael@paquier.xyz 970 :GNC 45 : RemoteSlot *remote_slot = palloc0_object(RemoteSlot);
971 : : Datum d;
672 akapila@postgresql.o 972 :CBC 45 : int col = 0;
973 : :
974 : 45 : remote_slot->name = TextDatumGetCString(slot_getattr(tupslot, ++col,
975 : : &isnull));
976 [ - + ]: 45 : Assert(!isnull);
977 : :
978 : 45 : remote_slot->plugin = TextDatumGetCString(slot_getattr(tupslot, ++col,
979 : : &isnull));
980 [ - + ]: 45 : Assert(!isnull);
981 : :
982 : : /*
983 : : * It is possible to get null values for LSN and Xmin if slot is
984 : : * invalidated on the primary server, so handle accordingly.
985 : : */
986 : 45 : d = slot_getattr(tupslot, ++col, &isnull);
987 [ + - ]: 45 : remote_slot->confirmed_lsn = isnull ? InvalidXLogRecPtr :
988 : 45 : DatumGetLSN(d);
989 : :
990 : 45 : d = slot_getattr(tupslot, ++col, &isnull);
991 [ + - ]: 45 : remote_slot->restart_lsn = isnull ? InvalidXLogRecPtr : DatumGetLSN(d);
992 : :
993 : 45 : d = slot_getattr(tupslot, ++col, &isnull);
994 [ + - ]: 45 : remote_slot->catalog_xmin = isnull ? InvalidTransactionId :
995 : 45 : DatumGetTransactionId(d);
996 : :
997 : 45 : remote_slot->two_phase = DatumGetBool(slot_getattr(tupslot, ++col,
998 : : &isnull));
999 [ - + ]: 45 : Assert(!isnull);
1000 : :
258 1001 : 45 : d = slot_getattr(tupslot, ++col, &isnull);
1002 [ + + ]: 45 : remote_slot->two_phase_at = isnull ? InvalidXLogRecPtr : DatumGetLSN(d);
1003 : :
672 1004 : 45 : remote_slot->failover = DatumGetBool(slot_getattr(tupslot, ++col,
1005 : : &isnull));
1006 [ - + ]: 45 : Assert(!isnull);
1007 : :
1008 : 45 : remote_slot->database = TextDatumGetCString(slot_getattr(tupslot,
1009 : : ++col, &isnull));
1010 [ - + ]: 45 : Assert(!isnull);
1011 : :
1012 : 45 : d = slot_getattr(tupslot, ++col, &isnull);
1013 [ - + ]: 45 : remote_slot->invalidated = isnull ? RS_INVAL_NONE :
672 akapila@postgresql.o 1014 :UBC 0 : GetSlotInvalidationCause(TextDatumGetCString(d));
1015 : :
1016 : : /* Sanity check */
672 akapila@postgresql.o 1017 [ - + ]:CBC 45 : Assert(col == SLOTSYNC_COLUMN_COUNT);
1018 : :
1019 : : /*
1020 : : * If restart_lsn, confirmed_lsn or catalog_xmin is invalid but the
1021 : : * slot is valid, that means we have fetched the remote_slot in its
1022 : : * RS_EPHEMERAL state. In such a case, don't sync it; we can always
1023 : : * sync it in the next sync cycle when the remote_slot is persisted
1024 : : * and has valid lsn(s) and xmin values.
1025 : : *
1026 : : * XXX: In future, if we plan to expose 'slot->data.persistency' in
1027 : : * pg_replication_slots view, then we can avoid fetching RS_EPHEMERAL
1028 : : * slots in the first place.
1029 : : */
41 alvherre@kurilemu.de 1030 [ + - ]:GNC 45 : if ((!XLogRecPtrIsValid(remote_slot->restart_lsn) ||
1031 [ + - ]: 45 : !XLogRecPtrIsValid(remote_slot->confirmed_lsn) ||
672 akapila@postgresql.o 1032 [ - + ]:CBC 45 : !TransactionIdIsValid(remote_slot->catalog_xmin)) &&
672 akapila@postgresql.o 1033 [ # # ]:UBC 0 : remote_slot->invalidated == RS_INVAL_NONE)
1034 : 0 : pfree(remote_slot);
1035 : : else
1036 : : /* Create list of remote slots */
672 akapila@postgresql.o 1037 :CBC 45 : remote_slot_list = lappend(remote_slot_list, remote_slot);
1038 : :
1039 : 45 : ExecClearTuple(tupslot);
1040 : : }
1041 : :
2 akapila@postgresql.o 1042 :GNC 25 : walrcv_clear_result(res);
1043 : :
1044 : 25 : return remote_slot_list;
1045 : : }
1046 : :
1047 : : /*
1048 : : * Synchronize slots.
1049 : : *
1050 : : * This function takes a list of remote slots and synchronizes them locally. It
1051 : : * creates the slots if not present on the standby and updates existing ones.
1052 : : *
1053 : : * If slot_persistence_pending is not NULL, it will be set to true if one or
1054 : : * more slots could not be persisted. This allows callers such as
1055 : : * SyncReplicationSlots() to retry those slots.
1056 : : *
1057 : : * Returns TRUE if any of the slots gets updated in this sync-cycle.
1058 : : */
1059 : : static bool
1060 : 25 : synchronize_slots(WalReceiverConn *wrconn, List *remote_slot_list,
1061 : : bool *slot_persistence_pending)
1062 : : {
1063 : 25 : bool some_slot_updated = false;
1064 : :
1065 : : /* Drop local slots that no longer need to be synced. */
672 akapila@postgresql.o 1066 :CBC 25 : drop_local_obsolete_slots(remote_slot_list);
1067 : :
1068 : : /* Now sync the slots locally */
1069 [ + - + + : 95 : foreach_ptr(RemoteSlot, remote_slot, remote_slot_list)
+ + ]
1070 : : {
1071 : 45 : Oid remote_dbid = get_database_oid(remote_slot->database, false);
1072 : :
1073 : : /*
1074 : : * Use shared lock to prevent a conflict with
1075 : : * ReplicationSlotsDropDBSlots(), trying to drop the same slot during
1076 : : * a drop-database operation.
1077 : : */
1078 : 45 : LockSharedObject(DatabaseRelationId, remote_dbid, 0, AccessShareLock);
1079 : :
2 akapila@postgresql.o 1080 :GNC 45 : some_slot_updated |= synchronize_one_slot(remote_slot, remote_dbid,
1081 : : slot_persistence_pending);
1082 : :
672 akapila@postgresql.o 1083 :CBC 45 : UnlockSharedObject(DatabaseRelationId, remote_dbid, 0, AccessShareLock);
1084 : : }
1085 : :
664 1086 : 25 : return some_slot_updated;
1087 : : }
1088 : :
1089 : : /*
1090 : : * Checks the remote server info.
1091 : : *
1092 : : * We ensure that the 'primary_slot_name' exists on the remote server and the
1093 : : * remote server is not a standby node.
1094 : : */
1095 : : static void
672 1096 : 13 : validate_remote_info(WalReceiverConn *wrconn)
1097 : : {
1098 : : #define PRIMARY_INFO_OUTPUT_COL_COUNT 2
1099 : : WalRcvExecResult *res;
1100 : 13 : Oid slotRow[PRIMARY_INFO_OUTPUT_COL_COUNT] = {BOOLOID, BOOLOID};
1101 : : StringInfoData cmd;
1102 : : bool isnull;
1103 : : TupleTableSlot *tupslot;
1104 : : bool remote_in_recovery;
1105 : : bool primary_slot_valid;
664 1106 : 13 : bool started_tx = false;
1107 : :
672 1108 : 13 : initStringInfo(&cmd);
1109 : 13 : appendStringInfo(&cmd,
1110 : : "SELECT pg_is_in_recovery(), count(*) = 1"
1111 : : " FROM pg_catalog.pg_replication_slots"
1112 : : " WHERE slot_type='physical' AND slot_name=%s",
1113 : : quote_literal_cstr(PrimarySlotName));
1114 : :
1115 : : /* The syscache access in walrcv_exec() needs a transaction env. */
664 1116 [ + + ]: 13 : if (!IsTransactionState())
1117 : : {
1118 : 4 : StartTransactionCommand();
1119 : 4 : started_tx = true;
1120 : : }
1121 : :
672 1122 : 13 : res = walrcv_exec(wrconn, cmd.data, PRIMARY_INFO_OUTPUT_COL_COUNT, slotRow);
1123 : 13 : pfree(cmd.data);
1124 : :
1125 [ - + ]: 13 : if (res->status != WALRCV_OK_TUPLES)
672 akapila@postgresql.o 1126 [ # # ]:UBC 0 : ereport(ERROR,
1127 : : errmsg("could not fetch primary slot name \"%s\" info from the primary server: %s",
1128 : : PrimarySlotName, res->err),
1129 : : errhint("Check if \"primary_slot_name\" is configured correctly."));
1130 : :
672 akapila@postgresql.o 1131 :CBC 13 : tupslot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
1132 [ - + ]: 13 : if (!tuplestore_gettupleslot(res->tuplestore, true, false, tupslot))
672 akapila@postgresql.o 1133 [ # # ]:UBC 0 : elog(ERROR,
1134 : : "failed to fetch tuple for the primary server slot specified by \"primary_slot_name\"");
1135 : :
672 akapila@postgresql.o 1136 :CBC 13 : remote_in_recovery = DatumGetBool(slot_getattr(tupslot, 1, &isnull));
1137 [ - + ]: 13 : Assert(!isnull);
1138 : :
1139 : : /*
1140 : : * Slot sync is currently not supported on a cascading standby. This is
1141 : : * because if we allow it, the primary server needs to wait for all the
1142 : : * cascading standbys, otherwise, logical subscribers can still be ahead
1143 : : * of one of the cascading standbys which we plan to promote. Thus, to
1144 : : * avoid this additional complexity, we restrict it for the time being.
1145 : : */
1146 [ + + ]: 13 : if (remote_in_recovery)
1147 [ + - ]: 1 : ereport(ERROR,
1148 : : errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1149 : : errmsg("cannot synchronize replication slots from a standby server"));
1150 : :
1151 : 12 : primary_slot_valid = DatumGetBool(slot_getattr(tupslot, 2, &isnull));
1152 [ - + ]: 12 : Assert(!isnull);
1153 : :
1154 [ - + ]: 12 : if (!primary_slot_valid)
672 akapila@postgresql.o 1155 [ # # ]:UBC 0 : ereport(ERROR,
1156 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1157 : : /* translator: second %s is a GUC variable name */
1158 : : errmsg("replication slot \"%s\" specified by \"%s\" does not exist on primary server",
1159 : : PrimarySlotName, "primary_slot_name"));
1160 : :
672 akapila@postgresql.o 1161 :CBC 12 : ExecClearTuple(tupslot);
1162 : 12 : walrcv_clear_result(res);
1163 : :
664 1164 [ + + ]: 12 : if (started_tx)
1165 : 4 : CommitTransactionCommand();
672 1166 : 12 : }
1167 : :
1168 : : /*
1169 : : * Checks if dbname is specified in 'primary_conninfo'.
1170 : : *
1171 : : * Error out if not specified otherwise return it.
1172 : : */
1173 : : char *
664 1174 : 14 : CheckAndGetDbnameFromConninfo(void)
1175 : : {
1176 : : char *dbname;
1177 : :
1178 : : /*
1179 : : * The slot synchronization needs a database connection for walrcv_exec to
1180 : : * work.
1181 : : */
1182 : 14 : dbname = walrcv_get_dbname_from_conninfo(PrimaryConnInfo);
1183 [ + + ]: 14 : if (dbname == NULL)
1184 [ + - ]: 1 : ereport(ERROR,
1185 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1186 : :
1187 : : /*
1188 : : * translator: first %s is a connection option; second %s is a GUC
1189 : : * variable name
1190 : : */
1191 : : errmsg("replication slot synchronization requires \"%s\" to be specified in \"%s\"",
1192 : : "dbname", "primary_conninfo"));
1193 : 13 : return dbname;
1194 : : }
1195 : :
1196 : : /*
1197 : : * Return true if all necessary GUCs for slot synchronization are set
1198 : : * appropriately, otherwise, return false.
1199 : : */
1200 : : bool
1201 : 17 : ValidateSlotSyncParams(int elevel)
1202 : : {
1203 : : /*
1204 : : * Logical slot sync/creation requires wal_level >= logical.
1205 : : */
1206 [ - + ]: 17 : if (wal_level < WAL_LEVEL_LOGICAL)
1207 : : {
135 fujii@postgresql.org 1208 [ # # ]:UBC 0 : ereport(elevel,
1209 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1210 : : errmsg("replication slot synchronization requires \"wal_level\" >= \"logical\""));
1211 : 0 : return false;
1212 : : }
1213 : :
1214 : : /*
1215 : : * A physical replication slot(primary_slot_name) is required on the
1216 : : * primary to ensure that the rows needed by the standby are not removed
1217 : : * after restarting, so that the synchronized slot on the standby will not
1218 : : * be invalidated.
1219 : : */
672 akapila@postgresql.o 1220 [ + - - + ]:CBC 17 : if (PrimarySlotName == NULL || *PrimarySlotName == '\0')
1221 : : {
664 akapila@postgresql.o 1222 [ # # ]:UBC 0 : ereport(elevel,
1223 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1224 : : /* translator: %s is a GUC variable name */
1225 : : errmsg("replication slot synchronization requires \"%s\" to be set", "primary_slot_name"));
1226 : 0 : return false;
1227 : : }
1228 : :
1229 : : /*
1230 : : * hot_standby_feedback must be enabled to cooperate with the physical
1231 : : * replication slot, which allows informing the primary about the xmin and
1232 : : * catalog_xmin values on the standby.
1233 : : */
672 akapila@postgresql.o 1234 [ + + ]:CBC 17 : if (!hot_standby_feedback)
1235 : : {
664 1236 [ + - ]: 1 : ereport(elevel,
1237 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1238 : : /* translator: %s is a GUC variable name */
1239 : : errmsg("replication slot synchronization requires \"%s\" to be enabled",
1240 : : "hot_standby_feedback"));
1241 : 1 : return false;
1242 : : }
1243 : :
1244 : : /*
1245 : : * The primary_conninfo is required to make connection to primary for
1246 : : * getting slots information.
1247 : : */
672 1248 [ + - - + ]: 16 : if (PrimaryConnInfo == NULL || *PrimaryConnInfo == '\0')
1249 : : {
664 akapila@postgresql.o 1250 [ # # ]:UBC 0 : ereport(elevel,
1251 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1252 : : /* translator: %s is a GUC variable name */
1253 : : errmsg("replication slot synchronization requires \"%s\" to be set",
1254 : : "primary_conninfo"));
1255 : 0 : return false;
1256 : : }
1257 : :
664 akapila@postgresql.o 1258 :CBC 16 : return true;
1259 : : }
1260 : :
1261 : : /*
1262 : : * Re-read the config file for slot synchronization.
1263 : : *
1264 : : * Exit or throw error if relevant GUCs have changed depending on whether
1265 : : * called from slot sync worker or from the SQL function pg_sync_replication_slots()
1266 : : */
1267 : : static void
1268 : 1 : slotsync_reread_config(void)
1269 : : {
1270 : 1 : char *old_primary_conninfo = pstrdup(PrimaryConnInfo);
1271 : 1 : char *old_primary_slotname = pstrdup(PrimarySlotName);
1272 : 1 : bool old_sync_replication_slots = sync_replication_slots;
1273 : 1 : bool old_hot_standby_feedback = hot_standby_feedback;
1274 : : bool conninfo_changed;
1275 : : bool primary_slotname_changed;
6 akapila@postgresql.o 1276 :GNC 1 : bool is_slotsync_worker = AmLogicalSlotSyncWorkerProcess();
1277 : 1 : bool parameter_changed = false;
1278 : :
1279 [ + - ]: 1 : if (is_slotsync_worker)
1280 [ - + ]: 1 : Assert(sync_replication_slots);
1281 : :
664 akapila@postgresql.o 1282 :CBC 1 : ConfigReloadPending = false;
1283 : 1 : ProcessConfigFile(PGC_SIGHUP);
1284 : :
1285 : 1 : conninfo_changed = strcmp(old_primary_conninfo, PrimaryConnInfo) != 0;
1286 : 1 : primary_slotname_changed = strcmp(old_primary_slotname, PrimarySlotName) != 0;
1287 : 1 : pfree(old_primary_conninfo);
1288 : 1 : pfree(old_primary_slotname);
1289 : :
1290 [ - + ]: 1 : if (old_sync_replication_slots != sync_replication_slots)
1291 : : {
6 akapila@postgresql.o 1292 [ # # ]:UNC 0 : if (is_slotsync_worker)
1293 : : {
1294 [ # # ]: 0 : ereport(LOG,
1295 : : /* translator: %s is a GUC variable name */
1296 : : errmsg("replication slot synchronization worker will stop because \"%s\" is disabled",
1297 : : "sync_replication_slots"));
1298 : :
1299 : 0 : proc_exit(0);
1300 : : }
1301 : :
1302 : 0 : parameter_changed = true;
1303 : : }
1304 : : else
1305 : : {
6 akapila@postgresql.o 1306 [ + - + - ]:GNC 1 : if (conninfo_changed ||
1307 : 1 : primary_slotname_changed ||
1308 [ + - ]: 1 : (old_hot_standby_feedback != hot_standby_feedback))
1309 : : {
1310 : :
1311 [ + - ]: 1 : if (is_slotsync_worker)
1312 : : {
1313 [ + - ]: 1 : ereport(LOG,
1314 : : errmsg("replication slot synchronization worker will restart because of a parameter change"));
1315 : :
1316 : : /*
1317 : : * Reset the last-start time for this worker so that the
1318 : : * postmaster can restart it without waiting for
1319 : : * SLOTSYNC_RESTART_INTERVAL_SEC.
1320 : : */
1321 : 1 : SlotSyncCtx->last_start_time = 0;
1322 : :
1323 : 1 : proc_exit(0);
1324 : : }
1325 : :
6 akapila@postgresql.o 1326 :UNC 0 : parameter_changed = true;
1327 : : }
1328 : : }
1329 : :
1330 : : /*
1331 : : * If we have reached here with a parameter change, we must be running in
1332 : : * SQL function, emit error in such a case.
1333 : : */
1334 [ # # ]: 0 : if (parameter_changed)
1335 : : {
1336 [ # # ]: 0 : Assert(!is_slotsync_worker);
1337 [ # # ]: 0 : ereport(ERROR,
1338 : : errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1339 : : errmsg("replication slot synchronization will stop because of a parameter change"));
1340 : : }
1341 : :
664 akapila@postgresql.o 1342 :UBC 0 : }
1343 : :
1344 : : /*
1345 : : * Interrupt handler for process performing slot synchronization.
1346 : : */
1347 : : static void
110 nathan@postgresql.or 1348 :GNC 38 : ProcessSlotSyncInterrupts(void)
1349 : : {
664 akapila@postgresql.o 1350 [ + + ]:CBC 38 : CHECK_FOR_INTERRUPTS();
1351 : :
8 1352 [ + + ]: 36 : if (SlotSyncCtx->stopSignaled)
1353 : : {
6 akapila@postgresql.o 1354 [ + - ]:GNC 1 : if (AmLogicalSlotSyncWorkerProcess())
1355 : : {
1356 [ + - ]: 1 : ereport(LOG,
1357 : : errmsg("replication slot synchronization worker will stop because promotion is triggered"));
1358 : :
1359 : 1 : proc_exit(0);
1360 : : }
1361 : : else
1362 : : {
1363 : : /*
1364 : : * For the backend executing SQL function
1365 : : * pg_sync_replication_slots().
1366 : : */
6 akapila@postgresql.o 1367 [ # # ]:UNC 0 : ereport(ERROR,
1368 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1369 : : errmsg("replication slot synchronization will stop because promotion is triggered"));
1370 : : }
1371 : : }
1372 : :
664 akapila@postgresql.o 1373 [ + + ]:CBC 35 : if (ConfigReloadPending)
1374 : 1 : slotsync_reread_config();
1375 : 34 : }
1376 : :
1377 : : /*
1378 : : * Connection cleanup function for slotsync worker.
1379 : : *
1380 : : * Called on slotsync worker exit.
1381 : : */
1382 : : static void
601 1383 : 4 : slotsync_worker_disconnect(int code, Datum arg)
1384 : : {
1385 : 4 : WalReceiverConn *wrconn = (WalReceiverConn *) DatumGetPointer(arg);
1386 : :
1387 : 4 : walrcv_disconnect(wrconn);
1388 : 4 : }
1389 : :
1390 : : /*
1391 : : * Cleanup function for slotsync worker.
1392 : : *
1393 : : * Called on slotsync worker exit.
1394 : : */
1395 : : static void
664 1396 : 4 : slotsync_worker_onexit(int code, Datum arg)
1397 : : {
1398 : : /*
1399 : : * We need to do slots cleanup here just like WalSndErrorCleanup() does.
1400 : : *
1401 : : * The startup process during promotion invokes ShutDownSlotSync() which
1402 : : * waits for slot sync to finish and it does that by checking the
1403 : : * 'syncing' flag. Thus the slot sync worker must be done with slots'
1404 : : * release and cleanup to avoid any dangling temporary slots or active
1405 : : * slots before it marks itself as finished syncing.
1406 : : */
1407 : :
1408 : : /* Make sure active replication slots are released */
601 1409 [ - + ]: 4 : if (MyReplicationSlot != NULL)
601 akapila@postgresql.o 1410 :UBC 0 : ReplicationSlotRelease();
1411 : :
1412 : : /* Also cleanup the temporary slots. */
601 akapila@postgresql.o 1413 :CBC 4 : ReplicationSlotCleanup(false);
1414 : :
664 1415 [ - + ]: 4 : SpinLockAcquire(&SlotSyncCtx->mutex);
1416 : :
1417 : 4 : SlotSyncCtx->pid = InvalidPid;
1418 : :
1419 : : /*
1420 : : * If syncing_slots is true, it indicates that the process errored out
1421 : : * without resetting the flag. So, we need to clean up shared memory and
1422 : : * reset the flag here.
1423 : : */
601 1424 [ + - ]: 4 : if (syncing_slots)
1425 : : {
1426 : 4 : SlotSyncCtx->syncing = false;
1427 : 4 : syncing_slots = false;
1428 : : }
1429 : :
664 1430 : 4 : SpinLockRelease(&SlotSyncCtx->mutex);
1431 : 4 : }
1432 : :
1433 : : /*
1434 : : * Sleep for long enough that we believe it's likely that the slots on primary
1435 : : * get updated.
1436 : : *
1437 : : * If there is no slot activity the wait time between sync-cycles will double
1438 : : * (to a maximum of 30s). If there is some slot activity the wait time between
1439 : : * sync-cycles is reset to the minimum (200ms).
1440 : : */
1441 : : static void
1442 : 17 : wait_for_slot_activity(bool some_slot_updated)
1443 : : {
1444 : : int rc;
1445 : :
1446 [ + + ]: 17 : if (!some_slot_updated)
1447 : : {
1448 : : /*
1449 : : * No slots were updated, so double the sleep time, but not beyond the
1450 : : * maximum allowable value.
1451 : : */
657 1452 : 10 : sleep_ms = Min(sleep_ms * 2, MAX_SLOTSYNC_WORKER_NAPTIME_MS);
1453 : : }
1454 : : else
1455 : : {
1456 : : /*
1457 : : * Some slots were updated since the last sleep, so reset the sleep
1458 : : * time.
1459 : : */
1460 : 7 : sleep_ms = MIN_SLOTSYNC_WORKER_NAPTIME_MS;
1461 : : }
1462 : :
664 1463 : 17 : rc = WaitLatch(MyLatch,
1464 : : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
1465 : : sleep_ms,
1466 : : WAIT_EVENT_REPLICATION_SLOTSYNC_MAIN);
1467 : :
1468 [ + + ]: 17 : if (rc & WL_LATCH_SET)
1469 : 4 : ResetLatch(MyLatch);
1470 : 17 : }
1471 : :
1472 : : /*
1473 : : * Emit an error if a concurrent sync call is in progress.
1474 : : * Otherwise, advertise that a sync is in progress.
1475 : : */
1476 : : static void
6 akapila@postgresql.o 1477 :GNC 13 : check_and_set_sync_info(pid_t sync_process_pid)
1478 : : {
601 akapila@postgresql.o 1479 [ - + ]:CBC 13 : SpinLockAcquire(&SlotSyncCtx->mutex);
1480 : :
1481 [ - + ]: 13 : if (SlotSyncCtx->syncing)
1482 : : {
601 akapila@postgresql.o 1483 :UBC 0 : SpinLockRelease(&SlotSyncCtx->mutex);
1484 [ # # ]: 0 : ereport(ERROR,
1485 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1486 : : errmsg("cannot synchronize replication slots concurrently"));
1487 : : }
1488 : :
1489 : : /* The pid must not be already assigned in SlotSyncCtx */
6 akapila@postgresql.o 1490 [ - + ]:GNC 13 : Assert(SlotSyncCtx->pid == InvalidPid);
1491 : :
601 akapila@postgresql.o 1492 :CBC 13 : SlotSyncCtx->syncing = true;
1493 : :
1494 : : /*
1495 : : * Advertise the required PID so that the startup process can kill the
1496 : : * slot sync process on promotion.
1497 : : */
6 akapila@postgresql.o 1498 :GNC 13 : SlotSyncCtx->pid = sync_process_pid;
1499 : :
601 akapila@postgresql.o 1500 :CBC 13 : SpinLockRelease(&SlotSyncCtx->mutex);
1501 : :
1502 : 13 : syncing_slots = true;
1503 : 13 : }
1504 : :
1505 : : /*
1506 : : * Reset syncing flag.
1507 : : */
1508 : : static void
14 nathan@postgresql.or 1509 :GNC 9 : reset_syncing_flag(void)
1510 : : {
601 akapila@postgresql.o 1511 [ - + ]:CBC 9 : SpinLockAcquire(&SlotSyncCtx->mutex);
1512 : 9 : SlotSyncCtx->syncing = false;
6 akapila@postgresql.o 1513 :GNC 9 : SlotSyncCtx->pid = InvalidPid;
601 akapila@postgresql.o 1514 :CBC 9 : SpinLockRelease(&SlotSyncCtx->mutex);
1515 : :
1516 : 9 : syncing_slots = false;
97 peter@eisentraut.org 1517 : 9 : }
1518 : :
1519 : : /*
1520 : : * The main loop of our worker process.
1521 : : *
1522 : : * It connects to the primary server, fetches logical failover slots
1523 : : * information periodically in order to create and sync the slots.
1524 : : *
1525 : : * Note: If any changes are made here, check if the corresponding SQL
1526 : : * function logic in SyncReplicationSlots() also needs to be changed.
1527 : : */
1528 : : void
299 1529 : 4 : ReplSlotSyncWorkerMain(const void *startup_data, size_t startup_data_len)
1530 : : {
664 akapila@postgresql.o 1531 : 4 : WalReceiverConn *wrconn = NULL;
1532 : : char *dbname;
1533 : : char *err;
1534 : : sigjmp_buf local_sigjmp_buf;
1535 : : StringInfoData app_name;
1536 : :
639 heikki.linnakangas@i 1537 [ - + ]: 4 : Assert(startup_data_len == 0);
1538 : :
664 akapila@postgresql.o 1539 : 4 : MyBackendType = B_SLOTSYNC_WORKER;
1540 : :
1541 : 4 : init_ps_display(NULL);
1542 : :
533 heikki.linnakangas@i 1543 [ - + ]: 4 : Assert(GetProcessingMode() == InitProcessing);
1544 : :
1545 : : /*
1546 : : * Create a per-backend PGPROC struct in shared memory. We must do this
1547 : : * before we access any shared memory.
1548 : : */
664 akapila@postgresql.o 1549 : 4 : InitProcess();
1550 : :
1551 : : /*
1552 : : * Early initialization.
1553 : : */
1554 : 4 : BaseInit();
1555 : :
1556 [ - + ]: 4 : Assert(SlotSyncCtx != NULL);
1557 : :
1558 : : /*
1559 : : * If an exception is encountered, processing resumes here.
1560 : : *
1561 : : * We just need to clean up, report the error, and go away.
1562 : : *
1563 : : * If we do not have this handling here, then since this worker process
1564 : : * operates at the bottom of the exception stack, ERRORs turn into FATALs.
1565 : : * Therefore, we create our own exception handler to catch ERRORs.
1566 : : */
1567 [ - + ]: 4 : if (sigsetjmp(local_sigjmp_buf, 1) != 0)
1568 : : {
1569 : : /* since not using PG_TRY, must reset error stack by hand */
664 akapila@postgresql.o 1570 :UBC 0 : error_context_stack = NULL;
1571 : :
1572 : : /* Prevents interrupts while cleaning up */
1573 : 0 : HOLD_INTERRUPTS();
1574 : :
1575 : : /* Report the error to the server log */
1576 : 0 : EmitErrorReport();
1577 : :
1578 : : /*
1579 : : * We can now go away. Note that because we called InitProcess, a
1580 : : * callback was registered to do ProcKill, which will clean up
1581 : : * necessary state.
1582 : : */
1583 : 0 : proc_exit(0);
1584 : : }
1585 : :
1586 : : /* We can now handle ereport(ERROR) */
664 akapila@postgresql.o 1587 :CBC 4 : PG_exception_stack = &local_sigjmp_buf;
1588 : :
1589 : : /* Setup signal handling */
601 1590 : 4 : pqsignal(SIGHUP, SignalHandlerForConfigReload);
8 1591 : 4 : pqsignal(SIGINT, StatementCancelHandler);
601 1592 : 4 : pqsignal(SIGTERM, die);
1593 : 4 : pqsignal(SIGFPE, FloatExceptionHandler);
1594 : 4 : pqsignal(SIGUSR1, procsignal_sigusr1_handler);
1595 : 4 : pqsignal(SIGUSR2, SIG_IGN);
1596 : 4 : pqsignal(SIGPIPE, SIG_IGN);
1597 : 4 : pqsignal(SIGCHLD, SIG_DFL);
1598 : :
1599 : 4 : check_and_set_sync_info(MyProcPid);
1600 : :
1601 [ + - ]: 4 : ereport(LOG, errmsg("slot sync worker started"));
1602 : :
1603 : : /* Register it as soon as SlotSyncCtx->pid is initialized. */
1604 : 4 : before_shmem_exit(slotsync_worker_onexit, (Datum) 0);
1605 : :
1606 : : /*
1607 : : * Establishes SIGALRM handler and initialize timeout module. It is needed
1608 : : * by InitPostgres to register different timeouts.
1609 : : */
1610 : 4 : InitializeTimeouts();
1611 : :
1612 : : /* Load the libpq-specific functions */
1613 : 4 : load_file("libpqwalreceiver", false);
1614 : :
1615 : : /*
1616 : : * Unblock signals (they were blocked when the postmaster forked us)
1617 : : */
664 1618 : 4 : sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
1619 : :
1620 : : /*
1621 : : * Set always-secure search path, so malicious users can't redirect user
1622 : : * code (e.g. operators).
1623 : : *
1624 : : * It's not strictly necessary since we won't be scanning or writing to
1625 : : * any user table locally, but it's good to retain it here for added
1626 : : * precaution.
1627 : : */
657 1628 : 4 : SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
1629 : :
664 1630 : 4 : dbname = CheckAndGetDbnameFromConninfo();
1631 : :
1632 : : /*
1633 : : * Connect to the database specified by the user in primary_conninfo. We
1634 : : * need a database connection for walrcv_exec to work which we use to
1635 : : * fetch slot information from the remote node. See comments atop
1636 : : * libpqrcv_exec.
1637 : : *
1638 : : * We do not specify a specific user here since the slot sync worker will
1639 : : * operate as a superuser. This is safe because the slot sync worker does
1640 : : * not interact with user tables, eliminating the risk of executing
1641 : : * arbitrary code within triggers.
1642 : : */
1643 : 4 : InitPostgres(dbname, InvalidOid, NULL, InvalidOid, 0, NULL);
1644 : :
1645 : 4 : SetProcessingMode(NormalProcessing);
1646 : :
1647 : 4 : initStringInfo(&app_name);
1648 [ + - ]: 4 : if (cluster_name[0])
1649 : 4 : appendStringInfo(&app_name, "%s_%s", cluster_name, "slotsync worker");
1650 : : else
616 drowley@postgresql.o 1651 :UBC 0 : appendStringInfoString(&app_name, "slotsync worker");
1652 : :
1653 : : /*
1654 : : * Establish the connection to the primary server for slot
1655 : : * synchronization.
1656 : : */
664 akapila@postgresql.o 1657 :CBC 4 : wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
1658 : : app_name.data, &err);
1659 : :
1660 [ - + ]: 4 : if (!wrconn)
664 akapila@postgresql.o 1661 [ # # ]:UBC 0 : ereport(ERROR,
1662 : : errcode(ERRCODE_CONNECTION_FAILURE),
1663 : : errmsg("synchronization worker \"%s\" could not connect to the primary server: %s",
1664 : : app_name.data, err));
1665 : :
105 akapila@postgresql.o 1666 :CBC 4 : pfree(app_name.data);
1667 : :
1668 : : /*
1669 : : * Register the disconnection callback.
1670 : : *
1671 : : * XXX: This can be combined with previous cleanup registration of
1672 : : * slotsync_worker_onexit() but that will need the connection to be made
1673 : : * global and we want to avoid introducing global for this purpose.
1674 : : */
601 1675 : 4 : before_shmem_exit(slotsync_worker_disconnect, PointerGetDatum(wrconn));
1676 : :
1677 : : /*
1678 : : * Using the specified primary server connection, check that we are not a
1679 : : * cascading standby and slot configured in 'primary_slot_name' exists on
1680 : : * the primary server.
1681 : : */
664 1682 : 4 : validate_remote_info(wrconn);
1683 : :
1684 : : /* Main loop to synchronize slots */
1685 : : for (;;)
1686 : 15 : {
1687 : 19 : bool some_slot_updated = false;
2 akapila@postgresql.o 1688 :GNC 19 : bool started_tx = false;
1689 : : List *remote_slots;
1690 : :
110 nathan@postgresql.or 1691 : 19 : ProcessSlotSyncInterrupts();
1692 : :
1693 : : /*
1694 : : * The syscache access in fetch_remote_slots() needs a transaction
1695 : : * env.
1696 : : */
2 akapila@postgresql.o 1697 [ + - ]: 15 : if (!IsTransactionState())
1698 : : {
1699 : 15 : StartTransactionCommand();
1700 : 15 : started_tx = true;
1701 : : }
1702 : :
1703 : 15 : remote_slots = fetch_remote_slots(wrconn, NIL);
1704 : 15 : some_slot_updated = synchronize_slots(wrconn, remote_slots, NULL);
1705 : 15 : list_free_deep(remote_slots);
1706 : :
1707 [ + - ]: 15 : if (started_tx)
1708 : 15 : CommitTransactionCommand();
1709 : :
664 akapila@postgresql.o 1710 :CBC 15 : wait_for_slot_activity(some_slot_updated);
1711 : : }
1712 : :
1713 : : /*
1714 : : * The slot sync worker can't get here because it will only stop when it
1715 : : * receives a stop request from the startup process, or when there is an
1716 : : * error.
1717 : : */
1718 : : Assert(false);
1719 : : }
1720 : :
1721 : : /*
1722 : : * Update the inactive_since property for synced slots.
1723 : : *
1724 : : * Note that this function is currently called when we shutdown the slot
1725 : : * sync machinery.
1726 : : */
1727 : : static void
621 1728 : 870 : update_synced_slots_inactive_since(void)
1729 : : {
1730 : 870 : TimestampTz now = 0;
1731 : :
1732 : : /*
1733 : : * We need to update inactive_since only when we are promoting standby to
1734 : : * correctly interpret the inactive_since if the standby gets promoted
1735 : : * without a restart. We don't want the slots to appear inactive for a
1736 : : * long time after promotion if they haven't been synchronized recently.
1737 : : * Whoever acquires the slot, i.e., makes the slot active, will reset it.
1738 : : */
1739 [ + + ]: 870 : if (!StandbyMode)
1740 : 824 : return;
1741 : :
1742 : : /* The slot sync worker or the SQL function mustn't be running by now */
601 1743 [ + - - + ]: 46 : Assert((SlotSyncCtx->pid == InvalidPid) && !SlotSyncCtx->syncing);
1744 : :
621 1745 : 46 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1746 : :
1747 [ + + ]: 499 : for (int i = 0; i < max_replication_slots; i++)
1748 : : {
1749 : 453 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
1750 : :
1751 : : /* Check if it is a synchronized slot */
1752 [ + + + + ]: 453 : if (s->in_use && s->data.synced)
1753 : : {
1754 [ - + ]: 3 : Assert(SlotIsLogical(s));
1755 : :
1756 : : /* The slot must not be acquired by any process */
601 1757 [ - + ]: 3 : Assert(s->active_pid == 0);
1758 : :
1759 : : /* Use the same inactive_since time for all the slots. */
621 1760 [ + + ]: 3 : if (now == 0)
1761 : 2 : now = GetCurrentTimestamp();
1762 : :
315 1763 : 3 : ReplicationSlotSetInactiveSince(s, now, true);
1764 : : }
1765 : : }
1766 : :
621 1767 : 46 : LWLockRelease(ReplicationSlotControlLock);
1768 : : }
1769 : :
1770 : : /*
1771 : : * Shut down slot synchronization.
1772 : : *
1773 : : * This function sets stopSignaled=true and wakes up the slot sync process
1774 : : * (either worker or backend running the SQL function pg_sync_replication_slots())
1775 : : * so that worker can exit or the SQL function pg_sync_replication_slots() can
1776 : : * finish. It also waits till the slot sync worker has exited or
1777 : : * pg_sync_replication_slots() has finished.
1778 : : */
1779 : : void
664 1780 : 870 : ShutDownSlotSync(void)
1781 : : {
1782 : : pid_t sync_process_pid;
1783 : :
1784 [ - + ]: 870 : SpinLockAcquire(&SlotSyncCtx->mutex);
1785 : :
1786 : 870 : SlotSyncCtx->stopSignaled = true;
1787 : :
1788 : : /*
1789 : : * Return if neither the slot sync worker is running nor the function
1790 : : * pg_sync_replication_slots() is executing.
1791 : : */
601 1792 [ + + ]: 870 : if (!SlotSyncCtx->syncing)
1793 : : {
664 1794 : 869 : SpinLockRelease(&SlotSyncCtx->mutex);
621 1795 : 869 : update_synced_slots_inactive_since();
664 1796 : 869 : return;
1797 : : }
1798 : :
6 akapila@postgresql.o 1799 :GNC 1 : sync_process_pid = SlotSyncCtx->pid;
1800 : :
664 akapila@postgresql.o 1801 :CBC 1 : SpinLockRelease(&SlotSyncCtx->mutex);
1802 : :
1803 : : /*
1804 : : * Signal process doing slotsync, if any. The process will stop upon
1805 : : * detecting that the stopSignaled flag is set to true.
1806 : : */
6 akapila@postgresql.o 1807 [ + - ]:GNC 1 : if (sync_process_pid != InvalidPid)
1808 : 1 : kill(sync_process_pid, SIGUSR1);
1809 : :
1810 : : /* Wait for slot sync to end */
1811 : : for (;;)
664 akapila@postgresql.o 1812 :UBC 0 : {
1813 : : int rc;
1814 : :
1815 : : /* Wait a bit, we don't expect to have to wait long */
664 akapila@postgresql.o 1816 :CBC 1 : rc = WaitLatch(MyLatch,
1817 : : WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
1818 : : 10L, WAIT_EVENT_REPLICATION_SLOTSYNC_SHUTDOWN);
1819 : :
1820 [ - + ]: 1 : if (rc & WL_LATCH_SET)
1821 : : {
664 akapila@postgresql.o 1822 :UBC 0 : ResetLatch(MyLatch);
1823 [ # # ]: 0 : CHECK_FOR_INTERRUPTS();
1824 : : }
1825 : :
664 akapila@postgresql.o 1826 [ - + ]:CBC 1 : SpinLockAcquire(&SlotSyncCtx->mutex);
1827 : :
1828 : : /* Ensure that no process is syncing the slots. */
601 1829 [ + - ]: 1 : if (!SlotSyncCtx->syncing)
664 1830 : 1 : break;
1831 : :
664 akapila@postgresql.o 1832 :UBC 0 : SpinLockRelease(&SlotSyncCtx->mutex);
1833 : : }
1834 : :
664 akapila@postgresql.o 1835 :CBC 1 : SpinLockRelease(&SlotSyncCtx->mutex);
1836 : :
621 1837 : 1 : update_synced_slots_inactive_since();
1838 : : }
1839 : :
1840 : : /*
1841 : : * SlotSyncWorkerCanRestart
1842 : : *
1843 : : * Return true, indicating worker is allowed to restart, if enough time has
1844 : : * passed since it was last launched to reach SLOTSYNC_RESTART_INTERVAL_SEC.
1845 : : * Otherwise return false.
1846 : : *
1847 : : * This is a safety valve to protect against continuous respawn attempts if the
1848 : : * worker is dying immediately at launch. Note that since we will retry to
1849 : : * launch the worker from the postmaster main loop, we will get another
1850 : : * chance later.
1851 : : */
1852 : : bool
664 1853 : 6 : SlotSyncWorkerCanRestart(void)
1854 : : {
1855 : 6 : time_t curtime = time(NULL);
1856 : :
1857 : : /*
1858 : : * If first time through, or time somehow went backwards, always update
1859 : : * last_start_time to match the current clock and allow worker start.
1860 : : * Otherwise allow it only once enough time has elapsed.
1861 : : */
56 tgl@sss.pgh.pa.us 1862 [ + + ]:GNC 6 : if (SlotSyncCtx->last_start_time == 0 ||
1863 [ + - ]: 2 : curtime < SlotSyncCtx->last_start_time ||
1864 [ - + ]: 2 : curtime - SlotSyncCtx->last_start_time >= SLOTSYNC_RESTART_INTERVAL_SEC)
1865 : : {
1866 : 4 : SlotSyncCtx->last_start_time = curtime;
1867 : 4 : return true;
1868 : : }
1869 : 2 : return false;
1870 : : }
1871 : :
1872 : : /*
1873 : : * Is current process syncing replication slots?
1874 : : *
1875 : : * Could be either backend executing SQL function or slot sync worker.
1876 : : */
1877 : : bool
672 akapila@postgresql.o 1878 :CBC 71 : IsSyncingReplicationSlots(void)
1879 : : {
1880 : 71 : return syncing_slots;
1881 : : }
1882 : :
1883 : : /*
1884 : : * Amount of shared memory required for slot synchronization.
1885 : : */
1886 : : Size
1887 : 3055 : SlotSyncShmemSize(void)
1888 : : {
1889 : 3055 : return sizeof(SlotSyncCtxStruct);
1890 : : }
1891 : :
1892 : : /*
1893 : : * Allocate and initialize the shared memory of slot synchronization.
1894 : : */
1895 : : void
1896 : 1069 : SlotSyncShmemInit(void)
1897 : : {
664 1898 : 1069 : Size size = SlotSyncShmemSize();
1899 : : bool found;
1900 : :
672 1901 : 1069 : SlotSyncCtx = (SlotSyncCtxStruct *)
664 1902 : 1069 : ShmemInitStruct("Slot Sync Data", size, &found);
1903 : :
672 1904 [ + - ]: 1069 : if (!found)
1905 : : {
664 1906 : 1069 : memset(SlotSyncCtx, 0, size);
1907 : 1069 : SlotSyncCtx->pid = InvalidPid;
672 1908 : 1069 : SpinLockInit(&SlotSyncCtx->mutex);
1909 : : }
1910 : 1069 : }
1911 : :
1912 : : /*
1913 : : * Error cleanup callback for slot sync SQL function.
1914 : : */
1915 : : static void
1916 : 1 : slotsync_failure_callback(int code, Datum arg)
1917 : : {
1918 : 1 : WalReceiverConn *wrconn = (WalReceiverConn *) DatumGetPointer(arg);
1919 : :
1920 : : /*
1921 : : * We need to do slots cleanup here just like WalSndErrorCleanup() does.
1922 : : *
1923 : : * The startup process during promotion invokes ShutDownSlotSync() which
1924 : : * waits for slot sync to finish and it does that by checking the
1925 : : * 'syncing' flag. Thus the SQL function must be done with slots' release
1926 : : * and cleanup to avoid any dangling temporary slots or active slots
1927 : : * before it marks itself as finished syncing.
1928 : : */
1929 : :
1930 : : /* Make sure active replication slots are released */
601 1931 [ - + ]: 1 : if (MyReplicationSlot != NULL)
601 akapila@postgresql.o 1932 :UBC 0 : ReplicationSlotRelease();
1933 : :
1934 : : /* Also cleanup the synced temporary slots. */
601 akapila@postgresql.o 1935 :CBC 1 : ReplicationSlotCleanup(true);
1936 : :
1937 : : /*
1938 : : * The set syncing_slots indicates that the process errored out without
1939 : : * resetting the flag. So, we need to clean up shared memory and reset the
1940 : : * flag here.
1941 : : */
1942 [ + - ]: 1 : if (syncing_slots)
1943 : 1 : reset_syncing_flag();
1944 : :
672 1945 : 1 : walrcv_disconnect(wrconn);
1946 : 1 : }
1947 : :
1948 : : /*
1949 : : * Helper function to extract slot names from a list of remote slots
1950 : : */
1951 : : static List *
2 akapila@postgresql.o 1952 :GNC 1 : extract_slot_names(List *remote_slots)
1953 : : {
1954 : 1 : List *slot_names = NIL;
1955 : :
1956 [ + - + + : 3 : foreach_ptr(RemoteSlot, remote_slot, remote_slots)
+ + ]
1957 : : {
1958 : : char *slot_name;
1959 : :
1960 : 1 : slot_name = pstrdup(remote_slot->name);
1961 : 1 : slot_names = lappend(slot_names, slot_name);
1962 : : }
1963 : :
1964 : 1 : return slot_names;
1965 : : }
1966 : :
1967 : : /*
1968 : : * Synchronize the failover enabled replication slots using the specified
1969 : : * primary server connection.
1970 : : *
1971 : : * Repeatedly fetches and updates replication slot information from the
1972 : : * primary until all slots are at least "sync ready".
1973 : : *
1974 : : * Exits early if promotion is triggered or certain critical
1975 : : * configuration parameters have changed.
1976 : : */
1977 : : void
672 akapila@postgresql.o 1978 :CBC 9 : SyncReplicationSlots(WalReceiverConn *wrconn)
1979 : : {
1980 [ + + ]: 9 : PG_ENSURE_ERROR_CLEANUP(slotsync_failure_callback, PointerGetDatum(wrconn));
1981 : : {
2 akapila@postgresql.o 1982 :GNC 9 : List *remote_slots = NIL;
1983 : 9 : List *slot_names = NIL; /* List of slot names to track */
1984 : :
6 1985 : 9 : check_and_set_sync_info(MyProcPid);
1986 : :
1987 : : /* Check for interrupts and config changes */
1988 : 9 : ProcessSlotSyncInterrupts();
1989 : :
672 akapila@postgresql.o 1990 :CBC 9 : validate_remote_info(wrconn);
1991 : :
1992 : : /* Retry until all the slots are sync-ready */
1993 : : for (;;)
2 akapila@postgresql.o 1994 :GNC 2 : {
1995 : 10 : bool slot_persistence_pending = false;
1996 : 10 : bool some_slot_updated = false;
1997 : :
1998 : : /* Check for interrupts and config changes */
1999 : 10 : ProcessSlotSyncInterrupts();
2000 : :
2001 : : /* We must be in a valid transaction state */
2002 [ - + ]: 10 : Assert(IsTransactionState());
2003 : :
2004 : : /*
2005 : : * Fetch remote slot info for the given slot_names. If slot_names
2006 : : * is NIL, fetch all failover-enabled slots. Note that we reuse
2007 : : * slot_names from the first iteration; re-fetching all failover
2008 : : * slots each time could cause an endless loop. Instead of
2009 : : * reprocessing only the pending slots in each iteration, it's
2010 : : * better to process all the slots received in the first
2011 : : * iteration. This ensures that by the time we're done, all slots
2012 : : * reflect the latest values.
2013 : : */
2014 : 10 : remote_slots = fetch_remote_slots(wrconn, slot_names);
2015 : :
2016 : : /* Attempt to synchronize slots */
2017 : 10 : some_slot_updated = synchronize_slots(wrconn, remote_slots,
2018 : : &slot_persistence_pending);
2019 : :
2020 : : /*
2021 : : * If slot_persistence_pending is true, extract slot names for
2022 : : * future iterations (only needed if we haven't done it yet)
2023 : : */
2024 [ + + + + ]: 10 : if (slot_names == NIL && slot_persistence_pending)
2025 : 1 : slot_names = extract_slot_names(remote_slots);
2026 : :
2027 : : /* Free the current remote_slots list */
2028 : 10 : list_free_deep(remote_slots);
2029 : :
2030 : : /* Done if all slots are persisted i.e are sync-ready */
2031 [ + + ]: 10 : if (!slot_persistence_pending)
2032 : 8 : break;
2033 : :
2034 : : /* wait before retrying again */
2035 : 2 : wait_for_slot_activity(some_slot_updated);
2036 : : }
2037 : :
2038 [ + + ]: 8 : if (slot_names)
2039 : 1 : list_free_deep(slot_names);
2040 : :
2041 : : /* Cleanup the synced temporary slots */
601 akapila@postgresql.o 2042 :CBC 8 : ReplicationSlotCleanup(true);
2043 : :
2044 : : /* We are done with sync, so reset sync flag */
2045 : 8 : reset_syncing_flag();
2046 : : }
672 2047 [ - + ]: 9 : PG_END_ENSURE_ERROR_CLEANUP(slotsync_failure_callback, PointerGetDatum(wrconn));
2048 : 8 : }
|
