/*-------------------------------------------------------------------------

 * 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 of 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 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

 * 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())

  {

    /*

     * 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.

     */

    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;

  }

  /*

   * Ensure to abort the activation process in cases where there is 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 processes starting after

   * this point will include the information required by logical decoding in

   * 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 in 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);

  /*

   * Re-check whether logical decoding got enabled while we waited for the

   * barrier above.

   */

  if (LogicalDecodingCtl->logical_decoding_enabled)

  {

    LogicalDecodingCtl->pending_disable = false;

    LWLockRelease(LogicalDecodingControlLock);

    return;

  }

  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.

 *

 * 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)

{

  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);

  /*

   * 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;

  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());

  LWLockAcquire(LogicalDecodingControlLock, LW_EXCLUSIVE);

  /*

   * With 'minimal' WAL level, no logical replication slot can exist (see

   * RestoreSlotFromDisk()), so the new status is always false. However,

   * logical decoding could have been enabled during recovery by replaying

   * an XLOG_LOGICAL_DECODING_STATUS_CHANGE record from WAL generated with a

   * higher wal_level, e.g. if the server crashed right after the last

   * logical slot was dropped and then restarted with wal_level='minimal'.

   * The code below disables logical decoding in that case.

   */

  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, 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)

  {

    /*

     * 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);

}