From 55d8c55e122a555e7e7c42a6eb8d52c0ab0cf8df Mon Sep 17 00:00:00 2001 From: PierrunoYT Date: Tue, 14 Jul 2026 21:55:02 +0200 Subject: [PATCH 1/2] fix(cron): reserve job IDs atomically --- internal/cron/store.go | 28 ++++++++++++--- internal/cron/store_test.go | 69 +++++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+), 5 deletions(-) diff --git a/internal/cron/store.go b/internal/cron/store.go index 47c743238..4e5a12e51 100644 --- a/internal/cron/store.go +++ b/internal/cron/store.go @@ -99,27 +99,35 @@ func (s *Store) Add(job Job) (Job, error) { job.Status = StatusActive } job.CreatedAt = s.now().UTC() - id, err := s.allocID() + id, err := s.reserveID() if err != nil { return Job{}, err } job.ID = id - if err := s.writeJob(job); err != nil { + if err := s.writeReservedJob(job); err != nil { return Job{}, err } return job, nil } -func (s *Store) allocID() (string, error) { +func (s *Store) reserveID() (string, error) { + if err := os.MkdirAll(s.root, 0o700); err != nil { + return "", fmt.Errorf("create cron store root: %w", err) + } base := s.now().UTC().Format("20060102-150405") for n := 0; n < 100; n++ { id := base if n > 0 { id = fmt.Sprintf("%s-%d", base, n) } - if _, err := os.Stat(filepath.Join(s.root, id)); errors.Is(err, os.ErrNotExist) { + err := os.Mkdir(s.jobDir(id), 0o700) + if err == nil { return id, nil } + if errors.Is(err, os.ErrExist) { + continue + } + return "", fmt.Errorf("reserve cron job id %q: %w", id, err) } return "", errors.New("could not allocate a unique cron job id") } @@ -127,7 +135,7 @@ func (s *Store) allocID() (string, error) { func (s *Store) jobDir(id string) string { return filepath.Join(s.root, id) } // validID rejects ids that could escape the store root (path separators or -// traversal). allocID-generated timestamp ids always pass; this guards +// traversal). reserveID-generated timestamp ids always pass; this guards // externally-supplied ids (get/update/remove/append). func validID(id string) bool { return id != "" && id != "." && id != ".." && !strings.ContainsAny(id, `/\`) && filepath.Base(id) == id @@ -153,6 +161,16 @@ func (s *Store) writeJob(job Job) error { return nil } +func (s *Store) writeReservedJob(job Job) error { + if err := s.writeJob(job); err != nil { + if cleanupErr := os.RemoveAll(s.jobDir(job.ID)); cleanupErr != nil { + return errors.Join(err, fmt.Errorf("remove failed cron job reservation: %w", cleanupErr)) + } + return err + } + return nil +} + func (s *Store) Get(id string) (Job, error) { if !validID(id) { return Job{}, fmt.Errorf("invalid cron job id %q", id) diff --git a/internal/cron/store_test.go b/internal/cron/store_test.go index 9aada15ec..079721dde 100644 --- a/internal/cron/store_test.go +++ b/internal/cron/store_test.go @@ -2,9 +2,11 @@ package cron import ( "errors" + "fmt" "os" "path/filepath" "strings" + "sync" "testing" "time" ) @@ -74,6 +76,73 @@ func TestStoreAddListGetRemove(t *testing.T) { } } +func TestStoreAddReservesUniqueIDsAcrossConcurrentStores(t *testing.T) { + root := t.TempDir() + now := func() time.Time { return time.Unix(1000, 0).UTC() } + const adds = 32 + + start := make(chan struct{}) + results := make(chan Job, adds) + errs := make(chan error, adds) + var wg sync.WaitGroup + for i := 0; i < adds; i++ { + wg.Add(1) + go func(index int) { + defer wg.Done() + store := NewStore(StoreOptions{RootDir: root, Now: now}) + <-start + job, err := store.Add(Job{Expr: "* * * * *", Prompt: fmt.Sprintf("job-%d", index)}) + if err != nil { + errs <- err + return + } + results <- job + }(i) + } + close(start) + wg.Wait() + close(results) + close(errs) + + for err := range errs { + t.Errorf("concurrent Add: %v", err) + } + seen := make(map[string]struct{}, adds) + reader := NewStore(StoreOptions{RootDir: root, Now: now}) + for job := range results { + if _, duplicate := seen[job.ID]; duplicate { + t.Errorf("concurrent Add returned duplicate ID %q", job.ID) + } + seen[job.ID] = struct{}{} + persisted, err := reader.Get(job.ID) + if err != nil { + t.Errorf("Get(%q): %v", job.ID, err) + continue + } + if persisted.Prompt != job.Prompt { + t.Errorf("Get(%q).Prompt = %q, want %q", job.ID, persisted.Prompt, job.Prompt) + } + } + if len(seen) != adds { + t.Fatalf("unique jobs = %d, want %d", len(seen), adds) + } +} + +func TestWriteReservedJobRemovesReservationOnFailure(t *testing.T) { + store := newTestStore(t) + job := Job{ID: "reserved", Expr: "* * * * *", Prompt: "job"} + if err := os.MkdirAll(filepath.Join(store.jobDir(job.ID), "metadata.json.tmp"), 0o700); err != nil { + t.Fatal(err) + } + + if err := store.writeReservedJob(job); err == nil { + t.Fatal("writeReservedJob error = nil, want metadata write failure") + } + if _, err := os.Stat(store.jobDir(job.ID)); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("failed reservation still exists: %v", err) + } +} + func TestStoreAppendRun(t *testing.T) { s := newTestStore(t) job, _ := s.Add(Job{Expr: "* * * * *", Prompt: "x", Status: StatusActive}) From ef7537b15421973c426360f420fbf6679f2df3ee Mon Sep 17 00:00:00 2001 From: PierrunoYT Date: Wed, 15 Jul 2026 04:37:49 +0200 Subject: [PATCH 2/2] fix(cron): hold the per-ID lock from reservation through persistence Address code review on PR #686: reserveID made the timestamp directory visible via os.Mkdir before metadata.json existed, but Remove (which takes the per-ID lock before deleting) never coordinated with that window. A concurrent `zero cron rm ` could delete the bare reservation and report success right after Add reserved it; writeJob's MkdirAll would then silently resurrect the directory and persist the job anyway. A second Add could also re-reserve the same id while the first writer was still running, putting two writers on the same metadata path. reserveID now acquires the same per-ID lock (used by Get/Update/ Mutate/Remove/AppendRun) for each candidate id before attempting os.Mkdir, and returns it still held; Add releases it only after writeReservedJob commits (or tears down) the reservation. A concurrent Remove for that id now blocks until the reservation is resolved one way or the other, and a concurrent Add attempting the same candidate id serializes through the lock rather than only racing the Mkdir. Adds two regression tests: one proving Remove waits for and removes the committed job rather than the bare reservation, and one proving a second reserveID call for the same candidate id blocks on the shared lock and moves on to the next candidate. Co-Authored-By: Claude Sonnet 5 --- internal/cron/store.go | 35 +++++++++++---- internal/cron/store_test.go | 87 +++++++++++++++++++++++++++++++++++++ 2 files changed, 113 insertions(+), 9 deletions(-) diff --git a/internal/cron/store.go b/internal/cron/store.go index 4e5a12e51..d23e79145 100644 --- a/internal/cron/store.go +++ b/internal/cron/store.go @@ -99,10 +99,17 @@ func (s *Store) Add(job Job) (Job, error) { job.Status = StatusActive } job.CreatedAt = s.now().UTC() - id, err := s.reserveID() + id, unlock, err := s.reserveID() if err != nil { return Job{}, err } + // Held from reservation through initial persistence: without this, Remove + // (which takes the same per-ID lock before deleting) could delete the + // bare reservation directory the instant after Mkdir and report success, + // then writeJob's MkdirAll would silently resurrect it and persist the + // job anyway — or a second Add could re-reserve the freed id while this + // writer is still running, putting two writers on the same metadata path. + defer unlock() job.ID = id if err := s.writeReservedJob(job); err != nil { return Job{}, err @@ -110,9 +117,14 @@ func (s *Store) Add(job Job) (Job, error) { return job, nil } -func (s *Store) reserveID() (string, error) { +// reserveID atomically reserves an unused job id (by os.Mkdir-ing its +// directory) and returns the per-ID lock held for that id, still acquired. +// The caller must release it (via the returned unlock) only once the +// reservation has been committed (metadata.json written) or torn down — +// see Add. +func (s *Store) reserveID() (string, func(), error) { if err := os.MkdirAll(s.root, 0o700); err != nil { - return "", fmt.Errorf("create cron store root: %w", err) + return "", nil, fmt.Errorf("create cron store root: %w", err) } base := s.now().UTC().Format("20060102-150405") for n := 0; n < 100; n++ { @@ -120,16 +132,21 @@ func (s *Store) reserveID() (string, error) { if n > 0 { id = fmt.Sprintf("%s-%d", base, n) } - err := os.Mkdir(s.jobDir(id), 0o700) - if err == nil { - return id, nil + unlock, err := s.lockJob(id) + if err != nil { + return "", nil, err + } + mkErr := os.Mkdir(s.jobDir(id), 0o700) + if mkErr == nil { + return id, unlock, nil } - if errors.Is(err, os.ErrExist) { + unlock() + if errors.Is(mkErr, os.ErrExist) { continue } - return "", fmt.Errorf("reserve cron job id %q: %w", id, err) + return "", nil, fmt.Errorf("reserve cron job id %q: %w", id, mkErr) } - return "", errors.New("could not allocate a unique cron job id") + return "", nil, errors.New("could not allocate a unique cron job id") } func (s *Store) jobDir(id string) string { return filepath.Join(s.root, id) } diff --git a/internal/cron/store_test.go b/internal/cron/store_test.go index 079721dde..9eb4d5789 100644 --- a/internal/cron/store_test.go +++ b/internal/cron/store_test.go @@ -128,6 +128,93 @@ func TestStoreAddReservesUniqueIDsAcrossConcurrentStores(t *testing.T) { } } +// TestAddDoesNotLeakReservationToConcurrentRemove reproduces the audit +// finding that Remove could observe and delete an in-flight Add's bare +// reservation directory (Mkdir'd but not yet holding metadata.json) and +// report success — after which Add's writeJob would silently resurrect the +// directory and persist the job anyway. Add now holds the same per-ID lock +// Remove takes, from reservation through initial persistence, so a +// concurrent Remove for that id must wait until the reservation is either +// committed or torn down, never observing the bare reservation in between. +func TestAddDoesNotLeakReservationToConcurrentRemove(t *testing.T) { + store := newTestStore(t) + + // Simulate the in-flight window Add now holds the lock across: the id is + // reserved (its directory exists) but metadata.json has not been written. + id, unlock, err := store.reserveID() + if err != nil { + t.Fatalf("reserveID: %v", err) + } + if _, statErr := os.Stat(store.jobDir(id)); statErr != nil { + t.Fatalf("reservation directory missing: %v", statErr) + } + + removeErr := make(chan error, 1) + go func() { + removeErr <- store.Remove(id) + }() + + // Remove cannot proceed past the shared per-ID lock we're holding, so it's + // safe to commit the job here, before releasing it: if Remove could race + // ahead of this write, it would delete the bare reservation instead of + // waiting to observe (and remove) the committed job. + job := Job{ID: id, Expr: "* * * * *", Prompt: "job", Status: StatusActive} + if err := store.writeJob(job); err != nil { + t.Fatalf("writeJob: %v", err) + } + unlock() + + if err := <-removeErr; err != nil { + t.Fatalf("Remove of the committed job: %v", err) + } + if _, err := os.Stat(store.jobDir(id)); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("job directory should be gone after Remove: %v", err) + } + if _, err := store.Get(id); !errors.Is(err, ErrJobNotFound) { + t.Fatalf("Get after Remove = %v, want ErrJobNotFound", err) + } +} + +// TestReserveIDSerializesOnPerIDLock covers the other half of the audit +// finding: a second Add attempting the SAME candidate id while the first +// reservation is still held must block on the shared per-ID lock (not just +// race the directory Mkdir), so it never ends up writing metadata.json for +// an id whose first writer is still in flight. It moves on to the next +// candidate id instead. +func TestReserveIDSerializesOnPerIDLock(t *testing.T) { + store := newTestStore(t) + + id, unlock, err := store.reserveID() + if err != nil { + t.Fatalf("first reserveID: %v", err) + } + + type reservation struct { + id string + err error + } + second := make(chan reservation, 1) + go func() { + id2, unlock2, err := store.reserveID() + if err == nil { + unlock2() + } + second <- reservation{id: id2, err: err} + }() + + // The second reserveID cannot observe an id until this lock is released: + // it contends on the same per-ID lock file before ever attempting Mkdir. + unlock() + + got := <-second + if got.err != nil { + t.Fatalf("second reserveID: %v", got.err) + } + if got.id == id { + t.Fatalf("second reserveID returned the first id %q while its reservation was still held", got.id) + } +} + func TestWriteReservedJobRemovesReservationOnFailure(t *testing.T) { store := newTestStore(t) job := Job{ID: "reserved", Expr: "* * * * *", Prompt: "job"}