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