diff --git a/docs/architecture/02-edge-server.md b/docs/architecture/02-edge-server.md index 27410398f..90aec5d9b 100644 --- a/docs/architecture/02-edge-server.md +++ b/docs/architecture/02-edge-server.md @@ -66,7 +66,7 @@ ProcessExecutor 配置 `RunTimeout`（默认 30 分钟）、`ShutdownGracePeriod `FileStore`（`internal/store/file_store.go`）对写入信号做 debounce，再保存全量 JSON 快照；显式 `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 移除；这不删除工作区文件。 +终端状态 runs（finished/failed/cancelled/completed_with_issues）按 `TerminalTTL` 超时或 `MaxTerminalRunsPerThread` 上限自动清理，级联删除关联 diffs/artifacts/previews/items/checkpoints。Checkpoint 在 run 完成时保留，在 run 清理或所属 thread 删除时随 run 移除；这不删除工作区文件。 SQLite 后台清理有删除时同步提交；失败通过 `LastPersistError` 和日志留痕，下一周期即使没有新删除也会重试。关闭时先停止并等待后台清理/checkpoint 任务，再做最终持久化与数据库关闭。 `EventBus`（`internal/events/bus.go`）是基于 channel 的发布/订阅模型：4 worker 并发 observer、子 channel 缓冲（256）、gap detection（`system.gap` 事件）；通过 `PersistFn` 钩子先持久化再广播。`EventLog` 是 append-only JSON-lines 事件日志（默认 50 MiB 上限，超限截断保留尾部 75%）。 diff --git a/edge-server/internal/store/sqlite_periodic_cleanup_test.go b/edge-server/internal/store/sqlite_periodic_cleanup_test.go new file mode 100644 index 000000000..daeae3dd9 --- /dev/null +++ b/edge-server/internal/store/sqlite_periodic_cleanup_test.go @@ -0,0 +1,199 @@ +package store + +import ( + "path/filepath" + "testing" + "testing/synctest" + "time" +) + +func TestSQLitePeriodicCleanupIsDurableBeforeClose(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, path := newSQLitePeriodicCleanupFixture(t, false) + advanceSQLiteCleanupTick() + if _, ok := s.GetRun("remove-run"); ok { + t.Fatal("periodic cleanup did not remove the expired run from memory") + } + if got := sqliteCleanupFixtureRows(t, s); got != 0 { + t.Errorf("periodic cleanup left %d durable run/evidence rows after removing them from memory", got) + } + stopSQLiteWithoutFinalPersist(t, s) + assertSQLiteCleanupReopen(t, path) + }) +} + +func TestSQLitePeriodicCleanupRetriesFailedCommit(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, path := newSQLitePeriodicCleanupFixture(t, true) + advanceSQLiteCleanupTick() + if _, ok := s.GetRun("remove-run"); ok { + t.Fatal("failed persist unexpectedly retained the run in memory") + } + if s.LastPersistError() == nil { + t.Fatal("periodic persist failure was not recorded") + } + if got := sqliteCleanupFixtureRows(t, s); got != 6 { + t.Fatalf("failed cleanup transaction left %d rows, want all six retained", got) + } + if _, err := s.db.Exec("DROP TRIGGER reject_periodic_cleanup"); err != nil { + t.Fatal(err) + } + // No business write or final Flush repairs the failed delete. The next + // actual background tick must retry it despite an empty cleanup result. + advanceSQLiteCleanupTick() + if err := s.LastPersistError(); err != nil { + t.Errorf("next cleanup tick did not clear recovered persist failure: %v", err) + } + if got := sqliteCleanupFixtureRows(t, s); got != 0 { + t.Errorf("recovered periodic cleanup left %d durable rows", got) + } + stopSQLiteWithoutFinalPersist(t, s) + assertSQLiteCleanupReopen(t, path) + }) +} + +func TestSQLiteCloseFlushesPendingCleanup(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, path := newSQLitePeriodicCleanupFixture(t, true) + advanceSQLiteCleanupTick() + if s.LastPersistError() == nil { + t.Fatal("expected the injected periodic persist failure") + } + if _, err := s.db.Exec("DROP TRIGGER reject_periodic_cleanup"); err != nil { + t.Fatal(err) + } + // Do not advance to another tick: Close must flush pending memory after + // its workers have stopped, then close the database. + s.Close() + if err := s.LastPersistError(); err != nil { + t.Fatalf("final cleanup persist failed: %v", err) + } + assertSQLiteCleanupReopen(t, path) + }) +} + +func TestSQLiteCloseWaitsForBackgroundWork(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, err := NewSQLite(filepath.Join(t.TempDir(), "close.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(s.Close) + release := make(chan struct{}) + s.backgroundWG.Go(func() { <-release }) + closed := make(chan struct{}) + go func() { + s.Close() + close(closed) + }() + synctest.Wait() + select { + case <-closed: + t.Error("Close returned before owned background work finished") + default: + } + if err := s.db.Ping(); err != nil { + t.Errorf("database closed before owned background work finished: %v", err) + } + close(release) + synctest.Wait() + select { + case <-closed: + default: + t.Error("Close did not finish after owned background work completed") + } + if err := s.db.Ping(); err == nil { + t.Error("database remains open after Close") + } + }) +} + +// Virtual time drives the constructor's real production-cadence loops. No +// shortened production interval or wall-clock sleep is needed in these tests. +func advanceSQLiteCleanupTick() { + synctest.Wait() + <-time.After(sqliteBackgroundLoopInterval) + synctest.Wait() +} + +func newSQLitePeriodicCleanupFixture(t *testing.T, rejectDelete bool) (*SQLiteStore, string) { + t.Helper() + s, path := newSQLiteCheckpointDeltaFixture(t) + if _, err := s.UpsertRunDiffFile(RunDiffFile{RunID: "remove-run", Path: "before.txt", Diff: "old evidence", Status: "modified"}); err != nil { + t.Fatal(err) + } + if _, err := s.UpsertArtifact(Artifact{ID: "expired-artifact", RunID: "remove-run", Path: "before.txt"}); err != nil { + t.Fatal(err) + } + if _, err := s.UpsertPreview(Preview{ID: "expired-preview", RunID: "remove-run", Status: "ready"}); err != nil { + t.Fatal(err) + } + if _, err := s.CreateItem(Item{ID: "expired-item", ProjectID: "project", ThreadID: "remove-thread", RunID: "remove-run", Type: "event"}); err != nil { + t.Fatal(err) + } + s.store.mu.Lock() + run := s.store.runs["remove-run"] + run.Status = "finished" + run.FinishedAt = time.Now().Add(-sqliteCleanupTerminalTTL - time.Hour).UTC().Format(time.RFC3339) + s.store.runs[run.ID] = run + s.store.mu.Unlock() + if err := s.syncPersist(); err != nil { + t.Fatal(err) + } + if got := sqliteCleanupFixtureRows(t, s); got != 6 { + t.Fatalf("fixture durable rows = %d, want run and five evidence rows", got) + } + if rejectDelete { + _, err := s.db.Exec("CREATE TRIGGER reject_periodic_cleanup BEFORE DELETE ON agenthub_store_rows " + + "WHEN OLD.row_kind = 'run' AND OLD.row_id = 'remove-run' " + + "BEGIN SELECT RAISE(ABORT, 'injected periodic cleanup failure'); END") + if err != nil { + t.Fatal(err) + } + } + return s, path +} + +func sqliteCleanupFixtureRows(t *testing.T, s *SQLiteStore) int { + t.Helper() + var count int + if err := s.db.QueryRow("SELECT COUNT(*) FROM agenthub_store_rows WHERE json_extract(payload, '$.runId') = ?", "remove-run").Scan(&count); err != nil { + t.Fatal(err) + } + return count +} + +func stopSQLiteWithoutFinalPersist(t *testing.T, s *SQLiteStore) { + t.Helper() + s.closeOnce.Do(func() { + close(s.stopCheckpoint) + s.backgroundWG.Wait() + if err := s.db.Close(); err != nil { + t.Fatal(err) + } + }) +} + +func assertSQLiteCleanupReopen(t *testing.T, path string) { + t.Helper() + restored, err := NewSQLite(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(restored.Close) + if got := sqliteCleanupFixtureRows(t, restored); got != 0 { + t.Errorf("reopen restored %d expired run/evidence rows", got) + } + if _, ok := restored.GetRun("remove-run"); ok { + t.Error("expired run reappeared after reopen") + } + if _, ok := restored.GetRunCheckpoint("remove-run"); ok { + t.Error("expired checkpoint reappeared after reopen") + } + if run, ok := restored.GetRun("keep-run"); !ok || run.Status != "queued" { + t.Error("unrelated queued run was removed or changed") + } + if cp, ok := restored.GetRunCheckpoint("keep-run"); !ok || len(cp.Files) != 1 || cp.Files[0].Content != "keep evidence" { + t.Error("unrelated queued checkpoint was removed or changed") + } +} diff --git a/edge-server/internal/store/sqlite_store.go b/edge-server/internal/store/sqlite_store.go index 46ea0fcd1..1413b4c1b 100644 --- a/edge-server/internal/store/sqlite_store.go +++ b/edge-server/internal/store/sqlite_store.go @@ -39,6 +39,7 @@ type SQLiteStore struct { store *Store persistMu sync.Mutex closeOnce sync.Once + backgroundWG sync.WaitGroup lastErr error lastSnapshot fileSnapshot // rowsSeeded marks agenthub_store_rows as the durable source of truth: @@ -46,7 +47,7 @@ type SQLiteStore struct { // (write amplification — the payload is only the pre-rows fallback and // is kept fresh until the first commit that durably seeds the rows). rowsSeeded bool - stopCheckpoint chan struct{} // closed to stop periodic WAL checkpoint + stopCheckpoint chan struct{} // closed to stop both background loops } func NewSQLite(path string) (*SQLiteStore, error) { @@ -77,11 +78,11 @@ func NewSQLite(path string) (*SQLiteStore, error) { // in-memory page cache small. On Windows the Go runtime is reluctant to // return freed memory to the OS, so keeping the WAL small is critical. s.stopCheckpoint = make(chan struct{}) - go s.checkpointLoop(sqliteBackgroundLoopInterval) + s.backgroundWG.Go(func() { s.checkpointLoop(sqliteBackgroundLoopInterval) }) // Periodically clean up old terminal runs to prevent unbounded Store map growth. // Without this, Store maps only shrink when a new run is created, // which may never happen on an idle server. - go s.cleanupLoop(sqliteBackgroundLoopInterval) + s.backgroundWG.Go(func() { s.cleanupLoop(sqliteBackgroundLoopInterval) }) return s, nil } @@ -103,16 +104,27 @@ func (s *SQLiteStore) checkpointLoop(interval time.Duration) { } } -// cleanupLoop periodically removes old terminal runs to prevent unbounded -// in-memory Store map growth. Without this, Store maps only shrink when a -// new run is explicitly created — which may never happen on an idle server. +// cleanupLoop periodically removes old terminal runs and persists the deletes. +// Failed commits remain pending for the next tick, even if no new runs expire. func (s *SQLiteStore) cleanupLoop(interval time.Duration) { ticker := time.NewTicker(interval) defer ticker.Stop() + pendingPersist := false for { select { case <-ticker.C: - s.store.CleanupRuns(sqlitePeriodicCleanupOptions()) + result := s.store.CleanupRuns(sqlitePeriodicCleanupOptions()) + // A failed commit already removed data from memory. Retry it even + // when the next tick has no additional runs to remove. + pendingPersist = pendingPersist || shouldSyncAfterCleanup(result) + if !pendingPersist { + continue + } + if err := s.syncPersist(); err != nil { + slog.Warn("sqlite store: periodic cleanup persist failed", "error", err) + continue + } + pendingPersist = false case <-s.stopCheckpoint: return } @@ -122,6 +134,7 @@ func (s *SQLiteStore) cleanupLoop(interval time.Duration) { func (s *SQLiteStore) Close() { s.closeOnce.Do(func() { close(s.stopCheckpoint) + s.backgroundWG.Wait() // Final checkpoint to shrink the WAL before close. Failures are logged // (not returned) to keep the shutdown signature; LastPersistError still // reflects the final persist outcome.