origin.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * origin.c
 *    Logical replication progress tracking support.
 *
 * Copyright (c) 2013-2024, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *    src/backend/replication/logical/origin.c
 *
 * NOTES
 *
 * This file provides the following:
 * * An infrastructure to name nodes in a replication setup
 * * A facility to efficiently store and persist replication progress in an
 *   efficient and durable manner.
 *
 * Replication origin consist out of a descriptive, user defined, external
 * name and a short, thus space efficient, internal 2 byte one. This split
 * exists because replication origin have to be stored in WAL and shared
 * memory and long descriptors would be inefficient.  For now only use 2 bytes
 * for the internal id of a replication origin as it seems unlikely that there
 * soon will be more than 65k nodes in one replication setup; and using only
 * two bytes allow us to be more space efficient.
 *
 * Replication progress is tracked in a shared memory table
 * (ReplicationState) that's dumped to disk every checkpoint. Entries
 * ('slots') in this table are identified by the internal id. That's the case
 * because it allows to increase replication progress during crash
 * recovery. To allow doing so we store the original LSN (from the originating
 * system) of a transaction in the commit record. That allows to recover the
 * precise replayed state after crash recovery; without requiring synchronous
 * commits. Allowing logical replication to use asynchronous commit is
 * generally good for performance, but especially important as it allows a
 * single threaded replay process to keep up with a source that has multiple
 * backends generating changes concurrently.  For efficiency and simplicity
 * reasons a backend can setup one replication origin that's from then used as
 * the source of changes produced by the backend, until reset again.
 *
 * This infrastructure is intended to be used in cooperation with logical
 * decoding. When replaying from a remote system the configured origin is
 * provided to output plugins, allowing prevention of replication loops and
 * other filtering.
 *
 * There are several levels of locking at work:
 *
 * * To create and drop replication origins an exclusive lock on
 *   pg_replication_slot is required for the duration. That allows us to
 *   safely and conflict free assign new origins using a dirty snapshot.
 *
 * * When creating an in-memory replication progress slot the ReplicationOrigin
 *   LWLock has to be held exclusively; when iterating over the replication
 *   progress a shared lock has to be held, the same when advancing the
 *   replication progress of an individual backend that has not setup as the
 *   session's replication origin.
 *
 * * When manipulating or looking at the remote_lsn and local_lsn fields of a
 *   replication progress slot that slot's lwlock has to be held. That's
 *   primarily because we do not assume 8 byte writes (the LSN) is atomic on
 *   all our platforms, but it also simplifies memory ordering concerns
 *   between the remote and local lsn. We use a lwlock instead of a spinlock
 *   so it's less harmful to hold the lock over a WAL write
 *   (cf. AdvanceReplicationProgress).
 *
 * ---------------------------------------------------------------------------
 */
 
#include "postgres.h"
 
#include <unistd.h>
#include <sys/stat.h>
 
#include "access/genam.h"
#include "access/htup_details.h"
#include "access/table.h"
#include "access/xact.h"
#include "access/xloginsert.h"
#include "catalog/catalog.h"
#include "catalog/indexing.h"
#include "catalog/pg_subscription.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "nodes/execnodes.h"
#include "pgstat.h"
#include "replication/origin.h"
#include "replication/slot.h"
#include "storage/condition_variable.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/pg_lsn.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"
 
/* paths for replication origin checkpoint files */
#define PG_REPLORIGIN_CHECKPOINT_FILENAME PG_LOGICAL_DIR "/replorigin_checkpoint"
#define PG_REPLORIGIN_CHECKPOINT_TMPFILE PG_REPLORIGIN_CHECKPOINT_FILENAME ".tmp"
 
/*
 * Replay progress of a single remote node.
 */
typedef struct ReplicationState
{
    /*
     * Local identifier for the remote node.
     */
    RepOriginId roident;
 
    /*
     * Location of the latest commit from the remote side.
     */
    XLogRecPtr  remote_lsn;
 
    /*
     * Remember the local lsn of the commit record so we can XLogFlush() to it
     * during a checkpoint so we know the commit record actually is safe on
     * disk.
     */
    XLogRecPtr  local_lsn;
 
    /*
     * PID of backend that's acquired slot, or 0 if none.
     */
    int         acquired_by;
 
    /*
     * Condition variable that's signaled when acquired_by changes.
     */
    ConditionVariable origin_cv;
 
    /*
     * Lock protecting remote_lsn and local_lsn.
     */
    LWLock      lock;
} ReplicationState;
 
/*
 * On disk version of ReplicationState.
 */
typedef struct ReplicationStateOnDisk
{
    RepOriginId roident;
    XLogRecPtr  remote_lsn;
} ReplicationStateOnDisk;
 
 
typedef struct ReplicationStateCtl
{
    /* Tranche to use for per-origin LWLocks */
    int         tranche_id;
    /* Array of length max_replication_slots */
    ReplicationState states[FLEXIBLE_ARRAY_MEMBER];
} ReplicationStateCtl;
 
/* external variables */
RepOriginId replorigin_session_origin = InvalidRepOriginId; /* assumed identity */
XLogRecPtr  replorigin_session_origin_lsn = InvalidXLogRecPtr;
TimestampTz replorigin_session_origin_timestamp = 0;
 
/*
 * Base address into a shared memory array of replication states of size
 * max_replication_slots.
 *
 * XXX: Should we use a separate variable to size this rather than
 * max_replication_slots?
 */
static ReplicationState *replication_states;
 
/*
 * Actual shared memory block (replication_states[] is now part of this).
 */
static ReplicationStateCtl *replication_states_ctl;
 
/*
 * We keep a pointer to this backend's ReplicationState to avoid having to
 * search the replication_states array in replorigin_session_advance for each
 * remote commit.  (Ownership of a backend's own entry can only be changed by
 * that backend.)
 */
static ReplicationState *session_replication_state = NULL;
 
/* Magic for on disk files. */
#define REPLICATION_STATE_MAGIC ((uint32) 0x1257DADE)
 
static void
replorigin_check_prerequisites(bool check_slots, bool recoveryOK)
{
    if (check_slots && max_replication_slots == 0)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("cannot query or manipulate replication origin when \"max_replication_slots\" is 0")));
 
    if (!recoveryOK && RecoveryInProgress())
        ereport(ERROR,
                (errcode(ERRCODE_READ_ONLY_SQL_TRANSACTION),
                 errmsg("cannot manipulate replication origins during recovery")));
}
 
 
/*
 * IsReservedOriginName
 *      True iff name is either "none" or "any".
 */
static bool
IsReservedOriginName(const char *name)
{
    return ((pg_strcasecmp(name, LOGICALREP_ORIGIN_NONE) == 0) ||
            (pg_strcasecmp(name, LOGICALREP_ORIGIN_ANY) == 0));
}
 
/* ---------------------------------------------------------------------------
 * Functions for working with replication origins themselves.
 * ---------------------------------------------------------------------------
 */
 
/*
 * Check for a persistent replication origin identified by name.
 *
 * Returns InvalidOid if the node isn't known yet and missing_ok is true.
 */
RepOriginId
replorigin_by_name(const char *roname, bool missing_ok)
{
    Form_pg_replication_origin ident;
    Oid         roident = InvalidOid;
    HeapTuple   tuple;
    Datum       roname_d;
 
    roname_d = CStringGetTextDatum(roname);
 
    tuple = SearchSysCache1(REPLORIGNAME, roname_d);
    if (HeapTupleIsValid(tuple))
    {
        ident = (Form_pg_replication_origin) GETSTRUCT(tuple);
        roident = ident->roident;
        ReleaseSysCache(tuple);
    }
    else if (!missing_ok)
        ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_OBJECT),
                 errmsg("replication origin \"%s\" does not exist",
                        roname)));
 
    return roident;
}
 
/*
 * Create a replication origin.
 *
 * Needs to be called in a transaction.
 */
RepOriginId
replorigin_create(const char *roname)
{
    Oid         roident;
    HeapTuple   tuple = NULL;
    Relation    rel;
    Datum       roname_d;
    SnapshotData SnapshotDirty;
    SysScanDesc scan;
    ScanKeyData key;
 
    roname_d = CStringGetTextDatum(roname);
 
    Assert(IsTransactionState());
 
    /*
     * We need the numeric replication origin to be 16bit wide, so we cannot
     * rely on the normal oid allocation. Instead we simply scan
     * pg_replication_origin for the first unused id. That's not particularly
     * efficient, but this should be a fairly infrequent operation - we can
     * easily spend a bit more code on this when it turns out it needs to be
     * faster.
     *
     * We handle concurrency by taking an exclusive lock (allowing reads!)
     * over the table for the duration of the search. Because we use a "dirty
     * snapshot" we can read rows that other in-progress sessions have
     * written, even though they would be invisible with normal snapshots. Due
     * to the exclusive lock there's no danger that new rows can appear while
     * we're checking.
     */
    InitDirtySnapshot(SnapshotDirty);
 
    rel = table_open(ReplicationOriginRelationId, ExclusiveLock);
 
    for (roident = InvalidOid + 1; roident < PG_UINT16_MAX; roident++)
    {
        bool        nulls[Natts_pg_replication_origin];
        Datum       values[Natts_pg_replication_origin];
        bool        collides;
 
        CHECK_FOR_INTERRUPTS();
 
        ScanKeyInit(&key,
                    Anum_pg_replication_origin_roident,
                    BTEqualStrategyNumber, F_OIDEQ,
                    ObjectIdGetDatum(roident));
 
        scan = systable_beginscan(rel, ReplicationOriginIdentIndex,
                                  true /* indexOK */ ,
                                  &SnapshotDirty,
                                  1, &key);
 
        collides = HeapTupleIsValid(systable_getnext(scan));
 
        systable_endscan(scan);
 
        if (!collides)
        {
            /*
             * Ok, found an unused roident, insert the new row and do a CCI,
             * so our callers can look it up if they want to.
             */
            memset(&nulls, 0, sizeof(nulls));
 
            values[Anum_pg_replication_origin_roident - 1] = ObjectIdGetDatum(roident);
            values[Anum_pg_replication_origin_roname - 1] = roname_d;
 
            tuple = heap_form_tuple(RelationGetDescr(rel), values, nulls);
            CatalogTupleInsert(rel, tuple);
            CommandCounterIncrement();
            break;
        }
    }
 
    /* now release lock again,  */
    table_close(rel, ExclusiveLock);
 
    if (tuple == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                 errmsg("could not find free replication origin ID")));
 
    heap_freetuple(tuple);
    return roident;
}
 
/*
 * Helper function to drop a replication origin.
 */
static void
replorigin_state_clear(RepOriginId roident, bool nowait)
{
    int         i;
 
    /*
     * Clean up the slot state info, if there is any matching slot.
     */
restart:
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    for (i = 0; i < max_replication_slots; i++)
    {
        ReplicationState *state = &replication_states[i];
 
        if (state->roident == roident)
        {
            /* found our slot, is it busy? */
            if (state->acquired_by != 0)
            {
                ConditionVariable *cv;
 
                if (nowait)
                    ereport(ERROR,
                            (errcode(ERRCODE_OBJECT_IN_USE),
                             errmsg("could not drop replication origin with ID %d, in use by PID %d",
                                    state->roident,
                                    state->acquired_by)));
 
                /*
                 * We must wait and then retry.  Since we don't know which CV
                 * to wait on until here, we can't readily use
                 * ConditionVariablePrepareToSleep (calling it here would be
                 * wrong, since we could miss the signal if we did so); just
                 * use ConditionVariableSleep directly.
                 */
                cv = &state->origin_cv;
 
                LWLockRelease(ReplicationOriginLock);
 
                ConditionVariableSleep(cv, WAIT_EVENT_REPLICATION_ORIGIN_DROP);
                goto restart;
            }
 
            /* first make a WAL log entry */
            {
                xl_replorigin_drop xlrec;
 
                xlrec.node_id = roident;
                XLogBeginInsert();
                XLogRegisterData((char *) (&xlrec), sizeof(xlrec));
                XLogInsert(RM_REPLORIGIN_ID, XLOG_REPLORIGIN_DROP);
            }
 
            /* then clear the in-memory slot */
            state->roident = InvalidRepOriginId;
            state->remote_lsn = InvalidXLogRecPtr;
            state->local_lsn = InvalidXLogRecPtr;
            break;
        }
    }
    LWLockRelease(ReplicationOriginLock);
    ConditionVariableCancelSleep();
}
 
/*
 * Drop replication origin (by name).
 *
 * Needs to be called in a transaction.
 */
void
replorigin_drop_by_name(const char *name, bool missing_ok, bool nowait)
{
    RepOriginId roident;
    Relation    rel;
    HeapTuple   tuple;
 
    Assert(IsTransactionState());
 
    rel = table_open(ReplicationOriginRelationId, RowExclusiveLock);
 
    roident = replorigin_by_name(name, missing_ok);
 
    /* Lock the origin to prevent concurrent drops. */
    LockSharedObject(ReplicationOriginRelationId, roident, 0,
                     AccessExclusiveLock);
 
    tuple = SearchSysCache1(REPLORIGIDENT, ObjectIdGetDatum(roident));
    if (!HeapTupleIsValid(tuple))
    {
        if (!missing_ok)
            elog(ERROR, "cache lookup failed for replication origin with ID %d",
                 roident);
 
        /*
         * We don't need to retain the locks if the origin is already dropped.
         */
        UnlockSharedObject(ReplicationOriginRelationId, roident, 0,
                           AccessExclusiveLock);
        table_close(rel, RowExclusiveLock);
        return;
    }
 
    replorigin_state_clear(roident, nowait);
 
    /*
     * Now, we can delete the catalog entry.
     */
    CatalogTupleDelete(rel, &tuple->t_self);
    ReleaseSysCache(tuple);
 
    CommandCounterIncrement();
 
    /* We keep the lock on pg_replication_origin until commit */
    table_close(rel, NoLock);
}
 
/*
 * Lookup replication origin via its oid and return the name.
 *
 * The external name is palloc'd in the calling context.
 *
 * Returns true if the origin is known, false otherwise.
 */
bool
replorigin_by_oid(RepOriginId roident, bool missing_ok, char **roname)
{
    HeapTuple   tuple;
    Form_pg_replication_origin ric;
 
    Assert(OidIsValid((Oid) roident));
    Assert(roident != InvalidRepOriginId);
    Assert(roident != DoNotReplicateId);
 
    tuple = SearchSysCache1(REPLORIGIDENT,
                            ObjectIdGetDatum((Oid) roident));
 
    if (HeapTupleIsValid(tuple))
    {
        ric = (Form_pg_replication_origin) GETSTRUCT(tuple);
        *roname = text_to_cstring(&ric->roname);
        ReleaseSysCache(tuple);
 
        return true;
    }
    else
    {
        *roname = NULL;
 
        if (!missing_ok)
            ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_OBJECT),
                     errmsg("replication origin with ID %d does not exist",
                            roident)));
 
        return false;
    }
}
 
 
/* ---------------------------------------------------------------------------
 * Functions for handling replication progress.
 * ---------------------------------------------------------------------------
 */
 
Size
ReplicationOriginShmemSize(void)
{
    Size        size = 0;
 
    /*
     * XXX: max_replication_slots is arguably the wrong thing to use, as here
     * we keep the replay state of *remote* transactions. But for now it seems
     * sufficient to reuse it, rather than introduce a separate GUC.
     */
    if (max_replication_slots == 0)
        return size;
 
    size = add_size(size, offsetof(ReplicationStateCtl, states));
 
    size = add_size(size,
                    mul_size(max_replication_slots, sizeof(ReplicationState)));
    return size;
}
 
void
ReplicationOriginShmemInit(void)
{
    bool        found;
 
    if (max_replication_slots == 0)
        return;
 
    replication_states_ctl = (ReplicationStateCtl *)
        ShmemInitStruct("ReplicationOriginState",
                        ReplicationOriginShmemSize(),
                        &found);
    replication_states = replication_states_ctl->states;
 
    if (!found)
    {
        int         i;
 
        MemSet(replication_states_ctl, 0, ReplicationOriginShmemSize());
 
        replication_states_ctl->tranche_id = LWTRANCHE_REPLICATION_ORIGIN_STATE;
 
        for (i = 0; i < max_replication_slots; i++)
        {
            LWLockInitialize(&replication_states[i].lock,
                             replication_states_ctl->tranche_id);
            ConditionVariableInit(&replication_states[i].origin_cv);
        }
    }
}
 
/* ---------------------------------------------------------------------------
 * Perform a checkpoint of each replication origin's progress with respect to
 * the replayed remote_lsn. Make sure that all transactions we refer to in the
 * checkpoint (local_lsn) are actually on-disk. This might not yet be the case
 * if the transactions were originally committed asynchronously.
 *
 * We store checkpoints in the following format:
 * +-------+------------------------+------------------+-----+--------+
 * | MAGIC | ReplicationStateOnDisk | struct Replic... | ... | CRC32C | EOF
 * +-------+------------------------+------------------+-----+--------+
 *
 * So its just the magic, followed by the statically sized
 * ReplicationStateOnDisk structs. Note that the maximum number of
 * ReplicationState is determined by max_replication_slots.
 * ---------------------------------------------------------------------------
 */
void
CheckPointReplicationOrigin(void)
{
    const char *tmppath = PG_REPLORIGIN_CHECKPOINT_TMPFILE;
    const char *path = PG_REPLORIGIN_CHECKPOINT_FILENAME;
    int         tmpfd;
    int         i;
    uint32      magic = REPLICATION_STATE_MAGIC;
    pg_crc32c   crc;
 
    if (max_replication_slots == 0)
        return;
 
    INIT_CRC32C(crc);
 
    /* make sure no old temp file is remaining */
    if (unlink(tmppath) < 0 && errno != ENOENT)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not remove file \"%s\": %m",
                        tmppath)));
 
    /*
     * no other backend can perform this at the same time; only one checkpoint
     * can happen at a time.
     */
    tmpfd = OpenTransientFile(tmppath,
                              O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
    if (tmpfd < 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not create file \"%s\": %m",
                        tmppath)));
 
    /* write magic */
    errno = 0;
    if ((write(tmpfd, &magic, sizeof(magic))) != sizeof(magic))
    {
        /* if write didn't set errno, assume problem is no disk space */
        if (errno == 0)
            errno = ENOSPC;
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not write to file \"%s\": %m",
                        tmppath)));
    }
    COMP_CRC32C(crc, &magic, sizeof(magic));
 
    /* prevent concurrent creations/drops */
    LWLockAcquire(ReplicationOriginLock, LW_SHARED);
 
    /* write actual data */
    for (i = 0; i < max_replication_slots; i++)
    {
        ReplicationStateOnDisk disk_state;
        ReplicationState *curstate = &replication_states[i];
        XLogRecPtr  local_lsn;
 
        if (curstate->roident == InvalidRepOriginId)
            continue;
 
        /* zero, to avoid uninitialized padding bytes */
        memset(&disk_state, 0, sizeof(disk_state));
 
        LWLockAcquire(&curstate->lock, LW_SHARED);
 
        disk_state.roident = curstate->roident;
 
        disk_state.remote_lsn = curstate->remote_lsn;
        local_lsn = curstate->local_lsn;
 
        LWLockRelease(&curstate->lock);
 
        /* make sure we only write out a commit that's persistent */
        XLogFlush(local_lsn);
 
        errno = 0;
        if ((write(tmpfd, &disk_state, sizeof(disk_state))) !=
            sizeof(disk_state))
        {
            /* if write didn't set errno, assume problem is no disk space */
            if (errno == 0)
                errno = ENOSPC;
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not write to file \"%s\": %m",
                            tmppath)));
        }
 
        COMP_CRC32C(crc, &disk_state, sizeof(disk_state));
    }
 
    LWLockRelease(ReplicationOriginLock);
 
    /* write out the CRC */
    FIN_CRC32C(crc);
    errno = 0;
    if ((write(tmpfd, &crc, sizeof(crc))) != sizeof(crc))
    {
        /* if write didn't set errno, assume problem is no disk space */
        if (errno == 0)
            errno = ENOSPC;
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not write to file \"%s\": %m",
                        tmppath)));
    }
 
    if (CloseTransientFile(tmpfd) != 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m",
                        tmppath)));
 
    /* fsync, rename to permanent file, fsync file and directory */
    durable_rename(tmppath, path, PANIC);
}
 
/*
 * Recover replication replay status from checkpoint data saved earlier by
 * CheckPointReplicationOrigin.
 *
 * This only needs to be called at startup and *not* during every checkpoint
 * read during recovery (e.g. in HS or PITR from a base backup) afterwards. All
 * state thereafter can be recovered by looking at commit records.
 */
void
StartupReplicationOrigin(void)
{
    const char *path = PG_REPLORIGIN_CHECKPOINT_FILENAME;
    int         fd;
    int         readBytes;
    uint32      magic = REPLICATION_STATE_MAGIC;
    int         last_state = 0;
    pg_crc32c   file_crc;
    pg_crc32c   crc;
 
    /* don't want to overwrite already existing state */
#ifdef USE_ASSERT_CHECKING
    static bool already_started = false;
 
    Assert(!already_started);
    already_started = true;
#endif
 
    if (max_replication_slots == 0)
        return;
 
    INIT_CRC32C(crc);
 
    elog(DEBUG2, "starting up replication origin progress state");
 
    fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
 
    /*
     * might have had max_replication_slots == 0 last run, or we just brought
     * up a standby.
     */
    if (fd < 0 && errno == ENOENT)
        return;
    else if (fd < 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m",
                        path)));
 
    /* verify magic, that is written even if nothing was active */
    readBytes = read(fd, &magic, sizeof(magic));
    if (readBytes != sizeof(magic))
    {
        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, sizeof(magic))));
    }
    COMP_CRC32C(crc, &magic, sizeof(magic));
 
    if (magic != REPLICATION_STATE_MAGIC)
        ereport(PANIC,
                (errmsg("replication checkpoint has wrong magic %u instead of %u",
                        magic, REPLICATION_STATE_MAGIC)));
 
    /* we can skip locking here, no other access is possible */
 
    /* recover individual states, until there are no more to be found */
    while (true)
    {
        ReplicationStateOnDisk disk_state;
 
        readBytes = read(fd, &disk_state, sizeof(disk_state));
 
        /* no further data */
        if (readBytes == sizeof(crc))
        {
            /* not pretty, but simple ... */
            file_crc = *(pg_crc32c *) &disk_state;
            break;
        }
 
        if (readBytes < 0)
        {
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": %m",
                            path)));
        }
 
        if (readBytes != sizeof(disk_state))
        {
            ereport(PANIC,
                    (errcode_for_file_access(),
                     errmsg("could not read file \"%s\": read %d of %zu",
                            path, readBytes, sizeof(disk_state))));
        }
 
        COMP_CRC32C(crc, &disk_state, sizeof(disk_state));
 
        if (last_state == max_replication_slots)
            ereport(PANIC,
                    (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                     errmsg("could not find free replication state, increase \"max_replication_slots\"")));
 
        /* copy data to shared memory */
        replication_states[last_state].roident = disk_state.roident;
        replication_states[last_state].remote_lsn = disk_state.remote_lsn;
        last_state++;
 
        ereport(LOG,
                (errmsg("recovered replication state of node %d to %X/%X",
                        disk_state.roident,
                        LSN_FORMAT_ARGS(disk_state.remote_lsn))));
    }
 
    /* now check checksum */
    FIN_CRC32C(crc);
    if (file_crc != crc)
        ereport(PANIC,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("replication slot checkpoint has wrong checksum %u, expected %u",
                        crc, file_crc)));
 
    if (CloseTransientFile(fd) != 0)
        ereport(PANIC,
                (errcode_for_file_access(),
                 errmsg("could not close file \"%s\": %m",
                        path)));
}
 
void
replorigin_redo(XLogReaderState *record)
{
    uint8       info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 
    switch (info)
    {
        case XLOG_REPLORIGIN_SET:
            {
                xl_replorigin_set *xlrec =
                    (xl_replorigin_set *) XLogRecGetData(record);
 
                replorigin_advance(xlrec->node_id,
                                   xlrec->remote_lsn, record->EndRecPtr,
                                   xlrec->force /* backward */ ,
                                   false /* WAL log */ );
                break;
            }
        case XLOG_REPLORIGIN_DROP:
            {
                xl_replorigin_drop *xlrec;
                int         i;
 
                xlrec = (xl_replorigin_drop *) XLogRecGetData(record);
 
                for (i = 0; i < max_replication_slots; i++)
                {
                    ReplicationState *state = &replication_states[i];
 
                    /* found our slot */
                    if (state->roident == xlrec->node_id)
                    {
                        /* reset entry */
                        state->roident = InvalidRepOriginId;
                        state->remote_lsn = InvalidXLogRecPtr;
                        state->local_lsn = InvalidXLogRecPtr;
                        break;
                    }
                }
                break;
            }
        default:
            elog(PANIC, "replorigin_redo: unknown op code %u", info);
    }
}
 
 
/*
 * Tell the replication origin progress machinery that a commit from 'node'
 * that originated at the LSN remote_commit on the remote node was replayed
 * successfully and that we don't need to do so again. In combination with
 * setting up replorigin_session_origin_lsn and replorigin_session_origin
 * that ensures we won't lose knowledge about that after a crash if the
 * transaction had a persistent effect (think of asynchronous commits).
 *
 * local_commit needs to be a local LSN of the commit so that we can make sure
 * upon a checkpoint that enough WAL has been persisted to disk.
 *
 * Needs to be called with a RowExclusiveLock on pg_replication_origin,
 * unless running in recovery.
 */
void
replorigin_advance(RepOriginId node,
                   XLogRecPtr remote_commit, XLogRecPtr local_commit,
                   bool go_backward, bool wal_log)
{
    int         i;
    ReplicationState *replication_state = NULL;
    ReplicationState *free_state = NULL;
 
    Assert(node != InvalidRepOriginId);
 
    /* we don't track DoNotReplicateId */
    if (node == DoNotReplicateId)
        return;
 
    /*
     * XXX: For the case where this is called by WAL replay, it'd be more
     * efficient to restore into a backend local hashtable and only dump into
     * shmem after recovery is finished. Let's wait with implementing that
     * till it's shown to be a measurable expense
     */
 
    /* Lock exclusively, as we may have to create a new table entry. */
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    /*
     * Search for either an existing slot for the origin, or a free one we can
     * use.
     */
    for (i = 0; i < max_replication_slots; i++)
    {
        ReplicationState *curstate = &replication_states[i];
 
        /* remember where to insert if necessary */
        if (curstate->roident == InvalidRepOriginId &&
            free_state == NULL)
        {
            free_state = curstate;
            continue;
        }
 
        /* not our slot */
        if (curstate->roident != node)
        {
            continue;
        }
 
        /* ok, found slot */
        replication_state = curstate;
 
        LWLockAcquire(&replication_state->lock, LW_EXCLUSIVE);
 
        /* Make sure it's not used by somebody else */
        if (replication_state->acquired_by != 0)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_IN_USE),
                     errmsg("replication origin with ID %d is already active for PID %d",
                            replication_state->roident,
                            replication_state->acquired_by)));
        }
 
        break;
    }
 
    if (replication_state == NULL && free_state == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                 errmsg("could not find free replication state slot for replication origin with ID %d",
                        node),
                 errhint("Increase \"max_replication_slots\" and try again.")));
 
    if (replication_state == NULL)
    {
        /* initialize new slot */
        LWLockAcquire(&free_state->lock, LW_EXCLUSIVE);
        replication_state = free_state;
        Assert(replication_state->remote_lsn == InvalidXLogRecPtr);
        Assert(replication_state->local_lsn == InvalidXLogRecPtr);
        replication_state->roident = node;
    }
 
    Assert(replication_state->roident != InvalidRepOriginId);
 
    /*
     * If somebody "forcefully" sets this slot, WAL log it, so it's durable
     * and the standby gets the message. Primarily this will be called during
     * WAL replay (of commit records) where no WAL logging is necessary.
     */
    if (wal_log)
    {
        xl_replorigin_set xlrec;
 
        xlrec.remote_lsn = remote_commit;
        xlrec.node_id = node;
        xlrec.force = go_backward;
 
        XLogBeginInsert();
        XLogRegisterData((char *) (&xlrec), sizeof(xlrec));
 
        XLogInsert(RM_REPLORIGIN_ID, XLOG_REPLORIGIN_SET);
    }
 
    /*
     * Due to - harmless - race conditions during a checkpoint we could see
     * values here that are older than the ones we already have in memory. We
     * could also see older values for prepared transactions when the prepare
     * is sent at a later point of time along with commit prepared and there
     * are other transactions commits between prepare and commit prepared. See
     * ReorderBufferFinishPrepared. Don't overwrite those.
     */
    if (go_backward || replication_state->remote_lsn < remote_commit)
        replication_state->remote_lsn = remote_commit;
    if (local_commit != InvalidXLogRecPtr &&
        (go_backward || replication_state->local_lsn < local_commit))
        replication_state->local_lsn = local_commit;
    LWLockRelease(&replication_state->lock);
 
    /*
     * Release *after* changing the LSNs, slot isn't acquired and thus could
     * otherwise be dropped anytime.
     */
    LWLockRelease(ReplicationOriginLock);
}
 
 
XLogRecPtr
replorigin_get_progress(RepOriginId node, bool flush)
{
    int         i;
    XLogRecPtr  local_lsn = InvalidXLogRecPtr;
    XLogRecPtr  remote_lsn = InvalidXLogRecPtr;
 
    /* prevent slots from being concurrently dropped */
    LWLockAcquire(ReplicationOriginLock, LW_SHARED);
 
    for (i = 0; i < max_replication_slots; i++)
    {
        ReplicationState *state;
 
        state = &replication_states[i];
 
        if (state->roident == node)
        {
            LWLockAcquire(&state->lock, LW_SHARED);
 
            remote_lsn = state->remote_lsn;
            local_lsn = state->local_lsn;
 
            LWLockRelease(&state->lock);
 
            break;
        }
    }
 
    LWLockRelease(ReplicationOriginLock);
 
    if (flush && local_lsn != InvalidXLogRecPtr)
        XLogFlush(local_lsn);
 
    return remote_lsn;
}
 
/*
 * Tear down a (possibly) configured session replication origin during process
 * exit.
 */
static void
ReplicationOriginExitCleanup(int code, Datum arg)
{
    ConditionVariable *cv = NULL;
 
    if (session_replication_state == NULL)
        return;
 
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    if (session_replication_state->acquired_by == MyProcPid)
    {
        cv = &session_replication_state->origin_cv;
 
        session_replication_state->acquired_by = 0;
        session_replication_state = NULL;
    }
 
    LWLockRelease(ReplicationOriginLock);
 
    if (cv)
        ConditionVariableBroadcast(cv);
}
 
/*
 * Setup a replication origin in the shared memory struct if it doesn't
 * already exist and cache access to the specific ReplicationSlot so the
 * array doesn't have to be searched when calling
 * replorigin_session_advance().
 *
 * Normally only one such cached origin can exist per process so the cached
 * value can only be set again after the previous value is torn down with
 * replorigin_session_reset(). For this normal case pass acquired_by = 0
 * (meaning the slot is not allowed to be already acquired by another process).
 *
 * However, sometimes multiple processes can safely re-use the same origin slot
 * (for example, multiple parallel apply processes can safely use the same
 * origin, provided they maintain commit order by allowing only one process to
 * commit at a time). For this case the first process must pass acquired_by =
 * 0, and then the other processes sharing that same origin can pass
 * acquired_by = PID of the first process.
 */
void
replorigin_session_setup(RepOriginId node, int acquired_by)
{
    static bool registered_cleanup;
    int         i;
    int         free_slot = -1;
 
    if (!registered_cleanup)
    {
        on_shmem_exit(ReplicationOriginExitCleanup, 0);
        registered_cleanup = true;
    }
 
    Assert(max_replication_slots > 0);
 
    if (session_replication_state != NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("cannot setup replication origin when one is already setup")));
 
    /* Lock exclusively, as we may have to create a new table entry. */
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    /*
     * Search for either an existing slot for the origin, or a free one we can
     * use.
     */
    for (i = 0; i < max_replication_slots; i++)
    {
        ReplicationState *curstate = &replication_states[i];
 
        /* remember where to insert if necessary */
        if (curstate->roident == InvalidRepOriginId &&
            free_slot == -1)
        {
            free_slot = i;
            continue;
        }
 
        /* not our slot */
        if (curstate->roident != node)
            continue;
 
        else if (curstate->acquired_by != 0 && acquired_by == 0)
        {
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_IN_USE),
                     errmsg("replication origin with ID %d is already active for PID %d",
                            curstate->roident, curstate->acquired_by)));
        }
 
        /* ok, found slot */
        session_replication_state = curstate;
        break;
    }
 
 
    if (session_replication_state == NULL && free_slot == -1)
        ereport(ERROR,
                (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                 errmsg("could not find free replication state slot for replication origin with ID %d",
                        node),
                 errhint("Increase \"max_replication_slots\" and try again.")));
    else if (session_replication_state == NULL)
    {
        /* initialize new slot */
        session_replication_state = &replication_states[free_slot];
        Assert(session_replication_state->remote_lsn == InvalidXLogRecPtr);
        Assert(session_replication_state->local_lsn == InvalidXLogRecPtr);
        session_replication_state->roident = node;
    }
 
 
    Assert(session_replication_state->roident != InvalidRepOriginId);
 
    if (acquired_by == 0)
        session_replication_state->acquired_by = MyProcPid;
    else if (session_replication_state->acquired_by != acquired_by)
        elog(ERROR, "could not find replication state slot for replication origin with OID %u which was acquired by %d",
             node, acquired_by);
 
    LWLockRelease(ReplicationOriginLock);
 
    /* probably this one is pointless */
    ConditionVariableBroadcast(&session_replication_state->origin_cv);
}
 
/*
 * Reset replay state previously setup in this session.
 *
 * This function may only be called if an origin was setup with
 * replorigin_session_setup().
 */
void
replorigin_session_reset(void)
{
    ConditionVariable *cv;
 
    Assert(max_replication_slots != 0);
 
    if (session_replication_state == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("no replication origin is configured")));
 
    LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);
 
    session_replication_state->acquired_by = 0;
    cv = &session_replication_state->origin_cv;
    session_replication_state = NULL;
 
    LWLockRelease(ReplicationOriginLock);
 
    ConditionVariableBroadcast(cv);
}
 
/*
 * Do the same work replorigin_advance() does, just on the session's
 * configured origin.
 *
 * This is noticeably cheaper than using replorigin_advance().
 */
void
replorigin_session_advance(XLogRecPtr remote_commit, XLogRecPtr local_commit)
{
    Assert(session_replication_state != NULL);
    Assert(session_replication_state->roident != InvalidRepOriginId);
 
    LWLockAcquire(&session_replication_state->lock, LW_EXCLUSIVE);
    if (session_replication_state->local_lsn < local_commit)
        session_replication_state->local_lsn = local_commit;
    if (session_replication_state->remote_lsn < remote_commit)
        session_replication_state->remote_lsn = remote_commit;
    LWLockRelease(&session_replication_state->lock);
}
 
/*
 * Ask the machinery about the point up to which we successfully replayed
 * changes from an already setup replication origin.
 */
XLogRecPtr
replorigin_session_get_progress(bool flush)
{
    XLogRecPtr  remote_lsn;
    XLogRecPtr  local_lsn;
 
    Assert(session_replication_state != NULL);
 
    LWLockAcquire(&session_replication_state->lock, LW_SHARED);
    remote_lsn = session_replication_state->remote_lsn;
    local_lsn = session_replication_state->local_lsn;
    LWLockRelease(&session_replication_state->lock);
 
    if (flush && local_lsn != InvalidXLogRecPtr)
        XLogFlush(local_lsn);
 
    return remote_lsn;
}
 
 
 
/* ---------------------------------------------------------------------------
 * SQL functions for working with replication origin.
 *
 * These mostly should be fairly short wrappers around more generic functions.
 * ---------------------------------------------------------------------------
 */
 
/*
 * Create replication origin for the passed in name, and return the assigned
 * oid.
 */
Datum
pg_replication_origin_create(PG_FUNCTION_ARGS)
{
    char       *name;
    RepOriginId roident;
 
    replorigin_check_prerequisites(false, false);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
 
    /*
     * Replication origins "any and "none" are reserved for system options.
     * The origins "pg_xxx" are reserved for internal use.
     */
    if (IsReservedName(name) || IsReservedOriginName(name))
        ereport(ERROR,
                (errcode(ERRCODE_RESERVED_NAME),
                 errmsg("replication origin name \"%s\" is reserved",
                        name),
                 errdetail("Origin names \"%s\", \"%s\", and names starting with \"pg_\" are reserved.",
                           LOGICALREP_ORIGIN_ANY, LOGICALREP_ORIGIN_NONE)));
 
    /*
     * If built with appropriate switch, whine when regression-testing
     * conventions for replication origin names are violated.
     */
#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
    if (strncmp(name, "regress_", 8) != 0)
        elog(WARNING, "replication origins created by regression test cases should have names starting with \"regress_\"");
#endif
 
    roident = replorigin_create(name);
 
    pfree(name);
 
    PG_RETURN_OID(roident);
}
 
/*
 * Drop replication origin.
 */
Datum
pg_replication_origin_drop(PG_FUNCTION_ARGS)
{
    char       *name;
 
    replorigin_check_prerequisites(false, false);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
 
    replorigin_drop_by_name(name, false, true);
 
    pfree(name);
 
    PG_RETURN_VOID();
}
 
/*
 * Return oid of a replication origin.
 */
Datum
pg_replication_origin_oid(PG_FUNCTION_ARGS)
{
    char       *name;
    RepOriginId roident;
 
    replorigin_check_prerequisites(false, false);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    roident = replorigin_by_name(name, true);
 
    pfree(name);
 
    if (OidIsValid(roident))
        PG_RETURN_OID(roident);
    PG_RETURN_NULL();
}
 
/*
 * Setup a replication origin for this session.
 */
Datum
pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
{
    char       *name;
    RepOriginId origin;
 
    replorigin_check_prerequisites(true, false);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    origin = replorigin_by_name(name, false);
    replorigin_session_setup(origin, 0);
 
    replorigin_session_origin = origin;
 
    pfree(name);
 
    PG_RETURN_VOID();
}
 
/*
 * Reset previously setup origin in this session
 */
Datum
pg_replication_origin_session_reset(PG_FUNCTION_ARGS)
{
    replorigin_check_prerequisites(true, false);
 
    replorigin_session_reset();
 
    replorigin_session_origin = InvalidRepOriginId;
    replorigin_session_origin_lsn = InvalidXLogRecPtr;
    replorigin_session_origin_timestamp = 0;
 
    PG_RETURN_VOID();
}
 
/*
 * Has a replication origin been setup for this session.
 */
Datum
pg_replication_origin_session_is_setup(PG_FUNCTION_ARGS)
{
    replorigin_check_prerequisites(false, false);
 
    PG_RETURN_BOOL(replorigin_session_origin != InvalidRepOriginId);
}
 
 
/*
 * Return the replication progress for origin setup in the current session.
 *
 * If 'flush' is set to true it is ensured that the returned value corresponds
 * to a local transaction that has been flushed. This is useful if asynchronous
 * commits are used when replaying replicated transactions.
 */
Datum
pg_replication_origin_session_progress(PG_FUNCTION_ARGS)
{
    XLogRecPtr  remote_lsn = InvalidXLogRecPtr;
    bool        flush = PG_GETARG_BOOL(0);
 
    replorigin_check_prerequisites(true, false);
 
    if (session_replication_state == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("no replication origin is configured")));
 
    remote_lsn = replorigin_session_get_progress(flush);
 
    if (remote_lsn == InvalidXLogRecPtr)
        PG_RETURN_NULL();
 
    PG_RETURN_LSN(remote_lsn);
}
 
Datum
pg_replication_origin_xact_setup(PG_FUNCTION_ARGS)
{
    XLogRecPtr  location = PG_GETARG_LSN(0);
 
    replorigin_check_prerequisites(true, false);
 
    if (session_replication_state == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("no replication origin is configured")));
 
    replorigin_session_origin_lsn = location;
    replorigin_session_origin_timestamp = PG_GETARG_TIMESTAMPTZ(1);
 
    PG_RETURN_VOID();
}
 
Datum
pg_replication_origin_xact_reset(PG_FUNCTION_ARGS)
{
    replorigin_check_prerequisites(true, false);
 
    replorigin_session_origin_lsn = InvalidXLogRecPtr;
    replorigin_session_origin_timestamp = 0;
 
    PG_RETURN_VOID();
}
 
 
Datum
pg_replication_origin_advance(PG_FUNCTION_ARGS)
{
    text       *name = PG_GETARG_TEXT_PP(0);
    XLogRecPtr  remote_commit = PG_GETARG_LSN(1);
    RepOriginId node;
 
    replorigin_check_prerequisites(true, false);
 
    /* lock to prevent the replication origin from vanishing */
    LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
 
    node = replorigin_by_name(text_to_cstring(name), false);
 
    /*
     * Can't sensibly pass a local commit to be flushed at checkpoint - this
     * xact hasn't committed yet. This is why this function should be used to
     * set up the initial replication state, but not for replay.
     */
    replorigin_advance(node, remote_commit, InvalidXLogRecPtr,
                       true /* go backward */ , true /* WAL log */ );
 
    UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
 
    PG_RETURN_VOID();
}
 
 
/*
 * Return the replication progress for an individual replication origin.
 *
 * If 'flush' is set to true it is ensured that the returned value corresponds
 * to a local transaction that has been flushed. This is useful if asynchronous
 * commits are used when replaying replicated transactions.
 */
Datum
pg_replication_origin_progress(PG_FUNCTION_ARGS)
{
    char       *name;
    bool        flush;
    RepOriginId roident;
    XLogRecPtr  remote_lsn = InvalidXLogRecPtr;
 
    replorigin_check_prerequisites(true, true);
 
    name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
    flush = PG_GETARG_BOOL(1);
 
    roident = replorigin_by_name(name, false);
    Assert(OidIsValid(roident));
 
    remote_lsn = replorigin_get_progress(roident, flush);
 
    if (remote_lsn == InvalidXLogRecPtr)
        PG_RETURN_NULL();
 
    PG_RETURN_LSN(remote_lsn);
}
 
 
Datum
pg_show_replication_origin_status(PG_FUNCTION_ARGS)
{
    ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
    int         i;
#define REPLICATION_ORIGIN_PROGRESS_COLS 4
 
    /* we want to return 0 rows if slot is set to zero */
    replorigin_check_prerequisites(false, true);
 
    InitMaterializedSRF(fcinfo, 0);
 
    /* prevent slots from being concurrently dropped */
    LWLockAcquire(ReplicationOriginLock, LW_SHARED);
 
    /*
     * Iterate through all possible replication_states, display if they are
     * filled. Note that we do not take any locks, so slightly corrupted/out
     * of date values are a possibility.
     */
    for (i = 0; i < max_replication_slots; i++)
    {
        ReplicationState *state;
        Datum       values[REPLICATION_ORIGIN_PROGRESS_COLS];
        bool        nulls[REPLICATION_ORIGIN_PROGRESS_COLS];
        char       *roname;
 
        state = &replication_states[i];
 
        /* unused slot, nothing to display */
        if (state->roident == InvalidRepOriginId)
            continue;
 
        memset(values, 0, sizeof(values));
        memset(nulls, 1, sizeof(nulls));
 
        values[0] = ObjectIdGetDatum(state->roident);
        nulls[0] = false;
 
        /*
         * We're not preventing the origin to be dropped concurrently, so
         * silently accept that it might be gone.
         */
        if (replorigin_by_oid(state->roident, true,
                              &roname))
        {
            values[1] = CStringGetTextDatum(roname);
            nulls[1] = false;
        }
 
        LWLockAcquire(&state->lock, LW_SHARED);
 
        values[2] = LSNGetDatum(state->remote_lsn);
        nulls[2] = false;
 
        values[3] = LSNGetDatum(state->local_lsn);
        nulls[3] = false;
 
        LWLockRelease(&state->lock);
 
        tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
                             values, nulls);
    }
 
    LWLockRelease(ReplicationOriginLock);
 
#undef REPLICATION_ORIGIN_PROGRESS_COLS
 
    return (Datum) 0;
}