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