fix: harden run persistence and error surfacing from code review

Snapshot store paths under lock before async runs, roll back failed
start/save state, emit UI events only after successful persistence,
surface log write failures, and sync stale YAML docs to JSON.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
mixeme
2026-06-29 21:33:50 +03:00
parent 09c5edc993
commit e9fc9eaba0
16 changed files with 230 additions and 88 deletions
+33 -10
View File
@@ -42,11 +42,17 @@ func (s *Service) CreateJob(job domain.Job) (domain.Job, error) {
record := uiRecord(job.ID, job.Name, "Created", "Job was added")
prependLog(runtime, record)
err := s.store.SaveJobs(s.jobs)
if err != nil {
s.jobs = s.jobs[:len(s.jobs)-1]
delete(s.runtimes, job.ID)
delete(s.schedules, job.ID)
s.mu.Unlock()
return domain.Job{}, err
}
s.mu.Unlock()
s.emit(RunRecorded{Record: record})
s.emit(JobChanged{JobID: job.ID})
return job, err
return job, nil
}
// UpdateJob replaces the durable configuration of the job with the same ID,
@@ -82,9 +88,12 @@ func (s *Service) UpdateJob(job domain.Job) error {
err := s.store.SaveJobs(s.jobs)
s.mu.Unlock()
if err != nil {
return err
}
s.emit(RunRecorded{Record: record})
s.emit(JobChanged{JobID: job.ID})
return err
return nil
}
// DeleteJob removes the job with the given ID along with its runtime and cached
@@ -105,9 +114,12 @@ func (s *Service) DeleteJob(id int) error {
err := s.store.SaveJobs(s.jobs)
s.mu.Unlock()
if err != nil {
return err
}
s.emit(RunRecorded{Record: record})
s.emit(JobChanged{JobID: 0})
return err
return nil
}
// SetEnabled enables or disables a single job. Enabling moves it back to "Ready"
@@ -140,15 +152,19 @@ func (s *Service) SetEnabled(id int, enabled bool) error {
err := s.store.SaveJobs(s.jobs)
s.mu.Unlock()
if err != nil {
return err
}
s.emit(RunRecorded{Record: record})
s.emit(JobChanged{JobID: id})
return err
return nil
}
// SetGlobalPause flips the global pause that gates all execution, scheduled and
// manual. Each enabled job's next-run text reflects the new state immediately so
// the list view is understandable before the next tick. A "Paused"/"Resumed"
// scheduler activity record and a SchedulerStateChanged event are emitted.
// SetGlobalPause flips the global pause that gates scheduled execution.
// Manual "Run now" remains available while paused. Each enabled job's next-run
// text reflects the new state immediately so the list view is understandable
// before the next tick. A "Paused"/"Resumed" scheduler activity record and a
// SchedulerStateChanged event are emitted.
func (s *Service) SetGlobalPause(paused bool) error {
s.mu.Lock()
s.paused = paused
@@ -165,13 +181,16 @@ func (s *Service) SetGlobalPause(paused bool) error {
}
s.mu.Unlock()
if err != nil {
return err
}
state, detail := "Resumed", "All job execution resumed"
if paused {
state, detail = "Paused", "All job execution paused"
}
s.emit(RunRecorded{Record: uiRecord(0, "Scheduler", state, detail)})
s.emit(SchedulerStateChanged{Paused: paused})
return err
return nil
}
// ShouldNotifyOnFailure reports whether the user has enabled desktop
@@ -344,6 +363,10 @@ func validateJob(job domain.Job) error {
if job.Name == "" || job.Schedule == "" || job.Command == "" {
return errors.New("name, schedule, and command are required")
}
policy := strings.TrimSpace(job.OverlapPolicy)
if policy != "" && policy != string(domain.OverlapPolicySkip) && policy != string(domain.OverlapPolicyQueue) {
return errors.New("overlap policy must be 'skip', 'queue', or empty")
}
return nil
}
+21 -18
View File
@@ -100,6 +100,9 @@ func TestCreateJobValidates(t *testing.T) {
if got := svc.Jobs(); len(got) != 0 {
t.Errorf("invalid job should not be stored, jobs = %+v", got)
}
if _, err := svc.CreateJob(domain.Job{Name: "A", Schedule: "@every 1m", Command: "echo", OverlapPolicy: "invalid"}); err == nil {
t.Error("expected error for invalid overlap policy")
}
}
func TestUpdateJobKeepsRuntimeAndReflectsDisable(t *testing.T) {
@@ -247,11 +250,11 @@ func TestRunNowUsesRunnerAndRecords(t *testing.T) {
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}})
done := make(chan domain.RunRecord, 1)
svc.runJob = func(_ context.Context, job *domain.Job, trigger string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, trigger string, _ string) (domain.RunRecord, error) {
if trigger != "Manual" {
t.Errorf("trigger = %q, want Manual", trigger)
}
return domain.RunRecord{Time: "2026-06-19 12:00:00", JobID: job.ID, JobName: job.Name, State: "Success", Output: "ok"}
return domain.RunRecord{Time: "2026-06-19 12:00:00", JobID: job.ID, JobName: job.Name, State: "Success", Output: "ok"}, nil
}
svc.Subscribe(ObserverFunc(func(e Event) {
if rr, ok := e.(RunRecorded); ok && rr.Record.JobID == 1 {
@@ -295,11 +298,11 @@ func TestRunNowRefusedWhileAlreadyRunning(t *testing.T) {
entered := make(chan struct{}, 1)
release := make(chan struct{})
var calls int32
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) (domain.RunRecord, error) {
atomic.AddInt32(&calls, 1)
entered <- struct{}{}
<-release
return domain.RunRecord{Time: "2026-06-19 12:00:00", JobID: job.ID, JobName: job.Name, State: "Success"}
return domain.RunRecord{Time: "2026-06-19 12:00:00", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
}
done := make(chan struct{}, 1)
svc.Subscribe(ObserverFunc(func(e Event) {
@@ -338,12 +341,12 @@ func TestRunNowRefusedWhileAlreadyRunning(t *testing.T) {
func TestRunNowAllowedWhilePaused(t *testing.T) {
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}})
done := make(chan struct{}, 1)
svc.runJob = func(context.Context, *domain.Job, string, string) domain.RunRecord {
svc.runJob = func(context.Context, *domain.Job, string, string) (domain.RunRecord, error) {
select {
case done <- struct{}{}:
default:
}
return domain.RunRecord{State: "Success"}
return domain.RunRecord{State: "Success"}, nil
}
if err := svc.SetGlobalPause(true); err != nil {
t.Fatalf("SetGlobalPause: %v", err)
@@ -363,11 +366,11 @@ func TestRunDueStartsDueJob(t *testing.T) {
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}})
done := make(chan domain.RunRecord, 1)
svc.runJob = func(_ context.Context, job *domain.Job, trigger string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, trigger string, _ string) (domain.RunRecord, error) {
if trigger != "Schedule" {
t.Errorf("trigger = %q, want Schedule", trigger)
}
return domain.RunRecord{Time: "2026-06-19 12:00:00", JobID: job.ID, JobName: job.Name, State: "Success", Output: "ok"}
return domain.RunRecord{Time: "2026-06-19 12:00:00", JobID: job.ID, JobName: job.Name, State: "Success", Output: "ok"}, nil
}
svc.Subscribe(ObserverFunc(func(e Event) {
if rr, ok := e.(RunRecorded); ok && rr.Record.JobID == 1 && rr.Record.State == "Success" {
@@ -394,9 +397,9 @@ func TestRunDueStartsDueJob(t *testing.T) {
func TestRunDueSkipsJobNotYetDue(t *testing.T) {
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}})
var ran int32
svc.runJob = func(context.Context, *domain.Job, string, string) domain.RunRecord {
svc.runJob = func(context.Context, *domain.Job, string, string) (domain.RunRecord, error) {
atomic.AddInt32(&ran, 1)
return domain.RunRecord{}
return domain.RunRecord{}, nil
}
// Next-due is ~1m out, so nothing is due "now".
@@ -415,9 +418,9 @@ func TestRunDueSkipsJobInRunningState(t *testing.T) {
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}})
var calls int32
svc.runJob = func(context.Context, *domain.Job, string, string) domain.RunRecord {
svc.runJob = func(context.Context, *domain.Job, string, string) (domain.RunRecord, error) {
atomic.AddInt32(&calls, 1)
return domain.RunRecord{State: "Success"}
return domain.RunRecord{State: "Success"}, nil
}
// Force the job into "Running" with a past NextDue, simulating an in-flight
@@ -439,9 +442,9 @@ func TestRunDueSkipsJobInRunningState(t *testing.T) {
func TestRunDueDoesNothingWhilePaused(t *testing.T) {
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}})
var ran int32
svc.runJob = func(context.Context, *domain.Job, string, string) domain.RunRecord {
svc.runJob = func(context.Context, *domain.Job, string, string) (domain.RunRecord, error) {
atomic.AddInt32(&ran, 1)
return domain.RunRecord{}
return domain.RunRecord{}, nil
}
if err := svc.SetGlobalPause(true); err != nil {
t.Fatalf("SetGlobalPause: %v", err)
@@ -469,12 +472,12 @@ func TestStartDrivesRunDueOnTick(t *testing.T) {
svc := newTempService(t, []domain.Job{{ID: 1, Name: "A", Schedule: "@every 1m", Command: "echo", Enabled: true}})
done := make(chan struct{}, 1)
svc.runJob = func(context.Context, *domain.Job, string, string) domain.RunRecord {
svc.runJob = func(context.Context, *domain.Job, string, string) (domain.RunRecord, error) {
select {
case done <- struct{}{}:
default:
}
return domain.RunRecord{State: "Success"}
return domain.RunRecord{State: "Success"}, nil
}
clock := &appFakeClock{ticks: make(chan time.Time, 1), now: time.Now().Add(2 * time.Minute)}
@@ -593,13 +596,13 @@ func TestServiceRebuiltFromPausedStoreStartsPaused(t *testing.T) {
var ran int32
runStarted := make(chan struct{}, 1)
svc2.runJob = func(context.Context, *domain.Job, string, string) domain.RunRecord {
svc2.runJob = func(context.Context, *domain.Job, string, string) (domain.RunRecord, error) {
atomic.AddInt32(&ran, 1)
select {
case runStarted <- struct{}{}:
default:
}
return domain.RunRecord{}
return domain.RunRecord{}, nil
}
// RunDue must not start any job while paused: the scheduler stays paused after
+2
View File
@@ -9,7 +9,9 @@ import (
// store.Paths.DesktopIcon so ApplyAutostart can reference it.
func (s *Service) InstallDesktopIcon(appID string, iconBytes []byte) {
if iconPath, err := desktop.InstallDesktopIntegration(appID, s.store.Paths.ExecutablePath, iconBytes); err == nil {
s.mu.Lock()
s.store.Paths.DesktopIcon = iconPath
s.mu.Unlock()
}
}
+50 -15
View File
@@ -36,11 +36,13 @@ func (s *Service) RunNow(id int) error {
s.mu.Unlock()
return errors.New("another job is already running (sequential mode)")
}
err := s.startRunLocked(job, runtime, "Manual")
err := s.startRunLocked(job, runtime, "Manual", time.Now())
s.mu.Unlock()
// Reflect the "Running" transition; the run's completion emits again later.
s.emit(JobChanged{JobID: id})
if err == nil {
// Reflect the "Running" transition; the run's completion emits again later.
s.emit(JobChanged{JobID: id})
}
return err
}
@@ -86,8 +88,9 @@ func (s *Service) RunDue(now time.Time) {
// tick once the in-flight run has finished.
continue
}
if err := s.startRunLocked(job, runtime, "Schedule"); err != nil {
if err := s.startRunLocked(job, runtime, "Schedule", now); err != nil {
startErr = err
continue
}
started = append(started, job.ID)
running = true
@@ -103,22 +106,47 @@ func (s *Service) RunDue(now time.Time) {
}
}
// runEnv snapshots path and retention settings for one background run so
// executeRun does not read store.Paths or store.Config without holding mu.
type runEnv struct {
logsDir string
maxFiles int
maxAge int
}
// startRunLocked transitions a job to "Running", advances its NextDue to the next
// scheduled occurrence, persists that, and launches the run on a background
// goroutine. Advancing (rather than zeroing) NextDue keeps the schedule marching
// while the run is in flight, which is what lets RunDue notice a fresh occurrence
// firing during a long run and apply the overlap policy. The caller must hold mu.
func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, trigger string) error {
// now is the reference time for next-due advancement and the running placeholder.
func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, trigger string, now time.Time) error {
jobCopy := *job
prevState := runtime.LastState
prevNextRun := runtime.NextRun
prevOutput := runtime.Output
prevNextDue := runtime.NextDue
runtime.LastState = "Running"
runtime.NextRun = "Running"
runtime.Output = runningOutput(jobCopy, trigger, time.Now())
s.advanceNextDueLocked(job, runtime, time.Now())
err := s.store.SaveJobs(s.jobs)
runtime.Output = runningOutput(jobCopy, trigger, now)
s.advanceNextDueLocked(job, runtime, now)
if err := s.store.SaveJobs(s.jobs); err != nil {
runtime.LastState = prevState
runtime.NextRun = prevNextRun
runtime.Output = prevOutput
runtime.NextDue = prevNextDue
return err
}
env := runEnv{
logsDir: s.store.Paths.LogsDir,
maxFiles: s.store.Config.MaxLogFiles,
maxAge: s.store.Config.MaxLogAgeDays,
}
// Capture ctx under the lock so a concurrent Start/Stop cannot swap it out
// from under the goroutine after we release mu.
go s.executeRun(s.ctx, jobCopy, trigger)
return err
go s.executeRun(s.ctx, jobCopy, trigger, env)
return nil
}
// executeRun runs the job off the lock, then records the result back through the
@@ -126,11 +154,12 @@ func (s *Service) startRunLocked(job *domain.Job, runtime *domain.JobRuntime, tr
// running (the "queue" overlap policy), and it is still enabled and the scheduler
// is not paused, the deferred run is started immediately. It runs on its own
// goroutine.
func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger string) {
record := s.runJob(ctx, &jobCopy, trigger, s.store.Paths.LogsDir)
func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger string, env runEnv) {
record, logErr := s.runJob(ctx, &jobCopy, trigger, env.logsDir)
s.mu.Lock()
var cleanupErr, saveErr error
var rerunStarted bool
if current := s.findByIDLocked(jobCopy.ID); current != nil {
runtime := s.runtimeForLocked(current)
runtime.LastRun = record.Time
@@ -143,15 +172,19 @@ func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger st
if rerun {
// A scheduled occurrence fired while this run was active under the
// "queue" policy; start that deferred run now.
saveErr = s.startRunLocked(current, runtime, "Schedule")
saveErr = s.startRunLocked(current, runtime, "Schedule", time.Now())
rerunStarted = saveErr == nil
} else {
s.refreshNextRunLocked(current, runtime)
saveErr = s.store.SaveJobs(s.jobs)
}
cleanupErr = runner.CleanupLogs(s.store.Paths.LogsDir, s.store.Config.MaxLogFiles, s.store.Config.MaxLogAgeDays)
cleanupErr = runner.CleanupLogs(env.logsDir, env.maxFiles, env.maxAge)
}
s.mu.Unlock()
if logErr != nil {
s.emit(ErrorOccurred{Err: fmt.Errorf("write run log for %q: %w", jobCopy.Name, logErr)})
}
if cleanupErr != nil {
s.emit(ErrorOccurred{Err: fmt.Errorf("log cleanup after run %q: %w", jobCopy.Name, cleanupErr)})
}
@@ -159,7 +192,9 @@ func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger st
s.emit(ErrorOccurred{Err: fmt.Errorf("save jobs after run %q: %w", jobCopy.Name, saveErr)})
}
s.emit(RunRecorded{Record: record})
s.emit(JobChanged{JobID: jobCopy.ID})
if !rerunStarted {
s.emit(JobChanged{JobID: jobCopy.ID})
}
}
// effectiveOverlapPolicy resolves the overlap policy that actually governs a
+16 -16
View File
@@ -116,10 +116,10 @@ func TestRunDueParallelStartsAllDueJobs(t *testing.T) {
entered := make(chan int, 2)
release := make(chan struct{})
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) (domain.RunRecord, error) {
entered <- job.ID
<-release
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
}
done := completions(svc)
@@ -157,10 +157,10 @@ func TestRunDueSequentialSerializes(t *testing.T) {
entered := make(chan int, 2)
release := make(chan struct{})
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) (domain.RunRecord, error) {
entered <- job.ID
<-release
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
}
done := completions(svc)
@@ -194,11 +194,11 @@ func TestRunDueSkipDropsOverlap(t *testing.T) {
entered := make(chan int, 2)
release := make(chan struct{})
var calls int32
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) (domain.RunRecord, error) {
atomic.AddInt32(&calls, 1)
entered <- job.ID
<-release
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
}
done := completions(svc)
@@ -240,11 +240,11 @@ func TestRunDueQueueRerunsAfterFinish(t *testing.T) {
entered := make(chan int, 2)
release := make(chan struct{})
var calls int32
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) (domain.RunRecord, error) {
atomic.AddInt32(&calls, 1)
entered <- job.ID
<-release
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
}
done := completions(svc)
@@ -297,11 +297,11 @@ func TestRunDuePerJobQueueOverridesGlobalSkip(t *testing.T) {
entered := make(chan int, 2)
release := make(chan struct{})
var calls int32
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) (domain.RunRecord, error) {
atomic.AddInt32(&calls, 1)
entered <- job.ID
<-release
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
}
done := completions(svc)
@@ -347,11 +347,11 @@ func TestRunDuePerJobSkipOverridesGlobalQueue(t *testing.T) {
entered := make(chan int, 2)
release := make(chan struct{})
var calls int32
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) (domain.RunRecord, error) {
atomic.AddInt32(&calls, 1)
entered <- job.ID
<-release
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
}
done := completions(svc)
@@ -395,10 +395,10 @@ func TestRunDueEmptyOverlapInheritsGlobal(t *testing.T) {
entered := make(chan int, 2)
release := make(chan struct{})
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) (domain.RunRecord, error) {
entered <- job.ID
<-release
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
}
done := completions(svc)
@@ -437,10 +437,10 @@ func TestRunNowSequentialGuard(t *testing.T) {
entered := make(chan int, 2)
release := make(chan struct{})
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) domain.RunRecord {
svc.runJob = func(_ context.Context, job *domain.Job, _ string, _ string) (domain.RunRecord, error) {
entered <- job.ID
<-release
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}
return domain.RunRecord{Time: "t", JobID: job.ID, JobName: job.Name, State: "Success"}, nil
}
done := completions(svc)
+1 -1
View File
@@ -51,7 +51,7 @@ type Service struct {
// processes. ctx is the lifecycle context passed to runs; Start replaces it
// with a cancelable context so Stop can abort in-flight runs, and until Start
// it is context.Background().
runJob func(ctx context.Context, job *domain.Job, trigger string, logsDir string) domain.RunRecord
runJob func(ctx context.Context, job *domain.Job, trigger string, logsDir string) (domain.RunRecord, error)
ctx context.Context
// sched is the timing loop installed by Start; cancel tears down ctx on Stop.