Hi, On Thu, Sep 24, 2026 at 10:55:03AM +0530, shveta malik wrote: > I had a look at patch001 as well. I have 2 questions: > > 1) > Would it be better to use a PG_TRY/PG_CATCH block in patch 001, > similar to patch 002? Currently, the slot is released via > PG_ENSURE_ERROR_CLEANUP, while we rely on top-level error cleanup for > lock-release. Using TRY/CATCH would let us explicitly release both the > slot and I/O lock together, making the error handling consistent > across both patches. We could even reuse persist_slot_invalidation() > with a small change to pass update_inactive_since from the caller.
Yeah, makes sense. I changed 0001 to use PG_TRY/PG_CATCH and moved the common cleanup into ReplicationSlotPersistInvalidation(), with update_inactive_since passed by the caller. The I/O lock is still acquired by each caller because 0001 must hold it before claiming the inactive slot to serialize concurrent internal invalidators. > 2) > + /* Let caller know */ > + invalidated = true; > + LWLockRelease(&s->io_in_progress_lock); > ReplicationSlotRelease(); > > Wouldn't it be better (and safer) to release the slot before releasing > the I/O lock? > > Currently, concurrent invalidators are protected by the > 'invalidation_cause == RS_INVAL_NONE' check after acquiring the lock. > But releasing the slot first would close this race window entirely. It > would also make the order consistent with Patch 002 and the > error-handling flow in Patch 001 itself. The current ordering should be safe because the invalidation has already been published, so a concurrent invalidator exits before considering active_proc. That said, I agree that releasing the slot first means this ordering no longer relies on that check and makes the success and error paths consistent. So, done in the attached. It also adds the check you suggested for the synchronized slot's shared memory state after a successful synchronization. Regards, -- Bertrand Drouvot PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com
>From 22308850ac3dbdbd3724286127bd017f73cb369d Mon Sep 17 00:00:00 2001 From: Bertrand Drouvot <[email protected]> Date: Wed, 26 Aug 2026 05:34:51 +0000 Subject: [PATCH v6 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 failed save now leaves both the shared slot and its disk image valid. On ERROR, release ownership claimed for an inactive slot before releasing the I/O lock 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 tests using an injection point to cover failed saves, subsequent checkpointing and immediate restart, and concurrent internal invalidators. 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 | 229 +++++++++++---- src/include/replication/slot.h | 3 + src/test/recovery/meson.build | 2 + .../t/057_replslot_invalidation_durability.pl | 125 ++++++++ .../t/058_slot_invalidation_concurrency.pl | 273 ++++++++++++++++++ 5 files changed, 582 insertions(+), 50 deletions(-) 43.6% src/backend/replication/ 54.9% src/test/recovery/t/ diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 63ce6d27885..ef54833e0f9 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -185,13 +185,16 @@ 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 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 +772,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 +788,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 +821,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 +836,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; @@ -1168,7 +1184,46 @@ 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. On success, the caller + * retains both. On ERROR, release slot ownership before the I/O lock and + * update inactive_since only if requested. + */ +void +ReplicationSlotPersistInvalidation(ReplicationSlotInvalidationCause cause, + bool clear_restart_lsn, + bool update_inactive_since) +{ + 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)); + + PG_TRY(); + { + SaveSlotToPath(slot, path, ERROR, cause, clear_restart_lsn); + } + PG_CATCH(); + { + HOLD_INTERRUPTS(); /* match the upcoming RESUME_INTERRUPTS */ + ReplicationSlotReleaseInternal(update_inactive_since); + LWLockRelease(&slot->io_in_progress_lock); + PG_RE_THROW(); + } + PG_END_TRY(); } /* @@ -2005,6 +2060,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 +2114,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 +2135,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 +2145,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 +2155,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 +2183,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 +2243,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,10 +2253,14 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, LWLockRelease(ReplicationSlotControlLock); released_lock = true; - /* Make sure the invalidated state persists across server restart */ - ReplicationSlotMarkDirty(); - ReplicationSlotSave(); + ReplicationSlotPersistInvalidation(invalidation_cause, + invalidation_cause == RS_INVAL_WAL_REMOVED, + false); + + /* Let caller know */ + invalidated = true; ReplicationSlotRelease(); + LWLockRelease(&s->io_in_progress_lock); ReportSlotInvalidation(invalidation_cause, false, active_pid, slotname, restart_lsn, @@ -2380,7 +2468,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 +2581,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 +2605,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 +2620,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 +2630,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 +2651,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 +2678,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 +2706,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 +2728,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 +2745,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 +2762,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 +2789,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..885d3af236c 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -344,6 +344,9 @@ 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, + bool update_inactive_since); 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 0e6094ee5b32b80a66999c5774fd35ad8a846cfd Mon Sep 17 00:00:00 2001 From: Bertrand Drouvot <[email protected]> Date: Wed, 26 Aug 2026 05:36:40 +0000 Subject: [PATCH v6 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 | 13 +- src/backend/replication/slot.c | 2 +- .../t/057_replslot_invalidation_durability.pl | 135 ++++++++++++++++++ 3 files changed, 142 insertions(+), 8 deletions(-) 12.8% src/backend/replication/logical/ 84.7% src/test/recovery/t/ diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index c0403893e23..8191a3c4892 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -829,13 +829,12 @@ 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); - - /* Make sure the invalidated state persists across server restart */ - ReplicationSlotMarkDirty(); - ReplicationSlotSave(); + LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE); + ReplicationSlotPersistInvalidation(remote_slot->invalidated, false, + true); + 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 ef54833e0f9..a645a9f7add 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -1204,7 +1204,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..483f038b767 100644 --- a/src/test/recovery/t/057_replslot_invalidation_durability.pl +++ b/src/test/recovery/t/057_replslot_invalidation_durability.pl @@ -122,4 +122,139 @@ 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()'); + +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 is published'); + +$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
