agora inbox for pgsql-hackers@postgresql.org
help / color / mirror / Atom feedFrom: Bertrand Drouvot <bertranddrouvot.pg@gmail.com>
Subject: [PATCH v3 1/2] Persist slot invalidations before publishing them
Date: Wed, 26 Aug 2026 05:34:51 +0000
InvalidatePossiblyObsoleteSlot() marks an inactive replication slot invalid
in shared memory before saving it. If the save fails, or the server crashes
before it completes, startup can restore a valid slot after resources required
by that slot have been removed.
Add ReplicationSlotPersistInvalidation(), which writes and fsyncs an invalidated
copy while the shared slot remains valid. Hold io_in_progress_lock until the
invalidation is published so checkpoints and other slot savers cannot persist
a stale image after publication.
A write failure now leaves both the shared slot and its disk image valid.
Ensure that errors also release ownership claimed for inactive slots while
preserving inactive_since.
Also serialize concurrent internal invalidators with the slot's I/O lock. This
makes a second invalidator wait instead of treating the first as a regular slot
user and terminating it.
Add an injection-point test covering save failure, subsequent checkpointing,
and immediate restart. This changes neither the on disk slot format nor the
ReplicationSlot shared memory layout.
Author: Bertrand Drouvot <bertranddrouvot.pg@gmail.com>
Reviewed-by: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Reviewed-by: Miłosz Bieniek <bieniek.milosz@proton.me>
Reviewed-by: JoongHyuk Shin <sjh910805@gmail.com>
Reviewed-by: Rui Zhao <zhaorui126@gmail.com>
Discussion: https://postgr.es/m/ao7u5I9OeIR72kGp%40bdtpg
Backpatch-through: 14
---
src/backend/replication/slot.c | 234 +++++++++++----
src/include/replication/slot.h | 2 +
src/test/recovery/meson.build | 2 +
.../t/057_replslot_invalidation_durability.pl | 125 ++++++++
.../t/058_slot_invalidation_concurrency.pl | 273 ++++++++++++++++++
5 files changed, 587 insertions(+), 49 deletions(-)
45.2% src/backend/replication/
53.5% src/test/recovery/t/
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 63ce6d27885..8f2b681b626 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -185,13 +185,17 @@ static SyncStandbySlotsConfigData *synchronized_standby_slots_config;
static XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr;
static void ReplicationSlotShmemExit(int code, Datum arg);
+static void ReplicationSlotReleaseInternal(bool update_inactive_since);
+static void ReplicationSlotInvalidationErrorCleanup(int code, Datum arg);
static bool IsSlotForConflictCheck(const char *name);
static void ReplicationSlotDropPtr(ReplicationSlot *slot);
/* internal persistency functions */
static void RestoreSlotFromDisk(const char *name);
static void CreateSlotOnDisk(ReplicationSlot *slot);
-static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel);
+static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel,
+ ReplicationSlotInvalidationCause invalidation_cause,
+ bool clear_restart_lsn);
/*
* Register shared memory space needed for replication slots.
@@ -769,6 +773,15 @@ retry:
*/
void
ReplicationSlotRelease(void)
+{
+ ReplicationSlotReleaseInternal(true);
+}
+
+/*
+ * Release the replication slot, optionally preserving inactive_since.
+ */
+static void
+ReplicationSlotReleaseInternal(bool update_inactive_since)
{
ReplicationSlot *slot = MyReplicationSlot;
char *slotname = NULL; /* keep compiler quiet */
@@ -776,6 +789,7 @@ ReplicationSlotRelease(void)
TimestampTz now = 0;
Assert(slot != NULL && slot->active_proc != INVALID_PROC_NUMBER);
+ Assert(update_inactive_since || slot->data.persistency == RS_PERSISTENT);
is_logical = SlotIsLogical(slot);
@@ -808,10 +822,12 @@ ReplicationSlotRelease(void)
}
/*
- * Set the time since the slot has become inactive. We get the current
- * time beforehand to avoid system call while holding the spinlock.
+ * Set the time since the slot has become inactive, unless the caller
+ * needs to preserve it. Get the current time beforehand to avoid a
+ * system call while holding the spinlock.
*/
- now = GetCurrentTimestamp();
+ if (update_inactive_since)
+ now = GetCurrentTimestamp();
if (slot->data.persistency == RS_PERSISTENT)
{
@@ -821,11 +837,12 @@ ReplicationSlotRelease(void)
*/
SpinLockAcquire(&slot->mutex);
slot->active_proc = INVALID_PROC_NUMBER;
- ReplicationSlotSetInactiveSince(slot, now, false);
+ if (update_inactive_since)
+ ReplicationSlotSetInactiveSince(slot, now, false);
SpinLockRelease(&slot->mutex);
ConditionVariableBroadcast(&slot->active_cv);
}
- else
+ else if (update_inactive_since)
ReplicationSlotSetInactiveSince(slot, now, true);
MyReplicationSlot = NULL;
@@ -850,6 +867,22 @@ ReplicationSlotRelease(void)
}
}
+/*
+ * Roll back an internal invalidation after an error.
+ *
+ * On ERROR, release slot ownership before normal error cleanup calls
+ * LWLockReleaseAll() to release the I/O lock. During process exit,
+ * shmem_exit() has already released LWLocks before invoking this callback.
+ */
+static void
+ReplicationSlotInvalidationErrorCleanup(int code, Datum arg)
+{
+ ReplicationSlot *slot = (ReplicationSlot *) DatumGetPointer(arg);
+
+ if (MyReplicationSlot == slot)
+ ReplicationSlotReleaseInternal(false);
+}
+
/*
* Cleanup temporary slots created in current session.
*
@@ -1168,7 +1201,32 @@ ReplicationSlotSave(void)
Assert(MyReplicationSlot != NULL);
sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(MyReplicationSlot->data.name));
- SaveSlotToPath(MyReplicationSlot, path, ERROR);
+ SaveSlotToPath(MyReplicationSlot, path, ERROR, RS_INVAL_NONE, false);
+}
+
+/*
+ * Persist an invalidated image of the acquired slot before publishing the
+ * invalidation in shared memory. The caller must own the slot and hold its
+ * I/O lock. The lock is released on success and left held for error cleanup
+ * otherwise.
+ */
+void
+ReplicationSlotPersistInvalidation(ReplicationSlotInvalidationCause cause,
+ bool clear_restart_lsn)
+{
+ char path[MAXPGPATH];
+ ReplicationSlot *slot = MyReplicationSlot;
+
+ Assert(slot != NULL);
+ Assert(slot->data.persistency == RS_PERSISTENT);
+ Assert(slot->data.invalidated == RS_INVAL_NONE);
+ Assert(cause != RS_INVAL_NONE);
+ Assert(!clear_restart_lsn || cause == RS_INVAL_WAL_REMOVED);
+ Assert(LWLockHeldByMeInMode(&slot->io_in_progress_lock, LW_EXCLUSIVE));
+
+ sprintf(path, "%s/%s", PG_REPLSLOT_DIR, NameStr(slot->data.name));
+
+ SaveSlotToPath(slot, path, ERROR, cause, clear_restart_lsn);
}
/*
@@ -2005,6 +2063,49 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
break;
}
+ /*
+ * Serializing on the slot's I/O lock ensures that an internal
+ * invalidator cannot be mistaken for a process using the slot. Avoid
+ * waiting for the lock while holding ReplicationSlotControlLock.
+ */
+ if (!LWLockConditionalAcquire(&s->io_in_progress_lock, LW_EXCLUSIVE))
+ {
+ /*
+ * Avoid waiting for an unrelated slot save. The check after
+ * acquiring the lock remains authoritative.
+ */
+ if (possible_causes & RS_INVAL_IDLE_TIMEOUT)
+ now = GetCurrentTimestamp();
+
+ SpinLockAcquire(&s->mutex);
+
+ if (s->data.invalidated == RS_INVAL_NONE)
+ invalidation_cause = DetermineSlotInvalidationCause(possible_causes,
+ s, oldestLSN,
+ dboid,
+ snapshotConflictHorizon,
+ &inactive_since, now);
+
+ SpinLockRelease(&s->mutex);
+
+ if (invalidation_cause == RS_INVAL_NONE)
+ {
+ if (released_lock)
+ LWLockRelease(ReplicationSlotControlLock);
+
+ break;
+ }
+
+ LWLockRelease(ReplicationSlotControlLock);
+ released_lock = true;
+
+ if (LWLockAcquireOrWait(&s->io_in_progress_lock, LW_EXCLUSIVE))
+ LWLockRelease(&s->io_in_progress_lock);
+
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+ continue;
+ }
+
if (possible_causes & RS_INVAL_IDLE_TIMEOUT)
{
/*
@@ -2016,10 +2117,9 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
/*
* Check if the slot needs to be invalidated. If it needs to be
- * invalidated, and is not currently acquired, acquire it and mark it
- * as having been invalidated. 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.
+ * invalidated and is not currently acquired, acquire it. We do this
+ * with the spinlock held to avoid races where restart_lsn moves
+ * forward or the slot is dropped.
*/
SpinLockAcquire(&s->mutex);
@@ -2038,6 +2138,7 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
if (invalidation_cause == RS_INVAL_NONE)
{
SpinLockRelease(&s->mutex);
+ LWLockRelease(&s->io_in_progress_lock);
if (released_lock)
LWLockRelease(ReplicationSlotControlLock);
break;
@@ -2047,9 +2148,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
active_proc = s->active_proc;
/*
- * If the slot can be acquired, do so and mark it invalidated
- * immediately. Otherwise we'll signal the owning process, below, and
- * retry.
+ * If the slot can be acquired, do so. Otherwise we'll signal the
+ * owning process, below, and retry.
*
* Note: Unlike other slot attributes, slot's inactive_since can't be
* changed until the acquired slot is released or the owning process
@@ -2058,22 +2158,9 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
*/
if (active_proc == INVALID_PROC_NUMBER)
{
+ Assert(s->data.persistency == RS_PERSISTENT);
MyReplicationSlot = s;
s->active_proc = MyProcNumber;
- s->data.invalidated = invalidation_cause;
-
- /*
- * XXX: We should consider not overwriting restart_lsn and instead
- * just rely on .invalidated.
- */
- if (invalidation_cause == RS_INVAL_WAL_REMOVED)
- {
- s->data.restart_lsn = InvalidXLogRecPtr;
- s->last_saved_restart_lsn = InvalidXLogRecPtr;
- }
-
- /* Let caller know */
- invalidated = true;
}
else
{
@@ -2099,11 +2186,11 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
{
/*
* Prepare the sleep on the slot's condition variable before
- * releasing the lock, to close a possible race condition if the
- * slot is released before the sleep below.
+ * releasing either lock.
*/
ConditionVariablePrepareToSleep(&s->active_cv);
+ LWLockRelease(&s->io_in_progress_lock);
LWLockRelease(ReplicationSlotControlLock);
released_lock = true;
@@ -2159,8 +2246,8 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
else
{
/*
- * We hold the slot now and have already invalidated it; flush it
- * to ensure that state persists.
+ * We hold the slot now. Persist its invalidation before
+ * publishing it in shared memory.
*
* Don't want to hold ReplicationSlotControlLock across file
* system operations, so release it now but be sure to tell caller
@@ -2169,9 +2256,17 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
LWLockRelease(ReplicationSlotControlLock);
released_lock = true;
- /* Make sure the invalidated state persists across server restart */
- ReplicationSlotMarkDirty();
- ReplicationSlotSave();
+ PG_ENSURE_ERROR_CLEANUP(ReplicationSlotInvalidationErrorCleanup,
+ PointerGetDatum(s));
+ {
+ ReplicationSlotPersistInvalidation(invalidation_cause,
+ invalidation_cause == RS_INVAL_WAL_REMOVED);
+ }
+ PG_END_ENSURE_ERROR_CLEANUP(ReplicationSlotInvalidationErrorCleanup,
+ PointerGetDatum(s));
+
+ /* Let caller know */
+ invalidated = true;
ReplicationSlotRelease();
ReportSlotInvalidation(invalidation_cause, false, active_pid,
@@ -2380,7 +2475,7 @@ CheckPointReplicationSlots(bool is_shutdown)
if (s->last_saved_restart_lsn != s->data.restart_lsn)
last_saved_restart_lsn_updated = true;
- SaveSlotToPath(s, path, LOG);
+ SaveSlotToPath(s, path, LOG, RS_INVAL_NONE, false);
}
LWLockRelease(ReplicationSlotAllocationLock);
@@ -2493,7 +2588,7 @@ CreateSlotOnDisk(ReplicationSlot *slot)
/* Write the actual state file. */
slot->dirty = true; /* signal that we really need to write */
- SaveSlotToPath(slot, tmppath, ERROR);
+ SaveSlotToPath(slot, tmppath, ERROR, RS_INVAL_NONE, false);
/* Rename the directory into place. */
if (rename(tmppath, path) != 0)
@@ -2517,9 +2612,15 @@ CreateSlotOnDisk(ReplicationSlot *slot)
/*
* Shared functionality between saving and creating a replication slot.
+ *
+ * When invalidation_cause is set, the caller has already acquired the slot's
+ * I/O lock. On error, leave the lock held for error cleanup. The lock is
+ * released after successful shared-memory publication.
*/
static void
-SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
+SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel,
+ ReplicationSlotInvalidationCause invalidation_cause,
+ bool clear_restart_lsn)
{
char tmppath[MAXPGPATH];
char path[MAXPGPATH];
@@ -2527,6 +2628,9 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
ReplicationSlotOnDisk cp;
bool was_dirty;
+ Assert(!clear_restart_lsn || invalidation_cause == RS_INVAL_WAL_REMOVED);
+ Assert(invalidation_cause == RS_INVAL_NONE || elevel >= ERROR);
+
/* first check whether there's something to write out */
SpinLockAcquire(&slot->mutex);
was_dirty = slot->dirty;
@@ -2534,10 +2638,16 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
SpinLockRelease(&slot->mutex);
/* and don't do anything if there's nothing to write */
- if (!was_dirty)
+ if (!was_dirty && invalidation_cause == RS_INVAL_NONE)
return;
- LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE);
+ if (invalidation_cause != RS_INVAL_NONE)
+ Assert(LWLockHeldByMeInMode(&slot->io_in_progress_lock,
+ LW_EXCLUSIVE));
+ else
+ LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE);
+
+ INJECTION_POINT("replication-slot-save-error", NameStr(slot->data.name));
/* silence valgrind :( */
memset(&cp, 0, sizeof(ReplicationSlotOnDisk));
@@ -2549,14 +2659,14 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
if (fd < 0)
{
/*
- * If not an ERROR, then release the lock before returning. In case
- * of an ERROR, the error recovery path automatically releases the
- * lock, but no harm in explicitly releasing even in that case. Note
- * that LWLockRelease() could affect errno.
+ * Keep a caller-owned lock until its error cleanup has rolled back
+ * any associated shared-memory state. Note that LWLockRelease() could
+ * affect errno.
*/
int save_errno = errno;
- LWLockRelease(&slot->io_in_progress_lock);
+ if (invalidation_cause == RS_INVAL_NONE)
+ LWLockRelease(&slot->io_in_progress_lock);
errno = save_errno;
ereport(elevel,
(errcode_for_file_access(),
@@ -2576,6 +2686,20 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
SpinLockRelease(&slot->mutex);
+ if (invalidation_cause != RS_INVAL_NONE)
+ {
+ Assert(cp.slotdata.invalidated == RS_INVAL_NONE);
+
+ cp.slotdata.invalidated = invalidation_cause;
+
+ /*
+ * XXX: We should consider not overwriting restart_lsn and instead
+ * just rely on .invalidated.
+ */
+ if (clear_restart_lsn)
+ cp.slotdata.restart_lsn = InvalidXLogRecPtr;
+ }
+
COMP_CRC32C(cp.checksum,
(char *) (&cp) + ReplicationSlotOnDiskNotChecksummedSize,
ReplicationSlotOnDiskChecksummedSize);
@@ -2590,7 +2714,8 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
pgstat_report_wait_end();
CloseTransientFile(fd);
unlink(tmppath);
- LWLockRelease(&slot->io_in_progress_lock);
+ if (invalidation_cause == RS_INVAL_NONE)
+ LWLockRelease(&slot->io_in_progress_lock);
/* if write didn't set errno, assume problem is no disk space */
errno = save_errno ? save_errno : ENOSPC;
@@ -2611,7 +2736,8 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
pgstat_report_wait_end();
CloseTransientFile(fd);
unlink(tmppath);
- LWLockRelease(&slot->io_in_progress_lock);
+ if (invalidation_cause == RS_INVAL_NONE)
+ LWLockRelease(&slot->io_in_progress_lock);
errno = save_errno;
ereport(elevel,
@@ -2627,7 +2753,8 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
int save_errno = errno;
unlink(tmppath);
- LWLockRelease(&slot->io_in_progress_lock);
+ if (invalidation_cause == RS_INVAL_NONE)
+ LWLockRelease(&slot->io_in_progress_lock);
errno = save_errno;
ereport(elevel,
@@ -2643,7 +2770,8 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
int save_errno = errno;
unlink(tmppath);
- LWLockRelease(&slot->io_in_progress_lock);
+ if (invalidation_cause == RS_INVAL_NONE)
+ LWLockRelease(&slot->io_in_progress_lock);
errno = save_errno;
ereport(elevel,
@@ -2669,6 +2797,14 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
* already and remember the confirmed_flush LSN value.
*/
SpinLockAcquire(&slot->mutex);
+ if (invalidation_cause != RS_INVAL_NONE)
+ {
+ Assert(slot->data.invalidated == RS_INVAL_NONE);
+
+ slot->data.invalidated = invalidation_cause;
+ if (clear_restart_lsn)
+ slot->data.restart_lsn = InvalidXLogRecPtr;
+ }
if (!slot->just_dirtied)
slot->dirty = false;
slot->last_saved_confirmed_flush = cp.slotdata.confirmed_flush;
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 9b29444cbca..80d48020a87 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -344,6 +344,8 @@ extern void ReplicationSlotAcquire(const char *name, bool nowait,
extern void ReplicationSlotRelease(void);
extern void ReplicationSlotCleanup(bool synced_only);
extern void ReplicationSlotSave(void);
+extern void ReplicationSlotPersistInvalidation(ReplicationSlotInvalidationCause cause,
+ bool clear_restart_lsn);
extern void ReplicationSlotMarkDirty(void);
/* misc stuff */
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index 72113c5ac6e..e74b547a961 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -65,6 +65,8 @@ tests += {
't/054_unlogged_sequence_promotion.pl',
't/055_cascade_reconnect.pl',
't/056_standby_snapshot_export.pl',
+ 't/057_replslot_invalidation_durability.pl',
+ 't/058_slot_invalidation_concurrency.pl',
],
},
}
diff --git a/src/test/recovery/t/057_replslot_invalidation_durability.pl b/src/test/recovery/t/057_replslot_invalidation_durability.pl
new file mode 100644
index 00000000000..247d7b00dc1
--- /dev/null
+++ b/src/test/recovery/t/057_replslot_invalidation_durability.pl
@@ -0,0 +1,125 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+#
+# Test that replication slot invalidation is persisted before it is published.
+#
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+
+use Test::More;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+ plan skip_all => 'Injection points not supported by this build';
+}
+
+my $node = PostgreSQL::Test::Cluster->new('primary');
+$node->init(allows_streaming => 1, extra => ['--wal-segsize=1']);
+$node->append_conf(
+ 'postgresql.conf', qq(
+checkpoint_timeout = 1h
+min_wal_size = 2MB
+max_wal_size = 64MB
+wal_keep_size = 0
+max_slot_wal_keep_size = -1
+log_checkpoints = on
+));
+$node->start;
+
+if (!$node->check_extension('injection_points'))
+{
+ plan skip_all => 'Extension injection_points not installed';
+}
+
+$node->safe_psql('postgres', 'CREATE EXTENSION injection_points');
+$node->safe_psql('postgres',
+ q{SELECT pg_create_physical_replication_slot('target_slot', true)});
+$node->safe_psql('postgres', 'CHECKPOINT');
+
+my ($restart_lsn, $restart_segment) = split(
+ /\|/,
+ $node->safe_psql(
+ 'postgres',
+ q{
+SELECT restart_lsn, pg_walfile_name(restart_lsn)
+FROM pg_replication_slots
+WHERE slot_name = 'target_slot'
+}));
+my $restart_segment_path = $node->data_dir . "/pg_wal/$restart_segment";
+my $inactive_since = $node->safe_psql(
+ 'postgres',
+ q{
+SELECT inactive_since
+FROM pg_replication_slots
+WHERE slot_name = 'target_slot'
+});
+
+$node->append_conf('postgresql.conf', 'max_slot_wal_keep_size = 1MB');
+$node->reload;
+$node->advance_wal(8);
+
+my $current_segment = $node->safe_psql('postgres',
+ 'SELECT pg_walfile_name(pg_current_wal_lsn())');
+isnt($current_segment, $restart_segment,
+ 'target slot requires an older WAL segment');
+ok(-f $restart_segment_path,
+ "target slot WAL segment $restart_segment exists before invalidation");
+
+$node->safe_psql(
+ 'postgres', q{
+SELECT injection_points_attach(
+ 'replication-slot-save-error', 'error', 'target_slot')
+});
+
+my ($ret, $stdout, $stderr) = $node->psql('postgres', 'CHECKPOINT');
+like(
+ $stderr,
+ qr/checkpoint request failed/,
+ 'injected slot save error failed the checkpoint');
+
+$node->safe_psql('postgres',
+ q{SELECT injection_points_detach('replication-slot-save-error')});
+
+is( $node->safe_psql(
+ 'postgres',
+ qq{
+SELECT NOT active, invalidation_reason IS NULL,
+ restart_lsn = '$restart_lsn',
+ inactive_since = '$inactive_since'::timestamptz
+FROM pg_replication_slots
+WHERE slot_name = 'target_slot'
+}),
+ 't|t|t|t',
+ 'failed save leaves the valid slot unchanged');
+ok( -f $restart_segment_path,
+ "target slot WAL segment $restart_segment survives the failed checkpoint"
+);
+
+$node->append_conf('postgresql.conf', 'max_slot_wal_keep_size = -1');
+$node->reload;
+$node->safe_psql('postgres', 'CHECKPOINT');
+
+ok(-f $restart_segment_path,
+ "target slot WAL segment $restart_segment survives the next checkpoint");
+
+$node->stop('immediate');
+$node->start;
+
+is( $node->safe_psql(
+ 'postgres',
+ qq{
+SELECT NOT active, invalidation_reason IS NULL,
+ restart_lsn = '$restart_lsn'
+FROM pg_replication_slots
+WHERE slot_name = 'target_slot'
+}),
+ 't|t|t',
+ 'target slot restores with its original restart LSN');
+ok(-f $restart_segment_path,
+ "target slot WAL segment $restart_segment exists after restart");
+
+$node->stop;
+
+done_testing();
diff --git a/src/test/recovery/t/058_slot_invalidation_concurrency.pl b/src/test/recovery/t/058_slot_invalidation_concurrency.pl
new file mode 100644
index 00000000000..1259d7be425
--- /dev/null
+++ b/src/test/recovery/t/058_slot_invalidation_concurrency.pl
@@ -0,0 +1,273 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+#
+# Test concurrent invalidation of the same replication slot.
+#
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Time::HiRes qw(usleep);
+
+use Test::More;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+ plan skip_all => 'Injection points not supported by this build';
+}
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 1, extra => ['--wal-segsize=1']);
+$primary->append_conf(
+ 'postgresql.conf', qq(
+wal_level = logical
+autovacuum = off
+checkpoint_timeout = 1h
+max_wal_size = 64MB
+));
+$primary->start;
+
+if (!$primary->check_extension('injection_points'))
+{
+ plan skip_all => 'Extension injection_points not installed';
+}
+
+$primary->safe_psql('postgres', 'CREATE EXTENSION injection_points');
+$primary->safe_psql('postgres',
+ q{SELECT pg_create_physical_replication_slot('phys')});
+$primary->backup('backup');
+
+my $standby = PostgreSQL::Test::Cluster->new('standby');
+$standby->init_from_backup($primary, 'backup', has_streaming => 1);
+$standby->append_conf(
+ 'postgresql.conf', qq(
+primary_slot_name = 'phys'
+hot_standby_feedback = on
+checkpoint_timeout = 1h
+max_wal_size = 64MB
+max_slot_wal_keep_size = 1MB
+log_checkpoints = on
+));
+$standby->start;
+$primary->wait_for_replay_catchup($standby);
+
+my $injection_point = 'replication-slot-save-error';
+
+sub set_primary_wal_level
+{
+ my ($wal_level) = @_;
+
+ $primary->append_conf('postgresql.conf', "wal_level = $wal_level");
+ $primary->restart;
+}
+
+sub create_lagging_slot
+{
+ my ($slot_name) = @_;
+
+ $standby->create_logical_slot_on_standby($primary, $slot_name,
+ 'postgres');
+ $primary->advance_wal(8);
+ $primary->safe_psql('postgres', 'CHECKPOINT');
+ $primary->wait_for_replay_catchup($standby);
+ $standby->safe_psql(
+ 'postgres',
+ qq{
+SELECT injection_points_attach(
+ '$injection_point', 'wait', '$slot_name')
+});
+}
+
+sub backend_pid
+{
+ my ($backend_type) = @_;
+
+ return $standby->safe_psql(
+ 'postgres',
+ qq{
+SELECT pid
+FROM pg_stat_activity
+WHERE backend_type = '$backend_type'
+});
+}
+
+sub start_restartpoint
+{
+ my $checkpoint =
+ $standby->background_psql('postgres', on_error_stop => 0);
+
+ $checkpoint->set_query_timer_restart();
+ $checkpoint->query_until(
+ qr/checkpoint started/,
+ q(\echo checkpoint started
+CHECKPOINT;
+));
+
+ return $checkpoint;
+}
+
+sub finish_restartpoint
+{
+ my ($checkpoint) = @_;
+ my (undef, $error) = $checkpoint->query('SELECT 1', verbose => 0);
+
+ is($error, 0, 'restartpoint succeeds');
+ $checkpoint->quit;
+}
+
+sub wait_for_replication_slot_io
+{
+ my ($pid) = @_;
+
+ $standby->poll_query_until(
+ 'postgres',
+ qq{
+SELECT wait_event = 'ReplicationSlotIO'
+FROM pg_stat_activity
+WHERE pid = $pid
+}) or die "process $pid did not wait for replication slot I/O";
+}
+
+sub wake_invalidator
+{
+ my ($slot_name) = @_;
+ my $invalidation_reason;
+
+ foreach (1 .. 10 * $PostgreSQL::Test::Utils::timeout_default)
+ {
+ my $waiting = $standby->safe_psql(
+ 'postgres',
+ qq{
+SELECT count(*) > 0
+FROM pg_stat_activity
+WHERE wait_event = '$injection_point'
+});
+
+ $standby->safe_psql('postgres',
+ qq{SELECT injection_points_wakeup('$injection_point')})
+ if $waiting eq 't';
+
+ $invalidation_reason = $standby->safe_psql(
+ 'postgres',
+ qq{
+SELECT invalidation_reason
+FROM pg_replication_slots
+WHERE slot_name = '$slot_name'
+});
+
+ last if $invalidation_reason ne '' && $waiting eq 'f';
+ usleep(100_000);
+ }
+
+ die "timed out waiting for slot $slot_name to be invalidated"
+ if !defined($invalidation_reason) || $invalidation_reason eq '';
+
+ return $invalidation_reason;
+}
+
+# The checkpointer starts invalidation before the startup process.
+my $slot_name = 'checkpointer_first';
+create_lagging_slot($slot_name);
+my $startup_pid = backend_pid('startup');
+my $checkpointer_pid = backend_pid('checkpointer');
+my $log_start = -s $standby->logfile;
+
+my $checkpoint = start_restartpoint();
+$standby->wait_for_event('checkpointer', $injection_point);
+
+is( $standby->safe_psql(
+ 'postgres',
+ qq{
+SELECT active_pid = $checkpointer_pid
+FROM pg_replication_slots
+WHERE slot_name = '$slot_name'
+}),
+ 't',
+ 'checkpointer owns the slot while persisting invalidation');
+
+set_primary_wal_level('replica');
+wait_for_replication_slot_io($startup_pid);
+
+ok( !$standby->log_contains(
+ qr/terminating process $checkpointer_pid to release replication slot
+ \s+"$slot_name"/x,
+ $log_start),
+ 'startup process does not terminate the checkpointer');
+
+my $invalidation_reason = wake_invalidator($slot_name);
+finish_restartpoint($checkpoint);
+
+is($invalidation_reason, 'wal_removed',
+ 'slot is invalidated by the checkpointer');
+ok( !$standby->log_contains(
+ qr/canceling statement due to conflict with recovery/, $log_start),
+ 'checkpointer does not receive a recovery conflict');
+
+$primary->wait_for_replay_catchup($standby);
+$standby->safe_psql('postgres',
+ qq{SELECT injection_points_detach('$injection_point')});
+
+# The startup process starts invalidation before the checkpointer.
+set_primary_wal_level('logical');
+$primary->wait_for_replay_catchup($standby);
+
+$slot_name = 'startup_first';
+create_lagging_slot($slot_name);
+$startup_pid = backend_pid('startup');
+$checkpointer_pid = backend_pid('checkpointer');
+$log_start = -s $standby->logfile;
+
+set_primary_wal_level('replica');
+$standby->wait_for_event('startup', $injection_point);
+
+is( $standby->safe_psql(
+ 'postgres',
+ qq{
+SELECT active_pid = $startup_pid
+FROM pg_replication_slots
+WHERE slot_name = '$slot_name'
+}),
+ 't',
+ 'startup process owns the slot while persisting invalidation');
+
+$checkpoint = start_restartpoint();
+wait_for_replication_slot_io($checkpointer_pid);
+
+ok( !$standby->log_contains(
+ qr/terminating process $startup_pid to release replication slot
+ \s+"$slot_name"/x,
+ $log_start),
+ 'checkpointer does not terminate the startup process');
+
+$standby->safe_psql('postgres',
+ qq{SELECT injection_points_wakeup('$injection_point')});
+finish_restartpoint($checkpoint);
+$standby->wait_for_log(
+ qr/invalidating obsolete replication slot "$slot_name"/, $log_start);
+
+$primary->advance_wal(1);
+$primary->wait_for_replay_catchup($standby);
+is(backend_pid('startup'), $startup_pid, 'startup process survives');
+ok($standby->is_alive, 'standby remains running');
+
+if ($standby->is_alive)
+{
+ is( $standby->safe_psql(
+ 'postgres',
+ qq{
+SELECT invalidation_reason
+FROM pg_replication_slots
+WHERE slot_name = '$slot_name'
+}),
+ 'wal_level_insufficient',
+ 'slot is invalidated by the startup process');
+
+ $standby->safe_psql('postgres',
+ qq{SELECT injection_points_detach('$injection_point')});
+
+ $standby->stop;
+}
+
+$primary->stop;
+
+done_testing();
--
2.34.1
--1qxyLN8W7y4rfpkV
Content-Type: text/x-diff; charset=utf-8
Content-Disposition: attachment;
filename="v3-0002-Persist-synchronized-slot-invalidations-before-pu.patch"
Content-Transfer-Encoding: 8bit
Message-ID: <no-message-id-1974184@localhost>
Permalink: ../no-message-id-1974184@localhost/
Also on: postgresql.org/message-id/no-message-id-1974184@localhost
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: pgsql-hackers@postgresql.org
Cc: bertranddrouvot.pg@gmail.com
Subject: Re: [PATCH v3 1/2] Persist slot invalidations before publishing them
In-Reply-To: <no-message-id-1974184@localhost>
* 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