Skip to main content

max / makenotwork

36.8 KB · 787 lines History Blame Raw
1 //! Background scheduler, runs periodic jobs on a fixed interval.
2 //!
3 //! Every tick: publish scheduled items/posts, send onboarding emails, dispatch builds.
4 //! Every 5 ticks: sandbox cleanup. Every tick: webhook retry, refund escalation, transaction cleanup.
5 //! Daily: subscription checks, bounce monitoring, session pruning, IP scrubbing, account deletion.
6 //! Weekly: storage drift correction, sales count integrity.
7
8 mod announcements;
9 mod cleanup;
10 /// Re-exported for integration tests: abandoned multipart sessions hold billed
11 /// parts, so the reap path is worth exercising directly. The rest of `cleanup`
12 /// stays private.
13 pub use cleanup::abort_orphan_multipart_sessions;
14 #[doc(hidden)]
15 pub use cleanup::drain_pending_s3_deletions_for_test;
16 mod integrity;
17 mod mt_threads;
18 mod synckit_warnings;
19 mod webhooks;
20
21 use tokio::sync::watch;
22 use tokio::task::JoinHandle;
23
24 use axum::extract::FromRef;
25
26 use crate::constants;
27 use crate::db;
28 use crate::{AppState, Integrations};
29
30 // Re-export public API used by route handlers
31 pub use announcements::{send_blog_post_announcements, send_release_announcements};
32 pub use mt_threads::{spawn_mt_thread_for_blog_post, spawn_mt_thread_for_item};
33
34 /// Advisory lock ID for single-instance scheduler coordination.
35 /// Prevents duplicate job execution during rolling deploys.
36 // Arbitrary fixed key, any stable, unique i64 works for pg_advisory_lock. (The
37 // value is not a real ASCII encoding of anything; an earlier comment claimed
38 // "MNW_SCH" but the literal has a stray nibble and decodes to no such string.
39 // Left as-is because changing it during a rolling deploy would briefly let old
40 // and new instances hold different keys and both run the scheduler.)
41 const SCHEDULER_ADVISORY_LOCK_ID: i64 = 0x04D4_E575_F534_3484;
42
43 /// Weekly drift correction interval in scheduler ticks (10,080 = 7 days at 60s).
44 const DRIFT_CORRECTION_INTERVAL: u64 = 10_080;
45
46 /// Daily interval in scheduler ticks (1,440 = 24h at 60s).
47 const DAILY_INTERVAL: u64 = 1440;
48
49 /// Sandbox cleanup interval in scheduler ticks (5 = 5min at 60s).
50 const SANDBOX_CLEANUP_INTERVAL: u64 = 5;
51
52 /// Hourly interval in scheduler ticks (60 = 1h at 60s). Used by the SyncKit
53 /// usage-warning job.
54 const HOURLY_INTERVAL: u64 = 60;
55
56 /// Soft tick-duration ceiling. A tick longer than this logs WARN; longer
57 /// than `TICK_DURATION_ALERT_SECS` also opens a WAM ticket (rate-limited).
58 /// Tuned conservatively, at 60s interval, anything over 30s means we're
59 /// burning more than half the budget and the next tick will skip.
60 const TICK_DURATION_WARN_SECS: u64 = 30;
61 const TICK_DURATION_ALERT_SECS: u64 = 50;
62 /// Hard ceiling on a single tick. The tick task is awaited while THIS loop task
63 /// holds the cross-instance advisory lock, so a job that hangs (e.g. a query
64 /// blocked on a row lock) would otherwise freeze all background maintenance on
65 /// every instance forever, the panic/overrun alerts only fire on completion or
66 /// panic, never on a hang. Past this bound we abort the tick, alert, and release
67 /// the lock so the next tick (or another instance) recovers. Far above the ~50s
68 /// healthy-tick ceiling, so it only trips on a true wedge.
69 const TICK_WATCHDOG_SECS: u64 = 300;
70
71 /// An import job whose liveness heartbeat is older than this is treated as
72 /// crashed and failed by the hourly reaper. Heartbeats bump after every 50-item
73 /// chunk, so 30 minutes of silence means the owning process is gone, not slow.
74 const STUCK_IMPORT_SECS: i64 = 1800;
75
76 /// Determine which scheduled job groups should run for a given tick.
77 ///
78 /// Returns `(sandbox_cleanup, hourly_jobs, daily_jobs, weekly_jobs)`.
79 fn jobs_for_tick(tick: u64) -> (bool, bool, bool, bool) {
80 let sandbox = tick.is_multiple_of(SANDBOX_CLEANUP_INTERVAL);
81 let hourly = tick.is_multiple_of(HOURLY_INTERVAL);
82 let daily = tick == 1 || tick.is_multiple_of(DAILY_INTERVAL);
83 let weekly = tick.is_multiple_of(DRIFT_CORRECTION_INTERVAL);
84 (sandbox, hourly, daily, weekly)
85 }
86
87 /// Spawn the background scheduler loop. Drop `shutdown_tx` to stop it.
88 pub fn spawn_scheduler(state: AppState, mut shutdown_rx: watch::Receiver<()>) -> JoinHandle<()> {
89 tokio::spawn(async move {
90 tracing::info!(
91 "Scheduler started (interval={}s)",
92 constants::SCHEDULER_INTERVAL_SECS
93 );
94
95 let mut interval = tokio::time::interval(std::time::Duration::from_secs(
96 constants::SCHEDULER_INTERVAL_SECS,
97 ));
98 // Skip (not burst) missed ticks: a long-running pass (storage recalc,
99 // drift checks under the advisory lock) must not trigger a catch-up
100 // burst of back-to-back ticks on the next wake.
101 interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
102 interval.tick().await; // consume immediate first tick
103
104 let mut tick_count: u64 = 0;
105 let mut last_overrun_alert: Option<std::time::Instant> = None;
106 let mut last_panic_alert: Option<std::time::Instant> = None;
107
108 // Dedicated 1-connection pool for the advisory lock, created once. Holding
109 // the lock across a tick on this pool no longer borrows from the 25-conn
110 // request pool (was an accepted residual; Run 9). Falls back to the request
111 // pool if the side-pool can't be created so the scheduler still runs.
112 let lock_pool = sqlx::postgres::PgPoolOptions::new()
113 .max_connections(1)
114 .connect(&state.config.database_url)
115 .await
116 .map_err(|e| tracing::warn!(error = ?e, "scheduler: dedicated lock pool unavailable, using request pool"))
117 .ok();
118
119 loop {
120 tokio::select! {
121 _ = interval.tick() => {}
122 _ = shutdown_rx.changed() => {
123 tracing::info!("Scheduler shutting down");
124 return;
125 }
126 }
127
128 tick_count += 1;
129 let tick_started = std::time::Instant::now();
130
131 // Pin advisory lock to a dedicated connection held for the entire tick.
132 // pg_try_advisory_lock is session-scoped, holding the connection prevents
133 // another instance from acquiring the lock until this tick completes. The
134 // connection comes from the dedicated `lock_pool` (1 conn), or the request
135 // pool if that pool couldn't be created.
136 let lock_source = lock_pool.as_ref().unwrap_or(&state.db);
137 let mut lock_conn = match lock_source.acquire().await {
138 Ok(conn) => conn,
139 Err(e) => {
140 tracing::warn!(error = ?e, "scheduler: failed to acquire connection for advisory lock, skipping tick");
141 continue;
142 }
143 };
144 let locked: bool = match sqlx::query_scalar("SELECT pg_try_advisory_lock($1)")
145 .bind(SCHEDULER_ADVISORY_LOCK_ID)
146 .fetch_one(&mut *lock_conn)
147 .await
148 {
149 Ok(v) => v,
150 Err(e) => {
151 tracing::warn!(error = ?e, "scheduler: failed to acquire advisory lock, skipping tick");
152 continue;
153 }
154 };
155 if !locked {
156 tracing::debug!("scheduler: advisory lock held by another instance, skipping tick");
157 continue;
158 }
159
160 // Run every periodic job in a supervised child task. A panic in any
161 // single job then aborts only this tick (the next tick re-runs the
162 // full set); without supervision the panic unwinds the scheduler
163 // loop and silently halts ALL background maintenance, including the
164 // overrun/panic alerting below (Run #2 Performance CRITICAL). The
165 // advisory lock stays held on `lock_conn` in this loop task for the
166 // whole tick, so no other instance runs concurrently while the
167 // child task works.
168 let tick_jobs = {
169 let state = state.clone();
170 tokio::spawn(async move {
171 // Publish scheduled items. The announcement fan-out (4 DB
172 // roundtrips per item) gets spawned off the lock-held tick so a
173 // burst of releases doesn't extend the advisory-lock hold time.
174 match db::items::publish_scheduled_items(&state.db).await {
175 Ok(items) => {
176 for item in &items {
177 tracing::info!(
178 item_id = %item.id,
179 title = %item.title,
180 "scheduler published item"
181 );
182 let db_pool = state.db.clone();
183 let mailer = state.email.clone();
184 let cfg = state.config.clone();
185 let item_for_announce = item.clone();
186 state.bg.spawn("release-announcements", async move {
187 announcements::send_release_announcements(
188 &db_pool,
189 &mailer,
190 &cfg,
191 &item_for_announce,
192 )
193 .await;
194 });
195 if item.mt_thread_id.is_none() {
196 mt_threads::spawn_mt_thread_for_item_by_lookup(
197 &state.db,
198 &state.bg,
199 &Integrations::from_ref(&state),
200 &state.config,
201 item,
202 );
203 }
204 }
205 }
206 Err(e) => {
207 tracing::error!(error = ?e, "scheduler failed to publish items");
208 }
209 }
210
211 // Send onboarding drip emails
212 announcements::send_onboarding_emails(&state.db, &state.email, &state.config)
213 .await;
214
215 // Dispatch pending builds. Project the runner's slice from the
216 // scheduler's `AppState` at the call site (same seam as the stripe
217 // dispatcher below) so the runner declares only what it needs.
218 crate::build_runner::dispatch_pending_build(&axum::extract::FromRef::from_ref(
219 &state,
220 ))
221 .await;
222
223 // Publish scheduled blog posts, same off-lock pattern as items.
224 match db::blog_posts::publish_scheduled_blog_posts(&state.db).await {
225 Ok(posts) => {
226 for post in &posts {
227 tracing::info!(
228 post_id = %post.id,
229 title = %post.title,
230 "scheduler published blog post"
231 );
232 let db_pool = state.db.clone();
233 let mailer = state.email.clone();
234 let cfg = state.config.clone();
235 let post_for_announce = post.clone();
236 state.bg.spawn("blog-post-announcements", async move {
237 announcements::send_blog_post_announcements(
238 &db_pool,
239 &mailer,
240 &cfg,
241 &post_for_announce,
242 )
243 .await;
244 });
245 if post.mt_thread_id.is_none() {
246 mt_threads::spawn_mt_thread_for_blog_post_by_lookup(
247 &state.db,
248 &state.bg,
249 &Integrations::from_ref(&state),
250 &state.config,
251 post,
252 );
253 }
254 }
255 }
256 Err(e) => {
257 tracing::error!(error = ?e, "scheduler failed to publish blog posts");
258 }
259 }
260
261 // Clean up expired idempotency keys (every tick is fine, cheap DELETE)
262 if let Err(e) = db::idempotency::cleanup_expired(&state.db).await {
263 tracing::error!(error = ?e, "failed to clean up expired idempotency keys");
264 }
265
266 let (run_sandbox, run_hourly, run_daily, run_weekly) =
267 jobs_for_tick(tick_count);
268
269 // Clean up expired sandbox accounts (every 5 ticks = 5 min at 60s interval)
270 if run_sandbox {
271 cleanup::cleanup_sandbox_accounts(&state).await;
272 cleanup::retry_pending_s3_deletions(&state).await;
273 let report = crate::scanning::spool::reap_orphans(std::path::Path::new(
274 constants::SCAN_SPOOL_DIR,
275 ));
276 if report.deleted > 0 || report.errors > 0 {
277 tracing::info!(
278 deleted = report.deleted,
279 errors = report.errors,
280 "scan spool reaper swept orphans"
281 );
282 }
283 }
284
285 // Payments + storage outbound I/O: webhook retry, refund/credit
286 // escalation, platform-credit settle/reverse, stale-transaction and
287 // orphaned-upload cleanup. This block issues up to ~310 serial
288 // Stripe/S3 round-trips (30-90s worst case). Run it on the background
289 // pool rather than inline so the scheduler advisory lock is released as
290 // soon as the DB-only jobs finish instead of being held for the full
291 // outbound-I/O duration (fuzz-2026-07-06 F2, the hold tripped the 50s
292 // WAM tick-overrun alert and could eat the 60s interval). Every
293 // operation here is already safe to run without the single-instance
294 // tick lock, per-event `pg_try_advisory_xact_lock` dedup, `FOR UPDATE
295 // SKIP LOCKED` claims, and deterministic Stripe idempotency keys, so a
296 // run overlapping the next tick's just skips locked rows. Internal
297 // ordering (settle before escalate) is preserved by the sequence here.
298 {
299 let job_state = state.clone();
300 state.bg.spawn("scheduler-payments-io", async move {
301 webhooks::retry_failed_webhooks(&job_state).await;
302 webhooks::escalate_stale_refunds(&job_state).await;
303 webhooks::settle_platform_credits(&job_state).await;
304 webhooks::escalate_stale_platform_credits(&job_state).await;
305 webhooks::reverse_refunded_platform_credits(&job_state).await;
306 cleanup::cleanup_stale_pending_transactions(&job_state).await;
307 cleanup::cleanup_orphaned_uploads(&job_state).await;
308 });
309 }
310
311 // Hourly: scan SyncKit apps for 75/90/100% cap breaches and email
312 // the app owner. Cheap query, single JOIN on a small table.
313 if run_hourly {
314 synckit_warnings::check_and_send_warnings(&state).await;
315 cleanup::purge_old_scan_jobs(&state).await;
316 // Fail imports stranded in `processing` by a crashed process.
317 // Heartbeats bump per chunk, so a 30-min-stale beat is dead.
318 match db::imports::reap_stuck_import_jobs(&state.db, STUCK_IMPORT_SECS)
319 .await
320 {
321 Ok(n) if n > 0 => tracing::warn!(
322 reaped = n,
323 "failed stuck import jobs (stale heartbeat)"
324 ),
325 Ok(_) => {}
326 Err(e) => {
327 tracing::error!(error = ?e, "failed to reap stuck import jobs");
328 }
329 }
330 }
331
332 // Weekly storage drift correction + integrity checks
333 if run_weekly {
334 integrity::recalculate_all_storage_used(&state).await;
335 integrity::check_sales_count_drift(&state).await;
336 match db::synckit_billing::recalculate_synckit_app_storage(&state.db).await
337 {
338 Ok(n) => {
339 if n > 0 {
340 tracing::info!(
341 corrected = n,
342 "synckit app storage drift corrected"
343 );
344 }
345 }
346 Err(e) => {
347 tracing::error!(error = ?e, "synckit storage drift correction failed");
348 }
349 }
350 }
351
352 // Daily checks (every 1440 ticks at 60s interval, plus first tick after startup)
353 if run_daily {
354 integrity::check_stale_subscriptions(&state).await;
355 integrity::check_email_bounce_spike(&state).await;
356
357 // Enforce post-grace item hiding (canceled 30+ days ago). Daily
358 // is ample for a 30-day grace window, and it's self-draining
359 // (mark_grace_enforced_batch), so running it every 60s tick only
360 // re-issued an empty query, moved here off the hot tick path.
361 integrity::enforce_post_grace_hiding(&state).await;
362
363 // Prune session records inactive for 90+ days
364 let session_threshold = chrono::Utc::now() - chrono::Duration::days(90);
365 match db::sessions::prune_expired_sessions(&state.db, session_threshold)
366 .await
367 {
368 Ok(n) => {
369 if n > 0 {
370 tracing::info!(pruned = n, "pruned expired session records");
371 }
372 let _ = db::scheduler_jobs::record_job_run(
373 &state.db,
374 "session_prune",
375 n as i64,
376 )
377 .await;
378 }
379 Err(e) => {
380 tracing::error!(error = ?e, "failed to prune expired sessions");
381 }
382 }
383
384 // Drop dead password-reset tokens (consumed/expired > 7 days).
385 match db::auth::prune_password_reset_tokens(&state.db).await {
386 Ok(n) => {
387 if n > 0 {
388 tracing::info!(pruned = n, "pruned old password-reset tokens");
389 }
390 }
391 Err(e) => {
392 tracing::error!(error = ?e, "failed to prune password-reset tokens");
393 }
394 }
395
396 // Scrub IP addresses older than 30 days (privacy policy commitment)
397 cleanup::scrub_stale_ip_addresses(&state).await;
398
399 // Delete abandoned custom-page drafts older than 30 days.
400 match db::custom_pages::delete_drafts_older_than(&state.db, 30).await {
401 Ok(n) => {
402 if n > 0 {
403 tracing::info!(deleted = n, "pruned old custom-page drafts");
404 }
405 }
406 Err(e) => {
407 tracing::error!(error = ?e, "failed to prune custom-page drafts");
408 }
409 }
410
411 // Delete terminated accounts whose 30-day export window has expired
412 cleanup::delete_expired_terminated_accounts(&state).await;
413
414 // Delete self-deleted creator accounts whose 90-day content grace period has expired
415 cleanup::delete_expired_content_removal_accounts(&state).await;
416
417 // Permanently delete soft-deleted items older than 7 days
418 cleanup::purge_expired_deleted_items(&state).await;
419
420 // Clean up stale and unavailable cart items
421 cleanup::cleanup_cart_items(&state).await;
422
423 // Prune page view aggregates older than 2 years
424 match db::page_views::prune_old_views(&state.db, 730).await {
425 Ok(n) => {
426 if n > 0 {
427 tracing::info!(pruned = n, "pruned old page view records");
428 }
429 }
430 Err(e) => tracing::error!(error = ?e, "failed to prune page views"),
431 }
432
433 // Prune processed-webhook dedup markers older than 30 days. Stripe
434 // won't redeliver events that old, so they no longer prevent a
435 // duplicate; the table is otherwise append-only and grows one row
436 // per webhook forever (Run #21 Performance SERIOUS).
437 match db::webhook_events::prune_processed_events(&state.db, 30).await {
438 Ok(n) => {
439 if n > 0 {
440 tracing::info!(
441 pruned = n,
442 "pruned old processed-webhook markers"
443 );
444 }
445 }
446 Err(e) => {
447 tracing::error!(error = ?e, "failed to prune processed-webhook markers");
448 }
449 }
450
451 // Health/sync/OAuth prune+compaction. These ran in the monitor
452 // loop under a SECOND advisory lock (a parallel maintenance
453 // scheduler, Perf-S3 mandatory surprise); consolidated here so
454 // all periodic maintenance lives under the one scheduler lock and
455 // there is a single place to add a daily job.
456 match db::monitor::prune_health_history(
457 &state.db,
458 constants::HEALTH_HISTORY_RETAIN_DAYS,
459 )
460 .await
461 {
462 Ok(n) => {
463 if n > 0 {
464 tracing::info!(
465 deleted = n,
466 "pruned old health history records"
467 );
468 }
469 }
470 Err(e) => tracing::warn!(error = ?e, "failed to prune health history"),
471 }
472 match db::synckit::prune_sync_log(
473 &state.db,
474 constants::SYNC_LOG_RETAIN_DAYS,
475 )
476 .await
477 {
478 Ok(n) => {
479 if n > 0 {
480 tracing::info!(deleted = n, "pruned old sync log records");
481 }
482 }
483 Err(e) => tracing::warn!(error = ?e, "failed to prune sync log"),
484 }
485 match db::synckit::compact_all_sync_logs(
486 &state.db,
487 constants::SYNC_LOG_COMPACT_MIN_AGE_DAYS,
488 )
489 .await
490 {
491 Ok(n) => {
492 if n > 0 {
493 tracing::info!(
494 deleted = n,
495 "compacted sync log (cursor-based)"
496 );
497 }
498 }
499 Err(e) => tracing::warn!(error = ?e, "failed to compact sync log"),
500 }
501 match db::oauth::cleanup_expired_oauth_codes(&state.db).await {
502 Ok(n) => {
503 if n > 0 {
504 tracing::info!(deleted = n, "cleaned up expired OAuth codes");
505 }
506 }
507 Err(e) => tracing::warn!(error = ?e, "failed to clean up OAuth codes"),
508 }
509 match db::oauth::cleanup_expired_refresh_tokens(&state.db).await {
510 Ok(n) => {
511 if n > 0 {
512 tracing::info!(
513 deleted = n,
514 "cleaned up expired OAuth refresh tokens"
515 );
516 }
517 }
518 Err(e) => {
519 tracing::warn!(error = ?e, "failed to clean up OAuth refresh tokens");
520 }
521 }
522 }
523 })
524 };
525
526 // Supervise the job task with BOTH a panic guard and a hang
527 // watchdog. A panic surfaces as a JoinError; a hang trips the
528 // watchdog. Either way we log, (rate-limited) alert, and fall
529 // through to release the advisory lock so the freeze can't outlive
530 // this tick. The watchdog is the load-bearing half: without it a
531 // wedged query holds the lock and halts maintenance on every
532 // instance with no signal (Run 21 Performance).
533 let tick_abort = tick_jobs.abort_handle();
534 match tokio::time::timeout(
535 std::time::Duration::from_secs(TICK_WATCHDOG_SECS),
536 tick_jobs,
537 )
538 .await
539 {
540 Ok(Ok(())) => {}
541 Ok(Err(join_err)) => {
542 tracing::error!(
543 tick = tick_count, error = ?join_err,
544 "scheduler job task panicked; tick aborted, loop continues"
545 );
546 if let Some(ref wam) = state.wam {
547 let cooldown_ok = last_panic_alert.is_none_or(|t| {
548 t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS
549 });
550 if cooldown_ok {
551 let body = format!(
552 "tick #{tick_count}: a scheduled job panicked ({join_err}). \
553 The tick was aborted; the scheduler loop survived and will \
554 re-run all jobs on the next tick."
555 );
556 wam.create_ticket(
557 "Scheduler job panicked",
558 Some(&body),
559 "high",
560 "scheduler-job-panic",
561 None,
562 )
563 .await;
564 last_panic_alert = Some(std::time::Instant::now());
565 }
566 }
567 }
568 Err(_elapsed) => {
569 // The tick exceeded the watchdog, abort it so it stops
570 // holding DB connections, then release the lock below.
571 tick_abort.abort();
572 tracing::error!(
573 tick = tick_count,
574 watchdog_secs = TICK_WATCHDOG_SECS,
575 "scheduler tick exceeded watchdog; aborted and releasing advisory lock so maintenance can recover"
576 );
577 if let Some(ref wam) = state.wam {
578 let cooldown_ok = last_panic_alert.is_none_or(|t| {
579 t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS
580 });
581 if cooldown_ok {
582 let body = format!(
583 "tick #{tick_count}: a scheduled job hung past {TICK_WATCHDOG_SECS}s \
584 (likely a query blocked on a lock). The tick was aborted and the \
585 advisory lock released so the next tick recovers."
586 );
587 wam.create_ticket(
588 "Scheduler tick hung (watchdog)",
589 Some(&body),
590 "high",
591 "scheduler-tick-hang",
592 None,
593 )
594 .await;
595 last_panic_alert = Some(std::time::Instant::now());
596 }
597 }
598 }
599 }
600
601 // Explicitly release the advisory lock (defense-in-depth: also released
602 // when lock_conn is dropped, but explicit unlock survives refactors that
603 // might move lock_conn into a shorter-lived scope).
604 let _ = sqlx::query("SELECT pg_advisory_unlock($1)")
605 .bind(SCHEDULER_ADVISORY_LOCK_ID)
606 .execute(&mut *lock_conn)
607 .await;
608
609 // Tick-duration accounting. WARN at TICK_DURATION_WARN_SECS so the
610 // log surfaces it; raise a WAM ticket past TICK_DURATION_ALERT_SECS
611 // with a 1-hour cooldown so a chronic overrun doesn't flood tickets.
612 let tick_duration = tick_started.elapsed();
613 let tick_secs = tick_duration.as_secs();
614 if tick_secs >= TICK_DURATION_WARN_SECS {
615 tracing::warn!(
616 tick = tick_count,
617 duration_secs = tick_secs,
618 "scheduler tick exceeded soft duration ceiling"
619 );
620 }
621 if tick_secs >= TICK_DURATION_ALERT_SECS
622 && let Some(ref wam) = state.wam
623 {
624 let cooldown_ok = last_overrun_alert
625 .is_none_or(|t| t.elapsed().as_secs() >= constants::ALERT_COOLDOWN_SECS);
626 if cooldown_ok {
627 let title = format!("Scheduler tick overran: {tick_secs}s");
628 let body = format!(
629 "tick #{tick_count} took {tick_secs}s (interval is {}s).",
630 constants::SCHEDULER_INTERVAL_SECS
631 );
632 wam.create_ticket(&title, Some(&body), "high", "scheduler-tick-overrun", None)
633 .await;
634 last_overrun_alert = Some(std::time::Instant::now());
635 }
636 }
637 }
638 })
639 }
640
641 #[cfg(test)]
642 mod tests {
643 use super::*;
644
645 // ── Tick cadence tests ──
646
647 #[test]
648 fn tick_1_runs_daily_not_weekly() {
649 let (sandbox, _hourly, daily, weekly) = jobs_for_tick(1);
650 assert!(!sandbox, "tick 1 is not a multiple of 5");
651 assert!(daily, "tick 1 should trigger daily jobs (first-tick rule)");
652 assert!(!weekly, "tick 1 should not trigger weekly jobs");
653 }
654
655 #[test]
656 fn tick_5_runs_sandbox_cleanup() {
657 let (sandbox, _hourly, daily, weekly) = jobs_for_tick(5);
658 assert!(sandbox);
659 assert!(!daily);
660 assert!(!weekly);
661 }
662
663 #[test]
664 fn tick_1440_runs_daily_and_sandbox() {
665 let (sandbox, _hourly, daily, weekly) = jobs_for_tick(1440);
666 assert!(sandbox, "1440 is divisible by 5");
667 assert!(daily, "1440 is the daily interval");
668 assert!(!weekly);
669 }
670
671 #[test]
672 fn tick_10080_runs_all_three() {
673 let (sandbox, _hourly, daily, weekly) = jobs_for_tick(10_080);
674 assert!(sandbox, "10080 is divisible by 5");
675 assert!(daily, "10080 is divisible by 1440");
676 assert!(weekly, "10080 is the weekly interval");
677 }
678
679 #[test]
680 fn normal_tick_runs_nothing_special() {
681 let (sandbox, _hourly, daily, weekly) = jobs_for_tick(7);
682 assert!(!sandbox);
683 assert!(!daily);
684 assert!(!weekly);
685 }
686
687 #[test]
688 fn second_daily_tick() {
689 let (_, _, daily, _) = jobs_for_tick(2880);
690 assert!(daily, "2880 = 2 * 1440");
691 }
692
693 #[test]
694 fn second_weekly_tick() {
695 let (_, _, _, weekly) = jobs_for_tick(20_160);
696 assert!(weekly, "20160 = 2 * 10080");
697 }
698
699 // ── Interval constant sanity checks ──
700
701 #[test]
702 fn drift_correction_interval_is_7_days() {
703 assert_eq!(DRIFT_CORRECTION_INTERVAL, 7 * 24 * 60);
704 }
705
706 #[test]
707 fn daily_interval_is_24_hours() {
708 assert_eq!(DAILY_INTERVAL, 24 * 60);
709 }
710
711 #[test]
712 fn sandbox_cleanup_interval_is_5_minutes() {
713 assert_eq!(SANDBOX_CLEANUP_INTERVAL, 5);
714 }
715
716 // ── Adversarial tests (test-fuzz) ──
717
718 #[test]
719 fn tick_0_runs_nothing() {
720 // Tick 0 should not run anything: 0 % N == 0 for all N,
721 // but tick 0 never happens in practice (counter starts at 0, increments before use).
722 // Test what would happen if it did.
723 let (sandbox, _hourly, daily, weekly) = jobs_for_tick(0);
724 // 0.is_multiple_of(N) is true for all N, so these all fire
725 assert!(sandbox, "0 is a multiple of 5");
726 assert!(daily, "0 is a multiple of 1440");
727 assert!(weekly, "0 is a multiple of 10080");
728 }
729
730 #[test]
731 fn large_tick_values() {
732 // 52 weeks of ticks, exact multiple of weekly interval
733 let fifty_two_weeks = 52 * DRIFT_CORRECTION_INTERVAL;
734 let (sandbox, _hourly, daily, weekly) = jobs_for_tick(fifty_two_weeks);
735 assert!(sandbox);
736 assert!(daily, "{fifty_two_weeks} should be divisible by 1440");
737 assert!(weekly, "{fifty_two_weeks} should be divisible by 10080");
738
739 // Large non-aligned tick
740 let (_, _, daily, weekly) = jobs_for_tick(999_999);
741 assert!(!daily, "999999 is not divisible by 1440");
742 assert!(!weekly, "999999 is not divisible by 10080");
743 }
744
745 #[test]
746 fn daily_not_on_partial_day() {
747 // Tick 720 = half a day, should not trigger daily
748 let (_, _, daily, _) = jobs_for_tick(720);
749 assert!(!daily, "720 ticks is only half a day");
750 }
751
752 #[test]
753 fn weekly_not_on_partial_week() {
754 // 5040 = 3.5 days, should not trigger weekly
755 let (_, _, _, weekly) = jobs_for_tick(5040);
756 assert!(!weekly, "5040 is only 3.5 days");
757 }
758
759 #[test]
760 fn sandbox_every_5_ticks_consecutively() {
761 for tick in 1..=25 {
762 let (sandbox, _, _, _) = jobs_for_tick(tick);
763 if tick % 5 == 0 {
764 assert!(sandbox, "tick {tick} should run sandbox cleanup");
765 } else {
766 assert!(!sandbox, "tick {tick} should NOT run sandbox cleanup");
767 }
768 }
769 }
770
771 #[test]
772 fn intervals_are_coprime_aware() {
773 // Verify weekly is an exact multiple of daily
774 assert_eq!(
775 DRIFT_CORRECTION_INTERVAL % DAILY_INTERVAL,
776 0,
777 "weekly interval must be exact multiple of daily"
778 );
779 // Verify sandbox fits evenly into daily
780 assert_eq!(
781 DAILY_INTERVAL % SANDBOX_CLEANUP_INTERVAL,
782 0,
783 "sandbox interval must divide evenly into daily"
784 );
785 }
786 }
787