// Package cron is the Go replacement for notifications/scheduler.py's // APScheduler pair (warm-cities + IndexNow). The Python side chose APScheduler // over a Postgres-native queue because exactly one process ever runs these // jobs — no fan-out, backpressure, or dead-letter need — and that shape // carries over unchanged: this daemon is the sole scheduler, so plain // per-job tickers are all the machinery required. procrastinate/pgqueuer // remain the documented upgrade path if that ever changes — see // docs/architecture/repo-topology-and-infrastructure.md §7. // // The jobs themselves stay in Python (called via the internal HTTP routes); // this package owns only the timing. package cron import ( "context" "log/slog" "sync" "time" ) // Job is one recurring task: a name for logs, an interval, and the callback // (in practice an apiclient method hitting a /internal/jobs/* route). type Job struct { Name string Interval time.Duration Run func(ctx context.Context) error } // Run drives every job on its own ticker until ctx is cancelled, then returns // once all job goroutines have stopped. It never returns early: a failing job // logs and keeps ticking — one bad tick (backend briefly down, timeout) must // never kill the schedule, because nothing restarts it short of a container // restart. func Run(ctx context.Context, logger *slog.Logger, jobs ...Job) { var wg sync.WaitGroup for _, job := range jobs { wg.Add(1) go func(job Job) { defer wg.Done() runJob(ctx, logger, job) }(job) } wg.Wait() } func runJob(ctx context.Context, logger *slog.Logger, job Job) { // A plain ticker's first fire is one interval from now, NOT at startup — // deliberately kept from the Python scheduler (its add_job had no // next_run_time override for the same reason): warm-cities already runs at // deploy time via the deploy hook, so an immediate run on every daemon // boot/restart would just re-spend work (and, for warm-cities, archive-fetch // quota) that the deploy already paid for. ticker := time.NewTicker(job.Interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: start := time.Now() // Run synchronously in this loop: a job can never overlap with // itself by construction. That matters most for warm-cities, // which spends the archive-fetch quota — two overlapping runs would // double-spend it. if err := job.Run(ctx); err != nil { if ctx.Err() != nil { // Shutdown raced the job; the cancellation is the story, // not the (expected) aborted call. return } logger.Error("cron job failed; will retry next tick", "job", job.Name, "err", err, "elapsed", time.Since(start)) } else { logger.Info("cron job ok", "job", job.Name, "elapsed", time.Since(start)) } // If the job overran its interval, a tick may have fired mid-run. // Skip it rather than running again back-to-back — a late tick is // the same double-spend as an overlapping one, just serialized. // (Go 1.23+ tickers already drop fires with no ready receiver; // this drain pins the skip semantics rather than relying on the // runtime's channel buffering.) select { case <-ticker.C: logger.Warn("cron job overran its interval; skipping the tick that fired mid-run", "job", job.Name, "interval", job.Interval, "elapsed", time.Since(start)) default: } } } }