LCOV - code coverage report
Current view: top level - src/backend/replication/logical - snapbuild.c (source / functions) Hit Total Coverage
Test: PostgreSQL 15devel Lines: 383 496 77.2 %
Date: 2021-12-09 04:09:06 Functions: 27 28 96.4 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * snapbuild.c
       4             :  *
       5             :  *    Infrastructure for building historic catalog snapshots based on contents
       6             :  *    of the WAL, for the purpose of decoding heapam.c style values in the
       7             :  *    WAL.
       8             :  *
       9             :  * NOTES:
      10             :  *
      11             :  * We build snapshots which can *only* be used to read catalog contents and we
      12             :  * do so by reading and interpreting the WAL stream. The aim is to build a
      13             :  * snapshot that behaves the same as a freshly taken MVCC snapshot would have
      14             :  * at the time the XLogRecord was generated.
      15             :  *
      16             :  * To build the snapshots we reuse the infrastructure built for Hot
      17             :  * Standby. The in-memory snapshots we build look different than HS' because
      18             :  * we have different needs. To successfully decode data from the WAL we only
      19             :  * need to access catalog tables and (sys|rel|cat)cache, not the actual user
      20             :  * tables since the data we decode is wholly contained in the WAL
      21             :  * records. Also, our snapshots need to be different in comparison to normal
      22             :  * MVCC ones because in contrast to those we cannot fully rely on the clog and
      23             :  * pg_subtrans for information about committed transactions because they might
      24             :  * commit in the future from the POV of the WAL entry we're currently
      25             :  * decoding. This definition has the advantage that we only need to prevent
      26             :  * removal of catalog rows, while normal table's rows can still be
      27             :  * removed. This is achieved by using the replication slot mechanism.
      28             :  *
      29             :  * As the percentage of transactions modifying the catalog normally is fairly
      30             :  * small in comparisons to ones only manipulating user data, we keep track of
      31             :  * the committed catalog modifying ones inside [xmin, xmax) instead of keeping
      32             :  * track of all running transactions like it's done in a normal snapshot. Note
      33             :  * that we're generally only looking at transactions that have acquired an
      34             :  * xid. That is we keep a list of transactions between snapshot->(xmin, xmax)
      35             :  * that we consider committed, everything else is considered aborted/in
      36             :  * progress. That also allows us not to care about subtransactions before they
      37             :  * have committed which means this module, in contrast to HS, doesn't have to
      38             :  * care about suboverflowed subtransactions and similar.
      39             :  *
      40             :  * One complexity of doing this is that to e.g. handle mixed DDL/DML
      41             :  * transactions we need Snapshots that see intermediate versions of the
      42             :  * catalog in a transaction. During normal operation this is achieved by using
      43             :  * CommandIds/cmin/cmax. The problem with that however is that for space
      44             :  * efficiency reasons only one value of that is stored
      45             :  * (cf. combocid.c). Since combo CIDs are only available in memory we log
      46             :  * additional information which allows us to get the original (cmin, cmax)
      47             :  * pair during visibility checks. Check the reorderbuffer.c's comment above
      48             :  * ResolveCminCmaxDuringDecoding() for details.
      49             :  *
      50             :  * To facilitate all this we need our own visibility routine, as the normal
      51             :  * ones are optimized for different usecases.
      52             :  *
      53             :  * To replace the normal catalog snapshots with decoding ones use the
      54             :  * SetupHistoricSnapshot() and TeardownHistoricSnapshot() functions.
      55             :  *
      56             :  *
      57             :  *
      58             :  * The snapbuild machinery is starting up in several stages, as illustrated
      59             :  * by the following graph describing the SnapBuild->state transitions:
      60             :  *
      61             :  *         +-------------------------+
      62             :  *    +----|         START           |-------------+
      63             :  *    |    +-------------------------+             |
      64             :  *    |                 |                          |
      65             :  *    |                 |                          |
      66             :  *    |        running_xacts #1                    |
      67             :  *    |                 |                          |
      68             :  *    |                 |                          |
      69             :  *    |                 v                          |
      70             :  *    |    +-------------------------+             v
      71             :  *    |    |   BUILDING_SNAPSHOT     |------------>|
      72             :  *    |    +-------------------------+             |
      73             :  *    |                 |                          |
      74             :  *    |                 |                          |
      75             :  *    | running_xacts #2, xacts from #1 finished   |
      76             :  *    |                 |                          |
      77             :  *    |                 |                          |
      78             :  *    |                 v                          |
      79             :  *    |    +-------------------------+             v
      80             :  *    |    |       FULL_SNAPSHOT     |------------>|
      81             :  *    |    +-------------------------+             |
      82             :  *    |                 |                          |
      83             :  * running_xacts        |                      saved snapshot
      84             :  * with zero xacts      |                 at running_xacts's lsn
      85             :  *    |                 |                          |
      86             :  *    | running_xacts with xacts from #2 finished  |
      87             :  *    |                 |                          |
      88             :  *    |                 v                          |
      89             :  *    |    +-------------------------+             |
      90             :  *    +--->|SNAPBUILD_CONSISTENT  |<------------+
      91             :  *         +-------------------------+
      92             :  *
      93             :  * Initially the machinery is in the START stage. When an xl_running_xacts
      94             :  * record is read that is sufficiently new (above the safe xmin horizon),
      95             :  * there's a state transition. If there were no running xacts when the
      96             :  * running_xacts record was generated, we'll directly go into CONSISTENT
      97             :  * state, otherwise we'll switch to the BUILDING_SNAPSHOT state. Having a full
      98             :  * snapshot means that all transactions that start henceforth can be decoded
      99             :  * in their entirety, but transactions that started previously can't. In
     100             :  * FULL_SNAPSHOT we'll switch into CONSISTENT once all those previously
     101             :  * running transactions have committed or aborted.
     102             :  *
     103             :  * Only transactions that commit after CONSISTENT state has been reached will
     104             :  * be replayed, even though they might have started while still in
     105             :  * FULL_SNAPSHOT. That ensures that we'll reach a point where no previous
     106             :  * changes has been exported, but all the following ones will be. That point
     107             :  * is a convenient point to initialize replication from, which is why we
     108             :  * export a snapshot at that point, which *can* be used to read normal data.
     109             :  *
     110             :  * Copyright (c) 2012-2021, PostgreSQL Global Development Group
     111             :  *
     112             :  * IDENTIFICATION
     113             :  *    src/backend/replication/snapbuild.c
     114             :  *
     115             :  *-------------------------------------------------------------------------
     116             :  */
     117             : 
     118             : #include "postgres.h"
     119             : 
     120             : #include <sys/stat.h>
     121             : #include <unistd.h>
     122             : 
     123             : #include "access/heapam_xlog.h"
     124             : #include "access/transam.h"
     125             : #include "access/xact.h"
     126             : #include "miscadmin.h"
     127             : #include "pgstat.h"
     128             : #include "replication/logical.h"
     129             : #include "replication/reorderbuffer.h"
     130             : #include "replication/snapbuild.h"
     131             : #include "storage/block.h"        /* debugging output */
     132             : #include "storage/fd.h"
     133             : #include "storage/lmgr.h"
     134             : #include "storage/proc.h"
     135             : #include "storage/procarray.h"
     136             : #include "storage/standby.h"
     137             : #include "utils/builtins.h"
     138             : #include "utils/memutils.h"
     139             : #include "utils/snapmgr.h"
     140             : #include "utils/snapshot.h"
     141             : 
     142             : /*
     143             :  * This struct contains the current state of the snapshot building
     144             :  * machinery. Besides a forward declaration in the header, it is not exposed
     145             :  * to the public, so we can easily change its contents.
     146             :  */
     147             : struct SnapBuild
     148             : {
     149             :     /* how far are we along building our first full snapshot */
     150             :     SnapBuildState state;
     151             : 
     152             :     /* private memory context used to allocate memory for this module. */
     153             :     MemoryContext context;
     154             : 
     155             :     /* all transactions < than this have committed/aborted */
     156             :     TransactionId xmin;
     157             : 
     158             :     /* all transactions >= than this are uncommitted */
     159             :     TransactionId xmax;
     160             : 
     161             :     /*
     162             :      * Don't replay commits from an LSN < this LSN. This can be set externally
     163             :      * but it will also be advanced (never retreat) from within snapbuild.c.
     164             :      */
     165             :     XLogRecPtr  start_decoding_at;
     166             : 
     167             :     /*
     168             :      * LSN at which two-phase decoding was enabled or LSN at which we found a
     169             :      * consistent point at the time of slot creation.
     170             :      *
     171             :      * The prepared transactions, that were skipped because previously
     172             :      * two-phase was not enabled or are not covered by initial snapshot, need
     173             :      * to be sent later along with commit prepared and they must be before
     174             :      * this point.
     175             :      */
     176             :     XLogRecPtr  two_phase_at;
     177             : 
     178             :     /*
     179             :      * Don't start decoding WAL until the "xl_running_xacts" information
     180             :      * indicates there are no running xids with an xid smaller than this.
     181             :      */
     182             :     TransactionId initial_xmin_horizon;
     183             : 
     184             :     /* Indicates if we are building full snapshot or just catalog one. */
     185             :     bool        building_full_snapshot;
     186             : 
     187             :     /*
     188             :      * Snapshot that's valid to see the catalog state seen at this moment.
     189             :      */
     190             :     Snapshot    snapshot;
     191             : 
     192             :     /*
     193             :      * LSN of the last location we are sure a snapshot has been serialized to.
     194             :      */
     195             :     XLogRecPtr  last_serialized_snapshot;
     196             : 
     197             :     /*
     198             :      * The reorderbuffer we need to update with usable snapshots et al.
     199             :      */
     200             :     ReorderBuffer *reorder;
     201             : 
     202             :     /*
     203             :      * TransactionId at which the next phase of initial snapshot building will
     204             :      * happen. InvalidTransactionId if not known (i.e. SNAPBUILD_START), or
     205             :      * when no next phase necessary (SNAPBUILD_CONSISTENT).
     206             :      */
     207             :     TransactionId next_phase_at;
     208             : 
     209             :     /*
     210             :      * Array of transactions which could have catalog changes that committed
     211             :      * between xmin and xmax.
     212             :      */
     213             :     struct
     214             :     {
     215             :         /* number of committed transactions */
     216             :         size_t      xcnt;
     217             : 
     218             :         /* available space for committed transactions */
     219             :         size_t      xcnt_space;
     220             : 
     221             :         /*
     222             :          * Until we reach a CONSISTENT state, we record commits of all
     223             :          * transactions, not just the catalog changing ones. Record when that
     224             :          * changes so we know we cannot export a snapshot safely anymore.
     225             :          */
     226             :         bool        includes_all_transactions;
     227             : 
     228             :         /*
     229             :          * Array of committed transactions that have modified the catalog.
     230             :          *
     231             :          * As this array is frequently modified we do *not* keep it in
     232             :          * xidComparator order. Instead we sort the array when building &
     233             :          * distributing a snapshot.
     234             :          *
     235             :          * TODO: It's unclear whether that reasoning has much merit. Every
     236             :          * time we add something here after becoming consistent will also
     237             :          * require distributing a snapshot. Storing them sorted would
     238             :          * potentially also make it easier to purge (but more complicated wrt
     239             :          * wraparound?). Should be improved if sorting while building the
     240             :          * snapshot shows up in profiles.
     241             :          */
     242             :         TransactionId *xip;
     243             :     }           committed;
     244             : };
     245             : 
     246             : /*
     247             :  * Starting a transaction -- which we need to do while exporting a snapshot --
     248             :  * removes knowledge about the previously used resowner, so we save it here.
     249             :  */
     250             : static ResourceOwner SavedResourceOwnerDuringExport = NULL;
     251             : static bool ExportInProgress = false;
     252             : 
     253             : /* ->committed manipulation */
     254             : static void SnapBuildPurgeCommittedTxn(SnapBuild *builder);
     255             : 
     256             : /* snapshot building/manipulation/distribution functions */
     257             : static Snapshot SnapBuildBuildSnapshot(SnapBuild *builder);
     258             : 
     259             : static void SnapBuildFreeSnapshot(Snapshot snap);
     260             : 
     261             : static void SnapBuildSnapIncRefcount(Snapshot snap);
     262             : 
     263             : static void SnapBuildDistributeNewCatalogSnapshot(SnapBuild *builder, XLogRecPtr lsn);
     264             : 
     265             : /* xlog reading helper functions for SnapBuildProcessRunningXacts */
     266             : static bool SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running);
     267             : static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff);
     268             : 
     269             : /* serialization functions */
     270             : static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn);
     271             : static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn);
     272             : 
     273             : /*
     274             :  * Allocate a new snapshot builder.
     275             :  *
     276             :  * xmin_horizon is the xid >= which we can be sure no catalog rows have been
     277             :  * removed, start_lsn is the LSN >= we want to replay commits.
     278             :  */
     279             : SnapBuild *
     280        1070 : AllocateSnapshotBuilder(ReorderBuffer *reorder,
     281             :                         TransactionId xmin_horizon,
     282             :                         XLogRecPtr start_lsn,
     283             :                         bool need_full_snapshot,
     284             :                         XLogRecPtr two_phase_at)
     285             : {
     286             :     MemoryContext context;
     287             :     MemoryContext oldcontext;
     288             :     SnapBuild  *builder;
     289             : 
     290             :     /* allocate memory in own context, to have better accountability */
     291        1070 :     context = AllocSetContextCreate(CurrentMemoryContext,
     292             :                                     "snapshot builder context",
     293             :                                     ALLOCSET_DEFAULT_SIZES);
     294        1070 :     oldcontext = MemoryContextSwitchTo(context);
     295             : 
     296        1070 :     builder = palloc0(sizeof(SnapBuild));
     297             : 
     298        1070 :     builder->state = SNAPBUILD_START;
     299        1070 :     builder->context = context;
     300        1070 :     builder->reorder = reorder;
     301             :     /* Other struct members initialized by zeroing via palloc0 above */
     302             : 
     303        1070 :     builder->committed.xcnt = 0;
     304        1070 :     builder->committed.xcnt_space = 128; /* arbitrary number */
     305        1070 :     builder->committed.xip =
     306        1070 :         palloc0(builder->committed.xcnt_space * sizeof(TransactionId));
     307        1070 :     builder->committed.includes_all_transactions = true;
     308             : 
     309        1070 :     builder->initial_xmin_horizon = xmin_horizon;
     310        1070 :     builder->start_decoding_at = start_lsn;
     311        1070 :     builder->building_full_snapshot = need_full_snapshot;
     312        1070 :     builder->two_phase_at = two_phase_at;
     313             : 
     314        1070 :     MemoryContextSwitchTo(oldcontext);
     315             : 
     316        1070 :     return builder;
     317             : }
     318             : 
     319             : /*
     320             :  * Free a snapshot builder.
     321             :  */
     322             : void
     323         938 : FreeSnapshotBuilder(SnapBuild *builder)
     324             : {
     325         938 :     MemoryContext context = builder->context;
     326             : 
     327             :     /* free snapshot explicitly, that contains some error checking */
     328         938 :     if (builder->snapshot != NULL)
     329             :     {
     330         322 :         SnapBuildSnapDecRefcount(builder->snapshot);
     331         322 :         builder->snapshot = NULL;
     332             :     }
     333             : 
     334             :     /* other resources are deallocated via memory context reset */
     335         938 :     MemoryContextDelete(context);
     336         938 : }
     337             : 
     338             : /*
     339             :  * Free an unreferenced snapshot that has previously been built by us.
     340             :  */
     341             : static void
     342        1688 : SnapBuildFreeSnapshot(Snapshot snap)
     343             : {
     344             :     /* make sure we don't get passed an external snapshot */
     345             :     Assert(snap->snapshot_type == SNAPSHOT_HISTORIC_MVCC);
     346             : 
     347             :     /* make sure nobody modified our snapshot */
     348             :     Assert(snap->curcid == FirstCommandId);
     349             :     Assert(!snap->suboverflowed);
     350             :     Assert(!snap->takenDuringRecovery);
     351             :     Assert(snap->regd_count == 0);
     352             : 
     353             :     /* slightly more likely, so it's checked even without c-asserts */
     354        1688 :     if (snap->copied)
     355           0 :         elog(ERROR, "cannot free a copied snapshot");
     356             : 
     357        1688 :     if (snap->active_count)
     358           0 :         elog(ERROR, "cannot free an active snapshot");
     359             : 
     360        1688 :     pfree(snap);
     361        1688 : }
     362             : 
     363             : /*
     364             :  * In which state of snapshot building are we?
     365             :  */
     366             : SnapBuildState
     367     3561294 : SnapBuildCurrentState(SnapBuild *builder)
     368             : {
     369     3561294 :     return builder->state;
     370             : }
     371             : 
     372             : /*
     373             :  * Return the LSN at which the two-phase decoding was first enabled.
     374             :  */
     375             : XLogRecPtr
     376          48 : SnapBuildGetTwoPhaseAt(SnapBuild *builder)
     377             : {
     378          48 :     return builder->two_phase_at;
     379             : }
     380             : 
     381             : /*
     382             :  * Set the LSN at which two-phase decoding is enabled.
     383             :  */
     384             : void
     385           8 : SnapBuildSetTwoPhaseAt(SnapBuild *builder, XLogRecPtr ptr)
     386             : {
     387           8 :     builder->two_phase_at = ptr;
     388           8 : }
     389             : 
     390             : /*
     391             :  * Should the contents of transaction ending at 'ptr' be decoded?
     392             :  */
     393             : bool
     394      398258 : SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
     395             : {
     396      398258 :     return ptr < builder->start_decoding_at;
     397             : }
     398             : 
     399             : /*
     400             :  * Increase refcount of a snapshot.
     401             :  *
     402             :  * This is used when handing out a snapshot to some external resource or when
     403             :  * adding a Snapshot as builder->snapshot.
     404             :  */
     405             : static void
     406        6902 : SnapBuildSnapIncRefcount(Snapshot snap)
     407             : {
     408        6902 :     snap->active_count++;
     409        6902 : }
     410             : 
     411             : /*
     412             :  * Decrease refcount of a snapshot and free if the refcount reaches zero.
     413             :  *
     414             :  * Externally visible, so that external resources that have been handed an
     415             :  * IncRef'ed Snapshot can adjust its refcount easily.
     416             :  */
     417             : void
     418        6742 : SnapBuildSnapDecRefcount(Snapshot snap)
     419             : {
     420             :     /* make sure we don't get passed an external snapshot */
     421             :     Assert(snap->snapshot_type == SNAPSHOT_HISTORIC_MVCC);
     422             : 
     423             :     /* make sure nobody modified our snapshot */
     424             :     Assert(snap->curcid == FirstCommandId);
     425             :     Assert(!snap->suboverflowed);
     426             :     Assert(!snap->takenDuringRecovery);
     427             : 
     428             :     Assert(snap->regd_count == 0);
     429             : 
     430             :     Assert(snap->active_count > 0);
     431             : 
     432             :     /* slightly more likely, so it's checked even without casserts */
     433        6742 :     if (snap->copied)
     434           0 :         elog(ERROR, "cannot free a copied snapshot");
     435             : 
     436        6742 :     snap->active_count--;
     437        6742 :     if (snap->active_count == 0)
     438        1688 :         SnapBuildFreeSnapshot(snap);
     439        6742 : }
     440             : 
     441             : /*
     442             :  * Build a new snapshot, based on currently committed catalog-modifying
     443             :  * transactions.
     444             :  *
     445             :  * In-progress transactions with catalog access are *not* allowed to modify
     446             :  * these snapshots; they have to copy them and fill in appropriate ->curcid
     447             :  * and ->subxip/subxcnt values.
     448             :  */
     449             : static Snapshot
     450        2010 : SnapBuildBuildSnapshot(SnapBuild *builder)
     451             : {
     452             :     Snapshot    snapshot;
     453             :     Size        ssize;
     454             : 
     455             :     Assert(builder->state >= SNAPBUILD_FULL_SNAPSHOT);
     456             : 
     457        2010 :     ssize = sizeof(SnapshotData)
     458        2010 :         + sizeof(TransactionId) * builder->committed.xcnt
     459        2010 :         + sizeof(TransactionId) * 1 /* toplevel xid */ ;
     460             : 
     461        2010 :     snapshot = MemoryContextAllocZero(builder->context, ssize);
     462             : 
     463        2010 :     snapshot->snapshot_type = SNAPSHOT_HISTORIC_MVCC;
     464             : 
     465             :     /*
     466             :      * We misuse the original meaning of SnapshotData's xip and subxip fields
     467             :      * to make the more fitting for our needs.
     468             :      *
     469             :      * In the 'xip' array we store transactions that have to be treated as
     470             :      * committed. Since we will only ever look at tuples from transactions
     471             :      * that have modified the catalog it's more efficient to store those few
     472             :      * that exist between xmin and xmax (frequently there are none).
     473             :      *
     474             :      * Snapshots that are used in transactions that have modified the catalog
     475             :      * also use the 'subxip' array to store their toplevel xid and all the
     476             :      * subtransaction xids so we can recognize when we need to treat rows as
     477             :      * visible that are not in xip but still need to be visible. Subxip only
     478             :      * gets filled when the transaction is copied into the context of a
     479             :      * catalog modifying transaction since we otherwise share a snapshot
     480             :      * between transactions. As long as a txn hasn't modified the catalog it
     481             :      * doesn't need to treat any uncommitted rows as visible, so there is no
     482             :      * need for those xids.
     483             :      *
     484             :      * Both arrays are qsort'ed so that we can use bsearch() on them.
     485             :      */
     486             :     Assert(TransactionIdIsNormal(builder->xmin));
     487             :     Assert(TransactionIdIsNormal(builder->xmax));
     488             : 
     489        2010 :     snapshot->xmin = builder->xmin;
     490        2010 :     snapshot->xmax = builder->xmax;
     491             : 
     492             :     /* store all transactions to be treated as committed by this snapshot */
     493        2010 :     snapshot->xip =
     494        2010 :         (TransactionId *) ((char *) snapshot + sizeof(SnapshotData));
     495        2010 :     snapshot->xcnt = builder->committed.xcnt;
     496        2010 :     memcpy(snapshot->xip,
     497        2010 :            builder->committed.xip,
     498        2010 :            builder->committed.xcnt * sizeof(TransactionId));
     499             : 
     500             :     /* sort so we can bsearch() */
     501        2010 :     qsort(snapshot->xip, snapshot->xcnt, sizeof(TransactionId), xidComparator);
     502             : 
     503             :     /*
     504             :      * Initially, subxip is empty, i.e. it's a snapshot to be used by
     505             :      * transactions that don't modify the catalog. Will be filled by
     506             :      * ReorderBufferCopySnap() if necessary.
     507             :      */
     508        2010 :     snapshot->subxcnt = 0;
     509        2010 :     snapshot->subxip = NULL;
     510             : 
     511        2010 :     snapshot->suboverflowed = false;
     512        2010 :     snapshot->takenDuringRecovery = false;
     513        2010 :     snapshot->copied = false;
     514        2010 :     snapshot->curcid = FirstCommandId;
     515        2010 :     snapshot->active_count = 0;
     516        2010 :     snapshot->regd_count = 0;
     517        2010 :     snapshot->snapXactCompletionCount = 0;
     518             : 
     519        2010 :     return snapshot;
     520             : }
     521             : 
     522             : /*
     523             :  * Build the initial slot snapshot and convert it to a normal snapshot that
     524             :  * is understood by HeapTupleSatisfiesMVCC.
     525             :  *
     526             :  * The snapshot will be usable directly in current transaction or exported
     527             :  * for loading in different transaction.
     528             :  */
     529             : Snapshot
     530         176 : SnapBuildInitialSnapshot(SnapBuild *builder)
     531             : {
     532             :     Snapshot    snap;
     533             :     TransactionId xid;
     534             :     TransactionId *newxip;
     535         176 :     int         newxcnt = 0;
     536             : 
     537             :     Assert(!FirstSnapshotSet);
     538             :     Assert(XactIsoLevel == XACT_REPEATABLE_READ);
     539             : 
     540         176 :     if (builder->state != SNAPBUILD_CONSISTENT)
     541           0 :         elog(ERROR, "cannot build an initial slot snapshot before reaching a consistent state");
     542             : 
     543         176 :     if (!builder->committed.includes_all_transactions)
     544           0 :         elog(ERROR, "cannot build an initial slot snapshot, not all transactions are monitored anymore");
     545             : 
     546             :     /* so we don't overwrite the existing value */
     547         176 :     if (TransactionIdIsValid(MyProc->xmin))
     548           0 :         elog(ERROR, "cannot build an initial slot snapshot when MyProc->xmin already is valid");
     549             : 
     550         176 :     snap = SnapBuildBuildSnapshot(builder);
     551             : 
     552             :     /*
     553             :      * We know that snap->xmin is alive, enforced by the logical xmin
     554             :      * mechanism. Due to that we can do this without locks, we're only
     555             :      * changing our own value.
     556             :      */
     557             : #ifdef USE_ASSERT_CHECKING
     558             :     {
     559             :         TransactionId safeXid;
     560             : 
     561             :         LWLockAcquire(ProcArrayLock, LW_SHARED);
     562             :         safeXid = GetOldestSafeDecodingTransactionId(false);
     563             :         LWLockRelease(ProcArrayLock);
     564             : 
     565             :         Assert(TransactionIdPrecedesOrEquals(safeXid, snap->xmin));
     566             :     }
     567             : #endif
     568             : 
     569         176 :     MyProc->xmin = snap->xmin;
     570             : 
     571             :     /* allocate in transaction context */
     572             :     newxip = (TransactionId *)
     573         176 :         palloc(sizeof(TransactionId) * GetMaxSnapshotXidCount());
     574             : 
     575             :     /*
     576             :      * snapbuild.c builds transactions in an "inverted" manner, which means it
     577             :      * stores committed transactions in ->xip, not ones in progress. Build a
     578             :      * classical snapshot by marking all non-committed transactions as
     579             :      * in-progress. This can be expensive.
     580             :      */
     581         176 :     for (xid = snap->xmin; NormalTransactionIdPrecedes(xid, snap->xmax);)
     582             :     {
     583             :         void       *test;
     584             : 
     585             :         /*
     586             :          * Check whether transaction committed using the decoding snapshot
     587             :          * meaning of ->xip.
     588             :          */
     589           0 :         test = bsearch(&xid, snap->xip, snap->xcnt,
     590             :                        sizeof(TransactionId), xidComparator);
     591             : 
     592           0 :         if (test == NULL)
     593             :         {
     594           0 :             if (newxcnt >= GetMaxSnapshotXidCount())
     595           0 :                 ereport(ERROR,
     596             :                         (errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
     597             :                          errmsg("initial slot snapshot too large")));
     598             : 
     599           0 :             newxip[newxcnt++] = xid;
     600             :         }
     601             : 
     602           0 :         TransactionIdAdvance(xid);
     603             :     }
     604             : 
     605             :     /* adjust remaining snapshot fields as needed */
     606         176 :     snap->snapshot_type = SNAPSHOT_MVCC;
     607         176 :     snap->xcnt = newxcnt;
     608         176 :     snap->xip = newxip;
     609             : 
     610         176 :     return snap;
     611             : }
     612             : 
     613             : /*
     614             :  * Export a snapshot so it can be set in another session with SET TRANSACTION
     615             :  * SNAPSHOT.
     616             :  *
     617             :  * For that we need to start a transaction in the current backend as the
     618             :  * importing side checks whether the source transaction is still open to make
     619             :  * sure the xmin horizon hasn't advanced since then.
     620             :  */
     621             : const char *
     622           0 : SnapBuildExportSnapshot(SnapBuild *builder)
     623             : {
     624             :     Snapshot    snap;
     625             :     char       *snapname;
     626             : 
     627           0 :     if (IsTransactionOrTransactionBlock())
     628           0 :         elog(ERROR, "cannot export a snapshot from within a transaction");
     629             : 
     630           0 :     if (SavedResourceOwnerDuringExport)
     631           0 :         elog(ERROR, "can only export one snapshot at a time");
     632             : 
     633           0 :     SavedResourceOwnerDuringExport = CurrentResourceOwner;
     634           0 :     ExportInProgress = true;
     635             : 
     636           0 :     StartTransactionCommand();
     637             : 
     638             :     /* There doesn't seem to a nice API to set these */
     639           0 :     XactIsoLevel = XACT_REPEATABLE_READ;
     640           0 :     XactReadOnly = true;
     641             : 
     642           0 :     snap = SnapBuildInitialSnapshot(builder);
     643             : 
     644             :     /*
     645             :      * now that we've built a plain snapshot, make it active and use the
     646             :      * normal mechanisms for exporting it
     647             :      */
     648           0 :     snapname = ExportSnapshot(snap);
     649             : 
     650           0 :     ereport(LOG,
     651             :             (errmsg_plural("exported logical decoding snapshot: \"%s\" with %u transaction ID",
     652             :                            "exported logical decoding snapshot: \"%s\" with %u transaction IDs",
     653             :                            snap->xcnt,
     654             :                            snapname, snap->xcnt)));
     655           0 :     return snapname;
     656             : }
     657             : 
     658             : /*
     659             :  * Ensure there is a snapshot and if not build one for current transaction.
     660             :  */
     661             : Snapshot
     662          88 : SnapBuildGetOrBuildSnapshot(SnapBuild *builder, TransactionId xid)
     663             : {
     664             :     Assert(builder->state == SNAPBUILD_CONSISTENT);
     665             : 
     666             :     /* only build a new snapshot if we don't have a prebuilt one */
     667          88 :     if (builder->snapshot == NULL)
     668             :     {
     669           0 :         builder->snapshot = SnapBuildBuildSnapshot(builder);
     670             :         /* increase refcount for the snapshot builder */
     671           0 :         SnapBuildSnapIncRefcount(builder->snapshot);
     672             :     }
     673             : 
     674          88 :     return builder->snapshot;
     675             : }
     676             : 
     677             : /*
     678             :  * Reset a previously SnapBuildExportSnapshot()'ed snapshot if there is
     679             :  * any. Aborts the previously started transaction and resets the resource
     680             :  * owner back to its original value.
     681             :  */
     682             : void
     683        4098 : SnapBuildClearExportedSnapshot(void)
     684             : {
     685             :     ResourceOwner tmpResOwner;
     686             : 
     687             :     /* nothing exported, that is the usual case */
     688        4098 :     if (!ExportInProgress)
     689        4098 :         return;
     690             : 
     691           0 :     if (!IsTransactionState())
     692           0 :         elog(ERROR, "clearing exported snapshot in wrong transaction state");
     693             : 
     694             :     /*
     695             :      * AbortCurrentTransaction() takes care of resetting the snapshot state,
     696             :      * so remember SavedResourceOwnerDuringExport.
     697             :      */
     698           0 :     tmpResOwner = SavedResourceOwnerDuringExport;
     699             : 
     700             :     /* make sure nothing could have ever happened */
     701           0 :     AbortCurrentTransaction();
     702             : 
     703           0 :     CurrentResourceOwner = tmpResOwner;
     704             : }
     705             : 
     706             : /*
     707             :  * Clear snapshot export state during transaction abort.
     708             :  */
     709             : void
     710       24300 : SnapBuildResetExportedSnapshotState(void)
     711             : {
     712       24300 :     SavedResourceOwnerDuringExport = NULL;
     713       24300 :     ExportInProgress = false;
     714       24300 : }
     715             : 
     716             : /*
     717             :  * Handle the effects of a single heap change, appropriate to the current state
     718             :  * of the snapshot builder and returns whether changes made at (xid, lsn) can
     719             :  * be decoded.
     720             :  */
     721             : bool
     722     2798040 : SnapBuildProcessChange(SnapBuild *builder, TransactionId xid, XLogRecPtr lsn)
     723             : {
     724             :     /*
     725             :      * We can't handle data in transactions if we haven't built a snapshot
     726             :      * yet, so don't store them.
     727             :      */
     728     2798040 :     if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
     729           0 :         return false;
     730             : 
     731             :     /*
     732             :      * No point in keeping track of changes in transactions that we don't have
     733             :      * enough information about to decode. This means that they started before
     734             :      * we got into the SNAPBUILD_FULL_SNAPSHOT state.
     735             :      */
     736     2798046 :     if (builder->state < SNAPBUILD_CONSISTENT &&
     737           6 :         TransactionIdPrecedes(xid, builder->next_phase_at))
     738           0 :         return false;
     739             : 
     740             :     /*
     741             :      * If the reorderbuffer doesn't yet have a snapshot, add one now, it will
     742             :      * be needed to decode the change we're currently processing.
     743             :      */
     744     2798040 :     if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
     745             :     {
     746             :         /* only build a new snapshot if we don't have a prebuilt one */
     747        3648 :         if (builder->snapshot == NULL)
     748             :         {
     749         418 :             builder->snapshot = SnapBuildBuildSnapshot(builder);
     750             :             /* increase refcount for the snapshot builder */
     751         418 :             SnapBuildSnapIncRefcount(builder->snapshot);
     752             :         }
     753             : 
     754             :         /*
     755             :          * Increase refcount for the transaction we're handing the snapshot
     756             :          * out to.
     757             :          */
     758        3648 :         SnapBuildSnapIncRefcount(builder->snapshot);
     759        3648 :         ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
     760             :                                      builder->snapshot);
     761             :     }
     762             : 
     763     2798040 :     return true;
     764             : }
     765             : 
     766             : /*
     767             :  * Do CommandId/combo CID handling after reading an xl_heap_new_cid record.
     768             :  * This implies that a transaction has done some form of write to system
     769             :  * catalogs.
     770             :  */
     771             : void
     772       34822 : SnapBuildProcessNewCid(SnapBuild *builder, TransactionId xid,
     773             :                        XLogRecPtr lsn, xl_heap_new_cid *xlrec)
     774             : {
     775             :     CommandId   cid;
     776             : 
     777             :     /*
     778             :      * we only log new_cid's if a catalog tuple was modified, so mark the
     779             :      * transaction as containing catalog modifications
     780             :      */
     781       34822 :     ReorderBufferXidSetCatalogChanges(builder->reorder, xid, lsn);
     782             : 
     783       34822 :     ReorderBufferAddNewTupleCids(builder->reorder, xlrec->top_xid, lsn,
     784             :                                  xlrec->target_node, xlrec->target_tid,
     785             :                                  xlrec->cmin, xlrec->cmax,
     786             :                                  xlrec->combocid);
     787             : 
     788             :     /* figure out new command id */
     789       34822 :     if (xlrec->cmin != InvalidCommandId &&
     790       31600 :         xlrec->cmax != InvalidCommandId)
     791        5902 :         cid = Max(xlrec->cmin, xlrec->cmax);
     792       28920 :     else if (xlrec->cmax != InvalidCommandId)
     793        3222 :         cid = xlrec->cmax;
     794       25698 :     else if (xlrec->cmin != InvalidCommandId)
     795       25698 :         cid = xlrec->cmin;
     796             :     else
     797             :     {
     798           0 :         cid = InvalidCommandId; /* silence compiler */
     799           0 :         elog(ERROR, "xl_heap_new_cid record without a valid CommandId");
     800             :     }
     801             : 
     802       34822 :     ReorderBufferAddNewCommandId(builder->reorder, xid, lsn, cid + 1);
     803       34822 : }
     804             : 
     805             : /*
     806             :  * Add a new Snapshot to all transactions we're decoding that currently are
     807             :  * in-progress so they can see new catalog contents made by the transaction
     808             :  * that just committed. This is necessary because those in-progress
     809             :  * transactions will use the new catalog's contents from here on (at the very
     810             :  * least everything they do needs to be compatible with newer catalog
     811             :  * contents).
     812             :  */
     813             : static void
     814        1412 : SnapBuildDistributeNewCatalogSnapshot(SnapBuild *builder, XLogRecPtr lsn)
     815             : {
     816             :     dlist_iter  txn_i;
     817             :     ReorderBufferTXN *txn;
     818             : 
     819             :     /*
     820             :      * Iterate through all toplevel transactions. This can include
     821             :      * subtransactions which we just don't yet know to be that, but that's
     822             :      * fine, they will just get an unnecessary snapshot queued.
     823             :      */
     824        2874 :     dlist_foreach(txn_i, &builder->reorder->toplevel_by_lsn)
     825             :     {
     826        1462 :         txn = dlist_container(ReorderBufferTXN, node, txn_i.cur);
     827             : 
     828             :         Assert(TransactionIdIsValid(txn->xid));
     829             : 
     830             :         /*
     831             :          * If we don't have a base snapshot yet, there are no changes in this
     832             :          * transaction which in turn implies we don't yet need a snapshot at
     833             :          * all. We'll add a snapshot when the first change gets queued.
     834             :          *
     835             :          * NB: This works correctly even for subtransactions because
     836             :          * ReorderBufferAssignChild() takes care to transfer the base snapshot
     837             :          * to the top-level transaction, and while iterating the changequeue
     838             :          * we'll get the change from the subtxn.
     839             :          */
     840        1462 :         if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, txn->xid))
     841           4 :             continue;
     842             : 
     843             :         /*
     844             :          * We don't need to add snapshot to prepared transactions as they
     845             :          * should not see the new catalog contents.
     846             :          */
     847        1458 :         if (rbtxn_prepared(txn) || rbtxn_skip_prepared(txn))
     848          44 :             continue;
     849             : 
     850        1414 :         elog(DEBUG2, "adding a new snapshot to %u at %X/%X",
     851             :              txn->xid, LSN_FORMAT_ARGS(lsn));
     852             : 
     853             :         /*
     854             :          * increase the snapshot's refcount for the transaction we are handing
     855             :          * it out to
     856             :          */
     857        1414 :         SnapBuildSnapIncRefcount(builder->snapshot);
     858        1414 :         ReorderBufferAddSnapshot(builder->reorder, txn->xid, lsn,
     859             :                                  builder->snapshot);
     860             :     }
     861        1412 : }
     862             : 
     863             : /*
     864             :  * Keep track of a new catalog changing transaction that has committed.
     865             :  */
     866             : static void
     867        1422 : SnapBuildAddCommittedTxn(SnapBuild *builder, TransactionId xid)
     868             : {
     869             :     Assert(TransactionIdIsValid(xid));
     870             : 
     871        1422 :     if (builder->committed.xcnt == builder->committed.xcnt_space)
     872             :     {
     873           0 :         builder->committed.xcnt_space = builder->committed.xcnt_space * 2 + 1;
     874             : 
     875           0 :         elog(DEBUG1, "increasing space for committed transactions to %u",
     876             :              (uint32) builder->committed.xcnt_space);
     877             : 
     878           0 :         builder->committed.xip = repalloc(builder->committed.xip,
     879           0 :                                           builder->committed.xcnt_space * sizeof(TransactionId));
     880             :     }
     881             : 
     882             :     /*
     883             :      * TODO: It might make sense to keep the array sorted here instead of
     884             :      * doing it every time we build a new snapshot. On the other hand this
     885             :      * gets called repeatedly when a transaction with subtransactions commits.
     886             :      */
     887        1422 :     builder->committed.xip[builder->committed.xcnt++] = xid;
     888        1422 : }
     889             : 
     890             : /*
     891             :  * Remove knowledge about transactions we treat as committed that are smaller
     892             :  * than ->xmin. Those won't ever get checked via the ->committed array but via
     893             :  * the clog machinery, so we don't need to waste memory on them.
     894             :  */
     895             : static void
     896         306 : SnapBuildPurgeCommittedTxn(SnapBuild *builder)
     897             : {
     898             :     int         off;
     899             :     TransactionId *workspace;
     900         306 :     int         surviving_xids = 0;
     901             : 
     902             :     /* not ready yet */
     903         306 :     if (!TransactionIdIsNormal(builder->xmin))
     904           0 :         return;
     905             : 
     906             :     /* TODO: Neater algorithm than just copying and iterating? */
     907             :     workspace =
     908         306 :         MemoryContextAlloc(builder->context,
     909         306 :                            builder->committed.xcnt * sizeof(TransactionId));
     910             : 
     911             :     /* copy xids that still are interesting to workspace */
     912         412 :     for (off = 0; off < builder->committed.xcnt; off++)
     913             :     {
     914         106 :         if (NormalTransactionIdPrecedes(builder->committed.xip[off],
     915             :                                         builder->xmin))
     916             :             ;                   /* remove */
     917             :         else
     918           0 :             workspace[surviving_xids++] = builder->committed.xip[off];
     919             :     }
     920             : 
     921             :     /* copy workspace back to persistent state */
     922         306 :     memcpy(builder->committed.xip, workspace,
     923             :            surviving_xids * sizeof(TransactionId));
     924             : 
     925         306 :     elog(DEBUG3, "purged committed transactions from %u to %u, xmin: %u, xmax: %u",
     926             :          (uint32) builder->committed.xcnt, (uint32) surviving_xids,
     927             :          builder->xmin, builder->xmax);
     928         306 :     builder->committed.xcnt = surviving_xids;
     929             : 
     930         306 :     pfree(workspace);
     931             : }
     932             : 
     933             : /*
     934             :  * Handle everything that needs to be done when a transaction commits
     935             :  */
     936             : void
     937        3570 : SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
     938             :                    int nsubxacts, TransactionId *subxacts)
     939             : {
     940             :     int         nxact;
     941             : 
     942        3570 :     bool        needs_snapshot = false;
     943        3570 :     bool        needs_timetravel = false;
     944        3570 :     bool        sub_needs_timetravel = false;
     945             : 
     946        3570 :     TransactionId xmax = xid;
     947             : 
     948             :     /*
     949             :      * Transactions preceding BUILDING_SNAPSHOT will neither be decoded, nor
     950             :      * will they be part of a snapshot.  So we don't need to record anything.
     951             :      */
     952        3570 :     if (builder->state == SNAPBUILD_START ||
     953        3570 :         (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
     954           0 :          TransactionIdPrecedes(xid, builder->next_phase_at)))
     955             :     {
     956             :         /* ensure that only commits after this are getting replayed */
     957           0 :         if (builder->start_decoding_at <= lsn)
     958           0 :             builder->start_decoding_at = lsn + 1;
     959           0 :         return;
     960             :     }
     961             : 
     962        3570 :     if (builder->state < SNAPBUILD_CONSISTENT)
     963             :     {
     964             :         /* ensure that only commits after this are getting replayed */
     965          10 :         if (builder->start_decoding_at <= lsn)
     966           4 :             builder->start_decoding_at = lsn + 1;
     967             : 
     968             :         /*
     969             :          * If building an exportable snapshot, force xid to be tracked, even
     970             :          * if the transaction didn't modify the catalog.
     971             :          */
     972          10 :         if (builder->building_full_snapshot)
     973             :         {
     974           0 :             needs_timetravel = true;
     975             :         }
     976             :     }
     977             : 
     978        6030 :     for (nxact = 0; nxact < nsubxacts; nxact++)
     979             :     {
     980        2460 :         TransactionId subxid = subxacts[nxact];
     981             : 
     982             :         /*
     983             :          * Add subtransaction to base snapshot if catalog modifying, we don't
     984             :          * distinguish to toplevel transactions there.
     985             :          */
     986        2460 :         if (ReorderBufferXidHasCatalogChanges(builder->reorder, subxid))
     987             :         {
     988          10 :             sub_needs_timetravel = true;
     989          10 :             needs_snapshot = true;
     990             : 
     991          10 :             elog(DEBUG1, "found subtransaction %u:%u with catalog changes",
     992             :                  xid, subxid);
     993             : 
     994          10 :             SnapBuildAddCommittedTxn(builder, subxid);
     995             : 
     996          10 :             if (NormalTransactionIdFollows(subxid, xmax))
     997          10 :                 xmax = subxid;
     998             :         }
     999             : 
    1000             :         /*
    1001             :          * If we're forcing timetravel we also need visibility information
    1002             :          * about subtransaction, so keep track of subtransaction's state, even
    1003             :          * if not catalog modifying.  Don't need to distribute a snapshot in
    1004             :          * that case.
    1005             :          */
    1006        2450 :         else if (needs_timetravel)
    1007             :         {
    1008           0 :             SnapBuildAddCommittedTxn(builder, subxid);
    1009           0 :             if (NormalTransactionIdFollows(subxid, xmax))
    1010           0 :                 xmax = subxid;
    1011             :         }
    1012             :     }
    1013             : 
    1014             :     /* if top-level modified catalog, it'll need a snapshot */
    1015        3570 :     if (ReorderBufferXidHasCatalogChanges(builder->reorder, xid))
    1016             :     {
    1017        1412 :         elog(DEBUG2, "found top level transaction %u, with catalog changes",
    1018             :              xid);
    1019        1412 :         needs_snapshot = true;
    1020        1412 :         needs_timetravel = true;
    1021        1412 :         SnapBuildAddCommittedTxn(builder, xid);
    1022             :     }
    1023        2158 :     else if (sub_needs_timetravel)
    1024             :     {
    1025             :         /* track toplevel txn as well, subxact alone isn't meaningful */
    1026           0 :         SnapBuildAddCommittedTxn(builder, xid);
    1027             :     }
    1028        2158 :     else if (needs_timetravel)
    1029             :     {
    1030           0 :         elog(DEBUG2, "forced transaction %u to do timetravel", xid);
    1031             : 
    1032           0 :         SnapBuildAddCommittedTxn(builder, xid);
    1033             :     }
    1034             : 
    1035        3570 :     if (!needs_timetravel)
    1036             :     {
    1037             :         /* record that we cannot export a general snapshot anymore */
    1038        2158 :         builder->committed.includes_all_transactions = false;
    1039             :     }
    1040             : 
    1041             :     Assert(!needs_snapshot || needs_timetravel);
    1042             : 
    1043             :     /*
    1044             :      * Adjust xmax of the snapshot builder, we only do that for committed,
    1045             :      * catalog modifying, transactions, everything else isn't interesting for
    1046             :      * us since we'll never look at the respective rows.
    1047             :      */
    1048        3570 :     if (needs_timetravel &&
    1049        2824 :         (!TransactionIdIsValid(builder->xmax) ||
    1050        1412 :          TransactionIdFollowsOrEquals(xmax, builder->xmax)))
    1051             :     {
    1052        1410 :         builder->xmax = xmax;
    1053        1410 :         TransactionIdAdvance(builder->xmax);
    1054             :     }
    1055             : 
    1056             :     /* if there's any reason to build a historic snapshot, do so now */
    1057        3570 :     if (needs_snapshot)
    1058             :     {
    1059             :         /*
    1060             :          * If we haven't built a complete snapshot yet there's no need to hand
    1061             :          * it out, it wouldn't (and couldn't) be used anyway.
    1062             :          */
    1063        1412 :         if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
    1064           0 :             return;
    1065             : 
    1066             :         /*
    1067             :          * Decrease the snapshot builder's refcount of the old snapshot, note
    1068             :          * that it still will be used if it has been handed out to the
    1069             :          * reorderbuffer earlier.
    1070             :          */
    1071        1412 :         if (builder->snapshot)
    1072        1410 :             SnapBuildSnapDecRefcount(builder->snapshot);
    1073             : 
    1074        1412 :         builder->snapshot = SnapBuildBuildSnapshot(builder);
    1075             : 
    1076             :         /* we might need to execute invalidations, add snapshot */
    1077        1412 :         if (!ReorderBufferXidHasBaseSnapshot(builder->reorder, xid))
    1078             :         {
    1079           6 :             SnapBuildSnapIncRefcount(builder->snapshot);
    1080           6 :             ReorderBufferSetBaseSnapshot(builder->reorder, xid, lsn,
    1081             :                                          builder->snapshot);
    1082             :         }
    1083             : 
    1084             :         /* refcount of the snapshot builder for the new snapshot */
    1085        1412 :         SnapBuildSnapIncRefcount(builder->snapshot);
    1086             : 
    1087             :         /* add a new catalog snapshot to all currently running transactions */
    1088        1412 :         SnapBuildDistributeNewCatalogSnapshot(builder, lsn);
    1089             :     }
    1090             : }
    1091             : 
    1092             : 
    1093             : /* -----------------------------------
    1094             :  * Snapshot building functions dealing with xlog records
    1095             :  * -----------------------------------
    1096             :  */
    1097             : 
    1098             : /*
    1099             :  * Process a running xacts record, and use its information to first build a
    1100             :  * historic snapshot and later to release resources that aren't needed
    1101             :  * anymore.
    1102             :  */
    1103             : void
    1104        1336 : SnapBuildProcessRunningXacts(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
    1105             : {
    1106             :     ReorderBufferTXN *txn;
    1107             :     TransactionId xmin;
    1108             : 
    1109             :     /*
    1110             :      * If we're not consistent yet, inspect the record to see whether it
    1111             :      * allows to get closer to being consistent. If we are consistent, dump
    1112             :      * our snapshot so others or we, after a restart, can use it.
    1113             :      */
    1114        1336 :     if (builder->state < SNAPBUILD_CONSISTENT)
    1115             :     {
    1116             :         /* returns false if there's no point in performing cleanup just yet */
    1117        1070 :         if (!SnapBuildFindSnapshot(builder, lsn, running))
    1118        1030 :             return;
    1119             :     }
    1120             :     else
    1121         266 :         SnapBuildSerialize(builder, lsn);
    1122             : 
    1123             :     /*
    1124             :      * Update range of interesting xids based on the running xacts
    1125             :      * information. We don't increase ->xmax using it, because once we are in
    1126             :      * a consistent state we can do that ourselves and much more efficiently
    1127             :      * so, because we only need to do it for catalog transactions since we
    1128             :      * only ever look at those.
    1129             :      *
    1130             :      * NB: We only increase xmax when a catalog modifying transaction commits
    1131             :      * (see SnapBuildCommitTxn).  Because of this, xmax can be lower than
    1132             :      * xmin, which looks odd but is correct and actually more efficient, since
    1133             :      * we hit fast paths in heapam_visibility.c.
    1134             :      */
    1135         306 :     builder->xmin = running->oldestRunningXid;
    1136             : 
    1137             :     /* Remove transactions we don't need to keep track off anymore */
    1138         306 :     SnapBuildPurgeCommittedTxn(builder);
    1139             : 
    1140             :     /*
    1141             :      * Advance the xmin limit for the current replication slot, to allow
    1142             :      * vacuum to clean up the tuples this slot has been protecting.
    1143             :      *
    1144             :      * The reorderbuffer might have an xmin among the currently running
    1145             :      * snapshots; use it if so.  If not, we need only consider the snapshots
    1146             :      * we'll produce later, which can't be less than the oldest running xid in
    1147             :      * the record we're reading now.
    1148             :      */
    1149         306 :     xmin = ReorderBufferGetOldestXmin(builder->reorder);
    1150         306 :     if (xmin == InvalidTransactionId)
    1151         270 :         xmin = running->oldestRunningXid;
    1152         306 :     elog(DEBUG3, "xmin: %u, xmax: %u, oldest running: %u, oldest xmin: %u",
    1153             :          builder->xmin, builder->xmax, running->oldestRunningXid, xmin);
    1154         306 :     LogicalIncreaseXminForSlot(lsn, xmin);
    1155             : 
    1156             :     /*
    1157             :      * Also tell the slot where we can restart decoding from. We don't want to
    1158             :      * do that after every commit because changing that implies an fsync of
    1159             :      * the logical slot's state file, so we only do it every time we see a
    1160             :      * running xacts record.
    1161             :      *
    1162             :      * Do so by looking for the oldest in progress transaction (determined by
    1163             :      * the first LSN of any of its relevant records). Every transaction
    1164             :      * remembers the last location we stored the snapshot to disk before its
    1165             :      * beginning. That point is where we can restart from.
    1166             :      */
    1167             : 
    1168             :     /*
    1169             :      * Can't know about a serialized snapshot's location if we're not
    1170             :      * consistent.
    1171             :      */
    1172         306 :     if (builder->state < SNAPBUILD_CONSISTENT)
    1173          30 :         return;
    1174             : 
    1175         276 :     txn = ReorderBufferGetOldestTXN(builder->reorder);
    1176             : 
    1177             :     /*
    1178             :      * oldest ongoing txn might have started when we didn't yet serialize
    1179             :      * anything because we hadn't reached a consistent state yet.
    1180             :      */
    1181         276 :     if (txn != NULL && txn->restart_decoding_lsn != InvalidXLogRecPtr)
    1182          10 :         LogicalIncreaseRestartDecodingForSlot(lsn, txn->restart_decoding_lsn);
    1183             : 
    1184             :     /*
    1185             :      * No in-progress transaction, can reuse the last serialized snapshot if
    1186             :      * we have one.
    1187             :      */
    1188         266 :     else if (txn == NULL &&
    1189         240 :              builder->reorder->current_restart_decoding_lsn != InvalidXLogRecPtr &&
    1190         236 :              builder->last_serialized_snapshot != InvalidXLogRecPtr)
    1191         236 :         LogicalIncreaseRestartDecodingForSlot(lsn,
    1192             :                                               builder->last_serialized_snapshot);
    1193             : }
    1194             : 
    1195             : 
    1196             : /*
    1197             :  * Build the start of a snapshot that's capable of decoding the catalog.
    1198             :  *
    1199             :  * Helper function for SnapBuildProcessRunningXacts() while we're not yet
    1200             :  * consistent.
    1201             :  *
    1202             :  * Returns true if there is a point in performing internal maintenance/cleanup
    1203             :  * using the xl_running_xacts record.
    1204             :  */
    1205             : static bool
    1206        1070 : SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *running)
    1207             : {
    1208             :     /* ---
    1209             :      * Build catalog decoding snapshot incrementally using information about
    1210             :      * the currently running transactions. There are several ways to do that:
    1211             :      *
    1212             :      * a) There were no running transactions when the xl_running_xacts record
    1213             :      *    was inserted, jump to CONSISTENT immediately. We might find such a
    1214             :      *    state while waiting on c)'s sub-states.
    1215             :      *
    1216             :      * b) This (in a previous run) or another decoding slot serialized a
    1217             :      *    snapshot to disk that we can use.  Can't use this method for the
    1218             :      *    initial snapshot when slot is being created and needs full snapshot
    1219             :      *    for export or direct use, as that snapshot will only contain catalog
    1220             :      *    modifying transactions.
    1221             :      *
    1222             :      * c) First incrementally build a snapshot for catalog tuples
    1223             :      *    (BUILDING_SNAPSHOT), that requires all, already in-progress,
    1224             :      *    transactions to finish.  Every transaction starting after that
    1225             :      *    (FULL_SNAPSHOT state), has enough information to be decoded.  But
    1226             :      *    for older running transactions no viable snapshot exists yet, so
    1227             :      *    CONSISTENT will only be reached once all of those have finished.
    1228             :      * ---
    1229             :      */
    1230             : 
    1231             :     /*
    1232             :      * xl_running_xact record is older than what we can use, we might not have
    1233             :      * all necessary catalog rows anymore.
    1234             :      */
    1235        1070 :     if (TransactionIdIsNormal(builder->initial_xmin_horizon) &&
    1236         432 :         NormalTransactionIdPrecedes(running->oldestRunningXid,
    1237             :                                     builder->initial_xmin_horizon))
    1238             :     {
    1239           0 :         ereport(DEBUG1,
    1240             :                 (errmsg_internal("skipping snapshot at %X/%X while building logical decoding snapshot, xmin horizon too low",
    1241             :                                  LSN_FORMAT_ARGS(lsn)),
    1242             :                  errdetail_internal("initial xmin horizon of %u vs the snapshot's %u",
    1243             :                                     builder->initial_xmin_horizon, running->oldestRunningXid)));
    1244             : 
    1245             : 
    1246           0 :         SnapBuildWaitSnapshot(running, builder->initial_xmin_horizon);
    1247             : 
    1248           0 :         return true;
    1249             :     }
    1250             : 
    1251             :     /*
    1252             :      * a) No transaction were running, we can jump to consistent.
    1253             :      *
    1254             :      * This is not affected by races around xl_running_xacts, because we can
    1255             :      * miss transaction commits, but currently not transactions starting.
    1256             :      *
    1257             :      * NB: We might have already started to incrementally assemble a snapshot,
    1258             :      * so we need to be careful to deal with that.
    1259             :      */
    1260        1070 :     if (running->oldestRunningXid == running->nextXid)
    1261             :     {
    1262        1026 :         if (builder->start_decoding_at == InvalidXLogRecPtr ||
    1263         610 :             builder->start_decoding_at <= lsn)
    1264             :             /* can decode everything after this */
    1265         416 :             builder->start_decoding_at = lsn + 1;
    1266             : 
    1267             :         /* As no transactions were running xmin/xmax can be trivially set. */
    1268        1026 :         builder->xmin = running->nextXid; /* < are finished */
    1269        1026 :         builder->xmax = running->nextXid; /* >= are running */
    1270             : 
    1271             :         /* so we can safely use the faster comparisons */
    1272             :         Assert(TransactionIdIsNormal(builder->xmin));
    1273             :         Assert(TransactionIdIsNormal(builder->xmax));
    1274             : 
    1275        1026 :         builder->state = SNAPBUILD_CONSISTENT;
    1276        1026 :         builder->next_phase_at = InvalidTransactionId;
    1277             : 
    1278        1026 :         ereport(LOG,
    1279             :                 (errmsg("logical decoding found consistent point at %X/%X",
    1280             :                         LSN_FORMAT_ARGS(lsn)),
    1281             :                  errdetail("There are no running transactions.")));
    1282             : 
    1283        1026 :         return false;
    1284             :     }
    1285             :     /* b) valid on disk state and not building full snapshot */
    1286          88 :     else if (!builder->building_full_snapshot &&
    1287          44 :              SnapBuildRestore(builder, lsn))
    1288             :     {
    1289             :         /* there won't be any state to cleanup */
    1290           4 :         return false;
    1291             :     }
    1292             : 
    1293             :     /*
    1294             :      * c) transition from START to BUILDING_SNAPSHOT.
    1295             :      *
    1296             :      * In START state, and a xl_running_xacts record with running xacts is
    1297             :      * encountered.  In that case, switch to BUILDING_SNAPSHOT state, and
    1298             :      * record xl_running_xacts->nextXid.  Once all running xacts have finished
    1299             :      * (i.e. they're all >= nextXid), we have a complete catalog snapshot.  It
    1300             :      * might look that we could use xl_running_xact's ->xids information to
    1301             :      * get there quicker, but that is problematic because transactions marked
    1302             :      * as running, might already have inserted their commit record - it's
    1303             :      * infeasible to change that with locking.
    1304             :      */
    1305          40 :     else if (builder->state == SNAPBUILD_START)
    1306             :     {
    1307          20 :         builder->state = SNAPBUILD_BUILDING_SNAPSHOT;
    1308          20 :         builder->next_phase_at = running->nextXid;
    1309             : 
    1310             :         /*
    1311             :          * Start with an xmin/xmax that's correct for future, when all the
    1312             :          * currently running transactions have finished. We'll update both
    1313             :          * while waiting for the pending transactions to finish.
    1314             :          */
    1315          20 :         builder->xmin = running->nextXid; /* < are finished */
    1316          20 :         builder->xmax = running->nextXid; /* >= are running */
    1317             : 
    1318             :         /* so we can safely use the faster comparisons */
    1319             :         Assert(TransactionIdIsNormal(builder->xmin));
    1320             :         Assert(TransactionIdIsNormal(builder->xmax));
    1321             : 
    1322          20 :         ereport(LOG,
    1323             :                 (errmsg("logical decoding found initial starting point at %X/%X",
    1324             :                         LSN_FORMAT_ARGS(lsn)),
    1325             :                  errdetail("Waiting for transactions (approximately %d) older than %u to end.",
    1326             :                            running->xcnt, running->nextXid)));
    1327             : 
    1328          20 :         SnapBuildWaitSnapshot(running, running->nextXid);
    1329             :     }
    1330             : 
    1331             :     /*
    1332             :      * c) transition from BUILDING_SNAPSHOT to FULL_SNAPSHOT.
    1333             :      *
    1334             :      * In BUILDING_SNAPSHOT state, and this xl_running_xacts' oldestRunningXid
    1335             :      * is >= than nextXid from when we switched to BUILDING_SNAPSHOT.  This
    1336             :      * means all transactions starting afterwards have enough information to
    1337             :      * be decoded.  Switch to FULL_SNAPSHOT.
    1338             :      */
    1339          30 :     else if (builder->state == SNAPBUILD_BUILDING_SNAPSHOT &&
    1340          10 :              TransactionIdPrecedesOrEquals(builder->next_phase_at,
    1341             :                                            running->oldestRunningXid))
    1342             :     {
    1343          10 :         builder->state = SNAPBUILD_FULL_SNAPSHOT;
    1344          10 :         builder->next_phase_at = running->nextXid;
    1345             : 
    1346          10 :         ereport(LOG,
    1347             :                 (errmsg("logical decoding found initial consistent point at %X/%X",
    1348             :                         LSN_FORMAT_ARGS(lsn)),
    1349             :                  errdetail("Waiting for transactions (approximately %d) older than %u to end.",
    1350             :                            running->xcnt, running->nextXid)));
    1351             : 
    1352          10 :         SnapBuildWaitSnapshot(running, running->nextXid);
    1353             :     }
    1354             : 
    1355             :     /*
    1356             :      * c) transition from FULL_SNAPSHOT to CONSISTENT.
    1357             :      *
    1358             :      * In FULL_SNAPSHOT state (see d) ), and this xl_running_xacts'
    1359             :      * oldestRunningXid is >= than nextXid from when we switched to
    1360             :      * FULL_SNAPSHOT.  This means all transactions that are currently in
    1361             :      * progress have a catalog snapshot, and all their changes have been
    1362             :      * collected.  Switch to CONSISTENT.
    1363             :      */
    1364          20 :     else if (builder->state == SNAPBUILD_FULL_SNAPSHOT &&
    1365          10 :              TransactionIdPrecedesOrEquals(builder->next_phase_at,
    1366             :                                            running->oldestRunningXid))
    1367             :     {
    1368          10 :         builder->state = SNAPBUILD_CONSISTENT;
    1369          10 :         builder->next_phase_at = InvalidTransactionId;
    1370             : 
    1371          10 :         ereport(LOG,
    1372             :                 (errmsg("logical decoding found consistent point at %X/%X",
    1373             :                         LSN_FORMAT_ARGS(lsn)),
    1374             :                  errdetail("There are no old transactions anymore.")));
    1375             :     }
    1376             : 
    1377             :     /*
    1378             :      * We already started to track running xacts and need to wait for all
    1379             :      * in-progress ones to finish. We fall through to the normal processing of
    1380             :      * records so incremental cleanup can be performed.
    1381             :      */
    1382          40 :     return true;
    1383             : 
    1384             : }
    1385             : 
    1386             : /* ---
    1387             :  * Iterate through xids in record, wait for all older than the cutoff to
    1388             :  * finish.  Then, if possible, log a new xl_running_xacts record.
    1389             :  *
    1390             :  * This isn't required for the correctness of decoding, but to:
    1391             :  * a) allow isolationtester to notice that we're currently waiting for
    1392             :  *    something.
    1393             :  * b) log a new xl_running_xacts record where it'd be helpful, without having
    1394             :  *    to wait for bgwriter or checkpointer.
    1395             :  * ---
    1396             :  */
    1397             : static void
    1398          30 : SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutoff)
    1399             : {
    1400             :     int         off;
    1401             : 
    1402          60 :     for (off = 0; off < running->xcnt; off++)
    1403             :     {
    1404          30 :         TransactionId xid = running->xids[off];
    1405             : 
    1406             :         /*
    1407             :          * Upper layers should prevent that we ever need to wait on ourselves.
    1408             :          * Check anyway, since failing to do so would either result in an
    1409             :          * endless wait or an Assert() failure.
    1410             :          */
    1411          30 :         if (TransactionIdIsCurrentTransactionId(xid))
    1412           0 :             elog(ERROR, "waiting for ourselves");
    1413             : 
    1414          30 :         if (TransactionIdFollows(xid, cutoff))
    1415           0 :             continue;
    1416             : 
    1417          30 :         XactLockTableWait(xid, NULL, NULL, XLTW_None);
    1418             :     }
    1419             : 
    1420             :     /*
    1421             :      * All transactions we needed to finish finished - try to ensure there is
    1422             :      * another xl_running_xacts record in a timely manner, without having to
    1423             :      * wait for bgwriter or checkpointer to log one.  During recovery we can't
    1424             :      * enforce that, so we'll have to wait.
    1425             :      */
    1426          30 :     if (!RecoveryInProgress())
    1427             :     {
    1428          30 :         LogStandbySnapshot();
    1429             :     }
    1430          30 : }
    1431             : 
    1432             : /* -----------------------------------
    1433             :  * Snapshot serialization support
    1434             :  * -----------------------------------
    1435             :  */
    1436             : 
    1437             : /*
    1438             :  * We store current state of struct SnapBuild on disk in the following manner:
    1439             :  *
    1440             :  * struct SnapBuildOnDisk;
    1441             :  * TransactionId * committed.xcnt; (*not xcnt_space*)
    1442             :  *
    1443             :  */
    1444             : typedef struct SnapBuildOnDisk
    1445             : {
    1446             :     /* first part of this struct needs to be version independent */
    1447             : 
    1448             :     /* data not covered by checksum */
    1449             :     uint32      magic;
    1450             :     pg_crc32c   checksum;
    1451             : 
    1452             :     /* data covered by checksum */
    1453             : 
    1454             :     /* version, in case we want to support pg_upgrade */
    1455             :     uint32      version;
    1456             :     /* how large is the on disk data, excluding the constant sized part */
    1457             :     uint32      length;
    1458             : 
    1459             :     /* version dependent part */
    1460             :     SnapBuild   builder;
    1461             : 
    1462             :     /* variable amount of TransactionIds follows */
    1463             : } SnapBuildOnDisk;
    1464             : 
    1465             : #define SnapBuildOnDiskConstantSize \
    1466             :     offsetof(SnapBuildOnDisk, builder)
    1467             : #define SnapBuildOnDiskNotChecksummedSize \
    1468             :     offsetof(SnapBuildOnDisk, version)
    1469             : 
    1470             : #define SNAPBUILD_MAGIC 0x51A1E001
    1471             : #define SNAPBUILD_VERSION 4
    1472             : 
    1473             : /*
    1474             :  * Store/Load a snapshot from disk, depending on the snapshot builder's state.
    1475             :  *
    1476             :  * Supposed to be used by external (i.e. not snapbuild.c) code that just read
    1477             :  * a record that's a potential location for a serialized snapshot.
    1478             :  */
    1479             : void
    1480          30 : SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
    1481             : {
    1482          30 :     if (builder->state < SNAPBUILD_CONSISTENT)
    1483           0 :         SnapBuildRestore(builder, lsn);
    1484             :     else
    1485          30 :         SnapBuildSerialize(builder, lsn);
    1486          30 : }
    1487             : 
    1488             : /*
    1489             :  * Serialize the snapshot 'builder' at the location 'lsn' if it hasn't already
    1490             :  * been done by another decoding process.
    1491             :  */
    1492             : static void
    1493         296 : SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
    1494             : {
    1495             :     Size        needed_length;
    1496         296 :     SnapBuildOnDisk *ondisk = NULL;
    1497             :     char       *ondisk_c;
    1498             :     int         fd;
    1499             :     char        tmppath[MAXPGPATH];
    1500             :     char        path[MAXPGPATH];
    1501             :     int         ret;
    1502             :     struct stat stat_buf;
    1503             :     Size        sz;
    1504             : 
    1505             :     Assert(lsn != InvalidXLogRecPtr);
    1506             :     Assert(builder->last_serialized_snapshot == InvalidXLogRecPtr ||
    1507             :            builder->last_serialized_snapshot <= lsn);
    1508             : 
    1509             :     /*
    1510             :      * no point in serializing if we cannot continue to work immediately after
    1511             :      * restoring the snapshot
    1512             :      */
    1513         296 :     if (builder->state < SNAPBUILD_CONSISTENT)
    1514           0 :         return;
    1515             : 
    1516             :     /* consistent snapshots have no next phase */
    1517             :     Assert(builder->next_phase_at == InvalidTransactionId);
    1518             : 
    1519             :     /*
    1520             :      * We identify snapshots by the LSN they are valid for. We don't need to
    1521             :      * include timelines in the name as each LSN maps to exactly one timeline
    1522             :      * unless the user used pg_resetwal or similar. If a user did so, there's
    1523             :      * no hope continuing to decode anyway.
    1524             :      */
    1525         296 :     sprintf(path, "pg_logical/snapshots/%X-%X.snap",
    1526         296 :             LSN_FORMAT_ARGS(lsn));
    1527             : 
    1528             :     /*
    1529             :      * first check whether some other backend already has written the snapshot
    1530             :      * for this LSN. It's perfectly fine if there's none, so we accept ENOENT
    1531             :      * as a valid state. Everything else is an unexpected error.
    1532             :      */
    1533         296 :     ret = stat(path, &stat_buf);
    1534             : 
    1535         296 :     if (ret != 0 && errno != ENOENT)
    1536           0 :         ereport(ERROR,
    1537             :                 (errcode_for_file_access(),
    1538             :                  errmsg("could not stat file \"%s\": %m", path)));
    1539             : 
    1540         296 :     else if (ret == 0)
    1541             :     {
    1542             :         /*
    1543             :          * somebody else has already serialized to this point, don't overwrite
    1544             :          * but remember location, so we don't need to read old data again.
    1545             :          *
    1546             :          * To be sure it has been synced to disk after the rename() from the
    1547             :          * tempfile filename to the real filename, we just repeat the fsync.
    1548             :          * That ought to be cheap because in most scenarios it should already
    1549             :          * be safely on disk.
    1550             :          */
    1551          62 :         fsync_fname(path, false);
    1552          62 :         fsync_fname("pg_logical/snapshots", true);
    1553             : 
    1554          62 :         builder->last_serialized_snapshot = lsn;
    1555          62 :         goto out;
    1556             :     }
    1557             : 
    1558             :     /*
    1559             :      * there is an obvious race condition here between the time we stat(2) the
    1560             :      * file and us writing the file. But we rename the file into place
    1561             :      * atomically and all files created need to contain the same data anyway,
    1562             :      * so this is perfectly fine, although a bit of a resource waste. Locking
    1563             :      * seems like pointless complication.
    1564             :      */
    1565         234 :     elog(DEBUG1, "serializing snapshot to %s", path);
    1566             : 
    1567             :     /* to make sure only we will write to this tempfile, include pid */
    1568         234 :     sprintf(tmppath, "pg_logical/snapshots/%X-%X.snap.%d.tmp",
    1569         234 :             LSN_FORMAT_ARGS(lsn), MyProcPid);
    1570             : 
    1571             :     /*
    1572             :      * Unlink temporary file if it already exists, needs to have been before a
    1573             :      * crash/error since we won't enter this function twice from within a
    1574             :      * single decoding slot/backend and the temporary file contains the pid of
    1575             :      * the current process.
    1576             :      */
    1577         234 :     if (unlink(tmppath) != 0 && errno != ENOENT)
    1578           0 :         ereport(ERROR,
    1579             :                 (errcode_for_file_access(),
    1580             :                  errmsg("could not remove file \"%s\": %m", tmppath)));
    1581             : 
    1582         234 :     needed_length = sizeof(SnapBuildOnDisk) +
    1583         234 :         sizeof(TransactionId) * builder->committed.xcnt;
    1584             : 
    1585         234 :     ondisk_c = MemoryContextAllocZero(builder->context, needed_length);
    1586         234 :     ondisk = (SnapBuildOnDisk *) ondisk_c;
    1587         234 :     ondisk->magic = SNAPBUILD_MAGIC;
    1588         234 :     ondisk->version = SNAPBUILD_VERSION;
    1589         234 :     ondisk->length = needed_length;
    1590         234 :     INIT_CRC32C(ondisk->checksum);
    1591         234 :     COMP_CRC32C(ondisk->checksum,
    1592             :                 ((char *) ondisk) + SnapBuildOnDiskNotChecksummedSize,
    1593             :                 SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);
    1594         234 :     ondisk_c += sizeof(SnapBuildOnDisk);
    1595             : 
    1596         234 :     memcpy(&ondisk->builder, builder, sizeof(SnapBuild));
    1597             :     /* NULL-ify memory-only data */
    1598         234 :     ondisk->builder.context = NULL;
    1599         234 :     ondisk->builder.snapshot = NULL;
    1600         234 :     ondisk->builder.reorder = NULL;
    1601         234 :     ondisk->builder.committed.xip = NULL;
    1602             : 
    1603         234 :     COMP_CRC32C(ondisk->checksum,
    1604             :                 &ondisk->builder,
    1605             :                 sizeof(SnapBuild));
    1606             : 
    1607             :     /* copy committed xacts */
    1608         234 :     sz = sizeof(TransactionId) * builder->committed.xcnt;
    1609         234 :     memcpy(ondisk_c, builder->committed.xip, sz);
    1610         234 :     COMP_CRC32C(ondisk->checksum, ondisk_c, sz);
    1611         234 :     ondisk_c += sz;
    1612             : 
    1613         234 :     FIN_CRC32C(ondisk->checksum);
    1614             : 
    1615             :     /* we have valid data now, open tempfile and write it there */
    1616         234 :     fd = OpenTransientFile(tmppath,
    1617             :                            O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
    1618         234 :     if (fd < 0)
    1619           0 :         ereport(ERROR,
    1620             :                 (errcode_for_file_access(),
    1621             :                  errmsg("could not open file \"%s\": %m", tmppath)));
    1622             : 
    1623         234 :     errno = 0;
    1624         234 :     pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_WRITE);
    1625         234 :     if ((write(fd, ondisk, needed_length)) != needed_length)
    1626             :     {
    1627           0 :         int         save_errno = errno;
    1628             : 
    1629           0 :         CloseTransientFile(fd);
    1630             : 
    1631             :         /* if write didn't set errno, assume problem is no disk space */
    1632           0 :         errno = save_errno ? save_errno : ENOSPC;
    1633           0 :         ereport(ERROR,
    1634             :                 (errcode_for_file_access(),
    1635             :                  errmsg("could not write to file \"%s\": %m", tmppath)));
    1636             :     }
    1637         234 :     pgstat_report_wait_end();
    1638             : 
    1639             :     /*
    1640             :      * fsync the file before renaming so that even if we crash after this we
    1641             :      * have either a fully valid file or nothing.
    1642             :      *
    1643             :      * It's safe to just ERROR on fsync() here because we'll retry the whole
    1644             :      * operation including the writes.
    1645             :      *
    1646             :      * TODO: Do the fsync() via checkpoints/restartpoints, doing it here has
    1647             :      * some noticeable overhead since it's performed synchronously during
    1648             :      * decoding?
    1649             :      */
    1650         234 :     pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_SYNC);
    1651         234 :     if (pg_fsync(fd) != 0)
    1652             :     {
    1653           0 :         int         save_errno = errno;
    1654             : 
    1655           0 :         CloseTransientFile(fd);
    1656           0 :         errno = save_errno;
    1657           0 :         ereport(ERROR,
    1658             :                 (errcode_for_file_access(),
    1659             :                  errmsg("could not fsync file \"%s\": %m", tmppath)));
    1660             :     }
    1661         234 :     pgstat_report_wait_end();
    1662             : 
    1663         234 :     if (CloseTransientFile(fd) != 0)
    1664           0 :         ereport(ERROR,
    1665             :                 (errcode_for_file_access(),
    1666             :                  errmsg("could not close file \"%s\": %m", tmppath)));
    1667             : 
    1668         234 :     fsync_fname("pg_logical/snapshots", true);
    1669             : 
    1670             :     /*
    1671             :      * We may overwrite the work from some other backend, but that's ok, our
    1672             :      * snapshot is valid as well, we'll just have done some superfluous work.
    1673             :      */
    1674         234 :     if (rename(tmppath, path) != 0)
    1675             :     {
    1676           0 :         ereport(ERROR,
    1677             :                 (errcode_for_file_access(),
    1678             :                  errmsg("could not rename file \"%s\" to \"%s\": %m",
    1679             :                         tmppath, path)));
    1680             :     }
    1681             : 
    1682             :     /* make sure we persist */
    1683         234 :     fsync_fname(path, false);
    1684         234 :     fsync_fname("pg_logical/snapshots", true);
    1685             : 
    1686             :     /*
    1687             :      * Now there's no way we can loose the dumped state anymore, remember this
    1688             :      * as a serialization point.
    1689             :      */
    1690         234 :     builder->last_serialized_snapshot = lsn;
    1691             : 
    1692         296 : out:
    1693         296 :     ReorderBufferSetRestartPoint(builder->reorder,
    1694             :                                  builder->last_serialized_snapshot);
    1695             :     /* be tidy */
    1696         296 :     if (ondisk)
    1697         234 :         pfree(ondisk);
    1698             : }
    1699             : 
    1700             : /*
    1701             :  * Restore a snapshot into 'builder' if previously one has been stored at the
    1702             :  * location indicated by 'lsn'. Returns true if successful, false otherwise.
    1703             :  */
    1704             : static bool
    1705          44 : SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
    1706             : {
    1707             :     SnapBuildOnDisk ondisk;
    1708             :     int         fd;
    1709             :     char        path[MAXPGPATH];
    1710             :     Size        sz;
    1711             :     int         readBytes;
    1712             :     pg_crc32c   checksum;
    1713             : 
    1714             :     /* no point in loading a snapshot if we're already there */
    1715          44 :     if (builder->state == SNAPBUILD_CONSISTENT)
    1716           0 :         return false;
    1717             : 
    1718          44 :     sprintf(path, "pg_logical/snapshots/%X-%X.snap",
    1719          44 :             LSN_FORMAT_ARGS(lsn));
    1720             : 
    1721          44 :     fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
    1722             : 
    1723          44 :     if (fd < 0 && errno == ENOENT)
    1724          40 :         return false;
    1725           4 :     else if (fd < 0)
    1726           0 :         ereport(ERROR,
    1727             :                 (errcode_for_file_access(),
    1728             :                  errmsg("could not open file \"%s\": %m", path)));
    1729             : 
    1730             :     /* ----
    1731             :      * Make sure the snapshot had been stored safely to disk, that's normally
    1732             :      * cheap.
    1733             :      * Note that we do not need PANIC here, nobody will be able to use the
    1734             :      * slot without fsyncing, and saving it won't succeed without an fsync()
    1735             :      * either...
    1736             :      * ----
    1737             :      */
    1738           4 :     fsync_fname(path, false);
    1739           4 :     fsync_fname("pg_logical/snapshots", true);
    1740             : 
    1741             : 
    1742             :     /* read statically sized portion of snapshot */
    1743           4 :     pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_READ);
    1744           4 :     readBytes = read(fd, &ondisk, SnapBuildOnDiskConstantSize);
    1745           4 :     pgstat_report_wait_end();
    1746           4 :     if (readBytes != SnapBuildOnDiskConstantSize)
    1747             :     {
    1748           0 :         int         save_errno = errno;
    1749             : 
    1750           0 :         CloseTransientFile(fd);
    1751             : 
    1752           0 :         if (readBytes < 0)
    1753             :         {
    1754           0 :             errno = save_errno;
    1755           0 :             ereport(ERROR,
    1756             :                     (errcode_for_file_access(),
    1757             :                      errmsg("could not read file \"%s\": %m", path)));
    1758             :         }
    1759             :         else
    1760           0 :             ereport(ERROR,
    1761             :                     (errcode(ERRCODE_DATA_CORRUPTED),
    1762             :                      errmsg("could not read file \"%s\": read %d of %zu",
    1763             :                             path, readBytes,
    1764             :                             (Size) SnapBuildOnDiskConstantSize)));
    1765             :     }
    1766             : 
    1767           4 :     if (ondisk.magic != SNAPBUILD_MAGIC)
    1768           0 :         ereport(ERROR,
    1769             :                 (errcode(ERRCODE_DATA_CORRUPTED),
    1770             :                  errmsg("snapbuild state file \"%s\" has wrong magic number: %u instead of %u",
    1771             :                         path, ondisk.magic, SNAPBUILD_MAGIC)));
    1772             : 
    1773           4 :     if (ondisk.version != SNAPBUILD_VERSION)
    1774           0 :         ereport(ERROR,
    1775             :                 (errcode(ERRCODE_DATA_CORRUPTED),
    1776             :                  errmsg("snapbuild state file \"%s\" has unsupported version: %u instead of %u",
    1777             :                         path, ondisk.version, SNAPBUILD_VERSION)));
    1778             : 
    1779           4 :     INIT_CRC32C(checksum);
    1780           4 :     COMP_CRC32C(checksum,
    1781             :                 ((char *) &ondisk) + SnapBuildOnDiskNotChecksummedSize,
    1782             :                 SnapBuildOnDiskConstantSize - SnapBuildOnDiskNotChecksummedSize);
    1783             : 
    1784             :     /* read SnapBuild */
    1785           4 :     pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_READ);
    1786           4 :     readBytes = read(fd, &ondisk.builder, sizeof(SnapBuild));
    1787           4 :     pgstat_report_wait_end();
    1788           4 :     if (readBytes != sizeof(SnapBuild))
    1789             :     {
    1790           0 :         int         save_errno = errno;
    1791             : 
    1792           0 :         CloseTransientFile(fd);
    1793             : 
    1794           0 :         if (readBytes < 0)
    1795             :         {
    1796           0 :             errno = save_errno;
    1797           0 :             ereport(ERROR,
    1798             :                     (errcode_for_file_access(),
    1799             :                      errmsg("could not read file \"%s\": %m", path)));
    1800             :         }
    1801             :         else
    1802           0 :             ereport(ERROR,
    1803             :                     (errcode(ERRCODE_DATA_CORRUPTED),
    1804             :                      errmsg("could not read file \"%s\": read %d of %zu",
    1805             :                             path, readBytes, sizeof(SnapBuild))));
    1806             :     }
    1807           4 :     COMP_CRC32C(checksum, &ondisk.builder, sizeof(SnapBuild));
    1808             : 
    1809             :     /* restore committed xacts information */
    1810           4 :     sz = sizeof(TransactionId) * ondisk.builder.committed.xcnt;
    1811           4 :     ondisk.builder.committed.xip = MemoryContextAllocZero(builder->context, sz);
    1812           4 :     pgstat_report_wait_start(WAIT_EVENT_SNAPBUILD_READ);
    1813           4 :     readBytes = read(fd, ondisk.builder.committed.xip, sz);
    1814           4 :     pgstat_report_wait_end();
    1815           4 :     if (readBytes != sz)
    1816             :     {
    1817           0 :         int         save_errno = errno;
    1818             : 
    1819           0 :         CloseTransientFile(fd);
    1820             : 
    1821           0 :         if (readBytes < 0)
    1822             :         {
    1823           0 :             errno = save_errno;
    1824           0 :             ereport(ERROR,
    1825             :                     (errcode_for_file_access(),
    1826             :                      errmsg("could not read file \"%s\": %m", path)));
    1827             :         }
    1828             :         else
    1829           0 :             ereport(ERROR,
    1830             :                     (errcode(ERRCODE_DATA_CORRUPTED),
    1831             :                      errmsg("could not read file \"%s\": read %d of %zu",
    1832             :                             path, readBytes, sz)));
    1833             :     }
    1834           4 :     COMP_CRC32C(checksum, ondisk.builder.committed.xip, sz);
    1835             : 
    1836           4 :     if (CloseTransientFile(fd) != 0)
    1837           0 :         ereport(ERROR,
    1838             :                 (errcode_for_file_access(),
    1839             :                  errmsg("could not close file \"%s\": %m", path)));
    1840             : 
    1841           4 :     FIN_CRC32C(checksum);
    1842             : 
    1843             :     /* verify checksum of what we've read */
    1844           4 :     if (!EQ_CRC32C(checksum, ondisk.checksum))
    1845           0 :         ereport(ERROR,
    1846             :                 (errcode(ERRCODE_DATA_CORRUPTED),
    1847             :                  errmsg("checksum mismatch for snapbuild state file \"%s\": is %u, should be %u",
    1848             :                         path, checksum, ondisk.checksum)));
    1849             : 
    1850             :     /*
    1851             :      * ok, we now have a sensible snapshot here, figure out if it has more
    1852             :      * information than we have.
    1853             :      */
    1854             : 
    1855             :     /*
    1856             :      * We are only interested in consistent snapshots for now, comparing
    1857             :      * whether one incomplete snapshot is more "advanced" seems to be
    1858             :      * unnecessarily complex.
    1859             :      */
    1860           4 :     if (ondisk.builder.state < SNAPBUILD_CONSISTENT)
    1861           0 :         goto snapshot_not_interesting;
    1862             : 
    1863             :     /*
    1864             :      * Don't use a snapshot that requires an xmin that we cannot guarantee to
    1865             :      * be available.
    1866             :      */
    1867           4 :     if (TransactionIdPrecedes(ondisk.builder.xmin, builder->initial_xmin_horizon))
    1868           0 :         goto snapshot_not_interesting;
    1869             : 
    1870             :     /* consistent snapshots have no next phase */
    1871             :     Assert(ondisk.builder.next_phase_at == InvalidTransactionId);
    1872             : 
    1873             :     /* ok, we think the snapshot is sensible, copy over everything important */
    1874           4 :     builder->xmin = ondisk.builder.xmin;
    1875           4 :     builder->xmax = ondisk.builder.xmax;
    1876           4 :     builder->state = ondisk.builder.state;
    1877             : 
    1878           4 :     builder->committed.xcnt = ondisk.builder.committed.xcnt;
    1879             :     /* We only allocated/stored xcnt, not xcnt_space xids ! */
    1880             :     /* don't overwrite preallocated xip, if we don't have anything here */
    1881           4 :     if (builder->committed.xcnt > 0)
    1882             :     {
    1883           4 :         pfree(builder->committed.xip);
    1884           4 :         builder->committed.xcnt_space = ondisk.builder.committed.xcnt;
    1885           4 :         builder->committed.xip = ondisk.builder.committed.xip;
    1886             :     }
    1887           4 :     ondisk.builder.committed.xip = NULL;
    1888             : 
    1889             :     /* our snapshot is not interesting anymore, build a new one */
    1890           4 :     if (builder->snapshot != NULL)
    1891             :     {
    1892           0 :         SnapBuildSnapDecRefcount(builder->snapshot);
    1893             :     }
    1894           4 :     builder->snapshot = SnapBuildBuildSnapshot(builder);
    1895           4 :     SnapBuildSnapIncRefcount(builder->snapshot);
    1896             : 
    1897           4 :     ReorderBufferSetRestartPoint(builder->reorder, lsn);
    1898             : 
    1899             :     Assert(builder->state == SNAPBUILD_CONSISTENT);
    1900             : 
    1901           4 :     ereport(LOG,
    1902             :             (errmsg("logical decoding found consistent point at %X/%X",
    1903             :                     LSN_FORMAT_ARGS(lsn)),
    1904             :              errdetail("Logical decoding will begin using saved snapshot.")));
    1905           4 :     return true;
    1906             : 
    1907           0 : snapshot_not_interesting:
    1908           0 :     if (ondisk.builder.committed.xip != NULL)
    1909           0 :         pfree(ondisk.builder.committed.xip);
    1910           0 :     return false;
    1911             : }
    1912             : 
    1913             : /*
    1914             :  * Remove all serialized snapshots that are not required anymore because no
    1915             :  * slot can need them. This doesn't actually have to run during a checkpoint,
    1916             :  * but it's a convenient point to schedule this.
    1917             :  *
    1918             :  * NB: We run this during checkpoints even if logical decoding is disabled so
    1919             :  * we cleanup old slots at some point after it got disabled.
    1920             :  */
    1921             : void
    1922        4142 : CheckPointSnapBuild(void)
    1923             : {
    1924             :     XLogRecPtr  cutoff;
    1925             :     XLogRecPtr  redo;
    1926             :     DIR        *snap_dir;
    1927             :     struct dirent *snap_de;
    1928             :     char        path[MAXPGPATH + 21];
    1929             : 
    1930             :     /*
    1931             :      * We start off with a minimum of the last redo pointer. No new
    1932             :      * replication slot will start before that, so that's a safe upper bound
    1933             :      * for removal.
    1934             :      */
    1935        4142 :     redo = GetRedoRecPtr();
    1936             : 
    1937             :     /* now check for the restart ptrs from existing slots */
    1938        4142 :     cutoff = ReplicationSlotsComputeLogicalRestartLSN();
    1939             : 
    1940             :     /* don't start earlier than the restart lsn */
    1941        4142 :     if (redo < cutoff)
    1942           0 :         cutoff = redo;
    1943             : 
    1944        4142 :     snap_dir = AllocateDir("pg_logical/snapshots");
    1945       12640 :     while ((snap_de = ReadDir(snap_dir, "pg_logical/snapshots")) != NULL)
    1946             :     {
    1947             :         uint32      hi;
    1948             :         uint32      lo;
    1949             :         XLogRecPtr  lsn;
    1950             :         struct stat statbuf;
    1951             : 
    1952        8498 :         if (strcmp(snap_de->d_name, ".") == 0 ||
    1953        4356 :             strcmp(snap_de->d_name, "..") == 0)
    1954        8284 :             continue;
    1955             : 
    1956         214 :         snprintf(path, sizeof(path), "pg_logical/snapshots/%s", snap_de->d_name);
    1957             : 
    1958         214 :         if (lstat(path, &statbuf) == 0 && !S_ISREG(statbuf.st_mode))
    1959             :         {
    1960           0 :             elog(DEBUG1, "only regular files expected: %s", path);
    1961           0 :             continue;
    1962             :         }
    1963             : 
    1964             :         /*
    1965             :          * temporary filenames from SnapBuildSerialize() include the LSN and
    1966             :          * everything but are postfixed by .$pid.tmp. We can just remove them
    1967             :          * the same as other files because there can be none that are
    1968             :          * currently being written that are older than cutoff.
    1969             :          *
    1970             :          * We just log a message if a file doesn't fit the pattern, it's
    1971             :          * probably some editors lock/state file or similar...
    1972             :          */
    1973         214 :         if (sscanf(snap_de->d_name, "%X-%X.snap", &hi, &lo) != 2)
    1974             :         {
    1975           0 :             ereport(LOG,
    1976             :                     (errmsg("could not parse file name \"%s\"", path)));
    1977           0 :             continue;
    1978             :         }
    1979             : 
    1980         214 :         lsn = ((uint64) hi) << 32 | lo;
    1981             : 
    1982             :         /* check whether we still need it */
    1983         214 :         if (lsn < cutoff || cutoff == InvalidXLogRecPtr)
    1984             :         {
    1985         154 :             elog(DEBUG1, "removing snapbuild snapshot %s", path);
    1986             : 
    1987             :             /*
    1988             :              * It's not particularly harmful, though strange, if we can't
    1989             :              * remove the file here. Don't prevent the checkpoint from
    1990             :              * completing, that'd be a cure worse than the disease.
    1991             :              */
    1992         154 :             if (unlink(path) < 0)
    1993             :             {
    1994           0 :                 ereport(LOG,
    1995             :                         (errcode_for_file_access(),
    1996             :                          errmsg("could not remove file \"%s\": %m",
    1997             :                                 path)));
    1998           0 :                 continue;
    1999             :             }
    2000             :         }
    2001             :     }
    2002        4142 :     FreeDir(snap_dir);
    2003        4142 : }

Generated by: LCOV version 1.14