From 1153b33af7f4d06dac64b409c94234acd3176846 Mon Sep 17 00:00:00 2001 From: Kevin Brown Date: Wed, 9 Sep 2026 23:28:03 -0600 Subject: [PATCH] Sweep job workspaces after 30 days MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit data/jobs grew without bound: nothing ever deleted a job's workspace once created, regardless of outcome. Add a daily age-based sweep mirroring the existing Drive-file retention sweep in internal/mailgun/retention.go — same shape, same crash-safety properties, applied to local job directories instead of published Sheets. A completed job (jobs.completed_at IS NOT NULL) past 30 days has its data/jobs/ workspace removed. The row itself is kept, with a new files_deleted_at marker set instead of deleting it, matching how email_deliveries.deleted_at already works: `jobs get ` keeps answering after the files are gone, and a swept row is never offered to the sweep again. Only completed_at gates the query, so a queued or running job can never be swept no matter how old. Verified the migration against a copy of the real production database (135 job rows): row count preserved, every existing row correctly gets files_deleted_at=NULL, a second Open() is a no-op, and ExpiredJobs correctly identifies the 51 jobs that would be swept on first deploy. Co-Authored-By: Claude Sonnet 5 --- cmd/filemill/main.go | 4 ++ internal/app/retention.go | 85 +++++++++++++++++++++++ internal/app/retention_test.go | 121 +++++++++++++++++++++++++++++++++ internal/store/store.go | 75 +++++++++++++++++++- internal/store/store_test.go | 87 ++++++++++++++++++++++++ 5 files changed, 371 insertions(+), 1 deletion(-) create mode 100644 internal/app/retention.go create mode 100644 internal/app/retention_test.go diff --git a/cmd/filemill/main.go b/cmd/filemill/main.go index 9ce6984..02894f6 100644 --- a/cmd/filemill/main.go +++ b/cmd/filemill/main.go @@ -117,6 +117,10 @@ func main() { } else { mailLog.Print("integration disabled: no Mailgun environment variables set") } + // Job workspaces accumulate whether or not email is configured, so + // this sweep runs unconditionally in continuous mode rather than + // nested under the mailgun branch above. + go application.SweepExpiredJobs(ctx) } runErr := application.Run(ctx, once) if server != nil { diff --git a/internal/app/retention.go b/internal/app/retention.go new file mode 100644 index 0000000..921f92f --- /dev/null +++ b/internal/app/retention.go @@ -0,0 +1,85 @@ +package app + +import ( + "context" + "os" + "path/filepath" + "time" +) + +// jobRetentionPeriod is how long a completed job's workspace stays on disk +// before the sweep deletes it. By then the sender already has the result — +// attached, or (for sheets-link) uploaded to Drive — so the local copy is a +// backup nobody is waiting on. +// +// A var, not a const: mailgun's retention test ages records through a fake +// Engine, but App talks to a concrete *store.Store with no such seam, so its +// own test shrinks this instead of waiting 30 real days. +var jobRetentionPeriod = 30 * 24 * time.Hour + +const ( + // jobSweepInterval mirrors mailgun's retention sweep: retention is measured + // in days, so checking daily is ample. + jobSweepInterval = 24 * time.Hour + // jobFirstSweepDelay staggers the startup sweep just past the noisy first + // moments of a restart, while still guaranteeing one runs. + jobFirstSweepDelay = time.Minute +) + +// SweepExpiredJobs runs the job-retention sweep until ctx is cancelled. +// +// This is the same pattern as mailgun.Service.SweepExpired, applied to local +// job workspaces instead of published Drive files: the first sweep runs +// shortly after startup rather than a full interval later, because the +// worker restarts on every config reload and crash, and a plain 24-hour +// ticker on a machine that restarts daily would never sweep anything at all. +func (a *App) SweepExpiredJobs(ctx context.Context) { + timer := time.NewTimer(jobFirstSweepDelay) + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return + case <-timer.C: + } + if err := a.sweepExpiredJobs(); err != nil { + a.log.Printf("job sweep: %v", err) + } + timer.Reset(jobSweepInterval) + } +} + +// sweepExpiredJobs deletes the on-disk workspace of every completed job past +// the retention horizon. Only completed_at is checked (see +// store.ExpiredJobs), so a job still queued or running is never touched. +// +// A workspace that fails to delete is logged and left unmarked, so the next +// sweep retries it; one stuck directory must not strand the rest of the +// batch. RemoveAll is idempotent — a workspace already gone counts as +// deleted — so marking a job after a crash-interrupted sweep is never a +// problem. +// +// A sweep that deleted something says so; an idle one stays silent, so the +// line means something when it does appear — the same reasoning as +// mailgun.Service.sweepExpired. +func (a *App) sweepExpiredJobs() error { + ids, err := a.store.ExpiredJobs(time.Now().UTC().Add(-jobRetentionPeriod)) + if err != nil { + return err + } + deleted := 0 + for _, id := range ids { + if err := os.RemoveAll(filepath.Join(a.data, "jobs", id)); err != nil { + a.log.Printf("job sweep: delete %s: %v", id, err) + continue + } + deleted++ + if err := a.store.MarkJobFilesDeleted(id); err != nil { + a.log.Printf("job sweep: record deletion of %s: %v", id, err) + } + } + if deleted > 0 { + a.log.Printf("job sweep: deleted %d job workspace(s) older than %d days", deleted, int(jobRetentionPeriod.Hours()/24)) + } + return nil +} diff --git a/internal/app/retention_test.go b/internal/app/retention_test.go new file mode 100644 index 0000000..666e6cd --- /dev/null +++ b/internal/app/retention_test.go @@ -0,0 +1,121 @@ +package app + +import ( + "os" + "path/filepath" + "testing" + "time" + + "filemill/internal/store" +) + +// A completed job past the retention horizon has its workspace deleted; one +// well inside the horizon is left alone. jobRetentionPeriod is temporarily +// shrunk so the test does not wait 30 real days for a job it just completed. +func TestSweepExpiredJobsDeletesOldWorkspaceOnly(t *testing.T) { + restore := jobRetentionPeriod + defer func() { jobRetentionPeriod = restore }() + + root := t.TempDir() + writeTransformers(t, root, `transformers: + - operation: copy_rename + command: ["python.exe", "x.py"] + extensions: [txt] +`) + a, err := Open(root) + if err != nil { + t.Fatal(err) + } + defer a.Close() + + src := filepath.Join(root, "input.txt") + if err := os.WriteFile(src, []byte("hi"), 0644); err != nil { + t.Fatal(err) + } + + oldID, err := a.Submit("copy_rename", src) + if err != nil { + t.Fatal(err) + } + if err := a.store.Complete(oldID, store.StatusSucceeded, "ok"); err != nil { + t.Fatal(err) + } + time.Sleep(500 * time.Millisecond) + + // Shrink the horizon only now, so oldID (completed 500ms ago) falls well + // outside it while freshID (completed a moment from now) stays well inside. + jobRetentionPeriod = 100 * time.Millisecond + + freshID, err := a.Submit("copy_rename", src) + if err != nil { + t.Fatal(err) + } + if err := a.store.Complete(freshID, store.StatusSucceeded, "ok"); err != nil { + t.Fatal(err) + } + + if err := a.sweepExpiredJobs(); err != nil { + t.Fatalf("sweepExpiredJobs: %v", err) + } + + if _, err := os.Stat(filepath.Join(root, "data", "jobs", oldID)); !os.IsNotExist(err) { + t.Errorf("old job workspace still exists (stat err = %v)", err) + } + if _, err := os.Stat(filepath.Join(root, "data", "jobs", freshID)); err != nil { + t.Errorf("fresh job workspace was removed: %v", err) + } + + // The row survives; only the files are gone. A second sweep must not + // touch it again (nothing new expired) or error on the missing directory. + job, err := a.Job(oldID) + if err != nil { + t.Fatalf("Job(oldID) after sweep: %v", err) + } + if job.Status != store.StatusSucceeded { + t.Errorf("swept job status = %q, want %q", job.Status, store.StatusSucceeded) + } + if err := a.sweepExpiredJobs(); err != nil { + t.Fatalf("second sweepExpiredJobs: %v", err) + } +} + +// A job whose workspace never existed to begin with — say a previous sweep +// crashed after RemoveAll but before marking the row — must not make the +// sweep error out. RemoveAll on an absent path is success, not failure. +func TestSweepExpiredJobsToleratesAlreadyMissingWorkspace(t *testing.T) { + restore := jobRetentionPeriod + jobRetentionPeriod = time.Millisecond + defer func() { jobRetentionPeriod = restore }() + + root := t.TempDir() + writeTransformers(t, root, `transformers: + - operation: copy_rename + command: ["python.exe", "x.py"] + extensions: [txt] +`) + a, err := Open(root) + if err != nil { + t.Fatal(err) + } + defer a.Close() + + src := filepath.Join(root, "input.txt") + if err := os.WriteFile(src, []byte("hi"), 0644); err != nil { + t.Fatal(err) + } + id, err := a.Submit("copy_rename", src) + if err != nil { + t.Fatal(err) + } + if err := a.store.Complete(id, store.StatusSucceeded, "ok"); err != nil { + t.Fatal(err) + } + if err := os.RemoveAll(filepath.Join(root, "data", "jobs", id)); err != nil { + t.Fatal(err) + } + time.Sleep(5 * time.Millisecond) + + if err := a.sweepExpiredJobs(); err != nil { + t.Fatalf("sweepExpiredJobs must tolerate an already-missing workspace: %v", err) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index c200855..cefaee2 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -71,12 +71,21 @@ func Open(path string) (*Store, error) { s := &Store{db: db} _, err = db.Exec(`CREATE TABLE IF NOT EXISTS jobs ( id TEXT PRIMARY KEY, operation TEXT NOT NULL, status TEXT NOT NULL, input_name TEXT NOT NULL, - message TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, started_at TEXT, completed_at TEXT + message TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL, started_at TEXT, completed_at TEXT, + files_deleted_at TEXT ); CREATE INDEX IF NOT EXISTS jobs_status_created ON jobs(status, created_at);`) if err != nil { db.Close() return nil, err } + if err := migrateJobFilesDeletedColumn(db); err != nil { + db.Close() + return nil, fmt.Errorf("migrate jobs: %w", err) + } + if _, err := db.Exec(`CREATE INDEX IF NOT EXISTS jobs_sweep ON jobs(files_deleted_at, completed_at)`); err != nil { + db.Close() + return nil, err + } // Unique on the pair, not on message_id alone: one email addressed to two // FileMill addresses arrives as two deliveries sharing a Message-Id, and // each is its own unit of work. A Mailgun retry repeats the recipient as @@ -173,6 +182,39 @@ func migrateSubmissionKey(db *sql.DB) error { return tx.Commit() } +// migrateJobFilesDeletedColumn adds files_deleted_at to jobs on a database +// created before the age-based job sweep existed. It is a no-op once the +// column exists, so it costs one PRAGMA per start on every database after +// the first — the same trade the CREATE TABLE IF NOT EXISTS statements above +// already make. +// +// A plain ALTER TABLE ADD COLUMN suffices here, unlike migrateSubmissionKey's +// rebuild: SQLite can add a nullable column in place, it just can't drop or +// widen a constraint that way. +func migrateJobFilesDeletedColumn(db *sql.DB) error { + rows, err := db.Query(`PRAGMA table_info(jobs)`) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + var cid, notnull, pk int + var name, ctype string + var dflt sql.NullString + if err := rows.Scan(&cid, &name, &ctype, ¬null, &dflt, &pk); err != nil { + return err + } + if name == "files_deleted_at" { + return nil + } + } + if err := rows.Err(); err != nil { + return err + } + _, err = db.Exec(`ALTER TABLE jobs ADD COLUMN files_deleted_at TEXT`) + return err +} + // uniqueIndexColumns returns one column list per unique index on a table, // including the implicit index behind a UNIQUE constraint. Reading the shape // this way rather than matching text in sqlite_master keeps the migration's @@ -398,6 +440,37 @@ func (s *Store) Complete(id, status, message string) error { _, err := s.db.Exec("UPDATE jobs SET status=?, message=?, completed_at=? WHERE id=?", status, message, time.Now().UTC().Format(time.RFC3339Nano), id) return err } +// ExpiredJobs returns ids of jobs that finished before cutoff and whose +// workspace the job sweep has not yet deleted from data/jobs. Only +// completed_at is checked, not created_at, so a job still queued or running +// is never a candidate no matter how old it is. +func (s *Store) ExpiredJobs(cutoff time.Time) ([]string, error) { + rows, err := s.db.Query("SELECT id FROM jobs WHERE files_deleted_at IS NULL AND completed_at IS NOT NULL AND completed_at