diff --git a/server/src/domain/jobs/job.repository.interface.ts b/server/src/domain/jobs/job.repository.interface.ts index e431675a0..5a9bf901e 100644 --- a/server/src/domain/jobs/job.repository.interface.ts +++ b/server/src/domain/jobs/job.repository.interface.ts @@ -17,6 +17,9 @@ export interface IJobsRepository { // Atomically claim a job claimDue(type: JobType, now: number): Promise; + // locks up to `limit` due rows in one batch + claimDueBatch(type: JobType, limit: number, now: number): Promise; + // Completes a repeating job recordSuccess(id: string, nextScheduledAt: number, intervalMs: number, now: number): Promise; @@ -34,7 +37,7 @@ export interface IJobsRepository { // Job CRUD // ******************** - // Create jobs. Takes a JobSeed — the repo owns volatile lease/counter defaults on insert. + // Create jobs. Takes a JobSeed upsertJob(job: JobSeed): Promise; // Pause/Resume @@ -46,7 +49,7 @@ export interface IJobsRepository { // Delete, drop all rows deleteById(refId: string): Promise; - // Delete a single typed row for a monitor (e.g. drop just the geo row, keep check/evaluate) + // Delete a single job of type deleteByIdAndType(refId: string, type: JobType): Promise; // Get job diff --git a/server/src/domain/jobs/job.repository.mongo.ts b/server/src/domain/jobs/job.repository.mongo.ts index a3a80f221..a8c0a5d12 100644 --- a/server/src/domain/jobs/job.repository.mongo.ts +++ b/server/src/domain/jobs/job.repository.mongo.ts @@ -49,6 +49,33 @@ class MongoJobsRepository implements IJobsRepository { return doc ? this.toEntity(doc) : null; }; + claimDueBatch = async (type: JobType, limit: number, now: number) => { + if (limit <= 0) return []; + const dueAndFree = { + type, + isActive: true, + nextScheduledAt: { $lte: now }, // due + $or: [{ lockedUntil: null }, { lockedUntil: { $lt: now } }], // free or expired lock + }; + + // 1. Pick the oldest-due candidates + const candidates = await JobModel.find(dueAndFree).sort({ nextScheduledAt: 1 }).limit(limit).select({ _id: 1 }).lean<{ _id: string }[]>(); + if (candidates.length === 0) return []; + const ids = candidates.map((d) => d._id); + + // 2. lock them. Re-check free/expired so we never steal another worker's live lock + // only the rows this updateMany actually flips end up with claimed + const lockedUntil = now + LOCK_MS; + await JobModel.updateMany( + { _id: { $in: ids }, $or: [{ lockedUntil: null }, { lockedUntil: { $lt: now } }] }, + { $set: { lockedBy: this.workerId, lockedUntil } } + ); + + // 3. Read back exactly the rows we now hold from this batch, keyed by the lockedUntil time + const docs = await JobModel.find({ _id: { $in: ids }, lockedBy: this.workerId, lockedUntil }).lean(); + return docs.map(this.toEntity); + }; + recordSuccess = async (id: string, scheduledRunAt: number, intervalMs: number, now: number) => { // Fixed rate job scheduling diff --git a/server/src/domain/jobs/job.type.ts b/server/src/domain/jobs/job.type.ts index 7ffd0f2f8..70d635f81 100644 --- a/server/src/domain/jobs/job.type.ts +++ b/server/src/domain/jobs/job.type.ts @@ -1,6 +1,6 @@ export const JOB_TYPES = ["check", "geo-check", "evaluate", "cleanup-orphaned", "cleanup-retention"] as const; export type JobType = (typeof JOB_TYPES)[number]; -export const LOCK_MS = 30_000; +export const LOCK_MS = 60_000; export const BACKOFF_MS = 5_000; export const PARKED = Number.MAX_SAFE_INTEGER; diff --git a/server/src/worker/worker.db-queue.ts b/server/src/worker/worker.db-queue.ts index cb56ac477..c333491df 100644 --- a/server/src/worker/worker.db-queue.ts +++ b/server/src/worker/worker.db-queue.ts @@ -17,7 +17,7 @@ const SERVICE_NAME = "JobQueue"; const POLL_MS = 250; const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000; const CONCURRENCY: Record = { - check: 50, + check: 500, "geo-check": 20, evaluate: 20, "cleanup-orphaned": 1, @@ -205,13 +205,13 @@ export class DBQueueWorker implements IWorker { const tick = async () => { if (this.stopped) return; try { - // Drain rows up to concurrency limit - while (this.inFlight[type] < CONCURRENCY[type]) { - const job = await this.jobsRepository.claimDue(type, Date.now()); // Claim a job - if (!job) break; // we're done, no jobs due - this.inFlight[type]++; // increment in flight jobs to stay within concurrency limits + // Claim a batch sized to the free capacity + const capacity = CONCURRENCY[type] - this.inFlight[type]; + const jobs = await this.jobsRepository.claimDueBatch(type, capacity, Date.now()); + for (const job of jobs) { + this.inFlight[type]++; // stay within concurrency limits this.runJob(job).finally(() => { - this.inFlight[type]--; // job is done, decrement in flight count + this.inFlight[type]--; // job done, free the slot }); } } catch (error: unknown) {