LCOV - code coverage report
Current view: top level - src/backend/replication - slotfuncs.c (source / functions) Hit Total Coverage
Test: PostgreSQL 17devel Lines: 317 337 94.1 %
Date: 2024-03-28 10:11:15 Functions: 15 16 93.8 %
Legend: Lines: hit not hit

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

Generated by: LCOV version 1.14