From c82ad716dffdb3491ad8e18f93a9d1d0b451139f Mon Sep 17 00:00:00 2001 From: Bertrand Drouvot Date: Wed, 26 Aug 2026 05:34:51 +0000 Subject: [PATCH v1 1/2] Persist slot invalidations before publishing them 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: Discussion: Backpatch-through: 14 --- src/backend/replication/slot.c | 140 +++++++++++++----- src/include/replication/slot.h | 2 + src/test/recovery/meson.build | 1 + .../t/056_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 39ec8c4946d..a74b9c64a4a 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -64,6 +64,7 @@ tests += { 't/053_standby_login_event_trigger.pl', 't/054_unlogged_sequence_promotion.pl', 't/055_cascade_reconnect.pl', + 't/056_replslot_invalidation_durability.pl', ], }, } diff --git a/src/test/recovery/t/056_replslot_invalidation_durability.pl b/src/test/recovery/t/056_replslot_invalidation_durability.pl new file mode 100644 index 00000000000..247d7b00dc1 --- /dev/null +++ b/src/test/recovery/t/056_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