30089661594

Committed 24 Jul 2026 11:28AM UTC coverage: 80.756% (+0.4%) from 80.352%

Build # 30089661594

Build Type

Pull #1043

github

Committed by

web-flow

Commit Message

Merge 583fb7624 into e7b975a46

Pull Request Pull Request #1043: feat: add queue management handlers

Coverage Stats

5829 of 7763 branches covered (75.09%)

Branch coverage included in aggregate %.

270 of 296 new or added lines in 6 files covered. (91.22%)

11066 of 13158 relevant lines covered (84.1%)

409.56 hits per line

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

90.53

/src/internal/queue/overflow.ts

import { ErrorCode, StorageBackendError } from '@internal/errors'
import type { QueryResultRow } from 'pg'
import type {
  DatabaseExecutor,
  DatabaseQueryOptions,
  DatabaseTransaction,
  TransactionOptions,
} from '../database/connection'
import { quoteQualifiedIdentifier } from '../database/sql'
import { PG_BOSS_SCHEMA } from './constants'

const CREATED_STATE = 'created'
const QUEUE_OVERFLOW_ADVISORY_LOCK_KEY = '-5525285245963000612'
export const JOB_OVERFLOW_LIST_LIMIT_DEFAULT = 50
export const JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT = 10000
export const QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE =
  'Backup requires at least one queue, event type, or tenant filter unless confirmAll is true'

export type QueueOverflowSource = 'job' | 'backup'
export type QueueOverflowGroupBy = 'summary' | 'tenant'

export interface QueueOverflowFilters {
  eventTypes?: readonly string[]
  name?: string
  tenantRefs?: readonly string[]
}

interface QueueOverflowOperationOptions {
  signal?: AbortSignal
}

export interface ListQueueOverflowOptions
  extends QueueOverflowFilters,
    QueueOverflowOperationOptions {
  groupBy?: QueueOverflowGroupBy
  limit?: number
  source?: QueueOverflowSource
}

export interface MoveQueueOverflowOptions
  extends QueueOverflowFilters,
    QueueOverflowOperationOptions {
  limit?: number
}

export interface BackupQueueOverflowOptions extends MoveQueueOverflowOptions {
  confirmAll?: boolean
}

interface QueueOverflowWhereClause {
  sql: string
  values: unknown[]
}

interface QueueOverflowAggregateRow extends QueryResultRow {
  count: number | string
  group_count: number | string
  total_count: number | string
}

interface QueueOverflowTotalCountRow extends QueryResultRow {
  total_count: number | string
}

interface QueueOverflowSummaryDbRow extends QueueOverflowAggregateRow {
  event_type: string | null
  name: string
}

interface QueueOverflowTenantDbRow extends QueueOverflowAggregateRow {
  tenant_ref: string | null
}

interface QueueOverflowCountRow extends QueryResultRow {
  moved_count: number | string
}

interface QueueOverflowRestoreCountRow extends QueueOverflowCountRow {
  selected_count: number | string
}

interface QueueOverflowEngineRow extends QueryResultRow {
  is_oriole: boolean
}

interface QueueOverflowTableRow extends QueryResultRow {
  table_name: string | null
}

interface QueueOverflowIndexRow extends QueryResultRow {
  index_name: string
}

export interface QueueOverflowDatabase {
  beginTransaction(
    options?: TransactionOptions & DatabaseQueryOptions
  ): Promise<DatabaseTransaction>
}

function normalizeOverflowString(value: string | undefined) {
  const normalized = value?.trim()
  return normalized ? normalized : undefined
}

function normalizeStringList(values: readonly string[] | undefined): string[] | undefined {
  if (!values) {
    return undefined
  }

  const normalized = [...new Set(values.map((value) => value.trim()).filter(Boolean))]
  return normalized.length ? normalized : undefined
}

export function parseCommaSeparatedList(value: string | undefined) {
  return value ? normalizeStringList(value.split(',')) : undefined
}

export function normalizeQueueOverflowFilters(filters: QueueOverflowFilters): QueueOverflowFilters {
  return {
    name: normalizeOverflowString(filters.name),
    eventTypes: normalizeStringList(filters.eventTypes),
    tenantRefs: normalizeStringList(filters.tenantRefs),
  }
}

export function buildQueueOverflowWhereClause(
  filters: QueueOverflowFilters
): QueueOverflowWhereClause {
  const normalizedFilters = normalizeQueueOverflowFilters(filters)
  const clauses = ['state = $1']
  const values: unknown[] = [CREATED_STATE]

  if (normalizedFilters.name) {
    values.push(normalizedFilters.name)
    clauses.push(`name = $${values.length}`)
  }

  if (normalizedFilters.eventTypes?.length) {
    values.push(normalizedFilters.eventTypes)
    clauses.push(`data->'event'->>'type' = ANY($${values.length}::text[])`)
  }

  if (normalizedFilters.tenantRefs?.length) {
    values.push(normalizedFilters.tenantRefs)
    clauses.push(`data->'tenant'->>'ref' = ANY($${values.length}::text[])`)
  }

  return {
    sql: clauses.join(' AND '),
    values,
  }
}

export class QueueOverflowStorePg {
  private readonly backupTableName: string
  private readonly jobTable: string
  private readonly backupTable: string

  constructor(
    private readonly db: QueueOverflowDatabase,
    schema = PG_BOSS_SCHEMA
  ) {
    const jobTableName = `${schema}.job`
    this.backupTableName = `${schema}.job_overflow_backup`
    this.jobTable = quoteQualifiedIdentifier(jobTableName)
    this.backupTable = quoteQualifiedIdentifier(this.backupTableName)
  }

  async countCreated(options: QueueOverflowOperationOptions = {}) {
    return this.withMaintenanceTransaction(options.signal, async (transaction) => {
      const result = await transaction.query<QueueOverflowTotalCountRow>(
        {
          text: `
            SELECT COUNT(*)::bigint AS total_count
            FROM ${this.jobTable}
            WHERE state = $1
          `,
          values: [CREATED_STATE],
        },
        { signal: options.signal }
      )

      return {
        totalCount: Number(result.rows[0]?.total_count ?? 0),
      }
    })
  }

  async list(options: ListQueueOverflowOptions) {
    const source = options.source ?? 'job'
    const groupBy = options.groupBy ?? 'summary'
    const limit = options.limit ?? JOB_OVERFLOW_LIST_LIMIT_DEFAULT
    const filters = normalizeQueueOverflowFilters(options)

    assertPositiveSafeInteger(limit, 'limit')

    return this.withMaintenanceTransaction(options.signal, async (transaction) => {
      const sourceTableExists =
        source === 'backup' ? await this.backupTableExists(transaction, options.signal) : true

      if (!sourceTableExists) {
        return {
          sourceTableExists,
          data: [],
          filters,
          groupBy,
          groupCount: 0,
          hasMore: false,
          source,
          totalCount: 0,
        }
      }

      const tableName = source === 'backup' ? this.backupTable : this.jobTable
      const whereClause = buildQueueOverflowWhereClause(filters)
      const values = [...whereClause.values, limit]
      const limitPlaceholder = `$${values.length}`

      if (groupBy === 'tenant') {
        const result = await transaction.query<QueueOverflowTenantDbRow>(
          {
            text: `
              SELECT
                data->'tenant'->>'ref' AS tenant_ref,
                COUNT(*)::bigint AS count,
                COUNT(*) OVER ()::bigint AS group_count,
                SUM(COUNT(*)) OVER ()::bigint AS total_count
              FROM ${tableName}
              WHERE ${whereClause.sql}
              GROUP BY data->'tenant'->>'ref'
              ORDER BY count DESC, tenant_ref ASC
              LIMIT ${limitPlaceholder}
            `,
            values,
          },
          { signal: options.signal }
        )

        const groupCount = Number(result.rows[0]?.group_count ?? 0)

        return {
          sourceTableExists,
          data: result.rows.map((row) => ({
            count: Number(row.count),
            tenantRef: row.tenant_ref,
          })),
          filters,
          groupBy,
          groupCount,
          hasMore: groupCount > result.rows.length,
          source,
          totalCount: Number(result.rows[0]?.total_count ?? 0),
        }
      }

      const result = await transaction.query<QueueOverflowSummaryDbRow>(
        {
          text: `
            SELECT
              name,
              data->'event'->>'type' AS event_type,
              COUNT(*)::bigint AS count,
              COUNT(*) OVER ()::bigint AS group_count,
              SUM(COUNT(*)) OVER ()::bigint AS total_count
            FROM ${tableName}
            WHERE ${whereClause.sql}
            GROUP BY name, data->'event'->>'type'
            ORDER BY count DESC, name ASC, event_type ASC
            LIMIT ${limitPlaceholder}
          `,
          values,
        },
        { signal: options.signal }
      )

      const groupCount = Number(result.rows[0]?.group_count ?? 0)

      return {
        sourceTableExists,
        data: result.rows.map((row) => ({
          count: Number(row.count),
          eventType: row.event_type,
          name: row.name,
        })),
        filters,
        groupBy,
        groupCount,
        hasMore: groupCount > result.rows.length,
        source,
        totalCount: Number(result.rows[0]?.total_count ?? 0),
      }
    })
  }

  async backup(options: BackupQueueOverflowOptions) {
    const filters = normalizeQueueOverflowFilters(options)
    if (
      options.confirmAll !== true &&
      !filters.name &&
      !filters.eventTypes?.length &&
      !filters.tenantRefs?.length
    ) {
      throw new Error(QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE)
    }

    if (options.limit !== undefined) {
      assertPositiveSafeInteger(options.limit, 'limit')
    }

    return this.withMaintenanceTransaction(options.signal, async (transaction) => {
      await this.acquireMaintenanceLock(transaction, options.signal)
      const backupTableCreated = await this.ensureBackupTable(transaction, options.signal)

      await transaction.query(`LOCK TABLE ${this.jobTable} IN SHARE ROW EXCLUSIVE MODE`, {
        signal: options.signal,
      })

      const movedCount = await this.moveJobs(
        transaction,
        filters,
        this.jobTable,
        this.backupTable,
        options.limit,
        options.signal
      )

      return {
        backupTableCreated,
        filters,
        limit: options.limit ?? null,
        movedCount,
      }
    })
  }

  async restore(options: MoveQueueOverflowOptions) {
    const limit = options.limit ?? JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT
    assertPositiveSafeInteger(limit, 'limit')
    const filters = normalizeQueueOverflowFilters(options)

    return this.withMaintenanceTransaction(options.signal, async (transaction) => {
      await this.acquireMaintenanceLock(transaction, options.signal)
      const backupTableExists = await this.backupTableExists(transaction, options.signal)

      if (!backupTableExists) {
        return {
          backupTableExists,
          conflictCount: 0,
          filters,
          hasMore: false,
          limit,
          movedCount: 0,
        }
      }

      const engine = await transaction.query<QueueOverflowEngineRow>(
        `SELECT EXISTS (
          SELECT 1 FROM pg_extension WHERE extname = 'orioledb'
        ) AS is_oriole`,
        { signal: options.signal }
      )

      if (engine.rows[0]?.is_oriole) {
        throw new StorageBackendError({
          code: ErrorCode.NotSupported,
          httpStatusCode: 409,
          message: 'Queue overflow restore is not supported on OrioleDB',
        })
      }

      await transaction.query(`LOCK TABLE ${this.jobTable} IN SHARE ROW EXCLUSIVE MODE`, {
        signal: options.signal,
      })

      const { conflictCount, hasMore, movedCount } = await this.restoreJobs(
        transaction,
        filters,
        limit,
        options.signal
      )

      return {
        backupTableExists,
        conflictCount,
        filters,
        hasMore,
        limit,
        movedCount,
      }
    })
  }

  private async backupTableExists(db: DatabaseExecutor, signal?: AbortSignal) {
    const result = await db.query<QueueOverflowTableRow>(
      {
        text: 'SELECT to_regclass($1) AS table_name',
        values: [this.backupTableName],
      },
      { signal }
    )

    return Boolean(result.rows[0]?.table_name)
  }

  private async ensureBackupTable(db: DatabaseExecutor, signal?: AbortSignal) {
    const existed = await this.backupTableExists(db, signal)

    await db.query(
      `CREATE TABLE IF NOT EXISTS ${this.backupTable} (
        LIKE ${this.jobTable} INCLUDING DEFAULTS,
        PRIMARY KEY (name, id)
      )`,
      {
        signal,
      }
    )

    const legacyIndexes = await db.query<QueueOverflowIndexRow>(
      {
        text: `
          SELECT format('%I.%I', index_namespace.nspname, index_class.relname) AS index_name
          FROM pg_index AS backup_index
          JOIN pg_class AS index_class ON index_class.oid = backup_index.indexrelid
          JOIN pg_namespace AS index_namespace ON index_namespace.oid = index_class.relnamespace
          WHERE backup_index.indrelid = to_regclass($1)
            AND backup_index.indisunique
            AND backup_index.indpred IS NOT NULL
        `,
        values: [this.backupTableName],
      },
      { signal }
    )

    for (const index of legacyIndexes.rows) {
      await db.query(`DROP INDEX IF EXISTS ${index.index_name}`, { signal })
    }

    return !existed
  }

  private async moveJobs(
    db: DatabaseExecutor,
    filters: QueueOverflowFilters,
    sourceTable: string,
    targetTable: string,
    limit?: number,
    signal?: AbortSignal
  ) {
    const whereClause = buildQueueOverflowWhereClause(filters)
    const values = [...whereClause.values]
    let selectionBoundary = ''

    if (limit !== undefined) {
      values.push(limit)
      selectionBoundary = `ORDER BY name, id LIMIT $${values.length}`
    }

    const result = await db.query<QueueOverflowCountRow>(
      {
        text: `
          WITH selected AS (
            SELECT name, id
            FROM ${sourceTable}
            WHERE ${whereClause.sql}
            ${selectionBoundary}
          ),
          moved AS (
            DELETE FROM ${sourceTable} AS source_job
            USING selected
            WHERE source_job.name = selected.name
              AND source_job.id = selected.id
            RETURNING source_job.*
          ),
          inserted AS (
            INSERT INTO ${targetTable}
            SELECT * FROM moved
            RETURNING 1
          )
          SELECT COUNT(*)::bigint AS moved_count FROM inserted
        `,
        values,
      },
      { signal }
    )

    return Number(result.rows[0]?.moved_count ?? 0)
  }

  private async restoreJobs(
    db: DatabaseExecutor,
    filters: QueueOverflowFilters,
    limit: number,
    signal?: AbortSignal
  ) {
    const whereClause = buildQueueOverflowWhereClause(filters)
    const values = [...whereClause.values]
    values.push(limit)
    const limitPlaceholder = `$${values.length}`
    const result = await db.query<QueueOverflowRestoreCountRow>(
      {
        text: `
          WITH selected AS (
            SELECT *
            FROM ${this.backupTable}
            WHERE ${whereClause.sql}
            ORDER BY name, id
            LIMIT ${limitPlaceholder}
          ),
          inserted AS (
            INSERT INTO ${this.jobTable}
            SELECT * FROM selected
            ORDER BY name, id
            ON CONFLICT DO NOTHING
            RETURNING 1
          ),
          deleted AS (
            DELETE FROM ${this.backupTable} AS source_job
            USING selected
            WHERE source_job.name = selected.name
              AND source_job.id = selected.id
            RETURNING 1
          )
          SELECT
            (SELECT COUNT(*) FROM selected) AS selected_count,
            (SELECT COUNT(*) FROM inserted) AS moved_count
        `,
        values,
      },
      { signal }
    )

    const selectedCount = Number(result.rows[0]?.selected_count ?? 0)
    const movedCount = Number(result.rows[0]?.moved_count ?? 0)

    return {
      conflictCount: selectedCount - movedCount,
      hasMore: selectedCount === limit,
      movedCount,
    }
  }

  private async acquireMaintenanceLock(db: DatabaseExecutor, signal?: AbortSignal): Promise<void> {
    await db.query(
      {
        text: 'SELECT pg_advisory_xact_lock($1::bigint)',
        values: [QUEUE_OVERFLOW_ADVISORY_LOCK_KEY],
      },
      { signal }
    )
  }

  private async withMaintenanceTransaction<T>(
    signal: AbortSignal | undefined,
    fn: (transaction: DatabaseTransaction) => Promise<T>
  ): Promise<T> {
    const transaction = await this.db.beginTransaction({ signal })

    try {
      await transaction.query(
        {
          text: `
            SELECT
              set_config('statement_timeout', $1, true),
              set_config('lock_timeout', $2, true)
          `,
          values: ['0', '30s'],
        },
        { signal }
      )

      const result = await fn(transaction)
      await transaction.commit()
      return result
    } catch (error) {
      try {
        await transaction.rollback()
      } catch (rollbackError) {
        await logQueueOverflowRollbackFailure(error, rollbackError)
      }

      throw error
    }
  }
}

async function logQueueOverflowRollbackFailure(
  originalError: unknown,
  rollbackError: unknown
): Promise<void> {
  try {
    const { logger, logSchema } = await import('@internal/monitoring')
    logSchema.warning(logger, '[QueueOverflow] Failed to rollback maintenance transaction', {
      type: 'pgboss',
      error: rollbackError,
      metadata: JSON.stringify({ originalError: String(originalError) }),
    })
  } catch (loggingError) {
    console.error('[QueueOverflow] Failed to log maintenance transaction rollback failure', {
      originalError,
      rollbackError,
      loggingError,
    })
  }
}

function assertPositiveSafeInteger(value: number, field: string): void {
  if (!Number.isSafeInteger(value) || value <= 0) {
    throw new Error(`${field} must be a positive safe integer`)
  }
}

1	import { ErrorCode, StorageBackendError } from '@internal/errors'
2	import type { QueryResultRow } from 'pg'
3	import type {
4	DatabaseExecutor,
5	DatabaseQueryOptions,
6	DatabaseTransaction,
7	TransactionOptions,
8	} from '../database/connection'
9	import { quoteQualifiedIdentifier } from '../database/sql'
10	import { PG_BOSS_SCHEMA } from './constants'
11
12	const CREATED_STATE = 'created'	88✔
13	const QUEUE_OVERFLOW_ADVISORY_LOCK_KEY = '-5525285245963000612'	88✔
14	export const JOB_OVERFLOW_LIST_LIMIT_DEFAULT = 50	88✔
15	export const JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT = 10000	88✔
16	export const QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE =
17	'Backup requires at least one queue, event type, or tenant filter unless confirmAll is true'	88✔
18
19	export type QueueOverflowSource = 'job' \| 'backup'
20	export type QueueOverflowGroupBy = 'summary' \| 'tenant'
21
22	export interface QueueOverflowFilters {
23	eventTypes?: readonly string[]
24	name?: string
25	tenantRefs?: readonly string[]
26	}
27
28	interface QueueOverflowOperationOptions {
29	signal?: AbortSignal
30	}
31
32	export interface ListQueueOverflowOptions
33	extends QueueOverflowFilters,
34	QueueOverflowOperationOptions {
35	groupBy?: QueueOverflowGroupBy
36	limit?: number
37	source?: QueueOverflowSource
38	}
39
40	export interface MoveQueueOverflowOptions
41	extends QueueOverflowFilters,
42	QueueOverflowOperationOptions {
43	limit?: number
44	}
45
46	export interface BackupQueueOverflowOptions extends MoveQueueOverflowOptions {
47	confirmAll?: boolean
48	}
49
50	interface QueueOverflowWhereClause {
51	sql: string
52	values: unknown[]
53	}
54
55	interface QueueOverflowAggregateRow extends QueryResultRow {
56	count: number \| string
57	group_count: number \| string
58	total_count: number \| string
59	}
60
61	interface QueueOverflowTotalCountRow extends QueryResultRow {
62	total_count: number \| string
63	}
64
65	interface QueueOverflowSummaryDbRow extends QueueOverflowAggregateRow {
66	event_type: string \| null
67	name: string
68	}
69
70	interface QueueOverflowTenantDbRow extends QueueOverflowAggregateRow {
71	tenant_ref: string \| null
72	}
73
74	interface QueueOverflowCountRow extends QueryResultRow {
75	moved_count: number \| string
76	}
77
78	interface QueueOverflowRestoreCountRow extends QueueOverflowCountRow {
79	selected_count: number \| string
80	}
81
82	interface QueueOverflowEngineRow extends QueryResultRow {
83	is_oriole: boolean
84	}
85
86	interface QueueOverflowTableRow extends QueryResultRow {
87	table_name: string \| null
88	}
89
90	interface QueueOverflowIndexRow extends QueryResultRow {
91	index_name: string
92	}
93
94	export interface QueueOverflowDatabase {
95	beginTransaction(
96	options?: TransactionOptions & DatabaseQueryOptions
97	): Promise<DatabaseTransaction>
98	}
99
100	function normalizeOverflowString(value: string \| undefined) {
101	const normalized = value?.trim()	56✔
102	return normalized ? normalized : undefined	56✔
103	}
104
105	function normalizeStringList(values: readonly string[] \| undefined): string[] \| undefined {
106	if (!values) {	115✔
107	return undefined	88✔
108	}
109
110	const normalized = [...new Set(values.map((value) => value.trim()).filter(Boolean))]	44✔
111	return normalized.length ? normalized : undefined	27!
112	}
113
114	export function parseCommaSeparatedList(value: string \| undefined) {
115	return value ? normalizeStringList(value.split(',')) : undefined	50✔
116	}
117
118	export function normalizeQueueOverflowFilters(filters: QueueOverflowFilters): QueueOverflowFilters {
119	return {	56✔
120	name: normalizeOverflowString(filters.name),
121	eventTypes: normalizeStringList(filters.eventTypes),
122	tenantRefs: normalizeStringList(filters.tenantRefs),
123	}
124	}
125
126	export function buildQueueOverflowWhereClause(
127	filters: QueueOverflowFilters
128	): QueueOverflowWhereClause {
129	const normalizedFilters = normalizeQueueOverflowFilters(filters)	26✔
130	const clauses = ['state = $1']	26✔
131	const values: unknown[] = [CREATED_STATE]	26✔
132
133	if (normalizedFilters.name) {	26✔
134	values.push(normalizedFilters.name)	21✔
135	clauses.push(`name = $${values.length}`)	21✔
136	}
137
138	if (normalizedFilters.eventTypes?.length) {	26✔
139	values.push(normalizedFilters.eventTypes)	4✔
140	clauses.push(`data->'event'->>'type' = ANY($${values.length}::text[])`)	4✔
141	}
142
143	if (normalizedFilters.tenantRefs?.length) {	26✔
144	values.push(normalizedFilters.tenantRefs)	8✔
145	clauses.push(`data->'tenant'->>'ref' = ANY($${values.length}::text[])`)	8✔
146	}
147
148	return {	26✔
149	sql: clauses.join(' AND '),
150	values,
151	}
152	}
153
154	export class QueueOverflowStorePg {
155	private readonly backupTableName: string
156	private readonly jobTable: string
157	private readonly backupTable: string
158
159	constructor(
160	private readonly db: QueueOverflowDatabase,	28✔
161	schema = PG_BOSS_SCHEMA	28✔
162	) {
163	const jobTableName = `${schema}.job`	28✔
164	this.backupTableName = `${schema}.job_overflow_backup`	28✔
165	this.jobTable = quoteQualifiedIdentifier(jobTableName)	28✔
166	this.backupTable = quoteQualifiedIdentifier(this.backupTableName)	28✔
167	}
168
169	async countCreated(options: QueueOverflowOperationOptions = {}) {	4✔
170	return this.withMaintenanceTransaction(options.signal, async (transaction) => {	4✔
171	const result = await transaction.query<QueueOverflowTotalCountRow>(	4✔
172	{
173	text: `
174	SELECT COUNT(*)::bigint AS total_count
175	FROM ${this.jobTable}
176	WHERE state = $1
177	`,
178	values: [CREATED_STATE],
179	},
180	{ signal: options.signal }
181	)
182
183	return {	4✔
184	totalCount: Number(result.rows[0]?.total_count ?? 0),	4!
185	}
186	})
187	}
188
189	async list(options: ListQueueOverflowOptions) {
190	const source = options.source ?? 'job'	4✔
191	const groupBy = options.groupBy ?? 'summary'	4✔
192	const limit = options.limit ?? JOB_OVERFLOW_LIST_LIMIT_DEFAULT	4!
193	const filters = normalizeQueueOverflowFilters(options)	4✔
194
195	assertPositiveSafeInteger(limit, 'limit')	4✔
196
197	return this.withMaintenanceTransaction(options.signal, async (transaction) => {	4✔
198	const sourceTableExists =
199	source === 'backup' ? await this.backupTableExists(transaction, options.signal) : true	4✔
200
201	if (!sourceTableExists) {	4✔
202	return {	1✔
203	sourceTableExists,
204	data: [],
205	filters,
206	groupBy,
207	groupCount: 0,
208	hasMore: false,
209	source,
210	totalCount: 0,
211	}
212	}
213
214	const tableName = source === 'backup' ? this.backupTable : this.jobTable	3!
215	const whereClause = buildQueueOverflowWhereClause(filters)	4✔
216	const values = [...whereClause.values, limit]	4✔
217	const limitPlaceholder = `$${values.length}`	4✔
218
219	if (groupBy === 'tenant') {	4✔
220	const result = await transaction.query<QueueOverflowTenantDbRow>(	1✔
221	{
222	text: `
223	SELECT
224	data->'tenant'->>'ref' AS tenant_ref,
225	COUNT(*)::bigint AS count,
226	COUNT(*) OVER ()::bigint AS group_count,
227	SUM(COUNT(*)) OVER ()::bigint AS total_count
228	FROM ${tableName}
229	WHERE ${whereClause.sql}
230	GROUP BY data->'tenant'->>'ref'
231	ORDER BY count DESC, tenant_ref ASC
232	LIMIT ${limitPlaceholder}
233	`,
234	values,
235	},
236	{ signal: options.signal }
237	)
238
239	const groupCount = Number(result.rows[0]?.group_count ?? 0)	1!
240
241	return {	1✔
242	sourceTableExists,
243	data: result.rows.map((row) => ({	2✔
244	count: Number(row.count),
245	tenantRef: row.tenant_ref,
246	})),
247	filters,
248	groupBy,
249	groupCount,
250	hasMore: groupCount > result.rows.length,
251	source,
252	totalCount: Number(result.rows[0]?.total_count ?? 0),	1!
253	}
254	}
255
256	const result = await transaction.query<QueueOverflowSummaryDbRow>(	2✔
257	{
258	text: `
259	SELECT
260	name,
261	data->'event'->>'type' AS event_type,
262	COUNT(*)::bigint AS count,
263	COUNT(*) OVER ()::bigint AS group_count,
264	SUM(COUNT(*)) OVER ()::bigint AS total_count
265	FROM ${tableName}
266	WHERE ${whereClause.sql}
267	GROUP BY name, data->'event'->>'type'
268	ORDER BY count DESC, name ASC, event_type ASC
269	LIMIT ${limitPlaceholder}
270	`,
271	values,
272	},
273	{ signal: options.signal }
274	)
275
276	const groupCount = Number(result.rows[0]?.group_count ?? 0)	2!
277
278	return {	4✔
279	sourceTableExists,
280	data: result.rows.map((row) => ({	2✔
281	count: Number(row.count),
282	eventType: row.event_type,
283	name: row.name,
284	})),
285	filters,
286	groupBy,
287	groupCount,
288	hasMore: groupCount > result.rows.length,
289	source,
290	totalCount: Number(result.rows[0]?.total_count ?? 0),	4!
291	}
292	})
293	}
294
295	async backup(options: BackupQueueOverflowOptions) {
296	const filters = normalizeQueueOverflowFilters(options)	13✔
297	if (	13✔
298	options.confirmAll !== true &&	29✔
299	!filters.name &&
300	!filters.eventTypes?.length &&
301	!filters.tenantRefs?.length
302	) {
303	throw new Error(QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE)	2✔
304	}
305
306	if (options.limit !== undefined) {	11✔
307	assertPositiveSafeInteger(options.limit, 'limit')	2✔
308	}
309
310	return this.withMaintenanceTransaction(options.signal, async (transaction) => {	11✔
311	await this.acquireMaintenanceLock(transaction, options.signal)	11✔
312	const backupTableCreated = await this.ensureBackupTable(transaction, options.signal)	11✔
313
314	await transaction.query(`LOCK TABLE ${this.jobTable} IN SHARE ROW EXCLUSIVE MODE`, {	11✔
315	signal: options.signal,
316	})
317
318	const movedCount = await this.moveJobs(	11✔
319	transaction,
320	filters,
321	this.jobTable,
322	this.backupTable,
323	options.limit,
324	options.signal
325	)
326
327	return {	11✔
328	backupTableCreated,
329	filters,
330	limit: options.limit ?? null,	20✔
331	movedCount,
332	}
333	})
334	}
335
336	async restore(options: MoveQueueOverflowOptions) {
337	const limit = options.limit ?? JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT	12✔
338	assertPositiveSafeInteger(limit, 'limit')	12✔
339	const filters = normalizeQueueOverflowFilters(options)	12✔
340
341	return this.withMaintenanceTransaction(options.signal, async (transaction) => {	12✔
342	await this.acquireMaintenanceLock(transaction, options.signal)	12✔
343	const backupTableExists = await this.backupTableExists(transaction, options.signal)	12✔
344
345	if (!backupTableExists) {	12✔
346	return {	1✔
347	backupTableExists,
348	conflictCount: 0,
349	filters,
350	hasMore: false,
351	limit,
352	movedCount: 0,
353	}
354	}
355
356	const engine = await transaction.query<QueueOverflowEngineRow>(	11✔
357	`SELECT EXISTS (
358	SELECT 1 FROM pg_extension WHERE extname = 'orioledb'
359	) AS is_oriole`,
360	{ signal: options.signal }
361	)
362
363	if (engine.rows[0]?.is_oriole) {	11✔
364	throw new StorageBackendError({	1✔
365	code: ErrorCode.NotSupported,
366	httpStatusCode: 409,
367	message: 'Queue overflow restore is not supported on OrioleDB',
368	})
369	}
370
371	await transaction.query(`LOCK TABLE ${this.jobTable} IN SHARE ROW EXCLUSIVE MODE`, {	10✔
372	signal: options.signal,
373	})
374
375	const { conflictCount, hasMore, movedCount } = await this.restoreJobs(	10✔
376	transaction,
377	filters,
378	limit,
379	options.signal
380	)
381
382	return {	8✔
383	backupTableExists,
384	conflictCount,
385	filters,
386	hasMore,
387	limit,
388	movedCount,
389	}
390	})
391	}
392
393	private async backupTableExists(db: DatabaseExecutor, signal?: AbortSignal) {
394	const result = await db.query<QueueOverflowTableRow>(	24✔
395	{
396	text: 'SELECT to_regclass($1) AS table_name',
397	values: [this.backupTableName],
398	},
399	{ signal }
400	)
401
402	return Boolean(result.rows[0]?.table_name)	24✔
403	}
404
405	private async ensureBackupTable(db: DatabaseExecutor, signal?: AbortSignal) {
406	const existed = await this.backupTableExists(db, signal)	11✔
407
408	await db.query(	11✔
409	`CREATE TABLE IF NOT EXISTS ${this.backupTable} (
410	LIKE ${this.jobTable} INCLUDING DEFAULTS,
411	PRIMARY KEY (name, id)
412	)`,
413	{
414	signal,
415	}
416	)
417
418	const legacyIndexes = await db.query<QueueOverflowIndexRow>(	11✔
419	{
420	text: `
421	SELECT format('%I.%I', index_namespace.nspname, index_class.relname) AS index_name
422	FROM pg_index AS backup_index
423	JOIN pg_class AS index_class ON index_class.oid = backup_index.indexrelid
424	JOIN pg_namespace AS index_namespace ON index_namespace.oid = index_class.relnamespace
425	WHERE backup_index.indrelid = to_regclass($1)
426	AND backup_index.indisunique
427	AND backup_index.indpred IS NOT NULL
428	`,
429	values: [this.backupTableName],
430	},
431	{ signal }
432	)
433
434	for (const index of legacyIndexes.rows) {	11✔
435	await db.query(`DROP INDEX IF EXISTS ${index.index_name}`, { signal })	2✔
436	}
437
438	return !existed	11✔
439	}
440
441	private async moveJobs(
442	db: DatabaseExecutor,
443	filters: QueueOverflowFilters,
444	sourceTable: string,
445	targetTable: string,
446	limit?: number,
447	signal?: AbortSignal
448	) {
449	const whereClause = buildQueueOverflowWhereClause(filters)	11✔
450	const values = [...whereClause.values]	11✔
451	let selectionBoundary = ''	11✔
452
453	if (limit !== undefined) {	11✔
454	values.push(limit)	2✔
455	selectionBoundary = `ORDER BY name, id LIMIT $${values.length}`	2✔
456	}
457
458	const result = await db.query<QueueOverflowCountRow>(	11✔
459	{
460	text: `
461	WITH selected AS (
462	SELECT name, id
463	FROM ${sourceTable}
464	WHERE ${whereClause.sql}
465	${selectionBoundary}
466	),
467	moved AS (
468	DELETE FROM ${sourceTable} AS source_job
469	USING selected
470	WHERE source_job.name = selected.name
471	AND source_job.id = selected.id
472	RETURNING source_job.*
473	),
474	inserted AS (
475	INSERT INTO ${targetTable}
476	SELECT * FROM moved
477	RETURNING 1
478	)
479	SELECT COUNT(*)::bigint AS moved_count FROM inserted
480	`,
481	values,
482	},
483	{ signal }
484	)
485
486	return Number(result.rows[0]?.moved_count ?? 0)	11!
487	}
488
489	private async restoreJobs(
490	db: DatabaseExecutor,
491	filters: QueueOverflowFilters,
492	limit: number,
493	signal?: AbortSignal
494	) {
495	const whereClause = buildQueueOverflowWhereClause(filters)	10✔
496	const values = [...whereClause.values]	10✔
497	values.push(limit)	10✔
498	const limitPlaceholder = `$${values.length}`	10✔
499	const result = await db.query<QueueOverflowRestoreCountRow>(	10✔
500	{
501	text: `
502	WITH selected AS (
503	SELECT *
504	FROM ${this.backupTable}
505	WHERE ${whereClause.sql}
506	ORDER BY name, id
507	LIMIT ${limitPlaceholder}
508	),
509	inserted AS (
510	INSERT INTO ${this.jobTable}
511	SELECT * FROM selected
512	ORDER BY name, id
513	ON CONFLICT DO NOTHING
514	RETURNING 1
515	),
516	deleted AS (
517	DELETE FROM ${this.backupTable} AS source_job
518	USING selected
519	WHERE source_job.name = selected.name
520	AND source_job.id = selected.id
521	RETURNING 1
522	)
523	SELECT
524	(SELECT COUNT(*) FROM selected) AS selected_count,
525	(SELECT COUNT(*) FROM inserted) AS moved_count
526	`,
527	values,
528	},
529	{ signal }
530	)
531
532	const selectedCount = Number(result.rows[0]?.selected_count ?? 0)	8!
533	const movedCount = Number(result.rows[0]?.moved_count ?? 0)	10!
534
535	return {	10✔
536	conflictCount: selectedCount - movedCount,
537	hasMore: selectedCount === limit,
538	movedCount,
539	}
540	}
541
542	private async acquireMaintenanceLock(db: DatabaseExecutor, signal?: AbortSignal): Promise<void> {
543	await db.query(	23✔
544	{
545	text: 'SELECT pg_advisory_xact_lock($1::bigint)',
546	values: [QUEUE_OVERFLOW_ADVISORY_LOCK_KEY],
547	},
548	{ signal }
549	)
550	}
551
552	private async withMaintenanceTransaction<T>(
553	signal: AbortSignal \| undefined,
554	fn: (transaction: DatabaseTransaction) => Promise<T>
555	): Promise<T> {
556	const transaction = await this.db.beginTransaction({ signal })	31✔
557
558	try {	31✔
559	await transaction.query(	31✔
560	{
561	text: `
562	SELECT
563	set_config('statement_timeout', $1, true),
564	set_config('lock_timeout', $2, true)
565	`,
566	values: ['0', '30s'],
567	},
568	{ signal }
569	)
570
571	const result = await fn(transaction)	31✔
572	await transaction.commit()	28✔
573	return result	28✔
574	} catch (error) {
575	try {	3✔
576	await transaction.rollback()	3✔
577	} catch (rollbackError) {
NEW 578	await logQueueOverflowRollbackFailure(error, rollbackError)	×
579	}
580
581	throw error	3✔
582	}
583	}
584	}
585
586	async function logQueueOverflowRollbackFailure(
587	originalError: unknown,
588	rollbackError: unknown
589	): Promise<void> {
NEW 590	try {	×
NEW 591	const { logger, logSchema } = await import('@internal/monitoring')	×
NEW 592	logSchema.warning(logger, '[QueueOverflow] Failed to rollback maintenance transaction', {	×
593	type: 'pgboss',
594	error: rollbackError,
595	metadata: JSON.stringify({ originalError: String(originalError) }),
596	})
597	} catch (loggingError) {
NEW 598	console.error('[QueueOverflow] Failed to log maintenance transaction rollback failure', {	×
599	originalError,
600	rollbackError,
601	loggingError,
602	})
603	}
604	}
605
606	function assertPositiveSafeInteger(value: number, field: string): void {
607	if (!Number.isSafeInteger(value) \|\| value <= 0) {	18!
NEW 608	throw new Error(`${field} must be a positive safe integer`)	×
609	}
610	}

supabase / storage / 30089661594

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