From 423f51e23b7a242e03eef0f5ff04179fcc458cae Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Thu, 30 Jul 2026 06:29:15 +0000 Subject: [PATCH v7] 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 of them throws an error in between, the slot is released by the top-level error handler (the sigsetjmp block in PostgresMain()). But when such a function is called from a PL/pgSQL, PL/Perl, PL/Python or PL/Tcl function with exception handling, the error is caught in an internal subtransaction and is not re-thrown, so it never reaches that handler. Nothing in the (sub)transaction abort path releases the slot either. The slot is therefore left acquired after the caught error, and the next slot operation in the session fails an assertion, or in a non-assertion build silently overwrites the reference and leaves the slot behind, which then keeps holding back WAL removal and the catalog xmin (blocking vacuum) and can no longer be acquired. An error that instead reaches the top-level handler is not affected, which is the case for a plain ROLLBACK TO SAVEPOINT, or a PL/pgSQL EXCEPTION block whose condition does not match the error and thus re-throws it. Fix by recording the subtransaction that acquires the slot and releasing the slot when that subtransaction aborts, in 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 Author: Bharath Rupireddy Co-authored-by: Satyanarayana Narlapuram Co-authored-by: Masahiko Sawada Reviewed-by: Fujii Masao Reviewed-by: vignesh C Reviewed-by: shveta malik Reviewed-by: Hou Zhijie Reviewed-by: Kyotaro Horiguchi Discussion: https://postgr.es/m/CAHg+QDeuf9tCq3ce=kgFMJP0m=PZC+wi6B=yS+7V0vNXjLS31w@mail.gmail.com Backpatch-through: 14 --- contrib/test_decoding/expected/slot.out | 120 ++++++++++++++++++++++++ contrib/test_decoding/sql/slot.sql | 66 +++++++++++++ src/backend/access/transam/xact.c | 5 + src/backend/replication/slot.c | 86 +++++++++++++++++ src/include/replication/slot.h | 2 + 5 files changed, 279 insertions(+) diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out index 7de03c79f6f..6221b30ac98 100644 --- a/contrib/test_decoding/expected/slot.out +++ b/contrib/test_decoding/expected/slot.out @@ -466,3 +466,123 @@ 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: error caught in an EXCEPTION block, which uses a subxact +-- internally. 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 handler does not match the error, so it is re-thrown and reaches +-- the top-level handler, which releases the slot. 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. Releasing a temporary slot on +-- error does not drop it, so it would keep holding back WAL removal and the +-- catalog xmin. The session's temporary slots are dropped as well, so none is +-- left behind. +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 caught in a subxact that held no slot does not drop +-- the session's temporary slots. This differs from a top-level error (see the +-- XXX in AtEOSubXact_ReplicationSlot()). +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..0a380382725 100644 --- a/contrib/test_decoding/sql/slot.sql +++ b/contrib/test_decoding/sql/slot.sql @@ -190,3 +190,69 @@ 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: error caught in an EXCEPTION block, which uses a subxact +-- internally. 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 handler does not match the error, so it is re-thrown and reaches +-- the top-level handler, which releases the slot. 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. Releasing a temporary slot on +-- error does not drop it, so it would keep holding back WAL removal and the +-- catalog xmin. The session's temporary slots are dropped as well, so none is +-- left behind. +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 caught in a subxact that held no slot does not drop +-- the session's temporary slots. This differs from a top-level error (see the +-- XXX in AtEOSubXact_ReplicationSlot()). +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..f658209bd45 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -50,6 +50,7 @@ #include "replication/logicallauncher.h" #include "replication/logicalworker.h" #include "replication/origin.h" +#include "replication/slot.h" #include "replication/snapbuild.h" #include "replication/syncrep.h" #include "storage/aio_subsys.h" @@ -5204,6 +5205,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 +5383,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 c7ab82a28e1..31dec1223da 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -40,6 +40,7 @@ #include #include "access/transam.h" +#include "access/xact.h" #include "access/xlog_internal.h" #include "access/xlogrecovery.h" #include "common/file_utils.h" @@ -59,6 +60,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 +159,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 */ @@ -517,6 +526,7 @@ ReplicationSlotCreate(const char *name, bool db_specific, slot->active_proc = MyProcNumber; SpinLockRelease(&slot->mutex); MyReplicationSlot = slot; + acquiredInSubId = GetCurrentSubTransactionId(); LWLockRelease(ReplicationSlotControlLock); @@ -722,6 +732,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 @@ -846,6 +857,80 @@ 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 an exception block (in PL/pgSQL, + * PL/Perl, PL/Python or PL/Tcl) catches the error in an internal subxact + * without re-throwing, so the top-level handler is not reached and the slot + * stays 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. + */ + 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()), but that always + * happens below the acquiring subxact, so those aborts have a different + * (deeper) id and do not match here. Decoding also runs with a historic + * snapshot set up, so assert that it is not. + */ + Assert(!HistoricSnapshotActive()); + + if (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, including an + * error in slot handling itself. + * + * XXX This only runs when the aborting subxact held a slot. A caught + * error that held no slot (for example an unrelated error in a PL/pgSQL + * EXCEPTION block) does not drop the session's temporary slots, unlike a + * top-level error, which always does. To also cover that, the cleanup + * would have to run at the early exit above, on every aborting subxact. + * But a slot must be released before it can be cleaned up + * (ReplicationSlotCleanup() requires no slot to be held), so that path + * would need its own MyReplicationSlot == NULL guard to skip the case + * where decoding still holds a slot. That is harder to reason about, so + * the cleanup is kept here after the release. + * + * XXX Alternatively we could 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); } /* @@ -1040,6 +1125,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