From ec1be609ea19e06d98776c326c46fe3565d2a4d6 Mon Sep 17 00:00:00 2001 From: Sebastien Tardif Date: Wed, 2 Sep 2026 13:25:36 -0700 Subject: [PATCH 1/2] fix: write run history as one complete JSONL line Encode each RunRecord into a buffer, write that one complete JSONL line, and return Close errors. Drop a leftover incomplete last line on read and trim it before the next append so crawlctl run, status, and logs stay usable after a short write. Complete corrupt lines still fail closed. Signed-off-by: Sebastien Tardif --- scheduler/run.go | 63 +++++++++++++++++++++++--- scheduler/scheduler_test.go | 90 +++++++++++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+), 7 deletions(-) diff --git a/scheduler/run.go b/scheduler/run.go index 1b06f0f..fbbdf49 100644 --- a/scheduler/run.go +++ b/scheduler/run.go @@ -2,6 +2,7 @@ package scheduler import ( "bufio" + "bytes" "context" "crypto/rand" "encoding/hex" @@ -279,26 +280,74 @@ func appendHistory(path string, record RunRecord) error { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return err } - file, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) + file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0o600) if err != nil { return err } - defer file.Close() - enc := json.NewEncoder(file) - return enc.Encode(record) + if err := trimIncompleteHistoryTail(file); err != nil { + _ = file.Close() + return err + } + info, err := file.Stat() + if err != nil { + _ = file.Close() + return err + } + start := info.Size() + var buf bytes.Buffer + if err := json.NewEncoder(&buf).Encode(record); err != nil { + _ = file.Close() + return err + } + if _, err := file.Write(buf.Bytes()); err != nil { + _ = file.Truncate(start) + _ = file.Close() + return err + } + return file.Close() +} + +func trimIncompleteHistoryTail(file *os.File) error { + info, err := file.Stat() + if err != nil { + return err + } + size := info.Size() + if size == 0 { + return nil + } + data := make([]byte, size) + if _, err := file.ReadAt(data, 0); err != nil { + return err + } + keep := completeHistoryPrefix(data) + if int64(len(keep)) == size { + return nil + } + return file.Truncate(int64(len(keep))) +} + +func completeHistoryPrefix(data []byte) []byte { + if len(data) == 0 || data[len(data)-1] == '\n' { + return data + } + if n := bytes.LastIndexByte(data, '\n'); n >= 0 { + return data[:n+1] + } + return nil } func ReadHistory(path string) ([]RunRecord, error) { - file, err := os.Open(path) + data, err := os.ReadFile(path) if err != nil { if errors.Is(err, os.ErrNotExist) { return nil, nil } return nil, err } - defer file.Close() + data = completeHistoryPrefix(data) var records []RunRecord - scanner := bufio.NewScanner(file) + scanner := bufio.NewScanner(bytes.NewReader(data)) for scanner.Scan() { var record RunRecord if err := json.Unmarshal(scanner.Bytes(), &record); err != nil { diff --git a/scheduler/scheduler_test.go b/scheduler/scheduler_test.go index 6cacc98..97b7181 100644 --- a/scheduler/scheduler_test.go +++ b/scheduler/scheduler_test.go @@ -1,6 +1,7 @@ package scheduler import ( + "bytes" "context" "os" "path/filepath" @@ -255,3 +256,92 @@ func TestDefaultPathsCustomConfigKeepsStateNearby(t *testing.T) { t.Fatalf("history = %s, want state next to config", paths.History) } } + +func TestReadHistoryIgnoresTruncatedLastLine(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "runs.jsonl") + complete := `{"id":"1","job":"ok","command":["true"],"started_at":"2026-01-01T00:00:00Z","finished_at":"2026-01-01T00:00:01Z","duration_ms":1,"exit_code":0,"status":"success","log_path":"ok.log"}` + "\n" + if err := os.WriteFile(path, []byte(complete+`{"id":"2","job":"ok"`), 0o600); err != nil { + t.Fatal(err) + } + history, err := ReadHistory(path) + if err != nil { + t.Fatalf("history: %v", err) + } + if len(history) != 1 || history[0].ID != "1" { + t.Fatalf("history = %#v", history) + } +} + +func TestRunDoesNotRefuseOnTruncatedHistoryLine(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("shell command path differs on windows") + } + dir := t.TempDir() + paths := Paths{LogDir: filepath.Join(dir, "logs"), StateDir: filepath.Join(dir, "state"), LockPath: filepath.Join(dir, "state", "lock"), History: filepath.Join(dir, "state", "runs.jsonl")} + if err := os.MkdirAll(paths.StateDir, 0o755); err != nil { + t.Fatal(err) + } + complete := `{"id":"1","job":"ok","command":["true"],"started_at":"2026-01-01T00:00:00Z","finished_at":"2026-01-01T00:00:01Z","duration_ms":1,"exit_code":0,"status":"success","log_path":"ok.log"}` + "\n" + if err := os.WriteFile(paths.History, []byte(complete+`{"id":"2","job":"ok"`), 0o600); err != nil { + t.Fatal(err) + } + cfg := DefaultConfig() + cfg.Jobs["ok"] = Job{Enabled: true, Command: []string{"sh", "-c", "echo ok"}} + records, err := Run(context.Background(), RunOptions{Config: cfg, Paths: paths, Names: []string{"ok"}}) + if err != nil { + t.Fatalf("run: %v", err) + } + if len(records) != 1 || records[0].Status != "success" { + t.Fatalf("records = %#v", records) + } + history, err := ReadHistory(paths.History) + if err != nil { + t.Fatalf("history: %v", err) + } + if len(history) != 2 || history[0].ID != "1" || history[1].ID == "" || history[1].ID == "1" { + t.Fatalf("history = %#v", history) + } + data, err := os.ReadFile(paths.History) + if err != nil { + t.Fatal(err) + } + if len(data) == 0 || data[len(data)-1] != '\n' { + t.Fatalf("history file = %q, want complete JSONL", data) + } +} + +func TestAppendHistoryWritesCompleteJSONLLine(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "runs.jsonl") + record := RunRecord{ + ID: "rec1", + Job: "ok", + Command: []string{"echo", "ok"}, + Status: "success", + StartedAt: "2026-08-29T00:00:00Z", + FinishedAt: "2026-08-29T00:00:01Z", + DurationMs: 1000, + LogPath: "/tmp/ok.log", + } + if err := appendHistory(path, record); err != nil { + t.Fatalf("appendHistory: %v", err) + } + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + if len(data) == 0 || data[len(data)-1] != '\n' { + t.Fatalf("history = %q, want one newline-terminated JSONL line", data) + } + if bytes.Count(data, []byte{'\n'}) != 1 { + t.Fatalf("history = %q, want exactly one line", data) + } + history, err := ReadHistory(path) + if err != nil { + t.Fatalf("ReadHistory: %v", err) + } + if len(history) != 1 || history[0].ID != record.ID || history[0].Job != record.Job { + t.Fatalf("history = %#v", history) + } +} From 18a5527ec136ca12e8263b4c432470f4f193e0f7 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 4 Sep 2026 02:11:59 -0700 Subject: [PATCH 2/2] fix(scheduler): allow history truncation on Windows Go removes FILE_WRITE_DATA when opening O_APPEND handles on Windows, so recovery and failed-write rollback cannot truncate them. Open a read/write handle and seek to the validated append position under the scheduler's existing writer lock. The new cross-platform truncation and rollback regressions exposed the issue in Windows CI. Full local checks and real CLI file-size-limit recovery proof pass with the corrected handle mode. Co-authored-by: Sebastien Tardif --- scheduler/run.go | 8 +++++++- scheduler/scheduler_test.go | 4 ++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/scheduler/run.go b/scheduler/run.go index d0abf9c..cdf5bac 100644 --- a/scheduler/run.go +++ b/scheduler/run.go @@ -280,7 +280,9 @@ func appendHistory(path string, record RunRecord) error { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return err } - file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0o600) + // The scheduler lock serializes writes. Avoid O_APPEND: Windows append + // handles lack the write-data access required for recovery truncation. + file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) if err != nil { return err } @@ -290,6 +292,7 @@ func appendHistory(path string, record RunRecord) error { type historyFile interface { io.Writer io.ReaderAt + io.Seeker Stat() (os.FileInfo, error) Truncate(int64) error Close() error @@ -327,6 +330,9 @@ func appendHistoryFile(file historyFile, record RunRecord) (err error) { if err := json.NewEncoder(&buf).Encode(record); err != nil { return err } + if _, err := file.Seek(start, io.SeekStart); err != nil { + return err + } n, err := file.Write(buf.Bytes()) if err == nil && n != buf.Len() { err = io.ErrShortWrite diff --git a/scheduler/scheduler_test.go b/scheduler/scheduler_test.go index 53ea1b5..3198335 100644 --- a/scheduler/scheduler_test.go +++ b/scheduler/scheduler_test.go @@ -475,7 +475,7 @@ func TestHistoryWriteFailurePreservesPriorRecords(t *testing.T) { if err := os.WriteFile(path, original, 0o600); err != nil { t.Fatal(err) } - file, err := os.OpenFile(path, os.O_RDWR|os.O_APPEND, 0o600) + file, err := os.OpenFile(path, os.O_RDWR, 0o600) if err != nil { t.Fatal(err) } @@ -511,7 +511,7 @@ func TestHistoryWriteFailurePreservesPriorRecords(t *testing.T) { func TestHistoryReturnsCloseFailure(t *testing.T) { path := filepath.Join(t.TempDir(), "runs.jsonl") - file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0o600) + file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) if err != nil { t.Fatal(err) }