Line data Source code
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 72 : 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 72 : ReplicationSlotCreate(name, false,
47 : temporary ? RS_TEMPORARY : RS_PERSISTENT, false,
48 : false, false);
49 :
50 72 : if (immediately_reserve)
51 : {
52 : /* Reserve WAL as the user asked for it */
53 32 : if (XLogRecPtrIsInvalid(restart_lsn))
54 24 : ReplicationSlotReserveWal();
55 : else
56 8 : MyReplicationSlot->data.restart_lsn = restart_lsn;
57 :
58 : /* Write this slot to disk */
59 32 : ReplicationSlotMarkDirty();
60 32 : ReplicationSlotSave();
61 : }
62 72 : }
63 :
64 : /*
65 : * SQL function for creating a new physical (streaming replication)
66 : * replication slot.
67 : */
68 : Datum
69 64 : pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
70 : {
71 64 : Name name = PG_GETARG_NAME(0);
72 64 : bool immediately_reserve = PG_GETARG_BOOL(1);
73 64 : 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 64 : if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
81 0 : elog(ERROR, "return type must be a row type");
82 :
83 64 : CheckSlotPermissions();
84 :
85 64 : CheckSlotRequirements();
86 :
87 64 : create_physical_replication_slot(NameStr(*name),
88 : immediately_reserve,
89 : temporary,
90 : InvalidXLogRecPtr);
91 :
92 64 : values[0] = NameGetDatum(&MyReplicationSlot->data.name);
93 64 : nulls[0] = false;
94 :
95 64 : if (immediately_reserve)
96 : {
97 24 : values[1] = LSNGetDatum(MyReplicationSlot->data.restart_lsn);
98 24 : nulls[1] = false;
99 : }
100 : else
101 40 : nulls[1] = true;
102 :
103 64 : tuple = heap_form_tuple(tupdesc, values, nulls);
104 64 : result = HeapTupleGetDatum(tuple);
105 :
106 64 : ReplicationSlotRelease();
107 :
108 64 : 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 240 : 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 240 : 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 240 : 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 230 : ctx = CreateInitDecodingContext(plugin, NIL,
151 : false, /* just catalogs is OK */
152 : restart_lsn,
153 230 : 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 224 : if (find_startpoint)
163 212 : DecodingContextFindStartpoint(ctx);
164 :
165 : /* don't need the decoding context anymore */
166 220 : FreeDecodingContext(ctx);
167 220 : }
168 :
169 : /*
170 : * SQL function for creating a new logical replication slot.
171 : */
172 : Datum
173 228 : pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
174 : {
175 228 : Name name = PG_GETARG_NAME(0);
176 228 : Name plugin = PG_GETARG_NAME(1);
177 228 : bool temporary = PG_GETARG_BOOL(2);
178 228 : bool two_phase = PG_GETARG_BOOL(3);
179 228 : 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 228 : if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
187 0 : elog(ERROR, "return type must be a row type");
188 :
189 228 : CheckSlotPermissions();
190 :
191 226 : CheckLogicalDecodingRequirements();
192 :
193 226 : create_logical_replication_slot(NameStr(*name),
194 226 : NameStr(*plugin),
195 : temporary,
196 : two_phase,
197 : failover,
198 : InvalidXLogRecPtr,
199 : true);
200 :
201 208 : values[0] = NameGetDatum(&MyReplicationSlot->data.name);
202 208 : values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);
203 :
204 208 : memset(nulls, 0, sizeof(nulls));
205 :
206 208 : tuple = heap_form_tuple(tupdesc, values, nulls);
207 208 : result = HeapTupleGetDatum(tuple);
208 :
209 : /* ok, slot is now fully created, mark it as persistent if needed */
210 208 : if (!temporary)
211 198 : ReplicationSlotPersist();
212 208 : ReplicationSlotRelease();
213 :
214 208 : PG_RETURN_DATUM(result);
215 : }
216 :
217 :
218 : /*
219 : * SQL function for dropping a replication slot.
220 : */
221 : Datum
222 250 : pg_drop_replication_slot(PG_FUNCTION_ARGS)
223 : {
224 250 : Name name = PG_GETARG_NAME(0);
225 :
226 250 : CheckSlotPermissions();
227 :
228 246 : CheckSlotRequirements();
229 :
230 246 : ReplicationSlotDrop(NameStr(*name), true);
231 :
232 234 : 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 548 : pg_get_replication_slots(PG_FUNCTION_ARGS)
241 : {
242 : #define PG_GET_REPLICATION_SLOTS_COLS 19
243 548 : 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 548 : InitMaterializedSRF(fcinfo, 0);
254 :
255 548 : currlsn = GetXLogWriteRecPtr();
256 :
257 548 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
258 4496 : for (slotno = 0; slotno < max_replication_slots; slotno++)
259 : {
260 3948 : 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 3948 : if (!slot->in_use)
269 3048 : continue;
270 :
271 : /* Copy slot contents while holding spinlock, then examine at leisure */
272 900 : SpinLockAcquire(&slot->mutex);
273 900 : slot_contents = *slot;
274 900 : SpinLockRelease(&slot->mutex);
275 :
276 900 : memset(values, 0, sizeof(values));
277 900 : memset(nulls, 0, sizeof(nulls));
278 :
279 900 : i = 0;
280 900 : values[i++] = NameGetDatum(&slot_contents.data.name);
281 :
282 900 : if (slot_contents.data.database == InvalidOid)
283 252 : nulls[i++] = true;
284 : else
285 648 : values[i++] = NameGetDatum(&slot_contents.data.plugin);
286 :
287 900 : if (slot_contents.data.database == InvalidOid)
288 252 : values[i++] = CStringGetTextDatum("physical");
289 : else
290 648 : values[i++] = CStringGetTextDatum("logical");
291 :
292 900 : if (slot_contents.data.database == InvalidOid)
293 252 : nulls[i++] = true;
294 : else
295 648 : values[i++] = ObjectIdGetDatum(slot_contents.data.database);
296 :
297 900 : values[i++] = BoolGetDatum(slot_contents.data.persistency == RS_TEMPORARY);
298 900 : values[i++] = BoolGetDatum(slot_contents.active_pid != 0);
299 :
300 900 : if (slot_contents.active_pid != 0)
301 288 : values[i++] = Int32GetDatum(slot_contents.active_pid);
302 : else
303 612 : nulls[i++] = true;
304 :
305 900 : if (slot_contents.data.xmin != InvalidTransactionId)
306 120 : values[i++] = TransactionIdGetDatum(slot_contents.data.xmin);
307 : else
308 780 : nulls[i++] = true;
309 :
310 900 : if (slot_contents.data.catalog_xmin != InvalidTransactionId)
311 708 : values[i++] = TransactionIdGetDatum(slot_contents.data.catalog_xmin);
312 : else
313 192 : nulls[i++] = true;
314 :
315 900 : if (slot_contents.data.restart_lsn != InvalidXLogRecPtr)
316 872 : values[i++] = LSNGetDatum(slot_contents.data.restart_lsn);
317 : else
318 28 : nulls[i++] = true;
319 :
320 900 : if (slot_contents.data.confirmed_flush != InvalidXLogRecPtr)
321 602 : values[i++] = LSNGetDatum(slot_contents.data.confirmed_flush);
322 : else
323 298 : nulls[i++] = true;
324 :
325 : /*
326 : * If the slot has not been invalidated, test availability from
327 : * restart_lsn.
328 : */
329 900 : if (slot_contents.data.invalidated != RS_INVAL_NONE)
330 62 : walstate = WALAVAIL_REMOVED;
331 : else
332 838 : walstate = GetWALAvailability(slot_contents.data.restart_lsn);
333 :
334 900 : switch (walstate)
335 : {
336 22 : case WALAVAIL_INVALID_LSN:
337 22 : nulls[i++] = true;
338 22 : break;
339 :
340 810 : case WALAVAIL_RESERVED:
341 810 : values[i++] = CStringGetTextDatum("reserved");
342 810 : break;
343 :
344 4 : case WALAVAIL_EXTENDED:
345 4 : values[i++] = CStringGetTextDatum("extended");
346 4 : break;
347 :
348 2 : case WALAVAIL_UNRESERVED:
349 2 : values[i++] = CStringGetTextDatum("unreserved");
350 2 : break;
351 :
352 62 : 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 62 : if (!XLogRecPtrIsInvalid(slot_contents.data.restart_lsn))
365 : {
366 : int pid;
367 :
368 56 : SpinLockAcquire(&slot->mutex);
369 56 : pid = slot->active_pid;
370 56 : slot_contents.data.restart_lsn = slot->data.restart_lsn;
371 56 : SpinLockRelease(&slot->mutex);
372 56 : if (pid != 0)
373 : {
374 0 : values[i++] = CStringGetTextDatum("unreserved");
375 0 : walstate = WALAVAIL_UNRESERVED;
376 0 : break;
377 : }
378 : }
379 62 : values[i++] = CStringGetTextDatum("lost");
380 62 : 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 900 : if (walstate == WALAVAIL_REMOVED || max_slot_wal_keep_size_mb < 0)
388 890 : nulls[i++] = true;
389 : else
390 : {
391 : XLogSegNo targetSeg;
392 : uint64 slotKeepSegs;
393 : uint64 keepSegs;
394 : XLogSegNo failSeg;
395 : XLogRecPtr failLSN;
396 :
397 10 : XLByteToSeg(slot_contents.data.restart_lsn, targetSeg, wal_segment_size);
398 :
399 : /* determine how many segments can be kept by slots */
400 10 : slotKeepSegs = XLogMBVarToSegs(max_slot_wal_keep_size_mb, wal_segment_size);
401 : /* ditto for wal_keep_size */
402 10 : keepSegs = XLogMBVarToSegs(wal_keep_size_mb, wal_segment_size);
403 :
404 : /* if currpos reaches failLSN, we lose our segment */
405 10 : failSeg = targetSeg + Max(slotKeepSegs, keepSegs) + 1;
406 10 : XLogSegNoOffsetToRecPtr(failSeg, 0, wal_segment_size, failLSN);
407 :
408 10 : values[i++] = Int64GetDatum(failLSN - currlsn);
409 : }
410 :
411 900 : values[i++] = BoolGetDatum(slot_contents.data.two_phase);
412 :
413 900 : if (slot_contents.inactive_since > 0)
414 638 : values[i++] = TimestampTzGetDatum(slot_contents.inactive_since);
415 : else
416 262 : nulls[i++] = true;
417 :
418 900 : cause = slot_contents.data.invalidated;
419 :
420 900 : if (SlotIsPhysical(&slot_contents))
421 252 : 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 648 : if (cause == RS_INVAL_HORIZON ||
429 : cause == RS_INVAL_WAL_LEVEL)
430 56 : values[i++] = BoolGetDatum(true);
431 : else
432 592 : values[i++] = BoolGetDatum(false);
433 : }
434 :
435 900 : if (cause == RS_INVAL_NONE)
436 838 : nulls[i++] = true;
437 : else
438 62 : values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]);
439 :
440 900 : values[i++] = BoolGetDatum(slot_contents.data.failover);
441 :
442 900 : values[i++] = BoolGetDatum(slot_contents.data.synced);
443 :
444 : Assert(i == PG_GET_REPLICATION_SLOTS_COLS);
445 :
446 900 : tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
447 : values, nulls);
448 : }
449 :
450 548 : LWLockRelease(ReplicationSlotControlLock);
451 :
452 548 : 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 2 : pg_physical_replication_slot_advance(XLogRecPtr moveto)
464 : {
465 2 : XLogRecPtr startlsn = MyReplicationSlot->data.restart_lsn;
466 2 : XLogRecPtr retlsn = startlsn;
467 :
468 : Assert(moveto != InvalidXLogRecPtr);
469 :
470 2 : if (startlsn < moveto)
471 : {
472 2 : SpinLockAcquire(&MyReplicationSlot->mutex);
473 2 : MyReplicationSlot->data.restart_lsn = moveto;
474 2 : SpinLockRelease(&MyReplicationSlot->mutex);
475 2 : 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 2 : ReplicationSlotMarkDirty();
483 :
484 : /*
485 : * Wake up logical walsenders holding logical failover slots after
486 : * updating the restart_lsn of the physical slot.
487 : */
488 2 : PhysicalWakeupLogicalWalSnd();
489 : }
490 :
491 2 : return retlsn;
492 : }
493 :
494 : /*
495 : * Advance our logical replication slot forward. See
496 : * LogicalSlotAdvanceAndCheckSnapState for details.
497 : */
498 : static XLogRecPtr
499 10 : pg_logical_replication_slot_advance(XLogRecPtr moveto)
500 : {
501 10 : return LogicalSlotAdvanceAndCheckSnapState(moveto, NULL);
502 : }
503 :
504 : /*
505 : * SQL function for moving the position in a replication slot.
506 : */
507 : Datum
508 16 : pg_replication_slot_advance(PG_FUNCTION_ARGS)
509 : {
510 16 : Name slotname = PG_GETARG_NAME(0);
511 16 : 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 16 : CheckSlotPermissions();
523 :
524 16 : if (XLogRecPtrIsInvalid(moveto))
525 2 : 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 14 : if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
531 0 : 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 14 : if (!RecoveryInProgress())
538 14 : moveto = Min(moveto, GetFlushRecPtr(NULL));
539 : else
540 0 : moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
541 :
542 : /* Acquire the slot so we "own" it */
543 14 : ReplicationSlotAcquire(NameStr(*slotname), true);
544 :
545 : /* A slot whose restart_lsn has never been reserved cannot be advanced */
546 14 : if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
547 2 : 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 12 : if (OidIsValid(MyReplicationSlot->data.database))
560 10 : minlsn = MyReplicationSlot->data.confirmed_flush;
561 : else
562 2 : minlsn = MyReplicationSlot->data.restart_lsn;
563 :
564 12 : if (moveto < minlsn)
565 0 : 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 12 : if (OidIsValid(MyReplicationSlot->data.database))
572 10 : endlsn = pg_logical_replication_slot_advance(moveto);
573 : else
574 2 : endlsn = pg_physical_replication_slot_advance(moveto);
575 :
576 12 : values[0] = NameGetDatum(&MyReplicationSlot->data.name);
577 12 : 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 12 : ReplicationSlotsComputeRequiredXmin(false);
584 12 : ReplicationSlotsComputeRequiredLSN();
585 :
586 12 : ReplicationSlotRelease();
587 :
588 : /* Return the reached position. */
589 12 : values[1] = LSNGetDatum(endlsn);
590 12 : nulls[1] = false;
591 :
592 12 : tuple = heap_form_tuple(tupdesc, values, nulls);
593 12 : result = HeapTupleGetDatum(tuple);
594 :
595 12 : PG_RETURN_DATUM(result);
596 : }
597 :
598 : /*
599 : * Helper function of copying a replication slot.
600 : */
601 : static Datum
602 28 : copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
603 : {
604 28 : Name src_name = PG_GETARG_NAME(0);
605 28 : Name dst_name = PG_GETARG_NAME(1);
606 28 : 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 28 : if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
620 0 : elog(ERROR, "return type must be a row type");
621 :
622 28 : CheckSlotPermissions();
623 :
624 28 : if (logical_slot)
625 16 : CheckLogicalDecodingRequirements();
626 : else
627 12 : CheckSlotRequirements();
628 :
629 28 : 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 30 : for (int i = 0; i < max_replication_slots; i++)
643 : {
644 30 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
645 :
646 30 : if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
647 : {
648 : /* Copy the slot contents while holding spinlock */
649 28 : SpinLockAcquire(&s->mutex);
650 28 : first_slot_contents = *s;
651 28 : SpinLockRelease(&s->mutex);
652 28 : src = s;
653 28 : break;
654 : }
655 : }
656 :
657 28 : LWLockRelease(ReplicationSlotControlLock);
658 :
659 28 : if (src == NULL)
660 0 : ereport(ERROR,
661 : (errcode(ERRCODE_UNDEFINED_OBJECT),
662 : errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
663 :
664 28 : src_islogical = SlotIsLogical(&first_slot_contents);
665 28 : src_restart_lsn = first_slot_contents.data.restart_lsn;
666 28 : temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
667 28 : plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
668 :
669 : /* Check type of replication slot */
670 28 : if (src_islogical != logical_slot)
671 4 : 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 24 : if (XLogRecPtrIsInvalid(src_restart_lsn))
681 2 : 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 22 : if (PG_NARGS() >= 3)
687 12 : temporary = PG_GETARG_BOOL(2);
688 22 : if (PG_NARGS() >= 4)
689 : {
690 : Assert(logical_slot);
691 8 : plugin = NameStr(*(PG_GETARG_NAME(3)));
692 : }
693 :
694 : /* Create new slot and acquire it */
695 22 : 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 14 : 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 8 : 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 20 : SpinLockAcquire(&src->mutex);
740 20 : second_slot_contents = *src;
741 20 : SpinLockRelease(&src->mutex);
742 :
743 20 : copy_effective_xmin = second_slot_contents.effective_xmin;
744 20 : copy_effective_catalog_xmin = second_slot_contents.effective_catalog_xmin;
745 :
746 20 : copy_xmin = second_slot_contents.data.xmin;
747 20 : copy_catalog_xmin = second_slot_contents.data.catalog_xmin;
748 20 : copy_restart_lsn = second_slot_contents.data.restart_lsn;
749 20 : copy_confirmed_flush = second_slot_contents.data.confirmed_flush;
750 :
751 : /* for existence check */
752 20 : copy_name = NameStr(second_slot_contents.data.name);
753 20 : 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 20 : if (copy_restart_lsn < src_restart_lsn ||
766 20 : src_islogical != copy_islogical ||
767 20 : strcmp(copy_name, NameStr(*src_name)) != 0)
768 0 : 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 20 : if (src_islogical && XLogRecPtrIsInvalid(copy_confirmed_flush))
775 0 : 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 20 : SpinLockAcquire(&MyReplicationSlot->mutex);
783 20 : MyReplicationSlot->effective_xmin = copy_effective_xmin;
784 20 : MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
785 :
786 20 : MyReplicationSlot->data.xmin = copy_xmin;
787 20 : MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
788 20 : MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
789 20 : MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
790 20 : SpinLockRelease(&MyReplicationSlot->mutex);
791 :
792 20 : ReplicationSlotMarkDirty();
793 20 : ReplicationSlotsComputeRequiredXmin(false);
794 20 : ReplicationSlotsComputeRequiredLSN();
795 20 : 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 20 : if (logical_slot && !temporary)
810 6 : ReplicationSlotPersist();
811 :
812 : /* All done. Set up the return values */
813 20 : values[0] = NameGetDatum(dst_name);
814 20 : nulls[0] = false;
815 20 : if (!XLogRecPtrIsInvalid(MyReplicationSlot->data.confirmed_flush))
816 : {
817 12 : values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);
818 12 : nulls[1] = false;
819 : }
820 : else
821 8 : nulls[1] = true;
822 :
823 20 : tuple = heap_form_tuple(tupdesc, values, nulls);
824 20 : result = HeapTupleGetDatum(tuple);
825 :
826 20 : ReplicationSlotRelease();
827 :
828 20 : PG_RETURN_DATUM(result);
829 : }
830 :
831 : /* The wrappers below are all to appease opr_sanity */
832 : Datum
833 8 : pg_copy_logical_replication_slot_a(PG_FUNCTION_ARGS)
834 : {
835 8 : return copy_replication_slot(fcinfo, true);
836 : }
837 :
838 : Datum
839 0 : pg_copy_logical_replication_slot_b(PG_FUNCTION_ARGS)
840 : {
841 0 : return copy_replication_slot(fcinfo, true);
842 : }
843 :
844 : Datum
845 8 : pg_copy_logical_replication_slot_c(PG_FUNCTION_ARGS)
846 : {
847 8 : return copy_replication_slot(fcinfo, true);
848 : }
849 :
850 : Datum
851 4 : pg_copy_physical_replication_slot_a(PG_FUNCTION_ARGS)
852 : {
853 4 : return copy_replication_slot(fcinfo, false);
854 : }
855 :
856 : Datum
857 8 : pg_copy_physical_replication_slot_b(PG_FUNCTION_ARGS)
858 : {
859 8 : 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 22 : pg_sync_replication_slots(PG_FUNCTION_ARGS)
868 : {
869 : WalReceiverConn *wrconn;
870 : char *err;
871 : StringInfoData app_name;
872 :
873 22 : CheckSlotPermissions();
874 :
875 20 : if (!RecoveryInProgress())
876 2 : ereport(ERROR,
877 : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
878 : errmsg("replication slots can only be synchronized to a standby server"));
879 :
880 18 : ValidateSlotSyncParams(ERROR);
881 :
882 : /* Load the libpq-specific functions */
883 18 : load_file("libpqwalreceiver", false);
884 :
885 18 : (void) CheckAndGetDbnameFromConninfo();
886 :
887 16 : initStringInfo(&app_name);
888 16 : if (cluster_name[0])
889 16 : appendStringInfo(&app_name, "%s_slotsync", cluster_name);
890 : else
891 0 : appendStringInfoString(&app_name, "slotsync");
892 :
893 : /* Connect to the primary server. */
894 16 : wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
895 : app_name.data, &err);
896 16 : pfree(app_name.data);
897 :
898 16 : if (!wrconn)
899 0 : 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 16 : SyncReplicationSlots(wrconn);
905 :
906 14 : walrcv_disconnect(wrconn);
907 :
908 14 : PG_RETURN_VOID();
909 : }
|
