fix(resume): write inflight.json atomically

changelog.d/2705-inflight-atomic.fixed.md

@@ -0,0 +1 @@
+`inflight.json` — the record that lets an interrupted turn resume — is now staged in a temp file and renamed over the old one, so a crash in the middle of the write leaves the previous complete notes instead of an empty or half-written file. Every turn and queued message the crash cut off is still picked up at the next start, where an empty file used to drop them all silently. A file that still cannot be parsed (a hand-edit, a disk error) is now reported once with its path and the reason instead of passing for "nothing was in flight", and a temp name a crashed write left behind is swept by the next host that writes. See [#2705](https://github.com/ThinkFlowLab/FlyBot/issues/2705).

core/src/host/resume.js

@@ -5,10 +5,12 @@
 // running Bot turn and each message it delivers until a turn takes it up, in
 // inflight.json; what is still noted at the next start is picked up once.
 import { randomUUID } from 'node:crypto'
-import { readFileSync } from 'node:fs'
-import { join } from 'node:path'
+import { readFileSync, readdirSync, unlinkSync } from 'node:fs'
+import { unlink } from 'node:fs/promises'
+import { basename, join } from 'node:path'
 import { SOURCE_KIND } from './constants.js'
-import { ensurePrivateDir, writePrivateFile } from './private-files.js'
+import { ensurePrivateDir, replacePrivateFile } from './private-files.js'
+import { tempWritable } from './real-dir.js'
 import { OTHER, settled } from './turn-origin.js'
 import { retiredBot } from './store.js'
 
@@ -36,12 +38,25 @@ const triggerOf = info => ({
   ...(info.senderRoomId ? { senderRoomId: info.senderRoomId } : {}),
 })
 
-const readNotes = (file) => {
+// A file that is not the notes cannot be resumed safely, but that is not the same answer as
+// "nothing was in flight": every turn it named is dropped, so it is named once — path and reason —
+// instead of passing for a clean start (#2705).
+const readNotes = (file, warn) => {
+  let text
   try {
-    const parsed = JSON.parse(readFileSync(file, 'utf8'))
+    text = readFileSync(file, 'utf8')
+  } catch (error) {
+    // A missing file is the ordinary case: nothing was noted, or the last write cleared it.
+    if (error?.code !== 'ENOENT') warn('ds-bot: reading %s failed (%s); nothing is picked up from it', file, error?.message ?? error)
+    return { turns: {}, queued: {} }
+  }
+  try {
+    const parsed = JSON.parse(text)
+    if (parsed === null || typeof parsed !== 'object' || Array.isArray(parsed)) throw new Error('not the object of turns and queued messages')
     const object = value => (value !== null && typeof value === 'object' && !Array.isArray(value) ? value : {})
-    return { turns: object(parsed?.turns), queued: object(parsed?.queued) }
-  } catch {
+    return { turns: object(parsed.turns), queued: object(parsed.queued) }
+  } catch (error) {
+    warn('ds-bot: %s could not be read as the notes of a running turn (%s); the turns and messages it named are not picked up', file, error?.message ?? error)
     return { turns: {}, queued: {} }
   }
 }
@@ -51,7 +66,7 @@ export function install(rt) {
   const file = join(home, 'inflight.json')
 
   // What the last run left; read before this run notes anything.
-  let interrupted = readNotes(file)
+  let interrupted = readNotes(file, rt.warn)
 
   /** @type {Record<string, {botId:string, roomId?:string, since:number, origin?:string, trigger?:any, approval?:{tool:string, reason:string}, resumed?:boolean}>} */
   let turns = {}
@@ -61,6 +76,20 @@ export function install(rt) {
   let stopped = false
   let latest
   let due = false
+  // A write this host never finished left its temp name behind. It holds no state — the target it
+  // was going to replace is still the last complete notes — so the first write of the next host
+  // sweeps it. Only a writing host does: a read-only window leaves the home alone.
+  let swept = false
+  const sweepTemps = () => {
+    swept = true
+    const prefix = `${basename(file)}.`
+    let names = []
+    try { names = readdirSync(home) } catch { return }
+    for (const name of names) {
+      if (!name.startsWith(prefix) || !name.endsWith('.tmp')) continue
+      try { unlinkSync(join(home, name)) } catch { /* another host may be writing it */ }
+    }
+  }
   const persist = () => {
     // Only the host that saves the team runs its work again.
     if (stopped || rt.readOnly() !== undefined) return
@@ -69,15 +98,26 @@ export function install(rt) {
     // write the file once per change.
     if (due) return
     due = true
-    // Written in place: a rename over the file fails on Windows while anything reads
-    // it, and a torn write only loses what one restart would pick up. The notes name the
-    // Session, the room and the message a turn was handed, so the file is private like the
-    // team file: created 0600 in a 0700 home, and an existing file keeps the mode it has (#2697).
+    // Renamed into place, never written through: the file is the only record of the turns a crash
+    // cut off, and a crash inside an in-place write left it empty or half written — every one of
+    // them dropped on the next start (#2705). The notes name the Session, the room and the message
+    // a turn was handed, so the temp name is private like the team file: created 0600 in a 0700
+    // home, at the mode an existing file already has (#2697).
     writing = writing.then(async () => {
       due = false
       const snapshot = latest
       await ensurePrivateDir(home)
-      await writePrivateFile(file, snapshot)
+      if (!swept) sweepTemps()
+      const temp = `${file}.${process.pid}.${randomUUID()}.tmp`
+      // `writeFile` follows a link at the temp name, which would take the notes outside the home
+      // and the rename after them would put the link where the notes belong (temp-name sweep).
+      if (!tempWritable(temp)) throw new Error(`${temp} is not a plain file of its own, so the notes were not written there`)
+      try {
+        await replacePrivateFile(temp, file, snapshot)
+      } catch (error) {
+        await unlink(temp).catch(() => { /* nothing was staged, or the rename already took it */ })
+        throw error
+      }
     }).catch(error => rt.warn('ds-bot: noting running turns failed: %s', error))
   }
   const originNote = (id) => {

core/test/resume-atomic.test.js

@@ -0,0 +1,200 @@
+// `inflight.json` is the one record of the turns a crash cut off, so the write that keeps it must
+// never leave the file empty or half written: the writer stages a temp name beside it and renames
+// it over, so a crash at any point leaves either the previous complete notes or the new complete
+// notes (#2705). These cases run real child processes and a real `kill -9` in a temp home, so the
+// window is the one the shipped code has, and a file nothing can parse is still reported instead of
+// passing for "nothing was in flight".
+import assert from 'node:assert/strict'
+import { spawn } from 'node:child_process'
+import { existsSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { test } from 'node:test'
+import { fileURLToPath } from 'node:url'
+
+const RESUME = fileURLToPath(new URL('../src/host/resume.js', import.meta.url))
+// The core modules resolve `@deepseek-ai/*` through this hook; the child processes need it too.
+const HARNESS = fileURLToPath(new URL('./harness-resolve.mjs', import.meta.url))
+
+const NOTE = {
+  turns: { 'sess-cut-off': { botId: 'bot-1', since: 1, turn: 3, trigger: { role: 'request', exchangeId: 'ex-1', senderSessionId: 'sess-peer' } } },
+  queued: { 'sess-peer': [{ at: 2, args: { toId: 'bot-1', sessionId: 'sess-cut-off', exchangeId: 'ex-2', text: 'answer me' } }] },
+}
+const NOTE_TEXT = JSON.stringify(NOTE, null, 2)
+
+const children = []
+const start = (code, args = []) => {
+  const child = spawn(process.execPath, ['--import', HARNESS, '--input-type=module', '-e', code, ...args], { stdio: ['ignore', 'pipe', 'pipe'] })
+  children.push(child)
+  let output = ''
+  let errors = ''
+  child.stdout.setEncoding('utf8')
+  child.stderr.setEncoding('utf8')
+  child.stdout.on('data', chunk => { output += chunk })
+  child.stderr.on('data', chunk => { errors += chunk })
+  const exited = new Promise(resolve => child.on('exit', (code_, signal) => resolve({ code: code_, signal })))
+  return { child, exited, get output() { return output }, get errors() { return errors } }
+}
+// Waits for the child's last printed line, which carries its answer as JSON.
+const answer = async (run, tag, timeoutMs = 30_000) => {
+  const deadline = Date.now() + timeoutMs
+  while (!new RegExp(`^@@${tag} `, 'm').test(run.output) && Date.now() < deadline) await new Promise(resolve => setTimeout(resolve, 5))
+  const matched = run.output.match(new RegExp(`^@@${tag} (.*)$`, 'm'))
+  assert.ok(matched, `the child never answered ${tag}\nstdout: ${run.output}\nstderr: ${run.errors}`)
+  return JSON.parse(matched[1])
+}
+// A real install of the module under test: the pickup timer is captured, not run, so the case can
+// read what the next start would schedule without the pickup itself racing the assertions.
+const STUB = `
+const home = process.argv[1]
+const timers = []
+const warns = []
+globalThis.setTimeout = (fn, ms) => { timers.push(ms); return { unref() {} } }
+const effects = []
+const ctx = { on() {}, effect(fn) { effects.push(fn) }, sessionController: undefined }
+const rt = {
+  ctx, home, load: async () => {}, readOnly: () => undefined, deliver: async () => {},
+  chatOf: id => id, botOf: () => undefined, isBotSession: () => false,
+  warn: (...args) => warns.push(args.map(value => String(value)).join(' ')),
+}
+`
+const READER = `
+import { install as installResume } from ${JSON.stringify(RESUME)}
+${STUB}
+installResume(rt)
+for (const fn of effects) fn()
+process.stdout.write('@@read ' + JSON.stringify({ timers, warns }) + '\\n')
+`
+const readNextStart = async (home) => {
+  const run = start(READER, [home])
+  const seen = await answer(run, 'read')
+  await run.exited
+  return seen
+}
+// A real writer: it notes one delivery and then sits there, so the parent can kill it inside the
+// window a real crash would land in.
+const WRITER = `
+import { install as installResume } from ${JSON.stringify(RESUME)}
+${STUB}
+installResume(rt)
+for (const fn of effects) fn()
+process.stdout.write('@@ready ' + JSON.stringify({}) + '\\n')
+// A real handle: the stubbed timer above would otherwise let the loop empty and the child exit
+// before the write it is here for.
+setInterval(() => {}, 1000)
+await rt.deliver({ toId: 'bot-1', text: 'x'.repeat(40 * 1024 * 1024) })
+await new Promise(() => {})
+`
+const tempNames = home => readdirSync(home).filter(name => name.startsWith('inflight.json.') && name.endsWith('.tmp'))
+// The window a crash has to land in: the write is under way, so the in-place writer has already
+// shortened the target and the atomic one has a temp name growing beside it.
+const cutWindow = (home, previous) => {
+  for (const name of tempNames(home)) {
+    const size = statSync(join(home, name)).size
+    if (size > 0) return { temp: name, size }
+  }
+  const target = join(home, 'inflight.json')
+  if (!existsSync(target)) return { target: 'missing' }
+  const size = statSync(target).size
+  return size < previous ? { target: size } : undefined
+}
+const scratch = (t) => {
+  const home = mkdtempSync(join(tmpdir(), 'ds-bot-resume-atomic-'))
+  t.after(() => rmSync(home, { recursive: true, force: true, maxRetries: 5 }))
+  return home
+}
+
+test('a write cut off by kill -9 keeps the complete notes on disk, and the next start picks them up', { timeout: 60_000 }, async (t) => {
+  const home = scratch(t)
+  writeFileSync(join(home, 'inflight.json'), NOTE_TEXT, { mode: 0o600 })
+
+  const run = start(WRITER, [home])
+  await answer(run, 'ready')
+  let window
+  const deadline = Date.now() + 30_000
+  while (window === undefined && Date.now() < deadline) {
+    window = cutWindow(home, NOTE_TEXT.length)
+    if (window === undefined) await new Promise(resolve => setTimeout(resolve, 1))
+  }
+  assert.ok(window, 'the writer never showed a write in flight')
+  run.child.kill('SIGKILL')
+  const { signal } = await run.exited
+  assert.equal(signal, 'SIGKILL')
+  t.diagnostic(`cut the write inside the window: ${JSON.stringify(window)}`)
+
+  // Whatever the crash landed on, the target itself is a complete record: a torn write in a temp
+  // name that was never renamed over it cannot shorten it.
+  const bytes = readFileSync(join(home, 'inflight.json'), 'utf8')
+  assert.equal(bytes.length, NOTE_TEXT.length, 'the killed write left the previous notes, not a half-written snapshot')
+  let onDisk
+  assert.doesNotThrow(() => { onDisk = JSON.parse(bytes) }, 'the killed write left a complete inflight.json behind')
+  assert.deepEqual(onDisk, NOTE, 'the notes on disk are the last complete ones')
+  const next = await readNextStart(home)
+  assert.deepEqual(next.timers, [3000], 'the next start schedules the pickup of the cut-off turn')
+  assert.deepEqual(next.warns, [], 'a complete file says nothing')
+})
+
+test('a file nothing can parse is reported once instead of passing for "nothing was in flight"', { timeout: 30_000 }, async (t) => {
+  const home = scratch(t)
+  const file = join(home, 'inflight.json')
+
+  writeFileSync(file, '', { mode: 0o600 })
+  const empty = await readNextStart(home)
+  assert.deepEqual(empty.timers, [], 'nothing is picked up from a file with no notes in it')
+  assert.equal(empty.warns.length, 1, 'the empty file is named once')
+  assert.match(empty.warns[0], /inflight\.json/)
+  assert.match(empty.warns[0], /not picked up|nothing is picked up|could not be read/i)
+
+  writeFileSync(file, NOTE_TEXT.slice(0, Math.floor(NOTE_TEXT.length / 2)), { mode: 0o600 })
+  const half = await readNextStart(home)
+  assert.deepEqual(half.timers, [])
+  assert.equal(half.warns.length, 1, 'the half-written file is named once')
+
+  // A file that parses but is not the notes shape cannot be resumed either, and says so.
+  writeFileSync(file, '[]', { mode: 0o600 })
+  const wrong = await readNextStart(home)
+  assert.deepEqual(wrong.timers, [])
+  assert.equal(wrong.warns.length, 1)
+
+  // A missing file is the ordinary case: no pickup, and nothing to report.
+  rmSync(file)
+  const absent = await readNextStart(home)
+  assert.deepEqual(absent, { timers: [], warns: [] })
+})
+
+test('a temp name a crash left behind is not read as state, and the next write clears it', { timeout: 30_000 }, async (t) => {
+  const home = scratch(t)
+  const stale = join(home, 'inflight.json.999999.deadbeef.tmp')
+  writeFileSync(stale, '{"turns":{},"queued":{}}', { mode: 0o600 })
+
+  const run = start(`
+    import { existsSync, readFileSync } from 'node:fs'
+    import { install as installResume } from ${JSON.stringify(RESUME)}
+    const home = process.argv[1]
+    const effects = []
+    const ctx = { on() {}, effect(fn) { effects.push(fn) }, sessionController: undefined }
+    const rt = {
+      ctx, home, load: async () => {}, readOnly: () => undefined, deliver: async () => {},
+      chatOf: id => id, botOf: () => undefined, isBotSession: () => false,
+      warn: message => process.stderr.write('warn: ' + String(message) + '\\n'),
+    }
+    installResume(rt)
+    for (const fn of effects) fn()
+    await rt.deliver({ toId: 'bot-1', text: 'after the crash' })
+    const notes = home + '/inflight.json'
+    for (let i = 0; i < 2000; i += 1) {
+      await new Promise(resolve => setImmediate(resolve))
+      try { if (JSON.parse(readFileSync(notes, 'utf8')).queued !== undefined) break } catch {}
+    }
+    process.stdout.write('@@wrote ' + JSON.stringify({ exists: existsSync(notes) }) + '\\n')
+  `, [home])
+  const wrote = await answer(run, 'wrote')
+  await run.exited
+  assert.equal(wrote.exists, true)
+  assert.equal(existsSync(stale), false, 'the crashed write\'s temp name is not left in the home')
+  assert.match(readFileSync(join(home, 'inflight.json'), 'utf8'), /after the crash/)
+})
+
+test.after(() => {
+  for (const child of children) if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL')
+})