diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index 332a6338cb..982af6daab 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -280,6 +280,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi signal?.throwIfAborted() const headerFrame = decodedFrames.next() signal?.throwIfAborted() + /* v8 ignore next -- a non-empty structural frame list makes the decoder yield its first frame or throw. */ if (headerFrame.done) throw new Error('empty or header-less Zstandard session log') assertZstdHeaderFrame(headerFrame.value) const scanner = new SessionLogScanner(headerFrame.value) diff --git a/packages/session-persistence/session-persistence-jsonl/src/zstd-private-decoder.ts b/packages/session-persistence/session-persistence-jsonl/src/zstd-private-decoder.ts index f4d19b2bd0..6cc88aadf7 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/zstd-private-decoder.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/zstd-private-decoder.ts @@ -41,6 +41,7 @@ function privateZstdStream( const errorKey = Reflect.ownKeys(stream).find((key): key is symbol => ( typeof key === 'symbol' && key.description === 'kError' )) + /* v8 ignore next -- one test runtime exposes one Node-private shape; the Node 22/24/26 matrix checks compatibility. */ if ( typeof handle !== 'object' || handle === null || typeof (handle as { writeSync?: unknown }).writeSync !== 'function' @@ -48,6 +49,7 @@ function privateZstdStream( || candidate._writeState.length < 2 || typeof candidate._defaultFlushFlag !== 'number' || errorKey === undefined + || candidate[errorKey] !== null ) return undefined return { stream: stream as NodeZstdPrivateStream, errorKey } } @@ -81,10 +83,13 @@ export class NodePrivateZstdFrameDecoder implements ZstdFrameDecoder { static create(): NodePrivateZstdFrameDecoder | undefined { const stream = createZstdDecompress({ chunkSize: DECODE_CHUNK_SIZE }) const privateAccess = privateZstdStream(stream) + /* v8 ignore next -- reached only when a supported Node release changes its private stream shape. */ if (privateAccess !== undefined) { return new NodePrivateZstdFrameDecoder(privateAccess.stream, privateAccess.errorKey) } + /* v8 ignore next -- the active Node runtime passed the private-shape probe above. */ stream.close() + /* v8 ignore next -- the active Node runtime passed the private-shape probe above. */ return undefined } @@ -111,6 +116,7 @@ export class NodePrivateZstdFrameDecoder implements ZstdFrameDecoder { /** Decode one frame; its returned scratch view remains valid until the next call. */ private decodeFrame(input: Buffer): Buffer { const handle = this.stream._handle + /* v8 ignore next -- decode() rejects closed instances before entering this private frame operation. */ if (this.closed || handle === null) throw new Error('cannot decode with a closed Zstandard frame decoder') let inputOffset = 0 @@ -125,11 +131,11 @@ export class NodePrivateZstdFrameDecoder implements ZstdFrameDecoder { inputRemaining, this.output, 0, - DECODE_CHUNK_SIZE, + this.output.length, ) if (this.decoderError !== undefined) throw this.decoderError const internalError = this.stream[this.errorKey] - if (internalError !== undefined && internalError !== null) { + if (internalError !== null) { if (internalError instanceof Error) throw internalError throw new Error('Zstandard decoder exposed a non-Error internal failure') } @@ -137,21 +143,23 @@ export class NodePrivateZstdFrameDecoder implements ZstdFrameDecoder { const outputAfter = this.stream._writeState[0] const inputAfter = this.stream._writeState[1] const consumed = inputRemaining - inputAfter - const produced = DECODE_CHUNK_SIZE - outputAfter + const produced = this.output.length - outputAfter if (produced > 0) { outputBytes += produced + /* v8 ignore next -- Buffer cannot materialize a frame beyond its own process-wide maximum length. */ if (outputBytes > bufferConstants.MAX_LENGTH) { throw new Error(`Zstandard frame output exceeds ${bufferConstants.MAX_LENGTH} bytes`) } } if (outputAfter !== 0) { + /* v8 ignore next -- structurally scanned ranges contain exactly one complete frame and no trailing bytes. */ if (inputAfter !== 0) throw new Error('Zstandard frame decoder left trailing input') const finalChunk = this.output.subarray(0, produced) if (fullChunks.length === 0) return finalChunk if (produced > 0) fullChunks.push(Buffer.from(finalChunk)) - const [onlyChunk] = fullChunks - return fullChunks.length === 1 && onlyChunk !== undefined + const onlyChunk = fullChunks[0] as Buffer + return fullChunks.length === 1 ? onlyChunk : Buffer.concat(fullChunks, outputBytes) } diff --git a/packages/session-persistence/session-persistence-jsonl/src/zstd.ts b/packages/session-persistence/session-persistence-jsonl/src/zstd.ts index 0a4bb9f2ea..01b4d459f0 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/zstd.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/zstd.ts @@ -6,7 +6,7 @@ */ import { - constants, zstdCompress, zstdDecompress, zstdDecompressSync, type ZstdOptions, + constants, zstdCompress, zstdDecompress, type ZstdOptions, } from 'node:zlib' import { promisify } from 'node:util' import { NodePrivateZstdFrameDecoder } from './zstd-private-decoder.ts' @@ -121,17 +121,6 @@ export async function decompressZstdFrame(input: Buffer): Promise { return zstdDecompressAsync(input) } -/** - * Synchronously decompress one complete frame and validate its checksum. - * Complete-log readers time-slice repeated calls so the event loop regains - * control without paying one asynchronous native dispatch per frame. - * @param input - one structurally complete Zstandard frame. - * @returns the frame plaintext. - */ -export function decompressZstdFrameSync(input: Buffer): Buffer { - return zstdDecompressSync(input) -} - /** Common lifecycle for interchangeable synchronous multi-frame decoders. */ export interface ZstdFrameDecoder { /** diff --git a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts index db1624cb8f..bebdee7870 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts @@ -694,6 +694,46 @@ describe('SessionPersistenceJsonl: write path (session/event → flush)', () => describe('SessionPersistenceJsonl: scanLog unit', () => { + it('requires exactly one newline-terminated header record', () => { + const header = JSON.stringify(toHeaderLine(meta('scanner-header'))) + expect(() => new SessionLogScanner(Buffer.alloc(0))).toThrow(/header-less/) + expect(() => new SessionLogScanner(Buffer.from(header))).toThrow(/header-less/) + expect(() => new SessionLogScanner(Buffer.from(`${header}\n${header}\n`))).toThrow(/header-less/) + }) + + it('handles empty writes, boundary newlines, torn fragments, and scanner completion', () => { + const header = Buffer.from(`${JSON.stringify(toHeaderLine(meta('scanner-lifecycle')))}\n`) + const event = Buffer.from(JSON.stringify(oneTurnLog()[0])) + const scanner = new SessionLogScanner(header) + + scanner.write(Buffer.alloc(0)) + scanner.write(event) + scanner.write(Buffer.from('\nignored torn tail')) + const result = scanner.finish() + + expect(result.events).toEqual([oneTurnLog()[0]]) + expect(result.committedBytes).toBe(header.length + event.length + 1) + expect(() =>{ scanner.write(Buffer.from('\n')) }).toThrow(/finished/) + }) + + it('keeps scanning after a tolerable corrupt suffix until a committed turn end appears', () => { + const header = Buffer.from(`${JSON.stringify(toHeaderLine(meta('scanner-corrupt-suffix')))}\n`) + const scanner = new SessionLogScanner(header) + scanner.write(Buffer.from([ + JSON.stringify(oneTurnLog()[0]), + '{not json', + JSON.stringify({ type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } }), + '', + ].join('\n'))) + expect(scanner.finish().events).toEqual([oneTurnLog()[0]]) + + const committed = new SessionLogScanner(header) + expect(() =>{ committed.write(Buffer.from([ + JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }), + '', + ].join('\n'))) }).toThrow(/seq gap in committed region/) + }) + it('incrementally scans records split across reusable decoder chunks', () => { const header = Buffer.from(`${JSON.stringify(toHeaderLine(meta('incremental')))}\n`) const body = Buffer.from(`${oneTurnLog().map(event => JSON.stringify(event)).join('\n').replace('"hi"', '"你好"')}\n`) diff --git a/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts index 13936f449f..44c6655386 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts @@ -10,8 +10,8 @@ import type { SessionEvent } from '@deepseek-ai/dsh-session' import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl' import { logPath, scanLog, sessionDir, toHeaderLine, type JsonlCompression } from '../src/format.ts' import { - compressZstdFrame, createZstdFrameDecoder, decompressZstdFrame, decompressZstdFrameSync, decompressZstdPrefix, - scanZstdFrames, + compressZstdFrame, createZstdFrameDecoder, decompressZstdFrame, decompressZstdPrefix, scanZstdFrames, + type ZstdFrameDecoder, } from '../src/zstd.ts' import { NodePrivateZstdFrameDecoder } from '../src/zstd-private-decoder.ts' import { PublicZstdFrameDecoder } from '../src/zstd-public-decoder.ts' @@ -157,7 +157,6 @@ describe('Zstandard frame structure', () => { expect(first[4]! & 0x04).toBe(0x04) expect(second[4]! & 0x04).toBe(0x04) expect((await decompressZstdFrame(first)).toString()).toBe('header\n') - expect(decompressZstdFrameSync(second).toString()).toBe('event\n') const decoder = createZstdFrameDecoder() try { const plaintext = Array.from(decoder.decode(stream, scanZstdFrames(stream).frames), chunk => Buffer.from(chunk)) @@ -185,6 +184,96 @@ describe('Zstandard frame structure', () => { } }) + it('falls back to the public decoder when the private Node contract is unavailable', () => { + vi.spyOn(NodePrivateZstdFrameDecoder, 'create').mockReturnValue(undefined) + const decoder = createZstdFrameDecoder() + expect(decoder).toBeInstanceOf(PublicZstdFrameDecoder) + decoder.close() + }) + + it('enforces decoder lifecycle and checksum errors through both implementations', async () => { + const frame = await compressZstdFrame('frame\n') + const range = [{ start: 0, end: frame.length }] + const corrupt = Buffer.from(frame) + corrupt[corrupt.length - 1] = corrupt[corrupt.length - 1]! ^ 0xFF + const factories: Array<() => ZstdFrameDecoder> = [ + () => new PublicZstdFrameDecoder(), + () => NodePrivateZstdFrameDecoder.create()!, + ] + + for (const create of factories) { + const interrupted = create() + const iterator = interrupted.decode(frame, range) + expect(iterator.next().value?.toString()).toBe('frame\n') + iterator.return() + expect(() => Array.from(interrupted.decode(frame, range))).toThrow(/already started/) + interrupted.close() + + const closed = create() + closed.close() + closed.close() + expect(() => Array.from(closed.decode(frame, range))).toThrow(/closed/) + + const invalid = create() + expect(() => Array.from(invalid.decode(corrupt, range))).toThrow(/frame at byte 0 failed validation/) + } + }) + + it('assembles private-decoder output at and beyond its reusable chunk boundary', async () => { + for (const length of [8, 9]) { + const plaintext = Buffer.alloc(length, 0x61) + const frame = await compressZstdFrame(plaintext) + const decoder = NodePrivateZstdFrameDecoder.create()! + ;(decoder as unknown as { output: Buffer }).output = Buffer.allocUnsafe(8) + const [decoded] = Array.from( + decoder.decode(frame, [{ start: 0, end: frame.length }]), + chunk => Buffer.from(chunk), + ) + expect(decoded).toEqual(plaintext) + } + }) + + it('normalizes private decoder stream failures', async () => { + interface PrivateDecoderInternals { + stream: { + [key: symbol]: unknown + emit(event: string, error: Error): boolean + } + errorKey: symbol + } + const frame = await compressZstdFrame('frame\n') + const range = [{ start: 0, end: frame.length }] + + const emitted = NodePrivateZstdFrameDecoder.create()! + const emittedInternals = emitted as unknown as PrivateDecoderInternals + const first = new Error('first emitted decoder failure') + emittedInternals.stream.emit('error', first) + emittedInternals.stream.emit('error', new Error('later emitted decoder failure')) + try { + Array.from(emitted.decode(frame, range)) + throw new Error('expected emitted decoder failure') + } catch (error) { + expect((error as Error).cause).toBe(first) + } + + for (const internalFailure of [new Error('internal decoder failure'), 'not an Error']) { + const decoder = NodePrivateZstdFrameDecoder.create()! + const internals = decoder as unknown as PrivateDecoderInternals + internals.stream[internals.errorKey] = internalFailure + try { + Array.from(decoder.decode(frame, range)) + throw new Error('expected internal decoder failure') + } catch (error) { + const cause = (error as Error).cause + if (internalFailure instanceof Error) { + expect(cause).toBe(internalFailure) + } else { + expect(cause).toMatchObject({ message: 'Zstandard decoder exposed a non-Error internal failure' }) + } + } + } + }) + it('distinguishes incomplete frame regions from invalid complete structure', () => { expect(scanZstdFrames(MAGIC.subarray(0, 2))).toEqual({ frames: [], tornStart: 0 }) expect(scanZstdFrames(MAGIC)).toEqual({ frames: [], tornStart: 0 })