aster_forge_tasks/
schedule.rs

1//! Scheduled runtime task catalog and runner primitives.
2//!
3//! A scheduled task is a product-owned runtime job with a stable name and interval. Forge keeps
4//! the reusable coordination contract here: products register catalog entries, a store atomically
5//! claims due firings, and the runner records one panic-protected execution before advancing the
6//! next due timestamp. Concrete persistence is supplied by another crate, typically
7//! `aster_forge_db`.
8
9use std::future::Future;
10use std::marker::PhantomData;
11use std::time::Duration;
12
13use chrono::{DateTime, Utc};
14use futures::FutureExt;
15use tokio_util::sync::CancellationToken;
16use tracing::Instrument;
17
18use crate::runtime::panic_payload_message;
19use crate::{
20    BackgroundTasks, RecordedTaskHooks, RegisteredRuntimeTaskKind, periodic_sleep_duration,
21};
22
23/// One scheduled runtime task entry registered by a product runtime.
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub struct ScheduledTaskCatalogEntry<'a> {
26    /// Product namespace.
27    pub namespace: &'a str,
28    /// Stable task wire name.
29    pub task_name: &'a str,
30    /// Operator-facing display name.
31    pub display_name: &'a str,
32    /// First due timestamp used when inserting a new catalog row.
33    pub first_run_at: DateTime<Utc>,
34}
35
36/// Request to atomically claim one due scheduled task firing.
37#[derive(Debug, Clone, Copy, PartialEq, Eq)]
38pub struct ScheduledTaskClaimRequest<'a> {
39    /// Product namespace.
40    pub namespace: &'a str,
41    /// Stable task wire name.
42    pub task_name: &'a str,
43    /// Process-unique runtime owner id.
44    pub owner_id: &'a str,
45    /// Current timestamp.
46    pub now: DateTime<Utc>,
47    /// Claim TTL. Another runtime may reclaim after this duration.
48    pub claim_ttl: Duration,
49}
50
51/// Claimed scheduled task firing.
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct ScheduledTaskClaim {
54    /// Stable row identifier.
55    pub task_id: String,
56    /// Product namespace.
57    pub namespace: String,
58    /// Stable task wire name.
59    pub task_name: String,
60    /// Runtime owner id that owns this claim.
61    pub owner_id: String,
62    /// Due timestamp that was claimed.
63    pub scheduled_at: DateTime<Utc>,
64    /// Claim acquisition timestamp.
65    pub claimed_at: DateTime<Utc>,
66    /// Claim expiry timestamp.
67    pub claim_expires_at: DateTime<Utc>,
68}
69
70/// Completion update for a claimed scheduled task.
71#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct ScheduledTaskCompletion {
73    /// Claimed firing to complete.
74    pub claim: ScheduledTaskClaim,
75    /// Runtime completion timestamp.
76    pub finished_at: DateTime<Utc>,
77    /// Next due timestamp after this completion.
78    pub next_run_at: DateTime<Utc>,
79}
80
81/// Renewal update for a claim whose task body is still running.
82///
83/// The store must apply the same ownership predicate as completion (task id,
84/// owner id, and claim acquisition timestamp), so a renewal can never revive a
85/// claim another runtime has already reclaimed.
86#[derive(Debug, Clone, Copy, PartialEq, Eq)]
87pub struct ScheduledTaskClaimRenewal<'a> {
88    /// Owned claim to renew.
89    pub claim: &'a ScheduledTaskClaim,
90    /// Renewal timestamp.
91    pub now: DateTime<Utc>,
92    /// Fresh claim TTL applied from `now`.
93    pub claim_ttl: Duration,
94}
95
96/// Persistence contract used by scheduled task runners.
97#[async_trait::async_trait]
98pub trait ScheduledTaskStore: Clone + Send + Sync + 'static {
99    /// Store error type.
100    type Error: std::fmt::Display + Send + Sync + 'static;
101
102    /// Ensures one scheduled task is present in the catalog.
103    async fn ensure_scheduled_task(
104        &self,
105        entry: ScheduledTaskCatalogEntry<'_>,
106    ) -> std::result::Result<(), Self::Error>;
107
108    /// Attempts to claim one due scheduled task firing.
109    async fn claim_scheduled_task(
110        &self,
111        request: ScheduledTaskClaimRequest<'_>,
112    ) -> std::result::Result<Option<ScheduledTaskClaim>, Self::Error>;
113
114    /// Renews an owned claim while the task body is still running.
115    ///
116    /// Returning `Ok(false)` means the ownership predicate did not match, so
117    /// the worker must treat the claim as lost and stop renewing. Returning
118    /// `Err(_)` is treated as transient and retried on the next renewal tick.
119    async fn renew_scheduled_task_claim(
120        &self,
121        renewal: ScheduledTaskClaimRenewal<'_>,
122    ) -> std::result::Result<bool, Self::Error>;
123
124    /// Completes a claimed firing and advances the next due timestamp.
125    async fn complete_scheduled_task(
126        &self,
127        completion: ScheduledTaskCompletion,
128    ) -> std::result::Result<bool, Self::Error>;
129}
130
131/// Configuration for one scheduled periodic runtime task worker.
132pub struct ScheduledPeriodicTask<Name, State, Store, IntervalFn, TaskFn, PanicFn, RecordFn> {
133    /// Product task identifier.
134    pub name: Name,
135    /// Product namespace.
136    pub namespace: &'static str,
137    /// Stable task wire name.
138    pub task_name: &'static str,
139    /// Operator-facing display name.
140    pub display_name: &'static str,
141    /// Process-unique runtime owner id.
142    pub owner_id: String,
143    /// Claim TTL used to recover from crashed workers.
144    pub claim_ttl: Duration,
145    /// Reads the latest product-configured interval.
146    pub interval_fn: IntervalFn,
147    /// Optional upper bound for positive jitter.
148    pub jitter_cap: Option<Duration>,
149    /// Shared shutdown token.
150    pub shutdown_token: CancellationToken,
151    /// Product runtime state passed to callbacks.
152    pub state: State,
153    /// Scheduled task store.
154    pub store: Store,
155    /// Product callbacks for execution, panic conversion, and recording.
156    pub hooks: RecordedTaskHooks<TaskFn, PanicFn, RecordFn>,
157}
158
159/// Configuration for a leased group of scheduled runtime tasks.
160///
161/// This is the high-level entrypoint for multi-instance Aster services. Forge
162/// generates the process owner id, supervises the runtime lease, creates the
163/// lease-scoped [`BackgroundTasks`] group, and wires every declared scheduled
164/// task into the shared catalog store. Product code only declares singleton
165/// workers and scheduled task bodies through [`ScheduledRuntimeTaskGroup`].
166#[derive(Clone)]
167pub struct LeasedScheduledRuntimeConfig<
168    Name,
169    Outcome,
170    State,
171    LeaseStore,
172    ScheduleStore,
173    PanicFn,
174    RecordFn,
175> {
176    namespace: &'static str,
177    lease_id: String,
178    lease_store: LeaseStore,
179    schedule_store: ScheduleStore,
180    claim_ttl: Duration,
181    lease_ttl: Duration,
182    lease_renew_interval: Duration,
183    lease_standby_retry_interval: Duration,
184    state: State,
185    panic_outcome: PanicFn,
186    record_outcome: RecordFn,
187    _name: PhantomData<fn() -> Name>,
188    _outcome: PhantomData<fn() -> Outcome>,
189}
190
191impl<Name, Outcome, State, LeaseStore, ScheduleStore, PanicFn, RecordFn>
192    LeasedScheduledRuntimeConfig<Name, Outcome, State, LeaseStore, ScheduleStore, PanicFn, RecordFn>
193{
194    /// Creates configuration for one leased scheduled runtime task group.
195    pub fn new<RecordFut>(
196        namespace: &'static str,
197        lease_id: impl Into<String>,
198        lease_store: LeaseStore,
199        schedule_store: ScheduleStore,
200        state: State,
201        panic_outcome: PanicFn,
202        record_outcome: RecordFn,
203    ) -> Self
204    where
205        PanicFn: Fn(String) -> Outcome,
206        RecordFn:
207            Fn(State, Name, ScheduledTaskClaim, DateTime<Utc>, DateTime<Utc>, Outcome) -> RecordFut,
208        RecordFut: Future<Output = ()> + Send + 'static,
209    {
210        Self {
211            namespace,
212            lease_id: lease_id.into(),
213            lease_store,
214            schedule_store,
215            claim_ttl: Duration::from_mins(2),
216            lease_ttl: aster_forge_runtime::DEFAULT_RUNTIME_LEASE_TTL,
217            lease_renew_interval: Duration::from_secs(10),
218            lease_standby_retry_interval: aster_forge_runtime::DEFAULT_RUNTIME_LEASE_RETRY_INTERVAL,
219            state,
220            panic_outcome,
221            record_outcome,
222            _name: PhantomData,
223            _outcome: PhantomData,
224        }
225    }
226
227    /// Sets the scheduled task claim TTL.
228    #[must_use]
229    pub const fn claim_ttl(mut self, claim_ttl: Duration) -> Self {
230        self.claim_ttl = claim_ttl;
231        self
232    }
233
234    /// Sets the runtime lease TTL.
235    #[must_use]
236    pub const fn lease_ttl(mut self, lease_ttl: Duration) -> Self {
237        self.lease_ttl = lease_ttl;
238        self
239    }
240
241    /// Sets the runtime lease renewal interval for the active owner.
242    #[must_use]
243    pub const fn lease_renew_interval(mut self, lease_renew_interval: Duration) -> Self {
244        self.lease_renew_interval = lease_renew_interval;
245        self
246    }
247
248    /// Sets the standby retry interval while another process owns the lease.
249    #[must_use]
250    pub const fn lease_standby_retry_interval(
251        mut self,
252        lease_standby_retry_interval: Duration,
253    ) -> Self {
254        self.lease_standby_retry_interval = lease_standby_retry_interval;
255        self
256    }
257
258    /// Runs this configured leased scheduled runtime group until shutdown.
259    ///
260    /// Prefer this method at product entrypoints because it keeps the call
261    /// shaped like a component declaration: configure shared resources once,
262    /// then declare workers and scheduled tasks in the closure.
263    pub async fn run<ConfigureFn>(self, shutdown_token: CancellationToken, configure: ConfigureFn)
264    where
265        Name: RegisteredRuntimeTaskKind + Send + Sync + 'static,
266        State: Clone + Send + Sync + 'static,
267        LeaseStore: aster_forge_runtime::RuntimeLeaseStore,
268        ScheduleStore: ScheduledTaskStore,
269        ConfigureFn: for<'a> FnMut(
270                &mut ScheduledRuntimeTaskGroup<
271                    'a,
272                    Name,
273                    State,
274                    ScheduleStore,
275                    PanicFn,
276                    RecordFn,
277                    Outcome,
278                >,
279            ) + Send
280            + 'static,
281        PanicFn: Clone + Fn(String) -> Outcome + Send + Sync + 'static,
282        RecordFn: Clone + Send + Sync + 'static,
283        Outcome: Send + 'static,
284    {
285        run_leased_scheduled_runtime_tasks(self, shutdown_token, configure).await;
286    }
287
288    fn into_parts(
289        self,
290    ) -> LeasedScheduledRuntimeParts<State, LeaseStore, ScheduleStore, PanicFn, RecordFn> {
291        let owner_id = aster_forge_runtime::new_runtime_lease_owner_id();
292        let lease_config =
293            aster_forge_runtime::RuntimeLeaseConfig::new(self.lease_id, owner_id.clone())
294                .ttl(self.lease_ttl)
295                .renew_interval(self.lease_renew_interval)
296                .standby_retry_interval(self.lease_standby_retry_interval);
297        LeasedScheduledRuntimeParts {
298            namespace: self.namespace,
299            owner_id,
300            lease_store: self.lease_store,
301            schedule_store: self.schedule_store,
302            claim_ttl: self.claim_ttl,
303            state: self.state,
304            panic_outcome: self.panic_outcome,
305            record_outcome: self.record_outcome,
306            lease_config,
307        }
308    }
309}
310
311struct LeasedScheduledRuntimeParts<State, LeaseStore, ScheduleStore, PanicFn, RecordFn> {
312    namespace: &'static str,
313    owner_id: String,
314    lease_store: LeaseStore,
315    schedule_store: ScheduleStore,
316    claim_ttl: Duration,
317    state: State,
318    panic_outcome: PanicFn,
319    record_outcome: RecordFn,
320    lease_config: aster_forge_runtime::RuntimeLeaseConfig,
321}
322
323/// Lease-scoped task group used by product registration closures.
324///
325/// A value of this type exists only while Forge is building the worker group
326/// for one lease acquisition. Use [`Self::worker`] for singleton workers that
327/// should run only on the active owner, and [`Self::scheduled`] for tasks that
328/// should additionally coordinate each firing through the scheduled task
329/// catalog.
330pub struct ScheduledRuntimeTaskGroup<'a, Name, State, Store, PanicFn, RecordFn, Outcome> {
331    tasks: &'a mut BackgroundTasks,
332    namespace: &'static str,
333    owner_id: String,
334    claim_ttl: Duration,
335    shutdown_token: CancellationToken,
336    state: State,
337    store: Store,
338    panic_outcome: PanicFn,
339    record_outcome: RecordFn,
340    _name: std::marker::PhantomData<Name>,
341    _outcome: std::marker::PhantomData<Outcome>,
342}
343
344impl<Name, State, Store, PanicFn, RecordFn, Outcome>
345    ScheduledRuntimeTaskGroup<'_, Name, State, Store, PanicFn, RecordFn, Outcome>
346where
347    State: Clone + Send + Sync + 'static,
348{
349    /// Spawns one lease-scoped singleton worker into this group.
350    pub fn worker<WorkerFn, WorkerFut>(&mut self, worker: WorkerFn)
351    where
352        WorkerFn: FnOnce(CancellationToken, State) -> WorkerFut,
353        WorkerFut: Future<Output = ()> + Send + 'static,
354    {
355        self.tasks
356            .push(worker(self.shutdown_token.clone(), self.state.clone()));
357    }
358
359    /// Returns a clone of the lease-scoped shutdown token.
360    pub fn shutdown_token(&self) -> CancellationToken {
361        self.shutdown_token.clone()
362    }
363
364    /// Returns a clone of the product runtime state.
365    pub fn state(&self) -> State {
366        self.state.clone()
367    }
368}
369
370impl<Name, State, Store, PanicFn, RecordFn, Outcome>
371    ScheduledRuntimeTaskGroup<'_, Name, State, Store, PanicFn, RecordFn, Outcome>
372where
373    Name: RegisteredRuntimeTaskKind + Send + Sync + 'static,
374    State: Clone + Send + Sync + 'static,
375    Store: ScheduledTaskStore,
376    PanicFn: Clone + Fn(String) -> Outcome + Send + Sync + 'static,
377    RecordFn: Clone + Send + Sync + 'static,
378    Outcome: Send + 'static,
379{
380    /// Registers one scheduled runtime task in the lease-scoped worker group.
381    pub fn scheduled<IntervalFn, TaskFn, TaskFut, RecordFut>(
382        &mut self,
383        name: Name,
384        interval_fn: IntervalFn,
385        jitter_cap: Option<Duration>,
386        task_fn: TaskFn,
387    ) where
388        IntervalFn: Fn(&State) -> Duration + Send + Sync + 'static,
389        TaskFn: Fn(State) -> TaskFut + Send + Sync + 'static,
390        TaskFut: Future<Output = Outcome> + Send + 'static,
391        RecordFn:
392            Fn(State, Name, ScheduledTaskClaim, DateTime<Utc>, DateTime<Utc>, Outcome) -> RecordFut,
393        RecordFut: Future<Output = ()> + Send + 'static,
394    {
395        self.tasks
396            .push(run_scheduled_periodic_task(ScheduledPeriodicTask {
397                name,
398                namespace: self.namespace,
399                task_name: name.as_str(),
400                display_name: name.display_name(),
401                owner_id: self.owner_id.clone(),
402                claim_ttl: self.claim_ttl,
403                interval_fn,
404                jitter_cap,
405                shutdown_token: self.shutdown_token.clone(),
406                state: self.state.clone(),
407                store: self.store.clone(),
408                hooks: RecordedTaskHooks::new(
409                    task_fn,
410                    self.panic_outcome.clone(),
411                    self.record_outcome.clone(),
412                ),
413            }));
414    }
415}
416
417/// Runs a lease-supervised scheduled runtime task group until shutdown.
418///
419/// Forge owns the lifecycle glue: owner id generation, runtime lease
420/// supervision, lease-scoped shutdown token creation, scheduled task catalog
421/// registration, and graceful worker shutdown. Product code supplies the
422/// runtime config and a closure that declares workers and scheduled tasks.
423async fn run_leased_scheduled_runtime_tasks<
424    Name,
425    Outcome,
426    State,
427    LeaseStore,
428    ScheduleStore,
429    ConfigureFn,
430    PanicFn,
431    RecordFn,
432>(
433    config: LeasedScheduledRuntimeConfig<
434        Name,
435        Outcome,
436        State,
437        LeaseStore,
438        ScheduleStore,
439        PanicFn,
440        RecordFn,
441    >,
442    shutdown_token: CancellationToken,
443    mut configure: ConfigureFn,
444) where
445    Name: RegisteredRuntimeTaskKind + Send + Sync + 'static,
446    State: Clone + Send + Sync + 'static,
447    LeaseStore: aster_forge_runtime::RuntimeLeaseStore,
448    ScheduleStore: ScheduledTaskStore,
449    ConfigureFn: for<'a> FnMut(
450            &mut ScheduledRuntimeTaskGroup<
451                'a,
452                Name,
453                State,
454                ScheduleStore,
455                PanicFn,
456                RecordFn,
457                Outcome,
458            >,
459        ) + Send
460        + 'static,
461    PanicFn: Clone + Fn(String) -> Outcome + Send + Sync + 'static,
462    RecordFn: Clone + Send + Sync + 'static,
463    Outcome: Send + 'static,
464{
465    let parts = config.into_parts();
466    let LeasedScheduledRuntimeParts {
467        namespace,
468        owner_id,
469        lease_store,
470        schedule_store,
471        claim_ttl,
472        state,
473        panic_outcome,
474        record_outcome,
475        lease_config,
476    } = parts;
477
478    aster_forge_runtime::run_runtime_lease_supervisor(
479        lease_store,
480        lease_config,
481        shutdown_token,
482        move |leased_shutdown_token| {
483            let mut tasks = BackgroundTasks::with_shutdown_token(leased_shutdown_token);
484            let group_shutdown_token = tasks.shutdown_token();
485            let mut group = ScheduledRuntimeTaskGroup {
486                tasks: &mut tasks,
487                namespace,
488                owner_id: owner_id.clone(),
489                claim_ttl,
490                shutdown_token: group_shutdown_token,
491                state: state.clone(),
492                store: schedule_store.clone(),
493                panic_outcome: panic_outcome.clone(),
494                record_outcome: record_outcome.clone(),
495                _name: std::marker::PhantomData,
496                _outcome: std::marker::PhantomData,
497            };
498            configure(&mut group);
499            tasks
500        },
501        |background_tasks| async move {
502            background_tasks.shutdown().await;
503        },
504    )
505    .await;
506}
507
508/// Runs a scheduled periodic task until shutdown.
509///
510/// Unlike [`crate::run_periodic_task`], this runner first claims a due catalog row. If the row is
511/// not due, or another process owns a fresh claim, the worker skips that iteration. Successful and
512/// failed task outcomes both complete the claim and advance `next_run_at`; crashes and process
513/// exits before completion are recovered by claim expiry.
514///
515/// While the task body runs, a renewal loop extends the claim at
516/// [`scheduled_claim_renew_interval`] ticks, so a task that outlives `claim_ttl` is not reclaimed
517/// and executed twice by another runtime. Renewal failures never abort the task body: a lost
518/// claim only stops the renewal loop, and completion still guards on ownership.
519pub async fn run_scheduled_periodic_task<
520    Name,
521    State,
522    Store,
523    IntervalFn,
524    TaskFn,
525    TaskFut,
526    PanicFn,
527    RecordFn,
528    RecordFut,
529    Outcome,
530>(
531    task: ScheduledPeriodicTask<Name, State, Store, IntervalFn, TaskFn, PanicFn, RecordFn>,
532) where
533    Name: Copy + Send + 'static,
534    State: Clone + Send + Sync + 'static,
535    Store: ScheduledTaskStore,
536    IntervalFn: Fn(&State) -> Duration + Send + Sync + 'static,
537    TaskFn: Fn(State) -> TaskFut + Send + Sync + 'static,
538    TaskFut: Future<Output = Outcome> + Send + 'static,
539    PanicFn: Fn(String) -> Outcome + Send + Sync + 'static,
540    RecordFn: Fn(State, Name, ScheduledTaskClaim, DateTime<Utc>, DateTime<Utc>, Outcome) -> RecordFut
541        + Send
542        + Sync
543        + 'static,
544    RecordFut: Future<Output = ()> + Send + 'static,
545    Outcome: Send + 'static,
546{
547    if task.shutdown_token.is_cancelled() {
548        return;
549    }
550
551    run_scheduled_periodic_iteration(&task)
552        .instrument(tracing::info_span!("bg_task", task.name = task.task_name))
553        .await;
554
555    loop {
556        let sleep_duration =
557            periodic_sleep_duration((task.interval_fn)(&task.state), task.jitter_cap);
558        tokio::select! {
559            biased;
560            () = task.shutdown_token.cancelled() => break,
561            () = tokio::time::sleep(sleep_duration) => {}
562        }
563
564        if task.shutdown_token.is_cancelled() {
565            break;
566        }
567
568        run_scheduled_periodic_iteration(&task)
569            .instrument(tracing::info_span!("bg_task", task.name = task.task_name))
570            .await;
571    }
572}
573
574/// Derives the claim renewal tick from the claim TTL.
575///
576/// Renewing three times per TTL window means two consecutive missed ticks still
577/// leave one renewal before expiry. The floor keeps `tokio::time::interval`
578/// away from a zero period for pathological TTLs.
579#[must_use]
580pub fn scheduled_claim_renew_interval(claim_ttl: Duration) -> Duration {
581    (claim_ttl / 3).max(Duration::from_millis(10))
582}
583
584/// Renews one owned scheduled task claim until stopped or the claim is lost.
585///
586/// Mirrors the background task heartbeat loop: `Ok(false)` means the ownership
587/// predicate no longer matches (another runtime reclaimed the firing), so the
588/// loop stops; `Err(_)` is logged and retried on the next tick. The loop never
589/// aborts the running task body — completion still guards on ownership.
590pub async fn run_scheduled_claim_renewal_loop<Store>(
591    store: Store,
592    claim: ScheduledTaskClaim,
593    claim_ttl: Duration,
594    interval: Duration,
595    stop_token: CancellationToken,
596) where
597    Store: ScheduledTaskStore,
598{
599    let mut ticker = tokio::time::interval(interval);
600    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
601    ticker.tick().await;
602
603    loop {
604        tokio::select! {
605            () = stop_token.cancelled() => return,
606            _ = ticker.tick() => {
607                let renewal = ScheduledTaskClaimRenewal {
608                    claim: &claim,
609                    now: Utc::now(),
610                    claim_ttl,
611                };
612                let result = tokio::select! {
613                    () = stop_token.cancelled() => return,
614                    result = store.renew_scheduled_task_claim(renewal) => result,
615                };
616
617                match result {
618                    Ok(true) => {}
619                    Ok(false) => {
620                        tracing::warn!(
621                            task.name = claim.task_name,
622                            "scheduled task claim lost; stopping claim renewal"
623                        );
624                        return;
625                    }
626                    Err(error) => {
627                        tracing::warn!(
628                            task.name = claim.task_name,
629                            error = %error,
630                            "scheduled task claim renewal failed; retrying next tick"
631                        );
632                    }
633                }
634            }
635        }
636    }
637}
638
639#[expect(
640    clippy::too_many_lines,
641    reason = "Catalog ensure, fenced claim, renewal, panic capture, recording, and completion form one scheduled iteration."
642)]
643async fn run_scheduled_periodic_iteration<
644    Name,
645    State,
646    Store,
647    IntervalFn,
648    TaskFn,
649    TaskFut,
650    PanicFn,
651    RecordFn,
652    RecordFut,
653    Outcome,
654>(
655    task: &ScheduledPeriodicTask<Name, State, Store, IntervalFn, TaskFn, PanicFn, RecordFn>,
656) where
657    Name: Copy + Send + 'static,
658    State: Clone + Send + Sync + 'static,
659    Store: ScheduledTaskStore,
660    IntervalFn: Fn(&State) -> Duration + Send + Sync + 'static,
661    TaskFn: Fn(State) -> TaskFut + Send + Sync + 'static,
662    TaskFut: Future<Output = Outcome> + Send + 'static,
663    PanicFn: Fn(String) -> Outcome + Send + Sync + 'static,
664    RecordFn: Fn(State, Name, ScheduledTaskClaim, DateTime<Utc>, DateTime<Utc>, Outcome) -> RecordFut
665        + Send
666        + Sync
667        + 'static,
668    RecordFut: Future<Output = ()> + Send + 'static,
669    Outcome: Send + 'static,
670{
671    let now = Utc::now();
672    if let Err(error) = task
673        .store
674        .ensure_scheduled_task(ScheduledTaskCatalogEntry {
675            namespace: task.namespace,
676            task_name: task.task_name,
677            display_name: task.display_name,
678            first_run_at: now,
679        })
680        .await
681    {
682        tracing::warn!(
683            task.name = task.task_name,
684            error = %error,
685            "failed to ensure scheduled task catalog row"
686        );
687        return;
688    }
689
690    let claim = match task
691        .store
692        .claim_scheduled_task(ScheduledTaskClaimRequest {
693            namespace: task.namespace,
694            task_name: task.task_name,
695            owner_id: &task.owner_id,
696            now,
697            claim_ttl: task.claim_ttl,
698        })
699        .await
700    {
701        Ok(Some(claim)) => claim,
702        Ok(None) => return,
703        Err(error) => {
704            tracing::warn!(
705                task.name = task.task_name,
706                error = %error,
707                "failed to claim scheduled task"
708            );
709            return;
710        }
711    };
712
713    let renewal_stop = task.shutdown_token.child_token();
714    let renewal_handle = tokio::spawn(run_scheduled_claim_renewal_loop(
715        task.store.clone(),
716        claim.clone(),
717        task.claim_ttl,
718        scheduled_claim_renew_interval(task.claim_ttl),
719        renewal_stop.clone(),
720    ));
721
722    let started_at = Utc::now();
723    let outcome = match std::panic::AssertUnwindSafe((task.hooks.task_fn)(task.state.clone()))
724        .catch_unwind()
725        .await
726    {
727        Ok(outcome) => outcome,
728        Err(panic) => {
729            let panic_message = panic_payload_message(&panic);
730            tracing::error!(
731                task.name = task.task_name,
732                "scheduled task panicked: {panic_message}"
733            );
734            (task.hooks.panic_outcome)(panic_message)
735        }
736    };
737    let finished_at = Utc::now();
738
739    renewal_stop.cancel();
740    if let Err(error) = renewal_handle.await {
741        tracing::warn!(
742            task.name = task.task_name,
743            error = %error,
744            "scheduled task claim renewal worker stopped unexpectedly"
745        );
746    }
747
748    let record_result = std::panic::AssertUnwindSafe((task.hooks.record_outcome)(
749        task.state.clone(),
750        task.name,
751        claim.clone(),
752        started_at,
753        finished_at,
754        outcome,
755    ))
756    .catch_unwind()
757    .await;
758    if let Err(panic) = record_result {
759        let panic_message = panic_payload_message(&panic);
760        tracing::error!(
761            task.name = task.task_name,
762            "scheduled task outcome recorder panicked: {panic_message}"
763        );
764        return;
765    }
766
767    let Some(next_run_at) = next_scheduled_run_at(finished_at, (task.interval_fn)(&task.state))
768    else {
769        tracing::warn!(
770            task.name = task.task_name,
771            "scheduled task interval overflowed while computing next run"
772        );
773        return;
774    };
775
776    match task
777        .store
778        .complete_scheduled_task(ScheduledTaskCompletion {
779            claim,
780            finished_at,
781            next_run_at,
782        })
783        .await
784    {
785        Ok(true) => {}
786        Ok(false) => {
787            tracing::warn!(
788                task.name = task.task_name,
789                "scheduled task claim was not completed because ownership changed"
790            );
791        }
792        Err(error) => {
793            tracing::warn!(
794                task.name = task.task_name,
795                error = %error,
796                "failed to complete scheduled task claim"
797            );
798        }
799    }
800}
801
802/// Computes the next run timestamp after a completed scheduled task firing.
803#[must_use]
804pub fn next_scheduled_run_at(
805    finished_at: DateTime<Utc>,
806    interval: Duration,
807) -> Option<DateTime<Utc>> {
808    let interval = chrono::Duration::from_std(interval).ok()?;
809    finished_at.checked_add_signed(interval)
810}
811
812#[cfg(test)]
813mod tests {
814    use std::sync::Arc;
815    use std::sync::Mutex;
816    use std::sync::atomic::{AtomicUsize, Ordering};
817
818    use async_trait::async_trait;
819    use chrono::{TimeZone, Utc};
820    use tokio_util::sync::CancellationToken;
821
822    use super::{
823        LeasedScheduledRuntimeConfig, ScheduledPeriodicTask, ScheduledRuntimeTaskGroup,
824        ScheduledTaskCatalogEntry, ScheduledTaskClaim, ScheduledTaskClaimRequest,
825        ScheduledTaskCompletion, ScheduledTaskStore, next_scheduled_run_at,
826        run_scheduled_claim_renewal_loop, run_scheduled_periodic_task,
827        scheduled_claim_renew_interval,
828    };
829    use crate::{RecordedTaskHooks, RegisteredRuntimeTaskKind};
830
831    #[derive(Clone)]
832    struct MemoryScheduleStore {
833        calls: Arc<AtomicUsize>,
834        completions: Arc<AtomicUsize>,
835        renewals: Arc<AtomicUsize>,
836        renewal_script: Arc<Mutex<std::collections::VecDeque<Result<bool, String>>>>,
837    }
838
839    #[expect(
840        clippy::trivially_copy_pass_by_ref,
841        reason = "The scheduled task interval callback receives shared state by reference; this fixture uses unit state."
842    )]
843    fn test_interval(_: &()) -> std::time::Duration {
844        std::time::Duration::from_mins(1)
845    }
846
847    #[derive(Clone)]
848    struct AlwaysAcquireLeaseStore {
849        acquired: Arc<AtomicUsize>,
850        released: Arc<AtomicUsize>,
851    }
852
853    impl AlwaysAcquireLeaseStore {
854        fn new() -> Self {
855            Self {
856                acquired: Arc::new(AtomicUsize::new(0)),
857                released: Arc::new(AtomicUsize::new(0)),
858            }
859        }
860    }
861
862    #[async_trait]
863    impl aster_forge_runtime::RuntimeLeaseStore for AlwaysAcquireLeaseStore {
864        type Error = String;
865
866        async fn try_acquire(
867            &self,
868            _claim: aster_forge_runtime::RuntimeLeaseClaim<'_>,
869        ) -> Result<aster_forge_runtime::RuntimeLeaseAcquire, Self::Error> {
870            self.acquired.fetch_add(1, Ordering::SeqCst);
871            Ok(aster_forge_runtime::RuntimeLeaseAcquire::Acquired)
872        }
873
874        async fn renew(
875            &self,
876            _lease_id: &str,
877            _owner_id: &str,
878            _now: chrono::DateTime<Utc>,
879            _expires_at: chrono::DateTime<Utc>,
880        ) -> Result<bool, Self::Error> {
881            Ok(true)
882        }
883
884        async fn release(&self, _lease_id: &str, _owner_id: &str) -> Result<(), Self::Error> {
885            self.released.fetch_add(1, Ordering::SeqCst);
886            Ok(())
887        }
888    }
889
890    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
891    enum TestRuntimeTask {
892        Cleanup,
893    }
894
895    impl RegisteredRuntimeTaskKind for TestRuntimeTask {
896        fn as_str(self) -> &'static str {
897            "cleanup"
898        }
899
900        fn display_name(self) -> &'static str {
901            "Cleanup"
902        }
903
904        fn from_wire_value(value: &str) -> Option<Self> {
905            (value == "cleanup").then_some(Self::Cleanup)
906        }
907    }
908
909    #[async_trait]
910    impl ScheduledTaskStore for MemoryScheduleStore {
911        type Error = String;
912
913        async fn ensure_scheduled_task(
914            &self,
915            entry: ScheduledTaskCatalogEntry<'_>,
916        ) -> Result<(), Self::Error> {
917            assert_eq!(entry.namespace, "aster_test");
918            assert_eq!(entry.task_name, "cleanup");
919            Ok(())
920        }
921
922        async fn claim_scheduled_task(
923            &self,
924            request: ScheduledTaskClaimRequest<'_>,
925        ) -> Result<Option<ScheduledTaskClaim>, Self::Error> {
926            if self.calls.fetch_add(1, Ordering::SeqCst) > 0 {
927                return Ok(None);
928            }
929            Ok(Some(ScheduledTaskClaim {
930                task_id: "aster_test:cleanup".to_string(),
931                namespace: request.namespace.to_string(),
932                task_name: request.task_name.to_string(),
933                owner_id: request.owner_id.to_string(),
934                scheduled_at: request.now,
935                claimed_at: request.now,
936                claim_expires_at: request.now,
937            }))
938        }
939
940        async fn renew_scheduled_task_claim(
941            &self,
942            renewal: super::ScheduledTaskClaimRenewal<'_>,
943        ) -> Result<bool, Self::Error> {
944            assert_eq!(renewal.claim.task_name, "cleanup");
945            assert!(!renewal.claim_ttl.is_zero());
946            self.renewals.fetch_add(1, Ordering::SeqCst);
947            let scripted = self
948                .renewal_script
949                .lock()
950                .expect("renewal script should lock")
951                .pop_front();
952            scripted.unwrap_or(Ok(true))
953        }
954
955        async fn complete_scheduled_task(
956            &self,
957            completion: ScheduledTaskCompletion,
958        ) -> Result<bool, Self::Error> {
959            assert_eq!(completion.claim.task_name, "cleanup");
960            assert!(completion.next_run_at >= completion.finished_at);
961            self.completions.fetch_add(1, Ordering::SeqCst);
962            Ok(true)
963        }
964    }
965
966    fn memory_store() -> MemoryScheduleStore {
967        MemoryScheduleStore {
968            calls: Arc::new(AtomicUsize::new(0)),
969            completions: Arc::new(AtomicUsize::new(0)),
970            renewals: Arc::new(AtomicUsize::new(0)),
971            renewal_script: Arc::new(Mutex::new(std::collections::VecDeque::new())),
972        }
973    }
974
975    #[tokio::test]
976    async fn scheduled_periodic_task_claims_records_and_completes_one_due_run() {
977        let shutdown = CancellationToken::new();
978        let ran = Arc::new(AtomicUsize::new(0));
979        let recorded = Arc::new(AtomicUsize::new(0));
980        let store = memory_store();
981        let completions = store.completions.clone();
982        let ran_for_task = ran.clone();
983        let recorded_for_hook = recorded.clone();
984        let shutdown_for_hook = shutdown.clone();
985
986        run_scheduled_periodic_task(ScheduledPeriodicTask {
987            name: "cleanup",
988            namespace: "aster_test",
989            task_name: "cleanup",
990            display_name: "Cleanup",
991            owner_id: "runtime-a".to_string(),
992            claim_ttl: std::time::Duration::from_secs(30),
993            interval_fn: test_interval,
994            jitter_cap: None,
995            shutdown_token: shutdown.clone(),
996            state: (),
997            store,
998            hooks: RecordedTaskHooks::new(
999                move |()| {
1000                    let ran = ran_for_task.clone();
1001                    async move {
1002                        ran.fetch_add(1, Ordering::SeqCst);
1003                        "ok"
1004                    }
1005                },
1006                |_| "panic",
1007                move |(),
1008                      _name: &str,
1009                      claim: ScheduledTaskClaim,
1010                      _started_at,
1011                      _finished_at,
1012                      outcome| {
1013                    let recorded = recorded_for_hook.clone();
1014                    let shutdown = shutdown_for_hook.clone();
1015                    async move {
1016                        assert_eq!(claim.task_name, "cleanup");
1017                        assert_eq!(outcome, "ok");
1018                        recorded.fetch_add(1, Ordering::SeqCst);
1019                        shutdown.cancel();
1020                    }
1021                },
1022            ),
1023        })
1024        .await;
1025
1026        assert_eq!(ran.load(Ordering::SeqCst), 1);
1027        assert_eq!(recorded.load(Ordering::SeqCst), 1);
1028        assert_eq!(completions.load(Ordering::SeqCst), 1);
1029    }
1030
1031    #[tokio::test]
1032    async fn scheduled_periodic_task_records_panic_outcome_and_completes_claim() {
1033        let shutdown = CancellationToken::new();
1034        let store = memory_store();
1035        let completions = store.completions.clone();
1036        let recorded = Arc::new(Mutex::new(Vec::new()));
1037        let recorded_for_hook = recorded.clone();
1038        let shutdown_for_hook = shutdown.clone();
1039
1040        run_scheduled_periodic_task(ScheduledPeriodicTask {
1041            name: "cleanup",
1042            namespace: "aster_test",
1043            task_name: "cleanup",
1044            display_name: "Cleanup",
1045            owner_id: "runtime-a".to_string(),
1046            claim_ttl: std::time::Duration::from_secs(30),
1047            interval_fn: test_interval,
1048            jitter_cap: None,
1049            shutdown_token: shutdown.clone(),
1050            state: (),
1051            store,
1052            hooks: RecordedTaskHooks::new(
1053                move |()| async move {
1054                    panic!("scheduled body failed");
1055                    #[expect(
1056                        unreachable_code,
1057                        reason = "The test intentionally panics to exercise scheduled-task failure handling."
1058                    )]
1059                    "ok".to_string()
1060                },
1061                |message| format!("panic:{message}"),
1062                move |(), _name, _claim, _started_at, _finished_at, outcome| {
1063                    let recorded = recorded_for_hook.clone();
1064                    let shutdown = shutdown_for_hook.clone();
1065                    async move {
1066                        recorded
1067                            .lock()
1068                            .expect("recorded outcomes should lock")
1069                            .push(outcome);
1070                        shutdown.cancel();
1071                    }
1072                },
1073            ),
1074        })
1075        .await;
1076
1077        assert_eq!(
1078            recorded
1079                .lock()
1080                .expect("recorded outcomes should lock")
1081                .as_slice(),
1082            ["panic:scheduled body failed"]
1083        );
1084        assert_eq!(completions.load(Ordering::SeqCst), 1);
1085    }
1086
1087    #[tokio::test]
1088    async fn scheduled_periodic_task_does_not_complete_claim_when_recorder_panics() {
1089        let shutdown = CancellationToken::new();
1090        let store = memory_store();
1091        let completions = store.completions.clone();
1092
1093        run_scheduled_periodic_task(ScheduledPeriodicTask {
1094            name: "cleanup",
1095            namespace: "aster_test",
1096            task_name: "cleanup",
1097            display_name: "Cleanup",
1098            owner_id: "runtime-a".to_string(),
1099            claim_ttl: std::time::Duration::from_secs(30),
1100            interval_fn: test_interval,
1101            jitter_cap: None,
1102            shutdown_token: shutdown.clone(),
1103            state: (),
1104            store,
1105            hooks: RecordedTaskHooks::new(
1106                move |()| {
1107                    let shutdown = shutdown.clone();
1108                    async move {
1109                        shutdown.cancel();
1110                        "ok"
1111                    }
1112                },
1113                |_| "panic",
1114                move |(), _name, _claim, _started_at, _finished_at, _outcome| async move {
1115                    panic!("record failed");
1116                },
1117            ),
1118        })
1119        .await;
1120
1121        assert_eq!(completions.load(Ordering::SeqCst), 0);
1122    }
1123
1124    #[tokio::test]
1125    async fn leased_scheduled_runtime_group_runs_worker_and_scheduled_task() {
1126        let lease_store = AlwaysAcquireLeaseStore::new();
1127        let acquired = lease_store.acquired.clone();
1128        let released = lease_store.released.clone();
1129        let schedule_store = memory_store();
1130        let completions = schedule_store.completions.clone();
1131        let worker_runs = Arc::new(AtomicUsize::new(0));
1132        let scheduled_runs = Arc::new(AtomicUsize::new(0));
1133        let recorded_runs = Arc::new(AtomicUsize::new(0));
1134        let shutdown = CancellationToken::new();
1135        let config = LeasedScheduledRuntimeConfig::new(
1136            "aster_test",
1137            "aster_test.background",
1138            lease_store,
1139            schedule_store,
1140            (),
1141            |_| "panic",
1142            {
1143                let recorded_runs = recorded_runs.clone();
1144                let shutdown = shutdown.clone();
1145                move |(), _name, claim: ScheduledTaskClaim, _started_at, _finished_at, outcome| {
1146                    let recorded_runs = recorded_runs.clone();
1147                    let shutdown = shutdown.clone();
1148                    async move {
1149                        assert_eq!(claim.task_name, "cleanup");
1150                        assert_eq!(outcome, "ok");
1151                        recorded_runs.fetch_add(1, Ordering::SeqCst);
1152                        shutdown.cancel();
1153                    }
1154                }
1155            },
1156        )
1157        .claim_ttl(std::time::Duration::from_secs(30))
1158        .lease_ttl(std::time::Duration::from_secs(30))
1159        .lease_renew_interval(std::time::Duration::from_secs(10))
1160        .lease_standby_retry_interval(std::time::Duration::from_secs(5));
1161        let worker_runs_for_group = worker_runs.clone();
1162        let scheduled_runs_for_group = scheduled_runs.clone();
1163
1164        config
1165            .run(
1166                shutdown.clone(),
1167                move |group: &mut ScheduledRuntimeTaskGroup<
1168                    '_,
1169                    TestRuntimeTask,
1170                    (),
1171                    _,
1172                    _,
1173                    _,
1174                    &'static str,
1175                >| {
1176                    let worker_runs = worker_runs_for_group.clone();
1177                    group.worker(move |shutdown_token, ()| async move {
1178                        worker_runs.fetch_add(1, Ordering::SeqCst);
1179                        shutdown_token.cancelled().await;
1180                    });
1181                    let scheduled_runs = scheduled_runs_for_group.clone();
1182                    group.scheduled(TestRuntimeTask::Cleanup, test_interval, None, move |()| {
1183                        let scheduled_runs = scheduled_runs.clone();
1184                        async move {
1185                            scheduled_runs.fetch_add(1, Ordering::SeqCst);
1186                            "ok"
1187                        }
1188                    });
1189                },
1190            )
1191            .await;
1192
1193        assert_eq!(acquired.load(Ordering::SeqCst), 1);
1194        assert_eq!(released.load(Ordering::SeqCst), 1);
1195        assert_eq!(worker_runs.load(Ordering::SeqCst), 1);
1196        assert_eq!(scheduled_runs.load(Ordering::SeqCst), 1);
1197        assert_eq!(recorded_runs.load(Ordering::SeqCst), 1);
1198        assert_eq!(completions.load(Ordering::SeqCst), 1);
1199    }
1200
1201    #[test]
1202    fn next_scheduled_run_at_adds_interval() {
1203        let finished_at = Utc.with_ymd_and_hms(2026, 6, 26, 1, 2, 3).unwrap();
1204        assert_eq!(
1205            next_scheduled_run_at(finished_at, std::time::Duration::from_mins(1)),
1206            Some(Utc.with_ymd_and_hms(2026, 6, 26, 1, 3, 3).unwrap())
1207        );
1208    }
1209
1210    fn test_claim() -> ScheduledTaskClaim {
1211        let now = Utc.with_ymd_and_hms(2026, 6, 26, 1, 0, 0).unwrap();
1212        ScheduledTaskClaim {
1213            task_id: "aster_test:cleanup".to_string(),
1214            namespace: "aster_test".to_string(),
1215            task_name: "cleanup".to_string(),
1216            owner_id: "runtime-a".to_string(),
1217            scheduled_at: now,
1218            claimed_at: now,
1219            claim_expires_at: now,
1220        }
1221    }
1222
1223    #[test]
1224    fn claim_renew_interval_is_one_third_of_ttl_with_floor() {
1225        assert_eq!(
1226            scheduled_claim_renew_interval(std::time::Duration::from_mins(2)),
1227            std::time::Duration::from_secs(40)
1228        );
1229        assert_eq!(
1230            scheduled_claim_renew_interval(std::time::Duration::from_millis(30)),
1231            std::time::Duration::from_millis(10)
1232        );
1233        // A zero TTL never reaches the store, but the interval must not panic.
1234        assert_eq!(
1235            scheduled_claim_renew_interval(std::time::Duration::ZERO),
1236            std::time::Duration::from_millis(10)
1237        );
1238    }
1239
1240    #[tokio::test]
1241    async fn claim_renewal_loop_renews_until_stopped() {
1242        let store = memory_store();
1243        let renewals = store.renewals.clone();
1244        let stop = CancellationToken::new();
1245
1246        let handle = tokio::spawn(run_scheduled_claim_renewal_loop(
1247            store,
1248            test_claim(),
1249            std::time::Duration::from_secs(30),
1250            std::time::Duration::from_millis(1),
1251            stop.clone(),
1252        ));
1253
1254        while renewals.load(Ordering::SeqCst) == 0 {
1255            tokio::task::yield_now().await;
1256        }
1257        stop.cancel();
1258        handle.await.expect("renewal loop should stop cleanly");
1259        assert!(renewals.load(Ordering::SeqCst) >= 1);
1260    }
1261
1262    #[tokio::test]
1263    async fn claim_renewal_loop_stops_when_claim_is_lost() {
1264        let store = memory_store();
1265        store
1266            .renewal_script
1267            .lock()
1268            .expect("renewal script should lock")
1269            .push_back(Ok(false));
1270        let renewals = store.renewals.clone();
1271
1272        // Ok(false) means the ownership predicate no longer matches: the loop
1273        // must stop on its own instead of hammering a reclaimed row.
1274        run_scheduled_claim_renewal_loop(
1275            store,
1276            test_claim(),
1277            std::time::Duration::from_secs(30),
1278            std::time::Duration::from_millis(1),
1279            CancellationToken::new(),
1280        )
1281        .await;
1282
1283        assert_eq!(renewals.load(Ordering::SeqCst), 1);
1284    }
1285
1286    #[tokio::test]
1287    async fn claim_renewal_loop_retries_after_transient_errors() {
1288        let store = memory_store();
1289        {
1290            let mut script = store
1291                .renewal_script
1292                .lock()
1293                .expect("renewal script should lock");
1294            script.push_back(Err("database temporarily unavailable".to_string()));
1295            script.push_back(Err("database temporarily unavailable".to_string()));
1296        }
1297        let renewals = store.renewals.clone();
1298        let stop = CancellationToken::new();
1299
1300        let handle = tokio::spawn(run_scheduled_claim_renewal_loop(
1301            store,
1302            test_claim(),
1303            std::time::Duration::from_secs(30),
1304            std::time::Duration::from_millis(1),
1305            stop.clone(),
1306        ));
1307
1308        while renewals.load(Ordering::SeqCst) < 3 {
1309            tokio::task::yield_now().await;
1310        }
1311        stop.cancel();
1312        handle.await.expect("renewal loop should stop cleanly");
1313        assert!(renewals.load(Ordering::SeqCst) >= 3);
1314    }
1315
1316    #[tokio::test]
1317    async fn scheduled_periodic_task_renews_claim_while_task_body_runs() {
1318        let shutdown = CancellationToken::new();
1319        let store = memory_store();
1320        let renewals = store.renewals.clone();
1321        let completions = store.completions.clone();
1322        let renewals_for_task = renewals.clone();
1323        let shutdown_for_hook = shutdown.clone();
1324
1325        run_scheduled_periodic_task(ScheduledPeriodicTask {
1326            name: "cleanup",
1327            namespace: "aster_test",
1328            task_name: "cleanup",
1329            display_name: "Cleanup",
1330            owner_id: "runtime-a".to_string(),
1331            // 30ms TTL derives a 10ms renewal interval, so a body that waits for
1332            // two renewals provably outlives the original claim window.
1333            claim_ttl: std::time::Duration::from_millis(30),
1334            interval_fn: test_interval,
1335            jitter_cap: None,
1336            shutdown_token: shutdown.clone(),
1337            state: (),
1338            store,
1339            hooks: RecordedTaskHooks::new(
1340                move |()| {
1341                    let renewals = renewals_for_task.clone();
1342                    async move {
1343                        tokio::time::timeout(std::time::Duration::from_secs(5), async move {
1344                            while renewals.load(Ordering::SeqCst) < 2 {
1345                                tokio::task::yield_now().await;
1346                            }
1347                        })
1348                        .await
1349                        .expect("claim should be renewed while the task body runs");
1350                        "ok"
1351                    }
1352                },
1353                |_| "panic",
1354                move |(), _name, _claim, _started_at, _finished_at, _outcome| {
1355                    let shutdown = shutdown_for_hook.clone();
1356                    async move {
1357                        shutdown.cancel();
1358                    }
1359                },
1360            ),
1361        })
1362        .await;
1363
1364        assert_eq!(completions.load(Ordering::SeqCst), 1);
1365        let renewed = renewals.load(Ordering::SeqCst);
1366        assert!(renewed >= 2);
1367        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1368        assert_eq!(
1369            renewals.load(Ordering::SeqCst),
1370            renewed,
1371            "renewal loop must stop before the claim is completed"
1372        );
1373    }
1374}
aster_forge_tasks/schedule.rs

aster_forge_tasks/
schedule.rs
