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