perf(jsonl): scan decoded frames incrementally
This commit is contained in:
imccyu
@@ -217,106 +217,157 @@ export function eventLines(events: readonly SessionEvent[], packChunks: boolean)
|
||||
return records.map(record => JSON.stringify(record)).join('\n')
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a JSONL log buffer into its preserved event prefix (the header is line
|
||||
* 0). Event lines pass through verbatim; packed chunk rows expand back into
|
||||
* their events, so callers see one contiguous event list regardless of layout.
|
||||
* Fully written events in an interrupted final turn remain part of the
|
||||
* prefix. The first unparsable record or seq gap after the last `turn/end`
|
||||
* marks a tolerated torn tail; the same hole in the committed region rejects.
|
||||
*
|
||||
* @param buffer - the raw bytes of the log file (header line first).
|
||||
* @returns the header, the preserved event prefix, and `committedBytes` — the
|
||||
* byte offset the next append truncates any torn tail to.
|
||||
*/
|
||||
export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionEvent[]; committedBytes: number } {
|
||||
const text = buffer.toString('utf8')
|
||||
// Track complete lines by byte offset: a non-newline tail is torn and ignored,
|
||||
// and a running counter avoids rescanning a long multi-byte log.
|
||||
const lines: { text: string; endByte: number }[] = []
|
||||
let start = 0
|
||||
let byteOffset = 0
|
||||
for (let i = 0; i < text.length; i++) {
|
||||
if (text[i] === '\n') {
|
||||
const lineText = text.slice(start, i)
|
||||
byteOffset += Buffer.byteLength(lineText, 'utf8') + 1 // +1 for the '\n' (a 1-byte char)
|
||||
lines.push({ text: lineText, endByte: byteOffset })
|
||||
start = i + 1
|
||||
}
|
||||
interface SessionLogScan {
|
||||
meta: SessionHeader
|
||||
events: SessionEvent[]
|
||||
committedBytes: number
|
||||
}
|
||||
|
||||
/** Parse one complete header record supplied independently from event rows. */
|
||||
function parseHeaderRecord(record: Buffer): SessionHeader {
|
||||
if (record.length === 0 || record.at(-1) !== 0x0A || record.indexOf(0x0A) !== record.length - 1) {
|
||||
throw new Error('empty or header-less session log')
|
||||
}
|
||||
|
||||
const [headerEntry, ...eventEntries] = lines
|
||||
if (headerEntry === undefined) throw new Error('empty or header-less session log')
|
||||
|
||||
// Line 0 is the header.
|
||||
let parsedHeader: unknown
|
||||
let parsed: unknown
|
||||
try {
|
||||
parsedHeader = JSON.parse(headerEntry.text)
|
||||
parsed = JSON.parse(record.subarray(0, -1).toString('utf8'))
|
||||
} catch {
|
||||
throw new Error('corrupt session log: header line is not valid JSON')
|
||||
}
|
||||
if (!isHeaderLine(parsedHeader)) {
|
||||
if (!isHeaderLine(parsed)) {
|
||||
throw new Error('corrupt session log: first line is not a session header')
|
||||
}
|
||||
const headerLine = parsedHeader
|
||||
return fromHeaderLine(parsed)
|
||||
}
|
||||
|
||||
// Parse and decode every complete line first so the last valid `turn/end`
|
||||
// determines whether an earlier hole interrupts an otherwise closed
|
||||
// execution or belongs to a tolerable final suffix. One line yields one
|
||||
// event, or a whole run for a packed chunk row; a row-tagged line that fails
|
||||
// row validation is a hole, exactly like unparsable JSON.
|
||||
interface Parsed { ok: boolean; events?: SessionEvent[]; endByte: number }
|
||||
const parsed: Parsed[] = eventEntries.map((entry) => {
|
||||
try {
|
||||
return { ok: true, events: decodeStorageRecord(JSON.parse(entry.text)), endByte: entry.endByte }
|
||||
} catch {
|
||||
return { ok: false, endByte: entry.endByte }
|
||||
}
|
||||
})
|
||||
/**
|
||||
* Incrementally scan complete JSONL event records after an independently
|
||||
* supplied header record. Newline search and byte offsets stay on raw buffers;
|
||||
* only complete records are decoded to UTF-8. A fragment crossing writes is
|
||||
* copied because a decoder may reuse its output buffer after `write()` returns.
|
||||
*/
|
||||
export class SessionLogScanner {
|
||||
private readonly meta: SessionHeader
|
||||
private readonly events: SessionEvent[] = []
|
||||
private fragments: Buffer[] = []
|
||||
private fragmentBytes = 0
|
||||
private inputBytes: number
|
||||
private committedBytes: number
|
||||
private eventLine = 0
|
||||
private issue: Error | undefined
|
||||
private finished = false
|
||||
|
||||
// The last index (into eventEntries) that ends in a valid `turn/end`. A hole
|
||||
// before this boundary cannot be a torn final suffix because later execution
|
||||
// already closed. Standalone events after it remain part of the preserved
|
||||
// contiguous prefix. A packed row never stores a turn/end, so only
|
||||
// single-event lines can match.
|
||||
let lastTurnEnd = -1
|
||||
for (let i = parsed.length - 1; i >= 0; i--) {
|
||||
const p = parsed[i]
|
||||
if (p?.ok && p.events?.some(e => e.type === 'turn/end')) { lastTurnEnd = i; break }
|
||||
/**
|
||||
* Create an event scanner from exactly one newline-terminated header record.
|
||||
* @param headerRecord - the complete first JSONL record, including its newline.
|
||||
*/
|
||||
constructor(headerRecord: Buffer) {
|
||||
this.meta = parseHeaderRecord(headerRecord)
|
||||
this.inputBytes = headerRecord.length
|
||||
this.committedBytes = headerRecord.length
|
||||
}
|
||||
|
||||
// Preserve the contiguous prefix, including a complete interrupted turn;
|
||||
// holes through the last committed boundary throw, while later holes stop.
|
||||
// Contiguity is a cursor over seqs (not the line index): a packed row
|
||||
// advances the cursor by its whole run.
|
||||
const preserved: SessionEvent[] = []
|
||||
let lastPreservedLine = -1
|
||||
scan: for (let i = 0; i < parsed.length; i++) {
|
||||
const p = parsed[i]
|
||||
if (!p?.ok || p.events === undefined) {
|
||||
if (i <= lastTurnEnd) throw new Error(`corrupt session log: unparsable committed event at line ${i + 1}`)
|
||||
break // torn tail fragment after the last turn/end — stop, tolerate
|
||||
}
|
||||
for (const event of p.events) {
|
||||
if (event.seq !== preserved.length) {
|
||||
if (i <= lastTurnEnd) {
|
||||
throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${preserved.length}, got ${event.seq})`)
|
||||
}
|
||||
break scan // gap after the last turn/end — torn tail, stop
|
||||
/**
|
||||
* Consume the next raw plaintext chunk, retaining only an incomplete final record.
|
||||
* @param chunk - bytes immediately following all previously supplied bytes.
|
||||
*/
|
||||
write(chunk: Buffer): void {
|
||||
if (this.finished) throw new Error('cannot write to a finished session log scanner')
|
||||
const chunkStart = this.inputBytes
|
||||
this.inputBytes += chunk.length
|
||||
let lineStart = 0
|
||||
for (
|
||||
let newline = chunk.indexOf(0x0A);
|
||||
newline !== -1;
|
||||
newline = chunk.indexOf(0x0A, lineStart)
|
||||
) {
|
||||
const fragment = chunk.subarray(lineStart, newline)
|
||||
let line = fragment
|
||||
if (this.fragments.length > 0) {
|
||||
if (fragment.length > 0) this.fragments.push(fragment)
|
||||
line = Buffer.concat(this.fragments, this.fragmentBytes + fragment.length)
|
||||
this.fragments = []
|
||||
this.fragmentBytes = 0
|
||||
}
|
||||
preserved.push(event)
|
||||
this.consumeEventLine(line, chunkStart + newline + 1)
|
||||
lineStart = newline + 1
|
||||
}
|
||||
if (lineStart < chunk.length) {
|
||||
const fragment = Buffer.from(chunk.subarray(lineStart))
|
||||
this.fragments.push(fragment)
|
||||
this.fragmentBytes += fragment.length
|
||||
}
|
||||
lastPreservedLine = i
|
||||
}
|
||||
|
||||
// committedBytes = end of the last FULLY preserved line (header if none): the
|
||||
// next append truncates any torn bytes past this point before writing the
|
||||
// synthetic closers + new events. A line is preserved whole or not at all —
|
||||
// a mid-row seq gap discards the whole row, keeping the truncation offset on
|
||||
// a line boundary.
|
||||
const lastPreserved = parsed[lastPreservedLine]
|
||||
const committedBytes = lastPreserved !== undefined ? lastPreserved.endByte : headerEntry.endByte
|
||||
return { meta: fromHeaderLine(headerLine), events: preserved, committedBytes }
|
||||
/**
|
||||
* Snapshot progress before appending a recoverable torn-frame prefix.
|
||||
* @returns byte, committed-prefix, and expanded-event cursors.
|
||||
*/
|
||||
checkpoint(): { inputBytes: number; committedBytes: number; eventCount: number } {
|
||||
return {
|
||||
inputBytes: this.inputBytes,
|
||||
committedBytes: this.committedBytes,
|
||||
eventCount: this.events.length,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Finish scanning, ignoring a final record without a newline as a torn tail.
|
||||
* @returns the header, contiguous event prefix, and safe truncation offset.
|
||||
*/
|
||||
finish(): SessionLogScan {
|
||||
this.finished = true
|
||||
return { meta: this.meta, events: this.events, committedBytes: this.committedBytes }
|
||||
}
|
||||
|
||||
/** Decode one complete event row and update the contiguous prefix. */
|
||||
private consumeEventLine(line: Buffer, endByte: number): void {
|
||||
this.eventLine += 1
|
||||
let decoded: SessionEvent[]
|
||||
try {
|
||||
decoded = decodeStorageRecord(JSON.parse(line.toString('utf8')))
|
||||
} catch {
|
||||
this.issue ??= new Error(`corrupt session log: unparsable committed event at line ${this.eventLine}`)
|
||||
return
|
||||
}
|
||||
|
||||
if (this.issue !== undefined) {
|
||||
if (decoded.some(event => event.type === 'turn/end')) throw this.issue
|
||||
return
|
||||
}
|
||||
|
||||
const rowStart = this.events.length
|
||||
for (const event of decoded) {
|
||||
if (event.seq !== this.events.length) {
|
||||
const expected = this.events.length
|
||||
this.events.length = rowStart
|
||||
this.issue = new Error(
|
||||
`corrupt session log: seq gap in committed region at line ${this.eventLine} `
|
||||
+ `(expected ${expected}, got ${event.seq})`,
|
||||
)
|
||||
if (decoded.some(candidate => candidate.type === 'turn/end')) throw this.issue
|
||||
return
|
||||
}
|
||||
this.events.push(event)
|
||||
}
|
||||
this.committedBytes = endByte
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Parse a complete or torn JSONL buffer into its preserved event prefix. This
|
||||
* compatibility wrapper supplies the first record separately, then delegates
|
||||
* event rows to {@link SessionLogScanner}.
|
||||
*
|
||||
* @param buffer - the raw bytes of the log file (header line first).
|
||||
* @returns the header, preserved event prefix, and byte offset safe to append at.
|
||||
*/
|
||||
export function scanLog(buffer: Buffer): SessionLogScan {
|
||||
const headerEnd = buffer.indexOf(0x0A)
|
||||
if (headerEnd === -1) throw new Error('empty or header-less session log')
|
||||
const scanner = new SessionLogScanner(buffer.subarray(0, headerEnd + 1))
|
||||
scanner.write(buffer.subarray(headerEnd + 1))
|
||||
return scanner.finish()
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -11,6 +11,8 @@ import z from 'schemastery'
|
||||
import { readdirSync } from 'node:fs'
|
||||
import { open, mkdir, readFile, readdir, realpath, link, rm, stat, truncate } from 'node:fs/promises'
|
||||
import { dirname, join, resolve } from 'node:path'
|
||||
import { performance } from 'node:perf_hooks'
|
||||
import { scheduler } from 'node:timers/promises'
|
||||
import { randomBytes } from 'node:crypto'
|
||||
import {
|
||||
DEFAULT_PREPARED_SESSION_CACHE_SIZE, SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
|
||||
@@ -19,16 +21,27 @@ import {
|
||||
} from '@deepseek-ai/dsh-session-persistence'
|
||||
import type { SessionEvent, SessionId, SessionHeader, SessionPreparation } from '@deepseek-ai/dsh-session'
|
||||
import {
|
||||
encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, projectDir, scanLog, sessionDir, toHeaderLine,
|
||||
encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, projectDir, scanLog, sessionDir,
|
||||
SessionLogScanner, toHeaderLine,
|
||||
type JsonlCompression,
|
||||
} from './format.ts'
|
||||
import { compressZstdFrame, decompressZstdFrame, decompressZstdPrefix, scanZstdFrames } from './zstd.ts'
|
||||
import {
|
||||
compressZstdFrame, createZstdFrameDecoder, decompressZstdFrame, decompressZstdPrefix, scanZstdFrames,
|
||||
} from './zstd.ts'
|
||||
import { ensureDurableDirectoryWin32, publishNewFileWin32 } from './win32.ts'
|
||||
|
||||
export type { JsonlCompression } from './format.ts'
|
||||
|
||||
const DEFAULT_PACK_CHUNKS = true
|
||||
const DEFAULT_COMPRESSION: JsonlCompression = 'zstd'
|
||||
const ZSTD_DECODE_YIELD_INTERVAL_MS = 1000
|
||||
|
||||
/** Assert that the independently decodable first frame contains only the header record. */
|
||||
function assertZstdHeaderFrame(plaintext: Buffer): void {
|
||||
if (plaintext.length === 0 || plaintext.indexOf(0x0A) !== plaintext.length - 1) {
|
||||
throw new Error('corrupt Zstandard session log: first frame is not exactly one header line')
|
||||
}
|
||||
}
|
||||
|
||||
/** Loader schema for the JSONL artifact's physical encoding. */
|
||||
export const JsonlCompressionSchema: z<JsonlCompression> = z.union([
|
||||
@@ -260,61 +273,66 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
signal?.throwIfAborted()
|
||||
if (frames.length === 0) throw new Error('empty or header-less Zstandard session log')
|
||||
|
||||
const plaintextFrames: Buffer[] = []
|
||||
for (const frame of frames) {
|
||||
let plaintext: Buffer
|
||||
try {
|
||||
const decoder = createZstdFrameDecoder()
|
||||
let yieldDeadline = performance.now() + ZSTD_DECODE_YIELD_INTERVAL_MS
|
||||
try {
|
||||
const decodedFrames = decoder.decode(buffer, frames)
|
||||
signal?.throwIfAborted()
|
||||
const headerFrame = decodedFrames.next()
|
||||
signal?.throwIfAborted()
|
||||
if (headerFrame.done) throw new Error('empty or header-less Zstandard session log')
|
||||
assertZstdHeaderFrame(headerFrame.value)
|
||||
const scanner = new SessionLogScanner(headerFrame.value)
|
||||
|
||||
let remainingFrames = frames.length - 1
|
||||
for (const plaintext of decodedFrames) {
|
||||
signal?.throwIfAborted()
|
||||
plaintext = await decompressZstdFrame(buffer.subarray(frame.start, frame.end))
|
||||
} catch (error) {
|
||||
/* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */
|
||||
if (signal?.aborted) signal.throwIfAborted()
|
||||
throw new Error(`corrupt Zstandard session log: frame at byte ${frame.start} failed validation`, { cause: error })
|
||||
scanner.write(plaintext)
|
||||
remainingFrames -= 1
|
||||
if (remainingFrames > 0 && performance.now() >= yieldDeadline) {
|
||||
await scheduler.yield()
|
||||
signal?.throwIfAborted()
|
||||
yieldDeadline = performance.now() + ZSTD_DECODE_YIELD_INTERVAL_MS
|
||||
}
|
||||
}
|
||||
signal?.throwIfAborted()
|
||||
plaintextFrames.push(plaintext)
|
||||
}
|
||||
const complete = scanner.checkpoint()
|
||||
if (complete.committedBytes !== complete.inputBytes) {
|
||||
throw new Error('corrupt Zstandard session log: complete frame contains a torn JSONL record')
|
||||
}
|
||||
if (tornStart === undefined) {
|
||||
const prefix = scanner.finish()
|
||||
return { meta: prefix.meta, events: prefix.events }
|
||||
}
|
||||
|
||||
const headerFrame = plaintextFrames[0]
|
||||
if (headerFrame === undefined || headerFrame.length === 0 || headerFrame.indexOf(0x0A) !== headerFrame.length - 1) {
|
||||
throw new Error('corrupt Zstandard session log: first frame is not exactly one header line')
|
||||
}
|
||||
signal?.throwIfAborted()
|
||||
const completePlaintext = Buffer.concat(plaintextFrames)
|
||||
signal?.throwIfAborted()
|
||||
const completePrefix = scanLog(completePlaintext)
|
||||
signal?.throwIfAborted()
|
||||
if (completePrefix.committedBytes !== completePlaintext.length) {
|
||||
throw new Error('corrupt Zstandard session log: complete frame contains a torn JSONL record')
|
||||
}
|
||||
if (tornStart === undefined) {
|
||||
return { meta: completePrefix.meta, events: completePrefix.events }
|
||||
}
|
||||
|
||||
let recoveredPlaintext: Buffer = Buffer.alloc(0)
|
||||
try {
|
||||
let recoveredPlaintext: Buffer = Buffer.alloc(0)
|
||||
try {
|
||||
signal?.throwIfAborted()
|
||||
recoveredPlaintext = await decompressZstdPrefix(buffer.subarray(tornStart))
|
||||
} catch {
|
||||
/* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */
|
||||
if (signal?.aborted) signal.throwIfAborted()
|
||||
// A structurally incomplete final frame may end before Node's decoder can
|
||||
// emit any plaintext; the complete prior frames remain recoverable.
|
||||
}
|
||||
signal?.throwIfAborted()
|
||||
recoveredPlaintext = await decompressZstdPrefix(buffer.subarray(tornStart))
|
||||
} catch {
|
||||
scanner.write(recoveredPlaintext)
|
||||
const recoveredPrefix = scanner.finish()
|
||||
signal?.throwIfAborted()
|
||||
return {
|
||||
meta: recoveredPrefix.meta,
|
||||
events: recoveredPrefix.events,
|
||||
tornMarker: {
|
||||
truncateTo: tornStart,
|
||||
recoveredEvents: recoveredPrefix.events.slice(complete.eventCount),
|
||||
},
|
||||
}
|
||||
} catch (error) {
|
||||
/* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */
|
||||
if (signal?.aborted) signal.throwIfAborted()
|
||||
// A structurally incomplete final frame may end before Node's decoder can
|
||||
// emit any plaintext; the complete prior frames remain recoverable.
|
||||
}
|
||||
signal?.throwIfAborted()
|
||||
const recoveredPrefix = scanLog(Buffer.concat([completePlaintext, recoveredPlaintext]))
|
||||
signal?.throwIfAborted()
|
||||
/* v8 ignore next 3 -- appending plaintext cannot shorten the already-scanned complete prefix */
|
||||
if (recoveredPrefix.events.length < completePrefix.events.length) {
|
||||
throw new Error('corrupt Zstandard session log: recovered prefix does not extend complete frames')
|
||||
}
|
||||
return {
|
||||
meta: recoveredPrefix.meta,
|
||||
events: recoveredPrefix.events,
|
||||
tornMarker: {
|
||||
truncateTo: tornStart,
|
||||
recoveredEvents: recoveredPrefix.events.slice(completePrefix.events.length),
|
||||
},
|
||||
throw error
|
||||
} finally {
|
||||
decoder.close()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -662,9 +680,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
throw new Error('corrupt Zstandard session log: header frame failed validation', { cause: error })
|
||||
}
|
||||
signal?.throwIfAborted()
|
||||
if (plaintext.length === 0 || plaintext.indexOf(0x0A) !== plaintext.length - 1) {
|
||||
throw new Error('corrupt Zstandard session log: first frame is not exactly one header line')
|
||||
}
|
||||
assertZstdHeaderFrame(plaintext)
|
||||
return plaintext.subarray(0, -1).toString('utf8')
|
||||
}
|
||||
} finally {
|
||||
|
||||
Reference in New Issue
Block a user