From 69ffad6af7a8bbbabaf3030ca417463a96cfb0c8 Mon Sep 17 00:00:00 2001 From: Nikolay Samokhvalov Date: Wed, 8 Jul 2026 06:11:24 +0000 Subject: [PATCH v3 1/2] Do not attribute logical decoding aborts to xact_rollback Logical decoding aborts the current transaction after decoding committed transactions to clean up catalog access and other transaction-local state. In a logical walsender this can be a top-level abort, so pg_stat_database.xact_rollback is incremented even though no user-visible transaction rolled back. Keep these internal cleanup aborts out of xact_rollback by adding AbortCurrentTransactionWithoutXactCounters(), a narrow wrapper around AbortCurrentTransaction() that suppresses only the next pg_stat_database xact_commit/xact_rollback counter update while preserving the rest of transaction cleanup. Add a TAP test that fails without the fix: five committed transactions decoded by a subscription produce a publisher xact_rollback delta of 5 when the walsender exits. With the fix, the xact_rollback delta remains 0. The test also verifies that decoded transactions do not affect xact_commit by comparing against a control walsender shutdown with no decoded transactions. Reported-by: Rafael Thofehrn Castro Discussion: https://postgr.es/m/CAM527d_EbU5Li4a5FdKQjYsdF-4Lqr_i3jXmZOm7Wbb%3DQ2KzTw%40mail.gmail.com --- src/backend/access/transam/xact.c | 23 ++++ .../replication/logical/reorderbuffer.c | 18 ++- src/backend/utils/activity/pgstat_xact.c | 32 ++++- src/include/access/xact.h | 1 + src/include/pgstat.h | 2 + src/test/subscription/meson.build | 1 + .../t/039_publisher_xact_rollback.pl | 124 ++++++++++++++++++ 7 files changed, 196 insertions(+), 5 deletions(-) create mode 100644 src/test/subscription/t/039_publisher_xact_rollback.pl diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 3a89149016f..d74b231caf3 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -3512,6 +3512,29 @@ AbortCurrentTransaction(void) } } +/* + * AbortCurrentTransactionWithoutXactCounters + * + * Like AbortCurrentTransaction(), but don't count the abort in + * pg_stat_database.xact_rollback. For internal cleanup aborts that don't + * represent a user-visible transaction outcome. + */ +void +AbortCurrentTransactionWithoutXactCounters(void) +{ + pgstat_suppress_xact_counters(); + AbortCurrentTransaction(); + + /* + * AbortCurrentTransaction() is a no-op when already idle (e.g. an abort at + * TBLOCK_DEFAULT/TRANS_DEFAULT), in which case AtEOXact_PgStat() never runs + * to consume the flag. That can't happen at the current call sites, which + * always abort a live transaction, but clear it unconditionally anyway so + * set and clear stay paired defensively and the flag cannot leak. + */ + pgstat_clear_xact_counter_suppression(); +} + /* * AbortCurrentTransactionInternal - a function doing an iteration of work * regarding handling the current transaction abort. In the case of diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index 059ed860314..49bb87ab1ec 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -2666,9 +2666,14 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, * Aborting the current (sub-)transaction as a whole has the right * semantics. We want all locks acquired in here to be released, not * reassigned to the parent and we do not want any database access - * have persistent effects. + * have persistent effects. In the !using_subtxn case this is a + * top-level abort for internal cleanup; keep it out of + * pg_stat_database.xact_rollback. */ - AbortCurrentTransaction(); + if (using_subtxn) + AbortCurrentTransaction(); + else + AbortCurrentTransactionWithoutXactCounters(); /* make sure there's no cache pollution */ if (rbtxn_distr_inval_overflowed(txn)) @@ -2729,9 +2734,14 @@ ReorderBufferProcessTXN(ReorderBuffer *rb, ReorderBufferTXN *txn, /* * Force cache invalidation to happen outside of a valid transaction - * to prevent catalog access as we just caught an error. + * to prevent catalog access as we just caught an error. As above, + * keep the top-level internal cleanup abort out of + * pg_stat_database.xact_rollback. */ - AbortCurrentTransaction(); + if (using_subtxn) + AbortCurrentTransaction(); + else + AbortCurrentTransactionWithoutXactCounters(); /* make sure there's no cache pollution */ if (rbtxn_distr_inval_overflowed(txn)) diff --git a/src/backend/utils/activity/pgstat_xact.c b/src/backend/utils/activity/pgstat_xact.c index 3e1978775e1..d1be3733e4d 100644 --- a/src/backend/utils/activity/pgstat_xact.c +++ b/src/backend/utils/activity/pgstat_xact.c @@ -31,6 +31,7 @@ static void AtEOSubXact_PgStat_DroppedStats(PgStat_SubXactStatus *xact_state, bool isCommit, int nestDepth); static PgStat_SubXactStatus *pgStatXactStack = NULL; +static bool pgStatSkipXactCounters = false; /* @@ -40,8 +41,15 @@ void AtEOXact_PgStat(bool isCommit, bool parallel) { PgStat_SubXactStatus *xact_state; + bool skip_xact_counters = pgStatSkipXactCounters; - AtEOXact_PgStat_Database(isCommit, parallel); + /* + * Consume the suppression flag. Only the xact_commit/xact_rollback bump is + * skipped; transactional stats below must still be processed. + */ + pgStatSkipXactCounters = false; + if (!skip_xact_counters) + AtEOXact_PgStat_Database(isCommit, parallel); /* handle transactional stats information */ xact_state = pgStatXactStack; @@ -59,6 +67,28 @@ AtEOXact_PgStat(bool isCommit, bool parallel) pgstat_clear_snapshot(); } +/* + * Suppress the xact_commit/xact_rollback bump at the next top-level transaction + * end. For internal aborts that don't represent a user-visible transaction + * outcome; see AbortCurrentTransactionWithoutXactCounters(). + */ +void +pgstat_suppress_xact_counters(void) +{ + Assert(!pgStatSkipXactCounters); + pgStatSkipXactCounters = true; +} + +/* + * Clear the suppression flag set by pgstat_suppress_xact_counters(), in case + * the intervening abort never reached AtEOXact_PgStat() to consume it. + */ +void +pgstat_clear_xact_counter_suppression(void) +{ + pgStatSkipXactCounters = false; +} + /* * When committing, drop stats for objects dropped in the transaction. When * aborting, drop stats for objects created in the transaction. diff --git a/src/include/access/xact.h b/src/include/access/xact.h index a8cbdf247c8..9a22a2e9a67 100644 --- a/src/include/access/xact.h +++ b/src/include/access/xact.h @@ -467,6 +467,7 @@ extern void SaveTransactionCharacteristics(SavedTransactionCharacteristics *s); extern void RestoreTransactionCharacteristics(const SavedTransactionCharacteristics *s); extern void CommitTransactionCommand(void); extern void AbortCurrentTransaction(void); +extern void AbortCurrentTransactionWithoutXactCounters(void); extern void BeginTransactionBlock(void); extern bool EndTransactionBlock(bool chain); extern bool PrepareTransactionBlock(const char *gid); diff --git a/src/include/pgstat.h b/src/include/pgstat.h index 58a44857f13..ff0a2e16467 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -828,6 +828,8 @@ extern PgStat_StatSubEntry *pgstat_fetch_stat_subscription(Oid subid); */ extern void AtEOXact_PgStat(bool isCommit, bool parallel); +extern void pgstat_suppress_xact_counters(void); +extern void pgstat_clear_xact_counter_suppression(void); extern void AtEOSubXact_PgStat(bool isCommit, int nestDepth); extern void AtPrepare_PgStat(void); extern void PostPrepare_PgStat(void); diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build index e71e95c6297..268fa8c3e9c 100644 --- a/src/test/subscription/meson.build +++ b/src/test/subscription/meson.build @@ -48,6 +48,7 @@ tests += { 't/036_sequences.pl', 't/037_except.pl', 't/038_walsnd_shutdown_timeout.pl', + 't/039_publisher_xact_rollback.pl', 't/100_bugs.pl', ], }, diff --git a/src/test/subscription/t/039_publisher_xact_rollback.pl b/src/test/subscription/t/039_publisher_xact_rollback.pl new file mode 100644 index 00000000000..c75634425d0 --- /dev/null +++ b/src/test/subscription/t/039_publisher_xact_rollback.pl @@ -0,0 +1,124 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +# Check that pg_stat_database.xact_rollback on a logical-replication +# publisher is not inflated by the walsender's internal catalog-cleanup +# aborts. ReorderBufferProcessTXN() ends each decoded transaction with +# AbortCurrentTransaction(); in the walsender that is a top-level abort. +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +sub get_xact_stats +{ + my ($node) = @_; + + return split /\|/, $node->safe_psql('template1', q{ + SELECT xact_commit, xact_rollback + FROM pg_stat_database + WHERE datname = 'postgres' + }); +} + +# Wait for this subscription's walsender to disappear from pg_stat_activity. +# +# A walsender registers its stats flush (pgstat_shutdown_hook) with +# before_shmem_exit() and clears its pg_stat_activity entry +# (pgstat_beshutdown_hook) with on_shmem_exit(). shmem_exit() runs all +# before_shmem_exit callbacks before any on_shmem_exit callback, so by the time +# the walsender is gone from pg_stat_activity its xact_commit/xact_rollback +# counts have already been flushed to shared stats. Observing its exit is +# therefore sufficient synchronization for reading the flushed counters. +sub wait_for_walsender_exit +{ + my ($node) = @_; + + $node->poll_query_until( + 'template1', q{ + SELECT count(*) = 0 FROM pg_stat_activity + WHERE backend_type = 'walsender' AND application_name = 's' + }) + or die 's walsender did not exit'; +} + +my $node_publisher = PostgreSQL::Test::Cluster->new('publisher'); +$node_publisher->init(allows_streaming => 'logical'); +# Autovacuum would commit on the postgres database and perturb the xact_commit +# delta this test compares between the control and decoding runs. +$node_publisher->append_conf('postgresql.conf', 'autovacuum = off'); +$node_publisher->start; + +my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber'); +$node_subscriber->init; +$node_subscriber->start; + +$node_publisher->safe_psql('postgres', + 'CREATE TABLE t (id int PRIMARY KEY)'); +$node_subscriber->safe_psql('postgres', + 'CREATE TABLE t (id int PRIMARY KEY)'); + +$node_publisher->safe_psql('postgres', 'CREATE PUBLICATION p FOR TABLE t'); + +my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres'; +$node_subscriber->safe_psql('postgres', + "CREATE SUBSCRIPTION s CONNECTION '$publisher_connstr' PUBLICATION p"); + +$node_subscriber->wait_for_subscription_sync($node_publisher, 's'); + +# Measure a walsender shutdown without any decoded committed transactions. +# The absolute xact_commit delta is not asserted: walsender shutdown performs +# fixed session/transaction bookkeeping that may change independently of this +# test. The control run captures that non-decoding delta so the decoding run +# can assert it is unchanged by decoded transactions. +my ($control_base_commit, $control_base_rollback) = + get_xact_stats($node_publisher); + +$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION s DISABLE'); +wait_for_walsender_exit($node_publisher); + +my ($control_final_commit, $control_final_rollback) = + get_xact_stats($node_publisher); + +my $control_commit_delta = $control_final_commit - $control_base_commit; +my $control_rollback_delta = $control_final_rollback - $control_base_rollback; + +$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION s ENABLE'); + +# Five autocommit INSERTs: each becomes one decoded committed transaction on +# the walsender. Without the fix, those produce five spurious rollbacks after +# DISABLE. xact_commit may change due to fixed walsender bookkeeping, but it +# must not change as a function of decoded transactions. +my $n = 5; +$node_publisher->safe_psql('postgres', + join('', map { "INSERT INTO t VALUES ($_);\n" } 1 .. $n)); + +$node_publisher->wait_for_catchup('s'); + +# Baseline after catchup and before walsender exit, so the deltas measure only +# the stats flushed by this subscription's walsender during shutdown. +my ($base_commit, $base_rollback) = get_xact_stats($node_publisher); + +# Disabling the subscription terminates the walsender; its shutdown hook +# flushes pgstat counters to shared stats before it leaves pg_stat_activity. +$node_subscriber->safe_psql('postgres', 'ALTER SUBSCRIPTION s DISABLE'); +wait_for_walsender_exit($node_publisher); + +my ($final_commit, $final_rollback) = get_xact_stats($node_publisher); + +cmp_ok( + $control_rollback_delta, '==', 0, + 'walsender shutdown without decoded transactions does not inflate publisher xact_rollback' +); + +cmp_ok( + $final_rollback - $base_rollback, '==', 0, + 'walsender does not inflate publisher xact_rollback for decoded transactions' +); + +cmp_ok( + $final_commit - $base_commit, '==', $control_commit_delta, + 'walsender does not change publisher xact_commit as a function of decoded transactions' +); + +done_testing(); -- 2.50.1 (Apple Git-155)