Merge pull request #746 from bcomnes/minPriority

Implement minPriority and maxPriority jobFetch options

Author: timgit

docs/api/jobs.md

@@ -227,6 +227,14 @@ Returns an array of jobs from a queue
 
     If `true`, jobs with a `startAfter` timestamp in the future will be fetched. Useful for fetching jobs immediately without waiting for a retry delay.
 
+  * `minPriority`, int
+
+    If set, only fetch jobs with a priority greater than or equal to this value. If used together with `maxPriority`, `minPriority` must be less than or equal to `maxPriority`.
+
+  * `maxPriority`, int
+
+    If set, only fetch jobs with a priority less than or equal to this value. If used together with `minPriority`, `minPriority` must be less than or equal to `maxPriority`.
+
     ```js
     interface JobWithMetadata<T = object> {
       id: string;

docs/api/workers.md

@@ -33,6 +33,14 @@ The default options for `work()` is 1 job every 2 seconds.
 
   Same as in [`fetch()`](#fetch)
 
+* **minPriority**, int
+
+  Same as in [`fetch()`](#fetch)
+
+* **maxPriority**, int
+
+  Same as in [`fetch()`](#fetch)
+
 * **pollingIntervalSeconds**, int, *(default=2)*
 
   Interval to check for new jobs in seconds, must be >=0.5 (500ms)

packages/proxy/src/contracts.ts

@@ -79,7 +79,12 @@ export const fetchOptionsSchema = z.object({
   ignoreStartAfter: z.boolean().optional(),
   groupConcurrency: z.union([z.number(), groupConcurrencyConfigSchema]).optional(),
   ignoreGroups: z.array(z.string()).nullable().optional(),
-}) satisfies z.ZodType<types.HttpFetchOptions>
+  minPriority: z.number().int().optional(),
+  maxPriority: z.number().int().optional(),
+}).refine(
+  data => data.minPriority == null || data.maxPriority == null || data.minPriority <= data.maxPriority,
+  { message: 'minPriority must be <= maxPriority' }
+) satisfies z.ZodType<types.HttpFetchOptions>
 
 export const findJobsOptionsSchema = z.object({
   id: z.string().optional(),

src/attorney.ts

@@ -101,6 +101,18 @@ function validateGroupConcurrencyValue (value: any, optionName: string) {
   }
 }
 
+function validatePriorityRangeConfig (config: any) {
+  if (config.minPriority !== undefined) {
+    assert(Number.isInteger(config.minPriority), 'minPriority must be an integer')
+  }
+  if (config.maxPriority !== undefined) {
+    assert(Number.isInteger(config.maxPriority), 'maxPriority must be an integer')
+  }
+  if (config.minPriority !== undefined && config.maxPriority !== undefined) {
+    assert(config.minPriority <= config.maxPriority, 'minPriority must be <= maxPriority')
+  }
+}
+
 function validateGroupConcurrencyConfig (config: any) {
   const hasGlobal = config.groupConcurrency != null
   const hasLocal = config.localGroupConcurrency != null
@@ -168,6 +180,7 @@ function checkWorkArgs (name: string, args: any[]): {
   assert(!('includeMetadata' in options) || typeof options.includeMetadata === 'boolean', 'includeMetadata must be a boolean')
   assert(!('priority' in options) || typeof options.priority === 'boolean', 'priority must be a boolean')
   assert(!('localConcurrency' in options) || (Number.isInteger(options.localConcurrency) && options.localConcurrency >= 1), 'localConcurrency must be an integer >= 1')
+  validatePriorityRangeConfig(options)
   validateGroupConcurrencyConfig(options)
   validateHeartbeatRefreshConfig(options)
 
@@ -181,6 +194,7 @@ function checkFetchArgs (name: string, options: any) {
   assert(!('includeMetadata' in options) || typeof options.includeMetadata === 'boolean', 'includeMetadata must be a boolean')
   assert(!('priority' in options) || typeof options.priority === 'boolean', 'priority must be a boolean')
   assert(!('ignoreStartAfter' in options) || typeof options.ignoreStartAfter === 'boolean', 'ignoreStartAfter must be a boolean')
+  validatePriorityRangeConfig(options)
 }
 
 function getConfig (value: string | types.ConstructorOptions): types.ResolvedConstructorOptions {

src/manager.ts

@@ -327,7 +327,9 @@ class Manager extends EventEmitter implements types.EventsMixin {
       localGroupConcurrency,
       groupConcurrency,
       orderByCreatedOn = true,
-      heartbeatRefreshSeconds
+      heartbeatRefreshSeconds,
+      minPriority,
+      maxPriority,
     } = options
 
     if (localGroupConcurrency != null) {
@@ -341,7 +343,7 @@ class Manager extends EventEmitter implements types.EventsMixin {
         const ignoreGroups = localGroupConcurrency != null
           ? this.#getGroupsAtLocalCapacity(name)
           : undefined
-        return this.fetch<ReqData>(name, { batchSize, includeMetadata, priority, orderByCreatedOn, groupConcurrency, ignoreGroups })
+        return this.fetch<ReqData>(name, { batchSize, includeMetadata, priority, orderByCreatedOn, groupConcurrency, ignoreGroups, minPriority, maxPriority })
       }
 
       const onFetch = async (jobs: types.Job<ReqData>[]) => {

src/plans.ts

@@ -795,6 +795,8 @@ interface FetchJobOptions {
   ignoreSingletons: string[] | null
   ignoreGroups?: string[] | null
   groupConcurrency?: number | GroupConcurrencyConfig
+  minPriority?: number
+  maxPriority?: number
 }
 
 interface FetchQueryParams {
@@ -803,13 +805,17 @@ interface FetchQueryParams {
   ignoreGroupsParam: string
   defaultGroupLimitParam: string
   tiersParam: string
+  minPriorityParam: string
+  maxPriorityParam: string
 }
 
 function buildFetchParams (options: FetchJobOptions): FetchQueryParams {
-  const { ignoreSingletons, ignoreGroups, groupConcurrency } = options
+  const { ignoreSingletons, ignoreGroups, groupConcurrency, minPriority, maxPriority } = options
   const hasIgnoreSingletons = ignoreSingletons != null && ignoreSingletons.length > 0
   const hasIgnoreGroups = ignoreGroups != null && ignoreGroups.length > 0
   const hasGroupConcurrency = groupConcurrency != null
+  const hasMinPriority = minPriority != null
+  const hasMaxPriority = maxPriority != null
   const groupConcurrencyConfig = hasGroupConcurrency
     ? (typeof groupConcurrency === 'number' ? { default: groupConcurrency } : groupConcurrency)
     : null
@@ -821,6 +827,8 @@ function buildFetchParams (options: FetchJobOptions): FetchQueryParams {
   let ignoreGroupsParam = ''
   let defaultGroupLimitParam = ''
   let tiersParam = ''
+  let minPriorityParam = ''
+  let maxPriorityParam = ''
 
   if (hasIgnoreSingletons) {
     paramIndex++
@@ -846,16 +854,30 @@ function buildFetchParams (options: FetchJobOptions): FetchQueryParams {
     }
   }
 
-  return { values, ignoreSingletonsParam, ignoreGroupsParam, defaultGroupLimitParam, tiersParam }
+  if (hasMinPriority) {
+    paramIndex++
+    minPriorityParam = `$${paramIndex}::int`
+    values.push(minPriority)
+  }
+
+  if (hasMaxPriority) {
+    paramIndex++
+    maxPriorityParam = `$${paramIndex}::int`
+    values.push(maxPriority)
+  }
+
+  return { values, ignoreSingletonsParam, ignoreGroupsParam, defaultGroupLimitParam, tiersParam, minPriorityParam, maxPriorityParam }
 }
 
 function fetchNextJob (options: FetchJobOptions): SqlQuery {
-  const { schema, table, name, policy, limit, includeMetadata, priority = true, orderByCreatedOn = true, ignoreStartAfter = false, groupConcurrency } = options
+  const { schema, table, name, policy, limit, includeMetadata, priority = true, orderByCreatedOn = true, ignoreStartAfter = false, groupConcurrency, minPriority, maxPriority } = options
 
   const singletonFetch = limit > 1 && (policy === QUEUE_POLICIES.singleton || policy === QUEUE_POLICIES.stately)
   const hasIgnoreSingletons = options.ignoreSingletons != null && options.ignoreSingletons.length > 0
   const hasIgnoreGroups = options.ignoreGroups != null && options.ignoreGroups.length > 0
   const hasGroupConcurrency = groupConcurrency != null
+  const hasMinPriority = minPriority != null
+  const hasMaxPriority = maxPriority != null
   const hasTiers = hasGroupConcurrency &&
     typeof groupConcurrency === 'object' &&
     groupConcurrency.tiers &&
@@ -868,7 +890,9 @@ function fetchNextJob (options: FetchJobOptions): SqlQuery {
     `state < '${JOB_STATES.active}'`,
     !ignoreStartAfter ? 'start_after < now()' : '',
     hasIgnoreSingletons ? `singleton_key <> ALL(${params.ignoreSingletonsParam})` : '',
-    hasIgnoreGroups ? `(group_id IS NULL OR group_id <> ALL(${params.ignoreGroupsParam}))` : ''
+    hasIgnoreGroups ? `(group_id IS NULL OR group_id <> ALL(${params.ignoreGroupsParam}))` : '',
+    hasMinPriority ? `priority >= ${params.minPriorityParam}` : '',
+    hasMaxPriority ? `priority <= ${params.maxPriorityParam}` : ''
   ].filter(Boolean).join(' AND ')
 
   const selectCols = [

src/types.ts

@@ -315,6 +315,20 @@ export interface JobFetchOptions {
    * @default false
    */
   ignoreStartAfter?: boolean;
+  /**
+   * Only fetch jobs with a priority greater than or equal to this value.
+   * Useful for reserving worker capacity exclusively for higher-priority jobs.
+   * Must be an integer. If both `minPriority` and `maxPriority` are set,
+   * `minPriority` must be less than or equal to `maxPriority`.
+   */
+  minPriority?: number;
+  /**
+   * Only fetch jobs with a priority less than or equal to this value.
+   * Useful for workers dedicated to lower-priority background work.
+   * Must be an integer. If both `minPriority` and `maxPriority` are set,
+   * `minPriority` must be less than or equal to `maxPriority`.
+   */
+  maxPriority?: number;
 }
 
 export interface WorkConcurrencyOptions {

test/priorityTest.ts

@@ -1,6 +1,7 @@
 import { expect } from 'vitest'
 import * as helper from './testHelper.ts'
 import { ctx } from './hooks.ts'
+import { assertTruthy } from './testHelper.ts'
 
 describe('priority', function () {
   it('higher priority job', async function () {
@@ -46,4 +47,104 @@ describe('priority', function () {
     expect(job2.id).toBe(medium)
     expect(job3.id).toBe(high)
   })
+
+  it('minPriority skips jobs below threshold', async function () {
+    ctx.boss = await helper.start(ctx.bossConfig)
+
+    await ctx.boss.send(ctx.schema, null, { priority: -10 })
+    const normal = await ctx.boss.send(ctx.schema, null, { priority: 0 })
+
+    const [job] = await ctx.boss.fetch(ctx.schema, { minPriority: 0 })
+
+    expect(job.id).toBe(normal)
+  })
+
+  it('minPriority returns nothing when all jobs are below threshold', async function () {
+    ctx.boss = await helper.start(ctx.bossConfig)
+
+    await ctx.boss.send(ctx.schema, null, { priority: -10 })
+    await ctx.boss.send(ctx.schema, null, { priority: -5 })
+
+    const jobs = await ctx.boss.fetch(ctx.schema, { minPriority: 0 })
+
+    expect(jobs.length).toBe(0)
+  })
+
+  it('maxPriority skips jobs above threshold', async function () {
+    ctx.boss = await helper.start(ctx.bossConfig)
+
+    const low = await ctx.boss.send(ctx.schema, null, { priority: -10 })
+    await ctx.boss.send(ctx.schema, null, { priority: 5 })
+
+    const [job] = await ctx.boss.fetch(ctx.schema, { maxPriority: 0 })
+
+    expect(job.id).toBe(low)
+  })
+
+  it('maxPriority returns nothing when all jobs are above threshold', async function () {
+    ctx.boss = await helper.start(ctx.bossConfig)
+
+    await ctx.boss.send(ctx.schema, null, { priority: 5 })
+    await ctx.boss.send(ctx.schema, null, { priority: 10 })
+
+    const jobs = await ctx.boss.fetch(ctx.schema, { maxPriority: 0 })
+
+    expect(jobs.length).toBe(0)
+  })
+
+  it('minPriority and maxPriority together fetch only jobs in range', async function () {
+    ctx.boss = await helper.start(ctx.bossConfig)
+
+    await ctx.boss.send(ctx.schema, null, { priority: -10 })
+    const inRange = await ctx.boss.send(ctx.schema, null, { priority: 5 })
+    await ctx.boss.send(ctx.schema, null, { priority: 20 })
+
+    const [job] = await ctx.boss.fetch(ctx.schema, { minPriority: 1, maxPriority: 10 })
+
+    expect(job.id).toBe(inRange)
+  })
+
+  it('worker with minPriority skips jobs below threshold', async function () {
+    ctx.boss = await helper.start({ ...ctx.bossConfig, __test__enableSpies: true })
+
+    const spy = ctx.boss.getSpy(ctx.schema)
+
+    const skipped = await ctx.boss.send(ctx.schema, null, { priority: -10 })
+    const normal = await ctx.boss.send(ctx.schema, null, { priority: 0 })
+
+    assertTruthy(skipped)
+    assertTruthy(normal)
+
+    await ctx.boss.work(ctx.schema, { minPriority: 0 }, async () => {})
+
+    await spy.waitForJobWithId(normal, 'completed')
+    await ctx.boss.offWork(ctx.schema)
+
+    const [remainingJob] = await ctx.boss.findJobs(ctx.schema, { id: skipped })
+
+    assertTruthy(remainingJob)
+    expect(remainingJob.state).toBe('created')
+  })
+
+  it('worker with maxPriority skips jobs above threshold', async function () {
+    ctx.boss = await helper.start({ ...ctx.bossConfig, __test__enableSpies: true })
+
+    const spy = ctx.boss.getSpy(ctx.schema)
+
+    const low = await ctx.boss.send(ctx.schema, null, { priority: -10 })
+    const skipped = await ctx.boss.send(ctx.schema, null, { priority: 5 })
+
+    assertTruthy(low)
+    assertTruthy(skipped)
+
+    await ctx.boss.work(ctx.schema, { maxPriority: 0 }, async () => {})
+
+    await spy.waitForJobWithId(low, 'completed')
+    await ctx.boss.offWork(ctx.schema)
+
+    const [remainingJob] = await ctx.boss.findJobs(ctx.schema, { id: skipped })
+
+    assertTruthy(remainingJob)
+    expect(remainingJob.state).toBe('created')
+  })
 })