src/backend/replication/logical/logicalctl.c - master - balmaster/postgres

src/backend/replication/logical/logicalctl.c
644 строки21 KB
Masahiko Sawada
Fix race when logical decoding activation is concurrently interrupted.
09 июн 2026, 21:19
09 июн 2026, 21:1993a3e68
/*-------------------------------------------------------------------------
 * logicalctl.c
 *		Functionality to control logical decoding status online.
 *
 * This module enables dynamic control of logical decoding availability.
 * Logical decoding becomes active under two conditions: when the wal_level
 * parameter is set to 'logical', or when at least one valid logical replication
 * slot exists with wal_level set to 'replica'. The system disables logical
 * decoding when neither condition is met. Therefore, the dynamic control
 * of logical decoding availability is required only when wal_level is set
 * to 'replica'. Logical decoding is always enabled when wal_level='logical'
 * and always disabled when wal_level='minimal'.
 *
 * The core concept of dynamically enabling and disabling logical decoding
 * is to separately control two aspects: writing information required for
 * logical decoding to WAL records, and using logical decoding itself. During
 * activation, we first enable logical WAL writing while keeping logical
 * decoding disabled. This change is reflected in the read-only
 * effective_wal_level GUC parameter. Once we ensure that all processes have
 * updated to the latest effective_wal_level value, we then enable logical
 * decoding. Deactivation follows a similar careful, multi-step process
 * in reverse order.
 *
 * While activation occurs synchronously right after creating the first
 * logical slot, deactivation happens asynchronously through the checkpointer
 * process. This design avoids a race condition at the end of recovery; see
 * the comments in UpdateLogicalDecodingStatusEndOfRecovery() for details.
 * Asynchronous deactivation also avoids excessive toggling of the logical
 * decoding status in workloads that repeatedly create and drop a single
 * logical slot. On the other hand, this lazy approach can delay changes
 * to effective_wal_level and the disabling logical decoding, especially
 * when the checkpointer is busy with other tasks. We chose this lazy approach
 * in all deactivation paths to keep the implementation simple, even though
 * laziness is strictly required only for end-of-recovery cases. Future work
 * might address this limitation either by using a dedicated worker instead
 * of the checkpointer, or by implementing synchronous waiting during slot
 * drops if workloads are significantly affected by the lazy deactivation
 * of logical decoding.
 *
 * Standby servers use the primary server's effective_wal_level and logical
 * decoding status. Unlike normal activation and deactivation, these
 * are updated simultaneously without status change coordination, solely by
 * replaying XLOG_LOGICAL_DECODING_STATUS_CHANGE records. The local wal_level
 * setting has no effect during this time. Upon promotion, we update the
 * logical decoding status based on local conditions: the wal_level value and
 * the presence of logical slots.
 *
 * In the future, we could extend support to include automatic transitions
 * of effective_wal_level between 'minimal' and 'logical' WAL levels. However,
 * this enhancement would require additional coordination mechanisms and
 * careful implementation of operations such as terminating walsenders and
 * archiver processes while carefully considering the sequence of operations
 * to ensure system stability during these transitions.
 *
 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * IDENTIFICATION
 * 	  src/backend/replication/logical/logicalctl.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "access/xloginsert.h"
#include "catalog/pg_control.h"
#include "miscadmin.h"
#include "replication/slot.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/procsignal.h"
#include "storage/subsystems.h"
#include "utils/injection_point.h"

/*
 * Struct for controlling the logical decoding status.
 *
 * This struct is protected by LogicalDecodingControlLock.
 */
typedef struct LogicalDecodingCtlData
{
	/*
	 * This is the authoritative value used by all processes to determine
	 * whether to write additional information required by logical decoding to
	 * WAL. Since this information could be checked frequently, each process
	 * caches this value in XLogLogicalInfo for better performance.
	 */
	bool		xlog_logical_info;

	/* True if logical decoding is available in the system */
	bool		logical_decoding_enabled;

	/* True if logical decoding might need to be disabled */
	bool		pending_disable;
} LogicalDecodingCtlData;

static LogicalDecodingCtlData *LogicalDecodingCtl = NULL;

static void LogicalDecodingCtlShmemRequest(void *arg);

const ShmemCallbacks LogicalDecodingCtlShmemCallbacks = {
	.request_fn = LogicalDecodingCtlShmemRequest,
};

/*
 * A process-local cache of LogicalDecodingCtl->xlog_logical_info. This is
 * initialized at process startup, and updated when processing the process
 * barrier signal in ProcessBarrierUpdateXLogLogicalInfo(). If the process
 * is in an XID-assigned transaction, the cache update is delayed until the
 * transaction ends. See the comments for XLogLogicalInfoUpdatePending for details.
 */
bool		XLogLogicalInfo = false;

/*
 * When receiving the PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO signal, if
 * an XID is assigned to the current transaction, the process sets this flag and
 * delays the XLogLogicalInfo update until the transaction ends. This ensures
 * that the XLogLogicalInfo value (typically accessed via XLogLogicalInfoActive)
 * remains consistent throughout the transaction.
 */
static bool XLogLogicalInfoUpdatePending = false;

static void update_xlog_logical_info(void);
static void abort_logical_decoding_activation(int code, Datum arg);
static void write_logical_decoding_status_update_record(bool status);

static void
LogicalDecodingCtlShmemRequest(void *arg)
{
	ShmemRequestStruct(.name = "Logical decoding control",
					   .size = sizeof(LogicalDecodingCtlData),
					   .ptr = (void **) &LogicalDecodingCtl,
		);
}

/*
 * Initialize the logical decoding status in shmem at server startup. This
 * must be called ONCE during postmaster or standalone-backend startup.
 */
void
StartupLogicalDecodingStatus(bool last_status)
{
	/* Logical decoding is always disabled when 'minimal' WAL level */
	if (wal_level == WAL_LEVEL_MINIMAL)
		return;

	/*
	 * Set the initial logical decoding status based on the last status. If
	 * logical decoding was enabled before the last shutdown, it remains
	 * enabled as we might have set wal_level='logical' or have at least one
	 * logical slot.
	 */
	LogicalDecodingCtl->xlog_logical_info = last_status;
	LogicalDecodingCtl->logical_decoding_enabled = last_status;
}

/*
 * Update the XLogLogicalInfo cache.
 */
static inline void
update_xlog_logical_info(void)
{
	XLogLogicalInfo = IsXLogLogicalInfoEnabled();
}

/*
 * Initialize XLogLogicalInfo backend-private cache. This routine is called
 * during process initialization.
 */
void
InitializeProcessXLogLogicalInfo(void)
{
	update_xlog_logical_info();
}

/*
 * This routine is called when we are told to update XLogLogicalInfo
 * by a ProcSignalBarrier.
 */
bool
ProcessBarrierUpdateXLogLogicalInfo(void)
{
	if (GetTopTransactionIdIfAny() != InvalidTransactionId)
	{
		/* Delay updating XLogLogicalInfo until the transaction end */
		XLogLogicalInfoUpdatePending = true;
	}
	else
		update_xlog_logical_info();

	return true;
}

/*
 * Check the shared memory state and return true if logical decoding is
 * enabled on the system.
 */
bool
IsLogicalDecodingEnabled(void)
{
	bool		enabled;

	LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
	enabled = LogicalDecodingCtl->logical_decoding_enabled;
	LWLockRelease(LogicalDecodingControlLock);

	return enabled;
}

/*
 * Returns true if logical WAL logging is enabled based on the shared memory
 * status.
 */
bool
IsXLogLogicalInfoEnabled(void)
{
	bool		xlog_logical_info;

	LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
	xlog_logical_info = LogicalDecodingCtl->xlog_logical_info;
	LWLockRelease(LogicalDecodingControlLock);

	return xlog_logical_info;
}

/*
 * Reset the local cache at end of the transaction.
 */
void
AtEOXact_LogicalCtl(void)
{
	/* Update the local cache if there is a pending update */
	if (XLogLogicalInfoUpdatePending)
	{
		update_xlog_logical_info();
		XLogLogicalInfoUpdatePending = false;
	}
}

/*
 * Writes an XLOG_LOGICAL_DECODING_STATUS_CHANGE WAL record with the given
 * status.
 */
static void
write_logical_decoding_status_update_record(bool status)
{
	XLogRecPtr	recptr;

	XLogBeginInsert();
	XLogRegisterData(&status, sizeof(bool));
	recptr = XLogInsert(RM_XLOG_ID, XLOG_LOGICAL_DECODING_STATUS_CHANGE);
	XLogFlush(recptr);
}

/*
 * A PG_ENSURE_ERROR_CLEANUP callback for activating logical decoding.
 *
 * Rather than directly reverting xlog_logical_info here, we request
 * that the checkpointer handle it via the normal disable path. This
 * avoids race conditions when multiple backends attempt concurrent
 * activation: the checkpointer will reset xlog_logical_info when
 * no valid logical slots exist.
 */
static void
abort_logical_decoding_activation(int code, Datum arg)
{
	elog(DEBUG1, "aborting logical decoding activation process");
	RequestDisableLogicalDecoding();
}

/*
 * 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.
 *
 * Note that there is no interlock between logical decoding activation
 * and slot creation. To ensure enabling logical decoding, the caller
 * needs to call this function after creating a logical slot before
 * initializing the logical decoding context.
 */
void
EnsureLogicalDecodingEnabled(void)
{
	Assert(MyReplicationSlot);
	Assert(wal_level >= WAL_LEVEL_REPLICA);

	/* Logical decoding is always enabled */
	if (wal_level >= WAL_LEVEL_LOGICAL)
		return;

	if (RecoveryInProgress())
	{
		/*
		 * CheckLogicalDecodingRequirements() must have already errored out if
		 * logical decoding is not enabled since we cannot enable the logical
		 * decoding status during recovery.
		 */
		Assert(IsLogicalDecodingEnabled());
		return;
	}

	/*
	 * Ensure to abort the activation process in cases where there in an
	 * interruption during the wait.
	 */
	PG_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0);
	{
		EnableLogicalDecoding();
	}
	PG_END_ENSURE_ERROR_CLEANUP(abort_logical_decoding_activation, (Datum) 0);
}

/*
 * A workhorse function to enable logical decoding.
 */
void
EnableLogicalDecoding(void)
{
	bool		in_recovery;

	LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);

	/* Return if it is already enabled */
	if (LogicalDecodingCtl->logical_decoding_enabled)
	{
		LogicalDecodingCtl->pending_disable = false;
		LWLockRelease(LogicalDecodingControlLock);
		return;
	}

	/*
	 * Set logical info WAL logging in shmem. All process starts after this
	 * point will include the information required by logical decoding to WAL
	 * records.
	 */
	LogicalDecodingCtl->xlog_logical_info = true;

	LWLockRelease(LogicalDecodingControlLock);

	/*
	 * Tell all running processes to reflect the xlog_logical_info update, and
	 * wait. This ensures that all running processes have enabled logical
	 * information WAL logging.
	 */
	WaitForProcSignalBarrier(
							 EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO));

	INJECTION_POINT("logical-decoding-activation", NULL);

	in_recovery = RecoveryInProgress();

	/*
	 * There could be some transactions that might have started with the old
	 * status, but we don't need to wait for these transactions to complete as
	 * long as they have valid XIDs. These transactions will appear in the
	 * xl_running_xacts record and therefore the snapshot builder will not try
	 * to decode the transaction during the logical decoding initialization.
	 *
	 * There is a theoretical case where a transaction decides whether to
	 * include logical-info to WAL records before getting an XID. In this
	 * case, the transaction won't appear in xl_running_xacts.
	 *
	 * For operations that do not require an XID assignment, the process
	 * starts including logical-info immediately upon receiving the signal
	 * (barrier). If such an operation checks the effective_wal_level multiple
	 * times within a single execution, the resulting WAL records might be
	 * inconsistent (i.e., logical-info is included in some records but not in
	 * others). However, this is harmless because logical decoding generally
	 * ignores WAL records that are not associated with an assigned XID.
	 *
	 * One might think we need to wait for all running transactions, including
	 * those without XIDs and read-only transactions, to finish before
	 * enabling logical decoding. However, such a requirement would force the
	 * slot creation to wait for a potentially very long time due to
	 * long-running read queries, which is practically unacceptable.
	 */

	LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);

	START_CRIT_SECTION();

	/*
	 * We enable logical decoding first, followed by writing the WAL record.
	 * This sequence ensures logical decoding becomes available on the primary
	 * first.
	 */
	LogicalDecodingCtl->logical_decoding_enabled = true;

	if (!in_recovery)
		write_logical_decoding_status_update_record(true);

	LogicalDecodingCtl->pending_disable = false;

	END_CRIT_SECTION();

	LWLockRelease(LogicalDecodingControlLock);

	/*
	 * We log the activation message after releasing the slot lock. This is
	 * safe because the activation is performed while holding a logical slot,
	 * meaning, a concurrent deactivation cannot interleave its log message
	 * ahead of ours.
	 */
	if (!in_recovery)
		ereport(LOG,
				errmsg("logical decoding is enabled upon creating a new logical replication slot"));
}

/*
 * Initiate a request for disabling logical decoding.
 *
 * Note that this function does not verify whether logical slots exist. The
 * checkpointer will verify if logical decoding should actually be disabled.
 */
void
RequestDisableLogicalDecoding(void)
{
	if (wal_level != WAL_LEVEL_REPLICA)
		return;

	/*
	 * It's possible that we might not actually need to disable logical
	 * decoding if someone creates a new logical slot concurrently. We set the
	 * flag anyway and the checkpointer will check it and disable logical
	 * decoding if necessary.
	 */
	LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);
	LogicalDecodingCtl->pending_disable = true;
	LWLockRelease(LogicalDecodingControlLock);

	WakeupCheckpointer();

	elog(DEBUG1, "requested disabling logical decoding");
}

/*
 * Disable logical decoding if necessary.
 *
 * This function disables logical decoding upon a request initiated by
 * RequestDisableLogicalDecoding(). Otherwise, it performs no action.
 */
void
DisableLogicalDecodingIfNecessary(void)
{
	bool		pending_disable;

	if (wal_level != WAL_LEVEL_REPLICA)
		return;

	/*
	 * Sanity check as we cannot disable logical decoding while holding a
	 * logical slot.
	 */
	Assert(!MyReplicationSlot);

	if (RecoveryInProgress())
		return;

	LWLockAcquire(LogicalDecodingControlLock, LW_SHARED);
	pending_disable = LogicalDecodingCtl->pending_disable;
	LWLockRelease(LogicalDecodingControlLock);

	/* Quick return if no pending disable request */
	if (!pending_disable)
		return;

	DisableLogicalDecoding();
}

/*
 * A workhorse function to disable logical decoding.
 */
void
DisableLogicalDecoding(void)
{
	bool		in_recovery = RecoveryInProgress();
	bool		was_enabled;

	LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);

	/*
	 * Check if we can disable logical decoding.
	 *
	 * Nothing to do if both flags are already off, or if valid slots exist
	 * (skip the slot check during recovery because the existing slots will be
	 * invalidated after disabling logical decoding.)
	 */
	if ((!LogicalDecodingCtl->logical_decoding_enabled &&
		 !LogicalDecodingCtl->xlog_logical_info) ||
		(!in_recovery && CheckLogicalSlotExists()))
	{
		LogicalDecodingCtl->pending_disable = false;
		LWLockRelease(LogicalDecodingControlLock);
		return;
	}

	/*
	 * Remember if logical decoding was enabled. An interrupted activation can
	 * leave xlog_logical_info=true while logical_decoding_enabled remains
	 * false.
	 */
	was_enabled = LogicalDecodingCtl->logical_decoding_enabled;

	START_CRIT_SECTION();

	/*
	 * We need to disable logical decoding first and then disable logical
	 * information WAL logging in order to ensure that no logical decoding
	 * processes WAL records with insufficient information.
	 */
	LogicalDecodingCtl->logical_decoding_enabled = false;

	/* Write the WAL to disable logical decoding on standbys too */
	if (!in_recovery && was_enabled)
		write_logical_decoding_status_update_record(false);

	/* Now disable logical information WAL logging */
	LogicalDecodingCtl->xlog_logical_info = false;
	LogicalDecodingCtl->pending_disable = false;

	END_CRIT_SECTION();

	/*
	 * Logging under the lock guarantees our "is disabled" message appears in
	 * the server log before its eventual "is enabled", making server log
	 * diagnostics easy.
	 */
	if (!in_recovery && was_enabled)
		ereport(LOG,
				errmsg("logical decoding is disabled because there are no valid logical replication slots"));

	LWLockRelease(LogicalDecodingControlLock);

	/*
	 * Tell all running processes to reflect the xlog_logical_info update.
	 * Unlike when enabling logical decoding, we don't need to wait for all
	 * processes to complete it in this case. We already disabled logical
	 * decoding and it's always safe to write logical information to WAL
	 * records, even when not strictly required. Therefore, we don't need to
	 * wait for all running transactions to finish either.
	 */
	EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO);
}

/*
 * Updates the logical decoding status at end of recovery, and ensures that
 * all running processes have the updated XLogLogicalInfo status. This
 * function must be called before accepting writes.
 */
void
UpdateLogicalDecodingStatusEndOfRecovery(void)
{
	bool		new_status = false;

	Assert(RecoveryInProgress());

	/*
	 * With 'minimal' WAL level, there are no logical replication slots during
	 * recovery. Logical decoding is always disabled, so there is no need to
	 * synchronize XLogLogicalInfo.
	 */
	if (wal_level == WAL_LEVEL_MINIMAL)
	{
		Assert(!IsXLogLogicalInfoEnabled() && !IsLogicalDecodingEnabled());
		return;
	}

	LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);

	if (wal_level == WAL_LEVEL_LOGICAL || CheckLogicalSlotExists())
		new_status = true;

	/*
	 * When recovery ends, we need to either enable or disable logical
	 * decoding based on the wal_level setting and the presence of logical
	 * slots. We need to note that concurrent slot creation and deletion could
	 * happen but WAL writes are still not permitted until recovery fully
	 * completes. Here's how we handle concurrent toggling of logical
	 * decoding:
	 *
	 * For 'enable' case, if there's a concurrent disable request before
	 * recovery fully completes, the checkpointer will handle it after
	 * recovery is done. This means there might be a brief period after
	 * recovery where logical decoding remains enabled even with no logical
	 * replication slots present. This temporary state is not new - it can
	 * 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.
	 */
	if (new_status != LogicalDecodingCtl->logical_decoding_enabled)
	{
		/*
		 * Update both the logical decoding status and logical WAL logging
		 * status. Unlike toggling these status during non-recovery, we don't
		 * need to worry about the operation order as WAL writes are still not
		 * permitted.
		 */
		LogicalDecodingCtl->xlog_logical_info = new_status;
		LogicalDecodingCtl->logical_decoding_enabled = new_status;

		elog(DEBUG1,
			 "update logical decoding status to %d at the end of recovery",
			 new_status);

		/*
		 * Now that we updated the logical decoding status, clear the pending
		 * disable flag. It's possible that a concurrent process drops the
		 * last logical slot and initiates the pending disable again. The
		 * checkpointer process will check it.
		 */
		LogicalDecodingCtl->pending_disable = false;

		LWLockRelease(LogicalDecodingControlLock);

		write_logical_decoding_status_update_record(new_status);
	}
	else
		LWLockRelease(LogicalDecodingControlLock);

	/*
	 * Ensure all running processes have the updated status. We don't need to
	 * wait for running transactions to finish as we don't accept any writes
	 * yet. On the other hand, we need to wait for synchronizing
	 * XLogLogicalInfo even if we've not updated the status above as the
	 * status have been turned on and off during recovery, having running
	 * processes have different status on their local caches.
	 */
	if (IsUnderPostmaster)
		WaitForProcSignalBarrier(
								 EmitProcSignalBarrier(PROCSIGNAL_BARRIER_UPDATE_XLOG_LOGICAL_INFO));

	INJECTION_POINT("startup-logical-decoding-status-change-end-of-recovery", NULL);
}