|
|
@@ -19,6 +19,7 @@ import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-se
|
|
|
import {
|
|
|
encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
|
|
|
} from './format.ts'
|
|
|
+import { ensureDurableDirectoryWin32, publishNewFileWin32 } from './win32.ts'
|
|
|
|
|
|
/** Plugin config: where the JSONL backend keeps its session logs (`root` is required — no default). */
|
|
|
export interface Config {
|
|
|
@@ -160,33 +161,35 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
|
|
|
|
|
// --- materialization / append / repair (file mechanics) ---
|
|
|
|
|
|
- /** Atomically write the header line + first batch (temp-write, fsync, collision-safe hard-link publish). */
|
|
|
+ /** Atomically write the header line + first batch (temp-write, fsync, publish). */
|
|
|
private async materialize(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
|
|
|
const dir = sessionDir(this.root, meta.cwd)
|
|
|
- await mkdir(this.root, { recursive: true, mode: 0o700 })
|
|
|
- await this.syncDir(dirname(this.root))
|
|
|
- await mkdir(dir, { recursive: true, mode: 0o700 })
|
|
|
- await this.syncDir(this.root)
|
|
|
const finalPath = logPath(this.root, meta.cwd, meta.id)
|
|
|
- // Materialization is the first write; an existing log is an id collision.
|
|
|
- /* v8 ignore next 3 -- createCore guards collisions before materialize; this is a TOCTOU backstop */
|
|
|
- if (await this.exists(finalPath)) {
|
|
|
- throw new Error(`refusing to materialize "${meta.id}": a log already exists on disk (load/resume it instead)`)
|
|
|
+ const content = this.initialLogContent(meta, events)
|
|
|
+ /* v8 ignore next -- native Windows coverage exercises this platform dispatch; Linux covers the POSIX peer */
|
|
|
+ if (process.platform === 'win32') {
|
|
|
+ await this.materializeWin32(dir, finalPath, meta.id, content)
|
|
|
+ } else {
|
|
|
+ await this.materializePosix(dir, finalPath, meta.id, content)
|
|
|
}
|
|
|
+ }
|
|
|
+
|
|
|
+ private initialLogContent(meta: SessionHeader, events: readonly SessionEvent[]): string {
|
|
|
const header = JSON.stringify(toHeaderLine(meta))
|
|
|
const body = events.map(eventLine).join('\n')
|
|
|
- const content = header + '\n' + body + '\n'
|
|
|
+ return header + '\n' + body + '\n'
|
|
|
+ }
|
|
|
|
|
|
- const tmp = `${finalPath}.${randomBytes(6).toString('hex')}.tmp`
|
|
|
- const handle = await open(tmp, 'wx', 0o600)
|
|
|
- try {
|
|
|
- await handle.writeFile(content)
|
|
|
- await handle.sync()
|
|
|
- } finally {
|
|
|
- await handle.close()
|
|
|
- }
|
|
|
- // Publish with link()+unlink(): unlike rename(), link fails if another
|
|
|
- // process materialized the same id first.
|
|
|
+ private async materializePosix(dir: string, finalPath: string, id: SessionId, content: string): Promise<void> {
|
|
|
+ await mkdir(this.root, { recursive: true, mode: 0o700 })
|
|
|
+ await this.syncDirPosix(dirname(this.root))
|
|
|
+ await mkdir(dir, { recursive: true, mode: 0o700 })
|
|
|
+ await this.syncDirPosix(this.root)
|
|
|
+ await this.rejectExistingLog(finalPath, id)
|
|
|
+ const tmp = await this.writeSyncedTempFile(finalPath, content)
|
|
|
+ // Publish via link()+unlink(), NOT rename(): link fails with EEXIST if the
|
|
|
+ // final path already exists, so two processes materializing the same id
|
|
|
+ // concurrently cannot clobber each other. rename() would silently overwrite.
|
|
|
let linked = false
|
|
|
try {
|
|
|
await link(tmp, finalPath)
|
|
|
@@ -197,10 +200,13 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
|
|
/* v8 ignore next -- link failure is the TOCTOU/IO race guarded above; not reachable in test */
|
|
|
if (!linked) await rm(tmp, { force: true })
|
|
|
}
|
|
|
- // The published link becomes crash-durable only after its directory fsync.
|
|
|
- await this.syncDir(dir)
|
|
|
- // Best-effort temp cleanup: the log is already published and durable, so a failure to
|
|
|
- // remove the (now-redundant) temp hard link must not reject the append.
|
|
|
+ // link() succeeded — the log is published. fsync the directory so the new
|
|
|
+ // entry survives a power loss: the new link is not crash-durable until the
|
|
|
+ // parent directory's metadata is synced.
|
|
|
+ await this.syncDirPosix(dir)
|
|
|
+ // Best-effort temp cleanup: the log is already published and durable, so a
|
|
|
+ // failure to remove the (now-redundant) temp hard link must NOT reject the
|
|
|
+ // append. Swallow only the rm failure; nothing else of consequence runs here.
|
|
|
try {
|
|
|
await rm(tmp, { force: true })
|
|
|
} catch {
|
|
|
@@ -208,8 +214,47 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- /** fsync a directory so a just-created or published entry inside it is crash-durable. */
|
|
|
- private async syncDir(dir: string): Promise<void> {
|
|
|
+ /* v8 ignore start -- native Windows coverage exercises this integration path */
|
|
|
+ private async materializeWin32(dir: string, finalPath: string, id: SessionId, content: string): Promise<void> {
|
|
|
+ await ensureDurableDirectoryWin32(this.root)
|
|
|
+ await ensureDurableDirectoryWin32(dir)
|
|
|
+ await this.rejectExistingLog(finalPath, id)
|
|
|
+ const tmp = await this.writeSyncedTempFile(finalPath, content)
|
|
|
+ try {
|
|
|
+ await publishNewFileWin32(tmp, finalPath)
|
|
|
+ } catch (error) {
|
|
|
+ await rm(tmp, { force: true })
|
|
|
+ throw error
|
|
|
+ }
|
|
|
+ }
|
|
|
+ /* v8 ignore stop */
|
|
|
+
|
|
|
+ private async rejectExistingLog(finalPath: string, id: SessionId): Promise<void> {
|
|
|
+ // Never publish over an existing committed log: materialize is the FIRST
|
|
|
+ // write of a session the backend believes is new. A file here means a
|
|
|
+ // different session shares this id on disk — reject loudly. (createCore
|
|
|
+ // already guards the create path, so this is unreachable-in-practice TOCTOU
|
|
|
+ // defense.)
|
|
|
+ /* v8 ignore next 3 -- createCore guards collisions before materialize; this is a TOCTOU backstop */
|
|
|
+ if (await this.exists(finalPath)) {
|
|
|
+ throw new Error(`refusing to materialize "${id}": a log already exists on disk (load/resume it instead)`)
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private async writeSyncedTempFile(finalPath: string, content: string): Promise<string> {
|
|
|
+ const tmp = `${finalPath}.${randomBytes(6).toString('hex')}.tmp`
|
|
|
+ const handle = await open(tmp, 'wx', 0o600)
|
|
|
+ try {
|
|
|
+ await handle.writeFile(content)
|
|
|
+ await handle.sync()
|
|
|
+ } finally {
|
|
|
+ await handle.close()
|
|
|
+ }
|
|
|
+ return tmp
|
|
|
+ }
|
|
|
+
|
|
|
+ /** fsync a POSIX directory so a just-created/renamed entry is crash-durable. */
|
|
|
+ private async syncDirPosix(dir: string): Promise<void> {
|
|
|
const handle = await open(dir, 'r')
|
|
|
try {
|
|
|
await handle.sync()
|
|
|
@@ -226,17 +271,37 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
|
|
private async appendLines(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
|
|
|
const path = logPath(this.root, meta.cwd, meta.id)
|
|
|
const handle = await open(path, 'a')
|
|
|
+ let closed = false
|
|
|
+ const closeAppendHandle = async (): Promise<void> => {
|
|
|
+ if (closed) return
|
|
|
+ closed = true
|
|
|
+ await handle.close()
|
|
|
+ }
|
|
|
+
|
|
|
try {
|
|
|
const { size: before } = await handle.stat()
|
|
|
try {
|
|
|
await handle.writeFile(events.map(eventLine).join('\n') + '\n')
|
|
|
await handle.sync()
|
|
|
} catch (error) {
|
|
|
- // Roll back whatever bytes landed so a retry starts from a clean EOF.
|
|
|
- await handle.truncate(before)
|
|
|
- await handle.sync()
|
|
|
+ try {
|
|
|
+ await closeAppendHandle()
|
|
|
+ await this.rollbackAppend(path, before)
|
|
|
+ } catch (rollbackError) {
|
|
|
+ throw new AggregateError([error, rollbackError], `failed to roll back append to "${path}"`)
|
|
|
+ }
|
|
|
throw error
|
|
|
}
|
|
|
+ } finally {
|
|
|
+ await closeAppendHandle()
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private async rollbackAppend(path: string, size: number): Promise<void> {
|
|
|
+ const handle = await open(path, 'r+')
|
|
|
+ try {
|
|
|
+ await handle.truncate(size)
|
|
|
+ await handle.sync()
|
|
|
} finally {
|
|
|
await handle.close()
|
|
|
}
|
|
|
@@ -322,9 +387,8 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
|
|
await handle.close()
|
|
|
return true
|
|
|
} catch (error) {
|
|
|
- // Only ENOENT means absent. A permission/I/O error must surface, not be
|
|
|
- // collapsed to `false` — otherwise load() reports "not found" and collision
|
|
|
- // checks proceed under a false absence assumption.
|
|
|
+ // Only ENOENT means absent. A permission/I/O error must surface rather
|
|
|
+ // than letting load or collision checks proceed under false absence.
|
|
|
if (isENOENT(error)) return false
|
|
|
throw error
|
|
|
}
|