30832958929

Committed 03 Aug 2026 04:36PM UTC coverage: 53.989% (-26.5%) from 80.494%

Build # 30832958929

Build Type

Pull #1292

github

Committed by

web-flow

Commit Message

Merge 477971401 into 42e86a3a7

Pull Request Pull Request #1292: feat: new queing system

Coverage Stats

3711 of 7412 branches covered (50.07%)

Branch coverage included in aggregate %.

140 of 337 new or added lines in 38 files covered. (41.54%)

3365 existing lines in 150 files now uncovered.

6941 of 12318 relevant lines covered (56.35%)

77.58 hits per line

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

29.49

/src/internal/queue/instance.ts

import { ERRORS } from '@internal/errors'
import { logger, logSchema } from '@internal/monitoring'
import type { PgBossCtx } from '@supabase-labs/wave-adapter-pgboss'
import { pgboss } from '@supabase-labs/wave-adapter-pgboss'
import type { AnyWave, HandlerFor, TopicMap, TopicRegistry, Wave } from '@supabase-labs/wave-core'
import { createSyncWave, createWave } from '@supabase-labs/wave-core'
import type { PgBoss } from 'pg-boss'
import { getConfig } from '../../config'
import { createQueueBoss, queueDefaults } from './boss'
import { schedulingMetrics, syncFallback, syncModeGuard, tenantDisableEvents } from './middleware'

const { pgQueueEnable, pgQueueEnableWorkers, pgQueueSchemaV2 } = getConfig()

/** A storage wave: the app's typed produce/invoke surface over the pgboss adapter. */
export type StorageWave<M extends TopicMap> = Wave<M, PgBossCtx>

/**
 * What the events layer supplies at startup — the queue layer knows no concrete events beyond
 * the `meta` each topic declares alongside its pg-boss extension (both merged onto core's
 * `CreateTopicOptions` — see `storage/events/topics.ts`); the middlewares below read it
 * straight off `setup.topics`, so there is no separate lookup to wire.
 */
export interface QueueOptions<M extends TopicMap> {
  topics: TopicRegistry<M>
  handlers: ReadonlyArray<HandlerFor<M>>
  /** Optional teardown for handler-owned resources (v1's per-event `onClose`). */
  onStop?: () => Promise<void>
}

const queueStopTimeoutMs = 25_000

/** The held singleton is topic-erased (module state can't be generic): `AnyWave` is wave's
 * erased-holder type — every concrete `Wave<...>` assigns here cast-free (ADR-0062), and the
 * typed surface is restored by the accessors below (`startQueue`'s return, `getWave<M>`). */
let instance: AnyWave | undefined
let stopping: Promise<void> | undefined

const boss = createQueueBoss({ enableWorkers: pgQueueEnableWorkers ?? true })

/**
 * The queue layer is a singleton, so the app's events layer can call `queue()` from anywhere and
 * get the same instance. The first call must provide the topics and handlers; subsequent calls
 * may omit them (or provide the same values).
 * @param opts
 */
function createSyncWaveInstance<M extends TopicMap>(opts: QueueOptions<M>) {
  return createSyncWave({
    topics: opts.topics,
    handlers: opts.handlers,
    middleware: [tenantDisableEvents(), syncModeGuard()],
  })
}

function createWaveInstance<M extends TopicMap>(opts: QueueOptions<M>): Wave<TopicRegistry<M>> {
  const workersEnabled = pgQueueEnableWorkers ?? true
  const { topics, handlers } = opts

  if (!pgQueueEnable) {
    return createSyncWaveInstance(opts)
  }

  const pgBossAdapter = pgboss({
    boss,
    schema: pgQueueSchemaV2,
    queue: queueDefaults(),
  })

  return createWave(pgBossAdapter, {
    topics,
    handlers: workersEnabled ? handlers : [],
    middleware: [tenantDisableEvents(), schedulingMetrics(), syncFallback()],
    pollIdleIntervalMs: 5_000,
    closeTimeout: 20_000,
  })
}

/**
 * Start the queue, idempotently. Three shapes, all behind the same `Wave` interface:
 *
 * - `PG_QUEUE_ENABLE=false` → a sync wave IS the app's wave: every produce runs its handler
 *   inline (nested sends included), no queue connection exists at all (v1 parity).
 * - queue enabled, workers enabled → full wave: producers append, workers consume, and a
 *   handler-attached sync wave backs the produce-failure fallback.
 * - queue enabled, workers disabled (producer-only API nodes) → same, minus consuming: the
 *   wave gets no handlers, while the sync wave still attaches them so the fallback can run.
 */
export async function startQueue<M extends TopicMap>(
  queueOpts: QueueOptions<M>,
  opts: { signal?: AbortSignal } = {}
): Promise<StorageWave<M>> {
  const { topics, handlers } = queueOpts

  if (instance) {
    return instance as StorageWave<M>
  }
  if (opts.signal?.aborted) {
    throw ERRORS.Aborted('Cannot start queue with aborted signal')
  }

  instance = createWaveInstance({
    topics,
    handlers,
    onStop: queueOpts.onStop,
  })

  await instance.start()

  if (opts.signal) {
    opts.signal.addEventListener(
      'abort',
      () => {
        logSchema.info(logger, '[Queue] Stopping', { type: 'queue' })
        stopQueue(queueOpts.onStop)
          .then(() => {
            logSchema.info(logger, '[Queue] Exited', { type: 'queue' })
          })
          .catch((e) => {
            logSchema.error(logger, '[Queue] Error while stopping queue', {
              error: e,
              type: 'queue',
            })
          })
      },
      { once: true }
    )
  }

  return instance as StorageWave<M>
}

/** The running wave. Producers reach it through the typed accessor in storage/events (or,
 * from inside the topic registry's own module graph, this generic accessor with a type-only
 * import of the registry's topic map — avoiding the runtime import cycle a value import would
 * create). */
export function getWave<M extends TopicMap>(): StorageWave<M> {
  if (!instance) {
    throw new Error('queue is not started (call startQueue first)')
  }
  return instance as StorageWave<M>
}

export function isQueueStarted(): boolean {
  return instance !== undefined
}

let bossForTesting: Pick<PgBoss, 'getQueueStats'> | undefined

/** TEST SEAM: install a stub wave as the running instance, and optionally a stats backend
 * consulted by `queueSize`. Test stubs are partial by nature, so the widening cast lives
 * HERE, once — callers pass plain objects with just the members their test touches. */
export function setWaveForTesting(
  wave: Partial<AnyWave>,
  boss?: Pick<PgBoss, 'getQueueStats'>
): void {
  instance = wave as AnyWave
  bossForTesting = boss
}

/** TEST SEAM: build and install the sync wave — the exact shape `startQueue` uses when the
 * queue is env-disabled — regardless of `pgQueueEnable`, so a test file that flips the flag on
 * (to exercise queue-enabled app branches) still never constructs a real pg-boss. */
export async function startSyncWaveForTesting<M extends TopicMap>(
  opts: QueueOptions<M>
): Promise<StorageWave<M>> {
  const wave = createSyncWaveInstance(opts)
  await wave.start()
  instance = wave
  return instance as StorageWave<M>
}

/** Pending jobs on a topic's bare queue (v1 `Queue.getQueueSize`, via pg-boss v12 queue stats).
 * Requires a real queue backend — throws in env-disabled sync mode. */
export async function queueSize(topic: string): Promise<number> {
  const stats = await (bossForTesting ?? boss).getQueueStats(topic)
  return stats[0]?.queuedCount ?? 0
}

/**
 * Drain and stop: wave.close() drains workers under their closeTimeout and detaches the
 * adapter (the caller-owned boss survives it), then the boss stops — gracefully in
 * production, as v1 did — the whole teardown raced against a hard 25s bound.
 */
export async function stopQueue(onStop?: () => Promise<void>): Promise<void> {
  if (!instance) return
  if (stopping) return stopping

  const { isProduction } = getConfig()
  const current = instance

  stopping = (async () => {
    try {
      await Promise.race([
        (async () => {
          await current.close()
          if (boss) {
            await boss.stop({ timeout: 20_000, graceful: isProduction, close: true })
          }
          await onStop?.()
        })(),
        new Promise((_, reject) =>
          setTimeout(() => reject(new Error('Queue stop timeout')), queueStopTimeoutMs)
        ),
      ])
    } finally {
      instance = undefined
      stopping = undefined
    }
  })()

  return stopping
}

1	import { ERRORS } from '@internal/errors'
2	import { logger, logSchema } from '@internal/monitoring'
3	import type { PgBossCtx } from '@supabase-labs/wave-adapter-pgboss'
4	import { pgboss } from '@supabase-labs/wave-adapter-pgboss'
5	import type { AnyWave, HandlerFor, TopicMap, TopicRegistry, Wave } from '@supabase-labs/wave-core'
6	import { createSyncWave, createWave } from '@supabase-labs/wave-core'
7	import type { PgBoss } from 'pg-boss'
8	import { getConfig } from '../../config'
9	import { createQueueBoss, queueDefaults } from './boss'
10	import { schedulingMetrics, syncFallback, syncModeGuard, tenantDisableEvents } from './middleware'
11
12	const { pgQueueEnable, pgQueueEnableWorkers, pgQueueSchemaV2 } = getConfig()	39✔
13
14	/** A storage wave: the app's typed produce/invoke surface over the pgboss adapter. */
15	export type StorageWave<M extends TopicMap> = Wave<M, PgBossCtx>
16
17	/**
18	* What the events layer supplies at startup — the queue layer knows no concrete events beyond
19	* the `meta` each topic declares alongside its pg-boss extension (both merged onto core's
20	* `CreateTopicOptions` — see `storage/events/topics.ts`); the middlewares below read it
21	* straight off `setup.topics`, so there is no separate lookup to wire.
22	*/
23	export interface QueueOptions<M extends TopicMap> {
24	topics: TopicRegistry<M>
25	handlers: ReadonlyArray<HandlerFor<M>>
26	/** Optional teardown for handler-owned resources (v1's per-event `onClose`). */
27	onStop?: () => Promise<void>
28	}
29
30	const queueStopTimeoutMs = 25_000	39✔
31
32	/** The held singleton is topic-erased (module state can't be generic): `AnyWave` is wave's
33	* erased-holder type — every concrete `Wave<...>` assigns here cast-free (ADR-0062), and the
34	* typed surface is restored by the accessors below (`startQueue`'s return, `getWave<M>`). */
35	let instance: AnyWave \| undefined
36	let stopping: Promise<void> \| undefined
37
38	const boss = createQueueBoss({ enableWorkers: pgQueueEnableWorkers ?? true })	39!
39
40	/**
41	* The queue layer is a singleton, so the app's events layer can call `queue()` from anywhere and
42	* get the same instance. The first call must provide the topics and handlers; subsequent calls
43	* may omit them (or provide the same values).
44	* @param opts
45	*/
46	function createSyncWaveInstance<M extends TopicMap>(opts: QueueOptions<M>) {
NEW 47	return createSyncWave({	×
48	topics: opts.topics,
49	handlers: opts.handlers,
50	middleware: [tenantDisableEvents(), syncModeGuard()],
51	})
52	}
53
54	function createWaveInstance<M extends TopicMap>(opts: QueueOptions<M>): Wave<TopicRegistry<M>> {
NEW 55	const workersEnabled = pgQueueEnableWorkers ?? true	×
NEW 56	const { topics, handlers } = opts	×
57
NEW 58	if (!pgQueueEnable) {	×
NEW 59	return createSyncWaveInstance(opts)	×
60	}
61
NEW 62	const pgBossAdapter = pgboss({	×
63	boss,
64	schema: pgQueueSchemaV2,
65	queue: queueDefaults(),
66	})
67
NEW 68	return createWave(pgBossAdapter, {	×
69	topics,
70	handlers: workersEnabled ? handlers : [],	×
71	middleware: [tenantDisableEvents(), schedulingMetrics(), syncFallback()],
72	pollIdleIntervalMs: 5_000,
73	closeTimeout: 20_000,
74	})
75	}
76
77	/**
78	* Start the queue, idempotently. Three shapes, all behind the same `Wave` interface:
79	*
80	* - `PG_QUEUE_ENABLE=false` → a sync wave IS the app's wave: every produce runs its handler
81	* inline (nested sends included), no queue connection exists at all (v1 parity).
82	* - queue enabled, workers enabled → full wave: producers append, workers consume, and a
83	* handler-attached sync wave backs the produce-failure fallback.
84	* - queue enabled, workers disabled (producer-only API nodes) → same, minus consuming: the
85	* wave gets no handlers, while the sync wave still attaches them so the fallback can run.
86	*/
87	export async function startQueue<M extends TopicMap>(
88	queueOpts: QueueOptions<M>,
89	opts: { signal?: AbortSignal } = {}	×
90	): Promise<StorageWave<M>> {
NEW 91	const { topics, handlers } = queueOpts	×
92
NEW 93	if (instance) {	×
NEW 94	return instance as StorageWave<M>	×
95	}
NEW 96	if (opts.signal?.aborted) {	×
NEW 97	throw ERRORS.Aborted('Cannot start queue with aborted signal')	×
98	}
99
NEW 100	instance = createWaveInstance({	×
101	topics,
102	handlers,
103	onStop: queueOpts.onStop,
104	})
105
NEW 106	await instance.start()	×
107
NEW 108	if (opts.signal) {	×
NEW 109	opts.signal.addEventListener(	×
110	'abort',
111	() => {
NEW 112	logSchema.info(logger, '[Queue] Stopping', { type: 'queue' })	×
NEW 113	stopQueue(queueOpts.onStop)	×
114	.then(() => {
NEW 115	logSchema.info(logger, '[Queue] Exited', { type: 'queue' })	×
116	})
117	.catch((e) => {
NEW 118	logSchema.error(logger, '[Queue] Error while stopping queue', {	×
119	error: e,
120	type: 'queue',
121	})
122	})
123	},
124	{ once: true }
125	)
126	}
127
NEW 128	return instance as StorageWave<M>	×
129	}
130
131	/** The running wave. Producers reach it through the typed accessor in storage/events (or,
132	* from inside the topic registry's own module graph, this generic accessor with a type-only
133	* import of the registry's topic map — avoiding the runtime import cycle a value import would
134	* create). */
135	export function getWave<M extends TopicMap>(): StorageWave<M> {
NEW 136	if (!instance) {	×
NEW 137	throw new Error('queue is not started (call startQueue first)')	×
138	}
NEW 139	return instance as StorageWave<M>	×
140	}
141
142	export function isQueueStarted(): boolean {
143	return instance !== undefined	3✔
144	}
145
146	let bossForTesting: Pick<PgBoss, 'getQueueStats'> \| undefined
147
148	/** TEST SEAM: install a stub wave as the running instance, and optionally a stats backend
149	* consulted by `queueSize`. Test stubs are partial by nature, so the widening cast lives
150	* HERE, once — callers pass plain objects with just the members their test touches. */
151	export function setWaveForTesting(
152	wave: Partial<AnyWave>,
153	boss?: Pick<PgBoss, 'getQueueStats'>
154	): void {
155	instance = wave as AnyWave	2✔
156	bossForTesting = boss	2✔
157	}
158
159	/** TEST SEAM: build and install the sync wave — the exact shape `startQueue` uses when the
160	* queue is env-disabled — regardless of `pgQueueEnable`, so a test file that flips the flag on
161	* (to exercise queue-enabled app branches) still never constructs a real pg-boss. */
162	export async function startSyncWaveForTesting<M extends TopicMap>(
163	opts: QueueOptions<M>
164	): Promise<StorageWave<M>> {
NEW 165	const wave = createSyncWaveInstance(opts)	×
NEW 166	await wave.start()	×
NEW 167	instance = wave	×
NEW 168	return instance as StorageWave<M>	×
169	}
170
171	/** Pending jobs on a topic's bare queue (v1 `Queue.getQueueSize`, via pg-boss v12 queue stats).
172	* Requires a real queue backend — throws in env-disabled sync mode. */
173	export async function queueSize(topic: string): Promise<number> {
NEW 174	const stats = await (bossForTesting ?? boss).getQueueStats(topic)	×
NEW 175	return stats[0]?.queuedCount ?? 0	×
176	}
177
178	/**
179	* Drain and stop: wave.close() drains workers under their closeTimeout and detaches the
180	* adapter (the caller-owned boss survives it), then the boss stops — gracefully in
181	* production, as v1 did — the whole teardown raced against a hard 25s bound.
182	*/
183	export async function stopQueue(onStop?: () => Promise<void>): Promise<void> {
184	if (!instance) return	4✔
185	if (stopping) return stopping	3✔
186
187	const { isProduction } = getConfig()	2✔
188	const current = instance	2✔
189
190	stopping = (async () => {	2✔
191	try {	2✔
192	await Promise.race([	2✔
193	(async () => {
194	await current.close()	2✔
NEW 195	if (boss) {	×
NEW 196	await boss.stop({ timeout: 20_000, graceful: isProduction, close: true })	×
197	}
NEW 198	await onStop?.()	×
199	})(),
200	new Promise((_, reject) =>
201	setTimeout(() => reject(new Error('Queue stop timeout')), queueStopTimeoutMs)	2✔
202	),
203	])
204	} finally {
205	instance = undefined	2✔
206	stopping = undefined	2✔
207	}
208	})()
209
210	return stopping	2✔
211	}

supabase / storage / 30832958929

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