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