?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
nbtree: fix read page recheck typo.
[pgsql.git] / src / backend / replication / slotfuncs.c
blobc7bfbb15e0a18408e230ec245bdc87ee8a49e386
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 {
43 Assert(!MyReplicationSlot);
44
45 /* acquire replication slot, this will check for conflicting names */
46 ReplicationSlotCreate(name, false,
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))
54 ReplicationSlotReserveWal();
55 else
56 MyReplicationSlot->data.restart_lsn = restart_lsn;
57
58 /* Write this slot to disk */
59 ReplicationSlotMarkDirty();
60 ReplicationSlotSave();
61 }
62 }
63
64 /*
65 * SQL function for creating a new physical (streaming replication)
66 * replication slot.
67 */
68 Datum
69 pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
70 {
71 Name name = PG_GETARG_NAME(0);
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
83 CheckSlotPermissions();
84
85 CheckSlotRequirements();
86
87 create_physical_replication_slot(NameStr(*name),
88 immediately_reserve,
89 temporary,
90 InvalidXLogRecPtr);
91
92 values[0] = NameGetDatum(&MyReplicationSlot->data.name);
93 nulls[0] = false;
94
95 if (immediately_reserve)
96 {
97 values[1] = LSNGetDatum(MyReplicationSlot->data.restart_lsn);
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
106 ReplicationSlotRelease();
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
121 create_logical_replication_slot(char *name, char *plugin,
122 bool temporary, bool two_phase,
123 bool failover,
124 XLogRecPtr restart_lsn,
125 bool find_startpoint)
126 {
127 LogicalDecodingContext *ctx = NULL;
128
129 Assert(!MyReplicationSlot);
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 */
139 ReplicationSlotCreate(name, true,
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 */
150 ctx = CreateInitDecodingContext(plugin, NIL,
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)
163 DecodingContextFindStartpoint(ctx);
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
173 pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
174 {
175 Name name = PG_GETARG_NAME(0);
176 Name plugin = PG_GETARG_NAME(1);
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
189 CheckSlotPermissions();
190
191 CheckLogicalDecodingRequirements();
192
193 create_logical_replication_slot(NameStr(*name),
194 NameStr(*plugin),
195 temporary,
196 two_phase,
197 failover,
198 InvalidXLogRecPtr,
199 true);
200
201 values[0] = NameGetDatum(&MyReplicationSlot->data.name);
202 values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);
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)
211 ReplicationSlotPersist();
212 ReplicationSlotRelease();
213
214 PG_RETURN_DATUM(result);
215 }
216
217
218 /*
219 * SQL function for dropping a replication slot.
220 */
221 Datum
222 pg_drop_replication_slot(PG_FUNCTION_ARGS)
223 {
224 Name name = PG_GETARG_NAME(0);
225
226 CheckSlotPermissions();
227
228 CheckSlotRequirements();
229
230 ReplicationSlotDrop(NameStr(*name), true);
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
240 pg_get_replication_slots(PG_FUNCTION_ARGS)
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 {
260 ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[slotno];
261 ReplicationSlot slot_contents;
262 Datum values[PG_GET_REPLICATION_SLOTS_COLS];
263 bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
264 WALAvailability walstate;
265 int i;
266 ReplicationSlotInvalidationCause cause;
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 {
336 case WALAVAIL_INVALID_LSN:
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 */
400 slotKeepSegs = XLogMBVarToSegs(max_slot_wal_keep_size_mb, wal_segment_size);
401 /* ditto for wal_keep_size */
402 keepSegs = XLogMBVarToSegs(wal_keep_size_mb, wal_segment_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
438 values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]);
439
440 values[i++] = BoolGetDatum(slot_contents.data.failover);
441
442 values[i++] = BoolGetDatum(slot_contents.data.synced);
443
444 Assert(i == PG_GET_REPLICATION_SLOTS_COLS);
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
463 pg_physical_replication_slot_advance(XLogRecPtr moveto)
464 {
465 XLogRecPtr startlsn = MyReplicationSlot->data.restart_lsn;
466 XLogRecPtr retlsn = startlsn;
467
468 Assert(moveto != InvalidXLogRecPtr);
469
470 if (startlsn < moveto)
471 {
472 SpinLockAcquire(&MyReplicationSlot->mutex);
473 MyReplicationSlot->data.restart_lsn = moveto;
474 SpinLockRelease(&MyReplicationSlot->mutex);
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 */
482 ReplicationSlotMarkDirty();
483
484 /*
485 * Wake up logical walsenders holding logical failover slots after
486 * updating the restart_lsn of the physical slot.
487 */
488 PhysicalWakeupLogicalWalSnd();
489 }
490
491 return retlsn;
492 }
493
494 /*
495 * Advance our logical replication slot forward. See
496 * LogicalSlotAdvanceAndCheckSnapState for details.
497 */
498 static XLogRecPtr
499 pg_logical_replication_slot_advance(XLogRecPtr moveto)
500 {
501 return LogicalSlotAdvanceAndCheckSnapState(moveto, NULL);
502 }
503
504 /*
505 * SQL function for moving the position in a replication slot.
506 */
507 Datum
508 pg_replication_slot_advance(PG_FUNCTION_ARGS)
509 {
510 Name slotname = PG_GETARG_NAME(0);
511 XLogRecPtr moveto = PG_GETARG_LSN(1);
512 XLogRecPtr endlsn;
513 XLogRecPtr minlsn;
514 TupleDesc tupdesc;
515 Datum values[2];
516 bool nulls[2];
517 HeapTuple tuple;
518 Datum result;
519
520 Assert(!MyReplicationSlot);
521
522 CheckSlotPermissions();
523
524 if (XLogRecPtrIsInvalid(moveto))
525 ereport(ERROR,
526 (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
527 errmsg("invalid target WAL LSN")));
528
529 /* Build a tuple descriptor for our result type */
530 if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
531 elog(ERROR, "return type must be a row type");
532
533 /*
534 * We can't move slot past what's been flushed/replayed so clamp the
535 * target position accordingly.
536 */
537 if (!RecoveryInProgress())
538 moveto = Min(moveto, GetFlushRecPtr(NULL));
539 else
540 moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
541
542 /* Acquire the slot so we "own" it */
543 ReplicationSlotAcquire(NameStr(*slotname), true);
544
545 /* A slot whose restart_lsn has never been reserved cannot be advanced */
546 if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
547 ereport(ERROR,
548 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
549 errmsg("replication slot \"%s\" cannot be advanced",
550 NameStr(*slotname)),
551 errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));
552
553 /*
554 * Check if the slot is not moving backwards. Physical slots rely simply
555 * on restart_lsn as a minimum point, while logical slots have confirmed
556 * consumption up to confirmed_flush, meaning that in both cases data
557 * older than that is not available anymore.
558 */
559 if (OidIsValid(MyReplicationSlot->data.database))
560 minlsn = MyReplicationSlot->data.confirmed_flush;
561 else
562 minlsn = MyReplicationSlot->data.restart_lsn;
563
564 if (moveto < minlsn)
565 ereport(ERROR,
566 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
567 errmsg("cannot advance replication slot to %X/%X, minimum is %X/%X",
568 LSN_FORMAT_ARGS(moveto), LSN_FORMAT_ARGS(minlsn))));
569
570 /* Do the actual slot update, depending on the slot type */
571 if (OidIsValid(MyReplicationSlot->data.database))
572 endlsn = pg_logical_replication_slot_advance(moveto);
573 else
574 endlsn = pg_physical_replication_slot_advance(moveto);
575
576 values[0] = NameGetDatum(&MyReplicationSlot->data.name);
577 nulls[0] = false;
578
579 /*
580 * Recompute the minimum LSN and xmin across all slots to adjust with the
581 * advancing potentially done.
582 */
583 ReplicationSlotsComputeRequiredXmin(false);
584 ReplicationSlotsComputeRequiredLSN();
585
586 ReplicationSlotRelease();
587
588 /* Return the reached position. */
589 values[1] = LSNGetDatum(endlsn);
590 nulls[1] = false;
591
592 tuple = heap_form_tuple(tupdesc, values, nulls);
593 result = HeapTupleGetDatum(tuple);
594
595 PG_RETURN_DATUM(result);
596 }
597
598 /*
599 * Helper function of copying a replication slot.
600 */
601 static Datum
602 copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
603 {
604 Name src_name = PG_GETARG_NAME(0);
605 Name dst_name = PG_GETARG_NAME(1);
606 ReplicationSlot *src = NULL;
607 ReplicationSlot first_slot_contents;
608 ReplicationSlot second_slot_contents;
609 XLogRecPtr src_restart_lsn;
610 bool src_islogical;
611 bool temporary;
612 char *plugin;
613 Datum values[2];
614 bool nulls[2];
615 Datum result;
616 TupleDesc tupdesc;
617 HeapTuple tuple;
618
619 if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
620 elog(ERROR, "return type must be a row type");
621
622 CheckSlotPermissions();
623
624 if (logical_slot)
625 CheckLogicalDecodingRequirements();
626 else
627 CheckSlotRequirements();
628
629 LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
630
631 /*
632 * We need to prevent the source slot's reserved WAL from being removed,
633 * but we don't want to lock that slot for very long, and it can advance
634 * in the meantime. So obtain the source slot's data, and create a new
635 * slot using its restart_lsn. Afterwards we lock the source slot again
636 * and verify that the data we copied (name, type) has not changed
637 * incompatibly. No inconvenient WAL removal can occur once the new slot
638 * is created -- but since WAL removal could have occurred before we
639 * managed to create the new slot, we advance the new slot's restart_lsn
640 * to the source slot's updated restart_lsn the second time we lock it.
641 */
642 for (int i = 0; i < max_replication_slots; i++)
643 {
644 ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
645
646 if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
647 {
648 /* Copy the slot contents while holding spinlock */
649 SpinLockAcquire(&s->mutex);
650 first_slot_contents = *s;
651 SpinLockRelease(&s->mutex);
652 src = s;
653 break;
654 }
655 }
656
657 LWLockRelease(ReplicationSlotControlLock);
658
659 if (src == NULL)
660 ereport(ERROR,
661 (errcode(ERRCODE_UNDEFINED_OBJECT),
662 errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
663
664 src_islogical = SlotIsLogical(&first_slot_contents);
665 src_restart_lsn = first_slot_contents.data.restart_lsn;
666 temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
667 plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
668
669 /* Check type of replication slot */
670 if (src_islogical != logical_slot)
671 ereport(ERROR,
672 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
673 src_islogical ?
674 errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
675 NameStr(*src_name)) :
676 errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
677 NameStr(*src_name))));
678
679 /* Copying non-reserved slot doesn't make sense */
680 if (XLogRecPtrIsInvalid(src_restart_lsn))
681 ereport(ERROR,
682 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
683 errmsg("cannot copy a replication slot that doesn't reserve WAL")));
684
685 /* Overwrite params from optional arguments */
686 if (PG_NARGS() >= 3)
687 temporary = PG_GETARG_BOOL(2);
688 if (PG_NARGS() >= 4)
689 {
690 Assert(logical_slot);
691 plugin = NameStr(*(PG_GETARG_NAME(3)));
692 }
693
694 /* Create new slot and acquire it */
695 if (logical_slot)
696 {
697 /*
698 * We must not try to read WAL, since we haven't reserved it yet --
699 * hence pass find_startpoint false. confirmed_flush will be set
700 * below, by copying from the source slot.
701 *
702 * To avoid potential issues with the slot synchronization where the
703 * restart_lsn of a replication slot can go backward, we set the
704 * failover option to false here. This situation occurs when a slot
705 * on the primary server is dropped and immediately replaced with a
706 * new slot of the same name, created by copying from another existing
707 * slot. However, the slot synchronization will only observe the
708 * restart_lsn of the same slot going backward.
709 */
710 create_logical_replication_slot(NameStr(*dst_name),
711 plugin,
712 temporary,
713 false,
714 false,
715 src_restart_lsn,
716 false);
717 }
718 else
719 create_physical_replication_slot(NameStr(*dst_name),
720 true,
721 temporary,
722 src_restart_lsn);
723
724 /*
725 * Update the destination slot to current values of the source slot;
726 * recheck that the source slot is still the one we saw previously.
727 */
728 {
729 TransactionId copy_effective_xmin;
730 TransactionId copy_effective_catalog_xmin;
731 TransactionId copy_xmin;
732 TransactionId copy_catalog_xmin;
733 XLogRecPtr copy_restart_lsn;
734 XLogRecPtr copy_confirmed_flush;
735 bool copy_islogical;
736 char *copy_name;
737
738 /* Copy data of source slot again */
739 SpinLockAcquire(&src->mutex);
740 second_slot_contents = *src;
741 SpinLockRelease(&src->mutex);
742
743 copy_effective_xmin = second_slot_contents.effective_xmin;
744 copy_effective_catalog_xmin = second_slot_contents.effective_catalog_xmin;
745
746 copy_xmin = second_slot_contents.data.xmin;
747 copy_catalog_xmin = second_slot_contents.data.catalog_xmin;
748 copy_restart_lsn = second_slot_contents.data.restart_lsn;
749 copy_confirmed_flush = second_slot_contents.data.confirmed_flush;
750
751 /* for existence check */
752 copy_name = NameStr(second_slot_contents.data.name);
753 copy_islogical = SlotIsLogical(&second_slot_contents);
754
755 /*
756 * Check if the source slot still exists and is valid. We regard it as
757 * invalid if the type of replication slot or name has been changed,
758 * or the restart_lsn either is invalid or has gone backward. (The
759 * restart_lsn could go backwards if the source slot is dropped and
760 * copied from an older slot during installation.)
761 *
762 * Since erroring out will release and drop the destination slot we
763 * don't need to release it here.
764 */
765 if (copy_restart_lsn < src_restart_lsn ||
766 src_islogical != copy_islogical ||
767 strcmp(copy_name, NameStr(*src_name)) != 0)
768 ereport(ERROR,
769 (errmsg("could not copy replication slot \"%s\"",
770 NameStr(*src_name)),
771 errdetail("The source replication slot was modified incompatibly during the copy operation.")));
772
773 /* The source slot must have a consistent snapshot */
774 if (src_islogical && XLogRecPtrIsInvalid(copy_confirmed_flush))
775 ereport(ERROR,
776 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
777 errmsg("cannot copy unfinished logical replication slot \"%s\"",
778 NameStr(*src_name)),
779 errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));
780
781 /* Install copied values again */
782 SpinLockAcquire(&MyReplicationSlot->mutex);
783 MyReplicationSlot->effective_xmin = copy_effective_xmin;
784 MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
785
786 MyReplicationSlot->data.xmin = copy_xmin;
787 MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
788 MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
789 MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
790 SpinLockRelease(&MyReplicationSlot->mutex);
791
792 ReplicationSlotMarkDirty();
793 ReplicationSlotsComputeRequiredXmin(false);
794 ReplicationSlotsComputeRequiredLSN();
795 ReplicationSlotSave();
796
797 #ifdef USE_ASSERT_CHECKING
798 /* Check that the restart_lsn is available */
799 {
800 XLogSegNo segno;
801
802 XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
803 Assert(XLogGetLastRemovedSegno() < segno);
804 }
805 #endif
806 }
807
808 /* target slot fully created, mark as persistent if needed */
809 if (logical_slot && !temporary)
810 ReplicationSlotPersist();
811
812 /* All done. Set up the return values */
813 values[0] = NameGetDatum(dst_name);
814 nulls[0] = false;
815 if (!XLogRecPtrIsInvalid(MyReplicationSlot->data.confirmed_flush))
816 {
817 values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);
818 nulls[1] = false;
819 }
820 else
821 nulls[1] = true;
822
823 tuple = heap_form_tuple(tupdesc, values, nulls);
824 result = HeapTupleGetDatum(tuple);
825
826 ReplicationSlotRelease();
827
828 PG_RETURN_DATUM(result);
829 }
830
831 /* The wrappers below are all to appease opr_sanity */
832 Datum
833 pg_copy_logical_replication_slot_a(PG_FUNCTION_ARGS)
834 {
835 return copy_replication_slot(fcinfo, true);
836 }
837
838 Datum
839 pg_copy_logical_replication_slot_b(PG_FUNCTION_ARGS)
840 {
841 return copy_replication_slot(fcinfo, true);
842 }
843
844 Datum
845 pg_copy_logical_replication_slot_c(PG_FUNCTION_ARGS)
846 {
847 return copy_replication_slot(fcinfo, true);
848 }
849
850 Datum
851 pg_copy_physical_replication_slot_a(PG_FUNCTION_ARGS)
852 {
853 return copy_replication_slot(fcinfo, false);
854 }
855
856 Datum
857 pg_copy_physical_replication_slot_b(PG_FUNCTION_ARGS)
858 {
859 return copy_replication_slot(fcinfo, false);
860 }
861
862 /*
863 * Synchronize failover enabled replication slots to a standby server
864 * from the primary server.
865 */
866 Datum
867 pg_sync_replication_slots(PG_FUNCTION_ARGS)
868 {
869 WalReceiverConn *wrconn;
870 char *err;
871 StringInfoData app_name;
872
873 CheckSlotPermissions();
874
875 if (!RecoveryInProgress())
876 ereport(ERROR,
877 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
878 errmsg("replication slots can only be synchronized to a standby server"));
879
880 ValidateSlotSyncParams(ERROR);
881
882 /* Load the libpq-specific functions */
883 load_file("libpqwalreceiver", false);
884
885 (void) CheckAndGetDbnameFromConninfo();
886
887 initStringInfo(&app_name);
888 if (cluster_name[0])
889 appendStringInfo(&app_name, "%s_slotsync", cluster_name);
890 else
891 appendStringInfoString(&app_name, "slotsync");
892
893 /* Connect to the primary server. */
894 wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
895 app_name.data, &err);
896 pfree(app_name.data);
897
898 if (!wrconn)
899 ereport(ERROR,
900 errcode(ERRCODE_CONNECTION_FAILURE),
901 errmsg("synchronization worker \"%s\" could not connect to the primary server: %s",
902 app_name.data, err));
903
904 SyncReplicationSlots(wrconn);
905
906 walrcv_disconnect(wrconn);
907
908 PG_RETURN_VOID();
909 }
Postgresql Clone from git://git.postgresql.org/git/postgresql.git