perf(session-persistence-jsonl): bound listing reads with a shared concurrency limit
list(), the root encoding check, and historical corpus fingerprints read project directories, Session directories, headers, and stats one at a time. On a network filesystem every listing paid the summed round trips. These reads now run through one p-limit limiter per backend (listConcurrency, default 3), shared by concurrent listings; results and the first reported failure keep the sequential order.
This commit is contained in:
1 parent
0af3e0dbc3
commit
6aa157a05a
10 files changed
+266
-49
No files matched your search
@@ -114,6 +114,7 @@ External packages installed for runtime use or distributed inside the prebuilt b
|
||||
| [`node-addon-require-builtin`](https://www.npmjs.com/package/node-addon-require-builtin) | MIT |
|
||||
| [`node-pty`](https://github.com/microsoft/node-pty) | MIT |
|
||||
| [`open`](https://github.com/sindresorhus/open) | MIT |
|
||||
| [`p-limit`](https://github.com/sindresorhus/p-limit) | MIT |
|
||||
| [`papaparse`](https://github.com/mholt/PapaParse) | MIT |
|
||||
| [`pdfjs-dist`](https://github.com/mozilla/pdf.js) | Apache-2.0 |
|
||||
| [`picomatch`](https://github.com/micromatch/picomatch) | MIT |
|
||||
|
||||
@@ -2769,10 +2769,10 @@ export type SessionLogCompressionLevel = 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9
|
||||
|
||||
## `@deepseek-ai/dsh-session-persistence-jsonl`
|
||||
|
||||
- `source`: [`packages/session/session-persistence-jsonl/src/index.ts:90`](../packages/session/session-persistence-jsonl/src/index.ts)
|
||||
- `source`: [`packages/session/session-persistence-jsonl/src/index.ts:94`](../packages/session/session-persistence-jsonl/src/index.ts)
|
||||
|
||||
```ts config-catalog
|
||||
/** Plugin config for the JSONL backend's root and physical encoding. */
|
||||
/** Plugin config for the JSONL backend's root, physical encoding, and listing concurrency. */
|
||||
export interface Config {
|
||||
/**
|
||||
* Root directory for all session files. Required (no default): a default of
|
||||
@@ -2784,6 +2784,12 @@ export interface Config {
|
||||
root: string
|
||||
/** Physical encoding; defaults to checksummed Zstandard frames. */
|
||||
compression?: JsonlCompression
|
||||
/**
|
||||
* Maximum directory, header, and stat reads in flight across all concurrent
|
||||
* listings of this backend (`list()`, the root encoding check, and historical
|
||||
* fingerprints). Defaults to 3.
|
||||
*/
|
||||
listConcurrency?: number
|
||||
}
|
||||
|
||||
/** Physical encoding selected for JSONL session artifacts. */
|
||||
|
||||
@@ -2771,10 +2771,10 @@ export type SessionLogCompressionLevel = 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9
|
||||
|
||||
## `@deepseek-ai/dsh-session-persistence-jsonl`
|
||||
|
||||
- `source`: [`packages/session/session-persistence-jsonl/src/index.ts:90`](../packages/session/session-persistence-jsonl/src/index.ts)
|
||||
- `source`: [`packages/session/session-persistence-jsonl/src/index.ts:94`](../packages/session/session-persistence-jsonl/src/index.ts)
|
||||
|
||||
```ts config-catalog
|
||||
/** Plugin config for the JSONL backend's root and physical encoding. */
|
||||
/** Plugin config for the JSONL backend's root, physical encoding, and listing concurrency. */
|
||||
export interface Config {
|
||||
/**
|
||||
* Root directory for all session files. Required (no default): a default of
|
||||
@@ -2786,6 +2786,12 @@ export interface Config {
|
||||
root: string
|
||||
/** Physical encoding; defaults to checksummed Zstandard frames. */
|
||||
compression?: JsonlCompression
|
||||
/**
|
||||
* Maximum directory, header, and stat reads in flight across all concurrent
|
||||
* listings of this backend (`list()`, the root encoding check, and historical
|
||||
* fingerprints). Defaults to 3.
|
||||
*/
|
||||
listConcurrency?: number
|
||||
}
|
||||
|
||||
/** Physical encoding selected for JSONL session artifacts. */
|
||||
|
||||
@@ -21,8 +21,8 @@
|
||||
en: 37c10302bbbde268
|
||||
zh: 3d5c027bac868efd
|
||||
/deepseek-ai-dsh-session-persistence-jsonl/use-this-package/minimal-configuration:
|
||||
en: 22e8c3b2da45edff
|
||||
zh: d9effd892b514520
|
||||
en: b6c2f0f11f2fa9cf
|
||||
zh: 9fac03fcbe5bbad4
|
||||
/deepseek-ai-dsh-session-persistence-jsonl/use-this-package/on-disk-layout:
|
||||
en: e20cf32540d7c785
|
||||
zh: 67e182529e9211ca
|
||||
|
||||
@@ -46,6 +46,7 @@ Choose this backend when consumers benefit from one artifact per session — nav
|
||||
|---|---|---|
|
||||
| `root` | required | Root directory for all session files |
|
||||
| `compression` | `'zstd'` | Physical encoding: `'zstd'` checksummed frames, or `'none'` newline-delimited UTF-8 text |
|
||||
| `listConcurrency` | `3` | Maximum directory, header, and stat reads in flight, shared by every concurrent listing of this backend |
|
||||
|
||||
Live-event write batching is not configuration: the batching window is the seam's internal scheduling policy inside each write handle.
|
||||
|
||||
|
||||
@@ -46,6 +46,7 @@ kind: "package-reference"
|
||||
|---|---|---|
|
||||
| `root` | 必填 | 所有会话文件的根目录 |
|
||||
| `compression` | `'zstd'` | 物理编码:`'zstd'` 带校验和帧,或 `'none'` 换行分隔 UTF-8 文本 |
|
||||
| `listConcurrency` | `3` | 同时进行的目录、header 与 stat 读取上限,由该后端所有并发的列举共用 |
|
||||
|
||||
实时事件的写入批处理不是配置:批处理窗口是该 seam 在每个写句柄内部的调度策略。
|
||||
|
||||
|
||||
@@ -41,7 +41,8 @@
|
||||
"@deepseek-ai/dsh-session-format-v3-to-v4": "workspace:*",
|
||||
"@deepseek-ai/node-addon-system": "workspace:~",
|
||||
"@deepseek-ai/schemastery": "workspace:~",
|
||||
"koffi": "3.1.1"
|
||||
"koffi": "3.1.1",
|
||||
"p-limit": "7.3.3"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-session": "workspace:*",
|
||||
|
||||
@@ -19,6 +19,7 @@ import { dirname, join, resolve } from 'node:path'
|
||||
import { performance } from 'node:perf_hooks'
|
||||
import { scheduler } from 'node:timers/promises'
|
||||
import { createHash, randomBytes } from 'node:crypto'
|
||||
import pLimit, { type LimitFunction } from 'p-limit'
|
||||
import {
|
||||
SessionPersistence, SessionPersistenceRevision, SessionFormatUnsupportedError,
|
||||
SessionPersistenceCorruptionError,
|
||||
@@ -86,7 +87,10 @@ export const JsonlCompressionSchema: z<JsonlCompression> = z.union([
|
||||
z.const('none'),
|
||||
]).default(DEFAULT_COMPRESSION)
|
||||
|
||||
/** Plugin config for the JSONL backend's root and physical encoding. */
|
||||
/** Default maximum listing filesystem operations in flight per backend. */
|
||||
const DEFAULT_LIST_CONCURRENCY = 3
|
||||
|
||||
/** Plugin config for the JSONL backend's root, physical encoding, and listing concurrency. */
|
||||
export interface Config {
|
||||
/**
|
||||
* Root directory for all session files. Required (no default): a default of
|
||||
@@ -98,6 +102,12 @@ export interface Config {
|
||||
root: string
|
||||
/** Physical encoding; defaults to checksummed Zstandard frames. */
|
||||
compression?: JsonlCompression
|
||||
/**
|
||||
* Maximum directory, header, and stat reads in flight across all concurrent
|
||||
* listings of this backend (`list()`, the root encoding check, and historical
|
||||
* fingerprints). Defaults to 3.
|
||||
*/
|
||||
listConcurrency?: number
|
||||
}
|
||||
|
||||
/** Logical header paired with the catalog's selected physical-generation classification. */
|
||||
@@ -202,6 +212,36 @@ function isENOENT(error: unknown): boolean {
|
||||
return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT'
|
||||
}
|
||||
|
||||
/**
|
||||
* Run `task` over `items` through the backend's shared listing limiter and
|
||||
* settle every call. Callers consume the outcomes in input order, so the first
|
||||
* failure they report matches a sequential loop whatever the completion order.
|
||||
* A limited task must never wait on the same limiter.
|
||||
* @param limit - the backend's shared listing limiter.
|
||||
* @param items - inputs in their sequential order.
|
||||
* @param task - per-item operation.
|
||||
* @returns one outcome per item, in input order.
|
||||
*/
|
||||
function settleLimited<T, R>(
|
||||
limit: LimitFunction,
|
||||
items: readonly T[],
|
||||
task: (item: T) => Promise<R>,
|
||||
): Promise<PromiseSettledResult<R>[]> {
|
||||
return Promise.allSettled(items.map(item => limit(() => task(item))))
|
||||
}
|
||||
|
||||
/**
|
||||
* Unwrap outcomes from {@link settleLimited}, rethrowing the first rejection in input order.
|
||||
* @param outcomes - outcomes in input order.
|
||||
* @returns the fulfilled values in input order.
|
||||
*/
|
||||
function settledValues<R>(outcomes: readonly PromiseSettledResult<R>[]): R[] {
|
||||
return outcomes.map((outcome) => {
|
||||
if (outcome.status === 'rejected') throw outcome.reason
|
||||
return outcome.value
|
||||
})
|
||||
}
|
||||
|
||||
/** Whether a filesystem-owned failure should retain its original errno and path. */
|
||||
function isErrnoException(error: unknown): error is NodeJS.ErrnoException {
|
||||
return typeof (error as NodeJS.ErrnoException | null)?.code === 'string'
|
||||
@@ -254,6 +294,12 @@ class JsonlSessionPersistence extends SessionPersistence {
|
||||
|
||||
private root: string
|
||||
private compression: JsonlCompression
|
||||
/**
|
||||
* Bounds directory, header, and stat I/O across every concurrent listing of
|
||||
* this backend (`list()`, the root encoding check, historical fingerprints).
|
||||
* Queued tasks of a cancelled listing run and reject at their signal check.
|
||||
*/
|
||||
private readonly listLimit: LimitFunction
|
||||
private rootEncodingCheck: Promise<void> | undefined
|
||||
private readonly generationFormat: Omit<JsonlGenerationFormatAdapter, 'createRestore'>
|
||||
/**
|
||||
@@ -279,6 +325,7 @@ class JsonlSessionPersistence extends SessionPersistence {
|
||||
// Resolve once so later process.cwd() changes cannot split one backend across roots.
|
||||
this.root = resolve(config.root)
|
||||
this.compression = config.compression ?? DEFAULT_COMPRESSION
|
||||
this.listLimit = pLimit(config.listConcurrency ?? DEFAULT_LIST_CONCURRENCY)
|
||||
this.generationFormat = {
|
||||
currentVersion: sessionFormatCatalog.currentVersion,
|
||||
encodeHeader: (header, inheritedEventCount) =>
|
||||
@@ -482,24 +529,30 @@ class JsonlSessionPersistence extends SessionPersistence {
|
||||
const artifacts = await this.listArtifacts(signal)
|
||||
const corpusRevision = artifacts.some(artifact => artifact.sourceVersion < SESSION_FORMAT_VERSION)
|
||||
? await this.historicalCorpusRevision(signal) : undefined
|
||||
for (const artifact of artifacts) {
|
||||
const statted = settledValues(await settleLimited(this.listLimit, artifacts, async (artifact) => {
|
||||
signal?.throwIfAborted()
|
||||
try {
|
||||
const identity = await stat(artifact.path, { bigint: true })
|
||||
signal?.throwIfAborted()
|
||||
listed.add(artifact.header.id)
|
||||
snapshots.push({
|
||||
header: artifact.header,
|
||||
formatStatus: artifact.formatStatus,
|
||||
revision: artifact.sourceVersion < SESSION_FORMAT_VERSION
|
||||
? SessionPersistenceRevision(`${fileRevision(identity)}:${corpusRevision}`)
|
||||
: fileRevision(identity),
|
||||
sizeBytes: Number(identity.size),
|
||||
})
|
||||
return { artifact, identity }
|
||||
} catch (error: unknown) {
|
||||
signal?.throwIfAborted()
|
||||
if (!isENOENT(error)) throw error
|
||||
return undefined
|
||||
}
|
||||
}))
|
||||
for (const entry of statted) {
|
||||
if (entry === undefined) continue
|
||||
const { artifact, identity } = entry
|
||||
listed.add(artifact.header.id)
|
||||
snapshots.push({
|
||||
header: artifact.header,
|
||||
formatStatus: artifact.formatStatus,
|
||||
revision: artifact.sourceVersion < SESSION_FORMAT_VERSION
|
||||
? SessionPersistenceRevision(`${fileRevision(identity)}:${corpusRevision}`)
|
||||
: fileRevision(identity),
|
||||
sizeBytes: Number(identity.size),
|
||||
})
|
||||
}
|
||||
for (const [id, entry] of pending) {
|
||||
if (!listed.has(id)) snapshots.push({ header: entry.header, revision: entry.revision })
|
||||
@@ -1023,32 +1076,59 @@ class JsonlSessionPersistence extends SessionPersistence {
|
||||
|
||||
/** Enumerate selected physical generations without interpreting their headers or bodies. */
|
||||
private async listGenerations(signal?: AbortSignal): Promise<ResolvedJsonlGeneration[]> {
|
||||
const sources: ResolvedJsonlGeneration[] = []
|
||||
for (const project of await this.listProjectDirs(signal)) {
|
||||
for (const dir of await this.listSessionDirs(project, signal)) {
|
||||
signal?.throwIfAborted()
|
||||
const selected = await this.resolveGenerationInDirectory(dir, signal)
|
||||
if (selected !== undefined) sources.push(selected)
|
||||
const selected = await this.mapSessionDirectories((dir) => {
|
||||
signal?.throwIfAborted()
|
||||
return this.resolveGenerationInDirectory(dir, signal)
|
||||
}, signal)
|
||||
return selected.filter(source => source !== undefined)
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply `task` to every Session directory through the shared listing
|
||||
* limiter. Results and the first failure follow the sequential
|
||||
* project-then-directory order; a project whose listing fails ends the
|
||||
* traversal after the directories of the projects before it.
|
||||
* @param task - per-directory operation; must not wait on the listing limiter.
|
||||
* @param signal - optional cancellation for directory listings.
|
||||
* @returns per-directory results in traversal order.
|
||||
*/
|
||||
private async mapSessionDirectories<R>(
|
||||
task: (dir: string) => Promise<R>,
|
||||
signal?: AbortSignal,
|
||||
): Promise<R[]> {
|
||||
const limit = this.listLimit
|
||||
// The root read counts against the same bound as every other listing read.
|
||||
const projectDirs = await limit(() => this.listProjectDirs(signal))
|
||||
const projects = await settleLimited(limit, projectDirs, project => this.listSessionDirs(project, signal))
|
||||
const dirs: string[] = []
|
||||
let projectFailure: PromiseRejectedResult | undefined
|
||||
for (const outcome of projects) {
|
||||
if (outcome.status === 'rejected') {
|
||||
projectFailure = outcome
|
||||
break
|
||||
}
|
||||
// Appended one by one: spreading a large project as call arguments exceeds the engine's argument limit.
|
||||
for (const dir of outcome.value) dirs.push(dir)
|
||||
}
|
||||
return sources
|
||||
const results = settledValues(await settleLimited(limit, dirs, task))
|
||||
if (projectFailure !== undefined) throw projectFailure.reason
|
||||
return results
|
||||
}
|
||||
|
||||
/** Historical logical events depend on the corpus, including members with unreadable headers. */
|
||||
private async historicalCorpusRevision(signal?: AbortSignal): Promise<string> {
|
||||
const paths = (await this.listGenerations(signal)).map(source => source.sourcePath).sort()
|
||||
const hash = createHash('sha256')
|
||||
for (const path of paths) {
|
||||
const entries = settledValues(await settleLimited(this.listLimit, paths, async (path) => {
|
||||
signal?.throwIfAborted()
|
||||
let revision: string
|
||||
try {
|
||||
revision = fileRevision(await stat(path, { bigint: true }))
|
||||
return { path, revision: fileRevision(await stat(path, { bigint: true })) }
|
||||
} catch (error: unknown) {
|
||||
if (!isENOENT(error)) throw error
|
||||
revision = 'missing'
|
||||
return { path, revision: 'missing' }
|
||||
}
|
||||
hash.update(JSON.stringify([path, revision]))
|
||||
}
|
||||
}))
|
||||
const hash = createHash('sha256')
|
||||
for (const { path, revision } of entries) hash.update(JSON.stringify([path, revision]))
|
||||
signal?.throwIfAborted()
|
||||
return hash.digest('hex')
|
||||
}
|
||||
@@ -1059,20 +1139,24 @@ class JsonlSessionPersistence extends SessionPersistence {
|
||||
signal?.throwIfAborted()
|
||||
await this.ensureRootEncoding()
|
||||
signal?.throwIfAborted()
|
||||
const artifacts: Array<GenerationHeader & { path: string; sourceVersion: number }> = []
|
||||
const ids = new Set<SessionId>()
|
||||
for (const selected of await this.listGenerations(signal)) {
|
||||
const generations = await this.listGenerations(signal)
|
||||
const headers = await settleLimited(this.listLimit, generations, async (selected) => {
|
||||
signal?.throwIfAborted()
|
||||
let metadata: GenerationHeader | undefined
|
||||
try {
|
||||
metadata = await this.readGenerationHeader(selected, undefined, signal)
|
||||
return { selected, metadata: await this.readGenerationHeader(selected, undefined, signal) }
|
||||
} catch (error: unknown) {
|
||||
if (error instanceof SessionFormatUnsupportedError || error instanceof SessionPersistenceCorruptionError) continue
|
||||
if (error instanceof SessionFormatUnsupportedError || error instanceof SessionPersistenceCorruptionError) {
|
||||
return { selected, metadata: undefined }
|
||||
}
|
||||
throw error
|
||||
}
|
||||
if (metadata === undefined) {
|
||||
continue
|
||||
}
|
||||
})
|
||||
const artifacts: Array<GenerationHeader & { path: string; sourceVersion: number }> = []
|
||||
const ids = new Set<SessionId>()
|
||||
for (const outcome of headers) {
|
||||
if (outcome.status === 'rejected') throw outcome.reason
|
||||
const { selected, metadata } = outcome.value
|
||||
if (metadata === undefined) continue
|
||||
if (ids.has(metadata.header.id)) {
|
||||
throw new Error(`duplicate JSONL session id "${metadata.header.id}" appears in multiple project directories`)
|
||||
}
|
||||
@@ -1613,12 +1697,10 @@ class JsonlSessionPersistence extends SessionPersistence {
|
||||
}
|
||||
|
||||
private async checkRootEncoding(): Promise<void> {
|
||||
for (const project of await this.listProjectDirs()) {
|
||||
for (const dir of await this.listSessionDirs(project)) {
|
||||
const incompatible = await this.findOppositeGenerationInDirectory(dir)
|
||||
if (incompatible !== undefined) throw this.encodingMismatch(incompatible)
|
||||
}
|
||||
}
|
||||
await this.mapSessionDirectories(async (dir) => {
|
||||
const incompatible = await this.findOppositeGenerationInDirectory(dir)
|
||||
if (incompatible !== undefined) throw this.encodingMismatch(incompatible)
|
||||
})
|
||||
}
|
||||
|
||||
private async rejectLegacyFlatArtifact(
|
||||
@@ -1722,10 +1804,11 @@ class JsonlSessionPersistence extends SessionPersistence {
|
||||
/** JSONL persistence plugin label. */
|
||||
export const name = 'session-persistence-jsonl'
|
||||
|
||||
/** Loader schema for the backend's root and physical encoding. */
|
||||
/** Loader schema for the backend's root, physical encoding, and listing concurrency. */
|
||||
export const Config: z<Config> = z.object({
|
||||
root: z.string().required(),
|
||||
compression: JsonlCompressionSchema,
|
||||
listConcurrency: z.natural().min(1).default(DEFAULT_LIST_CONCURRENCY),
|
||||
})
|
||||
|
||||
/**
|
||||
@@ -1733,7 +1816,7 @@ export const Config: z<Config> = z.object({
|
||||
* consumer unloads triggered by removing this provider. Consumers must own
|
||||
* cleanup in their injecting scope; concurrent root disposal is not ordered.
|
||||
* @param ctx - the owning plugin context.
|
||||
* @param config - the storage root and physical encoding.
|
||||
* @param config - the storage root, physical encoding, and listing concurrency.
|
||||
* @returns resolves when the service is ready.
|
||||
*/
|
||||
export async function apply(ctx: Context, config: Config): Promise<void> {
|
||||
|
||||
@@ -39,6 +39,18 @@ const readdirFailure = vi.hoisted(() => ({
|
||||
error: undefined as Error | undefined,
|
||||
}))
|
||||
|
||||
/**
|
||||
* While active, every `readdir` and `stat` waits `delayMs`, records how many
|
||||
* such calls overlap, and fails with the error registered for its path.
|
||||
*/
|
||||
const listingProbe = vi.hoisted(() => ({
|
||||
active: false,
|
||||
delayMs: 0,
|
||||
inFlight: 0,
|
||||
maxInFlight: 0,
|
||||
failures: new Map<string, Error>(),
|
||||
}))
|
||||
|
||||
const readTally = vi.hoisted(() => ({
|
||||
/** Physical whole-file reads per path suffix; keyed by session id segment. */
|
||||
bySuffix: new Map<string, number>(),
|
||||
@@ -62,9 +74,22 @@ const pausedRead = vi.hoisted(() => ({
|
||||
|
||||
vi.mock('node:fs/promises', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import('node:fs/promises')>()
|
||||
const probeListingRead = async <T>(path: string, read: () => Promise<T>): Promise<T> => {
|
||||
listingProbe.inFlight += 1
|
||||
listingProbe.maxInFlight = Math.max(listingProbe.maxInFlight, listingProbe.inFlight)
|
||||
try {
|
||||
await new Promise(resolvePromise => setTimeout(resolvePromise, listingProbe.delayMs))
|
||||
const failure = listingProbe.failures.get(path)
|
||||
if (failure !== undefined) throw failure
|
||||
return await read()
|
||||
} finally {
|
||||
listingProbe.inFlight -= 1
|
||||
}
|
||||
}
|
||||
return {
|
||||
...actual,
|
||||
stat: (async (...args: Parameters<typeof actual.stat>) => {
|
||||
if (listingProbe.active) return probeListingRead(String(args[0]), () => actual.stat(...args))
|
||||
if (String(args[0]) === statFailure.path && statFailure.error !== undefined) throw statFailure.error
|
||||
const identity = await actual.stat(...args)
|
||||
if (String(args[0]) !== statRace.path || !('mtimeNs' in identity)) return identity
|
||||
@@ -95,6 +120,7 @@ vi.mock('node:fs/promises', async (importOriginal) => {
|
||||
}
|
||||
}) as typeof actual.readFile,
|
||||
readdir: (async (...args: Parameters<typeof actual.readdir>) => {
|
||||
if (listingProbe.active) return probeListingRead(String(args[0]), () => actual.readdir(...args))
|
||||
if (String(args[0]) === readdirFailure.path && readdirFailure.error !== undefined) {
|
||||
throw readdirFailure.error
|
||||
}
|
||||
@@ -2789,3 +2815,78 @@ describe('JsonlSessionPersistence: edge cases', () => {
|
||||
expect((await readAll(ctx.sessionPersistence, m.id)).events).toEqual(events)
|
||||
})
|
||||
})
|
||||
|
||||
describe('JsonlSessionPersistence listing concurrency', () => {
|
||||
let listingRoot: string
|
||||
let listingCtx: Context
|
||||
|
||||
beforeEach(async () => {
|
||||
listingRoot = await mkdtemp(join(tmpdir(), 'dsh-jsonl-listing-'))
|
||||
listingCtx = new Context()
|
||||
await listingCtx.plugin(JsonlSessionPersistence, { root: listingRoot, compression: 'none', listConcurrency: 2 })
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
listingProbe.active = false
|
||||
listingProbe.delayMs = 0
|
||||
listingProbe.inFlight = 0
|
||||
listingProbe.maxInFlight = 0
|
||||
listingProbe.failures.clear()
|
||||
await listingCtx.fiber.dispose()
|
||||
await rm(listingRoot, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
/** Persist one-turn Sessions under one cwd. */
|
||||
async function seed(cwd: string, ids: readonly string[]): Promise<void> {
|
||||
for (const id of ids) await writeLog(listingCtx.sessionPersistence, meta(id, cwd), oneTurnLog())
|
||||
}
|
||||
|
||||
/** Directory entries in the order the backend's listing reads them. */
|
||||
async function directoriesOf(dir: string): Promise<string[]> {
|
||||
return (await readdir(dir, { withFileTypes: true })).filter(entry => entry.isDirectory()).map(entry => join(dir, entry.name))
|
||||
}
|
||||
|
||||
it('validates listConcurrency as a positive integer that defaults to 3', () => {
|
||||
expect(JsonlSessionPersistence.Config({ root: listingRoot }).listConcurrency).toBe(3)
|
||||
expect(JsonlSessionPersistence.Config({ root: listingRoot, listConcurrency: 1 }).listConcurrency).toBe(1)
|
||||
expect(() => JsonlSessionPersistence.Config({ root: listingRoot, listConcurrency: 0 })).toThrow()
|
||||
expect(() => JsonlSessionPersistence.Config({ root: listingRoot, listConcurrency: 1.5 })).toThrow()
|
||||
})
|
||||
|
||||
it('bounds directory and stat reads across concurrent listings of one backend', async () => {
|
||||
await seed('/work-a', ['bound-a1', 'bound-a2', 'bound-a3'])
|
||||
await seed('/work-b', ['bound-b1', 'bound-b2', 'bound-b3'])
|
||||
const expected = ['bound-a1', 'bound-a2', 'bound-a3', 'bound-b1', 'bound-b2', 'bound-b3']
|
||||
expect((await listingCtx.sessionPersistence.list()).map(s => s.header.id).sort()).toEqual(expected)
|
||||
|
||||
listingProbe.delayMs = 5
|
||||
listingProbe.active = true
|
||||
const listings = await Promise.all([
|
||||
listingCtx.sessionPersistence.list(),
|
||||
listingCtx.sessionPersistence.list(),
|
||||
])
|
||||
listingProbe.active = false
|
||||
|
||||
expect(listingProbe.maxInFlight).toBe(2)
|
||||
for (const listing of listings) expect(listing.map(s => s.header.id).sort()).toEqual(expected)
|
||||
})
|
||||
|
||||
it('reports the failure a sequential traversal would reach first', async () => {
|
||||
await seed('/work-a', ['order-a1', 'order-a2'])
|
||||
await seed('/work-b', ['order-b1', 'order-b2'])
|
||||
await listingCtx.sessionPersistence.list()
|
||||
const [earlierProject, laterProject] = await directoriesOf(listingRoot) as [string, string]
|
||||
const [earlierSession, laterSession] = await directoriesOf(earlierProject) as [string, string]
|
||||
const projectFailure = Object.assign(new Error('later project denied'), { code: 'EACCES' })
|
||||
const laterSessionFailure = Object.assign(new Error('later session denied'), { code: 'EACCES' })
|
||||
const earlierSessionFailure = Object.assign(new Error('earlier session denied'), { code: 'EACCES' })
|
||||
listingProbe.active = true
|
||||
|
||||
listingProbe.failures.set(laterProject, projectFailure)
|
||||
await expect(listingCtx.sessionPersistence.list()).rejects.toBe(projectFailure)
|
||||
listingProbe.failures.set(laterSession, laterSessionFailure)
|
||||
await expect(listingCtx.sessionPersistence.list()).rejects.toBe(laterSessionFailure)
|
||||
listingProbe.failures.set(earlierSession, earlierSessionFailure)
|
||||
await expect(listingCtx.sessionPersistence.list()).rejects.toBe(earlierSessionFailure)
|
||||
})
|
||||
})
|
||||
Generated
+17
@@ -11984,6 +11984,9 @@ importers:
|
||||
koffi:
|
||||
specifier: 3.1.1
|
||||
version: 3.1.1
|
||||
p-limit:
|
||||
specifier: 7.3.3
|
||||
version: 7.3.3
|
||||
devDependencies:
|
||||
'@deepseek-ai/cordis':
|
||||
specifier: workspace:~
|
||||
@@ -21624,6 +21627,10 @@ packages:
|
||||
resolution: {integrity: sha512-TYOanM3wGwNGsZN2cVTYPArw454xnXj5qmWF1bEoAc4+cU/ol7GVh7odevjp1FNHduHc3KZMcFduxU5Xc6uJRQ==}
|
||||
engines: {node: '>=10'}
|
||||
|
||||
[email protected]:
|
||||
resolution: {integrity: sha512-SKgVvDU5TNIxQ4r30U75BNLC6a+6GOEjngrk/ysM5+Zu3oJuk1SC5ApzIzY/7PjCEjoVPMWJ0zho/4QUiEB7TQ==}
|
||||
engines: {node: '>=20'}
|
||||
|
||||
[email protected]:
|
||||
resolution: {integrity: sha512-LaNjtRWUBY++zB5nE/NwcaoMylSPk+S+ZHNB1TzdbMJMny6dynpAGt7X/tl/QYq3TIeE6nxHppbo2LGymrG5Pw==}
|
||||
engines: {node: '>=10'}
|
||||
@@ -23137,6 +23144,10 @@ packages:
|
||||
resolution: {integrity: sha512-rVksvsnNCdJ/ohGc6xgPwyN8eheCxsiLM8mxuE/t/mOVqJewPuO1miLpTHQiRgTKCLexL4MeAFVagts7HmNZ2Q==}
|
||||
engines: {node: '>=10'}
|
||||
|
||||
[email protected]:
|
||||
resolution: {integrity: sha512-4LCcse/U2MHZ63HAJVE+v71o7yOdIe4cZ70Wpf8D/IyjDKYQLV5GD46B+hSTjJsvV5PztjvHoU580EftxjDZFQ==}
|
||||
engines: {node: '>=12.20'}
|
||||
|
||||
[email protected]:
|
||||
resolution: {integrity: sha512-CzhO+pFNo8ajLM2d2IW/R93ipy99LWjtwblvC1RsoSUMZgyLbYFr221TnSNT7GjGdYui6P459mw9JH/g/zW2ug==}
|
||||
engines: {node: '>=18'}
|
||||
@@ -29369,6 +29380,10 @@ snapshots:
|
||||
dependencies:
|
||||
yocto-queue: 0.1.0
|
||||
|
||||
[email protected]:
|
||||
dependencies:
|
||||
yocto-queue: 1.2.2
|
||||
|
||||
[email protected]:
|
||||
dependencies:
|
||||
p-limit: 3.1.0
|
||||
@@ -30922,6 +30937,8 @@ snapshots:
|
||||
|
||||
[email protected]: {}
|
||||
|
||||
[email protected]: {}
|
||||
|
||||
[email protected]: {}
|
||||
|
||||
[email protected]:
|
||||
|
||||
Reference in new issue
Block a user