30826909455

Committed 03 Aug 2026 03:18PM UTC coverage: 53.88% (-26.5%) from 80.366%

Build # 30826909455

Build Type

Pull #1292

github

Committed by

web-flow

Commit Message

Merge 478fb900a into 1e33d8708

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

Coverage Stats

3689 of 7393 branches covered (49.9%)

Branch coverage included in aggregate %.

140 of 332 new or added lines in 38 files covered. (42.17%)

3352 existing lines in 149 files now uncovered.

6920 of 12297 relevant lines covered (56.27%)

76.57 hits per line

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

31.51

/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 createWaveInstance<M extends TopicMap>(opts: QueueOptions<M>): Wave<TopicRegistry<M>> {
  const workersEnabled = pgQueueEnableWorkers ?? true
  const { topics, handlers } = opts

  if (!pgQueueEnable) {
    return createSyncWave({
      topics,
      handlers,
      middleware: [tenantDisableEvents(), syncModeGuard()],
    })
  }

  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
}

/** 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()	36✔
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	36✔
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 })	36!
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 createWaveInstance<M extends TopicMap>(opts: QueueOptions<M>): Wave<TopicRegistry<M>> {
NEW 47	const workersEnabled = pgQueueEnableWorkers ?? true	×
NEW 48	const { topics, handlers } = opts	×
49
NEW 50	if (!pgQueueEnable) {	×
NEW 51	return createSyncWave({	×
52	topics,
53	handlers,
54	middleware: [tenantDisableEvents(), syncModeGuard()],
55	})
56	}
57
NEW 58	const pgBossAdapter = pgboss({	×
59	boss,
60	schema: pgQueueSchemaV2,
61	queue: queueDefaults(),
62	})
63
NEW 64	return createWave(pgBossAdapter, {	×
65	topics,
66	handlers: workersEnabled ? handlers : [],	×
67	middleware: [tenantDisableEvents(), schedulingMetrics(), syncFallback()],
68	pollIdleIntervalMs: 5_000,
69	closeTimeout: 20_000,
70	})
71	}
72
73	/**
74	* Start the queue, idempotently. Three shapes, all behind the same `Wave` interface:
75	*
76	* - `PG_QUEUE_ENABLE=false` → a sync wave IS the app's wave: every produce runs its handler
77	* inline (nested sends included), no queue connection exists at all (v1 parity).
78	* - queue enabled, workers enabled → full wave: producers append, workers consume, and a
79	* handler-attached sync wave backs the produce-failure fallback.
80	* - queue enabled, workers disabled (producer-only API nodes) → same, minus consuming: the
81	* wave gets no handlers, while the sync wave still attaches them so the fallback can run.
82	*/
83	export async function startQueue<M extends TopicMap>(
84	queueOpts: QueueOptions<M>,
85	opts: { signal?: AbortSignal } = {}	×
86	): Promise<StorageWave<M>> {
NEW 87	const { topics, handlers } = queueOpts	×
88
NEW 89	if (instance) {	×
NEW 90	return instance as StorageWave<M>	×
91	}
NEW 92	if (opts.signal?.aborted) {	×
NEW 93	throw ERRORS.Aborted('Cannot start queue with aborted signal')	×
94	}
95
NEW 96	instance = createWaveInstance({	×
97	topics,
98	handlers,
99	onStop: queueOpts.onStop,
100	})
101
NEW 102	await instance.start()	×
103
NEW 104	if (opts.signal) {	×
NEW 105	opts.signal.addEventListener(	×
106	'abort',
107	() => {
NEW 108	logSchema.info(logger, '[Queue] Stopping', { type: 'queue' })	×
NEW 109	stopQueue(queueOpts.onStop)	×
110	.then(() => {
NEW 111	logSchema.info(logger, '[Queue] Exited', { type: 'queue' })	×
112	})
113	.catch((e) => {
NEW 114	logSchema.error(logger, '[Queue] Error while stopping queue', {	×
115	error: e,
116	type: 'queue',
117	})
118	})
119	},
120	{ once: true }
121	)
122	}
123
NEW 124	return instance as StorageWave<M>	×
125	}
126
127	/** The running wave. Producers reach it through the typed accessor in storage/events (or,
128	* from inside the topic registry's own module graph, this generic accessor with a type-only
129	* import of the registry's topic map — avoiding the runtime import cycle a value import would
130	* create). */
131	export function getWave<M extends TopicMap>(): StorageWave<M> {
NEW 132	if (!instance) {	×
NEW 133	throw new Error('queue is not started (call startQueue first)')	×
134	}
NEW 135	return instance as StorageWave<M>	×
136	}
137
138	export function isQueueStarted(): boolean {
139	return instance !== undefined	3✔
140	}
141
142	let bossForTesting: Pick<PgBoss, 'getQueueStats'> \| undefined
143
144	/** TEST SEAM: install a stub wave as the running instance, and optionally a stats backend
145	* consulted by `queueSize`. Test stubs are partial by nature, so the widening cast lives
146	* HERE, once — callers pass plain objects with just the members their test touches. */
147	export function setWaveForTesting(
148	wave: Partial<AnyWave>,
149	boss?: Pick<PgBoss, 'getQueueStats'>
150	): void {
151	instance = wave as AnyWave	2✔
152	bossForTesting = boss	2✔
153	}
154
155	/** Pending jobs on a topic's bare queue (v1 `Queue.getQueueSize`, via pg-boss v12 queue stats).
156	* Requires a real queue backend — throws in env-disabled sync mode. */
157	export async function queueSize(topic: string): Promise<number> {
NEW 158	const stats = await (bossForTesting ?? boss).getQueueStats(topic)	×
NEW 159	return stats[0]?.queuedCount ?? 0	×
160	}
161
162	/**
163	* Drain and stop: wave.close() drains workers under their closeTimeout and detaches the
164	* adapter (the caller-owned boss survives it), then the boss stops — gracefully in
165	* production, as v1 did — the whole teardown raced against a hard 25s bound.
166	*/
167	export async function stopQueue(onStop?: () => Promise<void>): Promise<void> {
168	if (!instance) return	4✔
169	if (stopping) return stopping	3✔
170
171	const { isProduction } = getConfig()	2✔
172	const current = instance	2✔
173
174	stopping = (async () => {	2✔
175	try {	2✔
176	await Promise.race([	2✔
177	(async () => {
178	await current.close()	2✔
NEW 179	if (boss) {	×
NEW 180	await boss.stop({ timeout: 20_000, graceful: isProduction, close: true })	×
181	}
NEW 182	await onStop?.()	×
183	})(),
184	new Promise((_, reject) =>
185	setTimeout(() => reject(new Error('Queue stop timeout')), queueStopTimeoutMs)	2✔
186	),
187	])
188	} finally {
189	instance = undefined	2✔
190	stopping = undefined	2✔
191	}
192	})()
193
194	return stopping	2✔
195	}

supabase / storage / 30826909455

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