Skip to main content

max / makenotwork

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