From efd65bde8e1e41f1645448a17d4a10e71c20db3e Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy Date: Sat, 19 Sep 2026 06:04:37 +0000 Subject: [PATCH v19] 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. When one of them throws an error, the slot is normally released by the top-level error handler (the sigsetjmp block in PostgresMain()). But PL/pgSQL, PL/Perl, PL/Python and PL/Tcl run the statements they protect in an internal subtransaction, and when a matching handler catches the error there, it is never re-thrown and the top-level handler is never reached. Nothing on the (sub)transaction abort path releases the slot either, so it stays acquired. The next slot operation in the session then hits an assertion failure, or in a non-assertion build silently overwrites MyReplicationSlot. The leaked slot keeps holding back WAL removal and the catalog xmin, and can no longer be acquired. Fix this by recording the subtransaction that acquires the slot and releasing the slot when that subtransaction aborts. We cannot simply release the slot on every subtransaction abort, because logical decoding starts and aborts an internal (sub)transaction for each decoded transaction while holding the slot. Unlike the top-level handler, this path only releases the slot and doesn't drop the session's temporary slots. An error caught in a subtransaction is normally meant to be handled so that the session carries on, so we leave them in place, which is also what happened before this fix. Update the documentation accordingly. Note that a slot whose creation fails partway is still left behind holding resources, but that is a pre-existing problem not specific to subtransactions. Reported-by: SATYANARAYANA NARLAPURAM Author: Bharath Rupireddy Suggested-by: Masahiko Sawada Reviewed-by: Fujii Masao Reviewed-by: shveta malik Reviewed-by: Hou Zhijie Reviewed-by: Kyotaro Horiguchi Reviewed-by: Ashutosh Sharma Reviewed-by: Amit Kapila Reviewed-by: Chao Li Reviewed-by: kedar anavardekar Discussion: https://postgr.es/m/CAHg+QDeuf9tCq3ce=kgFMJP0m=PZC+wi6B=yS+7V0vNXjLS31w@mail.gmail.com Backpatch-through: 14 --- contrib/test_decoding/expected/slot.out | 89 +++++++++++++++++++++++++ contrib/test_decoding/sql/slot.sql | 46 +++++++++++++ doc/src/sgml/catalogs.sgml | 5 +- doc/src/sgml/func.sgml | 8 ++- doc/src/sgml/protocol.sgml | 5 +- src/backend/access/transam/xact.c | 2 + src/backend/replication/slot.c | 67 +++++++++++++++++++ src/backend/tcop/postgres.c | 5 +- src/include/replication/slot.h | 1 + 9 files changed, 223 insertions(+), 5 deletions(-) diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out index 217c45eae96..2debff28ec5 100644 --- a/contrib/test_decoding/expected/slot.out +++ b/contrib/test_decoding/expected/slot.out @@ -388,3 +388,92 @@ SELECT pg_drop_replication_slot('copied_slot2_notemp'); (1 row) +-- +-- Test that a replication slot gets released when the subtransaction holding +-- it aborts on an error caught by a PL/pgSQL exception block. +-- +SELECT 'init' FROM pg_create_logical_replication_slot('regress_subxact_slot', 'test_decoding'); + ?column? +---------- + init +(1 row) + +-- An error caught by a PL/pgSQL exception block releases the slot. +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 active FROM pg_replication_slots WHERE slot_name = 'regress_subxact_slot'; + active +-------- + f +(1 row) + +SELECT count(*) >= 0 AS peek_ok + FROM pg_logical_slot_peek_changes('regress_subxact_slot', NULL, NULL); + peek_ok +--------- + t +(1 row) + +-- An error not caught by a PL/pgSQL exception block releases the slot too. +-- 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 active FROM pg_replication_slots WHERE slot_name = 'regress_subxact_slot'; + active +-------- + f +(1 row) + +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) + +-- An error caught by a PL/pgSQL exception block releases the session's +-- temporary slot, but does not drop it (unlike a top-level 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(*) = 1 AS temp_slot_kept + FROM pg_replication_slots WHERE slot_name = 'regress_subxact_temp_slot'; + temp_slot_kept +---------------- + t +(1 row) + +SELECT pg_drop_replication_slot('regress_subxact_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 0a1398726ee..5d161f765be 100644 --- a/contrib/test_decoding/sql/slot.sql +++ b/contrib/test_decoding/sql/slot.sql @@ -173,3 +173,49 @@ ORDER BY o.slot_name, c.slot_name; SELECT pg_drop_replication_slot('orig_slot2'); SELECT pg_drop_replication_slot('copied_slot2_no_change'); SELECT pg_drop_replication_slot('copied_slot2_notemp'); + +-- +-- Test that a replication slot gets released when the subtransaction holding +-- it aborts on an error caught by a PL/pgSQL exception block. +-- + +SELECT 'init' FROM pg_create_logical_replication_slot('regress_subxact_slot', 'test_decoding'); + +-- An error caught by a PL/pgSQL exception block releases the slot. +DO $$ +BEGIN + PERFORM pg_replication_slot_advance('regress_subxact_slot', '0/1'); +EXCEPTION WHEN OTHERS THEN + RAISE NOTICE 'caught SQLSTATE %', SQLSTATE; +END $$; +SELECT active FROM pg_replication_slots WHERE slot_name = 'regress_subxact_slot'; +SELECT count(*) >= 0 AS peek_ok + FROM pg_logical_slot_peek_changes('regress_subxact_slot', NULL, NULL); + +-- An error not caught by a PL/pgSQL exception block releases the slot too. +-- 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 active FROM pg_replication_slots WHERE slot_name = 'regress_subxact_slot'; +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'); + +-- An error caught by a PL/pgSQL exception block releases the session's +-- temporary slot, but does not drop it (unlike a top-level 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(*) = 1 AS temp_slot_kept + FROM pg_replication_slots WHERE slot_name = 'regress_subxact_temp_slot'; +SELECT pg_drop_replication_slot('regress_subxact_temp_slot'); diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml index 8256b6f3e64..7d022cdc336 100644 --- a/doc/src/sgml/catalogs.sgml +++ b/doc/src/sgml/catalogs.sgml @@ -11488,7 +11488,10 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx True if this is a temporary replication slot. Temporary slots are not saved to disk and are automatically dropped on error or when - the session has finished. + the session has finished. An error raised and caught in a + subtransaction, for example by a + PL/pgSQL exception block, does not drop + them. diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml index d1bf8e32fcf..4f1208529d6 100644 --- a/doc/src/sgml/func.sgml +++ b/doc/src/sgml/func.sgml @@ -25942,7 +25942,9 @@ postgres=# SELECT * FROM pg_walfile_name_offset(pg_stop_backup()); parameter, temporary, when set to true, specifies that the slot should not be permanently stored to disk and is only meant for use by the current session. Temporary slots are also - released upon any error. This function corresponds + dropped when an error is reported. An error caught inside a + subtransaction, for example by a PL/pgSQL + exception block, does not drop them. This function corresponds to the replication protocol command CREATE_REPLICATION_SLOT ... PHYSICAL. @@ -25980,7 +25982,9 @@ postgres=# SELECT * FROM pg_walfile_name_offset(pg_stop_backup()); parameter, temporary, when set to true, specifies that the slot should not be permanently stored to disk and is only meant for use by the current session. Temporary slots are also - released upon any error. The optional fourth parameter, + dropped when an error is reported. An error caught inside a + subtransaction, for example by a PL/pgSQL + exception block, does not drop them. The optional fourth parameter, twophase, when set to true, specifies that the decoding of prepared transactions is enabled for this slot. A call to this function has the same effect as the replication diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml index 5e8ed769c47..41181ff8b60 100644 --- a/doc/src/sgml/protocol.sgml +++ b/doc/src/sgml/protocol.sgml @@ -2048,7 +2048,10 @@ The commands accepted in replication mode are: Specify that this replication slot is a temporary one. Temporary slots are not saved to disk and are automatically dropped on error - or when the session has finished. + or when the session has finished. An error raised and caught in a + subtransaction, for example by a + PL/pgSQL exception block, does not drop + them. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index ef0fe7a5361..93daeb2d1cd 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -4918,6 +4918,7 @@ CommitSubTransaction(void) s->parent->curTransactionOwner); AtEOSubXact_LargeObject(true, s->subTransactionId, s->parent->subTransactionId); + AtEOSubXact_ReplicationSlot(true, s->subTransactionId); AtSubCommit_Notify(); CallSubXactCallbacks(SUBXACT_EVENT_COMMIT_SUB, s->subTransactionId, @@ -5086,6 +5087,7 @@ AbortSubTransaction(void) s->parent->curTransactionOwner); AtEOSubXact_LargeObject(false, s->subTransactionId, s->parent->subTransactionId); + AtEOSubXact_ReplicationSlot(false, s->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 ef2f8787eea..c87deeefad3 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 "common/string.h" #include "miscadmin.h" @@ -49,6 +50,7 @@ #include "storage/proc.h" #include "storage/procarray.h" #include "utils/builtins.h" +#include "utils/snapmgr.h" /* * Replication slot on-disk data structure. @@ -95,6 +97,13 @@ ReplicationSlotCtlData *ReplicationSlotCtl = NULL; /* My backend's replication slot in the shared memory array */ ReplicationSlot *MyReplicationSlot = NULL; +/* + * Subxact 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 MyReplicationSlotSubId = InvalidSubTransactionId; + /* GUCs */ int max_replication_slots = 0; /* the maximum number of replication * slots */ @@ -345,6 +354,7 @@ ReplicationSlotCreate(const char *name, bool db_specific, slot->active_pid = MyProcPid; SpinLockRelease(&slot->mutex); MyReplicationSlot = slot; + MyReplicationSlotSubId = GetCurrentSubTransactionId(); LWLockRelease(ReplicationSlotControlLock); @@ -485,6 +495,7 @@ retry: /* We made this slot active, so it's ours now. */ MyReplicationSlot = s; + MyReplicationSlotSubId = GetCurrentSubTransactionId(); } /* @@ -546,6 +557,61 @@ ReplicationSlotRelease(void) MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING; ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags; LWLockRelease(ProcArrayLock); + + /* The slot is no longer acquired in any subxact. */ + MyReplicationSlotSubId = InvalidSubTransactionId; +} + +/* + * At subxact end, release the replication slot if the subtransaction + * where the slot was acquired is aborted. + */ +void +AtEOSubXact_ReplicationSlot(bool isCommit, SubTransactionId mySubid) +{ + /* Nothing to do unless the slot was acquired in this subxact. */ + if (MyReplicationSlotSubId != mySubid) + return; + + /* + * The subxact that acquired the slot is committing with the slot still + * held. No slot function in the core code does that today, though an + * extension might. A slot left unreleased by mistake cannot be told apart + * from one the caller means to release later. Leave it to the caller. + */ + if (isCommit) + { + MyReplicationSlotSubId = InvalidSubTransactionId; + 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, so the slot is + * still held and must be released. MyReplicationSlotSubId is set only + * when a slot is held and cleared when it is released, so a matching + * subxact id means the slot is ours. + * + * We only release the slot here and do not drop the session's temporary + * slots, unlike the top-level error handler in PostgresMain(). An error + * caught within a subtransaction, for example by a PL/pgSQL exception + * block, is normally meant to be handled so the session carries on, + * unlike a top-level error, so a temporary slot is left in place. That + * matches the temporary slot behavior that predates this callback and is + * simpler to reason about; the slot lives on until the session ends or a + * top-level error occurs, as documented. + */ + ReplicationSlotRelease(); } /* @@ -611,6 +677,7 @@ ReplicationSlotDropAcquired(void) /* slot isn't acquired anymore */ MyReplicationSlot = NULL; + MyReplicationSlotSubId = InvalidSubTransactionId; ReplicationSlotDropPtr(slot); } diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index 0e64b04bd16..46eab4a05bb 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -4348,7 +4348,10 @@ PostgresMain(int argc, char *argv[], * need to be able to start and abort transactions while having a slot * acquired. But we never need to hold them across top level errors, * so releasing here is fine. There's another cleanup in ProcKill() - * ensuring we'll correctly cleanup on FATAL errors as well. + * ensuring we'll correctly cleanup on FATAL errors as well, and + * AtEOSubXact_ReplicationSlot() takes care of an error caught in a + * subtransaction. The latter only releases the slot and, unlike here, + * does not drop the session's temporary slots. */ if (MyReplicationSlot != NULL) ReplicationSlotRelease(); diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 30b0a03caee..f5b30c241f0 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -203,6 +203,7 @@ extern void ReplicationSlotDrop(const char *name, bool nowait); extern void ReplicationSlotAcquire(const char *name, bool nowait); extern void ReplicationSlotRelease(void); extern void ReplicationSlotCleanup(void); +extern void AtEOSubXact_ReplicationSlot(bool isCommit, SubTransactionId mySubid); extern void ReplicationSlotSave(void); extern void ReplicationSlotMarkDirty(void); -- 2.47.3