/*
* Enable logical decoding if disabled.
*
- * If this function is called during recovery, it simply returns without
- * action since the logical decoding status change is not allowed during
- * this time. The logical decoding status depends on the status on the primary.
- * The caller should use CheckLogicalDecodingRequirements() before calling this
- * function to make sure that the logical decoding status can be modified.
+ * If this function is called during recovery, it just checks that logical
+ * decoding is still enabled, since the logical decoding status cannot be
+ * changed during this time. The logical decoding status depends on the
+ * status on the primary.
*
* Note that there is no interlock between logical decoding activation
* and slot creation. To ensure enabling logical decoding, the caller
if (RecoveryInProgress())
{
/*
- * CheckLogicalDecodingRequirements() must have already errored out if
- * logical decoding is not enabled since we cannot enable the logical
- * decoding status during recovery.
+ * The caller has already checked that logical decoding is enabled via
+ * CheckLogicalDecodingRequirements(), but the status could have been
+ * disabled concurrently before we created our slot: either by
+ * replaying an XLOG_LOGICAL_DECODING_STATUS_CHANGE record, or by
+ * UpdateLogicalDecodingStatusEndOfRecovery() upon promotion. We
+ * cannot enable logical decoding during recovery, so raise an error.
+ *
+ * Our slot has already been created, so its in_use flag is set and
+ * the slot scans performed by a deactivation can see it. It
+ * guarantees that this check doesn't miss a concurrent deactivation:
+ * UpdateLogicalDecodingStatusEndOfRecovery() won't disable logical
+ * decoding since CheckLogicalSlotExists() finds our valid slot, and
+ * replaying a status change record after this check invalidates our
+ * slot, so this slot creation fails afterwards anyway (by a recovery
+ * conflict or the requirement re-check in
+ * CreateInitDecodingContext()). Hence, this check only needs to catch
+ * deactivations that completed before our slot's in_use flag was set.
*/
- Assert(IsLogicalDecodingEnabled());
+ if (!IsLogicalDecodingEnabled())
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("logical decoding on standby requires \"effective_wal_level\" >= \"logical\" on the primary"),
+ errdetail("Logical decoding was concurrently disabled during the logical replication slot creation.")));
+
return;
}
*
* Note that this function does not verify whether logical slots exist. The
* checkpointer will verify if logical decoding should actually be disabled.
+ *
+ * This may be called during recovery, for example when a standby invalidates
+ * its last valid logical slot. That is safe because the queued request is only
+ * acted upon outside recovery. See the RecoveryInProgress() check in
+ * DisableLogicalDecodingIfNecessary().
*/
void
RequestDisableLogicalDecoding(void)
*/
Assert(!MyReplicationSlot);
+ /*
+ * During recovery the logical decoding status follows the primary via WAL
+ * replay, so we must not disable it here. A pending_disable request
+ * queued during recovery, for example by a local slot invalidation, is
+ * intentionally left for the end-of-recovery transition or the
+ * post-promotion checkpointer to act on. See
+ * UpdateLogicalDecodingStatusEndOfRecovery().
+ */
if (RecoveryInProgress())
return;
* already occur due to the checkpointer's asynchronous deactivation
* process.
*
- * For 'disable' case, backend cannot create logical replication slots
- * during recovery (see checks in CheckLogicalDecodingRequirements()),
- * which prevents a race condition between disabling logical decoding and
- * concurrent slot creation.
+ * For 'disable' case, a backend concurrently creating a logical slot on a
+ * standby could have passed its CheckLogicalDecodingRequirements() check
+ * when creating its slot only after our slot check above. Such a backend
+ * rechecks the status after creating the slot in
+ * EnsureLogicalDecodingEnabled() and raises an error if logical decoding
+ * has been disabled meanwhile, so it cannot end up with a logical slot
+ * while logical decoding remains disabled. (If recovery has already ended
+ * by the time of the recheck, the backend instead enables logical
+ * decoding by itself, which is fine after promotion.)
*/
if (new_status != LogicalDecodingCtl->logical_decoding_enabled)
{
#include "pgstat.h"
#include "postmaster/interrupt.h"
#include "replication/logical.h"
+#include "replication/logicalctl.h"
#include "replication/slotsync.h"
#include "replication/snapbuild.h"
#include "storage/ipc.h"
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,
wait_for_logical_decoding_disabled($primary);
pass("the activation process aborted");
+ $psql_create_slot->quit;
+
# Test concurrent activation processes run and one is interrupted.
$psql_create_slot = $primary->background_psql('postgres');
test_wal_level($primary, "replica|logical",
"effective_wal_level remains 'logical' even after the concurrent activation is interrupted"
);
+
+ $psql_create_slot->quit;
+
+ # Test races between the deactivation of logical decoding and slot
+ # creation during recovery. Logical decoding can be deactivated while a
+ # logical slot is being created on a standby, either by replaying an
+ # XLOG_LOGICAL_DECODING_STATUS_CHANGE record or by the end-of-recovery
+ # transition upon promotion. Slot creation paths must detect the
+ # deactivation after creating their slot and fail cleanly, instead
+ # of leaving a slot that cannot be decoded.
+
+ # Restore the disabled state, and disable autovacuum so that the primary
+ # assigns no XIDs, keeping the slot synchronization test below
+ # deterministic.
+ $primary->safe_psql('postgres',
+ qq[select pg_drop_replication_slot('test_slot2')]);
+ wait_for_logical_decoding_disabled($primary);
+ $primary->safe_psql(
+ 'postgres', qq[
+alter system set autovacuum = off;
+select pg_reload_conf();
+]);
+
+ # Initialize standby5 node, setting up the requirements for slot
+ # synchronization.
+ $primary->safe_psql('postgres',
+ qq[select pg_create_physical_replication_slot('phys_slot')]);
+ my $standby5 = PostgreSQL::Test::Cluster->new('standby5');
+ $standby5->init_from_backup($primary, 'my_backup', has_streaming => 1);
+ my $connstr = $primary->connstr;
+ $standby5->append_conf(
+ 'postgresql.conf', qq[
+primary_slot_name = 'phys_slot'
+primary_conninfo = '$connstr dbname=postgres'
+hot_standby_feedback = on
+]);
+ $standby5->start;
+
+ # Test that slot synchronization doesn't persist a slot whose remote
+ # slot information was fetched before a deactivation was replayed.
+ # Otherwise, the slot would survive with a restart_lsn preceding the
+ # deactivation, as the slot invalidation performed by the replay cannot
+ # find the slot that is not created yet.
+
+ # Enable logical decoding by creating a failover slot, and wait for the
+ # standby to replay the activation.
+ $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");
+
+ # Start slot synchronization, stopping it after fetching the remote slot
+ # information but before creating the local slot.
+ my $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 the last logical slot on the primary and wait for the standby to
+ # replay the deactivation.
+ $primary->safe_psql('postgres',
+ qq[select pg_drop_replication_slot('sync_slot')]);
+ wait_for_logical_decoding_disabled($primary);
+ $primary->wait_for_replay_catchup($standby5);
+ test_wal_level($standby5, "replica|replica",
+ "logical decoding got deactivated on standby5");
+
+ # Resume the slot synchronization; it must skip persisting the slot.
+ $log_offset = -s $standby5->logfile;
+ $standby5->safe_psql('postgres',
+ qq[select injection_points_wakeup('replication-slot-create-begin')]);
+ $standby5->wait_for_log(
+ qr/could not synchronize replication slot "sync_slot"/, $log_offset);
+ $psql_sync_slot->quit;
+ is( $standby5->safe_psql(
+ 'postgres', qq[select count(*) from pg_replication_slots]),
+ '0',
+ "no synced slot is left behind on standby5");
+
+ # Test that logical slot creation on a standby fails cleanly if logical
+ # decoding is concurrently deactivated by the end-of-recovery transition
+ # upon promotion.
+
+ # Enable logical decoding again.
+ $primary->safe_psql('postgres',
+ qq[select pg_create_logical_replication_slot('test_slot3', 'test_decoding')]
+ );
+ $primary->wait_for_replay_catchup($standby5);
+ test_wal_level($standby5, "replica|logical",
+ "logical decoding got activated again on standby5");
+
+ # Hold the startup process right after the end-of-recovery status
+ # change, before recovery actually ends.
+ $standby5->safe_psql('postgres',
+ qq[select injection_points_attach('startup-logical-decoding-status-change-end-of-recovery', 'wait')]
+ );
+
+ # Start creating a logical slot, stopping it before creating the slot.
+ $psql_create_slot =
+ $standby5->background_psql('postgres', on_error_stop => 0);
+ $psql_create_slot->query_until(
+ qr/create_standby5_slot/,
+ q(\echo create_standby5_slot
+select injection_points_set_local();
+select injection_points_attach('replication-slot-create-begin', 'wait');
+select pg_create_logical_replication_slot('standby5_slot', 'test_decoding');
+));
+ $standby5->wait_for_event('client backend',
+ 'replication-slot-create-begin');
+ note("injection_point 'replication-slot-create-begin' is reached");
+
+ # Promote the standby without waiting; the startup process deactivates
+ # logical decoding as no logical slot, and stops at the injection point,
+ # still in recovery.
+ $standby5->safe_psql('postgres', qq[select pg_promote(false)]);
+ $standby5->wait_for_event('startup',
+ 'startup-logical-decoding-status-change-end-of-recovery');
+
+ # Resume the slot creation; it must fail cleanly.
+ $log_offset = -s $standby5->logfile;
+ $standby5->safe_psql('postgres',
+ qq[select injection_points_wakeup('replication-slot-create-begin')]);
+ $standby5->wait_for_log(
+ qr/ERROR: .* logical decoding on standby requires "effective_wal_level" >= "logical" on the primary/,
+ $log_offset);
+ $psql_create_slot->quit;
+ is( $standby5->safe_psql(
+ 'postgres', qq[select count(*) from pg_replication_slots]),
+ '0',
+ "no slot is left behind on standby5");
+
+ # Let the promotion complete.
+ $standby5->safe_psql('postgres',
+ qq[select injection_points_wakeup('startup-logical-decoding-status-change-end-of-recovery')]
+ );
+ $standby5->safe_psql('postgres',
+ qq[select injection_points_detach('startup-logical-decoding-status-change-end-of-recovery')]
+ );
+ $standby5->poll_query_until('postgres',
+ qq[select not pg_is_in_recovery()])
+ or die "timed out waiting for promotion";
+
+ # Retrying the slot creation on the promoted standby5 must succeed and
+ # activate logical decoding.
+ $standby5->safe_psql('postgres',
+ qq[select pg_create_logical_replication_slot('standby5_slot', 'test_decoding')]
+ );
+ test_wal_level($standby5, "replica|logical",
+ "logical decoding got activated on the promoted standby5 after retry"
+ );
}
$primary->stop;
projects / thirdparty / postgresql.git / commitdiff
