9949043480

Committed 16 Jul 2024 01:09AM UTC coverage: 94.41% (-5.6%) from 100.0%

Build # 9949043480

Build Type

Pull #425

github

Committed by

web-flow

Commit Message

Merge 6b98fbe61 into f1c1636ca

Pull Request Pull Request #425: v10

Run Details

466 of 546 branches covered (85.35%)

351 of 365 new or added lines in 10 files covered. (96.16%)

40 existing lines in 5 files now uncovered.

912 of 966 relevant lines covered (94.41%)

803.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.62

/src/manager.js

const assert = require('assert')
const EventEmitter = require('events')
const { randomUUID } = require('crypto')
const debounce = require('lodash.debounce')
const { serializeError: stringify } = require('serialize-error')
const pMap = require('p-map')
const { delay } = require('./tools')
const Attorney = require('./attorney')
const Worker = require('./worker')
const plans = require('./plans')

const { QUEUES: TIMEKEEPER_QUEUES } = require('./timekeeper')
const { QUEUE_POLICY } = plans

const INTERNAL_QUEUES = Object.values(TIMEKEEPER_QUEUES).reduce((acc, i) => ({ ...acc, [i]: i }), {})

const WIP_EVENT_INTERVAL = 2000
const WIP_DEBOUNCE_OPTIONS = { leading: true, trailing: true, maxWait: WIP_EVENT_INTERVAL }

const events = {
  error: 'error',
  wip: 'wip'
}

const resolveWithinSeconds = async (promise, seconds) => {
  const timeout = Math.max(1, seconds) * 1000
  const reject = delay(timeout, `handler execution exceeded ${timeout}ms`)

  let result

  try {
    result = await Promise.race([promise, reject])
  } finally {
    reject.abort()
  }

  return result
}

class Manager extends EventEmitter {
  constructor (db, config) {
    super()

    this.config = config
    this.db = db

    this.events = events
    this.workers = new Map()

    this.nextJobCommand = plans.fetchNextJob(config.schema)
    this.insertJobCommand = plans.insertJob(config.schema)
    this.insertJobsCommand = plans.insertJobs(config.schema)
    this.completeJobsCommand = plans.completeJobs(config.schema)
    this.cancelJobsCommand = plans.cancelJobs(config.schema)
    this.resumeJobsCommand = plans.resumeJobs(config.schema)
    this.failJobsByIdCommand = plans.failJobsById(config.schema)
    this.getJobByIdCommand = plans.getJobById(config.schema)
    this.getArchivedJobByIdCommand = plans.getArchivedJobById(config.schema)
    this.subscribeCommand = plans.subscribe(config.schema)
    this.unsubscribeCommand = plans.unsubscribe(config.schema)
    this.getQueuesForEventCommand = plans.getQueuesForEvent(config.schema)

    // exported api to index
    this.functions = [
      this.complete,
      this.cancel,
      this.resume,
      this.fail,
      this.fetch,
      this.work,
      this.offWork,
      this.notifyWorker,
      this.publish,
      this.subscribe,
      this.unsubscribe,
      this.insert,
      this.send,
      this.sendDebounced,
      this.sendThrottled,
      this.sendAfter,
      this.createQueue,
      this.updateQueue,
      this.getQueue,
      this.deleteQueue,
      this.purgeQueue,
      this.getQueueSize,
      this.clearStorage,
      this.getJobById
    ]

    this.emitWipThrottled = debounce(() => this.emit(events.wip, this.getWipData()), WIP_EVENT_INTERVAL, WIP_DEBOUNCE_OPTIONS)
  }

  start () {
    this.stopped = false
  }

  async stop () {
    this.stopped = true

    for (const worker of this.workers.values()) {
      if (!INTERNAL_QUEUES[worker.name]) {
        await this.offWork(worker.name)
      }
    }
  }

  async failWip () {
    for (const worker of this.workers.values()) {
      const jobIds = worker.jobs.map(j => j.id)
      if (jobIds.length) {
        await this.fail(worker.name, jobIds, 'pg-boss shut down while active')
      }
    }
  }

  async work (name, ...args) {
    const { options, callback } = Attorney.checkWorkArgs(name, args, this.config)
    return await this.watch(name, options, callback)
  }

  addWorker (worker) {
    this.workers.set(worker.id, worker)
  }

  removeWorker (worker) {
    this.workers.delete(worker.id)
  }

  getWorkers () {
    return Array.from(this.workers.values())
  }

  emitWip (name) {
    if (!INTERNAL_QUEUES[name]) {
      this.emitWipThrottled()
    }
  }

  getWipData (options = {}) {
    const { includeInternal = false } = options

    const data = this.getWorkers()
      .map(({
        id,
        name,
        options,
        state,
        jobs,
        createdOn,
        lastFetchedOn,
        lastJobStartedOn,
        lastJobEndedOn,
        lastError,
        lastErrorOn
      }) => ({
        id,
        name,
        options,
        state,
        count: jobs.length,
        createdOn,
        lastFetchedOn,
        lastJobStartedOn,
        lastJobEndedOn,
        lastError,
        lastErrorOn
      }))
      .filter(i => i.count > 0 && (!INTERNAL_QUEUES[i.name] || includeInternal))

    return data
  }

  async watch (name, options, callback) {
    if (this.stopped) {
      throw new Error('Workers are disabled. pg-boss is stopped')
    }

    const {
      newJobCheckInterval: interval = this.config.newJobCheckInterval,
      batchSize,
      teamSize = 1,
      teamConcurrency = 1,
      teamRefill: refill = false,
      includeMetadata = false,
      priority = true
    } = options

    const id = randomUUID({ disableEntropyCache: true })

    let queueSize = 0

    let refillTeamPromise
    let resolveRefillTeam

    // Setup a promise that onFetch can await for when at least one
    // job is finished and so the team is ready to be topped up
    const createTeamRefillPromise = () => {
      refillTeamPromise = new Promise((resolve) => { resolveRefillTeam = resolve })
    }

    createTeamRefillPromise()

    const onRefill = () => {
      queueSize--
      resolveRefillTeam()
      createTeamRefillPromise()
    }

    const fetch = () => this.fetch(name, batchSize || (teamSize - queueSize), { includeMetadata, priority })

    const onFetch = async (jobs) => {
      if (this.config.__test__throw_worker) {
        throw new Error('__test__throw_worker')
      }

      this.emitWip(name)

      if (batchSize) {
        const maxExpiration = jobs.reduce((acc, i) => Math.max(acc, i.expire_in_seconds), 0)

        await resolveWithinSeconds(Promise.all([callback(jobs)]), maxExpiration)
          .then(() => this.complete(name, jobs.map(job => job.id)))
          .catch(err => this.fail(name, jobs.map(job => job.id), err))
      } else {
        if (refill) {
          queueSize += jobs.length || 1
        }

        const allTeamPromise = pMap(jobs, job =>
          resolveWithinSeconds(callback(job), job.expire_in_seconds)
            .then(result => this.complete(name, job.id, result))
            .catch(err => this.fail(name, job.id, err))
            .then(() => refill ? onRefill() : null)
        , { concurrency: teamConcurrency }
        ).catch(() => {}) // allow promises & non-promises to live together in harmony

        if (refill) {
          if (queueSize < teamSize) {
            return
          } else {
            await refillTeamPromise
          }
        } else {
          await allTeamPromise
        }
      }

      this.emitWip(name)
    }

    const onError = error => {
      this.emit(events.error, { ...error, message: error.message, stack: error.stack, queue: name, worker: id })
    }

    const worker = new Worker({ id, name, options, interval, fetch, onFetch, onError })

    this.addWorker(worker)

    worker.start()

    return id
  }

  async offWork (value) {
    assert(value, 'Missing required argument')

    const query = (typeof value === 'string')
      ? { filter: i => i.name === value }
      : (typeof value === 'object' && value.id)
          ? { filter: i => i.id === value.id }
          : null

    assert(query, 'Invalid argument. Expected string or object: { id }')

    const workers = this.getWorkers().filter(i => query.filter(i) && !i.stopping && !i.stopped)

    if (workers.length === 0) {
      return
    }

    for (const worker of workers) {
      worker.stop()
    }

    setImmediate(async () => {
      while (!workers.every(w => w.stopped)) {
        await delay(1000)
      }

      for (const worker of workers) {
        this.removeWorker(worker)
      }
    })
  }

  notifyWorker (workerId) {
    if (this.workers.has(workerId)) {
      this.workers.get(workerId).notify()
    }
  }

  async subscribe (event, name) {
    assert(event, 'Missing required argument')
    assert(name, 'Missing required argument')

    return await this.db.executeSql(this.subscribeCommand, [event, name])
  }

  async unsubscribe (event, name) {
    assert(event, 'Missing required argument')
    assert(name, 'Missing required argument')

    return await this.db.executeSql(this.unsubscribeCommand, [event, name])
  }

  async publish (event, ...args) {
    assert(event, 'Missing required argument')

    const result = await this.db.executeSql(this.getQueuesForEventCommand, [event])

    if (!result || result.rowCount === 0) {
      return []
    }

    return await Promise.all(result.rows.map(({ name }) => this.send(name, ...args)))
  }

  async send (...args) {
    const { name, data, options } = Attorney.checkSendArgs(args, this.config)
    return await this.createJob(name, data, options)
  }

  async sendAfter (name, data, options, after) {
    options = options ? { ...options } : {}
    options.startAfter = after

    const result = Attorney.checkSendArgs([name, data, options], this.config)

    return await this.createJob(result.name, result.data, result.options)
  }

  async sendThrottled (name, data, options, seconds, key) {
    options = options ? { ...options } : {}
    options.singletonSeconds = seconds
    options.singletonNextSlot = false
    options.singletonKey = key

    const result = Attorney.checkSendArgs([name, data, options], this.config)

    return await this.createJob(result.name, result.data, result.options)
  }

  async sendDebounced (name, data, options, seconds, key) {
    options = options ? { ...options } : {}
    options.singletonSeconds = seconds
    options.singletonNextSlot = true
    options.singletonKey = key

    const result = Attorney.checkSendArgs([name, data, options], this.config)

    return await this.createJob(result.name, result.data, result.options)
  }

  async createJob (name, data, options, singletonOffset = 0) {
    const {
      id = null,
      db: wrapper,
      priority,
      startAfter,
      singletonKey = null,
      singletonSeconds,
      deadLetter = null,
      expireIn,
      expireInDefault,
      keepUntil,
      keepUntilDefault,
      retryLimit,
      retryLimitDefault,
      retryDelay,
      retryDelayDefault,
      retryBackoff,
      retryBackoffDefault
    } = options

    const values = [
      id, // 1
      name, // 2
      data, // 3
      priority, // 4
      startAfter, // 5
      singletonKey, // 6
      singletonSeconds, // 7
      singletonOffset, // 8
      deadLetter, // 9
      expireIn, // 10
      expireInDefault, // 11
      keepUntil, // 12
      keepUntilDefault, // 13
      retryLimit, // 14
      retryLimitDefault, // 15
      retryDelay, // 16
      retryDelayDefault, // 17
      retryBackoff, // 18
      retryBackoffDefault // 19
    ]

    const db = wrapper || this.db
    const result = await db.executeSql(this.insertJobCommand, values)

    if (result && result.rowCount === 1) {
      return result.rows[0].id
    }

    if (!options.singletonNextSlot) {
      return null
    }

    // delay starting by the offset to honor throttling config
    options.startAfter = this.getDebounceStartAfter(singletonSeconds, this.timekeeper.clockSkew)

    // toggle off next slot config for round 2
    options.singletonNextSlot = false

    singletonOffset = singletonSeconds

    return await this.createJob(name, data, options, singletonOffset)
  }

  async insert (jobs, options = {}) {
    assert(Array.isArray(jobs), 'jobs argument should be an array')

    const db = options.db || this.db

    const params = [
      JSON.stringify(jobs), // 1
      this.config.expireIn, // 2
      this.config.keepUntil, // 3
      this.config.retryLimit, // 4
      this.config.retryDelay, // 5
      this.config.retryBackoff // 6
    ]

    return await db.executeSql(this.insertJobsCommand, params)
  }

  getDebounceStartAfter (singletonSeconds, clockOffset) {
    const debounceInterval = singletonSeconds * 1000

    const now = Date.now() + clockOffset

    const slot = Math.floor(now / debounceInterval) * debounceInterval

    // prevent startAfter=0 during debouncing
    let startAfter = (singletonSeconds - Math.floor((now - slot) / 1000)) || 1

    if (singletonSeconds > 1) {
      startAfter++
    }

    return startAfter
  }

  async fetch (name, batchSize, options = {}) {
    const values = Attorney.checkFetchArgs(name, batchSize, options)
    const db = options.db || this.db
    const nextJobSql = this.nextJobCommand({ ...options })
    const statementValues = [values.name, batchSize || 1]

    let result

    try {
      result = await db.executeSql(nextJobSql, statementValues)
    } catch (err) {
      // errors from fetchquery should only be unique constraint violations
    }

    if (!result || result.rows.length === 0) {
      return null
    }

    return result.rows.length === 1 && !batchSize ? result.rows[0] : result.rows
  }

  mapCompletionIdArg (id, funcName) {
    const errorMessage = `${funcName}() requires an id`

    assert(id, errorMessage)

    const ids = Array.isArray(id) ? id : [id]

    assert(ids.length, errorMessage)

    return ids
  }

  mapCompletionDataArg (data) {
    if (data === null || typeof data === 'undefined' || typeof data === 'function') { return null }

    const result = (typeof data === 'object' && !Array.isArray(data))
      ? data
      : { value: data }

    return stringify(result)
  }

  mapCompletionResponse (ids, result) {
    return {
      jobs: ids,
      requested: ids.length,
      updated: result && result.rows ? parseInt(result.rows[0].count) : 0
    }
  }

  async complete (name, id, data, options = {}) {
    assert(name, 'Missing queue name argument')
    const db = options.db || this.db
    const ids = this.mapCompletionIdArg(id, 'complete')
    const result = await db.executeSql(this.completeJobsCommand, [name, ids, this.mapCompletionDataArg(data)])
    return this.mapCompletionResponse(ids, result)
  }

  async fail (name, id, data, options = {}) {
    assert(name, 'Missing queue name argument')
    const db = options.db || this.db
    const ids = this.mapCompletionIdArg(id, 'fail')
    const result = await db.executeSql(this.failJobsByIdCommand, [name, ids, this.mapCompletionDataArg(data)])
    return this.mapCompletionResponse(ids, result)
  }

  async cancel (name, id, options = {}) {
    assert(name, 'Missing queue name argument')
    const db = options.db || this.db
    const ids = this.mapCompletionIdArg(id, 'cancel')
    const result = await db.executeSql(this.cancelJobsCommand, [name, ids])
    return this.mapCompletionResponse(ids, result)
  }

  async resume (name, id, options = {}) {
    assert(name, 'Missing queue name argument')
    const db = options.db || this.db
    const ids = this.mapCompletionIdArg(id, 'resume')
    const result = await db.executeSql(this.resumeJobsCommand, [name, ids])
    return this.mapCompletionResponse(ids, result)
  }

  async createQueue (name, options = {}) {
    assert(name, 'Missing queue name argument')

    Attorney.assertQueueName(name)

    const { policy = QUEUE_POLICY.standard } = options

    assert(policy in QUEUE_POLICY, `${policy} is not a valid queue policy`)

    const {
      retryLimit,
      retryDelay,
      retryBackoff,
      expireInSeconds,
      retentionMinutes,
      deadLetter
    } = Attorney.checkQueueArgs(name, options)

    const paritionSql = plans.createPartition(this.config.schema, name)

    await this.db.executeSql(paritionSql)

    const sql = plans.createQueue(this.config.schema, name)

    const params = [
      name,
      policy,
      retryLimit,
      retryDelay,
      retryBackoff,
      expireInSeconds,
      retentionMinutes,
      deadLetter
    ]

    await this.db.executeSql(sql, params)
  }

  async updateQueue (name, options = {}) {
    assert(name, 'Missing queue name argument')

    const {
      retryLimit,
      retryDelay,
      retryBackoff,
      expireInSeconds,
      retentionMinutes,
      deadLetter
    } = Attorney.checkQueueArgs(name, options)

    const sql = plans.updateQueue(this.config.schema)

    const params = [
      name,
      retryLimit,
      retryDelay,
      retryBackoff,
      expireInSeconds,
      retentionMinutes,
      deadLetter
    ]

    await this.db.executeSql(sql, params)
  }

  async getQueue (name) {
    assert(name, 'Missing queue name argument')

    const sql = plans.getQueueByName(this.config.schema)
    const result = await this.db.executeSql(sql, [name])

    if (result.rows.length === 0) {
      return null
    }

    const {
      policy,
      retry_limit: retryLimit,
      retry_delay: retryDelay,
      retry_backoff: retryBackoff,
      expire_seconds: expireInSeconds,
      retention_minutes: retentionMinutes,
      dead_letter: deadLetter
    } = result.rows[0]

    return {
      name,
      policy,
      retryLimit,
      retryDelay,
      retryBackoff,
      expireInSeconds,
      retentionMinutes,
      deadLetter
    }
  }

  async deleteQueue (name) {
    assert(name, 'Missing queue name argument')

    const queueSql = plans.getQueueByName(this.config.schema)
    const result = await this.db.executeSql(queueSql, [name])

    if (result?.rows?.length) {
      Attorney.assertQueueName(name)
      const sql = plans.dropPartition(this.config.schema, name)
      await this.db.executeSql(sql)
    }

    const sql = plans.deleteQueueRecords(this.config.schema)
    const result2 = await this.db.executeSql(sql, [name])
    return result2?.rowCount || null
  }

  async purgeQueue (queue) {
    assert(queue, 'Missing queue name argument')
    const sql = plans.purgeQueue(this.config.schema)
    await this.db.executeSql(sql, [queue])
  }

  async clearStorage () {
    const sql = plans.clearStorage(this.config.schema)
    await this.db.executeSql(sql)
  }

  async getQueueSize (queue, options) {
    assert(queue, 'Missing queue name argument')

    const sql = plans.getQueueSize(this.config.schema, options)

    const result = await this.db.executeSql(sql, [queue])

    return result ? parseFloat(result.rows[0].count) : null
  }

  async getJobById (queue, id, options = {}) {
    const db = options.db || this.db
    const result1 = await db.executeSql(this.getJobByIdCommand, [queue, id])

    if (result1 && result1.rows && result1.rows.length === 1) {
      return result1.rows[0]
    }

    const result2 = await db.executeSql(this.getArchivedJobByIdCommand, [queue, id])

    if (result2 && result2.rows && result2.rows.length === 1) {
      return result2.rows[0]
    }

    return null
  }
}

module.exports = Manager

1	const assert = require('assert')	4✔
2	const EventEmitter = require('events')	4✔
3	const { randomUUID } = require('crypto')	4✔
4	const debounce = require('lodash.debounce')	4✔
5	const { serializeError: stringify } = require('serialize-error')	4✔
6	const pMap = require('p-map')	4✔
7	const { delay } = require('./tools')	4✔
8	const Attorney = require('./attorney')	4✔
9	const Worker = require('./worker')	4✔
10	const plans = require('./plans')	4✔
11
12	const { QUEUES: TIMEKEEPER_QUEUES } = require('./timekeeper')	4✔
13	const { QUEUE_POLICY } = plans	4✔
14
15	const INTERNAL_QUEUES = Object.values(TIMEKEEPER_QUEUES).reduce((acc, i) => ({ ...acc, [i]: i }), {})	4✔
16
17	const WIP_EVENT_INTERVAL = 2000	4✔
18	const WIP_DEBOUNCE_OPTIONS = { leading: true, trailing: true, maxWait: WIP_EVENT_INTERVAL }	4✔
19
20	const events = {	4✔
21	error: 'error',
22	wip: 'wip'
23	}
24
25	const resolveWithinSeconds = async (promise, seconds) => {	4✔
26	const timeout = Math.max(1, seconds) * 1000	56✔
27	const reject = delay(timeout, `handler execution exceeded ${timeout}ms`)	56✔
28
29	let result
30
31	try {	56✔
32	result = await Promise.race([promise, reject])	56✔
33	} finally {
34	reject.abort()	56✔
35	}
36
37	return result	46✔
38	}
39
40	class Manager extends EventEmitter {
41	constructor (db, config) {
42	super()	182✔
43
44	this.config = config	182✔
45	this.db = db	182✔
46
47	this.events = events	182✔
48	this.workers = new Map()	182✔
49
50	this.nextJobCommand = plans.fetchNextJob(config.schema)	182✔
51	this.insertJobCommand = plans.insertJob(config.schema)	182✔
52	this.insertJobsCommand = plans.insertJobs(config.schema)	182✔
53	this.completeJobsCommand = plans.completeJobs(config.schema)	182✔
54	this.cancelJobsCommand = plans.cancelJobs(config.schema)	182✔
55	this.resumeJobsCommand = plans.resumeJobs(config.schema)	182✔
56	this.failJobsByIdCommand = plans.failJobsById(config.schema)	182✔
57	this.getJobByIdCommand = plans.getJobById(config.schema)	182✔
58	this.getArchivedJobByIdCommand = plans.getArchivedJobById(config.schema)	182✔
59	this.subscribeCommand = plans.subscribe(config.schema)	182✔
60	this.unsubscribeCommand = plans.unsubscribe(config.schema)	182✔
61	this.getQueuesForEventCommand = plans.getQueuesForEvent(config.schema)	182✔
62
63	// exported api to index
64	this.functions = [	182✔
65	this.complete,
66	this.cancel,
67	this.resume,
68	this.fail,
69	this.fetch,
70	this.work,
71	this.offWork,
72	this.notifyWorker,
73	this.publish,
74	this.subscribe,
75	this.unsubscribe,
76	this.insert,
77	this.send,
78	this.sendDebounced,
79	this.sendThrottled,
80	this.sendAfter,
81	this.createQueue,
82	this.updateQueue,
83	this.getQueue,
84	this.deleteQueue,
85	this.purgeQueue,
86	this.getQueueSize,
87	this.clearStorage,
88	this.getJobById
89	]
90
91	this.emitWipThrottled = debounce(() => this.emit(events.wip, this.getWipData()), WIP_EVENT_INTERVAL, WIP_DEBOUNCE_OPTIONS)	182✔
92	}
93
94	start () {
95	this.stopped = false	181✔
96	}
97
98	async stop () {
99	this.stopped = true	180✔
100
101	for (const worker of this.workers.values()) {	180✔
102	if (!INTERNAL_QUEUES[worker.name]) {	37✔
103	await this.offWork(worker.name)	28✔
104	}
105	}
106	}
107
108	async failWip () {
109	for (const worker of this.workers.values()) {	178✔
110	const jobIds = worker.jobs.map(j => j.id)	17✔
111	if (jobIds.length) {	17✔
112	await this.fail(worker.name, jobIds, 'pg-boss shut down while active')	3✔
113	}
114	}
115	}
116
117	async work (name, ...args) {
118	const { options, callback } = Attorney.checkWorkArgs(name, args, this.config)	44✔
119	return await this.watch(name, options, callback)	41✔
120	}
121
122	addWorker (worker) {
123	this.workers.set(worker.id, worker)	40✔
124	}
125
126	removeWorker (worker) {
127	this.workers.delete(worker.id)	38✔
128	}
129
130	getWorkers () {
131	return Array.from(this.workers.values())	274✔
132	}
133
134	emitWip (name) {
135	if (!INTERNAL_QUEUES[name]) {	88✔
136	this.emitWipThrottled()	80✔
137	}
138	}
139
140	getWipData (options = {}) {	64✔
141	const { includeInternal = false } = options	234✔
142
143	const data = this.getWorkers()	234✔
144	.map(({
145	id,
146	name,
147	options,
148	state,
149	jobs,
150	createdOn,
151	lastFetchedOn,
152	lastJobStartedOn,
153	lastJobEndedOn,
154	lastError,
155	lastErrorOn
156	}) => ({	83✔
157	id,
158	name,
159	options,
160	state,
161	count: jobs.length,
162	createdOn,
163	lastFetchedOn,
164	lastJobStartedOn,
165	lastJobEndedOn,
166	lastError,
167	lastErrorOn
168	}))
169	.filter(i => i.count > 0 && (!INTERNAL_QUEUES[i.name] \|\| includeInternal))	83!
170
171	return data	234✔
172	}
173
174	async watch (name, options, callback) {
175	if (this.stopped) {	41✔
176	throw new Error('Workers are disabled. pg-boss is stopped')	1✔
177	}
178
179	const {
180	newJobCheckInterval: interval = this.config.newJobCheckInterval,	×
181	batchSize,
182	teamSize = 1,	26✔
183	teamConcurrency = 1,	27✔
184	teamRefill: refill = false,	38✔
185	includeMetadata = false,	39✔
186	priority = true	40✔
187	} = options	40✔
188
189	const id = randomUUID({ disableEntropyCache: true })	40✔
190
191	let queueSize = 0	40✔
192
193	let refillTeamPromise
194	let resolveRefillTeam
195
196	// Setup a promise that onFetch can await for when at least one
197	// job is finished and so the team is ready to be topped up
198	const createTeamRefillPromise = () => {	40✔
199	refillTeamPromise = new Promise((resolve) => { resolveRefillTeam = resolve })	49✔
200	}
201
202	createTeamRefillPromise()	40✔
203
204	const onRefill = () => {	40✔
205	queueSize--	9✔
206	resolveRefillTeam()	9✔
207	createTeamRefillPromise()	9✔
208	}
209
210	const fetch = () => this.fetch(name, batchSize \|\| (teamSize - queueSize), { includeMetadata, priority })	25,999✔
211
212	const onFetch = async (jobs) => {	40✔
213	if (this.config.__test__throw_worker) {	45✔
214	throw new Error('__test__throw_worker')	1✔
215	}
216
217	this.emitWip(name)	44✔
218
219	if (batchSize) {	44✔
220	const maxExpiration = jobs.reduce((acc, i) => Math.max(acc, i.expire_in_seconds), 0)	7✔
221
222	await resolveWithinSeconds(Promise.all([callback(jobs)]), maxExpiration)	3✔
223	.then(() => this.complete(name, jobs.map(job => job.id)))	5✔
224	.catch(err => this.fail(name, jobs.map(job => job.id), err))	2✔
225	} else {
226	if (refill) {	41✔
227	queueSize += jobs.length \|\| 1	5!
228	}
229
230	const allTeamPromise = pMap(jobs, job =>	41✔
231	resolveWithinSeconds(callback(job), job.expire_in_seconds)	53✔
232	.then(result => this.complete(name, job.id, result))	44✔
233	.catch(err => this.fail(name, job.id, err))	16✔
234	.then(() => refill ? onRefill() : null)	46✔
235	, { concurrency: teamConcurrency }
236	).catch(() => {}) // allow promises & non-promises to live together in harmony
237
238	if (refill) {	41✔
239	if (queueSize < teamSize) {	5!
UNCOV 240	return	×
241	} else {
242	await refillTeamPromise	5✔
243	}
244	} else {
245	await allTeamPromise	36✔
246	}
247	}
248
249	this.emitWip(name)	44✔
250	}
251
252	const onError = error => {	40✔
253	this.emit(events.error, { ...error, message: error.message, stack: error.stack, queue: name, worker: id })	1✔
254	}
255
256	const worker = new Worker({ id, name, options, interval, fetch, onFetch, onError })	40✔
257
258	this.addWorker(worker)	40✔
259
260	worker.start()	40✔
261
262	return id	40✔
263	}
264
265	async offWork (value) {
266	assert(value, 'Missing required argument')	41✔
267
268	const query = (typeof value === 'string')	40✔
269	? { filter: i => i.name === value }	40✔
270	: (typeof value === 'object' && value.id)	3!
271	? { filter: i => i.id === value.id }	1✔
272	: null
273
274	assert(query, 'Invalid argument. Expected string or object: { id }')	40✔
275
276	const workers = this.getWorkers().filter(i => query.filter(i) && !i.stopping && !i.stopped)	41✔
277
278	if (workers.length === 0) {	40✔
279	return	3✔
280	}
281
282	for (const worker of workers) {	37✔
283	worker.stop()	38✔
284	}
285
286	setImmediate(async () => {	37✔
287	while (!workers.every(w => w.stopped)) {	68✔
288	await delay(1000)	30✔
289	}
290
291	for (const worker of workers) {	37✔
292	this.removeWorker(worker)	38✔
293	}
294	})
295	}
296
297	notifyWorker (workerId) {
298	if (this.workers.has(workerId)) {	1!
299	this.workers.get(workerId).notify()	1✔
300	}
301	}
302
303	async subscribe (event, name) {
304	assert(event, 'Missing required argument')	5✔
305	assert(name, 'Missing required argument')	5✔
306
307	return await this.db.executeSql(this.subscribeCommand, [event, name])	5✔
308	}
309
310	async unsubscribe (event, name) {
311	assert(event, 'Missing required argument')	4✔
312	assert(name, 'Missing required argument')	3✔
313
314	return await this.db.executeSql(this.unsubscribeCommand, [event, name])	2✔
315	}
316
317	async publish (event, ...args) {
318	assert(event, 'Missing required argument')	11✔
319
320	const result = await this.db.executeSql(this.getQueuesForEventCommand, [event])	10✔
321
322	if (!result \|\| result.rowCount === 0) {	10✔
323	return []	6✔
324	}
325
326	return await Promise.all(result.rows.map(({ name }) => this.send(name, ...args)))	6✔
327	}
328
329	async send (...args) {
330	const { name, data, options } = Attorney.checkSendArgs(args, this.config)	183✔
331	return await this.createJob(name, data, options)	180✔
332	}
333
334	async sendAfter (name, data, options, after) {
335	options = options ? { ...options } : {}	1!
336	options.startAfter = after	1✔
337
338	const result = Attorney.checkSendArgs([name, data, options], this.config)	1✔
339
340	return await this.createJob(result.name, result.data, result.options)	1✔
341	}
342
343	async sendThrottled (name, data, options, seconds, key) {
344	options = options ? { ...options } : {}	2!
345	options.singletonSeconds = seconds	2✔
346	options.singletonNextSlot = false	2✔
347	options.singletonKey = key	2✔
348
349	const result = Attorney.checkSendArgs([name, data, options], this.config)	2✔
350
351	return await this.createJob(result.name, result.data, result.options)	2✔
352	}
353
354	async sendDebounced (name, data, options, seconds, key) {
355	options = options ? { ...options } : {}	3!
356	options.singletonSeconds = seconds	3✔
357	options.singletonNextSlot = true	3✔
358	options.singletonKey = key	3✔
359
360	const result = Attorney.checkSendArgs([name, data, options], this.config)	3✔
361
362	return await this.createJob(result.name, result.data, result.options)	3✔
363	}
364
365	async createJob (name, data, options, singletonOffset = 0) {	186✔
366	const {
367	id = null,	189✔
368	db: wrapper,
369	priority,
370	startAfter,
371	singletonKey = null,	176✔
372	singletonSeconds,
373	deadLetter = null,	188✔
374	expireIn,
375	expireInDefault,
376	keepUntil,
377	keepUntilDefault,
378	retryLimit,
379	retryLimitDefault,
380	retryDelay,
381	retryDelayDefault,
382	retryBackoff,
383	retryBackoffDefault
384	} = options	189✔
385
386	const values = [	189✔
387	id, // 1
388	name, // 2
389	data, // 3
390	priority, // 4
391	startAfter, // 5
392	singletonKey, // 6
393	singletonSeconds, // 7
394	singletonOffset, // 8
395	deadLetter, // 9
396	expireIn, // 10
397	expireInDefault, // 11
398	keepUntil, // 12
399	keepUntilDefault, // 13
400	retryLimit, // 14
401	retryLimitDefault, // 15
402	retryDelay, // 16
403	retryDelayDefault, // 17
404	retryBackoff, // 18
405	retryBackoffDefault // 19
406	]
407
408	const db = wrapper \|\| this.db	189✔
409	const result = await db.executeSql(this.insertJobCommand, values)	189✔
410
411	if (result && result.rowCount === 1) {	189✔
412	return result.rows[0].id	171✔
413	}
414
415	if (!options.singletonNextSlot) {	18✔
416	return null	15✔
417	}
418
419	// delay starting by the offset to honor throttling config
420	options.startAfter = this.getDebounceStartAfter(singletonSeconds, this.timekeeper.clockSkew)	3✔
421
422	// toggle off next slot config for round 2
423	options.singletonNextSlot = false	3✔
424
425	singletonOffset = singletonSeconds	3✔
426
427	return await this.createJob(name, data, options, singletonOffset)	3✔
428	}
429
430	async insert (jobs, options = {}) {	3✔
431	assert(Array.isArray(jobs), 'jobs argument should be an array')	4✔
432
433	const db = options.db \|\| this.db	4✔
434
435	const params = [	4✔
436	JSON.stringify(jobs), // 1
437	this.config.expireIn, // 2
438	this.config.keepUntil, // 3
439	this.config.retryLimit, // 4
440	this.config.retryDelay, // 5
441	this.config.retryBackoff // 6
442	]
443
444	return await db.executeSql(this.insertJobsCommand, params)	4✔
445	}
446
447	getDebounceStartAfter (singletonSeconds, clockOffset) {
448	const debounceInterval = singletonSeconds * 1000	3✔
449
450	const now = Date.now() + clockOffset	3✔
451
452	const slot = Math.floor(now / debounceInterval) * debounceInterval	3✔
453
454	// prevent startAfter=0 during debouncing
455	let startAfter = (singletonSeconds - Math.floor((now - slot) / 1000)) \|\| 1	3!
456
457	if (singletonSeconds > 1) {	3!
458	startAfter++	3✔
459	}
460
461	return startAfter	3✔
462	}
463
464	async fetch (name, batchSize, options = {}) {	74✔
465	const values = Attorney.checkFetchArgs(name, batchSize, options)	26,076✔
466	const db = options.db \|\| this.db	26,075✔
467	const nextJobSql = this.nextJobCommand({ ...options })	26,075✔
468	const statementValues = [values.name, batchSize \|\| 1]	26,075✔
469
470	let result
471
472	try {	26,075✔
473	result = await db.executeSql(nextJobSql, statementValues)	26,075✔
474	} catch (err) {
475	// errors from fetchquery should only be unique constraint violations
476	}
477
478	if (!result \|\| result.rows.length === 0) {	26,075✔
479	return null	25,967✔
480	}
481
482	return result.rows.length === 1 && !batchSize ? result.rows[0] : result.rows	108✔
483	}
484
485	mapCompletionIdArg (id, funcName) {
486	const errorMessage = `${funcName}() requires an id`	99✔
487
488	assert(id, errorMessage)	99✔
489
490	const ids = Array.isArray(id) ? id : [id]	99✔
491
492	assert(ids.length, errorMessage)	99✔
493
494	return ids	99✔
495	}
496
497	mapCompletionDataArg (data) {
498	if (data === null \|\| typeof data === 'undefined' \|\| typeof data === 'function') { return null }	91✔
499
500	const result = (typeof data === 'object' && !Array.isArray(data))	36✔
501	? data
502	: { value: data }
503
504	return stringify(result)	36✔
505	}
506
507	mapCompletionResponse (ids, result) {
508	return {	85✔
509	jobs: ids,
510	requested: ids.length,
511	updated: result && result.rows ? parseInt(result.rows[0].count) : 0	255!
512	}
513	}
514
515	async complete (name, id, data, options = {}) {	58✔
516	assert(name, 'Missing queue name argument')	59✔
517	const db = options.db \|\| this.db	58✔
518	const ids = this.mapCompletionIdArg(id, 'complete')	58✔
519	const result = await db.executeSql(this.completeJobsCommand, [name, ids, this.mapCompletionDataArg(data)])	58✔
520	return this.mapCompletionResponse(ids, result)	51✔
521	}
522
523	async fail (name, id, data, options = {}) {	33✔
524	assert(name, 'Missing queue name argument')	34✔
525	const db = options.db \|\| this.db	33✔
526	const ids = this.mapCompletionIdArg(id, 'fail')	33✔
527	const result = await db.executeSql(this.failJobsByIdCommand, [name, ids, this.mapCompletionDataArg(data)])	33✔
528	return this.mapCompletionResponse(ids, result)	26✔
529	}
530
531	async cancel (name, id, options = {}) {	5✔
532	assert(name, 'Missing queue name argument')	7✔
533	const db = options.db \|\| this.db	6✔
534	const ids = this.mapCompletionIdArg(id, 'cancel')	6✔
535	const result = await db.executeSql(this.cancelJobsCommand, [name, ids])	6✔
536	return this.mapCompletionResponse(ids, result)	6✔
537	}
538
539	async resume (name, id, options = {}) {	2✔
540	assert(name, 'Missing queue name argument')	3✔
541	const db = options.db \|\| this.db	2✔
542	const ids = this.mapCompletionIdArg(id, 'resume')	2✔
543	const result = await db.executeSql(this.resumeJobsCommand, [name, ids])	2✔
544	return this.mapCompletionResponse(ids, result)	2✔
545	}
546
547	async createQueue (name, options = {}) {	148✔
548	assert(name, 'Missing queue name argument')	158✔
549
550	Attorney.assertQueueName(name)	158✔
551
552	const { policy = QUEUE_POLICY.standard } = options	158✔
553
554	assert(policy in QUEUE_POLICY, `${policy} is not a valid queue policy`)	158✔
555
556	const {
557	retryLimit,
558	retryDelay,
559	retryBackoff,
560	expireInSeconds,
561	retentionMinutes,
562	deadLetter
563	} = Attorney.checkQueueArgs(name, options)	157✔
564
565	const paritionSql = plans.createPartition(this.config.schema, name)	157✔
566
567	await this.db.executeSql(paritionSql)	157✔
568
569	const sql = plans.createQueue(this.config.schema, name)	157✔
570
571	const params = [	157✔
572	name,
573	policy,
574	retryLimit,
575	retryDelay,
576	retryBackoff,
577	expireInSeconds,
578	retentionMinutes,
579	deadLetter
580	]
581
582	await this.db.executeSql(sql, params)	157✔
583	}
584
585	async updateQueue (name, options = {}) {	×
586	assert(name, 'Missing queue name argument')	1✔
587
588	const {
589	retryLimit,
590	retryDelay,
591	retryBackoff,
592	expireInSeconds,
593	retentionMinutes,
594	deadLetter
595	} = Attorney.checkQueueArgs(name, options)	1✔
596
597	const sql = plans.updateQueue(this.config.schema)	1✔
598
599	const params = [	1✔
600	name,
601	retryLimit,
602	retryDelay,
603	retryBackoff,
604	expireInSeconds,
605	retentionMinutes,
606	deadLetter
607	]
608
609	await this.db.executeSql(sql, params)	1✔
610	}
611
612	async getQueue (name) {
613	assert(name, 'Missing queue name argument')	3✔
614
615	const sql = plans.getQueueByName(this.config.schema)	3✔
616	const result = await this.db.executeSql(sql, [name])	3✔
617
618	if (result.rows.length === 0) {	3✔
619	return null	1✔
620	}
621
622	const {
623	policy,
624	retry_limit: retryLimit,
625	retry_delay: retryDelay,
626	retry_backoff: retryBackoff,
627	expire_seconds: expireInSeconds,
628	retention_minutes: retentionMinutes,
629	dead_letter: deadLetter
630	} = result.rows[0]	2✔
631
632	return {	2✔
633	name,
634	policy,
635	retryLimit,
636	retryDelay,
637	retryBackoff,
638	expireInSeconds,
639	retentionMinutes,
640	deadLetter
641	}
642	}
643
644	async deleteQueue (name) {
645	assert(name, 'Missing queue name argument')	2✔
646
647	const queueSql = plans.getQueueByName(this.config.schema)	2✔
648	const result = await this.db.executeSql(queueSql, [name])	2✔
649
650	if (result?.rows?.length) {	2✔
651	Attorney.assertQueueName(name)	1✔
652	const sql = plans.dropPartition(this.config.schema, name)	1✔
653	await this.db.executeSql(sql)	1✔
654	}
655
656	const sql = plans.deleteQueueRecords(this.config.schema)	2✔
657	const result2 = await this.db.executeSql(sql, [name])	2✔
658	return result2?.rowCount \|\| null	2✔
659	}
660
661	async purgeQueue (queue) {
662	assert(queue, 'Missing queue name argument')	3✔
663	const sql = plans.purgeQueue(this.config.schema)	3✔
664	await this.db.executeSql(sql, [queue])	3✔
665	}
666
667	async clearStorage () {
668	const sql = plans.clearStorage(this.config.schema)	1✔
669	await this.db.executeSql(sql)	1✔
670	}
671
672	async getQueueSize (queue, options) {
673	assert(queue, 'Missing queue name argument')	6✔
674
675	const sql = plans.getQueueSize(this.config.schema, options)	6✔
676
677	const result = await this.db.executeSql(sql, [queue])	6✔
678
679	return result ? parseFloat(result.rows[0].count) : null	6!
680	}
681
682	async getJobById (queue, id, options = {}) {	29✔
683	const db = options.db \|\| this.db	31✔
684	const result1 = await db.executeSql(this.getJobByIdCommand, [queue, id])	31✔
685
686	if (result1 && result1.rows && result1.rows.length === 1) {	31✔
687	return result1.rows[0]	29✔
688	}
689
690	const result2 = await db.executeSql(this.getArchivedJobByIdCommand, [queue, id])	2✔
691
692	if (result2 && result2.rows && result2.rows.length === 1) {	2✔
693	return result2.rows[0]	1✔
694	}
695
696	return null	1✔
697	}
698	}
699
700	module.exports = Manager	4✔

timgit / pg-boss / 9949043480

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous