use crate::{
platform::{self, TimerHandle},
registry::{
CallbackAcceptance, CallbackRole, CallbackToken, OrdinaryCallback, ProviderHandle,
ProviderHandles, RegisterError, RegistrationClaim, RegistryEffect, RegistryError,
RegistryTransition, TimerRegistry, WatchdogCallback,
},
schedule::{ScheduleError, TimerCadence, TimerDirective, TimerSchedule},
snapshot::{
DeclarationLifetime, TimerCompletion, TimerControlFailure, TimerEpoch, TimerIdentity,
TimerInventorySnapshot, TimerPolicy, TimerRunResult, 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) };
}
#[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 { .. } => Self::OwnershipInvariant,
RegistryError::Schedule(error) => Self::Schedule(error),
RegistryError::MissingCallback | RegistryError::ProviderHandleAlreadyOwned => {
Self::OwnershipInvariant
}
}
}
}
pub fn initialize_runtime() -> Result<TimerEpoch, TimerError> {
let epoch = TimerEpoch::new(platform::canister_version(), platform::time_ns());
RUNTIME.with(|runtime| {
let mut runtime = runtime
.try_borrow_mut()
.map_err(|_| TimerError::RuntimeBusy)?;
if let Some(registry) = runtime.as_ref() {
return Ok(registry.epoch());
}
*runtime = Some(TimerRegistry::new(epoch));
Ok(epoch)
})
}
struct CallbackContext {
token: CallbackToken,
}
impl CallbackContext {
const fn new(token: CallbackToken) -> Self {
Self { token }
}
fn claim(&self) -> RegistrationClaim {
RegistrationClaim::from_callback(&self.token)
}
const fn identity(&self) -> &TimerIdentity {
self.token.identity()
}
fn cancel(&self) -> Result<(), TimerError> {
cancel_claim(&self.claim(), Some(&self.token))
}
fn schedule_once(&self, schedule: TimerSchedule) -> Result<(), TimerError> {
ensure_once_claim(&self.claim(), Some(&self.token), schedule)
}
fn schedule_recurring(&self) -> Result<(), TimerError> {
ensure_recurring_claim(&self.claim(), Some(&self.token))
}
fn schedule_watchdog_immediately(&self) -> Result<(), TimerError> {
ensure_watchdog_immediately_claim(&self.claim(), Some(&self.token))
}
fn reconcile_ordinary(&self, schedule: Option<TimerSchedule>) -> Result<(), TimerError> {
reconcile_ordinary_claim(&self.claim(), Some(&self.token), schedule)
}
}
pub struct OnceContext {
inner: CallbackContext,
}
impl OnceContext {
const fn new(token: CallbackToken) -> Self {
Self {
inner: CallbackContext::new(token),
}
}
#[must_use]
pub const fn identity(&self) -> &TimerIdentity {
self.inner.identity()
}
pub fn ensure_scheduled(&self, schedule: TimerSchedule) -> Result<(), TimerError> {
self.inner.schedule_once(schedule)
}
pub fn reconcile_schedule(&self, schedule: Option<TimerSchedule>) -> Result<(), TimerError> {
self.inner.reconcile_ordinary(schedule)
}
pub fn cancel(&self) -> Result<(), TimerError> {
self.inner.cancel()
}
}
pub struct AfterCompletionContext {
inner: CallbackContext,
}
impl AfterCompletionContext {
const fn new(token: CallbackToken) -> Self {
Self {
inner: CallbackContext::new(token),
}
}
#[must_use]
pub const fn identity(&self) -> &TimerIdentity {
self.inner.identity()
}
pub fn ensure_scheduled(&self) -> Result<(), TimerError> {
self.inner.schedule_recurring()
}
pub fn reconcile_schedule(&self, schedule: Option<TimerSchedule>) -> Result<(), TimerError> {
self.inner.reconcile_ordinary(schedule)
}
pub fn cancel(&self) -> Result<(), TimerError> {
self.inner.cancel()
}
}
pub struct WatchdogContext {
inner: CallbackContext,
}
impl WatchdogContext {
const fn new(token: CallbackToken) -> Self {
Self {
inner: CallbackContext::new(token),
}
}
#[must_use]
pub const fn identity(&self) -> &TimerIdentity {
self.inner.identity()
}
pub fn ensure_scheduled(&self) -> Result<(), TimerError> {
self.inner.schedule_recurring()
}
pub fn ensure_scheduled_immediately(&self) -> Result<(), TimerError> {
self.inner.schedule_watchdog_immediately()
}
pub fn cancel(&self) -> Result<(), TimerError> {
self.inner.cancel()
}
}
#[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,
}
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 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,
lifetime: DeclarationLifetime,
callback: F,
) -> Result<OnceRegistration, TimerError>
where
F: FnMut(OnceContext) -> Fut + 'static,
Fut: Future<Output = TimerRunResult> + 'static,
{
let callback = erase_ordinary_callback(callback, OnceContext::new);
let claim = with_registry_mut(|registry| {
registry
.register_once_with_callback(identity, lifetime, callback)
.map_err(TimerError::from)
})?;
Ok(OnceRegistration { claim })
}
pub fn register_after_completion<F, Fut>(
identity: TimerIdentity,
cadence: TimerCadence,
lifetime: DeclarationLifetime,
callback: F,
) -> Result<AfterCompletionRegistration, TimerError>
where
F: FnMut(AfterCompletionContext) -> Fut + 'static,
Fut: Future<Output = TimerRunResult> + 'static,
{
let callback = erase_ordinary_callback(callback, AfterCompletionContext::new);
let claim = with_registry_mut(|registry| {
registry
.register_after_completion_with_callback(identity, cadence, lifetime, callback)
.map_err(TimerError::from)
})?;
Ok(AfterCompletionRegistration { claim })
}
pub fn register_watchdog<F>(
identity: TimerIdentity,
cadence: TimerCadence,
lifetime: DeclarationLifetime,
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 = with_registry_mut(|registry| {
registry
.register_watchdog_with_callback(identity, cadence, lifetime, callback)
.map_err(TimerError::from)
})?;
Ok(WatchdogRegistration { claim })
}
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 = TimerRunResult> + 'static,
{
let registration = reconcile_registration(registration, identity, TimerPolicy::Once, || {
register_once(identity.clone(), DeclarationLifetime::Retained, callback)
})?;
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 = TimerRunResult> + 'static,
{
let registration = reconcile_registration(
registration,
identity,
TimerPolicy::AfterCompletion { cadence },
|| {
register_after_completion(
identity.clone(),
cadence,
DeclarationLifetime::Retained,
callback,
)
},
)?;
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,
{
let registration = reconcile_registration(
registration,
identity,
TimerPolicy::Watchdog { cadence },
|| {
register_watchdog(
identity.clone(),
cadence,
DeclarationLifetime::Retained,
callback,
)
},
)?;
match desired {
WatchdogReconcileState::Inactive => registration.cancel(),
WatchdogReconcileState::Scheduled => registration.ensure_scheduled(),
WatchdogReconcileState::ScheduledImmediately => registration.ensure_scheduled_immediately(),
}
}
fn reconcile_registration<'a, Registration>(
registration: &'a mut Option<Registration>,
identity: &TimerIdentity,
policy: TimerPolicy,
register: impl FnOnce() -> Result<Registration, TimerError>,
) -> Result<&'a Registration, TimerError>
where
Registration: RegistrationClaimOwner,
{
if registration.is_none() {
*registration = Some(register()?);
}
let registration = registration
.as_ref()
.ok_or(TimerError::ReconciliationConflict)?;
verify_declaration(registration.registration_claim(), identity, policy)?;
Ok(registration)
}
fn verify_declaration(
claim: &RegistrationClaim,
identity: &TimerIdentity,
policy: TimerPolicy,
) -> Result<(), TimerError> {
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)
})
}
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, F, Fut>(
mut callback: F,
context: fn(CallbackToken) -> Context,
) -> OrdinaryCallback
where
F: FnMut(Context) -> Fut + 'static,
Fut: Future<Output = TimerRunResult> + 'static,
{
Rc::new(RefCell::new(Box::new(move |token| {
Box::pin(callback(context(token)))
})))
}
fn apply_claim_transition(
claim: &RegistrationClaim,
context: Option<&CallbackToken>,
operation: impl FnOnce(&mut TimerRegistry) -> Result<RegistryTransition, RegistryError>,
) -> Result<(), TimerError> {
let transition = with_registry_mut(|registry| {
validate_context(registry, context)?;
operation(registry).map_err(TimerError::from)
})?;
finish_claim_transition(claim, transition, ProviderHandles::default())
}
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> {
if schedule.is_none() {
let (handles, transition) = with_registry_mut(|registry| {
validate_context(registry, context)?;
registry
.validate_ordinary_claim(claim)
.map_err(TimerError::from)?;
let handles = registry
.take_provider_handles_for_claim(claim)
.map_err(TimerError::from)?;
let transition = registry
.reconcile_ordinary(claim, platform::time_ns(), None)
.map_err(TimerError::from);
Ok((handles, transition))
})?;
return finish_detached_claim_transition(claim, handles, transition);
}
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 cancel_claim(
claim: &RegistrationClaim,
context: Option<&CallbackToken>,
) -> Result<(), TimerError> {
apply_detached_claim_transition(claim, context, |registry| registry.cancel(claim))
}
fn validate_context(
registry: &TimerRegistry,
context: Option<&CallbackToken>,
) -> Result<(), TimerError> {
context.map_or(Ok(()), |token| {
registry
.validate_running_context(token)
.map_err(TimerError::from)
})
}
fn unregister_claim(claim: &RegistrationClaim) -> Result<(), TimerError> {
apply_detached_claim_transition(claim, None, |registry| registry.unregister(claim))
}
fn apply_detached_claim_transition(
claim: &RegistrationClaim,
context: Option<&CallbackToken>,
operation: impl FnOnce(&mut TimerRegistry) -> Result<RegistryTransition, RegistryError>,
) -> Result<(), TimerError> {
let (handles, transition) = with_registry_mut(|registry| {
validate_context(registry, context)?;
let handles = registry
.take_provider_handles_for_claim(claim)
.map_err(TimerError::from)?;
let transition = operation(registry).map_err(TimerError::from);
Ok((handles, transition))
})?;
finish_detached_claim_transition(claim, handles, transition)
}
fn finish_detached_claim_transition(
claim: &RegistrationClaim,
handles: ProviderHandles,
transition: Result<RegistryTransition, TimerError>,
) -> Result<(), TimerError> {
match transition {
Ok(transition) => finish_claim_transition(claim, transition, handles),
Err(error) => match restore_provider_handles(handles) {
Ok(()) => Err(error),
Err(restoration_error) => retire_failed_claim(claim, restoration_error),
},
}
}
fn finish_claim_transition(
claim: &RegistrationClaim,
transition: RegistryTransition,
handles: ProviderHandles,
) -> Result<(), TimerError> {
match finish_transition(transition, handles) {
result @ (Ok(()) | Err(TimerError::ControlFailure(_))) => result,
Err(error) => retire_failed_claim(claim, error),
}
}
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 effect = transition.into_effect();
apply_effect(&effect, handles)?;
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, 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)?;
arm_wakeup(effect)
}
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, .. } => {
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);
}
dispatch_watchdog_effect(effect)
}
}
}
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 arm_wakeup(effect: &RegistryEffect) -> Result<(), TimerError> {
let RegistryEffect::ArmWakeup {
token, delay_ns, ..
} = effect
else {
return Err(TimerError::OwnershipInvariant);
};
let task_token = token.clone();
let handle = platform::set_timer(Duration::from_nanos(*delay_ns), async move {
dispatch_wakeup(task_token).await;
});
if let Err((error, handle)) = install_provider_handle(token, handle) {
platform::clear_timer(handle);
return Err(error);
}
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(())
}
fn dispatch_watchdog_effect(effect: &RegistryEffect) -> Result<(), TimerError> {
let RegistryEffect::DispatchWatchdog {
successor,
successor_delay_ns,
work,
..
} = effect
else {
return Err(TimerError::OwnershipInvariant);
};
let successor_token = successor.clone();
let successor_handle =
platform::set_timer(Duration::from_nanos(*successor_delay_ns), async move {
dispatch_watchdog_scheduler(&successor_token);
});
if let Err((error, handle)) = install_provider_handle(successor, successor_handle) {
platform::clear_timer(handle);
return Err(error);
}
let work_token = work.clone();
let work_handle = platform::set_timer(Duration::ZERO, async move {
dispatch_watchdog_work(&work_token);
});
if let Err((error, handle)) = install_provider_handle(work, work_handle) {
platform::clear_timer(handle);
clear_entry_provider_handles(successor.identity())?;
return Err(error);
}
if let Err(error) = confirm_effect(effect) {
clear_entry_provider_handles(successor.identity())?;
return Err(error);
}
Ok(())
}
fn install_provider_handle(
token: &CallbackToken,
handle: TimerHandle,
) -> Result<(), (TimerError, TimerHandle)> {
#[cfg(test)]
if take_provider_install_fault() {
return Err((TimerError::OwnershipInvariant, handle));
}
RUNTIME.with(|runtime| {
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)),
}
})
}
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();
match install_provider_handle(&token, handle) {
Ok(()) => Ok(()),
Err((error, handle)) => {
platform::clear_timer(handle);
Err(error)
}
}
}
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(ProviderHandles::from_parts(
registry.take_wakeup_handle(identity),
registry.take_work_handle(identity),
))
})?;
clear_provider_handles(handles);
Ok(())
}
#[allow(clippy::future_not_send)] async fn dispatch_wakeup(token: CallbackToken) {
match token.role() {
CallbackRole::OrdinaryWork => dispatch_ordinary(token).await,
CallbackRole::WatchdogScheduler => dispatch_watchdog_scheduler(&token),
CallbackRole::WatchdogWork => {}
}
}
#[allow(clippy::future_not_send)] async fn dispatch_ordinary(token: CallbackToken) {
let measurement = CallbackMeasurementStart::capture();
let accepted = with_registry_mut(|registry| {
registry.consume_provider_handle(&token);
Ok(registry.begin_ordinary(&token))
});
match accepted {
Ok(CallbackAcceptance::Accepted) => {}
Ok(CallbackAcceptance::Stale) => return,
Err(error) => trap_callback_failure("ordinary callback acceptance", &error),
}
let callback = match with_registry(|registry| {
registry.ordinary_callback(&token).map_err(TimerError::from)
}) {
Ok(callback) => callback,
Err(TimerError::OwnershipInvariant) => {
fail_ordinary_dispatch(&token);
return;
}
Err(error) => trap_callback_failure("ordinary callback lookup", &error),
};
let future = {
let Ok(mut callback) = callback.try_borrow_mut() else {
fail_ordinary_dispatch(&token);
return;
};
callback(token.clone())
};
let result = future.await;
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());
record_callback_measurements(&token, measurement.finish());
}
fn fail_ordinary_dispatch(token: &CallbackToken) {
let transition = with_registry_mut(|registry| {
registry
.complete_ordinary(
token,
platform::time_ns(),
TimerRunResult::new(TimerCompletion::invariant_failure(0), TimerDirective::Stop),
)
.map_err(TimerError::from)
});
let transition = transition.unwrap_or_else(|error| {
trap_callback_failure("ordinary invariant-failure completion", &error)
});
finish_callback_transition(token, transition, ProviderHandles::default());
}
fn dispatch_watchdog_scheduler(token: &CallbackToken) {
let measurement = CallbackMeasurementStart::capture();
let transition = with_registry_mut(|registry| {
registry.consume_provider_handle(token);
Ok(registry.begin_watchdog_scheduler(token, platform::time_ns()))
});
let transition = transition
.unwrap_or_else(|error| trap_callback_failure("watchdog scheduler transition", &error));
let accepted = !matches!(transition.effect(), RegistryEffect::None);
finish_callback_transition(token, transition, ProviderHandles::default());
if accepted {
record_callback_measurements(token, measurement.finish());
}
}
fn dispatch_watchdog_work(token: &CallbackToken) {
let measurement = CallbackMeasurementStart::capture();
let accepted = with_registry_mut(|registry| {
registry.consume_provider_handle(token);
Ok(registry.begin_watchdog_work(token))
});
match accepted {
Ok(CallbackAcceptance::Accepted) => {}
Ok(CallbackAcceptance::Stale) => return,
Err(error) => trap_callback_failure("watchdog work acceptance", &error),
}
let callback =
match with_registry(|registry| registry.watchdog_callback(token).map_err(TimerError::from))
{
Ok(callback) => callback,
Err(error) => trap_callback_failure("watchdog callback lookup", &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 claim = RegistrationClaim::from_callback(token);
let completed = with_registry_mut(|registry| {
let handles = registry
.take_provider_handles_for_claim(&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 @ (TimerError::NotInitialized
| TimerError::RuntimeBusy
| TimerError::Register(_)
| TimerError::Schedule(_)
| TimerError::RegistrationExpired
| TimerError::OwnershipInvariant
| TimerError::ReconciliationConflict),
) => {
if token.role() == CallbackRole::WatchdogWork {
trap_callback_failure("watchdog provider-handle completion", &error);
}
fail_provider_binding(token).unwrap_or_else(|binding_error| {
trap_callback_failure("provider-binding failure cleanup", &binding_error)
});
}
}
}
fn fail_provider_binding(token: &CallbackToken) -> Result<(), TimerError> {
let claim = RegistrationClaim::from_callback(token);
fail_claim_provider_binding(&claim)
}
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)
});
clear_provider_handles(failed?);
Ok(())
}
#[derive(Clone, Copy)]
struct CallbackMeasurementStart {
instructions_before: u64,
memory_start: platform::MemoryPages,
}
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: platform::MemoryPages,
memory_end: platform::MemoryPages,
}
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;