Close the plan: screenshots and a measured performance pass
- docs/screenshots: nineteen pages captured with headless Chromium from a running instance; README gains a screenshot section - tests/perf: opt-in load test (NOEMA_PERF=1) seeding 60k events, 6k memories and 1.5k beliefs across two minds, timing sixteen hot reads against a 400 ms ceiling and rejecting sequential-scan plans - migration 0016: indexes the test found missing (memories by mind and recency, beliefs by mind and confidence, events by mind and category, belief versions by mind and time) - docs/performance.md records the numbers and how to rerun
This commit is contained in:
Andy Dixon
@@ -0,0 +1,191 @@
|
||||
// Package perf seeds a mind with a realistic volume of state and times the
|
||||
// hot read paths. It runs only with NOEMA_PERF=1 because seeding takes a
|
||||
// while; CI runs it nightly, developers run it before touching queries.
|
||||
//
|
||||
// NOEMA_PERF=1 NOEMA_TEST_DATABASE_URL=... go test -run TestHotPaths -v ./tests/perf/
|
||||
package perf
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/andydixon/noema/internal/beliefs"
|
||||
evstore "github.com/andydixon/noema/internal/events"
|
||||
"github.com/andydixon/noema/internal/memory"
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
"github.com/andydixon/noema/internal/snapshots"
|
||||
"github.com/andydixon/noema/internal/storage"
|
||||
"github.com/andydixon/noema/migrations"
|
||||
"github.com/andydixon/noema/pkg/concepts"
|
||||
"github.com/andydixon/noema/pkg/events"
|
||||
"github.com/andydixon/noema/pkg/ids"
|
||||
)
|
||||
|
||||
const (
|
||||
nEvents = 60_000
|
||||
nMemories = 6_000
|
||||
nBeliefs = 1_500
|
||||
ceiling = 400 * time.Millisecond // any hot read slower than this is a regression
|
||||
)
|
||||
|
||||
var vocab = strings.Fields("payments database migration latency deploy atlas server disk cpu alert go rust python andy project service host backup replica queue cache timeout retry release rollback metric dashboard incident postmortem")
|
||||
|
||||
func phrase(i int) string {
|
||||
var w []string
|
||||
for k := 0; k < 8; k++ {
|
||||
w = append(w, vocab[(i*7+k*3)%len(vocab)])
|
||||
}
|
||||
return fmt.Sprintf("%s (item %d)", strings.Join(w, " "), i)
|
||||
}
|
||||
|
||||
func TestHotPaths(t *testing.T) {
|
||||
if os.Getenv("NOEMA_PERF") == "" {
|
||||
t.Skip("set NOEMA_PERF=1 to run the performance pass")
|
||||
}
|
||||
pool := storage.TestPool(t, migrations.FS)
|
||||
ctx := context.Background()
|
||||
ms := minds.NewStore(pool)
|
||||
es := evstore.NewStore(pool, evstore.NewBus(8), nil)
|
||||
mem := memory.NewStore(ctx, pool, nil)
|
||||
bs := beliefs.NewStore(pool)
|
||||
m, err := ms.Create(ctx, minds.CreateParams{Slug: "perf", DisplayName: "Perf"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
other, _ := ms.Create(ctx, minds.CreateParams{Slug: "noise", DisplayName: "Noise"})
|
||||
|
||||
seed := time.Now()
|
||||
batch := make([]events.New, 0, 500)
|
||||
flush := func(mind ids.ID) {
|
||||
if len(batch) == 0 {
|
||||
return
|
||||
}
|
||||
if _, err := es.Append(ctx, batch...); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
batch = batch[:0]
|
||||
}
|
||||
types := []events.Type{events.Observation, events.ThoughtSelected, events.MemoryRetrieved, events.BeliefUpdated, events.WorkspaceBroadcast, events.CycleCompleted}
|
||||
for i := 0; i < nEvents; i++ {
|
||||
mind := m.ID
|
||||
if i%5 == 0 {
|
||||
mind = other.ID // a second mind so scoping matters
|
||||
}
|
||||
batch = append(batch, events.New{MindID: mind, Type: types[i%len(types)], Source: "perf", OccurredAt: seed.Add(-time.Duration(nEvents-i) * time.Second), Summary: phrase(i), Salience: float64(i%10) / 10, Confidence: 0.8})
|
||||
if len(batch) == 500 {
|
||||
flush(mind)
|
||||
}
|
||||
}
|
||||
flush(m.ID)
|
||||
for i := 0; i < nMemories; i++ {
|
||||
kind := concepts.Episodic
|
||||
if i%4 == 0 {
|
||||
kind = concepts.Semantic
|
||||
}
|
||||
if _, err := mem.Create(ctx, pool, memory.New{MindID: m.ID, Kind: kind, Epistemic: concepts.Observation, Content: phrase(i), Source: "perf", Trust: events.TrustUser, Confidence: 0.7, Importance: float64(i%10) / 10, OccurredAt: seed.Add(-time.Duration(i) * time.Minute)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
emit := func(ctx context.Context, q storage.Querier, n events.New) (events.Event, error) {
|
||||
out, err := es.AppendIn(ctx, q, n)
|
||||
if err != nil {
|
||||
return events.Event{}, err
|
||||
}
|
||||
return out[0], nil
|
||||
}
|
||||
for i := 0; i < nBeliefs; i++ {
|
||||
if _, err := bs.Create(ctx, pool, emit, beliefs.New{MindID: m.ID, Proposition: phrase(i * 3), Subject: vocab[i%len(vocab)], Epistemic: concepts.Claim, Source: "perf",
|
||||
Evidence: []beliefs.EvidenceInput{{Direction: "for", Kind: "memory", RefID: ids.New().String(), Source: "user", Weight: 0.4, Credibility: 0.8}}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
t.Logf("seeded %d events, %d memories, %d beliefs in %s", nEvents, nMemories, nBeliefs, time.Since(seed).Round(time.Millisecond))
|
||||
|
||||
type probe struct {
|
||||
name string
|
||||
fn func() error
|
||||
}
|
||||
probes := []probe{
|
||||
{"events recent 50", func() error { _, err := es.Query(ctx, evstore.Filter{MindID: m.ID, Limit: 50}); return err }},
|
||||
{"events by type", func() error {
|
||||
_, err := es.Query(ctx, evstore.Filter{MindID: m.ID, Types: []events.Type{events.BeliefUpdated}, Limit: 50})
|
||||
return err
|
||||
}},
|
||||
{"events after seq (cursor)", func() error {
|
||||
_, err := es.Query(ctx, evstore.Filter{MindID: m.ID, AfterSeq: 1000, Ascending: true, Limit: 200})
|
||||
return err
|
||||
}},
|
||||
{"events search", func() error {
|
||||
_, err := es.Query(ctx, evstore.Filter{MindID: m.ID, Search: "rollback", Limit: 50})
|
||||
return err
|
||||
}},
|
||||
{"events stats", func() error { _, err := es.Stats(ctx, m.ID, time.Time{}); return err }},
|
||||
{"memory list 50", func() error { _, err := mem.List(ctx, m.ID, memory.Filter{Limit: 50}); return err }},
|
||||
{"memory text search", func() error {
|
||||
_, err := mem.List(ctx, m.ID, memory.Filter{Search: "payments latency", Limit: 50})
|
||||
return err
|
||||
}},
|
||||
{"memory fused retrieve", func() error {
|
||||
_, err := mem.Retrieve(ctx, m.ID, memory.Query{Text: "database migration latency alert", Limit: 10})
|
||||
return err
|
||||
}},
|
||||
{"memory stats", func() error { _, err := mem.Stats(ctx, m.ID); return err }},
|
||||
{"beliefs list", func() error { _, err := bs.List(ctx, m.ID, beliefs.Filter{Limit: 50}); return err }},
|
||||
{"beliefs search", func() error { _, err := bs.Search(ctx, m.ID, "deploy rollback", 20); return err }},
|
||||
{"beliefs contested", func() error { _, err := bs.Contested(ctx, m.ID, 20); return err }},
|
||||
{"beliefs uncertain", func() error { _, err := bs.Uncertain(ctx, m.ID, 20); return err }},
|
||||
{"beliefs stats", func() error { _, err := bs.Stats(ctx, m.ID); return err }},
|
||||
{"beliefs recent changes", func() error { _, err := bs.RecentChanges(ctx, m.ID, seed.Add(-time.Hour), 50); return err }},
|
||||
{"snapshot capture", func() error { _, err := snapshots.Capture(ctx, pool, es, m); return err }},
|
||||
}
|
||||
failed := false
|
||||
for _, p := range probes {
|
||||
p.fn() //nolint:errcheck // warm the cache once; we time the steady state
|
||||
start := time.Now()
|
||||
var err error
|
||||
for i := 0; i < 3; i++ {
|
||||
err = p.fn()
|
||||
}
|
||||
d := time.Since(start) / 3
|
||||
mark := ""
|
||||
if d > ceiling {
|
||||
mark, failed = " <-- over ceiling", true
|
||||
}
|
||||
t.Logf("%-28s %8s%s", p.name, d.Round(100*time.Microsecond), mark)
|
||||
if err != nil {
|
||||
t.Errorf("%s: %v", p.name, err)
|
||||
}
|
||||
}
|
||||
// Plans: the hot event and memory queries must be index-driven.
|
||||
plans := map[string]string{
|
||||
"events recent": `EXPLAIN SELECT id FROM cognitive_events WHERE mind_id=$1 ORDER BY seq DESC LIMIT 50`,
|
||||
"events cursor": `EXPLAIN SELECT id FROM cognitive_events WHERE mind_id=$1 AND seq > 1000 ORDER BY seq ASC LIMIT 200`,
|
||||
"memory list": `EXPLAIN SELECT id FROM memories WHERE mind_id=$1 AND status='active' ORDER BY occurred_at DESC, id DESC LIMIT 50`,
|
||||
"beliefs list": `EXPLAIN SELECT id FROM beliefs WHERE mind_id=$1 AND status NOT IN ('superseded') ORDER BY confidence DESC LIMIT 50`,
|
||||
}
|
||||
for name, q := range plans {
|
||||
rows, err := pool.Query(ctx, q, m.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var plan []string
|
||||
for rows.Next() {
|
||||
var line string
|
||||
rows.Scan(&line) //nolint:errcheck
|
||||
plan = append(plan, strings.TrimSpace(line))
|
||||
}
|
||||
rows.Close()
|
||||
joined := strings.Join(plan, " | ")
|
||||
if strings.Contains(joined, "Seq Scan on cognitive_events") || strings.Contains(joined, "Seq Scan on memories") || strings.Contains(joined, "Seq Scan on beliefs") {
|
||||
t.Errorf("%s uses a sequential scan: %s", name, joined)
|
||||
}
|
||||
t.Logf("plan %-14s %s", name, plan[0])
|
||||
}
|
||||
if failed {
|
||||
t.Fatalf("a hot read exceeded %s", ceiling)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user