From b701870cf91535cf507ac9fd34d4bf33261bd9c7 Mon Sep 17 00:00:00 2001 From: Junyi Date: Tue, 29 Jul 2025 15:37:34 +0800 Subject: [PATCH] fix(plugin-workflow): fix bigint id issue when save job (#7292) * fix(plugin-workflow): fix bigint id issue when save job * fix(plugin-workflow): fix test cases * fix(plugin-workflow-request): fault-tolerant for job not found --- .../src/server/DelayInstruction.ts | 26 ++++++---- .../src/server/RequestInstruction.ts | 32 ++++++++----- .../plugin-workflow/src/server/Processor.ts | 47 ++++++++++++++----- 3 files changed, 72 insertions(+), 33 deletions(-) diff --git a/packages/plugins/@nocobase/plugin-workflow-delay/src/server/DelayInstruction.ts b/packages/plugins/@nocobase/plugin-workflow-delay/src/server/DelayInstruction.ts index f5b05990068..bb5bc06b317 100644 --- a/packages/plugins/@nocobase/plugin-workflow-delay/src/server/DelayInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow-delay/src/server/DelayInstruction.ts @@ -23,7 +23,7 @@ interface DelayConfig { } export default class extends Instruction { - timers: Map = new Map(); + timers: Map = new Map(); constructor(public workflow: WorkflowPlugin) { super(workflow); @@ -72,26 +72,32 @@ export default class extends Instruction { }; schedule(job) { - const now = new Date(); - const createdAt = Date.parse(job.createdAt); - const delay = createdAt + job.result - now.getTime(); + const createdAt = new Date(job.createdAt).getTime(); + const delay = createdAt + job.result - Date.now(); if (delay > 0) { - const trigger = this.trigger.bind(this, job); - this.timers.set(job.id, setTimeout(trigger, delay)); + const trigger = this.trigger.bind(this, job.id); + this.timers.set(job.id.toString(), setTimeout(trigger, delay)); } else { this.trigger(job); } } - async trigger(job) { + async trigger(jobOrId: JobModel | string) { + const { model } = this.workflow.app.db.getCollection('jobs'); + const job = + jobOrId instanceof model + ? jobOrId + : await this.workflow.app.db.getRepository('jobs').findOne({ filterByTk: jobOrId }); if (!job.execution) { job.execution = await job.getExecution(); } if (job.execution.status === EXECUTION_STATUS.STARTED) { this.workflow.resume(job); } - if (this.timers.get(job.id)) { - this.timers.delete(job.id); + const idStr = job.id.toString(); + if (this.timers.get(idStr)) { + clearTimeout(this.timers.get(idStr)); + this.timers.delete(idStr); } } @@ -104,10 +110,10 @@ export default class extends Instruction { nodeKey: node.key, upstreamId: prevJob?.id ?? null, }); - job.node = node; // add to schedule this.schedule(job); + processor.logger.debug(`delay node (${node.id}) will resume after ${duration}ms`); return null; } diff --git a/packages/plugins/@nocobase/plugin-workflow-request/src/server/RequestInstruction.ts b/packages/plugins/@nocobase/plugin-workflow-request/src/server/RequestInstruction.ts index 4f848f458be..e92779e4530 100644 --- a/packages/plugins/@nocobase/plugin-workflow-request/src/server/RequestInstruction.ts +++ b/packages/plugins/@nocobase/plugin-workflow-request/src/server/RequestInstruction.ts @@ -10,7 +10,7 @@ import axios, { AxiosRequestConfig } from 'axios'; import { trim } from 'lodash'; -import { Processor, Instruction, JOB_STATUS, FlowNodeModel } from '@nocobase/plugin-workflow'; +import { Processor, Instruction, JOB_STATUS, FlowNodeModel, IJob } from '@nocobase/plugin-workflow'; import PluginFileManagerServer, { AttachmentModel } from '@nocobase/plugin-file-manager'; import { Application } from '@nocobase/server'; import { Readable } from 'stream'; @@ -183,22 +183,24 @@ export default class extends Instruction { } } - const job = processor.saveJob({ + const { id } = processor.saveJob({ status: JOB_STATUS.PENDING, nodeId: node.id, nodeKey: node.key, upstreamId: prevJob?.id ?? null, }); + const jobDone: IJob = { + status: JOB_STATUS.PENDING, + }; + // eslint-disable-next-line promise/catch-or-return request(config, this.workflow.app) .then((response) => { processor.logger.info(`request (#${node.id}) response success, status: ${response.status}`); - job.set({ - status: JOB_STATUS.RESOLVED, - result: responseSuccess(response, config.onlyData), - }); + jobDone.status = JOB_STATUS.RESOLVED; + jobDone.result = responseSuccess(response, config.onlyData); }) .catch((error) => { if (error.isAxiosError) { @@ -213,14 +215,22 @@ export default class extends Instruction { processor.logger.error(`request (#${node.id}) failed unexpectedly: ${error.message}`); } - job.set({ - status: JOB_STATUS.FAILED, - result: responseFailure(error), - }); + jobDone.status = JOB_STATUS.FAILED; + jobDone.result = responseFailure(error); }) .finally(() => { processor.logger.debug(`request (#${node.id}) ended, resume workflow...`); - setTimeout(() => { + setTimeout(async () => { + const job = await this.workflow.app.db.getRepository('jobs').findOne({ + filterByTk: id, + }); + if (!job) { + processor.logger.error( + `request job (${id}) not found, execution (${processor.execution.id}) cannot be resumed.`, + ); + return; + } + job.set(jobDone); job.execution = processor.execution; this.workflow.resume(job); }); diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts index 4d79943d01b..e4f6bdac72e 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts @@ -16,7 +16,7 @@ import set from 'lodash/set'; import type Plugin from './Plugin'; import { EXECUTION_STATUS, JOB_STATUS } from './constants'; import { Runner } from './instructions'; -import { ExecutionModel, FlowNodeModel, JobModel } from './types'; +import type { ExecutionModel, FlowNodeModel, JobModel } from './types'; export interface ProcessorOptions extends Transactionable { plugin: Plugin; @@ -55,7 +55,7 @@ export default class Processor { /** * @experimental */ - nodesMap = new Map(); + nodesMap = new Map(); private jobsMapByNodeKey: { [key: string]: JobModel } = {}; private jobResultsMapByNodeKey: { [key: string]: any } = {}; @@ -283,7 +283,25 @@ export default class Processor { if (job.isNewRecord) { newJobs.push(job); } else { - await job.save({ transaction: this.mainTransaction }); + const JobCollection = this.options.plugin.db.getCollection('jobs'); + const changes = []; + if (job.changed('status')) { + changes.push([`status`, job.status]); + job.changed('status', false); + } + if (job.changed('result')) { + changes.push([`result`, JSON.stringify(job.result)]); + job.changed('result', false); + } + if (changes.length) { + await this.options.plugin.db.sequelize.query( + `UPDATE ${JobCollection.quotedTableName()} SET ${changes.map(([key]) => `${key} = ?`)} WHERE id='${ + job.id + }'`, + { replacements: changes.map(([, value]) => value), transaction: this.mainTransaction }, + ); + } + // await job.save({ transaction: this.mainTransaction }); } } if (newJobs.length) { @@ -323,13 +341,18 @@ export default class Processor { job = payload; job.set('updatedAt', new Date()); } else { - job = model.build({ - ...payload, - id: this.options.plugin.snowflake.getUniqueID().toString(), - createdAt: new Date(), - updatedAt: new Date(), - executionId: this.execution.id, - }); + job = model.build( + { + ...payload, + id: this.options.plugin.snowflake.getUniqueID().toString(), + createdAt: new Date(), + updatedAt: new Date(), + executionId: this.execution.id, + }, + { + isNewRecord: true, + }, + ); } this.jobsToSave.set(job.id, job); @@ -417,7 +440,7 @@ export default class Processor { /** * @experimental */ - public getScope(sourceNodeId: number, includeSelfScope = false) { + public getScope(sourceNodeId: number | string, includeSelfScope = false) { const node = this.nodesMap.get(sourceNodeId); const systemFns = {}; const scope = { @@ -448,7 +471,7 @@ export default class Processor { /** * @experimental */ - public getParsedValue(value, sourceNodeId: number, { additionalScope = {}, includeSelfScope = false } = {}) { + public getParsedValue(value, sourceNodeId: number | string, { additionalScope = {}, includeSelfScope = false } = {}) { const template = parse(value); const scope = Object.assign(this.getScope(sourceNodeId, includeSelfScope), additionalScope); template.parameters.forEach(({ key }) => {