Fix stream abort for a transaction that was never streamed.

Commit 072ee847ad taught logical decoding to discard the changes of a
transaction that is already known to be aborted when it is picked for
eviction. That path reuses ReorderBufferTruncateTXN(), which marks
every subtransaction that still has in-memory changes as
streamed. Since nothing is streamed in that path, and the top-level
transaction is never marked, a subtransaction ends up flagged as
streamed even though the output plugin has never seen it. Decoding the
subsequent abort record then makes ReorderBufferAbort() invoke the
stream_abort callback for that subtransaction.

For pgoutput this sends a Stream Abort ('A') message to a subscriber
that requested streaming = off, and it does so regardless of the
negotiated protocol version, so even a client speaking a version that
predates transaction streaming receives a message it cannot
parse. test_decoding dereferences a NULL pointer and crashes, since it
allocates its per-transaction state in the begin or stream start
callback, neither of which runs for a transaction discarded as
aborted.

This commit fixes this by marking a subtransaction as streamed only
when it has changes and its top-level transaction is already marked as
streamed. All streaming call sites mark the top-level transaction
before truncating it, so their behavior is unchanged, while the
abort-discard path never marks the top-level transaction and therefore
now leaves its subtransactions unmarked.

Backpatch to v18, where commit 072ee847ad was introduced.

Bug: #19616
Reported-by: Tyler Smart <tyler@smarts.io>
Author: Andrey Rachitskiy <pl0h0yp1@gmail.com>
Reviewed-by: Hayato Kuroda <kuroda.hayato@fujitsu.com>
Reviewed-by: Fujii Masao <masao.fujii@gmail.com>
Reviewed-by: Masahiko Sawada <sawada.mshk@gmail.com>
Discussion: https://postgr.es/m/19616-f6153af509910853@postgresql.org
Backpatch-through: 18

contrib/test_decoding/expected/stream.out

@@ -134,6 +134,29 @@ SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL,
 (1 row)
 
 RESET debug_logical_replication_streaming;
+-- bug #19616
+--
+-- An aborted top-level transaction that is discarded at eviction must not
+-- leave its subtransactions marked as streamed.  Otherwise, decoding its
+-- abort record invokes the stream abort callback for a subtransaction the
+-- output plugin has never seen, even though streaming was never requested.
+-- The trailing committed transaction is required to flush the ROLLBACK
+-- record; without it decoding would stop before reaching the abort.
+BEGIN;
+SAVEPOINT s;
+INSERT INTO stream_test VALUES ('subxact-change');
+RELEASE SAVEPOINT s;
+INSERT INTO stream_test SELECT 'toplevel-change' || g.i FROM generate_series(1, 5000) g(i);
+ROLLBACK;
+INSERT INTO stream_test VALUES ('after-abort');
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+                                    data                                    
+----------------------------------------------------------------------------
+ BEGIN
+ table public.stream_test: INSERT: data[text]:'after-abort' i[integer]:null
+ COMMIT
+(3 rows)
+
 DROP TABLE stream_test;
 SELECT pg_drop_replication_slot('regression_slot');
  pg_drop_replication_slot 

contrib/test_decoding/sql/stream.sql

@@ -65,5 +65,22 @@ COMMIT;
 SELECT count(*) FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1', 'stream-changes', '1');
 RESET debug_logical_replication_streaming;
 
+-- bug #19616
+--
+-- An aborted top-level transaction that is discarded at eviction must not
+-- leave its subtransactions marked as streamed.  Otherwise, decoding its
+-- abort record invokes the stream abort callback for a subtransaction the
+-- output plugin has never seen, even though streaming was never requested.
+-- The trailing committed transaction is required to flush the ROLLBACK
+-- record; without it decoding would stop before reaching the abort.
+BEGIN;
+SAVEPOINT s;
+INSERT INTO stream_test VALUES ('subxact-change');
+RELEASE SAVEPOINT s;
+INSERT INTO stream_test SELECT 'toplevel-change' || g.i FROM generate_series(1, 5000) g(i);
+ROLLBACK;
+INSERT INTO stream_test VALUES ('after-abort');
+SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL, 'include-xids', '0', 'skip-empty-xacts', '1');
+
 DROP TABLE stream_test;
 SELECT pg_drop_replication_slot('regression_slot');

src/backend/replication/logical/reorderbuffer.c

@@ -2132,26 +2132,36 @@ ReorderBufferSaveTXNSnapshot(ReorderBuffer *rb, ReorderBufferTXN *txn,
 }
 
 /*
- * Mark the given transaction as streamed if it's a top-level transaction
- * or has changes.
+ * Mark the given transaction as streamed, if appropriate.
+ *
+ * A top-level transaction is always marked.  A subtransaction is marked
+ * only when it has changes and its top-level transaction is already
+ * marked as streamed.
  */
 static void
 ReorderBufferMaybeMarkTXNStreamed(ReorderBuffer *rb, ReorderBufferTXN *txn)
 {
 	/*
-	 * The top-level transaction, is marked as streamed always, even if it
-	 * does not contain any changes (that is, when all the changes are in
-	 * subtransactions).
-	 *
-	 * For subtransactions, we only mark them as streamed when there are
-	 * changes in them.
-	 *
-	 * We do it this way because of aborts - we don't want to send aborts for
-	 * XIDs the downstream is not aware of. And of course, it always knows
-	 * about the top-level xact (we send the XID in all messages), but we
-	 * never stream XIDs of empty subxacts.
+	 * The top-level transaction is marked as streamed always, even if it does
+	 * not contain any changes (that is, when all the changes are in
+	 * subtransactions).  The downstream always knows about it, since we send
+	 * its XID in every message.
 	 */
-	if (rbtxn_is_toptxn(txn) || (txn->nentries_mem != 0))
+	if (rbtxn_is_toptxn(txn))
+	{
+		/* We only reach here when streaming is supported. */
+		Assert(ReorderBufferCanStream(rb));
+		txn->txn_flags |= RBTXN_IS_STREAMED;
+		return;
+	}
+
+	/*
+	 * A subtransaction is marked only when it has changes, and only when its
+	 * top-level transaction has already been marked as streamed.  We never
+	 * stream XIDs of empty subxacts, and we must not send an abort for an XID
+	 * the downstream has never heard of.
+	 */
+	if (txn->nentries_mem != 0 && rbtxn_is_streamed(rbtxn_get_toptxn(txn)))
 		txn->txn_flags |= RBTXN_IS_STREAMED;
 }