From 6eb5b70b83272e94d180d47d0b01070ef7adb6fe Mon Sep 17 00:00:00 2001 From: Kevin Brown Date: Wed, 28 Oct 2026 21:58:23 -0600 Subject: [PATCH] Make the job sweep test time-independent and fix timestamp ordering TestSweepExpiredJobsDeletesOldWorkspaceOnly shrank the retention period to 100ms and needed the fresh job's completion and the sweep to finish inside it. A slow CI runner took longer, the fresh job counted as expired, and the test failed. App now has a clock tests can set (the same pattern as mailgun.Service), and the test puts the cutoff exactly on a boundary taken between the two completions, so a slow machine only widens the margins. jobRetentionPeriod becomes a constant again. Underneath was a real ordering bug. The store wrote timestamps with RFC3339Nano, which trims trailing zeros, and compares them as text in SQL, so "10:00:00.5Z" sorted before "10:00:00Z": a job completed half a second after a cutoff counted as expired. The alert tables already used a fixed-width format for exactly this reason; it is now the store-wide timeFormat and every timestamp write uses it. Reads still parse with RFC3339Nano, which accepts both, so existing rows need no migration and can misorder against new ones only within the same second. Co-Authored-By: Claude Opus 5.5 --- internal/app/app.go | 10 +++ internal/app/loops_test.go | 2 +- internal/app/retention.go | 10 +-- internal/app/retention_test.go | 25 +++---- internal/app/sweep_alert_test.go | 17 ++--- internal/store/alert.go | 19 ++---- internal/store/store.go | 32 ++++++--- internal/store/store_test.go | 111 ++++++++++++++++++++++++++++++- 8 files changed, 168 insertions(+), 58 deletions(-) diff --git a/internal/app/app.go b/internal/app/app.go index 8dbebb3..42269dd 100644 --- a/internal/app/app.go +++ b/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) { diff --git a/internal/app/loops_test.go b/internal/app/loops_test.go index 520cd8e..316e031 100644 --- a/internal/app/loops_test.go +++ b/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 }) diff --git a/internal/app/retention.go b/internal/app/retention.go index 2c92e92..298d815 100644 --- a/internal/app/retention.go +++ b/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 } diff --git a/internal/app/retention_test.go b/internal/app/retention_test.go index 666e6cd..0a84d18 100644 --- a/internal/app/retention_test.go +++ b/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) diff --git a/internal/app/sweep_alert_test.go b/internal/app/sweep_alert_test.go index cf6a484..6ffef83 100644 --- a/internal/app/sweep_alert_test.go +++ b/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) diff --git a/internal/store/alert.go b/internal/store/alert.go index fa7b77a..fb82e84 100644 --- a/internal/store/alert.go +++ b/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) } diff --git a/internal/store/store.go b/internal/store/store.go index a85702c..7130f96 100644 --- a/internal/store/store.go +++ b/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", "a@example.com", "b@example.com", "subject") + if err != nil { + t.Fatal(err) + } + if err := s.MarkEmailDelivered(id); err != nil { + t.Fatal(err) + } + if err := s.PutDelivery(id, 0, "file", "https://docs.google.com/file"); err != nil { + t.Fatal(err) + } + if err := s.MarkDeliveryDeleted(id, 0); err != nil { + t.Fatal(err) + } + + for _, query := range []string{ + "SELECT created_at FROM jobs", + "SELECT started_at FROM jobs", + "SELECT completed_at FROM jobs", + "SELECT files_deleted_at FROM jobs WHERE id='job'", + "SELECT delivered_at FROM email_submissions", + "SELECT created_at FROM email_deliveries", + "SELECT deleted_at FROM email_deliveries", + } { + rows, err := s.db.Query(query) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var raw string + if err := rows.Scan(&raw); err != nil { + t.Fatal(err) + } + if _, err := time.Parse(timeFormat, raw); err != nil { + t.Errorf("%s: %q is not fixed-width: %v", query, raw, err) + } + } + rows.Close() + } +} + // A published Drive file is recorded before the reply is sent, so a retry after // a failed send finds the record and reuses the link instead of uploading a // second copy. The record is keyed per output file, because one job may declare