perf: keep file I/O off Service.mu and untie StartOnly from the app context

Phase 7 of the whole-project review (findings 3.2 and 3.3).

Service.mu is the lock the Fyne main thread takes on every Jobs() and
Runtime() call, so anything blocking inside it makes a UI refresh wait on
the disk. Three things did:

- Every SaveJobs/SaveConfig was a marshal, fsync, and rename under mu.
  Writes are now prepared under the lock (Store.PrepareSaveJobs /
  PrepareSaveConfig snapshot the payload and target path) and run after
  it is released. deferSaveLocked takes saveMu while mu is still held, so
  writes still reach the file in the order their snapshots were taken and
  an older snapshot can never land on top of a newer one.
- executeRun ran runner.CleanupLogs under mu after every run. It needs
  only the values already snapshotted into runEnv, so it now runs after
  the unlock — including when the job is gone, since the run still wrote
  a log file that retention covers.
- adoptJobsLocked ran runner.SeedStats under mu, reached from
  UpdateSettings on the UI thread. Seeding moved out into
  applySeededStatsLocked; UpdateSettings now reads the new jobs file and
  seeds its statistics before taking the lock, and re-checks the
  "no jobs-file switch while running" guard once it has it.

SeedStats also opened every log file twice — once to find the job, again
to read the result. readLogSummary reads job_id, state, and duration in
one pass, so each log is opened once.

StartOnly runs were built with exec.CommandContext on the app's lifecycle
context. os/exec keeps a watcher goroutine alive until Wait returns or the
context is done, and StartOnly never calls Wait, so one goroutine leaked
per run and would then try to kill a process whose handle startJobOnly had
already released. The invocation now uses context.Background(), whose nil
Done channel means no watcher is started at all.

Regression tests: TestRunJobStartOnlyLeavesNoContextWatcher (fails with 5
leaked goroutines on the old code), TestConcurrentJobOperationsLeaveTheFileMatchingMemory,
and TestUpdateSettingsSeedsAdoptedJobsFromLogs. STANDARDS gains the
no-I/O-under-mu rule and the "a StartOnly process outlives GoSentry" entry.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-06 21:44:51 +03:00
parent 98c820e3bd
commit 0c8442a8d1
12 changed files with 406 additions and 115 deletions
+13 -9
View File
@@ -136,10 +136,13 @@ example window-maximized detection, which would need per-OS native calls).
`UpdateSettings` has one extra step: when the configured jobs file changes `UpdateSettings` has one extra step: when the configured jobs file changes
and a file already exists at the new path, that file is authoritative. The and a file already exists at the new path, that file is authoritative. The
Service loads it, calls `adoptJobsLocked` to rebuild the jobs slice, runtime Service loads it, calls `adoptJobsLocked` to rebuild the jobs slice, runtime
map, schedule cache, next-run times, and log-seeded statistics around it, and map, schedule cache, and next-run times around it, applies the statistics
emits `JobsLoaded` plus a broad `JobChanged`. A path with no file behind it seeded from the new logs directory, and emits `JobsLoaded` plus a broad
receives the current jobs instead. Adoption drops all runtime state, so it is `JobChanged`. A path with no file behind it receives the current jobs instead.
refused while a job is running. Adoption drops all runtime state, so it is refused while a job is running.
Reading the new file and seeding its statistics both happen before `mu` is
taken (the no-I/O-under-`mu` rule in [STANDARDS.md](STANDARDS.md)), so the
running-job check is re-evaluated under the lock before anything is replaced.
3. Scheduled run: 3. Scheduled run:
`scheduler.Scheduler` fires a tick every second. On each tick it calls `scheduler.Scheduler` fires a tick every second. On each tick it calls
@@ -163,8 +166,9 @@ example window-maximized detection, which would need per-OS native calls).
6. History update: 6. History update:
When a run goroutine completes, `Service` updates the job's runtime When a run goroutine completes, `Service` updates the job's runtime
(including the statistics aggregate), saves JSON, triggers log cleanup, and (including the statistics aggregate) under `mu`, then — after releasing it —
emits `RunRecorded`. The UI observer appends the record to the History tab. runs log cleanup and emits `RunRecorded`. Nothing is saved: a run changes only
`JobRuntime`, which is never persisted. The UI observer appends the record to the History tab.
History rows exist only for the current process session; restarting the app History rows exist only for the current process session; restarting the app
clears the table (aggregate stats in the details panel are still seeded from clears the table (aggregate stats in the details panel are still seeded from
log files). log files).
@@ -220,9 +224,9 @@ resolves the effective duration under `mu` and `startRunLocked` snapshots it int
resolved duration as an argument, so the runner stays ignorant of the global resolved duration as an argument, so the runner stays ignorant of the global
config: a positive duration applies the timeout via `context.WithTimeout` and config: a positive duration applies the timeout via `context.WithTimeout` and
reports `Timed out after <timeout>` on expiry; a non-positive duration runs reports `Timed out after <timeout>` on expiry; a non-positive duration runs
without a deadline, bounded only by `ctx` (app shutdown). `StartOnly` jobs run on without a deadline, bounded only by `ctx` (app shutdown). `StartOnly` jobs are
the untimed context and so measure launch latency only, unaffected by the run built on `context.Background()` instead — neither the timeout nor app shutdown
timeout. applies to them — and so measure launch latency only.
### Run-time statistics ### Run-time statistics
+12
View File
@@ -39,6 +39,11 @@ the app icon (experimental).**
longer than its own interval no longer accumulates an unbounded backlog that longer than its own interval no longer accumulates an unbounded backlog that
then runs back-to-back indefinitely. The job details pane now shows the then runs back-to-back indefinitely. The job details pane now shows the
queued-run count (", N queued") whenever it is non-zero. queued-run count (", N queued") whenever it is non-zero.
- **Start-only jobs are no longer tied to the application's lifetime.** A job
with *Start only* checked is launched on an uncancelable context, so quitting
GoSentry (or a run context being cancelled) can no longer try to kill a
process it deliberately stopped waiting for. This also removes a goroutine
that leaked on every start-only run and lived until the app exited.
- The History tab no longer grows without bound: it keeps the newest 1000 - The History tab no longer grows without bound: it keeps the newest 1000
records and drops the oldest, the way a job's own activity list is capped. records and drops the oldest, the way a job's own activity list is capped.
Column widths are also folded in one record at a time instead of being Column widths are also folded in one record at a time instead of being
@@ -64,6 +69,13 @@ the app icon (experimental).**
- App-side failure-notification timing is appended to `logs/notify-timing.log` - App-side failure-notification timing is appended to `logs/notify-timing.log`
for diagnosing toast delay (OS latency excluded). `scripts/measure-windows-toast.ps1` for diagnosing toast delay (OS latency excluded). `scripts/measure-windows-toast.ps1`
measures the PowerShell baseline on Windows. measures the PowerShell baseline on Windows.
- File I/O no longer happens while `Service.mu` is held — that is the lock the
UI thread takes on every job and runtime read, so a JSON write, the
post-run log cleanup, or the startup log scan used to make a UI refresh wait
on the disk. Saves are now prepared under the lock and written after it is
released, in preparation order, so `jobs.json` still ends up matching the
in-memory list. Seeding statistics from logs also opens each log file once
instead of twice.
## 1.0.1 - 2026-08-04 ## 1.0.1 - 2026-08-04
+17
View File
@@ -11,6 +11,15 @@ in [ARCHITECTURE.md](ARCHITECTURE.md); test conventions in [TESTS.md](TESTS.md).
- Fixes with severity ≥ medium → regression test. - Fixes with severity ≥ medium → regression test.
- Documented intentional behavior → section below, not a backlog bug. - Documented intentional behavior → section below, not a backlog bug.
- UI view constructors accept `*app.Service`; call `app.Open()` only from `run.go`. - UI view constructors accept `*app.Service`; call `app.Open()` only from `run.go`.
- **No blocking file I/O under `Service.mu`.** It is the lock the Fyne main
thread takes on every `Jobs()` and `Runtime()` call, so a JSON write, a
log-directory scan, or a pass over every log header inside it makes a UI
refresh wait on the disk. Mutate state under the lock, snapshot what the I/O
needs, and run the I/O after `mu.Unlock()` — the way `emit()` already is.
Store writes go through `Service.deferSaveLocked` and `Store.PrepareSaveJobs` /
`Store.PrepareSaveConfig`, which take `saveMu` while `mu` is still held so
writes still reach the file in the order their snapshots were taken; log
cleanup and `runner.SeedStats` run from plain snapshots.
- A size that must follow the theme is **measured at build time, not written as - A size that must follow the theme is **measured at build time, not written as
a pixel constant.** `theme.Padding()` and text metrics depend on the running a pixel constant.** `theme.Padding()` and text metrics depend on the running
app's theme, text size, and DPI, so a hand-tuned number is only correct for app's theme, text size, and DPI, so a hand-tuned number is only correct for
@@ -62,6 +71,14 @@ change to their shape has to stay compatible on its own.
= 0) and is overridable per job (`Job.TimeoutSeconds *int`: unset = inherit the = 0) and is overridable per job (`Job.TimeoutSeconds *int`: unset = inherit the
global default, 0 = no timeout, positive = seconds). Neither zero may be global default, 0 = no timeout, positive = seconds). Neither zero may be
normalized away on load — 0 is a value, not a missing field. normalized away on load — 0 is a value, not a missing field.
- **A `StartOnly` process is expected to outlive GoSentry.** The option exists to
launch something and let go of it, so the runner builds that invocation on
`context.Background()`, not on the application's lifecycle context: quitting
GoSentry (or cancelling a run) does not stop a process it started this way, and
`Service.Stop()` reaches only jobs the runner is still waiting on. The
uncancelable context is also what keeps `os/exec` from leaving a watcher
goroutine per run — it only starts one when the context can be done, and
`StartOnly` never calls `Wait` to end it.
- **History tab is session-only.** `JobRuntime.Logs` exists only in memory for the - **History tab is session-only.** `JobRuntime.Logs` exists only in memory for the
current process. Log files on disk feed aggregate statistics via `SeedStats` current process. Log files on disk feed aggregate statistics via `SeedStats`
only. See [ARCHITECTURE.md](ARCHITECTURE.md). only. See [ARCHITECTURE.md](ARCHITECTURE.md).
+3
View File
@@ -166,6 +166,8 @@ Tests all mutating operations on the Service, scheduler integration, and setting
| `TestUpdateSettingsAdoptsExistingJobsFile` | Verifies that selecting a jobs file that already exists replaces the job list with its contents, rebuilds runtimes, and emits `JobsLoaded`. | | `TestUpdateSettingsAdoptsExistingJobsFile` | Verifies that selecting a jobs file that already exists replaces the job list with its contents, rebuilds runtimes, and emits `JobsLoaded`. |
| `TestUpdateSettingsKeepsJobsWhenTheNewFileIsMissing` | Verifies that a path with no file behind it receives the current jobs instead (the rename/relocate case). | | `TestUpdateSettingsKeepsJobsWhenTheNewFileIsMissing` | Verifies that a path with no file behind it receives the current jobs instead (the rename/relocate case). |
| `TestUpdateSettingsRefusesJobsFileSwitchWhileRunning` | Verifies that switching the jobs file is refused (and not persisted) while a job runs, while unrelated settings still save. | | `TestUpdateSettingsRefusesJobsFileSwitchWhileRunning` | Verifies that switching the jobs file is refused (and not persisted) while a job runs, while unrelated settings still save. |
| `TestUpdateSettingsSeedsAdoptedJobsFromLogs` | Verifies that statistics reconstructed from the new logs directory still reach the runtime map, now that the log scan happens before `UpdateSettings` takes `mu`. |
| `TestConcurrentJobOperationsLeaveTheFileMatchingMemory` | Verifies that saves prepared under `mu` and run after it is released still land in mutation order, so `jobs.json` matches the in-memory list after concurrent create/disable operations. |
| `TestSetJobListViewPersistsToConfigFile` | Verifies the Jobs-list density preference reaches `gosentry.json`, so the chosen view reopens after a restart. | | `TestSetJobListViewPersistsToConfigFile` | Verifies the Jobs-list density preference reaches `gosentry.json`, so the chosen view reopens after a restart. |
| `TestSetJobListViewNormalizesUnknownValue` | Verifies anything but `"compact"` is stored as `"detailed"`, so the config never gains a value no reader understands. | | `TestSetJobListViewNormalizesUnknownValue` | Verifies anything but `"compact"` is stored as `"detailed"`, so the config never gains a value no reader understands. |
| `TestPrependLogCapsActivityList` | Verifies that the activity log never grows beyond its maximum cap. | | `TestPrependLogCapsActivityList` | Verifies that the activity log never grows beyond its maximum cap. |
@@ -319,6 +321,7 @@ Tests command execution, exit code handling, output capture, and the run timeout
|------|---------| |------|---------|
| `TestRunJobStartOnlyDoesNotWaitForExitCode` | Verifies that `StartOnly: true` jobs launch and return "OK" immediately without waiting for the process to exit. | | `TestRunJobStartOnlyDoesNotWaitForExitCode` | Verifies that `StartOnly: true` jobs launch and return "OK" immediately without waiting for the process to exit. |
| `TestRunJobStartOnlyReportsStartFailure` | Verifies that `StartOnly: true` jobs still report "Failed" if the process cannot be started. | | `TestRunJobStartOnlyReportsStartFailure` | Verifies that `StartOnly: true` jobs still report "Failed" if the process cannot be started. |
| `TestRunJobStartOnlyLeavesNoContextWatcher` | Verifies that a start-only run leaves no `os/exec` context-watcher goroutine behind, since it never calls `Wait` and the started process is meant to outlive the app. |
--- ---
+60 -31
View File
@@ -43,15 +43,23 @@ func (s *Service) CreateJob(job domain.Job) (domain.Job, error) {
s.parseScheduleLocked(&job) s.parseScheduleLocked(&job)
record := uiRecord(job.ID, job.Name, "Created", "Job was added") record := uiRecord(job.ID, job.Name, "Created", "Job was added")
prependLog(runtime, record) prependLog(runtime, record)
err := s.store.SaveJobs(s.jobs) save := s.deferSaveLocked(s.store.PrepareSaveJobs(s.jobs))
if err != nil { s.mu.Unlock()
s.jobs = s.jobs[:len(s.jobs)-1]
if err := save(); err != nil {
// The write is atomic, so a failure left the file holding the previous
// list: take the job back out so memory matches what is on disk. Another
// operation may have run in between, so it is removed by ID rather than by
// truncating the slice.
s.mu.Lock()
if index := s.indexByIDLocked(job.ID); index >= 0 {
s.jobs = append(s.jobs[:index], s.jobs[index+1:]...)
}
delete(s.runtimes, job.ID) delete(s.runtimes, job.ID)
delete(s.schedules, job.ID) delete(s.schedules, job.ID)
s.mu.Unlock() s.mu.Unlock()
return domain.Job{}, err return domain.Job{}, err
} }
s.mu.Unlock()
s.emit(RunRecorded{Record: record}) s.emit(RunRecorded{Record: record})
s.emit(JobChanged{JobID: job.ID}) s.emit(JobChanged{JobID: job.ID})
return job, nil return job, nil
@@ -87,10 +95,10 @@ func (s *Service) UpdateJob(job domain.Job) error {
s.refreshNextRunLocked(existing, runtime) s.refreshNextRunLocked(existing, runtime)
record := uiRecord(job.ID, job.Name, "Updated", "Job settings changed") record := uiRecord(job.ID, job.Name, "Updated", "Job settings changed")
prependLog(runtime, record) prependLog(runtime, record)
err := s.store.SaveJobs(s.jobs) save := s.deferSaveLocked(s.store.PrepareSaveJobs(s.jobs))
s.mu.Unlock() s.mu.Unlock()
if err != nil { if err := save(); err != nil {
return err return err
} }
s.emit(RunRecorded{Record: record}) s.emit(RunRecorded{Record: record})
@@ -113,10 +121,10 @@ func (s *Service) DeleteJob(id int) error {
delete(s.runtimes, id) delete(s.runtimes, id)
delete(s.schedules, id) delete(s.schedules, id)
record := uiRecord(id, deleted.Name, "Deleted", "Job was removed") record := uiRecord(id, deleted.Name, "Deleted", "Job was removed")
err := s.store.SaveJobs(s.jobs) save := s.deferSaveLocked(s.store.PrepareSaveJobs(s.jobs))
s.mu.Unlock() s.mu.Unlock()
if err != nil { if err := save(); err != nil {
return err return err
} }
s.emit(RunRecorded{Record: record}) s.emit(RunRecorded{Record: record})
@@ -155,10 +163,10 @@ func (s *Service) SetEnabled(id int, enabled bool) error {
record = uiRecord(id, job.Name, "Paused", "Job was disabled") record = uiRecord(id, job.Name, "Paused", "Job was disabled")
} }
prependLog(runtime, record) prependLog(runtime, record)
err := s.store.SaveJobs(s.jobs) save := s.deferSaveLocked(s.store.PrepareSaveJobs(s.jobs))
s.mu.Unlock() s.mu.Unlock()
if err != nil { if err := save(); err != nil {
return err return err
} }
s.emit(RunRecorded{Record: record}) s.emit(RunRecorded{Record: record})
@@ -188,10 +196,10 @@ func (s *Service) SetGlobalPause(paused bool) error {
} }
s.refreshNextRunFromLocked(job, runtime, now) s.refreshNextRunFromLocked(job, runtime, now)
} }
err := s.store.SaveConfig() save := s.deferSaveLocked(s.store.PrepareSaveConfig())
s.mu.Unlock() s.mu.Unlock()
if err != nil { if err := save(); err != nil {
return err return err
} }
state, detail := "Resumed", "All job execution resumed" state, detail := "Resumed", "All job execution resumed"
@@ -219,9 +227,9 @@ func (s *Service) SetJobListView(view domain.JobListView) error {
return nil return nil
} }
s.store.Config.JobListView = view s.store.Config.JobListView = view
err := s.store.SaveConfig() save := s.deferSaveLocked(s.store.PrepareSaveConfig())
s.mu.Unlock() s.mu.Unlock()
return err return save()
} }
// ShouldNotifyOnFailure reports whether the user has enabled desktop // ShouldNotifyOnFailure reports whether the user has enabled desktop
@@ -251,54 +259,75 @@ func (s *Service) UpdateSettings(config domain.Config) error {
config.JobsFile = strings.TrimSpace(config.JobsFile) config.JobsFile = strings.TrimSpace(config.JobsFile)
s.mu.Lock() s.mu.Lock()
jobsPath := storage.ResolveConfiguredPath(s.store.Paths.AppDir, config.JobsFile) // AppDir is fixed for the process and only UpdateSettings itself — a UI
// action — can move JobsPath, so this snapshot stays valid across the reads
// below.
appDir := s.store.Paths.AppDir
jobsPath := storage.ResolveConfiguredPath(appDir, config.JobsFile)
switching := jobsPath != s.store.Paths.JobsPath switching := jobsPath != s.store.Paths.JobsPath
if switching && s.anyRunningLocked() { running := s.anyRunningLocked()
s.mu.Unlock() s.mu.Unlock()
if switching && running {
return errors.New("cannot change the jobs file while a job is running") return errors.New("cannot change the jobs file while a job is running")
} }
// Read the new file before anything is written, so a file that cannot be // Read the new file, and reconstruct its jobs' statistics from the logs the
// parsed leaves both the config and the current jobs untouched. // new config points at, before anything is written and while no lock is held:
// both are file I/O, and SeedStats opens every log in the directory. A file
// that cannot be parsed leaves both the config and the current jobs untouched.
var adopted []domain.Job var adopted []domain.Job
var seeds map[int]runner.SeededStats
if switching { if switching {
jobs, found, err := storage.LoadJobsFile(jobsPath) jobs, found, err := storage.LoadJobsFile(jobsPath)
if err != nil { if err != nil {
s.mu.Unlock()
return fmt.Errorf("read jobs file %s: %w", jobsPath, err) return fmt.Errorf("read jobs file %s: %w", jobsPath, err)
} }
if found { if found {
adopted = jobs adopted = jobs
seeds = runner.SeedStats(storage.ResolveConfiguredPath(appDir, config.LogsDir), jobs, config.MaxLogFiles)
} }
} }
s.store.Config = config s.mu.Lock()
if err := s.store.SaveConfig(); err != nil { // The guard above was evaluated before the reads, off the lock, so re-check
// it: a scheduled run may have started in the meantime, and adoption drops
// every runtime.
if switching && s.anyRunningLocked() {
s.mu.Unlock() s.mu.Unlock()
return err return errors.New("cannot change the jobs file while a job is running")
} }
s.store.Config = config
saveConfig := s.store.PrepareSaveConfig()
if adopted != nil { if adopted != nil {
s.adoptJobsLocked(adopted) s.adoptJobsLocked(adopted)
s.applySeededStatsLocked(seeds)
} }
// SaveConfig re-resolved the paths from the new config, so SaveJobs writes to // PrepareSaveConfig re-resolved the paths from the new config, so the jobs
// the (possibly new) jobs file and cleanup targets the new logs dir. Adopted // write targets the (possibly new) jobs file and cleanup targets the new logs
// jobs are written back too, which persists the IDs and defaults that // dir. Adopted jobs are written back too, which persists the IDs and defaults
// normalization filled in, exactly as loading them at startup would. // that normalization filled in, exactly as loading them at startup would. The
if err := s.store.SaveJobs(s.jobs); err != nil { // jobs write is skipped when the config write fails, because both writes run
s.mu.Unlock() // in the order prepared and stop at the first error.
return err save := s.deferSaveLocked(saveConfig, s.store.PrepareSaveJobs(s.jobs))
}
loaded := len(s.jobs) loaded := len(s.jobs)
logsDir := s.store.Paths.LogsDir logsDir := s.store.Paths.LogsDir
maxFiles := s.store.Config.MaxLogFiles maxFiles := s.store.Config.MaxLogFiles
maxAge := s.store.Config.MaxLogAgeDays maxAge := s.store.Config.MaxLogAgeDays
s.mu.Unlock() s.mu.Unlock()
saveErr := save()
if adopted != nil { if adopted != nil {
// A broad JobChanged redraws the job list; JobsLoaded tells the user in // A broad JobChanged redraws the job list; JobsLoaded tells the user in
// History which file those jobs came from, since nothing was asked. // History which file those jobs came from, since nothing was asked. Both
// are emitted even when the write failed: the adopted jobs are already the
// in-memory list, and a job list the user cannot see would be worse than
// the error they are about to be shown.
s.emit(JobsLoaded{Path: jobsPath, Count: loaded}) s.emit(JobsLoaded{Path: jobsPath, Count: loaded})
s.emit(JobChanged{}) s.emit(JobChanged{})
} }
if saveErr != nil {
return saveErr
}
return runner.CleanupLogs(logsDir, maxFiles, maxAge) return runner.CleanupLogs(logsDir, maxFiles, maxAge)
} }
+98
View File
@@ -3,8 +3,10 @@ package app
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"fmt"
"os" "os"
"path/filepath" "path/filepath"
"sync"
"sync/atomic" "sync/atomic"
"testing" "testing"
"time" "time"
@@ -725,6 +727,102 @@ func TestUpdateSettingsRefusesJobsFileSwitchWhileRunning(t *testing.T) {
waitRecord(t, done) waitRecord(t, done)
} }
// Adoption reconstructs the adopted jobs' aggregate statistics from the log
// files the new configuration points at. That scan opens every log in the
// directory, so UpdateSettings runs it before taking the state lock; this pins
// that its result still reaches the runtime map.
func TestUpdateSettingsSeedsAdoptedJobsFromLogs(t *testing.T) {
svc := newTempService(t, []domain.Job{{ID: 1, Name: "Local", Schedule: "@every 1m", Command: "echo local", Enabled: true}})
logsDir := svc.store.Paths.LogsDir
if err := os.MkdirAll(logsDir, 0o755); err != nil {
t.Fatal(err)
}
log := "time: 2026-08-05 10:00:00\njob_id: 7\njob_name: Adopted\ntrigger: Schedule\nstate: Failed\ndetail: boom\nduration: 1500\n\nstdout:\n<empty>\n"
if err := os.WriteFile(filepath.Join(logsDir, "20260805-100000_Adopted.log"), []byte(log), 0o644); err != nil {
t.Fatal(err)
}
shared := filepath.Join(svc.store.Paths.AppDir, "shared.json")
data, err := json.Marshal(domain.JobsFile{Jobs: []domain.Job{
{ID: 7, Name: "Adopted", Schedule: "@every 5m", Command: "echo adopted", Enabled: true},
}})
if err != nil {
t.Fatal(err)
}
if err := os.WriteFile(shared, data, 0o644); err != nil {
t.Fatal(err)
}
config := svc.store.Config
config.JobsFile = shared
if err := svc.UpdateSettings(config); err != nil {
t.Fatalf("UpdateSettings: %v", err)
}
runtime := svc.Runtime(7)
if runtime == nil {
t.Fatal("the adopted job has no runtime")
}
if runtime.RunCount != 1 || runtime.FailCount != 1 || runtime.LastDurationMS != 1500 {
t.Errorf("seeded stats: RunCount=%d FailCount=%d LastDurationMS=%d, want 1/1/1500",
runtime.RunCount, runtime.FailCount, runtime.LastDurationMS)
}
}
// Job saves run after mu is released, so one operation can be writing while
// another mutates state. deferSaveLocked takes its own lock while mu is still
// held, which is what keeps writes in mutation order: whatever changed the list
// last also wrote it last, so the file ends up matching memory instead of
// holding an older snapshot.
func TestConcurrentJobOperationsLeaveTheFileMatchingMemory(t *testing.T) {
svc := newTempService(t, nil)
const workers = 8
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
job, err := svc.CreateJob(domain.Job{Name: fmt.Sprintf("Job %d", i), Schedule: "@every 1m", Command: "echo hi", Enabled: true})
if err != nil {
t.Errorf("CreateJob %d: %v", i, err)
return
}
if err := svc.SetEnabled(job.ID, false); err != nil {
t.Errorf("SetEnabled %d: %v", job.ID, err)
}
}(i)
}
wg.Wait()
memory := svc.Jobs()
if len(memory) != workers {
t.Fatalf("jobs in memory = %d, want %d", len(memory), workers)
}
saved, found, err := storage.LoadJobsFile(svc.store.Paths.JobsPath)
if err != nil || !found {
t.Fatalf("read jobs file: found=%v err=%v", found, err)
}
if len(saved) != len(memory) {
t.Fatalf("jobs on disk = %d, want %d: the last write must be the last mutation", len(saved), len(memory))
}
onDisk := make(map[int]domain.Job, len(saved))
for _, job := range saved {
onDisk[job.ID] = job
}
for _, job := range memory {
got, ok := onDisk[job.ID]
if !ok {
t.Errorf("job %d (%q) is in memory but missing from the file", job.ID, job.Name)
continue
}
if got.Name != job.Name || got.Enabled != job.Enabled {
t.Errorf("job %d on disk = %q/%v, want %q/%v", job.ID, got.Name, got.Enabled, job.Name, job.Enabled)
}
}
}
func TestPrependLogCapsActivityList(t *testing.T) { func TestPrependLogCapsActivityList(t *testing.T) {
runtime := &domain.JobRuntime{} runtime := &domain.JobRuntime{}
for i := 0; i < maxJobLogs+10; i++ { for i := 0; i < maxJobLogs+10; i++ {
+6 -2
View File
@@ -147,7 +147,6 @@ func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger st
record, logErr := s.runJob(ctx, &jobCopy, trigger, env.logsDir, env.timeout) record, logErr := s.runJob(ctx, &jobCopy, trigger, env.logsDir, env.timeout)
s.mu.Lock() s.mu.Lock()
var cleanupErr error
var rerunStarted bool var rerunStarted bool
if current := s.findByIDLocked(jobCopy.ID); current != nil { if current := s.findByIDLocked(jobCopy.ID); current != nil {
runtime := s.runtimeForLocked(current) runtime := s.runtimeForLocked(current)
@@ -166,10 +165,15 @@ func (s *Service) executeRun(ctx context.Context, jobCopy domain.Job, trigger st
} else { } else {
s.refreshNextRunLocked(current, runtime) s.refreshNextRunLocked(current, runtime)
} }
cleanupErr = runner.CleanupLogs(env.logsDir, env.maxFiles, env.maxAge)
} }
s.mu.Unlock() s.mu.Unlock()
// Cleanup is a directory scan plus up to MaxLogFiles unlinks. It needs only
// the values already snapshotted into runEnv, so it runs after mu is released
// rather than making every UI refresh wait behind it. It runs even when the
// job is gone, because the run still wrote a log file that retention covers.
cleanupErr := runner.CleanupLogs(env.logsDir, env.maxFiles, env.maxAge)
if logErr != nil { if logErr != nil {
s.emit(ErrorOccurred{Err: fmt.Errorf("write run log for %q: %w", jobCopy.Name, logErr)}) s.emit(ErrorOccurred{Err: fmt.Errorf("write run log for %q: %w", jobCopy.Name, logErr)})
} }
+48 -8
View File
@@ -26,7 +26,10 @@ import (
// it; unexported helpers ending in "Locked" assume the caller already holds it. // it; unexported helpers ending in "Locked" assume the caller already holds it.
// The Service must never call back into the UI (or any code that might re-enter // The Service must never call back into the UI (or any code that might re-enter
// the Service) while holding mu — in particular emit() is always called after // the Service) while holding mu — in particular emit() is always called after
// mu is released. // mu is released. Blocking file I/O follows the same rule: mu is the lock the
// Fyne main thread takes on every Jobs() and Runtime() call, so a JSON write, a
// log-directory scan, or a pass over every log header must not happen inside it
// (see deferSaveLocked, executeRun, and applySeededStatsLocked).
type Service struct { type Service struct {
mu sync.Mutex mu sync.Mutex
store *storage.Store store *storage.Store
@@ -56,6 +59,13 @@ type Service struct {
// do not exercise autostart; Open() wires it via autostart.New(). // do not exercise autostart; Open() wires it via autostart.New().
manager autostart.Manager manager autostart.Manager
// saveMu serializes the store writes that operations prepare under mu and run
// after releasing it. It is taken while mu is still held and released once the
// write is done, so writes reach the file in the same order their snapshots
// were taken and an older snapshot can never land on top of a newer one.
// Nothing may take mu while holding saveMu.
saveMu sync.Mutex
// observers and their guard live in events.go. dispatchMu is separate from mu // observers and their guard live in events.go. dispatchMu is separate from mu
// so that emitting an event never requires (or is held under) the state lock: // so that emitting an event never requires (or is held under) the state lock:
// the Service must release mu before dispatching, per the locking contract. // the Service must release mu before dispatching, per the locking contract.
@@ -63,6 +73,26 @@ type Service struct {
observers []Observer observers []Observer
} }
// deferSaveLocked prepares the store writes for the caller to run after mu is
// released, and takes saveMu now so a later operation's write cannot overtake
// this one. The caller must hold mu, must unlock it before calling the returned
// function, and must call that function exactly once. Keeping the marshal, the
// fsync, and the rename out of the critical section is what stops a settings
// change or a job edit from blocking a scheduler tick or a finishing run. The
// writes run in the order given and stop at the first error.
func (s *Service) deferSaveLocked(writes ...func() error) func() error {
s.saveMu.Lock()
return func() error {
defer s.saveMu.Unlock()
for _, write := range writes {
if err := write(); err != nil {
return err
}
}
return nil
}
}
// NewService wires the Service to a loaded store and its jobs. It builds the // NewService wires the Service to a loaded store and its jobs. It builds the
// initial runtime map from the durable jobs so every job has transient state // initial runtime map from the durable jobs so every job has transient state
// from the moment the Service exists, and parses each job's schedule once. The // from the moment the Service exists, and parses each job's schedule once. The
@@ -77,18 +107,19 @@ func NewService(store *storage.Store, jobs []domain.Job) *Service {
// No lock is needed here: construction is single-threaded, before Start // No lock is needed here: construction is single-threaded, before Start
// launches the timing loop. // launches the timing loop.
s.adoptJobsLocked(jobs) s.adoptJobsLocked(jobs)
s.applySeededStatsLocked(runner.SeedStats(store.Paths.LogsDir, s.jobs, store.Config.MaxLogFiles))
return s return s
} }
// adoptJobsLocked makes jobs the Service's durable state and rebuilds everything // adoptJobsLocked makes jobs the Service's durable state and rebuilds everything
// derived from it: the runtime map, the parsed-schedule cache, each job's first // derived from it: the runtime map, the parsed-schedule cache, and each job's
// next-run — so the Service is ready to schedule the moment it exists, mirroring // first next-run — so the Service is ready to schedule the moment it exists,
// the old scheduler's reset-on-construction — and the statistics seeded from // mirroring the old scheduler's reset-on-construction.
// existing log files, so the details panel shows accumulated run history
// immediately rather than only runs since this process started.
// //
// It backs both construction and a Settings change that points at a different // It backs both construction and a Settings change that points at a different
// jobs file. The caller must hold mu. // jobs file. Statistics seeded from existing log files are applied separately by
// applySeededStatsLocked, because reconstructing them is file I/O. The caller
// must hold mu.
func (s *Service) adoptJobsLocked(jobs []domain.Job) { func (s *Service) adoptJobsLocked(jobs []domain.Job) {
s.jobs = jobs s.jobs = jobs
s.runtimes = domain.NewRuntimes(jobs) s.runtimes = domain.NewRuntimes(jobs)
@@ -100,7 +131,16 @@ func (s *Service) adoptJobsLocked(jobs []domain.Job) {
s.parseScheduleLocked(job) s.parseScheduleLocked(job)
s.refreshNextRunFromLocked(job, s.runtimes[job.ID], now) s.refreshNextRunFromLocked(job, s.runtimes[job.ID], now)
} }
for id, seed := range runner.SeedStats(s.store.Paths.LogsDir, s.jobs, s.store.Config.MaxLogFiles) { }
// applySeededStatsLocked folds statistics reconstructed from existing log files
// into the runtime map, so the details panel shows accumulated run history
// immediately rather than only runs since this process started. It is separate
// from adoptJobsLocked because producing the seeds opens every log file in the
// directory, which must not happen under mu: callers compute the map first and
// apply it here. The caller must hold mu.
func (s *Service) applySeededStatsLocked(seeds map[int]runner.SeededStats) {
for id, seed := range seeds {
runtime := s.runtimes[id] runtime := s.runtimes[id]
if runtime == nil { if runtime == nil {
continue continue
+9 -1
View File
@@ -36,7 +36,15 @@ func RunJob(ctx context.Context, job *domain.Job, trigger string, logsDir string
var detail string var detail string
var durationMS int64 var durationMS int64
if job.StartOnly { if job.StartOnly {
invocation := jobInvocation(ctx, *job) // A StartOnly process is deliberately never waited for, so it must not be
// tied to any cancelable context: exec.CommandContext leaves a watcher
// goroutine alive until Wait returns or the context is done, and since
// StartOnly never calls Wait that goroutine would live for the rest of the
// process — one per run — and then try to kill a process whose handle
// startJobOnly has already released. context.Background() has a nil Done
// channel, so os/exec starts no watcher at all and the started process is
// left to outlive GoSentry, which is the point of the option.
invocation := jobInvocation(context.Background(), *job)
// StartOnly jobs don't wait for process exit, so the duration measures // StartOnly jobs don't wait for process exit, so the duration measures
// launch latency (time to spawn the process) rather than run time. // launch latency (time to spawn the process) rather than run time.
state, detail, output, durationMS = startJobOnly(invocation, *job, started) state, detail, output, durationMS = startJobOnly(invocation, *job, started)
+52
View File
@@ -401,6 +401,58 @@ func TestRunJobZeroTimeoutMeansNoTimeout(t *testing.T) {
} }
} }
// A StartOnly run must not leave a watcher goroutine behind. exec.CommandContext
// keeps one alive until Wait returns or the context is done, and StartOnly never
// waits, so binding it to the caller's cancelable context would leak one
// goroutine per run for the lifetime of the app — and then, on shutdown, kill a
// process whose handle startJobOnly has already released.
func TestRunJobStartOnlyLeavesNoContextWatcher(t *testing.T) {
command := "sh"
arguments := "-c\nexit 0"
if runtime.GOOS == "windows" {
command = `C:\Windows\System32\cmd.exe`
arguments = "/C\nexit /b 0"
}
job := domain.Job{
ID: 53,
Name: "Start Only Goroutines",
Command: command,
Arguments: arguments,
StartOnly: true,
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
const runs = 5
before := settledGoroutines()
for i := 0; i < runs; i++ {
if _, err := RunJob(ctx, &job, "Manual", t.TempDir(), 30*time.Second); err != nil {
t.Fatal(err)
}
}
// Counted before cancel on purpose: a watcher would still be parked on
// ctx.Done() at this point, and cancelling first would release it.
if leaked := settledGoroutines() - before; leaked > 1 {
t.Errorf("%d goroutines left after %d StartOnly runs, want none tied to the run context", leaked, runs)
}
}
// settledGoroutines returns the goroutine count once it has stopped falling, so
// a goroutine that is still on its way out is not mistaken for a leak.
func settledGoroutines() int {
lowest := runtime.NumGoroutine()
for stable, i := 0, 0; stable < 3 && i < 100; i++ {
time.Sleep(10 * time.Millisecond)
if count := runtime.NumGoroutine(); count < lowest {
lowest, stable = count, 0
continue
}
stable++
}
return lowest
}
func TestRunJobStartOnlyIgnoresTimeout(t *testing.T) { func TestRunJobStartOnlyIgnoresTimeout(t *testing.T) {
command := "sh" command := "sh"
arguments := "-c\nsleep 5" arguments := "-c\nsleep 5"
+51 -55
View File
@@ -45,8 +45,8 @@ func SeedStats(logsDir string, jobs []domain.Job, maxFiles int) map[int]SeededSt
return result return result
} }
byID := make(map[int][]string) byID := make(map[int][]logSummary)
byName := make(map[string][]string) byName := make(map[string][]logSummary)
for _, entry := range entries { for _, entry := range entries {
if entry.IsDir() { if entry.IsDir() {
continue continue
@@ -55,9 +55,10 @@ func SeedStats(logsDir string, jobs []domain.Job, maxFiles int) map[int]SeededSt
if !strings.HasSuffix(strings.ToLower(name), ".log") { if !strings.HasSuffix(strings.ToLower(name), ".log") {
continue continue
} }
path := filepath.Join(logsDir, name) summary := readLogSummary(filepath.Join(logsDir, name))
if jobID, ok := readLogJobID(path); ok { summary.name = name
byID[jobID] = append(byID[jobID], name) if summary.hasJobID {
byID[summary.jobID] = append(byID[summary.jobID], summary)
continue continue
} }
base := name[:len(name)-len(".log")] base := name[:len(name)-len(".log")]
@@ -65,7 +66,7 @@ func SeedStats(logsDir string, jobs []domain.Job, maxFiles int) map[int]SeededSt
if idx < 0 { if idx < 0 {
continue continue
} }
byName[base[idx+1:]] = append(byName[base[idx+1:]], name) byName[base[idx+1:]] = append(byName[base[idx+1:]], summary)
} }
for _, job := range jobs { for _, job := range jobs {
@@ -76,37 +77,37 @@ func SeedStats(logsDir string, jobs []domain.Job, maxFiles int) map[int]SeededSt
if len(files) == 0 { if len(files) == 0 {
continue continue
} }
// The timestamp prefix sorts chronologically, so a lexical sort puts the // The timestamp prefix sorts chronologically, so a lexical sort by file
// oldest first; keep the newest maxFiles to honor the retention bound. // name puts the oldest first; keep the newest maxFiles to honor the
sort.Strings(files) // retention bound.
sort.Slice(files, func(i, j int) bool { return files[i].name < files[j].name })
if maxFiles > 0 && len(files) > maxFiles { if maxFiles > 0 && len(files) > maxFiles {
files = files[len(files)-maxFiles:] files = files[len(files)-maxFiles:]
} }
result[job.ID] = aggregateLogStats(logsDir, files) result[job.ID] = aggregateLogStats(files)
} }
return result return result
} }
// aggregateLogStats folds the header of each log file (oldest first) into one // aggregateLogStats folds the already-read header of each log file (oldest
// SeededStats. Files lacking a duration line contribute to the run/fail counts // first) into one SeededStats. Files lacking a duration line contribute to the
// but not to the duration aggregates. // run/fail counts but not to the duration aggregates.
func aggregateLogStats(logsDir string, files []string) SeededStats { func aggregateLogStats(files []logSummary) SeededStats {
var stats SeededStats var stats SeededStats
var durationSum int64 var durationSum int64
var durationCount int var durationCount int
for _, file := range files { for _, file := range files {
state, durationMS, hasDuration := readLogHeader(filepath.Join(logsDir, file))
stats.RunCount++ stats.RunCount++
if state == "Failed" { if file.state == "Failed" {
stats.FailCount++ stats.FailCount++
} }
if hasDuration { if file.hasDuration {
// Files are oldest first, so the last assignment is the newest run. // Files are oldest first, so the last assignment is the newest run.
stats.LastDurationMS = durationMS stats.LastDurationMS = file.durationMS
if durationMS > stats.MaxDurationMS { if file.durationMS > stats.MaxDurationMS {
stats.MaxDurationMS = durationMS stats.MaxDurationMS = file.durationMS
} }
durationSum += durationMS durationSum += file.durationMS
durationCount++ durationCount++
} }
} }
@@ -117,39 +118,29 @@ func aggregateLogStats(logsDir string, files []string) SeededStats {
return stats return stats
} }
// readLogJobID reads the job_id field from a log file header. // logSummary is everything SeedStats needs from one run log: the file name it
func readLogJobID(path string) (int, bool) { // sorts by, which job wrote it, how the run ended, and how long it took.
file, err := os.Open(path) type logSummary struct {
if err != nil { name string
return 0, false jobID int
} hasJobID bool
defer file.Close() state string
durationMS int64
scanner := bufio.NewScanner(file) hasDuration bool
for scanner.Scan() {
line := scanner.Text()
if line == "" {
break
}
if rest, ok := strings.CutPrefix(line, "job_id: "); ok {
id, err := strconv.Atoi(strings.TrimSpace(rest))
if err != nil {
return 0, false
}
return id, true
}
}
return 0, false
} }
// readLogHeader reads the "state" and "duration" fields from a log file's // readLogSummary reads the job_id, state, and duration fields from a log file's
// header (the lines before the first blank line). hasDuration reports whether a // header (the lines before the first blank line) in a single pass, so seeding
// well-formed duration line was present, distinguishing a legacy duration-less // opens each log once rather than once to find its job and again to read its
// log from one that genuinely recorded a zero-millisecond run. // result. The has* flags report whether a well-formed line was present,
func readLogHeader(path string) (state string, durationMS int64, hasDuration bool) { // distinguishing a legacy log written before the field existed from one that
// genuinely recorded a zero value. An unreadable file yields a zero summary,
// which falls back to matching by the job name in the file name.
func readLogSummary(path string) logSummary {
var summary logSummary
file, err := os.Open(path) file, err := os.Open(path)
if err != nil { if err != nil {
return "", 0, false return summary
} }
defer file.Close() defer file.Close()
@@ -159,14 +150,19 @@ func readLogHeader(path string) (state string, durationMS int64, hasDuration boo
if line == "" { if line == "" {
break // end of header break // end of header
} }
if rest, ok := strings.CutPrefix(line, "state: "); ok { if rest, ok := strings.CutPrefix(line, "job_id: "); ok {
state = strings.TrimSpace(rest) if id, err := strconv.Atoi(strings.TrimSpace(rest)); err == nil {
summary.jobID = id
summary.hasJobID = true
}
} else if rest, ok := strings.CutPrefix(line, "state: "); ok {
summary.state = strings.TrimSpace(rest)
} else if rest, ok := strings.CutPrefix(line, "duration: "); ok { } else if rest, ok := strings.CutPrefix(line, "duration: "); ok {
if value, err := strconv.ParseInt(strings.TrimSpace(rest), 10, 64); err == nil { if value, err := strconv.ParseInt(strings.TrimSpace(rest), 10, 64); err == nil {
durationMS = value summary.durationMS = value
hasDuration = true summary.hasDuration = true
} }
} }
} }
return state, durationMS, hasDuration return summary
} }
+35 -7
View File
@@ -65,19 +65,47 @@ func OpenStore() (*Store, []domain.Job, error) {
return store, jobs, nil return store, jobs, nil
} }
func (s *Store) SaveConfig() error { // PrepareSaveConfig re-resolves the derived paths from the current config and
// snapshots everything the write needs, returning the write itself as a closure.
// It exists so a caller that guards the Store with its own lock can do the file
// I/O — a marshal, an fsync, and a rename — after releasing that lock: the
// snapshot cannot change under the closure, so running it unlocked is safe.
// Prepared writes must be run in the order they were prepared, or an older
// snapshot can land on top of a newer one.
func (s *Store) PrepareSaveConfig() func() error {
s.applyConfigPaths() s.applyConfigPaths()
if err := os.MkdirAll(s.Paths.AppDir, 0o755); err != nil { dir := s.Paths.AppDir
path := s.Paths.ConfigPath
config := s.Config
return func() error {
if err := os.MkdirAll(dir, 0o755); err != nil {
return err return err
} }
return writeJSON(s.Paths.ConfigPath, s.Config) return writeJSON(path, config)
}
}
// PrepareSaveJobs is PrepareSaveConfig for the jobs file. The jobs slice is
// copied, so the caller may keep mutating its own slice as soon as this returns.
func (s *Store) PrepareSaveJobs(jobs []domain.Job) func() error {
dir := s.Paths.JobsDir
path := s.Paths.JobsPath
snapshot := make([]domain.Job, len(jobs))
copy(snapshot, jobs)
return func() error {
if err := os.MkdirAll(dir, 0o755); err != nil {
return err
}
return writeJSON(path, domain.JobsFile{Jobs: snapshot})
}
}
func (s *Store) SaveConfig() error {
return s.PrepareSaveConfig()()
} }
func (s *Store) SaveJobs(jobs []domain.Job) error { func (s *Store) SaveJobs(jobs []domain.Job) error {
if err := os.MkdirAll(s.Paths.JobsDir, 0o755); err != nil { return s.PrepareSaveJobs(jobs)()
return err
}
return writeJSON(s.Paths.JobsPath, domain.JobsFile{Jobs: jobs})
} }
func loadOrCreateConfig(paths Paths) (domain.Config, error) { func loadOrCreateConfig(paths Paths) (domain.Config, error) {