diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 7b38c167eac..d8ac785b1f4 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -8782,7 +8782,7 @@ xlog_redo(XLogReaderState *record) * subsequent WAL records, which may not contain logical information. */ if (status) - EnableLogicalDecoding(); + EnableLogicalDecoding(lsn); else DisableLogicalDecoding(); diff --git a/src/backend/replication/logical/logicalctl.c b/src/backend/replication/logical/logicalctl.c index 5089717ace5..572f1a07e93 100644 --- a/src/backend/replication/logical/logicalctl.c +++ b/src/backend/replication/logical/logicalctl.c @@ -64,6 +64,7 @@ #include "postgres.h" #include "access/xloginsert.h" +#include "access/xlogrecovery.h" #include "catalog/pg_control.h" #include "miscadmin.h" #include "replication/slot.h" @@ -95,6 +96,22 @@ typedef struct LogicalDecodingCtlData /* True if logical decoding might need to be disabled */ bool pending_disable; + + /* + * End LSN of the last XLOG_LOGICAL_DECODING_STATUS_CHANGE record that + * enabled logical decoding, or InvalidXLogRecPtr if none has been + * replayed since the server started. Checking it with + * logical_decoding_enabled tells from which point on WAL was written with + * logical decoding enabled; see StandbyLogicalDecodingEnabledSince(). + * + * WAL records that disable logical decoding are deliberately not tracked + * here. logical_decoding_enabled is false while decoding is off, which is + * all the check needs. + * + * This is maintained only during recovery and is not persisted, so it + * says nothing about a status change replayed in an earlier server run. + */ + XLogRecPtr last_replayed_enable_lsn; } LogicalDecodingCtlData; static LogicalDecodingCtlData *LogicalDecodingCtl = NULL; @@ -144,6 +161,7 @@ LogicalDecodingCtlShmemInit(void *arg) LogicalDecodingCtl->xlog_logical_info = false; LogicalDecodingCtl->logical_decoding_enabled = false; LogicalDecodingCtl->pending_disable = false; + LogicalDecodingCtl->last_replayed_enable_lsn = InvalidXLogRecPtr; } /* @@ -220,6 +238,31 @@ IsLogicalDecodingEnabled(void) return enabled; } +/* + * Return true if logical decoding has been enabled continuously from the given + * LSN up to the current replay position, that is, if the WAL in that range + * was written with logical decoding enabled. + * + * The given LSN must have been replayed already; nothing can be said about + * WAL this server has not replayed yet. The caller is responsible for checking + * that. + */ +bool +StandbyLogicalDecodingEnabledSince(XLogRecPtr lsn) +{ + bool result; + + Assert(RecoveryInProgress()); + Assert(lsn <= GetXLogReplayRecPtr(NULL)); + + LWLockAcquire(LogicalDecodingControlLock, LW_SHARED); + result = LogicalDecodingCtl->logical_decoding_enabled && + lsn >= LogicalDecodingCtl->last_replayed_enable_lsn; + LWLockRelease(LogicalDecodingControlLock); + + return result; +} + /* * Returns true if logical WAL logging is enabled based on the shared memory * status. @@ -344,21 +387,36 @@ EnsureLogicalDecodingEnabled(void) */ PG_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0); { - EnableLogicalDecoding(); + EnableLogicalDecoding(InvalidXLogRecPtr); } PG_END_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0); } /* * A workhorse function to enable logical decoding. + * + * lsn is the end LSN of the XLOG_LOGICAL_DECODING_STATUS_CHANGE record + * being replayed, and is InvalidXLogRecPtr when not called from redo. */ void -EnableLogicalDecoding(void) +EnableLogicalDecoding(XLogRecPtr lsn) { bool in_recovery; LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE); + /* + * Remember where logical decoding was enabled. This has to happen before + * the early return below, because replay can reach here with the status + * already on. For example, CreateCheckPoint() fixes the redo point before + * it records logicalDecodingEnabled, so a checkpoint can claim logical + * decoding is enabled while the record that enabled it still follows the + * redo point. + */ + Assert(RecoveryInProgress() == XLogRecPtrIsValid(lsn)); + if (XLogRecPtrIsValid(lsn)) + LogicalDecodingCtl->last_replayed_enable_lsn = lsn; + /* Return if it is already enabled */ if (LogicalDecodingCtl->logical_decoding_enabled) { diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index c184cdf1c17..aa4cc436ad7 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -709,41 +709,6 @@ update_and_persist_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid, return false; } - /* - * Do not persist the slot if logical decoding got disabled concurrently. - * This can happen if the last logical slot on the primary was dropped and - * the corresponding XLOG_LOGICAL_DECODING_STATUS_CHANGE record was - * replayed after we fetched the remote slot information: WAL records - * following the slot's restart_lsn might lack the information required by - * logical decoding, and the slot invalidation performed when replaying - * the record could not find our slot as it was not created yet. - * - * It is important to perform this check after creating the slot and - * 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 (!IsLogicalDecodingEnabled()) - { - ereport(LOG, - errmsg("could not synchronize replication slot \"%s\"", - remote_slot->name), - errdetail("Logical decoding was concurrently disabled.")); - - if (slot_persistence_pending) - *slot_persistence_pending = true; - - return false; - } - ReplicationSlotPersist(); ereport(LOG, @@ -874,6 +839,7 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid, { NameData plugin_name; TransactionId xmin_horizon = InvalidTransactionId; + XLogRecPtr replay_lsn; /* Skip creating the local slot if remote_slot is invalidated already */ if (remote_slot->invalidated != RS_INVAL_NONE) @@ -892,6 +858,63 @@ synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid, remote_slot->failover, true); + /* + * The remote slot information can predate a status change record that + * this standby has already replayed. That happens when the last + * logical slot on the primary is dropped, and possibly re-created + * with the same name, after fetch_remote_slots() ran. The resulting + * deactivation could not invalidate our slot because it did not exist + * yet, and WAL following the (stale) remote restart_lsn may lack the + * information logical decoding needs. Checking only whether logical + * decoding is enabled is not enough, as it can have been disabled and + * enabled again in the meantime. + * + * The check has to come after ReplicationSlotCreate(), which makes + * the slot both visible and acquired. A deactivation replayed from + * here on finds the slot in InvalidatePossiblyObsoleteSlot(), signals + * a recovery conflict and waits for the slot to be released before + * invalidating it (only in hot standby, which slot synchronization + * requires anyway). Replay therefore cannot get past that record + * behind our back, so the slot never needs to be rechecked before + * being persisted. + * + * The check only runs once replay has reached the remote restart_lsn; + * otherwise it is skipped and the slot is kept as-is. Without this, a + * standby lagging behind the primary (replay paused, or a large + * recovery_min_apply_delay) could fetch a live, valid restart_lsn + * from the primary and have it rejected by + * StandbyLogicalDecodingEnabledSince(), whose answer reflects only + * WAL replayed so far and says nothing about an LSN replay hasn't + * reached yet. That would drop a perfectly good slot every cycle. + * Skipping the check is safe: only a deactivation at or after the + * remote restart_lsn can leave the WAL that follows it without the + * information logical decoding needs, and replay has not reached such + * a record yet. Replaying it later invalidates the slot whether or + * not it has been persisted by then. + * + * Even so, the comparison uses the remote restart_lsn rather than the + * local one, so a slot that would have been usable may be dropped; + * the next cycle fetches fresh information. The slot cannot be kept, + * as it would go on using the stale restart_lsn. + */ + replay_lsn = GetXLogReplayRecPtr(NULL); + if (remote_slot->restart_lsn <= replay_lsn && + !StandbyLogicalDecodingEnabledSince(remote_slot->restart_lsn)) + { + ereport(LOG, + errmsg("could not synchronize replication slot \"%s\"", + remote_slot->name), + errdetail("Logical decoding was disabled after the remote slot's restart LSN %X/%08X.", + LSN_FORMAT_ARGS(remote_slot->restart_lsn))); + + ReplicationSlotDropAcquired(false); + + if (slot_persistence_pending) + *slot_persistence_pending = true; + + return false; + } + /* For shorter lines. */ slot = MyReplicationSlot; diff --git a/src/include/replication/logicalctl.h b/src/include/replication/logicalctl.h index 0bc1302f130..97cacb02e4e 100644 --- a/src/include/replication/logicalctl.h +++ b/src/include/replication/logicalctl.h @@ -14,14 +14,17 @@ #ifndef LOGICALCTL_H #define LOGICALCTL_H +#include "access/xlogdefs.h" + extern void StartupLogicalDecodingStatus(bool last_status); extern void InitializeProcessXLogLogicalInfo(void); extern bool ProcessBarrierUpdateXLogLogicalInfo(void); extern bool IsLogicalDecodingEnabled(void); +extern bool StandbyLogicalDecodingEnabledSince(XLogRecPtr lsn); extern bool IsXLogLogicalInfoEnabled(void); extern void AtEOXact_LogicalCtl(void); extern void EnsureLogicalDecodingEnabled(void); -extern void EnableLogicalDecoding(void); +extern void EnableLogicalDecoding(XLogRecPtr lsn); extern void RequestDisableLogicalDecoding(void); extern void DisableLogicalDecodingIfNecessary(void); extern void DisableLogicalDecoding(void); diff --git a/src/test/recovery/t/051_effective_wal_level.pl b/src/test/recovery/t/051_effective_wal_level.pl index 4f770e10522..ec8c3dc0684 100644 --- a/src/test/recovery/t/051_effective_wal_level.pl +++ b/src/test/recovery/t/051_effective_wal_level.pl @@ -627,6 +627,86 @@ select pg_sync_replication_slots(); '0', "no synced slot is left behind on standby5"); + # Test the same race, but where the slot is re-created on the primary + # before the slot synchronization resumes. Logical decoding is enabled + # again at the time the local slot is created, so checking the logical + # decoding status alone cannot tell that the remote slot information + # predates the deactivation. + + $primary->safe_psql('postgres', + qq[select pg_create_logical_replication_slot('sync_slot', 'test_decoding', false, false, true)] + ); + $primary->wait_for_replay_catchup($standby5); + test_wal_level($standby5, "replica|logical", + "logical decoding got activated on standby5 for the re-creation test" + ); + + $psql_sync_slot = $standby5->background_psql('postgres'); + $psql_sync_slot->query_until( + qr/sync_slots/, + q(\echo sync_slots +select injection_points_set_local(); +select injection_points_attach('replication-slot-create-begin', 'wait'); +select pg_sync_replication_slots(); +)); + $standby5->wait_for_event('client backend', + 'replication-slot-create-begin'); + note("injection_point 'replication-slot-create-begin' is reached"); + + # Drop and re-create the slot, and wait for the standby to replay both + # the deactivation and the activation. + $primary->safe_psql('postgres', + qq[select pg_drop_replication_slot('sync_slot')]); + wait_for_logical_decoding_disabled($primary); + $primary->safe_psql('postgres', + qq[select pg_create_logical_replication_slot('sync_slot', 'test_decoding', false, false, true)] + ); + my $restart_lsn = $primary->safe_psql('postgres', + qq[select restart_lsn from pg_replication_slots where slot_name = 'sync_slot'] + ); + $primary->wait_for_replay_catchup($standby5); + test_wal_level($standby5, "replica|logical", + "logical decoding got deactivated and activated again on standby5"); + + # Resume the slot synchronization. It must drop the slot created from + # the stale information, and re-create it from the re-created remote slot + # on retry. + $log_offset = -s $standby5->logfile; + $standby5->safe_psql( + 'postgres', qq[ +select injection_points_detach('replication-slot-create-begin'); +select injection_points_wakeup('replication-slot-create-begin'); +]); + $standby5->wait_for_log( + qr/could not synchronize replication slot "sync_slot".*\n.*DETAIL: Logical decoding was disabled after the remote slot's restart LSN/, + $log_offset); + $standby5->poll_query_until('postgres', + qq[select restart_lsn >= '$restart_lsn' from pg_replication_slots where slot_name = 'sync_slot'] + ) + or die + "timed out waiting for the slot to be re-created from the re-created remote slot"; + + # The slot created on retry might not be persisted until the remote slot + # catches up with the catalog_xmin computed locally. Drop the remote slot + # to let the slot synchronization finish, keeping logical decoding enabled + # with another slot as the slot synchronization requires it. + $primary->safe_psql( + 'postgres', qq[ +select pg_create_logical_replication_slot('test_slot4', 'test_decoding'); +select pg_drop_replication_slot('sync_slot'); +]); + $primary->wait_for_replay_catchup($standby5); + $psql_sync_slot->quit; + $primary->safe_psql('postgres', + qq[select pg_drop_replication_slot('test_slot4')]); + wait_for_logical_decoding_disabled($primary); + $primary->wait_for_replay_catchup($standby5); + is( $standby5->safe_psql( + 'postgres', qq[select count(*) from pg_replication_slots]), + '0', + "no synced slot is left behind on standby5 after the re-creation test" + ); + # Test that logical slot creation on a standby fails cleanly if logical # decoding is concurrently deactivated by the end-of-recovery transition # upon promotion, which cannot find the slot that is not created yet.