fix(replay): don't execute the replay target height as a checkpoint round

`deliver_batches()` derived `requires_full_state_hash` partly from its
`max_batch_height_to_deliver` argument, so the last batch of a bounded
delivery was always flagged as requiring a full state hash:

    let persist_batch = Some(height) == max_batch_height_to_deliver;
    let requires_full_state_hash = block.payload.is_summary() || persist_batch;

That flag does not only decide whether a checkpoint is written: it also
selects `ExecutionRoundType::CheckpointRound`, which *changes execution*.
A checkpoint round charges every canister for resource allocation and
usage, bypassing the `CHARGE_INTERVAL_ROUNDS` gate, and aborts all paused
executions instead of only those above a limit.

Only `ic-replay` passes `Some(..)` here, and it always does -- even
without `--replay-until-height`, it passes `Some(finalized_height)`. So
the last replayed height was executed differently from the way the subnet
executed that very same height, and the resulting state differed in the
canisters' cycle balances and consumed cycles.

That difference used to be invisible to the certified state. Since the
current certification version was bumped to `V29`, `/subnet/<subnet_id>/metrics`
includes `CanisterStates::total_consumed_cycles()`, so it now changes the
certification hash, and `ic-replay` reports

    Hash mismatch! State divergence detected for outstanding shares!

against the subnet's certification shares at that height, refusing to
proceed without manual inspection. Subnet recoveries replay to the highest
certification share height, which is essentially never a summary height,
so every recovery is affected.

Derive `requires_full_state_hash` from the block alone, and have
`ic-replay` create the checkpoint it needs by always delivering an extra
batch at the end, one height above the last replayed block. The replayed
heights are then executed exactly as the subnet executed them, and the
checkpoint round happens at a height no node ever certified.

The replayed height is therefore one above the subnet's; account for it in
`ValidateReplayStep`, which already models this via `extra_batches`.

Both added tests fail without the corresponding change.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: mraszyk

rs/consensus/src/consensus/batch_delivery.rs

@@ -219,10 +219,15 @@ pub(crate) fn deliver_batches_with_result_processor(
         };
         let (consensus_responses, canister_http_spent) =
             generate_responses_to_subnet_calls(&block, &mut batch_stats, log);
-        // This flag can only be true, if we've called deliver_batches with a height
-        // limit.  In this case we also want to have a checkpoint for that last height.
-        let persist_batch = Some(height) == max_batch_height_to_deliver;
-        let requires_full_state_hash = block.payload.is_summary() || persist_batch;
+        // Must be derived from the block alone, never from how far the caller chose
+        // to deliver: this flag also selects `ExecutionRoundType::CheckpointRound`
+        // (see `StateMachineImpl::execute_round`), which changes execution -- it
+        // forces resource allocation charging and aborts *all* paused executions.
+        // Deriving it from `max_batch_height_to_deliver` would make the last
+        // replayed round diverge from the round the subnet actually executed at
+        // that height. `ic-replay` creates the checkpoint it needs by delivering an
+        // extra batch instead, see `Player::deliver_extra_batch`.
+        let requires_full_state_hash = block.payload.is_summary();
         let batch_content = match block.payload.as_ref() {
             BlockPayload::Summary(_summary_payload) => BatchContent::Data {
                 batch_messages: BatchMessages::default(),
@@ -577,10 +582,13 @@ mod tests {
     //! Finalizer unit tests
     use super::*;
     use crate::consensus::batch_delivery::generate_responses_to_remote_dkgs;
+    use ic_consensus_mocks::{Dependencies, DependenciesBuilder};
     use ic_crypto_test_utils_ni_dkg::dummy_transcript_for_tests;
     use ic_logger::replica_logger::no_op_logger;
     use ic_management_canister_types_private::{SetupInitialDKGResponse, VetKdCurve, VetKdKeyId};
-    use ic_test_utilities_types::ids::subnet_test_id;
+    use ic_test_utilities::message_routing::FakeMessageRouting;
+    use ic_test_utilities_registry::SubnetRecordBuilder;
+    use ic_test_utilities_types::ids::{node_test_id, subnet_test_id};
     use ic_types::{
         PrincipalId, RegistryVersion, SubnetId,
         batch::{BatchPayload, ValidationContext},
@@ -601,6 +609,54 @@ mod tests {
 
     const TARGET_ID: NiDkgTargetId = NiDkgTargetId::new([8; 32]);
 
+    /// `requires_full_state_hash` selects `ExecutionRoundType::CheckpointRound`,
+    /// which changes execution. It must therefore depend only on the block, so
+    /// that a caller bounding the delivery (i.e. `ic-replay`) still executes each
+    /// round exactly the way the subnet executed it.
+    #[test]
+    fn requires_full_state_hash_ignores_max_batch_height_to_deliver() {
+        ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| {
+            let dkg_interval_length = 9;
+            let node_ids = [node_test_id(0)];
+            let record = SubnetRecordBuilder::from(&node_ids)
+                .with_dkg_interval_length(dkg_interval_length)
+                .build();
+            let subnet_id = subnet_test_id(0);
+            let Dependencies {
+                registry, mut pool, ..
+            } = DependenciesBuilder::single_subnet(pool_config, subnet_id, vec![(1, record)])
+                .build();
+
+            // Summary blocks are at heights 0, 10, ...; finalize a few rounds and
+            // stop short of the next summary height.
+            let target_height = Height::from(5);
+            pool.advance_round_normal_operation_n(target_height.get());
+
+            let membership = Membership::new(pool.get_cache(), registry.clone(), subnet_id);
+            let message_routing = FakeMessageRouting::new();
+
+            let last_delivered = deliver_batches(
+                &message_routing,
+                &membership,
+                &PoolReader::new(&pool),
+                registry.as_ref(),
+                subnet_id,
+                &no_op_logger(),
+                Some(target_height),
+            )
+            .expect("failed to deliver batches");
+            assert_eq!(last_delivered, target_height);
+
+            let batches = message_routing.batches.read().unwrap();
+            let last_batch = batches.last().expect("no batch was delivered");
+            assert_eq!(last_batch.batch_number, target_height);
+            assert!(
+                !last_batch.requires_full_state_hash(),
+                "the batch at the delivery bound must not be a checkpoint round"
+            );
+        })
+    }
+
     const EXPECTED_FRESH_SUBNET_ID_STR: &str =
         "icdrs-3sfmz-hm6r3-cdzf5-cfroa-3cddh-aght7-azz25-eo34b-4strl-wae";
 

rs/execution_environment/src/scheduler/tests/charging.rs

@@ -118,6 +118,59 @@ fn charging_happens_on_average_once_every_charge_interval_rounds() {
     assert_eq!(num_charges, NUM_ROUNDS / CHARGE_INTERVAL_ROUNDS);
 }
 
+/// The round type is observable in the canisters' cycle state: on a round that is
+/// not a multiple of `CHARGE_INTERVAL_ROUNDS`, a `CheckpointRound` charges for
+/// resource allocation while an `OrdinaryRound` does not.
+///
+/// This is why `requires_full_state_hash` -- which selects the round type -- must
+/// be derived from the block alone and never from how far a caller chose to
+/// deliver batches (see `deliver_batches()` in `ic_consensus`). Replaying a round
+/// with the wrong round type produces a state that differs from the one the
+/// subnet computed for that same height.
+#[test]
+fn round_type_decides_whether_a_non_charging_round_charges() {
+    // A round on which charging is not otherwise due.
+    const ROUND: u64 = CHARGE_INTERVAL_ROUNDS + 1;
+    assert!(!ROUND.is_multiple_of(CHARGE_INTERVAL_ROUNDS));
+
+    let mut test = SchedulerTestBuilder::new().build();
+
+    // Charging handles time=0 as a special case, so it should be set to some
+    // non-zero time.
+    let initial_time = Time::from_nanos_since_unix_epoch(1_000_000_000_000);
+    test.set_time(initial_time);
+
+    let canister = test.create_canister_with(
+        Cycles::new(1_000_000_000_000_000),
+        ComputeAllocation::zero(),
+        MemoryAllocation::from(NumBytes::from(1 << 30)),
+        None,
+        Some(initial_time),
+        None,
+    );
+
+    // Enough time has passed that a charge is due whenever charging is attempted.
+    test.advance_time(test.duration_between_allocation_charges());
+
+    // An ordinary round at `ROUND` does not charge...
+    let balance_before = test.canister_state(canister).system_state.balance();
+    test.advance_to_round(ExecutionRound::new(ROUND));
+    test.execute_round(ExecutionRoundType::OrdinaryRound);
+    assert_eq!(
+        test.canister_state(canister).system_state.balance(),
+        balance_before,
+        "an ordinary round charged for resource allocation"
+    );
+
+    // ...while a checkpoint round at the very same round does.
+    test.advance_to_round(ExecutionRound::new(ROUND));
+    test.execute_round(ExecutionRoundType::CheckpointRound);
+    assert!(
+        test.canister_state(canister).system_state.balance() < balance_before,
+        "a checkpoint round did not charge for resource allocation"
+    );
+}
+
 #[test]
 fn charging_for_message_memory_works() {
     let mut test = SchedulerTestBuilder::new()

rs/recovery/src/app_subnet_recovery.rs

@@ -522,8 +522,11 @@ impl RecoveryIterator<StepType, StepTypeIter> for AppSubnetRecovery {
             ))),
 
             StepType::ValidateReplayOutput => Ok(Box::new(
+                // `ic-replay` always delivers one extra batch at the end, to
+                // create the checkpoint that is uploaded below; the replayed
+                // height is therefore one above the subnet's.
                 self.recovery
-                    .get_validate_replay_step(self.params.subnet_id, 0),
+                    .get_validate_replay_step(self.params.subnet_id, 1),
             )),
 
             StepType::UploadState => {

rs/recovery/src/nns_recovery_failover_nodes.rs

@@ -363,8 +363,11 @@ impl RecoveryIterator<StepType, StepTypeIter> for NNSRecoveryFailoverNodes {
             )),
 
             StepType::ValidateReplayOutput => Ok(Box::new(
+                // `ic-replay` always delivers one extra batch at the end, to
+                // create the checkpoint that is uploaded below; the replayed
+                // height is therefore one above the subnet's.
                 self.recovery
-                    .get_validate_replay_step(self.params.subnet_id, 0),
+                    .get_validate_replay_step(self.params.subnet_id, 1),
             )),
 
             StepType::UpdateRegistryLocalStore => Ok(Box::new(

rs/recovery/src/nns_recovery_same_nodes.rs

@@ -449,7 +449,10 @@ impl RecoveryIterator<StepType, StepTypeIter> for NNSRecoverySameNodes {
             }
             StepType::ValidateReplayOutput => Ok(Box::new(self.recovery.get_validate_replay_step(
                 self.params.subnet_id,
-                u64::from(self.params.upgrade_version.is_some()),
+                // `ic-replay` always delivers one extra batch at the end, to create
+                // the checkpoint that is uploaded below, on top of the one delivered
+                // to update the registry local store during an upgrade.
+                1 + u64::from(self.params.upgrade_version.is_some()),
             ))),
 
             StepType::UpdateRegistryLocalStore => {

rs/replay/src/player.rs

@@ -420,34 +420,32 @@ impl Player {
             Default::default()
         };
 
-        let (latest_context_time, extra_batch_delivery) = self.deliver_extra_batch(
+        let (latest_context_time, last_batch_height, msgs) = self.deliver_extra_batch(
             self.message_routing.as_ref(),
             self.consensus_pool.as_ref(),
             extra,
         );
 
-        if let Some((last_batch_height, msgs)) = extra_batch_delivery {
-            self.wait_for_state(last_batch_height);
-            // We only want to persist the checkpoint after the latest batch.
-            self.state_manager.remove_states_below(last_batch_height);
-
-            // check if the extra messages have been delivered successfully
-            let get_latest_status = self.ingress_history_reader.get_latest_status();
-            for msg in msgs {
-                match get_latest_status(&msg.ingress.id()) {
-                    IngressStatus::Known {
-                        state: IngressState::Completed(WasmResult::Reply(bytes)),
-                        ..
-                    } => match msg.print {
-                        Some(printer) => printer(bytes),
-                        _ => println!(
-                            "Ingress id={} response={}",
-                            msg.ingress.id(),
-                            hex::encode(bytes)
-                        ),
-                    },
-                    status => panic!("Execution of {} has failed: {:?}", msg.ingress.id(), status),
-                }
+        self.wait_for_state(last_batch_height);
+        // We only want to persist the checkpoint after the latest batch.
+        self.state_manager.remove_states_below(last_batch_height);
+
+        // check if the extra messages have been delivered successfully
+        let get_latest_status = self.ingress_history_reader.get_latest_status();
+        for msg in msgs {
+            match get_latest_status(&msg.ingress.id()) {
+                IngressStatus::Known {
+                    state: IngressState::Completed(WasmResult::Reply(bytes)),
+                    ..
+                } => match msg.print {
+                    Some(printer) => printer(bytes),
+                    _ => println!(
+                        "Ingress id={} response={}",
+                        msg.ingress.id(),
+                        hex::encode(bytes)
+                    ),
+                },
+                status => panic!("Execution of {} has failed: {:?}", msg.ingress.id(), status),
             }
         }
 
@@ -736,7 +734,7 @@ impl Player {
         message_routing: &dyn MessageRouting,
         pool: Option<&ConsensusPoolImpl>,
         mut extra: F,
-    ) -> (Time, Option<(Height, Vec<IngressWithPrinter>)>) {
+    ) -> (Time, Height, Vec<IngressWithPrinter>) {
         let (registry_version, time, randomness, replica_version) = match pool {
             None => (
                 self.registry.get_latest_version(),
@@ -766,9 +764,13 @@ impl Player {
         };
 
         let extra_msgs = extra(self, time);
-        if extra_msgs.is_empty() {
-            return (time, None);
-        }
+        // `deliver_batches()` deliberately does not force a checkpoint at the replay
+        // target height: that height has to be executed exactly the way the subnet
+        // executed it, so that the resulting certified state can be compared against
+        // the subnet's certification shares (see `redeliver_certifications`).
+        // Therefore we always deliver at least one extra batch here, whose (final)
+        // round is the one that creates the checkpoint.
+        let no_extra_msgs = extra_msgs.is_empty();
 
         let extra_ingresses = extra_msgs
             .iter()
@@ -786,7 +788,7 @@ impl Player {
                 chain_key_data: Default::default(),
                 consensus_responses: Vec::new(),
                 canister_http_spent: Default::default(),
-                requires_full_state_hash: false,
+                requires_full_state_hash: no_extra_msgs,
             },
             // Use a fake randomness here since we don't have random tape for extra messages
             randomness,
@@ -796,7 +798,11 @@ impl Player {
             replica_version,
         };
 
-        println!("extra_batch created with new ingress");
+        if no_extra_msgs {
+            println!("extra_batch created to trigger checkpoint creation");
+        } else {
+            println!("extra_batch created with new ingress");
+        }
 
         loop {
             match message_routing.deliver_batch(extra_batch.clone()) {
@@ -841,7 +847,7 @@ impl Player {
                 }
             }
         }
-        (time, Some((extra_batch.batch_number, extra_msgs)))
+        (time, extra_batch.batch_number, extra_msgs)
     }
 
     fn certify_state_with_dummy_certification(&self) {