LCOV - code coverage report
Current view: top level - src/backend/replication/logical - logical.c (source / functions) Hit Total Coverage
Test: PostgreSQL 13devel Lines: 331 358 92.5 %
Date: 2019-11-15 23:07:02 Functions: 23 23 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  * logical.c
       3             :  *     PostgreSQL logical decoding coordination
       4             :  *
       5             :  * Copyright (c) 2012-2019, PostgreSQL Global Development Group
       6             :  *
       7             :  * IDENTIFICATION
       8             :  *    src/backend/replication/logical/logical.c
       9             :  *
      10             :  * NOTES
      11             :  *    This file coordinates interaction between the various modules that
      12             :  *    together provide logical decoding, primarily by providing so
      13             :  *    called LogicalDecodingContexts. The goal is to encapsulate most of the
      14             :  *    internal complexity for consumers of logical decoding, so they can
      15             :  *    create and consume a changestream with a low amount of code. Builtin
      16             :  *    consumers are the walsender and SQL SRF interface, but it's possible to
      17             :  *    add further ones without changing core code, e.g. to consume changes in
      18             :  *    a bgworker.
      19             :  *
      20             :  *    The idea is that a consumer provides three callbacks, one to read WAL,
      21             :  *    one to prepare a data write, and a final one for actually writing since
      22             :  *    their implementation depends on the type of consumer.  Check
      23             :  *    logicalfuncs.c for an example implementation of a fairly simple consumer
      24             :  *    and an implementation of a WAL reading callback that's suitable for
      25             :  *    simple consumers.
      26             :  *-------------------------------------------------------------------------
      27             :  */
      28             : 
      29             : #include "postgres.h"
      30             : 
      31             : #include "access/xact.h"
      32             : #include "access/xlog_internal.h"
      33             : #include "fmgr.h"
      34             : #include "miscadmin.h"
      35             : #include "replication/decode.h"
      36             : #include "replication/logical.h"
      37             : #include "replication/origin.h"
      38             : #include "replication/reorderbuffer.h"
      39             : #include "replication/snapbuild.h"
      40             : #include "storage/proc.h"
      41             : #include "storage/procarray.h"
      42             : #include "utils/memutils.h"
      43             : 
      44             : /* data for errcontext callback */
      45             : typedef struct LogicalErrorCallbackState
      46             : {
      47             :     LogicalDecodingContext *ctx;
      48             :     const char *callback_name;
      49             :     XLogRecPtr  report_location;
      50             : } LogicalErrorCallbackState;
      51             : 
      52             : /* wrappers around output plugin callbacks */
      53             : static void output_plugin_error_callback(void *arg);
      54             : static void startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt,
      55             :                                bool is_init);
      56             : static void shutdown_cb_wrapper(LogicalDecodingContext *ctx);
      57             : static void begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn);
      58             : static void commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
      59             :                               XLogRecPtr commit_lsn);
      60             : static void change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
      61             :                               Relation relation, ReorderBufferChange *change);
      62             : static void truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
      63             :                                 int nrelations, Relation relations[], ReorderBufferChange *change);
      64             : static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
      65             :                                XLogRecPtr message_lsn, bool transactional,
      66             :                                const char *prefix, Size message_size, const char *message);
      67             : 
      68             : static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, char *plugin);
      69             : 
      70             : /*
      71             :  * Make sure the current settings & environment are capable of doing logical
      72             :  * decoding.
      73             :  */
      74             : void
      75         564 : CheckLogicalDecodingRequirements(void)
      76             : {
      77         564 :     CheckSlotRequirements();
      78             : 
      79             :     /*
      80             :      * NB: Adding a new requirement likely means that RestoreSlotFromDisk()
      81             :      * needs the same check.
      82             :      */
      83             : 
      84         564 :     if (wal_level < WAL_LEVEL_LOGICAL)
      85           0 :         ereport(ERROR,
      86             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
      87             :                  errmsg("logical decoding requires wal_level >= logical")));
      88             : 
      89         564 :     if (MyDatabaseId == InvalidOid)
      90           0 :         ereport(ERROR,
      91             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
      92             :                  errmsg("logical decoding requires a database connection")));
      93             : 
      94             :     /* ----
      95             :      * TODO: We got to change that someday soon...
      96             :      *
      97             :      * There's basically three things missing to allow this:
      98             :      * 1) We need to be able to correctly and quickly identify the timeline a
      99             :      *    LSN belongs to
     100             :      * 2) We need to force hot_standby_feedback to be enabled at all times so
     101             :      *    the primary cannot remove rows we need.
     102             :      * 3) support dropping replication slots referring to a database, in
     103             :      *    dbase_redo. There can't be any active ones due to HS recovery
     104             :      *    conflicts, so that should be relatively easy.
     105             :      * ----
     106             :      */
     107         564 :     if (RecoveryInProgress())
     108           0 :         ereport(ERROR,
     109             :                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
     110             :                  errmsg("logical decoding cannot be used while in recovery")));
     111         564 : }
     112             : 
     113             : /*
     114             :  * Helper function for CreateInitDecodingContext() and
     115             :  * CreateDecodingContext() performing common tasks.
     116             :  */
     117             : static LogicalDecodingContext *
     118         548 : StartupDecodingContext(List *output_plugin_options,
     119             :                        XLogRecPtr start_lsn,
     120             :                        TransactionId xmin_horizon,
     121             :                        bool need_full_snapshot,
     122             :                        bool fast_forward,
     123             :                        XLogPageReadCB read_page,
     124             :                        LogicalOutputPluginWriterPrepareWrite prepare_write,
     125             :                        LogicalOutputPluginWriterWrite do_write,
     126             :                        LogicalOutputPluginWriterUpdateProgress update_progress)
     127             : {
     128             :     ReplicationSlot *slot;
     129             :     MemoryContext context,
     130             :                 old_context;
     131             :     LogicalDecodingContext *ctx;
     132             : 
     133             :     /* shorter lines... */
     134         548 :     slot = MyReplicationSlot;
     135             : 
     136         548 :     context = AllocSetContextCreate(CurrentMemoryContext,
     137             :                                     "Logical decoding context",
     138             :                                     ALLOCSET_DEFAULT_SIZES);
     139         548 :     old_context = MemoryContextSwitchTo(context);
     140         548 :     ctx = palloc0(sizeof(LogicalDecodingContext));
     141             : 
     142         548 :     ctx->context = context;
     143             : 
     144             :     /*
     145             :      * (re-)load output plugins, so we detect a bad (removed) output plugin
     146             :      * now.
     147             :      */
     148         548 :     if (!fast_forward)
     149         544 :         LoadOutputPlugin(&ctx->callbacks, NameStr(slot->data.plugin));
     150             : 
     151             :     /*
     152             :      * Now that the slot's xmin has been set, we can announce ourselves as a
     153             :      * logical decoding backend which doesn't need to be checked individually
     154             :      * when computing the xmin horizon because the xmin is enforced via
     155             :      * replication slots.
     156             :      *
     157             :      * We can only do so if we're outside of a transaction (i.e. the case when
     158             :      * streaming changes via walsender), otherwise an already setup
     159             :      * snapshot/xid would end up being ignored. That's not a particularly
     160             :      * bothersome restriction since the SQL interface can't be used for
     161             :      * streaming anyway.
     162             :      */
     163         546 :     if (!IsTransactionOrTransactionBlock())
     164             :     {
     165          80 :         LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
     166          80 :         MyPgXact->vacuumFlags |= PROC_IN_LOGICAL_DECODING;
     167          80 :         LWLockRelease(ProcArrayLock);
     168             :     }
     169             : 
     170         546 :     ctx->slot = slot;
     171             : 
     172         546 :     ctx->reader = XLogReaderAllocate(wal_segment_size, NULL, read_page, ctx);
     173         546 :     if (!ctx->reader)
     174           0 :         ereport(ERROR,
     175             :                 (errcode(ERRCODE_OUT_OF_MEMORY),
     176             :                  errmsg("out of memory")));
     177             : 
     178         546 :     ctx->reorder = ReorderBufferAllocate();
     179         546 :     ctx->snapshot_builder =
     180         546 :         AllocateSnapshotBuilder(ctx->reorder, xmin_horizon, start_lsn,
     181             :                                 need_full_snapshot);
     182             : 
     183         546 :     ctx->reorder->private_data = ctx;
     184             : 
     185             :     /* wrap output plugin callbacks, so we can add error context information */
     186         546 :     ctx->reorder->begin = begin_cb_wrapper;
     187         546 :     ctx->reorder->apply_change = change_cb_wrapper;
     188         546 :     ctx->reorder->apply_truncate = truncate_cb_wrapper;
     189         546 :     ctx->reorder->commit = commit_cb_wrapper;
     190         546 :     ctx->reorder->message = message_cb_wrapper;
     191             : 
     192         546 :     ctx->out = makeStringInfo();
     193         546 :     ctx->prepare_write = prepare_write;
     194         546 :     ctx->write = do_write;
     195         546 :     ctx->update_progress = update_progress;
     196             : 
     197         546 :     ctx->output_plugin_options = output_plugin_options;
     198             : 
     199         546 :     ctx->fast_forward = fast_forward;
     200             : 
     201         546 :     MemoryContextSwitchTo(old_context);
     202             : 
     203         546 :     return ctx;
     204             : }
     205             : 
     206             : /*
     207             :  * Create a new decoding context, for a new logical slot.
     208             :  *
     209             :  * plugin -- contains the name of the output plugin
     210             :  * output_plugin_options -- contains options passed to the output plugin
     211             :  * restart_lsn -- if given as invalid, it's this routine's responsibility to
     212             :  *      mark WAL as reserved by setting a convenient restart_lsn for the slot.
     213             :  *      Otherwise, we set for decoding to start from the given LSN without
     214             :  *      marking WAL reserved beforehand.  In that scenario, it's up to the
     215             :  *      caller to guarantee that WAL remains available.
     216             :  * read_page, prepare_write, do_write, update_progress --
     217             :  *      callbacks that perform the use-case dependent, actual, work.
     218             :  *
     219             :  * Needs to be called while in a memory context that's at least as long lived
     220             :  * as the decoding context because further memory contexts will be created
     221             :  * inside it.
     222             :  *
     223             :  * Returns an initialized decoding context after calling the output plugin's
     224             :  * startup function.
     225             :  */
     226             : LogicalDecodingContext *
     227         260 : CreateInitDecodingContext(char *plugin,
     228             :                           List *output_plugin_options,
     229             :                           bool need_full_snapshot,
     230             :                           XLogRecPtr restart_lsn,
     231             :                           XLogPageReadCB read_page,
     232             :                           LogicalOutputPluginWriterPrepareWrite prepare_write,
     233             :                           LogicalOutputPluginWriterWrite do_write,
     234             :                           LogicalOutputPluginWriterUpdateProgress update_progress)
     235             : {
     236         260 :     TransactionId xmin_horizon = InvalidTransactionId;
     237             :     ReplicationSlot *slot;
     238             :     LogicalDecodingContext *ctx;
     239             :     MemoryContext old_context;
     240             : 
     241             :     /* shorter lines... */
     242         260 :     slot = MyReplicationSlot;
     243             : 
     244             :     /* first some sanity checks that are unlikely to be violated */
     245         260 :     if (slot == NULL)
     246           0 :         elog(ERROR, "cannot perform logical decoding without an acquired slot");
     247             : 
     248         260 :     if (plugin == NULL)
     249           0 :         elog(ERROR, "cannot initialize logical decoding without a specified plugin");
     250             : 
     251             :     /* Make sure the passed slot is suitable. These are user facing errors. */
     252         260 :     if (SlotIsPhysical(slot))
     253           0 :         ereport(ERROR,
     254             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
     255             :                  errmsg("cannot use physical replication slot for logical decoding")));
     256             : 
     257         260 :     if (slot->data.database != MyDatabaseId)
     258           0 :         ereport(ERROR,
     259             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
     260             :                  errmsg("replication slot \"%s\" was not created in this database",
     261             :                         NameStr(slot->data.name))));
     262             : 
     263         484 :     if (IsTransactionState() &&
     264         224 :         GetTopTransactionIdIfAny() != InvalidTransactionId)
     265           4 :         ereport(ERROR,
     266             :                 (errcode(ERRCODE_ACTIVE_SQL_TRANSACTION),
     267             :                  errmsg("cannot create logical replication slot in transaction that has performed writes")));
     268             : 
     269             :     /* register output plugin name with slot */
     270         256 :     SpinLockAcquire(&slot->mutex);
     271         256 :     StrNCpy(NameStr(slot->data.plugin), plugin, NAMEDATALEN);
     272         256 :     SpinLockRelease(&slot->mutex);
     273             : 
     274         256 :     if (XLogRecPtrIsInvalid(restart_lsn))
     275         244 :         ReplicationSlotReserveWal();
     276             :     else
     277             :     {
     278          12 :         SpinLockAcquire(&slot->mutex);
     279          12 :         slot->data.restart_lsn = restart_lsn;
     280          12 :         SpinLockRelease(&slot->mutex);
     281             :     }
     282             : 
     283             :     /* ----
     284             :      * This is a bit tricky: We need to determine a safe xmin horizon to start
     285             :      * decoding from, to avoid starting from a running xacts record referring
     286             :      * to xids whose rows have been vacuumed or pruned
     287             :      * already. GetOldestSafeDecodingTransactionId() returns such a value, but
     288             :      * without further interlock its return value might immediately be out of
     289             :      * date.
     290             :      *
     291             :      * So we have to acquire the ProcArrayLock to prevent computation of new
     292             :      * xmin horizons by other backends, get the safe decoding xid, and inform
     293             :      * the slot machinery about the new limit. Once that's done the
     294             :      * ProcArrayLock can be released as the slot machinery now is
     295             :      * protecting against vacuum.
     296             :      *
     297             :      * Note that, temporarily, the data, not just the catalog, xmin has to be
     298             :      * reserved if a data snapshot is to be exported.  Otherwise the initial
     299             :      * data snapshot created here is not guaranteed to be valid. After that
     300             :      * the data xmin doesn't need to be managed anymore and the global xmin
     301             :      * should be recomputed. As we are fine with losing the pegged data xmin
     302             :      * after crash - no chance a snapshot would get exported anymore - we can
     303             :      * get away with just setting the slot's
     304             :      * effective_xmin. ReplicationSlotRelease will reset it again.
     305             :      *
     306             :      * ----
     307             :      */
     308         256 :     LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
     309             : 
     310         256 :     xmin_horizon = GetOldestSafeDecodingTransactionId(!need_full_snapshot);
     311             : 
     312         256 :     SpinLockAcquire(&slot->mutex);
     313         256 :     slot->effective_catalog_xmin = xmin_horizon;
     314         256 :     slot->data.catalog_xmin = xmin_horizon;
     315         256 :     if (need_full_snapshot)
     316          76 :         slot->effective_xmin = xmin_horizon;
     317         256 :     SpinLockRelease(&slot->mutex);
     318             : 
     319         256 :     ReplicationSlotsComputeRequiredXmin(true);
     320             : 
     321         256 :     LWLockRelease(ProcArrayLock);
     322             : 
     323         256 :     ReplicationSlotMarkDirty();
     324         256 :     ReplicationSlotSave();
     325             : 
     326         256 :     ctx = StartupDecodingContext(NIL, restart_lsn, xmin_horizon,
     327             :                                  need_full_snapshot, false,
     328             :                                  read_page, prepare_write, do_write,
     329             :                                  update_progress);
     330             : 
     331             :     /* call output plugin initialization callback */
     332         254 :     old_context = MemoryContextSwitchTo(ctx->context);
     333         254 :     if (ctx->callbacks.startup_cb != NULL)
     334         254 :         startup_cb_wrapper(ctx, &ctx->options, true);
     335         254 :     MemoryContextSwitchTo(old_context);
     336             : 
     337         254 :     ctx->reorder->output_rewrites = ctx->options.receive_rewrites;
     338             : 
     339         254 :     return ctx;
     340             : }
     341             : 
     342             : /*
     343             :  * Create a new decoding context, for a logical slot that has previously been
     344             :  * used already.
     345             :  *
     346             :  * start_lsn
     347             :  *      The LSN at which to start decoding.  If InvalidXLogRecPtr, restart
     348             :  *      from the slot's confirmed_flush; otherwise, start from the specified
     349             :  *      location (but move it forwards to confirmed_flush if it's older than
     350             :  *      that, see below).
     351             :  *
     352             :  * output_plugin_options
     353             :  *      options passed to the output plugin.
     354             :  *
     355             :  * fast_forward
     356             :  *      bypass the generation of logical changes.
     357             :  *
     358             :  * read_page, prepare_write, do_write, update_progress
     359             :  *      callbacks that have to be filled to perform the use-case dependent,
     360             :  *      actual work.
     361             :  *
     362             :  * Needs to be called while in a memory context that's at least as long lived
     363             :  * as the decoding context because further memory contexts will be created
     364             :  * inside it.
     365             :  *
     366             :  * Returns an initialized decoding context after calling the output plugin's
     367             :  * startup function.
     368             :  */
     369             : LogicalDecodingContext *
     370         296 : CreateDecodingContext(XLogRecPtr start_lsn,
     371             :                       List *output_plugin_options,
     372             :                       bool fast_forward,
     373             :                       XLogPageReadCB read_page,
     374             :                       LogicalOutputPluginWriterPrepareWrite prepare_write,
     375             :                       LogicalOutputPluginWriterWrite do_write,
     376             :                       LogicalOutputPluginWriterUpdateProgress update_progress)
     377             : {
     378             :     LogicalDecodingContext *ctx;
     379             :     ReplicationSlot *slot;
     380             :     MemoryContext old_context;
     381             : 
     382             :     /* shorter lines... */
     383         296 :     slot = MyReplicationSlot;
     384             : 
     385             :     /* first some sanity checks that are unlikely to be violated */
     386         296 :     if (slot == NULL)
     387           0 :         elog(ERROR, "cannot perform logical decoding without an acquired slot");
     388             : 
     389             :     /* make sure the passed slot is suitable, these are user facing errors */
     390         296 :     if (SlotIsPhysical(slot))
     391           2 :         ereport(ERROR,
     392             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
     393             :                  (errmsg("cannot use physical replication slot for logical decoding"))));
     394             : 
     395         294 :     if (slot->data.database != MyDatabaseId)
     396           2 :         ereport(ERROR,
     397             :                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
     398             :                  (errmsg("replication slot \"%s\" was not created in this database",
     399             :                          NameStr(slot->data.name)))));
     400             : 
     401         292 :     if (start_lsn == InvalidXLogRecPtr)
     402             :     {
     403             :         /* continue from last position */
     404         284 :         start_lsn = slot->data.confirmed_flush;
     405             :     }
     406           8 :     else if (start_lsn < slot->data.confirmed_flush)
     407             :     {
     408             :         /*
     409             :          * It might seem like we should error out in this case, but it's
     410             :          * pretty common for a client to acknowledge a LSN it doesn't have to
     411             :          * do anything for, and thus didn't store persistently, because the
     412             :          * xlog records didn't result in anything relevant for logical
     413             :          * decoding. Clients have to be able to do that to support synchronous
     414             :          * replication.
     415             :          */
     416           0 :         elog(DEBUG1, "cannot stream from %X/%X, minimum is %X/%X, forwarding",
     417             :              (uint32) (start_lsn >> 32), (uint32) start_lsn,
     418             :              (uint32) (slot->data.confirmed_flush >> 32),
     419             :              (uint32) slot->data.confirmed_flush);
     420             : 
     421           0 :         start_lsn = slot->data.confirmed_flush;
     422             :     }
     423             : 
     424         292 :     ctx = StartupDecodingContext(output_plugin_options,
     425             :                                  start_lsn, InvalidTransactionId, false,
     426             :                                  fast_forward, read_page, prepare_write,
     427             :                                  do_write, update_progress);
     428             : 
     429             :     /* call output plugin initialization callback */
     430         292 :     old_context = MemoryContextSwitchTo(ctx->context);
     431         292 :     if (ctx->callbacks.startup_cb != NULL)
     432         288 :         startup_cb_wrapper(ctx, &ctx->options, false);
     433         286 :     MemoryContextSwitchTo(old_context);
     434             : 
     435         286 :     ctx->reorder->output_rewrites = ctx->options.receive_rewrites;
     436             : 
     437         286 :     ereport(LOG,
     438             :             (errmsg("starting logical decoding for slot \"%s\"",
     439             :                     NameStr(slot->data.name)),
     440             :              errdetail("Streaming transactions committing after %X/%X, reading WAL from %X/%X.",
     441             :                        (uint32) (slot->data.confirmed_flush >> 32),
     442             :                        (uint32) slot->data.confirmed_flush,
     443             :                        (uint32) (slot->data.restart_lsn >> 32),
     444             :                        (uint32) slot->data.restart_lsn)));
     445             : 
     446         286 :     return ctx;
     447             : }
     448             : 
     449             : /*
     450             :  * Returns true if a consistent initial decoding snapshot has been built.
     451             :  */
     452             : bool
     453         286 : DecodingContextReady(LogicalDecodingContext *ctx)
     454             : {
     455         286 :     return SnapBuildCurrentState(ctx->snapshot_builder) == SNAPBUILD_CONSISTENT;
     456             : }
     457             : 
     458             : /*
     459             :  * Read from the decoding slot, until it is ready to start extracting changes.
     460             :  */
     461             : void
     462         254 : DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
     463             : {
     464             :     XLogRecPtr  startptr;
     465         254 :     ReplicationSlot *slot = ctx->slot;
     466             : 
     467             :     /* Initialize from where to start reading WAL. */
     468         254 :     startptr = slot->data.restart_lsn;
     469             : 
     470         254 :     elog(DEBUG1, "searching for logical decoding starting point, starting at %X/%X",
     471             :          (uint32) (slot->data.restart_lsn >> 32),
     472             :          (uint32) slot->data.restart_lsn);
     473             : 
     474             :     /* Wait for a consistent starting point */
     475             :     for (;;)
     476          32 :     {
     477             :         XLogRecord *record;
     478         286 :         char       *err = NULL;
     479             : 
     480             :         /* the read_page callback waits for new WAL */
     481         286 :         record = XLogReadRecord(ctx->reader, startptr, &err);
     482         286 :         if (err)
     483           0 :             elog(ERROR, "%s", err);
     484         286 :         if (!record)
     485           0 :             elog(ERROR, "no record found"); /* shouldn't happen */
     486             : 
     487         286 :         startptr = InvalidXLogRecPtr;
     488             : 
     489         286 :         LogicalDecodingProcessRecord(ctx, ctx->reader);
     490             : 
     491             :         /* only continue till we found a consistent spot */
     492         286 :         if (DecodingContextReady(ctx))
     493         254 :             break;
     494             : 
     495          32 :         CHECK_FOR_INTERRUPTS();
     496             :     }
     497             : 
     498         254 :     SpinLockAcquire(&slot->mutex);
     499         254 :     slot->data.confirmed_flush = ctx->reader->EndRecPtr;
     500         254 :     SpinLockRelease(&slot->mutex);
     501         254 : }
     502             : 
     503             : /*
     504             :  * Free a previously allocated decoding context, invoking the shutdown
     505             :  * callback if necessary.
     506             :  */
     507             : void
     508         494 : FreeDecodingContext(LogicalDecodingContext *ctx)
     509             : {
     510         494 :     if (ctx->callbacks.shutdown_cb != NULL)
     511         490 :         shutdown_cb_wrapper(ctx);
     512             : 
     513         494 :     ReorderBufferFree(ctx->reorder);
     514         494 :     FreeSnapshotBuilder(ctx->snapshot_builder);
     515         494 :     XLogReaderFree(ctx->reader);
     516         494 :     MemoryContextDelete(ctx->context);
     517         494 : }
     518             : 
     519             : /*
     520             :  * Prepare a write using the context's output routine.
     521             :  */
     522             : void
     523      293130 : OutputPluginPrepareWrite(struct LogicalDecodingContext *ctx, bool last_write)
     524             : {
     525      293130 :     if (!ctx->accept_writes)
     526           0 :         elog(ERROR, "writes are only accepted in commit, begin and change callbacks");
     527             : 
     528      293130 :     ctx->prepare_write(ctx, ctx->write_location, ctx->write_xid, last_write);
     529      293130 :     ctx->prepared_write = true;
     530      293130 : }
     531             : 
     532             : /*
     533             :  * Perform a write using the context's output routine.
     534             :  */
     535             : void
     536      293130 : OutputPluginWrite(struct LogicalDecodingContext *ctx, bool last_write)
     537             : {
     538      293130 :     if (!ctx->prepared_write)
     539           0 :         elog(ERROR, "OutputPluginPrepareWrite needs to be called before OutputPluginWrite");
     540             : 
     541      293130 :     ctx->write(ctx, ctx->write_location, ctx->write_xid, last_write);
     542      293130 :     ctx->prepared_write = false;
     543      293130 : }
     544             : 
     545             : /*
     546             :  * Update progress tracking (if supported).
     547             :  */
     548             : void
     549         238 : OutputPluginUpdateProgress(struct LogicalDecodingContext *ctx)
     550             : {
     551         238 :     if (!ctx->update_progress)
     552           0 :         return;
     553             : 
     554         238 :     ctx->update_progress(ctx, ctx->write_location, ctx->write_xid);
     555             : }
     556             : 
     557             : /*
     558             :  * Load the output plugin, lookup its output plugin init function, and check
     559             :  * that it provides the required callbacks.
     560             :  */
     561             : static void
     562         544 : LoadOutputPlugin(OutputPluginCallbacks *callbacks, char *plugin)
     563             : {
     564             :     LogicalOutputPluginInit plugin_init;
     565             : 
     566         544 :     plugin_init = (LogicalOutputPluginInit)
     567             :         load_external_function(plugin, "_PG_output_plugin_init", false, NULL);
     568             : 
     569         542 :     if (plugin_init == NULL)
     570           0 :         elog(ERROR, "output plugins have to declare the _PG_output_plugin_init symbol");
     571             : 
     572             :     /* ask the output plugin to fill the callback struct */
     573         542 :     plugin_init(callbacks);
     574             : 
     575         542 :     if (callbacks->begin_cb == NULL)
     576           0 :         elog(ERROR, "output plugins have to register a begin callback");
     577         542 :     if (callbacks->change_cb == NULL)
     578           0 :         elog(ERROR, "output plugins have to register a change callback");
     579         542 :     if (callbacks->commit_cb == NULL)
     580           0 :         elog(ERROR, "output plugins have to register a commit callback");
     581         542 : }
     582             : 
     583             : static void
     584           6 : output_plugin_error_callback(void *arg)
     585             : {
     586           6 :     LogicalErrorCallbackState *state = (LogicalErrorCallbackState *) arg;
     587             : 
     588             :     /* not all callbacks have an associated LSN  */
     589           6 :     if (state->report_location != InvalidXLogRecPtr)
     590           0 :         errcontext("slot \"%s\", output plugin \"%s\", in the %s callback, associated LSN %X/%X",
     591           0 :                    NameStr(state->ctx->slot->data.name),
     592           0 :                    NameStr(state->ctx->slot->data.plugin),
     593             :                    state->callback_name,
     594           0 :                    (uint32) (state->report_location >> 32),
     595           0 :                    (uint32) state->report_location);
     596             :     else
     597          12 :         errcontext("slot \"%s\", output plugin \"%s\", in the %s callback",
     598           6 :                    NameStr(state->ctx->slot->data.name),
     599           6 :                    NameStr(state->ctx->slot->data.plugin),
     600             :                    state->callback_name);
     601           6 : }
     602             : 
     603             : static void
     604         542 : startup_cb_wrapper(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init)
     605             : {
     606             :     LogicalErrorCallbackState state;
     607             :     ErrorContextCallback errcallback;
     608             : 
     609             :     Assert(!ctx->fast_forward);
     610             : 
     611             :     /* Push callback + info on the error context stack */
     612         542 :     state.ctx = ctx;
     613         542 :     state.callback_name = "startup";
     614         542 :     state.report_location = InvalidXLogRecPtr;
     615         542 :     errcallback.callback = output_plugin_error_callback;
     616         542 :     errcallback.arg = (void *) &state;
     617         542 :     errcallback.previous = error_context_stack;
     618         542 :     error_context_stack = &errcallback;
     619             : 
     620             :     /* set output state */
     621         542 :     ctx->accept_writes = false;
     622             : 
     623             :     /* do the actual work: call callback */
     624         542 :     ctx->callbacks.startup_cb(ctx, opt, is_init);
     625             : 
     626             :     /* Pop the error context stack */
     627         536 :     error_context_stack = errcallback.previous;
     628         536 : }
     629             : 
     630             : static void
     631         490 : shutdown_cb_wrapper(LogicalDecodingContext *ctx)
     632             : {
     633             :     LogicalErrorCallbackState state;
     634             :     ErrorContextCallback errcallback;
     635             : 
     636             :     Assert(!ctx->fast_forward);
     637             : 
     638             :     /* Push callback + info on the error context stack */
     639         490 :     state.ctx = ctx;
     640         490 :     state.callback_name = "shutdown";
     641         490 :     state.report_location = InvalidXLogRecPtr;
     642         490 :     errcallback.callback = output_plugin_error_callback;
     643         490 :     errcallback.arg = (void *) &state;
     644         490 :     errcallback.previous = error_context_stack;
     645         490 :     error_context_stack = &errcallback;
     646             : 
     647             :     /* set output state */
     648         490 :     ctx->accept_writes = false;
     649             : 
     650             :     /* do the actual work: call callback */
     651         490 :     ctx->callbacks.shutdown_cb(ctx);
     652             : 
     653             :     /* Pop the error context stack */
     654         490 :     error_context_stack = errcallback.previous;
     655         490 : }
     656             : 
     657             : 
     658             : /*
     659             :  * Callbacks for ReorderBuffer which add in some more information and then call
     660             :  * output_plugin.h plugins.
     661             :  */
     662             : static void
     663         890 : begin_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn)
     664             : {
     665         890 :     LogicalDecodingContext *ctx = cache->private_data;
     666             :     LogicalErrorCallbackState state;
     667             :     ErrorContextCallback errcallback;
     668             : 
     669             :     Assert(!ctx->fast_forward);
     670             : 
     671             :     /* Push callback + info on the error context stack */
     672         890 :     state.ctx = ctx;
     673         890 :     state.callback_name = "begin";
     674         890 :     state.report_location = txn->first_lsn;
     675         890 :     errcallback.callback = output_plugin_error_callback;
     676         890 :     errcallback.arg = (void *) &state;
     677         890 :     errcallback.previous = error_context_stack;
     678         890 :     error_context_stack = &errcallback;
     679             : 
     680             :     /* set output state */
     681         890 :     ctx->accept_writes = true;
     682         890 :     ctx->write_xid = txn->xid;
     683         890 :     ctx->write_location = txn->first_lsn;
     684             : 
     685             :     /* do the actual work: call callback */
     686         890 :     ctx->callbacks.begin_cb(ctx, txn);
     687             : 
     688             :     /* Pop the error context stack */
     689         890 :     error_context_stack = errcallback.previous;
     690         890 : }
     691             : 
     692             : static void
     693         890 : commit_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
     694             :                   XLogRecPtr commit_lsn)
     695             : {
     696         890 :     LogicalDecodingContext *ctx = cache->private_data;
     697             :     LogicalErrorCallbackState state;
     698             :     ErrorContextCallback errcallback;
     699             : 
     700             :     Assert(!ctx->fast_forward);
     701             : 
     702             :     /* Push callback + info on the error context stack */
     703         890 :     state.ctx = ctx;
     704         890 :     state.callback_name = "commit";
     705         890 :     state.report_location = txn->final_lsn; /* beginning of commit record */
     706         890 :     errcallback.callback = output_plugin_error_callback;
     707         890 :     errcallback.arg = (void *) &state;
     708         890 :     errcallback.previous = error_context_stack;
     709         890 :     error_context_stack = &errcallback;
     710             : 
     711             :     /* set output state */
     712         890 :     ctx->accept_writes = true;
     713         890 :     ctx->write_xid = txn->xid;
     714         890 :     ctx->write_location = txn->end_lsn; /* points to the end of the record */
     715             : 
     716             :     /* do the actual work: call callback */
     717         890 :     ctx->callbacks.commit_cb(ctx, txn, commit_lsn);
     718             : 
     719             :     /* Pop the error context stack */
     720         890 :     error_context_stack = errcallback.previous;
     721         890 : }
     722             : 
     723             : static void
     724      293944 : change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
     725             :                   Relation relation, ReorderBufferChange *change)
     726             : {
     727      293944 :     LogicalDecodingContext *ctx = cache->private_data;
     728             :     LogicalErrorCallbackState state;
     729             :     ErrorContextCallback errcallback;
     730             : 
     731             :     Assert(!ctx->fast_forward);
     732             : 
     733             :     /* Push callback + info on the error context stack */
     734      293944 :     state.ctx = ctx;
     735      293944 :     state.callback_name = "change";
     736      293944 :     state.report_location = change->lsn;
     737      293944 :     errcallback.callback = output_plugin_error_callback;
     738      293944 :     errcallback.arg = (void *) &state;
     739      293944 :     errcallback.previous = error_context_stack;
     740      293944 :     error_context_stack = &errcallback;
     741             : 
     742             :     /* set output state */
     743      293944 :     ctx->accept_writes = true;
     744      293944 :     ctx->write_xid = txn->xid;
     745             : 
     746             :     /*
     747             :      * report this change's lsn so replies from clients can give an up2date
     748             :      * answer. This won't ever be enough (and shouldn't be!) to confirm
     749             :      * receipt of this transaction, but it might allow another transaction's
     750             :      * commit to be confirmed with one message.
     751             :      */
     752      293944 :     ctx->write_location = change->lsn;
     753             : 
     754      293944 :     ctx->callbacks.change_cb(ctx, txn, relation, change);
     755             : 
     756             :     /* Pop the error context stack */
     757      293944 :     error_context_stack = errcallback.previous;
     758      293944 : }
     759             : 
     760             : static void
     761          16 : truncate_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
     762             :                     int nrelations, Relation relations[], ReorderBufferChange *change)
     763             : {
     764          16 :     LogicalDecodingContext *ctx = cache->private_data;
     765             :     LogicalErrorCallbackState state;
     766             :     ErrorContextCallback errcallback;
     767             : 
     768             :     Assert(!ctx->fast_forward);
     769             : 
     770          16 :     if (!ctx->callbacks.truncate_cb)
     771           0 :         return;
     772             : 
     773             :     /* Push callback + info on the error context stack */
     774          16 :     state.ctx = ctx;
     775          16 :     state.callback_name = "truncate";
     776          16 :     state.report_location = change->lsn;
     777          16 :     errcallback.callback = output_plugin_error_callback;
     778          16 :     errcallback.arg = (void *) &state;
     779          16 :     errcallback.previous = error_context_stack;
     780          16 :     error_context_stack = &errcallback;
     781             : 
     782             :     /* set output state */
     783          16 :     ctx->accept_writes = true;
     784          16 :     ctx->write_xid = txn->xid;
     785             : 
     786             :     /*
     787             :      * report this change's lsn so replies from clients can give an up2date
     788             :      * answer. This won't ever be enough (and shouldn't be!) to confirm
     789             :      * receipt of this transaction, but it might allow another transaction's
     790             :      * commit to be confirmed with one message.
     791             :      */
     792          16 :     ctx->write_location = change->lsn;
     793             : 
     794          16 :     ctx->callbacks.truncate_cb(ctx, txn, nrelations, relations, change);
     795             : 
     796             :     /* Pop the error context stack */
     797          16 :     error_context_stack = errcallback.previous;
     798             : }
     799             : 
     800             : bool
     801     2365442 : filter_by_origin_cb_wrapper(LogicalDecodingContext *ctx, RepOriginId origin_id)
     802             : {
     803             :     LogicalErrorCallbackState state;
     804             :     ErrorContextCallback errcallback;
     805             :     bool        ret;
     806             : 
     807             :     Assert(!ctx->fast_forward);
     808             : 
     809             :     /* Push callback + info on the error context stack */
     810     2365442 :     state.ctx = ctx;
     811     2365442 :     state.callback_name = "filter_by_origin";
     812     2365442 :     state.report_location = InvalidXLogRecPtr;
     813     2365442 :     errcallback.callback = output_plugin_error_callback;
     814     2365442 :     errcallback.arg = (void *) &state;
     815     2365442 :     errcallback.previous = error_context_stack;
     816     2365442 :     error_context_stack = &errcallback;
     817             : 
     818             :     /* set output state */
     819     2365442 :     ctx->accept_writes = false;
     820             : 
     821             :     /* do the actual work: call callback */
     822     2365442 :     ret = ctx->callbacks.filter_by_origin_cb(ctx, origin_id);
     823             : 
     824             :     /* Pop the error context stack */
     825     2365442 :     error_context_stack = errcallback.previous;
     826             : 
     827     2365442 :     return ret;
     828             : }
     829             : 
     830             : static void
     831          16 : message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
     832             :                    XLogRecPtr message_lsn, bool transactional,
     833             :                    const char *prefix, Size message_size, const char *message)
     834             : {
     835          16 :     LogicalDecodingContext *ctx = cache->private_data;
     836             :     LogicalErrorCallbackState state;
     837             :     ErrorContextCallback errcallback;
     838             : 
     839             :     Assert(!ctx->fast_forward);
     840             : 
     841          16 :     if (ctx->callbacks.message_cb == NULL)
     842           0 :         return;
     843             : 
     844             :     /* Push callback + info on the error context stack */
     845          16 :     state.ctx = ctx;
     846          16 :     state.callback_name = "message";
     847          16 :     state.report_location = message_lsn;
     848          16 :     errcallback.callback = output_plugin_error_callback;
     849          16 :     errcallback.arg = (void *) &state;
     850          16 :     errcallback.previous = error_context_stack;
     851          16 :     error_context_stack = &errcallback;
     852             : 
     853             :     /* set output state */
     854          16 :     ctx->accept_writes = true;
     855          16 :     ctx->write_xid = txn != NULL ? txn->xid : InvalidTransactionId;
     856          16 :     ctx->write_location = message_lsn;
     857             : 
     858             :     /* do the actual work: call callback */
     859          16 :     ctx->callbacks.message_cb(ctx, txn, message_lsn, transactional, prefix,
     860             :                               message_size, message);
     861             : 
     862             :     /* Pop the error context stack */
     863          16 :     error_context_stack = errcallback.previous;
     864             : }
     865             : 
     866             : /*
     867             :  * Set the required catalog xmin horizon for historic snapshots in the current
     868             :  * replication slot.
     869             :  *
     870             :  * Note that in the most cases, we won't be able to immediately use the xmin
     871             :  * to increase the xmin horizon: we need to wait till the client has confirmed
     872             :  * receiving current_lsn with LogicalConfirmReceivedLocation().
     873             :  */
     874             : void
     875         158 : LogicalIncreaseXminForSlot(XLogRecPtr current_lsn, TransactionId xmin)
     876             : {
     877         158 :     bool        updated_xmin = false;
     878             :     ReplicationSlot *slot;
     879             : 
     880         158 :     slot = MyReplicationSlot;
     881             : 
     882             :     Assert(slot != NULL);
     883             : 
     884         158 :     SpinLockAcquire(&slot->mutex);
     885             : 
     886             :     /*
     887             :      * don't overwrite if we already have a newer xmin. This can happen if we
     888             :      * restart decoding in a slot.
     889             :      */
     890         158 :     if (TransactionIdPrecedesOrEquals(xmin, slot->data.catalog_xmin))
     891             :     {
     892             :     }
     893             : 
     894             :     /*
     895             :      * If the client has already confirmed up to this lsn, we directly can
     896             :      * mark this as accepted. This can happen if we restart decoding in a
     897             :      * slot.
     898             :      */
     899          54 :     else if (current_lsn <= slot->data.confirmed_flush)
     900             :     {
     901          12 :         slot->candidate_catalog_xmin = xmin;
     902          12 :         slot->candidate_xmin_lsn = current_lsn;
     903             : 
     904             :         /* our candidate can directly be used */
     905          12 :         updated_xmin = true;
     906             :     }
     907             : 
     908             :     /*
     909             :      * Only increase if the previous values have been applied, otherwise we
     910             :      * might never end up updating if the receiver acks too slowly.
     911             :      */
     912          42 :     else if (slot->candidate_xmin_lsn == InvalidXLogRecPtr)
     913             :     {
     914          10 :         slot->candidate_catalog_xmin = xmin;
     915          10 :         slot->candidate_xmin_lsn = current_lsn;
     916             :     }
     917         158 :     SpinLockRelease(&slot->mutex);
     918             : 
     919             :     /* candidate already valid with the current flush position, apply */
     920         158 :     if (updated_xmin)
     921          12 :         LogicalConfirmReceivedLocation(slot->data.confirmed_flush);
     922         158 : }
     923             : 
     924             : /*
     925             :  * Mark the minimal LSN (restart_lsn) we need to read to replay all
     926             :  * transactions that have not yet committed at current_lsn.
     927             :  *
     928             :  * Just like LogicalIncreaseXminForSlot this only takes effect when the
     929             :  * client has confirmed to have received current_lsn.
     930             :  */
     931             : void
     932         126 : LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart_lsn)
     933             : {
     934         126 :     bool        updated_lsn = false;
     935             :     ReplicationSlot *slot;
     936             : 
     937         126 :     slot = MyReplicationSlot;
     938             : 
     939             :     Assert(slot != NULL);
     940             :     Assert(restart_lsn != InvalidXLogRecPtr);
     941             :     Assert(current_lsn != InvalidXLogRecPtr);
     942             : 
     943         126 :     SpinLockAcquire(&slot->mutex);
     944             : 
     945             :     /* don't overwrite if have a newer restart lsn */
     946         126 :     if (restart_lsn <= slot->data.restart_lsn)
     947             :     {
     948             :     }
     949             : 
     950             :     /*
     951             :      * We might have already flushed far enough to directly accept this lsn,
     952             :      * in this case there is no need to check for existing candidate LSNs
     953             :      */
     954         126 :     else if (current_lsn <= slot->data.confirmed_flush)
     955             :     {
     956          82 :         slot->candidate_restart_valid = current_lsn;
     957          82 :         slot->candidate_restart_lsn = restart_lsn;
     958             : 
     959             :         /* our candidate can directly be used */
     960          82 :         updated_lsn = true;
     961             :     }
     962             : 
     963             :     /*
     964             :      * Only increase if the previous values have been applied, otherwise we
     965             :      * might never end up updating if the receiver acks too slowly. A missed
     966             :      * value here will just cause some extra effort after reconnecting.
     967             :      */
     968         126 :     if (slot->candidate_restart_valid == InvalidXLogRecPtr)
     969             :     {
     970          14 :         slot->candidate_restart_valid = current_lsn;
     971          14 :         slot->candidate_restart_lsn = restart_lsn;
     972             : 
     973          14 :         elog(DEBUG1, "got new restart lsn %X/%X at %X/%X",
     974             :              (uint32) (restart_lsn >> 32), (uint32) restart_lsn,
     975             :              (uint32) (current_lsn >> 32), (uint32) current_lsn);
     976             :     }
     977             :     else
     978             :     {
     979         112 :         elog(DEBUG1, "failed to increase restart lsn: proposed %X/%X, after %X/%X, current candidate %X/%X, current after %X/%X, flushed up to %X/%X",
     980             :              (uint32) (restart_lsn >> 32), (uint32) restart_lsn,
     981             :              (uint32) (current_lsn >> 32), (uint32) current_lsn,
     982             :              (uint32) (slot->candidate_restart_lsn >> 32),
     983             :              (uint32) slot->candidate_restart_lsn,
     984             :              (uint32) (slot->candidate_restart_valid >> 32),
     985             :              (uint32) slot->candidate_restart_valid,
     986             :              (uint32) (slot->data.confirmed_flush >> 32),
     987             :              (uint32) slot->data.confirmed_flush
     988             :             );
     989             :     }
     990         126 :     SpinLockRelease(&slot->mutex);
     991             : 
     992             :     /* candidates are already valid with the current flush position, apply */
     993         126 :     if (updated_lsn)
     994          82 :         LogicalConfirmReceivedLocation(slot->data.confirmed_flush);
     995         126 : }
     996             : 
     997             : /*
     998             :  * Handle a consumer's confirmation having received all changes up to lsn.
     999             :  */
    1000             : void
    1001        2000 : LogicalConfirmReceivedLocation(XLogRecPtr lsn)
    1002             : {
    1003             :     Assert(lsn != InvalidXLogRecPtr);
    1004             : 
    1005             :     /* Do an unlocked check for candidate_lsn first. */
    1006        3982 :     if (MyReplicationSlot->candidate_xmin_lsn != InvalidXLogRecPtr ||
    1007        1982 :         MyReplicationSlot->candidate_restart_valid != InvalidXLogRecPtr)
    1008         108 :     {
    1009         108 :         bool        updated_xmin = false;
    1010         108 :         bool        updated_restart = false;
    1011             : 
    1012         108 :         SpinLockAcquire(&MyReplicationSlot->mutex);
    1013             : 
    1014         108 :         MyReplicationSlot->data.confirmed_flush = lsn;
    1015             : 
    1016             :         /* if we're past the location required for bumping xmin, do so */
    1017         126 :         if (MyReplicationSlot->candidate_xmin_lsn != InvalidXLogRecPtr &&
    1018          18 :             MyReplicationSlot->candidate_xmin_lsn <= lsn)
    1019             :         {
    1020             :             /*
    1021             :              * We have to write the changed xmin to disk *before* we change
    1022             :              * the in-memory value, otherwise after a crash we wouldn't know
    1023             :              * that some catalog tuples might have been removed already.
    1024             :              *
    1025             :              * Ensure that by first writing to ->xmin and only update
    1026             :              * ->effective_xmin once the new state is synced to disk. After a
    1027             :              * crash ->effective_xmin is set to ->xmin.
    1028             :              */
    1029          36 :             if (TransactionIdIsValid(MyReplicationSlot->candidate_catalog_xmin) &&
    1030          18 :                 MyReplicationSlot->data.catalog_xmin != MyReplicationSlot->candidate_catalog_xmin)
    1031             :             {
    1032          18 :                 MyReplicationSlot->data.catalog_xmin = MyReplicationSlot->candidate_catalog_xmin;
    1033          18 :                 MyReplicationSlot->candidate_catalog_xmin = InvalidTransactionId;
    1034          18 :                 MyReplicationSlot->candidate_xmin_lsn = InvalidXLogRecPtr;
    1035          18 :                 updated_xmin = true;
    1036             :             }
    1037             :         }
    1038             : 
    1039         204 :         if (MyReplicationSlot->candidate_restart_valid != InvalidXLogRecPtr &&
    1040          96 :             MyReplicationSlot->candidate_restart_valid <= lsn)
    1041             :         {
    1042             :             Assert(MyReplicationSlot->candidate_restart_lsn != InvalidXLogRecPtr);
    1043             : 
    1044          94 :             MyReplicationSlot->data.restart_lsn = MyReplicationSlot->candidate_restart_lsn;
    1045          94 :             MyReplicationSlot->candidate_restart_lsn = InvalidXLogRecPtr;
    1046          94 :             MyReplicationSlot->candidate_restart_valid = InvalidXLogRecPtr;
    1047          94 :             updated_restart = true;
    1048             :         }
    1049             : 
    1050         108 :         SpinLockRelease(&MyReplicationSlot->mutex);
    1051             : 
    1052             :         /* first write new xmin to disk, so we know what's up after a crash */
    1053         108 :         if (updated_xmin || updated_restart)
    1054             :         {
    1055         106 :             ReplicationSlotMarkDirty();
    1056         106 :             ReplicationSlotSave();
    1057         106 :             elog(DEBUG1, "updated xmin: %u restart: %u", updated_xmin, updated_restart);
    1058             :         }
    1059             : 
    1060             :         /*
    1061             :          * Now the new xmin is safely on disk, we can let the global value
    1062             :          * advance. We do not take ProcArrayLock or similar since we only
    1063             :          * advance xmin here and there's not much harm done by a concurrent
    1064             :          * computation missing that.
    1065             :          */
    1066         108 :         if (updated_xmin)
    1067             :         {
    1068          18 :             SpinLockAcquire(&MyReplicationSlot->mutex);
    1069          18 :             MyReplicationSlot->effective_catalog_xmin = MyReplicationSlot->data.catalog_xmin;
    1070          18 :             SpinLockRelease(&MyReplicationSlot->mutex);
    1071             : 
    1072          18 :             ReplicationSlotsComputeRequiredXmin(false);
    1073          18 :             ReplicationSlotsComputeRequiredLSN();
    1074             :         }
    1075             :     }
    1076             :     else
    1077             :     {
    1078        1892 :         SpinLockAcquire(&MyReplicationSlot->mutex);
    1079        1892 :         MyReplicationSlot->data.confirmed_flush = lsn;
    1080        1892 :         SpinLockRelease(&MyReplicationSlot->mutex);
    1081             :     }
    1082        2000 : }

Generated by: LCOV version 1.13