mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 16:03:47 +00:00
fix(ai-vault): harden Cursor sidecar discovery cancel, retention, and caps
Rethrow scan cancellation from every enumeration phase, rank newest-first before the aggregate byte budget, enforce the 16 MiB cap on actual verified read bytes, and hard-bound cold directory walks with cancel-aware streaming.
This commit is contained in:
@@ -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 () => {
|
||||
|
||||
@@ -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<void>
|
||||
}
|
||||
|
||||
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<T>(
|
||||
tracker: VerifiedReadTracker,
|
||||
run: () => Promise<T>
|
||||
): Promise<T> {
|
||||
const previous = tracker.gate
|
||||
let release!: () => void
|
||||
tracker.gate = new Promise<void>((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<ParsedCursorCandidate | null> {
|
||||
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
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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<void> {
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
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[] {
|
||||
|
||||
@@ -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<Map<string, Bucket>> {
|
||||
// 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<string, Bucket>,
|
||||
args: {
|
||||
caps: CursorSidecarScanCaps
|
||||
response: CursorSidecarScanResponse
|
||||
cancellation: CursorSidecarScanCancellation
|
||||
}
|
||||
args: ScanArgs
|
||||
): Promise<Bucket[]> {
|
||||
// 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<CursorSidecarScanCandidate[]> {
|
||||
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<CursorSidecarScanCandidate | null> => {
|
||||
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<CursorSidecarScanCandidate[]> {
|
||||
// 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<CursorSidecarScanCandidate | null> {
|
||||
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 {
|
||||
|
||||
Reference in New Issue
Block a user