Make the job sweep test time-independent and fix timestamp ordering

internal/app/app.go

@@ -51,7 +51,17 @@ type App struct {
 	log        *log.Logger
 	logFile    *os.File
 	reporter   alert.Reporter
+	now        func() time.Time // nil means time.Now; tests set a fake clock
 }
+
+// clock is the time the job sweep measures retention against.
+func (a *App) clock() time.Time {
+	if a.now != nil {
+		return a.now()
+	}
+	return time.Now()
+}
+
 type OutputFile struct{ Name, Path string }
 
 func Open(root string) (*App, error) {

internal/app/loops_test.go

@@ -10,7 +10,7 @@ import (
 func TestJobSweepTickSurvivesPanic(t *testing.T) {
 	a, rep := newTestApp(t, fakeTransformer(t, "succeed")...)
 	completedJob(t, a)
-	expireEverything(t)
+	expireEverything(t, a)
 	saved := removeWorkspace
 	removeWorkspace = func(string) error { panic("boom") }
 	t.Cleanup(func() { removeWorkspace = saved })

internal/app/retention.go

@@ -15,12 +15,8 @@ import (
 // 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
+// backup nobody is waiting on. Tests move App's clock instead of waiting.
+const jobRetentionPeriod = 30 * 24 * time.Hour
 
 // removeWorkspace deletes one job's workspace. A var so the sweep's test can
 // make a delete fail, which the filesystem won't do on request.
@@ -89,7 +85,7 @@ func (a *App) sweepJobsTick() {
 // 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))
+	ids, err := a.store.ExpiredJobs(a.clock().Add(-jobRetentionPeriod))
 	if err != nil {
 		return err
 	}

internal/app/retention_test.go

@@ -10,12 +10,11 @@ import (
 )
 
 // 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.
+// inside the horizon is left alone. The sweep's clock is set 30 days past a
+// boundary taken between the two completions, so the cutoff falls exactly
+// there. A slow machine only widens the gaps on either side of it - an earlier
+// version raced a 100ms horizon against the real clock and lost on slow CI.
 func TestSweepExpiredJobsDeletesOldWorkspaceOnly(t *testing.T) {
-	restore := jobRetentionPeriod
-	defer func() { jobRetentionPeriod = restore }()
-
 	root := t.TempDir()
 	writeTransformers(t, root, `transformers:
   - operation: copy_rename
@@ -40,11 +39,10 @@ func TestSweepExpiredJobsDeletesOldWorkspaceOnly(t *testing.T) {
 	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
+	// The pauses only make the three times distinct on Windows' coarse clock.
+	time.Sleep(20 * time.Millisecond)
+	boundary := time.Now()
+	time.Sleep(20 * time.Millisecond)
 
 	freshID, err := a.Submit("copy_rename", src)
 	if err != nil {
@@ -53,6 +51,7 @@ func TestSweepExpiredJobsDeletesOldWorkspaceOnly(t *testing.T) {
 	if err := a.store.Complete(freshID, store.StatusSucceeded, "ok"); err != nil {
 		t.Fatal(err)
 	}
+	a.now = func() time.Time { return boundary.Add(jobRetentionPeriod) }
 
 	if err := a.sweepExpiredJobs(); err != nil {
 		t.Fatalf("sweepExpiredJobs: %v", err)
@@ -83,10 +82,6 @@ func TestSweepExpiredJobsDeletesOldWorkspaceOnly(t *testing.T) {
 // 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
@@ -113,7 +108,7 @@ func TestSweepExpiredJobsToleratesAlreadyMissingWorkspace(t *testing.T) {
 	if err := os.RemoveAll(filepath.Join(root, "data", "jobs", id)); err != nil {
 		t.Fatal(err)
 	}
-	time.Sleep(5 * time.Millisecond)
+	expireEverything(t, a)
 
 	if err := a.sweepExpiredJobs(); err != nil {
 		t.Fatalf("sweepExpiredJobs must tolerate an already-missing workspace: %v", err)

internal/app/sweep_alert_test.go

@@ -11,15 +11,12 @@ import (
 	"filemill/internal/store"
 )
 
-// expireEverything makes the job sweep treat every completed job as past
-// retention, with no waiting. The horizon is a whole hour in the future, not a
-// nanosecond: the store compares RFC3339Nano text, which trims trailing zeros,
-// so times a fraction of a second apart can sort the wrong way.
-func expireEverything(t *testing.T) {
+// expireEverything makes the job sweep treat every job completed so far as
+// past retention, with no waiting, by running the sweep's clock one retention
+// period plus an hour ahead.
+func expireEverything(t *testing.T, a *App) {
 	t.Helper()
-	saved := jobRetentionPeriod
-	jobRetentionPeriod = -time.Hour
-	t.Cleanup(func() { jobRetentionPeriod = saved })
+	a.now = func() time.Time { return time.Now().Add(jobRetentionPeriod + time.Hour) }
 }
 
 func completedJob(t *testing.T, a *App) store.Job {
@@ -36,7 +33,7 @@ func completedJob(t *testing.T, a *App) store.Job {
 func TestSweepExpiredJobsReportsDeleteFailures(t *testing.T) {
 	a, rep := newTestApp(t, fakeTransformer(t, "succeed")...)
 	j := completedJob(t, a)
-	expireEverything(t)
+	expireEverything(t, a)
 	saved := removeWorkspace
 	removeWorkspace = func(string) error { return errors.New("access is denied") }
 	t.Cleanup(func() { removeWorkspace = saved })
@@ -59,7 +56,7 @@ func TestSweepExpiredJobsReportsDeleteFailures(t *testing.T) {
 func TestCleanJobSweepDoesNotReport(t *testing.T) {
 	a, rep := newTestApp(t, fakeTransformer(t, "succeed")...)
 	j := completedJob(t, a)
-	expireEverything(t)
+	expireEverything(t, a)
 
 	if err := a.sweepExpiredJobs(); err != nil {
 		t.Fatalf("sweepExpiredJobs: %v", err)

internal/store/alert.go

@@ -6,17 +6,10 @@ import (
 	"time"
 )
 
-// alertTimeFormat is fixed-width, unlike RFC3339Nano, which trims trailing
-// zeros from the fraction: "10:00:00Z" sorts after "10:00:00.5Z". The alert
-// windows are compared as strings in SQL, so the width matters.
-const alertTimeFormat = "2006-01-02T15:04:05.000000000Z07:00"
-
 // alertSendsRetention is the longest window the alert caps read, the rolling
 // 24h of the daily cap. Older rows are pruned on each write.
 const alertSendsRetention = 24 * time.Hour
 
-func formatAlertTime(t time.Time) string { return t.UTC().Format(alertTimeFormat) }
-
 // LastAlert returns when category last sent an alert and how many have been
 // suppressed since. Zero values mean it has no record; a zero sentAt with a
 // count means a global cap held it back before it ever sent.
@@ -33,7 +26,7 @@ func (s *Store) LastAlert(category string) (time.Time, int, error) {
 	if !sentAt.Valid {
 		return time.Time{}, suppressed, nil
 	}
-	t, err := time.Parse(alertTimeFormat, sentAt.String)
+	t, err := time.Parse(timeFormat, sentAt.String)
 	if err != nil {
 		return time.Time{}, 0, fmt.Errorf("alert_state %s: %w", category, err)
 	}
@@ -50,13 +43,13 @@ func (s *Store) RecordAlertSent(category string, at time.Time) error {
 	}
 	defer tx.Rollback()
 	if _, err := tx.Exec(`INSERT INTO alert_state(category,last_sent_at,suppressed) VALUES(?,?,0)
- ON CONFLICT(category) DO UPDATE SET last_sent_at=excluded.last_sent_at`, category, formatAlertTime(at)); err != nil {
+ ON CONFLICT(category) DO UPDATE SET last_sent_at=excluded.last_sent_at`, category, formatTime(at)); err != nil {
 		return err
 	}
-	if _, err := tx.Exec("INSERT INTO alert_sends(sent_at) VALUES(?)", formatAlertTime(at)); err != nil {
+	if _, err := tx.Exec("INSERT INTO alert_sends(sent_at) VALUES(?)", formatTime(at)); err != nil {
 		return err
 	}
-	if _, err := tx.Exec("DELETE FROM alert_sends WHERE sent_at<=?", formatAlertTime(at.Add(-alertSendsRetention))); err != nil {
+	if _, err := tx.Exec("DELETE FROM alert_sends WHERE sent_at<=?", formatTime(at.Add(-alertSendsRetention))); err != nil {
 		return err
 	}
 	return tx.Commit()
@@ -78,7 +71,7 @@ func (s *Store) RecordAlertSuppressed(category string) error {
 
 // AlertSendsSince returns every alert send after t, oldest first.
 func (s *Store) AlertSendsSince(t time.Time) ([]time.Time, error) {
-	rows, err := s.db.Query("SELECT sent_at FROM alert_sends WHERE sent_at>? ORDER BY sent_at", formatAlertTime(t))
+	rows, err := s.db.Query("SELECT sent_at FROM alert_sends WHERE sent_at>? ORDER BY sent_at", formatTime(t))
 	if err != nil {
 		return nil, err
 	}
@@ -89,7 +82,7 @@ func (s *Store) AlertSendsSince(t time.Time) ([]time.Time, error) {
 		if err := rows.Scan(&raw); err != nil {
 			return nil, err
 		}
-		at, err := time.Parse(alertTimeFormat, raw)
+		at, err := time.Parse(timeFormat, raw)
 		if err != nil {
 			return nil, fmt.Errorf("alert_sends: %w", err)
 		}

internal/store/store.go

@@ -9,6 +9,18 @@ import (
 	_ "modernc.org/sqlite"
 )
 
+// timeFormat is how every timestamp is stored. It is fixed-width, unlike
+// RFC3339Nano, which trims trailing zeros from the fraction: "10:00:00Z" sorts
+// after "10:00:00.5Z". Retention cutoffs, alert windows, and queue order are
+// all compared as strings in SQL, so the width matters.
+//
+// Rows written before jobs and deliveries used it are still RFC3339Nano. Reads
+// parse them with time.RFC3339Nano, which accepts both, and comparing them
+// against fixed-width text can misorder only within the same second.
+const timeFormat = "2006-01-02T15:04:05.000000000Z07:00"
+
+func formatTime(t time.Time) string { return t.UTC().Format(timeFormat) }
+
 type Job struct {
 	ID, Operation, Status, InputName, Message string
 	CreatedAt                                 time.Time
@@ -354,7 +366,7 @@ func (s *Store) PendingEmails() ([]EmailSubmission, error) {
 // has already been mailed out.
 func (s *Store) PutDelivery(submissionID int64, outputIndex int, fileID, link string) error {
 	_, err := s.db.Exec("INSERT OR IGNORE INTO email_deliveries(submission_id,output_index,file_id,link,created_at) VALUES(?,?,?,?,?)",
-		submissionID, outputIndex, fileID, link, time.Now().UTC().Format(time.RFC3339Nano))
+		submissionID, outputIndex, fileID, link, formatTime(time.Now()))
 	return err
 }
 
@@ -380,7 +392,7 @@ func (s *Store) Delivery(submissionID int64, outputIndex int) (Delivery, bool, e
 // retention sweep has not yet deleted.
 func (s *Store) ExpiredDeliveries(cutoff time.Time) ([]Delivery, error) {
 	rows, err := s.db.Query("SELECT submission_id,output_index,file_id,link,created_at FROM email_deliveries WHERE deleted_at IS NULL AND created_at<? ORDER BY created_at",
-		cutoff.UTC().Format(time.RFC3339Nano))
+		formatTime(cutoff))
 	if err != nil {
 		return nil, err
 	}
@@ -401,17 +413,17 @@ func (s *Store) ExpiredDeliveries(cutoff time.Time) ([]Delivery, error) {
 // MarkDeliveryDeleted records that the sweep removed the Drive file.
 func (s *Store) MarkDeliveryDeleted(submissionID int64, outputIndex int) error {
 	_, err := s.db.Exec("UPDATE email_deliveries SET deleted_at=? WHERE submission_id=? AND output_index=?",
-		time.Now().UTC().Format(time.RFC3339Nano), submissionID, outputIndex)
+		formatTime(time.Now()), submissionID, outputIndex)
 	return err
 }
 
 func (s *Store) MarkEmailDelivered(id int64) error {
-	_, err := s.db.Exec("UPDATE email_submissions SET delivered_at=? WHERE id=?", time.Now().UTC().Format(time.RFC3339Nano), id)
+	_, err := s.db.Exec("UPDATE email_submissions SET delivered_at=? WHERE id=?", formatTime(time.Now()), id)
 	return err
 }
 func (s *Store) Close() error { return s.db.Close() }
 func (s *Store) Create(j Job) error {
-	_, err := s.db.Exec("INSERT INTO jobs(id, operation, status, input_name, created_at) VALUES(?,?,?,?,?)", j.ID, j.Operation, StatusQueued, j.InputName, j.CreatedAt.UTC().Format(time.RFC3339Nano))
+	_, err := s.db.Exec("INSERT INTO jobs(id, operation, status, input_name, created_at) VALUES(?,?,?,?,?)", j.ID, j.Operation, StatusQueued, j.InputName, formatTime(j.CreatedAt))
 	return err
 }
 func (s *Store) Next() (*Job, error) {
@@ -429,7 +441,7 @@ func (s *Store) Next() (*Job, error) {
 		return nil, err
 	}
 	j.CreatedAt, _ = time.Parse(time.RFC3339Nano, created)
-	res, err := tx.Exec("UPDATE jobs SET status=?, started_at=? WHERE id=? AND status=?", StatusRunning, time.Now().UTC().Format(time.RFC3339Nano), j.ID, StatusQueued)
+	res, err := tx.Exec("UPDATE jobs SET status=?, started_at=? WHERE id=? AND status=?", StatusRunning, formatTime(time.Now()), j.ID, StatusQueued)
 	if err != nil {
 		return nil, err
 	}
@@ -444,7 +456,7 @@ func (s *Store) Next() (*Job, error) {
 	return &j, nil
 }
 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)
+	_, err := s.db.Exec("UPDATE jobs SET status=?, message=?, completed_at=? WHERE id=?", status, message, formatTime(time.Now()), id)
 	return err
 }
 
@@ -460,7 +472,7 @@ const interruptedMessage = "interrupted: FileMill stopped while this job was run
 // the delivery loop takes as finished.
 func (s *Store) InterruptRunning() (int, error) {
 	res, err := s.db.Exec("UPDATE jobs SET status=?, message=?, completed_at=? WHERE status=?",
-		StatusInterrupted, interruptedMessage, time.Now().UTC().Format(time.RFC3339Nano), StatusRunning)
+		StatusInterrupted, interruptedMessage, formatTime(time.Now()), StatusRunning)
 	if err != nil {
 		return 0, err
 	}
@@ -474,7 +486,7 @@ func (s *Store) InterruptRunning() (int, error) {
 // 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))
+		formatTime(cutoff))
 	if err != nil {
 		return nil, err
 	}
@@ -495,7 +507,7 @@ func (s *Store) ExpiredJobs(cutoff time.Time) ([]string, error) {
 // 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)
+		formatTime(time.Now()), id)
 	return err
 }