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

Generated by: LCOV version 1.13