From 40e717fb82404bae76edb55f1355cda103df1e09 Mon Sep 17 00:00:00 2001 From: Bertrand Drouvot Date: Wed, 26 Aug 2026 05:34:51 +0000 Subject: [PATCH v3 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 Reviewed-by: Kyotaro Horiguchi Reviewed-by: Miłosz Bieniek Reviewed-by: JoongHyuk Shin Reviewed-by: Rui Zhao Discussion: https://postgr.es/m/ao7u5I9OeIR72kGp%40bdtpg Backpatch-through: 14 --- src/backend/replication/slot.c | 234 +++++++++++---- 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, 587 insertions(+), 49 deletions(-) 45.2% 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..8f2b681b626 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,32 @@ 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 lock is released on success and left held for error cleanup + * otherwise. + */ +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 +2063,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 +2117,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 +2138,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 +2148,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 +2158,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 +2186,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 +2246,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 +2256,17 @@ 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; 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,15 @@ 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. On error, leave the lock held for error cleanup. The lock is + * released after successful shared-memory publication. */ 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 +2628,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 +2638,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 +2659,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 +2686,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 +2714,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 +2736,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 +2753,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 +2770,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,6 +2797,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..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