1use crate::{
8 control::{TimerControl, TimerControlAction, TimerControlError, TimerRegistration, WakeupArm},
9 platform::{MemoryPages, TimerHandle},
10 schedule::{DirectiveError, ScheduleError, TimerCadence, TimerDirective, TimerSchedule},
11 snapshot::{
12 DeclarationLifetime, InactiveReason, MemoryPageExtent, MemoryPageSample,
13 OrdinaryRuntimeStateSnapshot, TimerCompletion, TimerCompletionOutcome, TimerControlFailure,
14 TimerDirectiveSnapshot, TimerEpoch, TimerIdentity, TimerInventorySnapshot,
15 TimerObservabilitySnapshot, TimerPolicy, TimerRegistrationId, TimerRunResult,
16 TimerRuntimeStateSnapshot, TimerSchedulingMode, TimerSnapshot, WatchdogAttemptSnapshot,
17 WatchdogAttemptStatus, WatchdogDecision, WatchdogRunResult, WatchdogRuntimeStateSnapshot,
18 },
19};
20use std::{cell::RefCell, collections::BTreeMap, future::Future, pin::Pin, rc::Rc};
21use thiserror::Error;
22
23pub const MAX_TIMER_REGISTRATIONS: usize = 64;
25
26#[derive(Debug, Eq, PartialEq)]
28pub struct RegistrationClaim {
29 identity: TimerIdentity,
30 claim_generation: u64,
31}
32
33#[derive(Clone, Copy, Debug, Eq, PartialEq)]
35pub enum CallbackRole {
36 OrdinaryWork,
37 WatchdogScheduler,
38 WatchdogWork,
39}
40
41#[derive(Clone, Debug, Eq, PartialEq)]
43pub struct CallbackToken {
44 identity: TimerIdentity,
45 claim_generation: u64,
46 callback_generation: u64,
47 role: CallbackRole,
48}
49
50impl CallbackToken {
51 const fn new(
52 identity: TimerIdentity,
53 claim_generation: u64,
54 callback_generation: u64,
55 role: CallbackRole,
56 ) -> Self {
57 Self {
58 identity,
59 claim_generation,
60 callback_generation,
61 role,
62 }
63 }
64
65 pub(crate) const fn identity(&self) -> &TimerIdentity {
66 &self.identity
67 }
68
69 #[cfg(test)]
70 pub(crate) const fn callback_generation(&self) -> u64 {
71 self.callback_generation
72 }
73
74 pub(crate) const fn role(&self) -> CallbackRole {
75 self.role
76 }
77
78 pub(crate) fn belongs_to_same_claim(&self, other: &Self) -> bool {
79 self.identity == other.identity && self.claim_generation == other.claim_generation
80 }
81}
82
83impl RegistrationClaim {
84 pub(crate) const fn identity(&self) -> &TimerIdentity {
85 &self.identity
86 }
87
88 pub(crate) const fn claim_generation(&self) -> u64 {
89 self.claim_generation
90 }
91
92 pub(crate) fn from_callback(token: &CallbackToken) -> Self {
93 Self {
94 identity: token.identity.clone(),
95 claim_generation: token.claim_generation,
96 }
97 }
98}
99
100#[derive(Debug, Eq, PartialEq)]
102pub enum RegistryEffect {
103 None,
104 ArmWakeup {
105 token: CallbackToken,
106 deadline_ns: u64,
107 delay_ns: u64,
108 arm: WakeupArm,
109 },
110 ClearCallbacks {
111 identity: TimerIdentity,
112 handles: CallbacksToClear,
113 },
114 DispatchWatchdog {
115 successor: CallbackToken,
116 successor_deadline_ns: u64,
117 successor_delay_ns: u64,
118 work: CallbackToken,
119 },
120}
121
122#[derive(Clone, Copy, Debug, Eq, PartialEq)]
124pub enum CallbacksToClear {
125 Wakeup,
126 Work,
127 WakeupAndWork,
128}
129
130impl CallbacksToClear {
131 pub(crate) const fn includes_wakeup(self) -> bool {
132 matches!(self, Self::Wakeup | Self::WakeupAndWork)
133 }
134
135 pub(crate) const fn includes_work(self) -> bool {
136 matches!(self, Self::Work | Self::WakeupAndWork)
137 }
138
139 const fn wakeup_and_maybe_work(include_work: bool) -> Self {
140 if include_work {
141 Self::WakeupAndWork
142 } else {
143 Self::Wakeup
144 }
145 }
146}
147
148impl RegistryEffect {
149 pub(crate) fn has_valid_shape(&self) -> bool {
150 match self {
151 Self::None | Self::ClearCallbacks { .. } => true,
152 Self::ArmWakeup { token, arm, .. } => match token.role {
153 CallbackRole::OrdinaryWork => true,
154 CallbackRole::WatchdogScheduler => {
155 matches!(arm, WakeupArm::Initial | WakeupArm::Replacement)
156 }
157 CallbackRole::WatchdogWork => false,
158 },
159 Self::DispatchWatchdog {
160 successor, work, ..
161 } => {
162 successor.role == CallbackRole::WatchdogScheduler
163 && work.role == CallbackRole::WatchdogWork
164 && successor.belongs_to_same_claim(work)
165 }
166 }
167 }
168}
169
170#[derive(Debug, Eq, PartialEq)]
172pub struct RegistryTransition {
173 effect: RegistryEffect,
174 failure: Option<TimerControlFailure>,
175}
176
177impl RegistryTransition {
178 const fn normal(effect: RegistryEffect) -> Self {
179 Self {
180 effect,
181 failure: None,
182 }
183 }
184
185 const fn terminal(effect: RegistryEffect, failure: TimerControlFailure) -> Self {
186 Self {
187 effect,
188 failure: Some(failure),
189 }
190 }
191
192 pub(crate) const fn effect(&self) -> &RegistryEffect {
193 &self.effect
194 }
195
196 pub(crate) const fn failure(&self) -> Option<TimerControlFailure> {
197 self.failure
198 }
199
200 pub(crate) fn into_effect(self) -> RegistryEffect {
201 self.effect
202 }
203}
204
205#[derive(Clone, Copy, Debug, Eq, PartialEq)]
207pub enum CallbackAcceptance {
208 Accepted,
209 Stale,
210}
211
212#[non_exhaustive]
214#[derive(Clone, Debug, Eq, Error, PartialEq)]
215pub enum RegisterError {
216 #[error("timer identity is already registered: {0:?}")]
218 IdentityAlreadyRegistered(TimerIdentity),
219 #[error("timer registry capacity of {max} declarations has been reached")]
221 CapacityExceeded {
222 max: usize,
224 },
225 #[error("timer registration claim generation exhausted")]
227 ClaimGenerationExhausted,
228}
229
230#[derive(Clone, Debug, Eq, Error, PartialEq)]
232pub enum RegistryError {
233 #[error("timer registration is no longer present")]
234 UnknownRegistration,
235 #[error("timer registration claim is stale")]
236 StaleRegistration,
237 #[error("timer policy invariant failed for {actual}")]
238 PolicyMismatch { actual: &'static str },
239 #[error("timer callback token is stale")]
240 StaleCallback,
241 #[error("timer registration has no ordinary callback")]
242 MissingCallback,
243 #[error("timer registration already owns the provider handle for this callback role")]
244 ProviderHandleAlreadyOwned,
245 #[error(transparent)]
246 Schedule(#[from] ScheduleError),
247}
248
249type OrdinaryFuture = Pin<Box<dyn Future<Output = TimerRunResult>>>;
250pub type OrdinaryCallback = Rc<RefCell<Box<dyn FnMut(CallbackToken) -> OrdinaryFuture>>>;
251pub type WatchdogCallback = Rc<RefCell<Box<dyn FnMut(CallbackToken) -> WatchdogRunResult>>>;
252
253enum EntryCallback {
254 #[cfg(test)]
255 None,
256 Ordinary(OrdinaryCallback),
257 Watchdog(WatchdogCallback),
258}
259
260struct OwnedProviderHandle {
261 callback_generation: u64,
262 role: CallbackRole,
263 handle: TimerHandle,
264}
265
266#[must_use = "restore or clear the detached provider handle"]
268pub struct ProviderHandle {
269 token: CallbackToken,
270 handle: TimerHandle,
271}
272
273impl ProviderHandle {
274 pub(crate) fn into_parts(self) -> (CallbackToken, TimerHandle) {
275 (self.token, self.handle)
276 }
277}
278
279#[must_use = "restore or clear every detached provider handle"]
281#[derive(Default)]
282pub struct ProviderHandles {
283 wakeup: Option<ProviderHandle>,
284 work: Option<ProviderHandle>,
285}
286
287impl ProviderHandles {
288 pub(crate) const fn from_parts(
289 wakeup: Option<ProviderHandle>,
290 work: Option<ProviderHandle>,
291 ) -> Self {
292 Self { wakeup, work }
293 }
294
295 pub(crate) const fn take_wakeup(&mut self) -> Option<ProviderHandle> {
296 self.wakeup.take()
297 }
298
299 pub(crate) const fn take_work(&mut self) -> Option<ProviderHandle> {
300 self.work.take()
301 }
302}
303
304#[derive(Clone, Copy, Debug, Eq, PartialEq)]
305struct PendingSchedule {
306 deadline_ns: u64,
307 requested_delay_ns: Option<u64>,
308 mode: TimerSchedulingMode,
309}
310
311impl PendingSchedule {
312 fn resolve(schedule: TimerSchedule, now_ns: u64) -> Result<Self, ScheduleError> {
313 let resolved = schedule.resolve(now_ns)?;
314 Ok(Self {
315 deadline_ns: resolved.deadline_ns,
316 requested_delay_ns: resolved.requested_delay_ns,
317 mode: match schedule {
318 TimerSchedule::After(_) => TimerSchedulingMode::Once,
319 TimerSchedule::At(_) => TimerSchedulingMode::Deadline,
320 },
321 })
322 }
323}
324
325#[derive(Clone, Copy, Debug, Eq, PartialEq)]
326enum OrdinaryPending {
327 Cancel,
328 Reconcile(PendingSchedule),
329 Unregister,
330 Schedule(PendingSchedule),
331}
332
333#[derive(Clone, Copy, Debug, Eq, PartialEq)]
334enum OrdinaryRequest {
335 EnsureOnce,
336 EnsureRecurring,
337 Reconcile,
338}
339
340#[derive(Clone, Copy, Debug, Eq, PartialEq)]
341enum WatchdogPending {
342 Cancel,
343 Ensure,
344 EnsureImmediately,
345 Reconcile(PendingSchedule),
346 Unregister,
347}
348
349#[derive(Clone, Copy, Debug, Eq, PartialEq)]
350enum WatchdogScheduleRequest {
351 Cadence,
352 Immediate,
353 Reconcile(PendingSchedule),
354}
355
356#[derive(Debug)]
357enum EntryControl {
358 Ordinary {
359 control: TimerControl,
360 pending: Option<OrdinaryPending>,
361 inactive_reason: InactiveReason,
362 },
363 Watchdog(WatchdogControl),
364}
365
366#[derive(Clone, Copy, Debug, Eq, PartialEq)]
367enum WatchdogState {
368 Inactive,
369 Scheduled {
370 scheduler_generation: u64,
371 deadline_ns: u64,
372 },
373 AwaitingWork {
374 successor_generation: u64,
375 successor_deadline_ns: u64,
376 attempt_generation: u64,
377 attempt_status: WatchdogAttemptStatus,
378 },
379}
380
381#[derive(Debug)]
382struct WatchdogControl {
383 scheduler_generation: u64,
384 attempt_generation: u64,
385 state: WatchdogState,
386 pending: Option<WatchdogPending>,
387 inactive_reason: InactiveReason,
388}
389
390impl Default for WatchdogControl {
391 fn default() -> Self {
392 Self {
393 scheduler_generation: 0,
394 attempt_generation: 0,
395 state: WatchdogState::Inactive,
396 pending: None,
397 inactive_reason: InactiveReason::NeverScheduled,
398 }
399 }
400}
401
402impl WatchdogControl {
403 fn arm_scheduler(
404 &mut self,
405 claim: &RegistrationClaim,
406 now_ns: u64,
407 deadline_ns: u64,
408 arm: WakeupArm,
409 ) -> RegistryTransition {
410 let Some(generation) = self.scheduler_generation.checked_add(1) else {
411 let cleanup = if arm.replaces_existing() {
412 clear_callbacks(claim.identity.clone(), CallbacksToClear::Wakeup)
413 } else {
414 RegistryEffect::None
415 };
416 return self.terminate(cleanup, TimerControlFailure::GenerationExhausted);
417 };
418 self.scheduler_generation = generation;
419 self.state = WatchdogState::Scheduled {
420 scheduler_generation: generation,
421 deadline_ns,
422 };
423 self.pending = None;
424 RegistryTransition::normal(RegistryEffect::ArmWakeup {
425 token: token_for(claim, generation, CallbackRole::WatchdogScheduler),
426 deadline_ns,
427 delay_ns: deadline_ns.saturating_sub(now_ns),
428 arm,
429 })
430 }
431
432 const fn next_dispatch_generations(&self) -> Option<(u64, u64)> {
433 let Some(scheduler_generation) = self.scheduler_generation.checked_add(1) else {
434 return None;
435 };
436 let Some(attempt_generation) = self.attempt_generation.checked_add(1) else {
437 return None;
438 };
439 Some((scheduler_generation, attempt_generation))
440 }
441
442 const fn terminate(
443 &mut self,
444 effect: RegistryEffect,
445 failure: TimerControlFailure,
446 ) -> RegistryTransition {
447 self.state = WatchdogState::Inactive;
448 self.pending = None;
449 self.inactive_reason = InactiveReason::ControlFailure(failure);
450 RegistryTransition::terminal(effect, failure)
451 }
452}
453
454struct Entry {
455 claim_generation: u64,
456 policy: TimerPolicy,
457 lifetime: DeclarationLifetime,
458 control: EntryControl,
459 scheduling_mode: TimerSchedulingMode,
460 latest_directive: Option<TimerDirectiveSnapshot>,
461 latest_requested_delay_ns: Option<u64>,
462 latest_armed_delay_ns: Option<u64>,
463 confirmed_wakeup_generation: Option<u64>,
464 confirmed_work_generation: Option<u64>,
465 callback: EntryCallback,
466 wakeup: Option<OwnedProviderHandle>,
467 work: Option<OwnedProviderHandle>,
468 observability: TimerObservabilitySnapshot,
469}
470
471impl Entry {
472 fn new(
473 claim_generation: u64,
474 policy: TimerPolicy,
475 lifetime: DeclarationLifetime,
476 epoch: TimerEpoch,
477 callback: EntryCallback,
478 ) -> Self {
479 let control = match policy {
480 TimerPolicy::Once | TimerPolicy::AfterCompletion { .. } => EntryControl::Ordinary {
481 control: TimerControl::default(),
482 pending: None,
483 inactive_reason: InactiveReason::NeverScheduled,
484 },
485 TimerPolicy::Watchdog { .. } => EntryControl::Watchdog(WatchdogControl::default()),
486 };
487 let scheduling_mode = match policy {
488 TimerPolicy::Once => TimerSchedulingMode::Once,
489 TimerPolicy::AfterCompletion { .. } => TimerSchedulingMode::AfterCompletion,
490 TimerPolicy::Watchdog { .. } => TimerSchedulingMode::Watchdog,
491 };
492
493 Self {
494 claim_generation,
495 policy,
496 lifetime,
497 control,
498 scheduling_mode,
499 latest_directive: None,
500 latest_requested_delay_ns: None,
501 latest_armed_delay_ns: None,
502 confirmed_wakeup_generation: None,
503 confirmed_work_generation: None,
504 callback,
505 wakeup: None,
506 work: None,
507 observability: TimerObservabilitySnapshot::new(epoch),
508 }
509 }
510
511 const fn snapshot(&self, identity: TimerIdentity) -> TimerSnapshot {
512 let state = match &self.control {
513 EntryControl::Ordinary {
514 control,
515 inactive_reason,
516 ..
517 } => match control.registration() {
518 TimerRegistration::Unregistered => TimerRuntimeStateSnapshot::Inactive {
519 reason: *inactive_reason,
520 },
521 TimerRegistration::Scheduled {
522 generation,
523 deadline_ns,
524 } => TimerRuntimeStateSnapshot::Ordinary(OrdinaryRuntimeStateSnapshot::Scheduled {
525 generation,
526 deadline_ns,
527 }),
528 TimerRegistration::Running { generation } => {
529 TimerRuntimeStateSnapshot::Ordinary(OrdinaryRuntimeStateSnapshot::Running {
530 generation,
531 })
532 }
533 },
534 EntryControl::Watchdog(control) => match control.state {
535 WatchdogState::Inactive => TimerRuntimeStateSnapshot::Inactive {
536 reason: control.inactive_reason,
537 },
538 WatchdogState::Scheduled {
539 scheduler_generation,
540 deadline_ns,
541 } => TimerRuntimeStateSnapshot::Watchdog(WatchdogRuntimeStateSnapshot::Scheduled {
542 scheduler_generation,
543 deadline_ns,
544 }),
545 WatchdogState::AwaitingWork {
546 successor_generation,
547 successor_deadline_ns,
548 attempt_generation,
549 attempt_status,
550 } => TimerRuntimeStateSnapshot::Watchdog(
551 WatchdogRuntimeStateSnapshot::AwaitingWork {
552 successor_generation,
553 successor_deadline_ns,
554 attempt: WatchdogAttemptSnapshot::new(attempt_generation, attempt_status),
555 },
556 ),
557 },
558 };
559
560 TimerSnapshot::new(
561 identity,
562 TimerRegistrationId::new(self.observability.epoch(), self.claim_generation),
563 self.policy,
564 self.lifetime,
565 state,
566 self.scheduling_mode,
567 self.latest_directive,
568 self.latest_requested_delay_ns,
569 self.latest_armed_delay_ns,
570 &self.observability,
571 )
572 }
573
574 fn take_wakeup_handle(&mut self, identity: &TimerIdentity) -> Option<ProviderHandle> {
575 self.wakeup
576 .take()
577 .map(|owned| detach_provider_handle(identity, self.claim_generation, owned))
578 }
579
580 fn take_work_handle(&mut self, identity: &TimerIdentity) -> Option<ProviderHandle> {
581 self.work
582 .take()
583 .map(|owned| detach_provider_handle(identity, self.claim_generation, owned))
584 }
585
586 fn take_provider_handles(&mut self, identity: &TimerIdentity) -> ProviderHandles {
587 ProviderHandles {
588 wakeup: self.take_wakeup_handle(identity),
589 work: self.take_work_handle(identity),
590 }
591 }
592
593 const fn provider_slot_mut(&mut self, role: CallbackRole) -> &mut Option<OwnedProviderHandle> {
594 match role {
595 CallbackRole::OrdinaryWork | CallbackRole::WatchdogScheduler => &mut self.wakeup,
596 CallbackRole::WatchdogWork => &mut self.work,
597 }
598 }
599
600 const fn owns_token_claim(&self, token: &CallbackToken) -> bool {
601 self.claim_generation == token.claim_generation
602 }
603}
604
605pub struct TimerRegistry {
607 epoch: TimerEpoch,
608 next_claim_generation: u64,
609 entries: BTreeMap<TimerIdentity, Entry>,
610}
611
612impl TimerRegistry {
613 pub(crate) const fn new(epoch: TimerEpoch) -> Self {
614 Self {
615 epoch,
616 next_claim_generation: 0,
617 entries: BTreeMap::new(),
618 }
619 }
620
621 #[cfg(test)]
622 pub(crate) fn len(&self) -> usize {
623 self.entries.len()
624 }
625
626 #[cfg(test)]
627 pub(crate) fn is_empty(&self) -> bool {
628 self.entries.is_empty()
629 }
630
631 pub(crate) const fn epoch(&self) -> TimerEpoch {
632 self.epoch
633 }
634
635 #[cfg(test)]
636 pub(crate) fn register_once(
637 &mut self,
638 identity: TimerIdentity,
639 lifetime: DeclarationLifetime,
640 ) -> Result<RegistrationClaim, RegisterError> {
641 self.register(identity, TimerPolicy::Once, lifetime, EntryCallback::None)
642 }
643
644 #[cfg(test)]
645 pub(crate) fn register_after_completion(
646 &mut self,
647 identity: TimerIdentity,
648 cadence: TimerCadence,
649 lifetime: DeclarationLifetime,
650 ) -> Result<RegistrationClaim, RegisterError> {
651 self.register(
652 identity,
653 TimerPolicy::AfterCompletion { cadence },
654 lifetime,
655 EntryCallback::None,
656 )
657 }
658
659 #[cfg(test)]
660 pub(crate) fn register_watchdog(
661 &mut self,
662 identity: TimerIdentity,
663 cadence: TimerCadence,
664 lifetime: DeclarationLifetime,
665 ) -> Result<RegistrationClaim, RegisterError> {
666 self.register(
667 identity,
668 TimerPolicy::Watchdog { cadence },
669 lifetime,
670 EntryCallback::None,
671 )
672 }
673
674 pub(crate) fn register_once_with_callback(
675 &mut self,
676 identity: TimerIdentity,
677 lifetime: DeclarationLifetime,
678 callback: OrdinaryCallback,
679 ) -> Result<RegistrationClaim, RegisterError> {
680 self.register(
681 identity,
682 TimerPolicy::Once,
683 lifetime,
684 EntryCallback::Ordinary(callback),
685 )
686 }
687
688 pub(crate) fn register_after_completion_with_callback(
689 &mut self,
690 identity: TimerIdentity,
691 cadence: TimerCadence,
692 lifetime: DeclarationLifetime,
693 callback: OrdinaryCallback,
694 ) -> Result<RegistrationClaim, RegisterError> {
695 self.register(
696 identity,
697 TimerPolicy::AfterCompletion { cadence },
698 lifetime,
699 EntryCallback::Ordinary(callback),
700 )
701 }
702
703 pub(crate) fn register_watchdog_with_callback(
704 &mut self,
705 identity: TimerIdentity,
706 cadence: TimerCadence,
707 lifetime: DeclarationLifetime,
708 callback: WatchdogCallback,
709 ) -> Result<RegistrationClaim, RegisterError> {
710 self.register(
711 identity,
712 TimerPolicy::Watchdog { cadence },
713 lifetime,
714 EntryCallback::Watchdog(callback),
715 )
716 }
717
718 fn register(
719 &mut self,
720 identity: TimerIdentity,
721 policy: TimerPolicy,
722 lifetime: DeclarationLifetime,
723 callback: EntryCallback,
724 ) -> Result<RegistrationClaim, RegisterError> {
725 if self.entries.contains_key(&identity) {
726 return Err(RegisterError::IdentityAlreadyRegistered(identity));
727 }
728 if self.entries.len() == MAX_TIMER_REGISTRATIONS {
729 return Err(RegisterError::CapacityExceeded {
730 max: MAX_TIMER_REGISTRATIONS,
731 });
732 }
733 let claim_generation = self
734 .next_claim_generation
735 .checked_add(1)
736 .ok_or(RegisterError::ClaimGenerationExhausted)?;
737
738 self.next_claim_generation = claim_generation;
739 self.entries.insert(
740 identity.clone(),
741 Entry::new(claim_generation, policy, lifetime, self.epoch, callback),
742 );
743 Ok(RegistrationClaim {
744 identity,
745 claim_generation,
746 })
747 }
748
749 pub(crate) fn ensure_once(
750 &mut self,
751 claim: &RegistrationClaim,
752 now_ns: u64,
753 schedule: TimerSchedule,
754 ) -> Result<RegistryTransition, RegistryError> {
755 self.request_ordinary(
756 claim,
757 now_ns,
758 PendingSchedule::resolve(schedule, now_ns)?,
759 OrdinaryRequest::EnsureOnce,
760 )
761 }
762
763 pub(crate) fn reconcile_ordinary(
769 &mut self,
770 claim: &RegistrationClaim,
771 now_ns: u64,
772 schedule: Option<TimerSchedule>,
773 ) -> Result<RegistryTransition, RegistryError> {
774 self.validate_ordinary_claim(claim)?;
775 let Some(schedule) = schedule else {
776 return self.cancel(claim);
777 };
778 self.request_ordinary(
779 claim,
780 now_ns,
781 PendingSchedule::resolve(schedule, now_ns)?,
782 OrdinaryRequest::Reconcile,
783 )
784 }
785
786 pub(crate) fn validate_ordinary_claim(
787 &self,
788 claim: &RegistrationClaim,
789 ) -> Result<(), RegistryError> {
790 let entry = self.entry(claim)?;
791 if matches!(entry.control, EntryControl::Ordinary { .. }) {
792 Ok(())
793 } else {
794 Err(RegistryError::PolicyMismatch {
795 actual: entry.policy.label(),
796 })
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.policy {
807 TimerPolicy::Once => Err(RegistryError::PolicyMismatch { actual: "once" }),
808 TimerPolicy::AfterCompletion { cadence } => {
809 let already_scheduled = matches!(
810 entry.control,
811 EntryControl::Ordinary {
812 ref control,
813 ..
814 } if matches!(
815 control.registration(),
816 TimerRegistration::Scheduled { .. }
817 )
818 );
819 if already_scheduled {
820 let entry = self.entry_mut(claim)?;
821 entry.observability.counters_mut().record_schedule_request();
822 entry.observability.counters_mut().record_coalesced();
823 entry.latest_requested_delay_ns = Some(cadence.as_nanos());
824 return Ok(RegistryTransition::normal(RegistryEffect::None));
825 }
826 let deadline_ns = cadence.deadline_after(now_ns)?;
827 self.request_ordinary(
828 claim,
829 now_ns,
830 PendingSchedule {
831 deadline_ns,
832 requested_delay_ns: Some(cadence.as_nanos()),
833 mode: TimerSchedulingMode::AfterCompletion,
834 },
835 OrdinaryRequest::EnsureRecurring,
836 )
837 }
838 TimerPolicy::Watchdog { cadence } => {
839 self.ensure_watchdog(claim, now_ns, cadence, WatchdogScheduleRequest::Cadence)
840 }
841 }
842 }
843
844 pub(crate) fn ensure_watchdog_immediately(
845 &mut self,
846 claim: &RegistrationClaim,
847 now_ns: u64,
848 ) -> Result<RegistryTransition, RegistryError> {
849 let entry = self.entry(claim)?;
850 let TimerPolicy::Watchdog { cadence } = entry.policy else {
851 return Err(RegistryError::PolicyMismatch {
852 actual: entry.policy.label(),
853 });
854 };
855 self.ensure_watchdog(claim, now_ns, cadence, WatchdogScheduleRequest::Immediate)
856 }
857
858 pub(crate) fn reconcile_watchdog_schedule(
859 &mut self,
860 claim: &RegistrationClaim,
861 now_ns: u64,
862 schedule: Option<TimerSchedule>,
863 ) -> Result<RegistryTransition, RegistryError> {
864 let entry = self.entry(claim)?;
865 let TimerPolicy::Watchdog { cadence } = entry.policy else {
866 return Err(RegistryError::PolicyMismatch {
867 actual: entry.policy.label(),
868 });
869 };
870 let Some(schedule) = schedule else {
871 return self.cancel(claim);
872 };
873 let requested = PendingSchedule::resolve(schedule, now_ns)?;
874 self.ensure_watchdog(
875 claim,
876 now_ns,
877 cadence,
878 WatchdogScheduleRequest::Reconcile(requested),
879 )
880 }
881
882 fn request_ordinary(
883 &mut self,
884 claim: &RegistrationClaim,
885 now_ns: u64,
886 requested: PendingSchedule,
887 request: OrdinaryRequest,
888 ) -> Result<RegistryTransition, RegistryError> {
889 let entry = self.entry_mut(claim)?;
890 let policy_matches = match request {
891 OrdinaryRequest::EnsureOnce => matches!(entry.policy, TimerPolicy::Once),
892 OrdinaryRequest::EnsureRecurring => {
893 matches!(entry.policy, TimerPolicy::AfterCompletion { .. })
894 }
895 OrdinaryRequest::Reconcile => !matches!(entry.policy, TimerPolicy::Watchdog { .. }),
896 };
897 if !policy_matches {
898 return Err(RegistryError::PolicyMismatch {
899 actual: entry.policy.label(),
900 });
901 }
902 entry.observability.counters_mut().record_schedule_request();
903 entry.latest_requested_delay_ns = requested.requested_delay_ns;
904
905 let EntryControl::Ordinary {
906 control, pending, ..
907 } = &mut entry.control
908 else {
909 return Err(RegistryError::PolicyMismatch {
910 actual: entry.policy.label(),
911 });
912 };
913 let was_running = matches!(control.registration(), TimerRegistration::Running { .. });
914 let action = match request {
915 OrdinaryRequest::EnsureOnce | OrdinaryRequest::EnsureRecurring => {
916 control.schedule(requested.deadline_ns)
917 }
918 OrdinaryRequest::Reconcile => control.reconcile(requested.deadline_ns),
919 };
920
921 let action = match action {
922 Ok(action) => action,
923 Err(error) => {
924 return Ok(terminal_ordinary(entry, claim.identity.clone(), error));
925 }
926 };
927
928 if was_running {
929 *pending = Some(select_pending_ordinary(*pending, request, requested));
930 }
931 if matches!(request, OrdinaryRequest::Reconcile) {
932 entry.scheduling_mode = requested.mode;
933 }
934
935 Ok(apply_ordinary_action(
936 entry,
937 claim.identity.clone(),
938 now_ns,
939 action,
940 requested,
941 ))
942 }
943
944 fn ensure_watchdog(
945 &mut self,
946 claim: &RegistrationClaim,
947 now_ns: u64,
948 cadence: TimerCadence,
949 request: WatchdogScheduleRequest,
950 ) -> Result<RegistryTransition, RegistryError> {
951 let entry = self.entry_mut(claim)?;
952 if !matches!(entry.policy, TimerPolicy::Watchdog { .. }) {
953 return Err(RegistryError::PolicyMismatch {
954 actual: entry.policy.label(),
955 });
956 }
957 entry.observability.counters_mut().record_schedule_request();
958 let requested_delay_ns = match request {
959 WatchdogScheduleRequest::Cadence => Some(cadence.as_nanos()),
960 WatchdogScheduleRequest::Immediate => Some(0),
961 WatchdogScheduleRequest::Reconcile(requested) => requested.requested_delay_ns,
962 };
963 entry.latest_requested_delay_ns = requested_delay_ns;
964
965 let EntryControl::Watchdog(control) = &mut entry.control else {
966 return Err(RegistryError::PolicyMismatch {
967 actual: entry.policy.label(),
968 });
969 };
970 match control.state {
971 WatchdogState::Inactive => {
972 let deadline_ns = match request {
973 WatchdogScheduleRequest::Cadence => cadence.deadline_after(now_ns)?,
974 WatchdogScheduleRequest::Immediate => now_ns,
975 WatchdogScheduleRequest::Reconcile(requested) => requested.deadline_ns,
976 };
977 let transition =
978 control.arm_scheduler(claim, now_ns, deadline_ns, WakeupArm::Initial);
979 if transition.failure().is_none() {
980 entry.scheduling_mode = match request {
981 WatchdogScheduleRequest::Cadence => TimerSchedulingMode::Watchdog,
982 WatchdogScheduleRequest::Immediate => TimerSchedulingMode::Continuation,
983 WatchdogScheduleRequest::Reconcile(requested) => requested.mode,
984 };
985 }
986 Ok(transition)
987 }
988 WatchdogState::Scheduled { deadline_ns, .. }
989 if match request {
990 WatchdogScheduleRequest::Immediate => deadline_ns > now_ns,
991 WatchdogScheduleRequest::Reconcile(requested) => {
992 deadline_ns != requested.deadline_ns
993 }
994 WatchdogScheduleRequest::Cadence => false,
995 } =>
996 {
997 let (deadline_ns, mode) = match request {
998 WatchdogScheduleRequest::Reconcile(requested) => {
999 (requested.deadline_ns, requested.mode)
1000 }
1001 WatchdogScheduleRequest::Immediate | WatchdogScheduleRequest::Cadence => {
1002 (now_ns, TimerSchedulingMode::Continuation)
1003 }
1004 };
1005 let transition =
1006 control.arm_scheduler(claim, now_ns, deadline_ns, WakeupArm::Replacement);
1007 if transition.failure().is_none() {
1008 entry.scheduling_mode = mode;
1009 }
1010 Ok(transition)
1011 }
1012 WatchdogState::Scheduled { .. } => {
1013 entry.observability.counters_mut().record_coalesced();
1014 Ok(RegistryTransition::normal(RegistryEffect::None))
1015 }
1016 WatchdogState::AwaitingWork {
1017 attempt_status: WatchdogAttemptStatus::Dispatched,
1018 ..
1019 } => {
1020 if let WatchdogScheduleRequest::Reconcile(requested) = request {
1021 control.pending = Some(WatchdogPending::Reconcile(requested));
1022 }
1023 entry.observability.counters_mut().record_coalesced();
1024 Ok(RegistryTransition::normal(RegistryEffect::None))
1025 }
1026 WatchdogState::AwaitingWork {
1027 attempt_status: WatchdogAttemptStatus::Running,
1028 ..
1029 } => {
1030 control.pending = Some(select_pending_watchdog(control.pending, request, now_ns));
1031 entry.observability.counters_mut().record_coalesced();
1032 Ok(RegistryTransition::normal(RegistryEffect::None))
1033 }
1034 }
1035 }
1036
1037 pub(crate) fn cancel(
1038 &mut self,
1039 claim: &RegistrationClaim,
1040 ) -> Result<RegistryTransition, RegistryError> {
1041 let identity = claim.identity.clone();
1042 let (transition, remove) = {
1043 let entry = self.entry_mut(claim)?;
1044 match &mut entry.control {
1045 EntryControl::Ordinary {
1046 control,
1047 pending,
1048 inactive_reason,
1049 } => {
1050 let before = control.registration();
1051 let action = match control.cancel() {
1052 Ok(action) => action,
1053 Err(error) => {
1054 return Ok(terminal_ordinary(entry, identity.clone(), error));
1055 }
1056 };
1057 let mut remove = false;
1058 let transition = match (before, action) {
1059 (TimerRegistration::Unregistered, TimerControlAction::None) => {
1060 remove =
1061 matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped);
1062 RegistryTransition::normal(RegistryEffect::None)
1063 }
1064 (TimerRegistration::Running { .. }, TimerControlAction::None) => {
1065 if !matches!(*pending, Some(OrdinaryPending::Unregister)) {
1066 *pending = Some(OrdinaryPending::Cancel);
1067 }
1068 RegistryTransition::normal(RegistryEffect::None)
1069 }
1070 (_, TimerControlAction::Clear) => {
1071 *inactive_reason = InactiveReason::Cancelled;
1072 entry.observability.counters_mut().record_cancellation();
1073 remove =
1074 matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped);
1075 RegistryTransition::normal(clear_callbacks(
1076 identity.clone(),
1077 CallbacksToClear::Wakeup,
1078 ))
1079 }
1080 (TimerRegistration::Scheduled { .. }, TimerControlAction::None)
1081 | (_, TimerControlAction::Arm { .. } | TimerControlAction::Disarm { .. }) =>
1082 {
1083 let clear_wakeup = control.terminate();
1084 *pending = None;
1085 *inactive_reason = InactiveReason::ControlFailure(
1086 TimerControlFailure::DirectiveNotAllowed,
1087 );
1088 RegistryTransition::terminal(
1089 clear_wakeup_if(identity.clone(), clear_wakeup),
1090 TimerControlFailure::DirectiveNotAllowed,
1091 )
1092 }
1093 };
1094 (transition, remove)
1095 }
1096 EntryControl::Watchdog(control) => {
1097 let cancels_immediately = matches!(
1098 control.state,
1099 WatchdogState::Scheduled { .. }
1100 | WatchdogState::AwaitingWork {
1101 attempt_status: WatchdogAttemptStatus::Dispatched,
1102 ..
1103 }
1104 );
1105 let transition = cancel_watchdog(control, &identity);
1106 if cancels_immediately && transition.failure().is_none() {
1107 entry.observability.counters_mut().record_cancellation();
1108 }
1109 let stopped = matches!(control.state, WatchdogState::Inactive);
1110 (
1111 transition,
1112 stopped && matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped),
1113 )
1114 }
1115 }
1116 };
1117
1118 if remove {
1119 self.entries.remove(&identity);
1120 }
1121 Ok(transition)
1122 }
1123
1124 pub(crate) fn unregister(
1130 &mut self,
1131 claim: &RegistrationClaim,
1132 ) -> Result<RegistryTransition, RegistryError> {
1133 let identity = claim.identity.clone();
1134 let running_role = {
1135 let entry = self.entry(claim)?;
1136 match &entry.control {
1137 EntryControl::Ordinary { control, .. }
1138 if matches!(control.registration(), TimerRegistration::Running { .. }) =>
1139 {
1140 Some(CallbackRole::OrdinaryWork)
1141 }
1142 EntryControl::Watchdog(control)
1143 if matches!(
1144 control.state,
1145 WatchdogState::AwaitingWork {
1146 attempt_status: WatchdogAttemptStatus::Running,
1147 ..
1148 }
1149 ) =>
1150 {
1151 Some(CallbackRole::WatchdogWork)
1152 }
1153 EntryControl::Ordinary { .. } | EntryControl::Watchdog(_) => None,
1154 }
1155 };
1156
1157 match running_role {
1158 Some(CallbackRole::OrdinaryWork) => {
1159 let entry = self.entry_mut(claim)?;
1160 let EntryControl::Ordinary {
1161 control,
1162 pending,
1163 inactive_reason,
1164 } = &mut entry.control
1165 else {
1166 return Err(RegistryError::PolicyMismatch {
1167 actual: entry.policy.label(),
1168 });
1169 };
1170 let transition = match control.cancel() {
1171 Ok(TimerControlAction::None) => {
1172 *pending = Some(OrdinaryPending::Unregister);
1173 RegistryTransition::normal(RegistryEffect::None)
1174 }
1175 Ok(_) => RegistryTransition::terminal(
1176 RegistryEffect::None,
1177 TimerControlFailure::DirectiveNotAllowed,
1178 ),
1179 Err(error) => {
1180 let failure = map_control_failure(error);
1181 let clear_wakeup = control.terminate();
1182 *pending = None;
1183 *inactive_reason = InactiveReason::ControlFailure(failure);
1184 RegistryTransition::terminal(
1185 clear_wakeup_if(identity.clone(), clear_wakeup),
1186 failure,
1187 )
1188 }
1189 };
1190 if transition.failure().is_some() {
1191 self.entries.remove(&identity);
1192 }
1193 Ok(transition)
1194 }
1195 Some(CallbackRole::WatchdogWork) => {
1196 let entry = self.entry_mut(claim)?;
1197 let EntryControl::Watchdog(control) = &mut entry.control else {
1198 return Err(RegistryError::PolicyMismatch {
1199 actual: entry.policy.label(),
1200 });
1201 };
1202 control.pending = Some(WatchdogPending::Unregister);
1203 Ok(RegistryTransition::normal(RegistryEffect::None))
1204 }
1205 Some(CallbackRole::WatchdogScheduler) => Err(RegistryError::StaleCallback),
1206 None => {
1207 let transition = self.cancel(claim)?;
1208 self.entries.remove(&identity);
1209 Ok(transition)
1210 }
1211 }
1212 }
1213
1214 pub(crate) fn begin_ordinary(&mut self, token: &CallbackToken) -> CallbackAcceptance {
1215 let Some(entry) = self.entries.get_mut(token.identity()) else {
1216 return CallbackAcceptance::Stale;
1217 };
1218 if token.role != CallbackRole::OrdinaryWork || !entry.owns_token_claim(token) {
1219 entry.observability.counters_mut().record_stale_wakeup();
1220 return CallbackAcceptance::Stale;
1221 }
1222 let EntryControl::Ordinary { control, .. } = &mut entry.control else {
1223 entry.observability.counters_mut().record_stale_wakeup();
1224 return CallbackAcceptance::Stale;
1225 };
1226 if control.begin(token.callback_generation) {
1227 entry.observability.counters_mut().record_work_started();
1228 CallbackAcceptance::Accepted
1229 } else {
1230 entry.observability.counters_mut().record_stale_wakeup();
1231 CallbackAcceptance::Stale
1232 }
1233 }
1234
1235 #[allow(clippy::too_many_lines)] pub(crate) fn complete_ordinary(
1237 &mut self,
1238 token: &CallbackToken,
1239 now_ns: u64,
1240 result: TimerRunResult,
1241 ) -> Result<RegistryTransition, RegistryError> {
1242 let identity = token.identity.clone();
1243 let (transition, remove) = {
1244 let entry = self.entry_by_token_mut(token, CallbackRole::OrdinaryWork)?;
1245 let EntryControl::Ordinary {
1246 control,
1247 pending,
1248 inactive_reason,
1249 } = &mut entry.control
1250 else {
1251 return Err(RegistryError::StaleCallback);
1252 };
1253 if control.registration()
1254 != (TimerRegistration::Running {
1255 generation: token.callback_generation,
1256 })
1257 {
1258 return Err(RegistryError::StaleCallback);
1259 }
1260
1261 let completion = result.completion();
1262 let pending_command = *pending;
1263 let terminal_pending = matches!(
1264 pending_command,
1265 Some(OrdinaryPending::Cancel | OrdinaryPending::Unregister)
1266 );
1267 if completion.outcome() == TimerCompletionOutcome::InvariantFailure {
1268 let transition =
1269 invariant_completion(entry, token.callback_generation, completion, now_ns);
1270 let remove = matches!(pending_command, Some(OrdinaryPending::Unregister))
1271 || matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped);
1272 (transition, remove)
1273 } else if !terminal_pending
1274 && matches!(entry.policy, TimerPolicy::Once)
1275 && matches!(result.directive(), TimerDirective::RecurAfterCompletion)
1276 {
1277 let transition = terminal_completion(
1278 entry,
1279 token.callback_generation,
1280 completion.work_count(),
1281 now_ns,
1282 TimerControlFailure::DirectiveNotAllowed,
1283 );
1284 let remove = matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped);
1285 (transition, remove)
1286 } else {
1287 let effective_directive = if terminal_pending {
1288 TimerDirective::Stop
1289 } else {
1290 result.directive()
1291 };
1292 let cadence = match entry.policy {
1293 TimerPolicy::AfterCompletion { cadence } => Some(cadence),
1294 TimerPolicy::Once => None,
1295 TimerPolicy::Watchdog { .. } => {
1296 return Err(RegistryError::StaleCallback);
1297 }
1298 };
1299 let resolved = match effective_directive.resolve(now_ns, cadence) {
1300 Ok(value) => value,
1301 Err(error) => {
1302 let transition = terminal_completion(
1303 entry,
1304 token.callback_generation,
1305 completion.work_count(),
1306 now_ns,
1307 map_directive_failure(error),
1308 );
1309 let remove = matches!(pending_command, Some(OrdinaryPending::Unregister))
1310 || matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped);
1311 return Ok(remove_after(
1312 &mut self.entries,
1313 &identity,
1314 transition,
1315 remove,
1316 ));
1317 }
1318 };
1319 let directive_snapshot = TimerDirectiveSnapshot::try_from(effective_directive)
1320 .map_err(RegistryError::Schedule)?;
1321 let callback_schedule = resolved.deadline_ns.map(|deadline_ns| PendingSchedule {
1322 deadline_ns,
1323 requested_delay_ns: resolved.requested_delay_ns,
1324 mode: directive_snapshot
1325 .scheduling_mode()
1326 .unwrap_or(entry.scheduling_mode),
1327 });
1328 let selected_schedule =
1329 select_completion_schedule(pending_command, callback_schedule);
1330 let action = match control.complete(
1331 token.callback_generation,
1332 selected_schedule.map(|value| value.deadline_ns),
1333 terminal_pending,
1334 ) {
1335 Ok(action) => action,
1336 Err(TimerControlError::StaleCompletion) => {
1337 return Err(RegistryError::StaleCallback);
1338 }
1339 Err(error) => {
1340 let transition = terminal_completion(
1341 entry,
1342 token.callback_generation,
1343 completion.work_count(),
1344 now_ns,
1345 map_control_failure(error),
1346 );
1347 let remove = matches!(pending_command, Some(OrdinaryPending::Unregister))
1348 || matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped);
1349 return Ok(remove_after(
1350 &mut self.entries,
1351 &identity,
1352 transition,
1353 remove,
1354 ));
1355 }
1356 };
1357 *pending = None;
1358 entry.latest_directive = Some(directive_snapshot);
1359 entry.observability.record_completion(completion, now_ns);
1360
1361 let mut remove = false;
1362 let transition = match action {
1363 TimerControlAction::Arm {
1364 generation,
1365 deadline_ns,
1366 kind,
1367 } => {
1368 let selected = selected_schedule.unwrap_or(PendingSchedule {
1369 deadline_ns,
1370 requested_delay_ns: None,
1371 mode: entry.scheduling_mode,
1372 });
1373 entry.scheduling_mode = selected.mode;
1374 entry.latest_requested_delay_ns = selected.requested_delay_ns;
1375 RegistryTransition::normal(RegistryEffect::ArmWakeup {
1376 token: CallbackToken::new(
1377 identity.clone(),
1378 entry.claim_generation,
1379 generation,
1380 CallbackRole::OrdinaryWork,
1381 ),
1382 deadline_ns,
1383 delay_ns: deadline_ns.saturating_sub(now_ns),
1384 arm: kind,
1385 })
1386 }
1387 TimerControlAction::Disarm { cancelled } => {
1388 *inactive_reason = if cancelled {
1389 entry.observability.counters_mut().record_cancellation();
1390 InactiveReason::Cancelled
1391 } else {
1392 InactiveReason::Stopped
1393 };
1394 remove = matches!(pending_command, Some(OrdinaryPending::Unregister))
1395 || matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped);
1396 RegistryTransition::normal(RegistryEffect::None)
1397 }
1398 TimerControlAction::None | TimerControlAction::Clear => {
1399 let clear_wakeup = control.terminate();
1400 *inactive_reason = InactiveReason::ControlFailure(
1401 TimerControlFailure::DirectiveNotAllowed,
1402 );
1403 remove = matches!(pending_command, Some(OrdinaryPending::Unregister))
1404 || matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped);
1405 RegistryTransition::terminal(
1406 clear_wakeup_if(identity.clone(), clear_wakeup),
1407 TimerControlFailure::DirectiveNotAllowed,
1408 )
1409 }
1410 };
1411 (transition, remove)
1412 }
1413 };
1414
1415 Ok(remove_after(
1416 &mut self.entries,
1417 &identity,
1418 transition,
1419 remove,
1420 ))
1421 }
1422
1423 #[allow(clippy::too_many_lines)] pub(crate) fn begin_watchdog_scheduler(
1425 &mut self,
1426 token: &CallbackToken,
1427 now_ns: u64,
1428 ) -> RegistryTransition {
1429 let Some(entry) = self.entries.get_mut(token.identity()) else {
1430 return RegistryTransition::normal(RegistryEffect::None);
1431 };
1432 if token.role != CallbackRole::WatchdogScheduler || !entry.owns_token_claim(token) {
1433 entry.observability.counters_mut().record_stale_wakeup();
1434 return RegistryTransition::normal(RegistryEffect::None);
1435 }
1436 let TimerPolicy::Watchdog { cadence } = entry.policy else {
1437 entry.observability.counters_mut().record_stale_wakeup();
1438 return RegistryTransition::normal(RegistryEffect::None);
1439 };
1440 let EntryControl::Watchdog(control) = &mut entry.control else {
1441 entry.observability.counters_mut().record_stale_wakeup();
1442 return RegistryTransition::normal(RegistryEffect::None);
1443 };
1444
1445 let accepted = match control.state {
1446 WatchdogState::Scheduled {
1447 scheduler_generation,
1448 ..
1449 } => scheduler_generation == token.callback_generation,
1450 WatchdogState::AwaitingWork {
1451 successor_generation,
1452 attempt_status: WatchdogAttemptStatus::Dispatched,
1453 ..
1454 } => successor_generation == token.callback_generation,
1455 WatchdogState::Inactive
1456 | WatchdogState::AwaitingWork {
1457 attempt_status: WatchdogAttemptStatus::Running,
1458 ..
1459 } => false,
1460 };
1461 if !accepted {
1462 entry.observability.counters_mut().record_stale_wakeup();
1463 return RegistryTransition::normal(RegistryEffect::None);
1464 }
1465
1466 entry
1467 .observability
1468 .counters_mut()
1469 .record_scheduler_started();
1470 if matches!(control.state, WatchdogState::AwaitingWork { .. }) {
1471 entry.observability.record_unacknowledged(now_ns);
1472 }
1473 let Some((successor_generation, attempt_generation)) = control.next_dispatch_generations()
1474 else {
1475 return control.terminate(
1476 clear_callbacks(token.identity.clone(), CallbacksToClear::Work),
1477 TimerControlFailure::GenerationExhausted,
1478 );
1479 };
1480 let Ok(successor_deadline_ns) = cadence.deadline_after(now_ns) else {
1481 return control.terminate(
1482 clear_callbacks(token.identity.clone(), CallbacksToClear::Work),
1483 TimerControlFailure::DeadlineOverflow,
1484 );
1485 };
1486
1487 control.scheduler_generation = successor_generation;
1488 control.attempt_generation = attempt_generation;
1489 control.state = WatchdogState::AwaitingWork {
1490 successor_generation,
1491 successor_deadline_ns,
1492 attempt_generation,
1493 attempt_status: WatchdogAttemptStatus::Dispatched,
1494 };
1495 control.pending = None;
1496 entry.scheduling_mode = TimerSchedulingMode::Watchdog;
1497 RegistryTransition::normal(RegistryEffect::DispatchWatchdog {
1498 successor: CallbackToken::new(
1499 token.identity.clone(),
1500 entry.claim_generation,
1501 successor_generation,
1502 CallbackRole::WatchdogScheduler,
1503 ),
1504 successor_deadline_ns,
1505 successor_delay_ns: cadence.as_nanos(),
1506 work: CallbackToken::new(
1507 token.identity.clone(),
1508 entry.claim_generation,
1509 attempt_generation,
1510 CallbackRole::WatchdogWork,
1511 ),
1512 })
1513 }
1514
1515 pub(crate) fn begin_watchdog_work(&mut self, token: &CallbackToken) -> CallbackAcceptance {
1516 let Some(entry) = self.entries.get_mut(token.identity()) else {
1517 return CallbackAcceptance::Stale;
1518 };
1519 if token.role != CallbackRole::WatchdogWork || !entry.owns_token_claim(token) {
1520 entry.observability.counters_mut().record_stale_work();
1521 return CallbackAcceptance::Stale;
1522 }
1523 let EntryControl::Watchdog(control) = &mut entry.control else {
1524 entry.observability.counters_mut().record_stale_work();
1525 return CallbackAcceptance::Stale;
1526 };
1527 match &mut control.state {
1528 WatchdogState::AwaitingWork {
1529 attempt_generation,
1530 attempt_status,
1531 ..
1532 } if *attempt_generation == token.callback_generation
1533 && *attempt_status == WatchdogAttemptStatus::Dispatched =>
1534 {
1535 *attempt_status = WatchdogAttemptStatus::Running;
1536 entry.observability.counters_mut().record_work_started();
1537 CallbackAcceptance::Accepted
1538 }
1539 WatchdogState::Inactive
1540 | WatchdogState::Scheduled { .. }
1541 | WatchdogState::AwaitingWork { .. } => {
1542 entry.observability.counters_mut().record_stale_work();
1543 CallbackAcceptance::Stale
1544 }
1545 }
1546 }
1547
1548 #[allow(clippy::too_many_lines)] pub(crate) fn complete_watchdog_work(
1550 &mut self,
1551 token: &CallbackToken,
1552 now_ns: u64,
1553 result: WatchdogRunResult,
1554 ) -> Result<RegistryTransition, RegistryError> {
1555 let identity = token.identity.clone();
1556 let (transition, remove) = {
1557 let entry = self.entry_by_token_mut(token, CallbackRole::WatchdogWork)?;
1558 let EntryControl::Watchdog(control) = &mut entry.control else {
1559 return Err(RegistryError::StaleCallback);
1560 };
1561 let TimerPolicy::Watchdog { cadence } = entry.policy else {
1562 return Err(RegistryError::StaleCallback);
1563 };
1564 let (successor_generation, successor_deadline_ns) = match control.state {
1565 WatchdogState::AwaitingWork {
1566 successor_generation,
1567 successor_deadline_ns,
1568 attempt_generation,
1569 attempt_status: WatchdogAttemptStatus::Running,
1570 } if attempt_generation == token.callback_generation => {
1571 (successor_generation, successor_deadline_ns)
1572 }
1573 WatchdogState::Inactive
1574 | WatchdogState::Scheduled { .. }
1575 | WatchdogState::AwaitingWork { .. } => {
1576 return Err(RegistryError::StaleCallback);
1577 }
1578 };
1579
1580 let completion = result.completion();
1581 entry.observability.record_completion(completion, now_ns);
1582 let decision = if completion.outcome() == TimerCompletionOutcome::InvariantFailure {
1583 WatchdogDecision::Stop
1584 } else {
1585 match control.pending {
1586 Some(WatchdogPending::Cancel | WatchdogPending::Unregister) => {
1587 WatchdogDecision::Stop
1588 }
1589 Some(WatchdogPending::Ensure) => WatchdogDecision::Continue,
1590 Some(WatchdogPending::EnsureImmediately) => {
1591 WatchdogDecision::ContinueImmediately
1592 }
1593 Some(WatchdogPending::Reconcile(requested)) => {
1594 WatchdogDecision::ScheduleAt(requested.deadline_ns)
1595 }
1596 None => result.decision(),
1597 }
1598 };
1599 let cancelled = matches!(control.pending, Some(WatchdogPending::Cancel));
1600 let unregister = matches!(control.pending, Some(WatchdogPending::Unregister));
1601 let pending_schedule = match control.pending {
1602 Some(WatchdogPending::Reconcile(requested)) => Some(requested),
1603 _ => None,
1604 };
1605 control.pending = None;
1606
1607 match decision {
1608 WatchdogDecision::Continue => {
1609 control.state = WatchdogState::Scheduled {
1610 scheduler_generation: successor_generation,
1611 deadline_ns: successor_deadline_ns,
1612 };
1613 entry.scheduling_mode = TimerSchedulingMode::Watchdog;
1614 entry.latest_requested_delay_ns = Some(cadence.as_nanos());
1615 (RegistryTransition::normal(RegistryEffect::None), false)
1616 }
1617 WatchdogDecision::ContinueImmediately | WatchdogDecision::ScheduleAt(_) => {
1618 let (deadline_ns, retain_successor) =
1619 if let WatchdogDecision::ScheduleAt(deadline_ns) = decision {
1620 entry.scheduling_mode = pending_schedule
1621 .map_or(TimerSchedulingMode::Deadline, |requested| requested.mode);
1622 entry.latest_requested_delay_ns =
1623 pending_schedule.and_then(|requested| requested.requested_delay_ns);
1624 (deadline_ns, successor_deadline_ns == deadline_ns)
1625 } else {
1626 entry.scheduling_mode = TimerSchedulingMode::Continuation;
1627 entry.latest_requested_delay_ns = Some(0);
1628 (now_ns, successor_deadline_ns <= now_ns)
1629 };
1630 if retain_successor {
1631 control.state = WatchdogState::Scheduled {
1632 scheduler_generation: successor_generation,
1633 deadline_ns: successor_deadline_ns,
1634 };
1635 (RegistryTransition::normal(RegistryEffect::None), false)
1636 } else if let Some(generation) = control.scheduler_generation.checked_add(1) {
1637 control.scheduler_generation = generation;
1638 control.state = WatchdogState::Scheduled {
1639 scheduler_generation: generation,
1640 deadline_ns,
1641 };
1642 (
1643 RegistryTransition::normal(RegistryEffect::ArmWakeup {
1644 token: CallbackToken::new(
1645 identity.clone(),
1646 entry.claim_generation,
1647 generation,
1648 CallbackRole::WatchdogScheduler,
1649 ),
1650 deadline_ns,
1651 delay_ns: deadline_ns.saturating_sub(now_ns),
1652 arm: WakeupArm::Replacement,
1653 }),
1654 false,
1655 )
1656 } else {
1657 (
1658 control.terminate(
1659 clear_callbacks(identity.clone(), CallbacksToClear::Wakeup),
1660 TimerControlFailure::GenerationExhausted,
1661 ),
1662 matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped),
1663 )
1664 }
1665 }
1666 WatchdogDecision::Stop => {
1667 control.state = WatchdogState::Inactive;
1668 control.inactive_reason = if cancelled {
1669 entry.observability.counters_mut().record_cancellation();
1670 InactiveReason::Cancelled
1671 } else if completion.outcome() == TimerCompletionOutcome::InvariantFailure {
1672 InactiveReason::InvariantFailure
1673 } else {
1674 InactiveReason::Stopped
1675 };
1676 (
1677 RegistryTransition::normal(clear_callbacks(
1678 identity.clone(),
1679 CallbacksToClear::Wakeup,
1680 )),
1681 unregister
1682 || matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped),
1683 )
1684 }
1685 }
1686 };
1687
1688 Ok(remove_after(
1689 &mut self.entries,
1690 &identity,
1691 transition,
1692 remove,
1693 ))
1694 }
1695
1696 pub(crate) fn snapshot(&self, identity: &TimerIdentity) -> Option<TimerSnapshot> {
1697 self.entries
1698 .get(identity)
1699 .map(|entry| entry.snapshot(identity.clone()))
1700 }
1701
1702 pub(crate) fn inventory(&self) -> TimerInventorySnapshot {
1703 TimerInventorySnapshot::new(
1704 self.epoch,
1705 self.entries
1706 .iter()
1707 .map(|(identity, entry)| entry.snapshot(identity.clone()))
1708 .collect(),
1709 )
1710 }
1711
1712 pub(crate) fn consecutive_expected_failures(&self, identity: &TimerIdentity) -> Option<u64> {
1713 self.entries.get(identity).map(|entry| {
1714 entry
1715 .observability
1716 .outcomes()
1717 .consecutive_expected_failures()
1718 })
1719 }
1720
1721 pub(crate) fn has_armed_wakeup(
1722 &self,
1723 claim: &RegistrationClaim,
1724 ) -> Result<bool, RegistryError> {
1725 Ok(self.entry(claim)?.wakeup.is_some())
1726 }
1727
1728 pub(crate) fn declaration_matches(
1729 &self,
1730 claim: &RegistrationClaim,
1731 policy: TimerPolicy,
1732 lifetime: DeclarationLifetime,
1733 ) -> Result<bool, RegistryError> {
1734 let entry = self.entry(claim)?;
1735 Ok(entry.policy == policy && entry.lifetime == lifetime)
1736 }
1737
1738 pub(crate) fn record_callback_measurements(
1739 &mut self,
1740 token: &CallbackToken,
1741 instructions: u64,
1742 memory_start: MemoryPages,
1743 memory_end: MemoryPages,
1744 ) -> Result<(), RegistryError> {
1745 let Some(entry) = self.entries.get_mut(token.identity()) else {
1748 return Ok(());
1749 };
1750 if !entry.owns_token_claim(token) {
1753 return Ok(());
1754 }
1755
1756 let memory = memory_sample(memory_start, memory_end);
1757 match (&entry.control, token.role) {
1758 (EntryControl::Watchdog(_), CallbackRole::WatchdogScheduler) => entry
1759 .observability
1760 .record_scheduler_measurements(instructions, memory),
1761 (EntryControl::Ordinary { .. }, CallbackRole::OrdinaryWork)
1762 | (EntryControl::Watchdog(_), CallbackRole::WatchdogWork) => entry
1763 .observability
1764 .record_work_measurements(instructions, memory),
1765 _ => {
1766 return Err(RegistryError::PolicyMismatch {
1767 actual: entry.policy.label(),
1768 });
1769 }
1770 }
1771 Ok(())
1772 }
1773
1774 pub(crate) fn fail_registration(
1775 &mut self,
1776 claim: &RegistrationClaim,
1777 failure: TimerControlFailure,
1778 ) -> Result<ProviderHandles, RegistryError> {
1779 let identity = claim.identity.clone();
1780 let (handles, remove) = {
1781 let entry = self.entry_mut(claim)?;
1782 match &mut entry.control {
1783 EntryControl::Ordinary {
1784 control,
1785 pending,
1786 inactive_reason,
1787 } => {
1788 control.terminate();
1789 *pending = None;
1790 *inactive_reason = InactiveReason::ControlFailure(failure);
1791 }
1792 EntryControl::Watchdog(control) => {
1793 control.state = WatchdogState::Inactive;
1794 control.pending = None;
1795 control.inactive_reason = InactiveReason::ControlFailure(failure);
1796 }
1797 }
1798 let handles = entry.take_provider_handles(&identity);
1799 (
1800 handles,
1801 matches!(entry.lifetime, DeclarationLifetime::RemoveWhenStopped),
1802 )
1803 };
1804 if remove {
1805 self.entries.remove(&identity);
1806 }
1807 Ok(handles)
1808 }
1809
1810 pub(crate) fn ordinary_callback(
1811 &self,
1812 token: &CallbackToken,
1813 ) -> Result<OrdinaryCallback, RegistryError> {
1814 let entry = self.running_work_entry(token)?;
1815 match &entry.callback {
1816 EntryCallback::Ordinary(callback) => Ok(Rc::clone(callback)),
1817 EntryCallback::Watchdog(_) => Err(RegistryError::MissingCallback),
1818 #[cfg(test)]
1819 EntryCallback::None => Err(RegistryError::MissingCallback),
1820 }
1821 }
1822
1823 pub(crate) fn watchdog_callback(
1824 &self,
1825 token: &CallbackToken,
1826 ) -> Result<WatchdogCallback, RegistryError> {
1827 let entry = self.running_work_entry(token)?;
1828 match &entry.callback {
1829 EntryCallback::Watchdog(callback) => Ok(Rc::clone(callback)),
1830 EntryCallback::Ordinary(_) => Err(RegistryError::MissingCallback),
1831 #[cfg(test)]
1832 EntryCallback::None => Err(RegistryError::MissingCallback),
1833 }
1834 }
1835
1836 pub(crate) fn validate_running_context(
1837 &self,
1838 token: &CallbackToken,
1839 ) -> Result<(), RegistryError> {
1840 self.running_work_entry(token).map(|_| ())
1841 }
1842
1843 pub(crate) fn install_provider_handle(
1844 &mut self,
1845 token: &CallbackToken,
1846 handle: TimerHandle,
1847 ) -> Result<(), (RegistryError, TimerHandle)> {
1848 let entry = match self.entry_by_token_mut(token, token.role) {
1849 Ok(entry) => entry,
1850 Err(error) => return Err((error, handle)),
1851 };
1852 let valid = match (&entry.control, token.role) {
1853 (EntryControl::Ordinary { control, .. }, CallbackRole::OrdinaryWork) => matches!(
1854 control.registration(),
1855 TimerRegistration::Scheduled { generation, .. }
1856 if generation == token.callback_generation
1857 ),
1858 (EntryControl::Watchdog(control), CallbackRole::WatchdogScheduler) => {
1859 matches!(
1860 control.state,
1861 WatchdogState::Scheduled {
1862 scheduler_generation,
1863 ..
1864 } if scheduler_generation == token.callback_generation
1865 ) || matches!(
1866 control.state,
1867 WatchdogState::AwaitingWork {
1868 successor_generation,
1869 ..
1870 } if successor_generation == token.callback_generation
1871 )
1872 }
1873 (EntryControl::Watchdog(control), CallbackRole::WatchdogWork) => matches!(
1874 control.state,
1875 WatchdogState::AwaitingWork {
1876 attempt_generation,
1877 attempt_status: WatchdogAttemptStatus::Dispatched,
1878 ..
1879 } if attempt_generation == token.callback_generation
1880 ),
1881 (
1882 EntryControl::Ordinary { .. },
1883 CallbackRole::WatchdogScheduler | CallbackRole::WatchdogWork,
1884 )
1885 | (EntryControl::Watchdog(_), CallbackRole::OrdinaryWork) => false,
1886 };
1887 if !valid {
1888 return Err((RegistryError::StaleCallback, handle));
1889 }
1890 let slot = entry.provider_slot_mut(token.role);
1891 if slot.is_some() {
1892 return Err((RegistryError::ProviderHandleAlreadyOwned, handle));
1893 }
1894 *slot = Some(OwnedProviderHandle {
1895 callback_generation: token.callback_generation,
1896 role: token.role,
1897 handle,
1898 });
1899 Ok(())
1900 }
1901
1902 pub(crate) fn take_wakeup_handle(
1903 &mut self,
1904 identity: &TimerIdentity,
1905 ) -> Option<ProviderHandle> {
1906 let entry = self.entries.get_mut(identity)?;
1907 entry.take_wakeup_handle(identity)
1908 }
1909
1910 pub(crate) fn take_work_handle(&mut self, identity: &TimerIdentity) -> Option<ProviderHandle> {
1911 let entry = self.entries.get_mut(identity)?;
1912 entry.take_work_handle(identity)
1913 }
1914
1915 pub(crate) fn take_provider_handles_for_claim(
1916 &mut self,
1917 claim: &RegistrationClaim,
1918 ) -> Result<ProviderHandles, RegistryError> {
1919 let identity = claim.identity.clone();
1920 let entry = self.entry_mut(claim)?;
1921 Ok(entry.take_provider_handles(&identity))
1922 }
1923
1924 pub(crate) fn consume_provider_handle(&mut self, token: &CallbackToken) {
1925 let Some(entry) = self.entries.get_mut(token.identity()) else {
1926 return;
1927 };
1928 if !entry.owns_token_claim(token) {
1929 return;
1930 }
1931 let slot = entry.provider_slot_mut(token.role);
1932 let matches_token = slot.as_ref().is_some_and(|owned| {
1933 owned.callback_generation == token.callback_generation && owned.role == token.role
1934 });
1935 if matches_token {
1936 *slot = None;
1937 }
1938 }
1939
1940 pub(crate) fn confirm_effect_applied(
1946 &mut self,
1947 effect: &RegistryEffect,
1948 ) -> Result<(), RegistryError> {
1949 if !effect.has_valid_shape() {
1950 return Err(RegistryError::StaleCallback);
1951 }
1952 match effect {
1953 RegistryEffect::None | RegistryEffect::ClearCallbacks { .. } => Ok(()),
1954 RegistryEffect::ArmWakeup {
1955 token, delay_ns, ..
1956 } => {
1957 let entry = self.entry_by_token_mut(token, token.role)?;
1958 let valid_generation = match (&entry.control, token.role) {
1959 (EntryControl::Ordinary { control, .. }, CallbackRole::OrdinaryWork) => {
1960 matches!(
1961 control.registration(),
1962 TimerRegistration::Scheduled { generation, .. }
1963 if generation == token.callback_generation
1964 )
1965 }
1966 (EntryControl::Watchdog(control), CallbackRole::WatchdogScheduler) => matches!(
1967 control.state,
1968 WatchdogState::Scheduled {
1969 scheduler_generation,
1970 ..
1971 } if scheduler_generation == token.callback_generation
1972 ),
1973 (EntryControl::Ordinary { .. } | EntryControl::Watchdog(_), _) => false,
1974 };
1975 if !valid_generation {
1976 return Err(RegistryError::StaleCallback);
1977 }
1978 if entry.confirmed_wakeup_generation == Some(token.callback_generation) {
1979 return Ok(());
1980 }
1981 entry.confirmed_wakeup_generation = Some(token.callback_generation);
1982 entry.latest_armed_delay_ns = Some(*delay_ns);
1983 entry.observability.counters_mut().record_wakeup_armed();
1984 Ok(())
1985 }
1986 RegistryEffect::DispatchWatchdog {
1987 successor,
1988 successor_delay_ns,
1989 work,
1990 ..
1991 } => {
1992 let successor_callback_generation = successor.callback_generation;
1993 let successor_role = successor.role;
1994 let entry = self.entry_by_token_mut(successor, successor_role)?;
1995 let EntryControl::Watchdog(control) = &entry.control else {
1996 return Err(RegistryError::StaleCallback);
1997 };
1998 if !matches!(
1999 control.state,
2000 WatchdogState::AwaitingWork {
2001 successor_generation,
2002 attempt_generation,
2003 ..
2004 } if successor_generation == successor_callback_generation
2005 && attempt_generation == work.callback_generation
2006 ) {
2007 return Err(RegistryError::StaleCallback);
2008 }
2009 if entry.confirmed_wakeup_generation == Some(successor_callback_generation)
2010 && entry.confirmed_work_generation == Some(work.callback_generation)
2011 {
2012 return Ok(());
2013 }
2014 entry.confirmed_wakeup_generation = Some(successor_callback_generation);
2015 entry.confirmed_work_generation = Some(work.callback_generation);
2016 entry.latest_armed_delay_ns = Some(*successor_delay_ns);
2017 entry.observability.counters_mut().record_wakeup_armed();
2018 entry.observability.counters_mut().record_work_dispatched();
2019 Ok(())
2020 }
2021 }
2022 }
2023
2024 fn entry(&self, claim: &RegistrationClaim) -> Result<&Entry, RegistryError> {
2025 let entry = self
2026 .entries
2027 .get(claim.identity())
2028 .ok_or(RegistryError::UnknownRegistration)?;
2029 if entry.claim_generation != claim.claim_generation() {
2030 return Err(RegistryError::StaleRegistration);
2031 }
2032 Ok(entry)
2033 }
2034
2035 fn entry_mut(&mut self, claim: &RegistrationClaim) -> Result<&mut Entry, RegistryError> {
2036 let entry = self
2037 .entries
2038 .get_mut(claim.identity())
2039 .ok_or(RegistryError::UnknownRegistration)?;
2040 if entry.claim_generation != claim.claim_generation() {
2041 return Err(RegistryError::StaleRegistration);
2042 }
2043 Ok(entry)
2044 }
2045
2046 fn entry_by_token_mut(
2047 &mut self,
2048 token: &CallbackToken,
2049 role: CallbackRole,
2050 ) -> Result<&mut Entry, RegistryError> {
2051 let entry = self
2052 .entries
2053 .get_mut(token.identity())
2054 .ok_or(RegistryError::StaleCallback)?;
2055 if !entry.owns_token_claim(token) || token.role != role {
2056 return Err(RegistryError::StaleCallback);
2057 }
2058 Ok(entry)
2059 }
2060
2061 fn running_work_entry(&self, token: &CallbackToken) -> Result<&Entry, RegistryError> {
2062 let entry = self
2063 .entries
2064 .get(token.identity())
2065 .ok_or(RegistryError::StaleCallback)?;
2066 if !entry.owns_token_claim(token) {
2067 return Err(RegistryError::StaleCallback);
2068 }
2069 let active = match (&entry.control, token.role) {
2070 (EntryControl::Ordinary { control, .. }, CallbackRole::OrdinaryWork) => matches!(
2071 control.registration(),
2072 TimerRegistration::Running { generation }
2073 if generation == token.callback_generation
2074 ),
2075 (EntryControl::Watchdog(control), CallbackRole::WatchdogWork) => matches!(
2076 control.state,
2077 WatchdogState::AwaitingWork {
2078 attempt_generation,
2079 attempt_status: WatchdogAttemptStatus::Running,
2080 ..
2081 } if attempt_generation == token.callback_generation
2082 ),
2083 (
2084 EntryControl::Ordinary { .. },
2085 CallbackRole::WatchdogScheduler | CallbackRole::WatchdogWork,
2086 )
2087 | (
2088 EntryControl::Watchdog(_),
2089 CallbackRole::OrdinaryWork | CallbackRole::WatchdogScheduler,
2090 ) => false,
2091 };
2092 if active {
2093 Ok(entry)
2094 } else {
2095 Err(RegistryError::StaleCallback)
2096 }
2097 }
2098}
2099
2100fn token_for(
2101 claim: &RegistrationClaim,
2102 callback_generation: u64,
2103 role: CallbackRole,
2104) -> CallbackToken {
2105 CallbackToken::new(
2106 claim.identity.clone(),
2107 claim.claim_generation,
2108 callback_generation,
2109 role,
2110 )
2111}
2112
2113fn detach_provider_handle(
2114 identity: &TimerIdentity,
2115 claim_generation: u64,
2116 owned: OwnedProviderHandle,
2117) -> ProviderHandle {
2118 ProviderHandle {
2119 token: CallbackToken::new(
2120 identity.clone(),
2121 claim_generation,
2122 owned.callback_generation,
2123 owned.role,
2124 ),
2125 handle: owned.handle,
2126 }
2127}
2128
2129const fn memory_sample(start: MemoryPages, end: MemoryPages) -> MemoryPageSample {
2130 MemoryPageSample::new(
2131 MemoryPageExtent::new(start.wasm(), start.stable()),
2132 MemoryPageExtent::new(end.wasm(), end.stable()),
2133 )
2134}
2135
2136const fn clear_callbacks(identity: TimerIdentity, handles: CallbacksToClear) -> RegistryEffect {
2137 RegistryEffect::ClearCallbacks { identity, handles }
2138}
2139
2140fn clear_wakeup_if(identity: TimerIdentity, clear_wakeup: bool) -> RegistryEffect {
2141 if clear_wakeup {
2142 clear_callbacks(identity, CallbacksToClear::Wakeup)
2143 } else {
2144 RegistryEffect::None
2145 }
2146}
2147
2148fn apply_ordinary_action(
2149 entry: &mut Entry,
2150 identity: TimerIdentity,
2151 now_ns: u64,
2152 action: TimerControlAction,
2153 requested: PendingSchedule,
2154) -> RegistryTransition {
2155 match action {
2156 TimerControlAction::Arm {
2157 generation,
2158 deadline_ns,
2159 kind,
2160 } => {
2161 entry.scheduling_mode = requested.mode;
2162 RegistryTransition::normal(RegistryEffect::ArmWakeup {
2163 token: CallbackToken::new(
2164 identity,
2165 entry.claim_generation,
2166 generation,
2167 CallbackRole::OrdinaryWork,
2168 ),
2169 deadline_ns,
2170 delay_ns: deadline_ns.saturating_sub(now_ns),
2171 arm: kind,
2172 })
2173 }
2174 TimerControlAction::None => {
2175 entry.observability.counters_mut().record_coalesced();
2176 RegistryTransition::normal(RegistryEffect::None)
2177 }
2178 TimerControlAction::Clear | TimerControlAction::Disarm { .. } => {
2179 terminal_ordinary(entry, identity, TimerControlError::StaleCompletion)
2180 }
2181 }
2182}
2183
2184fn terminal_ordinary(
2185 entry: &mut Entry,
2186 identity: TimerIdentity,
2187 error: TimerControlError,
2188) -> RegistryTransition {
2189 let failure = map_control_failure(error);
2190 let EntryControl::Ordinary {
2191 control,
2192 pending,
2193 inactive_reason,
2194 } = &mut entry.control
2195 else {
2196 return RegistryTransition::terminal(RegistryEffect::None, failure);
2197 };
2198 let clear_wakeup = control.terminate();
2199 *pending = None;
2200 *inactive_reason = InactiveReason::ControlFailure(failure);
2201 RegistryTransition::terminal(clear_wakeup_if(identity, clear_wakeup), failure)
2202}
2203
2204fn terminal_completion(
2205 entry: &mut Entry,
2206 generation: u64,
2207 work_count: u64,
2208 now_ns: u64,
2209 failure: TimerControlFailure,
2210) -> RegistryTransition {
2211 let EntryControl::Ordinary {
2212 control,
2213 pending,
2214 inactive_reason,
2215 } = &mut entry.control
2216 else {
2217 return RegistryTransition::terminal(RegistryEffect::None, failure);
2218 };
2219 if control.registration() == (TimerRegistration::Running { generation }) {
2220 control.terminate();
2221 }
2222 *pending = None;
2223 *inactive_reason = InactiveReason::ControlFailure(failure);
2224 entry.latest_directive = Some(TimerDirectiveSnapshot::Stop);
2225 entry
2226 .observability
2227 .record_completion(TimerCompletion::invariant_failure(work_count), now_ns);
2228 RegistryTransition::terminal(RegistryEffect::None, failure)
2229}
2230
2231fn invariant_completion(
2232 entry: &mut Entry,
2233 generation: u64,
2234 completion: TimerCompletion,
2235 now_ns: u64,
2236) -> RegistryTransition {
2237 let EntryControl::Ordinary {
2238 control,
2239 pending,
2240 inactive_reason,
2241 } = &mut entry.control
2242 else {
2243 return RegistryTransition::normal(RegistryEffect::None);
2244 };
2245 if control.registration() == (TimerRegistration::Running { generation }) {
2246 control.terminate();
2247 }
2248 *pending = None;
2249 *inactive_reason = InactiveReason::InvariantFailure;
2250 entry.latest_directive = Some(TimerDirectiveSnapshot::Stop);
2251 entry.observability.record_completion(completion, now_ns);
2252 RegistryTransition::normal(RegistryEffect::None)
2253}
2254
2255const fn map_control_failure(error: TimerControlError) -> TimerControlFailure {
2256 match error {
2257 TimerControlError::GenerationExhausted => TimerControlFailure::GenerationExhausted,
2258 TimerControlError::StaleCompletion => TimerControlFailure::DirectiveNotAllowed,
2259 }
2260}
2261
2262const fn map_directive_failure(error: DirectiveError) -> TimerControlFailure {
2263 match error {
2264 DirectiveError::Schedule(ScheduleError::DeadlineOverflow) => {
2265 TimerControlFailure::DeadlineOverflow
2266 }
2267 DirectiveError::Schedule(ScheduleError::DelayOutOfRange) => {
2268 TimerControlFailure::DelayOutOfRange
2269 }
2270 DirectiveError::Schedule(ScheduleError::ZeroCadence) | DirectiveError::MissingCadence => {
2271 TimerControlFailure::DirectiveNotAllowed
2272 }
2273 }
2274}
2275
2276const fn select_completion_schedule(
2277 pending: Option<OrdinaryPending>,
2278 callback: Option<PendingSchedule>,
2279) -> Option<PendingSchedule> {
2280 match pending {
2281 Some(OrdinaryPending::Cancel | OrdinaryPending::Unregister) => None,
2282 Some(OrdinaryPending::Reconcile(pending)) => Some(pending),
2283 Some(OrdinaryPending::Schedule(pending)) => match callback {
2284 Some(callback) if callback.deadline_ns < pending.deadline_ns => Some(callback),
2285 Some(_) | None => Some(pending),
2286 },
2287 None => callback,
2288 }
2289}
2290
2291const fn select_pending_ordinary(
2292 current: Option<OrdinaryPending>,
2293 request: OrdinaryRequest,
2294 requested: PendingSchedule,
2295) -> OrdinaryPending {
2296 if matches!(current, Some(OrdinaryPending::Unregister)) {
2297 return OrdinaryPending::Unregister;
2298 }
2299 match request {
2300 OrdinaryRequest::Reconcile => OrdinaryPending::Reconcile(requested),
2301 OrdinaryRequest::EnsureOnce | OrdinaryRequest::EnsureRecurring => match current {
2302 Some(OrdinaryPending::Reconcile(current) | OrdinaryPending::Schedule(current))
2303 if current.deadline_ns <= requested.deadline_ns =>
2304 {
2305 OrdinaryPending::Schedule(current)
2306 }
2307 Some(
2308 OrdinaryPending::Cancel
2309 | OrdinaryPending::Reconcile(_)
2310 | OrdinaryPending::Schedule(_),
2311 )
2312 | None => OrdinaryPending::Schedule(requested),
2313 Some(OrdinaryPending::Unregister) => OrdinaryPending::Unregister,
2314 },
2315 }
2316}
2317
2318const fn select_pending_watchdog(
2319 pending: Option<WatchdogPending>,
2320 request: WatchdogScheduleRequest,
2321 now_ns: u64,
2322) -> WatchdogPending {
2323 match (pending, request) {
2324 (Some(WatchdogPending::Unregister), _) => WatchdogPending::Unregister,
2325 (_, WatchdogScheduleRequest::Reconcile(requested)) => WatchdogPending::Reconcile(requested),
2326 (Some(WatchdogPending::Reconcile(requested)), WatchdogScheduleRequest::Cadence) => {
2327 WatchdogPending::Reconcile(requested)
2328 }
2329 (Some(WatchdogPending::Reconcile(requested)), WatchdogScheduleRequest::Immediate)
2330 if requested.deadline_ns <= now_ns =>
2331 {
2332 WatchdogPending::Reconcile(requested)
2333 }
2334 (Some(WatchdogPending::EnsureImmediately), _) | (_, WatchdogScheduleRequest::Immediate) => {
2335 WatchdogPending::EnsureImmediately
2336 }
2337 (
2338 Some(WatchdogPending::Cancel | WatchdogPending::Ensure) | None,
2339 WatchdogScheduleRequest::Cadence,
2340 ) => WatchdogPending::Ensure,
2341 }
2342}
2343
2344fn cancel_watchdog(control: &mut WatchdogControl, identity: &TimerIdentity) -> RegistryTransition {
2345 match control.state {
2346 WatchdogState::Inactive => RegistryTransition::normal(RegistryEffect::None),
2347 WatchdogState::AwaitingWork {
2348 attempt_status: WatchdogAttemptStatus::Running,
2349 ..
2350 } => {
2351 if !matches!(control.pending, Some(WatchdogPending::Unregister)) {
2352 control.pending = Some(WatchdogPending::Cancel);
2353 }
2354 RegistryTransition::normal(RegistryEffect::None)
2355 }
2356 WatchdogState::Scheduled { .. }
2357 | WatchdogState::AwaitingWork {
2358 attempt_status: WatchdogAttemptStatus::Dispatched,
2359 ..
2360 } => {
2361 let clear_work = matches!(control.state, WatchdogState::AwaitingWork { .. });
2362 let Some((scheduler_generation, attempt_generation)) =
2363 control.next_dispatch_generations()
2364 else {
2365 return control.terminate(
2366 clear_callbacks(
2367 identity.clone(),
2368 CallbacksToClear::wakeup_and_maybe_work(clear_work),
2369 ),
2370 TimerControlFailure::GenerationExhausted,
2371 );
2372 };
2373 control.scheduler_generation = scheduler_generation;
2374 control.attempt_generation = attempt_generation;
2375 control.state = WatchdogState::Inactive;
2376 control.pending = None;
2377 control.inactive_reason = InactiveReason::Cancelled;
2378 RegistryTransition::normal(clear_callbacks(
2379 identity.clone(),
2380 CallbacksToClear::wakeup_and_maybe_work(clear_work),
2381 ))
2382 }
2383 }
2384}
2385
2386fn remove_after(
2387 entries: &mut BTreeMap<TimerIdentity, Entry>,
2388 identity: &TimerIdentity,
2389 transition: RegistryTransition,
2390 remove: bool,
2391) -> RegistryTransition {
2392 if remove {
2393 entries.remove(identity);
2394 }
2395 transition
2396}
2397
2398#[cfg(test)]
2399mod tests;