Hi, On Wed, Sep 23, 2026 at 12:00:39PM +0530, shveta malik wrote: > On Wed, Sep 23, 2026 at 11:37 AM Bertrand Drouvot > <[email protected]> wrote: > > > > Another possibility would be to make ReplicationSlotPersistInvalidation() > > always > > leave the caller acquired I/O lock held. Slotsync could then release that > > specific > > lock in a PG_CATCH() block, something like: > > Yes, I agree. I find this approach much better for 2 reasons: > > 1) The caller has better control over the lock, which makes sense > since it is the one acquiring it. > 2) It makes the code much more understandable. Earlier, it took me a > while to figure out exactly where the io_in_progress_lock was getting > released, especially looking at the slotsync patch where it was > acquired right before calling ReplicationSlotPersistInvalidation(). > > > " > > PG_CATCH(); > > { > > HOLD_INTERRUPTS(); > > LWLockRelease(&slot->io_in_progress_lock); > > PG_RE_THROW(); > > } > > PG_END_TRY(); > > > > LWLockRelease(&slot->io_in_progress_lock); > > " > > > > This would avoid both LWLockReleaseAll() and using LWLockHeldByMe() for > > normal > > control flow. Does that sound preferable? > > Yes.
Thanks! Done that way in v4 attached. Regards, -- Bertrand Drouvot PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com
>From 9ddcf170e434f4cc8bdc7be5ab5a6fd329b92ee0 Mon Sep 17 00:00:00 2001 From: Bertrand Drouvot <[email protected]> Date: Wed, 26 Aug 2026 05:34:51 +0000 Subject: [PATCH v4 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. 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 <[email protected]> Reviewed-by: Kyotaro Horiguchi <[email protected]> Reviewed-by: Miłosz Bieniek <[email protected]> Reviewed-by: JoongHyuk Shin <[email protected]> Reviewed-by: Rui Zhao <[email protected]> Reviewed-by: shveta malik <[email protected]> Discussion: https://postgr.es/m/ao7u5I9OeIR72kGp%40bdtpg Backpatch-through: 14 --- src/backend/replication/slot.c | 236 +++++++++++---- 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, 588 insertions(+), 50 deletions(-) 45.1% 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..642babebeab 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,31 @@ 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 caller is responsible for releasing the lock. + */ +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 +2062,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 +2116,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 +2137,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 +2147,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 +2157,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 +2185,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 +2245,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 +2255,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(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; + LWLockRelease(&s->io_in_progress_lock); 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,14 @@ 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 and remains responsible for releasing it. */ 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 +2627,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 +2637,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 +2658,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 +2685,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 +2713,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 +2735,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 +2752,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 +2769,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,13 +2796,22 @@ 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; slot->last_saved_restart_lsn = cp.slotdata.restart_lsn; SpinLockRelease(&slot->mutex); - LWLockRelease(&slot->io_in_progress_lock); + if (invalidation_cause == RS_INVAL_NONE) + LWLockRelease(&slot->io_in_progress_lock); } /* 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
>From 532ef5eeb81191bf3210e15b39c407401c4b7771 Mon Sep 17 00:00:00 2001 From: Bertrand Drouvot <[email protected]> Date: Wed, 26 Aug 2026 05:36:40 +0000 Subject: [PATCH v4 2/2] Persist synchronized slot invalidations before publishing them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit synchronize_one_slot() publishes a remote slot's invalidation before saving the local synchronized slot. A save failure therefore leaves the shared slot invalid while its disk image remains valid. On the next synchronization, drop_local_obsolete_slots() retains the local slot because the remote slot is also invalidated. However, synchronize_one_slot() sees the local slot already invalidated and skips the save, so the failed save is not retried directly. Use ReplicationSlotPersistInvalidation() so the invalidated image is durable before publication. Preserve the local restart LSN and recompute resource horizons only after the invalidation becomes durable and visible. A failed save now leaves the local slot valid, allowing the next synchronization to retry. Add a primary and standby test covering the failure, a direct retry, and an immediate restart that verifies durable invalidation. Author: Bertrand Drouvot <[email protected]> Reviewed-by: Kyotaro Horiguchi <[email protected]> Reviewed-by: Miłosz Bieniek <[email protected]> Reviewed-by: JoongHyuk Shin <[email protected]> Reviewed-by: Rui Zhao <[email protected]> Reviewed-by: shveta malik <[email protected]> Discussion: https://postgr.es/m/ao7u5I9OeIR72kGp%40bdtpg Backpatch-through: 17 --- src/backend/replication/logical/slotsync.c | 28 +++- src/backend/replication/slot.c | 2 +- .../t/057_replslot_invalidation_durability.pl | 125 ++++++++++++++++++ 3 files changed, 148 insertions(+), 7 deletions(-) 20.4% src/backend/replication/logical/ 77.2% src/test/recovery/t/ diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index c0403893e23..9121a80ec36 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -829,13 +829,29 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid, if (slot->data.invalidated == RS_INVAL_NONE && remote_slot->invalidated != RS_INVAL_NONE) { - SpinLockAcquire(&slot->mutex); - slot->data.invalidated = remote_slot->invalidated; - SpinLockRelease(&slot->mutex); + LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE); + + PG_TRY(); + { + ReplicationSlotPersistInvalidation(remote_slot->invalidated, + false); + } + PG_CATCH(); + { + /* + * Release ownership before making the I/O lock available to + * concurrent invalidators. + */ + HOLD_INTERRUPTS(); /* match the upcoming RESUME_INTERRUPTS */ + ReplicationSlotRelease(); + LWLockRelease(&slot->io_in_progress_lock); + PG_RE_THROW(); + } + PG_END_TRY(); - /* Make sure the invalidated state persists across server restart */ - ReplicationSlotMarkDirty(); - ReplicationSlotSave(); + LWLockRelease(&slot->io_in_progress_lock); + ReplicationSlotsComputeRequiredXmin(false); + ReplicationSlotsComputeRequiredLSN(); slot_updated = true; } diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 642babebeab..65b97dfab91 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -1217,7 +1217,7 @@ ReplicationSlotPersistInvalidation(ReplicationSlotInvalidationCause cause, ReplicationSlot *slot = MyReplicationSlot; Assert(slot != NULL); - Assert(slot->data.persistency == RS_PERSISTENT); + Assert(slot->data.persistency != RS_EPHEMERAL); Assert(slot->data.invalidated == RS_INVAL_NONE); Assert(cause != RS_INVAL_NONE); Assert(!clear_restart_lsn || cause == RS_INVAL_WAL_REMOVED); diff --git a/src/test/recovery/t/057_replslot_invalidation_durability.pl b/src/test/recovery/t/057_replslot_invalidation_durability.pl index 247d7b00dc1..7f9bfb2b694 100644 --- a/src/test/recovery/t/057_replslot_invalidation_durability.pl +++ b/src/test/recovery/t/057_replslot_invalidation_durability.pl @@ -122,4 +122,129 @@ ok(-f $restart_segment_path, $node->stop; +# Check that slot synchronization also persists an invalidation before +# publishing it. +my $primary = PostgreSQL::Test::Cluster->new('sync_primary'); +$primary->init(allows_streaming => 'logical', extra => ['--wal-segsize=1']); +$primary->append_conf( + 'postgresql.conf', qq( +autovacuum = off +checkpoint_timeout = 1h +max_wal_size = 64MB +)); +$primary->start; +$primary->safe_psql('postgres', 'CREATE EXTENSION injection_points'); +$primary->safe_psql('postgres', + q{SELECT pg_create_physical_replication_slot('sync_phys')}); +$primary->backup('sync_backup'); + +my $standby = PostgreSQL::Test::Cluster->new('sync_standby'); +$standby->init_from_backup( + $primary, 'sync_backup', + has_streaming => 1, + has_restoring => 1); +my $primary_connstr = $primary->connstr; +$standby->append_conf( + 'postgresql.conf', qq( +checkpoint_timeout = 1h +hot_standby_feedback = on +primary_slot_name = 'sync_phys' +primary_conninfo = '$primary_connstr dbname=postgres' +)); +$standby->start; +$primary->wait_for_replay_catchup($standby); + +$primary->safe_psql( + 'postgres', + q{SELECT pg_create_logical_replication_slot( + 'sync_slot', 'pgoutput', false, false, true)}); + +my $slot_synced = 'f'; +foreach (1 .. 10) +{ + $primary->safe_psql('postgres', 'SELECT pg_log_standby_snapshot()'); + $primary->wait_for_replay_catchup($standby); + $standby->safe_psql('postgres', 'SELECT pg_sync_replication_slots()'); + $slot_synced = $standby->safe_psql( + 'postgres', + q{ +SELECT count(*) = 1 +FROM pg_replication_slots +WHERE slot_name = 'sync_slot' + AND synced + AND NOT temporary + AND invalidation_reason IS NULL +}); + last if $slot_synced eq 't'; +} +is($slot_synced, 't', 'valid failover slot is synchronized'); +my $sync_restart_lsn = $standby->safe_psql( + 'postgres', + q{ +SELECT restart_lsn +FROM pg_replication_slots +WHERE slot_name = 'sync_slot' +}); + +$primary->append_conf('postgresql.conf', 'max_slot_wal_keep_size = 1MB'); +$primary->reload; +$primary->advance_wal(8); +$primary->wait_for_replay_catchup($standby); +$primary->safe_psql('postgres', 'CHECKPOINT'); + +is( $primary->safe_psql( + 'postgres', + q{ +SELECT invalidation_reason +FROM pg_replication_slots +WHERE slot_name = 'sync_slot' +}), + 'wal_removed', + 'failover slot is invalidated on the primary'); + +$standby->safe_psql( + 'postgres', + q{ +SELECT injection_points_attach( + 'replication-slot-save-error', 'error', 'sync_slot') +}); + +($ret, $stdout, $stderr) = + $standby->psql('postgres', 'SELECT pg_sync_replication_slots()'); +like( + $stderr, + qr/error triggered for injection point replication-slot-save-error/, + 'injected error prevents synchronized invalidation from being saved'); + +is( $standby->safe_psql( + 'postgres', + q{ +SELECT invalidation_reason IS NULL +FROM pg_replication_slots +WHERE slot_name = 'sync_slot' +}), + 't', + 'failed save leaves synchronized slot valid'); + +$standby->safe_psql('postgres', + q{SELECT injection_points_detach('replication-slot-save-error')}); + +$standby->safe_psql('postgres', 'SELECT pg_sync_replication_slots()'); + +$standby->stop('immediate'); +$standby->start; + +is( $standby->safe_psql( + 'postgres', + qq{ +SELECT invalidation_reason, restart_lsn = '$sync_restart_lsn' +FROM pg_replication_slots +WHERE slot_name = 'sync_slot' +}), + 'wal_removed|t', + 'retried synchronized invalidation and restart LSN survive restart'); + +$standby->stop; +$primary->stop; + done_testing(); -- 2.34.1
