mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 16:03:47 +00:00
fix(ai-vault): isolate Cursor read budgets by storage
This commit is contained in:
@@ -98,11 +98,11 @@ type VerifiedReadTracker = {
|
||||
truncated: NonNullable<SessionFileDiscovery['cursorDiscoveryTruncated']>
|
||||
}
|
||||
>
|
||||
aggregateReturnedBytes: number
|
||||
/** Serializes reserve/read/commit so concurrent parses cannot overshoot the cap. */
|
||||
gate: Promise<void>
|
||||
budgetsByStorageKey: Map<string, VerifiedReadBudget>
|
||||
}
|
||||
|
||||
type VerifiedReadBudget = { returnedBytes: number; gate: Promise<void> }
|
||||
|
||||
function createVerifiedReadTracker(
|
||||
discoveries: readonly SessionFileDiscovery[] | undefined
|
||||
): VerifiedReadTracker {
|
||||
@@ -127,16 +127,16 @@ function createVerifiedReadTracker(
|
||||
truncated: discovery.cursorDiscoveryTruncated
|
||||
})
|
||||
}
|
||||
return { byStorageKey, aggregateReturnedBytes: 0, gate: Promise.resolve() }
|
||||
return { byStorageKey, budgetsByStorageKey: new Map() }
|
||||
}
|
||||
|
||||
async function withVerifiedReadGate<T>(
|
||||
tracker: VerifiedReadTracker,
|
||||
budget: VerifiedReadBudget,
|
||||
run: () => Promise<T>
|
||||
): Promise<T> {
|
||||
const previous = tracker.gate
|
||||
const previous = budget.gate
|
||||
let release!: () => void
|
||||
tracker.gate = new Promise<void>((resolve) => {
|
||||
budget.gate = new Promise<void>((resolve) => {
|
||||
release = resolve
|
||||
})
|
||||
await previous
|
||||
@@ -147,6 +147,16 @@ async function withVerifiedReadGate<T>(
|
||||
}
|
||||
}
|
||||
|
||||
function verifiedReadBudget(tracker: VerifiedReadTracker, storageKey: string): VerifiedReadBudget {
|
||||
const existing = tracker.budgetsByStorageKey.get(storageKey)
|
||||
if (existing) {
|
||||
return existing
|
||||
}
|
||||
const created = { returnedBytes: 0, gate: Promise.resolve() }
|
||||
tracker.budgetsByStorageKey.set(storageKey, created)
|
||||
return created
|
||||
}
|
||||
|
||||
async function parseCursorGroups(
|
||||
groups: readonly CursorCandidateSelectionGroup<SessionFileCandidate>[],
|
||||
parsed: ParsedCursorCandidate[],
|
||||
@@ -218,6 +228,7 @@ async function parseSidecarWithAggregateCap(
|
||||
): Promise<ParsedCursorCandidate | null> {
|
||||
const storageKey = candidate.cursorStorageContextKey ?? 'native'
|
||||
const storage = verifiedReads.byStorageKey.get(storageKey)
|
||||
const budget = verifiedReadBudget(verifiedReads, storageKey)
|
||||
// Unknown size reserves the per-sidecar max so concurrent admission cannot overshoot.
|
||||
const estimatedBytes =
|
||||
candidate.file.sizeBytes === undefined
|
||||
@@ -225,14 +236,14 @@ async function parseSidecarWithAggregateCap(
|
||||
: Math.min(Math.max(0, candidate.file.sizeBytes), CURSOR_SIDECAR_MAX_BYTES)
|
||||
|
||||
// Gate covers reserve → verified read → commit of actual returned bytes.
|
||||
return withVerifiedReadGate(verifiedReads, async () => {
|
||||
if (verifiedReads.aggregateReturnedBytes >= CURSOR_REMOTE_MAX_AGGREGATE_BYTES) {
|
||||
return withVerifiedReadGate(budget, async () => {
|
||||
if (budget.returnedBytes >= CURSOR_REMOTE_MAX_AGGREGATE_BYTES) {
|
||||
if (storage) {
|
||||
storage.truncated.sidecarBytes = true
|
||||
}
|
||||
return null
|
||||
}
|
||||
if (verifiedReads.aggregateReturnedBytes + estimatedBytes > CURSOR_REMOTE_MAX_AGGREGATE_BYTES) {
|
||||
if (budget.returnedBytes + estimatedBytes > CURSOR_REMOTE_MAX_AGGREGATE_BYTES) {
|
||||
if (storage) {
|
||||
storage.truncated.sidecarBytes = true
|
||||
}
|
||||
@@ -252,7 +263,7 @@ async function parseSidecarWithAggregateCap(
|
||||
// Count the logical verified-read work even when the payload is dropped.
|
||||
storage.counters.boundedReads += 1
|
||||
}
|
||||
if (verifiedReads.aggregateReturnedBytes + readBytes > CURSOR_REMOTE_MAX_AGGREGATE_BYTES) {
|
||||
if (budget.returnedBytes + readBytes > CURSOR_REMOTE_MAX_AGGREGATE_BYTES) {
|
||||
if (storage) {
|
||||
storage.truncated.sidecarBytes = true
|
||||
}
|
||||
@@ -261,7 +272,7 @@ async function parseSidecarWithAggregateCap(
|
||||
if (storage) {
|
||||
storage.counters.returnedBytes += readBytes
|
||||
}
|
||||
verifiedReads.aggregateReturnedBytes += readBytes
|
||||
budget.returnedBytes += readBytes
|
||||
}
|
||||
|
||||
if (result.issue) {
|
||||
|
||||
@@ -0,0 +1,162 @@
|
||||
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 {
|
||||
CURSOR_REMOTE_MAX_AGGREGATE_BYTES,
|
||||
CURSOR_SIDECAR_MAX_BYTES
|
||||
} from '../../shared/cursor-sidecar-scan'
|
||||
import { startSpan } from '../observability/tracer'
|
||||
import { processLocalCursorCandidates } from './session-scanner-cursor-local-pipeline'
|
||||
import { createSessionParseStats } from './session-scanner-parse-cache'
|
||||
import type {
|
||||
FileWithMtime,
|
||||
SessionFileCandidate,
|
||||
SessionFileDiscovery
|
||||
} from './session-scanner-types'
|
||||
|
||||
const roots: string[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })))
|
||||
})
|
||||
|
||||
describe('Cursor verified-read budgets by storage context', () => {
|
||||
it('does not let native reads consume the WSL ingress budget', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-cursor-storage-budget-'))
|
||||
roots.push(root)
|
||||
const nativeRoot = join(root, 'native', '.cursor', 'chats')
|
||||
const wslRoot = join(root, 'wsl', '.cursor', 'chats')
|
||||
const nativePayloadBytes = CURSOR_SIDECAR_MAX_BYTES - 44
|
||||
const nativeFileCount = Math.floor(CURSOR_REMOTE_MAX_AGGREGATE_BYTES / nativePayloadBytes)
|
||||
const wslPayloadBytes = 5_000
|
||||
const nativePayload = sidecarPayload(nativePayloadBytes)
|
||||
const nativeFiles: FileWithMtime[] = []
|
||||
expect(nativePayloadBytes * nativeFileCount + wslPayloadBytes).toBeGreaterThan(
|
||||
CURSOR_REMOTE_MAX_AGGREGATE_BYTES
|
||||
)
|
||||
|
||||
for (let index = 0; index < nativeFileCount; index += 1) {
|
||||
nativeFiles.push(
|
||||
await addSession(
|
||||
nativeRoot,
|
||||
`native-${String(index).padStart(3, '0')}`,
|
||||
nativePayload,
|
||||
10_000
|
||||
)
|
||||
)
|
||||
}
|
||||
const wslFile = await addSession(wslRoot, 'wsl-session', sidecarPayload(wslPayloadBytes), 1_000)
|
||||
const discoveries = [
|
||||
sidecarDiscovery('native', nativeRoot, await realpath(nativeRoot), nativeFiles),
|
||||
sidecarDiscovery('wsl:ubuntu', wslRoot, await realpath(wslRoot), [wslFile])
|
||||
]
|
||||
const span = startSpan('cursor-storage-budget-test')
|
||||
const issues = []
|
||||
|
||||
try {
|
||||
const result = await processLocalCursorCandidates({
|
||||
candidates: discoveries.flatMap(discoveryCandidates),
|
||||
discoveries,
|
||||
executionHostId: 'local',
|
||||
issues,
|
||||
limit: 100,
|
||||
parseStats: createSessionParseStats(),
|
||||
platform: 'linux',
|
||||
scopeLimit: 100,
|
||||
span
|
||||
})
|
||||
|
||||
expect(issues).toEqual([])
|
||||
expect(result.sessions.map((session) => session.filePath)).toContain(wslFile.path)
|
||||
expect(discoveries[0].cursorDiscoveryCounters?.returnedBytes).toBe(
|
||||
nativePayloadBytes * nativeFileCount
|
||||
)
|
||||
expect(discoveries[1].cursorDiscoveryCounters?.returnedBytes).toBe(wslPayloadBytes)
|
||||
expect(
|
||||
discoveries.map((discovery) => discovery.cursorDiscoveryTruncated?.sidecarBytes)
|
||||
).toEqual([false, false])
|
||||
} finally {
|
||||
span.end()
|
||||
}
|
||||
}, 30_000)
|
||||
})
|
||||
|
||||
function sidecarDiscovery(
|
||||
storageKey: string,
|
||||
rootDir: string,
|
||||
expectedRootRealPath: string,
|
||||
files: FileWithMtime[]
|
||||
): SessionFileDiscovery {
|
||||
return {
|
||||
agent: 'cursor',
|
||||
cursorDiscoveryCounters: {
|
||||
boundedReads: 0,
|
||||
bucketReaddir: 0,
|
||||
elapsedMs: 0,
|
||||
fileLstat: 0,
|
||||
returnedBytes: 0,
|
||||
rootReaddir: 0,
|
||||
scopeRealpath: 0
|
||||
},
|
||||
cursorDiscoveryTruncated: {
|
||||
buckets: false,
|
||||
scopePaths: false,
|
||||
sessionDirs: false,
|
||||
sidecarBytes: false
|
||||
},
|
||||
cursorExpectedRootRealPath: expectedRootRealPath,
|
||||
cursorLayout: 'sidecar',
|
||||
cursorStorageContextKey: storageKey,
|
||||
files,
|
||||
rootDir
|
||||
}
|
||||
}
|
||||
|
||||
function discoveryCandidates(discovery: SessionFileDiscovery): SessionFileCandidate[] {
|
||||
return discovery.files.map((file) => ({
|
||||
agent: 'cursor',
|
||||
codexHome: null,
|
||||
cursorExpectedRootRealPath: discovery.cursorExpectedRootRealPath,
|
||||
cursorLayout: 'sidecar',
|
||||
cursorStorageContextKey: discovery.cursorStorageContextKey,
|
||||
file
|
||||
}))
|
||||
}
|
||||
|
||||
async function addSession(
|
||||
chatsDir: string,
|
||||
sessionId: string,
|
||||
content: string,
|
||||
mtimeMs: number
|
||||
): Promise<FileWithMtime> {
|
||||
const bucket = createHash('md5').update(sessionId).digest('hex')
|
||||
const sessionDir = join(chatsDir, bucket, sessionId)
|
||||
const metaPath = join(sessionDir, 'meta.json')
|
||||
const storePath = join(sessionDir, 'store.db')
|
||||
await mkdir(sessionDir, { recursive: true })
|
||||
await Promise.all([writeFile(metaPath, content), writeFile(storePath, '')])
|
||||
const timestamp = new Date(mtimeMs)
|
||||
await Promise.all([
|
||||
utimes(metaPath, timestamp, timestamp),
|
||||
utimes(storePath, timestamp, timestamp)
|
||||
])
|
||||
const [meta, store] = await Promise.all([lstat(metaPath), lstat(storePath)])
|
||||
return {
|
||||
cursorStoreMtimeMs: store.mtimeMs,
|
||||
dev: meta.dev,
|
||||
ino: meta.ino,
|
||||
modifiedAt: meta.mtime.toISOString(),
|
||||
mtimeMs: meta.mtimeMs,
|
||||
nlink: meta.nlink,
|
||||
path: metaPath,
|
||||
sizeBytes: meta.size
|
||||
}
|
||||
}
|
||||
|
||||
function sidecarPayload(byteLength: number): string {
|
||||
const prefix = '{"createdAtMs":1,"updatedAtMs":2,"hasConversation":true,"title":"session","pad":"'
|
||||
const suffix = '"}'
|
||||
return `${prefix}${'a'.repeat(byteLength - Buffer.byteLength(prefix + suffix))}${suffix}`
|
||||
}
|
||||
Reference in New Issue
Block a user