From 6fed1aa8d90009a6178beb3a8dc6198b982c3af4 Mon Sep 17 00:00:00 2001 From: Bertrand Drouvot Date: Wed, 26 Aug 2026 05:36:40 +0000 Subject: [PATCH v1 2/2] Persist synchronized slot invalidations before publishing them 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. It also prevents synchronization from retrying the save 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, restart, retry, and final durable invalidation. Author: Bertrand Drouvot Reviewed-by: Discussion: Backpatch-through: 17 --- src/backend/replication/logical/slotsync.c | 10 +- src/backend/replication/slot.c | 2 +- .../t/056_replslot_invalidation_durability.pl | 138 ++++++++++++++++++ 3 files changed, 142 insertions(+), 8 deletions(-) 10.0% src/backend/replication/logical/ 3.0% src/backend/replication/ 86.8% src/test/recovery/t/ diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index c0403893e23..51c19c60cf9 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -829,13 +829,9 @@ 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(); + ReplicationSlotPersistInvalidation(remote_slot->invalidated, false); + ReplicationSlotsComputeRequiredXmin(false); + ReplicationSlotsComputeRequiredLSN(); slot_updated = true; } diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index b5746eb6283..02cec90a20b 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -1211,7 +1211,7 @@ ReplicationSlotPersistInvalidation(ReplicationSlotInvalidationCause cause, char path[MAXPGPATH]; Assert(MyReplicationSlot != NULL); - Assert(MyReplicationSlot->data.persistency == RS_PERSISTENT); + Assert(MyReplicationSlot->data.persistency != RS_EPHEMERAL); Assert(MyReplicationSlot->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/056_replslot_invalidation_durability.pl b/src/test/recovery/t/056_replslot_invalidation_durability.pl index 247d7b00dc1..a360d6ee8f9 100644 --- a/src/test/recovery/t/056_replslot_invalidation_durability.pl +++ b/src/test/recovery/t/056_replslot_invalidation_durability.pl @@ -122,4 +122,142 @@ 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->stop('immediate'); +$standby->start; + +is( $standby->safe_psql( + 'postgres', + q{ +SELECT invalidation_reason IS NULL +FROM pg_replication_slots +WHERE slot_name = 'sync_slot' +}), + 't', + 'valid synchronized slot survives restart after failed save'); + +$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', + 'synchronized invalidation and restart LSN survive restart'); + +$standby->stop; +$primary->stop; + done_testing(); -- 2.34.1