All files / core/src/jobs jobs.usecases.ts

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
import type { TaskDefinitionRegistry } from '../tasks/task-definition.types';
import type { Job } from './jobs.types';
 
export async function processJob({ job, taskDefinitionRegistry }: { job: Job; taskDefinitionRegistry: TaskDefinitionRegistry }) {
  const taskDefinition = taskDefinitionRegistry.get(job.taskName);
 
  const maxRetries = job.maxRetries ?? taskDefinition.options?.maxRetries ?? 0;
  const attempts = maxRetries + 1;
 
  // Using two variables to track the failure state and the last error
  // to handle falsy errors
  let hasFailed = false;
  let lastError: unknown | undefined;
 
  for (let i = 0; i < attempts; i++) {
    try {
      const result = await taskDefinition.handler({ data: job.data, context: { workerId: job.id } });
      return result;
    } catch (error) {
      hasFailed = true;
      lastError = error;
    }
  }
 
  if (hasFailed) {
    throw lastError;
  }
}