From 12ccfc67ef85e4d5f36076cd0d182a01ce4708b4 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
Date: Thu, 6 Aug 2026 19:01:55 +0000
Subject: [PATCH v11] Fix replication slot leak on error caught in a
 subtransaction.

The SQL-callable replication slot functions, such as
pg_replication_slot_advance(), acquire a slot and release it
before returning. If one throws an error in between, the slot is
released by the top-level error handler (the sigsetjmp block in
PostgresMain()). But PL/pgSQL, PL/Perl, PL/Python and PL/Tcl run
an error-handling block in an internal subtransaction: when an
error is raised there, they abort the subtransaction and, if a
handler matches, catch the error without re-throwing it. The
top-level handler is then never reached, and nothing in the
(sub)transaction abort path releases the slot either.

The slot is therefore left acquired after the caught error. The
next slot operation in the session then fails an assertion, or in
a non-assertion build silently overwrites the reference and leaks
the slot, which keeps holding back WAL removal and the catalog
xmin (blocking vacuum) and can no longer be acquired.

An error that does reach the top-level handler is unaffected.
That is the case for a plain ROLLBACK TO SAVEPOINT, or for a
PL/pgSQL exception handler whose conditions do not match the
error, which then re-throws it.

Fix by recording the subtransaction that acquires the slot and
releasing the slot when that subtransaction aborts, the same way
other subtransaction-scoped resources are released, for example
in AtEOSubXact_LargeObject() and AtEOSubXact_Files(). The slot
cannot simply be released on every subtransaction abort, because
logical decoding starts and aborts an internal transaction or
subtransaction for each decoded transaction while holding the
slot. Those internal aborts are below the acquiring
subtransaction and so are left alone.

Releasing a temporary slot does not drop it, so it would keep
holding resources until the session ends. To avoid that, when
releasing the held slot on abort also drop the session's
temporary slots, the same way the top-level error handler already
does.

Reported-by: Satyanarayana Narlapuram <satyanarlapuram@gmail.com>
Author: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
Co-authored-by: Satyanarayana Narlapuram <satyanarlapuram@gmail.com>
Co-authored-by: Masahiko Sawada <sawada.mshk@gmail.com>
Reviewed-by: Fujii Masao <masao.fujii@gmail.com>
Reviewed-by: vignesh C <vignesh21@gmail.com>
Reviewed-by: shveta malik <shveta.malik@gmail.com>
Reviewed-by: Hou Zhijie <houzj.fnst@fujitsu.com>
Reviewed-by: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Reviewed-by: Ashutosh Sharma <ashu.coek88@gmail.com>
Discussion: https://postgr.es/m/CAHg+QDeuf9tCq3ce=kgFMJP0m=PZC+wi6B=yS+7V0vNXjLS31w@mail.gmail.com
Backpatch-through: 14
---
 contrib/test_decoding/expected/slot.out | 119 ++++++++++++++++++++++++
 contrib/test_decoding/sql/slot.sql      |  65 +++++++++++++
 src/backend/access/transam/xact.c       |   4 +
 src/backend/replication/slot.c          |  93 ++++++++++++++++++
 src/include/replication/slot.h          |   2 +
 5 files changed, 283 insertions(+)

diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out
index 7de03c79f6f..d42b2c4f480 100644
--- a/contrib/test_decoding/expected/slot.out
+++ b/contrib/test_decoding/expected/slot.out
@@ -466,3 +466,122 @@ SELECT pg_drop_replication_slot('physical_slot');
  
 (1 row)
 
+-- A slot function that errors out must still release the slot, otherwise the
+-- next slot operation in the session fails an assertion or leaves the slot
+-- behind. Advancing a freshly created slot to a low LSN always errors.
+-- Test 1: a block with an EXCEPTION clause runs its statements in a subxact
+-- and catches an error raised there. The slot must be released so the next
+-- slot operation works.
+SELECT 'init' FROM pg_create_logical_replication_slot('regress_subxact_slot', 'test_decoding');
+ ?column? 
+----------
+ init
+(1 row)
+
+DO $$
+BEGIN
+    PERFORM pg_replication_slot_advance('regress_subxact_slot', '0/1');
+EXCEPTION WHEN OTHERS THEN
+    RAISE NOTICE 'caught SQLSTATE %', SQLSTATE;
+END $$;
+NOTICE:  caught SQLSTATE 55000
+SELECT count(*) >= 0 AS peek_ok
+    FROM pg_logical_slot_peek_changes('regress_subxact_slot', NULL, NULL);
+ peek_ok 
+---------
+ t
+(1 row)
+
+-- Test 2: the EXCEPTION clause does not match, so the error is not caught. The
+-- slot is still released, so the next slot operation works. Show only the
+-- SQLSTATE for stable output.
+\set VERBOSITY sqlstate
+DO $$
+BEGIN
+    PERFORM pg_replication_slot_advance('regress_subxact_slot', '0/1');
+EXCEPTION WHEN division_by_zero THEN
+    RAISE NOTICE 'unreachable';
+END $$;
+ERROR:  55000
+\set VERBOSITY default
+SELECT count(*) >= 0 AS peek_ok
+    FROM pg_logical_slot_peek_changes('regress_subxact_slot', NULL, NULL);
+ peek_ok 
+---------
+ t
+(1 row)
+
+SELECT pg_drop_replication_slot('regress_subxact_slot');
+ pg_drop_replication_slot 
+--------------------------
+ 
+(1 row)
+
+-- Test 3: same as Test 1 for a temporary slot, which is dropped rather than
+-- just released, so it is not left behind after the error.
+SELECT 'init' FROM pg_create_logical_replication_slot('regress_subxact_temp_slot', 'test_decoding', true);
+ ?column? 
+----------
+ init
+(1 row)
+
+DO $$
+BEGIN
+    PERFORM pg_replication_slot_advance('regress_subxact_temp_slot', '0/1');
+EXCEPTION WHEN OTHERS THEN
+    RAISE NOTICE 'caught SQLSTATE %', SQLSTATE;
+END $$;
+NOTICE:  caught SQLSTATE 55000
+SELECT count(*) = 0 AS temp_slot_dropped
+    FROM pg_replication_slots WHERE slot_name = 'regress_subxact_temp_slot';
+ temp_slot_dropped 
+-------------------
+ t
+(1 row)
+
+-- Test 4: a top-level error always drops the session's temporary slots, even
+-- one unrelated to the error and never acquired here.
+SELECT 'init' FROM pg_create_logical_replication_slot('regress_toplevel_temp_slot', 'test_decoding', true);
+ ?column? 
+----------
+ init
+(1 row)
+
+SELECT 1 / 0;
+ERROR:  division by zero
+SELECT count(*) = 0 AS temp_slot_dropped
+    FROM pg_replication_slots WHERE slot_name = 'regress_toplevel_temp_slot';
+ temp_slot_dropped 
+-------------------
+ t
+(1 row)
+
+-- Test 5: an error raised and caught in a subxact that held no slot does
+-- not drop the session's temporary slots, unlike a top-level error, which
+-- always does.
+SELECT 'init' FROM pg_create_logical_replication_slot('regress_kept_temp_slot', 'test_decoding', true);
+ ?column? 
+----------
+ init
+(1 row)
+
+DO $$
+BEGIN
+    PERFORM 1 / 0;
+EXCEPTION WHEN division_by_zero THEN
+    RAISE NOTICE 'caught SQLSTATE %', SQLSTATE;
+END $$;
+NOTICE:  caught SQLSTATE 22012
+SELECT count(*) = 1 AS temp_slot_kept
+    FROM pg_replication_slots WHERE slot_name = 'regress_kept_temp_slot';
+ temp_slot_kept 
+----------------
+ t
+(1 row)
+
+SELECT pg_drop_replication_slot('regress_kept_temp_slot');
+ pg_drop_replication_slot 
+--------------------------
+ 
+(1 row)
+
diff --git a/contrib/test_decoding/sql/slot.sql b/contrib/test_decoding/sql/slot.sql
index 580e3ae3bef..b5b745f80c2 100644
--- a/contrib/test_decoding/sql/slot.sql
+++ b/contrib/test_decoding/sql/slot.sql
@@ -190,3 +190,68 @@ SELECT pg_drop_replication_slot('failover_true_slot');
 SELECT pg_drop_replication_slot('failover_false_slot');
 SELECT pg_drop_replication_slot('failover_default_slot');
 SELECT pg_drop_replication_slot('physical_slot');
+
+-- A slot function that errors out must still release the slot, otherwise the
+-- next slot operation in the session fails an assertion or leaves the slot
+-- behind. Advancing a freshly created slot to a low LSN always errors.
+
+-- Test 1: a block with an EXCEPTION clause runs its statements in a subxact
+-- and catches an error raised there. The slot must be released so the next
+-- slot operation works.
+SELECT 'init' FROM pg_create_logical_replication_slot('regress_subxact_slot', 'test_decoding');
+DO $$
+BEGIN
+    PERFORM pg_replication_slot_advance('regress_subxact_slot', '0/1');
+EXCEPTION WHEN OTHERS THEN
+    RAISE NOTICE 'caught SQLSTATE %', SQLSTATE;
+END $$;
+SELECT count(*) >= 0 AS peek_ok
+    FROM pg_logical_slot_peek_changes('regress_subxact_slot', NULL, NULL);
+
+-- Test 2: the EXCEPTION clause does not match, so the error is not caught. The
+-- slot is still released, so the next slot operation works. Show only the
+-- SQLSTATE for stable output.
+\set VERBOSITY sqlstate
+DO $$
+BEGIN
+    PERFORM pg_replication_slot_advance('regress_subxact_slot', '0/1');
+EXCEPTION WHEN division_by_zero THEN
+    RAISE NOTICE 'unreachable';
+END $$;
+\set VERBOSITY default
+SELECT count(*) >= 0 AS peek_ok
+    FROM pg_logical_slot_peek_changes('regress_subxact_slot', NULL, NULL);
+SELECT pg_drop_replication_slot('regress_subxact_slot');
+
+-- Test 3: same as Test 1 for a temporary slot, which is dropped rather than
+-- just released, so it is not left behind after the error.
+SELECT 'init' FROM pg_create_logical_replication_slot('regress_subxact_temp_slot', 'test_decoding', true);
+DO $$
+BEGIN
+    PERFORM pg_replication_slot_advance('regress_subxact_temp_slot', '0/1');
+EXCEPTION WHEN OTHERS THEN
+    RAISE NOTICE 'caught SQLSTATE %', SQLSTATE;
+END $$;
+SELECT count(*) = 0 AS temp_slot_dropped
+    FROM pg_replication_slots WHERE slot_name = 'regress_subxact_temp_slot';
+
+-- Test 4: a top-level error always drops the session's temporary slots, even
+-- one unrelated to the error and never acquired here.
+SELECT 'init' FROM pg_create_logical_replication_slot('regress_toplevel_temp_slot', 'test_decoding', true);
+SELECT 1 / 0;
+SELECT count(*) = 0 AS temp_slot_dropped
+    FROM pg_replication_slots WHERE slot_name = 'regress_toplevel_temp_slot';
+
+-- Test 5: an error raised and caught in a subxact that held no slot does
+-- not drop the session's temporary slots, unlike a top-level error, which
+-- always does.
+SELECT 'init' FROM pg_create_logical_replication_slot('regress_kept_temp_slot', 'test_decoding', true);
+DO $$
+BEGIN
+    PERFORM 1 / 0;
+EXCEPTION WHEN division_by_zero THEN
+    RAISE NOTICE 'caught SQLSTATE %', SQLSTATE;
+END $$;
+SELECT count(*) = 1 AS temp_slot_kept
+    FROM pg_replication_slots WHERE slot_name = 'regress_kept_temp_slot';
+SELECT pg_drop_replication_slot('regress_kept_temp_slot');
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 3a89149016f..d143ba6c3a8 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -5204,6 +5204,8 @@ CommitSubTransaction(void)
 						s->parent->curTransactionOwner);
 	AtEOSubXact_LargeObject(true, s->subTransactionId,
 							s->parent->subTransactionId);
+	AtEOSubXact_ReplicationSlot(true, s->subTransactionId,
+								s->parent->subTransactionId);
 	AtSubCommit_Notify();
 
 	CallSubXactCallbacks(SUBXACT_EVENT_COMMIT_SUB, s->subTransactionId,
@@ -5380,6 +5382,8 @@ AbortSubTransaction(void)
 						   s->parent->curTransactionOwner);
 		AtEOSubXact_LargeObject(false, s->subTransactionId,
 								s->parent->subTransactionId);
+		AtEOSubXact_ReplicationSlot(false, s->subTransactionId,
+									s->parent->subTransactionId);
 		AtSubAbort_Notify();
 
 		/* Advertise the fact that we aborted in pg_xact. */
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 1a0ff682068..5d77d24c5f1 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -59,6 +59,7 @@
 #include "utils/builtins.h"
 #include "utils/guc_hooks.h"
 #include "utils/injection_point.h"
+#include "utils/snapmgr.h"
 #include "utils/varlena.h"
 #include "utils/wait_event.h"
 
@@ -157,6 +158,13 @@ const ShmemCallbacks ReplicationSlotsShmemCallbacks = {
 /* My backend's replication slot in the shared memory array */
 ReplicationSlot *MyReplicationSlot = NULL;
 
+/*
+ * Subtransaction that acquired MyReplicationSlot, or invalid if none is held
+ * or it was acquired with no transaction in progress (as a walsender does).
+ * Used to release the slot when that subxact aborts.
+ */
+static SubTransactionId acquiredInSubId = InvalidSubTransactionId;
+
 /* GUC variables */
 int			max_replication_slots = 10; /* the maximum number of replication
 										 * slots */
@@ -519,6 +527,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
 	slot->active_proc = MyProcNumber;
 	SpinLockRelease(&slot->mutex);
 	MyReplicationSlot = slot;
+	acquiredInSubId = GetCurrentSubTransactionId();
 
 	LWLockRelease(ReplicationSlotControlLock);
 
@@ -724,6 +733,7 @@ retry:
 
 	/* We made this slot active, so it's ours now. */
 	MyReplicationSlot = s;
+	acquiredInSubId = GetCurrentSubTransactionId();
 
 	/*
 	 * We need to check for invalidation after making the slot ours to avoid
@@ -848,6 +858,88 @@ ReplicationSlotRelease(void)
 
 		pfree(slotname);
 	}
+
+	/* The slot is no longer acquired in any subxact. */
+	acquiredInSubId = InvalidSubTransactionId;
+}
+
+/*
+ * Release the replication slot at subxact end if it was acquired here.
+ *
+ * A slot function acquires a slot and releases it before returning. On error
+ * the top-level error handler releases it. But PL/pgSQL, PL/Perl, PL/Python and
+ * PL/Tcl run an error-handling block in an internal subxact, and when an error
+ * there is caught the top-level handler is never reached, so the slot would
+ * otherwise stay acquired. Release it when the subxact that acquired it aborts,
+ * the same way AtEOSubXact_LargeObject() and other subxact-scoped resources are
+ * handled. The subxact id is used rather than a nesting level because levels
+ * are reused across subxacts while ids are not.
+ */
+void
+AtEOSubXact_ReplicationSlot(bool isCommit, SubTransactionId mySubid,
+							SubTransactionId parentSubid)
+{
+	/* Nothing to do unless the slot was acquired in this subxact. */
+	if (acquiredInSubId != mySubid)
+		return;
+
+	/*
+	 * On commit, hand the slot to the parent subxact. A slot held across a
+	 * subxact commit (acquired but not yet released by its caller) would
+	 * otherwise be attributed to a gone subxact, and we still want it
+	 * released if the parent, or an ancestor, later aborts.
+	 *
+	 * This acts as a safety check against a slot function that misses
+	 * releasing the slot before returning. Today none do, so the check above
+	 * returns early and we never get here with a slot still held. Handling it
+	 * anyway, the same way AtEOSubXact_LargeObject() and AtEOSubXact_Files()
+	 * do, keeps such a slot from being leaked.
+	 */
+	if (isCommit)
+	{
+		acquiredInSubId = parentSubid;
+		return;
+	}
+
+	/*
+	 * We must not get here while decoding is running. Decoding starts and
+	 * aborts an internal (sub)transaction while holding the slot, for each
+	 * decoded transaction (ReorderBufferProcessTXN()) and when executing
+	 * invalidations (ReorderBufferImmediateInvalidation()). However, those
+	 * subtransactions are always nested below the one that acquired the slot,
+	 * so their subtransaction ids are deeper and do not match here. Decoding
+	 * also runs with a historic snapshot set up, so assert that it is not.
+	 */
+	Assert(!HistoricSnapshotActive());
+
+	/*
+	 * The aborting subxact is the one that acquired the slot, and its id is
+	 * never invalid, so acquiredInSubId is valid here. It is set only when a
+	 * slot is held, and cleared when the slot is released, so the slot must
+	 * still be held.
+	 */
+	Assert(acquiredInSubId != InvalidSubTransactionId);
+	Assert(MyReplicationSlot != NULL);
+	ReplicationSlotRelease();
+
+	/*
+	 * Also drop this session's temporary slots, as the top-level error
+	 * handler does. Otherwise a temporary slot could be left behind holding
+	 * back WAL removal and the catalog xmin after an error.
+	 *
+	 * Note that this only runs when the aborting subxact held a slot. A
+	 * caught error that held no slot (for example an unrelated error caught
+	 * by a PL/pgSQL EXCEPTION clause) does not drop the session's temporary
+	 * slots, unlike a top-level error, which always does. Covering that too
+	 * would mean running the cleanup on every aborting subxact, which is
+	 * harder to reason about, so it is kept here after the release.
+	 *
+	 * Note also that we could instead keep the temporary slots and treat the
+	 * error as recoverable, since the subxact was caught and the session goes
+	 * on. But the error may be in the slot handling itself, leaving the slot
+	 * in a doubtful state, so dropping it is the safer choice.
+	 */
+	ReplicationSlotCleanup(false);
 }
 
 /*
@@ -1042,6 +1134,7 @@ ReplicationSlotDropAcquired(bool try_disable)
 
 	/* slot isn't acquired anymore */
 	MyReplicationSlot = NULL;
+	acquiredInSubId = InvalidSubTransactionId;
 
 	ReplicationSlotDropPtr(slot);
 
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 9b29444cbca..31f8755dc39 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -343,6 +343,8 @@ extern void ReplicationSlotAcquire(const char *name, bool nowait,
 								   bool error_if_invalid);
 extern void ReplicationSlotRelease(void);
 extern void ReplicationSlotCleanup(bool synced_only);
+extern void AtEOSubXact_ReplicationSlot(bool isCommit, SubTransactionId mySubid,
+										SubTransactionId parentSubid);
 extern void ReplicationSlotSave(void);
 extern void ReplicationSlotMarkDirty(void);
 
-- 
2.47.3

