slotfuncs.c File Reference

#include "postgres.h"
#include "access/htup_details.h"
#include "access/xlog_internal.h"
#include "access/xlogrecovery.h"
#include "access/xlogutils.h"
#include "funcapi.h"
#include "replication/logical.h"
#include "replication/slot.h"
#include "replication/slotsync.h"
#include "storage/proc.h"
#include "utils/builtins.h"
#include "utils/guc.h"
#include "utils/pg_lsn.h"

Include dependency graph for slotfuncs.c:

Go to the source code of this file.

Macros
#define	PG_GET_REPLICATION_SLOTS_COLS   21

Functions
static void	create_physical_replication_slot (char *name, bool immediately_reserve, bool temporary, XLogRecPtr restart_lsn)
Datum	pg_create_physical_replication_slot (PG_FUNCTION_ARGS)
static void	create_logical_replication_slot (char *name, char *plugin, bool temporary, bool two_phase, bool failover, XLogRecPtr restart_lsn, bool find_startpoint)
Datum	pg_create_logical_replication_slot (PG_FUNCTION_ARGS)
Datum	pg_drop_replication_slot (PG_FUNCTION_ARGS)
Datum	pg_get_replication_slots (PG_FUNCTION_ARGS)
static XLogRecPtr	pg_physical_replication_slot_advance (XLogRecPtr moveto)
static XLogRecPtr	pg_logical_replication_slot_advance (XLogRecPtr moveto)
Datum	pg_replication_slot_advance (PG_FUNCTION_ARGS)
static Datum	copy_replication_slot (FunctionCallInfo fcinfo, bool logical_slot)
Datum	pg_copy_logical_replication_slot_a (PG_FUNCTION_ARGS)
Datum	pg_copy_logical_replication_slot_b (PG_FUNCTION_ARGS)
Datum	pg_copy_logical_replication_slot_c (PG_FUNCTION_ARGS)
Datum	pg_copy_physical_replication_slot_a (PG_FUNCTION_ARGS)
Datum	pg_copy_physical_replication_slot_b (PG_FUNCTION_ARGS)
Datum	pg_sync_replication_slots (PG_FUNCTION_ARGS)

Variables
static const char *	SlotSyncSkipReasonNames []

Macro Definition Documentation

PG_GET_REPLICATION_SLOTS_COLS

#define PG_GET_REPLICATION_SLOTS_COLS   21

Function Documentation

copy_replication_slot()

static Datum copy_replication_slot ( FunctionCallInfo  fcinfo, bool  logical_slot  )

static

Definition at line 629 of file slotfuncs.c.

{
    Name        src_name = PG_GETARG_NAME(0);
    Name        dst_name = PG_GETARG_NAME(1);
    ReplicationSlot *src = NULL;
    ReplicationSlot first_slot_contents;
    ReplicationSlot second_slot_contents;
    XLogRecPtr  src_restart_lsn;
    bool        src_islogical;
    bool        temporary;
    char       *plugin;
    Datum       values[2];
    bool        nulls[2];
    Datum       result;
    TupleDesc   tupdesc;
    HeapTuple   tuple;
 
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    CheckSlotPermissions();
 
    if (logical_slot)
        CheckLogicalDecodingRequirements(false);
    else
        CheckSlotRequirements(false);
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    /*
     * We need to prevent the source slot's reserved WAL from being removed,
     * but we don't want to lock that slot for very long, and it can advance
     * in the meantime.  So obtain the source slot's data, and create a new
     * slot using its restart_lsn.  Afterwards we lock the source slot again
     * and verify that the data we copied (name, type) has not changed
     * incompatibly.  No inconvenient WAL removal can occur once the new slot
     * is created -- but since WAL removal could have occurred before we
     * managed to create the new slot, we advance the new slot's restart_lsn
     * to the source slot's updated restart_lsn the second time we lock it.
     */
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
        if (s->in_use && strcmp(NameStr(s->data.name), NameStr(*src_name)) == 0)
        {
            /* Copy the slot contents while holding spinlock */
            SpinLockAcquire(&s->mutex);
            first_slot_contents = *s;
            SpinLockRelease(&s->mutex);
            src = s;
            break;
        }
    }
 
    LWLockRelease(ReplicationSlotControlLock);
 
    if (src == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_OBJECT),
                 errmsg("replication slot \"%s\" does not exist", NameStr(*src_name))));
 
    src_islogical = SlotIsLogical(&first_slot_contents);
    src_restart_lsn = first_slot_contents.data.restart_lsn;
    temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
    plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
 
    /* Check type of replication slot */
    if (src_islogical != logical_slot)
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 src_islogical ?
                 errmsg("cannot copy physical replication slot \"%s\" as a logical replication slot",
                        NameStr(*src_name)) :
                 errmsg("cannot copy logical replication slot \"%s\" as a physical replication slot",
                        NameStr(*src_name))));
 
    /* Copying non-reserved slot doesn't make sense */
    if (!XLogRecPtrIsValid(src_restart_lsn))
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("cannot copy a replication slot that doesn't reserve WAL")));
 
    /* Cannot copy an invalidated replication slot */
    if (first_slot_contents.data.invalidated != RS_INVAL_NONE)
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("cannot copy invalidated replication slot \"%s\"",
                       NameStr(*src_name)));
 
    /* Overwrite params from optional arguments */
    if (PG_NARGS() >= 3)
        temporary = PG_GETARG_BOOL(2);
    if (PG_NARGS() >= 4)
    {
        Assert(logical_slot);
        plugin = NameStr(*(PG_GETARG_NAME(3)));
    }
 
    /* Create new slot and acquire it */
    if (logical_slot)
    {
        /*
         * We must not try to read WAL, since we haven't reserved it yet --
         * hence pass find_startpoint false.  confirmed_flush will be set
         * below, by copying from the source slot.
         *
         * We don't copy the failover option to prevent potential issues with
         * slot synchronization. For instance, if a slot was synchronized to
         * the standby, then dropped on the primary, and immediately recreated
         * by copying from another existing slot with much earlier restart_lsn
         * and confirmed_flush_lsn, the slot synchronization would only
         * observe the LSN of the same slot moving backward. As slot
         * synchronization does not copy the restart_lsn and
         * confirmed_flush_lsn backward (see update_local_synced_slot() for
         * details), if a failover happens before the primary's slot catches
         * up, logical replication cannot continue using the synchronized slot
         * on the promoted standby because the slot retains the restart_lsn
         * and confirmed_flush_lsn that are much later than expected.
         */
        create_logical_replication_slot(NameStr(*dst_name),
                                        plugin,
                                        temporary,
                                        false,
                                        false,
                                        src_restart_lsn,
                                        false);
    }
    else
        create_physical_replication_slot(NameStr(*dst_name),
                                         true,
                                         temporary,
                                         src_restart_lsn);
 
    /*
     * Update the destination slot to current values of the source slot;
     * recheck that the source slot is still the one we saw previously.
     */
    {
        TransactionId copy_effective_xmin;
        TransactionId copy_effective_catalog_xmin;
        TransactionId copy_xmin;
        TransactionId copy_catalog_xmin;
        XLogRecPtr  copy_restart_lsn;
        XLogRecPtr  copy_confirmed_flush;
        bool        copy_islogical;
        char       *copy_name;
 
        /* Copy data of source slot again */
        SpinLockAcquire(&src->mutex);
        second_slot_contents = *src;
        SpinLockRelease(&src->mutex);
 
        copy_effective_xmin = second_slot_contents.effective_xmin;
        copy_effective_catalog_xmin = second_slot_contents.effective_catalog_xmin;
 
        copy_xmin = second_slot_contents.data.xmin;
        copy_catalog_xmin = second_slot_contents.data.catalog_xmin;
        copy_restart_lsn = second_slot_contents.data.restart_lsn;
        copy_confirmed_flush = second_slot_contents.data.confirmed_flush;
 
        /* for existence check */
        copy_name = NameStr(second_slot_contents.data.name);
        copy_islogical = SlotIsLogical(&second_slot_contents);
 
        /*
         * Check if the source slot still exists and is valid. We regard it as
         * invalid if the type of replication slot or name has been changed,
         * or the restart_lsn either is invalid or has gone backward. (The
         * restart_lsn could go backwards if the source slot is dropped and
         * copied from an older slot during installation.)
         *
         * Since erroring out will release and drop the destination slot we
         * don't need to release it here.
         */
        if (copy_restart_lsn < src_restart_lsn ||
            src_islogical != copy_islogical ||
            strcmp(copy_name, NameStr(*src_name)) != 0)
            ereport(ERROR,
                    (errmsg("could not copy replication slot \"%s\"",
                            NameStr(*src_name)),
                     errdetail("The source replication slot was modified incompatibly during the copy operation.")));
 
        /* The source slot must have a consistent snapshot */
        if (src_islogical && !XLogRecPtrIsValid(copy_confirmed_flush))
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("cannot copy unfinished logical replication slot \"%s\"",
                            NameStr(*src_name)),
                     errhint("Retry when the source replication slot's confirmed_flush_lsn is valid.")));
 
        /*
         * Copying an invalid slot doesn't make sense. Note that the source
         * slot can become invalid after we create the new slot and copy the
         * data of source slot. This is possible because the operations in
         * InvalidateObsoleteReplicationSlots() are not serialized with this
         * function. Even though we can't detect such a case here, the copied
         * slot will become invalid in the next checkpoint cycle.
         */
        if (second_slot_contents.data.invalidated != RS_INVAL_NONE)
            ereport(ERROR,
                    errmsg("cannot copy replication slot \"%s\"",
                           NameStr(*src_name)),
                    errdetail("The source replication slot was invalidated during the copy operation."));
 
        /* Install copied values again */
        SpinLockAcquire(&MyReplicationSlot->mutex);
        MyReplicationSlot->effective_xmin = copy_effective_xmin;
        MyReplicationSlot->effective_catalog_xmin = copy_effective_catalog_xmin;
 
        MyReplicationSlot->data.xmin = copy_xmin;
        MyReplicationSlot->data.catalog_xmin = copy_catalog_xmin;
        MyReplicationSlot->data.restart_lsn = copy_restart_lsn;
        MyReplicationSlot->data.confirmed_flush = copy_confirmed_flush;
        SpinLockRelease(&MyReplicationSlot->mutex);
 
        ReplicationSlotMarkDirty();
        ReplicationSlotsComputeRequiredXmin(false);
        ReplicationSlotsComputeRequiredLSN();
        ReplicationSlotSave();
 
#ifdef USE_ASSERT_CHECKING
        /* Check that the restart_lsn is available */
        {
            XLogSegNo   segno;
 
            XLByteToSeg(copy_restart_lsn, segno, wal_segment_size);
            Assert(XLogGetLastRemovedSegno() < segno);
        }
#endif
    }
 
    /* target slot fully created, mark as persistent if needed */
    if (logical_slot && !temporary)
        ReplicationSlotPersist();
 
    /* All done.  Set up the return values */
    values[0] = NameGetDatum(dst_name);
    nulls[0] = false;
    if (XLogRecPtrIsValid(MyReplicationSlot->data.confirmed_flush))
    {
        values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);
        nulls[1] = false;
    }
    else
        nulls[1] = true;
 
    tuple = heap_form_tuple(tupdesc, values, nulls);
    result = HeapTupleGetDatum(tuple);
 
    ReplicationSlotRelease();
 
    PG_RETURN_DATUM(result);
}

References Assert, ReplicationSlotPersistentData::catalog_xmin, CheckLogicalDecodingRequirements(), CheckSlotPermissions(), CheckSlotRequirements(), ReplicationSlotPersistentData::confirmed_flush, create_logical_replication_slot(), create_physical_replication_slot(), ReplicationSlot::data, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, elog, ereport, errcode(), errdetail(), errhint(), errmsg, ERROR, fb(), get_call_result_type(), heap_form_tuple(), HeapTupleGetDatum(), i, ReplicationSlot::in_use, LSNGetDatum(), LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, MyReplicationSlot, ReplicationSlotPersistentData::name, NameGetDatum(), NameStr, PG_GETARG_BOOL, PG_GETARG_NAME, PG_NARGS, PG_RETURN_DATUM, plugin, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotMarkDirty(), ReplicationSlotPersist(), ReplicationSlotRelease(), ReplicationSlotSave(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotPersistentData::restart_lsn, result, RS_INVAL_NONE, RS_TEMPORARY, SlotIsLogical, SpinLockAcquire(), SpinLockRelease(), TYPEFUNC_COMPOSITE, values, wal_segment_size, XLByteToSeg, XLogGetLastRemovedSegno(), XLogRecPtrIsValid, and ReplicationSlotPersistentData::xmin.

Referenced by pg_copy_logical_replication_slot_a(), pg_copy_logical_replication_slot_b(), pg_copy_logical_replication_slot_c(), pg_copy_physical_replication_slot_a(), and pg_copy_physical_replication_slot_b().

create_logical_replication_slot()

static void create_logical_replication_slot ( char *  name, char *  plugin, bool  temporary, bool  two_phase, bool  failover, XLogRecPtr  restart_lsn, bool  find_startpoint  )

static

Definition at line 129 of file slotfuncs.c.

{
    LogicalDecodingContext *ctx = NULL;
 
    Assert(!MyReplicationSlot);
 
    /*
     * Acquire a logical decoding slot, this will check for conflicting names.
     * Initially create persistent slot as ephemeral - that allows us to
     * nicely handle errors during initialization because it'll get dropped if
     * this transaction fails. We'll make it persistent at the end. Temporary
     * slots can be created as temporary from beginning as they get dropped on
     * error as well.
     */
    ReplicationSlotCreate(name, true,
                          temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase,
                          false, failover, false);
 
    /*
     * Ensure the logical decoding is enabled before initializing the logical
     * decoding context.
     */
    EnsureLogicalDecodingEnabled();
    Assert(IsLogicalDecodingEnabled());
 
    /*
     * Create logical decoding context to find start point or, if we don't
     * need it, to 1) bump slot's restart_lsn and xmin 2) check plugin sanity.
     *
     * Note: when !find_startpoint this is still important, because it's at
     * this point that the output plugin is validated.
     */
    ctx = CreateInitDecodingContext(plugin, NIL,
                                    false,  /* just catalogs is OK */
                                    false,  /* not repack */
                                    restart_lsn,
                                    XL_ROUTINE(.page_read = read_local_xlog_page,
                                               .segment_open = wal_segment_open,
                                               .segment_close = wal_segment_close),
                                    NULL, NULL, NULL);
 
    /*
     * If caller needs us to determine the decoding start point, do so now.
     * This might take a while.
     */
    if (find_startpoint)
        DecodingContextFindStartpoint(ctx);
 
    /* don't need the decoding context anymore */
    FreeDecodingContext(ctx);
}

References Assert, CreateInitDecodingContext(), DecodingContextFindStartpoint(), EnsureLogicalDecodingEnabled(), failover, fb(), FreeDecodingContext(), IsLogicalDecodingEnabled(), MyReplicationSlot, name, NIL, plugin, read_local_xlog_page(), ReplicationSlotCreate(), RS_EPHEMERAL, RS_TEMPORARY, two_phase, wal_segment_close(), wal_segment_open(), and XL_ROUTINE.

Referenced by copy_replication_slot(), and pg_create_logical_replication_slot().

create_physical_replication_slot()

static void create_physical_replication_slot ( char *  name, bool  immediately_reserve, bool  temporary, XLogRecPtr  restart_lsn  )

static

Definition at line 48 of file slotfuncs.c.

{
    Assert(!MyReplicationSlot);
 
    /* acquire replication slot, this will check for conflicting names */
    ReplicationSlotCreate(name, false,
                          temporary ? RS_TEMPORARY : RS_PERSISTENT, false,
                          false, false, false);
 
    if (immediately_reserve)
    {
        /* Reserve WAL as the user asked for it */
        if (!XLogRecPtrIsValid(restart_lsn))
            ReplicationSlotReserveWal();
        else
            MyReplicationSlot->data.restart_lsn = restart_lsn;
 
        /* Write this slot to disk */
        ReplicationSlotMarkDirty();
        ReplicationSlotSave();
    }
}

References Assert, ReplicationSlot::data, fb(), MyReplicationSlot, name, ReplicationSlotCreate(), ReplicationSlotMarkDirty(), ReplicationSlotReserveWal(), ReplicationSlotSave(), ReplicationSlotPersistentData::restart_lsn, RS_PERSISTENT, RS_TEMPORARY, and XLogRecPtrIsValid.

Referenced by copy_replication_slot(), and pg_create_physical_replication_slot().

pg_copy_logical_replication_slot_a()

Datum pg_copy_logical_replication_slot_a

(

PG_FUNCTION_ARGS

)

Definition at line 886 of file slotfuncs.c.

{
    return copy_replication_slot(fcinfo, true);
}

References copy_replication_slot().

pg_copy_logical_replication_slot_b()

Datum pg_copy_logical_replication_slot_b

(

PG_FUNCTION_ARGS

)

Definition at line 892 of file slotfuncs.c.

{
    return copy_replication_slot(fcinfo, true);
}

References copy_replication_slot().

pg_copy_logical_replication_slot_c()

Datum pg_copy_logical_replication_slot_c

(

PG_FUNCTION_ARGS

)

Definition at line 898 of file slotfuncs.c.

{
    return copy_replication_slot(fcinfo, true);
}

References copy_replication_slot().

pg_copy_physical_replication_slot_a()

Datum pg_copy_physical_replication_slot_a

(

PG_FUNCTION_ARGS

)

Definition at line 904 of file slotfuncs.c.

{
    return copy_replication_slot(fcinfo, false);
}

References copy_replication_slot().

pg_copy_physical_replication_slot_b()

Datum pg_copy_physical_replication_slot_b

(

PG_FUNCTION_ARGS

)

Definition at line 910 of file slotfuncs.c.

{
    return copy_replication_slot(fcinfo, false);
}

References copy_replication_slot().

pg_create_logical_replication_slot()

Datum pg_create_logical_replication_slot

(

PG_FUNCTION_ARGS

)

Definition at line 189 of file slotfuncs.c.

{
    Name        name = PG_GETARG_NAME(0);
    Name        plugin = PG_GETARG_NAME(1);
    bool        temporary = PG_GETARG_BOOL(2);
    bool        two_phase = PG_GETARG_BOOL(3);
    bool        failover = PG_GETARG_BOOL(4);
    Datum       result;
    TupleDesc   tupdesc;
    HeapTuple   tuple;
    Datum       values[2];
    bool        nulls[2];
 
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    CheckSlotPermissions();
 
    CheckLogicalDecodingRequirements(false);
 
    create_logical_replication_slot(NameStr(*name),
                                    NameStr(*plugin),
                                    temporary,
                                    two_phase,
                                    failover,
                                    InvalidXLogRecPtr,
                                    true);
 
    values[0] = NameGetDatum(&MyReplicationSlot->data.name);
    values[1] = LSNGetDatum(MyReplicationSlot->data.confirmed_flush);
 
    memset(nulls, 0, sizeof(nulls));
 
    tuple = heap_form_tuple(tupdesc, values, nulls);
    result = HeapTupleGetDatum(tuple);
 
    /* ok, slot is now fully created, mark it as persistent if needed */
    if (!temporary)
        ReplicationSlotPersist();
    ReplicationSlotRelease();
 
    PG_RETURN_DATUM(result);
}

References CheckLogicalDecodingRequirements(), CheckSlotPermissions(), ReplicationSlotPersistentData::confirmed_flush, create_logical_replication_slot(), ReplicationSlot::data, elog, ERROR, failover, fb(), get_call_result_type(), heap_form_tuple(), HeapTupleGetDatum(), InvalidXLogRecPtr, LSNGetDatum(), MyReplicationSlot, name, ReplicationSlotPersistentData::name, NameGetDatum(), NameStr, PG_GETARG_BOOL, PG_GETARG_NAME, PG_RETURN_DATUM, plugin, ReplicationSlotPersist(), ReplicationSlotRelease(), result, two_phase, TYPEFUNC_COMPOSITE, and values.

pg_create_physical_replication_slot()

Datum pg_create_physical_replication_slot

(

PG_FUNCTION_ARGS

)

Definition at line 77 of file slotfuncs.c.

{
    Name        name = PG_GETARG_NAME(0);
    bool        immediately_reserve = PG_GETARG_BOOL(1);
    bool        temporary = PG_GETARG_BOOL(2);
    Datum       values[2];
    bool        nulls[2];
    TupleDesc   tupdesc;
    HeapTuple   tuple;
    Datum       result;
 
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    CheckSlotPermissions();
 
    CheckSlotRequirements(false);
 
    create_physical_replication_slot(NameStr(*name),
                                     immediately_reserve,
                                     temporary,
                                     InvalidXLogRecPtr);
 
    values[0] = NameGetDatum(&MyReplicationSlot->data.name);
    nulls[0] = false;
 
    if (immediately_reserve)
    {
        values[1] = LSNGetDatum(MyReplicationSlot->data.restart_lsn);
        nulls[1] = false;
    }
    else
        nulls[1] = true;
 
    tuple = heap_form_tuple(tupdesc, values, nulls);
    result = HeapTupleGetDatum(tuple);
 
    ReplicationSlotRelease();
 
    PG_RETURN_DATUM(result);
}

References CheckSlotPermissions(), CheckSlotRequirements(), create_physical_replication_slot(), ReplicationSlot::data, elog, ERROR, fb(), get_call_result_type(), heap_form_tuple(), HeapTupleGetDatum(), InvalidXLogRecPtr, LSNGetDatum(), MyReplicationSlot, name, ReplicationSlotPersistentData::name, NameGetDatum(), NameStr, PG_GETARG_BOOL, PG_GETARG_NAME, PG_RETURN_DATUM, ReplicationSlotRelease(), ReplicationSlotPersistentData::restart_lsn, result, TYPEFUNC_COMPOSITE, and values.

pg_drop_replication_slot()

Datum pg_drop_replication_slot

(

PG_FUNCTION_ARGS

)

Definition at line 238 of file slotfuncs.c.

{
    Name        name = PG_GETARG_NAME(0);
 
    CheckSlotPermissions();
 
    CheckSlotRequirements(false);
 
    ReplicationSlotDrop(NameStr(*name), true);
 
    PG_RETURN_VOID();
}

References CheckSlotPermissions(), CheckSlotRequirements(), name, NameStr, PG_GETARG_NAME, PG_RETURN_VOID, and ReplicationSlotDrop().

pg_get_replication_slots()

Datum pg_get_replication_slots

(

PG_FUNCTION_ARGS

)

Definition at line 256 of file slotfuncs.c.

{
#define PG_GET_REPLICATION_SLOTS_COLS 21
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    XLogRecPtr  currlsn;
    int         slotno;
 
    /*
     * We don't require any special permission to see this function's data
     * because nothing should be sensitive. The most critical being the slot
     * name, which shouldn't contain anything particularly sensitive.
     */
 
    InitMaterializedSRF(fcinfo, 0);
 
    currlsn = GetXLogWriteRecPtr();
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (slotno = 0; slotno < max_replication_slots + max_repack_replication_slots; slotno++)
    {
        ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[slotno];
        ReplicationSlot slot_contents;
        Datum       values[PG_GET_REPLICATION_SLOTS_COLS];
        bool        nulls[PG_GET_REPLICATION_SLOTS_COLS];
        WALAvailability walstate;
        int         i;
        ReplicationSlotInvalidationCause cause;
 
        if (!slot->in_use)
            continue;
 
        /* Copy slot contents while holding spinlock, then examine at leisure */
        SpinLockAcquire(&slot->mutex);
        slot_contents = *slot;
        SpinLockRelease(&slot->mutex);
 
        memset(values, 0, sizeof(values));
        memset(nulls, 0, sizeof(nulls));
 
        i = 0;
        values[i++] = NameGetDatum(&slot_contents.data.name);
 
        if (slot_contents.data.database == InvalidOid)
            nulls[i++] = true;
        else
            values[i++] = NameGetDatum(&slot_contents.data.plugin);
 
        if (slot_contents.data.database == InvalidOid)
            values[i++] = CStringGetTextDatum("physical");
        else
            values[i++] = CStringGetTextDatum("logical");
 
        if (slot_contents.data.database == InvalidOid)
            nulls[i++] = true;
        else
            values[i++] = ObjectIdGetDatum(slot_contents.data.database);
 
        values[i++] = BoolGetDatum(slot_contents.data.persistency == RS_TEMPORARY);
        values[i++] = BoolGetDatum(slot_contents.active_proc != INVALID_PROC_NUMBER);
 
        if (slot_contents.active_proc != INVALID_PROC_NUMBER)
            values[i++] = Int32GetDatum(GetPGProcByNumber(slot_contents.active_proc)->pid);
        else
            nulls[i++] = true;
 
        if (slot_contents.data.xmin != InvalidTransactionId)
            values[i++] = TransactionIdGetDatum(slot_contents.data.xmin);
        else
            nulls[i++] = true;
 
        if (slot_contents.data.catalog_xmin != InvalidTransactionId)
            values[i++] = TransactionIdGetDatum(slot_contents.data.catalog_xmin);
        else
            nulls[i++] = true;
 
        if (XLogRecPtrIsValid(slot_contents.data.restart_lsn))
            values[i++] = LSNGetDatum(slot_contents.data.restart_lsn);
        else
            nulls[i++] = true;
 
        if (XLogRecPtrIsValid(slot_contents.data.confirmed_flush))
            values[i++] = LSNGetDatum(slot_contents.data.confirmed_flush);
        else
            nulls[i++] = true;
 
        /*
         * If the slot has not been invalidated, test availability from
         * restart_lsn.
         */
        if (slot_contents.data.invalidated != RS_INVAL_NONE)
            walstate = WALAVAIL_REMOVED;
        else
            walstate = GetWALAvailability(slot_contents.data.restart_lsn);
 
        switch (walstate)
        {
            case WALAVAIL_INVALID_LSN:
                nulls[i++] = true;
                break;
 
            case WALAVAIL_RESERVED:
                values[i++] = CStringGetTextDatum("reserved");
                break;
 
            case WALAVAIL_EXTENDED:
                values[i++] = CStringGetTextDatum("extended");
                break;
 
            case WALAVAIL_UNRESERVED:
                values[i++] = CStringGetTextDatum("unreserved");
                break;
 
            case WALAVAIL_REMOVED:
 
                /*
                 * If we read the restart_lsn long enough ago, maybe that file
                 * has been removed by now.  However, the walsender could have
                 * moved forward enough that it jumped to another file after
                 * we looked.  If checkpointer signalled the process to
                 * termination, then it's definitely lost; but if a process is
                 * still alive, then "unreserved" seems more appropriate.
                 *
                 * If we do change it, save the state for safe_wal_size below.
                 */
                if (XLogRecPtrIsValid(slot_contents.data.restart_lsn))
                {
                    ProcNumber  procno;
 
                    SpinLockAcquire(&slot->mutex);
                    procno = slot->active_proc;
                    slot_contents.data.restart_lsn = slot->data.restart_lsn;
                    SpinLockRelease(&slot->mutex);
                    if (procno != INVALID_PROC_NUMBER)
                    {
                        values[i++] = CStringGetTextDatum("unreserved");
                        walstate = WALAVAIL_UNRESERVED;
                        break;
                    }
                }
                values[i++] = CStringGetTextDatum("lost");
                break;
        }
 
        /*
         * safe_wal_size is only computed for slots that have not been lost,
         * and only if there's a configured maximum size.
         */
        if (walstate == WALAVAIL_REMOVED || max_slot_wal_keep_size_mb < 0)
            nulls[i++] = true;
        else
        {
            XLogSegNo   targetSeg;
            uint64      slotKeepSegs;
            uint64      keepSegs;
            XLogSegNo   failSeg;
            XLogRecPtr  failLSN;
 
            XLByteToSeg(slot_contents.data.restart_lsn, targetSeg, wal_segment_size);
 
            /* determine how many segments can be kept by slots */
            slotKeepSegs = XLogMBVarToSegs(max_slot_wal_keep_size_mb, wal_segment_size);
            /* ditto for wal_keep_size */
            keepSegs = XLogMBVarToSegs(wal_keep_size_mb, wal_segment_size);
 
            /* if currpos reaches failLSN, we lose our segment */
            failSeg = targetSeg + Max(slotKeepSegs, keepSegs) + 1;
            XLogSegNoOffsetToRecPtr(failSeg, 0, wal_segment_size, failLSN);
 
            values[i++] = Int64GetDatum(failLSN - currlsn);
        }
 
        values[i++] = BoolGetDatum(slot_contents.data.two_phase);
 
        if (slot_contents.data.two_phase &&
            XLogRecPtrIsValid(slot_contents.data.two_phase_at))
            values[i++] = LSNGetDatum(slot_contents.data.two_phase_at);
        else
            nulls[i++] = true;
 
        if (slot_contents.inactive_since > 0)
            values[i++] = TimestampTzGetDatum(slot_contents.inactive_since);
        else
            nulls[i++] = true;
 
        cause = slot_contents.data.invalidated;
 
        if (SlotIsPhysical(&slot_contents))
            nulls[i++] = true;
        else
        {
            /*
             * rows_removed and wal_level_insufficient are the only two
             * reasons for the logical slot's conflict with recovery.
             */
            if (cause == RS_INVAL_HORIZON ||
                cause == RS_INVAL_WAL_LEVEL)
                values[i++] = BoolGetDatum(true);
            else
                values[i++] = BoolGetDatum(false);
        }
 
        if (cause == RS_INVAL_NONE)
            nulls[i++] = true;
        else
            values[i++] = CStringGetTextDatum(GetSlotInvalidationCauseName(cause));
 
        values[i++] = BoolGetDatum(slot_contents.data.failover);
 
        values[i++] = BoolGetDatum(slot_contents.data.synced);
 
        if (slot_contents.slotsync_skip_reason == SS_SKIP_NONE)
            nulls[i++] = true;
        else
            values[i++] = CStringGetTextDatum(SlotSyncSkipReasonNames[slot_contents.slotsync_skip_reason]);
 
        Assert(i == PG_GET_REPLICATION_SLOTS_COLS);
 
        tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
                             values, nulls);
    }
 
    LWLockRelease(ReplicationSlotControlLock);
 
    return (Datum) 0;
}

References ReplicationSlot::active_proc, Assert, BoolGetDatum(), CStringGetTextDatum, ReplicationSlot::data, fb(), GetPGProcByNumber, GetSlotInvalidationCauseName(), GetWALAvailability(), GetXLogWriteRecPtr(), i, ReplicationSlot::in_use, InitMaterializedSRF(), Int32GetDatum(), Int64GetDatum(), INVALID_PROC_NUMBER, InvalidOid, InvalidTransactionId, LSNGetDatum(), LW_SHARED, LWLockAcquire(), LWLockRelease(), Max, max_repack_replication_slots, max_replication_slots, max_slot_wal_keep_size_mb, ReplicationSlot::mutex, NameGetDatum(), ObjectIdGetDatum(), PG_GET_REPLICATION_SLOTS_COLS, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotPersistentData::restart_lsn, RS_INVAL_HORIZON, RS_INVAL_NONE, RS_INVAL_WAL_LEVEL, RS_TEMPORARY, SlotIsPhysical, SlotSyncSkipReasonNames, SpinLockAcquire(), SpinLockRelease(), SS_SKIP_NONE, TimestampTzGetDatum(), TransactionIdGetDatum(), tuplestore_putvalues(), values, wal_keep_size_mb, wal_segment_size, WALAVAIL_EXTENDED, WALAVAIL_INVALID_LSN, WALAVAIL_REMOVED, WALAVAIL_RESERVED, WALAVAIL_UNRESERVED, XLByteToSeg, XLogMBVarToSegs, XLogRecPtrIsValid, and XLogSegNoOffsetToRecPtr.

pg_logical_replication_slot_advance()

static XLogRecPtr pg_logical_replication_slot_advance ( XLogRecPtr  moveto )

static

Definition at line 526 of file slotfuncs.c.

{
    return LogicalSlotAdvanceAndCheckSnapState(moveto, NULL);
}

References fb(), and LogicalSlotAdvanceAndCheckSnapState().

Referenced by pg_replication_slot_advance().

pg_physical_replication_slot_advance()

static XLogRecPtr pg_physical_replication_slot_advance ( XLogRecPtr  moveto )

static

Definition at line 490 of file slotfuncs.c.

{
    XLogRecPtr  startlsn = MyReplicationSlot->data.restart_lsn;
    XLogRecPtr  retlsn = startlsn;
 
    Assert(XLogRecPtrIsValid(moveto));
 
    if (startlsn < moveto)
    {
        SpinLockAcquire(&MyReplicationSlot->mutex);
        MyReplicationSlot->data.restart_lsn = moveto;
        SpinLockRelease(&MyReplicationSlot->mutex);
        retlsn = moveto;
 
        /*
         * Dirty the slot so as it is written out at the next checkpoint. Note
         * that the LSN position advanced may still be lost in the event of a
         * crash, but this makes the data consistent after a clean shutdown.
         */
        ReplicationSlotMarkDirty();
 
        /*
         * Wake up logical walsenders holding logical failover slots after
         * updating the restart_lsn of the physical slot.
         */
        PhysicalWakeupLogicalWalSnd();
    }
 
    return retlsn;
}

References Assert, ReplicationSlot::data, fb(), ReplicationSlot::mutex, MyReplicationSlot, PhysicalWakeupLogicalWalSnd(), ReplicationSlotMarkDirty(), ReplicationSlotPersistentData::restart_lsn, SpinLockAcquire(), SpinLockRelease(), and XLogRecPtrIsValid.

Referenced by pg_replication_slot_advance().

pg_replication_slot_advance()

Datum pg_replication_slot_advance

(

PG_FUNCTION_ARGS

)

Definition at line 535 of file slotfuncs.c.

{
    Name        slotname = PG_GETARG_NAME(0);
    XLogRecPtr  moveto = PG_GETARG_LSN(1);
    XLogRecPtr  endlsn;
    XLogRecPtr  minlsn;
    TupleDesc   tupdesc;
    Datum       values[2];
    bool        nulls[2];
    HeapTuple   tuple;
    Datum       result;
 
    Assert(!MyReplicationSlot);
 
    CheckSlotPermissions();
 
    if (!XLogRecPtrIsValid(moveto))
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("invalid target WAL LSN")));
 
    /* Build a tuple descriptor for our result type */
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    /*
     * We can't move slot past what's been flushed/replayed so clamp the
     * target position accordingly.
     */
    if (!RecoveryInProgress())
        moveto = Min(moveto, GetFlushRecPtr(NULL));
    else
        moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
 
    /* Acquire the slot so we "own" it */
    ReplicationSlotAcquire(NameStr(*slotname), true, true);
 
    /* A slot whose restart_lsn has never been reserved cannot be advanced */
    if (!XLogRecPtrIsValid(MyReplicationSlot->data.restart_lsn))
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("replication slot \"%s\" cannot be advanced",
                        NameStr(*slotname)),
                 errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));
 
    /*
     * Check if the slot is not moving backwards.  Physical slots rely simply
     * on restart_lsn as a minimum point, while logical slots have confirmed
     * consumption up to confirmed_flush, meaning that in both cases data
     * older than that is not available anymore.
     */
    if (OidIsValid(MyReplicationSlot->data.database))
        minlsn = MyReplicationSlot->data.confirmed_flush;
    else
        minlsn = MyReplicationSlot->data.restart_lsn;
 
    if (moveto < minlsn)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("cannot advance replication slot to %X/%08X, minimum is %X/%08X",
                        LSN_FORMAT_ARGS(moveto), LSN_FORMAT_ARGS(minlsn))));
 
    /* Do the actual slot update, depending on the slot type */
    if (OidIsValid(MyReplicationSlot->data.database))
        endlsn = pg_logical_replication_slot_advance(moveto);
    else
        endlsn = pg_physical_replication_slot_advance(moveto);
 
    values[0] = NameGetDatum(&MyReplicationSlot->data.name);
    nulls[0] = false;
 
    /*
     * Recompute the minimum LSN and xmin across all slots to adjust with the
     * advancing potentially done.
     */
    ReplicationSlotsComputeRequiredXmin(false);
    ReplicationSlotsComputeRequiredLSN();
 
    ReplicationSlotRelease();
 
    /* Return the reached position. */
    values[1] = LSNGetDatum(endlsn);
    nulls[1] = false;
 
    tuple = heap_form_tuple(tupdesc, values, nulls);
    result = HeapTupleGetDatum(tuple);
 
    PG_RETURN_DATUM(result);
}

References Assert, CheckSlotPermissions(), ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, ReplicationSlotPersistentData::database, elog, ereport, errcode(), errdetail(), errmsg, ERROR, fb(), get_call_result_type(), GetFlushRecPtr(), GetXLogReplayRecPtr(), heap_form_tuple(), HeapTupleGetDatum(), LSN_FORMAT_ARGS, LSNGetDatum(), Min, MyReplicationSlot, ReplicationSlotPersistentData::name, NameGetDatum(), NameStr, OidIsValid, PG_GETARG_LSN, PG_GETARG_NAME, pg_logical_replication_slot_advance(), pg_physical_replication_slot_advance(), PG_RETURN_DATUM, RecoveryInProgress(), ReplicationSlotAcquire(), ReplicationSlotRelease(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotPersistentData::restart_lsn, result, TYPEFUNC_COMPOSITE, values, and XLogRecPtrIsValid.

pg_sync_replication_slots()

Datum pg_sync_replication_slots

(

PG_FUNCTION_ARGS

)

Definition at line 920 of file slotfuncs.c.

{
    WalReceiverConn *wrconn;
    char       *err;
    StringInfoData app_name;
 
    CheckSlotPermissions();
 
    if (!RecoveryInProgress())
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("replication slots can only be synchronized to a standby server"));
 
    ValidateSlotSyncParams(ERROR);
 
    /* Load the libpq-specific functions */
    load_file("libpqwalreceiver", false);
 
    (void) CheckAndGetDbnameFromConninfo();
 
    initStringInfo(&app_name);
    if (cluster_name[0])
        appendStringInfo(&app_name, "%s_slotsync", cluster_name);
    else
        appendStringInfoString(&app_name, "slotsync");
 
    /* Connect to the primary server. */
    wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
                            app_name.data, &err);
 
    if (!wrconn)
        ereport(ERROR,
                errcode(ERRCODE_CONNECTION_FAILURE),
                errmsg("synchronization worker \"%s\" could not connect to the primary server: %s",
                       app_name.data, err));
 
    pfree(app_name.data);
 
    SyncReplicationSlots(wrconn);
 
    walrcv_disconnect(wrconn);
 
    PG_RETURN_VOID();
}

References appendStringInfo(), appendStringInfoString(), CheckAndGetDbnameFromConninfo(), CheckSlotPermissions(), cluster_name, ereport, err(), errcode(), errmsg, ERROR, fb(), initStringInfo(), load_file(), pfree(), PG_RETURN_VOID, PrimaryConnInfo, RecoveryInProgress(), SyncReplicationSlots(), ValidateSlotSyncParams(), walrcv_connect, walrcv_disconnect, and wrconn.

Variable Documentation

SlotSyncSkipReasonNames

const char* SlotSyncSkipReasonNames[]

static

Initial value:

= {
    [SS_SKIP_NONE] = "none",
    [SS_SKIP_WAL_NOT_FLUSHED] = "wal_not_flushed",
    [SS_SKIP_WAL_OR_ROWS_REMOVED] = "wal_or_rows_removed",
    [SS_SKIP_NO_CONSISTENT_SNAPSHOT] = "no_consistent_snapshot",
    [SS_SKIP_INVALID] = "slot_invalidated"
}

Definition at line 31 of file slotfuncs.c.

                                               {
    [SS_SKIP_NONE] = "none",
    [SS_SKIP_WAL_NOT_FLUSHED] = "wal_not_flushed",
    [SS_SKIP_WAL_OR_ROWS_REMOVED] = "wal_or_rows_removed",
    [SS_SKIP_NO_CONSISTENT_SNAPSHOT] = "no_consistent_snapshot",
    [SS_SKIP_INVALID] = "slot_invalidated"
};

Referenced by pg_get_replication_slots().