agora inbox for pgsql-hackers@postgresql.org
help / color / mirror / Atom feedFrom: Bertrand Drouvot <bertranddrouvot.pg@gmail.com>
Subject: [PATCH v2 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.
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>
Discussion: https://postgr.es/m/ao7u5I9OeIR72kGp%40bdtpg
Backpatch-through: 14
---
src/backend/replication/slot.c | 140 +++++++++++++-----
src/include/replication/slot.h | 2 +
src/test/recovery/meson.build | 1 +
.../t/057_replslot_invalidation_durability.pl | 125 ++++++++++++++++
4 files changed, 235 insertions(+), 33 deletions(-)
56.6% src/backend/replication/
41.1% src/test/recovery/t/
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 63ce6d27885..b5746eb6283 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 ReplicationSlotReleaseOnError(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,18 @@ ReplicationSlotRelease(void)
}
}
+/*
+ * Release a slot claimed internally for invalidation after an error.
+ */
+static void
+ReplicationSlotReleaseOnError(int code, Datum arg)
+{
+ ReplicationSlot *slot = (ReplicationSlot *) DatumGetPointer(arg);
+
+ if (MyReplicationSlot == slot)
+ ReplicationSlotReleaseInternal(false);
+}
+
/*
* Cleanup temporary slots created in current session.
*
@@ -1168,7 +1197,29 @@ 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.
+ */
+void
+ReplicationSlotPersistInvalidation(ReplicationSlotInvalidationCause cause,
+ bool clear_restart_lsn)
+{
+ char path[MAXPGPATH];
+
+ Assert(MyReplicationSlot != NULL);
+ Assert(MyReplicationSlot->data.persistency == RS_PERSISTENT);
+ Assert(MyReplicationSlot->data.invalidated == RS_INVAL_NONE);
+ Assert(cause != RS_INVAL_NONE);
+ Assert(!clear_restart_lsn || cause == RS_INVAL_WAL_REMOVED);
+
+ sprintf(path, "%s/%s", PG_REPLSLOT_DIR,
+ NameStr(MyReplicationSlot->data.name));
+
+ SaveSlotToPath(MyReplicationSlot, path, ERROR, cause, clear_restart_lsn);
}
/*
@@ -2047,9 +2098,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 +2108,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
{
@@ -2159,8 +2196,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 +2206,18 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes,
LWLockRelease(ReplicationSlotControlLock);
released_lock = true;
- /* Make sure the invalidated state persists across server restart */
- ReplicationSlotMarkDirty();
- ReplicationSlotSave();
+ PG_ENSURE_ERROR_CLEANUP(ReplicationSlotReleaseOnError,
+ PointerGetDatum(s));
+ {
+ ReplicationSlotPersistInvalidation(
+ invalidation_cause,
+ invalidation_cause == RS_INVAL_WAL_REMOVED);
+ }
+ PG_END_ENSURE_ERROR_CLEANUP(ReplicationSlotReleaseOnError,
+ PointerGetDatum(s));
+
+ /* Let caller know */
+ invalidated = true;
ReplicationSlotRelease();
ReportSlotInvalidation(invalidation_cause, false, active_pid,
@@ -2380,7 +2426,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 +2539,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)
@@ -2519,7 +2565,9 @@ CreateSlotOnDisk(ReplicationSlot *slot)
* Shared functionality between saving and creating a replication slot.
*/
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 +2575,8 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
ReplicationSlotOnDisk cp;
bool was_dirty;
+ Assert(!clear_restart_lsn || invalidation_cause == RS_INVAL_WAL_REMOVED);
+
/* first check whether there's something to write out */
SpinLockAcquire(&slot->mutex);
was_dirty = slot->dirty;
@@ -2534,9 +2584,11 @@ 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;
+ INJECTION_POINT("replication-slot-save-error", NameStr(slot->data.name));
+
LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE);
/* silence valgrind :( */
@@ -2576,6 +2628,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);
@@ -2669,6 +2735,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..9248a7390a6 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -65,6 +65,7 @@ tests += {
't/054_unlogged_sequence_promotion.pl',
't/055_cascade_reconnect.pl',
't/056_standby_snapshot_export.pl',
+ 't/057_replslot_invalidation_durability.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();
--
2.34.1
--Cg2b3K2/P/2A21mh
Content-Type: text/x-diff; charset=utf-8
Content-Disposition: attachment;
filename="v2-0002-Persist-synchronized-slot-invalidations-before-pu.patch"
Content-Transfer-Encoding: 8bit
Message-ID: <no-message-id-1952404@localhost>
Permalink: ../no-message-id-1952404@localhost/
Also on: postgresql.org/message-id/no-message-id-1952404@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 v2 1/2] Persist slot invalidations before publishing them
In-Reply-To: <no-message-id-1952404@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