canic_core/workflow/runtime/timer/
mod.rs

1//! Module: workflow::runtime::timer
2//!
3//! Responsibility: initialize the shared timer provider and coordinate authority suspension.
4//! Does not own: domain schedules, recurrence, provider state, or control-plane callbacks.
5//! Boundary: exact domain owners retain native claims; lifecycle uses detached native once work.
6
7use crate::{
8    InternalError,
9    ops::{
10        ic::IcOps,
11        runtime::env::EnvOps,
12        storage::async_job_recovery::{AsyncJobOwner, AsyncJobRecoveryOps},
13    },
14    workflow::{placement::acknowledgement::PlacementAcknowledgementWorkflow, runtime},
15};
16use ic_timers::{
17    DeclarationLifetime, OnceContext, OnceRegistration, ScheduleError, TimerCadence,
18    TimerCompletion, TimerDirective, TimerError as ProviderError, TimerIdentity,
19    TimerIdentityError, TimerRegistrationStatus, TimerRunResult, TimerSchedule, TimerSnapshot,
20    WatchdogReconcileState, WatchdogRegistration, WatchdogRunResult, initialize_runtime,
21    reconcile_watchdog, register_once, timer_inventory,
22};
23use std::{
24    cell::{Cell, RefCell},
25    collections::BTreeSet,
26    future::Future,
27    thread::LocalKey,
28    time::Duration,
29};
30use thiserror::Error;
31
32const RECOVERY_WATCHDOG_CADENCE: Duration = Duration::from_secs(30);
33
34thread_local! {
35    static CORE_RECOVERY_WATCHDOG: RefCell<Option<WatchdogRegistration>> = const { RefCell::new(None) };
36    static NEXT_LIFECYCLE_ID: Cell<u64> = const { Cell::new(0) };
37    static TIMERS_SUSPENDED: Cell<bool> = const { Cell::new(false) };
38}
39
40/// Failure from Canic's bounded native custody, suspension, or lifecycle coordination.
41#[non_exhaustive]
42#[derive(Debug, Error)]
43pub enum TimerError {
44    #[error("Canic timer claim custody is already borrowed")]
45    CustodyBusy,
46    #[error("Canic lifecycle timer identity allocation is exhausted")]
47    LifecycleIdentityExhausted,
48    #[error(transparent)]
49    Identity(#[from] TimerIdentityError),
50    #[error("Canic timer claim is missing")]
51    MissingClaim,
52    #[error(transparent)]
53    Provider(#[from] ProviderError),
54    #[error("timer registration rollback failed after {primary}: {cleanup}")]
55    RegistrationRollback {
56        primary: Box<Self>,
57        cleanup: Box<Self>,
58    },
59    #[error("Canic timer claim is running and cannot be suspended: {0}")]
60    RunningClaim(String),
61    #[error(transparent)]
62    Schedule(#[from] ScheduleError),
63    #[error("Canic timers are suspended for an authority snapshot")]
64    Suspended,
65    #[error("authority snapshots do not support a timer outside Canic custody: {0}")]
66    UnmanagedClaim(String),
67    #[error("Canic timer claim has the wrong scheduling policy")]
68    WrongPolicy,
69}
70
71impl From<TimerError> for InternalError {
72    fn from(_error: TimerError) -> Self {
73        Self::invariant()
74    }
75}
76
77/// Authority coordination over exact native timer owners.
78pub struct TimerAuthorityWorkflow;
79
80impl TimerAuthorityWorkflow {
81    /// Initialize only the shared timer runtime for a canister with no declared jobs yet.
82    pub(crate) fn initialize_shared_runtime() -> Result<(), TimerError> {
83        initialize_runtime()?;
84        Ok(())
85    }
86
87    /// Initialize the shared runtime; non-root claims remain lazy and domain-owned.
88    pub(crate) fn initialize_nonroot_runtime() -> Result<(), TimerError> {
89        Self::initialize_shared_runtime()
90    }
91
92    /// Initialize the shared runtime; the control plane declares Root-only claims later.
93    pub(crate) fn initialize_root_runtime() -> Result<(), TimerError> {
94        Self::initialize_shared_runtime()
95    }
96
97    /// Restore volatile suspension from the durable authority fence.
98    pub(crate) fn restore_snapshot_suspension(sealed: bool) {
99        TIMERS_SUSPENDED.with(|suspended| suspended.set(sealed));
100    }
101
102    /// Return whether durable lifecycle restoration left timer owners suspended.
103    #[must_use]
104    pub(crate) fn is_suspended() -> bool {
105        TIMERS_SUSPENDED.with(Cell::get)
106    }
107
108    /// Arm the one non-root watchdog after an exact domain owner reconstructs demand.
109    pub(crate) fn ensure_async_job_recovery_watchdog() -> Result<(), TimerError> {
110        require_active()?;
111        if EnvOps::is_root() {
112            return Ok(());
113        }
114        reconcile_core_recovery_watchdog(
115            WatchdogReconcileState::Scheduled,
116            Self::recover_expired_async_jobs,
117        )
118    }
119
120    /// Arm the same watchdog with the active role's automatic top-up recovery owner.
121    pub(crate) fn ensure_async_job_recovery_watchdog_with_automatic_topup() -> Result<(), TimerError>
122    {
123        require_active()?;
124        reconcile_core_recovery_watchdog(
125            WatchdogReconcileState::Scheduled,
126            Self::recover_expired_async_jobs_with_automatic_topup,
127        )
128    }
129
130    /// Prove that every Root-owned timer and business attempt can be suspended.
131    pub(crate) fn require_root_resumable() -> Result<(), TimerError> {
132        require_no_active_async_job_attempts()?;
133        let mut identities = BTreeSet::from([
134            canister_pool_timer_identity()?,
135            recovery_watchdog_identity()?,
136        ]);
137        for identity in [
138            runtime::auth::RuntimeAuthWorkflow::claimed_root_issuer_renewal_timer_identity()?,
139            runtime::intent::IntentCleanupWorkflow::claimed_timer_identity()?,
140            runtime::log::LogRetentionWorkflow::claimed_timer_identity()?,
141            runtime::cycles::CycleWorkflow::claimed_timer_identity()?,
142            PlacementAcknowledgementWorkflow::claimed_timer_identity()?,
143            claimed_core_recovery_watchdog_identity()?,
144        ]
145        .into_iter()
146        .flatten()
147        {
148            identities.insert(identity);
149        }
150        require_observed_claims_resumable(
151            &identities,
152            timer_inventory()?
153                .into_timers()
154                .into_iter()
155                .map(|snapshot| (snapshot.identity().clone(), snapshot.registration_status())),
156        )
157    }
158
159    /// Prove that Coordinator has no live private lifecycle work to snapshot.
160    pub(crate) fn require_coordinator_resumable() -> Result<(), TimerError> {
161        require_observed_claims_resumable(
162            &BTreeSet::new(),
163            timer_inventory()?
164                .into_timers()
165                .into_iter()
166                .map(|snapshot| (snapshot.identity().clone(), snapshot.registration_status())),
167        )
168    }
169
170    /// Disarm exact Root core-owned claims without affecting another provider owner.
171    pub(crate) fn suspend_root() -> Result<(), TimerError> {
172        Self::require_root_resumable()?;
173        TIMERS_SUSPENDED.with(|suspended| suspended.set(true));
174
175        runtime::auth::RuntimeAuthWorkflow::cancel_root_issuer_renewal_timer()?;
176        runtime::intent::IntentCleanupWorkflow::cancel_timer()?;
177        runtime::log::LogRetentionWorkflow::cancel_timer()?;
178        runtime::cycles::CycleWorkflow::cancel_timer()?;
179        PlacementAcknowledgementWorkflow::cancel_timer()?;
180        cancel_core_recovery_watchdog()?;
181        Ok(())
182    }
183
184    /// Seal a Coordinator only when its native inventory is empty.
185    pub(crate) fn suspend_coordinator() -> Result<(), TimerError> {
186        Self::require_coordinator_resumable()?;
187        TIMERS_SUSPENDED.with(|suspended| suspended.set(true));
188        Ok(())
189    }
190
191    /// End Root suspension before exact domain owners reconstruct current demand.
192    pub(crate) fn resume_root() {
193        TIMERS_SUSPENDED.with(|suspended| suspended.set(false));
194    }
195
196    /// End Coordinator suspension; it owns no fixed background claims.
197    pub(crate) fn resume_coordinator() {
198        TIMERS_SUSPENDED.with(|suspended| suspended.set(false));
199    }
200
201    /// Schedule one private lifecycle deferral as a direct remove-on-stop native claim.
202    pub(crate) fn defer_lifecycle_once(
203        delay: Duration,
204        label: impl Into<String>,
205        task: impl Future<Output = ()> + 'static,
206    ) -> Result<(), TimerError> {
207        register_lifecycle_once(delay, label.into(), async move {
208            task.await;
209            TimerRunResult::new(TimerCompletion::success(1), TimerDirective::Stop)
210        })
211    }
212
213    /// Schedule lifecycle work whose typed completion must remain observable.
214    pub(crate) fn defer_lifecycle_result_once(
215        delay: Duration,
216        label: impl Into<String>,
217        task: impl Future<Output = TimerRunResult> + 'static,
218    ) -> Result<(), TimerError> {
219        register_lifecycle_once(delay, label.into(), task)
220    }
221
222    /// Recover expired core-owned business attempts for the one role-native watchdog.
223    pub(crate) fn recover_expired_async_jobs(now_ns: u64) -> u64 {
224        let mut recovered = 0u64;
225        if runtime::auth::RuntimeAuthWorkflow::recover_expired_root_issuer_renewal(now_ns) {
226            recovered = recovered.saturating_add(1);
227        }
228        if PlacementAcknowledgementWorkflow::recover_expired_timer(now_ns) {
229            recovered = recovered.saturating_add(1);
230        }
231        recovered
232    }
233
234    /// Recover the base set plus automatic top-up for a capability-bearing non-root.
235    pub(crate) fn recover_expired_async_jobs_with_automatic_topup(now_ns: u64) -> u64 {
236        let recovered = Self::recover_expired_async_jobs(now_ns);
237        if runtime::cycles::CycleWorkflow::recover_expired_timer(now_ns) {
238            return recovered.saturating_add(1);
239        }
240        recovered
241    }
242
243    /// Return the shared canonical timer inventory in deterministic identity order.
244    pub fn statuses() -> Result<Vec<TimerSnapshot>, TimerError> {
245        Ok(timer_inventory()?.into_timers())
246    }
247}
248
249fn require_no_active_async_job_attempts() -> Result<(), TimerError> {
250    let owners = [
251        (
252            AsyncJobOwner::AuthRenewal,
253            runtime::auth::RuntimeAuthWorkflow::root_issuer_renewal_timer_identity()?,
254        ),
255        (
256            AsyncJobOwner::PlacementReceiptAcknowledgement,
257            PlacementAcknowledgementWorkflow::timer_identity()?,
258        ),
259        (
260            AsyncJobOwner::CanisterPoolMaintenance,
261            canister_pool_timer_identity()?,
262        ),
263        (
264            AsyncJobOwner::CycleTopup,
265            runtime::cycles::CycleWorkflow::timer_identity()?,
266        ),
267    ];
268    for (owner, identity) in owners {
269        if AsyncJobRecoveryOps::active_lease_deadline(owner).is_some() {
270            return Err(TimerError::RunningClaim(format_identity(&identity)));
271        }
272    }
273    Ok(())
274}
275
276fn reconcile_core_recovery_watchdog(
277    desired: WatchdogReconcileState,
278    recover: fn(u64) -> u64,
279) -> Result<(), TimerError> {
280    let identity = recovery_watchdog_identity()?;
281    let cadence = TimerCadence::new(RECOVERY_WATCHDOG_CADENCE)?;
282    CORE_RECOVERY_WATCHDOG
283        .try_with(|registration| {
284            let mut registration = registration
285                .try_borrow_mut()
286                .map_err(|_| TimerError::CustodyBusy)?;
287            reconcile_watchdog(
288                &mut registration,
289                &identity,
290                cadence,
291                desired,
292                move |_context| run_core_recovery_watchdog(recover),
293            )
294            .map_err(TimerError::from)
295        })
296        .map_err(|_| TimerError::CustodyBusy)?
297}
298
299fn claimed_core_recovery_watchdog_identity() -> Result<Option<TimerIdentity>, TimerError> {
300    CORE_RECOVERY_WATCHDOG
301        .try_with(|registration| {
302            let registration = registration
303                .try_borrow()
304                .map_err(|_| TimerError::CustodyBusy)?;
305            Ok(registration
306                .as_ref()
307                .map(|registration| registration.identity().clone()))
308        })
309        .map_err(|_| TimerError::CustodyBusy)?
310}
311
312fn cancel_core_recovery_watchdog() -> Result<(), TimerError> {
313    CORE_RECOVERY_WATCHDOG
314        .try_with(|registration| {
315            let registration = registration
316                .try_borrow()
317                .map_err(|_| TimerError::CustodyBusy)?;
318            if let Some(registration) = registration.as_ref() {
319                registration.cancel()?;
320            }
321            Ok(())
322        })
323        .map_err(|_| TimerError::CustodyBusy)?
324}
325
326fn run_core_recovery_watchdog(recover: fn(u64) -> u64) -> WatchdogRunResult {
327    let recovered = recover(IcOps::now_nanos());
328    let completion = if recovered == 0 {
329        TimerCompletion::no_work()
330    } else {
331        TimerCompletion::success(recovered)
332    };
333    WatchdogRunResult::new(completion, ic_timers::WatchdogDecision::Continue)
334}
335
336pub fn recovery_watchdog_identity() -> Result<TimerIdentity, TimerError> {
337    TimerIdentity::try_new("canic", "async_job_recovery", "watchdog").map_err(Into::into)
338}
339
340fn canister_pool_timer_identity() -> Result<TimerIdentity, TimerError> {
341    TimerIdentity::try_new("canic", "canister_pool", "maintain").map_err(Into::into)
342}
343
344fn register_lifecycle_once(
345    delay: Duration,
346    label: String,
347    task: impl Future<Output = TimerRunResult> + 'static,
348) -> Result<(), TimerError> {
349    require_active()?;
350    let identity = next_lifecycle_identity(label)?;
351    let mut task = Some(task);
352    let registration = register_once(
353        identity,
354        DeclarationLifetime::RemoveWhenStopped,
355        move |_context: OnceContext| {
356            let task = task.take();
357            async move {
358                match task {
359                    Some(task) => task.await,
360                    None => TimerRunResult::new(
361                        TimerCompletion::invariant_failure(0),
362                        TimerDirective::Stop,
363                    ),
364                }
365            }
366        },
367    )?;
368    if let Err(primary) = registration.ensure_scheduled(TimerSchedule::After(delay)) {
369        return match registration.unregister() {
370            Ok(()) => Err(primary.into()),
371            Err(cleanup) => Err(TimerError::RegistrationRollback {
372                primary: Box::new(primary.into()),
373                cleanup: Box::new(cleanup.into()),
374            }),
375        };
376    }
377    drop(registration);
378    Ok(())
379}
380
381fn next_lifecycle_identity(label: String) -> Result<TimerIdentity, TimerError> {
382    let id = NEXT_LIFECYCLE_ID.with(|next| {
383        let id = next
384            .get()
385            .checked_add(1)
386            .ok_or(TimerError::LifecycleIdentityExhausted)?;
387        next.set(id);
388        Ok::<_, TimerError>(id)
389    })?;
390    TimerIdentity::try_new("canic", format!("lifecycle-{id}"), label).map_err(Into::into)
391}
392
393pub fn require_active() -> Result<(), TimerError> {
394    if TIMERS_SUSPENDED.with(Cell::get) {
395        return Err(TimerError::Suspended);
396    }
397    Ok(())
398}
399
400/// Borrow one exact domain owner's native once registration without moving custody.
401pub fn with_owned_once<T>(
402    owner: &'static LocalKey<RefCell<Option<OnceRegistration>>>,
403    operation: impl FnOnce(&OnceRegistration) -> T,
404) -> Result<Option<T>, TimerError> {
405    owner
406        .try_with(|registration| {
407            let registration = registration
408                .try_borrow()
409                .map_err(|_| TimerError::CustodyBusy)?;
410            Ok::<_, TimerError>(registration.as_ref().map(operation))
411        })
412        .map_err(|_| TimerError::CustodyBusy)?
413}
414
415/// Retain one exact domain owner's native once registration with rollback on rejection.
416pub fn retain_owned_once(
417    owner: &'static LocalKey<RefCell<Option<OnceRegistration>>>,
418    registration: OnceRegistration,
419) -> Result<(), TimerError> {
420    retain_with_rollback(
421        registration,
422        |registration| {
423            owner.with(|current| {
424                let Ok(mut current) = current.try_borrow_mut() else {
425                    return Err((TimerError::CustodyBusy, registration));
426                };
427                if current.is_some() {
428                    return Err((TimerError::WrongPolicy, registration));
429                }
430                *current = Some(registration);
431                Ok(())
432            })
433        },
434        |registration| registration.unregister().map_err(TimerError::from),
435    )
436}
437
438fn retain_with_rollback<T>(
439    claim: T,
440    retain: impl FnOnce(T) -> Result<(), (TimerError, T)>,
441    cleanup: impl FnOnce(T) -> Result<(), TimerError>,
442) -> Result<(), TimerError> {
443    match retain(claim) {
444        Ok(()) => Ok(()),
445        Err((primary, claim)) => match cleanup(claim) {
446            Ok(()) => Err(primary),
447            Err(cleanup) => Err(TimerError::RegistrationRollback {
448                primary: Box::new(primary),
449                cleanup: Box::new(cleanup),
450            }),
451        },
452    }
453}
454
455fn format_identity(identity: &TimerIdentity) -> String {
456    format!(
457        "{}/{}/{}",
458        identity.owner(),
459        identity.subsystem(),
460        identity.name()
461    )
462}
463
464fn require_observed_claims_resumable(
465    claimed: &BTreeSet<TimerIdentity>,
466    observed: impl IntoIterator<Item = (TimerIdentity, TimerRegistrationStatus)>,
467) -> Result<(), TimerError> {
468    for (identity, registration) in observed {
469        if !claimed.contains(&identity) {
470            return Err(TimerError::UnmanagedClaim(format_identity(&identity)));
471        }
472        if registration == TimerRegistrationStatus::Running {
473            return Err(TimerError::RunningClaim(format_identity(&identity)));
474        }
475    }
476    Ok(())
477}
478
479#[cfg(test)]
480mod tests {
481    use super::*;
482    use crate::ops::storage::async_job_recovery::AsyncJobClaim;
483    use std::{cell::Cell, collections::BTreeSet};
484
485    #[test]
486    fn fixed_claim_identities_are_exact_and_unique() {
487        let identities = [
488            runtime::intent::IntentCleanupWorkflow::timer_identity()
489                .expect("intent cleanup identity"),
490            runtime::log::LogRetentionWorkflow::timer_identity().expect("log retention identity"),
491            runtime::auth::RuntimeAuthWorkflow::root_issuer_renewal_timer_identity()
492                .expect("auth renewal identity"),
493            runtime::cycles::CycleWorkflow::timer_identity().expect("cycle top-up identity"),
494            PlacementAcknowledgementWorkflow::timer_identity()
495                .expect("placement acknowledgement identity"),
496            recovery_watchdog_identity().expect("recovery watchdog identity"),
497            canister_pool_timer_identity().expect("canister pool identity"),
498        ]
499        .into_iter()
500        .collect::<BTreeSet<_>>();
501
502        assert_eq!(identities.len(), 7);
503        assert!(
504            identities
505                .iter()
506                .all(|identity| identity.owner() == "canic")
507        );
508    }
509
510    #[test]
511    fn failed_custody_insertion_runs_registration_cleanup() {
512        let cleaned = Cell::new(false);
513        let error = retain_with_rollback(
514            17u8,
515            |claim| Err((TimerError::CustodyBusy, claim)),
516            |claim| {
517                assert_eq!(claim, 17);
518                cleaned.set(true);
519                Ok(())
520            },
521        )
522        .expect_err("custody rejection must propagate");
523
524        assert!(matches!(error, TimerError::CustodyBusy));
525        assert!(cleaned.get());
526    }
527
528    #[test]
529    fn authority_snapshot_rejects_a_claim_outside_canic_custody() {
530        let external = TimerIdentity::try_new("companion-framework", "snapshot", "unmanaged")
531            .expect("external identity");
532        assert!(matches!(
533            require_observed_claims_resumable(
534                &BTreeSet::new(),
535                [(external, TimerRegistrationStatus::Unregistered)]
536            ),
537            Err(TimerError::UnmanagedClaim(identity))
538                if identity == "companion-framework/snapshot/unmanaged"
539        ));
540    }
541
542    #[test]
543    fn authority_snapshot_rejects_a_running_canic_claim() {
544        let identity =
545            TimerIdentity::try_new("canic", "cycles", "topup").expect("Canic timer identity");
546        assert!(matches!(
547            require_observed_claims_resumable(
548                &BTreeSet::from([identity.clone()]),
549                [(identity, TimerRegistrationStatus::Running)]
550            ),
551            Err(TimerError::RunningClaim(identity)) if identity == "canic/cycles/topup"
552        ));
553    }
554
555    #[test]
556    fn authority_snapshot_rejects_a_watchdog_dispatched_async_job_attempt() {
557        let owner = AsyncJobOwner::CanisterPoolMaintenance;
558        AsyncJobRecoveryOps::abandon(owner);
559        assert!(matches!(
560            AsyncJobRecoveryOps::claim(owner, 10, 20),
561            Ok(AsyncJobClaim::Acquired(_))
562        ));
563
564        assert!(matches!(
565            require_no_active_async_job_attempts(),
566            Err(TimerError::RunningClaim(identity))
567                if identity == "canic/canister_pool/maintain"
568        ));
569        AsyncJobRecoveryOps::abandon(owner);
570    }
571}
canic_core/workflow/runtime/timer/mod.rs

canic_core/workflow/runtime/timer/
mod.rs
