Egress runtime and retry

Service surface

Egress uses globalPrefix: /egress (port 3002 in dev). Important routes:

Kafka-driven execution

HandleIngressJobs (apps/egress-service/src/jobs/ingress.ts) subscribes to:

• dip.v2.*.regular
• dip.v2.*.priority
• dip.v2.*.retry

with bulkProcess: true. Messages carry data.jobId. The handler batches work (respecting DipV2Common.BATCH_SIZE), then Promise.allSettled → taskExecutorService.executeJob(jobId) for each valid id.

So async ingress and retry topics converge on the same executor path as sync.

TaskExecutorService

apps/egress-service/src/tasks/task-executor.ts injects TreeExecutionEngine and a map of TaskExecutor implementations by task type string. executeJob(jobId) loads the job and runs the tree (see Task types reference).

Sync execution (SyncExecutionService.executeAndSummarize)

Used by POST /egress/internal/sync/execute (SyncExecuteDto: jobIds: string[]):

1. Loads jobs by id; filters to PENDING.
2. Runs executeJob for each pending job in parallel (Promise.allSettled).
3. Reloads jobs and all dip job tasks, transforms via JobTransformer, returns { jobs } for ingress to serialize.

Independent jobs run in parallel; within a job, the tree engine defines order.

Retry API (POST /egress/retry)

Body: RetryDto — jobIds (Mongo ObjectId[]). RetryService calls executeJob for each id in parallel (Promise.allSettled) and returns a success message.

The DIP console uses this from EventJobsGrid / connector jobs (see frontend retryJobsByIds).

Job statuses

Execution outcomes update DipJobsModel.status (DipJobStatusEnum). See Event log and jobs for the full list. isRetryable and maxRetry on the job (from the connector–event mapping) interact with cron/retry topics — tune at mapping level.

Related

• Ingress async and sync — when sync calls egress and timeout behaviour.