From 565429c8e1408214a836baf9456f4a69efdce8d3 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 10 Aug 2026 15:54:41 -0700 Subject: [PATCH] fix(ai-vault): preserve bounded Cursor read concurrency --- .../session-scanner-cursor-local-pipeline.ts | 160 ++++++++---------- .../session-scanner-cursor-sidecar.ts | 4 +- ...sion-scanner-cursor-storage-budget.test.ts | 70 +++++++- src/relay/cursor-sidecar-scan.test.ts | 34 ++++ src/relay/cursor-sidecar-scan.ts | 2 +- 5 files changed, 174 insertions(+), 96 deletions(-) diff --git a/src/main/ai-vault/session-scanner-cursor-local-pipeline.ts b/src/main/ai-vault/session-scanner-cursor-local-pipeline.ts index 3dea973744b..c6cc37bb62e 100644 --- a/src/main/ai-vault/session-scanner-cursor-local-pipeline.ts +++ b/src/main/ai-vault/session-scanner-cursor-local-pipeline.ts @@ -103,7 +103,7 @@ type VerifiedReadTracker = { budgetsByStorageKey: Map } -type VerifiedReadBudget = { chargedBytes: number; gate: Promise } +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( - budget: VerifiedReadBudget, - run: () => Promise -): Promise { - const previous = budget.gate - let release!: () => void - budget.gate = new Promise((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 { diff --git a/src/main/ai-vault/session-scanner-cursor-sidecar.ts b/src/main/ai-vault/session-scanner-cursor-sidecar.ts index 0467e214541..45f4fd59b9e 100644 --- a/src/main/ai-vault/session-scanner-cursor-sidecar.ts +++ b/src/main/ai-vault/session-scanner-cursor-sidecar.ts @@ -55,13 +55,14 @@ export async function parseCursorSidecarFile(args: { targetPlatform?: NodeJS.Platform executionHostId?: ExecutionHostId expectedRootRealPath?: string + maxBytes?: number }): Promise { 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 { const cached = sidecarCache.get(args.file.path) if ( diff --git a/src/main/ai-vault/session-scanner-cursor-storage-budget.test.ts b/src/main/ai-vault/session-scanner-cursor-storage-budget.test.ts index ef74f71e521..590c9a1a81c 100644 --- a/src/main/ai-vault/session-scanner-cursor-storage-budget.test.ts +++ b/src/main/ai-vault/session-scanner-cursor-storage-budget.test.ts @@ -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((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() diff --git a/src/relay/cursor-sidecar-scan.test.ts b/src/relay/cursor-sidecar-scan.test.ts index 491109aa179..2d59ae0314e 100644 --- a/src/relay/cursor-sidecar-scan.test.ts +++ b/src/relay/cursor-sidecar-scan.test.ts @@ -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') diff --git a/src/relay/cursor-sidecar-scan.ts b/src/relay/cursor-sidecar-scan.ts index d1385613694..a0a8644c176 100644 --- a/src/relay/cursor-sidecar-scan.ts +++ b/src/relay/cursor-sidecar-scan.ts @@ -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')