1mod 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
29pub const MAX_TIMER_REGISTRATIONS: usize = 64;
31
32#[derive(Debug, Eq, PartialEq)]
34pub struct RegistrationClaim {
35 identity: TimerIdentity,
36 claim_generation: u64,
37}
38
39#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub enum CallbackRole {
42 OrdinaryWork,
43 WatchdogScheduler,
44 WatchdogWork,
45}
46
47#[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 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#[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
131pub struct RegistryTransition {
133 effect: RegistryEffect,
134 failure: Option<TimerControlFailure>,
135 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#[non_exhaustive]
184#[derive(Clone, Debug, Eq, Error, PartialEq)]
185pub enum RegisterError {
186 #[error("timer identity is already registered: {0:?}")]
188 IdentityAlreadyRegistered(TimerIdentity),
189 #[error("timer registry capacity of {max} declarations has been reached")]
191 CapacityExceeded {
192 max: usize,
194 },
195 #[error("timer registration claim generation exhausted")]
197 ClaimGenerationExhausted,
198}
199
200#[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#[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#[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
279enum 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 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 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 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 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 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
637pub 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 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 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 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 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 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 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 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 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 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 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 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 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 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 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 #[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 fn entry_by_token_mut(&mut self, token: &CallbackToken) -> Result<&mut Entry, RegistryError> {
1772 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
1833fn 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
1866fn 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;