mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 08:03:37 +00:00
fix(persistence): bound best-effort flushes
This commit is contained in:
@@ -162,9 +162,7 @@ describe('registerOrcaProfileHandlers', () => {
|
||||
expect(flush.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
setActiveOrcaProfileMock.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY
|
||||
)
|
||||
expect(flush.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
onBeforeRelaunch.mock.invocationCallOrder[0] ?? Number.POSITIVE_INFINITY
|
||||
)
|
||||
expect(flush).toHaveBeenCalledBefore(onBeforeRelaunch)
|
||||
expect(appRelaunchMock).not.toHaveBeenCalled()
|
||||
|
||||
await vi.advanceTimersByTimeAsync(150)
|
||||
|
||||
@@ -416,6 +416,43 @@ describe('async persistence write path avoids synchronous fs syscalls', () => {
|
||||
expect(JSON.parse(readFileSync(dataFile(dir), 'utf-8')).ui.sidebarWidth).toBe(602)
|
||||
})
|
||||
|
||||
it('bounds a best-effort flush to one state generation', async () => {
|
||||
const dir = makeDir()
|
||||
const store = await createStore(dir)
|
||||
let releaseRename!: () => void
|
||||
const renameRelease = new Promise<void>((resolve) => {
|
||||
releaseRename = resolve
|
||||
})
|
||||
let signalRename!: () => void
|
||||
const renameStarted = new Promise<void>((resolve) => {
|
||||
signalRename = resolve
|
||||
})
|
||||
let held = false
|
||||
fsCalls.waitAsync = (fn, target) => {
|
||||
if (held || fn !== 'rename' || !target.startsWith(dataFile(dir))) {
|
||||
return null
|
||||
}
|
||||
held = true
|
||||
signalRename()
|
||||
return renameRelease
|
||||
}
|
||||
fsCalls.dirPrefix = dir
|
||||
fsCalls.recording = true
|
||||
|
||||
store.updateUI({ sidebarWidth: 621 })
|
||||
const flush = store.flushPendingAsync()
|
||||
await renameStarted
|
||||
store.updateUI({ sidebarWidth: 622 })
|
||||
releaseRename()
|
||||
await flush
|
||||
fsCalls.recording = false
|
||||
|
||||
expect(JSON.parse(readFileSync(dataFile(dir), 'utf-8')).ui.sidebarWidth).toBe(621)
|
||||
|
||||
await store.flushPendingOrThrowAsync()
|
||||
expect(JSON.parse(readFileSync(dataFile(dir), 'utf-8')).ui.sidebarWidth).toBe(622)
|
||||
})
|
||||
|
||||
it('the throwing async barrier drains mutations made during sidecar I/O', async () => {
|
||||
const dir = makeDir()
|
||||
const store = await createStore(dir)
|
||||
|
||||
@@ -7163,7 +7163,8 @@ export class Store {
|
||||
}
|
||||
|
||||
flushPendingAsync(): Promise<void> {
|
||||
return this.flushCurrentStateAsync(false).catch(() => {})
|
||||
// Best-effort callers must not livelock while the live app keeps mutating state.
|
||||
return this.flushCurrentStateAsync(false, undefined, false).catch(() => {})
|
||||
}
|
||||
|
||||
flushPendingOrThrowAsync(options: { signal?: AbortSignal } = {}): Promise<void> {
|
||||
@@ -7173,7 +7174,11 @@ export class Store {
|
||||
return this.flushCurrentStateAsync(false, options.signal)
|
||||
}
|
||||
|
||||
private async flushCurrentStateAsync(final: boolean, signal?: AbortSignal): Promise<void> {
|
||||
private async flushCurrentStateAsync(
|
||||
final: boolean,
|
||||
signal?: AbortSignal,
|
||||
drainToStableGeneration = true
|
||||
): Promise<void> {
|
||||
for (;;) {
|
||||
if (signal?.aborted) {
|
||||
throw new Error('Persistence flush aborted')
|
||||
@@ -7200,7 +7205,7 @@ export class Store {
|
||||
if (signal?.aborted) {
|
||||
throw new Error('Persistence flush aborted')
|
||||
}
|
||||
if (generation === this.writeGeneration) {
|
||||
if (!drainToStableGeneration || generation === this.writeGeneration) {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,10 +2,9 @@ import { app } from 'electron'
|
||||
import { join } from 'node:path'
|
||||
import type { StatsSummary } from '../../shared/types'
|
||||
import type { StatsEvent, StatsAggregates } from './types'
|
||||
import { loadStatsFile } from './stats-file-loader'
|
||||
import { loadStatsFile, STATS_SCHEMA_VERSION } from './stats-file-loader'
|
||||
import { StatsSnapshotWriter } from './stats-snapshot-writer'
|
||||
|
||||
const STATS_SCHEMA_VERSION = 1
|
||||
const MAX_EVENTS = 10_000
|
||||
// Why: countedPRs is a deduplication registry that grows with every PR created
|
||||
// through Orca. Without a cap, a heavily-used instance accumulates thousands of
|
||||
@@ -225,13 +224,7 @@ export class StatsCollector {
|
||||
}
|
||||
this.writeTimer = setTimeout(() => {
|
||||
this.writeTimer = null
|
||||
// Why: the debounced save is fun-stats telemetry, not crash-critical
|
||||
// state, so it uses the async writer to move the ~900KB tmp-file write
|
||||
// off the main thread (the stringify stays sync — see prepareWritePayload).
|
||||
// A chatty multi-agent session re-arms this every 5s; a fully-sync write
|
||||
// is a recurring main-thread stall. The quit path uses flushAsync(); the sync
|
||||
// flush() remains for callers that cannot await, and writeToDiskSync keeps the
|
||||
// two paths race-safe.
|
||||
// Why async: a chatty session can write ~900KB every 5s and stall the main thread.
|
||||
void this.enqueueWrite().catch((err) => {
|
||||
console.error('[stats] Failed to write stats:', err)
|
||||
})
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { existsSync, readFileSync } from 'node:fs'
|
||||
import type { StatsAggregates, StatsFile } from './types'
|
||||
|
||||
const STATS_SCHEMA_VERSION = 1
|
||||
export const STATS_SCHEMA_VERSION = 1
|
||||
|
||||
function defaultAggregates(): StatsAggregates {
|
||||
return {
|
||||
|
||||
@@ -55,7 +55,7 @@ export class StatsSnapshotWriter {
|
||||
}
|
||||
}
|
||||
}
|
||||
const { tmpFile, json, generation } = this.preparePayload(serialize)
|
||||
const { tmpFile, json, generation } = this.preparePayload(statsFile, serialize)
|
||||
writeFileSync(tmpFile, json, 'utf-8')
|
||||
renameSync(tmpFile, statsFile)
|
||||
this.lastCommittedGeneration = Math.max(this.lastCommittedGeneration, generation)
|
||||
@@ -65,14 +65,14 @@ export class StatsSnapshotWriter {
|
||||
return this.pendingWrite ?? Promise.resolve()
|
||||
}
|
||||
|
||||
private preparePayload(serialize: () => string): {
|
||||
private preparePayload(finalPath: string, serialize: () => string): {
|
||||
tmpFile: string
|
||||
json: string
|
||||
generation: number
|
||||
} {
|
||||
const generation = ++this.writeGeneration
|
||||
return {
|
||||
tmpFile: durableWriteTempPath(this.resolveFile()),
|
||||
tmpFile: durableWriteTempPath(finalPath),
|
||||
json: serialize(),
|
||||
generation
|
||||
}
|
||||
@@ -101,7 +101,7 @@ export class StatsSnapshotWriter {
|
||||
private async writeToDiskAsync(serialize: () => string): Promise<void> {
|
||||
const statsFile = this.resolveFile()
|
||||
await mkdir(dirname(statsFile), { recursive: true }).catch(() => {})
|
||||
const { tmpFile, json, generation } = this.preparePayload(serialize)
|
||||
const { tmpFile, json, generation } = this.preparePayload(statsFile, serialize)
|
||||
let renamed = false
|
||||
try {
|
||||
await writeFile(tmpFile, json, 'utf-8')
|
||||
|
||||
Reference in New Issue
Block a user