Re: ERROR: subtransaction logged without previous top-level txn record

Hi,

Our customer also encountered this issue and I've looked into it. The problem is
reproduced well enough using the instructions in the previous message.

The check leading to this ERROR is too strict, it forbids legit behaviours. Say
we have in WAL

[ <xl_xact_assignment_1> <restart_lsn> <subxact_change> <xl_xact_assignment_1> <commit> confirmed_flush_lsn> ]

- First xl_xact_assignment record is beyond reading, i.e. earlier
  restart_lsn, where ready snapshot will be taken from disk.
- After restart_lsn there is some change of a subxact.
- After that, there is second xl_xact_assignment (for another subxact)
  revealing relationship between top and first subxact, where this ERROR fires.

Such transaction won't be streamed because we hadn't seen it in full. It must be
finished before streaming will start, i.e. before confirmed_flush_lsn.

Of course, the easiest fix is to just throw the check out. However, supposing
that someone would probably want to relax it instead, I considered ways to
accomplish this. Something like 'if we are still in SNAPSHOT_FULL and xid is
before SnapBuildNextPhaseAt, just ignore xl_xact_assignment record, we haven't
seen such xact in full and definitely won't stream it.' That led to discovery of
another bug in the place which I had found suspicious long before.

Snapbuilder enters into SNAPBUILD_CONSISTENT immediately after deserializing the
snapshot. Generally this is incorrect because SNAPBUILD_CONSISTENT means not
just complete snapshot (snapshot by itself in FULL state is just good as in
CONSISTENT), but also reorderbuffer filled with all currently running
xacts. This is painless for decoding sessions with existing slots because they
won't stream anything before confirmed_flush_lsn is reached anyway, at which
point all transactions which hadn't got into reorderbuffer would definitely
finish. However, new slots might be created too early, thus losing (not
decoding) parts of transactions committed after freshly created
confirmed_flush_lsn. This can happen under the following extremely unlucky
circumstances:
  - New slot creation reserves point in WAL since which it would read it
    (GetXLogInsertRecPtr);
  - It logs xl_running_xacts to start assembling a snapshot;
  - Running decoding session with another slot quickly reads this
    xl_running_xacts and serializes its snapshot;
  - New slot reads xl_running_xacts and picks this snapshot up, saying that it
    is ready to stream henceforth, though its reorderbuffer is empty.

Exact reproducing steps:

-- session 1
create table t (i int);
select pg_create_logical_replication_slot('slot_1', 'test_decoding');

-- session 2
begin;
insert into t values (1);

-- session 3, start slot creation
select pg_create_logical_replication_slot('slot_2', 'test_decoding');
-- stop (with gdb or something) it at DecodingContextFindStartpoint(ctx);

-- session 1
-- xl_running_xacts is dumped by ReplicationSlotReserveWal in previous command, no
-- need to sleep; our snap will be immediately serialized there
SELECT data FROM pg_logical_slot_get_changes('slot_1', NULL, NULL, 'include-xids', '1', 'skip-empty-xacts', '0');

-- continue slot_2 creation

-- session 2: insert some more and commit
insert into t values (1);
commit;

-- now this would find second insert, but not the first one
SELECT data FROM pg_logical_slot_get_changes('slot_2', NULL, NULL, 'include-xids', '1', 'skip-empty-xacts', '0');


What we can do here? Initially I was like, ok, then let's get into FULL_SNAPSHOT
upon deserializing the snap and wait for all xacts finish as usual. However, to
my surprise I've found this impossible. That is, snapbuilder has no way to
enforce that we go into CONSISTENT only when we have seen all running xacts
completely without risk of skipping legit transactions. Specifically, after
deserializing FULL snapshot snapbuilder must iterate over WAL further until all
running xacts finish, as we must see with correct snapshots all changes of every
transaction we are going to stream. However, snapbuilder can't *immediately*
notice this point, because
 - Snapbuilder updates xmin (first running xact) by taking it from xl_running_xacts
   (c.f. SnapBuildProcessRunningXacts). Even if we guarantee that, for
   each possible WAL reading starting position, there is always an an
   xl_running_xacts records logged right before the earliest possible
   streaming point -- IOW, after all xacts which we can't stream had
   finished (which is currently true btw, as slot's advancement is
   considered only at xl_running_xacts) -- that would not be enough due
   to races around xl_running_xacts, i.e with WAL like
   [ <T1> <restart_lsn> <T1 commit> <confirmed_flush_lsn, xrx> <T2 commit> ]
   T2 might be skipped if T1 is shown as running in xl_running_xacts.
 - Tracking xmin manually by recoding commits is not only inefficient,
   it just not feasible because serialized snapshot is not full: it
   contains only committed catalog-modifying xacts. Thus, we can't
   distinguish non-catalog-modifying xact committed before serialized
   snapshot from not yet committed one.

Which means only code external to snapbuilder knows the earliest point suitable
for streaming; slot advancement machinery ensures that <restart_lsn,
confirmed_flush_lsn> pair is always good. So possible fix is the following: if
snapbuilder's user knows exact LSN since which streaming is safe (existing slot,
essentially), trust him and and switch into CONSISTENT state after deserializing
snapshot as before. OTOH, if he doesn't know it (new slot creation), go via
usual FULL -> CONSISTENT procedure; we might transition into CONSISTENT a bit
later than it became possible, but there is nothing bad about that.

First attached patch implements this. I don't particularly like it, but the only
alternative which I see is to rework slots advancement logic to make
<restart_lsn, confirmed_flush_lsn> pair such that there is always
xl_running_xacts before confirmed_flush_lsn which confirms all xacts running as
of restart_lsn have finished. This unnecessary complexity looks much worse.


As for the check in the topic, I nonetheless propose to remove it completely, as
in second attached patch. Saying for sure whether xact of some record
encountered after snapshot was deserialized can be streamed or not requires to
know nextXid (first not yet running xid) as of this snapshot's lsn -- all xids <
nextXid possibly hadn't been seen in full and are not subject to
decoding. However, generally we don't know nextXid which is taken from
xl_running_xacts; in particular snapshot can be serizalized/deserialized at
XLOG_END_OF_RECOVERY. Changing that for the sake of the check in question is not
worthwhile, so just throw it out instead of trying to relax.

In fact, I don't see what is so important about seeing the top xact first at
all. During COMMIT decoding we'll know all subxids anyway, so why care?


P.S. While digging this, I have noted that return values of
SnapBuildFindSnapshot seem pretty random to me. Basically returning 'true'
performs immediately 4 things:
 - update xmin
 - purge old xip entries
 - advance xmin of the slot
 - if CONSISTENT, advance lsn (earliest serialized snap)

The latter two make sense only after slot created or confirmed_flush_lsn
reached. The first two make sense even immediately after deserializing the
snapshot (because it is serialized *before* updating xmin and xip); generally,
always when committed xids are tracked. Then why cleanup is done when xmin
horizon is too low? Why it is not performed after restoring serialized snapshot?
On the whole, I find this not very important as all these operations are pretty
cheap and harmless if executed too early -- it would be simpler just do them
always.


--
Arseny Sher
Postgres Professional: http://www.postgrespro.com
The Russian Postgres Company


From 86a03671c3bdbaee3dabe02a9a1f8e2e23098652 Mon Sep 17 00:00:00 2001
From: Arseny Sher <sher-ars@yandex.ru>
Date: Tue, 22 Oct 2019 19:02:14 +0300
Subject: [PATCH 1/2] Fix serialized snapshot usage for new logical slots.

Previously, snapbuilder entered into SNAPBUILD_CONSISTENT immediately after
deserializing the snapshot. Generally this is incorrect because
SNAPBUILD_CONSISTENT means not just complete snapshot, but also reorderbuffer
filled with all currently running xacts. This is painless for decoding sessions
with existing slots because they won't stream anything before
confirmed_flush_lsn is reached anyway, at which point all transactions which
hadn't got into reorderbuffer would definitely finish. However, new slots might
be created too early, thus losing (not decoding) parts of transactions committed
after freshly created confirmed_flush_lsn. This can happen under the following
extremely unlucky circumstances:
 - New slot creation reserves point in WAL since which it would read it
   (GetXLogInsertRecPtr);
 - It logs xl_running_xacts to start assembling a snapshot;
 - Running decoding session with another slot quickly reads this
   xl_running_xacts and serializes its snapshot;
 - New slot reads xl_running_xacts and picks this snapshot up, saying that it
   is ready to stream henceforth, though its reorderbuffer is empty.

It turns out, as comment to AllocateSnapshotBuilder explains, that snapbuilder
can't locate *earliest* possible point for streaming on its own. Thus, if
snapbuild users know it (decoding session for existing slot), we trust them and
switch after snapshot deserialization into SNAPBUILD_CONSISTENT as
previously. However, if it is not known (new slot creation), switch into
SNAPBUILD_FULL_SNAPSHOT and wait for all running xacts to finish as usual.
---
 src/backend/replication/logical/logical.c   |   2 +-
 src/backend/replication/logical/snapbuild.c | 141 ++++++++++++++++++++++------
 2 files changed, 115 insertions(+), 28 deletions(-)

diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index da265f5294..67ed52e56f 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -327,7 +327,7 @@ CreateInitDecodingContext(char *plugin,
     ReplicationSlotMarkDirty();
     ReplicationSlotSave();
 
-    ctx = StartupDecodingContext(NIL, restart_lsn, xmin_horizon,
+    ctx = StartupDecodingContext(NIL, InvalidXLogRecPtr, xmin_horizon,
                                  need_full_snapshot, false,
                                  read_page, prepare_write, do_write,
                                  update_progress);
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index 0bd1d0f954..3ed178a4f8 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -165,7 +165,7 @@ struct SnapBuild
 
     /*
      * Don't replay commits from an LSN < this LSN. This can be set externally
-     * but it will also be advanced (never retreat) from within snapbuild.c.
+     * or established by snapbuild.c once consistent snapshot is assembled.
      */
     XLogRecPtr    start_decoding_at;
 
@@ -275,7 +275,7 @@ static void SnapBuildWaitSnapshot(xl_running_xacts *running, TransactionId cutof
 
 /* serialization functions */
 static void SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn);
-static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn);
+static bool SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn, TransactionId nextXid);
 
 /*
  * Return TransactionId after which the next phase of initial snapshot
@@ -309,7 +309,42 @@ SnapBuildStartNextPhaseAt(SnapBuild *builder, TransactionId at)
  * Allocate a new snapshot builder.
  *
  * xmin_horizon is the xid >= which we can be sure no catalog rows have been
- * removed, start_lsn is the LSN >= we want to replay commits.
+ * removed.
+ * start_lsn is InvalidXLogRecPtr or LSN >= we want to replay commits.
+ *  InvalidXLogRecPtr is used during slot creation; snapbuild will assemble
+ *  consistent snapshot and (if continue decoding -- no core code does that
+ *  currently) stream all further commits.
+ *  If normal lsn is passed, caller *must* be sure that WAL reading starts
+ *  early enough to build the snapshot and pick up the first change of
+ *  earliest xact to stream. Slot creation and advancement machinery
+ *  guarantees that slot's <restart_lsn, confirmed_flush_lsn> pair always
+ *  satisfies this.
+ *
+ *    The condition is the caller's responsibility because in some cases
+ *    snapbuilder has no way to enforce this rule without risk of skipping legit
+ *    transactions. Specifically, after constructing (usually deserializing) FULL
+ *    snapshot snapbuilder must iterate over WAL further until all running xacts
+ *    from the snap finish, as we must see with correct snapshots all changes of
+ *    every transaction we are going to stream. However, snapbuild can't
+ *  *immediately* notice this point, because
+ *    - Snapbuilder updates xmin by taking it from xl_running_xacts
+ *      (c.f. SnapBuildProcessRunningXacts). Even if we guarantee that, for
+ *      each possible WAL reading starting position, there is always an an
+ *      xl_running_xacts records logged right before the earliest possible
+ *      streaming point -- IOW, after all xacts which we can't stream had
+ *      finished (which is currently true btw, as slot's advancement is
+ *      considered only at xl_running_xacts) -- that would not be enough due
+ *      to races around xl_running_xacts, i.e with WAL like
+ *      [ <T1> <restart_lsn> <T1 commit> <confirmed_flush_lsn, xrx> <T2 commit> ]
+ *      T2 might be skipped if T1 is shown as running in xl_running_xacts.
+ *      - Tracking xmin manually by recoding commits is not only inefficient,
+ *      it just not feasible because serialized snapshot is not full: it
+ *      contains only committed catalog-modifying xacts. Thus, we can't
+ *      distinguish non-catalog-modifying xact committed before serialized
+ *      snapshot from not yet committed one.
+ *  So, trust the caller: once given start_lsn is reached, it means we must
+ *  have reached the point where all further xacts can be streamed, FULL ->
+ *  CONSISTENT transition.
  */
 SnapBuild *
 AllocateSnapshotBuilder(ReorderBuffer *reorder,
@@ -408,7 +443,8 @@ SnapBuildCurrentState(SnapBuild *builder)
 bool
 SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
 {
-    return ptr < builder->start_decoding_at;
+    return XLogRecPtrIsInvalid(builder->start_decoding_at) ||
+        ptr < builder->start_decoding_at;
 }
 
 /*
@@ -945,16 +981,16 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
          TransactionIdPrecedes(xid, SnapBuildNextPhaseAt(builder))))
     {
         /* ensure that only commits after this are getting replayed */
-        if (builder->start_decoding_at <= lsn)
-            builder->start_decoding_at = lsn + 1;
+        Assert(XLogRecPtrIsInvalid(builder->start_decoding_at)    ||
+               builder->start_decoding_at > lsn);
         return;
     }
 
     if (builder->state < SNAPBUILD_CONSISTENT)
     {
         /* ensure that only commits after this are getting replayed */
-        if (builder->start_decoding_at <= lsn)
-            builder->start_decoding_at = lsn + 1;
+        Assert(XLogRecPtrIsInvalid(builder->start_decoding_at)    ||
+               builder->start_decoding_at > lsn);
 
         /*
          * If building an exportable snapshot, force xid to be tracked, even
@@ -966,6 +1002,16 @@ SnapBuildCommitTxn(SnapBuild *builder, XLogRecPtr lsn, TransactionId xid,
         }
     }
 
+    if (!XLogRecPtrIsInvalid(builder->start_decoding_at) &&
+        builder->start_decoding_at <= lsn)
+    {
+        /*
+         * We are going to stream this xact, so must already have fine
+         * snapshot.
+         */
+        Assert(builder->state == SNAPBUILD_CONSISTENT);
+    }
+
     for (nxact = 0; nxact < nsubxacts; nxact++)
     {
         TransactionId subxid = subxacts[nxact];
@@ -1250,10 +1296,10 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
      */
     if (running->oldestRunningXid == running->nextXid)
     {
-        if (builder->start_decoding_at == InvalidXLogRecPtr ||
-            builder->start_decoding_at <= lsn)
+        if (XLogRecPtrIsInvalid(builder->start_decoding_at))
             /* can decode everything after this */
             builder->start_decoding_at = lsn + 1;
+        Assert(builder->start_decoding_at >= lsn);
 
         /* As no transactions were running xmin/xmax can be trivially set. */
         builder->xmin = running->nextXid;    /* < are finished */
@@ -1275,9 +1321,9 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
     }
     /* b) valid on disk state and not building full snapshot */
     else if (!builder->building_full_snapshot &&
-             SnapBuildRestore(builder, lsn))
+             SnapBuildRestore(builder, lsn, running->nextXid))
     {
-        /* there won't be any state to cleanup */
+        /* there won't be much state to cleanup */
         return false;
     }
 
@@ -1358,6 +1404,10 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
     {
         builder->state = SNAPBUILD_CONSISTENT;
         SnapBuildStartNextPhaseAt(builder, InvalidTransactionId);
+        if (XLogRecPtrIsInvalid(builder->start_decoding_at))
+            /* can decode everything after this */
+            builder->start_decoding_at = lsn + 1;
+        Assert(builder->start_decoding_at >= lsn);
 
         ereport(LOG,
                 (errmsg("logical decoding found consistent point at %X/%X",
@@ -1471,8 +1521,8 @@ typedef struct SnapBuildOnDisk
 void
 SnapBuildSerializationPoint(SnapBuild *builder, XLogRecPtr lsn)
 {
-    if (builder->state < SNAPBUILD_CONSISTENT)
-        SnapBuildRestore(builder, lsn);
+    if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
+        SnapBuildRestore(builder, lsn, InvalidTransactionId);
     else
         SnapBuildSerialize(builder, lsn);
 }
@@ -1499,10 +1549,12 @@ SnapBuildSerialize(SnapBuild *builder, XLogRecPtr lsn)
            builder->last_serialized_snapshot <= lsn);
 
     /*
-     * no point in serializing if we cannot continue to work immediately after
-     * restoring the snapshot
+     * No point in serializing if the snapshot is not complete.
+     * However, FULL snapshot is just as good as CONSISTENT; difference
+     * between these states is not snapshot property, but whether we have
+     * filled reorderbuffer with all currently running xacts.
      */
-    if (builder->state < SNAPBUILD_CONSISTENT)
+    if (builder->state < SNAPBUILD_FULL_SNAPSHOT)
         return;
 
     /*
@@ -1688,10 +1740,15 @@ out:
 
 /*
  * Restore a snapshot into 'builder' if previously one has been stored at the
- * location indicated by 'lsn'. Returns true if successful, false otherwise.
+ * location indicated by 'lsn'.
+ * nextXid is first not yet running xid as of this lsn or InvalidTransactionId;
+ *   unless external code knows where it is safe to start streaming, we can
+ *   use serialized snapshot only if we are aware which xids finish we must
+ *   wait to be able to stream all further commits.
+ *  Returns true if successful, false otherwise.
  */
 static bool
-SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
+SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn, TransactionId nextXid)
 {
     SnapBuildOnDisk ondisk;
     int            fd;
@@ -1701,7 +1758,7 @@ SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
     pg_crc32c    checksum;
 
     /* no point in loading a snapshot if we're already there */
-    if (builder->state == SNAPBUILD_CONSISTENT)
+    if (builder->state >= SNAPBUILD_FULL_SNAPSHOT)
         return false;
 
     sprintf(path, "pg_logical/snapshots/%X-%X.snap",
@@ -1884,11 +1941,47 @@ SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
     if (TransactionIdPrecedes(ondisk.builder.xmin, builder->initial_xmin_horizon))
         goto snapshot_not_interesting;
 
+    /*
+     * Don't use snapshot if external code doesn't know where it is safe to
+     * start streaming and we have no idea which xids to wait for.
+     */
+    if (XLogRecPtrIsInvalid(builder->start_decoding_at) &&
+        !TransactionIdIsValid(nextXid))
+        goto snapshot_not_interesting;
 
     /* ok, we think the snapshot is sensible, copy over everything important */
     builder->xmin = ondisk.builder.xmin;
     builder->xmax = ondisk.builder.xmax;
-    builder->state = ondisk.builder.state;
+    Assert(ondisk.builder.state >= SNAPBUILD_FULL_SNAPSHOT);
+
+    if (XLogRecPtrIsInvalid(builder->start_decoding_at))
+    {
+        /*
+         * Snapshot is fine, now we need to wait till we see all further
+         * commits since the xact's first record.
+         */
+        builder->state = SNAPBUILD_FULL_SNAPSHOT;
+        SnapBuildStartNextPhaseAt(builder, nextXid);
+    }
+    else
+    {
+        /*
+         * If external code (c.f. AllocateSnapshotBuilder) knows we would pick
+         * up all xacts in full before start_decoding_at, just go directly
+         * into CONSISTENT. Though we probably can't stream right now (as we
+         * haven't seen beginnings of some xacts), no xact will be streamed
+         * before start_decoding_at, and we can't be sure to switch into
+         * CONSISTENT later in time anyway.
+         */
+        builder->state = SNAPBUILD_CONSISTENT;
+        SnapBuildStartNextPhaseAt(builder, InvalidTransactionId);
+
+        ereport(LOG,
+                (errmsg("logical decoding found consistent point at %X/%X",
+                        (uint32) (lsn >> 32), (uint32) lsn),
+                 errdetail("Logical decoding will begin using saved snapshot.")));
+    }
+
 
     builder->committed.xcnt = ondisk.builder.committed.xcnt;
     /* We only allocated/stored xcnt, not xcnt_space xids ! */
@@ -1911,12 +2004,6 @@ SnapBuildRestore(SnapBuild *builder, XLogRecPtr lsn)
 
     ReorderBufferSetRestartPoint(builder->reorder, lsn);
 
-    Assert(builder->state == SNAPBUILD_CONSISTENT);
-
-    ereport(LOG,
-            (errmsg("logical decoding found consistent point at %X/%X",
-                    (uint32) (lsn >> 32), (uint32) lsn),
-             errdetail("Logical decoding will begin using saved snapshot.")));
     return true;
 
 snapshot_not_interesting:
-- 
2.11.0

From ab798a087d99ea8d7d0df4c296fee5787cf5394e Mon Sep 17 00:00:00 2001
From: Arseny Sher <sher-ars@yandex.ru>
Date: Wed, 23 Oct 2019 15:56:46 +0300
Subject: [PATCH 2/2] Stop demanding that top xact must be seen before subxact
 in decoding.

Manifested as
ERROR:  subtransaction logged without previous top-level txn record
, this check forbids legit behaviours like
 - First xl_xact_assignment record is beyond reading, i.e. earlier
   restart_lsn.
 - After restart_lsn there is some change of a subxact.
 - After that, there is second xl_xact_assignment (for another subxact)
   revealing relationship between top and first subxact.

Such transaction won't be streamed anyway because we hadn't seen it in full;
confirmed_flush_lsn must be past all these records. Saying for sure whether xact
of some record encountered after snapshot was deserialized can be streamed or
not requires to know nextXid (first not yet running xid) as of this snapshot's
lsn -- all xids < nextXid possibly hadn't been seen in full and are not subject
to decoding. However, generally we don't know nextXid which is taken from
xl_running_xacts; in particular snapshot can be serizalized/deserialized at
XLOG_END_OF_RECOVERY. Changing that for the sake of the check in question is not
worthwhile, so just throw it out instead of trying to relax.
---
 src/backend/replication/logical/reorderbuffer.c | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 8ce28ad629..6faba6077e 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -778,9 +778,6 @@ ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid,
     txn = ReorderBufferTXNByXid(rb, xid, true, &new_top, lsn, true);
     subtxn = ReorderBufferTXNByXid(rb, subxid, true, &new_sub, lsn, false);
 
-    if (new_top && !new_sub)
-        elog(ERROR, "subtransaction logged without previous top-level txn record");
-
     if (!new_sub)
     {
         if (subtxn->is_known_as_subxact)
-- 
2.11.0