mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 16:03:47 +00:00
fix: harden Cursor sidecar discovery bounds
This commit is contained in:
@@ -264,6 +264,28 @@ describe('local Cursor sidecar discovery bounds', () => {
|
||||
).rejects.toThrow('cursor_sidecar_scan_cancelled')
|
||||
})
|
||||
|
||||
it('honors cancellation that lands while resolving a missing chats root', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-cursor-cancel-missing-'))
|
||||
tempRoots.push(root)
|
||||
let checks = 0
|
||||
const signal = {
|
||||
get aborted() {
|
||||
checks += 1
|
||||
return checks > 1
|
||||
}
|
||||
} as AbortSignal
|
||||
|
||||
await expect(
|
||||
discoverLocalCursorSidecarsBounded({
|
||||
chatsDir: join(root, 'missing-chats'),
|
||||
scopePaths: [],
|
||||
issues: [],
|
||||
signal
|
||||
})
|
||||
).rejects.toThrow('cursor_sidecar_scan_cancelled')
|
||||
expect(checks).toBe(2)
|
||||
})
|
||||
|
||||
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)
|
||||
@@ -276,7 +298,7 @@ describe('local Cursor sidecar discovery bounds', () => {
|
||||
const signal = {
|
||||
get aborted() {
|
||||
checks += 1
|
||||
// First check (post-realpath) passes; later enumeration-phase checks abort.
|
||||
// The pre-realpath check passes; the post-realpath check observes cancellation.
|
||||
return checks > 1
|
||||
}
|
||||
} as AbortSignal
|
||||
|
||||
@@ -12,7 +12,7 @@ import { cursorBucketForCwd } from '../main/ai-vault/session-scanner-cursor-path
|
||||
|
||||
const roots: string[] = []
|
||||
const context = { clientId: 1, isStale: () => false }
|
||||
const CHECKS_BEFORE_FIRST_VERIFIED_READ = 8
|
||||
const CHECKS_BEFORE_FIRST_VERIFIED_READ = 9
|
||||
|
||||
async function createRoot(): Promise<string> {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-cursor-scan-'))
|
||||
@@ -89,6 +89,48 @@ describe('scanCursorSidecars', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('shares bounded session capacity across exact scope buckets', async () => {
|
||||
const root = await createRoot()
|
||||
const chatsRoot = join(root, 'chats')
|
||||
const scopes = [join(root, 'a-workspace'), join(root, 'z-workspace')]
|
||||
await Promise.all(scopes.map((scope) => mkdir(scope)))
|
||||
const firstBucket = cursorBucketForCwd(scopes[0], process.platform)
|
||||
await Promise.all([
|
||||
addSession(chatsRoot, firstBucket, 'first-a'),
|
||||
addSession(chatsRoot, firstBucket, 'first-b'),
|
||||
addSession(chatsRoot, cursorBucketForCwd(scopes[1], process.platform), 'second')
|
||||
])
|
||||
const request = defaultCursorSidecarScanRequest(chatsRoot, scopes, process.platform)
|
||||
request.maxSessionDirs = 2
|
||||
|
||||
const result = await scanCursorSidecars(request, context)
|
||||
|
||||
expect(result.sidecars.map((sidecar) => sidecar.sessionId).sort()).toEqual([
|
||||
'first-a',
|
||||
'second'
|
||||
])
|
||||
expect(result.truncated.sessionDirs).toBe(true)
|
||||
})
|
||||
|
||||
it('keeps exact scope sessions within the requested cap', async () => {
|
||||
const root = await createRoot()
|
||||
const chatsRoot = join(root, 'chats')
|
||||
const scope = join(root, 'workspace')
|
||||
await mkdir(scope)
|
||||
const bucket = cursorBucketForCwd(scope, process.platform)
|
||||
await Promise.all([
|
||||
addSession(chatsRoot, bucket, 'first-scoped'),
|
||||
addSession(chatsRoot, bucket, 'second-scoped')
|
||||
])
|
||||
const request = defaultCursorSidecarScanRequest(chatsRoot, [scope], process.platform)
|
||||
request.maxSessionDirs = 1
|
||||
|
||||
const result = await scanCursorSidecars(request, context)
|
||||
|
||||
expect(result.sidecars).toHaveLength(1)
|
||||
expect(result.truncated.sessionDirs).toBe(true)
|
||||
})
|
||||
|
||||
it.skipIf(process.platform === 'win32')(
|
||||
'keeps missing files silent and rejects symlinked sidecars',
|
||||
async () => {
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
export type CursorSidecarScanCancellation = {
|
||||
throwIfCancelled: () => void
|
||||
}
|
||||
|
||||
export function isCursorSidecarScanCancelledError(error: unknown): boolean {
|
||||
return error instanceof Error && error.message === 'cursor_sidecar_scan_cancelled'
|
||||
}
|
||||
|
||||
export function cursorSidecarScanCancellationFromSignal(
|
||||
signal?: AbortSignal
|
||||
): CursorSidecarScanCancellation {
|
||||
return {
|
||||
throwIfCancelled: () => {
|
||||
if (signal?.aborted) {
|
||||
throw new Error('cursor_sidecar_scan_cancelled')
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { Dirent } from 'node:fs'
|
||||
import { mkdir, mkdtemp, readdir, rm, writeFile } from 'node:fs/promises'
|
||||
import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
@@ -85,22 +85,14 @@ describe('cursor sidecar directory examination bounds', () => {
|
||||
expect(onDirentCount).toBe(budget)
|
||||
})
|
||||
|
||||
it('hard-bounds the readdir fallback without retaining an unbounded listing', async () => {
|
||||
it('fails bounded when directory streaming is unsupported', async () => {
|
||||
const root = await tempDir('orca-cursor-dir-fallback-')
|
||||
const budget = 16
|
||||
const names = await fillDirectory(root, budget + 8)
|
||||
await fillDirectory(root, budget + 8)
|
||||
|
||||
let readdirCalls = 0
|
||||
let observedListingLength = 0
|
||||
setStreamDirectoryIoForTests({
|
||||
opendir: async () => {
|
||||
throw unsupportedDirectoryStreamError()
|
||||
},
|
||||
readdirWithFileTypes: async (dirPath) => {
|
||||
readdirCalls += 1
|
||||
const entries = await readdir(dirPath, { withFileTypes: true })
|
||||
observedListingLength = entries.length
|
||||
return entries
|
||||
}
|
||||
})
|
||||
|
||||
@@ -113,13 +105,8 @@ describe('cursor sidecar directory examination bounds', () => {
|
||||
{ maxEntriesExamined: budget }
|
||||
)
|
||||
|
||||
expect(readdirCalls).toBe(1)
|
||||
expect(result.entriesExamined).toBe(budget)
|
||||
expect(result.examinationTruncated).toBe(true)
|
||||
expect(visited).toHaveLength(budget)
|
||||
expect(observedListingLength).toBe(names.length)
|
||||
// After return, the eager listing must not remain as a retained examination set.
|
||||
expect(visited.every((name) => names.includes(name))).toBe(true)
|
||||
expect(result).toEqual({ entriesExamined: 0, examinationTruncated: true })
|
||||
expect(visited).toEqual([])
|
||||
})
|
||||
|
||||
it('propagates cancellation from onDirent during examination', async () => {
|
||||
@@ -140,29 +127,6 @@ describe('cursor sidecar directory examination bounds', () => {
|
||||
expect(seen).toBe(10)
|
||||
})
|
||||
|
||||
it('propagates cancellation on the readdir fallback path', async () => {
|
||||
const root = await tempDir('orca-cursor-dir-cancel-fallback-')
|
||||
await fillDirectory(root, 40)
|
||||
setStreamDirectoryIoForTests({
|
||||
opendir: async () => {
|
||||
throw unsupportedDirectoryStreamError()
|
||||
}
|
||||
})
|
||||
let seen = 0
|
||||
await expect(
|
||||
streamDirectoryNames(root, () => undefined, {
|
||||
maxEntriesExamined: 40,
|
||||
onDirent: () => {
|
||||
seen += 1
|
||||
if (seen >= 5) {
|
||||
throw new Error('cursor_sidecar_scan_cancelled')
|
||||
}
|
||||
}
|
||||
})
|
||||
).rejects.toThrow('cursor_sidecar_scan_cancelled')
|
||||
expect(seen).toBe(5)
|
||||
})
|
||||
|
||||
it('keeps listLexicographic examination-truncated when the walk hits the budget with more left', async () => {
|
||||
const root = await tempDir('orca-cursor-dir-list-trunc-')
|
||||
await fillDirectory(root, 20, 'z')
|
||||
@@ -226,7 +190,7 @@ describe('cursor sidecar directory examination bounds', () => {
|
||||
expect(listed.names).toHaveLength(64)
|
||||
}, 60_000)
|
||||
|
||||
it('fallback path never visits more than the examination budget on a 10k store', async () => {
|
||||
it('unsupported streaming does not traverse a 10k store', async () => {
|
||||
const root = await tempDir('orca-cursor-dir-10k-fallback-')
|
||||
await fillDirectory(root, 10_000)
|
||||
setStreamDirectoryIoForTests({
|
||||
@@ -243,9 +207,8 @@ describe('cursor sidecar directory examination bounds', () => {
|
||||
},
|
||||
{ maxEntriesExamined: budget }
|
||||
)
|
||||
expect(visits).toBe(budget)
|
||||
expect(result.entriesExamined).toBe(budget)
|
||||
expect(result.examinationTruncated).toBe(true)
|
||||
expect(visits).toBe(0)
|
||||
expect(result).toEqual({ entriesExamined: 0, examinationTruncated: true })
|
||||
}, 60_000)
|
||||
|
||||
it('retainLexicographic keeps only the first `limit` names in order', () => {
|
||||
@@ -257,6 +220,32 @@ describe('cursor sidecar directory examination bounds', () => {
|
||||
expect(selected).toEqual(['a', 'b', 'm'])
|
||||
})
|
||||
|
||||
it('bounds comparisons for reverse-ordered retained names', () => {
|
||||
const originalLocaleCompare = String.prototype.localeCompare
|
||||
let comparisons = 0
|
||||
const localeCompare = vi.spyOn(String.prototype, 'localeCompare').mockImplementation(function (
|
||||
this: string,
|
||||
value: string
|
||||
) {
|
||||
comparisons += 1
|
||||
return originalLocaleCompare.call(this, value)
|
||||
})
|
||||
const selected: string[] = []
|
||||
|
||||
try {
|
||||
for (let index = 255; index >= 0; index -= 1) {
|
||||
retainLexicographic(selected, String(index).padStart(3, '0'), 64)
|
||||
}
|
||||
} finally {
|
||||
localeCompare.mockRestore()
|
||||
}
|
||||
|
||||
expect(comparisons).toBeLessThan(2_500)
|
||||
expect(selected).toEqual(
|
||||
Array.from({ length: 64 }, (_, index) => String(index).padStart(3, '0'))
|
||||
)
|
||||
})
|
||||
|
||||
it('treats a zero examination budget as already truncated', async () => {
|
||||
const root = await tempDir('orca-cursor-dir-zero-')
|
||||
await fillDirectory(root, 3)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { Dir, Dirent } from 'node:fs'
|
||||
import { opendir, readdir, realpath } from 'node:fs/promises'
|
||||
import { opendir, realpath } from 'node:fs/promises'
|
||||
import { posix, win32 } from 'node:path'
|
||||
|
||||
/** Per-directory dirent examination budget; keeps cold scans hard-bounded. */
|
||||
@@ -18,12 +18,18 @@ export function retainLexicographic(selected: string[], name: string, limit: num
|
||||
if (name.localeCompare(last) >= 0) {
|
||||
return true
|
||||
}
|
||||
let index = limit - 1
|
||||
while (index > 0 && name.localeCompare(selected[index - 1]) < 0) {
|
||||
selected[index] = selected[index - 1]
|
||||
index -= 1
|
||||
let low = 0
|
||||
let high = limit
|
||||
while (low < high) {
|
||||
const middle = Math.floor((low + high) / 2)
|
||||
if (name.localeCompare(selected[middle]) < 0) {
|
||||
high = middle
|
||||
} else {
|
||||
low = middle + 1
|
||||
}
|
||||
}
|
||||
selected[index] = name
|
||||
selected.splice(low, 0, name)
|
||||
selected.pop()
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -36,22 +42,18 @@ export type StreamDirectoryNamesOptions = {
|
||||
|
||||
type StreamDirectoryIo = {
|
||||
opendir: (path: string) => ReturnType<typeof opendir>
|
||||
readdirWithFileTypes: (path: string) => Promise<Dirent[]>
|
||||
}
|
||||
|
||||
const defaultStreamDirectoryIo: StreamDirectoryIo = {
|
||||
opendir,
|
||||
readdirWithFileTypes: (path) => readdir(path, { withFileTypes: true })
|
||||
opendir
|
||||
}
|
||||
let streamDirectoryIo: StreamDirectoryIo = defaultStreamDirectoryIo
|
||||
|
||||
/** Test isolation for opendir vs readdir-fallback paths. */
|
||||
/** Test isolation for streamed and unsupported directory paths. */
|
||||
export function setStreamDirectoryIoForTests(next?: Partial<StreamDirectoryIo>): void {
|
||||
streamDirectoryIo = next
|
||||
? {
|
||||
opendir: next.opendir ?? defaultStreamDirectoryIo.opendir,
|
||||
readdirWithFileTypes:
|
||||
next.readdirWithFileTypes ?? defaultStreamDirectoryIo.readdirWithFileTypes
|
||||
opendir: next.opendir ?? defaultStreamDirectoryIo.opendir
|
||||
}
|
||||
: defaultStreamDirectoryIo
|
||||
}
|
||||
@@ -82,12 +84,12 @@ export async function streamDirectoryNames(
|
||||
await directory.close().catch(() => undefined)
|
||||
}
|
||||
} catch (error) {
|
||||
// Fallback keeps remote-wire hosts that only expose readdir working.
|
||||
if (!isUnsupportedDirectoryStream(error)) {
|
||||
throw error
|
||||
}
|
||||
// readdir is eager and cannot preserve this function's memory/IO bound.
|
||||
return { entriesExamined: 0, examinationTruncated: true }
|
||||
}
|
||||
return examineReaddirFallback(dirPath, visit, options, maxEntries)
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -198,33 +200,6 @@ async function examineDirectoryStream(
|
||||
return { entriesExamined, examinationTruncated }
|
||||
}
|
||||
|
||||
async function examineReaddirFallback(
|
||||
dirPath: string,
|
||||
visit: (name: string, entry: Dirent) => void,
|
||||
options: StreamDirectoryNamesOptions,
|
||||
maxEntries: number
|
||||
): Promise<{ entriesExamined: number; examinationTruncated: boolean }> {
|
||||
// Why: readdir eagerly allocates the full listing. Shrink immediately to the
|
||||
// examination window so adversarial directories cannot keep an unbounded
|
||||
// Dirent array alive past the budget (peak alloc is platform-limited).
|
||||
const entries = await streamDirectoryIo.readdirWithFileTypes(dirPath)
|
||||
let entriesExamined = 0
|
||||
try {
|
||||
const examinationTruncated = entries.length > maxEntries
|
||||
if (examinationTruncated) {
|
||||
entries.length = maxEntries
|
||||
}
|
||||
for (const entry of entries) {
|
||||
entriesExamined += 1
|
||||
options.onDirent?.()
|
||||
visit(entry.name, entry)
|
||||
}
|
||||
return { entriesExamined, examinationTruncated }
|
||||
} finally {
|
||||
entries.length = 0
|
||||
}
|
||||
}
|
||||
|
||||
function isUnsupportedDirectoryStream(error: unknown): boolean {
|
||||
const code = (error as NodeJS.ErrnoException | null)?.code
|
||||
return code === 'ERR_FEATURE_UNAVAILABLE_ON_PLATFORM' || code === 'ENOTSUP'
|
||||
|
||||
@@ -2,16 +2,34 @@ import { createHash } from 'node:crypto'
|
||||
import type { Stats } from 'node:fs'
|
||||
import { lstat, realpath } from 'node:fs/promises'
|
||||
import { join } from 'node:path'
|
||||
import type { CursorSidecarScanRequest, CursorSidecarScanResponse } from './cursor-sidecar-scan'
|
||||
import {
|
||||
CURSOR_REMOTE_MAX_SESSION_DIRS,
|
||||
type CursorSidecarScanRequest,
|
||||
type CursorSidecarScanResponse
|
||||
} from './cursor-sidecar-scan'
|
||||
import {
|
||||
isCursorSidecarScanCancelledError,
|
||||
type CursorSidecarScanCancellation
|
||||
} from './cursor-sidecar-scan-cancellation'
|
||||
import {
|
||||
CURSOR_DIR_MAX_ENTRIES_EXAMINED,
|
||||
listLexicographicDirectoryNames,
|
||||
resolveTargetScopePathVariants,
|
||||
safeBasename
|
||||
resolveTargetScopePathVariants
|
||||
} from './cursor-sidecar-scan-directory'
|
||||
import {
|
||||
CURSOR_SIDECAR_SCAN_CONCURRENCY,
|
||||
retainCursorSidecarSessions,
|
||||
type CursorSidecarScanBucket as Bucket,
|
||||
type CursorSidecarScanSession
|
||||
} from './cursor-sidecar-scan-session-retention'
|
||||
|
||||
export {
|
||||
cursorSidecarScanCancellationFromSignal,
|
||||
isCursorSidecarScanCancelledError,
|
||||
type CursorSidecarScanCancellation
|
||||
} from './cursor-sidecar-scan-cancellation'
|
||||
|
||||
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
|
||||
|
||||
@@ -23,11 +41,6 @@ export type CursorSidecarScanCaps = {
|
||||
aggregateBytes: number
|
||||
}
|
||||
|
||||
export type CursorSidecarScanCancellation = {
|
||||
throwIfCancelled: () => void
|
||||
}
|
||||
|
||||
type Bucket = { name: string; path: string; scopeCwd: string | null }
|
||||
type ScanArgs = {
|
||||
caps: CursorSidecarScanCaps
|
||||
response: CursorSidecarScanResponse
|
||||
@@ -42,22 +55,6 @@ 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 {
|
||||
return {
|
||||
throwIfCancelled: () => {
|
||||
if (signal?.aborted) {
|
||||
throw new Error('cursor_sidecar_scan_cancelled')
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export async function discoverCursorSidecarCandidates(args: {
|
||||
request: CursorSidecarScanRequest
|
||||
caps: CursorSidecarScanCaps
|
||||
@@ -71,10 +68,12 @@ export async function discoverCursorSidecarCandidates(args: {
|
||||
const pathPlatform = args.pathPlatform ?? process.platform
|
||||
const chatsRoot = args.request.chatsRoot
|
||||
let rootRealPath: string
|
||||
args.cancellation.throwIfCancelled()
|
||||
try {
|
||||
rootRealPath = await realpath(chatsRoot)
|
||||
} catch (error) {
|
||||
rethrowCancel(error)
|
||||
args.cancellation.throwIfCancelled()
|
||||
if (!isMissing(error)) {
|
||||
addIssue(args.response, chatsRoot, error)
|
||||
}
|
||||
@@ -85,7 +84,12 @@ export async function discoverCursorSidecarCandidates(args: {
|
||||
|
||||
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 sessions = await retainCursorSidecarSessions({
|
||||
buckets: [...direct.values(), ...enumerated],
|
||||
sessionLimit: args.caps.sessions,
|
||||
response: args.response,
|
||||
cancellation: args.cancellation
|
||||
})
|
||||
return { rootRealPath, candidates: await eligibleCandidates(sessions, args) }
|
||||
}
|
||||
|
||||
@@ -155,86 +159,16 @@ async function enumeratedBuckets(
|
||||
}
|
||||
}
|
||||
|
||||
async function retainSessions(
|
||||
buckets: readonly Bucket[],
|
||||
args: ScanArgs
|
||||
): Promise<(Bucket & { sessionId: string })[]> {
|
||||
const retained: (Bucket & { sessionId: string })[] = []
|
||||
for (
|
||||
let index = 0;
|
||||
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((bucket) => listBucketSessions(bucket, args, retained.length))
|
||||
)
|
||||
args.cancellation.throwIfCancelled()
|
||||
for (let listingIndex = 0; listingIndex < listings.length; listingIndex += 1) {
|
||||
const { bucket, names, truncated } = listings[listingIndex]
|
||||
if (retained.length >= args.caps.sessions) {
|
||||
if (
|
||||
listings.slice(listingIndex).some((listing) => listing.truncated || listing.names.length)
|
||||
) {
|
||||
args.response.truncated.sessionDirs = true
|
||||
}
|
||||
break
|
||||
}
|
||||
const capacity = args.caps.sessions - retained.length
|
||||
retained.push(...names.slice(0, capacity).map((sessionId) => ({ ...bucket, sessionId })))
|
||||
if (truncated || names.length > capacity) {
|
||||
args.response.truncated.sessionDirs = true
|
||||
}
|
||||
}
|
||||
if (retained.length >= args.caps.sessions && index + batch.length < buckets.length) {
|
||||
args.response.truncated.sessionDirs = true
|
||||
}
|
||||
}
|
||||
return retained
|
||||
}
|
||||
|
||||
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 }
|
||||
}
|
||||
}
|
||||
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 }
|
||||
}
|
||||
}
|
||||
|
||||
async function eligibleCandidates(
|
||||
sessions: readonly (Bucket & { sessionId: string })[],
|
||||
sessions: readonly CursorSidecarScanSession[],
|
||||
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) {
|
||||
for (let index = 0; index < sessions.length; index += CURSOR_SIDECAR_SCAN_CONCURRENCY) {
|
||||
args.cancellation.throwIfCancelled()
|
||||
const batch = sessions.slice(index, index + BUCKET_READ_CONCURRENCY)
|
||||
const batch = sessions.slice(index, index + CURSOR_SIDECAR_SCAN_CONCURRENCY)
|
||||
const results = await Promise.all(batch.map((session) => inspectSession(session, args)))
|
||||
args.cancellation.throwIfCancelled()
|
||||
for (const result of results) {
|
||||
@@ -247,7 +181,7 @@ async function eligibleCandidates(
|
||||
}
|
||||
|
||||
async function inspectSession(
|
||||
session: Bucket & { sessionId: string },
|
||||
session: CursorSidecarScanSession,
|
||||
args: ScanArgs
|
||||
): Promise<CursorSidecarScanCandidate | null> {
|
||||
const sessionDir = join(session.path, session.sessionId)
|
||||
@@ -287,9 +221,25 @@ function retainByNewestThenAggregate(
|
||||
Math.max(left.meta.mtimeMs, left.store.mtimeMs) ||
|
||||
`${left.name}\0${left.sessionId}`.localeCompare(`${right.name}\0${right.sessionId}`)
|
||||
)
|
||||
const representatives = new Set<CursorSidecarScanCandidate>()
|
||||
const representedBuckets = new Set<string>()
|
||||
for (const candidate of ranked) {
|
||||
if (candidate.scopeCwd && !representedBuckets.has(candidate.name)) {
|
||||
representedBuckets.add(candidate.name)
|
||||
representatives.add(candidate)
|
||||
}
|
||||
}
|
||||
const prioritized = [
|
||||
...representatives,
|
||||
...ranked.filter((candidate) => !representatives.has(candidate))
|
||||
]
|
||||
const retained: CursorSidecarScanCandidate[] = []
|
||||
let aggregateBytes = 0
|
||||
for (const candidate of ranked) {
|
||||
for (const candidate of prioritized) {
|
||||
if (retained.length >= CURSOR_REMOTE_MAX_SESSION_DIRS) {
|
||||
args.response.truncated.sessionDirs = true
|
||||
break
|
||||
}
|
||||
if (aggregateBytes + candidate.meta.size > args.caps.aggregateBytes) {
|
||||
args.response.truncated.sidecarBytes = true
|
||||
break
|
||||
|
||||
@@ -0,0 +1,203 @@
|
||||
import { lstat } from 'node:fs/promises'
|
||||
import type { CursorSidecarScanResponse } from './cursor-sidecar-scan'
|
||||
import {
|
||||
isCursorSidecarScanCancelledError,
|
||||
type CursorSidecarScanCancellation
|
||||
} from './cursor-sidecar-scan-cancellation'
|
||||
import {
|
||||
CURSOR_DIR_MAX_ENTRIES_EXAMINED,
|
||||
listLexicographicDirectoryNames,
|
||||
safeBasename
|
||||
} from './cursor-sidecar-scan-directory'
|
||||
|
||||
export const CURSOR_SIDECAR_SCAN_CONCURRENCY = 8
|
||||
const CANCEL_CHECK_EVERY_DIRENTS = 64
|
||||
|
||||
export type CursorSidecarScanBucket = { name: string; path: string; scopeCwd: string | null }
|
||||
export type CursorSidecarScanSession = CursorSidecarScanBucket & { sessionId: string }
|
||||
|
||||
type RetentionArgs = {
|
||||
buckets: readonly CursorSidecarScanBucket[]
|
||||
sessionLimit: number
|
||||
response: CursorSidecarScanResponse
|
||||
cancellation: CursorSidecarScanCancellation
|
||||
}
|
||||
|
||||
export async function retainCursorSidecarSessions(
|
||||
args: RetentionArgs
|
||||
): Promise<CursorSidecarScanSession[]> {
|
||||
const retained: CursorSidecarScanSession[] = []
|
||||
const scopedBuckets = args.buckets.filter((bucket) => bucket.scopeCwd !== null)
|
||||
const unscopedBuckets = args.buckets.filter((bucket) => bucket.scopeCwd === null)
|
||||
const scopedBucketLimit = scopedBuckets.length
|
||||
? Math.max(1, Math.floor(CURSOR_DIR_MAX_ENTRIES_EXAMINED / scopedBuckets.length))
|
||||
: 0
|
||||
const scopedListings = await listBucketSessionsBatched(scopedBuckets, args, scopedBucketLimit)
|
||||
retainScopedSessionListings(scopedListings, retained, args)
|
||||
if (retained.length >= args.sessionLimit) {
|
||||
if (unscopedBuckets.length) {
|
||||
args.response.truncated.sessionDirs = true
|
||||
}
|
||||
return retained
|
||||
}
|
||||
await retainUnscopedSessions(unscopedBuckets, retained, args)
|
||||
return retained
|
||||
}
|
||||
|
||||
async function listBucketSessionsBatched(
|
||||
buckets: readonly CursorSidecarScanBucket[],
|
||||
args: RetentionArgs,
|
||||
limit: number
|
||||
): Promise<BucketSessionListing[]> {
|
||||
const listings: BucketSessionListing[] = []
|
||||
for (let index = 0; index < buckets.length; index += CURSOR_SIDECAR_SCAN_CONCURRENCY) {
|
||||
args.cancellation.throwIfCancelled()
|
||||
const batch = buckets.slice(index, index + CURSOR_SIDECAR_SCAN_CONCURRENCY)
|
||||
listings.push(
|
||||
...(await Promise.all(batch.map((bucket) => listBucketSessions(bucket, args, limit, limit))))
|
||||
)
|
||||
args.cancellation.throwIfCancelled()
|
||||
}
|
||||
return listings
|
||||
}
|
||||
|
||||
function retainScopedSessionListings(
|
||||
listings: readonly BucketSessionListing[],
|
||||
retained: CursorSidecarScanSession[],
|
||||
args: RetentionArgs
|
||||
): void {
|
||||
let depth = 0
|
||||
while (retained.length < args.sessionLimit) {
|
||||
let found = false
|
||||
for (const { bucket, names } of listings) {
|
||||
const sessionId = names[depth]
|
||||
if (sessionId === undefined) {
|
||||
continue
|
||||
}
|
||||
found = true
|
||||
retained.push({ ...bucket, sessionId })
|
||||
if (retained.length >= args.sessionLimit) {
|
||||
break
|
||||
}
|
||||
}
|
||||
if (!found) {
|
||||
break
|
||||
}
|
||||
depth += 1
|
||||
}
|
||||
const listedCount = listings.reduce((total, listing) => total + listing.names.length, 0)
|
||||
if (listings.some((listing) => listing.truncated) || listedCount > retained.length) {
|
||||
args.response.truncated.sessionDirs = true
|
||||
}
|
||||
}
|
||||
|
||||
async function retainUnscopedSessions(
|
||||
buckets: readonly CursorSidecarScanBucket[],
|
||||
retained: CursorSidecarScanSession[],
|
||||
args: RetentionArgs
|
||||
): Promise<void> {
|
||||
for (let index = 0; index < buckets.length; index += CURSOR_SIDECAR_SCAN_CONCURRENCY) {
|
||||
args.cancellation.throwIfCancelled()
|
||||
const batch = buckets.slice(index, index + CURSOR_SIDECAR_SCAN_CONCURRENCY)
|
||||
const listings = await Promise.all(
|
||||
batch.map((bucket) =>
|
||||
listBucketSessions(
|
||||
bucket,
|
||||
args,
|
||||
args.sessionLimit - retained.length,
|
||||
CURSOR_DIR_MAX_ENTRIES_EXAMINED
|
||||
)
|
||||
)
|
||||
)
|
||||
args.cancellation.throwIfCancelled()
|
||||
retainUnscopedSessionListings(listings, retained, args)
|
||||
if (retained.length >= args.sessionLimit && index + batch.length < buckets.length) {
|
||||
args.response.truncated.sessionDirs = true
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function retainUnscopedSessionListings(
|
||||
listings: readonly BucketSessionListing[],
|
||||
retained: CursorSidecarScanSession[],
|
||||
args: RetentionArgs
|
||||
): void {
|
||||
for (let index = 0; index < listings.length; index += 1) {
|
||||
const { bucket, names, truncated } = listings[index]
|
||||
if (retained.length >= args.sessionLimit) {
|
||||
if (listings.slice(index).some((listing) => listing.truncated || listing.names.length)) {
|
||||
args.response.truncated.sessionDirs = true
|
||||
}
|
||||
break
|
||||
}
|
||||
const capacity = args.sessionLimit - retained.length
|
||||
const selected = names.slice(0, capacity)
|
||||
retained.push(...selected.map((sessionId) => ({ ...bucket, sessionId })))
|
||||
if (truncated || names.length > capacity) {
|
||||
args.response.truncated.sessionDirs = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type BucketSessionListing = {
|
||||
bucket: CursorSidecarScanBucket
|
||||
names: string[]
|
||||
truncated: boolean
|
||||
}
|
||||
|
||||
async function listBucketSessions(
|
||||
bucket: CursorSidecarScanBucket,
|
||||
args: RetentionArgs,
|
||||
limit: number,
|
||||
maxEntriesExamined: number
|
||||
): Promise<BucketSessionListing> {
|
||||
try {
|
||||
if (bucket.scopeCwd) {
|
||||
args.response.counters.fileLstat++
|
||||
const stats = await lstat(bucket.path)
|
||||
if (!stats.isDirectory() || stats.isSymbolicLink()) {
|
||||
return { bucket, names: [], truncated: false }
|
||||
}
|
||||
}
|
||||
args.response.counters.bucketReaddir++
|
||||
const listed = await listLexicographicDirectoryNames({
|
||||
dirPath: bucket.path,
|
||||
limit: Math.max(0, limit),
|
||||
maxEntriesExamined,
|
||||
accept: (name, entry) => entry.isDirectory() && !entry.isSymbolicLink() && safeBasename(name),
|
||||
onDirent: createDirentCancelChecker(args.cancellation)
|
||||
})
|
||||
return { bucket, names: listed.names, truncated: listed.truncated }
|
||||
} catch (error) {
|
||||
if (isCursorSidecarScanCancelledError(error)) {
|
||||
throw error
|
||||
}
|
||||
if (!isMissing(error)) {
|
||||
addIssue(args.response, bucket.path, error)
|
||||
}
|
||||
return { bucket, names: [], truncated: false }
|
||||
}
|
||||
}
|
||||
|
||||
function createDirentCancelChecker(cancellation: CursorSidecarScanCancellation): () => void {
|
||||
let seen = 0
|
||||
return () => {
|
||||
seen += 1
|
||||
if (seen % CANCEL_CHECK_EVERY_DIRENTS === 0) {
|
||||
cancellation.throwIfCancelled()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function addIssue(response: CursorSidecarScanResponse, path: string, error: unknown): void {
|
||||
response.issues.push({
|
||||
path,
|
||||
message: error instanceof Error ? error.message.slice(0, 1_024) : 'Cursor scan failed.'
|
||||
})
|
||||
}
|
||||
|
||||
function isMissing(error: unknown): boolean {
|
||||
const code = (error as NodeJS.ErrnoException | null)?.code
|
||||
return code === 'ENOENT' || code === 'ENOTDIR'
|
||||
}
|
||||
Reference in New Issue
Block a user