fix: create task when waiting (#352)

> 下游同步 cnpmcore 项目时，cnpmcore 会同时发送 [version, tag] 两个独立的事件
有可能下游在处理 version change 时，tag 尚未创建完成
但由于 task targetName 幂等，导致 tag 的 change 事件未响应，造成 tag 异常

* 创建同步任务时进行判断，如果任务还未开始才放弃创建

app/core/service/TaskService.ts

@@ -28,16 +28,21 @@ export class TaskService extends AbstractService {
   public async createTask(task: Task, addTaskQueueOnExists: boolean) {
     const existsTask = await this.taskRepository.findTaskByTargetName(task.targetName, task.type);
     if (existsTask) {
-      if (addTaskQueueOnExists && existsTask.state === TaskState.Waiting) {
-        const queueLength = await this.getTaskQueueLength(task.type);
-        if (queueLength < this.config.cnpmcore.taskQueueHighWaterSize) {
-          // make sure waiting task in queue
-          await this.queueAdapter.push<string>(task.type, existsTask.taskId);
-          this.logger.info('[TaskService.createTask:exists-to-queue] taskType: %s, targetName: %s, taskId: %s, queue size: %s',
-            task.type, task.targetName, task.taskId, queueLength);
+      // 如果任务还未被触发，就不继续重复创建
+      // 如果任务正在执行，可能任务状态已更新，这种情况需要继续创建
+      if (existsTask.state === TaskState.Waiting) {
+        // 提高任务的优先级
+        if (addTaskQueueOnExists) {
+          const queueLength = await this.getTaskQueueLength(task.type);
+          if (queueLength < this.config.cnpmcore.taskQueueHighWaterSize) {
+            // make sure waiting task in queue
+            await this.queueAdapter.push<string>(task.type, existsTask.taskId);
+            this.logger.info('[TaskService.createTask:exists-to-queue] taskType: %s, targetName: %s, taskId: %s, queue size: %s',
+              task.type, task.targetName, task.taskId, queueLength);
+          }
         }
+        return existsTask;
       }
-      return existsTask;
     }
     await this.taskRepository.saveTask(task);
     await this.queueAdapter.push<string>(task.type, task.taskId);

test/core/service/PackageSyncerService/createTask.test.ts

@@ -1,18 +1,27 @@
 import assert = require('assert');
 import { app, mock } from 'egg-mock/bootstrap';
 import { Context } from 'egg';
+import { setTimeout } from 'timers/promises';
 import { PackageSyncerService } from '../../../../app/core/service/PackageSyncerService';
 import { TestUtil } from '../../../TestUtil';
+import { Task } from 'app/core/entity/Task';
+import { TaskState } from '../../../../app/common/enum/Task';
+import { TaskRepository } from '../../../../app/repository/TaskRepository';
+import { TaskService } from '../../../../app/core/service/TaskService';
 
 describe('test/core/service/PackageSyncerService/createTask.test.ts', () => {
   let ctx: Context;
   const pkgName = '@cnpmcore/foo';
   const username = 'mock_username';
   let packageSyncerService: PackageSyncerService;
+  let taskRepository: TaskRepository;
+  let taskService: TaskService;
 
   beforeEach(async () => {
     ctx = await app.mockModuleContext();
     packageSyncerService = await ctx.getEggObject(PackageSyncerService);
+    taskRepository = await ctx.getEggObject(TaskRepository);
+    taskService = await ctx.getEggObject(TaskService);
 
     await TestUtil.createPackage({
       name: pkgName,
@@ -56,4 +65,25 @@ describe('test/core/service/PackageSyncerService/createTask.test.ts', () => {
     });
     assert(task);
   });
+
+  it('should create task when processing', async () => {
+    mock(PackageSyncerService.prototype, 'executeTask', async (task: Task) => {
+      task.state = TaskState.Processing;
+      await taskRepository.saveTask(task);
+      await setTimeout(2);
+      await taskService.finishTask(task, TaskState.Success);
+    });
+    const task = await packageSyncerService.createTask(pkgName);
+    const res = await Promise.all([ packageSyncerService.executeTask(task), (async () => {
+      await setTimeout(1);
+      return await packageSyncerService.createTask(pkgName);
+    })() ]);
+    assert(res[1].taskId !== task.taskId);
+  });
+
+  it('should not duplicate task when waiting', async () => {
+    const task = await packageSyncerService.createTask(pkgName);
+    const newTask = await packageSyncerService.createTask(pkgName);
+    assert(newTask.taskId === task.taskId);
+  });
 });

test/port/controller/PackageSyncController/createSyncTask.test.ts

@@ -5,7 +5,6 @@ import { app, mock } from 'egg-mock/bootstrap';
 import { TestUtil } from 'test/TestUtil';
 import { Task as TaskModel } from 'app/repository/model/Task';
 import { PackageSyncerService } from 'app/core/service/PackageSyncerService';
-import { TaskState } from 'app/common/enum/Task';
 
 describe('test/port/controller/PackageSyncController/createSyncTask.test.ts', () => {
   let publisher: any;
@@ -286,7 +285,7 @@ describe('test/port/controller/PackageSyncController/createSyncTask.test.ts', ()
       assert(res.body.id === firstTaskId);
     });
 
-    it('should dont create exists processing task update less than 1 min', async () => {
+    it('should dont create exists waiting task', async () => {
       let res = await app.httpRequest()
         .put('/-/package/koa/syncs')
         .expect(201);
@@ -295,13 +294,12 @@ describe('test/port/controller/PackageSyncController/createSyncTask.test.ts', ()
       assert(res.body.id);
       const firstTaskId = res.body.id;
 
-      await TaskModel.update({ taskId: firstTaskId }, { state: TaskState.Processing });
       // again dont create
       res = await app.httpRequest()
         .put('/-/package/koa/syncs')
         .expect(201);
       assert(res.body.ok === true);
-      assert(res.body.state === 'processing');
+      assert(res.body.state === 'waiting');
       assert(res.body.id === firstTaskId);
 
       // update bigger than 1 min, same task return
@@ -310,7 +308,7 @@ describe('test/port/controller/PackageSyncController/createSyncTask.test.ts', ()
         .put('/-/package/koa/syncs')
         .expect(201);
       assert(res.body.ok === true);
-      assert(res.body.state === 'processing');
+      assert(res.body.state === 'waiting');
       assert(res.body.id === firstTaskId);
     });
   });