Skip to main content

max / makenotwork

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