use crate::{
callback::{AfterCompletionRunResult, OnceRunResult, OrdinaryRunResult},
platform::{self, TimerHandle},
registry::{
CallbackRole, CallbackToken, OrdinaryCallback, ProviderHandle, ProviderHandles,
RegisterError, RegistrationClaim, RegistryEffect, RegistryError, RegistryTransition,
TimerRegistry, WatchdogCallback,
},
schedule::{OrdinaryDirective, ScheduleError, TimerCadence, TimerSchedule},
snapshot::{
DeclarationLifetime, MemoryPageExtent, TimerCompletion, TimerControlFailure, TimerEpoch,
TimerIdentity, TimerInventorySnapshot, TimerPolicy, TimerSnapshot, WatchdogRunResult,
},
};
use std::{cell::RefCell, future::Future, rc::Rc, time::Duration};
use thiserror::Error;
thread_local! {
static RUNTIME: RefCell<Option<TimerRegistry>> = const { RefCell::new(None) };
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TimerStart {
Retained {
initial: Option<TimerSchedule>,
},
RemoveWhenStopped {
initial: TimerSchedule,
},
}
impl TimerStart {
const fn into_parts(self) -> (DeclarationLifetime, Option<TimerSchedule>) {
match self {
Self::Retained { initial } => (DeclarationLifetime::Retained, initial),
Self::RemoveWhenStopped { initial } => {
(DeclarationLifetime::RemoveWhenStopped, Some(initial))
}
}
}
}
#[non_exhaustive]
#[derive(Debug, Error)]
pub enum TimerError {
#[error("timer runtime is not initialized")]
NotInitialized,
#[error("timer runtime is already borrowed")]
RuntimeBusy,
#[error(transparent)]
Register(#[from] RegisterError),
#[error(transparent)]
Schedule(#[from] ScheduleError),
#[error("timer registration is no longer authoritative")]
RegistrationExpired,
#[error("timer control failed: {0:?}")]
ControlFailure(TimerControlFailure),
#[error("timer runtime ownership invariant failed")]
OwnershipInvariant,
#[error("timer lifecycle reconciliation conflicts with the canonical declaration")]
ReconciliationConflict,
}
impl From<RegistryError> for TimerError {
fn from(value: RegistryError) -> Self {
match value {
RegistryError::UnknownRegistration
| RegistryError::StaleRegistration
| RegistryError::StaleCallback => Self::RegistrationExpired,
RegistryError::PolicyMismatch { .. } | RegistryError::ProviderHandleAlreadyOwned => {
Self::OwnershipInvariant
}
RegistryError::Schedule(error) => Self::Schedule(error),
}
}
}
pub fn initialize_runtime() -> Result<TimerEpoch, TimerError> {
RUNTIME.with(|runtime| {
let mut runtime = runtime
.try_borrow_mut()
.map_err(|_| TimerError::RuntimeBusy)?;
let registry = runtime.get_or_insert_with(|| {
TimerRegistry::new(TimerEpoch::new(
platform::canister_version(),
platform::time_ns(),
))
});
Ok(registry.epoch())
})
}
pub struct OnceContext {
token: CallbackToken,
}
impl OnceContext {
const fn new(token: CallbackToken) -> Self {
Self { token }
}
#[must_use]
pub const fn identity(&self) -> &TimerIdentity {
self.token.identity()
}
pub fn ensure_scheduled(&self, schedule: TimerSchedule) -> Result<(), TimerError> {
ensure_once_claim(self.token.claim(), Some(&self.token), schedule)
}
pub fn reconcile_schedule(&self, schedule: Option<TimerSchedule>) -> Result<(), TimerError> {
reconcile_ordinary_claim(self.token.claim(), Some(&self.token), schedule)
}
pub fn cancel(&self) -> Result<(), TimerError> {
cancel_claim(self.token.claim(), Some(&self.token))
}
}
pub struct AfterCompletionContext {
token: CallbackToken,
}
impl AfterCompletionContext {
const fn new(token: CallbackToken) -> Self {
Self { token }
}
#[must_use]
pub const fn identity(&self) -> &TimerIdentity {
self.token.identity()
}
pub fn ensure_scheduled(&self) -> Result<(), TimerError> {
ensure_recurring_claim(self.token.claim(), Some(&self.token))
}
pub fn reconcile_schedule(&self, schedule: Option<TimerSchedule>) -> Result<(), TimerError> {
reconcile_ordinary_claim(self.token.claim(), Some(&self.token), schedule)
}
pub fn cancel(&self) -> Result<(), TimerError> {
cancel_claim(self.token.claim(), Some(&self.token))
}
}
pub struct WatchdogContext {
token: CallbackToken,
}
impl WatchdogContext {
const fn new(token: CallbackToken) -> Self {
Self { token }
}
#[must_use]
pub const fn identity(&self) -> &TimerIdentity {
self.token.identity()
}
pub fn ensure_scheduled(&self) -> Result<(), TimerError> {
ensure_recurring_claim(self.token.claim(), Some(&self.token))
}
pub fn ensure_scheduled_immediately(&self) -> Result<(), TimerError> {
ensure_watchdog_immediately_claim(self.token.claim(), Some(&self.token))
}
pub fn reconcile_schedule(&self, schedule: Option<TimerSchedule>) -> Result<(), TimerError> {
reconcile_watchdog_claim(self.token.claim(), Some(&self.token), schedule)
}
pub fn cancel(&self) -> Result<(), TimerError> {
cancel_claim(self.token.claim(), Some(&self.token))
}
}
#[must_use = "retain the registration claim so the timer remains controllable"]
pub struct OnceRegistration {
claim: RegistrationClaim,
}
impl OnceRegistration {
#[must_use]
pub const fn identity(&self) -> &TimerIdentity {
self.claim.identity()
}
pub fn has_armed_wakeup(&self) -> Result<bool, TimerError> {
has_armed_wakeup_claim(&self.claim)
}
pub fn ensure_scheduled(&self, schedule: TimerSchedule) -> Result<(), TimerError> {
ensure_once_claim(&self.claim, None, schedule)
}
pub fn reconcile_schedule(&self, schedule: Option<TimerSchedule>) -> Result<(), TimerError> {
reconcile_ordinary_claim(&self.claim, None, schedule)
}
pub fn cancel(&self) -> Result<(), TimerError> {
cancel_claim(&self.claim, None)
}
pub fn unregister(self) -> Result<(), TimerError> {
unregister_claim(&self.claim)
}
}
#[must_use = "retain the registration claim so the timer remains controllable"]
pub struct AfterCompletionRegistration {
claim: RegistrationClaim,
}
#[must_use = "retain the registration claim so the timer remains controllable"]
pub struct WatchdogRegistration {
claim: RegistrationClaim,
}
trait RegistrationClaimOwner {
fn registration_claim(&self) -> &RegistrationClaim;
}
impl RegistrationClaimOwner for OnceRegistration {
fn registration_claim(&self) -> &RegistrationClaim {
&self.claim
}
}
impl RegistrationClaimOwner for AfterCompletionRegistration {
fn registration_claim(&self) -> &RegistrationClaim {
&self.claim
}
}
impl RegistrationClaimOwner for WatchdogRegistration {
fn registration_claim(&self) -> &RegistrationClaim {
&self.claim
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TimerReconcileState {
Inactive,
Scheduled,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum WatchdogReconcileState {
Inactive,
Scheduled,
ScheduledImmediately,
ScheduledAt(u64),
}
impl WatchdogRegistration {
#[must_use]
pub const fn identity(&self) -> &TimerIdentity {
self.claim.identity()
}
pub fn has_armed_wakeup(&self) -> Result<bool, TimerError> {
has_armed_wakeup_claim(&self.claim)
}
pub fn ensure_scheduled(&self) -> Result<(), TimerError> {
ensure_recurring_claim(&self.claim, None)
}
pub fn ensure_scheduled_immediately(&self) -> Result<(), TimerError> {
ensure_watchdog_immediately_claim(&self.claim, None)
}
pub fn reconcile_schedule(&self, schedule: Option<TimerSchedule>) -> Result<(), TimerError> {
reconcile_watchdog_claim(&self.claim, None, schedule)
}
pub fn cancel(&self) -> Result<(), TimerError> {
cancel_claim(&self.claim, None)
}
pub fn unregister(self) -> Result<(), TimerError> {
unregister_claim(&self.claim)
}
}
impl AfterCompletionRegistration {
#[must_use]
pub const fn identity(&self) -> &TimerIdentity {
self.claim.identity()
}
pub fn has_armed_wakeup(&self) -> Result<bool, TimerError> {
has_armed_wakeup_claim(&self.claim)
}
pub fn ensure_scheduled(&self) -> Result<(), TimerError> {
ensure_recurring_claim(&self.claim, None)
}
pub fn reconcile_schedule(&self, schedule: Option<TimerSchedule>) -> Result<(), TimerError> {
reconcile_ordinary_claim(&self.claim, None, schedule)
}
pub fn cancel(&self) -> Result<(), TimerError> {
cancel_claim(&self.claim, None)
}
pub fn unregister(self) -> Result<(), TimerError> {
unregister_claim(&self.claim)
}
}
pub fn register_once<F, Fut>(
identity: TimerIdentity,
start: TimerStart,
callback: F,
) -> Result<OnceRegistration, TimerError>
where
F: FnMut(OnceContext) -> Fut + 'static,
Fut: Future<Output = OnceRunResult> + 'static,
{
let callback = erase_ordinary_callback(callback, OnceContext::new, OrdinaryRunResult::from);
let claim = register_initial(
start,
|registry, lifetime| {
registry.register_once_with_callback(identity, lifetime, Rc::clone(&callback))
},
TimerRegistry::reconcile_ordinary,
);
drop(callback);
Ok(OnceRegistration { claim: claim? })
}
pub fn register_after_completion<F, Fut>(
identity: TimerIdentity,
cadence: TimerCadence,
start: TimerStart,
callback: F,
) -> Result<AfterCompletionRegistration, TimerError>
where
F: FnMut(AfterCompletionContext) -> Fut + 'static,
Fut: Future<Output = AfterCompletionRunResult> + 'static,
{
let callback = erase_ordinary_callback(
callback,
AfterCompletionContext::new,
OrdinaryRunResult::from,
);
let claim = register_initial(
start,
|registry, lifetime| {
registry.register_after_completion_with_callback(
identity,
cadence,
lifetime,
Rc::clone(&callback),
)
},
TimerRegistry::reconcile_ordinary,
);
drop(callback);
Ok(AfterCompletionRegistration { claim: claim? })
}
pub fn register_watchdog<F>(
identity: TimerIdentity,
cadence: TimerCadence,
start: TimerStart,
callback: F,
) -> Result<WatchdogRegistration, TimerError>
where
F: FnMut(WatchdogContext) -> WatchdogRunResult + 'static,
{
let mut callback = callback;
let callback: WatchdogCallback = Rc::new(RefCell::new(Box::new(move |token| {
callback(WatchdogContext::new(token))
})));
let claim = register_initial(
start,
|registry, lifetime| {
registry.register_watchdog_with_callback(
identity,
cadence,
lifetime,
Rc::clone(&callback),
)
},
TimerRegistry::reconcile_watchdog_schedule,
);
drop(callback);
Ok(WatchdogRegistration { claim: claim? })
}
fn register_initial(
start: TimerStart,
register: impl FnOnce(
&mut TimerRegistry,
DeclarationLifetime,
) -> Result<RegistrationClaim, RegisterError>,
reconcile: fn(
&mut TimerRegistry,
&RegistrationClaim,
u64,
Option<TimerSchedule>,
) -> Result<RegistryTransition, RegistryError>,
) -> Result<RegistrationClaim, TimerError> {
let (lifetime, initial) = start.into_parts();
let (claim, requested) = with_registry_mut(|registry| {
let requested = match initial {
Some(schedule) => {
let now_ns = platform::time_ns();
schedule.resolve(now_ns)?;
Some((schedule, now_ns))
}
None => None,
};
Ok((register(registry, lifetime)?, requested))
})?;
if let Some((schedule, now_ns)) = requested
&& let Err(error) = apply_claim_transition(&claim, None, |registry| {
reconcile(registry, &claim, now_ns, Some(schedule))
})
{
return Err(cleanup_creation_error(&claim, error));
}
Ok(claim)
}
fn cleanup_creation_error(claim: &RegistrationClaim, error: TimerError) -> TimerError {
match unregister_claim(claim) {
Ok(()) | Err(TimerError::RegistrationExpired) => error,
Err(cleanup_error) => cleanup_error,
}
}
pub fn reconcile_once<F, Fut>(
registration: &mut Option<OnceRegistration>,
identity: &TimerIdentity,
desired: Option<TimerSchedule>,
callback: F,
) -> Result<(), TimerError>
where
F: FnMut(OnceContext) -> Fut + 'static,
Fut: Future<Output = OnceRunResult> + 'static,
{
reconcile_registration(
registration,
identity,
TimerPolicy::Once,
|| {
register_once(
identity.clone(),
TimerStart::Retained { initial: None },
callback,
)
},
|registration| registration.reconcile_schedule(desired),
)
}
pub fn reconcile_after_completion<F, Fut>(
registration: &mut Option<AfterCompletionRegistration>,
identity: &TimerIdentity,
cadence: TimerCadence,
desired: TimerReconcileState,
callback: F,
) -> Result<(), TimerError>
where
F: FnMut(AfterCompletionContext) -> Fut + 'static,
Fut: Future<Output = AfterCompletionRunResult> + 'static,
{
reconcile_registration(
registration,
identity,
TimerPolicy::AfterCompletion { cadence },
|| {
register_after_completion(
identity.clone(),
cadence,
TimerStart::Retained { initial: None },
callback,
)
},
|registration| match desired {
TimerReconcileState::Inactive => registration.cancel(),
TimerReconcileState::Scheduled => registration.ensure_scheduled(),
},
)
}
pub fn reconcile_watchdog<F>(
registration: &mut Option<WatchdogRegistration>,
identity: &TimerIdentity,
cadence: TimerCadence,
desired: WatchdogReconcileState,
callback: F,
) -> Result<(), TimerError>
where
F: FnMut(WatchdogContext) -> WatchdogRunResult + 'static,
{
reconcile_registration(
registration,
identity,
TimerPolicy::Watchdog { cadence },
|| {
register_watchdog(
identity.clone(),
cadence,
TimerStart::Retained { initial: None },
callback,
)
},
|registration| match desired {
WatchdogReconcileState::Inactive => registration.cancel(),
WatchdogReconcileState::Scheduled => registration.ensure_scheduled(),
WatchdogReconcileState::ScheduledImmediately => {
registration.ensure_scheduled_immediately()
}
WatchdogReconcileState::ScheduledAt(deadline_ns) => {
registration.reconcile_schedule(Some(TimerSchedule::At(deadline_ns)))
}
},
)
}
fn reconcile_registration<Registration>(
registration: &mut Option<Registration>,
identity: &TimerIdentity,
policy: TimerPolicy,
register: impl FnOnce() -> Result<Registration, TimerError>,
reconcile: impl FnOnce(&Registration) -> Result<(), TimerError>,
) -> Result<(), TimerError>
where
Registration: RegistrationClaimOwner,
{
if registration.is_none() {
let created = register()?;
if let Err(error) = reconcile(&created) {
return Err(cleanup_creation_error(created.registration_claim(), error));
}
*registration = Some(created);
return Ok(());
}
let registration = registration
.as_ref()
.ok_or(TimerError::ReconciliationConflict)?;
let claim = registration.registration_claim();
if claim.identity() != identity {
return Err(TimerError::ReconciliationConflict);
}
with_registry(|registry| {
registry
.declaration_matches(claim, policy, DeclarationLifetime::Retained)
.map_err(TimerError::from)?
.then_some(())
.ok_or(TimerError::ReconciliationConflict)
})?;
drop(register);
reconcile(registration)
}
pub fn timer_snapshot(identity: &TimerIdentity) -> Result<Option<TimerSnapshot>, TimerError> {
with_registry(|registry| Ok(registry.snapshot(identity)))
}
pub fn timer_inventory() -> Result<TimerInventorySnapshot, TimerError> {
with_registry(|registry| Ok(registry.inventory()))
}
pub fn consecutive_expected_failures(identity: &TimerIdentity) -> Result<Option<u64>, TimerError> {
with_registry(|registry| Ok(registry.consecutive_expected_failures(identity)))
}
fn has_armed_wakeup_claim(claim: &RegistrationClaim) -> Result<bool, TimerError> {
with_registry(|registry| registry.has_armed_wakeup(claim).map_err(TimerError::from))
}
fn erase_ordinary_callback<Context: 'static, Output: 'static, F, Fut>(
mut callback: F,
context: fn(CallbackToken) -> Context,
erase_result: fn(Output) -> OrdinaryRunResult,
) -> OrdinaryCallback
where
F: FnMut(Context) -> Fut + 'static,
Fut: Future<Output = Output> + 'static,
{
Rc::new(RefCell::new(Box::new(move |token| {
let future = callback(context(token));
Box::pin(async move { erase_result(future.await) })
})))
}
fn apply_claim_transition(
claim: &RegistrationClaim,
context: Option<&CallbackToken>,
operation: impl FnOnce(&mut TimerRegistry) -> Result<RegistryTransition, RegistryError>,
) -> Result<(), TimerError> {
let (transition, handles) = with_registry_mut(|registry| {
if let Some(token) = context {
registry
.validate_running_context(token)
.map_err(TimerError::from)?;
}
registry
.transition_claim(claim, operation)
.map_err(TimerError::from)
})?;
match finish_transition(transition, handles) {
result @ (Ok(()) | Err(TimerError::ControlFailure(_))) => result,
Err(error) => retire_failed_claim(claim, error),
}
}
fn ensure_once_claim(
claim: &RegistrationClaim,
context: Option<&CallbackToken>,
schedule: TimerSchedule,
) -> Result<(), TimerError> {
apply_claim_transition(claim, context, |registry| {
registry.ensure_once(claim, platform::time_ns(), schedule)
})
}
fn reconcile_ordinary_claim(
claim: &RegistrationClaim,
context: Option<&CallbackToken>,
schedule: Option<TimerSchedule>,
) -> Result<(), TimerError> {
apply_claim_transition(claim, context, |registry| {
registry.reconcile_ordinary(claim, platform::time_ns(), schedule)
})
}
fn ensure_recurring_claim(
claim: &RegistrationClaim,
context: Option<&CallbackToken>,
) -> Result<(), TimerError> {
apply_claim_transition(claim, context, |registry| {
registry.ensure_recurring(claim, platform::time_ns())
})
}
fn ensure_watchdog_immediately_claim(
claim: &RegistrationClaim,
context: Option<&CallbackToken>,
) -> Result<(), TimerError> {
apply_claim_transition(claim, context, |registry| {
registry.ensure_watchdog_immediately(claim, platform::time_ns())
})
}
fn reconcile_watchdog_claim(
claim: &RegistrationClaim,
context: Option<&CallbackToken>,
schedule: Option<TimerSchedule>,
) -> Result<(), TimerError> {
apply_claim_transition(claim, context, |registry| {
registry.reconcile_watchdog_schedule(claim, platform::time_ns(), schedule)
})
}
fn cancel_claim(
claim: &RegistrationClaim,
context: Option<&CallbackToken>,
) -> Result<(), TimerError> {
apply_claim_transition(claim, context, |registry| registry.cancel(claim))
}
fn unregister_claim(claim: &RegistrationClaim) -> Result<(), TimerError> {
apply_claim_transition(claim, None, |registry| registry.unregister(claim))
}
fn retire_failed_claim(claim: &RegistrationClaim, error: TimerError) -> Result<(), TimerError> {
match fail_claim_provider_binding(claim) {
Ok(()) | Err(TimerError::RegistrationExpired) => Err(error),
Err(cleanup_error) => Err(cleanup_error),
}
}
fn finish_transition(
transition: RegistryTransition,
handles: ProviderHandles,
) -> Result<(), TimerError> {
let failure = transition.failure();
let applied = apply_effect(transition.effect(), handles);
drop(transition);
applied?;
failure.map_or(Ok(()), |failure| Err(TimerError::ControlFailure(failure)))
}
fn apply_effect(effect: &RegistryEffect, mut handles: ProviderHandles) -> Result<(), TimerError> {
if !effect.has_valid_shape() {
clear_provider_handles(handles);
return Err(TimerError::OwnershipInvariant);
}
match effect {
RegistryEffect::None => restore_provider_handles(handles),
RegistryEffect::ArmWakeup {
token,
delay_ns,
arm,
} => {
if arm.replaces_existing() {
let replaced = take_detached_or_owned_handle(handles.take_wakeup(), |registry| {
registry.take_wakeup_handle(token.identity())
})?;
if let Some(replaced) = replaced {
clear_provider_handle(replaced);
}
}
restore_provider_handles(handles)?;
let handle = platform::set_timer(
Duration::from_nanos(*delay_ns),
dispatch_wakeup(token.clone()),
);
bind_provider_handle(token, handle)?;
if let Err(error) = confirm_effect(effect) {
let handle = with_registry_mut(|registry| {
registry
.take_wakeup_handle(token.identity())
.ok_or(TimerError::OwnershipInvariant)
})?;
clear_provider_handle(handle);
return Err(error);
}
Ok(())
}
RegistryEffect::ClearCallbacks {
identity,
handles: selected,
} => {
if selected.includes_wakeup() {
let wakeup = take_detached_or_owned_handle(handles.take_wakeup(), |registry| {
registry.take_wakeup_handle(identity)
})?;
if let Some(wakeup) = wakeup {
clear_provider_handle(wakeup);
}
}
if selected.includes_work() {
let work = take_detached_or_owned_handle(handles.take_work(), |registry| {
registry.take_work_handle(identity)
})?;
if let Some(work) = work {
clear_provider_handle(work);
}
}
restore_provider_handles(handles)
}
RegistryEffect::DispatchWatchdog {
successor,
successor_delay_ns,
work,
} => {
if let Some(wakeup) = handles.take_wakeup() {
clear_provider_handle(wakeup);
}
let replaced_work = take_detached_or_owned_handle(handles.take_work(), |registry| {
registry.take_work_handle(successor.identity())
})?;
if let Some(replaced_work) = replaced_work {
clear_provider_handle(replaced_work);
}
let successor_token = successor.clone();
let successor_handle =
platform::set_timer(Duration::from_nanos(*successor_delay_ns), async move {
dispatch_watchdog_scheduler(&successor_token);
});
bind_provider_handle(successor, successor_handle)?;
let work_token = work.clone();
let work_handle = platform::set_timer(Duration::ZERO, async move {
dispatch_watchdog_work(&work_token);
});
if let Err(error) =
bind_provider_handle(work, work_handle).and_then(|()| confirm_effect(effect))
{
clear_entry_provider_handles(successor.identity())?;
return Err(error);
}
Ok(())
}
}
}
fn take_detached_or_owned_handle(
detached: Option<ProviderHandle>,
take_owned: impl FnOnce(&mut TimerRegistry) -> Option<ProviderHandle>,
) -> Result<Option<ProviderHandle>, TimerError> {
detached.map_or_else(
|| with_registry_mut(|registry| Ok(take_owned(registry))),
|handle| Ok(Some(handle)),
)
}
fn bind_provider_handle(token: &CallbackToken, handle: TimerHandle) -> Result<(), TimerError> {
RUNTIME
.with(|runtime| {
#[cfg(test)]
if take_provider_install_fault() {
return Err((TimerError::OwnershipInvariant, handle));
}
let Ok(mut runtime) = runtime.try_borrow_mut() else {
return Err((TimerError::RuntimeBusy, handle));
};
let Some(registry) = runtime.as_mut() else {
return Err((TimerError::NotInitialized, handle));
};
match registry.install_provider_handle(token, handle) {
Ok(()) => Ok(()),
Err((error, handle)) => Err((TimerError::from(error), handle)),
}
})
.map_err(|(error, handle)| {
platform::clear_timer(handle);
error
})
}
fn confirm_effect(effect: &RegistryEffect) -> Result<(), TimerError> {
#[cfg(test)]
if take_provider_confirmation_fault() {
return Err(TimerError::OwnershipInvariant);
}
with_registry_mut(|registry| {
registry
.confirm_effect_applied(effect)
.map_err(TimerError::from)
})
}
fn restore_provider_handles(mut handles: ProviderHandles) -> Result<(), TimerError> {
let wakeup_failure = handles
.take_wakeup()
.and_then(|handle| restore_provider_handle(handle).err());
let work_failure = handles
.take_work()
.and_then(|handle| restore_provider_handle(handle).err());
wakeup_failure.or(work_failure).map_or(Ok(()), Err)
}
fn restore_provider_handle(handle: ProviderHandle) -> Result<(), TimerError> {
let (token, handle) = handle.into_parts();
bind_provider_handle(&token, handle)
}
fn clear_provider_handle(handle: ProviderHandle) {
let (_, handle) = handle.into_parts();
platform::clear_timer(handle);
}
fn clear_provider_handles(mut handles: ProviderHandles) {
if let Some(wakeup) = handles.take_wakeup() {
clear_provider_handle(wakeup);
}
if let Some(work) = handles.take_work() {
clear_provider_handle(work);
}
}
fn clear_entry_provider_handles(identity: &TimerIdentity) -> Result<(), TimerError> {
let handles = with_registry_mut(|registry| Ok(registry.take_provider_handles(identity)))?;
clear_provider_handles(handles);
Ok(())
}
#[expect(
clippy::future_not_send,
reason = "IC callbacks and canister-local state are single-threaded."
)]
fn dispatch_wakeup(token: CallbackToken) -> impl Future<Output = ()> {
let mut delivery = OrdinaryDelivery {
token,
completed: false,
};
async move {
match delivery.token.role() {
CallbackRole::OrdinaryWork => dispatch_ordinary(&delivery.token).await,
CallbackRole::WatchdogScheduler => dispatch_watchdog_scheduler(&delivery.token),
CallbackRole::WatchdogWork => {}
}
delivery.completed = true;
drop(delivery);
}
}
struct OrdinaryDelivery {
token: CallbackToken,
completed: bool,
}
impl Drop for OrdinaryDelivery {
fn drop(&mut self) {
if self.completed || self.token.role() != CallbackRole::OrdinaryWork {
return;
}
let removed = RUNTIME.with(|runtime| {
let Ok(mut runtime) = runtime.try_borrow_mut() else {
let runtime = runtime.try_borrow().map_err(|_| TimerError::RuntimeBusy)?;
return if runtime
.as_ref()
.is_some_and(|registry| registry.owns_ordinary_delivery(&self.token))
{
Err(TimerError::RuntimeBusy)
} else {
Ok(None)
};
};
Ok(runtime
.as_mut()
.and_then(|registry| registry.abandon_ordinary(&self.token, platform::time_ns())))
});
let removed = removed
.unwrap_or_else(|error| trap_callback_failure("ordinary delivery abandonment", &error));
drop(removed);
}
}
#[expect(
clippy::future_not_send,
reason = "IC callbacks and canister-local state are single-threaded."
)]
async fn dispatch_ordinary(token: &CallbackToken) {
let measurement = CallbackMeasurementStart::capture();
let callback = with_registry_mut(|registry| Ok(registry.begin_ordinary(token)));
let callback = match callback {
Ok(Some(callback)) => callback,
Ok(None) => return,
Err(error) => trap_callback_failure("ordinary callback acceptance", &error),
};
let future = {
let Ok(mut callback) = callback.try_borrow_mut() else {
finish_ordinary_callback(
token,
OrdinaryRunResult::new(
TimerCompletion::invariant_failure(0),
OrdinaryDirective::Stop,
),
);
return;
};
callback(token.clone())
};
let result = future.await;
finish_ordinary_callback(token, result);
record_callback_measurements(token, measurement.finish());
}
fn finish_ordinary_callback(token: &CallbackToken, result: OrdinaryRunResult) {
let transition = with_registry_mut(|registry| {
registry
.complete_ordinary(token, platform::time_ns(), result)
.map_err(TimerError::from)
});
let transition = transition
.unwrap_or_else(|error| trap_callback_failure("ordinary callback completion", &error));
finish_callback_transition(token, transition, ProviderHandles::default());
}
fn dispatch_watchdog_scheduler(token: &CallbackToken) {
let measurement = CallbackMeasurementStart::capture();
let dispatched = with_registry_mut(|registry| {
let handles = registry
.take_watchdog_scheduler_handles(token)
.map_err(TimerError::from)?;
Ok((
registry.begin_watchdog_scheduler(token, platform::time_ns()),
handles,
))
});
let (transition, handles) = dispatched
.unwrap_or_else(|error| trap_callback_failure("watchdog scheduler transition", &error));
let accepted = !matches!(transition.effect(), RegistryEffect::None);
finish_callback_transition(token, transition, handles);
if accepted {
record_callback_measurements(token, measurement.finish());
}
}
fn dispatch_watchdog_work(token: &CallbackToken) {
let measurement = CallbackMeasurementStart::capture();
let callback = with_registry_mut(|registry| Ok(registry.begin_watchdog_work(token)));
let callback = match callback {
Ok(Some(callback)) => callback,
Ok(None) => return,
Err(error) => trap_callback_failure("watchdog work acceptance", &error),
};
let result = {
let Ok(mut callback) = callback.try_borrow_mut() else {
trap_callback_failure(
"watchdog callback ownership",
&TimerError::OwnershipInvariant,
);
};
callback(token.clone())
};
finish_watchdog_dispatch(token, result);
record_callback_measurements(token, measurement.finish());
}
fn finish_watchdog_dispatch(token: &CallbackToken, result: WatchdogRunResult) {
let completed = with_registry_mut(|registry| {
let handles = registry
.take_provider_handles_for_claim(token.claim())
.map_err(TimerError::from)?;
#[cfg(test)]
{
if take_watchdog_completion_fault() {
return Err(TimerError::OwnershipInvariant);
}
}
let transition = registry
.complete_watchdog_work(token, platform::time_ns(), result)
.map_err(TimerError::from)?;
Ok((transition, handles))
});
let (transition, handles) =
completed.unwrap_or_else(|error| trap_callback_failure("watchdog work completion", &error));
finish_callback_transition(token, transition, handles);
}
fn finish_callback_transition(
token: &CallbackToken,
transition: RegistryTransition,
handles: ProviderHandles,
) {
match finish_transition(transition, handles) {
Ok(()) | Err(TimerError::ControlFailure(_)) => {}
Err(error) => {
if token.role() == CallbackRole::WatchdogWork {
trap_callback_failure("watchdog provider-handle completion", &error);
}
fail_claim_provider_binding(token.claim()).unwrap_or_else(|binding_error| {
trap_callback_failure("provider-binding failure cleanup", &binding_error)
});
}
}
}
fn fail_claim_provider_binding(claim: &RegistrationClaim) -> Result<(), TimerError> {
let failed = with_registry_mut(|registry| {
registry
.fail_registration(claim, TimerControlFailure::ProviderBindingFailed)
.map_err(TimerError::from)
});
let (handles, transition) = failed?;
clear_provider_handles(handles);
drop(transition);
Ok(())
}
#[derive(Clone, Copy)]
struct CallbackMeasurementStart {
instructions_before: u64,
memory_start: MemoryPageExtent,
}
impl CallbackMeasurementStart {
fn capture() -> Self {
let memory_start = platform::memory_pages();
let instructions_before = platform::instruction_counter();
Self {
instructions_before,
memory_start,
}
}
fn finish(self) -> CallbackMeasurement {
let instructions = platform::instruction_counter().saturating_sub(self.instructions_before);
let memory_end = platform::memory_pages();
CallbackMeasurement {
instructions,
memory_start: self.memory_start,
memory_end,
}
}
}
#[derive(Clone, Copy)]
struct CallbackMeasurement {
instructions: u64,
memory_start: MemoryPageExtent,
memory_end: MemoryPageExtent,
}
fn record_callback_measurements(token: &CallbackToken, measurement: CallbackMeasurement) {
with_registry_mut(|registry| {
registry
.record_callback_measurements(
token,
measurement.instructions,
measurement.memory_start,
measurement.memory_end,
)
.map_err(TimerError::from)
})
.unwrap_or_else(|error| trap_callback_failure("callback measurement accounting", &error));
}
fn trap_callback_failure(context: &str, error: &TimerError) -> ! {
platform::trap(&format!("ic-timers {context} failed: {error}"))
}
fn with_registry<T>(
operation: impl FnOnce(&TimerRegistry) -> Result<T, TimerError>,
) -> Result<T, TimerError> {
RUNTIME.with(|runtime| {
let runtime = runtime.try_borrow().map_err(|_| TimerError::RuntimeBusy)?;
let registry = runtime.as_ref().ok_or(TimerError::NotInitialized)?;
operation(registry)
})
}
fn with_registry_mut<T>(
operation: impl FnOnce(&mut TimerRegistry) -> Result<T, TimerError>,
) -> Result<T, TimerError> {
RUNTIME.with(|runtime| {
let mut runtime = runtime
.try_borrow_mut()
.map_err(|_| TimerError::RuntimeBusy)?;
let registry = runtime.as_mut().ok_or(TimerError::NotInitialized)?;
operation(registry)
})
}
#[cfg(test)]
fn reset_for_test(now_ns: u64, canister_version: u64) {
platform::reset(now_ns, canister_version);
WATCHDOG_COMPLETION_FAULT.with(|fault| fault.set(false));
PROVIDER_INSTALL_FAULT_AFTER.with(|fault| fault.set(None));
PROVIDER_CONFIRMATION_FAULT.with(|fault| fault.set(false));
RUNTIME.with(|runtime| {
*runtime.borrow_mut() = None;
});
}
#[cfg(test)]
thread_local! {
static WATCHDOG_COMPLETION_FAULT: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
static PROVIDER_INSTALL_FAULT_AFTER: std::cell::Cell<Option<u64>> = const { std::cell::Cell::new(None) };
static PROVIDER_CONFIRMATION_FAULT: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}
#[cfg(test)]
fn inject_watchdog_completion_fault() {
WATCHDOG_COMPLETION_FAULT.with(|fault| fault.set(true));
}
#[cfg(test)]
fn take_watchdog_completion_fault() -> bool {
WATCHDOG_COMPLETION_FAULT.with(|fault| fault.replace(false))
}
#[cfg(test)]
fn inject_provider_install_fault() {
inject_provider_install_fault_after(0);
}
#[cfg(test)]
fn inject_provider_install_fault_after(successful_installs: u64) {
PROVIDER_INSTALL_FAULT_AFTER.with(|fault| fault.set(Some(successful_installs)));
}
#[cfg(test)]
fn take_provider_install_fault() -> bool {
PROVIDER_INSTALL_FAULT_AFTER.with(|fault| match fault.get() {
Some(0) => {
fault.set(None);
true
}
Some(remaining) => {
fault.set(Some(remaining - 1));
false
}
None => false,
})
}
#[cfg(test)]
fn inject_provider_confirmation_fault() {
PROVIDER_CONFIRMATION_FAULT.with(|fault| fault.set(true));
}
#[cfg(test)]
fn take_provider_confirmation_fault() -> bool {
PROVIDER_CONFIRMATION_FAULT.with(|fault| fault.replace(false))
}
#[cfg(test)]
mod tests;