From 5bb7de2f8080b1f361b612afadb8559e3d325d72 Mon Sep 17 00:00:00 2001 From: Bertrand Drouvot Date: Wed, 26 Aug 2026 05:34:51 +0000 Subject: [PATCH v2 1/2] Persist slot invalidations before publishing them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Reviewed-by: Kyotaro Horiguchi Reviewed-by: Miłosz Bieniek 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