LCOV - code coverage report
Current view: top level - src/backend/replication/logical - snapbuild.c (source / functions) Coverage Total Hit
Test: PostgreSQL 20devel Lines: 86.0 % 566 487
Test Date: 2026-10-13 03:15:50 Functions: 100.0 % 32 32
Legend: Lines:     hit not hit
Branches: + taken - not taken # not executed
Branches: 60.4 % 402 243

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

Generated by: LCOV version 2.0-1