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-2020, PostgreSQL Global Development Group
8  *
9  *
10  * IDENTIFICATION
11  * src/backend/replication/slot.c
12  *
13  * NOTES
14  *
15  * Replication slots are used to keep state about replication streams
16  * originating from this cluster. Their primary purpose is to prevent the
17  * premature removal of WAL or of old tuple versions in a manner that would
18  * interfere with replication; they are also useful for monitoring purposes.
19  * Slots need to be permanent (to allow restarts), crash-safe, and allocatable
20  * on standbys (to support cascading setups). The requirement that slots be
21  * usable on standbys precludes storing them in the system catalogs.
22  *
23  * Each replication slot gets its own directory inside the $PGDATA/pg_replslot
24  * directory. Inside that directory the state file will contain the slot's
25  * own data. Additional data can be stored alongside that file if required.
26  * While the server is running, the state data is also cached in memory for
27  * efficiency.
28  *
29  * ReplicationSlotAllocationLock must be taken in exclusive mode to allocate
30  * or free a slot. ReplicationSlotControlLock must be taken in shared mode
31  * to iterate over the slots, and in exclusive mode to change the in_use flag
32  * of a slot. The remaining data in each slot is protected by its mutex.
33  *
34  *-------------------------------------------------------------------------
35  */
36 
37 #include "postgres.h"
38 
39 #include <unistd.h>
40 #include <sys/stat.h>
41 
42 #include "access/transam.h"
43 #include "access/xlog_internal.h"
44 #include "common/string.h"
45 #include "miscadmin.h"
46 #include "pgstat.h"
47 #include "replication/slot.h"
48 #include "storage/fd.h"
49 #include "storage/proc.h"
50 #include "storage/procarray.h"
51 #include "utils/builtins.h"
52 
53 /*
54  * Replication slot on-disk data structure.
55  */
56 typedef struct ReplicationSlotOnDisk
57 {
58  /* first part of this struct needs to be version independent */
59 
60  /* data not covered by checksum */
63 
64  /* data covered by checksum */
67 
68  /*
69  * The actual data in the slot that follows can differ based on the above
70  * 'version'.
71  */
72 
75 
76 /* size of version independent data */
77 #define ReplicationSlotOnDiskConstantSize \
78  offsetof(ReplicationSlotOnDisk, slotdata)
79 /* size of the part of the slot not covered by the checksum */
80 #define SnapBuildOnDiskNotChecksummedSize \
81  offsetof(ReplicationSlotOnDisk, version)
82 /* size of the part covered by the checksum */
83 #define SnapBuildOnDiskChecksummedSize \
84  sizeof(ReplicationSlotOnDisk) - SnapBuildOnDiskNotChecksummedSize
85 /* size of the slot data that is version dependent */
86 #define ReplicationSlotOnDiskV2Size \
87  sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize
88 
89 #define SLOT_MAGIC 0x1051CA1 /* format identifier */
90 #define SLOT_VERSION 2 /* version for new files */
91 
92 /* Control array for replication slot management */
94 
95 /* My backend's replication slot in the shared memory array */
97 
98 /* GUCs */
99 int max_replication_slots = 0; /* the maximum number of replication
100  * slots */
101 
102 static ReplicationSlot *SearchNamedReplicationSlot(const char *name);
104  const char *name, SlotAcquireBehavior behavior);
105 static void ReplicationSlotDropAcquired(void);
106 static void ReplicationSlotDropPtr(ReplicationSlot *slot);
107 
108 /* internal persistency functions */
109 static void RestoreSlotFromDisk(const char *name);
110 static void CreateSlotOnDisk(ReplicationSlot *slot);
111 static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel);
112 
113 /*
114  * Report shared-memory space needed by ReplicationSlotsShmemInit.
115  */
116 Size
118 {
119  Size size = 0;
120 
121  if (max_replication_slots == 0)
122  return size;
123 
124  size = offsetof(ReplicationSlotCtlData, replication_slots);
125  size = add_size(size,
127 
128  return size;
129 }
130 
131 /*
132  * Allocate and initialize shared memory for replication slots.
133  */
134 void
136 {
137  bool found;
138 
139  if (max_replication_slots == 0)
140  return;
141 
142  ReplicationSlotCtl = (ReplicationSlotCtlData *)
143  ShmemInitStruct("ReplicationSlot Ctl", ReplicationSlotsShmemSize(),
144  &found);
145 
146  if (!found)
147  {
148  int i;
149 
150  /* First time through, so initialize */
151  MemSet(ReplicationSlotCtl, 0, ReplicationSlotsShmemSize());
152 
153  for (i = 0; i < max_replication_slots; i++)
154  {
155  ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[i];
156 
157  /* everything else is zeroed by the memset above */
158  SpinLockInit(&slot->mutex);
162  }
163  }
164 }
165 
166 /*
167  * Check whether the passed slot name is valid and report errors at elevel.
168  *
169  * Slot names may consist out of [a-z0-9_]{1,NAMEDATALEN-1} which should allow
170  * the name to be used as a directory name on every supported OS.
171  *
172  * Returns whether the directory name is valid or not if elevel < ERROR.
173  */
174 bool
176 {
177  const char *cp;
178 
179  if (strlen(name) == 0)
180  {
181  ereport(elevel,
182  (errcode(ERRCODE_INVALID_NAME),
183  errmsg("replication slot name \"%s\" is too short",
184  name)));
185  return false;
186  }
187 
188  if (strlen(name) >= NAMEDATALEN)
189  {
190  ereport(elevel,
191  (errcode(ERRCODE_NAME_TOO_LONG),
192  errmsg("replication slot name \"%s\" is too long",
193  name)));
194  return false;
195  }
196 
197  for (cp = name; *cp; cp++)
198  {
199  if (!((*cp >= 'a' && *cp <= 'z')
200  || (*cp >= '0' && *cp <= '9')
201  || (*cp == '_')))
202  {
203  ereport(elevel,
204  (errcode(ERRCODE_INVALID_NAME),
205  errmsg("replication slot name \"%s\" contains invalid character",
206  name),
207  errhint("Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
208  return false;
209  }
210  }
211  return true;
212 }
213 
214 /*
215  * Create a new replication slot and mark it as used by this backend.
216  *
217  * name: Name of the slot
218  * db_specific: logical decoding is db specific; if the slot is going to
219  * be used for that pass true, otherwise false.
220  */
221 void
222 ReplicationSlotCreate(const char *name, bool db_specific,
223  ReplicationSlotPersistency persistency)
224 {
225  ReplicationSlot *slot = NULL;
226  int i;
227 
228  Assert(MyReplicationSlot == NULL);
229 
231 
232  /*
233  * If some other backend ran this code concurrently with us, we'd likely
234  * both allocate the same slot, and that would be bad. We'd also be at
235  * risk of missing a name collision. Also, we don't want to try to create
236  * a new slot while somebody's busy cleaning up an old one, because we
237  * might both be monkeying with the same directory.
238  */
239  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
240 
241  /*
242  * Check for name collision, and identify an allocatable slot. We need to
243  * hold ReplicationSlotControlLock in shared mode for this, so that nobody
244  * else can change the in_use flags while we're looking at them.
245  */
246  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
247  for (i = 0; i < max_replication_slots; i++)
248  {
249  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
250 
251  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
252  ereport(ERROR,
254  errmsg("replication slot \"%s\" already exists", name)));
255  if (!s->in_use && slot == NULL)
256  slot = s;
257  }
258  LWLockRelease(ReplicationSlotControlLock);
259 
260  /* If all slots are in use, we're out of luck. */
261  if (slot == NULL)
262  ereport(ERROR,
263  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
264  errmsg("all replication slots are in use"),
265  errhint("Free one or increase max_replication_slots.")));
266 
267  /*
268  * Since this slot is not in use, nobody should be looking at any part of
269  * it other than the in_use field unless they're trying to allocate it.
270  * And since we hold ReplicationSlotAllocationLock, nobody except us can
271  * be doing that. So it's safe to initialize the slot.
272  */
273  Assert(!slot->in_use);
274  Assert(slot->active_pid == 0);
275 
276  /* first initialize persistent data */
277  memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
278  StrNCpy(NameStr(slot->data.name), name, NAMEDATALEN);
279  slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
280  slot->data.persistency = persistency;
281 
282  /* and then data only present in shared memory */
283  slot->just_dirtied = false;
284  slot->dirty = false;
291 
292  /*
293  * Create the slot on disk. We haven't actually marked the slot allocated
294  * yet, so no special cleanup is required if this errors out.
295  */
296  CreateSlotOnDisk(slot);
297 
298  /*
299  * We need to briefly prevent any other backend from iterating over the
300  * slots while we flip the in_use flag. We also need to set the active
301  * flag while holding the ControlLock as otherwise a concurrent
302  * ReplicationSlotAcquire() could acquire the slot as well.
303  */
304  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
305 
306  slot->in_use = true;
307 
308  /* We can now mark the slot active, and that makes it our slot. */
309  SpinLockAcquire(&slot->mutex);
310  Assert(slot->active_pid == 0);
311  slot->active_pid = MyProcPid;
312  SpinLockRelease(&slot->mutex);
313  MyReplicationSlot = slot;
314 
315  LWLockRelease(ReplicationSlotControlLock);
316 
317  /*
318  * Now that the slot has been marked as in_use and active, it's safe to
319  * let somebody else try to allocate a slot.
320  */
321  LWLockRelease(ReplicationSlotAllocationLock);
322 
323  /* Let everybody know we've modified this slot */
325 }
326 
327 /*
328  * Search for the named replication slot.
329  *
330  * Return the replication slot if found, otherwise NULL.
331  *
332  * The caller must hold ReplicationSlotControlLock in shared mode.
333  */
334 static ReplicationSlot *
336 {
337  int i;
338  ReplicationSlot *slot = NULL;
339 
340  Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock,
341  LW_SHARED));
342 
343  for (i = 0; i < max_replication_slots; i++)
344  {
345  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
346 
347  if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
348  {
349  slot = s;
350  break;
351  }
352  }
353 
354  return slot;
355 }
356 
357 /*
358  * Find a previously created slot and mark it as used by this process.
359  *
360  * The return value is only useful if behavior is SAB_Inquire, in which
361  * it's zero if we successfully acquired the slot, -1 if the slot no longer
362  * exists, or the PID of the owning process otherwise. If behavior is
363  * SAB_Error, then trying to acquire an owned slot is an error.
364  * If SAB_Block, we sleep until the slot is released by the owning process.
365  */
366 int
368 {
369  return ReplicationSlotAcquireInternal(NULL, name, behavior);
370 }
371 
372 /*
373  * Mark the specified slot as used by this process.
374  *
375  * Only one of slot and name can be specified.
376  * If slot == NULL, search for the slot with the given name.
377  *
378  * See comments about the return value in ReplicationSlotAcquire().
379  */
380 static int
382  SlotAcquireBehavior behavior)
383 {
384  ReplicationSlot *s;
385  int active_pid;
386 
387  AssertArg((slot == NULL) ^ (name == NULL));
388 
389 retry:
390  Assert(MyReplicationSlot == NULL);
391 
392  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
393 
394  /*
395  * Search for the slot with the specified name if the slot to acquire is
396  * not given. If the slot is not found, we either return -1 or error out.
397  */
398  s = slot ? slot : SearchNamedReplicationSlot(name);
399  if (s == NULL || !s->in_use)
400  {
401  LWLockRelease(ReplicationSlotControlLock);
402 
403  if (behavior == SAB_Inquire)
404  return -1;
405  ereport(ERROR,
406  (errcode(ERRCODE_UNDEFINED_OBJECT),
407  errmsg("replication slot \"%s\" does not exist",
408  name ? name : NameStr(slot->data.name))));
409  }
410 
411  /*
412  * This is the slot we want; check if it's active under some other
413  * process. In single user mode, we don't need this check.
414  */
415  if (IsUnderPostmaster)
416  {
417  /*
418  * Get ready to sleep on the slot in case it is active if SAB_Block.
419  * (We may end up not sleeping, but we don't want to do this while
420  * holding the spinlock.)
421  */
422  if (behavior == SAB_Block)
424 
425  SpinLockAcquire(&s->mutex);
426  if (s->active_pid == 0)
427  s->active_pid = MyProcPid;
428  active_pid = s->active_pid;
429  SpinLockRelease(&s->mutex);
430  }
431  else
432  active_pid = MyProcPid;
433  LWLockRelease(ReplicationSlotControlLock);
434 
435  /*
436  * If we found the slot but it's already active in another process, we
437  * either error out, return the PID of the owning process, or retry
438  * after a short wait, as caller specified.
439  */
440  if (active_pid != MyProcPid)
441  {
442  if (behavior == SAB_Error)
443  ereport(ERROR,
444  (errcode(ERRCODE_OBJECT_IN_USE),
445  errmsg("replication slot \"%s\" is active for PID %d",
446  NameStr(s->data.name), active_pid)));
447  else if (behavior == SAB_Inquire)
448  return active_pid;
449 
450  /* Wait here until we get signaled, and then restart */
454  goto retry;
455  }
456  else if (behavior == SAB_Block)
457  ConditionVariableCancelSleep(); /* no sleep needed after all */
458 
459  /* Let everybody know we've modified this slot */
461 
462  /* We made this slot active, so it's ours now. */
463  MyReplicationSlot = s;
464 
465  /* success */
466  return 0;
467 }
468 
469 /*
470  * Release the replication slot that this backend considers to own.
471  *
472  * This or another backend can re-acquire the slot later.
473  * Resources this slot requires will be preserved.
474  */
475 void
477 {
479 
480  Assert(slot != NULL && slot->active_pid != 0);
481 
482  if (slot->data.persistency == RS_EPHEMERAL)
483  {
484  /*
485  * Delete the slot. There is no !PANIC case where this is allowed to
486  * fail, all that may happen is an incomplete cleanup of the on-disk
487  * data.
488  */
490  }
491 
492  /*
493  * If slot needed to temporarily restrain both data and catalog xmin to
494  * create the catalog snapshot, remove that temporary constraint.
495  * Snapshots can only be exported while the initial snapshot is still
496  * acquired.
497  */
498  if (!TransactionIdIsValid(slot->data.xmin) &&
500  {
501  SpinLockAcquire(&slot->mutex);
503  SpinLockRelease(&slot->mutex);
505  }
506 
507  if (slot->data.persistency == RS_PERSISTENT)
508  {
509  /*
510  * Mark persistent slot inactive. We're not freeing it, just
511  * disconnecting, but wake up others that may be waiting for it.
512  */
513  SpinLockAcquire(&slot->mutex);
514  slot->active_pid = 0;
515  SpinLockRelease(&slot->mutex);
517  }
518 
519  MyReplicationSlot = NULL;
520 
521  /* might not have been set when we've been a plain slot */
522  LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
524  LWLockRelease(ProcArrayLock);
525 }
526 
527 /*
528  * Cleanup all temporary slots created in current session.
529  */
530 void
532 {
533  int i;
534 
535  Assert(MyReplicationSlot == NULL);
536 
537 restart:
538  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
539  for (i = 0; i < max_replication_slots; i++)
540  {
541  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
542 
543  if (!s->in_use)
544  continue;
545 
546  SpinLockAcquire(&s->mutex);
547  if (s->active_pid == MyProcPid)
548  {
550  SpinLockRelease(&s->mutex);
551  LWLockRelease(ReplicationSlotControlLock); /* avoid deadlock */
552 
554 
556  goto restart;
557  }
558  else
559  SpinLockRelease(&s->mutex);
560  }
561 
562  LWLockRelease(ReplicationSlotControlLock);
563 }
564 
565 /*
566  * Permanently drop replication slot identified by the passed in name.
567  */
568 void
569 ReplicationSlotDrop(const char *name, bool nowait)
570 {
571  Assert(MyReplicationSlot == NULL);
572 
573  (void) ReplicationSlotAcquire(name, nowait ? SAB_Error : SAB_Block);
574 
576 }
577 
578 /*
579  * Permanently drop the currently acquired replication slot.
580  */
581 static void
583 {
585 
586  Assert(MyReplicationSlot != NULL);
587 
588  /* slot isn't acquired anymore */
589  MyReplicationSlot = NULL;
590 
592 }
593 
594 /*
595  * Permanently drop the replication slot which will be released by the point
596  * this function returns.
597  */
598 static void
600 {
601  char path[MAXPGPATH];
602  char tmppath[MAXPGPATH];
603 
604  /*
605  * If some other backend ran this code concurrently with us, we might try
606  * to delete a slot with a certain name while someone else was trying to
607  * create a slot with the same name.
608  */
609  LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
610 
611  /* Generate pathnames. */
612  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
613  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
614 
615  /*
616  * Rename the slot directory on disk, so that we'll no longer recognize
617  * this as a valid slot. Note that if this fails, we've got to mark the
618  * slot inactive before bailing out. If we're dropping an ephemeral or a
619  * temporary slot, we better never fail hard as the caller won't expect
620  * the slot to survive and this might get called during error handling.
621  */
622  if (rename(path, tmppath) == 0)
623  {
624  /*
625  * We need to fsync() the directory we just renamed and its parent to
626  * make sure that our changes are on disk in a crash-safe fashion. If
627  * fsync() fails, we can't be sure whether the changes are on disk or
628  * not. For now, we handle that by panicking;
629  * StartupReplicationSlots() will try to straighten it out after
630  * restart.
631  */
633  fsync_fname(tmppath, true);
634  fsync_fname("pg_replslot", true);
636  }
637  else
638  {
639  bool fail_softly = slot->data.persistency != RS_PERSISTENT;
640 
641  SpinLockAcquire(&slot->mutex);
642  slot->active_pid = 0;
643  SpinLockRelease(&slot->mutex);
644 
645  /* wake up anyone waiting on this slot */
647 
648  ereport(fail_softly ? WARNING : ERROR,
650  errmsg("could not rename file \"%s\" to \"%s\": %m",
651  path, tmppath)));
652  }
653 
654  /*
655  * The slot is definitely gone. Lock out concurrent scans of the array
656  * long enough to kill it. It's OK to clear the active PID here without
657  * grabbing the mutex because nobody else can be scanning the array here,
658  * and nobody can be attached to this slot and thus access it without
659  * scanning the array.
660  *
661  * Also wake up processes waiting for it.
662  */
663  LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
664  slot->active_pid = 0;
665  slot->in_use = false;
666  LWLockRelease(ReplicationSlotControlLock);
668 
669  /*
670  * Slot is dead and doesn't prevent resource removal anymore, recompute
671  * limits.
672  */
675 
676  /*
677  * If removing the directory fails, the worst thing that will happen is
678  * that the user won't be able to create a new slot with the same name
679  * until the next server restart. We warn about it, but that's all.
680  */
681  if (!rmtree(tmppath, true))
683  (errmsg("could not remove directory \"%s\"", tmppath)));
684 
685  /*
686  * We release this at the very end, so that nobody starts trying to create
687  * a slot while we're still cleaning up the detritus of the old one.
688  */
689  LWLockRelease(ReplicationSlotAllocationLock);
690 }
691 
692 /*
693  * Serialize the currently acquired slot's state from memory to disk, thereby
694  * guaranteeing the current state will survive a crash.
695  */
696 void
698 {
699  char path[MAXPGPATH];
700 
701  Assert(MyReplicationSlot != NULL);
702 
703  sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
704  SaveSlotToPath(MyReplicationSlot, path, ERROR);
705 }
706 
707 /*
708  * Signal that it would be useful if the currently acquired slot would be
709  * flushed out to disk.
710  *
711  * Note that the actual flush to disk can be delayed for a long time, if
712  * required for correctness explicitly do a ReplicationSlotSave().
713  */
714 void
716 {
718 
719  Assert(MyReplicationSlot != NULL);
720 
721  SpinLockAcquire(&slot->mutex);
722  MyReplicationSlot->just_dirtied = true;
723  MyReplicationSlot->dirty = true;
724  SpinLockRelease(&slot->mutex);
725 }
726 
727 /*
728  * Convert a slot that's marked as RS_EPHEMERAL to a RS_PERSISTENT slot,
729  * guaranteeing it will be there after an eventual crash.
730  */
731 void
733 {
735 
736  Assert(slot != NULL);
738 
739  SpinLockAcquire(&slot->mutex);
741  SpinLockRelease(&slot->mutex);
742 
745 }
746 
747 /*
748  * Compute the oldest xmin across all slots and store it in the ProcArray.
749  *
750  * If already_locked is true, ProcArrayLock has already been acquired
751  * exclusively.
752  */
753 void
755 {
756  int i;
758  TransactionId agg_catalog_xmin = InvalidTransactionId;
759 
760  Assert(ReplicationSlotCtl != NULL);
761 
762  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
763 
764  for (i = 0; i < max_replication_slots; i++)
765  {
766  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
767  TransactionId effective_xmin;
768  TransactionId effective_catalog_xmin;
769 
770  if (!s->in_use)
771  continue;
772 
773  SpinLockAcquire(&s->mutex);
774  effective_xmin = s->effective_xmin;
775  effective_catalog_xmin = s->effective_catalog_xmin;
776  SpinLockRelease(&s->mutex);
777 
778  /* check the data xmin */
779  if (TransactionIdIsValid(effective_xmin) &&
780  (!TransactionIdIsValid(agg_xmin) ||
781  TransactionIdPrecedes(effective_xmin, agg_xmin)))
782  agg_xmin = effective_xmin;
783 
784  /* check the catalog xmin */
785  if (TransactionIdIsValid(effective_catalog_xmin) &&
786  (!TransactionIdIsValid(agg_catalog_xmin) ||
787  TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
788  agg_catalog_xmin = effective_catalog_xmin;
789  }
790 
791  LWLockRelease(ReplicationSlotControlLock);
792 
793  ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
794 }
795 
796 /*
797  * Compute the oldest restart LSN across all slots and inform xlog module.
798  *
799  * Note: while max_slot_wal_keep_size is theoretically relevant for this
800  * purpose, we don't try to account for that, because this module doesn't
801  * know what to compare against.
802  */
803 void
805 {
806  int i;
807  XLogRecPtr min_required = InvalidXLogRecPtr;
808 
809  Assert(ReplicationSlotCtl != NULL);
810 
811  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
812  for (i = 0; i < max_replication_slots; i++)
813  {
814  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
815  XLogRecPtr restart_lsn;
816 
817  if (!s->in_use)
818  continue;
819 
820  SpinLockAcquire(&s->mutex);
821  restart_lsn = s->data.restart_lsn;
822  SpinLockRelease(&s->mutex);
823 
824  if (restart_lsn != InvalidXLogRecPtr &&
825  (min_required == InvalidXLogRecPtr ||
826  restart_lsn < min_required))
827  min_required = restart_lsn;
828  }
829  LWLockRelease(ReplicationSlotControlLock);
830 
831  XLogSetReplicationSlotMinimumLSN(min_required);
832 }
833 
834 /*
835  * Compute the oldest WAL LSN required by *logical* decoding slots..
836  *
837  * Returns InvalidXLogRecPtr if logical decoding is disabled or no logical
838  * slots exist.
839  *
840  * NB: this returns a value >= ReplicationSlotsComputeRequiredLSN(), since it
841  * ignores physical replication slots.
842  *
843  * The results aren't required frequently, so we don't maintain a precomputed
844  * value like we do for ComputeRequiredLSN() and ComputeRequiredXmin().
845  */
848 {
849  XLogRecPtr result = InvalidXLogRecPtr;
850  int i;
851 
852  if (max_replication_slots <= 0)
853  return InvalidXLogRecPtr;
854 
855  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
856 
857  for (i = 0; i < max_replication_slots; i++)
858  {
859  ReplicationSlot *s;
860  XLogRecPtr restart_lsn;
861 
862  s = &ReplicationSlotCtl->replication_slots[i];
863 
864  /* cannot change while ReplicationSlotCtlLock is held */
865  if (!s->in_use)
866  continue;
867 
868  /* we're only interested in logical slots */
869  if (!SlotIsLogical(s))
870  continue;
871 
872  /* read once, it's ok if it increases while we're checking */
873  SpinLockAcquire(&s->mutex);
874  restart_lsn = s->data.restart_lsn;
875  SpinLockRelease(&s->mutex);
876 
877  if (restart_lsn == InvalidXLogRecPtr)
878  continue;
879 
880  if (result == InvalidXLogRecPtr ||
881  restart_lsn < result)
882  result = restart_lsn;
883  }
884 
885  LWLockRelease(ReplicationSlotControlLock);
886 
887  return result;
888 }
889 
890 /*
891  * ReplicationSlotsCountDBSlots -- count the number of slots that refer to the
892  * passed database oid.
893  *
894  * Returns true if there are any slots referencing the database. *nslots will
895  * be set to the absolute number of slots in the database, *nactive to ones
896  * currently active.
897  */
898 bool
899 ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
900 {
901  int i;
902 
903  *nslots = *nactive = 0;
904 
905  if (max_replication_slots <= 0)
906  return false;
907 
908  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
909  for (i = 0; i < max_replication_slots; i++)
910  {
911  ReplicationSlot *s;
912 
913  s = &ReplicationSlotCtl->replication_slots[i];
914 
915  /* cannot change while ReplicationSlotCtlLock is held */
916  if (!s->in_use)
917  continue;
918 
919  /* only logical slots are database specific, skip */
920  if (!SlotIsLogical(s))
921  continue;
922 
923  /* not our database, skip */
924  if (s->data.database != dboid)
925  continue;
926 
927  /* count slots with spinlock held */
928  SpinLockAcquire(&s->mutex);
929  (*nslots)++;
930  if (s->active_pid != 0)
931  (*nactive)++;
932  SpinLockRelease(&s->mutex);
933  }
934  LWLockRelease(ReplicationSlotControlLock);
935 
936  if (*nslots > 0)
937  return true;
938  return false;
939 }
940 
941 /*
942  * ReplicationSlotsDropDBSlots -- Drop all db-specific slots relating to the
943  * passed database oid. The caller should hold an exclusive lock on the
944  * pg_database oid for the database to prevent creation of new slots on the db
945  * or replay from existing slots.
946  *
947  * Another session that concurrently acquires an existing slot on the target DB
948  * (most likely to drop it) may cause this function to ERROR. If that happens
949  * it may have dropped some but not all slots.
950  *
951  * This routine isn't as efficient as it could be - but we don't drop
952  * databases often, especially databases with lots of slots.
953  */
954 void
956 {
957  int i;
958 
959  if (max_replication_slots <= 0)
960  return;
961 
962 restart:
963  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
964  for (i = 0; i < max_replication_slots; i++)
965  {
966  ReplicationSlot *s;
967  char *slotname;
968  int active_pid;
969 
970  s = &ReplicationSlotCtl->replication_slots[i];
971 
972  /* cannot change while ReplicationSlotCtlLock is held */
973  if (!s->in_use)
974  continue;
975 
976  /* only logical slots are database specific, skip */
977  if (!SlotIsLogical(s))
978  continue;
979 
980  /* not our database, skip */
981  if (s->data.database != dboid)
982  continue;
983 
984  /* acquire slot, so ReplicationSlotDropAcquired can be reused */
985  SpinLockAcquire(&s->mutex);
986  /* can't change while ReplicationSlotControlLock is held */
987  slotname = NameStr(s->data.name);
988  active_pid = s->active_pid;
989  if (active_pid == 0)
990  {
991  MyReplicationSlot = s;
992  s->active_pid = MyProcPid;
993  }
994  SpinLockRelease(&s->mutex);
995 
996  /*
997  * Even though we hold an exclusive lock on the database object a
998  * logical slot for that DB can still be active, e.g. if it's
999  * concurrently being dropped by a backend connected to another DB.
1000  *
1001  * That's fairly unlikely in practice, so we'll just bail out.
1002  */
1003  if (active_pid)
1004  ereport(ERROR,
1005  (errcode(ERRCODE_OBJECT_IN_USE),
1006  errmsg("replication slot \"%s\" is active for PID %d",
1007  slotname, active_pid)));
1008 
1009  /*
1010  * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
1011  * holding ReplicationSlotControlLock over filesystem operations,
1012  * release ReplicationSlotControlLock and use
1013  * ReplicationSlotDropAcquired.
1014  *
1015  * As that means the set of slots could change, restart scan from the
1016  * beginning each time we release the lock.
1017  */
1018  LWLockRelease(ReplicationSlotControlLock);
1020  goto restart;
1021  }
1022  LWLockRelease(ReplicationSlotControlLock);
1023 }
1024 
1025 
1026 /*
1027  * Check whether the server's configuration supports using replication
1028  * slots.
1029  */
1030 void
1032 {
1033  /*
1034  * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
1035  * needs the same check.
1036  */
1037 
1038  if (max_replication_slots == 0)
1039  ereport(ERROR,
1040  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1041  errmsg("replication slots can only be used if max_replication_slots > 0")));
1042 
1044  ereport(ERROR,
1045  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1046  errmsg("replication slots can only be used if wal_level >= replica")));
1047 }
1048 
1049 /*
1050  * Reserve WAL for the currently active slot.
1051  *
1052  * Compute and set restart_lsn in a manner that's appropriate for the type of
1053  * the slot and concurrency safe.
1054  */
1055 void
1057 {
1059 
1060  Assert(slot != NULL);
1062 
1063  /*
1064  * The replication slot mechanism is used to prevent removal of required
1065  * WAL. As there is no interlock between this routine and checkpoints, WAL
1066  * segments could concurrently be removed when a now stale return value of
1067  * ReplicationSlotsComputeRequiredLSN() is used. In the unlikely case that
1068  * this happens we'll just retry.
1069  */
1070  while (true)
1071  {
1072  XLogSegNo segno;
1073  XLogRecPtr restart_lsn;
1074 
1075  /*
1076  * For logical slots log a standby snapshot and start logical decoding
1077  * at exactly that position. That allows the slot to start up more
1078  * quickly.
1079  *
1080  * That's not needed (or indeed helpful) for physical slots as they'll
1081  * start replay at the last logged checkpoint anyway. Instead return
1082  * the location of the last redo LSN. While that slightly increases
1083  * the chance that we have to retry, it's where a base backup has to
1084  * start replay at.
1085  */
1086  if (!RecoveryInProgress() && SlotIsLogical(slot))
1087  {
1088  XLogRecPtr flushptr;
1089 
1090  /* start at current insert position */
1091  restart_lsn = GetXLogInsertRecPtr();
1092  SpinLockAcquire(&slot->mutex);
1093  slot->data.restart_lsn = restart_lsn;
1094  SpinLockRelease(&slot->mutex);
1095 
1096  /* make sure we have enough information to start */
1097  flushptr = LogStandbySnapshot();
1098 
1099  /* and make sure it's fsynced to disk */
1100  XLogFlush(flushptr);
1101  }
1102  else
1103  {
1104  restart_lsn = GetRedoRecPtr();
1105  SpinLockAcquire(&slot->mutex);
1106  slot->data.restart_lsn = restart_lsn;
1107  SpinLockRelease(&slot->mutex);
1108  }
1109 
1110  /* prevent WAL removal as fast as possible */
1112 
1113  /*
1114  * If all required WAL is still there, great, otherwise retry. The
1115  * slot should prevent further removal of WAL, unless there's a
1116  * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
1117  * the new restart_lsn above, so normally we should never need to loop
1118  * more than twice.
1119  */
1121  if (XLogGetLastRemovedSegno() < segno)
1122  break;
1123  }
1124 }
1125 
1126 /*
1127  * Mark any slot that points to an LSN older than the given segment
1128  * as invalid; it requires WAL that's about to be removed.
1129  *
1130  * NB - this runs as part of checkpoint, so avoid raising errors if possible.
1131  */
1132 void
1134 {
1135  XLogRecPtr oldestLSN;
1136 
1137  XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
1138 
1139 restart:
1140  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
1141  for (int i = 0; i < max_replication_slots; i++)
1142  {
1143  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
1144  XLogRecPtr restart_lsn = InvalidXLogRecPtr;
1145  NameData slotname;
1146  int wspid;
1147  int last_signaled_pid = 0;
1148 
1149  if (!s->in_use)
1150  continue;
1151 
1152  SpinLockAcquire(&s->mutex);
1153  slotname = s->data.name;
1154  restart_lsn = s->data.restart_lsn;
1155  SpinLockRelease(&s->mutex);
1156 
1157  if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn >= oldestLSN)
1158  continue;
1159  LWLockRelease(ReplicationSlotControlLock);
1161 
1162  /* Get ready to sleep on the slot in case it is active */
1164 
1165  for (;;)
1166  {
1167  /*
1168  * Try to mark this slot as used by this process.
1169  *
1170  * Note that ReplicationSlotAcquireInternal(SAB_Inquire)
1171  * should not cancel the prepared condition variable
1172  * if this slot is active in other process. Because in this case
1173  * we have to wait on that CV for the process owning
1174  * the slot to be terminated, later.
1175  */
1176  wspid = ReplicationSlotAcquireInternal(s, NULL, SAB_Inquire);
1177 
1178  /*
1179  * Exit the loop if we successfully acquired the slot or
1180  * the slot was dropped during waiting for the owning process
1181  * to be terminated. For example, the latter case is likely to
1182  * happen when the slot is temporary because it's automatically
1183  * dropped by the termination of the owning process.
1184  */
1185  if (wspid <= 0)
1186  break;
1187 
1188  /*
1189  * Signal to terminate the process that owns the slot.
1190  *
1191  * There is the race condition where other process may own
1192  * the slot after the process using it was terminated and before
1193  * this process owns it. To handle this case, we signal again
1194  * if the PID of the owning process is changed than the last.
1195  *
1196  * XXX This logic assumes that the same PID is not reused
1197  * very quickly.
1198  */
1199  if (last_signaled_pid != wspid)
1200  {
1201  ereport(LOG,
1202  (errmsg("terminating process %d because replication slot \"%s\" is too far behind",
1203  wspid, NameStr(slotname))));
1204  (void) kill(wspid, SIGTERM);
1205  last_signaled_pid = wspid;
1206  }
1207 
1210  }
1212 
1213  /*
1214  * Do nothing here and start from scratch if the slot has
1215  * already been dropped.
1216  */
1217  if (wspid == -1)
1218  goto restart;
1219 
1220  ereport(LOG,
1221  (errmsg("invalidating slot \"%s\" because its restart_lsn %X/%X exceeds max_slot_wal_keep_size",
1222  NameStr(slotname),
1223  (uint32) (restart_lsn >> 32),
1224  (uint32) restart_lsn)));
1225 
1226  SpinLockAcquire(&s->mutex);
1229  SpinLockRelease(&s->mutex);
1230 
1231  /* Make sure the invalidated state persists across server restart */
1235 
1236  /* if we did anything, start from scratch */
1237  goto restart;
1238  }
1239  LWLockRelease(ReplicationSlotControlLock);
1240 }
1241 
1242 /*
1243  * Flush all replication slots to disk.
1244  *
1245  * This needn't actually be part of a checkpoint, but it's a convenient
1246  * location.
1247  */
1248 void
1250 {
1251  int i;
1252 
1253  elog(DEBUG1, "performing replication slot checkpoint");
1254 
1255  /*
1256  * Prevent any slot from being created/dropped while we're active. As we
1257  * explicitly do *not* want to block iterating over replication_slots or
1258  * acquiring a slot we cannot take the control lock - but that's OK,
1259  * because holding ReplicationSlotAllocationLock is strictly stronger, and
1260  * enough to guarantee that nobody can change the in_use bits on us.
1261  */
1262  LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
1263 
1264  for (i = 0; i < max_replication_slots; i++)
1265  {
1266  ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
1267  char path[MAXPGPATH];
1268 
1269  if (!s->in_use)
1270  continue;
1271 
1272  /* save the slot to disk, locking is handled in SaveSlotToPath() */
1273  sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
1274  SaveSlotToPath(s, path, LOG);
1275  }
1276  LWLockRelease(ReplicationSlotAllocationLock);
1277 }
1278 
1279 /*
1280  * Load all replication slots from disk into memory at server startup. This
1281  * needs to be run before we start crash recovery.
1282  */
1283 void
1285 {
1286  DIR *replication_dir;
1287  struct dirent *replication_de;
1288 
1289  elog(DEBUG1, "starting up replication slots");
1290 
1291  /* restore all slots by iterating over all on-disk entries */
1292  replication_dir = AllocateDir("pg_replslot");
1293  while ((replication_de = ReadDir(replication_dir, "pg_replslot")) != NULL)
1294  {
1295  struct stat statbuf;
1296  char path[MAXPGPATH + 12];
1297 
1298  if (strcmp(replication_de->d_name, ".") == 0 ||
1299  strcmp(replication_de->d_name, "..") == 0)
1300  continue;
1301 
1302  snprintf(path, sizeof(path), "pg_replslot/%s", replication_de->d_name);
1303 
1304  /* we're only creating directories here, skip if it's not our's */
1305  if (lstat(path, &statbuf) == 0 && !S_ISDIR(statbuf.st_mode))
1306  continue;
1307 
1308  /* we crashed while a slot was being setup or deleted, clean up */
1309  if (pg_str_endswith(replication_de->d_name, ".tmp"))
1310  {
1311  if (!rmtree(path, true))
1312  {
1313  ereport(WARNING,
1314  (errmsg("could not remove directory \"%s\"",
1315  path)));
1316  continue;
1317  }
1318  fsync_fname("pg_replslot", true);
1319  continue;
1320  }
1321 
1322  /* looks like a slot in a normal state, restore */
1323  RestoreSlotFromDisk(replication_de->d_name);
1324  }
1325  FreeDir(replication_dir);
1326 
1327  /* currently no slots exist, we're done. */
1328  if (max_replication_slots <= 0)
1329  return;
1330 
1331  /* Now that we have recovered all the data, compute replication xmin */
1334 }
1335 
1336 /* ----
1337  * Manipulation of on-disk state of replication slots
1338  *
1339  * NB: none of the routines below should take any notice whether a slot is the
1340  * current one or not, that's all handled a layer above.
1341  * ----
1342  */
1343 static void
1345 {
1346  char tmppath[MAXPGPATH];
1347  char path[MAXPGPATH];
1348  struct stat st;
1349 
1350  /*
1351  * No need to take out the io_in_progress_lock, nobody else can see this
1352  * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
1353  * takes out the lock, if we'd take the lock here, we'd deadlock.
1354  */
1355 
1356  sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
1357  sprintf(tmppath, "pg_replslot/%s.tmp", NameStr(slot->data.name));
1358 
1359  /*
1360  * It's just barely possible that some previous effort to create or drop a
1361  * slot with this name left a temp directory lying around. If that seems
1362  * to be the case, try to remove it. If the rmtree() fails, we'll error
1363  * out at the MakePGDirectory() below, so we don't bother checking
1364  * success.
1365  */
1366  if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
1367  rmtree(tmppath, true);
1368 
1369  /* Create and fsync the temporary slot directory. */
1370  if (MakePGDirectory(tmppath) < 0)
1371  ereport(ERROR,
1373  errmsg("could not create directory \"%s\": %m",
1374  tmppath)));
1375  fsync_fname(tmppath, true);
1376 
1377  /* Write the actual state file. */
1378  slot->dirty = true; /* signal that we really need to write */
1379  SaveSlotToPath(slot, tmppath, ERROR);
1380 
1381  /* Rename the directory into place. */
1382  if (rename(tmppath, path) != 0)
1383  ereport(ERROR,
1385  errmsg("could not rename file \"%s\" to \"%s\": %m",
1386  tmppath, path)));
1387 
1388  /*
1389  * If we'd now fail - really unlikely - we wouldn't know whether this slot
1390  * would persist after an OS crash or not - so, force a restart. The
1391  * restart would try to fsync this again till it works.
1392  */
1394 
1395  fsync_fname(path, true);
1396  fsync_fname("pg_replslot", true);
1397 
1398  END_CRIT_SECTION();
1399 }
1400 
1401 /*
1402  * Shared functionality between saving and creating a replication slot.
1403  */
1404 static void
1405 SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
1406 {
1407  char tmppath[MAXPGPATH];
1408  char path[MAXPGPATH];
1409  int fd;
1411  bool was_dirty;
1412 
1413  /* first check whether there's something to write out */
1414  SpinLockAcquire(&slot->mutex);
1415  was_dirty = slot->dirty;
1416  slot->just_dirtied = false;
1417  SpinLockRelease(&slot->mutex);
1418 
1419  /* and don't do anything if there's nothing to write */
1420  if (!was_dirty)
1421  return;
1422 
1424 
1425  /* silence valgrind :( */
1426  memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
1427 
1428  sprintf(tmppath, "%s/state.tmp", dir);
1429  sprintf(path, "%s/state", dir);
1430 
1431  fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
1432  if (fd < 0)
1433  {
1434  /*
1435  * If not an ERROR, then release the lock before returning. In case
1436  * of an ERROR, the error recovery path automatically releases the
1437  * lock, but no harm in explicitly releasing even in that case. Note
1438  * that LWLockRelease() could affect errno.
1439  */
1440  int save_errno = errno;
1441 
1443  errno = save_errno;
1444  ereport(elevel,
1446  errmsg("could not create file \"%s\": %m",
1447  tmppath)));
1448  return;
1449  }
1450 
1451  cp.magic = SLOT_MAGIC;
1452  INIT_CRC32C(cp.checksum);
1453  cp.version = SLOT_VERSION;
1455 
1456  SpinLockAcquire(&slot->mutex);
1457 
1458  memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
1459 
1460  SpinLockRelease(&slot->mutex);
1461 
1462  COMP_CRC32C(cp.checksum,
1463  (char *) (&cp) + SnapBuildOnDiskNotChecksummedSize,
1465  FIN_CRC32C(cp.checksum);
1466 
1467  errno = 0;
1469  if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
1470  {
1471  int save_errno = errno;
1472 
1474  CloseTransientFile(fd);
1476 
1477  /* if write didn't set errno, assume problem is no disk space */
1478  errno = save_errno ? save_errno : ENOSPC;
1479  ereport(elevel,
1481  errmsg("could not write to file \"%s\": %m",
1482  tmppath)));
1483  return;
1484  }
1486 
1487  /* fsync the temporary file */
1489  if (pg_fsync(fd) != 0)
1490  {
1491  int save_errno = errno;
1492 
1494  CloseTransientFile(fd);
1496  errno = save_errno;
1497  ereport(elevel,
1499  errmsg("could not fsync file \"%s\": %m",
1500  tmppath)));
1501  return;
1502  }
1504 
1505  if (CloseTransientFile(fd) != 0)
1506  {
1507  int save_errno = errno;
1508 
1510  errno = save_errno;
1511  ereport(elevel,
1513  errmsg("could not close file \"%s\": %m",
1514  tmppath)));
1515  return;
1516  }
1517 
1518  /* rename to permanent file, fsync file and directory */
1519  if (rename(tmppath, path) != 0)
1520  {
1521  int save_errno = errno;
1522 
1524  errno = save_errno;
1525  ereport(elevel,
1527  errmsg("could not rename file \"%s\" to \"%s\": %m",
1528  tmppath, path)));
1529  return;
1530  }
1531 
1532  /*
1533  * Check CreateSlotOnDisk() for the reasoning of using a critical section.
1534  */
1536 
1537  fsync_fname(path, false);
1538  fsync_fname(dir, true);
1539  fsync_fname("pg_replslot", true);
1540 
1541  END_CRIT_SECTION();
1542 
1543  /*
1544  * Successfully wrote, unset dirty bit, unless somebody dirtied again
1545  * already.
1546  */
1547  SpinLockAcquire(&slot->mutex);
1548  if (!slot->just_dirtied)
1549  slot->dirty = false;
1550  SpinLockRelease(&slot->mutex);
1551 
1553 }
1554 
1555 /*
1556  * Load a single slot from disk into memory.
1557  */
1558 static void
1560 {
1562  int i;
1563  char slotdir[MAXPGPATH + 12];
1564  char path[MAXPGPATH + 22];
1565  int fd;
1566  bool restored = false;
1567  int readBytes;
1569 
1570  /* no need to lock here, no concurrent access allowed yet */
1571 
1572  /* delete temp file if it exists */
1573  sprintf(slotdir, "pg_replslot/%s", name);
1574  sprintf(path, "%s/state.tmp", slotdir);
1575  if (unlink(path) < 0 && errno != ENOENT)
1576  ereport(PANIC,
1578  errmsg("could not remove file \"%s\": %m", path)));
1579 
1580  sprintf(path, "%s/state", slotdir);
1581 
1582  elog(DEBUG1, "restoring replication slot from \"%s\"", path);
1583 
1584  /* on some operating systems fsyncing a file requires O_RDWR */
1585  fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
1586 
1587  /*
1588  * We do not need to handle this as we are rename()ing the directory into
1589  * place only after we fsync()ed the state file.
1590  */
1591  if (fd < 0)
1592  ereport(PANIC,
1594  errmsg("could not open file \"%s\": %m", path)));
1595 
1596  /*
1597  * Sync state file before we're reading from it. We might have crashed
1598  * while it wasn't synced yet and we shouldn't continue on that basis.
1599  */
1601  if (pg_fsync(fd) != 0)
1602  ereport(PANIC,
1604  errmsg("could not fsync file \"%s\": %m",
1605  path)));
1607 
1608  /* Also sync the parent directory */
1610  fsync_fname(slotdir, true);
1611  END_CRIT_SECTION();
1612 
1613  /* read part of statefile that's guaranteed to be version independent */
1615  readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
1617  if (readBytes != ReplicationSlotOnDiskConstantSize)
1618  {
1619  if (readBytes < 0)
1620  ereport(PANIC,
1622  errmsg("could not read file \"%s\": %m", path)));
1623  else
1624  ereport(PANIC,
1626  errmsg("could not read file \"%s\": read %d of %zu",
1627  path, readBytes,
1629  }
1630 
1631  /* verify magic */
1632  if (cp.magic != SLOT_MAGIC)
1633  ereport(PANIC,
1635  errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
1636  path, cp.magic, SLOT_MAGIC)));
1637 
1638  /* verify version */
1639  if (cp.version != SLOT_VERSION)
1640  ereport(PANIC,
1642  errmsg("replication slot file \"%s\" has unsupported version %u",
1643  path, cp.version)));
1644 
1645  /* boundary check on length */
1647  ereport(PANIC,
1649  errmsg("replication slot file \"%s\" has corrupted length %u",
1650  path, cp.length)));
1651 
1652  /* Now that we know the size, read the entire file */
1654  readBytes = read(fd,
1655  (char *) &cp + ReplicationSlotOnDiskConstantSize,
1656  cp.length);
1658  if (readBytes != cp.length)
1659  {
1660  if (readBytes < 0)
1661  ereport(PANIC,
1663  errmsg("could not read file \"%s\": %m", path)));
1664  else
1665  ereport(PANIC,
1667  errmsg("could not read file \"%s\": read %d of %zu",
1668  path, readBytes, (Size) cp.length)));
1669  }
1670 
1671  if (CloseTransientFile(fd) != 0)
1672  ereport(PANIC,
1674  errmsg("could not close file \"%s\": %m", path)));
1675 
1676  /* now verify the CRC */
1677  INIT_CRC32C(checksum);
1678  COMP_CRC32C(checksum,
1679  (char *) &cp + SnapBuildOnDiskNotChecksummedSize,
1681  FIN_CRC32C(checksum);
1682 
1683  if (!EQ_CRC32C(checksum, cp.checksum))
1684  ereport(PANIC,
1685  (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
1686  path, checksum, cp.checksum)));
1687 
1688  /*
1689  * If we crashed with an ephemeral slot active, don't restore but delete
1690  * it.
1691  */
1693  {
1694  if (!rmtree(slotdir, true))
1695  {
1696  ereport(WARNING,
1697  (errmsg("could not remove directory \"%s\"",
1698  slotdir)));
1699  }
1700  fsync_fname("pg_replslot", true);
1701  return;
1702  }
1703 
1704  /*
1705  * Verify that requirements for the specific slot type are met. That's
1706  * important because if these aren't met we're not guaranteed to retain
1707  * all the necessary resources for the slot.
1708  *
1709  * NB: We have to do so *after* the above checks for ephemeral slots,
1710  * because otherwise a slot that shouldn't exist anymore could prevent
1711  * restarts.
1712  *
1713  * NB: Changing the requirements here also requires adapting
1714  * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
1715  */
1717  ereport(FATAL,
1718  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1719  errmsg("logical replication slot \"%s\" exists, but wal_level < logical",
1720  NameStr(cp.slotdata.name)),
1721  errhint("Change wal_level to be logical or higher.")));
1722  else if (wal_level < WAL_LEVEL_REPLICA)
1723  ereport(FATAL,
1724  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1725  errmsg("physical replication slot \"%s\" exists, but wal_level < replica",
1726  NameStr(cp.slotdata.name)),
1727  errhint("Change wal_level to be replica or higher.")));
1728 
1729  /* nothing can be active yet, don't lock anything */
1730  for (i = 0; i < max_replication_slots; i++)
1731  {
1732  ReplicationSlot *slot;
1733 
1734  slot = &ReplicationSlotCtl->replication_slots[i];
1735 
1736  if (slot->in_use)
1737  continue;
1738 
1739  /* restore the entire set of persistent data */
1740  memcpy(&slot->data, &cp.slotdata,
1742 
1743  /* initialize in memory state */
1744  slot->effective_xmin = cp.slotdata.xmin;
1746 
1751 
1752  slot->in_use = true;
1753  slot->active_pid = 0;
1754 
1755  restored = true;
1756  break;
1757  }
1758 
1759  if (!restored)
1760  ereport(FATAL,
1761  (errmsg("too many replication slots active before shutdown"),
1762  errhint("Increase max_replication_slots and try again.")));
1763 }
#define INIT_CRC32C(crc)
Definition: pg_crc32c.h:41
static void RestoreSlotFromDisk(const char *name)
Definition: slot.c:1559
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:93
void CheckSlotRequirements(void)
Definition: slot.c:1031
TransactionId candidate_catalog_xmin
Definition: slot.h:158
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
bool LWLockHeldByMeInMode(LWLock *l, LWLockMode mode)
Definition: lwlock.c:1946
#define DEBUG1
Definition: elog.h:25
int MyProcPid
Definition: globals.c:40
int errhint(const char *fmt,...)
Definition: elog.c:1071
Size ReplicationSlotsShmemSize(void)
Definition: slot.c:117
#define PROC_IN_LOGICAL_DECODING
Definition: proc.h:57
int wal_segment_size
Definition: xlog.c:116
uint32 TransactionId
Definition: c.h:520
bool pg_str_endswith(const char *str, const char *end)
Definition: string.c:31
#define write(a, b, c)
Definition: win32.h:14
#define SLOT_MAGIC
Definition: slot.c:89
uint32 pg_crc32c
Definition: pg_crc32c.h:38
int wal_level
Definition: xlog.c:106
#define SpinLockInit(lock)
Definition: spin.h:60
#define END_CRIT_SECTION()
Definition: miscadmin.h:134
void fsync_fname(const char *fname, bool isdir)
Definition: fd.c:632
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency)
Definition: slot.c:222
ReplicationSlotPersistency persistency
Definition: slot.h:61
#define START_CRIT_SECTION()
Definition: miscadmin.h:132
void ConditionVariableBroadcast(ConditionVariable *cv)
int errcode(int sqlerrcode)
Definition: elog.c:610
#define MemSet(start, val, len)
Definition: c.h:978
#define kill(pid, sig)
Definition: win32_port.h:426
void ReplicationSlotSave(void)
Definition: slot.c:697
#define SnapBuildOnDiskNotChecksummedSize
Definition: slot.c:80
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
Definition: slot.c:599
ReplicationSlotPersistentData data
Definition: slot.h:143
#define LOG
Definition: elog.h:26
unsigned int Oid
Definition: postgres_ext.h:31
bool RecoveryInProgress(void)
Definition: xlog.c:8072
Definition: dirent.h:9
#define PANIC
Definition: elog.h:53
void XLogFlush(XLogRecPtr record)
Definition: xlog.c:2847
static int fd(const char *x, int i)
Definition: preproc-init.c:105
#define PG_BINARY
Definition: c.h:1240
static void CreateSlotOnDisk(ReplicationSlot *slot)
Definition: slot.c:1344
void ReplicationSlotsShmemInit(void)
Definition: slot.c:135
static void SaveSlotToPath(ReplicationSlot *slot, const char *path, int elevel)
Definition: slot.c:1405
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
Definition: xlog.c:2726
PGXACT * MyPgXact
Definition: proc.c:68
uint8 vacuumFlags
Definition: proc.h:240
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1812
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
#define NAMEDATALEN
#define sprintf
Definition: port.h:195
bool ReplicationSlotValidateName(const char *name, int elevel)
Definition: slot.c:175
#define SpinLockAcquire(lock)
Definition: spin.h:62
void ConditionVariableInit(ConditionVariable *cv)
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3959
void ReplicationSlotReserveWal(void)
Definition: slot.c:1056
static void ReplicationSlotDropAcquired(void)
Definition: slot.c:582
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:804
ReplicationSlotPersistentData slotdata
Definition: slot.c:73
static ReplicationSlot * SearchNamedReplicationSlot(const char *name)
Definition: slot.c:335
void ConditionVariableCancelSleep(void)
XLogRecPtr LogStandbySnapshot(void)
Definition: standby.c:923
Definition: dirent.c:25
#define ERROR
Definition: elog.h:43
bool ConditionVariableTimedSleep(ConditionVariable *cv, long timeout, uint32 wait_event_info)
int OpenTransientFile(const char *fileName, int fileFlags)
Definition: fd.c:2372
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:392
#define FATAL
Definition: elog.h:52
XLogRecPtr GetXLogInsertRecPtr(void)
Definition: xlog.c:11491
#define MAXPGPATH
Definition: slot.h:43
void ReplicationSlotPersist(void)
Definition: slot.c:732
TransactionId effective_xmin
Definition: slot.h:139
Definition: c.h:616
XLogRecPtr candidate_restart_valid
Definition: slot.h:160
void StartupReplicationSlots(void)
Definition: slot.c:1284
bool IsUnderPostmaster
Definition: globals.c:109
uint64 XLogSegNo
Definition: xlogdefs.h:41
SlotAcquireBehavior
Definition: slot.h:40
int errcode_for_file_access(void)
Definition: elog.c:633
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void)
Definition: slot.c:847
TransactionId catalog_xmin
Definition: slot.h:77
#define InvalidTransactionId
Definition: transam.h:31
unsigned int uint32
Definition: c.h:374
DIR * AllocateDir(const char *dirname)
Definition: fd.c:2583
void ReplicationSlotRelease(void)
Definition: slot.c:476
static void pgstat_report_wait_end(void)
Definition: pgstat.h:1381
TransactionId xmin
Definition: slot.h:69
#define EQ_CRC32C(c1, c2)
Definition: pg_crc32c.h:42
#define SlotIsLogical(slot)
Definition: slot.h:165
#define AssertArg(condition)
Definition: c.h:747
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
Definition: transam.c:300
pg_crc32c checksum
Definition: slot.c:62
#define ERRCODE_DATA_CORRUPTED
Definition: pg_basebackup.c:45
struct ReplicationSlotOnDisk ReplicationSlotOnDisk
void LWLockInitialize(LWLock *lock, int tranche_id)
Definition: lwlock.c:745
int CloseTransientFile(int fd)
Definition: fd.c:2549
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29
#define WARNING
Definition: elog.h:40
#define stat(a, b)
Definition: win32_port.h:255
bool rmtree(const char *path, bool rmtopdir)
Definition: rmtree.c:42
bool in_use
Definition: slot.h:119
static int elevel
Definition: vacuumlazy.c:330
#define SpinLockRelease(lock)
Definition: spin.h:64
Size mul_size(Size s1, Size s2)
Definition: shmem.c:515
bool just_dirtied
Definition: slot.h:125
Size add_size(Size s1, Size s2)
Definition: shmem.c:498
TransactionId effective_catalog_xmin
Definition: slot.h:140
Oid MyDatabaseId
Definition: globals.c:85
#define SLOT_VERSION
Definition: slot.c:90
#define InvalidOid
Definition: postgres_ext.h:36
#define ereport(elevel,...)
Definition: elog.h:144
int MakePGDirectory(const char *directoryName)
Definition: fd.c:3582
ReplicationSlot * MyReplicationSlot
Definition: slot.c:96
int max_replication_slots
Definition: slot.c:99
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define ReplicationSlotOnDiskV2Size
Definition: slot.c:86
uint64 XLogRecPtr
Definition: xlogdefs.h:21
#define Assert(condition)
Definition: c.h:745
#define StrNCpy(dst, src, len)
Definition: c.h:951
struct dirent * ReadDir(DIR *dir, const char *dirname)
Definition: fd.c:2649
XLogRecPtr restart_lsn
Definition: slot.h:80
#define ReplicationSlotOnDiskConstantSize
Definition: slot.c:77
size_t Size
Definition: c.h:473
uint32 version
Definition: slot.c:65
static void pgstat_report_wait_start(uint32 wait_event_info)
Definition: pgstat.h:1357
#define SnapBuildOnDiskChecksummedSize
Definition: slot.c:83
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1208
XLogRecPtr GetRedoRecPtr(void)
Definition: xlog.c:8364
static int ReplicationSlotAcquireInternal(ReplicationSlot *slot, const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:381
ConditionVariable active_cv
Definition: slot.h:149
const char * name
Definition: encode.c:561
bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
Definition: slot.c:899
#define S_ISDIR(m)
Definition: win32_port.h:296
#define lstat(path, sb)
Definition: win32_port.h:244
XLogRecPtr candidate_xmin_lsn
Definition: slot.h:159
void ReplicationSlotDrop(const char *name, bool nowait)
Definition: slot.c:569
ReplicationSlotPersistency
Definition: slot.h:32
int errmsg(const char *fmt,...)
Definition: elog.c:824
int ReplicationSlotAcquire(const char *name, SlotAcquireBehavior behavior)
Definition: slot.c:367
pid_t active_pid
Definition: slot.h:122
void InvalidateObsoleteReplicationSlots(XLogSegNo oldestSegno)
Definition: slot.c:1133
#define elog(elevel,...)
Definition: elog.h:214
int i
Definition: slot.h:42
#define NameStr(name)
Definition: c.h:622
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
Definition: procarray.c:3092
void ReplicationSlotCleanup(void)
Definition: slot.c:531
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:99
int pg_fsync(int fd)
Definition: fd.c:345
ReplicationSlot replication_slots[1]
Definition: slot.h:176
char d_name[MAX_PATH]
Definition: dirent.h:14
XLogRecPtr invalidated_at
Definition: slot.h:83
slock_t mutex
Definition: slot.h:116
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define COMP_CRC32C(crc, data, len)
Definition: pg_crc32c.h:89
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:31
void CheckPointReplicationSlots(void)
Definition: slot.c:1249
#define FIN_CRC32C(crc)
Definition: pg_crc32c.h:94
#define snprintf
Definition: port.h:193
void ReplicationSlotsDropDBSlots(Oid dboid)
Definition: slot.c:955
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:754
bool dirty
Definition: slot.h:126
#define read(a, b, c)
Definition: win32.h:13
int FreeDir(DIR *dir)
Definition: fd.c:2701
XLogRecPtr candidate_restart_lsn
Definition: slot.h:161
#define offsetof(type, field)
Definition: c.h:668
void ReplicationSlotMarkDirty(void)
Definition: slot.c:715
LWLock io_in_progress_lock
Definition: slot.h:146
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)