|
|
@@ -19,33 +19,45 @@ import {
|
|
|
type FileHandle,
|
|
|
} from 'node:fs/promises'
|
|
|
import { basename, dirname, join } from 'node:path'
|
|
|
+import { performance } from 'node:perf_hooks'
|
|
|
+import { pipeline, Readable } from 'node:stream'
|
|
|
+import { scheduler } from 'node:timers/promises'
|
|
|
+import { constants, createZstdCompress } from 'node:zlib'
|
|
|
+import { Session } from '@deepseek-ai/dsh-session'
|
|
|
+import type {
|
|
|
+ SessionFormatArtifact,
|
|
|
+ SessionFormatJsonValue,
|
|
|
+ SessionFormatRestore,
|
|
|
+} from '@deepseek-ai/dsh-session-format'
|
|
|
+import { validateStoredEvents } from '@deepseek-ai/dsh-session-persistence'
|
|
|
import type { JsonlCompression } from './format.ts'
|
|
|
-import { generationLogFilename, logSuffix } from './format.ts'
|
|
|
+import { generationLogFilename, logSuffix, SessionLogScanner } from './format.ts'
|
|
|
import { publishNewFileWin32 } from './win32.ts'
|
|
|
import {
|
|
|
compressZstdFrame,
|
|
|
createZstdFrameDecoder,
|
|
|
- decompressZstdFrame,
|
|
|
decompressZstdPrefix,
|
|
|
scanZstdFrames,
|
|
|
} from './zstd.ts'
|
|
|
|
|
|
-/** Parsed JSONL values supplied to the format catalog. */
|
|
|
-export interface JsonlDecodedGeneration {
|
|
|
- readonly header: Record<string, unknown>
|
|
|
- readonly rows: readonly unknown[]
|
|
|
+/** Internal scheduling bounds: preserve old decode cadence and cap each synchronous encode slice. */
|
|
|
+const MIGRATION_DECODE_YIELD_INTERVAL_MS = 500
|
|
|
+const MIGRATION_WORK_CHUNK_BYTES = 1024 * 1024
|
|
|
+const MIGRATION_WRITE_CHUNK_BYTES = 4 * 1024 * 1024
|
|
|
+const ZSTD_CHECKSUM_OPTIONS = {
|
|
|
+ chunkSize: MIGRATION_WORK_CHUNK_BYTES,
|
|
|
+ params: { [constants.ZSTD_c_checksumFlag]: 1 },
|
|
|
}
|
|
|
|
|
|
-/** Current JSONL values returned by the format catalog for physical encoding. */
|
|
|
-export interface JsonlCurrentGeneration extends JsonlDecodedGeneration {}
|
|
|
-
|
|
|
/** Pure adapter between backend-owned JSONL framing and the format catalog. */
|
|
|
export interface JsonlGenerationFormatAdapter {
|
|
|
readonly currentVersion: number
|
|
|
- /** Convert one detached historical generation to exact current JSON values. */
|
|
|
- migrate(source: JsonlDecodedGeneration): JsonlCurrentGeneration
|
|
|
- /** Validate one decoded current generation, including after committed reopen. */
|
|
|
- validateCurrent(candidate: JsonlCurrentGeneration): void
|
|
|
+ /** Create the single-pass codec and migration state for a historical header. */
|
|
|
+ createRestore(header: Record<string, unknown>): SessionFormatRestore
|
|
|
+ /** Encode one current header record without materializing body rows. */
|
|
|
+ encodeHeader(header: SessionFormatArtifact['header'], inheritedEventCount: number): SessionFormatJsonValue
|
|
|
+ /** Encode one current event record. */
|
|
|
+ encodeEvent(event: SessionFormatArtifact['events'][number]): SessionFormatJsonValue
|
|
|
/** Classify a supported-version artifact that policy refuses to migrate. */
|
|
|
isUnsupportedMigrationError?(error: unknown): error is Error
|
|
|
}
|
|
|
@@ -64,9 +76,31 @@ export interface EnsureJsonlGenerationOptions {
|
|
|
readonly validateHistoricalHeader?: (
|
|
|
header: Readonly<Record<string, unknown>>,
|
|
|
) => void | Promise<void>
|
|
|
+ /** Validate the staged file in an isolated worker before publication. */
|
|
|
+ readonly verifyCurrentFile: (
|
|
|
+ path: string,
|
|
|
+ compression: JsonlCompression,
|
|
|
+ expectedId: string,
|
|
|
+ expectedEventCount: number,
|
|
|
+ expectedPrefix?: JsonlExpectedPrefix,
|
|
|
+ signal?: AbortSignal,
|
|
|
+ ) => Promise<JsonlVerifiedGeneration>
|
|
|
readonly signal?: AbortSignal
|
|
|
}
|
|
|
|
|
|
+/** Small physical identity returned by an isolated generation verifier. */
|
|
|
+export interface JsonlVerifiedGeneration {
|
|
|
+ readonly identity: JsonlPhysicalIdentity
|
|
|
+ readonly bytes: number
|
|
|
+ readonly digest: string
|
|
|
+}
|
|
|
+
|
|
|
+/** Physical byte prefix already proven to be a valid complete generation. */
|
|
|
+export interface JsonlExpectedPrefix {
|
|
|
+ readonly bytes: number
|
|
|
+ readonly digest: string
|
|
|
+}
|
|
|
+
|
|
|
/** Result of current classification or exclusive publication. */
|
|
|
export type EnsureJsonlGenerationResult =
|
|
|
| {
|
|
|
@@ -88,11 +122,6 @@ export type EnsureJsonlGenerationResult =
|
|
|
export class JsonlGenerationNewerVersionError extends Error {
|
|
|
override readonly name = 'JsonlGenerationNewerVersionError'
|
|
|
|
|
|
- /**
|
|
|
- * @param storedVersion - version read from the highest stored generation.
|
|
|
- * @param currentVersion - version this build writes.
|
|
|
- * @param storedId - minimally decoded identity used in the refusal diagnostic.
|
|
|
- */
|
|
|
constructor(
|
|
|
readonly storedVersion: number,
|
|
|
readonly currentVersion: number,
|
|
|
@@ -143,10 +172,8 @@ export interface JsonlPhysicalIdentity {
|
|
|
readonly ctimeNs: bigint
|
|
|
}
|
|
|
|
|
|
-/** One revision-stable physical artifact read reusable by the immediate backend hook. */
|
|
|
-export interface JsonlPhysicalSnapshot {
|
|
|
- readonly bytes: Buffer
|
|
|
- readonly identity: JsonlPhysicalIdentity
|
|
|
+/** One revision-stable physical artifact returned to the immediate backend decoder. */
|
|
|
+export interface JsonlPhysicalSnapshot extends StablePhysicalFile {
|
|
|
readonly headerValue: Record<string, unknown>
|
|
|
readonly headerRecord: Buffer
|
|
|
}
|
|
|
@@ -162,11 +189,6 @@ interface JsonlPhysicalHeader {
|
|
|
readonly record: Buffer
|
|
|
}
|
|
|
|
|
|
-interface DecodedPhysicalJsonl {
|
|
|
- readonly bytes: Buffer
|
|
|
- readonly torn: boolean
|
|
|
-}
|
|
|
-
|
|
|
interface GenerationFileSystem {
|
|
|
open(path: string, flags: string, mode?: number): Promise<FileHandle>
|
|
|
readFile(path: string, signal?: AbortSignal): Promise<Buffer>
|
|
|
@@ -189,10 +211,24 @@ interface JsonlGenerationInternals {
|
|
|
readonly barrier: (phase: GenerationBarrierPhase, attempt: number) => void | Promise<void>
|
|
|
}
|
|
|
|
|
|
-type JsonlGenerationTestOverrides = Partial<Omit<JsonlGenerationInternals, 'fs'>> & {
|
|
|
+/** Dependency overrides for an isolated generation runtime. */
|
|
|
+export type JsonlGenerationRuntimeOverrides = Partial<Omit<JsonlGenerationInternals, 'fs'>> & {
|
|
|
readonly fs?: Partial<GenerationFileSystem>
|
|
|
}
|
|
|
|
|
|
+/** Bound generation operations used by production defaults and deterministic tests. */
|
|
|
+export interface JsonlGenerationRuntime {
|
|
|
+ readStable(path: string, signal?: AbortSignal): Promise<StablePhysicalFile>
|
|
|
+ ensure(options: EnsureJsonlGenerationOptions): Promise<EnsureJsonlGenerationResult>
|
|
|
+ verify(
|
|
|
+ path: string,
|
|
|
+ compression: JsonlCompression,
|
|
|
+ expectedId: string,
|
|
|
+ expectedEventCount: number,
|
|
|
+ expectedPrefix?: JsonlExpectedPrefix,
|
|
|
+ ): Promise<JsonlVerifiedGeneration>
|
|
|
+}
|
|
|
+
|
|
|
const defaultFileSystem: GenerationFileSystem = {
|
|
|
open: (path, flags, mode) => fsOpen(path, flags, mode),
|
|
|
readFile: (path, signal) => fsReadFile(path, signal === undefined ? undefined : { signal }),
|
|
|
@@ -240,7 +276,7 @@ export async function readStableJsonlFile(
|
|
|
path: string,
|
|
|
signal?: AbortSignal,
|
|
|
): Promise<StablePhysicalFile> {
|
|
|
- return readStableSnapshot(path, signal, defaultFileSystem)
|
|
|
+ return defaultGenerationRuntime.readStable(path, signal)
|
|
|
}
|
|
|
|
|
|
async function readStableSnapshot(
|
|
|
@@ -289,120 +325,299 @@ function parseJson(text: string, subject: string): unknown {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-function parseGeneration(bytes: Buffer, recoverSuffix = false): JsonlDecodedGeneration {
|
|
|
- /* v8 ignore next -- decodePhysicalJsonl supplies a non-empty newline-terminated prefix. */
|
|
|
- if (bytes.length === 0 || bytes.at(-1) !== 0x0A) {
|
|
|
- throw new Error('empty or header-less session log')
|
|
|
+/** Incremental JSONL parser that retains only one cross-frame record fragment. */
|
|
|
+class MigratingJsonlRows {
|
|
|
+ private fragments: Buffer[] = []
|
|
|
+ private fragmentBytes = 0
|
|
|
+ private rowIndex = 0
|
|
|
+ private issue: Error | undefined
|
|
|
+
|
|
|
+ constructor(private readonly restore: SessionFormatRestore) {}
|
|
|
+
|
|
|
+ /** Consume plaintext bytes following the independently decoded header. */
|
|
|
+ write(chunk: Buffer): void {
|
|
|
+ /* jscpd:ignore-start -- migration parsing and readable-log scanning own different recovery and byte-accounting state. */
|
|
|
+ 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
|
|
|
+ }
|
|
|
+ this.consume(line)
|
|
|
+ lineStart = newline + 1
|
|
|
+ }
|
|
|
+ if (lineStart < chunk.length) {
|
|
|
+ const fragment = Buffer.from(chunk.subarray(lineStart))
|
|
|
+ this.fragments.push(fragment)
|
|
|
+ this.fragmentBytes += fragment.length
|
|
|
+ }
|
|
|
+ /* jscpd:ignore-end */
|
|
|
}
|
|
|
- const records = bytes.toString('utf8').slice(0, -1).split('\n')
|
|
|
- const parsedHeader = parseJson(records[0] as string, 'header line')
|
|
|
- storedVersion(parsedHeader)
|
|
|
- const rows: unknown[] = []
|
|
|
- let issue: Error | undefined
|
|
|
- for (const [index, record] of records.slice(1).entries()) {
|
|
|
+
|
|
|
+ /** Refuse a record fragment left by structurally complete Zstandard frames. */
|
|
|
+ assertCompleteFramesEndOnRecord(): void {
|
|
|
+ if (this.fragments.length > 0) {
|
|
|
+ throw new Error('corrupt Zstandard session log: complete frame contains a torn JSONL record')
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ finish(): SessionFormatArtifact {
|
|
|
+ return this.restore.finish()
|
|
|
+ }
|
|
|
+
|
|
|
+ private consume(line: Buffer): void {
|
|
|
+ const index = this.rowIndex
|
|
|
+ this.rowIndex += 1
|
|
|
let row: unknown
|
|
|
try {
|
|
|
- row = parseJson(record, `row ${index + 1}`)
|
|
|
- } catch (error) {
|
|
|
- if (!recoverSuffix) throw error
|
|
|
- issue ??= error as Error
|
|
|
- continue
|
|
|
+ row = parseJson(line.toString('utf8'), `row ${index + 1}`)
|
|
|
+ } catch (error: unknown) {
|
|
|
+ this.issue ??= asError(error)
|
|
|
+ return
|
|
|
}
|
|
|
- if (issue !== undefined) {
|
|
|
+ if (this.issue !== undefined) {
|
|
|
if (typeof row === 'object' && row !== null
|
|
|
- && (row as { type?: unknown }).type === 'turn/end') throw issue
|
|
|
- continue
|
|
|
+ && (row as { type?: unknown }).type === 'turn/end') throw this.issue
|
|
|
+ return
|
|
|
}
|
|
|
- rows.push(row)
|
|
|
+ this.restore.decodeRow(row)
|
|
|
}
|
|
|
- return { header: parsedHeader as Record<string, unknown>, rows }
|
|
|
}
|
|
|
|
|
|
-function stringifyJson(value: unknown, subject: string): string {
|
|
|
- let text: unknown
|
|
|
- try {
|
|
|
- text = JSON.stringify(value)
|
|
|
- } catch (error) {
|
|
|
- throw new Error(`${subject} is not lossless JSON`, { cause: error })
|
|
|
- }
|
|
|
- if (typeof text !== 'string') throw new Error(`${subject} is not lossless JSON`)
|
|
|
- return text
|
|
|
+interface StartedMigrationStream {
|
|
|
+ readonly parser: MigratingJsonlRows
|
|
|
}
|
|
|
|
|
|
-function encodeLogicalJsonl(generation: JsonlCurrentGeneration): Buffer {
|
|
|
- const records = [
|
|
|
- stringifyJson(generation.header, 'migrated session header'),
|
|
|
- ...generation.rows.map((row, index) => stringifyJson(row, `migrated session row ${index + 1}`)),
|
|
|
- ]
|
|
|
- return Buffer.from(`${records.join('\n')}\n`)
|
|
|
+async function startMigrationStream(
|
|
|
+ headerRecord: Buffer,
|
|
|
+ format: JsonlGenerationFormatAdapter,
|
|
|
+ validateHistoricalHeader?: EnsureJsonlGenerationOptions['validateHistoricalHeader'],
|
|
|
+): Promise<StartedMigrationStream> {
|
|
|
+ const value = parseJson(headerRecord.subarray(0, -1).toString('utf8'), 'header line')
|
|
|
+ const header = value as Record<string, unknown>
|
|
|
+ const validation = validateHistoricalHeader?.(header)
|
|
|
+ if (validation !== undefined) await validation
|
|
|
+ const stream = format.createRestore(header)
|
|
|
+ return { parser: new MigratingJsonlRows(stream) }
|
|
|
}
|
|
|
|
|
|
-function assertIndependentHeaderFrame(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')
|
|
|
+async function consumeMigrationBytes(
|
|
|
+ rows: MigratingJsonlRows,
|
|
|
+ chunks: Iterable<Buffer>,
|
|
|
+ signal?: AbortSignal,
|
|
|
+): Promise<void> {
|
|
|
+ signal?.throwIfAborted()
|
|
|
+ let yieldDeadline = performance.now() + MIGRATION_DECODE_YIELD_INTERVAL_MS
|
|
|
+ for (const bytes of chunks) {
|
|
|
+ for (let offset = 0; offset < bytes.length; offset += MIGRATION_WORK_CHUNK_BYTES) {
|
|
|
+ rows.write(bytes.subarray(offset, offset + MIGRATION_WORK_CHUNK_BYTES))
|
|
|
+ if (performance.now() < yieldDeadline) continue
|
|
|
+ await scheduler.yield()
|
|
|
+ signal?.throwIfAborted()
|
|
|
+ yieldDeadline = performance.now() + MIGRATION_DECODE_YIELD_INTERVAL_MS
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-async function decodeZstdJsonl(bytes: Buffer, signal?: AbortSignal): Promise<DecodedPhysicalJsonl> {
|
|
|
+async function decodeStreamingMigration(
|
|
|
+ bytes: Buffer,
|
|
|
+ compression: JsonlCompression,
|
|
|
+ format: JsonlGenerationFormatAdapter,
|
|
|
+ validateHistoricalHeader: EnsureJsonlGenerationOptions['validateHistoricalHeader'],
|
|
|
+ signal?: AbortSignal,
|
|
|
+): Promise<SessionFormatArtifact> {
|
|
|
signal?.throwIfAborted()
|
|
|
+ if (compression === 'none') {
|
|
|
+ const headerEnd = bytes.indexOf(0x0A)
|
|
|
+ /* v8 ignore next -- ensureCurrent's physical-header preflight already requires this newline. */
|
|
|
+ if (headerEnd === -1) throw new Error('empty or header-less session log')
|
|
|
+ const stream = await startMigrationStream(
|
|
|
+ bytes.subarray(0, headerEnd + 1),
|
|
|
+ format,
|
|
|
+ validateHistoricalHeader,
|
|
|
+ )
|
|
|
+ signal?.throwIfAborted()
|
|
|
+ const bodyEnd = bytes.lastIndexOf(0x0A)
|
|
|
+ if (bodyEnd > headerEnd) {
|
|
|
+ await consumeMigrationBytes(
|
|
|
+ stream.parser,
|
|
|
+ [bytes.subarray(headerEnd + 1, bodyEnd + 1)],
|
|
|
+ signal,
|
|
|
+ )
|
|
|
+ }
|
|
|
+ return stream.parser.finish()
|
|
|
+ }
|
|
|
+
|
|
|
const { frames, tornStart } = scanZstdFrames(bytes)
|
|
|
- /* v8 ignore next -- the independent header probe already established the first frame. */
|
|
|
+ /* v8 ignore next -- ensureCurrent's physical-header preflight already requires a complete header frame. */
|
|
|
if (frames.length === 0) throw new Error('empty or header-less Zstandard session log')
|
|
|
- const complete: Buffer[] = []
|
|
|
- for (const [index, frame] of frames.entries()) {
|
|
|
+ const decoder = createZstdFrameDecoder()
|
|
|
+ try {
|
|
|
+ const decoded = decoder.decode(bytes, frames)
|
|
|
+ const first = decoded.next()
|
|
|
+ /* v8 ignore next -- a non-empty structural frame list yields once or throws. */
|
|
|
+ if (first.done) throw new Error('empty or header-less Zstandard session log')
|
|
|
+ assertIndependentHeaderFrame(first.value)
|
|
|
+ const stream = await startMigrationStream(
|
|
|
+ first.value,
|
|
|
+ format,
|
|
|
+ validateHistoricalHeader,
|
|
|
+ )
|
|
|
signal?.throwIfAborted()
|
|
|
- const plaintext = await decompressZstdFrame(bytes.subarray(frame.start, frame.end))
|
|
|
- if (index === 0) assertIndependentHeaderFrame(plaintext)
|
|
|
- complete.push(plaintext)
|
|
|
- }
|
|
|
- const completeBytes = Buffer.concat(complete)
|
|
|
- if (completeBytes.at(-1) !== 0x0A) {
|
|
|
- throw new Error('corrupt Zstandard session log: complete frame contains a torn JSONL record')
|
|
|
+ await consumeMigrationBytes(stream.parser, decoded, signal)
|
|
|
+ stream.parser.assertCompleteFramesEndOnRecord()
|
|
|
+ if (tornStart !== undefined) {
|
|
|
+ let recovered: Buffer = Buffer.alloc(0)
|
|
|
+ try {
|
|
|
+ recovered = await decompressZstdPrefix(bytes.subarray(tornStart))
|
|
|
+ } catch {
|
|
|
+ /* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent. */
|
|
|
+ if (signal?.aborted) signal.throwIfAborted()
|
|
|
+ }
|
|
|
+ signal?.throwIfAborted()
|
|
|
+ const newline = recovered.lastIndexOf(0x0A)
|
|
|
+ if (newline !== -1) {
|
|
|
+ await consumeMigrationBytes(
|
|
|
+ stream.parser,
|
|
|
+ [recovered.subarray(0, newline + 1)],
|
|
|
+ signal,
|
|
|
+ )
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return stream.parser.finish()
|
|
|
+ } finally {
|
|
|
+ decoder.close()
|
|
|
}
|
|
|
- if (tornStart === undefined) return { bytes: completeBytes, torn: false }
|
|
|
+}
|
|
|
|
|
|
- let recovered = Buffer.alloc(0)
|
|
|
- try {
|
|
|
- recovered = Buffer.from(await decompressZstdPrefix(bytes.subarray(tornStart)))
|
|
|
- } catch {
|
|
|
- /* v8 ignore next -- an abort racing decoder failure is timing-dependent */
|
|
|
- if (signal?.aborted) signal.throwIfAborted()
|
|
|
- // A structurally torn frame may produce no plaintext; prior frames remain valid.
|
|
|
+/**
|
|
|
+ * Read and validate one complete current generation for an isolated verifier.
|
|
|
+ * @param path - staged or competing current-generation path.
|
|
|
+ * @param compression - configured physical encoding.
|
|
|
+ * @param expectedId - Session identity expected in the header.
|
|
|
+ * @param expectedEventCount - exact logical event count expected after decoding.
|
|
|
+ * @param expectedPrefix - verified migration prefix; an append tail may be present and is not validated.
|
|
|
+ * @returns stable physical identity and digest for publication comparison.
|
|
|
+ */
|
|
|
+export async function verifyJsonlCurrentGeneration(
|
|
|
+ path: string,
|
|
|
+ compression: JsonlCompression,
|
|
|
+ expectedId: string,
|
|
|
+ expectedEventCount: number,
|
|
|
+ expectedPrefix?: JsonlExpectedPrefix,
|
|
|
+): Promise<JsonlVerifiedGeneration> {
|
|
|
+ return defaultGenerationRuntime.verify(path, compression, expectedId, expectedEventCount, expectedPrefix)
|
|
|
+}
|
|
|
+
|
|
|
+async function verifyCurrentGeneration(
|
|
|
+ path: string,
|
|
|
+ compression: JsonlCompression,
|
|
|
+ expectedId: string,
|
|
|
+ expectedEventCount: number,
|
|
|
+ fs: GenerationFileSystem,
|
|
|
+ expectedPrefix?: JsonlExpectedPrefix,
|
|
|
+): Promise<JsonlVerifiedGeneration> {
|
|
|
+ const before = await fs.stat(path)
|
|
|
+ const bytes = await fs.readFile(path)
|
|
|
+ const after = await fs.stat(path)
|
|
|
+ if (expectedPrefix !== undefined) {
|
|
|
+ if (bytes.length < expectedPrefix.bytes) {
|
|
|
+ throw new Error('target bytes are shorter than the migrated generation')
|
|
|
+ }
|
|
|
+ const digest = createHash('sha256').update(bytes.subarray(0, expectedPrefix.bytes)).digest('hex')
|
|
|
+ if (digest !== expectedPrefix.digest) {
|
|
|
+ throw new Error('target bytes do not begin with the migrated generation')
|
|
|
+ }
|
|
|
+ return { identity: after, bytes: expectedPrefix.bytes, digest }
|
|
|
}
|
|
|
- signal?.throwIfAborted()
|
|
|
- const newline = recovered.lastIndexOf(0x0A)
|
|
|
+ if (identity(before) !== identity(after)) {
|
|
|
+ throw new Error('current session generation changed during verification')
|
|
|
+ }
|
|
|
+ const snapshot = { bytes, identity: after }
|
|
|
+ const generation = decodeCurrentGeneration(snapshot.bytes, compression)
|
|
|
+ validateStoredEvents(generation.meta, generation.events, { kind: 'jsonl', path })
|
|
|
+ if (generation.meta.id !== expectedId) {
|
|
|
+ throw new Error(`current session generation contains id "${generation.meta.id}", expected "${expectedId}"`)
|
|
|
+ }
|
|
|
+ if (generation.events.length !== expectedEventCount) {
|
|
|
+ throw new Error(
|
|
|
+ `current session generation contains ${generation.events.length} events, expected ${expectedEventCount}`,
|
|
|
+ )
|
|
|
+ }
|
|
|
+ Session.fromRestore(
|
|
|
+ generation.meta.id,
|
|
|
+ generation.events,
|
|
|
+ generation.meta,
|
|
|
+ generation.inheritedEventCount,
|
|
|
+ )
|
|
|
return {
|
|
|
- bytes: newline === -1
|
|
|
- ? completeBytes
|
|
|
- : Buffer.concat([completeBytes, recovered.subarray(0, newline + 1)]),
|
|
|
- torn: true,
|
|
|
+ identity: snapshot.identity,
|
|
|
+ bytes: snapshot.bytes.length,
|
|
|
+ digest: createHash('sha256').update(snapshot.bytes).digest('hex'),
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-async function decodePhysicalJsonl(
|
|
|
+function decodeCurrentGeneration(
|
|
|
bytes: Buffer,
|
|
|
compression: JsonlCompression,
|
|
|
- signal?: AbortSignal,
|
|
|
-): Promise<DecodedPhysicalJsonl> {
|
|
|
- if (compression === 'zstd') return decodeZstdJsonl(bytes, signal)
|
|
|
- signal?.throwIfAborted()
|
|
|
- const newline = bytes.lastIndexOf(0x0A)
|
|
|
- /* v8 ignore next -- physical header classification already found a newline in the same stable bytes. */
|
|
|
- if (newline === -1) throw new Error('empty or header-less session log')
|
|
|
- return { bytes: bytes.subarray(0, newline + 1), torn: newline + 1 !== bytes.length }
|
|
|
+): ReturnType<SessionLogScanner['finish']> {
|
|
|
+ if (compression === 'none') {
|
|
|
+ const headerEnd = bytes.indexOf(0x0A)
|
|
|
+ if (headerEnd === -1) throw new Error('empty or header-less session log')
|
|
|
+ const scanner = new SessionLogScanner(bytes.subarray(0, headerEnd + 1), 'strict')
|
|
|
+ scanner.write(bytes.subarray(headerEnd + 1))
|
|
|
+ return finishCurrentGenerationScan(scanner)
|
|
|
+ }
|
|
|
+ const { frames, tornStart } = scanZstdFrames(bytes)
|
|
|
+ if (frames.length === 0) throw new Error('empty or header-less Zstandard session log')
|
|
|
+ if (tornStart !== undefined) throw new Error('current session generation has a torn physical tail')
|
|
|
+ const decoder = createZstdFrameDecoder()
|
|
|
+ try {
|
|
|
+ const plaintext = decoder.decode(bytes, frames)
|
|
|
+ const header = plaintext.next()
|
|
|
+ /* v8 ignore next -- a non-empty structural frame list yields once or throws. */
|
|
|
+ if (header.done) throw new Error('empty or header-less Zstandard session log')
|
|
|
+ assertIndependentHeaderFrame(header.value)
|
|
|
+ const scanner = new SessionLogScanner(header.value, 'strict')
|
|
|
+ for (const chunk of plaintext) scanner.write(chunk)
|
|
|
+ return finishCurrentGenerationScan(scanner)
|
|
|
+ } finally {
|
|
|
+ decoder.close()
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
-async function encodePhysicalJsonl(
|
|
|
- logical: Buffer,
|
|
|
- generation: JsonlCurrentGeneration,
|
|
|
- compression: JsonlCompression,
|
|
|
-): Promise<Buffer> {
|
|
|
- if (compression === 'none') return logical
|
|
|
- const header = Buffer.from(`${stringifyJson(generation.header, 'migrated session header')}\n`)
|
|
|
- const headerFrame = await compressZstdFrame(header)
|
|
|
- if (generation.rows.length === 0) return headerFrame
|
|
|
- const body = logical.subarray(header.length)
|
|
|
- return Buffer.concat([headerFrame, await compressZstdFrame(body)])
|
|
|
+function finishCurrentGenerationScan(
|
|
|
+ scanner: SessionLogScanner,
|
|
|
+): ReturnType<SessionLogScanner['finish']> {
|
|
|
+ const inputBytes = scanner.checkpoint().inputBytes
|
|
|
+ const decoded = scanner.finish()
|
|
|
+ if (decoded.committedBytes !== inputBytes) throw new Error('current session generation has a torn physical tail')
|
|
|
+ return decoded
|
|
|
+}
|
|
|
+
|
|
|
+function stringifyJson(value: unknown, subject: string): string {
|
|
|
+ let text: unknown
|
|
|
+ try {
|
|
|
+ text = JSON.stringify(value)
|
|
|
+ } catch (error) {
|
|
|
+ throw new Error(`${subject} is not lossless JSON`, { cause: error })
|
|
|
+ }
|
|
|
+ if (typeof text !== 'string') throw new Error(`${subject} is not lossless JSON`)
|
|
|
+ return text
|
|
|
+}
|
|
|
+
|
|
|
+function assertIndependentHeaderFrame(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')
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
function readRawHeader(bytes: Buffer): JsonlPhysicalHeader {
|
|
|
@@ -441,8 +656,7 @@ function readPhysicalHeader(
|
|
|
compression: JsonlCompression,
|
|
|
signal: AbortSignal | undefined,
|
|
|
): JsonlPhysicalHeader {
|
|
|
- if (compression === 'zstd') return readZstdHeader(bytes, signal)
|
|
|
- return readRawHeader(bytes)
|
|
|
+ return compression === 'zstd' ? readZstdHeader(bytes, signal) : readRawHeader(bytes)
|
|
|
}
|
|
|
|
|
|
function assertGenerationPaths(
|
|
|
@@ -477,48 +691,127 @@ async function syncDirectory(path: string, internals: JsonlGenerationInternals):
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+interface StreamedMigrationStage {
|
|
|
+ readonly path: string
|
|
|
+ readonly bytes: number
|
|
|
+ readonly digest: string
|
|
|
+}
|
|
|
+
|
|
|
+/** Produce bounded JSONL chunks while yielding between main-thread encoding slices. */
|
|
|
+async function* encodeMigrationRows(
|
|
|
+ artifact: SessionFormatArtifact,
|
|
|
+ format: JsonlGenerationFormatAdapter,
|
|
|
+ signal?: AbortSignal,
|
|
|
+): AsyncGenerator<Buffer, void, void> {
|
|
|
+ signal?.throwIfAborted()
|
|
|
+ let lines: string[] = []
|
|
|
+ let bytes = 0
|
|
|
+ for (const value of artifact.events) {
|
|
|
+ const line = `${stringifyJson(format.encodeEvent(value), `migrated Session event ${value.seq}`)}\n`
|
|
|
+ const lineBytes = Buffer.byteLength(line)
|
|
|
+ if (bytes > 0 && bytes + lineBytes > MIGRATION_WORK_CHUNK_BYTES) {
|
|
|
+ yield Buffer.from(lines.join(''))
|
|
|
+ await scheduler.yield()
|
|
|
+ signal?.throwIfAborted()
|
|
|
+ lines = []
|
|
|
+ bytes = 0
|
|
|
+ }
|
|
|
+ lines.push(line)
|
|
|
+ bytes += lineBytes
|
|
|
+ }
|
|
|
+ yield Buffer.from(lines.join(''))
|
|
|
+}
|
|
|
+
|
|
|
+async function writeMigrationChunks(
|
|
|
+ chunks: AsyncIterable<Buffer>,
|
|
|
+ write: (chunk: Buffer) => Promise<void>,
|
|
|
+): Promise<void> {
|
|
|
+ let pending: Buffer[] = []
|
|
|
+ let bytes = 0
|
|
|
+ for await (const chunk of chunks) {
|
|
|
+ pending.push(chunk)
|
|
|
+ bytes += chunk.length
|
|
|
+ if (bytes < MIGRATION_WRITE_CHUNK_BYTES) continue
|
|
|
+ await write(pending.length === 1 ? pending[0] as Buffer : Buffer.concat(pending, bytes))
|
|
|
+ pending = []
|
|
|
+ bytes = 0
|
|
|
+ }
|
|
|
+ if (bytes > 0) await write(pending.length === 1 ? pending[0] as Buffer : Buffer.concat(pending, bytes))
|
|
|
+}
|
|
|
+
|
|
|
+/** Encode directly into one synced stage without a whole-artifact row or byte buffer. */
|
|
|
async function writeSyncedTemp(
|
|
|
currentPath: string,
|
|
|
suffix: string,
|
|
|
- bytes: Buffer,
|
|
|
+ compression: JsonlCompression,
|
|
|
+ artifact: SessionFormatArtifact,
|
|
|
+ format: JsonlGenerationFormatAdapter,
|
|
|
+ signal: AbortSignal | undefined,
|
|
|
internals: JsonlGenerationInternals,
|
|
|
-): Promise<string> {
|
|
|
+): Promise<StreamedMigrationStage> {
|
|
|
+ signal?.throwIfAborted()
|
|
|
+ let path: string
|
|
|
+ let handle: FileHandle
|
|
|
for (;;) {
|
|
|
- const path = join(dirname(currentPath), `session.migration.${internals.randomToken()}${suffix}.tmp`)
|
|
|
- let handle: FileHandle
|
|
|
+ path = join(dirname(currentPath), `session.migration.${internals.randomToken()}${suffix}.tmp`)
|
|
|
try {
|
|
|
handle = await internals.fs.open(path, 'wx', 0o600)
|
|
|
+ break
|
|
|
} catch (error) {
|
|
|
if (isEEXIST(error)) continue
|
|
|
throw error
|
|
|
}
|
|
|
- let failure: unknown
|
|
|
- try {
|
|
|
- await handle.writeFile(bytes)
|
|
|
- await handle.sync()
|
|
|
- } catch (error: unknown) {
|
|
|
- failure = error
|
|
|
- }
|
|
|
- try {
|
|
|
- await handle.close()
|
|
|
- } catch (error: unknown) {
|
|
|
- failure = failure === undefined
|
|
|
- ? error
|
|
|
- : new AggregateError([failure, error], `failed to write and close migration stage "${path}"`)
|
|
|
- }
|
|
|
- if (failure !== undefined) {
|
|
|
- const writeError = failure instanceof Error
|
|
|
- ? failure
|
|
|
- : new Error('migration stage write failed with a non-Error rejection', { cause: failure })
|
|
|
- try {
|
|
|
- await internals.fs.rm(path)
|
|
|
- } catch (cleanupError: unknown) {
|
|
|
- throw new AggregateError([writeError, cleanupError], `failed to clean migration stage "${path}"`)
|
|
|
+ }
|
|
|
+ const hash = createHash('sha256')
|
|
|
+ let bytes = 0
|
|
|
+ const write = async (chunk: Buffer): Promise<void> => {
|
|
|
+ await handle.writeFile(chunk)
|
|
|
+ hash.update(chunk)
|
|
|
+ bytes += chunk.length
|
|
|
+ }
|
|
|
+ let failure: unknown
|
|
|
+ try {
|
|
|
+ const headerValue = format.encodeHeader(artifact.header, artifact.inheritedEventCount)
|
|
|
+ const header = Buffer.from(`${stringifyJson(headerValue, 'migrated session header')}\n`)
|
|
|
+ await write(compression === 'zstd' ? await compressZstdFrame(header) : header)
|
|
|
+ if (artifact.events.length > 0) {
|
|
|
+ const rows = encodeMigrationRows(artifact, format, signal)
|
|
|
+ if (compression === 'none') {
|
|
|
+ await writeMigrationChunks(rows, write)
|
|
|
+ } else {
|
|
|
+ await new Promise<void>((resolve, reject) => {
|
|
|
+ pipeline(
|
|
|
+ Readable.from(rows, { objectMode: false, highWaterMark: MIGRATION_WORK_CHUNK_BYTES }),
|
|
|
+ createZstdCompress(ZSTD_CHECKSUM_OPTIONS),
|
|
|
+ async (source) => { await writeMigrationChunks(source as AsyncIterable<Buffer>, write) },
|
|
|
+ (error: Error | null | undefined) => {
|
|
|
+ if (error instanceof Error) reject(error)
|
|
|
+ else resolve()
|
|
|
+ },
|
|
|
+ )
|
|
|
+ })
|
|
|
}
|
|
|
- throw writeError
|
|
|
}
|
|
|
- return path
|
|
|
+ signal?.throwIfAborted()
|
|
|
+ await handle.sync()
|
|
|
+ } catch (error: unknown) {
|
|
|
+ failure = error
|
|
|
}
|
|
|
+ try {
|
|
|
+ await handle.close()
|
|
|
+ } catch (error: unknown) {
|
|
|
+ failure = failure === undefined
|
|
|
+ ? error
|
|
|
+ : new AggregateError([failure, error], `failed to write and close migration stage "${path}"`)
|
|
|
+ }
|
|
|
+ if (failure !== undefined) {
|
|
|
+ const writeError = failure instanceof Error
|
|
|
+ ? failure
|
|
|
+ : new Error('migration stage write failed with a non-Error rejection', { cause: failure })
|
|
|
+ await removeTemporary(path, writeError, internals)
|
|
|
+ throw writeError
|
|
|
+ }
|
|
|
+ return { path, bytes, digest: hash.digest('hex') }
|
|
|
}
|
|
|
|
|
|
/** Remove one temporary file without hiding the operation failure that made it disposable. */
|
|
|
@@ -550,31 +843,6 @@ async function removeCommittedTemporary(
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-async function validatePhysicalCurrent(
|
|
|
- path: string,
|
|
|
- compression: JsonlCompression,
|
|
|
- format: JsonlGenerationFormatAdapter,
|
|
|
- signal: AbortSignal | undefined,
|
|
|
- internals: JsonlGenerationInternals,
|
|
|
-): Promise<JsonlPhysicalSnapshot> {
|
|
|
- const snapshot = await readStableSnapshot(path, signal, internals.fs)
|
|
|
- const decoded = await decodePhysicalJsonl(snapshot.bytes, compression, signal)
|
|
|
- if (decoded.torn) throw new Error('staged current session generation has a torn physical tail')
|
|
|
- const generation = parseGeneration(decoded.bytes)
|
|
|
- if (storedVersion(generation.header) !== format.currentVersion) {
|
|
|
- throw new Error(`staged session generation is not current v${format.currentVersion}`)
|
|
|
- }
|
|
|
- format.validateCurrent(generation)
|
|
|
- const headerEnd = decoded.bytes.indexOf(0x0A)
|
|
|
- /* v8 ignore next -- parseGeneration already required the header newline. */
|
|
|
- if (headerEnd === -1) throw new Error('empty or header-less session log')
|
|
|
- return {
|
|
|
- ...snapshot,
|
|
|
- headerValue: generation.header,
|
|
|
- headerRecord: Buffer.from(decoded.bytes.subarray(0, headerEnd + 1)),
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
async function publishCurrentExclusive(
|
|
|
staged: string,
|
|
|
currentPath: string,
|
|
|
@@ -609,15 +877,12 @@ function asError(error: unknown): Error {
|
|
|
})
|
|
|
}
|
|
|
|
|
|
-async function reopenExpectedCurrent(
|
|
|
+async function inspectExpectedCurrent<T>(
|
|
|
currentPath: string,
|
|
|
- expectedBytes: Buffer,
|
|
|
- compression: JsonlCompression,
|
|
|
- format: JsonlGenerationFormatAdapter,
|
|
|
- signal: AbortSignal | undefined,
|
|
|
checkCanonicalTargetName: boolean,
|
|
|
internals: JsonlGenerationInternals,
|
|
|
-): Promise<JsonlPhysicalSnapshot> {
|
|
|
+ inspect: () => Promise<T>,
|
|
|
+): Promise<T> {
|
|
|
try {
|
|
|
if (checkCanonicalTargetName) {
|
|
|
const expectedName = basename(currentPath)
|
|
|
@@ -631,23 +896,16 @@ async function reopenExpectedCurrent(
|
|
|
}
|
|
|
const info = await internals.fs.lstat(currentPath)
|
|
|
if (info.isSymbolicLink() || !info.isFile()) {
|
|
|
- const kind = info.isSymbolicLink() ? 'symbolic link' : 'non-regular file'
|
|
|
- throw new Error(`target is a ${kind}`)
|
|
|
+ throw new Error(`target is a ${info.isSymbolicLink() ? 'symbolic link' : 'non-regular file'}`)
|
|
|
}
|
|
|
- const snapshot = await validatePhysicalCurrent(currentPath, compression, format, signal, internals)
|
|
|
- if (snapshot.bytes.length < expectedBytes.length
|
|
|
- || !snapshot.bytes.subarray(0, expectedBytes.length).equals(expectedBytes)) {
|
|
|
- throw new Error('target bytes do not begin with the migrated generation')
|
|
|
- }
|
|
|
- return snapshot
|
|
|
+ return await inspect()
|
|
|
} catch (error: unknown) {
|
|
|
- if (signal?.aborted) signal.throwIfAborted()
|
|
|
if (isErrnoException(error)) throw error
|
|
|
throw new JsonlGenerationTargetConflictError(currentPath, asError(error))
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-function withOverrides(overrides: JsonlGenerationTestOverrides): JsonlGenerationInternals {
|
|
|
+function withOverrides(overrides: JsonlGenerationRuntimeOverrides): JsonlGenerationInternals {
|
|
|
return {
|
|
|
...defaultInternals,
|
|
|
...overrides,
|
|
|
@@ -655,6 +913,35 @@ function withOverrides(overrides: JsonlGenerationTestOverrides): JsonlGeneration
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+async function reopenExpectedCurrent(
|
|
|
+ currentPath: string,
|
|
|
+ staged: StreamedMigrationStage,
|
|
|
+ compression: JsonlCompression,
|
|
|
+ expectedId: string,
|
|
|
+ expectedEventCount: number,
|
|
|
+ verifyCurrentFile: EnsureJsonlGenerationOptions['verifyCurrentFile'],
|
|
|
+ signal: AbortSignal | undefined,
|
|
|
+ checkCanonicalTargetName: boolean,
|
|
|
+ internals: JsonlGenerationInternals,
|
|
|
+): Promise<JsonlPhysicalSnapshot> {
|
|
|
+ return inspectExpectedCurrent(currentPath, checkCanonicalTargetName, internals, async () => {
|
|
|
+ const verified = await verifyCurrentFile(
|
|
|
+ currentPath,
|
|
|
+ compression,
|
|
|
+ expectedId,
|
|
|
+ expectedEventCount,
|
|
|
+ staged,
|
|
|
+ signal,
|
|
|
+ )
|
|
|
+ if (verified.bytes !== staged.bytes || verified.digest !== staged.digest) {
|
|
|
+ throw new Error('target bytes differ from the migrated generation')
|
|
|
+ }
|
|
|
+ const snapshot = await readStableSnapshot(currentPath, signal, internals.fs)
|
|
|
+ const header = readPhysicalHeader(snapshot.bytes, compression, signal)
|
|
|
+ return { ...snapshot, headerValue: header.value, headerRecord: header.record }
|
|
|
+ })
|
|
|
+}
|
|
|
+
|
|
|
async function ensureCurrent(
|
|
|
options: EnsureJsonlGenerationOptions,
|
|
|
internals: JsonlGenerationInternals,
|
|
|
@@ -681,68 +968,87 @@ async function ensureCurrent(
|
|
|
)
|
|
|
}
|
|
|
if (sourceVersion > format.currentVersion) {
|
|
|
- throw new JsonlGenerationNewerVersionError(sourceVersion, format.currentVersion, storedId(quickHeader.value))
|
|
|
+ throw new JsonlGenerationNewerVersionError(
|
|
|
+ sourceVersion,
|
|
|
+ format.currentVersion,
|
|
|
+ storedId(quickHeader.value),
|
|
|
+ )
|
|
|
}
|
|
|
if (sourceVersion === format.currentVersion) {
|
|
|
return {
|
|
|
status: 'current',
|
|
|
version: quickVersion,
|
|
|
path: sourcePath,
|
|
|
- snapshot: { ...source, headerValue: quickHeader.value, headerRecord: quickHeader.record },
|
|
|
+ snapshot: {
|
|
|
+ ...source,
|
|
|
+ headerValue: quickHeader.value,
|
|
|
+ headerRecord: quickHeader.record,
|
|
|
+ },
|
|
|
}
|
|
|
}
|
|
|
- const validation = options.validateHistoricalHeader?.(quickHeader.value)
|
|
|
- if (validation !== undefined) await validation
|
|
|
-
|
|
|
- const decodedSource = await decodePhysicalJsonl(source.bytes, compression, signal)
|
|
|
- const parsedSource = parseGeneration(decodedSource.bytes, true)
|
|
|
- const fromVersion = storedVersion(parsedSource.header)
|
|
|
- /* v8 ignore next -- both headers come from the same stable physical snapshot. */
|
|
|
- if (fromVersion !== quickVersion) throw new Error('session format changed within one stable physical snapshot')
|
|
|
- const sourceFingerprint = fingerprint(source.identity, source.bytes)
|
|
|
-
|
|
|
- let migrated: JsonlCurrentGeneration
|
|
|
+ let artifact: SessionFormatArtifact
|
|
|
try {
|
|
|
- migrated = format.migrate(parsedSource)
|
|
|
+ artifact = await decodeStreamingMigration(
|
|
|
+ source.bytes,
|
|
|
+ compression,
|
|
|
+ format,
|
|
|
+ options.validateHistoricalHeader,
|
|
|
+ signal,
|
|
|
+ )
|
|
|
} catch (error: unknown) {
|
|
|
if (format.isUnsupportedMigrationError?.(error) === true) {
|
|
|
- throw new JsonlGenerationUnsupportedMigrationError(fromVersion, error)
|
|
|
+ throw new JsonlGenerationUnsupportedMigrationError(sourceVersion, error)
|
|
|
}
|
|
|
throw error
|
|
|
}
|
|
|
- if (storedVersion(migrated.header) !== format.currentVersion) {
|
|
|
- throw new Error(`format migration returned v${storedVersion(migrated.header)}, expected v${format.currentVersion}`)
|
|
|
+ if (artifact.header.version !== format.currentVersion) {
|
|
|
+ throw new Error(`format migration returned v${artifact.header.version}, expected v${format.currentVersion}`)
|
|
|
}
|
|
|
- const logical = encodeLogicalJsonl(migrated)
|
|
|
- const physical = await encodePhysicalJsonl(logical, migrated, compression)
|
|
|
- let staged = await writeSyncedTemp(currentPath, suffix, physical, internals)
|
|
|
+
|
|
|
+ await scheduler.yield()
|
|
|
+ signal?.throwIfAborted()
|
|
|
+ const sourceFingerprint = fingerprint(source.identity, source.bytes)
|
|
|
+ const eventCount = artifact.events.length
|
|
|
+ let staged = await writeSyncedTemp(currentPath, suffix, compression, artifact, format, signal, internals)
|
|
|
let failure: unknown
|
|
|
try {
|
|
|
- await validatePhysicalCurrent(staged, compression, format, signal, internals)
|
|
|
+ const verifiedStage = await options.verifyCurrentFile(
|
|
|
+ staged.path,
|
|
|
+ compression,
|
|
|
+ artifact.header.id,
|
|
|
+ eventCount,
|
|
|
+ undefined,
|
|
|
+ signal,
|
|
|
+ )
|
|
|
+ if (verifiedStage.bytes !== staged.bytes || verifiedStage.digest !== staged.digest) {
|
|
|
+ throw new Error('staged session generation changed during verification')
|
|
|
+ }
|
|
|
await internals.barrier('before-source-check', attempt)
|
|
|
const beforePublish = await readStableSnapshot(sourcePath, signal, internals.fs)
|
|
|
if (fingerprint(beforePublish.identity, beforePublish.bytes) !== sourceFingerprint) continue
|
|
|
|
|
|
- const published = await publishCurrentExclusive(staged, currentPath, internals)
|
|
|
- if (published && internals.platform === 'win32') staged = ''
|
|
|
+ const published = await publishCurrentExclusive(staged.path, currentPath, internals)
|
|
|
+ if (published && internals.platform === 'win32') staged = { ...staged, path: '' }
|
|
|
await internals.barrier('after-publication', attempt)
|
|
|
signal?.throwIfAborted()
|
|
|
const committed = await reopenExpectedCurrent(
|
|
|
currentPath,
|
|
|
- physical,
|
|
|
+ staged,
|
|
|
compression,
|
|
|
- format,
|
|
|
+ artifact.header.id,
|
|
|
+ eventCount,
|
|
|
+ options.verifyCurrentFile,
|
|
|
signal,
|
|
|
!published,
|
|
|
internals,
|
|
|
)
|
|
|
- if (staged !== '') {
|
|
|
- await removeCommittedTemporary(staged, internals)
|
|
|
- staged = ''
|
|
|
+ if (staged.path !== '') {
|
|
|
+ await removeCommittedTemporary(staged.path, internals)
|
|
|
+ staged = { ...staged, path: '' }
|
|
|
}
|
|
|
return {
|
|
|
status: 'migrated',
|
|
|
- fromVersion,
|
|
|
+ fromVersion: sourceVersion,
|
|
|
toVersion: format.currentVersion,
|
|
|
path: currentPath,
|
|
|
sourcePath,
|
|
|
@@ -752,32 +1058,38 @@ async function ensureCurrent(
|
|
|
failure = error
|
|
|
throw error
|
|
|
} finally {
|
|
|
- if (staged !== '') await removeTemporary(staged, failure, internals)
|
|
|
+ if (staged.path !== '') await removeTemporary(staged.path, failure, internals)
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * Ensure one resolved generation has a current-format successor. Current input reads one
|
|
|
- * coherent physical snapshot, inspects only its independently readable header,
|
|
|
- * invokes no body decoder or migration callback, and returns that snapshot for
|
|
|
- * the immediate body-reading backend hook. Historical input remains unchanged;
|
|
|
- * only a previously absent current filename can be published.
|
|
|
- * @param options - resolved source and target, configured encoding, format adapter, and cancellation.
|
|
|
- * @returns whether the source was already current or which immutable successor was published.
|
|
|
+ * Ensure one resolved generation has a current-format successor before returning.
|
|
|
+ * @param options - resolved source, current target, format adapter, verification, and cancellation.
|
|
|
+ * @returns the current source or the verified and reopened migrated successor.
|
|
|
*/
|
|
|
export function ensureJsonlGenerationCurrent(
|
|
|
options: EnsureJsonlGenerationOptions,
|
|
|
): Promise<EnsureJsonlGenerationResult> {
|
|
|
- return ensureCurrent(options, defaultInternals)
|
|
|
+ return defaultGenerationRuntime.ensure(options)
|
|
|
}
|
|
|
|
|
|
-/** Private deterministic filesystem, platform, and race seams for package tests. */
|
|
|
-export const __jsonlGenerationTest = {
|
|
|
- ensure(
|
|
|
- options: EnsureJsonlGenerationOptions,
|
|
|
- overrides: JsonlGenerationTestOverrides,
|
|
|
- ): Promise<EnsureJsonlGenerationResult> {
|
|
|
- return ensureCurrent(options, withOverrides(overrides))
|
|
|
- },
|
|
|
+/**
|
|
|
+ * Create one generation runtime with fixed filesystem and publication dependencies.
|
|
|
+ * @param overrides - deterministic filesystem, platform, and race dependencies.
|
|
|
+ * @returns bound generation operations.
|
|
|
+ */
|
|
|
+export function createJsonlGenerationRuntime(
|
|
|
+ overrides: JsonlGenerationRuntimeOverrides = {},
|
|
|
+): JsonlGenerationRuntime {
|
|
|
+ const internals = withOverrides(overrides)
|
|
|
+ return {
|
|
|
+ readStable: (path, signal) => readStableSnapshot(path, signal, internals.fs),
|
|
|
+ ensure: options => ensureCurrent(options, internals),
|
|
|
+ verify: (path, compression, expectedId, expectedEventCount, expectedPrefix) => verifyCurrentGeneration(
|
|
|
+ path, compression, expectedId, expectedEventCount, internals.fs, expectedPrefix,
|
|
|
+ ),
|
|
|
+ }
|
|
|
}
|
|
|
+
|
|
|
+const defaultGenerationRuntime = createJsonlGenerationRuntime()
|
imccyu