From a5c08a9913dea62fcd088be6a403b03afb1de5a5 Mon Sep 17 00:00:00 2001 From: Rui Zhao Date: Mon, 14 Sep 2026 11:47:00 +0800 Subject: [PATCH] Wait for a concurrent invalidation of a slot instead of terminating it With the invalidation persisted before it is published, a slot being invalidated looks like a slot in use for the duration of a file write: it has an active_pid and no invalidation reason. A second invalidator that finds it in that state signals its owner as it would signal a walsender. On a standby the startup process and the checkpointer both invalidate slots, and the checkpointer signals with SIGTERM, which makes the startup process exit and takes the standby down. Examine the slot with its io_in_progress_lock held, and keep that lock from the moment the slot is claimed until the invalidation is published. A concurrent invalidator then sees either the invalidation or a slot that nobody is invalidating. The wait for the lock is done without ReplicationSlotControlLock, like the other waits in that function. Add a test with the two processes in both orders. --- src/backend/replication/logical/slotsync.c | 1 + src/backend/replication/slot.c | 49 +++- src/test/recovery/meson.build | 1 + .../recovery/t/058_slot_invalidation_race.pl | 263 ++++++++++++++++++ 4 files changed, 311 insertions(+), 3 deletions(-) create mode 100644 src/test/recovery/t/058_slot_invalidation_race.pl diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index 51c19c60cf..cd91ab1e9a 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -829,6 +829,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid, if (slot->data.invalidated == RS_INVAL_NONE && remote_slot->invalidated != RS_INVAL_NONE) { + LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE); ReplicationSlotPersistInvalidation(remote_slot->invalidated, false); ReplicationSlotsComputeRequiredXmin(false); ReplicationSlotsComputeRequiredLSN(); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 02cec90a20..3805b89560 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -1203,6 +1203,8 @@ ReplicationSlotSave(void) /* * Persist an invalidated image of the acquired slot before publishing the * invalidation in shared memory. + * + * The caller holds the slot's io_in_progress_lock; it is released on return. */ void ReplicationSlotPersistInvalidation(ReplicationSlotInvalidationCause cause, @@ -1211,6 +1213,8 @@ ReplicationSlotPersistInvalidation(ReplicationSlotInvalidationCause cause, char path[MAXPGPATH]; Assert(MyReplicationSlot != NULL); + Assert(LWLockHeldByMeInMode(&MyReplicationSlot->io_in_progress_lock, + LW_EXCLUSIVE)); Assert(MyReplicationSlot->data.persistency != RS_EPHEMERAL); Assert(MyReplicationSlot->data.invalidated == RS_INVAL_NONE); Assert(cause != RS_INVAL_NONE); @@ -2016,6 +2020,12 @@ DetermineSlotInvalidationCause(uint32 possible_causes, ReplicationSlot *s, * * Acquires the given slot and mark it invalid, if necessary and possible. * + * The slot is examined with its io_in_progress_lock held, and the lock is + * kept from the moment the slot is claimed until the invalidation has been + * published in shared memory. A concurrent invalidator thus either sees the + * invalidation or a slot nobody is invalidating, and does not take the + * process persisting the invalidation for a process using the slot. + * * Returns true if the slot was invalidated. * * Set *released_lock_out if ReplicationSlotControlLock was released in the @@ -2056,6 +2066,22 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, break; } + /* + * Getting the io_in_progress_lock may mean waiting for a write of the + * slot; don't do that with ReplicationSlotControlLock held. + */ + if (!LWLockConditionalAcquire(&s->io_in_progress_lock, LW_EXCLUSIVE)) + { + LWLockRelease(ReplicationSlotControlLock); + released_lock = true; + + LWLockAcquire(&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) { /* @@ -2089,6 +2115,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; @@ -2134,6 +2161,12 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, if (active_proc != INVALID_PROC_NUMBER) { + /* + * The owner is not invalidating the slot, or we would have seen + * the invalidation: it is using the slot. + */ + LWLockRelease(&s->io_in_progress_lock); + /* * Prepare the sleep on the slot's condition variable before * releasing the lock, to close a possible race condition if the @@ -2196,8 +2229,9 @@ InvalidatePossiblyObsoleteSlot(uint32 possible_causes, else { /* - * We hold the slot now. Persist its invalidation before - * publishing it in shared memory. + * We hold the slot and its io_in_progress_lock now. Persist the + * invalidation before publishing it in shared memory; the lock is + * released once that is done. * * Don't want to hold ReplicationSlotControlLock across file * system operations, so release it now but be sure to tell caller @@ -2589,7 +2623,16 @@ SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel, INJECTION_POINT("replication-slot-save-error", NameStr(slot->data.name)); - LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE); + /* + * An invalidation is written with the io_in_progress_lock already held by + * the caller, from before it claimed the slot; see + * InvalidatePossiblyObsoleteSlot(). The lock is released below all the + * same. + */ + if (invalidation_cause == RS_INVAL_NONE) + LWLockAcquire(&slot->io_in_progress_lock, LW_EXCLUSIVE); + else + Assert(LWLockHeldByMeInMode(&slot->io_in_progress_lock, LW_EXCLUSIVE)); /* silence valgrind :( */ memset(&cp, 0, sizeof(ReplicationSlotOnDisk)); diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index 9248a7390a..698a934c06 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -66,6 +66,7 @@ tests += { 't/055_cascade_reconnect.pl', 't/056_standby_snapshot_export.pl', 't/057_replslot_invalidation_durability.pl', + 't/058_slot_invalidation_race.pl', ], }, } diff --git a/src/test/recovery/t/058_slot_invalidation_race.pl b/src/test/recovery/t/058_slot_invalidation_race.pl new file mode 100644 index 0000000000..33387885e3 --- /dev/null +++ b/src/test/recovery/t/058_slot_invalidation_race.pl @@ -0,0 +1,263 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group +# +# Two processes can try to invalidate the same replication slot at about +# the same time. On a standby, the startup process invalidates logical +# slots when the primary disables logical decoding, and the checkpointer +# invalidates slots whose WAL a restartpoint is about to remove. The one +# that comes second has to wait for the other, not terminate it. +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('bkp'); + +my $standby = PostgreSQL::Test::Cluster->new('standby'); +$standby->init_from_backup($primary, 'bkp', 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'; + +# Set the primary's wal_level and restart it. Going to replica disables +# logical decoding, which the startup process of the standby applies by +# invalidating the logical slots of the standby. +sub set_primary_wal_level +{ + my ($wal_level) = @_; + + $primary->append_conf('postgresql.conf', "wal_level = $wal_level"); + $primary->restart; +} + +# Create a logical slot on the standby and make its restart_lsn older than +# what max_slot_wal_keep_size lets a restartpoint keep. Replay a checkpoint +# record so that a restartpoint is possible afterwards. The slot's save +# will stop at the injection point. +sub create_lagging_slot +{ + my ($slot) = @_; + + $standby->create_logical_slot_on_standby($primary, $slot, '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')}); +} + +sub pid_of +{ + my ($backend_type) = @_; + + return $standby->safe_psql('postgres', + qq{SELECT pid FROM pg_stat_activity + WHERE backend_type = '$backend_type'}); +} + +# Start a CHECKPOINT on the standby in the background, and return the +# handle to finish it with. +sub start_restartpoint +{ + my ($out, $err) = @_; + + return IPC::Run::start( + [ + 'psql', '-X', '-d', $standby->connstr('postgres'), '-c', + 'CHECKPOINT' + ], + '>' => $out, + '2>' => $err); +} + +# Wait until the process waits for the slot, or is done with it. +sub wait_for_second_invalidator +{ + my ($pid, $done_re, $logstart) = @_; + my $primary_lsn = $primary->lsn('write'); + + foreach (1 .. 10 * $PostgreSQL::Test::Utils::timeout_default) + { + return + if $standby->log_contains($done_re, $logstart); + return + if $standby->safe_psql( + 'postgres', + qq{SELECT wait_event IN ('ReplicationSlotIO', + 'ReplicationSlotDrop', '$injection_point') + OR pg_last_wal_replay_lsn() >= '$primary_lsn' + FROM pg_stat_activity WHERE pid = $pid}) eq 't'; + usleep(100_000); + } + die "timed out waiting for process $pid to deal with the slot"; +} + +# Wake up whoever waits at the injection point, until nobody does and +# the slot is invalidated. +sub wake_until_invalidated +{ + my ($slot) = @_; + my $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'; + $reason = $standby->safe_psql( + 'postgres', + qq{SELECT invalidation_reason FROM pg_replication_slots + WHERE slot_name = '$slot'}); + last if $reason ne '' && $waiting eq 'f'; + usleep(100_000); + } + return $reason; +} + +# +# Checkpointer first, startup process second. +# +my $slot = 'checkpointer_first'; +create_lagging_slot($slot); +my $startup_pid = pid_of('startup'); +my $checkpointer_pid = pid_of('checkpointer'); +my $logstart = -s $standby->logfile; + +my ($ckpt_out, $ckpt_err) = ('', ''); +my $ckpt = start_restartpoint(\$ckpt_out, \$ckpt_err); +$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'}), + 't', + 'checkpointer holds the slot while persisting its invalidation'); + +set_primary_wal_level('replica'); +wait_for_second_invalidator($startup_pid, + qr/invalidating obsolete replication slot "$slot"/, $logstart); +ok( !$standby->log_contains( + qr/terminating process $checkpointer_pid to release replication slot "$slot"/, + $logstart), + 'startup process does not terminate the checkpointer'); + +my $reason = wake_until_invalidated($slot); +$ckpt->finish; +is($ckpt_err, '', 'restartpoint succeeds'); +is($reason, 'wal_removed', 'slot invalidated by the checkpointer'); +ok( !$standby->log_contains( + qr/canceling statement due to conflict with recovery/, $logstart), + 'checkpointer sees no recovery conflict'); +$primary->wait_for_replay_catchup($standby); +$standby->safe_psql('postgres', + qq{SELECT injection_points_detach('$injection_point')}); + +# +# Startup process first, checkpointer second. +# +set_primary_wal_level('logical'); +$primary->wait_for_replay_catchup($standby); +$slot = 'startup_first'; +create_lagging_slot($slot); +$startup_pid = pid_of('startup'); +$checkpointer_pid = pid_of('checkpointer'); +$logstart = -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'}), + 't', + 'startup process holds the slot while persisting its invalidation'); + +($ckpt_out, $ckpt_err) = ('', ''); +$ckpt = start_restartpoint(\$ckpt_out, \$ckpt_err); +$standby->wait_for_log(qr/restartpoint starting/, $logstart); +wait_for_second_invalidator($checkpointer_pid, qr/restartpoint complete/, + $logstart); +ok( !$standby->log_contains( + qr/terminating process $startup_pid to release replication slot "$slot"/, + $logstart), + 'checkpointer does not terminate the startup process'); + +$standby->safe_psql('postgres', + qq{SELECT injection_points_wakeup('$injection_point')}); +$ckpt->finish; +is($ckpt_err, '', 'restartpoint succeeds'); +$standby->wait_for_log(qr/invalidating obsolete replication slot "$slot"/, + $logstart); + +# A startup process that got a SIGTERM exits as soon as it is back in its +# loop, and takes the standby down with it. +foreach (1 .. 20) +{ + last + if $standby->log_contains( + qr/startup process \(PID $startup_pid\) exited/, $logstart); + usleep(100_000); +} +ok( !$standby->log_contains( + qr/startup process \(PID $startup_pid\) exited/, $logstart), + 'startup process survives'); +ok($standby->is_alive, 'standby is up'); + +if ($standby->is_alive) +{ + is( $standby->safe_psql( + 'postgres', + qq{SELECT invalidation_reason FROM pg_replication_slots + WHERE slot_name = '$slot'}), + 'wal_level_insufficient', + 'slot invalidated by the startup process'); + $standby->safe_psql('postgres', + qq{SELECT injection_points_detach('$injection_point')}); + $standby->stop; +} +$primary->stop; + +done_testing(); -- 2.43.7