diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts index b61db15b6..1fcde4a8d 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts @@ -306,13 +306,18 @@ export default class Processor { return null; } - findBranchLastJob(node: FlowNodeModel): JobModel | null { + findBranchLastJob(node: FlowNodeModel, job: JobModel): JobModel | null { + const allJobs = Array.from(this.jobsMap.values()); + const branchJobs = []; for (let n = this.findBranchEndNode(node); n && n !== node.upstream; n = n.upstream) { - const jobs = Array.from(this.jobsMap.values()) - .filter((item) => item.nodeId === n.id) - .sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime()); - if (jobs.length) { - return jobs[jobs.length - 1]; + branchJobs.push(...allJobs.filter((item) => item.nodeId === n.id)); + } + branchJobs.sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime()); + for (let i = branchJobs.length - 1; i >= 0; i -= 1) { + for (let j = branchJobs[i]; j && j.id !== job.id; j = this.jobsMap.get(j.upstreamId)) { + if (j.upstreamId === job.id) { + return branchJobs[i]; + } } } return null; diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/Processor.test.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/Processor.test.ts index f6573c5b3..ba9f918ba 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/Processor.test.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/__tests__/Processor.test.ts @@ -470,5 +470,68 @@ describe('workflow > Processor', () => { const jobs = await e2.getJobs({ order: [['id', 'ASC']] }); expect(jobs.length).toEqual(5); }); + + it('loop branch contains parallel branches', async () => { + const n1 = await workflow.createNode({ + type: 'loop', + config: { + target: 2, + }, + }); + + const n2 = await workflow.createNode({ + type: 'parallel', + branchIndex: 0, + upstreamId: n1.id, + config: { + mode: 'any', + }, + }); + + const n3 = await workflow.createNode({ + type: 'condition', + config: { + rejectOnFalse: true, + calculation: { + calculator: '<', + operands: [`{{$scopes.${n1.id}.item}}`, 1], + }, + }, + branchIndex: 0, + upstreamId: n2.id, + }); + const n4 = await workflow.createNode({ + type: 'echo', + upstreamId: n3.id, + }); + await n3.setDownstream(n4); + + const n5 = await workflow.createNode({ + type: 'condition', + config: { + rejectOnFalse: true, + calculation: { + calculator: '<', + operands: [`{{$scopes.${n1.id}.item}}`, 1], + }, + }, + branchIndex: 1, + upstreamId: n2.id, + }); + const n6 = await workflow.createNode({ + type: 'echo', + upstreamId: n5.id, + }); + await n5.setDownstream(n6); + + const post = await PostRepo.create({ values: { title: 't1' } }); + + await sleep(1000); + + const [e1] = await workflow.getExecutions(); + expect(e1.status).toEqual(EXECUTION_STATUS.FAILED); + const e1jobs = await e1.getJobs(); + expect(e1jobs.length).toBe(7); + }); }); }); diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/parallel.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/parallel.ts index acd1bd91d..166d8fae5 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/parallel.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/instructions/parallel.ts @@ -81,7 +81,7 @@ export default { await processor.run(branch, job); // find last job of the branch - return processor.findBranchLastJob(branch); + return processor.findBranchLastJob(branch, job); }), Promise.resolve(), );