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

canic_core/workflow/runtime/timer/
mod.rs
