Skip to main content

ic_timers/registry/
mod.rs

1//! Bounded canonical registry and provider-neutral policy transition engine.
2//!
3//! The registry emits provider-neutral effects. The runtime binds them to
4//! linear provider handles while the registry remains the sole logical state
5//! authority.
6
7mod effect;
8
9use effect::WatchdogDispatch;
10pub use effect::{RegistryEffect, WakeupToken};
11
12use crate::{
13    callback::OrdinaryRunResult,
14    control::{OrdinaryPending, TimerControl, TimerRegistration, WakeupArm},
15    platform::TimerHandle,
16    schedule::{OrdinaryDirective, ResolvedSchedule, ScheduleError, TimerCadence, TimerSchedule},
17    snapshot::{
18        DeclarationLifetime, InactiveReason, MemoryPageExtent, MemoryPageSample,
19        OrdinaryRuntimeStateSnapshot, TimerCompletion, TimerCompletionOutcome, TimerControlFailure,
20        TimerDirectiveSnapshot, TimerEpoch, TimerIdentity, TimerInventorySnapshot,
21        TimerObservabilitySnapshot, TimerPolicy, TimerRegistrationId, TimerRuntimeStateSnapshot,
22        TimerSchedulingMode, TimerSnapshot, WatchdogAttemptStatus, WatchdogDecision,
23        WatchdogRunResult, WatchdogRuntimeStateSnapshot,
24    },
25};
26use std::{cell::RefCell, collections::BTreeMap, fmt, future::Future, pin::Pin, rc::Rc};
27use thiserror::Error;
28
29/// Maximum declarations owned by one canonical registry.
30pub const MAX_TIMER_REGISTRATIONS: usize = 64;
31
32/// Opaque logical ownership claim returned by pure registration.
33#[derive(Debug, Eq, PartialEq)]
34pub struct RegistrationClaim {
35    identity: TimerIdentity,
36    claim_generation: u64,
37}
38
39/// Internal callback role carried by a generation-checked dispatch token.
40#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub enum CallbackRole {
42    OrdinaryWork,
43    WatchdogScheduler,
44    WatchdogWork,
45}
46
47/// Identity and generations an internal callback must present.
48#[derive(Debug, Eq, PartialEq)]
49pub struct CallbackToken {
50    claim: RegistrationClaim,
51    callback_generation: u64,
52    role: CallbackRole,
53}
54
55impl CallbackToken {
56    const fn new(
57        identity: TimerIdentity,
58        claim_generation: u64,
59        callback_generation: u64,
60        role: CallbackRole,
61    ) -> Self {
62        Self {
63            claim: RegistrationClaim {
64                identity,
65                claim_generation,
66            },
67            callback_generation,
68            role,
69        }
70    }
71
72    pub(crate) const fn identity(&self) -> &TimerIdentity {
73        self.claim.identity()
74    }
75
76    pub(crate) const fn claim(&self) -> &RegistrationClaim {
77        &self.claim
78    }
79
80    #[cfg(test)]
81    pub(crate) const fn callback_generation(&self) -> u64 {
82        self.callback_generation
83    }
84
85    pub(crate) const fn role(&self) -> CallbackRole {
86        self.role
87    }
88}
89
90impl Clone for CallbackToken {
91    fn clone(&self) -> Self {
92        // Queued deliveries need owned tokens. Retained registration capabilities
93        // remain non-clone, and running contexts borrow this token's exact claim.
94        Self::new(
95            self.identity().clone(),
96            self.claim.claim_generation,
97            self.callback_generation,
98            self.role,
99        )
100    }
101}
102
103impl RegistrationClaim {
104    pub(crate) const fn identity(&self) -> &TimerIdentity {
105        &self.identity
106    }
107
108    pub(crate) const fn claim_generation(&self) -> u64 {
109        self.claim_generation
110    }
111}
112
113/// Non-empty subset of provider callbacks cleared by one effect.
114#[derive(Clone, Copy, Debug, Eq, PartialEq)]
115pub enum CallbacksToClear {
116    Wakeup,
117    Work,
118    WakeupAndWork,
119}
120
121impl CallbacksToClear {
122    pub(crate) const fn includes_wakeup(self) -> bool {
123        matches!(self, Self::Wakeup | Self::WakeupAndWork)
124    }
125
126    pub(crate) const fn includes_work(self) -> bool {
127        matches!(self, Self::Work | Self::WakeupAndWork)
128    }
129}
130
131/// One pure transition and any terminal checked-control failure it produced.
132pub struct RegistryTransition {
133    effect: RegistryEffect,
134    failure: Option<TimerControlFailure>,
135    // Removal transfers callback captures out of the registry borrow. Keep
136    // them alive until runtime has also applied the provider cleanup.
137    removed: Option<Box<Entry>>,
138}
139
140impl fmt::Debug for RegistryTransition {
141    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
142        formatter
143            .debug_struct("RegistryTransition")
144            .field("effect", &self.effect)
145            .field("failure", &self.failure)
146            .field("removed", &self.removed.is_some())
147            .finish()
148    }
149}
150
151impl RegistryTransition {
152    const fn normal(effect: RegistryEffect) -> Self {
153        Self {
154            effect,
155            failure: None,
156            removed: None,
157        }
158    }
159
160    const fn terminal(effect: RegistryEffect, failure: TimerControlFailure) -> Self {
161        Self {
162            effect,
163            failure: Some(failure),
164            removed: None,
165        }
166    }
167
168    pub(crate) const fn effect(&self) -> &RegistryEffect {
169        &self.effect
170    }
171
172    pub(crate) const fn failure(&self) -> Option<TimerControlFailure> {
173        self.failure
174    }
175
176    #[cfg(test)]
177    pub(crate) fn into_effect(self) -> RegistryEffect {
178        self.effect
179    }
180}
181
182/// Failure to claim one canonical timer identity.
183#[non_exhaustive]
184#[derive(Clone, Debug, Eq, Error, PartialEq)]
185pub enum RegisterError {
186    /// The identity already has one canonical claimant.
187    #[error("timer identity is already registered: {0:?}")]
188    IdentityAlreadyRegistered(TimerIdentity),
189    /// The fixed registry capacity has been reached.
190    #[error("timer registry capacity of {max} declarations has been reached")]
191    CapacityExceeded {
192        /// Fixed registry capacity.
193        max: usize,
194    },
195    /// Logical registration claim generations are exhausted.
196    #[error("timer registration claim generation exhausted")]
197    ClaimGenerationExhausted,
198}
199
200/// Invalid request against a logical registration claim.
201#[derive(Clone, Debug, Eq, Error, PartialEq)]
202pub enum RegistryError {
203    #[error("timer registration is no longer present")]
204    UnknownRegistration,
205    #[error("timer registration claim is stale")]
206    StaleRegistration,
207    #[error("timer policy invariant failed for {actual}")]
208    PolicyMismatch { actual: &'static str },
209    #[error("timer callback token is stale")]
210    StaleCallback,
211    #[error("timer registration already owns the provider handle for this callback role")]
212    ProviderHandleAlreadyOwned,
213    #[error(transparent)]
214    Schedule(#[from] ScheduleError),
215}
216
217type OrdinaryFuture = Pin<Box<dyn Future<Output = OrdinaryRunResult>>>;
218pub type OrdinaryCallback = Rc<RefCell<dyn FnMut(CallbackToken) -> OrdinaryFuture>>;
219pub type WatchdogCallback = Rc<RefCell<dyn FnMut(CallbackToken) -> WatchdogRunResult>>;
220
221struct OwnedProviderHandle {
222    callback_generation: u64,
223    handle: TimerHandle,
224}
225
226/// One temporarily detached exact provider capability.
227#[must_use = "restore or clear the detached provider handle"]
228pub struct ProviderHandle {
229    token: CallbackToken,
230    handle: TimerHandle,
231}
232
233impl ProviderHandle {
234    pub(crate) fn into_parts(self) -> (CallbackToken, TimerHandle) {
235        (self.token, self.handle)
236    }
237}
238
239/// The at-most-two provider capabilities owned by one timer entry.
240#[must_use = "restore or clear every detached provider handle"]
241#[derive(Default)]
242pub struct ProviderHandles {
243    wakeup: Option<ProviderHandle>,
244    work: Option<ProviderHandle>,
245}
246
247impl ProviderHandles {
248    pub(crate) const fn take_wakeup(&mut self) -> Option<ProviderHandle> {
249        self.wakeup.take()
250    }
251
252    pub(crate) const fn take_work(&mut self) -> Option<ProviderHandle> {
253        self.work.take()
254    }
255}
256
257#[derive(Clone, Copy, Debug, Eq, PartialEq)]
258enum OrdinaryRequest {
259    Ensure,
260    Reconcile,
261}
262
263#[derive(Clone, Copy, Debug, Eq, PartialEq)]
264enum WatchdogPending {
265    Cancel,
266    Ensure,
267    EnsureImmediately,
268    Reconcile(ResolvedSchedule),
269    Unregister,
270}
271
272#[derive(Clone, Copy, Debug, Eq, PartialEq)]
273enum WatchdogScheduleRequest {
274    Cadence,
275    Immediate,
276    Reconcile(ResolvedSchedule),
277}
278
279// One payload owns policy, callback type and control together. An ordinary
280// entry without cadence is Once; one with cadence is AfterCompletion.
281enum EntryKind {
282    Ordinary {
283        cadence: Option<TimerCadence>,
284        callback: OrdinaryCallback,
285        control: TimerControl,
286    },
287    Watchdog {
288        cadence: TimerCadence,
289        callback: WatchdogCallback,
290        control: WatchdogControl,
291    },
292}
293
294impl EntryKind {
295    const fn wakeup_role(&self) -> CallbackRole {
296        match self {
297            Self::Ordinary { .. } => CallbackRole::OrdinaryWork,
298            Self::Watchdog { .. } => CallbackRole::WatchdogScheduler,
299        }
300    }
301
302    fn ordinary(cadence: Option<TimerCadence>, callback: OrdinaryCallback) -> Self {
303        Self::Ordinary {
304            cadence,
305            callback,
306            control: TimerControl::default(),
307        }
308    }
309
310    fn watchdog(cadence: TimerCadence, callback: WatchdogCallback) -> Self {
311        Self::Watchdog {
312            cadence,
313            callback,
314            control: WatchdogControl::default(),
315        }
316    }
317
318    const fn policy(&self) -> TimerPolicy {
319        match self {
320            Self::Ordinary { cadence: None, .. } => TimerPolicy::Once,
321            Self::Ordinary {
322                cadence: Some(cadence),
323                ..
324            } => TimerPolicy::AfterCompletion { cadence: *cadence },
325            Self::Watchdog { cadence, .. } => TimerPolicy::Watchdog { cadence: *cadence },
326        }
327    }
328}
329
330#[derive(Clone, Copy, Debug, Eq, PartialEq)]
331enum WatchdogState {
332    Inactive {
333        reason: InactiveReason,
334    },
335    Scheduled {
336        deadline_ns: u64,
337    },
338    AwaitingWork {
339        successor_deadline_ns: u64,
340        attempt_status: WatchdogAttemptStatus,
341        pending: Option<WatchdogPending>,
342    },
343}
344
345#[derive(Debug)]
346struct WatchdogControl {
347    // One allocation history survives inactive state. A dispatch stamps its
348    // successor and work with one generation, distinguished by callback role.
349    // Requests while awaiting work cannot replace that successor until completion.
350    generation: u64,
351    state: WatchdogState,
352}
353
354impl Default for WatchdogControl {
355    fn default() -> Self {
356        Self {
357            generation: 0,
358            state: WatchdogState::Inactive {
359                reason: InactiveReason::NeverScheduled,
360            },
361        }
362    }
363}
364
365impl WatchdogControl {
366    fn arm_scheduler(
367        &mut self,
368        claim: &RegistrationClaim,
369        now_ns: u64,
370        deadline_ns: u64,
371        arm: WakeupArm,
372    ) -> RegistryTransition {
373        let Some(generation) = self.generation.checked_add(1) else {
374            let cleanup = clear_wakeup_if(arm.replaces_existing());
375            return self.terminate(cleanup, TimerControlFailure::GenerationExhausted);
376        };
377        self.generation = generation;
378        self.state = WatchdogState::Scheduled { deadline_ns };
379        RegistryTransition::normal(RegistryEffect::ArmWakeup {
380            token: WakeupToken::scheduler(claim, generation),
381            delay_ns: deadline_ns.saturating_sub(now_ns),
382            arm,
383        })
384    }
385
386    const fn terminate(
387        &mut self,
388        effect: RegistryEffect,
389        failure: TimerControlFailure,
390    ) -> RegistryTransition {
391        self.state = WatchdogState::Inactive {
392            reason: InactiveReason::ControlFailure(failure),
393        };
394        RegistryTransition::terminal(effect, failure)
395    }
396}
397
398struct Entry {
399    claim_generation: u64,
400    lifetime: DeclarationLifetime,
401    kind: EntryKind,
402    scheduling_mode: TimerSchedulingMode,
403    latest_directive: Option<TimerDirectiveSnapshot>,
404    latest_requested_delay_ns: Option<u64>,
405    latest_armed_delay_ns: Option<u64>,
406    confirmed_wakeup_generation: Option<u64>,
407    wakeup: Option<OwnedProviderHandle>,
408    work: Option<OwnedProviderHandle>,
409    observability: TimerObservabilitySnapshot,
410}
411
412impl Entry {
413    const fn new(claim_generation: u64, kind: EntryKind, lifetime: DeclarationLifetime) -> Self {
414        let scheduling_mode = match kind.policy() {
415            TimerPolicy::Once => TimerSchedulingMode::Once,
416            TimerPolicy::AfterCompletion { .. } => TimerSchedulingMode::AfterCompletion,
417            TimerPolicy::Watchdog { .. } => TimerSchedulingMode::Watchdog,
418        };
419
420        Self {
421            claim_generation,
422            lifetime,
423            kind,
424            scheduling_mode,
425            latest_directive: None,
426            latest_requested_delay_ns: None,
427            latest_armed_delay_ns: None,
428            confirmed_wakeup_generation: None,
429            wakeup: None,
430            work: None,
431            observability: TimerObservabilitySnapshot::EMPTY,
432        }
433    }
434
435    /// Validate the role, live generation and state once for binding or restoration.
436    const fn provider_handle_slot(
437        &mut self,
438        token: &CallbackToken,
439        final_arm: bool,
440    ) -> Result<&mut Option<OwnedProviderHandle>, RegistryError> {
441        match (&self.kind, token.role) {
442            (EntryKind::Ordinary { control, .. }, CallbackRole::OrdinaryWork)
443                if matches!(
444                    control.registration,
445                    TimerRegistration::Scheduled { .. }
446                        if control.generation() == token.callback_generation
447                ) =>
448            {
449                Ok(&mut self.wakeup)
450            }
451            (EntryKind::Watchdog { control, .. }, CallbackRole::WatchdogScheduler)
452                if matches!(
453                    control.state,
454                    WatchdogState::Scheduled { .. } | WatchdogState::AwaitingWork { .. }
455                        if control.generation == token.callback_generation
456                            && (!final_arm || matches!(control.state, WatchdogState::Scheduled { .. }))
457                ) =>
458            {
459                Ok(&mut self.wakeup)
460            }
461            (EntryKind::Watchdog { control, .. }, CallbackRole::WatchdogWork)
462                if matches!(
463                    control.state,
464                    WatchdogState::AwaitingWork {
465                        attempt_status: WatchdogAttemptStatus::Dispatched,
466                        ..
467                    } if control.generation == token.callback_generation
468                ) =>
469            {
470                Ok(&mut self.work)
471            }
472            _ => Err(RegistryError::StaleCallback),
473        }
474    }
475
476    fn record_arm(&mut self, token: &CallbackToken, delay_ns: u64) {
477        if self.confirmed_wakeup_generation == Some(token.callback_generation) {
478            return;
479        }
480        self.confirmed_wakeup_generation = Some(token.callback_generation);
481        self.latest_armed_delay_ns = Some(delay_ns);
482        let counters = self.observability.counters_mut();
483        counters.record_wakeup_armed();
484        if token.role == CallbackRole::WatchdogWork {
485            counters.record_work_dispatched();
486        }
487    }
488
489    const fn snapshot(&self, identity: TimerIdentity, epoch: TimerEpoch) -> TimerSnapshot {
490        let state = match &self.kind {
491            EntryKind::Ordinary { control, .. } => match control.registration {
492                TimerRegistration::Inactive { reason } => {
493                    TimerRuntimeStateSnapshot::Inactive { reason }
494                }
495                TimerRegistration::Scheduled { deadline_ns } => {
496                    TimerRuntimeStateSnapshot::Ordinary(OrdinaryRuntimeStateSnapshot::Scheduled {
497                        generation: control.generation(),
498                        deadline_ns,
499                    })
500                }
501                TimerRegistration::Running { .. } => {
502                    TimerRuntimeStateSnapshot::Ordinary(OrdinaryRuntimeStateSnapshot::Running {
503                        generation: control.generation(),
504                    })
505                }
506            },
507            EntryKind::Watchdog { control, .. } => match control.state {
508                WatchdogState::Inactive { reason } => {
509                    TimerRuntimeStateSnapshot::Inactive { reason }
510                }
511                WatchdogState::Scheduled { deadline_ns } => {
512                    TimerRuntimeStateSnapshot::Watchdog(WatchdogRuntimeStateSnapshot::Scheduled {
513                        scheduler_generation: control.generation,
514                        deadline_ns,
515                    })
516                }
517                WatchdogState::AwaitingWork {
518                    successor_deadline_ns,
519                    attempt_status,
520                    ..
521                } => TimerRuntimeStateSnapshot::Watchdog(
522                    WatchdogRuntimeStateSnapshot::AwaitingWork {
523                        successor_generation: control.generation,
524                        successor_deadline_ns,
525                        attempt_status,
526                    },
527                ),
528            },
529        };
530
531        TimerSnapshot::new(
532            identity,
533            TimerRegistrationId::new(epoch, self.claim_generation),
534            self.kind.policy(),
535            self.lifetime,
536            state,
537            self.scheduling_mode,
538            self.latest_directive,
539            self.latest_requested_delay_ns,
540            self.latest_armed_delay_ns,
541            &self.observability,
542        )
543    }
544
545    /// Decide terminal removal from the entry already selected and authorized
546    /// by the transition owner. Runtime detaches handles before that transition.
547    const fn remove_on_failure(&self, transition: &RegistryTransition) -> bool {
548        transition.failure().is_some()
549            && matches!(self.lifetime, DeclarationLifetime::RemoveWhenStopped)
550    }
551
552    fn take_wakeup_handle(&mut self, identity: &TimerIdentity) -> Option<ProviderHandle> {
553        let role = self.kind.wakeup_role();
554        self.wakeup
555            .take()
556            .map(|owned| detach_provider_handle(identity, self.claim_generation, role, owned))
557    }
558
559    fn take_work_handle(&mut self, identity: &TimerIdentity) -> Option<ProviderHandle> {
560        self.work.take().map(|owned| {
561            detach_provider_handle(
562                identity,
563                self.claim_generation,
564                CallbackRole::WatchdogWork,
565                owned,
566            )
567        })
568    }
569
570    fn take_provider_handles(&mut self, identity: &TimerIdentity) -> ProviderHandles {
571        ProviderHandles {
572            wakeup: self.take_wakeup_handle(identity),
573            work: self.take_work_handle(identity),
574        }
575    }
576
577    // The caller selected this entry by identity. Fired handles are consumed
578    // independently of callback acceptance, but only for the exact claim,
579    // policy role and owned slot generation.
580    fn consume_provider_handle(&mut self, token: &CallbackToken) {
581        if !self.owns_token_claim(token) {
582            return;
583        }
584        let slot = match token.role {
585            CallbackRole::OrdinaryWork | CallbackRole::WatchdogScheduler
586                if token.role == self.kind.wakeup_role() =>
587            {
588                &mut self.wakeup
589            }
590            CallbackRole::WatchdogWork if matches!(self.kind, EntryKind::Watchdog { .. }) => {
591                &mut self.work
592            }
593            _ => return,
594        };
595        if slot
596            .as_ref()
597            .is_some_and(|owned| owned.callback_generation == token.callback_generation)
598        {
599            *slot = None;
600        }
601    }
602
603    const fn owns_token_claim(&self, token: &CallbackToken) -> bool {
604        self.claim_generation == token.claim().claim_generation()
605    }
606
607    // Registry lookups select this entry by token identity first.
608    const fn owns_running_work(&self, token: &CallbackToken) -> bool {
609        if !self.owns_token_claim(token) {
610            return false;
611        }
612        match (&self.kind, token.role) {
613            (EntryKind::Ordinary { control, .. }, CallbackRole::OrdinaryWork) => matches!(
614                control.registration,
615                TimerRegistration::Running { .. }
616                    if control.generation() == token.callback_generation
617            ),
618            (EntryKind::Watchdog { control, .. }, CallbackRole::WatchdogWork) => matches!(
619                control.state,
620                WatchdogState::AwaitingWork {
621                    attempt_status: WatchdogAttemptStatus::Running,
622                    ..
623                } if control.generation == token.callback_generation
624            ),
625            (
626                EntryKind::Ordinary { .. },
627                CallbackRole::WatchdogScheduler | CallbackRole::WatchdogWork,
628            )
629            | (
630                EntryKind::Watchdog { .. },
631                CallbackRole::OrdinaryWork | CallbackRole::WatchdogScheduler,
632            ) => false,
633        }
634    }
635}
636
637/// Pure, fixed-capacity canonical registry.
638pub struct TimerRegistry {
639    epoch: TimerEpoch,
640    next_claim_generation: u64,
641    entries: BTreeMap<TimerIdentity, Entry>,
642}
643
644impl TimerRegistry {
645    pub(crate) const fn new(epoch: TimerEpoch) -> Self {
646        Self {
647            epoch,
648            next_claim_generation: 0,
649            entries: BTreeMap::new(),
650        }
651    }
652
653    pub(crate) const fn epoch(&self) -> TimerEpoch {
654        self.epoch
655    }
656
657    pub(crate) fn register_once_with_callback(
658        &mut self,
659        identity: TimerIdentity,
660        lifetime: DeclarationLifetime,
661        callback: OrdinaryCallback,
662    ) -> Result<RegistrationClaim, RegisterError> {
663        self.register(identity, EntryKind::ordinary(None, callback), lifetime)
664    }
665
666    pub(crate) fn register_after_completion_with_callback(
667        &mut self,
668        identity: TimerIdentity,
669        cadence: TimerCadence,
670        lifetime: DeclarationLifetime,
671        callback: OrdinaryCallback,
672    ) -> Result<RegistrationClaim, RegisterError> {
673        self.register(
674            identity,
675            EntryKind::ordinary(Some(cadence), callback),
676            lifetime,
677        )
678    }
679
680    pub(crate) fn register_watchdog_with_callback(
681        &mut self,
682        identity: TimerIdentity,
683        cadence: TimerCadence,
684        lifetime: DeclarationLifetime,
685        callback: WatchdogCallback,
686    ) -> Result<RegistrationClaim, RegisterError> {
687        self.register(identity, EntryKind::watchdog(cadence, callback), lifetime)
688    }
689
690    fn register(
691        &mut self,
692        identity: TimerIdentity,
693        kind: EntryKind,
694        lifetime: DeclarationLifetime,
695    ) -> Result<RegistrationClaim, RegisterError> {
696        if self.entries.contains_key(&identity) {
697            return Err(RegisterError::IdentityAlreadyRegistered(identity));
698        }
699        if self.entries.len() == MAX_TIMER_REGISTRATIONS {
700            return Err(RegisterError::CapacityExceeded {
701                max: MAX_TIMER_REGISTRATIONS,
702            });
703        }
704        let claim_generation = self
705            .next_claim_generation
706            .checked_add(1)
707            .ok_or(RegisterError::ClaimGenerationExhausted)?;
708
709        self.next_claim_generation = claim_generation;
710        self.entries.insert(
711            identity.clone(),
712            Entry::new(claim_generation, kind, lifetime),
713        );
714        Ok(RegistrationClaim {
715            identity,
716            claim_generation,
717        })
718    }
719
720    /// Apply public control with handles left in place for no-op and rejected
721    /// requests. Removed entries carry their handles and captures out together.
722    pub(crate) fn transition_claim(
723        &mut self,
724        claim: &RegistrationClaim,
725        operation: impl FnOnce(&mut Self) -> Result<RegistryTransition, RegistryError>,
726    ) -> Result<(RegistryTransition, ProviderHandles), RegistryError> {
727        self.entry(claim)?;
728        let mut transition = operation(self)?;
729        let handles = if let Some(entry) = &mut transition.removed {
730            entry.take_provider_handles(claim.identity())
731        } else if matches!(transition.effect(), RegistryEffect::None) {
732            ProviderHandles::default()
733        } else {
734            self.take_provider_handles(claim.identity())
735        };
736        Ok((transition, handles))
737    }
738
739    /// Apply an initial schedule already checked before claim allocation.
740    pub(crate) fn reconcile_initial(
741        &mut self,
742        claim: &RegistrationClaim,
743        now_ns: u64,
744        requested: ResolvedSchedule,
745    ) -> Result<RegistryTransition, RegistryError> {
746        match self.entry(claim)?.kind {
747            EntryKind::Ordinary { .. } => {
748                self.request_ordinary(claim, now_ns, requested, OrdinaryRequest::Reconcile)
749            }
750            EntryKind::Watchdog { .. } => {
751                self.ensure_watchdog(claim, now_ns, WatchdogScheduleRequest::Reconcile(requested))
752            }
753        }
754    }
755
756    pub(crate) fn ensure_once(
757        &mut self,
758        claim: &RegistrationClaim,
759        now_ns: u64,
760        schedule: TimerSchedule,
761    ) -> Result<RegistryTransition, RegistryError> {
762        let requested = schedule.resolve(now_ns)?;
763        let entry = self.entry(claim)?;
764        if !matches!(entry.kind, EntryKind::Ordinary { cadence: None, .. }) {
765            return Err(RegistryError::PolicyMismatch {
766                actual: entry.kind.policy().label(),
767            });
768        }
769        self.request_ordinary(claim, now_ns, requested, OrdinaryRequest::Ensure)
770    }
771
772    /// Reconcile an ordinary declaration to one exact desired schedule.
773    ///
774    /// Unlike `ensure`, reconciliation may move an existing deadline later.
775    /// `None` cancels live work; declaration lifetime determines whether the
776    /// callback authority remains registered.
777    pub(crate) fn reconcile_ordinary(
778        &mut self,
779        claim: &RegistrationClaim,
780        now_ns: u64,
781        schedule: Option<TimerSchedule>,
782    ) -> Result<RegistryTransition, RegistryError> {
783        let entry = self.entry(claim)?;
784        if !matches!(entry.kind, EntryKind::Ordinary { .. }) {
785            return Err(RegistryError::PolicyMismatch {
786                actual: entry.kind.policy().label(),
787            });
788        }
789        let Some(schedule) = schedule else {
790            return self.cancel(claim);
791        };
792        self.request_ordinary(
793            claim,
794            now_ns,
795            schedule.resolve(now_ns)?,
796            OrdinaryRequest::Reconcile,
797        )
798    }
799
800    pub(crate) fn ensure_recurring(
801        &mut self,
802        claim: &RegistrationClaim,
803        now_ns: u64,
804    ) -> Result<RegistryTransition, RegistryError> {
805        let entry = self.entry(claim)?;
806        match &entry.kind {
807            EntryKind::Ordinary { cadence: None, .. } => {
808                Err(RegistryError::PolicyMismatch { actual: "once" })
809            }
810            EntryKind::Ordinary {
811                cadence: Some(cadence),
812                control,
813                ..
814            } => {
815                let cadence = *cadence;
816                // Existing authority satisfies recurrence without calculating
817                // an unused successor, even when now + cadence would overflow.
818                let deadline_ns = match control.registration {
819                    TimerRegistration::Scheduled { deadline_ns } => deadline_ns,
820                    TimerRegistration::Inactive { .. } | TimerRegistration::Running { .. } => {
821                        cadence.deadline_after(now_ns)?
822                    }
823                };
824                self.request_ordinary(
825                    claim,
826                    now_ns,
827                    ResolvedSchedule {
828                        deadline_ns,
829                        requested_delay_ns: Some(cadence.as_nanos()),
830                        mode: TimerSchedulingMode::AfterCompletion,
831                    },
832                    OrdinaryRequest::Ensure,
833                )
834            }
835            EntryKind::Watchdog { .. } => {
836                self.ensure_watchdog(claim, now_ns, WatchdogScheduleRequest::Cadence)
837            }
838        }
839    }
840
841    pub(crate) fn ensure_watchdog_immediately(
842        &mut self,
843        claim: &RegistrationClaim,
844        now_ns: u64,
845    ) -> Result<RegistryTransition, RegistryError> {
846        self.ensure_watchdog(claim, now_ns, WatchdogScheduleRequest::Immediate)
847    }
848
849    pub(crate) fn reconcile_watchdog_schedule(
850        &mut self,
851        claim: &RegistrationClaim,
852        now_ns: u64,
853        schedule: Option<TimerSchedule>,
854    ) -> Result<RegistryTransition, RegistryError> {
855        let entry = self.entry(claim)?;
856        if !matches!(entry.kind, EntryKind::Watchdog { .. }) {
857            return Err(RegistryError::PolicyMismatch {
858                actual: entry.kind.policy().label(),
859            });
860        }
861        let Some(schedule) = schedule else {
862            return self.cancel(claim);
863        };
864        let requested = schedule.resolve(now_ns)?;
865        self.ensure_watchdog(claim, now_ns, WatchdogScheduleRequest::Reconcile(requested))
866    }
867
868    fn request_ordinary(
869        &mut self,
870        claim: &RegistrationClaim,
871        now_ns: u64,
872        requested: ResolvedSchedule,
873        request: OrdinaryRequest,
874    ) -> Result<RegistryTransition, RegistryError> {
875        let entry = self.entry_mut(claim)?;
876        let actual = entry.kind.policy().label();
877        let EntryKind::Ordinary { control, .. } = &mut entry.kind else {
878            return Err(RegistryError::PolicyMismatch { actual });
879        };
880        entry.observability.counters_mut().record_schedule_request();
881        entry.latest_requested_delay_ns = requested.requested_delay_ns;
882
883        let arm = match request {
884            OrdinaryRequest::Ensure => control.schedule(requested.deadline_ns),
885            OrdinaryRequest::Reconcile => control.reconcile(requested.deadline_ns),
886        };
887
888        let arm = match arm {
889            Ok(arm) => arm,
890            Err(error) => {
891                let clear_wakeup = control.terminate(InactiveReason::ControlFailure(error));
892                let transition = RegistryTransition::terminal(clear_wakeup_if(clear_wakeup), error);
893                let remove = entry.remove_on_failure(&transition);
894                return Ok(remove_after(
895                    &mut self.entries,
896                    claim.identity(),
897                    transition,
898                    remove,
899                ));
900            }
901        };
902
903        if let TimerRegistration::Running { pending } = &mut control.registration {
904            *pending = Some(select_pending_ordinary(*pending, request, requested));
905        }
906        if arm.is_some() || matches!(request, OrdinaryRequest::Reconcile) {
907            entry.scheduling_mode = requested.mode;
908        }
909
910        let Some(arm) = arm else {
911            entry.observability.counters_mut().record_coalesced();
912            return Ok(RegistryTransition::normal(RegistryEffect::None));
913        };
914        Ok(RegistryTransition::normal(RegistryEffect::ArmWakeup {
915            token: WakeupToken::ordinary(claim, control.generation()),
916            delay_ns: requested.deadline_ns.saturating_sub(now_ns),
917            arm,
918        }))
919    }
920
921    fn ensure_watchdog(
922        &mut self,
923        claim: &RegistrationClaim,
924        now_ns: u64,
925        request: WatchdogScheduleRequest,
926    ) -> Result<RegistryTransition, RegistryError> {
927        let entry = self.entry_mut(claim)?;
928        let actual = entry.kind.policy().label();
929        let EntryKind::Watchdog {
930            cadence, control, ..
931        } = &mut entry.kind
932        else {
933            return Err(RegistryError::PolicyMismatch { actual });
934        };
935        let cadence = *cadence;
936        let (mode, requested_delay_ns) = match request {
937            WatchdogScheduleRequest::Cadence => {
938                (TimerSchedulingMode::Watchdog, Some(cadence.as_nanos()))
939            }
940            WatchdogScheduleRequest::Immediate => (TimerSchedulingMode::Continuation, Some(0)),
941            WatchdogScheduleRequest::Reconcile(requested) => {
942                (requested.mode, requested.requested_delay_ns)
943            }
944        };
945        let transition = match &mut control.state {
946            WatchdogState::Inactive { .. } => {
947                let deadline_ns = match request {
948                    WatchdogScheduleRequest::Cadence => cadence.deadline_after(now_ns)?,
949                    WatchdogScheduleRequest::Immediate => now_ns,
950                    WatchdogScheduleRequest::Reconcile(requested) => requested.deadline_ns,
951                };
952                control.arm_scheduler(claim, now_ns, deadline_ns, WakeupArm::Initial)
953            }
954            WatchdogState::Scheduled { deadline_ns }
955                if match request {
956                    WatchdogScheduleRequest::Immediate => *deadline_ns > now_ns,
957                    WatchdogScheduleRequest::Reconcile(requested) => {
958                        *deadline_ns != requested.deadline_ns
959                    }
960                    WatchdogScheduleRequest::Cadence => false,
961                } =>
962            {
963                let deadline_ns = match request {
964                    WatchdogScheduleRequest::Reconcile(requested) => requested.deadline_ns,
965                    WatchdogScheduleRequest::Immediate | WatchdogScheduleRequest::Cadence => now_ns,
966                };
967                control.arm_scheduler(claim, now_ns, deadline_ns, WakeupArm::Replacement)
968            }
969            WatchdogState::Scheduled { .. } => RegistryTransition::normal(RegistryEffect::None),
970            WatchdogState::AwaitingWork {
971                attempt_status: WatchdogAttemptStatus::Dispatched,
972                pending,
973                ..
974            } => {
975                if let WatchdogScheduleRequest::Reconcile(requested) = request {
976                    *pending = Some(WatchdogPending::Reconcile(requested));
977                }
978                RegistryTransition::normal(RegistryEffect::None)
979            }
980            WatchdogState::AwaitingWork {
981                attempt_status: WatchdogAttemptStatus::Running,
982                pending,
983                ..
984            } => {
985                *pending = Some(select_pending_watchdog(*pending, request, now_ns));
986                RegistryTransition::normal(RegistryEffect::None)
987            }
988        };
989        if matches!(transition.effect(), RegistryEffect::ArmWakeup { .. }) {
990            entry.scheduling_mode = mode;
991        }
992        if matches!(
993            (transition.effect(), transition.failure()),
994            (RegistryEffect::None, None)
995        ) {
996            entry.observability.counters_mut().record_coalesced();
997        }
998        entry.observability.counters_mut().record_schedule_request();
999        entry.latest_requested_delay_ns = requested_delay_ns;
1000        let remove = entry.remove_on_failure(&transition);
1001        Ok(remove_after(
1002            &mut self.entries,
1003            claim.identity(),
1004            transition,
1005            remove,
1006        ))
1007    }
1008
1009    pub(crate) fn cancel(
1010        &mut self,
1011        claim: &RegistrationClaim,
1012    ) -> Result<RegistryTransition, RegistryError> {
1013        let identity = claim.identity();
1014        let (effect, remove) = {
1015            let entry = self.entry_mut(claim)?;
1016            let (effect, stopped) = match &mut entry.kind {
1017                EntryKind::Ordinary { control, .. } => {
1018                    let effect = match &mut control.registration {
1019                        TimerRegistration::Inactive { .. } => RegistryEffect::None,
1020                        TimerRegistration::Scheduled { .. } => {
1021                            control.terminate(InactiveReason::Cancelled);
1022                            entry.observability.counters_mut().record_cancellation();
1023                            RegistryEffect::ClearCallbacks {
1024                                handles: CallbacksToClear::Wakeup,
1025                            }
1026                        }
1027                        TimerRegistration::Running { pending } => {
1028                            if !matches!(*pending, Some(OrdinaryPending::Unregister)) {
1029                                *pending = Some(OrdinaryPending::Cancel);
1030                            }
1031                            RegistryEffect::None
1032                        }
1033                    };
1034                    (
1035                        effect,
1036                        matches!(control.registration, TimerRegistration::Inactive { .. }),
1037                    )
1038                }
1039                EntryKind::Watchdog { control, .. } => {
1040                    let handles = match &mut control.state {
1041                        WatchdogState::Inactive { .. } => None,
1042                        WatchdogState::Scheduled { .. } => Some(CallbacksToClear::Wakeup),
1043                        WatchdogState::AwaitingWork {
1044                            attempt_status: WatchdogAttemptStatus::Dispatched,
1045                            ..
1046                        } => Some(CallbacksToClear::WakeupAndWork),
1047                        WatchdogState::AwaitingWork {
1048                            attempt_status: WatchdogAttemptStatus::Running,
1049                            pending,
1050                            ..
1051                        } => {
1052                            if !matches!(*pending, Some(WatchdogPending::Unregister)) {
1053                                *pending = Some(WatchdogPending::Cancel);
1054                            }
1055                            None
1056                        }
1057                    };
1058                    let effect = handles.map_or(RegistryEffect::None, |handles| {
1059                        control.state = WatchdogState::Inactive {
1060                            reason: InactiveReason::Cancelled,
1061                        };
1062                        entry.observability.counters_mut().record_cancellation();
1063                        RegistryEffect::ClearCallbacks { handles }
1064                    });
1065                    let stopped = matches!(control.state, WatchdogState::Inactive { .. });
1066                    (effect, stopped)
1067                }
1068            };
1069            (
1070                effect,
1071                stopped && matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped),
1072            )
1073        };
1074
1075        Ok(remove_after(
1076            &mut self.entries,
1077            identity,
1078            RegistryTransition::normal(effect),
1079            remove,
1080        ))
1081    }
1082
1083    /// Remove the declaration owned by one exact logical claim.
1084    ///
1085    /// Removal requested by running work is finalized by normal completion or,
1086    /// for ordinary work, confirmed delivery abandonment. A request made in a
1087    /// trapping message rolls back with that message's heap mutations; an
1088    /// already committed pending request remains eligible for finalization.
1089    pub(crate) fn unregister(
1090        &mut self,
1091        claim: &RegistrationClaim,
1092    ) -> Result<RegistryTransition, RegistryError> {
1093        let identity = claim.identity();
1094        let entry = self.entry_mut(claim)?;
1095        match &mut entry.kind {
1096            EntryKind::Ordinary {
1097                control:
1098                    TimerControl {
1099                        registration: TimerRegistration::Running { pending },
1100                        ..
1101                    },
1102                ..
1103            } => {
1104                *pending = Some(OrdinaryPending::Unregister);
1105                Ok(RegistryTransition::normal(RegistryEffect::None))
1106            }
1107            EntryKind::Watchdog {
1108                control:
1109                    WatchdogControl {
1110                        state:
1111                            WatchdogState::AwaitingWork {
1112                                attempt_status: WatchdogAttemptStatus::Running,
1113                                pending,
1114                                ..
1115                            },
1116                        ..
1117                    },
1118                ..
1119            } => {
1120                *pending = Some(WatchdogPending::Unregister);
1121                Ok(RegistryTransition::normal(RegistryEffect::None))
1122            }
1123            EntryKind::Ordinary { .. } | EntryKind::Watchdog { .. } => {
1124                let transition = self.cancel(claim)?;
1125                Ok(remove_after(&mut self.entries, identity, transition, true))
1126            }
1127        }
1128    }
1129
1130    /// Consume this delivery's handle, then accept ordinary work and return its
1131    /// callback.
1132    pub(crate) fn begin_ordinary(&mut self, token: &CallbackToken) -> Option<OrdinaryCallback> {
1133        let entry = self.entries.get_mut(token.identity())?;
1134        entry.consume_provider_handle(token);
1135        if token.role != CallbackRole::OrdinaryWork || !entry.owns_token_claim(token) {
1136            entry.observability.counters_mut().record_stale_wakeup();
1137            return None;
1138        }
1139        let EntryKind::Ordinary {
1140            control, callback, ..
1141        } = &mut entry.kind
1142        else {
1143            entry.observability.counters_mut().record_stale_wakeup();
1144            return None;
1145        };
1146        if control.begin(token.callback_generation) {
1147            entry.observability.counters_mut().record_work_started();
1148            Some(Rc::clone(callback))
1149        } else {
1150            entry.observability.counters_mut().record_stale_wakeup();
1151            None
1152        }
1153    }
1154
1155    pub(crate) fn complete_ordinary(
1156        &mut self,
1157        token: &CallbackToken,
1158        now_ns: u64,
1159        result: OrdinaryRunResult,
1160    ) -> Result<RegistryTransition, RegistryError> {
1161        let identity = token.identity();
1162        let (transition, remove) = {
1163            let entry = self.running_work_entry_mut(token)?;
1164            let EntryKind::Ordinary {
1165                cadence, control, ..
1166            } = &mut entry.kind
1167            else {
1168                return Err(RegistryError::StaleCallback);
1169            };
1170            let cadence = *cadence;
1171            let completion = result.completion();
1172            let pending_command = control.pending();
1173            let remove_on_stop = matches!(pending_command, Some(OrdinaryPending::Unregister))
1174                || matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped);
1175            if completion.outcome() == TimerCompletionOutcome::InvariantFailure {
1176                let transition = stop_ordinary_completion(entry, completion, now_ns, None);
1177                (transition, remove_on_stop)
1178            } else {
1179                let prepared = prepare_ordinary_completion(
1180                    pending_command,
1181                    result.directive(),
1182                    now_ns,
1183                    cadence,
1184                )
1185                .and_then(|(selected, directive_snapshot, cancelled)| {
1186                    selected.map_or(Ok(()), |selected| {
1187                        control.arm_deadline(selected.deadline_ns)
1188                    })?;
1189                    Ok((selected, directive_snapshot, cancelled))
1190                });
1191                match prepared {
1192                    Err(failure) => (
1193                        stop_ordinary_completion(entry, completion, now_ns, Some(failure)),
1194                        remove_on_stop,
1195                    ),
1196                    Ok((selected_schedule, directive_snapshot, cancelled)) => {
1197                        entry.latest_directive = Some(directive_snapshot);
1198                        entry.observability.record_completion(completion, now_ns);
1199
1200                        let remove = selected_schedule.is_none() && remove_on_stop;
1201                        let transition = if let Some(selected) = selected_schedule {
1202                            entry.scheduling_mode = selected.mode;
1203                            entry.latest_requested_delay_ns = selected.requested_delay_ns;
1204                            RegistryTransition::normal(RegistryEffect::ArmWakeup {
1205                                token: WakeupToken::ordinary(token.claim(), control.generation()),
1206                                delay_ns: selected.deadline_ns.saturating_sub(now_ns),
1207                                arm: WakeupArm::Initial,
1208                            })
1209                        } else {
1210                            let reason = if cancelled {
1211                                entry.observability.counters_mut().record_cancellation();
1212                                InactiveReason::Cancelled
1213                            } else {
1214                                InactiveReason::Stopped
1215                            };
1216                            control.terminate(reason);
1217                            RegistryTransition::normal(RegistryEffect::None)
1218                        };
1219                        (transition, remove)
1220                    }
1221                }
1222            }
1223        };
1224
1225        Ok(remove_after(
1226            &mut self.entries,
1227            identity,
1228            transition,
1229            remove,
1230        ))
1231    }
1232
1233    /// Whether a confirmed ordinary delivery still owns scheduled or running work.
1234    /// Detached scheduled handles belong to their synchronous transition owner.
1235    pub(crate) fn owns_ordinary_delivery(&self, token: &CallbackToken) -> bool {
1236        let Some(entry) = self.entries.get(token.identity()) else {
1237            return false;
1238        };
1239        let EntryKind::Ordinary { control, .. } = &entry.kind else {
1240            return false;
1241        };
1242        token.role == CallbackRole::OrdinaryWork
1243            && entry.owns_token_claim(token)
1244            && control.generation() == token.callback_generation
1245            && entry.confirmed_wakeup_generation == Some(token.callback_generation)
1246            && match control.registration {
1247                TimerRegistration::Running { .. } => true,
1248                TimerRegistration::Scheduled { .. } => entry
1249                    .wakeup
1250                    .as_ref()
1251                    .is_some_and(|owned| owned.callback_generation == token.callback_generation),
1252                TimerRegistration::Inactive { .. } => false,
1253            }
1254    }
1255
1256    /// Retire a dropped delivery without provider calls or fabricated completion.
1257    /// Return removed callback ownership so capture destructors run after the
1258    /// runtime releases its registry borrow, including during CDK cleanup.
1259    pub(crate) fn abandon_ordinary(
1260        &mut self,
1261        token: &CallbackToken,
1262        now_ns: u64,
1263    ) -> Option<OrdinaryCallback> {
1264        if !self.owns_ordinary_delivery(token) {
1265            return None;
1266        }
1267        let entry = self.entries.get_mut(token.identity())?;
1268        let EntryKind::Ordinary { control, .. } = &mut entry.kind else {
1269            return None;
1270        };
1271        let remove = matches!(control.pending(), Some(OrdinaryPending::Unregister))
1272            || matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped);
1273        control.terminate(InactiveReason::Abandoned);
1274        // The provider/executor is already dropping this future. Forget the
1275        // consumed handle; calling clear_timer here could reborrow its task map.
1276        entry.wakeup = None;
1277        entry.observability.record_unacknowledged(now_ns);
1278        if !remove {
1279            return None;
1280        }
1281        self.entries.remove(token.identity()).and_then(|entry| {
1282            if let EntryKind::Ordinary { callback, .. } = entry.kind {
1283                Some(callback)
1284            } else {
1285                None
1286            }
1287        })
1288    }
1289
1290    pub(crate) fn begin_watchdog_scheduler(
1291        &mut self,
1292        token: &CallbackToken,
1293        now_ns: u64,
1294    ) -> RegistryTransition {
1295        let Some(entry) = self.entries.get_mut(token.identity()) else {
1296            return RegistryTransition::normal(RegistryEffect::None);
1297        };
1298        if token.role != CallbackRole::WatchdogScheduler || !entry.owns_token_claim(token) {
1299            entry.observability.counters_mut().record_stale_wakeup();
1300            return RegistryTransition::normal(RegistryEffect::None);
1301        }
1302        let EntryKind::Watchdog {
1303            cadence, control, ..
1304        } = &mut entry.kind
1305        else {
1306            entry.observability.counters_mut().record_stale_wakeup();
1307            return RegistryTransition::normal(RegistryEffect::None);
1308        };
1309        let cadence = *cadence;
1310
1311        let accepted = match control.state {
1312            WatchdogState::Scheduled { .. }
1313            | WatchdogState::AwaitingWork {
1314                attempt_status: WatchdogAttemptStatus::Dispatched,
1315                ..
1316            } => control.generation == token.callback_generation,
1317            WatchdogState::Inactive { .. }
1318            | WatchdogState::AwaitingWork {
1319                attempt_status: WatchdogAttemptStatus::Running,
1320                ..
1321            } => false,
1322        };
1323        if !accepted {
1324            entry.observability.counters_mut().record_stale_wakeup();
1325            return RegistryTransition::normal(RegistryEffect::None);
1326        }
1327
1328        entry
1329            .observability
1330            .counters_mut()
1331            .record_scheduler_started();
1332        if matches!(control.state, WatchdogState::AwaitingWork { .. }) {
1333            entry.observability.record_unacknowledged(now_ns);
1334        }
1335        let next_dispatch = control
1336            .generation
1337            .checked_add(1)
1338            .ok_or(TimerControlFailure::GenerationExhausted)
1339            .and_then(|generation| {
1340                cadence
1341                    .deadline_after(now_ns)
1342                    .map(|deadline_ns| (generation, deadline_ns))
1343                    .map_err(|_| TimerControlFailure::DeadlineOverflow)
1344            });
1345        let (generation, successor_deadline_ns) = match next_dispatch {
1346            Ok(next) => next,
1347            Err(failure) => {
1348                let transition = control.terminate(
1349                    RegistryEffect::ClearCallbacks {
1350                        handles: CallbacksToClear::Work,
1351                    },
1352                    failure,
1353                );
1354                let remove = entry.remove_on_failure(&transition);
1355                return remove_after(&mut self.entries, token.identity(), transition, remove);
1356            }
1357        };
1358
1359        control.generation = generation;
1360        control.state = WatchdogState::AwaitingWork {
1361            successor_deadline_ns,
1362            attempt_status: WatchdogAttemptStatus::Dispatched,
1363            pending: None,
1364        };
1365        entry.scheduling_mode = TimerSchedulingMode::Watchdog;
1366        RegistryTransition::normal(RegistryEffect::DispatchWatchdog {
1367            dispatch: WatchdogDispatch::new(token.claim(), generation),
1368            successor_delay_ns: cadence.as_nanos(),
1369        })
1370    }
1371
1372    /// Consume this delivery's work handle, then accept Watchdog work and return
1373    /// its callback.
1374    pub(crate) fn begin_watchdog_work(
1375        &mut self,
1376        token: &CallbackToken,
1377    ) -> Option<WatchdogCallback> {
1378        let entry = self.entries.get_mut(token.identity())?;
1379        entry.consume_provider_handle(token);
1380        if token.role != CallbackRole::WatchdogWork || !entry.owns_token_claim(token) {
1381            entry.observability.counters_mut().record_stale_work();
1382            return None;
1383        }
1384        let EntryKind::Watchdog {
1385            control, callback, ..
1386        } = &mut entry.kind
1387        else {
1388            entry.observability.counters_mut().record_stale_work();
1389            return None;
1390        };
1391        match &mut control.state {
1392            WatchdogState::AwaitingWork { attempt_status, .. }
1393                if control.generation == token.callback_generation
1394                    && *attempt_status == WatchdogAttemptStatus::Dispatched =>
1395            {
1396                *attempt_status = WatchdogAttemptStatus::Running;
1397                entry.observability.counters_mut().record_work_started();
1398                Some(Rc::clone(callback))
1399            }
1400            WatchdogState::Inactive { .. }
1401            | WatchdogState::Scheduled { .. }
1402            | WatchdogState::AwaitingWork { .. } => {
1403                entry.observability.counters_mut().record_stale_work();
1404                None
1405            }
1406        }
1407    }
1408
1409    #[expect(
1410        clippy::too_many_lines,
1411        reason = "One atomic successor and request-arbitration transition."
1412    )]
1413    pub(crate) fn complete_watchdog_work(
1414        &mut self,
1415        token: &CallbackToken,
1416        now_ns: u64,
1417        result: WatchdogRunResult,
1418    ) -> Result<RegistryTransition, RegistryError> {
1419        let identity = token.identity();
1420        let (transition, remove) = {
1421            let entry = self.running_work_entry_mut(token)?;
1422            let EntryKind::Watchdog {
1423                cadence, control, ..
1424            } = &mut entry.kind
1425            else {
1426                return Err(RegistryError::StaleCallback);
1427            };
1428            let cadence = *cadence;
1429            let WatchdogState::AwaitingWork {
1430                successor_deadline_ns,
1431                pending,
1432                ..
1433            } = control.state
1434            else {
1435                return Err(RegistryError::StaleCallback);
1436            };
1437
1438            let completion = result.completion();
1439            entry.observability.record_completion(completion, now_ns);
1440            let decision = if completion.outcome() == TimerCompletionOutcome::InvariantFailure {
1441                WatchdogDecision::Stop
1442            } else {
1443                match pending {
1444                    Some(WatchdogPending::Cancel | WatchdogPending::Unregister) => {
1445                        WatchdogDecision::Stop
1446                    }
1447                    Some(WatchdogPending::Ensure) => WatchdogDecision::Continue,
1448                    Some(WatchdogPending::EnsureImmediately) => {
1449                        WatchdogDecision::ContinueImmediately
1450                    }
1451                    Some(WatchdogPending::Reconcile(requested)) => {
1452                        WatchdogDecision::ScheduleAt(requested.deadline_ns)
1453                    }
1454                    None => result.decision(),
1455                }
1456            };
1457            let cancelled = matches!(pending, Some(WatchdogPending::Cancel));
1458            let pending_schedule = match pending {
1459                Some(WatchdogPending::Reconcile(requested)) => Some(requested),
1460                _ => None,
1461            };
1462
1463            let transition = match decision {
1464                WatchdogDecision::Continue => {
1465                    control.state = WatchdogState::Scheduled {
1466                        deadline_ns: successor_deadline_ns,
1467                    };
1468                    entry.scheduling_mode = TimerSchedulingMode::Watchdog;
1469                    entry.latest_requested_delay_ns = Some(cadence.as_nanos());
1470                    RegistryTransition::normal(RegistryEffect::None)
1471                }
1472                WatchdogDecision::ContinueImmediately | WatchdogDecision::ScheduleAt(_) => {
1473                    let (deadline_ns, retain_successor) =
1474                        if let WatchdogDecision::ScheduleAt(deadline_ns) = decision {
1475                            entry.scheduling_mode = pending_schedule
1476                                .map_or(TimerSchedulingMode::Deadline, |requested| requested.mode);
1477                            entry.latest_requested_delay_ns =
1478                                pending_schedule.and_then(|requested| requested.requested_delay_ns);
1479                            (deadline_ns, successor_deadline_ns == deadline_ns)
1480                        } else {
1481                            entry.scheduling_mode = TimerSchedulingMode::Continuation;
1482                            entry.latest_requested_delay_ns = Some(0);
1483                            (now_ns, successor_deadline_ns <= now_ns)
1484                        };
1485                    if retain_successor {
1486                        control.state = WatchdogState::Scheduled {
1487                            deadline_ns: successor_deadline_ns,
1488                        };
1489                        RegistryTransition::normal(RegistryEffect::None)
1490                    } else {
1491                        control.arm_scheduler(
1492                            token.claim(),
1493                            now_ns,
1494                            deadline_ns,
1495                            WakeupArm::Replacement,
1496                        )
1497                    }
1498                }
1499                WatchdogDecision::Stop => {
1500                    let reason = if completion.outcome() == TimerCompletionOutcome::InvariantFailure
1501                    {
1502                        InactiveReason::InvariantFailure
1503                    } else if cancelled {
1504                        entry.observability.counters_mut().record_cancellation();
1505                        InactiveReason::Cancelled
1506                    } else {
1507                        InactiveReason::Stopped
1508                    };
1509                    control.state = WatchdogState::Inactive { reason };
1510                    RegistryTransition::normal(RegistryEffect::ClearCallbacks {
1511                        handles: CallbacksToClear::Wakeup,
1512                    })
1513                }
1514            };
1515            let remove = matches!(control.state, WatchdogState::Inactive { .. })
1516                && (matches!(pending, Some(WatchdogPending::Unregister))
1517                    || matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped));
1518            (transition, remove)
1519        };
1520
1521        Ok(remove_after(
1522            &mut self.entries,
1523            identity,
1524            transition,
1525            remove,
1526        ))
1527    }
1528
1529    pub(crate) fn snapshot(&self, identity: &TimerIdentity) -> Option<TimerSnapshot> {
1530        self.entries
1531            .get(identity)
1532            .map(|entry| entry.snapshot(identity.clone(), self.epoch))
1533    }
1534
1535    pub(crate) fn inventory(&self) -> TimerInventorySnapshot {
1536        TimerInventorySnapshot::new(
1537            self.epoch,
1538            self.entries
1539                .iter()
1540                .map(|(identity, entry)| entry.snapshot(identity.clone(), self.epoch))
1541                .collect(),
1542        )
1543    }
1544
1545    pub(crate) fn consecutive_expected_failures(&self, identity: &TimerIdentity) -> Option<u64> {
1546        self.entries.get(identity).map(|entry| {
1547            entry
1548                .observability
1549                .outcomes()
1550                .consecutive_expected_failures()
1551        })
1552    }
1553
1554    pub(crate) fn has_armed_wakeup(
1555        &self,
1556        claim: &RegistrationClaim,
1557    ) -> Result<bool, RegistryError> {
1558        Ok(self.entry(claim)?.wakeup.is_some())
1559    }
1560
1561    pub(crate) fn declaration_matches(
1562        &self,
1563        claim: &RegistrationClaim,
1564        policy: TimerPolicy,
1565        lifetime: DeclarationLifetime,
1566    ) -> Result<bool, RegistryError> {
1567        let entry = self.entry(claim)?;
1568        Ok(entry.kind.policy() == policy && entry.lifetime == lifetime)
1569    }
1570
1571    pub(crate) fn record_callback_measurements(
1572        &mut self,
1573        token: &CallbackToken,
1574        instructions: u64,
1575        memory_start: MemoryPageExtent,
1576        memory_end: MemoryPageExtent,
1577    ) -> Result<(), RegistryError> {
1578        // A normal remove-on-stop completion can delete its entry before the
1579        // post-run measurement is committed, leaving nothing to observe.
1580        // Identity reuse must not let a late callback write into a newer
1581        // registration's observations.
1582        let Ok(entry) = self.entry_mut(token.claim()) else {
1583            return Ok(());
1584        };
1585
1586        let memory = MemoryPageSample::new(memory_start, memory_end);
1587        match (&entry.kind, token.role) {
1588            (EntryKind::Watchdog { .. }, CallbackRole::WatchdogScheduler) => entry
1589                .observability
1590                .record_scheduler_measurements(instructions, memory),
1591            (EntryKind::Ordinary { .. }, CallbackRole::OrdinaryWork)
1592            | (EntryKind::Watchdog { .. }, CallbackRole::WatchdogWork) => entry
1593                .observability
1594                .record_work_measurements(instructions, memory),
1595            _ => {
1596                return Err(RegistryError::PolicyMismatch {
1597                    actual: entry.kind.policy().label(),
1598                });
1599            }
1600        }
1601        Ok(())
1602    }
1603
1604    pub(crate) fn fail_registration(
1605        &mut self,
1606        claim: &RegistrationClaim,
1607        failure: TimerControlFailure,
1608    ) -> Result<(ProviderHandles, RegistryTransition), RegistryError> {
1609        let identity = claim.identity();
1610        let (handles, remove) = {
1611            let entry = self.entry_mut(claim)?;
1612            match &mut entry.kind {
1613                EntryKind::Ordinary { control, .. } => {
1614                    control.terminate(InactiveReason::ControlFailure(failure));
1615                }
1616                EntryKind::Watchdog { control, .. } => {
1617                    control.state = WatchdogState::Inactive {
1618                        reason: InactiveReason::ControlFailure(failure),
1619                    };
1620                }
1621            }
1622            let handles = entry.take_provider_handles(identity);
1623            (
1624                handles,
1625                matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped),
1626            )
1627        };
1628        let transition = remove_after(
1629            &mut self.entries,
1630            identity,
1631            RegistryTransition::normal(RegistryEffect::None),
1632            remove,
1633        );
1634        Ok((handles, transition))
1635    }
1636
1637    pub(crate) fn validate_running_context(
1638        &self,
1639        token: &CallbackToken,
1640    ) -> Result<(), RegistryError> {
1641        let entry = self
1642            .entries
1643            .get(token.identity())
1644            .ok_or(RegistryError::StaleCallback)?;
1645        if entry.owns_running_work(token) {
1646            Ok(())
1647        } else {
1648            Err(RegistryError::StaleCallback)
1649        }
1650    }
1651
1652    /// Bind a handle and, for the final arm, commit its successful-arm facts.
1653    /// A missing delay restores ownership or stages a Watchdog successor only.
1654    pub(crate) fn install_provider_handle(
1655        &mut self,
1656        token: &CallbackToken,
1657        handle: TimerHandle,
1658        armed_delay_ns: Option<u64>,
1659    ) -> Result<(), (RegistryError, TimerHandle)> {
1660        let entry = match self.entry_by_token_mut(token) {
1661            Ok(entry) => entry,
1662            Err(error) => return Err((error, handle)),
1663        };
1664        let slot = match entry.provider_handle_slot(token, armed_delay_ns.is_some()) {
1665            Ok(slot) => slot,
1666            Err(error) => return Err((error, handle)),
1667        };
1668        if slot.is_some() {
1669            return Err((RegistryError::ProviderHandleAlreadyOwned, handle));
1670        }
1671        *slot = Some(OwnedProviderHandle {
1672            callback_generation: token.callback_generation,
1673            handle,
1674        });
1675        if let Some(delay_ns) = armed_delay_ns {
1676            entry.record_arm(token, delay_ns);
1677        }
1678        Ok(())
1679    }
1680
1681    #[cfg(test)]
1682    pub(crate) fn take_wakeup_handle(
1683        &mut self,
1684        identity: &TimerIdentity,
1685    ) -> Option<ProviderHandle> {
1686        let entry = self.entries.get_mut(identity)?;
1687        entry.take_wakeup_handle(identity)
1688    }
1689
1690    pub(crate) fn take_provider_handles(&mut self, identity: &TimerIdentity) -> ProviderHandles {
1691        self.entries
1692            .get_mut(identity)
1693            .map_or_else(ProviderHandles::default, |entry| {
1694                entry.take_provider_handles(identity)
1695            })
1696    }
1697
1698    pub(crate) fn take_provider_handles_for_claim(
1699        &mut self,
1700        claim: &RegistrationClaim,
1701    ) -> Result<ProviderHandles, RegistryError> {
1702        let identity = claim.identity();
1703        let entry = self.entry_mut(claim)?;
1704        Ok(entry.take_provider_handles(identity))
1705    }
1706
1707    /// Consume the delivered scheduler handle before detaching the remaining
1708    /// capabilities. A scheduler transition can remove a transient declaration.
1709    pub(crate) fn take_watchdog_scheduler_handles(
1710        &mut self,
1711        token: &CallbackToken,
1712    ) -> Result<ProviderHandles, RegistryError> {
1713        let entry = match self.entry_mut(token.claim()) {
1714            Ok(entry) => entry,
1715            // Removed or superseded claims are normal stale callback delivery.
1716            Err(RegistryError::UnknownRegistration | RegistryError::StaleRegistration) => {
1717                return Ok(ProviderHandles::default());
1718            }
1719            Err(error) => return Err(error),
1720        };
1721        entry.consume_provider_handle(token);
1722        Ok(entry.take_provider_handles(token.identity()))
1723    }
1724
1725    /// Model successful platform application in provider-neutral transition tests.
1726    #[cfg(test)]
1727    pub(crate) fn confirm_effect_applied(
1728        &mut self,
1729        effect: &RegistryEffect,
1730    ) -> Result<(), RegistryError> {
1731        let (token, delay_ns) = match effect {
1732            RegistryEffect::None | RegistryEffect::ClearCallbacks { .. } => return Ok(()),
1733            RegistryEffect::ArmWakeup {
1734                token, delay_ns, ..
1735            } => (token.token().clone(), *delay_ns),
1736            RegistryEffect::DispatchWatchdog {
1737                dispatch,
1738                successor_delay_ns,
1739            } => (dispatch.work(), *successor_delay_ns),
1740        };
1741        let entry = self.entry_by_token_mut(&token)?;
1742        entry.provider_handle_slot(&token, true)?;
1743        entry.record_arm(&token, delay_ns);
1744        Ok(())
1745    }
1746
1747    fn entry(&self, claim: &RegistrationClaim) -> Result<&Entry, RegistryError> {
1748        let entry = self
1749            .entries
1750            .get(claim.identity())
1751            .ok_or(RegistryError::UnknownRegistration)?;
1752        if entry.claim_generation != claim.claim_generation() {
1753            return Err(RegistryError::StaleRegistration);
1754        }
1755        Ok(entry)
1756    }
1757
1758    fn entry_mut(&mut self, claim: &RegistrationClaim) -> Result<&mut Entry, RegistryError> {
1759        let entry = self
1760            .entries
1761            .get_mut(claim.identity())
1762            .ok_or(RegistryError::UnknownRegistration)?;
1763        if entry.claim_generation != claim.claim_generation() {
1764            return Err(RegistryError::StaleRegistration);
1765        }
1766        Ok(entry)
1767    }
1768
1769    // Select identity and exact claim; callers validate the role and generation
1770    // against their operation's policy state.
1771    fn entry_by_token_mut(&mut self, token: &CallbackToken) -> Result<&mut Entry, RegistryError> {
1772        // Missing and superseded claims are both stale callback delivery.
1773        self.entry_mut(token.claim())
1774            .map_err(|_| RegistryError::StaleCallback)
1775    }
1776
1777    fn running_work_entry_mut(
1778        &mut self,
1779        token: &CallbackToken,
1780    ) -> Result<&mut Entry, RegistryError> {
1781        let entry = self
1782            .entries
1783            .get_mut(token.identity())
1784            .ok_or(RegistryError::StaleCallback)?;
1785        if entry.owns_running_work(token) {
1786            Ok(entry)
1787        } else {
1788            Err(RegistryError::StaleCallback)
1789        }
1790    }
1791}
1792
1793fn token_for(
1794    claim: &RegistrationClaim,
1795    callback_generation: u64,
1796    role: CallbackRole,
1797) -> CallbackToken {
1798    CallbackToken::new(
1799        claim.identity.clone(),
1800        claim.claim_generation,
1801        callback_generation,
1802        role,
1803    )
1804}
1805
1806fn detach_provider_handle(
1807    identity: &TimerIdentity,
1808    claim_generation: u64,
1809    role: CallbackRole,
1810    owned: OwnedProviderHandle,
1811) -> ProviderHandle {
1812    ProviderHandle {
1813        token: CallbackToken::new(
1814            identity.clone(),
1815            claim_generation,
1816            owned.callback_generation,
1817            role,
1818        ),
1819        handle: owned.handle,
1820    }
1821}
1822
1823const fn clear_wakeup_if(clear_wakeup: bool) -> RegistryEffect {
1824    if clear_wakeup {
1825        RegistryEffect::ClearCallbacks {
1826            handles: CallbacksToClear::Wakeup,
1827        }
1828    } else {
1829        RegistryEffect::None
1830    }
1831}
1832
1833// Callers have validated the exact running generation. Checked completion
1834// failures leave that state unchanged before this shared stop finalization.
1835fn stop_ordinary_completion(
1836    entry: &mut Entry,
1837    completion: TimerCompletion,
1838    now_ns: u64,
1839    failure: Option<TimerControlFailure>,
1840) -> RegistryTransition {
1841    let (reason, completion, transition) = failure.map_or_else(
1842        || {
1843            (
1844                InactiveReason::InvariantFailure,
1845                completion,
1846                RegistryTransition::normal(RegistryEffect::None),
1847            )
1848        },
1849        |failure| {
1850            (
1851                InactiveReason::ControlFailure(failure),
1852                TimerCompletion::invariant_failure(completion.work_count()),
1853                RegistryTransition::terminal(RegistryEffect::None, failure),
1854            )
1855        },
1856    );
1857    let EntryKind::Ordinary { control, .. } = &mut entry.kind else {
1858        return transition;
1859    };
1860    control.terminate(reason);
1861    entry.latest_directive = Some(TimerDirectiveSnapshot::Stop);
1862    entry.observability.record_completion(completion, now_ns);
1863    transition
1864}
1865
1866// Select authoritative commands before validating a callback proposal they discard.
1867// Keep the callback observation when a pending ensure wins; reconciliation instead
1868// reports its exact deadline while retaining the already checked schedule metadata.
1869fn prepare_ordinary_completion(
1870    pending: Option<OrdinaryPending>,
1871    directive: OrdinaryDirective,
1872    now_ns: u64,
1873    cadence: Option<TimerCadence>,
1874) -> Result<(Option<ResolvedSchedule>, TimerDirectiveSnapshot, bool), TimerControlFailure> {
1875    match pending {
1876        Some(OrdinaryPending::Cancel | OrdinaryPending::Unregister) => {
1877            Ok((None, TimerDirectiveSnapshot::Stop, true))
1878        }
1879        Some(OrdinaryPending::Reconcile(pending)) => Ok((
1880            Some(pending),
1881            TimerDirectiveSnapshot::ScheduleAt {
1882                deadline_ns: pending.deadline_ns,
1883            },
1884            false,
1885        )),
1886        Some(OrdinaryPending::Schedule(_)) | None => {
1887            let (callback, snapshot) = directive.resolve(now_ns, cadence)?;
1888            let selected = if let Some(OrdinaryPending::Schedule(pending)) = pending {
1889                match callback {
1890                    Some(callback) if callback.deadline_ns < pending.deadline_ns => Some(callback),
1891                    Some(_) | None => Some(pending),
1892                }
1893            } else {
1894                callback
1895            };
1896            Ok((selected, snapshot, false))
1897        }
1898    }
1899}
1900
1901const fn select_pending_ordinary(
1902    current: Option<OrdinaryPending>,
1903    request: OrdinaryRequest,
1904    requested: ResolvedSchedule,
1905) -> OrdinaryPending {
1906    match (current, request) {
1907        (Some(OrdinaryPending::Unregister), _) => OrdinaryPending::Unregister,
1908        (_, OrdinaryRequest::Reconcile) => OrdinaryPending::Reconcile(requested),
1909        (
1910            Some(OrdinaryPending::Reconcile(current) | OrdinaryPending::Schedule(current)),
1911            OrdinaryRequest::Ensure,
1912        ) if current.deadline_ns <= requested.deadline_ns => OrdinaryPending::Schedule(current),
1913        (_, OrdinaryRequest::Ensure) => OrdinaryPending::Schedule(requested),
1914    }
1915}
1916
1917const fn select_pending_watchdog(
1918    pending: Option<WatchdogPending>,
1919    request: WatchdogScheduleRequest,
1920    now_ns: u64,
1921) -> WatchdogPending {
1922    match (pending, request) {
1923        (Some(WatchdogPending::Unregister), _) => WatchdogPending::Unregister,
1924        (_, WatchdogScheduleRequest::Reconcile(requested)) => WatchdogPending::Reconcile(requested),
1925        (Some(WatchdogPending::Reconcile(requested)), WatchdogScheduleRequest::Cadence) => {
1926            WatchdogPending::Reconcile(requested)
1927        }
1928        (Some(WatchdogPending::Reconcile(requested)), WatchdogScheduleRequest::Immediate)
1929            if requested.deadline_ns <= now_ns =>
1930        {
1931            WatchdogPending::Reconcile(requested)
1932        }
1933        (Some(WatchdogPending::EnsureImmediately), _) | (_, WatchdogScheduleRequest::Immediate) => {
1934            WatchdogPending::EnsureImmediately
1935        }
1936        (
1937            Some(WatchdogPending::Cancel | WatchdogPending::Ensure) | None,
1938            WatchdogScheduleRequest::Cadence,
1939        ) => WatchdogPending::Ensure,
1940    }
1941}
1942
1943fn remove_after(
1944    entries: &mut BTreeMap<TimerIdentity, Entry>,
1945    identity: &TimerIdentity,
1946    mut transition: RegistryTransition,
1947    remove: bool,
1948) -> RegistryTransition {
1949    if remove && let Some(entry) = entries.remove(identity) {
1950        transition.removed = Some(Box::new(entry));
1951    }
1952    transition
1953}
1954
1955#[cfg(test)]
1956mod tests;