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

canic_core/workflow/runtime/timer/
mod.rs
