From 1fd8168416bc01bfc6e62facf82644659daf890e Mon Sep 17 00:00:00 2001 From: Hsukqi Lee Date: Sat, 14 Nov 2026 19:21:08 +0800 Subject: [PATCH 1/2] fix: deliver reviewed audit repairs --- internal/engine/engine.go | 21 +++ internal/engine/persistence_linux_test.go | 207 ++++++++++++++++++++++ internal/engine/runs.go | 4 + 3 files changed, 232 insertions(+) create mode 100644 internal/engine/persistence_linux_test.go diff --git a/internal/engine/engine.go b/internal/engine/engine.go index e3b73b8..a4cabb9 100644 --- a/internal/engine/engine.go +++ b/internal/engine/engine.go @@ -46,6 +46,8 @@ type Engine struct { hub *Hub st *store + // Lock order: persistMu -> mu. Never wait for a run while holding persistMu. + persistMu sync.Mutex mu sync.Mutex sessions map[string]*Session runs map[string]*Run @@ -214,6 +216,8 @@ func (e *Engine) persistLoop() { // Flush writes dirty sessions and runs to disk. func (e *Engine) Flush() { + e.persistMu.Lock() + defer e.persistMu.Unlock() type rec struct { kind, id string data []byte @@ -225,6 +229,9 @@ func (e *Engine) Flush() { b, err := json.Marshal(r) if err == nil { recs = append(recs, rec{"runs", id, b}) + } else { + slog.Error("encode failed", "kind", "runs", "id", id, "err", err) + continue } } delete(e.dirtyRuns, id) @@ -234,6 +241,9 @@ func (e *Engine) Flush() { b, err := json.Marshal(s) if err == nil { recs = append(recs, rec{"sessions", id, b}) + } else { + slog.Error("encode failed", "kind", "sessions", "id", id, "err", err) + continue } } delete(e.dirtySessions, id) @@ -242,7 +252,16 @@ func (e *Engine) Flush() { for _, r := range recs { if err := e.st.saveRaw(r.kind, r.id, r.data); err != nil { slog.Error("persist failed", "kind", r.kind, "id", r.id, "err", err) + e.mu.Lock() + if r.kind == "runs" && e.runs[r.id] != nil { + e.dirtyRuns[r.id] = true + } + if r.kind == "sessions" && e.sessions[r.id] != nil { + e.dirtySessions[r.id] = true + } + e.mu.Unlock() } + // Success acknowledges only this snapshot. Updates during I/O stay dirty. } } @@ -440,6 +459,8 @@ func (e *Engine) DeleteSession(id string) error { case <-time.After(20 * time.Second): } } + e.persistMu.Lock() + defer e.persistMu.Unlock() e.mu.Lock() delete(e.sessions, id) delete(e.dirtySessions, id) diff --git a/internal/engine/persistence_linux_test.go b/internal/engine/persistence_linux_test.go new file mode 100644 index 0000000..f4dd9e1 --- /dev/null +++ b/internal/engine/persistence_linux_test.go @@ -0,0 +1,207 @@ +package engine + +import ( + "encoding/json" + "io" + "os" + "path/filepath" + "strings" + "syscall" + "testing" + "time" +) + +// These scenarios use the production loader and filesystem, including a FIFO +// to hold an actual WriteFile in flight without adding a storage test hook. +func persistenceEngine(t *testing.T) (*Engine, string, string) { + t.Helper() + dir := t.TempDir() + if err := os.MkdirAll(filepath.Join(dir, "runs"), 0755); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(filepath.Join(dir, "sessions"), 0755); err != nil { + t.Fatal(err) + } + for path, data := range map[string]string{ + "runs/r.json": `{"id":"r","sessionId":"s","status":"running","steps":[]}`, + "sessions/s.json": `{"id":"s","title":"before","runIds":["r"]}`, + } { + if err := os.WriteFile(filepath.Join(dir, path), []byte(data), 0644); err != nil { + t.Fatal(err) + } + } + e, err := New(Config{DataDir: dir, Workspace: t.TempDir()}) + if err != nil { + t.Fatal(err) + } + return e, filepath.Join(dir, "sessions/s.json"), filepath.Join(dir, "runs/r.json") +} + +func TestFlushRetriesSessionAndRun(t *testing.T) { + e, sessionPath, runPath := persistenceEngine(t) + title := "after" + if _, err := e.UpdateSession("s", &title, nil, nil); err != nil { + t.Fatal(err) + } + for _, path := range []string{sessionPath, runPath} { + if err := os.Mkdir(path+".tmp", 0755); err != nil { + t.Fatal(err) + } + } + e.Flush() + for _, path := range []string{sessionPath, runPath} { + if err := os.Remove(path + ".tmp"); err != nil { + t.Fatal(err) + } + } + e.Flush() // no business mutation after recovery + reloaded, err := New(e.Config()) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(reloaded.ListSessions()), `"title":"after"`) { + t.Error("failed session write was not retried") + } + data, err := os.ReadFile(runPath) + if err != nil { + t.Fatal(err) + } + var run Run + if err := json.Unmarshal(data, &run); err != nil { + t.Fatal(err) + } + if run.Status != StatusInterrupted { + t.Errorf("failed run write was not retried: %s", run.Status) + } +} + +func TestFlushOrdersUpdatesAndDeletion(t *testing.T) { + for _, operation := range []string{"flush", "session", "run", "cleanup"} { + t.Run(operation, func(t *testing.T) { + e, path, runPath := persistenceEngine(t) + e.Flush() + title := strings.Repeat("x", 2<<20) // larger than the FIFO buffer + if _, err := e.UpdateSession("s", &title, nil, nil); err != nil { + t.Fatal(err) + } + if err := syscall.Mkfifo(path+".tmp", 0600); err != nil { + t.Fatal(err) + } + first := make(chan struct{}) + go func() { e.Flush(); close(first) }() + reader, err := os.Open(path + ".tmp") + if err != nil { + t.Fatal(err) + } + defer reader.Close() + if _, err := io.ReadFull(reader, make([]byte, 1)); err != nil { + t.Fatal(err) + } + // WriteFile has the FIFO open and is blocked on its full buffer. + // Replace only the directory entry so its later rename remains real. + if err := os.Remove(path + ".tmp"); err != nil { + t.Fatal(err) + } + old, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path+".tmp", old, 0644); err != nil { + t.Fatal(err) + } + title = "latest" + if _, err := e.UpdateSession("s", &title, nil, nil); err != nil { + t.Fatal(err) + } + second := make(chan error, 1) + go func() { + switch operation { + case "flush": + e.Flush() + case "session": + second <- e.DeleteSession("s") + return + case "run": + second <- e.DeleteRun("r") + return + case "cleanup": + if n := e.CleanupRuns("", 0, false); n != 1 { + second <- io.ErrUnexpectedEOF + return + } + } + second <- nil + }() + var early bool + select { + case err := <-second: + early = true + if err != nil { + t.Error(err) + } + t.Error("commit overtook an in-flight Flush") + case <-time.After(100 * time.Millisecond): + } + if _, err := io.Copy(io.Discard, reader); err != nil { + t.Fatal(err) + } + <-first + if !early { + if err := <-second; err != nil { + t.Fatal(err) + } + } + e.Flush() + reloaded, err := New(e.Config()) + if err != nil { + t.Fatal(err) + } + if operation == "flush" { + if !strings.Contains(string(reloaded.ListSessions()), `"title":"latest"`) { + t.Error("update during I/O lost on reload") + } + } else { + for _, file := range []string{path, runPath} { + if _, err := os.Stat(file); !os.IsNotExist(err) { + t.Errorf("deleted file resurrected: %s (%v)", file, err) + } + } + if string(reloaded.ListSessions()) != "[]" || string(reloaded.ListRuns(RunFilter{})) != "[]" { + t.Error("deleted records returned on reload") + } + } + }) + } +} + +func TestDeleteSessionWaitsForActiveRun(t *testing.T) { + e := newTestEngine(t) + s, err := e.CreateSession("active", "", "") + if err != nil { + t.Fatal(err) + } + raw, err := e.StartChat(ChatRequest{SessionID: s.ID, Text: "SLOW"}) + if err != nil { + t.Fatal(err) + } + id := runIDOf(t, raw) + waitFor(t, "active turn", func() bool { return e.snapshot(id).Steps[0].TurnID != "" }) + done := make(chan error, 1) + go func() { done <- e.DeleteSession(s.ID) }() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("deletion and finishing Flush deadlocked") + } + e.Flush() + reloaded, err := New(e.Config()) + if err != nil { + t.Fatal(err) + } + if string(reloaded.ListSessions()) != "[]" || string(reloaded.ListRuns(RunFilter{})) != "[]" { + t.Fatal("active run survived session deletion") + } +} diff --git a/internal/engine/runs.go b/internal/engine/runs.go index e198c0b..c92ba0b 100644 --- a/internal/engine/runs.go +++ b/internal/engine/runs.go @@ -40,6 +40,8 @@ func (e *Engine) RunStats(nodeID string) RunStats { // the manager's answers about it and its logs. A session // left without runs goes too. func (e *Engine) DeleteRun(id string) error { + e.persistMu.Lock() + defer e.persistMu.Unlock() e.mu.Lock() r := e.runs[id] switch { @@ -60,6 +62,8 @@ func (e *Engine) DeleteRun(id string) error { // nodeID is empty) that ended more than olderThan ago; 0 means every // finished run. With dryRun it only counts them. func (e *Engine) CleanupRuns(nodeID string, olderThan time.Duration, dryRun bool) int { + e.persistMu.Lock() + defer e.persistMu.Unlock() cutoff := time.Now().Add(-olderThan) e.mu.Lock() var pick []*Run From b7dea06379eb1a7e8a6b9d3780b29a17156845c5 Mon Sep 17 00:00:00 2001 From: Hsukqi Lee Date: Sat, 14 Nov 2026 22:58:35 +0800 Subject: [PATCH 2/2] test: wait for thread creation before persistence snapshot --- internal/engine/engine_test.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/internal/engine/engine_test.go b/internal/engine/engine_test.go index 0123978..c7b1bc5 100644 --- a/internal/engine/engine_test.go +++ b/internal/engine/engine_test.go @@ -492,7 +492,10 @@ func TestPersistenceMarksInterruptedRuns(t *testing.T) { s, _ := e.CreateSession("", "", "") raw, _ := e.StartChat(ChatRequest{SessionID: s.ID, Text: "SLOW"}) id := runIDOf(t, raw) - waitFor(t, "running", func() bool { return e.snapshot(id).Status == StatusRunning }) + waitFor(t, "thread started", func() bool { + r := e.snapshot(id) + return r.Status == StatusRunning && r.Steps[0].ThreadID != "" + }) e.Flush() e2, err := New(e.cfg) if err != nil {