fix(plugin-workflow): fix missed preparing (#1337)

This commit is contained in:
Junyi 2023-01-08 00:56:43 +08:00 committed by GitHub
parent f6769341bd
commit 322194f24a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 47 additions and 35 deletions

View File

@ -2,4 +2,5 @@ import { AppLoggerOptions } from '@nocobase/logger';
export default { export default {
transports: process.env.LOGGER_TRANSPORT || ['console', 'dailyRotateFile'], transports: process.env.LOGGER_TRANSPORT || ['console', 'dailyRotateFile'],
level: process.env.LOGGER_LEVEL || (process.env.APP_ENV === 'development' ? 'debug' : 'info')
} as AppLoggerOptions; } as AppLoggerOptions;

View File

@ -149,6 +149,8 @@ export default class WorkflowPlugin extends Plugin {
this.events.push([workflow, context, options]); this.events.push([workflow, context, options]);
this.app.logger.debug(`[Workflow] new event triggered, now events: ${this.events.length}`);
if (this.events.length > 1) { if (this.events.length > 1) {
return; return;
} }
@ -164,6 +166,7 @@ export default class WorkflowPlugin extends Plugin {
} }
const [workflow, context, options] = event; const [workflow, context, options] = event;
let valid = true;
if (options.context?.executionId) { if (options.context?.executionId) {
// NOTE: no transaction here for read-uncommitted execution // NOTE: no transaction here for read-uncommitted execution
const existed = await workflow.countExecutions({ const existed = await workflow.countExecutions({
@ -173,48 +176,52 @@ export default class WorkflowPlugin extends Plugin {
}); });
if (existed) { if (existed) {
console.warn(`workflow ${workflow.id} has already been triggered in same execution (${options.context.executionId}), and newly triggering will be skipped.`); this.app.logger.warn(`[Workflow] workflow ${workflow.id} has already been triggered in same execution (${options.context.executionId}), and newly triggering will be skipped.`);
return; valid = false;
} }
} }
const execution = await this.db.sequelize.transaction(async transaction => { if (valid) {
const execution = await workflow.createExecution({ const execution = await this.db.sequelize.transaction(async transaction => {
context, const execution = await workflow.createExecution({
key: workflow.key, context,
status: EXECUTION_STATUS.CREATED, key: workflow.key,
useTransaction: workflow.useTransaction, status: EXECUTION_STATUS.CREATED,
}, { transaction }); useTransaction: workflow.useTransaction,
}, { transaction });
const executed = await workflow.countExecutions({ transaction }); const executed = await workflow.countExecutions({ transaction });
// NOTE: not to trigger afterUpdate hook here // NOTE: not to trigger afterUpdate hook here
await workflow.update({ executed }, { transaction, hooks: false }); await workflow.update({ executed }, { transaction, hooks: false });
const allExecuted = await (<typeof ExecutionModel>execution.constructor).count({ const allExecuted = await (<typeof ExecutionModel>execution.constructor).count({
where: { where: {
key: workflow.key key: workflow.key
}, },
transaction transaction
}); });
await (<typeof WorkflowModel>workflow.constructor).update({ await (<typeof WorkflowModel>workflow.constructor).update({
allExecuted allExecuted
}, { }, {
where: { where: {
key: workflow.key key: workflow.key
}, },
individualHooks: true, individualHooks: true,
transaction transaction
});
execution.workflow = workflow;
return execution;
}); });
execution.workflow = workflow; this.app.logger.debug(`[Workflow] execution of workflow ${workflow.id} created as ${execution.id}`);
return execution; // 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]);
} }
if (this.events.length) { if (this.events.length) {
@ -267,9 +274,13 @@ export default class WorkflowPlugin extends Plugin {
const processor = this.createProcessor(execution); const processor = this.createProcessor(execution);
console.log('workflow processing:', new Date(), execution.workflowId, execution.id); this.app.logger.info(`[Workflow] execution ${execution.id} ${job ? 'resuming' : 'starting'} ...`);
await (job ? processor.resume(job) : processor.start()); try {
await (job ? processor.resume(job) : processor.start());
} catch (err) {
this.app.logger.error(`[Workflow] ${err.message}`, err);
}
this.executing = null; this.executing = null;