v3-0001-xact_rollback-decoding-fix.patch

From 69ffad6af7a8bbbabaf3030ca417463a96cfb0c8 Mon Sep 17 00:00:00 2001
From: Nikolay Samokhvalov <nik@postgres.ai>
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)