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

canic_core/workflow/runtime/timer/
mod.rs
