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