On Thu, Jul 16, 2026 at 6:52 AM Masahiko Sawada <[email protected]> wrote: > For slot synchronization, the local slot could be created and > persisted based on the remote slot information fetched before the > deactivation was replayed, leaving a valid slot whose restart_lsn > precedes the deactivation. Decoding such a slot after a failover fails > with: > > ERROR: unexpected logical decoding status change 0 > > These races are confined to the narrow window between checking the > logical decoding status and the new slot becoming visible; once the > slot is visible, the invalidation performed by the deactivation > already covers it. So the fix is simple: re-check the logical decoding > status after the new slot becomes visible. Regular slot creation > raises an error and slot synchronization skips persisting the slot. If > the deactivation happens after the recheck instead, it is guaranteed > to invalidate the now-visible slot as usual. The attached 0002 > implements this.
The disable/re-enable case described in the comment above the final IsLogicalDecodingEnabled() check in update_and_persist_local_synced_slot() is reachable. On b73d13c3, the reproducer uses this sequence: 1. Slot sync fetches failover slot S and pauses at replication-slot-create-begin, before creating the local slot. 2. The primary drops S. The standby replays the logical-decoding deactivation while no local S exists to invalidate. 3. The primary recreates S. The standby replays the reactivation. 4. The old slot sync resumes with the first incarnation's restart_lsn. The final IsLogicalDecodingEnabled() check now returns true, so the old slot information is persisted. After promoting the standby, decoding that slot fails with: ERROR: unexpected logical decoding status change 0 The attached patch adds a logical-decoding status generation. Slot sync records it before fetching remote slot information and refuses to persist a new slot if the generation changed in the meantime. It drops the temporary slot so that the next attempt fetches the current incarnation. The new injection-point test fails without the fix because the stale slot is persisted. With the fix, it verifies that the replacement slot is fetched and that decoding succeeds after promotion. The existing 051_effective_wal_level test and the core regression tests also pass. This work was done by our new AI harness for Postgres testing. Nik
From 952bd42f9ed3ab2e392e2760b6148d16a2c721b6 Mon Sep 17 00:00:00 2001 From: Nik Samokhvalov <[email protected]> Date: Mon, 21 Sep 2026 22:14:53 -0700 Subject: [PATCH v1] Fix slot sync across logical decoding disable/re-enable Slot synchronization checks whether logical decoding is enabled before persisting a newly created slot. That misses a disable/re-enable cycle: the old synchronization can persist slot information fetched before the cycle, leaving a slot whose restart LSN precedes the deactivation record. Decoding from that slot after promotion then fails with an unexpected logical decoding status change. Track a generation for logical decoding status changes. Record it before fetching remote slots and require both an unchanged generation and enabled status before persistence. If the check fails, drop the temporary slot so the next attempt refetches the current incarnation. Add an injection-point test for the disable/re-enable race. --- src/backend/replication/logical/logicalctl.c | 39 ++++++++ src/backend/replication/logical/slotsync.c | 31 +++--- src/include/replication/logicalctl.h | 2 + src/test/recovery/meson.build | 1 + src/test/recovery/t/057_slotsync_wal_level.pl | 98 +++++++++++++++++++ 5 files changed, 160 insertions(+), 11 deletions(-) create mode 100644 src/test/recovery/t/057_slotsync_wal_level.pl diff --git a/src/backend/replication/logical/logicalctl.c b/src/backend/replication/logical/logicalctl.c index e5340880fa..27bc092d6b 100644 --- a/src/backend/replication/logical/logicalctl.c +++ b/src/backend/replication/logical/logicalctl.c @@ -95,6 +95,9 @@ typedef struct LogicalDecodingCtlData /* True if logical decoding might need to be disabled */ bool pending_disable; + + /* Incremented whenever logical_decoding_enabled changes */ + uint64 status_generation; } LogicalDecodingCtlData; static LogicalDecodingCtlData *LogicalDecodingCtl = NULL; @@ -210,6 +213,38 @@ IsLogicalDecodingEnabled(void) return enabled; } +/* + * Return the generation of the current logical decoding status. + */ +uint64 +GetLogicalDecodingStatusGeneration(void) +{ + uint64 generation; + + LWLockAcquire(LogicalDecodingControlLock, LW_SHARED); + generation = LogicalDecodingCtl->status_generation; + LWLockRelease(LogicalDecodingControlLock); + + return generation; +} + +/* + * Return true if logical decoding is enabled and its status has not changed + * since the given generation was read. + */ +bool +LogicalDecodingStatusMatches(uint64 generation) +{ + bool matches; + + LWLockAcquire(LogicalDecodingControlLock, LW_SHARED); + matches = LogicalDecodingCtl->logical_decoding_enabled && + LogicalDecodingCtl->status_generation == generation; + LWLockRelease(LogicalDecodingControlLock); + + return matches; +} + /* * Returns true if logical WAL logging is enabled based on the shared memory * status. @@ -421,6 +456,7 @@ EnableLogicalDecoding(void) * first. */ LogicalDecodingCtl->logical_decoding_enabled = true; + LogicalDecodingCtl->status_generation++; if (!in_recovery) write_logical_decoding_status_update_record(true); @@ -558,6 +594,8 @@ DisableLogicalDecoding(void) * processes WAL records with insufficient information. */ LogicalDecodingCtl->logical_decoding_enabled = false; + if (was_enabled) + LogicalDecodingCtl->status_generation++; /* Write the WAL to disable logical decoding on standbys too */ if (!in_recovery && was_enabled) @@ -653,6 +691,7 @@ UpdateLogicalDecodingStatusEndOfRecovery(void) */ LogicalDecodingCtl->xlog_logical_info = new_status; LogicalDecodingCtl->logical_decoding_enabled = new_status; + LogicalDecodingCtl->status_generation++; elog(DEBUG1, "update logical decoding status to %d at the end of recovery", diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index c184cdf1c1..22b486e21a 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -177,6 +177,9 @@ typedef struct RemoteSlot /* RS_INVAL_NONE if valid, or the reason of invalidation */ ReplicationSlotInvalidationCause invalidated; + + /* Local logical decoding status when this slot was fetched */ + uint64 logical_decoding_generation; } RemoteSlot; static void slotsync_failure_callback(int code, Datum arg); @@ -722,25 +725,23 @@ update_and_persist_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid, * before persisting it. This way, even if the status change record is * replayed after this check, the replay will invalidate our slot. * - * If the check fails, we keep the temporary slot and let the caller - * retry; the next cycle fetches the remote slot information again and - * will drop this slot as the remote slot no longer exists. - * - * XXX: this check cannot detect the case where logical decoding is - * already re-enabled by a slot creation on the primary at this point. - * Detecting that would require comparing the slot's restart_lsn with the - * LSN at which logical decoding was last enabled. + * If the check fails, drop the temporary slot and let the caller retry. + * The next cycle fetches the remote slot information again, which is + * necessary if logical decoding was disabled and then re-enabled in the + * meantime. */ - if (!IsLogicalDecodingEnabled()) + if (!LogicalDecodingStatusMatches(remote_slot->logical_decoding_generation)) { ereport(LOG, errmsg("could not synchronize replication slot \"%s\"", remote_slot->name), - errdetail("Logical decoding was concurrently disabled.")); + errdetail("Logical decoding status changed concurrently.")); if (slot_persistence_pending) *slot_persistence_pending = true; + ReplicationSlotDropAcquired(false); + return false; } @@ -922,7 +923,8 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid, slot_updated = true; } - ReplicationSlotRelease(); + if (MyReplicationSlot) + ReplicationSlotRelease(); return slot_updated; } @@ -947,6 +949,10 @@ fetch_remote_slots(WalReceiverConn *wrconn, List *slot_names) TupleTableSlot *tupslot; List *remote_slot_list = NIL; StringInfoData query; + uint64 logical_decoding_generation; + + /* Detect status changes while fetching and synchronizing these slots. */ + logical_decoding_generation = GetLogicalDecodingStatusGeneration(); initStringInfo(&query); appendStringInfoString(&query, @@ -993,6 +999,9 @@ fetch_remote_slots(WalReceiverConn *wrconn, List *slot_names) Datum d; int col = 0; + remote_slot->logical_decoding_generation = + logical_decoding_generation; + remote_slot->name = TextDatumGetCString(slot_getattr(tupslot, ++col, &isnull)); Assert(!isnull); diff --git a/src/include/replication/logicalctl.h b/src/include/replication/logicalctl.h index 0bc1302f13..5befdd18d6 100644 --- a/src/include/replication/logicalctl.h +++ b/src/include/replication/logicalctl.h @@ -18,6 +18,8 @@ extern void StartupLogicalDecodingStatus(bool last_status); extern void InitializeProcessXLogLogicalInfo(void); extern bool ProcessBarrierUpdateXLogLogicalInfo(void); extern bool IsLogicalDecodingEnabled(void); +extern uint64 GetLogicalDecodingStatusGeneration(void); +extern bool LogicalDecodingStatusMatches(uint64 generation); extern bool IsXLogLogicalInfoEnabled(void); extern void AtEOXact_LogicalCtl(void); extern void EnsureLogicalDecodingEnabled(void); diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index 72113c5ac6..5a82677a6e 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -65,6 +65,7 @@ tests += { 't/054_unlogged_sequence_promotion.pl', 't/055_cascade_reconnect.pl', 't/056_standby_snapshot_export.pl', + 't/057_slotsync_wal_level.pl', ], }, } diff --git a/src/test/recovery/t/057_slotsync_wal_level.pl b/src/test/recovery/t/057_slotsync_wal_level.pl new file mode 100644 index 0000000000..ce1ff65f3c --- /dev/null +++ b/src/test/recovery/t/057_slotsync_wal_level.pl @@ -0,0 +1,98 @@ +# Copyright (c) 2026, PostgreSQL Global Development Group + +use strict; +use warnings FATAL => 'all'; +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +plan skip_all => 'injection points not enabled' + unless $ENV{enable_injection_points} eq 'yes'; + +my $primary = PostgreSQL::Test::Cluster->new('primary'); +$primary->init(allows_streaming => 1); +$primary->append_conf('postgresql.conf', qq[ +autovacuum = off +log_min_messages = debug1 +]); +$primary->start; +$primary->safe_psql('postgres', q[ +create extension injection_points; +select pg_create_physical_replication_slot('phys_slot'); +]); +$primary->backup('backup'); + +my $standby = PostgreSQL::Test::Cluster->new('standby'); +$standby->init_from_backup($primary, 'backup', has_streaming => 1); +my $connstr = $primary->connstr; +$standby->append_conf('postgresql.conf', qq[ +primary_slot_name = 'phys_slot' +primary_conninfo = '$connstr dbname=postgres' +hot_standby_feedback = on +]); +$standby->start; + +$primary->safe_psql('postgres', q[ +select pg_create_logical_replication_slot( + 'sync_slot', 'test_decoding', false, false, true) +]); +my $old_restart_lsn = $primary->safe_psql('postgres', + q[select restart_lsn from pg_replication_slots where slot_name = 'sync_slot']); +$primary->wait_for_replay_catchup($standby); +is($standby->safe_psql('postgres', q[show effective_wal_level]), + 'logical', 'standby replayed activation'); + +my $sync = $standby->background_psql('postgres'); +$sync->query_until(qr/sync_started/, q(\echo sync_started +select injection_points_set_local(); +select injection_points_attach('replication-slot-create-begin', 'wait'); +select pg_sync_replication_slots(); +)); +$standby->wait_for_event('client backend', + 'replication-slot-create-begin'); +pass('slot sync fetched the first incarnation'); + +$primary->safe_psql('postgres', + q[select pg_drop_replication_slot('sync_slot')]); +$primary->poll_query_until('postgres', + q[select current_setting('effective_wal_level') = 'replica']) + or die 'timed out waiting for deactivation'; +$primary->wait_for_replay_catchup($standby); +is($standby->safe_psql('postgres', q[show effective_wal_level]), + 'replica', 'standby replayed deactivation'); + +$primary->safe_psql('postgres', q[ +select pg_create_logical_replication_slot( + 'sync_slot', 'test_decoding', false, false, true) +]); +my $new_restart_lsn = $primary->safe_psql('postgres', + q[select restart_lsn from pg_replication_slots where slot_name = 'sync_slot']); +isnt($new_restart_lsn, $old_restart_lsn, + 'second incarnation has a different restart LSN'); +$primary->wait_for_replay_catchup($standby); +is($standby->safe_psql('postgres', q[show effective_wal_level]), + 'logical', 'standby replayed reactivation'); + +$standby->safe_psql('postgres', q[ +select injection_points_detach('replication-slot-create-begin'); +select injection_points_wakeup('replication-slot-create-begin') +]); +$sync->quit; + +is($standby->safe_psql('postgres', + qq[select synced, temporary, invalidation_reason is null, + restart_lsn >= '$new_restart_lsn'::pg_lsn + from pg_replication_slots where slot_name = 'sync_slot']), + 't|f|t|t', 'slot was refetched before it was persisted'); + +$standby->promote; +my ($ret, $stdout, $stderr) = $standby->psql('postgres', + q[select * from pg_logical_slot_peek_changes('sync_slot', null, null)]); +is($ret, 0, 'decoding the refetched slot succeeds'); +unlike($stderr, qr/unexpected logical decoding status change/, + 'decoding does not reach a status change record'); + +$standby->stop; +$primary->stop; + +done_testing(); -- 2.50.1 (Apple Git-155)
