TLA Line data Source code
1 : /*-------------------------------------------------------------------------
2 : *
3 : * slot.c
4 : * Replication slot management.
5 : *
6 : *
7 : * Copyright (c) 2012-2023, PostgreSQL Global Development Group
8 : *
9 : *
10 : * IDENTIFICATION
11 : * src/backend/replication/slot.c
12 : *
13 : * NOTES
14 : *
15 : * Replication slots are used to keep state about replication streams
16 : * originating from this cluster. Their primary purpose is to prevent the
17 : * premature removal of WAL or of old tuple versions in a manner that would
18 : * interfere with replication; they are also useful for monitoring purposes.
19 : * Slots need to be permanent (to allow restarts), crash-safe, and allocatable
20 : * on standbys (to support cascading setups). The requirement that slots be
21 : * usable on standbys precludes storing them in the system catalogs.
22 : *
23 : * Each replication slot gets its own directory inside the $PGDATA/pg_replslot
24 : * directory. Inside that directory the state file will contain the slot's
25 : * own data. Additional data can be stored alongside that file if required.
26 : * While the server is running, the state data is also cached in memory for
27 : * efficiency.
28 : *
29 : * ReplicationSlotAllocationLock must be taken in exclusive mode to allocate
30 : * or free a slot. ReplicationSlotControlLock must be taken in shared mode
31 : * to iterate over the slots, and in exclusive mode to change the in_use flag
32 : * of a slot. The remaining data in each slot is protected by its mutex.
33 : *
34 : *-------------------------------------------------------------------------
35 : */
36 :
37 : #include "postgres.h"
38 :
39 : #include <unistd.h>
40 : #include <sys/stat.h>
41 :
42 : #include "access/transam.h"
43 : #include "access/xlog_internal.h"
44 : #include "access/xlogrecovery.h"
45 : #include "common/file_utils.h"
46 : #include "common/string.h"
47 : #include "miscadmin.h"
48 : #include "pgstat.h"
49 : #include "replication/slot.h"
50 : #include "storage/fd.h"
51 : #include "storage/ipc.h"
52 : #include "storage/proc.h"
53 : #include "storage/procarray.h"
54 : #include "utils/builtins.h"
55 :
56 : /*
57 : * Replication slot on-disk data structure.
58 : */
59 : typedef struct ReplicationSlotOnDisk
60 : {
61 : /* first part of this struct needs to be version independent */
62 :
63 : /* data not covered by checksum */
64 : uint32 magic;
65 : pg_crc32c checksum;
66 :
67 : /* data covered by checksum */
68 : uint32 version;
69 : uint32 length;
70 :
71 : /*
72 : * The actual data in the slot that follows can differ based on the above
73 : * 'version'.
74 : */
75 :
76 : ReplicationSlotPersistentData slotdata;
77 : } ReplicationSlotOnDisk;
78 :
79 : /* size of version independent data */
80 : #define ReplicationSlotOnDiskConstantSize \
81 : offsetof(ReplicationSlotOnDisk, slotdata)
82 : /* size of the part of the slot not covered by the checksum */
83 : #define ReplicationSlotOnDiskNotChecksummedSize \
84 : offsetof(ReplicationSlotOnDisk, version)
85 : /* size of the part covered by the checksum */
86 : #define ReplicationSlotOnDiskChecksummedSize \
87 : sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize
88 : /* size of the slot data that is version dependent */
89 : #define ReplicationSlotOnDiskV2Size \
90 : sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize
91 :
92 : #define SLOT_MAGIC 0x1051CA1 /* format identifier */
93 : #define SLOT_VERSION 3 /* version for new files */
94 :
95 : /* Control array for replication slot management */
96 : ReplicationSlotCtlData *ReplicationSlotCtl = NULL;
97 :
98 : /* My backend's replication slot in the shared memory array */
99 : ReplicationSlot *MyReplicationSlot = NULL;
100 :
101 : /* GUC variable */
102 : int max_replication_slots = 10; /* the maximum number of replication
103 : * slots */
104 :
105 : static void ReplicationSlotShmemExit(int code, Datum arg);
106 : static void ReplicationSlotDropAcquired(void);
107 : static void ReplicationSlotDropPtr(ReplicationSlot *slot);
108 :
109 : /* internal persistency functions */
110 : static void RestoreSlotFromDisk(const char *name);
111 : static void CreateSlotOnDisk(ReplicationSlot *slot);
112 : static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel);
113 :
114 : /*
115 : * Report shared-memory space needed by ReplicationSlotsShmemInit.
116 : */
117 : Size
118 GIC 6390 : ReplicationSlotsShmemSize(void)
119 : {
120 CBC 6390 : Size size = 0;
121 :
122 6390 : if (max_replication_slots == 0)
123 UIC 0 : return size;
124 ECB :
125 GBC 6390 : size = offsetof(ReplicationSlotCtlData, replication_slots);
126 GIC 6390 : size = add_size(size,
127 ECB : mul_size(max_replication_slots, sizeof(ReplicationSlot)));
128 :
129 GIC 6390 : return size;
130 : }
131 ECB :
132 : /*
133 : * Allocate and initialize shared memory for replication slots.
134 : */
135 : void
136 GIC 1826 : ReplicationSlotsShmemInit(void)
137 : {
138 ECB : bool found;
139 :
140 GIC 1826 : if (max_replication_slots == 0)
141 UIC 0 : return;
142 ECB :
143 GBC 1826 : ReplicationSlotCtl = (ReplicationSlotCtlData *)
144 GIC 1826 : ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
145 ECB : &found);
146 :
147 GIC 1826 : if (!found)
148 : {
149 ECB : int i;
150 :
151 : /* First time through, so initialize */
152 GIC 2189 : MemSet(ReplicationSlotCtl, 0, ReplicationSlotsShmemSize());
153 :
154 CBC 19967 : for (i = 0; i < max_replication_slots; i++)
155 : {
156 18141 : ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[i];
157 :
158 ECB : /* everything else is zeroed by the memset above */
159 GIC 18141 : SpinLockInit(&slot->mutex);
160 18141 : LWLockInitialize(&slot->io_in_progress_lock,
161 ECB : LWTRANCHE_REPLICATION_SLOT_IO);
162 CBC 18141 : ConditionVariableInit(&slot->active_cv);
163 : }
164 ECB : }
165 : }
166 :
167 : /*
168 : * Register the callback for replication slot cleanup and releasing.
169 : */
170 : void
171 GIC 13289 : ReplicationSlotInitialize(void)
172 : {
173 CBC 13289 : before_shmem_exit(ReplicationSlotShmemExit, 0);
174 GIC 13289 : }
175 ECB :
176 : /*
177 : * Release and cleanup replication slots.
178 : */
179 : static void
180 GIC 13289 : ReplicationSlotShmemExit(int code, Datum arg)
181 : {
182 ECB : /* Make sure active replication slots are released */
183 GIC 13289 : if (MyReplicationSlot != NULL)
184 153 : ReplicationSlotRelease();
185 ECB :
186 : /* Also cleanup all the temporary slots. */
187 GIC 13289 : ReplicationSlotCleanup();
188 13289 : }
189 ECB :
190 : /*
191 : * Check whether the passed slot name is valid and report errors at elevel.
192 : *
193 : * Slot names may consist out of [a-z0-9_]{1,NAMEDATALEN-1} which should allow
194 : * the name to be used as a directory name on every supported OS.
195 : *
196 : * Returns whether the directory name is valid or not if elevel < ERROR.
197 : */
198 : bool
199 GIC 603 : ReplicationSlotValidateName(const char *name, int elevel)
200 : {
201 ECB : const char *cp;
202 :
203 GIC 603 : if (strlen(name) == 0)
204 : {
205 CBC 3 : ereport(elevel,
206 : (errcode(ERRCODE_INVALID_NAME),
207 ECB : errmsg("replication slot name \"%s\" is too short",
208 : name)));
209 UIC 0 : return false;
210 : }
211 EUB :
212 GIC 600 : if (strlen(name) >= NAMEDATALEN)
213 : {
214 LBC 0 : ereport(elevel,
215 : (errcode(ERRCODE_NAME_TOO_LONG),
216 EUB : errmsg("replication slot name \"%s\" is too long",
217 : name)));
218 UIC 0 : return false;
219 : }
220 EUB :
221 GIC 13080 : for (cp = name; *cp; cp++)
222 : {
223 CBC 12481 : if (!((*cp >= 'a' && *cp <= 'z')
224 GIC 6433 : || (*cp >= '0' && *cp <= '9')
225 CBC 1189 : || (*cp == '_')))
226 ECB : {
227 CBC 1 : ereport(elevel,
228 : (errcode(ERRCODE_INVALID_NAME),
229 ECB : errmsg("replication slot name \"%s\" contains invalid character",
230 : name),
231 : errhint("Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
232 UIC 0 : return false;
233 : }
234 EUB : }
235 GIC 599 : return true;
236 : }
237 ECB :
238 : /*
239 : * Create a new replication slot and mark it as used by this backend.
240 : *
241 : * name: Name of the slot
242 : * db_specific: logical decoding is db specific; if the slot is going to
243 : * be used for that pass true, otherwise false.
244 : * two_phase: Allows decoding of prepared transactions. We allow this option
245 : * to be enabled only at the slot creation time. If we allow this option
246 : * to be changed during decoding then it is quite possible that we skip
247 : * prepare first time because this option was not enabled. Now next time
248 : * during getting changes, if the two_phase option is enabled it can skip
249 : * prepare because by that time start decoding point has been moved. So the
250 : * user will only get commit prepared.
251 : */
252 : void
253 GIC 474 : ReplicationSlotCreate(const char *name, bool db_specific,
254 : ReplicationSlotPersistency persistency, bool two_phase)
255 ECB : {
256 GIC 474 : ReplicationSlot *slot = NULL;
257 : int i;
258 ECB :
259 GIC 474 : Assert(MyReplicationSlot == NULL);
260 :
261 CBC 474 : ReplicationSlotValidateName(name, ERROR);
262 :
263 ECB : /*
264 : * If some other backend ran this code concurrently with us, we'd likely
265 : * both allocate the same slot, and that would be bad. We'd also be at
266 : * risk of missing a name collision. Also, we don't want to try to create
267 : * a new slot while somebody's busy cleaning up an old one, because we
268 : * might both be monkeying with the same directory.
269 : */
270 GIC 473 : LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
271 :
272 ECB : /*
273 : * Check for name collision, and identify an allocatable slot. We need to
274 : * hold ReplicationSlotControlLock in shared mode for this, so that nobody
275 : * else can change the in_use flags while we're looking at them.
276 : */
277 GIC 473 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
278 4383 : for (i = 0; i < max_replication_slots; i++)
279 ECB : {
280 CBC 3913 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
281 :
282 3913 : if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
283 GIC 3 : ereport(ERROR,
284 ECB : (errcode(ERRCODE_DUPLICATE_OBJECT),
285 : errmsg("replication slot \"%s\" already exists", name)));
286 GIC 3910 : if (!s->in_use && slot == NULL)
287 469 : slot = s;
288 ECB : }
289 CBC 470 : LWLockRelease(ReplicationSlotControlLock);
290 :
291 ECB : /* If all slots are in use, we're out of luck. */
292 GIC 470 : if (slot == NULL)
293 1 : ereport(ERROR,
294 ECB : (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
295 : errmsg("all replication slots are in use"),
296 : errhint("Free one or increase max_replication_slots.")));
297 :
298 : /*
299 : * Since this slot is not in use, nobody should be looking at any part of
300 : * it other than the in_use field unless they're trying to allocate it.
301 : * And since we hold ReplicationSlotAllocationLock, nobody except us can
302 : * be doing that. So it's safe to initialize the slot.
303 : */
304 GIC 469 : Assert(!slot->in_use);
305 469 : Assert(slot->active_pid == 0);
306 ECB :
307 : /* first initialize persistent data */
308 GIC 469 : memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
309 469 : namestrcpy(&slot->data.name, name);
310 CBC 469 : slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
311 469 : slot->data.persistency = persistency;
312 469 : slot->data.two_phase = two_phase;
313 469 : slot->data.two_phase_at = InvalidXLogRecPtr;
314 ECB :
315 : /* and then data only present in shared memory */
316 GIC 469 : slot->just_dirtied = false;
317 469 : slot->dirty = false;
318 CBC 469 : slot->effective_xmin = InvalidTransactionId;
319 469 : slot->effective_catalog_xmin = InvalidTransactionId;
320 469 : slot->candidate_catalog_xmin = InvalidTransactionId;
321 469 : slot->candidate_xmin_lsn = InvalidXLogRecPtr;
322 469 : slot->candidate_restart_valid = InvalidXLogRecPtr;
323 469 : slot->candidate_restart_lsn = InvalidXLogRecPtr;
324 ECB :
325 : /*
326 : * Create the slot on disk. We haven't actually marked the slot allocated
327 : * yet, so no special cleanup is required if this errors out.
328 : */
329 GIC 469 : CreateSlotOnDisk(slot);
330 :
331 ECB : /*
332 : * We need to briefly prevent any other backend from iterating over the
333 : * slots while we flip the in_use flag. We also need to set the active
334 : * flag while holding the ControlLock as otherwise a concurrent
335 : * ReplicationSlotAcquire() could acquire the slot as well.
336 : */
337 GIC 469 : LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
338 :
339 CBC 469 : slot->in_use = true;
340 :
341 ECB : /* We can now mark the slot active, and that makes it our slot. */
342 GIC 469 : SpinLockAcquire(&slot->mutex);
343 469 : Assert(slot->active_pid == 0);
344 CBC 469 : slot->active_pid = MyProcPid;
345 469 : SpinLockRelease(&slot->mutex);
346 469 : MyReplicationSlot = slot;
347 ECB :
348 CBC 469 : LWLockRelease(ReplicationSlotControlLock);
349 :
350 ECB : /*
351 : * Create statistics entry for the new logical slot. We don't collect any
352 : * stats for physical slots, so no need to create an entry for the same.
353 : * See ReplicationSlotDropPtr for why we need to do this before releasing
354 : * ReplicationSlotAllocationLock.
355 : */
356 GIC 469 : if (SlotIsLogical(slot))
357 344 : pgstat_create_replslot(slot);
358 ECB :
359 : /*
360 : * Now that the slot has been marked as in_use and active, it's safe to
361 : * let somebody else try to allocate a slot.
362 : */
363 GIC 469 : LWLockRelease(ReplicationSlotAllocationLock);
364 :
365 ECB : /* Let everybody know we've modified this slot */
366 GIC 469 : ConditionVariableBroadcast(&slot->active_cv);
367 469 : }
368 ECB :
369 : /*
370 : * Search for the named replication slot.
371 : *
372 : * Return the replication slot if found, otherwise NULL.
373 : */
374 : ReplicationSlot *
375 GIC 985 : SearchNamedReplicationSlot(const char *name, bool need_lock)
376 : {
377 ECB : int i;
378 GIC 985 : ReplicationSlot *slot = NULL;
379 :
380 CBC 985 : if (need_lock)
381 GIC 66 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
382 ECB :
383 CBC 1643 : for (i = 0; i < max_replication_slots; i++)
384 : {
385 1627 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
386 :
387 1627 : if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
388 : {
389 969 : slot = s;
390 GIC 969 : break;
391 ECB : }
392 : }
393 :
394 GIC 985 : if (need_lock)
395 66 : LWLockRelease(ReplicationSlotControlLock);
396 ECB :
397 CBC 985 : return slot;
398 : }
399 ECB :
400 : /*
401 : * Return the index of the replication slot in
402 : * ReplicationSlotCtl->replication_slots.
403 : *
404 : * This is mainly useful to have an efficient key for storing replication slot
405 : * stats.
406 : */
407 : int
408 GIC 6891 : ReplicationSlotIndex(ReplicationSlot *slot)
409 : {
410 CBC 6891 : Assert(slot >= ReplicationSlotCtl->replication_slots &&
411 : slot < ReplicationSlotCtl->replication_slots + max_replication_slots);
412 ECB :
413 GIC 6891 : return slot - ReplicationSlotCtl->replication_slots;
414 : }
415 ECB :
416 : /*
417 : * If the slot at 'index' is unused, return false. Otherwise 'name' is set to
418 : * the slot's name and true is returned.
419 : *
420 : * This likely is only useful for pgstat_replslot.c during shutdown, in other
421 : * cases there are obvious TOCTOU issues.
422 : */
423 : bool
424 GIC 43 : ReplicationSlotName(int index, Name name)
425 : {
426 ECB : ReplicationSlot *slot;
427 : bool found;
428 :
429 GIC 43 : slot = &ReplicationSlotCtl->replication_slots[index];
430 :
431 ECB : /*
432 : * Ensure that the slot cannot be dropped while we copy the name. Don't
433 : * need the spinlock as the name of an existing slot cannot change.
434 : */
435 GIC 43 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
436 43 : found = slot->in_use;
437 CBC 43 : if (slot->in_use)
438 43 : namestrcpy(name, NameStr(slot->data.name));
439 43 : LWLockRelease(ReplicationSlotControlLock);
440 ECB :
441 CBC 43 : return found;
442 : }
443 ECB :
444 : /*
445 : * Find a previously created slot and mark it as used by this process.
446 : *
447 : * An error is raised if nowait is true and the slot is currently in use. If
448 : * nowait is false, we sleep until the slot is released by the owning process.
449 : */
450 : void
451 GIC 913 : ReplicationSlotAcquire(const char *name, bool nowait)
452 : {
453 ECB : ReplicationSlot *s;
454 : int active_pid;
455 :
456 GNC 913 : Assert(name != NULL);
457 :
458 CBC 913 : retry:
459 GIC 913 : Assert(MyReplicationSlot == NULL);
460 ECB :
461 CBC 913 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
462 :
463 ECB : /*
464 : * Search for the slot with the specified name if the slot to acquire is
465 : * not given. If the slot is not found, we either return -1 or error out.
466 : */
467 GIC 913 : s = SearchNamedReplicationSlot(name, false);
468 913 : if (s == NULL || !s->in_use)
469 ECB : {
470 CBC 9 : LWLockRelease(ReplicationSlotControlLock);
471 :
472 9 : ereport(ERROR,
473 : (errcode(ERRCODE_UNDEFINED_OBJECT),
474 ECB : errmsg("replication slot \"%s\" does not exist",
475 : name)));
476 : }
477 :
478 : /*
479 : * This is the slot we want; check if it's active under some other
480 : * process. In single user mode, we don't need this check.
481 : */
482 GIC 904 : if (IsUnderPostmaster)
483 : {
484 ECB : /*
485 : * Get ready to sleep on the slot in case it is active. (We may end
486 : * up not sleeping, but we don't want to do this while holding the
487 : * spinlock.)
488 : */
489 GIC 904 : if (!nowait)
490 189 : ConditionVariablePrepareToSleep(&s->active_cv);
491 ECB :
492 CBC 904 : SpinLockAcquire(&s->mutex);
493 GIC 904 : if (s->active_pid == 0)
494 CBC 804 : s->active_pid = MyProcPid;
495 904 : active_pid = s->active_pid;
496 904 : SpinLockRelease(&s->mutex);
497 ECB : }
498 : else
499 UIC 0 : active_pid = MyProcPid;
500 GIC 904 : LWLockRelease(ReplicationSlotControlLock);
501 EUB :
502 ECB : /*
503 : * If we found the slot but it's already active in another process, we
504 : * wait until the owning process signals us that it's been released, or
505 : * error out.
506 : */
507 GIC 904 : if (active_pid != MyProcPid)
508 : {
509 LBC 0 : if (!nowait)
510 : {
511 EUB : /* Wait here until we get signaled, and then restart */
512 UIC 0 : ConditionVariableSleep(&s->active_cv,
513 : WAIT_EVENT_REPLICATION_SLOT_DROP);
514 UBC 0 : ConditionVariableCancelSleep();
515 UIC 0 : goto retry;
516 EUB : }
517 :
518 UIC 0 : ereport(ERROR,
519 : (errcode(ERRCODE_OBJECT_IN_USE),
520 EUB : errmsg("replication slot \"%s\" is active for PID %d",
521 : NameStr(s->data.name), active_pid)));
522 : }
523 GIC 904 : else if (!nowait)
524 189 : ConditionVariableCancelSleep(); /* no sleep needed after all */
525 ECB :
526 : /* Let everybody know we've modified this slot */
527 GIC 904 : ConditionVariableBroadcast(&s->active_cv);
528 :
529 ECB : /* We made this slot active, so it's ours now. */
530 GIC 904 : MyReplicationSlot = s;
531 :
532 ECB : /*
533 : * The call to pgstat_acquire_replslot() protects against stats for a
534 : * different slot, from before a restart or such, being present during
535 : * pgstat_report_replslot().
536 : */
537 GIC 904 : if (SlotIsLogical(s))
538 767 : pgstat_acquire_replslot(s);
539 CBC 904 : }
540 ECB :
541 : /*
542 : * Release the replication slot that this backend considers to own.
543 : *
544 : * This or another backend can re-acquire the slot later.
545 : * Resources this slot requires will be preserved.
546 : */
547 : void
548 GIC 1092 : ReplicationSlotRelease(void)
549 : {
550 CBC 1092 : ReplicationSlot *slot = MyReplicationSlot;
551 :
552 1092 : Assert(slot != NULL && slot->active_pid != 0);
553 :
554 1092 : if (slot->data.persistency == RS_EPHEMERAL)
555 : {
556 ECB : /*
557 : * Delete the slot. There is no !PANIC case where this is allowed to
558 : * fail, all that may happen is an incomplete cleanup of the on-disk
559 : * data.
560 : */
561 GIC 5 : ReplicationSlotDropAcquired();
562 : }
563 ECB :
564 : /*
565 : * If slot needed to temporarily restrain both data and catalog xmin to
566 : * create the catalog snapshot, remove that temporary constraint.
567 : * Snapshots can only be exported while the initial snapshot is still
568 : * acquired.
569 : */
570 GIC 1092 : if (!TransactionIdIsValid(slot->data.xmin) &&
571 1087 : TransactionIdIsValid(slot->effective_xmin))
572 ECB : {
573 CBC 155 : SpinLockAcquire(&slot->mutex);
574 GIC 155 : slot->effective_xmin = InvalidTransactionId;
575 CBC 155 : SpinLockRelease(&slot->mutex);
576 155 : ReplicationSlotsComputeRequiredXmin(false);
577 ECB : }
578 :
579 GIC 1092 : if (slot->data.persistency == RS_PERSISTENT)
580 : {
581 ECB : /*
582 : * Mark persistent slot inactive. We're not freeing it, just
583 : * disconnecting, but wake up others that may be waiting for it.
584 : */
585 GIC 886 : SpinLockAcquire(&slot->mutex);
586 886 : slot->active_pid = 0;
587 CBC 886 : SpinLockRelease(&slot->mutex);
588 886 : ConditionVariableBroadcast(&slot->active_cv);
589 ECB : }
590 :
591 GIC 1092 : MyReplicationSlot = NULL;
592 :
593 ECB : /* might not have been set when we've been a plain slot */
594 GIC 1092 : LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
595 1092 : MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING;
596 CBC 1092 : ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
597 1092 : LWLockRelease(ProcArrayLock);
598 1092 : }
599 ECB :
600 : /*
601 : * Cleanup all temporary slots created in current session.
602 : */
603 : void
604 GIC 31075 : ReplicationSlotCleanup(void)
605 : {
606 ECB : int i;
607 :
608 GIC 31075 : Assert(MyReplicationSlot == NULL);
609 :
610 CBC 31075 : restart:
611 GIC 31176 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
612 CBC 338131 : for (i = 0; i < max_replication_slots; i++)
613 ECB : {
614 CBC 307056 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
615 :
616 307056 : if (!s->in_use)
617 GIC 297495 : continue;
618 ECB :
619 CBC 9561 : SpinLockAcquire(&s->mutex);
620 GIC 9561 : if (s->active_pid == MyProcPid)
621 ECB : {
622 CBC 101 : Assert(s->data.persistency == RS_TEMPORARY);
623 GIC 101 : SpinLockRelease(&s->mutex);
624 CBC 101 : LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
625 ECB :
626 CBC 101 : ReplicationSlotDropPtr(s);
627 :
628 101 : ConditionVariableBroadcast(&s->active_cv);
629 GIC 101 : goto restart;
630 ECB : }
631 : else
632 GIC 9460 : SpinLockRelease(&s->mutex);
633 : }
634 ECB :
635 GIC 31075 : LWLockRelease(ReplicationSlotControlLock);
636 31075 : }
637 ECB :
638 : /*
639 : * Permanently drop replication slot identified by the passed in name.
640 : */
641 : void
642 GIC 299 : ReplicationSlotDrop(const char *name, bool nowait)
643 : {
644 CBC 299 : Assert(MyReplicationSlot == NULL);
645 :
646 299 : ReplicationSlotAcquire(name, nowait);
647 :
648 294 : ReplicationSlotDropAcquired();
649 GIC 294 : }
650 ECB :
651 : /*
652 : * Permanently drop the currently acquired replication slot.
653 : */
654 : static void
655 GIC 304 : ReplicationSlotDropAcquired(void)
656 : {
657 CBC 304 : ReplicationSlot *slot = MyReplicationSlot;
658 :
659 304 : Assert(MyReplicationSlot != NULL);
660 :
661 ECB : /* slot isn't acquired anymore */
662 GIC 304 : MyReplicationSlot = NULL;
663 :
664 CBC 304 : ReplicationSlotDropPtr(slot);
665 GIC 304 : }
666 ECB :
667 : /*
668 : * Permanently drop the replication slot which will be released by the point
669 : * this function returns.
670 : */
671 : static void
672 GIC 405 : ReplicationSlotDropPtr(ReplicationSlot *slot)
673 : {
674 ECB : char path[MAXPGPATH];
675 : char tmppath[MAXPGPATH];
676 :
677 : /*
678 : * If some other backend ran this code concurrently with us, we might try
679 : * to delete a slot with a certain name while someone else was trying to
680 : * create a slot with the same name.
681 : */
682 GIC 405 : LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
683 :
684 ECB : /* Generate pathnames. */
685 GIC 405 : sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
686 405 : sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
687 ECB :
688 : /*
689 : * Rename the slot directory on disk, so that we'll no longer recognize
690 : * this as a valid slot. Note that if this fails, we've got to mark the
691 : * slot inactive before bailing out. If we're dropping an ephemeral or a
692 : * temporary slot, we better never fail hard as the caller won't expect
693 : * the slot to survive and this might get called during error handling.
694 : */
695 GIC 405 : if (rename(path, tmppath) == 0)
696 : {
697 ECB : /*
698 : * We need to fsync() the directory we just renamed and its parent to
699 : * make sure that our changes are on disk in a crash-safe fashion. If
700 : * fsync() fails, we can't be sure whether the changes are on disk or
701 : * not. For now, we handle that by panicking;
702 : * StartupReplicationSlots() will try to straighten it out after
703 : * restart.
704 : */
705 GIC 405 : START_CRIT_SECTION();
706 405 : fsync_fname(tmppath, true);
707 CBC 405 : fsync_fname("pg_replslot", true);
708 405 : END_CRIT_SECTION();
709 ECB : }
710 : else
711 : {
712 UIC 0 : bool fail_softly = slot->data.persistency != RS_PERSISTENT;
713 :
714 UBC 0 : SpinLockAcquire(&slot->mutex);
715 UIC 0 : slot->active_pid = 0;
716 UBC 0 : SpinLockRelease(&slot->mutex);
717 EUB :
718 : /* wake up anyone waiting on this slot */
719 UIC 0 : ConditionVariableBroadcast(&slot->active_cv);
720 :
721 UBC 0 : ereport(fail_softly ? WARNING : ERROR,
722 : (errcode_for_file_access(),
723 EUB : errmsg("could not rename file \"%s\" to \"%s\": %m",
724 : path, tmppath)));
725 : }
726 :
727 : /*
728 : * The slot is definitely gone. Lock out concurrent scans of the array
729 : * long enough to kill it. It's OK to clear the active PID here without
730 : * grabbing the mutex because nobody else can be scanning the array here,
731 : * and nobody can be attached to this slot and thus access it without
732 : * scanning the array.
733 : *
734 : * Also wake up processes waiting for it.
735 : */
736 GIC 405 : LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
737 405 : slot->active_pid = 0;
738 CBC 405 : slot->in_use = false;
739 405 : LWLockRelease(ReplicationSlotControlLock);
740 405 : ConditionVariableBroadcast(&slot->active_cv);
741 ECB :
742 : /*
743 : * Slot is dead and doesn't prevent resource removal anymore, recompute
744 : * limits.
745 : */
746 GIC 405 : ReplicationSlotsComputeRequiredXmin(false);
747 405 : ReplicationSlotsComputeRequiredLSN();
748 ECB :
749 : /*
750 : * If removing the directory fails, the worst thing that will happen is
751 : * that the user won't be able to create a new slot with the same name
752 : * until the next server restart. We warn about it, but that's all.
753 : */
754 GIC 405 : if (!rmtree(tmppath, true))
755 UIC 0 : ereport(WARNING,
756 ECB : (errmsg("could not remove directory \"%s\"", tmppath)));
757 EUB :
758 : /*
759 : * Drop the statistics entry for the replication slot. Do this while
760 : * holding ReplicationSlotAllocationLock so that we don't drop a
761 : * statistics entry for another slot with the same name just created in
762 : * another session.
763 : */
764 GIC 405 : if (SlotIsLogical(slot))
765 298 : pgstat_drop_replslot(slot);
766 ECB :
767 : /*
768 : * We release this at the very end, so that nobody starts trying to create
769 : * a slot while we're still cleaning up the detritus of the old one.
770 : */
771 GIC 405 : LWLockRelease(ReplicationSlotAllocationLock);
772 405 : }
773 ECB :
774 : /*
775 : * Serialize the currently acquired slot's state from memory to disk, thereby
776 : * guaranteeing the current state will survive a crash.
777 : */
778 : void
779 GIC 971 : ReplicationSlotSave(void)
780 : {
781 ECB : char path[MAXPGPATH];
782 :
783 GIC 971 : Assert(MyReplicationSlot != NULL);
784 :
785 CBC 971 : sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
786 GIC 971 : SaveSlotToPath(MyReplicationSlot, path, ERROR);
787 CBC 971 : }
788 ECB :
789 : /*
790 : * Signal that it would be useful if the currently acquired slot would be
791 : * flushed out to disk.
792 : *
793 : * Note that the actual flush to disk can be delayed for a long time, if
794 : * required for correctness explicitly do a ReplicationSlotSave().
795 : */
796 : void
797 GIC 18803 : ReplicationSlotMarkDirty(void)
798 : {
799 CBC 18803 : ReplicationSlot *slot = MyReplicationSlot;
800 :
801 18803 : Assert(MyReplicationSlot != NULL);
802 :
803 18803 : SpinLockAcquire(&slot->mutex);
804 GIC 18803 : MyReplicationSlot->just_dirtied = true;
805 CBC 18803 : MyReplicationSlot->dirty = true;
806 18803 : SpinLockRelease(&slot->mutex);
807 18803 : }
808 ECB :
809 : /*
810 : * Convert a slot that's marked as RS_EPHEMERAL to a RS_PERSISTENT slot,
811 : * guaranteeing it will be there after an eventual crash.
812 : */
813 : void
814 GIC 331 : ReplicationSlotPersist(void)
815 : {
816 CBC 331 : ReplicationSlot *slot = MyReplicationSlot;
817 :
818 331 : Assert(slot != NULL);
819 GIC 331 : Assert(slot->data.persistency != RS_PERSISTENT);
820 ECB :
821 CBC 331 : SpinLockAcquire(&slot->mutex);
822 GIC 331 : slot->data.persistency = RS_PERSISTENT;
823 CBC 331 : SpinLockRelease(&slot->mutex);
824 ECB :
825 CBC 331 : ReplicationSlotMarkDirty();
826 GIC 331 : ReplicationSlotSave();
827 CBC 331 : }
828 ECB :
829 : /*
830 : * Compute the oldest xmin across all slots and store it in the ProcArray.
831 : *
832 : * If already_locked is true, ProcArrayLock has already been acquired
833 : * exclusively.
834 : */
835 : void
836 GIC 2189 : ReplicationSlotsComputeRequiredXmin(bool already_locked)
837 : {
838 ECB : int i;
839 GIC 2189 : TransactionId agg_xmin = InvalidTransactionId;
840 2189 : TransactionId agg_catalog_xmin = InvalidTransactionId;
841 ECB :
842 CBC 2189 : Assert(ReplicationSlotCtl != NULL);
843 :
844 2189 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
845 :
846 22253 : for (i = 0; i < max_replication_slots; i++)
847 : {
848 20064 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
849 : TransactionId effective_xmin;
850 ECB : TransactionId effective_catalog_xmin;
851 : bool invalidated;
852 :
853 GIC 20064 : if (!s->in_use)
854 18542 : continue;
855 ECB :
856 CBC 1522 : SpinLockAcquire(&s->mutex);
857 GIC 1522 : effective_xmin = s->effective_xmin;
858 CBC 1522 : effective_catalog_xmin = s->effective_catalog_xmin;
859 GNC 1522 : invalidated = s->data.invalidated != RS_INVAL_NONE;
860 CBC 1522 : SpinLockRelease(&s->mutex);
861 ECB :
862 : /* invalidated slots need not apply */
863 GIC 1522 : if (invalidated)
864 CBC 21 : continue;
865 ECB :
866 : /* check the data xmin */
867 GIC 1501 : if (TransactionIdIsValid(effective_xmin) &&
868 CBC 3 : (!TransactionIdIsValid(agg_xmin) ||
869 3 : TransactionIdPrecedes(effective_xmin, agg_xmin)))
870 210 : agg_xmin = effective_xmin;
871 ECB :
872 : /* check the catalog xmin */
873 GIC 1501 : if (TransactionIdIsValid(effective_catalog_xmin) &&
874 CBC 599 : (!TransactionIdIsValid(agg_catalog_xmin) ||
875 599 : TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
876 789 : agg_catalog_xmin = effective_catalog_xmin;
877 ECB : }
878 :
879 GIC 2189 : LWLockRelease(ReplicationSlotControlLock);
880 ECB :
881 GIC 2189 : ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
882 CBC 2189 : }
883 ECB :
884 : /*
885 : * Compute the oldest restart LSN across all slots and inform xlog module.
886 : *
887 : * Note: while max_slot_wal_keep_size is theoretically relevant for this
888 : * purpose, we don't try to account for that, because this module doesn't
889 : * know what to compare against.
890 : */
891 : void
892 GIC 19626 : ReplicationSlotsComputeRequiredLSN(void)
893 ECB : {
894 : int i;
895 GIC 19626 : XLogRecPtr min_required = InvalidXLogRecPtr;
896 ECB :
897 GIC 19626 : Assert(ReplicationSlotCtl != NULL);
898 ECB :
899 GIC 19626 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
900 CBC 212248 : for (i = 0; i < max_replication_slots; i++)
901 ECB : {
902 GIC 192622 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
903 ECB : XLogRecPtr restart_lsn;
904 : bool invalidated;
905 :
906 GIC 192622 : if (!s->in_use)
907 173627 : continue;
908 ECB :
909 CBC 18995 : SpinLockAcquire(&s->mutex);
910 GIC 18995 : restart_lsn = s->data.restart_lsn;
911 GNC 18995 : invalidated = s->data.invalidated != RS_INVAL_NONE;
912 CBC 18995 : SpinLockRelease(&s->mutex);
913 ECB :
914 : /* invalidated slots need not apply */
915 GNC 18995 : if (invalidated)
916 38 : continue;
917 :
918 CBC 18957 : if (restart_lsn != InvalidXLogRecPtr &&
919 648 : (min_required == InvalidXLogRecPtr ||
920 : restart_lsn < min_required))
921 GIC 18389 : min_required = restart_lsn;
922 ECB : }
923 CBC 19626 : LWLockRelease(ReplicationSlotControlLock);
924 :
925 19626 : XLogSetReplicationSlotMinimumLSN(min_required);
926 19626 : }
927 :
928 ECB : /*
929 : * Compute the oldest WAL LSN required by *logical* decoding slots..
930 : *
931 : * Returns InvalidXLogRecPtr if logical decoding is disabled or no logical
932 : * slots exist.
933 : *
934 : * NB: this returns a value >= ReplicationSlotsComputeRequiredLSN(), since it
935 : * ignores physical replication slots.
936 : *
937 : * The results aren't required frequently, so we don't maintain a precomputed
938 : * value like we do for ComputeRequiredLSN() and ComputeRequiredXmin().
939 : */
940 : XLogRecPtr
941 GIC 4726 : ReplicationSlotsComputeLogicalRestartLSN(void)
942 : {
943 4726 : XLogRecPtr result = InvalidXLogRecPtr;
944 : int i;
945 :
946 4726 : if (max_replication_slots <= 0)
947 UIC 0 : return InvalidXLogRecPtr;
948 ECB :
949 GIC 4726 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
950 ECB :
951 GIC 51678 : for (i = 0; i < max_replication_slots; i++)
952 : {
953 ECB : ReplicationSlot *s;
954 EUB : XLogRecPtr restart_lsn;
955 : bool invalidated;
956 :
957 CBC 46952 : s = &ReplicationSlotCtl->replication_slots[i];
958 :
959 ECB : /* cannot change while ReplicationSlotCtlLock is held */
960 GIC 46952 : if (!s->in_use)
961 46714 : continue;
962 :
963 : /* we're only interested in logical slots */
964 238 : if (!SlotIsLogical(s))
965 CBC 124 : continue;
966 :
967 : /* read once, it's ok if it increases while we're checking */
968 114 : SpinLockAcquire(&s->mutex);
969 114 : restart_lsn = s->data.restart_lsn;
970 GNC 114 : invalidated = s->data.invalidated != RS_INVAL_NONE;
971 GIC 114 : SpinLockRelease(&s->mutex);
972 :
973 : /* invalidated slots need not apply */
974 GNC 114 : if (invalidated)
975 UNC 0 : continue;
976 :
977 CBC 114 : if (restart_lsn == InvalidXLogRecPtr)
978 LBC 0 : continue;
979 :
980 GIC 114 : if (result == InvalidXLogRecPtr ||
981 ECB : restart_lsn < result)
982 CBC 98 : result = restart_lsn;
983 ECB : }
984 :
985 GIC 4726 : LWLockRelease(ReplicationSlotControlLock);
986 :
987 CBC 4726 : return result;
988 EUB : }
989 :
990 ECB : /*
991 EUB : * ReplicationSlotsCountDBSlots -- count the number of slots that refer to the
992 : * passed database oid.
993 ECB : *
994 : * Returns true if there are any slots referencing the database. *nslots will
995 : * be set to the absolute number of slots in the database, *nactive to ones
996 : * currently active.
997 : */
998 : bool
999 GIC 21 : ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
1000 ECB : {
1001 : int i;
1002 :
1003 GIC 21 : *nslots = *nactive = 0;
1004 :
1005 21 : if (max_replication_slots <= 0)
1006 UIC 0 : return false;
1007 :
1008 GIC 21 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1009 218 : for (i = 0; i < max_replication_slots; i++)
1010 : {
1011 : ReplicationSlot *s;
1012 ECB :
1013 GIC 197 : s = &ReplicationSlotCtl->replication_slots[i];
1014 :
1015 : /* cannot change while ReplicationSlotCtlLock is held */
1016 CBC 197 : if (!s->in_use)
1017 GIC 183 : continue;
1018 ECB :
1019 EUB : /* only logical slots are database specific, skip */
1020 GIC 14 : if (!SlotIsLogical(s))
1021 CBC 8 : continue;
1022 ECB :
1023 : /* not our database, skip */
1024 GIC 6 : if (s->data.database != dboid)
1025 3 : continue;
1026 ECB :
1027 : /* NB: intentionally counting invalidated slots */
1028 :
1029 : /* count slots with spinlock held */
1030 GIC 3 : SpinLockAcquire(&s->mutex);
1031 CBC 3 : (*nslots)++;
1032 3 : if (s->active_pid != 0)
1033 GIC 1 : (*nactive)++;
1034 3 : SpinLockRelease(&s->mutex);
1035 ECB : }
1036 CBC 21 : LWLockRelease(ReplicationSlotControlLock);
1037 :
1038 GIC 21 : if (*nslots > 0)
1039 CBC 3 : return true;
1040 18 : return false;
1041 : }
1042 :
1043 : /*
1044 : * ReplicationSlotsDropDBSlots -- Drop all db-specific slots relating to the
1045 ECB : * passed database oid. The caller should hold an exclusive lock on the
1046 : * pg_database oid for the database to prevent creation of new slots on the db
1047 : * or replay from existing slots.
1048 : *
1049 : * Another session that concurrently acquires an existing slot on the target DB
1050 : * (most likely to drop it) may cause this function to ERROR. If that happens
1051 : * it may have dropped some but not all slots.
1052 : *
1053 : * This routine isn't as efficient as it could be - but we don't drop
1054 : * databases often, especially databases with lots of slots.
1055 : */
1056 : void
1057 GIC 29 : ReplicationSlotsDropDBSlots(Oid dboid)
1058 : {
1059 : int i;
1060 :
1061 29 : if (max_replication_slots <= 0)
1062 UIC 0 : return;
1063 :
1064 GIC 29 : restart:
1065 34 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1066 303 : for (i = 0; i < max_replication_slots; i++)
1067 : {
1068 : ReplicationSlot *s;
1069 : char *slotname;
1070 : int active_pid;
1071 :
1072 CBC 274 : s = &ReplicationSlotCtl->replication_slots[i];
1073 :
1074 : /* cannot change while ReplicationSlotCtlLock is held */
1075 GIC 274 : if (!s->in_use)
1076 CBC 252 : continue;
1077 EUB :
1078 : /* only logical slots are database specific, skip */
1079 CBC 22 : if (!SlotIsLogical(s))
1080 10 : continue;
1081 ECB :
1082 : /* not our database, skip */
1083 GIC 12 : if (s->data.database != dboid)
1084 7 : continue;
1085 :
1086 : /* NB: intentionally including invalidated slots */
1087 :
1088 : /* acquire slot, so ReplicationSlotDropAcquired can be reused */
1089 CBC 5 : SpinLockAcquire(&s->mutex);
1090 : /* can't change while ReplicationSlotControlLock is held */
1091 GIC 5 : slotname = NameStr(s->data.name);
1092 CBC 5 : active_pid = s->active_pid;
1093 5 : if (active_pid == 0)
1094 : {
1095 GIC 5 : MyReplicationSlot = s;
1096 CBC 5 : s->active_pid = MyProcPid;
1097 ECB : }
1098 GIC 5 : SpinLockRelease(&s->mutex);
1099 :
1100 ECB : /*
1101 : * Even though we hold an exclusive lock on the database object a
1102 : * logical slot for that DB can still be active, e.g. if it's
1103 : * concurrently being dropped by a backend connected to another DB.
1104 : *
1105 : * That's fairly unlikely in practice, so we'll just bail out.
1106 : */
1107 GIC 5 : if (active_pid)
1108 LBC 0 : ereport(ERROR,
1109 ECB : (errcode(ERRCODE_OBJECT_IN_USE),
1110 : errmsg("replication slot \"%s\" is active for PID %d",
1111 : slotname, active_pid)));
1112 :
1113 : /*
1114 : * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1115 : * holding ReplicationSlotControlLock over filesystem operations,
1116 : * release ReplicationSlotControlLock and use
1117 : * ReplicationSlotDropAcquired.
1118 : *
1119 : * As that means the set of slots could change, restart scan from the
1120 : * beginning each time we release the lock.
1121 : */
1122 GIC 5 : LWLockRelease(ReplicationSlotControlLock);
1123 5 : ReplicationSlotDropAcquired();
1124 CBC 5 : goto restart;
1125 EUB : }
1126 GIC 29 : LWLockRelease(ReplicationSlotControlLock);
1127 : }
1128 :
1129 :
1130 : /*
1131 : * Check whether the server's configuration supports using replication
1132 : * slots.
1133 : */
1134 : void
1135 1320 : CheckSlotRequirements(void)
1136 : {
1137 : /*
1138 : * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1139 ECB : * needs the same check.
1140 : */
1141 :
1142 GIC 1320 : if (max_replication_slots == 0)
1143 LBC 0 : ereport(ERROR,
1144 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1145 : errmsg("replication slots can only be used if max_replication_slots > 0")));
1146 :
1147 GIC 1320 : if (wal_level < WAL_LEVEL_REPLICA)
1148 UIC 0 : ereport(ERROR,
1149 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1150 : errmsg("replication slots can only be used if wal_level >= replica")));
1151 GIC 1320 : }
1152 ECB :
1153 : /*
1154 : * Check whether the user has privilege to use replication slots.
1155 : */
1156 : void
1157 GIC 441 : CheckSlotPermissions(void)
1158 : {
1159 GNC 441 : if (!has_rolreplication(GetUserId()))
1160 GBC 4 : ereport(ERROR,
1161 : (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1162 : errmsg("permission denied to use replication slots"),
1163 : errdetail("Only roles with the %s attribute may use replication slots.",
1164 : "REPLICATION")));
1165 GIC 437 : }
1166 ECB :
1167 EUB : /*
1168 : * Reserve WAL for the currently active slot.
1169 : *
1170 ECB : * Compute and set restart_lsn in a manner that's appropriate for the type of
1171 : * the slot and concurrency safe.
1172 : */
1173 : void
1174 GIC 442 : ReplicationSlotReserveWal(void)
1175 : {
1176 CBC 442 : ReplicationSlot *slot = MyReplicationSlot;
1177 :
1178 442 : Assert(slot != NULL);
1179 442 : Assert(slot->data.restart_lsn == InvalidXLogRecPtr);
1180 :
1181 : /*
1182 : * The replication slot mechanism is used to prevent removal of required
1183 : * WAL. As there is no interlock between this routine and checkpoints, WAL
1184 ECB : * segments could concurrently be removed when a now stale return value of
1185 : * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1186 : * this happens we'll just retry.
1187 : */
1188 : while (true)
1189 UIC 0 : {
1190 : XLogSegNo segno;
1191 : XLogRecPtr restart_lsn;
1192 :
1193 ECB : /*
1194 : * For logical slots log a standby snapshot and start logical decoding
1195 : * at exactly that position. That allows the slot to start up more
1196 : * quickly. But on a standby we cannot do WAL writes, so just use the
1197 : * replay pointer; effectively, an attempt to create a logical slot on
1198 : * standby will cause it to wait for an xl_running_xact record to be
1199 : * logged independently on the primary, so that a snapshot can be
1200 : * built using the record.
1201 : *
1202 : * None of this is needed (or indeed helpful) for physical slots as
1203 : * they'll start replay at the last logged checkpoint anyway. Instead
1204 : * return the location of the last redo LSN. While that slightly
1205 : * increases the chance that we have to retry, it's where a base
1206 : * backup has to start replay at.
1207 : */
1208 GNC 442 : if (SlotIsPhysical(slot))
1209 GIC 106 : restart_lsn = GetRedoRecPtr();
1210 GNC 336 : else if (RecoveryInProgress())
1211 21 : restart_lsn = GetXLogReplayRecPtr(NULL);
1212 : else
1213 315 : restart_lsn = GetXLogInsertRecPtr();
1214 :
1215 442 : SpinLockAcquire(&slot->mutex);
1216 442 : slot->data.restart_lsn = restart_lsn;
1217 442 : SpinLockRelease(&slot->mutex);
1218 ECB :
1219 : /* prevent WAL removal as fast as possible */
1220 CBC 442 : ReplicationSlotsComputeRequiredLSN();
1221 ECB :
1222 : /*
1223 : * If all required WAL is still there, great, otherwise retry. The
1224 : * slot should prevent further removal of WAL, unless there's a
1225 : * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1226 : * the new restart_lsn above, so normally we should never need to loop
1227 : * more than twice.
1228 : */
1229 GIC 442 : XLByteToSeg(slot->data.restart_lsn, segno, wal_segment_size);
1230 CBC 442 : if (XLogGetLastRemovedSegno() < segno)
1231 GIC 442 : break;
1232 : }
1233 :
1234 GNC 442 : if (!RecoveryInProgress() && SlotIsLogical(slot))
1235 : {
1236 : XLogRecPtr flushptr;
1237 :
1238 : /* make sure we have enough information to start */
1239 315 : flushptr = LogStandbySnapshot();
1240 :
1241 : /* and make sure it's fsynced to disk */
1242 315 : XLogFlush(flushptr);
1243 : }
1244 GIC 442 : }
1245 :
1246 : /*
1247 : * Report that replication slot needs to be invalidated
1248 : */
1249 : static void
1250 GNC 20 : ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
1251 : bool terminating,
1252 : int pid,
1253 : NameData slotname,
1254 : XLogRecPtr restart_lsn,
1255 : XLogRecPtr oldestLSN,
1256 : TransactionId snapshotConflictHorizon)
1257 : {
1258 : StringInfoData err_detail;
1259 20 : bool hint = false;
1260 :
1261 20 : initStringInfo(&err_detail);
1262 :
1263 20 : switch (cause)
1264 : {
1265 5 : case RS_INVAL_WAL_REMOVED:
1266 5 : hint = true;
1267 5 : appendStringInfo(&err_detail, _("The slot's restart_lsn %X/%X exceeds the limit by %llu bytes."),
1268 5 : LSN_FORMAT_ARGS(restart_lsn),
1269 5 : (unsigned long long) (oldestLSN - restart_lsn));
1270 5 : break;
1271 12 : case RS_INVAL_HORIZON:
1272 12 : appendStringInfo(&err_detail, _("The slot conflicted with xid horizon %u."),
1273 : snapshotConflictHorizon);
1274 12 : break;
1275 :
1276 3 : case RS_INVAL_WAL_LEVEL:
1277 3 : appendStringInfo(&err_detail, _("Logical decoding on standby requires wal_level to be at least logical on the primary server"));
1278 3 : break;
1279 UNC 0 : case RS_INVAL_NONE:
1280 0 : pg_unreachable();
1281 : }
1282 :
1283 GNC 20 : ereport(LOG,
1284 : terminating ?
1285 : errmsg("terminating process %d to release replication slot \"%s\"",
1286 : pid, NameStr(slotname)) :
1287 : errmsg("invalidating obsolete replication slot \"%s\"",
1288 : NameStr(slotname)),
1289 : errdetail_internal("%s", err_detail.data),
1290 : hint ? errhint("You might need to increase max_slot_wal_keep_size.") : 0);
1291 :
1292 20 : pfree(err_detail.data);
1293 20 : }
1294 :
1295 : /*
1296 : * Helper for InvalidateObsoleteReplicationSlots
1297 : *
1298 : * Acquires the given slot and mark it invalid, if necessary and possible.
1299 : *
1300 ECB : * Returns whether ReplicationSlotControlLock was released in the interim (and
1301 : * in that case we're not holding the lock at return, otherwise we are).
1302 : *
1303 : * Sets *invalidated true if the slot was invalidated. (Untouched otherwise.)
1304 : *
1305 : * This is inherently racy, because we release the LWLock
1306 : * for syscalls, so caller must restart if we return true.
1307 : */
1308 : static bool
1309 GNC 201 : InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
1310 : ReplicationSlot *s,
1311 : XLogRecPtr oldestLSN,
1312 : Oid dboid, TransactionId snapshotConflictHorizon,
1313 ECB : bool *invalidated)
1314 : {
1315 GIC 201 : int last_signaled_pid = 0;
1316 CBC 201 : bool released_lock = false;
1317 :
1318 ECB : for (;;)
1319 GIC 7 : {
1320 : XLogRecPtr restart_lsn;
1321 : NameData slotname;
1322 208 : int active_pid = 0;
1323 GNC 208 : ReplicationSlotInvalidationCause conflict = RS_INVAL_NONE;
1324 :
1325 CBC 208 : Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
1326 :
1327 GIC 208 : if (!s->in_use)
1328 : {
1329 UIC 0 : if (released_lock)
1330 0 : LWLockRelease(ReplicationSlotControlLock);
1331 0 : break;
1332 : }
1333 :
1334 ECB : /*
1335 : * Check if the slot needs to be invalidated. If it needs to be
1336 : * invalidated, and is not currently acquired, acquire it and mark it
1337 : * as having been invalidated. We do this with the spinlock held to
1338 : * avoid race conditions -- for example the restart_lsn could move
1339 : * forward, or the slot could be dropped.
1340 : */
1341 CBC 208 : SpinLockAcquire(&s->mutex);
1342 ECB :
1343 CBC 208 : restart_lsn = s->data.restart_lsn;
1344 ECB :
1345 : /*
1346 : * If the slot is already invalid or is a non conflicting slot, we
1347 : * don't need to do anything.
1348 : */
1349 GNC 208 : if (s->data.invalidated == RS_INVAL_NONE)
1350 : {
1351 174 : switch (cause)
1352 : {
1353 117 : case RS_INVAL_WAL_REMOVED:
1354 117 : if (s->data.restart_lsn != InvalidXLogRecPtr &&
1355 108 : s->data.restart_lsn < oldestLSN)
1356 5 : conflict = cause;
1357 117 : break;
1358 52 : case RS_INVAL_HORIZON:
1359 52 : if (!SlotIsLogical(s))
1360 22 : break;
1361 : /* invalid DB oid signals a shared relation */
1362 30 : if (dboid != InvalidOid && dboid != s->data.database)
1363 UNC 0 : break;
1364 GNC 30 : if (TransactionIdIsValid(s->effective_xmin) &&
1365 UNC 0 : TransactionIdPrecedesOrEquals(s->effective_xmin,
1366 : snapshotConflictHorizon))
1367 0 : conflict = cause;
1368 GNC 60 : else if (TransactionIdIsValid(s->effective_catalog_xmin) &&
1369 30 : TransactionIdPrecedesOrEquals(s->effective_catalog_xmin,
1370 : snapshotConflictHorizon))
1371 12 : conflict = cause;
1372 30 : break;
1373 5 : case RS_INVAL_WAL_LEVEL:
1374 5 : if (SlotIsLogical(s))
1375 3 : conflict = cause;
1376 5 : break;
1377 UNC 0 : case RS_INVAL_NONE:
1378 0 : pg_unreachable();
1379 : }
1380 : }
1381 :
1382 : /* if there's no conflict, we're done */
1383 GNC 208 : if (conflict == RS_INVAL_NONE)
1384 : {
1385 CBC 188 : SpinLockRelease(&s->mutex);
1386 188 : if (released_lock)
1387 LBC 0 : LWLockRelease(ReplicationSlotControlLock);
1388 GBC 188 : break;
1389 EUB : }
1390 :
1391 GIC 20 : slotname = s->data.name;
1392 CBC 20 : active_pid = s->active_pid;
1393 :
1394 : /*
1395 : * If the slot can be acquired, do so and mark it invalidated
1396 : * immediately. Otherwise we'll signal the owning process, below, and
1397 : * retry.
1398 : */
1399 GIC 20 : if (active_pid == 0)
1400 : {
1401 CBC 13 : MyReplicationSlot = s;
1402 13 : s->active_pid = MyProcPid;
1403 GNC 13 : s->data.invalidated = conflict;
1404 :
1405 : /*
1406 : * XXX: We should consider not overwriting restart_lsn and instead
1407 : * just rely on .invalidated.
1408 : */
1409 13 : if (conflict == RS_INVAL_WAL_REMOVED)
1410 3 : s->data.restart_lsn = InvalidXLogRecPtr;
1411 :
1412 : /* Let caller know */
1413 GIC 13 : *invalidated = true;
1414 : }
1415 :
1416 20 : SpinLockRelease(&s->mutex);
1417 :
1418 20 : if (active_pid != 0)
1419 : {
1420 : /*
1421 : * Prepare the sleep on the slot's condition variable before
1422 : * releasing the lock, to close a possible race condition if the
1423 : * slot is released before the sleep below.
1424 ECB : */
1425 GIC 7 : ConditionVariablePrepareToSleep(&s->active_cv);
1426 :
1427 7 : LWLockRelease(ReplicationSlotControlLock);
1428 7 : released_lock = true;
1429 :
1430 ECB : /*
1431 : * Signal to terminate the process that owns the slot, if we
1432 : * haven't already signalled it. (Avoidance of repeated
1433 : * signalling is the only reason for there to be a loop in this
1434 : * routine; otherwise we could rely on caller's restart loop.)
1435 : *
1436 : * There is the race condition that other process may own the slot
1437 : * after its current owner process is terminated and before this
1438 : * process owns it. To handle that, we signal only if the PID of
1439 : * the owning process has changed from the previous time. (This
1440 : * logic assumes that the same PID is not reused very quickly.)
1441 : */
1442 CBC 7 : if (last_signaled_pid != active_pid)
1443 : {
1444 GNC 7 : ReportSlotInvalidation(conflict, true, active_pid,
1445 : slotname, restart_lsn,
1446 : oldestLSN, snapshotConflictHorizon);
1447 :
1448 7 : if (MyBackendType == B_STARTUP)
1449 5 : (void) SendProcSignal(active_pid,
1450 : PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT,
1451 : InvalidBackendId);
1452 : else
1453 2 : (void) kill(active_pid, SIGTERM);
1454 :
1455 GIC 7 : last_signaled_pid = active_pid;
1456 : }
1457 :
1458 : /* Wait until the slot is released. */
1459 7 : ConditionVariableSleep(&s->active_cv,
1460 : WAIT_EVENT_REPLICATION_SLOT_DROP);
1461 :
1462 ECB : /*
1463 : * Re-acquire lock and start over; we expect to invalidate the
1464 : * slot next time (unless another process acquires the slot in the
1465 : * meantime).
1466 : */
1467 GIC 7 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1468 7 : continue;
1469 : }
1470 ECB : else
1471 : {
1472 : /*
1473 : * We hold the slot now and have already invalidated it; flush it
1474 : * to ensure that state persists.
1475 : *
1476 : * Don't want to hold ReplicationSlotControlLock across file
1477 : * system operations, so release it now but be sure to tell caller
1478 : * to restart from scratch.
1479 : */
1480 CBC 13 : LWLockRelease(ReplicationSlotControlLock);
1481 13 : released_lock = true;
1482 :
1483 ECB : /* Make sure the invalidated state persists across server restart */
1484 GBC 13 : ReplicationSlotMarkDirty();
1485 CBC 13 : ReplicationSlotSave();
1486 GBC 13 : ReplicationSlotRelease();
1487 GNC 13 : pgstat_drop_replslot(s);
1488 :
1489 13 : ReportSlotInvalidation(conflict, false, active_pid,
1490 : slotname, restart_lsn,
1491 : oldestLSN, snapshotConflictHorizon);
1492 ECB :
1493 : /* done with this slot for now */
1494 CBC 13 : break;
1495 ECB : }
1496 : }
1497 :
1498 GBC 201 : Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock));
1499 EUB :
1500 GIC 201 : return released_lock;
1501 : }
1502 :
1503 : /*
1504 : * Invalidate slots that require resources about to be removed.
1505 ECB : *
1506 : * Returns true when any slot have got invalidated.
1507 EUB : *
1508 : * Whether a slot needs to be invalidated depends on the cause. A slot is
1509 : * removed if it:
1510 : * - RS_INVAL_WAL_REMOVED: requires a LSN older than the given segment
1511 : * - RS_INVAL_HORIZON: requires a snapshot <= the given horizon in the given
1512 : * db; dboid may be InvalidOid for shared relations
1513 : * - RS_INVAL_WAL_LEVEL: is logical
1514 : *
1515 ECB : * NB - this runs as part of checkpoint, so avoid raising errors if possible.
1516 : */
1517 : bool
1518 GNC 2383 : InvalidateObsoleteReplicationSlots(ReplicationSlotInvalidationCause cause,
1519 : XLogSegNo oldestSegno, Oid dboid,
1520 : TransactionId snapshotConflictHorizon)
1521 ECB : {
1522 : XLogRecPtr oldestLSN;
1523 GIC 2383 : bool invalidated = false;
1524 :
1525 GNC 2383 : Assert(cause != RS_INVAL_HORIZON || TransactionIdIsValid(snapshotConflictHorizon));
1526 2383 : Assert(cause != RS_INVAL_WAL_REMOVED || oldestSegno > 0);
1527 2383 : Assert(cause != RS_INVAL_NONE);
1528 :
1529 2383 : if (max_replication_slots == 0)
1530 UNC 0 : return invalidated;
1531 :
1532 GIC 2383 : XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1533 :
1534 2396 : restart:
1535 CBC 2396 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1536 GIC 25961 : for (int i = 0; i < max_replication_slots; i++)
1537 ECB : {
1538 CBC 23578 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
1539 ECB :
1540 GIC 23578 : if (!s->in_use)
1541 23377 : continue;
1542 :
1543 GNC 201 : if (InvalidatePossiblyObsoleteSlot(cause, s, oldestLSN, dboid,
1544 : snapshotConflictHorizon,
1545 : &invalidated))
1546 : {
1547 ECB : /* if the lock was released, start from scratch */
1548 CBC 13 : goto restart;
1549 : }
1550 : }
1551 2383 : LWLockRelease(ReplicationSlotControlLock);
1552 :
1553 : /*
1554 ECB : * If any slots have been invalidated, recalculate the resource limits.
1555 : */
1556 CBC 2383 : if (invalidated)
1557 : {
1558 GIC 8 : ReplicationSlotsComputeRequiredXmin(false);
1559 8 : ReplicationSlotsComputeRequiredLSN();
1560 : }
1561 :
1562 2383 : return invalidated;
1563 ECB : }
1564 :
1565 : /*
1566 : * Flush all replication slots to disk.
1567 : *
1568 : * This needn't actually be part of a checkpoint, but it's a convenient
1569 : * location.
1570 : */
1571 : void
1572 GIC 2363 : CheckPointReplicationSlots(void)
1573 : {
1574 : int i;
1575 :
1576 2363 : elog(DEBUG1, "performing replication slot checkpoint");
1577 :
1578 : /*
1579 : * Prevent any slot from being created/dropped while we're active. As we
1580 ECB : * explicitly do *not* want to block iterating over replication_slots or
1581 : * acquiring a slot we cannot take the control lock - but that's OK,
1582 : * because holding ReplicationSlotAllocationLock is strictly stronger, and
1583 : * enough to guarantee that nobody can change the in_use bits on us.
1584 : */
1585 GIC 2363 : LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1586 ECB :
1587 CBC 25839 : for (i = 0; i < max_replication_slots; i++)
1588 : {
1589 GIC 23476 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
1590 : char path[MAXPGPATH];
1591 ECB :
1592 GIC 23476 : if (!s->in_use)
1593 CBC 23357 : continue;
1594 :
1595 : /* save the slot to disk, locking is handled in SaveSlotToPath() */
1596 GIC 119 : sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1597 CBC 119 : SaveSlotToPath(s, path, LOG);
1598 : }
1599 GIC 2363 : LWLockRelease(ReplicationSlotAllocationLock);
1600 2363 : }
1601 :
1602 : /*
1603 : * Load all replication slots from disk into memory at server startup. This
1604 : * needs to be run before we start crash recovery.
1605 ECB : */
1606 : void
1607 GIC 1176 : StartupReplicationSlots(void)
1608 : {
1609 : DIR *replication_dir;
1610 : struct dirent *replication_de;
1611 :
1612 1176 : elog(DEBUG1, "starting up replication slots");
1613 :
1614 : /* restore all slots by iterating over all on-disk entries */
1615 1176 : replication_dir = AllocateDir("pg_replslot");
1616 3562 : while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1617 : {
1618 ECB : char path[MAXPGPATH + 12];
1619 : PGFileType de_type;
1620 :
1621 GIC 2386 : if (strcmp(replication_de->d_name, ".") == 0 ||
1622 CBC 1210 : strcmp(replication_de->d_name, "..") == 0)
1623 2352 : continue;
1624 ECB :
1625 CBC 34 : snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1626 GNC 34 : de_type = get_dirent_type(path, replication_de, false, DEBUG1);
1627 :
1628 ECB : /* we're only creating directories here, skip if it's not our's */
1629 GNC 34 : if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_DIR)
1630 UIC 0 : continue;
1631 :
1632 : /* we crashed while a slot was being setup or deleted, clean up */
1633 CBC 34 : if (pg_str_endswith(replication_de->d_name, ".tmp"))
1634 : {
1635 UIC 0 : if (!rmtree(path, true))
1636 : {
1637 LBC 0 : ereport(WARNING,
1638 : (errmsg("could not remove directory \"%s\"",
1639 ECB : path)));
1640 UIC 0 : continue;
1641 : }
1642 0 : fsync_fname("pg_replslot", true);
1643 0 : continue;
1644 : }
1645 :
1646 : /* looks like a slot in a normal state, restore */
1647 GIC 34 : RestoreSlotFromDisk(replication_de->d_name);
1648 : }
1649 1176 : FreeDir(replication_dir);
1650 :
1651 : /* currently no slots exist, we're done. */
1652 1176 : if (max_replication_slots <= 0)
1653 UIC 0 : return;
1654 :
1655 : /* Now that we have recovered all the data, compute replication xmin */
1656 GIC 1176 : ReplicationSlotsComputeRequiredXmin(false);
1657 CBC 1176 : ReplicationSlotsComputeRequiredLSN();
1658 : }
1659 :
1660 : /* ----
1661 : * Manipulation of on-disk state of replication slots
1662 ECB : *
1663 : * NB: none of the routines below should take any notice whether a slot is the
1664 : * current one or not, that's all handled a layer above.
1665 : * ----
1666 : */
1667 : static void
1668 CBC 469 : CreateSlotOnDisk(ReplicationSlot *slot)
1669 EUB : {
1670 : char tmppath[MAXPGPATH];
1671 ECB : char path[MAXPGPATH];
1672 : struct stat st;
1673 :
1674 : /*
1675 : * No need to take out the io_in_progress_lock, nobody else can see this
1676 : * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
1677 : * takes out the lock, if we'd take the lock here, we'd deadlock.
1678 : */
1679 :
1680 CBC 469 : sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
1681 GIC 469 : sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
1682 ECB :
1683 : /*
1684 : * It's just barely possible that some previous effort to create or drop a
1685 : * slot with this name left a temp directory lying around. If that seems
1686 : * to be the case, try to remove it. If the rmtree() fails, we'll error
1687 : * out at the MakePGDirectory() below, so we don't bother checking
1688 : * success.
1689 : */
1690 CBC 469 : if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
1691 UIC 0 : rmtree(tmppath, true);
1692 :
1693 : /* Create and fsync the temporary slot directory. */
1694 GIC 469 : if (MakePGDirectory(tmppath) < 0)
1695 LBC 0 : ereport(ERROR,
1696 : (errcode_for_file_access(),
1697 ECB : errmsg("could not create directory \"%s\": %m",
1698 : tmppath)));
1699 GIC 469 : fsync_fname(tmppath, true);
1700 :
1701 ECB : /* Write the actual state file. */
1702 GIC 469 : slot->dirty = true; /* signal that we really need to write */
1703 469 : SaveSlotToPath(slot, tmppath, ERROR);
1704 :
1705 : /* Rename the directory into place. */
1706 469 : if (rename(tmppath, path) != 0)
1707 UIC 0 : ereport(ERROR,
1708 : (errcode_for_file_access(),
1709 : errmsg("could not rename file \"%s\" to \"%s\": %m",
1710 : tmppath, path)));
1711 ECB :
1712 : /*
1713 : * If we'd now fail - really unlikely - we wouldn't know whether this slot
1714 : * would persist after an OS crash or not - so, force a restart. The
1715 : * restart would try to fsync this again till it works.
1716 : */
1717 GIC 469 : START_CRIT_SECTION();
1718 :
1719 469 : fsync_fname(path, true);
1720 469 : fsync_fname("pg_replslot", true);
1721 :
1722 469 : END_CRIT_SECTION();
1723 469 : }
1724 ECB :
1725 : /*
1726 : * Shared functionality between saving and creating a replication slot.
1727 : */
1728 : static void
1729 GIC 1559 : SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
1730 : {
1731 ECB : char tmppath[MAXPGPATH];
1732 : char path[MAXPGPATH];
1733 : int fd;
1734 : ReplicationSlotOnDisk cp;
1735 : bool was_dirty;
1736 :
1737 : /* first check whether there's something to write out */
1738 CBC 1559 : SpinLockAcquire(&slot->mutex);
1739 1559 : was_dirty = slot->dirty;
1740 GIC 1559 : slot->just_dirtied = false;
1741 1559 : SpinLockRelease(&slot->mutex);
1742 :
1743 : /* and don't do anything if there's nothing to write */
1744 1559 : if (!was_dirty)
1745 79 : return;
1746 ECB :
1747 GIC 1480 : LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE);
1748 :
1749 : /* silence valgrind :( */
1750 1480 : memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
1751 ECB :
1752 GIC 1480 : sprintf(tmppath, "%s/state.tmp", dir);
1753 1480 : sprintf(path, "%s/state", dir);
1754 ECB :
1755 CBC 1480 : fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1756 GIC 1480 : if (fd < 0)
1757 : {
1758 : /*
1759 : * If not an ERROR, then release the lock before returning. In case
1760 ECB : * of an ERROR, the error recovery path automatically releases the
1761 : * lock, but no harm in explicitly releasing even in that case. Note
1762 : * that LWLockRelease() could affect errno.
1763 : */
1764 LBC 0 : int save_errno = errno;
1765 ECB :
1766 UIC 0 : LWLockRelease(&slot->io_in_progress_lock);
1767 0 : errno = save_errno;
1768 LBC 0 : ereport(elevel,
1769 EUB : (errcode_for_file_access(),
1770 : errmsg("could not create file \"%s\": %m",
1771 : tmppath)));
1772 LBC 0 : return;
1773 : }
1774 EUB :
1775 GIC 1480 : cp.magic = SLOT_MAGIC;
1776 GBC 1480 : INIT_CRC32C(cp.checksum);
1777 GIC 1480 : cp.version = SLOT_VERSION;
1778 1480 : cp.length = ReplicationSlotOnDiskV2Size;
1779 EUB :
1780 GIC 1480 : SpinLockAcquire(&slot->mutex);
1781 EUB :
1782 GBC 1480 : memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
1783 :
1784 GIC 1480 : SpinLockRelease(&slot->mutex);
1785 :
1786 CBC 1480 : COMP_CRC32C(cp.checksum,
1787 : (char *) (&cp) + ReplicationSlotOnDiskNotChecksummedSize,
1788 ECB : ReplicationSlotOnDiskChecksummedSize);
1789 GIC 1480 : FIN_CRC32C(cp.checksum);
1790 :
1791 CBC 1480 : errno = 0;
1792 GBC 1480 : pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_WRITE);
1793 GIC 1480 : if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
1794 : {
1795 LBC 0 : int save_errno = errno;
1796 ECB :
1797 UIC 0 : pgstat_report_wait_end();
1798 0 : CloseTransientFile(fd);
1799 0 : LWLockRelease(&slot->io_in_progress_lock);
1800 :
1801 : /* if write didn't set errno, assume problem is no disk space */
1802 0 : errno = save_errno ? save_errno : ENOSPC;
1803 0 : ereport(elevel,
1804 : (errcode_for_file_access(),
1805 : errmsg("could not write to file \"%s\": %m",
1806 : tmppath)));
1807 LBC 0 : return;
1808 : }
1809 GIC 1480 : pgstat_report_wait_end();
1810 :
1811 : /* fsync the temporary file */
1812 1480 : pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_SYNC);
1813 1480 : if (pg_fsync(fd) != 0)
1814 : {
1815 UIC 0 : int save_errno = errno;
1816 :
1817 0 : pgstat_report_wait_end();
1818 0 : CloseTransientFile(fd);
1819 LBC 0 : LWLockRelease(&slot->io_in_progress_lock);
1820 0 : errno = save_errno;
1821 UIC 0 : ereport(elevel,
1822 : (errcode_for_file_access(),
1823 : errmsg("could not fsync file \"%s\": %m",
1824 : tmppath)));
1825 0 : return;
1826 : }
1827 GIC 1480 : pgstat_report_wait_end();
1828 :
1829 CBC 1480 : if (CloseTransientFile(fd) != 0)
1830 EUB : {
1831 UIC 0 : int save_errno = errno;
1832 :
1833 LBC 0 : LWLockRelease(&slot->io_in_progress_lock);
1834 UBC 0 : errno = save_errno;
1835 UIC 0 : ereport(elevel,
1836 : (errcode_for_file_access(),
1837 : errmsg("could not close file \"%s\": %m",
1838 ECB : tmppath)));
1839 UIC 0 : return;
1840 : }
1841 ECB :
1842 : /* rename to permanent file, fsync file and directory */
1843 GIC 1480 : if (rename(tmppath, path) != 0)
1844 : {
1845 LBC 0 : int save_errno = errno;
1846 EUB :
1847 UIC 0 : LWLockRelease(&slot->io_in_progress_lock);
1848 0 : errno = save_errno;
1849 0 : ereport(elevel,
1850 : (errcode_for_file_access(),
1851 : errmsg("could not rename file \"%s\" to \"%s\": %m",
1852 : tmppath, path)));
1853 0 : return;
1854 : }
1855 :
1856 ECB : /*
1857 : * Check CreateSlotOnDisk() for the reasoning of using a critical section.
1858 : */
1859 CBC 1480 : START_CRIT_SECTION();
1860 :
1861 1480 : fsync_fname(path, false);
1862 1480 : fsync_fname(dir, true);
1863 GIC 1480 : fsync_fname("pg_replslot", true);
1864 :
1865 1480 : END_CRIT_SECTION();
1866 :
1867 : /*
1868 ECB : * Successfully wrote, unset dirty bit, unless somebody dirtied again
1869 : * already.
1870 : */
1871 GIC 1480 : SpinLockAcquire(&slot->mutex);
1872 1480 : if (!slot->just_dirtied)
1873 1479 : slot->dirty = false;
1874 1480 : SpinLockRelease(&slot->mutex);
1875 :
1876 1480 : LWLockRelease(&slot->io_in_progress_lock);
1877 ECB : }
1878 :
1879 : /*
1880 : * Load a single slot from disk into memory.
1881 : */
1882 : static void
1883 CBC 34 : RestoreSlotFromDisk(const char *name)
1884 ECB : {
1885 : ReplicationSlotOnDisk cp;
1886 : int i;
1887 : char slotdir[MAXPGPATH + 12];
1888 : char path[MAXPGPATH + 22];
1889 : int fd;
1890 GIC 34 : bool restored = false;
1891 ECB : int readBytes;
1892 : pg_crc32c checksum;
1893 :
1894 : /* no need to lock here, no concurrent access allowed yet */
1895 :
1896 : /* delete temp file if it exists */
1897 GIC 34 : sprintf(slotdir, "pg_replslot/%s", name);
1898 34 : sprintf(path, "%s/state.tmp", slotdir);
1899 34 : if (unlink(path) < 0 && errno != ENOENT)
1900 UIC 0 : ereport(PANIC,
1901 : (errcode_for_file_access(),
1902 : errmsg("could not remove file \"%s\": %m", path)));
1903 EUB :
1904 GIC 34 : sprintf(path, "%s/state", slotdir);
1905 EUB :
1906 GBC 34 : elog(DEBUG1, "restoring replication slot from \"%s\"", path);
1907 EUB :
1908 : /* on some operating systems fsyncing a file requires O_RDWR */
1909 GIC 34 : fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
1910 :
1911 EUB : /*
1912 : * We do not need to handle this as we are rename()ing the directory into
1913 : * place only after we fsync()ed the state file.
1914 ECB : */
1915 CBC 34 : if (fd < 0)
1916 LBC 0 : ereport(PANIC,
1917 ECB : (errcode_for_file_access(),
1918 : errmsg("could not open file \"%s\": %m", path)));
1919 :
1920 : /*
1921 : * Sync state file before we're reading from it. We might have crashed
1922 : * while it wasn't synced yet and we shouldn't continue on that basis.
1923 : */
1924 GIC 34 : pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC);
1925 CBC 34 : if (pg_fsync(fd) != 0)
1926 UIC 0 : ereport(PANIC,
1927 : (errcode_for_file_access(),
1928 ECB : errmsg("could not fsync file \"%s\": %m",
1929 : path)));
1930 CBC 34 : pgstat_report_wait_end();
1931 ECB :
1932 : /* Also sync the parent directory */
1933 GIC 34 : START_CRIT_SECTION();
1934 GBC 34 : fsync_fname(slotdir, true);
1935 GIC 34 : END_CRIT_SECTION();
1936 EUB :
1937 : /* read part of statefile that's guaranteed to be version independent */
1938 GBC 34 : pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
1939 GIC 34 : readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
1940 34 : pgstat_report_wait_end();
1941 GBC 34 : if (readBytes != ReplicationSlotOnDiskConstantSize)
1942 EUB : {
1943 UIC 0 : if (readBytes < 0)
1944 0 : ereport(PANIC,
1945 : (errcode_for_file_access(),
1946 EUB : errmsg("could not read file \"%s\": %m", path)));
1947 : else
1948 LBC 0 : ereport(PANIC,
1949 : (errcode(ERRCODE_DATA_CORRUPTED),
1950 : errmsg("could not read file \"%s\": read %d of %zu",
1951 ECB : path, readBytes,
1952 : (Size) ReplicationSlotOnDiskConstantSize)));
1953 : }
1954 EUB :
1955 : /* verify magic */
1956 GBC 34 : if (cp.magic != SLOT_MAGIC)
1957 UBC 0 : ereport(PANIC,
1958 EUB : (errcode(ERRCODE_DATA_CORRUPTED),
1959 : errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
1960 : path, cp.magic, SLOT_MAGIC)));
1961 :
1962 : /* verify version */
1963 GIC 34 : if (cp.version != SLOT_VERSION)
1964 UBC 0 : ereport(PANIC,
1965 : (errcode(ERRCODE_DATA_CORRUPTED),
1966 ECB : errmsg("replication slot file \"%s\" has unsupported version %u",
1967 : path, cp.version)));
1968 :
1969 : /* boundary check on length */
1970 GBC 34 : if (cp.length != ReplicationSlotOnDiskV2Size)
1971 UIC 0 : ereport(PANIC,
1972 EUB : (errcode(ERRCODE_DATA_CORRUPTED),
1973 : errmsg("replication slot file \"%s\" has corrupted length %u",
1974 : path, cp.length)));
1975 :
1976 : /* Now that we know the size, read the entire file */
1977 GIC 34 : pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
1978 GBC 68 : readBytes = read(fd,
1979 : (char *) &cp + ReplicationSlotOnDiskConstantSize,
1980 GIC 34 : cp.length);
1981 34 : pgstat_report_wait_end();
1982 CBC 34 : if (readBytes != cp.length)
1983 : {
1984 UBC 0 : if (readBytes < 0)
1985 UIC 0 : ereport(PANIC,
1986 EUB : (errcode_for_file_access(),
1987 : errmsg("could not read file \"%s\": %m", path)));
1988 : else
1989 UIC 0 : ereport(PANIC,
1990 : (errcode(ERRCODE_DATA_CORRUPTED),
1991 : errmsg("could not read file \"%s\": read %d of %zu",
1992 EUB : path, readBytes, (Size) cp.length)));
1993 : }
1994 :
1995 GIC 34 : if (CloseTransientFile(fd) != 0)
1996 UIC 0 : ereport(PANIC,
1997 : (errcode_for_file_access(),
1998 ECB : errmsg("could not close file \"%s\": %m", path)));
1999 :
2000 : /* now verify the CRC */
2001 CBC 34 : INIT_CRC32C(checksum);
2002 34 : COMP_CRC32C(checksum,
2003 : (char *) &cp + ReplicationSlotOnDiskNotChecksummedSize,
2004 ECB : ReplicationSlotOnDiskChecksummedSize);
2005 GIC 34 : FIN_CRC32C(checksum);
2006 :
2007 34 : if (!EQ_CRC32C(checksum, cp.checksum))
2008 UIC 0 : ereport(PANIC,
2009 : (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
2010 ECB : path, checksum, cp.checksum)));
2011 :
2012 : /*
2013 : * If we crashed with an ephemeral slot active, don't restore but delete
2014 : * it.
2015 : */
2016 GIC 34 : if (cp.slotdata.persistency != RS_PERSISTENT)
2017 : {
2018 UIC 0 : if (!rmtree(slotdir, true))
2019 : {
2020 0 : ereport(WARNING,
2021 : (errmsg("could not remove directory \"%s\"",
2022 ECB : slotdir)));
2023 : }
2024 UIC 0 : fsync_fname("pg_replslot", true);
2025 0 : return;
2026 : }
2027 :
2028 : /*
2029 ECB : * Verify that requirements for the specific slot type are met. That's
2030 : * important because if these aren't met we're not guaranteed to retain
2031 : * all the necessary resources for the slot.
2032 : *
2033 : * NB: We have to do so *after* the above checks for ephemeral slots,
2034 : * because otherwise a slot that shouldn't exist anymore could prevent
2035 : * restarts.
2036 : *
2037 : * NB: Changing the requirements here also requires adapting
2038 : * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
2039 EUB : */
2040 GIC 34 : if (cp.slotdata.database != InvalidOid && wal_level < WAL_LEVEL_LOGICAL)
2041 UIC 0 : ereport(FATAL,
2042 : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2043 ECB : errmsg("logical replication slot \"%s\" exists, but wal_level < logical",
2044 : NameStr(cp.slotdata.name)),
2045 : errhint("Change wal_level to be logical or higher.")));
2046 GIC 34 : else if (wal_level < WAL_LEVEL_REPLICA)
2047 UIC 0 : ereport(FATAL,
2048 ECB : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2049 : errmsg("physical replication slot \"%s\" exists, but wal_level < replica",
2050 : NameStr(cp.slotdata.name)),
2051 : errhint("Change wal_level to be replica or higher.")));
2052 :
2053 : /* nothing can be active yet, don't lock anything */
2054 CBC 50 : for (i = 0; i < max_replication_slots; i++)
2055 EUB : {
2056 : ReplicationSlot *slot;
2057 :
2058 GIC 50 : slot = &ReplicationSlotCtl->replication_slots[i];
2059 :
2060 50 : if (slot->in_use)
2061 16 : continue;
2062 :
2063 ECB : /* restore the entire set of persistent data */
2064 CBC 34 : memcpy(&slot->data, &cp.slotdata,
2065 EUB : sizeof(ReplicationSlotPersistentData));
2066 :
2067 : /* initialize in memory state */
2068 GIC 34 : slot->effective_xmin = cp.slotdata.xmin;
2069 CBC 34 : slot->effective_catalog_xmin = cp.slotdata.catalog_xmin;
2070 :
2071 GIC 34 : slot->candidate_catalog_xmin = InvalidTransactionId;
2072 CBC 34 : slot->candidate_xmin_lsn = InvalidXLogRecPtr;
2073 34 : slot->candidate_restart_lsn = InvalidXLogRecPtr;
2074 34 : slot->candidate_restart_valid = InvalidXLogRecPtr;
2075 :
2076 GIC 34 : slot->in_use = true;
2077 CBC 34 : slot->active_pid = 0;
2078 ECB :
2079 CBC 34 : restored = true;
2080 34 : break;
2081 : }
2082 EUB :
2083 GBC 34 : if (!restored)
2084 UIC 0 : ereport(FATAL,
2085 : (errmsg("too many replication slots active before shutdown"),
2086 : errhint("Increase max_replication_slots and try again.")));
2087 EUB : }
|
