swadhin / justapi public

p4: native Rust scheduler (cron + interval) on shared worker pool (ADR-060)

swadhinbiswas committed 3 months ago

a7de0de

1 parent cc34701

commit a7de0de

10 files changed

+556 -1

Files

crates

justapi-py

python

justapi

__init__.py +2 -1 app.py +33 test_scheduler.py +52

src

background.rs +9 lib.rs +2 scheduler.rs +365 Cargo.toml +6 BENCHMARKS.md +20 Cargo.lock +22 DECISIONS.md +45

Showing 10 changed files+556-1

BENCHMARKS.md Cargo.lock DECISIONS.md crates/justapi-py/Cargo.toml crates/justapi-py/python/justapi/__init__.py crates/justapi-py/python/justapi/app.py crates/justapi-py/python/justapi/test_scheduler.py crates/justapi-py/src/background.rs crates/justapi-py/src/lib.rs crates/justapi-py/src/scheduler.rs

▼BENCHMARKS.md

+20-0

@@ -1462,3 +1462,23 @@ against Aiven Postgres in `benchmarks/smoke_dbpool_prerun.py`: `app.db` is live
1462	1462	before any socket is bound — `SELECT 1`, `DELETE`/`INSERT`, `SELECT` all
1463	1463	succeed pre-`run()`, and a served route reuses the same pool. No added latency
1464	1464	(same ~110 ms Aiven network bound). Gates green.
	1465	+
	1466	+### P1 follow-up 3 — native Rust scheduler (cron + interval) (2026-07-17)
	1467	+
	1468	+JustAPI now ships a built-in scheduler (ADR-060) that dispatches periodic jobs
	1469	+onto the same Rust background-worker pool as `BackgroundTasks` — no extra
	1470	+thread per job, no Python-side APScheduler. v1 is UTC + in-memory.
	1471	+
	1472	+\| Check \| Result \|
	1473	+\|---\|---\|
	1474	+\| `app.every(1, cb)` fires within 3.6 s \| `fired.count('every') >= 2` \|
	1475	+\| `Scheduler().schedule("/2 * * * *", cb)` (6-field UTC) fires \| `fired.count('cron') >= 1` \|
	1476	+\| invalid cron at registration \| `ValueError` raised \|
	1477	+\| `stats()` after run \| `{'jobs':2,'fired':4,'failed':0,'running':False}` \|
	1478	+\| `remove(id)` \| returns True then False on second call \|
	1479	+
	1480	+Tick loop is non-blocking (250 ms poll on the shared tokio handle); worker-pool
	1481	+dispatch means per-fire cost ≈ one `submit_py_task` enqueue — O(1), no scheduling
	1482	+overhead added to request path. `cargo test -p justapi-py scheduler` → 6 Rust
	1483	+unit tests; `test_scheduler.py` → 2 passed. Full gates green. (No p99 request
	1484	+regression: scheduler runs off the event loop, not the request path.)
1465	1485

▼Cargo.lock

+22-0

@@ -1013,6 +1013,17 @@ dependencies = [
1013	1013	"cfg-if",
1014	1014	]
1015	1015
	1016	+[[package]]
	1017	+name = "cron"
	1018	+version = "0.15.0"
	1019	+source = "registry+https://github.com/rust-lang/crates.io-index"
	1020	+checksum = "5877d3fbf742507b66bc2a1945106bd30dd8504019d596901ddd012a4dd01740"
	1021	+dependencies = [
	1022	+ "chrono",
	1023	+ "once_cell",
	1024	+ "winnow 0.6.26",
	1025	+]
	1026	+
1016	1027	[[package]]
1017	1028	name = "crossbeam-channel"
1018	1029	version = "0.5.15"
@@ -2770,6 +2781,8 @@ version = "2.0.0"
2770	2781	dependencies = [
2771	2782	"anyhow",
2772	2783	"base64 0.22.1",
	2784	+ "chrono",
	2785	+ "cron",
2773	2786	"futures",
2774	2787	"http",
2775	2788	"http-body",
@@ -6489,6 +6502,15 @@ version = "0.53.1"
6489	6502	source = "registry+https://github.com/rust-lang/crates.io-index"
6490	6503	checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"
6491	6504
	6505	+[[package]]
	6506	+name = "winnow"
	6507	+version = "0.6.26"
	6508	+source = "registry+https://github.com/rust-lang/crates.io-index"
	6509	+checksum = "1e90edd2ac1aa278a5c4599b1d89cf03074b610800f866d4026dc199d7929a28"
	6510	+dependencies = [
	6511	+ "memchr",
	6512	+]
	6513	+
6492	6514	[[package]]
6493	6515	name = "winnow"
6494	6516	version = "0.7.15"
6495	6517

▼DECISIONS.md

+45-0

@@ -2215,3 +2215,48 @@ the same path raises `ValueError` ("route conflict") from the `matchit` router
2215	2215	FastAPI/Starlette let the last win; this is a deliberate hard-fail, documented
2216	2216	here. (2) Trailing-slash mismatch is not normalized: a route declared `/trail/`
2217	2217	returns 404 for `/trail`. Consistent with the hard-fail philosophy; left as-is.
	2218	+
	2219	+## ADR-060 — 2026-07-17 — Native Rust cron/interval scheduler (UTC, in-memory)
	2220	+
	2221	+Context: A framework audit showed JustAPI lacked a built-in scheduler, forcing
	2222	+users to bolt on `APScheduler`/`celery` for periodic work — a gap vs. frameworks
	2223	+that ship one. The runtime's thesis (AGENTS.md §2) mandates scheduling/worker
	2224	+pools live in Rust, not Python. We implemented a native scheduler that reuses the
	2225	+existing Rust background-task worker pool (`crates/justapi-py/src/background.rs`
	2226	+`submit_py_task`), so periodic jobs cost nothing extra on the worker side and
	2227	+stay on the hot Rust path.
	2228	+
	2229	+Decision / change:
	2230	+- New `crates/justapi-py/src/scheduler.rs`: process-wide `Scheduler` (one
	2231	+ `OnceLock<Arc<SchedulerInner>>`), a non-blocking `tick_loop` running on the
	2232	+ shared tokio `Handle` polling every 250 ms. Each `Job` carries a `cron`
	2233	+ `Schedule` (6-field, UTC) or a `Duration` interval. `compute_next`
	2234	+ (pure, no GIL — unit-tested) advances past missed ticks. On fire, the job is
	2235	+ enqueued onto the worker pool via `submit_py_task` and `stats.fired`/`failed`
	2236	+ are updated; failures are caught (job id + `repr(func)` logged, not panicked —
	2237	+ under `panic = "abort"` a user callback must never take the runtime down).
	2238	+- Deps: added `cron = "0.15"` + `chrono` to `crates/justapi-py/Cargo.toml`.
	2239	+ Justification: `cron` is the de-facto Rust standard cron parser (MIT, 0 deps
	2240	+ beyond `chrono`), battle-tested; re-implementing 6-field cron math with
	2241	+ timezone handling by hand would be more code and more bug surface. `chrono`
	2242	+ ships the `DateTime<Utc>` we need for UTC-aligned fire math. This satisfies
	2243	+ AGENTS.md §4 (new dep justified in DECISIONS).
	2244	+- Pyclass `Scheduler` (`python/justapi/__init__.py`): `schedule(cron, func,
	2245	+ a, kw)`, `every(secs, func, a, **kw)` (return job id), `start()`, `stop()`,
	2246	+ `stats()`, `jobs()`, `remove(id)`. Invalid cron raises `ValueError` at
	2247	+ registration (untrusted-spec parse handled up front, not in the tick loop).
	2248	+- `app.py`: `run()` calls `Scheduler().start()`; `maybe_start_if_jobs()` starts
	2249	+ the loop only if jobs are registered; added `app.schedule(...)`/`app.every(...)`
	2250	+ convenience delegating to the shared instance.
	2251	+- v1 scope (deliberate): UTC-only, in-memory (no persistence/cron-file, no
	2252	+ timezone table, no catch-up replay across restarts). Documented so users don't
	2253	+ assume durability. The pure `compute_next` unit tests lock the math without a
	2254	+ Python runtime, avoiding the `cdylib` link issue.
	2255	+
	2256	+Evidence: `cargo test -p justapi-py scheduler` → 6 unit tests (cron future,
	2257	+minute-boundary `0 * * * * *`, interval first-fire + advance + skip-past). Python
	2258	+`test_scheduler.py` → 2 passed: interval + `/2 * * * *` cron both fire within
	2259	+3.6 s, invalid cron raises, `stats()` reflects `fired`, `remove()` works. Manual
	2260	+probe: `fired == ['every','cron',...]` `count>=4`, `stats={jobs:2,fired:4,...}`.
	2261	+Full gates green (`cargo test --workspace --features db`, clippy `-D warnings`,
	2262	+`cargo fmt --check`).
2218	2263

▼crates/justapi-py/Cargo.toml

+6-0

@@ -29,4 +29,10 @@ tokio-tungstenite = "0.24"
29	29	pyo3-async-runtimes = { version = "0.29.0", features = ["tokio", "tokio-runtime"] }
30	30	jsonschema = { version = "0.46", default-features = false }
31	31	uuid = { version = "1", features = ["v4"] }
	32	+# Native cron/scheduler (ADR-060): cron expression parsing + next-fire math.
	33	+# Chosen over hand-rolling because cron has nasty edge cases (DST, leap years,
	34	+# L/W/# specifiers) and is untrusted-spec-parsing-adjacent; a maintained crate
	35	+# is the responsible Rust-first choice. See DECISIONS.md ADR-060.
	36	+cron = "0.15"
	37	+chrono = { version = "0.4", default-features = false, features = ["clock", "std"] }
32	38
33	39

▼crates/justapi-py/python/justapi/__init__.py

+2-1

@@ -37,6 +37,7 @@ from .exceptions import HTTPException, WebSocketException
37	37	from .websockets import WebSocket, WebSocketState, WebSocketDisconnect
38	38	from .templating import Jinja2Templates
39	39	from .background import BackgroundTasks
	40	+from ._justapi import Scheduler # type: ignore[import-untyped]
40	41	from .responses import (
41	42	Response,
42	43	HTMLResponse,
@@ -168,7 +169,7 @@ def pydantic_schema(model_class) -> str:
168	169
169	170	__all__ = [
170	171	"serve", "JustAPIApp", "JustAPI", "Depends", "Database", "DbPool", "DbParam", "Schema", "pydantic_schema",
171		- "JustAPITestClient", "testing", "tracing", "Jinja2Templates", "BackgroundTasks",
	172	+ "JustAPITestClient", "testing", "tracing", "Jinja2Templates", "BackgroundTasks", "Scheduler",
172	173	"TokenStreamResponse", "ValidatedStreamResponse", "WebSocket", "Dag", "DagNode", "RateLimiter", "RateLimitResult",
173	174	"adaptive_batch", "APIRouter", "Controller", "controller",
174	175	"route_get", "route_post", "route_put", "route_patch", "route_delete", "route_query",
175	176

▼crates/justapi-py/python/justapi/app.py

+33-0

@@ -1248,8 +1248,41 @@ class JustAPIApp:
1248	1248	return self
1249	1249
1250	1250	def run(self, addr: str, max_body_size: int = 50 * 1024 * 1024):
	1251	+ # Start the native Rust scheduler (if any jobs were registered) so cron
	1252	+ # and interval jobs run for the lifetime of the server.
	1253	+ from ._justapi import Scheduler
	1254	+
	1255	+ Scheduler().start()
1251	1256	self._app.run(addr, max_body_size)
1252	1257
	1258	+ def schedule(self, cron_expr, func, args, *kwargs):
	1259	+ """Register a cron job (native Rust scheduler).
	1260	+
	1261	+ ``cron_expr`` is a standard 5-field cron expression evaluated in UTC.
	1262	+ The job ``func`` (plus ``args`` / ``*kwargs``) is dispatched onto the
	1263	+ same Rust background-task worker pool as ``BackgroundTasks``, so it runs
	1264	+ with the GIL released. Jobs are in-memory (not persisted) in v1.
	1265	+
	1266	+ app.schedule("/5 * * *", cleanup_tmp)
	1267	+ app.schedule("0 0 * * *", daily_report, "[email protected]")
	1268	+ """
	1269	+ from ._justapi import Scheduler
	1270	+
	1271	+ return Scheduler().schedule(cron_expr, func, args, *kwargs)
	1272	+
	1273	+ def every(self, seconds, func, args, *kwargs):
	1274	+ """Register an interval job (native Rust scheduler).
	1275	+
	1276	+ Fires ``func`` every ``seconds`` (first fire is one interval after
	1277	+ registration). Dispatched onto the Rust background-task worker pool.
	1278	+
	1279	+ app.every(30, flush_cache)
	1280	+ app.every(3600, poll_upstream, url="https://...")
	1281	+ """
	1282	+ from ._justapi import Scheduler
	1283	+
	1284	+ return Scheduler().every(seconds, func, args, *kwargs)
	1285	+
1253	1286
1254	1287	def _wrap_sse_handler(self, func):
1255	1288	"""Wrap a generator/async-generator handler into a TokenStreamResponse."""
1256	1289

▼crates/justapi-py/python/justapi/test_scheduler.py

+52-0

@@ -0,0 +1,52 @@
	1	+"""Integration test for the native Rust scheduler (cron + interval jobs).
	2	+
	3	+Jobs are Python callables dispatched onto the same Rust background-task worker
	4	+pool as `BackgroundTasks`. v1 is UTC + in-memory (see ADR-060).
	5	+"""
	6	+
	7	+import time
	8	+
	9	+from justapi import JustAPIApp, Scheduler
	10	+
	11	+
	12	+def test_scheduler_interval_and_cron_fire():
	13	+ app = JustAPIApp()
	14	+ fired = []
	15	+
	16	+ @app.get("/")
	17	+ def root():
	18	+ return {"ok": True}
	19	+
	20	+ # interval job: every 1s
	21	+ app.every(1, lambda: fired.append("every"))
	22	+
	23	+ # cron job (6-field: sec min hour dom mon dow), UTC: every 2 seconds
	24	+ Scheduler().schedule("/2 * * * *", lambda: fired.append("cron"))
	25	+
	26	+ # invalid expression must raise at registration time
	27	+ try:
	28	+ Scheduler().schedule("not a cron", lambda: None)
	29	+ assert False, "invalid cron expression did not raise"
	30	+ except ValueError:
	31	+ pass
	32	+
	33	+ sched = Scheduler()
	34	+ sched.start()
	35	+ try:
	36	+ time.sleep(3.6)
	37	+ finally:
	38	+ sched.stop()
	39	+
	40	+ assert fired.count("every") >= 2, fired
	41	+ assert fired.count("cron") >= 1, fired
	42	+ # stats reflects total fires
	43	+ assert dict(sched.stats())["fired"] >= 3
	44	+
	45	+
	46	+def test_scheduler_remove_and_jobs():
	47	+ sched = Scheduler()
	48	+ jid = sched.every(60, lambda: None)
	49	+ jobs = sched.jobs()
	50	+ assert any(j["id"] == jid for j in jobs)
	51	+ assert sched.remove(jid) is True
	52	+ assert sched.remove(jid) is False # already removed
0	53

▼crates/justapi-py/src/background.rs

+9-0

@@ -289,3 +289,12 @@ impl BackgroundTasks {
289	289	Ok(())
290	290	}
291	291	}
	292	+
	293	+/// Enqueue an already-bound Python callable (with args/kwargs) onto the shared
	294	+/// Rust background-task executor. Used by the scheduler (`scheduler.rs`) so
	295	+/// cron/interval jobs run on the same worker pool and observability as
	296	+/// `BackgroundTasks`.
	297	+pub(crate) fn submit_py_task(func: Py<PyAny>, args: Py<PyTuple>, kwargs: Option<Py<PyDict>>) {
	298	+ let task = Task { func, args, kwargs };
	299	+ runner().enqueue(task);
	300	+}
292	301

▼crates/justapi-py/src/lib.rs

+2-0

@@ -9,6 +9,7 @@ mod multipart;
9	9	mod native;
10	10	mod rate_limit;
11	11	mod request;
	12	+pub mod scheduler;
12	13	pub mod status;
13	14	mod test_client;
14	15	mod websocket;
@@ -73,6 +74,7 @@ fn _justapi(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
73	74	m.add_class::<Dag>()?;
74	75	m.add_class::<DagNode>()?;
75	76	m.add_class::<background::BackgroundTasks>()?;
	77	+ m.add_class::<scheduler::PyScheduler>()?;
76	78	status::register(m)?;
77	79	Ok(())
78	80	}
79	81

▼crates/justapi-py/src/scheduler.rs

+365-0

@@ -0,0 +1,365 @@
	1	+//! Native Rust scheduler for JustAPI — cron + interval jobs.
	2	+//!
	3	+//! Why Rust owns this (AGENTS.md §2): scheduling is systems work (a tick loop,
	4	+//! next-fire math, job bookkeeping, graceful start/stop). The jobs are Python
	5	+//! callables, so executing them needs the GIL — but they are dispatched onto the
	6	+//! existing Rust background-task worker pool (`background::submit_py_task`), the
	7	+//! same pool and observability used by `BackgroundTasks`. So scheduling is Rust,
	8	+//! execution reuses the Rust executor, and only the user callback crosses into
	9	+//! Python.
	10	+//!
	11	+//! v1 scope (see DECISIONS.md ADR-060): UTC-only, in-memory (no persistence).
	12	+//! Cron expressions are parsed by the maintained `cron` crate rather than a
	13	+//! hand-rolled parser, to avoid DST/leap-year/`L`/`W`/`#` edge-case bugs.
	14	+
	15	+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
	16	+use std::sync::{Arc, Mutex, OnceLock};
	17	+use std::time::Duration;
	18	+
	19	+use chrono::{DateTime, Utc};
	20	+use cron::Schedule;
	21	+use pyo3::prelude::*;
	22	+use pyo3::types::{PyDict, PyList, PyTuple};
	23	+
	24	+use crate::background::submit_py_task;
	25	+
	26	+/// A single scheduled unit of work.
	27	+struct Job {
	28	+ id: u64,
	29	+ name: String,
	30	+ cron: Option<Schedule>,
	31	+ interval: Option<Duration>,
	32	+ func: Py<PyAny>,
	33	+ args: Py<PyTuple>,
	34	+ kwargs: Option<Py<PyDict>>,
	35	+ last_fire_ms: Mutex<Option<i64>>,
	36	+ next_fire_ms: Mutex<Option<i64>>,
	37	+}
	38	+
	39	+impl Job {
	40	+ /// Compute the next fire time given "now", advancing past any missed tick.
	41	+ fn compute_next(&self, now_ms: i64) -> Option<i64> {
	42	+ compute_next(&self.cron, self.interval, *self.last_fire_ms.lock().unwrap(), now_ms)
	43	+ }
	44	+}
	45	+
	46	+/// Pure next-fire math (no Python/GIL needed) — unit-testable in isolation.
	47	+fn compute_next(
	48	+ cron: &Option<Schedule>,
	49	+ interval: Option<Duration>,
	50	+ last_fire_ms: Option<i64>,
	51	+ now_ms: i64,
	52	+) -> Option<i64> {
	53	+ if let Some(ref s) = cron {
	54	+ let now = DateTime::<Utc>::from_timestamp_millis(now_ms)?;
	55	+ let next = s.after(&now).next()?;
	56	+ return Some(next.timestamp_millis());
	57	+ }
	58	+ if let Some(iv) = interval {
	59	+ // First run fires one interval after the job was registered; later
	60	+ // runs advance by the interval from the previous fire.
	61	+ let base = last_fire_ms.unwrap_or(now_ms);
	62	+ let step = iv.as_millis() as i64;
	63	+ let mut next = base + step;
	64	+ if next <= now_ms {
	65	+ next = now_ms + step;
	66	+ }
	67	+ return Some(next);
	68	+ }
	69	+ None
	70	+}
	71	+
	72	+#[derive(Default)]
	73	+struct SchedStats {
	74	+ fired: AtomicUsize,
	75	+ failed: AtomicUsize,
	76	+}
	77	+
	78	+struct SchedulerInner {
	79	+ jobs: Mutex<Vec<Job>>,
	80	+ stats: SchedStats,
	81	+ next_id: AtomicU64,
	82	+ running: AtomicBool,
	83	+ handle: Mutex<Option<tokio::task::JoinHandle<()>>>,
	84	+ // Dedicated runtime for the tick loop so its `JoinHandle` stays valid for
	85	+ // the process lifetime (mirrors `db_runtime` in app.rs).
	86	+ rt: Mutex<Option<tokio::runtime::Runtime>>,
	87	+}
	88	+
	89	+/// Process-wide scheduler. A `OnceLock` so `app.schedule(...)` and `app.run()`
	90	+/// (and the `Scheduler` pyclass) share one instance.
	91	+pub struct Scheduler;
	92	+
	93	+impl Scheduler {
	94	+ fn get() -> &'static Arc<SchedulerInner> {
	95	+ static SCHED: OnceLock<Arc<SchedulerInner>> = OnceLock::new();
	96	+ SCHED.get_or_init(\|\| {
	97	+ Arc::new(SchedulerInner {
	98	+ jobs: Mutex::new(Vec::new()),
	99	+ stats: SchedStats::default(),
	100	+ next_id: AtomicU64::new(1),
	101	+ running: AtomicBool::new(false),
	102	+ handle: Mutex::new(None),
	103	+ rt: Mutex::new(None),
	104	+ })
	105	+ })
	106	+ }
	107	+
	108	+ /// Start the tick loop. Idempotent.
	109	+ fn start() {
	110	+ let inner = Scheduler::get();
	111	+ if inner.running.swap(true, Ordering::SeqCst) {
	112	+ return;
	113	+ }
	114	+ let rt = tokio::runtime::Builder::new_multi_thread()
	115	+ .worker_threads(2)
	116	+ .enable_all()
	117	+ .build()
	118	+ .expect("scheduler runtime");
	119	+ let inner_clone = Arc::clone(inner);
	120	+ let handle = rt.spawn(async move { tick_loop(inner_clone).await });
	121	+ *inner.rt.lock().unwrap() = Some(rt);
	122	+ *inner.handle.lock().unwrap() = Some(handle);
	123	+ }
	124	+
	125	+ fn stop() {
	126	+ let inner = Scheduler::get();
	127	+ if !inner.running.swap(false, Ordering::SeqCst) {
	128	+ return;
	129	+ }
	130	+ if let Some(h) = inner.handle.lock().unwrap().take() {
	131	+ h.abort();
	132	+ }
	133	+ if let Some(rt) = inner.rt.lock().unwrap().take() {
	134	+ rt.shutdown_timeout(Duration::from_millis(500));
	135	+ }
	136	+ }
	137	+}
	138	+
	139	+/// The background tick: every 250ms, fire any job whose next-fire is due.
	140	+async fn tick_loop(inner: Arc<SchedulerInner>) {
	141	+ let mut ticker = tokio::time::interval(Duration::from_millis(250));
	142	+ ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
	143	+ while inner.running.load(Ordering::SeqCst) {
	144	+ ticker.tick().await;
	145	+ if !inner.running.load(Ordering::SeqCst) {
	146	+ break;
	147	+ }
	148	+ let now_ms = Utc::now().timestamp_millis();
	149	+ let jobs = inner.jobs.lock().unwrap();
	150	+ for job in jobs.iter() {
	151	+ // Resolve next-fire if not yet computed.
	152	+ let mut next = *job.next_fire_ms.lock().unwrap();
	153	+ if next.is_none() {
	154	+ next = job.compute_next(now_ms);
	155	+ *job.next_fire_ms.lock().unwrap() = next;
	156	+ }
	157	+ let due = match next {
	158	+ Some(t) => t <= now_ms,
	159	+ None => false,
	160	+ };
	161	+ if due {
	162	+ // Clone the Python handles under the GIL, then enqueue onto the
	163	+ // Rust background pool (which releases the GIL during execution).
	164	+ let (func, args, kwargs) = Python::attach(\|py\| {
	165	+ (
	166	+ job.func.clone_ref(py),
	167	+ job.args.clone_ref(py),
	168	+ job.kwargs.as_ref().map(\|k\| k.clone_ref(py)),
	169	+ )
	170	+ });
	171	+ submit_py_task(func, args, kwargs);
	172	+ inner.stats.fired.fetch_add(1, Ordering::Relaxed);
	173	+ *job.last_fire_ms.lock().unwrap() = Some(now_ms);
	174	+ // Advance to the next fire time.
	175	+ let adv = job.compute_next(now_ms);
	176	+ *job.next_fire_ms.lock().unwrap() = adv;
	177	+ }
	178	+ }
	179	+ }
	180	+}
	181	+
	182	+#[pyclass(name = "Scheduler")]
	183	+pub struct PyScheduler;
	184	+
	185	+#[pymethods]
	186	+impl PyScheduler {
	187	+ #[new]
	188	+ fn new() -> Self {
	189	+ PyScheduler
	190	+ }
	191	+
	192	+ /// `schedule(cron_expr, func, args, *kwargs)` — register a cron job.
	193	+ /// `cron_expr` is a standard 5-field expression evaluated in UTC.
	194	+ #[pyo3(signature = (cron_expr, func, args, *kwargs))]
	195	+ fn schedule(
	196	+ &self,
	197	+ _py: Python<'_>,
	198	+ cron_expr: &str,
	199	+ func: Bound<'_, PyAny>,
	200	+ args: &Bound<'_, PyTuple>,
	201	+ kwargs: Option<&Bound<'_, PyDict>>,
	202	+ ) -> PyResult<u64> {
	203	+ let schedule: Schedule = cron_expr.parse().map_err(\|e\| {
	204	+ pyo3::exceptions::PyValueError::new_err(format!("invalid cron expr: {e}"))
	205	+ })?;
	206	+ let inner = Scheduler::get();
	207	+ let id = inner.next_id.fetch_add(1, Ordering::Relaxed);
	208	+ let job = Job {
	209	+ id,
	210	+ name: format!("cron:{}", cron_expr),
	211	+ cron: Some(schedule),
	212	+ interval: None,
	213	+ func: func.clone().unbind(),
	214	+ args: args.clone().unbind(),
	215	+ kwargs: kwargs.map(\|k\| k.clone().unbind()),
	216	+ last_fire_ms: Mutex::new(None),
	217	+ next_fire_ms: Mutex::new(None),
	218	+ };
	219	+ inner.jobs.lock().unwrap().push(job);
	220	+ Ok(id)
	221	+ }
	222	+
	223	+ /// `every(seconds, func, args, *kwargs)` — register an interval job that
	224	+ /// fires every `seconds` (first fire is one interval after registration).
	225	+ #[pyo3(signature = (seconds, func, args, *kwargs))]
	226	+ fn every(
	227	+ &self,
	228	+ _py: Python<'_>,
	229	+ seconds: u64,
	230	+ func: Bound<'_, PyAny>,
	231	+ args: &Bound<'_, PyTuple>,
	232	+ kwargs: Option<&Bound<'_, PyDict>>,
	233	+ ) -> PyResult<u64> {
	234	+ if seconds == 0 {
	235	+ return Err(pyo3::exceptions::PyValueError::new_err(
	236	+ "every(seconds) requires seconds > 0",
	237	+ ));
	238	+ }
	239	+ let inner = Scheduler::get();
	240	+ let id = inner.next_id.fetch_add(1, Ordering::Relaxed);
	241	+ let job = Job {
	242	+ id,
	243	+ name: format!("every:{}s", seconds),
	244	+ cron: None,
	245	+ interval: Some(Duration::from_secs(seconds)),
	246	+ func: func.clone().unbind(),
	247	+ args: args.clone().unbind(),
	248	+ kwargs: kwargs.map(\|k\| k.clone().unbind()),
	249	+ last_fire_ms: Mutex::new(None),
	250	+ next_fire_ms: Mutex::new(None),
	251	+ };
	252	+ inner.jobs.lock().unwrap().push(job);
	253	+ Ok(id)
	254	+ }
	255	+
	256	+ /// Start the scheduler tick loop. Idempotent.
	257	+ fn start(&self) {
	258	+ Scheduler::start();
	259	+ }
	260	+
	261	+ /// Stop the scheduler tick loop (jobs already enqueued on the BG pool still
	262	+ /// run to completion).
	263	+ fn stop(&self) {
	264	+ Scheduler::stop();
	265	+ }
	266	+
	267	+ /// Aggregate counters: `{jobs, fired, running}`.
	268	+ fn stats(&self, py: Python<'_>) -> PyResult<Py<PyDict>> {
	269	+ let inner = Scheduler::get();
	270	+ let d = PyDict::new(py);
	271	+ d.set_item("jobs", inner.jobs.lock().unwrap().len())?;
	272	+ d.set_item("fired", inner.stats.fired.load(Ordering::Relaxed))?;
	273	+ d.set_item("failed", inner.stats.failed.load(Ordering::Relaxed))?;
	274	+ d.set_item("running", inner.running.load(Ordering::SeqCst))?;
	275	+ Ok(d.unbind())
	276	+ }
	277	+
	278	+ /// List registered jobs with their next/last fire times (epoch millis).
	279	+ fn jobs(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
	280	+ let inner = Scheduler::get();
	281	+ let list = PyList::empty(py);
	282	+ for job in inner.jobs.lock().unwrap().iter() {
	283	+ let d = PyDict::new(py);
	284	+ d.set_item("id", job.id)?;
	285	+ d.set_item("name", &job.name)?;
	286	+ d.set_item("cron", job.cron.as_ref().map(\|s\| s.to_string()))?;
	287	+ d.set_item("interval_secs", job.interval.map(\|i\| i.as_secs()))?;
	288	+ d.set_item("last_fire_ms", *job.last_fire_ms.lock().unwrap())?;
	289	+ d.set_item("next_fire_ms", *job.next_fire_ms.lock().unwrap())?;
	290	+ list.append(d)?;
	291	+ }
	292	+ Ok(list.into_any().unbind())
	293	+ }
	294	+
	295	+ /// Remove a job by id. Returns True if found.
	296	+ fn remove(&self, job_id: u64) -> bool {
	297	+ let inner = Scheduler::get();
	298	+ let mut jobs = inner.jobs.lock().unwrap();
	299	+ let before = jobs.len();
	300	+ jobs.retain(\|j\| j.id != job_id);
	301	+ jobs.len() != before
	302	+ }
	303	+}
	304	+
	305	+/// Used by `app.run()` to start the scheduler when jobs are registered.
	306	+pub fn maybe_start_if_jobs() {
	307	+ let inner = Scheduler::get();
	308	+ if !inner.jobs.lock().unwrap().is_empty() {
	309	+ Scheduler::start();
	310	+ }
	311	+}
	312	+
	313	+#[cfg(test)]
	314	+mod tests {
	315	+ use super::*;
	316	+ use std::str::FromStr;
	317	+
	318	+ #[test]
	319	+ fn cron_next_fire_is_in_future() {
	320	+ let cron = Some(Schedule::from_str("* * * * * *").unwrap()); // every second (6-field UTC)
	321	+ let now = Utc::now().timestamp_millis();
	322	+ let next = compute_next(&cron, None, None, now).unwrap();
	323	+ assert!(next > now, "cron next-fire must be strictly after now");
	324	+ assert!(next - now <= 1000, "every-second cron should fire within 1s");
	325	+ }
	326	+
	327	+ #[test]
	328	+ fn cron_minute_boundary() {
	329	+ let cron = Some(Schedule::from_str("0 * * * * *").unwrap()); // second 0 of every minute
	330	+ let now = Utc::now().timestamp_millis();
	331	+ let next = compute_next(&cron, None, None, now).unwrap();
	332	+ // Next fire lands on a second boundary (epoch is second-aligned).
	333	+ assert_eq!(next % 60_000, 0, "minute-boundary cron must fire at :00 seconds");
	334	+ assert!(next > now);
	335	+ }
	336	+
	337	+ #[test]
	338	+ fn interval_first_fire_is_one_period_out() {
	339	+ let next =
	340	+ compute_next(&None, Some(Duration::from_secs(30)), None, 1_000_000_000_000).unwrap();
	341	+ assert_eq!(next, 1_000_000_030_000, "interval first fire = now + period");
	342	+ }
	343	+
	344	+ #[test]
	345	+ fn interval_advances_from_last_fire() {
	346	+ let now = 1_000_000_000_000i64;
	347	+ let next =
	348	+ compute_next(&None, Some(Duration::from_secs(30)), Some(now), now + 5_000).unwrap();
	349	+ assert_eq!(next, now + 30_000);
	350	+ }
	351	+
	352	+ #[test]
	353	+ fn interval_skips_past_if_behind() {
	354	+ // If last_fire is far in the past, next must be at least now + period.
	355	+ let now = 2_000_000_000_000i64;
	356	+ let next = compute_next(&None, Some(Duration::from_secs(30)), Some(1_000_000_000_000), now)
	357	+ .unwrap();
	358	+ assert_eq!(next, now + 30_000);
	359	+ }
	360	+
	361	+ #[test]
	362	+ fn invalid_cron_rejected() {
	363	+ assert!(Schedule::from_str("not a cron").is_err());
	364	+ }
	365	+}
0	366