Branch data Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * slotfuncs.c
4 : : * Support functions for replication slots
5 : : *
6 : : * Copyright (c) 2012-2026, 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 "replication/logical.h"
21 : : #include "replication/slot.h"
22 : : #include "replication/slotsync.h"
23 : : #include "storage/proc.h"
24 : : #include "utils/builtins.h"
25 : : #include "utils/guc.h"
26 : : #include "utils/pg_lsn.h"
27 : :
28 : : /*
29 : : * Map SlotSyncSkipReason enum values to human-readable names.
30 : : */
31 : : static const char *SlotSyncSkipReasonNames[] = {
32 : : [SS_SKIP_NONE] = "none",
33 : : [SS_SKIP_WAL_NOT_FLUSHED] = "wal_not_flushed",
34 : : [SS_SKIP_WAL_OR_ROWS_REMOVED] = "wal_or_rows_removed",
35 : : [SS_SKIP_NO_CONSISTENT_SNAPSHOT] = "no_consistent_snapshot",
36 : : [SS_SKIP_INVALID] = "slot_invalidated"
37 : : };
38 : :
39 : : /*
40 : : * Helper function for creating a new physical replication slot with
41 : : * given arguments. Note that this function doesn't release the created
42 : : * slot.
43 : : *
44 : : * If restart_lsn is a valid value, we use it without WAL reservation
45 : : * routine. So the caller must guarantee that WAL is available.
46 : : */
47 : : static void
48 : 48 : create_physical_replication_slot(char *name, bool immediately_reserve,
49 : : bool temporary, XLogRecPtr restart_lsn)
50 : : {
51 : : Assert(!MyReplicationSlot);
52 : :
53 : : /* acquire replication slot, this will check for conflicting names */
54 [ + + ]: 48 : ReplicationSlotCreate(name, false,
55 : : temporary ? RS_TEMPORARY : RS_PERSISTENT, false,
56 : : false, false, false);
57 : :
58 [ + + ]: 48 : if (immediately_reserve)
59 : : {
60 : : /* Reserve WAL as the user asked for it */
61 [ + + ]: 22 : if (!XLogRecPtrIsValid(restart_lsn))
62 : 17 : ReplicationSlotReserveWal();
63 : : else
64 : 5 : MyReplicationSlot->data.restart_lsn = restart_lsn;
65 : :
66 : : /* Write this slot to disk */
67 : 22 : ReplicationSlotMarkDirty();
68 : 22 : ReplicationSlotSave();
69 : : }
70 : 48 : }
71 : :
72 : : /*
73 : : * SQL function for creating a new physical (streaming replication)
74 : : * replication slot.
75 : : */
76 : : Datum
77 : 43 : pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
78 : : {
79 : 43 : Name name = PG_GETARG_NAME(0);
80 : 43 : bool immediately_reserve = PG_GETARG_BOOL(1);
81 : 43 : bool temporary = PG_GETARG_BOOL(2);
82 : : Datum values[2];
83 : : bool nulls[2];
84 : : TupleDesc tupdesc;
85 : : HeapTuple tuple;
86 : : Datum result;
87 : :
88 [ - + ]: 43 : if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
89 [ # # ]: 0 : elog(ERROR, "return type must be a row type");
90 : :
91 : 43 : CheckSlotPermissions();
92 : :
93 : 43 : CheckSlotRequirements(false);
94 : :
95 : 43 : create_physical_replication_slot(NameStr(*name),
96 : : immediately_reserve,
97 : : temporary,
98 : : InvalidXLogRecPtr);
99 : :
100 : 43 : values[0] = NameGetDatum(&MyReplicationSlot->data.name);
101 : 43 : nulls[0] = false;
102 : :
103 [ + + ]: 43 : if (immediately_reserve)
104 : : {
105 : 17 : values[1] = LSNGetDatum(MyReplicationSlot->data.restart_lsn);
106 : 17 : nulls[1] = false;
107 : : }
108 : : else
109 : 26 : nulls[1] = true;
110 : :
111 : 43 : tuple = heap_form_tuple(tupdesc, values, nulls);
112 : 43 : result = HeapTupleGetDatum(tuple);
113 : :
114 : 43 : ReplicationSlotRelease();
115 : :
116 : 43 : PG_RETURN_DATUM(result);
117 : : }
118 : :
119 : :
120 : : /*
121 : : * Helper function for creating a new logical replication slot with
122 : : * given arguments. Note that this function doesn't release the created
123 : : * slot.
124 : : *
125 : : * When find_startpoint is false, the slot's confirmed_flush is not set; it's
126 : : * caller's responsibility to ensure it's set to something sensible.
127 : : */
128 : : static void
129 : 154 : create_logical_replication_slot(char *name, char *plugin,
130 : : bool temporary, bool two_phase,
131 : : bool failover,
132 : : XLogRecPtr restart_lsn,
133 : : bool find_startpoint)
134 : : {
135 : 154 : LogicalDecodingContext *ctx = NULL;
136 : :
137 : : Assert(!MyReplicationSlot);
138 : :
139 : : /*
140 : : * Acquire a logical decoding slot, this will check for conflicting names.
141 : : * Initially create persistent slot as ephemeral - that allows us to
142 : : * nicely handle errors during initialization because it'll get dropped if
143 : : * this transaction fails. We'll make it persistent at the end. Temporary
144 : : * slots can be created as temporary from beginning as they get dropped on
145 : : * error as well.
146 : : */
147 [ + + ]: 154 : ReplicationSlotCreate(name, true,
148 : : temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase,
149 : : false, failover, false);
150 : :
151 : : /*
152 : : * Ensure the logical decoding is enabled before initializing the logical
153 : : * decoding context.
154 : : */
155 : 149 : EnsureLogicalDecodingEnabled();
156 : :
157 : : /*
158 : : * Outside of recovery, holding a valid logical slot prevents logical
159 : : * decoding from being disabled. During recovery, however, replaying a
160 : : * status change record can disable it at any time regardless of slot
161 : : * existence, so we cannot assert that it is still enabled here. That is
162 : : * harmless: such a replay invalidates slots, so this slot creation fails
163 : : * afterwards.
164 : : */
165 : : Assert(RecoveryInProgress() || IsLogicalDecodingEnabled());
166 : :
167 : : /*
168 : : * Create logical decoding context to find start point or, if we don't
169 : : * need it, to 1) bump slot's restart_lsn and xmin 2) check plugin sanity.
170 : : *
171 : : * Note: when !find_startpoint this is still important, because it's at
172 : : * this point that the output plugin is validated.
173 : : */
174 : 146 : ctx = CreateInitDecodingContext(plugin, NIL,
175 : : false, /* just catalogs is OK */
176 : : false, /* not repack */
177 : : restart_lsn,
178 : 146 : XL_ROUTINE(.page_read = read_local_xlog_page,
179 : : .segment_open = wal_segment_open,
180 : : .segment_close = wal_segment_close),
181 : : NULL, NULL, NULL);
182 : :
183 : : /*
184 : : * If caller needs us to determine the decoding start point, do so now.
185 : : * This might take a while.
186 : : */
187 [ + + ]: 140 : if (find_startpoint)
188 : 133 : DecodingContextFindStartpoint(ctx);
189 : :
190 : : /* don't need the decoding context anymore */
191 : 138 : FreeDecodingContext(ctx);
192 : 138 : }
193 : :
194 : : /*
195 : : * SQL function for creating a new logical replication slot.
196 : : */
197 : : Datum
198 : 147 : pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
199 : : {
200 : 147 : Name name = PG_GETARG_NAME(0);
201 : 147 : Name plugin = PG_GETARG_NAME(1);
202 : 147 : bool temporary = PG_GETARG_BOOL(2);
203 : 147 : bool two_phase = PG_GETARG_BOOL(3);
204 : 147 : bool failover = PG_GETARG_BOOL(4);
205 : : Datum result;
206 : : TupleDesc tupdesc;
207 : : HeapTuple tuple;
208 : : Datum values[2];
209 : : bool nulls[2];
210 : :
211 [ - + ]: 147 : if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
212 [ # # ]: 0 : elog(ERROR, "return type must be a row type");
213 : :
214 : 147 : CheckSlotPermissions();
215 : :
216 : 146 : CheckLogicalDecodingRequirements(false);
217 : :
218 : 146 : create_logical_replication_slot(NameStr(*name),
219 : 146 : NameStr(*plugin),
220 : : temporary,
221 : : two_phase,
222 : : failover,
223 : : InvalidXLogRecPtr,
224 : : true);
225 : :
226 : 131 : values[0] = NameGetDatum(&MyReplicationSlot->data.name);
227 : 131 : values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);
228 : :
229 : 131 : memset(nulls, 0, sizeof(nulls));
230 : :
231 : 131 : tuple = heap_form_tuple(tupdesc, values, nulls);
232 : 131 : result = HeapTupleGetDatum(tuple);
233 : :
234 : : /* ok, slot is now fully created, mark it as persistent if needed */
235 [ + + ]: 131 : if (!temporary)
236 : 125 : ReplicationSlotPersist();
237 : 131 : ReplicationSlotRelease();
238 : :
239 : 131 : PG_RETURN_DATUM(result);
240 : : }
241 : :
242 : :
243 : : /*
244 : : * SQL function for dropping a replication slot.
245 : : */
246 : : Datum
247 : 155 : pg_drop_replication_slot(PG_FUNCTION_ARGS)
248 : : {
249 : 155 : Name name = PG_GETARG_NAME(0);
250 : :
251 : 155 : CheckSlotPermissions();
252 : :
253 : 153 : CheckSlotRequirements(false);
254 : :
255 : 153 : ReplicationSlotDrop(NameStr(*name), true);
256 : :
257 : 146 : PG_RETURN_VOID();
258 : : }
259 : :
260 : : /*
261 : : * pg_get_replication_slots - SQL SRF showing all replication slots
262 : : * that currently exist on the database cluster.
263 : : */
264 : : Datum
265 : 393 : pg_get_replication_slots(PG_FUNCTION_ARGS)
266 : : {
267 : : #define PG_GET_REPLICATION_SLOTS_COLS 21
268 : 393 : ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
269 : : XLogRecPtr currlsn;
270 : : int slotno;
271 : :
272 : : /*
273 : : * We don't require any special permission to see this function's data
274 : : * because nothing should be sensitive. The most critical being the slot
275 : : * name, which shouldn't contain anything particularly sensitive.
276 : : */
277 : :
278 : 393 : InitMaterializedSRF(fcinfo, 0);
279 : :
280 : 393 : currlsn = GetXLogWriteRecPtr();
281 : :
282 : 393 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
283 [ + + ]: 5390 : for (slotno = 0; slotno < max_replication_slots + max_repack_replication_slots; slotno++)
284 : : {
285 : 4997 : ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[slotno];
286 : : ReplicationSlot slot_contents;
287 : : Datum values[PG_GET_REPLICATION_SLOTS_COLS];
288 : : bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
289 : : WALAvailability walstate;
290 : : int i;
291 : : ReplicationSlotInvalidationCause cause;
292 : :
293 [ + + ]: 4997 : if (!slot->in_use)
294 : 4336 : continue;
295 : :
296 : : /* Copy slot contents while holding spinlock, then examine at leisure */
297 : 661 : SpinLockAcquire(&slot->mutex);
298 : 661 : slot_contents = *slot;
299 : 661 : SpinLockRelease(&slot->mutex);
300 : :
301 : 661 : memset(values, 0, sizeof(values));
302 : 661 : memset(nulls, 0, sizeof(nulls));
303 : :
304 : 661 : i = 0;
305 : 661 : values[i++] = NameGetDatum(&slot_contents.data.name);
306 : :
307 [ + + ]: 661 : if (slot_contents.data.database == InvalidOid)
308 : 195 : nulls[i++] = true;
309 : : else
310 : 466 : values[i++] = NameGetDatum(&slot_contents.data.plugin);
311 : :
312 [ + + ]: 661 : if (slot_contents.data.database == InvalidOid)
313 : 195 : values[i++] = CStringGetTextDatum("physical");
314 : : else
315 : 466 : values[i++] = CStringGetTextDatum("logical");
316 : :
317 [ + + ]: 661 : if (slot_contents.data.database == InvalidOid)
318 : 195 : nulls[i++] = true;
319 : : else
320 : 466 : values[i++] = ObjectIdGetDatum(slot_contents.data.database);
321 : :
322 : 661 : values[i++] = BoolGetDatum(slot_contents.data.persistency == RS_TEMPORARY);
323 : 661 : values[i++] = BoolGetDatum(slot_contents.active_proc != INVALID_PROC_NUMBER);
324 : :
325 [ + + ]: 661 : if (slot_contents.active_proc != INVALID_PROC_NUMBER)
326 : 252 : values[i++] = Int32GetDatum(GetPGProcByNumber(slot_contents.active_proc)->pid);
327 : : else
328 : 409 : nulls[i++] = true;
329 : :
330 [ + + ]: 661 : if (slot_contents.data.xmin != InvalidTransactionId)
331 : 105 : values[i++] = TransactionIdGetDatum(slot_contents.data.xmin);
332 : : else
333 : 556 : nulls[i++] = true;
334 : :
335 [ + + ]: 661 : if (slot_contents.data.catalog_xmin != InvalidTransactionId)
336 : 513 : values[i++] = TransactionIdGetDatum(slot_contents.data.catalog_xmin);
337 : : else
338 : 148 : nulls[i++] = true;
339 : :
340 [ + + ]: 661 : if (XLogRecPtrIsValid(slot_contents.data.restart_lsn))
341 : 628 : values[i++] = LSNGetDatum(slot_contents.data.restart_lsn);
342 : : else
343 : 33 : nulls[i++] = true;
344 : :
345 [ + + ]: 661 : if (XLogRecPtrIsValid(slot_contents.data.confirmed_flush))
346 : 433 : values[i++] = LSNGetDatum(slot_contents.data.confirmed_flush);
347 : : else
348 : 228 : nulls[i++] = true;
349 : :
350 : : /*
351 : : * If the slot has not been invalidated, test availability from
352 : : * restart_lsn.
353 : : */
354 [ + + ]: 661 : if (slot_contents.data.invalidated != RS_INVAL_NONE)
355 : 37 : walstate = WALAVAIL_REMOVED;
356 : : else
357 : 624 : walstate = GetWALAvailability(slot_contents.data.restart_lsn);
358 : :
359 [ + + + + : 661 : switch (walstate)
+ - ]
360 : : {
361 : 29 : case WALAVAIL_INVALID_LSN:
362 : 29 : nulls[i++] = true;
363 : 29 : break;
364 : :
365 : 592 : case WALAVAIL_RESERVED:
366 : 592 : values[i++] = CStringGetTextDatum("reserved");
367 : 592 : break;
368 : :
369 : 2 : case WALAVAIL_EXTENDED:
370 : 2 : values[i++] = CStringGetTextDatum("extended");
371 : 2 : break;
372 : :
373 : 1 : case WALAVAIL_UNRESERVED:
374 : 1 : values[i++] = CStringGetTextDatum("unreserved");
375 : 1 : break;
376 : :
377 : 37 : case WALAVAIL_REMOVED:
378 : :
379 : : /*
380 : : * If we read the restart_lsn long enough ago, maybe that file
381 : : * has been removed by now. However, the walsender could have
382 : : * moved forward enough that it jumped to another file after
383 : : * we looked. If checkpointer signalled the process to
384 : : * termination, then it's definitely lost; but if a process is
385 : : * still alive, then "unreserved" seems more appropriate.
386 : : *
387 : : * If we do change it, save the state for safe_wal_size below.
388 : : */
389 [ + + ]: 37 : if (XLogRecPtrIsValid(slot_contents.data.restart_lsn))
390 : : {
391 : : ProcNumber procno;
392 : :
393 : 33 : SpinLockAcquire(&slot->mutex);
394 : 33 : procno = slot->active_proc;
395 : 33 : slot_contents.data.restart_lsn = slot->data.restart_lsn;
396 : 33 : SpinLockRelease(&slot->mutex);
397 [ - + ]: 33 : if (procno != INVALID_PROC_NUMBER)
398 : : {
399 : 0 : values[i++] = CStringGetTextDatum("unreserved");
400 : 0 : walstate = WALAVAIL_UNRESERVED;
401 : 0 : break;
402 : : }
403 : : }
404 : 37 : values[i++] = CStringGetTextDatum("lost");
405 : 37 : break;
406 : : }
407 : :
408 : : /*
409 : : * safe_wal_size is only computed for slots that have not been lost,
410 : : * and only if there's a configured maximum size.
411 : : */
412 [ + + + + ]: 661 : if (walstate == WALAVAIL_REMOVED || max_slot_wal_keep_size_mb < 0)
413 : 653 : nulls[i++] = true;
414 : : else
415 : : {
416 : : XLogSegNo targetSeg;
417 : : uint64 slotKeepSegs;
418 : : uint64 keepSegs;
419 : : XLogSegNo failSeg;
420 : : XLogRecPtr failLSN;
421 : :
422 : 8 : XLByteToSeg(slot_contents.data.restart_lsn, targetSeg, wal_segment_size);
423 : :
424 : : /* determine how many segments can be kept by slots */
425 : 8 : slotKeepSegs = XLogMBVarToSegs(max_slot_wal_keep_size_mb, wal_segment_size);
426 : : /* ditto for wal_keep_size */
427 : 8 : keepSegs = XLogMBVarToSegs(wal_keep_size_mb, wal_segment_size);
428 : :
429 : : /* if currpos reaches failLSN, we lose our segment */
430 : 8 : failSeg = targetSeg + Max(slotKeepSegs, keepSegs) + 1;
431 : 8 : XLogSegNoOffsetToRecPtr(failSeg, 0, wal_segment_size, failLSN);
432 : :
433 : 8 : values[i++] = Int64GetDatum(failLSN - currlsn);
434 : : }
435 : :
436 : 661 : values[i++] = BoolGetDatum(slot_contents.data.two_phase);
437 : :
438 [ + + ]: 661 : if (slot_contents.data.two_phase &&
439 [ + - ]: 18 : XLogRecPtrIsValid(slot_contents.data.two_phase_at))
440 : 18 : values[i++] = LSNGetDatum(slot_contents.data.two_phase_at);
441 : : else
442 : 643 : nulls[i++] = true;
443 : :
444 [ + + ]: 661 : if (slot_contents.inactive_since > 0)
445 : 420 : values[i++] = TimestampTzGetDatum(slot_contents.inactive_since);
446 : : else
447 : 241 : nulls[i++] = true;
448 : :
449 : 661 : cause = slot_contents.data.invalidated;
450 : :
451 [ + + ]: 661 : if (SlotIsPhysical(&slot_contents))
452 : 195 : nulls[i++] = true;
453 : : else
454 : : {
455 : : /*
456 : : * rows_removed and wal_level_insufficient are the only two
457 : : * reasons for the logical slot's conflict with recovery.
458 : : */
459 [ + + + + ]: 466 : if (cause == RS_INVAL_HORIZON ||
460 : : cause == RS_INVAL_WAL_LEVEL)
461 : 29 : values[i++] = BoolGetDatum(true);
462 : : else
463 : 437 : values[i++] = BoolGetDatum(false);
464 : : }
465 : :
466 [ + + ]: 661 : if (cause == RS_INVAL_NONE)
467 : 624 : nulls[i++] = true;
468 : : else
469 : 37 : values[i++] = CStringGetTextDatum(GetSlotInvalidationCauseName(cause));
470 : :
471 : 661 : values[i++] = BoolGetDatum(slot_contents.data.failover);
472 : :
473 : 661 : values[i++] = BoolGetDatum(slot_contents.data.synced);
474 : :
475 [ + + ]: 661 : if (slot_contents.slotsync_skip_reason == SS_SKIP_NONE)
476 : 657 : nulls[i++] = true;
477 : : else
478 : 4 : values[i++] = CStringGetTextDatum(SlotSyncSkipReasonNames[slot_contents.slotsync_skip_reason]);
479 : :
480 : : Assert(i == PG_GET_REPLICATION_SLOTS_COLS);
481 : :
482 : 661 : tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
483 : : values, nulls);
484 : : }
485 : :
486 : 393 : LWLockRelease(ReplicationSlotControlLock);
487 : :
488 : 393 : return (Datum) 0;
489 : : }
490 : :
491 : : /*
492 : : * Helper function for advancing our physical replication slot forward.
493 : : *
494 : : * The LSN position to move to is compared simply to the slot's restart_lsn,
495 : : * knowing that any position older than that would be removed by successive
496 : : * checkpoints.
497 : : */
498 : : static XLogRecPtr
499 : 7 : pg_physical_replication_slot_advance(XLogRecPtr moveto)
500 : : {
501 : 7 : XLogRecPtr startlsn = MyReplicationSlot->data.restart_lsn;
502 : 7 : XLogRecPtr retlsn = startlsn;
503 : :
504 : : Assert(XLogRecPtrIsValid(moveto));
505 : :
506 [ + - ]: 7 : if (startlsn < moveto)
507 : : {
508 : 7 : SpinLockAcquire(&MyReplicationSlot->mutex);
509 : 7 : MyReplicationSlot->data.restart_lsn = moveto;
510 : 7 : SpinLockRelease(&MyReplicationSlot->mutex);
511 : 7 : retlsn = moveto;
512 : :
513 : : /*
514 : : * Dirty the slot so as it is written out at the next checkpoint. Note
515 : : * that the LSN position advanced may still be lost in the event of a
516 : : * crash, but this makes the data consistent after a clean shutdown.
517 : : */
518 : 7 : ReplicationSlotMarkDirty();
519 : :
520 : : /*
521 : : * Wake up logical walsenders holding logical failover slots after
522 : : * updating the restart_lsn of the physical slot.
523 : : */
524 : 7 : PhysicalWakeupLogicalWalSnd();
525 : : }
526 : :
527 : 7 : return retlsn;
528 : : }
529 : :
530 : : /*
531 : : * Advance our logical replication slot forward. See
532 : : * LogicalSlotAdvanceAndCheckSnapState for details.
533 : : */
534 : : static XLogRecPtr
535 : 11 : pg_logical_replication_slot_advance(XLogRecPtr moveto)
536 : : {
537 : 11 : return LogicalSlotAdvanceAndCheckSnapState(moveto, NULL);
538 : : }
539 : :
540 : : /*
541 : : * SQL function for moving the position in a replication slot.
542 : : */
543 : : Datum
544 : 21 : pg_replication_slot_advance(PG_FUNCTION_ARGS)
545 : : {
546 : 21 : Name slotname = PG_GETARG_NAME(0);
547 : 21 : XLogRecPtr moveto = PG_GETARG_LSN(1);
548 : : XLogRecPtr endlsn;
549 : : XLogRecPtr minlsn;
550 : : TupleDesc tupdesc;
551 : : Datum values[2];
552 : : bool nulls[2];
553 : : HeapTuple tuple;
554 : : Datum result;
555 : :
556 : : Assert(!MyReplicationSlot);
557 : :
558 : 21 : CheckSlotPermissions();
559 : :
560 [ + + ]: 21 : if (!XLogRecPtrIsValid(moveto))
561 [ + - ]: 1 : ereport(ERROR,
562 : : (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
563 : : errmsg("invalid target WAL LSN")));
564 : :
565 : : /* Build a tuple descriptor for our result type */
566 [ - + ]: 20 : if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
567 [ # # ]: 0 : elog(ERROR, "return type must be a row type");
568 : :
569 : : /*
570 : : * We can't move slot past what's been flushed/replayed so clamp the
571 : : * target position accordingly.
572 : : */
573 [ + - ]: 20 : if (!RecoveryInProgress())
574 [ + + ]: 20 : moveto = Min(moveto, GetFlushRecPtr(NULL));
575 : : else
576 [ # # ]: 0 : moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
577 : :
578 : : /* Acquire the slot so we "own" it */
579 : 20 : ReplicationSlotAcquire(NameStr(*slotname), true, true);
580 : :
581 : : /* A slot whose restart_lsn has never been reserved cannot be advanced */
582 [ + + ]: 19 : if (!XLogRecPtrIsValid(MyReplicationSlot->data.restart_lsn))
583 [ + - ]: 1 : ereport(ERROR,
584 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
585 : : errmsg("replication slot \"%s\" cannot be advanced",
586 : : NameStr(*slotname)),
587 : : errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));
588 : :
589 : : /*
590 : : * Check if the slot is not moving backwards. Physical slots rely simply
591 : : * on restart_lsn as a minimum point, while logical slots have confirmed
592 : : * consumption up to confirmed_flush, meaning that in both cases data
593 : : * older than that is not available anymore.
594 : : */
595 [ + + ]: 18 : if (OidIsValid(MyReplicationSlot->data.database))
596 : 11 : minlsn = MyReplicationSlot->data.confirmed_flush;
597 : : else
598 : 7 : minlsn = MyReplicationSlot->data.restart_lsn;
599 : :
600 [ - + ]: 18 : if (moveto < minlsn)
601 [ # # ]: 0 : ereport(ERROR,
602 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
603 : : errmsg("cannot advance replication slot to %X/%08X, minimum is %X/%08X",
604 : : LSN_FORMAT_ARGS(moveto), LSN_FORMAT_ARGS(minlsn))));
605 : :
606 : : /* Do the actual slot update, depending on the slot type */
607 [ + + ]: 18 : if (OidIsValid(MyReplicationSlot->data.database))
608 : 11 : endlsn = pg_logical_replication_slot_advance(moveto);
609 : : else
610 : 7 : endlsn = pg_physical_replication_slot_advance(moveto);
611 : :
612 : 18 : values[0] = NameGetDatum(&MyReplicationSlot->data.name);
613 : 18 : nulls[0] = false;
614 : :
615 : : /*
616 : : * Recompute the minimum LSN and xmin across all slots to adjust with the
617 : : * advancing potentially done.
618 : : */
619 : 18 : ReplicationSlotsComputeRequiredXmin(false);
620 : 18 : ReplicationSlotsComputeRequiredLSN();
621 : :
622 : 18 : ReplicationSlotRelease();
623 : :
624 : : /* Return the reached position. */
625 : 18 : values[1] = LSNGetDatum(endlsn);
626 : 18 : nulls[1] = false;
627 : :
628 : 18 : tuple = heap_form_tuple(tupdesc, values, nulls);
629 : 18 : result = HeapTupleGetDatum(tuple);
630 : :
631 : 18 : PG_RETURN_DATUM(result);
632 : : }
633 : :
634 : : /*
635 : : * Helper function of copying a replication slot.
636 : : */
637 : : static Datum
638 : 17 : copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
639 : : {
640 : 17 : Name src_name = PG_GETARG_NAME(0);
641 : 17 : Name dst_name = PG_GETARG_NAME(1);
642 : 17 : ReplicationSlot *src = NULL;
643 : : ReplicationSlot first_slot_contents;
644 : : ReplicationSlot second_slot_contents;
645 : : XLogRecPtr src_restart_lsn;
646 : : bool src_islogical;
647 : : bool temporary;
648 : : char *plugin;
649 : : Datum values[2];
650 : : bool nulls[2];
651 : : Datum result;
652 : : TupleDesc tupdesc;
653 : : HeapTuple tuple;
654 : :
655 [ - + ]: 17 : if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
656 [ # # ]: 0 : elog(ERROR, "return type must be a row type");
657 : :
658 : 17 : CheckSlotPermissions();
659 : :
660 [ + + ]: 17 : if (logical_slot)
661 : 10 : CheckLogicalDecodingRequirements(false);
662 : : else
663 : 7 : CheckSlotRequirements(false);
664 : :
665 : 17 : LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
666 : :
667 : : /*
668 : : * We need to prevent the source slot's reserved WAL from being removed,
669 : : * but we don't want to lock that slot for very long, and it can advance
670 : : * in the meantime. So obtain the source slot's data, and create a new
671 : : * slot using its restart_lsn. Afterwards we lock the source slot again
672 : : * and verify that the data we copied (name, type) has not changed
673 : : * incompatibly. No inconvenient WAL removal can occur once the new slot
674 : : * is created -- but since WAL removal could have occurred before we
675 : : * managed to create the new slot, we advance the new slot's restart_lsn
676 : : * to the source slot's updated restart_lsn the second time we lock it.
677 : : */
678 [ + - ]: 20 : for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
679 : : {
680 : 20 : ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
681 : :
682 [ + - + + ]: 20 : if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
683 : : {
684 : : /* Copy the slot contents while holding spinlock */
685 : 17 : SpinLockAcquire(&s->mutex);
686 : 17 : first_slot_contents = *s;
687 : 17 : SpinLockRelease(&s->mutex);
688 : 17 : src = s;
689 : 17 : break;
690 : : }
691 : : }
692 : :
693 : 17 : LWLockRelease(ReplicationSlotControlLock);
694 : :
695 [ - + ]: 17 : if (src == NULL)
696 [ # # ]: 0 : ereport(ERROR,
697 : : (errcode(ERRCODE_UNDEFINED_OBJECT),
698 : : errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
699 : :
700 : 17 : src_islogical = SlotIsLogical(&first_slot_contents);
701 : 17 : src_restart_lsn = first_slot_contents.data.restart_lsn;
702 : 17 : temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
703 [ + + ]: 17 : plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
704 : :
705 : : /* Check type of replication slot */
706 [ + + ]: 17 : if (src_islogical != logical_slot)
707 [ + - + + ]: 2 : ereport(ERROR,
708 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
709 : : src_islogical ?
710 : : errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
711 : : NameStr(*src_name)) :
712 : : errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
713 : : NameStr(*src_name))));
714 : :
715 : : /* Copying non-reserved slot doesn't make sense */
716 [ + + ]: 15 : if (!XLogRecPtrIsValid(src_restart_lsn))
717 [ + - ]: 1 : ereport(ERROR,
718 : : (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
719 : : errmsg("cannot copy a replication slot that doesn't reserve WAL")));
720 : :
721 : : /* Cannot copy an invalidated replication slot */
722 [ + + ]: 14 : if (first_slot_contents.data.invalidated != RS_INVAL_NONE)
723 [ + - ]: 1 : ereport(ERROR,
724 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
725 : : errmsg("cannot copy invalidated replication slot \"%s\"",
726 : : NameStr(*src_name)));
727 : :
728 : : /* Overwrite params from optional arguments */
729 [ + + ]: 13 : if (PG_NARGS() >= 3)
730 : 6 : temporary = PG_GETARG_BOOL(2);
731 [ + + ]: 13 : if (PG_NARGS() >= 4)
732 : : {
733 : : Assert(logical_slot);
734 : 4 : plugin = NameStr(*(PG_GETARG_NAME(3)));
735 : : }
736 : :
737 : : /* Create new slot and acquire it */
738 [ + + ]: 13 : if (logical_slot)
739 : : {
740 : : /*
741 : : * We must not try to read WAL, since we haven't reserved it yet --
742 : : * hence pass find_startpoint false. confirmed_flush will be set
743 : : * below, by copying from the source slot.
744 : : *
745 : : * We don't copy the failover option to prevent potential issues with
746 : : * slot synchronization. For instance, if a slot was synchronized to
747 : : * the standby, then dropped on the primary, and immediately recreated
748 : : * by copying from another existing slot with much earlier restart_lsn
749 : : * and confirmed_flush_lsn, the slot synchronization would only
750 : : * observe the LSN of the same slot moving backward. As slot
751 : : * synchronization does not copy the restart_lsn and
752 : : * confirmed_flush_lsn backward (see update_local_synced_slot() for
753 : : * details), if a failover happens before the primary's slot catches
754 : : * up, logical replication cannot continue using the synchronized slot
755 : : * on the promoted standby because the slot retains the restart_lsn
756 : : * and confirmed_flush_lsn that are much later than expected.
757 : : */
758 : 8 : create_logical_replication_slot(NameStr(*dst_name),
759 : : plugin,
760 : : temporary,
761 : : false,
762 : : false,
763 : : src_restart_lsn,
764 : : false);
765 : : }
766 : : else
767 : 5 : create_physical_replication_slot(NameStr(*dst_name),
768 : : true,
769 : : temporary,
770 : : src_restart_lsn);
771 : :
772 : : /*
773 : : * Update the destination slot to current values of the source slot;
774 : : * recheck that the source slot is still the one we saw previously.
775 : : */
776 : : {
777 : : TransactionId copy_effective_xmin;
778 : : TransactionId copy_effective_catalog_xmin;
779 : : TransactionId copy_xmin;
780 : : TransactionId copy_catalog_xmin;
781 : : XLogRecPtr copy_restart_lsn;
782 : : XLogRecPtr copy_confirmed_flush;
783 : : bool copy_islogical;
784 : : char *copy_name;
785 : :
786 : : /* Copy data of source slot again */
787 : 12 : SpinLockAcquire(&src->mutex);
788 : 12 : second_slot_contents = *src;
789 : 12 : SpinLockRelease(&src->mutex);
790 : :
791 : 12 : copy_effective_xmin = second_slot_contents.effective_xmin;
792 : 12 : copy_effective_catalog_xmin = second_slot_contents.effective_catalog_xmin;
793 : :
794 : 12 : copy_xmin = second_slot_contents.data.xmin;
795 : 12 : copy_catalog_xmin = second_slot_contents.data.catalog_xmin;
796 : 12 : copy_restart_lsn = second_slot_contents.data.restart_lsn;
797 : 12 : copy_confirmed_flush = second_slot_contents.data.confirmed_flush;
798 : :
799 : : /* for existence check */
800 : 12 : copy_name = NameStr(second_slot_contents.data.name);
801 : 12 : copy_islogical = SlotIsLogical(&second_slot_contents);
802 : :
803 : : /*
804 : : * Check if the source slot still exists and is valid. We regard it as
805 : : * invalid if the type of replication slot or name has been changed,
806 : : * or the restart_lsn either is invalid or has gone backward. (The
807 : : * restart_lsn could go backwards if the source slot is dropped and
808 : : * copied from an older slot during installation.)
809 : : *
810 : : * Since erroring out will release and drop the destination slot we
811 : : * don't need to release it here.
812 : : */
813 [ + - + - ]: 12 : if (copy_restart_lsn < src_restart_lsn ||
814 : 12 : src_islogical != copy_islogical ||
815 [ - + ]: 12 : strcmp(copy_name, NameStr(*src_name)) != 0)
816 [ # # ]: 0 : ereport(ERROR,
817 : : (errmsg("could not copy replication slot \"%s\"",
818 : : NameStr(*src_name)),
819 : : errdetail("The source replication slot was modified incompatibly during the copy operation.")));
820 : :
821 : : /* The source slot must have a consistent snapshot */
822 [ + + - + ]: 12 : if (src_islogical && !XLogRecPtrIsValid(copy_confirmed_flush))
823 [ # # ]: 0 : ereport(ERROR,
824 : : (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
825 : : errmsg("cannot copy unfinished logical replication slot \"%s\"",
826 : : NameStr(*src_name)),
827 : : errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));
828 : :
829 : : /*
830 : : * Copying an invalid slot doesn't make sense. Note that the source
831 : : * slot can become invalid after we create the new slot and copy the
832 : : * data of source slot. This is possible because the operations in
833 : : * InvalidateObsoleteReplicationSlots() are not serialized with this
834 : : * function. Even though we can't detect such a case here, the copied
835 : : * slot will become invalid in the next checkpoint cycle.
836 : : */
837 [ - + ]: 12 : if (second_slot_contents.data.invalidated != RS_INVAL_NONE)
838 [ # # ]: 0 : ereport(ERROR,
839 : : errmsg("cannot copy replication slot \"%s\"",
840 : : NameStr(*src_name)),
841 : : errdetail("The source replication slot was invalidated during the copy operation."));
842 : :
843 : : /* Install copied values again */
844 : 12 : SpinLockAcquire(&MyReplicationSlot->mutex);
845 : 12 : MyReplicationSlot->effective_xmin = copy_effective_xmin;
846 : 12 : MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
847 : :
848 : 12 : MyReplicationSlot->data.xmin = copy_xmin;
849 : 12 : MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
850 : 12 : MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
851 : 12 : MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
852 : 12 : SpinLockRelease(&MyReplicationSlot->mutex);
853 : :
854 : 12 : ReplicationSlotMarkDirty();
855 : 12 : ReplicationSlotsComputeRequiredXmin(false);
856 : 12 : ReplicationSlotsComputeRequiredLSN();
857 : 12 : ReplicationSlotSave();
858 : :
859 : : #ifdef USE_ASSERT_CHECKING
860 : : /* Check that the restart_lsn is available */
861 : : {
862 : : XLogSegNo segno;
863 : :
864 : : XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
865 : : Assert(XLogGetLastRemovedSegno() < segno);
866 : : }
867 : : #endif
868 : : }
869 : :
870 : : /* target slot fully created, mark as persistent if needed */
871 [ + + + + ]: 12 : if (logical_slot && !temporary)
872 : 4 : ReplicationSlotPersist();
873 : :
874 : : /* All done. Set up the return values */
875 : 12 : values[0] = NameGetDatum(dst_name);
876 : 12 : nulls[0] = false;
877 [ + + ]: 12 : if (XLogRecPtrIsValid(MyReplicationSlot->data.confirmed_flush))
878 : : {
879 : 7 : values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);
880 : 7 : nulls[1] = false;
881 : : }
882 : : else
883 : 5 : nulls[1] = true;
884 : :
885 : 12 : tuple = heap_form_tuple(tupdesc, values, nulls);
886 : 12 : result = HeapTupleGetDatum(tuple);
887 : :
888 : 12 : ReplicationSlotRelease();
889 : :
890 : 12 : PG_RETURN_DATUM(result);
891 : : }
892 : :
893 : : /* The wrappers below are all to appease opr_sanity */
894 : : Datum
895 : 4 : pg_copy_logical_replication_slot_a(PG_FUNCTION_ARGS)
896 : : {
897 : 4 : return copy_replication_slot(fcinfo, true);
898 : : }
899 : :
900 : : Datum
901 : 0 : pg_copy_logical_replication_slot_b(PG_FUNCTION_ARGS)
902 : : {
903 : 0 : return copy_replication_slot(fcinfo, true);
904 : : }
905 : :
906 : : Datum
907 : 6 : pg_copy_logical_replication_slot_c(PG_FUNCTION_ARGS)
908 : : {
909 : 6 : return copy_replication_slot(fcinfo, true);
910 : : }
911 : :
912 : : Datum
913 : 2 : pg_copy_physical_replication_slot_a(PG_FUNCTION_ARGS)
914 : : {
915 : 2 : return copy_replication_slot(fcinfo, false);
916 : : }
917 : :
918 : : Datum
919 : 5 : pg_copy_physical_replication_slot_b(PG_FUNCTION_ARGS)
920 : : {
921 : 5 : return copy_replication_slot(fcinfo, false);
922 : : }
923 : :
924 : : /*
925 : : * Synchronize failover enabled replication slots to a standby server
926 : : * from the primary server.
927 : : */
928 : : Datum
929 : 13 : pg_sync_replication_slots(PG_FUNCTION_ARGS)
930 : : {
931 : : WalReceiverConn *wrconn;
932 : : char *err;
933 : : StringInfoData app_name;
934 : :
935 : 13 : CheckSlotPermissions();
936 : :
937 [ + + ]: 12 : if (!RecoveryInProgress())
938 [ + - ]: 1 : ereport(ERROR,
939 : : errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
940 : : errmsg("replication slots can only be synchronized to a standby server"));
941 : :
942 : 11 : ValidateSlotSyncParams(ERROR);
943 : :
944 : : /* Load the libpq-specific functions */
945 : 11 : load_file("libpqwalreceiver", false);
946 : :
947 : 11 : (void) CheckAndGetDbnameFromConninfo();
948 : :
949 : 10 : initStringInfo(&app_name);
950 [ + - ]: 10 : if (cluster_name[0])
951 : 10 : appendStringInfo(&app_name, "%s_slotsync", cluster_name);
952 : : else
953 : 0 : appendStringInfoString(&app_name, "slotsync");
954 : :
955 : : /* Connect to the primary server. */
956 : 10 : wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
957 : : app_name.data, &err);
958 : :
959 [ - + ]: 10 : if (!wrconn)
960 [ # # ]: 0 : ereport(ERROR,
961 : : errcode(ERRCODE_CONNECTION_FAILURE),
962 : : errmsg("synchronization worker \"%s\" could not connect to the primary server: %s",
963 : : app_name.data, err));
964 : :
965 : 10 : pfree(app_name.data);
966 : :
967 : 10 : SyncReplicationSlots(wrconn);
968 : :
969 : 9 : walrcv_disconnect(wrconn);
970 : :
971 : 9 : PG_RETURN_VOID();
972 : : }
|
