From abf6dc29080d1c0d9e69a06ecfaa89d8b7ee5426 Mon Sep 17 00:00:00 2001 From: Alex Holliday Date: Mon, 15 Jun 2026 12:21:36 -0700 Subject: [PATCH] single worker ID --- server/src/config/services.ts | 10 +++++++--- server/src/domain/jobs/job.repository.mongo.ts | 4 +--- server/src/worker/worker.db-queue.ts | 12 ++++++------ 3 files changed, 14 insertions(+), 12 deletions(-) diff --git a/server/src/config/services.ts b/server/src/config/services.ts index 2afd263c9..092471b8c 100644 --- a/server/src/config/services.ts +++ b/server/src/config/services.ts @@ -1,4 +1,6 @@ import { Mongoose } from "mongoose"; +import { hostname } from "node:os"; +import { randomUUID } from "node:crypto"; import MongoDB from "../db/db.mongo.js"; import { IDb } from "@/db/db.interface.js"; import { EnvConfig, ISettingsService, SettingsService } from "@/domain/app-settings/app-settings.service.js"; @@ -98,7 +100,6 @@ import MongoTeamsRepository from "@/domain/teams/team.repository.model.js"; import { IUsersRepository } from "@/domain/users/user.repository.interface.js"; import MongoUsersRepository from "@/domain/users/user.repository.mongo.js"; import { ILogger } from "@/utils/logger.js"; -import { type QueueMode } from "@/domain/app-settings/app-settings.type.js"; import { NotificationReactor } from "@/worker/reactors/reactor.notification.js"; import { ReactorDispatcher } from "@/worker/reactors/reactor.dispatcher.js"; import { IncidentReactor } from "@/worker/reactors/reactor.incident.js"; @@ -177,7 +178,9 @@ export const initializeServices = async ({ const incidentsRepository = new MongoIncidentsRepository(); const teamsRepository = new MongoTeamsRepository(); const maintenanceWindowsRepository = new MongoMaintenanceWindowsRepository(); - const jobsRepository = new MongoJobsRepository(); + // One identity per process + const workerId = `${hostname()}:${process.pid}:${randomUUID()}`; + const jobsRepository = new MongoJobsRepository(workerId); const queueWorkersRepository = new MongoQueueWorkersRepository(); // Inject settings repository into settings service (now that DB is connected) @@ -305,7 +308,8 @@ export const initializeServices = async ({ workerHelper, queueWorkersRepository, envSettings.queueMode, - envSettings.queuePrimaryProcesses + envSettings.queuePrimaryProcesses, + workerId ); // Business services diff --git a/server/src/domain/jobs/job.repository.mongo.ts b/server/src/domain/jobs/job.repository.mongo.ts index 7b409c911..f7609ba01 100644 --- a/server/src/domain/jobs/job.repository.mongo.ts +++ b/server/src/domain/jobs/job.repository.mongo.ts @@ -1,11 +1,9 @@ import { IJobsRepository, JobPageQuery } from "@/domain/jobs/job.repository.interface.js"; import JobModel, { JobDocument } from "@/domain/jobs/job.model.js"; import { JobType, Job, JobSeed, BACKOFF_MS, LOCK_MS, PARKED, jobId } from "@/domain/jobs/job.type.js"; -import { hostname } from "node:os"; -import { randomUUID } from "node:crypto"; class MongoJobsRepository implements IJobsRepository { - private readonly workerId = `${hostname()}:${process.pid}:${randomUUID()}`; + constructor(private readonly workerId: string) {} private toEntity(doc: JobDocument): Job { return { diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index 3d106445b..d8c997453 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -1,8 +1,6 @@ import { Monitor, supportsGeoCheck } from "@/domain/monitors/monitor.types.js"; import { IWorker, WorkerJobsPagination, WorkerJobSummary, WorkerMetrics } from "@/worker/worker.interface.js"; import { type Job, type JobSeed, type JobType, jobId, LOCK_MS } from "@/domain/jobs/job.type.js"; -import { randomUUID } from "node:crypto"; -import { hostname } from "node:os"; import { ILogger } from "@/utils/logger.js"; import { IJobsRepository } from "@/domain/jobs/job.repository.interface.js"; import { IMonitorsRepository } from "@/domain/monitors/monitor.repository.interface.js"; @@ -32,7 +30,6 @@ const CONCURRENCY: Record = { export class DBQueueWorker implements IWorker { static SERVICE_NAME = SERVICE_NAME; - private readonly workerId = `${hostname()}:${process.pid}:${randomUUID().slice(0, 8)}`; private stopped = false; private timers = new Map(); private inFlight: Record = { @@ -56,7 +53,8 @@ export class DBQueueWorker implements IWorker { private helper: IWorkerHelper, private queueWorkersRepository: IQueueWorkersRepository, private queueMode: QueueMode, - private queuePrimaryProcesses: boolean + private queuePrimaryProcesses: boolean, + private readonly workerId: string // same id MongoJobsRepository stamps into lockedBy ) {} get serviceName() { @@ -76,7 +74,8 @@ export class DBQueueWorker implements IWorker { helper: IWorkerHelper, queueWorkersRepository: IQueueWorkersRepository, queueMode: QueueMode, - queuePrimaryProcesses: boolean + queuePrimaryProcesses: boolean, + workerId: string ): Promise { const instance = new DBQueueWorker( logger, @@ -91,7 +90,8 @@ export class DBQueueWorker implements IWorker { helper, queueWorkersRepository, queueMode, - queuePrimaryProcesses + queuePrimaryProcesses, + workerId ); await instance.init(); return instance;