PostgreSQL Source Code git master
slot.c
Go to the documentation of this file.
1/*-------------------------------------------------------------------------
2 *
3 * slot.c
4 * Replication slot management.
5 *
6 *
7 * Copyright (c) 2012-2025, 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 directory
24 * $PGDATA / PG_REPLSLOT_DIR. Inside that directory the state file will
25 * contain the slot's own data. Additional data can be stored alongside that
26 * file if required. While the server is running, the state data is also
27 * cached in memory for 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"
44#include "access/xlogrecovery.h"
45#include "common/file_utils.h"
46#include "common/string.h"
47#include "miscadmin.h"
48#include "pgstat.h"
51#include "replication/slot.h"
53#include "storage/fd.h"
54#include "storage/ipc.h"
55#include "storage/proc.h"
56#include "storage/procarray.h"
57#include "utils/builtins.h"
58#include "utils/guc_hooks.h"
60#include "utils/varlena.h"
61
62/*
63 * Replication slot on-disk data structure.
64 */
66{
67 /* first part of this struct needs to be version independent */
68
69 /* data not covered by checksum */
72
73 /* data covered by checksum */
76
77 /*
78 * The actual data in the slot that follows can differ based on the above
79 * 'version'.
80 */
81
84
85/*
86 * Struct for the configuration of synchronized_standby_slots.
87 *
88 * Note: this must be a flat representation that can be held in a single chunk
89 * of guc_malloc'd memory, so that it can be stored as the "extra" data for the
90 * synchronized_standby_slots GUC.
91 */
92typedef struct
93{
94 /* Number of slot names in the slot_names[] */
96
97 /*
98 * slot_names contains 'nslotnames' consecutive null-terminated C strings.
99 */
100 char slot_names[FLEXIBLE_ARRAY_MEMBER];
102
103/*
104 * Lookup table for slot invalidation causes.
105 */
107{
109 const char *cause_name;
111
113 {RS_INVAL_NONE, "none"},
114 {RS_INVAL_WAL_REMOVED, "wal_removed"},
115 {RS_INVAL_HORIZON, "rows_removed"},
116 {RS_INVAL_WAL_LEVEL, "wal_level_insufficient"},
117 {RS_INVAL_IDLE_TIMEOUT, "idle_timeout"},
118};
119
120/*
121 * Ensure that the lookup table is up-to-date with the enums defined in
122 * ReplicationSlotInvalidationCause.
123 */
125 "array length mismatch");
126
127/* size of version independent data */
128#define ReplicationSlotOnDiskConstantSize \
129 offsetof(ReplicationSlotOnDisk, slotdata)
130/* size of the part of the slot not covered by the checksum */
131#define ReplicationSlotOnDiskNotChecksummedSize \
132 offsetof(ReplicationSlotOnDisk, version)
133/* size of the part covered by the checksum */
134#define ReplicationSlotOnDiskChecksummedSize \
135 sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize
136/* size of the slot data that is version dependent */
137#define ReplicationSlotOnDiskV2Size \
138 sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize
139
140#define SLOT_MAGIC 0x1051CA1 /* format identifier */
141#define SLOT_VERSION 5 /* version for new files */
142
143/* Control array for replication slot management */
145
146/* My backend's replication slot in the shared memory array */
148
149/* GUC variables */
150int max_replication_slots = 10; /* the maximum number of replication
151 * slots */
152
153/*
154 * Invalidate replication slots that have remained idle longer than this
155 * duration; '0' disables it.
156 */
158
159/*
160 * This GUC lists streaming replication standby server slot names that
161 * logical WAL sender processes will wait for.
162 */
164
165/* This is the parsed and cached configuration for synchronized_standby_slots */
167
168/*
169 * Oldest LSN that has been confirmed to be flushed to the standbys
170 * corresponding to the physical slots specified in the synchronized_standby_slots GUC.
171 */
173
174static void ReplicationSlotShmemExit(int code, Datum arg);
175static void ReplicationSlotDropPtr(ReplicationSlot *slot);
176
177/* internal persistency functions */
178static void RestoreSlotFromDisk(const char *name);
179static void CreateSlotOnDisk(ReplicationSlot *slot);
180static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel);
181
182/*
183 * Report shared-memory space needed by ReplicationSlotsShmemInit.
184 */
185Size
187{
188 Size size = 0;
189
190 if (max_replication_slots == 0)
191 return size;
192
193 size = offsetof(ReplicationSlotCtlData, replication_slots);
194 size = add_size(size,
196
197 return size;
198}
199
200/*
201 * Allocate and initialize shared memory for replication slots.
202 */
203void
205{
206 bool found;
207
208 if (max_replication_slots == 0)
209 return;
210
212 ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
213 &found);
214
215 if (!found)
216 {
217 int i;
218
219 /* First time through, so initialize */
221
222 for (i = 0; i < max_replication_slots; i++)
223 {
225
226 /* everything else is zeroed by the memset above */
227 SpinLockInit(&slot->mutex);
231 }
232 }
233}
234
235/*
236 * Register the callback for replication slot cleanup and releasing.
237 */
238void
240{
242}
243
244/*
245 * Release and cleanup replication slots.
246 */
247static void
249{
250 /* Make sure active replication slots are released */
251 if (MyReplicationSlot != NULL)
253
254 /* Also cleanup all the temporary slots. */
256}
257
258/*
259 * Check whether the passed slot name is valid and report errors at elevel.
260 *
261 * Slot names may consist out of [a-z0-9_]{1,NAMEDATALEN-1} which should allow
262 * the name to be used as a directory name on every supported OS.
263 *
264 * Returns whether the directory name is valid or not if elevel < ERROR.
265 */
266bool
267ReplicationSlotValidateName(const char *name, int elevel)
268{
269 const char *cp;
270
271 if (strlen(name) == 0)
272 {
273 ereport(elevel,
274 (errcode(ERRCODE_INVALID_NAME),
275 errmsg("replication slot name \"%s\" is too short",
276 name)));
277 return false;
278 }
279
280 if (strlen(name) >= NAMEDATALEN)
281 {
282 ereport(elevel,
283 (errcode(ERRCODE_NAME_TOO_LONG),
284 errmsg("replication slot name \"%s\" is too long",
285 name)));
286 return false;
287 }
288
289 for (cp = name; *cp; cp++)
290 {
291 if (!((*cp >= 'a' && *cp <= 'z')
292 || (*cp >= '0' && *cp <= '9')
293 || (*cp == '_')))
294 {
295 ereport(elevel,
296 (errcode(ERRCODE_INVALID_NAME),
297 errmsg("replication slot name \"%s\" contains invalid character",
298 name),
299 errhint("Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
300 return false;
301 }
302 }
303 return true;
304}
305
306/*
307 * Create a new replication slot and mark it as used by this backend.
308 *
309 * name: Name of the slot
310 * db_specific: logical decoding is db specific; if the slot is going to
311 * be used for that pass true, otherwise false.
312 * two_phase: Allows decoding of prepared transactions. We allow this option
313 * to be enabled only at the slot creation time. If we allow this option
314 * to be changed during decoding then it is quite possible that we skip
315 * prepare first time because this option was not enabled. Now next time
316 * during getting changes, if the two_phase option is enabled it can skip
317 * prepare because by that time start decoding point has been moved. So the
318 * user will only get commit prepared.
319 * failover: If enabled, allows the slot to be synced to standbys so
320 * that logical replication can be resumed after failover.
321 * synced: True if the slot is synchronized from the primary server.
322 */
323void
324ReplicationSlotCreate(const char *name, bool db_specific,
325 ReplicationSlotPersistency persistency,
326 bool two_phase, bool failover, bool synced)
327{
328 ReplicationSlot *slot = NULL;
329 int i;
330
331 Assert(MyReplicationSlot == NULL);
332
334
335 if (failover)
336 {
337 /*
338 * Do not allow users to create the failover enabled slots on the
339 * standby as we do not support sync to the cascading standby.
340 *
341 * However, failover enabled slots can be created during slot
342 * synchronization because we need to retain the same values as the
343 * remote slot.
344 */
347 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
348 errmsg("cannot enable failover for a replication slot created on the standby"));
349
350 /*
351 * Do not allow users to create failover enabled temporary slots,
352 * because temporary slots will not be synced to the standby.
353 *
354 * However, failover enabled temporary slots can be created during
355 * slot synchronization. See the comments atop slotsync.c for details.
356 */
357 if (persistency == RS_TEMPORARY && !IsSyncingReplicationSlots())
359 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
360 errmsg("cannot enable failover for a temporary replication slot"));
361 }
362
363 /*
364 * If some other backend ran this code concurrently with us, we'd likely
365 * both allocate the same slot, and that would be bad. We'd also be at
366 * risk of missing a name collision. Also, we don't want to try to create
367 * a new slot while somebody's busy cleaning up an old one, because we
368 * might both be monkeying with the same directory.
369 */
370 LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
371
372 /*
373 * Check for name collision, and identify an allocatable slot. We need to
374 * hold ReplicationSlotControlLock in shared mode for this, so that nobody
375 * else can change the in_use flags while we're looking at them.
376 */
377 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
378 for (i = 0; i < max_replication_slots; i++)
379 {
381
382 if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
385 errmsg("replication slot \"%s\" already exists", name)));
386 if (!s->in_use && slot == NULL)
387 slot = s;
388 }
389 LWLockRelease(ReplicationSlotControlLock);
390
391 /* If all slots are in use, we're out of luck. */
392 if (slot == NULL)
394 (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
395 errmsg("all replication slots are in use"),
396 errhint("Free one or increase \"max_replication_slots\".")));
397
398 /*
399 * Since this slot is not in use, nobody should be looking at any part of
400 * it other than the in_use field unless they're trying to allocate it.
401 * And since we hold ReplicationSlotAllocationLock, nobody except us can
402 * be doing that. So it's safe to initialize the slot.
403 */
404 Assert(!slot->in_use);
405 Assert(slot->active_pid == 0);
406
407 /* first initialize persistent data */
408 memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
409 namestrcpy(&slot->data.name, name);
410 slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
411 slot->data.persistency = persistency;
412 slot->data.two_phase = two_phase;
414 slot->data.failover = failover;
415 slot->data.synced = synced;
416
417 /* and then data only present in shared memory */
418 slot->just_dirtied = false;
419 slot->dirty = false;
427 slot->inactive_since = 0;
428
429 /*
430 * Create the slot on disk. We haven't actually marked the slot allocated
431 * yet, so no special cleanup is required if this errors out.
432 */
433 CreateSlotOnDisk(slot);
434
435 /*
436 * We need to briefly prevent any other backend from iterating over the
437 * slots while we flip the in_use flag. We also need to set the active
438 * flag while holding the ControlLock as otherwise a concurrent
439 * ReplicationSlotAcquire() could acquire the slot as well.
440 */
441 LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
442
443 slot->in_use = true;
444
445 /* We can now mark the slot active, and that makes it our slot. */
446 SpinLockAcquire(&slot->mutex);
447 Assert(slot->active_pid == 0);
448 slot->active_pid = MyProcPid;
449 SpinLockRelease(&slot->mutex);
450 MyReplicationSlot = slot;
451
452 LWLockRelease(ReplicationSlotControlLock);
453
454 /*
455 * Create statistics entry for the new logical slot. We don't collect any
456 * stats for physical slots, so no need to create an entry for the same.
457 * See ReplicationSlotDropPtr for why we need to do this before releasing
458 * ReplicationSlotAllocationLock.
459 */
460 if (SlotIsLogical(slot))
462
463 /*
464 * Now that the slot has been marked as in_use and active, it's safe to
465 * let somebody else try to allocate a slot.
466 */
467 LWLockRelease(ReplicationSlotAllocationLock);
468
469 /* Let everybody know we've modified this slot */
471}
472
473/*
474 * Search for the named replication slot.
475 *
476 * Return the replication slot if found, otherwise NULL.
477 */
479SearchNamedReplicationSlot(const char *name, bool need_lock)
480{
481 int i;
482 ReplicationSlot *slot = NULL;
483
484 if (need_lock)
485 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
486
487 for (i = 0; i < max_replication_slots; i++)
488 {
490
491 if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
492 {
493 slot = s;
494 break;
495 }
496 }
497
498 if (need_lock)
499 LWLockRelease(ReplicationSlotControlLock);
500
501 return slot;
502}
503
504/*
505 * Return the index of the replication slot in
506 * ReplicationSlotCtl->replication_slots.
507 *
508 * This is mainly useful to have an efficient key for storing replication slot
509 * stats.
510 */
511int
513{
515 slot < ReplicationSlotCtl->replication_slots + max_replication_slots);
516
518}
519
520/*
521 * If the slot at 'index' is unused, return false. Otherwise 'name' is set to
522 * the slot's name and true is returned.
523 *
524 * This likely is only useful for pgstat_replslot.c during shutdown, in other
525 * cases there are obvious TOCTOU issues.
526 */
527bool
529{
530 ReplicationSlot *slot;
531 bool found;
532
534
535 /*
536 * Ensure that the slot cannot be dropped while we copy the name. Don't
537 * need the spinlock as the name of an existing slot cannot change.
538 */
539 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
540 found = slot->in_use;
541 if (slot->in_use)
543 LWLockRelease(ReplicationSlotControlLock);
544
545 return found;
546}
547
548/*
549 * Find a previously created slot and mark it as used by this process.
550 *
551 * An error is raised if nowait is true and the slot is currently in use. If
552 * nowait is false, we sleep until the slot is released by the owning process.
553 *
554 * An error is raised if error_if_invalid is true and the slot is found to
555 * be invalid. It should always be set to true, except when we are temporarily
556 * acquiring the slot and don't intend to change it.
557 */
558void
559ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid)
560{
562 int active_pid;
563
564 Assert(name != NULL);
565
566retry:
567 Assert(MyReplicationSlot == NULL);
568
569 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
570
571 /* Check if the slot exits with the given name. */
573 if (s == NULL || !s->in_use)
574 {
575 LWLockRelease(ReplicationSlotControlLock);
576
578 (errcode(ERRCODE_UNDEFINED_OBJECT),
579 errmsg("replication slot \"%s\" does not exist",
580 name)));
581 }
582
583 /*
584 * This is the slot we want; check if it's active under some other
585 * process. In single user mode, we don't need this check.
586 */
588 {
589 /*
590 * Get ready to sleep on the slot in case it is active. (We may end
591 * up not sleeping, but we don't want to do this while holding the
592 * spinlock.)
593 */
594 if (!nowait)
596
597 /*
598 * It is important to reset the inactive_since under spinlock here to
599 * avoid race conditions with slot invalidation. See comments related
600 * to inactive_since in InvalidatePossiblyObsoleteSlot.
601 */
603 if (s->active_pid == 0)
605 active_pid = s->active_pid;
608 }
609 else
610 {
611 active_pid = MyProcPid;
613 }
614 LWLockRelease(ReplicationSlotControlLock);
615
616 /*
617 * If we found the slot but it's already active in another process, we
618 * wait until the owning process signals us that it's been released, or
619 * error out.
620 */
621 if (active_pid != MyProcPid)
622 {
623 if (!nowait)
624 {
625 /* Wait here until we get signaled, and then restart */
627 WAIT_EVENT_REPLICATION_SLOT_DROP);
629 goto retry;
630 }
631
633 (errcode(ERRCODE_OBJECT_IN_USE),
634 errmsg("replication slot \"%s\" is active for PID %d",
635 NameStr(s->data.name), active_pid)));
636 }
637 else if (!nowait)
638 ConditionVariableCancelSleep(); /* no sleep needed after all */
639
640 /* We made this slot active, so it's ours now. */
642
643 /*
644 * We need to check for invalidation after making the slot ours to avoid
645 * the possible race condition with the checkpointer that can otherwise
646 * invalidate the slot immediately after the check.
647 */
648 if (error_if_invalid && s->data.invalidated != RS_INVAL_NONE)
650 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
651 errmsg("can no longer access replication slot \"%s\"",
652 NameStr(s->data.name)),
653 errdetail("This replication slot has been invalidated due to \"%s\".",
655
656 /* Let everybody know we've modified this slot */
658
659 /*
660 * The call to pgstat_acquire_replslot() protects against stats for a
661 * different slot, from before a restart or such, being present during
662 * pgstat_report_replslot().
663 */
664 if (SlotIsLogical(s))
666
667
668 if (am_walsender)
669 {
672 ? errmsg("acquired logical replication slot \"%s\"",
673 NameStr(s->data.name))
674 : errmsg("acquired physical replication slot \"%s\"",
675 NameStr(s->data.name)));
676 }
677}
678
679/*
680 * Release the replication slot that this backend considers to own.
681 *
682 * This or another backend can re-acquire the slot later.
683 * Resources this slot requires will be preserved.
684 */
685void
687{
689 char *slotname = NULL; /* keep compiler quiet */
690 bool is_logical = false; /* keep compiler quiet */
691 TimestampTz now = 0;
692
693 Assert(slot != NULL && slot->active_pid != 0);
694
695 if (am_walsender)
696 {
697 slotname = pstrdup(NameStr(slot->data.name));
698 is_logical = SlotIsLogical(slot);
699 }
700
701 if (slot->data.persistency == RS_EPHEMERAL)
702 {
703 /*
704 * Delete the slot. There is no !PANIC case where this is allowed to
705 * fail, all that may happen is an incomplete cleanup of the on-disk
706 * data.
707 */
709 }
710
711 /*
712 * If slot needed to temporarily restrain both data and catalog xmin to
713 * create the catalog snapshot, remove that temporary constraint.
714 * Snapshots can only be exported while the initial snapshot is still
715 * acquired.
716 */
717 if (!TransactionIdIsValid(slot->data.xmin) &&
719 {
720 SpinLockAcquire(&slot->mutex);
722 SpinLockRelease(&slot->mutex);
724 }
725
726 /*
727 * Set the time since the slot has become inactive. We get the current
728 * time beforehand to avoid system call while holding the spinlock.
729 */
731
732 if (slot->data.persistency == RS_PERSISTENT)
733 {
734 /*
735 * Mark persistent slot inactive. We're not freeing it, just
736 * disconnecting, but wake up others that may be waiting for it.
737 */
738 SpinLockAcquire(&slot->mutex);
739 slot->active_pid = 0;
741 SpinLockRelease(&slot->mutex);
743 }
744 else
746
747 MyReplicationSlot = NULL;
748
749 /* might not have been set when we've been a plain slot */
750 LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
751 MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING;
753 LWLockRelease(ProcArrayLock);
754
755 if (am_walsender)
756 {
758 is_logical
759 ? errmsg("released logical replication slot \"%s\"",
760 slotname)
761 : errmsg("released physical replication slot \"%s\"",
762 slotname));
763
764 pfree(slotname);
765 }
766}
767
768/*
769 * Cleanup temporary slots created in current session.
770 *
771 * Cleanup only synced temporary slots if 'synced_only' is true, else
772 * cleanup all temporary slots.
773 */
774void
775ReplicationSlotCleanup(bool synced_only)
776{
777 int i;
778
779 Assert(MyReplicationSlot == NULL);
780
781restart:
782 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
783 for (i = 0; i < max_replication_slots; i++)
784 {
786
787 if (!s->in_use)
788 continue;
789
791 if ((s->active_pid == MyProcPid &&
792 (!synced_only || s->data.synced)))
793 {
796 LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
797
799
801 goto restart;
802 }
803 else
805 }
806
807 LWLockRelease(ReplicationSlotControlLock);
808}
809
810/*
811 * Permanently drop replication slot identified by the passed in name.
812 */
813void
814ReplicationSlotDrop(const char *name, bool nowait)
815{
816 Assert(MyReplicationSlot == NULL);
817
818 ReplicationSlotAcquire(name, nowait, false);
819
820 /*
821 * Do not allow users to drop the slots which are currently being synced
822 * from the primary to the standby.
823 */
826 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
827 errmsg("cannot drop replication slot \"%s\"", name),
828 errdetail("This replication slot is being synchronized from the primary server."));
829
831}
832
833/*
834 * Change the definition of the slot identified by the specified name.
835 */
836void
837ReplicationSlotAlter(const char *name, const bool *failover,
838 const bool *two_phase)
839{
840 bool update_slot = false;
841
842 Assert(MyReplicationSlot == NULL);
843 Assert(failover || two_phase);
844
845 ReplicationSlotAcquire(name, false, true);
846
849 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
850 errmsg("cannot use %s with a physical replication slot",
851 "ALTER_REPLICATION_SLOT"));
852
853 if (RecoveryInProgress())
854 {
855 /*
856 * Do not allow users to alter the slots which are currently being
857 * synced from the primary to the standby.
858 */
861 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
862 errmsg("cannot alter replication slot \"%s\"", name),
863 errdetail("This replication slot is being synchronized from the primary server."));
864
865 /*
866 * Do not allow users to enable failover on the standby as we do not
867 * support sync to the cascading standby.
868 */
869 if (failover && *failover)
871 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
872 errmsg("cannot enable failover for a replication slot"
873 " on the standby"));
874 }
875
876 if (failover)
877 {
878 /*
879 * Do not allow users to enable failover for temporary slots as we do
880 * not support syncing temporary slots to the standby.
881 */
882 if (*failover && MyReplicationSlot->data.persistency == RS_TEMPORARY)
884 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
885 errmsg("cannot enable failover for a temporary replication slot"));
886
887 if (MyReplicationSlot->data.failover != *failover)
888 {
890 MyReplicationSlot->data.failover = *failover;
892
893 update_slot = true;
894 }
895 }
896
898 {
902
903 update_slot = true;
904 }
905
906 if (update_slot)
907 {
910 }
911
913}
914
915/*
916 * Permanently drop the currently acquired replication slot.
917 */
918void
920{
922
923 Assert(MyReplicationSlot != NULL);
924
925 /* slot isn't acquired anymore */
926 MyReplicationSlot = NULL;
927
929}
930
931/*
932 * Permanently drop the replication slot which will be released by the point
933 * this function returns.
934 */
935static void
937{
938 char path[MAXPGPATH];
939 char tmppath[MAXPGPATH];
940
941 /*
942 * If some other backend ran this code concurrently with us, we might try
943 * to delete a slot with a certain name while someone else was trying to
944 * create a slot with the same name.
945 */
946 LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
947
948 /* Generate pathnames. */
949 sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(slot->data.name));
950 sprintf(tmppath, "%s/%s.tmp", PG_REPLSLOT_DIR, NameStr(slot->data.name));
951
952 /*
953 * Rename the slot directory on disk, so that we'll no longer recognize
954 * this as a valid slot. Note that if this fails, we've got to mark the
955 * slot inactive before bailing out. If we're dropping an ephemeral or a
956 * temporary slot, we better never fail hard as the caller won't expect
957 * the slot to survive and this might get called during error handling.
958 */
959 if (rename(path, tmppath) == 0)
960 {
961 /*
962 * We need to fsync() the directory we just renamed and its parent to
963 * make sure that our changes are on disk in a crash-safe fashion. If
964 * fsync() fails, we can't be sure whether the changes are on disk or
965 * not. For now, we handle that by panicking;
966 * StartupReplicationSlots() will try to straighten it out after
967 * restart.
968 */
970 fsync_fname(tmppath, true);
973 }
974 else
975 {
976 bool fail_softly = slot->data.persistency != RS_PERSISTENT;
977
978 SpinLockAcquire(&slot->mutex);
979 slot->active_pid = 0;
980 SpinLockRelease(&slot->mutex);
981
982 /* wake up anyone waiting on this slot */
984
985 ereport(fail_softly ? WARNING : ERROR,
987 errmsg("could not rename file \"%s\" to \"%s\": %m",
988 path, tmppath)));
989 }
990
991 /*
992 * The slot is definitely gone. Lock out concurrent scans of the array
993 * long enough to kill it. It's OK to clear the active PID here without
994 * grabbing the mutex because nobody else can be scanning the array here,
995 * and nobody can be attached to this slot and thus access it without
996 * scanning the array.
997 *
998 * Also wake up processes waiting for it.
999 */
1000 LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
1001 slot->active_pid = 0;
1002 slot->in_use = false;
1003 LWLockRelease(ReplicationSlotControlLock);
1005
1006 /*
1007 * Slot is dead and doesn't prevent resource removal anymore, recompute
1008 * limits.
1009 */
1012
1013 /*
1014 * If removing the directory fails, the worst thing that will happen is
1015 * that the user won't be able to create a new slot with the same name
1016 * until the next server restart. We warn about it, but that's all.
1017 */
1018 if (!rmtree(tmppath, true))
1020 (errmsg("could not remove directory \"%s\"", tmppath)));
1021
1022 /*
1023 * Drop the statistics entry for the replication slot. Do this while
1024 * holding ReplicationSlotAllocationLock so that we don't drop a
1025 * statistics entry for another slot with the same name just created in
1026 * another session.
1027 */
1028 if (SlotIsLogical(slot))
1030
1031 /*
1032 * We release this at the very end, so that nobody starts trying to create
1033 * a slot while we're still cleaning up the detritus of the old one.
1034 */
1035 LWLockRelease(ReplicationSlotAllocationLock);
1036}
1037
1038/*
1039 * Serialize the currently acquired slot's state from memory to disk, thereby
1040 * guaranteeing the current state will survive a crash.
1041 */
1042void
1044{
1045 char path[MAXPGPATH];
1046
1047 Assert(MyReplicationSlot != NULL);
1048
1051}
1052
1053/*
1054 * Signal that it would be useful if the currently acquired slot would be
1055 * flushed out to disk.
1056 *
1057 * Note that the actual flush to disk can be delayed for a long time, if
1058 * required for correctness explicitly do a ReplicationSlotSave().
1059 */
1060void
1062{
1064
1065 Assert(MyReplicationSlot != NULL);
1066
1067 SpinLockAcquire(&slot->mutex);
1069 MyReplicationSlot->dirty = true;
1070 SpinLockRelease(&slot->mutex);
1071}
1072
1073/*
1074 * Convert a slot that's marked as RS_EPHEMERAL or RS_TEMPORARY to a
1075 * RS_PERSISTENT slot, guaranteeing it will be there after an eventual crash.
1076 */
1077void
1079{
1081
1082 Assert(slot != NULL);
1084
1085 SpinLockAcquire(&slot->mutex);
1087 SpinLockRelease(&slot->mutex);
1088
1091}
1092
1093/*
1094 * Compute the oldest xmin across all slots and store it in the ProcArray.
1095 *
1096 * If already_locked is true, ProcArrayLock has already been acquired
1097 * exclusively.
1098 */
1099void
1101{
1102 int i;
1104 TransactionId agg_catalog_xmin = InvalidTransactionId;
1105
1106 Assert(ReplicationSlotCtl != NULL);
1107
1108 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1109
1110 for (i = 0; i < max_replication_slots; i++)
1111 {
1113 TransactionId effective_xmin;
1114 TransactionId effective_catalog_xmin;
1115 bool invalidated;
1116
1117 if (!s->in_use)
1118 continue;
1119
1121 effective_xmin = s->effective_xmin;
1122 effective_catalog_xmin = s->effective_catalog_xmin;
1123 invalidated = s->data.invalidated != RS_INVAL_NONE;
1125
1126 /* invalidated slots need not apply */
1127 if (invalidated)
1128 continue;
1129
1130 /* check the data xmin */
1131 if (TransactionIdIsValid(effective_xmin) &&
1132 (!TransactionIdIsValid(agg_xmin) ||
1133 TransactionIdPrecedes(effective_xmin, agg_xmin)))
1134 agg_xmin = effective_xmin;
1135
1136 /* check the catalog xmin */
1137 if (TransactionIdIsValid(effective_catalog_xmin) &&
1138 (!TransactionIdIsValid(agg_catalog_xmin) ||
1139 TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
1140 agg_catalog_xmin = effective_catalog_xmin;
1141 }
1142
1143 LWLockRelease(ReplicationSlotControlLock);
1144
1145 ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
1146}
1147
1148/*
1149 * Compute the oldest restart LSN across all slots and inform xlog module.
1150 *
1151 * Note: while max_slot_wal_keep_size is theoretically relevant for this
1152 * purpose, we don't try to account for that, because this module doesn't
1153 * know what to compare against.
1154 */
1155void
1157{
1158 int i;
1159 XLogRecPtr min_required = InvalidXLogRecPtr;
1160
1161 Assert(ReplicationSlotCtl != NULL);
1162
1163 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1164 for (i = 0; i < max_replication_slots; i++)
1165 {
1167 XLogRecPtr restart_lsn;
1168 bool invalidated;
1169
1170 if (!s->in_use)
1171 continue;
1172
1174 restart_lsn = s->data.restart_lsn;
1175 invalidated = s->data.invalidated != RS_INVAL_NONE;
1177
1178 /* invalidated slots need not apply */
1179 if (invalidated)
1180 continue;
1181
1182 if (restart_lsn != InvalidXLogRecPtr &&
1183 (min_required == InvalidXLogRecPtr ||
1184 restart_lsn < min_required))
1185 min_required = restart_lsn;
1186 }
1187 LWLockRelease(ReplicationSlotControlLock);
1188
1190}
1191
1192/*
1193 * Compute the oldest WAL LSN required by *logical* decoding slots..
1194 *
1195 * Returns InvalidXLogRecPtr if logical decoding is disabled or no logical
1196 * slots exist.
1197 *
1198 * NB: this returns a value >= ReplicationSlotsComputeRequiredLSN(), since it
1199 * ignores physical replication slots.
1200 *
1201 * The results aren't required frequently, so we don't maintain a precomputed
1202 * value like we do for ComputeRequiredLSN() and ComputeRequiredXmin().
1203 */
1206{
1208 int i;
1209
1210 if (max_replication_slots <= 0)
1211 return InvalidXLogRecPtr;
1212
1213 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1214
1215 for (i = 0; i < max_replication_slots; i++)
1216 {
1217 ReplicationSlot *s;
1218 XLogRecPtr restart_lsn;
1219 bool invalidated;
1220
1222
1223 /* cannot change while ReplicationSlotCtlLock is held */
1224 if (!s->in_use)
1225 continue;
1226
1227 /* we're only interested in logical slots */
1228 if (!SlotIsLogical(s))
1229 continue;
1230
1231 /* read once, it's ok if it increases while we're checking */
1233 restart_lsn = s->data.restart_lsn;
1234 invalidated = s->data.invalidated != RS_INVAL_NONE;
1236
1237 /* invalidated slots need not apply */
1238 if (invalidated)
1239 continue;
1240
1241 if (restart_lsn == InvalidXLogRecPtr)
1242 continue;
1243
1244 if (result == InvalidXLogRecPtr ||
1245 restart_lsn < result)
1246 result = restart_lsn;
1247 }
1248
1249 LWLockRelease(ReplicationSlotControlLock);
1250
1251 return result;
1252}
1253
1254/*
1255 * ReplicationSlotsCountDBSlots -- count the number of slots that refer to the
1256 * passed database oid.
1257 *
1258 * Returns true if there are any slots referencing the database. *nslots will
1259 * be set to the absolute number of slots in the database, *nactive to ones
1260 * currently active.
1261 */
1262bool
1263ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
1264{
1265 int i;
1266
1267 *nslots = *nactive = 0;
1268
1269 if (max_replication_slots <= 0)
1270 return false;
1271
1272 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1273 for (i = 0; i < max_replication_slots; i++)
1274 {
1275 ReplicationSlot *s;
1276
1278
1279 /* cannot change while ReplicationSlotCtlLock is held */
1280 if (!s->in_use)
1281 continue;
1282
1283 /* only logical slots are database specific, skip */
1284 if (!SlotIsLogical(s))
1285 continue;
1286
1287 /* not our database, skip */
1288 if (s->data.database != dboid)
1289 continue;
1290
1291 /* NB: intentionally counting invalidated slots */
1292
1293 /* count slots with spinlock held */
1295 (*nslots)++;
1296 if (s->active_pid != 0)
1297 (*nactive)++;
1299 }
1300 LWLockRelease(ReplicationSlotControlLock);
1301
1302 if (*nslots > 0)
1303 return true;
1304 return false;
1305}
1306
1307/*
1308 * ReplicationSlotsDropDBSlots -- Drop all db-specific slots relating to the
1309 * passed database oid. The caller should hold an exclusive lock on the
1310 * pg_database oid for the database to prevent creation of new slots on the db
1311 * or replay from existing slots.
1312 *
1313 * Another session that concurrently acquires an existing slot on the target DB
1314 * (most likely to drop it) may cause this function to ERROR. If that happens
1315 * it may have dropped some but not all slots.
1316 *
1317 * This routine isn't as efficient as it could be - but we don't drop
1318 * databases often, especially databases with lots of slots.
1319 */
1320void
1322{
1323 int i;
1324
1325 if (max_replication_slots <= 0)
1326 return;
1327
1328restart:
1329 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1330 for (i = 0; i < max_replication_slots; i++)
1331 {
1332 ReplicationSlot *s;
1333 char *slotname;
1334 int active_pid;
1335
1337
1338 /* cannot change while ReplicationSlotCtlLock is held */
1339 if (!s->in_use)
1340 continue;
1341
1342 /* only logical slots are database specific, skip */
1343 if (!SlotIsLogical(s))
1344 continue;
1345
1346 /* not our database, skip */
1347 if (s->data.database != dboid)
1348 continue;
1349
1350 /* NB: intentionally including invalidated slots */
1351
1352 /* acquire slot, so ReplicationSlotDropAcquired can be reused */
1354 /* can't change while ReplicationSlotControlLock is held */
1355 slotname = NameStr(s->data.name);
1356 active_pid = s->active_pid;
1357 if (active_pid == 0)
1358 {
1360 s->active_pid = MyProcPid;
1361 }
1363
1364 /*
1365 * Even though we hold an exclusive lock on the database object a
1366 * logical slot for that DB can still be active, e.g. if it's
1367 * concurrently being dropped by a backend connected to another DB.
1368 *
1369 * That's fairly unlikely in practice, so we'll just bail out.
1370 *
1371 * The slot sync worker holds a shared lock on the database before
1372 * operating on synced logical slots to avoid conflict with the drop
1373 * happening here. The persistent synced slots are thus safe but there
1374 * is a possibility that the slot sync worker has created a temporary
1375 * slot (which stays active even on release) and we are trying to drop
1376 * that here. In practice, the chances of hitting this scenario are
1377 * less as during slot synchronization, the temporary slot is
1378 * immediately converted to persistent and thus is safe due to the
1379 * shared lock taken on the database. So, we'll just bail out in such
1380 * a case.
1381 *
1382 * XXX: We can consider shutting down the slot sync worker before
1383 * trying to drop synced temporary slots here.
1384 */
1385 if (active_pid)
1386 ereport(ERROR,
1387 (errcode(ERRCODE_OBJECT_IN_USE),
1388 errmsg("replication slot \"%s\" is active for PID %d",
1389 slotname, active_pid)));
1390
1391 /*
1392 * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1393 * holding ReplicationSlotControlLock over filesystem operations,
1394 * release ReplicationSlotControlLock and use
1395 * ReplicationSlotDropAcquired.
1396 *
1397 * As that means the set of slots could change, restart scan from the
1398 * beginning each time we release the lock.
1399 */
1400 LWLockRelease(ReplicationSlotControlLock);
1402 goto restart;
1403 }
1404 LWLockRelease(ReplicationSlotControlLock);
1405}
1406
1407
1408/*
1409 * Check whether the server's configuration supports using replication
1410 * slots.
1411 */
1412void
1414{
1415 /*
1416 * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1417 * needs the same check.
1418 */
1419
1420 if (max_replication_slots == 0)
1421 ereport(ERROR,
1422 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1423 errmsg("replication slots can only be used if \"max_replication_slots\" > 0")));
1424
1426 ereport(ERROR,
1427 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1428 errmsg("replication slots can only be used if \"wal_level\" >= \"replica\"")));
1429}
1430
1431/*
1432 * Check whether the user has privilege to use replication slots.
1433 */
1434void
1436{
1438 ereport(ERROR,
1439 (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1440 errmsg("permission denied to use replication slots"),
1441 errdetail("Only roles with the %s attribute may use replication slots.",
1442 "REPLICATION")));
1443}
1444
1445/*
1446 * Reserve WAL for the currently active slot.
1447 *
1448 * Compute and set restart_lsn in a manner that's appropriate for the type of
1449 * the slot and concurrency safe.
1450 */
1451void
1453{
1455
1456 Assert(slot != NULL);
1458
1459 /*
1460 * The replication slot mechanism is used to prevent removal of required
1461 * WAL. As there is no interlock between this routine and checkpoints, WAL
1462 * segments could concurrently be removed when a now stale return value of
1463 * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1464 * this happens we'll just retry.
1465 */
1466 while (true)
1467 {
1468 XLogSegNo segno;
1469 XLogRecPtr restart_lsn;
1470
1471 /*
1472 * For logical slots log a standby snapshot and start logical decoding
1473 * at exactly that position. That allows the slot to start up more
1474 * quickly. But on a standby we cannot do WAL writes, so just use the
1475 * replay pointer; effectively, an attempt to create a logical slot on
1476 * standby will cause it to wait for an xl_running_xact record to be
1477 * logged independently on the primary, so that a snapshot can be
1478 * built using the record.
1479 *
1480 * None of this is needed (or indeed helpful) for physical slots as
1481 * they'll start replay at the last logged checkpoint anyway. Instead
1482 * return the location of the last redo LSN. While that slightly
1483 * increases the chance that we have to retry, it's where a base
1484 * backup has to start replay at.
1485 */
1486 if (SlotIsPhysical(slot))
1487 restart_lsn = GetRedoRecPtr();
1488 else if (RecoveryInProgress())
1489 restart_lsn = GetXLogReplayRecPtr(NULL);
1490 else
1491 restart_lsn = GetXLogInsertRecPtr();
1492
1493 SpinLockAcquire(&slot->mutex);
1494 slot->data.restart_lsn = restart_lsn;
1495 SpinLockRelease(&slot->mutex);
1496
1497 /* prevent WAL removal as fast as possible */
1499
1500 /*
1501 * If all required WAL is still there, great, otherwise retry. The
1502 * slot should prevent further removal of WAL, unless there's a
1503 * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1504 * the new restart_lsn above, so normally we should never need to loop
1505 * more than twice.
1506 */
1508 if (XLogGetLastRemovedSegno() < segno)
1509 break;
1510 }
1511
1512 if (!RecoveryInProgress() && SlotIsLogical(slot))
1513 {
1514 XLogRecPtr flushptr;
1515
1516 /* make sure we have enough information to start */
1517 flushptr = LogStandbySnapshot();
1518
1519 /* and make sure it's fsynced to disk */
1520 XLogFlush(flushptr);
1521 }
1522}
1523
1524/*
1525 * Report that replication slot needs to be invalidated
1526 */
1527static void
1529 bool terminating,
1530 int pid,
1531 NameData slotname,
1532 XLogRecPtr restart_lsn,
1533 XLogRecPtr oldestLSN,
1534 TransactionId snapshotConflictHorizon,
1535 long slot_idle_seconds)
1536{
1537 StringInfoData err_detail;
1538 StringInfoData err_hint;
1539
1540 initStringInfo(&err_detail);
1541 initStringInfo(&err_hint);
1542
1543 switch (cause)
1544 {
1546 {
1547 unsigned long long ex = oldestLSN - restart_lsn;
1548
1549 appendStringInfo(&err_detail,
1550 ngettext("The slot's restart_lsn %X/%X exceeds the limit by %llu byte.",
1551 "The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
1552 ex),
1553 LSN_FORMAT_ARGS(restart_lsn),
1554 ex);
1555 /* translator: %s is a GUC variable name */
1556 appendStringInfo(&err_hint, _("You might need to increase \"%s\"."),
1557 "max_slot_wal_keep_size");
1558 break;
1559 }
1560 case RS_INVAL_HORIZON:
1561 appendStringInfo(&err_detail, _("The slot conflicted with xid horizon %u."),
1562 snapshotConflictHorizon);
1563 break;
1564
1565 case RS_INVAL_WAL_LEVEL:
1566 appendStringInfoString(&err_detail, _("Logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary server."));
1567 break;
1568
1570 {
1571 int minutes = slot_idle_seconds / SECS_PER_MINUTE;
1572 int secs = slot_idle_seconds % SECS_PER_MINUTE;
1573
1574 /* translator: %s is a GUC variable name */
1575 appendStringInfo(&err_detail, _("The slot's idle time of %dmin %02ds exceeds the configured \"%s\" duration of %dmin."),
1576 minutes, secs, "idle_replication_slot_timeout",
1578 /* translator: %s is a GUC variable name */
1579 appendStringInfo(&err_hint, _("You might need to increase \"%s\"."),
1580 "idle_replication_slot_timeout");
1581 break;
1582 }
1583 case RS_INVAL_NONE:
1585 }
1586
1587 ereport(LOG,
1588 terminating ?
1589 errmsg("terminating process %d to release replication slot \"%s\"",
1590 pid, NameStr(slotname)) :
1591 errmsg("invalidating obsolete replication slot \"%s\"",
1592 NameStr(slotname)),
1593 errdetail_internal("%s", err_detail.data),
1594 err_hint.len ? errhint("%s", err_hint.data) : 0);
1595
1596 pfree(err_detail.data);
1597 pfree(err_hint.data);
1598}
1599
1600/*
1601 * Can we invalidate an idle replication slot?
1602 *
1603 * Idle timeout invalidation is allowed only when:
1604 *
1605 * 1. Idle timeout is set
1606 * 2. Slot has reserved WAL
1607 * 3. Slot is inactive
1608 * 4. The slot is not being synced from the primary while the server is in
1609 * recovery. This is because synced slots are always considered to be
1610 * inactive because they don't perform logical decoding to produce changes.
1611 */
1612static inline bool
1614{
1617 s->inactive_since > 0 &&
1618 !(RecoveryInProgress() && s->data.synced));
1619}
1620
1621/*
1622 * DetermineSlotInvalidationCause - Determine the cause for which a slot
1623 * becomes invalid among the given possible causes.
1624 *
1625 * This function sequentially checks all possible invalidation causes and
1626 * returns the first one for which the slot is eligible for invalidation.
1627 */
1630 XLogRecPtr oldestLSN, Oid dboid,
1631 TransactionId snapshotConflictHorizon,
1632 TransactionId initial_effective_xmin,
1633 TransactionId initial_catalog_effective_xmin,
1634 XLogRecPtr initial_restart_lsn,
1635 TimestampTz *inactive_since, TimestampTz now)
1636{
1637 Assert(possible_causes != RS_INVAL_NONE);
1638
1639 if (possible_causes & RS_INVAL_WAL_REMOVED)
1640 {
1641 if (initial_restart_lsn != InvalidXLogRecPtr &&
1642 initial_restart_lsn < oldestLSN)
1643 return RS_INVAL_WAL_REMOVED;
1644 }
1645
1646 if (possible_causes & RS_INVAL_HORIZON)
1647 {
1648 /* invalid DB oid signals a shared relation */
1649 if (SlotIsLogical(s) &&
1650 (dboid == InvalidOid || dboid == s->data.database))
1651 {
1652 if (TransactionIdIsValid(initial_effective_xmin) &&
1653 TransactionIdPrecedesOrEquals(initial_effective_xmin,
1654 snapshotConflictHorizon))
1655 return RS_INVAL_HORIZON;
1656 else if (TransactionIdIsValid(initial_catalog_effective_xmin) &&
1657 TransactionIdPrecedesOrEquals(initial_catalog_effective_xmin,
1658 snapshotConflictHorizon))
1659 return RS_INVAL_HORIZON;
1660 }
1661 }
1662
1663 if (possible_causes & RS_INVAL_WAL_LEVEL)
1664 {
1665 if (SlotIsLogical(s))
1666 return RS_INVAL_WAL_LEVEL;
1667 }
1668
1669 if (possible_causes & RS_INVAL_IDLE_TIMEOUT)
1670 {
1671 Assert(now > 0);
1672
1673 if (CanInvalidateIdleSlot(s))
1674 {
1675 /*
1676 * We simulate the invalidation due to idle_timeout as the minimum
1677 * time idle time is one minute which makes tests take a long
1678 * time.
1679 */
1680#ifdef USE_INJECTION_POINTS
1681 if (IS_INJECTION_POINT_ATTACHED("slot-timeout-inval"))
1682 {
1683 *inactive_since = 0; /* since the beginning of time */
1684 return RS_INVAL_IDLE_TIMEOUT;
1685 }
1686#endif
1687
1688 /*
1689 * Check if the slot needs to be invalidated due to
1690 * idle_replication_slot_timeout GUC.
1691 */
1694 {
1695 *inactive_since = s->inactive_since;
1696 return RS_INVAL_IDLE_TIMEOUT;
1697 }
1698 }
1699 }
1700
1701 return RS_INVAL_NONE;
1702}
1703
1704/*
1705 * Helper for InvalidateObsoleteReplicationSlots
1706 *
1707 * Acquires the given slot and mark it invalid, if necessary and possible.
1708 *
1709 * Returns whether ReplicationSlotControlLock was released in the interim (and
1710 * in that case we're not holding the lock at return, otherwise we are).
1711 *
1712 * Sets *invalidated true if the slot was invalidated. (Untouched otherwise.)
1713 *
1714 * This is inherently racy, because we release the LWLock
1715 * for syscalls, so caller must restart if we return true.
1716 */
1717static bool
1719 ReplicationSlot *s,
1720 XLogRecPtr oldestLSN,
1721 Oid dboid, TransactionId snapshotConflictHorizon,
1722 bool *invalidated)
1723{
1724 int last_signaled_pid = 0;
1725 bool released_lock = false;
1726 bool terminated = false;
1727 TransactionId initial_effective_xmin = InvalidTransactionId;
1728 TransactionId initial_catalog_effective_xmin = InvalidTransactionId;
1729 XLogRecPtr initial_restart_lsn = InvalidXLogRecPtr;
1731 TimestampTz inactive_since = 0;
1732
1733 for (;;)
1734 {
1735 XLogRecPtr restart_lsn;
1736 NameData slotname;
1737 int active_pid = 0;
1739 TimestampTz now = 0;
1740 long slot_idle_secs = 0;
1741
1742 Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
1743
1744 if (!s->in_use)
1745 {
1746 if (released_lock)
1747 LWLockRelease(ReplicationSlotControlLock);
1748 break;
1749 }
1750
1751 if (possible_causes & RS_INVAL_IDLE_TIMEOUT)
1752 {
1753 /*
1754 * Assign the current time here to avoid system call overhead
1755 * while holding the spinlock in subsequent code.
1756 */
1758 }
1759
1760 /*
1761 * Check if the slot needs to be invalidated. If it needs to be
1762 * invalidated, and is not currently acquired, acquire it and mark it
1763 * as having been invalidated. We do this with the spinlock held to
1764 * avoid race conditions -- for example the restart_lsn could move
1765 * forward, or the slot could be dropped.
1766 */
1768
1769 restart_lsn = s->data.restart_lsn;
1770
1771 /* we do nothing if the slot is already invalid */
1772 if (s->data.invalidated == RS_INVAL_NONE)
1773 {
1774 /*
1775 * The slot's mutex will be released soon, and it is possible that
1776 * those values change since the process holding the slot has been
1777 * terminated (if any), so record them here to ensure that we
1778 * would report the correct invalidation cause.
1779 *
1780 * Unlike other slot attributes, slot's inactive_since can't be
1781 * changed until the acquired slot is released or the owning
1782 * process is terminated. So, the inactive slot can only be
1783 * invalidated immediately without being terminated.
1784 */
1785 if (!terminated)
1786 {
1787 initial_restart_lsn = s->data.restart_lsn;
1788 initial_effective_xmin = s->effective_xmin;
1789 initial_catalog_effective_xmin = s->effective_catalog_xmin;
1790 }
1791
1792 invalidation_cause = DetermineSlotInvalidationCause(possible_causes,
1793 s, oldestLSN,
1794 dboid,
1795 snapshotConflictHorizon,
1796 initial_effective_xmin,
1797 initial_catalog_effective_xmin,
1798 initial_restart_lsn,
1799 &inactive_since,
1800 now);
1801 }
1802
1803 /*
1804 * The invalidation cause recorded previously should not change while
1805 * the process owning the slot (if any) has been terminated.
1806 */
1807 Assert(!(invalidation_cause_prev != RS_INVAL_NONE && terminated &&
1808 invalidation_cause_prev != invalidation_cause));
1809
1810 /* if there's no invalidation, we're done */
1811 if (invalidation_cause == RS_INVAL_NONE)
1812 {
1814 if (released_lock)
1815 LWLockRelease(ReplicationSlotControlLock);
1816 break;
1817 }
1818
1819 slotname = s->data.name;
1820 active_pid = s->active_pid;
1821
1822 /*
1823 * If the slot can be acquired, do so and mark it invalidated
1824 * immediately. Otherwise we'll signal the owning process, below, and
1825 * retry.
1826 */
1827 if (active_pid == 0)
1828 {
1830 s->active_pid = MyProcPid;
1831 s->data.invalidated = invalidation_cause;
1832
1833 /*
1834 * XXX: We should consider not overwriting restart_lsn and instead
1835 * just rely on .invalidated.
1836 */
1837 if (invalidation_cause == RS_INVAL_WAL_REMOVED)
1839
1840 /* Let caller know */
1841 *invalidated = true;
1842 }
1843
1845
1846 /*
1847 * The logical replication slots shouldn't be invalidated as GUC
1848 * max_slot_wal_keep_size is set to -1 and
1849 * idle_replication_slot_timeout is set to 0 during the binary
1850 * upgrade. See check_old_cluster_for_valid_slots() where we ensure
1851 * that no invalidated before the upgrade.
1852 */
1853 Assert(!(*invalidated && SlotIsLogical(s) && IsBinaryUpgrade));
1854
1855 /*
1856 * Calculate the idle time duration of the slot if slot is marked
1857 * invalidated with RS_INVAL_IDLE_TIMEOUT.
1858 */
1859 if (invalidation_cause == RS_INVAL_IDLE_TIMEOUT)
1860 {
1861 int slot_idle_usecs;
1862
1863 TimestampDifference(inactive_since, now, &slot_idle_secs,
1864 &slot_idle_usecs);
1865 }
1866
1867 if (active_pid != 0)
1868 {
1869 /*
1870 * Prepare the sleep on the slot's condition variable before
1871 * releasing the lock, to close a possible race condition if the
1872 * slot is released before the sleep below.
1873 */
1875
1876 LWLockRelease(ReplicationSlotControlLock);
1877 released_lock = true;
1878
1879 /*
1880 * Signal to terminate the process that owns the slot, if we
1881 * haven't already signalled it. (Avoidance of repeated
1882 * signalling is the only reason for there to be a loop in this
1883 * routine; otherwise we could rely on caller's restart loop.)
1884 *
1885 * There is the race condition that other process may own the slot
1886 * after its current owner process is terminated and before this
1887 * process owns it. To handle that, we signal only if the PID of
1888 * the owning process has changed from the previous time. (This
1889 * logic assumes that the same PID is not reused very quickly.)
1890 */
1891 if (last_signaled_pid != active_pid)
1892 {
1893 ReportSlotInvalidation(invalidation_cause, true, active_pid,
1894 slotname, restart_lsn,
1895 oldestLSN, snapshotConflictHorizon,
1896 slot_idle_secs);
1897
1898 if (MyBackendType == B_STARTUP)
1899 (void) SendProcSignal(active_pid,
1902 else
1903 (void) kill(active_pid, SIGTERM);
1904
1905 last_signaled_pid = active_pid;
1906 terminated = true;
1907 invalidation_cause_prev = invalidation_cause;
1908 }
1909
1910 /* Wait until the slot is released. */
1912 WAIT_EVENT_REPLICATION_SLOT_DROP);
1913
1914 /*
1915 * Re-acquire lock and start over; we expect to invalidate the
1916 * slot next time (unless another process acquires the slot in the
1917 * meantime).
1918 */
1919 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1920 continue;
1921 }
1922 else
1923 {
1924 /*
1925 * We hold the slot now and have already invalidated it; flush it
1926 * to ensure that state persists.
1927 *
1928 * Don't want to hold ReplicationSlotControlLock across file
1929 * system operations, so release it now but be sure to tell caller
1930 * to restart from scratch.
1931 */
1932 LWLockRelease(ReplicationSlotControlLock);
1933 released_lock = true;
1934
1935 /* Make sure the invalidated state persists across server restart */
1939
1940 ReportSlotInvalidation(invalidation_cause, false, active_pid,
1941 slotname, restart_lsn,
1942 oldestLSN, snapshotConflictHorizon,
1943 slot_idle_secs);
1944
1945 /* done with this slot for now */
1946 break;
1947 }
1948 }
1949
1950 Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock));
1951
1952 return released_lock;
1953}
1954
1955/*
1956 * Invalidate slots that require resources about to be removed.
1957 *
1958 * Returns true when any slot have got invalidated.
1959 *
1960 * Whether a slot needs to be invalidated depends on the invalidation cause.
1961 * A slot is invalidated if it:
1962 * - RS_INVAL_WAL_REMOVED: requires a LSN older than the given segment
1963 * - RS_INVAL_HORIZON: requires a snapshot <= the given horizon in the given
1964 * db; dboid may be InvalidOid for shared relations
1965 * - RS_INVAL_WAL_LEVEL: is logical and wal_level is insufficient
1966 * - RS_INVAL_IDLE_TIMEOUT: has been idle longer than the configured
1967 * "idle_replication_slot_timeout" duration.
1968 *
1969 * Note: This function attempts to invalidate the slot for multiple possible
1970 * causes in a single pass, minimizing redundant iterations. The "cause"
1971 * parameter can be a MASK representing one or more of the defined causes.
1972 *
1973 * NB - this runs as part of checkpoint, so avoid raising errors if possible.
1974 */
1975bool
1977 XLogSegNo oldestSegno, Oid dboid,
1978 TransactionId snapshotConflictHorizon)
1979{
1980 XLogRecPtr oldestLSN;
1981 bool invalidated = false;
1982
1983 Assert(!(possible_causes & RS_INVAL_HORIZON) || TransactionIdIsValid(snapshotConflictHorizon));
1984 Assert(!(possible_causes & RS_INVAL_WAL_REMOVED) || oldestSegno > 0);
1985 Assert(possible_causes != RS_INVAL_NONE);
1986
1987 if (max_replication_slots == 0)
1988 return invalidated;
1989
1990 XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1991
1992restart:
1993 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1994 for (int i = 0; i < max_replication_slots; i++)
1995 {
1997
1998 if (!s->in_use)
1999 continue;
2000
2001 if (InvalidatePossiblyObsoleteSlot(possible_causes, s, oldestLSN, dboid,
2002 snapshotConflictHorizon,
2003 &invalidated))
2004 {
2005 /* if the lock was released, start from scratch */
2006 goto restart;
2007 }
2008 }
2009 LWLockRelease(ReplicationSlotControlLock);
2010
2011 /*
2012 * If any slots have been invalidated, recalculate the resource limits.
2013 */
2014 if (invalidated)
2015 {
2018 }
2019
2020 return invalidated;
2021}
2022
2023/*
2024 * Flush all replication slots to disk.
2025 *
2026 * It is convenient to flush dirty replication slots at the time of checkpoint.
2027 * Additionally, in case of a shutdown checkpoint, we also identify the slots
2028 * for which the confirmed_flush LSN has been updated since the last time it
2029 * was saved and flush them.
2030 */
2031void
2033{
2034 int i;
2035
2036 elog(DEBUG1, "performing replication slot checkpoint");
2037
2038 /*
2039 * Prevent any slot from being created/dropped while we're active. As we
2040 * explicitly do *not* want to block iterating over replication_slots or
2041 * acquiring a slot we cannot take the control lock - but that's OK,
2042 * because holding ReplicationSlotAllocationLock is strictly stronger, and
2043 * enough to guarantee that nobody can change the in_use bits on us.
2044 */
2045 LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
2046
2047 for (i = 0; i < max_replication_slots; i++)
2048 {
2050 char path[MAXPGPATH];
2051
2052 if (!s->in_use)
2053 continue;
2054
2055 /* save the slot to disk, locking is handled in SaveSlotToPath() */
2056 sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(s->data.name));
2057
2058 /*
2059 * Slot's data is not flushed each time the confirmed_flush LSN is
2060 * updated as that could lead to frequent writes. However, we decide
2061 * to force a flush of all logical slot's data at the time of shutdown
2062 * if the confirmed_flush LSN is changed since we last flushed it to
2063 * disk. This helps in avoiding an unnecessary retreat of the
2064 * confirmed_flush LSN after restart.
2065 */
2066 if (is_shutdown && SlotIsLogical(s))
2067 {
2069
2070 if (s->data.invalidated == RS_INVAL_NONE &&
2072 {
2073 s->just_dirtied = true;
2074 s->dirty = true;
2075 }
2077 }
2078
2079 SaveSlotToPath(s, path, LOG);
2080 }
2081 LWLockRelease(ReplicationSlotAllocationLock);
2082}
2083
2084/*
2085 * Load all replication slots from disk into memory at server startup. This
2086 * needs to be run before we start crash recovery.
2087 */
2088void
2090{
2091 DIR *replication_dir;
2092 struct dirent *replication_de;
2093
2094 elog(DEBUG1, "starting up replication slots");
2095
2096 /* restore all slots by iterating over all on-disk entries */
2097 replication_dir = AllocateDir(PG_REPLSLOT_DIR);
2098 while ((replication_de = ReadDir(replication_dir, PG_REPLSLOT_DIR)) != NULL)
2099 {
2100 char path[MAXPGPATH + sizeof(PG_REPLSLOT_DIR)];
2101 PGFileType de_type;
2102
2103 if (strcmp(replication_de->d_name, ".") == 0 ||
2104 strcmp(replication_de->d_name, "..") == 0)
2105 continue;
2106
2107 snprintf(path, sizeof(path), "%s/%s", PG_REPLSLOT_DIR, replication_de->d_name);
2108 de_type = get_dirent_type(path, replication_de, false, DEBUG1);
2109
2110 /* we're only creating directories here, skip if it's not our's */
2111 if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_DIR)
2112 continue;
2113
2114 /* we crashed while a slot was being setup or deleted, clean up */
2115 if (pg_str_endswith(replication_de->d_name, ".tmp"))
2116 {
2117 if (!rmtree(path, true))
2118 {
2120 (errmsg("could not remove directory \"%s\"",
2121 path)));
2122 continue;
2123 }
2125 continue;
2126 }
2127
2128 /* looks like a slot in a normal state, restore */
2129 RestoreSlotFromDisk(replication_de->d_name);
2130 }
2131 FreeDir(replication_dir);
2132
2133 /* currently no slots exist, we're done. */
2134 if (max_replication_slots <= 0)
2135 return;
2136
2137 /* Now that we have recovered all the data, compute replication xmin */
2140}
2141
2142/* ----
2143 * Manipulation of on-disk state of replication slots
2144 *
2145 * NB: none of the routines below should take any notice whether a slot is the
2146 * current one or not, that's all handled a layer above.
2147 * ----
2148 */
2149static void
2151{
2152 char tmppath[MAXPGPATH];
2153 char path[MAXPGPATH];
2154 struct stat st;
2155
2156 /*
2157 * No need to take out the io_in_progress_lock, nobody else can see this
2158 * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
2159 * takes out the lock, if we'd take the lock here, we'd deadlock.
2160 */
2161
2162 sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(slot->data.name));
2163 sprintf(tmppath, "%s/%s.tmp", PG_REPLSLOT_DIR, NameStr(slot->data.name));
2164
2165 /*
2166 * It's just barely possible that some previous effort to create or drop a
2167 * slot with this name left a temp directory lying around. If that seems
2168 * to be the case, try to remove it. If the rmtree() fails, we'll error
2169 * out at the MakePGDirectory() below, so we don't bother checking
2170 * success.
2171 */
2172 if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
2173 rmtree(tmppath, true);
2174
2175 /* Create and fsync the temporary slot directory. */
2176 if (MakePGDirectory(tmppath) < 0)
2177 ereport(ERROR,
2179 errmsg("could not create directory \"%s\": %m",
2180 tmppath)));
2181 fsync_fname(tmppath, true);
2182
2183 /* Write the actual state file. */
2184 slot->dirty = true; /* signal that we really need to write */
2185 SaveSlotToPath(slot, tmppath, ERROR);
2186
2187 /* Rename the directory into place. */
2188 if (rename(tmppath, path) != 0)
2189 ereport(ERROR,
2191 errmsg("could not rename file \"%s\" to \"%s\": %m",
2192 tmppath, path)));
2193
2194 /*
2195 * If we'd now fail - really unlikely - we wouldn't know whether this slot
2196 * would persist after an OS crash or not - so, force a restart. The
2197 * restart would try to fsync this again till it works.
2198 */
2200
2201 fsync_fname(path, true);
2203
2205}
2206
2207/*
2208 * Shared functionality between saving and creating a replication slot.
2209 */
2210static void
2211SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
2212{
2213 char tmppath[MAXPGPATH];
2214 char path[MAXPGPATH];
2215 int fd;
2217 bool was_dirty;
2218
2219 /* first check whether there's something to write out */
2220 SpinLockAcquire(&slot->mutex);
2221 was_dirty = slot->dirty;
2222 slot->just_dirtied = false;
2223 SpinLockRelease(&slot->mutex);
2224
2225 /* and don't do anything if there's nothing to write */
2226 if (!was_dirty)
2227 return;
2228
2230
2231 /* silence valgrind :( */
2232 memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
2233
2234 sprintf(tmppath, "%s/state.tmp", dir);
2235 sprintf(path, "%s/state", dir);
2236
2237 fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
2238 if (fd < 0)
2239 {
2240 /*
2241 * If not an ERROR, then release the lock before returning. In case
2242 * of an ERROR, the error recovery path automatically releases the
2243 * lock, but no harm in explicitly releasing even in that case. Note
2244 * that LWLockRelease() could affect errno.
2245 */
2246 int save_errno = errno;
2247
2249 errno = save_errno;
2250 ereport(elevel,
2252 errmsg("could not create file \"%s\": %m",
2253 tmppath)));
2254 return;
2255 }
2256
2257 cp.magic = SLOT_MAGIC;
2259 cp.version = SLOT_VERSION;
2261
2262 SpinLockAcquire(&slot->mutex);
2263
2264 memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
2265
2266 SpinLockRelease(&slot->mutex);
2267
2271 FIN_CRC32C(cp.checksum);
2272
2273 errno = 0;
2274 pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_WRITE);
2275 if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
2276 {
2277 int save_errno = errno;
2278
2282
2283 /* if write didn't set errno, assume problem is no disk space */
2284 errno = save_errno ? save_errno : ENOSPC;
2285 ereport(elevel,
2287 errmsg("could not write to file \"%s\": %m",
2288 tmppath)));
2289 return;
2290 }
2292
2293 /* fsync the temporary file */
2294 pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_SYNC);
2295 if (pg_fsync(fd) != 0)
2296 {
2297 int save_errno = errno;
2298
2302 errno = save_errno;
2303 ereport(elevel,
2305 errmsg("could not fsync file \"%s\": %m",
2306 tmppath)));
2307 return;
2308 }
2310
2311 if (CloseTransientFile(fd) != 0)
2312 {
2313 int save_errno = errno;
2314
2316 errno = save_errno;
2317 ereport(elevel,
2319 errmsg("could not close file \"%s\": %m",
2320 tmppath)));
2321 return;
2322 }
2323
2324 /* rename to permanent file, fsync file and directory */
2325 if (rename(tmppath, path) != 0)
2326 {
2327 int save_errno = errno;
2328
2330 errno = save_errno;
2331 ereport(elevel,
2333 errmsg("could not rename file \"%s\" to \"%s\": %m",
2334 tmppath, path)));
2335 return;
2336 }
2337
2338 /*
2339 * Check CreateSlotOnDisk() for the reasoning of using a critical section.
2340 */
2342
2343 fsync_fname(path, false);
2344 fsync_fname(dir, true);
2346
2348
2349 /*
2350 * Successfully wrote, unset dirty bit, unless somebody dirtied again
2351 * already and remember the confirmed_flush LSN value.
2352 */
2353 SpinLockAcquire(&slot->mutex);
2354 if (!slot->just_dirtied)
2355 slot->dirty = false;
2357 SpinLockRelease(&slot->mutex);
2358
2360}
2361
2362/*
2363 * Load a single slot from disk into memory.
2364 */
2365static void
2367{
2369 int i;
2370 char slotdir[MAXPGPATH + sizeof(PG_REPLSLOT_DIR)];
2371 char path[MAXPGPATH + sizeof(PG_REPLSLOT_DIR) + 10];
2372 int fd;
2373 bool restored = false;
2374 int readBytes;
2375 pg_crc32c checksum;
2376 TimestampTz now = 0;
2377
2378 /* no need to lock here, no concurrent access allowed yet */
2379
2380 /* delete temp file if it exists */
2381 sprintf(slotdir, "%s/%s", PG_REPLSLOT_DIR, name);
2382 sprintf(path, "%s/state.tmp", slotdir);
2383 if (unlink(path) < 0 && errno != ENOENT)
2384 ereport(PANIC,
2386 errmsg("could not remove file \"%s\": %m", path)));
2387
2388 sprintf(path, "%s/state", slotdir);
2389
2390 elog(DEBUG1, "restoring replication slot from \"%s\"", path);
2391
2392 /* on some operating systems fsyncing a file requires O_RDWR */
2393 fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
2394
2395 /*
2396 * We do not need to handle this as we are rename()ing the directory into
2397 * place only after we fsync()ed the state file.
2398 */
2399 if (fd < 0)
2400 ereport(PANIC,
2402 errmsg("could not open file \"%s\": %m", path)));
2403
2404 /*
2405 * Sync state file before we're reading from it. We might have crashed
2406 * while it wasn't synced yet and we shouldn't continue on that basis.
2407 */
2408 pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC);
2409 if (pg_fsync(fd) != 0)
2410 ereport(PANIC,
2412 errmsg("could not fsync file \"%s\": %m",
2413 path)));
2415
2416 /* Also sync the parent directory */
2418 fsync_fname(slotdir, true);
2420
2421 /* read part of statefile that's guaranteed to be version independent */
2422 pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
2423 readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
2425 if (readBytes != ReplicationSlotOnDiskConstantSize)
2426 {
2427 if (readBytes < 0)
2428 ereport(PANIC,
2430 errmsg("could not read file \"%s\": %m", path)));
2431 else
2432 ereport(PANIC,
2434 errmsg("could not read file \"%s\": read %d of %zu",
2435 path, readBytes,
2437 }
2438
2439 /* verify magic */
2440 if (cp.magic != SLOT_MAGIC)
2441 ereport(PANIC,
2443 errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
2444 path, cp.magic, SLOT_MAGIC)));
2445
2446 /* verify version */
2447 if (cp.version != SLOT_VERSION)
2448 ereport(PANIC,
2450 errmsg("replication slot file \"%s\" has unsupported version %u",
2451 path, cp.version)));
2452
2453 /* boundary check on length */
2455 ereport(PANIC,
2457 errmsg("replication slot file \"%s\" has corrupted length %u",
2458 path, cp.length)));
2459
2460 /* Now that we know the size, read the entire file */
2461 pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
2462 readBytes = read(fd,
2463 (char *) &cp + ReplicationSlotOnDiskConstantSize,
2464 cp.length);
2466 if (readBytes != cp.length)
2467 {
2468 if (readBytes < 0)
2469 ereport(PANIC,
2471 errmsg("could not read file \"%s\": %m", path)));
2472 else
2473 ereport(PANIC,
2475 errmsg("could not read file \"%s\": read %d of %zu",
2476 path, readBytes, (Size) cp.length)));
2477 }
2478
2479 if (CloseTransientFile(fd) != 0)
2480 ereport(PANIC,
2482 errmsg("could not close file \"%s\": %m", path)));
2483
2484 /* now verify the CRC */
2485 INIT_CRC32C(checksum);
2486 COMP_CRC32C(checksum,
2489 FIN_CRC32C(checksum);
2490
2491 if (!EQ_CRC32C(checksum, cp.checksum))
2492 ereport(PANIC,
2493 (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
2494 path, checksum, cp.checksum)));
2495
2496 /*
2497 * If we crashed with an ephemeral slot active, don't restore but delete
2498 * it.
2499 */
2501 {
2502 if (!rmtree(slotdir, true))
2503 {
2505 (errmsg("could not remove directory \"%s\"",
2506 slotdir)));
2507 }
2509 return;
2510 }
2511
2512 /*
2513 * Verify that requirements for the specific slot type are met. That's
2514 * important because if these aren't met we're not guaranteed to retain
2515 * all the necessary resources for the slot.
2516 *
2517 * NB: We have to do so *after* the above checks for ephemeral slots,
2518 * because otherwise a slot that shouldn't exist anymore could prevent
2519 * restarts.
2520 *
2521 * NB: Changing the requirements here also requires adapting
2522 * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
2523 */
2524 if (cp.slotdata.database != InvalidOid)
2525 {
2527 ereport(FATAL,
2528 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2529 errmsg("logical replication slot \"%s\" exists, but \"wal_level\" < \"logical\"",
2530 NameStr(cp.slotdata.name)),
2531 errhint("Change \"wal_level\" to be \"logical\" or higher.")));
2532
2533 /*
2534 * In standby mode, the hot standby must be enabled. This check is
2535 * necessary to ensure logical slots are invalidated when they become
2536 * incompatible due to insufficient wal_level. Otherwise, if the
2537 * primary reduces wal_level < logical while hot standby is disabled,
2538 * logical slots would remain valid even after promotion.
2539 */
2541 ereport(FATAL,
2542 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2543 errmsg("logical replication slot \"%s\" exists on the standby, but \"hot_standby\" = \"off\"",
2544 NameStr(cp.slotdata.name)),
2545 errhint("Change \"hot_standby\" to be \"on\".")));
2546 }
2547 else if (wal_level < WAL_LEVEL_REPLICA)
2548 ereport(FATAL,
2549 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2550 errmsg("physical replication slot \"%s\" exists, but \"wal_level\" < \"replica\"",
2551 NameStr(cp.slotdata.name)),
2552 errhint("Change \"wal_level\" to be \"replica\" or higher.")));
2553
2554 /* nothing can be active yet, don't lock anything */
2555 for (i = 0; i < max_replication_slots; i++)
2556 {
2557 ReplicationSlot *slot;
2558
2560
2561 if (slot->in_use)
2562 continue;
2563
2564 /* restore the entire set of persistent data */
2565 memcpy(&slot->data, &cp.slotdata,
2567
2568 /* initialize in memory state */
2569 slot->effective_xmin = cp.slotdata.xmin;
2572
2577
2578 slot->in_use = true;
2579 slot->active_pid = 0;
2580
2581 /*
2582 * Set the time since the slot has become inactive after loading the
2583 * slot from the disk into memory. Whoever acquires the slot i.e.
2584 * makes the slot active will reset it. Use the same inactive_since
2585 * time for all the slots.
2586 */
2587 if (now == 0)
2589
2591
2592 restored = true;
2593 break;
2594 }
2595
2596 if (!restored)
2597 ereport(FATAL,
2598 (errmsg("too many replication slots active before shutdown"),
2599 errhint("Increase \"max_replication_slots\" and try again.")));
2600}
2601
2602/*
2603 * Maps an invalidation reason for a replication slot to
2604 * ReplicationSlotInvalidationCause.
2605 */
2607GetSlotInvalidationCause(const char *cause_name)
2608{
2609 Assert(cause_name);
2610
2611 /* Search lookup table for the cause having this name */
2612 for (int i = 0; i <= RS_INVAL_MAX_CAUSES; i++)
2613 {
2614 if (strcmp(SlotInvalidationCauses[i].cause_name, cause_name) == 0)
2616 }
2617
2618 Assert(false);
2619 return RS_INVAL_NONE; /* to keep compiler quiet */
2620}
2621
2622/*
2623 * Maps an ReplicationSlotInvalidationCause to the invalidation
2624 * reason for a replication slot.
2625 */
2626const char *
2628{
2629 /* Search lookup table for the name of this cause */
2630 for (int i = 0; i <= RS_INVAL_MAX_CAUSES; i++)
2631 {
2632 if (SlotInvalidationCauses[i].cause == cause)
2634 }
2635
2636 Assert(false);
2637 return "none"; /* to keep compiler quiet */
2638}
2639
2640/*
2641 * A helper function to validate slots specified in GUC synchronized_standby_slots.
2642 *
2643 * The rawname will be parsed, and the result will be saved into *elemlist.
2644 */
2645static bool
2646validate_sync_standby_slots(char *rawname, List **elemlist)
2647{
2648 bool ok;
2649
2650 /* Verify syntax and parse string into a list of identifiers */
2651 ok = SplitIdentifierString(rawname, ',', elemlist);
2652
2653 if (!ok)
2654 {
2655 GUC_check_errdetail("List syntax is invalid.");
2656 }
2657 else if (MyProc)
2658 {
2659 /*
2660 * Check that each specified slot exist and is physical.
2661 *
2662 * Because we need an LWLock, we cannot do this on processes without a
2663 * PGPROC, so we skip it there; but see comments in
2664 * StandbySlotsHaveCaughtup() as to why that's not a problem.
2665 */
2666 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
2667
2668 foreach_ptr(char, name, *elemlist)
2669 {
2670 ReplicationSlot *slot;
2671
2672 slot = SearchNamedReplicationSlot(name, false);
2673
2674 if (!slot)
2675 {
2676 GUC_check_errdetail("Replication slot \"%s\" does not exist.",
2677 name);
2678 ok = false;
2679 break;
2680 }
2681
2682 if (!SlotIsPhysical(slot))
2683 {
2684 GUC_check_errdetail("\"%s\" is not a physical replication slot.",
2685 name);
2686 ok = false;
2687 break;
2688 }
2689 }
2690
2691 LWLockRelease(ReplicationSlotControlLock);
2692 }
2693
2694 return ok;
2695}
2696
2697/*
2698 * GUC check_hook for synchronized_standby_slots
2699 */
2700bool
2702{
2703 char *rawname;
2704 char *ptr;
2705 List *elemlist;
2706 int size;
2707 bool ok;
2709
2710 if ((*newval)[0] == '\0')
2711 return true;
2712
2713 /* Need a modifiable copy of the GUC string */
2714 rawname = pstrdup(*newval);
2715
2716 /* Now verify if the specified slots exist and have correct type */
2717 ok = validate_sync_standby_slots(rawname, &elemlist);
2718
2719 if (!ok || elemlist == NIL)
2720 {
2721 pfree(rawname);
2722 list_free(elemlist);
2723 return ok;
2724 }
2725
2726 /* Compute the size required for the SyncStandbySlotsConfigData struct */
2727 size = offsetof(SyncStandbySlotsConfigData, slot_names);
2728 foreach_ptr(char, slot_name, elemlist)
2729 size += strlen(slot_name) + 1;
2730
2731 /* GUC extra value must be guc_malloc'd, not palloc'd */
2732 config = (SyncStandbySlotsConfigData *) guc_malloc(LOG, size);
2733
2734 /* Transform the data into SyncStandbySlotsConfigData */
2735 config->nslotnames = list_length(elemlist);
2736
2737 ptr = config->slot_names;
2738 foreach_ptr(char, slot_name, elemlist)
2739 {
2740 strcpy(ptr, slot_name);
2741 ptr += strlen(slot_name) + 1;
2742 }
2743
2744 *extra = config;
2745
2746 pfree(rawname);
2747 list_free(elemlist);
2748 return true;
2749}
2750
2751/*
2752 * GUC assign_hook for synchronized_standby_slots
2753 */
2754void
2756{
2757 /*
2758 * The standby slots may have changed, so we must recompute the oldest
2759 * LSN.
2760 */
2762
2764}
2765
2766/*
2767 * Check if the passed slot_name is specified in the synchronized_standby_slots GUC.
2768 */
2769bool
2770SlotExistsInSyncStandbySlots(const char *slot_name)
2771{
2772 const char *standby_slot_name;
2773
2774 /* Return false if there is no value in synchronized_standby_slots */
2776 return false;
2777
2778 /*
2779 * XXX: We are not expecting this list to be long so a linear search
2780 * shouldn't hurt but if that turns out not to be true then we can cache
2781 * this information for each WalSender as well.
2782 */
2783 standby_slot_name = synchronized_standby_slots_config->slot_names;
2784 for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
2785 {
2786 if (strcmp(standby_slot_name, slot_name) == 0)
2787 return true;
2788
2789 standby_slot_name += strlen(standby_slot_name) + 1;
2790 }
2791
2792 return false;
2793}
2794
2795/*
2796 * Return true if the slots specified in synchronized_standby_slots have caught up to
2797 * the given WAL location, false otherwise.
2798 *
2799 * The elevel parameter specifies the error level used for logging messages
2800 * related to slots that do not exist, are invalidated, or are inactive.
2801 */
2802bool
2803StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
2804{
2805 const char *name;
2806 int caught_up_slot_num = 0;
2807 XLogRecPtr min_restart_lsn = InvalidXLogRecPtr;
2808
2809 /*
2810 * Don't need to wait for the standbys to catch up if there is no value in
2811 * synchronized_standby_slots.
2812 */
2814 return true;
2815
2816 /*
2817 * Don't need to wait for the standbys to catch up if we are on a standby
2818 * server, since we do not support syncing slots to cascading standbys.
2819 */
2820 if (RecoveryInProgress())
2821 return true;
2822
2823 /*
2824 * Don't need to wait for the standbys to catch up if they are already
2825 * beyond the specified WAL location.
2826 */
2828 ss_oldest_flush_lsn >= wait_for_lsn)
2829 return true;
2830
2831 /*
2832 * To prevent concurrent slot dropping and creation while filtering the
2833 * slots, take the ReplicationSlotControlLock outside of the loop.
2834 */
2835 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
2836
2838 for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
2839 {
2840 XLogRecPtr restart_lsn;
2841 bool invalidated;
2842 bool inactive;
2843 ReplicationSlot *slot;
2844
2845 slot = SearchNamedReplicationSlot(name, false);
2846
2847 /*
2848 * If a slot name provided in synchronized_standby_slots does not
2849 * exist, report a message and exit the loop.
2850 *
2851 * Though validate_sync_standby_slots (the GUC check_hook) tries to
2852 * avoid this, it can nonetheless happen because the user can specify
2853 * a nonexistent slot name before server startup. That function cannot
2854 * validate such a slot during startup, as ReplicationSlotCtl is not
2855 * initialized by then. Also, the user might have dropped one slot.
2856 */
2857 if (!slot)
2858 {
2859 ereport(elevel,
2860 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2861 errmsg("replication slot \"%s\" specified in parameter \"%s\" does not exist",
2862 name, "synchronized_standby_slots"),
2863 errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
2864 name),
2865 errhint("Create the replication slot \"%s\" or amend parameter \"%s\".",
2866 name, "synchronized_standby_slots"));
2867 break;
2868 }
2869
2870 /* Same as above: if a slot is not physical, exit the loop. */
2871 if (SlotIsLogical(slot))
2872 {
2873 ereport(elevel,
2874 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2875 errmsg("cannot specify logical replication slot \"%s\" in parameter \"%s\"",
2876 name, "synchronized_standby_slots"),
2877 errdetail("Logical replication is waiting for correction on replication slot \"%s\".",
2878 name),
2879 errhint("Remove the logical replication slot \"%s\" from parameter \"%s\".",
2880 name, "synchronized_standby_slots"));
2881 break;
2882 }
2883
2884 SpinLockAcquire(&slot->mutex);
2885 restart_lsn = slot->data.restart_lsn;
2886 invalidated = slot->data.invalidated != RS_INVAL_NONE;
2887 inactive = slot->active_pid == 0;
2888 SpinLockRelease(&slot->mutex);
2889
2890 if (invalidated)
2891 {
2892 /* Specified physical slot has been invalidated */
2893 ereport(elevel,
2894 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2895 errmsg("physical replication slot \"%s\" specified in parameter \"%s\" has been invalidated",
2896 name, "synchronized_standby_slots"),
2897 errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
2898 name),
2899 errhint("Drop and recreate the replication slot \"%s\", or amend parameter \"%s\".",
2900 name, "synchronized_standby_slots"));
2901 break;
2902 }
2903
2904 if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn)
2905 {
2906 /* Log a message if no active_pid for this physical slot */
2907 if (inactive)
2908 ereport(elevel,
2909 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2910 errmsg("replication slot \"%s\" specified in parameter \"%s\" does not have active_pid",
2911 name, "synchronized_standby_slots"),
2912 errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
2913 name),
2914 errhint("Start the standby associated with the replication slot \"%s\", or amend parameter \"%s\".",
2915 name, "synchronized_standby_slots"));
2916
2917 /* Continue if the current slot hasn't caught up. */
2918 break;
2919 }
2920
2921 Assert(restart_lsn >= wait_for_lsn);
2922
2923 if (XLogRecPtrIsInvalid(min_restart_lsn) ||
2924 min_restart_lsn > restart_lsn)
2925 min_restart_lsn = restart_lsn;
2926
2927 caught_up_slot_num++;
2928
2929 name += strlen(name) + 1;
2930 }
2931
2932 LWLockRelease(ReplicationSlotControlLock);
2933
2934 /*
2935 * Return false if not all the standbys have caught up to the specified
2936 * WAL location.
2937 */
2938 if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
2939 return false;
2940
2941 /* The ss_oldest_flush_lsn must not retreat. */
2943 min_restart_lsn >= ss_oldest_flush_lsn);
2944
2945 ss_oldest_flush_lsn = min_restart_lsn;
2946
2947 return true;
2948}
2949
2950/*
2951 * Wait for physical standbys to confirm receiving the given lsn.
2952 *
2953 * Used by logical decoding SQL functions. It waits for physical standbys
2954 * corresponding to the physical slots specified in the synchronized_standby_slots GUC.
2955 */
2956void
2958{
2959 /*
2960 * Don't need to wait for the standby to catch up if the current acquired
2961 * slot is not a logical failover slot, or there is no value in
2962 * synchronized_standby_slots.
2963 */
2965 return;
2966
2968
2969 for (;;)
2970 {
2972
2974 {
2975 ConfigReloadPending = false;
2977 }
2978
2979 /* Exit if done waiting for every slot. */
2980 if (StandbySlotsHaveCaughtup(wait_for_lsn, WARNING))
2981 break;
2982
2983 /*
2984 * Wait for the slots in the synchronized_standby_slots to catch up,
2985 * but use a timeout (1s) so we can also check if the
2986 * synchronized_standby_slots has been changed.
2987 */
2989 WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION);
2990 }
2991
2993}
2994
2995/*
2996 * GUC check_hook for idle_replication_slot_timeout
2997 *
2998 * The value of idle_replication_slot_timeout must be set to 0 during
2999 * a binary upgrade. See start_postmaster() in pg_upgrade for more details.
3000 */
3001bool
3003{
3004 if (IsBinaryUpgrade && *newval != 0)
3005 {
3006 GUC_check_errdetail("\"%s\" must be set to 0 during binary upgrade mode.",
3007 "idle_replication_slot_timeout");
3008 return false;
3009 }
3010
3011 return true;
3012}
void TimestampDifference(TimestampTz start_time, TimestampTz stop_time, long *secs, int *microsecs)
Definition: timestamp.c:1720
bool TimestampDifferenceExceedsSeconds(TimestampTz start_time, TimestampTz stop_time, int threshold_sec)
Definition: timestamp.c:1794
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1644
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1608
#define NameStr(name)
Definition: c.h:717
#define ngettext(s, p, n)
Definition: c.h:1152
#define PG_USED_FOR_ASSERTS_ONLY
Definition: c.h:224
#define PG_BINARY
Definition: c.h:1244
#define FLEXIBLE_ARRAY_MEMBER
Definition: c.h:434
#define pg_unreachable()
Definition: c.h:332
uint32_t uint32
Definition: c.h:502
#define lengthof(array)
Definition: c.h:759
#define MemSet(start, val, len)
Definition: c.h:991
uint32 TransactionId
Definition: c.h:623
size_t Size
Definition: c.h:576
bool ConditionVariableCancelSleep(void)
bool ConditionVariableTimedSleep(ConditionVariable *cv, long timeout, uint32 wait_event_info)
void ConditionVariableBroadcast(ConditionVariable *cv)
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
void ConditionVariableInit(ConditionVariable *cv)
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
int64 TimestampTz
Definition: timestamp.h:39
#define SECS_PER_MINUTE
Definition: timestamp.h:128
int errdetail_internal(const char *fmt,...)
Definition: elog.c:1230
int errcode_for_file_access(void)
Definition: elog.c:876
int errdetail(const char *fmt,...)
Definition: elog.c:1203
int errhint(const char *fmt,...)
Definition: elog.c:1317
int errcode(int sqlerrcode)
Definition: elog.c:853
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define _(x)
Definition: elog.c:90
#define LOG
Definition: elog.h:31
#define FATAL
Definition: elog.h:41
#define WARNING
Definition: elog.h:36
#define PANIC
Definition: elog.h:42
#define DEBUG1
Definition: elog.h:30
#define ERROR
Definition: elog.h:39
#define elog(elevel,...)
Definition: elog.h:225
#define ereport(elevel,...)
Definition: elog.h:149
int MakePGDirectory(const char *directoryName)
Definition: fd.c:3936
int FreeDir(DIR *dir)
Definition: fd.c:2983
int CloseTransientFile(int fd)
Definition: fd.c:2831
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:755
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2865
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2931
int pg_fsync(int fd)
Definition: fd.c:385
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2655
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
Definition: file_utils.c:526
PGFileType
Definition: file_utils.h:19
@ PGFILETYPE_DIR
Definition: file_utils.h:23
@ PGFILETYPE_ERROR
Definition: file_utils.h:20
bool IsBinaryUpgrade
Definition: globals.c:120
int MyProcPid
Definition: globals.c:46
bool IsUnderPostmaster
Definition: globals.c:119
Oid MyDatabaseId
Definition: globals.c:93
void ProcessConfigFile(GucContext context)
Definition: guc-file.l:120
void * guc_malloc(int elevel, size_t size)
Definition: guc.c:638
#define newval
#define GUC_check_errdetail
Definition: guc.h:481
GucSource
Definition: guc.h:112
@ PGC_SIGHUP
Definition: guc.h:75
Assert(PointerIsAligned(start, uint64))
#define IS_INJECTION_POINT_ATTACHED(name)
#define write(a, b, c)
Definition: win32.h:14
#define read(a, b, c)
Definition: win32.h:13
volatile sig_atomic_t ConfigReloadPending
Definition: interrupt.c:27
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:337
int i
Definition: isn.c:74
void list_free(List *list)
Definition: list.c:1546
bool LWLockHeldByMe(LWLock *lock)
Definition: lwlock.c:1967
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1179
bool LWLockHeldByMeInMode(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:2011
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1899
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:718
@ LWTRANCHE_REPLICATION_SLOT_IO
Definition: lwlock.h:191
@ LW_SHARED
Definition: lwlock.h:115
@ LW_EXCLUSIVE
Definition: lwlock.h:114
char * pstrdup(const char *in)
Definition: mcxt.c:1699
void pfree(void *pointer)
Definition: mcxt.c:1524
#define START_CRIT_SECTION()
Definition: miscadmin.h:149
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
@ B_STARTUP
Definition: miscadmin.h:363
#define END_CRIT_SECTION()
Definition: miscadmin.h:151
Oid GetUserId(void)
Definition: miscinit.c:517
BackendType MyBackendType
Definition: miscinit.c:64
bool has_rolreplication(Oid roleid)
Definition: miscinit.c:736
void namestrcpy(Name name, const char *str)
Definition: name.c:233
void * arg
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:41
#define NAMEDATALEN
#define MAXPGPATH
uint32 pg_crc32c
Definition: pg_crc32c.h:38
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:98
#define EQ_CRC32C(c1, c2)
Definition: pg_crc32c.h:42
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:103
static int list_length(const List *l)
Definition: pg_list.h:152
#define NIL
Definition: pg_list.h:68
#define foreach_ptr(type, var, lst)
Definition: pg_list.h:469
static bool two_phase
static rewind_source * source
Definition: pg_rewind.c:89
void pgstat_create_replslot(ReplicationSlot *slot)
void pgstat_acquire_replslot(ReplicationSlot *slot)
void pgstat_drop_replslot(ReplicationSlot *slot)
#define sprintf
Definition: port.h:241
#define snprintf
Definition: port.h:239
uintptr_t Datum
Definition: postgres.h:69
#define InvalidOid
Definition: postgres_ext.h:37
unsigned int Oid
Definition: postgres_ext.h:32
static int fd(const char *x, int i)
Definition: preproc-init.c:105
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3943
#define INVALID_PROC_NUMBER
Definition: procnumber.h:26
int SendProcSignal(pid_t pid, ProcSignalReason reason, ProcNumber procNumber)
Definition: procsignal.c:283
@ PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT
Definition: procsignal.h:46
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:50
Size add_size(Size s1, Size s2)
Definition: shmem.c:488
Size mul_size(Size s1, Size s2)
Definition: shmem.c:505
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:382
int ReplicationSlotIndex(ReplicationSlot *slot)
Definition: slot.c:512
void ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid)
Definition: slot.c:559
static const SlotInvalidationCauseMap SlotInvalidationCauses[]
Definition: slot.c:112
char * synchronized_standby_slots
Definition: slot.c:163
void assign_synchronized_standby_slots(const char *newval, void *extra)
Definition: slot.c:2755
#define ReplicationSlotOnDiskChecksummedSize
Definition: slot.c:134
void CheckPointReplicationSlots(bool is_shutdown)
Definition: slot.c:2032
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency, bool two_phase, bool failover, bool synced)
Definition: slot.c:324
void ReplicationSlotDropAcquired(void)
Definition: slot.c:919
void ReplicationSlotMarkDirty(void)
Definition: slot.c:1061
void ReplicationSlotReserveWal(void)
Definition: slot.c:1452
bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
Definition: slot.c:1263
struct SlotInvalidationCauseMap SlotInvalidationCauseMap
static XLogRecPtr ss_oldest_flush_lsn
Definition: slot.c:172
static ReplicationSlotInvalidationCause DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, TransactionId initial_effective_xmin, TransactionId initial_catalog_effective_xmin, XLogRecPtr initial_restart_lsn, TimestampTz *inactive_since, TimestampTz now)
Definition: slot.c:1629
void ReplicationSlotsDropDBSlots(Oid dboid)
Definition: slot.c:1321
#define ReplicationSlotOnDiskNotChecksummedSize
Definition: slot.c:131
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void)
Definition: slot.c:1205
ReplicationSlotInvalidationCause GetSlotInvalidationCause(const char *cause_name)
Definition: slot.c:2607
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:1100
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:2366
int idle_replication_slot_timeout_mins
Definition: slot.c:157
void ReplicationSlotPersist(void)
Definition: slot.c:1078
static void ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, bool terminating, int pid, NameData slotname, XLogRecPtr restart_lsn, XLogRecPtr oldestLSN, TransactionId snapshotConflictHorizon, long slot_idle_seconds)
Definition: slot.c:1528
ReplicationSlot * MyReplicationSlot
Definition: slot.c:147
static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
Definition: slot.c:2211
void ReplicationSlotDrop(const char *name, bool nowait)
Definition: slot.c:814
bool SlotExistsInSyncStandbySlots(const char *slot_name)
Definition: slot.c:2770
static bool validate_sync_standby_slots(char *rawname, List **elemlist)
Definition: slot.c:2646
void ReplicationSlotSave(void)
Definition: slot.c:1043
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
Definition: slot.c:479
static void CreateSlotOnDisk(ReplicationSlot *slot)
Definition: slot.c:2150
#define ReplicationSlotOnDiskV2Size
Definition: slot.c:137
void CheckSlotPermissions(void)
Definition: slot.c:1435
bool ReplicationSlotName(int index, Name name)
Definition: slot.c:528
bool check_synchronized_standby_slots(char **newval, void **extra, GucSource source)
Definition: slot.c:2701
void ReplicationSlotsShmemInit(void)
Definition: slot.c:204
void ReplicationSlotAlter(const char *name, const bool *failover, const bool *two_phase)
Definition: slot.c:837
void ReplicationSlotRelease(void)
Definition: slot.c:686
int max_replication_slots
Definition: slot.c:150
StaticAssertDecl(lengthof(SlotInvalidationCauses)==(RS_INVAL_MAX_CAUSES+1), "array length mismatch")
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:144
#define SLOT_VERSION
Definition: slot.c:141
struct ReplicationSlotOnDisk ReplicationSlotOnDisk
void WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
Definition: slot.c:2957
bool StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
Definition: slot.c:2803
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:1156
void ReplicationSlotCleanup(bool synced_only)
Definition: slot.c:775
void ReplicationSlotInitialize(void)
Definition: slot.c:239
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:936
void StartupReplicationSlots(void)
Definition: slot.c:2089
static bool InvalidatePossiblyObsoleteSlot(uint32 possible_causes, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, bool *invalidated)
Definition: slot.c:1718
static bool CanInvalidateIdleSlot(ReplicationSlot *s)
Definition: slot.c:1613
void CheckSlotRequirements(void)
Definition: slot.c:1413
#define SLOT_MAGIC
Definition: slot.c:140
bool InvalidateObsoleteReplicationSlots(uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, TransactionId snapshotConflictHorizon)
Definition: slot.c:1976
static SyncStandbySlotsConfigData * synchronized_standby_slots_config
Definition: slot.c:166
#define ReplicationSlotOnDiskConstantSize
Definition: slot.c:128
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:186
const char * GetSlotInvalidationCauseName(ReplicationSlotInvalidationCause cause)
Definition: slot.c:2627
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:267
static void ReplicationSlotShmemExit(int code, Datum arg)
Definition: slot.c:248
bool check_idle_replication_slot_timeout(int *newval, void **extra, GucSource source)
Definition: slot.c:3002
#define RS_INVAL_MAX_CAUSES
Definition: slot.h:65
ReplicationSlotPersistency
Definition: slot.h:37
@ RS_PERSISTENT
Definition: slot.h:38
@ RS_EPHEMERAL
Definition: slot.h:39
@ RS_TEMPORARY
Definition: slot.h:40
#define SlotIsPhysical(slot)
Definition: slot.h:220
#define PG_REPLSLOT_DIR
Definition: slot.h:21
ReplicationSlotInvalidationCause
Definition: slot.h:52
@ RS_INVAL_WAL_REMOVED
Definition: slot.h:55
@ RS_INVAL_IDLE_TIMEOUT
Definition: slot.h:61
@ RS_INVAL_HORIZON
Definition: slot.h:57
@ RS_INVAL_WAL_LEVEL
Definition: slot.h:59
@ RS_INVAL_NONE
Definition: slot.h:53
#define SlotIsLogical(slot)
Definition: slot.h:221
static void ReplicationSlotSetInactiveSince(ReplicationSlot *s, TimestampTz ts, bool acquire_lock)
Definition: slot.h:239
bool IsSyncingReplicationSlots(void)
Definition: slotsync.c:1647
#define SpinLockInit(lock)
Definition: spin.h:57
#define SpinLockRelease(lock)
Definition: spin.h:61
#define SpinLockAcquire(lock)
Definition: spin.h:59
PGPROC * MyProc
Definition: proc.c:66
PROC_HDR * ProcGlobal
Definition: proc.c:78
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:1281
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:30
bool pg_str_endswith(const char *str, const char *end)
Definition: string.c:31
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:145
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:230
void initStringInfo(StringInfo str)
Definition: stringinfo.c:97
Definition: dirent.c:26
Definition: pg_list.h:54
uint8 statusFlags
Definition: proc.h:243
int pgxactoff
Definition: proc.h:185
uint8 * statusFlags
Definition: proc.h:387
ReplicationSlot replication_slots[1]
Definition: slot.h:232
uint32 version
Definition: slot.c:74
ReplicationSlotPersistentData slotdata
Definition: slot.c:82
pg_crc32c checksum
Definition: slot.c:71
TransactionId xmin
Definition: slot.h:89
TransactionId catalog_xmin
Definition: slot.h:97
XLogRecPtr confirmed_flush
Definition: slot.h:111
ReplicationSlotPersistency persistency
Definition: slot.h:81
ReplicationSlotInvalidationCause invalidated
Definition: slot.h:103
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:201
TransactionId effective_catalog_xmin
Definition: slot.h:182
slock_t mutex
Definition: slot.h:158
XLogRecPtr candidate_restart_valid
Definition: slot.h:202
XLogRecPtr last_saved_confirmed_flush
Definition: slot.h:210
pid_t active_pid
Definition: slot.h:164
bool in_use
Definition: slot.h:161
TransactionId effective_xmin
Definition: slot.h:181
bool just_dirtied
Definition: slot.h:167
XLogRecPtr candidate_restart_lsn
Definition: slot.h:203
LWLock io_in_progress_lock
Definition: slot.h:188
ConditionVariable active_cv
Definition: slot.h:191
TransactionId candidate_catalog_xmin
Definition: slot.h:200
bool dirty
Definition: slot.h:168
ReplicationSlotPersistentData data
Definition: slot.h:185
TimestampTz inactive_since
Definition: slot.h:217
const char * cause_name
Definition: slot.c:109
ReplicationSlotInvalidationCause cause
Definition: slot.c:108
char slot_names[FLEXIBLE_ARRAY_MEMBER]
Definition: slot.c:100
ConditionVariable wal_confirm_rcv_cv
Definition: dirent.h:10
char d_name[MAX_PATH]
Definition: dirent.h:15
Definition: type.h:96
Definition: c.h:712
unsigned short st_mode
Definition: win32_port.h:258
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:280
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
Definition: transam.c:299
#define InvalidTransactionId
Definition: transam.h:31
#define TransactionIdIsValid(xid)
Definition: transam.h:41
bool SplitIdentifierString(char *rawstring, char separator, List **namelist)
Definition: varlena.c:3525
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: wait_event.h:85
static void pgstat_report_wait_end(void)
Definition: wait_event.h:101
const char * name
bool am_walsender
Definition: walsender.c:116
bool log_replication_commands
Definition: walsender.c:126
WalSndCtlData * WalSndCtl
Definition: walsender.c:110
#define stat
Definition: win32_port.h:274
#define S_ISDIR(m)
Definition: win32_port.h:315
#define kill(pid, sig)
Definition: win32_port.h:493
bool RecoveryInProgress(void)
Definition: xlog.c:6380
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3764
bool EnableHotStandby
Definition: xlog.c:121
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:6483
int wal_level
Definition: xlog.c:131
int wal_segment_size
Definition: xlog.c:143
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2676
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:9470
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2790
@ WAL_LEVEL_REPLICA
Definition: xlog.h:75
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:76
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
uint64 XLogSegNo
Definition: xlogdefs.h:48
bool StandbyMode
Definition: xlogrecovery.c:148
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)