public inbox for [email protected]
help / color / mirror / Atom feedFrom: Nisha Moond <[email protected]>
To: vignesh C <[email protected]>
Cc: shveta malik <[email protected]>
Cc: Bharath Rupireddy <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Amit Kapila <[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: Thu, 21 Nov 2024 17:35:01 +0530
Message-ID: <CABdArM6A3xsedesUWaQYnOM+mAGGj--N2ohoLrk2tO-W8UfR5g@mail.gmail.com> (raw)
In-Reply-To: <CALDaNm0g86wD2=bQdFOy0smsP0MZWyz0CUqXej=Qi-hCEeqkag@mail.gmail.com>
References: <CAD21AoAd0BX0L17YxSCbSNbj=3V7dvLVGY0WhwLHAt724rNCZA@mail.gmail.com>
<[email protected]>
<CAJpy0uBthb-RG=Xg+mM-G6Xy01JLcBsizJuz+xbti08qP_W7AQ@mail.gmail.com>
<CALj2ACVfDBcbybgyga7ia_KiQPAfEi_3=VDOGf2VU8RkJZSQpw@mail.gmail.com>
<[email protected]>
<CALj2ACUmp-F+irHYfNhmo_9yaCo7aGi7ZB20KQcDR3S9eT56hg@mail.gmail.com>
<Zg1WqRS/[email protected]>
<CALj2ACUBiRLHF7VYD-0nYVHSvVBSCQ_5OMbP4tE8Ys_rpvQ--A@mail.gmail.com>
<[email protected]>
<CALj2ACXbUzp-9tixeUK_dws5k+ARJaXxj8cj9W3adA0XNUS4Hg@mail.gmail.com>
<[email protected]>
<CALj2ACW8SYiRL8vGAhez=FdrzxO0W5pVFvN8GF7yKCO8_c=QdQ@mail.gmail.com>
<CAA4eK1+-ipXB+LuUiDNQ5L+sWJFpzK89k-Oc7xOG37AefrptoQ@mail.gmail.com>
<CALj2ACWhTY9mRc7sOpuao=GGidRjCVXc0SNtJ_DNman4nh5CcQ@mail.gmail.com>
<CALj2ACVY+Fd5vC0VjW=5VDK9mmt-Y+PDZxnBp8ngGAZc24Vv9g@mail.gmail.com>
<CALj2ACX0FVrx9f0c2g19Rb-WAJhm3wdyttde3YQtkotL6bUveA@mail.gmail.com>
<CALj2ACXQNGXokgx8APwdxrG4MHMF=cOz6XQtUL7EHua9oUfkgA@mail.gmail.com>
<CAFPTHDZPOc7pY4E0zJ2om0MZLMt+UuiX-VLsJyRO_aSyR8DsnA@mail.gmail.com>
<CALj2ACVu0EJ+QQ7W0oHbL_+hCtk+p-00MNhE4hfjFcXaMepMfg@mail.gmail.com>
<CAA4eK1KVbicrQQbMLEQiHEi9=VJpxrErrK7CiMamwaMPfff3WA@mail.gmail.com>
<CALj2ACXe8+xSNdMXTMaSRWUwX7v61Ad4iddUwnn=djSwx3GLLg@mail.gmail.com>
<CAHut+PuFzCHPCiZbpoQX59kgZbebuWT0gR0O7rOe4t_sdYu=OA@mail.gmail.com>
<CALj2ACUJg2faYd-_Wr3NWoj_GtbW5Bje3G9+o595GyEhcBkbrg@mail.gmail.com>
<CAHut+PuC7nwuT3+36zsLxtBAcUHrSaysrHb0_dxTXqMnFBKZBA@mail.gmail.com>
<CALj2ACWXQT3_HY40ceqKf1DadjLQP6b1r=0sZRh-xhAOd-b0pA@mail.gmail.com>
<CAHut+PtJxxaVhjDCFPqE_V63VaAN_bL+PHRMak=dkghoGrZc0Q@mail.gmail.com>
<CALj2ACXjkM4nEE5dXeV=NA2NmbE3AmonKEgudYBMqU3qO9-9fw@mail.gmail.com>
<CAJpy0uAwxc49Dz6t=-y_-z-MU+A4RWX4BR3Zri_jj2qgGMq_8g@mail.gmail.com>
<CAJpy0uC6nN3SLbEuCvz7-CpaPdNdXxH=feW5MhYQch-JWV0tLg@mail.gmail.com>
<CAJpy0uBXXJC6f04+FU1axKaU+p78wN0SEhUNE9XoqbjXj=hhgw@mail.gmail.com>
<CAJpy0uB48aCJX8gpwwv-kGoOK3WYXWatB=1pgr__UW-OjwKkDg@mail.gmail.com>
<CABdArM44oW0dQa6Vv5c4xcY54rvhaKsBOpbDVaYzf0F3hUZFPw@mail.gmail.com>
<CALDaNm2VQW_gpOJ-QWkEA_h18DN31ELEz2_7QmwWCAg9=Zew4A@mail.gmail.com>
<CABdArM5141FHGvfNYD=pCybLDpmEWbRXFJW3HnEoieMScZ62-g@mail.gmail.com>
<CALDaNm0g86wD2=bQdFOy0smsP0MZWyz0CUqXej=Qi-hCEeqkag@mail.gmail.com>
On Wed, Nov 20, 2024 at 1:29 PM vignesh C <[email protected]> wrote:
>
> On Tue, 19 Nov 2024 at 12:43, Nisha Moond <[email protected]> wrote:
> >
> > Attached is the v49 patch set:
> > - Fixed the bug reported in [1].
> > - Addressed comments in [2] and [3].
> >
> > I've split the patch into two, implementing the suggested idea in
> > comment #5 of [2] separately in 001:
> >
> > Patch-001: Adds additional error reports (for all invalidation types)
> > in ReplicationSlotAcquire() for invalid slots when error_if_invalid =
> > true.
> > Patch-002: The original patch with comments addressed.
>
> This Assert can fail:
>
Attached v50 patch-set addressing review comments in [1] and [2].
Regarding the assert issue reported in [2]:
- For temporary replication slots, the current session's pid serves as
the active_pid for the slot, which is expected behavior.
- Therefore, the ASSERT has been removed in v50. Now, if a temporary
slot qualifies for a timeout invalidation, the holding process will be
terminated, and the slot will be invalidated.
[1] https://www.postgresql.org/message-id/CALDaNm2UUTfJczjR-rEQwKgmx%3DiFnuMnR1cXv7ccB%2BO9P15mYg%40mail...
[2] https://www.postgresql.org/message-id/CALDaNm0g86wD2%3DbQdFOy0smsP0MZWyz0CUqXej%3DQi-hCEeqkag%40mail...
--
Thanks,
Nisha
Attachments:
[application/octet-stream] v50-0001-Add-error-handling-while-acquiring-a-replication.patch (7.9K, ../CABdArM6A3xsedesUWaQYnOM+mAGGj--N2ohoLrk2tO-W8UfR5g@mail.gmail.com/2-v50-0001-Add-error-handling-while-acquiring-a-replication.patch)
download | inline diff:
From 6df6c33d07ec27a8d633e17c004cf2d6bb61c55b Mon Sep 17 00:00:00 2001
From: Nisha Moond <[email protected]>
Date: Mon, 18 Nov 2024 16:13:26 +0530
Subject: [PATCH v50 1/2] Add error handling while acquiring a replication slot
In ReplicationSlotAcquire(), raise an error for invalid slots if caller
specify error_if_invalid=true.
---
.../replication/logical/logicalfuncs.c | 2 +-
src/backend/replication/logical/slotsync.c | 4 +-
src/backend/replication/slot.c | 48 +++++++++++++++++--
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 +-
8 files changed, 55 insertions(+), 12 deletions(-)
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 d62186a510..69a32422bf 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 6828100cf1..0cbfd4fbaf 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -535,9 +535,12 @@ 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 has been
+ * invalidated previously.
*/
void
-ReplicationSlotAcquire(const char *name, bool nowait)
+ReplicationSlotAcquire(const char *name, bool nowait, bool error_if_invalid)
{
ReplicationSlot *s;
int active_pid;
@@ -615,6 +618,45 @@ retry:
/* We made this slot active, so it's ours now. */
MyReplicationSlot = s;
+ /*
+ * An error is raised if error_if_invalid is true and the slot has been
+ * previously invalidated.
+ */
+ if (error_if_invalid &&
+ s->data.invalidated != RS_INVAL_NONE)
+ {
+ StringInfoData err_detail;
+
+ initStringInfo(&err_detail);
+ appendStringInfo(&err_detail, _("This slot has been invalidated because "));
+
+ switch (s->data.invalidated)
+ {
+ case RS_INVAL_WAL_REMOVED:
+ appendStringInfo(&err_detail, _("the required WAL has been removed."));
+ break;
+
+ case RS_INVAL_HORIZON:
+ appendStringInfo(&err_detail, _("the required rows have been removed."));
+ break;
+
+ case RS_INVAL_WAL_LEVEL:
+ appendStringInfo(&err_detail, _("wal_level is insufficient for slot."));
+ break;
+
+ case RS_INVAL_NONE:
+ pg_unreachable();
+ }
+
+ ereport(ERROR,
+ errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("can no longer get changes from replication slot \"%s\"",
+ NameStr(s->data.name)),
+ errdetail_internal("%s", err_detail.data));
+
+ pfree(err_detail.data);
+ }
+
/*
* The call to pgstat_acquire_replslot() protects against stats for a
* different slot, from before a restart or such, being present during
@@ -785,7 +827,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 +854,7 @@ ReplicationSlotAlter(const char *name, const bool *failover,
Assert(MyReplicationSlot == NULL);
Assert(failover || two_phase);
- ReplicationSlotAcquire(name, false);
+ ReplicationSlotAcquire(name, false, false);
if (SlotIsPhysical(MyReplicationSlot))
ereport(ERROR,
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 488a161b3e..578cff64c8 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 371eef3ddd..b36ae90b2c 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 8a45b5827e..3322848e03 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, false);
Assert(SlotIsLogical(MyReplicationSlot));
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 45582cf9d8..2c325fd942 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -249,7 +249,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 efb4ba3af1..333e040e7f 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 slot has been invalidated because the required WAL has been removed",
$logstart))
{
$failed = 1;
--
2.34.1
[application/octet-stream] v50-0002-Introduce-inactive_timeout-based-replication-slo.patch (27.3K, ../CABdArM6A3xsedesUWaQYnOM+mAGGj--N2ohoLrk2tO-W8UfR5g@mail.gmail.com/3-v50-0002-Introduce-inactive_timeout-based-replication-slo.patch)
download | inline diff:
From 8871cb932a047f04ca4a6c38d24dacd77b6fef6c Mon Sep 17 00:00:00 2001
From: Nisha Moond <[email protected]>
Date: Mon, 18 Nov 2024 16:54:50 +0530
Subject: [PATCH v50 2/2] 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 replication_slot_inactive_timeout.
When set, postgres invalidates slots (during non-shutdown
checkpoints) that are inactive for longer than this amount of
time.
Note that the inactive 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.
---
doc/src/sgml/config.sgml | 35 ++++
doc/src/sgml/system-views.sgml | 12 +-
src/backend/replication/logical/slotsync.c | 13 +-
src/backend/replication/slot.c | 160 +++++++++++++--
src/backend/utils/misc/guc_tables.c | 12 ++
src/backend/utils/misc/postgresql.conf.sample | 1 +
src/include/replication/slot.h | 22 ++
src/include/utils/guc_hooks.h | 2 +
src/test/recovery/meson.build | 1 +
src/test/recovery/t/050_invalidate_slots.pl | 190 ++++++++++++++++++
10 files changed, 421 insertions(+), 27 deletions(-)
create mode 100644 src/test/recovery/t/050_invalidate_slots.pl
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index a84e60c09b..25ca232a02 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4585,6 +4585,41 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows
</listitem>
</varlistentry>
+ <varlistentry id="guc-replication-slot-inactive-timeout" xreflabel="replication_slot_inactive_timeout">
+ <term><varname>replication_slot_inactive_timeout</varname> (<type>integer</type>)
+ <indexterm>
+ <primary><varname>replication_slot_inactive_timeout</varname> configuration parameter</primary>
+ </indexterm>
+ </term>
+ <listitem>
+ <para>
+ Invalidate replication slots that are inactive for longer than this
+ amount of time. If this value is specified without units,
+ it is taken as seconds. A value of zero (which is default) disables
+ the inactive timeout invalidation mechanism. 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 inactive timeout occurs during checkpoint.
+ 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 inactive 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
+ <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-track-commit-timestamp" xreflabel="track_commit_timestamp">
<term><varname>track_commit_timestamp</varname> (<type>boolean</type>)
<indexterm>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index 61d28e701f..7eec92eac9 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2567,8 +2567,9 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
</para>
<para>
The time since the slot has become inactive.
- <literal>NULL</literal> if the slot is currently being used.
- Note that for slots on the standby that are being synced from a
+ <literal>NULL</literal> if the slot is currently being used. Once the
+ slot is invalidated, this value will remain unchanged until we shutdown
+ the server. 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> indicates the last
@@ -2618,6 +2619,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 remained
+ inactive beyond the duration specified by the
+ <xref linkend="guc-replication-slot-inactive-timeout"/> parameter.
+ </para>
+ </listitem>
</itemizedlist>
</para></entry>
</row>
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 69a32422bf..4dc2811f10 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -1508,7 +1508,7 @@ ReplSlotSyncWorkerMain(char *startup_data, size_t startup_data_len)
static void
update_synced_slots_inactive_since(void)
{
- TimestampTz now = 0;
+ TimestampTz now;
/*
* We need to update inactive_since only when we are promoting standby to
@@ -1523,6 +1523,9 @@ update_synced_slots_inactive_since(void)
/* The slot sync worker or SQL function mustn't be running by now */
Assert((SlotSyncCtx->pid == InvalidPid) && !SlotSyncCtx->syncing);
+ /* Use same inactive_since time for all slots */
+ now = GetCurrentTimestamp();
+
LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
for (int i = 0; i < max_replication_slots; i++)
@@ -1537,13 +1540,7 @@ update_synced_slots_inactive_since(void)
/* The slot must not be acquired by any process */
Assert(s->active_pid == 0);
- /* Use the same inactive_since time for all the slots. */
- 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 0cbfd4fbaf..d65b8a6376 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");
@@ -140,6 +141,7 @@ ReplicationSlot *MyReplicationSlot = NULL;
/* GUC variables */
int max_replication_slots = 10; /* the maximum number of replication
* slots */
+int replication_slot_inactive_timeout_sec = 0;
/*
* This GUC lists streaming replication standby server slot names that
@@ -644,6 +646,11 @@ retry:
appendStringInfo(&err_detail, _("wal_level is insufficient for slot."));
break;
+ case RS_INVAL_INACTIVE_TIMEOUT:
+ appendStringInfo(&err_detail, _("inactivity exceeded the time limit set by \"%s\"."),
+ "replication_slot_inactive_timeout");
+ break;
+
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -745,16 +752,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;
@@ -1550,7 +1553,8 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
NameData slotname,
XLogRecPtr restart_lsn,
XLogRecPtr oldestLSN,
- TransactionId snapshotConflictHorizon)
+ TransactionId snapshotConflictHorizon,
+ TimestampTz inactive_since)
{
StringInfoData err_detail;
bool hint = false;
@@ -1580,6 +1584,15 @@ 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:
+ Assert(inactive_since > 0);
+ appendStringInfo(&err_detail,
+ _("The slot has been inactive since %s, exceeding the time limit set by \"%s\"."),
+ timestamptz_to_str(inactive_since),
+ "replication_slot_inactive_timeout");
+ break;
+
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -1596,6 +1609,30 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
pfree(err_detail.data);
}
+/*
+ * Is inactive timeout invalidation possible for this replication slot?
+ *
+ * Inactive timeout invalidation is allowed only when:
+ *
+ * 1. Inactive timeout is set
+ * 2. Slot is inactive
+ * 3. The slot is not being synced from the primary while the server
+ * is in recovery
+ *
+ * Note that the inactive 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
+IsSlotInactiveTimeoutPossible(ReplicationSlot *s)
+{
+ return (replication_slot_inactive_timeout_sec > 0 &&
+ s->inactive_since > 0 &&
+ !(RecoveryInProgress() && s->data.synced));
+}
+
/*
* Helper for InvalidateObsoleteReplicationSlots
*
@@ -1623,6 +1660,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 (;;)
{
@@ -1630,6 +1668,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
NameData slotname;
int active_pid = 0;
ReplicationSlotInvalidationCause invalidation_cause = RS_INVAL_NONE;
+ TimestampTz now = 0;
Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
@@ -1640,6 +1679,16 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
break;
}
+ if (cause == RS_INVAL_INACTIVE_TIMEOUT &&
+ IsSlotInactiveTimeoutPossible(s))
+ {
+ /*
+ * 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
@@ -1693,6 +1742,20 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
if (SlotIsLogical(s))
invalidation_cause = cause;
break;
+ case RS_INVAL_INACTIVE_TIMEOUT:
+
+ /*
+ * Check if the slot needs to be invalidated due to
+ * replication_slot_inactive_timeout GUC.
+ */
+ if (now &&
+ TimestampDifferenceExceeds(s->inactive_since, now,
+ replication_slot_inactive_timeout_sec * 1000))
+ {
+ invalidation_cause = cause;
+ inactive_since = s->inactive_since;
+ }
+ break;
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -1718,11 +1781,13 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
active_pid = s->active_pid;
/*
- * If the slot can be acquired, do so and mark it invalidated
- * immediately. Otherwise we'll signal the owning process, below, and
- * retry.
+ * If the slot can be acquired, do so and mark it as invalidated. If
+ * the slot is already ours, mark it as invalidated. Otherwise, we'll
+ * signal the owning process below and retry.
*/
- if (active_pid == 0)
+ if (active_pid == 0 ||
+ (MyReplicationSlot == s &&
+ active_pid == MyProcPid))
{
MyReplicationSlot = s;
s->active_pid = MyProcPid;
@@ -1777,7 +1842,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,
@@ -1823,7 +1889,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;
@@ -1846,6 +1913,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 has occurred
*
* NB - this runs as part of checkpoint, so avoid raising errors if possible.
*/
@@ -1898,7 +1966,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
@@ -1956,6 +2025,38 @@ 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 inactive_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 inactive_timeout invalidation check closer to
+ * wal_removed in CreateCheckPoint and CreateRestartPoint?
+ */
+ InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT,
+ 0,
+ InvalidOid,
+ InvalidTransactionId);
+ }
}
/*
@@ -2250,6 +2351,7 @@ RestoreSlotFromDisk(const char *name)
bool restored = false;
int readBytes;
pg_crc32c checksum;
+ TimestampTz now;
/* no need to lock here, no concurrent access allowed yet */
@@ -2410,6 +2512,9 @@ RestoreSlotFromDisk(const char *name)
NameStr(cp.slotdata.name)),
errhint("Change \"wal_level\" to be \"replica\" or higher.")));
+ /* Use same inactive_since time for all slots */
+ now = GetCurrentTimestamp();
+
/* nothing can be active yet, don't lock anything */
for (i = 0; i < max_replication_slots; i++)
{
@@ -2440,9 +2545,11 @@ 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();
+ slot->inactive_since = now;
restored = true;
break;
@@ -2845,3 +2952,22 @@ WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
ConditionVariableCancelSleep();
}
+
+/*
+ * GUC check_hook for replication_slot_inactive_timeout
+ *
+ * We don't allow the value of replication_slot_inactive_timeout other than 0
+ * during the binary upgrade.
+ */
+bool
+check_replication_slot_inactive_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.",
+ "replication_slot_inactive_timeout");
+ return false;
+ }
+
+ return true;
+}
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 8a67f01200..f5a6bf4c61 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -3028,6 +3028,18 @@ struct config_int ConfigureNamesInt[] =
NULL, NULL, NULL
},
+ {
+ {"replication_slot_inactive_timeout", PGC_SIGHUP, REPLICATION_SENDING,
+ gettext_noop("Sets the amount of time a replication slot can remain inactive before "
+ "it will be invalidated."),
+ NULL,
+ GUC_UNIT_S
+ },
+ &replication_slot_inactive_timeout_sec,
+ 0, 0, INT_MAX,
+ check_replication_slot_inactive_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 39a3ac2312..7c6ae1baa2 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -336,6 +336,7 @@
#wal_sender_timeout = 60s # in milliseconds; 0 disables
#track_commit_timestamp = off # collect timestamp of transaction commit
# (change requires restart)
+#replication_slot_inactive_timeout = 0 # in seconds; 0 disables
# - Primary Server -
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 2c325fd942..98f858659c 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,
+ /* inactive slot timeout has occurred */
+ RS_INVAL_INACTIVE_TIMEOUT,
} ReplicationSlotInvalidationCause;
extern PGDLLIMPORT const char *const SlotInvalidationCauses[];
@@ -224,6 +226,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
*/
@@ -233,6 +254,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot;
/* GUCs */
extern PGDLLIMPORT int max_replication_slots;
extern PGDLLIMPORT char *synchronized_standby_slots;
+extern PGDLLIMPORT int replication_slot_inactive_timeout_sec;
/* shmem initialization functions */
extern Size ReplicationSlotsShmemSize(void);
diff --git a/src/include/utils/guc_hooks.h b/src/include/utils/guc_hooks.h
index 5813dba0a2..e36b6cfe21 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_replication_slot_inactive_timeout(int *newval, void **extra,
+ GucSource source);
#endif /* GUC_HOOKS_H */
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..6a10cd0a54
--- /dev/null
+++ b/src/test/recovery/t/050_invalidate_slots.pl
@@ -0,0 +1,190 @@
+# 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);
+
+# =============================================================================
+# Testcase start
+#
+# Test invalidation of streaming standby slot and logical failover slot on the
+# primary due to inactive 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
+autovacuum = off
+});
+$primary->start;
+
+# Take backup
+my $backup_name = 'my_backup';
+$primary->backup($backup_name);
+
+# 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'
+primary_conninfo = '$connstr dbname=postgres'
+));
+
+# 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);
+]);
+
+$standby1->start;
+
+# Wait until the standby has replayed enough data
+$primary->wait_for_catchup($standby1);
+
+my $logstart = -s $standby1->logfile;
+
+# Set timeout GUC on the standby to verify that the next checkpoint will not
+# invalidate synced slots.
+my $inactive_timeout_1s = 1;
+$standby1->safe_psql(
+ 'postgres', qq[
+ ALTER SYSTEM SET replication_slot_inactive_timeout TO '${inactive_timeout_1s}s';
+]);
+$standby1->reload;
+
+# 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
+sleep($inactive_timeout_1s + 1);
+
+# Despite inactive timeout being set, the synced slot won't get invalidated on
+# its own on the standby.
+$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');
+
+$logstart = -s $primary->logfile;
+
+# Set timeout GUC so that the next checkpoint will invalidate inactive slots
+$primary->safe_psql(
+ 'postgres', qq[
+ ALTER SYSTEM SET replication_slot_inactive_timeout TO '${inactive_timeout_1s}s';
+]);
+$primary->reload;
+
+# 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
+# inactive timeout.
+wait_for_slot_invalidation($primary, 'sync_slot1', $logstart,
+ $inactive_timeout_1s);
+
+# Re-sync the primary slots to the standby. Note that the primary slot was
+# already invalidated (above) due to inactive 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 = 'inactive_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,
+ $inactive_timeout_1s);
+
+# Testcase end
+# =============================================================================
+
+# Wait for slot to first become inactive and then get invalidated
+sub wait_for_slot_invalidation
+{
+ my ($node, $slot, $offset, $inactive_timeout_1s) = @_;
+ my $node_name = $node->name;
+
+ trigger_slot_invalidation($node, $slot, $offset, $inactive_timeout_1s);
+
+ # 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 get changes from 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, $inactive_timeout_1s) = @_;
+ my $node_name = $node->name;
+ my $invalidated = 0;
+
+ # Give enough time to avoid multiple checkpoints
+ sleep($inactive_timeout_1s + 1);
+
+ # 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 'inactive_timeout'
+ $node->poll_query_until(
+ 'postgres', qq[
+ SELECT COUNT(slot_name) = 1 FROM pg_replication_slots
+ WHERE slot_name = '$slot' AND
+ invalidation_reason = 'inactive_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 (98+ 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]
Subject: Re: Introduce XID age and inactive timeout based replication slot invalidation
In-Reply-To: <CABdArM6A3xsedesUWaQYnOM+mAGGj--N2ohoLrk2tO-W8UfR5g@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