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