10016541428

Committed 20 Jul 2024 01:41AM UTC coverage: 93.17% (-6.8%) from 100.0%

Build # 10016541428

Build Type

Pull #425

github

Committed by

web-flow

Commit Message

Merge b0d7e7540 into f1c1636ca

Pull Request Pull Request #425: v10

Run Details

451 of 532 branches covered (84.77%)

357 of 381 new or added lines in 10 files covered. (93.7%)

40 existing lines in 5 files now uncovered.

873 of 937 relevant lines covered (93.17%)

805.36 hits per line

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

99.61

/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_POLICIES } = 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.expireInSeconds), 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.expireInSeconds)
            .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 { rows } = await this.db.executeSql(this.getQueuesForEventCommand, [event])

    return await Promise.all(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 { rows } = await db.executeSql(this.insertJobCommand, values)

    if (rows.length === 1) {
      return 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_POLICIES.standard } = options

    assert(policy in QUEUE_POLICIES, `${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 { rows } = await this.db.executeSql(queueSql, [name])

    if (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)
    await this.db.executeSql(sql, [name])
  }

  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_POLICIES } = 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	57✔
27	const reject = delay(timeout, `handler execution exceeded ${timeout}ms`)	57✔
28
29	let result
30
31	try {	57✔
32	result = await Promise.race([promise, reject])	57✔
33	} finally {
34	reject.abort()	57✔
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)	19✔
111	if (jobIds.length) {	19✔
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())	276✔
132	}
133
134	emitWip (name) {
135	if (!INTERNAL_QUEUES[name]) {	90✔
136	this.emitWipThrottled()	82✔
137	}
138	}
139
140	getWipData (options = {}) {	66✔
141	const { includeInternal = false } = options	237✔
142
143	const data = this.getWorkers()	237✔
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	}) => ({	87✔
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))	87!
170
171	return data	237✔
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,231✔
211
212	const onFetch = async (jobs) => {	40✔
213	if (this.config.__test__throw_worker) {	46✔
214	throw new Error('__test__throw_worker')	1✔
215	}
216
217	this.emitWip(name)	45✔
218
219	if (batchSize) {	45✔
220	const maxExpiration = jobs.reduce((acc, i) => Math.max(acc, i.expireInSeconds), 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) {	42✔
227	queueSize += jobs.length \|\| 1	5!
228	}
229
230	const allTeamPromise = pMap(jobs, job =>	42✔
231	resolveWithinSeconds(callback(job), job.expireInSeconds)	54✔
232	.then(result => this.complete(name, job.id, result))	44✔
233	.catch(err => this.fail(name, job.id, err))	17✔
234	.then(() => refill ? onRefill() : null)	47✔
235	, { concurrency: teamConcurrency }
236	).catch(() => {}) // allow promises & non-promises to live together in harmony
237
238	if (refill) {	42✔
239	if (queueSize < teamSize) {	5!
UNCOV 240	return	×
241	} else {
242	await refillTeamPromise	5✔
243	}
244	} else {
245	await allTeamPromise	37✔
246	}
247	}
248
249	this.emitWip(name)	45✔
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')	40✔
267
268	const query = (typeof value === 'string')	39✔
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 }')	39✔
275
276	const workers = this.getWorkers().filter(i => query.filter(i) && !i.stopping && !i.stopped)	41✔
277
278	if (workers.length === 0) {	39✔
279	return	2✔
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)) {	71✔
288	await delay(1000)	33✔
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 { rows } = await this.db.executeSql(this.getQueuesForEventCommand, [event])	10✔
321
322	return await Promise.all(rows.map(({ name }) => this.send(name, ...args)))	10✔
323	}
324
325	async send (...args) {
326	const { name, data, options } = Attorney.checkSendArgs(args, this.config)	186✔
327	return await this.createJob(name, data, options)	183✔
328	}
329
330	async sendAfter (name, data, options, after) {
331	options = options ? { ...options } : {}	1!
332	options.startAfter = after	1✔
333
334	const result = Attorney.checkSendArgs([name, data, options], this.config)	1✔
335
336	return await this.createJob(result.name, result.data, result.options)	1✔
337	}
338
339	async sendThrottled (name, data, options, seconds, key) {
340	options = options ? { ...options } : {}	2!
341	options.singletonSeconds = seconds	2✔
342	options.singletonNextSlot = false	2✔
343	options.singletonKey = key	2✔
344
345	const result = Attorney.checkSendArgs([name, data, options], this.config)	2✔
346
347	return await this.createJob(result.name, result.data, result.options)	2✔
348	}
349
350	async sendDebounced (name, data, options, seconds, key) {
351	options = options ? { ...options } : {}	3!
352	options.singletonSeconds = seconds	3✔
353	options.singletonNextSlot = true	3✔
354	options.singletonKey = key	3✔
355
356	const result = Attorney.checkSendArgs([name, data, options], this.config)	3✔
357
358	return await this.createJob(result.name, result.data, result.options)	3✔
359	}
360
361	async createJob (name, data, options, singletonOffset = 0) {	189✔
362	const {
363	id = null,	192✔
364	db: wrapper,
365	priority,
366	startAfter,
367	singletonKey = null,	176✔
368	singletonSeconds,
369	deadLetter = null,	191✔
370	expireIn,
371	expireInDefault,
372	keepUntil,
373	keepUntilDefault,
374	retryLimit,
375	retryLimitDefault,
376	retryDelay,
377	retryDelayDefault,
378	retryBackoff,
379	retryBackoffDefault
380	} = options	192✔
381
382	const values = [	192✔
383	id, // 1
384	name, // 2
385	data, // 3
386	priority, // 4
387	startAfter, // 5
388	singletonKey, // 6
389	singletonSeconds, // 7
390	singletonOffset, // 8
391	deadLetter, // 9
392	expireIn, // 10
393	expireInDefault, // 11
394	keepUntil, // 12
395	keepUntilDefault, // 13
396	retryLimit, // 14
397	retryLimitDefault, // 15
398	retryDelay, // 16
399	retryDelayDefault, // 17
400	retryBackoff, // 18
401	retryBackoffDefault // 19
402	]
403
404	const db = wrapper \|\| this.db	192✔
405	const { rows } = await db.executeSql(this.insertJobCommand, values)	192✔
406
407	if (rows.length === 1) {	191✔
408	return rows[0].id	171✔
409	}
410
411	if (!options.singletonNextSlot) {	20✔
412	return null	17✔
413	}
414
415	// delay starting by the offset to honor throttling config
416	options.startAfter = this.getDebounceStartAfter(singletonSeconds, this.timekeeper.clockSkew)	3✔
417
418	// toggle off next slot config for round 2
419	options.singletonNextSlot = false	3✔
420
421	singletonOffset = singletonSeconds	3✔
422
423	return await this.createJob(name, data, options, singletonOffset)	3✔
424	}
425
426	async insert (jobs, options = {}) {	3✔
427	assert(Array.isArray(jobs), 'jobs argument should be an array')	4✔
428
429	const db = options.db \|\| this.db	4✔
430
431	const params = [	4✔
432	JSON.stringify(jobs), // 1
433	this.config.expireIn, // 2
434	this.config.keepUntil, // 3
435	this.config.retryLimit, // 4
436	this.config.retryDelay, // 5
437	this.config.retryBackoff // 6
438	]
439
440	return await db.executeSql(this.insertJobsCommand, params)	4✔
441	}
442
443	getDebounceStartAfter (singletonSeconds, clockOffset) {
444	const debounceInterval = singletonSeconds * 1000	3✔
445
446	const now = Date.now() + clockOffset	3✔
447
448	const slot = Math.floor(now / debounceInterval) * debounceInterval	3✔
449
450	// prevent startAfter=0 during debouncing
451	let startAfter = (singletonSeconds - Math.floor((now - slot) / 1000)) \|\| 1	3!
452
453	if (singletonSeconds > 1) {	3!
454	startAfter++	3✔
455	}
456
457	return startAfter	3✔
458	}
459
460	async fetch (name, batchSize, options = {}) {	74✔
461	const values = Attorney.checkFetchArgs(name, batchSize, options)	25,308✔
462	const db = options.db \|\| this.db	25,307✔
463	const nextJobSql = this.nextJobCommand({ ...options })	25,307✔
464	const statementValues = [values.name, batchSize \|\| 1]	25,307✔
465
466	let result
467
468	try {	25,307✔
469	result = await db.executeSql(nextJobSql, statementValues)	25,307✔
470	} catch (err) {
471	// errors from fetchquery should only be unique constraint violations
472	}
473
474	if (!result \|\| result.rows.length === 0) {	25,307✔
475	return null	25,198✔
476	}
477
478	return result.rows.length === 1 && !batchSize ? result.rows[0] : result.rows	109✔
479	}
480
481	mapCompletionIdArg (id, funcName) {
482	const errorMessage = `${funcName}() requires an id`	100✔
483
484	assert(id, errorMessage)	100✔
485
486	const ids = Array.isArray(id) ? id : [id]	100✔
487
488	assert(ids.length, errorMessage)	100✔
489
490	return ids	100✔
491	}
492
493	mapCompletionDataArg (data) {
494	if (data === null \|\| typeof data === 'undefined' \|\| typeof data === 'function') { return null }	92✔
495
496	const result = (typeof data === 'object' && !Array.isArray(data))	37✔
497	? data
498	: { value: data }
499
500	return stringify(result)	37✔
501	}
502
503	mapCompletionResponse (ids, result) {
504	return {	86✔
505	jobs: ids,
506	requested: ids.length,
507	updated: result && result.rows ? parseInt(result.rows[0].count) : 0	258!
508	}
509	}
510
511	async complete (name, id, data, options = {}) {	58✔
512	assert(name, 'Missing queue name argument')	59✔
513	const db = options.db \|\| this.db	58✔
514	const ids = this.mapCompletionIdArg(id, 'complete')	58✔
515	const result = await db.executeSql(this.completeJobsCommand, [name, ids, this.mapCompletionDataArg(data)])	58✔
516	return this.mapCompletionResponse(ids, result)	51✔
517	}
518
519	async fail (name, id, data, options = {}) {	34✔
520	assert(name, 'Missing queue name argument')	35✔
521	const db = options.db \|\| this.db	34✔
522	const ids = this.mapCompletionIdArg(id, 'fail')	34✔
523	const result = await db.executeSql(this.failJobsByIdCommand, [name, ids, this.mapCompletionDataArg(data)])	34✔
524	return this.mapCompletionResponse(ids, result)	27✔
525	}
526
527	async cancel (name, id, options = {}) {	5✔
528	assert(name, 'Missing queue name argument')	7✔
529	const db = options.db \|\| this.db	6✔
530	const ids = this.mapCompletionIdArg(id, 'cancel')	6✔
531	const result = await db.executeSql(this.cancelJobsCommand, [name, ids])	6✔
532	return this.mapCompletionResponse(ids, result)	6✔
533	}
534
535	async resume (name, id, options = {}) {	2✔
536	assert(name, 'Missing queue name argument')	3✔
537	const db = options.db \|\| this.db	2✔
538	const ids = this.mapCompletionIdArg(id, 'resume')	2✔
539	const result = await db.executeSql(this.resumeJobsCommand, [name, ids])	2✔
540	return this.mapCompletionResponse(ids, result)	2✔
541	}
542
543	async createQueue (name, options = {}) {	148✔
544	assert(name, 'Missing queue name argument')	158✔
545
546	Attorney.assertQueueName(name)	158✔
547
548	const { policy = QUEUE_POLICIES.standard } = options	158✔
549
550	assert(policy in QUEUE_POLICIES, `${policy} is not a valid queue policy`)	158✔
551
552	const {
553	retryLimit,
554	retryDelay,
555	retryBackoff,
556	expireInSeconds,
557	retentionMinutes,
558	deadLetter
559	} = Attorney.checkQueueArgs(name, options)	157✔
560
561	const paritionSql = plans.createPartition(this.config.schema, name)	157✔
562
563	await this.db.executeSql(paritionSql)	157✔
564
565	const sql = plans.createQueue(this.config.schema, name)	157✔
566
567	const params = [	157✔
568	name,
569	policy,
570	retryLimit,
571	retryDelay,
572	retryBackoff,
573	expireInSeconds,
574	retentionMinutes,
575	deadLetter
576	]
577
578	await this.db.executeSql(sql, params)	157✔
579	}
580
581	async updateQueue (name, options = {}) {	×
582	assert(name, 'Missing queue name argument')	1✔
583
584	const {
585	retryLimit,
586	retryDelay,
587	retryBackoff,
588	expireInSeconds,
589	retentionMinutes,
590	deadLetter
591	} = Attorney.checkQueueArgs(name, options)	1✔
592
593	const sql = plans.updateQueue(this.config.schema)	1✔
594
595	const params = [	1✔
596	name,
597	retryLimit,
598	retryDelay,
599	retryBackoff,
600	expireInSeconds,
601	retentionMinutes,
602	deadLetter
603	]
604
605	await this.db.executeSql(sql, params)	1✔
606	}
607
608	async getQueue (name) {
609	assert(name, 'Missing queue name argument')	3✔
610
611	const sql = plans.getQueueByName(this.config.schema)	3✔
612	const result = await this.db.executeSql(sql, [name])	3✔
613
614	if (result.rows.length === 0) {	3✔
615	return null	1✔
616	}
617
618	const {
619	policy,
620	retry_limit: retryLimit,
621	retry_delay: retryDelay,
622	retry_backoff: retryBackoff,
623	expire_seconds: expireInSeconds,
624	retention_minutes: retentionMinutes,
625	dead_letter: deadLetter
626	} = result.rows[0]	2✔
627
628	return {	2✔
629	name,
630	policy,
631	retryLimit,
632	retryDelay,
633	retryBackoff,
634	expireInSeconds,
635	retentionMinutes,
636	deadLetter
637	}
638	}
639
640	async deleteQueue (name) {
641	assert(name, 'Missing queue name argument')	2✔
642
643	const queueSql = plans.getQueueByName(this.config.schema)	2✔
644	const { rows } = await this.db.executeSql(queueSql, [name])	2✔
645
646	if (rows.length) {	2✔
647	Attorney.assertQueueName(name)	1✔
648	const sql = plans.dropPartition(this.config.schema, name)	1✔
649	await this.db.executeSql(sql)	1✔
650	}
651
652	const sql = plans.deleteQueueRecords(this.config.schema)	2✔
653	await this.db.executeSql(sql, [name])	2✔
654	}
655
656	async purgeQueue (queue) {
657	assert(queue, 'Missing queue name argument')	3✔
658	const sql = plans.purgeQueue(this.config.schema)	3✔
659	await this.db.executeSql(sql, [queue])	3✔
660	}
661
662	async clearStorage () {
663	const sql = plans.clearStorage(this.config.schema)	1✔
664	await this.db.executeSql(sql)	1✔
665	}
666
667	async getQueueSize (queue, options) {
668	assert(queue, 'Missing queue name argument')	6✔
669
670	const sql = plans.getQueueSize(this.config.schema, options)	6✔
671
672	const result = await this.db.executeSql(sql, [queue])	6✔
673
674	return result ? parseFloat(result.rows[0].count) : null	6!
675	}
676
677	async getJobById (queue, id, options = {}) {	29✔
678	const db = options.db \|\| this.db	31✔
679	const result1 = await db.executeSql(this.getJobByIdCommand, [queue, id])	31✔
680
681	if (result1 && result1.rows && result1.rows.length === 1) {	31✔
682	return result1.rows[0]	29✔
683	}
684
685	const result2 = await db.executeSql(this.getArchivedJobByIdCommand, [queue, id])	2✔
686
687	if (result2 && result2.rows && result2.rows.length === 1) {	2✔
688	return result2.rows[0]	1✔
689	}
690
691	return null	1✔
692	}
693	}
694
695	module.exports = Manager	4✔

timgit / pg-boss / 10016541428

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