From 3e326515b46ade47692012dbcc45e1e563ce8d7b Mon Sep 17 00:00:00 2001 From: Delicious233 <101502465+DeliciousBuding@users.noreply.github.com> Date: Sun, 6 Sep 2026 14:35:57 +0800 Subject: [PATCH] test(store): benchmark snapshot persistence and SQLite facade contention Refs #2333. Keep production persistence, pool, and timer semantics unchanged. Co-authored-by: Codex --- docs/architecture/02-edge-server.md | 8 +- .../internal/store/persistence_bench_test.go | 189 +++++++++++++++++ .../store/sqlite_contention_bench_test.go | 193 ++++++++++++++++++ 3 files changed, 388 insertions(+), 2 deletions(-) create mode 100644 edge-server/internal/store/persistence_bench_test.go create mode 100644 edge-server/internal/store/sqlite_contention_bench_test.go diff --git a/docs/architecture/02-edge-server.md b/docs/architecture/02-edge-server.md index 686f4499f..7062670f9 100644 --- a/docs/architecture/02-edge-server.md +++ b/docs/architecture/02-edge-server.md @@ -2,7 +2,7 @@ > 子文档 | 主索引：[architecture.md](../architecture.md) > -> 最后更新：2026-09-02 +> 最后更新：2026-09-06 ## 职责 @@ -62,7 +62,11 @@ ProcessExecutor 配置 `RunTimeout`（默认 30 分钟）、`ShutdownGracePeriod ## Store 与事件持久化 -`Store`（`internal/store/store.go`）是核心内存数据结构，管理 Project、Thread、Run、Item、Pin、Diff、Artifact、Preview、UserProfile、AgentProfile、Settings 的全量 CRUD。`SQLiteStore`（`internal/store/sqlite_store.go`）通过 snapshot 持久化到 SQLite（WAL 模式，定期 checkpoint），支持崩溃恢复。终端状态 runs（completed/failed/cancelled/completed_with_issues）按 `TerminalTTL` 超时或 `MaxTerminalRunsPerThread` 上限自动清理，级联删除关联 diffs/artifacts/previews/items。 +`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 读者的争用直接代替业务读取测量。 + +终端状态 runs（completed/failed/cancelled/completed_with_issues）按 `TerminalTTL` 超时或 `MaxTerminalRunsPerThread` 上限自动清理，级联删除关联 diffs/artifacts/previews/items。 `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/persistence_bench_test.go b/edge-server/internal/store/persistence_bench_test.go new file mode 100644 index 000000000..74de7baa2 --- /dev/null +++ b/edge-server/internal/store/persistence_bench_test.go @@ -0,0 +1,189 @@ +package store + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "os" + "path/filepath" + "reflect" + "strings" + "testing" +) + +// BenchmarkFileStorePersistence separates snapshot copying, JSON encoding and +// filesystem work. The save/flush cases use real temp-file Sync and Rename; +// encode discards bytes without keeping a second snapshot-sized buffer. These +// are explicit flushes of fixed resident state, NOT debounce/request frequency. +func BenchmarkFileStorePersistence(b *testing.B) { + for _, runs := range []int{10, 100, 1000} { + b.Run(fmt.Sprintf("runs=%d", runs), func(b *testing.B) { + s := persistenceBenchmarkStore(b, runs) + want := s.snapshot() + var encoded bytes.Buffer + if err := encodePersistenceBenchmark(&encoded, want); err != nil { + b.Fatal(err) + } + payloadBytes := int64(encoded.Len()) + + b.Run("snapshot", func(b *testing.B) { + var got fileSnapshot + b.ReportAllocs() + for b.Loop() { + got = s.snapshot() + } + checkPersistenceBenchmarkSnapshot(b, got, want) + }) + b.Run("encode", func(b *testing.B) { + sink := &persistenceBenchmarkSink{} + b.SetBytes(payloadBytes) + b.ReportAllocs() + for b.Loop() { + sink.bytes = 0 + if err := encodePersistenceBenchmark(sink, want); err != nil { + b.Fatal(err) + } + if sink.bytes != payloadBytes { + b.Fatalf("encoded bytes = %d, want %d", sink.bytes, payloadBytes) + } + } + b.ReportMetric(float64(payloadBytes), "payload-B/op") + }) + b.Run("save", func(b *testing.B) { + path := filepath.Join(b.TempDir(), "store.json") + if err := saveFileSnapshot(path, want); err != nil { + b.Fatal(err) + } + b.SetBytes(payloadBytes) + b.ReportAllocs() + for b.Loop() { + if err := saveFileSnapshot(path, want); err != nil { + b.Fatal(err) + } + } + checkPersistenceBenchmarkFile(b, path, want, encoded.Bytes()) + b.ReportMetric(float64(payloadBytes), "payload-B/op") + }) + b.Run("flush", func(b *testing.B) { + path := filepath.Join(b.TempDir(), "store.json") + f, err := NewFile(path) + if err != nil { + b.Fatal(err) + } + b.Cleanup(f.Close) + // Fixture construction uses the validated Store APIs; installing + // its snapshot here avoids timing a backlog of debounce signals. + f.store.applySnapshot(want) + f.Flush() + if err := f.LastPersistError(); err != nil { + b.Fatal(err) + } + b.SetBytes(payloadBytes) + b.ReportAllocs() + for b.Loop() { + f.Flush() + if err := f.LastPersistError(); err != nil { + b.Fatal(err) + } + } + checkPersistenceBenchmarkFile(b, path, want, encoded.Bytes()) + b.ReportMetric(float64(payloadBytes), "payload-B/op") + }) + }) + } +} + +// All paths/content below are generated metadata; no user workspace is read. +// Ten queued runs per thread, ten 512-byte items, two artifact metadata records, +// and one checkpoint containing a 256-byte file per run. The remaining entity +// collections are empty. This is a declared resident-size fixture, not telemetry. +func persistenceBenchmarkStore(b *testing.B, runs int) *Store { + b.Helper() + s := New() + if _, err := s.CreateProject("project", "Benchmark", "owner"); err != nil { + b.Fatal(err) + } + content := strings.Repeat("x", 512) + for i := 0; i < runs; i++ { + threadID := fmt.Sprintf("thread-%04d", i/10) + if i%10 == 0 { + if _, err := s.CreateThread(threadID, "project", "Benchmark", "", "", ""); err != nil { + b.Fatal(err) + } + } + runID := fmt.Sprintf("run-%06d", i) + if _, err := s.CreateRun(runID, "project", threadID); err != nil { + b.Fatal(err) + } + for j := 0; j < 10; j++ { + _, err := s.CreateItem(Item{ + ID: fmt.Sprintf("item-%06d-%02d", i, j), ProjectID: "project", + ThreadID: threadID, RunID: runID, Type: "message", Role: "assistant", + Status: "completed", Content: content, + }) + if err != nil { + b.Fatal(err) + } + } + for j := 0; j < 2; j++ { + path := fmt.Sprintf("output/result-%02d.txt", j) + _, err := s.UpsertArtifact(Artifact{ + ID: fmt.Sprintf("artifact-%06d-%02d", i, j), RunID: runID, + ThreadID: threadID, Kind: "file", Path: path, SizeBytes: 512, + ContentSource: &ArtifactContentSource{Kind: "edge_file", Path: path, Readable: true}, + }) + if err != nil { + b.Fatal(err) + } + } + _, err := s.UpsertRunCheckpoint(RunCheckpoint{ + ID: fmt.Sprintf("checkpoint-%06d", i), RunID: runID, WorkDir: "fixture", + FileCount: 1, TotalBytes: 256, + Files: []CheckpointFile{{Path: "src/input.txt", Size: 256, Content: content[:256]}}, + }) + if err != nil { + b.Fatal(err) + } + } + return s +} + +func encodePersistenceBenchmark(w io.Writer, snapshot fileSnapshot) error { + encoder := json.NewEncoder(w) + encoder.SetIndent("", " ") // Match saveFileSnapshot, including the trailing newline. + return encoder.Encode(snapshot) +} + +type persistenceBenchmarkSink struct{ bytes int64 } + +func (s *persistenceBenchmarkSink) Write(p []byte) (int, error) { + s.bytes += int64(len(p)) + return len(p), nil +} + +func checkPersistenceBenchmarkFile(b *testing.B, path string, want fileSnapshot, encoded []byte) { + b.Helper() + actual, err := os.ReadFile(path) + if err != nil { + b.Fatal(err) + } + if !bytes.Equal(actual, encoded) { + b.Fatal("persisted bytes differ from the indented JSON control") + } + restored := New() + if err := loadFileSnapshot(path, restored); err != nil { + b.Fatal(err) + } + checkPersistenceBenchmarkSnapshot(b, restored.snapshot(), want) +} + +func checkPersistenceBenchmarkSnapshot(b *testing.B, got, want fileSnapshot) { + b.Helper() + g, w := reflect.ValueOf(got), reflect.ValueOf(want) + for i := 0; i < g.NumField(); i++ { + if !reflect.DeepEqual(g.Field(i).Interface(), w.Field(i).Interface()) { + b.Fatalf("restored snapshot differs in %s", g.Type().Field(i).Name) + } + } +} diff --git a/edge-server/internal/store/sqlite_contention_bench_test.go b/edge-server/internal/store/sqlite_contention_bench_test.go new file mode 100644 index 000000000..770b5ae1c --- /dev/null +++ b/edge-server/internal/store/sqlite_contention_bench_test.go @@ -0,0 +1,193 @@ +package store + +import ( + "fmt" + "math" + "math/bits" + "path/filepath" + "sync" + "testing" + "time" +) + +// BenchmarkSQLiteFacadeContention uses the real in-memory GetRun/ListRuns +// facade while one writer durably changes a fixed run. It does NOT add SQL +// readers or change the pool/PRAGMAs. Readers repeatedly inspect one active +// thread; resident runs in other threads remain in the fixture. Unthrottled +// reader loops are contention controls, not production request-rate estimates. +// Run with -run '^$' -bench '^BenchmarkSQLiteFacadeContention$' -benchmem -cpu=4. +func BenchmarkSQLiteFacadeContention(b *testing.B) { + for _, runs := range []int{10, 100, 1000} { + for _, readers := range []int{0, 1, 4} { + b.Run(fmt.Sprintf("runs=%d/readers=%d", runs, readers), func(b *testing.B) { + path := filepath.Join(b.TempDir(), "store.db") + s, err := NewSQLite(path) + if err != nil { + b.Fatal(err) + } + b.Cleanup(s.Close) + s.store.applySnapshot(persistenceBenchmarkStore(b, runs).snapshot()) + s.Flush() + if err := s.LastPersistError(); err != nil { + b.Fatal(err) + } + var changesBefore int64 + if err := s.db.QueryRow("SELECT total_changes()").Scan(&changesBefore); err != nil { + b.Fatal(err) + } + before := s.db.Stats() + if before.MaxOpenConnections != 1 { + b.Fatal("benchmark must retain the constructor's single-connection pool") + } + + start, stop := make(chan struct{}), make(chan struct{}) + var ready, finished sync.WaitGroup + var stopOnce sync.Once + stopReaders := func() { + stopOnce.Do(func() { close(stop) }) + finished.Wait() + } + defer stopReaders() + readStats := make([]persistenceBenchmarkLatency, readers) + readErrors := make([]int, readers) + for i := 0; i < readers; i++ { + ready.Add(1) + finished.Add(1) + go func() { + defer finished.Done() + ready.Done() + <-start + for { + select { + case <-stop: + return + default: + } + began := time.Now() + run, ok := s.GetRun("run-000000") + list := s.ListRuns("thread-0000") + readStats[i].add(time.Since(began)) + if !ok || run.ID != "run-000000" || run.ThreadID != "thread-0000" || len(list) != 10 { + readErrors[i]++ + return + } + } + }() + } + ready.Wait() + var writeStats persistenceBenchmarkLatency + var readWindow time.Time + writes := 0 + b.ReportAllocs() // Process-wide: includes the readers, not just the writer. + for b.Loop() { + if writes == 0 { + readWindow = time.Now() + close(start) + } + began := time.Now() + // An increasing value forces a durable delta without growing + // the entity set; it is not a retry policy/frequency simulation. + run, ok := s.SetRunRetryCount("run-000000", writes+1) + writeStats.add(time.Since(began)) + if !ok || run.RetryCount != writes+1 { + b.Fatalf("write failed: ok=%v error=%v", ok, s.LastPersistError()) + } + writes++ + } + stopReaders() + readSeconds := time.Since(readWindow).Seconds() + after := s.db.Stats() + var changesAfter int64 + if err := s.db.QueryRow("SELECT total_changes()").Scan(&changesAfter); err != nil { + b.Fatal(err) + } + b.ReportMetric(float64(changesAfter-changesBefore)/float64(writes), "sql-row-changes/op") + var reads persistenceBenchmarkLatency + for i := range readStats { + if readErrors[i] != 0 { + b.Fatalf("reader %d returned invalid data", i) + } + reads.merge(readStats[i]) + } + if err := s.LastPersistError(); err != nil { + b.Fatal(err) + } + b.ReportMetric(float64(writes)/b.Elapsed().Seconds(), "writes/s") + b.ReportMetric(float64(after.WaitCount-before.WaitCount)/float64(writes), "db-waits/op") + b.ReportMetric(float64(after.WaitDuration-before.WaitDuration)/float64(writes), "db-wait-ns/op") + b.ReportMetric(float64(after.OpenConnections), "db-open-conns") + writeStats.report(b, "write") + if readers > 0 { + b.ReportMetric(float64(reads.count)/readSeconds, "read-cycles/s") + reads.report(b, "read-cycle") + } + + // Read back before Close can mask a missing write with its extra + // flush. Compare all fields/content/order through the actual load + // path and independently check the final monotonic mutation. + want := s.store.snapshot() + if want.Runs["run-000000"].RetryCount != writes { + b.Fatal("final write is missing from memory") + } + restored, err := NewSQLite(path) + if err != nil { + b.Fatal(err) + } + b.Cleanup(restored.Close) + checkPersistenceBenchmarkSnapshot(b, restored.store.snapshot(), want) + }) + } + } +} + +// Each goroutine owns its histogram until it has stopped. Power-of-two +// nanosecond buckets bound memory independent of operation count; metrics are +// clock-observed percentile UPPER BOUNDS (not exact latencies). Zero clock +// readings are counted separately and reported as zero, never as a 1 ns result. +// One read cycle = GetRun+ListRuns. +// Sampling clocks/bucket updates add overhead; these are instrumented workload +// measurements, not a replacement for an uninstrumented microbenchmark. +type persistenceBenchmarkLatency struct { + buckets [64]uint64 + count uint64 + zero uint64 +} + +func (h *persistenceBenchmarkLatency) add(d time.Duration) { + h.count++ + if d <= 0 { + h.zero++ + return + } + h.buckets[bits.Len64(uint64(d)-1)]++ +} + +func (h *persistenceBenchmarkLatency) merge(other persistenceBenchmarkLatency) { + for i, count := range other.buckets { + h.buckets[i] += count + } + h.count += other.count + h.zero += other.zero +} + +func (h *persistenceBenchmarkLatency) report(b *testing.B, label string) { + b.Helper() + b.ReportMetric(float64(h.count), label+"-samples") + b.ReportMetric(float64(h.zero), label+"-clock-zero") + for _, percentile := range []uint64{50, 95, 99} { + rank := (h.count*percentile + 99) / 100 + unit := fmt.Sprintf("%s-p%d-observed-upper-ns", label, percentile) + if rank <= h.zero && h.count > 0 { + b.ReportMetric(0, unit) + continue + } + count := h.zero + for i, n := range h.buckets { + count += n + if count >= rank && h.count > 0 { + b.ReportMetric(math.Ldexp(1, i), unit) + break + } + } + } +}