single worker ID

This commit is contained in:
Alex Holliday committed 2026-06-15 12:21:36 -07:00
1 parent 7ce6068757
commit abf6dc2908
3 files changed
+14 -12

No files matched your search

+7 -3
View File
@@ -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
@@ -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 {
+6 -6
View File
@@ -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<JobType, number> = {
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<JobType | "heartbeat", NodeJS.Timeout>();
private inFlight: Record<JobType, number> = {
@@ -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<DBQueueWorker> {
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;