Preserve required !catalog tuples while computing initial decoding snapshot.

The logical decoding machinery already preserved all the required
catalog tuples, which is sufficient in the course of normal logical
decoding, but did not guarantee that non-catalog tuples were preserved
during computation of the initial snapshot when creating a slot over
the replication protocol.

This could cause a corrupted initial snapshot being exported.  The
time window for issues is usually not terribly large, but on a busy
server it's perfectly possible to it hit it.  Ongoing decoding is not
affected by this bug.

To avoid increased overhead for the SQL API, only retain additional
tuples when a logical slot is being created over the replication
protocol.  To do so this commit changes the signature of
CreateInitDecodingContext(), but it seems unlikely that it's being
used in an extension, so that's probably ok.

In a drive-by fix, fix handling of
ReplicationSlotsComputeRequiredXmin's already_locked argument, which
should only apply to ProcArrayLock, not ReplicationSlotControlLock.

Reported-By: Erik Rijkers
Analyzed-By: Petr Jelinek
Author: Petr Jelinek, heavily editorialized by Andres Freund
Reviewed-By: Andres Freund
Discussion: https://postgr.es/m/9a897b86-46e1-9915-ee4c-da02e4ff6a95@2ndquadrant.com
Backport: 9.4, where logical decoding was introduced.

diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 1512be5322c254b7913815eab81fd0256ba1f838..40432fe640786d7b7324b406215a6b1b42ac27e5 100644 (file)
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -210,6 +210,7 @@ StartupDecodingContext(List *output_plugin_options,
 LogicalDecodingContext *
 CreateInitDecodingContext(char *plugin,
                                                  List *output_plugin_options,
+                                                 bool need_full_snapshot,
                                                  XLogPageReadCB read_page,
                                                  LogicalOutputPluginWriterPrepareWrite prepare_write,
                                                  LogicalOutputPluginWriterWrite do_write)
@@ -267,23 +268,31 @@ CreateInitDecodingContext(char *plugin,
         * the slot machinery about the new limit. Once that's done the
         * ProcArrayLock can be released as the slot machinery now is
         * protecting against vacuum.
+        *
+        * Note that, temporarily, the data, not just the catalog, xmin has to be
+        * reserved if a data snapshot is to be exported.  Otherwise the initial
+        * data snapshot created here is not guaranteed to be valid. After that
+        * the data xmin doesn't need to be managed anymore and the global xmin
+        * should be recomputed. As we are fine with losing the pegged data xmin
+        * after crash - no chance a snapshot would get exported anymore - we can
+        * get away with just setting the slot's
+        * effective_xmin. ReplicationSlotRelease will reset it again.
+        *
         * ----
         */
        LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
 
-       slot->effective_catalog_xmin = GetOldestSafeDecodingTransactionId();
-       slot->data.catalog_xmin = slot->effective_catalog_xmin;
+       xmin_horizon = GetOldestSafeDecodingTransactionId(need_full_snapshot);
+
+       slot->effective_catalog_xmin = xmin_horizon;
+       slot->data.catalog_xmin = xmin_horizon;
+       if (need_full_snapshot)
+               slot->effective_xmin = xmin_horizon;
 
        ReplicationSlotsComputeRequiredXmin(true);
 
        LWLockRelease(ProcArrayLock);
 
-       /*
-        * tell the snapshot builder to only assemble snapshot once reaching the
-        * running_xact's record with the respective xmin.
-        */
-       xmin_horizon = slot->data.catalog_xmin;
-
        ReplicationSlotMarkDirty();
        ReplicationSlotSave();
 

diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index 6ae09ce400478563930f70e7c3914f88404a63d1..727ef7785377c1a4c1d910902dfcc861e1ece102 100644 (file)
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -551,6 +551,18 @@ SnapBuildExportSnapshot(SnapBuild *builder)
         * mechanism. Due to that we can do this without locks, we're only
         * changing our own value.
         */
+#ifdef USE_ASSERT_CHECKING
+       {
+               TransactionId safeXid;
+
+               LWLockAcquire(ProcArrayLock, LW_SHARED);
+               safeXid = GetOldestSafeDecodingTransactionId(true);
+               LWLockRelease(ProcArrayLock);
+
+               Assert(TransactionIdPrecedesOrEquals(safeXid, snap->xmin));
+       }
+#endif
+
        MyPgXact->xmin = snap->xmin;
 
        /* allocate in transaction context */

diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 0b2575ee9d0b3b9f7b418202d2c7c0293bf1aba6..d64cfc6dc9e744834c5ad4d8ad0f0e390daf9274 100644 (file)
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -397,6 +397,22 @@ ReplicationSlotRelease(void)
                SpinLockRelease(&slot->mutex);
        }
 
+
+       /*
+        * If slot needed to temporarily restrain both data and catalog xmin to
+        * create the catalog snapshot, remove that temporary constraint.
+        * Snapshots can only be exported while the initial snapshot is still
+        * acquired.
+        */
+       if (!TransactionIdIsValid(slot->data.xmin) &&
+               TransactionIdIsValid(slot->effective_xmin))
+       {
+               SpinLockAcquire(&slot->mutex);
+               slot->effective_xmin = InvalidTransactionId;
+               SpinLockRelease(&slot->mutex);
+               ReplicationSlotsComputeRequiredXmin(false);
+       }
+
        MyReplicationSlot = NULL;
 
        /* might not have been set when we've been a plain slot */
@@ -574,6 +590,9 @@ ReplicationSlotPersist(void)
 
 /*
  * Compute the oldest xmin across all slots and store it in the ProcArray.
+ *
+ * If already_locked is true, ProcArrayLock has already been acquired
+ * exclusively.
  */
 void
 ReplicationSlotsComputeRequiredXmin(bool already_locked)
@@ -584,8 +603,7 @@ ReplicationSlotsComputeRequiredXmin(bool already_locked)
 
        Assert(ReplicationSlotCtl != NULL);
 
-       if (!already_locked)
-               LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+       LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
 
        for (i = 0; i < max_replication_slots; i++)
        {
@@ -614,8 +632,7 @@ ReplicationSlotsComputeRequiredXmin(bool already_locked)
                        agg_catalog_xmin = effective_catalog_xmin;
        }
 
-       if (!already_locked)
-               LWLockRelease(ReplicationSlotControlLock);
+       LWLockRelease(ReplicationSlotControlLock);
 
        ProcArraySetReplicationSlotXmin(agg_xmin, agg_catalog_xmin, already_locked);
 }

diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index f9087619d2b1a5bfd8f3f767d1351d12c2479d53..de8cd2a67e4f3b2cd74c967203c8130320528462 100644 (file)
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -125,8 +125,8 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
        /*
         * Create logical decoding context, to build the initial snapshot.
         */
-       ctx = CreateInitDecodingContext(
-                                                                       NameStr(*plugin), NIL,
+       ctx = CreateInitDecodingContext(NameStr(*plugin), NIL,
+                                                                       false, /* do not build snapshot */
                                                                        logical_read_local_xlog_page, NULL, NULL);
 
        /* build initial snapshot, might take a while */

diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index b2c059e6b0bea8bb5e67eb1a990a4d070fba61f9..834bf947a37a1741a01e7414fa0fbee9d3f0540e 100644 (file)
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -813,6 +813,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
                LogicalDecodingContext *ctx;
 
                ctx = CreateInitDecodingContext(cmd->plugin, NIL,
+                                                                               true, /* build snapshot */
                                                                                logical_read_xlog_page,
                                                                                WalSndPrepareWrite, WalSndWriteData);
 

diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 651c561dfd50bba325cb593e0d41b29d47c876d0..a3a0207872d1bacd5d947a5bdfcbabfed51538a2 100644 (file)
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -2153,7 +2153,7 @@ GetOldestActiveTransactionId(void)
  * that the caller will immediately use the xid to peg the xmin horizon.
  */
 TransactionId
-GetOldestSafeDecodingTransactionId(void)
+GetOldestSafeDecodingTransactionId(bool catalogOnly)
 {
        ProcArrayStruct *arrayP = procArray;
        TransactionId oldestSafeXid;
@@ -2176,9 +2176,17 @@ GetOldestSafeDecodingTransactionId(void)
        /*
         * If there's already a slot pegging the xmin horizon, we can start with
         * that value, it's guaranteed to be safe since it's computed by this
-        * routine initially and has been enforced since.
+        * routine initially and has been enforced since.  We can always use the
+        * slot's general xmin horizon, but the catalog horizon is only usable
+        * when we only catalog data is going to be looked at.
         */
-       if (TransactionIdIsValid(procArray->replication_slot_catalog_xmin) &&
+       if (TransactionIdIsValid(procArray->replication_slot_xmin) &&
+               TransactionIdPrecedes(procArray->replication_slot_xmin,
+                                                         oldestSafeXid))
+               oldestSafeXid = procArray->replication_slot_xmin;
+
+       if (catalogOnly &&
+               TransactionIdIsValid(procArray->replication_slot_catalog_xmin) &&
                TransactionIdPrecedes(procArray->replication_slot_catalog_xmin,
                                                          oldestSafeXid))
                oldestSafeXid = procArray->replication_slot_catalog_xmin;

diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index 947000e63f8989eb78f9ea767659249198499409..ad3ff5eacba85be4e4e9f0cf5c7c6089ea4d0709 100644 (file)
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -79,6 +79,7 @@ extern void CheckLogicalDecodingRequirements(void);
 
 extern LogicalDecodingContext *CreateInitDecodingContext(char *plugin,
                                                  List *output_plugin_options,
+                                                 bool need_full_snapshot,
                                                  XLogPageReadCB read_page,
                                                  LogicalOutputPluginWriterPrepareWrite prepare_write,
                                                  LogicalOutputPluginWriterWrite do_write);

diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h
index ecc3c4a4f63bac5361d1df066b2183c48244e334..f6af33340b855bbca2a893b3d1fb836c44ef4eb9 100644 (file)
--- a/src/include/storage/procarray.h
+++ b/src/include/storage/procarray.h
@@ -55,7 +55,7 @@ extern bool TransactionIdIsInProgress(TransactionId xid);
 extern bool TransactionIdIsActive(TransactionId xid);
 extern TransactionId GetOldestXmin(Relation rel, bool ignoreVacuum);
 extern TransactionId GetOldestActiveTransactionId(void);
-extern TransactionId GetOldestSafeDecodingTransactionId(void);
+extern TransactionId GetOldestSafeDecodingTransactionId(bool catalogOnly);
 
 extern VirtualTransactionId *GetVirtualXIDsDelayingChkpt(int *nvxids);
 extern bool HaveVirtualXIDsDelayingChkpt(VirtualTransactionId *vxids, int nvxids);