diff --git a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts index cedf29901..ae0b5df72 100644 --- a/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts +++ b/server/src/service/infrastructure/JobQueues/LessSimpleQueue.ts @@ -27,6 +27,7 @@ export class LessSimpleQueue implements IJobQueue { private monitorsRepository: IMonitorsRepository; private workersRepository: IQueueWorkersRepository; private readonly scheduler: Scheduler; + private queueMode: QueueMode; private heartbeatListener: ((workerId: string) => void) | null = null; constructor( @@ -34,13 +35,15 @@ export class LessSimpleQueue implements IJobQueue { helper: IQueueHelper, monitorsRepository: IMonitorsRepository, workersRepository: IQueueWorkersRepository, - scheduler: Scheduler + scheduler: Scheduler, + queueMode: QueueMode ) { this.logger = logger; this.helper = helper; this.monitorsRepository = monitorsRepository; this.workersRepository = workersRepository; this.scheduler = scheduler; + this.queueMode = queueMode; this.registerListeners(); } @@ -137,7 +140,7 @@ export class LessSimpleQueue implements IJobQueue { const scheduler = new Scheduler(store, { concurrency: 50, }); - const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler); + const instance = new LessSimpleQueue(logger, helper, monitorsRepository, workersRepository, scheduler, queueMode); await instance.init(queueMode); return instance; @@ -422,8 +425,14 @@ export class LessSimpleQueue implements IJobQueue { }; flushQueues = async () => { - const stopRes = await this.scheduler.stop(); - const flushRes = await this.scheduler.flushJobs(); + if (this.queueMode !== "primary") { + // Don't allow workers to flush, though this isn't currently reachable since workers have no API endpoints + this.logger.warn({ message: "Ignoring flush on non-primary worker", service: SERVICE_NAME, method: "flushQueues" }); + return { success: false }; + } + + const flushRes = await this.scheduler.flushJobs(); // Less Simple Queue must flush before stopping as it needs DB + const stopRes = await this.scheduler.stop(); // Destructive, other workers will abort their jobs const initRes = await this.init(); return { success: Boolean(stopRes && flushRes && initRes),