PostgreSQL Source Code  git master
slotfuncs.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * slotfuncs.c
4  * Support functions for replication slots
5  *
6  * Copyright (c) 2012-2024, PostgreSQL Global Development Group
7  *
8  * IDENTIFICATION
9  * src/backend/replication/slotfuncs.c
10  *
11  *-------------------------------------------------------------------------
12  */
13 #include "postgres.h"
14 
15 #include "access/htup_details.h"
16 #include "access/xlog_internal.h"
17 #include "access/xlogrecovery.h"
18 #include "access/xlogutils.h"
19 #include "funcapi.h"
20 #include "miscadmin.h"
21 #include "replication/decode.h"
22 #include "replication/logical.h"
23 #include "replication/slot.h"
24 #include "replication/slotsync.h"
25 #include "utils/builtins.h"
26 #include "utils/guc.h"
27 #include "utils/inval.h"
28 #include "utils/pg_lsn.h"
29 #include "utils/resowner.h"
30 
31 /*
32  * Helper function for creating a new physical replication slot with
33  * given arguments. Note that this function doesn't release the created
34  * slot.
35  *
36  * If restart_lsn is a valid value, we use it without WAL reservation
37  * routine. So the caller must guarantee that WAL is available.
38  */
39 static void
40 create_physical_replication_slot(char *name, bool immediately_reserve,
41  bool temporary, XLogRecPtr restart_lsn)
42 {
44 
45  /* acquire replication slot, this will check for conflicting names */
47  temporary ? RS_TEMPORARY : RS_PERSISTENT, false,
48  false, false);
49 
50  if (immediately_reserve)
51  {
52  /* Reserve WAL as the user asked for it */
53  if (XLogRecPtrIsInvalid(restart_lsn))
55  else
56  MyReplicationSlot->data.restart_lsn = restart_lsn;
57 
58  /* Write this slot to disk */
61  }
62 }
63 
64 /*
65  * SQL function for creating a new physical (streaming replication)
66  * replication slot.
67  */
68 Datum
70 {
72  bool immediately_reserve = PG_GETARG_BOOL(1);
73  bool temporary = PG_GETARG_BOOL(2);
74  Datum values[2];
75  bool nulls[2];
76  TupleDesc tupdesc;
77  HeapTuple tuple;
78  Datum result;
79 
80  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
81  elog(ERROR, "return type must be a row type");
82 
84 
86 
88  immediately_reserve,
89  temporary,
91 
93  nulls[0] = false;
94 
95  if (immediately_reserve)
96  {
98  nulls[1] = false;
99  }
100  else
101  nulls[1] = true;
102 
103  tuple = heap_form_tuple(tupdesc, values, nulls);
104  result = HeapTupleGetDatum(tuple);
105 
107 
108  PG_RETURN_DATUM(result);
109 }
110 
111 
112 /*
113  * Helper function for creating a new logical replication slot with
114  * given arguments. Note that this function doesn't release the created
115  * slot.
116  *
117  * When find_startpoint is false, the slot's confirmed_flush is not set; it's
118  * caller's responsibility to ensure it's set to something sensible.
119  */
120 static void
122  bool temporary, bool two_phase,
123  bool failover,
124  XLogRecPtr restart_lsn,
125  bool find_startpoint)
126 {
127  LogicalDecodingContext *ctx = NULL;
128 
130 
131  /*
132  * Acquire a logical decoding slot, this will check for conflicting names.
133  * Initially create persistent slot as ephemeral - that allows us to
134  * nicely handle errors during initialization because it'll get dropped if
135  * this transaction fails. We'll make it persistent at the end. Temporary
136  * slots can be created as temporary from beginning as they get dropped on
137  * error as well.
138  */
140  temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase,
141  failover, false);
142 
143  /*
144  * Create logical decoding context to find start point or, if we don't
145  * need it, to 1) bump slot's restart_lsn and xmin 2) check plugin sanity.
146  *
147  * Note: when !find_startpoint this is still important, because it's at
148  * this point that the output plugin is validated.
149  */
151  false, /* just catalogs is OK */
152  restart_lsn,
153  XL_ROUTINE(.page_read = read_local_xlog_page,
154  .segment_open = wal_segment_open,
155  .segment_close = wal_segment_close),
156  NULL, NULL, NULL);
157 
158  /*
159  * If caller needs us to determine the decoding start point, do so now.
160  * This might take a while.
161  */
162  if (find_startpoint)
164 
165  /* don't need the decoding context anymore */
166  FreeDecodingContext(ctx);
167 }
168 
169 /*
170  * SQL function for creating a new logical replication slot.
171  */
172 Datum
174 {
175  Name name = PG_GETARG_NAME(0);
177  bool temporary = PG_GETARG_BOOL(2);
178  bool two_phase = PG_GETARG_BOOL(3);
179  bool failover = PG_GETARG_BOOL(4);
180  Datum result;
181  TupleDesc tupdesc;
182  HeapTuple tuple;
183  Datum values[2];
184  bool nulls[2];
185 
186  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
187  elog(ERROR, "return type must be a row type");
188 
190 
192 
194  NameStr(*plugin),
195  temporary,
196  two_phase,
197  failover,
199  true);
200 
203 
204  memset(nulls, 0, sizeof(nulls));
205 
206  tuple = heap_form_tuple(tupdesc, values, nulls);
207  result = HeapTupleGetDatum(tuple);
208 
209  /* ok, slot is now fully created, mark it as persistent if needed */
210  if (!temporary)
213 
214  PG_RETURN_DATUM(result);
215 }
216 
217 
218 /*
219  * SQL function for dropping a replication slot.
220  */
221 Datum
223 {
224  Name name = PG_GETARG_NAME(0);
225 
227 
229 
231 
232  PG_RETURN_VOID();
233 }
234 
235 /*
236  * pg_get_replication_slots - SQL SRF showing all replication slots
237  * that currently exist on the database cluster.
238  */
239 Datum
241 {
242 #define PG_GET_REPLICATION_SLOTS_COLS 19
243  ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
244  XLogRecPtr currlsn;
245  int slotno;
246 
247  /*
248  * We don't require any special permission to see this function's data
249  * because nothing should be sensitive. The most critical being the slot
250  * name, which shouldn't contain anything particularly sensitive.
251  */
252 
253  InitMaterializedSRF(fcinfo, 0);
254 
255  currlsn = GetXLogWriteRecPtr();
256 
257  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
258  for (slotno = 0; slotno < max_replication_slots; slotno++)
259  {
261  ReplicationSlot slot_contents;
263  bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
264  WALAvailability walstate;
265  int i;
267 
268  if (!slot->in_use)
269  continue;
270 
271  /* Copy slot contents while holding spinlock, then examine at leisure */
272  SpinLockAcquire(&slot->mutex);
273  slot_contents = *slot;
274  SpinLockRelease(&slot->mutex);
275 
276  memset(values, 0, sizeof(values));
277  memset(nulls, 0, sizeof(nulls));
278 
279  i = 0;
280  values[i++] = NameGetDatum(&slot_contents.data.name);
281 
282  if (slot_contents.data.database == InvalidOid)
283  nulls[i++] = true;
284  else
285  values[i++] = NameGetDatum(&slot_contents.data.plugin);
286 
287  if (slot_contents.data.database == InvalidOid)
288  values[i++] = CStringGetTextDatum("physical");
289  else
290  values[i++] = CStringGetTextDatum("logical");
291 
292  if (slot_contents.data.database == InvalidOid)
293  nulls[i++] = true;
294  else
295  values[i++] = ObjectIdGetDatum(slot_contents.data.database);
296 
297  values[i++] = BoolGetDatum(slot_contents.data.persistency == RS_TEMPORARY);
298  values[i++] = BoolGetDatum(slot_contents.active_pid != 0);
299 
300  if (slot_contents.active_pid != 0)
301  values[i++] = Int32GetDatum(slot_contents.active_pid);
302  else
303  nulls[i++] = true;
304 
305  if (slot_contents.data.xmin != InvalidTransactionId)
306  values[i++] = TransactionIdGetDatum(slot_contents.data.xmin);
307  else
308  nulls[i++] = true;
309 
310  if (slot_contents.data.catalog_xmin != InvalidTransactionId)
311  values[i++] = TransactionIdGetDatum(slot_contents.data.catalog_xmin);
312  else
313  nulls[i++] = true;
314 
315  if (slot_contents.data.restart_lsn != InvalidXLogRecPtr)
316  values[i++] = LSNGetDatum(slot_contents.data.restart_lsn);
317  else
318  nulls[i++] = true;
319 
320  if (slot_contents.data.confirmed_flush != InvalidXLogRecPtr)
321  values[i++] = LSNGetDatum(slot_contents.data.confirmed_flush);
322  else
323  nulls[i++] = true;
324 
325  /*
326  * If the slot has not been invalidated, test availability from
327  * restart_lsn.
328  */
329  if (slot_contents.data.invalidated != RS_INVAL_NONE)
330  walstate = WALAVAIL_REMOVED;
331  else
332  walstate = GetWALAvailability(slot_contents.data.restart_lsn);
333 
334  switch (walstate)
335  {
337  nulls[i++] = true;
338  break;
339 
340  case WALAVAIL_RESERVED:
341  values[i++] = CStringGetTextDatum("reserved");
342  break;
343 
344  case WALAVAIL_EXTENDED:
345  values[i++] = CStringGetTextDatum("extended");
346  break;
347 
348  case WALAVAIL_UNRESERVED:
349  values[i++] = CStringGetTextDatum("unreserved");
350  break;
351 
352  case WALAVAIL_REMOVED:
353 
354  /*
355  * If we read the restart_lsn long enough ago, maybe that file
356  * has been removed by now. However, the walsender could have
357  * moved forward enough that it jumped to another file after
358  * we looked. If checkpointer signalled the process to
359  * termination, then it's definitely lost; but if a process is
360  * still alive, then "unreserved" seems more appropriate.
361  *
362  * If we do change it, save the state for safe_wal_size below.
363  */
364  if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
365  {
366  int pid;
367 
368  SpinLockAcquire(&slot->mutex);
369  pid = slot->active_pid;
370  slot_contents.data.restart_lsn = slot->data.restart_lsn;
371  SpinLockRelease(&slot->mutex);
372  if (pid != 0)
373  {
374  values[i++] = CStringGetTextDatum("unreserved");
375  walstate = WALAVAIL_UNRESERVED;
376  break;
377  }
378  }
379  values[i++] = CStringGetTextDatum("lost");
380  break;
381  }
382 
383  /*
384  * safe_wal_size is only computed for slots that have not been lost,
385  * and only if there's a configured maximum size.
386  */
387  if (walstate == WALAVAIL_REMOVED || max_slot_wal_keep_size_mb < 0)
388  nulls[i++] = true;
389  else
390  {
391  XLogSegNo targetSeg;
392  uint64 slotKeepSegs;
393  uint64 keepSegs;
394  XLogSegNo failSeg;
395  XLogRecPtr failLSN;
396 
397  XLByteToSeg(slot_contents.data.restart_lsn, targetSeg, wal_segment_size);
398 
399  /* determine how many segments can be kept by slots */
401  /* ditto for wal_keep_size */
403 
404  /* if currpos reaches failLSN, we lose our segment */
405  failSeg = targetSeg + Max(slotKeepSegs, keepSegs) + 1;
406  XLogSegNoOffsetToRecPtr(failSeg, 0, wal_segment_size, failLSN);
407 
408  values[i++] = Int64GetDatum(failLSN - currlsn);
409  }
410 
411  values[i++] = BoolGetDatum(slot_contents.data.two_phase);
412 
413  if (slot_contents.inactive_since > 0)
414  values[i++] = TimestampTzGetDatum(slot_contents.inactive_since);
415  else
416  nulls[i++] = true;
417 
418  cause = slot_contents.data.invalidated;
419 
420  if (SlotIsPhysical(&slot_contents))
421  nulls[i++] = true;
422  else
423  {
424  /*
425  * rows_removed and wal_level_insufficient are the only two
426  * reasons for the logical slot's conflict with recovery.
427  */
428  if (cause == RS_INVAL_HORIZON ||
429  cause == RS_INVAL_WAL_LEVEL)
430  values[i++] = BoolGetDatum(true);
431  else
432  values[i++] = BoolGetDatum(false);
433  }
434 
435  if (cause == RS_INVAL_NONE)
436  nulls[i++] = true;
437  else
439 
440  values[i++] = BoolGetDatum(slot_contents.data.failover);
441 
442  values[i++] = BoolGetDatum(slot_contents.data.synced);
443 
445 
446  tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
447  values, nulls);
448  }
449 
450  LWLockRelease(ReplicationSlotControlLock);
451 
452  return (Datum) 0;
453 }
454 
455 /*
456  * Helper function for advancing our physical replication slot forward.
457  *
458  * The LSN position to move to is compared simply to the slot's restart_lsn,
459  * knowing that any position older than that would be removed by successive
460  * checkpoints.
461  */
462 static XLogRecPtr
464 {
466  XLogRecPtr retlsn = startlsn;
467 
468  Assert(moveto != InvalidXLogRecPtr);
469 
470  if (startlsn < moveto)
471  {
475  retlsn = moveto;
476 
477  /*
478  * Dirty the slot so as it is written out at the next checkpoint. Note
479  * that the LSN position advanced may still be lost in the event of a
480  * crash, but this makes the data consistent after a clean shutdown.
481  */
483 
484  /*
485  * Wake up logical walsenders holding logical failover slots after
486  * updating the restart_lsn of the physical slot.
487  */
489  }
490 
491  return retlsn;
492 }
493 
494 /*
495  * Helper function for advancing our logical replication slot forward.
496  *
497  * The slot's restart_lsn is used as start point for reading records, while
498  * confirmed_flush is used as base point for the decoding context.
499  *
500  * We cannot just do LogicalConfirmReceivedLocation to update confirmed_flush,
501  * because we need to digest WAL to advance restart_lsn allowing to recycle
502  * WAL and removal of old catalog tuples. As decoding is done in fast_forward
503  * mode, no changes are generated anyway.
504  */
505 static XLogRecPtr
507 {
509  ResourceOwner old_resowner = CurrentResourceOwner;
510  XLogRecPtr retlsn;
511 
512  Assert(moveto != InvalidXLogRecPtr);
513 
514  PG_TRY();
515  {
516  /*
517  * Create our decoding context in fast_forward mode, passing start_lsn
518  * as InvalidXLogRecPtr, so that we start processing from my slot's
519  * confirmed_flush.
520  */
522  NIL,
523  true, /* fast_forward */
524  XL_ROUTINE(.page_read = read_local_xlog_page,
525  .segment_open = wal_segment_open,
526  .segment_close = wal_segment_close),
527  NULL, NULL, NULL);
528 
529  /*
530  * Wait for specified streaming replication standby servers (if any)
531  * to confirm receipt of WAL up to moveto lsn.
532  */
534 
535  /*
536  * Start reading at the slot's restart_lsn, which we know to point to
537  * a valid record.
538  */
540 
541  /* invalidate non-timetravel entries */
543 
544  /* Decode records until we reach the requested target */
545  while (ctx->reader->EndRecPtr < moveto)
546  {
547  char *errm = NULL;
548  XLogRecord *record;
549 
550  /*
551  * Read records. No changes are generated in fast_forward mode,
552  * but snapbuilder/slot statuses are updated properly.
553  */
554  record = XLogReadRecord(ctx->reader, &errm);
555  if (errm)
556  elog(ERROR, "could not find record while advancing replication slot: %s",
557  errm);
558 
559  /*
560  * Process the record. Storage-level changes are ignored in
561  * fast_forward mode, but other modules (such as snapbuilder)
562  * might still have critical updates to do.
563  */
564  if (record)
566 
568  }
569 
570  /*
571  * Logical decoding could have clobbered CurrentResourceOwner during
572  * transaction management, so restore the executor's value. (This is
573  * a kluge, but it's not worth cleaning up right now.)
574  */
575  CurrentResourceOwner = old_resowner;
576 
577  if (ctx->reader->EndRecPtr != InvalidXLogRecPtr)
578  {
580 
581  /*
582  * If only the confirmed_flush LSN has changed the slot won't get
583  * marked as dirty by the above. Callers on the walsender
584  * interface are expected to keep track of their own progress and
585  * don't need it written out. But SQL-interface users cannot
586  * specify their own start positions and it's harder for them to
587  * keep track of their progress, so we should make more of an
588  * effort to save it for them.
589  *
590  * Dirty the slot so it is written out at the next checkpoint. The
591  * LSN position advanced to may still be lost on a crash but this
592  * makes the data consistent after a clean shutdown.
593  */
595  }
596 
598 
599  /* free context, call shutdown callback */
600  FreeDecodingContext(ctx);
601 
603  }
604  PG_CATCH();
605  {
606  /* clear all timetravel entries */
608 
609  PG_RE_THROW();
610  }
611  PG_END_TRY();
612 
613  return retlsn;
614 }
615 
616 /*
617  * SQL function for moving the position in a replication slot.
618  */
619 Datum
621 {
622  Name slotname = PG_GETARG_NAME(0);
623  XLogRecPtr moveto = PG_GETARG_LSN(1);
624  XLogRecPtr endlsn;
625  XLogRecPtr minlsn;
626  TupleDesc tupdesc;
627  Datum values[2];
628  bool nulls[2];
629  HeapTuple tuple;
630  Datum result;
631 
633 
635 
636  if (XLogRecPtrIsInvalid(moveto))
637  ereport(ERROR,
638  (errmsg("invalid target WAL LSN")));
639 
640  /* Build a tuple descriptor for our result type */
641  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
642  elog(ERROR, "return type must be a row type");
643 
644  /*
645  * We can't move slot past what's been flushed/replayed so clamp the
646  * target position accordingly.
647  */
648  if (!RecoveryInProgress())
649  moveto = Min(moveto, GetFlushRecPtr(NULL));
650  else
651  moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
652 
653  /* Acquire the slot so we "own" it */
654  ReplicationSlotAcquire(NameStr(*slotname), true);
655 
656  /* A slot whose restart_lsn has never been reserved cannot be advanced */
658  ereport(ERROR,
659  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
660  errmsg("replication slot \"%s\" cannot be advanced",
661  NameStr(*slotname)),
662  errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));
663 
664  /*
665  * Check if the slot is not moving backwards. Physical slots rely simply
666  * on restart_lsn as a minimum point, while logical slots have confirmed
667  * consumption up to confirmed_flush, meaning that in both cases data
668  * older than that is not available anymore.
669  */
672  else
674 
675  if (moveto < minlsn)
676  ereport(ERROR,
677  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
678  errmsg("cannot advance replication slot to %X/%X, minimum is %X/%X",
679  LSN_FORMAT_ARGS(moveto), LSN_FORMAT_ARGS(minlsn))));
680 
681  /* Do the actual slot update, depending on the slot type */
683  endlsn = pg_logical_replication_slot_advance(moveto);
684  else
685  endlsn = pg_physical_replication_slot_advance(moveto);
686 
688  nulls[0] = false;
689 
690  /*
691  * Recompute the minimum LSN and xmin across all slots to adjust with the
692  * advancing potentially done.
693  */
696 
698 
699  /* Return the reached position. */
700  values[1] = LSNGetDatum(endlsn);
701  nulls[1] = false;
702 
703  tuple = heap_form_tuple(tupdesc, values, nulls);
704  result = HeapTupleGetDatum(tuple);
705 
706  PG_RETURN_DATUM(result);
707 }
708 
709 /*
710  * Helper function of copying a replication slot.
711  */
712 static Datum
713 copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
714 {
715  Name src_name = PG_GETARG_NAME(0);
716  Name dst_name = PG_GETARG_NAME(1);
717  ReplicationSlot *src = NULL;
718  ReplicationSlot first_slot_contents;
719  ReplicationSlot second_slot_contents;
720  XLogRecPtr src_restart_lsn;
721  bool src_islogical;
722  bool temporary;
723  char *plugin;
724  Datum values[2];
725  bool nulls[2];
726  Datum result;
727  TupleDesc tupdesc;
728  HeapTuple tuple;
729 
730  if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
731  elog(ERROR, "return type must be a row type");
732 
734 
735  if (logical_slot)
737  else
739 
740  LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
741 
742  /*
743  * We need to prevent the source slot's reserved WAL from being removed,
744  * but we don't want to lock that slot for very long, and it can advance
745  * in the meantime. So obtain the source slot's data, and create a new
746  * slot using its restart_lsn. Afterwards we lock the source slot again
747  * and verify that the data we copied (name, type) has not changed
748  * incompatibly. No inconvenient WAL removal can occur once the new slot
749  * is created -- but since WAL removal could have occurred before we
750  * managed to create the new slot, we advance the new slot's restart_lsn
751  * to the source slot's updated restart_lsn the second time we lock it.
752  */
753  for (int i = 0; i < max_replication_slots; i++)
754  {
756 
757  if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
758  {
759  /* Copy the slot contents while holding spinlock */
760  SpinLockAcquire(&s->mutex);
761  first_slot_contents = *s;
762  SpinLockRelease(&s->mutex);
763  src = s;
764  break;
765  }
766  }
767 
768  LWLockRelease(ReplicationSlotControlLock);
769 
770  if (src == NULL)
771  ereport(ERROR,
772  (errcode(ERRCODE_UNDEFINED_OBJECT),
773  errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
774 
775  src_islogical = SlotIsLogical(&first_slot_contents);
776  src_restart_lsn = first_slot_contents.data.restart_lsn;
777  temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
778  plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
779 
780  /* Check type of replication slot */
781  if (src_islogical != logical_slot)
782  ereport(ERROR,
783  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
784  src_islogical ?
785  errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
786  NameStr(*src_name)) :
787  errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
788  NameStr(*src_name))));
789 
790  /* Copying non-reserved slot doesn't make sense */
791  if (XLogRecPtrIsInvalid(src_restart_lsn))
792  ereport(ERROR,
793  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
794  errmsg("cannot copy a replication slot that doesn't reserve WAL")));
795 
796  /* Overwrite params from optional arguments */
797  if (PG_NARGS() >= 3)
798  temporary = PG_GETARG_BOOL(2);
799  if (PG_NARGS() >= 4)
800  {
801  Assert(logical_slot);
802  plugin = NameStr(*(PG_GETARG_NAME(3)));
803  }
804 
805  /* Create new slot and acquire it */
806  if (logical_slot)
807  {
808  /*
809  * We must not try to read WAL, since we haven't reserved it yet --
810  * hence pass find_startpoint false. confirmed_flush will be set
811  * below, by copying from the source slot.
812  *
813  * To avoid potential issues with the slot synchronization where the
814  * restart_lsn of a replication slot can go backward, we set the
815  * failover option to false here. This situation occurs when a slot
816  * on the primary server is dropped and immediately replaced with a
817  * new slot of the same name, created by copying from another existing
818  * slot. However, the slot synchronization will only observe the
819  * restart_lsn of the same slot going backward.
820  */
822  plugin,
823  temporary,
824  false,
825  false,
826  src_restart_lsn,
827  false);
828  }
829  else
831  true,
832  temporary,
833  src_restart_lsn);
834 
835  /*
836  * Update the destination slot to current values of the source slot;
837  * recheck that the source slot is still the one we saw previously.
838  */
839  {
840  TransactionId copy_effective_xmin;
841  TransactionId copy_effective_catalog_xmin;
842  TransactionId copy_xmin;
843  TransactionId copy_catalog_xmin;
844  XLogRecPtr copy_restart_lsn;
845  XLogRecPtr copy_confirmed_flush;
846  bool copy_islogical;
847  char *copy_name;
848 
849  /* Copy data of source slot again */
850  SpinLockAcquire(&src->mutex);
851  second_slot_contents = *src;
852  SpinLockRelease(&src->mutex);
853 
854  copy_effective_xmin = second_slot_contents.effective_xmin;
855  copy_effective_catalog_xmin = second_slot_contents.effective_catalog_xmin;
856 
857  copy_xmin = second_slot_contents.data.xmin;
858  copy_catalog_xmin = second_slot_contents.data.catalog_xmin;
859  copy_restart_lsn = second_slot_contents.data.restart_lsn;
860  copy_confirmed_flush = second_slot_contents.data.confirmed_flush;
861 
862  /* for existence check */
863  copy_name = NameStr(second_slot_contents.data.name);
864  copy_islogical = SlotIsLogical(&second_slot_contents);
865 
866  /*
867  * Check if the source slot still exists and is valid. We regard it as
868  * invalid if the type of replication slot or name has been changed,
869  * or the restart_lsn either is invalid or has gone backward. (The
870  * restart_lsn could go backwards if the source slot is dropped and
871  * copied from an older slot during installation.)
872  *
873  * Since erroring out will release and drop the destination slot we
874  * don't need to release it here.
875  */
876  if (copy_restart_lsn < src_restart_lsn ||
877  src_islogical != copy_islogical ||
878  strcmp(copy_name, NameStr(*src_name)) != 0)
879  ereport(ERROR,
880  (errmsg("could not copy replication slot \"%s\"",
881  NameStr(*src_name)),
882  errdetail("The source replication slot was modified incompatibly during the copy operation.")));
883 
884  /* The source slot must have a consistent snapshot */
885  if (src_islogical && XLogRecPtrIsInvalid(copy_confirmed_flush))
886  ereport(ERROR,
887  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
888  errmsg("cannot copy unfinished logical replication slot \"%s\"",
889  NameStr(*src_name)),
890  errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));
891 
892  /* Install copied values again */
894  MyReplicationSlot->effective_xmin = copy_effective_xmin;
895  MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
896 
897  MyReplicationSlot->data.xmin = copy_xmin;
898  MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
899  MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
900  MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
902 
907 
908 #ifdef USE_ASSERT_CHECKING
909  /* Check that the restart_lsn is available */
910  {
911  XLogSegNo segno;
912 
913  XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
914  Assert(XLogGetLastRemovedSegno() < segno);
915  }
916 #endif
917  }
918 
919  /* target slot fully created, mark as persistent if needed */
920  if (logical_slot && !temporary)
922 
923  /* All done. Set up the return values */
924  values[0] = NameGetDatum(dst_name);
925  nulls[0] = false;
927  {
929  nulls[1] = false;
930  }
931  else
932  nulls[1] = true;
933 
934  tuple = heap_form_tuple(tupdesc, values, nulls);
935  result = HeapTupleGetDatum(tuple);
936 
938 
939  PG_RETURN_DATUM(result);
940 }
941 
942 /* The wrappers below are all to appease opr_sanity */
943 Datum
945 {
946  return copy_replication_slot(fcinfo, true);
947 }
948 
949 Datum
951 {
952  return copy_replication_slot(fcinfo, true);
953 }
954 
955 Datum
957 {
958  return copy_replication_slot(fcinfo, true);
959 }
960 
961 Datum
963 {
964  return copy_replication_slot(fcinfo, false);
965 }
966 
967 Datum
969 {
970  return copy_replication_slot(fcinfo, false);
971 }
972 
973 /*
974  * Synchronize failover enabled replication slots to a standby server
975  * from the primary server.
976  */
977 Datum
979 {
981  char *err;
982  StringInfoData app_name;
983 
985 
986  if (!RecoveryInProgress())
987  ereport(ERROR,
988  errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
989  errmsg("replication slots can only be synchronized to a standby server"));
990 
992 
993  /* Load the libpq-specific functions */
994  load_file("libpqwalreceiver", false);
995 
997 
998  initStringInfo(&app_name);
999  if (cluster_name[0])
1000  appendStringInfo(&app_name, "%s_slotsync", cluster_name);
1001  else
1002  appendStringInfoString(&app_name, "slotsync");
1003 
1004  /* Connect to the primary server. */
1005  wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
1006  app_name.data, &err);
1007  pfree(app_name.data);
1008 
1009  if (!wrconn)
1010  ereport(ERROR,
1011  errcode(ERRCODE_CONNECTION_FAILURE),
1012  errmsg("could not connect to the primary server: %s", err));
1013 
1015 
1017 
1018  PG_RETURN_VOID();
1019 }
static Datum values[MAXATTR]
Definition: bootstrap.c:152
#define CStringGetTextDatum(s)
Definition: builtins.h:97
#define NameStr(name)
Definition: c.h:733
#define Min(x, y)
Definition: c.h:991
#define Max(x, y)
Definition: c.h:985
uint32 TransactionId
Definition: c.h:639
#define OidIsValid(objectId)
Definition: c.h:762
void LogicalDecodingProcessRecord(LogicalDecodingContext *ctx, XLogReaderState *record)
Definition: decode.c:88
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:144
int errdetail(const char *fmt,...)
Definition: elog.c:1205
int errhint(const char *fmt,...)
Definition: elog.c:1319
int errcode(int sqlerrcode)
Definition: elog.c:859
int errmsg(const char *fmt,...)
Definition: elog.c:1072
#define PG_RE_THROW()
Definition: elog.h:411
#define PG_TRY(...)
Definition: elog.h:370
#define PG_END_TRY(...)
Definition: elog.h:395
#define ERROR
Definition: elog.h:39
#define PG_CATCH(...)
Definition: elog.h:380
#define elog(elevel,...)
Definition: elog.h:224
#define ereport(elevel,...)
Definition: elog.h:149
void err(int eval, const char *fmt,...)
Definition: err.c:43
Datum Int64GetDatum(int64 X)
Definition: fmgr.c:1807
#define PG_RETURN_VOID()
Definition: fmgr.h:349
#define PG_NARGS()
Definition: fmgr.h:203
#define PG_GETARG_NAME(n)
Definition: fmgr.h:278
#define PG_GETARG_BOOL(n)
Definition: fmgr.h:274
#define PG_RETURN_DATUM(x)
Definition: fmgr.h:353
#define PG_FUNCTION_ARGS
Definition: fmgr.h:193
void InitMaterializedSRF(FunctionCallInfo fcinfo, bits32 flags)
Definition: funcapi.c:76
TypeFuncClass get_call_result_type(FunctionCallInfo fcinfo, Oid *resultTypeId, TupleDesc *resultTupleDesc)
Definition: funcapi.c:276
@ TYPEFUNC_COMPOSITE
Definition: funcapi.h:149
static Datum HeapTupleGetDatum(const HeapTupleData *tuple)
Definition: funcapi.h:230
char * cluster_name
Definition: guc_tables.c:538
HeapTuple heap_form_tuple(TupleDesc tupleDescriptor, const Datum *values, const bool *isnull)
Definition: heaptuple.c:1116
void InvalidateSystemCaches(void)
Definition: inval.c:792
int i
Definition: isn.c:73
Assert(fmt[strlen(fmt) - 1] !='\n')
void LogicalConfirmReceivedLocation(XLogRecPtr lsn)
Definition: logical.c:1827
void FreeDecodingContext(LogicalDecodingContext *ctx)
Definition: logical.c:686
void DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
Definition: logical.c:642
void CheckLogicalDecodingRequirements(void)
Definition: logical.c:108
LogicalDecodingContext * CreateInitDecodingContext(const char *plugin, List *output_plugin_options, bool need_full_snapshot, XLogRecPtr restart_lsn, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:328
LogicalDecodingContext * CreateDecodingContext(XLogRecPtr start_lsn, List *output_plugin_options, bool fast_forward, XLogReaderRoutine *xl_routine, LogicalOutputPluginWriterPrepareWrite prepare_write, LogicalOutputPluginWriterWrite do_write, LogicalOutputPluginWriterUpdateProgress update_progress)
Definition: logical.c:494
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1169
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1782
@ LW_SHARED
Definition: lwlock.h:115
void pfree(void *pointer)
Definition: mcxt.c:1508
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
#define NIL
Definition: pg_list.h:68
#define PG_GETARG_LSN(n)
Definition: pg_lsn.h:33
static Datum LSNGetDatum(XLogRecPtr X)
Definition: pg_lsn.h:28
static bool two_phase
static const char * plugin
static Datum TransactionIdGetDatum(TransactionId X)
Definition: postgres.h:272
uintptr_t Datum
Definition: postgres.h:64
static Datum BoolGetDatum(bool X)
Definition: postgres.h:102
static Datum ObjectIdGetDatum(Oid X)
Definition: postgres.h:252
static Datum NameGetDatum(const NameData *X)
Definition: postgres.h:373
static Datum Int32GetDatum(int32 X)
Definition: postgres.h:212
#define InvalidOid
Definition: postgres_ext.h:36
ResourceOwner CurrentResourceOwner
Definition: resowner.c:165
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency, bool two_phase, bool failover, bool synced)
Definition: slot.c:309
void ReplicationSlotMarkDirty(void)
Definition: slot.c:1009
void ReplicationSlotReserveWal(void)
Definition: slot.c:1400
void ReplicationSlotAcquire(const char *name, bool nowait)
Definition: slot.c:540
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
Definition: slot.c:1048
void ReplicationSlotPersist(void)
Definition: slot.c:1026
ReplicationSlot * MyReplicationSlot
Definition: slot.c:138
void ReplicationSlotDrop(const char *name, bool nowait)
Definition: slot.c:783
void ReplicationSlotSave(void)
Definition: slot.c:991
void CheckSlotPermissions(void)
Definition: slot.c:1383
const char *const SlotInvalidationCauses[]
Definition: slot.c:105
void ReplicationSlotRelease(void)
Definition: slot.c:652
int max_replication_slots
Definition: slot.c:141
ReplicationSlotCtlData * ReplicationSlotCtl
Definition: slot.c:135
void WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
Definition: slot.c:2746
void ReplicationSlotsComputeRequiredLSN(void)
Definition: slot.c:1104
void CheckSlotRequirements(void)
Definition: slot.c:1361
@ RS_PERSISTENT
Definition: slot.h:35
@ RS_EPHEMERAL
Definition: slot.h:36
@ RS_TEMPORARY
Definition: slot.h:37
#define SlotIsPhysical(slot)
Definition: slot.h:209
ReplicationSlotInvalidationCause
Definition: slot.h:48
@ RS_INVAL_HORIZON
Definition: slot.h:53
@ RS_INVAL_WAL_LEVEL
Definition: slot.h:55
@ RS_INVAL_NONE
Definition: slot.h:49
#define SlotIsLogical(slot)
Definition: slot.h:210
Datum pg_get_replication_slots(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:240
Datum pg_copy_physical_replication_slot_a(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:962
static void create_logical_replication_slot(char *name, char *plugin, bool temporary, bool two_phase, bool failover, XLogRecPtr restart_lsn, bool find_startpoint)
Definition: slotfuncs.c:121
Datum pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:173
#define PG_GET_REPLICATION_SLOTS_COLS
static Datum copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
Definition: slotfuncs.c:713
Datum pg_copy_logical_replication_slot_c(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:956
Datum pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:69
Datum pg_copy_logical_replication_slot_a(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:944
Datum pg_copy_physical_replication_slot_b(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:968
Datum pg_sync_replication_slots(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:978
Datum pg_copy_logical_replication_slot_b(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:950
static XLogRecPtr pg_logical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:506
static XLogRecPtr pg_physical_replication_slot_advance(XLogRecPtr moveto)
Definition: slotfuncs.c:463
Datum pg_replication_slot_advance(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:620
static void create_physical_replication_slot(char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
Definition: slotfuncs.c:40
Datum pg_drop_replication_slot(PG_FUNCTION_ARGS)
Definition: slotfuncs.c:222
void SyncReplicationSlots(WalReceiverConn *wrconn)
Definition: slotsync.c:1442
char * CheckAndGetDbnameFromConninfo(void)
Definition: slotsync.c:895
bool ValidateSlotSyncParams(int elevel)
Definition: slotsync.c:921
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:97
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:182
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
XLogReaderState * reader
Definition: logical.h:42
ReplicationSlot replication_slots[1]
Definition: slot.h:221
TransactionId xmin
Definition: slot.h:82
TransactionId catalog_xmin
Definition: slot.h:90
XLogRecPtr restart_lsn
Definition: slot.h:93
XLogRecPtr confirmed_flush
Definition: slot.h:104
ReplicationSlotPersistency persistency
Definition: slot.h:74
ReplicationSlotInvalidationCause invalidated
Definition: slot.h:96
TransactionId effective_catalog_xmin
Definition: slot.h:175
slock_t mutex
Definition: slot.h:151
pid_t active_pid
Definition: slot.h:157
bool in_use
Definition: slot.h:154
TransactionId effective_xmin
Definition: slot.h:174
ReplicationSlotPersistentData data
Definition: slot.h:178
TimestampTz inactive_since
Definition: slot.h:206
TupleDesc setDesc
Definition: execnodes.h:340
Tuplestorestate * setResult
Definition: execnodes.h:339
XLogRecPtr EndRecPtr
Definition: xlogreader.h:207
Definition: c.h:728
#define InvalidTransactionId
Definition: transam.h:31
void tuplestore_putvalues(Tuplestorestate *state, TupleDesc tdesc, const Datum *values, const bool *isnull)
Definition: tuplestore.c:750
static Datum TimestampTzGetDatum(TimestampTz X)
Definition: timestamp.h:52
const char * name
static WalReceiverConn * wrconn
Definition: walreceiver.c:92
#define walrcv_connect(conninfo, replication, logical, must_use_password, appname, err)
Definition: walreceiver.h:432
#define walrcv_disconnect(conn)
Definition: walreceiver.h:464
void PhysicalWakeupLogicalWalSnd(void)
Definition: walsender.c:1733
bool RecoveryInProgress(void)
Definition: xlog.c:6201
XLogSegNo XLogGetLastRemovedSegno(void)
Definition: xlog.c:3688
int wal_keep_size_mb
Definition: xlog.c:116
int wal_segment_size
Definition: xlog.c:143
WALAvailability GetWALAvailability(XLogRecPtr targetLSN)
Definition: xlog.c:7703
int max_slot_wal_keep_size_mb
Definition: xlog.c:135
XLogRecPtr GetFlushRecPtr(TimeLineID *insertTLI)
Definition: xlog.c:6366
XLogRecPtr GetXLogWriteRecPtr(void)
Definition: xlog.c:9282
WALAvailability
Definition: xlog.h:185
@ WALAVAIL_REMOVED
Definition: xlog.h:191
@ WALAVAIL_RESERVED
Definition: xlog.h:187
@ WALAVAIL_UNRESERVED
Definition: xlog.h:190
@ WALAVAIL_EXTENDED
Definition: xlog.h:188
@ WALAVAIL_INVALID_LSN
Definition: xlog.h:186
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define XLogMBVarToSegs(mbvar, wal_segsz_bytes)
#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
XLogRecord * XLogReadRecord(XLogReaderState *state, char **errormsg)
Definition: xlogreader.c:389
void XLogBeginRead(XLogReaderState *state, XLogRecPtr RecPtr)
Definition: xlogreader.c:231
#define XL_ROUTINE(...)
Definition: xlogreader.h:117
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)
char * PrimaryConnInfo
Definition: xlogrecovery.c:96
void wal_segment_close(XLogReaderState *state)
Definition: xlogutils.c:842
void wal_segment_open(XLogReaderState *state, XLogSegNo nextSegNo, TimeLineID *tli_p)
Definition: xlogutils.c:817
int read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page)
Definition: xlogutils.c:861