fix(plugin-workflow): fix parallel bug in loop (#2703)

This commit is contained in:
Junyi 2023-09-22 17:59:36 +08:00 committed by GitHub
parent 0ed164ff09
commit f1cf3cc45b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 75 additions and 7 deletions

View File

@ -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;

View File

@ -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);
});
});
});

View File

@ -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(),
);