fix: handle ConflictError in JobHandler and add corresponding tests

src/job/models/jobHandler.ts

@@ -5,6 +5,7 @@ import type { LayerNameFormats } from '@map-colonies/raster-shared';
 import { TaskHandler as QueueClient } from '@map-colonies/mc-priority-queue';
 import { SpanStatusCode } from '@opentelemetry/api';
 import type { Logger } from '@map-colonies/js-logger';
+import { ConflictError } from '@map-colonies/error-types';
 import type { IConfig, JobAndTaskTelemetry, PollingConfig, StepKey } from '../../common/interfaces';
 import { JobTrackerClient } from '../../httpClients/jobTrackerClient';
 
@@ -70,15 +71,14 @@ export class JobHandler {
     const taskAttempts = task.attempts + 1; // rejecting the task increments the attempts
     logger.info({ msg: 'task attempts', taskAttempts, maxAttempts: this.pollingConfig.maxTaskAttempts });
     const reachedMaxAttempts = taskAttempts >= this.pollingConfig.maxTaskAttempts;
-    const isRecoverable = !(error instanceof ZodError) && !reachedMaxAttempts;
-
-    await this.queueClient.reject(job.id, task.id, isRecoverable, reason);
-
-    logger.error({ msg, reason, error, reachedMaxAttempts, isRecoverable });
+    const isRecoverable = !(error instanceof ZodError) && !(error instanceof ConflictError) && !reachedMaxAttempts;
     taskTracker?.failure(errName);
     tracingSpan?.setStatus({ code: SpanStatusCode.ERROR, message: reason });
     tracingSpan?.recordException(error instanceof Error ? error : new Error(reason));
 
+    logger.error({ msg, reason, error, reachedMaxAttempts, isRecoverable });
+    await this.queueClient.reject(job.id, task.id, isRecoverable, reason);
+
     if (!isRecoverable) {
       await this.jobTrackerClient.notify(task);
     }

tests/unit/job/jobHnadler/jobHandler.spec.ts

@@ -1,4 +1,5 @@
 import { ZodError } from 'zod';
+import { ConflictError } from '@map-colonies/error-types';
 import { JobAndTaskTelemetry } from '../../../../src/common/interfaces';
 import { registerDefaultConfig } from '../../mocks/configMock';
 import { ingestionNewJob } from '../../mocks/jobsMockData';
@@ -39,5 +40,20 @@ describe('JobHandler', () => {
       expect(jobTrackerClientMock.notify).toHaveBeenCalledWith(task);
       /* eslint-enable @typescript-eslint/unbound-method */
     });
+
+    it('should handle ConflictError as unrecoverable error', async () => {
+      const { newJobHandler, queueClientMock, jobTrackerClientMock } = setupJobHandlerTest();
+      const job = ingestionNewJob;
+      const task = createTasksTaskForIngestionNew;
+      const telemetry: JobAndTaskTelemetry = { taskTracker: undefined, tracingSpan: undefined };
+      const error = new ConflictError('Job already aborted');
+
+      await newJobHandler['handleError'](error, job, task, telemetry);
+
+      /* eslint-disable @typescript-eslint/unbound-method */
+      expect(queueClientMock.reject).toHaveBeenCalledWith(job.id, task.id, false, error.message);
+      expect(jobTrackerClientMock.notify).toHaveBeenCalledWith(task);
+      /* eslint-enable @typescript-eslint/unbound-method */
+    });
   });
 });

tests/unit/task/tileMergeTaskManager/tileMergeTaskManager.spec.ts

@@ -2,6 +2,7 @@
 import { randomUUID } from 'crypto';
 import nock from 'nock';
 import { TileOutputFormat } from '@map-colonies/raster-shared';
+import { ConflictError } from '@map-colonies/error-types';
 import { configMock, registerDefaultConfig } from '../../mocks/configMock';
 import { ingestionNewJob } from '../../mocks/jobsMockData';
 import type { MergeTaskParameters, JobResumeState, MergeTilesTaskParams } from '../../../../src/common/interfaces';
@@ -477,5 +478,30 @@ describe('tileMergeTaskManager', () => {
       // Should reject when HTTP requests fail
       await expect(tileMergeTaskManager.pushTasks(mockInitTask, jobId, ingestionNewJob.type, tasks)).rejects.toThrow();
     });
+
+    it('should handle ConflictError during batch publishing and stop immediately', async () => {
+      const { tileMergeTaskManager } = testContext;
+
+      const jobId = randomUUID();
+
+      // Setup HTTP mock to return 409 Conflict
+      const jobManagerBaseUrl = configMock.get<string>('jobManagement.config.jobManagerBaseUrl');
+      const createTasksPath = `/jobs/${jobId}/tasks`;
+
+      nock(jobManagerBaseUrl).post(createTasksPath).reply(409, { message: 'Job already aborted' }).persist();
+
+      const tasks = tileMergeTaskManager.buildTasks(buildTasksParams, mockInitTask);
+
+      // Spy on enqueueTasks to verify it's only called once (stops after conflict)
+      const enqueueTasksSpy = jest.spyOn(tileMergeTaskManager as unknown as { enqueueTasks: jest.Func }, 'enqueueTasks');
+
+      // Should throw ConflictError when job manager returns 409
+      await expect(tileMergeTaskManager.pushTasks(mockInitTask, jobId, ingestionNewJob.type, tasks)).rejects.toThrow(ConflictError);
+
+      // Verify that enqueueTasks was called (at least once before the conflict)
+      expect(enqueueTasksSpy).toHaveBeenCalled();
+
+      enqueueTasksSpy.mockRestore();
+    });
   });
 });