From 2f892f00d7b81ce4096a36cadbef3e6312003654 Mon Sep 17 00:00:00 2001 From: Junyi Date: Wed, 20 Dec 2023 10:13:08 +0800 Subject: [PATCH] refactor(plugin-workflow): add logs and try/catch for preparing (#3236) --- .../plugin-workflow/src/server/Plugin.ts | 132 +++++++++++------- .../plugin-workflow/src/server/Processor.ts | 4 +- 2 files changed, 82 insertions(+), 54 deletions(-) diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts index 50156d747..4b05e6804 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/Plugin.ts @@ -25,9 +25,11 @@ import UpdateInstruction from './instructions/UpdateInstruction'; import type { ExecutionModel, JobModel, WorkflowModel } from './types'; +type ID = number | string; + type Pending = [ExecutionModel, JobModel?]; -type ID = number | string; +type CachedEvent = [WorkflowModel, any, { context?: any }]; export default class WorkflowPlugin extends Plugin { instructions: Registry = new Registry(); @@ -37,7 +39,7 @@ export default class WorkflowPlugin extends Plugin { private ready = false; private executing: Promise | null = null; private pending: Pending[] = []; - private events: [WorkflowModel, any, { context?: any }][] = []; + private events: CachedEvent[] = []; private loggerCache: LRUCache; @@ -221,7 +223,9 @@ export default class WorkflowPlugin extends Plugin { }); this.ready = false; - await this.prepare(); + if (this.events.length) { + await this.prepare(); + } if (this.executing) { await this.executing; } @@ -244,15 +248,22 @@ export default class WorkflowPlugin extends Plugin { } public trigger(workflow: WorkflowModel, context: object, options: { context?: any } = {}): void { + const logger = this.getLogger(workflow.id); + if (!this.ready) { + logger.warn(`app is not ready, event of workflow ${workflow.id} will be ignored`); + logger.debug(`ignored event data:`, { data: context }); + return; + } // `null` means not to trigger - if (!this.ready || context == null) { + if (context == null) { + logger.warn(`workflow ${workflow.id} event data context is null, event will be ignored`); return; } this.events.push([workflow, context, options]); - this.getLogger(workflow.id).info(`new event triggered, now events: ${this.events.length}`); - this.getLogger(workflow.id).debug(`event data:`, { + logger.info(`new event triggered, now events: ${this.events.length}`); + logger.debug(`event data:`, { data: context, }); @@ -268,7 +279,9 @@ export default class WorkflowPlugin extends Plugin { if (!job.execution) { job.execution = await job.getExecution(); } - + this.getLogger(job.execution.workflowId).info( + `execution (${job.execution.id}) resuming from job (${job.id}) added to pending list`, + ); this.pending.push([job.execution, job]); this.dispatch(); } @@ -277,14 +290,9 @@ export default class WorkflowPlugin extends Plugin { return new Processor(execution, { ...options, plugin: this }); } - private prepare = async () => { - const [event] = this.events; - if (!event) { - return; - } + private async createExecution(event: CachedEvent): Promise { const [workflow, context, options] = event; - let valid = true; if (options.context?.executionId) { // NOTE: no transaction here for read-uncommitted execution const existed = await workflow.countExecutions({ @@ -298,55 +306,69 @@ export default class WorkflowPlugin extends Plugin { `workflow ${workflow.id} has already been triggered in same execution (${options.context.executionId}), and newly triggering will be skipped.`, ); - valid = false; + return null; } } - if (valid) { - const execution = await this.db.sequelize.transaction(async (transaction) => { - const execution = await workflow.createExecution( - { - context, + const execution = await this.db.sequelize.transaction(async (transaction) => { + const execution = await workflow.createExecution( + { + context, + key: workflow.key, + status: EXECUTION_STATUS.QUEUEING, + }, + { transaction }, + ); + + await workflow.increment(['executed', 'allExecuted'], { transaction }); + // NOTE: https://sequelize.org/api/v6/class/src/model.js~model#instance-method-increment + if (this.db.options.dialect !== 'postgres') { + await workflow.reload({ transaction }); + } + + await (workflow.constructor).update( + { + allExecuted: workflow.allExecuted, + }, + { + where: { key: workflow.key, - status: EXECUTION_STATUS.QUEUEING, }, - { transaction }, - ); + transaction, + }, + ); - await workflow.increment(['executed', 'allExecuted'], { transaction }); - // NOTE: https://sequelize.org/api/v6/class/src/model.js~model#instance-method-increment - if (this.db.options.dialect !== 'postgres') { - await workflow.reload({ transaction }); - } + execution.workflow = workflow; - await (workflow.constructor).update( - { - allExecuted: workflow.allExecuted, - }, - { - where: { - key: workflow.key, - }, - transaction, - }, - ); + return execution; + }); - execution.workflow = workflow; + this.getLogger(workflow.id).info(`execution of workflow ${workflow.id} created as ${execution.id}`); - return execution; - }); - - this.getLogger(workflow.id).debug(`execution of workflow ${workflow.id} created as ${execution.id}`, { - data: execution.context, - }); - - // NOTE: cache first execution for most cases - if (!this.executing && !this.pending.length) { - this.pending.push([execution]); - } + // NOTE: cache first execution for most cases + if (!this.executing && !this.pending.length) { + this.pending.push([execution]); } - this.events.shift(); + return execution; + } + + private prepare = async () => { + const event = this.events.shift(); + if (!event) { + this.getLogger('dispatcher').warn(`events queue is empty, no need to prepare`); + return; + } + + const logger = this.getLogger(event[0].id); + logger.info(`preparing execution for event`); + + try { + await this.createExecution(event); + } catch (err) { + logger.error(`failed to create execution: ${err.message}`, err); + // this.events.push(event); // NOTE: retry will cause infinite loop + } if (this.events.length) { await this.prepare(); @@ -356,7 +378,13 @@ export default class WorkflowPlugin extends Plugin { }; private dispatch() { - if (!this.ready || this.executing) { + if (!this.ready) { + this.getLogger('dispatcher').warn(`app is not ready, new dispatching will be ignored`); + return; + } + + if (this.executing) { + this.getLogger('dispatcher').warn(`workflow executing is not finished, new dispatching will be ignored`); return; } diff --git a/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts b/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts index d7ed3dbac..d1dccf60c 100644 --- a/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts +++ b/packages/plugins/@nocobase/plugin-workflow/src/server/Processor.ts @@ -104,7 +104,7 @@ export default class Processor { public async start() { const { execution } = this; if (execution.status !== EXECUTION_STATUS.STARTED) { - throw new Error(`execution was ended with status ${execution.status}`); + throw new Error(`execution was ended with status ${execution.status} before, could not be started again`); } await this.prepare(); if (this.nodes.length) { @@ -118,7 +118,7 @@ export default class Processor { public async resume(job: JobModel) { const { execution } = this; if (execution.status !== EXECUTION_STATUS.STARTED) { - throw new Error(`execution was ended with status ${execution.status}`); + throw new Error(`execution was ended with status ${execution.status} before, could not be resumed`); } await this.prepare(); const node = this.nodesMap.get(job.nodeId);