public inbox for [email protected]
help / color / mirror / Atom feedFrom: SATYANARAYANA NARLAPURAM <[email protected]>
To: shveta malik <[email protected]>
Cc: vignesh C <[email protected]>
Cc: Fujii Masao <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Cc: shveta malik <[email protected]>
Subject: Re: [PATCH] Release replication slot on error in SQL-callable slot functions
Date: Mon, 25 May 2026 12:01:37 -0700
Message-ID: <CAHg+QDf5PVyFgesBNs1GvOnuk_khoXifo96A7QW1EJ8zhhBxyw@mail.gmail.com> (raw)
In-Reply-To: <CAJpy0uBShUF_xm0=BVWivpWHt-4zs__k_3wL1RRjpi0Av8nsog@mail.gmail.com>
References: <CAHg+QDeuf9tCq3ce=kgFMJP0m=PZC+wi6B=yS+7V0vNXjLS31w@mail.gmail.com>
<CAHGQGwFZaWj8DctXuhWQZwSqi631=NKzQJyDV4yqT1Qapt8MFQ@mail.gmail.com>
<CALDaNm1Jjun=by60V-4EpLZe4pAKy0qVZ7ptyHGVCuDyKfo2xQ@mail.gmail.com>
<CAHg+QDcu2x0mjkBSqRxP_8EQ6UmpuX_jMgdKLDkAL1=N6wzZCQ@mail.gmail.com>
<CAJpy0uCmW_NUZN8mw26onvfoFzH_oMrFSKhLUhz896nDgf8c7Q@mail.gmail.com>
<CAHg+QDcf_9prAX=TaSO3UUiCLVD53bEw-KLqzAEXi+ud7h+Z4w@mail.gmail.com>
<CAJpy0uBShUF_xm0=BVWivpWHt-4zs__k_3wL1RRjpi0Av8nsog@mail.gmail.com>
Hi,
On Mon, May 25, 2026 at 2:58 AM shveta malik <[email protected]> wrote:
> On Mon, May 25, 2026 at 12:42 PM SATYANARAYANA NARLAPURAM
> <[email protected]> wrote:
> >
> > Hi
> >
> > On Fri, May 22, 2026 at 2:16 AM shveta malik <[email protected]>
> wrote:
> >>
> >> Thanks for reporting the issue. I could reproduce the same issue with
> >> all these as well:
> >>
> >> pg_logical_slot_peek_changes
> >> pg_logical_slot_get_binary_changes
> >> pg_logical_slot_peek_binary_changes
> >
> >
> > Please find the attached v2 patch that addressed these three cases as
> well.
> >
>
> Thank You for addressuing these cases. A few comments:
>
> 1)
>
> +-- Test 2: session remains usable after the error (MyReplicationSlot
> cleared)
>
> It shoudl be part of 'Test 1' itself and thus should not be named as 'Test
> 2'
>
> 2)
> --------
> +-- Test 4: copy_replication_slot with max_replication_slots exceeded.
> +-- We reduce max_replication_slots artificially by filling all remaining
> slots.
> +-- Instead, trigger an error by copying to an already-existing name.
> +DO $$
> +BEGIN
> + PERFORM pg_copy_logical_replication_slot('regression_slot_t3',
> 'regression_slot_t3');
> +EXCEPTION WHEN OTHERS THEN
> + RAISE NOTICE 'caught: %', SQLERRM;
> +END;
> +$$;
> +-- The original slot must still exist and be usable
> +SELECT count(*) = 1 AS orig_slot_ok FROM pg_replication_slots
> + WHERE slot_name = 'regression_slot_t3';
> -----------
>
> I don't think we can hit the Assert with above test (at-least I could
> not). Since creation of slot itself will fail as the slot with
> same-name already exists, MyReplicationSlot will never be set and thus
> Assert will not be hit. A better testcase will be below which fails
> during LoadOutputPlugin() after slot-creation and MyReplicationSlot is
> set already.
>
> SELECT pg_create_logical_replication_slot('src_slot', 'test_decoding');
>
> DO $$
> BEGIN
> PERFORM pg_copy_logical_replication_slot('src_slot', 'dst_slot',
> false, 'nonexistent_plugin');
> EXCEPTION WHEN others THEN
> RAISE NOTICE 'caught: %', SQLERRM;
> END $$;
>
> SELECT count(*) FROM pg_logical_slot_get_changes('src_slot', NULL, NULL);
>
> 3)
> So overall these are the problematic APIs:
>
> pg_create_logical_replication_slot
> pg_replication_slot_advance
> pg_copy_logical_replication_slot
> pg_logical_slot_peek_binary_changes
> pg_logical_slot_peek_changes
> pg_logical_slot_get_changes
> pg_logical_slot_get_binary_changes
>
> First 3 are are mutually exclusive fixes fow which we have added
> testcases. Last 4 are addressed by fixing common function
> pg_logical_slot_get_changes_guts(). I think we should add a test case
> for at-least any one of these APIs to cover
> pg_logical_slot_get_changes_guts().
>
Thanks for reviewing. Please review the attached v3 patch.
Thanks,
Satya
Attachments:
[application/octet-stream] v3-0001-Release-replication-slot-on-error-in-slot-SQL-functions.patch (15.7K, 3-v3-0001-Release-replication-slot-on-error-in-slot-SQL-functions.patch)
download | inline diff:
diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out
index 7de03c79f6f..962a4277b75 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)
+--
+-- Test that replication slots are properly released or dropped on error,
+-- even when the error is caught by a PL/pgSQL EXCEPTION handler (which
+-- doesn't terminate the session).
+--
+-- pg_create_logical_replication_slot: error during slot creation should
+-- drop the slot.
+DO $$
+BEGIN
+ PERFORM pg_create_logical_replication_slot('regression_slot_error', 'nonexistent_plugin_xyz', true);
+EXCEPTION WHEN OTHERS THEN
+ RAISE NOTICE 'caught: %', SQLERRM;
+END;
+$$;
+NOTICE: caught: could not access file "nonexistent_plugin_xyz": No such file or directory
+SELECT count(*) = 0 AS slot_was_dropped FROM pg_replication_slots
+ WHERE slot_name = 'regression_slot_error';
+ slot_was_dropped
+------------------
+ t
+(1 row)
+
+SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_t3', 'test_decoding', true);
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT count(*) = 1 AS slot_exists FROM pg_replication_slots
+ WHERE slot_name = 'regression_slot_t3';
+ slot_exists
+-------------
+ t
+(1 row)
+
+-- pg_replication_slot_advance: error after acquiring the slot should
+-- release it so the session stays usable.
+SELECT slot_name FROM pg_replication_slot_advance('regression_slot_t3', pg_current_wal_lsn());
+ slot_name
+--------------------
+ regression_slot_t3
+(1 row)
+
+DO $$
+BEGIN
+ PERFORM pg_replication_slot_advance('regression_slot_t3', '0/1');
+EXCEPTION WHEN OTHERS THEN
+ RAISE NOTICE 'caught expected error';
+END;
+$$;
+NOTICE: caught expected error
+-- the session is still healthy
+SELECT slot_name FROM pg_replication_slot_advance('regression_slot_t3', pg_current_wal_lsn());
+ slot_name
+--------------------
+ regression_slot_t3
+(1 row)
+
+-- pg_copy_logical_replication_slot: error after creating the destination
+-- slot should drop it.
+DO $$
+BEGIN
+ PERFORM pg_copy_logical_replication_slot('regression_slot_t3', 'regression_slot_dst', false, 'nonexistent_plugin_xyz');
+EXCEPTION WHEN OTHERS THEN
+ RAISE NOTICE 'caught: %', SQLERRM;
+END;
+$$;
+NOTICE: caught: could not access file "nonexistent_plugin_xyz": No such file or directory
+-- the destination slot must not exist (it was dropped on error)
+SELECT count(*) = 0 AS dst_slot_dropped FROM pg_replication_slots
+ WHERE slot_name = 'regression_slot_dst';
+ dst_slot_dropped
+------------------
+ t
+(1 row)
+
+-- the session is still usable
+SELECT count(*) >= 0 AS changes_ok FROM pg_logical_slot_get_changes('regression_slot_t3', NULL, NULL);
+ changes_ok
+------------
+ t
+(1 row)
+
+-- pg_logical_slot_get_changes: error after acquiring the slot should
+-- release it.
+SELECT 'init' FROM pg_create_physical_replication_slot('regression_slot_phy', true);
+ ?column?
+----------
+ init
+(1 row)
+
+DO $$
+BEGIN
+ PERFORM pg_logical_slot_get_changes('regression_slot_phy', NULL, NULL);
+EXCEPTION WHEN OTHERS THEN
+ RAISE NOTICE 'caught: %', SQLERRM;
+END;
+$$;
+NOTICE: caught: cannot use physical replication slot for logical decoding
+-- the session is still usable
+SELECT slot_name FROM pg_replication_slot_advance('regression_slot_t3', pg_current_wal_lsn());
+ slot_name
+--------------------
+ regression_slot_t3
+(1 row)
+
+SELECT pg_drop_replication_slot('regression_slot_phy');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+-- cleanup
+SELECT pg_drop_replication_slot('regression_slot_t3');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
diff --git a/contrib/test_decoding/sql/slot.sql b/contrib/test_decoding/sql/slot.sql
index 580e3ae3bef..1b327f518aa 100644
--- a/contrib/test_decoding/sql/slot.sql
+++ b/contrib/test_decoding/sql/slot.sql
@@ -190,3 +190,70 @@ 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');
+
+--
+-- Test that replication slots are properly released or dropped on error,
+-- even when the error is caught by a PL/pgSQL EXCEPTION handler (which
+-- doesn't terminate the session).
+--
+
+-- pg_create_logical_replication_slot: error during slot creation should
+-- drop the slot.
+DO $$
+BEGIN
+ PERFORM pg_create_logical_replication_slot('regression_slot_error', 'nonexistent_plugin_xyz', true);
+EXCEPTION WHEN OTHERS THEN
+ RAISE NOTICE 'caught: %', SQLERRM;
+END;
+$$;
+SELECT count(*) = 0 AS slot_was_dropped FROM pg_replication_slots
+ WHERE slot_name = 'regression_slot_error';
+
+SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot_t3', 'test_decoding', true);
+SELECT count(*) = 1 AS slot_exists FROM pg_replication_slots
+ WHERE slot_name = 'regression_slot_t3';
+
+-- pg_replication_slot_advance: error after acquiring the slot should
+-- release it so the session stays usable.
+SELECT slot_name FROM pg_replication_slot_advance('regression_slot_t3', pg_current_wal_lsn());
+DO $$
+BEGIN
+ PERFORM pg_replication_slot_advance('regression_slot_t3', '0/1');
+EXCEPTION WHEN OTHERS THEN
+ RAISE NOTICE 'caught expected error';
+END;
+$$;
+-- the session is still healthy
+SELECT slot_name FROM pg_replication_slot_advance('regression_slot_t3', pg_current_wal_lsn());
+
+-- pg_copy_logical_replication_slot: error after creating the destination
+-- slot should drop it.
+DO $$
+BEGIN
+ PERFORM pg_copy_logical_replication_slot('regression_slot_t3', 'regression_slot_dst', false, 'nonexistent_plugin_xyz');
+EXCEPTION WHEN OTHERS THEN
+ RAISE NOTICE 'caught: %', SQLERRM;
+END;
+$$;
+-- the destination slot must not exist (it was dropped on error)
+SELECT count(*) = 0 AS dst_slot_dropped FROM pg_replication_slots
+ WHERE slot_name = 'regression_slot_dst';
+-- the session is still usable
+SELECT count(*) >= 0 AS changes_ok FROM pg_logical_slot_get_changes('regression_slot_t3', NULL, NULL);
+
+-- pg_logical_slot_get_changes: error after acquiring the slot should
+-- release it.
+SELECT 'init' FROM pg_create_physical_replication_slot('regression_slot_phy', true);
+DO $$
+BEGIN
+ PERFORM pg_logical_slot_get_changes('regression_slot_phy', NULL, NULL);
+EXCEPTION WHEN OTHERS THEN
+ RAISE NOTICE 'caught: %', SQLERRM;
+END;
+$$;
+-- the session is still usable
+SELECT slot_name FROM pg_replication_slot_advance('regression_slot_t3', pg_current_wal_lsn());
+SELECT pg_drop_replication_slot('regression_slot_phy');
+
+-- cleanup
+SELECT pg_drop_replication_slot('regression_slot_t3');
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 71fbaf72269..aa56e90bfab 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -197,10 +197,10 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
else
end_of_wal = GetXLogReplayRecPtr(NULL);
- ReplicationSlotAcquire(NameStr(*name), true, true);
-
PG_TRY();
{
+ ReplicationSlotAcquire(NameStr(*name), true, true);
+
/* restart at slot's confirmed_flush */
ctx = CreateDecodingContext(InvalidXLogRecPtr,
options,
@@ -320,6 +320,9 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
/* clear all timetravel entries */
InvalidateSystemCaches();
+ if (MyReplicationSlot != NULL)
+ ReplicationSlotRelease();
+
PG_RE_THROW();
}
PG_END_TRY();
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 16fbd383735..acc643ac749 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -148,38 +148,55 @@ create_logical_replication_slot(char *name, char *plugin,
temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase,
false, failover, false);
- /*
- * Ensure the logical decoding is enabled before initializing the logical
- * decoding context.
- */
- EnsureLogicalDecodingEnabled();
- Assert(IsLogicalDecodingEnabled());
+ PG_TRY();
+ {
+ /*
+ * Ensure the logical decoding is enabled before initializing the logical
+ * decoding context.
+ */
+ EnsureLogicalDecodingEnabled();
+ Assert(IsLogicalDecodingEnabled());
- /*
- * Create logical decoding context to find start point or, if we don't
- * need it, to 1) bump slot's restart_lsn and xmin 2) check plugin sanity.
- *
- * Note: when !find_startpoint this is still important, because it's at
- * this point that the output plugin is validated.
- */
- ctx = CreateInitDecodingContext(plugin, NIL,
- false, /* just catalogs is OK */
- false, /* not repack */
- restart_lsn,
- XL_ROUTINE(.page_read = read_local_xlog_page,
- .segment_open = wal_segment_open,
- .segment_close = wal_segment_close),
- NULL, NULL, NULL);
+ /*
+ * Create logical decoding context to find start point or, if we don't
+ * need it, to 1) bump slot's restart_lsn and xmin 2) check plugin sanity.
+ *
+ * Note: when !find_startpoint this is still important, because it's at
+ * this point that the output plugin is validated.
+ */
+ ctx = CreateInitDecodingContext(plugin, NIL,
+ false, /* just catalogs is OK */
+ false, /* not repack */
+ restart_lsn,
+ XL_ROUTINE(.page_read = read_local_xlog_page,
+ .segment_open = wal_segment_open,
+ .segment_close = wal_segment_close),
+ NULL, NULL, NULL);
- /*
- * If caller needs us to determine the decoding start point, do so now.
- * This might take a while.
- */
- if (find_startpoint)
- DecodingContextFindStartpoint(ctx);
+ /*
+ * If caller needs us to determine the decoding start point, do so now.
+ * This might take a while.
+ */
+ if (find_startpoint)
+ DecodingContextFindStartpoint(ctx);
- /* don't need the decoding context anymore */
- FreeDecodingContext(ctx);
+ /* don't need the decoding context anymore */
+ FreeDecodingContext(ctx);
+ }
+ PG_CATCH();
+ {
+ /*
+ * Drop the slot on error. ReplicationSlotRelease() only drops
+ * RS_EPHEMERAL slots, so for RS_TEMPORARY slots we must explicitly
+ * call ReplicationSlotDropAcquired() to avoid leaving the slot
+ * behind (e.g. when the error is caught by a PL/pgSQL EXCEPTION
+ * handler that doesn't terminate the session).
+ */
+ if (MyReplicationSlot != NULL)
+ ReplicationSlotDropAcquired();
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
}
/*
@@ -566,49 +583,58 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
else
moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
- /* Acquire the slot so we "own" it */
- ReplicationSlotAcquire(NameStr(*slotname), true, true);
-
- /* A slot whose restart_lsn has never been reserved cannot be advanced */
- if (!XLogRecPtrIsValid(MyReplicationSlot->data.restart_lsn))
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("replication slot \"%s\" cannot be advanced",
- NameStr(*slotname)),
- errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));
+ PG_TRY();
+ {
+ /* Acquire the slot so we "own" it */
+ ReplicationSlotAcquire(NameStr(*slotname), true, true);
+ /* A slot whose restart_lsn has never been reserved cannot be advanced */
+ if (!XLogRecPtrIsValid(MyReplicationSlot->data.restart_lsn))
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("replication slot \"%s\" cannot be advanced",
+ NameStr(*slotname)),
+ errdetail("This slot has never previously reserved WAL, or it has been invalidated.")));
- /*
- * Check if the slot is not moving backwards. Physical slots rely simply
- * on restart_lsn as a minimum point, while logical slots have confirmed
- * consumption up to confirmed_flush, meaning that in both cases data
- * older than that is not available anymore.
- */
- if (OidIsValid(MyReplicationSlot->data.database))
- minlsn = MyReplicationSlot->data.confirmed_flush;
- else
- minlsn = MyReplicationSlot->data.restart_lsn;
+ /*
+ * Check if the slot is not moving backwards. Physical slots rely simply
+ * on restart_lsn as a minimum point, while logical slots have confirmed
+ * consumption up to confirmed_flush, meaning that in both cases data
+ * older than that is not available anymore.
+ */
+ if (OidIsValid(MyReplicationSlot->data.database))
+ minlsn = MyReplicationSlot->data.confirmed_flush;
+ else
+ minlsn = MyReplicationSlot->data.restart_lsn;
- if (moveto < minlsn)
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("cannot advance replication slot to %X/%08X, minimum is %X/%08X",
- LSN_FORMAT_ARGS(moveto), LSN_FORMAT_ARGS(minlsn))));
+ if (moveto < minlsn)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot advance replication slot to %X/%08X, minimum is %X/%08X",
+ LSN_FORMAT_ARGS(moveto), LSN_FORMAT_ARGS(minlsn))));
- /* Do the actual slot update, depending on the slot type */
- if (OidIsValid(MyReplicationSlot->data.database))
- endlsn = pg_logical_replication_slot_advance(moveto);
- else
- endlsn = pg_physical_replication_slot_advance(moveto);
+ /* Do the actual slot update, depending on the slot type */
+ if (OidIsValid(MyReplicationSlot->data.database))
+ endlsn = pg_logical_replication_slot_advance(moveto);
+ else
+ endlsn = pg_physical_replication_slot_advance(moveto);
- values[0] = NameGetDatum(&MyReplicationSlot->data.name);
- nulls[0] = false;
+ values[0] = NameGetDatum(&MyReplicationSlot->data.name);
+ nulls[0] = false;
- /*
- * Recompute the minimum LSN and xmin across all slots to adjust with the
- * advancing potentially done.
- */
- ReplicationSlotsComputeRequiredXmin(false);
- ReplicationSlotsComputeRequiredLSN();
+ /*
+ * Recompute the minimum LSN and xmin across all slots to adjust with the
+ * advancing potentially done.
+ */
+ ReplicationSlotsComputeRequiredXmin(false);
+ ReplicationSlotsComputeRequiredLSN();
+ }
+ PG_CATCH();
+ {
+ if (MyReplicationSlot != NULL)
+ ReplicationSlotRelease();
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
ReplicationSlotRelease();
@@ -763,7 +789,13 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
/*
* Update the destination slot to current values of the source slot;
* recheck that the source slot is still the one we saw previously.
+ *
+ * Use PG_TRY to ensure we drop the destination slot if any validation
+ * error occurs. Without this, an error caught by a PL/pgSQL EXCEPTION
+ * handler would leave MyReplicationSlot set, crashing on the next slot
+ * operation.
*/
+ PG_TRY();
{
TransactionId copy_effective_xmin;
TransactionId copy_effective_catalog_xmin;
@@ -797,9 +829,6 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
* or the restart_lsn either is invalid or has gone backward. (The
* restart_lsn could go backwards if the source slot is dropped and
* copied from an older slot during installation.)
- *
- * Since erroring out will release and drop the destination slot we
- * don't need to release it here.
*/
if (copy_restart_lsn < src_restart_lsn ||
src_islogical != copy_islogical ||
@@ -857,6 +886,18 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
}
#endif
}
+ PG_CATCH();
+ {
+ /*
+ * Drop the newly-created destination slot on error. Same as in
+ * create_logical_replication_slot(): use ReplicationSlotDropAcquired()
+ * to handle both RS_EPHEMERAL and RS_TEMPORARY slots.
+ */
+ if (MyReplicationSlot != NULL)
+ ReplicationSlotDropAcquired();
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
/* target slot fully created, mark as persistent if needed */
if (logical_slot && !temporary)
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: [PATCH] Release replication slot on error in SQL-callable slot functions
In-Reply-To: <CAHg+QDf5PVyFgesBNs1GvOnuk_khoXifo96A7QW1EJ8zhhBxyw@mail.gmail.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox