add a batch claim method
This commit is contained in:
1 parent
8b9c41464a
commit
47162f2064
4 files changed
+40
-10
No files matched your search
@@ -17,6 +17,9 @@ export interface IJobsRepository {
|
||||
// Atomically claim a job
|
||||
claimDue(type: JobType, now: number): Promise<Job | null>;
|
||||
|
||||
// locks up to `limit` due rows in one batch
|
||||
claimDueBatch(type: JobType, limit: number, now: number): Promise<Job[]>;
|
||||
|
||||
// Completes a repeating job
|
||||
recordSuccess(id: string, nextScheduledAt: number, intervalMs: number, now: number): Promise<boolean>;
|
||||
|
||||
@@ -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<boolean>;
|
||||
|
||||
// Pause/Resume
|
||||
@@ -46,7 +49,7 @@ export interface IJobsRepository {
|
||||
// Delete, drop all rows
|
||||
deleteById(refId: string): Promise<boolean>;
|
||||
|
||||
// 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<boolean>;
|
||||
|
||||
// Get job
|
||||
|
||||
@@ -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<JobDocument[]>();
|
||||
return docs.map(this.toEntity);
|
||||
};
|
||||
|
||||
recordSuccess = async (id: string, scheduledRunAt: number, intervalMs: number, now: number) => {
|
||||
// Fixed rate job scheduling
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ const SERVICE_NAME = "JobQueue";
|
||||
const POLL_MS = 250;
|
||||
const CLEANUP_INTERVAL_MS = 24 * 60 * 60 * 1000;
|
||||
const CONCURRENCY: Record<JobType, number> = {
|
||||
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) {
|
||||
|
||||
Reference in new issue
Block a user