v5-0004-Add-inactive_timeout-based-replication-slot-inval.patch

From 598e036d72f47375b1071c01b2de546dea5c1681 Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
Date: Tue, 20 Feb 2024 05:45:54 +0000
Subject: [PATCH v5 4/4] Add inactive_timeout based replication slot
 invalidation

Currently postgres has the ability to invalidate inactive
replication slots based on the amount of WAL (set via
max_slot_wal_keep_size GUC) that will be needed for the slots in
case they become active. However, choosing a default value for
max_slot_wal_keep_size is tricky. Because the amount of WAL a
customer generates, and their allocated storage will vary greatly
in production, making it difficult to pin down a one-size-fits-all
value. It is often easy for developers to set a timeout of say 1
or 2 or 3 days, after which the inactive slots get dropped.

To achieve the above, postgres uses replication slot metric
inactive_at (the time at which the slot became inactive), and a
new GUC inactive_replication_slot_timeout. The checkpointer then
looks at all replication slots invalidating the inactive slots
based on the timeout set.
---
 doc/src/sgml/config.sgml                      | 18 +++++
 src/backend/access/transam/xlog.c             | 10 +++
 src/backend/replication/slot.c                | 21 +++++
 src/backend/replication/slotfuncs.c           |  8 ++
 src/backend/utils/misc/guc_tables.c           | 12 +++
 src/backend/utils/misc/postgresql.conf.sample |  1 +
 src/include/replication/slot.h                |  4 +
 src/test/recovery/meson.build                 |  1 +
 src/test/recovery/t/050_invalidate_slots.pl   | 79 +++++++++++++++++++
 9 files changed, 154 insertions(+)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 6c1c5f421f..4904541607 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4426,6 +4426,24 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"'  # Windows
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-inactive-replication-slot-timeout" xreflabel="inactive_replication_slot_timeout">
+      <term><varname>inactive_replication_slot_timeout</varname> (<type>integer</type>)
+      <indexterm>
+       <primary><varname>inactive_replication_slot_timeout</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Invalidate replication slots that are inactive for longer than this
+        amount of time at the next checkpoint. If this value is specified
+        without units, it is taken as seconds. A value of zero (which is
+        default) disables the timeout mechanism. This parameter can only be
+        set in the <filename>postgresql.conf</filename> file or on the server
+        command line.
+       </para>
+      </listitem>
+     </varlistentry>
+
      <varlistentry id="guc-track-commit-timestamp" xreflabel="track_commit_timestamp">
       <term><varname>track_commit_timestamp</varname> (<type>boolean</type>)
       <indexterm>
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 5fe76c46a1..ad8786c9f7 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -7164,6 +7164,11 @@ CreateCheckPoint(int flags)
 		InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0,
 										   InvalidOid, InvalidTransactionId);
 
+	/* Invalidate inactive replication slots based on timeout */
+	if (inactive_replication_slot_timeout > 0)
+		InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT, 0,
+										   InvalidOid, InvalidTransactionId);
+
 	/*
 	 * Delete old log files, those no longer needed for last checkpoint to
 	 * prevent the disk holding the xlog from growing full.
@@ -7613,6 +7618,11 @@ CreateRestartPoint(int flags)
 		InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0,
 										   InvalidOid, InvalidTransactionId);
 
+	/* Invalidate inactive replication slots based on timeout */
+	if (inactive_replication_slot_timeout > 0)
+		InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT, 0,
+										   InvalidOid, InvalidTransactionId);
+
 	/*
 	 * Retreat _logSegNo using the current end of xlog replayed or received,
 	 * whichever is later.
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index ce51c6d909..aef027e4f6 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -103,6 +103,7 @@ ReplicationSlot *MyReplicationSlot = NULL;
 int			max_replication_slots = 10; /* the maximum number of replication
 										 * slots */
 int			max_slot_xid_age = 0;
+int			inactive_replication_slot_timeout = 0;
 
 static void ReplicationSlotShmemExit(int code, Datum arg);
 static void ReplicationSlotDropPtr(ReplicationSlot *slot);
@@ -1445,6 +1446,9 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause,
 		case RS_INVAL_XID_AGE:
 			appendStringInfoString(&err_detail, _("The replication slot's xmin or catalog_xmin reached the age specified by max_slot_xid_age."));
 			break;
+		case RS_INVAL_INACTIVE_TIMEOUT:
+			appendStringInfoString(&err_detail, _("The slot has been inactive for more than the time specified by inactive_replication_slot_timeout."));
+			break;
 		case RS_INVAL_NONE:
 			pg_unreachable();
 	}
@@ -1597,6 +1601,20 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
 						}
 					}
 					break;
+				case RS_INVAL_INACTIVE_TIMEOUT:
+					if (s->data.last_inactive_at > 0)
+					{
+						TimestampTz now;
+
+						Assert(s->data.persistency == RS_PERSISTENT);
+						Assert(s->active_pid == 0);
+
+						now = GetCurrentTimestamp();
+						if (TimestampDifferenceExceeds(s->data.last_inactive_at, now,
+													   inactive_replication_slot_timeout * 1000))
+							conflict = cause;
+					}
+					break;
 				case RS_INVAL_NONE:
 					pg_unreachable();
 			}
@@ -1752,6 +1770,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause,
  *   db; dboid may be InvalidOid for shared relations
  * - RS_INVAL_WAL_LEVEL: is logical
  * - RS_INVAL_XID_AGE: slot's xmin or catalog_xmin has reached the age
+ * - RS_INVAL_INACTIVE_TIMEOUT: inactive slot timeout occurs
  *
  * NB - this runs as part of checkpoint, so avoid raising errors if possible.
  */
@@ -2372,6 +2391,8 @@ GetSlotInvalidationCause(char *conflict_reason)
 		return RS_INVAL_WAL_LEVEL;
 	else if (strcmp(conflict_reason, SLOT_INVAL_XID_AGED_TEXT) == 0)
 		return RS_INVAL_XID_AGE;
+	else if (strcmp(conflict_reason, SLOT_INVAL_INACTIVE_TIMEOUT) == 0)
+		return RS_INVAL_INACTIVE_TIMEOUT;
 	else
 		Assert(0);
 
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 6a12db27fd..b5f077f9f3 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -435,6 +435,10 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 				case RS_INVAL_XID_AGE:
 					values[i++] = CStringGetTextDatum(SLOT_INVAL_XID_AGED_TEXT);
 					break;
+
+				case RS_INVAL_INACTIVE_TIMEOUT:
+					values[i++] = CStringGetTextDatum(SLOT_INVAL_INACTIVE_TIMEOUT);
+					break;
 			}
 		}
 
@@ -463,6 +467,10 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
 			case RS_INVAL_XID_AGE:
 				values[i++] = CStringGetTextDatum(SLOT_INVAL_XID_AGED_TEXT);
 				break;
+
+			case RS_INVAL_INACTIVE_TIMEOUT:
+				values[i++] = CStringGetTextDatum(SLOT_INVAL_INACTIVE_TIMEOUT);
+				break;
 		}
 
 		if (slot_contents.data.last_inactive_at > 0)
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index 298bb8ea85..28780b8c87 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -2923,6 +2923,18 @@ struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"inactive_replication_slot_timeout", PGC_SIGHUP, REPLICATION_SENDING,
+			gettext_noop("Sets the amount of time to wait before invalidating an "
+						 "inactive replication slot."),
+			NULL,
+			GUC_UNIT_S
+		},
+		&inactive_replication_slot_timeout,
+		0, 0, INT_MAX,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"commit_delay", PGC_SUSET, WAL_SETTINGS,
 			gettext_noop("Sets the delay in microseconds between transaction commit and "
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 96fe198c23..ccb79e5a67 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -252,6 +252,7 @@
 #recovery_prefetch = try	# prefetch pages referenced in the WAL?
 #wal_decode_buffer_size = 512kB	# lookahead window used for prefetching
 				# (change requires restart)
+#inactive_replication_slot_timeout = 0	# in seconds; 0 disables
 
 # - Archiving -
 
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 87a56aa28a..c5174f7c8d 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -52,6 +52,8 @@ typedef enum ReplicationSlotInvalidationCause
 	RS_INVAL_WAL_LEVEL,
 	/* slot's xmin or catalog_xmin has reached the age */
 	RS_INVAL_XID_AGE,
+	/* inactive slot timeout has occurred */
+	RS_INVAL_INACTIVE_TIMEOUT,
 } ReplicationSlotInvalidationCause;
 
 /*
@@ -62,6 +64,7 @@ typedef enum ReplicationSlotInvalidationCause
 #define SLOT_INVAL_HORIZON_TEXT     "rows_removed"
 #define SLOT_INVAL_WAL_LEVEL_TEXT   "wal_level_insufficient"
 #define SLOT_INVAL_XID_AGED_TEXT	"xid_aged"
+#define SLOT_INVAL_INACTIVE_TIMEOUT	"inactive_timeout"
 
 /*
  * On-Disk data of a replication slot, preserved across restarts.
@@ -239,6 +242,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot;
 /* GUCs */
 extern PGDLLIMPORT int max_replication_slots;
 extern PGDLLIMPORT int max_slot_xid_age;
+extern PGDLLIMPORT int inactive_replication_slot_timeout;
 
 /* shmem initialization functions */
 extern Size ReplicationSlotsShmemSize(void);
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index bf087ac2a9..e07b941d73 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -46,6 +46,7 @@ tests += {
       't/038_save_logical_slots_shutdown.pl',
       't/039_end_of_wal.pl',
       't/040_standby_failover_slots_sync.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
index 2f482b56e8..4c66dd4a4e 100644
--- a/src/test/recovery/t/050_invalidate_slots.pl
+++ b/src/test/recovery/t/050_invalidate_slots.pl
@@ -105,4 +105,83 @@ $primary->poll_query_until(
   or die
   "Timed out while waiting for replication slot sb1_slot to be invalidated";
 
+$primary->safe_psql(
+	'postgres', qq[
+    SELECT pg_create_physical_replication_slot('sb2_slot');
+]);
+
+$primary->safe_psql(
+	'postgres', qq[
+    ALTER SYSTEM SET max_slot_xid_age = 0;
+]);
+$primary->reload;
+
+# Create a standby linking to the primary using the replication slot
+my $standby2 = PostgreSQL::Test::Cluster->new('standby2');
+$standby2->init_from_backup($primary, $backup_name, has_streaming => 1);
+$standby2->append_conf(
+	'postgresql.conf', q{
+primary_slot_name = 'sb2_slot'
+});
+$standby2->start;
+
+# Wait until standby has replayed enough data
+$primary->wait_for_catchup($standby2);
+
+# 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_count = 0 AS OK
+		FROM pg_replication_slots WHERE slot_name = 'sb2_slot';
+]);
+is($result, "t|t",
+	'check the inactive replication slot info for an active slot');
+
+# Set timeout so that the next checkpoint will invalidate the inactive
+# replication slot.
+$primary->safe_psql(
+	'postgres', qq[
+    ALTER SYSTEM SET inactive_replication_slot_timeout TO '1s';
+]);
+$primary->reload;
+
+$logstart = -s $primary->logfile;
+
+# Stop standby to make the replication slot on primary inactive
+$standby2->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
+		inactive_count = 1 AND slot_name = 'sb2_slot';
+])
+  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++)
+{
+	$primary->safe_psql('postgres', "CHECKPOINT");
+	if ($primary->log_contains(
+			'invalidating obsolete replication slot "sb2_slot"', $logstart))
+	{
+		$invalidated = 1;
+		last;
+	}
+	usleep(100_000);
+}
+ok($invalidated, 'check that slot sb2_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 = 'sb2_slot' AND
+		invalidation_reason = 'inactive_timeout';
+])
+  or die
+  "Timed out while waiting for inactive replication slot sb2_slot to be invalidated";
+
 done_testing();
-- 
2.34.1