Sweep job workspaces after 30 days

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 {

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
+}

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)
+	}
+}

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, &notnull, &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<? ORDER BY completed_at",
+		cutoff.UTC().Format(time.RFC3339Nano))
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var ids []string
+	for rows.Next() {
+		var id string
+		if err := rows.Scan(&id); err != nil {
+			return nil, err
+		}
+		ids = append(ids, id)
+	}
+	return ids, rows.Err()
+}
+
+// MarkJobFilesDeleted records that the sweep removed a job's workspace from
+// disk. The job row itself is kept, so Get still answers after the files are
+// gone — only this one column changes.
+func (s *Store) MarkJobFilesDeleted(id string) error {
+	_, err := s.db.Exec("UPDATE jobs SET files_deleted_at=? WHERE id=?",
+		time.Now().UTC().Format(time.RFC3339Nano), id)
+	return err
+}
+
 func (s *Store) Get(id string) (Job, error) {
 	var j Job
 	var c string