mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
perf(stats): stop retaining and rewriting the raw stats event log (#35852)
* perf(stats): stop retaining and rewriting the raw stats event log (#35544) The collector kept up to 10K raw events (~203 B each since #14657 switched meta to pane keys and added worktree ids) and re-serialized the whole ~2 MB file every 5 s. No consumer reads events; persist aggregates only, keep the v1 file shape with an empty events array, and compact legacy files on load. * test(stats): use a small event burst for the retained-size bound
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
import { mkdtempSync, readdirSync, readFileSync, rmSync } from 'node:fs'
|
||||
import { mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'
|
||||
import type * as FsPromises from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
@@ -123,43 +123,49 @@ describe('StatsCollector async debounced save', () => {
|
||||
expect(JSON.parse(readFileSync(statsPath(), 'utf-8')).aggregates.totalAgentsSpawned).toBe(5)
|
||||
})
|
||||
|
||||
it('bounds retained events while a stalled write prevents serialization', async () => {
|
||||
it('persists aggregates without raw events and compacts a legacy event log', async () => {
|
||||
vi.useFakeTimers()
|
||||
const legacyEvents = Array.from({ length: 10_000 }, (_, index) => ({
|
||||
type: 'agent_start',
|
||||
at: index,
|
||||
worktreeId: 'repo::/Users/someone/workspaces/feature',
|
||||
meta: { sessionKey: 'tab:leaf' }
|
||||
}))
|
||||
writeFileSync(
|
||||
statsPath(),
|
||||
JSON.stringify({
|
||||
schemaVersion: 1,
|
||||
events: legacyEvents,
|
||||
aggregates: {
|
||||
totalAgentsSpawned: 7,
|
||||
totalPRsCreated: 0,
|
||||
totalAgentTimeMs: 0,
|
||||
countedPRs: [],
|
||||
firstEventAt: 0
|
||||
}
|
||||
})
|
||||
)
|
||||
const { StatsCollector, initStatsPath } = await importCollector()
|
||||
initStatsPath(userDataDir)
|
||||
const collector = new StatsCollector()
|
||||
|
||||
gate.blocked = true
|
||||
collector.record({ type: 'agent_start', at: 0 })
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
await vi.waitFor(() => expect(gate.writeFileCalls).toBe(1))
|
||||
await vi.waitFor(() =>
|
||||
expect(JSON.parse(readFileSync(statsPath(), 'utf-8')).events).toEqual([])
|
||||
)
|
||||
|
||||
const expectedEvents = Array.from({ length: 10_000 }, (_, index) => ({
|
||||
type: 'agent_start',
|
||||
at: index + 10_001
|
||||
}))
|
||||
try {
|
||||
for (let at = 1; at <= 20_000; at += 1) {
|
||||
collector.record({ type: 'agent_start', at })
|
||||
if (at % 5_000 === 0) {
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
expect(collector['events'].length).toBeLessThanOrEqual(10_000)
|
||||
}
|
||||
}
|
||||
expect(gate.writeFileCalls).toBe(1)
|
||||
expect(collector['events']).toEqual(expectedEvents)
|
||||
} finally {
|
||||
const flushed = collector.flushAsync()
|
||||
gate.blocked = false
|
||||
gate.waiters.splice(0).forEach((resolve) => resolve())
|
||||
await flushed
|
||||
for (let at = 1; at <= 100; at += 1) {
|
||||
collector.onAgentStart(`pane-${at}`, at, undefined, 'repo::/Users/someone/workspaces/feature')
|
||||
collector.onAgentStop(`pane-${at}`, at + 10)
|
||||
}
|
||||
await collector.flushAsync()
|
||||
|
||||
const persisted = JSON.parse(readFileSync(statsPath(), 'utf-8'))
|
||||
expect(persisted.events).toEqual(expectedEvents)
|
||||
expect(persisted.aggregates).toMatchObject({
|
||||
totalAgentsSpawned: 20_001,
|
||||
firstEventAt: 0
|
||||
const raw = readFileSync(statsPath(), 'utf-8')
|
||||
expect(raw.length).toBeLessThan(1_000)
|
||||
expect(JSON.parse(raw)).toMatchObject({
|
||||
schemaVersion: 1,
|
||||
events: [],
|
||||
aggregates: { totalAgentsSpawned: 107, totalAgentTimeMs: 1_000, firstEventAt: 0 }
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
+11
-16
@@ -1,11 +1,10 @@
|
||||
import { join } from 'node:path'
|
||||
import { getAppEnvironment } from '../../shared/app-environment'
|
||||
import type { StatsSummary } from '../../shared/process-stats-types'
|
||||
import type { StatsEvent, StatsAggregates } from './types'
|
||||
import type { StatsEvent, StatsAggregates, StatsFile } from './types'
|
||||
import { loadStatsFile, STATS_SCHEMA_VERSION } from './stats-file-loader'
|
||||
import { AtomicSnapshotWriter } from '../persistence/atomic-snapshot-writer'
|
||||
|
||||
const MAX_EVENTS = 10_000
|
||||
// Bound persistent PR dedup URLs at 2000 (~100KB) across long-lived installs.
|
||||
const MAX_COUNTED_PRS = 2_000
|
||||
// Stats are infrequent and UI-insensitive; batch writes more loosely than the main Store.
|
||||
@@ -25,7 +24,6 @@ function getStatsFile(): string {
|
||||
}
|
||||
|
||||
export class StatsCollector {
|
||||
private events: StatsEvent[]
|
||||
private aggregates: StatsAggregates
|
||||
private liveAgents = new Map<string, number>()
|
||||
private writeTimer: ReturnType<typeof setTimeout> | null = null
|
||||
@@ -38,8 +36,11 @@ export class StatsCollector {
|
||||
|
||||
constructor() {
|
||||
const data = loadStatsFile(getStatsFile())
|
||||
this.events = data.events
|
||||
this.aggregates = data.aggregates
|
||||
// Compact pre-#35544 files so later boots stop parsing the legacy event log.
|
||||
if (Array.isArray(data.events) && data.events.length > 0) {
|
||||
this.scheduleSave()
|
||||
}
|
||||
}
|
||||
|
||||
onAgentStarted(listener: (totalAgentsSpawned: number) => void): () => void {
|
||||
@@ -53,12 +54,8 @@ export class StatsCollector {
|
||||
return this.aggregates.totalAgentsSpawned
|
||||
}
|
||||
|
||||
/** Folds the event into aggregates; no consumer reads raw events, so they are not retained (#35544). */
|
||||
record(event: StatsEvent): void {
|
||||
this.events.push(event)
|
||||
// A stalled async write must not defer the in-memory retention limit.
|
||||
if (this.events.length > MAX_EVENTS) {
|
||||
this.events.splice(0, this.events.length - MAX_EVENTS)
|
||||
}
|
||||
this.updateAggregates(event)
|
||||
this.scheduleSave()
|
||||
}
|
||||
@@ -179,7 +176,6 @@ export class StatsCollector {
|
||||
}
|
||||
this.writeTimer = setTimeout(() => {
|
||||
this.writeTimer = null
|
||||
// Write ~900KB snapshots off-thread to avoid main stalls every five seconds.
|
||||
void this.enqueueWrite().catch((err) => {
|
||||
console.error('[stats] Failed to write stats:', err)
|
||||
})
|
||||
@@ -194,14 +190,13 @@ export class StatsCollector {
|
||||
}
|
||||
|
||||
private serialize(): string {
|
||||
if (this.events.length > MAX_EVENTS) {
|
||||
this.events = this.events.slice(-MAX_EVENTS)
|
||||
}
|
||||
return JSON.stringify({
|
||||
// Empty events keeps the v1 file shape for older builds and external readers.
|
||||
const file: StatsFile = {
|
||||
schemaVersion: STATS_SCHEMA_VERSION,
|
||||
events: this.events,
|
||||
events: [],
|
||||
aggregates: this.aggregates
|
||||
})
|
||||
}
|
||||
return JSON.stringify(file)
|
||||
}
|
||||
|
||||
private enqueueWrite(): Promise<void> {
|
||||
|
||||
@@ -6,7 +6,6 @@ export type StatsEventType =
|
||||
export type StatsEvent = {
|
||||
type: StatsEventType
|
||||
at: number // Unix epoch milliseconds.
|
||||
// Keep repo/workspace context for future breakdowns even when current aggregates omit it.
|
||||
repoId?: string
|
||||
worktreeId?: string
|
||||
meta?: Record<string, string | number>
|
||||
@@ -18,12 +17,13 @@ export type StatsAggregates = {
|
||||
totalAgentTimeMs: number
|
||||
// PR URLs deduplicate across restarts; collector caps this at 2000.
|
||||
countedPRs: string[]
|
||||
// Persist first event time independently: trimming the 10K event log must not move tracking-since forward.
|
||||
// Persisted because raw events are not retained.
|
||||
firstEventAt: number | null
|
||||
}
|
||||
|
||||
export type StatsFile = {
|
||||
schemaVersion: number
|
||||
// Always written empty since #35544; older files may still carry events, which load ignores.
|
||||
events: StatsEvent[]
|
||||
aggregates: StatsAggregates
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user