From f4987dd3843d129b41b0589e600db1c3d1d57cd1 Mon Sep 17 00:00:00 2001
From: Zsolt Parragi <zsolt.parragi@percona.com>
Date: Fri, 14 Aug 2026 17:06:02 +0000
Subject: [PATCH v9 1/4] Do not adopt data checksum state from another node
 during replay

Offline pg_checksums changes are local to one node, but replay adopted
the data checksum state carried by checkpoint records unconditionally.
After an offline enable on the primary, a standby whose pages were
never rewritten started verifying checksums it does not have; after an
offline change on a standby, the next replayed checkpoint silently
reverted it.

To fix, make the control file track this node's state alone, and have
replay cross-check the replayed state against it instead of adopting
it, warning once per divergent value and reporting when the states
agree again.  pg_control gains a watermark, normally the end LSN of
the newest XLOG2_CHECKSUMS record the node has written or applied, so
that replay can skip transition records whose effect the control file
already contains, and a flag marking a state last written by
pg_checksums, which recovery must never overwrite with a replayed one.

Since the control file may only claim "on" once every page on disk
carries a checksum, persisting the state is tied to flushes:
XLOG2_CHECKSUMS replay persists every state but "on" immediately and
leaves "on" to the next restartpoint, checkpoints and restartpoints
only persist a state their flush ran under from beginning to end, and
transitions publish their state under the new
DataChecksumTransitionLock so that the states carried by WAL records
match their WAL order.  The comments in xlog.c spell out the
individual rules.

Recovery from a base backup is the exception to not adopting: its
control file was copied at an arbitrary moment, so the state carried
by the starting checkpoint is the one the WAL from there on was
written under.  pg_rewind keeps the target's own state and clamps the
watermark to the divergence point, since a watermark set on the
target's abandoned fork could numerically cover transition records the
source wrote after the divergence.

Bump PG_CONTROL_VERSION.

Document the offline procedure for replication setups, the lockstep
one: stop all nodes, run pg_checksums on each of them, and only then
restart.  An offline change writes no WAL and has no ordering against
WAL a node has not replayed yet, so a node must not stop before
replaying all WAL of its upstream, or the change is overridden on
restart.

Author: Zsolt Parragi <zsolt.parragi@percona.com>
Author: Bertrand Drouvot <bertranddrouvot.pg@gmail.com>
---
 doc/src/sgml/ref/pg_checksums.sgml            |  51 +-
 doc/src/sgml/wal.sgml                         |  21 +
 src/backend/access/transam/xlog.c             | 547 ++++++++++++++++--
 .../utils/activity/wait_event_names.txt       |   1 +
 src/bin/pg_checksums/pg_checksums.c           |  10 +
 src/bin/pg_controldata/pg_controldata.c       |   4 +
 src/bin/pg_resetwal/pg_resetwal.c             |   7 +
 src/bin/pg_rewind/pg_rewind.c                 |  35 +-
 src/bin/pg_upgrade/controldata.c              |   2 +-
 src/include/catalog/pg_control.h              |  28 +-
 src/include/storage/lwlocklist.h              |   1 +
 src/test/modules/test_checksums/Makefile      |   2 +-
 src/test/modules/test_checksums/meson.build   |  10 +
 .../test_checksums/t/012_offline_standby.pl   | 289 +++++++++
 .../modules/test_checksums/t/013_rewind.pl    | 201 +++++++
 .../modules/test_checksums/t/014_lockstep.pl  | 182 ++++++
 .../t/015_standby_crash_after_disable.pl      | 125 ++++
 .../t/016_promote_enable_crash.pl             | 134 +++++
 .../test_checksums/t/017_restartpoint_race.pl | 138 +++++
 .../t/018_enable_crash_windows.pl             | 496 ++++++++++++++++
 .../t/019_standby_shutdown_catchup.pl         | 126 ++++
 .../t/020_cascade_divergence.pl               | 115 ++++
 .../t/021_rewind_divergent_transitions.pl     | 193 ++++++
 23 files changed, 2639 insertions(+), 79 deletions(-)
 create mode 100644 src/test/modules/test_checksums/t/012_offline_standby.pl
 create mode 100644 src/test/modules/test_checksums/t/013_rewind.pl
 create mode 100644 src/test/modules/test_checksums/t/014_lockstep.pl
 create mode 100644 src/test/modules/test_checksums/t/015_standby_crash_after_disable.pl
 create mode 100644 src/test/modules/test_checksums/t/016_promote_enable_crash.pl
 create mode 100644 src/test/modules/test_checksums/t/017_restartpoint_race.pl
 create mode 100644 src/test/modules/test_checksums/t/018_enable_crash_windows.pl
 create mode 100644 src/test/modules/test_checksums/t/019_standby_shutdown_catchup.pl
 create mode 100644 src/test/modules/test_checksums/t/020_cascade_divergence.pl
 create mode 100644 src/test/modules/test_checksums/t/021_rewind_divergent_transitions.pl

diff --git a/doc/src/sgml/ref/pg_checksums.sgml b/doc/src/sgml/ref/pg_checksums.sgml
index ae66fad3f0f..6f332ade462 100644
--- a/doc/src/sgml/ref/pg_checksums.sgml
+++ b/doc/src/sgml/ref/pg_checksums.sgml
@@ -242,15 +242,48 @@ PostgreSQL documentation
    data directory must not be started or else data loss may occur.
   </para>
   <para>
-   When using a replication setup with tools which perform direct copies
-   of relation file blocks (for example <xref linkend="app-pgrewind"/>),
-   enabling or disabling checksums can lead to page corruptions in the
-   shape of incorrect checksums if the operation is not done consistently
-   across all nodes. When enabling or disabling checksums in a replication
-   setup, it is thus recommended to stop all the clusters before switching
-   them all consistently. Destroying all standbys, performing the operation
-   on the primary and finally recreating the standbys from scratch is also
-   safe.
+   Enabling or disabling checksums with
+   <application>pg_checksums</application> changes only the local data
+   directory; the new state does not propagate over replication.  In a
+   replication setup the same change must be applied to every node: stop all
+   nodes, run <application>pg_checksums</application> on each of them, and
+   only then restart them.  Before stopping a standby, make sure it has
+   replayed all WAL of its upstream node, for example by stopping the
+   primary first and comparing
+   <function>pg_last_wal_replay_lsn()</function> with
+   <function>pg_last_wal_receive_lsn()</function> on the standby.  Tools
+   that copy relation file blocks directly between nodes, such as
+   <xref linkend="app-pgrewind"/>, likewise require both nodes to be in
+   the same data checksum state.
+  </para>
+  <para>
+   The replay requirement exists because an offline change is recorded
+   only in the control file and has no defined ordering against WAL the
+   node has not replayed yet; see
+   <xref linkend="checksums-offline-enable-disable"/>.  A node stopped
+   before replaying an online checksum state change applies that change
+   when it is restarted, overriding the offline change, and the states of
+   the nodes silently diverge until a later checkpoint record triggers
+   the warning described below.  Because of this it is best not to mix
+   the two mechanisms: change the state of a replication setup either
+   with the offline procedure above or with an online transition, and
+   make sure the previous change has reached every node before starting
+   the next one.
+  </para>
+  <para>
+   If the change is applied inconsistently, each node keeps its own state,
+   and a standby logs a warning when the state recorded in the replayed WAL
+   differs from its own.  The same warning can appear transiently while a
+   standby catches up over WAL written before a consistent change; it stops
+   once a checkpoint record carrying the new state has been replayed.
+  </para>
+  <para>
+   A standby whose data directory was never checksummed must not have
+   checksums enabled by catching up this way.  Converge the cluster by
+   running <application>pg_checksums</application> on it while stopped, by
+   enabling checksums online, or by recreating it from a base backup.  Note
+   that an online enable only starts from a primary whose checksums are off,
+   so if they are already enabled there, disable them online first.
   </para>
   <para>
    If <application>pg_checksums</application> is aborted or killed while
diff --git a/doc/src/sgml/wal.sgml b/doc/src/sgml/wal.sgml
index ec62d17fbcc..d88a832c2a8 100644
--- a/doc/src/sgml/wal.sgml
+++ b/doc/src/sgml/wal.sgml
@@ -317,6 +317,27 @@
     verify checksums, on an offline cluster.
    </para>
 
+   <para>
+    An offline change provides durability differently from an
+    <link linkend="checksums-online-enable-disable">online change</link>.
+    An online transition is WAL-logged: it is ordered against all other
+    WAL records, it is replayed after a crash, and it propagates to
+    standbys.  An offline change is recorded only in the cluster's
+    control file: it writes no WAL, it is invisible to replication, and
+    it has no defined ordering against WAL the node has not replayed
+    yet.  When a node later replays WAL that contains an online checksum
+    state change, that change takes effect on the node even if it was
+    written before the offline change was made.
+   </para>
+
+   <para>
+    An offline change only affects the data directory it is run on; the
+    new state does not propagate over replication.  In a replication setup
+    the same change must be applied to all nodes while all of them are
+    stopped, as described in <xref linkend="app-pgchecksums"/>.  A standby
+    whose state diverges logs a warning but keeps its local setting.
+   </para>
+
   </sect2>
 
   <sect2 id="checksums-online-enable-disable" xreflabel="Online Enabling of Checksums">
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index de4c96e135f..7487ca509f3 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -556,9 +556,17 @@ typedef struct XLogCtlData
 	 */
 	XLogRecPtr	lastFpwDisableRecPtr;
 
-	/* last data_checksum_version we've seen */
+	/* current data checksum state of this node */
 	uint32		data_checksum_version;
 
+	/*
+	 * In-memory copies of ControlFile->data_checksum_lsn and
+	 * ControlFile->data_checksum_is_local, see there.  Updated together with
+	 * data_checksum_version under info_lck.
+	 */
+	XLogRecPtr	data_checksum_lsn;
+	bool		data_checksum_is_local;
+
 	slock_t		info_lck;		/* locks shared variables shown above */
 
 	/*
@@ -690,6 +698,14 @@ static ChecksumStateType LocalDataChecksumState = 0;
  */
 int			data_checksums = 0;
 
+/*
+ * Whether replay of the next checkpoint-family record must adopt the data
+ * checksum state it carries.  Set when recovery starts from a base backup,
+ * where the state at the redo point takes precedence over the control file
+ * copied with the backup later.
+ */
+static bool adoptChecksumStateFromNextCheckpoint = false;
+
 /* For WALInsertLockAcquire/Release functions */
 static int	MyLockNo = 0;
 static bool holdingAllLocks = false;
@@ -730,6 +746,8 @@ static void ValidateXLOGDirectoryStructure(void);
 static void CleanupBackupHistory(void);
 static void UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force);
 static bool PerformRecoveryXLogAction(void);
+static void CheckReplayedDataChecksumState(uint32 replayed_version);
+static void AdoptReplayedDataChecksumState(uint32 new_version, XLogRecPtr lsn);
 static void InitControlFile(uint64 sysidentifier, uint32 data_checksum_version);
 static void WriteControlFile(void);
 static void ReadControlFile(void);
@@ -757,7 +775,7 @@ static void WALInsertLockAcquireExclusive(void);
 static void WALInsertLockRelease(void);
 static void WALInsertLockUpdateInsertingAt(XLogRecPtr insertingAt);
 
-static void XLogChecksums(uint32 new_type);
+static XLogRecPtr XLogChecksums(uint32 new_type);
 
 /*
  * Insert an XLOG record represented by an already-constructed chain of data
@@ -4774,6 +4792,7 @@ void
 SetDataChecksumsOnInProgress(void)
 {
 	uint64		barrier;
+	XLogRecPtr	recptr;
 
 	/*
 	 * The state transition is performed in a critical section with
@@ -4782,14 +4801,12 @@ SetDataChecksumsOnInProgress(void)
 	START_CRIT_SECTION();
 	MyProc->delayChkptFlags |= DELAY_CHKPT_START;
 
-	XLogChecksums(PG_DATA_CHECKSUM_INPROGRESS_ON);
-
-	SpinLockAcquire(&XLogCtl->info_lck);
-	XLogCtl->data_checksum_version = PG_DATA_CHECKSUM_INPROGRESS_ON;
-	SpinLockRelease(&XLogCtl->info_lck);
+	recptr = XLogChecksums(PG_DATA_CHECKSUM_INPROGRESS_ON);
 
 	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
 	ControlFile->data_checksum_version = PG_DATA_CHECKSUM_INPROGRESS_ON;
+	ControlFile->data_checksum_lsn = recptr;
+	ControlFile->data_checksum_is_local = false;
 	UpdateControlFile();
 	LWLockRelease(ControlFileLock);
 
@@ -4827,6 +4844,8 @@ void
 SetDataChecksumsOn(void)
 {
 	uint64		barrier;
+	bool		persist;
+	XLogRecPtr	recptr;
 
 	SpinLockAcquire(&XLogCtl->info_lck);
 
@@ -4847,23 +4866,11 @@ SetDataChecksumsOn(void)
 	SpinLockRelease(&XLogCtl->info_lck);
 
 	INJECTION_POINT("datachecksums-enable-checksums-delay", NULL);
+	INJECTION_POINT_LOAD("datachecksums-on-before-publish");
 	START_CRIT_SECTION();
 	MyProc->delayChkptFlags |= DELAY_CHKPT_START;
 
-	XLogChecksums(PG_DATA_CHECKSUM_VERSION);
-
-	SpinLockAcquire(&XLogCtl->info_lck);
-	XLogCtl->data_checksum_version = PG_DATA_CHECKSUM_VERSION;
-	SpinLockRelease(&XLogCtl->info_lck);
-
-	/*
-	 * Update the controlfile before waiting since if we have an immediate
-	 * shutdown while waiting we want to come back up with checksums enabled.
-	 */
-	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
-	ControlFile->data_checksum_version = PG_DATA_CHECKSUM_VERSION;
-	UpdateControlFile();
-	LWLockRelease(ControlFileLock);
+	recptr = XLogChecksums(PG_DATA_CHECKSUM_VERSION);
 
 	barrier = EmitProcSignalBarrier(PROCSIGNAL_BARRIER_CHECKSUM_ON);
 
@@ -4873,6 +4880,39 @@ SetDataChecksumsOn(void)
 	INJECTION_POINT("datachecksums-on-before-checkpoint", NULL);
 
 	RequestCheckpoint(CHECKPOINT_FORCE | CHECKPOINT_WAIT | CHECKPOINT_FAST);
+
+	INJECTION_POINT("datachecksums-on-after-checkpoint", NULL);
+
+	/*
+	 * Persist "on" only now that the checkpoint has flushed the pages the
+	 * transition rewrote.  Crash recovery initializes verification from the
+	 * control file but resumes from a checkpoint that can predate the
+	 * transition, so an "on" persisted earlier would have replay verify pages
+	 * whose rewrite never reached disk: the pages the worker found in shared
+	 * buffers are not written back by its ring buffer.
+	 *
+	 * The checkpoint above normally persists the state itself; this covers the
+	 * case where it started before the record written above and left the field
+	 * alone.  Crashing before this point is safe, as replay then re-establishes
+	 * "on" from the full page images of the rewrite.  Skip the write if the
+	 * state moved on meanwhile, since whatever moved it persists its own.
+	 * Compare the watermark rather than the state: a state comparison could
+	 * not tell our transition from a later round trip back to "on".
+	 */
+	SpinLockAcquire(&XLogCtl->info_lck);
+	persist = (XLogCtl->data_checksum_lsn == recptr);
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	if (persist)
+	{
+		LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+		ControlFile->data_checksum_version = PG_DATA_CHECKSUM_VERSION;
+		ControlFile->data_checksum_lsn = recptr;
+		ControlFile->data_checksum_is_local = false;
+		UpdateControlFile();
+		LWLockRelease(ControlFileLock);
+	}
+
 	WaitForProcSignalBarrier(barrier);
 }
 
@@ -4893,6 +4933,7 @@ void
 SetDataChecksumsOff(void)
 {
 	uint64		barrier;
+	XLogRecPtr	recptr;
 
 	SpinLockAcquire(&XLogCtl->info_lck);
 
@@ -4918,14 +4959,12 @@ SetDataChecksumsOff(void)
 		START_CRIT_SECTION();
 		MyProc->delayChkptFlags |= DELAY_CHKPT_START;
 
-		XLogChecksums(PG_DATA_CHECKSUM_INPROGRESS_OFF);
-
-		SpinLockAcquire(&XLogCtl->info_lck);
-		XLogCtl->data_checksum_version = PG_DATA_CHECKSUM_INPROGRESS_OFF;
-		SpinLockRelease(&XLogCtl->info_lck);
+		recptr = XLogChecksums(PG_DATA_CHECKSUM_INPROGRESS_OFF);
 
 		LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
 		ControlFile->data_checksum_version = PG_DATA_CHECKSUM_INPROGRESS_OFF;
+		ControlFile->data_checksum_lsn = recptr;
+		ControlFile->data_checksum_is_local = false;
 		UpdateControlFile();
 		LWLockRelease(ControlFileLock);
 
@@ -4956,14 +4995,12 @@ SetDataChecksumsOff(void)
 	/* Ensure that we don't incur a checkpoint during disabling checksums */
 	MyProc->delayChkptFlags |= DELAY_CHKPT_START;
 
-	XLogChecksums(PG_DATA_CHECKSUM_OFF);
-
-	SpinLockAcquire(&XLogCtl->info_lck);
-	XLogCtl->data_checksum_version = PG_DATA_CHECKSUM_OFF;
-	SpinLockRelease(&XLogCtl->info_lck);
+	recptr = XLogChecksums(PG_DATA_CHECKSUM_OFF);
 
 	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
 	ControlFile->data_checksum_version = PG_DATA_CHECKSUM_OFF;
+	ControlFile->data_checksum_lsn = recptr;
+	ControlFile->data_checksum_is_local = false;
 	UpdateControlFile();
 	LWLockRelease(ControlFileLock);
 
@@ -5001,6 +5038,131 @@ SetLocalDataChecksumState(uint32 data_checksum_version)
 	data_checksums = data_checksum_version;
 }
 
+/*
+ * Cross-check the data checksum state carried by a replayed checkpoint record
+ * against the state of this node.
+ *
+ * The state in the record belongs to whichever node wrote the WAL, and must
+ * not be adopted: an offline state change made with pg_checksums on one node
+ * of a replication set generates no WAL, and adopting would leak it into the
+ * other nodes through replay.  Backup label recovery is the one exception,
+ * see AdoptReplayedDataChecksumState().  XLOG_CHECKPOINT_ONLINE needs no call
+ * here, as an online checkpoint's state already traveled in the preceding
+ * XLOG_CHECKPOINT_REDO record.
+ *
+ * Only archive recovery can see a lasting mismatch, as only there can the WAL
+ * and the control file come from different nodes or different times.  In
+ * crash recovery a mismatch means replay resumed from a restartpoint
+ * predating an already-applied XLOG2_CHECKSUMS record, and replaying forward
+ * re-establishes the same state.
+ */
+static void
+CheckReplayedDataChecksumState(uint32 replayed_version)
+{
+	/*
+	 * Warn once per remote value, so a lasting mismatch does not flood the
+	 * log.  Matching states re-arm the warning.  Backend-local state is
+	 * enough: replay only runs in the startup process, and restarting it
+	 * re-arms as well.
+	 */
+	static uint32 last_warned_version = PG_UINT32_MAX;
+	uint32		local_version;
+
+	if (!ArchiveRecoveryRequested)
+		return;
+
+	/*
+	 * Re-replayed WAL below the consistency point was already cross-checked
+	 * before minRecoveryPoint was last persisted, and the persisted state can
+	 * legitimately be newer than what checkpoint records there carry:
+	 * XLOG2_CHECKSUMS replay persists most states ahead of the restartpoint
+	 * horizon.  In particular the checkpoint record recovery restarts from is
+	 * such a re-replay.
+	 */
+	if (!reachedConsistency)
+		return;
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	local_version = XLogCtl->data_checksum_version;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	if (replayed_version == local_version)
+	{
+		/*
+		 * Report convergence if this process warned before.  Nothing else
+		 * tells the operator that running pg_checksums on the other nodes, or
+		 * a rebuild, took effect.
+		 */
+		if (last_warned_version != PG_UINT32_MAX)
+			ereport(LOG,
+					errmsg("data checksum state \"%s\" of this node now agrees with the replayed WAL",
+						   get_checksum_state_string(local_version)));
+
+		last_warned_version = PG_UINT32_MAX;
+		return;
+	}
+
+	/* the nodes legitimately differ while an online transition runs */
+	if (replayed_version == PG_DATA_CHECKSUM_INPROGRESS_ON ||
+		replayed_version == PG_DATA_CHECKSUM_INPROGRESS_OFF ||
+		local_version == PG_DATA_CHECKSUM_INPROGRESS_ON ||
+		local_version == PG_DATA_CHECKSUM_INPROGRESS_OFF)
+		return;
+
+	if (replayed_version == last_warned_version)
+		return;
+	last_warned_version = replayed_version;
+
+	ereport(WARNING,
+			errmsg("data checksum state \"%s\" of this node does not match the state \"%s\" in the replayed WAL",
+				   get_checksum_state_string(local_version),
+				   get_checksum_state_string(replayed_version)),
+			errdetail("The data checksum state was most likely changed with pg_checksums on another node."),
+			errhint("Apply the same change with pg_checksums on the primary and all standby servers, or rebuild this server from a base backup."));
+}
+
+/*
+ * Adopt the data checksum state found at the redo point of backup label
+ * recovery.  Persist it immediately so that a crash before the first
+ * restartpoint does not resurrect the state copied with the backup; a crash
+ * at this point restarts from the same redo point, so the control file does
+ * not run ahead of the replay position.  If the value is unchanged the
+ * control file already carries it, so both the barrier and the persist are
+ * skipped.
+ *
+ * lsn is the location of the checkpoint-family record the state was taken
+ * from and becomes the new watermark: the adopted state covers everything
+ * below the redo point, which replay never revisits.
+ */
+static void
+AdoptReplayedDataChecksumState(uint32 new_version, XLogRecPtr lsn)
+{
+	bool		changed = false;
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	if (XLogCtl->data_checksum_version != new_version)
+	{
+		XLogCtl->data_checksum_version = new_version;
+		XLogCtl->data_checksum_lsn = lsn;
+		XLogCtl->data_checksum_is_local = false;
+		SetLocalDataChecksumState(new_version);
+		changed = true;
+	}
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	if (!changed)
+		return;
+
+	EmitAndWaitDataChecksumsBarrier(new_version);
+
+	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
+	ControlFile->data_checksum_version = new_version;
+	ControlFile->data_checksum_lsn = lsn;
+	ControlFile->data_checksum_is_local = false;
+	UpdateControlFile();
+	LWLockRelease(ControlFileLock);
+}
+
 /* guc hook */
 const char *
 show_data_checksums(void)
@@ -5454,6 +5616,8 @@ XLOGShmemInit(void *arg)
 
 	/* Use the checksum info from control file */
 	XLogCtl->data_checksum_version = ControlFile->data_checksum_version;
+	XLogCtl->data_checksum_lsn = ControlFile->data_checksum_lsn;
+	XLogCtl->data_checksum_is_local = ControlFile->data_checksum_is_local;
 	SetLocalDataChecksumState(XLogCtl->data_checksum_version);
 
 	SpinLockInit(&XLogCtl->Insert.insertpos_lck);
@@ -6031,6 +6195,52 @@ StartupXLOG(void)
 	SetCommitTsLimit(checkPoint.oldestCommitTsXid,
 					 checkPoint.newestCommitTsXid);
 
+	/*
+	 * When recovery starts from a base backup, the control file was copied at
+	 * an arbitrary moment and its data checksum state may differ from the
+	 * state at the redo point, which is what the WAL from there on was
+	 * written under.  Adopt the state of the starting checkpoint: a shutdown
+	 * checkpoint is not replayed, so take it from the record read above; the
+	 * redo point of an online checkpoint is its CHECKPOINT_REDO record, so
+	 * let the replay of that record adopt it.  Check backupStartPoint in
+	 * addition to the label: on a crash restart during backup recovery the
+	 * label file is already renamed away, but the start point persists until
+	 * the backup end record.
+	 *
+	 * Not for a base backup taken from a standby, though.  Its starting
+	 * checkpoint is the standby's last restartpoint, a record written by the
+	 * upstream primary, whose state is not the one the copied files were
+	 * written under.  The copied control file is already correct: a standby
+	 * persists its state only at restartpoint horizons and never claims
+	 * more than what reached disk.  Such backups are recognized by
+	 * backupEndPoint together with backupEndRequired; backupEndPoint is only
+	 * set for "BACKUP FROM: standby" labels and persists across a crash
+	 * restart.  pg_rewind writes a standby label as well, but no
+	 * backupEndPoint, and its recovery keeps adopting: the control file it
+	 * installs carries the target's own checksum state, which can lag the
+	 * redo point of the last common checkpoint the same way a restartpoint
+	 * horizon can.
+	 *
+	 * Never adopt over a state the control file's watermark or local flag
+	 * marks as newer than the starting checkpoint.  A pg_checksums change is
+	 * local to the node and generates no WAL, so nothing in the replayed WAL
+	 * could ever restore it once overwritten; and a watermark above the redo
+	 * point means the control file already contains the effect of every
+	 * transition record up to there.
+	 */
+	if ((haveBackupLabel || XLogRecPtrIsValid(ControlFile->backupStartPoint)) &&
+		!(XLogRecPtrIsValid(ControlFile->backupEndPoint) &&
+		  ControlFile->backupEndRequired) &&
+		!ControlFile->data_checksum_is_local &&
+		checkPoint.redo > ControlFile->data_checksum_lsn)
+	{
+		if (wasShutdown)
+			AdoptReplayedDataChecksumState(checkPoint.dataChecksumState,
+										   checkPoint.redo);
+		else
+			adoptChecksumStateFromNextCheckpoint = true;
+	}
+
 	/*
 	 * Clear out any old relcache cache files.  This is *necessary* if we do
 	 * any WAL replay, since that would probably result in the cache files
@@ -6633,11 +6843,7 @@ StartupXLOG(void)
 	if (XLogCtl->data_checksum_version == PG_DATA_CHECKSUM_INPROGRESS_ON)
 	{
 		XLogChecksums(PG_DATA_CHECKSUM_OFF);
-
-		SpinLockAcquire(&XLogCtl->info_lck);
-		XLogCtl->data_checksum_version = PG_DATA_CHECKSUM_OFF;
-		SetLocalDataChecksumState(XLogCtl->data_checksum_version);
-		SpinLockRelease(&XLogCtl->info_lck);
+		SetLocalDataChecksumState(PG_DATA_CHECKSUM_OFF);
 
 		EmitAndWaitDataChecksumsBarrier(PG_DATA_CHECKSUM_OFF);
 		ereport(WARNING,
@@ -6654,11 +6860,7 @@ StartupXLOG(void)
 	else if (XLogCtl->data_checksum_version == PG_DATA_CHECKSUM_INPROGRESS_OFF)
 	{
 		XLogChecksums(PG_DATA_CHECKSUM_OFF);
-
-		SpinLockAcquire(&XLogCtl->info_lck);
-		XLogCtl->data_checksum_version = PG_DATA_CHECKSUM_OFF;
-		SetLocalDataChecksumState(XLogCtl->data_checksum_version);
-		SpinLockRelease(&XLogCtl->info_lck);
+		SetLocalDataChecksumState(PG_DATA_CHECKSUM_OFF);
 
 		EmitAndWaitDataChecksumsBarrier(PG_DATA_CHECKSUM_OFF);
 	}
@@ -6814,6 +7016,23 @@ static bool
 PerformRecoveryXLogAction(void)
 {
 	bool		promoted = false;
+	bool		flushForChecksums;
+	uint32		checksum_state;
+
+	/*
+	 * The end-of-recovery record persists the data checksum state without
+	 * flushing the buffer pool, but the control file may only claim "on" once
+	 * every page on disk carries a checksum.  If replay entered that state
+	 * without a restartpoint following it, the pages rewritten by the
+	 * transition are still only in the buffer pool, so take the full
+	 * checkpoint below instead of the lightweight record.
+	 */
+	SpinLockAcquire(&XLogCtl->info_lck);
+	checksum_state = XLogCtl->data_checksum_version;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	flushForChecksums = (checksum_state == PG_DATA_CHECKSUM_VERSION &&
+						 ControlFile->data_checksum_version != checksum_state);
 
 	/*
 	 * Perform a checkpoint to update all our recovery activity to disk.
@@ -6829,7 +7048,7 @@ PerformRecoveryXLogAction(void)
 	 * fully out of recovery mode and already accepting queries.
 	 */
 	if (ArchiveRecoveryRequested && IsUnderPostmaster &&
-		PromoteIsTriggered())
+		PromoteIsTriggered() && !flushForChecksums)
 	{
 		promoted = true;
 
@@ -7436,6 +7655,7 @@ CreateCheckPoint(int flags)
 	uint32		freespace;
 	XLogRecPtr	PriorRedoPtr;
 	XLogRecPtr	last_important_lsn;
+	XLogRecPtr	checksumLsn;
 	VirtualTransactionId *vxids;
 	int			nvxids;
 	int			oldXLogAllowed = 0;
@@ -7548,11 +7768,14 @@ CreateCheckPoint(int flags)
 	checkPoint.wal_level = wal_level;
 
 	/*
-	 * Get the current data_checksum_version value from xlogctl, valid at the
-	 * time of the checkpoint.
+	 * Get the current data_checksum_version value from xlogctl.  This is
+	 * final only for a shutdown checkpoint, where no concurrent transition
+	 * is possible; an online checkpoint resamples it together with the redo
+	 * record below.
 	 */
 	SpinLockAcquire(&XLogCtl->info_lck);
 	checkPoint.dataChecksumState = XLogCtl->data_checksum_version;
+	checksumLsn = XLogCtl->data_checksum_lsn;
 	SpinLockRelease(&XLogCtl->info_lck);
 
 	if (shutdown)
@@ -7609,10 +7832,21 @@ CreateCheckPoint(int flags)
 	{
 		xl_checkpoint_redo redo_rec;
 
+		/*
+		 * Sample the data checksum state and insert the redo record under
+		 * DataChecksumTransitionLock, so that a concurrent transition cannot
+		 * insert its XLOG2_CHECKSUMS record between the sampling and the
+		 * insertion below.  Without this, the redo record could follow the
+		 * transition record in WAL while carrying the pre-transition state,
+		 * and recovery resuming here would never learn about the transition.
+		 * See XLogChecksums().
+		 */
+		LWLockAcquire(DataChecksumTransitionLock, LW_EXCLUSIVE);
 		WALInsertLockAcquire();
 		redo_rec.wal_level = wal_level;
 		SpinLockAcquire(&XLogCtl->info_lck);
 		redo_rec.data_checksum_version = XLogCtl->data_checksum_version;
+		checksumLsn = XLogCtl->data_checksum_lsn;
 		SpinLockRelease(&XLogCtl->info_lck);
 		WALInsertLockRelease();
 
@@ -7620,6 +7854,14 @@ CreateCheckPoint(int flags)
 		XLogBeginInsert();
 		XLogRegisterData(&redo_rec, sizeof(xl_checkpoint_redo));
 		(void) XLogInsert(RM_XLOG_ID, XLOG_CHECKPOINT_REDO);
+		LWLockRelease(DataChecksumTransitionLock);
+
+		/*
+		 * The checkpoint record must carry the same state as the redo record
+		 * just inserted: the sample taken before redo determination can be
+		 * stale by now, and the pair would otherwise disagree.
+		 */
+		checkPoint.dataChecksumState = redo_rec.data_checksum_version;
 
 		/*
 		 * XLogInsertRecord will have updated XLogCtl->Insert.RedoRecPtr in
@@ -7823,6 +8065,40 @@ CreateCheckPoint(int flags)
 	ControlFile->minRecoveryPoint = InvalidXLogRecPtr;
 	ControlFile->minRecoveryPointTLI = 0;
 
+	/*
+	 * Persist the data checksum state this node runs under.  Only the
+	 * top-level field tracks this node; ControlFile->checkPointCopy above is
+	 * a historical record used to resume replay.
+	 *
+	 * checkPoint.dataChecksumState was sampled under
+	 * DataChecksumTransitionLock together with the redo record, so it is the
+	 * state in effect at the redo point.  If it was "on",
+	 * the XLOG2_CHECKSUMS record announcing that precedes the redo point and
+	 * every page the transition rewrote was dirtied before it, so
+	 * CheckPointGuts() has just written all of them out.  Recording the state
+	 * here is what keeps a finished transition from being resolved as
+	 * interrupted when this checkpoint is the one crash recovery resumes
+	 * from: replay never sees the record announcing it.
+	 *
+	 * Persist it only if the state did not change while the flush was in
+	 * progress.  If it changed in between, the pages written out straddle two
+	 * states, and the newer one could claim checksums that pages already on
+	 * disk do not carry; leave the field to the next checkpoint then, the
+	 * transition itself has already persisted every state that is safe
+	 * without a flush.
+	 * Compare the watermark rather than the state: record positions are
+	 * unique, so a full round trip back to the sampled state cannot alias,
+	 * while its flushed pages straddle the intermediate states all the same.
+	 */
+	SpinLockAcquire(&XLogCtl->info_lck);
+	if (checksumLsn == XLogCtl->data_checksum_lsn)
+	{
+		ControlFile->data_checksum_version = checkPoint.dataChecksumState;
+		ControlFile->data_checksum_lsn = checksumLsn;
+		ControlFile->data_checksum_is_local = XLogCtl->data_checksum_is_local;
+	}
+	SpinLockRelease(&XLogCtl->info_lck);
+
 	/*
 	 * Persist unloggedLSN value. It's reset on crash recovery, so this goes
 	 * unused on non-shutdown checkpoints, but seems useful to store it always
@@ -7967,9 +8243,11 @@ CreateEndOfRecoveryRecord(void)
 	ControlFile->minRecoveryPoint = recptr;
 	ControlFile->minRecoveryPointTLI = xlrec.ThisTimeLineID;
 
-	/* start with the latest checksum version (as of the end of recovery) */
+	/* persist the data checksum state this node ended recovery with */
 	SpinLockAcquire(&XLogCtl->info_lck);
 	ControlFile->data_checksum_version = XLogCtl->data_checksum_version;
+	ControlFile->data_checksum_lsn = XLogCtl->data_checksum_lsn;
+	ControlFile->data_checksum_is_local = XLogCtl->data_checksum_is_local;
 	SpinLockRelease(&XLogCtl->info_lck);
 
 	UpdateControlFile();
@@ -8174,6 +8452,9 @@ CreateRestartPoint(int flags)
 	XLogRecPtr	endptr;
 	XLogSegNo	_logSegNo;
 	TimestampTz xtime;
+	uint32		checksum_state;
+	XLogRecPtr	checksum_lsn;
+	bool		checksum_is_local;
 
 	/* Concurrent checkpoint/restartpoint cannot happen */
 	Assert(!IsUnderPostmaster || MyBackendType == B_CHECKPOINTER);
@@ -8220,8 +8501,45 @@ CreateRestartPoint(int flags)
 		UpdateMinRecoveryPoint(InvalidXLogRecPtr, true);
 		if (flags & CHECKPOINT_IS_SHUTDOWN)
 		{
+			bool		catchUpChecksums;
+
+			/*
+			 * There is no new restartpoint to persist the data checksum state
+			 * with, but a cleanly stopped node should not leave the control
+			 * file behind the state replay reached: pg_checksums and
+			 * pg_rewind read it, and an in-progress state there makes them
+			 * refuse to run.  Catching it up needs the same guarantee a
+			 * restartpoint gives: that every page on disk carries a checksum,
+			 * so flush the buffer pool first.  Only a transition to
+			 * "on" that no restartpoint followed can get here; the other
+			 * states are already persisted by XLOG2_CHECKSUMS replay.  Replay
+			 * has ended by now, so the state cannot change under us.
+			 */
+			SpinLockAcquire(&XLogCtl->info_lck);
+			checksum_state = XLogCtl->data_checksum_version;
+			checksum_lsn = XLogCtl->data_checksum_lsn;
+			checksum_is_local = XLogCtl->data_checksum_is_local;
+			SpinLockRelease(&XLogCtl->info_lck);
+
+			catchUpChecksums =
+				(checksum_lsn != ControlFile->data_checksum_lsn &&
+				 XLogRecPtrIsValid(lastCheckPointRecPtr));
+
+			if (catchUpChecksums)
+			{
+				MemSet(&CheckpointStats, 0, sizeof(CheckpointStats));
+				CheckpointStats.ckpt_start_t = GetCurrentTimestamp();
+				CheckPointGuts(lastCheckPoint.redo, flags);
+			}
+
 			LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
 			ControlFile->state = DB_SHUTDOWNED_IN_RECOVERY;
+			if (catchUpChecksums)
+			{
+				ControlFile->data_checksum_version = checksum_state;
+				ControlFile->data_checksum_lsn = checksum_lsn;
+				ControlFile->data_checksum_is_local = checksum_is_local;
+			}
 			UpdateControlFile();
 			LWLockRelease(ControlFileLock);
 		}
@@ -8262,6 +8580,17 @@ CreateRestartPoint(int flags)
 	/* Update the process title */
 	update_checkpoint_display(flags, true, false);
 
+	/*
+	 * Note the data checksum state the flush below starts under.  Replay runs
+	 * concurrently and can change the state while the flush is in progress,
+	 * in which case the flush covers pages written under both states; see
+	 * where the state is persisted further down.
+	 */
+	SpinLockAcquire(&XLogCtl->info_lck);
+	checksum_state = XLogCtl->data_checksum_version;
+	checksum_lsn = XLogCtl->data_checksum_lsn;
+	SpinLockRelease(&XLogCtl->info_lck);
+
 	CheckPointGuts(lastCheckPoint.redo, flags);
 
 	/*
@@ -8321,8 +8650,26 @@ CreateRestartPoint(int flags)
 				ControlFile->state = DB_SHUTDOWNED_IN_RECOVERY;
 		}
 
-		/* we shall start with the latest checksum version */
-		ControlFile->data_checksum_version = lastCheckPoint.dataChecksumState;
+		/*
+		 * Persist the data checksum state of this node.  Not the state of the
+		 * replayed checkpoint: that one belongs to the node that wrote it and
+		 * may differ after an offline change on either side.
+		 * ControlFile->checkPointCopy above keeps the replayed value on
+		 * purpose, being a historical record used to resume replay rather
+		 * than a tracker of node state.
+		 *
+		 * Persist only if the flush above ran under one state throughout; see
+		 * CreateCheckPoint() for why, including why this compares the
+		 * watermark and not the state.
+		 */
+		SpinLockAcquire(&XLogCtl->info_lck);
+		if (checksum_lsn == XLogCtl->data_checksum_lsn)
+		{
+			ControlFile->data_checksum_version = checksum_state;
+			ControlFile->data_checksum_lsn = checksum_lsn;
+			ControlFile->data_checksum_is_local = XLogCtl->data_checksum_is_local;
+		}
+		SpinLockRelease(&XLogCtl->info_lck);
 
 		UpdateControlFile();
 	}
@@ -8763,9 +9110,21 @@ XLogReportParameters(void)
 }
 
 /*
- * Log the new state of checksums
+ * Log and publish the new state of checksums
+ *
+ * Inserting the record and publishing the new state must be atomic with
+ * respect to a checkpoint sampling the state for its XLOG_CHECKPOINT_REDO
+ * record: without that, a checkpoint could read the old state after the
+ * record is already in WAL and insert a redo record that both precedes the
+ * transition in WAL order and carries the pre-transition state.  Recovery
+ * resuming from such a redo point would never replay the transition record
+ * and resolve the finished transition as interrupted.
+ * DataChecksumTransitionLock serializes the two; see CreateCheckPoint().
+ *
+ * Returns the end LSN of the inserted record, which the caller persists
+ * together with the new state as the data checksum watermark.
  */
-static void
+static XLogRecPtr
 XLogChecksums(uint32 new_type)
 {
 	xl_checksum_state xlrec;
@@ -8773,12 +9132,28 @@ XLogChecksums(uint32 new_type)
 
 	xlrec.new_checksum_state = new_type;
 
+	LWLockAcquire(DataChecksumTransitionLock, LW_EXCLUSIVE);
+
 	XLogBeginInsert();
 	XLogRegisterData((char *) &xlrec, sizeof(xl_checksum_state));
 
 	recptr = XLogInsert(RM_XLOG2_ID, XLOG2_CHECKSUMS);
 	pg_atomic_write_u64(&XLogCtl->lastChecksumChangeRecPtr, recptr);
+
+	/* only loaded by SetDataChecksumsOn(), a no-op for the other callers */
+	INJECTION_POINT_CACHED("datachecksums-on-before-publish", NULL);
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	XLogCtl->data_checksum_version = new_type;
+	XLogCtl->data_checksum_lsn = recptr;
+	XLogCtl->data_checksum_is_local = false;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	LWLockRelease(DataChecksumTransitionLock);
+
 	XLogFlush(recptr);
+
+	return recptr;
 }
 
 /*
@@ -8966,11 +9341,19 @@ xlog_redo(XLogReaderState *record)
 		/* ControlFile->checkPointCopy always tracks the latest ckpt XID */
 		LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
 		ControlFile->checkPointCopy.nextXid = checkPoint.nextXid;
-		ControlFile->data_checksum_version = checkPoint.dataChecksumState;
 
 		UpdateControlFile();
 		LWLockRelease(ControlFileLock);
 
+		if (adoptChecksumStateFromNextCheckpoint)
+		{
+			adoptChecksumStateFromNextCheckpoint = false;
+			AdoptReplayedDataChecksumState(checkPoint.dataChecksumState,
+										   record->ReadRecPtr);
+		}
+		else
+			CheckReplayedDataChecksumState(checkPoint.dataChecksumState);
+
 		/*
 		 * We should've already switched to the new TLI before replaying this
 		 * record.
@@ -9206,19 +9589,17 @@ xlog_redo(XLogReaderState *record)
 	else if (info == XLOG_CHECKPOINT_REDO)
 	{
 		xl_checkpoint_redo redo_rec;
-		bool		new_state = false;
 
 		memcpy(&redo_rec, XLogRecGetData(record), sizeof(xl_checkpoint_redo));
 
-		SpinLockAcquire(&XLogCtl->info_lck);
-		XLogCtl->data_checksum_version = redo_rec.data_checksum_version;
-		SetLocalDataChecksumState(redo_rec.data_checksum_version);
-		if (redo_rec.data_checksum_version != ControlFile->data_checksum_version)
-			new_state = true;
-		SpinLockRelease(&XLogCtl->info_lck);
-
-		if (new_state)
-			EmitAndWaitDataChecksumsBarrier(redo_rec.data_checksum_version);
+		if (adoptChecksumStateFromNextCheckpoint)
+		{
+			adoptChecksumStateFromNextCheckpoint = false;
+			AdoptReplayedDataChecksumState(redo_rec.data_checksum_version,
+										   record->ReadRecPtr);
+		}
+		else
+			CheckReplayedDataChecksumState(redo_rec.data_checksum_version);
 	}
 	else if (info == XLOG_LOGICAL_DECODING_STATUS_CHANGE)
 	{
@@ -9280,25 +9661,63 @@ xlog2_redo(XLogReaderState *record)
 	{
 		xl_checksum_state state;
 		XLogRecPtr	lsn = record->EndRecPtr;
+		XLogRecPtr	watermark;
 
 		memcpy(&state, XLogRecGetData(record), sizeof(xl_checksum_state));
 
+		SpinLockAcquire(&XLogCtl->info_lck);
+		watermark = XLogCtl->data_checksum_lsn;
+		SpinLockRelease(&XLogCtl->info_lck);
+
+		/*
+		 * Skip records this node has already applied.  The control file
+		 * carries the watermark, so this holds across restarts: recovery
+		 * resuming below a record whose effect the control file already
+		 * contains must not re-apply it, or it would revert a state change
+		 * made with pg_checksums in between, which moves the state without
+		 * writing any record of its own.
+		 */
+		if (lsn <= watermark)
+			return;
+
 		/* advertise the location before the new state becomes visible */
 		pg_atomic_write_u64(&XLogCtl->lastChecksumChangeRecPtr, lsn);
 
 		SpinLockAcquire(&XLogCtl->info_lck);
 		XLogCtl->data_checksum_version = state.new_checksum_state;
+		XLogCtl->data_checksum_lsn = lsn;
+		XLogCtl->data_checksum_is_local = false;
+		SetLocalDataChecksumState(state.new_checksum_state);
 		SpinLockRelease(&XLogCtl->info_lck);
 
 		LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
-		ControlFile->data_checksum_version = state.new_checksum_state;
+
+		/*
+		 * Persist the new state, except when it is "on".  Only "on" verifies
+		 * checksums during reads, and between the last restartpoint and this
+		 * record there may be pages on disk flushed under the old state; a
+		 * crash-restart initializes verification from the control file and
+		 * replay reads those pages back, so the control file may only say
+		 * "on" once everything written under the transition has been flushed,
+		 * as restartpoints and the end of recovery do.
+		 * The opposite direction cannot wait for the restartpoint: once this
+		 * record is replayed, evicted pages are written without checksums,
+		 * and a control file still saying "on" would fail verification on
+		 * exactly those pages after a crash.
+		 */
+		if (state.new_checksum_state != PG_DATA_CHECKSUM_VERSION)
+		{
+			ControlFile->data_checksum_version = state.new_checksum_state;
+			ControlFile->data_checksum_lsn = lsn;
+			ControlFile->data_checksum_is_local = false;
+		}
 
 		/*
 		 * Update minRecoveryPoint to ensure that if recovery is aborted, we
 		 * recover back up to this point before allowing hot standby again.
-		 * The new state is durable in pg_control while its location is only
-		 * tracked in shared memory; a standby becoming consistent below this
-		 * record would let base backups resume checksum verification with the
+		 * The change location is only tracked in shared memory and is lost
+		 * over a restart; a standby becoming consistent below this record
+		 * would let base backups resume checksum verification with the
 		 * location unknown.  The local copies cannot be updated as long as
 		 * crash recovery is happening and we expect all the WAL to be
 		 * replayed.
diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt
index 256b3a3c02e..77fa543f9d8 100644
--- a/src/backend/utils/activity/wait_event_names.txt
+++ b/src/backend/utils/activity/wait_event_names.txt
@@ -371,6 +371,7 @@ WaitLSN	"Waiting to read or update shared Wait-for-LSN state."
 LogicalDecodingControl	"Waiting to read or update logical decoding status information."
 DataChecksumsWorker	"Waiting for data checksums worker."
 AioWorkerControl	"Waiting to update AIO worker information."
+DataChecksumTransition	"Waiting for a data checksum state transition to be written to WAL."
 
 #
 # END OF PREDEFINED LWLOCKS (DO NOT CHANGE THIS LINE)
diff --git a/src/bin/pg_checksums/pg_checksums.c b/src/bin/pg_checksums/pg_checksums.c
index 3b3ae23f1a6..20a99112838 100644
--- a/src/bin/pg_checksums/pg_checksums.c
+++ b/src/bin/pg_checksums/pg_checksums.c
@@ -648,6 +648,16 @@ main(int argc, char *argv[])
 		ControlFile->data_checksum_version =
 			(mode == PG_MODE_ENABLE) ? PG_DATA_CHECKSUM_VERSION : PG_DATA_CHECKSUM_OFF;
 
+		/*
+		 * Mark the state as changed locally, without a WAL record.  Recovery
+		 * then knows the state is newer than anything the WAL carries and
+		 * does not let a replayed checkpoint overwrite it.  The watermark is
+		 * left alone: any XLOG2_CHECKSUMS record this node had applied stays
+		 * covered, and only records above it, written after this change, take
+		 * effect again.
+		 */
+		ControlFile->data_checksum_is_local = true;
+
 		if (do_sync)
 		{
 			pg_log_info("syncing data directory");
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c
index 6fc87ed114d..c363ce3dbb7 100644
--- a/src/bin/pg_controldata/pg_controldata.c
+++ b/src/bin/pg_controldata/pg_controldata.c
@@ -349,6 +349,10 @@ main(int argc, char *argv[])
 		   (ControlFile->float8ByVal ? _("by value") : _("by reference")));
 	printf(_("Data page checksum version:           %u\n"),
 		   ControlFile->data_checksum_version);
+	printf(_("Data checksum watermark:              %X/%08X\n"),
+		   LSN_FORMAT_ARGS(ControlFile->data_checksum_lsn));
+	printf(_("Data checksum state is node-local:    %s\n"),
+		   (ControlFile->data_checksum_is_local ? _("yes") : _("no")));
 	printf(_("Default char data signedness:         %s\n"),
 		   (ControlFile->default_char_signedness ? _("signed") : _("unsigned")));
 	printf(_("Mock authentication nonce:            %s\n"),
diff --git a/src/bin/pg_resetwal/pg_resetwal.c b/src/bin/pg_resetwal/pg_resetwal.c
index 1542a56ca4b..cdfb9898066 100644
--- a/src/bin/pg_resetwal/pg_resetwal.c
+++ b/src/bin/pg_resetwal/pg_resetwal.c
@@ -923,6 +923,13 @@ RewriteControlFile(void)
 	ControlFile.backupEndPoint = InvalidXLogRecPtr;
 	ControlFile.backupEndRequired = false;
 
+	/*
+	 * The old WAL is gone and the new position may lie below the old
+	 * watermark, which would make replay ignore future checksum transition
+	 * records.  The state itself is kept.
+	 */
+	ControlFile.data_checksum_lsn = InvalidXLogRecPtr;
+
 	/*
 	 * Force the defaults for max_* settings. The values don't really matter
 	 * as long as wal_level='minimal'; the postmaster will reset these fields
diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c
index 2e86fd158d0..466f4223501 100644
--- a/src/bin/pg_rewind/pg_rewind.c
+++ b/src/bin/pg_rewind/pg_rewind.c
@@ -37,7 +37,8 @@ static void usage(const char *progname);
 static void perform_rewind(filemap_t *filemap, rewind_source *source,
 						   XLogRecPtr chkptrec,
 						   TimeLineID chkpttli,
-						   XLogRecPtr chkptredo);
+						   XLogRecPtr chkptredo,
+						   XLogRecPtr divergerec);
 
 static void createBackupLabel(XLogRecPtr startpoint, TimeLineID starttli,
 							  XLogRecPtr checkpointloc);
@@ -531,7 +532,8 @@ main(int argc, char **argv)
 	 * This is the point of no return. Once we start copying things, there is
 	 * no turning back!
 	 */
-	perform_rewind(filemap, source, chkptrec, chkpttli, chkptredo);
+	perform_rewind(filemap, source, chkptrec, chkpttli, chkptredo,
+				   divergerec);
 
 	if (showprogress)
 		pg_log_info("syncing target data directory");
@@ -566,7 +568,8 @@ static void
 perform_rewind(filemap_t *filemap, rewind_source *source,
 			   XLogRecPtr chkptrec,
 			   TimeLineID chkpttli,
-			   XLogRecPtr chkptredo)
+			   XLogRecPtr chkptredo,
+			   XLogRecPtr divergerec)
 {
 	XLogRecPtr	endrec;
 	TimeLineID	endtli;
@@ -738,6 +741,32 @@ perform_rewind(filemap_t *filemap, rewind_source *source,
 	ControlFile_new.minRecoveryPoint = endrec;
 	ControlFile_new.minRecoveryPointTLI = endtli;
 	ControlFile_new.state = DB_IN_ARCHIVE_RECOVERY;
+
+	/*
+	 * Keep the target's own data checksum state.  Most of the data directory
+	 * is still the target's: only blocks it changed since the divergence
+	 * were copied from the source, so the source's state says nothing about
+	 * the pages that stay.  Replay from the last common checkpoint applies
+	 * any WAL-logged transition the target has not seen (the watermark tells
+	 * them apart), which converges the rewound server to the source's state
+	 * whenever the WAL carries it.
+	 */
+	ControlFile_new.data_checksum_version = ControlFile_target.data_checksum_version;
+	ControlFile_new.data_checksum_lsn = ControlFile_target.data_checksum_lsn;
+	ControlFile_new.data_checksum_is_local = ControlFile_target.data_checksum_is_local;
+
+	/*
+	 * The watermark is only meaningful within the history the node replays.
+	 * Records at or below the divergence point are common to both histories
+	 * and stay covered, but a watermark above it was set by a transition
+	 * record on the target's own abandoned fork: numerically it can cover
+	 * transition records the source wrote after the divergence, and replay
+	 * would skip them as already applied.  Clamp it to the divergence point,
+	 * so that every transition record on the source's history takes effect.
+	 */
+	if (ControlFile_new.data_checksum_lsn > divergerec)
+		ControlFile_new.data_checksum_lsn = divergerec;
+
 	if (!dry_run)
 		update_controlfile(datadir_target, &ControlFile_new, do_sync);
 }
diff --git a/src/bin/pg_upgrade/controldata.c b/src/bin/pg_upgrade/controldata.c
index b3bd4ccde83..de539479bdc 100644
--- a/src/bin/pg_upgrade/controldata.c
+++ b/src/bin/pg_upgrade/controldata.c
@@ -431,7 +431,7 @@ get_control_data(ClusterInfo *cluster)
 			cluster->controldata.date_is_int = strstr(p, "64-bit integers") != NULL;
 			got_date_is_int = true;
 		}
-		else if ((p = strstr(bufin, "checksum")) != NULL)
+		else if ((p = strstr(bufin, "Data page checksum version:")) != NULL)
 		{
 			p = strchr(p, ':');
 
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index 7b5404460ec..6a83316827d 100644
--- a/src/include/catalog/pg_control.h
+++ b/src/include/catalog/pg_control.h
@@ -22,7 +22,7 @@
 
 
 /* Version identifier for this pg_control format */
-#define PG_CONTROL_VERSION	1903
+#define PG_CONTROL_VERSION	1904
 
 /* Nonce key length, see below */
 #define MOCK_AUTH_NONCE_LEN		32
@@ -237,6 +237,32 @@ typedef struct ControlFileData
 	/* Current data checksums state */
 	uint32		data_checksum_version;
 
+	/*
+	 * WAL position through which data checksum transitions are covered.
+	 * Replay ignores XLOG2_CHECKSUMS records ending at or below this point:
+	 * their effect is already contained in data_checksum_version, or an
+	 * offline pg_checksums change made after they were first applied
+	 * supersedes them.  Ordinarily this is the end of the newest such record
+	 * this node has written or applied, but a tool may store any position
+	 * that covers the same set of records.  InvalidXLogRecPtr if the node
+	 * has never written or applied such a record.
+	 *
+	 * The comparison has no timeline context, so the value is only valid
+	 * within the WAL history this node replays.  A tool that moves the node
+	 * to another history must clamp the watermark to the point where the
+	 * histories fork, as pg_rewind does, or reset it, as pg_resetwal does.
+	 */
+	XLogRecPtr	data_checksum_lsn;
+
+	/*
+	 * True when data_checksum_version was last set by pg_checksums rather
+	 * than by a WAL-logged transition.  Such a state is local to this node
+	 * and newer than anything the WAL carries, so recovery must not replace
+	 * it with a state taken from a checkpoint record.  Cleared by the next
+	 * WAL-logged transition.
+	 */
+	bool		data_checksum_is_local;
+
 	/*
 	 * True if the default signedness of char is "signed" on a platform where
 	 * the cluster is initialized.
diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h
index d7eb648bd27..8d858be9927 100644
--- a/src/include/storage/lwlocklist.h
+++ b/src/include/storage/lwlocklist.h
@@ -89,6 +89,7 @@ PG_LWLOCK(54, WaitLSN)
 PG_LWLOCK(55, LogicalDecodingControl)
 PG_LWLOCK(56, DataChecksumsWorker)
 PG_LWLOCK(57, AioWorkerControl)
+PG_LWLOCK(58, DataChecksumTransition)
 
 /*
  * There also exist several built-in LWLock tranches.  As with the predefined
diff --git a/src/test/modules/test_checksums/Makefile b/src/test/modules/test_checksums/Makefile
index 71455cd5577..80f54bf6d8c 100644
--- a/src/test/modules/test_checksums/Makefile
+++ b/src/test/modules/test_checksums/Makefile
@@ -9,7 +9,7 @@
 #
 #-------------------------------------------------------------------------
 
-EXTRA_INSTALL = src/test/modules/injection_points
+EXTRA_INSTALL = contrib/pg_buffercache src/test/modules/injection_points
 
 export enable_injection_points
 
diff --git a/src/test/modules/test_checksums/meson.build b/src/test/modules/test_checksums/meson.build
index fb7129d796f..e5d38fafb7d 100644
--- a/src/test/modules/test_checksums/meson.build
+++ b/src/test/modules/test_checksums/meson.build
@@ -35,6 +35,16 @@ tests += {
       't/009_fpi.pl',
       't/010_backup_straddle.pl',
       't/011_standby_straddle.pl',
+      't/012_offline_standby.pl',
+      't/013_rewind.pl',
+      't/014_lockstep.pl',
+      't/015_standby_crash_after_disable.pl',
+      't/016_promote_enable_crash.pl',
+      't/017_restartpoint_race.pl',
+      't/018_enable_crash_windows.pl',
+      't/019_standby_shutdown_catchup.pl',
+      't/020_cascade_divergence.pl',
+      't/021_rewind_divergent_transitions.pl',
     ],
   },
 }
diff --git a/src/test/modules/test_checksums/t/012_offline_standby.pl b/src/test/modules/test_checksums/t/012_offline_standby.pl
new file mode 100644
index 00000000000..241a1282ae9
--- /dev/null
+++ b/src/test/modules/test_checksums/t/012_offline_standby.pl
@@ -0,0 +1,289 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# Offline checksum changes with pg_checksums are local to one node.  A
+# standby must neither adopt the state of the primary from replayed
+# checkpoint records, nor lose its own offline change to them.
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use DataChecksums::Utils;
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 1, no_data_checksums => 1);
+$primary->append_conf('postgresql.conf', 'autovacuum = off');
+$primary->start;
+$primary->safe_psql('postgres',
+	"CREATE TABLE t AS SELECT generate_series(1,10000) AS a;");
+
+$primary->backup('backup');
+my $standby = PostgreSQL::Test::Cluster->new('standby');
+$standby->init_from_backup($primary, 'backup', has_streaming => 1);
+$standby->start;
+$primary->wait_for_catchup($standby);
+
+test_checksum_state($primary, 'off');
+test_checksum_state($standby, 'off');
+
+# Scenario 1: enable offline on the primary only.  The standby must
+# stay off, warn about the mismatch, and remain readable.
+$standby->stop;
+$primary->stop;
+$primary->checksum_enable_offline;
+$primary->start;
+$standby->start;
+
+test_checksum_state($primary, 'on');
+test_checksum_state($standby, 'off');
+
+my $logstart = -s $standby->logfile;
+$primary->safe_psql('postgres', "INSERT INTO t VALUES (0);");
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->wait_for_catchup($standby);
+
+test_checksum_state($standby, 'off');
+is($standby->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10001', 'standby readable after offline enable on the primary');
+
+$standby->wait_for_log(qr/does not match the state "on" in the replayed WAL/,
+	$logstart);
+
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->wait_for_catchup($standby);
+my $log = PostgreSQL::Test::Utils::slurp_file($standby->logfile, $logstart);
+my @warnings = $log =~ /(does not match the state)/g;
+is(scalar(@warnings), 1, 'mismatch warned once per remote value');
+
+# Matching states re-arm the warning: undo the divergence on the primary,
+# then diverge again to the same value, all without restarting the standby.
+$primary->stop;
+$primary->checksum_disable_offline;
+$logstart = -s $standby->logfile;
+$primary->start;
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->wait_for_catchup($standby);
+$log = PostgreSQL::Test::Utils::slurp_file($standby->logfile, $logstart);
+unlike(
+	$log,
+	qr/does not match the state/,
+	'no warning while the states match again');
+like($log, qr/now agrees with the replayed WAL/,
+	'convergence is reported once the states match again');
+
+$primary->stop;
+$primary->checksum_enable_offline;
+$primary->start;
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->wait_for_catchup($standby);
+$standby->wait_for_log(qr/does not match the state "on" in the replayed WAL/,
+	$logstart);
+test_checksum_state($standby, 'off');
+
+# The local state survives both clean and immediate restarts.
+$standby->restart;
+test_checksum_state($standby, 'off');
+$standby->stop('immediate');
+$standby->start;
+test_checksum_state($standby, 'off');
+
+# Still in scenario 1's divergence: take a base backup *from the
+# standby*.  A backup taken on a standby uses the last restartpoint as
+# its starting checkpoint (do_pg_backup_start()), so the record at the
+# redo point was written by the upstream primary and carries the
+# primary's state.  Recovery from such a backup must not adopt that
+# state: the files were copied from the standby, and the new node
+# would come up verifying checksums its files do not have.
+
+# Make sure the standby has written pages under its own "off" state, so
+# its files really do lack checksums.
+$primary->safe_psql('postgres',
+	"CREATE TABLE t2 AS SELECT generate_series(1,50000) AS a;");
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->wait_for_catchup($standby);
+$standby->safe_psql('postgres', "SELECT count(*) FROM t2;");
+
+$standby->backup('from_standby');
+my $newnode = PostgreSQL::Test::Cluster->new('newnode');
+$newnode->init_from_backup($standby, 'from_standby');
+
+# Stream from the primary, not from the standby the files were copied
+# from, so the new node replays the "on" primary's records.
+$newnode->enable_streaming($primary);
+$newnode->start;
+
+# The new node's files all came from a cluster running with checksums
+# off.  Anything but "off" here means it adopted the primary's state
+# through the checkpoint record at the redo point.
+my ($rc, $stdout, $stderr) = $newnode->psql('postgres',
+	"SELECT setting FROM pg_settings WHERE name = 'data_checksums';");
+is($rc, 0, 'the node copied from the standby accepts connections')
+  or diag("stderr: $stderr");
+is($stdout, 'off', 'backup of an "off" standby comes up with checksums off');
+
+# A checkpoint record streamed from the "on" primary must not flip the
+# state either.
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->wait_for_catchup($newnode);
+test_checksum_state($newnode, 'off');
+
+# And it must be able to read the pages the standby wrote without
+# checksums.
+(undef, $stdout, $stderr) =
+  $newnode->psql('postgres', "SELECT count(*) FROM t2;");
+is($stdout, '50000', 'pages copied from the standby are readable')
+  or diag("stderr: $stderr");
+
+my $newnode_log = PostgreSQL::Test::Utils::slurp_file($newnode->logfile);
+unlike(
+	$newnode_log,
+	qr/page verification failed/,
+	'no checksum verification failures on the node copied from the standby');
+
+$newnode->stop('immediate');
+
+# Converge the cluster: enable offline on the standby too.
+$standby->stop;
+$standby->checksum_enable_offline;
+$standby->start;
+test_checksum_state($standby, 'on');
+$primary->wait_for_catchup($standby);
+is($standby->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10001', 'standby readable after converging');
+
+# Scenario 2: disable offline on the standby only.  The replayed
+# checkpoint records of the still-enabled primary must not override it.
+$standby->stop;
+$standby->checksum_disable_offline;
+$standby->start;
+test_checksum_state($standby, 'off');
+test_checksum_state($primary, 'on');
+
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->wait_for_catchup($standby);
+test_checksum_state($standby, 'off');
+
+# Restartpoints must persist the local state, not the replayed copy.
+$standby->safe_psql('postgres', "CHECKPOINT;");
+$standby->restart;
+test_checksum_state($standby, 'off');
+$standby->stop('immediate');
+$standby->start;
+test_checksum_state($standby, 'off');
+
+is($standby->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10001', 'standby readable with checksums disabled locally');
+
+# Scenario 3: crash-restart right after an online transition, before the
+# next restartpoint.  Replay then resumes from an older restartpoint whose
+# checkpoint records still carry the pre-transition state.  Those must
+# still match the state seeded from the control file, and the transition
+# itself must be re-established by re-replaying the XLOG2_CHECKSUMS record,
+# without a spurious mismatch warning along the way.
+
+# Converge first: bring the standby back to "on" offline.
+$standby->stop;
+$standby->checksum_enable_offline;
+$standby->start;
+test_checksum_state($standby, 'on');
+test_checksum_state($primary, 'on');
+
+disable_data_checksums($primary, wait => 'off');
+$primary->wait_for_catchup($standby);
+wait_for_checksum_state($standby, 'off');
+
+# Crash-restart the standby immediately, before any restartpoint has had a
+# chance to persist the new state to its control file.
+$logstart = -s $standby->logfile;
+$standby->stop('immediate');
+$standby->start;
+$primary->wait_for_catchup($standby);
+
+test_checksum_state($standby, 'off');
+is($standby->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10001',
+	'standby readable after crash-restart across an online transition');
+
+$log = PostgreSQL::Test::Utils::slurp_file($standby->logfile, $logstart);
+unlike(
+	$log,
+	qr/does not match the state/,
+	'no spurious mismatch warning after crash-restart across an online transition'
+);
+
+# Scenario 4: a standby stopped while replaying an interrupted online
+# transition keeps the interrupted state in its own control file.  A
+# primary is never caught this way, as its checksums launcher resolves
+# inprogress-on back to off from its exit cleanup.  A standby has no
+# launcher; it carries forward whatever the last replayed record left
+# it in.
+
+# Block an online enable on the primary at inprogress-on with a
+# blocking temp table, same trick as in 004_offline.pl.
+my $bsession = $primary->background_psql('postgres');
+$bsession->query_safe('CREATE TEMPORARY TABLE tt (a integer);');
+enable_data_checksums($primary, wait => 'inprogress-on');
+
+# The standby picks up the in-progress state from the XLOG2_CHECKSUMS record.
+wait_for_checksum_state($standby, 'inprogress-on');
+
+# Stop the standby cleanly; its restartpoint persists inprogress-on to
+# its own control file, since nothing on a standby resolves it away.
+$standby->stop;
+$standby->start;
+wait_for_checksum_state($standby, 'inprogress-on');
+
+# The primary still sits at inprogress-on, so a base backup taken now
+# copies a control file and a redo point that both carry the
+# in-progress state; replay of the XLOG2_CHECKSUMS records completes
+# the transition on the new standby.
+$primary->backup('inprogress_backup');
+my $standby2 = PostgreSQL::Test::Cluster->new('standby2');
+$standby2->init_from_backup($primary, 'inprogress_backup',
+	has_streaming => 1);
+$standby2->start;
+
+# Backup label recovery adopts the redo point's state immediately,
+# before any further WAL is replayed.
+wait_for_checksum_state($standby2, 'inprogress-on');
+
+$bsession->quit;
+wait_for_checksum_state($primary, 'on');
+$primary->wait_for_catchup($standby);
+wait_for_checksum_state($standby, 'on');
+
+is( $standby->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10001', 'standby readable once the transition completes');
+
+# Scenario 4 continued: the backup taken mid-transition must also
+# complete the transition and stay readable.
+$primary->wait_for_catchup($standby2);
+wait_for_checksum_state($standby2, 'on');
+
+is($standby2->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10001', 'mid-transition backup standby readable after the transition');
+
+$standby2->stop;
+
+# Nothing along this standby's replay can legitimately disagree: it
+# started out at the same inprogress-on state as the primary.
+my $standby2_log = PostgreSQL::Test::Utils::slurp_file($standby2->logfile);
+unlike(
+	$standby2_log,
+	qr/does not match the state/,
+	'no mismatch warning on a backup taken mid-transition');
+
+# No extra CHECKPOINT needed: the transition's completion checkpoint
+# wrote every page with a checksum, and the shutdown restartpoint
+# flushes the rest.
+command_ok([ 'pg_checksums', '--check', '-D', $standby2->data_dir ],
+	'checksums valid on the mid-transition backup standby');
+
+$standby->stop;
+$primary->stop;
+done_testing();
diff --git a/src/test/modules/test_checksums/t/013_rewind.pl b/src/test/modules/test_checksums/t/013_rewind.pl
new file mode 100644
index 00000000000..a791e24317d
--- /dev/null
+++ b/src/test/modules/test_checksums/t/013_rewind.pl
@@ -0,0 +1,201 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# Test pg_rewind across an online data checksum enable.
+#
+# A clean switchover leaves the shutdown checkpoint of the old primary
+# as the last common checkpoint between the two nodes.  When data
+# checksums are enabled online on the new primary before the old one is
+# rewound, pg_rewind installs the control file of the new primary, which
+# already claims checksums are fully enabled, while replay begins at the
+# shutdown checkpoint whose record still carries the old state.  Replay
+# of the WAL stretch from before the enable must run with checksums off,
+# as recorded in the checkpoint, else it would verify pages which never
+# had checksums written and fail recovery.
+#
+# The new primary requests a checkpoint right after promotion, and
+# replaying its CHECKPOINT_REDO record would repair the state before
+# any interesting WAL is reached.  Hold the checkpointer on the standby
+# in a restartpoint over the promotion, like in the recovery test
+# 041_checkpoint_at_promote, so that the post-promotion writes end up
+# in WAL before the first checkpoint of the new timeline.
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use DataChecksums::Utils;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+	plan skip_all => 'Injection points not supported by this build';
+}
+
+# Old primary.  full_page_writes is off so that the updates done on the
+# promoted node do not carry full page images, forcing replay on the
+# rewound node to read the pages from disk.  wal_log_hints is required
+# by pg_rewind on a cluster without data checksums.
+my $node_a = PostgreSQL::Test::Cluster->new('node_a');
+$node_a->init(allows_streaming => 1, no_data_checksums => 1);
+$node_a->append_conf(
+	'postgresql.conf', qq[
+autovacuum = off
+full_page_writes = off
+wal_log_hints = on
+wal_keep_size = '1GB'
+log_checkpoints = on
+]);
+$node_a->start;
+
+if (!$node_a->check_extension('injection_points'))
+{
+	plan skip_all => 'Extension injection_points not installed';
+}
+
+$node_a->safe_psql('postgres',
+	"CREATE TABLE t AS SELECT generate_series(1,10000) AS a;");
+$node_a->safe_psql('postgres', "CREATE TABLE t_div (a int);");
+$node_a->safe_psql('postgres', "CREATE EXTENSION injection_points;");
+
+# Set the hint bits on t before taking the backup, so that reads on the
+# promoted node do not emit full page images for its pages later.
+$node_a->safe_psql('postgres', "CHECKPOINT;");
+$node_a->safe_psql('postgres', "SELECT count(*) FROM t;");
+
+$node_a->backup('backup');
+my $node_b = PostgreSQL::Test::Cluster->new('node_b');
+$node_b->init_from_backup($node_a, 'backup', has_streaming => 1);
+$node_b->start;
+
+$node_a->wait_for_catchup($node_b, 'replay', $node_a->lsn('insert'));
+test_checksum_state($node_a, 'off');
+test_checksum_state($node_b, 'off');
+
+# Hold the next restartpoint on the standby.
+$node_b->safe_psql('postgres',
+	"SELECT injection_points_attach('create-restart-point', 'wait');");
+
+# Give the restartpoint a checkpoint record to work on, then start it
+# in a background session; it will block on the injection point with
+# the checkpointer busy until released.
+$node_a->safe_psql('postgres', "CHECKPOINT;");
+$node_a->wait_for_catchup($node_b, 'replay', $node_a->lsn('insert'));
+
+my $bg_psql = $node_b->background_psql('postgres', on_error_stop => 0);
+$bg_psql->query_until(
+	qr/starting_restartpoint/, q(
+   \echo starting_restartpoint
+   CHECKPOINT;
+));
+$node_b->wait_for_event('checkpointer', 'create-restart-point');
+
+# Clean switchover: the shutdown checkpoint of A streams to B and
+# becomes the last common checkpoint.
+$node_a->stop('fast');
+
+my ($stdout, $stderr) = run_command([ 'pg_controldata', $node_a->data_dir ]);
+$stdout =~ /^Latest checkpoint location:\s*([0-9A-F\/]+)$/m
+  or die "checkpoint location missing from pg_controldata output";
+my $shutdown_ckpt = $1;
+$node_b->poll_query_until('postgres',
+	"SELECT pg_last_wal_replay_lsn() > '$shutdown_ckpt'::pg_lsn;")
+  or die "standby never replayed the shutdown checkpoint";
+
+my $logstart = -s $node_b->logfile;
+$node_b->promote;
+
+# Accidental restart of the old primary, diverging its timeline.
+$node_a->start;
+$node_a->safe_psql('postgres', "INSERT INTO t_div VALUES (1);");
+$node_a->stop('fast');
+
+# Updates on the new primary before its first checkpoint; replay of
+# these on the rewound node has to read the pages from disk.
+$node_b->safe_psql('postgres', "UPDATE t SET a = a WHERE a % 25 = 0;");
+ok( !$node_b->log_contains("checkpoint complete", $logstart),
+	"no checkpoint on the new timeline before the updates");
+
+# Release the checkpointer; the queued post-promotion checkpoint runs
+# after the updates.
+$node_b->safe_psql('postgres',
+	"SELECT injection_points_wakeup('create-restart-point');");
+$node_b->safe_psql('postgres',
+	"SELECT injection_points_detach('create-restart-point');");
+$bg_psql->quit;
+
+enable_data_checksums($node_b, wait => 'on');
+test_checksum_state($node_b, 'on');
+
+# pg_rewind refuses to run with full_page_writes disabled on the
+# source; the updates it was disabled for are already in WAL.
+$node_b->safe_psql('postgres', "ALTER SYSTEM SET full_page_writes = on;");
+$node_b->safe_psql('postgres', "SELECT pg_reload_conf();");
+
+command_ok(
+	[
+		'pg_rewind',
+		'--target-pgdata' => $node_a->data_dir,
+		'--source-server' => $node_b->connstr('postgres'),
+	],
+	'pg_rewind from the new primary');
+
+# Replay on the rewound node must start at the shutdown checkpoint of
+# the switchover, with the control file of the new primary.
+my $backup_label = slurp_file($node_a->data_dir . '/backup_label');
+$backup_label =~ /^CHECKPOINT LOCATION: ([0-9A-F\/]+)$/m
+  or die "checkpoint location missing from backup_label";
+is($1, $shutdown_ckpt, 'replay starts at the switchover checkpoint');
+
+($stdout, $stderr) = run_command(
+	[
+		'pg_waldump',
+		'-p' => $node_a->data_dir . '/pg_wal',
+		'-t' => 1,
+		'-s' => $shutdown_ckpt,
+		'-n' => 1,
+	]);
+like($stdout, qr/CHECKPOINT_SHUTDOWN/,
+	'last common checkpoint is a shutdown checkpoint');
+
+# pg_rewind keeps the target's own checksum state in the control file it
+# writes; the online enable reaches the rewound node through WAL replay
+# below, not through the copied control file.
+($stdout, $stderr) = run_command([ 'pg_controldata', $node_a->data_dir ]);
+like(
+	$stdout,
+	qr/^Data page checksum version:\s*0$/m,
+	'rewound node keeps its own checksum state in the control file');
+
+# Start the rewound node as a standby of the new primary.  Replay runs
+# through the pre-enable WAL stretch and the online enable.
+#
+# The rewind replaced the configuration files with those of the new
+# primary, so put the port back.
+my $connstr = $node_b->connstr;
+$node_a->append_conf(
+	'postgresql.conf', qq[
+port = @{[$node_a->port]}
+primary_conninfo = '$connstr application_name=@{[$node_a->name]}'
+]);
+$node_a->set_standby_mode;
+$node_a->start;
+
+$node_b->wait_for_catchup($node_a, 'replay', $node_b->lsn('insert'));
+test_checksum_state($node_a, 'on');
+
+is($node_a->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10000', 'data readable on the rewound node');
+is($node_a->safe_psql('postgres', "SELECT count(*) FROM t_div;"),
+	'0', 'divergent insert was rewound');
+
+$node_a->stop('fast');
+command_ok([ 'pg_checksums', '--check', '-D', $node_a->data_dir ],
+	'checksums valid on the rewound node');
+
+$node_b->stop;
+
+done_testing();
diff --git a/src/test/modules/test_checksums/t/014_lockstep.pl b/src/test/modules/test_checksums/t/014_lockstep.pl
new file mode 100644
index 00000000000..713f37fb22c
--- /dev/null
+++ b/src/test/modules/test_checksums/t/014_lockstep.pl
@@ -0,0 +1,182 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# The lockstep procedure for offline checksum changes in a replication
+# setup: stop all nodes, run pg_checksums on all of them, restart.
+# Replay of checkpoint records written before the change must not
+# revert the state of the standby.
+#
+# The same pair then runs the procedure in the disable direction, and
+# finally across a divergence: after a failover both nodes get their
+# checksums enabled offline, while the last common checkpoint still
+# carries "off".  pg_rewind must keep the offline "on" on the rewound
+# node, since no record in the replayed WAL could ever restore it.
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use DataChecksums::Utils;
+
+# wal_log_hints keeps the pair eligible for pg_rewind without checksums.
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 1, no_data_checksums => 1);
+$primary->append_conf(
+	'postgresql.conf', qq[
+autovacuum = off
+wal_keep_size = '1GB'
+wal_log_hints = on
+]);
+$primary->start;
+$primary->safe_psql('postgres',
+	"CREATE TABLE t AS SELECT generate_series(1,10000) AS a;");
+
+$primary->backup('backup');
+my $standby = PostgreSQL::Test::Cluster->new('standby');
+$standby->init_from_backup($primary, 'backup', has_streaming => 1);
+$standby->start;
+$primary->wait_for_catchup($standby);
+
+# Part 1: the lockstep procedure in the enable direction.  Stop the
+# standby first: the WAL written after this point is replayed only
+# after the offline switch, and every checkpoint record in it still
+# carries the old state.
+$standby->stop;
+$primary->safe_psql('postgres', "UPDATE t SET a = a WHERE a % 10 = 0;");
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->safe_psql('postgres', "UPDATE t SET a = a WHERE a % 10 = 1;");
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->stop;
+
+# The lockstep procedure.
+$primary->checksum_enable_offline;
+$standby->checksum_enable_offline;
+$primary->start;
+$standby->start;
+
+# The standby replays the pre-switch checkpoints; its state must not
+# revert to off.
+$primary->wait_for_catchup($standby);
+$standby->wait_for_log(qr/does not match the state "off" in the replayed WAL/,
+	0);
+test_checksum_state($standby, 'on');
+test_checksum_state($primary, 'on');
+
+is($standby->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10000', 'standby readable after lockstep enable');
+
+# Crash the standby and replay the same stretch again.
+$standby->stop('immediate');
+$standby->start;
+$primary->wait_for_catchup($standby);
+test_checksum_state($standby, 'on');
+is($standby->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10000', 'standby readable after crash restart');
+
+# Once a post-switch checkpoint has been replayed the states match and
+# no warning may be logged.
+my $logstart = -s $standby->logfile;
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->wait_for_catchup($standby);
+my $log = PostgreSQL::Test::Utils::slurp_file($standby->logfile, $logstart);
+unlike(
+	$log,
+	qr/does not match the state/,
+	'no mismatch warning once the states match');
+
+# Every page the standby wrote in this window must carry a checksum.
+$standby->safe_psql('postgres', 'CHECKPOINT;');
+$standby->stop;
+command_ok([ 'pg_checksums', '--check', '-D', $standby->data_dir ],
+	'checksums valid on the standby');
+
+# Part 2: the lockstep procedure in the other direction.  The standby
+# is already stopped; give it a pre-switch WAL stretch to replay, with
+# every checkpoint record in it still carrying "on" -- an explicit
+# checkpoint plus the shutdown checkpoint written by stop().
+$primary->safe_psql('postgres', "UPDATE t SET a = a WHERE a % 10 = 2;");
+$primary->safe_psql('postgres', "CHECKPOINT;");
+$primary->stop;
+
+$primary->checksum_disable_offline;
+$standby->checksum_disable_offline;
+
+my $disable_logstart = -s $standby->logfile;
+$primary->start;
+$standby->start;
+
+# The standby replays the pre-switch checkpoints; its state must not
+# revert to on.
+$primary->wait_for_catchup($standby);
+$standby->wait_for_log(qr/does not match the state "on" in the replayed WAL/,
+	$disable_logstart);
+test_checksum_state($standby, 'off');
+test_checksum_state($primary, 'off');
+
+is($standby->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10000', 'standby readable after lockstep disable');
+
+# Part 3: failover, divergence and pg_rewind across offline enables.
+# The roles swap from here on: the standby becomes the new primary and
+# the old primary is rewound to follow it, but the variables keep the
+# names they had above.
+#
+# Failover: promote the standby, then diverge the old primary.
+$standby->promote;
+$standby->safe_psql('postgres', "INSERT INTO t VALUES (0);");
+$primary->safe_psql('postgres', "INSERT INTO t VALUES (-1);");
+$primary->stop;
+
+# The lockstep procedure across the divergence: both nodes get their
+# checksums enabled offline.
+$primary->checksum_enable_offline;
+$standby->stop;
+$standby->checksum_enable_offline;
+$standby->start;
+test_checksum_state($standby, 'on');
+
+# The states match, so the rewind proceeds without complaint.
+command_ok(
+	[
+		'pg_rewind',
+		'--target-pgdata' => $primary->data_dir,
+		'--source-server' => $standby->connstr('postgres'),
+	],
+	'pg_rewind with checksums enabled offline on both nodes');
+
+# The rewind replaced the configuration files with those of the source,
+# so put the port back.  The copied file also carries the source's own
+# stale primary_conninfo; enable_streaming() below appends ours, which
+# wins as the later entry.
+$primary->append_conf(
+	'postgresql.conf', qq[
+port = @{[$primary->port]}
+]);
+$primary->enable_streaming($standby);
+
+my $rewind_logstart = -s $primary->logfile;
+$primary->start;
+$standby->wait_for_catchup($primary);
+
+is($primary->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10001', 'rewound server readable as a standby');
+
+# The offline enable survives: replay from the common checkpoint must
+# not resurrect the pre-divergence "off".
+test_checksum_state($primary, 'on');
+
+$log =
+  PostgreSQL::Test::Utils::slurp_file($primary->logfile, $rewind_logstart);
+unlike(
+	$log,
+	qr/does not match the state/,
+	'no divergence reported between the rewound server and its source');
+
+$primary->stop;
+$standby->stop;
+
+done_testing();
diff --git a/src/test/modules/test_checksums/t/015_standby_crash_after_disable.pl b/src/test/modules/test_checksums/t/015_standby_crash_after_disable.pl
new file mode 100644
index 00000000000..315b27b93d4
--- /dev/null
+++ b/src/test/modules/test_checksums/t/015_standby_crash_after_disable.pl
@@ -0,0 +1,125 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# Test that a standby crashing after a replayed online disable, but before
+# its next restartpoint, restarts cleanly.
+#
+# Pages dirtied before the XLOG2_CHECKSUMS record and evicted after it are
+# written out under the new "off" state, without checksums.  The control
+# file must follow the record immediately in this direction: a crash-restart
+# initializes checksum verification from the control file, and one still
+# saying "on" would fail verification on exactly those pages while replaying
+# records older than the state change.
+
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use DataChecksums::Utils;
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 1);
+$primary->append_conf(
+	'postgresql.conf', qq(
+autovacuum = off
+checkpoint_timeout = 1h
+max_wal_size = 10GB
+));
+$primary->start;
+
+test_checksum_state($primary, 'on');
+
+$primary->safe_psql('postgres',
+	"CREATE TABLE t AS SELECT generate_series(1,1000) AS a;");
+
+$primary->backup('backup');
+my $standby = PostgreSQL::Test::Cluster->new('standby');
+$standby->init_from_backup($primary, 'backup', has_streaming => 1);
+
+# Small buffer pool so that replaying a bulk load evicts the pages we care
+# about, and no restartpoints so the control file stays where it is.
+$standby->append_conf(
+	'postgresql.conf', qq(
+shared_buffers = 1MB
+checkpoint_timeout = 1h
+max_wal_size = 10GB
+bgwriter_delay = 10000
+));
+$standby->start;
+$primary->wait_for_catchup($standby);
+
+test_checksum_state($standby, 'on');
+
+# No full page images, so that replay has to read the pages of "t" back from
+# disk instead of overwriting them from the WAL.
+$primary->append_conf('postgresql.conf', 'full_page_writes = off');
+$primary->reload;
+$primary->safe_psql('postgres', 'CHECKPOINT;');
+
+# Establish the restartpoint that recovery will resume from after the crash.
+$standby->safe_psql('postgres', 'CHECKPOINT;');
+$primary->wait_for_catchup($standby);
+
+# Dirty the pages of "t" on the standby, *before* the state change.  They are
+# not flushed: the standby has no restartpoint from here on.
+$primary->safe_psql('postgres', 'UPDATE t SET a = a + 1;');
+$primary->wait_for_catchup($standby);
+
+# Online disable.  The standby replays it and switches to "off", but its
+# control file is not updated.
+$primary->safe_psql('postgres', 'SELECT pg_disable_data_checksums();');
+$primary->wait_for_catchup($standby);
+wait_for_checksum_state($standby, 'off');
+
+my ($ctl_before) = run_command([ 'pg_controldata', $standby->data_dir ]);
+my ($ctl_state) = $ctl_before =~ /Data page checksum version:\s+(\d+)/;
+note(
+	"standby control file data_checksum_version after the disable: "
+	  . "$ctl_state (0 = off, 1 = on)");
+
+# Force the standby to evict the dirty pages of "t" now that it is "off":
+# they are written back without checksums.
+$primary->safe_psql('postgres',
+	"CREATE TABLE filler AS SELECT generate_series(1,300000) AS a;");
+$primary->wait_for_catchup($standby);
+
+# Crash the standby before it gets a chance to run a restartpoint.
+$standby->stop('immediate');
+
+my $started = $standby->start(fail_ok => 1);
+ok($started,
+		'standby restarts after crashing between a checksum state '
+	  . 'change and the next restartpoint');
+
+my $log = PostgreSQL::Test::Utils::slurp_file($standby->logfile);
+unlike(
+	$log,
+	qr/page verification failed/,
+	'no checksum verification failures while replaying');
+unlike($log, qr/invalid page in block/, 'no invalid pages while replaying');
+
+if ($started)
+{
+	my ($rc, $stdout, $stderr) =
+	  $standby->psql('postgres', 'SELECT count(*) FROM t;');
+	is($rc, 0, 'the pages written while "off" are readable')
+	  or diag("stderr: $stderr");
+	$standby->stop('immediate');
+}
+else
+{
+	my @lines = grep { /FATAL|PANIC|invalid page|verification failed/ }
+	  split(/\n/, $log);
+	diag("standby log tail:\n" . join("\n", @lines[ -12 .. -1 ]))
+	  if @lines >= 12;
+	fail('the pages written while "off" are readable');
+}
+
+$primary->stop;
+
+done_testing();
diff --git a/src/test/modules/test_checksums/t/016_promote_enable_crash.pl b/src/test/modules/test_checksums/t/016_promote_enable_crash.pl
new file mode 100644
index 00000000000..97f84da5685
--- /dev/null
+++ b/src/test/modules/test_checksums/t/016_promote_enable_crash.pl
@@ -0,0 +1,134 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# Test a standby promoted right after replaying an online enable, before any
+# restartpoint has flushed the rewritten pages.
+#
+# The end-of-recovery record persists the checksum state without flushing the
+# buffer pool, while the control file still points at a restartpoint older than
+# the transition.  Persisting "on" there would make a crash before the
+# post-promotion checkpoint verify checksums over the pages that were flushed
+# while the state was still "off"; promotion must take the full checkpoint
+# instead.
+
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use DataChecksums::Utils;
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 1, no_data_checksums => 1);
+$primary->append_conf(
+	'postgresql.conf', qq(
+autovacuum = off
+checkpoint_timeout = 1h
+max_wal_size = 10GB
+full_page_writes = off
+));
+$primary->start;
+
+$primary->safe_psql('postgres', 'CREATE EXTENSION pg_buffercache;');
+$primary->safe_psql('postgres',
+	'CREATE TABLE t AS SELECT generate_series(1,10000) AS a;');
+
+$primary->backup('backup');
+my $standby = PostgreSQL::Test::Cluster->new('standby');
+$standby->init_from_backup($primary, 'backup', has_streaming => 1);
+$standby->append_conf(
+	'postgresql.conf', qq(
+shared_buffers = 512MB
+checkpoint_timeout = 1h
+max_wal_size = 10GB
+bgwriter_lru_maxpages = 0
+));
+$standby->start;
+$primary->wait_for_catchup($standby);
+
+test_checksum_state($primary, 'off');
+test_checksum_state($standby, 'off');
+
+# Establish the restartpoint that recovery will resume from after the crash.
+$primary->safe_psql('postgres', 'CHECKPOINT;');
+$primary->wait_for_catchup($standby);
+$standby->safe_psql('postgres', 'CHECKPOINT;');
+
+# Dirty the pages of "t" without full page images, then push them out to disk
+# on the standby while checksums are still off.
+$primary->safe_psql('postgres', 'UPDATE t SET a = a + 1;');
+$primary->wait_for_catchup($standby);
+
+my $relpath = $standby->safe_psql('postgres',
+	"SELECT pg_relation_filepath('t'::regclass);");
+my $evicted = $standby->safe_psql('postgres',
+	"SELECT pg_buffercache_evict_relation('t'::regclass);");
+note("evict_relation on the standby: $evicted, relpath $relpath");
+
+# Online enable on the primary; the standby replays it into its own buffers,
+# where the rewritten pages stay dirty (no restartpoint, no bgwriter).
+enable_data_checksums($primary, wait => 'on');
+$primary->wait_for_catchup($standby);
+wait_for_checksum_state($standby, 'on');
+
+my ($ctl) = run_command([ 'pg_controldata', $standby->data_dir ]);
+my ($before) = $ctl =~ /Data page checksum version:\s+(\d+)/;
+note("standby control file before the promotion: $before");
+
+# Promote.  The end-of-recovery record persists the live state without any
+# buffer flush; the checkpoint requested afterwards is not immediate.
+$standby->promote;
+$standby->poll_query_until('postgres', 'SELECT NOT pg_is_in_recovery();')
+  or die 'timed out waiting for the promotion';
+$standby->stop('immediate');
+
+($ctl) = run_command([ 'pg_controldata', $standby->data_dir ]);
+my ($after) = $ctl =~ /Data page checksum version:\s+(\d+)/;
+my ($ckpt) = $ctl =~ /Latest checkpoint location:\s+(\S+)/;
+my ($redo) = $ctl =~ /Latest checkpoint's REDO location:\s+(\S+)/;
+note(
+	"standby control file after the promotion: $after, checkpoint $ckpt, redo $redo"
+);
+
+my $page;
+open(my $fh, '<', $standby->data_dir . '/' . $relpath) or die $!;
+binmode $fh;
+read($fh, $page, 8192);
+close($fh);
+my ($pd_checksum) = unpack('x8 v', $page);
+note("on-disk pd_checksum of t block 0: $pd_checksum");
+
+my $started = $standby->start(fail_ok => 1);
+ok($started,
+	'promoted node restarts after crashing before its first checkpoint');
+
+my $log = PostgreSQL::Test::Utils::slurp_file($standby->logfile);
+unlike(
+	$log,
+	qr/page verification failed/,
+	'no checksum verification failures while replaying');
+unlike($log, qr/invalid page in block/, 'no invalid pages while replaying');
+
+if ($started)
+{
+	my ($rc, $stdout, $stderr) =
+	  $standby->psql('postgres', 'SELECT count(*) FROM t;');
+	is($rc, 0, 'table readable after the crash restart')
+	  or diag("stderr: $stderr");
+	$standby->stop('immediate');
+}
+else
+{
+	my @lines = grep { /FATAL|PANIC|invalid page|verification failed/ }
+	  split(/\n/, $log);
+	diag("log tail:\n" . join("\n", @lines));
+	fail('table readable after the crash restart');
+}
+
+$primary->stop;
+
+done_testing();
diff --git a/src/test/modules/test_checksums/t/017_restartpoint_race.pl b/src/test/modules/test_checksums/t/017_restartpoint_race.pl
new file mode 100644
index 00000000000..65aa834f366
--- /dev/null
+++ b/src/test/modules/test_checksums/t/017_restartpoint_race.pl
@@ -0,0 +1,138 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# Test a restartpoint whose flush races the replay of an online enable.
+#
+# Replay keeps running while CheckPointGuts() writes out the buffer pool, so
+# the state at the end of the flush can be newer than the one the flush ran
+# under.  Persisting it would claim checksums for pages the flush wrote out
+# before the transition, against a redo pointer that predates it.
+
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use DataChecksums::Utils;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+	plan skip_all => 'Injection points not supported by this build';
+}
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 1, no_data_checksums => 1);
+$primary->append_conf(
+	'postgresql.conf', qq(
+autovacuum = off
+checkpoint_timeout = 1h
+max_wal_size = 10GB
+full_page_writes = off
+));
+$primary->start;
+
+$primary->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
+$primary->safe_psql('postgres', 'CREATE EXTENSION pg_buffercache;');
+$primary->safe_psql('postgres',
+	'CREATE TABLE t AS SELECT generate_series(1,10000) AS a;');
+
+$primary->backup('backup');
+my $standby = PostgreSQL::Test::Cluster->new('standby');
+$standby->init_from_backup($primary, 'backup', has_streaming => 1);
+$standby->append_conf(
+	'postgresql.conf', qq(
+shared_buffers = 512MB
+checkpoint_timeout = 1h
+max_wal_size = 10GB
+bgwriter_lru_maxpages = 0
+));
+$standby->start;
+$primary->wait_for_catchup($standby);
+
+# C1: the checkpoint the pending restartpoint will target.
+$primary->safe_psql('postgres', 'CHECKPOINT;');
+$primary->wait_for_catchup($standby);
+
+# Dirty the pages of "t" without full page images and push them to disk on
+# the standby while it is still "off".
+$primary->safe_psql('postgres', 'UPDATE t SET a = a + 1;');
+$primary->wait_for_catchup($standby);
+
+my $relpath = $standby->safe_psql('postgres',
+	"SELECT pg_relation_filepath('t'::regclass);");
+my $evicted = $standby->safe_psql('postgres',
+	"SELECT pg_buffercache_evict_relation('t'::regclass);");
+note("evict_relation on the standby: $evicted, relpath $relpath");
+
+# Start a restartpoint for C1 and hold it right after CheckPointGuts().
+$standby->safe_psql('postgres',
+	"SELECT injection_points_attach('create-restart-point','wait');");
+my $bg = $standby->background_psql('postgres');
+$bg->query_until(qr//, "\\echo restartpoint\nCHECKPOINT;\n");
+$standby->poll_query_until('postgres',
+	"SELECT count(*) > 0 FROM pg_stat_activity WHERE wait_event = 'create-restart-point';"
+) or die 'timed out waiting for the restartpoint injection point';
+
+# The startup process keeps replaying while the checkpointer is held: the
+# whole online enable lands, and the rewritten pages stay dirty.
+enable_data_checksums($primary, wait => 'on');
+$primary->wait_for_catchup($standby);
+wait_for_checksum_state($standby, 'on');
+
+# Let the restartpoint finish; it persists the state it samples now.
+$standby->safe_psql('postgres',
+	"SELECT injection_points_wakeup('create-restart-point');");
+$standby->poll_query_until('postgres',
+	"SELECT count(*) = 0 FROM pg_stat_activity WHERE wait_event = 'create-restart-point';"
+) or die 'timed out waiting for the restartpoint to finish';
+$standby->safe_psql('postgres',
+	"SELECT injection_points_detach('create-restart-point');");
+
+$standby->stop('immediate');
+
+my ($ctl) = run_command([ 'pg_controldata', $standby->data_dir ]);
+my ($after) = $ctl =~ /Data page checksum version:\s+(\d+)/;
+my ($redo) = $ctl =~ /Latest checkpoint's REDO location:\s+(\S+)/;
+note("standby control file after the restartpoint: $after, redo $redo");
+
+my $page;
+open(my $fh, '<', $standby->data_dir . '/' . $relpath) or die $!;
+binmode $fh;
+read($fh, $page, 8192);
+close($fh);
+my ($pd_checksum) = unpack('x8 v', $page);
+note("on-disk pd_checksum of t block 0: $pd_checksum");
+
+my $started = $standby->start(fail_ok => 1);
+ok($started, 'standby restarts after a restartpoint that raced the enable');
+
+my $log = PostgreSQL::Test::Utils::slurp_file($standby->logfile);
+unlike(
+	$log,
+	qr/page verification failed/,
+	'no checksum verification failures while replaying');
+unlike($log, qr/invalid page in block/, 'no invalid pages while replaying');
+
+if ($started)
+{
+	my ($rc, $stdout, $stderr) =
+	  $standby->psql('postgres', 'SELECT count(*) FROM t;');
+	is($rc, 0, 'table readable after the crash restart')
+	  or diag("stderr: $stderr");
+	$standby->stop('immediate');
+}
+else
+{
+	my @lines = grep { /FATAL|PANIC|invalid page|verification failed/ }
+	  split(/\n/, $log);
+	diag("log tail:\n" . join("\n", @lines));
+	fail('table readable after the crash restart');
+}
+
+$primary->stop;
+
+done_testing();
diff --git a/src/test/modules/test_checksums/t/018_enable_crash_windows.pl b/src/test/modules/test_checksums/t/018_enable_crash_windows.pl
new file mode 100644
index 00000000000..17f96ff6272
--- /dev/null
+++ b/src/test/modules/test_checksums/t/018_enable_crash_windows.pl
@@ -0,0 +1,496 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# Crash and checkpoint windows inside an online data checksum enable on
+# a single node.  Each scenario starts from "off" on the same cluster,
+# runs the enable into a held injection point, and checks what a crash
+# or a concurrent checkpoint in that window may and may not do to the
+# control file and the pages on disk.
+
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use DataChecksums::Utils;
+
+use IPC::Run;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+	plan skip_all => 'Injection points not supported by this build';
+}
+
+# Bring the node back to "off" with no leftover table between
+# scenarios.  The crash restarts already resolve an interrupted enable
+# to "off", so only disable when something is left to disable.  Those
+# restarts also clear any attached injection points.  The checkpoint
+# bounds the WAL the next scenario's crash recovery has to replay.
+sub reset_scenario
+{
+	my ($node) = @_;
+
+	BAIL_OUT('node is not running after the previous scenario')
+	  unless $node->is_alive;
+
+	my $state = $node->safe_psql('postgres', 'SHOW data_checksums;');
+	disable_data_checksums($node, wait => 'off') if $state ne 'off';
+	$node->safe_psql('postgres', 'DROP TABLE IF EXISTS t;');
+	$node->safe_psql('postgres', 'CHECKPOINT;');
+	test_checksum_state($node, 'off');
+}
+
+my $node = PostgreSQL::Test::Cluster->new('crash_windows');
+$node->init(no_data_checksums => 1);
+
+# Scenario 1 needs the pages of "t" to reach disk without a full page
+# image, hence full_page_writes and wal_log_hints off.  Scenario 2
+# needs them to stay dirty in shared buffers until a checkpoint, hence
+# no bgwriter and a large shared_buffers.  wal_level is pinned because
+# init() writes "minimal" for a non-streaming node.  The rest merely
+# tolerate the settings.
+$node->append_conf(
+	'postgresql.conf', qq(
+autovacuum = off
+checkpoint_timeout = 1h
+wal_level = replica
+max_wal_size = 10GB
+full_page_writes = off
+wal_log_hints = off
+shared_buffers = 512MB
+bgwriter_lru_maxpages = 0
+));
+$node->start;
+
+$node->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
+$node->safe_psql('postgres', 'CREATE EXTENSION pg_buffercache;');
+
+test_checksum_state($node, 'off');
+
+# Scenario 1: a primary crashes inside SetDataChecksumsOn(), just before the
+# forced checkpoint that flushes the rewritten pages, with crash recovery
+# resuming from a checkpoint older than the transition.  The control file
+# must still say "inprogress-on" there: replay does not adopt the state of
+# the checkpoint record it resumes from, so an "on" written before the flush
+# would stay in effect while replay reads pages whose rewrite never reached
+# disk.  Here the pages went out through the rewriting worker's ring buffer,
+# so only the control file state discriminates; see the next scenario for
+# the case where they did not.
+{
+	$node->safe_psql('postgres',
+		'CREATE TABLE t AS SELECT generate_series(1,10000) AS a;');
+
+	# Establish the checkpoint that crash recovery will resume from.
+	$node->safe_psql('postgres', 'CHECKPOINT;');
+
+	# Dirty the pages of "t" without full page images, then push them out
+	# to disk while checksums are still off.
+	$node->safe_psql('postgres', 'UPDATE t SET a = a + 1;');
+	my $relpath = $node->safe_psql('postgres',
+		"SELECT pg_relation_filepath('t'::regclass);");
+	my $evicted = $node->safe_psql('postgres',
+		"SELECT pg_buffercache_evict_relation('t'::regclass);");
+	note("evict_relation on t: $evicted, relpath $relpath");
+
+	# Stop the enable right after the control file has been updated to
+	# "on" but before the checkpoint that flushes the rewritten pages.
+	$node->safe_psql('postgres',
+		"SELECT injection_points_attach('datachecksums-on-before-checkpoint','wait');"
+	);
+	$node->safe_psql('postgres', 'SELECT pg_enable_data_checksums();');
+
+	$node->poll_query_until('postgres',
+		"SELECT count(*) > 0 FROM pg_stat_activity WHERE wait_event = 'datachecksums-on-before-checkpoint';"
+	) or die 'timed out waiting for the injection point';
+
+	my ($ctl) = run_command([ 'pg_controldata', $node->data_dir ]);
+	my ($ctl_state) = $ctl =~ /Data page checksum version:\s+(\d+)/;
+	note("control file data_checksum_version before the crash: $ctl_state");
+	is($ctl_state, '3',
+		'control file still says "inprogress-on" before the checkpoint');
+
+	$node->stop('immediate');
+
+	# Show the on-disk checksum field of the first page of "t".
+	my $page;
+	open(my $fh, '<', $node->data_dir . '/' . $relpath) or die $!;
+	binmode $fh;
+	read($fh, $page, 8192);
+	close($fh);
+	my ($pd_checksum) = unpack('x8 v', $page);
+	note("on-disk pd_checksum of t block 0: $pd_checksum");
+
+	my $started = $node->start(fail_ok => 1);
+	ok($started, 'primary restarts after crashing inside the online enable');
+
+	my $log = PostgreSQL::Test::Utils::slurp_file($node->logfile);
+	unlike(
+		$log,
+		qr/page verification failed/,
+		'no checksum verification failures while replaying');
+	unlike($log, qr/invalid page in block/,
+		'no invalid pages while replaying');
+
+	if ($started)
+	{
+		my ($rc, $stdout, $stderr) =
+		  $node->psql('postgres', 'SELECT count(*) FROM t;');
+		is($rc, 0, 'table readable after the crash restart')
+		  or diag("stderr: $stderr");
+	}
+	else
+	{
+		my @lines = grep { /FATAL|PANIC|invalid page|verification failed/ }
+		  split(/\n/, $log);
+		diag("log tail:\n" . join("\n", @lines));
+		fail('table readable after the crash restart');
+	}
+}
+reset_scenario($node);
+
+# Scenario 2: the same crash as above, but with the rewritten pages resident
+# in shared buffers instead of gone out through the ring.  The rewriting
+# worker reads through a BAS_VACUUM ring, which writes pages back as the
+# ring recycles, but a page already resident in shared buffers is not read
+# through the ring: ReadBufferExtended() hands back the existing buffer, and
+# it stays dirty until a checkpoint.  The control file may therefore not
+# say "on" before that checkpoint has run, or crash recovery would resume
+# from a checkpoint older than the transition with verification already
+# enabled, and replay records that read those still-unchecksummed pages.
+# full_page_writes is off so that the records do not simply overwrite the
+# pages with a full page image.
+{
+	$node->safe_psql('postgres',
+		'CREATE TABLE t AS SELECT generate_series(1,100000) AS a;');
+	my $relpath = $node->safe_psql('postgres',
+		"SELECT pg_relation_filepath('t'::regclass);");
+
+	# Establish the checkpoint that crash recovery will resume from, with the
+	# pages of "t" written out while checksums are still off.
+	$node->safe_psql('postgres', 'CHECKPOINT;');
+
+	# Dirty those pages again without emitting full page images, and leave
+	# them in shared buffers.  Replay of these records has to read the
+	# pages from disk.
+	$node->safe_psql('postgres', 'UPDATE t SET a = a + 1;');
+
+	my $dirty_before = $node->safe_psql('postgres',
+			"SELECT count(*) FROM pg_buffercache "
+		  . "WHERE relfilenode = pg_relation_filenode('t'::regclass) "
+		  . "AND relforknumber = 0 AND isdirty;");
+	cmp_ok($dirty_before, '>', 0,
+		'pages of t are resident and dirty before enabling checksums');
+
+	# Hold the enabling right before the checkpoint that flushes the rewritten
+	# pages.
+	$node->safe_psql('postgres',
+		"SELECT injection_points_attach('datachecksums-on-before-checkpoint','wait');"
+	);
+
+	enable_data_checksums($node);
+	$node->wait_for_event('datachecksums launcher',
+		'datachecksums-on-before-checkpoint');
+
+	my ($ctl) = run_command([ 'pg_controldata', $node->data_dir ]);
+	my ($ctl_state) = $ctl =~ /Data page checksum version:\s+(\d+)/;
+	note("control file data_checksum_version before the crash: $ctl_state");
+	is($ctl_state, '3',
+		'control file still says "inprogress-on" before the checkpoint');
+
+	# The rewritten pages must still be sitting dirty in shared buffers, or
+	# the window this test is about does not exist.
+	my $dirty_after = $node->safe_psql('postgres',
+			"SELECT count(*) FROM pg_buffercache "
+		  . "WHERE relfilenode = pg_relation_filenode('t'::regclass) "
+		  . "AND relforknumber = 0 AND isdirty;");
+	cmp_ok($dirty_after, '>', 0,
+		'rewritten pages of t are still dirty before the checkpoint');
+
+	$node->stop('immediate');
+
+	# The on-disk copy is the one written before the transition, without a
+	# checksum.
+	my $page;
+	open(my $fh, '<', $node->data_dir . '/' . $relpath) or die $!;
+	binmode $fh;
+	read($fh, $page, 8192);
+	close($fh);
+	my ($pd_checksum) = unpack('x8 v', $page);
+	note("on-disk pd_checksum of t block 0: $pd_checksum");
+	is($pd_checksum, 0, 'block 0 of t on disk carries no checksum');
+
+	my $started = $node->start(fail_ok => 1);
+	ok($started, 'primary restarts after crashing inside the online enable');
+
+	my $log = PostgreSQL::Test::Utils::slurp_file($node->logfile);
+	unlike(
+		$log,
+		qr/page verification failed/,
+		'no checksum verification failures while replaying');
+	unlike($log, qr/invalid page in block/,
+		'no invalid pages while replaying');
+
+	if ($started)
+	{
+		my ($rc, $stdout, $stderr) =
+		  $node->psql('postgres', 'SELECT count(*) FROM t;');
+		is($rc, 0, 'table readable after the crash restart')
+		  or diag("stderr: $stderr");
+	}
+	else
+	{
+		my @lines = grep { /FATAL|PANIC|invalid page|verification failed/ }
+		  split(/\n/, $log);
+		diag("log tail:\n" . join("\n", @lines));
+		fail('table readable after the crash restart');
+	}
+}
+reset_scenario($node);
+
+# Scenario 3: a checkpoint that runs between the XLOG2_CHECKSUMS("on")
+# record and the control file write at the end of SetDataChecksumsOn() must
+# not make a crash throw the completed transition away.  SetDataChecksumsOn()
+# writes the record, flips shared memory to "on", emits the barrier and only
+# then requests the checkpoint that flushes the rewritten pages; the control
+# file is written after that checkpoint returns.  A crash in that window is
+# harmless only as long as recovery still starts before the record.  Any
+# checkpoint completing in the window moves the redo point past the record,
+# so recovery would never see it, would come up with the control file's
+# "inprogress-on" and StartupXLOG() would demote that to "off", even though
+# every page on disk carries a checksum by then.  CreateCheckPoint()
+# therefore persists the state the checkpoint ran under, the same way
+# CreateRestartPoint() does.  Here the window is made deterministic by
+# holding the launcher at the datachecksums-on-before-checkpoint injection
+# point and checkpointing from another session.
+{
+	$node->safe_psql('postgres',
+		'CREATE TABLE t AS SELECT generate_series(1,10000) AS a;');
+	my $relpath = $node->safe_psql('postgres',
+		"SELECT pg_relation_filepath('t'::regclass);");
+
+	# Hold the launcher after the record, the shared memory flip and the
+	# barrier, but before the checkpoint SetDataChecksumsOn() requests
+	# itself.
+	$node->safe_psql('postgres',
+		"SELECT injection_points_attach('datachecksums-on-before-checkpoint','wait');"
+	);
+
+	enable_data_checksums($node);
+	$node->wait_for_event('datachecksums launcher',
+		'datachecksums-on-before-checkpoint');
+
+	# Every backend already sees "on" and writes checksums.
+	test_checksum_state($node, 'on');
+
+	# ... while the control file still says "inprogress-on".
+	my ($ctl) = run_command([ 'pg_controldata', $node->data_dir ]);
+	my ($before) = $ctl =~ /Data page checksum version:\s+(\d+)/;
+	is($before, '3', 'control file says "inprogress-on" inside the window');
+
+	# A concurrent checkpoint.  It flushes the rewritten pages and moves
+	# the redo point past the XLOG2_CHECKSUMS("on") record, so it has to
+	# record the "on" state in the control file as well.
+	$node->safe_psql('postgres', 'CHECKPOINT;');
+
+	($ctl) = run_command([ 'pg_controldata', $node->data_dir ]);
+	my ($after) = $ctl =~ /Data page checksum version:\s+(\d+)/;
+	is($after, '1',
+		'the concurrent checkpoint records "on" in the control file');
+
+	$node->stop('immediate');
+
+	# The checkpoint flushed the rewritten pages, so they carry a checksum on
+	# disk: the transition really did complete.
+	my $page;
+	open(my $fh, '<', $node->data_dir . '/' . $relpath) or die $!;
+	binmode $fh;
+	read($fh, $page, 8192);
+	close($fh);
+	my ($pd_checksum) = unpack('x8 v', $page);
+	note("on-disk pd_checksum of t block 0: $pd_checksum");
+	isnt($pd_checksum, 0, 'block 0 of t on disk carries a checksum');
+
+	my $log_offset = -s $node->logfile;
+	$node->start;
+
+	# The transition is complete on disk, so the cluster has to come back
+	# "on".
+	test_checksum_state($node, 'on');
+
+	my $log =
+	  PostgreSQL::Test::Utils::slurp_file($node->logfile, $log_offset);
+	unlike(
+		$log,
+		qr/enabling data checksums was interrupted/,
+		'the completed transition is not reported as interrupted');
+}
+reset_scenario($node);
+
+# Scenario 4: a crash between the checkpoint an online enable requests and
+# the control file write that follows it must not lose the transition.  The
+# last steps of SetDataChecksumsOn() are
+#
+#	WAL record -> shmem -> barrier -> checkpoint -> persist
+#
+# The checkpoint is what makes "on" safe to persist, but it also moves the
+# redo point above the XLOG2_CHECKSUMS record that carries the new state.
+# Crash recovery started from that checkpoint therefore never replays the
+# record, so without the state the checkpoint itself persists, a crash in
+# the remaining window would bring the cluster back at "inprogress-on",
+# which StartupXLOG() resolves to "off", discarding a transition whose
+# pages are all on disk with a checksum.  The previous scenario exercises
+# the same window through a checkpoint requested by another session; this
+# one closes it from the other side: the transition's own checkpoint is the
+# one that moves the redo point, so persisting the state from
+# CreateCheckPoint() is what has to save it, not the write below the
+# injection point.
+{
+	$node->safe_psql('postgres',
+		'CREATE TABLE t AS SELECT generate_series(1,10000) AS a;');
+
+	# Hold the launcher after the checkpoint that licenses "on" has
+	# completed and before the state reaches the control file.
+	$node->safe_psql('postgres',
+		"SELECT injection_points_attach('datachecksums-on-after-checkpoint','wait');"
+	);
+
+	enable_data_checksums($node);
+	$node->wait_for_event('datachecksums launcher',
+		'datachecksums-on-after-checkpoint');
+
+	# The transition is complete as far as the running cluster is concerned.
+	test_checksum_state($node, 'on');
+
+	# The checkpoint has already recorded it, so the pending write below the
+	# injection point has nothing left to do.
+	my ($ctl) = run_command([ 'pg_controldata', $node->data_dir ]);
+	my ($version) = $ctl =~ /Data page checksum version:\s+(\d+)/;
+	is($version, '1',
+		'the requested checkpoint recorded "on" in the control file');
+
+	# Crash before the control file write that follows the checkpoint.
+	$node->stop('immediate');
+
+	$node->start;
+
+	# Every page on disk carries a checksum and the checkpoint that
+	# flushed them completed, so the cluster has to come back verifying
+	# them.
+	test_checksum_state($node, 'on');
+
+	is($node->safe_psql('postgres', 'SELECT count(*) FROM t;'),
+		'10000', 'relation readable after the crash');
+
+	my $log = PostgreSQL::Test::Utils::slurp_file($node->logfile);
+	unlike(
+		$log,
+		qr/enabling data checksums was interrupted/,
+		'the completed transition is not reported as interrupted');
+
+	# The data directory must be one the offline tools accept.
+	$node->stop;
+	$node->command_ok([ 'pg_checksums', '--check', '-D', $node->data_dir ],
+		'pg_checksums accepts the data directory');
+
+	# Leave the node running for the reset.
+	$node->start;
+}
+reset_scenario($node);
+
+# Scenario 5: a checkpoint racing SetDataChecksumsOn() between the insertion
+# of the XLOG2_CHECKSUMS("on") record and the shared memory update must not
+# insert an XLOG_CHECKPOINT_REDO record that follows the transition in WAL
+# order while still carrying "inprogress-on".  Recovery resuming from such a
+# redo point never replays the preceding transition record, comes up in
+# "inprogress-on" and resolves the finished transition as interrupted.
+# XLogChecksums() closes the window by inserting the record and publishing
+# the new state under DataChecksumTransitionLock, which CreateCheckPoint()
+# takes around sampling the state and inserting the redo record.  Here the
+# launcher is held between the two steps at the
+# datachecksums-on-before-publish injection point, and a concurrent
+# CHECKPOINT has to block on the lock instead of completing inside the
+# window.
+{
+	$node->safe_psql('postgres',
+		'CREATE TABLE t AS SELECT generate_series(1,10000) AS a;');
+
+	# The datachecksums-on-before-publish point fires inside a critical
+	# section, where the wait machinery must not allocate.  Waiting once
+	# at the datachecksums-enable-checksums-delay point, which the
+	# launcher runs outside the critical section, initializes it; see
+	# 050_redo_segment_missing.pl for the same recipe around
+	# create-checkpoint-run.
+	$node->safe_psql('postgres',
+		"SELECT injection_points_attach('datachecksums-enable-checksums-delay','wait');"
+	);
+
+	# Hold the launcher after the XLOG2_CHECKSUMS("on") record is in WAL but
+	# before the new state is published in shared memory.
+	$node->safe_psql('postgres',
+		"SELECT injection_points_attach('datachecksums-on-before-publish','wait');"
+	);
+
+	enable_data_checksums($node);
+	$node->wait_for_event('datachecksums launcher',
+		'datachecksums-enable-checksums-delay');
+	$node->safe_psql('postgres',
+		"SELECT injection_points_wakeup('datachecksums-enable-checksums-delay');");
+	$node->safe_psql('postgres',
+		"SELECT injection_points_detach('datachecksums-enable-checksums-delay');");
+
+	$node->wait_for_event('datachecksums launcher',
+		'datachecksums-on-before-publish');
+
+	# The record is in WAL, the published state is still the old one.
+	test_checksum_state($node, 'inprogress-on');
+
+	# A concurrent checkpoint.  It must block on DataChecksumTransitionLock
+	# before inserting its redo record rather than complete inside the window.
+	my $checkpointer = IPC::Run::start(
+		[ 'psql', '-XAtq', '-d', $node->connstr('postgres'), '-c', 'CHECKPOINT;' ],
+		'<' => '/dev/null',
+		'>' => '/dev/null',
+		'2>' => '/dev/null',
+		IPC::Run::timer($PostgreSQL::Test::Utils::timeout_default));
+
+	ok( $node->poll_query_until(
+			'postgres',
+			"SELECT wait_event = 'DataChecksumTransition' "
+			  . "FROM pg_stat_activity WHERE backend_type = 'checkpointer';"),
+		'concurrent checkpoint blocks on DataChecksumTransitionLock');
+
+	# Release the launcher; the checkpoint then samples the published "on" and
+	# its redo record follows the transition record.
+	$node->safe_psql('postgres',
+		"SELECT injection_points_wakeup('datachecksums-on-before-publish');");
+	$node->safe_psql('postgres',
+		"SELECT injection_points_detach('datachecksums-on-before-publish');");
+
+	$checkpointer->finish;
+
+	# Crash while the launcher's own checkpoint may still be in flight.
+	# Recovery resumes from the concurrent checkpoint's redo point, which
+	# now lies above the transition record and carries "on".
+	$node->stop('immediate');
+
+	my $log_offset = -s $node->logfile;
+	$node->start;
+
+	# The transition completed, so the cluster has to come back "on".
+	wait_for_checksum_state($node, 'on');
+
+	my $log =
+	  PostgreSQL::Test::Utils::slurp_file($node->logfile, $log_offset);
+	unlike(
+		$log,
+		qr/enabling data checksums was interrupted/,
+		'the completed transition is not reported as interrupted');
+
+	$node->stop;
+}
+
+done_testing();
diff --git a/src/test/modules/test_checksums/t/019_standby_shutdown_catchup.pl b/src/test/modules/test_checksums/t/019_standby_shutdown_catchup.pl
new file mode 100644
index 00000000000..aaa35f8f7dd
--- /dev/null
+++ b/src/test/modules/test_checksums/t/019_standby_shutdown_catchup.pl
@@ -0,0 +1,126 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# A standby stopped after replaying an online enable, but before the
+# checkpoint record that follows it on the primary.
+#
+# The shutdown restartpoint has no new checkpoint record to work from
+# and is skipped, so the control file would keep the in-progress state
+# the transition passed through, even though replay left the node at
+# "on" and rewrote every page.  pg_checksums reads that field and
+# refuses to run on an in-progress state, so the shutdown has to flush
+# and catch it up instead.
+#
+# The caught-up control file then carries the watermark of the "on"
+# record, which the second half of this test relies on: an offline
+# disable made while the standby is down must survive the re-replay of
+# that record on the next startup, or the change would be silently
+# reverted.
+
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use DataChecksums::Utils;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+	plan skip_all => 'Injection points not supported by this build';
+}
+
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 1, no_data_checksums => 1);
+$primary->append_conf(
+	'postgresql.conf', qq(
+autovacuum = off
+checkpoint_timeout = 1h
+max_wal_size = 10GB
+wal_keep_size = 1GB
+));
+$primary->start;
+
+$primary->safe_psql('postgres', 'CREATE EXTENSION injection_points;');
+$primary->safe_psql('postgres',
+	'CREATE TABLE t AS SELECT generate_series(1,10000) AS a;');
+
+$primary->backup('backup');
+my $standby = PostgreSQL::Test::Cluster->new('standby');
+$standby->init_from_backup($primary, 'backup', has_streaming => 1);
+$standby->append_conf(
+	'postgresql.conf', qq(
+checkpoint_timeout = 1h
+max_wal_size = 10GB
+));
+$standby->start;
+$primary->wait_for_catchup($standby);
+
+# Anchor the standby's last restartpoint here, so that nothing replayed
+# from now on gives the shutdown restartpoint a newer checkpoint record
+# to use.
+$primary->safe_psql('postgres', 'CHECKPOINT;');
+$primary->wait_for_catchup($standby);
+$standby->safe_psql('postgres', 'CHECKPOINT;');
+
+# Hold the enable right after the state change record, before the
+# checkpoint it requests once the transition is complete.
+$primary->safe_psql('postgres',
+	"SELECT injection_points_attach('datachecksums-on-before-checkpoint','wait');"
+);
+enable_data_checksums($primary);
+$primary->wait_for_event('datachecksums launcher',
+	'datachecksums-on-before-checkpoint');
+wait_for_checksum_state($primary, 'on');
+
+# The state change record is already flushed and streams on its own;
+# no checkpoint record follows while the launcher is held.
+$primary->wait_for_catchup($standby);
+wait_for_checksum_state($standby, 'on');
+
+$standby->stop('fast');
+
+my ($ctl) = run_command([ 'pg_controldata', $standby->data_dir ]);
+my ($state) = $ctl =~ /Data page checksum version:\s+(\d+)/;
+is($state, '1',
+	'shutdown catches the control file up with the replayed state');
+
+command_ok([ 'pg_checksums', '--check', '-D', $standby->data_dir ],
+	'pg_checksums verifies the stopped standby');
+
+# Disable checksums offline while the standby is down.  The restart
+# below resumes replay under the last restartpoint, re-reading the "on"
+# record; the watermark persisted with the catch-up above marks it as
+# already applied, so the offline change must survive.
+$standby->checksum_disable_offline;
+
+# Let the enable finish on the primary.
+$primary->safe_psql('postgres',
+	"SELECT injection_points_wakeup('datachecksums-on-before-checkpoint');");
+$primary->safe_psql('postgres',
+	"SELECT injection_points_detach('datachecksums-on-before-checkpoint');");
+
+# The launcher exits only after the checkpoint it requests completes,
+# so a checkpoint record now follows the transition in WAL.
+$primary->poll_query_until('postgres',
+		"SELECT count(*) = 0 FROM pg_catalog.pg_stat_activity "
+	  . "WHERE backend_type = 'datachecksums launcher';");
+
+my $log_offset = -s $standby->logfile;
+$standby->start;
+$primary->wait_for_catchup($standby);
+
+test_checksum_state($standby, 'off');
+
+# The nodes legitimately diverged, which replayed checkpoints report.
+$standby->wait_for_log(
+	qr/data checksum state "off" of this node does not match the state "on" in the replayed WAL/,
+	$log_offset);
+
+$standby->stop;
+$primary->stop;
+
+done_testing();
diff --git a/src/test/modules/test_checksums/t/020_cascade_divergence.pl b/src/test/modules/test_checksums/t/020_cascade_divergence.pl
new file mode 100644
index 00000000000..b4b474fa583
--- /dev/null
+++ b/src/test/modules/test_checksums/t/020_cascade_divergence.pl
@@ -0,0 +1,115 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# An offline data checksum state change made on one node of a cascading setup
+# is reported all the way down the chain.
+#
+# CheckReplayedDataChecksumState() is reached from xlog_redo() when a
+# checkpoint record (XLOG_CHECKPOINT_SHUTDOWN or XLOG_CHECKPOINT_REDO) is
+# replayed after consistency has been reached.  Those records are written by
+# the root primary only and are relayed verbatim by every intermediate
+# standby, so a cascaded standby that was changed offline learns about the
+# divergence too.  An intermediate standby's own restartpoints are not
+# WAL-logged and cannot surface it.
+#
+# Note that this makes detection checkpoint-driven: an idle or read-only
+# primary surfaces the divergence no sooner than its next checkpoint, so up to
+# checkpoint_timeout may pass before the operator sees anything.  Closing that
+# window would require comparing the state when a walreceiver connects.
+
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use DataChecksums::Utils;
+
+# checkpoint_timeout is deliberately long so that the only checkpoint record in
+# this test is the one requested explicitly below.
+my $conf = qq(
+autovacuum = off
+checkpoint_timeout = 1h
+max_wal_size = 10GB
+);
+
+my $primary = PostgreSQL::Test::Cluster->new('cascade_primary');
+$primary->init(allows_streaming => 1, no_data_checksums => 1);
+$primary->append_conf('postgresql.conf', $conf);
+$primary->start;
+$primary->safe_psql('postgres',
+	'CREATE TABLE t AS SELECT generate_series(1,1000) AS a;');
+
+$primary->backup('backup');
+my $standby = PostgreSQL::Test::Cluster->new('cascade_standby');
+$standby->init_from_backup($primary, 'backup', has_streaming => 1);
+$standby->append_conf('postgresql.conf', $conf);
+$standby->start;
+
+$standby->backup('backup');
+my $cascade = PostgreSQL::Test::Cluster->new('cascade_cascade');
+$cascade->init_from_backup($standby, 'backup', has_streaming => 1);
+$cascade->append_conf('postgresql.conf', $conf);
+$cascade->start;
+
+$primary->wait_for_catchup($standby);
+$standby->wait_for_catchup($cascade);
+
+test_checksum_state($primary, 'off');
+test_checksum_state($standby, 'off');
+test_checksum_state($cascade, 'off');
+
+# Change the cascaded standby offline, and only it.
+$cascade->stop;
+$cascade->command_ok([ 'pg_checksums', '--enable', '-D', $cascade->data_dir ],
+	'pg_checksums enables checksums on the cascaded standby only');
+
+my $logstart = -s $cascade->logfile;
+$cascade->start;
+
+test_checksum_state($cascade, 'on');
+
+# Ordinary WAL traffic carries no state, so the cascaded standby keeps
+# streaming from a chain whose state is "off" without noticing.
+$primary->safe_psql('postgres', 'INSERT INTO t VALUES (1);');
+$primary->wait_for_catchup($standby);
+$standby->wait_for_catchup($cascade);
+
+is($cascade->safe_psql('postgres', 'SELECT count(*) FROM t;'),
+	'1001', 'cascaded standby keeps streaming from a divergent chain');
+
+# Restartpoints on the intermediate standby are not WAL-logged, so this does
+# not surface anything on the cascaded standby either.
+$standby->safe_psql('postgres', 'CHECKPOINT;');
+$primary->safe_psql('postgres', 'INSERT INTO t VALUES (2);');
+$primary->wait_for_catchup($standby);
+$standby->wait_for_catchup($cascade);
+
+my $log = PostgreSQL::Test::Utils::slurp_file($cascade->logfile, $logstart);
+unlike(
+	$log,
+	qr/data checksum state .* does not match/,
+	'an intermediate restartpoint does not surface the divergence');
+
+# A checkpoint on the root primary does: the record is relayed down the whole
+# chain, so both the direct and the cascaded standby compare it against their
+# own state.  wait_for_catchup() below waits for replay, not just receipt, so
+# the record has been through xlog_redo() on both nodes once it returns.
+$primary->safe_psql('postgres', 'CHECKPOINT;');
+$primary->wait_for_catchup($standby);
+$standby->wait_for_catchup($cascade);
+
+$log = PostgreSQL::Test::Utils::slurp_file($cascade->logfile, $logstart);
+like(
+	$log,
+	qr/data checksum state .* does not match/,
+	'the cascaded standby reports the divergence on the primary checkpoint');
+
+$cascade->stop;
+$standby->stop;
+$primary->stop;
+
+done_testing();
diff --git a/src/test/modules/test_checksums/t/021_rewind_divergent_transitions.pl b/src/test/modules/test_checksums/t/021_rewind_divergent_transitions.pl
new file mode 100644
index 00000000000..a6c8a1091ee
--- /dev/null
+++ b/src/test/modules/test_checksums/t/021_rewind_divergent_transitions.pl
@@ -0,0 +1,193 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# pg_rewind across online data checksum transitions on both sides of a
+# divergence, with the two nodes trading roles between the scenarios.
+#
+# Scenario 1: after a switchover the new primary enables checksums
+# online, while the old primary restarts on its old timeline, advances
+# its WAL beyond the enable records and runs an online enable/disable
+# cycle of its own.  The old primary ends "off" with a checksum
+# watermark numerically above every checksum record the new primary has
+# written.  pg_rewind clamps the watermark it keeps to the divergence
+# point; without the clamp, replay on the rewound node would skip the
+# source's enable as already applied and stay "off" under an "on"
+# primary.
+#
+# Scenario 2: checksums are disabled again, and after another
+# switchover both nodes enable them online independently, so the
+# divergence checkpoint carries "off" while both control files say
+# "on".  The rewind is allowed, the target keeps its own "on" state,
+# and replay re-walks the source's enable onto the already enabled
+# node.
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+use FindBin;
+use lib $FindBin::RealBin;
+
+use DataChecksums::Utils;
+
+sub controldata_watermark
+{
+	my ($node) = @_;
+	my ($stdout) = run_command([ 'pg_controldata', $node->data_dir ]);
+	$stdout =~ /^Data checksum watermark:\s*([0-9A-F]+)\/([0-9A-F]+)$/m
+	  or die "watermark missing from pg_controldata output";
+	return (hex($1) << 32) + hex($2);
+}
+
+# Wait until the standby has replayed the shutdown checkpoint of the
+# stopped primary, so that a subsequent promotion diverges after it and
+# the shutdown checkpoint becomes the last common checkpoint.
+sub wait_for_shutdown_checkpoint_replay
+{
+	my ($primary, $standby) = @_;
+	my ($stdout) = run_command([ 'pg_controldata', $primary->data_dir ]);
+	$stdout =~ /^Latest checkpoint location:\s*([0-9A-F\/]+)$/m
+	  or die "checkpoint location missing from pg_controldata output";
+	my $shutdown_checkpoint = $1;
+
+	$standby->poll_query_until('postgres',
+		"SELECT pg_last_wal_replay_lsn() > '$shutdown_checkpoint'::pg_lsn;")
+	  or die "standby never replayed the shutdown checkpoint";
+}
+
+# Old primary, checksums off.  wal_log_hints is required by pg_rewind
+# on a cluster without data checksums.
+my $node_a = PostgreSQL::Test::Cluster->new('node_a');
+$node_a->init(allows_streaming => 1, no_data_checksums => 1);
+$node_a->append_conf(
+	'postgresql.conf', qq[
+autovacuum = off
+wal_log_hints = on
+wal_keep_size = '1GB'
+]);
+$node_a->start;
+$node_a->safe_psql('postgres',
+	"CREATE TABLE t AS SELECT generate_series(1,10000) AS a;");
+$node_a->safe_psql('postgres', "CREATE TABLE t_div (a int);");
+
+$node_a->backup('backup');
+my $node_b = PostgreSQL::Test::Cluster->new('node_b');
+$node_b->init_from_backup($node_a, 'backup', has_streaming => 1);
+$node_b->start;
+$node_a->wait_for_catchup($node_b, 'replay', $node_a->lsn('insert'));
+
+# Clean switchover to B; enable checksums online on it.
+$node_a->stop('fast');
+wait_for_shutdown_checkpoint_replay($node_a, $node_b);
+$node_b->promote;
+enable_data_checksums($node_b, wait => 'on');
+test_checksum_state($node_b, 'on');
+
+$node_b->stop('fast');
+my $watermark_b = controldata_watermark($node_b);
+$node_b->start;
+
+# Accidental restart of the old primary on the old timeline.  Advance
+# its WAL beyond the enable watermark of B, then run an online enable
+# and disable cycle: the node ends "off" with a watermark above every
+# checksum record B has written.
+$node_a->start;
+$node_a->safe_psql('postgres', "INSERT INTO t_div VALUES (1);");
+$node_a->safe_psql('postgres',
+	"CREATE TABLE t_pad AS SELECT generate_series(1,200000) AS a;");
+enable_data_checksums($node_a, wait => 'on');
+disable_data_checksums($node_a, wait => 'off');
+test_checksum_state($node_a, 'off');
+$node_a->stop('fast');
+
+my $watermark_a = controldata_watermark($node_a);
+die "test broken: target watermark not above the source's enable"
+  unless $watermark_a > $watermark_b;
+
+command_ok(
+	[
+		'pg_rewind',
+		'--target-pgdata' => $node_a->data_dir,
+		'--source-server' => $node_b->connstr('postgres'),
+	],
+	'pg_rewind from the promoted node');
+
+# Start the rewound node as a standby of B.  Replay from the last
+# common checkpoint runs through B's online enable, which the target
+# never saw, so the rewound node must converge to "on".
+my $connstr_b = $node_b->connstr;
+$node_a->append_conf(
+	'postgresql.conf', qq[
+port = @{[$node_a->port]}
+primary_conninfo = '$connstr_b application_name=@{[$node_a->name]}'
+]);
+$node_a->set_standby_mode;
+$node_a->start;
+
+$node_b->wait_for_catchup($node_a, 'replay', $node_b->lsn('insert'));
+test_checksum_state($node_a, 'on');
+
+is($node_a->safe_psql('postgres', "SELECT count(*) FROM t_div;"),
+	'0', 'divergent insert was rewound');
+
+# Scenario 2, reusing the pair with the roles reversed.  Disable
+# checksums online so the next divergence point carries "off", and let
+# A replay the change.
+disable_data_checksums($node_b, wait => 'off');
+$node_b->wait_for_catchup($node_a, 'replay', $node_b->lsn('insert'));
+test_checksum_state($node_a, 'off');
+
+# Clean switchover back to A; enable checksums online on it.
+$node_b->stop('fast');
+wait_for_shutdown_checkpoint_replay($node_b, $node_a);
+$node_a->promote;
+enable_data_checksums($node_a, wait => 'on');
+test_checksum_state($node_a, 'on');
+my $source_enable_watermark = controldata_watermark($node_a);
+
+# The old primary restarts on its old timeline and enables checksums
+# online independently: both control files say "on", the divergence
+# checkpoint says "off".
+$node_b->start;
+$node_b->safe_psql('postgres', "INSERT INTO t_div VALUES (2);");
+enable_data_checksums($node_b, wait => 'on');
+test_checksum_state($node_b, 'on');
+$node_b->stop('fast');
+
+command_ok(
+	[
+		'pg_rewind',
+		'--target-pgdata' => $node_b->data_dir,
+		'--source-server' => $node_a->connstr('postgres'),
+	],
+	'pg_rewind with checksums enabled on both nodes');
+
+my $connstr_a = $node_a->connstr;
+$node_b->append_conf(
+	'postgresql.conf', qq[
+port = @{[$node_b->port]}
+primary_conninfo = '$connstr_a application_name=@{[$node_b->name]}'
+]);
+$node_b->set_standby_mode;
+$node_b->start;
+
+$node_a->wait_for_catchup($node_b, 'replay', $node_a->lsn('insert'));
+test_checksum_state($node_b, 'on');
+
+is($node_b->safe_psql('postgres', "SELECT count(*) FROM t;"),
+	'10000', 'data readable on the rewound node');
+is($node_b->safe_psql('postgres', "SELECT count(*) FROM t_div;"),
+	'0', 'divergent insert was rewound');
+
+$node_b->stop('fast');
+is(controldata_watermark($node_b), $source_enable_watermark,
+	'rewound node replayed the source checksum transition');
+command_ok([ 'pg_checksums', '--check', '-D', $node_b->data_dir ],
+	'checksums valid on the rewound node');
+
+$node_a->stop('fast');
+command_ok([ 'pg_checksums', '--check', '-D', $node_a->data_dir ],
+	'checksums valid on the twice-rewound node');
+
+done_testing();
-- 
2.55.0

