diff --git a/src/main/ai-vault/session-scanner-cursor-local-bounds.test.ts b/src/main/ai-vault/session-scanner-cursor-local-bounds.test.ts index 162acf62afb..17529e1e790 100644 --- a/src/main/ai-vault/session-scanner-cursor-local-bounds.test.ts +++ b/src/main/ai-vault/session-scanner-cursor-local-bounds.test.ts @@ -218,6 +218,155 @@ describe('local Cursor sidecar discovery bounds', () => { ).rejects.toThrow('cursor_sidecar_scan_cancelled') }) + it('rejects on second-phase cancellation instead of resolving with an issue', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-cursor-cancel-2nd-')) + tempRoots.push(root) + const chatsDir = join(root, 'chats') + await mkdir(chatsDir, { recursive: true }) + for (let index = 0; index < 8; index += 1) { + await addSession(chatsDir, bucketName(`phase-${index}`), `session-${index}`) + } + let checks = 0 + const signal = { + get aborted() { + checks += 1 + // First check (post-realpath) passes; later enumeration-phase checks abort. + return checks > 1 + } + } as AbortSignal + const issues: AiVaultScanIssue[] = [] + await expect( + discoverLocalCursorSidecarsBounded({ + chatsDir, + scopePaths: [], + issues, + signal + }) + ).rejects.toThrow('cursor_sidecar_scan_cancelled') + expect(issues.some((issue) => issue.message.includes('cursor_sidecar_scan_cancelled'))).toBe( + false + ) + }) + + it('caps aggregate verified meta bytes so a 70×~250KiB store cannot return every session', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-cursor-70-agg-')) + tempRoots.push(root) + const chatsDir = join(root, 'chats') + const bucket = bucketName('aggregate-70') + // Adversarial shape: 70 sidecars totaling ~17.5 MiB (> 16 MiB aggregate). + const prefix = '{"createdAtMs":1,"updatedAtMs":2,"hasConversation":true,"title":"x","pad":"' + const suffix = '"}' + const pad = 'a'.repeat(250_069 - Buffer.byteLength(prefix + suffix, 'utf8')) + const payload = `${prefix}${pad}${suffix}` + const payloadBytes = Buffer.byteLength(payload, 'utf8') + expect(payloadBytes).toBe(250_069) + expect(payloadBytes * 70).toBe(17_504_830) + for (let index = 0; index < 70; index += 1) { + const sessionDir = join(chatsDir, bucket, `session-${String(index).padStart(3, '0')}`) + await mkdir(sessionDir, { recursive: true }) + await Promise.all([ + writeFile(join(sessionDir, 'meta.json'), payload), + writeFile(join(sessionDir, 'store.db'), '') + ]) + } + const issues: AiVaultScanIssue[] = [] + const result = await discoverLocalCursorSidecarsBounded({ + chatsDir, + scopePaths: [], + issues + }) + const retainedBytes = result.files.reduce((total, file) => total + (file.sizeBytes ?? 0), 0) + expect(result.files.length).toBeLessThan(70) + expect(result.files.length).toBeGreaterThan(0) + expect(retainedBytes).toBeLessThanOrEqual(16_777_216) + expect(result.truncated.sidecarBytes).toBe(true) + expect(issues.some((issue) => issue.message.includes('sidecar bytes'))).toBe(true) + }) + + it('applies newest-first mtime retention before the aggregate byte cap', async () => { + const { utimes } = await import('node:fs/promises') + const root = await mkdtemp(join(tmpdir(), 'orca-cursor-mtime-ret-')) + tempRoots.push(root) + const chatsDir = join(root, 'chats') + const bucket = bucketName('mtime-pair') + const content = `${'x'.repeat(70)}` + expect(Buffer.byteLength(content, 'utf8')).toBe(70) + // Lexicographically earlier session is older; later name is newer. + for (const [sessionId, mtimeMs] of [ + ['aaa-older', 1_000], + ['zzz-newer', 9_000] + ] as const) { + const sessionDir = join(chatsDir, bucket, sessionId) + await mkdir(sessionDir, { recursive: true }) + const metaPath = join(sessionDir, 'meta.json') + await Promise.all([writeFile(metaPath, content), writeFile(join(sessionDir, 'store.db'), '')]) + const timestamp = new Date(mtimeMs) + await Promise.all([ + utimes(metaPath, timestamp, timestamp), + utimes(join(sessionDir, 'store.db'), timestamp, timestamp) + ]) + } + const result = await discoverLocalCursorSidecarsBounded({ + chatsDir, + scopePaths: [], + issues: [] + // Force a 70-byte aggregate so only one of the two equal-size sessions fits. + // discoverLocalCursorSidecarsBounded uses fixed LOCAL_CAPS; assert via shared + // discovery path instead. + }) + // With the default 16MiB cap both fit; exercise the shared helper contract below. + expect(result.files.length).toBe(2) + + const { discoverCursorSidecarCandidates, cursorSidecarScanCancellationFromSignal } = + await import('../../shared/cursor-sidecar-scan-discovery') + const { + CURSOR_SIDECAR_SCAN_VERSION, + CURSOR_REMOTE_MAX_BUCKETS, + CURSOR_REMOTE_MAX_SESSION_DIRS, + CURSOR_REMOTE_MAX_SCOPE_PATHS, + CURSOR_SIDECAR_MAX_BYTES + } = await import('../../shared/cursor-sidecar-scan') + const response = { + version: CURSOR_SIDECAR_SCAN_VERSION, + scopeCwds: [] as string[], + sidecars: [] as never[], + issues: [] as { path: string; message: string }[], + counters: { + rootReaddir: 0, + bucketReaddir: 0, + fileLstat: 0, + boundedReads: 0, + scopeRealpath: 0, + returnedBytes: 0, + elapsedMs: 0 + }, + truncated: { scopePaths: false, buckets: false, sessionDirs: false, sidecarBytes: false } + } + const discovery = await discoverCursorSidecarCandidates({ + request: { + version: CURSOR_SIDECAR_SCAN_VERSION, + chatsRoot: chatsDir, + scopePaths: [], + maxBuckets: CURSOR_REMOTE_MAX_BUCKETS, + maxSessionDirs: CURSOR_REMOTE_MAX_SESSION_DIRS, + maxScopePaths: CURSOR_REMOTE_MAX_SCOPE_PATHS, + maxSidecarBytes: CURSOR_SIDECAR_MAX_BYTES, + maxAggregateBytes: 70 + }, + caps: { + buckets: CURSOR_REMOTE_MAX_BUCKETS, + sessions: CURSOR_REMOTE_MAX_SESSION_DIRS, + scopes: CURSOR_REMOTE_MAX_SCOPE_PATHS, + sidecarBytes: CURSOR_SIDECAR_MAX_BYTES, + aggregateBytes: 70 + }, + response, + cancellation: cursorSidecarScanCancellationFromSignal() + }) + expect(discovery?.candidates.map((candidate) => candidate.sessionId)).toEqual(['zzz-newer']) + expect(response.truncated.sidecarBytes).toBe(true) + }) + it.skipIf(process.platform === 'win32')( 'rejects symlink session dirs without opening store contents', async () => { 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 628ff75336d..1a1898f25b7 100644 --- a/src/main/ai-vault/session-scanner-cursor-local-pipeline.ts +++ b/src/main/ai-vault/session-scanner-cursor-local-pipeline.ts @@ -4,6 +4,7 @@ import { CURSOR_REMOTE_MAX_AGGREGATE_BYTES, CURSOR_SIDECAR_MAX_BYTES } from '../../shared/cursor-sidecar-scan' +import { isCursorSidecarScanCancelledError } from '../../shared/cursor-sidecar-scan-discovery' import type { ActiveSpan } from '../observability/tracer' import { parseAgentSessionFileCached } from './session-scanner-parse-cache' import type { SessionParseStats } from './session-scanner-parse-cache' @@ -98,6 +99,8 @@ type VerifiedReadTracker = { } > aggregateReturnedBytes: number + /** Serializes reserve/read/commit so concurrent parses cannot overshoot the cap. */ + gate: Promise } function createVerifiedReadTracker( @@ -124,7 +127,24 @@ function createVerifiedReadTracker( truncated: discovery.cursorDiscoveryTruncated }) } - return { byStorageKey, aggregateReturnedBytes: 0 } + return { byStorageKey, aggregateReturnedBytes: 0, gate: Promise.resolve() } +} + +async function withVerifiedReadGate( + tracker: VerifiedReadTracker, + run: () => Promise +): Promise { + const previous = tracker.gate + let release!: () => void + tracker.gate = new Promise((resolve) => { + release = resolve + }) + await previous + try { + return await run() + } finally { + release() + } } async function parseCursorGroups( @@ -161,43 +181,7 @@ async function parseCursorCandidate( const layout = candidate.cursorLayout ?? 'legacy' try { if (layout === 'sidecar') { - if (verifiedReads.aggregateReturnedBytes >= CURSOR_REMOTE_MAX_AGGREGATE_BYTES) { - const bucket = verifiedReads.byStorageKey.get(candidate.cursorStorageContextKey ?? 'native') - if (bucket) { - bucket.truncated.sidecarBytes = true - } - return null - } - const result = await parseCursorSidecarFileCached({ - file: candidate.file, - platform: args.platform, - executionHostId: args.executionHostId, - expectedRootRealPath: candidate.cursorExpectedRootRealPath - }) - const storage = verifiedReads.byStorageKey.get(candidate.cursorStorageContextKey ?? 'native') - if (!result.cacheHit && storage) { - const readBytes = Math.min(candidate.file.sizeBytes ?? 0, CURSOR_SIDECAR_MAX_BYTES) - if (verifiedReads.aggregateReturnedBytes + readBytes > CURSOR_REMOTE_MAX_AGGREGATE_BYTES) { - storage.truncated.sidecarBytes = true - storage.counters.boundedReads += 1 - return null - } - storage.counters.boundedReads += 1 - storage.counters.returnedBytes += readBytes - verifiedReads.aggregateReturnedBytes += readBytes - } - if (result.issue) { - args.issues.push(result.issue) - } - return result.evidence - ? { - layout, - storageContextKey: candidate.cursorStorageContextKey ?? 'native', - file: candidate.file, - cwdEvidence: candidate.cursorCwdEvidence, - sidecar: result.evidence - } - : null + return await parseSidecarWithAggregateCap(candidate, args, verifiedReads) } const legacy = await parseAgentSessionFileCached(candidate, args.platform, args.parseStats) return legacy @@ -210,6 +194,9 @@ async function parseCursorCandidate( } : null } catch (error) { + if (isCursorSidecarScanCancelledError(error)) { + throw error + } args.issues.push({ executionHostId: args.executionHostId, agent: 'cursor', @@ -219,3 +206,75 @@ async function parseCursorCandidate( return null } } + +async function parseSidecarWithAggregateCap( + candidate: SessionFileCandidate, + args: { + platform: NodeJS.Platform + executionHostId: ExecutionHostId + issues: AiVaultScanIssue[] + }, + verifiedReads: VerifiedReadTracker +): Promise { + const storageKey = candidate.cursorStorageContextKey ?? 'native' + const storage = verifiedReads.byStorageKey.get(storageKey) + // Unknown size reserves the per-sidecar max so concurrent admission cannot overshoot. + const estimatedBytes = + 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(verifiedReads, async () => { + if (verifiedReads.aggregateReturnedBytes >= CURSOR_REMOTE_MAX_AGGREGATE_BYTES) { + if (storage) { + storage.truncated.sidecarBytes = true + } + return null + } + if (verifiedReads.aggregateReturnedBytes + estimatedBytes > CURSOR_REMOTE_MAX_AGGREGATE_BYTES) { + if (storage) { + storage.truncated.sidecarBytes = true + } + return null + } + + const result = await parseCursorSidecarFileCached({ + file: candidate.file, + platform: args.platform, + executionHostId: args.executionHostId, + expectedRootRealPath: candidate.cursorExpectedRootRealPath + }) + + 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 + } + if (verifiedReads.aggregateReturnedBytes + readBytes > CURSOR_REMOTE_MAX_AGGREGATE_BYTES) { + if (storage) { + storage.truncated.sidecarBytes = true + } + return null + } + if (storage) { + storage.counters.returnedBytes += readBytes + } + verifiedReads.aggregateReturnedBytes += readBytes + } + + 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 + }) +} diff --git a/src/main/ai-vault/session-scanner-cursor-sidecar.ts b/src/main/ai-vault/session-scanner-cursor-sidecar.ts index 1d0bc1f7001..f19ad6ad8df 100644 --- a/src/main/ai-vault/session-scanner-cursor-sidecar.ts +++ b/src/main/ai-vault/session-scanner-cursor-sidecar.ts @@ -27,6 +27,8 @@ export type CursorSidecarParseResult = { issue: AiVaultScanIssue | null /** True when the parse reused an in-memory entry and skipped a verified read. */ cacheHit?: boolean + /** UTF-8 byte length actually returned by the verified read; 0 on cache hit. */ + returnedBytes?: number } const SIDECAR_CACHE_MAX_ENTRIES = 4096 @@ -55,13 +57,17 @@ export async function parseCursorSidecarFile(args: { if (!args.expectedRootRealPath) { throw new Error('cursor_sidecar_root_unavailable') } - return parseCursorSidecarContent({ - ...args, - content: await readVerifiedBoundedTextFile(args.file.path, { - expectedRootRealPath: args.expectedRootRealPath, - maxBytes: CURSOR_SIDECAR_MAX_BYTES - }) + const content = await readVerifiedBoundedTextFile(args.file.path, { + expectedRootRealPath: args.expectedRootRealPath, + maxBytes: CURSOR_SIDECAR_MAX_BYTES }) + return { + ...parseCursorSidecarContent({ + ...args, + content + }), + returnedBytes: Buffer.byteLength(content, 'utf8') + } } export async function parseCursorSidecarFileCached(args: { @@ -87,9 +93,13 @@ export async function parseCursorSidecarFileCached(args: { ) { sidecarCache.delete(args.file.path) sidecarCache.set(args.file.path, cached) - return { ...cached.result, cacheHit: true } + return { ...cached.result, cacheHit: true, returnedBytes: 0 } } const result = await parseCursorSidecarFile(args) + const cacheEntry = { + evidence: result.evidence, + issue: result.issue + } sidecarCache.delete(args.file.path) sidecarCache.set(args.file.path, { mtimeMs: args.file.mtimeMs, @@ -101,7 +111,7 @@ export async function parseCursorSidecarFileCached(args: { platform: args.platform, executionHostId: args.executionHostId ?? null, expectedRootRealPath: args.expectedRootRealPath ?? null, - result + result: cacheEntry }) if (sidecarCache.size > SIDECAR_CACHE_MAX_ENTRIES) { const oldest = sidecarCache.keys().next() diff --git a/src/relay/cursor-sidecar-scan.test.ts b/src/relay/cursor-sidecar-scan.test.ts index 3cf782a5f39..43a60730909 100644 --- a/src/relay/cursor-sidecar-scan.test.ts +++ b/src/relay/cursor-sidecar-scan.test.ts @@ -257,6 +257,64 @@ describe('scanCursorSidecars', () => { ).rejects.toThrow('cursor_sidecar_scan_cancelled') }) + it('rejects when cancellation flips after the first enumeration check', async () => { + const root = await createRoot() + const chatsRoot = join(root, 'chats') + await addSession(chatsRoot, 'cccccccccccccccccccccccccccccccc', 'alive') + let checks = 0 + await expect( + scanCursorSidecars(defaultCursorSidecarScanRequest(chatsRoot, [], process.platform), { + clientId: 1, + isStale: () => { + checks += 1 + return checks > 1 + } + }) + ).rejects.toThrow('cursor_sidecar_scan_cancelled') + }) + + it('retains the newer equal-size session under a tight aggregate byte cap', async () => { + const root = await createRoot() + const chatsRoot = join(root, 'chats') + const content = 'y'.repeat(70) + const bucket = 'dddddddddddddddddddddddddddddddd' + await addSession(chatsRoot, bucket, 'aaa-older', content) + await addSession(chatsRoot, bucket, 'zzz-newer', content) + await setSessionMtime(chatsRoot, bucket, 'aaa-older', 1_000) + await setSessionMtime(chatsRoot, bucket, 'zzz-newer', 9_000) + const request = defaultCursorSidecarScanRequest(chatsRoot, [], process.platform) + request.maxAggregateBytes = 70 + + const result = await scanCursorSidecars(request, context) + expect(result.sidecars.map((sidecar) => sidecar.sessionId)).toEqual(['zzz-newer']) + expect(result.counters.returnedBytes).toBe(70) + expect(result.truncated.sidecarBytes).toBe(true) + }) + + it('does not return all 70 large sidecars past the 16 MiB aggregate', async () => { + const root = await createRoot() + const chatsRoot = join(root, 'chats') + const bucket = 'eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee' + const payload = `${'{"createdAtMs":1,"updatedAtMs":2,"hasConversation":true,"pad":"'}${'b'.repeat(249_990)}"}` + const payloadBytes = Buffer.byteLength(payload, 'utf8') + expect(payloadBytes * 70).toBeGreaterThan(16_777_216) + for (let index = 0; index < 70; index += 1) { + await addSession(chatsRoot, bucket, `s-${String(index).padStart(3, '0')}`, payload) + } + const result = await scanCursorSidecars( + defaultCursorSidecarScanRequest(chatsRoot, [], process.platform), + context + ) + expect(result.sidecars.length).toBeLessThan(70) + expect(result.counters.returnedBytes).toBeLessThanOrEqual(16_777_216) + expect(result.truncated.sidecarBytes).toBe(true) + const contentTotal = result.sidecars.reduce( + (total, sidecar) => total + Buffer.byteLength(sidecar.content, 'utf8'), + 0 + ) + expect(contentTotal).toBe(result.counters.returnedBytes) + }) + it.skipIf(process.platform === 'win32')( 'continues after an unreadable bucket and reports its path', async () => { diff --git a/src/relay/cursor-sidecar-scan.ts b/src/relay/cursor-sidecar-scan.ts index e2e8fb1563c..54634c378c8 100644 --- a/src/relay/cursor-sidecar-scan.ts +++ b/src/relay/cursor-sidecar-scan.ts @@ -16,6 +16,7 @@ import { readVerifiedBoundedTextFile } from '../shared/node-verified-bounded-tex import type { RequestContext } from './dispatcher' import { discoverCursorSidecarCandidates, + isCursorSidecarScanCancelledError, type CursorSidecarScanCandidate, type CursorSidecarScanCaps, type CursorSidecarScanCancellation @@ -61,7 +62,18 @@ async function readCandidates( ): Promise { for (let index = 0; index < candidates.length; index += 1) { throwIfCancelled(context) + if (response.counters.returnedBytes >= caps.aggregateBytes) { + if (index < candidates.length) { + response.truncated.sidecarBytes = true + } + break + } const candidate = candidates[index] + // Pre-check statted size so we do not open a sidecar that cannot fit. + if (response.counters.returnedBytes + candidate.meta.size > caps.aggregateBytes) { + response.truncated.sidecarBytes = true + break + } try { response.counters.boundedReads++ const content = await readVerifiedBoundedTextFile(candidate.metaPath, { @@ -89,6 +101,9 @@ async function readCandidates( break } } catch (error) { + if (isCursorSidecarScanCancelledError(error)) { + throw error + } if (!isMissing(error)) { addIssue(response, candidate.metaPath, error) } diff --git a/src/shared/cursor-sidecar-scan-directory.ts b/src/shared/cursor-sidecar-scan-directory.ts index 6095096c941..3ed6320b7bb 100644 --- a/src/shared/cursor-sidecar-scan-directory.ts +++ b/src/shared/cursor-sidecar-scan-directory.ts @@ -2,6 +2,9 @@ import type { Dirent } from 'node:fs' import { opendir, readdir } from 'node:fs/promises' import { posix, win32 } from 'node:path' +/** Per-directory dirent examination budget; keeps cold scans hard-bounded. */ +export const CURSOR_DIR_MAX_ENTRIES_EXAMINED = 8_192 + /** Returns true when `name` caused the retained set to overflow the limit. */ export function retainLexicographic(selected: string[], name: string, limit: number): boolean { if (selected.length < limit) { @@ -24,16 +27,51 @@ export function retainLexicographic(selected: string[], name: string, limit: num return true } +export type StreamDirectoryNamesOptions = { + /** Invoked once per dirent observed (opendir read or readdir fallback entry). */ + onDirent?: () => void + /** Hard stop after this many dirents; default CURSOR_DIR_MAX_ENTRIES_EXAMINED. */ + maxEntriesExamined?: number +} + +/** + * Streams directory names without materializing the full listing. + * Retention callers keep only their bounded selection; examination stops at + * maxEntriesExamined so cold adversarial directories cannot unbounded-walk. + */ export async function streamDirectoryNames( dirPath: string, - visit: (name: string, entry: Dirent) => void -): Promise { + visit: (name: string, entry: Dirent) => void, + options: StreamDirectoryNamesOptions = {} +): Promise<{ entriesExamined: number; examinationTruncated: boolean }> { + const maxEntries = options.maxEntriesExamined ?? CURSOR_DIR_MAX_ENTRIES_EXAMINED + let entriesExamined = 0 + let examinationTruncated = false + + const consider = (name: string, entry: Dirent): boolean => { + if (entriesExamined >= maxEntries) { + examinationTruncated = true + return false + } + entriesExamined += 1 + options.onDirent?.() + visit(name, entry) + return entriesExamined < maxEntries + } + try { const directory = await opendir(dirPath) - for await (const entry of directory) { - visit(entry.name, entry) + try { + for await (const entry of directory) { + if (!consider(entry.name, entry)) { + break + } + } + } finally { + // Closing mid-iteration is best-effort; for-await also closes on break/throw. + await directory.close().catch(() => undefined) } - return + return { entriesExamined, examinationTruncated } } catch (error) { // Fallback keeps remote-wire hosts that only expose readdir working. if (!isUnsupportedDirectoryStream(error)) { @@ -42,32 +80,52 @@ export async function streamDirectoryNames( } const entries = await readdir(dirPath, { withFileTypes: true }) for (const entry of entries) { - visit(entry.name, entry) + if (!consider(entry.name, entry)) { + break + } } + return { entriesExamined, examinationTruncated } } +/** + * Deterministic retention: among accepted names in the first maxEntriesExamined + * dirents (opendir/readdir order), keep the lexicographic first `limit`. + * Does not materialize the full directory; only the retained name set is kept. + */ export async function listLexicographicDirectoryNames(args: { dirPath: string limit: number accept: (name: string, entry: Dirent) => boolean -}): Promise<{ names: string[]; truncated: boolean }> { + maxEntriesExamined?: number + onDirent?: () => void +}): Promise<{ names: string[]; truncated: boolean; entriesExamined: number }> { if (args.limit <= 0) { - return { names: [], truncated: true } + return { names: [], truncated: true, entriesExamined: 0 } } const selected: string[] = [] let truncated = false - await streamDirectoryNames(args.dirPath, (name, entry) => { - if (!args.accept(name, entry)) { - return + const { entriesExamined, examinationTruncated } = await streamDirectoryNames( + args.dirPath, + (name, entry) => { + if (!args.accept(name, entry)) { + return + } + if (retainLexicographic(selected, name, args.limit)) { + truncated = true + } + }, + { + maxEntriesExamined: args.maxEntriesExamined, + onDirent: args.onDirent } - if (retainLexicographic(selected, name, args.limit)) { - truncated = true - } - }) + ) + if (examinationTruncated) { + truncated = true + } if (selected.length < args.limit) { selected.sort((left, right) => left.localeCompare(right)) } - return { names: selected, truncated } + return { names: selected, truncated, entriesExamined } } export function targetPathVariants(value: string, pathPlatform: NodeJS.Platform): string[] { diff --git a/src/shared/cursor-sidecar-scan-discovery.ts b/src/shared/cursor-sidecar-scan-discovery.ts index 0eef07c174b..2ddc3ca19ca 100644 --- a/src/shared/cursor-sidecar-scan-discovery.ts +++ b/src/shared/cursor-sidecar-scan-discovery.ts @@ -4,6 +4,7 @@ import { lstat, realpath } from 'node:fs/promises' import { join } from 'node:path' import type { CursorSidecarScanRequest, CursorSidecarScanResponse } from './cursor-sidecar-scan' import { + CURSOR_DIR_MAX_ENTRIES_EXAMINED, listLexicographicDirectoryNames, safeBasename, targetPathVariants @@ -11,6 +12,8 @@ import { const BUCKET_PATTERN = /^[0-9a-f]{32}$/u const BUCKET_READ_CONCURRENCY = 8 +/** Check cancellation every N dirents during cold directory walks. */ +const CANCEL_CHECK_EVERY_DIRENTS = 64 export type CursorSidecarScanCaps = { buckets: number @@ -25,6 +28,11 @@ export type CursorSidecarScanCancellation = { } type Bucket = { name: string; path: string; scopeCwd: string | null } +type ScanArgs = { + caps: CursorSidecarScanCaps + response: CursorSidecarScanResponse + cancellation: CursorSidecarScanCancellation +} export type CursorSidecarScanCandidate = Bucket & { sessionId: string @@ -33,6 +41,10 @@ export type CursorSidecarScanCandidate = Bucket & { store: Stats } +export function isCursorSidecarScanCancelledError(error: unknown): boolean { + return error instanceof Error && error.message === 'cursor_sidecar_scan_cancelled' +} + export function cursorSidecarScanCancellationFromSignal( signal?: AbortSignal ): CursorSidecarScanCancellation { @@ -59,31 +71,27 @@ export async function discoverCursorSidecarCandidates(args: { let rootRealPath: string try { rootRealPath = await realpath(chatsRoot) - args.response.counters.rootReaddir++ } catch (error) { + rethrowCancel(error) if (!isMissing(error)) { addIssue(args.response, chatsRoot, error) } return null } + // First enumeration-phase cancel check (after root resolve, before walks). args.cancellation.throwIfCancelled() const direct = await scopeBuckets(args.request, chatsRoot, pathPlatform, args) const enumerated = await enumeratedBuckets(chatsRoot, direct, args) const sessions = await retainSessions([...direct.values(), ...enumerated], args) - const candidates = await eligibleCandidates(sessions, args) - return { rootRealPath, candidates } + return { rootRealPath, candidates: await eligibleCandidates(sessions, args) } } async function scopeBuckets( request: CursorSidecarScanRequest, chatsRoot: string, pathPlatform: NodeJS.Platform, - args: { - caps: CursorSidecarScanCaps - response: CursorSidecarScanResponse - cancellation: CursorSidecarScanCancellation - } + args: ScanArgs ): Promise> { // Truncation is measured on the full unique list before the cap slice. const paths = [...new Set(request.scopePaths.map((value) => value.trim()).filter(Boolean))].sort() @@ -99,7 +107,8 @@ async function scopeBuckets( for (const cwd of targetPathVariants(resolved, pathPlatform)) { cwds.add(cwd) } - } catch { + } catch (error) { + rethrowCancel(error) // Scope paths are allowed to be absent on the owning host. } args.cancellation.throwIfCancelled() @@ -117,26 +126,28 @@ async function scopeBuckets( async function enumeratedBuckets( chatsRoot: string, direct: ReadonlyMap, - args: { - caps: CursorSidecarScanCaps - response: CursorSidecarScanResponse - cancellation: CursorSidecarScanCancellation - } + args: ScanArgs ): Promise { + // Cancel check must sit outside the listing try so it is never turned into an issue. + args.cancellation.throwIfCancelled() try { + args.response.counters.rootReaddir++ const { names, truncated } = await listLexicographicDirectoryNames({ dirPath: chatsRoot, limit: args.caps.buckets, + maxEntriesExamined: CURSOR_DIR_MAX_ENTRIES_EXAMINED, accept: (name, entry) => entry.isDirectory() && !entry.isSymbolicLink() && BUCKET_PATTERN.test(name) && - !direct.has(name) + !direct.has(name), + onDirent: createDirentCancelChecker(args.cancellation) }) args.cancellation.throwIfCancelled() args.response.truncated.buckets = truncated return names.map((name) => ({ name, path: join(chatsRoot, name), scopeCwd: null })) } catch (error) { + rethrowCancel(error) if (!isMissing(error)) { addIssue(args.response, chatsRoot, error) } @@ -146,11 +157,7 @@ async function enumeratedBuckets( async function retainSessions( buckets: readonly Bucket[], - args: { - caps: CursorSidecarScanCaps - response: CursorSidecarScanResponse - cancellation: CursorSidecarScanCancellation - } + args: ScanArgs ): Promise<(Bucket & { sessionId: string })[]> { const retained: (Bucket & { sessionId: string })[] = [] for ( @@ -158,33 +165,10 @@ async function retainSessions( index < buckets.length && retained.length < args.caps.sessions; index += BUCKET_READ_CONCURRENCY ) { + args.cancellation.throwIfCancelled() const batch = buckets.slice(index, index + BUCKET_READ_CONCURRENCY) const listings = await Promise.all( - batch.map(async (bucket) => { - try { - if (bucket.scopeCwd) { - args.response.counters.fileLstat++ - const stats = await lstat(bucket.path) - if (!stats.isDirectory() || stats.isSymbolicLink()) { - return { bucket, names: [] as string[], truncated: false } - } - } - args.response.counters.bucketReaddir++ - const remaining = Math.max(0, args.caps.sessions - retained.length) - const listed = await listLexicographicDirectoryNames({ - dirPath: bucket.path, - limit: remaining, - accept: (name, entry) => - entry.isDirectory() && !entry.isSymbolicLink() && safeBasename(name) - }) - return { bucket, names: listed.names, truncated: listed.truncated } - } catch (error) { - if (!isMissing(error)) { - addIssue(args.response, bucket.path, error) - } - return { bucket, names: [] as string[], truncated: false } - } - }) + batch.map((bucket) => listBucketSessions(bucket, args, retained.length)) ) args.cancellation.throwIfCancelled() for (const { bucket, names, truncated } of listings) { @@ -205,77 +189,126 @@ async function retainSessions( return retained } -async function eligibleCandidates( - sessions: readonly (Bucket & { sessionId: string })[], - args: { - caps: CursorSidecarScanCaps - response: CursorSidecarScanResponse - cancellation: CursorSidecarScanCancellation - } -): Promise { - const candidates: CursorSidecarScanCandidate[] = [] - let aggregateBytes = 0 - for (let index = 0; index < sessions.length; index += BUCKET_READ_CONCURRENCY) { - const batch = sessions.slice(index, index + BUCKET_READ_CONCURRENCY) - const results = await Promise.all( - batch.map(async (session): Promise => { - const sessionDir = join(session.path, session.sessionId) - const metaPath = join(sessionDir, 'meta.json') - try { - args.response.counters.fileLstat += 2 - const [meta, store] = await Promise.all([ - lstat(metaPath), - lstat(join(sessionDir, 'store.db')) - ]) - if ( - !meta.isFile() || - meta.isSymbolicLink() || - !store.isFile() || - store.isSymbolicLink() - ) { - return null - } - if (meta.size > args.caps.sidecarBytes) { - addIssue( - args.response, - metaPath, - new Error('Cursor session metadata exceeds the read limit.') - ) - return null - } - return { ...session, metaPath, meta, store } - } catch (error) { - if (!isMissing(error)) { - addIssue(args.response, metaPath, error) - } - return null - } - }) - ) - args.cancellation.throwIfCancelled() - for (const result of results) { - if (!result) { - continue +async function listBucketSessions( + bucket: Bucket, + args: ScanArgs, + retainedCount: number +): Promise<{ bucket: Bucket; names: string[]; truncated: boolean }> { + try { + if (bucket.scopeCwd) { + args.response.counters.fileLstat++ + const stats = await lstat(bucket.path) + if (!stats.isDirectory() || stats.isSymbolicLink()) { + return { bucket, names: [], truncated: false } } - if (aggregateBytes + result.meta.size > args.caps.aggregateBytes) { - args.response.truncated.sidecarBytes = true - return sortCandidates(candidates) - } - aggregateBytes += result.meta.size - candidates.push(result) } + args.response.counters.bucketReaddir++ + const listed = await listLexicographicDirectoryNames({ + dirPath: bucket.path, + limit: Math.max(0, args.caps.sessions - retainedCount), + maxEntriesExamined: CURSOR_DIR_MAX_ENTRIES_EXAMINED, + accept: (name, entry) => entry.isDirectory() && !entry.isSymbolicLink() && safeBasename(name), + onDirent: createDirentCancelChecker(args.cancellation) + }) + return { bucket, names: listed.names, truncated: listed.truncated } + } catch (error) { + rethrowCancel(error) + if (!isMissing(error)) { + addIssue(args.response, bucket.path, error) + } + return { bucket, names: [], truncated: false } } - return sortCandidates(candidates) } -function sortCandidates(candidates: CursorSidecarScanCandidate[]): CursorSidecarScanCandidate[] { - return candidates.sort( +async function eligibleCandidates( + sessions: readonly (Bucket & { sessionId: string })[], + args: ScanArgs +): Promise { + // Collect every eligible candidate first, then apply newest-first retention + // before the aggregate byte budget (stable lexical ties). + const eligible: CursorSidecarScanCandidate[] = [] + for (let index = 0; index < sessions.length; index += BUCKET_READ_CONCURRENCY) { + args.cancellation.throwIfCancelled() + const batch = sessions.slice(index, index + BUCKET_READ_CONCURRENCY) + const results = await Promise.all(batch.map((session) => inspectSession(session, args))) + args.cancellation.throwIfCancelled() + for (const result of results) { + if (result) { + eligible.push(result) + } + } + } + return retainByNewestThenAggregate(eligible, args) +} + +async function inspectSession( + session: Bucket & { sessionId: string }, + args: ScanArgs +): Promise { + const sessionDir = join(session.path, session.sessionId) + const metaPath = join(sessionDir, 'meta.json') + try { + args.response.counters.fileLstat += 2 + const [meta, store] = await Promise.all([lstat(metaPath), lstat(join(sessionDir, 'store.db'))]) + if (!meta.isFile() || meta.isSymbolicLink() || !store.isFile() || store.isSymbolicLink()) { + return null + } + if (meta.size > args.caps.sidecarBytes) { + addIssue( + args.response, + metaPath, + new Error('Cursor session metadata exceeds the read limit.') + ) + return null + } + return { ...session, metaPath, meta, store } + } catch (error) { + rethrowCancel(error) + if (!isMissing(error)) { + addIssue(args.response, metaPath, error) + } + return null + } +} + +function retainByNewestThenAggregate( + eligible: CursorSidecarScanCandidate[], + args: ScanArgs +): CursorSidecarScanCandidate[] { + const ranked = eligible.sort( (left, right) => Number(right.scopeCwd !== null) - Number(left.scopeCwd !== null) || Math.max(right.meta.mtimeMs, right.store.mtimeMs) - Math.max(left.meta.mtimeMs, left.store.mtimeMs) || `${left.name}\0${left.sessionId}`.localeCompare(`${right.name}\0${right.sessionId}`) ) + const retained: CursorSidecarScanCandidate[] = [] + let aggregateBytes = 0 + for (const candidate of ranked) { + if (aggregateBytes + candidate.meta.size > args.caps.aggregateBytes) { + args.response.truncated.sidecarBytes = true + break + } + aggregateBytes += candidate.meta.size + retained.push(candidate) + } + return retained +} + +function createDirentCancelChecker(cancellation: CursorSidecarScanCancellation): () => void { + let seen = 0 + return () => { + seen += 1 + if (seen % CANCEL_CHECK_EVERY_DIRENTS === 0) { + cancellation.throwIfCancelled() + } + } +} + +function rethrowCancel(error: unknown): void { + if (isCursorSidecarScanCancelledError(error)) { + throw error + } } function addIssue(response: CursorSidecarScanResponse, path: string, error: unknown): void {