From d8c171162d0e00ec401a1cf1d9869d47ec5f15ef Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Fri, 30 Oct 2026 11:47:54 -0700 Subject: [PATCH 1/2] perf: stop rewriting unchanged sessions every pass and share one app snapshot per poll Compare session facts independently of elapsed mixer presentation and reuse committed rows between passes. Restrict inferred parent work to live and recently ended sessions, and resolve explicit edges from one local index. Keep dashboard single-flight sharing while evicting superseded snapshots. Batch MCP history clock reads, and expose CPU, heap and memory profiling for the desktop app and development server. The 1600-session, 25-live benchmark drops from 274 ms and 587374 allocations to 49 ms and 166243 allocations per pass. Native and UTC package tests, web cache race tests, and native, Linux and Windows vet checks pass. --- internal/cli/app.go | 36 ++++- internal/cli/app_test.go | 2 + internal/cli/collect_profile.go | 48 +++++- internal/cli/collect_profile_test.go | 6 + internal/cli/command_tree_test.go | 2 +- internal/cli/dev_serve.go | 12 +- internal/collector/pass.go | 148 ++++++++++++------ internal/collector/pass_bench_test.go | 64 ++++++++ .../collector/session_write_changes_test.go | 88 +++++++++++ internal/collector/sources.go | 13 +- internal/collector/state.go | 25 +-- internal/store/mcp_history.go | 37 +++-- internal/store/store.go | 16 +- internal/web/overview_cache_test.go | 45 ++++++ internal/web/snapshot.go | 7 + 15 files changed, 471 insertions(+), 78 deletions(-) create mode 100644 internal/collector/pass_bench_test.go diff --git a/internal/cli/app.go b/internal/cli/app.go index e9e8c272..0fc3d989 100644 --- a/internal/cli/app.go +++ b/internal/cli/app.go @@ -7,6 +7,8 @@ import ( "io" "log/slog" "os" + "strings" + "time" "github.com/labstack/driga/internal/cleanup" "github.com/labstack/driga/internal/config" @@ -21,8 +23,10 @@ import ( // AppCmd opens the desktop app, or focuses the running one. type AppCmd struct { - NotificationStdin bool `hidden:"" help:"Read a notification target from standard input."` - Notification string `hidden:"" help:"Desktop notification event target."` + CPUProfile string `name:"cpu-profile" type:"path" help:"Write a CPU profile of the app to this file."` + ProfileFor time.Duration `name:"profile-for" default:"2m" help:"How long --cpu-profile records before it stops."` + NotificationStdin bool `hidden:"" help:"Read a notification target from standard input."` + Notification string `hidden:"" help:"Desktop notification event target."` } var ( @@ -47,16 +51,24 @@ var ( func AppLaunch(args []string) bool { login := len(args) == 0 && os.Getenv("DRIGA_APP") != "1" && openedAtLogin() app := appLaunch(os.Getenv, login, args) - systemLaunch = app && (len(args) == 0 || len(args) > 1 && (args[1] == "--notification" || args[1] == "--notification-stdin") && os.Getenv("DRIGA_APP") == "1") + systemLaunch = app && (len(args) == 0 || os.Getenv("DRIGA_APP") == "1") _ = os.Unsetenv("DRIGA_APP") return app } // appLaunch decides app mode: `driga app`, or a bare launch by the system. func appLaunch(env func(string) string, openedAtLogin bool, args []string) bool { - if len(args) >= 1 && args[0] == "app" && (len(args) == 1 || (args[1] == "--notification" || args[1] == "--notification-stdin")) { - return true + if len(args) > 0 && args[0] == "app" { + if len(args) == 1 { + return true + } + flag, _, _ := strings.Cut(args[1], "=") + switch flag { + case "--notification", "--notification-stdin", "--cpu-profile", "--profile-for": + return true + } } + return len(args) == 0 && (env("DRIGA_APP") == "1" || openedAtLogin) } @@ -104,9 +116,23 @@ func (c AppCmd) Run(g *Globals) error { } args = []string{"app", "--notification", c.Notification} } + if c.CPUProfile != "" { + if len(args) == 0 { + args = []string{"app"} + } + args = append(args, "--cpu-profile", c.CPUProfile, "--profile-for", c.ProfileFor.String()) + } if handed, err := handOffApp(systemLaunch || c.NotificationStdin, args...); handed || err != nil { return err } + if c.CPUProfile != "" { + stopProfile, err := startCPUProfile(c.CPUProfile, c.ProfileFor) + if err != nil { + return err + } + defer stopProfile() + } + restore := appLogging() defer restore() // Opened from Finder or at login, an error here would reach no one: the diff --git a/internal/cli/app_test.go b/internal/cli/app_test.go index e6b52fa3..2cf4bae1 100644 --- a/internal/cli/app_test.go +++ b/internal/cli/app_test.go @@ -37,6 +37,8 @@ func TestAppLaunch(t *testing.T) { {"command", "1", false, []string{"top"}, false}, {"command at login", "", true, []string{"top"}, false}, {"app", "", false, []string{"app"}, true}, + {"profile equals", "", false, []string{"app", "--cpu-profile=/tmp/app.pprof"}, true}, + {"profile", "", false, []string{"app", "--cpu-profile", "/tmp/app.pprof"}, true}, {"app help", "", false, []string{"app", "--help"}, false}, {"notification", "", false, []string{"app", "--notification", `{}`}, true}, {"other value", "0", false, nil, false}, diff --git a/internal/cli/collect_profile.go b/internal/cli/collect_profile.go index 456749c0..59ca166e 100644 --- a/internal/cli/collect_profile.go +++ b/internal/cli/collect_profile.go @@ -1,14 +1,17 @@ package cli import ( + "encoding/json" "os" + "runtime" "runtime/pprof" "sync" "time" ) -// startCPUProfile records the collector's CPU use into path for d and writes -// allocations to path + ".alloc", so the live service can be measured in place. +// startCPUProfile records CPU use into path for d, cumulative allocations to +// path + ".alloc", a final collected heap to ".heap", and memory/goroutine +// samples every five seconds to ".mem.jsonl". // The returned stop is idempotent and flushes early on shutdown. func startCPUProfile(path string, d time.Duration) (func(), error) { f, err := os.Create(path) @@ -25,13 +28,54 @@ func startCPUProfile(path string, d time.Duration) (func(), error) { alloc.Close() return nil, err } + memory, err := os.Create(path + ".mem.jsonl") + if err != nil { + pprof.StopCPUProfile() + f.Close() + alloc.Close() + return nil, err + } + done, sampled := make(chan struct{}), make(chan struct{}) + go func() { + defer close(sampled) + defer memory.Close() + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + sample := func() { + var m runtime.MemStats + runtime.ReadMemStats(&m) + _ = json.NewEncoder(memory).Encode(struct { + At time.Time + HeapAlloc, HeapObjects, TotalAlloc, Sys uint64 + NumGC uint32 + Goroutines int + }{time.Now(), m.HeapAlloc, m.HeapObjects, m.TotalAlloc, m.Sys, m.NumGC, runtime.NumGoroutine()}) + } + sample() + for { + select { + case <-ticker.C: + sample() + case <-done: + sample() + return + } + } + }() var once sync.Once stop := func() { once.Do(func() { pprof.StopCPUProfile() + close(done) + <-sampled f.Close() pprof.Lookup("allocs").WriteTo(alloc, 0) alloc.Close() + if heap, err := os.Create(path + ".heap"); err == nil { + runtime.GC() + _ = pprof.WriteHeapProfile(heap) + _ = heap.Close() + } }) } t := time.AfterFunc(d, stop) diff --git a/internal/cli/collect_profile_test.go b/internal/cli/collect_profile_test.go index a9cbfb52..984fe66c 100644 --- a/internal/cli/collect_profile_test.go +++ b/internal/cli/collect_profile_test.go @@ -33,6 +33,12 @@ func TestCPUProfileStopsAfterDuration(t *testing.T) { } time.Sleep(10 * time.Millisecond) } + stop() + for _, suffix := range []string{".heap", ".mem.jsonl"} { + if fi, err := os.Stat(path + suffix); err != nil || fi.Size() == 0 { + t.Fatalf("missing %s: %v", suffix, err) + } + } // A second profile can start, so the first one released the profiler. again, err := startCPUProfile(filepath.Join(t.TempDir(), "again.pprof"), time.Hour) if err != nil { diff --git a/internal/cli/command_tree_test.go b/internal/cli/command_tree_test.go index b6f46db9..4005926f 100644 --- a/internal/cli/command_tree_test.go +++ b/internal/cli/command_tree_test.go @@ -22,7 +22,7 @@ func parseCommand(t *testing.T, args ...string) (*Root, *kong.Context, error) { } func TestCommandTree(t *testing.T) { paths := []string{ - "", "summary", "flow", "flow 12", "agents", "agents list", "agents show 12", "agents jump 12", "agents kill 12", "projects", "projects list", "projects show app", "projects summarize app other", "projects summarize --all", "inbox", "inbox list", "inbox ack item", "inbox ack --all", "inbox jump item", "orphans", "orphans list", "orphans kill item", "orphans kill --all", "machines", "machines list", "machines show", "machines show local", "machines reset-id", "cleanup", "cleanup list", "cleanup remove item", "cleanup remove --all --kind worktree", "top", "app", "ask what needs me", "mcp", "doctor", "init", "config path", "config show", "config edit", "service run", "service run --serve --listen 127.0.0.1:7411", "service run --scheduled --service-data-dir fixture --service-env DRIGA_CAPTURE_TEXT=titles", "service install", "service start", "service stop", "service restart", "service status", "service logs -f", "service uninstall", "hooks install", "hooks install claude", "hooks uninstall codex", "hooks status", "hooks status all", "fleet peers list", "fleet peers add host:7411 --device-id device -y", "fleet peers remove peer", "fleet viewers list", "fleet viewers revoke viewer", "fleet enroll --allow-viewer viewer --allow-ack -y", "fleet rekey", "fleet id", "fleet status", "version", "hook claude Stop", "feed tail --after-seq 1 -f", "dev agy-schema", "mcp --chat", "mcp --snapshot", + "", "summary", "flow", "flow 12", "agents", "agents list", "agents show 12", "agents jump 12", "agents kill 12", "projects", "projects list", "projects show app", "projects summarize app other", "projects summarize --all", "inbox", "inbox list", "inbox ack item", "inbox ack --all", "inbox jump item", "orphans", "orphans list", "orphans kill item", "orphans kill --all", "machines", "machines list", "machines show", "machines show local", "machines reset-id", "cleanup", "cleanup list", "cleanup remove item", "cleanup remove --all --kind worktree", "top", "app", "app --cpu-profile app.pprof --profile-for 1m", "dev serve --cpu-profile app.pprof --profile-for 1m", "ask what needs me", "mcp", "doctor", "init", "config path", "config show", "config edit", "service run", "service run --serve --listen 127.0.0.1:7411", "service run --scheduled --service-data-dir fixture --service-env DRIGA_CAPTURE_TEXT=titles", "service install", "service start", "service stop", "service restart", "service status", "service logs -f", "service uninstall", "hooks install", "hooks install claude", "hooks uninstall codex", "hooks status", "hooks status all", "fleet peers list", "fleet peers add host:7411 --device-id device -y", "fleet peers remove peer", "fleet viewers list", "fleet viewers revoke viewer", "fleet enroll --allow-viewer viewer --allow-ack -y", "fleet rekey", "fleet id", "fleet status", "version", "hook claude Stop", "feed tail --after-seq 1 -f", "dev agy-schema", "mcp --chat", "mcp --snapshot", } for _, path := range paths { t.Run(path, func(t *testing.T) { diff --git a/internal/cli/dev_serve.go b/internal/cli/dev_serve.go index eae5cfb9..9292c730 100644 --- a/internal/cli/dev_serve.go +++ b/internal/cli/dev_serve.go @@ -19,10 +19,20 @@ import ( // dev). It is the app without its window: the same page, data, Chat and // actions, but no native dialogs, menus, tray or notifications. type DevServeCmd struct { - Addr string `default:"127.0.0.1:47801" help:"Loopback address to listen on."` + CPUProfile string `name:"cpu-profile" type:"path" help:"Write a CPU profile of the app to this file."` + ProfileFor time.Duration `name:"profile-for" default:"2m" help:"How long --cpu-profile records before it stops."` + Addr string `default:"127.0.0.1:47801" help:"Loopback address to listen on."` } func (c DevServeCmd) Run(g *Globals) error { + if c.CPUProfile != "" { + stopProfile, err := startCPUProfile(c.CPUProfile, c.ProfileFor) + if err != nil { + return err + } + defer stopProfile() + } + host, _, err := net.SplitHostPort(c.Addr) if err != nil { return invalidUsage("--addr must be host:port") diff --git a/internal/collector/pass.go b/internal/collector/pass.go index d52574da..7d6a72c5 100644 --- a/internal/collector/pass.go +++ b/internal/collector/pass.go @@ -373,10 +373,12 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err return err } // Inbox grouping needs parent IDs before labels consume the unseen items. - parents := derive.LinkParents(ss, idx) - if err := explicitParents(tx, ss, parents); err != nil { + resolveParents, err := explicitParentResolver(tx, ss) + if err != nil { return err } + parents := linkPassParents(ss, idx, now) + resolveParents(parents) if err := syncInbox(tx, d, ss, parents, hooks, baseline); err != nil { return err } @@ -415,10 +417,8 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err return err } } - parents = derive.LinkParents(ss, idx) - if err := explicitParents(tx, ss, parents); err != nil { - return err - } + parents = linkPassParents(ss, idx, now) + resolveParents(parents) for i := range ss { ss[i].ParentID = nil if id, ok := parents[ss[i].ID]; ok { @@ -500,11 +500,10 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err } work.procs = slices.Clone(procs) } - for _, s := range ss { + for i, s := range ss { before := original[s.ID] - current := s - current.Completions, current.FilesTouched, current.Events = nil, nil, nil - if reflect.DeepEqual(before, current) { + if sameSessionRow(before, s) { + ss[i] = before continue } if _, err := tx.UpsertSession(s); err != nil { @@ -534,6 +533,12 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err return err } } + work.sessions = make([]model.Session, len(ss)) + for i, s := range ss { + // Keep only row data. Transcript batches must not accumulate in run state. + s.Events, s.Completions, s.FilesTouched = nil, nil, nil + work.sessions[i] = s + } r.Sessions, r.Procs = len(ss), len(procs) return tx.SetMeta("last_pass_at", now.Format(time.RFC3339Nano)) }) @@ -546,9 +551,18 @@ func Pass(ctx context.Context, d Deps, st *store.Store, p Plan) (PassReport, err state.procs = work.procs state.logs = work.logs state.panelSeen = work.panelSeen + state.sessions, state.sessionStore, state.sessionVersion = work.sessions, work.sessionStore, work.sessionVersion + state.sessionRevision = st.Revision() } if err == nil { + revision := st.Revision() err = alertInbox(ctx, d, st) + // Alert planning commits one session-neutral metadata transaction. Do not + // invalidate our row index for that transaction, but keep invalidation if + // another writer (including a notifier callback) committed as well. + if err == nil && d.Notifier != nil && !d.snapshot && state.sessionRevision == revision && st.Revision() == revision+1 { + state.sessionRevision = revision + 1 + } } return r, err } @@ -558,52 +572,53 @@ func sessionKey(a, id string) string { return a + "\x00" + id } // Keep unresolved adapter references in transactional metadata so a parent // first seen after a collector restart can still be linked. Session row IDs // are resolved only through the local machine's store view. -func explicitParents(tx *store.Tx, ss []model.Session, parents map[int64]int64) error { - stored, err := tx.Sessions(store.SessionFilter{IncludeEnded: true}) +func explicitParentResolver(tx *store.Tx, ss []model.Session) (func(map[int64]int64), error) { + refs, err := tx.MetaPrefix("session-parent:") if err != nil { - return err + return nil, err } - ids := make(map[string]int64, len(stored)) - for _, s := range stored { + ids := make(map[string]int64, len(ss)) + for _, s := range ss { ids[sessionKey(s.Adapter, s.ExternalID)] = s.ID } + type edge struct{ child, parent int64 } + var edges []edge for _, s := range ss { - key := "session-parent:" + sessionKey(s.Adapter, s.ExternalID) + key := sessionKey(s.Adapter, s.ExternalID) ref := s.ParentExternalID - if ref != "" { - if err := tx.SetMeta(key, ref); err != nil { - return err - } - } else { - ref, err = tx.Meta(key) - if err != nil && !errors.Is(err, store.ErrNotFound) { - return err - } - } if ref == "" { - continue - } - // Explicit evidence also suppresses an unrelated inferred parent while - // the referenced row is still missing. - delete(parents, s.ID) - id := ids[sessionKey(s.Adapter, ref)] - if id == 0 || id == s.ID { - continue - } - seen := map[int64]bool{s.ID: true} - cycle := false - for p := id; p != 0; p = parents[p] { - if seen[p] { - cycle = true - break + ref = refs[key] + } else if refs[key] != ref { + if err := tx.SetMeta("session-parent:"+key, ref); err != nil { + return nil, err } - seen[p] = true } - if !cycle { - parents[s.ID] = id + if ref != "" { + edges = append(edges, edge{s.ID, ids[sessionKey(s.Adapter, ref)]}) } } - return nil + // Only children with explicit evidence need resolution; both label phases + // reuse this index. Pending historical references still resolve on arrival. + return func(parents map[int64]int64) { + for _, e := range edges { + delete(parents, e.child) + if e.parent == 0 || e.parent == e.child { + continue + } + seen := map[int64]bool{e.child: true} + cycle := false + for p := e.parent; p != 0; p = parents[p] { + if seen[p] { + cycle = true + break + } + seen[p] = true + } + if !cycle { + parents[e.child] = e.parent + } + } + }, nil } func collectFeedEnabled(d Deps, dir string) (bool, error) { @@ -1188,3 +1203,46 @@ func spoolOffsetPrefix(d Deps) string { } return "spool/" } + +// Reconsider live and recently ended children. Historical links remain stable; +// explicit references are resolved against the complete local identity index. +func linkPassParents(ss []model.Session, idx derive.ProcIndex, now time.Time) map[int64]int64 { + candidates := make([]model.Session, 0, 32) + historical := make(map[int64]int64) + for _, s := range ss { + if s.EndedAt == nil || !s.EndedAt.Before(now.Add(-5*time.Minute)) { + candidates = append(candidates, s) + } else if s.ParentID != nil { + historical[s.ID] = *s.ParentID + } + } + for child, parent := range derive.LinkParents(candidates, idx) { + historical[child] = parent + } + return historical +} + +// Compare persisted facts, excluding adapter-only batches and clock-derived +// mixer presentation. A changed verdict, support, input or CPU sample still +// writes immediately; elapsed time alone does not create a new observation. +func sameSessionRow(a, b model.Session) bool { + normalize := func(s model.Session) model.Session { + s.MachineID = "" + s.ParentExternalID = "" + s.Events, s.Completions, s.FilesTouched = nil, nil, nil + s.Evidence = evidenceDuration.ReplaceAllString(s.Evidence, "") + if s.Mixer != nil { + m := *s.Mixer + m.ObservedAt = time.Time{} + m.Evidence = evidenceDuration.ReplaceAllString(m.Evidence, "") + m.Rows = slices.Clone(m.Rows) + for i := range m.Rows { + m.Rows[i].Age = 0 + m.Rows[i].Strength = 0 + } + s.Mixer = &m + } + return s + } + return reflect.DeepEqual(normalize(a), normalize(b)) +} diff --git a/internal/collector/pass_bench_test.go b/internal/collector/pass_bench_test.go new file mode 100644 index 00000000..1760fbba --- /dev/null +++ b/internal/collector/pass_bench_test.go @@ -0,0 +1,64 @@ +package collector + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/labstack/driga/internal/adapter" + "github.com/labstack/driga/internal/config" + "github.com/labstack/driga/internal/model" + "github.com/labstack/driga/internal/signatures" + "github.com/labstack/driga/internal/store" +) + +func BenchmarkCollectorPass(b *testing.B) { + now := time.Date(2026, 10, 30, 12, 0, 0, 0, time.UTC) + st, err := store.Open("", store.Memory) + if err != nil { + b.Fatal(err) + } + defer st.Close() + p := &fakePlatform{} + a := &fakeAdapter{name: "claude"} + err = st.Tx(context.Background(), func(tx *store.Tx) error { + for i := 0; i < 1600; i++ { + obs := model.SessionObs{Adapter: "claude", ExternalID: fmt.Sprint(i), Cwd: "/synthetic", StartedAt: now.Add(-24 * time.Hour), LastActivityAt: now.Add(-time.Hour), ConversationClaimed: true} + s := model.Session{SessionObs: obs, Label: "ended"} + ended := now.Add(-time.Hour) + s.EndedAt = &ended + if i < 25 { + s.EndedAt = nil + s.Label = "working" + obs.PID = int32(100 + i) + obs.ProcStart = obs.StartedAt + obs.RegistryStatus = "busy" + s.SessionObs = obs + a.obs = append(a.obs, obs) + p.procs = append(p.procs, model.Proc{PID: obs.PID, StartTime: obs.ProcStart, Cmdline: "claude", Argv: []string{"claude"}, Cwd: obs.Cwd}) + } + if _, err := tx.UpsertSession(s); err != nil { + return err + } + } + return nil + }) + if err != nil { + b.Fatal(err) + } + d := Deps{Cfg: config.Default(), Plat: p, Adapters: []adapter.Adapter{a}, Sigs: signatures.Builtin(), SpoolDir: b.TempDir(), State: NewState(), Now: func() time.Time { return now }} + for range 2 { + if _, err := Pass(context.Background(), d, st, Plan{Procs: true}); err != nil { + b.Fatal(err) + } + } + b.ReportAllocs() + b.ResetTimer() + for range b.N { + now = now.Add(time.Second) + if _, err := Pass(context.Background(), d, st, Plan{Procs: true}); err != nil { + b.Fatal(err) + } + } +} diff --git a/internal/collector/session_write_changes_test.go b/internal/collector/session_write_changes_test.go index c194e1e6..6c733008 100644 --- a/internal/collector/session_write_changes_test.go +++ b/internal/collector/session_write_changes_test.go @@ -1,15 +1,19 @@ package collector import ( + "context" "database/sql" "path/filepath" "testing" + "time" + "github.com/labstack/driga/internal/model" "github.com/labstack/driga/internal/store" ) func TestPassUpsertsOnlyChangedSessions(t *testing.T) { d, _, _, a := setup(t) + d.Notifier = &fakeNotifier{} path := filepath.Join(t.TempDir(), "db") st, err := store.Open(path, store.ReadWrite) if err != nil { @@ -29,6 +33,7 @@ func TestPassUpsertsOnlyChangedSessions(t *testing.T) { if _, err = db.Exec("UPDATE writes SET n=0"); err != nil { t.Fatal(err) } + d.Now = func() time.Time { return time.Date(2026, 10, 18, 12, 0, 1, 0, time.UTC) } pass(t, d, st, Plan{}) var n int if err := db.QueryRow("SELECT n FROM writes").Scan(&n); err != nil { @@ -37,6 +42,9 @@ func TestPassUpsertsOnlyChangedSessions(t *testing.T) { if n != 0 { t.Fatalf("unchanged session updated %d times", n) } + if d.State.sessionRevision != st.Revision() { + t.Fatal("session-neutral alert transaction invalidated session cache") + } a.obs[0].Title = "changed" pass(t, d, st, Plan{}) if err := db.QueryRow("SELECT n FROM writes").Scan(&n); err != nil { @@ -46,3 +54,83 @@ func TestPassUpsertsOnlyChangedSessions(t *testing.T) { t.Fatalf("changed session writes=%d", n) } } + +func TestPassSessionCacheSeesLocalAndExternalWrites(t *testing.T) { + d, _, _, a := setup(t) + path := filepath.Join(t.TempDir(), "db") + st, err := store.Open(path, store.ReadWrite) + if err != nil { + t.Fatal(err) + } + defer st.Close() + pass(t, d, st, Plan{Procs: true}) + a.obs = nil + for _, external := range []bool{false, true} { + writer := st + if external { + writer, err = store.Open(path, store.ReadWrite) + if err != nil { + t.Fatal(err) + } + } + err = writer.Tx(context.Background(), func(tx *store.Tx) error { + ss, err := tx.Sessions(store.SessionFilter{IncludeEnded: true}) + if err != nil { + return err + } + ss[0].Title = "written outside pass" + if external { + ss[0].Title = "external connection" + } + _, err = tx.UpsertSession(ss[0]) + return err + }) + if external { + writer.Close() + } + if err != nil { + t.Fatal(err) + } + pass(t, d, st, Plan{}) + want := "written outside pass" + if external { + want = "external connection" + } + if len(d.State.sessions) != 1 || d.State.sessions[0].Title != want { + t.Fatalf("cache failed to invalidate: %+v", d.State.sessions) + } + } +} + +func TestSessionRowComparisonKeepsFacts(t *testing.T) { + at := time.Date(2026, 10, 30, 12, 0, 0, 0, time.UTC) + base := model.Session{ID: 1, SessionObs: model.SessionObs{Adapter: "claude", ExternalID: "one"}, Mixer: &model.StateMixer{ObservedAt: at, Input: &model.StateInput{}, Verdict: "idle"}} + later := base + m := *base.Mixer + m.ObservedAt = at.Add(time.Second) + later.Mixer = &m + if !sameSessionRow(base, later) { + t.Fatal("clock alone changed row") + } + for _, mutate := range []func(*model.Session){ + func(s *model.Session) { s.Title = "changed" }, + func(s *model.Session) { s.RegistryUpdatedAt = at }, + func(s *model.Session) { s.LastPrompt = "changed" }, + func(s *model.Session) { s.Label = "stuck" }, + func(s *model.Session) { id := int64(2); s.ParentID = &id }, + func(s *model.Session) { + m := *s.Mixer + in := *m.Input + in.Proc = &model.Proc{PID: 42, CPUPercent: 7} + m.Input = &in + s.Mixer = &m + }, + func(s *model.Session) { m := *s.Mixer; m.Agreement++; s.Mixer = &m }, + } { + changed := later + mutate(&changed) + if sameSessionRow(base, changed) { + t.Fatalf("changed fact suppressed: %+v", changed) + } + } +} diff --git a/internal/collector/sources.go b/internal/collector/sources.go index e93ada3e..6cc77109 100644 --- a/internal/collector/sources.go +++ b/internal/collector/sources.go @@ -7,6 +7,7 @@ import ( "fmt" "log/slog" "path/filepath" + "slices" "sort" "strings" "sync" @@ -101,7 +102,17 @@ func collectSources(ctx context.Context, d Deps, st *store.Store, p Plan, state if err != nil { return r, err } - ss, err := st.Sessions(store.SessionFilter{IncludeEnded: true}) + version, err := st.DataVersion(ctx) + if err != nil { + return r, err + } + var ss []model.Session + if state.sessions != nil && state.sessionStore == st && state.sessionVersion == version && state.sessionRevision == st.Revision() { + ss = slices.Clone(state.sessions) + } else { + ss, err = st.Sessions(store.SessionFilter{IncludeEnded: true}) + } + state.sessionStore, state.sessionVersion = st, version if err != nil { return r, err } diff --git a/internal/collector/state.go b/internal/collector/state.go index 77a4af61..738b73b5 100644 --- a/internal/collector/state.go +++ b/internal/collector/state.go @@ -6,20 +6,25 @@ import ( "time" "github.com/labstack/driga/internal/model" + "github.com/labstack/driga/internal/store" ) // State retains sampling and process ownership evidence between passes. It is // private to a collector run and is never written to disk. type State struct { - mu sync.Mutex - prev map[int32]model.Proc - prevAt time.Time - prevMachine model.MachineSample - cwds map[int32]cwdResult - cwdsAt time.Time - cwdNext int - spawn map[string]int64 - gitFullAt time.Time + mu sync.Mutex + sessions []model.Session + sessionStore *store.Store + sessionVersion int64 + sessionRevision uint64 + prev map[int32]model.Proc + prevAt time.Time + prevMachine model.MachineSample + cwds map[int32]cwdResult + cwdsAt time.Time + cwdNext int + spawn map[string]int64 + gitFullAt time.Time // procs is the process table this run last wrote, or nil before the first // successful scan, when the stored rows are the only source. procs []model.Proc @@ -42,7 +47,7 @@ func NewState() *State { // replaces prev, spawn, procs and logs wholesale and never writes into them, so // they are shared; only the cwd cache is updated in place and is copied. func (s *State) copy() *State { - n := &State{prev: s.prev, spawn: s.spawn, procs: s.procs, logs: s.logs, cwds: make(map[int32]cwdResult, len(s.cwds))} + n := &State{sessions: s.sessions, sessionStore: s.sessionStore, sessionVersion: s.sessionVersion, sessionRevision: s.sessionRevision, prev: s.prev, spawn: s.spawn, procs: s.procs, logs: s.logs, cwds: make(map[int32]cwdResult, len(s.cwds))} n.prevAt, n.cwdsAt, n.gitFullAt = s.prevAt, s.cwdsAt, s.gitFullAt n.cwdNext = s.cwdNext n.prevMachine = s.prevMachine diff --git a/internal/store/mcp_history.go b/internal/store/mcp_history.go index aad85dea..9a2f46d9 100644 --- a/internal/store/mcp_history.go +++ b/internal/store/mcp_history.go @@ -128,12 +128,20 @@ func (tx *Tx) ObserveMCP(at time.Time, samples []model.MCPObservation) error { // Endpoints are chronological within each group; groups can interleave. func (q *querier) ScanMCPHistory(since, until time.Time, visit func(model.MCPObservation) error) error { since = maxTime(since, until.Add(-30*24*time.Hour)).Truncate(time.Minute) - rows, err := q.conn.QueryContext(q.ctx, `SELECT `+mcpStateColumns+`, - CASE WHEN s.resolution=60 THEN (SELECT MIN(at) FROM mcp_clock c WHERE c.machine_id=s.machine_id AND c.at>=MAX(s.first_at,?) AND c.at<=MIN(s.last_at,?)) ELSE s.last_at END, - CASE WHEN s.resolution=60 THEN (SELECT MAX(at) FROM mcp_clock c WHERE c.machine_id=s.machine_id AND c.at>=MAX(s.first_at,?) AND c.at<=MIN(s.last_at,?)) ELSE s.last_at END + // Resolve run endpoints against one bounded shared clock in memory. Two + // correlated MIN/MAX subqueries per state repeatedly reread the same pages. + clock, err := readMany(q, `SELECT at FROM mcp_clock WHERE machine_id=? AND at>=? AND at<=? ORDER BY at`, []any{q.machineID, unix(since), unix(until)}, func(s scanner) (int64, error) { + var at int64 + err := s.Scan(&at) + return at, err + }) + if err != nil { + return err + } + rows, err := q.conn.QueryContext(q.ctx, `SELECT `+mcpStateColumns+` FROM mcp_states s JOIN mcp_groups g ON g.id=s.group_id JOIN mcp_labels l ON l.id=s.labels_id WHERE s.machine_id=? AND s.first_at>=? AND s.first_at<=? AND s.last_at>=? - ORDER BY s.first_at,s.group_id`, unix(since), unix(until), unix(since), unix(until), q.machineID, unix(since.Truncate(time.Hour)), unix(until), unix(since)) + ORDER BY s.first_at,s.group_id`, q.machineID, unix(since.Truncate(time.Hour)), unix(until), unix(since)) if err != nil { return err } @@ -141,24 +149,33 @@ func (q *querier) ScanMCPHistory(since, until time.Time, visit func(model.MCPObs for rows.Next() { var r mcpState var start int64 - var first, last *int64 - err := rows.Scan(&r.id, &r.first, &r.last, &r.sample.Host.PID, &start, &r.sample.Key, &r.sample.Host.Name, &r.sample.Host.Exe, &r.sample.Host.Cwd, &r.sample.Name, &r.sample.Count, &r.sample.Footprint, &r.members, &r.sample.ResolutionSeconds, &first, &last) + var first, last int64 + err := rows.Scan(&r.id, &r.first, &r.last, &r.sample.Host.PID, &start, &r.sample.Key, &r.sample.Host.Name, &r.sample.Host.Exe, &r.sample.Host.Cwd, &r.sample.Name, &r.sample.Count, &r.sample.Footprint, &r.members, &r.sample.ResolutionSeconds) if err != nil { return err } - if first == nil || last == nil || *last > unix(until) { + first, last = r.last, r.last + if r.sample.ResolutionSeconds == 60 { + lo := sort.Search(len(clock), func(i int) bool { return clock[i] >= r.first }) + hi := sort.Search(len(clock), func(i int) bool { return clock[i] > r.last }) + if lo >= hi { + continue + } + first, last = clock[lo], clock[hi-1] + } + if last > unix(until) { continue } r.sample.Host.StartedAt = fromUnix(start) if err := json.Unmarshal([]byte(r.members), &r.sample.Members); err != nil { return err } - r.sample.At = fromUnix(*first) + r.sample.At = fromUnix(first) if err := visit(r.sample); err != nil { return err } - if *last != *first { - r.sample.At = fromUnix(*last) + if last != first { + r.sample.At = fromUnix(last) if err := visit(r.sample); err != nil { return err } diff --git a/internal/store/store.go b/internal/store/store.go index 90f3d6dd..3a9096fd 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -16,6 +16,7 @@ import ( "path/filepath" "runtime" "strings" + "sync/atomic" "time" "github.com/google/uuid" @@ -58,7 +59,8 @@ type querier struct { // there would wait for the connection held by the transaction and deadlock. type Store struct { querier - db *sql.DB + db *sql.DB + revision *atomic.Uint64 } // Tx provides writes and the same reads as Store, on the active transaction. @@ -204,7 +206,7 @@ func Open(path string, mode Mode) (_ *Store, retErr error) { if err := ValidateMachineID(id); err != nil { return nil, err } - return &Store{querier: querier{conn: db, ctx: ctx, machineID: id, localID: id, dataDir: dataDir}, db: db}, nil + return &Store{revision: new(atomic.Uint64), querier: querier{conn: db, ctx: ctx, machineID: id, localID: id, dataDir: dataDir}, db: db}, nil } func randomID() (string, error) { @@ -252,7 +254,11 @@ func (s *Store) Tx(ctx context.Context, fn func(*Tx) error) error { if err := fn(&Tx{querier{conn: tx, ctx: ctx, machineID: s.machineID, localID: s.localID, dataDir: s.dataDir}}); err != nil { return err } - return tx.Commit() + if err := tx.Commit(); err != nil { + return err + } + s.revision.Add(1) + return nil } // DataVersion changes when another connection commits a write. Compare values @@ -428,3 +434,7 @@ func checkSchema(ctx context.Context, db *sql.DB, path string) (have, want int64 } return have, want, nil } + +// Revision counts successful transactions on this connection, including local writes +// which SQLite data_version deliberately excludes. +func (s *Store) Revision() uint64 { return s.revision.Load() } diff --git a/internal/web/overview_cache_test.go b/internal/web/overview_cache_test.go index a001c7d3..fe320905 100644 --- a/internal/web/overview_cache_test.go +++ b/internal/web/overview_cache_test.go @@ -290,3 +290,48 @@ func TestCleanupSizesReachHeaderWithoutRelisting(t *testing.T) { t.Fatalf("removed candidate still counted: %v", v) } } + +func TestSnapshotCacheDropsSupersededVersions(t *testing.T) { + version := int64(1) + s := &Server{source: cacheSource{version: func(context.Context) (int64, error) { return version, nil }, load: func(context.Context, store.Scope) (viewer.Snapshot, error) { return viewer.Snapshot{}, nil }}} + for range 40 { + if _, err := s.cachedLoad(context.Background(), store.Scope{}, false); err != nil { + t.Fatal(err) + } + if len(s.cache) != 1 { + t.Fatalf("retained %d snapshot versions", len(s.cache)) + } + version++ + } +} + +func TestDashboardPollSharesSnapshotAndInvalidatesOnCommit(t *testing.T) { + base, st, _, _ := fixture(t) + source := &countingSource{Source: base.source} + s, err := New(source, Options{}) + if err != nil { + t.Fatal(err) + } + poll := func() { + t.Helper() + for _, spec := range []panel.Dashboard{panel.Header(), panel.Home(), + {Name: "Inbox", Panels: []panel.Panel{{ID: "inbox", Title: "Inbox", Viz: "inbox", Source: "inbox"}}}, + {Name: "Fleet", Panels: []panel.Panel{{ID: "machines", Title: "Machines", Viz: "table", Source: "machines"}, {ID: "sessions", Title: "Sessions", Viz: "table", Source: "sessions"}, {ID: "flow", Title: "Flow", Viz: "table", Source: "flow"}}}, + } { + if _, err := s.runPanels(context.Background(), panel.RunRequest{Dashboard: spec}); err != nil { + t.Fatal(err) + } + } + } + poll() + if source.loads != 1 { + t.Fatalf("poll built %d snapshots", source.loads) + } + if err := st.Tx(context.Background(), func(tx *store.Tx) error { return tx.SetMeta("cache-test", "changed") }); err != nil { + t.Fatal(err) + } + poll() + if source.loads != 2 || len(s.cache) != 1 { + t.Fatalf("commit: loads=%d retained=%d", source.loads, len(s.cache)) + } +} diff --git a/internal/web/snapshot.go b/internal/web/snapshot.go index 9a8f2f0f..844fafb4 100644 --- a/internal/web/snapshot.go +++ b/internal/web/snapshot.go @@ -75,6 +75,13 @@ func (s *Server) cachedLoad(ctx context.Context, scope store.Scope, all bool) (v s.cacheMu.Lock() delete(s.snapshotLoads, key) if load.err == nil { + // A new store version invalidates every old overview. Keeping those + // snapshots until the size cap retained entire process/event histories. + for old, cached := range s.cache { + if old.version != version || time.Since(cached.at) >= 15*time.Second { + delete(s.cache, old) + } + } if s.cache == nil || len(s.cache) >= 32 { s.cache = map[snapshotKey]cachedSnapshot{} } From ea1a452d513d159d719e9b59e11681e208721ff5 Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Fri, 30 Oct 2026 12:04:05 -0700 Subject: [PATCH 2/2] docs: regenerate the CLI reference for the app profiling flags --- site/src/content/docs/reference/cli.mdx | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/site/src/content/docs/reference/cli.mdx b/site/src/content/docs/reference/cli.mdx index 14922db6..b54579f6 100644 --- a/site/src/content/docs/reference/cli.mdx +++ b/site/src/content/docs/reference/cli.mdx @@ -480,9 +480,14 @@ driga top [flags] Open the Driga desktop app, or focus it. ```sh -driga app +driga app [flags] ``` +| Flag | Default | Description | +|---|---|---| +| `--cpu-profile=` | | Write a CPU profile of the app to this file. | +| `--profile-for=` | `2m` | How long --cpu-profile records before it stops. | + ## driga ask Ask a one-shot Chat question through Claude and driga's read-only tools.