appcore_scheduler/
runtime.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: runtime.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/22 15:41:18 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/02 13:24:05 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11use super::*;
12
13/// Cancellation handle for a registered task.
14#[derive(Clone)]
15pub struct TaskHandle {
16    id: String,
17    cancelled: Arc<AtomicBool>,
18    scheduler: Weak<SchedulerInner>,
19}
20
21impl TaskHandle {
22    /// Returns the task identity.
23    pub fn id(&self) -> &str {
24        &self.id
25    }
26
27    /// Requests cooperative cancellation.
28    pub fn cancel(&self) {
29        self.cancelled.store(true, Ordering::Release);
30        if let Some(scheduler) = self.scheduler.upgrade() {
31            scheduler.wakeup.notify_all();
32        }
33    }
34
35    /// Reports whether cancellation was requested.
36    pub fn is_cancelled(&self) -> bool {
37        self.cancelled.load(Ordering::Acquire)
38    }
39}
40
41/// Bounded local scheduler with cooperative cancellation and clean shutdown.
42pub struct Scheduler {
43    inner: Arc<SchedulerInner>,
44    coordinator: Mutex<Option<JoinHandle<()>>>,
45}
46
47struct SchedulerInner {
48    config: SchedulerConfig,
49    state: Mutex<SchedulerState>,
50    wakeup: Condvar,
51    shutdown: Arc<AtomicBool>,
52    active: AtomicUsize,
53    sequence: AtomicU64,
54    jitter_state: AtomicU64,
55    workers: Mutex<Vec<JoinHandle<()>>>,
56}
57
58#[derive(Default)]
59struct SchedulerState {
60    tasks: HashMap<String, TaskEntry>,
61}
62
63struct TaskEntry {
64    schedule: ParsedSchedule,
65    retry: RetryPolicy,
66    priority: u8,
67    callback: TaskCallback,
68    cancelled: Arc<AtomicBool>,
69    next_run: SystemTime,
70    running: bool,
71    attempts: u32,
72    order: u64,
73    last_error: Option<String>,
74    trace: Option<TraceContext>,
75}
76
77enum ParsedSchedule {
78    Once,
79    Interval(Duration),
80    Cron(Box<Schedule>),
81}
82
83impl Scheduler {
84    /// Starts a scheduler coordinator with validated bounds.
85    pub fn new(config: SchedulerConfig) -> Result<Self, SchedulerError> {
86        validate_config(&config)?;
87        let inner = Arc::new(SchedulerInner {
88            config,
89            state: Mutex::new(SchedulerState::default()),
90            wakeup: Condvar::new(),
91            shutdown: Arc::new(AtomicBool::new(false)),
92            active: AtomicUsize::new(0),
93            sequence: AtomicU64::new(1),
94            jitter_state: AtomicU64::new(now_seed()),
95            workers: Mutex::new(Vec::new()),
96        });
97        let coordinator_inner = Arc::clone(&inner);
98        let coordinator = thread::Builder::new()
99            .name("appcore-scheduler".to_string())
100            .spawn(move || coordinator_loop(coordinator_inner))
101            .map_err(|_| SchedulerError::WorkerPanicked)?;
102        Ok(Self {
103            inner,
104            coordinator: Mutex::new(Some(coordinator)),
105        })
106    }
107
108    /// Registers a task and callback.
109    pub fn schedule(
110        &self,
111        task: ScheduledTask,
112        callback: TaskCallback,
113    ) -> Result<TaskHandle, SchedulerError> {
114        if self.inner.shutdown.load(Ordering::Acquire) {
115            return Err(SchedulerError::Shutdown);
116        }
117        task.validate()?;
118        let (schedule, next_run) = parse_schedule(task.schedule)?;
119        let cancelled = Arc::new(AtomicBool::new(false));
120        let mut state = self.inner.state.lock();
121        if self.inner.shutdown.load(Ordering::Acquire) {
122            return Err(SchedulerError::Shutdown);
123        }
124        if state.tasks.contains_key(&task.id) {
125            return Err(SchedulerError::DuplicateTask(task.id));
126        }
127        if state.tasks.len() >= self.inner.config.max_tasks {
128            return Err(SchedulerError::CapacityExceeded {
129                max_tasks: self.inner.config.max_tasks,
130            });
131        }
132        let id = task.id;
133        state.tasks.insert(
134            id.clone(),
135            TaskEntry {
136                schedule,
137                retry: task.retry,
138                priority: task.priority,
139                callback,
140                cancelled: Arc::clone(&cancelled),
141                next_run,
142                running: false,
143                attempts: 0,
144                order: self.inner.sequence.fetch_add(1, Ordering::Relaxed),
145                last_error: None,
146                trace: task.trace,
147            },
148        );
149        drop(state);
150        self.inner.wakeup.notify_all();
151        Ok(TaskHandle {
152            id,
153            cancelled,
154            scheduler: Arc::downgrade(&self.inner),
155        })
156    }
157
158    /// Requests cancellation of a task by identity.
159    pub fn cancel(&self, task_id: &str) -> bool {
160        let state = self.inner.state.lock();
161        let Some(task) = state.tasks.get(task_id) else {
162            return false;
163        };
164        task.cancelled.store(true, Ordering::Release);
165        drop(state);
166        self.inner.wakeup.notify_all();
167        true
168    }
169
170    /// Returns current scheduler and task state.
171    pub fn snapshot(&self) -> SchedulerSnapshot {
172        let state = self.inner.state.lock();
173        let mut tasks = state
174            .tasks
175            .iter()
176            .map(|(id, task)| TaskSnapshot {
177                id: id.clone(),
178                priority: task.priority,
179                running: task.running,
180                attempts: task.attempts,
181                next_run: task.next_run,
182                last_error: task.last_error.clone(),
183                trace: task.trace.clone(),
184            })
185            .collect::<Vec<_>>();
186        tasks.sort_by(|left, right| left.id.cmp(&right.id));
187        SchedulerSnapshot {
188            shutdown: self.inner.shutdown.load(Ordering::Acquire),
189            active_tasks: self.inner.active.load(Ordering::Acquire),
190            tasks,
191        }
192    }
193
194    /// Stops accepting work, cancels tasks and joins all worker threads.
195    pub fn shutdown(&self) -> Result<(), SchedulerError> {
196        self.inner.shutdown.store(true, Ordering::Release);
197        {
198            let state = self.inner.state.lock();
199            for task in state.tasks.values() {
200                task.cancelled.store(true, Ordering::Release);
201            }
202        }
203        self.inner.wakeup.notify_all();
204        if let Some(coordinator) = self.coordinator.lock().take() {
205            coordinator
206                .join()
207                .map_err(|_| SchedulerError::WorkerPanicked)?;
208        }
209        Ok(())
210    }
211}
212
213impl Drop for Scheduler {
214    fn drop(&mut self) {
215        let _ = self.shutdown();
216    }
217}
218
219fn coordinator_loop(inner: Arc<SchedulerInner>) {
220    while !inner.shutdown.load(Ordering::Acquire) {
221        reap_finished_workers(&inner);
222        let available = inner
223            .config
224            .max_concurrent_tasks
225            .saturating_sub(inner.active.load(Ordering::Acquire));
226        let due = {
227            let now = SystemTime::now();
228            let mut state = inner.state.lock();
229            state
230                .tasks
231                .retain(|_, task| task.running || !task.cancelled.load(Ordering::Acquire));
232            let mut due = state
233                .tasks
234                .iter()
235                .filter(|(_, task)| !task.running && task.next_run <= now)
236                .map(|(id, task)| (id.clone(), task.priority, task.next_run, task.order))
237                .collect::<Vec<_>>();
238            due.sort_by(|left, right| {
239                right
240                    .1
241                    .cmp(&left.1)
242                    .then_with(|| left.2.cmp(&right.2))
243                    .then_with(|| left.3.cmp(&right.3))
244            });
245            due.truncate(available);
246            for (id, _, _, _) in &due {
247                if let Some(task) = state.tasks.get_mut(id) {
248                    task.running = true;
249                    task.attempts = task.attempts.saturating_add(1);
250                }
251            }
252            due.into_iter().map(|item| item.0).collect::<Vec<_>>()
253        };
254
255        for task_id in due {
256            spawn_task(&inner, task_id);
257        }
258
259        let mut state = inner.state.lock();
260        if !inner.shutdown.load(Ordering::Acquire) {
261            inner
262                .wakeup
263                .wait_for(&mut state, inner.config.poll_interval);
264        }
265    }
266
267    let workers = std::mem::take(&mut *inner.workers.lock());
268    for worker in workers {
269        let _ = worker.join();
270    }
271    inner.state.lock().tasks.clear();
272}
273
274fn spawn_task(inner: &Arc<SchedulerInner>, task_id: String) {
275    let (callback, context) = {
276        let state = inner.state.lock();
277        let Some(task) = state.tasks.get(&task_id) else {
278            return;
279        };
280        (
281            Arc::clone(&task.callback),
282            TaskContext::new(
283                task_id.clone(),
284                task.attempts,
285                Arc::clone(&task.cancelled),
286                Arc::clone(&inner.shutdown),
287                task.trace.clone(),
288            ),
289        )
290    };
291    inner.active.fetch_add(1, Ordering::AcqRel);
292    let worker_inner = Arc::clone(inner);
293    let worker_task_id = task_id.clone();
294    match thread::Builder::new()
295        .name("appcore-scheduler-worker".to_string())
296        .spawn(move || {
297            let result = catch_unwind(AssertUnwindSafe(|| callback(context)))
298                .unwrap_or_else(|_| Err("task panicked".to_string()));
299            complete_task(&worker_inner, &worker_task_id, result);
300        }) {
301        Ok(handle) => inner.workers.lock().push(handle),
302        Err(_) => {
303            complete_task(inner, &task_id, Err("worker could not start".to_string()));
304        }
305    }
306}
307
308fn complete_task(inner: &SchedulerInner, task_id: &str, result: TaskResult) {
309    let now = SystemTime::now();
310    let mut state = inner.state.lock();
311    let mut remove = false;
312    if let Some(task) = state.tasks.get_mut(task_id) {
313        task.running = false;
314        if inner.shutdown.load(Ordering::Acquire) || task.cancelled.load(Ordering::Acquire) {
315            remove = true;
316        } else if let Err(error) = result {
317            task.last_error = Some(redact_text(&error));
318            if task.attempts < task.retry.max_attempts {
319                if let Some(next_run) =
320                    now.checked_add(retry_delay(&task.retry, task.attempts, inner))
321                {
322                    task.next_run = next_run;
323                } else {
324                    task.last_error = Some("retry schedule exceeds clock range".to_string());
325                    remove = true;
326                }
327            } else {
328                task.attempts = 0;
329                remove = !schedule_next(&task.schedule, now, &mut task.next_run);
330            }
331        } else {
332            task.last_error = None;
333            task.attempts = 0;
334            remove = !schedule_next(&task.schedule, now, &mut task.next_run);
335        }
336    }
337    if remove {
338        state.tasks.remove(task_id);
339    }
340    drop(state);
341    inner.active.fetch_sub(1, Ordering::AcqRel);
342    inner.wakeup.notify_all();
343}
344
345fn schedule_next(schedule: &ParsedSchedule, now: SystemTime, next_run: &mut SystemTime) -> bool {
346    match schedule {
347        ParsedSchedule::Once => false,
348        ParsedSchedule::Interval(every) => {
349            let Some(next) = now.checked_add(*every) else {
350                return false;
351            };
352            *next_run = next;
353            true
354        }
355        ParsedSchedule::Cron(schedule) => {
356            let now: DateTime<Utc> = now.into();
357            let Some(next) = schedule.after(&now).next() else {
358                return false;
359            };
360            *next_run = next.into();
361            true
362        }
363    }
364}
365
366fn parse_schedule(schedule: TaskSchedule) -> Result<(ParsedSchedule, SystemTime), SchedulerError> {
367    let now = SystemTime::now();
368    match schedule {
369        TaskSchedule::Once { run_at } => Ok((ParsedSchedule::Once, run_at)),
370        TaskSchedule::Interval { every, start_at } => {
371            if every.is_zero() {
372                return Err(SchedulerError::InvalidSchedule("zero interval"));
373            }
374            let next_run = match start_at {
375                Some(start_at) => start_at,
376                None => now
377                    .checked_add(every)
378                    .ok_or(SchedulerError::InvalidSchedule(
379                        "interval exceeds clock range",
380                    ))?,
381            };
382            Ok((ParsedSchedule::Interval(every), next_run))
383        }
384        TaskSchedule::Cron { expression } => {
385            let schedule = Schedule::from_str(&expression)
386                .map_err(|error| SchedulerError::InvalidCron(error.to_string()))?;
387            let now_utc: DateTime<Utc> = now.into();
388            let next = schedule
389                .after(&now_utc)
390                .next()
391                .ok_or(SchedulerError::InvalidSchedule(
392                    "cron has no next occurrence",
393                ))?;
394            Ok((ParsedSchedule::Cron(Box::new(schedule)), next.into()))
395        }
396    }
397}
398
399fn retry_delay(policy: &RetryPolicy, attempt: u32, inner: &SchedulerInner) -> Duration {
400    let exponent = attempt.saturating_sub(1).min(31);
401    let factor = u128::from(policy.multiplier).saturating_pow(exponent);
402    let base_ms = policy.initial_backoff.as_millis().saturating_mul(factor);
403    let capped_ms = base_ms.min(policy.max_backoff.as_millis());
404    let jitter_max = policy.jitter.as_millis().min(u128::from(u64::MAX)) as u64;
405    let jitter = if jitter_max == 0 {
406        0
407    } else {
408        next_random(&inner.jitter_state) % jitter_max.saturating_add(1)
409    };
410    Duration::from_millis(
411        capped_ms
412            .min(u128::from(u64::MAX))
413            .saturating_add(u128::from(jitter))
414            .min(u128::from(u64::MAX)) as u64,
415    )
416}
417
418fn next_random(state: &AtomicU64) -> u64 {
419    let mut current = state.load(Ordering::Relaxed);
420    loop {
421        let mut next = current;
422        next ^= next << 13;
423        next ^= next >> 7;
424        next ^= next << 17;
425        match state.compare_exchange_weak(current, next, Ordering::Relaxed, Ordering::Relaxed) {
426            Ok(_) => return next,
427            Err(actual) => current = actual,
428        }
429    }
430}
431
432fn reap_finished_workers(inner: &SchedulerInner) {
433    let completed = {
434        let mut workers = inner.workers.lock();
435        let mut completed = Vec::new();
436        let mut index = 0;
437        while index < workers.len() {
438            if workers[index].is_finished() {
439                completed.push(workers.swap_remove(index));
440            } else {
441                index += 1;
442            }
443        }
444        completed
445    };
446    for worker in completed {
447        let _ = worker.join();
448    }
449}
450
451fn validate_config(config: &SchedulerConfig) -> Result<(), SchedulerError> {
452    if config.max_tasks == 0 {
453        return Err(SchedulerError::InvalidConfig("max_tasks must be positive"));
454    }
455    if config.max_concurrent_tasks == 0 {
456        return Err(SchedulerError::InvalidConfig(
457            "max_concurrent_tasks must be positive",
458        ));
459    }
460    if config.poll_interval.is_zero() {
461        return Err(SchedulerError::InvalidConfig(
462            "poll_interval must be positive",
463        ));
464    }
465    Ok(())
466}
467
468fn now_seed() -> u64 {
469    SystemTime::now()
470        .duration_since(UNIX_EPOCH)
471        .map(|duration| duration.as_nanos() as u64)
472        .unwrap_or(1)
473        .max(1)
474}
appcore_scheduler/runtime.rs

appcore_scheduler/
runtime.rs
