mirror of
https://github.com/wu736139669/hapi.git
synced 2026-10-31 23:31:39 +00:00
fix(hub,cli): run OpenCode usage scans off the process event loop
The 5-minute reconciliation read the multi-GB OpenCode store on the hub's event loop with bun:sqlite; a cold scan froze every request for ~88s and stalled each machine's runner the same way. - scan in a spawned `hapi usage-scan` child process (stdin/stdout JSON) - refresh on demand from the usage page (debounced) + a daily fallback - a runner whose hub is localhost skips the store the hub already owns - machine scans run every 30min and read the message table sequentially in one pass instead of per-session indexed lookups
This commit is contained in:
@@ -21,6 +21,7 @@ import { notifyCommand } from './notify'
|
||||
import { hubCommand } from './hub'
|
||||
import { pingPeerCommand } from './pingPeer'
|
||||
import { inspectPeerCommand } from './inspectPeer'
|
||||
import { usageScanCommand } from './usageScan'
|
||||
import type { CommandContext, CommandDefinition } from './types'
|
||||
|
||||
// Gemini CLI was sunset (Google stopped serving the consumer Gemini CLI on
|
||||
@@ -63,7 +64,10 @@ const COMMANDS: CommandDefinition[] = [
|
||||
runnerCommand,
|
||||
notifyCommand,
|
||||
pingPeerCommand,
|
||||
inspectPeerCommand
|
||||
inspectPeerCommand,
|
||||
// Internal: spawned by the hub to scan the local OpenCode store without
|
||||
// blocking its event loop (see hub/src/sync/usageReconciliation.ts).
|
||||
usageScanCommand
|
||||
]
|
||||
|
||||
const commandMap = new Map<string, CommandDefinition>()
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
import { Database } from 'bun:sqlite'
|
||||
import { describe, expect, it } from 'bun:test'
|
||||
import { mkdtempSync, rmSync } from 'node:fs'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { parseUsageScanRequest, runUsageScan } from './usageScan'
|
||||
|
||||
function createOpencodeStore(): string {
|
||||
const dir = mkdtempSync(join(tmpdir(), 'hapi-usage-scan-'))
|
||||
const dbPath = join(dir, 'opencode.db')
|
||||
const db = new Database(dbPath, { create: true })
|
||||
db.exec(`
|
||||
CREATE TABLE message (
|
||||
id TEXT PRIMARY KEY,
|
||||
session_id TEXT NOT NULL,
|
||||
time_created INTEGER NOT NULL,
|
||||
time_updated INTEGER NOT NULL,
|
||||
data TEXT NOT NULL
|
||||
)
|
||||
`)
|
||||
db.prepare('INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)').run(
|
||||
'msg-1',
|
||||
'ses_scan',
|
||||
Date.parse('2026-09-17T10:00:00'),
|
||||
Date.parse('2026-09-17T10:00:00'),
|
||||
JSON.stringify({
|
||||
role: 'assistant',
|
||||
providerID: 'opencode-go',
|
||||
modelID: 'deepseek-v4.1-flash',
|
||||
time: { created: Date.parse('2026-09-17T10:00:00') },
|
||||
tokens: { input: 7, output: 3, reasoning: 1, cache: { read: 5, write: 2 } }
|
||||
})
|
||||
)
|
||||
db.close()
|
||||
return dbPath
|
||||
}
|
||||
|
||||
describe('usage-scan command', () => {
|
||||
it('returns the hub-facing payload for the requested sessions', async () => {
|
||||
const dbPath = createOpencodeStore()
|
||||
try {
|
||||
const payload = await runUsageScan({ dbPath, sessionIds: ['ses_scan', 'ses_missing'] })
|
||||
expect(payload.results.ses_scan.messages).toBe(1)
|
||||
expect(payload.results.ses_scan.rows).toEqual([
|
||||
expect.objectContaining({
|
||||
day: '2026-09-17',
|
||||
model: 'opencode-go/deepseek-v4.1-flash',
|
||||
inputTokens: 14,
|
||||
outputTokens: 4,
|
||||
requests: 1
|
||||
})
|
||||
])
|
||||
expect(payload.results.ses_missing).toEqual({ rows: [], messages: 0 })
|
||||
} finally {
|
||||
rmSync(join(dbPath, '..'), { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('listAll scans every session present in the store', async () => {
|
||||
const dbPath = createOpencodeStore()
|
||||
try {
|
||||
const payload = await runUsageScan({ dbPath, listAll: true })
|
||||
expect(Object.keys(payload.results)).toEqual(['ses_scan'])
|
||||
expect(payload.results.ses_scan.messages).toBe(1)
|
||||
// The sequential all-sessions pass must match the indexed
|
||||
// per-session reads (this path feeds the runner's reports).
|
||||
const perSession = await runUsageScan({ dbPath, sessionIds: ['ses_scan'] })
|
||||
expect(payload.results.ses_scan).toEqual(perSession.results.ses_scan)
|
||||
} finally {
|
||||
rmSync(join(dbPath, '..'), { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects malformed requests', () => {
|
||||
expect(() => parseUsageScanRequest('not json')).toThrow()
|
||||
expect(() => parseUsageScanRequest('{"sessionIds":[]}')).toThrow()
|
||||
expect(parseUsageScanRequest('{"dbPath":"/x","sessionIds":["ses_a",3,""]}')).toEqual({
|
||||
dbPath: '/x',
|
||||
sessionIds: ['ses_a']
|
||||
})
|
||||
expect(parseUsageScanRequest('{"dbPath":"/x","listAll":true}')).toEqual({
|
||||
dbPath: '/x',
|
||||
listAll: true
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,62 @@
|
||||
import { readFileSync } from 'node:fs'
|
||||
import type { CommandDefinition } from './types'
|
||||
|
||||
/**
|
||||
* Internal command: aggregate OpenCode token usage.
|
||||
*
|
||||
* The hub and the runner spawn `hapi usage-scan` as a child process so the
|
||||
* synchronous bun:sqlite reads over the multi-GB OpenCode store never block
|
||||
* their event loops (a cold scan used to freeze the hub for a minute or
|
||||
* more). The request arrives as JSON on stdin, the result leaves as JSON on
|
||||
* stdout: `{ dbPath, sessionIds }` for known sessions, or
|
||||
* `{ dbPath, listAll: true }` to scan every session found in the store.
|
||||
*/
|
||||
|
||||
export type UsageScanRequest = {
|
||||
dbPath: string
|
||||
sessionIds?: string[]
|
||||
listAll?: boolean
|
||||
}
|
||||
|
||||
function readStdin(): string {
|
||||
// fd 0 works for both pipe and file-backed stdin; Bun's async iteration
|
||||
// over a file-backed stdin yields nothing.
|
||||
return readFileSync(0, 'utf8')
|
||||
}
|
||||
|
||||
export function parseUsageScanRequest(raw: string): UsageScanRequest {
|
||||
const parsed = JSON.parse(raw) as { dbPath?: unknown; sessionIds?: unknown; listAll?: unknown }
|
||||
if (typeof parsed.dbPath !== 'string') {
|
||||
throw new Error('usage-scan expects { dbPath: string, sessionIds?: string[], listAll?: boolean } on stdin')
|
||||
}
|
||||
if (parsed.listAll === true) {
|
||||
return { dbPath: parsed.dbPath, listAll: true }
|
||||
}
|
||||
if (!Array.isArray(parsed.sessionIds)) {
|
||||
throw new Error('usage-scan expects sessionIds or listAll: true on stdin')
|
||||
}
|
||||
const sessionIds = parsed.sessionIds.filter((id): id is string => typeof id === 'string' && id.length > 0)
|
||||
return { dbPath: parsed.dbPath, sessionIds }
|
||||
}
|
||||
|
||||
/** Payload shape the hub/runner parse (see hub/src/sync/usageReconciliation.ts). */
|
||||
export async function runUsageScan(request: UsageScanRequest): Promise<{ results: Record<string, { rows: unknown[]; messages: number }> }> {
|
||||
const { readOpencodeUsageForAllSessions, readOpencodeUsageForSessions } = await import('@hapi/protocol/opencodeUsage')
|
||||
const usage = request.listAll === true
|
||||
? await readOpencodeUsageForAllSessions(request.dbPath)
|
||||
: await readOpencodeUsageForSessions(request.dbPath, request.sessionIds ?? [])
|
||||
const results: Record<string, { rows: unknown[]; messages: number }> = {}
|
||||
for (const [sessionId, entry] of usage) {
|
||||
results[sessionId] = { rows: entry.rows, messages: entry.messages }
|
||||
}
|
||||
return { results }
|
||||
}
|
||||
|
||||
export const usageScanCommand: CommandDefinition = {
|
||||
name: 'usage-scan',
|
||||
requiresRuntimeAssets: false,
|
||||
run: async () => {
|
||||
const request = parseUsageScanRequest(readStdin())
|
||||
process.stdout.write(JSON.stringify(await runUsageScan(request)))
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import { listOpencodeSessionIds, readOpencodeUsageForSessions, resolveOpencodeDbPath } from '@hapi/protocol/opencodeUsage'
|
||||
import type { OpencodeUsageSessionReport } from '@hapi/protocol/usage'
|
||||
import type { OpencodeUsageReportRow, OpencodeUsageSessionReport } from '@hapi/protocol/usage'
|
||||
import { logger } from '@/ui/logger'
|
||||
|
||||
/**
|
||||
@@ -12,12 +12,18 @@ import { logger } from '@/ui/logger'
|
||||
* their reconciliation rows, which makes repeated or concurrent reports
|
||||
* idempotent (see hub/src/sync/usageReconciliation.ts).
|
||||
*
|
||||
* The scan runs in a spawned `hapi usage-scan` child process: the OpenCode
|
||||
* store is multi-GB in practice and `bun:sqlite` is synchronous, so an
|
||||
* in-process scan would stall this runner's event loop — machine RPCs
|
||||
* (spawn, model listing) queue behind it and can time out.
|
||||
*
|
||||
* The live chat pipeline is not involved at all — a report never becomes a
|
||||
* timeline message.
|
||||
*/
|
||||
|
||||
const SCAN_INTERVAL_MS = 5 * 60 * 1000
|
||||
const SCAN_INTERVAL_MS = 30 * 60 * 1000
|
||||
const INITIAL_DELAY_MS = 20_000
|
||||
const SCAN_TIMEOUT_MS = 2 * 60 * 1000
|
||||
|
||||
export type OpencodeUsageScannerOptions = {
|
||||
/** Best-effort delivery; the next scan re-sends on failure. */
|
||||
@@ -35,15 +41,82 @@ export type OpencodeUsageScanner = {
|
||||
runNow: () => void
|
||||
}
|
||||
|
||||
type ScannedUsage = Map<string, { rows: OpencodeUsageReportRow[]; messages: number }>
|
||||
|
||||
async function spawnUsageScan(request: { dbPath: string; listAll: true }): Promise<ScannedUsage | null> {
|
||||
if (typeof Bun === 'undefined') return null
|
||||
const executable = process.env.HAPI_CLI_EXECUTABLE?.trim() || process.execPath
|
||||
if (!executable) return null
|
||||
try {
|
||||
const proc = Bun.spawn({
|
||||
cmd: [executable, 'usage-scan'],
|
||||
stdin: 'pipe',
|
||||
stdout: 'pipe',
|
||||
stderr: 'ignore'
|
||||
})
|
||||
const timer = setTimeout(() => {
|
||||
try {
|
||||
proc.kill()
|
||||
} catch {
|
||||
}
|
||||
}, SCAN_TIMEOUT_MS)
|
||||
try {
|
||||
proc.stdin.write(JSON.stringify(request))
|
||||
proc.stdin.end()
|
||||
const [exitCode, stdout] = await Promise.all([proc.exited, new Response(proc.stdout).text()])
|
||||
if (exitCode !== 0 || stdout.trim().length === 0) return null
|
||||
const parsed = JSON.parse(stdout) as { results?: unknown }
|
||||
const results = parsed.results
|
||||
if (results === null || typeof results !== 'object' || Array.isArray(results)) return null
|
||||
const usage: ScannedUsage = new Map()
|
||||
for (const [sessionId, value] of Object.entries(results as Record<string, unknown>)) {
|
||||
const entry = value as { rows?: unknown; messages?: unknown } | null
|
||||
if (!entry || typeof entry !== 'object' || !Array.isArray(entry.rows)) return null
|
||||
usage.set(sessionId, {
|
||||
rows: entry.rows as OpencodeUsageReportRow[],
|
||||
messages: typeof entry.messages === 'number' ? entry.messages : entry.rows.length
|
||||
})
|
||||
}
|
||||
return usage
|
||||
} finally {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/** Probe once whether the current executable can actually run `usage-scan`. */
|
||||
let cliScanAvailablePromise: Promise<boolean> | null = null
|
||||
|
||||
function cliScanAvailable(): Promise<boolean> {
|
||||
cliScanAvailablePromise ??= (async () => {
|
||||
const probe = await spawnUsageScan({ dbPath: '', listAll: true })
|
||||
return probe !== null
|
||||
})()
|
||||
return cliScanAvailablePromise
|
||||
}
|
||||
|
||||
async function scanOpencodeStore(dbPath: string): Promise<ScannedUsage> {
|
||||
if (await cliScanAvailable()) {
|
||||
const scanned = await spawnUsageScan({ dbPath, listAll: true })
|
||||
if (scanned) return scanned
|
||||
throw new Error('opencode usage scan failed')
|
||||
}
|
||||
// Dev/test fallback (no hapi binary to spawn): read in-process. This is
|
||||
// deliberately not used in production — the multi-GB scan must never run
|
||||
// on this process's event loop.
|
||||
const opencodeSessionIds = await listOpencodeSessionIds(dbPath)
|
||||
if (opencodeSessionIds.length === 0) return new Map()
|
||||
return await readOpencodeUsageForSessions(dbPath, opencodeSessionIds)
|
||||
}
|
||||
|
||||
/** Collect snapshot rows for every session present in the OpenCode store. */
|
||||
export async function collectOpencodeUsageReports(dbPath: string): Promise<OpencodeUsageSessionReport[]> {
|
||||
const opencodeSessionIds = await listOpencodeSessionIds(dbPath)
|
||||
if (opencodeSessionIds.length === 0) return []
|
||||
const usageBySession = await readOpencodeUsageForSessions(dbPath, opencodeSessionIds)
|
||||
const usageBySession = await scanOpencodeStore(dbPath)
|
||||
const reports: OpencodeUsageSessionReport[] = []
|
||||
for (const opencodeSessionId of opencodeSessionIds) {
|
||||
const usage = usageBySession.get(opencodeSessionId)
|
||||
if (!usage || usage.rows.length === 0) continue
|
||||
for (const [opencodeSessionId, usage] of usageBySession) {
|
||||
if (usage.rows.length === 0) continue
|
||||
reports.push({ opencodeSessionId, rows: usage.rows })
|
||||
}
|
||||
return reports
|
||||
|
||||
@@ -1166,8 +1166,11 @@ export async function startRunner(options: { workspaceRoots?: string[] } = {}):
|
||||
apiMachine.connect();
|
||||
// Report this machine's OpenCode usage snapshots so the hub's token
|
||||
// dashboard does not under-count tool-loop turns on non-hub machines.
|
||||
// The scan reads the local OpenCode store; best-effort, never blocks.
|
||||
const opencodeUsageScanner = startOpencodeUsageScanner({
|
||||
// A runner whose hub is localhost would only rescan the store the hub
|
||||
// already reconciles itself (see hub/src/sync/usageReconciliation.ts),
|
||||
// so skip it there. The scan itself runs in a child process.
|
||||
const hubIsLocal = /^https?:\/\/(localhost|127\.0\.0\.1|\[::1\])(?::|\/|$)/i.test(configuration.apiUrl);
|
||||
const opencodeUsageScanner = hubIsLocal ? null : startOpencodeUsageScanner({
|
||||
report: (sessions) => apiMachine.reportOpencodeUsage(sessions)
|
||||
});
|
||||
scheduleCursorModelsPrewarm();
|
||||
@@ -1435,7 +1438,7 @@ export async function startRunner(options: { workspaceRoots?: string[] } = {}):
|
||||
clearInterval(restartOnStaleVersionAndHeartbeat);
|
||||
logger.debug('[RUNNER RUN] Health check interval cleared');
|
||||
}
|
||||
opencodeUsageScanner.stop();
|
||||
opencodeUsageScanner?.stop();
|
||||
|
||||
// Update runner state before shutting down
|
||||
await apiMachine.updateRunnerState((state: RunnerState | null) => ({
|
||||
|
||||
+5
-1
@@ -413,7 +413,11 @@ export async function startHub(options: StartHubOptions = {}): Promise<HubInstan
|
||||
corsOrigins,
|
||||
relayMode: relayFlag.enabled,
|
||||
officialWebUrl,
|
||||
dataDir: config.dataDir
|
||||
dataDir: config.dataDir,
|
||||
// The usage page refreshes OpenCode snapshots on open (debounced).
|
||||
refreshUsage: async () => {
|
||||
await usageReconciliation?.refresh()
|
||||
}
|
||||
})
|
||||
|
||||
// Start the bot if configured
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
import { existsSync } from 'node:fs'
|
||||
import { readOpencodeUsageForSessions, resolveOpencodeDbPath } from '@hapi/protocol/opencodeUsage'
|
||||
import {
|
||||
readOpencodeUsageForSessions,
|
||||
resolveOpencodeDbPath,
|
||||
type OpencodeSessionUsage
|
||||
} from '@hapi/protocol/opencodeUsage'
|
||||
import type { Store } from '../store'
|
||||
import type { ReconciledUsageRow } from '../store/usage'
|
||||
import type { UsageSource } from '../store/usageSources'
|
||||
@@ -21,10 +25,18 @@ import type { UsageSource } from '../store/usageSources'
|
||||
* hub machine. Every other machine's runner reports the same row shape over
|
||||
* the `opencode-usage-report` socket event (see `@hapi/protocol/opencodeUsage`
|
||||
* and cli/src/runner/opencodeUsageScanner.ts).
|
||||
*
|
||||
* The scan itself runs in a spawned `hapi usage-scan` child process: the
|
||||
* OpenCode store is multi-GB in practice, and `bun:sqlite` is synchronous, so
|
||||
* an in-process scan would block the hub's event loop for the whole read
|
||||
* (observed: 88s freezes under memory pressure). The hub only ever applies the
|
||||
* small aggregated result.
|
||||
*/
|
||||
|
||||
const RECONCILE_INTERVAL_MS = 5 * 60 * 1000
|
||||
const INITIAL_DELAY_MS = 15_000
|
||||
const FALLBACK_INTERVAL_MS = 24 * 60 * 60 * 1000
|
||||
const INITIAL_DELAY_MS = 60_000
|
||||
const REFRESH_MAX_AGE_MS = 5 * 60 * 1000
|
||||
const SCAN_TIMEOUT_MS = 2 * 60 * 1000
|
||||
|
||||
export type UsageReconciliationResult = {
|
||||
dbPath: string | null
|
||||
@@ -33,6 +45,15 @@ export type UsageReconciliationResult = {
|
||||
messages: number
|
||||
}
|
||||
|
||||
export type OpencodeUsageScanRequest = {
|
||||
dbPath: string
|
||||
sessionIds: string[]
|
||||
}
|
||||
|
||||
export type OpencodeUsageScan = (
|
||||
request: OpencodeUsageScanRequest
|
||||
) => Promise<Map<string, OpencodeSessionUsage>>
|
||||
|
||||
function asRecord(value: unknown): Record<string, unknown> | null {
|
||||
return value !== null && typeof value === 'object' && !Array.isArray(value)
|
||||
? value as Record<string, unknown>
|
||||
@@ -43,10 +64,87 @@ function asString(value: unknown): string | null {
|
||||
return typeof value === 'string' && value.trim().length > 0 ? value.trim() : null
|
||||
}
|
||||
|
||||
function parseScanOutput(stdout: string): Map<string, OpencodeSessionUsage> | null {
|
||||
const parsed = JSON.parse(stdout) as { results?: unknown }
|
||||
const results = asRecord(parsed.results)
|
||||
if (!results) return null
|
||||
const usage = new Map<string, OpencodeSessionUsage>()
|
||||
for (const [sessionId, value] of Object.entries(results)) {
|
||||
const entry = asRecord(value)
|
||||
if (!entry || !Array.isArray(entry.rows) || typeof entry.messages !== 'number') return null
|
||||
usage.set(sessionId, { rows: entry.rows as OpencodeSessionUsage['rows'], messages: entry.messages })
|
||||
}
|
||||
return usage
|
||||
}
|
||||
|
||||
async function spawnUsageScanCli(request: OpencodeUsageScanRequest): Promise<string | null> {
|
||||
const executable = process.env.HAPI_CLI_EXECUTABLE?.trim() || process.execPath
|
||||
if (!executable) return null
|
||||
const proc = Bun.spawn({
|
||||
cmd: [executable, 'usage-scan'],
|
||||
stdin: 'pipe',
|
||||
stdout: 'pipe',
|
||||
stderr: 'ignore'
|
||||
})
|
||||
const timer = setTimeout(() => {
|
||||
try {
|
||||
proc.kill()
|
||||
} catch {
|
||||
}
|
||||
}, SCAN_TIMEOUT_MS)
|
||||
try {
|
||||
proc.stdin.write(JSON.stringify(request))
|
||||
proc.stdin.end()
|
||||
const [exitCode, stdout] = await Promise.all([proc.exited, new Response(proc.stdout).text()])
|
||||
if (exitCode !== 0 || stdout.trim().length === 0) return null
|
||||
return stdout
|
||||
} catch {
|
||||
return null
|
||||
} finally {
|
||||
clearTimeout(timer)
|
||||
}
|
||||
}
|
||||
|
||||
/** Probe once whether the current executable can actually run `usage-scan`. */
|
||||
let scanModePromise: Promise<'cli' | 'in-process'> | null = null
|
||||
|
||||
function resolveScanMode(): Promise<'cli' | 'in-process'> {
|
||||
scanModePromise ??= (async () => {
|
||||
try {
|
||||
// Empty request: the CLI answers immediately, while a non-hapi
|
||||
// executable (e.g. `bun` under tests/dev) fails fast.
|
||||
const stdout = await spawnUsageScanCli({ dbPath: '', sessionIds: [] })
|
||||
return stdout !== null && parseScanOutput(stdout) !== null ? 'cli' : 'in-process'
|
||||
} catch {
|
||||
return 'in-process'
|
||||
}
|
||||
})()
|
||||
return scanModePromise
|
||||
}
|
||||
|
||||
/**
|
||||
* Default scan: a `hapi usage-scan` child process when available. A transient
|
||||
* child failure is reported as an error instead of falling back to an
|
||||
* in-process read — the whole point is to keep the read off this process.
|
||||
* Only when the executable cannot run the command at all (dev/test) do we
|
||||
* read in-process; those environments are not the multi-GB-store case.
|
||||
*/
|
||||
async function defaultOpencodeUsageScan(
|
||||
request: OpencodeUsageScanRequest
|
||||
): Promise<Map<string, OpencodeSessionUsage>> {
|
||||
if (await resolveScanMode() === 'cli') {
|
||||
const stdout = await spawnUsageScanCli(request)
|
||||
const usage = stdout === null ? null : parseScanOutput(stdout)
|
||||
if (usage) return usage
|
||||
throw new Error('opencode usage scan failed')
|
||||
}
|
||||
return await readOpencodeUsageForSessions(request.dbPath, request.sessionIds)
|
||||
}
|
||||
|
||||
/** Re-read every known OpenCode session from the local store and refresh snapshots. */
|
||||
export async function reconcileOpencodeUsage(
|
||||
store: Store,
|
||||
options?: { dbPath?: string | null; now?: number }
|
||||
options?: { dbPath?: string | null; now?: number; scan?: OpencodeUsageScan }
|
||||
): Promise<UsageReconciliationResult> {
|
||||
const dbPath = options?.dbPath === undefined ? resolveOpencodeDbPath() : options.dbPath
|
||||
const now = options?.now ?? Date.now()
|
||||
@@ -68,7 +166,11 @@ export async function reconcileOpencodeUsage(
|
||||
if (!targets.has(key)) targets.set(key, source)
|
||||
}
|
||||
|
||||
const usageBySession = await readOpencodeUsageForSessions(dbPath, [...targets.values()].map((target) => target.nativeSessionId))
|
||||
const scan = options?.scan ?? defaultOpencodeUsageScan
|
||||
const usageBySession = await scan({
|
||||
dbPath,
|
||||
sessionIds: [...targets.values()].map((target) => target.nativeSessionId)
|
||||
})
|
||||
let sessionsTouched = 0
|
||||
let rowsTotal = 0
|
||||
let messagesTotal = 0
|
||||
@@ -95,19 +197,30 @@ export async function reconcileOpencodeUsage(
|
||||
export type UsageReconciliationJob = {
|
||||
/** Idempotent. */
|
||||
stop: () => void
|
||||
/** Run one reconciliation pass immediately (used by tests). */
|
||||
runNow: () => void
|
||||
/** Run one reconciliation pass now (concurrent callers share the pass). */
|
||||
runNow: () => Promise<void>
|
||||
/** Run a pass unless the last one finished within `maxAgeMs`. */
|
||||
refresh: (maxAgeMs?: number) => Promise<void>
|
||||
}
|
||||
|
||||
/** Schedule periodic reconciliation while the hub is running. */
|
||||
/**
|
||||
* Schedule reconciliation while the hub is running.
|
||||
*
|
||||
* The dashboard refreshes on demand (`refresh`, triggered by the usage page),
|
||||
* so the timer is only a gentle safety net — one pass shortly after startup
|
||||
* and a daily fallback. The pass itself never blocks the hub (child process).
|
||||
*/
|
||||
export function startOpencodeUsageReconciliation(
|
||||
store: Store,
|
||||
options?: { intervalMs?: number; initialDelayMs?: number }
|
||||
): UsageReconciliationJob {
|
||||
let stopped = false
|
||||
const run = (): void => {
|
||||
if (stopped) return
|
||||
void reconcileOpencodeUsage(store)
|
||||
let lastCompletedAt = 0
|
||||
let inFlight: Promise<void> | null = null
|
||||
|
||||
const runPass = (): Promise<void> => {
|
||||
if (stopped) return Promise.resolve()
|
||||
const pass = reconcileOpencodeUsage(store)
|
||||
.then((result) => {
|
||||
if (result.messages > 0) {
|
||||
console.log(
|
||||
@@ -118,10 +231,24 @@ export function startOpencodeUsageReconciliation(
|
||||
.catch((error) => {
|
||||
console.warn('[UsageReconciliation] failed:', error instanceof Error ? error.message : error)
|
||||
})
|
||||
.finally(() => {
|
||||
lastCompletedAt = Date.now()
|
||||
inFlight = null
|
||||
})
|
||||
inFlight = pass
|
||||
return pass
|
||||
}
|
||||
|
||||
const initial = setTimeout(run, options?.initialDelayMs ?? INITIAL_DELAY_MS)
|
||||
const interval = setInterval(run, options?.intervalMs ?? RECONCILE_INTERVAL_MS)
|
||||
const runNow = (): Promise<void> => inFlight ?? runPass()
|
||||
|
||||
const refresh = async (maxAgeMs?: number): Promise<void> => {
|
||||
const maxAge = typeof maxAgeMs === 'number' && Number.isFinite(maxAgeMs) ? maxAgeMs : REFRESH_MAX_AGE_MS
|
||||
if (Date.now() - lastCompletedAt < maxAge) return
|
||||
await runNow()
|
||||
}
|
||||
|
||||
const initial = setTimeout(() => void runNow(), options?.initialDelayMs ?? INITIAL_DELAY_MS)
|
||||
const interval = setInterval(() => void runNow(), options?.intervalMs ?? FALLBACK_INTERVAL_MS)
|
||||
initial.unref?.()
|
||||
interval.unref?.()
|
||||
|
||||
@@ -131,6 +258,7 @@ export function startOpencodeUsageReconciliation(
|
||||
clearTimeout(initial)
|
||||
clearInterval(interval)
|
||||
},
|
||||
runNow: run
|
||||
runNow,
|
||||
refresh
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,10 +4,10 @@ import type { WebAppEnv } from '../middleware/auth'
|
||||
import type { Store } from '../../store'
|
||||
import { getUsageSummary } from '../../sync/usageService'
|
||||
|
||||
export function createUsageRoutes(store: Store): Hono<WebAppEnv> {
|
||||
export function createUsageRoutes(store: Store, refreshUsage?: () => Promise<void>): Hono<WebAppEnv> {
|
||||
const app = new Hono<WebAppEnv>()
|
||||
|
||||
app.get('/usage/summary', (c) => {
|
||||
app.get('/usage/summary', async (c) => {
|
||||
if (c.get('namespace') !== 'default') {
|
||||
return c.json({ error: 'Usage summary is only available to the hub owner' }, 403)
|
||||
}
|
||||
@@ -21,6 +21,16 @@ export function createUsageRoutes(store: Store): Hono<WebAppEnv> {
|
||||
} catch {
|
||||
return c.json({ error: 'Invalid timeZone' }, 400)
|
||||
}
|
||||
// Refresh OpenCode snapshots on demand (debounced by the job): the
|
||||
// scan runs in a child process, so waiting here never blocks the hub,
|
||||
// and the first page open after a quiet period shows fresh numbers.
|
||||
if (refreshUsage) {
|
||||
try {
|
||||
await refreshUsage()
|
||||
} catch {
|
||||
// A failed refresh leaves the previous snapshot in place.
|
||||
}
|
||||
}
|
||||
const response: UsageSummaryResponse = getUsageSummary(store, c.get('namespace'), range, timeZone)
|
||||
c.header('Cache-Control', 'no-store')
|
||||
return c.json(response)
|
||||
|
||||
@@ -239,6 +239,7 @@ function createWebApp(options: {
|
||||
relayMode?: boolean
|
||||
officialWebUrl?: string
|
||||
dataDir: string
|
||||
refreshUsage?: () => Promise<void>
|
||||
}): Hono<WebAppEnv> {
|
||||
const app = new Hono<WebAppEnv>()
|
||||
|
||||
@@ -324,7 +325,7 @@ function createWebApp(options: {
|
||||
if (teamService) {
|
||||
app.route('/api', createTeamsRoutes(teamService))
|
||||
}
|
||||
app.route('/api', createUsageRoutes(options.store))
|
||||
app.route('/api', createUsageRoutes(options.store, options.refreshUsage))
|
||||
app.route('/api', createStudioRoutes({
|
||||
store: options.store,
|
||||
getSyncEngine: options.getSyncEngine
|
||||
@@ -483,6 +484,7 @@ export async function startWebServer(options: {
|
||||
relayMode?: boolean
|
||||
officialWebUrl?: string
|
||||
dataDir: string
|
||||
refreshUsage?: () => Promise<void>
|
||||
}): Promise<BunServer<WebSocketData>> {
|
||||
const isCompiled = isBunCompiled()
|
||||
const embeddedAssetMap = isCompiled ? await loadEmbeddedAssetMap() : null
|
||||
@@ -498,7 +500,8 @@ export async function startWebServer(options: {
|
||||
embeddedAssetMap,
|
||||
dataDir: options.dataDir,
|
||||
relayMode: options.relayMode,
|
||||
officialWebUrl: options.officialWebUrl
|
||||
officialWebUrl: options.officialWebUrl,
|
||||
refreshUsage: options.refreshUsage
|
||||
})
|
||||
|
||||
const configuration = getConfiguration()
|
||||
|
||||
+70
-19
@@ -136,6 +136,32 @@ export async function listOpencodeSessionIds(dbPath: string): Promise<string[]>
|
||||
}
|
||||
}
|
||||
|
||||
/** Fold one message row's token counters into its (day, model) bucket. */
|
||||
function accumulateOpencodeUsage(
|
||||
buckets: Map<string, OpencodeUsageReportRow>,
|
||||
data: string
|
||||
): boolean {
|
||||
const usage = parseOpencodeAssistantUsage(data)
|
||||
if (!usage) return false
|
||||
const key = `${usage.day}|${usage.model}`
|
||||
const bucket = buckets.get(key) ?? {
|
||||
day: usage.day,
|
||||
model: usage.model,
|
||||
inputTokens: 0,
|
||||
outputTokens: 0,
|
||||
cacheReadTokens: 0,
|
||||
cacheCreationTokens: 0,
|
||||
requests: 0
|
||||
}
|
||||
bucket.inputTokens += usage.inputTokens
|
||||
bucket.outputTokens += usage.outputTokens
|
||||
bucket.cacheReadTokens += usage.cacheReadTokens
|
||||
bucket.cacheCreationTokens += usage.cacheCreationTokens
|
||||
bucket.requests += 1
|
||||
buckets.set(key, bucket)
|
||||
return true
|
||||
}
|
||||
|
||||
/**
|
||||
* Aggregate absolute per-(day, model) totals for the given OpenCode session
|
||||
* ids. Missing database or unknown session ids yield empty entries, so
|
||||
@@ -163,25 +189,7 @@ export async function readOpencodeUsageForSessions(
|
||||
const buckets = new Map<string, OpencodeUsageReportRow>()
|
||||
let messages = 0
|
||||
for (const row of messageRows) {
|
||||
const usage = parseOpencodeAssistantUsage(row.data)
|
||||
if (!usage) continue
|
||||
messages += 1
|
||||
const key = `${usage.day}|${usage.model}`
|
||||
const bucket = buckets.get(key) ?? {
|
||||
day: usage.day,
|
||||
model: usage.model,
|
||||
inputTokens: 0,
|
||||
outputTokens: 0,
|
||||
cacheReadTokens: 0,
|
||||
cacheCreationTokens: 0,
|
||||
requests: 0
|
||||
}
|
||||
bucket.inputTokens += usage.inputTokens
|
||||
bucket.outputTokens += usage.outputTokens
|
||||
bucket.cacheReadTokens += usage.cacheReadTokens
|
||||
bucket.cacheCreationTokens += usage.cacheCreationTokens
|
||||
bucket.requests += 1
|
||||
buckets.set(key, bucket)
|
||||
if (accumulateOpencodeUsage(buckets, row.data)) messages += 1
|
||||
}
|
||||
result.set(sessionId, { rows: Array.from(buckets.values()), messages })
|
||||
}
|
||||
@@ -191,3 +199,46 @@ export async function readOpencodeUsageForSessions(
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
/**
|
||||
* Like `readOpencodeUsageForSessions` but for every session present in the
|
||||
* store (the machine-level runner path). Reads the message table in a single
|
||||
* sequential pass: the per-session index queries above translate into one
|
||||
* scattered page fetch per message across the multi-GB store, while a table
|
||||
* scan reads the same pages in storage order.
|
||||
*/
|
||||
export async function readOpencodeUsageForAllSessions(
|
||||
dbPath: string
|
||||
): Promise<Map<string, OpencodeSessionUsage>> {
|
||||
const result = new Map<string, OpencodeSessionUsage>()
|
||||
if (!existsSync(dbPath)) return result
|
||||
|
||||
const { Database } = await import('bun:sqlite')
|
||||
const db = new Database(dbPath, { readonly: true })
|
||||
try {
|
||||
const rows = db.prepare('SELECT session_id, data FROM message').all() as Array<{
|
||||
session_id: string
|
||||
data: string
|
||||
}>
|
||||
const bucketsBySession = new Map<string, Map<string, OpencodeUsageReportRow>>()
|
||||
const messagesBySession = new Map<string, number>()
|
||||
for (const row of rows) {
|
||||
if (typeof row.session_id !== 'string' || row.session_id.length === 0) continue
|
||||
const buckets = bucketsBySession.get(row.session_id) ?? new Map<string, OpencodeUsageReportRow>()
|
||||
if (accumulateOpencodeUsage(buckets, row.data)) {
|
||||
messagesBySession.set(row.session_id, (messagesBySession.get(row.session_id) ?? 0) + 1)
|
||||
}
|
||||
bucketsBySession.set(row.session_id, buckets)
|
||||
}
|
||||
for (const [sessionId, buckets] of bucketsBySession) {
|
||||
result.set(sessionId, {
|
||||
rows: Array.from(buckets.values()),
|
||||
messages: messagesBySession.get(sessionId) ?? 0
|
||||
})
|
||||
}
|
||||
} finally {
|
||||
db.close()
|
||||
}
|
||||
|
||||
return result
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user