public inbox for [email protected]
help / color / mirror / Atom feedFrom: Nisha Moond <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: vignesh C <[email protected]>
Cc: Hayato Kuroda (Fujitsu) <[email protected]>
Cc: shveta malik <[email protected]>
Cc: Bharath Rupireddy <[email protected]>
Cc: Ajin Cherian <[email protected]>
Cc: Bertrand Drouvot <[email protected]>
Cc: Masahiko Sawada <[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: Fri, 31 Jan 2025 07:43:16 +0530
Message-ID: <CABdArM5oXMO63uETfEFLEan3UfNBiDwprdoyYYdeWP9fsYNg9A@mail.gmail.com> (raw)
In-Reply-To: <CABdArM6FKY6GjVu=7ak8CAYNK0e329h-JEVa3p0UE5niwP_=yA@mail.gmail.com>
References: <CALDaNm39dtWHoQej2wd74zYGphA0e54Rx6aTRoCEnMXZ5ysbow@mail.gmail.com>
<CABdArM45=j=DAFTEZA8p=oqYo-XqRwhaPCdKpciQCJx+jNeTJg@mail.gmail.com>
<TYAPR01MB5692B7687EE7981AA91BA5B9F5362@TYAPR01MB5692.jpnprd01.prod.outlook.com>
<CABdArM4N60XShbmxXgz4z8ENQGPOMhXSKzvmskdv8V=mQsYOOA@mail.gmail.com>
<CALDaNm0mTWwg0z4v-sorq08S2CdZmL2s+rh4nHpWeJaBQ2F+mg@mail.gmail.com>
<CAHut+Psi-mso-qSyMkHRP8e+psDFFmMp4NZTiL_yrFWidHkGMw@mail.gmail.com>
<CALDaNm3znxKvTv=MDmMBOSk6XurkKgYyh9cH49NLL7SmJ62Q_A@mail.gmail.com>
<CABdArM7CPR0Ag6OgtufATUn+x=Q_h3cdBcc39r+xeKYDks5Mfw@mail.gmail.com>
<CALDaNm1FQS04aG0C0gCRpvi-o-OTdq91y6Az34YKN-dVc9r5Ng@mail.gmail.com>
<CABdArM7LJopsdDyBAhjbsW3K-Yq=f75bnjaU5HRSsTkGWXibaQ@mail.gmail.com>
<CAHut+PvfSHzKgXJTd=PbJieaXFgR9+VhhR3+eb7jCFFj6kC5bA@mail.gmail.com>
<CABdArM6qPsZE9v7qymNTwiRt68DgOJ+LsAMhweTxZx1xr1cVOA@mail.gmail.com>
<CAA4eK1JO+3Q+wkyjhMK+h_5MveTLFY7q2KRbrg1ojtO6dLCaMw@mail.gmail.com>
<CABdArM4wO2gPfcjrFWWL=D18PyFeWJZJcJGY3uJMzqL+4LDGpw@mail.gmail.com>
<CAHut+Pt7hYB3kzNgmKJ16z46p9M+4U94k=b=kydPhmZtCtBYHA@mail.gmail.com>
<CAA4eK1+YyUWkR6WvNzVjgU+QTYtLFo0uofr5d3DKwkLcaOfAsQ@mail.gmail.com>
<CABdArM6FKY6GjVu=7ak8CAYNK0e329h-JEVa3p0UE5niwP_=yA@mail.gmail.com>
On Tue, Jan 28, 2025 at 5:28 PM Nisha Moond <[email protected]> wrote:
>
> On Tue, Jan 28, 2025 at 3:26 PM Amit Kapila <[email protected]> wrote:
> >
> > On Mon, Dec 30, 2024 at 11:05 AM Peter Smith <[email protected]> wrote:
> > >
> > > I think we are often too quick to throw out perfectly good tests.
> > > Citing that some similar GUCs don't do testing as a reason to skip
> > > them just seems to me like an example of "two wrongs don't make a
> > > right".
> > >
> > > There is a third option.
> > >
> > > Keep the tests. Because they take excessive time to run, that simply
> > > means you should run them *conditionally* based on the PG_TEST_EXTRA
> > > environment variable so they don't impact the normal BF execution. The
> > > documentation [1] says this env var is for "resource intensive" tests
> > > -- AFAIK this is exactly the scenario we find ourselves in, so is
> > > exactly what this env var was meant for.
> > >
> > > Search other *.pl tests for PG_TEST_EXTRA to see some examples.
> > >
> >
> > I don't see the long-running tests to be added under PG_TEST_EXTRA as
> > that will make it unusable after some point. Now, if multiple senior
> > members feel it is okay to add long-running tests under PG_TEST_EXTRA
> > then I am open to considering it. We can keep this test as a separate
> > patch so that the patch is being tested in CI or in manual tests
> > before commit.
> >
>
> Please find the attached v64 patches. The changes in this version
> w.r.t. older patch v63 are as -
> - The changes from the v63-0001 patch have been moved to a separate thread [1].
> - The v63-0002 patch has been split into two parts in v64:
> 1) 001 patch: Implements the main feature - inactive timeout-based
> slot invalidation.
> 2) 002 patch: Separates the TAP test "044_invalidate_inactive_slots"
> as suggested above.
>
> [1] https://www.postgresql.org/message-id/CABdArM6pBL5hPnSQ%2B5nEVMANcF4FCH7LQmgskXyiLY75TMnKpw%40mail.g...
>
Please find the v65 patch set attached with following changes:
- patch-0001 is the copy of v4-001 patch used as a base patch from
[1], as suggested by Peter in [2].
- patch-0002 is the main patch implementing the feature, this has
also addressed the comments from [2] and [3]
- patch-0003 adds an alternative approach for the TAP test using
injection points to force idle_timeout slot invalidation without
waiting for a minute. This test takes 2-3 seconds to complete.
- patch-0004 maintains the previous test(v64-0002) which is under
PG_TEST_EXTRA. Also, addressed Vignesh's comments [4] for the test and
now it takes 70-80 seconds to complete.
Note: Patches 0003 and 0004 contain the same TAP test but use
different verification methods. We need to decide which one to keep.
[1] https://www.postgresql.org/message-id/CALDaNm1ZUHnKm%2BPSjjqRrMxcLagrUTS6SADnEsQBfW8rMZFrDA%40mail.g...
[2] https://www.postgresql.org/message-id/CAHut%2BPtyUQGee6pHkNN3-ghYhWnY5p-3yWumK7zKupu0S1oVQQ%40mail.g...
[3] https://www.postgresql.org/message-id/CALDaNm1J_mdqCYjQZgfQMVhJrxndPem5ruxpG_67t4C_2My9WQ%40mail.gma...
[4] https://www.postgresql.org/message-id/CALDaNm2dAJB%3DfJ2X7EMb7meNTjMyL-%2B-xA93JL_jPkGF4%3DRUYw%40ma...
Thank you, Kuroda-san for providing the TAP test using injection
points (patch-0003).
--
Thanks,
Nisha
Attachments:
[application/octet-stream] v65-0001-Raise-Error-for-Invalid-Slots-in-ReplicationSlot.patch (13.6K, ../CABdArM5oXMO63uETfEFLEan3UfNBiDwprdoyYYdeWP9fsYNg9A@mail.gmail.com/2-v65-0001-Raise-Error-for-Invalid-Slots-in-ReplicationSlot.patch)
download | inline diff:
From aa47b7d71a12d18ced352e3771055753b993a122 Mon Sep 17 00:00:00 2001
From: Vignesh <[email protected]>
Date: Thu, 30 Jan 2025 18:15:15 +0530
Subject: [PATCH v65 1/4] Raise Error for Invalid Slots in
ReplicationSlotAcquire()
Once a replication slot is invalidated, it cannot be reused. However, a
process could still acquire an invalid slot and fail later.
For example, if a process acquires a logical slot that was invalidated due
to wal_removed, it will eventually fail in CreateDecodingContext() when
attempting to access the removed WAL. Similarly, for physical replication
slots, even if the slot is invalidated and invalidation_reason is set to
wal_removed, the walsender does not currently check for invalidation when
starting physical replication. Instead, replication starts, and an error
is only reported later by the standby when a missing WAL is detected.
This patch improves error handling by detecting invalid slots earlier.
If error_if_invalid=true is specified when calling ReplicationSlotAcquire(),
an error will be raised immediately instead of letting the process acquire the
slot and fail later due to the invalidated slot.
---
src/backend/replication/logical/logical.c | 20 ------------
.../replication/logical/logicalfuncs.c | 2 +-
src/backend/replication/logical/slotsync.c | 4 +--
src/backend/replication/slot.c | 32 +++++++++++--------
src/backend/replication/slotfuncs.c | 2 +-
src/backend/replication/walsender.c | 4 +--
src/backend/utils/adt/pg_upgrade_support.c | 2 +-
src/include/replication/slot.h | 3 +-
src/test/recovery/t/019_replslot_limit.pl | 2 +-
.../t/035_standby_logical_decoding.pl | 15 ++++-----
10 files changed, 35 insertions(+), 51 deletions(-)
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 0b25efafe2..2c8cf516bd 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -542,26 +542,6 @@ CreateDecodingContext(XLogRecPtr start_lsn,
errdetail("This replication slot is being synchronized from the primary server."),
errhint("Specify another replication slot."));
- /*
- * Check if slot has been invalidated due to max_slot_wal_keep_size. Avoid
- * "cannot get changes" wording in this errmsg because that'd be
- * confusingly ambiguous about no changes being available when called from
- * pg_logical_slot_get_changes_guts().
- */
- if (MyReplicationSlot->data.invalidated == RS_INVAL_WAL_REMOVED)
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("can no longer get changes from replication slot \"%s\"",
- NameStr(MyReplicationSlot->data.name)),
- errdetail("This slot has been invalidated because it exceeded the maximum reserved size.")));
-
- if (MyReplicationSlot->data.invalidated != RS_INVAL_NONE)
- ereport(ERROR,
- (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("can no longer get changes from replication slot \"%s\"",
- NameStr(MyReplicationSlot->data.name)),
- errdetail("This slot has been invalidated because it was conflicting with recovery.")));
-
Assert(MyReplicationSlot->data.invalidated == RS_INVAL_NONE);
Assert(MyReplicationSlot->data.restart_lsn != InvalidXLogRecPtr);
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 0148ec3678..ca53caac2f 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 f6945af1d4..be6f87f00b 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -446,7 +446,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();
}
@@ -665,7 +665,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
* pre-check to ensure that at least one of the slot properties is
* changed before acquiring the slot.
*/
- 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 b30e0473e1..74f7d565f0 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -535,9 +535,13 @@ 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.
+ *
+ * An error is raised if error_if_invalid is true and the slot is found to
+ * be invalid. It should always be set to true, except when we are temporarily
+ * acquiring the slot and doesn't intend to change it.
*/
void
-ReplicationSlotAcquire(const char *name, bool nowait)
+ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid)
{
ReplicationSlot *s;
int active_pid;
@@ -585,6 +589,18 @@ retry:
active_pid = MyProcPid;
LWLockRelease(ReplicationSlotControlLock);
+ /* We made this slot active, so it's ours now. */
+ MyReplicationSlot = s;
+
+ /* Invalid slots can't be modified or used before accessing the WAL. */
+ if (error_if_invalid && s->data.invalidated != RS_INVAL_NONE)
+ ereport(ERROR,
+ errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("can no longer access replication slot \"%s\"",
+ NameStr(s->data.name)),
+ errdetail("This replication slot has been invalidated due to \"%s\".",
+ SlotInvalidationCauses[s->data.invalidated]));
+
/*
* If we found the slot but it's already active in another process, we
* wait until the owning process signals us that it's been released, or
@@ -612,9 +628,6 @@ retry:
/* Let everybody know we've modified this slot */
ConditionVariableBroadcast(&s->active_cv);
- /* We made this slot active, so it's ours now. */
- MyReplicationSlot = s;
-
/*
* The call to pgstat_acquire_replslot() protects against stats for a
* different slot, from before a restart or such, being present during
@@ -785,7 +798,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
@@ -812,7 +825,7 @@ ReplicationSlotAlter(const char *name, const bool *failover,
Assert(MyReplicationSlot == NULL);
Assert(failover || two_phase);
- ReplicationSlotAcquire(name, false);
+ ReplicationSlotAcquire(name, false, true);
if (SlotIsPhysical(MyReplicationSlot))
ereport(ERROR,
@@ -820,13 +833,6 @@ ReplicationSlotAlter(const char *name, const bool *failover,
errmsg("cannot use %s with a physical replication slot",
"ALTER_REPLICATION_SLOT"));
- if (MyReplicationSlot->data.invalidated != RS_INVAL_NONE)
- ereport(ERROR,
- errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
- errmsg("cannot alter invalid replication slot \"%s\"", name),
- errdetail("This replication slot has been invalidated due to \"%s\".",
- SlotInvalidationCauses[MyReplicationSlot->data.invalidated]));
-
if (RecoveryInProgress())
{
/*
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 977146789f..8be4b8c65b 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -536,7 +536,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 bac504b554..446d10c1a7 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -816,7 +816,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),
@@ -1434,7 +1434,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 9a10907d05..d44f8c262b 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -298,7 +298,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, true);
Assert(SlotIsLogical(MyReplicationSlot));
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index bf62b36ad0..47ebdaecb6 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -253,7 +253,8 @@ extern void ReplicationSlotDropAcquired(void);
extern void ReplicationSlotAlter(const char *name, const bool *failover,
const bool *two_phase);
-extern void ReplicationSlotAcquire(const char *name, bool nowait);
+extern void ReplicationSlotAcquire(const char *name, bool nowait,
+ bool error_if_invalid);
extern void ReplicationSlotRelease(void);
extern void ReplicationSlotCleanup(bool synced_only);
extern void ReplicationSlotSave(void);
diff --git a/src/test/recovery/t/019_replslot_limit.pl b/src/test/recovery/t/019_replslot_limit.pl
index ae2ad5c933..6468784b83 100644
--- a/src/test/recovery/t/019_replslot_limit.pl
+++ b/src/test/recovery/t/019_replslot_limit.pl
@@ -234,7 +234,7 @@ my $failed = 0;
for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++)
{
if ($node_standby->log_contains(
- "requested WAL segment [0-9A-F]+ has already been removed",
+ "This replication slot has been invalidated due to \"wal_removed\".",
$logstart))
{
$failed = 1;
diff --git a/src/test/recovery/t/035_standby_logical_decoding.pl b/src/test/recovery/t/035_standby_logical_decoding.pl
index 7e794c5bea..505e85d1eb 100644
--- a/src/test/recovery/t/035_standby_logical_decoding.pl
+++ b/src/test/recovery/t/035_standby_logical_decoding.pl
@@ -533,7 +533,7 @@ check_slots_conflict_reason('vacuum_full_', 'rows_removed');
qq[ALTER_REPLICATION_SLOT vacuum_full_inactiveslot (failover);],
replication => 'database');
ok( $stderr =~
- /ERROR: cannot alter invalid replication slot "vacuum_full_inactiveslot"/
+ /ERROR: can no longer access replication slot "vacuum_full_inactiveslot"/
&& $stderr =~
/DETAIL: This replication slot has been invalidated due to "rows_removed"./,
"invalidated slot cannot be altered");
@@ -551,8 +551,7 @@ $handle =
# We are not able to read from the slot as it has been invalidated
check_pg_recvlogical_stderr($handle,
- "can no longer get changes from replication slot \"vacuum_full_activeslot\""
-);
+ "can no longer access replication slot \"vacuum_full_activeslot\"");
# Turn hot_standby_feedback back on
change_hot_standby_feedback_and_wait_for_xmins(1, 1);
@@ -632,8 +631,7 @@ $handle =
# We are not able to read from the slot as it has been invalidated
check_pg_recvlogical_stderr($handle,
- "can no longer get changes from replication slot \"row_removal_activeslot\""
-);
+ "can no longer access replication slot \"row_removal_activeslot\"");
##################################################
# Recovery conflict: Same as Scenario 2 but on a shared catalog table
@@ -668,7 +666,7 @@ $handle = make_slot_active($node_standby, 'shared_row_removal_', 0, \$stdout,
# We are not able to read from the slot as it has been invalidated
check_pg_recvlogical_stderr($handle,
- "can no longer get changes from replication slot \"shared_row_removal_activeslot\""
+ "can no longer access replication slot \"shared_row_removal_activeslot\""
);
##################################################
@@ -759,7 +757,7 @@ $handle = make_slot_active($node_standby, 'pruning_', 0, \$stdout, \$stderr);
# We are not able to read from the slot as it has been invalidated
check_pg_recvlogical_stderr($handle,
- "can no longer get changes from replication slot \"pruning_activeslot\"");
+ "can no longer access replication slot \"pruning_activeslot\"");
# Turn hot_standby_feedback back on
change_hot_standby_feedback_and_wait_for_xmins(1, 1);
@@ -818,8 +816,7 @@ $handle =
make_slot_active($node_standby, 'wal_level_', 0, \$stdout, \$stderr);
# as the slot has been invalidated we should not be able to read
check_pg_recvlogical_stderr($handle,
- "can no longer get changes from replication slot \"wal_level_activeslot\""
-);
+ "can no longer access replication slot \"wal_level_activeslot\"");
##################################################
# DROP DATABASE should drop its slots, including active slots.
--
2.34.1
[application/octet-stream] v65-0002-Introduce-inactive_timeout-based-replication-slo.patch (24.1K, ../CABdArM5oXMO63uETfEFLEan3UfNBiDwprdoyYYdeWP9fsYNg9A@mail.gmail.com/3-v65-0002-Introduce-inactive_timeout-based-replication-slo.patch)
download | inline diff:
From a07b916edbb25418acdadeeffd9cbd008fd7a315 Mon Sep 17 00:00:00 2001
From: Nisha Moond <[email protected]>
Date: Thu, 30 Jan 2025 12:51:11 +0530
Subject: [PATCH v65 2/4] Introduce 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
this GUC is a bit tricky. Because the amount of WAL a database
generates, and the allocated storage per instance will vary
greatly in production, making it difficult to pin down a
one-size-fits-all value.
It is often easy for users to set a timeout of say 1 or 2 or n
days, after which all the inactive slots get invalidated. This
commit introduces a GUC named idle_replication_slot_timeout.
When set, postgres invalidates slots (during non-shutdown
checkpoints) that are idle for longer than this amount of
time.
Note that the idle timeout invalidation mechanism is not applicable
for slots that do not reserve WAL or for slots on the standby server
that are being synced from the primary server (i.e., standby slots
having 'synced' field 'true'). Synced slots are always considered to be
inactive because they don't perform logical decoding to produce changes.
---
doc/src/sgml/config.sgml | 40 +++++
doc/src/sgml/logical-replication.sgml | 5 +
doc/src/sgml/system-views.sgml | 10 +-
src/backend/replication/logical/slotsync.c | 4 +-
src/backend/replication/slot.c | 170 ++++++++++++++++--
src/backend/utils/adt/timestamp.c | 18 ++
src/backend/utils/misc/guc_tables.c | 14 ++
src/backend/utils/misc/postgresql.conf.sample | 1 +
src/bin/pg_basebackup/pg_createsubscriber.c | 4 +
src/bin/pg_upgrade/server.c | 7 +
src/include/replication/slot.h | 22 +++
src/include/utils/guc_hooks.h | 2 +
src/include/utils/timestamp.h | 3 +
13 files changed, 277 insertions(+), 23 deletions(-)
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index a782f10998..a065fbbaab 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4423,6 +4423,46 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows
</listitem>
</varlistentry>
+ <varlistentry id="guc-idle-replication-slot-timeout" xreflabel="idle_replication_slot_timeout">
+ <term><varname>idle_replication_slot_timeout</varname> (<type>integer</type>)
+ <indexterm>
+ <primary><varname>idle_replication_slot_timeout</varname> configuration parameter</primary>
+ </indexterm>
+ </term>
+ <listitem>
+ <para>
+ Invalidate replication slots that have remained idle longer than this
+ duration. If this value is specified without units, it is taken as
+ minutes. A value of zero disables the idle timeout invalidation
+ mechanism. The default is one day. This parameter can only be set in
+ the <filename>postgresql.conf</filename> file or on the server command
+ line.
+ </para>
+
+ <para>
+ Slot invalidation due to idle timeout occurs during checkpoint.
+ Because checkpoints happen at <varname>checkpoint_timeout</varname>
+ intervals, there can be some lag between when the
+ <varname>idle_replication_slot_timeout</varname> was exceeded and when
+ the slot invalidation is triggered at the next checkpoint.
+ To avoid such lags, users can force a checkpoint to promptly invalidate
+ inactive slots. The duration of slot inactivity is calculated using the
+ slot's <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>inactive_since</structfield>
+ value.
+ </para>
+
+ <para>
+ Note that the idle timeout invalidation mechanism is not applicable
+ for slots that do not reserve WAL or for slots on the standby server
+ that are being synced from the primary server (i.e., standby slots
+ having <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>synced</structfield>
+ value <literal>true</literal>).
+ Synced slots are always considered to be inactive because they don't
+ perform logical decoding to produce changes.
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry id="guc-wal-sender-timeout" xreflabel="wal_sender_timeout">
<term><varname>wal_sender_timeout</varname> (<type>integer</type>)
<indexterm>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 613abcd28b..3d18e507bb 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -2390,6 +2390,11 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
plus some reserve for table synchronization.
</para>
+ <para>
+ Logical replication slots are also affected by
+ <link linkend="guc-idle-replication-slot-timeout"><varname>idle_replication_slot_timeout</varname></link>.
+ </para>
+
<para>
<link linkend="guc-max-wal-senders"><varname>max_wal_senders</varname></link>
should be set to at least the same as
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index 8e2b0a7927..7d3a0aa709 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2566,7 +2566,8 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
</para>
<para>
The time when the slot became inactive. <literal>NULL</literal> if the
- slot is currently being streamed.
+ slot is currently being streamed. If the slot becomes invalidated,
+ this value will remain unchanged until server shutdown.
Note that for slots on the standby that are being synced from a
primary server (whose <structfield>synced</structfield> field is
<literal>true</literal>), the <structfield>inactive_since</structfield>
@@ -2620,6 +2621,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>idle_timeout</literal> means that the slot has remained
+ idle longer than the configured
+ <xref linkend="guc-idle-replication-slot-timeout"/> duration.
+ </para>
+ </listitem>
</itemizedlist>
</para></entry>
</row>
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index be6f87f00b..987857b949 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -1541,9 +1541,7 @@ update_synced_slots_inactive_since(void)
if (now == 0)
now = GetCurrentTimestamp();
- SpinLockAcquire(&s->mutex);
- s->inactive_since = now;
- SpinLockRelease(&s->mutex);
+ ReplicationSlotSetInactiveSince(s, now, true);
}
}
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 74f7d565f0..2ebf366785 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_IDLE_TIMEOUT] = "idle_timeout",
};
/* Maximum number of invalidation causes */
-#define RS_INVAL_MAX_CAUSES RS_INVAL_WAL_LEVEL
+#define RS_INVAL_MAX_CAUSES RS_INVAL_IDLE_TIMEOUT
StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1),
"array length mismatch");
@@ -141,6 +142,12 @@ ReplicationSlot *MyReplicationSlot = NULL;
int max_replication_slots = 10; /* the maximum number of replication
* slots */
+/*
+ * Invalidate replication slots that have remained idle longer than this
+ * duration; '0' disables it.
+ */
+int idle_replication_slot_timeout_mins = HOURS_PER_DAY * MINS_PER_HOUR;
+
/*
* This GUC lists streaming replication standby server slot names that
* logical WAL sender processes will wait for.
@@ -716,16 +723,12 @@ ReplicationSlotRelease(void)
*/
SpinLockAcquire(&slot->mutex);
slot->active_pid = 0;
- slot->inactive_since = now;
+ ReplicationSlotSetInactiveSince(slot, now, false);
SpinLockRelease(&slot->mutex);
ConditionVariableBroadcast(&slot->active_cv);
}
else
- {
- SpinLockAcquire(&slot->mutex);
- slot->inactive_since = now;
- SpinLockRelease(&slot->mutex);
- }
+ ReplicationSlotSetInactiveSince(slot, now, true);
MyReplicationSlot = NULL;
@@ -1514,7 +1517,8 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
NameData slotname,
XLogRecPtr restart_lsn,
XLogRecPtr oldestLSN,
- TransactionId snapshotConflictHorizon)
+ TransactionId snapshotConflictHorizon,
+ TimestampTz inactive_since)
{
StringInfoData err_detail;
bool hint = false;
@@ -1544,6 +1548,16 @@ 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_IDLE_TIMEOUT:
+ Assert(inactive_since > 0);
+ /* translator: second %s is a GUC variable name */
+ appendStringInfo(&err_detail,
+ _("The slot has remained idle since %s, which is longer than the configured \"%s\" duration."),
+ timestamptz_to_str(inactive_since),
+ "idle_replication_slot_timeout");
+ break;
+
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -1560,6 +1574,32 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
pfree(err_detail.data);
}
+/*
+ * Can we invalidate an idle replication slot?
+ *
+ * Idle timeout invalidation is allowed only when:
+ *
+ * 1. Idle timeout is set
+ * 2. Slot has WAL reserved
+ * 3. Slot is inactive
+ * 4. The slot is not being synced from the primary while the server
+ * is in recovery
+ *
+ * Note that the idle timeout invalidation mechanism is not
+ * applicable for slots on the standby server that are being synced
+ * from the primary server (i.e., standby slots having 'synced' field 'true').
+ * Synced slots are always considered to be inactive because they don't
+ * perform logical decoding to produce changes.
+ */
+static inline bool
+CanInvalidateIdleSlot(ReplicationSlot *s)
+{
+ return (idle_replication_slot_timeout_mins > 0 &&
+ !XLogRecPtrIsInvalid(s->data.restart_lsn) &&
+ s->inactive_since > 0 &&
+ !(RecoveryInProgress() && s->data.synced));
+}
+
/*
* Helper for InvalidateObsoleteReplicationSlots
*
@@ -1587,6 +1627,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
TransactionId initial_catalog_effective_xmin = InvalidTransactionId;
XLogRecPtr initial_restart_lsn = InvalidXLogRecPtr;
ReplicationSlotInvalidationCause invalidation_cause_prev PG_USED_FOR_ASSERTS_ONLY = RS_INVAL_NONE;
+ TimestampTz inactive_since = 0;
for (;;)
{
@@ -1594,6 +1635,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
NameData slotname;
int active_pid = 0;
ReplicationSlotInvalidationCause invalidation_cause = RS_INVAL_NONE;
+ TimestampTz now = 0;
Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
@@ -1604,6 +1646,15 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
break;
}
+ if (cause == RS_INVAL_IDLE_TIMEOUT)
+ {
+ /*
+ * We get the current time beforehand to avoid system call while
+ * holding the spinlock.
+ */
+ now = GetCurrentTimestamp();
+ }
+
/*
* Check if the slot needs to be invalidated. If it needs to be
* invalidated, and is not currently acquired, acquire it and mark it
@@ -1657,6 +1708,21 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
if (SlotIsLogical(s))
invalidation_cause = cause;
break;
+ case RS_INVAL_IDLE_TIMEOUT:
+ Assert(now > 0);
+
+ /*
+ * Check if the slot needs to be invalidated due to
+ * idle_replication_slot_timeout GUC.
+ */
+ if (CanInvalidateIdleSlot(s) &&
+ TimestampDifferenceExceedsSeconds(s->inactive_since, now,
+ idle_replication_slot_timeout_mins * SECS_PER_MINUTE))
+ {
+ invalidation_cause = cause;
+ inactive_since = s->inactive_since;
+ }
+ break;
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -1707,9 +1773,10 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
/*
* The logical replication slots shouldn't be invalidated as GUC
- * max_slot_wal_keep_size is set to -1 during the binary upgrade. See
- * check_old_cluster_for_valid_slots() where we ensure that no
- * invalidated before the upgrade.
+ * max_slot_wal_keep_size is set to -1 and
+ * idle_replication_slot_timeout is set to 0 during the binary
+ * upgrade. See check_old_cluster_for_valid_slots() where we ensure
+ * that no invalidated before the upgrade.
*/
Assert(!(*invalidated && SlotIsLogical(s) && IsBinaryUpgrade));
@@ -1741,7 +1808,8 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
{
ReportSlotInvalidation(invalidation_cause, true, active_pid,
slotname, restart_lsn,
- oldestLSN, snapshotConflictHorizon);
+ oldestLSN, snapshotConflictHorizon,
+ inactive_since);
if (MyBackendType == B_STARTUP)
(void) SendProcSignal(active_pid,
@@ -1787,7 +1855,8 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
ReportSlotInvalidation(invalidation_cause, false, active_pid,
slotname, restart_lsn,
- oldestLSN, snapshotConflictHorizon);
+ oldestLSN, snapshotConflictHorizon,
+ inactive_since);
/* done with this slot for now */
break;
@@ -1802,14 +1871,16 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
/*
* Invalidate slots that require resources about to be removed.
*
- * Returns true when any slot have got invalidated.
+ * Returns true if there are any invalidated slots.
*
- * Whether a slot needs to be invalidated depends on the cause. A slot is
- * removed if it:
+ * Whether a slot needs to be invalidated depends on the invalidation cause.
+ * A slot is invalidated if it:
* - RS_INVAL_WAL_REMOVED: requires a LSN older than the given segment
* - 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_WAL_LEVEL: is logical and wal_level is insufficient
+ * - RS_INVAL_IDLE_TIMEOUT: has been idle longer than the configured
+ * "idle_replication_slot_timeout" duration.
*
* NB - this runs as part of checkpoint, so avoid raising errors if possible.
*/
@@ -1862,7 +1933,8 @@ restart:
}
/*
- * Flush all replication slots to disk.
+ * Flush all replication slots to disk. Also, invalidate obsolete slots during
+ * non-shutdown checkpoint.
*
* It is convenient to flush dirty replication slots at the time of checkpoint.
* Additionally, in case of a shutdown checkpoint, we also identify the slots
@@ -1920,6 +1992,45 @@ CheckPointReplicationSlots(bool is_shutdown)
SaveSlotToPath(s, path, LOG);
}
LWLockRelease(ReplicationSlotAllocationLock);
+
+ if (!is_shutdown)
+ {
+ elog(DEBUG1, "performing replication slot invalidation checks");
+
+ /*
+ * NB: We will make another pass over replication slots for
+ * invalidation checks to keep the code simple. Testing shows that
+ * there is no noticeable overhead (when compared with wal_removed
+ * invalidation) even if we were to do idle_timeout invalidation of
+ * thousands of replication slots here. If it is ever proven that this
+ * assumption is wrong, we will have to perform the invalidation
+ * checks in the above for loop with the following changes:
+ *
+ * - Acquire ControlLock lock once before the loop.
+ *
+ * - Call InvalidatePossiblyObsoleteSlot for each slot.
+ *
+ * - Handle the cases in which ControlLock gets released just like
+ * InvalidateObsoleteReplicationSlots does.
+ *
+ * - Avoid saving slot info to disk two times for each invalidated
+ * slot.
+ *
+ * XXX: Should we move idle_timeout invalidation check closer to
+ * wal_removed in CreateCheckPoint and CreateRestartPoint?
+ *
+ * XXX: Slot invalidation due to 'idle_timeout' applies only to
+ * released slots, and is based on the 'idle_replication_slot_timeout'
+ * GUC. Active slots currently in use for replication are excluded to
+ * prevent accidental invalidation. Slots where communication between
+ * the publisher and subscriber is down are also excluded, as they are
+ * managed by the 'wal_sender_timeout'.
+ */
+ InvalidateObsoleteReplicationSlots(RS_INVAL_IDLE_TIMEOUT,
+ 0,
+ InvalidOid,
+ InvalidTransactionId);
+ }
}
/*
@@ -2404,7 +2515,9 @@ RestoreSlotFromDisk(const char *name)
/*
* Set the time since the slot has become inactive after loading the
* slot from the disk into memory. Whoever acquires the slot i.e.
- * makes the slot active will reset it.
+ * makes the slot active will reset it. Avoid calling
+ * ReplicationSlotSetInactiveSince() here, as it will not set the time
+ * for invalid slots.
*/
slot->inactive_since = GetCurrentTimestamp();
@@ -2799,3 +2912,22 @@ WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
ConditionVariableCancelSleep();
}
+
+/*
+ * GUC check_hook for idle_replication_slot_timeout
+ *
+ * The value of idle_replication_slot_timeout must be set to 0 during
+ * a binary upgrade. See start_postmaster() in pg_upgrade for more details.
+ */
+bool
+check_idle_replication_slot_timeout(int *newval, void **extra, GucSource source)
+{
+ if (IsBinaryUpgrade && *newval != 0)
+ {
+ GUC_check_errdetail("The value of \"%s\" must be set to 0 during binary upgrade mode.",
+ "idle_replication_slot_timeout");
+ return false;
+ }
+
+ return true;
+}
diff --git a/src/backend/utils/adt/timestamp.c b/src/backend/utils/adt/timestamp.c
index ba9bae0506..9682f9dbdc 100644
--- a/src/backend/utils/adt/timestamp.c
+++ b/src/backend/utils/adt/timestamp.c
@@ -1786,6 +1786,24 @@ TimestampDifferenceExceeds(TimestampTz start_time,
return (diff >= msec * INT64CONST(1000));
}
+/*
+ * Check if the difference between two timestamps is >= a given
+ * threshold (expressed in seconds).
+ */
+bool
+TimestampDifferenceExceedsSeconds(TimestampTz start_time,
+ TimestampTz stop_time,
+ int threshold_sec)
+{
+ long secs;
+ int usecs;
+
+ /* Calculate the difference in seconds */
+ TimestampDifference(start_time, stop_time, &secs, &usecs);
+
+ return (secs >= threshold_sec);
+}
+
/*
* Convert a time_t to TimestampTz.
*
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 38cb9e970d..7cbba03bc1 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -3048,6 +3048,20 @@ struct config_int ConfigureNamesInt[] =
NULL, NULL, NULL
},
+ {
+ {"idle_replication_slot_timeout", PGC_SIGHUP, REPLICATION_SENDING,
+ gettext_noop("Sets the duration a replication slot can remain idle before "
+ "it is invalidated."),
+ NULL,
+ GUC_UNIT_MIN
+ },
+ &idle_replication_slot_timeout_mins,
+ HOURS_PER_DAY * MINS_PER_HOUR, /* 1 day */
+ 0,
+ INT_MAX / SECS_PER_MINUTE,
+ check_idle_replication_slot_timeout, NULL, NULL
+ },
+
{
{"commit_delay", PGC_SUSET, WAL_SETTINGS,
gettext_noop("Sets the delay in microseconds between transaction commit and "
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 079efa1baa..0ed9eb057e 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -329,6 +329,7 @@
#wal_sender_timeout = 60s # in milliseconds; 0 disables
#track_commit_timestamp = off # collect timestamp of transaction commit
# (change requires restart)
+#idle_replication_slot_timeout = 1d # in minutes; 0 disables
# - Primary Server -
diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index faf18ccf13..f2ef8d5ccc 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -1438,6 +1438,10 @@ start_standby_server(const struct CreateSubscriberOptions *opt, bool restricted_
appendPQExpBuffer(pg_ctl_cmd, "\"%s\" start -D ", pg_ctl_path);
appendShellString(pg_ctl_cmd, subscriber_dir);
appendPQExpBuffer(pg_ctl_cmd, " -s -o \"-c sync_replication_slots=off\"");
+
+ /* Prevent unintended slot invalidation */
+ appendPQExpBuffer(pg_ctl_cmd, " -o \"-c idle_replication_slot_timeout=0\"");
+
if (restricted_access)
{
appendPQExpBuffer(pg_ctl_cmd, " -o \"-p %s\"", opt->sub_port);
diff --git a/src/bin/pg_upgrade/server.c b/src/bin/pg_upgrade/server.c
index de6971cde6..873e5b5117 100644
--- a/src/bin/pg_upgrade/server.c
+++ b/src/bin/pg_upgrade/server.c
@@ -252,6 +252,13 @@ start_postmaster(ClusterInfo *cluster, bool report_and_exit_on_error)
if (GET_MAJOR_VERSION(cluster->major_version) >= 1700)
appendPQExpBufferStr(&pgoptions, " -c max_slot_wal_keep_size=-1");
+ /*
+ * Use idle_replication_slot_timeout=0 to prevent slot invalidation due to
+ * idle_timeout by checkpointer process during upgrade.
+ */
+ if (GET_MAJOR_VERSION(cluster->major_version) >= 1800)
+ appendPQExpBufferStr(&pgoptions, " -c idle_replication_slot_timeout=0");
+
/*
* Use -b to disable autovacuum and logical replication launcher
* (effective in PG17 or later for the latter).
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 47ebdaecb6..f3994ab000 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -56,6 +56,8 @@ typedef enum ReplicationSlotInvalidationCause
RS_INVAL_HORIZON,
/* wal_level insufficient for slot */
RS_INVAL_WAL_LEVEL,
+ /* idle slot timeout has occurred */
+ RS_INVAL_IDLE_TIMEOUT,
} ReplicationSlotInvalidationCause;
extern PGDLLIMPORT const char *const SlotInvalidationCauses[];
@@ -228,6 +230,25 @@ typedef struct ReplicationSlotCtlData
ReplicationSlot replication_slots[1];
} ReplicationSlotCtlData;
+/*
+ * Set slot's inactive_since property unless it was previously invalidated.
+ */
+static inline void
+ReplicationSlotSetInactiveSince(ReplicationSlot *s, TimestampTz now,
+ bool acquire_lock)
+{
+ if (s->data.invalidated != RS_INVAL_NONE)
+ return;
+
+ if (acquire_lock)
+ SpinLockAcquire(&s->mutex);
+
+ s->inactive_since = now;
+
+ if (acquire_lock)
+ SpinLockRelease(&s->mutex);
+}
+
/*
* Pointers to shared memory
*/
@@ -237,6 +258,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot;
/* GUCs */
extern PGDLLIMPORT int max_replication_slots;
extern PGDLLIMPORT char *synchronized_standby_slots;
+extern PGDLLIMPORT int idle_replication_slot_timeout_mins;
/* shmem initialization functions */
extern Size ReplicationSlotsShmemSize(void);
diff --git a/src/include/utils/guc_hooks.h b/src/include/utils/guc_hooks.h
index 87999218d6..951451a976 100644
--- a/src/include/utils/guc_hooks.h
+++ b/src/include/utils/guc_hooks.h
@@ -174,5 +174,7 @@ extern void assign_wal_sync_method(int new_wal_sync_method, void *extra);
extern bool check_synchronized_standby_slots(char **newval, void **extra,
GucSource source);
extern void assign_synchronized_standby_slots(const char *newval, void *extra);
+extern bool check_idle_replication_slot_timeout(int *newval, void **extra,
+ GucSource source);
#endif /* GUC_HOOKS_H */
diff --git a/src/include/utils/timestamp.h b/src/include/utils/timestamp.h
index d26f023fb8..e1d05d6779 100644
--- a/src/include/utils/timestamp.h
+++ b/src/include/utils/timestamp.h
@@ -143,5 +143,8 @@ extern int date2isoyear(int year, int mon, int mday);
extern int date2isoyearday(int year, int mon, int mday);
extern bool TimestampTimestampTzRequiresRewrite(void);
+extern bool TimestampDifferenceExceedsSeconds(TimestampTz start_time,
+ TimestampTz stop_time,
+ int threshold_sec);
#endif /* TIMESTAMP_H */
--
2.34.1
[application/octet-stream] v65-0003-Add-TAP-test-for-slot-invalidation-based-on-inac.patch (7.6K, ../CABdArM5oXMO63uETfEFLEan3UfNBiDwprdoyYYdeWP9fsYNg9A@mail.gmail.com/4-v65-0003-Add-TAP-test-for-slot-invalidation-based-on-inac.patch)
download | inline diff:
From 2c7a48b926787bad592e91d3270706eab1425077 Mon Sep 17 00:00:00 2001
From: Nisha Moond <[email protected]>
Date: Thu, 30 Jan 2025 21:07:12 +0530
Subject: [PATCH v65 3/4] Add TAP test for slot invalidation based on inactive
timeout.
This test uses injection points to bypass the time overhead caused by the
idle_replication_slot_timeout GUC, which has a minimum value of one minute.
---
src/backend/replication/slot.c | 5 +
src/test/recovery/meson.build | 1 +
.../t/044_invalidate_inactive_slots.pl | 176 ++++++++++++++++++
3 files changed, 182 insertions(+)
create mode 100644 src/test/recovery/t/044_invalidate_inactive_slots.pl
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 2ebf366785..2191033e5c 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -55,6 +55,7 @@
#include "storage/proc.h"
#include "storage/procarray.h"
#include "utils/builtins.h"
+#include "utils/injection_point.h"
#include "utils/guc_hooks.h"
#include "utils/varlena.h"
@@ -1711,6 +1712,10 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
case RS_INVAL_IDLE_TIMEOUT:
Assert(now > 0);
+ /* For testing timeout slot invalidation */
+ if (IS_INJECTION_POINT_ATTACHED("slot-time-out-inval"))
+ s->inactive_since = 1;
+
/*
* Check if the slot needs to be invalidated due to
* idle_replication_slot_timeout GUC.
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index 0428704dbf..057bcde143 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -52,6 +52,7 @@ tests += {
't/041_checkpoint_at_promote.pl',
't/042_low_level_backup.pl',
't/043_no_contrecord_switch.pl',
+ 't/044_invalidate_inactive_slots.pl',
],
},
}
diff --git a/src/test/recovery/t/044_invalidate_inactive_slots.pl b/src/test/recovery/t/044_invalidate_inactive_slots.pl
new file mode 100644
index 0000000000..ab948aba85
--- /dev/null
+++ b/src/test/recovery/t/044_invalidate_inactive_slots.pl
@@ -0,0 +1,176 @@
+# Copyright (c) 2025, 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;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+ plan skip_all => 'Injection points not supported by this build';
+}
+
+# =============================================================================
+# Testcase start
+#
+# Test invalidation of streaming standby slot and logical failover slot on the
+# primary due to idle timeout. Also, test logical failover slot synced to
+# the standby from the primary doesn't get invalidated on its own, but gets the
+# invalidated state from the primary.
+
+# Initialize primary
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 'logical');
+
+# Avoid unpredictability
+$primary->append_conf(
+ 'postgresql.conf', qq{
+checkpoint_timeout = 1h
+});
+$primary->start;
+
+# Take backup
+my $backup_name = 'my_backup';
+$primary->backup($backup_name);
+
+# Create sync slot on the primary
+$primary->psql('postgres',
+ q{SELECT pg_create_logical_replication_slot('sync_slot1', 'test_decoding', false, false, true);}
+);
+
+# Create standby slot on the primary
+$primary->safe_psql(
+ 'postgres', qq[
+ SELECT pg_create_physical_replication_slot(slot_name := 'sb_slot1', immediately_reserve := true);
+]);
+
+# Create standby
+my $standby1 = PostgreSQL::Test::Cluster->new('standby1');
+$standby1->init_from_backup($primary, $backup_name, has_streaming => 1);
+
+my $connstr = $primary->connstr;
+$standby1->append_conf(
+ 'postgresql.conf', qq(
+hot_standby_feedback = on
+primary_slot_name = 'sb_slot1'
+idle_replication_slot_timeout = 1
+primary_conninfo = '$connstr dbname=postgres'
+));
+$standby1->start;
+
+# Wait until the standby has replayed enough data
+$primary->wait_for_catchup($standby1);
+
+# Sync the primary slots to the standby
+$standby1->safe_psql('postgres', "SELECT pg_sync_replication_slots();");
+
+# Confirm that the logical failover slot is created on the standby
+is( $standby1->safe_psql(
+ 'postgres',
+ q{SELECT count(*) = 1 FROM pg_replication_slots
+ WHERE slot_name = 'sync_slot1' AND synced
+ AND NOT temporary
+ AND invalidation_reason IS NULL;}
+ ),
+ 't',
+ 'logical slot sync_slot1 is synced to standby');
+
+my $logstart = -s $primary->logfile;
+
+# Set timeout GUC so that the next checkpoint will invalidate inactive slots
+$primary->safe_psql(
+ 'postgres', qq[
+ ALTER SYSTEM SET idle_replication_slot_timeout TO '1min';
+]);
+$primary->reload;
+
+# Register an injection point on the primary to forcibly cause a slot timeout
+$primary->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
+
+# Check if the extension injection_points is available, as it may be
+# possible that this script is run with installcheck, where the module
+# would not be installed by default.
+if (!$primary->check_extension('injection_points'))
+{
+ plan skip_all => 'Extension injection_points not installed';
+}
+
+$primary->safe_psql('postgres',
+ "SELECT injection_points_attach('slot-time-out-inval', 'error');");
+
+# Wait for logical failover slot to become inactive on the primary. Note that
+# nobody has acquired the slot yet, so it must get invalidated due to
+# idle timeout.
+wait_for_slot_invalidation($primary, 'sync_slot1', $logstart);
+
+# Re-sync the primary slots to the standby. Note that the primary slot was
+# already invalidated (above) due to idle timeout. The standby must just
+# sync the invalidated state.
+$standby1->safe_psql('postgres', "SELECT pg_sync_replication_slots();");
+
+is( $standby1->safe_psql(
+ 'postgres',
+ q{SELECT count(*) = 1 FROM pg_replication_slots
+ WHERE slot_name = 'sync_slot1'
+ AND invalidation_reason = 'idle_timeout';}
+ ),
+ "t",
+ 'check that invalidation of synced slot sync_slot1 is synced on standby');
+
+# Make the standby slot on the primary inactive and check for invalidation
+$standby1->stop;
+wait_for_slot_invalidation($primary, 'sb_slot1', $logstart);
+
+# Testcase end
+# =============================================================================
+
+# Wait for slot to first become idle and then get invalidated
+sub wait_for_slot_invalidation
+{
+ my ($node, $slot, $offset) = @_;
+ my $node_name = $node->name;
+
+ trigger_slot_invalidation($node, $slot, $offset);
+
+ # Check that an invalidated slot cannot be acquired
+ my ($result, $stdout, $stderr);
+ ($result, $stdout, $stderr) = $node->psql(
+ 'postgres', qq[
+ SELECT pg_replication_slot_advance('$slot', '0/1');
+ ]);
+ ok( $stderr =~ /can no longer access replication slot "$slot"/,
+ "detected error upon trying to acquire invalidated slot $slot on node $node_name"
+ )
+ or die
+ "could not detect error upon trying to acquire invalidated slot $slot on node $node_name";
+}
+
+# Trigger slot invalidation and confirm it in the server log
+sub trigger_slot_invalidation
+{
+ my ($node, $slot, $offset) = @_;
+ my $node_name = $node->name;
+ my $invalidated = 0;
+
+ # Run a checkpoint
+ $node->safe_psql('postgres', "CHECKPOINT");
+
+ # The slot's invalidation should be logged
+ $node->wait_for_log(qr/invalidating obsolete replication slot \"$slot\"/,
+ $offset);
+
+ # Check that the invalidation reason is 'idle_timeout'
+ $node->poll_query_until(
+ 'postgres', qq[
+ SELECT COUNT(slot_name) = 1 FROM pg_replication_slots
+ WHERE slot_name = '$slot' AND
+ invalidation_reason = 'idle_timeout';
+ ])
+ or die
+ "Timed out while waiting for invalidation reason of slot $slot to be set on node $node_name";
+}
+
+done_testing();
--
2.34.1
[application/octet-stream] v65-0004-Add-TAP-test-for-slot-invalidation-based-on-inac.patch (10.1K, ../CABdArM5oXMO63uETfEFLEan3UfNBiDwprdoyYYdeWP9fsYNg9A@mail.gmail.com/5-v65-0004-Add-TAP-test-for-slot-invalidation-based-on-inac.patch)
download | inline diff:
From b4cac3e24c150c58d3c724e1dc67f5db5d305962 Mon Sep 17 00:00:00 2001
From: Nisha Moond <[email protected]>
Date: Wed, 29 Jan 2025 12:12:00 +0530
Subject: [PATCH v65 4/4] Add TAP test for slot invalidation based on inactive
timeout.
This patch adds the same test, but places it under PG_TEST_EXTRA instead
of using injection points.
Since the minimum value for GUC 'idle_replication_slot_timeout' is one minute,
the test takes more than a minute to complete and is disabled by default.
Use PG_TEST_EXTRA=idle_replication_slot_timeout with "make" to run the test.
---
.cirrus.tasks.yml | 2 +-
doc/src/sgml/regress.sgml | 10 +
src/test/recovery/README | 5 +
src/test/recovery/meson.build | 1 +
.../045_invalidate_inactive_slots_pg_extra.pl | 208 ++++++++++++++++++
5 files changed, 225 insertions(+), 1 deletion(-)
create mode 100644 src/test/recovery/t/045_invalidate_inactive_slots_pg_extra.pl
diff --git a/.cirrus.tasks.yml b/.cirrus.tasks.yml
index 18e944ca89..8d3c13fcee 100644
--- a/.cirrus.tasks.yml
+++ b/.cirrus.tasks.yml
@@ -20,7 +20,7 @@ env:
MTEST_ARGS: --print-errorlogs --no-rebuild -C build
PGCTLTIMEOUT: 120 # avoids spurious failures during parallel tests
TEMP_CONFIG: ${CIRRUS_WORKING_DIR}/src/tools/ci/pg_ci_base.conf
- PG_TEST_EXTRA: kerberos ldap ssl libpq_encryption load_balance
+ PG_TEST_EXTRA: kerberos ldap ssl libpq_encryption load_balance idle_replication_slot_timeout
# What files to preserve in case tests fail
diff --git a/doc/src/sgml/regress.sgml b/doc/src/sgml/regress.sgml
index 7c474559bd..f5b1f2f353 100644
--- a/doc/src/sgml/regress.sgml
+++ b/doc/src/sgml/regress.sgml
@@ -347,6 +347,16 @@ make check-world PG_TEST_EXTRA='kerberos ldap ssl load_balance libpq_encryption'
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>idle_replication_slot_timeout</literal></term>
+ <listitem>
+ <para>
+ Runs the test <filename>src/test/recovery/t/045_invalidate_inactive_slots_pg_extra.pl</filename>.
+ Not enabled by default because it is time consuming.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist>
Tests for features that are not supported by the current build
diff --git a/src/test/recovery/README b/src/test/recovery/README
index 896df0ad05..5c066fc41f 100644
--- a/src/test/recovery/README
+++ b/src/test/recovery/README
@@ -30,4 +30,9 @@ PG_TEST_EXTRA=wal_consistency_checking
to the "make" command. This is resource-intensive, so it's not done
by default.
+If you want to test idle_replication_slot_timeout, add
+PG_TEST_EXTRA=idle_replication_slot_timeout
+to the "make" command. This test takes over a minutes, so it's not done
+by default.
+
See src/test/perl/README for more info about running these tests.
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index 057bcde143..0a037b4b65 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -53,6 +53,7 @@ tests += {
't/042_low_level_backup.pl',
't/043_no_contrecord_switch.pl',
't/044_invalidate_inactive_slots.pl',
+ 't/045_invalidate_inactive_slots_pg_extra.pl',
],
},
}
diff --git a/src/test/recovery/t/045_invalidate_inactive_slots_pg_extra.pl b/src/test/recovery/t/045_invalidate_inactive_slots_pg_extra.pl
new file mode 100644
index 0000000000..577f69d05d
--- /dev/null
+++ b/src/test/recovery/t/045_invalidate_inactive_slots_pg_extra.pl
@@ -0,0 +1,208 @@
+# Copyright (c) 2025, 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;
+
+# The test takes over two minutes to complete. Run it only if
+# idle_replication_slot_timeout is specified in PG_TEST_EXTRA.
+if ( !$ENV{PG_TEST_EXTRA}
+ || $ENV{PG_TEST_EXTRA} !~ /\bidle_replication_slot_timeout\b/)
+{
+ plan skip_all =>
+ 'test idle_replication_slot_timeout not enabled in PG_TEST_EXTRA';
+}
+
+# =============================================================================
+# Testcase start
+#
+# Test invalidation of streaming standby slot and logical failover slot on the
+# primary due to idle timeout. Also, test logical failover slot synced to
+# the standby from the primary doesn't get invalidated on its own, but gets the
+# invalidated state from the primary.
+
+# Initialize primary
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 'logical');
+
+# Avoid unpredictability
+$primary->append_conf(
+ 'postgresql.conf', qq{
+checkpoint_timeout = 1h
+idle_replication_slot_timeout = 1
+});
+$primary->start;
+
+# Take backup
+my $backup_name = 'my_backup';
+$primary->backup($backup_name);
+
+# Create sync slot on the primary
+$primary->psql('postgres',
+ q{SELECT pg_create_logical_replication_slot('sync_slot1', 'test_decoding', false, false, true);}
+);
+
+# Create standby1's slot on the primary
+$primary->safe_psql(
+ 'postgres', qq[
+ SELECT pg_create_physical_replication_slot(slot_name := 'sb_slot1', immediately_reserve := true);
+]);
+
+# Create standby2's slot on the primary
+$primary->safe_psql(
+ 'postgres', qq[
+ SELECT pg_create_physical_replication_slot(slot_name := 'sb_slot2', immediately_reserve := true);
+]);
+
+# Create standby1
+my $standby1 = PostgreSQL::Test::Cluster->new('standby1');
+$standby1->init_from_backup($primary, $backup_name, has_streaming => 1);
+
+my $connstr = $primary->connstr;
+$standby1->append_conf(
+ 'postgresql.conf', qq(
+hot_standby_feedback = on
+primary_slot_name = 'sb_slot1'
+idle_replication_slot_timeout = 1
+primary_conninfo = '$connstr dbname=postgres'
+));
+$standby1->start;
+
+# Wait until the standby has replayed enough data
+$primary->wait_for_catchup($standby1);
+
+# Create standby2
+my $standby2 = PostgreSQL::Test::Cluster->new('standby2');
+$standby2->init_from_backup($primary, $backup_name, has_streaming => 1);
+
+$connstr = $primary->connstr;
+$standby2->append_conf(
+ 'postgresql.conf', qq(
+hot_standby_feedback = on
+primary_slot_name = 'sb_slot2'
+idle_replication_slot_timeout = 1
+primary_conninfo = '$connstr dbname=postgres'
+));
+$standby2->start;
+
+# Wait until the standby has replayed enough data
+$primary->wait_for_catchup($standby2);
+
+# Set timeout GUC on the standby to verify that the next checkpoint will not
+# invalidate synced slots.
+
+# Make the standby2's slot on the primary inactive
+$standby2->stop;
+
+# Sync the primary slots to the standby
+$standby1->safe_psql('postgres', "SELECT pg_sync_replication_slots();");
+
+# Confirm that the logical failover slot is created on the standby
+is( $standby1->safe_psql(
+ 'postgres',
+ q{SELECT count(*) = 1 FROM pg_replication_slots
+ WHERE slot_name = 'sync_slot1' AND synced
+ AND NOT temporary
+ AND invalidation_reason IS NULL;}
+ ),
+ 't',
+ 'logical slot sync_slot1 is synced to standby');
+
+# Give enough time for inactive_since to exceed the timeout
+sleep(61);
+
+# On standby, synced slots are not invalidated by the idle timeout
+# until the invalidation state is propagated from the primary.
+$standby1->safe_psql('postgres', "CHECKPOINT");
+is( $standby1->safe_psql(
+ 'postgres',
+ q{SELECT count(*) = 1 FROM pg_replication_slots
+ WHERE slot_name = 'sync_slot1'
+ AND invalidation_reason IS NULL;}
+ ),
+ 't',
+ 'check that synced slot sync_slot1 has not been invalidated on standby');
+
+my $logstart = -s $primary->logfile;
+
+# Wait for logical failover slot to become inactive on the primary. Note that
+# nobody has acquired the slot yet, so it must get invalidated due to
+# idle timeout as 61 seconds has elapsed and wait for another 10 seconds
+# to make test reliable.
+wait_for_slot_invalidation($primary, 'sync_slot1', $logstart, 10);
+
+# Re-sync the primary slots to the standby. Note that the primary slot was
+# already invalidated (above) due to idle timeout. The standby must just
+# sync the invalidated state.
+$standby1->safe_psql('postgres', "SELECT pg_sync_replication_slots();");
+
+is( $standby1->safe_psql(
+ 'postgres',
+ q{SELECT count(*) = 1 FROM pg_replication_slots
+ WHERE slot_name = 'sync_slot1'
+ AND invalidation_reason = 'idle_timeout';}
+ ),
+ "t",
+ 'check that invalidation of synced slot sync_slot1 is synced on standby');
+
+# By now standby2's slot must be invalidated due to idle timeout,
+# check for invalidation.
+wait_for_slot_invalidation($primary, 'sb_slot2', $logstart, 1);
+
+# Testcase end
+# =============================================================================
+
+# Wait for slot to first become idle and then get invalidated
+sub wait_for_slot_invalidation
+{
+ my ($node, $slot, $offset, $wait_time_secs) = @_;
+ my $node_name = $node->name;
+
+ trigger_slot_invalidation($node, $slot, $offset, $wait_time_secs);
+
+ # Check that an invalidated slot cannot be acquired
+ my ($result, $stdout, $stderr);
+ ($result, $stdout, $stderr) = $node->psql(
+ 'postgres', qq[
+ SELECT pg_replication_slot_advance('$slot', '0/1');
+ ]);
+ ok( $stderr =~ /can no longer access replication slot "$slot"/,
+ "detected error upon trying to acquire invalidated slot $slot on node $node_name"
+ )
+ or die
+ "could not detect error upon trying to acquire invalidated slot $slot on node $node_name";
+}
+
+# Trigger slot invalidation and confirm it in the server log
+sub trigger_slot_invalidation
+{
+ my ($node, $slot, $offset, $wait_time_secs) = @_;
+ my $node_name = $node->name;
+ my $invalidated = 0;
+
+ # Give enough time for inactive_since to exceed the timeout
+ sleep($wait_time_secs);
+
+ # Run a checkpoint
+ $node->safe_psql('postgres', "CHECKPOINT");
+
+ # The slot's invalidation should be logged
+ $node->wait_for_log(qr/invalidating obsolete replication slot \"$slot\"/,
+ $offset);
+
+ # Check that the invalidation reason is 'idle_timeout'
+ $node->poll_query_until(
+ 'postgres', qq[
+ SELECT COUNT(slot_name) = 1 FROM pg_replication_slots
+ WHERE slot_name = '$slot' AND
+ invalidation_reason = 'idle_timeout';
+ ])
+ or die
+ "Timed out while waiting for invalidation reason of slot $slot to be set on node $node_name";
+}
+
+done_testing();
--
2.34.1
view thread (50+ 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], [email protected], [email protected], [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: <CABdArM5oXMO63uETfEFLEan3UfNBiDwprdoyYYdeWP9fsYNg9A@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