canic-core 0.95.4

//! Module: workflow::runtime::intent
//!
//! Responsibility: schedule and run cleanup for expired pending intents.
//! Does not own: intent storage schemas, business policy, or endpoint authorization.
//! Boundary: runtime workflow timer coordinating intent storage cleanup and metrics.

use crate::{
    InternalError, InternalErrorOrigin,
    ids::IntentId,
    log,
    log::Topic,
    model::intent::{
        BeginLocalIntentInput, BeginReceiptBackedIntentInput, BeginReceiptBackedIntentResult,
        ReceiptBackedIntent, SettleReceiptBackedIntentInput, SettleReceiptBackedIntentResult,
        TerminalEvidenceDecision, is_canic_owned_intent_resource_key,
    },
    ops::{
        ic::IcOps,
        runtime::metrics::intent::{
            IntentMetricOperation, IntentMetricOutcome, IntentMetricReason, IntentMetricSurface,
            IntentMetrics,
        },
        storage::intent::{IntentStoreOps, ReceiptBackedIntentOps},
    },
    workflow::runtime::timer::{TimerDirective, TimerKey, TimerRunResult, TimerWorkflow},
};

const CLEANUP_BATCH_SIZE: usize = 32;
const NANOS_PER_SECOND: u64 = 1_000_000_000;

/// Direct workflow for locally decidable, expirable reservations.
pub struct LocalIntentWorkflow;

impl LocalIntentWorkflow {
    pub fn begin(input: BeginLocalIntentInput) -> Result<IntentId, InternalError> {
        ensure_consumer_resource_key(&input.resource_key)?;
        let now = IcOps::now_secs();
        if let Some(limit) = input.reservation_limit {
            let current = IntentStoreOps::totals(&input.resource_key).reserved_qty;
            let next = current.checked_add(input.quantity).ok_or_else(|| {
                record_intent(
                    IntentMetricSurface::Local,
                    IntentMetricOperation::CapacityCheck,
                    IntentMetricOutcome::Failed,
                    IntentMetricReason::Overflow,
                );
                InternalError::invariant(
                    InternalErrorOrigin::Workflow,
                    "local intent reservation overflow",
                )
            })?;
            if next > limit {
                record_intent(
                    IntentMetricSurface::Local,
                    IntentMetricOperation::CapacityCheck,
                    IntentMetricOutcome::Failed,
                    IntentMetricReason::Capacity,
                );
                return Err(InternalError::domain(
                    InternalErrorOrigin::Domain,
                    format!(
                        "local intent capacity exceeded key={} in_flight={current} requested={} limit={limit}",
                        input.resource_key, input.quantity
                    ),
                ));
            }
            record_intent(
                IntentMetricSurface::Local,
                IntentMetricOperation::CapacityCheck,
                IntentMetricOutcome::Completed,
                IntentMetricReason::Ok,
            );
        }

        let intent_id = IntentStoreOps::allocate_intent_id().inspect_err(|_| {
            record_intent(
                IntentMetricSurface::Local,
                IntentMetricOperation::Reserve,
                IntentMetricOutcome::Failed,
                IntentMetricReason::StorageFailed,
            );
        })?;
        IntentStoreOps::try_reserve(
            intent_id,
            input.resource_key,
            input.quantity,
            now,
            input.ttl_secs,
            now,
        )
        .inspect_err(|_| {
            record_intent(
                IntentMetricSurface::Local,
                IntentMetricOperation::Reserve,
                IntentMetricOutcome::Failed,
                IntentMetricReason::StorageFailed,
            );
        })?;
        record_intent(
            IntentMetricSurface::Local,
            IntentMetricOperation::Reserve,
            IntentMetricOutcome::Completed,
            IntentMetricReason::Ok,
        );
        IntentCleanupWorkflow::schedule_intent(intent_id)?;
        Ok(intent_id)
    }

    pub fn commit(intent_id: IntentId) -> Result<(), InternalError> {
        ensure_consumer_local_intent(intent_id)?;
        IntentStoreOps::commit_at(intent_id, IcOps::now_secs()).inspect_err(|_| {
            record_intent(
                IntentMetricSurface::Local,
                IntentMetricOperation::Commit,
                IntentMetricOutcome::Failed,
                IntentMetricReason::StorageFailed,
            );
        })?;
        record_intent(
            IntentMetricSurface::Local,
            IntentMetricOperation::Commit,
            IntentMetricOutcome::Completed,
            IntentMetricReason::Ok,
        );
        IntentCleanupWorkflow::reconcile_after_terminal();
        Ok(())
    }

    pub fn rollback(intent_id: IntentId) -> Result<(), InternalError> {
        ensure_consumer_local_intent(intent_id)?;
        IntentStoreOps::abort(intent_id).inspect_err(|_| {
            record_intent(
                IntentMetricSurface::Local,
                IntentMetricOperation::Abort,
                IntentMetricOutcome::Failed,
                IntentMetricReason::StorageFailed,
            );
        })?;
        record_intent(
            IntentMetricSurface::Local,
            IntentMetricOperation::Abort,
            IntentMetricOutcome::Completed,
            IntentMetricReason::Ok,
        );
        IntentCleanupWorkflow::reconcile_after_terminal();
        Ok(())
    }
}

/// Non-awaiting workflow facade for exact-key receipt-backed operations.
pub struct ReceiptBackedIntentWorkflow;

impl ReceiptBackedIntentWorkflow {
    pub fn begin_or_load(
        input: &BeginReceiptBackedIntentInput,
    ) -> Result<BeginReceiptBackedIntentResult, InternalError> {
        ensure_consumer_resource_key(&input.resource_key)?;
        Self::begin_or_load_authorized(input)
    }

    pub(crate) fn begin_canic_owned_or_load(
        input: &BeginReceiptBackedIntentInput,
    ) -> Result<BeginReceiptBackedIntentResult, InternalError> {
        ensure_canic_owned_resource_key(&input.resource_key)?;
        Self::begin_or_load_authorized(input)
    }

    fn begin_or_load_authorized(
        input: &BeginReceiptBackedIntentInput,
    ) -> Result<BeginReceiptBackedIntentResult, InternalError> {
        let result =
            ReceiptBackedIntentOps::begin_or_load(input, IcOps::now_nanos()).inspect_err(|_| {
                record_intent(
                    IntentMetricSurface::ReceiptBacked,
                    IntentMetricOperation::Reserve,
                    IntentMetricOutcome::Failed,
                    IntentMetricReason::StorageFailed,
                );
            })?;
        if matches!(result, BeginReceiptBackedIntentResult::Created { .. }) {
            record_intent(
                IntentMetricSurface::ReceiptBacked,
                IntentMetricOperation::Reserve,
                IntentMetricOutcome::Completed,
                IntentMetricReason::Ok,
            );
        }
        Ok(result)
    }

    pub fn load(
        operation_id: crate::model::replay::OperationId,
    ) -> Result<Option<ReceiptBackedIntent>, InternalError> {
        let intent = ReceiptBackedIntentOps::load(operation_id)?;
        if let Some(intent) = &intent {
            ensure_consumer_resource_key(&intent.resource_key)?;
        }
        Ok(intent)
    }

    pub fn settle_if_pending(
        input: &SettleReceiptBackedIntentInput,
    ) -> Result<SettleReceiptBackedIntentResult, InternalError> {
        if let Some(intent) = ReceiptBackedIntentOps::load(input.operation_id)? {
            ensure_consumer_resource_key(&intent.resource_key)?;
        }
        Self::settle_if_pending_authorized(input)
    }

    pub(crate) fn settle_canic_owned_if_pending(
        input: &SettleReceiptBackedIntentInput,
    ) -> Result<SettleReceiptBackedIntentResult, InternalError> {
        if let Some(intent) = ReceiptBackedIntentOps::load(input.operation_id)? {
            ensure_canic_owned_resource_key(&intent.resource_key)?;
        }
        Self::settle_if_pending_authorized(input)
    }

    fn settle_if_pending_authorized(
        input: &SettleReceiptBackedIntentInput,
    ) -> Result<SettleReceiptBackedIntentResult, InternalError> {
        let operation = match input.evidence.decision {
            TerminalEvidenceDecision::Committed => IntentMetricOperation::Commit,
            TerminalEvidenceDecision::RolledBack => IntentMetricOperation::Abort,
        };
        let result = ReceiptBackedIntentOps::settle_if_pending(input, IcOps::now_nanos())
            .inspect_err(|_| {
                record_intent(
                    IntentMetricSurface::ReceiptBacked,
                    operation,
                    IntentMetricOutcome::Failed,
                    IntentMetricReason::StorageFailed,
                );
            })?;
        if matches!(result, SettleReceiptBackedIntentResult::Settled { .. }) {
            record_intent(
                IntentMetricSurface::ReceiptBacked,
                operation,
                IntentMetricOutcome::Completed,
                IntentMetricReason::Ok,
            );
        }
        Ok(result)
    }
}

fn ensure_consumer_resource_key(
    resource_key: &crate::ids::IntentResourceKey,
) -> Result<(), InternalError> {
    if is_canic_owned_intent_resource_key(resource_key) {
        return Err(InternalError::invalid_input(
            "intent resource keys beginning with 'canic:' are reserved for Canic runtime authority",
        ));
    }
    Ok(())
}

fn ensure_consumer_local_intent(intent_id: IntentId) -> Result<(), InternalError> {
    if let Some(intent) = IntentStoreOps::load(intent_id)? {
        ensure_consumer_resource_key(&intent.resource_key)?;
    }
    Ok(())
}

fn ensure_canic_owned_resource_key(
    resource_key: &crate::ids::IntentResourceKey,
) -> Result<(), InternalError> {
    if !is_canic_owned_intent_resource_key(resource_key) {
        return Err(InternalError::invariant(
            InternalErrorOrigin::Workflow,
            "Canic-owned intent must use the reserved 'canic:' resource namespace",
        ));
    }
    Ok(())
}

///
/// IntentCleanupWorkflow
///

pub struct IntentCleanupWorkflow;

impl IntentCleanupWorkflow {
    /// Reconstruct the sole live cleanup deadline from the stable expiry index.
    pub fn start() -> Result<(), InternalError> {
        Self::reconcile()
    }

    fn run_due_batch() -> TimerRunResult {
        Self::run_due_batch_at(IcOps::now_secs())
    }

    fn run_due_batch_at(now: u64) -> TimerRunResult {
        let result = Self::cleanup_due_batch(now);
        let aborted = match result {
            Ok(aborted) => aborted,
            Err(err) => {
                record_cleanup_intent(
                    IntentMetricOperation::Cleanup,
                    IntentMetricOutcome::Failed,
                    IntentMetricReason::StorageFailed,
                );
                log!(Topic::Memory, Warn, "intent cleanup batch failed: {err}");
                return TimerRunResult::invariant_failure();
            }
        };

        let directive = match Self::next_directive(now) {
            Ok(directive) => directive,
            Err(err) => {
                log!(
                    Topic::Memory,
                    Warn,
                    "intent cleanup deadline reconciliation failed: {err}"
                );
                return TimerRunResult {
                    outcome: crate::domain::runtime::TimerExecutionOutcome::InvariantFailure,
                    work_count: u64::try_from(aborted).unwrap_or(u64::MAX),
                    directive: TimerDirective::Stop,
                };
            }
        };

        if aborted == 0 {
            record_cleanup_intent(
                IntentMetricOperation::Cleanup,
                IntentMetricOutcome::Completed,
                IntentMetricReason::NoExpired,
            );
            TimerRunResult::no_work(directive)
        } else {
            record_cleanup_intent(
                IntentMetricOperation::Cleanup,
                IntentMetricOutcome::Completed,
                IntentMetricReason::Ok,
            );
            log!(
                Topic::Memory,
                Info,
                "intent cleanup: aborted={aborted} batch_limit={CLEANUP_BATCH_SIZE}"
            );
            TimerRunResult::success(u64::try_from(aborted).unwrap_or(u64::MAX), directive)
        }
    }

    fn cleanup_due_batch(now_secs: u64) -> Result<usize, InternalError> {
        let due = IntentStoreOps::list_due_expiry_intents(now_secs, CLEANUP_BATCH_SIZE)?;
        let mut aborted = 0;
        for intent_id in due {
            match IntentStoreOps::abort_intent_if_pending(intent_id) {
                Ok(true) => {
                    record_cleanup_intent(
                        IntentMetricOperation::Abort,
                        IntentMetricOutcome::Completed,
                        IntentMetricReason::Expired,
                    );
                    aborted += 1;
                }
                Ok(false) => {
                    return Err(InternalError::invariant(
                        InternalErrorOrigin::Workflow,
                        format!("due intent {intent_id} ceased to be pending before cleanup"),
                    ));
                }
                Err(err) => {
                    record_cleanup_intent(
                        IntentMetricOperation::Abort,
                        IntentMetricOutcome::Failed,
                        IntentMetricReason::StorageFailed,
                    );
                    return Err(
                        err.with_diagnostic_context(format!("abort due local intent {intent_id}"))
                    );
                }
            }
        }
        Ok(aborted)
    }

    pub(crate) fn schedule_intent(intent_id: IntentId) -> Result<(), InternalError> {
        if let Some(due_at_secs) = IntentStoreOps::cleanup_due_at_secs(intent_id)? {
            Self::schedule_at(due_at_secs)?;
        }
        Ok(())
    }

    pub(crate) fn reconcile_after_terminal() {
        if let Err(err) = Self::reconcile() {
            record_cleanup_intent(
                IntentMetricOperation::Cleanup,
                IntentMetricOutcome::Failed,
                IntentMetricReason::StorageFailed,
            );
            log!(
                Topic::Memory,
                Warn,
                "intent cleanup timer reconciliation failed after terminal transition: {err}"
            );
        }
    }

    fn reconcile() -> Result<(), InternalError> {
        let deadline_ns = IntentStoreOps::next_expiry_at_secs()?
            .map(Self::deadline_ns)
            .transpose()?;
        TimerWorkflow::reconcile_at(TimerKey::IntentCleanup, deadline_ns, || async {
            Self::run_due_batch()
        });
        Ok(())
    }

    fn schedule_at(due_at_secs: u64) -> Result<(), InternalError> {
        let deadline_ns = Self::deadline_ns(due_at_secs)?;
        TimerWorkflow::schedule_at(TimerKey::IntentCleanup, deadline_ns, || async {
            Self::run_due_batch()
        });
        Ok(())
    }

    fn next_directive(now_secs: u64) -> Result<TimerDirective, InternalError> {
        match IntentStoreOps::next_expiry_at_secs()? {
            None => Ok(TimerDirective::Stop),
            Some(due_at_secs) if due_at_secs <= now_secs => Ok(TimerDirective::ContinueImmediately),
            Some(due_at_secs) => Ok(TimerDirective::ScheduleAt(Self::deadline_ns(due_at_secs)?)),
        }
    }

    fn deadline_ns(due_at_secs: u64) -> Result<u64, InternalError> {
        due_at_secs.checked_mul(NANOS_PER_SECOND).ok_or_else(|| {
            InternalError::invariant(
                InternalErrorOrigin::Workflow,
                format!("intent cleanup deadline overflows nanoseconds: {due_at_secs}"),
            )
        })
    }
}

// Record a cleanup-surface intent metric with fixed labels only.
fn record_cleanup_intent(
    operation: IntentMetricOperation,
    outcome: IntentMetricOutcome,
    reason: IntentMetricReason,
) {
    IntentMetrics::record(IntentMetricSurface::Cleanup, operation, outcome, reason);
}

fn record_intent(
    surface: IntentMetricSurface,
    operation: IntentMetricOperation,
    outcome: IntentMetricOutcome,
    reason: IntentMetricReason,
) {
    IntentMetrics::record(surface, operation, outcome, reason);
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        cdk::types::Principal,
        dto::error::ErrorCode,
        ids::IntentResourceKey,
        model::{
            intent::{PayloadBinding, TerminalEvidence},
            replay::OperationId,
        },
        test::seams,
    };

    fn canic_receipt_input() -> BeginReceiptBackedIntentInput {
        BeginReceiptBackedIntentInput {
            operation_id: OperationId::from_bytes([1; 32]),
            payload_binding: PayloadBinding::new([2; 32]),
            resource_key: IntentResourceKey::new(format!("canic:placement:{}", "a".repeat(64))),
            quantity: 1,
            reservation_limit: 1,
        }
    }

    fn assert_reserved_namespace_error(err: &InternalError) {
        assert_eq!(
            err.public_error().map(|error| error.code),
            Some(ErrorCode::InvalidInput)
        );
    }

    #[test]
    fn consumer_begin_rejects_canic_owned_resource_namespace() {
        let _guard = seams::lock();
        IntentStoreOps::reset_for_tests();

        let local_error = LocalIntentWorkflow::begin(BeginLocalIntentInput {
            resource_key: IntentResourceKey::new("canic:consumer"),
            quantity: 1,
            ttl_secs: None,
            reservation_limit: Some(1),
        })
        .expect_err("consumer local intent must not enter Canic namespace");
        assert_reserved_namespace_error(&local_error);

        let receipt_error = ReceiptBackedIntentWorkflow::begin_or_load(&canic_receipt_input())
            .expect_err("consumer receipt intent must not enter Canic namespace");
        assert_reserved_namespace_error(&receipt_error);
        assert!(
            ReceiptBackedIntentOps::load(canic_receipt_input().operation_id)
                .expect("load receipt-backed intent")
                .is_none()
        );
    }

    #[test]
    fn consumer_cannot_commit_or_rollback_canic_owned_local_intent() {
        let _guard = seams::lock();
        IntentStoreOps::reset_for_tests();
        let intent_id = IntentStoreOps::allocate_intent_id().expect("allocate internal intent");
        IntentStoreOps::try_reserve(
            intent_id,
            IntentResourceKey::new("canic:test"),
            1,
            10,
            None,
            10,
        )
        .expect("reserve internal intent");

        let commit_error = LocalIntentWorkflow::commit(intent_id)
            .expect_err("consumer commit must reject Canic-owned intent");
        assert_reserved_namespace_error(&commit_error);
        let rollback_error = LocalIntentWorkflow::rollback(intent_id)
            .expect_err("consumer rollback must reject Canic-owned intent");
        assert_reserved_namespace_error(&rollback_error);
        assert!(
            IntentStoreOps::is_pending_for_tests(intent_id)
                .expect("internal intent state remains readable")
        );
    }

    #[test]
    fn consumer_receipt_operations_cannot_observe_or_settle_canic_owned_intent() {
        let _guard = seams::lock();
        IntentStoreOps::reset_for_tests();
        let input = canic_receipt_input();

        assert!(matches!(
            ReceiptBackedIntentWorkflow::begin_canic_owned_or_load(&input)
                .expect("Canic-owned begin succeeds"),
            BeginReceiptBackedIntentResult::Created { revision: 1 }
        ));

        let load_error = ReceiptBackedIntentWorkflow::load(input.operation_id)
            .expect_err("consumer load must reject Canic-owned intent");
        assert_reserved_namespace_error(&load_error);

        let settle = SettleReceiptBackedIntentInput {
            operation_id: input.operation_id,
            expected_revision: 1,
            expected_payload_binding: input.payload_binding,
            evidence: TerminalEvidence::new(
                Principal::from_slice(&[3; 29]),
                TerminalEvidenceDecision::Committed,
                [4; 32],
            ),
        };
        let settle_error = ReceiptBackedIntentWorkflow::settle_if_pending(&settle)
            .expect_err("consumer settlement must reject Canic-owned intent");
        assert_reserved_namespace_error(&settle_error);
        assert!(matches!(
            ReceiptBackedIntentWorkflow::settle_canic_owned_if_pending(&settle)
                .expect("Canic-owned settlement succeeds"),
            SettleReceiptBackedIntentResult::Settled { .. }
        ));
    }

    #[test]
    fn consumer_receipt_namespace_remains_available_outside_canic_prefix() {
        let _guard = seams::lock();
        IntentStoreOps::reset_for_tests();
        let mut input = canic_receipt_input();
        input.resource_key = IntentResourceKey::new("app:placement");

        assert!(matches!(
            ReceiptBackedIntentWorkflow::begin_or_load(&input)
                .expect("consumer namespace remains available"),
            BeginReceiptBackedIntentResult::Created { revision: 1 }
        ));
    }

    #[test]
    fn cleanup_due_work_is_bounded_and_continues_until_the_expiry_index_is_empty() {
        let _guard = seams::lock();
        IntentStoreOps::reset_for_tests();
        let resource_key = IntentResourceKey::new("cleanup:bounded");
        for seed in 1..=33 {
            IntentStoreOps::try_reserve(IntentId(seed), resource_key.clone(), 1, 10, Some(0), 10)
                .expect("reserve due intent");
        }
        IntentStoreOps::try_reserve(IntentId(34), resource_key, 1, 10, None, 10)
            .expect("reserve TTL-free intent");

        assert_eq!(
            IntentCleanupWorkflow::cleanup_due_batch(11).expect("first bounded batch"),
            CLEANUP_BATCH_SIZE
        );
        assert!(matches!(
            IntentCleanupWorkflow::next_directive(11).expect("continuation directive"),
            TimerDirective::ContinueImmediately
        ));
        assert_eq!(
            IntentCleanupWorkflow::cleanup_due_batch(11).expect("second bounded batch"),
            1
        );
        assert!(matches!(
            IntentCleanupWorkflow::next_directive(11).expect("terminal directive"),
            TimerDirective::Stop
        ));
        assert_eq!(IntentStoreOps::pending_total().expect("pending total"), 1);
        assert_eq!(IntentStoreOps::expiry_index_total_for_tests(), 0);
    }

    #[test]
    fn cleanup_failure_preserves_the_due_intent_and_exact_expiry_entry() {
        let _guard = seams::lock();
        IntentStoreOps::reset_for_tests();
        let intent_id = IntentId(41);
        IntentStoreOps::try_reserve(
            intent_id,
            IntentResourceKey::new("cleanup:failure"),
            1,
            10,
            Some(0),
            10,
        )
        .expect("reserve due intent");
        IntentStoreOps::clear_totals_for_tests();

        IntentCleanupWorkflow::cleanup_due_batch(11)
            .expect_err("missing totals must preserve typed storage failure");
        assert!(IntentStoreOps::is_pending_for_tests(intent_id).expect("pending intent remains"));
        assert_eq!(IntentStoreOps::expiry_index_total_for_tests(), 1);
        assert_eq!(
            IntentStoreOps::list_due_expiry_intents(11, CLEANUP_BATCH_SIZE)
                .expect("due intent remains indexed"),
            vec![intent_id]
        );

        let result = IntentCleanupWorkflow::run_due_batch_at(11);
        assert_eq!(
            result.outcome,
            crate::domain::runtime::TimerExecutionOutcome::InvariantFailure
        );
        assert_eq!(result.directive, TimerDirective::Stop);
    }
}