starred/orca
1
0
Fork 0
mirror of https://github.com/stablyai/orca.git synced 2026-10-23 08:03:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ai-vault): preserve bounded Cursor read concurrency

Browse Source
This commit is contained in:
Brennan Benson
2026-08-10 15:54:41 -07:00
parent 8c2e4b7863
commit 565429c8e1
5 changed files with 174 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/ai-vault/session-scanner-cursor-local-pipeline.ts
+73 -87
View File
@@ -103,7 +103,7 @@ type VerifiedReadTracker = {
budgetsByStorageKey: Map<string, VerifiedReadBudget>
}
type VerifiedReadBudget = { chargedBytes: number; gate: Promise<void> }
type VerifiedReadBudget = { chargedBytes: number; reservedBytes: number }
function createVerifiedReadTracker(
discoveries: readonly SessionFileDiscovery[] | undefined
@@ -132,29 +132,12 @@ function createVerifiedReadTracker(
return { byStorageKey, budgetsByStorageKey: new Map() }
}
async function withVerifiedReadGate<T>(
budget: VerifiedReadBudget,
run: () => Promise<T>
): Promise<T> {
const previous = budget.gate
let release!: () => void
budget.gate = new Promise<void>((resolve) => {
release = resolve
})
await previous
try {
return await run()
} finally {
release()
}
}
function verifiedReadBudget(tracker: VerifiedReadTracker, storageKey: string): VerifiedReadBudget {
const existing = tracker.budgetsByStorageKey.get(storageKey)
if (existing) {
return existing
}
const created = { chargedBytes: 0, gate: Promise.resolve() }
const created = { chargedBytes: 0, reservedBytes: 0 }
tracker.budgetsByStorageKey.set(storageKey, created)
return created
}
@@ -174,9 +157,16 @@ async function parseCursorGroups(
const candidates = groups.flatMap((group) => group.candidates)
for (let index = 0; index < candidates.length; index += CURSOR_PARSE_CONCURRENCY) {
const batch = candidates.slice(index, index + CURSOR_PARSE_CONCURRENCY)
const results = await Promise.all(
const settled = await Promise.allSettled(
batch.map((candidate) => parseCursorCandidate(candidate, args, verifiedReads))
)
const rejected = settled.find(
(result): result is PromiseRejectedResult => result.status === 'rejected'
)
if (rejected) {
throw rejected.reason
}
const results = settled.map((result) => (result.status === 'fulfilled' ? result.value : null))
parsed.push(...results.filter((result): result is ParsedCursorCandidate => Boolean(result)))
}
}
@@ -239,80 +229,76 @@ async function parseSidecarWithAggregateCap(
candidate.file.sizeBytes === undefined
? CURSOR_SIDECAR_MAX_BYTES
: Math.min(Math.max(0, candidate.file.sizeBytes), CURSOR_SIDECAR_MAX_BYTES)
// Gate covers reserve → verified read → commit of actual returned bytes.
return withVerifiedReadGate(budget, async () => {
throwIfAiVaultScanCancelled(args.signal)
if (budget.chargedBytes >= CURSOR_REMOTE_MAX_AGGREGATE_BYTES) {
if (storage) {
storage.truncated.sidecarBytes = true
}
return null
}
if (budget.chargedBytes + estimatedBytes > CURSOR_REMOTE_MAX_AGGREGATE_BYTES) {
if (storage) {
storage.truncated.sidecarBytes = true
}
return null
throwIfAiVaultScanCancelled(args.signal)
const availableBytes =
CURSOR_REMOTE_MAX_AGGREGATE_BYTES - budget.chargedBytes - budget.reservedBytes
if (availableBytes <= 0 || estimatedBytes > availableBytes) {
if (storage) {
storage.truncated.sidecarBytes = true
}
return null
}
const reservedBytes = Math.min(CURSOR_SIDECAR_MAX_BYTES, availableBytes)
budget.reservedBytes += reservedBytes
let result
try {
result = await parseCursorSidecarFileCached({
file: candidate.file,
platform: args.platform,
targetPlatform: candidate.cursorTargetPlatform,
executionHostId: args.executionHostId,
expectedRootRealPath: candidate.cursorExpectedRootRealPath
})
} catch (error) {
if (storage) {
storage.counters.boundedReads += 1
}
let result
try {
result = await parseCursorSidecarFileCached({
file: candidate.file,
platform: args.platform,
targetPlatform: candidate.cursorTargetPlatform,
executionHostId: args.executionHostId,
expectedRootRealPath: candidate.cursorExpectedRootRealPath,
maxBytes: reservedBytes
})
} catch (error) {
settleVerifiedReadReservation(
budget,
reservedBytes,
isVerifiedReadTooLargeError(error)
? CURSOR_REMOTE_MAX_AGGREGATE_BYTES
: CURSOR_SIDECAR_MAX_BYTES
)
if (storage) {
storage.counters.boundedReads += 1
if (isVerifiedReadTooLargeError(error)) {
budget.chargedBytes = CURSOR_REMOTE_MAX_AGGREGATE_BYTES
if (storage) {
storage.truncated.sidecarBytes = true
}
} else {
// A failed remote read hides how many bytes it consumed; charge its full ceiling.
budget.chargedBytes += CURSOR_SIDECAR_MAX_BYTES
}
throwIfAiVaultScanCancelled(args.signal)
throw error
}
if (!result.cacheHit) {
const readBytes = result.returnedBytes ?? 0
if (storage) {
// Count the logical verified-read work even when the payload is dropped.
storage.counters.boundedReads += 1
storage.counters.returnedBytes += readBytes
}
budget.chargedBytes += readBytes
if (budget.chargedBytes > CURSOR_REMOTE_MAX_AGGREGATE_BYTES) {
if (storage) {
storage.truncated.sidecarBytes = true
}
throwIfAiVaultScanCancelled(args.signal)
return null
storage.truncated.sidecarBytes = true
}
}
throwIfAiVaultScanCancelled(args.signal)
throw error
}
if (result.issue) {
args.issues.push(result.issue)
}
return result.evidence
? {
layout: 'sidecar' as const,
storageContextKey: storageKey,
file: candidate.file,
cwdEvidence: candidate.cursorCwdEvidence,
sidecar: result.evidence
}
: null
})
const readBytes = result.cacheHit ? 0 : (result.returnedBytes ?? 0)
settleVerifiedReadReservation(budget, reservedBytes, readBytes)
if (!result.cacheHit && storage) {
storage.counters.boundedReads += 1
storage.counters.returnedBytes += readBytes
}
throwIfAiVaultScanCancelled(args.signal)
if (result.issue) {
args.issues.push(result.issue)
}
return result.evidence
? {
layout: 'sidecar' as const,
storageContextKey: storageKey,
file: candidate.file,
cwdEvidence: candidate.cursorCwdEvidence,
sidecar: result.evidence
}
: null
}
function settleVerifiedReadReservation(
budget: VerifiedReadBudget,
reservedBytes: number,
chargedBytes: number
): void {
budget.reservedBytes -= reservedBytes
const available = CURSOR_REMOTE_MAX_AGGREGATE_BYTES - budget.chargedBytes - budget.reservedBytes
budget.chargedBytes += Math.min(chargedBytes, Math.max(0, available))
}
function isVerifiedReadTooLargeError(error: unknown): boolean {
src/main/ai-vault/session-scanner-cursor-sidecar.ts
+3 -1
View File
@@ -55,13 +55,14 @@ export async function parseCursorSidecarFile(args: {
targetPlatform?: NodeJS.Platform
executionHostId?: ExecutionHostId
expectedRootRealPath?: string
maxBytes?: number
}): Promise<CursorSidecarParseResult> {
if (!args.expectedRootRealPath) {
throw new Error('cursor_sidecar_root_unavailable')
}
const content = await readVerifiedBoundedTextFile(args.file.path, {
expectedRootRealPath: args.expectedRootRealPath,
maxBytes: CURSOR_SIDECAR_MAX_BYTES
maxBytes: args.maxBytes ?? CURSOR_SIDECAR_MAX_BYTES
})
return {
...parseCursorSidecarContent({
@@ -78,6 +79,7 @@ export async function parseCursorSidecarFileCached(args: {
targetPlatform?: NodeJS.Platform
executionHostId?: ExecutionHostId
expectedRootRealPath?: string
maxBytes?: number
}): Promise<CursorSidecarParseResult> {
const cached = sidecarCache.get(args.file.path)
if (
src/main/ai-vault/session-scanner-cursor-storage-budget.test.ts
+63 -7
View File
@@ -2,7 +2,7 @@ import { createHash } from 'node:crypto'
import { lstat, mkdir, mkdtemp, realpath, rm, utimes, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, describe, expect, it } from 'vitest'
import { afterEach, describe, expect, it, vi } from 'vitest'
import type { AiVaultScanIssue } from '../../shared/ai-vault-types'
import {
CURSOR_REMOTE_MAX_AGGREGATE_BYTES,
@@ -10,6 +10,7 @@ import {
} from '../../shared/cursor-sidecar-scan'
import { startSpan } from '../observability/tracer'
import { processLocalCursorCandidates } from './session-scanner-cursor-local-pipeline'
import * as cursorSidecarParser from './session-scanner-cursor-sidecar'
import { createSessionParseStats } from './session-scanner-parse-cache'
import type {
FileWithMtime,
@@ -20,10 +21,63 @@ import type {
const roots: string[] = []
afterEach(async () => {
vi.restoreAllMocks()
cursorSidecarParser.resetCursorSidecarParseCacheForTests()
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })))
})
describe('Cursor verified-read budgets by storage context', () => {
it.each(['native', 'wsl:ubuntu'])(
'keeps eight cold %s verified reads concurrent within one storage budget',
async (storageKey) => {
const root = await mkdtemp(join(tmpdir(), 'orca-cursor-storage-concurrency-'))
roots.push(root)
const chatsRoot = join(root, '.cursor', 'chats')
const files = await Promise.all(
Array.from({ length: 8 }, (_, index) =>
addSession(chatsRoot, `session-${index}`, sidecarPayload(1_000), 1_000)
)
)
const discovery = sidecarDiscovery(storageKey, chatsRoot, await realpath(chatsRoot), files)
const originalParse = cursorSidecarParser.parseCursorSidecarFileCached
let activeReads = 0
let peakReads = 0
vi.spyOn(cursorSidecarParser, 'parseCursorSidecarFileCached').mockImplementation(
async (args) => {
activeReads += 1
peakReads = Math.max(peakReads, activeReads)
await new Promise<void>((resolve) => setImmediate(resolve))
try {
return await originalParse(args)
} finally {
activeReads -= 1
}
}
)
const span = startSpan('cursor-storage-concurrency-test')
try {
const result = await processLocalCursorCandidates({
candidates: discoveryCandidates(discovery),
discoveries: [discovery],
executionHostId: 'local',
issues: [],
limit: 20,
parseStats: createSessionParseStats(),
platform: 'linux',
scopeLimit: 20,
span
})
expect(result.sessions).toHaveLength(8)
expect(peakReads).toBe(8)
expect(discovery.cursorDiscoveryCounters?.boundedReads).toBe(8)
} finally {
span.end()
}
}
)
it('stops after the first verified read when cancellation lands during parsing', async () => {
const root = await mkdtemp(join(tmpdir(), 'orca-cursor-parse-cancel-'))
roots.push(root)
@@ -63,7 +117,7 @@ describe('Cursor verified-read budgets by storage context', () => {
}
})
it('charges a raced verified read and skips later candidates after budget exhaustion', async () => {
it('bounds a raced verified read by the remaining aggregate budget', async () => {
const root = await mkdtemp(join(tmpdir(), 'orca-cursor-storage-race-'))
roots.push(root)
const chatsRoot = join(root, '.cursor', 'chats')
@@ -89,13 +143,14 @@ describe('Cursor verified-read budgets by storage context', () => {
const files = [...fillerFiles, ...racedFiles]
const discovery = sidecarDiscovery('native', chatsRoot, await realpath(chatsRoot), files)
const span = startSpan('cursor-storage-race-test')
const issues: AiVaultScanIssue[] = []
try {
const result = await processLocalCursorCandidates({
candidates: discoveryCandidates(discovery),
discoveries: [discovery],
executionHostId: 'local',
issues: [],
issues,
limit: 100,
parseStats: createSessionParseStats(),
platform: 'linux',
@@ -106,11 +161,12 @@ describe('Cursor verified-read budgets by storage context', () => {
expect(result.sessions).toHaveLength(fillerCount)
expect(discovery.cursorDiscoveryCounters?.boundedReads).toBe(fillerCount + 1)
expect(discovery.cursorDiscoveryCounters?.returnedBytes).toBe(fillerTotal + racedBytes)
expect(discovery.cursorDiscoveryCounters?.returnedBytes).toBeGreaterThan(
expect(discovery.cursorDiscoveryCounters?.returnedBytes).toBe(fillerTotal)
expect(discovery.cursorDiscoveryCounters?.returnedBytes).toBeLessThanOrEqual(
CURSOR_REMOTE_MAX_AGGREGATE_BYTES
)
expect(discovery.cursorDiscoveryTruncated?.sidecarBytes).toBe(true)
expect(issues).toContainEqual(expect.objectContaining({ message: 'file_too_large' }))
} finally {
span.end()
}
@@ -145,10 +201,10 @@ describe('Cursor verified-read budgets by storage context', () => {
})
expect(result.sessions).toEqual([])
expect(discovery.cursorDiscoveryCounters?.boundedReads).toBe(1)
expect(discovery.cursorDiscoveryCounters?.boundedReads).toBe(2)
expect(discovery.cursorDiscoveryCounters?.returnedBytes).toBe(0)
expect(discovery.cursorDiscoveryTruncated?.sidecarBytes).toBe(true)
expect(issues).toHaveLength(1)
expect(issues).toHaveLength(2)
expect(issues[0]?.message).toContain('file_too_large')
} finally {
span.end()
src/relay/cursor-sidecar-scan.test.ts
+34
View File
@@ -435,6 +435,40 @@ describe('scanCursorSidecars', () => {
expect(result.issues).toContainEqual(expect.objectContaining({ message: 'file_too_large' }))
})
it('bounds a raced sidecar by the remaining aggregate budget', async () => {
const root = await createRoot()
const chatsRoot = join(root, 'chats')
const bucket = 'dededededededededededededededede'
const racePath = join(chatsRoot, bucket, 'race', 'meta.json')
await Promise.all([
addSession(chatsRoot, bucket, 'filler', 'f'.repeat(90)),
addSession(chatsRoot, bucket, 'race', 'r'.repeat(5))
])
await setSessionMtime(chatsRoot, bucket, 'filler', 10_000)
await setSessionMtime(chatsRoot, bucket, 'race', 1_000)
const request = defaultCursorSidecarScanRequest(chatsRoot, [], process.platform)
request.maxAggregateBytes = 100
let checks = 0
const result = await scanCursorSidecars(request, {
clientId: 1,
isStale: () => {
checks += 1
if (checks === CHECKS_BEFORE_FIRST_VERIFIED_READ) {
writeFileSync(racePath, 'x'.repeat(20))
}
return false
}
})
expect(result.sidecars.map((sidecar) => sidecar.sessionId)).toEqual(['filler'])
expect(result.counters.returnedBytes).toBe(90)
expect(result.truncated.sidecarBytes).toBe(true)
expect(result.issues).toContainEqual(
expect.objectContaining({ path: racePath, message: 'file_too_large' })
)
})
it('prioritizes cancellation that lands during a failed raced read', async () => {
const root = await createRoot()
const chatsRoot = join(root, 'chats')
src/relay/cursor-sidecar-scan.ts
+1 -1
View File
@@ -79,7 +79,7 @@ async function readCandidates(
response.counters.boundedReads++
const content = await readVerifiedBoundedTextFile(candidate.metaPath, {
expectedRootRealPath: rootRealPath,
maxBytes: caps.sidecarBytes
maxBytes: Math.min(caps.sidecarBytes, caps.aggregateBytes - chargedBytes)
})
throwIfCancelled(context)
const bytes = Buffer.byteLength(content, 'utf8')