public inbox for [email protected]
help / color / mirror / Atom feedFrom: Bharath Rupireddy <[email protected]>
To: Bertrand Drouvot <[email protected]>
Cc: Amit Kapila <[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, 22 Mar 2024 13:45:01 +0530
Message-ID: <CALj2ACWuNHJ3qBMmmy6i1YA9V8JDRSaciL2bBM+72bRcqiBE2g@mail.gmail.com> (raw)
In-Reply-To: <[email protected]>
References: <[email protected]>
<CALj2ACXib_+Kuy3_WbTBJkURYGwO7ngS8+HYATReJtXShNCtNw@mail.gmail.com>
<CAA4eK1KZyjdn7oXS3d7qJU5ag9ss4xSk0Jkjsh5NOrRdkPkV1g@mail.gmail.com>
<[email protected]>
<CALj2ACUvK7ShPyirAURWf62=qOQqu=NwnyL5CMfVjM5Ody7Oxw@mail.gmail.com>
<[email protected]>
<CALj2ACWk5a-LfBPziAjjFA--syoUSD_4jY6dZyhU8g8XuD-i7g@mail.gmail.com>
<CAA4eK1+46kJF3Z2Tv=xdkyy0nuurrS7RfzqvqY8bq1MF-JLM=A@mail.gmail.com>
<CALj2ACVKncd+JZR8BeXSgPO4akqic9kO99Z6pCfeXQKwmye2Og@mail.gmail.com>
<CAA4eK1JwJdsKMxczGO9tYgEzxp_L2=j0QTeDDk2Tx4-ipGSppw@mail.gmail.com>
<[email protected]>
On Fri, Mar 22, 2024 at 12:39 PM Bertrand Drouvot
<[email protected]> wrote:
>
> > > Please find the v14-0001 patch for now.
>
> Thanks!
>
> > LGTM. Let's wait for Bertrand to see if he has more comments on 0001
> > and then I'll push it.
>
> LGTM too.
Thanks. Here I'm implementing the following:
0001 Track invalidation_reason in pg_replication_slots
0002 Track last_inactive_at in pg_replication_slots
0003 Allow setting inactive_timeout for replication slots via SQL API
0004 Introduce new SQL funtion pg_alter_replication_slot
0005 Allow setting inactive_timeout in the replication command
0006 Add inactive_timeout based replication slot invalidation
1. Keep it last_inactive_at as a shared memory variable, but always
set it at restart if the slot's inactive_timeout has non-zero value
and reset it as soon as someone acquires that slot so that if the slot
doesn't get acquired till inactive_timeout, checkpointer will
invalidate the slot.
2. Ensure with pg_alter_replication_slot one could "only" alter the
timeout property for the time being, if not that could lead to the
subscription inconsistency.
3. Have some notes in the CREATE and ALTER SUBSCRIPTION docs about
using an existing slot to leverage inactive_timeout feature.
4. last_inactive_at should also be set to the current time during slot
creation because if one creates a slot and does nothing with it then
it's the time it starts to be inactive.
5. We don't set last_inactive_at to GetCurrentTimestamp() for failover slots.
6. Leave the patch that added support for inactive_timeout in subscriptions.
Please see the attached v14 patch set. No change in the attached
v14-0001 from the previous patch.
--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
Attachments:
[application/x-patch] v14-0001-Track-invalidation_reason-in-pg_replication_slot.patch (24.3K, ../CALj2ACWuNHJ3qBMmmy6i1YA9V8JDRSaciL2bBM+72bRcqiBE2g@mail.gmail.com/2-v14-0001-Track-invalidation_reason-in-pg_replication_slot.patch)
download | inline diff:
From 9ce64d4e6b629d70516bb2a16c5cbfa458c3a244 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Fri, 22 Mar 2024 02:56:12 +0000
Subject: [PATCH v14 1/6] Track invalidation_reason in pg_replication_slots
Up until now, reason for replication slot invalidation is not
tracked in pg_replication_slots. A recent commit 007693f2a added
'conflict_reason' to show the reasons for slot invalidation, but
only for logical slots.
This commit adds a new column 'invalidation_reason' to show
invalidation reasons for both physical and logical slots. And,
this commit also turns 'conflict_reason' text column to
'conflicting' boolean column (effectively reverting commit
007693f2a). The 'conflicting' column is true for inavlidation
reasons 'rows_removed' and 'wal_level_insufficient', because they
are the ones making the slot conflict with recovery. When
'conflicting' is true, one can now look at the new
'invalidation_reason' column for reason for logical slots conflict
with recovery.
The new 'invalidation_reason' column will also be useful when we
add more invalidation reasons in the future commit.
Author: Bharath Rupireddy
Reviewed-by: Bertrand Drouvot, Amit Kapila
Reviewed-by: shveta malik
Discussion: https://www.postgresql.org/message-id/ZfR7HuzFEswakt/a%40ip-10-97-1-34.eu-west-3.compute.internal
---
doc/src/sgml/ref/pgupgrade.sgml | 4 +-
doc/src/sgml/system-views.sgml | 25 +++++++---
src/backend/catalog/system_views.sql | 3 +-
src/backend/replication/logical/slotsync.c | 2 +-
src/backend/replication/slot.c | 49 +++++++++----------
src/backend/replication/slotfuncs.c | 25 +++++++---
src/bin/pg_upgrade/info.c | 4 +-
src/include/catalog/pg_proc.dat | 6 +--
src/include/replication/slot.h | 2 +-
.../t/035_standby_logical_decoding.pl | 35 ++++++-------
.../t/040_standby_failover_slots_sync.pl | 4 +-
src/test/regress/expected/rules.out | 5 +-
12 files changed, 93 insertions(+), 71 deletions(-)
diff --git a/doc/src/sgml/ref/pgupgrade.sgml b/doc/src/sgml/ref/pgupgrade.sgml
index 58c6c2df8b..8de52bf752 100644
--- a/doc/src/sgml/ref/pgupgrade.sgml
+++ b/doc/src/sgml/ref/pgupgrade.sgml
@@ -453,8 +453,8 @@ make prefix=/usr/local/pgsql.new install
<para>
All slots on the old cluster must be usable, i.e., there are no slots
whose
- <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>conflict_reason</structfield>
- is not <literal>NULL</literal>.
+ <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>conflicting</structfield>
+ is not <literal>true</literal>.
</para>
</listitem>
<listitem>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index be90edd0e2..b5da476c20 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2525,13 +2525,24 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
<row>
<entry role="catalog_table_entry"><para role="column_definition">
- <structfield>conflict_reason</structfield> <type>text</type>
+ <structfield>conflicting</structfield> <type>bool</type>
</para>
<para>
- The reason for the logical slot's conflict with recovery. It is always
- NULL for physical slots, as well as for logical slots which are not
- invalidated. The non-NULL values indicate that the slot is marked
- as invalidated. Possible values are:
+ True if this logical slot conflicted with recovery (and so is now
+ invalidated). When this column is true, check
+ <structfield>invalidation_reason</structfield> column for the conflict
+ reason. Always NULL for physical slots.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>invalidation_reason</structfield> <type>text</type>
+ </para>
+ <para>
+ The reason for the slot's invalidation. It is set for both logical and
+ physical slots. <literal>NULL</literal> if the slot is not invalidated.
+ Possible values are:
<itemizedlist spacing="compact">
<listitem>
<para>
@@ -2542,14 +2553,14 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
<listitem>
<para>
<literal>rows_removed</literal> means that the required rows have
- been removed.
+ been removed. It is set only for logical slots.
</para>
</listitem>
<listitem>
<para>
<literal>wal_level_insufficient</literal> means that the
primary doesn't have a <xref linkend="guc-wal-level"/> sufficient to
- perform logical decoding.
+ perform logical decoding. It is set only for logical slots.
</para>
</listitem>
</itemizedlist>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 04227a72d1..f69b7f5580 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1023,7 +1023,8 @@ CREATE VIEW pg_replication_slots AS
L.wal_status,
L.safe_wal_size,
L.two_phase,
- L.conflict_reason,
+ L.conflicting,
+ L.invalidation_reason,
L.failover,
L.synced
FROM pg_get_replication_slots() AS L
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 7b180bdb5c..30480960c5 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -663,7 +663,7 @@ synchronize_slots(WalReceiverConn *wrconn)
bool started_tx = false;
const char *query = "SELECT slot_name, plugin, confirmed_flush_lsn,"
" restart_lsn, catalog_xmin, two_phase, failover,"
- " database, conflict_reason"
+ " database, invalidation_reason"
" FROM pg_catalog.pg_replication_slots"
" WHERE failover and NOT temporary";
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 91ca397857..cdf0c450c5 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -1525,14 +1525,14 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
XLogRecPtr initial_effective_xmin = InvalidXLogRecPtr;
XLogRecPtr initial_catalog_effective_xmin = InvalidXLogRecPtr;
XLogRecPtr initial_restart_lsn = InvalidXLogRecPtr;
- ReplicationSlotInvalidationCause conflict_prev PG_USED_FOR_ASSERTS_ONLY = RS_INVAL_NONE;
+ ReplicationSlotInvalidationCause invalidation_cause_prev PG_USED_FOR_ASSERTS_ONLY = RS_INVAL_NONE;
for (;;)
{
XLogRecPtr restart_lsn;
NameData slotname;
int active_pid = 0;
- ReplicationSlotInvalidationCause conflict = RS_INVAL_NONE;
+ ReplicationSlotInvalidationCause invalidation_cause = RS_INVAL_NONE;
Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
@@ -1554,17 +1554,14 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
restart_lsn = s->data.restart_lsn;
- /*
- * If the slot is already invalid or is a non conflicting slot, we
- * don't need to do anything.
- */
+ /* we do nothing if the slot is already invalid */
if (s->data.invalidated == RS_INVAL_NONE)
{
/*
* The slot's mutex will be released soon, and it is possible that
* those values change since the process holding the slot has been
* terminated (if any), so record them here to ensure that we
- * would report the correct conflict cause.
+ * would report the correct invalidation cause.
*/
if (!terminated)
{
@@ -1578,7 +1575,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
case RS_INVAL_WAL_REMOVED:
if (initial_restart_lsn != InvalidXLogRecPtr &&
initial_restart_lsn < oldestLSN)
- conflict = cause;
+ invalidation_cause = cause;
break;
case RS_INVAL_HORIZON:
if (!SlotIsLogical(s))
@@ -1589,15 +1586,15 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
if (TransactionIdIsValid(initial_effective_xmin) &&
TransactionIdPrecedesOrEquals(initial_effective_xmin,
snapshotConflictHorizon))
- conflict = cause;
+ invalidation_cause = cause;
else if (TransactionIdIsValid(initial_catalog_effective_xmin) &&
TransactionIdPrecedesOrEquals(initial_catalog_effective_xmin,
snapshotConflictHorizon))
- conflict = cause;
+ invalidation_cause = cause;
break;
case RS_INVAL_WAL_LEVEL:
if (SlotIsLogical(s))
- conflict = cause;
+ invalidation_cause = cause;
break;
case RS_INVAL_NONE:
pg_unreachable();
@@ -1605,14 +1602,14 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
}
/*
- * The conflict cause recorded previously should not change while the
- * process owning the slot (if any) has been terminated.
+ * The invalidation cause recorded previously should not change while
+ * the process owning the slot (if any) has been terminated.
*/
- Assert(!(conflict_prev != RS_INVAL_NONE && terminated &&
- conflict_prev != conflict));
+ Assert(!(invalidation_cause_prev != RS_INVAL_NONE && terminated &&
+ invalidation_cause_prev != invalidation_cause));
- /* if there's no conflict, we're done */
- if (conflict == RS_INVAL_NONE)
+ /* if there's no invalidation, we're done */
+ if (invalidation_cause == RS_INVAL_NONE)
{
SpinLockRelease(&s->mutex);
if (released_lock)
@@ -1632,13 +1629,13 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
{
MyReplicationSlot = s;
s->active_pid = MyProcPid;
- s->data.invalidated = conflict;
+ s->data.invalidated = invalidation_cause;
/*
* XXX: We should consider not overwriting restart_lsn and instead
* just rely on .invalidated.
*/
- if (conflict == RS_INVAL_WAL_REMOVED)
+ if (invalidation_cause == RS_INVAL_WAL_REMOVED)
s->data.restart_lsn = InvalidXLogRecPtr;
/* Let caller know */
@@ -1681,7 +1678,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
*/
if (last_signaled_pid != active_pid)
{
- ReportSlotInvalidation(conflict, true, active_pid,
+ ReportSlotInvalidation(invalidation_cause, true, active_pid,
slotname, restart_lsn,
oldestLSN, snapshotConflictHorizon);
@@ -1694,7 +1691,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
last_signaled_pid = active_pid;
terminated = true;
- conflict_prev = conflict;
+ invalidation_cause_prev = invalidation_cause;
}
/* Wait until the slot is released. */
@@ -1727,7 +1724,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
ReplicationSlotSave();
ReplicationSlotRelease();
- ReportSlotInvalidation(conflict, false, active_pid,
+ ReportSlotInvalidation(invalidation_cause, false, active_pid,
slotname, restart_lsn,
oldestLSN, snapshotConflictHorizon);
@@ -2356,21 +2353,21 @@ RestoreSlotFromDisk(const char *name)
}
/*
- * Maps a conflict reason for a replication slot to
+ * Maps an invalidation reason for a replication slot to
* ReplicationSlotInvalidationCause.
*/
ReplicationSlotInvalidationCause
-GetSlotInvalidationCause(const char *conflict_reason)
+GetSlotInvalidationCause(const char *invalidation_reason)
{
ReplicationSlotInvalidationCause cause;
ReplicationSlotInvalidationCause result = RS_INVAL_NONE;
bool found PG_USED_FOR_ASSERTS_ONLY = false;
- Assert(conflict_reason);
+ Assert(invalidation_reason);
for (cause = RS_INVAL_NONE; cause <= RS_INVAL_MAX_CAUSES; cause++)
{
- if (strcmp(SlotInvalidationCauses[cause], conflict_reason) == 0)
+ if (strcmp(SlotInvalidationCauses[cause], invalidation_reason) == 0)
{
found = true;
result = cause;
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index ad79e1fccd..4232c1e52e 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -239,7 +239,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_get_replication_slots(PG_FUNCTION_ARGS)
{
-#define PG_GET_REPLICATION_SLOTS_COLS 17
+#define PG_GET_REPLICATION_SLOTS_COLS 18
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
XLogRecPtr currlsn;
int slotno;
@@ -263,6 +263,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
bool nulls[PG_GET_REPLICATION_SLOTS_COLS];
WALAvailability walstate;
int i;
+ ReplicationSlotInvalidationCause cause;
if (!slot->in_use)
continue;
@@ -409,18 +410,28 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
values[i++] = BoolGetDatum(slot_contents.data.two_phase);
- if (slot_contents.data.database == InvalidOid)
+ cause = slot_contents.data.invalidated;
+
+ if (SlotIsPhysical(&slot_contents))
nulls[i++] = true;
else
{
- ReplicationSlotInvalidationCause cause = slot_contents.data.invalidated;
-
- if (cause == RS_INVAL_NONE)
- nulls[i++] = true;
+ /*
+ * rows_removed and wal_level_insufficient are the only two
+ * reasons for the logical slot's conflict with recovery.
+ */
+ if (cause == RS_INVAL_HORIZON ||
+ cause == RS_INVAL_WAL_LEVEL)
+ values[i++] = BoolGetDatum(true);
else
- values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]);
+ values[i++] = BoolGetDatum(false);
}
+ if (cause == RS_INVAL_NONE)
+ nulls[i++] = true;
+ else
+ values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]);
+
values[i++] = BoolGetDatum(slot_contents.data.failover);
values[i++] = BoolGetDatum(slot_contents.data.synced);
diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c
index b5b8d11602..95c22a7200 100644
--- a/src/bin/pg_upgrade/info.c
+++ b/src/bin/pg_upgrade/info.c
@@ -676,13 +676,13 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
* removed.
*/
res = executeQueryOrDie(conn, "SELECT slot_name, plugin, two_phase, failover, "
- "%s as caught_up, conflict_reason IS NOT NULL as invalid "
+ "%s as caught_up, invalidation_reason IS NOT NULL as invalid "
"FROM pg_catalog.pg_replication_slots "
"WHERE slot_type = 'logical' AND "
"database = current_database() AND "
"temporary IS FALSE;",
live_check ? "FALSE" :
- "(CASE WHEN conflict_reason IS NOT NULL THEN FALSE "
+ "(CASE WHEN invalidation_reason IS NOT NULL THEN FALSE "
"ELSE (SELECT pg_catalog.binary_upgrade_logical_slot_has_caught_up(slot_name)) "
"END)");
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 042f66f714..71c74350a0 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11133,9 +11133,9 @@
proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', prorettype => 'record',
proargtypes => '',
- proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,text,bool,bool}',
- proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflict_reason,failover,synced}',
+ proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,text,bool,bool}',
+ proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,invalidation_reason,failover,synced}',
prosrc => 'pg_get_replication_slots' },
{ oid => '3786', descr => 'set up a logical replication slot',
proname => 'pg_create_logical_replication_slot', provolatile => 'v',
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 425effad21..7f25a083ee 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -273,7 +273,7 @@ extern void CheckPointReplicationSlots(bool is_shutdown);
extern void CheckSlotRequirements(void);
extern void CheckSlotPermissions(void);
extern ReplicationSlotInvalidationCause
- GetSlotInvalidationCause(const char *conflict_reason);
+ GetSlotInvalidationCause(const char *invalidation_reason);
extern bool SlotExistsInStandbySlotNames(const char *slot_name);
extern bool StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel);
diff --git a/src/test/recovery/t/035_standby_logical_decoding.pl b/src/test/recovery/t/035_standby_logical_decoding.pl
index 88b03048c4..8d6740c734 100644
--- a/src/test/recovery/t/035_standby_logical_decoding.pl
+++ b/src/test/recovery/t/035_standby_logical_decoding.pl
@@ -168,7 +168,7 @@ sub change_hot_standby_feedback_and_wait_for_xmins
}
}
-# Check conflict_reason in pg_replication_slots.
+# Check reason for conflict in pg_replication_slots.
sub check_slots_conflict_reason
{
my ($slot_prefix, $reason) = @_;
@@ -178,15 +178,15 @@ sub check_slots_conflict_reason
$res = $node_standby->safe_psql(
'postgres', qq(
- select conflict_reason from pg_replication_slots where slot_name = '$active_slot';));
+ select invalidation_reason from pg_replication_slots where slot_name = '$active_slot' and conflicting;));
- is($res, "$reason", "$active_slot conflict_reason is $reason");
+ is($res, "$reason", "$active_slot reason for conflict is $reason");
$res = $node_standby->safe_psql(
'postgres', qq(
- select conflict_reason from pg_replication_slots where slot_name = '$inactive_slot';));
+ select invalidation_reason from pg_replication_slots where slot_name = '$inactive_slot' and conflicting;));
- is($res, "$reason", "$inactive_slot conflict_reason is $reason");
+ is($res, "$reason", "$inactive_slot reason for conflict is $reason");
}
# Drop the slots, re-create them, change hot_standby_feedback,
@@ -293,13 +293,13 @@ $node_primary->safe_psql('testdb',
qq[SELECT * FROM pg_create_physical_replication_slot('$primary_slotname');]
);
-# Check conflict_reason is NULL for physical slot
+# Check conflicting is NULL for physical slot
$res = $node_primary->safe_psql(
'postgres', qq[
- SELECT conflict_reason is null FROM pg_replication_slots where slot_name = '$primary_slotname';]
+ SELECT conflicting is null FROM pg_replication_slots where slot_name = '$primary_slotname';]
);
-is($res, 't', "Physical slot reports conflict_reason as NULL");
+is($res, 't', "Physical slot reports conflicting as NULL");
my $backup_name = 'b1';
$node_primary->backup($backup_name);
@@ -524,7 +524,7 @@ $node_primary->wait_for_replay_catchup($node_standby);
# Check invalidation in the logfile and in pg_stat_database_conflicts
check_for_invalidation('vacuum_full_', 1, 'with vacuum FULL on pg_class');
-# Verify conflict_reason is 'rows_removed' in pg_replication_slots
+# Verify reason for conflict is 'rows_removed' in pg_replication_slots
check_slots_conflict_reason('vacuum_full_', 'rows_removed');
# Ensure that replication slot stats are not removed after invalidation.
@@ -551,7 +551,7 @@ change_hot_standby_feedback_and_wait_for_xmins(1, 1);
##################################################
$node_standby->restart;
-# Verify conflict_reason is retained across a restart.
+# Verify reason for conflict is retained across a restart.
check_slots_conflict_reason('vacuum_full_', 'rows_removed');
##################################################
@@ -560,7 +560,8 @@ check_slots_conflict_reason('vacuum_full_', 'rows_removed');
# Get the restart_lsn from an invalidated slot
my $restart_lsn = $node_standby->safe_psql('postgres',
- "SELECT restart_lsn from pg_replication_slots WHERE slot_name = 'vacuum_full_activeslot' and conflict_reason is not null;"
+ "SELECT restart_lsn FROM pg_replication_slots
+ WHERE slot_name = 'vacuum_full_activeslot' AND conflicting;"
);
chomp($restart_lsn);
@@ -611,7 +612,7 @@ $node_primary->wait_for_replay_catchup($node_standby);
# Check invalidation in the logfile and in pg_stat_database_conflicts
check_for_invalidation('row_removal_', $logstart, 'with vacuum on pg_class');
-# Verify conflict_reason is 'rows_removed' in pg_replication_slots
+# Verify reason for conflict is 'rows_removed' in pg_replication_slots
check_slots_conflict_reason('row_removal_', 'rows_removed');
$handle =
@@ -647,7 +648,7 @@ $node_primary->wait_for_replay_catchup($node_standby);
check_for_invalidation('shared_row_removal_', $logstart,
'with vacuum on pg_authid');
-# Verify conflict_reason is 'rows_removed' in pg_replication_slots
+# Verify reason for conflict is 'rows_removed' in pg_replication_slots
check_slots_conflict_reason('shared_row_removal_', 'rows_removed');
$handle = make_slot_active($node_standby, 'shared_row_removal_', 0, \$stdout,
@@ -700,8 +701,8 @@ ok( $node_standby->poll_query_until(
is( $node_standby->safe_psql(
'postgres',
q[select bool_or(conflicting) from
- (select conflict_reason is not NULL as conflicting
- from pg_replication_slots WHERE slot_type = 'logical')]),
+ (select conflicting from pg_replication_slots
+ where slot_type = 'logical')]),
'f',
'Logical slots are reported as non conflicting');
@@ -739,7 +740,7 @@ $node_primary->wait_for_replay_catchup($node_standby);
# Check invalidation in the logfile and in pg_stat_database_conflicts
check_for_invalidation('pruning_', $logstart, 'with on-access pruning');
-# Verify conflict_reason is 'rows_removed' in pg_replication_slots
+# Verify reason for conflict is 'rows_removed' in pg_replication_slots
check_slots_conflict_reason('pruning_', 'rows_removed');
$handle = make_slot_active($node_standby, 'pruning_', 0, \$stdout, \$stderr);
@@ -783,7 +784,7 @@ $node_primary->wait_for_replay_catchup($node_standby);
# Check invalidation in the logfile and in pg_stat_database_conflicts
check_for_invalidation('wal_level_', $logstart, 'due to wal_level');
-# Verify conflict_reason is 'wal_level_insufficient' in pg_replication_slots
+# Verify reason for conflict is 'wal_level_insufficient' in pg_replication_slots
check_slots_conflict_reason('wal_level_', 'wal_level_insufficient');
$handle =
diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl
index 0ea1f3d323..f47bfd78eb 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -228,7 +228,7 @@ $standby1->safe_psql('postgres', "CHECKPOINT");
# Check if the synced slot is invalidated
is( $standby1->safe_psql(
'postgres',
- q{SELECT conflict_reason = 'wal_removed' FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';}
+ q{SELECT invalidation_reason = 'wal_removed' FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';}
),
"t",
'synchronized slot has been invalidated');
@@ -274,7 +274,7 @@ $standby1->wait_for_log(qr/dropped replication slot "lsub1_slot" of dbid [0-9]+/
# flagged as 'synced'
is( $standby1->safe_psql(
'postgres',
- q{SELECT conflict_reason IS NULL AND synced AND NOT temporary FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';}
+ q{SELECT invalidation_reason IS NULL AND synced AND NOT temporary FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';}
),
"t",
'logical slot is re-synced');
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 84e359f6ed..18829ea586 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1473,10 +1473,11 @@ pg_replication_slots| SELECT l.slot_name,
l.wal_status,
l.safe_wal_size,
l.two_phase,
- l.conflict_reason,
+ l.conflicting,
+ l.invalidation_reason,
l.failover,
l.synced
- FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflict_reason, failover, synced)
+ FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, invalidation_reason, failover, synced)
LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
pg_roles| SELECT pg_authid.rolname,
pg_authid.rolsuper,
--
2.34.1
[application/x-patch] v14-0002-Track-last_inactive_at-in-pg_replication_slots.patch (6.9K, ../CALj2ACWuNHJ3qBMmmy6i1YA9V8JDRSaciL2bBM+72bRcqiBE2g@mail.gmail.com/3-v14-0002-Track-last_inactive_at-in-pg_replication_slots.patch)
download | inline diff:
From eb5bb019863bb7eaa872357b3d9985f59dc56e9f Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Fri, 22 Mar 2024 02:59:07 +0000
Subject: [PATCH v14 2/6] Track last_inactive_at in pg_replication_slots
---
doc/src/sgml/system-views.sgml | 11 +++++++++++
src/backend/catalog/system_views.sql | 3 ++-
src/backend/replication/slot.c | 16 ++++++++++++++++
src/backend/replication/slotfuncs.c | 7 ++++++-
src/include/catalog/pg_proc.dat | 6 +++---
src/include/replication/slot.h | 3 +++
src/test/regress/expected/rules.out | 5 +++--
7 files changed, 44 insertions(+), 7 deletions(-)
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index b5da476c20..61378b3d4b 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2592,6 +2592,17 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>last_inactive_at</structfield> <type>timestamptz</type>
+ </para>
+ <para>
+ The time at which the slot became inactive.
+ <literal>NULL</literal> if the slot is currently actively being
+ used.
+ </para></entry>
+ </row>
+
</tbody>
</tgroup>
</table>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index f69b7f5580..e17f979c7f 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1026,7 +1026,8 @@ CREATE VIEW pg_replication_slots AS
L.conflicting,
L.invalidation_reason,
L.failover,
- L.synced
+ L.synced,
+ L.last_inactive_at
FROM pg_get_replication_slots() AS L
LEFT JOIN pg_database D ON (L.datoid = D.oid);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index cdf0c450c5..146f0fbf84 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -409,6 +409,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
slot->candidate_restart_valid = InvalidXLogRecPtr;
slot->candidate_restart_lsn = InvalidXLogRecPtr;
slot->last_saved_confirmed_flush = InvalidXLogRecPtr;
+ slot->last_inactive_at = 0;
/*
* Create the slot on disk. We haven't actually marked the slot allocated
@@ -622,6 +623,13 @@ retry:
if (SlotIsLogical(s))
pgstat_acquire_replslot(s);
+ if (s->data.persistency == RS_PERSISTENT)
+ {
+ SpinLockAcquire(&s->mutex);
+ s->last_inactive_at = 0;
+ SpinLockRelease(&s->mutex);
+ }
+
if (am_walsender)
{
ereport(log_replication_commands ? LOG : DEBUG1,
@@ -691,6 +699,13 @@ ReplicationSlotRelease(void)
ConditionVariableBroadcast(&slot->active_cv);
}
+ if (slot->data.persistency == RS_PERSISTENT)
+ {
+ SpinLockAcquire(&slot->mutex);
+ slot->last_inactive_at = GetCurrentTimestamp();
+ SpinLockRelease(&slot->mutex);
+ }
+
MyReplicationSlot = NULL;
/* might not have been set when we've been a plain slot */
@@ -2341,6 +2356,7 @@ RestoreSlotFromDisk(const char *name)
slot->in_use = true;
slot->active_pid = 0;
+ slot->last_inactive_at = 0;
restored = true;
break;
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 4232c1e52e..75300f24b6 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -239,7 +239,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_get_replication_slots(PG_FUNCTION_ARGS)
{
-#define PG_GET_REPLICATION_SLOTS_COLS 18
+#define PG_GET_REPLICATION_SLOTS_COLS 19
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
XLogRecPtr currlsn;
int slotno;
@@ -436,6 +436,11 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
values[i++] = BoolGetDatum(slot_contents.data.synced);
+ if (slot_contents.last_inactive_at > 0)
+ values[i++] = TimestampTzGetDatum(slot_contents.last_inactive_at);
+ else
+ nulls[i++] = true;
+
Assert(i == PG_GET_REPLICATION_SLOTS_COLS);
tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 71c74350a0..bf13448ad4 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11133,9 +11133,9 @@
proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', prorettype => 'record',
proargtypes => '',
- proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,text,bool,bool}',
- proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,invalidation_reason,failover,synced}',
+ proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,text,bool,bool,timestamptz}',
+ proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,invalidation_reason,failover,synced,last_inactive_at}',
prosrc => 'pg_get_replication_slots' },
{ oid => '3786', descr => 'set up a logical replication slot',
proname => 'pg_create_logical_replication_slot', provolatile => 'v',
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 7f25a083ee..b4bb7f5e99 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -201,6 +201,9 @@ typedef struct ReplicationSlot
* forcibly flushed or not.
*/
XLogRecPtr last_saved_confirmed_flush;
+
+ /* When did this slot become inactive last time? */
+ TimestampTz last_inactive_at;
} ReplicationSlot;
#define SlotIsPhysical(slot) ((slot)->data.database == InvalidOid)
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 18829ea586..effee2879e 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1476,8 +1476,9 @@ pg_replication_slots| SELECT l.slot_name,
l.conflicting,
l.invalidation_reason,
l.failover,
- l.synced
- FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, invalidation_reason, failover, synced)
+ l.synced,
+ l.last_inactive_at
+ FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, invalidation_reason, failover, synced, last_inactive_at)
LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
pg_roles| SELECT pg_authid.rolname,
pg_authid.rolsuper,
--
2.34.1
[application/x-patch] v14-0003-Allow-setting-inactive_timeout-for-replication-s.patch (35.4K, ../CALj2ACWuNHJ3qBMmmy6i1YA9V8JDRSaciL2bBM+72bRcqiBE2g@mail.gmail.com/4-v14-0003-Allow-setting-inactive_timeout-for-replication-s.patch)
download | inline diff:
From 1fe40bcdd725858eab8d414d80f1234dcd0a4835 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Fri, 22 Mar 2024 03:55:02 +0000
Subject: [PATCH v14 3/6] Allow setting inactive_timeout for replication slots
via SQL API
---
contrib/test_decoding/expected/slot.out | 102 ++++++++++++++++++
contrib/test_decoding/sql/slot.sql | 34 ++++++
doc/src/sgml/func.sgml | 18 ++--
doc/src/sgml/system-views.sgml | 9 ++
src/backend/catalog/system_functions.sql | 2 +
src/backend/catalog/system_views.sql | 3 +-
src/backend/replication/logical/slotsync.c | 17 ++-
src/backend/replication/slot.c | 20 +++-
src/backend/replication/slotfuncs.c | 31 +++++-
src/backend/replication/walsender.c | 4 +-
src/bin/pg_upgrade/info.c | 6 +-
src/bin/pg_upgrade/pg_upgrade.c | 5 +-
src/bin/pg_upgrade/pg_upgrade.h | 2 +
src/bin/pg_upgrade/t/003_logical_slots.pl | 11 +-
src/include/catalog/pg_proc.dat | 22 ++--
src/include/replication/slot.h | 5 +-
.../t/040_standby_failover_slots_sync.pl | 13 ++-
src/test/regress/expected/rules.out | 5 +-
18 files changed, 266 insertions(+), 43 deletions(-)
diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out
index 349ab2d380..6771520afb 100644
--- a/contrib/test_decoding/expected/slot.out
+++ b/contrib/test_decoding/expected/slot.out
@@ -466,3 +466,105 @@ SELECT pg_drop_replication_slot('physical_slot');
(1 row)
+-- Test negative value for inactive_timeout option for slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_fail_slot', inactive_timeout := -300); -- error
+ERROR: "inactive_timeout" must not be negative
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_fail_slot', plugin := 'test_decoding', inactive_timeout := -600); -- error
+ERROR: "inactive_timeout" must not be negative
+-- Test inactive_timeout option for temporary slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_fail_slot', temporary := true, inactive_timeout := 300); -- error
+ERROR: cannot set inactive_timeout for a temporary replication slot
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_fail_slot', plugin := 'test_decoding', temporary := true, inactive_timeout := 600); -- error
+ERROR: cannot set inactive_timeout for a temporary replication slot
+-- Test inactive_timeout option of physical slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot1', immediately_reserve := true, inactive_timeout := 300);
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot2');
+ ?column?
+----------
+ init
+(1 row)
+
+-- Copy physical slot with inactive_timeout option set.
+SELECT 'copy' FROM pg_copy_physical_replication_slot(src_slot_name := 'it_phy_slot1', dst_slot_name := 'it_phy_slot3');
+ ?column?
+----------
+ copy
+(1 row)
+
+SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
+ slot_name | slot_type | inactive_timeout
+--------------+-----------+------------------
+ it_phy_slot1 | physical | 300
+ it_phy_slot2 | physical | 0
+ it_phy_slot3 | physical | 300
+(3 rows)
+
+SELECT pg_drop_replication_slot('it_phy_slot1');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('it_phy_slot2');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('it_phy_slot3');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+-- Test inactive_timeout option of logical slots.
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot1', plugin := 'test_decoding', inactive_timeout := 600);
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot2', plugin := 'test_decoding');
+ ?column?
+----------
+ init
+(1 row)
+
+-- Copy logical slot with inactive_timeout option set.
+SELECT 'copy' FROM pg_copy_logical_replication_slot(src_slot_name := 'it_log_slot1', dst_slot_name := 'it_log_slot3');
+ ?column?
+----------
+ copy
+(1 row)
+
+SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
+ slot_name | slot_type | inactive_timeout
+--------------+-----------+------------------
+ it_log_slot1 | logical | 600
+ it_log_slot2 | logical | 0
+ it_log_slot3 | logical | 600
+(3 rows)
+
+SELECT pg_drop_replication_slot('it_log_slot1');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('it_log_slot2');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('it_log_slot3');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
diff --git a/contrib/test_decoding/sql/slot.sql b/contrib/test_decoding/sql/slot.sql
index 580e3ae3be..443e91da07 100644
--- a/contrib/test_decoding/sql/slot.sql
+++ b/contrib/test_decoding/sql/slot.sql
@@ -190,3 +190,37 @@ SELECT pg_drop_replication_slot('failover_true_slot');
SELECT pg_drop_replication_slot('failover_false_slot');
SELECT pg_drop_replication_slot('failover_default_slot');
SELECT pg_drop_replication_slot('physical_slot');
+
+-- Test negative value for inactive_timeout option for slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_fail_slot', inactive_timeout := -300); -- error
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_fail_slot', plugin := 'test_decoding', inactive_timeout := -600); -- error
+
+-- Test inactive_timeout option for temporary slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_fail_slot', temporary := true, inactive_timeout := 300); -- error
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_fail_slot', plugin := 'test_decoding', temporary := true, inactive_timeout := 600); -- error
+
+-- Test inactive_timeout option of physical slots.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot1', immediately_reserve := true, inactive_timeout := 300);
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot2');
+
+-- Copy physical slot with inactive_timeout option set.
+SELECT 'copy' FROM pg_copy_physical_replication_slot(src_slot_name := 'it_phy_slot1', dst_slot_name := 'it_phy_slot3');
+
+SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
+
+SELECT pg_drop_replication_slot('it_phy_slot1');
+SELECT pg_drop_replication_slot('it_phy_slot2');
+SELECT pg_drop_replication_slot('it_phy_slot3');
+
+-- Test inactive_timeout option of logical slots.
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot1', plugin := 'test_decoding', inactive_timeout := 600);
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot2', plugin := 'test_decoding');
+
+-- Copy logical slot with inactive_timeout option set.
+SELECT 'copy' FROM pg_copy_logical_replication_slot(src_slot_name := 'it_log_slot1', dst_slot_name := 'it_log_slot3');
+
+SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
+
+SELECT pg_drop_replication_slot('it_log_slot1');
+SELECT pg_drop_replication_slot('it_log_slot2');
+SELECT pg_drop_replication_slot('it_log_slot3');
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index 8ecc02f2b9..afaafa35ad 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -28373,7 +28373,7 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
<indexterm>
<primary>pg_create_physical_replication_slot</primary>
</indexterm>
- <function>pg_create_physical_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type> <optional>, <parameter>immediately_reserve</parameter> <type>boolean</type>, <parameter>temporary</parameter> <type>boolean</type> </optional> )
+ <function>pg_create_physical_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type> <optional>, <parameter>immediately_reserve</parameter> <type>boolean</type>, <parameter>temporary</parameter> <type>boolean</type>, <parameter>inactive_timeout</parameter> <type>integer</type> </optional>)
<returnvalue>record</returnvalue>
( <parameter>slot_name</parameter> <type>name</type>,
<parameter>lsn</parameter> <type>pg_lsn</type> )
@@ -28390,9 +28390,12 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
parameter, <parameter>temporary</parameter>, when set to true, specifies that
the slot should not be permanently stored to disk and is only meant
for use by the current session. Temporary slots are also
- released upon any error. This function corresponds
- to the replication protocol command <literal>CREATE_REPLICATION_SLOT
- ... PHYSICAL</literal>.
+ released upon any error. The optional fourth
+ parameter, <parameter>inactive_timeout</parameter>, when set to a
+ non-zero value, specifies the amount of time in seconds the slot is
+ allowed to be inactive. This function corresponds to the replication
+ protocol command
+ <literal>CREATE_REPLICATION_SLOT ... PHYSICAL</literal>.
</para></entry>
</row>
@@ -28417,7 +28420,7 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
<indexterm>
<primary>pg_create_logical_replication_slot</primary>
</indexterm>
- <function>pg_create_logical_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type>, <parameter>plugin</parameter> <type>name</type> <optional>, <parameter>temporary</parameter> <type>boolean</type>, <parameter>twophase</parameter> <type>boolean</type>, <parameter>failover</parameter> <type>boolean</type> </optional> )
+ <function>pg_create_logical_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type>, <parameter>plugin</parameter> <type>name</type> <optional>, <parameter>temporary</parameter> <type>boolean</type>, <parameter>twophase</parameter> <type>boolean</type>, <parameter>failover</parameter> <type>boolean</type>, <parameter>inactive_timeout</parameter> <type>integer</type> </optional> )
<returnvalue>record</returnvalue>
( <parameter>slot_name</parameter> <type>name</type>,
<parameter>lsn</parameter> <type>pg_lsn</type> )
@@ -28436,7 +28439,10 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
<parameter>failover</parameter>, when set to true,
specifies that this slot is enabled to be synced to the
standbys so that logical replication can be resumed after
- failover. A call to this function has the same effect as
+ failover. The optional sixth parameter,
+ <parameter>inactive_timeout</parameter>, when set to a
+ non-zero value, specifies the amount of time in seconds the slot is
+ allowed to be inactive. A call to this function has the same effect as
the replication protocol command
<literal>CREATE_REPLICATION_SLOT ... LOGICAL</literal>.
</para></entry>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index 61378b3d4b..f8838b1a23 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2761,6 +2761,15 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
ID of role
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>inactive_timeout</structfield> <type>integer</type>
+ </para>
+ <para>
+ The amount of time in seconds the slot is allowed to be inactive.
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql
index fe2bb50f46..af27616657 100644
--- a/src/backend/catalog/system_functions.sql
+++ b/src/backend/catalog/system_functions.sql
@@ -469,6 +469,7 @@ AS 'pg_logical_emit_message_bytea';
CREATE OR REPLACE FUNCTION pg_create_physical_replication_slot(
IN slot_name name, IN immediately_reserve boolean DEFAULT false,
IN temporary boolean DEFAULT false,
+ IN inactive_timeout int DEFAULT 0,
OUT slot_name name, OUT lsn pg_lsn)
RETURNS RECORD
LANGUAGE INTERNAL
@@ -480,6 +481,7 @@ CREATE OR REPLACE FUNCTION pg_create_logical_replication_slot(
IN temporary boolean DEFAULT false,
IN twophase boolean DEFAULT false,
IN failover boolean DEFAULT false,
+ IN inactive_timeout int DEFAULT 0,
OUT slot_name name, OUT lsn pg_lsn)
RETURNS RECORD
LANGUAGE INTERNAL
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index e17f979c7f..6648b125d5 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1027,7 +1027,8 @@ CREATE VIEW pg_replication_slots AS
L.invalidation_reason,
L.failover,
L.synced,
- L.last_inactive_at
+ L.last_inactive_at,
+ L.inactive_timeout
FROM pg_get_replication_slots() AS L
LEFT JOIN pg_database D ON (L.datoid = D.oid);
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 30480960c5..c01876ceeb 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -131,6 +131,7 @@ typedef struct RemoteSlot
char *database;
bool two_phase;
bool failover;
+ int inactive_timeout;
XLogRecPtr restart_lsn;
XLogRecPtr confirmed_lsn;
TransactionId catalog_xmin;
@@ -167,7 +168,8 @@ update_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid)
remote_slot->two_phase == slot->data.two_phase &&
remote_slot->failover == slot->data.failover &&
remote_slot->confirmed_lsn == slot->data.confirmed_flush &&
- strcmp(remote_slot->plugin, NameStr(slot->data.plugin)) == 0)
+ strcmp(remote_slot->plugin, NameStr(slot->data.plugin)) == 0 &&
+ remote_slot->inactive_timeout == slot->data.inactive_timeout)
return false;
/* Avoid expensive operations while holding a spinlock. */
@@ -182,6 +184,7 @@ update_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid)
slot->data.confirmed_flush = remote_slot->confirmed_lsn;
slot->data.catalog_xmin = remote_slot->catalog_xmin;
slot->effective_catalog_xmin = remote_slot->catalog_xmin;
+ slot->data.inactive_timeout = remote_slot->inactive_timeout;
SpinLockRelease(&slot->mutex);
if (xmin_changed)
@@ -607,7 +610,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
ReplicationSlotCreate(remote_slot->name, true, RS_TEMPORARY,
remote_slot->two_phase,
remote_slot->failover,
- true);
+ true, 0);
/* For shorter lines. */
slot = MyReplicationSlot;
@@ -627,6 +630,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
SpinLockAcquire(&slot->mutex);
slot->effective_catalog_xmin = xmin_horizon;
slot->data.catalog_xmin = xmin_horizon;
+ slot->data.inactive_timeout = remote_slot->inactive_timeout;
SpinLockRelease(&slot->mutex);
ReplicationSlotsComputeRequiredXmin(true);
LWLockRelease(ProcArrayLock);
@@ -652,9 +656,9 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
static bool
synchronize_slots(WalReceiverConn *wrconn)
{
-#define SLOTSYNC_COLUMN_COUNT 9
+#define SLOTSYNC_COLUMN_COUNT 10
Oid slotRow[SLOTSYNC_COLUMN_COUNT] = {TEXTOID, TEXTOID, LSNOID,
- LSNOID, XIDOID, BOOLOID, BOOLOID, TEXTOID, TEXTOID};
+ LSNOID, XIDOID, BOOLOID, BOOLOID, TEXTOID, TEXTOID, INT4OID};
WalRcvExecResult *res;
TupleTableSlot *tupslot;
@@ -663,7 +667,7 @@ synchronize_slots(WalReceiverConn *wrconn)
bool started_tx = false;
const char *query = "SELECT slot_name, plugin, confirmed_flush_lsn,"
" restart_lsn, catalog_xmin, two_phase, failover,"
- " database, invalidation_reason"
+ " database, invalidation_reason, inactive_timeout"
" FROM pg_catalog.pg_replication_slots"
" WHERE failover and NOT temporary";
@@ -743,6 +747,9 @@ synchronize_slots(WalReceiverConn *wrconn)
remote_slot->invalidated = isnull ? RS_INVAL_NONE :
GetSlotInvalidationCause(TextDatumGetCString(d));
+ remote_slot->inactive_timeout = DatumGetInt32(slot_getattr(tupslot, ++col,
+ &isnull));
+
/* Sanity check */
Assert(col == SLOTSYNC_COLUMN_COUNT);
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 146f0fbf84..195771920f 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -129,7 +129,7 @@ StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1),
sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize
#define SLOT_MAGIC 0x1051CA1 /* format identifier */
-#define SLOT_VERSION 5 /* version for new files */
+#define SLOT_VERSION 6 /* version for new files */
/* Control array for replication slot management */
ReplicationSlotCtlData *ReplicationSlotCtl = NULL;
@@ -304,11 +304,14 @@ ReplicationSlotValidateName(const char *name, int elevel)
* failover: If enabled, allows the slot to be synced to standbys so
* that logical replication can be resumed after failover.
* synced: True if the slot is synchronized from the primary server.
+ * inactive_timeout: The amount of time in seconds the slot is allowed to be
+ * inactive.
*/
void
ReplicationSlotCreate(const char *name, bool db_specific,
ReplicationSlotPersistency persistency,
- bool two_phase, bool failover, bool synced)
+ bool two_phase, bool failover, bool synced,
+ int inactive_timeout)
{
ReplicationSlot *slot = NULL;
int i;
@@ -345,6 +348,18 @@ ReplicationSlotCreate(const char *name, bool db_specific,
errmsg("cannot enable failover for a temporary replication slot"));
}
+ if (inactive_timeout > 0)
+ {
+ /*
+ * Do not allow users to set inactive_timeout for temporary slots,
+ * because temporary slots will not be saved to the disk.
+ */
+ if (persistency == RS_TEMPORARY)
+ ereport(ERROR,
+ errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot set inactive_timeout for a temporary replication slot"));
+ }
+
/*
* If some other backend ran this code concurrently with us, we'd likely
* both allocate the same slot, and that would be bad. We'd also be at
@@ -398,6 +413,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
slot->data.two_phase_at = InvalidXLogRecPtr;
slot->data.failover = failover;
slot->data.synced = synced;
+ slot->data.inactive_timeout = inactive_timeout;
/* and then data only present in shared memory */
slot->just_dirtied = false;
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 75300f24b6..326682138b 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -38,14 +38,15 @@
*/
static void
create_physical_replication_slot(char *name, bool immediately_reserve,
- bool temporary, XLogRecPtr restart_lsn)
+ bool temporary, int inactive_timeout,
+ XLogRecPtr restart_lsn)
{
Assert(!MyReplicationSlot);
/* acquire replication slot, this will check for conflicting names */
ReplicationSlotCreate(name, false,
temporary ? RS_TEMPORARY : RS_PERSISTENT, false,
- false, false);
+ false, false, inactive_timeout);
if (immediately_reserve)
{
@@ -71,6 +72,7 @@ pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
Name name = PG_GETARG_NAME(0);
bool immediately_reserve = PG_GETARG_BOOL(1);
bool temporary = PG_GETARG_BOOL(2);
+ int inactive_timeout = PG_GETARG_INT32(3);
Datum values[2];
bool nulls[2];
TupleDesc tupdesc;
@@ -84,9 +86,15 @@ pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
CheckSlotRequirements();
+ if (inactive_timeout < 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+ errmsg("\"inactive_timeout\" must not be negative")));
+
create_physical_replication_slot(NameStr(*name),
immediately_reserve,
temporary,
+ inactive_timeout,
InvalidXLogRecPtr);
values[0] = NameGetDatum(&MyReplicationSlot->data.name);
@@ -120,7 +128,7 @@ pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
static void
create_logical_replication_slot(char *name, char *plugin,
bool temporary, bool two_phase,
- bool failover,
+ bool failover, int inactive_timeout,
XLogRecPtr restart_lsn,
bool find_startpoint)
{
@@ -138,7 +146,7 @@ create_logical_replication_slot(char *name, char *plugin,
*/
ReplicationSlotCreate(name, true,
temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase,
- failover, false);
+ failover, false, inactive_timeout);
/*
* Create logical decoding context to find start point or, if we don't
@@ -177,6 +185,7 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
bool temporary = PG_GETARG_BOOL(2);
bool two_phase = PG_GETARG_BOOL(3);
bool failover = PG_GETARG_BOOL(4);
+ int inactive_timeout = PG_GETARG_INT32(5);
Datum result;
TupleDesc tupdesc;
HeapTuple tuple;
@@ -190,11 +199,17 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
CheckLogicalDecodingRequirements();
+ if (inactive_timeout < 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+ errmsg("\"inactive_timeout\" must not be negative")));
+
create_logical_replication_slot(NameStr(*name),
NameStr(*plugin),
temporary,
two_phase,
failover,
+ inactive_timeout,
InvalidXLogRecPtr,
true);
@@ -239,7 +254,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_get_replication_slots(PG_FUNCTION_ARGS)
{
-#define PG_GET_REPLICATION_SLOTS_COLS 19
+#define PG_GET_REPLICATION_SLOTS_COLS 20
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
XLogRecPtr currlsn;
int slotno;
@@ -441,6 +456,8 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
else
nulls[i++] = true;
+ values[i++] = Int32GetDatum(slot_contents.data.inactive_timeout);
+
Assert(i == PG_GET_REPLICATION_SLOTS_COLS);
tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
@@ -720,6 +737,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
XLogRecPtr src_restart_lsn;
bool src_islogical;
bool temporary;
+ int inactive_timeout;
char *plugin;
Datum values[2];
bool nulls[2];
@@ -776,6 +794,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
src_restart_lsn = first_slot_contents.data.restart_lsn;
temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
+ inactive_timeout = first_slot_contents.data.inactive_timeout;
/* Check type of replication slot */
if (src_islogical != logical_slot)
@@ -823,6 +842,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
temporary,
false,
false,
+ inactive_timeout,
src_restart_lsn,
false);
}
@@ -830,6 +850,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
create_physical_replication_slot(NameStr(*dst_name),
true,
temporary,
+ inactive_timeout,
src_restart_lsn);
/*
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index bc40c454de..5315c08650 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1221,7 +1221,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
{
ReplicationSlotCreate(cmd->slotname, false,
cmd->temporary ? RS_TEMPORARY : RS_PERSISTENT,
- false, false, false);
+ false, false, false, 0);
if (reserve_wal)
{
@@ -1252,7 +1252,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
*/
ReplicationSlotCreate(cmd->slotname, true,
cmd->temporary ? RS_TEMPORARY : RS_EPHEMERAL,
- two_phase, failover, false);
+ two_phase, failover, false, 0);
/*
* Do options check early so that we can bail before calling the
diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c
index 95c22a7200..12626987f0 100644
--- a/src/bin/pg_upgrade/info.c
+++ b/src/bin/pg_upgrade/info.c
@@ -676,7 +676,8 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
* removed.
*/
res = executeQueryOrDie(conn, "SELECT slot_name, plugin, two_phase, failover, "
- "%s as caught_up, invalidation_reason IS NOT NULL as invalid "
+ "%s as caught_up, invalidation_reason IS NOT NULL as invalid, "
+ "inactive_timeout "
"FROM pg_catalog.pg_replication_slots "
"WHERE slot_type = 'logical' AND "
"database = current_database() AND "
@@ -696,6 +697,7 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
int i_failover;
int i_caught_up;
int i_invalid;
+ int i_inactive_timeout;
slotinfos = (LogicalSlotInfo *) pg_malloc(sizeof(LogicalSlotInfo) * num_slots);
@@ -705,6 +707,7 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
i_failover = PQfnumber(res, "failover");
i_caught_up = PQfnumber(res, "caught_up");
i_invalid = PQfnumber(res, "invalid");
+ i_inactive_timeout = PQfnumber(res, "inactive_timeout");
for (int slotnum = 0; slotnum < num_slots; slotnum++)
{
@@ -716,6 +719,7 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
curr->failover = (strcmp(PQgetvalue(res, slotnum, i_failover), "t") == 0);
curr->caught_up = (strcmp(PQgetvalue(res, slotnum, i_caught_up), "t") == 0);
curr->invalid = (strcmp(PQgetvalue(res, slotnum, i_invalid), "t") == 0);
+ curr->inactive_timeout = atooid(PQgetvalue(res, slotnum, i_inactive_timeout));
}
}
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index f6143b6bc4..2656056103 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -931,9 +931,10 @@ create_logical_replication_slots(void)
appendPQExpBuffer(query, ", ");
appendStringLiteralConn(query, slot_info->plugin, conn);
- appendPQExpBuffer(query, ", false, %s, %s);",
+ appendPQExpBuffer(query, ", false, %s, %s, %d);",
slot_info->two_phase ? "true" : "false",
- slot_info->failover ? "true" : "false");
+ slot_info->failover ? "true" : "false",
+ slot_info->inactive_timeout);
PQclear(executeQueryOrDie(conn, "%s", query->data));
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h
index 92bcb693fb..eb86d000b1 100644
--- a/src/bin/pg_upgrade/pg_upgrade.h
+++ b/src/bin/pg_upgrade/pg_upgrade.h
@@ -162,6 +162,8 @@ typedef struct
bool invalid; /* if true, the slot is unusable */
bool failover; /* is the slot designated to be synced to the
* physical standby? */
+ int inactive_timeout; /* The amount of time in seconds the slot
+ * is allowed to be inactive. */
} LogicalSlotInfo;
typedef struct
diff --git a/src/bin/pg_upgrade/t/003_logical_slots.pl b/src/bin/pg_upgrade/t/003_logical_slots.pl
index 83d71c3084..6e82d2cb7b 100644
--- a/src/bin/pg_upgrade/t/003_logical_slots.pl
+++ b/src/bin/pg_upgrade/t/003_logical_slots.pl
@@ -153,14 +153,17 @@ like(
# TEST: Successful upgrade
# Preparations for the subsequent test:
-# 1. Setup logical replication (first, cleanup slots from the previous tests)
+# 1. Setup logical replication (first, cleanup slots from the previous tests,
+# and then create slot for this test with inactive_timeout set).
my $old_connstr = $oldpub->connstr . ' dbname=postgres';
+my $inactive_timeout = 3600;
$oldpub->start;
$oldpub->safe_psql(
'postgres', qq[
SELECT * FROM pg_drop_replication_slot('test_slot1');
SELECT * FROM pg_drop_replication_slot('test_slot2');
+ SELECT pg_create_logical_replication_slot(slot_name := 'regress_sub', plugin := 'pgoutput', inactive_timeout := $inactive_timeout);
CREATE PUBLICATION regress_pub FOR ALL TABLES;
]);
@@ -172,7 +175,7 @@ $sub->start;
$sub->safe_psql(
'postgres', qq[
CREATE TABLE tbl (a int);
- CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (two_phase = 'true', failover = 'true')
+ CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (slot_name = 'regress_sub', create_slot = false, two_phase = 'true', failover = 'true')
]);
$sub->wait_for_subscription_sync($oldpub, 'regress_sub');
@@ -192,8 +195,8 @@ command_ok([@pg_upgrade_cmd], 'run of pg_upgrade of old cluster');
# Check that the slot 'regress_sub' has migrated to the new cluster
$newpub->start;
my $result = $newpub->safe_psql('postgres',
- "SELECT slot_name, two_phase, failover FROM pg_replication_slots");
-is($result, qq(regress_sub|t|t), 'check the slot exists on new cluster');
+ "SELECT slot_name, two_phase, failover, inactive_timeout = $inactive_timeout FROM pg_replication_slots");
+is($result, qq(regress_sub|t|t|t), 'check the slot exists on new cluster');
# Update the connection
my $new_connstr = $newpub->connstr . ' dbname=postgres';
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index bf13448ad4..50db6b68d0 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11105,10 +11105,10 @@
# replication slots
{ oid => '3779', descr => 'create a physical replication slot',
proname => 'pg_create_physical_replication_slot', provolatile => 'v',
- proparallel => 'u', prorettype => 'record', proargtypes => 'name bool bool',
- proallargtypes => '{name,bool,bool,name,pg_lsn}',
- proargmodes => '{i,i,i,o,o}',
- proargnames => '{slot_name,immediately_reserve,temporary,slot_name,lsn}',
+ proparallel => 'u', prorettype => 'record', proargtypes => 'name bool bool int4',
+ proallargtypes => '{name,bool,bool,int4,name,pg_lsn}',
+ proargmodes => '{i,i,i,i,o,o}',
+ proargnames => '{slot_name,immediately_reserve,temporary,inactive_timeout,slot_name,lsn}',
prosrc => 'pg_create_physical_replication_slot' },
{ oid => '4220',
descr => 'copy a physical replication slot, changing temporality',
@@ -11133,17 +11133,17 @@
proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', prorettype => 'record',
proargtypes => '',
- proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,text,bool,bool,timestamptz}',
- proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,invalidation_reason,failover,synced,last_inactive_at}',
+ proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,text,bool,bool,timestamptz,int4}',
+ proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,invalidation_reason,failover,synced,last_inactive_at,inactive_timeout}',
prosrc => 'pg_get_replication_slots' },
{ oid => '3786', descr => 'set up a logical replication slot',
proname => 'pg_create_logical_replication_slot', provolatile => 'v',
proparallel => 'u', prorettype => 'record',
- proargtypes => 'name name bool bool bool',
- proallargtypes => '{name,name,bool,bool,bool,name,pg_lsn}',
- proargmodes => '{i,i,i,i,i,o,o}',
- proargnames => '{slot_name,plugin,temporary,twophase,failover,slot_name,lsn}',
+ proargtypes => 'name name bool bool bool int4',
+ proallargtypes => '{name,name,bool,bool,bool,int4,name,pg_lsn}',
+ proargmodes => '{i,i,i,i,i,i,o,o}',
+ proargnames => '{slot_name,plugin,temporary,twophase,failover,inactive_timeout,slot_name,lsn}',
prosrc => 'pg_create_logical_replication_slot' },
{ oid => '4222',
descr => 'copy a logical replication slot, changing temporality and plugin',
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index b4bb7f5e99..ff62542b03 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -127,6 +127,9 @@ typedef struct ReplicationSlotPersistentData
* for logical slots on the primary server.
*/
bool failover;
+
+ /* The amount of time in seconds the slot is allowed to be inactive. */
+ int inactive_timeout;
} ReplicationSlotPersistentData;
/*
@@ -239,7 +242,7 @@ extern void ReplicationSlotsShmemInit(void);
extern void ReplicationSlotCreate(const char *name, bool db_specific,
ReplicationSlotPersistency persistency,
bool two_phase, bool failover,
- bool synced);
+ bool synced, int inactive_timeout);
extern void ReplicationSlotPersist(void);
extern void ReplicationSlotDrop(const char *name, bool nowait);
extern void ReplicationSlotDropAcquired(void);
diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl
index f47bfd78eb..3dd780beab 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -152,8 +152,9 @@ log_min_messages = 'debug2'
$primary->append_conf('postgresql.conf', "log_min_messages = 'debug2'");
$primary->reload;
+my $inactive_timeout = 3600;
$primary->psql('postgres',
- q{SELECT pg_create_logical_replication_slot('lsub2_slot', 'test_decoding', false, false, true);}
+ "SELECT pg_create_logical_replication_slot('lsub2_slot', 'test_decoding', false, false, true, $inactive_timeout);"
);
$primary->psql('postgres',
@@ -190,6 +191,16 @@ is( $standby1->safe_psql(
"t",
'logical slots have synced as true on standby');
+# Confirm that the synced slot on the standby has got inactive_timeout from the
+# primary.
+is( $standby1->safe_psql(
+ 'postgres',
+ "SELECT inactive_timeout = $inactive_timeout FROM pg_replication_slots
+ WHERE slot_name = 'lsub2_slot' AND synced AND NOT temporary;"
+ ),
+ "t",
+ 'synced logical slot has got inactive_timeout on standby');
+
##################################################
# Test that the synchronized slot will be dropped if the corresponding remote
# slot on the primary server has been dropped.
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index effee2879e..adf7b8947f 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1477,8 +1477,9 @@ pg_replication_slots| SELECT l.slot_name,
l.invalidation_reason,
l.failover,
l.synced,
- l.last_inactive_at
- FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, invalidation_reason, failover, synced, last_inactive_at)
+ l.last_inactive_at,
+ l.inactive_timeout
+ FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, invalidation_reason, failover, synced, last_inactive_at, inactive_timeout)
LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
pg_roles| SELECT pg_authid.rolname,
pg_authid.rolsuper,
--
2.34.1
[application/x-patch] v14-0004-Introduce-new-SQL-funtion-pg_alter_replication_s.patch (15.7K, ../CALj2ACWuNHJ3qBMmmy6i1YA9V8JDRSaciL2bBM+72bRcqiBE2g@mail.gmail.com/5-v14-0004-Introduce-new-SQL-funtion-pg_alter_replication_s.patch)
download | inline diff:
From a0163b1f67dad275ff84fd1c5ebe290a19ebeb07 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Fri, 22 Mar 2024 06:32:39 +0000
Subject: [PATCH v14 4/6] Introduce new SQL funtion pg_alter_replication_slot
This commit adds a new function pg_alter_replication_slot to alter
the given property of a replication slot. It is similar to
replication protocol command ALTER_REPLICATION_SLOT, except that
for now it allows only inactive_timeout property to be set. The
reason for disallowing failover property to be altered via this
function is to avoid inconsistency with the catalog
pg_subscription on the logical subscriber. Because, the subscriber
won't know the altered value of its replication slot on the
publisher.
---
contrib/test_decoding/expected/slot.out | 44 ++++++++++++++-
contrib/test_decoding/sql/slot.sql | 10 ++++
doc/src/sgml/func.sgml | 21 ++++++++
src/backend/replication/slot.c | 22 ++++----
src/backend/replication/slotfuncs.c | 66 ++++++++++++++++++++++-
src/bin/pg_upgrade/t/003_logical_slots.pl | 14 +++--
src/include/catalog/pg_proc.dat | 5 ++
src/include/replication/slot.h | 2 +
8 files changed, 167 insertions(+), 17 deletions(-)
diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out
index 6771520afb..5b8dbf6f52 100644
--- a/contrib/test_decoding/expected/slot.out
+++ b/contrib/test_decoding/expected/slot.out
@@ -496,13 +496,27 @@ SELECT 'copy' FROM pg_copy_physical_replication_slot(src_slot_name := 'it_phy_sl
copy
(1 row)
+-- Test alter physical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot4');
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_phy_slot4', inactive_timeout := 900);
+ ?column?
+----------
+ alter
+(1 row)
+
SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
slot_name | slot_type | inactive_timeout
--------------+-----------+------------------
it_phy_slot1 | physical | 300
it_phy_slot2 | physical | 0
it_phy_slot3 | physical | 300
-(3 rows)
+ it_phy_slot4 | physical | 900
+(4 rows)
SELECT pg_drop_replication_slot('it_phy_slot1');
pg_drop_replication_slot
@@ -522,6 +536,12 @@ SELECT pg_drop_replication_slot('it_phy_slot3');
(1 row)
+SELECT pg_drop_replication_slot('it_phy_slot4');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
-- Test inactive_timeout option of logical slots.
SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot1', plugin := 'test_decoding', inactive_timeout := 600);
?column?
@@ -542,13 +562,27 @@ SELECT 'copy' FROM pg_copy_logical_replication_slot(src_slot_name := 'it_log_slo
copy
(1 row)
+-- Test alter logical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot4', plugin := 'test_decoding');
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_log_slot4', inactive_timeout := 900);
+ ?column?
+----------
+ alter
+(1 row)
+
SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
slot_name | slot_type | inactive_timeout
--------------+-----------+------------------
it_log_slot1 | logical | 600
it_log_slot2 | logical | 0
it_log_slot3 | logical | 600
-(3 rows)
+ it_log_slot4 | logical | 900
+(4 rows)
SELECT pg_drop_replication_slot('it_log_slot1');
pg_drop_replication_slot
@@ -568,3 +602,9 @@ SELECT pg_drop_replication_slot('it_log_slot3');
(1 row)
+SELECT pg_drop_replication_slot('it_log_slot4');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
diff --git a/contrib/test_decoding/sql/slot.sql b/contrib/test_decoding/sql/slot.sql
index 443e91da07..6785714cc7 100644
--- a/contrib/test_decoding/sql/slot.sql
+++ b/contrib/test_decoding/sql/slot.sql
@@ -206,11 +206,16 @@ SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot
-- Copy physical slot with inactive_timeout option set.
SELECT 'copy' FROM pg_copy_physical_replication_slot(src_slot_name := 'it_phy_slot1', dst_slot_name := 'it_phy_slot3');
+-- Test alter physical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_physical_replication_slot(slot_name := 'it_phy_slot4');
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_phy_slot4', inactive_timeout := 900);
+
SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
SELECT pg_drop_replication_slot('it_phy_slot1');
SELECT pg_drop_replication_slot('it_phy_slot2');
SELECT pg_drop_replication_slot('it_phy_slot3');
+SELECT pg_drop_replication_slot('it_phy_slot4');
-- Test inactive_timeout option of logical slots.
SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot1', plugin := 'test_decoding', inactive_timeout := 600);
@@ -219,8 +224,13 @@ SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot2
-- Copy logical slot with inactive_timeout option set.
SELECT 'copy' FROM pg_copy_logical_replication_slot(src_slot_name := 'it_log_slot1', dst_slot_name := 'it_log_slot3');
+-- Test alter logical slot with inactive_timeout option set.
+SELECT 'init' FROM pg_create_logical_replication_slot(slot_name := 'it_log_slot4', plugin := 'test_decoding');
+SELECT 'alter' FROM pg_alter_replication_slot(slot_name := 'it_log_slot4', inactive_timeout := 900);
+
SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;
SELECT pg_drop_replication_slot('it_log_slot1');
SELECT pg_drop_replication_slot('it_log_slot2');
SELECT pg_drop_replication_slot('it_log_slot3');
+SELECT pg_drop_replication_slot('it_log_slot4');
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index afaafa35ad..22c8e0d39c 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -28829,6 +28829,27 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
</entry>
</row>
+ <row>
+ <entry role="func_table_entry"><para role="func_signature">
+ <indexterm>
+ <primary>pg_alter_replication_slot</primary>
+ </indexterm>
+ <function>pg_alter_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type>, <parameter>inactive_timeout</parameter> <type>integer</type> )
+ <returnvalue>void</returnvalue>
+ </para>
+ <para>
+ Alters the given property of a replication slot
+ named <parameter>slot_name</parameter>. Same as replication protocol
+ command <literal>ALTER_REPLICATION_SLOT</literal>, except that it
+ allows only <parameter>inactive_timeout</parameter> property to be set.
+ The reason for disallowing <parameter>failover</parameter> property to
+ be altered via this function is to avoid inconsistency with the catalog
+ <structname>pg_subscription</structname> on the logical subscriber.
+ Because, the subscriber won't know the altered value of its
+ replication slot on the publisher.
+ </para></entry>
+ </row>
+
</tbody>
</tgroup>
</table>
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 195771920f..5644765a7e 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -162,7 +162,6 @@ static void ReplicationSlotDropPtr(ReplicationSlot *slot);
/* internal persistency functions */
static void RestoreSlotFromDisk(const char *name);
static void CreateSlotOnDisk(ReplicationSlot *slot);
-static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel);
/*
* Report shared-memory space needed by ReplicationSlotsShmemInit.
@@ -865,6 +864,7 @@ ReplicationSlotAlter(const char *name, bool failover)
ReplicationSlotRelease();
}
+
/*
* Permanently drop the currently acquired replication slot.
*/
@@ -1000,7 +1000,7 @@ ReplicationSlotSave(void)
Assert(MyReplicationSlot != NULL);
sprintf(path, "pg_replslot/%s", NameStr(MyReplicationSlot->data.name));
- SaveSlotToPath(MyReplicationSlot, path, ERROR);
+ ReplicationSlotSaveToPath(MyReplicationSlot, path, ERROR);
}
/*
@@ -1863,7 +1863,10 @@ CheckPointReplicationSlots(bool is_shutdown)
if (!s->in_use)
continue;
- /* save the slot to disk, locking is handled in SaveSlotToPath() */
+ /*
+ * Save the slot to disk, locking is handled in
+ * ReplicationSlotSaveToPath.
+ */
sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
/*
@@ -1889,7 +1892,7 @@ CheckPointReplicationSlots(bool is_shutdown)
SpinLockRelease(&s->mutex);
}
- SaveSlotToPath(s, path, LOG);
+ ReplicationSlotSaveToPath(s, path, LOG);
}
LWLockRelease(ReplicationSlotAllocationLock);
}
@@ -1968,8 +1971,9 @@ CreateSlotOnDisk(ReplicationSlot *slot)
/*
* No need to take out the io_in_progress_lock, nobody else can see this
- * slot yet, so nobody else will write. We're reusing SaveSlotToPath which
- * takes out the lock, if we'd take the lock here, we'd deadlock.
+ * slot yet, so nobody else will write. We're reusing
+ * ReplicationSlotSaveToPath which takes out the lock, if we'd take the
+ * lock here, we'd deadlock.
*/
sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
@@ -1995,7 +1999,7 @@ CreateSlotOnDisk(ReplicationSlot *slot)
/* Write the actual state file. */
slot->dirty = true; /* signal that we really need to write */
- SaveSlotToPath(slot, tmppath, ERROR);
+ ReplicationSlotSaveToPath(slot, tmppath, ERROR);
/* Rename the directory into place. */
if (rename(tmppath, path) != 0)
@@ -2020,8 +2024,8 @@ CreateSlotOnDisk(ReplicationSlot *slot)
/*
* Shared functionality between saving and creating a replication slot.
*/
-static void
-SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
+void
+ReplicationSlotSaveToPath(ReplicationSlot *slot, const char *dir, int elevel)
{
char tmppath[MAXPGPATH];
char path[MAXPGPATH];
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 326682138b..d6ef14fba6 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -229,7 +229,6 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
PG_RETURN_DATUM(result);
}
-
/*
* SQL function for dropping a replication slot.
*/
@@ -1038,3 +1037,68 @@ pg_sync_replication_slots(PG_FUNCTION_ARGS)
PG_RETURN_VOID();
}
+
+/*
+ * SQL function for altering given properties of a replication slot.
+ */
+Datum
+pg_alter_replication_slot(PG_FUNCTION_ARGS)
+{
+ Name name = PG_GETARG_NAME(0);
+ int inactive_timeout = PG_GETARG_INT32(1);
+ ReplicationSlot *slot;
+ char path[MAXPGPATH];
+
+ CheckSlotPermissions();
+
+ CheckSlotRequirements();
+
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+
+ /* Check if the slot exits with the given name. */
+ slot = SearchNamedReplicationSlot(NameStr(*name), false);
+
+ if (!slot)
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("replication slot \"%s\" does not exist",
+ NameStr(*name))));
+
+ /*
+ * Do not allow users to set inactive_timeout for temporary slots because
+ * temporary, slots will not be saved to the disk.
+ */
+ if (slot->data.persistency == RS_TEMPORARY)
+ ereport(ERROR,
+ errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot set inactive_timeout for a temporary replication slot"));
+
+ LWLockRelease(ReplicationSlotControlLock);
+
+ if (inactive_timeout < 0)
+ ereport(ERROR,
+ (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
+ errmsg("\"inactive_timeout\" must not be negative")));
+
+ /*
+ * We need to briefly prevent any other backend from acquiring the slot
+ * while we set the property. Without holding the ControlLock exclusively,
+ * a concurrent ReplicationSlotAcquire() could acquire the slot as well.
+ */
+ LWLockAcquire(ReplicationSlotControlLock, LW_EXCLUSIVE);
+
+ SpinLockAcquire(&slot->mutex);
+ slot->data.inactive_timeout = inactive_timeout;
+
+ /* Make sure the invalidated state persists across server restart */
+ slot->just_dirtied = true;
+ slot->dirty = true;
+ SpinLockRelease(&slot->mutex);
+
+ sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
+ ReplicationSlotSaveToPath(slot, path, ERROR);
+
+ LWLockRelease(ReplicationSlotControlLock);
+
+ PG_RETURN_VOID();
+}
diff --git a/src/bin/pg_upgrade/t/003_logical_slots.pl b/src/bin/pg_upgrade/t/003_logical_slots.pl
index 6e82d2cb7b..b79db24f42 100644
--- a/src/bin/pg_upgrade/t/003_logical_slots.pl
+++ b/src/bin/pg_upgrade/t/003_logical_slots.pl
@@ -153,17 +153,14 @@ like(
# TEST: Successful upgrade
# Preparations for the subsequent test:
-# 1. Setup logical replication (first, cleanup slots from the previous tests,
-# and then create slot for this test with inactive_timeout set).
+# 1. Setup logical replication (first, cleanup slots from the previous tests)
my $old_connstr = $oldpub->connstr . ' dbname=postgres';
-my $inactive_timeout = 3600;
$oldpub->start;
$oldpub->safe_psql(
'postgres', qq[
SELECT * FROM pg_drop_replication_slot('test_slot1');
SELECT * FROM pg_drop_replication_slot('test_slot2');
- SELECT pg_create_logical_replication_slot(slot_name := 'regress_sub', plugin := 'pgoutput', inactive_timeout := $inactive_timeout);
CREATE PUBLICATION regress_pub FOR ALL TABLES;
]);
@@ -175,7 +172,7 @@ $sub->start;
$sub->safe_psql(
'postgres', qq[
CREATE TABLE tbl (a int);
- CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (slot_name = 'regress_sub', create_slot = false, two_phase = 'true', failover = 'true')
+ CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (two_phase = 'true', failover = 'true')
]);
$sub->wait_for_subscription_sync($oldpub, 'regress_sub');
@@ -185,6 +182,13 @@ my $twophase_query =
$sub->poll_query_until('postgres', $twophase_query)
or die "Timed out while waiting for subscriber to enable twophase";
+# Alter slot to set inactive_timeout
+my $inactive_timeout = 3600;
+$oldpub->safe_psql(
+ 'postgres', qq[
+ SELECT pg_alter_replication_slot(slot_name := 'regress_sub', inactive_timeout := $inactive_timeout);
+]);
+
# 2. Temporarily disable the subscription
$sub->safe_psql('postgres', "ALTER SUBSCRIPTION regress_sub DISABLE");
$oldpub->stop;
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 50db6b68d0..b83e2f39b1 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11222,6 +11222,11 @@
proname => 'pg_sync_replication_slots', provolatile => 'v', proparallel => 'u',
prorettype => 'void', proargtypes => '',
prosrc => 'pg_sync_replication_slots' },
+{ oid => '9039', descr => 'alter given properties of a replication slot',
+ proname => 'pg_alter_replication_slot', provolatile => 'v', proparallel => 'u',
+ prorettype => 'void', proargtypes => 'name int4',
+ proargnames => '{slot_name,inactive_timeout}',
+ prosrc => 'pg_alter_replication_slot' },
# event triggers
{ oid => '3566', descr => 'list objects dropped by the current command',
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index ff62542b03..a8d7d42a07 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -252,6 +252,8 @@ extern void ReplicationSlotAcquire(const char *name, bool nowait);
extern void ReplicationSlotRelease(void);
extern void ReplicationSlotCleanup(void);
extern void ReplicationSlotSave(void);
+extern void ReplicationSlotSaveToPath(ReplicationSlot *slot, const char *dir,
+ int elevel);
extern void ReplicationSlotMarkDirty(void);
/* misc stuff */
--
2.34.1
[application/x-patch] v14-0005-Allow-setting-inactive_timeout-in-the-replicatio.patch (17.9K, ../CALj2ACWuNHJ3qBMmmy6i1YA9V8JDRSaciL2bBM+72bRcqiBE2g@mail.gmail.com/6-v14-0005-Allow-setting-inactive_timeout-in-the-replicatio.patch)
download | inline diff:
From bcc949abc9521eac3c16353fdf2783ced57eebae Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Fri, 22 Mar 2024 06:55:45 +0000
Subject: [PATCH v14 5/6] Allow setting inactive_timeout in the replication
commands
---
doc/src/sgml/protocol.sgml | 20 ++++++
src/backend/commands/subscriptioncmds.c | 6 +-
.../libpqwalreceiver/libpqwalreceiver.c | 61 ++++++++++++++++---
src/backend/replication/logical/tablesync.c | 1 +
src/backend/replication/slot.c | 30 ++++++++-
src/backend/replication/walreceiver.c | 2 +-
src/backend/replication/walsender.c | 38 +++++++++---
src/include/replication/slot.h | 3 +-
src/include/replication/walreceiver.h | 11 ++--
src/test/recovery/t/001_stream_rep.pl | 50 +++++++++++++++
10 files changed, 195 insertions(+), 27 deletions(-)
diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index a5cb19357f..2ffa1b470a 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2068,6 +2068,16 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>INACTIVE_TIMEOUT [ <replaceable class="parameter">integer</replaceable> ]</literal></term>
+ <listitem>
+ <para>
+ If set to a non-zero value, specifies the amount of time in seconds
+ the slot is allowed to be inactive. The default is zero.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist>
<para>
@@ -2168,6 +2178,16 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>INACTIVE_TIMEOUT [ <replaceable class="parameter">integer</replaceable> ]</literal></term>
+ <listitem>
+ <para>
+ If set to a non-zero value, specifies the amount of time in seconds
+ the slot is allowed to be inactive. The default is zero.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist>
</listitem>
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 5a47fa984d..4562de49c4 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -827,7 +827,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
twophase_enabled = true;
walrcv_create_slot(wrconn, opts.slot_name, false, twophase_enabled,
- opts.failover, CRS_NOEXPORT_SNAPSHOT, NULL);
+ opts.failover, 0, CRS_NOEXPORT_SNAPSHOT, NULL);
if (twophase_enabled)
UpdateTwoPhaseState(subid, LOGICALREP_TWOPHASE_STATE_ENABLED);
@@ -849,7 +849,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
else if (opts.slot_name &&
(opts.failover || walrcv_server_version(wrconn) >= 170000))
{
- walrcv_alter_slot(wrconn, opts.slot_name, opts.failover);
+ walrcv_alter_slot(wrconn, opts.slot_name, &opts.failover, NULL);
}
}
PG_FINALLY();
@@ -1541,7 +1541,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PG_TRY();
{
- walrcv_alter_slot(wrconn, sub->slotname, opts.failover);
+ walrcv_alter_slot(wrconn, sub->slotname, &opts.failover, NULL);
}
PG_FINALLY();
{
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 761bf0f677..126250a076 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -77,10 +77,11 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
bool temporary,
bool two_phase,
bool failover,
+ int inactive_timeout,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover);
+ bool *failover, int *inactive_timeout);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -1008,7 +1009,8 @@ libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
*/
static char *
libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
- bool temporary, bool two_phase, bool failover,
+ bool temporary, bool two_phase,
+ bool failover, int inactive_timeout,
CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
{
PGresult *res;
@@ -1048,6 +1050,15 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
appendStringInfoChar(&cmd, ' ');
}
+ if (inactive_timeout > 0)
+ {
+ appendStringInfo(&cmd, "INACTIVE_TIMEOUT %d", inactive_timeout);
+ if (use_new_options_syntax)
+ appendStringInfoString(&cmd, ", ");
+ else
+ appendStringInfoChar(&cmd, ' ');
+ }
+
if (use_new_options_syntax)
{
switch (snapshot_action)
@@ -1084,10 +1095,24 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
}
else
{
+ appendStringInfoString(&cmd, " PHYSICAL ");
if (use_new_options_syntax)
- appendStringInfoString(&cmd, " PHYSICAL (RESERVE_WAL)");
- else
- appendStringInfoString(&cmd, " PHYSICAL RESERVE_WAL");
+ appendStringInfoChar(&cmd, '(');
+
+ appendStringInfoString(&cmd, "RESERVE_WAL");
+
+ if (inactive_timeout > 0)
+ {
+ if (use_new_options_syntax)
+ appendStringInfoString(&cmd, ", ");
+ else
+ appendStringInfoChar(&cmd, ' ');
+
+ appendStringInfo(&cmd, "INACTIVE_TIMEOUT %d", inactive_timeout);
+ }
+
+ if (use_new_options_syntax)
+ appendStringInfoChar(&cmd, ')');
}
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
@@ -1121,15 +1146,33 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover)
+ bool *failover, int *inactive_timeout)
{
StringInfoData cmd;
PGresult *res;
+ bool specified_prev_opt = false;
initStringInfo(&cmd);
- appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s )",
- quote_identifier(slotname),
- failover ? "true" : "false");
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s (",
+ quote_identifier(slotname));
+
+ if (failover != NULL)
+ {
+ appendStringInfo(&cmd, "FAILOVER %s",
+ *failover ? "true" : "false");
+ specified_prev_opt = true;
+ }
+
+ if (inactive_timeout != NULL)
+ {
+ if (specified_prev_opt)
+ appendStringInfoString(&cmd, ", ");
+
+ appendStringInfo(&cmd, "INACTIVE_TIMEOUT %d", *inactive_timeout);
+ specified_prev_opt = true;
+ }
+
+ appendStringInfoChar(&cmd, ')');
res = libpqrcv_PQexec(conn->streamConn, cmd.data);
pfree(cmd.data);
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 1061d5b61b..59f8e5fbaa 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -1431,6 +1431,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
walrcv_create_slot(LogRepWorkerWalRcvConn,
slotname, false /* permanent */ , false /* two_phase */ ,
MySubscription->failover,
+ 0,
CRS_USE_SNAPSHOT, origin_startpos);
/*
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 5644765a7e..3680a608c3 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -807,8 +807,10 @@ ReplicationSlotDrop(const char *name, bool nowait)
* Change the definition of the slot identified by the specified name.
*/
void
-ReplicationSlotAlter(const char *name, bool failover)
+ReplicationSlotAlter(const char *name, bool failover, int inactive_timeout)
{
+ bool lock_acquired;
+
Assert(MyReplicationSlot == NULL);
ReplicationSlotAcquire(name, false);
@@ -851,10 +853,36 @@ ReplicationSlotAlter(const char *name, bool failover)
errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot enable failover for a temporary replication slot"));
+ /*
+ * Do not allow users to set inactive_timeout for temporary slots because
+ * temporary, slots will not be saved to the disk.
+ */
+ if (inactive_timeout > 0 && MyReplicationSlot->data.persistency == RS_TEMPORARY)
+ ereport(ERROR,
+ errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot set inactive_timeout for a temporary replication slot"));
+
+ lock_acquired = false;
if (MyReplicationSlot->data.failover != failover)
{
SpinLockAcquire(&MyReplicationSlot->mutex);
+ lock_acquired = true;
MyReplicationSlot->data.failover = failover;
+ }
+
+ if (MyReplicationSlot->data.inactive_timeout != inactive_timeout)
+ {
+ if (!lock_acquired)
+ {
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ lock_acquired = true;
+ }
+
+ MyReplicationSlot->data.inactive_timeout = inactive_timeout;
+ }
+
+ if (lock_acquired)
+ {
SpinLockRelease(&MyReplicationSlot->mutex);
ReplicationSlotMarkDirty();
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index acda5f68d9..ac2ebb0c69 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -389,7 +389,7 @@ WalReceiverMain(char *startup_data, size_t startup_data_len)
"pg_walreceiver_%lld",
(long long int) walrcv_get_backend_pid(wrconn));
- walrcv_create_slot(wrconn, slotname, true, false, false, 0, NULL);
+ walrcv_create_slot(wrconn, slotname, true, false, false, 0, 0, NULL);
SpinLockAcquire(&walrcv->mutex);
strlcpy(walrcv->slotname, slotname, NAMEDATALEN);
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 5315c08650..0420274247 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1123,13 +1123,15 @@ static void
parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
bool *reserve_wal,
CRSSnapshotAction *snapshot_action,
- bool *two_phase, bool *failover)
+ bool *two_phase, bool *failover,
+ int *inactive_timeout)
{
ListCell *lc;
bool snapshot_action_given = false;
bool reserve_wal_given = false;
bool two_phase_given = false;
bool failover_given = false;
+ bool inactive_timeout_given = false;
/* Parse options */
foreach(lc, cmd->options)
@@ -1188,6 +1190,15 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
failover_given = true;
*failover = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "inactive_timeout") == 0)
+ {
+ if (inactive_timeout_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ inactive_timeout_given = true;
+ *inactive_timeout = defGetInt32(defel);
+ }
else
elog(ERROR, "unrecognized option: %s", defel->defname);
}
@@ -1205,6 +1216,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
bool reserve_wal = false;
bool two_phase = false;
bool failover = false;
+ int inactive_timeout = 0;
CRSSnapshotAction snapshot_action = CRS_EXPORT_SNAPSHOT;
DestReceiver *dest;
TupOutputState *tstate;
@@ -1215,13 +1227,13 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
Assert(!MyReplicationSlot);
parseCreateReplSlotOptions(cmd, &reserve_wal, &snapshot_action, &two_phase,
- &failover);
+ &failover, &inactive_timeout);
if (cmd->kind == REPLICATION_KIND_PHYSICAL)
{
ReplicationSlotCreate(cmd->slotname, false,
cmd->temporary ? RS_TEMPORARY : RS_PERSISTENT,
- false, false, false, 0);
+ false, false, false, inactive_timeout);
if (reserve_wal)
{
@@ -1252,7 +1264,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
*/
ReplicationSlotCreate(cmd->slotname, true,
cmd->temporary ? RS_TEMPORARY : RS_EPHEMERAL,
- two_phase, failover, false, 0);
+ two_phase, failover, false, inactive_timeout);
/*
* Do options check early so that we can bail before calling the
@@ -1411,9 +1423,11 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
* Process extra options given to ALTER_REPLICATION_SLOT.
*/
static void
-ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
+ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover,
+ int *inactive_timeout)
{
bool failover_given = false;
+ bool inactive_timeout_given = false;
/* Parse options */
foreach_ptr(DefElem, defel, cmd->options)
@@ -1427,6 +1441,15 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
failover_given = true;
*failover = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "inactive_timeout") == 0)
+ {
+ if (inactive_timeout_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ inactive_timeout_given = true;
+ *inactive_timeout = defGetInt32(defel);
+ }
else
elog(ERROR, "unrecognized option: %s", defel->defname);
}
@@ -1439,9 +1462,10 @@ static void
AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
{
bool failover = false;
+ int inactive_timeout = 0;
- ParseAlterReplSlotOptions(cmd, &failover);
- ReplicationSlotAlter(cmd->slotname, failover);
+ ParseAlterReplSlotOptions(cmd, &failover, &inactive_timeout);
+ ReplicationSlotAlter(cmd->slotname, failover, inactive_timeout);
}
/*
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index a8d7d42a07..9cd4bf98e5 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -246,7 +246,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
extern void ReplicationSlotPersist(void);
extern void ReplicationSlotDrop(const char *name, bool nowait);
extern void ReplicationSlotDropAcquired(void);
-extern void ReplicationSlotAlter(const char *name, bool failover);
+extern void ReplicationSlotAlter(const char *name, bool failover,
+ int inactive_timeout);
extern void ReplicationSlotAcquire(const char *name, bool nowait);
extern void ReplicationSlotRelease(void);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 12f71fa99b..038812fd24 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -366,6 +366,7 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
bool temporary,
bool two_phase,
bool failover,
+ int inactive_timeout,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
@@ -377,7 +378,7 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
*/
typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
const char *slotname,
- bool failover);
+ bool *failover, int *inactive_timeout);
/*
* walrcv_get_backend_pid_fn
@@ -453,10 +454,10 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
#define walrcv_send(conn, buffer, nbytes) \
WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
-#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \
- WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
-#define walrcv_alter_slot(conn, slotname, failover) \
- WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover)
+#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, inactive_timeout, snapshot_action, lsn) \
+ WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, inactive_timeout, snapshot_action, lsn)
+#define walrcv_alter_slot(conn, slotname, failover, inactive_timeout) \
+ WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover, inactive_timeout)
#define walrcv_get_backend_pid(conn) \
WalReceiverFunctions->walrcv_get_backend_pid(conn)
#define walrcv_exec(conn, exec, nRetTypes, retTypes) \
diff --git a/src/test/recovery/t/001_stream_rep.pl b/src/test/recovery/t/001_stream_rep.pl
index 5311ade509..db00b6aa24 100644
--- a/src/test/recovery/t/001_stream_rep.pl
+++ b/src/test/recovery/t/001_stream_rep.pl
@@ -604,4 +604,54 @@ ok( pump_until(
'base backup cleanly canceled');
$sigchld_bb->finish();
+# Drop any existing slots on the primary, for the follow-up tests.
+$node_primary->safe_psql('postgres',
+ "SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots;");
+
+# Test setting inactive_timeout option via replication commands.
+$node_primary->append_conf(
+ 'postgresql.conf', qq(
+wal_level = logical
+));
+$node_primary->restart;
+
+$node_primary->psql(
+ 'postgres',
+ "CREATE_REPLICATION_SLOT it_phy_slot1 PHYSICAL (RESERVE_WAL, INACTIVE_TIMEOUT 100);",
+ extra_params => [ '-d', $connstr_db ]);
+
+$node_primary->psql(
+ 'postgres',
+ "CREATE_REPLICATION_SLOT it_phy_slot2 PHYSICAL (RESERVE_WAL);",
+ extra_params => [ '-d', $connstr_db ]);
+
+$node_primary->psql(
+ 'postgres',
+ "ALTER_REPLICATION_SLOT it_phy_slot2 (INACTIVE_TIMEOUT 200);",
+ extra_params => [ '-d', $connstr_db ]);
+
+$node_primary->psql(
+ 'postgres',
+ "CREATE_REPLICATION_SLOT it_log_slot1 LOGICAL pgoutput (TWO_PHASE, INACTIVE_TIMEOUT 300);",
+ extra_params => [ '-d', $connstr_db ]);
+
+$node_primary->psql(
+ 'postgres',
+ "CREATE_REPLICATION_SLOT it_log_slot2 LOGICAL pgoutput;",
+ extra_params => [ '-d', $connstr_db ]);
+
+$node_primary->psql(
+ 'postgres',
+ "ALTER_REPLICATION_SLOT it_log_slot2 (INACTIVE_TIMEOUT 400);",
+ extra_params => [ '-d', $connstr_db ]);
+
+my $slot_info_expected = 'it_log_slot1|logical|300
+it_log_slot2|logical|400
+it_phy_slot1|physical|100
+it_phy_slot2|physical|0';
+
+my $slot_info = $node_primary->safe_psql('postgres',
+ qq[SELECT slot_name, slot_type, inactive_timeout FROM pg_replication_slots ORDER BY 1;]);
+is($slot_info, $slot_info_expected, "replication slots with inactive_timeout on primary exist");
+
done_testing();
--
2.34.1
[application/x-patch] v14-0006-Add-inactive_timeout-based-replication-slot-inva.patch (32.1K, ../CALj2ACWuNHJ3qBMmmy6i1YA9V8JDRSaciL2bBM+72bRcqiBE2g@mail.gmail.com/7-v14-0006-Add-inactive_timeout-based-replication-slot-inva.patch)
download | inline diff:
From 0a30fab6610a1197f58124d276369ac89f7cba99 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Fri, 22 Mar 2024 07:54:21 +0000
Subject: [PATCH v14 6/6] Add inactive_timeout based replication slot
invalidation
---
doc/src/sgml/func.sgml | 12 +-
doc/src/sgml/system-views.sgml | 10 +-
.../replication/logical/logicalfuncs.c | 4 +-
src/backend/replication/logical/slotsync.c | 8 +-
src/backend/replication/slot.c | 240 ++++++++++++++++--
src/backend/replication/slotfuncs.c | 27 +-
src/backend/replication/walsender.c | 12 +-
src/backend/tcop/postgres.c | 2 +-
src/backend/utils/adt/pg_upgrade_support.c | 4 +-
src/include/replication/slot.h | 11 +-
src/test/recovery/meson.build | 1 +
src/test/recovery/t/050_invalidate_slots.pl | 170 +++++++++++++
12 files changed, 455 insertions(+), 46 deletions(-)
create mode 100644 src/test/recovery/t/050_invalidate_slots.pl
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index 22c8e0d39c..4826e45c7d 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -28393,8 +28393,8 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
released upon any error. The optional fourth
parameter, <parameter>inactive_timeout</parameter>, when set to a
non-zero value, specifies the amount of time in seconds the slot is
- allowed to be inactive. This function corresponds to the replication
- protocol command
+ allowed to be inactive before getting invalidated.
+ This function corresponds to the replication protocol command
<literal>CREATE_REPLICATION_SLOT ... PHYSICAL</literal>.
</para></entry>
</row>
@@ -28439,12 +28439,12 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
<parameter>failover</parameter>, when set to true,
specifies that this slot is enabled to be synced to the
standbys so that logical replication can be resumed after
- failover. The optional sixth parameter,
+ failover. The optional sixth parameter,
<parameter>inactive_timeout</parameter>, when set to a
non-zero value, specifies the amount of time in seconds the slot is
- allowed to be inactive. A call to this function has the same effect as
- the replication protocol command
- <literal>CREATE_REPLICATION_SLOT ... LOGICAL</literal>.
+ allowed to be inactive before getting invalidated.
+ A call to this function has the same effect as the replication protocol
+ command <literal>CREATE_REPLICATION_SLOT ... LOGICAL</literal>.
</para></entry>
</row>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index f8838b1a23..8e7d9c9105 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2563,6 +2563,13 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
perform logical decoding. It is set only for logical slots.
</para>
</listitem>
+ <listitem>
+ <para>
+ <literal>inactive_timeout</literal> means that the slot has been
+ inactive for the duration specified by slot's
+ <literal>inactive_timeout</literal> parameter.
+ </para>
+ </listitem>
</itemizedlist>
</para></entry>
</row>
@@ -2767,7 +2774,8 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
<structfield>inactive_timeout</structfield> <type>integer</type>
</para>
<para>
- The amount of time in seconds the slot is allowed to be inactive.
+ The amount of time in seconds the slot is allowed to be inactive before
+ getting invalidated.
</para></entry>
</row>
</tbody>
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index b4dd5cce75..53cf8bbd42 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();
{
@@ -309,7 +309,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
/* free context, call shutdown callback */
FreeDecodingContext(ctx);
- ReplicationSlotRelease();
+ ReplicationSlotRelease(true);
InvalidateSystemCaches();
}
PG_CATCH();
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index c01876ceeb..5aba117e2b 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -319,7 +319,7 @@ drop_local_obsolete_slots(List *remote_slot_list)
if (synced_slot)
{
- ReplicationSlotAcquire(NameStr(local_slot->data.name), true);
+ ReplicationSlotAcquire(NameStr(local_slot->data.name), true, false);
ReplicationSlotDropAcquired();
}
@@ -529,7 +529,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
* InvalidatePossiblyObsoleteSlot() where it invalidates slot directly
* if the slot is not acquired by other processes.
*/
- ReplicationSlotAcquire(remote_slot->name, true);
+ ReplicationSlotAcquire(remote_slot->name, true, false);
Assert(slot == MyReplicationSlot);
@@ -554,7 +554,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
/* Skip the sync of an invalidated slot */
if (slot->data.invalidated != RS_INVAL_NONE)
{
- ReplicationSlotRelease();
+ ReplicationSlotRelease(false);
return slot_updated;
}
@@ -640,7 +640,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
slot_updated = true;
}
- ReplicationSlotRelease();
+ ReplicationSlotRelease(false);
return slot_updated;
}
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 3680a608c3..0acf1d1960 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -107,10 +107,11 @@ const char *const SlotInvalidationCauses[] = {
[RS_INVAL_WAL_REMOVED] = "wal_removed",
[RS_INVAL_HORIZON] = "rows_removed",
[RS_INVAL_WAL_LEVEL] = "wal_level_insufficient",
+ [RS_INVAL_INACTIVE_TIMEOUT] = "inactive_timeout",
};
/* Maximum number of invalidation causes */
-#define RS_INVAL_MAX_CAUSES RS_INVAL_WAL_LEVEL
+#define RS_INVAL_MAX_CAUSES RS_INVAL_INACTIVE_TIMEOUT
StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1),
"array length mismatch");
@@ -158,6 +159,9 @@ static XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr;
static void ReplicationSlotShmemExit(int code, Datum arg);
static void ReplicationSlotDropPtr(ReplicationSlot *slot);
+static bool InvalidateSlotForInactiveTimeout(ReplicationSlot *slot,
+ bool need_control_lock,
+ bool need_mutex);
/* internal persistency functions */
static void RestoreSlotFromDisk(const char *name);
@@ -233,7 +237,7 @@ ReplicationSlotShmemExit(int code, Datum arg)
{
/* Make sure active replication slots are released */
if (MyReplicationSlot != NULL)
- ReplicationSlotRelease();
+ ReplicationSlotRelease(true);
/* Also cleanup all the temporary slots. */
ReplicationSlotCleanup();
@@ -424,7 +428,19 @@ ReplicationSlotCreate(const char *name, bool db_specific,
slot->candidate_restart_valid = InvalidXLogRecPtr;
slot->candidate_restart_lsn = InvalidXLogRecPtr;
slot->last_saved_confirmed_flush = InvalidXLogRecPtr;
- slot->last_inactive_at = 0;
+
+ /*
+ * We set last_inactive_at after creation of the slot so that the
+ * inactive_timeout if set is honored.
+ *
+ * There's no point in allowing failover slots to get invalidated based on
+ * slot's inactive_timeout parameter on standby. The failover slots simply
+ * get synced from the primary on the standby.
+ */
+ if (!(RecoveryInProgress() && slot->data.failover))
+ slot->last_inactive_at = GetCurrentTimestamp();
+ else
+ slot->last_inactive_at = 0;
/*
* Create the slot on disk. We haven't actually marked the slot allocated
@@ -550,9 +566,14 @@ ReplicationSlotName(int index, Name name)
*
* An error is raised if nowait is true and the slot is currently in use. If
* nowait is false, we sleep until the slot is released by the owning process.
+ *
+ * If check_for_invalidation is true, the slot is checked for invalidation
+ * based on its inactive_timeout parameter and an error is raised after making
+ * the slot ours.
*/
void
-ReplicationSlotAcquire(const char *name, bool nowait)
+ReplicationSlotAcquire(const char *name, bool nowait,
+ bool check_for_invalidation)
{
ReplicationSlot *s;
int active_pid;
@@ -630,6 +651,42 @@ retry:
/* We made this slot active, so it's ours now. */
MyReplicationSlot = s;
+ /*
+ * Check if the given slot can be invalidated based on its
+ * inactive_timeout parameter. If yes, persist the invalidated state to
+ * disk and then error out. We do this only after making the slot ours to
+ * avoid anyone else acquiring it while we check for its invalidation.
+ */
+ if (check_for_invalidation)
+ {
+ /* The slot is ours by now */
+ Assert(s->active_pid == MyProcPid);
+
+ /*
+ * Well, the slot is not yet ours really unless we check for the
+ * invalidation below.
+ */
+ s->active_pid = 0;
+ if (InvalidateReplicationSlotForInactiveTimeout(s, true, true, true))
+ {
+ /*
+ * If the slot has been invalidated, recalculate the resource
+ * limits.
+ */
+ ReplicationSlotsComputeRequiredXmin(false);
+ ReplicationSlotsComputeRequiredLSN();
+
+ /* Might need it for slot clean up on error, so restore it */
+ s->active_pid = MyProcPid;
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot acquire invalidated replication slot \"%s\"",
+ NameStr(MyReplicationSlot->data.name)),
+ errdetail("This slot has been invalidated because of its inactive_timeout parameter.")));
+ }
+ s->active_pid = MyProcPid;
+ }
+
/*
* The call to pgstat_acquire_replslot() protects against stats for a
* different slot, from before a restart or such, being present during
@@ -663,7 +720,7 @@ retry:
* Resources this slot requires will be preserved.
*/
void
-ReplicationSlotRelease(void)
+ReplicationSlotRelease(bool set_last_inactive_at)
{
ReplicationSlot *slot = MyReplicationSlot;
char *slotname = NULL; /* keep compiler quiet */
@@ -714,11 +771,20 @@ ReplicationSlotRelease(void)
ConditionVariableBroadcast(&slot->active_cv);
}
- if (slot->data.persistency == RS_PERSISTENT)
+ if (set_last_inactive_at &&
+ slot->data.persistency == RS_PERSISTENT)
{
- SpinLockAcquire(&slot->mutex);
- slot->last_inactive_at = GetCurrentTimestamp();
- SpinLockRelease(&slot->mutex);
+ /*
+ * There's no point in allowing failover slots to get invalidated
+ * based on slot's inactive_timeout parameter on standby. The failover
+ * slots simply get synced from the primary on the standby.
+ */
+ if (!(RecoveryInProgress() && slot->data.failover))
+ {
+ SpinLockAcquire(&slot->mutex);
+ slot->last_inactive_at = GetCurrentTimestamp();
+ SpinLockRelease(&slot->mutex);
+ }
}
MyReplicationSlot = NULL;
@@ -788,7 +854,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
@@ -813,7 +879,7 @@ ReplicationSlotAlter(const char *name, bool failover, int inactive_timeout)
Assert(MyReplicationSlot == NULL);
- ReplicationSlotAcquire(name, false);
+ ReplicationSlotAcquire(name, false, true);
if (SlotIsPhysical(MyReplicationSlot))
ereport(ERROR,
@@ -889,7 +955,7 @@ ReplicationSlotAlter(const char *name, bool failover, int inactive_timeout)
ReplicationSlotSave();
}
- ReplicationSlotRelease();
+ ReplicationSlotRelease(true);
}
@@ -1542,6 +1608,9 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
case RS_INVAL_WAL_LEVEL:
appendStringInfoString(&err_detail, _("Logical decoding on standby requires wal_level >= logical on the primary server."));
break;
+ case RS_INVAL_INACTIVE_TIMEOUT:
+ appendStringInfoString(&err_detail, _("The slot has been inactive for more than the time specified by slot's inactive_timeout parameter."));
+ break;
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -1655,6 +1724,10 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
if (SlotIsLogical(s))
invalidation_cause = cause;
break;
+ case RS_INVAL_INACTIVE_TIMEOUT:
+ if (InvalidateReplicationSlotForInactiveTimeout(s, false, false, false))
+ invalidation_cause = cause;
+ break;
case RS_INVAL_NONE:
pg_unreachable();
}
@@ -1781,7 +1854,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
/* Make sure the invalidated state persists across server restart */
ReplicationSlotMarkDirty();
ReplicationSlotSave();
- ReplicationSlotRelease();
+ ReplicationSlotRelease(true);
ReportSlotInvalidation(invalidation_cause, false, active_pid,
slotname, restart_lsn,
@@ -1808,6 +1881,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
* - RS_INVAL_HORIZON: requires a snapshot <= the given horizon in the given
* db; dboid may be InvalidOid for shared relations
* - RS_INVAL_WAL_LEVEL: is logical
+ * - RS_INVAL_INACTIVE_TIMEOUT: inactive slot timeout occurs
*
* NB - this runs as part of checkpoint, so avoid raising errors if possible.
*/
@@ -1859,6 +1933,110 @@ restart:
return invalidated;
}
+/*
+ * Invalidate given slot based on its inactive_timeout parameter.
+ *
+ * Returns true if the slot has got invalidated.
+ *
+ * NB - this function also runs as part of checkpoint, so avoid raising errors
+ * if possible.
+ */
+bool
+InvalidateReplicationSlotForInactiveTimeout(ReplicationSlot *slot,
+ bool need_control_lock,
+ bool need_mutex,
+ bool persist_state)
+{
+ if (!InvalidateSlotForInactiveTimeout(slot, need_control_lock, need_mutex))
+ return false;
+
+ Assert(slot->active_pid == 0);
+
+ SpinLockAcquire(&slot->mutex);
+ slot->data.invalidated = RS_INVAL_INACTIVE_TIMEOUT;
+
+ /* Make sure the invalidated state persists across server restart */
+ slot->just_dirtied = true;
+ slot->dirty = true;
+ SpinLockRelease(&slot->mutex);
+
+ if (persist_state)
+ {
+ char path[MAXPGPATH];
+
+ sprintf(path, "pg_replslot/%s", NameStr(slot->data.name));
+ ReplicationSlotSaveToPath(slot, path, ERROR);
+ }
+
+ ReportSlotInvalidation(RS_INVAL_INACTIVE_TIMEOUT, false, 0,
+ slot->data.name, InvalidXLogRecPtr,
+ InvalidXLogRecPtr, InvalidTransactionId);
+
+ return true;
+}
+
+/*
+ * Helper for InvalidateReplicationSlotForInactiveTimeout
+ */
+static bool
+InvalidateSlotForInactiveTimeout(ReplicationSlot *slot,
+ bool need_control_lock,
+ bool need_mutex)
+{
+ ReplicationSlotInvalidationCause inavidation_cause = RS_INVAL_NONE;
+
+ if (slot->last_inactive_at == 0 ||
+ slot->data.inactive_timeout == 0)
+ return false;
+
+ /* inactive_timeout is only tracked for permanent slots */
+ if (slot->data.persistency != RS_PERSISTENT)
+ return false;
+
+ /*
+ * There's no point in allowing failover slots to get invalidated based on
+ * slot's inactive_timeout parameter on standby. The failover slots simply
+ * get synced from the primary on the standby.
+ */
+ if (RecoveryInProgress() && slot->data.failover)
+ return false;
+
+ if (need_control_lock)
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+
+ Assert(LWLockHeldByMeInMode(ReplicationSlotControlLock, LW_SHARED));
+
+ /*
+ * Check if the slot needs to be invalidated due to inactive_timeout. We
+ * do this with the spinlock held to avoid race conditions -- for example
+ * the restart_lsn could move forward, or the slot could be dropped.
+ */
+ if (need_mutex)
+ SpinLockAcquire(&slot->mutex);
+
+ if (slot->last_inactive_at > 0 &&
+ slot->data.inactive_timeout > 0)
+ {
+ TimestampTz now;
+
+ /* last_inactive_at is only tracked for inactive slots */
+ Assert(slot->active_pid == 0);
+
+ now = GetCurrentTimestamp();
+ if (TimestampDifferenceExceeds(slot->last_inactive_at, now,
+ slot->data.inactive_timeout * 1000))
+ inavidation_cause = RS_INVAL_INACTIVE_TIMEOUT;
+ }
+
+ if (need_mutex)
+ SpinLockRelease(&slot->mutex);
+
+ if (need_control_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+
+ return (inavidation_cause == RS_INVAL_INACTIVE_TIMEOUT);
+}
+
/*
* Flush all replication slots to disk.
*
@@ -1871,6 +2049,7 @@ void
CheckPointReplicationSlots(bool is_shutdown)
{
int i;
+ bool invalidated = false;
elog(DEBUG1, "performing replication slot checkpoint");
@@ -1892,10 +2071,11 @@ CheckPointReplicationSlots(bool is_shutdown)
continue;
/*
- * Save the slot to disk, locking is handled in
- * ReplicationSlotSaveToPath.
+ * Here's an opportunity to invalidate inactive replication slots
+ * based on timeout, so let's do it.
*/
- sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
+ if (InvalidateReplicationSlotForInactiveTimeout(s, true, true, false))
+ invalidated = true;
/*
* Slot's data is not flushed each time the confirmed_flush LSN is
@@ -1920,9 +2100,21 @@ CheckPointReplicationSlots(bool is_shutdown)
SpinLockRelease(&s->mutex);
}
+ /*
+ * Save the slot to disk, locking is handled in
+ * ReplicationSlotSaveToPath.
+ */
+ sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
ReplicationSlotSaveToPath(s, path, LOG);
}
LWLockRelease(ReplicationSlotAllocationLock);
+
+ /* If the slot has been invalidated, recalculate the resource limits */
+ if (invalidated)
+ {
+ ReplicationSlotsComputeRequiredXmin(false);
+ ReplicationSlotsComputeRequiredLSN();
+ }
}
/*
@@ -2404,7 +2596,21 @@ RestoreSlotFromDisk(const char *name)
slot->in_use = true;
slot->active_pid = 0;
- slot->last_inactive_at = 0;
+
+ /*
+ * We set last_inactive_at only if inactive_timeout of the slot is
+ * specified so that the timeout is honored after the slot is restored
+ * from the disk.
+ *
+ * There's no point in allowing failover slots to get invalidated
+ * based on slot's inactive_timeout parameter on standby. The failover
+ * slots simply get synced from the primary on the standby.
+ */
+ if (slot->data.inactive_timeout > 0 &&
+ !(RecoveryInProgress() && slot->data.failover))
+ slot->last_inactive_at = GetCurrentTimestamp();
+ else
+ slot->last_inactive_at = 0;
restored = true;
break;
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index d6ef14fba6..7cc5c8bdf6 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -111,7 +111,7 @@ pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
tuple = heap_form_tuple(tupdesc, values, nulls);
result = HeapTupleGetDatum(tuple);
- ReplicationSlotRelease();
+ ReplicationSlotRelease(false);
PG_RETURN_DATUM(result);
}
@@ -224,7 +224,7 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
/* ok, slot is now fully created, mark it as persistent if needed */
if (!temporary)
ReplicationSlotPersist();
- ReplicationSlotRelease();
+ ReplicationSlotRelease(false);
PG_RETURN_DATUM(result);
}
@@ -257,6 +257,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
XLogRecPtr currlsn;
int slotno;
+ bool invalidated = false;
/*
* We don't require any special permission to see this function's data
@@ -287,6 +288,13 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
slot_contents = *slot;
SpinLockRelease(&slot->mutex);
+ /*
+ * Here's an opportunity to invalidate inactive replication slots
+ * based on timeout, so let's do it.
+ */
+ if (InvalidateReplicationSlotForInactiveTimeout(slot, false, true, true))
+ invalidated = true;
+
memset(values, 0, sizeof(values));
memset(nulls, 0, sizeof(nulls));
@@ -465,6 +473,15 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
LWLockRelease(ReplicationSlotControlLock);
+ /*
+ * If the slot has been invalidated, recalculate the resource limits
+ */
+ if (invalidated)
+ {
+ ReplicationSlotsComputeRequiredXmin(false);
+ ReplicationSlotsComputeRequiredLSN();
+ }
+
return (Datum) 0;
}
@@ -667,7 +684,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
moveto = Min(moveto, GetXLogReplayRecPtr(NULL));
/* Acquire the slot so we "own" it */
- ReplicationSlotAcquire(NameStr(*slotname), true);
+ ReplicationSlotAcquire(NameStr(*slotname), true, true);
/* A slot whose restart_lsn has never been reserved cannot be advanced */
if (XLogRecPtrIsInvalid(MyReplicationSlot->data.restart_lsn))
@@ -710,7 +727,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
ReplicationSlotsComputeRequiredXmin(false);
ReplicationSlotsComputeRequiredLSN();
- ReplicationSlotRelease();
+ ReplicationSlotRelease(true);
/* Return the reached position. */
values[1] = LSNGetDatum(endlsn);
@@ -954,7 +971,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
tuple = heap_form_tuple(tupdesc, values, nulls);
result = HeapTupleGetDatum(tuple);
- ReplicationSlotRelease();
+ ReplicationSlotRelease(false);
PG_RETURN_DATUM(result);
}
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 0420274247..b6795048cc 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -334,7 +334,7 @@ WalSndErrorCleanup(void)
wal_segment_close(xlogreader);
if (MyReplicationSlot != NULL)
- ReplicationSlotRelease();
+ ReplicationSlotRelease(true);
ReplicationSlotCleanup();
@@ -846,7 +846,7 @@ StartReplication(StartReplicationCmd *cmd)
if (cmd->slotname)
{
- ReplicationSlotAcquire(cmd->slotname, true);
+ ReplicationSlotAcquire(cmd->slotname, true, true);
if (SlotIsLogical(MyReplicationSlot))
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
@@ -992,7 +992,7 @@ StartReplication(StartReplicationCmd *cmd)
}
if (cmd->slotname)
- ReplicationSlotRelease();
+ ReplicationSlotRelease(true);
/*
* Copy is finished now. Send a single-row result set indicating the next
@@ -1407,7 +1407,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
do_tup_output(tstate, values, nulls);
end_tup_output(tstate);
- ReplicationSlotRelease();
+ ReplicationSlotRelease(false);
}
/*
@@ -1483,7 +1483,7 @@ StartLogicalReplication(StartReplicationCmd *cmd)
Assert(!MyReplicationSlot);
- ReplicationSlotAcquire(cmd->slotname, true);
+ ReplicationSlotAcquire(cmd->slotname, true, true);
/*
* Force a disconnect, so that the decoding code doesn't need to care
@@ -1545,7 +1545,7 @@ StartLogicalReplication(StartReplicationCmd *cmd)
WalSndLoop(XLogSendLogical);
FreeDecodingContext(logical_decoding_ctx);
- ReplicationSlotRelease();
+ ReplicationSlotRelease(true);
replication_active = false;
if (got_STOPPING)
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index fd4199a098..749de2741e 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -4407,7 +4407,7 @@ PostgresMain(const char *dbname, const char *username)
* callback ensuring correct cleanup on FATAL errors.
*/
if (MyReplicationSlot != NULL)
- ReplicationSlotRelease();
+ ReplicationSlotRelease(true);
/* We also want to cleanup temporary slots on error. */
ReplicationSlotCleanup();
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index c54b08fe18..d56ecf4137 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -299,7 +299,7 @@ binary_upgrade_logical_slot_has_caught_up(PG_FUNCTION_ARGS)
slot_name = PG_GETARG_NAME(0);
/* Acquire the given slot */
- ReplicationSlotAcquire(NameStr(*slot_name), true);
+ ReplicationSlotAcquire(NameStr(*slot_name), true, false);
Assert(SlotIsLogical(MyReplicationSlot));
@@ -310,7 +310,7 @@ binary_upgrade_logical_slot_has_caught_up(PG_FUNCTION_ARGS)
found_pending_wal = LogicalReplicationSlotHasPendingWal(end_of_wal);
/* Clean up */
- ReplicationSlotRelease();
+ ReplicationSlotRelease(false);
PG_RETURN_BOOL(!found_pending_wal);
}
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 9cd4bf98e5..bd4ad48ce8 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -53,6 +53,8 @@ typedef enum ReplicationSlotInvalidationCause
RS_INVAL_HORIZON,
/* wal_level insufficient for slot */
RS_INVAL_WAL_LEVEL,
+ /* inactive slot timeout has occurred */
+ RS_INVAL_INACTIVE_TIMEOUT,
} ReplicationSlotInvalidationCause;
extern PGDLLIMPORT const char *const SlotInvalidationCauses[];
@@ -249,8 +251,9 @@ extern void ReplicationSlotDropAcquired(void);
extern void ReplicationSlotAlter(const char *name, bool failover,
int inactive_timeout);
-extern void ReplicationSlotAcquire(const char *name, bool nowait);
-extern void ReplicationSlotRelease(void);
+extern void ReplicationSlotAcquire(const char *name, bool nowait,
+ bool check_for_invalidation);
+extern void ReplicationSlotRelease(bool set_last_inactive_at);
extern void ReplicationSlotCleanup(void);
extern void ReplicationSlotSave(void);
extern void ReplicationSlotSaveToPath(ReplicationSlot *slot, const char *dir,
@@ -270,6 +273,10 @@ extern bool InvalidateObsoleteReplicationSlots(ReplicationSlotInvalidationCause
XLogSegNo oldestSegno,
Oid dboid,
TransactionId snapshotConflictHorizon);
+extern bool InvalidateReplicationSlotForInactiveTimeout(ReplicationSlot *slot,
+ bool need_control_lock,
+ bool need_mutex,
+ bool persist_state);
extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_lock);
extern int ReplicationSlotIndex(ReplicationSlot *slot);
extern bool ReplicationSlotName(int index, Name name);
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index b1eb77b1ec..708a2a3798 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -51,6 +51,7 @@ tests += {
't/040_standby_failover_slots_sync.pl',
't/041_checkpoint_at_promote.pl',
't/042_low_level_backup.pl',
+ 't/050_invalidate_slots.pl',
],
},
}
diff --git a/src/test/recovery/t/050_invalidate_slots.pl b/src/test/recovery/t/050_invalidate_slots.pl
new file mode 100644
index 0000000000..6adaa1d648
--- /dev/null
+++ b/src/test/recovery/t/050_invalidate_slots.pl
@@ -0,0 +1,170 @@
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test for replication slots invalidation
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Utils;
+use PostgreSQL::Test::Cluster;
+use Test::More;
+use Time::HiRes qw(usleep);
+
+# Check for invalidation of slot in server log.
+sub check_slots_invalidation_in_server_log
+{
+ my ($node, $slot_name, $offset) = @_;
+ my $invalidated = 0;
+
+ for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++)
+ {
+ $node->safe_psql('postgres', "CHECKPOINT");
+ if ($node->log_contains(
+ "invalidating obsolete replication slot \"$slot_name\"", $offset))
+ {
+ $invalidated = 1;
+ last;
+ }
+ usleep(100_000);
+ }
+ ok($invalidated, "check that slot $slot_name invalidation has been logged");
+}
+
+# =============================================================================
+# Testcase start: Invalidate streaming standby's slot due to inactive_timeout
+#
+
+# Initialize primary node
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 'logical');
+
+# Avoid checkpoint during the test, otherwise, the test can get unpredictable
+$primary->append_conf(
+ 'postgresql.conf', q{
+checkpoint_timeout = 1h
+autovacuum = off
+});
+$primary->start;
+
+# Take backup
+my $backup_name = 'my_backup';
+$primary->backup($backup_name);
+
+# Create a standby linking to the primary using the replication slot
+my $standby1 = PostgreSQL::Test::Cluster->new('standby1');
+$standby1->init_from_backup($primary, $backup_name, has_streaming => 1);
+$standby1->append_conf(
+ 'postgresql.conf', q{
+primary_slot_name = 'sb1_slot'
+});
+
+# Set timeout so that the slot when inactive will get invalidated after the
+# timeout.
+my $inactive_timeout = 1;
+$primary->safe_psql(
+ 'postgres', qq[
+ SELECT pg_create_physical_replication_slot(slot_name := 'sb1_slot', inactive_timeout := $inactive_timeout);
+]);
+
+$standby1->start;
+
+# Wait until standby has replayed enough data
+$primary->wait_for_catchup($standby1);
+
+# The inactive replication slot info should be null when the slot is active
+my $result = $primary->safe_psql(
+ 'postgres', qq[
+ SELECT last_inactive_at IS NULL, inactive_timeout = $inactive_timeout
+ FROM pg_replication_slots WHERE slot_name = 'sb1_slot';
+]);
+is($result, "t|t",
+ 'check the inactive replication slot info for an active slot');
+
+my $logstart = -s $primary->logfile;
+
+# Stop standby to make the replication slot on primary inactive
+$standby1->stop;
+
+# Wait for the inactive replication slot info to be updated
+$primary->poll_query_until(
+ 'postgres', qq[
+ SELECT COUNT(slot_name) = 1 FROM pg_replication_slots
+ WHERE last_inactive_at IS NOT NULL
+ AND slot_name = 'sb1_slot'
+ AND inactive_timeout = $inactive_timeout;
+])
+ or die
+ "Timed out while waiting for inactive replication slot info to be updated";
+
+check_slots_invalidation_in_server_log($primary, 'sb1_slot', $logstart);
+
+# Wait for the inactive replication slots to be invalidated.
+$primary->poll_query_until(
+ 'postgres', qq[
+ SELECT COUNT(slot_name) = 1 FROM pg_replication_slots
+ WHERE slot_name = 'sb1_slot' AND
+ invalidation_reason = 'inactive_timeout';
+])
+ or die
+ "Timed out while waiting for inactive replication slot sb1_slot to be invalidated";
+
+# Testcase end: Invalidate streaming standby's slot due to inactive_timeout
+# =============================================================================
+
+# =============================================================================
+# Testcase start: Invalidate logical subscriber's slot due to inactive_timeout
+my $publisher = $primary;
+
+# Create subscriber node
+my $subscriber = PostgreSQL::Test::Cluster->new('sub');
+$subscriber->init;
+$subscriber->start;
+
+# Create tables
+$publisher->safe_psql('postgres', "CREATE TABLE test_tbl (id int)");
+$subscriber->safe_psql('postgres', "CREATE TABLE test_tbl (id int)");
+
+# Insert some data
+$subscriber->safe_psql('postgres',
+ "INSERT INTO test_tbl VALUES (generate_series(1, 5));");
+
+# Setup logical replication
+my $publisher_connstr = $publisher->connstr . ' dbname=postgres';
+$publisher->safe_psql('postgres', "CREATE PUBLICATION pub FOR ALL TABLES");
+$subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION sub CONNECTION '$publisher_connstr' PUBLICATION pub WITH (slot_name = 'lsub1_slot')"
+);
+
+$subscriber->wait_for_subscription_sync($publisher, 'sub');
+
+$result = $subscriber->safe_psql('postgres', "SELECT count(*) FROM test_tbl");
+
+is($result, qq(5), "check initial copy was done");
+
+# Alter slot to set inactive_timeout
+$publisher->safe_psql(
+ 'postgres', qq[
+ SELECT pg_alter_replication_slot(slot_name := 'lsub1_slot', inactive_timeout := $inactive_timeout);
+]);
+
+$logstart = -s $publisher->logfile;
+
+# Stop subscriber to make the replication slot on publisher inactive
+$subscriber->stop;
+
+# Wait for the inactive replication slot info to be updated
+$publisher->poll_query_until(
+ 'postgres', qq[
+ SELECT COUNT(slot_name) = 1 FROM pg_replication_slots
+ WHERE last_inactive_at IS NOT NULL
+ AND slot_name = 'lsub1_slot'
+ AND inactive_timeout = $inactive_timeout;
+])
+ or die
+ "Timed out while waiting for inactive replication slot info to be updated";
+
+check_slots_invalidation_in_server_log($publisher, 'lsub1_slot', $logstart);
+
+# Testcase end: Invalidate logical subscriber's slot due to inactive_timeout
+# =============================================================================
+
+done_testing();
--
2.34.1
view thread (43+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: Introduce XID age and inactive timeout based replication slot invalidation
In-Reply-To: <CALj2ACWuNHJ3qBMmmy6i1YA9V8JDRSaciL2bBM+72bRcqiBE2g@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