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 new file mode 100644 index 000000000..42312cd3d --- /dev/null +++ b/edge-server/internal/store/file_store_persist_loop_test.go @@ -0,0 +1,164 @@ +package store + +import ( + "fmt" + "os" + "path/filepath" + "testing" + "testing/synctest" + "time" +) + +func TestFileStorePersistsDuringContinuousWrites(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + 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) + addFilePersistLoopItem(t, s, fmt.Sprintf("busy-%d", i)) + synctest.Wait() + } + if err := s.LastPersistError(); err != nil { + t.Fatal(err) + } + // 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 := 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 +}