From ebda7d2b12ce4860ffc1e0d6e091fa14e824c0c3 Mon Sep 17 00:00:00 2001 From: Andrey Rachitskiy Date: Fri, 14 Aug 2026 09:27:00 +0500 Subject: [PATCH v5] Don't mark discarded aborted subxacts as streamed. ReorderBufferTruncateTXN is also used when discarding already-aborted transactions at eviction. Marking every subxact with in-memory changes as streamed in that path made a later abort emit stream_abort to clients that never enabled streaming. Mark a subxact as streamed only when it has changes and its top-level transaction is already marked as streamed. The streaming call sites mark the top-level xact first, so this keeps their behavior. The abort-discard path never marks the top-level xact, so its subxacts are left unmarked. BUG #19616 Reported-by: Tyler Smart Author: Andrey Rachitskiy Reviewed-by: Hayato Kuroda (Fujitsu) Reviewed-by: Fujii Masao Reviewed-by: Masahiko Sawada Discussion: https://www.postgresql.org/message-id/19616-f6153af509910853%40postgresql.org --- contrib/test_decoding/expected/stream.out | 40 +++++++++++++++++++ contrib/test_decoding/sql/stream.sql | 23 +++++++++++ .../replication/logical/reorderbuffer.c | 38 +++++++++++------- 3 files changed, 87 insertions(+), 14 deletions(-) diff --git a/contrib/test_decoding/expected/stream.out b/contrib/test_decoding/expected/stream.out index 9879e02ca84..7a92019933e 100644 --- a/contrib/test_decoding/expected/stream.out +++ b/contrib/test_decoding/expected/stream.out @@ -141,3 +141,43 @@ SELECT pg_drop_replication_slot('regression_slot'); (1 row) +-- bug #19616 +-- pgoutput protocol compatibility could be broken for an aborted xact +-- discarded at eviction while a subxact remained in memory. +-- Stream Abort ('A'), a streaming-only message, could then reach a client +-- that did not enable streaming. +CREATE TABLE stream_abort_test(data text); +CREATE PUBLICATION stream_pub FOR TABLE stream_abort_test; +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_pgoutput', 'pgoutput'); + ?column? +---------- + init +(1 row) + +BEGIN; +SAVEPOINT s; +INSERT INTO stream_abort_test VALUES ('subtransaction-change'); +RELEASE SAVEPOINT s; +INSERT INTO stream_abort_test SELECT repeat('x', 1000) FROM generate_series(1, 5000) g(i); +ROLLBACK; +INSERT INTO stream_abort_test VALUES ('after-abort'); +SELECT chr(get_byte(data, 0)) AS msgtype, count(*) +FROM pg_logical_slot_peek_binary_changes('regression_slot_pgoutput', NULL, NULL, + 'proto_version', '4', 'streaming', 'off', 'publication_names', 'stream_pub') +GROUP BY 1 ORDER BY 1; + msgtype | count +---------+------- + B | 1 + C | 1 + I | 1 + R | 1 +(4 rows) + +SELECT pg_drop_replication_slot('regression_slot_pgoutput'); + pg_drop_replication_slot +-------------------------- + +(1 row) + +DROP PUBLICATION stream_pub; +DROP TABLE stream_abort_test; diff --git a/contrib/test_decoding/sql/stream.sql b/contrib/test_decoding/sql/stream.sql index f1269403e0a..ff40910d313 100644 --- a/contrib/test_decoding/sql/stream.sql +++ b/contrib/test_decoding/sql/stream.sql @@ -67,3 +67,26 @@ RESET debug_logical_replication_streaming; DROP TABLE stream_test; SELECT pg_drop_replication_slot('regression_slot'); + +-- bug #19616 +-- pgoutput protocol compatibility could be broken for an aborted xact +-- discarded at eviction while a subxact remained in memory. +-- Stream Abort ('A'), a streaming-only message, could then reach a client +-- that did not enable streaming. +CREATE TABLE stream_abort_test(data text); +CREATE PUBLICATION stream_pub FOR TABLE stream_abort_test; +SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_pgoutput', 'pgoutput'); +BEGIN; +SAVEPOINT s; +INSERT INTO stream_abort_test VALUES ('subtransaction-change'); +RELEASE SAVEPOINT s; +INSERT INTO stream_abort_test SELECT repeat('x', 1000) FROM generate_series(1, 5000) g(i); +ROLLBACK; +INSERT INTO stream_abort_test VALUES ('after-abort'); +SELECT chr(get_byte(data, 0)) AS msgtype, count(*) +FROM pg_logical_slot_peek_binary_changes('regression_slot_pgoutput', NULL, NULL, + 'proto_version', '4', 'streaming', 'off', 'publication_names', 'stream_pub') +GROUP BY 1 ORDER BY 1; +SELECT pg_drop_replication_slot('regression_slot_pgoutput'); +DROP PUBLICATION stream_pub; +DROP TABLE stream_abort_test; 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)) + { + /* 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 (rbtxn_is_toptxn(txn) || (txn->nentries_mem != 0)) + if (txn->nentries_mem != 0 && rbtxn_is_streamed(rbtxn_get_toptxn(txn))) txn->txn_flags |= RBTXN_IS_STREAMED; } -- 2.53.0