fix(pipeline-tasks): 修复日志有概率丢失最后几条的问题

日志记录结束前，任务信息已入库存档。

src/pipeline-tasks/pipeline-task.consumer.spec.ts

@@ -223,11 +223,9 @@ describe('PipelineTaskConsumer', () => {
         update: jest.fn().mockImplementation(() => undefined),
       } as unknown) as Job;
 
-      const runScript = jest
+      jest.spyOn(consumer, 'runScript').mockImplementation(async () => {
-        .spyOn(consumer, 'runScript')
+        throw new Error('bad message');
-        .mockImplementation(async () => {
+      });
-          throw new Error('bad message');
-        });
       const updateTask = jest.spyOn(tasksService, 'updateTask');
 
       await consumer.doTask(job);

src/pipeline-tasks/pipeline-task.consumer.ts

@@ -113,19 +113,30 @@ export class PipelineTaskConsumer {
         shell: true,
         cwd: workspaceRoot,
       });
+      let loggingCount = 0; // semaphore
+
       sub.stderr.on('data', (data: Buffer) => {
         const str = data.toString();
-        this.logsService.recordLog(
+        loggingCount++;
-          PipelineTaskLogMessage.create(task, unit, str, true),
+        this.logsService
-        );
+          .recordLog(PipelineTaskLogMessage.create(task, unit, str, true))
+          .finally(() => loggingCount--);
       });
       sub.stdout.on('data', (data: Buffer) => {
         const str = data.toString();
-        this.logsService.recordLog(
+        loggingCount++;
-          PipelineTaskLogMessage.create(task, unit, str, false),
+        this.logsService
-        );
+          .recordLog(PipelineTaskLogMessage.create(task, unit, str, false))
+          .finally(() => loggingCount--);
       });
-      sub.addListener('close', (code) => {
+      sub.addListener('close', async (code) => {
+        await new Promise<void>(async (resolve) => {
+          for (let i = 0; i < 10 && loggingCount > 0; i++) {
+            await new Promise((resolve) => setTimeout(resolve, 500));
+            log('waiting logging... (%dx500ms)', i);
+          }
+          resolve();
+        });
         if (code === 0) {
           return resolve();
         }