diff --git a/desktop/wailskit/chat.go b/desktop/wailskit/chat.go index 321019939..1c3cd9c90 100644 --- a/desktop/wailskit/chat.go +++ b/desktop/wailskit/chat.go @@ -3319,7 +3319,7 @@ func (b *ChatBridge) recordMetric(ev interface{}) { me.Model = b.currentSes.Model me.Vendor = b.currentSes.Vendor me.Endpoint = b.currentSes.Endpoint - b.currentSes.Metrics = append(b.currentSes.Metrics, me) + b.currentSes.AppendMetricEvent(me) // #3086: lock-guarded append (was a bare slice append) b.currentSes.AppendMetricForEndpoint(b.currentSes.Vendor, b.currentSes.Endpoint, me) } b.metricEvents = append(b.metricEvents, me) diff --git a/internal/im/daemon_bridge.go b/internal/im/daemon_bridge.go index fcfc05a97..dac45bed7 100644 --- a/internal/im/daemon_bridge.go +++ b/internal/im/daemon_bridge.go @@ -1197,7 +1197,7 @@ func (b *DaemonBridge) recordMetric(ev metrics.MetricEvent) { ev.Model = ses.Model ev.Vendor = ses.Vendor ev.Endpoint = ses.Endpoint - ses.Metrics = append(ses.Metrics, ev) + ses.AppendMetricEvent(ev) // #3086: lock-guarded append (was a bare slice append) ses.AppendMetricForEndpoint(ses.Vendor, ses.Endpoint, ev) } b.mu.Unlock() diff --git a/internal/session/endpoint_stats.go b/internal/session/endpoint_stats.go index 866003d1d..a5df20602 100644 --- a/internal/session/endpoint_stats.go +++ b/internal/session/endpoint_stats.go @@ -33,20 +33,82 @@ func (s *Session) ensureEndpointStatsLocked() { } } +// AddUsageHistoryEntry appends to UsageHistory under endpointStatsMu. +// #3086: cross-goroutine callback appends (`ses.UsageHistory = append(...)` +// in TUI/IM/desktop bridges) raced the endpoint-stats rebuild readers that +// range the same slice - `slice = append(slice, x)` updates len and pointer +// in separate steps, so a concurrent range can see a torn view or panic on +// the stale-array/new-len window. Appends and reads of the source slices +// now all pass through this mutex. Safe under higher-level session/bridge +// locks (the mutex is nested, never the other way around). +func (s *Session) AddUsageHistoryEntry(entry UsageEntry) { + if s == nil { + return + } + s.endpointStatsMu.Lock() + s.UsageHistory = append(s.UsageHistory, entry) + s.endpointStatsMu.Unlock() +} + +// AppendMetricEvent appends to Metrics under endpointStatsMu - the +// Metrics-side twin of AddUsageHistoryEntry (#3086). +func (s *Session) AppendMetricEvent(ev metrics.MetricEvent) { + if s == nil { + return + } + s.endpointStatsMu.Lock() + s.Metrics = append(s.Metrics, ev) + s.endpointStatsMu.Unlock() +} + +// UsageHistorySnapshot returns a copy of UsageHistory taken under the +// same lock the appends hold. Exported for TUI/desktop read paths that +// run off the session goroutine (cost snapshot per frame, /cost export) +// - #3086. +func (s *Session) UsageHistorySnapshot() []UsageEntry { + return s.usageHistorySnapshot() +} + +// usageHistorySnapshot returns a copy of UsageHistory taken under the +// same lock the appends hold (#3086). +func (s *Session) usageHistorySnapshot() []UsageEntry { + s.endpointStatsMu.RLock() + snap := append([]UsageEntry(nil), s.UsageHistory...) + s.endpointStatsMu.RUnlock() + return snap +} + +// MetricsSnapshot returns a copy of Metrics taken under the same lock +// the appends hold. Exported for TUI read paths that capture the slice +// in closures crossing goroutines (/trace export) - #3086. +func (s *Session) MetricsSnapshot() []metrics.MetricEvent { + return s.metricsSnapshot() +} + +// metricsSnapshot returns a copy of Metrics taken under the same lock the +// appends hold (#3086). +func (s *Session) metricsSnapshot() []metrics.MetricEvent { + s.endpointStatsMu.RLock() + snap := append([]metrics.MetricEvent(nil), s.Metrics...) + s.endpointStatsMu.RUnlock() + return snap +} + func (s *Session) RebuildEndpointStats() { if s == nil { return } usageByEndpoint := make(map[string]provider.TokenUsage) metricsByEndpoint := make(map[string][]metrics.MetricEvent) - for _, entry := range s.UsageHistory { + usageHistory := s.usageHistorySnapshot() // #3086: read under the append lock + for _, entry := range usageHistory { key := EndpointStatsKey(entry.Vendor, entry.Endpoint) if key == "" { continue } usageByEndpoint[key] = usageByEndpoint[key].Add(entry.Usage) } - for _, ev := range s.Metrics { + for _, ev := range s.metricsSnapshot() { // #3086: read under the append lock key := EndpointStatsKey(ev.Vendor, ev.Endpoint) if key == "" { continue @@ -108,9 +170,10 @@ func (s *Session) UsageForEndpoint(vendor, endpoint string) provider.TokenUsage s.endpointStatsMu.RLock() usage, ok := s.EndpointUsage[key] hasBuckets := len(s.EndpointUsage) > 0 + hasHistory := len(s.UsageHistory) > 0 s.endpointStatsMu.RUnlock() if key == "" { - if !hasBuckets && len(s.UsageHistory) == 0 { + if !hasBuckets && !hasHistory { return s.TokenUsage } return provider.TokenUsage{} @@ -118,7 +181,7 @@ func (s *Session) UsageForEndpoint(vendor, endpoint string) provider.TokenUsage if ok { return usage } - if !hasBuckets && len(s.UsageHistory) > 0 { + if !hasBuckets && hasHistory { s.RebuildEndpointStats() s.endpointStatsMu.RLock() usage, ok = s.EndpointUsage[key] @@ -127,8 +190,11 @@ func (s *Session) UsageForEndpoint(vendor, endpoint string) provider.TokenUsage return usage } } + // #3086: hasHistory was read under the lock at the top of this + // function; appends since then only shrink the empty-history window, + // never reopen it. sessionKey := EndpointStatsKey(s.Vendor, s.Endpoint) - if len(s.UsageHistory) == 0 && (sessionKey == key || sessionKey == "") { + if !hasHistory && (sessionKey == key || sessionKey == "") { return s.TokenUsage } return provider.TokenUsage{} @@ -142,18 +208,19 @@ func (s *Session) MetricsForEndpoint(vendor, endpoint string) []metrics.MetricEv s.endpointStatsMu.RLock() events, ok := s.EndpointMetrics[key] hasBuckets := len(s.EndpointMetrics) > 0 + metricsSrc := append([]metrics.MetricEvent(nil), s.Metrics...) // #3086: read under the append lock s.endpointStatsMu.RUnlock() if key == "" { if !hasBuckets { hasMetadata := false - for _, ev := range s.Metrics { + for _, ev := range metricsSrc { if strings.TrimSpace(ev.Vendor) != "" || strings.TrimSpace(ev.Endpoint) != "" { hasMetadata = true break } } if !hasMetadata { - return append([]metrics.MetricEvent(nil), s.Metrics...) + return metricsSrc } } return nil @@ -161,7 +228,7 @@ func (s *Session) MetricsForEndpoint(vendor, endpoint string) []metrics.MetricEv if ok { return append([]metrics.MetricEvent(nil), events...) } - if !hasBuckets && len(s.Metrics) > 0 { + if !hasBuckets && len(metricsSrc) > 0 { s.RebuildEndpointStats() s.endpointStatsMu.RLock() events, ok = s.EndpointMetrics[key] @@ -175,7 +242,7 @@ func (s *Session) MetricsForEndpoint(vendor, endpoint string) []metrics.MetricEv } } hasMetadata := false - for _, ev := range s.Metrics { + for _, ev := range metricsSrc { if strings.TrimSpace(ev.Vendor) != "" || strings.TrimSpace(ev.Endpoint) != "" { hasMetadata = true break @@ -186,7 +253,7 @@ func (s *Session) MetricsForEndpoint(vendor, endpoint string) []metrics.MetricEv } sessionKey := EndpointStatsKey(s.Vendor, s.Endpoint) if sessionKey == key || sessionKey == "" { - return append([]metrics.MetricEvent(nil), s.Metrics...) + return metricsSrc } return nil } diff --git a/internal/session/zz_issue3086_test.go b/internal/session/zz_issue3086_test.go new file mode 100644 index 000000000..3790850f1 --- /dev/null +++ b/internal/session/zz_issue3086_test.go @@ -0,0 +1,122 @@ +package session + +// Regression probes for #3086 (endpoint_stats data race): the +// endpointStatsMu mutex covered only the aggregate maps; the SOURCE slices +// (UsageHistory/Metrics) were appended by cross-goroutine callbacks +// (TUI/IM/desktop) as bare `s.X = append(s.X, e)` while +// UsageForEndpoint/MetricsForEndpoint/rebuild ranged them lock-free. +// +// These probes run under `go test -race`: before the fix the concurrent +// append+read trips the race detector; after it, all access passes +// through endpointStatsMu-guarded methods. + +import ( + "sync" + "testing" + + "github.com/topcheer/ggcode/internal/metrics" + "github.com/topcheer/ggcode/internal/provider" +) + +// TestIssue3086_UsageRebuildVsConcurrentAppend: one goroutine appends via +// AddUsageHistoryEntry while the reader path triggers the lazy rebuild. +func TestIssue3086_UsageRebuildVsConcurrentAppend(t *testing.T) { + s := &Session{Vendor: "v", Endpoint: "e"} + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + for i := 0; i < 200; i++ { + s.AddUsageHistoryEntry(UsageEntry{ + Vendor: "v", + Endpoint: "e", + Usage: provider.TokenUsage{InputTokens: 1}, + }) + } + }() + go func() { + defer wg.Done() + for i := 0; i < 200; i++ { + _ = s.UsageForEndpoint("v", "e") + } + }() + wg.Wait() + // The lazy rebuild fires at most once (while the aggregate map is + // empty), so a mid-race reader freezes a partial aggregate - that is + // the documented lazy-migration semantics, not a race. Rebuild once + // after the writers settle to assert every guarded append landed. + s.RebuildEndpointStats() + if got := s.UsageForEndpoint("v", "e"); got.InputTokens != 200 { + t.Fatalf("aggregated usage = %d, want 200", got.InputTokens) + } +} + +// TestIssue3086_MetricsReadVsConcurrentAppend: same shape for Metrics. +func TestIssue3086_MetricsReadVsConcurrentAppend(t *testing.T) { + s := &Session{Vendor: "v", Endpoint: "e"} + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + for i := 0; i < 200; i++ { + s.AppendMetricEvent(metrics.MetricEvent{Vendor: "v", Endpoint: "e", TurnIndex: i}) + } + }() + go func() { + defer wg.Done() + for i := 0; i < 200; i++ { + _ = s.MetricsForEndpoint("v", "e") + } + }() + wg.Wait() + // Same lazy-migration semantics as the usage probe: rebuild after the + // writers settle. + s.RebuildEndpointStats() + if got := len(s.MetricsForEndpoint("v", "e")); got != 200 { + t.Fatalf("metrics count = %d, want 200", got) + } +} + +// TestIssue3086_RebuildVsConcurrentAppend: the explicit rebuild path races +// too (it ranges both source slices). +func TestIssue3086_RebuildVsConcurrentAppend(t *testing.T) { + s := &Session{Vendor: "v", Endpoint: "e"} + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + for i := 0; i < 100; i++ { + s.AddUsageHistoryEntry(UsageEntry{Vendor: "v", Endpoint: "e", Usage: provider.TokenUsage{OutputTokens: 1}}) + s.AppendMetricEvent(metrics.MetricEvent{Vendor: "v", Endpoint: "e"}) + } + }() + go func() { + defer wg.Done() + for i := 0; i < 100; i++ { + s.RebuildEndpointStats() + } + }() + wg.Wait() + s.RebuildEndpointStats() + if got := s.UsageForEndpoint("v", "e"); got.OutputTokens != 100 { + t.Fatalf("aggregated output = %d, want 100", got.OutputTokens) + } +} + +// TestIssue3086_GuardedMethodsSemantics: the new appends behave like plain +// slice appends and the legacy zero-value paths still hold. +func TestIssue3086_GuardedMethodsSemantics(t *testing.T) { + s := &Session{} + s.AddUsageHistoryEntry(UsageEntry{Vendor: "v", Endpoint: "e", Usage: provider.TokenUsage{InputTokens: 3}}) + s.AppendMetricEvent(metrics.MetricEvent{Vendor: "v", Endpoint: "e"}) + if len(s.UsageHistory) != 1 || len(s.Metrics) != 1 { + t.Fatalf("guarded append counts: usage=%d metrics=%d, want 1/1", len(s.UsageHistory), len(s.Metrics)) + } + if got := s.UsageForEndpoint("v", "e"); got.InputTokens != 3 { + t.Fatalf("lazy rebuild after guarded append: %v", got) + } + // Nil-safe, matching package conventions. + var nilS *Session + nilS.AddUsageHistoryEntry(UsageEntry{}) + nilS.AppendMetricEvent(metrics.MetricEvent{}) +} diff --git a/internal/tui/commands_slash_info.go b/internal/tui/commands_slash_info.go index 93895531d..39c58ceff 100644 --- a/internal/tui/commands_slash_info.go +++ b/internal/tui/commands_slash_info.go @@ -244,7 +244,10 @@ func (m *Model) exportTraceSession(id string, otelFormat bool) tea.Cmd { endpoint = m.session.Endpoint model = m.session.Model createdAt = m.session.CreatedAt - events = m.session.Metrics + // #3086: copy under the append lock - the bare slice reference + // captured here crossed into the command goroutine while agent + // callbacks appended it. + events = m.session.MetricsSnapshot() } store := m.sessionStore tr := m.t diff --git a/internal/tui/model.go b/internal/tui/model.go index 3ab4882b7..720c465dc 100644 --- a/internal/tui/model.go +++ b/internal/tui/model.go @@ -1092,7 +1092,10 @@ func (m *Model) recordSessionUsage(usage provider.TokenUsage, source string) { // Keep in-memory UsageHistory in sync with disk. AppendUsageEntry // only writes to the JSONL file; without this append, /cost reads // stale data that only reflects what was loaded at startup. - ses.UsageHistory = append(ses.UsageHistory, entry) + // #3086: append via the lock-guarded method - a bare slice append + // here raced the endpoint-stats rebuild readers ranging the same + // slice from the UI render path. + ses.AddUsageHistoryEntry(entry) mu.Unlock() // Disk I/O is async — the TUI event loop must not block on fsync. @@ -1132,7 +1135,7 @@ func (m *Model) recordSessionMetric(ev metrics.MetricEvent) { ev.Vendor = m.session.Vendor ev.Endpoint = m.session.Endpoint } - m.session.Metrics = append(m.session.Metrics, ev) + m.session.AppendMetricEvent(ev) // #3086: lock-guarded append (was a bare slice append) m.session.AppendMetricForEndpoint(m.session.Vendor, m.session.Endpoint, ev) m.session.UpdatedAt = time.Now() ses := m.session diff --git a/internal/tui/view_status.go b/internal/tui/view_status.go index e2a3aa351..d400a4326 100644 --- a/internal/tui/view_status.go +++ b/internal/tui/view_status.go @@ -402,12 +402,16 @@ func (m *Model) sessionCostSnapshot() (float64, provider.TokenUsage) { c := m.costCache c.mu.Lock() defer c.mu.Unlock() - if c.sid == m.session.ID && c.n == len(m.session.UsageHistory) { + // #3086: read UsageHistory through the lock-guarded snapshot - the + // usage callbacks append it from agent-stream goroutines while this + // runs on the View thread every frame. + history := m.session.UsageHistorySnapshot() + if c.sid == m.session.ID && c.n == len(history) { return c.cost, c.usage } var total float64 var agg provider.TokenUsage - for _, entry := range m.session.UsageHistory { + for _, entry := range history { agg = agg.Add(entry.Usage) if entry.Usage.Total() == 0 { continue @@ -428,7 +432,7 @@ func (m *Model) sessionCostSnapshot() (float64, provider.TokenUsage) { float64(u.CacheWrite)*rate.CacheWritePerM/1e6 } c.sid = m.session.ID - c.n = len(m.session.UsageHistory) + c.n = len(history) c.cost = total c.usage = agg return total, agg @@ -447,7 +451,7 @@ func (m Model) sessionCostHint() string { } // Fallback: aggregate from UsageHistory if session-level total is still 0 // (aggregation is cached — see sessionCostSnapshot). - if usage.Total() == 0 && len(m.session.UsageHistory) > 0 { + if usage.Total() == 0 && len(m.session.UsageHistorySnapshot()) > 0 { _, usage = m.sessionCostSnapshot() } totalTokens := usage.Total()