public inbox for [email protected]
help / color / mirror / Atom feedFrom: Bharath Rupireddy <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Bertrand Drouvot <[email protected]>
Cc: Nathan Bossart <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: Re: Introduce XID age and inactive timeout based replication slot invalidation
Date: Sun, 24 Mar 2024 15:05:44 +0530
Message-ID: <CALj2ACU6tV0bh_MiFBYU7-VK2cj8Gas73htcQjoOv0uvSEG_HA@mail.gmail.com> (raw)
In-Reply-To: <CAA4eK1L_UP=ZjCj=+cGEauYXHC1icUEfx16p+B7qrAHZuCx5Ew@mail.gmail.com>
References: <CAA4eK1KZyjdn7oXS3d7qJU5ag9ss4xSk0Jkjsh5NOrRdkPkV1g@mail.gmail.com>
<[email protected]>
<CALj2ACUvK7ShPyirAURWf62=qOQqu=NwnyL5CMfVjM5Ody7Oxw@mail.gmail.com>
<[email protected]>
<CALj2ACWk5a-LfBPziAjjFA--syoUSD_4jY6dZyhU8g8XuD-i7g@mail.gmail.com>
<CAA4eK1+46kJF3Z2Tv=xdkyy0nuurrS7RfzqvqY8bq1MF-JLM=A@mail.gmail.com>
<CALj2ACVKncd+JZR8BeXSgPO4akqic9kO99Z6pCfeXQKwmye2Og@mail.gmail.com>
<CAA4eK1JwJdsKMxczGO9tYgEzxp_L2=j0QTeDDk2Tx4-ipGSppw@mail.gmail.com>
<[email protected]>
<CALj2ACWuNHJ3qBMmmy6i1YA9V8JDRSaciL2bBM+72bRcqiBE2g@mail.gmail.com>
<[email protected]>
<CALj2ACVUCo48FXWAcxC9LEV6P=xgd-8O=e9aPmTRUZvvH_L7Zw@mail.gmail.com>
<CAA4eK1LjNB8jBgqLHtf+FKnd=SfMz3TXXs871p7KjG4VHnxafg@mail.gmail.com>
<CALj2ACXQmGx-L073=8CUSOk3hhe=HL_e6dveYeR9WJcEh+7jvg@mail.gmail.com>
<CAA4eK1L_UP=ZjCj=+cGEauYXHC1icUEfx16p+B7qrAHZuCx5Ew@mail.gmail.com>
On Sun, Mar 24, 2024 at 10:40 AM Amit Kapila <[email protected]> wrote:
>
> > For instance, setting last_inactive_time_1 to an invalid value fails
> > with the following error:
> >
> > error running SQL: 'psql:<stdin>:1: ERROR: invalid input syntax for
> > type timestamp with time zone: "foo"
> > LINE 1: SELECT last_inactive_time > 'foo'::timestamptz FROM pg_repli...
> >
>
> It would be found at a later point. It would be probably better to
> verify immediately after the test that fetches the last_inactive_time
> value.
Agree. I've added a few more checks explicitly to verify the
last_inactive_time is sane with the following:
qq[SELECT '$last_inactive_time'::timestamptz > to_timestamp(0)
AND '$last_inactive_time'::timestamptz >
'$slot_creation_time'::timestamptz;]
I've attached the v18 patch set here. I've also addressed earlier
review comments from Amit, Ajin Cherian. Note that I've added new
invalidation mechanism tests in a separate TAP test file just because
I don't want to clutter or bloat any of the existing files and spread
tests for physical slots and logical slots into separate existing TAP
files.
--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
Attachments:
[application/octet-stream] v18-0001-Track-last_inactive_time-in-pg_replication_slots.patch (14.4K, ../CALj2ACU6tV0bh_MiFBYU7-VK2cj8Gas73htcQjoOv0uvSEG_HA@mail.gmail.com/2-v18-0001-Track-last_inactive_time-in-pg_replication_slots.patch)
download | inline diff:
From 79c3967c0dc25ec2741f7fe979b2b97939e2eeb1 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Sun, 24 Mar 2024 07:12:55 +0000
Subject: [PATCH v18 1/5] Track last_inactive_time in pg_replication_slots.
Till now, the time at which the replication slot became inactive
is not tracked directly in pg_replication_slots. This commit adds
a new property called last_inactive_time for this. It is set to 0
whenever a slot is made active/acquired and set to current
timestamp whenever the slot is inactive/released or restored from
the disk.
The new property will be useful on production servers to debug and
analyze inactive replication slots. It will also help to know the
lifetime of a replication slot - one can know how long a streaming
standby, logical subscriber, or replication slot consumer is down.
The new property will be useful to implement inactive timeout based
replication slot invalidation in a future commit.
Author: Bharath Rupireddy
Reviewed-by: Bertrand Drouvot, Amit Kapila
Discussion: https://www.postgresql.org/message-id/CALj2ACW4aUe-_uFQOjdWCEN-xXoLGhmvRFnL8SNw_TZ5nJe+aw@mail.gmail.com
---
doc/src/sgml/system-views.sgml | 11 ++
src/backend/catalog/system_views.sql | 1 +
src/backend/replication/slot.c | 27 ++++
src/backend/replication/slotfuncs.c | 7 +-
src/include/catalog/pg_proc.dat | 6 +-
src/include/replication/slot.h | 3 +
src/test/recovery/t/019_replslot_limit.pl | 148 ++++++++++++++++++++++
src/test/regress/expected/rules.out | 3 +-
8 files changed, 201 insertions(+), 5 deletions(-)
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index b5da476c20..2b36b5fef1 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2523,6 +2523,17 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>last_inactive_time</structfield> <type>timestamptz</type>
+ </para>
+ <para>
+ The time at which the slot became inactive.
+ <literal>NULL</literal> if the slot is currently actively being
+ used.
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>conflicting</structfield> <type>bool</type>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index f69b7f5580..bc70ff193e 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1023,6 +1023,7 @@ CREATE VIEW pg_replication_slots AS
L.wal_status,
L.safe_wal_size,
L.two_phase,
+ L.last_inactive_time,
L.conflicting,
L.invalidation_reason,
L.failover,
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index cdf0c450c5..0f48d6dc7c 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -409,6 +409,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
slot->candidate_restart_valid = InvalidXLogRecPtr;
slot->candidate_restart_lsn = InvalidXLogRecPtr;
slot->last_saved_confirmed_flush = InvalidXLogRecPtr;
+ slot->last_inactive_time = 0;
/*
* Create the slot on disk. We haven't actually marked the slot allocated
@@ -622,6 +623,11 @@ retry:
if (SlotIsLogical(s))
pgstat_acquire_replslot(s);
+ /* The slot is active by now, so reset the last inactive time. */
+ SpinLockAcquire(&s->mutex);
+ s->last_inactive_time = 0;
+ SpinLockRelease(&s->mutex);
+
if (am_walsender)
{
ereport(log_replication_commands ? LOG : DEBUG1,
@@ -645,6 +651,7 @@ ReplicationSlotRelease(void)
ReplicationSlot *slot = MyReplicationSlot;
char *slotname = NULL; /* keep compiler quiet */
bool is_logical = false; /* keep compiler quiet */
+ TimestampTz now;
Assert(slot != NULL && slot->active_pid != 0);
@@ -679,6 +686,12 @@ ReplicationSlotRelease(void)
ReplicationSlotsComputeRequiredXmin(false);
}
+ /*
+ * Set the last inactive time after marking slot inactive. We get current
+ * time beforehand to avoid system call while holding the lock.
+ */
+ now = GetCurrentTimestamp();
+
if (slot->data.persistency == RS_PERSISTENT)
{
/*
@@ -687,9 +700,16 @@ ReplicationSlotRelease(void)
*/
SpinLockAcquire(&slot->mutex);
slot->active_pid = 0;
+ slot->last_inactive_time = now;
SpinLockRelease(&slot->mutex);
ConditionVariableBroadcast(&slot->active_cv);
}
+ else
+ {
+ SpinLockAcquire(&slot->mutex);
+ slot->last_inactive_time = now;
+ SpinLockRelease(&slot->mutex);
+ }
MyReplicationSlot = NULL;
@@ -2342,6 +2362,13 @@ RestoreSlotFromDisk(const char *name)
slot->in_use = true;
slot->active_pid = 0;
+ /*
+ * We set last inactive time after loading the slot from the disk into
+ * memory. Whoever acquires the slot i.e. makes the slot active will
+ * anyway reset it.
+ */
+ slot->last_inactive_time = GetCurrentTimestamp();
+
restored = true;
break;
}
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 4232c1e52e..24f5e6d90a 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -239,7 +239,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_get_replication_slots(PG_FUNCTION_ARGS)
{
-#define PG_GET_REPLICATION_SLOTS_COLS 18
+#define PG_GET_REPLICATION_SLOTS_COLS 19
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
XLogRecPtr currlsn;
int slotno;
@@ -410,6 +410,11 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
values[i++] = BoolGetDatum(slot_contents.data.two_phase);
+ if (slot_contents.last_inactive_time > 0)
+ values[i++] = TimestampTzGetDatum(slot_contents.last_inactive_time);
+ else
+ nulls[i++] = true;
+
cause = slot_contents.data.invalidated;
if (SlotIsPhysical(&slot_contents))
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 71c74350a0..0d26e5b422 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11133,9 +11133,9 @@
proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', prorettype => 'record',
proargtypes => '',
- proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,text,bool,bool}',
- proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,invalidation_reason,failover,synced}',
+ proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,timestamptz,bool,text,bool,bool}',
+ proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,last_inactive_time,conflicting,invalidation_reason,failover,synced}',
prosrc => 'pg_get_replication_slots' },
{ oid => '3786', descr => 'set up a logical replication slot',
proname => 'pg_create_logical_replication_slot', provolatile => 'v',
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 7f25a083ee..2f18433ecc 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -201,6 +201,9 @@ typedef struct ReplicationSlot
* forcibly flushed or not.
*/
XLogRecPtr last_saved_confirmed_flush;
+
+ /* The time at which this slot become inactive */
+ TimestampTz last_inactive_time;
} ReplicationSlot;
#define SlotIsPhysical(slot) ((slot)->data.database == InvalidOid)
diff --git a/src/test/recovery/t/019_replslot_limit.pl b/src/test/recovery/t/019_replslot_limit.pl
index fe00370c3e..bff84cc9c4 100644
--- a/src/test/recovery/t/019_replslot_limit.pl
+++ b/src/test/recovery/t/019_replslot_limit.pl
@@ -410,4 +410,152 @@ kill 'CONT', $receiverpid;
$node_primary3->stop;
$node_standby3->stop;
+# =============================================================================
+# Testcase start: Check last_inactive_time property of streaming standby's slot
+#
+
+# Initialize primary node
+my $primary4 = PostgreSQL::Test::Cluster->new('primary4');
+$primary4->init(allows_streaming => 'logical');
+$primary4->start;
+
+# Take backup
+$backup_name = 'my_backup4';
+$primary4->backup($backup_name);
+
+# Create a standby linking to the primary using the replication slot
+my $standby4 = PostgreSQL::Test::Cluster->new('standby4');
+$standby4->init_from_backup($primary4, $backup_name, has_streaming => 1);
+
+my $sb4_slot = 'sb4_slot';
+$standby4->append_conf('postgresql.conf', "primary_slot_name = '$sb4_slot'");
+
+my $slot_creation_time = $primary4->safe_psql(
+ 'postgres', qq[
+ SELECT current_timestamp;
+]);
+
+$primary4->safe_psql(
+ 'postgres', qq[
+ SELECT pg_create_physical_replication_slot(slot_name := '$sb4_slot');
+]);
+
+# Get last_inactive_time value after slot's creation. Note that the slot is still
+# inactive unless it's used by the standby below.
+my $last_inactive_time = $primary4->safe_psql('postgres',
+ qq(SELECT last_inactive_time FROM pg_replication_slots WHERE slot_name = '$sb4_slot' AND last_inactive_time IS NOT NULL;)
+);
+
+# Check that the captured time is sane
+is( $primary4->safe_psql(
+ 'postgres',
+ qq[SELECT '$last_inactive_time'::timestamptz > to_timestamp(0) AND '$last_inactive_time'::timestamptz > '$slot_creation_time'::timestamptz;]
+ ),
+ 't',
+ 'last inactive time for an active physical slot is sane');
+
+$standby4->start;
+
+# Wait until standby has replayed enough data
+$primary4->wait_for_catchup($standby4);
+
+# Now the slot is active so last_inactive_time value must be NULL
+is( $primary4->safe_psql(
+ 'postgres',
+ qq[SELECT last_inactive_time IS NULL FROM pg_replication_slots WHERE slot_name = '$sb4_slot';]
+ ),
+ 't',
+ 'last inactive time for an active physical slot is NULL');
+
+# Stop the standby to check its last_inactive_time value is updated
+$standby4->stop;
+
+# Let's also restart the primary so that the last_inactive_time is set upon
+# loading the slot from disk.
+$primary4->restart;
+
+is( $primary4->safe_psql(
+ 'postgres',
+ qq[SELECT last_inactive_time > '$last_inactive_time'::timestamptz FROM pg_replication_slots WHERE slot_name = '$sb4_slot' AND last_inactive_time IS NOT NULL;]
+ ),
+ 't',
+ 'last inactive time for an inactive physical slot is updated correctly');
+
+$standby4->stop;
+
+# Testcase end: Check last_inactive_time property of streaming standby's slot
+# =============================================================================
+
+# =============================================================================
+# Testcase start: Check last_inactive_time property of logical subscriber's slot
+my $publisher4 = $primary4;
+
+# Create subscriber node
+my $subscriber4 = PostgreSQL::Test::Cluster->new('subscriber4');
+$subscriber4->init;
+
+# Setup logical replication
+my $publisher4_connstr = $publisher4->connstr . ' dbname=postgres';
+$publisher4->safe_psql('postgres', "CREATE PUBLICATION pub FOR ALL TABLES");
+
+$slot_creation_time = $publisher4->safe_psql(
+ 'postgres', qq[
+ SELECT current_timestamp;
+]);
+
+my $lsub4_slot = 'lsub4_slot';
+$publisher4->safe_psql('postgres',
+ "SELECT pg_create_logical_replication_slot(slot_name := '$lsub4_slot', plugin := 'pgoutput');"
+);
+
+# Get last_inactive_time value after slot's creation. Note that the slot is still
+# inactive unless it's used by the subscriber below.
+$last_inactive_time = $publisher4->safe_psql('postgres',
+ qq(SELECT last_inactive_time FROM pg_replication_slots WHERE slot_name = '$lsub4_slot' AND last_inactive_time IS NOT NULL;)
+);
+
+# Check that the captured time is sane
+is( $publisher4->safe_psql(
+ 'postgres',
+ qq[SELECT '$last_inactive_time'::timestamptz > to_timestamp(0) AND '$last_inactive_time'::timestamptz > '$slot_creation_time'::timestamptz;]
+ ),
+ 't',
+ 'last inactive time for an active physical slot is sane');
+
+$subscriber4->start;
+$subscriber4->safe_psql('postgres',
+ "CREATE SUBSCRIPTION sub CONNECTION '$publisher4_connstr' PUBLICATION pub WITH (slot_name = '$lsub4_slot', create_slot = false)"
+);
+
+# Wait until subscriber has caught up
+$subscriber4->wait_for_subscription_sync($publisher4, 'sub');
+
+# Now the slot is active so last_inactive_time value must be NULL
+is( $publisher4->safe_psql(
+ 'postgres',
+ qq[SELECT last_inactive_time IS NULL FROM pg_replication_slots WHERE slot_name = '$lsub4_slot';]
+ ),
+ 't',
+ 'last inactive time for an active logical slot is NULL');
+
+# Stop the subscriber to check its last_inactive_time value is updated
+$subscriber4->stop;
+
+# Let's also restart the publisher so that the last_inactive_time is set upon
+# loading the slot from disk.
+$publisher4->restart;
+
+is( $publisher4->safe_psql(
+ 'postgres',
+ qq[SELECT last_inactive_time > '$last_inactive_time'::timestamptz FROM pg_replication_slots WHERE slot_name = '$lsub4_slot' AND last_inactive_time IS NOT NULL;]
+ ),
+ 't',
+ 'last inactive time for an inactive logical slot is updated correctly');
+
+# Testcase end: Check last_inactive_time property of logical subscriber's slot
+# =============================================================================
+
+$publisher4->stop;
+$subscriber4->stop;
+
done_testing();
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 18829ea586..dfcbaec387 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1473,11 +1473,12 @@ pg_replication_slots| SELECT l.slot_name,
l.wal_status,
l.safe_wal_size,
l.two_phase,
+ l.last_inactive_time,
l.conflicting,
l.invalidation_reason,
l.failover,
l.synced
- FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, invalidation_reason, failover, synced)
+ FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, last_inactive_time, conflicting, invalidation_reason, failover, synced)
LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
pg_roles| SELECT pg_authid.rolname,
pg_authid.rolsuper,
--
2.34.1
[application/octet-stream] v18-0002-Allow-setting-inactive_timeout-for-replication-s.patch (36.3K, ../CALj2ACU6tV0bh_MiFBYU7-VK2cj8Gas73htcQjoOv0uvSEG_HA@mail.gmail.com/3-v18-0002-Allow-setting-inactive_timeout-for-replication-s.patch)
download | inline diff:
From 0d80cfad9658c7303d036833636b22293fd25109 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Sun, 24 Mar 2024 07:14:51 +0000
Subject: [PATCH v18 2/5] Allow setting inactive_timeout for replication slots
via SQL API.
This commit adds a new replication slot property called
inactive_timeout specifying the amount of time in seconds the slot
is allowed to be inactive. It is added to slot's persistent data
structure to survive during server restarts. It will be synced to
failover slots on the standby, and also will be carried over to
the new cluster as part of pg_upgrade.
This commit particularly lets one specify the inactive_timeout for
a slot via SQL functions pg_create_physical_replication_slot and
pg_create_logical_replication_slot.
The new property will be useful to implement inactive timeout based
replication slot invalidation in a future commit.
Author: Bharath Rupireddy
Reviewed-by: Bertrand Drouvot, Amit Kapila
Discussion: https://www.postgresql.org/message-id/CALj2ACW4aUe-_uFQOjdWCEN-xXoLGhmvRFnL8SNw_TZ5nJe+aw@mail.gmail.com
---
contrib/test_decoding/expected/slot.out | 102 ++++++++++++++++++
contrib/test_decoding/sql/slot.sql | 34 ++++++
doc/src/sgml/func.sgml | 18 ++--
doc/src/sgml/system-views.sgml | 9 ++
src/backend/catalog/system_functions.sql | 2 +
src/backend/catalog/system_views.sql | 1 +
src/backend/replication/logical/slotsync.c | 17 ++-
src/backend/replication/slot.c | 20 +++-
src/backend/replication/slotfuncs.c | 31 +++++-
src/backend/replication/walsender.c | 4 +-
src/bin/pg_upgrade/info.c | 6 +-
src/bin/pg_upgrade/pg_upgrade.c | 5 +-
src/bin/pg_upgrade/pg_upgrade.h | 2 +
src/bin/pg_upgrade/t/003_logical_slots.pl | 11 +-
src/include/catalog/pg_proc.dat | 22 ++--
src/include/replication/slot.h | 5 +-
.../t/040_standby_failover_slots_sync.pl | 13 ++-
src/test/regress/expected/rules.out | 3 +-
18 files changed, 264 insertions(+), 41 deletions(-)
diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out
index 349ab2d380..6771520afb 100644
--- a/contrib/test_decoding/expected/slot.out
+++ b/contrib/test_decoding/expected/slot.out
@@ -466,3 +466,105 @@ SELECT pg_drop_replication_slot('physical_slot');
(1 row)
+-- Test negative value for inactive_timeout option for slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_fail_slot', inactive_timeout := -300); -- error
+ERROR: "inactive_timeout" must not be negative
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_fail_slot', plugin := 'test_decoding', inactive_timeout := -600); -- error
+ERROR: "inactive_timeout" must not be negative
+-- Test inactive_timeout option for temporary slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_fail_slot', temporary := true, inactive_timeout := 300); -- error
+ERROR: cannot set inactive_timeout for a temporary replication slot
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_fail_slot', plugin := 'test_decoding', temporary := true, inactive_timeout := 600); -- error
+ERROR: cannot set inactive_timeout for a temporary replication slot
+-- Test inactive_timeout option of physical slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot1', immediately_reserve := true, inactive_timeout := 300);
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot2');
+ ?column?
+----------
+ init
+(1 row)
+
+-- Copy physical slot with inactive_timeout option set.
+SELECT 'copy' FROM pg_copy_physical_replication_slot(src_slot_name := 'it_phy_slot1', dst_slot_name := 'it_phy_slot3');
+ ?column?
+----------
+ copy
+(1 row)
+
+SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
+ slot_name | slot_type | inactive_timeout
+--------------+-----------+------------------
+ it_phy_slot1 | physical | 300
+ it_phy_slot2 | physical | 0
+ it_phy_slot3 | physical | 300
+(3 rows)
+
+SELECT pg_drop_replication_slot('it_phy_slot1');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('it_phy_slot2');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('it_phy_slot3');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+-- Test inactive_timeout option of logical slots.
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot1', plugin := 'test_decoding', inactive_timeout := 600);
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot2', plugin := 'test_decoding');
+ ?column?
+----------
+ init
+(1 row)
+
+-- Copy logical slot with inactive_timeout option set.
+SELECT 'copy' FROM pg_copy_logical_replication_slot(src_slot_name := 'it_log_slot1', dst_slot_name := 'it_log_slot3');
+ ?column?
+----------
+ copy
+(1 row)
+
+SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
+ slot_name | slot_type | inactive_timeout
+--------------+-----------+------------------
+ it_log_slot1 | logical | 600
+ it_log_slot2 | logical | 0
+ it_log_slot3 | logical | 600
+(3 rows)
+
+SELECT pg_drop_replication_slot('it_log_slot1');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('it_log_slot2');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('it_log_slot3');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
diff --git a/contrib/test_decoding/sql/slot.sql b/contrib/test_decoding/sql/slot.sql
index 580e3ae3be..443e91da07 100644
--- a/contrib/test_decoding/sql/slot.sql
+++ b/contrib/test_decoding/sql/slot.sql
@@ -190,3 +190,37 @@ 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 negative value for inactive_timeout option for slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_fail_slot', inactive_timeout := -300); -- error
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_fail_slot', plugin := 'test_decoding', inactive_timeout := -600); -- error
+
+-- Test inactive_timeout option for temporary slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_fail_slot', temporary := true, inactive_timeout := 300); -- error
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_fail_slot', plugin := 'test_decoding', temporary := true, inactive_timeout := 600); -- error
+
+-- Test inactive_timeout option of physical slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot1', immediately_reserve := true, inactive_timeout := 300);
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot2');
+
+-- Copy physical slot with inactive_timeout option set.
+SELECT 'copy' FROM pg_copy_physical_replication_slot(src_slot_name := 'it_phy_slot1', dst_slot_name := 'it_phy_slot3');
+
+SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
+
+SELECT pg_drop_replication_slot('it_phy_slot1');
+SELECT pg_drop_replication_slot('it_phy_slot2');
+SELECT pg_drop_replication_slot('it_phy_slot3');
+
+-- Test inactive_timeout option of logical slots.
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot1', plugin := 'test_decoding', inactive_timeout := 600);
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot2', plugin := 'test_decoding');
+
+-- Copy logical slot with inactive_timeout option set.
+SELECT 'copy' FROM pg_copy_logical_replication_slot(src_slot_name := 'it_log_slot1', dst_slot_name := 'it_log_slot3');
+
+SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
+
+SELECT pg_drop_replication_slot('it_log_slot1');
+SELECT pg_drop_replication_slot('it_log_slot2');
+SELECT pg_drop_replication_slot('it_log_slot3');
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index 8ecc02f2b9..afaafa35ad 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -28373,7 +28373,7 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
<indexterm>
<primary>pg_create_physical_replication_slot</primary>
</indexterm>
- <function>pg_create_physical_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type> <optional>, <parameter>immediately_reserve</parameter> <type>boolean</type>, <parameter>temporary</parameter> <type>boolean</type> </optional> )
+ <function>pg_create_physical_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type> <optional>, <parameter>immediately_reserve</parameter> <type>boolean</type>, <parameter>temporary</parameter> <type>boolean</type>, <parameter>inactive_timeout</parameter> <type>integer</type> </optional>)
<returnvalue>record</returnvalue>
( <parameter>slot_name</parameter> <type>name</type>,
<parameter>lsn</parameter> <type>pg_lsn</type> )
@@ -28390,9 +28390,12 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
parameter, <parameter>temporary</parameter>, 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
- to the replication protocol command <literal>CREATE_REPLICATION_SLOT
- ... PHYSICAL</literal>.
+ released upon any error. The optional fourth
+ parameter, <parameter>inactive_timeout</parameter>, when set to a
+ non-zero value, specifies the amount of time in seconds the slot is
+ allowed to be inactive. This function corresponds to the replication
+ protocol command
+ <literal>CREATE_REPLICATION_SLOT ... PHYSICAL</literal>.
</para></entry>
</row>
@@ -28417,7 +28420,7 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
<indexterm>
<primary>pg_create_logical_replication_slot</primary>
</indexterm>
- <function>pg_create_logical_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type>, <parameter>plugin</parameter> <type>name</type> <optional>, <parameter>temporary</parameter> <type>boolean</type>, <parameter>twophase</parameter> <type>boolean</type>, <parameter>failover</parameter> <type>boolean</type> </optional> )
+ <function>pg_create_logical_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type>, <parameter>plugin</parameter> <type>name</type> <optional>, <parameter>temporary</parameter> <type>boolean</type>, <parameter>twophase</parameter> <type>boolean</type>, <parameter>failover</parameter> <type>boolean</type>, <parameter>inactive_timeout</parameter> <type>integer</type> </optional> )
<returnvalue>record</returnvalue>
( <parameter>slot_name</parameter> <type>name</type>,
<parameter>lsn</parameter> <type>pg_lsn</type> )
@@ -28436,7 +28439,10 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
<parameter>failover</parameter>, when set to true,
specifies that this slot is enabled to be synced to the
standbys so that logical replication can be resumed after
- failover. A call to this function has the same effect as
+ failover. The optional sixth parameter,
+ <parameter>inactive_timeout</parameter>, when set to a
+ non-zero value, specifies the amount of time in seconds the slot is
+ allowed to be inactive. A call to this function has the same effect as
the replication protocol command
<literal>CREATE_REPLICATION_SLOT ... LOGICAL</literal>.
</para></entry>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index 2b36b5fef1..dddbaa070f 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2534,6 +2534,15 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>inactive_timeout</structfield> <type>integer</type>
+ </para>
+ <para>
+ The amount of time in seconds the slot is allowed to be inactive.
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>conflicting</structfield> <type>bool</type>
diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql
index fe2bb50f46..af27616657 100644
--- a/src/backend/catalog/system_functions.sql
+++ b/src/backend/catalog/system_functions.sql
@@ -469,6 +469,7 @@ AS 'pg_logical_emit_message_bytea';
CREATE OR REPLACE FUNCTION pg_create_physical_replication_slot(
IN slot_name name, IN immediately_reserve boolean DEFAULT false,
IN temporary boolean DEFAULT false,
+ IN inactive_timeout int DEFAULT 0,
OUT slot_name name, OUT lsn pg_lsn)
RETURNS RECORD
LANGUAGE INTERNAL
@@ -480,6 +481,7 @@ CREATE OR REPLACE FUNCTION pg_create_logical_replication_slot(
IN temporary boolean DEFAULT false,
IN twophase boolean DEFAULT false,
IN failover boolean DEFAULT false,
+ IN inactive_timeout int DEFAULT 0,
OUT slot_name name, OUT lsn pg_lsn)
RETURNS RECORD
LANGUAGE INTERNAL
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index bc70ff193e..40d7ad469d 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1024,6 +1024,7 @@ CREATE VIEW pg_replication_slots AS
L.safe_wal_size,
L.two_phase,
L.last_inactive_time,
+ L.inactive_timeout,
L.conflicting,
L.invalidation_reason,
L.failover,
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 30480960c5..c01876ceeb 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -131,6 +131,7 @@ typedef struct RemoteSlot
char *database;
bool two_phase;
bool failover;
+ int inactive_timeout;
XLogRecPtr restart_lsn;
XLogRecPtr confirmed_lsn;
TransactionId catalog_xmin;
@@ -167,7 +168,8 @@ update_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid)
remote_slot->two_phase == slot->data.two_phase &&
remote_slot->failover == slot->data.failover &&
remote_slot->confirmed_lsn == slot->data.confirmed_flush &&
- strcmp(remote_slot->plugin, NameStr(slot->data.plugin)) == 0)
+ strcmp(remote_slot->plugin, NameStr(slot->data.plugin)) == 0 &&
+ remote_slot->inactive_timeout == slot->data.inactive_timeout)
return false;
/* Avoid expensive operations while holding a spinlock. */
@@ -182,6 +184,7 @@ update_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid)
slot->data.confirmed_flush = remote_slot->confirmed_lsn;
slot->data.catalog_xmin = remote_slot->catalog_xmin;
slot->effective_catalog_xmin = remote_slot->catalog_xmin;
+ slot->data.inactive_timeout = remote_slot->inactive_timeout;
SpinLockRelease(&slot->mutex);
if (xmin_changed)
@@ -607,7 +610,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
ReplicationSlotCreate(remote_slot->name, true, RS_TEMPORARY,
remote_slot->two_phase,
remote_slot->failover,
- true);
+ true, 0);
/* For shorter lines. */
slot = MyReplicationSlot;
@@ -627,6 +630,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
SpinLockAcquire(&slot->mutex);
slot->effective_catalog_xmin = xmin_horizon;
slot->data.catalog_xmin = xmin_horizon;
+ slot->data.inactive_timeout = remote_slot->inactive_timeout;
SpinLockRelease(&slot->mutex);
ReplicationSlotsComputeRequiredXmin(true);
LWLockRelease(ProcArrayLock);
@@ -652,9 +656,9 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
static bool
synchronize_slots(WalReceiverConn *wrconn)
{
-#define SLOTSYNC_COLUMN_COUNT 9
+#define SLOTSYNC_COLUMN_COUNT 10
Oid slotRow[SLOTSYNC_COLUMN_COUNT] = {TEXTOID, TEXTOID, LSNOID,
- LSNOID, XIDOID, BOOLOID, BOOLOID, TEXTOID, TEXTOID};
+ LSNOID, XIDOID, BOOLOID, BOOLOID, TEXTOID, TEXTOID, INT4OID};
WalRcvExecResult *res;
TupleTableSlot *tupslot;
@@ -663,7 +667,7 @@ synchronize_slots(WalReceiverConn *wrconn)
bool started_tx = false;
const char *query = "SELECT slot_name, plugin, confirmed_flush_lsn,"
" restart_lsn, catalog_xmin, two_phase, failover,"
- " database, invalidation_reason"
+ " database, invalidation_reason, inactive_timeout"
" FROM pg_catalog.pg_replication_slots"
" WHERE failover and NOT temporary";
@@ -743,6 +747,9 @@ synchronize_slots(WalReceiverConn *wrconn)
remote_slot->invalidated = isnull ? RS_INVAL_NONE :
GetSlotInvalidationCause(TextDatumGetCString(d));
+ remote_slot->inactive_timeout = DatumGetInt32(slot_getattr(tupslot, ++col,
+ &isnull));
+
/* Sanity check */
Assert(col == SLOTSYNC_COLUMN_COUNT);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 0f48d6dc7c..852a657e97 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -129,7 +129,7 @@ StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1),
sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize
#define SLOT_MAGIC 0x1051CA1 /* format identifier */
-#define SLOT_VERSION 5 /* version for new files */
+#define SLOT_VERSION 6 /* version for new files */
/* Control array for replication slot management */
ReplicationSlotCtlData *ReplicationSlotCtl = NULL;
@@ -304,11 +304,14 @@ ReplicationSlotValidateName(const char *name, int elevel)
* failover: If enabled, allows the slot to be synced to standbys so
* that logical replication can be resumed after failover.
* synced: True if the slot is synchronized from the primary server.
+ * inactive_timeout: The amount of time in seconds the slot is allowed to be
+ * inactive.
*/
void
ReplicationSlotCreate(const char *name, bool db_specific,
ReplicationSlotPersistency persistency,
- bool two_phase, bool failover, bool synced)
+ bool two_phase, bool failover, bool synced,
+ int inactive_timeout)
{
ReplicationSlot *slot = NULL;
int i;
@@ -345,6 +348,18 @@ ReplicationSlotCreate(const char *name, bool db_specific,
errmsg("cannot enable failover for a temporary replication slot"));
}
+ if (inactive_timeout > 0)
+ {
+ /*
+ * Do not allow users to set inactive_timeout for temporary slots,
+ * because temporary slots will not be saved to the disk.
+ */
+ if (persistency == RS_TEMPORARY)
+ ereport(ERROR,
+ errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot set inactive_timeout for a temporary replication slot"));
+ }
+
/*
* If some other backend ran this code concurrently with us, we'd likely
* both allocate the same slot, and that would be bad. We'd also be at
@@ -398,6 +413,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
slot->data.two_phase_at = InvalidXLogRecPtr;
slot->data.failover = failover;
slot->data.synced = synced;
+ slot->data.inactive_timeout = inactive_timeout;
/* and then data only present in shared memory */
slot->just_dirtied = false;
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 24f5e6d90a..fb79401c50 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -38,14 +38,15 @@
*/
static void
create_physical_replication_slot(char *name, bool immediately_reserve,
- bool temporary, XLogRecPtr restart_lsn)
+ bool temporary, int inactive_timeout,
+ XLogRecPtr restart_lsn)
{
Assert(!MyReplicationSlot);
/* acquire replication slot, this will check for conflicting names */
ReplicationSlotCreate(name, false,
temporary ? RS_TEMPORARY : RS_PERSISTENT, false,
- false, false);
+ false, false, inactive_timeout);
if (immediately_reserve)
{
@@ -71,6 +72,7 @@ pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
Name name = PG_GETARG_NAME(0);
bool immediately_reserve = PG_GETARG_BOOL(1);
bool temporary = PG_GETARG_BOOL(2);
+ int inactive_timeout = PG_GETARG_INT32(3);
Datum values[2];
bool nulls[2];
TupleDesc tupdesc;
@@ -84,9 +86,15 @@ pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
CheckSlotRequirements();
+ if (inactive_timeout < 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+ errmsg("\"inactive_timeout\" must not be negative")));
+
create_physical_replication_slot(NameStr(*name),
immediately_reserve,
temporary,
+ inactive_timeout,
InvalidXLogRecPtr);
values[0] = NameGetDatum(&MyReplicationSlot->data.name);
@@ -120,7 +128,7 @@ pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
static void
create_logical_replication_slot(char *name, char *plugin,
bool temporary, bool two_phase,
- bool failover,
+ bool failover, int inactive_timeout,
XLogRecPtr restart_lsn,
bool find_startpoint)
{
@@ -138,7 +146,7 @@ create_logical_replication_slot(char *name, char *plugin,
*/
ReplicationSlotCreate(name, true,
temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase,
- failover, false);
+ failover, false, inactive_timeout);
/*
* Create logical decoding context to find start point or, if we don't
@@ -177,6 +185,7 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
bool temporary = PG_GETARG_BOOL(2);
bool two_phase = PG_GETARG_BOOL(3);
bool failover = PG_GETARG_BOOL(4);
+ int inactive_timeout = PG_GETARG_INT32(5);
Datum result;
TupleDesc tupdesc;
HeapTuple tuple;
@@ -190,11 +199,17 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
CheckLogicalDecodingRequirements();
+ if (inactive_timeout < 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+ errmsg("\"inactive_timeout\" must not be negative")));
+
create_logical_replication_slot(NameStr(*name),
NameStr(*plugin),
temporary,
two_phase,
failover,
+ inactive_timeout,
InvalidXLogRecPtr,
true);
@@ -239,7 +254,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_get_replication_slots(PG_FUNCTION_ARGS)
{
-#define PG_GET_REPLICATION_SLOTS_COLS 19
+#define PG_GET_REPLICATION_SLOTS_COLS 20
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
XLogRecPtr currlsn;
int slotno;
@@ -415,6 +430,8 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
else
nulls[i++] = true;
+ values[i++] = Int32GetDatum(slot_contents.data.inactive_timeout);
+
cause = slot_contents.data.invalidated;
if (SlotIsPhysical(&slot_contents))
@@ -720,6 +737,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
XLogRecPtr src_restart_lsn;
bool src_islogical;
bool temporary;
+ int inactive_timeout;
char *plugin;
Datum values[2];
bool nulls[2];
@@ -776,6 +794,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
src_restart_lsn = first_slot_contents.data.restart_lsn;
temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
+ inactive_timeout = first_slot_contents.data.inactive_timeout;
/* Check type of replication slot */
if (src_islogical != logical_slot)
@@ -823,6 +842,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
temporary,
false,
false,
+ inactive_timeout,
src_restart_lsn,
false);
}
@@ -830,6 +850,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
create_physical_replication_slot(NameStr(*dst_name),
true,
temporary,
+ inactive_timeout,
src_restart_lsn);
/*
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index bc40c454de..5315c08650 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1221,7 +1221,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
{
ReplicationSlotCreate(cmd->slotname, false,
cmd->temporary ? RS_TEMPORARY : RS_PERSISTENT,
- false, false, false);
+ false, false, false, 0);
if (reserve_wal)
{
@@ -1252,7 +1252,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
*/
ReplicationSlotCreate(cmd->slotname, true,
cmd->temporary ? RS_TEMPORARY : RS_EPHEMERAL,
- two_phase, failover, false);
+ two_phase, failover, false, 0);
/*
* Do options check early so that we can bail before calling the
diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c
index 95c22a7200..12626987f0 100644
--- a/src/bin/pg_upgrade/info.c
+++ b/src/bin/pg_upgrade/info.c
@@ -676,7 +676,8 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
* removed.
*/
res = executeQueryOrDie(conn, "SELECT slot_name, plugin, two_phase, failover, "
- "%s as caught_up, invalidation_reason IS NOT NULL as invalid "
+ "%s as caught_up, invalidation_reason IS NOT NULL as invalid, "
+ "inactive_timeout "
"FROM pg_catalog.pg_replication_slots "
"WHERE slot_type = 'logical' AND "
"database = current_database() AND "
@@ -696,6 +697,7 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
int i_failover;
int i_caught_up;
int i_invalid;
+ int i_inactive_timeout;
slotinfos = (LogicalSlotInfo *) pg_malloc(sizeof(LogicalSlotInfo) * num_slots);
@@ -705,6 +707,7 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
i_failover = PQfnumber(res, "failover");
i_caught_up = PQfnumber(res, "caught_up");
i_invalid = PQfnumber(res, "invalid");
+ i_inactive_timeout = PQfnumber(res, "inactive_timeout");
for (int slotnum = 0; slotnum < num_slots; slotnum++)
{
@@ -716,6 +719,7 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
curr->failover = (strcmp(PQgetvalue(res, slotnum, i_failover), "t") == 0);
curr->caught_up = (strcmp(PQgetvalue(res, slotnum, i_caught_up), "t") == 0);
curr->invalid = (strcmp(PQgetvalue(res, slotnum, i_invalid), "t") == 0);
+ curr->inactive_timeout = atooid(PQgetvalue(res, slotnum, i_inactive_timeout));
}
}
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index f6143b6bc4..2656056103 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -931,9 +931,10 @@ create_logical_replication_slots(void)
appendPQExpBuffer(query, ", ");
appendStringLiteralConn(query, slot_info->plugin, conn);
- appendPQExpBuffer(query, ", false, %s, %s);",
+ appendPQExpBuffer(query, ", false, %s, %s, %d);",
slot_info->two_phase ? "true" : "false",
- slot_info->failover ? "true" : "false");
+ slot_info->failover ? "true" : "false",
+ slot_info->inactive_timeout);
PQclear(executeQueryOrDie(conn, "%s", query->data));
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h
index 92bcb693fb..eb86d000b1 100644
--- a/src/bin/pg_upgrade/pg_upgrade.h
+++ b/src/bin/pg_upgrade/pg_upgrade.h
@@ -162,6 +162,8 @@ typedef struct
bool invalid; /* if true, the slot is unusable */
bool failover; /* is the slot designated to be synced to the
* physical standby? */
+ int inactive_timeout; /* The amount of time in seconds the slot
+ * is allowed to be inactive. */
} LogicalSlotInfo;
typedef struct
diff --git a/src/bin/pg_upgrade/t/003_logical_slots.pl b/src/bin/pg_upgrade/t/003_logical_slots.pl
index 83d71c3084..6e82d2cb7b 100644
--- a/src/bin/pg_upgrade/t/003_logical_slots.pl
+++ b/src/bin/pg_upgrade/t/003_logical_slots.pl
@@ -153,14 +153,17 @@ like(
# TEST: Successful upgrade
# Preparations for the subsequent test:
-# 1. Setup logical replication (first, cleanup slots from the previous tests)
+# 1. Setup logical replication (first, cleanup slots from the previous tests,
+# and then create slot for this test with inactive_timeout set).
my $old_connstr = $oldpub->connstr . ' dbname=postgres';
+my $inactive_timeout = 3600;
$oldpub->start;
$oldpub->safe_psql(
'postgres', qq[
SELECT * FROM pg_drop_replication_slot('test_slot1');
SELECT * FROM pg_drop_replication_slot('test_slot2');
+ SELECT pg_create_logical_replication_slot(slot_name := 'regress_sub', plugin := 'pgoutput', inactive_timeout := $inactive_timeout);
CREATE PUBLICATION regress_pub FOR ALL TABLES;
]);
@@ -172,7 +175,7 @@ $sub->start;
$sub->safe_psql(
'postgres', qq[
CREATE TABLE tbl (a int);
- CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (two_phase = 'true', failover = 'true')
+ CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (slot_name = 'regress_sub', create_slot = false, two_phase = 'true', failover = 'true')
]);
$sub->wait_for_subscription_sync($oldpub, 'regress_sub');
@@ -192,8 +195,8 @@ command_ok([@pg_upgrade_cmd], 'run of pg_upgrade of old cluster');
# Check that the slot 'regress_sub' has migrated to the new cluster
$newpub->start;
my $result = $newpub->safe_psql('postgres',
- "SELECT slot_name, two_phase, failover FROM pg_replication_slots");
-is($result, qq(regress_sub|t|t), 'check the slot exists on new cluster');
+ "SELECT slot_name, two_phase, failover, inactive_timeout = $inactive_timeout FROM pg_replication_slots");
+is($result, qq(regress_sub|t|t|t), 'check the slot exists on new cluster');
# Update the connection
my $new_connstr = $newpub->connstr . ' dbname=postgres';
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 0d26e5b422..a09da44b6a 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11105,10 +11105,10 @@
# replication slots
{ oid => '3779', descr => 'create a physical replication slot',
proname => 'pg_create_physical_replication_slot', provolatile => 'v',
- proparallel => 'u', prorettype => 'record', proargtypes => 'name bool bool',
- proallargtypes => '{name,bool,bool,name,pg_lsn}',
- proargmodes => '{i,i,i,o,o}',
- proargnames => '{slot_name,immediately_reserve,temporary,slot_name,lsn}',
+ proparallel => 'u', prorettype => 'record', proargtypes => 'name bool bool int4',
+ proallargtypes => '{name,bool,bool,int4,name,pg_lsn}',
+ proargmodes => '{i,i,i,i,o,o}',
+ proargnames => '{slot_name,immediately_reserve,temporary,inactive_timeout,slot_name,lsn}',
prosrc => 'pg_create_physical_replication_slot' },
{ oid => '4220',
descr => 'copy a physical replication slot, changing temporality',
@@ -11133,17 +11133,17 @@
proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', prorettype => 'record',
proargtypes => '',
- proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,timestamptz,bool,text,bool,bool}',
- proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,last_inactive_time,conflicting,invalidation_reason,failover,synced}',
+ proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,timestamptz,int4,bool,text,bool,bool}',
+ proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,last_inactive_time,inactive_timeout,conflicting,invalidation_reason,failover,synced}',
prosrc => 'pg_get_replication_slots' },
{ oid => '3786', descr => 'set up a logical replication slot',
proname => 'pg_create_logical_replication_slot', provolatile => 'v',
proparallel => 'u', prorettype => 'record',
- proargtypes => 'name name bool bool bool',
- proallargtypes => '{name,name,bool,bool,bool,name,pg_lsn}',
- proargmodes => '{i,i,i,i,i,o,o}',
- proargnames => '{slot_name,plugin,temporary,twophase,failover,slot_name,lsn}',
+ proargtypes => 'name name bool bool bool int4',
+ proallargtypes => '{name,name,bool,bool,bool,int4,name,pg_lsn}',
+ proargmodes => '{i,i,i,i,i,i,o,o}',
+ proargnames => '{slot_name,plugin,temporary,twophase,failover,inactive_timeout,slot_name,lsn}',
prosrc => 'pg_create_logical_replication_slot' },
{ oid => '4222',
descr => 'copy a logical replication slot, changing temporality and plugin',
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 2f18433ecc..24623cfdc1 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -127,6 +127,9 @@ typedef struct ReplicationSlotPersistentData
* for logical slots on the primary server.
*/
bool failover;
+
+ /* The amount of time in seconds the slot is allowed to be inactive */
+ int inactive_timeout;
} ReplicationSlotPersistentData;
/*
@@ -239,7 +242,7 @@ extern void ReplicationSlotsShmemInit(void);
extern void ReplicationSlotCreate(const char *name, bool db_specific,
ReplicationSlotPersistency persistency,
bool two_phase, bool failover,
- bool synced);
+ bool synced, int inactive_timeout);
extern void ReplicationSlotPersist(void);
extern void ReplicationSlotDrop(const char *name, bool nowait);
extern void ReplicationSlotDropAcquired(void);
diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl
index f47bfd78eb..3dd780beab 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -152,8 +152,9 @@ log_min_messages = 'debug2'
$primary->append_conf('postgresql.conf', "log_min_messages = 'debug2'");
$primary->reload;
+my $inactive_timeout = 3600;
$primary->psql('postgres',
- q{SELECT pg_create_logical_replication_slot('lsub2_slot', 'test_decoding', false, false, true);}
+ "SELECT pg_create_logical_replication_slot('lsub2_slot', 'test_decoding', false, false, true, $inactive_timeout);"
);
$primary->psql('postgres',
@@ -190,6 +191,16 @@ is( $standby1->safe_psql(
"t",
'logical slots have synced as true on standby');
+# Confirm that the synced slot on the standby has got inactive_timeout from the
+# primary.
+is( $standby1->safe_psql(
+ 'postgres',
+ "SELECT inactive_timeout = $inactive_timeout FROM pg_replication_slots
+ WHERE slot_name = 'lsub2_slot' AND synced AND NOT temporary;"
+ ),
+ "t",
+ 'synced logical slot has got inactive_timeout on standby');
+
##################################################
# Test that the synchronized slot will be dropped if the corresponding remote
# slot on the primary server has been dropped.
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index dfcbaec387..d532e23176 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1474,11 +1474,12 @@ pg_replication_slots| SELECT l.slot_name,
l.safe_wal_size,
l.two_phase,
l.last_inactive_time,
+ l.inactive_timeout,
l.conflicting,
l.invalidation_reason,
l.failover,
l.synced
- FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, last_inactive_time, conflicting, invalidation_reason, failover, synced)
+ FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, last_inactive_time, inactive_timeout, conflicting, invalidation_reason, failover, synced)
LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
pg_roles| SELECT pg_authid.rolname,
pg_authid.rolsuper,
--
2.34.1
[application/octet-stream] v18-0003-Introduce-new-SQL-funtion-pg_alter_replication_s.patch (15.8K, ../CALj2ACU6tV0bh_MiFBYU7-VK2cj8Gas73htcQjoOv0uvSEG_HA@mail.gmail.com/4-v18-0003-Introduce-new-SQL-funtion-pg_alter_replication_s.patch)
download | inline diff:
From 65be663680fbde9812392a4fa739633060625f82 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Sun, 24 Mar 2024 07:18:17 +0000
Subject: [PATCH v18 3/5] Introduce new SQL funtion pg_alter_replication_slot
This commit adds a new function pg_alter_replication_slot to alter
the given property of a replication slot. It is similar to
replication protocol command ALTER_REPLICATION_SLOT, except that
for now it allows only inactive_timeout property to be set. The
reason for disallowing failover property to be altered via this
function is to avoid inconsistency with the catalog
pg_subscription on the logical subscriber. Because, the subscriber
won't know the altered value of its replication slot on the
publisher.
Author: Bharath Rupireddy
Reviewed-by: Bertrand Drouvot, Amit Kapila
Discussion: https://www.postgresql.org/message-id/CALj2ACW4aUe-_uFQOjdWCEN-xXoLGhmvRFnL8SNw_TZ5nJe+aw@mail.gmail.com
---
contrib/test_decoding/expected/slot.out | 44 ++++++++++++++-
contrib/test_decoding/sql/slot.sql | 10 ++++
doc/src/sgml/func.sgml | 21 ++++++++
src/backend/replication/slot.c | 22 ++++----
src/backend/replication/slotfuncs.c | 66 ++++++++++++++++++++++-
src/bin/pg_upgrade/t/003_logical_slots.pl | 14 +++--
src/include/catalog/pg_proc.dat | 5 ++
src/include/replication/slot.h | 2 +
8 files changed, 167 insertions(+), 17 deletions(-)
diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out
index 6771520afb..5b8dbf6f52 100644
--- a/contrib/test_decoding/expected/slot.out
+++ b/contrib/test_decoding/expected/slot.out
@@ -496,13 +496,27 @@ SELECT 'copy' FROM pg_copy_physical_replication_slot(src_slot_name := 'it_phy_sl
copy
(1 row)
+-- Test alter physical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot4');
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_phy_slot4', inactive_timeout := 900);
+ ?column?
+----------
+ alter
+(1 row)
+
SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
slot_name | slot_type | inactive_timeout
--------------+-----------+------------------
it_phy_slot1 | physical | 300
it_phy_slot2 | physical | 0
it_phy_slot3 | physical | 300
-(3 rows)
+ it_phy_slot4 | physical | 900
+(4 rows)
SELECT pg_drop_replication_slot('it_phy_slot1');
pg_drop_replication_slot
@@ -522,6 +536,12 @@ SELECT pg_drop_replication_slot('it_phy_slot3');
(1 row)
+SELECT pg_drop_replication_slot('it_phy_slot4');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
-- Test inactive_timeout option of logical slots.
SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot1', plugin := 'test_decoding', inactive_timeout := 600);
?column?
@@ -542,13 +562,27 @@ SELECT 'copy' FROM pg_copy_logical_replication_slot(src_slot_name := 'it_log_slo
copy
(1 row)
+-- Test alter logical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot4', plugin := 'test_decoding');
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_log_slot4', inactive_timeout := 900);
+ ?column?
+----------
+ alter
+(1 row)
+
SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
slot_name | slot_type | inactive_timeout
--------------+-----------+------------------
it_log_slot1 | logical | 600
it_log_slot2 | logical | 0
it_log_slot3 | logical | 600
-(3 rows)
+ it_log_slot4 | logical | 900
+(4 rows)
SELECT pg_drop_replication_slot('it_log_slot1');
pg_drop_replication_slot
@@ -568,3 +602,9 @@ SELECT pg_drop_replication_slot('it_log_slot3');
(1 row)
+SELECT pg_drop_replication_slot('it_log_slot4');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
diff --git a/contrib/test_decoding/sql/slot.sql b/contrib/test_decoding/sql/slot.sql
index 443e91da07..6785714cc7 100644
--- a/contrib/test_decoding/sql/slot.sql
+++ b/contrib/test_decoding/sql/slot.sql
@@ -206,11 +206,16 @@ SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot
-- Copy physical slot with inactive_timeout option set.
SELECT 'copy' FROM pg_copy_physical_replication_slot(src_slot_name := 'it_phy_slot1', dst_slot_name := 'it_phy_slot3');
+-- Test alter physical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot4');
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_phy_slot4', inactive_timeout := 900);
+
SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
SELECT pg_drop_replication_slot('it_phy_slot1');
SELECT pg_drop_replication_slot('it_phy_slot2');
SELECT pg_drop_replication_slot('it_phy_slot3');
+SELECT pg_drop_replication_slot('it_phy_slot4');
-- Test inactive_timeout option of logical slots.
SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot1', plugin := 'test_decoding', inactive_timeout := 600);
@@ -219,8 +224,13 @@ SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot2
-- Copy logical slot with inactive_timeout option set.
SELECT 'copy' FROM pg_copy_logical_replication_slot(src_slot_name := 'it_log_slot1', dst_slot_name := 'it_log_slot3');
+-- Test alter logical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot4', plugin := 'test_decoding');
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_log_slot4', inactive_timeout := 900);
+
SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
SELECT pg_drop_replication_slot('it_log_slot1');
SELECT pg_drop_replication_slot('it_log_slot2');
SELECT pg_drop_replication_slot('it_log_slot3');
+SELECT pg_drop_replication_slot('it_log_slot4');
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index afaafa35ad..22c8e0d39c 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -28829,6 +28829,27 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
</entry>
</row>
+ <row>
+ <entry role="func_table_entry"><para role="func_signature">
+ <indexterm>
+ <primary>pg_alter_replication_slot</primary>
+ </indexterm>
+ <function>pg_alter_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type>, <parameter>inactive_timeout</parameter> <type>integer</type> )
+ <returnvalue>void</returnvalue>
+ </para>
+ <para>
+ Alters the given property of a replication slot
+ named <parameter>slot_name</parameter>. Same as replication protocol
+ command <literal>ALTER_REPLICATION_SLOT</literal>, except that it
+ allows only <parameter>inactive_timeout</parameter> property to be set.
+ The reason for disallowing <parameter>failover</parameter> property to
+ be altered via this function is to avoid inconsistency with the catalog
+ <structname>pg_subscription</structname> on the logical subscriber.
+ Because, the subscriber won't know the altered value of its
+ replication slot on the publisher.
+ </para></entry>
+ </row>
+
</tbody>
</tgroup>
</table>
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 852a657e97..3287aa2860 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -162,7 +162,6 @@ static void ReplicationSlotDropPtr(ReplicationSlot *slot);
/* internal persistency functions */
static void RestoreSlotFromDisk(const char *name);
static void CreateSlotOnDisk(ReplicationSlot *slot);
-static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel);
/*
* Report shared-memory space needed by ReplicationSlotsShmemInit.
@@ -870,6 +869,7 @@ ReplicationSlotAlter(const char *name, bool failover)
ReplicationSlotRelease();
}
+
/*
* Permanently drop the currently acquired replication slot.
*/
@@ -1005,7 +1005,7 @@ ReplicationSlotSave(void)
Assert(MyReplicationSlot != NULL);
sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
- SaveSlotToPath(MyReplicationSlot, path, ERROR);
+ ReplicationSlotSaveToPath(MyReplicationSlot, path, ERROR);
}
/*
@@ -1868,7 +1868,10 @@ CheckPointReplicationSlots(bool is_shutdown)
if (!s->in_use)
continue;
- /* save the slot to disk, locking is handled in SaveSlotToPath() */
+ /*
+ * Save the slot to disk, locking is handled in
+ * ReplicationSlotSaveToPath.
+ */
sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
/*
@@ -1894,7 +1897,7 @@ CheckPointReplicationSlots(bool is_shutdown)
SpinLockRelease(&s->mutex);
}
- SaveSlotToPath(s, path, LOG);
+ ReplicationSlotSaveToPath(s, path, LOG);
}
LWLockRelease(ReplicationSlotAllocationLock);
}
@@ -1973,8 +1976,9 @@ CreateSlotOnDisk(ReplicationSlot *slot)
/*
* No need to take out the io_in_progress_lock, nobody else can see this
- * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
- * takes out the lock, if we'd take the lock here, we'd deadlock.
+ * slot yet, so nobody else will write. We're reusing
+ * ReplicationSlotSaveToPath which takes out the lock, if we'd take the
+ * lock here, we'd deadlock.
*/
sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
@@ -2000,7 +2004,7 @@ CreateSlotOnDisk(ReplicationSlot *slot)
/* Write the actual state file. */
slot->dirty = true; /* signal that we really need to write */
- SaveSlotToPath(slot, tmppath, ERROR);
+ ReplicationSlotSaveToPath(slot, tmppath, ERROR);
/* Rename the directory into place. */
if (rename(tmppath, path) != 0)
@@ -2025,8 +2029,8 @@ CreateSlotOnDisk(ReplicationSlot *slot)
/*
* Shared functionality between saving and creating a replication slot.
*/
-static void
-SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
+void
+ReplicationSlotSaveToPath(ReplicationSlot *slot, const char *dir, int elevel)
{
char tmppath[MAXPGPATH];
char path[MAXPGPATH];
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index fb79401c50..dba80ac1bb 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -229,7 +229,6 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
PG_RETURN_DATUM(result);
}
-
/*
* SQL function for dropping a replication slot.
*/
@@ -1038,3 +1037,68 @@ pg_sync_replication_slots(PG_FUNCTION_ARGS)
PG_RETURN_VOID();
}
+
+/*
+ * SQL function for altering given properties of a replication slot.
+ */
+Datum
+pg_alter_replication_slot(PG_FUNCTION_ARGS)
+{
+ Name name = PG_GETARG_NAME(0);
+ int inactive_timeout = PG_GETARG_INT32(1);
+ ReplicationSlot *slot;
+ char path[MAXPGPATH];
+
+ CheckSlotPermissions();
+
+ CheckSlotRequirements();
+
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+
+ /* Check if the slot exits with the given name. */
+ slot = SearchNamedReplicationSlot(NameStr(*name), false);
+
+ if (!slot)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("replication slot \"%s\" does not exist",
+ NameStr(*name))));
+
+ /*
+ * Do not allow users to set inactive_timeout for temporary slots because
+ * temporary, slots will not be saved to the disk.
+ */
+ if (slot->data.persistency == RS_TEMPORARY)
+ ereport(ERROR,
+ errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot set inactive_timeout for a temporary replication slot"));
+
+ LWLockRelease(ReplicationSlotControlLock);
+
+ if (inactive_timeout < 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+ errmsg("\"inactive_timeout\" must not be negative")));
+
+ /*
+ * We need to briefly prevent any other backend from acquiring the slot
+ * while we set the property. Without holding the ControlLock exclusively,
+ * a concurrent ReplicationSlotAcquire() could acquire the slot as well.
+ */
+ LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
+
+ SpinLockAcquire(&slot->mutex);
+ slot->data.inactive_timeout = inactive_timeout;
+
+ /* Make sure the invalidated state persists across server restart */
+ slot->just_dirtied = true;
+ slot->dirty = true;
+ SpinLockRelease(&slot->mutex);
+
+ sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
+ ReplicationSlotSaveToPath(slot, path, ERROR);
+
+ LWLockRelease(ReplicationSlotControlLock);
+
+ PG_RETURN_VOID();
+}
diff --git a/src/bin/pg_upgrade/t/003_logical_slots.pl b/src/bin/pg_upgrade/t/003_logical_slots.pl
index 6e82d2cb7b..b79db24f42 100644
--- a/src/bin/pg_upgrade/t/003_logical_slots.pl
+++ b/src/bin/pg_upgrade/t/003_logical_slots.pl
@@ -153,17 +153,14 @@ like(
# TEST: Successful upgrade
# Preparations for the subsequent test:
-# 1. Setup logical replication (first, cleanup slots from the previous tests,
-# and then create slot for this test with inactive_timeout set).
+# 1. Setup logical replication (first, cleanup slots from the previous tests)
my $old_connstr = $oldpub->connstr . ' dbname=postgres';
-my $inactive_timeout = 3600;
$oldpub->start;
$oldpub->safe_psql(
'postgres', qq[
SELECT * FROM pg_drop_replication_slot('test_slot1');
SELECT * FROM pg_drop_replication_slot('test_slot2');
- SELECT pg_create_logical_replication_slot(slot_name := 'regress_sub', plugin := 'pgoutput', inactive_timeout := $inactive_timeout);
CREATE PUBLICATION regress_pub FOR ALL TABLES;
]);
@@ -175,7 +172,7 @@ $sub->start;
$sub->safe_psql(
'postgres', qq[
CREATE TABLE tbl (a int);
- CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (slot_name = 'regress_sub', create_slot = false, two_phase = 'true', failover = 'true')
+ CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (two_phase = 'true', failover = 'true')
]);
$sub->wait_for_subscription_sync($oldpub, 'regress_sub');
@@ -185,6 +182,13 @@ my $twophase_query =
$sub->poll_query_until('postgres', $twophase_query)
or die "Timed out while waiting for subscriber to enable twophase";
+# Alter slot to set inactive_timeout
+my $inactive_timeout = 3600;
+$oldpub->safe_psql(
+ 'postgres', qq[
+ SELECT pg_alter_replication_slot(slot_name := 'regress_sub', inactive_timeout := $inactive_timeout);
+]);
+
# 2. Temporarily disable the subscription
$sub->safe_psql('postgres', "ALTER SUBSCRIPTION regress_sub DISABLE");
$oldpub->stop;
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index a09da44b6a..9a8134aa46 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11222,6 +11222,11 @@
proname => 'pg_sync_replication_slots', provolatile => 'v', proparallel => 'u',
prorettype => 'void', proargtypes => '',
prosrc => 'pg_sync_replication_slots' },
+{ oid => '9039', descr => 'alter given properties of a replication slot',
+ proname => 'pg_alter_replication_slot', provolatile => 'v', proparallel => 'u',
+ prorettype => 'void', proargtypes => 'name int4',
+ proargnames => '{slot_name,inactive_timeout}',
+ prosrc => 'pg_alter_replication_slot' },
# event triggers
{ oid => '3566', descr => 'list objects dropped by the current command',
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 24623cfdc1..915edf7617 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -252,6 +252,8 @@ extern void ReplicationSlotAcquire(const char *name, bool nowait);
extern void ReplicationSlotRelease(void);
extern void ReplicationSlotCleanup(void);
extern void ReplicationSlotSave(void);
+extern void ReplicationSlotSaveToPath(ReplicationSlot *slot, const char *dir,
+ int elevel);
extern void ReplicationSlotMarkDirty(void);
/* misc stuff */
--
2.34.1
[application/octet-stream] v18-0004-Allow-setting-inactive_timeout-in-the-replicatio.patch (10.5K, ../CALj2ACU6tV0bh_MiFBYU7-VK2cj8Gas73htcQjoOv0uvSEG_HA@mail.gmail.com/5-v18-0004-Allow-setting-inactive_timeout-in-the-replicatio.patch)
download | inline diff:
From a1062c4c527693c6980dc2b63c5091eed19438e7 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Sun, 24 Mar 2024 08:08:24 +0000
Subject: [PATCH v18 4/5] Allow setting inactive_timeout in the replication
commands.
This commit allows replication connections to be able to set
inactive_timeout property of the slot using replication commands
CREATE_REPLICATION_SLOT and ALTER_REPLICATION_SLOT.
Author: Bharath Rupireddy
Reviewed-by: Bertrand Drouvot, Amit Kapila, Ajin Cherian
Discussion: https://www.postgresql.org/message-id/CALj2ACW4aUe-_uFQOjdWCEN-xXoLGhmvRFnL8SNw_TZ5nJe+aw@mail.gmail.com
---
doc/src/sgml/protocol.sgml | 20 +++++++++++
src/backend/replication/slot.c | 31 +++++++++++++++--
src/backend/replication/walsender.c | 38 ++++++++++++++++----
src/include/replication/slot.h | 3 +-
src/test/recovery/t/001_stream_rep.pl | 50 +++++++++++++++++++++++++++
5 files changed, 132 insertions(+), 10 deletions(-)
diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index a5cb19357f..2ffa1b470a 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2068,6 +2068,16 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>INACTIVE_TIMEOUT [ <replaceable class="parameter">integer</replaceable> ]</literal></term>
+ <listitem>
+ <para>
+ If set to a non-zero value, specifies the amount of time in seconds
+ the slot is allowed to be inactive. The default is zero.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist>
<para>
@@ -2168,6 +2178,16 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>INACTIVE_TIMEOUT [ <replaceable class="parameter">integer</replaceable> ]</literal></term>
+ <listitem>
+ <para>
+ If set to a non-zero value, specifies the amount of time in seconds
+ the slot is allowed to be inactive. The default is zero.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist>
</listitem>
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 3287aa2860..baf0b9aa72 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -812,8 +812,10 @@ ReplicationSlotDrop(const char *name, bool nowait)
* Change the definition of the slot identified by the specified name.
*/
void
-ReplicationSlotAlter(const char *name, bool failover)
+ReplicationSlotAlter(const char *name, bool failover, int inactive_timeout)
{
+ bool lock_acquired;
+
Assert(MyReplicationSlot == NULL);
ReplicationSlotAcquire(name, false);
@@ -856,10 +858,35 @@ ReplicationSlotAlter(const char *name, bool failover)
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot enable failover for a temporary replication slot"));
- if (MyReplicationSlot->data.failover != failover)
+ /*
+ * Do not allow users to set inactive_timeout for temporary slots because
+ * temporary, slots will not be saved to the disk.
+ */
+ if (inactive_timeout > 0 && MyReplicationSlot->data.persistency == RS_TEMPORARY)
+ ereport(ERROR,
+ errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot set inactive_timeout for a temporary replication slot"));
+
+ /*
+ * If we are to change any of the slot property, acquire the lock once and
+ * for all.
+ */
+ lock_acquired = false;
+ if (MyReplicationSlot->data.failover != failover ||
+ MyReplicationSlot->data.inactive_timeout != inactive_timeout)
{
SpinLockAcquire(&MyReplicationSlot->mutex);
+ lock_acquired = true;
+ }
+
+ if (MyReplicationSlot->data.failover != failover)
MyReplicationSlot->data.failover = failover;
+
+ if (MyReplicationSlot->data.inactive_timeout != inactive_timeout)
+ MyReplicationSlot->data.inactive_timeout = inactive_timeout;
+
+ if (lock_acquired)
+ {
SpinLockRelease(&MyReplicationSlot->mutex);
ReplicationSlotMarkDirty();
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 5315c08650..0420274247 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1123,13 +1123,15 @@ static void
parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
bool *reserve_wal,
CRSSnapshotAction *snapshot_action,
- bool *two_phase, bool *failover)
+ bool *two_phase, bool *failover,
+ int *inactive_timeout)
{
ListCell *lc;
bool snapshot_action_given = false;
bool reserve_wal_given = false;
bool two_phase_given = false;
bool failover_given = false;
+ bool inactive_timeout_given = false;
/* Parse options */
foreach(lc, cmd->options)
@@ -1188,6 +1190,15 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
failover_given = true;
*failover = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "inactive_timeout") == 0)
+ {
+ if (inactive_timeout_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ inactive_timeout_given = true;
+ *inactive_timeout = defGetInt32(defel);
+ }
else
elog(ERROR, "unrecognized option: %s", defel->defname);
}
@@ -1205,6 +1216,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
bool reserve_wal = false;
bool two_phase = false;
bool failover = false;
+ int inactive_timeout = 0;
CRSSnapshotAction snapshot_action = CRS_EXPORT_SNAPSHOT;
DestReceiver *dest;
TupOutputState *tstate;
@@ -1215,13 +1227,13 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
Assert(!MyReplicationSlot);
parseCreateReplSlotOptions(cmd, &reserve_wal, &snapshot_action, &two_phase,
- &failover);
+ &failover, &inactive_timeout);
if (cmd->kind == REPLICATION_KIND_PHYSICAL)
{
ReplicationSlotCreate(cmd->slotname, false,
cmd->temporary ? RS_TEMPORARY : RS_PERSISTENT,
- false, false, false, 0);
+ false, false, false, inactive_timeout);
if (reserve_wal)
{
@@ -1252,7 +1264,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
*/
ReplicationSlotCreate(cmd->slotname, true,
cmd->temporary ? RS_TEMPORARY : RS_EPHEMERAL,
- two_phase, failover, false, 0);
+ two_phase, failover, false, inactive_timeout);
/*
* Do options check early so that we can bail before calling the
@@ -1411,9 +1423,11 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
* Process extra options given to ALTER_REPLICATION_SLOT.
*/
static void
-ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
+ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover,
+ int *inactive_timeout)
{
bool failover_given = false;
+ bool inactive_timeout_given = false;
/* Parse options */
foreach_ptr(DefElem, defel, cmd->options)
@@ -1427,6 +1441,15 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
failover_given = true;
*failover = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "inactive_timeout") == 0)
+ {
+ if (inactive_timeout_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ inactive_timeout_given = true;
+ *inactive_timeout = defGetInt32(defel);
+ }
else
elog(ERROR, "unrecognized option: %s", defel->defname);
}
@@ -1439,9 +1462,10 @@ static void
AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
{
bool failover = false;
+ int inactive_timeout = 0;
- ParseAlterReplSlotOptions(cmd, &failover);
- ReplicationSlotAlter(cmd->slotname, failover);
+ ParseAlterReplSlotOptions(cmd, &failover, &inactive_timeout);
+ ReplicationSlotAlter(cmd->slotname, failover, inactive_timeout);
}
/*
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 915edf7617..ee9b385cf9 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -246,7 +246,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
extern void ReplicationSlotPersist(void);
extern void ReplicationSlotDrop(const char *name, bool nowait);
extern void ReplicationSlotDropAcquired(void);
-extern void ReplicationSlotAlter(const char *name, bool failover);
+extern void ReplicationSlotAlter(const char *name, bool failover,
+ int inactive_timeout);
extern void ReplicationSlotAcquire(const char *name, bool nowait);
extern void ReplicationSlotRelease(void);
diff --git a/src/test/recovery/t/001_stream_rep.pl b/src/test/recovery/t/001_stream_rep.pl
index 5311ade509..db00b6aa24 100644
--- a/src/test/recovery/t/001_stream_rep.pl
+++ b/src/test/recovery/t/001_stream_rep.pl
@@ -604,4 +604,54 @@ ok( pump_until(
'base backup cleanly canceled');
$sigchld_bb->finish();
+# Drop any existing slots on the primary, for the follow-up tests.
+$node_primary->safe_psql('postgres',
+ "SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots;");
+
+# Test setting inactive_timeout option via replication commands.
+$node_primary->append_conf(
+ 'postgresql.conf', qq(
+wal_level = logical
+));
+$node_primary->restart;
+
+$node_primary->psql(
+ 'postgres',
+ "CREATE_REPLICATION_SLOT it_phy_slot1 PHYSICAL (RESERVE_WAL, INACTIVE_TIMEOUT 100);",
+ extra_params => [ '-d', $connstr_db ]);
+
+$node_primary->psql(
+ 'postgres',
+ "CREATE_REPLICATION_SLOT it_phy_slot2 PHYSICAL (RESERVE_WAL);",
+ extra_params => [ '-d', $connstr_db ]);
+
+$node_primary->psql(
+ 'postgres',
+ "ALTER_REPLICATION_SLOT it_phy_slot2 (INACTIVE_TIMEOUT 200);",
+ extra_params => [ '-d', $connstr_db ]);
+
+$node_primary->psql(
+ 'postgres',
+ "CREATE_REPLICATION_SLOT it_log_slot1 LOGICAL pgoutput (TWO_PHASE, INACTIVE_TIMEOUT 300);",
+ extra_params => [ '-d', $connstr_db ]);
+
+$node_primary->psql(
+ 'postgres',
+ "CREATE_REPLICATION_SLOT it_log_slot2 LOGICAL pgoutput;",
+ extra_params => [ '-d', $connstr_db ]);
+
+$node_primary->psql(
+ 'postgres',
+ "ALTER_REPLICATION_SLOT it_log_slot2 (INACTIVE_TIMEOUT 400);",
+ extra_params => [ '-d', $connstr_db ]);
+
+my $slot_info_expected = 'it_log_slot1|logical|300
+it_log_slot2|logical|400
+it_phy_slot1|physical|100
+it_phy_slot2|physical|0';
+
+my $slot_info = $node_primary->safe_psql('postgres',
+ qq[SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;]);
+is($slot_info, $slot_info_expected, "replication slots with inactive_timeout on primary exist");
+
done_testing();
--
2.34.1
[application/octet-stream] v18-0005-Add-inactive_timeout-based-replication-slot-inva.patch (26.2K, ../CALj2ACU6tV0bh_MiFBYU7-VK2cj8Gas73htcQjoOv0uvSEG_HA@mail.gmail.com/6-v18-0005-Add-inactive_timeout-based-replication-slot-inva.patch)
download | inline diff:
From 2e1a8cba688291cf9150f0abbce89273832a644d Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Sun, 24 Mar 2024 08:48:41 +0000
Subject: [PATCH v18 5/5] Add inactive_timeout based replication slot
invalidation.
Till now, postgres has the ability to invalidate inactive
replication slots based on the amount of WAL (set via
max_slot_wal_keep_size GUC) that will be needed for the slots in
case they become active. However, choosing a default value for
max_slot_wal_keep_size is tricky. Because the amount of WAL a
customer generates, and their allocated storage will vary greatly
in production, making it difficult to pin down a one-size-fits-all
value. It is often easy for developers to set a timeout of say 1
or 2 or 3 days at slot level, after which the inactive slots get
dropped.
To achieve the above, postgres uses replication slot property
last_inactive_time (the time at which the slot became inactive),
and a new slot level parameter inactive_timeout and finds an
opportunity to invalidate the slot based on this new mechanism.
The invalidation check happens at various locations to help
being as latest as possible, these locations include the
following:
- Whenever the slot is acquired if the slot
gets invalidated due to this new mechanism, an error is
emitted.
- During checkpoint.
- Whenver pg_get_replication_slots() is called.
Note that this new invalidation mechanism won't kick-in for the
slots that are currently being synced from the primary to the
standby.
Author: Bharath Rupireddy
Reviewed-by: Bertrand Drouvot, Amit Kapila
Discussion: https://www.postgresql.org/message-id/CALj2ACW4aUe-_uFQOjdWCEN-xXoLGhmvRFnL8SNw_TZ5nJe+aw@mail.gmail.com
---
doc/src/sgml/func.sgml | 12 +-
doc/src/sgml/system-views.sgml | 10 +-
.../replication/logical/logicalfuncs.c | 2 +-
src/backend/replication/logical/slotsync.c | 4 +-
src/backend/replication/slot.c | 184 +++++++++++++++++-
src/backend/replication/slotfuncs.c | 19 +-
src/backend/replication/walsender.c | 4 +-
src/backend/utils/adt/pg_upgrade_support.c | 2 +-
src/include/replication/slot.h | 9 +-
src/test/recovery/meson.build | 1 +
src/test/recovery/t/050_invalidate_slots.pl | 170 ++++++++++++++++
11 files changed, 395 insertions(+), 22 deletions(-)
create mode 100644 src/test/recovery/t/050_invalidate_slots.pl
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index 22c8e0d39c..4826e45c7d 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -28393,8 +28393,8 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
released upon any error. The optional fourth
parameter, <parameter>inactive_timeout</parameter>, when set to a
non-zero value, specifies the amount of time in seconds the slot is
- allowed to be inactive. This function corresponds to the replication
- protocol command
+ allowed to be inactive before getting invalidated.
+ This function corresponds to the replication protocol command
<literal>CREATE_REPLICATION_SLOT ... PHYSICAL</literal>.
</para></entry>
</row>
@@ -28439,12 +28439,12 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
<parameter>failover</parameter>, when set to true,
specifies that this slot is enabled to be synced to the
standbys so that logical replication can be resumed after
- failover. The optional sixth parameter,
+ failover. The optional sixth parameter,
<parameter>inactive_timeout</parameter>, when set to a
non-zero value, specifies the amount of time in seconds the slot is
- allowed to be inactive. A call to this function has the same effect as
- the replication protocol command
- <literal>CREATE_REPLICATION_SLOT ... LOGICAL</literal>.
+ allowed to be inactive before getting invalidated.
+ A call to this function has the same effect as the replication protocol
+ command <literal>CREATE_REPLICATION_SLOT ... LOGICAL</literal>.
</para></entry>
</row>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index dddbaa070f..1722609d39 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2539,7 +2539,8 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
<structfield>inactive_timeout</structfield> <type>integer</type>
</para>
<para>
- The amount of time in seconds the slot is allowed to be inactive.
+ The amount of time in seconds the slot is allowed to be inactive
+ before getting invalidated.
</para></entry>
</row>
@@ -2583,6 +2584,13 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
perform logical decoding. It is set only for logical slots.
</para>
</listitem>
+ <listitem>
+ <para>
+ <literal>inactive_timeout</literal> means that the slot has been
+ inactive for the duration specified by slot's
+ <literal>inactive_timeout</literal> parameter.
+ </para>
+ </listitem>
</itemizedlist>
</para></entry>
</row>
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index b4dd5cce75..56fc1a45a9 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -197,7 +197,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
else
end_of_wal = GetXLogReplayRecPtr(NULL);
- ReplicationSlotAcquire(NameStr(*name), true);
+ ReplicationSlotAcquire(NameStr(*name), true, true);
PG_TRY();
{
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index c01876ceeb..7f1ffab23c 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -319,7 +319,7 @@ drop_local_obsolete_slots(List *remote_slot_list)
if (synced_slot)
{
- ReplicationSlotAcquire(NameStr(local_slot->data.name), true);
+ ReplicationSlotAcquire(NameStr(local_slot->data.name), true, false);
ReplicationSlotDropAcquired();
}
@@ -529,7 +529,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
* InvalidatePossiblyObsoleteSlot() where it invalidates slot directly
* if the slot is not acquired by other processes.
*/
- ReplicationSlotAcquire(remote_slot->name, true);
+ ReplicationSlotAcquire(remote_slot->name, true, false);
Assert(slot == MyReplicationSlot);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index baf0b9aa72..fae61020c4 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -107,10 +107,11 @@ const char *const SlotInvalidationCauses[] = {
[RS_INVAL_WAL_REMOVED] = "wal_removed",
[RS_INVAL_HORIZON] = "rows_removed",
[RS_INVAL_WAL_LEVEL] = "wal_level_insufficient",
+ [RS_INVAL_INACTIVE_TIMEOUT] = "inactive_timeout",
};
/* Maximum number of invalidation causes */
-#define RS_INVAL_MAX_CAUSES RS_INVAL_WAL_LEVEL
+#define RS_INVAL_MAX_CAUSES RS_INVAL_INACTIVE_TIMEOUT
StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1),
"array length mismatch");
@@ -158,6 +159,9 @@ static XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr;
static void ReplicationSlotShmemExit(int code, Datum arg);
static void ReplicationSlotDropPtr(ReplicationSlot *slot);
+static bool InvalidateSlotForInactiveTimeout(ReplicationSlot *slot,
+ bool need_control_lock,
+ bool need_mutex);
/* internal persistency functions */
static void RestoreSlotFromDisk(const char *name);
@@ -550,9 +554,14 @@ ReplicationSlotName(int index, Name name)
*
* An error is raised if nowait is true and the slot is currently in use. If
* nowait is false, we sleep until the slot is released by the owning process.
+ *
+ * If check_for_invalidation is true, the slot is checked for invalidation
+ * based on its inactive_timeout parameter and an error is raised after making
+ * the slot ours.
*/
void
-ReplicationSlotAcquire(const char *name, bool nowait)
+ReplicationSlotAcquire(const char *name, bool nowait,
+ bool check_for_invalidation)
{
ReplicationSlot *s;
int active_pid;
@@ -630,6 +639,42 @@ retry:
/* We made this slot active, so it's ours now. */
MyReplicationSlot = s;
+ /*
+ * Check if the given slot can be invalidated based on its
+ * inactive_timeout parameter. If yes, persist the invalidated state to
+ * disk and then error out. We do this only after making the slot ours to
+ * avoid anyone else acquiring it while we check for its invalidation.
+ */
+ if (check_for_invalidation)
+ {
+ /* The slot is ours by now */
+ Assert(s->active_pid == MyProcPid);
+
+ /*
+ * Well, the slot is not yet ours really unless we check for the
+ * invalidation below.
+ */
+ s->active_pid = 0;
+ if (InvalidateReplicationSlotForInactiveTimeout(s, true, true, true))
+ {
+ /*
+ * If the slot has been invalidated, recalculate the resource
+ * limits.
+ */
+ ReplicationSlotsComputeRequiredXmin(false);
+ ReplicationSlotsComputeRequiredLSN();
+
+ /* Might need it for slot clean up on error, so restore it */
+ s->active_pid = MyProcPid;
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot acquire invalidated replication slot \"%s\"",
+ NameStr(MyReplicationSlot->data.name)),
+ errdetail("This slot has been invalidated because of its inactive_timeout parameter.")));
+ }
+ s->active_pid = MyProcPid;
+ }
+
/*
* The call to pgstat_acquire_replslot() protects against stats for a
* different slot, from before a restart or such, being present during
@@ -793,7 +838,7 @@ ReplicationSlotDrop(const char *name, bool nowait)
{
Assert(MyReplicationSlot == NULL);
- ReplicationSlotAcquire(name, nowait);
+ ReplicationSlotAcquire(name, nowait, false);
/*
* Do not allow users to drop the slots which are currently being synced
@@ -818,7 +863,7 @@ ReplicationSlotAlter(const char *name, bool failover, int inactive_timeout)
Assert(MyReplicationSlot == NULL);
- ReplicationSlotAcquire(name, false);
+ ReplicationSlotAcquire(name, false, true);
if (SlotIsPhysical(MyReplicationSlot))
ereport(ERROR,
@@ -1546,6 +1591,9 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
case RS_INVAL_WAL_LEVEL:
appendStringInfoString(&err_detail, _("Logical decoding on standby requires wal_level >= logical on the primary server."));
break;
+ case RS_INVAL_INACTIVE_TIMEOUT:
+ appendStringInfoString(&err_detail, _("The slot has been inactive for more than the time specified by slot's inactive_timeout parameter."));
+ break;
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -1659,6 +1707,10 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
if (SlotIsLogical(s))
invalidation_cause = cause;
break;
+ case RS_INVAL_INACTIVE_TIMEOUT:
+ if (InvalidateReplicationSlotForInactiveTimeout(s, false, false, false))
+ invalidation_cause = cause;
+ break;
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -1812,6 +1864,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
* - RS_INVAL_HORIZON: requires a snapshot <= the given horizon in the given
* db; dboid may be InvalidOid for shared relations
* - RS_INVAL_WAL_LEVEL: is logical
+ * - RS_INVAL_INACTIVE_TIMEOUT: inactive slot timeout occurs
*
* NB - this runs as part of checkpoint, so avoid raising errors if possible.
*/
@@ -1863,6 +1916,109 @@ restart:
return invalidated;
}
+/*
+ * Invalidate given slot based on its inactive_timeout parameter.
+ *
+ * Returns true if the slot has got invalidated.
+ *
+ * NB - this function also runs as part of checkpoint, so avoid raising errors
+ * if possible.
+ */
+bool
+InvalidateReplicationSlotForInactiveTimeout(ReplicationSlot *slot,
+ bool need_control_lock,
+ bool need_mutex,
+ bool persist_state)
+{
+ if (!InvalidateSlotForInactiveTimeout(slot, need_control_lock, need_mutex))
+ return false;
+
+ Assert(slot->active_pid == 0);
+
+ SpinLockAcquire(&slot->mutex);
+ slot->data.invalidated = RS_INVAL_INACTIVE_TIMEOUT;
+
+ /* Make sure the invalidated state persists across server restart */
+ slot->just_dirtied = true;
+ slot->dirty = true;
+ SpinLockRelease(&slot->mutex);
+
+ if (persist_state)
+ {
+ char path[MAXPGPATH];
+
+ sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
+ ReplicationSlotSaveToPath(slot, path, ERROR);
+ }
+
+ ReportSlotInvalidation(RS_INVAL_INACTIVE_TIMEOUT, false, 0,
+ slot->data.name, InvalidXLogRecPtr,
+ InvalidXLogRecPtr, InvalidTransactionId);
+
+ return true;
+}
+
+/*
+ * Helper for InvalidateReplicationSlotForInactiveTimeout
+ */
+static bool
+InvalidateSlotForInactiveTimeout(ReplicationSlot *slot,
+ bool need_control_lock,
+ bool need_mutex)
+{
+ ReplicationSlotInvalidationCause inavidation_cause = RS_INVAL_NONE;
+
+ if (slot->last_inactive_time == 0 ||
+ slot->data.inactive_timeout == 0)
+ return false;
+
+ /* inactive_timeout is only tracked for permanent slots */
+ if (slot->data.persistency != RS_PERSISTENT)
+ return false;
+
+ /*
+ * Do not invalidate the slots which are currently being synced from the
+ * primary to the standby.
+ */
+ if (RecoveryInProgress() && slot->data.synced)
+ return false;
+
+ if (need_control_lock)
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+
+ Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
+
+ /*
+ * Check if the slot needs to be invalidated due to inactive_timeout. We
+ * do this with the spinlock held to avoid race conditions -- for example
+ * the restart_lsn could move forward, or the slot could be dropped.
+ */
+ if (need_mutex)
+ SpinLockAcquire(&slot->mutex);
+
+ if (slot->last_inactive_time > 0 &&
+ slot->data.inactive_timeout > 0)
+ {
+ TimestampTz now;
+
+ /* last_inactive_time is only tracked for inactive slots */
+ Assert(slot->active_pid == 0);
+
+ now = GetCurrentTimestamp();
+ if (TimestampDifferenceExceeds(slot->last_inactive_time, now,
+ slot->data.inactive_timeout * 1000))
+ inavidation_cause = RS_INVAL_INACTIVE_TIMEOUT;
+ }
+
+ if (need_mutex)
+ SpinLockRelease(&slot->mutex);
+
+ if (need_control_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+
+ return (inavidation_cause == RS_INVAL_INACTIVE_TIMEOUT);
+}
+
/*
* Flush all replication slots to disk.
*
@@ -1875,6 +2031,7 @@ void
CheckPointReplicationSlots(bool is_shutdown)
{
int i;
+ bool invalidated = false;
elog(DEBUG1, "performing replication slot checkpoint");
@@ -1896,10 +2053,11 @@ CheckPointReplicationSlots(bool is_shutdown)
continue;
/*
- * Save the slot to disk, locking is handled in
- * ReplicationSlotSaveToPath.
+ * Here's an opportunity to invalidate inactive replication slots
+ * based on timeout, so let's do it.
*/
- sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
+ if (InvalidateReplicationSlotForInactiveTimeout(s, true, true, false))
+ invalidated = true;
/*
* Slot's data is not flushed each time the confirmed_flush LSN is
@@ -1924,9 +2082,21 @@ CheckPointReplicationSlots(bool is_shutdown)
SpinLockRelease(&s->mutex);
}
+ /*
+ * Save the slot to disk, locking is handled in
+ * ReplicationSlotSaveToPath.
+ */
+ sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
ReplicationSlotSaveToPath(s, path, LOG);
}
LWLockRelease(ReplicationSlotAllocationLock);
+
+ /* If the slot has been invalidated, recalculate the resource limits */
+ if (invalidated)
+ {
+ ReplicationSlotsComputeRequiredXmin(false);
+ ReplicationSlotsComputeRequiredLSN();
+ }
}
/*
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index dba80ac1bb..aadba68c11 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -257,6 +257,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
XLogRecPtr currlsn;
int slotno;
+ bool invalidated = false;
/*
* We don't require any special permission to see this function's data
@@ -287,6 +288,13 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
slot_contents = *slot;
SpinLockRelease(&slot->mutex);
+ /*
+ * Here's an opportunity to invalidate inactive replication slots
+ * based on timeout, so let's do it.
+ */
+ if (InvalidateReplicationSlotForInactiveTimeout(slot, false, true, true))
+ invalidated = true;
+
memset(values, 0, sizeof(values));
memset(nulls, 0, sizeof(nulls));
@@ -465,6 +473,15 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
LWLockRelease(ReplicationSlotControlLock);
+ /*
+ * If the slot has been invalidated, recalculate the resource limits
+ */
+ if (invalidated)
+ {
+ ReplicationSlotsComputeRequiredXmin(false);
+ ReplicationSlotsComputeRequiredLSN();
+ }
+
return (Datum) 0;
}
@@ -667,7 +684,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
/* Acquire the slot so we "own" it */
- ReplicationSlotAcquire(NameStr(*slotname), true);
+ ReplicationSlotAcquire(NameStr(*slotname), true, true);
/* A slot whose restart_lsn has never been reserved cannot be advanced */
if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 0420274247..aa886412a5 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -846,7 +846,7 @@ StartReplication(StartReplicationCmd *cmd)
if (cmd->slotname)
{
- ReplicationSlotAcquire(cmd->slotname, true);
+ ReplicationSlotAcquire(cmd->slotname, true, true);
if (SlotIsLogical(MyReplicationSlot))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
@@ -1483,7 +1483,7 @@ StartLogicalReplication(StartReplicationCmd *cmd)
Assert(!MyReplicationSlot);
- ReplicationSlotAcquire(cmd->slotname, true);
+ ReplicationSlotAcquire(cmd->slotname, true, true);
/*
* Force a disconnect, so that the decoding code doesn't need to care
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index c54b08fe18..82956d58d3 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -299,7 +299,7 @@ binary_upgrade_logical_slot_has_caught_up(PG_FUNCTION_ARGS)
slot_name = PG_GETARG_NAME(0);
/* Acquire the given slot */
- ReplicationSlotAcquire(NameStr(*slot_name), true);
+ ReplicationSlotAcquire(NameStr(*slot_name), true, false);
Assert(SlotIsLogical(MyReplicationSlot));
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index ee9b385cf9..00ff8e5ef5 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -53,6 +53,8 @@ typedef enum ReplicationSlotInvalidationCause
RS_INVAL_HORIZON,
/* wal_level insufficient for slot */
RS_INVAL_WAL_LEVEL,
+ /* inactive slot timeout has occurred */
+ RS_INVAL_INACTIVE_TIMEOUT,
} ReplicationSlotInvalidationCause;
extern PGDLLIMPORT const char *const SlotInvalidationCauses[];
@@ -249,7 +251,8 @@ extern void ReplicationSlotDropAcquired(void);
extern void ReplicationSlotAlter(const char *name, bool failover,
int inactive_timeout);
-extern void ReplicationSlotAcquire(const char *name, bool nowait);
+extern void ReplicationSlotAcquire(const char *name, bool nowait,
+ bool check_for_invalidation);
extern void ReplicationSlotRelease(void);
extern void ReplicationSlotCleanup(void);
extern void ReplicationSlotSave(void);
@@ -270,6 +273,10 @@ extern bool InvalidateObsoleteReplicationSlots(ReplicationSlotInvalidationCause
XLogSegNo oldestSegno,
Oid dboid,
TransactionId snapshotConflictHorizon);
+extern bool InvalidateReplicationSlotForInactiveTimeout(ReplicationSlot *slot,
+ bool need_control_lock,
+ bool need_mutex,
+ bool persist_state);
extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
extern int ReplicationSlotIndex(ReplicationSlot *slot);
extern bool ReplicationSlotName(int index, Name name);
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index b1eb77b1ec..708a2a3798 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -51,6 +51,7 @@ tests += {
't/040_standby_failover_slots_sync.pl',
't/041_checkpoint_at_promote.pl',
't/042_low_level_backup.pl',
+ 't/050_invalidate_slots.pl',
],
},
}
diff --git a/src/test/recovery/t/050_invalidate_slots.pl b/src/test/recovery/t/050_invalidate_slots.pl
new file mode 100644
index 0000000000..77499dde07
--- /dev/null
+++ b/src/test/recovery/t/050_invalidate_slots.pl
@@ -0,0 +1,170 @@
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test for replication slots invalidation
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Utils;
+use PostgreSQL::Test::Cluster;
+use Test::More;
+use Time::HiRes qw(usleep);
+
+# Check for invalidation of slot in server log.
+sub check_slots_invalidation_in_server_log
+{
+ my ($node, $slot_name, $offset) = @_;
+ my $invalidated = 0;
+
+ for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++)
+ {
+ $node->safe_psql('postgres', "CHECKPOINT");
+ if ($node->log_contains(
+ "invalidating obsolete replication slot \"$slot_name\"", $offset))
+ {
+ $invalidated = 1;
+ last;
+ }
+ usleep(100_000);
+ }
+ ok($invalidated, "check that slot $slot_name invalidation has been logged");
+}
+
+# =============================================================================
+# Testcase start: Invalidate streaming standby's slot due to inactive_timeout
+#
+
+# Initialize primary node
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 'logical');
+
+# Avoid checkpoint during the test, otherwise, the test can get unpredictable
+$primary->append_conf(
+ 'postgresql.conf', q{
+checkpoint_timeout = 1h
+autovacuum = off
+});
+$primary->start;
+
+# Take backup
+my $backup_name = 'my_backup';
+$primary->backup($backup_name);
+
+# Create a standby linking to the primary using the replication slot
+my $standby1 = PostgreSQL::Test::Cluster->new('standby1');
+$standby1->init_from_backup($primary, $backup_name, has_streaming => 1);
+$standby1->append_conf(
+ 'postgresql.conf', q{
+primary_slot_name = 'sb1_slot'
+});
+
+# Set timeout so that the slot when inactive will get invalidated after the
+# timeout.
+my $inactive_timeout = 1;
+$primary->safe_psql(
+ 'postgres', qq[
+ SELECT pg_create_physical_replication_slot(slot_name := 'sb1_slot', inactive_timeout := $inactive_timeout);
+]);
+
+$standby1->start;
+
+# Wait until standby has replayed enough data
+$primary->wait_for_catchup($standby1);
+
+# Check inactive_timeout is what we've set above
+my $result = $primary->safe_psql(
+ 'postgres', qq[
+ SELECT inactive_timeout = $inactive_timeout
+ FROM pg_replication_slots WHERE slot_name = 'sb1_slot';
+]);
+is($result, "t",
+ 'check the inactive replication slot info for an active slot');
+
+my $logstart = -s $primary->logfile;
+
+# Stop standby to make the replication slot on primary inactive
+$standby1->stop;
+
+# Wait for the inactive replication slot info to be updated
+$primary->poll_query_until(
+ 'postgres', qq[
+ SELECT COUNT(slot_name) = 1 FROM pg_replication_slots
+ WHERE last_inactive_time IS NOT NULL
+ AND slot_name = 'sb1_slot'
+ AND inactive_timeout = $inactive_timeout;
+])
+ or die
+ "Timed out while waiting for inactive replication slot info to be updated";
+
+check_slots_invalidation_in_server_log($primary, 'sb1_slot', $logstart);
+
+# Wait for the inactive replication slots to be invalidated.
+$primary->poll_query_until(
+ 'postgres', qq[
+ SELECT COUNT(slot_name) = 1 FROM pg_replication_slots
+ WHERE slot_name = 'sb1_slot' AND
+ invalidation_reason = 'inactive_timeout';
+])
+ or die
+ "Timed out while waiting for inactive replication slot sb1_slot to be invalidated";
+
+# Testcase end: Invalidate streaming standby's slot due to inactive_timeout
+# =============================================================================
+
+# =============================================================================
+# Testcase start: Invalidate logical subscriber's slot due to inactive_timeout
+my $publisher = $primary;
+
+# Create subscriber node
+my $subscriber = PostgreSQL::Test::Cluster->new('sub');
+$subscriber->init;
+$subscriber->start;
+
+# Create tables
+$publisher->safe_psql('postgres', "CREATE TABLE test_tbl (id int)");
+$subscriber->safe_psql('postgres', "CREATE TABLE test_tbl (id int)");
+
+# Insert some data
+$subscriber->safe_psql('postgres',
+ "INSERT INTO test_tbl VALUES (generate_series(1, 5));");
+
+# Setup logical replication
+my $publisher_connstr = $publisher->connstr . ' dbname=postgres';
+$publisher->safe_psql('postgres', "CREATE PUBLICATION pub FOR ALL TABLES");
+$subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION sub CONNECTION '$publisher_connstr' PUBLICATION pub WITH (slot_name = 'lsub1_slot')"
+);
+
+$subscriber->wait_for_subscription_sync($publisher, 'sub');
+
+$result = $subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tbl");
+
+is($result, qq(5), "check initial copy was done");
+
+# Alter slot to set inactive_timeout
+$publisher->safe_psql(
+ 'postgres', qq[
+ SELECT pg_alter_replication_slot(slot_name := 'lsub1_slot', inactive_timeout := $inactive_timeout);
+]);
+
+$logstart = -s $publisher->logfile;
+
+# Stop subscriber to make the replication slot on publisher inactive
+$subscriber->stop;
+
+# Wait for the inactive replication slot info to be updated
+$publisher->poll_query_until(
+ 'postgres', qq[
+ SELECT COUNT(slot_name) = 1 FROM pg_replication_slots
+ WHERE last_inactive_time IS NOT NULL
+ AND slot_name = 'lsub1_slot'
+ AND inactive_timeout = $inactive_timeout;
+])
+ or die
+ "Timed out while waiting for inactive replication slot info to be updated";
+
+check_slots_invalidation_in_server_log($publisher, 'lsub1_slot', $logstart);
+
+# Testcase end: Invalidate logical subscriber's slot due to inactive_timeout
+# =============================================================================
+
+done_testing();
--
2.34.1
view thread (43+ messages) latest in thread
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]
Subject: Re: Introduce XID age and inactive timeout based replication slot invalidation
In-Reply-To: <CALj2ACU6tV0bh_MiFBYU7-VK2cj8Gas73htcQjoOv0uvSEG_HA@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