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..38e858ee4 --- /dev/null +++ b/plugins/workflow/test/resume-during-dispatch.test.js @@ -0,0 +1,59 @@ +// 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' + +const settle = (ms = 30) => new Promise(resolve => setTimeout(resolve, ms)) + +function bench() { + const gates = [] + 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 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() + + // #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') + + // #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') + + for (let round = 0; round < 3; round += 1) { + for (const resolve of gates.splice(0)) resolve() + await settle(20) + } +}) 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 }) + } + }) +}