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