canic-core 0.93.13

//! Module: workflow::pool::create_empty
//!
//! Responsibility: create empty pool canisters behind replay and cost guards.
//! Does not own: endpoint authorization, stable pool schemas, or management-call ops.
//! Boundary: pool workflow validates admin access, reserves replay/cost, then calls ops.

use crate::{
    InternalError, InternalErrorOrigin,
    cdk::types::{Cycles, Principal, TC},
    dto::{
        error::Error,
        pool::{CreateEmptyPoolRequest, PoolAdminResponse},
        rpc::RootRequestMetadata,
    },
    model::replay::{
        CommandKind, ExternalEffectDescriptor, OperationId, RecoveryReason, ReplayActor,
        ReplayPayloadHasher, ReplayReceipt,
    },
    ops::{
        cost_guard::{CostGuardOps, CostGuardPermit, CostGuardRequest},
        ic::{IcOps, mgmt::MgmtOps},
        replay::{
            self as replay_ops, POOL_CREATE_EMPTY_REPLAY_RESPONSE_SCHEMA_VERSION,
            guard::secs_to_ns,
            receipt::{
                ReplayReceiptDecision, ReplayReceiptReserveInput, ReplayReceiptStoreError,
                ReplayReceiptToken, abort_reserved_receipt, commit_staged_receipt_response,
                mark_costed_external_effect_in_flight, mark_recovery_required,
                replay_cost_guard_settlement, reserve_or_replay_receipt, stage_receipt_response,
            },
        },
        runtime::metrics::{
            pool::{PoolMetricOperation as MetricOperation, PoolMetricReason as MetricReason},
            recording::PoolMetricEvent as MetricEvent,
        },
        storage::pool::PoolOps,
    },
    replay_policy::CostClass,
    workflow::{cost_guard::map_cost_guard_reserve_error, pool::PoolWorkflow},
};

/// Default cycles allocated to freshly created pool canisters.
const POOL_CANISTER_CYCLES: u128 = 5 * TC;
const POOL_CREATE_EMPTY_REPLAY_COMMAND_KIND: &str = "pool.create_empty.v1";
const POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS: u64 = 300_000_000_000;
const POOL_CREATE_EMPTY_QUOTA_WINDOW_SECONDS: u64 = 60;
const POOL_CREATE_EMPTY_MAX_OPERATIONS_PER_WINDOW: u64 = 10;
const POOL_CREATE_EMPTY_MIN_CYCLES_AFTER_RESERVATION: u128 = TC;

impl PoolWorkflow {
    pub async fn pool_create_canister(
        request: CreateEmptyPoolRequest,
    ) -> Result<Principal, InternalError> {
        MetricEvent::started(MetricOperation::CreateEmpty);
        if let Err(err) = Self::require_pool_admin() {
            MetricEvent::failed(MetricOperation::CreateEmpty, &err);
            return Err(err);
        }

        let metadata = match pool_create_empty_replay_metadata(request.metadata) {
            Ok(metadata) => metadata,
            Err(err) => {
                MetricEvent::failed(MetricOperation::CreateEmpty, &err);
                return Err(err);
            }
        };
        let caller = IcOps::msg_caller();
        let (command_kind, token) = match reserve_pool_create_empty_replay(metadata, caller) {
            Ok(PoolCreateEmptyReplayReservation::Fresh {
                command_kind,
                token,
            }) => (command_kind, token),
            Ok(PoolCreateEmptyReplayReservation::Replay(pid)) => {
                MetricEvent::completed(MetricOperation::CreateEmpty, MetricReason::Ok);
                return Ok(pid);
            }
            Err(err) => {
                MetricEvent::failed(MetricOperation::CreateEmpty, &err);
                return Err(err);
            }
        };

        let cycles = Cycles::new(POOL_CANISTER_CYCLES);
        let controllers = match Self::pool_controllers() {
            Ok(controllers) => controllers,
            Err(err) => {
                abort_reserved_receipt(&token).map_err(map_pool_create_empty_replay_store_error)?;
                MetricEvent::failed(MetricOperation::CreateEmpty, &err);
                return Err(err);
            }
        };
        let cost_permit = match reserve_pool_create_empty_cost_guard(&command_kind, caller) {
            Ok(permit) => permit,
            Err(err) => {
                abort_reserved_receipt(&token).map_err(map_pool_create_empty_replay_store_error)?;
                MetricEvent::failed(MetricOperation::CreateEmpty, &err);
                return Err(err);
            }
        };

        if let Err(err) =
            mark_pool_create_empty_external_effect(&token, &command_kind, &cost_permit)
        {
            let err = CostGuardOps::recover_after_failure(&cost_permit, IcOps::now_secs(), err);
            MetricEvent::failed(MetricOperation::CreateEmpty, &err);
            return Err(err);
        }

        let pid =
            match MgmtOps::create_canister_with_permit(&cost_permit, controllers, cycles.clone())
                .await
            {
                Ok(pid) => pid,
                Err(err) => {
                    let err =
                        CostGuardOps::recover_after_failure(&cost_permit, IcOps::now_secs(), err);
                    mark_recovery_required(
                        &token,
                        RecoveryReason::ExternalEffectStatusUnknown,
                        secs_to_ns(IcOps::now_secs()),
                    )
                    .map_err(map_pool_create_empty_replay_store_error)?;
                    MetricEvent::failed(MetricOperation::CreateEmpty, &err);
                    return Err(err);
                }
            };

        let created_at = IcOps::now_secs();
        PoolOps::register_ready(pid, cycles, None, None, None, created_at);
        let response = PoolAdminResponse::Created { pid };
        match encode_pool_create_empty_response(&response) {
            Ok(response_bytes) => {
                commit_pool_create_empty_success(&token, &cost_permit, created_at, response_bytes)?;
            }
            Err(err) => {
                let err = preserve_pool_create_empty_response_failure(&token, &cost_permit, err);
                MetricEvent::failed(MetricOperation::CreateEmpty, &err);
                return Err(err);
            }
        }

        MetricEvent::completed(MetricOperation::CreateEmpty, MetricReason::Ok);

        Ok(pid)
    }
}

fn pool_create_empty_replay_metadata(
    metadata: Option<RootRequestMetadata>,
) -> Result<RootRequestMetadata, InternalError> {
    let metadata = metadata.ok_or_else(|| InternalError::public(Error::operation_id_required()))?;
    if metadata.ttl_ns == 0 {
        return Err(InternalError::public(Error::invalid(
            "pool create-empty replay metadata ttl_ns must be greater than zero",
        )));
    }
    if metadata.ttl_ns > POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS {
        return Err(InternalError::public(Error::invalid(format!(
            "pool create-empty replay metadata ttl_ns={} exceeds max {}",
            metadata.ttl_ns, POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS
        ))));
    }
    Ok(metadata)
}

///
/// PoolCreateEmptyReplayReservation
///
/// Replay reservation outcome for one empty-pool creation request.
/// Owned by pool workflow and mapped into either execution or cached response.
///
enum PoolCreateEmptyReplayReservation {
    Fresh {
        command_kind: CommandKind,
        token: Box<ReplayReceiptToken>,
    },
    Replay(Principal),
}

fn reserve_pool_create_empty_replay(
    metadata: RootRequestMetadata,
    caller: Principal,
) -> Result<PoolCreateEmptyReplayReservation, InternalError> {
    let command_kind = pool_create_empty_command_kind();
    let actor = ReplayActor::direct_caller(caller);
    let payload_hash = pool_create_empty_payload_hash(&command_kind, &actor);
    let now_secs = IcOps::now_secs();
    let replay_input = pool_create_empty_replay_input(
        command_kind.clone(),
        metadata.request_id,
        actor,
        payload_hash,
        secs_to_ns(now_secs),
        metadata.ttl_ns,
    )?;

    match reserve_or_replay_receipt(replay_input) {
        Ok(ReplayReceiptDecision::Fresh(token)) => Ok(PoolCreateEmptyReplayReservation::Fresh {
            command_kind,
            token: Box::new(token),
        }),
        Ok(ReplayReceiptDecision::RecoveryRequired {
            token,
            reason:
                reason @ (RecoveryReason::CostSettlementFailed | RecoveryReason::ResponseCommitFailed),
        }) => recover_pool_create_empty_response(&token, reason)
            .map(PoolCreateEmptyReplayReservation::Replay),
        Ok(decision) => map_pool_create_empty_replay_decision(decision)
            .map(PoolCreateEmptyReplayReservation::Replay),
        Err(err) => Err(map_pool_create_empty_replay_store_error(err)),
    }
}

fn pool_create_empty_replay_input(
    command_kind: CommandKind,
    request_id: [u8; 32],
    actor: ReplayActor,
    payload_hash: [u8; 32],
    now_ns: u64,
    ttl_ns: u64,
) -> Result<ReplayReceiptReserveInput, InternalError> {
    let expires_at_ns = now_ns.checked_add(ttl_ns).ok_or_else(|| {
        InternalError::public(Error::invalid(
            "pool create-empty replay metadata ttl_ns overflows nanoseconds",
        ))
    })?;
    Ok(ReplayReceiptReserveInput::new(
        command_kind,
        OperationId::from_bytes(request_id),
        actor,
        payload_hash,
        now_ns,
    )
    .with_expires_at_ns(expires_at_ns))
}

fn reserve_pool_create_empty_cost_guard(
    command_kind: &CommandKind,
    caller: Principal,
) -> Result<CostGuardPermit, InternalError> {
    CostGuardOps::reserve(CostGuardRequest {
        cost_class: CostClass::ManagementDeployment,
        command_kind: command_kind.clone(),
        quota_subject: caller,
        payer: IcOps::canister_self(),
        now_secs: IcOps::now_secs(),
        quota_window_secs: POOL_CREATE_EMPTY_QUOTA_WINDOW_SECONDS,
        max_operations_per_window: POOL_CREATE_EMPTY_MAX_OPERATIONS_PER_WINDOW,
        current_cycle_balance: IcOps::canister_cycle_balance().to_u128(),
        cycle_reservation_cycles: POOL_CANISTER_CYCLES,
        min_cycles_after_reservation: POOL_CREATE_EMPTY_MIN_CYCLES_AFTER_RESERVATION,
    })
    .map_err(map_cost_guard_reserve_error)
}

fn mark_pool_create_empty_external_effect(
    token: &ReplayReceiptToken,
    command_kind: &CommandKind,
    cost_permit: &CostGuardPermit,
) -> Result<(), InternalError> {
    mark_costed_external_effect_in_flight(
        token,
        ExternalEffectDescriptor::ManagementCreateCanister {
            command_kind: command_kind.clone(),
        },
        cost_permit.replay_settlement(),
        secs_to_ns(IcOps::now_secs()),
    )
    .map_err(map_pool_create_empty_replay_store_error)
}

fn commit_pool_create_empty_success(
    token: &ReplayReceiptToken,
    cost_permit: &CostGuardPermit,
    created_at: u64,
    response_bytes: Vec<u8>,
) -> Result<(), InternalError> {
    if let Err(err) = stage_receipt_response(
        token,
        POOL_CREATE_EMPTY_REPLAY_RESPONSE_SCHEMA_VERSION,
        response_bytes,
        secs_to_ns(created_at),
    ) {
        return Err(preserve_pool_create_empty_response_failure(
            token,
            cost_permit,
            map_pool_create_empty_replay_store_error(err),
        ));
    }
    if let Err(err) = CostGuardOps::complete(cost_permit, IcOps::now_secs()) {
        if let Err(recovery_err) = mark_recovery_required(
            token,
            RecoveryReason::CostSettlementFailed,
            secs_to_ns(IcOps::now_secs()),
        )
        .map_err(map_pool_create_empty_replay_store_error)
        {
            return Err(err.with_diagnostic_context(format!(
                "pool create-empty replay recovery marker failed: {recovery_err}"
            )));
        }
        return Err(err);
    }
    if let Err(err) = commit_staged_receipt_response(token, secs_to_ns(IcOps::now_secs())) {
        let mut err = map_pool_create_empty_replay_store_error(err);
        if let Err(recovery_err) = mark_recovery_required(
            token,
            RecoveryReason::ResponseCommitFailed,
            secs_to_ns(IcOps::now_secs()),
        )
        .map_err(map_pool_create_empty_replay_store_error)
        {
            err = err.with_diagnostic_context(format!(
                "pool create-empty replay recovery marker failed: {recovery_err}"
            ));
        }
        return Err(err);
    }
    Ok(())
}

fn preserve_pool_create_empty_response_failure(
    token: &ReplayReceiptToken,
    cost_permit: &CostGuardPermit,
    mut err: InternalError,
) -> InternalError {
    let reason = match CostGuardOps::complete(cost_permit, IcOps::now_secs()) {
        Ok(()) => RecoveryReason::ResponseCommitFailed,
        Err(settlement_err) => {
            err = err.with_diagnostic_context(format!(
                "pool create-empty cost settlement also failed: {settlement_err}"
            ));
            RecoveryReason::CostSettlementFailed
        }
    };
    if let Err(recovery_err) = mark_recovery_required(token, reason, secs_to_ns(IcOps::now_secs()))
        .map_err(map_pool_create_empty_replay_store_error)
    {
        err = err.with_diagnostic_context(format!(
            "pool create-empty replay recovery marker failed: {recovery_err}"
        ));
    }
    err
}

fn recover_pool_create_empty_response(
    token: &ReplayReceiptToken,
    reason: RecoveryReason,
) -> Result<Principal, InternalError> {
    let cost_settled = reason == RecoveryReason::CostSettlementFailed;
    if cost_settled {
        let settlement = replay_cost_guard_settlement(token)
            .map_err(map_pool_create_empty_replay_store_error)?;
        CostGuardOps::complete_replay_settlement(&settlement, IcOps::now_secs())?;
    }
    let receipt = match commit_staged_receipt_response(token, secs_to_ns(IcOps::now_secs())) {
        Ok(receipt) => receipt,
        Err(err) => {
            let mut err = map_pool_create_empty_replay_store_error(err);
            if cost_settled
                && let Err(recovery_err) = mark_recovery_required(
                    token,
                    RecoveryReason::ResponseCommitFailed,
                    secs_to_ns(IcOps::now_secs()),
                )
                .map_err(map_pool_create_empty_replay_store_error)
            {
                err = err.with_diagnostic_context(format!(
                    "pool create-empty response recovery marker failed: {recovery_err}"
                ));
            }
            return Err(err);
        }
    };
    decode_pool_create_empty_response(&receipt)
}

fn pool_create_empty_command_kind() -> CommandKind {
    CommandKind::new(POOL_CREATE_EMPTY_REPLAY_COMMAND_KIND)
        .expect("pool create-empty replay command kind is a valid static label")
}

fn pool_create_empty_payload_hash(command_kind: &CommandKind, actor: &ReplayActor) -> [u8; 32] {
    let mut hasher = ReplayPayloadHasher::new(command_kind, actor);
    hasher.hash_str("create_empty");
    hasher.finish()
}

fn map_pool_create_empty_replay_decision(
    decision: ReplayReceiptDecision,
) -> Result<Principal, InternalError> {
    match decision {
        ReplayReceiptDecision::Fresh(_) => Err(InternalError::invariant(
            InternalErrorOrigin::Workflow,
            "fresh pool create replay decision escaped",
        )),
        ReplayReceiptDecision::ReturnCommitted(receipt) => {
            decode_pool_create_empty_response(&receipt)
        }
        ReplayReceiptDecision::OperationInProgress => Err(InternalError::public(Error::conflict(
            "pool create-empty request is already in progress; retry later with the same request id",
        ))),
        ReplayReceiptDecision::ActorMismatch => Err(InternalError::public(Error::conflict(
            "pool create-empty request id was reused by a different caller",
        ))),
        ReplayReceiptDecision::PayloadMismatch => Err(InternalError::public(Error::conflict(
            "pool create-empty request id was reused with a different payload",
        ))),
        ReplayReceiptDecision::Expired => Err(InternalError::public(Error::conflict(
            "pool create-empty replay receipt expired; retry with a new request id",
        ))),
        ReplayReceiptDecision::RecoveryRequired { reason, .. } => {
            Err(InternalError::public(Error::conflict(format!(
                "pool create-empty request requires recovery before replay: {reason:?}"
            ))))
        }
        ReplayReceiptDecision::PendingActorQuotaExceeded { max_pending, .. } => {
            Err(InternalError::public(Error::exhausted(format!(
                "pool create-empty pending replay receipt quota exceeded for caller; max_pending={max_pending}"
            ))))
        }
        ReplayReceiptDecision::PendingCommandQuotaExceeded { max_pending, .. } => {
            Err(InternalError::public(Error::exhausted(format!(
                "pool create-empty pending replay receipt quota exceeded for command kind; max_pending={max_pending}"
            ))))
        }
    }
}

fn map_pool_create_empty_replay_store_error(err: ReplayReceiptStoreError) -> InternalError {
    match err {
        ReplayReceiptStoreError::ReceiptMissing => InternalError::workflow(
            InternalErrorOrigin::Workflow,
            "pool create-empty replay receipt is missing",
        ),
        ReplayReceiptStoreError::ReceiptDecodeFailed(message) => InternalError::workflow(
            InternalErrorOrigin::Workflow,
            format!("failed to decode pool create-empty replay receipt: {message}"),
        ),
        ReplayReceiptStoreError::StagedResponseMissing => InternalError::workflow(
            InternalErrorOrigin::Workflow,
            "pool create-empty replay receipt is missing staged response data",
        ),
        ReplayReceiptStoreError::CostGuardSettlementMissing => InternalError::workflow(
            InternalErrorOrigin::Workflow,
            "pool create-empty replay receipt is missing cost guard settlement identity",
        ),
    }
}

fn encode_pool_create_empty_response(
    response: &PoolAdminResponse,
) -> Result<Vec<u8>, InternalError> {
    replay_ops::encode_pool_create_empty_replay_response(response).map_err(|err| match err {
        replay_ops::ReplayCommitError::EncodeFailed(message) => InternalError::workflow(
            InternalErrorOrigin::Workflow,
            format!("failed to encode pool create-empty replay response: {message}"),
        ),
    })
}

fn decode_pool_create_empty_response(receipt: &ReplayReceipt) -> Result<Principal, InternalError> {
    replay_ops::decode_pool_create_empty_replay_response(receipt).map_err(|err| match err {
        replay_ops::ReplayDecodeError::DecodeFailed(message) => {
            InternalError::workflow(InternalErrorOrigin::Workflow, message)
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        dto::error::ErrorCode,
        model::replay::{
            REPLAY_PAYLOAD_HASH_SCHEMA_VERSION, REPLAY_RECEIPT_SCHEMA_VERSION, ReplayReceipt,
            ReplayReceiptStatus,
        },
        ops::{replay::receipt::reserve_or_replay_receipt, storage::replay::ReplayReceiptOps},
    };

    fn p(id: u8) -> Principal {
        Principal::from_slice(&[id; 29])
    }

    fn metadata(id: u8, ttl_ns: u64) -> RootRequestMetadata {
        RootRequestMetadata {
            request_id: [id; 32],
            ttl_ns,
        }
    }

    fn committed_pool_create_empty_receipt(pid: Principal) -> ReplayReceipt {
        let command_kind = pool_create_empty_command_kind();
        let actor = ReplayActor::direct_caller(p(2));
        ReplayReceipt {
            schema_version: REPLAY_RECEIPT_SCHEMA_VERSION,
            command_kind: command_kind.clone(),
            operation_id: OperationId::from_bytes([7; 32]),
            actor,
            payload_hash_schema_version: REPLAY_PAYLOAD_HASH_SCHEMA_VERSION,
            payload_hash: pool_create_empty_payload_hash(&command_kind, &actor),
            status: ReplayReceiptStatus::Committed,
            created_at_ns: 100,
            updated_at_ns: 200,
            expires_at_ns: Some(1_000),
            response_schema_version: Some(POOL_CREATE_EMPTY_REPLAY_RESPONSE_SCHEMA_VERSION),
            response_bytes: Some(
                encode_pool_create_empty_response(&PoolAdminResponse::Created { pid })
                    .expect("response encodes"),
            ),
            staged_response_schema_version: None,
            staged_response_bytes: None,
            cost_guard_settlement: None,
            effect: Some(ExternalEffectDescriptor::ManagementCreateCanister { command_kind }),
        }
    }

    #[test]
    fn pool_create_empty_replay_metadata_rejects_missing_or_invalid_ttl() {
        let missing = pool_create_empty_replay_metadata(None).expect_err("metadata is required");
        assert_eq!(
            missing.public_error().expect("public error").code,
            ErrorCode::OperationIdRequired
        );
        assert_eq!(
            missing.public_error().expect("public error").message,
            "operation_id is required for this command"
        );

        let zero = pool_create_empty_replay_metadata(Some(metadata(1, 0)))
            .expect_err("zero ttl is invalid");
        assert_eq!(
            zero.public_error().expect("public error").code,
            ErrorCode::InvalidInput
        );

        let too_large = pool_create_empty_replay_metadata(Some(metadata(
            1,
            POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS + 1,
        )))
        .expect_err("oversized ttl is invalid");
        assert_eq!(
            too_large.public_error().expect("public error").code,
            ErrorCode::InvalidInput
        );
    }

    #[test]
    fn pool_create_empty_replay_metadata_accepts_bounded_ttl() {
        let accepted = pool_create_empty_replay_metadata(Some(metadata(
            3,
            POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS,
        )))
        .expect("bounded ttl is accepted");

        assert_eq!(accepted.request_id, [3; 32]);
        assert_eq!(accepted.ttl_ns, POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS);
    }

    #[test]
    fn pool_create_empty_payload_hash_binds_actor() {
        let command_kind = pool_create_empty_command_kind();
        let actor_a = ReplayActor::direct_caller(p(2));
        let actor_b = ReplayActor::direct_caller(p(3));

        assert_ne!(
            pool_create_empty_payload_hash(&command_kind, &actor_a),
            pool_create_empty_payload_hash(&command_kind, &actor_b)
        );
    }

    #[test]
    fn pool_create_empty_replay_decision_returns_committed_created_response() {
        let pid = p(9);
        let decision =
            ReplayReceiptDecision::ReturnCommitted(committed_pool_create_empty_receipt(pid));

        assert_eq!(
            map_pool_create_empty_replay_decision(decision).expect("committed receipt replays"),
            pid
        );
    }

    #[test]
    fn pool_create_empty_does_not_commit_replay_when_cost_settlement_fails() {
        CostGuardOps::reset_for_tests();
        let command_kind = pool_create_empty_command_kind();
        let actor = ReplayActor::direct_caller(p(2));
        let replay_input = pool_create_empty_replay_input(
            command_kind.clone(),
            [240; 32],
            actor,
            pool_create_empty_payload_hash(&command_kind, &actor),
            1_000,
            POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS,
        )
        .expect("replay input");
        let ReplayReceiptDecision::Fresh(token) =
            reserve_or_replay_receipt(replay_input).expect("fresh replay receipt")
        else {
            panic!("expected fresh replay receipt");
        };
        let permit = CostGuardOps::reserve(CostGuardRequest {
            cost_class: CostClass::ManagementDeployment,
            command_kind: command_kind.clone(),
            quota_subject: p(2),
            payer: p(3),
            now_secs: 10,
            quota_window_secs: POOL_CREATE_EMPTY_QUOTA_WINDOW_SECONDS,
            max_operations_per_window: POOL_CREATE_EMPTY_MAX_OPERATIONS_PER_WINDOW,
            current_cycle_balance: 10 * TC,
            cycle_reservation_cycles: POOL_CANISTER_CYCLES,
            min_cycles_after_reservation: POOL_CREATE_EMPTY_MIN_CYCLES_AFTER_RESERVATION,
        })
        .expect("cost permit");
        mark_pool_create_empty_external_effect(&token, &command_kind, &permit)
            .expect("mark create effect");
        CostGuardOps::abort(&permit).expect("invalidate cost permit");
        let pid = p(240);
        let response_bytes = encode_pool_create_empty_response(&PoolAdminResponse::Created { pid })
            .expect("response bytes");

        commit_pool_create_empty_success(&token, &permit, 10, response_bytes)
            .expect_err("cost settlement failure must reject replay commit");

        let receipt = ReplayReceiptOps::get(token.key())
            .expect("receipt")
            .into_receipt()
            .expect("receipt decodes");
        assert_eq!(
            receipt.status,
            ReplayReceiptStatus::RecoveryRequired {
                reason: RecoveryReason::CostSettlementFailed
            }
        );
        assert!(receipt.response_bytes.is_none());
        assert!(receipt.staged_response_bytes.is_some());
        PoolOps::remove(&pid);
    }

    #[test]
    fn pool_create_empty_retry_finishes_cost_settlement_without_create_call() {
        CostGuardOps::reset_for_tests();
        ReplayReceiptOps::reset_for_tests();
        let caller = p(2);
        let (command_kind, token) = match reserve_pool_create_empty_replay(
            metadata(241, POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS),
            caller,
        )
        .expect("fresh reservation")
        {
            PoolCreateEmptyReplayReservation::Fresh {
                command_kind,
                token,
            } => (command_kind, token),
            PoolCreateEmptyReplayReservation::Replay(_) => panic!("expected fresh reservation"),
        };
        let permit = CostGuardOps::reserve(CostGuardRequest {
            cost_class: CostClass::ManagementDeployment,
            command_kind: command_kind.clone(),
            quota_subject: caller,
            payer: p(3),
            now_secs: IcOps::now_secs(),
            quota_window_secs: POOL_CREATE_EMPTY_QUOTA_WINDOW_SECONDS,
            max_operations_per_window: POOL_CREATE_EMPTY_MAX_OPERATIONS_PER_WINDOW,
            current_cycle_balance: 10 * TC,
            cycle_reservation_cycles: POOL_CANISTER_CYCLES,
            min_cycles_after_reservation: POOL_CREATE_EMPTY_MIN_CYCLES_AFTER_RESERVATION,
        })
        .expect("cost permit");
        mark_pool_create_empty_external_effect(&token, &command_kind, &permit)
            .expect("mark costed create effect");
        let pid = p(241);
        stage_receipt_response(
            &token,
            POOL_CREATE_EMPTY_REPLAY_RESPONSE_SCHEMA_VERSION,
            encode_pool_create_empty_response(&PoolAdminResponse::Created { pid })
                .expect("response bytes"),
            0,
        )
        .expect("stage response");
        mark_recovery_required(&token, RecoveryReason::CostSettlementFailed, 0)
            .expect("mark recovery");

        let replayed = reserve_pool_create_empty_replay(
            metadata(241, POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS),
            caller,
        )
        .expect("retry should finish accounting");
        assert!(matches!(
            replayed,
            PoolCreateEmptyReplayReservation::Replay(replayed_pid) if replayed_pid == pid
        ));
        let receipt = ReplayReceiptOps::get(token.key())
            .expect("receipt")
            .into_receipt()
            .expect("receipt decodes");
        assert_eq!(receipt.status, ReplayReceiptStatus::Committed);

        ReplayReceiptOps::reset_for_tests();
        CostGuardOps::reset_for_tests();
    }

    #[test]
    fn pool_create_empty_retry_promotes_staged_response_without_create_call() {
        ReplayReceiptOps::reset_for_tests();
        let caller = p(2);
        let token = match reserve_pool_create_empty_replay(
            metadata(242, POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS),
            caller,
        )
        .expect("fresh reservation")
        {
            PoolCreateEmptyReplayReservation::Fresh { token, .. } => token,
            PoolCreateEmptyReplayReservation::Replay(_) => panic!("expected fresh reservation"),
        };
        let pid = p(242);
        stage_receipt_response(
            &token,
            POOL_CREATE_EMPTY_REPLAY_RESPONSE_SCHEMA_VERSION,
            encode_pool_create_empty_response(&PoolAdminResponse::Created { pid })
                .expect("response bytes"),
            0,
        )
        .expect("stage response");
        mark_recovery_required(&token, RecoveryReason::ResponseCommitFailed, 0)
            .expect("mark response recovery");

        let replayed = reserve_pool_create_empty_replay(
            metadata(242, POOL_CREATE_EMPTY_MAX_REPLAY_TTL_NS),
            caller,
        )
        .expect("retry should promote response");
        assert!(matches!(
            replayed,
            PoolCreateEmptyReplayReservation::Replay(replayed_pid) if replayed_pid == pid
        ));
        let receipt = ReplayReceiptOps::get(token.key())
            .expect("receipt")
            .into_receipt()
            .expect("receipt decodes");
        assert_eq!(receipt.status, ReplayReceiptStatus::Committed);

        ReplayReceiptOps::reset_for_tests();
    }
}