From 33621c7949a73fa40a860ab2b7dd2ded3e2b80f3 Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Sun, 6 Sep 2026 21:21:40 +0800 Subject: [PATCH 1/2] test(store): reproduce FileStore persistence starvation Reproduce #2345 with real snapshot recovery before final Flush/Close and virtual-time continuous writes. The existing FileStore suite passes on the unchanged production base; this focused regression fails with 41 memory items and no durable items. Co-authored-by: Codex --- .../store/file_store_persist_loop_test.go | 52 +++++++++++++++++++ 1 file changed, 52 insertions(+) create mode 100644 edge-server/internal/store/file_store_persist_loop_test.go diff --git a/edge-server/internal/store/file_store_persist_loop_test.go b/edge-server/internal/store/file_store_persist_loop_test.go new file mode 100644 index 000000000..f4e63afa3 --- /dev/null +++ b/edge-server/internal/store/file_store_persist_loop_test.go @@ -0,0 +1,52 @@ +package store + +import ( + "fmt" + "path/filepath" + "testing" + "testing/synctest" + "time" +) + +func TestFileStorePersistsDuringContinuousWrites(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "store.json") + s, err := NewFile(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(s.Close) + if _, err := s.CreateProject("project", "Project", ""); err != nil { + t.Fatal(err) + } + if _, err := s.CreateThread("thread", "project", "Thread", "", "", ""); err != nil { + t.Fatal(err) + } + if _, err := s.CreateItem(Item{ID: "first-message", ProjectID: "project", ThreadID: "thread", Type: "user_message", Content: "must become durable while writes continue"}); err != nil { + t.Fatal(err) + } + synctest.Wait() + + // Drive the real constructor-owned loop with virtual time. A fresh + // write before each quiet-period deadline must not starve persistence. + for i := range 40 { + <-time.After(debounceInterval / 2) + if _, err := s.CreateItem(Item{ID: fmt.Sprintf("busy-%d", i), ProjectID: "project", ThreadID: "thread", Type: "event"}); err != nil { + t.Fatal(err) + } + synctest.Wait() + } + if err := s.LastPersistError(); err != nil { + t.Fatal(err) + } + // Exercise the actual disk restore path without Flush/Close repairing + // the snapshot. This is a scheduling regression, not a load benchmark. + restored := New() + if err := loadFileSnapshot(path, restored); err != nil { + t.Fatal(err) + } + if _, ok := restored.GetItem("first-message"); !ok { + t.Fatalf("continuous writes starved FileStore persistence for 20 batch intervals: memory items=%d, durable items=%d", len(s.ListThreadItems("thread")), len(restored.ListThreadItems("thread"))) + } + }) +} From 10f40b20205633838d050b70c599bea7bf1a5a8e Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Sun, 6 Sep 2026 21:37:15 +0800 Subject: [PATCH 2/2] fix(store): bound FileStore batching to the first pending write Keep the existing 50ms asynchronous coalescing window without resetting it for every mutation, so continuous writes cannot starve durable snapshots. Preserve idle, error and close semantics, with real-file virtual-time regressions and an aligned architecture description. Co-authored-by: Codex --- docs/architecture/02-edge-server.md | 4 +- edge-server/internal/store/file_store.go | 13 +- .../store/file_store_persist_loop_test.go | 160 +++++++++++++++--- 3 files changed, 149 insertions(+), 28 deletions(-) diff --git a/docs/architecture/02-edge-server.md b/docs/architecture/02-edge-server.md index 90aec5d9b..c21a7ebe6 100644 --- a/docs/architecture/02-edge-server.md +++ b/docs/architecture/02-edge-server.md @@ -64,7 +64,9 @@ ProcessExecutor 配置 `RunTimeout`（默认 30 分钟）、`ShutdownGracePeriod `Store`（`internal/store/store.go`）是核心内存数据结构，管理 Project、Thread、Run、Item、Pin、Diff、Artifact、Preview、UserProfile、AgentProfile、Settings 的全量 CRUD。`FileStore` 与 `SQLiteStore` 的常规 `Get*` / `List*` 委托这个内存 Store，不是直接 SQL 查询；业务读取主要经过内存锁，而不是 SQLite 连接池。 -`FileStore`（`internal/store/file_store.go`）对写入信号做 debounce，再保存全量 JSON 快照；显式 `Flush` 同步执行快照、编码和文件 Sync/rename。`SQLiteStore`（`internal/store/sqlite_store.go`）在写入后同步持久化快照差分到 SQLite（WAL 模式，定期 checkpoint），支持崩溃恢复。SQL 连接初始化与持久化串行化独立于普通业务读面，不能用额外 SQL 读者的争用直接代替业务读取测量。 +`FileStore`（`internal/store/file_store.go`）由首条待保存写入启动 50 ms 合批窗口，后续写入只合批、不推迟已有截止点；窗口到期后保存全量 JSON 快照，空闲时不轮询写盘。50 ms 是调度窗口，不是包含编码与文件 I/O 的完成时限；显式 `Flush` 同步执行快照、编码和文件 Sync/rename。 + +`SQLiteStore`（`internal/store/sqlite_store.go`）在写入后同步持久化快照差分到 SQLite（WAL 模式，定期 checkpoint），支持崩溃恢复。SQL 连接初始化与持久化串行化独立于普通业务读面，不能用额外 SQL 读者的争用直接代替业务读取测量。 终端状态 runs（finished/failed/cancelled/completed_with_issues）按 `TerminalTTL` 超时或 `MaxTerminalRunsPerThread` 上限自动清理，级联删除关联 diffs/artifacts/previews/items/checkpoints。Checkpoint 在 run 完成时保留，在 run 清理或所属 thread 删除时随 run 移除；这不删除工作区文件。 SQLite 后台清理有删除时同步提交；失败通过 `LastPersistError` 和日志留痕，下一周期即使没有新删除也会重试。关闭时先停止并等待后台清理/checkpoint 任务，再做最终持久化与数据库关闭。 diff --git a/edge-server/internal/store/file_store.go b/edge-server/internal/store/file_store.go index 80282bc74..983a45d6c 100644 --- a/edge-server/internal/store/file_store.go +++ b/edge-server/internal/store/file_store.go @@ -48,7 +48,8 @@ type fileSnapshot struct { } // FileStore wraps the in-memory store with a JSON snapshot saved asynchronously after writes. -// Writes are debounced — rapid mutations batch into a single disk write. +// Writes within a fixed window batch into one disk write; continuous traffic +// cannot postpone persistence by repeatedly restarting that window. type FileStore struct { path string @@ -144,7 +145,7 @@ func (f *FileStore) schedulePersist() { } } -// persistLoop runs in the background, debouncing persist calls. +// persistLoop coalesces writes until the first pending signal's deadline. func (f *FileStore) persistLoop() { defer close(f.done) @@ -152,6 +153,8 @@ func (f *FileStore) persistLoop() { if !timer.Stop() { <-timer.C } + defer timer.Stop() + pending := false for { select { @@ -161,8 +164,12 @@ func (f *FileStore) persistLoop() { _ = f.syncPersist() return } - timer.Reset(debounceInterval) + if !pending { + timer.Reset(debounceInterval) + pending = true + } case <-timer.C: + pending = false // Failure is surfaced via LastPersistError, matching Flush() semantics. _ = f.syncPersist() } diff --git a/edge-server/internal/store/file_store_persist_loop_test.go b/edge-server/internal/store/file_store_persist_loop_test.go index f4e63afa3..42312cd3d 100644 --- a/edge-server/internal/store/file_store_persist_loop_test.go +++ b/edge-server/internal/store/file_store_persist_loop_test.go @@ -2,6 +2,7 @@ package store import ( "fmt" + "os" "path/filepath" "testing" "testing/synctest" @@ -10,43 +11,154 @@ import ( func TestFileStorePersistsDuringContinuousWrites(t *testing.T) { synctest.Test(t, func(t *testing.T) { - path := filepath.Join(t.TempDir(), "store.json") - s, err := NewFile(path) - if err != nil { - t.Fatal(err) - } - t.Cleanup(s.Close) - if _, err := s.CreateProject("project", "Project", ""); err != nil { - t.Fatal(err) - } - if _, err := s.CreateThread("thread", "project", "Thread", "", "", ""); err != nil { - t.Fatal(err) - } - if _, err := s.CreateItem(Item{ID: "first-message", ProjectID: "project", ThreadID: "thread", Type: "user_message", Content: "must become durable while writes continue"}); err != nil { - t.Fatal(err) - } + s, path := newFilePersistLoopFixture(t) + addFilePersistLoopItem(t, s, "first-message") synctest.Wait() // Drive the real constructor-owned loop with virtual time. A fresh // write before each quiet-period deadline must not starve persistence. for i := range 40 { <-time.After(debounceInterval / 2) - if _, err := s.CreateItem(Item{ID: fmt.Sprintf("busy-%d", i), ProjectID: "project", ThreadID: "thread", Type: "event"}); err != nil { - t.Fatal(err) - } + addFilePersistLoopItem(t, s, fmt.Sprintf("busy-%d", i)) synctest.Wait() } if err := s.LastPersistError(); err != nil { t.Fatal(err) } - // Exercise the actual disk restore path without Flush/Close repairing - // the snapshot. This is a scheduling regression, not a load benchmark. - restored := New() - if err := loadFileSnapshot(path, restored); err != nil { + // Use real disk recovery before Flush/Close can repair the snapshot. + // A late item requires repeated progress, not only the first flush. + restored := readFilePersistLoopDisk(t, path) + for _, id := range []string{"first-message", "busy-36"} { + if _, ok := restored.GetItem(id); !ok { + t.Errorf("continuous writes starved FileStore persistence for 20 batch intervals: missing %s, memory items=%d, durable items=%d", id, len(s.ListThreadItems("thread")), len(restored.ListThreadItems("thread"))) + } + } + }) +} + +func TestFileStoreCoalescesWithoutPostponingBatchDeadline(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, path := newFilePersistLoopFixture(t) + addFilePersistLoopItem(t, s, "first-message") + synctest.Wait() + <-time.After(debounceInterval / 2) + addFilePersistLoopItem(t, s, "second-message") + synctest.Wait() + if got := len(readFilePersistLoopDisk(t, path).ListThreadItems("thread")); got != 0 { + t.Fatalf("batch persisted %d items before its first-write deadline", got) + } + <-time.After(debounceInterval / 2) + synctest.Wait() + restored := readFilePersistLoopDisk(t, path) + for _, id := range []string{"first-message", "second-message"} { + if _, ok := restored.GetItem(id); !ok { + t.Errorf("later write postponed the first batch: missing %s", id) + } + } + + addFilePersistLoopItem(t, s, "next-batch") + synctest.Wait() + <-time.After(debounceInterval / 2) + synctest.Wait() + if _, ok := readFilePersistLoopDisk(t, path).GetItem("next-batch"); ok { + t.Error("next batch was not coalesced") + } + <-time.After(debounceInterval / 2) + synctest.Wait() + if _, ok := readFilePersistLoopDisk(t, path).GetItem("next-batch"); !ok { + t.Error("next batch did not re-arm persistence") + } + }) +} + +func TestFileStoreIdleDoesNotPersistAndNewBatchReportsFailure(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, path := newFilePersistLoopFixture(t) + addFilePersistLoopItem(t, s, "first-message") + synctest.Wait() + <-time.After(debounceInterval) + synctest.Wait() + if _, ok := readFilePersistLoopDisk(t, path).GetItem("first-message"); !ok { + t.Fatal("initial batch did not persist") + } + // An actual filesystem failure makes an unwanted idle persist visible. + // The next real write below checks that the fault is effective. + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + if err := os.Mkdir(path, 0o755); err != nil { + t.Fatal(err) + } + <-time.After(4 * debounceInterval) + synctest.Wait() + if err := s.LastPersistError(); err != nil { + t.Errorf("idle store attempted persistence without a write: %v", err) + } + addFilePersistLoopItem(t, s, "after-idle") + synctest.Wait() + <-time.After(debounceInterval) + synctest.Wait() + if err := s.LastPersistError(); err == nil { + t.Error("new batch did not surface the real snapshot write failure") + } + if err := os.Remove(path); err != nil { + t.Fatal(err) + } + s.Flush() + if err := s.LastPersistError(); err != nil { + t.Fatalf("explicit Flush did not recover: %v", err) + } + if _, ok := readFilePersistLoopDisk(t, path).GetItem("after-idle"); !ok { + t.Error("failed batch was not retained for explicit Flush") + } + }) +} + +func TestFileStoreCloseFlushesPendingBatch(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, path := newFilePersistLoopFixture(t) + addFilePersistLoopItem(t, s, "pending-at-close") + synctest.Wait() + // No timer advance or explicit Flush: shutdown itself must save it. + s.Close() + if err := s.LastPersistError(); err != nil { t.Fatal(err) } - if _, ok := restored.GetItem("first-message"); !ok { - t.Fatalf("continuous writes starved FileStore persistence for 20 batch intervals: memory items=%d, durable items=%d", len(s.ListThreadItems("thread")), len(restored.ListThreadItems("thread"))) + if _, ok := readFilePersistLoopDisk(t, path).GetItem("pending-at-close"); !ok { + t.Error("Close did not persist the pending batch") } }) } + +func newFilePersistLoopFixture(t *testing.T) (*FileStore, string) { + t.Helper() + path := filepath.Join(t.TempDir(), "store.json") + s, err := NewFile(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(s.Close) + if _, err := s.CreateProject("project", "Project", ""); err != nil { + t.Fatal(err) + } + if _, err := s.CreateThread("thread", "project", "Thread", "", "", ""); err != nil { + t.Fatal(err) + } + return s, path +} + +func addFilePersistLoopItem(t *testing.T, s *FileStore, id string) { + t.Helper() + if _, err := s.CreateItem(Item{ID: id, ProjectID: "project", ThreadID: "thread", Type: "event"}); err != nil { + t.Fatal(err) + } +} + +func readFilePersistLoopDisk(t *testing.T, path string) *Store { + t.Helper() + restored := New() + if err := loadFileSnapshot(path, restored); err != nil { + t.Fatal(err) + } + return restored +}