/src/postgres/src/backend/postmaster/datachecksum_state.c

Source
/*-------------------------------------------------------------------------
 *
 * datachecksum_state.c
 *    Background worker for enabling or disabling data checksums online as
 *    well as functionality for manipulating data checksum state
 *
 * When enabling data checksums on a cluster at initdb time or when shut down
 * with pg_checksums, no extra process is required as each page is checksummed,
 * and verified, when accessed.  When enabling checksums on an already running
 * cluster, this worker will ensure that all pages are checksummed before
 * verification of the checksums is turned on. In the case of disabling
 * checksums, the state transition is performed only in the control file, no
 * changes are performed on the data pages.
 *
 * Checksums can be either enabled or disabled cluster-wide, with on/off being
 * the end state for data_checksums.
 *
 * 1. Enabling checksums
 * ---------------------
 * When enabling checksums in an online cluster, data_checksums will be set to
 * "inprogress-on" which signals that write operations MUST compute and write
 * the checksum on the data page, but during reading the checksum SHALL NOT be
 * verified. This ensures that all objects created while checksums are being
 * enabled will have checksums set, but reads won't fail due to missing or
 * invalid checksums. Invalid checksums can be present in case the cluster had
 * checksums enabled, then disabled them and updated the page while they were
 * disabled.
 *
 * The DataChecksumsWorker will compile a list of all databases at the start,
 * any databases created concurrently will see the in-progress state and will
 * be checksummed automatically.  All databases from the original list MUST BE
 * successfully processed in order for data checksums to be enabled, the only
 * exception are databases which are dropped before having been processed.
 *
 * For each database, all relations which have storage are read and every data
 * page is marked dirty to force a write with the checksum. This will generate
 * a lot of WAL as the entire database is read and written.
 *
 * If the processing is interrupted by a cluster crash or restart, it needs to
 * be restarted from the beginning again as state isn't persisted.
 *
 * 2. Disabling checksums
 * ----------------------
 * When disabling checksums, data_checksums will be set to "inprogress-off"
 * which signals that checksums are written but no longer need to be verified.
 * This ensures that backends which have not yet transitioned to the
 * "inprogress-off" state will still see valid checksums on pages.
 *
 * 3. Synchronization and Correctness
 * ----------------------------------
 * The processes involved in enabling or disabling data checksums in an
 * online cluster must be properly synchronized with the normal backends
 * serving concurrent queries to ensure correctness. Correctness is defined
 * as the following:
 *
 *    - Backends SHALL NOT violate the data_checksums state they have agreed to
 *      by acknowledging the procsignalbarrier:  This means that all backends
 *      MUST calculate and write data checksums during all states except off;
 *      MUST validate checksums only in the 'on' state.
 *    - Data checksums SHALL NOT be considered enabled cluster-wide until all
 *      currently connected backends have state "on": This means that all
 *      backends must wait on the procsignalbarrier to be acknowledged by all
 *      before proceeding to validate data checksums.
 *
 * There are two steps of synchronization required for changing data_checksums
 * in an online cluster: (i) changing state in the active backends ("on",
 * "off", "inprogress-on" and "inprogress-off"), and (ii) ensuring no
 * incompatible objects and processes are left in a database when workers end.
 * The former deals with cluster-wide agreement on data checksum state and the
 * latter with ensuring that any concurrent activity cannot break the data
 * checksum contract during processing.
 *
 * Synchronizing the state change is done with procsignal barriers. Before
 * updating the data_checksums state in the control file, all other backends must absorb the
 * barrier.  Barrier absorption will happen during interrupt processing, which
 * means that connected backends will change state at different times.  If
 * waiting for a barrier is done during startup, for example during replay, it
 * is important to realize that any locks held by the startup process might
 * cause deadlocks if backends end up waiting for those locks while startup
 * is waiting for a procsignalbarrier.
 *
 * 3.1 When Enabling Data Checksums
 * --------------------------------
 * A process which fails to observe data checksums being enabled can induce two
 * types of errors: failing to write the checksum when modifying the page and
 * failing to validate the data checksum on the page when reading it.
 *
 * When processing starts all backends belong to one of the below sets, with
 * one of Bd and Bi being empty:
 *
 * Bg: Backend updating the global state and emitting the procsignalbarrier
 * Bd: Backends in "off" state
 * Bi: Backends in "inprogress-on" state
 *
 * If processing is started in an online cluster then all backends are in Bd.
 * If processing was halted by the cluster shutting down (due to a crash or
 * intentional restart), the control file state "inprogress-on" will be
 * observed on system startup and all backends will be placed in Bd. The
 * control file state will also be set to "off".
 *
 * Backends transition Bd -> Bi via a procsignalbarrier which is emitted by the
 * DataChecksumsWorkerLauncherMain.  When all backends have acknowledged the
 * barrier then Bd will be empty and the next phase can begin: calculating and
 * writing data checksums with DataChecksumsWorkers.  When the
 * DataChecksumsWorker processes have finished writing checksums on all pages,
 * data checksums are enabled cluster-wide via another procsignalbarrier.
 * There are four sets of backends where Bd shall be an empty set:
 *
 * Bg: Backend updating the global state and emitting the procsignalbarrier
 * Bd: Backends in "off" state
 * Be: Backends in "on" state
 * Bi: Backends in "inprogress-on" state
 *
 * Backends in Bi and Be will write checksums when modifying a page, but only
 * backends in Be will verify the checksum during reading. The Bg backend is
 * blocked waiting for all backends in Bi to process interrupts and move to
 * Be. Any backend starting while Bg is waiting on the procsignalbarrier will
 * observe the global state being "on" and will thus automatically belong to
 * Be.  Checksums are enabled cluster-wide when Bi is an empty set. Bi and Be
 * are compatible sets while still operating based on their local state as
 * both write data checksums.
 *
 * 3.2 When Disabling Data Checksums
 * ---------------------------------
 * A process which fails to observe that data checksums have been disabled
 * can induce two types of errors: writing the checksum when modifying the
 * page and validating a data checksum which is no longer correct due to
 * modifications to the page. The former is not an error per se as data
 * integrity is maintained, but it is wasteful.  The latter will cause errors
 * in user operations.  Assuming the following sets of backends:
 *
 * Bg: Backend updating the global state and emitting the procsignalbarrier
 * Bd: Backends in "off" state
 * Be: Backends in "on" state
 * Bo: Backends in "inprogress-off" state
 * Bi: Backends in "inprogress-on" state
 *
 * Backends transition from the Be state to Bd like so: Be -> Bo -> Bd.  From
 * all other states, the transition can be straight to Bd.
 *
 * The goal is to transition all backends to Bd making the others empty sets.
 * Backends in Bo write data checksums, but don't validate them, such that
 * backends still in Be can continue to validate pages until the barrier has
 * been absorbed such that they are in Bo. Once all backends are in Bo, the
 * barrier to transition to "off" can be raised and all backends can safely
 * stop writing data checksums as no backend is enforcing data checksum
 * validation any longer.
 *
 * 4. Interaction with offline data checksum changes
 * -------------------------------------------------
 * Enabling or disabling checksums offline with pg_checksums uses none of the
 * machinery in this file, but the two mechanisms share the state kept in the
 * control file, so their interaction is documented here.
 *
 * pg_checksums writes the new state to the control file and sets
 * data_checksum_is_local, marking a state that no WAL record accounts for.
 * Recovery then does not adopt the state carried by a replayed checkpoint
 * record over it.  The control file also carries a watermark, the WAL
 * position through which data checksum transitions are covered.  Replay skips
 * transition records ending at or below the watermark, as their effect is
 * already contained in the control file, and applies records above it as
 * usual, whether they were written before or after an offline change.  This
 * is why an offline change in a replicated setup must be made on every node
 * while all of them are stopped and caught up; see the pg_checksums
 * documentation for the procedure.
 *
 * 5. Future opportunities for optimizations
 * -----------------------------------------
 * Below are some potential optimizations and improvements which were brought
 * up during reviews of this feature, but which weren't implemented in the
 * initial version. These are ideas listed without any validation on their
 * feasibility or potential payoff. More discussion on (most of) these can be
 * found on the -hackers threads linked to in the commit message of this
 * feature.
 *
 *   * Launching datachecksumsworker for resuming operation from the startup
 *     process: Currently users have to restart processing manually after a
 *     restart since dynamic background worker cannot be started from the
 *     postmaster. Changing the startup process could make restarting the
 *     processing automatic on cluster restart.
 *   * Avoid dirtying the page when checksums already match: Iff the checksum
 *     on the page happens to already match we still dirty the page. It should
 *     be enough to only do the log_newpage_buffer() call in that case.
 *   * Teach pg_checksums to avoid checksummed pages when pg_checksums is used
 *     to enable checksums on a cluster which is in inprogress-on state and
 *     may have checksummed pages (make pg_checksums be able to resume an
 *     online operation). This should only be attempted for wal_level minimal.
 *   * Restartability (not necessarily with page granularity).
 *   * Avoid processing databases which were created during inprogress-on.
 *     Right now all databases are processed regardless to be safe.
 *   * Teach CREATE DATABASE to calculate checksums for databases created
 *     during inprogress-on with a template database which has yet to be
 *     processed.
 *
 *
 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *    src/backend/postmaster/datachecksum_state.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/genam.h"
#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "access/xloginsert.h"
#include "catalog/indexing.h"
#include "catalog/pg_class.h"
#include "catalog/pg_database.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "common/relpath.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "postmaster/bgworker.h"
#include "postmaster/bgwriter.h"
#include "postmaster/datachecksum_state.h"
#include "storage/bufmgr.h"
#include "storage/checksum.h"
#include "storage/ipc.h"
#include "storage/latch.h"
#include "storage/lmgr.h"
#include "storage/lwlock.h"
#include "storage/procarray.h"
#include "storage/smgr.h"
#include "storage/subsystems.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/injection_point.h"
#include "utils/lsyscache.h"
#include "utils/ps_status.h"
#include "utils/syscache.h"
#include "utils/wait_event.h"

/*
 * Configuration of conditions which must match when absorbing a procsignal
 * barrier during data checksum enable/disable operations.  A single function
 * is used for absorbing all barriers, and the current and target states must
 * be defined as a from/to tuple in the checksum_barriers struct.
 */
typedef struct ChecksumBarrierCondition
{
  /* Current state of data checksums */
  int     from;
  /* Target state for data checksums */
  int     to;
} ChecksumBarrierCondition;

static const ChecksumBarrierCondition checksum_barriers[9] =
{
  /*
   * Disabling checksums: If checksums are currently enabled, disabling must
   * go through the 'inprogress-off' state.
   */
  {PG_DATA_CHECKSUM_VERSION, PG_DATA_CHECKSUM_INPROGRESS_OFF},
  {PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_OFF},

  /*
   * If checksums are in the process of being enabled, but are not yet being
   * verified, we can abort by going back to 'off' state.
   */
  {PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_OFF},

  /*
   * Enabling checksums must normally go through the 'inprogress-on' state.
   */
  {PG_DATA_CHECKSUM_OFF, PG_DATA_CHECKSUM_INPROGRESS_ON},
  {PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_VERSION},

  /*
   * If checksums are being disabled but all backends are still computing
   * checksums, we can go straight back to 'on'
   */
  {PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_VERSION},

  /*
   * If checksums are being enabled when launcher_exit is executed, state is
   * set to off since we cannot reach on at that point.
   */
  {PG_DATA_CHECKSUM_INPROGRESS_ON, PG_DATA_CHECKSUM_INPROGRESS_OFF},

  /*
   * Transitions that can happen when a new request is made while another is
   * currently being processed.
   */
  {PG_DATA_CHECKSUM_INPROGRESS_OFF, PG_DATA_CHECKSUM_INPROGRESS_ON},
  {PG_DATA_CHECKSUM_OFF, PG_DATA_CHECKSUM_INPROGRESS_OFF},
};

/* Possible operations the DataChecksumsWorker can perform */
typedef enum DataChecksumsWorkerOperation
{
  ENABLE_DATACHECKSUMS,
  DISABLE_DATACHECKSUMS,
} DataChecksumsWorkerOperation;

/* Possible states for a database entry which has been processed */
typedef enum
{
  DATACHECKSUMSWORKER_SUCCESSFUL = 0,
  DATACHECKSUMSWORKER_ABORTED,
  DATACHECKSUMSWORKER_FAILED,
  DATACHECKSUMSWORKER_DROPDB,
} DataChecksumsWorkerResult;

/*
 * Signaling between backends calling pg_enable/disable_data_checksums, the
 * checksums launcher process, and the checksums worker process.
 *
 * This struct is protected by DataChecksumsWorkerLock
 */
typedef struct DataChecksumsStateStruct
{
  /*
   * These are set by pg_{enable|disable}_data_checksums, to tell the
   * launcher what the target state is.
   */
  DataChecksumsWorkerOperation launch_operation;
  int     launch_cost_delay;
  int     launch_cost_limit;

  /*
   * Is a launcher process currently running?  This is set by the main
   * launcher process, after it has read the above launch_* parameters.
   */
  bool    launcher_running;

  /*
   * Every time a new worker is launched, it's assigned a unique invocation
   * number by incrementing this counter.
   */
  uint64    worker_invocation_counter;

  /*
   * Information about the current worker, if it's currently running.  These
   * are set by the worker launcher.
   */
  uint64    worker_invocation;  /* unique invocation number */
  Oid     database_oid; /* database it's processing */
  pid_t   worker_pid;   /* worker process's PID */

  /*
   * These fields indicate the target state that the worker is currently
   * running with.  They can be different from the corresponding launch_*
   * fields, if a new pg_enable/disable_data_checksums() call was made while
   * the launcher/worker was already running.  The worker will periodically
   * check if new cost settings have been requested, and if so will copy
   * them from the launch_* fields and reset cost throttling to match the
   * new values.
   */
  DataChecksumsWorkerOperation operation;
  int     cost_delay;
  int     cost_limit;

  /*
   * Signaling between the launcher and the worker process. Protected by
   * DataChecksumsWorkerLock.
   */

  /* result, set by worker before exiting */
  DataChecksumsWorkerResult worker_result;

  /*
   * Tells the worker process whether it should also process the shared
   * catalogs
   */
  bool    process_shared_catalogs;
} DataChecksumsStateStruct;

/* Shared memory segment for datachecksumsworker */
static DataChecksumsStateStruct *DataChecksumState;

typedef struct DataChecksumsWorkerDatabase
{
  Oid     dboid;
  char     *dbname;
} DataChecksumsWorkerDatabase;

/* Flag set by the interrupt handler */
static volatile sig_atomic_t abort_requested = false;

static uint64 worker_invocation;

/*
 * Have we set the DataChecksumsStateStruct->launcher_running flag?
 * If we have, we need to clear it before exiting!
 */
static volatile sig_atomic_t launcher_running = false;

/* Are we enabling data checksums, or disabling them? */
static DataChecksumsWorkerOperation operation;

/* Prototypes */
static void StartDataChecksumsWorkerLauncher(DataChecksumsWorkerOperation op,
                       int cost_delay,
                       int cost_limit);
static void ErrorOnInvalidDatabases(void);
static bool ProcessSingleRelationFork(Relation reln, ForkNumber forkNum, BufferAccessStrategy strategy);
static void ResetDataChecksumsProgressCounters(void);
static bool ProcessSingleRelationByOid(Oid relationId, BufferAccessStrategy strategy);
static BgwHandleStatus WaitForDataChecksumsWorkerState(BackgroundWorkerHandle *handle,
                             bool wait_for_startup,
                             pid_t *pidp,
                             uint32 wait_event);
static DataChecksumsWorkerResult ProcessDatabase(DataChecksumsWorkerDatabase *db);
static void launcher_exit(int code, Datum arg);
static void launcher_cancel_handler(SIGNAL_ARGS);
static void WaitForAllTransactionsToFinish(void);
static bool ProcessAllDatabases(void);
static void DataChecksumsShmemRequest(void *arg);
static bool DatabaseExists(Oid dboid);
static List *BuildDatabaseList(void);
static void FreeDatabaseList(List *dblist);
static List *BuildRelationList(bool temp_relations, bool include_shared);

const ShmemCallbacks DataChecksumsShmemCallbacks = {
  .request_fn = DataChecksumsShmemRequest,
};

#define CHECK_FOR_LAUNCHER_ABORT_REQUEST() \
  do {                             \
    Assert(MyBackendType == B_DATACHECKSUMSWORKER_LAUNCHER);  \
    LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);     \
    if (DataChecksumState->launch_operation != operation)     \
      abort_requested = true;                 \
    LWLockRelease(DataChecksumsWorkerLock);            \
  } while (0)

#define CHECK_FOR_WORKER_ABORT_REQUEST() \
  do {                             \
    Assert(MyBackendType == B_DATACHECKSUMSWORKER_WORKER);    \
    LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);     \
    if (DataChecksumState->worker_invocation != worker_invocation || \
      DataChecksumState->launch_operation != operation)     \
      abort_requested = true;                 \
    LWLockRelease(DataChecksumsWorkerLock);            \
  } while (0)


/*****************************************************************************
 * Functionality for manipulating the data checksum state in the cluster
 */

void
EmitAndWaitDataChecksumsBarrier(uint32 state)
{
  uint64    barrier;

  switch (state)
  {
    case PG_DATA_CHECKSUM_INPROGRESS_ON:
      barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON);
      WaitForProcSignalBarrier(barrier);
      break;

    case PG_DATA_CHECKSUM_INPROGRESS_OFF:
      barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF);
      WaitForProcSignalBarrier(barrier);
      break;

    case PG_DATA_CHECKSUM_VERSION:
      barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_ON);
      WaitForProcSignalBarrier(barrier);
      break;

    case PG_DATA_CHECKSUM_OFF:
      barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_OFF);
      WaitForProcSignalBarrier(barrier);
      break;

    default:
      Assert(false);
  }
}

/*
 * AbsorbDataChecksumsBarrier
 *    Generic function for absorbing data checksum state changes
 *
 * All procsignalbarriers regarding data checksum state changes are absorbed
 * with this function.  The set of conditions required for the state change to
 * be accepted are listed in the checksum_barriers struct, target_state is
 * used to look up the relevant entry.
 */
bool
AbsorbDataChecksumsBarrier(ProcSignalBarrierType barrier)
{
  uint32    target_state;
  int     current = data_checksums;
  bool    found = false;

  /*
   * Translate the barrier condition to the target state, doing it here
   * instead of in the procsignal code saves the latter from knowing about
   * checksum states.
   */
  switch (barrier)
  {
    case PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_ON:
      target_state = PG_DATA_CHECKSUM_INPROGRESS_ON;
      break;
    case PROCSIGNAL_BARRIER_CHECKSUM_ON:
      target_state = PG_DATA_CHECKSUM_VERSION;
      break;
    case PROCSIGNAL_BARRIER_CHECKSUM_INPROGRESS_OFF:
      target_state = PG_DATA_CHECKSUM_INPROGRESS_OFF;
      break;
    case PROCSIGNAL_BARRIER_CHECKSUM_OFF:
      target_state = PG_DATA_CHECKSUM_OFF;
      break;
    default:
      elog(ERROR, "incorrect barrier \"%d\" received", barrier);
  }

  /*
   * If the target state matches the current state then the barrier has been
   * repeated.
   */
  if (current == target_state)
    return true;

  /*
   * If the cluster is in recovery we skip the validation of current state
   * since the replay is trusted.
   */
  if (RecoveryInProgress())
  {
    SetLocalDataChecksumState(target_state);
    return true;
  }

  /*
   * Find the barrier condition definition for the target state. Not finding
   * a condition would be a grave programmer error as the states are a
   * discrete set.
   */
  for (size_t i = 0; i < lengthof(checksum_barriers) && !found; i++)
  {
    if (checksum_barriers[i].from == current && checksum_barriers[i].to == target_state)
      found = true;
  }

  /*
   * If the relevant state criteria aren't satisfied, throw an error which
   * will be caught by the procsignal machinery for a later retry.
   */
  if (!found)
    ereport(ERROR,
        errcode(ERRCODE_INVALID_PARAMETER_VALUE),
        errmsg("incorrect data checksum state %d for target state %d",
             current, target_state));

  SetLocalDataChecksumState(target_state);
  return true;
}


/*
 * Disables data checksums for the cluster, if applicable. Starts a background
 * worker which turns off the data checksums.
 */
Datum
disable_data_checksums(PG_FUNCTION_ARGS)
{
  PreventCommandDuringRecovery("pg_disable_data_checksums()");

  if (!superuser())
    ereport(ERROR,
        errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
        errmsg("must be superuser to change data checksum state"));

  StartDataChecksumsWorkerLauncher(DISABLE_DATACHECKSUMS, 0, 0);
  PG_RETURN_VOID();
}

/*
 * Enables data checksums for the cluster, if applicable.  Supports vacuum-
 * like cost based throttling to limit system load. Starts a background worker
 * which updates data checksums on existing data.
 */
Datum
enable_data_checksums(PG_FUNCTION_ARGS)
{
  int     cost_delay = PG_GETARG_INT32(0);
  int     cost_limit = PG_GETARG_INT32(1);

  PreventCommandDuringRecovery("pg_enable_data_checksums()");

  if (!superuser())
    ereport(ERROR,
        errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
        errmsg("must be superuser to change data checksum state"));

  if (cost_delay < 0)
    ereport(ERROR,
        errcode(ERRCODE_INVALID_PARAMETER_VALUE),
        errmsg("cost delay cannot be a negative value"));

  if (cost_limit <= 0)
    ereport(ERROR,
        errcode(ERRCODE_INVALID_PARAMETER_VALUE),
        errmsg("cost limit must be greater than zero"));

  /*
   * An invalid database cannot be connected to, so the worker would fail to
   * process it, and unlike a dropped database its files stay around.  Error
   * out early with a hint rather than failing halfway through processing. A
   * database which turns invalid after this check, for example from an
   * interrupted DROP DATABASE, instead makes its worker fail; the launcher
   * then aborts and leaves checksums disabled, since the invalid database's
   * files would otherwise be left without valid checksums.
   */
  ErrorOnInvalidDatabases();

  StartDataChecksumsWorkerLauncher(ENABLE_DATACHECKSUMS, cost_delay, cost_limit);

  PG_RETURN_VOID();
}


/*****************************************************************************
 * Functionality for running the datachecksumsworker and associated launcher
 */

/*
 * StartDataChecksumsWorkerLauncher
 *    Start the datachecksumsworker launcher process, if not running yet
 *
 * This is called to start data checksums processing for enabling as well as
 * disabling.
 */
static void
StartDataChecksumsWorkerLauncher(DataChecksumsWorkerOperation op,
                 int cost_delay,
                 int cost_limit)
{
  BackgroundWorker bgw;
  BackgroundWorkerHandle *bgw_handle;
  bool    running;

#ifdef USE_ASSERT_CHECKING
  /* The cost delay settings have no effect when disabling */
  if (op == DISABLE_DATACHECKSUMS)
    Assert(cost_delay == 0 && cost_limit == 0);
#endif

  INJECTION_POINT("datachecksumsworker-startup-delay", NULL);

  /* Store the desired state in shared memory */
  LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);

  DataChecksumState->launch_operation = op;
  DataChecksumState->launch_cost_delay = cost_delay;
  DataChecksumState->launch_cost_limit = cost_limit;

  /* Is the launcher already running? If so, what is it doing? */
  running = DataChecksumState->launcher_running;

  LWLockRelease(DataChecksumsWorkerLock);

  /*
   * Launch a new launcher process, if it's not running already.
   *
   * If the launcher is currently busy enabling the checksums, and we want
   * them disabled (or vice versa), the launcher will notice that at latest
   * when it's about to exit, and will loop back to process the new request.
   * So if the launcher is already running, we don't need to do anything
   * more here to abort it.
   *
   * If you call pg_enable/disable_data_checksums() twice in a row, before
   * the launcher has had a chance to start up, we still end up launching it
   * twice.  That's OK, the second invocation will see that a launcher is
   * already running and exit quickly.
   */
  if (!running)
  {
    if ((op == ENABLE_DATACHECKSUMS && DataChecksumsOn()) ||
      (op == DISABLE_DATACHECKSUMS && DataChecksumsOff()))
    {
      ereport(LOG,
          errmsg("data checksums already in desired state, exiting"));
      return;
    }

    /*
     * Prepare the BackgroundWorker and launch it.
     */
    memset(&bgw, 0, sizeof(bgw));
    bgw.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
    bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
    snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
    snprintf(bgw.bgw_function_name, BGW_MAXLEN, "DataChecksumsWorkerLauncherMain");
    snprintf(bgw.bgw_name, BGW_MAXLEN, "datachecksums launcher");
    snprintf(bgw.bgw_type, BGW_MAXLEN, "datachecksums launcher");
    bgw.bgw_restart_time = BGW_NEVER_RESTART;
    bgw.bgw_notify_pid = MyProcPid;
    bgw.bgw_main_arg = (Datum) 0;

    if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
      ereport(ERROR,
          errcode(ERRCODE_INSUFFICIENT_RESOURCES),
          errmsg("failed to start background worker to process data checksums"));
  }
  else
  {
    ereport(LOG,
        errmsg("data checksum processing already running"));
  }
}

/*
 * ProcessSingleRelationFork
 *    Enable data checksums in a single relation/fork.
 *
 * Returns true if successful, and false if *aborted*. On error, an actual
 * error is raised in the lower levels.
 */
static bool
ProcessSingleRelationFork(Relation reln, ForkNumber forkNum, BufferAccessStrategy strategy)
{
  BlockNumber numblocks = RelationGetNumberOfBlocksInFork(reln, forkNum);
  char    activity[NAMEDATALEN * 2 + 128];
  char     *relns;

  relns = get_namespace_name(RelationGetNamespace(reln));

  /* Report the current relation to pg_stat_activity */
  snprintf(activity, sizeof(activity) - 1, "processing: %s.%s (%s, %u blocks)",
       (relns ? relns : ""), RelationGetRelationName(reln), forkNames[forkNum], numblocks);
  pgstat_report_activity(STATE_RUNNING, activity);
  {
    const int index[] = {
      PROGRESS_DATACHECKSUMS_BLOCKS_TOTAL,
      PROGRESS_DATACHECKSUMS_BLOCKS_DONE
    };

    int64   vals[2];

    vals[0] = numblocks;
    vals[1] = 0;

    pgstat_progress_update_multi_param(2, index, vals);
  }
  if (relns)
    pfree(relns);

  /*
   * We are looping over the blocks which existed at the time of process
   * start, which is safe since new blocks are created with checksums set
   * already due to the state being "inprogress-on".
   */
  for (BlockNumber blknum = 0; blknum < numblocks; blknum++)
  {
    Buffer    buf = ReadBufferExtended(reln, forkNum, blknum, RBM_NORMAL, strategy);

    /* Need to get an exclusive lock to mark the buffer as dirty */
    LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);

    /*
     * Mark the buffer as dirty and force a full page write.  We have to
     * re-write the page to WAL even if the checksum hasn't changed,
     * because if there is a replica it might have a slightly different
     * version of the page with an invalid checksum, caused by unlogged
     * changes (e.g. hint bits) on the primary happening while checksums
     * were off. This can happen if there was a valid checksum on the page
     * at one point in the past, so only when checksums are first on, then
     * off, and then turned on again.  TODO: investigate if this could be
     * avoided if the checksum is calculated to be correct and wal_level
     * is set to "minimal".
     *
     * Unlogged relations don't need WAL since they are reset to their
     * init fork on recovery.  We still dirty the buffer so that the
     * checksum is written to disk at the next checkpoint.
     *
     * The init fork is an exception: it is WAL-logged so the standby can
     * materialize the relation after promotion (see
     * ResetUnloggedRelations()).  Skipping it here would leave the
     * standby with a stale init fork that, once copied to the main fork
     * on promotion, would fail checksum verification on every read.
     */
    START_CRIT_SECTION();
    MarkBufferDirty(buf);
    if (RelationNeedsWAL(reln) || forkNum == INIT_FORKNUM)
      log_newpage_buffer(buf, false);
    END_CRIT_SECTION();

    UnlockReleaseBuffer(buf);

    /* Check if we are asked to abort, the abortion will bubble up. */
    Assert(operation == ENABLE_DATACHECKSUMS);
    CHECK_FOR_WORKER_ABORT_REQUEST();
    if (abort_requested)
      return false;

    /* update the block counter */
    pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_BLOCKS_DONE,
                   (blknum + 1));

    /*
     * Processing is re-using the vacuum cost delay for process
     * throttling, hence why we call vacuum APIs here.
     */
    vacuum_delay_point(false);
  }

  return true;
}

/*
 * Initialize all data checksum progress counters to be displayed as NULL.
 */
static void
ResetDataChecksumsProgressCounters(void)
{
  const int index[] = {
    PROGRESS_DATACHECKSUMS_DBS_TOTAL,
    PROGRESS_DATACHECKSUMS_DBS_DONE,
    PROGRESS_DATACHECKSUMS_RELS_TOTAL,
    PROGRESS_DATACHECKSUMS_RELS_DONE,
    PROGRESS_DATACHECKSUMS_BLOCKS_TOTAL,
    PROGRESS_DATACHECKSUMS_BLOCKS_DONE,
  };

  int64   vals[lengthof(index)];

  for (size_t i = 0; i < lengthof(index); i++)
    vals[i] = -1;

  pgstat_progress_update_multi_param(lengthof(index), index, vals);
}

/*
 * ProcessSingleRelationByOid
 *    Process a single relation based on oid.
 *
 * Returns true if successful, and false if *aborted*. On error, an actual
 * error is raised in the lower levels.
 */
static bool
ProcessSingleRelationByOid(Oid relationId, BufferAccessStrategy strategy)
{
  Relation  rel;
  bool    aborted = false;

  StartTransactionCommand();

  rel = try_relation_open(relationId, AccessShareLock);
  if (rel == NULL)
  {
    /*
     * Relation no longer exists. We don't consider this an error since
     * there are no pages in it that need data checksums, and thus return
     * true. The worker operates off a list of relations generated at the
     * start of processing, so relations being dropped in the meantime is
     * to be expected.
     */
    CommitTransactionCommand();
    pgstat_report_activity(STATE_IDLE, NULL);
    return true;
  }

  for (ForkNumber fnum = 0; fnum <= MAX_FORKNUM; fnum++)
  {
    if (smgrexists(RelationGetSmgr(rel), fnum))
    {
      if (!ProcessSingleRelationFork(rel, fnum, strategy))
      {
        aborted = true;
        break;
      }
    }
  }
  relation_close(rel, AccessShareLock);

  CommitTransactionCommand();
  pgstat_report_activity(STATE_IDLE, NULL);

  return !aborted;
}

/*
 * WaitForDataChecksumsWorkerState
 *    Wait for a data checksums worker to start or stop.
 *
 * This is like WaitForBackgroundWorkerStartup() and
 * WaitForBackgroundWorkerShutdown(), except that it also reacts to SIGINT
 * received by the launcher.  The launcher owns the overall checksum
 * operation, so canceling it should stop the worker it has registered or is
 * currently running.
 *
 * If wait_for_startup is true, wait until the worker is no longer in
 * BGWH_NOT_YET_STARTED state, like WaitForBackgroundWorkerStartup().  If it
 * is false, wait until the worker reaches BGWH_STOPPED state, like
 * WaitForBackgroundWorkerShutdown().
 *
 * pidp is set to the worker's PID when startup succeeds, if it is not NULL.
 */
static BgwHandleStatus
WaitForDataChecksumsWorkerState(BackgroundWorkerHandle *handle,
                bool wait_for_startup,
                pid_t *pidp,
                uint32 wait_event)
{
  BgwHandleStatus status;
  bool    termination_requested = false;

  for (;;)
  {
    int     rc;
    pid_t   pid;

    CHECK_FOR_INTERRUPTS();

    status = GetBackgroundWorkerPid(handle, &pid);
    if (status == BGWH_STARTED && pidp)
      *pidp = pid;

    if (abort_requested && !termination_requested)
    {
      TerminateBackgroundWorker(handle);
      termination_requested = true;
    }

    /*
     * Startup waits for the worker to leave BGWH_NOT_YET_STARTED, while
     * shutdown waits for it to reach BGWH_STOPPED.
     */
    if (status == BGWH_STOPPED ||
      (wait_for_startup && status == BGWH_STARTED))
      break;

    rc = WaitLatch(MyLatch,
             WL_LATCH_SET | WL_POSTMASTER_DEATH, 0,
             wait_event);

    if (rc & WL_POSTMASTER_DEATH)
    {
      status = BGWH_POSTMASTER_DIED;
      break;
    }

    ResetLatch(MyLatch);
  }

  return status;
}

/*
 * ProcessDatabase
 *    Enable data checksums in a single database.
 *
 * We do this by launching a dynamic background worker into this database, and
 * waiting for it to finish.  We have to do this in a separate worker, since
 * each process can only be connected to one database during its lifetime.
 */
static DataChecksumsWorkerResult
ProcessDatabase(DataChecksumsWorkerDatabase *db)
{
  BackgroundWorker bgw;
  BackgroundWorkerHandle *bgw_handle;
  BgwHandleStatus status;
  pid_t   pid;
  uint64    invocation;
  char    activity[NAMEDATALEN + 64];
  DataChecksumsWorkerResult result;

  LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);

  /*
   * Initialize result to FAILED.  The worker will change it to SUCCESSFUL
   * if it completes successfully.
   */
  DataChecksumState->worker_result = DATACHECKSUMSWORKER_FAILED;
  DataChecksumState->worker_pid = InvalidPid;

  invocation = ++DataChecksumState->worker_invocation_counter;
  DataChecksumState->worker_invocation = invocation;
  DataChecksumState->database_oid = db->dboid;

  LWLockRelease(DataChecksumsWorkerLock);

  memset(&bgw, 0, sizeof(bgw));
  bgw.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION;
  bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
  snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
  snprintf(bgw.bgw_function_name, BGW_MAXLEN, "%s", "DataChecksumsWorkerMain");
  snprintf(bgw.bgw_name, BGW_MAXLEN, "datachecksums worker");
  snprintf(bgw.bgw_type, BGW_MAXLEN, "datachecksums worker");
  bgw.bgw_restart_time = BGW_NEVER_RESTART;
  bgw.bgw_notify_pid = MyProcPid;
  /* pass the invocation number to the worker process */
  bgw.bgw_main_arg = UInt64GetDatum(invocation);

  /*
   * If there are no worker slots available, there is little we can do.  If
   * we retry in a bit it's still unlikely that the user has managed to
   * reconfigure in the meantime and we'd be run through retries fast.
   */
  if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
  {
    ereport(WARNING,
        errmsg("could not start background worker for enabling data checksums in database \"%s\"",
             db->dbname),
        errhint("The \"%s\" setting might be too low.", "max_worker_processes"));
    return DATACHECKSUMSWORKER_FAILED;
  }

  /*
   * While this expects to wait for BGWORKER_STARTUP it may return _STOPPED
   * if the worker was terminated in the meantime so we must check status.
   */
  status = WaitForDataChecksumsWorkerState(bgw_handle, true, &pid,
                       WAIT_EVENT_BGWORKER_STARTUP);
  if (status == BGWH_STOPPED)
  {
    if (abort_requested)
    {
      result = DATACHECKSUMSWORKER_ABORTED;
      goto done;
    }

    /*
     * If the worker managed to start, and stop, before we got to waiting
     * for it we can see a STOPPED status here without it being a failure.
     */
    LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);
    Assert(DataChecksumState->worker_invocation == invocation);
    if (DataChecksumState->worker_result == DATACHECKSUMSWORKER_SUCCESSFUL)
    {
      LWLockRelease(DataChecksumsWorkerLock);
      pgstat_report_activity(STATE_IDLE, NULL);
      return DATACHECKSUMSWORKER_SUCCESSFUL;
    }
    LWLockRelease(DataChecksumsWorkerLock);

    ereport(WARNING,
        errmsg("could not start background worker for enabling data checksums in database \"%s\"",
             db->dbname),
        errhint("More details on the error might be found in the server log."));

    /*
     * Heuristic to see if the database was dropped, and if it was we can
     * treat it as not an error, else treat as fatal and error out.
     */
    if (DatabaseExists(db->dboid))
      return DATACHECKSUMSWORKER_FAILED;
    else
      return DATACHECKSUMSWORKER_DROPDB;
  }

  /*
   * If the postmaster crashed we cannot end up with a processed database so
   * we have no alternative other than exiting. When enabling checksums we
   * won't at this time have changed the data checksums state in pg_control
   * to enabled so when the cluster comes back up processing will have to be
   * restarted.
   */
  if (status == BGWH_POSTMASTER_DIED)
    ereport(FATAL,
        errcode(ERRCODE_ADMIN_SHUTDOWN),
        errmsg("cannot enable data checksums without the postmaster process"),
        errhint("Restart the database and restart data checksum processing by calling pg_enable_data_checksums()."));

  Assert(status == BGWH_STARTED);
  ereport(LOG,
      errmsg("initiating data checksum processing in database \"%s\"",
           db->dbname));

  /* Save the pid of the worker so we can signal it later */
  LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
  Assert(DataChecksumState->worker_invocation == invocation);
  DataChecksumState->worker_pid = pid;
  LWLockRelease(DataChecksumsWorkerLock);

  snprintf(activity, sizeof(activity) - 1,
       "Waiting for worker in database %s (pid %ld)", db->dbname, (long) pid);
  pgstat_report_activity(STATE_RUNNING, activity);

  status = WaitForDataChecksumsWorkerState(bgw_handle, false, NULL,
                       WAIT_EVENT_BGWORKER_SHUTDOWN);
  if (status == BGWH_POSTMASTER_DIED)
    ereport(FATAL,
        errcode(ERRCODE_ADMIN_SHUTDOWN),
        errmsg("postmaster exited during data checksum processing in \"%s\"",
             db->dbname),
        errhint("Restart the database and restart data checksum processing by calling pg_enable_data_checksums()."));

  LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
  Assert(DataChecksumState->worker_invocation == invocation);
  result = DataChecksumState->worker_result;
  DataChecksumState->worker_pid = InvalidPid;
  LWLockRelease(DataChecksumsWorkerLock);

  /*
   * A worker which started but failed before reporting a result has most
   * likely FATALed in InitPostgres.  If the database was dropped after we
   * built the database list then that is the expected outcome and not an
   * error, so apply the same heuristic as when the worker failed to start.
   */
  if (result == DATACHECKSUMSWORKER_FAILED && !DatabaseExists(db->dboid))
    result = DATACHECKSUMSWORKER_DROPDB;

  CHECK_FOR_LAUNCHER_ABORT_REQUEST();
  if (abort_requested)
    result = DATACHECKSUMSWORKER_ABORTED;

done:
  if (result == DATACHECKSUMSWORKER_ABORTED)
    ereport(LOG,
        errmsg("data checksums processing was aborted in database \"%s\"",
             db->dbname));
  pgstat_report_activity(STATE_IDLE, NULL);
  return result;
}

/*
 * launcher_exit
 *
 * Internal routine for cleaning up state when a launcher process which has
 * performed checksum operations exits. A launcher process which is exiting due
 * to a duplicate started launcher does not need to perform any cleanup and
 * this function should not be called. Otherwise, we need to clean up the abort
 * flag to ensure that processing can be started again if it was previously
 * aborted (note: started again, *not* restarted from where it left off).
 */
static void
launcher_exit(int code, Datum arg)
{
  abort_requested = false;

  if (launcher_running)
  {
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    if (DataChecksumState->worker_pid != InvalidPid)
    {
      ereport(LOG,
          errmsg("data checksums launcher exiting while worker is still running, signalling worker"));
      kill(DataChecksumState->worker_pid, SIGTERM);
      DataChecksumState->worker_pid = InvalidPid;
    }
    LWLockRelease(DataChecksumsWorkerLock);
  }

  /*
   * If the launcher is exiting before data checksums are enabled then set
   * the state to off since processing cannot be resumed.
   */
  if (DataChecksumsInProgressOn())
    SetDataChecksumsOff();

  LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
  if (launcher_running)
    DataChecksumState->launcher_running = false;
  launcher_running = false;
  LWLockRelease(DataChecksumsWorkerLock);
}

/*
 * launcher_cancel_handler
 *
 * Internal routine for reacting to SIGINT and flagging the launcher to abort.
 * If a worker is registered or running, the launcher will request worker
 * termination from its normal control flow.
 */
static void
launcher_cancel_handler(SIGNAL_ARGS)
{
  int     save_errno = errno;

  abort_requested = true;

  /*
   * Wake the launcher if it is waiting for transactions to finish or for a
   * worker to start up or shut down.
   */
  SetLatch(MyLatch);

  errno = save_errno;
}

/*
 * WaitForAllTransactionsToFinish
 *    Blocks awaiting all current transactions to finish
 *
 * Returns when all transactions which are active at the call of the function
 * have ended.
 *
 * NB: this will return early, if aborted by SIGINT or if the target state
 * is changed while we're running.
 */
static void
WaitForAllTransactionsToFinish(void)
{
  TransactionId waitforxid;

  LWLockAcquire(XidGenLock, LW_SHARED);
  waitforxid = XidFromFullTransactionId(TransamVariables->nextXid);
  LWLockRelease(XidGenLock);

  while (TransactionIdPrecedes(GetOldestActiveTransactionId(false, true), waitforxid))
  {
    char    activity[64];
    int     rc;

    /* Oldest running xid is older than us, so wait */
    snprintf(activity,
         sizeof(activity),
         "Waiting for transactions older than %u to end",
         waitforxid);
    pgstat_report_activity(STATE_RUNNING, activity);

    /* Retry every 3 seconds */
    ResetLatch(MyLatch);
    rc = WaitLatch(MyLatch,
             WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
             3000,
             WAIT_EVENT_CHECKSUM_ENABLE_STARTCONDITION);

    /*
     * If the postmaster died, bail out.  But first print a log message to
     * note that the checksumming didn't complete.
     */
    if (rc & WL_POSTMASTER_DEATH)
      ereport(FATAL,
          errcode(ERRCODE_ADMIN_SHUTDOWN),
          errmsg("postmaster exited during data checksums processing"),
          errhint("Data checksums processing must be restarted manually after cluster restart."));

    CHECK_FOR_INTERRUPTS();
    CHECK_FOR_LAUNCHER_ABORT_REQUEST();

    if (abort_requested)
      break;
  }

  pgstat_report_activity(STATE_IDLE, NULL);
  return;
}

/*
 * DataChecksumsWorkerLauncherMain
 *
 * Main function for launching dynamic background workers for processing data
 * checksums in databases. This function has the bgworker management, with
 * ProcessAllDatabases being responsible for looping over the databases and
 * initiating processing.
 */
void
DataChecksumsWorkerLauncherMain(Datum arg)
{

  ereport(DEBUG1,
      errmsg("background worker \"datachecksums launcher\" started"));

  pqsignal(SIGTERM, die);
  pqsignal(SIGINT, launcher_cancel_handler);
  pqsignal(SIGUSR1, procsignal_sigusr1_handler);
  pqsignal(SIGUSR2, PG_SIG_IGN);

  BackgroundWorkerUnblockSignals();

  MyBackendType = B_DATACHECKSUMSWORKER_LAUNCHER;
  init_ps_display(NULL);

  INJECTION_POINT("datachecksumsworker-launcher-delay", NULL);

  LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);

  if (DataChecksumState->launcher_running)
  {
    ereport(LOG,
        errmsg("background worker \"datachecksums launcher\" already running, exiting"));
    /* Launcher was already running, let it finish */
    LWLockRelease(DataChecksumsWorkerLock);
    return;
  }

  on_shmem_exit(launcher_exit, 0);
  launcher_running = true;

  /* Initialize a connection to shared catalogs only */
  BackgroundWorkerInitializeConnectionByOid(InvalidOid, InvalidOid, 0);

  operation = DataChecksumState->launch_operation;
  DataChecksumState->launcher_running = true;
  DataChecksumState->operation = operation;
  DataChecksumState->cost_delay = DataChecksumState->launch_cost_delay;
  DataChecksumState->cost_limit = DataChecksumState->launch_cost_limit;
  LWLockRelease(DataChecksumsWorkerLock);

  /*
   * The target state can change while we are busy enabling/disabling
   * checksums, if the user calls pg_disable/enable_data_checksums() before
   * we are finished with the previous request. In that case, we will loop
   * back here, to process the new request.
   */
again:

  pgstat_progress_start_command(PROGRESS_COMMAND_DATACHECKSUMS,
                  InvalidOid);
  ResetDataChecksumsProgressCounters();

  if (operation == ENABLE_DATACHECKSUMS)
  {
    /*
     * If we are asked to enable checksums in a cluster which already has
     * checksums enabled, exit immediately as there is nothing more to do.
     */
    if (DataChecksumsNeedVerify())
      goto done;

    ereport(LOG,
        errmsg("enabling data checksums requested, starting data checksum calculation"));

    /*
     * Set the state to inprogress-on and wait on the procsignal barrier.
     */
    pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
                   PROGRESS_DATACHECKSUMS_PHASE_ENABLING);
    SetDataChecksumsOnInProgress();

    /*
     * All backends are now in inprogress-on state and are writing data
     * checksums.  Start processing all data at rest.
     */
    if (!ProcessAllDatabases())
    {
      /*
       * If processing was canceled, or the target state changed during
       * processing, then it's not a failure. In the latter case, the
       * launcher will restart processing with the new target state.
       */
      CHECK_FOR_LAUNCHER_ABORT_REQUEST();
      if (abort_requested)
        goto done;
      ereport(ERROR,
          errcode(ERRCODE_INSUFFICIENT_RESOURCES),
          errmsg("unable to enable data checksums in cluster"));
    }

    /*
     * Data checksums have been set on all pages, set the state to on in
     * order to instruct backends to validate checksums on reading.
     */
    SetDataChecksumsOn();

    ereport(LOG,
        errmsg("data checksums are now enabled"));
  }
  else if (operation == DISABLE_DATACHECKSUMS)
  {
    ereport(LOG,
        errmsg("disabling data checksums requested"));

    pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
                   PROGRESS_DATACHECKSUMS_PHASE_DISABLING);
    SetDataChecksumsOff();
    ereport(LOG,
        errmsg("data checksums are now disabled"));
  }
  else
    Assert(false);

done:

  /*
   * This state will only be displayed for a fleeting moment, but for the
   * sake of correctness it is still added before ending the command.
   */
  pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
                 PROGRESS_DATACHECKSUMS_PHASE_DONE);

  /*
   * All done. But before we exit, check if the target state was changed
   * while we were running. In that case we will have to start all over
   * again.
   */
  LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
  if (DataChecksumState->launch_operation != operation)
  {
    DataChecksumState->operation = DataChecksumState->launch_operation;
    operation = DataChecksumState->launch_operation;
    DataChecksumState->cost_delay = DataChecksumState->launch_cost_delay;
    DataChecksumState->cost_limit = DataChecksumState->launch_cost_limit;

    /*
     * If the user started, but aborted processing, and then changed their
     * mind again before we had time to exit we need to clear the abort
     * flag.
     */
    abort_requested = false;
    LWLockRelease(DataChecksumsWorkerLock);
    goto again;
  }

  /* Shut down progress reporting as we are done */
  pgstat_progress_end_command();

  launcher_running = false;
  DataChecksumState->launcher_running = false;
  LWLockRelease(DataChecksumsWorkerLock);
}

/*
 * ProcessAllDatabases
 *    Compute the list of all databases and process checksums in each
 *
 * This will generate a list of databases to process for enabling checksums.
 * If a database encounters a failure then processing will end immediately and
 * return an error.
 */
static bool
ProcessAllDatabases(void)
{
  List     *DatabaseList;
  int     cumulative_total = 0;

  /* Set up so first run processes shared catalogs, not once in every db */
  LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
  DataChecksumState->process_shared_catalogs = true;
  LWLockRelease(DataChecksumsWorkerLock);

  /* Get a list of all databases to process */
  WaitForAllTransactionsToFinish();
  if (abort_requested)
    return false;
  DatabaseList = BuildDatabaseList();

  /*
   * Update progress reporting with the total number of databases we need to
   * process.  This number should not be changed during processing, the
   * columns for processed databases is instead increased such that it can
   * be compared against the total.
   */
  {
    const int index[] = {
      PROGRESS_DATACHECKSUMS_DBS_TOTAL,
      PROGRESS_DATACHECKSUMS_DBS_DONE,
    };

    int64   vals[2];

    vals[0] = list_length(DatabaseList);
    vals[1] = 0;

    pgstat_progress_update_multi_param(2, index, vals);
  }

  foreach_ptr(DataChecksumsWorkerDatabase, db, DatabaseList)
  {
    DataChecksumsWorkerResult result;

    result = ProcessDatabase(db);

#ifdef USE_INJECTION_POINTS
    /* Allow a test process to alter the result of the operation */
    if (IS_INJECTION_POINT_ATTACHED("datachecksumsworker-fail-db-result"))
    {
      result = DATACHECKSUMSWORKER_FAILED;
      INJECTION_POINT_CACHED("datachecksumsworker-fail-db-result",
                   db->dbname);
    }
#endif

    pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_DBS_DONE,
                   ++cumulative_total);

    if (result == DATACHECKSUMSWORKER_FAILED)
    {
      /*
       * Disable checksums on cluster, because we failed one of the
       * databases and this is an all or nothing process.
       */
      SetDataChecksumsOff();
      ereport(ERROR,
          errcode(ERRCODE_INSUFFICIENT_RESOURCES),
          errmsg("data checksums failed to get enabled in all databases, aborting"),
          errhint("The server log might have more information on the cause of the error."));
    }
    else if (result == DATACHECKSUMSWORKER_ABORTED || abort_requested)
    {
      /* Abort flag set, so exit the whole process */
      FreeDatabaseList(DatabaseList);
      return false;
    }
    else if (result == DATACHECKSUMSWORKER_DROPDB)
    {
      /*
       * Ignore databases that were dropped before their worker could
       * process them, and continue with the remaining databases.
       */
      continue;
    }

    /*
     * When one database has completed, it will have done shared catalogs
     * so we don't have to process them again.
     */
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    DataChecksumState->process_shared_catalogs = false;
    LWLockRelease(DataChecksumsWorkerLock);
  }

  FreeDatabaseList(DatabaseList);

  pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
                 PROGRESS_DATACHECKSUMS_PHASE_WAITING_BARRIER);
  return true;
}

/*
 * DataChecksumsShmemRequest
 *    Request datachecksumsworker-related shared memory
 */
static void
DataChecksumsShmemRequest(void *arg)
{
  ShmemRequestStruct(.name = "DataChecksumsWorker Data",
             .size = sizeof(DataChecksumsStateStruct),
             .ptr = (void **) &DataChecksumState,
    );
}

/*
 * DatabaseExists
 *
 * Scans the system catalog to check if a database with the given Oid exists
 * and returns true if it is found, even if it is marked invalid.  An invalid
 * database still has files that need checksums, so only a missing catalog row
 * proves that a concurrent DROP DATABASE completed.
 */
static bool
DatabaseExists(Oid dboid)
{
  Relation  rel;
  ScanKeyData skey;
  SysScanDesc scan;
  bool    found;
  HeapTuple tuple;

  StartTransactionCommand();

  /*
   * DROP DATABASE holds an exclusive lock on the database from before it
   * terminates the connections to it until it commits, so take a lock which
   * conflicts with it to wait out a drop which is in flight.  Without this
   * we can see a database whose worker was just killed by DROP DATABASE ...
   * WITH (FORCE) as still existing, and report a spurious failure.
   */
  LockSharedObject(DatabaseRelationId, dboid, 0, AccessShareLock);

  rel = table_open(DatabaseRelationId, AccessShareLock);
  ScanKeyInit(&skey,
        Anum_pg_database_oid,
        BTEqualStrategyNumber, F_OIDEQ,
        ObjectIdGetDatum(dboid));
  scan = systable_beginscan(rel, DatabaseOidIndexId, true, SnapshotSelf,
                1, &skey);
  tuple = systable_getnext(scan);
  found = HeapTupleIsValid(tuple);

  systable_endscan(scan);
  table_close(rel, AccessShareLock);

  CommitTransactionCommand();

  return found;
}

/*
 * ErrorOnInvalidDatabases
 *    Error out if the cluster contains an invalid database
 *
 * A database left invalid by an interrupted DROP DATABASE cannot be connected
 * to, so data checksums can never be enabled in it, while its files remain on
 * disk where checksum verification will find them.  Report it to the caller
 * so the user can drop it before retrying.  Called from a normal backend, so
 * unlike DatabaseExists we are already in a transaction.
 *
 * A cluster can contain more than one invalid database, but only the first one
 * found is reported; collecting them all is not worth the complexity here.  A
 * user with several of them gets the error again for the next one after
 * dropping the reported database, which the hint accounts for.
 */
static void
ErrorOnInvalidDatabases(void)
{
  Relation  rel;
  TableScanDesc scan;
  HeapTuple tup;

  rel = table_open(DatabaseRelationId, AccessShareLock);
  scan = table_beginscan_catalog(rel, 0, NULL);

  while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
  {
    Form_pg_database pgdb = (Form_pg_database) GETSTRUCT(tup);

    if (database_is_invalid_form(pgdb))
      ereport(ERROR,
          errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
          errmsg("cannot enable data checksums in a cluster with invalid database \"%s\"",
               NameStr(pgdb->datname)),
          errhint("Use DROP DATABASE to drop invalid databases."));
  }

  table_endscan(scan);
  table_close(rel, AccessShareLock);
}

/*
 * BuildDatabaseList
 *    Compile a list of all currently available databases in the cluster
 *
 * This creates the list of databases for the datachecksumsworker workers to
 * add checksums to. If the caller wants to ensure that no concurrently
 * running CREATE DATABASE calls exist, this needs to be preceded by a call
 * to WaitForAllTransactionsToFinish().
 */
static List *
BuildDatabaseList(void)
{
  List     *DatabaseList = NIL;
  Relation  rel;
  TableScanDesc scan;
  HeapTuple tup;
  MemoryContext ctx = CurrentMemoryContext;
  MemoryContext oldctx;

  StartTransactionCommand();

  rel = table_open(DatabaseRelationId, AccessShareLock);
  scan = table_beginscan_catalog(rel, 0, NULL);

  while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
  {
    Form_pg_database pgdb = (Form_pg_database) GETSTRUCT(tup);
    DataChecksumsWorkerDatabase *db;

    oldctx = MemoryContextSwitchTo(ctx);

    db = palloc0_object(DataChecksumsWorkerDatabase);

    db->dboid = pgdb->oid;
    db->dbname = pstrdup(NameStr(pgdb->datname));

    DatabaseList = lappend(DatabaseList, db);

    MemoryContextSwitchTo(oldctx);
  }

  table_endscan(scan);
  table_close(rel, AccessShareLock);

  CommitTransactionCommand();

  return DatabaseList;
}

static void
FreeDatabaseList(List *dblist)
{
  if (!dblist)
    return;

  foreach_ptr(DataChecksumsWorkerDatabase, db, dblist)
  {
    if (db->dbname != NULL)
      pfree(db->dbname);
  }

  list_free_deep(dblist);
}

/*
 * BuildRelationList
 *    Compile a list of relations in the database
 *
 * Returns a list of OIDs for the requested relation types. If temp_relations
 * is True then only temporary relations with storage are returned.  If
 * temp_relations is False then non-temporary relations with storage are
 * returned.  If include_shared is True then shared relations are included as
 * well in a non-temporary list. include_shared has no relevance when building
 * a list of temporary relations.
 */
static List *
BuildRelationList(bool temp_relations, bool include_shared)
{
  List     *RelationList = NIL;
  Relation  rel;
  TableScanDesc scan;
  HeapTuple tup;
  MemoryContext ctx = CurrentMemoryContext;
  MemoryContext oldctx;

  StartTransactionCommand();

  rel = table_open(RelationRelationId, AccessShareLock);
  scan = table_beginscan_catalog(rel, 0, NULL);

  while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
  {
    Form_pg_class pgc = (Form_pg_class) GETSTRUCT(tup);

    if (!RELKIND_HAS_STORAGE(pgc->relkind))
      continue;

    /* Only include temporary relations when explicitly asked to */
    if (pgc->relpersistence == RELPERSISTENCE_TEMP)
    {
      if (!temp_relations)
        continue;
    }
    else
    {
      /*
       * If we are only interested in temp relations then continue
       * immediately as the current relation isn't a temp relation.
       */
      if (temp_relations)
        continue;

      if (pgc->relisshared && !include_shared)
        continue;
    }

    oldctx = MemoryContextSwitchTo(ctx);
    RelationList = lappend_oid(RelationList, pgc->oid);
    MemoryContextSwitchTo(oldctx);
  }

  table_endscan(scan);
  table_close(rel, AccessShareLock);

  CommitTransactionCommand();

  return RelationList;
}

/*
 * DataChecksumsWorkerMain
 *
 * Main function for enabling checksums in a single database. This is the
 * function set as the bgw_function_name in the dynamic background worker
 * process initiated for each database by the worker launcher. After enabling
 * data checksums in each applicable relation in the database, it will wait for
 * all temporary relations that were present when the function started to
 * disappear before returning. This is required since we cannot rewrite
 * existing temporary relations with data checksums.
 */
void
DataChecksumsWorkerMain(Datum arg)
{
  Oid     dboid;
  List     *RelationList = NIL;
  List     *InitialTempTableList = NIL;
  BufferAccessStrategy strategy;
  bool    aborted = false;
  int64   rels_done;
  bool    process_shared;
#ifdef USE_INJECTION_POINTS
  bool    retried = false;
#endif

  worker_invocation = DatumGetUInt64(arg);

  operation = ENABLE_DATACHECKSUMS;

  pqsignal(SIGTERM, die);
  pqsignal(SIGUSR1, procsignal_sigusr1_handler);

  BackgroundWorkerUnblockSignals();

  MyBackendType = B_DATACHECKSUMSWORKER_WORKER;
  init_ps_display(NULL);

  LWLockAcquire(DataChecksumsWorkerLock, LW_SHARED);
  if (DataChecksumState->worker_invocation != worker_invocation)
  {
    LWLockRelease(DataChecksumsWorkerLock);
    return;
  }
  dboid = DataChecksumState->database_oid;
  LWLockRelease(DataChecksumsWorkerLock);

  BackgroundWorkerInitializeConnectionByOid(dboid, InvalidOid,
                        BGWORKER_BYPASS_ALLOWCONN);

  /* worker will have a separate entry in pg_stat_progress_data_checksums */
  pgstat_progress_start_command(PROGRESS_COMMAND_DATACHECKSUMS,
                  InvalidOid);
  ResetDataChecksumsProgressCounters();

  /*
   * Get a list of all temp tables present as we start in this database. We
   * need to wait until they are all gone before we exit.  For the list of
   * relations to enable checksums in, check if shared catalogs have been
   * processed already.
   */
  InitialTempTableList = BuildRelationList(true, false);
  LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
  if (DataChecksumState->worker_invocation != worker_invocation)
  {
    LWLockRelease(DataChecksumsWorkerLock);
    return;
  }
  process_shared = DataChecksumState->process_shared_catalogs;

  /*
   * Enable vacuum cost delay, if any.  While this process isn't doing any
   * vacuuming, we are re-using the infrastructure that vacuum cost delay
   * provides rather than inventing something bespoke. This is an internal
   * implementation detail and care should be taken to avoid it bleeding
   * through to the user to avoid confusion.
   *
   * VacuumUpdateCosts() propagates the values to the variables actually
   * read by vacuum_delay_point().
   */
  VacuumCostDelay = DataChecksumState->cost_delay;
  VacuumCostLimit = DataChecksumState->cost_limit;
  LWLockRelease(DataChecksumsWorkerLock);
  VacuumUpdateCosts();
  VacuumCostBalance = 0;

  /*
   * Create and set the vacuum strategy as our buffer strategy.
   */
  strategy = GetAccessStrategy(BAS_VACUUM);

  RelationList = BuildRelationList(false, process_shared);

  /* Update the total number of relations to be processed in this DB. */
  {
    const int index[] = {
      PROGRESS_DATACHECKSUMS_RELS_TOTAL,
      PROGRESS_DATACHECKSUMS_RELS_DONE
    };

    int64   vals[2];

    vals[0] = list_length(RelationList);
    vals[1] = 0;

    pgstat_progress_update_multi_param(2, index, vals);
  }

  /* Process the relations */
  rels_done = 0;
  foreach_oid(reloid, RelationList)
  {
    bool    costs_updated = false;

    if (!ProcessSingleRelationByOid(reloid, strategy))
    {
      aborted = true;
      break;
    }

    pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_RELS_DONE,
                   ++rels_done);
    CHECK_FOR_INTERRUPTS();
    CHECK_FOR_WORKER_ABORT_REQUEST();

    if (abort_requested)
      break;

    /*
     * Check if the cost settings changed during runtime and if so, update
     * to reflect the new values and signal that the access strategy needs
     * to be refreshed.
     */
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    if (DataChecksumState->worker_invocation != worker_invocation)
    {
      LWLockRelease(DataChecksumsWorkerLock);
      break;
    }
    if ((DataChecksumState->launch_cost_delay != DataChecksumState->cost_delay)
      || (DataChecksumState->launch_cost_limit != DataChecksumState->cost_limit))
    {
      costs_updated = true;
      VacuumCostDelay = DataChecksumState->launch_cost_delay;
      VacuumCostLimit = DataChecksumState->launch_cost_limit;
      VacuumUpdateCosts();

      DataChecksumState->cost_delay = DataChecksumState->launch_cost_delay;
      DataChecksumState->cost_limit = DataChecksumState->launch_cost_limit;
    }
    else
      costs_updated = false;
    LWLockRelease(DataChecksumsWorkerLock);

    if (costs_updated)
    {
      FreeAccessStrategy(strategy);
      strategy = GetAccessStrategy(BAS_VACUUM);
    }
  }

  list_free(RelationList);
  FreeAccessStrategy(strategy);

  if (aborted || abort_requested)
  {
    LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
    if (DataChecksumState->worker_invocation == worker_invocation)
      DataChecksumState->worker_result = DATACHECKSUMSWORKER_ABORTED;
    LWLockRelease(DataChecksumsWorkerLock);
    ereport(DEBUG1,
        errmsg("data checksum processing aborted in database OID %u",
             dboid));
    return;
  }

  /* The worker is about to wait for temporary tables to go away. */
  pgstat_progress_update_param(PROGRESS_DATACHECKSUMS_PHASE,
                 PROGRESS_DATACHECKSUMS_PHASE_WAITING_TEMPREL);

  /*
   * Wait for all temp tables that existed when we started to go away. This
   * is necessary since we cannot "reach" them to enable checksums. Any temp
   * tables created after we started will already have checksums in them
   * (due to the "inprogress-on" state), so no need to wait for those.
   */
  for (;;)
  {
    List     *CurrentTempTables;
    int     numleft;
    char    activity[64];

    CurrentTempTables = BuildRelationList(true, false);
    numleft = 0;
    foreach_oid(tmptbloid, InitialTempTableList)
    {
      if (list_member_oid(CurrentTempTables, tmptbloid))
        numleft++;
    }
    list_free(CurrentTempTables);

#ifdef USE_INJECTION_POINTS
    if (IS_INJECTION_POINT_ATTACHED("datachecksumsworker-fake-temptable-wait"))
    {
      /* Make sure to just cause one retry */
      if (!retried && numleft == 0)
      {
        numleft = 1;
        retried = true;

        INJECTION_POINT_CACHED("datachecksumsworker-fake-temptable-wait", NULL);
      }
    }
#endif

    if (numleft == 0)
      break;

    /*
     * At least one temp table is left to wait for, indicate in pgstat
     * activity and progress reporting.
     */
    snprintf(activity,
         sizeof(activity),
         "Waiting for %d temp tables to be removed", numleft);
    pgstat_report_activity(STATE_RUNNING, activity);

    /* Retry every 3 seconds */
    ResetLatch(MyLatch);
    (void) WaitLatch(MyLatch,
             WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
             3000,
             WAIT_EVENT_CHECKSUM_ENABLE_TEMPTABLE_WAIT);

    CHECK_FOR_INTERRUPTS();
    CHECK_FOR_WORKER_ABORT_REQUEST();

    if (aborted || abort_requested)
    {
      LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
      if (DataChecksumState->worker_invocation == worker_invocation)
        DataChecksumState->worker_result = DATACHECKSUMSWORKER_ABORTED;
      LWLockRelease(DataChecksumsWorkerLock);
      ereport(LOG,
          errmsg("data checksum processing aborted in database OID %u",
               dboid));
      return;
    }
  }

  list_free(InitialTempTableList);

  /* worker done */
  pgstat_progress_end_command();

  LWLockAcquire(DataChecksumsWorkerLock, LW_EXCLUSIVE);
  if (DataChecksumState->worker_invocation == worker_invocation)
    DataChecksumState->worker_result = DATACHECKSUMSWORKER_SUCCESSFUL;
  LWLockRelease(DataChecksumsWorkerLock);
}

Coverage Report

Created: 2026-09-28 06:55