fix(workflow): report an unreadable store instead of answering the fixture

changelog.d/2603.fixed.md

@@ -0,0 +1 @@
+A workflow store that cannot be read is no longer answered as a clean list: `workflow/state` now reports `store-unreadable` with the file and the reason instead of the factory fixture, so the Workflows page says the store is unreadable instead of showing workflows that are not yours, and the first save keeps the unreadable bytes in `workflows.json.broken-<stamp>` before it writes a fresh store ([#2603](https://github.com/ThinkFlowLab/FlyBot/issues/2603)).

plugins/workflow/src/service.js

@@ -3,7 +3,7 @@
 // (default 2). A live node stays running until its Bot replies.
 
 import { randomBytes } from 'node:crypto'
-import { lstatSync, mkdirSync, readFileSync, renameSync, writeFileSync } from 'node:fs'
+import { copyFileSync, lstatSync, mkdirSync, readFileSync, renameSync, writeFileSync } from 'node:fs'
 import { join } from 'node:path'
 import { releaseAll, spawnAll } from './ephemeral.js'
 import {
@@ -218,7 +218,12 @@
   setTimer = (fn, ms) => setTimeout(fn, ms), clearTimer = handle => clearTimeout(handle),
 } = {}) {
   const file = dir ? join(dir, FILE) : null
-  let state = load(file, warn) ?? seed()
+  const loaded = load(file, warn)
+  let state = loaded.state ?? seed()
+  // #2603: a store this service could not read stays a failure instead of turning
+  // into a clean list. `state()` answers it and the first `write()` keeps the bytes
+  // before replacing them; every other endpoint keeps running on the fixture.
+  let loadError = loaded.error
   state.routes ??= {}
   // migrateRuns() also normalizes the orphan list a deleted workflow left
   // (#1621), so a store written by an older version keeps those ids.
@@ -255,13 +260,13 @@
   }
 
   function clearRunTimers(runId) {
     for (const [key, entry] of [...timers]) {
       if (entry.runId === runId) clearTimerAt(key)
     }
     for (const key of [...warned]) {
       if (key.endsWith(`:${runId}`)) warned.delete(key)
     }
     for (const key of [...handOffTries.keys()]) {
       if (key.startsWith(`${runId}:`)) handOffTries.delete(key)
     }
   }
@@ -298,7 +303,7 @@
     // keep its deadline: without it the run holds its lane slot with nothing left
     // to free it until the owner presses Stop again (P-560).
     const stopRetry = run.status === 'running' && run.paused === true && run.releasePending === true
     for (const [key, entry] of [...timers]) {
       if (entry.runId !== run.id) continue
       if (run.status !== 'running' || (run.paused === true && !stopRetry) || !running.has(entry.nodeId)) clearTimerAt(key)
     }
@@ -372,12 +377,27 @@
     }).catch(error => warn(`flybot-workflow: step ${nodeId} could not be timed out: ${error?.message ?? error}`))
   }
 
+  // The bytes of a store this service could not read are kept before the first write
+  // replaces them, the way the core store takes `state.json.broken-<stamp>` (#2603).
+  function keepUnreadable() {
+    if (!loadError) return
+    const backup = `${file}.broken-${Date.now()}`
+    try {
+      copyFileSync(file, backup)
+    } catch (error) {
+      // A store that is gone by now has no bytes left to keep.
+      if (error?.code !== 'ENOENT') throw new Error(`Not saved: ${file} could not be read and its bytes could not be kept as ${backup} (${error?.code ?? error}).`)
+    }
+    loadError = null
+  }
+
   function write() {
     if (!persist || !file) return
     state.seq = seq
     mkdirSync(dir, { recursive: true })
     const tmp = `${file}.${process.pid}.tmp`
     if (!tempWritable(tmp)) throw new Error(`Not saved: ${tmp} is not a plain file of its own, so the workflow store was not written there.`)
+    keepUnreadable()
     writeFileSync(tmp, JSON.stringify(state))
     renameSync(tmp, file)
   }
@@ -571,7 +591,7 @@
       const missing = unmetRequired(item, id, run.produced, run.task)
       if (missing.length === 0) { started.push(id); continue }
       const preds = item.edges.filter(edge => edge.to === id && edge.loop !== true).map(edge => edge.from)
       const stuck = preds.length === 0 || preds.every(pred => status[pred] === 'done' || status[pred] === 'failed' || status[pred] === 'skipped')
       if (stuck) {
         status[id] = 'failed'
         blocked.push({ id, missing })
@@ -584,7 +604,7 @@
   function applyOp(item, run, op) {
     let working = run
     if (op.type === 'complete' && op.outputs && typeof op.outputs === 'object') {
       working = { ...run, produced: { ...(run.produced ?? {}), [op.nodeId]: op.outputs } }
     }
     let result
     if (op.type === 'start') result = { ...startFrontier(item, working.nodeStatus, working.fired), joined: false, layerSize: 0 }
@@ -1262,7 +1282,13 @@
   }
 
   return {
-    state(body) { return present(heldLimitOf(body)) },
+    state(body) {
+      // #2603: a store that could not be read is not answered as a list nothing in it
+      // came from. The factory templates are not the store, so `workflow/templates`
+      // still gets them from here.
+      if (loadError) return { ok: false, code: 'store-unreadable', error: loadError, templates: templateCatalog() }
+      return present(heldLimitOf(body))
+    },
     rename(body = {}) {
       if (readOnly()) return { ok: false, code: 'read-only', error: 'The host is read-only' }
       const workflowId = String(body.workflowId ?? '')
@@ -2257,27 +2283,33 @@
   state.laneFormat = true
 }
 
+// #2603: the reason a store could not be read travels beside the state instead of
+// ending at the warning. Every endpoint except `workflow/state` keeps running on the
+// fixture; the read is the one answer that must not pretend the fixture is the store.
 function load(file, warn) {
-  if (!file) return null
+  if (!file) return { state: null, error: null }
   let text
   try {
     text = readFileSync(file, 'utf8')
   } catch (error) {
-    if (error?.code === 'ENOENT') return null
-    warn(`flybot-workflow: could not read ${file}`)
-    return null
+    if (error?.code === 'ENOENT') return { state: null, error: null }
+    const problem = `the workflow store ${file} could not be read (${error?.code ?? error?.message ?? error})`
+    warn(`flybot-workflow: ${problem}; starting from the fixture`)
+    return { state: null, error: problem }
   }
   try {
     const parsed = JSON.parse(text)
     if (parsed?.version !== 1 || !Array.isArray(parsed.workflows) || typeof parsed.runs !== 'object') {
-      warn('flybot-workflow: the store is not a workflow file; starting from the fixture')
-      return null
+      const problem = `the workflow store ${file} is not a workflow file`
+      warn(`flybot-workflow: ${problem}; starting from the fixture`)
+      return { state: null, error: problem }
     }
-    return parsed
+    return { state: parsed, error: null }
   } catch (error) {
     // A file that does not parse is not a workflow store. Keep the plugin up
     // on the fixture instead of failing every endpoint.
-    warn(`flybot-workflow: the store did not parse (${error?.message ?? error}); starting from the fixture`)
-    return null
+    const problem = `the workflow store ${file} did not parse (${error?.message ?? error})`
+    warn(`flybot-workflow: ${problem}; starting from the fixture`)
+    return { state: null, error: problem }
   }
 }

plugins/workflow/test/store-unreadable.test.js

@@ -0,0 +1,55 @@
+// #2603: a crash or a full disk leaves the store truncated. `load()` warned and
+// returned null, so the read behind `workflow/state` answered `ok: true` with the
+// fixture and the first ordinary save renamed that fixture over the recoverable
+// bytes. The read now says the store could not be read, and the first write keeps
+// the bytes in `workflows.json.broken-<stamp>` before it replaces them.
+import assert from 'node:assert/strict'
+import { mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { test } from 'node:test'
+import { createWorkflowService } from '../src/service.js'
+
+const TRUNCATED = '{"version":1,"workflows":[{"id":"mine","nam'
+
+function unreadableStore() {
+  const dir = mkdtempSync(join(tmpdir(), 'flybot-wf-unreadable-'))
+  writeFileSync(join(dir, 'workflows.json'), TRUNCATED)
+  return dir
+}
+
+test('a store that did not parse is answered as a failure, not as the fixture', () => {
+  const dir = unreadableStore()
+  const warnings = []
+  const svc = createWorkflowService({ dir, persist: true, bots: () => [], warn: message => warnings.push(String(message)) })
+  try {
+    const read = svc.state({})
+    assert.equal(read.ok, false, 'a store that could not be read is not a clean list')
+    assert.equal(read.code, 'store-unreadable')
+    assert.match(read.error, /workflows\.json/, 'the answer names the file')
+    assert.match(read.error, /did not parse/, 'the answer names the failure')
+    assert.equal(Array.isArray(read.workflows), false, 'no fixture list is answered as the store')
+    assert.ok(warnings.some(message => /did not parse/.test(message)), 'the warning is still raised')
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})
+
+test('the first write keeps the unreadable bytes before it replaces them', () => {
+  const dir = unreadableStore()
+  const svc = createWorkflowService({ dir, persist: true, bots: () => [] })
+  try {
+    const created = svc.create({ name: 'After the crash', nodes: [{ id: 'plan', name: 'Planner' }], edges: [] })
+    assert.equal(created.ok, true, 'the plugin still saves')
+
+    const broken = readdirSync(dir).filter(name => name.startsWith('workflows.json.broken-'))
+    assert.equal(broken.length, 1, 'the unreadable file is kept once')
+    assert.equal(readFileSync(join(dir, broken[0]), 'utf8'), TRUNCATED, 'the kept copy holds the original bytes')
+
+    const live = JSON.parse(readFileSync(join(dir, 'workflows.json'), 'utf8'))
+    assert.ok(live.workflows.some(item => item.name === 'After the crash'), 'the live store parses again')
+    assert.equal(svc.state({}).ok, true, 'a store that can be read is a clean read again')
+  } finally {
+    rmSync(dir, { recursive: true, force: true })
+  }
+})