#include "postgres.h"
#include <unistd.h>
#include <sys/stat.h>
#include "access/transam.h"
#include "access/xlog_internal.h"
#include "access/xlogrecovery.h"
#include "common/file_utils.h"
#include "common/string.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/interrupt.h"
#include "replication/logicallauncher.h"
#include "replication/slotsync.h"
#include "replication/slot.h"
#include "replication/walsender_private.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/subsystems.h"
#include "utils/builtins.h"
#include "utils/guc_hooks.h"
#include "utils/injection_point.h"
#include "utils/varlena.h"
#include "utils/wait_event.h"

Include dependency graph for slot.c:

Data Structures
struct	ReplicationSlotOnDisk

struct	SyncStandbySlotsConfigData

struct	SlotInvalidationCauseMap

Macros
#define	ReplicationSlotOnDiskConstantSize offsetof(ReplicationSlotOnDisk, slotdata)

#define	ReplicationSlotOnDiskNotChecksummedSize offsetof(ReplicationSlotOnDisk, version)

#define	ReplicationSlotOnDiskChecksummedSize sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize

#define	ReplicationSlotOnDiskV2Size sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

#define	SLOT_MAGIC 0x1051CA1 /* format identifier */

#define	SLOT_VERSION 5 /* version for new files */

Typedefs
typedef struct ReplicationSlotOnDisk	ReplicationSlotOnDisk

typedef struct SlotInvalidationCauseMap	SlotInvalidationCauseMap

Functions
	StaticAssertDecl (lengthof(SlotInvalidationCauses)==(RS_INVAL_MAX_CAUSES+1), "array length mismatch")

static void	ReplicationSlotsShmemRequest (void *arg)

static void	ReplicationSlotsShmemInit (void *arg)

static void	ReplicationSlotShmemExit (int code, Datum arg)

static bool	IsSlotForConflictCheck (const char *name)

static void	ReplicationSlotDropPtr (ReplicationSlot *slot)

static void	RestoreSlotFromDisk (const char *name)

static void	CreateSlotOnDisk (ReplicationSlot *slot)

static void	SaveSlotToPath (ReplicationSlot slot, const char dir, int elevel)

void	ReplicationSlotInitialize (void)

bool	ReplicationSlotValidateName (const char *name, bool allow_reserved_name, int elevel)

bool	ReplicationSlotValidateNameInternal (const char name, bool allow_reserved_name, int err_code, char err_msg, char err_hint)

void	ReplicationSlotCreate (const char *name, bool db_specific, ReplicationSlotPersistency persistency, bool two_phase, bool repack, bool failover, bool synced)

ReplicationSlot *	SearchNamedReplicationSlot (const char *name, bool need_lock)

int	ReplicationSlotIndex (ReplicationSlot *slot)

bool	ReplicationSlotName (int index, Name name)

void	ReplicationSlotAcquire (const char *name, bool nowait, bool error_if_invalid)

void	ReplicationSlotRelease (void)

void	ReplicationSlotCleanup (bool synced_only)

void	ReplicationSlotDrop (const char *name, bool nowait)

void	ReplicationSlotAlter (const char name, const bool failover, const bool *two_phase)

void	ReplicationSlotDropAcquired (bool try_disable)

void	ReplicationSlotSave (void)

void	ReplicationSlotMarkDirty (void)

void	ReplicationSlotPersist (void)

void	ReplicationSlotsComputeRequiredXmin (bool already_locked)

void	ReplicationSlotsComputeRequiredLSN (void)

XLogRecPtr	ReplicationSlotsComputeLogicalRestartLSN (void)

bool	ReplicationSlotsCountDBSlots (Oid dboid, int nslots, int nactive)

void	ReplicationSlotsDropDBSlots (Oid dboid)

bool	CheckLogicalSlotExists (void)

void	CheckSlotRequirements (bool repack)

void	CheckSlotPermissions (void)

void	ReplicationSlotReserveWal (void)

static void	ReportSlotInvalidation (ReplicationSlotInvalidationCause cause, bool terminating, int pid, NameData slotname, XLogRecPtr restart_lsn, XLogRecPtr oldestLSN, TransactionId snapshotConflictHorizon, long slot_idle_seconds)

static bool	CanInvalidateIdleSlot (ReplicationSlot *s)

static ReplicationSlotInvalidationCause	DetermineSlotInvalidationCause (uint32 possible_causes, ReplicationSlot s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, TimestampTz inactive_since, TimestampTz now)

static bool	InvalidatePossiblyObsoleteSlot (uint32 possible_causes, ReplicationSlot s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, bool released_lock_out)

bool	InvalidateObsoleteReplicationSlots (uint32 possible_causes, XLogSegNo oldestSegno, Oid dboid, TransactionId snapshotConflictHorizon)

void	CheckPointReplicationSlots (bool is_shutdown)

void	StartupReplicationSlots (void)

ReplicationSlotInvalidationCause	GetSlotInvalidationCause (const char *cause_name)

const char *	GetSlotInvalidationCauseName (ReplicationSlotInvalidationCause cause)

static bool	validate_sync_standby_slots (char rawname, List *elemlist)

bool	check_synchronized_standby_slots (char newval, void extra, GucSource source)

void	assign_synchronized_standby_slots (const char newval, void extra)

bool	SlotExistsInSyncStandbySlots (const char *slot_name)

bool	StandbySlotsHaveCaughtup (XLogRecPtr wait_for_lsn, int elevel)

void	WaitForStandbyConfirmation (XLogRecPtr wait_for_lsn)

Macro Definition Documentation

◆ ReplicationSlotOnDiskChecksummedSize

#define ReplicationSlotOnDiskChecksummedSize sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize

Definition at line 137 of file slot.c.

                                                      {
    .request_fn = ReplicationSlotsShmemRequest,
    .init_fn = ReplicationSlotsShmemInit,
};
 
/* My backend's replication slot in the shared memory array */
ReplicationSlot *MyReplicationSlot = NULL;
 
/* GUC variables */
int         max_replication_slots = 10; /* the maximum number of replication
                                         * slots */
int         max_repack_replication_slots = 5;   /* the maximum number of slots
                                                 * for REPACK */
 
/*
 * Invalidate replication slots that have remained idle longer than this
 * duration; '0' disables it.
 */
int         idle_replication_slot_timeout_secs = 0;
 
/*
 * This GUC lists streaming replication standby server slot names that
 * logical WAL sender processes will wait for.
 */
char       *synchronized_standby_slots;
 
/* This is the parsed and cached configuration for synchronized_standby_slots */
static SyncStandbySlotsConfigData *synchronized_standby_slots_config;
 
/*
 * Oldest LSN that has been confirmed to be flushed to the standbys
 * corresponding to the physical slots specified in the synchronized_standby_slots GUC.
 */
static XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr;
 
static void ReplicationSlotShmemExit(int code, Datum arg);
static bool IsSlotForConflictCheck(const char *name);
static void ReplicationSlotDropPtr(ReplicationSlot *slot);
 
/* internal persistency functions */
static void RestoreSlotFromDisk(const char *name);
static void CreateSlotOnDisk(ReplicationSlot *slot);
static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel);
 
/*
 * Register shared memory space needed for replication slots.
 */
static void
ReplicationSlotsShmemRequest(void *arg)
{
    Size        size;
 
    if (max_replication_slots + max_repack_replication_slots == 0)
        return;
 
    size = offsetof(ReplicationSlotCtlData, replication_slots);
    size = add_size(size,
                    mul_size(max_replication_slots + max_repack_replication_slots,
                             sizeof(ReplicationSlot)));
    ShmemRequestStruct(.name = "ReplicationSlot Ctl",
                       .size = size,
                       .ptr = (void **) &ReplicationSlotCtl,
        );
}
 
/*
 * Initialize shared memory for replication slots.
 */
static void
ReplicationSlotsShmemInit(void *arg)
{
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[i];
 
        /* everything else is zeroed by the memset above */
        slot->active_proc = INVALID_PROC_NUMBER;
        SpinLockInit(&slot->mutex);
        LWLockInitialize(&slot->io_in_progress_lock,
                         LWTRANCHE_REPLICATION_SLOT_IO);
        ConditionVariableInit(&slot->active_cv);
    }
}
 
/*
 * Register the callback for replication slot cleanup and releasing.
 */
void
ReplicationSlotInitialize(void)
{
    before_shmem_exit(ReplicationSlotShmemExit, 0);
}
 
/*
 * Release and cleanup replication slots.
 */
static void
ReplicationSlotShmemExit(int code, Datum arg)
{
    /* Make sure active replication slots are released */
    if (MyReplicationSlot != NULL)
        ReplicationSlotRelease();
 
    /* Also cleanup all the temporary slots. */
    ReplicationSlotCleanup(false);
}
 
/*
 * Check whether the passed slot name is valid and report errors at elevel.
 *
 * See comments for ReplicationSlotValidateNameInternal().
 */
bool
ReplicationSlotValidateName(const char *name, bool allow_reserved_name,
                            int elevel)
{
    int         err_code;
    char       *err_msg = NULL;
    char       *err_hint = NULL;
 
    if (!ReplicationSlotValidateNameInternal(name, allow_reserved_name,
                                             &err_code, &err_msg, &err_hint))
    {
        /*
         * Use errmsg_internal() and errhint_internal() instead of errmsg()
         * and errhint(), since the messages from
         * ReplicationSlotValidateNameInternal() are already translated. This
         * avoids double translation.
         */
        ereport(elevel,
                errcode(err_code),
                errmsg_internal("%s", err_msg),
                (err_hint != NULL) ? errhint_internal("%s", err_hint) : 0);
 
        pfree(err_msg);
        if (err_hint != NULL)
            pfree(err_hint);
        return false;
    }
 
    return true;
}
 
/*
 * Check whether the passed slot name is valid.
 *
 * An error will be reported for a reserved replication slot name if
 * allow_reserved_name is set to false.
 *
 * Slot names may consist out of [a-z0-9_]{1,NAMEDATALEN-1} which should allow
 * the name to be used as a directory name on every supported OS.
 *
 * Returns true if the slot name is valid. Otherwise, returns false and stores
 * the error code, error message, and optional hint in err_code, err_msg, and
 * err_hint, respectively. The caller is responsible for freeing err_msg and
 * err_hint, which are palloc'd.
 */
bool
ReplicationSlotValidateNameInternal(const char *name, bool allow_reserved_name,
                                    int *err_code, char **err_msg, char **err_hint)
{
    const char *cp;
 
    if (strlen(name) == 0)
    {
        *err_code = ERRCODE_INVALID_NAME;
        *err_msg = psprintf(_("replication slot name \"%s\" is too short"), name);
        *err_hint = NULL;
        return false;
    }
 
    if (strlen(name) >= NAMEDATALEN)
    {
        *err_code = ERRCODE_NAME_TOO_LONG;
        *err_msg = psprintf(_("replication slot name \"%s\" is too long"), name);
        *err_hint = NULL;
        return false;
    }
 
    for (cp = name; *cp; cp++)
    {
        if (!((*cp >= 'a' && *cp <= 'z')
              || (*cp >= '0' && *cp <= '9')
              || (*cp == '_')))
        {
            *err_code = ERRCODE_INVALID_NAME;
            *err_msg = psprintf(_("replication slot name \"%s\" contains invalid character"), name);
            *err_hint = psprintf(_("Replication slot names may only contain lower case letters, numbers, and the underscore character."));
            return false;
        }
    }
 
    if (!allow_reserved_name && IsSlotForConflictCheck(name))
    {
        *err_code = ERRCODE_RESERVED_NAME;
        *err_msg = psprintf(_("replication slot name \"%s\" is reserved"), name);
        *err_hint = psprintf(_("The name \"%s\" is reserved for the conflict detection slot."),
                             CONFLICT_DETECTION_SLOT);
        return false;
    }
 
    return true;
}
 
/*
 * Return true if the replication slot name is "pg_conflict_detection".
 */
static bool
IsSlotForConflictCheck(const char *name)
{
    return (strcmp(name, CONFLICT_DETECTION_SLOT) == 0);
}
 
/*
 * Create a new replication slot and mark it as used by this backend.
 *
 * name: Name of the slot
 * db_specific: logical decoding is db specific; if the slot is going to
 *     be used for that pass true, otherwise false.
 * two_phase: If enabled, allows decoding of prepared transactions.
 * repack: If true, use a slot from the pool for REPACK.
 * failover: If enabled, allows the slot to be synced to standbys so
 *     that logical replication can be resumed after failover.
 * synced: True if the slot is synchronized from the primary server.
 */
void
ReplicationSlotCreate(const char *name, bool db_specific,
                      ReplicationSlotPersistency persistency,
                      bool two_phase, bool repack, bool failover, bool synced)
{
    ReplicationSlot *slot = NULL;
    int         startpoint,
                endpoint;
 
    Assert(MyReplicationSlot == NULL);
 
    /*
     * The logical launcher or pg_upgrade may create or migrate an internal
     * slot, so using a reserved name is allowed in these cases.
     */
    ReplicationSlotValidateName(name, IsBinaryUpgrade || IsLogicalLauncher(),
                                ERROR);
 
    if (failover)
    {
        /*
         * Do not allow users to create the failover enabled slots on the
         * standby as we do not support sync to the cascading standby.
         *
         * However, failover enabled slots can be created during slot
         * synchronization because we need to retain the same values as the
         * remote slot.
         */
        if (RecoveryInProgress() && !IsSyncingReplicationSlots())
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("cannot enable failover for a replication slot created on the standby"));
 
        /*
         * Do not allow users to create failover enabled temporary slots,
         * because temporary slots will not be synced to the standby.
         *
         * However, failover enabled temporary slots can be created during
         * slot synchronization. See the comments atop slotsync.c for details.
         */
        if (persistency == RS_TEMPORARY && !IsSyncingReplicationSlots())
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("cannot enable failover for a temporary replication slot"));
    }
 
    /*
     * If some other backend ran this code concurrently with us, we'd likely
     * both allocate the same slot, and that would be bad.  We'd also be at
     * risk of missing a name collision.  Also, we don't want to try to create
     * a new slot while somebody's busy cleaning up an old one, because we
     * might both be monkeying with the same directory.
     */
    LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
 
    /*
     * Check for name collision (across the whole array), and identify an
     * allocatable slot (in the array slice specific to our current use case:
     * either general, or REPACK only).  We need to hold
     * ReplicationSlotControlLock in shared mode for this, so that nobody else
     * can change the in_use flags while we're looking at them.
     */
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    startpoint = !repack ? 0 : max_replication_slots;
    endpoint = max_replication_slots + (repack ? max_repack_replication_slots : 0);
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
        if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
            ereport(ERROR,
                    (errcode(ERRCODE_DUPLICATE_OBJECT),
                     errmsg("replication slot \"%s\" already exists", name)));
 
        if (i >= startpoint && i < endpoint &&
            !s->in_use && slot == NULL)
            slot = s;
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    /* If all slots are in use, we're out of luck. */
    if (slot == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                 errmsg("all replication slots are in use"),
                 errhint("Free one or increase \"%s\".",
                         repack ? "max_repack_replication_slots" : "max_replication_slots")));
 
    /*
     * Since this slot is not in use, nobody should be looking at any part of
     * it other than the in_use field unless they're trying to allocate it.
     * And since we hold ReplicationSlotAllocationLock, nobody except us can
     * be doing that.  So it's safe to initialize the slot.
     */
    Assert(!slot->in_use);
    Assert(slot->active_proc == INVALID_PROC_NUMBER);
 
    /* first initialize persistent data */
    memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
    namestrcpy(&slot->data.name, name);
    slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
    slot->data.persistency = persistency;
    slot->data.two_phase = two_phase;
    slot->data.two_phase_at = InvalidXLogRecPtr;
    slot->data.failover = failover;
    slot->data.synced = synced;
 
    /* and then data only present in shared memory */
    slot->just_dirtied = false;
    slot->dirty = false;
    slot->effective_xmin = InvalidTransactionId;
    slot->effective_catalog_xmin = InvalidTransactionId;
    slot->candidate_catalog_xmin = InvalidTransactionId;
    slot->candidate_xmin_lsn = InvalidXLogRecPtr;
    slot->candidate_restart_valid = InvalidXLogRecPtr;
    slot->candidate_restart_lsn = InvalidXLogRecPtr;
    slot->last_saved_confirmed_flush = InvalidXLogRecPtr;
    slot->last_saved_restart_lsn = InvalidXLogRecPtr;
    slot->inactive_since = 0;
    slot->slotsync_skip_reason = SS_SKIP_NONE;
 
    /*
     * Create the slot on disk.  We haven't actually marked the slot allocated
     * yet, so no special cleanup is required if this errors out.
     */
    CreateSlotOnDisk(slot);
 
    /*
     * We need to briefly prevent any other backend from iterating over the
     * slots while we flip the in_use flag. We also need to set the active
     * flag while holding the ControlLock as otherwise a concurrent
     * ReplicationSlotAcquire() could acquire the slot as well.
     */
    LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
 
    slot->in_use = true;
 
    /* We can now mark the slot active, and that makes it our slot. */
    SpinLockAcquire(&slot->mutex);
    Assert(slot->active_proc == INVALID_PROC_NUMBER);
    slot->active_proc = MyProcNumber;
    SpinLockRelease(&slot->mutex);
    MyReplicationSlot = slot;
 
    LWLockRelease(ReplicationSlotControlLock);
 
    /*
     * Create statistics entry for the new logical slot. We don't collect any
     * stats for physical slots, so no need to create an entry for the same.
     * See ReplicationSlotDropPtr for why we need to do this before releasing
     * ReplicationSlotAllocationLock.
     */
    if (SlotIsLogical(slot))
        pgstat_create_replslot(slot);
 
    /*
     * Now that the slot has been marked as in_use and active, it's safe to
     * let somebody else try to allocate a slot.
     */
    LWLockRelease(ReplicationSlotAllocationLock);
 
    /* Let everybody know we've modified this slot */
    ConditionVariableBroadcast(&slot->active_cv);
}
 
/*
 * Search for the named replication slot.
 *
 * Return the replication slot if found, otherwise NULL.
 */
ReplicationSlot *
SearchNamedReplicationSlot(const char *name, bool need_lock)
{
    int         i;
    ReplicationSlot *slot = NULL;
 
    if (need_lock)
        LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
        if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
        {
            slot = s;
            break;
        }
    }
 
    if (need_lock)
        LWLockRelease(ReplicationSlotControlLock);
 
    return slot;
}
 
/*
 * Return the index of the replication slot in
 * ReplicationSlotCtl->replication_slots.
 *
 * This is mainly useful to have an efficient key for storing replication slot
 * stats.
 */
int
ReplicationSlotIndex(ReplicationSlot *slot)
{
    Assert(slot >= ReplicationSlotCtl->replication_slots &&
           slot < ReplicationSlotCtl->replication_slots +
           (max_replication_slots + max_repack_replication_slots));
 
    return slot - ReplicationSlotCtl->replication_slots;
}
 
/*
 * If the slot at 'index' is unused, return false. Otherwise 'name' is set to
 * the slot's name and true is returned.
 *
 * This likely is only useful for pgstat_replslot.c during shutdown, in other
 * cases there are obvious TOCTOU issues.
 */
bool
ReplicationSlotName(int index, Name name)
{
    ReplicationSlot *slot;
    bool        found;
 
    slot = &ReplicationSlotCtl->replication_slots[index];
 
    /*
     * Ensure that the slot cannot be dropped while we copy the name. Don't
     * need the spinlock as the name of an existing slot cannot change.
     */
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    found = slot->in_use;
    if (slot->in_use)
        namestrcpy(name, NameStr(slot->data.name));
    LWLockRelease(ReplicationSlotControlLock);
 
    return found;
}
 
/*
 * Find a previously created slot and mark it as used by this process.
 *
 * An error is raised if nowait is true and the slot is currently in use. If
 * nowait is false, we sleep until the slot is released by the owning process.
 *
 * An error is raised if error_if_invalid is true and the slot is found to
 * be invalid. It should always be set to true, except when we are temporarily
 * acquiring the slot and don't intend to change it.
 */
void
ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid)
{
    ReplicationSlot *s;
    ProcNumber  active_proc;
    int         active_pid;
 
    Assert(name != NULL);
 
retry:
    Assert(MyReplicationSlot == NULL);
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    /* Check if the slot exists with the given name. */
    s = SearchNamedReplicationSlot(name, false);
    if (s == NULL || !s->in_use)
    {
        LWLockRelease(ReplicationSlotControlLock);
 
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_OBJECT),
                 errmsg("replication slot \"%s\" does not exist",
                        name)));
    }
 
    /*
     * Do not allow users to acquire the reserved slot. This scenario may
     * occur if the launcher that owns the slot has terminated unexpectedly
     * due to an error, and a backend process attempts to reuse the slot.
     */
    if (!IsLogicalLauncher() && IsSlotForConflictCheck(name))
        ereport(ERROR,
                errcode(ERRCODE_UNDEFINED_OBJECT),
                errmsg("cannot acquire replication slot \"%s\"", name),
                errdetail("The slot is reserved for conflict detection and can only be acquired by logical replication launcher."));
 
    /*
     * This is the slot we want; check if it's active under some other
     * process.  In single user mode, we don't need this check.
     */
    if (IsUnderPostmaster)
    {
        /*
         * Get ready to sleep on the slot in case it is active.  (We may end
         * up not sleeping, but we don't want to do this while holding the
         * spinlock.)
         */
        if (!nowait)
            ConditionVariablePrepareToSleep(&s->active_cv);
 
        /*
         * It is important to reset the inactive_since under spinlock here to
         * avoid race conditions with slot invalidation. See comments related
         * to inactive_since in InvalidatePossiblyObsoleteSlot.
         */
        SpinLockAcquire(&s->mutex);
        if (s->active_proc == INVALID_PROC_NUMBER)
            s->active_proc = MyProcNumber;
        active_proc = s->active_proc;
        ReplicationSlotSetInactiveSince(s, 0, false);
        SpinLockRelease(&s->mutex);
    }
    else
    {
        s->active_proc = active_proc = MyProcNumber;
        ReplicationSlotSetInactiveSince(s, 0, true);
    }
    active_pid = GetPGProcByNumber(active_proc)->pid;
    LWLockRelease(ReplicationSlotControlLock);
 
    /*
     * If we found the slot but it's already active in another process, we
     * wait until the owning process signals us that it's been released, or
     * error out.
     */
    if (active_proc != MyProcNumber)
    {
        if (!nowait)
        {
            /* Wait here until we get signaled, and then restart */
            ConditionVariableSleep(&s->active_cv,
                                   WAIT_EVENT_REPLICATION_SLOT_DROP);
            ConditionVariableCancelSleep();
            goto retry;
        }
 
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_IN_USE),
                 errmsg("replication slot \"%s\" is active for PID %d",
                        NameStr(s->data.name), active_pid)));
    }
    else if (!nowait)
        ConditionVariableCancelSleep(); /* no sleep needed after all */
 
    /* We made this slot active, so it's ours now. */
    MyReplicationSlot = s;
 
    /*
     * We need to check for invalidation after making the slot ours to avoid
     * the possible race condition with the checkpointer that can otherwise
     * invalidate the slot immediately after the check.
     */
    if (error_if_invalid && s->data.invalidated != RS_INVAL_NONE)
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("can no longer access replication slot \"%s\"",
                       NameStr(s->data.name)),
                errdetail("This replication slot has been invalidated due to \"%s\".",
                          GetSlotInvalidationCauseName(s->data.invalidated)));
 
    /* Let everybody know we've modified this slot */
    ConditionVariableBroadcast(&s->active_cv);
 
    /*
     * The call to pgstat_acquire_replslot() protects against stats for a
     * different slot, from before a restart or such, being present during
     * pgstat_report_replslot().
     */
    if (SlotIsLogical(s))
        pgstat_acquire_replslot(s);
 
 
    if (am_walsender)
    {
        ereport(log_replication_commands ? LOG : DEBUG1,
                SlotIsLogical(s)
                ? errmsg("acquired logical replication slot \"%s\"",
                         NameStr(s->data.name))
                : errmsg("acquired physical replication slot \"%s\"",
                         NameStr(s->data.name)));
    }
}
 
/*
 * Release the replication slot that this backend considers to own.
 *
 * This or another backend can re-acquire the slot later.
 * Resources this slot requires will be preserved.
 */
void
ReplicationSlotRelease(void)
{
    ReplicationSlot *slot = MyReplicationSlot;
    char       *slotname = NULL;    /* keep compiler quiet */
    bool        is_logical;
    TimestampTz now = 0;
 
    Assert(slot != NULL && slot->active_proc != INVALID_PROC_NUMBER);
 
    is_logical = SlotIsLogical(slot);
 
    if (am_walsender)
        slotname = pstrdup(NameStr(slot->data.name));
 
    if (slot->data.persistency == RS_EPHEMERAL)
    {
        /*
         * If slot is ephemeral, we drop it upon release, and request logical
         * decoding be disabled.
         */
        ReplicationSlotDropAcquired(is_logical);
    }
    else
    {
        /*
         * If slot needed to temporarily restrain both data and catalog xmin
         * to create the catalog snapshot, remove that temporary constraint.
         * Snapshots can only be exported while the initial snapshot is still
         * acquired.
         */
        if (!TransactionIdIsValid(slot->data.xmin) &&
            TransactionIdIsValid(slot->effective_xmin))
        {
            SpinLockAcquire(&slot->mutex);
            slot->effective_xmin = InvalidTransactionId;
            SpinLockRelease(&slot->mutex);
            ReplicationSlotsComputeRequiredXmin(false);
        }
 
        /*
         * Set the time since the slot has become inactive. We get the current
         * time beforehand to avoid system call while holding the spinlock.
         */
        now = GetCurrentTimestamp();
 
        if (slot->data.persistency == RS_PERSISTENT)
        {
            /*
             * Mark persistent slot inactive.  We're not freeing it, just
             * disconnecting, but wake up others that may be waiting for it.
             */
            SpinLockAcquire(&slot->mutex);
            slot->active_proc = INVALID_PROC_NUMBER;
            ReplicationSlotSetInactiveSince(slot, now, false);
            SpinLockRelease(&slot->mutex);
            ConditionVariableBroadcast(&slot->active_cv);
        }
        else
            ReplicationSlotSetInactiveSince(slot, now, true);
 
        MyReplicationSlot = NULL;
    }
 
    /* might not have been set when we've been a plain slot */
    LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
    MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING;
    ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
    LWLockRelease(ProcArrayLock);
 
    if (am_walsender)
    {
        ereport(log_replication_commands ? LOG : DEBUG1,
                is_logical
                ? errmsg("released logical replication slot \"%s\"",
                         slotname)
                : errmsg("released physical replication slot \"%s\"",
                         slotname));
 
        pfree(slotname);
    }
}
 
/*
 * Cleanup temporary slots created in current session.
 *
 * Cleanup only synced temporary slots if 'synced_only' is true, else
 * cleanup all temporary slots.
 *
 * If it drops the last logical slot in the cluster, requests to disable
 * logical decoding.
 */
void
ReplicationSlotCleanup(bool synced_only)
{
    int         i;
    bool        found_valid_logicalslot;
    bool        dropped_logical = false;
 
    Assert(MyReplicationSlot == NULL);
 
restart:
    found_valid_logicalslot = false;
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
        if (!s->in_use)
            continue;
 
        SpinLockAcquire(&s->mutex);
 
        found_valid_logicalslot |=
            (SlotIsLogical(s) && s->data.invalidated == RS_INVAL_NONE);
 
        if ((s->active_proc == MyProcNumber &&
             (!synced_only || s->data.synced)))
        {
            Assert(s->data.persistency == RS_TEMPORARY);
            SpinLockRelease(&s->mutex);
            LWLockRelease(ReplicationSlotControlLock);  /* avoid deadlock */
 
            if (SlotIsLogical(s))
                dropped_logical = true;
 
            ReplicationSlotDropPtr(s);
 
            ConditionVariableBroadcast(&s->active_cv);
            goto restart;
        }
        else
            SpinLockRelease(&s->mutex);
    }
 
    LWLockRelease(ReplicationSlotControlLock);
 
    if (dropped_logical && !found_valid_logicalslot)
        RequestDisableLogicalDecoding();
}
 
/*
 * Permanently drop the replication slot identified by the passed-in name.
 *
 * If this is a logical slot, request that logical decoding be disabled.
 */
void
ReplicationSlotDrop(const char *name, bool nowait)
{
    ReplicationSlotAcquire(name, nowait, false);
 
    /*
     * Do not allow users to drop the slots which are currently being synced
     * from the primary to the standby.
     */
    if (RecoveryInProgress() && MyReplicationSlot->data.synced)
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("cannot drop replication slot \"%s\"", name),
                errdetail("This replication slot is being synchronized from the primary server."));
 
    ReplicationSlotDropAcquired(SlotIsLogical(MyReplicationSlot));
}
 
/*
 * Change the definition of the slot identified by the specified name.
 *
 * Altering the two_phase property of a slot requires caution on the
 * client-side. Enabling it at any random point during decoding has the
 * risk that transactions prepared before this change may be skipped by
 * the decoder, leading to missing prepare records on the client. So, we
 * enable it for subscription related slots only once the initial tablesync
 * is finished. See comments atop worker.c. Disabling it is safe only when
 * there are no pending prepared transaction, otherwise, the changes of
 * already prepared transactions can be replicated again along with their
 * corresponding commit leading to duplicate data or errors.
 */
void
ReplicationSlotAlter(const char *name, const bool *failover,
                     const bool *two_phase)
{
    bool        update_slot = false;
 
    Assert(MyReplicationSlot == NULL);
    Assert(failover || two_phase);
 
    ReplicationSlotAcquire(name, false, true);
 
    if (SlotIsPhysical(MyReplicationSlot))
        ereport(ERROR,
                errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                errmsg("cannot use %s with a physical replication slot",
                       "ALTER_REPLICATION_SLOT"));
 
    if (RecoveryInProgress())
    {
        /*
         * Do not allow users to alter the slots which are currently being
         * synced from the primary to the standby.
         */
        if (MyReplicationSlot->data.synced)
            ereport(ERROR,
                    errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                    errmsg("cannot alter replication slot \"%s\"", name),
                    errdetail("This replication slot is being synchronized from the primary server."));
 
        /*
         * Do not allow users to enable failover on the standby as we do not
         * support sync to the cascading standby.
         */
        if (failover && *failover)
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("cannot enable failover for a replication slot"
                           " on the standby"));
    }
 
    if (failover)
    {
        /*
         * Do not allow users to enable failover for temporary slots as we do
         * not support syncing temporary slots to the standby.
         */
        if (*failover && MyReplicationSlot->data.persistency == RS_TEMPORARY)
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("cannot enable failover for a temporary replication slot"));
 
        if (MyReplicationSlot->data.failover != *failover)
        {
            SpinLockAcquire(&MyReplicationSlot->mutex);
            MyReplicationSlot->data.failover = *failover;
            SpinLockRelease(&MyReplicationSlot->mutex);
 
            update_slot = true;
        }
    }
 
    if (two_phase && MyReplicationSlot->data.two_phase != *two_phase)
    {
        SpinLockAcquire(&MyReplicationSlot->mutex);
        MyReplicationSlot->data.two_phase = *two_phase;
        SpinLockRelease(&MyReplicationSlot->mutex);
 
        update_slot = true;
    }
 
    if (update_slot)
    {
        ReplicationSlotMarkDirty();
        ReplicationSlotSave();
    }
 
    ReplicationSlotRelease();
}
 
/*
 * Permanently drop the currently acquired replication slot.
 *
 * If caller requests it, have checkpointer attempt to disable logical
 * decoding.  Obviously, this should only be done if the slot is logical.
 */
void
ReplicationSlotDropAcquired(bool try_disable)
{
    ReplicationSlot *slot;
 
    Assert(MyReplicationSlot != NULL);
    slot = MyReplicationSlot;
 
    /* Can only disable logical decoding if slot is logical */
    Assert(!try_disable || SlotIsLogical(slot));
 
    /* slot isn't acquired anymore */
    MyReplicationSlot = NULL;
 
    ReplicationSlotDropPtr(slot);
 
    if (try_disable)
        RequestDisableLogicalDecoding();
}
 
/*
 * Permanently drop the replication slot which will be released by the point
 * this function returns.
 */
static void
ReplicationSlotDropPtr(ReplicationSlot *slot)
{
    char        path[MAXPGPATH];
    char        tmppath[MAXPGPATH];
 
    /*
     * If some other backend ran this code concurrently with us, we might try
     * to delete a slot with a certain name while someone else was trying to
     * create a slot with the same name.
     */
    LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
 
    /* Generate pathnames. */
    sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(slot->data.name));
    sprintf(tmppath, "%s/%s.tmp", PG_REPLSLOT_DIR, NameStr(slot->data.name));
 
    /*
     * Rename the slot directory on disk, so that we'll no longer recognize
     * this as a valid slot.  Note that if this fails, we've got to mark the
     * slot inactive before bailing out.  If we're dropping an ephemeral or a
     * temporary slot, we better never fail hard as the caller won't expect
     * the slot to survive and this might get called during error handling.
     */
    if (rename(path, tmppath) == 0)
    {
        /*
         * We need to fsync() the directory we just renamed and its parent to
         * make sure that our changes are on disk in a crash-safe fashion.  If
         * fsync() fails, we can't be sure whether the changes are on disk or
         * not.  For now, we handle that by panicking;
         * StartupReplicationSlots() will try to straighten it out after
         * restart.
         */
        START_CRIT_SECTION();
        fsync_fname(tmppath, true);
        fsync_fname(PG_REPLSLOT_DIR, true);
        END_CRIT_SECTION();
    }
    else
    {
        bool        fail_softly = slot->data.persistency != RS_PERSISTENT;
 
        SpinLockAcquire(&slot->mutex);
        slot->active_proc = INVALID_PROC_NUMBER;
        SpinLockRelease(&slot->mutex);
 
        /* wake up anyone waiting on this slot */
        ConditionVariableBroadcast(&slot->active_cv);
 
        ereport(fail_softly ? WARNING : ERROR,
                (errcode_for_file_access(),
                 errmsg("could not rename file \"%s\" to \"%s\": %m",
                        path, tmppath)));
    }
 
    /*
     * The slot is definitely gone.  Lock out concurrent scans of the array
     * long enough to kill it.  It's OK to clear the active PID here without
     * grabbing the mutex because nobody else can be scanning the array here,
     * and nobody can be attached to this slot and thus access it without
     * scanning the array.
     *
     * Also wake up processes waiting for it.
     */
    LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
    slot->active_proc = INVALID_PROC_NUMBER;
    slot->in_use = false;
    LWLockRelease(ReplicationSlotControlLock);
    ConditionVariableBroadcast(&slot->active_cv);
 
    /*
     * Slot is dead and doesn't prevent resource removal anymore, recompute
     * limits.
     */
    ReplicationSlotsComputeRequiredXmin(false);
    ReplicationSlotsComputeRequiredLSN();
 
    /*
     * If removing the directory fails, the worst thing that will happen is
     * that the user won't be able to create a new slot with the same name
     * until the next server restart.  We warn about it, but that's all.
     */
    if (!rmtree(tmppath, true))
        ereport(WARNING,
                (errmsg("could not remove directory \"%s\"", tmppath)));
 
    /*
     * Drop the statistics entry for the replication slot.  Do this while
     * holding ReplicationSlotAllocationLock so that we don't drop a
     * statistics entry for another slot with the same name just created in
     * another session.
     */
    if (SlotIsLogical(slot))
        pgstat_drop_replslot(slot);
 
    /*
     * We release this at the very end, so that nobody starts trying to create
     * a slot while we're still cleaning up the detritus of the old one.
     */
    LWLockRelease(ReplicationSlotAllocationLock);
}
 
/*
 * Serialize the currently acquired slot's state from memory to disk, thereby
 * guaranteeing the current state will survive a crash.
 */
void
ReplicationSlotSave(void)
{
    char        path[MAXPGPATH];
 
    Assert(MyReplicationSlot != NULL);
 
    sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(MyReplicationSlot->data.name));
    SaveSlotToPath(MyReplicationSlot, path, ERROR);
}
 
/*
 * Signal that it would be useful if the currently acquired slot would be
 * flushed out to disk.
 *
 * Note that the actual flush to disk can be delayed for a long time, if
 * required for correctness explicitly do a ReplicationSlotSave().
 */
void
ReplicationSlotMarkDirty(void)
{
    ReplicationSlot *slot = MyReplicationSlot;
 
    Assert(MyReplicationSlot != NULL);
 
    SpinLockAcquire(&slot->mutex);
    MyReplicationSlot->just_dirtied = true;
    MyReplicationSlot->dirty = true;
    SpinLockRelease(&slot->mutex);
}
 
/*
 * Convert a slot that's marked as RS_EPHEMERAL or RS_TEMPORARY to a
 * RS_PERSISTENT slot, guaranteeing it will be there after an eventual crash.
 */
void
ReplicationSlotPersist(void)
{
    ReplicationSlot *slot = MyReplicationSlot;
 
    Assert(slot != NULL);
    Assert(slot->data.persistency != RS_PERSISTENT);
 
    SpinLockAcquire(&slot->mutex);
    slot->data.persistency = RS_PERSISTENT;
    SpinLockRelease(&slot->mutex);
 
    ReplicationSlotMarkDirty();
    ReplicationSlotSave();
}
 
/*
 * Compute the oldest xmin across all slots and store it in the ProcArray.
 *
 * If already_locked is true, both the ReplicationSlotControlLock and the
 * ProcArrayLock have already been acquired exclusively. It is crucial that the
 * caller first acquires the ReplicationSlotControlLock, followed by the
 * ProcArrayLock, to prevent any undetectable deadlocks since this function
 * acquires them in that order.
 */
void
ReplicationSlotsComputeRequiredXmin(bool already_locked)
{
    int         i;
    TransactionId agg_xmin = InvalidTransactionId;
    TransactionId agg_catalog_xmin = InvalidTransactionId;
 
    Assert(ReplicationSlotCtl != NULL);
    Assert(!already_locked ||
           (LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_EXCLUSIVE) &&
            LWLockHeldByMeInMode(ProcArrayLock, LW_EXCLUSIVE)));
 
    /*
     * Hold the ReplicationSlotControlLock until after updating the slot xmin
     * values, so no backend updates the initial xmin for newly created slot
     * concurrently. A shared lock is used here to minimize lock contention,
     * especially when many slots exist and advancements occur frequently.
     * This is safe since an exclusive lock is taken during initial slot xmin
     * update in slot creation.
     *
     * One might think that we can hold the ProcArrayLock exclusively and
     * update the slot xmin values, but it could increase lock contention on
     * the ProcArrayLock, which is not great since this function can be called
     * at non-negligible frequency.
     *
     * Concurrent invocation of this function may cause the computed slot xmin
     * to regress. However, this is harmless because tuples prior to the most
     * recent xmin are no longer useful once advancement occurs (see
     * LogicalConfirmReceivedLocation where the slot's xmin value is flushed
     * before updating the effective_xmin). Thus, such regression merely
     * prevents VACUUM from prematurely removing tuples without causing the
     * early deletion of required data.
     */
    if (!already_locked)
        LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
        TransactionId effective_xmin;
        TransactionId effective_catalog_xmin;
        bool        invalidated;
 
        if (!s->in_use)
            continue;
 
        SpinLockAcquire(&s->mutex);
        effective_xmin = s->effective_xmin;
        effective_catalog_xmin = s->effective_catalog_xmin;
        invalidated = s->data.invalidated != RS_INVAL_NONE;
        SpinLockRelease(&s->mutex);
 
        /* invalidated slots need not apply */
        if (invalidated)
            continue;
 
        /* check the data xmin */
        if (TransactionIdIsValid(effective_xmin) &&
            (!TransactionIdIsValid(agg_xmin) ||
             TransactionIdPrecedes(effective_xmin, agg_xmin)))
            agg_xmin = effective_xmin;
 
        /* check the catalog xmin */
        if (TransactionIdIsValid(effective_catalog_xmin) &&
            (!TransactionIdIsValid(agg_catalog_xmin) ||
             TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
            agg_catalog_xmin = effective_catalog_xmin;
    }
 
    ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
 
    if (!already_locked)
        LWLockRelease(ReplicationSlotControlLock);
}
 
/*
 * Compute the oldest restart LSN across all slots and inform xlog module.
 *
 * Note: while max_slot_wal_keep_size is theoretically relevant for this
 * purpose, we don't try to account for that, because this module doesn't
 * know what to compare against.
 */
void
ReplicationSlotsComputeRequiredLSN(void)
{
    int         i;
    XLogRecPtr  min_required = InvalidXLogRecPtr;
 
    Assert(ReplicationSlotCtl != NULL);
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
        XLogRecPtr  restart_lsn;
        XLogRecPtr  last_saved_restart_lsn;
        bool        invalidated;
        ReplicationSlotPersistency persistency;
 
        if (!s->in_use)
            continue;
 
        SpinLockAcquire(&s->mutex);
        persistency = s->data.persistency;
        restart_lsn = s->data.restart_lsn;
        invalidated = s->data.invalidated != RS_INVAL_NONE;
        last_saved_restart_lsn = s->last_saved_restart_lsn;
        SpinLockRelease(&s->mutex);
 
        /* invalidated slots need not apply */
        if (invalidated)
            continue;
 
        /*
         * For persistent slot use last_saved_restart_lsn to compute the
         * oldest LSN for removal of WAL segments.  The segments between
         * last_saved_restart_lsn and restart_lsn might be needed by a
         * persistent slot in the case of database crash.  Non-persistent
         * slots can't survive the database crash, so we don't care about
         * last_saved_restart_lsn for them.
         */
        if (persistency == RS_PERSISTENT)
        {
            if (XLogRecPtrIsValid(last_saved_restart_lsn) &&
                restart_lsn > last_saved_restart_lsn)
            {
                restart_lsn = last_saved_restart_lsn;
            }
        }
 
        if (XLogRecPtrIsValid(restart_lsn) &&
            (!XLogRecPtrIsValid(min_required) ||
             restart_lsn < min_required))
            min_required = restart_lsn;
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    XLogSetReplicationSlotMinimumLSN(min_required);
}
 
/*
 * Compute the oldest WAL LSN required by *logical* decoding slots..
 *
 * Returns InvalidXLogRecPtr if logical decoding is disabled or no logical
 * slots exist.
 *
 * NB: this returns a value >= ReplicationSlotsComputeRequiredLSN(), since it
 * ignores physical replication slots.
 *
 * The results aren't required frequently, so we don't maintain a precomputed
 * value like we do for ComputeRequiredLSN() and ComputeRequiredXmin().
 */
XLogRecPtr
ReplicationSlotsComputeLogicalRestartLSN(void)
{
    XLogRecPtr  result = InvalidXLogRecPtr;
    int         i;
 
    if (max_replication_slots + max_repack_replication_slots <= 0)
        return InvalidXLogRecPtr;
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s;
        XLogRecPtr  restart_lsn;
        XLogRecPtr  last_saved_restart_lsn;
        bool        invalidated;
        ReplicationSlotPersistency persistency;
 
        s = &ReplicationSlotCtl->replication_slots[i];
 
        /* cannot change while ReplicationSlotCtlLock is held */
        if (!s->in_use)
            continue;
 
        /* we're only interested in logical slots */
        if (!SlotIsLogical(s))
            continue;
 
        /* read once, it's ok if it increases while we're checking */
        SpinLockAcquire(&s->mutex);
        persistency = s->data.persistency;
        restart_lsn = s->data.restart_lsn;
        invalidated = s->data.invalidated != RS_INVAL_NONE;
        last_saved_restart_lsn = s->last_saved_restart_lsn;
        SpinLockRelease(&s->mutex);
 
        /* invalidated slots need not apply */
        if (invalidated)
            continue;
 
        /*
         * For persistent slot use last_saved_restart_lsn to compute the
         * oldest LSN for removal of WAL segments.  The segments between
         * last_saved_restart_lsn and restart_lsn might be needed by a
         * persistent slot in the case of database crash.  Non-persistent
         * slots can't survive the database crash, so we don't care about
         * last_saved_restart_lsn for them.
         */
        if (persistency == RS_PERSISTENT)
        {
            if (XLogRecPtrIsValid(last_saved_restart_lsn) &&
                restart_lsn > last_saved_restart_lsn)
            {
                restart_lsn = last_saved_restart_lsn;
            }
        }
 
        if (!XLogRecPtrIsValid(restart_lsn))
            continue;
 
        if (!XLogRecPtrIsValid(result) ||
            restart_lsn < result)
            result = restart_lsn;
    }
 
    LWLockRelease(ReplicationSlotControlLock);
 
    return result;
}
 
/*
 * ReplicationSlotsCountDBSlots -- count the number of slots that refer to the
 * passed database oid.
 *
 * Returns true if there are any slots referencing the database. *nslots will
 * be set to the absolute number of slots in the database, *nactive to ones
 * currently active.
 */
bool
ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
{
    int         i;
 
    *nslots = *nactive = 0;
 
    if (max_replication_slots + max_repack_replication_slots <= 0)
        return false;
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s;
 
        s = &ReplicationSlotCtl->replication_slots[i];
 
        /* cannot change while ReplicationSlotCtlLock is held */
        if (!s->in_use)
            continue;
 
        /* only logical slots are database specific, skip */
        if (!SlotIsLogical(s))
            continue;
 
        /* not our database, skip */
        if (s->data.database != dboid)
            continue;
 
        /* NB: intentionally counting invalidated slots */
 
        /* count slots with spinlock held */
        SpinLockAcquire(&s->mutex);
        (*nslots)++;
        if (s->active_proc != INVALID_PROC_NUMBER)
            (*nactive)++;
        SpinLockRelease(&s->mutex);
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    if (*nslots > 0)
        return true;
    return false;
}
 
/*
 * ReplicationSlotsDropDBSlots -- Drop all db-specific slots relating to the
 * passed database oid. The caller should hold an exclusive lock on the
 * pg_database oid for the database to prevent creation of new slots on the db
 * or replay from existing slots.
 *
 * Another session that concurrently acquires an existing slot on the target DB
 * (most likely to drop it) may cause this function to ERROR. If that happens
 * it may have dropped some but not all slots.
 *
 * This routine isn't as efficient as it could be - but we don't drop
 * databases often, especially databases with lots of slots.
 *
 * If the last logical slot in the cluster is dropped, request to disable
 * logical decoding.
 */
void
ReplicationSlotsDropDBSlots(Oid dboid)
{
    int         i;
    bool        found_valid_logicalslot;
    bool        dropped = false;
 
    if (max_replication_slots + max_repack_replication_slots <= 0)
        return;
 
restart:
    found_valid_logicalslot = false;
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s;
        char       *slotname;
        ProcNumber  active_proc;
 
        s = &ReplicationSlotCtl->replication_slots[i];
 
        /* cannot change while ReplicationSlotCtlLock is held */
        if (!s->in_use)
            continue;
 
        /* only logical slots are database specific, skip */
        if (!SlotIsLogical(s))
            continue;
 
        /*
         * Check logical slots on other databases too so we can disable
         * logical decoding only if no slots in the cluster.
         */
        SpinLockAcquire(&s->mutex);
        found_valid_logicalslot |= (s->data.invalidated == RS_INVAL_NONE);
        SpinLockRelease(&s->mutex);
 
        /* not our database, skip */
        if (s->data.database != dboid)
            continue;
 
        /* NB: intentionally including invalidated slots to drop */
 
        /* acquire slot, so ReplicationSlotDropAcquired can be reused  */
        SpinLockAcquire(&s->mutex);
        /* can't change while ReplicationSlotControlLock is held */
        slotname = NameStr(s->data.name);
        active_proc = s->active_proc;
        if (active_proc == INVALID_PROC_NUMBER)
        {
            MyReplicationSlot = s;
            s->active_proc = MyProcNumber;
        }
        SpinLockRelease(&s->mutex);
 
        /*
         * Even though we hold an exclusive lock on the database object a
         * logical slot for that DB can still be active, e.g. if it's
         * concurrently being dropped by a backend connected to another DB.
         *
         * That's fairly unlikely in practice, so we'll just bail out.
         *
         * The slot sync worker holds a shared lock on the database before
         * operating on synced logical slots to avoid conflict with the drop
         * happening here. The persistent synced slots are thus safe but there
         * is a possibility that the slot sync worker has created a temporary
         * slot (which stays active even on release) and we are trying to drop
         * that here. In practice, the chances of hitting this scenario are
         * less as during slot synchronization, the temporary slot is
         * immediately converted to persistent and thus is safe due to the
         * shared lock taken on the database. So, we'll just bail out in such
         * a case.
         *
         * XXX: We can consider shutting down the slot sync worker before
         * trying to drop synced temporary slots here.
         */
        if (active_proc != INVALID_PROC_NUMBER)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_IN_USE),
                     errmsg("replication slot \"%s\" is active for PID %d",
                            slotname, GetPGProcByNumber(active_proc)->pid)));
 
        /*
         * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
         * holding ReplicationSlotControlLock over filesystem operations,
         * release ReplicationSlotControlLock and use
         * ReplicationSlotDropAcquired.
         *
         * As that means the set of slots could change, restart scan from the
         * beginning each time we release the lock.
         */
        LWLockRelease(ReplicationSlotControlLock);
        ReplicationSlotDropAcquired(false);
        dropped = true;
        goto restart;
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    if (dropped && !found_valid_logicalslot)
        RequestDisableLogicalDecoding();
}
 
/*
 * Returns true if there is at least one in-use valid logical replication slot.
 */
bool
CheckLogicalSlotExists(void)
{
    bool        found = false;
 
    if (max_replication_slots + max_repack_replication_slots <= 0)
        return false;
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s;
        bool        invalidated;
 
        s = &ReplicationSlotCtl->replication_slots[i];
 
        /* cannot change while ReplicationSlotCtlLock is held */
        if (!s->in_use)
            continue;
 
        if (SlotIsPhysical(s))
            continue;
 
        SpinLockAcquire(&s->mutex);
        invalidated = s->data.invalidated != RS_INVAL_NONE;
        SpinLockRelease(&s->mutex);
 
        if (invalidated)
            continue;
 
        found = true;
        break;
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    return found;
}
 
/*
 * Check whether the server's configuration supports using replication
 * slots.
 */
void
CheckSlotRequirements(bool repack)
{
    /*
     * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
     * needs the same check.
     */
 
    if (!repack && max_replication_slots == 0)
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("replication slots can only be used if \"%s\" > 0",
                       "max_replication_slots"));
 
    if (repack && max_repack_replication_slots == 0)
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("REPACK can only be used if \"%s\" > 0",
                       "max_repack_replication_slots"));
 
    if (wal_level < WAL_LEVEL_REPLICA)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("replication slots can only be used if \"wal_level\" >= \"replica\"")));
}
 
/*
 * Check whether the user has privilege to use replication slots.
 */
void
CheckSlotPermissions(void)
{
    if (!has_rolreplication(GetUserId()))
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                 errmsg("permission denied to use replication slots"),
                 errdetail("Only roles with the %s attribute may use replication slots.",
                           "REPLICATION")));
}
 
/*
 * Reserve WAL for the currently active slot.
 *
 * Compute and set restart_lsn in a manner that's appropriate for the type of
 * the slot and concurrency safe.
 */
void
ReplicationSlotReserveWal(void)
{
    ReplicationSlot *slot = MyReplicationSlot;
    XLogSegNo   segno;
    XLogRecPtr  restart_lsn;
 
    Assert(slot != NULL);
    Assert(!XLogRecPtrIsValid(slot->data.restart_lsn));
    Assert(!XLogRecPtrIsValid(slot->last_saved_restart_lsn));
 
    /*
     * The replication slot mechanism is used to prevent the removal of
     * required WAL.
     *
     * Acquire an exclusive lock to prevent the checkpoint process from
     * concurrently computing the minimum slot LSN (see
     * CheckPointReplicationSlots). This ensures that the WAL reserved for
     * replication cannot be removed during a checkpoint.
     *
     * The mechanism is reliable because if WAL reservation occurs first, the
     * checkpoint must wait for the restart_lsn update before determining the
     * minimum non-removable LSN. On the other hand, if the checkpoint happens
     * first, subsequent WAL reservations will select positions at or beyond
     * the redo pointer of that checkpoint.
     */
    LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
 
    /*
     * For logical slots log a standby snapshot and start logical decoding at
     * exactly that position. That allows the slot to start up more quickly.
     * But on a standby we cannot do WAL writes, so just use the replay
     * pointer; effectively, an attempt to create a logical slot on standby
     * will cause it to wait for an xl_running_xact record to be logged
     * independently on the primary, so that a snapshot can be built using the
     * record.
     *
     * None of this is needed (or indeed helpful) for physical slots as
     * they'll start replay at the last logged checkpoint anyway. Instead,
     * return the location of the last redo LSN, where a base backup has to
     * start replay at.
     */
    if (SlotIsPhysical(slot))
        restart_lsn = GetRedoRecPtr();
    else if (RecoveryInProgress())
        restart_lsn = GetXLogReplayRecPtr(NULL);
    else
        restart_lsn = GetXLogInsertRecPtr();
 
    SpinLockAcquire(&slot->mutex);
    slot->data.restart_lsn = restart_lsn;
    SpinLockRelease(&slot->mutex);
 
    /* prevent WAL removal as fast as possible */
    ReplicationSlotsComputeRequiredLSN();
 
    /* Checkpoint shouldn't remove the required WAL. */
    XLByteToSeg(slot->data.restart_lsn, segno, wal_segment_size);
    if (XLogGetLastRemovedSegno() >= segno)
        elog(ERROR, "WAL required by replication slot %s has been removed concurrently",
             NameStr(slot->data.name));
 
    LWLockRelease(ReplicationSlotAllocationLock);
 
    if (!RecoveryInProgress() && SlotIsLogical(slot))
    {
        XLogRecPtr  flushptr;
 
        /* make sure we have enough information to start */
        flushptr = LogStandbySnapshot();
 
        /* and make sure it's fsynced to disk */
        XLogFlush(flushptr);
    }
}
 
/*
 * Report that replication slot needs to be invalidated
 */
static void
ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
                       bool terminating,
                       int pid,
                       NameData slotname,
                       XLogRecPtr restart_lsn,
                       XLogRecPtr oldestLSN,
                       TransactionId snapshotConflictHorizon,
                       long slot_idle_seconds)
{
    StringInfoData err_detail;
    StringInfoData err_hint;
 
    initStringInfo(&err_detail);
    initStringInfo(&err_hint);
 
    switch (cause)
    {
        case RS_INVAL_WAL_REMOVED:
            {
                uint64      ex = oldestLSN - restart_lsn;
 
                appendStringInfo(&err_detail,
                                 ngettext("The slot's restart_lsn %X/%08X exceeds the limit by %" PRIu64 " byte.",
                                          "The slot's restart_lsn %X/%08X exceeds the limit by %" PRIu64 " bytes.",
                                          ex),
                                 LSN_FORMAT_ARGS(restart_lsn),
                                 ex);
                /* translator: %s is a GUC variable name */
                appendStringInfo(&err_hint, _("You might need to increase \"%s\"."),
                                 "max_slot_wal_keep_size");
                break;
            }
        case RS_INVAL_HORIZON:
            appendStringInfo(&err_detail, _("The slot conflicted with xid horizon %u."),
                             snapshotConflictHorizon);
            break;
 
        case RS_INVAL_WAL_LEVEL:
            appendStringInfoString(&err_detail, _("Logical decoding on standby requires the primary server to either set \"wal_level\" >= \"logical\" or have at least one logical slot when \"wal_level\" = \"replica\"."));
            break;
 
        case RS_INVAL_IDLE_TIMEOUT:
            {
                /* translator: %s is a GUC variable name */
                appendStringInfo(&err_detail, _("The slot's idle time of %lds exceeds the configured \"%s\" duration of %ds."),
                                 slot_idle_seconds, "idle_replication_slot_timeout",
                                 idle_replication_slot_timeout_secs);
                /* translator: %s is a GUC variable name */
                appendStringInfo(&err_hint, _("You might need to increase \"%s\"."),
                                 "idle_replication_slot_timeout");
                break;
            }
        case RS_INVAL_NONE:
            pg_unreachable();
    }
 
    ereport(LOG,
            terminating ?
            errmsg("terminating process %d to release replication slot \"%s\"",
                   pid, NameStr(slotname)) :
            errmsg("invalidating obsolete replication slot \"%s\"",
                   NameStr(slotname)),
            errdetail_internal("%s", err_detail.data),
            err_hint.len ? errhint("%s", err_hint.data) : 0);
 
    pfree(err_detail.data);
    pfree(err_hint.data);
}
 
/*
 * Can we invalidate an idle replication slot?
 *
 * Idle timeout invalidation is allowed only when:
 *
 * 1. Idle timeout is set
 * 2. Slot has reserved WAL
 * 3. Slot is inactive
 * 4. The slot is not being synced from the primary while the server is in
 *    recovery. This is because synced slots are always considered to be
 *    inactive because they don't perform logical decoding to produce changes.
 */
static inline bool
CanInvalidateIdleSlot(ReplicationSlot *s)
{
    return (idle_replication_slot_timeout_secs != 0 &&
            XLogRecPtrIsValid(s->data.restart_lsn) &&
            s->inactive_since > 0 &&
            !(RecoveryInProgress() && s->data.synced));
}
 
/*
 * DetermineSlotInvalidationCause - Determine the cause for which a slot
 * becomes invalid among the given possible causes.
 *
 * This function sequentially checks all possible invalidation causes and
 * returns the first one for which the slot is eligible for invalidation.
 */
static ReplicationSlotInvalidationCause
DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s,
                               XLogRecPtr oldestLSN, Oid dboid,
                               TransactionId snapshotConflictHorizon,
                               TimestampTz *inactive_since, TimestampTz now)
{
    Assert(possible_causes != RS_INVAL_NONE);
 
    if (possible_causes & RS_INVAL_WAL_REMOVED)
    {
        XLogRecPtr  restart_lsn = s->data.restart_lsn;
 
        if (XLogRecPtrIsValid(restart_lsn) &&
            restart_lsn < oldestLSN)
            return RS_INVAL_WAL_REMOVED;
    }
 
    if (possible_causes & RS_INVAL_HORIZON)
    {
        /* invalid DB oid signals a shared relation */
        if (SlotIsLogical(s) &&
            (dboid == InvalidOid || dboid == s->data.database))
        {
            TransactionId effective_xmin = s->effective_xmin;
            TransactionId catalog_effective_xmin = s->effective_catalog_xmin;
 
            if (TransactionIdIsValid(effective_xmin) &&
                TransactionIdPrecedesOrEquals(effective_xmin,
                                              snapshotConflictHorizon))
                return RS_INVAL_HORIZON;
            else if (TransactionIdIsValid(catalog_effective_xmin) &&
                     TransactionIdPrecedesOrEquals(catalog_effective_xmin,
                                                   snapshotConflictHorizon))
                return RS_INVAL_HORIZON;
        }
    }
 
    if (possible_causes & RS_INVAL_WAL_LEVEL)
    {
        if (SlotIsLogical(s))
            return RS_INVAL_WAL_LEVEL;
    }
 
    if (possible_causes & RS_INVAL_IDLE_TIMEOUT)
    {
        Assert(now > 0);
 
        if (CanInvalidateIdleSlot(s))
        {
            /*
             * Simulate the invalidation due to idle_timeout to test the
             * timeout behavior promptly, without waiting for it to trigger
             * naturally.
             */
#ifdef USE_INJECTION_POINTS
            if (IS_INJECTION_POINT_ATTACHED("slot-timeout-inval"))
            {
                *inactive_since = 0;    /* since the beginning of time */
                return RS_INVAL_IDLE_TIMEOUT;
            }
#endif
 
            /*
             * Check if the slot needs to be invalidated due to
             * idle_replication_slot_timeout GUC.
             */
            if (TimestampDifferenceExceedsSeconds(s->inactive_since, now,
                                                  idle_replication_slot_timeout_secs))
            {
                *inactive_since = s->inactive_since;
                return RS_INVAL_IDLE_TIMEOUT;
            }
        }
    }
 
    return RS_INVAL_NONE;
}
 
/*
 * Helper for InvalidateObsoleteReplicationSlots
 *
 * Acquires the given slot and mark it invalid, if necessary and possible.
 *
 * Returns true if the slot was invalidated.
 *
 * Set *released_lock_out if ReplicationSlotControlLock was released in the
 * interim (and in that case we're not holding the lock at return, otherwise
 * we are).
 *
 * This is inherently racy, because we release the LWLock
 * for syscalls, so caller must restart if we return true.
 */
static bool
InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
                               ReplicationSlot *s,
                               XLogRecPtr oldestLSN,
                               Oid dboid, TransactionId snapshotConflictHorizon,
                               bool *released_lock_out)
{
    int         last_signaled_pid = 0;
    bool        released_lock = false;
    bool        invalidated = false;
    TimestampTz inactive_since = 0;
 
    for (;;)
    {
        XLogRecPtr  restart_lsn;
        NameData    slotname;
        ProcNumber  active_proc;
        int         active_pid = 0;
        ReplicationSlotInvalidationCause invalidation_cause = RS_INVAL_NONE;
        TimestampTz now = 0;
        long        slot_idle_secs = 0;
 
        Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
 
        if (!s->in_use)
        {
            if (released_lock)
                LWLockRelease(ReplicationSlotControlLock);
            break;
        }
 
        if (possible_causes & RS_INVAL_IDLE_TIMEOUT)
        {
            /*
             * Assign the current time here to avoid system call overhead
             * while holding the spinlock in subsequent code.
             */
            now = GetCurrentTimestamp();
        }
 
        /*
         * Check if the slot needs to be invalidated. If it needs to be
         * invalidated, and is not currently acquired, acquire it and mark it
         * as having been invalidated.  We do this with the spinlock held to
         * avoid race conditions -- for example the restart_lsn could move
         * forward, or the slot could be dropped.
         */
        SpinLockAcquire(&s->mutex);
 
        restart_lsn = s->data.restart_lsn;
 
        /* we do nothing if the slot is already invalid */
        if (s->data.invalidated == RS_INVAL_NONE)
            invalidation_cause = DetermineSlotInvalidationCause(possible_causes,
                                                                s, oldestLSN,
                                                                dboid,
                                                                snapshotConflictHorizon,
                                                                &inactive_since,
                                                                now);
 
        /* if there's no invalidation, we're done */
        if (invalidation_cause == RS_INVAL_NONE)
        {
            SpinLockRelease(&s->mutex);
            if (released_lock)
                LWLockRelease(ReplicationSlotControlLock);
            break;
        }
 
        slotname = s->data.name;
        active_proc = s->active_proc;
 
        /*
         * If the slot can be acquired, do so and mark it invalidated
         * immediately.  Otherwise we'll signal the owning process, below, and
         * retry.
         *
         * Note: Unlike other slot attributes, slot's inactive_since can't be
         * changed until the acquired slot is released or the owning process
         * is terminated. So, the inactive slot can only be invalidated
         * immediately without being terminated.
         */
        if (active_proc == INVALID_PROC_NUMBER)
        {
            MyReplicationSlot = s;
            s->active_proc = MyProcNumber;
            s->data.invalidated = invalidation_cause;
 
            /*
             * XXX: We should consider not overwriting restart_lsn and instead
             * just rely on .invalidated.
             */
            if (invalidation_cause == RS_INVAL_WAL_REMOVED)
            {
                s->data.restart_lsn = InvalidXLogRecPtr;
                s->last_saved_restart_lsn = InvalidXLogRecPtr;
            }
 
            /* Let caller know */
            invalidated = true;
        }
        else
        {
            active_pid = GetPGProcByNumber(active_proc)->pid;
            Assert(active_pid != 0);
        }
 
        SpinLockRelease(&s->mutex);
 
        /*
         * Calculate the idle time duration of the slot if slot is marked
         * invalidated with RS_INVAL_IDLE_TIMEOUT.
         */
        if (invalidation_cause == RS_INVAL_IDLE_TIMEOUT)
        {
            int         slot_idle_usecs;
 
            TimestampDifference(inactive_since, now, &slot_idle_secs,
                                &slot_idle_usecs);
        }
 
        if (active_proc != INVALID_PROC_NUMBER)
        {
            /*
             * Prepare the sleep on the slot's condition variable before
             * releasing the lock, to close a possible race condition if the
             * slot is released before the sleep below.
             */
            ConditionVariablePrepareToSleep(&s->active_cv);
 
            LWLockRelease(ReplicationSlotControlLock);
            released_lock = true;
 
            /*
             * Signal to terminate the process that owns the slot, if we
             * haven't already signalled it.  (Avoidance of repeated
             * signalling is the only reason for there to be a loop in this
             * routine; otherwise we could rely on caller's restart loop.)
             *
             * There is the race condition that other process may own the slot
             * after its current owner process is terminated and before this
             * process owns it. To handle that, we signal only if the PID of
             * the owning process has changed from the previous time. (This
             * logic assumes that the same PID is not reused very quickly.)
             */
            if (last_signaled_pid != active_pid)
            {
                ReportSlotInvalidation(invalidation_cause, true, active_pid,
                                       slotname, restart_lsn,
                                       oldestLSN, snapshotConflictHorizon,
                                       slot_idle_secs);
 
                if (MyBackendType == B_STARTUP)
                    (void) SignalRecoveryConflict(GetPGProcByNumber(active_proc),
                                                  active_pid,
                                                  RECOVERY_CONFLICT_LOGICALSLOT);
                else
                    (void) kill(active_pid, SIGTERM);
 
                last_signaled_pid = active_pid;
            }
 
            /* Wait until the slot is released. */
            ConditionVariableSleep(&s->active_cv,
                                   WAIT_EVENT_REPLICATION_SLOT_DROP);
 
            /*
             * Re-acquire lock and start over; we expect to invalidate the
             * slot next time (unless another process acquires the slot in the
             * meantime).
             *
             * Note: It is possible for a slot to advance its restart_lsn or
             * xmin values sufficiently between when we release the mutex and
             * when we recheck, moving from a conflicting state to a non
             * conflicting state.  This is intentional and safe: if the slot
             * has caught up while we're busy here, the resources we were
             * concerned about (WAL segments or tuples) have not yet been
             * removed, and there's no reason to invalidate the slot.
             */
            LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
            continue;
        }
        else
        {
            /*
             * We hold the slot now and have already invalidated it; flush it
             * to ensure that state persists.
             *
             * Don't want to hold ReplicationSlotControlLock across file
             * system operations, so release it now but be sure to tell caller
             * to restart from scratch.
             */
            LWLockRelease(ReplicationSlotControlLock);
            released_lock = true;
 
            /* Make sure the invalidated state persists across server restart */
            ReplicationSlotMarkDirty();
            ReplicationSlotSave();
            ReplicationSlotRelease();
 
            ReportSlotInvalidation(invalidation_cause, false, active_pid,
                                   slotname, restart_lsn,
                                   oldestLSN, snapshotConflictHorizon,
                                   slot_idle_secs);
 
            /* done with this slot for now */
            break;
        }
    }
 
    Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock));
 
    *released_lock_out = released_lock;
    return invalidated;
}
 
/*
 * Invalidate slots that require resources about to be removed.
 *
 * Returns true when any slot have got invalidated.
 *
 * Whether a slot needs to be invalidated depends on the invalidation cause.
 * A slot is invalidated if it:
 * - RS_INVAL_WAL_REMOVED: requires a LSN older than the given segment
 * - RS_INVAL_HORIZON: requires a snapshot <= the given horizon in the given
 *   db; dboid may be InvalidOid for shared relations
 * - RS_INVAL_WAL_LEVEL: is a logical slot and effective_wal_level is not
 *   logical.
 * - RS_INVAL_IDLE_TIMEOUT: has been idle longer than the configured
 *   "idle_replication_slot_timeout" duration.
 *
 * Note: This function attempts to invalidate the slot for multiple possible
 * causes in a single pass, minimizing redundant iterations. The "cause"
 * parameter can be a MASK representing one or more of the defined causes.
 *
 * If it invalidates the last logical slot in the cluster, it requests to
 * disable logical decoding.
 *
 * NB - this runs as part of checkpoint, so avoid raising errors if possible.
 */
bool
InvalidateObsoleteReplicationSlots(uint32 possible_causes,
                                   XLogSegNo oldestSegno, Oid dboid,
                                   TransactionId snapshotConflictHorizon)
{
    XLogRecPtr  oldestLSN;
    bool        invalidated = false;
    bool        invalidated_logical = false;
    bool        found_valid_logicalslot;
 
    Assert(!(possible_causes & RS_INVAL_HORIZON) || TransactionIdIsValid(snapshotConflictHorizon));
    Assert(!(possible_causes & RS_INVAL_WAL_REMOVED) || oldestSegno > 0);
    Assert(possible_causes != RS_INVAL_NONE);
 
    if (max_replication_slots == 0 && max_repack_replication_slots == 0)
        return invalidated;
 
    XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
 
restart:
    found_valid_logicalslot = false;
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
        bool        released_lock = false;
 
        if (!s->in_use)
            continue;
 
        /* Prevent invalidation of logical slots during binary upgrade */
        if (SlotIsLogical(s) && IsBinaryUpgrade)
        {
            SpinLockAcquire(&s->mutex);
            found_valid_logicalslot |= (s->data.invalidated == RS_INVAL_NONE);
            SpinLockRelease(&s->mutex);
 
            continue;
        }
 
        if (InvalidatePossiblyObsoleteSlot(possible_causes, s, oldestLSN,
                                           dboid, snapshotConflictHorizon,
                                           &released_lock))
        {
            Assert(released_lock);
 
            /* Remember we have invalidated a physical or logical slot */
            invalidated = true;
 
            /*
             * Additionally, remember we have invalidated a logical slot as we
             * can request disabling logical decoding later.
             */
            if (SlotIsLogical(s))
                invalidated_logical = true;
        }
        else
        {
            /*
             * We need to check if the slot is invalidated here since
             * InvalidatePossiblyObsoleteSlot() returns false also if the slot
             * is already invalidated.
             */
            SpinLockAcquire(&s->mutex);
            found_valid_logicalslot |=
                (SlotIsLogical(s) && (s->data.invalidated == RS_INVAL_NONE));
            SpinLockRelease(&s->mutex);
        }
 
        /* if the lock was released, start from scratch */
        if (released_lock)
            goto restart;
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    /*
     * If any slots have been invalidated, recalculate the resource limits.
     */
    if (invalidated)
    {
        ReplicationSlotsComputeRequiredXmin(false);
        ReplicationSlotsComputeRequiredLSN();
    }
 
    /*
     * Request the checkpointer to disable logical decoding if no valid
     * logical slots remain. If called by the checkpointer during a
     * checkpoint, only the request is initiated; actual deactivation is
     * deferred until after the checkpoint completes.
     */
    if (invalidated_logical && !found_valid_logicalslot)
        RequestDisableLogicalDecoding();
 
    return invalidated;
}
 
/*
 * Flush all replication slots to disk.
 *
 * It is convenient to flush dirty replication slots at the time of checkpoint.
 * Additionally, in case of a shutdown checkpoint, we also identify the slots
 * for which the confirmed_flush LSN has been updated since the last time it
 * was saved and flush them.
 */
void
CheckPointReplicationSlots(bool is_shutdown)
{
    int         i;
    bool        last_saved_restart_lsn_updated = false;
 
    elog(DEBUG1, "performing replication slot checkpoint");
 
    /*
     * Prevent any slot from being created/dropped while we're active. As we
     * explicitly do *not* want to block iterating over replication_slots or
     * acquiring a slot we cannot take the control lock - but that's OK,
     * because holding ReplicationSlotAllocationLock is strictly stronger, and
     * enough to guarantee that nobody can change the in_use bits on us.
     *
     * Additionally, acquiring the Allocation lock is necessary to serialize
     * the slot flush process with concurrent slot WAL reservation. This
     * ensures that the WAL position being reserved is either flushed to disk
     * or is beyond or equal to the redo pointer of the current checkpoint
     * (See ReplicationSlotReserveWal for details).
     */
    LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
 
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
        char        path[MAXPGPATH];
 
        if (!s->in_use)
            continue;
 
        /* save the slot to disk, locking is handled in SaveSlotToPath() */
        sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(s->data.name));
 
        /*
         * Slot's data is not flushed each time the confirmed_flush LSN is
         * updated as that could lead to frequent writes.  However, we decide
         * to force a flush of all logical slot's data at the time of shutdown
         * if the confirmed_flush LSN is changed since we last flushed it to
         * disk.  This helps in avoiding an unnecessary retreat of the
         * confirmed_flush LSN after restart.
         */
        if (is_shutdown && SlotIsLogical(s))
        {
            SpinLockAcquire(&s->mutex);
 
            if (s->data.invalidated == RS_INVAL_NONE &&
                s->data.confirmed_flush > s->last_saved_confirmed_flush)
            {
                s->just_dirtied = true;
                s->dirty = true;
            }
            SpinLockRelease(&s->mutex);
        }
 
        /*
         * Track if we're going to update slot's last_saved_restart_lsn. We
         * need this to know if we need to recompute the required LSN.
         */
        if (s->last_saved_restart_lsn != s->data.restart_lsn)
            last_saved_restart_lsn_updated = true;
 
        SaveSlotToPath(s, path, LOG);
    }
    LWLockRelease(ReplicationSlotAllocationLock);
 
    /*
     * Recompute the required LSN if SaveSlotToPath() updated
     * last_saved_restart_lsn for any slot.
     */
    if (last_saved_restart_lsn_updated)
        ReplicationSlotsComputeRequiredLSN();
}
 
/*
 * Load all replication slots from disk into memory at server startup. This
 * needs to be run before we start crash recovery.
 */
void
StartupReplicationSlots(void)
{
    DIR        *replication_dir;
    struct dirent *replication_de;
 
    elog(DEBUG1, "starting up replication slots");
 
    /* restore all slots by iterating over all on-disk entries */
    replication_dir = AllocateDir(PG_REPLSLOT_DIR);
    while ((replication_de = ReadDir(replication_dir, PG_REPLSLOT_DIR)) != NULL)
    {
        char        path[MAXPGPATH + sizeof(PG_REPLSLOT_DIR)];
        PGFileType  de_type;
 
        if (strcmp(replication_de->d_name, ".") == 0 ||
            strcmp(replication_de->d_name, "..") == 0)
            continue;
 
        snprintf(path, sizeof(path), "%s/%s", PG_REPLSLOT_DIR, replication_de->d_name);
        de_type = get_dirent_type(path, replication_de, false, DEBUG1);
 
        /* we're only creating directories here, skip if it's not our's */
        if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_DIR)
            continue;
 
        /* we crashed while a slot was being setup or deleted, clean up */
        if (pg_str_endswith(replication_de->d_name, ".tmp"))
        {
            if (!rmtree(path, true))
            {
                ereport(WARNING,
                        (errmsg("could not remove directory \"%s\"",
                                path)));
                continue;
            }
            fsync_fname(PG_REPLSLOT_DIR, true);
            continue;
        }
 
        /* looks like a slot in a normal state, restore */
        RestoreSlotFromDisk(replication_de->d_name);
    }
    FreeDir(replication_dir);
 
    /* currently no slots exist, we're done. */
    if (max_replication_slots + max_repack_replication_slots <= 0)
        return;
 
    /* Now that we have recovered all the data, compute replication xmin */
    ReplicationSlotsComputeRequiredXmin(false);
    ReplicationSlotsComputeRequiredLSN();
}
 
/* ----
 * Manipulation of on-disk state of replication slots
 *
 * NB: none of the routines below should take any notice whether a slot is the
 * current one or not, that's all handled a layer above.
 * ----
 */
static void
CreateSlotOnDisk(ReplicationSlot *slot)
{
    char        tmppath[MAXPGPATH];
    char        path[MAXPGPATH];
    struct stat st;
 
    /*
     * No need to take out the io_in_progress_lock, nobody else can see this
     * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
     * takes out the lock, if we'd take the lock here, we'd deadlock.
     */
 
    sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(slot->data.name));
    sprintf(tmppath, "%s/%s.tmp", PG_REPLSLOT_DIR, NameStr(slot->data.name));
 
    /*
     * It's just barely possible that some previous effort to create or drop a
     * slot with this name left a temp directory lying around. If that seems
     * to be the case, try to remove it.  If the rmtree() fails, we'll error
     * out at the MakePGDirectory() below, so we don't bother checking
     * success.
     */
    if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
        rmtree(tmppath, true);
 
    /* Create and fsync the temporary slot directory. */
    if (MakePGDirectory(tmppath) < 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not create directory \"%s\": %m",
                        tmppath)));
    fsync_fname(tmppath, true);
 
    /* Write the actual state file. */
    slot->dirty = true;         /* signal that we really need to write */
    SaveSlotToPath(slot, tmppath, ERROR);
 
    /* Rename the directory into place. */
    if (rename(tmppath, path) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not rename file \"%s\" to \"%s\": %m",
                        tmppath, path)));
 
    /*
     * If we'd now fail - really unlikely - we wouldn't know whether this slot
     * would persist after an OS crash or not - so, force a restart. The
     * restart would try to fsync this again till it works.
     */
    START_CRIT_SECTION();
 
    fsync_fname(path, true);
    fsync_fname(PG_REPLSLOT_DIR, true);
 
    END_CRIT_SECTION();
}
 
/*
 * Shared functionality between saving and creating a replication slot.
 */
static void
SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
{
    char        tmppath[MAXPGPATH];
    char        path[MAXPGPATH];
    int         fd;
    ReplicationSlotOnDisk cp;
    bool        was_dirty;
 
    /* first check whether there's something to write out */
    SpinLockAcquire(&slot->mutex);
    was_dirty = slot->dirty;
    slot->just_dirtied = false;
    SpinLockRelease(&slot->mutex);
 
    /* and don't do anything if there's nothing to write */
    if (!was_dirty)
        return;
 
    LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE);
 
    /* silence valgrind :( */
    memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
 
    sprintf(tmppath, "%s/state.tmp", dir);
    sprintf(path, "%s/state", dir);
 
    fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
    if (fd < 0)
    {
        /*
         * If not an ERROR, then release the lock before returning.  In case
         * of an ERROR, the error recovery path automatically releases the
         * lock, but no harm in explicitly releasing even in that case.  Note
         * that LWLockRelease() could affect errno.
         */
        int         save_errno = errno;
 
        LWLockRelease(&slot->io_in_progress_lock);
        errno = save_errno;
        ereport(elevel,
                (errcode_for_file_access(),
                 errmsg("could not create file \"%s\": %m",
                        tmppath)));
        return;
    }
 
    cp.magic = SLOT_MAGIC;
    INIT_CRC32C(cp.checksum);
    cp.version = SLOT_VERSION;
    cp.length = ReplicationSlotOnDiskV2Size;
 
    SpinLockAcquire(&slot->mutex);
 
    memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
 
    SpinLockRelease(&slot->mutex);
 
    COMP_CRC32C(cp.checksum,
                (char *) (&cp) + ReplicationSlotOnDiskNotChecksummedSize,
                ReplicationSlotOnDiskChecksummedSize);
    FIN_CRC32C(cp.checksum);
 
    errno = 0;
    pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_WRITE);
    if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
    {
        int         save_errno = errno;
 
        pgstat_report_wait_end();
        CloseTransientFile(fd);
        unlink(tmppath);
        LWLockRelease(&slot->io_in_progress_lock);
 
        /* if write didn't set errno, assume problem is no disk space */
        errno = save_errno ? save_errno : ENOSPC;
        ereport(elevel,
                (errcode_for_file_access(),
                 errmsg("could not write to file \"%s\": %m",
                        tmppath)));
        return;
    }
    pgstat_report_wait_end();
 
    /* fsync the temporary file */
    pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_SYNC);
    if (pg_fsync(fd) != 0)
    {
        int         save_errno = errno;
 
        pgstat_report_wait_end();
        CloseTransientFile(fd);
        unlink(tmppath);
        LWLockRelease(&slot->io_in_progress_lock);
 
        errno = save_errno;
        ereport(elevel,
                (errcode_for_file_access(),
                 errmsg("could not fsync file \"%s\": %m",
                        tmppath)));
        return;
    }
    pgstat_report_wait_end();
 
    if (CloseTransientFile(fd) != 0)
    {
        int         save_errno = errno;
 
        unlink(tmppath);
        LWLockRelease(&slot->io_in_progress_lock);
 
        errno = save_errno;
        ereport(elevel,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m",
                        tmppath)));
        return;
    }
 
    /* rename to permanent file, fsync file and directory */
    if (rename(tmppath, path) != 0)
    {
        int         save_errno = errno;
 
        unlink(tmppath);
        LWLockRelease(&slot->io_in_progress_lock);
 
        errno = save_errno;
        ereport(elevel,
                (errcode_for_file_access(),
                 errmsg("could not rename file \"%s\" to \"%s\": %m",
                        tmppath, path)));
        return;
    }
 
    /*
     * Check CreateSlotOnDisk() for the reasoning of using a critical section.
     */
    START_CRIT_SECTION();
 
    fsync_fname(path, false);
    fsync_fname(dir, true);
    fsync_fname(PG_REPLSLOT_DIR, true);
 
    END_CRIT_SECTION();
 
    /*
     * Successfully wrote, unset dirty bit, unless somebody dirtied again
     * already and remember the confirmed_flush LSN value.
     */
    SpinLockAcquire(&slot->mutex);
    if (!slot->just_dirtied)
        slot->dirty = false;
    slot->last_saved_confirmed_flush = cp.slotdata.confirmed_flush;
    slot->last_saved_restart_lsn = cp.slotdata.restart_lsn;
    SpinLockRelease(&slot->mutex);
 
    LWLockRelease(&slot->io_in_progress_lock);
}
 
/*
 * Load a single slot from disk into memory.
 */
static void
RestoreSlotFromDisk(const char *name)
{
    ReplicationSlotOnDisk cp;
    int         i;
    char        slotdir[MAXPGPATH + sizeof(PG_REPLSLOT_DIR)];
    char        path[MAXPGPATH + sizeof(PG_REPLSLOT_DIR) + 10];
    int         fd;
    bool        restored = false;
    int         readBytes;
    pg_crc32c   checksum;
    TimestampTz now = 0;
 
    /* no need to lock here, no concurrent access allowed yet */
 
    /* delete temp file if it exists */
    sprintf(slotdir, "%s/%s", PG_REPLSLOT_DIR, name);
    sprintf(path, "%s/state.tmp", slotdir);
    if (unlink(path) < 0 && errno != ENOENT)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not remove file \"%s\": %m", path)));
 
    sprintf(path, "%s/state", slotdir);
 
    elog(DEBUG1, "restoring replication slot from \"%s\"", path);
 
    /* on some operating systems fsyncing a file requires O_RDWR */
    fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
 
    /*
     * We do not need to handle this as we are rename()ing the directory into
     * place only after we fsync()ed the state file.
     */
    if (fd < 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m", path)));
 
    /*
     * Sync state file before we're reading from it. We might have crashed
     * while it wasn't synced yet and we shouldn't continue on that basis.
     */
    pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC);
    if (pg_fsync(fd) != 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not fsync file \"%s\": %m",
                        path)));
    pgstat_report_wait_end();
 
    /* Also sync the parent directory */
    START_CRIT_SECTION();
    fsync_fname(slotdir, true);
    END_CRIT_SECTION();
 
    /* read part of statefile that's guaranteed to be version independent */
    pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
    readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
    pgstat_report_wait_end();
    if (readBytes != ReplicationSlotOnDiskConstantSize)
    {
        if (readBytes < 0)
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m", path)));
        else
            ereport(PANIC,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("could not read file \"%s\": read %d of %zu",
                            path, readBytes,
                            (Size) ReplicationSlotOnDiskConstantSize)));
    }
 
    /* verify magic */
    if (cp.magic != SLOT_MAGIC)
        ereport(PANIC,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
                        path, cp.magic, SLOT_MAGIC)));
 
    /* verify version */
    if (cp.version != SLOT_VERSION)
        ereport(PANIC,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("replication slot file \"%s\" has unsupported version %u",
                        path, cp.version)));
 
    /* boundary check on length */
    if (cp.length != ReplicationSlotOnDiskV2Size)
        ereport(PANIC,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("replication slot file \"%s\" has corrupted length %u",
                        path, cp.length)));
 
    /* Now that we know the size, read the entire file */
    pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
    readBytes = read(fd,
                     (char *) &cp + ReplicationSlotOnDiskConstantSize,
                     cp.length);
    pgstat_report_wait_end();
    if (readBytes != cp.length)
    {
        if (readBytes < 0)
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m", path)));
        else
            ereport(PANIC,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("could not read file \"%s\": read %d of %zu",
                            path, readBytes, (Size) cp.length)));
    }
 
    if (CloseTransientFile(fd) != 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m", path)));
 
    /* now verify the CRC */
    INIT_CRC32C(checksum);
    COMP_CRC32C(checksum,
                (char *) &cp + ReplicationSlotOnDiskNotChecksummedSize,
                ReplicationSlotOnDiskChecksummedSize);
    FIN_CRC32C(checksum);
 
    if (!EQ_CRC32C(checksum, cp.checksum))
        ereport(PANIC,
                (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
                        path, checksum, cp.checksum)));
 
    /*
     * If we crashed with an ephemeral slot active, don't restore but delete
     * it.
     */
    if (cp.slotdata.persistency != RS_PERSISTENT)
    {
        if (!rmtree(slotdir, true))
        {
            ereport(WARNING,
                    (errmsg("could not remove directory \"%s\"",
                            slotdir)));
        }
        fsync_fname(PG_REPLSLOT_DIR, true);
        return;
    }
 
    /*
     * Verify that requirements for the specific slot type are met. That's
     * important because if these aren't met we're not guaranteed to retain
     * all the necessary resources for the slot.
     *
     * NB: We have to do so *after* the above checks for ephemeral slots,
     * because otherwise a slot that shouldn't exist anymore could prevent
     * restarts.
     *
     * NB: Changing the requirements here also requires adapting
     * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
     */
    if (cp.slotdata.database != InvalidOid)
    {
        if (wal_level < WAL_LEVEL_REPLICA)
            ereport(FATAL,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("logical replication slot \"%s\" exists, but \"wal_level\" < \"replica\"",
                            NameStr(cp.slotdata.name)),
                     errhint("Change \"wal_level\" to be \"replica\" or higher.")));
 
        /*
         * In standby mode, the hot standby must be enabled. This check is
         * necessary to ensure logical slots are invalidated when they become
         * incompatible due to insufficient wal_level. Otherwise, if the
         * primary reduces effective_wal_level < logical while hot standby is
         * disabled, primary disable logical decoding while hot standby is
         * disabled, logical slots would remain valid even after promotion.
         */
        if (StandbyMode && !EnableHotStandby)
            ereport(FATAL,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("logical replication slot \"%s\" exists on the standby, but \"hot_standby\" = \"off\"",
                            NameStr(cp.slotdata.name)),
                     errhint("Change \"hot_standby\" to be \"on\".")));
    }
    else if (wal_level < WAL_LEVEL_REPLICA)
        ereport(FATAL,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("physical replication slot \"%s\" exists, but \"wal_level\" < \"replica\"",
                        NameStr(cp.slotdata.name)),
                 errhint("Change \"wal_level\" to be \"replica\" or higher.")));
 
    /*
     * Nothing can be active yet, don't lock anything.  Note we iterate up to
     * max_replication_slots instead of adding max_repack_replication_slots as
     * in all other places, because we must enforce the GUC value in case
     * there were more slots before the shutdown than what it is set up to
     * now.
     */
    for (i = 0; i < max_replication_slots; i++)
    {
        ReplicationSlot *slot;
 
        slot = &ReplicationSlotCtl->replication_slots[i];
 
        if (slot->in_use)
            continue;
 
        /* restore the entire set of persistent data */
        memcpy(&slot->data, &cp.slotdata,
               sizeof(ReplicationSlotPersistentData));
 
        /* initialize in memory state */
        slot->effective_xmin = cp.slotdata.xmin;
        slot->effective_catalog_xmin = cp.slotdata.catalog_xmin;
        slot->last_saved_confirmed_flush = cp.slotdata.confirmed_flush;
        slot->last_saved_restart_lsn = cp.slotdata.restart_lsn;
 
        slot->candidate_catalog_xmin = InvalidTransactionId;
        slot->candidate_xmin_lsn = InvalidXLogRecPtr;
        slot->candidate_restart_lsn = InvalidXLogRecPtr;
        slot->candidate_restart_valid = InvalidXLogRecPtr;
 
        slot->in_use = true;
        slot->active_proc = INVALID_PROC_NUMBER;
 
        /*
         * Set the time since the slot has become inactive after loading the
         * slot from the disk into memory. Whoever acquires the slot i.e.
         * makes the slot active will reset it. Use the same inactive_since
         * time for all the slots.
         */
        if (now == 0)
            now = GetCurrentTimestamp();
 
        ReplicationSlotSetInactiveSince(slot, now, false);
 
        restored = true;
        break;
    }
 
    if (!restored)
        ereport(FATAL,
                (errmsg("too many replication slots active before shutdown"),
                 errhint("Increase \"max_replication_slots\" and try again.")));
}
 
/*
 * Maps an invalidation reason for a replication slot to
 * ReplicationSlotInvalidationCause.
 */
ReplicationSlotInvalidationCause
GetSlotInvalidationCause(const char *cause_name)
{
    Assert(cause_name);
 
    /* Search lookup table for the cause having this name */
    for (int i = 0; i <= RS_INVAL_MAX_CAUSES; i++)
    {
        if (strcmp(SlotInvalidationCauses[i].cause_name, cause_name) == 0)
            return SlotInvalidationCauses[i].cause;
    }
 
    Assert(false);
    return RS_INVAL_NONE;       /* to keep compiler quiet */
}
 
/*
 * Maps a ReplicationSlotInvalidationCause to the invalidation
 * reason for a replication slot.
 */
const char *
GetSlotInvalidationCauseName(ReplicationSlotInvalidationCause cause)
{
    /* Search lookup table for the name of this cause */
    for (int i = 0; i <= RS_INVAL_MAX_CAUSES; i++)
    {
        if (SlotInvalidationCauses[i].cause == cause)
            return SlotInvalidationCauses[i].cause_name;
    }
 
    Assert(false);
    return "none";              /* to keep compiler quiet */
}
 
/*
 * A helper function to validate slots specified in GUC synchronized_standby_slots.
 *
 * The rawname will be parsed, and the result will be saved into *elemlist.
 */
static bool
validate_sync_standby_slots(char *rawname, List **elemlist)
{
    /* Verify syntax and parse string into a list of identifiers */
    if (!SplitIdentifierString(rawname, ',', elemlist))
    {
        GUC_check_errdetail("List syntax is invalid.");
        return false;
    }
 
    /* Iterate the list to validate each slot name */
    foreach_ptr(char, name, *elemlist)
    {
        int         err_code;
        char       *err_msg = NULL;
        char       *err_hint = NULL;
 
        if (!ReplicationSlotValidateNameInternal(name, false, &err_code,
                                                 &err_msg, &err_hint))
        {
            GUC_check_errcode(err_code);
            GUC_check_errdetail("%s", err_msg);
            if (err_hint != NULL)
                GUC_check_errhint("%s", err_hint);
            return false;
        }
    }
 
    return true;
}
 
/*
 * GUC check_hook for synchronized_standby_slots
 */
bool
check_synchronized_standby_slots(char **newval, void **extra, GucSource source)
{
    char       *rawname;
    char       *ptr;
    List       *elemlist;
    int         size;
    bool        ok;
    SyncStandbySlotsConfigData *config;
 
    if ((*newval)[0] == '\0')
        return true;
 
    /* Need a modifiable copy of the GUC string */
    rawname = pstrdup(*newval);
 
    /* Now verify if the specified slots exist and have correct type */
    ok = validate_sync_standby_slots(rawname, &elemlist);
 
    if (!ok || elemlist == NIL)
    {
        pfree(rawname);
        list_free(elemlist);
        return ok;
    }
 
    /* Compute the size required for the SyncStandbySlotsConfigData struct */
    size = offsetof(SyncStandbySlotsConfigData, slot_names);
    foreach_ptr(char, slot_name, elemlist)
        size += strlen(slot_name) + 1;
 
    /* GUC extra value must be guc_malloc'd, not palloc'd */
    config = (SyncStandbySlotsConfigData *) guc_malloc(LOG, size);
    if (!config)
        return false;
 
    /* Transform the data into SyncStandbySlotsConfigData */
    config->nslotnames = list_length(elemlist);
 
    ptr = config->slot_names;
    foreach_ptr(char, slot_name, elemlist)
    {
        strcpy(ptr, slot_name);
        ptr += strlen(slot_name) + 1;
    }
 
    *extra = config;
 
    pfree(rawname);
    list_free(elemlist);
    return true;
}
 
/*
 * GUC assign_hook for synchronized_standby_slots
 */
void
assign_synchronized_standby_slots(const char *newval, void *extra)
{
    /*
     * The standby slots may have changed, so we must recompute the oldest
     * LSN.
     */
    ss_oldest_flush_lsn = InvalidXLogRecPtr;
 
    synchronized_standby_slots_config = (SyncStandbySlotsConfigData *) extra;
}
 
/*
 * Check if the passed slot_name is specified in the synchronized_standby_slots GUC.
 */
bool
SlotExistsInSyncStandbySlots(const char *slot_name)
{
    const char *standby_slot_name;
 
    /* Return false if there is no value in synchronized_standby_slots */
    if (synchronized_standby_slots_config == NULL)
        return false;
 
    /*
     * XXX: We are not expecting this list to be long so a linear search
     * shouldn't hurt but if that turns out not to be true then we can cache
     * this information for each WalSender as well.
     */
    standby_slot_name = synchronized_standby_slots_config->slot_names;
    for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
    {
        if (strcmp(standby_slot_name, slot_name) == 0)
            return true;
 
        standby_slot_name += strlen(standby_slot_name) + 1;
    }
 
    return false;
}
 
/*
 * Return true if the slots specified in synchronized_standby_slots have caught up to
 * the given WAL location, false otherwise.
 *
 * The elevel parameter specifies the error level used for logging messages
 * related to slots that do not exist, are invalidated, or are inactive.
 */
bool
StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
{
    const char *name;
    int         caught_up_slot_num = 0;
    XLogRecPtr  min_restart_lsn = InvalidXLogRecPtr;
 
    /*
     * Don't need to wait for the standbys to catch up if there is no value in
     * synchronized_standby_slots.
     */
    if (synchronized_standby_slots_config == NULL)
        return true;
 
    /*
     * Don't need to wait for the standbys to catch up if we are on a standby
     * server, since we do not support syncing slots to cascading standbys.
     */
    if (RecoveryInProgress())
        return true;
 
    /*
     * Don't need to wait for the standbys to catch up if they are already
     * beyond the specified WAL location.
     */
    if (XLogRecPtrIsValid(ss_oldest_flush_lsn) &&
        ss_oldest_flush_lsn >= wait_for_lsn)
        return true;
 
    /*
     * To prevent concurrent slot dropping and creation while filtering the
     * slots, take the ReplicationSlotControlLock outside of the loop.
     */
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    name = synchronized_standby_slots_config->slot_names;
    for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
    {
        XLogRecPtr  restart_lsn;
        bool        invalidated;
        bool        inactive;
        ReplicationSlot *slot;
 
        slot = SearchNamedReplicationSlot(name, false);
 
        /*
         * If a slot name provided in synchronized_standby_slots does not
         * exist, report a message and exit the loop.
         */
        if (!slot)
        {
            ereport(elevel,
                    errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                    errmsg("replication slot \"%s\" specified in parameter \"%s\" does not exist",
                           name, "synchronized_standby_slots"),
                    errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
                              name),
                    errhint("Create the replication slot \"%s\" or amend parameter \"%s\".",
                            name, "synchronized_standby_slots"));
            break;
        }
 
        /* Same as above: if a slot is not physical, exit the loop. */
        if (SlotIsLogical(slot))
        {
            ereport(elevel,
                    errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                    errmsg("cannot specify logical replication slot \"%s\" in parameter \"%s\"",
                           name, "synchronized_standby_slots"),
                    errdetail("Logical replication is waiting for correction on replication slot \"%s\".",
                              name),
                    errhint("Remove the logical replication slot \"%s\" from parameter \"%s\".",
                            name, "synchronized_standby_slots"));
            break;
        }
 
        SpinLockAcquire(&slot->mutex);
        restart_lsn = slot->data.restart_lsn;
        invalidated = slot->data.invalidated != RS_INVAL_NONE;
        inactive = slot->active_proc == INVALID_PROC_NUMBER;
        SpinLockRelease(&slot->mutex);
 
        if (invalidated)
        {
            /* Specified physical slot has been invalidated */
            ereport(elevel,
                    errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                    errmsg("physical replication slot \"%s\" specified in parameter \"%s\" has been invalidated",
                           name, "synchronized_standby_slots"),
                    errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
                              name),
                    errhint("Drop and recreate the replication slot \"%s\", or amend parameter \"%s\".",
                            name, "synchronized_standby_slots"));
            break;
        }
 
        if (!XLogRecPtrIsValid(restart_lsn) || restart_lsn < wait_for_lsn)
        {
            /* Log a message if no active_pid for this physical slot */
            if (inactive)
                ereport(elevel,
                        errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                        errmsg("replication slot \"%s\" specified in parameter \"%s\" does not have active_pid",
                               name, "synchronized_standby_slots"),
                        errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
                                  name),
                        errhint("Start the standby associated with the replication slot \"%s\", or amend parameter \"%s\".",
                                name, "synchronized_standby_slots"));
 
            /* Continue if the current slot hasn't caught up. */
            break;
        }
 
        Assert(restart_lsn >= wait_for_lsn);
 
        if (!XLogRecPtrIsValid(min_restart_lsn) ||
            min_restart_lsn > restart_lsn)
            min_restart_lsn = restart_lsn;
 
        caught_up_slot_num++;
 
        name += strlen(name) + 1;
    }
 
    LWLockRelease(ReplicationSlotControlLock);
 
    /*
     * Return false if not all the standbys have caught up to the specified
     * WAL location.
     */
    if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
        return false;
 
    /* The ss_oldest_flush_lsn must not retreat. */
    Assert(!XLogRecPtrIsValid(ss_oldest_flush_lsn) ||
           min_restart_lsn >= ss_oldest_flush_lsn);
 
    ss_oldest_flush_lsn = min_restart_lsn;
 
    return true;
}
 
/*
 * Wait for physical standbys to confirm receiving the given lsn.
 *
 * Used by logical decoding SQL functions. It waits for physical standbys
 * corresponding to the physical slots specified in the synchronized_standby_slots GUC.
 */
void
WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
{
    /*
     * Don't need to wait for the standby to catch up if the current acquired
     * slot is not a logical failover slot, or there is no value in
     * synchronized_standby_slots.
     */
    if (!MyReplicationSlot->data.failover || !synchronized_standby_slots_config)
        return;
 
    ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv);
 
    for (;;)
    {
        CHECK_FOR_INTERRUPTS();
 
        if (ConfigReloadPending)
        {
            ConfigReloadPending = false;
            ProcessConfigFile(PGC_SIGHUP);
        }
 
        /* Exit if done waiting for every slot. */
        if (StandbySlotsHaveCaughtup(wait_for_lsn, WARNING))
            break;
 
        /*
         * Wait for the slots in the synchronized_standby_slots to catch up,
         * but use a timeout (1s) so we can also check if the
         * synchronized_standby_slots has been changed.
         */
        ConditionVariableTimedSleep(&WalSndCtl->wal_confirm_rcv_cv, 1000,
                                    WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION);
    }
 
    ConditionVariableCancelSleep();
}

◆ ReplicationSlotOnDiskConstantSize

#define ReplicationSlotOnDiskConstantSize offsetof(ReplicationSlotOnDisk, slotdata)

Definition at line 131 of file slot.c.

◆ ReplicationSlotOnDiskNotChecksummedSize

#define ReplicationSlotOnDiskNotChecksummedSize offsetof(ReplicationSlotOnDisk, version)

Definition at line 134 of file slot.c.

◆ ReplicationSlotOnDiskV2Size

#define ReplicationSlotOnDiskV2Size sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize

Definition at line 140 of file slot.c.

◆ SLOT_MAGIC

#define SLOT_MAGIC 0x1051CA1 /* format identifier */

Definition at line 143 of file slot.c.

◆ SLOT_VERSION

#define SLOT_VERSION 5 /* version for new files */

Definition at line 144 of file slot.c.

Typedef Documentation

◆ ReplicationSlotOnDisk

typedef struct ReplicationSlotOnDisk ReplicationSlotOnDisk

◆ SlotInvalidationCauseMap

typedef struct SlotInvalidationCauseMap SlotInvalidationCauseMap

Function Documentation

◆ assign_synchronized_standby_slots()

void assign_synchronized_standby_slots	(	const char *	newval,
		void *	extra
	)

Definition at line 3061 of file slot.c.

{
    /*
     * The standby slots may have changed, so we must recompute the oldest
     * LSN.
     */
    ss_oldest_flush_lsn = InvalidXLogRecPtr;
 
    synchronized_standby_slots_config = (SyncStandbySlotsConfigData *) extra;
}

References InvalidXLogRecPtr, ss_oldest_flush_lsn, and synchronized_standby_slots_config.

◆ CanInvalidateIdleSlot()

static bool CanInvalidateIdleSlot ( ReplicationSlot * s )

inlinestatic

Definition at line 1868 of file slot.c.

{
    return (idle_replication_slot_timeout_secs != 0 &&
            XLogRecPtrIsValid(s->data.restart_lsn) &&
            s->inactive_since > 0 &&
            !(RecoveryInProgress() && s->data.synced));
}

References ReplicationSlot::data, idle_replication_slot_timeout_secs, ReplicationSlot::inactive_since, RecoveryInProgress(), ReplicationSlotPersistentData::restart_lsn, ReplicationSlotPersistentData::synced, and XLogRecPtrIsValid.

Referenced by DetermineSlotInvalidationCause().

◆ check_synchronized_standby_slots()

bool check_synchronized_standby_slots	(	char **	newval,
		void **	extra,
		GucSource	source
	)

Definition at line 3005 of file slot.c.

{
    char       *rawname;
    char       *ptr;
    List       *elemlist;
    int         size;
    bool        ok;
    SyncStandbySlotsConfigData *config;
 
    if ((*newval)[0] == '\0')
        return true;
 
    /* Need a modifiable copy of the GUC string */
    rawname = pstrdup(*newval);
 
    /* Now verify if the specified slots exist and have correct type */
    ok = validate_sync_standby_slots(rawname, &elemlist);
 
    if (!ok || elemlist == NIL)
    {
        pfree(rawname);
        list_free(elemlist);
        return ok;
    }
 
    /* Compute the size required for the SyncStandbySlotsConfigData struct */
    size = offsetof(SyncStandbySlotsConfigData, slot_names);
    foreach_ptr(char, slot_name, elemlist)
        size += strlen(slot_name) + 1;
 
    /* GUC extra value must be guc_malloc'd, not palloc'd */
    config = (SyncStandbySlotsConfigData *) guc_malloc(LOG, size);
    if (!config)
        return false;
 
    /* Transform the data into SyncStandbySlotsConfigData */
    config->nslotnames = list_length(elemlist);
 
    ptr = config->slot_names;
    foreach_ptr(char, slot_name, elemlist)
    {
        strcpy(ptr, slot_name);
        ptr += strlen(slot_name) + 1;
    }
 
    *extra = config;
 
    pfree(rawname);
    list_free(elemlist);
    return true;
}

References fb(), foreach_ptr, guc_malloc(), list_free(), list_length(), LOG, newval, NIL, SyncStandbySlotsConfigData::nslotnames, pfree(), pstrdup(), SyncStandbySlotsConfigData::slot_names, and validate_sync_standby_slots().

◆ CheckLogicalSlotExists()

bool CheckLogicalSlotExists ( void )

Definition at line 1619 of file slot.c.

{
    bool        found = false;
 
    if (max_replication_slots + max_repack_replication_slots <= 0)
        return false;
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s;
        bool        invalidated;
 
        s = &ReplicationSlotCtl->replication_slots[i];
 
        /* cannot change while ReplicationSlotCtlLock is held */
        if (!s->in_use)
            continue;
 
        if (SlotIsPhysical(s))
            continue;
 
        SpinLockAcquire(&s->mutex);
        invalidated = s->data.invalidated != RS_INVAL_NONE;
        SpinLockRelease(&s->mutex);
 
        if (invalidated)
            continue;
 
        found = true;
        break;
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    return found;
}

References ReplicationSlot::data, fb(), i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, RS_INVAL_NONE, SlotIsPhysical, SpinLockAcquire(), and SpinLockRelease().

Referenced by DisableLogicalDecoding(), and UpdateLogicalDecodingStatusEndOfRecovery().

◆ CheckPointReplicationSlots()

void CheckPointReplicationSlots ( bool is_shutdown )

Definition at line 2320 of file slot.c.

{
    int         i;
    bool        last_saved_restart_lsn_updated = false;
 
    elog(DEBUG1, "performing replication slot checkpoint");
 
    /*
     * Prevent any slot from being created/dropped while we're active. As we
     * explicitly do *not* want to block iterating over replication_slots or
     * acquiring a slot we cannot take the control lock - but that's OK,
     * because holding ReplicationSlotAllocationLock is strictly stronger, and
     * enough to guarantee that nobody can change the in_use bits on us.
     *
     * Additionally, acquiring the Allocation lock is necessary to serialize
     * the slot flush process with concurrent slot WAL reservation. This
     * ensures that the WAL position being reserved is either flushed to disk
     * or is beyond or equal to the redo pointer of the current checkpoint
     * (See ReplicationSlotReserveWal for details).
     */
    LWLockAcquire(ReplicationSlotAllocationLock, LW_SHARED);
 
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
        char        path[MAXPGPATH];
 
        if (!s->in_use)
            continue;
 
        /* save the slot to disk, locking is handled in SaveSlotToPath() */
        sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(s->data.name));
 
        /*
         * Slot's data is not flushed each time the confirmed_flush LSN is
         * updated as that could lead to frequent writes.  However, we decide
         * to force a flush of all logical slot's data at the time of shutdown
         * if the confirmed_flush LSN is changed since we last flushed it to
         * disk.  This helps in avoiding an unnecessary retreat of the
         * confirmed_flush LSN after restart.
         */
        if (is_shutdown && SlotIsLogical(s))
        {
            SpinLockAcquire(&s->mutex);
 
            if (s->data.invalidated == RS_INVAL_NONE &&
                s->data.confirmed_flush > s->last_saved_confirmed_flush)
            {
                s->just_dirtied = true;
                s->dirty = true;
            }
            SpinLockRelease(&s->mutex);
        }
 
        /*
         * Track if we're going to update slot's last_saved_restart_lsn. We
         * need this to know if we need to recompute the required LSN.
         */
        if (s->last_saved_restart_lsn != s->data.restart_lsn)
            last_saved_restart_lsn_updated = true;
 
        SaveSlotToPath(s, path, LOG);
    }
    LWLockRelease(ReplicationSlotAllocationLock);
 
    /*
     * Recompute the required LSN if SaveSlotToPath() updated
     * last_saved_restart_lsn for any slot.
     */
    if (last_saved_restart_lsn_updated)
        ReplicationSlotsComputeRequiredLSN();
}

References ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, DEBUG1, ReplicationSlot::dirty, elog, fb(), i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, ReplicationSlot::just_dirtied, ReplicationSlot::last_saved_confirmed_flush, ReplicationSlot::last_saved_restart_lsn, LOG, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, MAXPGPATH, ReplicationSlot::mutex, ReplicationSlotPersistentData::name, NameStr, PG_REPLSLOT_DIR, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotsComputeRequiredLSN(), ReplicationSlotPersistentData::restart_lsn, RS_INVAL_NONE, SaveSlotToPath(), SlotIsLogical, SpinLockAcquire(), SpinLockRelease(), and sprintf.

Referenced by CheckPointGuts().

◆ CheckSlotPermissions()

void CheckSlotPermissions ( void )

Definition at line 1690 of file slot.c.

{
    if (!has_rolreplication(GetUserId()))
        ereport(ERROR,
                (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                 errmsg("permission denied to use replication slots"),
                 errdetail("Only roles with the %s attribute may use replication slots.",
                           "REPLICATION")));
}

References ereport, errcode(), errdetail(), errmsg, ERROR, fb(), GetUserId(), and has_rolreplication().

Referenced by copy_replication_slot(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_drop_replication_slot(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), and pg_sync_replication_slots().

◆ CheckSlotRequirements()

void CheckSlotRequirements ( bool repack )

Definition at line 1661 of file slot.c.

{
    /*
     * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
     * needs the same check.
     */
 
    if (!repack && max_replication_slots == 0)
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("replication slots can only be used if \"%s\" > 0",
                       "max_replication_slots"));
 
    if (repack && max_repack_replication_slots == 0)
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("REPACK can only be used if \"%s\" > 0",
                       "max_repack_replication_slots"));
 
    if (wal_level < WAL_LEVEL_REPLICA)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("replication slots can only be used if \"wal_level\" >= \"replica\"")));
}

References ereport, errcode(), errmsg, ERROR, fb(), max_repack_replication_slots, max_replication_slots, wal_level, and WAL_LEVEL_REPLICA.

Referenced by CheckLogicalDecodingRequirements(), copy_replication_slot(), pg_create_physical_replication_slot(), and pg_drop_replication_slot().

◆ CreateSlotOnDisk()

static void CreateSlotOnDisk ( ReplicationSlot * slot )

static

Definition at line 2459 of file slot.c.

{
    char        tmppath[MAXPGPATH];
    char        path[MAXPGPATH];
    struct stat st;
 
    /*
     * No need to take out the io_in_progress_lock, nobody else can see this
     * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
     * takes out the lock, if we'd take the lock here, we'd deadlock.
     */
 
    sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(slot->data.name));
    sprintf(tmppath, "%s/%s.tmp", PG_REPLSLOT_DIR, NameStr(slot->data.name));
 
    /*
     * It's just barely possible that some previous effort to create or drop a
     * slot with this name left a temp directory lying around. If that seems
     * to be the case, try to remove it.  If the rmtree() fails, we'll error
     * out at the MakePGDirectory() below, so we don't bother checking
     * success.
     */
    if (stat(tmppath, &st) == 0 && S_ISDIR(st.st_mode))
        rmtree(tmppath, true);
 
    /* Create and fsync the temporary slot directory. */
    if (MakePGDirectory(tmppath) < 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not create directory \"%s\": %m",
                        tmppath)));
    fsync_fname(tmppath, true);
 
    /* Write the actual state file. */
    slot->dirty = true;         /* signal that we really need to write */
    SaveSlotToPath(slot, tmppath, ERROR);
 
    /* Rename the directory into place. */
    if (rename(tmppath, path) != 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not rename file \"%s\" to \"%s\": %m",
                        tmppath, path)));
 
    /*
     * If we'd now fail - really unlikely - we wouldn't know whether this slot
     * would persist after an OS crash or not - so, force a restart. The
     * restart would try to fsync this again till it works.
     */
    START_CRIT_SECTION();
 
    fsync_fname(path, true);
    fsync_fname(PG_REPLSLOT_DIR, true);
 
    END_CRIT_SECTION();
}

References ReplicationSlot::data, ReplicationSlot::dirty, END_CRIT_SECTION, ereport, errcode_for_file_access(), errmsg, ERROR, fb(), fsync_fname(), MakePGDirectory(), MAXPGPATH, ReplicationSlotPersistentData::name, NameStr, PG_REPLSLOT_DIR, rmtree(), S_ISDIR, SaveSlotToPath(), sprintf, stat::st_mode, START_CRIT_SECTION, and stat.

Referenced by ReplicationSlotCreate().

◆ DetermineSlotInvalidationCause()

static ReplicationSlotInvalidationCause DetermineSlotInvalidationCause	(	uint32	possible_causes,
		ReplicationSlot *	s,
		XLogRecPtr	oldestLSN,
		Oid	dboid,
		TransactionId	snapshotConflictHorizon,
		TimestampTz *	inactive_since,
		TimestampTz	now
	)

static

Definition at line 1884 of file slot.c.

{
    Assert(possible_causes != RS_INVAL_NONE);
 
    if (possible_causes & RS_INVAL_WAL_REMOVED)
    {
        XLogRecPtr  restart_lsn = s->data.restart_lsn;
 
        if (XLogRecPtrIsValid(restart_lsn) &&
            restart_lsn < oldestLSN)
            return RS_INVAL_WAL_REMOVED;
    }
 
    if (possible_causes & RS_INVAL_HORIZON)
    {
        /* invalid DB oid signals a shared relation */
        if (SlotIsLogical(s) &&
            (dboid == InvalidOid || dboid == s->data.database))
        {
            TransactionId effective_xmin = s->effective_xmin;
            TransactionId catalog_effective_xmin = s->effective_catalog_xmin;
 
            if (TransactionIdIsValid(effective_xmin) &&
                TransactionIdPrecedesOrEquals(effective_xmin,
                                              snapshotConflictHorizon))
                return RS_INVAL_HORIZON;
            else if (TransactionIdIsValid(catalog_effective_xmin) &&
                     TransactionIdPrecedesOrEquals(catalog_effective_xmin,
                                                   snapshotConflictHorizon))
                return RS_INVAL_HORIZON;
        }
    }
 
    if (possible_causes & RS_INVAL_WAL_LEVEL)
    {
        if (SlotIsLogical(s))
            return RS_INVAL_WAL_LEVEL;
    }
 
    if (possible_causes & RS_INVAL_IDLE_TIMEOUT)
    {
        Assert(now > 0);
 
        if (CanInvalidateIdleSlot(s))
        {
            /*
             * Simulate the invalidation due to idle_timeout to test the
             * timeout behavior promptly, without waiting for it to trigger
             * naturally.
             */
#ifdef USE_INJECTION_POINTS
            if (IS_INJECTION_POINT_ATTACHED("slot-timeout-inval"))
            {
                *inactive_since = 0;    /* since the beginning of time */
                return RS_INVAL_IDLE_TIMEOUT;
            }
#endif
 
            /*
             * Check if the slot needs to be invalidated due to
             * idle_replication_slot_timeout GUC.
             */
            if (TimestampDifferenceExceedsSeconds(s->inactive_since, now,
                                                  idle_replication_slot_timeout_secs))
            {
                *inactive_since = s->inactive_since;
                return RS_INVAL_IDLE_TIMEOUT;
            }
        }
    }
 
    return RS_INVAL_NONE;
}

References Assert, CanInvalidateIdleSlot(), ReplicationSlot::data, ReplicationSlotPersistentData::database, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, fb(), idle_replication_slot_timeout_secs, ReplicationSlot::inactive_since, InvalidOid, IS_INJECTION_POINT_ATTACHED, now(), ReplicationSlotPersistentData::restart_lsn, RS_INVAL_HORIZON, RS_INVAL_IDLE_TIMEOUT, RS_INVAL_NONE, RS_INVAL_WAL_LEVEL, RS_INVAL_WAL_REMOVED, SlotIsLogical, TimestampDifferenceExceedsSeconds(), TransactionIdIsValid, TransactionIdPrecedesOrEquals(), and XLogRecPtrIsValid.

Referenced by InvalidatePossiblyObsoleteSlot().

◆ GetSlotInvalidationCause()

ReplicationSlotInvalidationCause GetSlotInvalidationCause ( const char * cause_name )

Definition at line 2932 of file slot.c.

{
    Assert(cause_name);
 
    /* Search lookup table for the cause having this name */
    for (int i = 0; i <= RS_INVAL_MAX_CAUSES; i++)
    {
        if (strcmp(SlotInvalidationCauses[i].cause_name, cause_name) == 0)
            return SlotInvalidationCauses[i].cause;
    }
 
    Assert(false);
    return RS_INVAL_NONE;       /* to keep compiler quiet */
}

References Assert, SlotInvalidationCauseMap::cause, fb(), i, RS_INVAL_MAX_CAUSES, RS_INVAL_NONE, and SlotInvalidationCauses.

Referenced by fetch_remote_slots().

◆ GetSlotInvalidationCauseName()

const char * GetSlotInvalidationCauseName ( ReplicationSlotInvalidationCause cause )

Definition at line 2952 of file slot.c.

{
    /* Search lookup table for the name of this cause */
    for (int i = 0; i <= RS_INVAL_MAX_CAUSES; i++)
    {
        if (SlotInvalidationCauses[i].cause == cause)
            return SlotInvalidationCauses[i].cause_name;
    }
 
    Assert(false);
    return "none";              /* to keep compiler quiet */
}

References Assert, SlotInvalidationCauseMap::cause_name, i, RS_INVAL_MAX_CAUSES, and SlotInvalidationCauses.

Referenced by pg_get_replication_slots(), and ReplicationSlotAcquire().

◆ InvalidateObsoleteReplicationSlots()

bool InvalidateObsoleteReplicationSlots	(	uint32	possible_causes,
		XLogSegNo	oldestSegno,
		Oid	dboid,
		TransactionId	snapshotConflictHorizon
	)

Definition at line 2216 of file slot.c.

{
    XLogRecPtr  oldestLSN;
    bool        invalidated = false;
    bool        invalidated_logical = false;
    bool        found_valid_logicalslot;
 
    Assert(!(possible_causes & RS_INVAL_HORIZON) || TransactionIdIsValid(snapshotConflictHorizon));
    Assert(!(possible_causes & RS_INVAL_WAL_REMOVED) || oldestSegno > 0);
    Assert(possible_causes != RS_INVAL_NONE);
 
    if (max_replication_slots == 0 && max_repack_replication_slots == 0)
        return invalidated;
 
    XLogSegNoOffsetToRecPtr(oldestSegno, 0, wal_segment_size, oldestLSN);
 
restart:
    found_valid_logicalslot = false;
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
        bool        released_lock = false;
 
        if (!s->in_use)
            continue;
 
        /* Prevent invalidation of logical slots during binary upgrade */
        if (SlotIsLogical(s) && IsBinaryUpgrade)
        {
            SpinLockAcquire(&s->mutex);
            found_valid_logicalslot |= (s->data.invalidated == RS_INVAL_NONE);
            SpinLockRelease(&s->mutex);
 
            continue;
        }
 
        if (InvalidatePossiblyObsoleteSlot(possible_causes, s, oldestLSN,
                                           dboid, snapshotConflictHorizon,
                                           &released_lock))
        {
            Assert(released_lock);
 
            /* Remember we have invalidated a physical or logical slot */
            invalidated = true;
 
            /*
             * Additionally, remember we have invalidated a logical slot as we
             * can request disabling logical decoding later.
             */
            if (SlotIsLogical(s))
                invalidated_logical = true;
        }
        else
        {
            /*
             * We need to check if the slot is invalidated here since
             * InvalidatePossiblyObsoleteSlot() returns false also if the slot
             * is already invalidated.
             */
            SpinLockAcquire(&s->mutex);
            found_valid_logicalslot |=
                (SlotIsLogical(s) && (s->data.invalidated == RS_INVAL_NONE));
            SpinLockRelease(&s->mutex);
        }
 
        /* if the lock was released, start from scratch */
        if (released_lock)
            goto restart;
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    /*
     * If any slots have been invalidated, recalculate the resource limits.
     */
    if (invalidated)
    {
        ReplicationSlotsComputeRequiredXmin(false);
        ReplicationSlotsComputeRequiredLSN();
    }
 
    /*
     * Request the checkpointer to disable logical decoding if no valid
     * logical slots remain. If called by the checkpointer during a
     * checkpoint, only the request is initiated; actual deactivation is
     * deferred until after the checkpoint completes.
     */
    if (invalidated_logical && !found_valid_logicalslot)
        RequestDisableLogicalDecoding();
 
    return invalidated;
}

References Assert, ReplicationSlot::data, fb(), i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, InvalidatePossiblyObsoleteSlot(), IsBinaryUpgrade, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), RequestDisableLogicalDecoding(), RS_INVAL_HORIZON, RS_INVAL_NONE, RS_INVAL_WAL_REMOVED, SlotIsLogical, SpinLockAcquire(), SpinLockRelease(), TransactionIdIsValid, wal_segment_size, and XLogSegNoOffsetToRecPtr.

Referenced by CreateCheckPoint(), CreateRestartPoint(), ResolveRecoveryConflictWithSnapshot(), and xlog_redo().

◆ InvalidatePossiblyObsoleteSlot()

static bool InvalidatePossiblyObsoleteSlot	(	uint32	possible_causes,
		ReplicationSlot *	s,
		XLogRecPtr	oldestLSN,
		Oid	dboid,
		TransactionId	snapshotConflictHorizon,
		bool *	released_lock_out
	)

static

Definition at line 1976 of file slot.c.

{
    int         last_signaled_pid = 0;
    bool        released_lock = false;
    bool        invalidated = false;
    TimestampTz inactive_since = 0;
 
    for (;;)
    {
        XLogRecPtr  restart_lsn;
        NameData    slotname;
        ProcNumber  active_proc;
        int         active_pid = 0;
        ReplicationSlotInvalidationCause invalidation_cause = RS_INVAL_NONE;
        TimestampTz now = 0;
        long        slot_idle_secs = 0;
 
        Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
 
        if (!s->in_use)
        {
            if (released_lock)
                LWLockRelease(ReplicationSlotControlLock);
            break;
        }
 
        if (possible_causes & RS_INVAL_IDLE_TIMEOUT)
        {
            /*
             * Assign the current time here to avoid system call overhead
             * while holding the spinlock in subsequent code.
             */
            now = GetCurrentTimestamp();
        }
 
        /*
         * Check if the slot needs to be invalidated. If it needs to be
         * invalidated, and is not currently acquired, acquire it and mark it
         * as having been invalidated.  We do this with the spinlock held to
         * avoid race conditions -- for example the restart_lsn could move
         * forward, or the slot could be dropped.
         */
        SpinLockAcquire(&s->mutex);
 
        restart_lsn = s->data.restart_lsn;
 
        /* we do nothing if the slot is already invalid */
        if (s->data.invalidated == RS_INVAL_NONE)
            invalidation_cause = DetermineSlotInvalidationCause(possible_causes,
                                                                s, oldestLSN,
                                                                dboid,
                                                                snapshotConflictHorizon,
                                                                &inactive_since,
                                                                now);
 
        /* if there's no invalidation, we're done */
        if (invalidation_cause == RS_INVAL_NONE)
        {
            SpinLockRelease(&s->mutex);
            if (released_lock)
                LWLockRelease(ReplicationSlotControlLock);
            break;
        }
 
        slotname = s->data.name;
        active_proc = s->active_proc;
 
        /*
         * If the slot can be acquired, do so and mark it invalidated
         * immediately.  Otherwise we'll signal the owning process, below, and
         * retry.
         *
         * Note: Unlike other slot attributes, slot's inactive_since can't be
         * changed until the acquired slot is released or the owning process
         * is terminated. So, the inactive slot can only be invalidated
         * immediately without being terminated.
         */
        if (active_proc == INVALID_PROC_NUMBER)
        {
            MyReplicationSlot = s;
            s->active_proc = MyProcNumber;
            s->data.invalidated = invalidation_cause;
 
            /*
             * XXX: We should consider not overwriting restart_lsn and instead
             * just rely on .invalidated.
             */
            if (invalidation_cause == RS_INVAL_WAL_REMOVED)
            {
                s->data.restart_lsn = InvalidXLogRecPtr;
                s->last_saved_restart_lsn = InvalidXLogRecPtr;
            }
 
            /* Let caller know */
            invalidated = true;
        }
        else
        {
            active_pid = GetPGProcByNumber(active_proc)->pid;
            Assert(active_pid != 0);
        }
 
        SpinLockRelease(&s->mutex);
 
        /*
         * Calculate the idle time duration of the slot if slot is marked
         * invalidated with RS_INVAL_IDLE_TIMEOUT.
         */
        if (invalidation_cause == RS_INVAL_IDLE_TIMEOUT)
        {
            int         slot_idle_usecs;
 
            TimestampDifference(inactive_since, now, &slot_idle_secs,
                                &slot_idle_usecs);
        }
 
        if (active_proc != INVALID_PROC_NUMBER)
        {
            /*
             * Prepare the sleep on the slot's condition variable before
             * releasing the lock, to close a possible race condition if the
             * slot is released before the sleep below.
             */
            ConditionVariablePrepareToSleep(&s->active_cv);
 
            LWLockRelease(ReplicationSlotControlLock);
            released_lock = true;
 
            /*
             * Signal to terminate the process that owns the slot, if we
             * haven't already signalled it.  (Avoidance of repeated
             * signalling is the only reason for there to be a loop in this
             * routine; otherwise we could rely on caller's restart loop.)
             *
             * There is the race condition that other process may own the slot
             * after its current owner process is terminated and before this
             * process owns it. To handle that, we signal only if the PID of
             * the owning process has changed from the previous time. (This
             * logic assumes that the same PID is not reused very quickly.)
             */
            if (last_signaled_pid != active_pid)
            {
                ReportSlotInvalidation(invalidation_cause, true, active_pid,
                                       slotname, restart_lsn,
                                       oldestLSN, snapshotConflictHorizon,
                                       slot_idle_secs);
 
                if (MyBackendType == B_STARTUP)
                    (void) SignalRecoveryConflict(GetPGProcByNumber(active_proc),
                                                  active_pid,
                                                  RECOVERY_CONFLICT_LOGICALSLOT);
                else
                    (void) kill(active_pid, SIGTERM);
 
                last_signaled_pid = active_pid;
            }
 
            /* Wait until the slot is released. */
            ConditionVariableSleep(&s->active_cv,
                                   WAIT_EVENT_REPLICATION_SLOT_DROP);
 
            /*
             * Re-acquire lock and start over; we expect to invalidate the
             * slot next time (unless another process acquires the slot in the
             * meantime).
             *
             * Note: It is possible for a slot to advance its restart_lsn or
             * xmin values sufficiently between when we release the mutex and
             * when we recheck, moving from a conflicting state to a non
             * conflicting state.  This is intentional and safe: if the slot
             * has caught up while we're busy here, the resources we were
             * concerned about (WAL segments or tuples) have not yet been
             * removed, and there's no reason to invalidate the slot.
             */
            LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
            continue;
        }
        else
        {
            /*
             * We hold the slot now and have already invalidated it; flush it
             * to ensure that state persists.
             *
             * Don't want to hold ReplicationSlotControlLock across file
             * system operations, so release it now but be sure to tell caller
             * to restart from scratch.
             */
            LWLockRelease(ReplicationSlotControlLock);
            released_lock = true;
 
            /* Make sure the invalidated state persists across server restart */
            ReplicationSlotMarkDirty();
            ReplicationSlotSave();
            ReplicationSlotRelease();
 
            ReportSlotInvalidation(invalidation_cause, false, active_pid,
                                   slotname, restart_lsn,
                                   oldestLSN, snapshotConflictHorizon,
                                   slot_idle_secs);
 
            /* done with this slot for now */
            break;
        }
    }
 
    Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock));
 
    *released_lock_out = released_lock;
    return invalidated;
}

References ReplicationSlot::active_cv, ReplicationSlot::active_proc, Assert, B_STARTUP, ConditionVariablePrepareToSleep(), ConditionVariableSleep(), ReplicationSlot::data, DetermineSlotInvalidationCause(), fb(), GetCurrentTimestamp(), GetPGProcByNumber, ReplicationSlot::in_use, INVALID_PROC_NUMBER, ReplicationSlotPersistentData::invalidated, InvalidXLogRecPtr, kill, ReplicationSlot::last_saved_restart_lsn, LW_SHARED, LWLockAcquire(), LWLockHeldByMe(), LWLockHeldByMeInMode(), LWLockRelease(), ReplicationSlot::mutex, MyBackendType, MyProcNumber, MyReplicationSlot, ReplicationSlotPersistentData::name, now(), RECOVERY_CONFLICT_LOGICALSLOT, ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotSave(), ReportSlotInvalidation(), ReplicationSlotPersistentData::restart_lsn, RS_INVAL_IDLE_TIMEOUT, RS_INVAL_NONE, RS_INVAL_WAL_REMOVED, SignalRecoveryConflict(), SpinLockAcquire(), SpinLockRelease(), and TimestampDifference().

Referenced by InvalidateObsoleteReplicationSlots().

◆ IsSlotForConflictCheck()

static bool IsSlotForConflictCheck ( const char * name )

static

Definition at line 360 of file slot.c.

{
    return (strcmp(name, CONFLICT_DETECTION_SLOT) == 0);
}

References CONFLICT_DETECTION_SLOT, fb(), and name.

Referenced by ReplicationSlotAcquire(), and ReplicationSlotValidateNameInternal().

◆ ReplicationSlotAcquire()

void ReplicationSlotAcquire	(	const char *	name,
		bool	nowait,
		bool	error_if_invalid
	)

Definition at line 629 of file slot.c.

{
    ReplicationSlot *s;
    ProcNumber  active_proc;
    int         active_pid;
 
    Assert(name != NULL);
 
retry:
    Assert(MyReplicationSlot == NULL);
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    /* Check if the slot exists with the given name. */
    s = SearchNamedReplicationSlot(name, false);
    if (s == NULL || !s->in_use)
    {
        LWLockRelease(ReplicationSlotControlLock);
 
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_OBJECT),
                 errmsg("replication slot \"%s\" does not exist",
                        name)));
    }
 
    /*
     * Do not allow users to acquire the reserved slot. This scenario may
     * occur if the launcher that owns the slot has terminated unexpectedly
     * due to an error, and a backend process attempts to reuse the slot.
     */
    if (!IsLogicalLauncher() && IsSlotForConflictCheck(name))
        ereport(ERROR,
                errcode(ERRCODE_UNDEFINED_OBJECT),
                errmsg("cannot acquire replication slot \"%s\"", name),
                errdetail("The slot is reserved for conflict detection and can only be acquired by logical replication launcher."));
 
    /*
     * This is the slot we want; check if it's active under some other
     * process.  In single user mode, we don't need this check.
     */
    if (IsUnderPostmaster)
    {
        /*
         * Get ready to sleep on the slot in case it is active.  (We may end
         * up not sleeping, but we don't want to do this while holding the
         * spinlock.)
         */
        if (!nowait)
            ConditionVariablePrepareToSleep(&s->active_cv);
 
        /*
         * It is important to reset the inactive_since under spinlock here to
         * avoid race conditions with slot invalidation. See comments related
         * to inactive_since in InvalidatePossiblyObsoleteSlot.
         */
        SpinLockAcquire(&s->mutex);
        if (s->active_proc == INVALID_PROC_NUMBER)
            s->active_proc = MyProcNumber;
        active_proc = s->active_proc;
        ReplicationSlotSetInactiveSince(s, 0, false);
        SpinLockRelease(&s->mutex);
    }
    else
    {
        s->active_proc = active_proc = MyProcNumber;
        ReplicationSlotSetInactiveSince(s, 0, true);
    }
    active_pid = GetPGProcByNumber(active_proc)->pid;
    LWLockRelease(ReplicationSlotControlLock);
 
    /*
     * If we found the slot but it's already active in another process, we
     * wait until the owning process signals us that it's been released, or
     * error out.
     */
    if (active_proc != MyProcNumber)
    {
        if (!nowait)
        {
            /* Wait here until we get signaled, and then restart */
            ConditionVariableSleep(&s->active_cv,
                                   WAIT_EVENT_REPLICATION_SLOT_DROP);
            ConditionVariableCancelSleep();
            goto retry;
        }
 
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_IN_USE),
                 errmsg("replication slot \"%s\" is active for PID %d",
                        NameStr(s->data.name), active_pid)));
    }
    else if (!nowait)
        ConditionVariableCancelSleep(); /* no sleep needed after all */
 
    /* We made this slot active, so it's ours now. */
    MyReplicationSlot = s;
 
    /*
     * We need to check for invalidation after making the slot ours to avoid
     * the possible race condition with the checkpointer that can otherwise
     * invalidate the slot immediately after the check.
     */
    if (error_if_invalid && s->data.invalidated != RS_INVAL_NONE)
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("can no longer access replication slot \"%s\"",
                       NameStr(s->data.name)),
                errdetail("This replication slot has been invalidated due to \"%s\".",
                          GetSlotInvalidationCauseName(s->data.invalidated)));
 
    /* Let everybody know we've modified this slot */
    ConditionVariableBroadcast(&s->active_cv);
 
    /*
     * The call to pgstat_acquire_replslot() protects against stats for a
     * different slot, from before a restart or such, being present during
     * pgstat_report_replslot().
     */
    if (SlotIsLogical(s))
        pgstat_acquire_replslot(s);
 
 
    if (am_walsender)
    {
        ereport(log_replication_commands ? LOG : DEBUG1,
                SlotIsLogical(s)
                ? errmsg("acquired logical replication slot \"%s\"",
                         NameStr(s->data.name))
                : errmsg("acquired physical replication slot \"%s\"",
                         NameStr(s->data.name)));
    }
}

References ReplicationSlot::active_cv, ReplicationSlot::active_proc, am_walsender, Assert, ConditionVariableBroadcast(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), ReplicationSlot::data, DEBUG1, ereport, errcode(), errdetail(), errmsg, ERROR, fb(), GetPGProcByNumber, GetSlotInvalidationCauseName(), ReplicationSlot::in_use, INVALID_PROC_NUMBER, ReplicationSlotPersistentData::invalidated, IsLogicalLauncher(), IsSlotForConflictCheck(), IsUnderPostmaster, LOG, log_replication_commands, LW_SHARED, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyProcNumber, MyReplicationSlot, name, ReplicationSlotPersistentData::name, NameStr, pgstat_acquire_replslot(), ReplicationSlotSetInactiveSince(), RS_INVAL_NONE, SearchNamedReplicationSlot(), SlotIsLogical, SpinLockAcquire(), and SpinLockRelease().

Referenced by acquire_conflict_slot_if_exists(), binary_upgrade_check_logical_slot_pending_wal(), drop_local_obsolete_slots(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), ReplicationSlotAlter(), ReplicationSlotDrop(), StartLogicalReplication(), StartReplication(), and synchronize_one_slot().

◆ ReplicationSlotAlter()

void ReplicationSlotAlter	(	const char *	name,
		const bool *	failover,
		const bool *	two_phase
	)

Definition at line 946 of file slot.c.

{
    bool        update_slot = false;
 
    Assert(MyReplicationSlot == NULL);
    Assert(failover || two_phase);
 
    ReplicationSlotAcquire(name, false, true);
 
    if (SlotIsPhysical(MyReplicationSlot))
        ereport(ERROR,
                errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                errmsg("cannot use %s with a physical replication slot",
                       "ALTER_REPLICATION_SLOT"));
 
    if (RecoveryInProgress())
    {
        /*
         * Do not allow users to alter the slots which are currently being
         * synced from the primary to the standby.
         */
        if (MyReplicationSlot->data.synced)
            ereport(ERROR,
                    errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                    errmsg("cannot alter replication slot \"%s\"", name),
                    errdetail("This replication slot is being synchronized from the primary server."));
 
        /*
         * Do not allow users to enable failover on the standby as we do not
         * support sync to the cascading standby.
         */
        if (failover && *failover)
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("cannot enable failover for a replication slot"
                           " on the standby"));
    }
 
    if (failover)
    {
        /*
         * Do not allow users to enable failover for temporary slots as we do
         * not support syncing temporary slots to the standby.
         */
        if (*failover && MyReplicationSlot->data.persistency == RS_TEMPORARY)
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("cannot enable failover for a temporary replication slot"));
 
        if (MyReplicationSlot->data.failover != *failover)
        {
            SpinLockAcquire(&MyReplicationSlot->mutex);
            MyReplicationSlot->data.failover = *failover;
            SpinLockRelease(&MyReplicationSlot->mutex);
 
            update_slot = true;
        }
    }
 
    if (two_phase && MyReplicationSlot->data.two_phase != *two_phase)
    {
        SpinLockAcquire(&MyReplicationSlot->mutex);
        MyReplicationSlot->data.two_phase = *two_phase;
        SpinLockRelease(&MyReplicationSlot->mutex);
 
        update_slot = true;
    }
 
    if (update_slot)
    {
        ReplicationSlotMarkDirty();
        ReplicationSlotSave();
    }
 
    ReplicationSlotRelease();
}

References Assert, ReplicationSlot::data, ereport, errcode(), errdetail(), errmsg, ERROR, failover, ReplicationSlotPersistentData::failover, fb(), ReplicationSlot::mutex, MyReplicationSlot, name, ReplicationSlotPersistentData::persistency, RecoveryInProgress(), ReplicationSlotAcquire(), ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotSave(), RS_TEMPORARY, SlotIsPhysical, SpinLockAcquire(), SpinLockRelease(), ReplicationSlotPersistentData::synced, two_phase, and ReplicationSlotPersistentData::two_phase.

Referenced by AlterReplicationSlot().

◆ ReplicationSlotCleanup()

void ReplicationSlotCleanup ( bool synced_only )

Definition at line 861 of file slot.c.

{
    int         i;
    bool        found_valid_logicalslot;
    bool        dropped_logical = false;
 
    Assert(MyReplicationSlot == NULL);
 
restart:
    found_valid_logicalslot = false;
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
        if (!s->in_use)
            continue;
 
        SpinLockAcquire(&s->mutex);
 
        found_valid_logicalslot |=
            (SlotIsLogical(s) && s->data.invalidated == RS_INVAL_NONE);
 
        if ((s->active_proc == MyProcNumber &&
             (!synced_only || s->data.synced)))
        {
            Assert(s->data.persistency == RS_TEMPORARY);
            SpinLockRelease(&s->mutex);
            LWLockRelease(ReplicationSlotControlLock);  /* avoid deadlock */
 
            if (SlotIsLogical(s))
                dropped_logical = true;
 
            ReplicationSlotDropPtr(s);
 
            ConditionVariableBroadcast(&s->active_cv);
            goto restart;
        }
        else
            SpinLockRelease(&s->mutex);
    }
 
    LWLockRelease(ReplicationSlotControlLock);
 
    if (dropped_logical && !found_valid_logicalslot)
        RequestDisableLogicalDecoding();
}

References ReplicationSlot::active_cv, ReplicationSlot::active_proc, Assert, ConditionVariableBroadcast(), ReplicationSlot::data, fb(), i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, MyProcNumber, MyReplicationSlot, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotDropPtr(), RequestDisableLogicalDecoding(), RS_INVAL_NONE, RS_TEMPORARY, SlotIsLogical, SpinLockAcquire(), SpinLockRelease(), and ReplicationSlotPersistentData::synced.

Referenced by PostgresMain(), ReplicationSlotShmemExit(), slotsync_failure_callback(), slotsync_worker_onexit(), SyncReplicationSlots(), and WalSndErrorCleanup().

◆ ReplicationSlotCreate()

void ReplicationSlotCreate	(	const char *	name,
		bool	db_specific,
		ReplicationSlotPersistency	persistency,
		bool	two_phase,
		bool	repack,
		bool	failover,
		bool	synced
	)

Definition at line 378 of file slot.c.

{
    ReplicationSlot *slot = NULL;
    int         startpoint,
                endpoint;
 
    Assert(MyReplicationSlot == NULL);
 
    /*
     * The logical launcher or pg_upgrade may create or migrate an internal
     * slot, so using a reserved name is allowed in these cases.
     */
    ReplicationSlotValidateName(name, IsBinaryUpgrade || IsLogicalLauncher(),
                                ERROR);
 
    if (failover)
    {
        /*
         * Do not allow users to create the failover enabled slots on the
         * standby as we do not support sync to the cascading standby.
         *
         * However, failover enabled slots can be created during slot
         * synchronization because we need to retain the same values as the
         * remote slot.
         */
        if (RecoveryInProgress() && !IsSyncingReplicationSlots())
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("cannot enable failover for a replication slot created on the standby"));
 
        /*
         * Do not allow users to create failover enabled temporary slots,
         * because temporary slots will not be synced to the standby.
         *
         * However, failover enabled temporary slots can be created during
         * slot synchronization. See the comments atop slotsync.c for details.
         */
        if (persistency == RS_TEMPORARY && !IsSyncingReplicationSlots())
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("cannot enable failover for a temporary replication slot"));
    }
 
    /*
     * If some other backend ran this code concurrently with us, we'd likely
     * both allocate the same slot, and that would be bad.  We'd also be at
     * risk of missing a name collision.  Also, we don't want to try to create
     * a new slot while somebody's busy cleaning up an old one, because we
     * might both be monkeying with the same directory.
     */
    LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
 
    /*
     * Check for name collision (across the whole array), and identify an
     * allocatable slot (in the array slice specific to our current use case:
     * either general, or REPACK only).  We need to hold
     * ReplicationSlotControlLock in shared mode for this, so that nobody else
     * can change the in_use flags while we're looking at them.
     */
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    startpoint = !repack ? 0 : max_replication_slots;
    endpoint = max_replication_slots + (repack ? max_repack_replication_slots : 0);
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
        if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
            ereport(ERROR,
                    (errcode(ERRCODE_DUPLICATE_OBJECT),
                     errmsg("replication slot \"%s\" already exists", name)));
 
        if (i >= startpoint && i < endpoint &&
            !s->in_use && slot == NULL)
            slot = s;
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    /* If all slots are in use, we're out of luck. */
    if (slot == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                 errmsg("all replication slots are in use"),
                 errhint("Free one or increase \"%s\".",
                         repack ? "max_repack_replication_slots" : "max_replication_slots")));
 
    /*
     * Since this slot is not in use, nobody should be looking at any part of
     * it other than the in_use field unless they're trying to allocate it.
     * And since we hold ReplicationSlotAllocationLock, nobody except us can
     * be doing that.  So it's safe to initialize the slot.
     */
    Assert(!slot->in_use);
    Assert(slot->active_proc == INVALID_PROC_NUMBER);
 
    /* first initialize persistent data */
    memset(&slot->data, 0, sizeof(ReplicationSlotPersistentData));
    namestrcpy(&slot->data.name, name);
    slot->data.database = db_specific ? MyDatabaseId : InvalidOid;
    slot->data.persistency = persistency;
    slot->data.two_phase = two_phase;
    slot->data.two_phase_at = InvalidXLogRecPtr;
    slot->data.failover = failover;
    slot->data.synced = synced;
 
    /* and then data only present in shared memory */
    slot->just_dirtied = false;
    slot->dirty = false;
    slot->effective_xmin = InvalidTransactionId;
    slot->effective_catalog_xmin = InvalidTransactionId;
    slot->candidate_catalog_xmin = InvalidTransactionId;
    slot->candidate_xmin_lsn = InvalidXLogRecPtr;
    slot->candidate_restart_valid = InvalidXLogRecPtr;
    slot->candidate_restart_lsn = InvalidXLogRecPtr;
    slot->last_saved_confirmed_flush = InvalidXLogRecPtr;
    slot->last_saved_restart_lsn = InvalidXLogRecPtr;
    slot->inactive_since = 0;
    slot->slotsync_skip_reason = SS_SKIP_NONE;
 
    /*
     * Create the slot on disk.  We haven't actually marked the slot allocated
     * yet, so no special cleanup is required if this errors out.
     */
    CreateSlotOnDisk(slot);
 
    /*
     * We need to briefly prevent any other backend from iterating over the
     * slots while we flip the in_use flag. We also need to set the active
     * flag while holding the ControlLock as otherwise a concurrent
     * ReplicationSlotAcquire() could acquire the slot as well.
     */
    LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
 
    slot->in_use = true;
 
    /* We can now mark the slot active, and that makes it our slot. */
    SpinLockAcquire(&slot->mutex);
    Assert(slot->active_proc == INVALID_PROC_NUMBER);
    slot->active_proc = MyProcNumber;
    SpinLockRelease(&slot->mutex);
    MyReplicationSlot = slot;
 
    LWLockRelease(ReplicationSlotControlLock);
 
    /*
     * Create statistics entry for the new logical slot. We don't collect any
     * stats for physical slots, so no need to create an entry for the same.
     * See ReplicationSlotDropPtr for why we need to do this before releasing
     * ReplicationSlotAllocationLock.
     */
    if (SlotIsLogical(slot))
        pgstat_create_replslot(slot);
 
    /*
     * Now that the slot has been marked as in_use and active, it's safe to
     * let somebody else try to allocate a slot.
     */
    LWLockRelease(ReplicationSlotAllocationLock);
 
    /* Let everybody know we've modified this slot */
    ConditionVariableBroadcast(&slot->active_cv);
}

References ReplicationSlot::active_cv, ReplicationSlot::active_proc, Assert, ReplicationSlot::candidate_catalog_xmin, ReplicationSlot::candidate_restart_lsn, ReplicationSlot::candidate_restart_valid, ReplicationSlot::candidate_xmin_lsn, ConditionVariableBroadcast(), CreateSlotOnDisk(), ReplicationSlot::data, ReplicationSlotPersistentData::database, ReplicationSlot::dirty, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, ereport, errcode(), ERRCODE_DUPLICATE_OBJECT, errhint(), errmsg, ERROR, failover, ReplicationSlotPersistentData::failover, fb(), i, ReplicationSlot::in_use, ReplicationSlot::inactive_since, INVALID_PROC_NUMBER, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, IsBinaryUpgrade, IsLogicalLauncher(), IsSyncingReplicationSlots(), ReplicationSlot::just_dirtied, ReplicationSlot::last_saved_confirmed_flush, ReplicationSlot::last_saved_restart_lsn, LW_EXCLUSIVE, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, MyDatabaseId, MyProcNumber, MyReplicationSlot, name, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), ReplicationSlotPersistentData::persistency, pgstat_create_replslot(), RecoveryInProgress(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotValidateName(), RS_TEMPORARY, SlotIsLogical, ReplicationSlot::slotsync_skip_reason, SpinLockAcquire(), SpinLockRelease(), SS_SKIP_NONE, ReplicationSlotPersistentData::synced, two_phase, ReplicationSlotPersistentData::two_phase, and ReplicationSlotPersistentData::two_phase_at.

Referenced by create_logical_replication_slot(), create_physical_replication_slot(), CreateConflictDetectionSlot(), CreateReplicationSlot(), repack_setup_logical_decoding(), and synchronize_one_slot().

◆ ReplicationSlotDrop()

void ReplicationSlotDrop	(	const char *	name,
		bool	nowait
	)

Definition at line 915 of file slot.c.

{
    ReplicationSlotAcquire(name, nowait, false);
 
    /*
     * Do not allow users to drop the slots which are currently being synced
     * from the primary to the standby.
     */
    if (RecoveryInProgress() && MyReplicationSlot->data.synced)
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("cannot drop replication slot \"%s\"", name),
                errdetail("This replication slot is being synchronized from the primary server."));
 
    ReplicationSlotDropAcquired(SlotIsLogical(MyReplicationSlot));
}

References ReplicationSlot::data, ereport, errcode(), errdetail(), errmsg, ERROR, fb(), MyReplicationSlot, name, RecoveryInProgress(), ReplicationSlotAcquire(), ReplicationSlotDropAcquired(), SlotIsLogical, and ReplicationSlotPersistentData::synced.

Referenced by DropReplicationSlot(), and pg_drop_replication_slot().

◆ ReplicationSlotDropAcquired()

void ReplicationSlotDropAcquired ( bool try_disable )

Definition at line 1031 of file slot.c.

{
    ReplicationSlot *slot;
 
    Assert(MyReplicationSlot != NULL);
    slot = MyReplicationSlot;
 
    /* Can only disable logical decoding if slot is logical */
    Assert(!try_disable || SlotIsLogical(slot));
 
    /* slot isn't acquired anymore */
    MyReplicationSlot = NULL;
 
    ReplicationSlotDropPtr(slot);
 
    if (try_disable)
        RequestDisableLogicalDecoding();
}

References Assert, fb(), MyReplicationSlot, ReplicationSlotDropPtr(), RequestDisableLogicalDecoding(), and SlotIsLogical.

Referenced by ApplyLauncherMain(), drop_local_obsolete_slots(), repack_cleanup_logical_decoding(), ReplicationSlotDrop(), ReplicationSlotRelease(), and ReplicationSlotsDropDBSlots().

◆ ReplicationSlotDropPtr()

static void ReplicationSlotDropPtr ( ReplicationSlot * slot )

static

Definition at line 1055 of file slot.c.

{
    char        path[MAXPGPATH];
    char        tmppath[MAXPGPATH];
 
    /*
     * If some other backend ran this code concurrently with us, we might try
     * to delete a slot with a certain name while someone else was trying to
     * create a slot with the same name.
     */
    LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
 
    /* Generate pathnames. */
    sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(slot->data.name));
    sprintf(tmppath, "%s/%s.tmp", PG_REPLSLOT_DIR, NameStr(slot->data.name));
 
    /*
     * Rename the slot directory on disk, so that we'll no longer recognize
     * this as a valid slot.  Note that if this fails, we've got to mark the
     * slot inactive before bailing out.  If we're dropping an ephemeral or a
     * temporary slot, we better never fail hard as the caller won't expect
     * the slot to survive and this might get called during error handling.
     */
    if (rename(path, tmppath) == 0)
    {
        /*
         * We need to fsync() the directory we just renamed and its parent to
         * make sure that our changes are on disk in a crash-safe fashion.  If
         * fsync() fails, we can't be sure whether the changes are on disk or
         * not.  For now, we handle that by panicking;
         * StartupReplicationSlots() will try to straighten it out after
         * restart.
         */
        START_CRIT_SECTION();
        fsync_fname(tmppath, true);
        fsync_fname(PG_REPLSLOT_DIR, true);
        END_CRIT_SECTION();
    }
    else
    {
        bool        fail_softly = slot->data.persistency != RS_PERSISTENT;
 
        SpinLockAcquire(&slot->mutex);
        slot->active_proc = INVALID_PROC_NUMBER;
        SpinLockRelease(&slot->mutex);
 
        /* wake up anyone waiting on this slot */
        ConditionVariableBroadcast(&slot->active_cv);
 
        ereport(fail_softly ? WARNING : ERROR,
                (errcode_for_file_access(),
                 errmsg("could not rename file \"%s\" to \"%s\": %m",
                        path, tmppath)));
    }
 
    /*
     * The slot is definitely gone.  Lock out concurrent scans of the array
     * long enough to kill it.  It's OK to clear the active PID here without
     * grabbing the mutex because nobody else can be scanning the array here,
     * and nobody can be attached to this slot and thus access it without
     * scanning the array.
     *
     * Also wake up processes waiting for it.
     */
    LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
    slot->active_proc = INVALID_PROC_NUMBER;
    slot->in_use = false;
    LWLockRelease(ReplicationSlotControlLock);
    ConditionVariableBroadcast(&slot->active_cv);
 
    /*
     * Slot is dead and doesn't prevent resource removal anymore, recompute
     * limits.
     */
    ReplicationSlotsComputeRequiredXmin(false);
    ReplicationSlotsComputeRequiredLSN();
 
    /*
     * If removing the directory fails, the worst thing that will happen is
     * that the user won't be able to create a new slot with the same name
     * until the next server restart.  We warn about it, but that's all.
     */
    if (!rmtree(tmppath, true))
        ereport(WARNING,
                (errmsg("could not remove directory \"%s\"", tmppath)));
 
    /*
     * Drop the statistics entry for the replication slot.  Do this while
     * holding ReplicationSlotAllocationLock so that we don't drop a
     * statistics entry for another slot with the same name just created in
     * another session.
     */
    if (SlotIsLogical(slot))
        pgstat_drop_replslot(slot);
 
    /*
     * We release this at the very end, so that nobody starts trying to create
     * a slot while we're still cleaning up the detritus of the old one.
     */
    LWLockRelease(ReplicationSlotAllocationLock);
}

References ReplicationSlot::active_cv, ReplicationSlot::active_proc, ConditionVariableBroadcast(), ReplicationSlot::data, END_CRIT_SECTION, ereport, errcode_for_file_access(), errmsg, ERROR, fb(), fsync_fname(), ReplicationSlot::in_use, INVALID_PROC_NUMBER, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAXPGPATH, ReplicationSlot::mutex, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotPersistentData::persistency, PG_REPLSLOT_DIR, pgstat_drop_replslot(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), rmtree(), RS_PERSISTENT, SlotIsLogical, SpinLockAcquire(), SpinLockRelease(), sprintf, START_CRIT_SECTION, and WARNING.

Referenced by ReplicationSlotCleanup(), and ReplicationSlotDropAcquired().

◆ ReplicationSlotIndex()

int ReplicationSlotIndex ( ReplicationSlot * slot )

Definition at line 581 of file slot.c.

{
    Assert(slot >= ReplicationSlotCtl->replication_slots &&
           slot < ReplicationSlotCtl->replication_slots +
           (max_replication_slots + max_repack_replication_slots));
 
    return slot - ReplicationSlotCtl->replication_slots;
}

References Assert, fb(), max_repack_replication_slots, max_replication_slots, ReplicationSlotCtlData::replication_slots, and ReplicationSlotCtl.

Referenced by get_replslot_index(), pgstat_acquire_replslot(), pgstat_create_replslot(), pgstat_drop_replslot(), pgstat_report_replslot(), pgstat_report_replslotsync(), and pgstat_reset_replslot().

◆ ReplicationSlotInitialize()

void ReplicationSlotInitialize ( void )

Definition at line 240 of file slot.c.

{
    before_shmem_exit(ReplicationSlotShmemExit, 0);
}

References before_shmem_exit(), and ReplicationSlotShmemExit().

Referenced by BaseInit().

◆ ReplicationSlotMarkDirty()

void ReplicationSlotMarkDirty ( void )

Definition at line 1180 of file slot.c.

{
    ReplicationSlot *slot = MyReplicationSlot;
 
    Assert(MyReplicationSlot != NULL);
 
    SpinLockAcquire(&slot->mutex);
    MyReplicationSlot->just_dirtied = true;
    MyReplicationSlot->dirty = true;
    SpinLockRelease(&slot->mutex);
}

References Assert, ReplicationSlot::dirty, fb(), ReplicationSlot::just_dirtied, ReplicationSlot::mutex, MyReplicationSlot, SpinLockAcquire(), and SpinLockRelease().

Referenced by copy_replication_slot(), create_physical_replication_slot(), CreateDecodingContext(), CreateInitDecodingContext(), CreateReplicationSlot(), init_conflict_slot_xmin(), InvalidatePossiblyObsoleteSlot(), LogicalConfirmReceivedLocation(), LogicalSlotAdvanceAndCheckSnapState(), pg_logical_slot_get_changes_guts(), pg_physical_replication_slot_advance(), PhysicalConfirmReceivedLocation(), PhysicalReplicationSlotNewXmin(), ReplicationSlotAlter(), ReplicationSlotPersist(), synchronize_one_slot(), update_conflict_slot_xmin(), and update_local_synced_slot().

◆ ReplicationSlotName()

bool ReplicationSlotName	(	int	index,
		Name	name
	)

Definition at line 598 of file slot.c.

{
    ReplicationSlot *slot;
    bool        found;
 
    slot = &ReplicationSlotCtl->replication_slots[index];
 
    /*
     * Ensure that the slot cannot be dropped while we copy the name. Don't
     * need the spinlock as the name of an existing slot cannot change.
     */
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    found = slot->in_use;
    if (slot->in_use)
        namestrcpy(name, NameStr(slot->data.name));
    LWLockRelease(ReplicationSlotControlLock);
 
    return found;
}

References ReplicationSlot::data, fb(), ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), name, ReplicationSlotPersistentData::name, NameStr, namestrcpy(), ReplicationSlotCtlData::replication_slots, and ReplicationSlotCtl.

Referenced by pgstat_replslot_to_serialized_name_cb().

◆ ReplicationSlotPersist()

void ReplicationSlotPersist ( void )

Definition at line 1197 of file slot.c.

{
    ReplicationSlot *slot = MyReplicationSlot;
 
    Assert(slot != NULL);
    Assert(slot->data.persistency != RS_PERSISTENT);
 
    SpinLockAcquire(&slot->mutex);
    slot->data.persistency = RS_PERSISTENT;
    SpinLockRelease(&slot->mutex);
 
    ReplicationSlotMarkDirty();
    ReplicationSlotSave();
}

References Assert, ReplicationSlot::data, fb(), ReplicationSlot::mutex, MyReplicationSlot, ReplicationSlotPersistentData::persistency, ReplicationSlotMarkDirty(), ReplicationSlotSave(), RS_PERSISTENT, SpinLockAcquire(), and SpinLockRelease().

Referenced by copy_replication_slot(), CreateReplicationSlot(), pg_create_logical_replication_slot(), and update_and_persist_local_synced_slot().

◆ ReplicationSlotRelease()

void ReplicationSlotRelease ( void )

Definition at line 769 of file slot.c.

{
    ReplicationSlot *slot = MyReplicationSlot;
    char       *slotname = NULL;    /* keep compiler quiet */
    bool        is_logical;
    TimestampTz now = 0;
 
    Assert(slot != NULL && slot->active_proc != INVALID_PROC_NUMBER);
 
    is_logical = SlotIsLogical(slot);
 
    if (am_walsender)
        slotname = pstrdup(NameStr(slot->data.name));
 
    if (slot->data.persistency == RS_EPHEMERAL)
    {
        /*
         * If slot is ephemeral, we drop it upon release, and request logical
         * decoding be disabled.
         */
        ReplicationSlotDropAcquired(is_logical);
    }
    else
    {
        /*
         * If slot needed to temporarily restrain both data and catalog xmin
         * to create the catalog snapshot, remove that temporary constraint.
         * Snapshots can only be exported while the initial snapshot is still
         * acquired.
         */
        if (!TransactionIdIsValid(slot->data.xmin) &&
            TransactionIdIsValid(slot->effective_xmin))
        {
            SpinLockAcquire(&slot->mutex);
            slot->effective_xmin = InvalidTransactionId;
            SpinLockRelease(&slot->mutex);
            ReplicationSlotsComputeRequiredXmin(false);
        }
 
        /*
         * Set the time since the slot has become inactive. We get the current
         * time beforehand to avoid system call while holding the spinlock.
         */
        now = GetCurrentTimestamp();
 
        if (slot->data.persistency == RS_PERSISTENT)
        {
            /*
             * Mark persistent slot inactive.  We're not freeing it, just
             * disconnecting, but wake up others that may be waiting for it.
             */
            SpinLockAcquire(&slot->mutex);
            slot->active_proc = INVALID_PROC_NUMBER;
            ReplicationSlotSetInactiveSince(slot, now, false);
            SpinLockRelease(&slot->mutex);
            ConditionVariableBroadcast(&slot->active_cv);
        }
        else
            ReplicationSlotSetInactiveSince(slot, now, true);
 
        MyReplicationSlot = NULL;
    }
 
    /* might not have been set when we've been a plain slot */
    LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
    MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING;
    ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags;
    LWLockRelease(ProcArrayLock);
 
    if (am_walsender)
    {
        ereport(log_replication_commands ? LOG : DEBUG1,
                is_logical
                ? errmsg("released logical replication slot \"%s\"",
                         slotname)
                : errmsg("released physical replication slot \"%s\"",
                         slotname));
 
        pfree(slotname);
    }
}

References ReplicationSlot::active_cv, ReplicationSlot::active_proc, am_walsender, Assert, ConditionVariableBroadcast(), ReplicationSlot::data, DEBUG1, ReplicationSlot::effective_xmin, ereport, errmsg, fb(), GetCurrentTimestamp(), INVALID_PROC_NUMBER, InvalidTransactionId, LOG, log_replication_commands, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyProc, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, now(), ReplicationSlotPersistentData::persistency, pfree(), PGPROC::pgxactoff, ProcGlobal, pstrdup(), ReplicationSlotDropAcquired(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotSetInactiveSince(), RS_EPHEMERAL, RS_PERSISTENT, SlotIsLogical, SpinLockAcquire(), SpinLockRelease(), PGPROC::statusFlags, PROC_HDR::statusFlags, TransactionIdIsValid, and ReplicationSlotPersistentData::xmin.

Referenced by binary_upgrade_check_logical_slot_pending_wal(), binary_upgrade_create_conflict_detection_slot(), copy_replication_slot(), CreateReplicationSlot(), InvalidatePossiblyObsoleteSlot(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_logical_slot_get_changes_guts(), pg_replication_slot_advance(), PostgresMain(), ReplicationSlotAlter(), ReplicationSlotShmemExit(), slotsync_failure_callback(), slotsync_worker_onexit(), StartLogicalReplication(), StartReplication(), synchronize_one_slot(), and WalSndErrorCleanup().

◆ ReplicationSlotReserveWal()

void ReplicationSlotReserveWal ( void )

Definition at line 1707 of file slot.c.

{
    ReplicationSlot *slot = MyReplicationSlot;
    XLogSegNo   segno;
    XLogRecPtr  restart_lsn;
 
    Assert(slot != NULL);
    Assert(!XLogRecPtrIsValid(slot->data.restart_lsn));
    Assert(!XLogRecPtrIsValid(slot->last_saved_restart_lsn));
 
    /*
     * The replication slot mechanism is used to prevent the removal of
     * required WAL.
     *
     * Acquire an exclusive lock to prevent the checkpoint process from
     * concurrently computing the minimum slot LSN (see
     * CheckPointReplicationSlots). This ensures that the WAL reserved for
     * replication cannot be removed during a checkpoint.
     *
     * The mechanism is reliable because if WAL reservation occurs first, the
     * checkpoint must wait for the restart_lsn update before determining the
     * minimum non-removable LSN. On the other hand, if the checkpoint happens
     * first, subsequent WAL reservations will select positions at or beyond
     * the redo pointer of that checkpoint.
     */
    LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
 
    /*
     * For logical slots log a standby snapshot and start logical decoding at
     * exactly that position. That allows the slot to start up more quickly.
     * But on a standby we cannot do WAL writes, so just use the replay
     * pointer; effectively, an attempt to create a logical slot on standby
     * will cause it to wait for an xl_running_xact record to be logged
     * independently on the primary, so that a snapshot can be built using the
     * record.
     *
     * None of this is needed (or indeed helpful) for physical slots as
     * they'll start replay at the last logged checkpoint anyway. Instead,
     * return the location of the last redo LSN, where a base backup has to
     * start replay at.
     */
    if (SlotIsPhysical(slot))
        restart_lsn = GetRedoRecPtr();
    else if (RecoveryInProgress())
        restart_lsn = GetXLogReplayRecPtr(NULL);
    else
        restart_lsn = GetXLogInsertRecPtr();
 
    SpinLockAcquire(&slot->mutex);
    slot->data.restart_lsn = restart_lsn;
    SpinLockRelease(&slot->mutex);
 
    /* prevent WAL removal as fast as possible */
    ReplicationSlotsComputeRequiredLSN();
 
    /* Checkpoint shouldn't remove the required WAL. */
    XLByteToSeg(slot->data.restart_lsn, segno, wal_segment_size);
    if (XLogGetLastRemovedSegno() >= segno)
        elog(ERROR, "WAL required by replication slot %s has been removed concurrently",
             NameStr(slot->data.name));
 
    LWLockRelease(ReplicationSlotAllocationLock);
 
    if (!RecoveryInProgress() && SlotIsLogical(slot))
    {
        XLogRecPtr  flushptr;
 
        /* make sure we have enough information to start */
        flushptr = LogStandbySnapshot();
 
        /* and make sure it's fsynced to disk */
        XLogFlush(flushptr);
    }
}

References Assert, ReplicationSlot::data, elog, ERROR, fb(), GetRedoRecPtr(), GetXLogInsertRecPtr(), GetXLogReplayRecPtr(), ReplicationSlot::last_saved_restart_lsn, LogStandbySnapshot(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, RecoveryInProgress(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotPersistentData::restart_lsn, SlotIsLogical, SlotIsPhysical, SpinLockAcquire(), SpinLockRelease(), wal_segment_size, XLByteToSeg, XLogFlush(), XLogGetLastRemovedSegno(), and XLogRecPtrIsValid.

Referenced by create_physical_replication_slot(), CreateInitDecodingContext(), and CreateReplicationSlot().

◆ ReplicationSlotSave()

void ReplicationSlotSave ( void )

Definition at line 1162 of file slot.c.

{
    char        path[MAXPGPATH];
 
    Assert(MyReplicationSlot != NULL);
 
    sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(MyReplicationSlot->data.name));
    SaveSlotToPath(MyReplicationSlot, path, ERROR);
}

References Assert, ReplicationSlot::data, ERROR, fb(), MAXPGPATH, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, PG_REPLSLOT_DIR, SaveSlotToPath(), and sprintf.

Referenced by copy_replication_slot(), create_physical_replication_slot(), CreateDecodingContext(), CreateInitDecodingContext(), CreateReplicationSlot(), init_conflict_slot_xmin(), InvalidatePossiblyObsoleteSlot(), LogicalConfirmReceivedLocation(), ReplicationSlotAlter(), ReplicationSlotPersist(), synchronize_one_slot(), and update_local_synced_slot().

◆ ReplicationSlotsComputeLogicalRestartLSN()

XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN ( void )

Definition at line 1374 of file slot.c.

{
    XLogRecPtr  result = InvalidXLogRecPtr;
    int         i;
 
    if (max_replication_slots + max_repack_replication_slots <= 0)
        return InvalidXLogRecPtr;
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s;
        XLogRecPtr  restart_lsn;
        XLogRecPtr  last_saved_restart_lsn;
        bool        invalidated;
        ReplicationSlotPersistency persistency;
 
        s = &ReplicationSlotCtl->replication_slots[i];
 
        /* cannot change while ReplicationSlotCtlLock is held */
        if (!s->in_use)
            continue;
 
        /* we're only interested in logical slots */
        if (!SlotIsLogical(s))
            continue;
 
        /* read once, it's ok if it increases while we're checking */
        SpinLockAcquire(&s->mutex);
        persistency = s->data.persistency;
        restart_lsn = s->data.restart_lsn;
        invalidated = s->data.invalidated != RS_INVAL_NONE;
        last_saved_restart_lsn = s->last_saved_restart_lsn;
        SpinLockRelease(&s->mutex);
 
        /* invalidated slots need not apply */
        if (invalidated)
            continue;
 
        /*
         * For persistent slot use last_saved_restart_lsn to compute the
         * oldest LSN for removal of WAL segments.  The segments between
         * last_saved_restart_lsn and restart_lsn might be needed by a
         * persistent slot in the case of database crash.  Non-persistent
         * slots can't survive the database crash, so we don't care about
         * last_saved_restart_lsn for them.
         */
        if (persistency == RS_PERSISTENT)
        {
            if (XLogRecPtrIsValid(last_saved_restart_lsn) &&
                restart_lsn > last_saved_restart_lsn)
            {
                restart_lsn = last_saved_restart_lsn;
            }
        }
 
        if (!XLogRecPtrIsValid(restart_lsn))
            continue;
 
        if (!XLogRecPtrIsValid(result) ||
            restart_lsn < result)
            result = restart_lsn;
    }
 
    LWLockRelease(ReplicationSlotControlLock);
 
    return result;
}

References ReplicationSlot::data, fb(), i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, InvalidXLogRecPtr, ReplicationSlot::last_saved_restart_lsn, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotPersistentData::restart_lsn, result, RS_INVAL_NONE, RS_PERSISTENT, SlotIsLogical, SpinLockAcquire(), SpinLockRelease(), and XLogRecPtrIsValid.

Referenced by CheckPointLogicalRewriteHeap(), and CheckPointSnapBuild().

◆ ReplicationSlotsComputeRequiredLSN()

void ReplicationSlotsComputeRequiredLSN ( void )

Definition at line 1304 of file slot.c.

{
    int         i;
    XLogRecPtr  min_required = InvalidXLogRecPtr;
 
    Assert(ReplicationSlotCtl != NULL);
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
        XLogRecPtr  restart_lsn;
        XLogRecPtr  last_saved_restart_lsn;
        bool        invalidated;
        ReplicationSlotPersistency persistency;
 
        if (!s->in_use)
            continue;
 
        SpinLockAcquire(&s->mutex);
        persistency = s->data.persistency;
        restart_lsn = s->data.restart_lsn;
        invalidated = s->data.invalidated != RS_INVAL_NONE;
        last_saved_restart_lsn = s->last_saved_restart_lsn;
        SpinLockRelease(&s->mutex);
 
        /* invalidated slots need not apply */
        if (invalidated)
            continue;
 
        /*
         * For persistent slot use last_saved_restart_lsn to compute the
         * oldest LSN for removal of WAL segments.  The segments between
         * last_saved_restart_lsn and restart_lsn might be needed by a
         * persistent slot in the case of database crash.  Non-persistent
         * slots can't survive the database crash, so we don't care about
         * last_saved_restart_lsn for them.
         */
        if (persistency == RS_PERSISTENT)
        {
            if (XLogRecPtrIsValid(last_saved_restart_lsn) &&
                restart_lsn > last_saved_restart_lsn)
            {
                restart_lsn = last_saved_restart_lsn;
            }
        }
 
        if (XLogRecPtrIsValid(restart_lsn) &&
            (!XLogRecPtrIsValid(min_required) ||
             restart_lsn < min_required))
            min_required = restart_lsn;
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    XLogSetReplicationSlotMinimumLSN(min_required);
}

References Assert, ReplicationSlot::data, fb(), i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, InvalidXLogRecPtr, ReplicationSlot::last_saved_restart_lsn, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, ReplicationSlotPersistentData::persistency, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotPersistentData::restart_lsn, RS_INVAL_NONE, RS_PERSISTENT, SpinLockAcquire(), SpinLockRelease(), XLogRecPtrIsValid, and XLogSetReplicationSlotMinimumLSN().

Referenced by CheckPointReplicationSlots(), copy_replication_slot(), InvalidateObsoleteReplicationSlots(), LogicalConfirmReceivedLocation(), pg_replication_slot_advance(), PhysicalConfirmReceivedLocation(), ReplicationSlotDropPtr(), ReplicationSlotReserveWal(), reserve_wal_for_local_slot(), StartupReplicationSlots(), and update_local_synced_slot().

◆ ReplicationSlotsComputeRequiredXmin()

void ReplicationSlotsComputeRequiredXmin ( bool already_locked )

Definition at line 1222 of file slot.c.

{
    int         i;
    TransactionId agg_xmin = InvalidTransactionId;
    TransactionId agg_catalog_xmin = InvalidTransactionId;
 
    Assert(ReplicationSlotCtl != NULL);
    Assert(!already_locked ||
           (LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_EXCLUSIVE) &&
            LWLockHeldByMeInMode(ProcArrayLock, LW_EXCLUSIVE)));
 
    /*
     * Hold the ReplicationSlotControlLock until after updating the slot xmin
     * values, so no backend updates the initial xmin for newly created slot
     * concurrently. A shared lock is used here to minimize lock contention,
     * especially when many slots exist and advancements occur frequently.
     * This is safe since an exclusive lock is taken during initial slot xmin
     * update in slot creation.
     *
     * One might think that we can hold the ProcArrayLock exclusively and
     * update the slot xmin values, but it could increase lock contention on
     * the ProcArrayLock, which is not great since this function can be called
     * at non-negligible frequency.
     *
     * Concurrent invocation of this function may cause the computed slot xmin
     * to regress. However, this is harmless because tuples prior to the most
     * recent xmin are no longer useful once advancement occurs (see
     * LogicalConfirmReceivedLocation where the slot's xmin value is flushed
     * before updating the effective_xmin). Thus, such regression merely
     * prevents VACUUM from prematurely removing tuples without causing the
     * early deletion of required data.
     */
    if (!already_locked)
        LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
        TransactionId effective_xmin;
        TransactionId effective_catalog_xmin;
        bool        invalidated;
 
        if (!s->in_use)
            continue;
 
        SpinLockAcquire(&s->mutex);
        effective_xmin = s->effective_xmin;
        effective_catalog_xmin = s->effective_catalog_xmin;
        invalidated = s->data.invalidated != RS_INVAL_NONE;
        SpinLockRelease(&s->mutex);
 
        /* invalidated slots need not apply */
        if (invalidated)
            continue;
 
        /* check the data xmin */
        if (TransactionIdIsValid(effective_xmin) &&
            (!TransactionIdIsValid(agg_xmin) ||
             TransactionIdPrecedes(effective_xmin, agg_xmin)))
            agg_xmin = effective_xmin;
 
        /* check the catalog xmin */
        if (TransactionIdIsValid(effective_catalog_xmin) &&
            (!TransactionIdIsValid(agg_catalog_xmin) ||
             TransactionIdPrecedes(effective_catalog_xmin, agg_catalog_xmin)))
            agg_catalog_xmin = effective_catalog_xmin;
    }
 
    ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
 
    if (!already_locked)
        LWLockRelease(ReplicationSlotControlLock);
}

References Assert, ReplicationSlot::data, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, fb(), i, ReplicationSlot::in_use, ReplicationSlotPersistentData::invalidated, InvalidTransactionId, LW_EXCLUSIVE, LW_SHARED, LWLockAcquire(), LWLockHeldByMeInMode(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, ProcArraySetReplicationSlotXmin(), ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, RS_INVAL_NONE, SpinLockAcquire(), SpinLockRelease(), TransactionIdIsValid, and TransactionIdPrecedes().

Referenced by copy_replication_slot(), CreateInitDecodingContext(), init_conflict_slot_xmin(), InvalidateObsoleteReplicationSlots(), LogicalConfirmReceivedLocation(), pg_replication_slot_advance(), PhysicalReplicationSlotNewXmin(), ReplicationSlotDropPtr(), ReplicationSlotRelease(), StartupReplicationSlots(), synchronize_one_slot(), update_conflict_slot_xmin(), and update_local_synced_slot().

◆ ReplicationSlotsCountDBSlots()

bool ReplicationSlotsCountDBSlots	(	Oid	dboid,
		int *	nslots,
		int *	nactive
	)

Definition at line 1453 of file slot.c.

{
    int         i;
 
    *nslots = *nactive = 0;
 
    if (max_replication_slots + max_repack_replication_slots <= 0)
        return false;
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s;
 
        s = &ReplicationSlotCtl->replication_slots[i];
 
        /* cannot change while ReplicationSlotCtlLock is held */
        if (!s->in_use)
            continue;
 
        /* only logical slots are database specific, skip */
        if (!SlotIsLogical(s))
            continue;
 
        /* not our database, skip */
        if (s->data.database != dboid)
            continue;
 
        /* NB: intentionally counting invalidated slots */
 
        /* count slots with spinlock held */
        SpinLockAcquire(&s->mutex);
        (*nslots)++;
        if (s->active_proc != INVALID_PROC_NUMBER)
            (*nactive)++;
        SpinLockRelease(&s->mutex);
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    if (*nslots > 0)
        return true;
    return false;
}

References ReplicationSlot::active_proc, ReplicationSlot::data, ReplicationSlotPersistentData::database, fb(), i, ReplicationSlot::in_use, INVALID_PROC_NUMBER, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, SlotIsLogical, SpinLockAcquire(), and SpinLockRelease().

Referenced by dropdb().

◆ ReplicationSlotsDropDBSlots()

void ReplicationSlotsDropDBSlots ( Oid dboid )

Definition at line 1514 of file slot.c.

{
    int         i;
    bool        found_valid_logicalslot;
    bool        dropped = false;
 
    if (max_replication_slots + max_repack_replication_slots <= 0)
        return;
 
restart:
    found_valid_logicalslot = false;
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s;
        char       *slotname;
        ProcNumber  active_proc;
 
        s = &ReplicationSlotCtl->replication_slots[i];
 
        /* cannot change while ReplicationSlotCtlLock is held */
        if (!s->in_use)
            continue;
 
        /* only logical slots are database specific, skip */
        if (!SlotIsLogical(s))
            continue;
 
        /*
         * Check logical slots on other databases too so we can disable
         * logical decoding only if no slots in the cluster.
         */
        SpinLockAcquire(&s->mutex);
        found_valid_logicalslot |= (s->data.invalidated == RS_INVAL_NONE);
        SpinLockRelease(&s->mutex);
 
        /* not our database, skip */
        if (s->data.database != dboid)
            continue;
 
        /* NB: intentionally including invalidated slots to drop */
 
        /* acquire slot, so ReplicationSlotDropAcquired can be reused  */
        SpinLockAcquire(&s->mutex);
        /* can't change while ReplicationSlotControlLock is held */
        slotname = NameStr(s->data.name);
        active_proc = s->active_proc;
        if (active_proc == INVALID_PROC_NUMBER)
        {
            MyReplicationSlot = s;
            s->active_proc = MyProcNumber;
        }
        SpinLockRelease(&s->mutex);
 
        /*
         * Even though we hold an exclusive lock on the database object a
         * logical slot for that DB can still be active, e.g. if it's
         * concurrently being dropped by a backend connected to another DB.
         *
         * That's fairly unlikely in practice, so we'll just bail out.
         *
         * The slot sync worker holds a shared lock on the database before
         * operating on synced logical slots to avoid conflict with the drop
         * happening here. The persistent synced slots are thus safe but there
         * is a possibility that the slot sync worker has created a temporary
         * slot (which stays active even on release) and we are trying to drop
         * that here. In practice, the chances of hitting this scenario are
         * less as during slot synchronization, the temporary slot is
         * immediately converted to persistent and thus is safe due to the
         * shared lock taken on the database. So, we'll just bail out in such
         * a case.
         *
         * XXX: We can consider shutting down the slot sync worker before
         * trying to drop synced temporary slots here.
         */
        if (active_proc != INVALID_PROC_NUMBER)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_IN_USE),
                     errmsg("replication slot \"%s\" is active for PID %d",
                            slotname, GetPGProcByNumber(active_proc)->pid)));
 
        /*
         * To avoid duplicating ReplicationSlotDropAcquired() and to avoid
         * holding ReplicationSlotControlLock over filesystem operations,
         * release ReplicationSlotControlLock and use
         * ReplicationSlotDropAcquired.
         *
         * As that means the set of slots could change, restart scan from the
         * beginning each time we release the lock.
         */
        LWLockRelease(ReplicationSlotControlLock);
        ReplicationSlotDropAcquired(false);
        dropped = true;
        goto restart;
    }
    LWLockRelease(ReplicationSlotControlLock);
 
    if (dropped && !found_valid_logicalslot)
        RequestDisableLogicalDecoding();
}

References ReplicationSlot::active_proc, ReplicationSlot::data, ReplicationSlotPersistentData::database, ereport, errcode(), errmsg, ERROR, fb(), GetPGProcByNumber, i, ReplicationSlot::in_use, INVALID_PROC_NUMBER, ReplicationSlotPersistentData::invalidated, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, MyProcNumber, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotDropAcquired(), RequestDisableLogicalDecoding(), RS_INVAL_NONE, SlotIsLogical, SpinLockAcquire(), and SpinLockRelease().

Referenced by dbase_redo(), and dropdb().

◆ ReplicationSlotShmemExit()

static void ReplicationSlotShmemExit	(	int	code,
		Datum	arg
	)

static

Definition at line 249 of file slot.c.

{
    /* Make sure active replication slots are released */
    if (MyReplicationSlot != NULL)
        ReplicationSlotRelease();
 
    /* Also cleanup all the temporary slots. */
    ReplicationSlotCleanup(false);
}

References fb(), MyReplicationSlot, ReplicationSlotCleanup(), and ReplicationSlotRelease().

Referenced by ReplicationSlotInitialize().

◆ ReplicationSlotsShmemInit()

static void ReplicationSlotsShmemInit ( void * arg )

static

Definition at line 221 of file slot.c.

{
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *slot = &ReplicationSlotCtl->replication_slots[i];
 
        /* everything else is zeroed by the memset above */
        slot->active_proc = INVALID_PROC_NUMBER;
        SpinLockInit(&slot->mutex);
        LWLockInitialize(&slot->io_in_progress_lock,
                         LWTRANCHE_REPLICATION_SLOT_IO);
        ConditionVariableInit(&slot->active_cv);
    }
}

References ReplicationSlot::active_cv, ReplicationSlot::active_proc, ConditionVariableInit(), fb(), i, INVALID_PROC_NUMBER, ReplicationSlot::io_in_progress_lock, LWLockInitialize(), max_repack_replication_slots, max_replication_slots, ReplicationSlot::mutex, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, and SpinLockInit().

◆ ReplicationSlotsShmemRequest()

static void ReplicationSlotsShmemRequest ( void * arg )

static

Definition at line 200 of file slot.c.

{
    Size        size;
 
    if (max_replication_slots + max_repack_replication_slots == 0)
        return;
 
    size = offsetof(ReplicationSlotCtlData, replication_slots);
    size = add_size(size,
                    mul_size(max_replication_slots + max_repack_replication_slots,
                             sizeof(ReplicationSlot)));
    ShmemRequestStruct(.name = "ReplicationSlot Ctl",
                       .size = size,
                       .ptr = (void **) &ReplicationSlotCtl,
        );
}

References add_size(), fb(), max_repack_replication_slots, max_replication_slots, mul_size(), name, ReplicationSlotCtl, and ShmemRequestStruct.

◆ ReplicationSlotValidateName()

bool ReplicationSlotValidateName	(	const char *	name,
		bool	allow_reserved_name,
		int	elevel
	)

Definition at line 265 of file slot.c.

{
    int         err_code;
    char       *err_msg = NULL;
    char       *err_hint = NULL;
 
    if (!ReplicationSlotValidateNameInternal(name, allow_reserved_name,
                                             &err_code, &err_msg, &err_hint))
    {
        /*
         * Use errmsg_internal() and errhint_internal() instead of errmsg()
         * and errhint(), since the messages from
         * ReplicationSlotValidateNameInternal() are already translated. This
         * avoids double translation.
         */
        ereport(elevel,
                errcode(err_code),
                errmsg_internal("%s", err_msg),
                (err_hint != NULL) ? errhint_internal("%s", err_hint) : 0);
 
        pfree(err_msg);
        if (err_hint != NULL)
            pfree(err_hint);
        return false;
    }
 
    return true;
}

References ereport, errcode(), errhint_internal(), errmsg_internal(), fb(), name, pfree(), and ReplicationSlotValidateNameInternal().

Referenced by parse_subscription_options(), ReplicationSlotCreate(), and StartupReorderBuffer().

◆ ReplicationSlotValidateNameInternal()

bool ReplicationSlotValidateNameInternal	(	const char *	name,
		bool	allow_reserved_name,
		int *	err_code,
		char **	err_msg,
		char **	err_hint
	)

Definition at line 310 of file slot.c.

{
    const char *cp;
 
    if (strlen(name) == 0)
    {
        *err_code = ERRCODE_INVALID_NAME;
        *err_msg = psprintf(_("replication slot name \"%s\" is too short"), name);
        *err_hint = NULL;
        return false;
    }
 
    if (strlen(name) >= NAMEDATALEN)
    {
        *err_code = ERRCODE_NAME_TOO_LONG;
        *err_msg = psprintf(_("replication slot name \"%s\" is too long"), name);
        *err_hint = NULL;
        return false;
    }
 
    for (cp = name; *cp; cp++)
    {
        if (!((*cp >= 'a' && *cp <= 'z')
              || (*cp >= '0' && *cp <= '9')
              || (*cp == '_')))
        {
            *err_code = ERRCODE_INVALID_NAME;
            *err_msg = psprintf(_("replication slot name \"%s\" contains invalid character"), name);
            *err_hint = psprintf(_("Replication slot names may only contain lower case letters, numbers, and the underscore character."));
            return false;
        }
    }
 
    if (!allow_reserved_name && IsSlotForConflictCheck(name))
    {
        *err_code = ERRCODE_RESERVED_NAME;
        *err_msg = psprintf(_("replication slot name \"%s\" is reserved"), name);
        *err_hint = psprintf(_("The name \"%s\" is reserved for the conflict detection slot."),
                             CONFLICT_DETECTION_SLOT);
        return false;
    }
 
    return true;
}

References _, CONFLICT_DETECTION_SLOT, fb(), IsSlotForConflictCheck(), name, NAMEDATALEN, and psprintf().

Referenced by check_primary_slot_name(), ReplicationSlotValidateName(), and validate_sync_standby_slots().

◆ ReportSlotInvalidation()

static void ReportSlotInvalidation	(	ReplicationSlotInvalidationCause	cause,
		bool	terminating,
		int	pid,
		NameData	slotname,
		XLogRecPtr	restart_lsn,
		XLogRecPtr	oldestLSN,
		TransactionId	snapshotConflictHorizon,
		long	slot_idle_seconds
	)

static

Definition at line 1786 of file slot.c.

{
    StringInfoData err_detail;
    StringInfoData err_hint;
 
    initStringInfo(&err_detail);
    initStringInfo(&err_hint);
 
    switch (cause)
    {
        case RS_INVAL_WAL_REMOVED:
            {
                uint64      ex = oldestLSN - restart_lsn;
 
                appendStringInfo(&err_detail,
                                 ngettext("The slot's restart_lsn %X/%08X exceeds the limit by %" PRIu64 " byte.",
                                          "The slot's restart_lsn %X/%08X exceeds the limit by %" PRIu64 " bytes.",
                                          ex),
                                 LSN_FORMAT_ARGS(restart_lsn),
                                 ex);
                /* translator: %s is a GUC variable name */
                appendStringInfo(&err_hint, _("You might need to increase \"%s\"."),
                                 "max_slot_wal_keep_size");
                break;
            }
        case RS_INVAL_HORIZON:
            appendStringInfo(&err_detail, _("The slot conflicted with xid horizon %u."),
                             snapshotConflictHorizon);
            break;
 
        case RS_INVAL_WAL_LEVEL:
            appendStringInfoString(&err_detail, _("Logical decoding on standby requires the primary server to either set \"wal_level\" >= \"logical\" or have at least one logical slot when \"wal_level\" = \"replica\"."));
            break;
 
        case RS_INVAL_IDLE_TIMEOUT:
            {
                /* translator: %s is a GUC variable name */
                appendStringInfo(&err_detail, _("The slot's idle time of %lds exceeds the configured \"%s\" duration of %ds."),
                                 slot_idle_seconds, "idle_replication_slot_timeout",
                                 idle_replication_slot_timeout_secs);
                /* translator: %s is a GUC variable name */
                appendStringInfo(&err_hint, _("You might need to increase \"%s\"."),
                                 "idle_replication_slot_timeout");
                break;
            }
        case RS_INVAL_NONE:
            pg_unreachable();
    }
 
    ereport(LOG,
            terminating ?
            errmsg("terminating process %d to release replication slot \"%s\"",
                   pid, NameStr(slotname)) :
            errmsg("invalidating obsolete replication slot \"%s\"",
                   NameStr(slotname)),
            errdetail_internal("%s", err_detail.data),
            err_hint.len ? errhint("%s", err_hint.data) : 0);
 
    pfree(err_detail.data);
    pfree(err_hint.data);
}

References _, appendStringInfo(), appendStringInfoString(), ereport, errdetail_internal(), errhint(), errmsg, fb(), idle_replication_slot_timeout_secs, initStringInfo(), LOG, LSN_FORMAT_ARGS, NameStr, ngettext, pfree(), pg_unreachable, RS_INVAL_HORIZON, RS_INVAL_IDLE_TIMEOUT, RS_INVAL_NONE, RS_INVAL_WAL_LEVEL, and RS_INVAL_WAL_REMOVED.

Referenced by InvalidatePossiblyObsoleteSlot().

◆ RestoreSlotFromDisk()

static void RestoreSlotFromDisk ( const char * name )

static

Definition at line 2683 of file slot.c.

{
    ReplicationSlotOnDisk cp;
    int         i;
    char        slotdir[MAXPGPATH + sizeof(PG_REPLSLOT_DIR)];
    char        path[MAXPGPATH + sizeof(PG_REPLSLOT_DIR) + 10];
    int         fd;
    bool        restored = false;
    int         readBytes;
    pg_crc32c   checksum;
    TimestampTz now = 0;
 
    /* no need to lock here, no concurrent access allowed yet */
 
    /* delete temp file if it exists */
    sprintf(slotdir, "%s/%s", PG_REPLSLOT_DIR, name);
    sprintf(path, "%s/state.tmp", slotdir);
    if (unlink(path) < 0 && errno != ENOENT)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not remove file \"%s\": %m", path)));
 
    sprintf(path, "%s/state", slotdir);
 
    elog(DEBUG1, "restoring replication slot from \"%s\"", path);
 
    /* on some operating systems fsyncing a file requires O_RDWR */
    fd = OpenTransientFile(path, O_RDWR | PG_BINARY);
 
    /*
     * We do not need to handle this as we are rename()ing the directory into
     * place only after we fsync()ed the state file.
     */
    if (fd < 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m", path)));
 
    /*
     * Sync state file before we're reading from it. We might have crashed
     * while it wasn't synced yet and we shouldn't continue on that basis.
     */
    pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_RESTORE_SYNC);
    if (pg_fsync(fd) != 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not fsync file \"%s\": %m",
                        path)));
    pgstat_report_wait_end();
 
    /* Also sync the parent directory */
    START_CRIT_SECTION();
    fsync_fname(slotdir, true);
    END_CRIT_SECTION();
 
    /* read part of statefile that's guaranteed to be version independent */
    pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
    readBytes = read(fd, &cp, ReplicationSlotOnDiskConstantSize);
    pgstat_report_wait_end();
    if (readBytes != ReplicationSlotOnDiskConstantSize)
    {
        if (readBytes < 0)
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m", path)));
        else
            ereport(PANIC,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("could not read file \"%s\": read %d of %zu",
                            path, readBytes,
                            (Size) ReplicationSlotOnDiskConstantSize)));
    }
 
    /* verify magic */
    if (cp.magic != SLOT_MAGIC)
        ereport(PANIC,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("replication slot file \"%s\" has wrong magic number: %u instead of %u",
                        path, cp.magic, SLOT_MAGIC)));
 
    /* verify version */
    if (cp.version != SLOT_VERSION)
        ereport(PANIC,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("replication slot file \"%s\" has unsupported version %u",
                        path, cp.version)));
 
    /* boundary check on length */
    if (cp.length != ReplicationSlotOnDiskV2Size)
        ereport(PANIC,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("replication slot file \"%s\" has corrupted length %u",
                        path, cp.length)));
 
    /* Now that we know the size, read the entire file */
    pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_READ);
    readBytes = read(fd,
                     (char *) &cp + ReplicationSlotOnDiskConstantSize,
                     cp.length);
    pgstat_report_wait_end();
    if (readBytes != cp.length)
    {
        if (readBytes < 0)
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m", path)));
        else
            ereport(PANIC,
                    (errcode(ERRCODE_DATA_CORRUPTED),
                     errmsg("could not read file \"%s\": read %d of %zu",
                            path, readBytes, (Size) cp.length)));
    }
 
    if (CloseTransientFile(fd) != 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m", path)));
 
    /* now verify the CRC */
    INIT_CRC32C(checksum);
    COMP_CRC32C(checksum,
                (char *) &cp + ReplicationSlotOnDiskNotChecksummedSize,
                ReplicationSlotOnDiskChecksummedSize);
    FIN_CRC32C(checksum);
 
    if (!EQ_CRC32C(checksum, cp.checksum))
        ereport(PANIC,
                (errmsg("checksum mismatch for replication slot file \"%s\": is %u, should be %u",
                        path, checksum, cp.checksum)));
 
    /*
     * If we crashed with an ephemeral slot active, don't restore but delete
     * it.
     */
    if (cp.slotdata.persistency != RS_PERSISTENT)
    {
        if (!rmtree(slotdir, true))
        {
            ereport(WARNING,
                    (errmsg("could not remove directory \"%s\"",
                            slotdir)));
        }
        fsync_fname(PG_REPLSLOT_DIR, true);
        return;
    }
 
    /*
     * Verify that requirements for the specific slot type are met. That's
     * important because if these aren't met we're not guaranteed to retain
     * all the necessary resources for the slot.
     *
     * NB: We have to do so *after* the above checks for ephemeral slots,
     * because otherwise a slot that shouldn't exist anymore could prevent
     * restarts.
     *
     * NB: Changing the requirements here also requires adapting
     * CheckSlotRequirements() and CheckLogicalDecodingRequirements().
     */
    if (cp.slotdata.database != InvalidOid)
    {
        if (wal_level < WAL_LEVEL_REPLICA)
            ereport(FATAL,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("logical replication slot \"%s\" exists, but \"wal_level\" < \"replica\"",
                            NameStr(cp.slotdata.name)),
                     errhint("Change \"wal_level\" to be \"replica\" or higher.")));
 
        /*
         * In standby mode, the hot standby must be enabled. This check is
         * necessary to ensure logical slots are invalidated when they become
         * incompatible due to insufficient wal_level. Otherwise, if the
         * primary reduces effective_wal_level < logical while hot standby is
         * disabled, primary disable logical decoding while hot standby is
         * disabled, logical slots would remain valid even after promotion.
         */
        if (StandbyMode && !EnableHotStandby)
            ereport(FATAL,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("logical replication slot \"%s\" exists on the standby, but \"hot_standby\" = \"off\"",
                            NameStr(cp.slotdata.name)),
                     errhint("Change \"hot_standby\" to be \"on\".")));
    }
    else if (wal_level < WAL_LEVEL_REPLICA)
        ereport(FATAL,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("physical replication slot \"%s\" exists, but \"wal_level\" < \"replica\"",
                        NameStr(cp.slotdata.name)),
                 errhint("Change \"wal_level\" to be \"replica\" or higher.")));
 
    /*
     * Nothing can be active yet, don't lock anything.  Note we iterate up to
     * max_replication_slots instead of adding max_repack_replication_slots as
     * in all other places, because we must enforce the GUC value in case
     * there were more slots before the shutdown than what it is set up to
     * now.
     */
    for (i = 0; i < max_replication_slots; i++)
    {
        ReplicationSlot *slot;
 
        slot = &ReplicationSlotCtl->replication_slots[i];
 
        if (slot->in_use)
            continue;
 
        /* restore the entire set of persistent data */
        memcpy(&slot->data, &cp.slotdata,
               sizeof(ReplicationSlotPersistentData));
 
        /* initialize in memory state */
        slot->effective_xmin = cp.slotdata.xmin;
        slot->effective_catalog_xmin = cp.slotdata.catalog_xmin;
        slot->last_saved_confirmed_flush = cp.slotdata.confirmed_flush;
        slot->last_saved_restart_lsn = cp.slotdata.restart_lsn;
 
        slot->candidate_catalog_xmin = InvalidTransactionId;
        slot->candidate_xmin_lsn = InvalidXLogRecPtr;
        slot->candidate_restart_lsn = InvalidXLogRecPtr;
        slot->candidate_restart_valid = InvalidXLogRecPtr;
 
        slot->in_use = true;
        slot->active_proc = INVALID_PROC_NUMBER;
 
        /*
         * Set the time since the slot has become inactive after loading the
         * slot from the disk into memory. Whoever acquires the slot i.e.
         * makes the slot active will reset it. Use the same inactive_since
         * time for all the slots.
         */
        if (now == 0)
            now = GetCurrentTimestamp();
 
        ReplicationSlotSetInactiveSince(slot, now, false);
 
        restored = true;
        break;
    }
 
    if (!restored)
        ereport(FATAL,
                (errmsg("too many replication slots active before shutdown"),
                 errhint("Increase \"max_replication_slots\" and try again.")));
}

References ReplicationSlot::active_proc, ReplicationSlot::candidate_catalog_xmin, ReplicationSlot::candidate_restart_lsn, ReplicationSlot::candidate_restart_valid, ReplicationSlot::candidate_xmin_lsn, CloseTransientFile(), COMP_CRC32C, ReplicationSlot::data, DEBUG1, ReplicationSlot::effective_catalog_xmin, ReplicationSlot::effective_xmin, elog, EnableHotStandby, END_CRIT_SECTION, EQ_CRC32C, ereport, errcode(), ERRCODE_DATA_CORRUPTED, errcode_for_file_access(), errhint(), errmsg, FATAL, fb(), fd(), FIN_CRC32C, fsync_fname(), GetCurrentTimestamp(), i, ReplicationSlot::in_use, INIT_CRC32C, INVALID_PROC_NUMBER, InvalidOid, InvalidTransactionId, InvalidXLogRecPtr, ReplicationSlot::last_saved_confirmed_flush, ReplicationSlot::last_saved_restart_lsn, max_replication_slots, MAXPGPATH, memcpy(), name, NameStr, now(), OpenTransientFile(), PANIC, PG_BINARY, pg_fsync(), PG_REPLSLOT_DIR, pgstat_report_wait_end(), pgstat_report_wait_start(), read, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotOnDiskChecksummedSize, ReplicationSlotOnDiskConstantSize, ReplicationSlotOnDiskNotChecksummedSize, ReplicationSlotOnDiskV2Size, ReplicationSlotSetInactiveSince(), rmtree(), RS_PERSISTENT, SLOT_MAGIC, SLOT_VERSION, sprintf, StandbyMode, START_CRIT_SECTION, wal_level, WAL_LEVEL_REPLICA, and WARNING.

Referenced by StartupReplicationSlots().

◆ SaveSlotToPath()

static void SaveSlotToPath	(	ReplicationSlot *	slot,
		const char *	dir,
		int	elevel
	)

static

Definition at line 2520 of file slot.c.

{
    char        tmppath[MAXPGPATH];
    char        path[MAXPGPATH];
    int         fd;
    ReplicationSlotOnDisk cp;
    bool        was_dirty;
 
    /* first check whether there's something to write out */
    SpinLockAcquire(&slot->mutex);
    was_dirty = slot->dirty;
    slot->just_dirtied = false;
    SpinLockRelease(&slot->mutex);
 
    /* and don't do anything if there's nothing to write */
    if (!was_dirty)
        return;
 
    LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE);
 
    /* silence valgrind :( */
    memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
 
    sprintf(tmppath, "%s/state.tmp", dir);
    sprintf(path, "%s/state", dir);
 
    fd = OpenTransientFile(tmppath, O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
    if (fd < 0)
    {
        /*
         * If not an ERROR, then release the lock before returning.  In case
         * of an ERROR, the error recovery path automatically releases the
         * lock, but no harm in explicitly releasing even in that case.  Note
         * that LWLockRelease() could affect errno.
         */
        int         save_errno = errno;
 
        LWLockRelease(&slot->io_in_progress_lock);
        errno = save_errno;
        ereport(elevel,
                (errcode_for_file_access(),
                 errmsg("could not create file \"%s\": %m",
                        tmppath)));
        return;
    }
 
    cp.magic = SLOT_MAGIC;
    INIT_CRC32C(cp.checksum);
    cp.version = SLOT_VERSION;
    cp.length = ReplicationSlotOnDiskV2Size;
 
    SpinLockAcquire(&slot->mutex);
 
    memcpy(&cp.slotdata, &slot->data, sizeof(ReplicationSlotPersistentData));
 
    SpinLockRelease(&slot->mutex);
 
    COMP_CRC32C(cp.checksum,
                (char *) (&cp) + ReplicationSlotOnDiskNotChecksummedSize,
                ReplicationSlotOnDiskChecksummedSize);
    FIN_CRC32C(cp.checksum);
 
    errno = 0;
    pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_WRITE);
    if ((write(fd, &cp, sizeof(cp))) != sizeof(cp))
    {
        int         save_errno = errno;
 
        pgstat_report_wait_end();
        CloseTransientFile(fd);
        unlink(tmppath);
        LWLockRelease(&slot->io_in_progress_lock);
 
        /* if write didn't set errno, assume problem is no disk space */
        errno = save_errno ? save_errno : ENOSPC;
        ereport(elevel,
                (errcode_for_file_access(),
                 errmsg("could not write to file \"%s\": %m",
                        tmppath)));
        return;
    }
    pgstat_report_wait_end();
 
    /* fsync the temporary file */
    pgstat_report_wait_start(WAIT_EVENT_REPLICATION_SLOT_SYNC);
    if (pg_fsync(fd) != 0)
    {
        int         save_errno = errno;
 
        pgstat_report_wait_end();
        CloseTransientFile(fd);
        unlink(tmppath);
        LWLockRelease(&slot->io_in_progress_lock);
 
        errno = save_errno;
        ereport(elevel,
                (errcode_for_file_access(),
                 errmsg("could not fsync file \"%s\": %m",
                        tmppath)));
        return;
    }
    pgstat_report_wait_end();
 
    if (CloseTransientFile(fd) != 0)
    {
        int         save_errno = errno;
 
        unlink(tmppath);
        LWLockRelease(&slot->io_in_progress_lock);
 
        errno = save_errno;
        ereport(elevel,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m",
                        tmppath)));
        return;
    }
 
    /* rename to permanent file, fsync file and directory */
    if (rename(tmppath, path) != 0)
    {
        int         save_errno = errno;
 
        unlink(tmppath);
        LWLockRelease(&slot->io_in_progress_lock);
 
        errno = save_errno;
        ereport(elevel,
                (errcode_for_file_access(),
                 errmsg("could not rename file \"%s\" to \"%s\": %m",
                        tmppath, path)));
        return;
    }
 
    /*
     * Check CreateSlotOnDisk() for the reasoning of using a critical section.
     */
    START_CRIT_SECTION();
 
    fsync_fname(path, false);
    fsync_fname(dir, true);
    fsync_fname(PG_REPLSLOT_DIR, true);
 
    END_CRIT_SECTION();
 
    /*
     * Successfully wrote, unset dirty bit, unless somebody dirtied again
     * already and remember the confirmed_flush LSN value.
     */
    SpinLockAcquire(&slot->mutex);
    if (!slot->just_dirtied)
        slot->dirty = false;
    slot->last_saved_confirmed_flush = cp.slotdata.confirmed_flush;
    slot->last_saved_restart_lsn = cp.slotdata.restart_lsn;
    SpinLockRelease(&slot->mutex);
 
    LWLockRelease(&slot->io_in_progress_lock);
}

References CloseTransientFile(), COMP_CRC32C, ReplicationSlot::data, ReplicationSlot::dirty, END_CRIT_SECTION, ereport, errcode_for_file_access(), errmsg, fb(), fd(), FIN_CRC32C, fsync_fname(), INIT_CRC32C, ReplicationSlot::io_in_progress_lock, ReplicationSlot::just_dirtied, ReplicationSlot::last_saved_confirmed_flush, ReplicationSlot::last_saved_restart_lsn, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAXPGPATH, memcpy(), ReplicationSlot::mutex, OpenTransientFile(), PG_BINARY, pg_fsync(), PG_REPLSLOT_DIR, pgstat_report_wait_end(), pgstat_report_wait_start(), ReplicationSlotOnDiskChecksummedSize, ReplicationSlotOnDiskNotChecksummedSize, ReplicationSlotOnDiskV2Size, SLOT_MAGIC, SLOT_VERSION, SpinLockAcquire(), SpinLockRelease(), sprintf, START_CRIT_SECTION, and write.

Referenced by CheckPointReplicationSlots(), CreateSlotOnDisk(), and ReplicationSlotSave().

◆ SearchNamedReplicationSlot()

ReplicationSlot * SearchNamedReplicationSlot	(	const char *	name,
		bool	need_lock
	)

Definition at line 548 of file slot.c.

{
    int         i;
    ReplicationSlot *slot = NULL;
 
    if (need_lock)
        LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    for (i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
        if (s->in_use && strcmp(name, NameStr(s->data.name)) == 0)
        {
            slot = s;
            break;
        }
    }
 
    if (need_lock)
        LWLockRelease(ReplicationSlotControlLock);
 
    return slot;
}

References ReplicationSlot::data, fb(), i, ReplicationSlot::in_use, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, name, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotCtlData::replication_slots, and ReplicationSlotCtl.

Referenced by acquire_conflict_slot_if_exists(), get_replslot_index(), pg_ls_replslotdir(), pgstat_reset_replslot(), ReadReplicationSlot(), ReplicationSlotAcquire(), StandbySlotsHaveCaughtup(), and synchronize_one_slot().

◆ SlotExistsInSyncStandbySlots()

bool SlotExistsInSyncStandbySlots ( const char * slot_name )

Definition at line 3076 of file slot.c.

{
    const char *standby_slot_name;
 
    /* Return false if there is no value in synchronized_standby_slots */
    if (synchronized_standby_slots_config == NULL)
        return false;
 
    /*
     * XXX: We are not expecting this list to be long so a linear search
     * shouldn't hurt but if that turns out not to be true then we can cache
     * this information for each WalSender as well.
     */
    standby_slot_name = synchronized_standby_slots_config->slot_names;
    for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
    {
        if (strcmp(standby_slot_name, slot_name) == 0)
            return true;
 
        standby_slot_name += strlen(standby_slot_name) + 1;
    }
 
    return false;
}

References fb(), i, SyncStandbySlotsConfigData::nslotnames, SyncStandbySlotsConfigData::slot_names, and synchronized_standby_slots_config.

Referenced by PhysicalWakeupLogicalWalSnd().

◆ StandbySlotsHaveCaughtup()

bool StandbySlotsHaveCaughtup	(	XLogRecPtr	wait_for_lsn,
		int	elevel
	)

Definition at line 3109 of file slot.c.

{
    const char *name;
    int         caught_up_slot_num = 0;
    XLogRecPtr  min_restart_lsn = InvalidXLogRecPtr;
 
    /*
     * Don't need to wait for the standbys to catch up if there is no value in
     * synchronized_standby_slots.
     */
    if (synchronized_standby_slots_config == NULL)
        return true;
 
    /*
     * Don't need to wait for the standbys to catch up if we are on a standby
     * server, since we do not support syncing slots to cascading standbys.
     */
    if (RecoveryInProgress())
        return true;
 
    /*
     * Don't need to wait for the standbys to catch up if they are already
     * beyond the specified WAL location.
     */
    if (XLogRecPtrIsValid(ss_oldest_flush_lsn) &&
        ss_oldest_flush_lsn >= wait_for_lsn)
        return true;
 
    /*
     * To prevent concurrent slot dropping and creation while filtering the
     * slots, take the ReplicationSlotControlLock outside of the loop.
     */
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    name = synchronized_standby_slots_config->slot_names;
    for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++)
    {
        XLogRecPtr  restart_lsn;
        bool        invalidated;
        bool        inactive;
        ReplicationSlot *slot;
 
        slot = SearchNamedReplicationSlot(name, false);
 
        /*
         * If a slot name provided in synchronized_standby_slots does not
         * exist, report a message and exit the loop.
         */
        if (!slot)
        {
            ereport(elevel,
                    errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                    errmsg("replication slot \"%s\" specified in parameter \"%s\" does not exist",
                           name, "synchronized_standby_slots"),
                    errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
                              name),
                    errhint("Create the replication slot \"%s\" or amend parameter \"%s\".",
                            name, "synchronized_standby_slots"));
            break;
        }
 
        /* Same as above: if a slot is not physical, exit the loop. */
        if (SlotIsLogical(slot))
        {
            ereport(elevel,
                    errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                    errmsg("cannot specify logical replication slot \"%s\" in parameter \"%s\"",
                           name, "synchronized_standby_slots"),
                    errdetail("Logical replication is waiting for correction on replication slot \"%s\".",
                              name),
                    errhint("Remove the logical replication slot \"%s\" from parameter \"%s\".",
                            name, "synchronized_standby_slots"));
            break;
        }
 
        SpinLockAcquire(&slot->mutex);
        restart_lsn = slot->data.restart_lsn;
        invalidated = slot->data.invalidated != RS_INVAL_NONE;
        inactive = slot->active_proc == INVALID_PROC_NUMBER;
        SpinLockRelease(&slot->mutex);
 
        if (invalidated)
        {
            /* Specified physical slot has been invalidated */
            ereport(elevel,
                    errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                    errmsg("physical replication slot \"%s\" specified in parameter \"%s\" has been invalidated",
                           name, "synchronized_standby_slots"),
                    errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
                              name),
                    errhint("Drop and recreate the replication slot \"%s\", or amend parameter \"%s\".",
                            name, "synchronized_standby_slots"));
            break;
        }
 
        if (!XLogRecPtrIsValid(restart_lsn) || restart_lsn < wait_for_lsn)
        {
            /* Log a message if no active_pid for this physical slot */
            if (inactive)
                ereport(elevel,
                        errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                        errmsg("replication slot \"%s\" specified in parameter \"%s\" does not have active_pid",
                               name, "synchronized_standby_slots"),
                        errdetail("Logical replication is waiting on the standby associated with replication slot \"%s\".",
                                  name),
                        errhint("Start the standby associated with the replication slot \"%s\", or amend parameter \"%s\".",
                                name, "synchronized_standby_slots"));
 
            /* Continue if the current slot hasn't caught up. */
            break;
        }
 
        Assert(restart_lsn >= wait_for_lsn);
 
        if (!XLogRecPtrIsValid(min_restart_lsn) ||
            min_restart_lsn > restart_lsn)
            min_restart_lsn = restart_lsn;
 
        caught_up_slot_num++;
 
        name += strlen(name) + 1;
    }
 
    LWLockRelease(ReplicationSlotControlLock);
 
    /*
     * Return false if not all the standbys have caught up to the specified
     * WAL location.
     */
    if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames)
        return false;
 
    /* The ss_oldest_flush_lsn must not retreat. */
    Assert(!XLogRecPtrIsValid(ss_oldest_flush_lsn) ||
           min_restart_lsn >= ss_oldest_flush_lsn);
 
    ss_oldest_flush_lsn = min_restart_lsn;
 
    return true;
}

References ReplicationSlot::active_proc, Assert, ReplicationSlot::data, ereport, errcode(), errdetail(), errhint(), errmsg, fb(), i, INVALID_PROC_NUMBER, ReplicationSlotPersistentData::invalidated, InvalidXLogRecPtr, LW_SHARED, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, name, SyncStandbySlotsConfigData::nslotnames, RecoveryInProgress(), ReplicationSlotPersistentData::restart_lsn, RS_INVAL_NONE, SearchNamedReplicationSlot(), SyncStandbySlotsConfigData::slot_names, SlotIsLogical, SpinLockAcquire(), SpinLockRelease(), ss_oldest_flush_lsn, synchronized_standby_slots_config, and XLogRecPtrIsValid.

Referenced by NeedToWaitForStandbys(), and WaitForStandbyConfirmation().

◆ StartupReplicationSlots()

void StartupReplicationSlots ( void )

Definition at line 2398 of file slot.c.

{
    DIR        *replication_dir;
    struct dirent *replication_de;
 
    elog(DEBUG1, "starting up replication slots");
 
    /* restore all slots by iterating over all on-disk entries */
    replication_dir = AllocateDir(PG_REPLSLOT_DIR);
    while ((replication_de = ReadDir(replication_dir, PG_REPLSLOT_DIR)) != NULL)
    {
        char        path[MAXPGPATH + sizeof(PG_REPLSLOT_DIR)];
        PGFileType  de_type;
 
        if (strcmp(replication_de->d_name, ".") == 0 ||
            strcmp(replication_de->d_name, "..") == 0)
            continue;
 
        snprintf(path, sizeof(path), "%s/%s", PG_REPLSLOT_DIR, replication_de->d_name);
        de_type = get_dirent_type(path, replication_de, false, DEBUG1);
 
        /* we're only creating directories here, skip if it's not our's */
        if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_DIR)
            continue;
 
        /* we crashed while a slot was being setup or deleted, clean up */
        if (pg_str_endswith(replication_de->d_name, ".tmp"))
        {
            if (!rmtree(path, true))
            {
                ereport(WARNING,
                        (errmsg("could not remove directory \"%s\"",
                                path)));
                continue;
            }
            fsync_fname(PG_REPLSLOT_DIR, true);
            continue;
        }
 
        /* looks like a slot in a normal state, restore */
        RestoreSlotFromDisk(replication_de->d_name);
    }
    FreeDir(replication_dir);
 
    /* currently no slots exist, we're done. */
    if (max_replication_slots + max_repack_replication_slots <= 0)
        return;
 
    /* Now that we have recovered all the data, compute replication xmin */
    ReplicationSlotsComputeRequiredXmin(false);
    ReplicationSlotsComputeRequiredLSN();
}

References AllocateDir(), DEBUG1, elog, ereport, errmsg, fb(), FreeDir(), fsync_fname(), get_dirent_type(), max_repack_replication_slots, max_replication_slots, MAXPGPATH, PG_REPLSLOT_DIR, pg_str_endswith(), PGFILETYPE_DIR, PGFILETYPE_ERROR, ReadDir(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), RestoreSlotFromDisk(), rmtree(), snprintf, and WARNING.

Referenced by StartupXLOG().

◆ StaticAssertDecl()

StaticAssertDecl	(	lengthof(SlotInvalidationCauses)	= `=(RS_INVAL_MAX_CAUSES+1)`,
		"array length mismatch"
	)

◆ validate_sync_standby_slots()

static bool validate_sync_standby_slots	(	char *	rawname,
		List **	elemlist
	)

static

Definition at line 2971 of file slot.c.

{
    /* Verify syntax and parse string into a list of identifiers */
    if (!SplitIdentifierString(rawname, ',', elemlist))
    {
        GUC_check_errdetail("List syntax is invalid.");
        return false;
    }
 
    /* Iterate the list to validate each slot name */
    foreach_ptr(char, name, *elemlist)
    {
        int         err_code;
        char       *err_msg = NULL;
        char       *err_hint = NULL;
 
        if (!ReplicationSlotValidateNameInternal(name, false, &err_code,
                                                 &err_msg, &err_hint))
        {
            GUC_check_errcode(err_code);
            GUC_check_errdetail("%s", err_msg);
            if (err_hint != NULL)
                GUC_check_errhint("%s", err_hint);
            return false;
        }
    }
 
    return true;
}

References fb(), foreach_ptr, GUC_check_errcode(), GUC_check_errdetail, GUC_check_errhint, name, ReplicationSlotValidateNameInternal(), and SplitIdentifierString().

Referenced by check_synchronized_standby_slots().

◆ WaitForStandbyConfirmation()

void WaitForStandbyConfirmation ( XLogRecPtr wait_for_lsn )

Definition at line 3257 of file slot.c.

{
    /*
     * Don't need to wait for the standby to catch up if the current acquired
     * slot is not a logical failover slot, or there is no value in
     * synchronized_standby_slots.
     */
    if (!MyReplicationSlot->data.failover || !synchronized_standby_slots_config)
        return;
 
    ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv);
 
    for (;;)
    {
        CHECK_FOR_INTERRUPTS();
 
        if (ConfigReloadPending)
        {
            ConfigReloadPending = false;
            ProcessConfigFile(PGC_SIGHUP);
        }
 
        /* Exit if done waiting for every slot. */
        if (StandbySlotsHaveCaughtup(wait_for_lsn, WARNING))
            break;
 
        /*
         * Wait for the slots in the synchronized_standby_slots to catch up,
         * but use a timeout (1s) so we can also check if the
         * synchronized_standby_slots has been changed.
         */
        ConditionVariableTimedSleep(&WalSndCtl->wal_confirm_rcv_cv, 1000,
                                    WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION);
    }
 
    ConditionVariableCancelSleep();
}

References CHECK_FOR_INTERRUPTS, ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableTimedSleep(), ConfigReloadPending, ReplicationSlot::data, ReplicationSlotPersistentData::failover, fb(), MyReplicationSlot, PGC_SIGHUP, ProcessConfigFile(), StandbySlotsHaveCaughtup(), synchronized_standby_slots_config, WalSndCtlData::wal_confirm_rcv_cv, WalSndCtl, and WARNING.

Referenced by LogicalSlotAdvanceAndCheckSnapState(), and pg_logical_slot_get_changes_guts().

Variable Documentation

◆ idle_replication_slot_timeout_secs

int idle_replication_slot_timeout_secs = 0

Definition at line 170 of file slot.c.

Referenced by CanInvalidateIdleSlot(), DetermineSlotInvalidationCause(), and ReportSlotInvalidation().

◆ max_repack_replication_slots

◆ max_replication_slots

◆ MyReplicationSlot

ReplicationSlot* MyReplicationSlot = NULL

Definition at line 158 of file slot.c.

Referenced by ApplyLauncherMain(), binary_upgrade_check_logical_slot_pending_wal(), compute_min_nonremovable_xid(), copy_replication_slot(), create_logical_replication_slot(), create_physical_replication_slot(), CreateConflictDetectionSlot(), CreateDecodingContext(), CreateInitDecodingContext(), CreateReplicationSlot(), DisableLogicalDecodingIfNecessary(), EnsureLogicalDecodingEnabled(), init_conflict_slot_xmin(), InvalidatePossiblyObsoleteSlot(), LogicalConfirmReceivedLocation(), LogicalIncreaseRestartDecodingForSlot(), LogicalIncreaseXminForSlot(), logicalrep_worker_launch(), LogicalReplicationSlotCheckPendingWal(), LogicalSlotAdvanceAndCheckSnapState(), NeedToWaitForStandbys(), pg_create_logical_replication_slot(), pg_create_physical_replication_slot(), pg_logical_slot_get_changes_guts(), pg_physical_replication_slot_advance(), pg_replication_slot_advance(), PhysicalConfirmReceivedLocation(), PhysicalReplicationSlotNewXmin(), PhysicalWakeupLogicalWalSnd(), PostgresMain(), ProcessStandbyHSFeedbackMessage(), ProcessStandbyReplyMessage(), ReorderBufferAllocate(), ReorderBufferFree(), ReorderBufferRestoreChanges(), ReorderBufferRestoreCleanup(), ReorderBufferSerializedPath(), ReorderBufferSerializeTXN(), ReplicationSlotAcquire(), ReplicationSlotAlter(), ReplicationSlotCleanup(), ReplicationSlotCreate(), ReplicationSlotDrop(), ReplicationSlotDropAcquired(), ReplicationSlotMarkDirty(), ReplicationSlotPersist(), ReplicationSlotRelease(), ReplicationSlotReserveWal(), ReplicationSlotSave(), ReplicationSlotsDropDBSlots(), ReplicationSlotShmemExit(), reserve_wal_for_local_slot(), slotsync_failure_callback(), slotsync_worker_onexit(), StartLogicalReplication(), StartReplication(), StartupDecodingContext(), synchronize_one_slot(), update_and_persist_local_synced_slot(), update_conflict_slot_xmin(), update_local_synced_slot(), update_slotsync_skip_stats(), WaitForStandbyConfirmation(), and WalSndErrorCleanup().

◆ ReplicationSlotCtl

◆ ReplicationSlotsShmemCallbacks

const ShmemCallbacks ReplicationSlotsShmemCallbacks

Initial value:

= {
    .request_fn = ReplicationSlotsShmemRequest,
    .init_fn = ReplicationSlotsShmemInit,
}

Definition at line 152 of file slot.c.

                                                      {
    .request_fn = ReplicationSlotsShmemRequest,
    .init_fn = ReplicationSlotsShmemInit,
};

◆ SlotInvalidationCauses

const SlotInvalidationCauseMap SlotInvalidationCauses[]

static

Initial value:

= {
    {RS_INVAL_NONE, "none"},
    {RS_INVAL_WAL_REMOVED, "wal_removed"},
    {RS_INVAL_HORIZON, "rows_removed"},
    {RS_INVAL_WAL_LEVEL, "wal_level_insufficient"},
    {RS_INVAL_IDLE_TIMEOUT, "idle_timeout"},
}

Definition at line 115 of file slot.c.

                                                                 {
    {RS_INVAL_NONE, "none"},
    {RS_INVAL_WAL_REMOVED, "wal_removed"},
    {RS_INVAL_HORIZON, "rows_removed"},
    {RS_INVAL_WAL_LEVEL, "wal_level_insufficient"},
    {RS_INVAL_IDLE_TIMEOUT, "idle_timeout"},
};

Referenced by GetSlotInvalidationCause(), and GetSlotInvalidationCauseName().

◆ ss_oldest_flush_lsn

XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr

static

Definition at line 185 of file slot.c.

Referenced by assign_synchronized_standby_slots(), and StandbySlotsHaveCaughtup().

◆ synchronized_standby_slots

char* synchronized_standby_slots

Definition at line 176 of file slot.c.

◆ synchronized_standby_slots_config

SyncStandbySlotsConfigData* synchronized_standby_slots_config

static

Definition at line 179 of file slot.c.

Referenced by assign_synchronized_standby_slots(), SlotExistsInSyncStandbySlots(), StandbySlotsHaveCaughtup(), and WaitForStandbyConfirmation().

Variables
static const SlotInvalidationCauseMap	SlotInvalidationCauses []

ReplicationSlotCtlData *	ReplicationSlotCtl = NULL

const ShmemCallbacks	ReplicationSlotsShmemCallbacks

ReplicationSlot *	MyReplicationSlot = NULL

int	max_replication_slots = 10

int	max_repack_replication_slots = 5

int	idle_replication_slot_timeout_secs = 0

char *	synchronized_standby_slots

static SyncStandbySlotsConfigData *	synchronized_standby_slots_config

static XLogRecPtr	ss_oldest_flush_lsn = InvalidXLogRecPtr

Data Structures

Macros

Typedefs

Functions

Variables

Macro Definition Documentation

◆ ReplicationSlotOnDiskChecksummedSize

◆ ReplicationSlotOnDiskConstantSize

◆ ReplicationSlotOnDiskNotChecksummedSize

◆ ReplicationSlotOnDiskV2Size

◆ SLOT_MAGIC

◆ SLOT_VERSION

Typedef Documentation

◆ ReplicationSlotOnDisk

◆ SlotInvalidationCauseMap

Function Documentation

◆ assign_synchronized_standby_slots()

◆ CanInvalidateIdleSlot()

◆ check_synchronized_standby_slots()

◆ CheckLogicalSlotExists()

◆ CheckPointReplicationSlots()

◆ CheckSlotPermissions()

◆ CheckSlotRequirements()

◆ CreateSlotOnDisk()

◆ DetermineSlotInvalidationCause()

◆ GetSlotInvalidationCause()

◆ GetSlotInvalidationCauseName()

◆ InvalidateObsoleteReplicationSlots()

◆ InvalidatePossiblyObsoleteSlot()

◆ IsSlotForConflictCheck()

◆ ReplicationSlotAcquire()

◆ ReplicationSlotAlter()

◆ ReplicationSlotCleanup()

◆ ReplicationSlotCreate()

◆ ReplicationSlotDrop()

◆ ReplicationSlotDropAcquired()

◆ ReplicationSlotDropPtr()

◆ ReplicationSlotIndex()

◆ ReplicationSlotInitialize()

◆ ReplicationSlotMarkDirty()

◆ ReplicationSlotName()

◆ ReplicationSlotPersist()

◆ ReplicationSlotRelease()

◆ ReplicationSlotReserveWal()

◆ ReplicationSlotSave()

◆ ReplicationSlotsComputeLogicalRestartLSN()

◆ ReplicationSlotsComputeRequiredLSN()

◆ ReplicationSlotsComputeRequiredXmin()

◆ ReplicationSlotsCountDBSlots()

◆ ReplicationSlotsDropDBSlots()

◆ ReplicationSlotShmemExit()

◆ ReplicationSlotsShmemInit()

◆ ReplicationSlotsShmemRequest()

◆ ReplicationSlotValidateName()

◆ ReplicationSlotValidateNameInternal()

◆ ReportSlotInvalidation()

◆ RestoreSlotFromDisk()

◆ SaveSlotToPath()

◆ SearchNamedReplicationSlot()

◆ SlotExistsInSyncStandbySlots()

◆ StandbySlotsHaveCaughtup()

◆ StartupReplicationSlots()

◆ StaticAssertDecl()

◆ validate_sync_standby_slots()

◆ WaitForStandbyConfirmation()

Variable Documentation

◆ idle_replication_slot_timeout_secs

◆ max_repack_replication_slots

◆ max_replication_slots

◆ MyReplicationSlot

◆ ReplicationSlotCtl

◆ ReplicationSlotsShmemCallbacks

◆ SlotInvalidationCauses

◆ ss_oldest_flush_lsn

◆ synchronized_standby_slots

◆ synchronized_standby_slots_config