.vscode/settings.json

@@ -1,5 +1,7 @@
 {
   "cSpell.words": [
-    "Repos"
+    "Repos",
+    "lpush",
+    "rpop"
   ]
 }

src/pipeline-tasks/pipeline-tasks.service.spec.ts

@@ -6,38 +6,161 @@ import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
 import { getQueueToken } from '@nestjs/bull';
 import { RedisService } from 'nestjs-redis';
 import { Pipeline } from '../pipelines/pipeline.entity';
+import { EntityNotFoundError } from 'typeorm/error/EntityNotFoundError';
+import { Repository } from 'typeorm';
+import { Queue } from 'bull';
+import { LockFailedException } from '../commons/exceptions/lock-failed.exception';
 
 describe('PipelineTasksService', () => {
   let service: PipelineTasksService;
+  let module: TestingModule;
+  let taskRepository: Repository<PipelineTask>;
+  let pipelineRepository: Repository<Pipeline>;
+  const getBasePipeline = () =>
+    ({
+      id: 'test',
+      name: '测试流水线',
+      branch: 'master',
+      workUnitMetadata: [],
+    } as Pipeline);
+  let redisClient;
+  let taskQueue: Queue;
+  const getTask = () =>
+    ({
+      pipelineId: 'test',
+      commit: 'test',
+      units: [],
+    } as PipelineTask);
 
   beforeEach(async () => {
-    const module: TestingModule = await Test.createTestingModule({
+    redisClient = (() => ({
+      set: jest.fn().mockImplementation(async () => 'OK'),
+      incr: jest.fn().mockImplementation(async () => 1),
+      decr: jest.fn().mockImplementation(async () => 0),
+      lpush: jest.fn().mockImplementation(async () => 1),
+      rpop: jest.fn().mockImplementation(async () => JSON.stringify(getTask())),
+    }))() as any;
+    taskQueue = (() => ({
+      add: jest.fn().mockImplementation(async () => null),
+    }))() as any;
+    module = await Test.createTestingModule({
       providers: [
         PipelineTasksService,
         {
           provide: getRepositoryToken(PipelineTask),
-          useValue: {},
+          useValue: new Repository(),
         },
-        PipelineTasksService,
         {
           provide: getRepositoryToken(Pipeline),
-          useValue: {},
+          useValue: new Repository(),
         },
         {
           provide: getQueueToken(PIPELINE_TASK_QUEUE),
-          useValue: {},
+          useValue: taskQueue,
         },
         {
           provide: RedisService,
-          useValue: {},
+          useValue: {
+            getClient: jest.fn(() => redisClient),
+          },
         },
       ],
     }).compile();
 
     service = module.get<PipelineTasksService>(PipelineTasksService);
+    taskRepository = module.get(getRepositoryToken(PipelineTask));
+    pipelineRepository = module.get(getRepositoryToken(Pipeline));
+    jest
+      .spyOn(taskRepository, 'save')
+      .mockImplementation(async (data: any) => data);
+    jest
+      .spyOn(taskRepository, 'create')
+      .mockImplementation((data: any) => data);
   });
 
   it('should be defined', () => {
     expect(service).toBeDefined();
   });
+
+  describe('addTask', () => {
+    beforeEach(() => {
+      jest
+        .spyOn(pipelineRepository, 'findOneOrFail')
+        .mockImplementation(async () => getBasePipeline());
+    });
+    it('pipeline not found', async () => {
+      jest.spyOn(taskRepository, 'findOneOrFail').mockImplementation(() => {
+        throw new EntityNotFoundError(Pipeline, {});
+      });
+      await expect(
+        service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
+      ).rejects;
+    });
+    it('create task on db', async () => {
+      const save = jest
+        .spyOn(taskRepository, 'save')
+        .mockImplementation(async (data: any) => data);
+      await service.addTask({ pipelineId: 'test', commit: 'test', units: [] }),
+        expect(save.mock.calls[0][0]).toMatchObject({
+          pipelineId: 'test',
+          commit: 'test',
+          units: [],
+        });
+    });
+    it('add task', async () => {
+      const lpush = jest.spyOn(redisClient, 'lpush');
+      const doNextTask = jest.spyOn(service, 'doNextTask');
+      await service.addTask({ pipelineId: 'test', commit: 'test', units: [] });
+      expect(typeof lpush.mock.calls[0][1] === 'string').toBeTruthy();
+      expect(JSON.parse(lpush.mock.calls[0][1] as string)).toMatchObject({
+        pipelineId: 'test',
+        commit: 'test',
+        units: [],
+        pipeline: getBasePipeline(),
+      });
+      expect(doNextTask).toHaveBeenCalledWith(getBasePipeline());
+    });
+  });
+
+  describe('doNextTask', () => {
+    it('add task to queue', async () => {
+      const decr = jest.spyOn(redisClient, 'decr');
+      const rpop = jest.spyOn(redisClient, 'rpop');
+      const add = jest.spyOn(taskQueue, 'add');
+
+      await service.doNextTask(getBasePipeline());
+
+      expect(add).toHaveBeenCalledWith(getTask());
+      expect(decr).toHaveBeenCalledTimes(1);
+      expect(rpop).toHaveBeenCalledTimes(1);
+    });
+    it('pipeline is busy', async () => {
+      let remainTimes = 3;
+
+      const incr = jest
+        .spyOn(redisClient, 'incr')
+        .mockImplementation(() => remainTimes--);
+      const rpop = jest.spyOn(redisClient, 'rpop');
+      const decr = jest.spyOn(redisClient, 'decr');
+      const add = jest.spyOn(taskQueue, 'add');
+
+      await service.doNextTask(getBasePipeline());
+
+      expect(rpop).toHaveBeenCalledTimes(1);
+      expect(incr).toHaveBeenCalledTimes(3);
+      expect(decr).toHaveBeenCalledTimes(3);
+      expect(add).toHaveBeenCalledWith(getTask());
+    });
+    it('pipeline always busy and timeout', async () => {
+      const incr = jest.spyOn(redisClient, 'incr').mockImplementation(() => 3);
+      const decr = jest.spyOn(redisClient, 'decr');
+
+      await expect(
+        service.doNextTask(getBasePipeline()),
+      ).rejects.toBeInstanceOf(LockFailedException);
+
+      expect(decr).toHaveBeenCalledTimes(5);
+      expect(incr).toHaveBeenCalledTimes(5);
+    }, 15_000);
+  });
 });

src/pipeline-tasks/pipeline-tasks.service.ts

@@ -8,6 +8,7 @@ import { Pipeline } from '../pipelines/pipeline.entity';
 import { InjectQueue } from '@nestjs/bull';
 import { PIPELINE_TASK_QUEUE } from './pipeline-tasks.constants';
 import { Queue } from 'bull';
+import { LockFailedException } from '../commons/exceptions/lock-failed.exception';
 
 @Injectable()
 export class PipelineTasksService {
@@ -30,26 +31,34 @@ export class PipelineTasksService {
 
     const [lckKey, tasksKey] = this.getRedisTokens(pipeline);
     const redis = this.redis.getClient();
-    await redis.set(lckKey, 0, 'EX', 10, 'NX');
-    const lckSemaphore = await redis.incr(lckKey);
-    if (lckSemaphore > 1) {
-      await this.redis
-        .getClient()
-        .lpush(tasksKey, JSON.stringify(task))
-        .finally(() => {
-          return redis.decr(lckKey);
-        });
-    } else {
-      this.queue.add(task);
-    }
+    await redis.lpush(tasksKey, JSON.stringify(task)).finally(() => {
+      return redis.decr(lckKey);
+    });
+    await this.doNextTask(pipeline);
   }
 
   async doNextTask(pipeline: Pipeline) {
-    const tasksKey = this.getRedisTokens(pipeline)[1];
+    const [lckKey, tasksKey] = this.getRedisTokens(pipeline);
     const redis = this.redis.getClient();
-    const task = JSON.parse((await redis.rpop(tasksKey)) ?? 'null');
+    await redis.set(lckKey, 0, 'EX', 10, 'NX');
+
+    await new Promise(async (resolve, reject) => {
+      for (let i = 0; i < 5; i++) {
+        if ((await redis.incr(lckKey)) === 1) {
+          resolve(undefined);
+          return;
+        }
+        await redis.decr(lckKey);
+        await new Promise((resolve) => setTimeout(resolve, 2000));
+      }
+      reject(new LockFailedException(lckKey));
+    });
+
+    const task = JSON.parse(
+      (await redis.rpop(tasksKey).finally(() => redis.decr(lckKey))) ?? 'null',
+    );
     if (task) {
-      this.queue.add(task);
+      await this.queue.add(task);
     }
   }
 

src/repos/repos.service.spec.ts

@@ -59,19 +59,19 @@ describe('ReposService', () => {
   it('getWorkspaceRoot', () => {
     expect(service.getWorkspaceRoot(getTest1Project())).toBeDefined();
   });
-  describe('listLogs', () => {
+  describe.skip('listLogs', () => {
     it('should be return logs', async () => {
       const result = await service.listLogs({ projectId: '1' });
       expect(result).toBeDefined();
     }, 20_000);
   });
-  describe('listBranch', () => {
+  describe.skip('listBranch', () => {
     it('should be return branches', async () => {
       const result = await service.listBranches({ projectId: '1' });
       expect(result).toBeDefined();
     }, 10_000);
   });
-  describe('checkoutBranch', () => {
+  describe.skip('checkoutBranch', () => {
     it('should be checkout', async () => {
       await service.checkoutBranch(getTest1Project(), 'master');
       const filePath = join(
@@ -108,7 +108,7 @@ describe('ReposService', () => {
     }, 30_000);
   });
 
-  describe('checkoutCommit', () => {
+  describe.skip('checkoutCommit', () => {
     it('should be checkout', async () => {
       await service.checkoutCommit(getTest1Project(), '498c782685');
       const filePath = join(