From 87aae6043f9925631697e377ccf097bc8b91d44d Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Mon, 9 Nov 2026 13:02:45 +0800 Subject: [PATCH 1/2] fix(workflow): keep a pause, a resume and a parked card when a dispatch lands A hand-off awaiting its dispatch stores the run record it grabbed before, so a pause or resume stored meanwhile was overwritten by that stale object. A parked card whose start the host refused left the queue for good while the claim still named its lane. And a store entry that is not a workflow threw out of load() instead of answering store-unreadable, so the damaged bytes were never kept. Fixes #2734 Fixes #2735 Fixes #2736 Docs-not-needed: no subsystem page owns these state transitions, and nothing but the three defects changes behaviour. --- plugins/workflow/src/service.js | 29 ++++++++--- .../test/pause-during-dispatch.test.js | 31 ++++++++++++ .../test/queued-card-not-dropped.test.js | 48 +++++++++++++++++++ .../test/resume-during-dispatch.test.js | 33 +++++++++++++ .../test/store-entry-unreadable.test.js | 34 +++++++++++++ 5 files changed, 168 insertions(+), 7 deletions(-) create mode 100644 plugins/workflow/test/pause-during-dispatch.test.js create mode 100644 plugins/workflow/test/queued-card-not-dropped.test.js create mode 100644 plugins/workflow/test/resume-during-dispatch.test.js create mode 100644 plugins/workflow/test/store-entry-unreadable.test.js diff --git a/plugins/workflow/src/service.js b/plugins/workflow/src/service.js index ea514dfbe..4bc1ac77b 100644 --- a/plugins/workflow/src/service.js +++ b/plugins/workflow/src/service.js @@ -1010,8 +1010,10 @@ export function createWorkflowService({ if (!best) return null const task = laneOf(best.item.id).queue.shift() // #1869: the row is rebuilt from the parked fields, so the brief has to be - // named here too or it is lost exactly when the run starts. - return { item: best.item, task: carriedTask(task) } + // named here too or it is lost exactly when the run starts. The parked row + // itself travels back too, so a start the host refuses puts the card where + // it was instead of taking it out of the queue for good. + return { item: best.item, task: carriedTask(task), row: task } } function noteUnstarted(item, error) { @@ -1029,6 +1031,10 @@ export function createWorkflowService({ if (!next) return const begun = beginRun(next.item, next.task) if (begun.ok === false) { + // The card never started, so it is still the owner's parked work. Taking it out + // of the queue here loses it: no run holds it, no claim is retried for it, and + // the next ingest answers "already" for a card in nobody's hands. + laneOf(next.item.id).queue.unshift(next.row) noteUnstarted(next.item, begun.error) return } @@ -1047,6 +1053,7 @@ export function createWorkflowService({ if (!next) return const started = await startLive(next.item, next.task) if (started?.ok === false) { + laneOf(next.item.id).queue.unshift(next.row) noteUnstarted(next.item, started.error) return } @@ -1801,8 +1808,11 @@ export function createWorkflowService({ let run = runId ? lane.active.find(itemRun => itemRun.id === runId) : lane.active.find(itemRun => itemRun.live === true && itemRun.status === 'running') if (!run || run.live !== true) return { ok: false, error: 'No live run to pause' } if (run.paused === true) return present() - run = { ...run, paused: true, updatedAt: now() } - storeRun(workflowId, run) + // Set on the record the lane holds, not on a copy: a hand-off waiting on its dispatch + // still holds that same object and stores it back when the dispatch lands, so a copy + // here is replaced by the hand-off's copy and the pause is silently lost. + run.paused = true + run.updatedAt = now() syncTimers(item, run) push(workflowId, [{ kind: 'system', @@ -1827,8 +1837,8 @@ export function createWorkflowService({ // silent skip. Stop again is the retry (#1476). if (run.releasePending === true) return { ok: false, error: 'This run is still waiting to release a Bot; stop it again to retry' } if (run.paused !== true) return present() - run = { ...run, paused: false, updatedAt: now() } - storeRun(workflowId, run) + run.paused = false + run.updatedAt = now() syncTimers(item, run) push(workflowId, [{ kind: 'system', @@ -2475,7 +2485,12 @@ function load(file, warn) { warn(`flybot-workflow: ${problem}; starting from the fixture`) return { state: null, error: problem, newer: version } } - if (version !== STORE_VERSION || !Array.isArray(parsed.workflows) || typeof parsed.runs !== 'object') { + // A store this build can parse but not read is the same failure as one that does not parse: + // the read has to say so instead of throwing out of present() on every render. A workflow + // entry that is not an object, or carries no nodes, is not one this build wrote. + const entriesReadable = Array.isArray(parsed.workflows) + && parsed.workflows.every(item => item !== null && typeof item === 'object' && Array.isArray(item.nodes) && item.nodes.length > 0) + if (version !== STORE_VERSION || !entriesReadable || parsed.runs === null || typeof parsed.runs !== 'object') { const problem = `the workflow store ${file} is not a workflow file` warn(`flybot-workflow: ${problem}; starting from the fixture`) return { state: null, error: problem } diff --git a/plugins/workflow/test/pause-during-dispatch.test.js b/plugins/workflow/test/pause-during-dispatch.test.js new file mode 100644 index 000000000..66eb4fa91 --- /dev/null +++ b/plugins/workflow/test/pause-during-dispatch.test.js @@ -0,0 +1,31 @@ +// A Pause that arrives while a step's dispatch is in flight must not be thrown away. +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { createWorkflowService } from '../src/service.js' +import { DELIVERY_ID } from '../src/template.js' + +test('pause during an in-flight step dispatch is not lost', async () => { + let release + const gate = new Promise(resolve => { release = resolve }) + let seq = 0 + const svc = createWorkflowService({ + persist: false, + now: () => 1_700_000_000_000, + bots: () => [{ id: 'main', name: 'Chief', isMain: true, remote: false }], + spawn: async () => { seq += 1; return `eph-${seq}` }, + release: () => {}, + dispatch: async () => { await gate; return { ok: true } }, + }) + const row = () => svc.state().workflows.find(item => item.id === DELIVERY_ID) + + const assigning = svc.assign({ workflowId: DELIVERY_ID, taskId: '#9001', title: 'Pause while dispatching' }) + await new Promise(resolve => setTimeout(resolve, 20)) // the dispatch is now blocked in flight + svc.pause({ workflowId: DELIVERY_ID }) + assert.equal(row().run.paused, true) + release() + await assigning + await new Promise(resolve => setTimeout(resolve, 20)) + + assert.equal(row().run.paused, true, 'the run came back unpaused after the dispatch landed') + assert.doesNotMatch(row().preview, /正在处理/) +}) diff --git a/plugins/workflow/test/queued-card-not-dropped.test.js b/plugins/workflow/test/queued-card-not-dropped.test.js new file mode 100644 index 000000000..e48d51231 --- /dev/null +++ b/plugins/workflow/test/queued-card-not-dropped.test.js @@ -0,0 +1,48 @@ +// A card that was parked and could not start is still the owner's parked work: it stays +// in the queue and starts on the next drain, instead of leaving the queue for good. +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { createWorkflowService } from '../src/service.js' +import { DELIVERY_ID } from '../src/template.js' + +function bench() { + const state = { armed: false, seed: 0 } + const svc = createWorkflowService({ + persist: false, + now: () => 1_700_000_000_000, + bots: () => [{ id: 'main', name: 'Chief', isMain: true, remote: false }], + spawn: async () => { + if (state.armed) throw new Error('the host could not start a Bot (transient)') + state.seed += 1 + return `eph-${state.seed}` + }, + release: () => {}, + dispatch: async () => ({ ok: true }), + }) + return { svc, state } +} + +const row = svc => svc.state().workflows.find(item => item.id === DELIVERY_ID) +const queued = svc => row(svc).queue.map(task => task.id) +const running = svc => row(svc).runs.filter(run => run.status === 'running').map(run => run.task?.id) + +test('a parked card whose start fails stays in the queue', async () => { + const { svc, state } = bench() + svc.setConcurrency({ workflowId: DELIVERY_ID, maxConcurrent: 1 }) + await svc.ingest({ id: '#A', title: 'First card', status: 'open', workflowId: DELIVERY_ID }) + await svc.ingest({ id: '#B', title: 'Second card', status: 'open', workflowId: DELIVERY_ID }) + assert.deepEqual(queued(svc), ['#B']) + + const first = row(svc).runs.find(run => run.status === 'running') + state.armed = true // free the slot; the spawn for the parked card fails once + await svc.reset({ workflowId: DELIVERY_ID, runId: first.id }) + state.armed = false + + assert.deepEqual(queued(svc), ['#B'], 'the parked card left the queue without starting') + assert.deepEqual(running(svc), []) + + // A later card joins the tail and drains the queue: the parked card starts first. + await svc.ingest({ id: '#C', title: 'Third card', status: 'open', workflowId: DELIVERY_ID }) + assert.deepEqual(running(svc), ['#B'], 'the parked card never got its turn') + assert.deepEqual(queued(svc), ['#C']) +}) diff --git a/plugins/workflow/test/resume-during-dispatch.test.js b/plugins/workflow/test/resume-during-dispatch.test.js new file mode 100644 index 000000000..53bdf7d68 --- /dev/null +++ b/plugins/workflow/test/resume-during-dispatch.test.js @@ -0,0 +1,33 @@ +// A Resume that arrives while a step's dispatch is in flight must not be thrown away either. +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { createWorkflowService } from '../src/service.js' +import { DELIVERY_ID } from '../src/template.js' + +test('resume during an in-flight step dispatch is not lost', async () => { + let release + const gate = new Promise(resolve => { release = resolve }) + let seq = 0 + const svc = createWorkflowService({ + persist: false, + now: () => 1_700_000_000_000, + bots: () => [{ id: 'main', name: 'Chief', isMain: true, remote: false }], + spawn: async () => { seq += 1; return `eph-${seq}` }, + release: () => {}, + dispatch: async () => { await gate; return { ok: true } }, + }) + const row = () => svc.state().workflows.find(item => item.id === DELIVERY_ID) + + const assigning = svc.assign({ workflowId: DELIVERY_ID, taskId: '#9002', title: 'Resume while dispatching' }) + await new Promise(resolve => setTimeout(resolve, 20)) // the dispatch is now blocked in flight + svc.pause({ workflowId: DELIVERY_ID }) + assert.equal(row().run.paused, true) + svc.resume({ workflowId: DELIVERY_ID }) + assert.notEqual(row().run.paused, true, 'the resume did not take effect') + + release() + await assigning + await new Promise(resolve => setTimeout(resolve, 20)) + + assert.notEqual(row().run.paused, true, 'the run came back paused after the dispatch landed') +}) diff --git a/plugins/workflow/test/store-entry-unreadable.test.js b/plugins/workflow/test/store-entry-unreadable.test.js new file mode 100644 index 000000000..973f3b1ed --- /dev/null +++ b/plugins/workflow/test/store-entry-unreadable.test.js @@ -0,0 +1,34 @@ +// A store that parses but whose workflow entries are not workflows is the same failure +// as a store that does not parse: the read answers `store-unreadable` instead of throwing. +import assert from 'node:assert/strict' +import { mkdtempSync, 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 STORE_VERSION = 1 + +function openWith(entry) { + const dir = mkdtempSync(join(tmpdir(), 'store-entry-')) + writeFileSync(join(dir, 'workflows.json'), JSON.stringify({ version: STORE_VERSION, workflows: [entry], runs: {} })) + const svc = createWorkflowService({ dir, persist: true, now: () => 1, bots: () => [], warn: () => {} }) + return { svc, dir } +} + +for (const [label, entry] of [ + ['an entry that is not an object', null], + ['an entry with no nodes', { id: 'w1', name: 'Broken' }], + ['an entry whose nodes are not a list', { id: 'w1', name: 'Broken', nodes: null, edges: [], layers: [] }], +]) { + test(`a store with ${label} answers store-unreadable`, () => { + const { svc, dir } = openWith(entry) + try { + const state = svc.state() + assert.equal(state.ok, false, 'the read answered a list instead of the failure') + assert.equal(state.code, 'store-unreadable') + } finally { + rmSync(dir, { recursive: true, force: true }) + } + }) +} From 24c7935f91e69dc65d458cbdc8ac31e40237f477 Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Mon, 9 Nov 2026 13:16:05 +0800 Subject: [PATCH 2/2] test(workflow): make the resume case red without the fix, with a pause control The previous resume case passed on the base tree because the pause it relied on was thrown away by the same landing, so it could not tell "the resume held" from "the pause was already lost". It now pauses and resumes the run while its dispatch is in flight, then does the same landing on a second card with no resume: the control is what fails without the pause fix, and the resume case is what fails without the resume fix. Docs-not-needed: test only, no behaviour or subsystem page changes. --- .../test/resume-during-dispatch.test.js | 56 ++++++++++++++----- 1 file changed, 41 insertions(+), 15 deletions(-) diff --git a/plugins/workflow/test/resume-during-dispatch.test.js b/plugins/workflow/test/resume-during-dispatch.test.js index 53bdf7d68..38e858ee4 100644 --- a/plugins/workflow/test/resume-during-dispatch.test.js +++ b/plugins/workflow/test/resume-during-dispatch.test.js @@ -1,12 +1,16 @@ -// A Resume that arrives while a step's dispatch is in flight must not be thrown away either. +// A Resume is thrown away the same way a Pause is: the hand-off that was already in +// flight stores the run record it grabbed before, so the resolve never reaches the lane. +// The second card is the control — unless a pause survives its own landing, the resume +// case above it would pass without the fix purely because the pause was lost too. import assert from 'node:assert/strict' import { test } from 'node:test' import { createWorkflowService } from '../src/service.js' import { DELIVERY_ID } from '../src/template.js' -test('resume during an in-flight step dispatch is not lost', async () => { - let release - const gate = new Promise(resolve => { release = resolve }) +const settle = (ms = 30) => new Promise(resolve => setTimeout(resolve, ms)) + +function bench() { + const gates = [] let seq = 0 const svc = createWorkflowService({ persist: false, @@ -14,20 +18,42 @@ test('resume during an in-flight step dispatch is not lost', async () => { bots: () => [{ id: 'main', name: 'Chief', isMain: true, remote: false }], spawn: async () => { seq += 1; return `eph-${seq}` }, release: () => {}, - dispatch: async () => { await gate; return { ok: true } }, + dispatch: async () => { await new Promise(resolve => { gates.push(resolve) }); return { ok: true } }, }) const row = () => svc.state().workflows.find(item => item.id === DELIVERY_ID) + const runOf = taskId => row().runs.find(run => run.task?.id === taskId) + return { svc, gates, runOf } +} + +test('resume during an in-flight step dispatch is not lost', async () => { + const { svc, gates, runOf } = bench() - const assigning = svc.assign({ workflowId: DELIVERY_ID, taskId: '#9002', title: 'Resume while dispatching' }) - await new Promise(resolve => setTimeout(resolve, 20)) // the dispatch is now blocked in flight - svc.pause({ workflowId: DELIVERY_ID }) - assert.equal(row().run.paused, true) - svc.resume({ workflowId: DELIVERY_ID }) - assert.notEqual(row().run.paused, true, 'the resume did not take effect') + // #9002 — the resume case: a pause and then a resume, both while the dispatch is in flight. + const first = svc.assign({ workflowId: DELIVERY_ID, taskId: '#9002', title: 'Resume while dispatching' }) + await settle() + const a = runOf('#9002').id + svc.pause({ workflowId: DELIVERY_ID, runId: a }) + assert.equal(runOf('#9002').paused, true, 'the pause did not take effect') + svc.resume({ workflowId: DELIVERY_ID, runId: a }) + assert.notEqual(runOf('#9002').paused, true, 'the resume did not take effect') + gates[0]() + await first + await settle() + assert.notEqual(runOf('#9002').paused, true, 'the resume was thrown away by the dispatch that landed') - release() - await assigning - await new Promise(resolve => setTimeout(resolve, 20)) + // #9003 — the control: the same landing with no resume has to leave the run paused. + const second = svc.assign({ workflowId: DELIVERY_ID, taskId: '#9003', title: 'Pause while dispatching' }) + await settle() + const b = runOf('#9003').id + svc.pause({ workflowId: DELIVERY_ID, runId: b }) + assert.equal(runOf('#9003').paused, true, 'the pause did not take effect') + gates[1]() + await second + await settle() + assert.equal(runOf('#9003').paused, true, 'the pause was thrown away by the dispatch that landed') - assert.notEqual(row().run.paused, true, 'the run came back paused after the dispatch landed') + for (let round = 0; round < 3; round += 1) { + for (const resolve of gates.splice(0)) resolve() + await settle(20) + } })