v11-0003-Avoid-concurrent-XID-age-slot-invalidation-attem.patch

From 7e17543594722054a9310a15da9b250d476e436e Mon Sep 17 00:00:00 2001
From: Bharath Rupireddy <bharath.rupireddyforpostgres@gmail.com>
Date: Fri, 17 Apr 2026 18:31:35 +0000
Subject: [PATCH v11 3/3] Avoid concurrent XID-age slot invalidation attempts

Multiple processes (autovacuum workers, backends running VACUUM)
can concurrently attempt to invalidate the same replication slot
due to XID age, causing them to wait on the same ConditionVariable
while waiting for a slow walsender to release the slot.

Add an invalidating_proc field to ReplicationSlot to track which
process is currently attempting invalidation. When a process sees
that another live process is already working on a slot, it skips
the slot and defers to a subsequent cycle. This prevents
unnecessary blocking during XID-age based slot invalidation.

Added an injection point in the walsender to mimic slow SIGTERM
processing and a TAP test for the concurrent slot invalidation.
---
 src/backend/replication/slot.c            |  75 ++++++++++
 src/backend/tcop/postgres.c               |  11 ++
 src/include/replication/slot.h            |   7 +
 src/test/recovery/t/019_replslot_limit.pl | 161 ++++++++++++++++++++++
 4 files changed, 254 insertions(+)

diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index fe43f5c8820..7cef325a888 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -233,6 +233,7 @@ ReplicationSlotsShmemInit(void *arg)
 
 		/* everything else is zeroed by the memset above */
 		slot->active_proc = INVALID_PROC_NUMBER;
+		slot->invalidating_proc = INVALID_PROC_NUMBER;
 		SpinLockInit(&slot->mutex);
 		LWLockInitialize(&slot->io_in_progress_lock,
 						 LWTRANCHE_REPLICATION_SLOT_IO);
@@ -501,6 +502,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
 	slot->last_saved_restart_lsn = InvalidXLogRecPtr;
 	slot->inactive_since = 0;
 	slot->slotsync_skip_reason = SS_SKIP_NONE;
+	slot->invalidating_proc = INVALID_PROC_NUMBER;
 
 	/*
 	 * Create the slot on disk.  We haven't actually marked the slot allocated
@@ -2052,6 +2054,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
 	int			last_signaled_pid = 0;
 	bool		released_lock = false;
 	bool		invalidated = false;
+	bool		am_invalidating = false;
 	TimestampTz inactive_since = 0;
 
 	for (;;)
@@ -2112,6 +2115,59 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
 			break;
 		}
 
+		/*
+		 * Skip XID-age invalidation if another process is already
+		 * invalidating this slot.
+		 *
+		 * Check if another process is already trying to invalidate this slot.
+		 * If so, skip it to avoid multiple processes blocking on the same CV
+		 * sleep. The first process will complete the invalidation attempt;
+		 * others defer to a subsequent cycle.
+		 *
+		 * We handle this only for XID-age invalidation because multiple
+		 * processes (autovacuum workers, backends running VACUUM, the
+		 * checkpointer) can attempt it concurrently, making it likely that
+		 * several end up blocking on the same ConditionVariable while waiting
+		 * for a slow walsender to release the slot. Invalidation due to other
+		 * causes can also involve multiple processes (e.g., on a standby, the
+		 * checkpointer and the startup process may attempt to invalidate a
+		 * slot for RS_INVAL_WAL_LEVEL and RS_INVAL_HORIZON respectively), but
+		 * such concurrent attempts are rare in practice.
+		 */
+		if (invalidation_cause == RS_INVAL_XID_AGE &&
+			s->invalidating_proc != INVALID_PROC_NUMBER)
+		{
+			int			invalidating_pid;
+
+			invalidating_pid = GetPGProcByNumber(s->invalidating_proc)->pid;
+
+			if (invalidating_pid != 0 &&
+				s->invalidating_proc != MyProcNumber)
+			{
+				/* Another live process is already invalidating this slot */
+				SpinLockRelease(&s->mutex);
+				if (released_lock)
+					LWLockRelease(ReplicationSlotControlLock);
+				break;
+			}
+
+			/*
+			 * The previously recorded process has exited (pid == 0) or it's
+			 * us. Reset and proceed with invalidation.
+			 */
+			s->invalidating_proc = INVALID_PROC_NUMBER;
+		}
+
+		/*
+		 * If the slot is active (we'll need to signal and wait), record
+		 * ourselves as the invalidating process.
+		 */
+		if (s->active_proc != INVALID_PROC_NUMBER)
+		{
+			s->invalidating_proc = MyProcNumber;
+			am_invalidating = true;
+		}
+
 		slotname = s->data.name;
 		active_proc = s->active_proc;
 
@@ -2131,6 +2187,12 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
 			s->active_proc = MyProcNumber;
 			s->data.invalidated = invalidation_cause;
 
+			/*
+			 * Clear the invalidating process since we have completed the
+			 * invalidation (acquired the slot and marked it invalid).
+			 */
+			s->invalidating_proc = INVALID_PROC_NUMBER;
+
 			/*
 			 * XXX: We should consider not overwriting restart_lsn and instead
 			 * just rely on .invalidated.
@@ -2255,6 +2317,19 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
 		}
 	}
 
+	/*
+	 * If we set invalidating_proc but exited without completing the
+	 * invalidation (e.g., the slot caught up while we were waiting, or the
+	 * slot was dropped), clear it so other processes don't skip this slot.
+	 */
+	if (am_invalidating && !invalidated)
+	{
+		SpinLockAcquire(&s->mutex);
+		if (s->invalidating_proc == MyProcNumber)
+			s->invalidating_proc = INVALID_PROC_NUMBER;
+		SpinLockRelease(&s->mutex);
+	}
+
 	Assert(released_lock == !LWLockHeldByMe(ReplicationSlotControlLock));
 
 	*released_lock_out = released_lock;
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 2c1f14b7889..846ee613243 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -3375,6 +3375,17 @@ ProcessInterrupts(void)
 		ProcDieSenderUid = 0;
 		QueryCancelPending = false; /* ProcDie trumps QueryCancel */
 		LockErrorCleanup();
+
+#ifdef USE_INJECTION_POINTS
+		/*
+		 * Injection point used to simulate a walsender that is slow to
+		 * respond to SIGTERM, allowing tests to verify concurrent slot
+		 * invalidation behavior.
+		 */
+		if (am_walsender)
+			INJECTION_POINT("walsender-before-sigterm-exit", NULL);
+#endif
+
 		/* As in quickdie, don't risk sending to client during auth */
 		if (ClientAuthInProgress && whereToSendOutput == DestRemote)
 			whereToSendOutput = DestNone;
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index cad1d89b05b..fb085447d23 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -284,6 +284,13 @@ typedef struct ReplicationSlot
 	 * slotsync_skip_reason provides no practical benefit.
 	 */
 	SlotSyncSkipReason slotsync_skip_reason;
+
+	/*
+	 * Process currently attempting to invalidate this slot.
+	 * INVALID_PROC_NUMBER means no invalidation is in progress. Protected by
+	 * the slot's mutex.
+	 */
+	ProcNumber	invalidating_proc;
 } ReplicationSlot;
 
 #define SlotIsPhysical(slot) ((slot)->data.database == InvalidOid)
diff --git a/src/test/recovery/t/019_replslot_limit.pl b/src/test/recovery/t/019_replslot_limit.pl
index 6f92034f3d5..0897f4388d7 100644
--- a/src/test/recovery/t/019_replslot_limit.pl
+++ b/src/test/recovery/t/019_replslot_limit.pl
@@ -845,4 +845,165 @@ $primary6->stop;
 # Testcase end: XID-age-based slot invalidation with autovacuum (production-like)
 # ================================================================================
 
+# ===============================================================================
+# Testcase start: Concurrent slot invalidation due to max_slot_xid_age GUC
+#
+# Two concurrent VACUUMs both try to invalidate the same active logical slot.
+# An injection point delays the walsender's SIGTERM processing so that vacuum1
+# blocks on the CV waiting for the slot to be released.  When vacuum2 runs, it
+# sees that vacuum1 is already invalidating the same slot and skips without
+# blocking. After the walsender is woken, vacuum1 completes the invalidation.
+
+# Skip if injection points are not available.
+if ($ENV{enable_injection_points} ne 'yes')
+{
+	done_testing();
+	exit;
+}
+
+my $primary7 = PostgreSQL::Test::Cluster->new('primary7');
+$primary7->init(allows_streaming => 'logical');
+
+my $max_slot_xid_age7 = 100;
+$primary7->append_conf(
+	'postgresql.conf', qq{
+max_slot_xid_age = $max_slot_xid_age7
+autovacuum = off
+shared_preload_libraries = 'injection_points'
+});
+
+$primary7->start;
+
+# Check if injection_points extension is available.
+if (!$primary7->check_extension('injection_points'))
+{
+	$primary7->stop;
+	done_testing();
+	exit;
+}
+
+$primary7->safe_psql('postgres', 'CREATE EXTENSION injection_points');
+
+# Helper to consume XIDs.
+$primary7->safe_psql(
+	'postgres', qq{
+	CREATE PROCEDURE consume_xid(cnt int)
+	AS \$\$
+	DECLARE
+	    i int;
+	BEGIN
+	    FOR i IN 1..cnt LOOP
+	        EXECUTE 'SELECT pg_current_xact_id()';
+	        COMMIT;
+	    END LOOP;
+	END;
+	\$\$ LANGUAGE plpgsql;
+});
+
+
+# Create a logical slot (gets catalog_xmin immediately).
+$primary7->safe_psql('postgres',
+	"SELECT pg_create_logical_replication_slot('lslot7', 'test_decoding')");
+
+# Hold the slot active via pg_recvlogical.
+my $pg_recvlog_stdout7 = '';
+my $pg_recvlog_stderr7 = '';
+my $connstr7 = $primary7->connstr('postgres');
+my $pg_recvlogical_handle7 = IPC::Run::start(
+	[
+		'pg_recvlogical', '-d', $connstr7,
+		'--slot', 'lslot7', '--start',
+		'-f', '/dev/null', '--no-loop'
+	],
+	'>', \$pg_recvlog_stdout7,
+	'2>', \$pg_recvlog_stderr7);
+
+# Wait for the slot to become active.
+$primary7->poll_query_until(
+	'postgres', qq[
+	SELECT active FROM pg_replication_slots WHERE slot_name = 'lslot7';
+]) or die "Timed out waiting for slot lslot7 to become active";
+
+# Make the walsender block before processing SIGTERM.
+$primary7->safe_psql('postgres',
+	"SELECT injection_points_attach('walsender-before-sigterm-exit', 'wait')");
+
+# Make the slot's catalog_xmin stale.
+$primary7->safe_psql('postgres',
+	qq{CALL consume_xid(2 * $max_slot_xid_age7)});
+
+# Launch vacuum1 on a catalog table: the logical slot holds catalog_xmin,
+# so only catalog VACUUMs see it as OldestXmin and trigger invalidation.
+# vacuum1 will SIGTERM the walsender, then block on the CV.
+my $vacuum1 = $primary7->background_psql('postgres');
+my $vacuum2 = $primary7->background_psql('postgres');
+
+$vacuum1->query_until(
+	qr/starting_vacuum/,
+	q(\echo starting_vacuum
+VACUUM pg_class;
+\echo vacuum1_done
+));
+
+# Wait for the walsender to hit the injection point.
+$primary7->wait_for_event('walsender', 'walsender-before-sigterm-exit');
+
+# Verify vacuum1 is blocked on the CV.
+$primary7->poll_query_until(
+	'postgres', qq[
+	SELECT count(*) = 1 FROM pg_stat_activity
+		WHERE wait_event = 'ReplicationSlotDrop'
+		AND backend_type = 'client backend';
+]) or die "Timed out waiting for vacuum1 to block on slot CV";
+
+# Launch vacuum2 on a different catalog table: it also computes the catalog
+# horizon and sees the slot needs invalidation, but finds vacuum1 is already
+# invalidating the same slot (via invalidating_proc) and skips.
+$vacuum2->query_until(
+	qr/starting_vacuum/,
+	q(\echo starting_vacuum
+VACUUM pg_type;
+\echo vacuum2_done
+));
+
+# vacuum2 completes without blocking.
+$vacuum2->query_until(qr/vacuum2_done/, '');
+
+# Verify only vacuum1 is waiting on ReplicationSlotDrop.
+$result = $primary7->safe_psql(
+	'postgres', qq[
+	SELECT count(*) FROM pg_stat_activity
+		WHERE wait_event = 'ReplicationSlotDrop'
+		AND backend_type = 'client backend';
+]);
+is($result, '1',
+	'only vacuum1 blocks on CV; vacuum2 skips via invalidating_proc');
+
+# Wake up the walsender so it can exit and release the slot.
+$primary7->safe_psql('postgres',
+	"SELECT injection_points_wakeup('walsender-before-sigterm-exit')");
+
+# Detach the injection point.
+$primary7->safe_psql('postgres',
+	"SELECT injection_points_detach('walsender-before-sigterm-exit')");
+
+# Wait for pg_recvlogical to exit.
+$pg_recvlogical_handle7->finish;
+
+# Wait for vacuum1 to complete now that the walsender has released the slot.
+$vacuum1->query_until(qr/vacuum1_done/, '');
+
+# Verify the slot was invalidated.
+wait_for_xid_aged_invalidation($primary7, 'lslot7');
+ok(1,
+	"concurrent VACUUM: vacuum1 blocks on CV, vacuum2 skips via invalidating_proc"
+);
+
+$vacuum1->quit;
+$vacuum2->quit;
+$primary7->stop;
+
+# Testcase end: Concurrent slot invalidation due to max_slot_xid_age GUC
+# ===============================================================================
+
 done_testing();
-- 
2.47.3