From aa4c52b808f76870f80189757af7217358544d60 Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Tue, 18 Aug 2026 11:56:49 -0700 Subject: [PATCH] Fix stream abort for a transaction that was never streamed. Commit 072ee847ad4 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 072ee847ad4 was introduced. Bug: #19616 Reported-by: Tyler Smart Author: Andrey Rachitskiy Reviewed-by: Hayato Kuroda Reviewed-by: Fujii Masao Reviewed-by: Masahiko Sawada Discussion: https://postgr.es/m/19616-f6153af509910853@postgresql.org Backpatch-through: 18 --- contrib/test_decoding/expected/stream.out | 23 +++++++++++ contrib/test_decoding/sql/stream.sql | 17 +++++++++ .../replication/logical/reorderbuffer.c | 38 ++++++++++++------- 3 files changed, 64 insertions(+), 14 deletions(-) diff --git a/contrib/test_decoding/expected/stream.out b/contrib/test_decoding/expected/stream.out index 9879e02ca84..0ec5c933610 100644 --- a/contrib/test_decoding/expected/stream.out +++ b/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 diff --git a/contrib/test_decoding/sql/stream.sql b/contrib/test_decoding/sql/stream.sql index f1269403e0a..5e45a8e9b64 100644 --- a/contrib/test_decoding/sql/stream.sql +++ b/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'); diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 6aed6346366..6567dd52aed 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/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; }