fix(journal): keep the prefix of a crash-truncated journal and record again (#2166)

changelog.d/2166-journal-incomplete-last-line.fixed.md

@@ -0,0 +1 @@
+The state journal recovers a last line a crash left half written instead of refusing the file on every start and every save: when the complete prefix before it replays, the journal is replaced with that prefix, the dropped line is warned once, and recording continues from it, so a person no longer has to delete `journal/state.jsonl` ([#2166](https://github.com/ThinkFlowLab/FlyBot/issues/2166)). A broken line above the last one, a `seq` gap, a `prev` mismatch or an unknown op still refuses the whole journal and leaves it as it is.

core/src/host/journal.js

@@ -263,11 +263,34 @@ export function openJournal({ path, warn, now = Date.now, maxBytes = JOURNAL_MAX
     if (!info.isFile() || info.nlink !== 1) throw new JournalError('the journal path is not a plain file')
     await chmod(path, 0o600)
     const text = await readFile(path, 'utf8')
-    const entries = parseJournal(text)
+    // A crash between the bytes of two appends leaves the last line half written, and replay
+    // refuses a file that does not end on a line boundary. That refusal used to be permanent:
+    // every later start and every save failed until a person deleted the journal. Only that shape
+    // is repaired — the complete prefix must be a journal replay accepts — with the atomic
+    // replacement a rotation uses, so the kept lines still replay and recording continues from
+    // them. A broken line above the last one still refuses the whole file, and a half line on top
+    // of a prefix that does not replay is refused too.
+    let kept = text
+    let entries
+    if (text !== '' && !text.endsWith('\n')) {
+      kept = text.slice(0, text.lastIndexOf('\n') + 1)
+      entries = parseJournal(kept)
+      const temp = `${path}.${process.pid}.tmp`
+      try {
+        await writeFile(temp, kept, { mode: 0o600, flag: 'wx' })
+        await rename(temp, path)
+      } catch (error) {
+        await unlink(temp).catch(() => { /* the temp file may not exist */ })
+        throw error
+      }
+      warn('ds-bot: state journal %s', `${path} dropped an incomplete last line left by a crash`)
+    } else {
+      entries = parseJournal(text)
+    }
     if (entries.length === 0) return
     lastText = stateText(replayEntries(entries, Infinity))
     seq = entries.at(-1).seq
-    bytes = Buffer.byteLength(text)
+    bytes = Buffer.byteLength(kept)
     oldest = entries[1]?.t
   }
   const opened = open().catch(stop)

core/test/state-journal.test.js

@@ -165,7 +165,9 @@ test('the writer stops recording and leaves the file when its journal does not r
   const home = tempHome()
   const path = join(home, 'journal', 'state.jsonl')
   mkdirSync(join(home, 'journal'))
-  const broken = `${JSON.stringify({ seq: 1, t: 1, op: 'base', next: { version: 2 } })}\n{"seq":2,"t":`
+  // A complete line that does not parse is the journal refusing itself; a half-written last line
+  // is the crash shape the recovery keeps the prefix of (#2166), so this fixture stays whole.
+  const broken = `${JSON.stringify({ seq: 1, t: 1, op: 'base', next: { version: 2 } })}\n{"seq":2,"t":}\n`
   writeFileSync(path, broken)
   const warnings = []
   const journal = openJournal({ path, warn: (...args) => warnings.push(args.join(' ')) })
@@ -668,3 +670,51 @@ test('a journal reopened after its file was removed starts with a base (#2154)',
   rmSync(home, { recursive: true, force: true })
   rmSync(outside, { recursive: true, force: true })
 })
+
+// A crash between the bytes of two appends leaves the last line half written, and replay refuses
+// a file that does not end on a line boundary. That refusal was permanent: every start and every
+// save failed until a person deleted the journal. Only that shape is repaired — the complete prefix
+// must be a journal replay accepts — and recording continues from the kept lines.
+test('a journal a crash left with half a last line keeps its complete prefix and records again (#2166)', async () => {
+  const home = tempHome()
+  mkdirSync(join(home, 'journal'))
+  const path = join(home, 'journal', 'state.jsonl')
+  const base = { seq: 1, t: 1, op: 'base', next: { version: 2, revision: 1 }, actor: 'host' }
+  const set = { seq: 2, t: 2, op: 'set', key: ['revision'], prev: 1, next: 2, actor: 'host' }
+  writeFileSync(path, `${lines([base, set])}{"seq":3,"t":3,"op":"del`)
+  const warnings = []
+  // A clock near the fixture's times: an aged journal would be pruned on the first write, which
+  // would fold the kept prefix into a base line and make the seq assertions say nothing.
+  const journal = openJournal({ path, warn: (...args) => warnings.push(args.join(' ')), now: () => 3 })
+  await journal.record(stateText({ version: 2, revision: 3 }))
+  assert.deepEqual(journal.status(), { recording: true })
+  const text = readFileSync(path, 'utf8')
+  assert.ok(text.endsWith('\n'), 'the repaired file ends on a line boundary')
+  assert.equal(text.includes('{"seq":3,"t":3,"op":"del'), false, 'the half line is gone')
+  const entries = parseJournal(text)
+  assert.deepEqual(entries.map(entry => entry.seq), [1, 2, 3], 'the seq continues from the kept prefix')
+  assert.equal(stateText(replayTo(text)), stateText({ version: 2, revision: 3 }))
+  assert.equal(warnings.length, 1, 'the dropped half line is announced once')
+  assert.match(warnings[0], /dropped an incomplete last line left by a crash/)
+  if (posix) assert.equal(statSync(path).mode & 0o777, 0o600)
+  rmSync(home, { recursive: true, force: true })
+})
+
+// The recovery is only for a half-written last line: a complete line that does not parse above it
+// is still the journal refusing itself, and the file is left exactly as it was.
+test('a journal with a broken line above the last one stays refused and is left as it was (#2166)', async () => {
+  const home = tempHome()
+  mkdirSync(join(home, 'journal'))
+  const path = join(home, 'journal', 'state.jsonl')
+  const base = { seq: 1, t: 1, op: 'base', next: { version: 2, revision: 1 }, actor: 'host' }
+  const broken = `${lines([base])}{"seq":2,"t":2,"op":"se\n`
+  writeFileSync(path, broken)
+  const warnings = []
+  const journal = openJournal({ path, warn: (...args) => warnings.push(args.join(' ')) })
+  await journal.record(stateText({ version: 2, revision: 2 }))
+  assert.equal(journal.status().recording, false)
+  assert.equal(readFileSync(path, 'utf8'), broken, 'a broken line above the last one is never trimmed away')
+  assert.equal(warnings.length, 1)
+  assert.match(warnings[0], /not recording/)
+  rmSync(home, { recursive: true, force: true })
+})

docs/subsystems/core-host.md

@@ -1,3 +1,3 @@
 ---
 subsystem: core-host
 owned_paths:
@@ -557,7 +557,7 @@
 - **On.** `stateJournal: true` in the `bot` config. Off (the default) nothing is opened or written, so every other test runs as before.
 - **What.** `src/host/journal.js`. After each successful `state.json` write, `store.js` hands the exact snapshot text to the journal, which appends the difference from the last recorded snapshot to `<home>/journal/state.jsonl` (file mode 0600, folder 0700). Lines are `{ seq, t, op, key, prev, next, botId?, actor }`; `op` is `base` (the whole team: the first line, and any change only a key reorder explains), `set`, `delete`, or `append` (`prevLength`, `drop`, the new items: a group log or other list that grew or slid). Paths go at most three keys deep (`bots.<id>.<field>`); `botId` is set on a `bots.<id>` path; `actor` is `host` for a save and `retention` for a folded base (per-call attribution is not recorded). Only the host that holds the state lock writes, so a read-only window records nothing.
 - **Retention.** When a write would pass 20 MB, or the oldest change is older than 7 days, the oldest lines are dropped down to 90% of each cap and folded into the base line (atomic rewrite), so the kept lines still replay. A team that alone exceeds the cap stops recording.
-- **Fail closed.** Replay (`replayTo(text, seq)`) refuses the whole journal on an unreadable or incomplete line, an unknown op, a gap in `seq`, a first line that is not a base, or a `prev` that does not match what the earlier lines built; a `seq` outside the file is refused, not approximated. On start the writer replays the existing file the same way; a journal that does not replay, a path that is not a plain file, or a failed write stops recording with a warning and leaves the file as it is. A writer that stopped is asked once more on the next save: it runs the start-up `open()` again, so a folder or file repaired meanwhile resumes recording — appending the change that save carried, or starting a new file with a base when the journal was removed — while a journal that still refuses stays stopped and that save still lands; the asking is silent, so a refused start-up or write is warned once, not once per save ([#2154](https://github.com/ThinkFlowLab/FlyBot/issues/2154)). Saving the team never depends on the journal. The folder above the journal must be a real directory too (#2056): a link, a plain file or a check that cannot run refuses before the `mkdir`/`chmod` that would follow it, so neither the permissions nor a write reach the folder a link names; a folder that does not exist yet is still this writer's to create, as before. Since [#2074](https://github.com/ThinkFlowLab/FlyBot/issues/2074) the same question is asked again when the write runs: a journal folder swapped for a link is refused with one warning and recording stops, and a `state.jsonl` swapped for a link is never written through. A hard link reads as a plain file to `lstat` while its bytes are also a name outside this home, so the start-up read, the append and — since [#2146](https://github.com/ThinkFlowLab/FlyBot/issues/2146) — the rotation also require the file to carry only this one name (`nlink === 1`): a shared inode is refused exactly like a link — one warning, nothing read, chmod-ed or appended — and a journal with one name records as before. The append path refuses the write, and so does the rotation, before it reads: a link or a second name there stops recording with one warning, so another file's lines are never read as the team's history and the `rename` never drops the other name. A writer that stopped says so off the host too: `rt.journal.status()` answers `{ recording, because }`, and the diagnostics report adds one line — `state journal: recording`, `state journal: stopped (<reason>)`, or `state journal: paused (this window is read-only)` in a window another dsh keeps read-only, which never reaches the write that records ([#2153](https://github.com/ThinkFlowLab/FlyBot/issues/2153)) — while a host without the journal adds none ([#2134](https://github.com/ThinkFlowLab/FlyBot/issues/2134)).
+- **Fail closed.** Replay (`replayTo(text, seq)`) refuses the whole journal on an unreadable or incomplete line, an unknown op, a gap in `seq`, a first line that is not a base, or a `prev` that does not match what the earlier lines built; a `seq` outside the file is refused, not approximated. On start the writer replays the existing file the same way; a journal that does not replay, a path that is not a plain file, or a failed write stops recording with a warning and leaves the file as it is. A last line left half written by a crash is the one shape the writer repairs instead of refusing: when the complete prefix before it replays, the file is replaced with that prefix (the same atomic temp-and-rename as a rotation) with one warning, `dropped an incomplete last line left by a crash`, and recording continues from it ([#2166](https://github.com/ThinkFlowLab/FlyBot/issues/2166)) — a broken line above the last one, a `seq` gap, a `prev` mismatch or an unknown op still stops recording and leaves the file as it is. A writer that stopped is asked once more on the next save: it runs the start-up `open()` again, so a folder or file repaired meanwhile resumes recording — appending the change that save carried, or starting a new file with a base when the journal was removed — while a journal that still refuses stays stopped and that save still lands; the asking is silent, so a refused start-up or write is warned once, not once per save ([#2154](https://github.com/ThinkFlowLab/FlyBot/issues/2154)). Saving the team never depends on the journal. The folder above the journal must be a real directory too (#2056): a link, a plain file or a check that cannot run refuses before the `mkdir`/`chmod` that would follow it, so neither the permissions nor a write reach the folder a link names; a folder that does not exist yet is still this writer's to create, as before. Since [#2074](https://github.com/ThinkFlowLab/FlyBot/issues/2074) the same question is asked again when the write runs: a journal folder swapped for a link is refused with one warning and recording stops, and a `state.jsonl` swapped for a link is never written through. A hard link reads as a plain file to `lstat` while its bytes are also a name outside this home, so the start-up read, the append and — since [#2146](https://github.com/ThinkFlowLab/FlyBot/issues/2146) — the rotation also require the file to carry only this one name (`nlink === 1`): a shared inode is refused exactly like a link — one warning, nothing read, chmod-ed or appended — and a journal with one name records as before. The append path refuses the write, and so does the rotation, before it reads: a link or a second name there stops recording with one warning, so another file's lines are never read as the team's history and the `rename` never drops the other name. A writer that stopped says so off the host too: `rt.journal.status()` answers `{ recording, because }`, and the diagnostics report adds one line — `state journal: recording`, `state journal: stopped (<reason>)`, or `state journal: paused (this window is read-only)` in a window another dsh keeps read-only, which never reaches the write that records ([#2153](https://github.com/ThinkFlowLab/FlyBot/issues/2153)) — while a host without the journal adds none ([#2134](https://github.com/ThinkFlowLab/FlyBot/issues/2134)).
 - **Replay.** `core/scripts/replay.mjs` (dev-only, not in the package, see [install](install.md)) prints or writes (`--out`, new file only, 0600) the team after a given `--to <seq>`; it makes no network or process calls. Test: `core/test/state-journal.test.js` (a 3-Bot round replays byte-identical at every step).
 
 ### Standing tool approvals (#540)

patches/ds-bot/files/src/host/journal.js

@@ -263,11 +263,34 @@ export function openJournal({ path, warn, now = Date.now, maxBytes = JOURNAL_MAX
     if (!info.isFile() || info.nlink !== 1) throw new JournalError('the journal path is not a plain file')
     await chmod(path, 0o600)
     const text = await readFile(path, 'utf8')
-    const entries = parseJournal(text)
+    // A crash between the bytes of two appends leaves the last line half written, and replay
+    // refuses a file that does not end on a line boundary. That refusal used to be permanent:
+    // every later start and every save failed until a person deleted the journal. Only that shape
+    // is repaired — the complete prefix must be a journal replay accepts — with the atomic
+    // replacement a rotation uses, so the kept lines still replay and recording continues from
+    // them. A broken line above the last one still refuses the whole file, and a half line on top
+    // of a prefix that does not replay is refused too.
+    let kept = text
+    let entries
+    if (text !== '' && !text.endsWith('\n')) {
+      kept = text.slice(0, text.lastIndexOf('\n') + 1)
+      entries = parseJournal(kept)
+      const temp = `${path}.${process.pid}.tmp`
+      try {
+        await writeFile(temp, kept, { mode: 0o600, flag: 'wx' })
+        await rename(temp, path)
+      } catch (error) {
+        await unlink(temp).catch(() => { /* the temp file may not exist */ })
+        throw error
+      }
+      warn('ds-bot: state journal %s', `${path} dropped an incomplete last line left by a crash`)
+    } else {
+      entries = parseJournal(text)
+    }
     if (entries.length === 0) return
     lastText = stateText(replayEntries(entries, Infinity))
     seq = entries.at(-1).seq
-    bytes = Buffer.byteLength(text)
+    bytes = Buffer.byteLength(kept)
     oldest = entries[1]?.t
   }
   const opened = open().catch(stop)

patches/ds-bot/files/test/state-journal.test.js

@@ -165,7 +165,9 @@ test('the writer stops recording and leaves the file when its journal does not r
   const home = tempHome()
   const path = join(home, 'journal', 'state.jsonl')
   mkdirSync(join(home, 'journal'))
-  const broken = `${JSON.stringify({ seq: 1, t: 1, op: 'base', next: { version: 2 } })}\n{"seq":2,"t":`
+  // A complete line that does not parse is the journal refusing itself; a half-written last line
+  // is the crash shape the recovery keeps the prefix of (#2166), so this fixture stays whole.
+  const broken = `${JSON.stringify({ seq: 1, t: 1, op: 'base', next: { version: 2 } })}\n{"seq":2,"t":}\n`
   writeFileSync(path, broken)
   const warnings = []
   const journal = openJournal({ path, warn: (...args) => warnings.push(args.join(' ')) })
@@ -668,3 +670,51 @@ test('a journal reopened after its file was removed starts with a base (#2154)',
   rmSync(home, { recursive: true, force: true })
   rmSync(outside, { recursive: true, force: true })
 })
+
+// A crash between the bytes of two appends leaves the last line half written, and replay refuses
+// a file that does not end on a line boundary. That refusal was permanent: every start and every
+// save failed until a person deleted the journal. Only that shape is repaired — the complete prefix
+// must be a journal replay accepts — and recording continues from the kept lines.
+test('a journal a crash left with half a last line keeps its complete prefix and records again (#2166)', async () => {
+  const home = tempHome()
+  mkdirSync(join(home, 'journal'))
+  const path = join(home, 'journal', 'state.jsonl')
+  const base = { seq: 1, t: 1, op: 'base', next: { version: 2, revision: 1 }, actor: 'host' }
+  const set = { seq: 2, t: 2, op: 'set', key: ['revision'], prev: 1, next: 2, actor: 'host' }
+  writeFileSync(path, `${lines([base, set])}{"seq":3,"t":3,"op":"del`)
+  const warnings = []
+  // A clock near the fixture's times: an aged journal would be pruned on the first write, which
+  // would fold the kept prefix into a base line and make the seq assertions say nothing.
+  const journal = openJournal({ path, warn: (...args) => warnings.push(args.join(' ')), now: () => 3 })
+  await journal.record(stateText({ version: 2, revision: 3 }))
+  assert.deepEqual(journal.status(), { recording: true })
+  const text = readFileSync(path, 'utf8')
+  assert.ok(text.endsWith('\n'), 'the repaired file ends on a line boundary')
+  assert.equal(text.includes('{"seq":3,"t":3,"op":"del'), false, 'the half line is gone')
+  const entries = parseJournal(text)
+  assert.deepEqual(entries.map(entry => entry.seq), [1, 2, 3], 'the seq continues from the kept prefix')
+  assert.equal(stateText(replayTo(text)), stateText({ version: 2, revision: 3 }))
+  assert.equal(warnings.length, 1, 'the dropped half line is announced once')
+  assert.match(warnings[0], /dropped an incomplete last line left by a crash/)
+  if (posix) assert.equal(statSync(path).mode & 0o777, 0o600)
+  rmSync(home, { recursive: true, force: true })
+})
+
+// The recovery is only for a half-written last line: a complete line that does not parse above it
+// is still the journal refusing itself, and the file is left exactly as it was.
+test('a journal with a broken line above the last one stays refused and is left as it was (#2166)', async () => {
+  const home = tempHome()
+  mkdirSync(join(home, 'journal'))
+  const path = join(home, 'journal', 'state.jsonl')
+  const base = { seq: 1, t: 1, op: 'base', next: { version: 2, revision: 1 }, actor: 'host' }
+  const broken = `${lines([base])}{"seq":2,"t":2,"op":"se\n`
+  writeFileSync(path, broken)
+  const warnings = []
+  const journal = openJournal({ path, warn: (...args) => warnings.push(args.join(' ')) })
+  await journal.record(stateText({ version: 2, revision: 2 }))
+  assert.equal(journal.status().recording, false)
+  assert.equal(readFileSync(path, 'utf8'), broken, 'a broken line above the last one is never trimmed away')
+  assert.equal(warnings.length, 1)
+  assert.match(warnings[0], /not recording/)
+  rmSync(home, { recursive: true, force: true })
+})