fix: complete cache deletion jobs only after overseer finished creating tasks (MAPCO-11779)

package.json

@@ -51,7 +51,7 @@
     "@map-colonies/mc-utils": "^5.1.0",
     "@map-colonies/openapi-express-viewer": "^3.0.0",
     "@map-colonies/prometheus": "^1.0.0",
-    "@map-colonies/raster-shared": "^8.3.0",
+    "@map-colonies/raster-shared": "9.0.0",
     "@map-colonies/read-pkg": "0.0.1",
     "@map-colonies/schemas": "^1.18.0",
     "@map-colonies/telemetry": "^6.0.0",

src/tasks/handlers/deleteCache/deleteCacheHandler.ts

@@ -1,6 +1,8 @@
 import type { Logger } from '@map-colonies/js-logger';
 import type { IJobResponse, ITaskResponse, JobManagerClient } from '@map-colonies/mc-priority-queue';
 import { injectable, inject } from 'tsyringe';
+import { BadRequestError } from '@map-colonies/error-types';
+import { cacheDeletionJobParamsSchema } from '@map-colonies/raster-shared';
 import type { ConfigType } from '@src/common/config';
 import type { TaskTypes } from '../../../common/interfaces';
 import { SERVICES } from '../../../common/constants';
@@ -27,7 +29,19 @@ export class DeleteCacheJobHandler extends JobHandler {
     this.initializeTaskOperations();
   }
 
+  /**
+   * Overseer streams the tasks onto the job in batches, so all known tasks being completed is not enough -
+   * the job is completed only once overseer flags that it finished creating tasks.
+   */
   public override isJobCompleted = (): boolean => {
-    return this.job.completedTasks === this.job.taskCount;
+    const result = cacheDeletionJobParamsSchema.safeParse(this.job.parameters);
+
+    if (!result.success) {
+      const errorMessage = `Failed to parse cache deletion job parameters: ${result.error.message}`;
+      this.logger.error({ msg: errorMessage, jobId: this.job.id });
+      throw new BadRequestError(errorMessage);
+    }
+
+    return this.job.completedTasks === this.job.taskCount && result.data.tasksCreationCompleted;
   };
 }

tests/integration/tasks/deleteCache/taskManager.spec.ts

@@ -7,7 +7,7 @@ import { initConfig } from '../../../../src/common/config';
 import { configMock } from '../../../mocks/configMock';
 import { getApp } from '../../../../src/app';
 import type { IJobManagerConfig, IJobDefinitionsConfig } from '../../../../src/common/interfaces';
-import { getDeleteCacheJobMock, getTaskMock } from '../../../mocks/jobMocks';
+import { getCacheDeletionJobParamsMock, getDeleteCacheJobMock, getTaskMock } from '../../../mocks/jobMocks';
 import { calculateJobPercentage } from '../../../../src/utils/jobUtils';
 import { SERVICES } from '../../../../src/common/constants';
 import { registerExternalValues } from '../../../../src/containerConfig';
@@ -91,6 +91,30 @@ describe('tasks', function () {
       expect(response).toSatisfyApiSpec();
     });
 
+    it.each(deleteCacheJobTypes)(
+      'should return 200 and only update progress when all tasks are completed but tasks creation is not completed - $jobTypeKey',
+      async ({ jobTypeKey }) => {
+        // mocks
+        const mockJob = getDeleteCacheJobMock(jobDefinitionsConfig.jobs[jobTypeKey], {
+          completedTasks: 5,
+          taskCount: 5,
+          parameters: getCacheDeletionJobParamsMock({ tasksCreationCompleted: false }),
+        });
+        const mockTask = getTaskMock(mockJob.id, { type: jobDefinitionsConfig.tasks.tilesDeletion, status: OperationStatus.COMPLETED });
+        nock(jobManagerConfigMock.jobManagerBaseUrl).post('/tasks/find', { id: mockTask.id }).reply(httpStatusCodes.OK, [mockTask]);
+        nock(jobManagerConfigMock.jobManagerBaseUrl)
+          .get(`/jobs/${mockJob.id}`)
+          .query({ shouldReturnTasks: false })
+          .reply(httpStatusCodes.OK, mockJob);
+        nock(jobManagerConfigMock.jobManagerBaseUrl).put(`/jobs/${mockJob.id}`, { percentage: 100 }).reply(httpStatusCodes.OK);
+        // action
+        const response = await requestSender.handleTaskNotification(mockTask.id);
+        // expectation
+        expect(response.status).toBe(httpStatusCodes.OK);
+        expect(response).toSatisfyApiSpec();
+      }
+    );
+
     it.each(deleteCacheJobTypes)(
       'should return 200 and fail the job when getting failed tiles-deletion task - $jobTypeKey',
       async ({ jobTypeKey }) => {
@@ -112,4 +136,25 @@ describe('tasks', function () {
       }
     );
   });
+
+  describe('Bad Path', function () {
+    it.each(deleteCacheJobTypes)(
+      'should return 400 and not complete the job when job parameters are invalid - $jobTypeKey',
+      async ({ jobTypeKey }) => {
+        // mocks
+        const mockJob = getDeleteCacheJobMock(jobDefinitionsConfig.jobs[jobTypeKey], { completedTasks: 5, taskCount: 5, parameters: {} });
+        const mockTask = getTaskMock(mockJob.id, { type: jobDefinitionsConfig.tasks.tilesDeletion, status: OperationStatus.COMPLETED });
+        nock(jobManagerConfigMock.jobManagerBaseUrl).post('/tasks/find', { id: mockTask.id }).reply(httpStatusCodes.OK, [mockTask]);
+        nock(jobManagerConfigMock.jobManagerBaseUrl)
+          .get(`/jobs/${mockJob.id}`)
+          .query({ shouldReturnTasks: false })
+          .reply(httpStatusCodes.OK, mockJob);
+        // action
+        const response = await requestSender.handleTaskNotification(mockTask.id);
+        // expectation
+        expect(response.status).toBe(httpStatusCodes.BAD_REQUEST);
+        expect(response).toSatisfyApiSpec();
+      }
+    );
+  });
 });

tests/mocks/jobMocks.ts

@@ -1,6 +1,7 @@
 import type { IJobResponse, ITaskResponse } from '@map-colonies/mc-priority-queue';
 import { OperationStatus } from '@map-colonies/mc-priority-queue';
 import { faker } from '@faker-js/faker';
+import type { CacheDeletionJobParams } from '@map-colonies/raster-shared';
 import type { JobType } from '../../src/common/interfaces';
 
 export const createTestJob = (jobType: JobType, overrides?: Partial<IJobResponse<unknown, unknown>>): IJobResponse<unknown, unknown> => {
@@ -76,6 +77,10 @@ export const getExportJobMock = (override?: Partial<IJobResponse<unknown, unknow
   return { ...defaultJobMock, ...override };
 };
 
+export const getCacheDeletionJobParamsMock = (override?: Partial<CacheDeletionJobParams>): CacheDeletionJobParams => {
+  return { ingestionJobId: faker.string.uuid(), tasksCreationCompleted: true, ...override };
+};
+
 export const getDeleteCacheJobMock = (
   jobType: string = 'Update_Delete_Cache',
   override?: Partial<IJobResponse<unknown, unknown>>
@@ -92,7 +97,7 @@ export const getDeleteCacheJobMock = (
     domain: 'RASTER',
     isCleaned: false,
     priority: 0,
-    parameters: {},
+    parameters: getCacheDeletionJobParamsMock(),
     expirationDate: undefined,
     internalId: faker.string.uuid(),
     producerName: undefined,

tests/unit/tasks/handlers/baseJobHandler.spec.ts

@@ -25,8 +25,6 @@ describe('BaseJobHandler', () => {
     { mockJob: createTestJob(jobDefinitionsConfig.jobs.update) },
     { mockJob: createTestJob(jobDefinitionsConfig.jobs.swapUpdate) },
     { mockJob: createTestJob(jobDefinitionsConfig.jobs.export) },
-    { mockJob: createTestJob(jobDefinitionsConfig.jobs.updateCacheDeletion) },
-    { mockJob: createTestJob(jobDefinitionsConfig.jobs.swapCacheDeletion) },
   ];
 
   const testCaseHandlerLog = '$mockJob.type handler';
@@ -138,10 +136,6 @@ describe('BaseJobHandler', () => {
   });
 
   describe('isJobCompleted', () => {
-    const nonDeleteCacheTestCases = testCases.filter(
-      ({ mockJob }) => mockJob.type !== jobDefinitionsConfig.jobs.updateCacheDeletion && mockJob.type !== jobDefinitionsConfig.jobs.swapCacheDeletion
-    ); // removing deleteCache job test cases as finalize task type is not handled there
-
     it.each(testCases)(`should return true when all tasks are completed - ${testCaseHandlerLog}`, (testCase) => {
       let { mockJob } = testCase;
       mockJob = { ...mockJob, completedTasks: 10, taskCount: 10 };
@@ -174,7 +168,7 @@ describe('BaseJobHandler', () => {
       expect(result).toBe(false);
     });
 
-    it.each(nonDeleteCacheTestCases)(`should return false in case task type is not finalize - ${testCaseHandlerLog}`, (testCase) => {
+    it.each(testCases)(`should return false in case task type is not finalize - ${testCaseHandlerLog}`, (testCase) => {
       let { mockJob } = testCase;
       mockJob = { ...mockJob, completedTasks: 10, taskCount: 10 };
       mockTask = getTaskMock<unknown>(mockJob.id, {

tests/unit/tasks/handlers/deleteCacheHandler.spec.ts

@@ -0,0 +1,62 @@
+import { jsLogger, type Logger } from '@map-colonies/js-logger';
+import { OperationStatus } from '@map-colonies/mc-priority-queue';
+import { BadRequestError } from '@map-colonies/error-types';
+import type { IJobDefinitionsConfig } from '../../../../src/common/interfaces';
+import { getCacheDeletionJobParamsMock, getDeleteCacheJobMock, getTaskMock } from '../../../mocks/jobMocks';
+import { registerDefaultConfig, clear as clearConfig, configMock } from '../../../mocks/configMock';
+import { queueClientMock } from '../../../mocks/mockJobManager';
+import { getJobHandler } from '../../../../src/tasks/handlers/jobHandlerFactory';
+
+describe('DeleteCacheJobHandler', () => {
+  let mockLogger: Logger;
+  beforeAll(async () => {
+    mockLogger = await jsLogger({ enabled: false });
+  });
+
+  registerDefaultConfig();
+  const jobDefinitionsConfig = configMock.get('jobDefinitions') as IJobDefinitionsConfig;
+  const testCases = [{ jobType: jobDefinitionsConfig.jobs.updateCacheDeletion }, { jobType: jobDefinitionsConfig.jobs.swapCacheDeletion }];
+
+  const testCaseHandlerLog = '$jobType handler';
+
+  beforeEach(() => {
+    registerDefaultConfig();
+  });
+
+  afterEach(() => {
+    clearConfig();
+    jest.resetAllMocks();
+  });
+
+  describe('isJobCompleted', () => {
+    const getHandler = (jobType: string, completedTasks: number, taskCount: number, parameters: unknown): ReturnType<typeof getJobHandler> => {
+      const mockJob = getDeleteCacheJobMock(jobType, { completedTasks, taskCount, parameters });
+      const mockTask = getTaskMock<unknown>(mockJob.id, { type: jobDefinitionsConfig.tasks.tilesDeletion, status: OperationStatus.COMPLETED });
+      return getJobHandler(mockJob.type, jobDefinitionsConfig, mockLogger, queueClientMock, configMock, mockJob, mockTask);
+    };
+
+    it.each(testCases)(`should return true when all tasks are completed and tasks creation is completed - ${testCaseHandlerLog}`, ({ jobType }) => {
+      const handler = getHandler(jobType, 10, 10, getCacheDeletionJobParamsMock({ tasksCreationCompleted: true }));
+
+      expect(handler.isJobCompleted(jobDefinitionsConfig.tasks.tilesDeletion)).toBe(true);
+    });
+
+    it.each(testCases)(`should return false when not all of the tasks are completed - ${testCaseHandlerLog}`, ({ jobType }) => {
+      const handler = getHandler(jobType, 5, 10, getCacheDeletionJobParamsMock({ tasksCreationCompleted: true }));
+
+      expect(handler.isJobCompleted(jobDefinitionsConfig.tasks.tilesDeletion)).toBe(false);
+    });
+
+    it.each(testCases)(`should return false when all tasks are completed but tasks creation is not - ${testCaseHandlerLog}`, ({ jobType }) => {
+      const handler = getHandler(jobType, 10, 10, getCacheDeletionJobParamsMock({ tasksCreationCompleted: false }));
+
+      expect(handler.isJobCompleted(jobDefinitionsConfig.tasks.tilesDeletion)).toBe(false);
+    });
+
+    it.each(testCases)(`should throw BadRequestError when job parameters are invalid - ${testCaseHandlerLog}`, ({ jobType }) => {
+      const handler = getHandler(jobType, 10, 10, {});
+
+      expect(() => handler.isJobCompleted(jobDefinitionsConfig.tasks.tilesDeletion)).toThrow(BadRequestError);
+    });
+  });
+});

tests/unit/tasks/handlers/jobHandler.spec.ts

@@ -2,7 +2,7 @@ import { jsLogger } from '@map-colonies/js-logger';
 import type { Logger } from '@map-colonies/js-logger';
 import { OperationStatus } from '@map-colonies/mc-priority-queue';
 import type { IJobDefinitionsConfig } from '../../../../src/common/interfaces';
-import { createTestJob, getTaskMock } from '../../../mocks/jobMocks';
+import { createTestJob, getCacheDeletionJobParamsMock, getTaskMock } from '../../../mocks/jobMocks';
 import { registerDefaultConfig, clear as clearConfig, configMock } from '../../../mocks/configMock';
 import { mockJobManager, queueClientMock } from '../../../mocks/mockJobManager';
 import { getJobHandler } from '../../../../src/tasks/handlers/jobHandlerFactory';
@@ -22,8 +22,14 @@ describe('JobHandler', () => {
     { mockJob: createTestJob(jobDefinitionsConfig.jobs.update), taskType: jobDefinitionsConfig.tasks.merge },
     { mockJob: createTestJob(jobDefinitionsConfig.jobs.swapUpdate), taskType: jobDefinitionsConfig.tasks.merge },
     { mockJob: createTestJob(jobDefinitionsConfig.jobs.export), taskType: jobDefinitionsConfig.tasks.export },
-    { mockJob: createTestJob(jobDefinitionsConfig.jobs.updateCacheDeletion), taskType: jobDefinitionsConfig.tasks.tilesDeletion },
-    { mockJob: createTestJob(jobDefinitionsConfig.jobs.swapCacheDeletion), taskType: jobDefinitionsConfig.tasks.tilesDeletion },
+    {
+      mockJob: createTestJob(jobDefinitionsConfig.jobs.updateCacheDeletion, { parameters: getCacheDeletionJobParamsMock() }),
+      taskType: jobDefinitionsConfig.tasks.tilesDeletion,
+    },
+    {
+      mockJob: createTestJob(jobDefinitionsConfig.jobs.swapCacheDeletion, { parameters: getCacheDeletionJobParamsMock() }),
+      taskType: jobDefinitionsConfig.tasks.tilesDeletion,
+    },
     { mockJob: createTestJob(jobDefinitionsConfig.jobs.deleteLayer), taskType: jobDefinitionsConfig.tasks.delete },
   ];