// Package maintenance runs the janitorial work that keeps persisted // data from accumulating without bound. // // It exists because cleanup used to have no owner. Functions that // deleted expired rows were written and then never called; files written // by one package had no counterpart anywhere that removed them. A // registry makes the set of janitorial jobs a single visible list, so a // new cache that forgets to register is obvious in review rather than // discovered years later as unbounded growth. // // Policies come from the classification in backend/datamap: // // - Derived data is swept against a live set computed from the data it // was derived from. Anything not in the live set is garbage. // - Cache data is evicted by age, because it has no owner to be // compared against and is merely expensive — not impossible — to // re-fetch. // // Sweeps are idempotent and safe to interrupt: each deletes only what it // has positively identified as unreferenced, so a partial run simply // leaves work for the next one. package maintenance import ( "context" "log/slog" "sync" "time" ) // Result reports what a single job reclaimed. type Result struct { // RowsDeleted counts database rows removed. RowsDeleted int64 // FilesDeleted counts files removed from disk. FilesDeleted int64 // BytesFreed is the size of those files. BytesFreed int64 } // empty reports whether the job found nothing to do, so quiet runs can // be logged at a lower level. func (r Result) empty() bool { return r.RowsDeleted == 0 && r.FilesDeleted == 0 } // Job is one unit of janitorial work. type Job struct { // Name identifies the job in logs and in the run record. Name string // MinInterval is the minimum time between runs. A job is skipped if // it ran more recently than this, so hooking the runner to a // frequently-firing trigger stays cheap. MinInterval time.Duration // Run performs the work. It must be idempotent and must respect // context cancellation. Run func(ctx context.Context) (Result, error) } // Runner holds the registered jobs and enforces their intervals. type Runner struct { mu sync.Mutex jobs []Job lastRun map[string]time.Time logger *slog.Logger } // NewRunner returns an empty runner. func NewRunner(logger *slog.Logger) *Runner { return &Runner{ lastRun: make(map[string]time.Time), logger: logger, } } // Register adds a job. Registering a name twice replaces the earlier // job, so wiring code can be re-run without accumulating duplicates. func (r *Runner) Register(job Job) { r.mu.Lock() defer r.mu.Unlock() for i, existing := range r.jobs { if existing.Name == job.Name { r.jobs[i] = job return } } r.jobs = append(r.jobs, job) } // JobNames returns the registered job names, for tests and diagnostics. func (r *Runner) JobNames() []string { r.mu.Lock() defer r.mu.Unlock() names := make([]string, 0, len(r.jobs)) for _, j := range r.jobs { names = append(names, j.Name) } return names } // due reports whether a job's interval has elapsed. func (r *Runner) due(job Job, now time.Time) bool { r.mu.Lock() defer r.mu.Unlock() last, ran := r.lastRun[job.Name] if !ran { return true } return now.Sub(last) >= job.MinInterval } func (r *Runner) markRun(name string, at time.Time) { r.mu.Lock() defer r.mu.Unlock() r.lastRun[name] = at } // snapshot copies the job list so a run does not hold the lock while // executing jobs. func (r *Runner) snapshot() []Job { r.mu.Lock() defer r.mu.Unlock() out := make([]Job, len(r.jobs)) copy(out, r.jobs) return out } // RunDue runs every job whose interval has elapsed. A job that fails is // logged and does not prevent the others from running; janitorial work // is best-effort by nature and the next run will retry. func (r *Runner) RunDue(ctx context.Context) Result { var total Result for _, job := range r.snapshot() { if ctx.Err() != nil { r.logger.Info("maintenance cancelled", "after", job.Name) break } now := time.Now() if !r.due(job, now) { continue } start := time.Now() result, err := job.Run(ctx) r.markRun(job.Name, now) if err != nil { r.logger.Warn("maintenance job failed", "job", job.Name, "err", err, "duration", time.Since(start), ) continue } total.RowsDeleted += result.RowsDeleted total.FilesDeleted += result.FilesDeleted total.BytesFreed += result.BytesFreed if result.empty() { r.logger.Debug("maintenance job found nothing", "job", job.Name, "duration", time.Since(start), ) continue } r.logger.Info("maintenance job reclaimed", "job", job.Name, "rows", result.RowsDeleted, "files", result.FilesDeleted, "bytes", result.BytesFreed, "duration", time.Since(start), ) } return total } // Start runs the due jobs immediately and then on every tick until the // context is cancelled. It returns straight away; the loop runs in its // own goroutine. func (r *Runner) Start(ctx context.Context, tick time.Duration) { go func() { r.RunDue(ctx) ticker := time.NewTicker(tick) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: r.RunDue(ctx) } } }() }