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, TimerReconcileState, TimerRegistrationStatus, TimerRunResult,
20    TimerSchedule, TimerSnapshot, 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            TimerReconcileState::Scheduled,
116            Self::recover_expired_async_jobs,
117        )
118    }
119
120    /// Arm the same watchdog with the compile-selected automatic top-up recovery owner.
121    pub(crate) fn ensure_async_job_recovery_watchdog_with_automatic_topup() -> Result<(), TimerError>
122    {
123        require_active()?;
124        if EnvOps::is_root() {
125            return Err(TimerError::WrongPolicy);
126        }
127        reconcile_core_recovery_watchdog(
128            TimerReconcileState::Scheduled,
129            Self::recover_expired_async_jobs_with_automatic_topup,
130        )
131    }
132
133    /// Prove that every Root-owned timer and business attempt can be suspended.
134    pub(crate) fn require_root_resumable() -> Result<(), TimerError> {
135        require_no_active_async_job_attempts()?;
136        let mut identities = BTreeSet::from([
137            canister_pool_timer_identity()?,
138            recovery_watchdog_identity()?,
139        ]);
140        for identity in [
141            runtime::auth::RuntimeAuthWorkflow::claimed_root_issuer_renewal_timer_identity()?,
142            runtime::intent::IntentCleanupWorkflow::claimed_timer_identity()?,
143            runtime::log::LogRetentionWorkflow::claimed_timer_identity()?,
144            PlacementAcknowledgementWorkflow::claimed_timer_identity()?,
145            claimed_core_recovery_watchdog_identity()?,
146        ]
147        .into_iter()
148        .flatten()
149        {
150            identities.insert(identity);
151        }
152        require_observed_claims_resumable(
153            &identities,
154            timer_inventory()?
155                .into_timers()
156                .into_iter()
157                .map(|snapshot| (snapshot.identity().clone(), snapshot.registration_status())),
158        )
159    }
160
161    /// Prove that Coordinator has no live private lifecycle work to snapshot.
162    pub(crate) fn require_coordinator_resumable() -> Result<(), TimerError> {
163        require_observed_claims_resumable(
164            &BTreeSet::new(),
165            timer_inventory()?
166                .into_timers()
167                .into_iter()
168                .map(|snapshot| (snapshot.identity().clone(), snapshot.registration_status())),
169        )
170    }
171
172    /// Disarm exact Root core-owned claims without affecting another provider owner.
173    pub(crate) fn suspend_root() -> Result<(), TimerError> {
174        Self::require_root_resumable()?;
175        TIMERS_SUSPENDED.with(|suspended| suspended.set(true));
176
177        runtime::auth::RuntimeAuthWorkflow::cancel_root_issuer_renewal_timer()?;
178        runtime::intent::IntentCleanupWorkflow::cancel_timer()?;
179        runtime::log::LogRetentionWorkflow::cancel_timer()?;
180        PlacementAcknowledgementWorkflow::cancel_timer()?;
181        cancel_core_recovery_watchdog()?;
182        Ok(())
183    }
184
185    /// Seal a Coordinator only when its native inventory is empty.
186    pub(crate) fn suspend_coordinator() -> Result<(), TimerError> {
187        Self::require_coordinator_resumable()?;
188        TIMERS_SUSPENDED.with(|suspended| suspended.set(true));
189        Ok(())
190    }
191
192    /// End Root suspension before exact domain owners reconstruct current demand.
193    pub(crate) fn resume_root() {
194        TIMERS_SUSPENDED.with(|suspended| suspended.set(false));
195    }
196
197    /// End Coordinator suspension; it owns no fixed background claims.
198    pub(crate) fn resume_coordinator() {
199        TIMERS_SUSPENDED.with(|suspended| suspended.set(false));
200    }
201
202    /// Schedule one private lifecycle deferral as a direct remove-on-stop native claim.
203    pub(crate) fn defer_lifecycle_once(
204        delay: Duration,
205        label: impl Into<String>,
206        task: impl Future<Output = ()> + 'static,
207    ) -> Result<(), TimerError> {
208        register_lifecycle_once(delay, label.into(), async move {
209            task.await;
210            TimerRunResult::new(TimerCompletion::success(1), TimerDirective::Stop)
211        })
212    }
213
214    /// Schedule lifecycle work whose typed completion must remain observable.
215    pub(crate) fn defer_lifecycle_result_once(
216        delay: Duration,
217        label: impl Into<String>,
218        task: impl Future<Output = TimerRunResult> + 'static,
219    ) -> Result<(), TimerError> {
220        register_lifecycle_once(delay, label.into(), task)
221    }
222
223    /// Recover expired core-owned business attempts for the one role-native watchdog.
224    pub(crate) fn recover_expired_async_jobs(now_ns: u64) -> u64 {
225        let mut recovered = 0u64;
226        if runtime::auth::RuntimeAuthWorkflow::recover_expired_root_issuer_renewal(now_ns) {
227            recovered = recovered.saturating_add(1);
228        }
229        if PlacementAcknowledgementWorkflow::recover_expired_timer(now_ns) {
230            recovered = recovered.saturating_add(1);
231        }
232        recovered
233    }
234
235    /// Recover the base set plus automatic top-up for a capability-bearing non-root.
236    pub(crate) fn recover_expired_async_jobs_with_automatic_topup(now_ns: u64) -> u64 {
237        let recovered = Self::recover_expired_async_jobs(now_ns);
238        if runtime::cycles::CycleWorkflow::recover_expired_timer(now_ns) {
239            return recovered.saturating_add(1);
240        }
241        recovered
242    }
243
244    /// Return the shared canonical timer inventory in deterministic identity order.
245    pub fn statuses() -> Result<Vec<TimerSnapshot>, TimerError> {
246        Ok(timer_inventory()?.into_timers())
247    }
248}
249
250fn require_no_active_async_job_attempts() -> Result<(), TimerError> {
251    let owners = [
252        (
253            AsyncJobOwner::AuthRenewal,
254            runtime::auth::RuntimeAuthWorkflow::root_issuer_renewal_timer_identity()?,
255        ),
256        (
257            AsyncJobOwner::PlacementReceiptAcknowledgement,
258            PlacementAcknowledgementWorkflow::timer_identity()?,
259        ),
260        (
261            AsyncJobOwner::CanisterPoolMaintenance,
262            canister_pool_timer_identity()?,
263        ),
264    ];
265    for (owner, identity) in owners {
266        if AsyncJobRecoveryOps::active_lease_deadline(owner).is_some() {
267            return Err(TimerError::RunningClaim(format_identity(&identity)));
268        }
269    }
270    Ok(())
271}
272
273fn reconcile_core_recovery_watchdog(
274    desired: TimerReconcileState,
275    recover: fn(u64) -> u64,
276) -> Result<(), TimerError> {
277    let identity = recovery_watchdog_identity()?;
278    let cadence = TimerCadence::new(RECOVERY_WATCHDOG_CADENCE)?;
279    CORE_RECOVERY_WATCHDOG
280        .try_with(|registration| {
281            let mut registration = registration
282                .try_borrow_mut()
283                .map_err(|_| TimerError::CustodyBusy)?;
284            reconcile_watchdog(
285                &mut registration,
286                &identity,
287                cadence,
288                desired,
289                move |_context| run_core_recovery_watchdog(recover),
290            )
291            .map_err(TimerError::from)
292        })
293        .map_err(|_| TimerError::CustodyBusy)?
294}
295
296fn claimed_core_recovery_watchdog_identity() -> Result<Option<TimerIdentity>, TimerError> {
297    CORE_RECOVERY_WATCHDOG
298        .try_with(|registration| {
299            let registration = registration
300                .try_borrow()
301                .map_err(|_| TimerError::CustodyBusy)?;
302            Ok(registration
303                .as_ref()
304                .map(|registration| registration.identity().clone()))
305        })
306        .map_err(|_| TimerError::CustodyBusy)?
307}
308
309fn cancel_core_recovery_watchdog() -> Result<(), TimerError> {
310    CORE_RECOVERY_WATCHDOG
311        .try_with(|registration| {
312            let registration = registration
313                .try_borrow()
314                .map_err(|_| TimerError::CustodyBusy)?;
315            if let Some(registration) = registration.as_ref() {
316                registration.cancel()?;
317            }
318            Ok(())
319        })
320        .map_err(|_| TimerError::CustodyBusy)?
321}
322
323fn run_core_recovery_watchdog(recover: fn(u64) -> u64) -> WatchdogRunResult {
324    let recovered = recover(IcOps::now_nanos());
325    let completion = if recovered == 0 {
326        TimerCompletion::no_work()
327    } else {
328        TimerCompletion::success(recovered)
329    };
330    WatchdogRunResult::new(completion, ic_timers::WatchdogDecision::Continue)
331}
332
333pub fn recovery_watchdog_identity() -> Result<TimerIdentity, TimerError> {
334    TimerIdentity::try_new("canic", "async_job_recovery", "watchdog").map_err(Into::into)
335}
336
337fn canister_pool_timer_identity() -> Result<TimerIdentity, TimerError> {
338    TimerIdentity::try_new("canic", "canister_pool", "maintain").map_err(Into::into)
339}
340
341fn register_lifecycle_once(
342    delay: Duration,
343    label: String,
344    task: impl Future<Output = TimerRunResult> + 'static,
345) -> Result<(), TimerError> {
346    require_active()?;
347    let identity = next_lifecycle_identity(label)?;
348    let mut task = Some(task);
349    let registration = register_once(
350        identity,
351        DeclarationLifetime::RemoveWhenStopped,
352        move |_context: OnceContext| {
353            let task = task.take();
354            async move {
355                match task {
356                    Some(task) => task.await,
357                    None => TimerRunResult::new(
358                        TimerCompletion::invariant_failure(0),
359                        TimerDirective::Stop,
360                    ),
361                }
362            }
363        },
364    )?;
365    if let Err(primary) = registration.ensure_scheduled(TimerSchedule::After(delay)) {
366        return match registration.unregister() {
367            Ok(()) => Err(primary.into()),
368            Err(cleanup) => Err(TimerError::RegistrationRollback {
369                primary: Box::new(primary.into()),
370                cleanup: Box::new(cleanup.into()),
371            }),
372        };
373    }
374    drop(registration);
375    Ok(())
376}
377
378fn next_lifecycle_identity(label: String) -> Result<TimerIdentity, TimerError> {
379    let id = NEXT_LIFECYCLE_ID.with(|next| {
380        let id = next
381            .get()
382            .checked_add(1)
383            .ok_or(TimerError::LifecycleIdentityExhausted)?;
384        next.set(id);
385        Ok::<_, TimerError>(id)
386    })?;
387    TimerIdentity::try_new("canic", format!("lifecycle-{id}"), label).map_err(Into::into)
388}
389
390pub fn require_active() -> Result<(), TimerError> {
391    if TIMERS_SUSPENDED.with(Cell::get) {
392        return Err(TimerError::Suspended);
393    }
394    Ok(())
395}
396
397/// Borrow one exact domain owner's native once registration without moving custody.
398pub fn with_owned_once<T>(
399    owner: &'static LocalKey<RefCell<Option<OnceRegistration>>>,
400    operation: impl FnOnce(&OnceRegistration) -> T,
401) -> Result<Option<T>, TimerError> {
402    owner
403        .try_with(|registration| {
404            let registration = registration
405                .try_borrow()
406                .map_err(|_| TimerError::CustodyBusy)?;
407            Ok::<_, TimerError>(registration.as_ref().map(operation))
408        })
409        .map_err(|_| TimerError::CustodyBusy)?
410}
411
412/// Retain one exact domain owner's native once registration with rollback on rejection.
413pub fn retain_owned_once(
414    owner: &'static LocalKey<RefCell<Option<OnceRegistration>>>,
415    registration: OnceRegistration,
416) -> Result<(), TimerError> {
417    retain_with_rollback(
418        registration,
419        |registration| {
420            owner.with(|current| {
421                let Ok(mut current) = current.try_borrow_mut() else {
422                    return Err((TimerError::CustodyBusy, registration));
423                };
424                if current.is_some() {
425                    return Err((TimerError::WrongPolicy, registration));
426                }
427                *current = Some(registration);
428                Ok(())
429            })
430        },
431        |registration| registration.unregister().map_err(TimerError::from),
432    )
433}
434
435fn retain_with_rollback<T>(
436    claim: T,
437    retain: impl FnOnce(T) -> Result<(), (TimerError, T)>,
438    cleanup: impl FnOnce(T) -> Result<(), TimerError>,
439) -> Result<(), TimerError> {
440    match retain(claim) {
441        Ok(()) => Ok(()),
442        Err((primary, claim)) => match cleanup(claim) {
443            Ok(()) => Err(primary),
444            Err(cleanup) => Err(TimerError::RegistrationRollback {
445                primary: Box::new(primary),
446                cleanup: Box::new(cleanup),
447            }),
448        },
449    }
450}
451
452fn format_identity(identity: &TimerIdentity) -> String {
453    format!(
454        "{}/{}/{}",
455        identity.owner(),
456        identity.subsystem(),
457        identity.name()
458    )
459}
460
461fn require_observed_claims_resumable(
462    claimed: &BTreeSet<TimerIdentity>,
463    observed: impl IntoIterator<Item = (TimerIdentity, TimerRegistrationStatus)>,
464) -> Result<(), TimerError> {
465    for (identity, registration) in observed {
466        if !claimed.contains(&identity) {
467            return Err(TimerError::UnmanagedClaim(format_identity(&identity)));
468        }
469        if registration == TimerRegistrationStatus::Running {
470            return Err(TimerError::RunningClaim(format_identity(&identity)));
471        }
472    }
473    Ok(())
474}
475
476#[cfg(test)]
477mod tests {
478    use super::*;
479    use crate::ops::storage::async_job_recovery::AsyncJobClaim;
480    use std::{cell::Cell, collections::BTreeSet};
481
482    #[test]
483    fn fixed_claim_identities_are_exact_and_unique() {
484        let identities = [
485            runtime::intent::IntentCleanupWorkflow::timer_identity()
486                .expect("intent cleanup identity"),
487            runtime::log::LogRetentionWorkflow::timer_identity().expect("log retention identity"),
488            runtime::auth::RuntimeAuthWorkflow::root_issuer_renewal_timer_identity()
489                .expect("auth renewal identity"),
490            runtime::cycles::CycleWorkflow::timer_identity().expect("cycle top-up identity"),
491            PlacementAcknowledgementWorkflow::timer_identity()
492                .expect("placement acknowledgement identity"),
493            recovery_watchdog_identity().expect("recovery watchdog identity"),
494            canister_pool_timer_identity().expect("canister pool identity"),
495        ]
496        .into_iter()
497        .collect::<BTreeSet<_>>();
498
499        assert_eq!(identities.len(), 7);
500        assert!(
501            identities
502                .iter()
503                .all(|identity| identity.owner() == "canic")
504        );
505    }
506
507    #[test]
508    fn failed_custody_insertion_runs_registration_cleanup() {
509        let cleaned = Cell::new(false);
510        let error = retain_with_rollback(
511            17u8,
512            |claim| Err((TimerError::CustodyBusy, claim)),
513            |claim| {
514                assert_eq!(claim, 17);
515                cleaned.set(true);
516                Ok(())
517            },
518        )
519        .expect_err("custody rejection must propagate");
520
521        assert!(matches!(error, TimerError::CustodyBusy));
522        assert!(cleaned.get());
523    }
524
525    #[test]
526    fn authority_snapshot_rejects_a_claim_outside_canic_custody() {
527        let external = TimerIdentity::try_new("companion-framework", "snapshot", "unmanaged")
528            .expect("external identity");
529        assert!(matches!(
530            require_observed_claims_resumable(
531                &BTreeSet::new(),
532                [(external, TimerRegistrationStatus::Unregistered)]
533            ),
534            Err(TimerError::UnmanagedClaim(identity))
535                if identity == "companion-framework/snapshot/unmanaged"
536        ));
537    }
538
539    #[test]
540    fn authority_snapshot_rejects_a_running_canic_claim() {
541        let identity =
542            TimerIdentity::try_new("canic", "cycles", "topup").expect("Canic timer identity");
543        assert!(matches!(
544            require_observed_claims_resumable(
545                &BTreeSet::from([identity.clone()]),
546                [(identity, TimerRegistrationStatus::Running)]
547            ),
548            Err(TimerError::RunningClaim(identity)) if identity == "canic/cycles/topup"
549        ));
550    }
551
552    #[test]
553    fn authority_snapshot_rejects_a_watchdog_dispatched_async_job_attempt() {
554        let owner = AsyncJobOwner::CanisterPoolMaintenance;
555        AsyncJobRecoveryOps::abandon(owner);
556        assert!(matches!(
557            AsyncJobRecoveryOps::claim(owner, 10, 20),
558            Ok(AsyncJobClaim::Acquired(_))
559        ));
560
561        assert!(matches!(
562            require_no_active_async_job_attempts(),
563            Err(TimerError::RunningClaim(identity))
564                if identity == "canic/canister_pool/maintain"
565        ));
566        AsyncJobRecoveryOps::abandon(owner);
567    }
568}
canic_core/workflow/runtime/timer/mod.rs

canic_core/workflow/runtime/timer/
mod.rs
