/src/postgres/src/backend/replication/logical/snapbuild.c

Source
/*-------------------------------------------------------------------------
 *
 * snapbuild.c
 *
 *    Infrastructure for building historic catalog snapshots based on contents
 *    of the WAL, for the purpose of decoding heapam.c style values in the
 *    WAL.
 *
 * NOTES:
 *
 * We build snapshots which can *only* be used to read catalog contents and we
 * do so by reading and interpreting the WAL stream. The aim is to build a
 * snapshot that behaves the same as a freshly taken MVCC snapshot would have
 * at the time the XLogRecord was generated.
 *
 * To build the snapshots we reuse the infrastructure built for Hot
 * Standby. The in-memory snapshots we build look different than HS' because
 * we have different needs. To successfully decode data from the WAL we only
 * need to access catalog tables and (sys|rel|cat)cache, not the actual user
 * tables since the data we decode is wholly contained in the WAL
 * records. Also, our snapshots need to be different in comparison to normal
 * MVCC ones because in contrast to those we cannot fully rely on the clog and
 * pg_subtrans for information about committed transactions because they might
 * commit in the future from the POV of the WAL entry we're currently
 * decoding. This definition has the advantage that we only need to prevent
 * removal of catalog rows, while normal table's rows can still be
 * removed. This is achieved by using the replication slot mechanism.
 *
 * As the percentage of transactions modifying the catalog normally is fairly
 * small in comparisons to ones only manipulating user data, we keep track of
 * the committed catalog modifying ones inside [xmin, xmax) instead of keeping
 * track of all running transactions like it's done in a normal snapshot. Note
 * that we're generally only looking at transactions that have acquired an
 * xid. That is we keep a list of transactions between snapshot->(xmin, xmax)
 * that we consider committed, everything else is considered aborted/in
 * progress. That also allows us not to care about subtransactions before they
 * have committed which means this module, in contrast to HS, doesn't have to
 * care about suboverflowed subtransactions and similar.
 *
 * One complexity of doing this is that to e.g. handle mixed DDL/DML
 * transactions we need Snapshots that see intermediate versions of the
 * catalog in a transaction. During normal operation this is achieved by using
 * CommandIds/cmin/cmax. The problem with that however is that for space
 * efficiency reasons, the cmin and cmax are not included in WAL records. We
 * cannot read the cmin/cmax from the tuple itself, either, because it is
 * reset on crash recovery. Even if we could, we could not decode combocids
 * which are only tracked in the original backend's memory. To work around
 * that, heapam writes an extra WAL record (XLOG_HEAP2_NEW_CID) every time a
 * catalog row is modified, which includes the cmin and cmax of the
 * tuple. During decoding, we insert the ctid->(cmin,cmax) mappings into the
 * reorder buffer, and use them at visibility checks instead of the cmin/cmax
 * on the tuple itself. Check the reorderbuffer.c's comment above
 * ResolveCminCmaxDuringDecoding() for details.
 *
 * To facilitate all this we need our own visibility routine, as the normal
 * ones are optimized for different usecases.
 *
 * To replace the normal catalog snapshots with decoding ones use the
 * SetupHistoricSnapshot() and TeardownHistoricSnapshot() functions.
 *
 *
 *
 * The snapbuild machinery is starting up in several stages, as illustrated
 * by the following graph describing the SnapBuild->state transitions:
 *
 *       +-------------------------+
 *    +----|     START       |-------------+
 *    |    +-------------------------+         |
 *    |         |              |
 *    |         |              |
 *    |      running_xacts #1            |
 *    |         |              |
 *    |         |              |
 *    |         v              |
 *    |    +-------------------------+         v
 *    |    |   BUILDING_SNAPSHOT   |------------>|
 *    |    +-------------------------+         |
 *    |         |              |
 *    |         |              |
 *    | running_xacts #2, xacts from #1 finished   |
 *    |         |              |
 *    |         |              |
 *    |         v              |
 *    |    +-------------------------+         v
 *    |    |     FULL_SNAPSHOT   |------------>|
 *    |    +-------------------------+         |
 *    |         |              |
 * running_xacts    |            saved snapshot
 * with zero xacts    |         at running_xacts's lsn
 *    |         |              |
 *    | running_xacts with xacts from #2 finished  |
 *    |         |              |
 *    |         v              |
 *    |    +-------------------------+         |
 *    +--->|SNAPBUILD_CONSISTENT   |<------------+
 *       +-------------------------+
 *
 * Initially the machinery is in the START stage. When an xl_running_xacts
 * record is read that is sufficiently new (above the safe xmin horizon),
 * there's a state transition. If there were no running xacts when the
 * xl_running_xacts record was generated, we'll directly go into CONSISTENT
 * state, otherwise we'll switch to the BUILDING_SNAPSHOT state. Having a full
 * snapshot means that all transactions that start henceforth can be decoded
 * in their entirety, but transactions that started previously can't. In
 * FULL_SNAPSHOT we'll switch into CONSISTENT once all those previously
 * running transactions have committed or aborted.
 *
 * Only transactions that commit after CONSISTENT state has been reached will
 * be replayed, even though they might have started while still in
 * FULL_SNAPSHOT. That ensures that we'll reach a point where no previous
 * changes has been exported, but all the following ones will be. That point
 * is a convenient point to initialize replication from, which is why we
 * export a snapshot at that point, which *can* be used to read normal data.
 *
 * Copyright (c) 2012-2026, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *    src/backend/replication/logical/snapbuild.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include <sys/stat.h>
#include <unistd.h>

#include "access/heapam_xlog.h"
#include "access/transam.h"
#include "access/xact.h"
#include "common/file_utils.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/logical.h"
#include "replication/reorderbuffer.h"
#include "replication/snapbuild.h"
#include "replication/snapbuild_internal.h"
#include "storage/fd.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/standby.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"
#include "utils/snapshot.h"
#include "utils/wait_event.h"


/*
 * Starting a transaction -- which we need to do while exporting a snapshot --
 * removes knowledge about the previously used resowner, so we save it here.
 */
static ResourceOwner SavedResourceOwnerDuringExport = NULL;
static bool ExportInProgress = false;

/* ->committed and ->catchange manipulation */
static void SnapBuildPurgeOlderTxn(SnapBuild *builder);

/* snapshot building/manipulation/distribution functions */
static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder);

static void SnapBuildFreeSnapshot(Snapshot snap);

static void SnapBuildSnapIncRefcount(Snapshot snap);

static void SnapBuildDistributeSnapshotAndInval(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid);

static inline bool SnapBuildXidHasCatalogChanges(SnapBuild *builder, TransactionId xid,
                         uint32 xinfo);

/* xlog reading helper functions for SnapBuildProcessRunningXacts */
static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn,
                  xl_running_xacts *running);
static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff);

/* serialization functions */
static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn);
static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn);
static void SnapBuildRestoreContents(int fd, void *dest, Size size, const char *path);

/*
 * Allocate a new snapshot builder.
 *
 * xmin_horizon is the xid >= which we can be sure no catalog rows have been
 * removed, start_lsn is the LSN >= we want to replay commits.
 */
SnapBuild *
AllocateSnapshotBuilder(ReorderBuffer *reorder,
            TransactionId xmin_horizon,
            XLogRecPtr start_lsn,
            bool need_full_snapshot,
            bool in_slot_creation,
            XLogRecPtr two_phase_at)
{
  MemoryContext context;
  MemoryContext oldcontext;
  SnapBuild  *builder;

  /* allocate memory in own context, to have better accountability */
  context = AllocSetContextCreate(CurrentMemoryContext,
                  "snapshot builder context",
                  ALLOCSET_DEFAULT_SIZES);
  oldcontext = MemoryContextSwitchTo(context);

  builder = palloc0_object(SnapBuild);

  builder->state = SNAPBUILD_START;
  builder->context = context;
  builder->reorder = reorder;
  /* Other struct members initialized by zeroing via palloc0 above */

  builder->committed.xcnt = 0;
  builder->committed.xcnt_space = 128;  /* arbitrary number */
  builder->committed.xip =
    palloc0_array(TransactionId, builder->committed.xcnt_space);
  builder->committed.includes_all_transactions = true;

  builder->catchange.xcnt = 0;
  builder->catchange.xip = NULL;

  builder->initial_xmin_horizon = xmin_horizon;
  builder->start_decoding_at = start_lsn;
  builder->in_slot_creation = in_slot_creation;
  builder->building_full_snapshot = need_full_snapshot;
  builder->two_phase_at = two_phase_at;

  MemoryContextSwitchTo(oldcontext);

  return builder;
}

/*
 * Free a snapshot builder.
 */
void
FreeSnapshotBuilder(SnapBuild *builder)
{
  MemoryContext context = builder->context;

  /* free snapshot explicitly, that contains some error checking */
  if (builder->snapshot != NULL)
  {
    SnapBuildSnapDecRefcount(builder->snapshot);
    builder->snapshot = NULL;
  }

  /* other resources are deallocated via memory context reset */
  MemoryContextDelete(context);
}

/*
 * Free an unreferenced snapshot that has previously been built by us.
 */
static void
SnapBuildFreeSnapshot(Snapshot snap)
{
  /* make sure we don't get passed an external snapshot */
  Assert(snap->snapshot_type == SNAPSHOT_HISTORIC_MVCC);

  /* make sure nobody modified our snapshot */
  Assert(snap->curcid == FirstCommandId);
  Assert(!snap->suboverflowed);
  Assert(!snap->takenDuringRecovery);
  Assert(snap->regd_count == 0);

  /* slightly more likely, so it's checked even without c-asserts */
  if (snap->copied)
    elog(ERROR, "cannot free a copied snapshot");

  if (snap->active_count)
    elog(ERROR, "cannot free an active snapshot");

  pfree(snap);
}

/*
 * In which state of snapshot building are we?
 */
SnapBuildState
SnapBuildCurrentState(SnapBuild *builder)
{
  return builder->state;
}

/*
 * Return the LSN at which the two-phase decoding was first enabled.
 */
XLogRecPtr
SnapBuildGetTwoPhaseAt(SnapBuild *builder)
{
  return builder->two_phase_at;
}

/*
 * Set the LSN at which two-phase decoding is enabled.
 */
void
SnapBuildSetTwoPhaseAt(SnapBuild *builder, XLogRecPtr ptr)
{
  builder->two_phase_at = ptr;
}

/*
 * Should the contents of transaction ending at 'ptr' be decoded?
 */
bool
SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
{
  return ptr < builder->start_decoding_at;
}

/*
 * Increase refcount of a snapshot.
 *
 * This is used when handing out a snapshot to some external resource or when
 * adding a Snapshot as builder->snapshot.
 */
static void
SnapBuildSnapIncRefcount(Snapshot snap)
{
  snap->active_count++;
}

/*
 * Decrease refcount of a snapshot and free if the refcount reaches zero.
 *
 * Externally visible, so that external resources that have been handed an
 * IncRef'ed Snapshot can adjust its refcount easily.
 */
void
SnapBuildSnapDecRefcount(Snapshot snap)
{
  /* make sure we don't get passed an external snapshot */
  Assert(snap->snapshot_type == SNAPSHOT_HISTORIC_MVCC);

  /* make sure nobody modified our snapshot */
  Assert(snap->curcid == FirstCommandId);
  Assert(!snap->suboverflowed);
  Assert(!snap->takenDuringRecovery);

  Assert(snap->regd_count == 0);

  Assert(snap->active_count > 0);

  /* slightly more likely, so it's checked even without casserts */
  if (snap->copied)
    elog(ERROR, "cannot free a copied snapshot");

  snap->active_count--;
  if (snap->active_count == 0)
    SnapBuildFreeSnapshot(snap);
}

/*
 * Build a new snapshot, based on currently committed catalog-modifying
 * transactions.
 *
 * In-progress transactions with catalog access are *not* allowed to modify
 * these snapshots; they have to copy them and fill in appropriate ->curcid
 * and ->subxip/subxcnt values.
 *
 * Note that some of the transactions in the returned snapshot->xip might not
 * have finished committing yet (we saw their commit WAL records, but we don't
 * know if they've removed themselves from procarray).  Callers that want to
 * use the returned snapshot as an MVCC one must wait for this to happen; see
 * SnapBuildInitialSnapshot().
 */
static Snapshot
SnapBuildBuildSnapshot(SnapBuild *builder)
{
  Snapshot  snapshot;
  Size    ssize;

  Assert(builder->state >= SNAPBUILD_FULL_SNAPSHOT);

  ssize = sizeof(SnapshotData)
    + sizeof(TransactionId) * builder->committed.xcnt
    + sizeof(TransactionId) * 1 /* toplevel xid */ ;

  snapshot = MemoryContextAllocZero(builder->context, ssize);

  snapshot->snapshot_type = SNAPSHOT_HISTORIC_MVCC;

  /*
   * We misuse the original meaning of SnapshotData's xip and subxip fields
   * to make the more fitting for our needs.
   *
   * In the 'xip' array we store transactions that have to be treated as
   * committed. Since we will only ever look at tuples from transactions
   * that have modified the catalog it's more efficient to store those few
   * that exist between xmin and xmax (frequently there are none).
   *
   * Snapshots that are used in transactions that have modified the catalog
   * also use the 'subxip' array to store their toplevel xid and all the
   * subtransaction xids so we can recognize when we need to treat rows as
   * visible that are not in xip but still need to be visible. Subxip only
   * gets filled when the transaction is copied into the context of a
   * catalog modifying transaction since we otherwise share a snapshot
   * between transactions. As long as a txn hasn't modified the catalog it
   * doesn't need to treat any uncommitted rows as visible, so there is no
   * need for those xids.
   *
   * Both arrays are qsort'ed so that we can use bsearch() on them.
   */
  Assert(TransactionIdIsNormal(builder->xmin));
  Assert(TransactionIdIsNormal(builder->xmax));

  snapshot->xmin = builder->xmin;
  snapshot->xmax = builder->xmax;

  /* store all transactions to be treated as committed by this snapshot */
  snapshot->xip =
    (TransactionId *) ((char *) snapshot + sizeof(SnapshotData));
  snapshot->xcnt = builder->committed.xcnt;
  memcpy(snapshot->xip,
       builder->committed.xip,
       builder->committed.xcnt * sizeof(TransactionId));

  /* sort so we can bsearch() */
  qsort(snapshot->xip, snapshot->xcnt, sizeof(TransactionId), xidComparator);

  /*
   * Initially, subxip is empty, i.e. it's a snapshot to be used by
   * transactions that don't modify the catalog. Will be filled by
   * ReorderBufferCopySnap() if necessary.
   */
  snapshot->subxcnt = 0;
  snapshot->subxip = NULL;

  snapshot->suboverflowed = false;
  snapshot->takenDuringRecovery = false;
  snapshot->copied = false;
  snapshot->curcid = FirstCommandId;
  snapshot->active_count = 0;
  snapshot->regd_count = 0;
  snapshot->snapXactCompletionCount = 0;

  return snapshot;
}

/*
 * Build the initial slot snapshot and convert it to a normal snapshot that
 * is understood by HeapTupleSatisfiesMVCC.
 *
 * The snapshot will be usable directly in current transaction or exported
 * for loading in different transaction.
 */
Snapshot
SnapBuildInitialSnapshot(SnapBuild *builder)
{
  Snapshot  snap;
  TransactionId xid;
  TransactionId safeXid;
  TransactionId *newxip;
  int     newxcnt = 0;

  Assert(XactIsoLevel == XACT_REPEATABLE_READ);
  Assert(builder->building_full_snapshot);

  /* don't allow older snapshots */
  InvalidateCatalogSnapshot();  /* about to overwrite MyProc->xmin */
  if (HaveRegisteredOrActiveSnapshot())
    elog(ERROR, "cannot build an initial slot snapshot when snapshots exist");
  Assert(!HistoricSnapshotActive());

  if (builder->state != SNAPBUILD_CONSISTENT)
    elog(ERROR, "cannot build an initial slot snapshot before reaching a consistent state");

  if (!builder->committed.includes_all_transactions)
    elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");

  /* so we don't overwrite the existing value */
  if (TransactionIdIsValid(MyProc->xmin))
    elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");

  snap = SnapBuildBuildSnapshot(builder);

  /*
   * Transactions appear in snap->xip as soon as their commit WAL records
   * are decoded, but that doesn't imply they can be considered committed in
   * a regular MVCC snapshot: that happens only when they remove themselves
   * from procarray.  Wait for this to happen; otherwise somebody using the
   * snapshot might set hint bits incorrectly.
   *
   * This is not needed during recovery, because the decoded commit record
   * has been replayed already.
   */
  if (!RecoveryInProgress())
  {
    RunningTransactions running;

    running = GetRunningTransactionData();
    LWLockRelease(ProcArrayLock);
    LWLockRelease(XidGenLock);

    /*
     * Note we walk the array up to running->xcnt only, omitting this wait
     * for subtransactions: the subxacts are covered by their top-level
     * transaction already, there's no need for an additional wait.
     */
    for (int i = 0; i < running->xcnt; i++)
    {
      TransactionId running_xid = running->xids[i];

      if (bsearch(&running_xid, snap->xip, snap->xcnt,
            sizeof(TransactionId), xidComparator) != NULL)
        XactLockTableWait(running_xid, NULL, NULL, XLTW_None);
    }
  }

  /*
   * Building an initial snapshot is expensive and an unenforced xmin
   * horizon would have bad consequences, therefore always double-check that
   * the horizon is enforced.
   */
  LWLockAcquire(ProcArrayLock, LW_SHARED);
  safeXid = GetOldestSafeDecodingTransactionId(false);
  LWLockRelease(ProcArrayLock);

  if (TransactionIdFollows(safeXid, snap->xmin))
    elog(ERROR, "cannot build an initial slot snapshot as oldest safe xid %u follows snapshot's xmin %u",
       safeXid, snap->xmin);

  /*
   * We know that snap->xmin is alive, enforced by the logical xmin
   * mechanism. Due to that we can do this without locks, we're only
   * changing our own value.
   */
  MyProc->xmin = snap->xmin;

  /* allocate in transaction context */
  newxip = palloc_array(TransactionId, GetMaxSnapshotXidCount());

  /*
   * snapbuild.c builds transactions in an "inverted" manner, which means it
   * stores committed transactions in ->xip, not ones in progress. Build a
   * classical snapshot by marking all non-committed transactions as
   * in-progress. This can be expensive.
   */
  for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)
  {
    void     *test;

    /*
     * Check whether transaction committed using the decoding snapshot
     * meaning of ->xip.
     */
    test = bsearch(&xid, snap->xip, snap->xcnt,
             sizeof(TransactionId), xidComparator);

    if (test == NULL)
    {
      if (newxcnt >= GetMaxSnapshotXidCount())
        ereport(ERROR,
            (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
             errmsg("initial slot snapshot too large")));

      newxip[newxcnt++] = xid;
    }

    TransactionIdAdvance(xid);
  }

  /* adjust remaining snapshot fields as needed */
  snap->snapshot_type = SNAPSHOT_MVCC;
  snap->xcnt = newxcnt;
  snap->xip = newxip;

  return snap;
}

/*
 * Export a snapshot so it can be set in another session with SET TRANSACTION
 * SNAPSHOT.
 *
 * For that we need to start a transaction in the current backend as the
 * importing side checks whether the source transaction is still open to make
 * sure the xmin horizon hasn't advanced since then.
 */
const char *
SnapBuildExportSnapshot(SnapBuild *builder)
{
  Snapshot  snap;
  char     *snapname;

  if (IsTransactionOrTransactionBlock())
    elog(ERROR, "cannot export a snapshot from within a transaction");

  if (SavedResourceOwnerDuringExport)
    elog(ERROR, "can only export one snapshot at a time");

  SavedResourceOwnerDuringExport = CurrentResourceOwner;
  ExportInProgress = true;

  StartTransactionCommand();

  /* There doesn't seem to a nice API to set these */
  XactIsoLevel = XACT_REPEATABLE_READ;
  XactReadOnly = true;

  snap = SnapBuildInitialSnapshot(builder);

  /*
   * now that we've built a plain snapshot, make it active and use the
   * normal mechanisms for exporting it
   */
  snapname = ExportSnapshot(snap);

  ereport(LOG,
      (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
               "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
               snap->xcnt,
               snapname, snap->xcnt)));
  return snapname;
}

/*
 * Ensure there is a snapshot and if not build one for current transaction.
 */
Snapshot
SnapBuildGetOrBuildSnapshot(SnapBuild *builder)
{
  Assert(builder->state == SNAPBUILD_CONSISTENT);

  /* only build a new snapshot if we don't have a prebuilt one */
  if (builder->snapshot == NULL)
  {
    builder->snapshot = SnapBuildBuildSnapshot(builder);
    /* increase refcount for the snapshot builder */
    SnapBuildSnapIncRefcount(builder->snapshot);
  }

  return builder->snapshot;
}

/*
 * Reset a previously SnapBuildExportSnapshot()'ed snapshot if there is
 * any. Aborts the previously started transaction and resets the resource
 * owner back to its original value.
 */
void
SnapBuildClearExportedSnapshot(void)
{
  ResourceOwner tmpResOwner;

  /* nothing exported, that is the usual case */
  if (!ExportInProgress)
    return;

  if (!IsTransactionState())
    elog(ERROR, "clearing exported snapshot in wrong transaction state");

  /*
   * AbortCurrentTransaction() takes care of resetting the snapshot state,
   * so remember SavedResourceOwnerDuringExport.
   */
  tmpResOwner = SavedResourceOwnerDuringExport;

  /* make sure nothing could have ever happened */
  AbortCurrentTransaction();

  CurrentResourceOwner = tmpResOwner;
}

/*
 * Clear snapshot export state during transaction abort.
 */
void
SnapBuildResetExportedSnapshotState(void)
{
  SavedResourceOwnerDuringExport = NULL;
  ExportInProgress = false;
}

/*
 * Handle the effects of a single heap change, appropriate to the current state
 * of the snapshot builder and returns whether changes made at (xid, lsn) can
 * be decoded.
 */
bool
SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
{
  /*
   * We can't handle data in transactions if we haven't built a snapshot
   * yet, so don't store them.
   */
  if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
    return false;

  /*
   * No point in keeping track of changes in transactions that we don't have
   * enough information about to decode. This means that they started before
   * we got into the SNAPBUILD_FULL_SNAPSHOT state.
   */
  if (builder->state < SNAPBUILD_CONSISTENT &&
    TransactionIdPrecedes(xid, builder->next_phase_at))
    return false;

  /*
   * If the reorderbuffer doesn't yet have a snapshot, add one now, it will
   * be needed to decode the change we're currently processing.
   */
  if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
  {
    /* only build a new snapshot if we don't have a prebuilt one */
    if (builder->snapshot == NULL)
    {
      builder->snapshot = SnapBuildBuildSnapshot(builder);
      /* increase refcount for the snapshot builder */
      SnapBuildSnapIncRefcount(builder->snapshot);
    }

    /*
     * Increase refcount for the transaction we're handing the snapshot
     * out to.
     */
    SnapBuildSnapIncRefcount(builder->snapshot);
    ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
                   builder->snapshot);
  }

  return true;
}

/*
 * Do CommandId/combo CID handling after reading an xl_heap_new_cid record.
 * This implies that a transaction has done some form of write to system
 * catalogs.
 */
void
SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid,
             XLogRecPtr lsn, xl_heap_new_cid *xlrec)
{
  CommandId cid;

  /*
   * we only log new_cid's if a catalog tuple was modified, so mark the
   * transaction as containing catalog modifications
   */
  ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);

  ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, xid, lsn,
                 xlrec->target_locator, xlrec->target_tid,
                 xlrec->cmin, xlrec->cmax,
                 xlrec->combocid);

  /* figure out new command id */
  if (xlrec->cmin != InvalidCommandId &&
    xlrec->cmax != InvalidCommandId)
    cid = Max(xlrec->cmin, xlrec->cmax);
  else if (xlrec->cmax != InvalidCommandId)
    cid = xlrec->cmax;
  else if (xlrec->cmin != InvalidCommandId)
    cid = xlrec->cmin;
  else
  {
    cid = InvalidCommandId; /* silence compiler */
    elog(ERROR, "xl_heap_new_cid record without a valid CommandId");
  }

  ReorderBufferAddNewCommandId(builder->reorder, xid, lsn, cid + 1);
}

/*
 * Add a new Snapshot and invalidation messages to all transactions we're
 * decoding that currently are in-progress so they can see new catalog contents
 * made by the transaction that just committed. This is necessary because those
 * in-progress transactions will use the new catalog's contents from here on
 * (at the very least everything they do needs to be compatible with newer
 * catalog contents).
 */
static void
SnapBuildDistributeSnapshotAndInval(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid)
{
  dlist_iter  txn_i;
  ReorderBufferTXN *txn;

  /*
   * Iterate through all toplevel transactions. This can include
   * subtransactions which we just don't yet know to be that, but that's
   * fine, they will just get an unnecessary snapshot and invalidations
   * queued.
   */
  dlist_foreach(txn_i, &builder->reorder->toplevel_by_lsn)
  {
    txn = dlist_container(ReorderBufferTXN, node, txn_i.cur);

    Assert(TransactionIdIsValid(txn->xid));

    /*
     * If we don't have a base snapshot yet, there are no changes in this
     * transaction which in turn implies we don't yet need a snapshot at
     * all. We'll add a snapshot when the first change gets queued.
     *
     * Similarly, we don't need to add invalidations to a transaction
     * whose base snapshot is not yet set. Once a base snapshot is built,
     * it will include the xids of committed transactions that have
     * modified the catalog, thus reflecting the new catalog contents. The
     * existing catalog cache will have already been invalidated after
     * processing the invalidations in the transaction that modified
     * catalogs, ensuring that a fresh cache is constructed during
     * decoding.
     *
     * NB: This works correctly even for subtransactions because
     * ReorderBufferAssignChild() takes care to transfer the base snapshot
     * to the top-level transaction, and while iterating the changequeue
     * we'll get the change from the subtxn.
     */
    if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, txn->xid))
      continue;

    /*
     * We don't need to add snapshot or invalidations to prepared
     * transactions as they should not see the new catalog contents.
     */
    if (rbtxn_is_prepared(txn))
      continue;

    elog(DEBUG2, "adding a new snapshot and invalidations to %u at %X/%08X",
       txn->xid, LSN_FORMAT_ARGS(lsn));

    /*
     * increase the snapshot's refcount for the transaction we are handing
     * it out to
     */
    SnapBuildSnapIncRefcount(builder->snapshot);
    ReorderBufferAddSnapshot(builder->reorder, txn->xid, lsn,
                 builder->snapshot);

    /*
     * Add invalidation messages to the reorder buffer of in-progress
     * transactions except the current committed transaction, for which we
     * will execute invalidations at the end.
     *
     * It is required, otherwise, we will end up using the stale catcache
     * contents built by the current transaction even after its decoding,
     * which should have been invalidated due to concurrent catalog
     * changing transaction.
     *
     * Distribute only the invalidation messages generated by the current
     * committed transaction. Invalidation messages received from other
     * transactions would have already been propagated to the relevant
     * in-progress transactions. This transaction would have processed
     * those invalidations, ensuring that subsequent transactions observe
     * a consistent cache state.
     */
    if (txn->xid != xid)
    {
      uint32    ninvalidations;
      SharedInvalidationMessage *msgs = NULL;

      ninvalidations = ReorderBufferGetInvalidations(builder->reorder,
                               xid, &msgs);

      if (ninvalidations > 0)
      {
        Assert(msgs != NULL);

        ReorderBufferAddDistributedInvalidations(builder->reorder,
                             txn->xid, lsn,
                             ninvalidations, msgs);
      }
    }
  }
}

/*
 * Keep track of a new catalog changing transaction that has committed.
 */
static void
SnapBuildAddCommittedTxn(SnapBuild *builder, TransactionId xid)
{
  Assert(TransactionIdIsValid(xid));

  if (builder->committed.xcnt == builder->committed.xcnt_space)
  {
    builder->committed.xcnt_space = builder->committed.xcnt_space * 2 + 1;

    elog(DEBUG1, "increasing space for committed transactions to %u",
       (uint32) builder->committed.xcnt_space);

    builder->committed.xip = repalloc_array(builder->committed.xip,
                        TransactionId,
                        builder->committed.xcnt_space);
  }

  /*
   * TODO: It might make sense to keep the array sorted here instead of
   * doing it every time we build a new snapshot. On the other hand this
   * gets called repeatedly when a transaction with subtransactions commits.
   */
  builder->committed.xip[builder->committed.xcnt++] = xid;
}

/*
 * Remove knowledge about transactions we treat as committed or containing catalog
 * changes that are smaller than ->xmin. Those won't ever get checked via
 * the ->committed or ->catchange array, respectively. The committed xids will
 * get checked via the clog machinery.
 *
 * We can ideally remove the transaction from catchange array once it is
 * finished (committed/aborted) but that could be costly as we need to maintain
 * the xids order in the array.
 */
static void
SnapBuildPurgeOlderTxn(SnapBuild *builder)
{
  TransactionId *workspace;
  int     surviving_xids = 0;

  /* not ready yet */
  if (!TransactionIdIsNormal(builder->xmin))
    return;

  /* TODO: Neater algorithm than just copying and iterating? */
  workspace =
    MemoryContextAlloc(builder->context,
               builder->committed.xcnt * sizeof(TransactionId));

  /* copy xids that still are interesting to workspace */
  for (size_t off = 0; off < builder->committed.xcnt; off++)
  {
    if (NormalTransactionIdPrecedes(builder->committed.xip[off],
                    builder->xmin))
      ;         /* remove */
    else
      workspace[surviving_xids++] = builder->committed.xip[off];
  }

  /* copy workspace back to persistent state */
  memcpy(builder->committed.xip, workspace,
       surviving_xids * sizeof(TransactionId));

  elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
     (uint32) builder->committed.xcnt, (uint32) surviving_xids,
     builder->xmin, builder->xmax);
  builder->committed.xcnt = surviving_xids;

  pfree(workspace);

  /*
   * Purge xids in ->catchange as well. The purged array must also be sorted
   * in xidComparator order.
   */
  if (builder->catchange.xcnt > 0)
  {
    size_t    off;

    /*
     * Since catchange.xip is sorted, we find the lower bound of xids that
     * are still interesting.
     */
    for (off = 0; off < builder->catchange.xcnt; off++)
    {
      if (TransactionIdFollowsOrEquals(builder->catchange.xip[off],
                       builder->xmin))
        break;
    }

    surviving_xids = builder->catchange.xcnt - off;

    if (surviving_xids > 0)
    {
      memmove(builder->catchange.xip, &(builder->catchange.xip[off]),
          surviving_xids * sizeof(TransactionId));
    }
    else
    {
      pfree(builder->catchange.xip);
      builder->catchange.xip = NULL;
    }

    elog(DEBUG3, "purged catalog modifying transactions from %u to %u, xmin: %u, xmax: %u",
       (uint32) builder->catchange.xcnt, (uint32) surviving_xids,
       builder->xmin, builder->xmax);
    builder->catchange.xcnt = surviving_xids;
  }
}

/*
 * Handle everything that needs to be done when a transaction commits
 */
void
SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
           int nsubxacts, TransactionId *subxacts, uint32 xinfo)
{
  int     nxact;

  bool    needs_snapshot = false;
  bool    needs_timetravel = false;
  bool    sub_needs_timetravel = false;

  TransactionId xmax = xid;

  /*
   * Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
   * will they be part of a snapshot.  So we don't need to record anything.
   */
  if (builder->state == SNAPBUILD_START ||
    (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
     TransactionIdPrecedes(xid, builder->next_phase_at)))
  {
    /* ensure that only commits after this are getting replayed */
    if (builder->start_decoding_at <= lsn)
      builder->start_decoding_at = lsn + 1;
    return;
  }

  if (builder->state < SNAPBUILD_CONSISTENT)
  {
    /* ensure that only commits after this are getting replayed */
    if (builder->start_decoding_at <= lsn)
      builder->start_decoding_at = lsn + 1;

    /*
     * If building an exportable snapshot, force xid to be tracked, even
     * if the transaction didn't modify the catalog.
     */
    if (builder->building_full_snapshot)
    {
      needs_timetravel = true;
    }
  }

  for (nxact = 0; nxact < nsubxacts; nxact++)
  {
    TransactionId subxid = subxacts[nxact];

    /*
     * Add subtransaction to base snapshot if catalog modifying, we don't
     * distinguish to toplevel transactions there.
     */
    if (SnapBuildXidHasCatalogChanges(builder, subxid, xinfo))
    {
      sub_needs_timetravel = true;
      needs_snapshot = true;

      elog(DEBUG1, "found subtransaction %u:%u with catalog changes",
         xid, subxid);

      SnapBuildAddCommittedTxn(builder, subxid);

      if (NormalTransactionIdFollows(subxid, xmax))
        xmax = subxid;
    }

    /*
     * If we're forcing timetravel we also need visibility information
     * about subtransaction, so keep track of subtransaction's state, even
     * if not catalog modifying.  Don't need to distribute a snapshot in
     * that case.
     */
    else if (needs_timetravel)
    {
      SnapBuildAddCommittedTxn(builder, subxid);
      if (NormalTransactionIdFollows(subxid, xmax))
        xmax = subxid;
    }
  }

  /* if top-level modified catalog, it'll need a snapshot */
  if (SnapBuildXidHasCatalogChanges(builder, xid, xinfo))
  {
    elog(DEBUG2, "found top level transaction %u, with catalog changes",
       xid);
    needs_snapshot = true;
    needs_timetravel = true;
    SnapBuildAddCommittedTxn(builder, xid);
  }
  else if (sub_needs_timetravel)
  {
    /* track toplevel txn as well, subxact alone isn't meaningful */
    elog(DEBUG2, "forced transaction %u to do timetravel due to one of its subtransactions",
       xid);
    needs_timetravel = true;
    SnapBuildAddCommittedTxn(builder, xid);
  }
  else if (needs_timetravel)
  {
    elog(DEBUG2, "forced transaction %u to do timetravel", xid);

    SnapBuildAddCommittedTxn(builder, xid);
  }

  if (!needs_timetravel)
  {
    /* record that we cannot export a general snapshot anymore */
    builder->committed.includes_all_transactions = false;
  }

  Assert(!needs_snapshot || needs_timetravel);

  /*
   * Adjust xmax of the snapshot builder, we only do that for committed,
   * catalog modifying, transactions, everything else isn't interesting for
   * us since we'll never look at the respective rows.
   */
  if (needs_timetravel &&
    (!TransactionIdIsValid(builder->xmax) ||
     TransactionIdFollowsOrEquals(xmax, builder->xmax)))
  {
    builder->xmax = xmax;
    TransactionIdAdvance(builder->xmax);
  }

  /* if there's any reason to build a historic snapshot, do so now */
  if (needs_snapshot)
  {
    /*
     * If we haven't built a complete snapshot yet there's no need to hand
     * it out, it wouldn't (and couldn't) be used anyway.
     */
    if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
      return;

    /*
     * Decrease the snapshot builder's refcount of the old snapshot, note
     * that it still will be used if it has been handed out to the
     * reorderbuffer earlier.
     */
    if (builder->snapshot)
      SnapBuildSnapDecRefcount(builder->snapshot);

    builder->snapshot = SnapBuildBuildSnapshot(builder);

    /* we might need to execute invalidations, add snapshot */
    if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
    {
      SnapBuildSnapIncRefcount(builder->snapshot);
      ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
                     builder->snapshot);
    }

    /* refcount of the snapshot builder for the new snapshot */
    SnapBuildSnapIncRefcount(builder->snapshot);

    /*
     * Add a new catalog snapshot and invalidations messages to all
     * currently running transactions.
     */
    SnapBuildDistributeSnapshotAndInval(builder, lsn, xid);
  }
}

/*
 * Check the reorder buffer and the snapshot to see if the given transaction has
 * modified catalogs.
 */
static inline bool
SnapBuildXidHasCatalogChanges(SnapBuild *builder, TransactionId xid,
                uint32 xinfo)
{
  if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))
    return true;

  /*
   * The transactions that have changed catalogs must have invalidation
   * info.
   */
  if (!(xinfo & XACT_XINFO_HAS_INVALS))
    return false;

  /* Check the catchange XID array */
  return ((builder->catchange.xcnt > 0) &&
      (bsearch(&xid, builder->catchange.xip, builder->catchange.xcnt,
           sizeof(TransactionId), xidComparator) != NULL));
}

/* -----------------------------------
 * Snapshot building functions dealing with xlog records
 * -----------------------------------
 */

/*
 * Process a running xacts record, and use its information to first build a
 * historic snapshot and later to release resources that aren't needed
 * anymore.
 */
void
SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
{
  ReorderBufferTXN *txn;
  TransactionId xmin;

  /*
   * If we're not consistent yet, inspect the record to see whether it
   * allows to get closer to being consistent. If we are consistent, dump
   * our snapshot so others or we, after a restart, can use it.
   */
  if (builder->state < SNAPBUILD_CONSISTENT)
  {
    /* returns false if there's no point in performing cleanup just yet */
    if (!SnapBuildFindSnapshot(builder, lsn, running))
      return;
  }
  else
    SnapBuildSerialize(builder, lsn);

  /*
   * Update range of interesting xids based on the running xacts
   * information. We don't increase ->xmax using it, because once we are in
   * a consistent state we can do that ourselves and much more efficiently
   * so, because we only need to do it for catalog transactions since we
   * only ever look at those.
   *
   * NB: We only increase xmax when a catalog modifying transaction commits
   * (see SnapBuildCommitTxn).  Because of this, xmax can be lower than
   * xmin, which looks odd but is correct and actually more efficient, since
   * we hit fast paths in heapam_visibility.c.
   */
  builder->xmin = running->oldestRunningXid;

  /* Remove transactions we don't need to keep track off anymore */
  SnapBuildPurgeOlderTxn(builder);

  /*
   * Advance the xmin limit for the current replication slot, to allow
   * vacuum to clean up the tuples this slot has been protecting.
   *
   * The reorderbuffer might have an xmin among the currently running
   * snapshots; use it if so.  If not, we need only consider the snapshots
   * we'll produce later, which can't be less than the oldest running xid in
   * the record we're reading now.
   */
  xmin = ReorderBufferGetOldestXmin(builder->reorder);
  if (xmin == InvalidTransactionId)
    xmin = running->oldestRunningXid;
  elog(DEBUG3, "xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
     builder->xmin, builder->xmax, running->oldestRunningXid, xmin);
  LogicalIncreaseXminForSlot(lsn, xmin);

  /*
   * Also tell the slot where we can restart decoding from. We don't want to
   * do that after every commit because changing that implies an fsync of
   * the logical slot's state file, so we only do it every time we see a
   * running xacts record.
   *
   * Do so by looking for the oldest in progress transaction (determined by
   * the first LSN of any of its relevant records). Every transaction
   * remembers the last location we stored the snapshot to disk before its
   * beginning. That point is where we can restart from.
   */

  /*
   * Can't know about a serialized snapshot's location if we're not
   * consistent.
   */
  if (builder->state < SNAPBUILD_CONSISTENT)
    return;

  txn = ReorderBufferGetOldestTXN(builder->reorder);

  /*
   * oldest ongoing txn might have started when we didn't yet serialize
   * anything because we hadn't reached a consistent state yet.
   */
  if (txn != NULL && XLogRecPtrIsValid(txn->restart_decoding_lsn))
    LogicalIncreaseRestartDecodingForSlot(lsn, txn->restart_decoding_lsn);

  /*
   * No in-progress transaction, can reuse the last serialized snapshot if
   * we have one.
   */
  else if (txn == NULL &&
       XLogRecPtrIsValid(builder->reorder->current_restart_decoding_lsn) &&
       XLogRecPtrIsValid(builder->last_serialized_snapshot))
    LogicalIncreaseRestartDecodingForSlot(lsn,
                        builder->last_serialized_snapshot);
}


/*
 * Build the start of a snapshot that's capable of decoding the catalog.
 *
 * Helper function for SnapBuildProcessRunningXacts() while we're not yet
 * consistent.
 *
 * Returns true if there is a point in performing internal maintenance/cleanup
 * using the xl_running_xacts record.
 */
static bool
SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
{
  /* ---
   * Build catalog decoding snapshot incrementally using information about
   * the currently running transactions. There are several ways to do that:
   *
   * a) There were no running transactions when the xl_running_xacts record
   *    was inserted, jump to CONSISTENT immediately. We might find such a
   *    state while waiting on c)'s sub-states.
   *
   * b) This (in a previous run) or another decoding slot serialized a
   *    snapshot to disk that we can use. Can't use this method while finding
   *    the start point for decoding changes as the restart LSN would be an
   *    arbitrary LSN but we need to find the start point to extract changes
   *    where we won't see the data for partial transactions. Also, we cannot
   *    use this method when a slot needs a full snapshot for export or direct
   *    use, as that snapshot will only contain catalog modifying transactions.
   *
   * c) First incrementally build a snapshot for catalog tuples
   *    (BUILDING_SNAPSHOT), that requires all, already in-progress,
   *    transactions to finish.  Every transaction starting after that
   *    (FULL_SNAPSHOT state), has enough information to be decoded.  But
   *    for older running transactions no viable snapshot exists yet, so
   *    CONSISTENT will only be reached once all of those have finished.
   * ---
   */

  /*
   * xl_running_xacts record is older than what we can use, we might not
   * have all necessary catalog rows anymore.
   */
  if (TransactionIdIsNormal(builder->initial_xmin_horizon) &&
    NormalTransactionIdPrecedes(running->oldestRunningXid,
                  builder->initial_xmin_horizon))
  {
    ereport(DEBUG1,
        errmsg_internal("skipping snapshot at %X/%08X while building logical decoding snapshot, xmin horizon too low",
                LSN_FORMAT_ARGS(lsn)),
        errdetail_internal("initial xmin horizon of %u vs the snapshot's %u",
                   builder->initial_xmin_horizon, running->oldestRunningXid));


    SnapBuildWaitSnapshot(running, builder->initial_xmin_horizon);

    return true;
  }

  /*
   * a) No transaction were running, we can jump to consistent.
   *
   * This is not affected by races around xl_running_xacts, because we can
   * miss transaction commits, but currently not transactions starting.
   *
   * NB: We might have already started to incrementally assemble a snapshot,
   * so we need to be careful to deal with that.
   */
  if (running->oldestRunningXid == running->nextXid)
  {
    if (!XLogRecPtrIsValid(builder->start_decoding_at) ||
      builder->start_decoding_at <= lsn)
      /* can decode everything after this */
      builder->start_decoding_at = lsn + 1;

    /* As no transactions were running xmin/xmax can be trivially set. */
    builder->xmin = running->nextXid; /* < are finished */
    builder->xmax = running->nextXid; /* >= are running */

    /* so we can safely use the faster comparisons */
    Assert(TransactionIdIsNormal(builder->xmin));
    Assert(TransactionIdIsNormal(builder->xmax));

    builder->state = SNAPBUILD_CONSISTENT;
    builder->next_phase_at = InvalidTransactionId;

    ereport(LogicalDecodingLogLevel(),
        errmsg("logical decoding found consistent point at %X/%08X",
             LSN_FORMAT_ARGS(lsn)),
        errdetail("There are no running transactions."));

    return false;
  }

  /*
   * b) valid on disk state and while neither building full snapshot nor
   * creating a slot.
   */
  else if (!builder->building_full_snapshot &&
       !builder->in_slot_creation &&
       SnapBuildRestore(builder, lsn))
  {
    /* there won't be any state to cleanup */
    return false;
  }

  /*
   * c) transition from START to BUILDING_SNAPSHOT.
   *
   * In START state, and a xl_running_xacts record with running xacts is
   * encountered.  In that case, switch to BUILDING_SNAPSHOT state, and
   * record xl_running_xacts->nextXid.  Once all running xacts have finished
   * (i.e. they're all >= nextXid), we have a complete catalog snapshot.  It
   * might look that we could use xl_running_xacts's ->xids information to
   * get there quicker, but that is problematic because transactions marked
   * as running, might already have inserted their commit record - it's
   * infeasible to change that with locking.
   */
  else if (builder->state == SNAPBUILD_START)
  {
    builder->state = SNAPBUILD_BUILDING_SNAPSHOT;
    builder->next_phase_at = running->nextXid;

    /*
     * Start with an xmin/xmax that's correct for future, when all the
     * currently running transactions have finished. We'll update both
     * while waiting for the pending transactions to finish.
     */
    builder->xmin = running->nextXid; /* < are finished */
    builder->xmax = running->nextXid; /* >= are running */

    /* so we can safely use the faster comparisons */
    Assert(TransactionIdIsNormal(builder->xmin));
    Assert(TransactionIdIsNormal(builder->xmax));

    ereport(LOG,
        errmsg("logical decoding found initial starting point at %X/%08X",
             LSN_FORMAT_ARGS(lsn)),
        errdetail("Waiting for transactions (approximately %d) older than %u to end.",
              running->xcnt, running->nextXid));

    SnapBuildWaitSnapshot(running, running->nextXid);
  }

  /*
   * c) transition from BUILDING_SNAPSHOT to FULL_SNAPSHOT.
   *
   * In BUILDING_SNAPSHOT state, and this xl_running_xacts' oldestRunningXid
   * is >= than nextXid from when we switched to BUILDING_SNAPSHOT.  This
   * means all transactions starting afterwards have enough information to
   * be decoded.  Switch to FULL_SNAPSHOT.
   */
  else if (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
       TransactionIdPrecedesOrEquals(builder->next_phase_at,
                       running->oldestRunningXid))
  {
    builder->state = SNAPBUILD_FULL_SNAPSHOT;
    builder->next_phase_at = running->nextXid;

    ereport(LOG,
        errmsg("logical decoding found initial consistent point at %X/%08X",
             LSN_FORMAT_ARGS(lsn)),
        errdetail("Waiting for transactions (approximately %d) older than %u to end.",
              running->xcnt, running->nextXid));

    SnapBuildWaitSnapshot(running, running->nextXid);
  }

  /*
   * c) transition from FULL_SNAPSHOT to CONSISTENT.
   *
   * In FULL_SNAPSHOT state, and this xl_running_xacts' oldestRunningXid is
   * >= than nextXid from when we switched to FULL_SNAPSHOT.  This means all
   * transactions that are currently in progress have a catalog snapshot,
   * and all their changes have been collected.  Switch to CONSISTENT.
   */
  else if (builder->state == SNAPBUILD_FULL_SNAPSHOT &&
       TransactionIdPrecedesOrEquals(builder->next_phase_at,
                       running->oldestRunningXid))
  {
    builder->state = SNAPBUILD_CONSISTENT;
    builder->next_phase_at = InvalidTransactionId;

    ereport(LogicalDecodingLogLevel(),
        errmsg("logical decoding found consistent point at %X/%08X",
             LSN_FORMAT_ARGS(lsn)),
        errdetail("There are no old transactions anymore."));
  }

  /*
   * We already started to track running xacts and need to wait for all
   * in-progress ones to finish. We fall through to the normal processing of
   * records so incremental cleanup can be performed.
   */
  return true;
}

/* ---
 * Iterate through xids in record, wait for all older than the cutoff to
 * finish.  Then, if possible, log a new xl_running_xacts record.
 *
 * This isn't required for the correctness of decoding, but to:
 * a) allow isolationtester to notice that we're currently waiting for
 *    something.
 * b) log a new xl_running_xacts record where it'd be helpful, without having
 *    to wait for bgwriter or checkpointer.
 * ---
 */
static void
SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff)
{
  int     off;

  for (off = 0; off < running->xcnt; off++)
  {
    TransactionId xid = running->xids[off];

    /*
     * Upper layers should prevent that we ever need to wait on ourselves.
     * Check anyway, since failing to do so would either result in an
     * endless wait or an Assert() failure.
     */
    if (TransactionIdIsCurrentTransactionId(xid))
      elog(ERROR, "waiting for ourselves");

    if (TransactionIdFollows(xid, cutoff))
      continue;

    XactLockTableWait(xid, NULL, NULL, XLTW_None);
  }

  /*
   * All transactions we needed to finish finished - try to ensure there is
   * another xl_running_xacts record in a timely manner, without having to
   * wait for bgwriter or checkpointer to log one.  During recovery we can't
   * enforce that, so we'll have to wait.
   */
  if (!RecoveryInProgress())
  {
    LogStandbySnapshot();
  }
}

#define SnapBuildOnDiskConstantSize \
  offsetof(SnapBuildOnDisk, builder)
#define SnapBuildOnDiskNotChecksummedSize \
  offsetof(SnapBuildOnDisk, version)

#define SNAPBUILD_MAGIC 0x51A1E001
#define SNAPBUILD_VERSION 6

/*
 * Store/Load a snapshot from disk, depending on the snapshot builder's state.
 *
 * Supposed to be used by external (i.e. not snapbuild.c) code that just read
 * a record that's a potential location for a serialized snapshot.
 */
void
SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
{
  if (builder->state < SNAPBUILD_CONSISTENT)
    SnapBuildRestore(builder, lsn);
  else
    SnapBuildSerialize(builder, lsn);
}

/*
 * Serialize the snapshot 'builder' at the location 'lsn' if it hasn't already
 * been done by another decoding process.
 */
static void
SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
{
  Size    needed_length;
  SnapBuildOnDisk *ondisk = NULL;
  TransactionId *catchange_xip = NULL;
  MemoryContext old_ctx;
  size_t    catchange_xcnt;
  char     *ondisk_c;
  int     fd;
  char    tmppath[MAXPGPATH];
  char    path[MAXPGPATH];
  int     ret;
  struct stat stat_buf;
  Size    sz;

  Assert(XLogRecPtrIsValid(lsn));
  Assert(!XLogRecPtrIsValid(builder->last_serialized_snapshot) ||
       builder->last_serialized_snapshot <= lsn);

  /*
   * no point in serializing if we cannot continue to work immediately after
   * restoring the snapshot
   */
  if (builder->state < SNAPBUILD_CONSISTENT)
    return;

  /* consistent snapshots have no next phase */
  Assert(builder->next_phase_at == InvalidTransactionId);

  /*
   * We identify snapshots by the LSN they are valid for. We don't need to
   * include timelines in the name as each LSN maps to exactly one timeline
   * unless the user used pg_resetwal or similar. If a user did so, there's
   * no hope continuing to decode anyway.
   */
  sprintf(path, "%s/%X-%X.snap",
      PG_LOGICAL_SNAPSHOTS_DIR,
      LSN_FORMAT_ARGS(lsn));

  /*
   * first check whether some other backend already has written the snapshot
   * for this LSN. It's perfectly fine if there's none, so we accept ENOENT
   * as a valid state. Everything else is an unexpected error.
   */
  ret = stat(path, &stat_buf);

  if (ret != 0 && errno != ENOENT)
    ereport(ERROR,
        (errcode_for_file_access(),
         errmsg("could not stat file \"%s\": %m", path)));

  else if (ret == 0)
  {
    /*
     * somebody else has already serialized to this point, don't overwrite
     * but remember location, so we don't need to read old data again.
     *
     * To be sure it has been synced to disk after the rename() from the
     * tempfile filename to the real filename, we just repeat the fsync.
     * That ought to be cheap because in most scenarios it should already
     * be safely on disk.
     */
    fsync_fname(path, false);
    fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);

    builder->last_serialized_snapshot = lsn;
    goto out;
  }

  /*
   * there is an obvious race condition here between the time we stat(2) the
   * file and us writing the file. But we rename the file into place
   * atomically and all files created need to contain the same data anyway,
   * so this is perfectly fine, although a bit of a resource waste. Locking
   * seems like pointless complication.
   */
  elog(DEBUG1, "serializing snapshot to %s", path);

  /* to make sure only we will write to this tempfile, include pid */
  sprintf(tmppath, "%s/%X-%X.snap.%d.tmp",
      PG_LOGICAL_SNAPSHOTS_DIR,
      LSN_FORMAT_ARGS(lsn), MyProcPid);

  /*
   * Unlink temporary file if it already exists, needs to have been before a
   * crash/error since we won't enter this function twice from within a
   * single decoding slot/backend and the temporary file contains the pid of
   * the current process.
   */
  if (unlink(tmppath) != 0 && errno != ENOENT)
    ereport(ERROR,
        (errcode_for_file_access(),
         errmsg("could not remove file \"%s\": %m", tmppath)));

  old_ctx = MemoryContextSwitchTo(builder->context);

  /* Get the catalog modifying transactions that are yet not committed */
  catchange_xip = ReorderBufferGetCatalogChangesXacts(builder->reorder);
  catchange_xcnt = dclist_count(&builder->reorder->catchange_txns);

  needed_length = sizeof(SnapBuildOnDisk) +
    sizeof(TransactionId) * (builder->committed.xcnt + catchange_xcnt);

  ondisk_c = palloc0(needed_length);
  ondisk = (SnapBuildOnDisk *) ondisk_c;
  ondisk->magic = SNAPBUILD_MAGIC;
  ondisk->version = SNAPBUILD_VERSION;
  ondisk->length = needed_length;
  INIT_CRC32C(ondisk->checksum);
  COMP_CRC32C(ondisk->checksum,
        ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
        SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);
  ondisk_c += sizeof(SnapBuildOnDisk);

  memcpy(&ondisk->builder, builder, sizeof(SnapBuild));
  /* NULL-ify memory-only data */
  ondisk->builder.context = NULL;
  ondisk->builder.snapshot = NULL;
  ondisk->builder.reorder = NULL;
  ondisk->builder.committed.xip = NULL;
  ondisk->builder.catchange.xip = NULL;
  /* update catchange only on disk data */
  ondisk->builder.catchange.xcnt = catchange_xcnt;

  COMP_CRC32C(ondisk->checksum,
        &ondisk->builder,
        sizeof(SnapBuild));

  /* copy committed xacts */
  if (builder->committed.xcnt > 0)
  {
    sz = sizeof(TransactionId) * builder->committed.xcnt;
    memcpy(ondisk_c, builder->committed.xip, sz);
    COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
    ondisk_c += sz;
  }

  /* copy catalog modifying xacts */
  if (catchange_xcnt > 0)
  {
    sz = sizeof(TransactionId) * catchange_xcnt;
    memcpy(ondisk_c, catchange_xip, sz);
    COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
    ondisk_c += sz;
  }

  FIN_CRC32C(ondisk->checksum);

  /* we have valid data now, open tempfile and write it there */
  fd = OpenTransientFile(tmppath,
               O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
  if (fd < 0)
    ereport(ERROR,
        (errcode_for_file_access(),
         errmsg("could not open file \"%s\": %m", tmppath)));

  errno = 0;
  pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_WRITE);
  if ((write(fd, ondisk, needed_length)) != needed_length)
  {
    int     save_errno = errno;

    CloseTransientFile(fd);

    /* if write didn't set errno, assume problem is no disk space */
    errno = save_errno ? save_errno : ENOSPC;
    ereport(ERROR,
        (errcode_for_file_access(),
         errmsg("could not write to file \"%s\": %m", tmppath)));
  }
  pgstat_report_wait_end();

  /*
   * fsync the file before renaming so that even if we crash after this we
   * have either a fully valid file or nothing.
   *
   * It's safe to just ERROR on fsync() here because we'll retry the whole
   * operation including the writes.
   *
   * TODO: Do the fsync() via checkpoints/restartpoints, doing it here has
   * some noticeable overhead since it's performed synchronously during
   * decoding?
   */
  pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_SYNC);
  if (pg_fsync(fd) != 0)
  {
    int     save_errno = errno;

    CloseTransientFile(fd);
    errno = save_errno;
    ereport(ERROR,
        (errcode_for_file_access(),
         errmsg("could not fsync file \"%s\": %m", tmppath)));
  }
  pgstat_report_wait_end();

  if (CloseTransientFile(fd) != 0)
    ereport(ERROR,
        (errcode_for_file_access(),
         errmsg("could not close file \"%s\": %m", tmppath)));

  fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);

  /*
   * We may overwrite the work from some other backend, but that's ok, our
   * snapshot is valid as well, we'll just have done some superfluous work.
   */
  if (rename(tmppath, path) != 0)
  {
    ereport(ERROR,
        (errcode_for_file_access(),
         errmsg("could not rename file \"%s\" to \"%s\": %m",
            tmppath, path)));
  }

  /* make sure we persist */
  fsync_fname(path, false);
  fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);

  /*
   * Now there's no way we can lose the dumped state anymore, remember this
   * as a serialization point.
   */
  builder->last_serialized_snapshot = lsn;

  MemoryContextSwitchTo(old_ctx);

out:
  ReorderBufferSetRestartPoint(builder->reorder,
                 builder->last_serialized_snapshot);
  /* be tidy */
  if (ondisk)
    pfree(ondisk);
  if (catchange_xip)
    pfree(catchange_xip);
}

/*
 * Restore the logical snapshot file contents to 'ondisk'.
 *
 * 'context' is the memory context where the catalog modifying/committed xid
 * will live.
 * If 'missing_ok' is true, will not throw an error if the file is not found.
 */
bool
SnapBuildRestoreSnapshot(SnapBuildOnDisk *ondisk, XLogRecPtr lsn,
             MemoryContext context, bool missing_ok)
{
  int     fd;
  pg_crc32c checksum;
  Size    sz;
  char    path[MAXPGPATH];

  sprintf(path, "%s/%X-%X.snap",
      PG_LOGICAL_SNAPSHOTS_DIR,
      LSN_FORMAT_ARGS(lsn));

  fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);

  if (fd < 0)
  {
    if (missing_ok && errno == ENOENT)
      return false;

    ereport(ERROR,
        (errcode_for_file_access(),
         errmsg("could not open file \"%s\": %m", path)));
  }

  /* ----
   * Make sure the snapshot had been stored safely to disk, that's normally
   * cheap.
   * Note that we do not need PANIC here, nobody will be able to use the
   * slot without fsyncing, and saving it won't succeed without an fsync()
   * either...
   * ----
   */
  fsync_fname(path, false);
  fsync_fname(PG_LOGICAL_SNAPSHOTS_DIR, true);

  /* read statically sized portion of snapshot */
  SnapBuildRestoreContents(fd, ondisk, SnapBuildOnDiskConstantSize, path);

  if (ondisk->magic != SNAPBUILD_MAGIC)
    ereport(ERROR,
        (errcode(ERRCODE_DATA_CORRUPTED),
         errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
            path, ondisk->magic, SNAPBUILD_MAGIC)));

  if (ondisk->version != SNAPBUILD_VERSION)
    ereport(ERROR,
        (errcode(ERRCODE_DATA_CORRUPTED),
         errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
            path, ondisk->version, SNAPBUILD_VERSION)));

  INIT_CRC32C(checksum);
  COMP_CRC32C(checksum,
        ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
        SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);

  /* read SnapBuild */
  SnapBuildRestoreContents(fd, &ondisk->builder, sizeof(SnapBuild), path);
  COMP_CRC32C(checksum, &ondisk->builder, sizeof(SnapBuild));

  /* restore committed xacts information */
  if (ondisk->builder.committed.xcnt > 0)
  {
    sz = sizeof(TransactionId) * ondisk->builder.committed.xcnt;
    ondisk->builder.committed.xip = MemoryContextAllocZero(context, sz);
    SnapBuildRestoreContents(fd, ondisk->builder.committed.xip, sz, path);
    COMP_CRC32C(checksum, ondisk->builder.committed.xip, sz);
  }

  /* restore catalog modifying xacts information */
  if (ondisk->builder.catchange.xcnt > 0)
  {
    sz = sizeof(TransactionId) * ondisk->builder.catchange.xcnt;
    ondisk->builder.catchange.xip = MemoryContextAllocZero(context, sz);
    SnapBuildRestoreContents(fd, ondisk->builder.catchange.xip, sz, path);
    COMP_CRC32C(checksum, ondisk->builder.catchange.xip, sz);
  }

  if (CloseTransientFile(fd) != 0)
    ereport(ERROR,
        (errcode_for_file_access(),
         errmsg("could not close file \"%s\": %m", path)));

  FIN_CRC32C(checksum);

  /* verify checksum of what we've read */
  if (!EQ_CRC32C(checksum, ondisk->checksum))
    ereport(ERROR,
        (errcode(ERRCODE_DATA_CORRUPTED),
         errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
            path, checksum, ondisk->checksum)));

  return true;
}

/*
 * Restore a snapshot into 'builder' if previously one has been stored at the
 * location indicated by 'lsn'. Returns true if successful, false otherwise.
 */
static bool
SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
{
  SnapBuildOnDisk ondisk;

  /* no point in loading a snapshot if we're already there */
  if (builder->state == SNAPBUILD_CONSISTENT)
    return false;

  /* validate and restore the snapshot to 'ondisk' */
  if (!SnapBuildRestoreSnapshot(&ondisk, lsn, builder->context, true))
    return false;

  /*
   * ok, we now have a sensible snapshot here, figure out if it has more
   * information than we have.
   */

  /*
   * We are only interested in consistent snapshots for now, comparing
   * whether one incomplete snapshot is more "advanced" seems to be
   * unnecessarily complex.
   */
  if (ondisk.builder.state < SNAPBUILD_CONSISTENT)
    goto snapshot_not_interesting;

  /*
   * Don't use a snapshot that requires an xmin that we cannot guarantee to
   * be available.
   */
  if (TransactionIdPrecedes(ondisk.builder.xmin, builder->initial_xmin_horizon))
    goto snapshot_not_interesting;

  /*
   * Consistent snapshots have no next phase. Reset next_phase_at as it is
   * possible that an old value may remain.
   */
  Assert(ondisk.builder.next_phase_at == InvalidTransactionId);
  builder->next_phase_at = InvalidTransactionId;

  /* ok, we think the snapshot is sensible, copy over everything important */
  builder->xmin = ondisk.builder.xmin;
  builder->xmax = ondisk.builder.xmax;
  builder->state = ondisk.builder.state;

  builder->committed.xcnt = ondisk.builder.committed.xcnt;
  /* We only allocated/stored xcnt, not xcnt_space xids ! */
  /* don't overwrite preallocated xip, if we don't have anything here */
  if (builder->committed.xcnt > 0)
  {
    pfree(builder->committed.xip);
    builder->committed.xcnt_space = ondisk.builder.committed.xcnt;
    builder->committed.xip = ondisk.builder.committed.xip;
  }
  ondisk.builder.committed.xip = NULL;

  /* set catalog modifying transactions */
  if (builder->catchange.xip)
    pfree(builder->catchange.xip);
  builder->catchange.xcnt = ondisk.builder.catchange.xcnt;
  builder->catchange.xip = ondisk.builder.catchange.xip;
  ondisk.builder.catchange.xip = NULL;

  /* our snapshot is not interesting anymore, build a new one */
  if (builder->snapshot != NULL)
  {
    SnapBuildSnapDecRefcount(builder->snapshot);
  }
  builder->snapshot = SnapBuildBuildSnapshot(builder);
  SnapBuildSnapIncRefcount(builder->snapshot);

  ReorderBufferSetRestartPoint(builder->reorder, lsn);

  Assert(builder->state == SNAPBUILD_CONSISTENT);

  ereport(LogicalDecodingLogLevel(),
      errmsg("logical decoding found consistent point at %X/%08X",
           LSN_FORMAT_ARGS(lsn)),
      errdetail("Logical decoding will begin using saved snapshot."));
  return true;

snapshot_not_interesting:
  if (ondisk.builder.committed.xip != NULL)
    pfree(ondisk.builder.committed.xip);
  if (ondisk.builder.catchange.xip != NULL)
    pfree(ondisk.builder.catchange.xip);
  return false;
}

/*
 * Read the contents of the serialized snapshot to 'dest'.
 */
static void
SnapBuildRestoreContents(int fd, void *dest, Size size, const char *path)
{
  ssize_t   readBytes;

  pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_READ);
  readBytes = read(fd, dest, size);
  pgstat_report_wait_end();
  if (readBytes != size)
  {
    int     save_errno = errno;

    CloseTransientFile(fd);

    if (readBytes < 0)
    {
      errno = save_errno;
      ereport(ERROR,
          (errcode_for_file_access(),
           errmsg("could not read file \"%s\": %m", path)));
    }
    else
      ereport(ERROR,
          (errcode(ERRCODE_DATA_CORRUPTED),
           errmsg("could not read file \"%s\": read %zd of %zu",
              path, readBytes, size)));
  }
}

/*
 * Remove all serialized snapshots that are not required anymore because no
 * slot can need them. This doesn't actually have to run during a checkpoint,
 * but it's a convenient point to schedule this.
 *
 * NB: We run this during checkpoints even if logical decoding is disabled so
 * we cleanup old slots at some point after it got disabled.
 */
void
CheckPointSnapBuild(void)
{
  XLogRecPtr  cutoff;
  XLogRecPtr  redo;
  DIR      *snap_dir;
  struct dirent *snap_de;
  char    path[MAXPGPATH + sizeof(PG_LOGICAL_SNAPSHOTS_DIR)];

  /*
   * We start off with a minimum of the last redo pointer. No new
   * replication slot will start before that, so that's a safe upper bound
   * for removal.
   */
  redo = GetRedoRecPtr();

  /* now check for the restart ptrs from existing slots */
  cutoff = ReplicationSlotsComputeLogicalRestartLSN();

  /* don't start earlier than the restart lsn */
  if (redo < cutoff)
    cutoff = redo;

  snap_dir = AllocateDir(PG_LOGICAL_SNAPSHOTS_DIR);
  while ((snap_de = ReadDir(snap_dir, PG_LOGICAL_SNAPSHOTS_DIR)) != NULL)
  {
    uint32    hi;
    uint32    lo;
    XLogRecPtr  lsn;
    PGFileType  de_type;

    if (strcmp(snap_de->d_name, ".") == 0 ||
      strcmp(snap_de->d_name, "..") == 0)
      continue;

    snprintf(path, sizeof(path), "%s/%s", PG_LOGICAL_SNAPSHOTS_DIR, snap_de->d_name);
    de_type = get_dirent_type(path, snap_de, false, DEBUG1);

    if (de_type != PGFILETYPE_ERROR && de_type != PGFILETYPE_REG)
    {
      elog(DEBUG1, "only regular files expected: %s", path);
      continue;
    }

    /*
     * temporary filenames from SnapBuildSerialize() include the LSN and
     * everything but are postfixed by .$pid.tmp. We can just remove them
     * the same as other files because there can be none that are
     * currently being written that are older than cutoff.
     *
     * We just log a message if a file doesn't fit the pattern, it's
     * probably some editors lock/state file or similar...
     */
    if (sscanf(snap_de->d_name, "%X-%X.snap", &hi, &lo) != 2)
    {
      ereport(LOG,
          (errmsg("could not parse file name \"%s\"", path)));
      continue;
    }

    lsn = ((uint64) hi) << 32 | lo;

    /* check whether we still need it */
    if (lsn < cutoff || !XLogRecPtrIsValid(cutoff))
    {
      elog(DEBUG1, "removing snapbuild snapshot %s", path);

      /*
       * It's not particularly harmful, though strange, if we can't
       * remove the file here. Don't prevent the checkpoint from
       * completing, that'd be a cure worse than the disease.
       */
      if (unlink(path) < 0)
      {
        ereport(LOG,
            (errcode_for_file_access(),
             errmsg("could not remove file \"%s\": %m",
                path)));
        continue;
      }
    }
  }
  FreeDir(snap_dir);
}

/*
 * Check if a logical snapshot at the specified point has been serialized.
 */
bool
SnapBuildSnapshotExists(XLogRecPtr lsn)
{
  char    path[MAXPGPATH];
  int     ret;
  struct stat stat_buf;

  sprintf(path, "%s/%X-%X.snap",
      PG_LOGICAL_SNAPSHOTS_DIR,
      LSN_FORMAT_ARGS(lsn));

  ret = stat(path, &stat_buf);

  if (ret != 0 && errno != ENOENT)
    ereport(ERROR,
        (errcode_for_file_access(),
         errmsg("could not stat file \"%s\": %m", path)));

  return ret == 0;
}

Coverage Report

Created: 2026-11-12 07:20