23638991080

Committed 27 Mar 2026 09:06AM UTC coverage: 79.332% (+0.001%) from 79.331%

Build # 23638991080

Build Type

push

github

Committed by

web-flow

Commit Message

feat: support for watt process runner (#917)

Run Details

5047 of 6664 branches covered (75.74%)

Branch coverage included in aggregate %.

14 of 18 new or added lines in 5 files covered. (77.78%)

3 existing lines in 2 files now uncovered.

28742 of 35928 relevant lines covered (80.0%)

587.97 hits per line

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

10.55

/src/internal/queue/queue.ts

import { ERRORS } from '@internal/errors'
import { QueueDB } from '@internal/queue/database'
import { Semaphore } from '@shopify/semaphore'
import PgBoss, { Db, Job, JobWithMetadata } from 'pg-boss'
import { getConfig } from '../../config'
import { logger, logSchema } from '../monitoring'
import { queueJobCompleted, queueJobError, queueJobRetryFailed } from '../monitoring/metrics'
import { Event } from './event'

type SubclassOfBaseClass = (new (
  payload: any
) => Event<any>) & {
  [K in keyof typeof Event]: (typeof Event)[K]
}

export const PG_BOSS_SCHEMA = 'pgboss_v10'

export abstract class Queue {
  protected static events: SubclassOfBaseClass[] = []
  private static pgBoss?: PgBoss
  private static pgBossDb?: PgBoss.Db

  static createPgBoss(opts: { db: Db; enableWorkers: boolean }) {
    const {
      isMultitenant,
      databaseURL,
      multitenantDatabasePoolUrl,
      multitenantDatabaseUrl,
      pgQueueConnectionURL,
      pgQueueArchiveCompletedAfterSeconds,
      pgQueueDeleteAfterDays,
      pgQueueDeleteAfterHours,
      pgQueueRetentionDays,
    } = getConfig()

    let url = pgQueueConnectionURL ?? databaseURL
    let migrate = true

    if (isMultitenant && !pgQueueConnectionURL) {
      if (!multitenantDatabaseUrl) {
        throw new Error(
          'running storage in multi-tenant but DB_MULTITENANT_DATABASE_URL is not set'
        )
      }
      url = multitenantDatabasePoolUrl || multitenantDatabaseUrl

      if (multitenantDatabasePoolUrl) {
        migrate = false
      }
    }

    return new PgBoss({
      connectionString: url,
      migrate,
      db: opts.db,
      schema: PG_BOSS_SCHEMA,
      ...(pgQueueDeleteAfterHours
        ? { deleteAfterHours: pgQueueDeleteAfterHours }
        : { deleteAfterDays: pgQueueDeleteAfterDays }),
      archiveCompletedAfterSeconds: pgQueueArchiveCompletedAfterSeconds,
      retentionDays: pgQueueRetentionDays,
      retryBackoff: true,
      retryLimit: 20,
      expireInHours: 23,
      maintenanceIntervalSeconds: 60 * 5,
      schedule: opts.enableWorkers,
      supervise: opts.enableWorkers,
    })
  }

  static async start(opts: {
    signal?: AbortSignal
    onMessage?: (job: Job) => void
    registerWorkers?: () => void
  }) {
    if (Queue.pgBoss) {
      return Queue.pgBoss
    }

    if (opts.signal?.aborted) {
      throw ERRORS.Aborted('Cannot start queue with aborted signal')
    }

    const {
      isMultitenant,
      databaseURL,
      multitenantDatabaseUrl,
      multitenantDatabasePoolUrl,
      pgQueueConnectionURL,
      pgQueueEnableWorkers,
      pgQueueReadWriteTimeout,
      pgQueueConcurrentTasksPerQueue,
      pgQueueMaxConnections,
    } = getConfig()

    let url = pgQueueConnectionURL || databaseURL

    if (isMultitenant && !pgQueueConnectionURL) {
      if (!multitenantDatabaseUrl && !multitenantDatabasePoolUrl) {
        throw new Error(
          'running storage in multi-tenant but DB_MULTITENANT_DATABASE_URL is not set'
        )
      }
      url = (multitenantDatabasePoolUrl || multitenantDatabaseUrl) as string
    }

    Queue.pgBossDb = new QueueDB({
      min: 0,
      max: pgQueueMaxConnections,
      connectionString: url,
      statement_timeout: pgQueueReadWriteTimeout > 0 ? pgQueueReadWriteTimeout : undefined,
    })

    Queue.pgBoss = this.createPgBoss({
      db: Queue.pgBossDb,
      enableWorkers: pgQueueEnableWorkers !== false,
    })

    Queue.pgBoss.on('error', (error) => {
      logSchema.error(logger, '[Queue] error', {
        type: 'queue',
        error,
      })
    })

    await Queue.pgBoss.start()

    if (opts.registerWorkers && pgQueueEnableWorkers) {
      opts.registerWorkers()
    }

    await Queue.callStart()
    await Queue.startWorkers({
      maxConcurrentTasks: pgQueueConcurrentTasksPerQueue,
      onMessage: opts.onMessage,
      signal: opts.signal,
    })

    if (opts.signal) {
      opts.signal.addEventListener(
        'abort',
        async () => {
          logSchema.info(logger, '[Queue] Stopping', {
            type: 'queue',
          })
          return Queue.stop()
            .then(async () => {
              logSchema.info(logger, '[Queue] Exited', {
                type: 'queue',
              })
            })
            .catch((e) => {
              logSchema.error(logger, '[Queue] Error while stopping queue', {
                error: e,
                type: 'queue',
              })
            })
            .finally(async () => {
              await Queue.callClose().catch(() => {
                // no-op
              })
            })
        },
        { once: true }
      )
    }

    return Queue.pgBoss
  }

  static getInstance() {
    if (!this.pgBoss) {
      throw new Error('pg boss not initialised')
    }

    return this.pgBoss
  }

  static getDb() {
    if (!this.pgBossDb) {
      throw new Error('pg boss not initialised')
    }

    return this.pgBossDb
  }

  static register<T extends SubclassOfBaseClass>(event: T) {
    Queue.events.push(event)
  }

  static async stop() {
    if (!this.pgBoss) {
      return
    }

    const boss = this.pgBoss
    const { isProduction } = getConfig()

    await boss.stop({
      timeout: 20 * 1000,
      graceful: isProduction,
      wait: true,
    })

    await this.callClose()

    Queue.pgBoss = undefined
  }

  protected static async startWorkers(opts: {
    maxConcurrentTasks: number
    signal?: AbortSignal
    onMessage?: (job: Job) => void
  }) {
    for (const event of Queue.events) {
      await Queue.registerTask(event, opts.maxConcurrentTasks, opts.onMessage, opts.signal)
    }
  }

  protected static callStart() {
    const events = Queue.events.map((event) => {
      return event.onStart()
    })

    return Promise.all(events)
  }

  protected static callClose() {
    const events = Queue.events.map((event) => {
      return event.onClose()
    })

    return Promise.all(events)
  }

  protected static async registerTask(
    event: SubclassOfBaseClass,
    maxConcurrentTasks: number,
    onMessage?: (job: Job) => void,
    signal?: AbortSignal
  ) {
    const queueName = event.getQueueName()
    const deadLetterName = event.deadLetterQueueName()

    const concurrentTaskCount = event.getWorkerOptions().concurrentTaskCount || maxConcurrentTasks
    try {
      // Create dead-letter queue and the normal queue
      const queueOptions = {
        policy: 'standard',
        ...event.getQueueOptions(),
      } as const

      // dead-letter
      await this.pgBoss?.createQueue(deadLetterName, {
        ...queueOptions,
        name: deadLetterName,
        retentionDays: 30,
        retryBackoff: true,
      })

      // // normal queue
      await this.pgBoss?.createQueue(queueName, {
        ...queueOptions,
        name: queueName,
        deadLetter: deadLetterName,
      })
    } catch {
      // no-op
    }

    return this.pollQueue(event, {
      concurrentTaskCount,
      onMessage,
      signal,
    })
  }

  protected static pollQueue(
    event: SubclassOfBaseClass,
    queueOpts: {
      concurrentTaskCount: number
      onMessage?: (job: Job) => void
      signal?: AbortSignal
    }
  ) {
    const semaphore = new Semaphore(queueOpts.concurrentTaskCount)
    const pollingInterval = (event.getWorkerOptions().pollingIntervalSeconds || 5) * 1000
    const batchSize =
      event.getWorkerOptions().batchSize ||
      queueOpts.concurrentTaskCount + Math.max(1, Math.floor(queueOpts.concurrentTaskCount * 1.2))

    logSchema.info(logger, `[Queue] Polling queue ${event.getQueueName()}`, {
      type: 'queue',
      metadata: JSON.stringify({
        queueName: event.getQueueName(),
        batchSize,
        pollingInterval,
      }),
    })

    let started = false
    let jobFetched = 0

    const interval = setInterval(async () => {
      if (started) {
        return
      }

      try {
        started = true
        const defaultFetch = {
          includeMetadata: true,
          batchSize,
        }

        const currentBatch = defaultFetch.batchSize - jobFetched

        if (currentBatch <= 0) {
          return
        }

        const jobs = await this.pgBoss?.fetch(event.getQueueName(), {
          ...event.getWorkerOptions(),
          ...defaultFetch,
          batchSize: currentBatch,
        })

        jobFetched += jobs?.length || 0

        if (jobFetched < defaultFetch.batchSize) {
          started = false
        }

        if (queueOpts.signal?.aborted) {
          started = false
          return
        }

        if (!jobs || (jobs && jobs.length === 0)) {
          started = false
          return
        }

        await Promise.allSettled(
          jobs.map(async (job) => {
            const lock = await semaphore.acquire()
            try {
              queueOpts.onMessage?.(job as Job)

              await event.handle(job, { signal: queueOpts.signal })

              await this.pgBoss?.complete(event.getQueueName(), job.id)
              queueJobCompleted.add(1, {
                name: event.getQueueName(),
              })
            } catch (e) {
              queueJobRetryFailed.add(1, {
                name: event.getQueueName(),
              })

              await this.pgBoss?.fail(event.getQueueName(), job.id)

              try {
                const dbJob: JobWithMetadata | null =
                  (job as JobWithMetadata).priority !== undefined
                    ? (job as JobWithMetadata)
                    : await Queue.getInstance().getJobById(event.getQueueName(), job.id)

                if (!dbJob) {
                  return
                }
                if (dbJob.retryCount >= dbJob.retryLimit) {
                  queueJobError.add(1, {
                    name: event.getQueueName(),
                  })
                }
              } catch (e) {
                logSchema.error(logger, `[Queue Handler] fetching job ${event.name}`, {
                  type: 'queue-task',
                  error: e,
                  metadata: JSON.stringify(job),
                })
              }

              logSchema.error(logger, `[Queue Handler] Error while processing job ${event.name}`, {
                type: 'queue-task',
                error: e,
                metadata: JSON.stringify(job),
              })

              throw e
            } finally {
              jobFetched = Math.max(0, jobFetched - 1)
              await lock.release()
            }
          })
        )
      } catch (e) {
        logSchema.error(logger, `[Queue] Error while polling queue ${event.name}`, {
          type: 'queue',
          error: e,
          metadata: JSON.stringify({
            queueName: event.getQueueName(),
            batchSize,
            pollingInterval,
          }),
        })
      } finally {
        started = false
      }
    }, pollingInterval)

    queueOpts.signal?.addEventListener('abort', () => {
      clearInterval(interval)
    })
  }
}

1	import { ERRORS } from '@internal/errors'	2✔
2	import { QueueDB } from '@internal/queue/database'	2✔
3	import { Semaphore } from '@shopify/semaphore'	2✔
4	import PgBoss, { Db, Job, JobWithMetadata } from 'pg-boss'	2✔
5	import { getConfig } from '../../config'	2✔
6	import { logger, logSchema } from '../monitoring'	2✔
7	import { queueJobCompleted, queueJobError, queueJobRetryFailed } from '../monitoring/metrics'	2✔
8	import { Event } from './event'	2✔
9		2✔
10	type SubclassOfBaseClass = (new (	2✔
11	payload: any	2✔
12	) => Event<any>) & {	2✔
13	[K in keyof typeof Event]: (typeof Event)[K]	2✔
14	}	2✔
15		2✔
16	export const PG_BOSS_SCHEMA = 'pgboss_v10'	2✔
17		2✔
18	export abstract class Queue {	2✔
19	protected static events: SubclassOfBaseClass[] = []	2✔
20	private static pgBoss?: PgBoss	2✔
21	private static pgBossDb?: PgBoss.Db	2✔
22		2✔
23	static createPgBoss(opts: { db: Db; enableWorkers: boolean }) {	2✔
24	const {	×
25	isMultitenant,	×
26	databaseURL,	×
27	multitenantDatabasePoolUrl,	×
28	multitenantDatabaseUrl,	×
29	pgQueueConnectionURL,	×
30	pgQueueArchiveCompletedAfterSeconds,	×
31	pgQueueDeleteAfterDays,	×
32	pgQueueDeleteAfterHours,	×
33	pgQueueRetentionDays,	×
34	} = getConfig()	×
35		×
36	let url = pgQueueConnectionURL ?? databaseURL	×
37	let migrate = true	×
38		×
39	if (isMultitenant && !pgQueueConnectionURL) {	×
40	if (!multitenantDatabaseUrl) {	×
41	throw new Error(	×
42	'running storage in multi-tenant but DB_MULTITENANT_DATABASE_URL is not set'	×
43	)	×
44	}	×
45	url = multitenantDatabasePoolUrl \|\| multitenantDatabaseUrl	×
46		×
47	if (multitenantDatabasePoolUrl) {	×
48	migrate = false	×
49	}	×
50	}	×
51		×
52	return new PgBoss({	×
53	connectionString: url,	×
54	migrate,	×
55	db: opts.db,	×
56	schema: PG_BOSS_SCHEMA,	×
57	...(pgQueueDeleteAfterHours	×
58	? { deleteAfterHours: pgQueueDeleteAfterHours }	×
59	: { deleteAfterDays: pgQueueDeleteAfterDays }),	×
60	archiveCompletedAfterSeconds: pgQueueArchiveCompletedAfterSeconds,	×
61	retentionDays: pgQueueRetentionDays,	×
62	retryBackoff: true,	×
63	retryLimit: 20,	×
64	expireInHours: 23,	×
65	maintenanceIntervalSeconds: 60 * 5,	×
66	schedule: opts.enableWorkers,	×
67	supervise: opts.enableWorkers,	×
68	})	×
69	}	×
70		2✔
71	static async start(opts: {	2✔
72	signal?: AbortSignal	×
73	onMessage?: (job: Job) => void	×
74	registerWorkers?: () => void	×
75	}) {	×
76	if (Queue.pgBoss) {	×
77	return Queue.pgBoss	×
78	}	×
79		×
80	if (opts.signal?.aborted) {	×
81	throw ERRORS.Aborted('Cannot start queue with aborted signal')	×
82	}	×
83		×
84	const {	×
85	isMultitenant,	×
86	databaseURL,	×
87	multitenantDatabaseUrl,	×
88	multitenantDatabasePoolUrl,	×
89	pgQueueConnectionURL,	×
90	pgQueueEnableWorkers,	×
91	pgQueueReadWriteTimeout,	×
92	pgQueueConcurrentTasksPerQueue,	×
93	pgQueueMaxConnections,	×
94	} = getConfig()	×
95		×
96	let url = pgQueueConnectionURL \|\| databaseURL	×
97		×
98	if (isMultitenant && !pgQueueConnectionURL) {	×
99	if (!multitenantDatabaseUrl && !multitenantDatabasePoolUrl) {	×
100	throw new Error(	×
101	'running storage in multi-tenant but DB_MULTITENANT_DATABASE_URL is not set'	×
102	)	×
103	}	×
104	url = (multitenantDatabasePoolUrl \|\| multitenantDatabaseUrl) as string	×
105	}	×
106		×
107	Queue.pgBossDb = new QueueDB({	×
108	min: 0,	×
109	max: pgQueueMaxConnections,	×
110	connectionString: url,	×
111	statement_timeout: pgQueueReadWriteTimeout > 0 ? pgQueueReadWriteTimeout : undefined,	×
112	})	×
113		×
114	Queue.pgBoss = this.createPgBoss({	×
115	db: Queue.pgBossDb,	×
116	enableWorkers: pgQueueEnableWorkers !== false,	×
117	})	×
118		×
119	Queue.pgBoss.on('error', (error) => {	×
120	logSchema.error(logger, '[Queue] error', {	×
121	type: 'queue',	×
122	error,	×
123	})	×
124	})	×
125		×
126	await Queue.pgBoss.start()	×
127		×
128	if (opts.registerWorkers && pgQueueEnableWorkers) {	×
129	opts.registerWorkers()	×
130	}	×
131		×
132	await Queue.callStart()	×
133	await Queue.startWorkers({	×
134	maxConcurrentTasks: pgQueueConcurrentTasksPerQueue,	×
135	onMessage: opts.onMessage,	×
136	signal: opts.signal,	×
137	})	×
138		×
139	if (opts.signal) {	×
140	opts.signal.addEventListener(	×
141	'abort',	×
142	async () => {	×
143	logSchema.info(logger, '[Queue] Stopping', {	×
144	type: 'queue',	×
145	})	×
146	return Queue.stop()	×
147	.then(async () => {	×
148	logSchema.info(logger, '[Queue] Exited', {	×
149	type: 'queue',	×
150	})	×
151	})	×
152	.catch((e) => {	×
153	logSchema.error(logger, '[Queue] Error while stopping queue', {	×
154	error: e,	×
155	type: 'queue',	×
156	})	×
157	})	×
158	.finally(async () => {	×
159	await Queue.callClose().catch(() => {	×
160	// no-op	×
161	})	×
162	})	×
163	},	×
164	{ once: true }	×
165	)	×
166	}	×
167		×
168	return Queue.pgBoss	×
169	}	×
170		2✔
171	static getInstance() {	2✔
172	if (!this.pgBoss) {	×
173	throw new Error('pg boss not initialised')	×
174	}	×
175		×
176	return this.pgBoss	×
177	}	×
178		2✔
179	static getDb() {	2✔
180	if (!this.pgBossDb) {	×
181	throw new Error('pg boss not initialised')	×
182	}	×
183		×
184	return this.pgBossDb	×
185	}	×
186		2✔
187	static register<T extends SubclassOfBaseClass>(event: T) {	2✔
188	Queue.events.push(event)	×
189	}	×
190		2✔
191	static async stop() {	2✔
192	if (!this.pgBoss) {	×
193	return	×
194	}	×
195		×
196	const boss = this.pgBoss	×
197	const { isProduction } = getConfig()	×
198		×
199	await boss.stop({	×
200	timeout: 20 * 1000,	×
201	graceful: isProduction,	×
202	wait: true,	×
203	})	×
204		×
NEW 205	await this.callClose()	×
206		×
207	Queue.pgBoss = undefined	×
208	}	×
209		2✔
210	protected static async startWorkers(opts: {	2✔
211	maxConcurrentTasks: number	×
212	signal?: AbortSignal	×
213	onMessage?: (job: Job) => void	×
214	}) {	×
215	for (const event of Queue.events) {	×
216	await Queue.registerTask(event, opts.maxConcurrentTasks, opts.onMessage, opts.signal)	×
217	}	×
218	}	×
219		2✔
220	protected static callStart() {	2✔
221	const events = Queue.events.map((event) => {	×
222	return event.onStart()	×
223	})	×
224		×
225	return Promise.all(events)	×
226	}	×
227		2✔
228	protected static callClose() {	2✔
229	const events = Queue.events.map((event) => {	×
230	return event.onClose()	×
231	})	×
232		×
233	return Promise.all(events)	×
234	}	×
235		2✔
236	protected static async registerTask(	2✔
237	event: SubclassOfBaseClass,	×
238	maxConcurrentTasks: number,	×
239	onMessage?: (job: Job) => void,	×
240	signal?: AbortSignal	×
241	) {	×
242	const queueName = event.getQueueName()	×
243	const deadLetterName = event.deadLetterQueueName()	×
244		×
245	const concurrentTaskCount = event.getWorkerOptions().concurrentTaskCount \|\| maxConcurrentTasks	×
246	try {	×
247	// Create dead-letter queue and the normal queue	×
248	const queueOptions = {	×
249	policy: 'standard',	×
250	...event.getQueueOptions(),	×
251	} as const	×
252		×
253	// dead-letter	×
254	await this.pgBoss?.createQueue(deadLetterName, {	×
255	...queueOptions,	×
256	name: deadLetterName,	×
257	retentionDays: 30,	×
258	retryBackoff: true,	×
259	})	×
260		×
261	// // normal queue	×
262	await this.pgBoss?.createQueue(queueName, {	×
263	...queueOptions,	×
264	name: queueName,	×
265	deadLetter: deadLetterName,	×
266	})	×
267	} catch {	×
268	// no-op	×
269	}	×
270		×
271	return this.pollQueue(event, {	×
272	concurrentTaskCount,	×
273	onMessage,	×
274	signal,	×
275	})	×
276	}	×
277		2✔
278	protected static pollQueue(	2✔
279	event: SubclassOfBaseClass,	×
280	queueOpts: {	×
281	concurrentTaskCount: number	×
282	onMessage?: (job: Job) => void	×
283	signal?: AbortSignal	×
284	}	×
285	) {	×
286	const semaphore = new Semaphore(queueOpts.concurrentTaskCount)	×
287	const pollingInterval = (event.getWorkerOptions().pollingIntervalSeconds \|\| 5) * 1000	×
288	const batchSize =	×
289	event.getWorkerOptions().batchSize \|\|	×
290	queueOpts.concurrentTaskCount + Math.max(1, Math.floor(queueOpts.concurrentTaskCount * 1.2))	×
291		×
292	logSchema.info(logger, `[Queue] Polling queue ${event.getQueueName()}`, {	×
293	type: 'queue',	×
294	metadata: JSON.stringify({	×
295	queueName: event.getQueueName(),	×
296	batchSize,	×
297	pollingInterval,	×
298	}),	×
299	})	×
300		×
301	let started = false	×
302	let jobFetched = 0	×
303		×
304	const interval = setInterval(async () => {	×
305	if (started) {	×
306	return	×
307	}	×
308		×
309	try {	×
310	started = true	×
311	const defaultFetch = {	×
312	includeMetadata: true,	×
313	batchSize,	×
314	}	×
315		×
316	const currentBatch = defaultFetch.batchSize - jobFetched	×
317		×
318	if (currentBatch <= 0) {	×
319	return	×
320	}	×
321		×
322	const jobs = await this.pgBoss?.fetch(event.getQueueName(), {	×
323	...event.getWorkerOptions(),	×
324	...defaultFetch,	×
325	batchSize: currentBatch,	×
326	})	×
327		×
328	jobFetched += jobs?.length \|\| 0	×
329		×
330	if (jobFetched < defaultFetch.batchSize) {	×
331	started = false	×
332	}	×
333		×
334	if (queueOpts.signal?.aborted) {	×
335	started = false	×
336	return	×
337	}	×
338		×
339	if (!jobs \|\| (jobs && jobs.length === 0)) {	×
340	started = false	×
341	return	×
342	}	×
343		×
344	await Promise.allSettled(	×
345	jobs.map(async (job) => {	×
346	const lock = await semaphore.acquire()	×
347	try {	×
348	queueOpts.onMessage?.(job as Job)	×
349		×
350	await event.handle(job, { signal: queueOpts.signal })	×
351		×
352	await this.pgBoss?.complete(event.getQueueName(), job.id)	×
353	queueJobCompleted.add(1, {	×
354	name: event.getQueueName(),	×
355	})	×
356	} catch (e) {	×
357	queueJobRetryFailed.add(1, {	×
358	name: event.getQueueName(),	×
359	})	×
360		×
361	await this.pgBoss?.fail(event.getQueueName(), job.id)	×
362		×
363	try {	×
364	const dbJob: JobWithMetadata \| null =	×
365	(job as JobWithMetadata).priority !== undefined	×
366	? (job as JobWithMetadata)	×
367	: await Queue.getInstance().getJobById(event.getQueueName(), job.id)	×
368		×
369	if (!dbJob) {	×
370	return	×
371	}	×
372	if (dbJob.retryCount >= dbJob.retryLimit) {	×
373	queueJobError.add(1, {	×
374	name: event.getQueueName(),	×
375	})	×
376	}	×
377	} catch (e) {	×
378	logSchema.error(logger, `[Queue Handler] fetching job ${event.name}`, {	×
379	type: 'queue-task',	×
380	error: e,	×
381	metadata: JSON.stringify(job),	×
382	})	×
383	}	×
384		×
385	logSchema.error(logger, `[Queue Handler] Error while processing job ${event.name}`, {	×
386	type: 'queue-task',	×
387	error: e,	×
388	metadata: JSON.stringify(job),	×
389	})	×
390		×
391	throw e	×
392	} finally {	×
393	jobFetched = Math.max(0, jobFetched - 1)	×
394	await lock.release()	×
395	}	×
396	})	×
397	)	×
398	} catch (e) {	×
399	logSchema.error(logger, `[Queue] Error while polling queue ${event.name}`, {	×
400	type: 'queue',	×
401	error: e,	×
402	metadata: JSON.stringify({	×
403	queueName: event.getQueueName(),	×
404	batchSize,	×
405	pollingInterval,	×
406	}),	×
407	})	×
408	} finally {	×
409	started = false	×
410	}	×
411	}, pollingInterval)	×
412		×
413	queueOpts.signal?.addEventListener('abort', () => {	×
414	clearInterval(interval)	×
415	})	×
416	}	×
417	}	2✔

supabase / storage / 23638991080

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