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: Thu, 21 Mar 2024 05:05:46 +0530
Message-ID: <CALj2ACUvK7ShPyirAURWf62=qOQqu=NwnyL5CMfVjM5Ody7Oxw@mail.gmail.com> (raw)
In-Reply-To: <[email protected]>
References: <CAA4eK1+C3QxAZ6UtpSn9umGB33YjtYJgxA_xVXCacATEfZG4YQ@mail.gmail.com>
<CALj2ACUNDR==n6ebq86ShBqdf64EasANuxmXQBDYHDHeN3G5Fg@mail.gmail.com>
<CAA4eK1LF28ZTwmf3D-OQWeMdQqjB56P+iAmfWM7Bcwow3icBJQ@mail.gmail.com>
<CALj2ACVtLt0rPnr4NS6c72DdtQmpyQA=qcMaVfn6vu7j5yQvzw@mail.gmail.com>
<CAA4eK1+7Y6bknNNentcNfkmCBfm3j2fLjA4JjVydz3TB2MTjVA@mail.gmail.com>
<CALj2ACUR0=ByDv2VcqqbT0x+0T0gde7Zosb0A_8syPM63WSoFw@mail.gmail.com>
<CAA4eK1K7awEe1PXmmAUOoc+VnUJ76cPbYAV7mwwcW+2OxNyo7A@mail.gmail.com>
<[email protected]>
<CALj2ACXib_+Kuy3_WbTBJkURYGwO7ngS8+HYATReJtXShNCtNw@mail.gmail.com>
<CAA4eK1KZyjdn7oXS3d7qJU5ag9ss4xSk0Jkjsh5NOrRdkPkV1g@mail.gmail.com>
<[email protected]>
On Wed, Mar 20, 2024 at 1:04 PM Bertrand Drouvot
<[email protected]> wrote:
>
> On Wed, Mar 20, 2024 at 08:58:05AM +0530, Amit Kapila wrote:
> > On Wed, Mar 20, 2024 at 12:49 AM Bharath Rupireddy
> > <[email protected]> wrote:
> > >
> > > Following are some open points:
> > >
> > > 1. Where to do inactive_timeout invalidation exactly if not the checkpointer.
> > >
> > I have suggested to do it at the time of CheckpointReplicationSlots()
> > and Bertrand suggested to do it whenever we resume using the slot. I
> > think we should follow both the suggestions.
>
> Agree. I also think that pg_get_replication_slots() would be a good place, so
> that queries would return the right invalidation status.
I've addressed review comments and attaching the v13 patches with the
following changes:
1. Invalidate replication slot due to inactive_timeout:
1.1 In CheckpointReplicationSlots() to help with automatic invalidation.
1.2 In pg_get_replication_slots to help readers see the latest slot information.
1.3 In ReplicationSlotAcquire for walsenders as typically walsenders
are the ones that use slots for longer durations for streaming
standbys and logical subscribers.
1.4 In ReplicationSlotAcquire when called from
pg_logical_slot_get_changes_guts to help with logical decoding clients
to disallow decoding from invalidated slots.
1.5 In ReplicationSlotAcquire when called from
pg_replication_slot_advance to help with disallowing advancing
invalidated slots.
2. Have a new input parameter bool check_for_invalidation for
ReplicationSlotAcquire(). When true, check for the inactive_timeout
invalidation, if invalidated, error out.
3. Have a new function to just do inactive_timeout invalidation.
4. Do not update last_inactive_at for failover slots on standby to not
invalidate failover slots on the standby.
5. In ReplicationSlotAcquire(), invalidate the slot before making it active.
6. Make last_inactive_at a shared-memory parameter as opposed to an
on-disk parameter to help not count the server downtime for inactive
time.
7. Let the failover slot on standby and pg_upgraded slots get
inactive_timeout parameter from the primary and old cluster
respectively.
Please see the attached v13 patches.
--
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com
Attachments:
[application/octet-stream] v13-0001-Track-invalidation_reason-in-pg_replication_slot.patch (25.1K, ../CALj2ACUvK7ShPyirAURWf62=qOQqu=NwnyL5CMfVjM5Ody7Oxw@mail.gmail.com/2-v13-0001-Track-invalidation_reason-in-pg_replication_slot.patch)
download | inline diff:
From 2750c2fd767579cab9a4295d9aca7ef0dd42b9b3 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Wed, 20 Mar 2024 22:37:15 +0000
Subject: [PATCH v13 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 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). One now can look at the
new invalidation_reason column for logical slots conflict with
recovery.
---
doc/src/sgml/ref/pgupgrade.sgml | 4 +-
doc/src/sgml/system-views.sgml | 63 +++++++++++--------
src/backend/catalog/system_views.sql | 5 +-
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 | 39 ++++++------
.../t/040_standby_failover_slots_sync.pl | 4 +-
src/test/regress/expected/rules.out | 7 ++-
12 files changed, 116 insertions(+), 94 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..95355743ca 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2525,34 +2525,13 @@ 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:
- <itemizedlist spacing="compact">
- <listitem>
- <para>
- <literal>wal_removed</literal> means that the required WAL has been
- removed.
- </para>
- </listitem>
- <listitem>
- <para>
- <literal>rows_removed</literal> means that the required rows have
- been removed.
- </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.
- </para>
- </listitem>
- </itemizedlist>
+ 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>
@@ -2581,6 +2560,38 @@ 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>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>
+ <literal>wal_removed</literal> means that the required WAL has been
+ removed.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ <literal>rows_removed</literal> means that the required rows have
+ been removed.
+ </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.
+ </para>
+ </listitem>
+ </itemizedlist>
+ </para></entry>
+ </row>
+
</tbody>
</tgroup>
</table>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 04227a72d1..cd22dad959 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1023,9 +1023,10 @@ CREATE VIEW pg_replication_slots AS
L.wal_status,
L.safe_wal_size,
L.two_phase,
- L.conflict_reason,
+ L.conflicting,
L.failover,
- L.synced
+ L.synced,
+ L.invalidation_reason
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 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..dfaac999f1 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,22 +410,32 @@ 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);
}
values[i++] = BoolGetDatum(slot_contents.data.failover);
values[i++] = BoolGetDatum(slot_contents.data.synced);
+ if (cause == RS_INVAL_NONE)
+ nulls[i++] = true;
+ else
+ values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]);
+
Assert(i == PG_GET_REPLICATION_SLOTS_COLS);
tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c
index b5b8d11602..34a157f792 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 conflicting 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..cf116bc548 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,bool,bool,text}',
+ 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,failover,synced,invalidation_reason}',
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..2203841ca1 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,
@@ -696,14 +697,14 @@ ok( $node_standby->poll_query_until(
'confl_active_logicalslot not updated'
) or die "Timed out waiting confl_active_logicalslot to be updated";
-# Verify slots are reported as non conflicting in pg_replication_slots
+# Verify slots are reported as valid in pg_replication_slots
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');
+ 'Logical slots are reported as valid');
# Turn hot_standby_feedback back on
change_hot_standby_feedback_and_wait_for_xmins(1, 0);
@@ -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..19c44c0cb7 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.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)
+ l.synced,
+ l.invalidation_reason
+ 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, failover, synced, invalidation_reason)
LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
pg_roles| SELECT pg_authid.rolname,
pg_authid.rolsuper,
--
2.34.1
[application/octet-stream] v13-0002-Track-last_inactive_at-for-replication-slots-in-.patch (6.3K, ../CALj2ACUvK7ShPyirAURWf62=qOQqu=NwnyL5CMfVjM5Ody7Oxw@mail.gmail.com/3-v13-0002-Track-last_inactive_at-for-replication-slots-in-.patch)
download | inline diff:
From b1e31a64e36ae27017813d7f1f602a0067c7cc39 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Wed, 20 Mar 2024 22:37:40 +0000
Subject: [PATCH v13 2/6] Track last_inactive_at for replication slots in
shared memory
---
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 +++--
6 files changed, 33 insertions(+), 7 deletions(-)
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index cd22dad959..2fa4272006 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.failover,
L.synced,
- L.invalidation_reason
+ L.invalidation_reason,
+ 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 dfaac999f1..2c33cc0c16 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)
else
values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]);
+ 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 cf116bc548..d89a223a60 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,bool,bool,text}',
- 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,failover,synced,invalidation_reason}',
+ proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text,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,failover,synced,invalidation_reason,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 19c44c0cb7..88fbd6a53c 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.failover,
l.synced,
- l.invalidation_reason
- 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, failover, synced, invalidation_reason)
+ l.invalidation_reason,
+ 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, failover, synced, invalidation_reason, 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/octet-stream] v13-0003-Allow-setting-inactive_timeout-for-replication-s.patch (33.2K, ../CALj2ACUvK7ShPyirAURWf62=qOQqu=NwnyL5CMfVjM5Ody7Oxw@mail.gmail.com/4-v13-0003-Allow-setting-inactive_timeout-for-replication-s.patch)
download | inline diff:
From 9339d8203cafc03f7590c6770e7966321b8e3ea5 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Wed, 20 Mar 2024 22:37:54 +0000
Subject: [PATCH v13 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/include/catalog/pg_proc.dat | 22 ++--
src/include/replication/slot.h | 5 +-
.../t/040_standby_failover_slots_sync.pl | 11 +-
src/test/regress/expected/rules.out | 5 +-
17 files changed, 257 insertions(+), 39 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 030ea8affd..9467ff86b3 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -28163,7 +28163,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> )
@@ -28180,9 +28180,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>
@@ -28207,7 +28210,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> )
@@ -28226,7 +28229,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 95355743ca..ec60c43038 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2750,6 +2750,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 2fa4272006..a43048ae93 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.failover,
L.synced,
L.invalidation_reason,
- 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 2c33cc0c16..55ff73cc78 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 34a157f792..6817e9be67 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/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index d89a223a60..80c281c8a5 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,bool,bool,text,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,failover,synced,invalidation_reason,last_inactive_at}',
+ proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text,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,failover,synced,invalidation_reason,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..e4e244effb 100644
--- a/src/test/recovery/t/040_standby_failover_slots_sync.pl
+++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl
@@ -153,7 +153,7 @@ $primary->append_conf('postgresql.conf', "log_min_messages = 'debug2'");
$primary->reload;
$primary->psql('postgres',
- q{SELECT pg_create_logical_replication_slot('lsub2_slot', 'test_decoding', false, false, true);}
+ q{SELECT pg_create_logical_replication_slot('lsub2_slot', 'test_decoding', false, false, true, 3600);}
);
$primary->psql('postgres',
@@ -190,6 +190,15 @@ 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',
+ q{SELECT inactive_timeout FROM pg_replication_slots WHERE slot_name = 'lsub2_slot' AND synced AND NOT temporary;}
+ ),
+ "3600",
+ '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 88fbd6a53c..1c683ceaca 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.failover,
l.synced,
l.invalidation_reason,
- 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, failover, synced, invalidation_reason, 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, failover, synced, invalidation_reason, 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/octet-stream] v13-0004-Allow-setting-inactive_timeout-in-the-replicatio.patch (17.9K, ../CALj2ACUvK7ShPyirAURWf62=qOQqu=NwnyL5CMfVjM5Ody7Oxw@mail.gmail.com/5-v13-0004-Allow-setting-inactive_timeout-in-the-replicatio.patch)
download | inline diff:
From faee83481f2e04d5ac5a62657d2eadecb7205247 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Wed, 20 Mar 2024 22:38:24 +0000
Subject: [PATCH v13 4/6] Allow setting inactive_timeout in the replication
command
---
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 195771920f..aba5e981d7 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -808,8 +808,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);
@@ -852,10 +854,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 ff62542b03..77def17386 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/octet-stream] v13-0005-Add-inactive_timeout-option-to-subscriptions.patch (62.5K, ../CALj2ACUvK7ShPyirAURWf62=qOQqu=NwnyL5CMfVjM5Ody7Oxw@mail.gmail.com/6-v13-0005-Add-inactive_timeout-option-to-subscriptions.patch)
download | inline diff:
From 4c547f888a329d759cd2b5df0d6a08ed39344912 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Wed, 20 Mar 2024 22:38:44 +0000
Subject: [PATCH v13 5/6] Add inactive_timeout option to subscriptions
---
doc/src/sgml/catalogs.sgml | 11 ++
doc/src/sgml/ref/alter_subscription.sgml | 5 +-
doc/src/sgml/ref/create_subscription.sgml | 12 ++
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_views.sql | 3 +-
src/backend/commands/subscriptioncmds.c | 83 +++++++++--
src/backend/replication/logical/tablesync.c | 2 +-
src/bin/pg_dump/pg_dump.c | 22 ++-
src/bin/pg_dump/pg_dump.h | 1 +
src/bin/pg_upgrade/t/003_logical_slots.pl | 6 +-
src/bin/psql/describe.c | 7 +-
src/bin/psql/tab-complete.c | 14 +-
src/include/catalog/pg_subscription.h | 9 ++
src/test/regress/expected/subscription.out | 152 ++++++++++----------
src/test/subscription/t/001_rep_changes.pl | 63 ++++++++
15 files changed, 288 insertions(+), 103 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index b7980eb499..4126e2d3cd 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -8028,6 +8028,17 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subinactivetimeout</structfield> <type>int4</type>
+ </para>
+ <para>
+ When set to a non-zero value, specifies the amount of time in seconds
+ the associated replication slots (i.e. the main slot and the table
+ sync slots) in the upstream database are allowed to be inactive.
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>subconninfo</structfield> <type>text</type>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 413ce68ce2..d02d6232de 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -227,8 +227,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>,
<link linkend="sql-createsubscription-params-with-password-required"><literal>password_required</literal></link>,
<link linkend="sql-createsubscription-params-with-run-as-owner"><literal>run_as_owner</literal></link>,
- <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>, and
- <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>.
+ <link linkend="sql-createsubscription-params-with-origin"><literal>origin</literal></link>,
+ <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-inactive-timeout"><literal>inactive_timeout</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
</para>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 15794731bb..7be4610921 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -414,6 +414,18 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</para>
</listitem>
</varlistentry>
+
+ <varlistentry id="sql-createsubscription-params-with-inactive-timeout">
+ <term><literal>inactive_timeout</literal> (<type>int4</type>)</term>
+ <listitem>
+ <para>
+ When set to a non-zero value, specifies the amount of time in seconds
+ the associated replication slots (i.e. the main slot and the table
+ sync slots) in the upstream database are allowed to be inactive.
+ The default is <literal>0</literal>.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 9efc9159f2..f874146e72 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -72,6 +72,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->passwordrequired = subform->subpasswordrequired;
sub->runasowner = subform->subrunasowner;
sub->failover = subform->subfailover;
+ sub->inactivetimeout = subform->subinactivetimeout;
/* Get conninfo */
datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index a43048ae93..6005315ce3 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1363,7 +1363,8 @@ REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
subpasswordrequired, subrunasowner, subfailover,
- subslotname, subsynccommit, subpublications, suborigin)
+ subinactivetimeout, subslotname, subsynccommit,
+ subpublications, suborigin)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 4562de49c4..c39f765af8 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -70,8 +70,9 @@
#define SUBOPT_PASSWORD_REQUIRED 0x00000800
#define SUBOPT_RUN_AS_OWNER 0x00001000
#define SUBOPT_FAILOVER 0x00002000
-#define SUBOPT_LSN 0x00004000
-#define SUBOPT_ORIGIN 0x00008000
+#define SUBOPT_INACTIVE_TIMEOUT 0x00004000
+#define SUBOPT_LSN 0x00008000
+#define SUBOPT_ORIGIN 0x00010000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -97,6 +98,7 @@ typedef struct SubOpts
bool passwordrequired;
bool runasowner;
bool failover;
+ int inactivetimeout;
char *origin;
XLogRecPtr lsn;
} SubOpts;
@@ -159,6 +161,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->runasowner = false;
if (IsSet(supported_opts, SUBOPT_FAILOVER))
opts->failover = false;
+ if (IsSet(supported_opts, SUBOPT_INACTIVE_TIMEOUT))
+ opts->inactivetimeout = 0;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
@@ -316,6 +320,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->specified_opts |= SUBOPT_FAILOVER;
opts->failover = defGetBoolean(defel);
}
+ else if (IsSet(supported_opts, SUBOPT_INACTIVE_TIMEOUT) &&
+ strcmp(defel->defname, "inactive_timeout") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_INACTIVE_TIMEOUT))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_INACTIVE_TIMEOUT;
+ opts->inactivetimeout = defGetInt32(defel);
+ }
else if (IsSet(supported_opts, SUBOPT_ORIGIN) &&
strcmp(defel->defname, "origin") == 0)
{
@@ -610,7 +623,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
- SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER | SUBOPT_ORIGIN);
+ SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
+ SUBOPT_INACTIVE_TIMEOUT | SUBOPT_ORIGIN);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -717,6 +731,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_subpasswordrequired - 1] = BoolGetDatum(opts.passwordrequired);
values[Anum_pg_subscription_subrunasowner - 1] = BoolGetDatum(opts.runasowner);
values[Anum_pg_subscription_subfailover - 1] = BoolGetDatum(opts.failover);
+ values[Anum_pg_subscription_subinactivetimeout - 1] = Int32GetDatum(opts.inactivetimeout);
values[Anum_pg_subscription_subconninfo - 1] =
CStringGetTextDatum(conninfo);
if (opts.slot_name)
@@ -827,7 +842,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
twophase_enabled = true;
walrcv_create_slot(wrconn, opts.slot_name, false, twophase_enabled,
- opts.failover, 0, CRS_NOEXPORT_SNAPSHOT, NULL);
+ opts.failover, opts.inactivetimeout,
+ CRS_NOEXPORT_SNAPSHOT, NULL);
if (twophase_enabled)
UpdateTwoPhaseState(subid, LOGICALREP_TWOPHASE_STATE_ENABLED);
@@ -840,16 +856,19 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
/*
* If the slot_name is specified without the create_slot option,
* it is possible that the user intends to use an existing slot on
- * the publisher, so here we alter the failover property of the
- * slot to match the failover value in subscription.
+ * the publisher, so here we alter the failover and
+ * inactive_timeout properties of the slot to match the failover
+ * and inactive_timeout values in subscription.
*
- * We do not need to change the failover to false if the server
- * does not support failover (e.g. pre-PG17).
+ * We do not need to change the failover to false and
+ * inactive_timeout to zero if the server does not support them
+ * (e.g. pre-PG17).
*/
else if (opts.slot_name &&
- (opts.failover || walrcv_server_version(wrconn) >= 170000))
+ (opts.failover || opts.inactivetimeout > 0 ||
+ walrcv_server_version(wrconn) >= 170000))
{
- walrcv_alter_slot(wrconn, opts.slot_name, &opts.failover, NULL);
+ walrcv_alter_slot(wrconn, opts.slot_name, &opts.failover, &opts.inactivetimeout);
}
}
PG_FINALLY();
@@ -1168,7 +1187,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
SUBOPT_STREAMING | SUBOPT_DISABLE_ON_ERR |
SUBOPT_PASSWORD_REQUIRED |
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
- SUBOPT_ORIGIN);
+ SUBOPT_INACTIVE_TIMEOUT | SUBOPT_ORIGIN);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1272,6 +1291,19 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_subfailover - 1] = true;
}
+ if (IsSet(opts.specified_opts, SUBOPT_INACTIVE_TIMEOUT))
+ {
+ if (!sub->slotname)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot set %s for a subscription that does not have a slot name",
+ "inactive_timeout")));
+
+ values[Anum_pg_subscription_subinactivetimeout - 1] =
+ BoolGetDatum(opts.inactivetimeout);
+ replaces[Anum_pg_subscription_subinactivetimeout - 1] = true;
+ }
+
if (IsSet(opts.specified_opts, SUBOPT_ORIGIN))
{
values[Anum_pg_subscription_suborigin - 1] =
@@ -1550,6 +1582,35 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
PG_END_TRY();
}
+ if (replaces[Anum_pg_subscription_subinactivetimeout - 1])
+ {
+ bool must_use_password;
+ char *err;
+ WalReceiverConn *wrconn;
+
+ /* Load the library providing us libpq calls. */
+ load_file("libpqwalreceiver", false);
+
+ /* Try to connect to the publisher. */
+ must_use_password = sub->passwordrequired && !sub->ownersuperuser;
+ wrconn = walrcv_connect(sub->conninfo, true, true, must_use_password,
+ sub->name, &err);
+ if (!wrconn)
+ ereport(ERROR,
+ (errcode(ERRCODE_CONNECTION_FAILURE),
+ errmsg("could not connect to the publisher: %s", err)));
+
+ PG_TRY();
+ {
+ walrcv_alter_slot(wrconn, sub->slotname, NULL, &opts.inactivetimeout);
+ }
+ PG_FINALLY();
+ {
+ walrcv_disconnect(wrconn);
+ }
+ PG_END_TRY();
+ }
+
table_close(rel, RowExclusiveLock);
ObjectAddressSet(myself, SubscriptionRelationId, subid);
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 59f8e5fbaa..c660f1e65e 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -1431,7 +1431,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
walrcv_create_slot(LogRepWorkerWalRcvConn,
slotname, false /* permanent */ , false /* two_phase */ ,
MySubscription->failover,
- 0,
+ MySubscription->inactivetimeout,
CRS_USE_SNAPSHOT, origin_startpos);
/*
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index d275b31605..a31d5a213f 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -4673,6 +4673,7 @@ getSubscriptions(Archive *fout)
int i_suboriginremotelsn;
int i_subenabled;
int i_subfailover;
+ int i_subinactivetimeout;
int i,
ntups;
@@ -4739,11 +4740,13 @@ getSubscriptions(Archive *fout)
if (dopt->binary_upgrade && fout->remoteVersion >= 170000)
appendPQExpBufferStr(query, " o.remote_lsn AS suboriginremotelsn,\n"
" s.subenabled,\n"
- " s.subfailover\n");
+ " s.subfailover,\n"
+ " s.subinactivetimeout\n");
else
appendPQExpBufferStr(query, " NULL AS suboriginremotelsn,\n"
" false AS subenabled,\n"
- " false AS subfailover\n");
+ " false AS subfailover,\n"
+ " 0 AS subinactivetimeout\n");
appendPQExpBufferStr(query,
"FROM pg_subscription s\n");
@@ -4783,6 +4786,7 @@ getSubscriptions(Archive *fout)
i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
i_subenabled = PQfnumber(res, "subenabled");
i_subfailover = PQfnumber(res, "subfailover");
+ i_subinactivetimeout = PQfnumber(res, "subinactivetimeout");
subinfo = pg_malloc(ntups * sizeof(SubscriptionInfo));
@@ -4829,6 +4833,8 @@ getSubscriptions(Archive *fout)
pg_strdup(PQgetvalue(res, i, i_subenabled));
subinfo[i].subfailover =
pg_strdup(PQgetvalue(res, i, i_subfailover));
+ subinfo[i].subinactivetimeout =
+ atooid(PQgetvalue(res, i, i_subinactivetimeout));
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
@@ -5110,6 +5116,18 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
appendPQExpBuffer(query, "ALTER SUBSCRIPTION %s SET(failover = true);\n", qsubname);
}
+ if (subinfo->subinactivetimeout > 0)
+ {
+ /*
+ * Preserve subscription's inactive_timeout option to be able to
+ * use it after the upgrade.
+ */
+ appendPQExpBufferStr(query,
+ "\n-- For binary upgrade, must preserve the subscriber's inactive_timeout option.\n");
+ appendPQExpBuffer(query, "ALTER SUBSCRIPTION %s SET(inactive_timeout = %d);\n",
+ qsubname, subinfo->subinactivetimeout);
+ }
+
if (strcmp(subinfo->subenabled, "t") == 0)
{
/*
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 9bc93520b4..bfaedcd7e2 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -667,6 +667,7 @@ typedef struct _SubscriptionInfo
char *suborigin;
char *suboriginremotelsn;
char *subfailover;
+ int subinactivetimeout;
} SubscriptionInfo;
/*
diff --git a/src/bin/pg_upgrade/t/003_logical_slots.pl b/src/bin/pg_upgrade/t/003_logical_slots.pl
index 83d71c3084..8aa34d66cc 100644
--- a/src/bin/pg_upgrade/t/003_logical_slots.pl
+++ b/src/bin/pg_upgrade/t/003_logical_slots.pl
@@ -172,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 (two_phase = 'true', failover = 'true')
+ CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (two_phase = 'true', failover = 'true', inactive_timeout = 3600)
]);
$sub->wait_for_subscription_sync($oldpub, 'regress_sub');
@@ -192,8 +192,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 = 3600 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/bin/psql/describe.c b/src/bin/psql/describe.c
index 6433497bcd..73fcfa421d 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6581,7 +6581,7 @@ describeSubscriptions(const char *pattern, bool verbose)
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
false, false, false, false, false, false, false, false, false, false,
- false};
+ false, false};
if (pset.sversion < 100000)
{
@@ -6650,6 +6650,11 @@ describeSubscriptions(const char *pattern, bool verbose)
", subfailover AS \"%s\"\n",
gettext_noop("Failover"));
+ if (pset.sversion >= 170000)
+ appendPQExpBuffer(&buf,
+ ", subinactivetimeout AS \"%s\"\n",
+ gettext_noop("Inactive timeout"));
+
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
", subconninfo AS \"%s\"\n",
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 56d723de8a..bf7349bae1 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1946,9 +1946,10 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("(", "PUBLICATION");
/* ALTER SUBSCRIPTION <name> SET ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SET", "("))
- COMPLETE_WITH("binary", "disable_on_error", "failover", "origin",
- "password_required", "run_as_owner", "slot_name",
- "streaming", "synchronous_commit");
+ COMPLETE_WITH("binary", "disable_on_error", "failover",
+ "inactive_timeout", "origin", "password_required",
+ "run_as_owner", "slot_name", "streaming",
+ "synchronous_commit");
/* ALTER SUBSCRIPTION <name> SKIP ( */
else if (HeadMatches("ALTER", "SUBSCRIPTION", MatchAny) && TailMatches("SKIP", "("))
COMPLETE_WITH("lsn");
@@ -3344,9 +3345,10 @@ psql_completion(const char *text, int start, int end)
/* Complete "CREATE SUBSCRIPTION <name> ... WITH ( <opt>" */
else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "("))
COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
- "disable_on_error", "enabled", "failover", "origin",
- "password_required", "run_as_owner", "slot_name",
- "streaming", "synchronous_commit", "two_phase");
+ "disable_on_error", "enabled", "failover",
+ "inactive_timeout", "origin", "password_required",
+ "run_as_owner", "slot_name", "streaming",
+ "synchronous_commit", "two_phase");
/* CREATE TRIGGER --- is allowed inside CREATE SCHEMA, so use TailMatches */
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 0aa14ec4a2..1113cdf690 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -98,6 +98,11 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
* slots) in the upstream database are enabled
* to be synchronized to the standbys. */
+ int32 subinactivetimeout; /* Associated replication slots (i.e. the
+ * main slot and the table sync slots) in
+ * the upstream database are allowed to be
+ * inactive this amount of time. */
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -151,6 +156,10 @@ typedef struct Subscription
* (i.e. the main slot and the table sync
* slots) in the upstream database are enabled
* to be synchronized to the standbys. */
+ int32 inactivetimeout; /* Associated replication slots (i.e. the
+ * main slot and the table sync slots) in
+ * the upstream database are allowed to be
+ * inactive this amount of time. */
char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 1eee6b17b8..5d5dc1561d 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -118,18 +118,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -147,10 +147,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -159,10 +159,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | f | 0 | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -178,10 +178,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/12345
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist2 | 0/12345
(1 row)
-- ok - with lsn = NONE
@@ -190,10 +190,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
BEGIN;
@@ -225,10 +225,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | local | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | f | 0 | local | dbname=regress_doesnotexist2 | 0/0
(1 row)
-- rename back to keep the rest simple
@@ -257,19 +257,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -281,27 +281,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication already exists
@@ -316,10 +316,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication used more than once
@@ -334,10 +334,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -373,10 +373,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
--fail - alter of two_phase option not supported.
@@ -385,10 +385,10 @@ ERROR: unrecognized subscription parameter: "two_phase"
-- but can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -398,10 +398,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -414,18 +414,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Inactive timeout | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+------------------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | f | 0 | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/subscription/t/001_rep_changes.pl b/src/test/subscription/t/001_rep_changes.pl
index 9ccebd890a..e6fd2297d2 100644
--- a/src/test/subscription/t/001_rep_changes.pl
+++ b/src/test/subscription/t/001_rep_changes.pl
@@ -531,6 +531,69 @@ $node_publisher->poll_query_until('postgres',
or die
"Timed out while waiting for apply to restart after renaming SUBSCRIPTION";
+# Check inactive_timeout options with subscriptions
+#
+# Setup logical replication that will only be used for this test
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub_it FOR ALL TABLES;");
+
+# Create subscription with inactive_timeout set
+my $inactive_timeout = 300;
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_it1 CONNECTION '$publisher_connstr' PUBLICATION tap_pub_it WITH (inactive_timeout = $inactive_timeout, enabled = false);"
+);
+
+$result = $node_publisher->safe_psql('postgres',
+ "SELECT inactive_timeout FROM pg_replication_slots WHERE slot_name = 'tap_sub_it1';"
+);
+is($result, $inactive_timeout,
+ "check inactive_timeout passed via create subscription is set on slot created on publisher"
+);
+
+$result = $node_subscriber->safe_psql('postgres',
+ "SELECT subinactivetimeout FROM pg_subscription WHERE subname = 'tap_sub_it1';"
+);
+is($result, $inactive_timeout,
+ "check inactive_timeout passed via create subscription is set on pg_subscription on subscriber"
+);
+
+# Alter subscription with inactive_timeout set to a different value
+$inactive_timeout = 600;
+$node_subscriber->safe_psql('postgres',
+ "ALTER SUBSCRIPTION tap_sub_it1 SET (inactive_timeout = $inactive_timeout);"
+);
+
+$result = $node_publisher->safe_psql('postgres',
+ "SELECT inactive_timeout FROM pg_replication_slots WHERE slot_name = 'tap_sub_it1';"
+);
+is($result, $inactive_timeout,
+ "check inactive_timeout passed via alter subscription is set on slot created on publisher"
+);
+
+# Create subscription with using a pre-existing slot with inactive_timeout set.
+# This is equivalent to alter subscription to change the inactive_timeout.
+$inactive_timeout = 900;
+$node_publisher->safe_psql('postgres',
+ "SELECT * FROM pg_create_logical_replication_slot('slot_sub_it2', 'pgoutput');"
+);
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION tap_sub_it2 CONNECTION '$publisher_connstr' PUBLICATION tap_pub_it WITH (create_slot = false, slot_name = 'slot_sub_it2', inactive_timeout = $inactive_timeout);"
+);
+
+$result = $node_publisher->safe_psql('postgres',
+ "SELECT inactive_timeout FROM pg_replication_slots WHERE slot_name = 'slot_sub_it2';"
+);
+is($result, $inactive_timeout,
+ "check inactive_timeout passed via create subscription with slot_name is set on slot created on publisher"
+);
+
+# Drop subscriptions as we don't need them anymore
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_it1");
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_it2");
+
+# Drop publication as we don't need it anymore
+$node_publisher->safe_psql('postgres', "DROP PUBLICATION tap_pub_it");
+
# check all the cleanup
$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub_renamed");
--
2.34.1
[application/octet-stream] v13-0006-Add-inactive_timeout-based-replication-slot-inva.patch (31.1K, ../CALj2ACUvK7ShPyirAURWf62=qOQqu=NwnyL5CMfVjM5Ody7Oxw@mail.gmail.com/7-v13-0006-Add-inactive_timeout-based-replication-slot-inva.patch)
download | inline diff:
From 89fd621fb800bb2300c16567742906001fca7158 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <[email protected]>
Date: Wed, 20 Mar 2024 23:04:24 +0000
Subject: [PATCH v13 6/6] Add inactive_timeout based replication slot
invalidation
---
doc/src/sgml/func.sgml | 12 +-
doc/src/sgml/ref/create_subscription.sgml | 4 +-
doc/src/sgml/system-views.sgml | 10 +-
.../replication/logical/logicalfuncs.c | 4 +-
src/backend/replication/logical/slotsync.c | 8 +-
src/backend/replication/slot.c | 205 +++++++++++++++++-
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 | 168 ++++++++++++++
13 files changed, 425 insertions(+), 43 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 9467ff86b3..af5ac09a4b 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -28183,8 +28183,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>
@@ -28229,12 +28229,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/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 7be4610921..472592c750 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -421,8 +421,8 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<para>
When set to a non-zero value, specifies the amount of time in seconds
the associated replication slots (i.e. the main slot and the table
- sync slots) in the upstream database are allowed to be inactive.
- The default is <literal>0</literal>.
+ sync slots) in the upstream database are allowed to be inactive before
+ getting invalidated. The default is <literal>0</literal>.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index ec60c43038..d063e989a6 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2588,6 +2588,13 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
perform logical decoding.
</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>
@@ -2756,7 +2763,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 aba5e981d7..d007ed126f 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);
@@ -234,7 +238,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();
@@ -551,9 +555,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;
@@ -631,6 +640,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
@@ -664,7 +709,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 */
@@ -715,11 +760,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;
@@ -789,7 +843,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
@@ -814,7 +868,7 @@ ReplicationSlotAlter(const char *name, bool failover, int inactive_timeout)
Assert(MyReplicationSlot == NULL);
- ReplicationSlotAcquire(name, false);
+ ReplicationSlotAcquire(name, false, false);
if (SlotIsPhysical(MyReplicationSlot))
ereport(ERROR,
@@ -890,7 +944,7 @@ ReplicationSlotAlter(const char *name, bool failover, int inactive_timeout)
ReplicationSlotSave();
}
- ReplicationSlotRelease();
+ ReplicationSlotRelease(false);
}
/*
@@ -1542,6 +1596,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 +1712,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 +1842,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 +1869,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 +1921,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));
+ SaveSlotToPath(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 +2037,7 @@ void
CheckPointReplicationSlots(bool is_shutdown)
{
int i;
+ bool invalidated = false;
elog(DEBUG1, "performing replication slot checkpoint");
@@ -1894,6 +2061,13 @@ CheckPointReplicationSlots(bool is_shutdown)
/* save the slot to disk, locking is handled in SaveSlotToPath() */
sprintf(path, "pg_replslot/%s", NameStr(s->data.name));
+ /*
+ * Here's an opportunity to invalidate inactive replication slots
+ * based on timeout, so let's do it.
+ */
+ if (InvalidateReplicationSlotForInactiveTimeout(s, true, true, false))
+ invalidated = true;
+
/*
* Slot's data is not flushed each time the confirmed_flush LSN is
* updated as that could lead to frequent writes. However, we decide
@@ -1920,6 +2094,13 @@ CheckPointReplicationSlots(bool is_shutdown)
SaveSlotToPath(s, path, LOG);
}
LWLockRelease(ReplicationSlotAllocationLock);
+
+ /* If the slot has been invalidated, recalculate the resource limits */
+ if (invalidated)
+ {
+ ReplicationSlotsComputeRequiredXmin(false);
+ ReplicationSlotsComputeRequiredLSN();
+ }
}
/*
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 55ff73cc78..60055a6bb4 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);
}
@@ -258,6 +258,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
@@ -288,6 +289,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));
@@ -466,6 +474,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;
}
@@ -668,7 +685,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))
@@ -711,7 +728,7 @@ pg_replication_slot_advance(PG_FUNCTION_ARGS)
ReplicationSlotsComputeRequiredXmin(false);
ReplicationSlotsComputeRequiredLSN();
- ReplicationSlotRelease();
+ ReplicationSlotRelease(true);
/* Return the reached position. */
values[1] = LSNGetDatum(endlsn);
@@ -955,7 +972,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 77def17386..6a91e8a63f 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 ReplicationSlotMarkDirty(void);
@@ -268,6 +271,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..d046e1d5d7
--- /dev/null
+++ b/src/test/recovery/t/050_invalidate_slots.pl
@@ -0,0 +1,168 @@
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+# Test for replication slots invalidation
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Utils;
+use PostgreSQL::Test::Cluster;
+use Test::More;
+use Time::HiRes qw(usleep);
+
+# =============================================================================
+# Testcase start: 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 = 5;
+$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";
+
+my $invalidated = 0;
+for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++)
+{
+ $primary->safe_psql('postgres', "CHECKPOINT");
+ if ($primary->log_contains(
+ 'invalidating obsolete replication slot "sb1_slot"', $logstart))
+ {
+ $invalidated = 1;
+ last;
+ }
+ usleep(100_000);
+}
+ok($invalidated, 'check that slot sb1_slot invalidation has been logged');
+
+# 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', inactive_timeout = $inactive_timeout)"
+);
+
+$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");
+
+$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";
+
+$invalidated = 0;
+for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++)
+{
+ $publisher->safe_psql('postgres', "CHECKPOINT");
+ if ($publisher->log_contains(
+ 'invalidating obsolete replication slot "lsub1_slot"', $logstart))
+ {
+ $invalidated = 1;
+ last;
+ }
+ usleep(100_000);
+}
+ok($invalidated, 'check that slot lsub1_slot invalidation has been logged');
+
+# Testcase end: Invalidate logical subscriber's slot due to inactive_timeout
+# =============================================================================
+
+done_testing();
--
2.34.1
view thread (18+ 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: <CALj2ACUvK7ShPyirAURWf62=qOQqu=NwnyL5CMfVjM5Ody7Oxw@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