slotsync.c File Reference

#include "postgres.h"
#include <time.h>
#include "access/xlog_internal.h"
#include "access/xlogrecovery.h"
#include "catalog/pg_database.h"
#include "libpq/pqsignal.h"
#include "pgstat.h"
#include "postmaster/interrupt.h"
#include "replication/logical.h"
#include "replication/slotsync.h"
#include "replication/snapbuild.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/subsystems.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/ps_status.h"
#include "utils/timeout.h"
#include "utils/wait_event.h"

Include dependency graph for slotsync.c:

Go to the source code of this file.

Data Structures
struct	SlotSyncCtxStruct
struct	RemoteSlot

Macros
#define	MIN_SLOTSYNC_WORKER_NAPTIME_MS   200
#define	MAX_SLOTSYNC_WORKER_NAPTIME_MS   30000 /* 30s */
#define	SLOTSYNC_RESTART_INTERVAL_SEC   10
#define	SLOTSYNC_COLUMN_COUNT   10
#define	PRIMARY_INFO_OUTPUT_COL_COUNT   2

Typedefs
typedef struct SlotSyncCtxStruct	SlotSyncCtxStruct
typedef struct RemoteSlot	RemoteSlot

Functions
static void	SlotSyncShmemRequest (void *arg)
static void	SlotSyncShmemInit (void *arg)
static void	slotsync_failure_callback (int code, Datum arg)
static void	update_synced_slots_inactive_since (void)
static void	update_slotsync_skip_stats (SlotSyncSkipReason skip_reason)
static bool	update_local_synced_slot (RemoteSlot *remote_slot, Oid remote_dbid)
static List *	get_local_synced_slots (void)
static bool	local_sync_slot_required (ReplicationSlot *local_slot, List *remote_slots)
static void	drop_local_obsolete_slots (List *remote_slot_list)
static void	reserve_wal_for_local_slot (XLogRecPtr restart_lsn)
static bool	update_and_persist_local_synced_slot (RemoteSlot *remote_slot, Oid remote_dbid, bool *slot_persistence_pending)
static bool	synchronize_one_slot (RemoteSlot *remote_slot, Oid remote_dbid, bool *slot_persistence_pending)
static List *	fetch_remote_slots (WalReceiverConn *wrconn, List *slot_names)
static bool	synchronize_slots (WalReceiverConn *wrconn, List *remote_slot_list, bool *slot_persistence_pending)
static void	validate_remote_info (WalReceiverConn *wrconn)
char *	CheckAndGetDbnameFromConninfo (void)
bool	ValidateSlotSyncParams (int elevel)
static void	slotsync_reread_config (void)
void	HandleSlotSyncMessageInterrupt (void)
void	ProcessSlotSyncMessage (void)
static void	slotsync_worker_disconnect (int code, Datum arg)
static void	slotsync_worker_onexit (int code, Datum arg)
static void	wait_for_slot_activity (bool some_slot_updated)
static void	check_and_set_sync_info (pid_t sync_process_pid)
static void	reset_syncing_flag (void)
void	ReplSlotSyncWorkerMain (const void *startup_data, size_t startup_data_len)
void	ShutDownSlotSync (void)
bool	SlotSyncWorkerCanRestart (void)
bool	IsSyncingReplicationSlots (void)
static List *	extract_slot_names (List *remote_slots)
void	SyncReplicationSlots (WalReceiverConn *wrconn)

Variables
static SlotSyncCtxStruct *	SlotSyncCtx = NULL
const ShmemCallbacks	SlotSyncShmemCallbacks
bool	sync_replication_slots = false
static long	sleep_ms = MIN_SLOTSYNC_WORKER_NAPTIME_MS
static bool	syncing_slots = false
volatile sig_atomic_t	SlotSyncShutdownPending = false

Macro Definition Documentation

MAX_SLOTSYNC_WORKER_NAPTIME_MS

#define MAX_SLOTSYNC_WORKER_NAPTIME_MS   30000 /* 30s */

Definition at line 140 of file slotsync.c.

MIN_SLOTSYNC_WORKER_NAPTIME_MS

#define MIN_SLOTSYNC_WORKER_NAPTIME_MS   200

Definition at line 139 of file slotsync.c.

PRIMARY_INFO_OUTPUT_COL_COUNT

#define PRIMARY_INFO_OUTPUT_COL_COUNT   2

SLOTSYNC_COLUMN_COUNT

#define SLOTSYNC_COLUMN_COUNT   10

SLOTSYNC_RESTART_INTERVAL_SEC

#define SLOTSYNC_RESTART_INTERVAL_SEC   10

Definition at line 145 of file slotsync.c.

Typedef Documentation

RemoteSlot

typedef struct RemoteSlot RemoteSlot

SlotSyncCtxStruct

typedef struct SlotSyncCtxStruct SlotSyncCtxStruct

Function Documentation

check_and_set_sync_info()

static void check_and_set_sync_info ( pid_t  sync_process_pid )

static

Definition at line 1491 of file slotsync.c.

{
    SpinLockAcquire(&SlotSyncCtx->mutex);
 
    /*
     * Exit immediately if promotion has been triggered.  This guards against
     * a new worker (or a call to pg_sync_replication_slots()) that starts
     * after the old worker was stopped by ShutDownSlotSync().
     */
    if (SlotSyncCtx->stopSignaled)
    {
        SpinLockRelease(&SlotSyncCtx->mutex);
 
        if (AmLogicalSlotSyncWorkerProcess())
        {
            ereport(DEBUG1,
                    errmsg("replication slot synchronization worker will not start because promotion was triggered"));
 
            proc_exit(0);
        }
        else
        {
            /*
             * For the backend executing SQL function
             * pg_sync_replication_slots().
             */
            ereport(ERROR,
                    errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                    errmsg("replication slot synchronization will not start because promotion was triggered"));
        }
    }
 
    if (SlotSyncCtx->syncing)
    {
        SpinLockRelease(&SlotSyncCtx->mutex);
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("cannot synchronize replication slots concurrently"));
    }
 
    /* The pid must not be already assigned in SlotSyncCtx */
    Assert(SlotSyncCtx->pid == InvalidPid);
 
    SlotSyncCtx->syncing = true;
 
    /*
     * Advertise the required PID so that the startup process can kill the
     * slot sync process on promotion.
     */
    SlotSyncCtx->pid = sync_process_pid;
 
    SpinLockRelease(&SlotSyncCtx->mutex);
 
    syncing_slots = true;
}

References AmLogicalSlotSyncWorkerProcess, Assert, DEBUG1, ereport, errcode(), errmsg, ERROR, fb(), InvalidPid, SlotSyncCtxStruct::mutex, SlotSyncCtxStruct::pid, proc_exit(), SlotSyncCtx, SpinLockAcquire(), SpinLockRelease(), SlotSyncCtxStruct::stopSignaled, SlotSyncCtxStruct::syncing, and syncing_slots.

Referenced by ReplSlotSyncWorkerMain(), and SyncReplicationSlots().

CheckAndGetDbnameFromConninfo()

char * CheckAndGetDbnameFromConninfo

(

void

)

Definition at line 1170 of file slotsync.c.

{
    char       *dbname;
 
    /*
     * The slot synchronization needs a database connection for walrcv_exec to
     * work.
     */
    dbname = walrcv_get_dbname_from_conninfo(PrimaryConnInfo);
    if (dbname == NULL)
        ereport(ERROR,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 
        /*
         * translator: first %s is a connection option; second %s is a GUC
         * variable name
         */
                errmsg("replication slot synchronization requires \"%s\" to be specified in \"%s\"",
                       "dbname", "primary_conninfo"));
    return dbname;
}

References dbname, ereport, errcode(), errmsg, ERROR, fb(), PrimaryConnInfo, and walrcv_get_dbname_from_conninfo.

Referenced by pg_sync_replication_slots(), and ReplSlotSyncWorkerMain().

drop_local_obsolete_slots()

static void drop_local_obsolete_slots ( List *  remote_slot_list )

static

Definition at line 535 of file slotsync.c.

{
    List       *local_slots = get_local_synced_slots();
 
    foreach_ptr(ReplicationSlot, local_slot, local_slots)
    {
        /* Drop the local slot if it is not required to be retained. */
        if (!local_sync_slot_required(local_slot, remote_slot_list))
        {
            Oid         slot_database = local_slot->data.database;
            bool        synced_slot;
 
            /*
             * Use shared lock to prevent a conflict with
             * ReplicationSlotsDropDBSlots(), trying to drop the same slot
             * during a drop-database operation.
             */
            LockSharedObject(DatabaseRelationId, slot_database, 0,
                             AccessShareLock);
 
            /*
             * In the small window between getting the slot to drop and
             * locking the database, there is a possibility of a parallel
             * database drop by the startup process and the creation of a new
             * slot by the user. This new user-created slot may end up using
             * the same shared memory as that of 'local_slot'.
             *
             * Because local_slot still points to a reusable slot-array entry,
             * its fields (name, database OID, invalidation state) may already
             * describe such a replacement slot by the time we reach here.
             * That means the drop decision made by local_sync_slot_required()
             * above could have been based on the replacement slot's data, and
             * slot_database could refer to an unrelated database. The recheck
             * below keeps us from actually dropping a user-created
             * replacement slot; the residual risk is confined to this cycle
             * (for example, briefly locking an unrelated database) and is
             * acceptable because the race is rare and non-fatal.
             */
            SpinLockAcquire(&local_slot->mutex);
            synced_slot = local_slot->in_use && local_slot->data.synced;
            SpinLockRelease(&local_slot->mutex);
 
            if (synced_slot)
            {
                NameData    slot_name = local_slot->data.name;
 
                /*
                 * Now acquire and drop the slot.  Note we purposely don't
                 * request logical decoding to be disabled here: since this is
                 * a standby, which derives its logical decoding state from
                 * the primary, it would be wrong to do so.
                 */
                ReplicationSlotAcquire(NameStr(slot_name), true, false);
                ReplicationSlotDropAcquired(false);
 
                ereport(LOG,
                        errmsg("dropped replication slot \"%s\" of database with OID %u",
                               NameStr(slot_name),
                               slot_database));
            }
 
            UnlockSharedObject(DatabaseRelationId, slot_database, 0,
                               AccessShareLock);
        }
    }
}

References AccessShareLock, nameData::data, ereport, errmsg, fb(), foreach_ptr, get_local_synced_slots(), local_sync_slot_required(), LockSharedObject(), LOG, NameStr, ReplicationSlotAcquire(), ReplicationSlotDropAcquired(), SpinLockAcquire(), SpinLockRelease(), and UnlockSharedObject().

Referenced by synchronize_slots().

extract_slot_names()

static List * extract_slot_names ( List *  remote_slots )

static

Definition at line 1996 of file slotsync.c.

{
    List       *slot_names = NIL;
 
    foreach_ptr(RemoteSlot, remote_slot, remote_slots)
    {
        char       *slot_name;
 
        slot_name = pstrdup(remote_slot->name);
        slot_names = lappend(slot_names, slot_name);
    }
 
    return slot_names;
}

References fb(), foreach_ptr, lappend(), NIL, and pstrdup().

Referenced by SyncReplicationSlots().

fetch_remote_slots()

static List * fetch_remote_slots ( WalReceiverConn *  wrconn, List *  slot_names  )

static

Definition at line 913 of file slotsync.c.

{
#define SLOTSYNC_COLUMN_COUNT 10
    Oid         slotRow[SLOTSYNC_COLUMN_COUNT] = {TEXTOID, TEXTOID, LSNOID,
    LSNOID, XIDOID, BOOLOID, LSNOID, BOOLOID, TEXTOID, TEXTOID};
 
    WalRcvExecResult *res;
    TupleTableSlot *tupslot;
    List       *remote_slot_list = NIL;
    StringInfoData query;
 
    initStringInfo(&query);
    appendStringInfoString(&query,
                           "SELECT slot_name, plugin, confirmed_flush_lsn,"
                           " restart_lsn, catalog_xmin, two_phase,"
                           " two_phase_at, failover,"
                           " database, invalidation_reason"
                           " FROM pg_catalog.pg_replication_slots"
                           " WHERE failover and NOT temporary");
 
    if (slot_names != NIL)
    {
        bool        first_slot = true;
 
        /*
         * Construct the query to fetch only the specified slots
         */
        appendStringInfoString(&query, " AND slot_name IN (");
 
        foreach_ptr(char, slot_name, slot_names)
        {
            if (!first_slot)
                appendStringInfoString(&query, ", ");
 
            appendStringInfoString(&query, quote_literal_cstr(slot_name));
            first_slot = false;
        }
        appendStringInfoChar(&query, ')');
    }
 
    /* Execute the query */
    res = walrcv_exec(wrconn, query.data, SLOTSYNC_COLUMN_COUNT, slotRow);
    pfree(query.data);
    if (res->status != WALRCV_OK_TUPLES)
        ereport(ERROR,
                errmsg("could not fetch failover logical slots info from the primary server: %s",
                       res->err));
 
    tupslot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
    while (tuplestore_gettupleslot(res->tuplestore, true, false, tupslot))
    {
        bool        isnull;
        RemoteSlot *remote_slot = palloc0_object(RemoteSlot);
        Datum       d;
        int         col = 0;
 
        remote_slot->name = TextDatumGetCString(slot_getattr(tupslot, ++col,
                                                             &isnull));
        Assert(!isnull);
 
        remote_slot->plugin = TextDatumGetCString(slot_getattr(tupslot, ++col,
                                                               &isnull));
        Assert(!isnull);
 
        /*
         * It is possible to get null values for LSN and Xmin if slot is
         * invalidated on the primary server, so handle accordingly.
         */
        d = slot_getattr(tupslot, ++col, &isnull);
        remote_slot->confirmed_lsn = isnull ? InvalidXLogRecPtr :
            DatumGetLSN(d);
 
        d = slot_getattr(tupslot, ++col, &isnull);
        remote_slot->restart_lsn = isnull ? InvalidXLogRecPtr : DatumGetLSN(d);
 
        d = slot_getattr(tupslot, ++col, &isnull);
        remote_slot->catalog_xmin = isnull ? InvalidTransactionId :
            DatumGetTransactionId(d);
 
        remote_slot->two_phase = DatumGetBool(slot_getattr(tupslot, ++col,
                                                           &isnull));
        Assert(!isnull);
 
        d = slot_getattr(tupslot, ++col, &isnull);
        remote_slot->two_phase_at = isnull ? InvalidXLogRecPtr : DatumGetLSN(d);
 
        remote_slot->failover = DatumGetBool(slot_getattr(tupslot, ++col,
                                                          &isnull));
        Assert(!isnull);
 
        remote_slot->database = TextDatumGetCString(slot_getattr(tupslot,
                                                                 ++col, &isnull));
        Assert(!isnull);
 
        d = slot_getattr(tupslot, ++col, &isnull);
        remote_slot->invalidated = isnull ? RS_INVAL_NONE :
            GetSlotInvalidationCause(TextDatumGetCString(d));
 
        /* Sanity check */
        Assert(col == SLOTSYNC_COLUMN_COUNT);
 
        /*
         * If restart_lsn, confirmed_lsn or catalog_xmin is invalid but the
         * slot is valid, that means we have fetched the remote_slot in its
         * RS_EPHEMERAL state. In such a case, don't sync it; we can always
         * sync it in the next sync cycle when the remote_slot is persisted
         * and has valid lsn(s) and xmin values.
         *
         * XXX: In future, if we plan to expose 'slot->data.persistency' in
         * pg_replication_slots view, then we can avoid fetching RS_EPHEMERAL
         * slots in the first place.
         */
        if ((!XLogRecPtrIsValid(remote_slot->restart_lsn) ||
             !XLogRecPtrIsValid(remote_slot->confirmed_lsn) ||
             !TransactionIdIsValid(remote_slot->catalog_xmin)) &&
            remote_slot->invalidated == RS_INVAL_NONE)
            pfree(remote_slot);
        else
            /* Create list of remote slots */
            remote_slot_list = lappend(remote_slot_list, remote_slot);
 
        ExecClearTuple(tupslot);
    }
 
    ExecDropSingleTupleTableSlot(tupslot);
    walrcv_clear_result(res);
 
    return remote_slot_list;
}

References appendStringInfoChar(), appendStringInfoString(), Assert, StringInfoData::data, DatumGetBool(), DatumGetLSN(), DatumGetTransactionId(), ereport, WalRcvExecResult::err, errmsg, ERROR, ExecClearTuple(), ExecDropSingleTupleTableSlot(), fb(), foreach_ptr, GetSlotInvalidationCause(), initStringInfo(), InvalidTransactionId, InvalidXLogRecPtr, lappend(), MakeSingleTupleTableSlot(), NIL, palloc0_object, pfree(), quote_literal_cstr(), RS_INVAL_NONE, slot_getattr(), SLOTSYNC_COLUMN_COUNT, WalRcvExecResult::status, TextDatumGetCString, TransactionIdIsValid, TTSOpsMinimalTuple, WalRcvExecResult::tupledesc, WalRcvExecResult::tuplestore, tuplestore_gettupleslot(), walrcv_clear_result(), walrcv_exec, WALRCV_OK_TUPLES, wrconn, and XLogRecPtrIsValid.

Referenced by ReplSlotSyncWorkerMain(), and SyncReplicationSlots().

get_local_synced_slots()

static List * get_local_synced_slots ( void  )

static

Definition at line 451 of file slotsync.c.

{
    List       *local_slots = NIL;
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
        /* Check if it is a synchronized slot */
        if (s->in_use && s->data.synced)
        {
            Assert(SlotIsLogical(s));
            local_slots = lappend(local_slots, s);
        }
    }
 
    LWLockRelease(ReplicationSlotControlLock);
 
    return local_slots;
}

References Assert, ReplicationSlot::data, fb(), i, ReplicationSlot::in_use, lappend(), LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, NIL, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, SlotIsLogical, and ReplicationSlotPersistentData::synced.

Referenced by drop_local_obsolete_slots().

HandleSlotSyncMessageInterrupt()

void HandleSlotSyncMessageInterrupt

(

void

)

Definition at line 1350 of file slotsync.c.

{
    InterruptPending = true;
    SlotSyncShutdownPending = true;
    /* latch will be set by procsignal_sigusr1_handler */
}

References InterruptPending, and SlotSyncShutdownPending.

Referenced by procsignal_sigusr1_handler().

IsSyncingReplicationSlots()

bool IsSyncingReplicationSlots

(

void

)

Definition at line 1928 of file slotsync.c.

{
    return syncing_slots;
}

References syncing_slots.

Referenced by CreateDecodingContext(), GetStandbyFlushRecPtr(), ProcessSlotSyncMessage(), and ReplicationSlotCreate().

local_sync_slot_required()

static bool local_sync_slot_required ( ReplicationSlot *  local_slot, List *  remote_slots  )

static

Definition at line 482 of file slotsync.c.

{
    bool        remote_exists = false;
    bool        locally_invalidated = false;
 
    foreach_ptr(RemoteSlot, remote_slot, remote_slots)
    {
        if (strcmp(remote_slot->name, NameStr(local_slot->data.name)) == 0)
        {
            remote_exists = true;
 
            /*
             * If remote slot is not invalidated but local slot is marked as
             * invalidated, then set locally_invalidated flag.
             */
            SpinLockAcquire(&local_slot->mutex);
            locally_invalidated =
                (remote_slot->invalidated == RS_INVAL_NONE) &&
                (local_slot->data.invalidated != RS_INVAL_NONE);
            SpinLockRelease(&local_slot->mutex);
 
            break;
        }
    }
 
    return (remote_exists && !locally_invalidated);
}

References fb(), foreach_ptr, NameStr, RS_INVAL_NONE, SpinLockAcquire(), and SpinLockRelease().

Referenced by drop_local_obsolete_slots().

ProcessSlotSyncMessage()

void ProcessSlotSyncMessage

(

void

)

Definition at line 1366 of file slotsync.c.

{
    SlotSyncShutdownPending = false;
 
    if (AmLogicalSlotSyncWorkerProcess())
    {
        ereport(LOG,
                errmsg("replication slot synchronization worker will stop because promotion is triggered"));
        proc_exit(0);
    }
    else
    {
        /*
         * If sync has already completed, there is no need to interrupt the
         * caller with an error.
         */
        if (!IsSyncingReplicationSlots())
            return;
 
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("replication slot synchronization will stop because promotion is triggered"));
    }
}

References AmLogicalSlotSyncWorkerProcess, ereport, errcode(), errmsg, ERROR, fb(), IsSyncingReplicationSlots(), LOG, proc_exit(), and SlotSyncShutdownPending.

Referenced by ProcessInterrupts().

ReplSlotSyncWorkerMain()

void ReplSlotSyncWorkerMain	(	const void *	startup_data,
		size_t	startup_data_len
	)

Definition at line 1571 of file slotsync.c.

{
    WalReceiverConn *wrconn = NULL;
    char       *dbname;
    char       *err;
    sigjmp_buf  local_sigjmp_buf;
    StringInfoData app_name;
 
    Assert(startup_data_len == 0);
 
    /* Release postmaster's working memory context */
    if (PostmasterContext)
    {
        MemoryContextDelete(PostmasterContext);
        PostmasterContext = NULL;
    }
 
    init_ps_display(NULL);
 
    Assert(GetProcessingMode() == InitProcessing);
 
    /*
     * Create a per-backend PGPROC struct in shared memory.  We must do this
     * before we access any shared memory.
     */
    InitProcess();
 
    /*
     * Early initialization.
     */
    BaseInit();
 
    Assert(SlotSyncCtx != NULL);
 
    /*
     * If an exception is encountered, processing resumes here.
     *
     * We just need to clean up, report the error, and go away.
     *
     * If we do not have this handling here, then since this worker process
     * operates at the bottom of the exception stack, ERRORs turn into FATALs.
     * Therefore, we create our own exception handler to catch ERRORs.
     */
    if (sigsetjmp(local_sigjmp_buf, 1) != 0)
    {
        /* since not using PG_TRY, must reset error stack by hand */
        error_context_stack = NULL;
 
        /* Prevents interrupts while cleaning up */
        HOLD_INTERRUPTS();
 
        /* Report the error to the server log */
        EmitErrorReport();
 
        /*
         * We can now go away.  Note that because we called InitProcess, a
         * callback was registered to do ProcKill, which will clean up
         * necessary state.
         */
        proc_exit(0);
    }
 
    /* We can now handle ereport(ERROR) */
    PG_exception_stack = &local_sigjmp_buf;
 
    /* Setup signal handling */
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    pqsignal(SIGINT, StatementCancelHandler);
    pqsignal(SIGTERM, die);
    pqsignal(SIGFPE, FloatExceptionHandler);
    pqsignal(SIGUSR1, procsignal_sigusr1_handler);
    pqsignal(SIGUSR2, PG_SIG_IGN);
    pqsignal(SIGPIPE, PG_SIG_IGN);
    pqsignal(SIGCHLD, PG_SIG_DFL);
 
    check_and_set_sync_info(MyProcPid);
 
    ereport(LOG, errmsg("slot sync worker started"));
 
    /* Register it as soon as SlotSyncCtx->pid is initialized. */
    before_shmem_exit(slotsync_worker_onexit, (Datum) 0);
 
    /*
     * Establishes SIGALRM handler and initialize timeout module. It is needed
     * by InitPostgres to register different timeouts.
     */
    InitializeTimeouts();
 
    /* Load the libpq-specific functions */
    load_file("libpqwalreceiver", false);
 
    /*
     * Unblock signals (they were blocked when the postmaster forked us)
     */
    sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
 
    /*
     * Set always-secure search path, so malicious users can't redirect user
     * code (e.g. operators).
     *
     * It's not strictly necessary since we won't be scanning or writing to
     * any user table locally, but it's good to retain it here for added
     * precaution.
     */
    SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
 
    dbname = CheckAndGetDbnameFromConninfo();
 
    /*
     * Connect to the database specified by the user in primary_conninfo. We
     * need a database connection for walrcv_exec to work which we use to
     * fetch slot information from the remote node. See comments atop
     * libpqrcv_exec.
     *
     * We do not specify a specific user here since the slot sync worker will
     * operate as a superuser. This is safe because the slot sync worker does
     * not interact with user tables, eliminating the risk of executing
     * arbitrary code within triggers.
     */
    InitPostgres(dbname, InvalidOid, NULL, InvalidOid, 0, NULL);
 
    SetProcessingMode(NormalProcessing);
 
    initStringInfo(&app_name);
    if (cluster_name[0])
        appendStringInfo(&app_name, "%s_%s", cluster_name, "slotsync worker");
    else
        appendStringInfoString(&app_name, "slotsync worker");
 
    /*
     * Establish the connection to the primary server for slot
     * synchronization.
     */
    wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
                            app_name.data, &err);
 
    if (!wrconn)
        ereport(ERROR,
                errcode(ERRCODE_CONNECTION_FAILURE),
                errmsg("synchronization worker \"%s\" could not connect to the primary server: %s",
                       app_name.data, err));
 
    pfree(app_name.data);
 
    /*
     * Register the disconnection callback.
     *
     * XXX: This can be combined with previous cleanup registration of
     * slotsync_worker_onexit() but that will need the connection to be made
     * global and we want to avoid introducing global for this purpose.
     */
    before_shmem_exit(slotsync_worker_disconnect, PointerGetDatum(wrconn));
 
    /*
     * Using the specified primary server connection, check that we are not a
     * cascading standby and slot configured in 'primary_slot_name' exists on
     * the primary server.
     */
    validate_remote_info(wrconn);
 
    /* Main loop to synchronize slots */
    for (;;)
    {
        bool        some_slot_updated = false;
        bool        started_tx = false;
        List       *remote_slots;
 
        CHECK_FOR_INTERRUPTS();
 
        if (ConfigReloadPending)
            slotsync_reread_config();
 
        /*
         * The syscache access in fetch_remote_slots() needs a transaction
         * env.
         */
        if (!IsTransactionState())
        {
            StartTransactionCommand();
            started_tx = true;
        }
 
        remote_slots = fetch_remote_slots(wrconn, NIL);
        some_slot_updated = synchronize_slots(wrconn, remote_slots, NULL);
        list_free_deep(remote_slots);
 
        if (started_tx)
            CommitTransactionCommand();
 
        wait_for_slot_activity(some_slot_updated);
    }
 
    /*
     * The slot sync worker can't get here because it will only stop when it
     * receives a stop request from the startup process, or when there is an
     * error.
     */
    Assert(false);
}

References appendStringInfo(), appendStringInfoString(), Assert, BaseInit(), before_shmem_exit(), check_and_set_sync_info(), CHECK_FOR_INTERRUPTS, CheckAndGetDbnameFromConninfo(), cluster_name, CommitTransactionCommand(), ConfigReloadPending, dbname, die, EmitErrorReport(), ereport, err(), errcode(), errmsg, ERROR, error_context_stack, fb(), fetch_remote_slots(), FloatExceptionHandler(), GetProcessingMode, HOLD_INTERRUPTS, init_ps_display(), InitializeTimeouts(), InitPostgres(), InitProcess(), InitProcessing, initStringInfo(), InvalidOid, IsTransactionState(), list_free_deep(), load_file(), LOG, MemoryContextDelete(), MyProcPid, NIL, NormalProcessing, pfree(), PG_exception_stack, PG_SIG_DFL, PG_SIG_IGN, PGC_S_OVERRIDE, PGC_SUSET, PointerGetDatum, PostmasterContext, pqsignal, PrimaryConnInfo, proc_exit(), procsignal_sigusr1_handler(), SetConfigOption(), SetProcessingMode, SIGCHLD, SIGHUP, SignalHandlerForConfigReload(), SIGPIPE, SIGUSR1, SIGUSR2, slotsync_reread_config(), slotsync_worker_disconnect(), slotsync_worker_onexit(), SlotSyncCtx, StartTransactionCommand(), StatementCancelHandler(), synchronize_slots(), UnBlockSig, validate_remote_info(), wait_for_slot_activity(), walrcv_connect, and wrconn.

reserve_wal_for_local_slot()

static void reserve_wal_for_local_slot ( XLogRecPtr  restart_lsn )

static

Definition at line 610 of file slotsync.c.

{
    XLogRecPtr  slot_min_lsn;
    XLogRecPtr  min_safe_lsn;
    XLogSegNo   segno;
    ReplicationSlot *slot = MyReplicationSlot;
 
    Assert(slot != NULL);
    Assert(!XLogRecPtrIsValid(slot->data.restart_lsn));
 
    /*
     * Acquire an exclusive lock to prevent the checkpoint process from
     * concurrently calculating the minimum slot LSN (see
     * CheckPointReplicationSlots), ensuring that if WAL reservation occurs
     * first, the checkpoint must wait for the restart_lsn update before
     * calculating the minimum LSN.
     *
     * Note: Unlike ReplicationSlotReserveWal(), this lock does not protect a
     * newly synced slot from being invalidated if a concurrent checkpoint has
     * invoked CheckPointReplicationSlots() before the WAL reservation here.
     * This can happen because the initial restart_lsn received from the
     * remote server can precede the redo pointer. Therefore, when selecting
     * the initial restart_lsn, we consider using the redo pointer or the
     * minimum slot LSN (if those values are greater than the remote
     * restart_lsn) instead of relying solely on the remote value.
     */
    LWLockAcquire(ReplicationSlotAllocationLock, LW_EXCLUSIVE);
 
    /*
     * Determine the minimum non-removable LSN by comparing the redo pointer
     * with the minimum slot LSN.
     *
     * The minimum slot LSN is considered because the redo pointer advances at
     * every checkpoint, even when replication slots are present on the
     * standby. In such scenarios, the redo pointer can exceed the remote
     * restart_lsn, while WALs preceding the remote restart_lsn remain
     * protected by a local replication slot.
     */
    min_safe_lsn = GetRedoRecPtr();
    slot_min_lsn = XLogGetReplicationSlotMinimumLSN();
 
    if (XLogRecPtrIsValid(slot_min_lsn) && min_safe_lsn > slot_min_lsn)
        min_safe_lsn = slot_min_lsn;
 
    /*
     * If the minimum safe LSN is greater than the given restart_lsn, use it
     * as the initial restart_lsn for the newly synced slot. Otherwise, use
     * the given remote restart_lsn.
     */
    SpinLockAcquire(&slot->mutex);
    slot->data.restart_lsn = Max(restart_lsn, min_safe_lsn);
    SpinLockRelease(&slot->mutex);
 
    ReplicationSlotsComputeRequiredLSN();
 
    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);
}

References Assert, ReplicationSlot::data, elog, ERROR, fb(), GetRedoRecPtr(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), Max, ReplicationSlot::mutex, MyReplicationSlot, ReplicationSlotPersistentData::name, NameStr, ReplicationSlotsComputeRequiredLSN(), ReplicationSlotPersistentData::restart_lsn, SpinLockAcquire(), SpinLockRelease(), wal_segment_size, XLByteToSeg, XLogGetLastRemovedSegno(), XLogGetReplicationSlotMinimumLSN(), and XLogRecPtrIsValid.

Referenced by synchronize_one_slot().

reset_syncing_flag()

static void reset_syncing_flag ( void  )

static

Definition at line 1551 of file slotsync.c.

{
    SpinLockAcquire(&SlotSyncCtx->mutex);
    SlotSyncCtx->syncing = false;
    SlotSyncCtx->pid = InvalidPid;
    SpinLockRelease(&SlotSyncCtx->mutex);
 
    syncing_slots = false;
}

References InvalidPid, SlotSyncCtxStruct::mutex, SlotSyncCtxStruct::pid, SlotSyncCtx, SpinLockAcquire(), SpinLockRelease(), SlotSyncCtxStruct::syncing, and syncing_slots.

Referenced by slotsync_failure_callback(), and SyncReplicationSlots().

ShutDownSlotSync()

void ShutDownSlotSync

(

void

)

Definition at line 1830 of file slotsync.c.

{
    pid_t       sync_process_pid;
 
    SpinLockAcquire(&SlotSyncCtx->mutex);
 
    SlotSyncCtx->stopSignaled = true;
 
    /*
     * Return if neither the slot sync worker is running nor the function
     * pg_sync_replication_slots() is executing.
     */
    if (!SlotSyncCtx->syncing)
    {
        SpinLockRelease(&SlotSyncCtx->mutex);
        update_synced_slots_inactive_since();
        return;
    }
 
    sync_process_pid = SlotSyncCtx->pid;
 
    SpinLockRelease(&SlotSyncCtx->mutex);
 
    /*
     * Signal process doing slotsync, if any, asking it to stop.
     */
    if (sync_process_pid != InvalidPid)
        SendProcSignal(sync_process_pid, PROCSIG_SLOTSYNC_MESSAGE,
                       INVALID_PROC_NUMBER);
 
    /* Wait for slot sync to end */
    for (;;)
    {
        int         rc;
 
        /* Wait a bit, we don't expect to have to wait long */
        rc = WaitLatch(MyLatch,
                       WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                       10L, WAIT_EVENT_REPLICATION_SLOTSYNC_SHUTDOWN);
 
        if (rc & WL_LATCH_SET)
        {
            ResetLatch(MyLatch);
            CHECK_FOR_INTERRUPTS();
        }
 
        SpinLockAcquire(&SlotSyncCtx->mutex);
 
        /* Ensure that no process is syncing the slots. */
        if (!SlotSyncCtx->syncing)
            break;
 
        SpinLockRelease(&SlotSyncCtx->mutex);
    }
 
    SpinLockRelease(&SlotSyncCtx->mutex);
 
    update_synced_slots_inactive_since();
}

References CHECK_FOR_INTERRUPTS, fb(), INVALID_PROC_NUMBER, InvalidPid, SlotSyncCtxStruct::mutex, MyLatch, SlotSyncCtxStruct::pid, PROCSIG_SLOTSYNC_MESSAGE, ResetLatch(), SendProcSignal(), SlotSyncCtx, SpinLockAcquire(), SpinLockRelease(), SlotSyncCtxStruct::stopSignaled, SlotSyncCtxStruct::syncing, update_synced_slots_inactive_since(), WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by FinishWalRecovery().

slotsync_failure_callback()

static void slotsync_failure_callback ( int  code, Datum  arg  )

static

Definition at line 1960 of file slotsync.c.

{
    WalReceiverConn *wrconn = (WalReceiverConn *) DatumGetPointer(arg);
 
    /*
     * We need to do slots cleanup here just like WalSndErrorCleanup() does.
     *
     * The startup process during promotion invokes ShutDownSlotSync() which
     * waits for slot sync to finish and it does that by checking the
     * 'syncing' flag. Thus the SQL function must be done with slots' release
     * and cleanup to avoid any dangling temporary slots or active slots
     * before it marks itself as finished syncing.
     */
 
    /* Make sure active replication slots are released */
    if (MyReplicationSlot != NULL)
        ReplicationSlotRelease();
 
    /* Also cleanup the synced temporary slots. */
    ReplicationSlotCleanup(true);
 
    /*
     * The set syncing_slots indicates that the process errored out without
     * resetting the flag. So, we need to clean up shared memory and reset the
     * flag here.
     */
    if (syncing_slots)
        reset_syncing_flag();
 
    walrcv_disconnect(wrconn);
}

References arg, DatumGetPointer(), fb(), MyReplicationSlot, ReplicationSlotCleanup(), ReplicationSlotRelease(), reset_syncing_flag(), syncing_slots, walrcv_disconnect, and wrconn.

Referenced by SyncReplicationSlots().

slotsync_reread_config()

static void slotsync_reread_config ( void  )

static

Definition at line 1266 of file slotsync.c.

{
    char       *old_primary_conninfo = pstrdup(PrimaryConnInfo);
    char       *old_primary_slotname = pstrdup(PrimarySlotName);
    bool        old_sync_replication_slots = sync_replication_slots;
    bool        old_hot_standby_feedback = hot_standby_feedback;
    bool        conninfo_changed;
    bool        primary_slotname_changed;
    bool        is_slotsync_worker = AmLogicalSlotSyncWorkerProcess();
    bool        parameter_changed = false;
 
    if (is_slotsync_worker)
        Assert(sync_replication_slots);
 
    ConfigReloadPending = false;
    ProcessConfigFile(PGC_SIGHUP);
 
    conninfo_changed = strcmp(old_primary_conninfo, PrimaryConnInfo) != 0;
    primary_slotname_changed = strcmp(old_primary_slotname, PrimarySlotName) != 0;
    pfree(old_primary_conninfo);
    pfree(old_primary_slotname);
 
    if (old_sync_replication_slots != sync_replication_slots)
    {
        if (is_slotsync_worker)
        {
            ereport(LOG,
            /* translator: %s is a GUC variable name */
                    errmsg("replication slot synchronization worker will stop because \"%s\" is disabled",
                           "sync_replication_slots"));
 
            proc_exit(0);
        }
 
        parameter_changed = true;
    }
    else
    {
        if (conninfo_changed ||
            primary_slotname_changed ||
            (old_hot_standby_feedback != hot_standby_feedback))
        {
 
            if (is_slotsync_worker)
            {
                ereport(LOG,
                        errmsg("replication slot synchronization worker will restart because of a parameter change"));
 
                /*
                 * Reset the last-start time for this worker so that the
                 * postmaster can restart it without waiting for
                 * SLOTSYNC_RESTART_INTERVAL_SEC.
                 */
                SlotSyncCtx->last_start_time = 0;
 
                proc_exit(0);
            }
 
            parameter_changed = true;
        }
    }
 
    /*
     * If we have reached here with a parameter change, we must be running in
     * SQL function, emit error in such a case.
     */
    if (parameter_changed)
    {
        Assert(!is_slotsync_worker);
        ereport(ERROR,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                errmsg("replication slot synchronization will stop because of a parameter change"));
    }
 
}

References AmLogicalSlotSyncWorkerProcess, Assert, ConfigReloadPending, ereport, errcode(), errmsg, ERROR, fb(), hot_standby_feedback, SlotSyncCtxStruct::last_start_time, LOG, pfree(), PGC_SIGHUP, PrimaryConnInfo, PrimarySlotName, proc_exit(), ProcessConfigFile(), pstrdup(), SlotSyncCtx, and sync_replication_slots.

Referenced by ReplSlotSyncWorkerMain(), and SyncReplicationSlots().

slotsync_worker_disconnect()

static void slotsync_worker_disconnect ( int  code, Datum  arg  )

static

Definition at line 1397 of file slotsync.c.

{
    WalReceiverConn *wrconn = (WalReceiverConn *) DatumGetPointer(arg);
 
    walrcv_disconnect(wrconn);
}

References arg, DatumGetPointer(), walrcv_disconnect, and wrconn.

Referenced by ReplSlotSyncWorkerMain().

slotsync_worker_onexit()

static void slotsync_worker_onexit ( int  code, Datum  arg  )

static

Definition at line 1410 of file slotsync.c.

{
    /*
     * We need to do slots cleanup here just like WalSndErrorCleanup() does.
     *
     * The startup process during promotion invokes ShutDownSlotSync() which
     * waits for slot sync to finish and it does that by checking the
     * 'syncing' flag. Thus the slot sync worker must be done with slots'
     * release and cleanup to avoid any dangling temporary slots or active
     * slots before it marks itself as finished syncing.
     */
 
    /* Make sure active replication slots are released */
    if (MyReplicationSlot != NULL)
        ReplicationSlotRelease();
 
    /* Also cleanup the temporary slots. */
    ReplicationSlotCleanup(false);
 
    SpinLockAcquire(&SlotSyncCtx->mutex);
 
    SlotSyncCtx->pid = InvalidPid;
 
    /*
     * If syncing_slots is true, it indicates that the process errored out
     * without resetting the flag. So, we need to clean up shared memory and
     * reset the flag here.
     */
    if (syncing_slots)
    {
        SlotSyncCtx->syncing = false;
        syncing_slots = false;
    }
 
    SpinLockRelease(&SlotSyncCtx->mutex);
}

References fb(), InvalidPid, SlotSyncCtxStruct::mutex, MyReplicationSlot, SlotSyncCtxStruct::pid, ReplicationSlotCleanup(), ReplicationSlotRelease(), SlotSyncCtx, SpinLockAcquire(), SpinLockRelease(), SlotSyncCtxStruct::syncing, and syncing_slots.

Referenced by ReplSlotSyncWorkerMain().

SlotSyncShmemInit()

static void SlotSyncShmemInit ( void *  arg )

static

Definition at line 1949 of file slotsync.c.

{
    memset(SlotSyncCtx, 0, sizeof(SlotSyncCtxStruct));
    SlotSyncCtx->pid = InvalidPid;
    SpinLockInit(&SlotSyncCtx->mutex);
}

References fb(), InvalidPid, SlotSyncCtxStruct::mutex, SlotSyncCtxStruct::pid, SlotSyncCtx, and SpinLockInit().

SlotSyncShmemRequest()

static void SlotSyncShmemRequest ( void *  arg )

static

Definition at line 1937 of file slotsync.c.

{
    ShmemRequestStruct(.name = "Slot Sync Data",
                       .size = sizeof(SlotSyncCtxStruct),
                       .ptr = (void **) &SlotSyncCtx,
        );
}

References name, ShmemRequestStruct, and SlotSyncCtx.

SlotSyncWorkerCanRestart()

bool SlotSyncWorkerCanRestart

(

void

)

Definition at line 1903 of file slotsync.c.

{
    time_t      curtime = time(NULL);
 
    /*
     * If first time through, or time somehow went backwards, always update
     * last_start_time to match the current clock and allow worker start.
     * Otherwise allow it only once enough time has elapsed.
     */
    if (SlotSyncCtx->last_start_time == 0 ||
        curtime < SlotSyncCtx->last_start_time ||
        curtime - SlotSyncCtx->last_start_time >= SLOTSYNC_RESTART_INTERVAL_SEC)
    {
        SlotSyncCtx->last_start_time = curtime;
        return true;
    }
    return false;
}

References fb(), SlotSyncCtxStruct::last_start_time, SLOTSYNC_RESTART_INTERVAL_SEC, and SlotSyncCtx.

Referenced by LaunchMissingBackgroundProcesses().

synchronize_one_slot()

static bool synchronize_one_slot ( RemoteSlot *  remote_slot, Oid  remote_dbid, bool *  slot_persistence_pending  )

static

Definition at line 746 of file slotsync.c.

{
    ReplicationSlot *slot;
    bool        slot_updated = false;
 
    /* Search for the named slot */
    if ((slot = SearchNamedReplicationSlot(remote_slot->name, true)))
    {
        bool        synced;
 
        SpinLockAcquire(&slot->mutex);
        synced = slot->data.synced;
        SpinLockRelease(&slot->mutex);
 
        /* User-created slot with the same name exists, raise ERROR. */
        if (!synced)
            ereport(ERROR,
                    errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                    errmsg("exiting from slot synchronization because same"
                           " name slot \"%s\" already exists on the standby",
                           remote_slot->name));
 
        /*
         * The slot has been synchronized before.
         *
         * It is important to acquire the slot here before checking
         * invalidation. If we don't acquire the slot first, there could be a
         * race condition that the local slot could be invalidated just after
         * checking the 'invalidated' flag here and we could end up
         * overwriting 'invalidated' flag to remote_slot's value. See
         * InvalidatePossiblyObsoleteSlot() where it invalidates slot directly
         * if the slot is not acquired by other processes.
         *
         * XXX: If it ever turns out that slot acquire/release is costly for
         * cases when none of the slot properties is changed then we can do a
         * pre-check to ensure that at least one of the slot properties is
         * changed before acquiring the slot.
         */
        ReplicationSlotAcquire(remote_slot->name, true, false);
 
        Assert(slot == MyReplicationSlot);
 
        /*
         * Copy the invalidation cause from remote only if local slot is not
         * invalidated locally, we don't want to overwrite existing one.
         */
        if (slot->data.invalidated == RS_INVAL_NONE &&
            remote_slot->invalidated != RS_INVAL_NONE)
        {
            SpinLockAcquire(&slot->mutex);
            slot->data.invalidated = remote_slot->invalidated;
            SpinLockRelease(&slot->mutex);
 
            /* Make sure the invalidated state persists across server restart */
            ReplicationSlotMarkDirty();
            ReplicationSlotSave();
 
            slot_updated = true;
        }
 
        /* Skip the sync of an invalidated slot */
        if (slot->data.invalidated != RS_INVAL_NONE)
        {
            update_slotsync_skip_stats(SS_SKIP_INVALID);
 
            ReplicationSlotRelease();
            return slot_updated;
        }
 
        /* Slot not ready yet, let's attempt to make it sync-ready now. */
        if (slot->data.persistency == RS_TEMPORARY)
        {
            slot_updated = update_and_persist_local_synced_slot(remote_slot,
                                                                remote_dbid,
                                                                slot_persistence_pending);
        }
 
        /* Slot ready for sync, so sync it. */
        else
        {
            /*
             * Sanity check: As long as the invalidations are handled
             * appropriately as above, this should never happen.
             *
             * We don't need to check restart_lsn here. See the comments in
             * update_local_synced_slot() for details.
             */
            if (remote_slot->confirmed_lsn < slot->data.confirmed_flush)
                ereport(ERROR,
                        errmsg_internal("cannot synchronize local slot \"%s\"",
                                        remote_slot->name),
                        errdetail_internal("Local slot's start streaming location LSN(%X/%08X) is ahead of remote slot's LSN(%X/%08X).",
                                           LSN_FORMAT_ARGS(slot->data.confirmed_flush),
                                           LSN_FORMAT_ARGS(remote_slot->confirmed_lsn)));
 
            slot_updated = update_local_synced_slot(remote_slot, remote_dbid);
        }
    }
    /* Otherwise create the slot first. */
    else
    {
        NameData    plugin_name;
        TransactionId xmin_horizon = InvalidTransactionId;
 
        /* Skip creating the local slot if remote_slot is invalidated already */
        if (remote_slot->invalidated != RS_INVAL_NONE)
            return false;
 
        /*
         * We create temporary slots instead of ephemeral slots here because
         * we want the slots to survive after releasing them. This is done to
         * avoid dropping and re-creating the slots in each synchronization
         * cycle if the restart_lsn or catalog_xmin of the remote slot has not
         * caught up.
         */
        ReplicationSlotCreate(remote_slot->name, true, RS_TEMPORARY,
                              remote_slot->two_phase,
                              false,
                              remote_slot->failover,
                              true);
 
        /* For shorter lines. */
        slot = MyReplicationSlot;
 
        /* Avoid expensive operations while holding a spinlock. */
        namestrcpy(&plugin_name, remote_slot->plugin);
 
        SpinLockAcquire(&slot->mutex);
        slot->data.database = remote_dbid;
        slot->data.plugin = plugin_name;
        SpinLockRelease(&slot->mutex);
 
        reserve_wal_for_local_slot(remote_slot->restart_lsn);
 
        LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
        LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
        xmin_horizon = GetOldestSafeDecodingTransactionId(true);
        SpinLockAcquire(&slot->mutex);
        slot->effective_catalog_xmin = xmin_horizon;
        slot->data.catalog_xmin = xmin_horizon;
        SpinLockRelease(&slot->mutex);
        ReplicationSlotsComputeRequiredXmin(true);
        LWLockRelease(ProcArrayLock);
        LWLockRelease(ReplicationSlotControlLock);
 
        update_and_persist_local_synced_slot(remote_slot, remote_dbid,
                                             slot_persistence_pending);
 
        slot_updated = true;
    }
 
    ReplicationSlotRelease();
 
    return slot_updated;
}

References Assert, ReplicationSlotPersistentData::catalog_xmin, ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, ReplicationSlotPersistentData::database, ReplicationSlot::effective_catalog_xmin, ereport, errcode(), errdetail_internal(), errmsg, errmsg_internal(), ERROR, fb(), GetOldestSafeDecodingTransactionId(), ReplicationSlotPersistentData::invalidated, InvalidTransactionId, LSN_FORMAT_ARGS, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), ReplicationSlot::mutex, MyReplicationSlot, namestrcpy(), ReplicationSlotPersistentData::persistency, ReplicationSlotPersistentData::plugin, ReplicationSlotAcquire(), ReplicationSlotCreate(), ReplicationSlotMarkDirty(), ReplicationSlotRelease(), ReplicationSlotSave(), ReplicationSlotsComputeRequiredXmin(), reserve_wal_for_local_slot(), RS_INVAL_NONE, RS_TEMPORARY, SearchNamedReplicationSlot(), SpinLockAcquire(), SpinLockRelease(), SS_SKIP_INVALID, ReplicationSlotPersistentData::synced, update_and_persist_local_synced_slot(), update_local_synced_slot(), and update_slotsync_skip_stats().

Referenced by synchronize_slots().

synchronize_slots()

static bool synchronize_slots ( WalReceiverConn *  wrconn, List *  remote_slot_list, bool *  slot_persistence_pending  )

static

Definition at line 1056 of file slotsync.c.

{
    bool        some_slot_updated = false;
 
    /* Drop local slots that no longer need to be synced. */
    drop_local_obsolete_slots(remote_slot_list);
 
    /* Now sync the slots locally */
    foreach_ptr(RemoteSlot, remote_slot, remote_slot_list)
    {
        Oid         remote_dbid = get_database_oid(remote_slot->database, false);
 
        /*
         * Use shared lock to prevent a conflict with
         * ReplicationSlotsDropDBSlots(), trying to drop the same slot during
         * a drop-database operation.
         */
        LockSharedObject(DatabaseRelationId, remote_dbid, 0, AccessShareLock);
 
        some_slot_updated |= synchronize_one_slot(remote_slot, remote_dbid,
                                                  slot_persistence_pending);
 
        UnlockSharedObject(DatabaseRelationId, remote_dbid, 0, AccessShareLock);
    }
 
    return some_slot_updated;
}

References AccessShareLock, drop_local_obsolete_slots(), fb(), foreach_ptr, get_database_oid(), LockSharedObject(), synchronize_one_slot(), and UnlockSharedObject().

Referenced by ReplSlotSyncWorkerMain(), and SyncReplicationSlots().

SyncReplicationSlots()

void SyncReplicationSlots

(

WalReceiverConn *

wrconn

)

Definition at line 2022 of file slotsync.c.

{
    PG_ENSURE_ERROR_CLEANUP(slotsync_failure_callback, PointerGetDatum(wrconn));
    {
        List       *remote_slots = NIL;
        List       *slot_names = NIL;   /* List of slot names to track */
        MemoryContext sync_retry_ctx;
 
        check_and_set_sync_info(MyProcPid);
 
        validate_remote_info(wrconn);
 
        /*
         * Setup and use a per-sync-cycle memory context, which is reset every
         * time we loop below. This avoids having to retail freeing the memory
         * used in each sync cycle.
         */
        sync_retry_ctx = AllocSetContextCreate(CurrentMemoryContext,
                                               "slot sync retry context",
                                               ALLOCSET_DEFAULT_SIZES);
 
        /* Retry until all the slots are sync-ready */
        for (;;)
        {
            bool        slot_persistence_pending = false;
            bool        some_slot_updated = false;
            MemoryContext oldctx;
 
            /* Check for interrupts and config changes */
            CHECK_FOR_INTERRUPTS();
 
            if (ConfigReloadPending)
                slotsync_reread_config();
 
            /* We must be in a valid transaction state */
            Assert(IsTransactionState());
 
            MemoryContextReset(sync_retry_ctx);
            oldctx = MemoryContextSwitchTo(sync_retry_ctx);
 
            /*
             * Fetch remote slot info for the given slot_names. If slot_names
             * is NIL, fetch all failover-enabled slots. Note that we reuse
             * slot_names from the first iteration; re-fetching all failover
             * slots each time could cause an endless loop. Instead of
             * reprocessing only the pending slots in each iteration, it's
             * better to process all the slots received in the first
             * iteration. This ensures that by the time we're done, all slots
             * reflect the latest values.
             */
            remote_slots = fetch_remote_slots(wrconn, slot_names);
 
            /* Attempt to synchronize slots */
            some_slot_updated = synchronize_slots(wrconn, remote_slots,
                                                  &slot_persistence_pending);
 
            /*
             * slot_names must survive later sync_retry_ctx resets, so copy it
             * in the outer context.
             */
            MemoryContextSwitchTo(oldctx);
 
            /*
             * If slot_persistence_pending is true, extract slot names for
             * future iterations (only needed if we haven't done it yet)
             */
            if (slot_names == NIL && slot_persistence_pending)
                slot_names = extract_slot_names(remote_slots);
 
            /* Done if all slots are persisted i.e are sync-ready */
            if (!slot_persistence_pending)
                break;
 
            /* wait before retrying again */
            wait_for_slot_activity(some_slot_updated);
        }
 
        MemoryContextDelete(sync_retry_ctx);
 
        if (slot_names)
            list_free_deep(slot_names);
 
        /* Cleanup the synced temporary slots */
        ReplicationSlotCleanup(true);
 
        /* We are done with sync, so reset sync flag */
        reset_syncing_flag();
    }
    PG_END_ENSURE_ERROR_CLEANUP(slotsync_failure_callback, PointerGetDatum(wrconn));
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert, check_and_set_sync_info(), CHECK_FOR_INTERRUPTS, ConfigReloadPending, CurrentMemoryContext, extract_slot_names(), fb(), fetch_remote_slots(), IsTransactionState(), list_free_deep(), MemoryContextDelete(), MemoryContextReset(), MemoryContextSwitchTo(), MyProcPid, NIL, PG_END_ENSURE_ERROR_CLEANUP, PG_ENSURE_ERROR_CLEANUP, PointerGetDatum, ReplicationSlotCleanup(), reset_syncing_flag(), slotsync_failure_callback(), slotsync_reread_config(), synchronize_slots(), validate_remote_info(), wait_for_slot_activity(), and wrconn.

Referenced by pg_sync_replication_slots().

update_and_persist_local_synced_slot()

static bool update_and_persist_local_synced_slot ( RemoteSlot *  remote_slot, Oid  remote_dbid, bool *  slot_persistence_pending  )

static

Definition at line 685 of file slotsync.c.

{
    ReplicationSlot *slot = MyReplicationSlot;
 
    /* Slotsync skip stats are handled in function update_local_synced_slot() */
    (void) update_local_synced_slot(remote_slot, remote_dbid);
 
    /*
     * Check if the slot cannot be synchronized. Refer to the comment atop the
     * file for details on this check.
     */
    if (slot->slotsync_skip_reason != SS_SKIP_NONE)
    {
        /*
         * We reach this point when the remote slot didn't catch up to locally
         * reserved position, or it cannot reach the consistent point from the
         * restart_lsn, or the WAL prior to the remote confirmed flush LSN has
         * not been received and flushed.
         *
         * We do not drop the slot because the restart_lsn and confirmed_lsn
         * can be ahead of the current location when recreating the slot in
         * the next cycle. It may take more time to create such a slot or
         * reach the consistent point. Therefore, we keep this slot and
         * attempt the synchronization in the next cycle.
         *
         * We also update the slot_persistence_pending parameter, so the SQL
         * function can retry.
         */
        if (slot_persistence_pending)
            *slot_persistence_pending = true;
 
        return false;
    }
 
    ReplicationSlotPersist();
 
    ereport(LOG,
            errmsg("newly created replication slot \"%s\" is sync-ready now",
                   remote_slot->name));
 
    return true;
}

References ereport, errmsg, fb(), LOG, MyReplicationSlot, ReplicationSlotPersist(), ReplicationSlot::slotsync_skip_reason, SS_SKIP_NONE, and update_local_synced_slot().

Referenced by synchronize_one_slot().

update_local_synced_slot()

static bool update_local_synced_slot ( RemoteSlot *  remote_slot, Oid  remote_dbid  )

static

Definition at line 221 of file slotsync.c.

{
    ReplicationSlot *slot = MyReplicationSlot;
    bool        updated_xmin_or_lsn = false;
    bool        updated_config = false;
    SlotSyncSkipReason skip_reason = SS_SKIP_NONE;
    XLogRecPtr  latestFlushPtr = GetStandbyFlushRecPtr(NULL);
 
    Assert(slot->data.invalidated == RS_INVAL_NONE);
 
    /*
     * Make sure that concerned WAL is received and flushed before syncing
     * slot to target lsn received from the primary server.
     */
    if (remote_slot->confirmed_lsn > latestFlushPtr)
    {
        update_slotsync_skip_stats(SS_SKIP_WAL_NOT_FLUSHED);
 
        /*
         * Can get here only if GUC 'synchronized_standby_slots' on the
         * primary server was not configured correctly.
         */
        ereport(LOG,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("skipping slot synchronization because the received slot sync"
                       " LSN %X/%08X for slot \"%s\" is ahead of the standby position %X/%08X",
                       LSN_FORMAT_ARGS(remote_slot->confirmed_lsn),
                       remote_slot->name,
                       LSN_FORMAT_ARGS(latestFlushPtr)));
 
        return false;
    }
 
    /*
     * Don't overwrite if we already have a newer catalog_xmin and
     * restart_lsn.
     */
    if (remote_slot->restart_lsn < slot->data.restart_lsn ||
        TransactionIdPrecedes(remote_slot->catalog_xmin,
                              slot->data.catalog_xmin))
    {
        /* Update slot sync skip stats */
        update_slotsync_skip_stats(SS_SKIP_WAL_OR_ROWS_REMOVED);
 
        /*
         * This can happen in following situations:
         *
         * If the slot is temporary, it means either the initial WAL location
         * reserved for the local slot is ahead of the remote slot's
         * restart_lsn or the initial xmin_horizon computed for the local slot
         * is ahead of the remote slot.
         *
         * If the slot is persistent, both restart_lsn and catalog_xmin of the
         * synced slot could still be ahead of the remote slot. Since we use
         * slot advance functionality to keep snapbuild/slot updated, it is
         * possible that the restart_lsn and catalog_xmin are advanced to a
         * later position than it has on the primary. This can happen when
         * slot advancing machinery finds running xacts record after reaching
         * the consistent state at a later point than the primary where it
         * serializes the snapshot and updates the restart_lsn.
         *
         * We LOG the message if the slot is temporary as it can help the user
         * to understand why the slot is not sync-ready. In the case of a
         * persistent slot, it would be a more common case and won't directly
         * impact the users, so we used DEBUG1 level to log the message.
         */
        ereport(slot->data.persistency == RS_TEMPORARY ? LOG : DEBUG1,
                errmsg("could not synchronize replication slot \"%s\"",
                       remote_slot->name),
                errdetail("Synchronization could lead to data loss, because the remote slot needs WAL at LSN %X/%08X and catalog xmin %u, but the standby has LSN %X/%08X and catalog xmin %u.",
                          LSN_FORMAT_ARGS(remote_slot->restart_lsn),
                          remote_slot->catalog_xmin,
                          LSN_FORMAT_ARGS(slot->data.restart_lsn),
                          slot->data.catalog_xmin));
 
        /*
         * Skip updating the configuration. This is required to avoid syncing
         * two_phase_at without syncing confirmed_lsn. Otherwise, the prepared
         * transaction between old confirmed_lsn and two_phase_at will
         * unexpectedly get decoded and sent to the downstream after
         * promotion. See comments in ReorderBufferFinishPrepared.
         */
        return false;
    }
 
    /*
     * Attempt to sync LSNs and xmins only if remote slot is ahead of local
     * slot.
     */
    if (remote_slot->confirmed_lsn > slot->data.confirmed_flush ||
        remote_slot->restart_lsn > slot->data.restart_lsn ||
        TransactionIdFollows(remote_slot->catalog_xmin,
                             slot->data.catalog_xmin))
    {
        /*
         * We can't directly copy the remote slot's LSN or xmin unless there
         * exists a consistent snapshot at that point. Otherwise, after
         * promotion, the slots may not reach a consistent point before the
         * confirmed_flush_lsn which can lead to a data loss. To avoid data
         * loss, we let slot machinery advance the slot which ensures that
         * snapbuilder/slot statuses are updated properly.
         */
        if (SnapBuildSnapshotExists(remote_slot->restart_lsn))
        {
            /*
             * Update the slot info directly if there is a serialized snapshot
             * at the restart_lsn, as the slot can quickly reach consistency
             * at restart_lsn by restoring the snapshot.
             */
            SpinLockAcquire(&slot->mutex);
            slot->data.restart_lsn = remote_slot->restart_lsn;
            slot->data.confirmed_flush = remote_slot->confirmed_lsn;
            slot->data.catalog_xmin = remote_slot->catalog_xmin;
            SpinLockRelease(&slot->mutex);
 
            updated_xmin_or_lsn = true;
        }
        else
        {
            bool        found_consistent_snapshot;
            XLogRecPtr  old_confirmed_lsn = slot->data.confirmed_flush;
            XLogRecPtr  old_restart_lsn = slot->data.restart_lsn;
            TransactionId old_catalog_xmin = slot->data.catalog_xmin;
 
            LogicalSlotAdvanceAndCheckSnapState(remote_slot->confirmed_lsn,
                                                &found_consistent_snapshot);
 
            /* Sanity check */
            if (slot->data.confirmed_flush != remote_slot->confirmed_lsn)
                ereport(ERROR,
                        errmsg_internal("synchronized confirmed_flush for slot \"%s\" differs from remote slot",
                                        remote_slot->name),
                        errdetail_internal("Remote slot has LSN %X/%08X but local slot has LSN %X/%08X.",
                                           LSN_FORMAT_ARGS(remote_slot->confirmed_lsn),
                                           LSN_FORMAT_ARGS(slot->data.confirmed_flush)));
 
            /*
             * If we can't reach a consistent snapshot, the slot won't be
             * persisted. See update_and_persist_local_synced_slot().
             */
            if (!found_consistent_snapshot)
            {
                Assert(MyReplicationSlot->data.persistency == RS_TEMPORARY);
 
                ereport(LOG,
                        errmsg("could not synchronize replication slot \"%s\"",
                               remote_slot->name),
                        errdetail("Synchronization could lead to data loss, because the standby could not build a consistent snapshot to decode WALs at LSN %X/%08X.",
                                  LSN_FORMAT_ARGS(slot->data.restart_lsn)));
 
                skip_reason = SS_SKIP_NO_CONSISTENT_SNAPSHOT;
            }
 
            /*
             * It is possible that the slot's xmin or LSNs are not updated,
             * when the synced slot has reached consistent snapshot state or
             * cannot build one at all.
             */
            updated_xmin_or_lsn = (old_confirmed_lsn != slot->data.confirmed_flush ||
                                   old_restart_lsn != slot->data.restart_lsn ||
                                   old_catalog_xmin != slot->data.catalog_xmin);
        }
    }
 
    /* Update slot sync skip stats */
    update_slotsync_skip_stats(skip_reason);
 
    if (remote_dbid != slot->data.database ||
        remote_slot->two_phase != slot->data.two_phase ||
        remote_slot->failover != slot->data.failover ||
        strcmp(remote_slot->plugin, NameStr(slot->data.plugin)) != 0 ||
        remote_slot->two_phase_at != slot->data.two_phase_at)
    {
        NameData    plugin_name;
 
        /* Avoid expensive operations while holding a spinlock. */
        namestrcpy(&plugin_name, remote_slot->plugin);
 
        SpinLockAcquire(&slot->mutex);
        slot->data.plugin = plugin_name;
        slot->data.database = remote_dbid;
        slot->data.two_phase = remote_slot->two_phase;
        slot->data.two_phase_at = remote_slot->two_phase_at;
        slot->data.failover = remote_slot->failover;
        SpinLockRelease(&slot->mutex);
 
        updated_config = true;
 
        /*
         * Ensure that there is no risk of sending prepared transactions
         * unexpectedly after the promotion.
         */
        Assert(slot->data.two_phase_at <= slot->data.confirmed_flush);
    }
 
    /*
     * We have to write the changed xmin to disk *before* we change the
     * in-memory value, otherwise after a crash we wouldn't know that some
     * catalog tuples might have been removed already.
     */
    if (updated_config || updated_xmin_or_lsn)
    {
        ReplicationSlotMarkDirty();
        ReplicationSlotSave();
    }
 
    /*
     * Now the new xmin is safely on disk, we can let the global value
     * advance. We do not take ProcArrayLock or similar since we only advance
     * xmin here and there's not much harm done by a concurrent computation
     * missing that.
     */
    if (updated_xmin_or_lsn)
    {
        SpinLockAcquire(&slot->mutex);
        slot->effective_catalog_xmin = remote_slot->catalog_xmin;
        SpinLockRelease(&slot->mutex);
 
        ReplicationSlotsComputeRequiredXmin(false);
        ReplicationSlotsComputeRequiredLSN();
    }
 
    return updated_config || updated_xmin_or_lsn;
}

References Assert, ReplicationSlotPersistentData::catalog_xmin, ReplicationSlotPersistentData::confirmed_flush, ReplicationSlot::data, ReplicationSlotPersistentData::database, DEBUG1, ReplicationSlot::effective_catalog_xmin, ereport, errcode(), errdetail(), errdetail_internal(), errmsg, errmsg_internal(), ERROR, ReplicationSlotPersistentData::failover, fb(), GetStandbyFlushRecPtr(), ReplicationSlotPersistentData::invalidated, LOG, LogicalSlotAdvanceAndCheckSnapState(), LSN_FORMAT_ARGS, ReplicationSlot::mutex, MyReplicationSlot, NameStr, namestrcpy(), ReplicationSlotPersistentData::persistency, ReplicationSlotPersistentData::plugin, ReplicationSlotMarkDirty(), ReplicationSlotSave(), ReplicationSlotsComputeRequiredLSN(), ReplicationSlotsComputeRequiredXmin(), ReplicationSlotPersistentData::restart_lsn, RS_INVAL_NONE, RS_TEMPORARY, SnapBuildSnapshotExists(), SpinLockAcquire(), SpinLockRelease(), SS_SKIP_NO_CONSISTENT_SNAPSHOT, SS_SKIP_NONE, SS_SKIP_WAL_NOT_FLUSHED, SS_SKIP_WAL_OR_ROWS_REMOVED, TransactionIdFollows(), TransactionIdPrecedes(), ReplicationSlotPersistentData::two_phase, ReplicationSlotPersistentData::two_phase_at, and update_slotsync_skip_stats().

Referenced by synchronize_one_slot(), and update_and_persist_local_synced_slot().

update_slotsync_skip_stats()

static void update_slotsync_skip_stats ( SlotSyncSkipReason  skip_reason )

static

Definition at line 189 of file slotsync.c.

{
    ReplicationSlot *slot;
 
    Assert(MyReplicationSlot);
 
    slot = MyReplicationSlot;
 
    /*
     * Update the slot sync related stats in pg_stat_replication_slots when a
     * slot sync is skipped
     */
    if (skip_reason != SS_SKIP_NONE)
        pgstat_report_replslotsync(slot);
 
    /* Update the slot sync skip reason */
    if (slot->slotsync_skip_reason != skip_reason)
    {
        SpinLockAcquire(&slot->mutex);
        slot->slotsync_skip_reason = skip_reason;
        SpinLockRelease(&slot->mutex);
    }
}

References Assert, fb(), ReplicationSlot::mutex, MyReplicationSlot, pgstat_report_replslotsync(), ReplicationSlot::slotsync_skip_reason, SpinLockAcquire(), SpinLockRelease(), and SS_SKIP_NONE.

Referenced by synchronize_one_slot(), and update_local_synced_slot().

update_synced_slots_inactive_since()

static void update_synced_slots_inactive_since ( void  )

static

Definition at line 1778 of file slotsync.c.

{
    TimestampTz now = 0;
 
    /*
     * We need to update inactive_since only when we are promoting standby to
     * correctly interpret the inactive_since if the standby gets promoted
     * without a restart. We don't want the slots to appear inactive for a
     * long time after promotion if they haven't been synchronized recently.
     * Whoever acquires the slot, i.e., makes the slot active, will reset it.
     */
    if (!StandbyMode)
        return;
 
    /* The slot sync worker or the SQL function mustn't be running by now */
    Assert((SlotSyncCtx->pid == InvalidPid) && !SlotSyncCtx->syncing);
 
    LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
    for (int i = 0; i < max_replication_slots + max_repack_replication_slots; i++)
    {
        ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
 
        /* Check if it is a synchronized slot */
        if (s->in_use && s->data.synced)
        {
            Assert(SlotIsLogical(s));
 
            /* The slot must not be acquired by any process */
            Assert(s->active_proc == INVALID_PROC_NUMBER);
 
            /* Use the same inactive_since time for all the slots. */
            if (now == 0)
                now = GetCurrentTimestamp();
 
            ReplicationSlotSetInactiveSince(s, now, true);
        }
    }
 
    LWLockRelease(ReplicationSlotControlLock);
}

References ReplicationSlot::active_proc, Assert, ReplicationSlot::data, fb(), GetCurrentTimestamp(), i, ReplicationSlot::in_use, INVALID_PROC_NUMBER, InvalidPid, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_repack_replication_slots, max_replication_slots, now(), SlotSyncCtxStruct::pid, ReplicationSlotCtlData::replication_slots, ReplicationSlotCtl, ReplicationSlotSetInactiveSince(), SlotIsLogical, SlotSyncCtx, StandbyMode, ReplicationSlotPersistentData::synced, and SlotSyncCtxStruct::syncing.

Referenced by ShutDownSlotSync().

validate_remote_info()

static void validate_remote_info ( WalReceiverConn *  wrconn )

static

Definition at line 1092 of file slotsync.c.

{
#define PRIMARY_INFO_OUTPUT_COL_COUNT 2
    WalRcvExecResult *res;
    Oid         slotRow[PRIMARY_INFO_OUTPUT_COL_COUNT] = {BOOLOID, BOOLOID};
    StringInfoData cmd;
    bool        isnull;
    TupleTableSlot *tupslot;
    bool        remote_in_recovery;
    bool        primary_slot_valid;
    bool        started_tx = false;
 
    initStringInfo(&cmd);
    appendStringInfo(&cmd,
                     "SELECT pg_is_in_recovery(), count(*) = 1"
                     " FROM pg_catalog.pg_replication_slots"
                     " WHERE slot_type='physical' AND slot_name=%s",
                     quote_literal_cstr(PrimarySlotName));
 
    /* The syscache access in walrcv_exec() needs a transaction env. */
    if (!IsTransactionState())
    {
        StartTransactionCommand();
        started_tx = true;
    }
 
    res = walrcv_exec(wrconn, cmd.data, PRIMARY_INFO_OUTPUT_COL_COUNT, slotRow);
    pfree(cmd.data);
 
    if (res->status != WALRCV_OK_TUPLES)
        ereport(ERROR,
                errmsg("could not fetch primary slot name \"%s\" info from the primary server: %s",
                       PrimarySlotName, res->err),
                errhint("Check if \"primary_slot_name\" is configured correctly."));
 
    tupslot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
    if (!tuplestore_gettupleslot(res->tuplestore, true, false, tupslot))
        elog(ERROR,
             "failed to fetch tuple for the primary server slot specified by \"primary_slot_name\"");
 
    remote_in_recovery = DatumGetBool(slot_getattr(tupslot, 1, &isnull));
    Assert(!isnull);
 
    /*
     * Slot sync is currently not supported on a cascading standby. This is
     * because if we allow it, the primary server needs to wait for all the
     * cascading standbys, otherwise, logical subscribers can still be ahead
     * of one of the cascading standbys which we plan to promote. Thus, to
     * avoid this additional complexity, we restrict it for the time being.
     */
    if (remote_in_recovery)
        ereport(ERROR,
                errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                errmsg("cannot synchronize replication slots from a standby server"));
 
    primary_slot_valid = DatumGetBool(slot_getattr(tupslot, 2, &isnull));
    Assert(!isnull);
 
    if (!primary_slot_valid)
        ereport(ERROR,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
        /* translator: second %s is a GUC variable name */
                errmsg("replication slot \"%s\" specified by \"%s\" does not exist on primary server",
                       PrimarySlotName, "primary_slot_name"));
 
    ExecDropSingleTupleTableSlot(tupslot);
    walrcv_clear_result(res);
 
    if (started_tx)
        CommitTransactionCommand();
}

References appendStringInfo(), Assert, CommitTransactionCommand(), StringInfoData::data, DatumGetBool(), elog, ereport, WalRcvExecResult::err, errcode(), errhint(), errmsg, ERROR, ExecDropSingleTupleTableSlot(), fb(), initStringInfo(), IsTransactionState(), MakeSingleTupleTableSlot(), pfree(), PRIMARY_INFO_OUTPUT_COL_COUNT, PrimarySlotName, quote_literal_cstr(), slot_getattr(), StartTransactionCommand(), WalRcvExecResult::status, TTSOpsMinimalTuple, WalRcvExecResult::tupledesc, WalRcvExecResult::tuplestore, tuplestore_gettupleslot(), walrcv_clear_result(), walrcv_exec, WALRCV_OK_TUPLES, and wrconn.

Referenced by ReplSlotSyncWorkerMain(), and SyncReplicationSlots().

ValidateSlotSyncParams()

bool ValidateSlotSyncParams

(

int

elevel

)

Definition at line 1197 of file slotsync.c.

{
    /*
     * Logical slot sync/creation requires logical decoding to be enabled.
     */
    if (!IsLogicalDecodingEnabled())
    {
        ereport(elevel,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                errmsg("replication slot synchronization requires \"effective_wal_level\" >= \"logical\" on the primary"),
                errhint("To enable logical decoding on primary, set \"wal_level\" >= \"logical\" or create at least one logical slot when \"wal_level\" = \"replica\"."));
 
        return false;
    }
 
    /*
     * A physical replication slot(primary_slot_name) is required on the
     * primary to ensure that the rows needed by the standby are not removed
     * after restarting, so that the synchronized slot on the standby will not
     * be invalidated.
     */
    if (PrimarySlotName == NULL || *PrimarySlotName == '\0')
    {
        ereport(elevel,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
        /* translator: %s is a GUC variable name */
                errmsg("replication slot synchronization requires \"%s\" to be set", "primary_slot_name"));
        return false;
    }
 
    /*
     * hot_standby_feedback must be enabled to cooperate with the physical
     * replication slot, which allows informing the primary about the xmin and
     * catalog_xmin values on the standby.
     */
    if (!hot_standby_feedback)
    {
        ereport(elevel,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
        /* translator: %s is a GUC variable name */
                errmsg("replication slot synchronization requires \"%s\" to be enabled",
                       "hot_standby_feedback"));
        return false;
    }
 
    /*
     * The primary_conninfo is required to make connection to primary for
     * getting slots information.
     */
    if (PrimaryConnInfo == NULL || *PrimaryConnInfo == '\0')
    {
        ereport(elevel,
                errcode(ERRCODE_INVALID_PARAMETER_VALUE),
        /* translator: %s is a GUC variable name */
                errmsg("replication slot synchronization requires \"%s\" to be set",
                       "primary_conninfo"));
        return false;
    }
 
    return true;
}

References ereport, errcode(), errhint(), errmsg, fb(), hot_standby_feedback, IsLogicalDecodingEnabled(), PrimaryConnInfo, and PrimarySlotName.

Referenced by LaunchMissingBackgroundProcesses(), and pg_sync_replication_slots().

wait_for_slot_activity()

static void wait_for_slot_activity ( bool  some_slot_updated )

static

Definition at line 1456 of file slotsync.c.

{
    int         rc;
 
    if (!some_slot_updated)
    {
        /*
         * No slots were updated, so double the sleep time, but not beyond the
         * maximum allowable value.
         */
        sleep_ms = Min(sleep_ms * 2, MAX_SLOTSYNC_WORKER_NAPTIME_MS);
    }
    else
    {
        /*
         * Some slots were updated since the last sleep, so reset the sleep
         * time.
         */
        sleep_ms = MIN_SLOTSYNC_WORKER_NAPTIME_MS;
    }
 
    rc = WaitLatch(MyLatch,
                   WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                   sleep_ms,
                   WAIT_EVENT_REPLICATION_SLOTSYNC_MAIN);
 
    if (rc & WL_LATCH_SET)
        ResetLatch(MyLatch);
}

References fb(), MAX_SLOTSYNC_WORKER_NAPTIME_MS, Min, MIN_SLOTSYNC_WORKER_NAPTIME_MS, MyLatch, ResetLatch(), sleep_ms, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by ReplSlotSyncWorkerMain(), and SyncReplicationSlots().

Variable Documentation

sleep_ms

long sleep_ms = MIN_SLOTSYNC_WORKER_NAPTIME_MS

static

Definition at line 142 of file slotsync.c.

Referenced by do_watch(), and wait_for_slot_activity().

SlotSyncCtx

SlotSyncCtxStruct* SlotSyncCtx = NULL

static

Definition at line 121 of file slotsync.c.

Referenced by check_and_set_sync_info(), ReplSlotSyncWorkerMain(), reset_syncing_flag(), ShutDownSlotSync(), slotsync_reread_config(), slotsync_worker_onexit(), SlotSyncShmemInit(), SlotSyncShmemRequest(), SlotSyncWorkerCanRestart(), and update_synced_slots_inactive_since().

SlotSyncShmemCallbacks

const ShmemCallbacks SlotSyncShmemCallbacks

Initial value:

= {
    .request_fn = SlotSyncShmemRequest,
    .init_fn = SlotSyncShmemInit,
}

Definition at line 126 of file slotsync.c.

                                              {
    .request_fn = SlotSyncShmemRequest,
    .init_fn = SlotSyncShmemInit,
};

SlotSyncShutdownPending

volatile sig_atomic_t SlotSyncShutdownPending = false

Definition at line 159 of file slotsync.c.

Referenced by HandleSlotSyncMessageInterrupt(), ProcessInterrupts(), and ProcessSlotSyncMessage().

sync_replication_slots

bool sync_replication_slots = false

Definition at line 132 of file slotsync.c.

Referenced by LaunchMissingBackgroundProcesses(), PostmasterMain(), slotsync_reread_config(), and xlog_redo().

syncing_slots

bool syncing_slots = false

static

Definition at line 152 of file slotsync.c.

Referenced by check_and_set_sync_info(), IsSyncingReplicationSlots(), reset_syncing_flag(), slotsync_failure_callback(), and slotsync_worker_onexit().