From a42d131b07f2a2917dd96466b02f6e40f6d4c13c Mon Sep 17 00:00:00 2001 From: Brendan Kellam Date: Tue, 7 Jul 2026 20:16:00 -0700 Subject: [PATCH] feat(backend): add BullMQ JobManager framework Introduces a JobManager over BullMQ: work-queue Workloads with a ProcessContext, CronWorkloads multiplexed onto a shared cron worker with a reconcile() sweep helper, a JobProducer that owns the queues and the deduplicated enqueue path, and a Redis-backed read model (status/jobDetail). Wired into the backend entrypoint with a demo workload. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/backend/src/index.ts | 48 ++++ packages/backend/src/jobManager.test.ts | 114 +++++++++ packages/backend/src/jobManager.ts | 309 ++++++++++++++++++++++++ packages/backend/src/jobProducer.ts | 32 +++ packages/backend/src/types.ts | 86 ++++++- 5 files changed, 588 insertions(+), 1 deletion(-) create mode 100644 packages/backend/src/jobManager.test.ts create mode 100644 packages/backend/src/jobManager.ts create mode 100644 packages/backend/src/jobProducer.ts diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts index b97fe248f..379ca388c 100644 --- a/packages/backend/src/index.ts +++ b/packages/backend/src/index.ts @@ -20,6 +20,8 @@ import { shutdownPosthog } from "./posthog.js"; import { PromClient } from './promClient.js'; import { RepoIndexManager } from "./repoIndexManager.js"; import { redis } from "./redis.js"; +import { BullMQJobManager, reconcile } from "./jobManager.js"; +import { QueueSpec, Workload } from "./types.js"; const logger = createLogger('backend-entrypoint'); @@ -83,6 +85,51 @@ const api = new Api( logger.info('Worker started.'); +// Background jobs run through the JobManager (BullMQ/Redis as the source of truth). Phase 0 +// wires the framework here in place of the old per-manager pollers; the real workloads +// (repo-index, connection-sync, permission syncers) are ported onto it in subsequent phases. +const jobManager = new BullMQJobManager(redis); + + +const demoSpec: QueueSpec<{ id: string }> = { + name: 'demo', + dedupKey: ({ id }) => `demo:${id}`, + jobOptions: { + attempts: 2, + backoff: { type: 'fixed', delayMs: 1000 }, + keep: { completed: 50, failed: 50 }, + }, +}; + +const demoWorkload: Workload<{ id: string }, { id: string; ranAt: number }> = { + spec: demoSpec, + concurrency: 2, + process: async ({ data: { id }, jobId, attemptsMade }) => { + logger.info(`demo: processing "${id}" (job ${jobId}, attempt ${attemptsMade})`); + await new Promise((resolve) => setTimeout(resolve, 1500)); + if (id === 'gamma') { + throw new Error(`demo: simulated failure processing "${id}"`); + } + return { id, ranAt: Date.now() }; + }, + onTerminalFailure: async ({ id }, err) => { + logger.warn(`demo: "${id}" failed terminally: ${err.message}`); + }, +}; +jobManager.register(demoWorkload); + +// The reconcile sweep for the demo queue, expressed as a cron workload: every 15s it triggers +// a fixed set into `demo`. Dedup keeps an in-flight item from re-queuing, but a finished one +// re-enqueues on the next tick, so the loop visibly cycles. +jobManager.registerCron(reconcile({ + name: 'demo-sweep', + schedule: { every: '15s' }, + target: 'demo', + scan: async () => ['alpha', 'beta', 'gamma'].map((id) => ({ id })), +})); + +await jobManager.start(); + const listenToShutdownSignals = () => { const signals = SHUTDOWN_SIGNALS; @@ -104,6 +151,7 @@ const listenToShutdownSignals = () => { await auditLogPruner.dispose() await attachmentPruner.dispose() await configManager.dispose() + await jobManager.stop(); await prisma.$disconnect(); await redis.quit(); diff --git a/packages/backend/src/jobManager.test.ts b/packages/backend/src/jobManager.test.ts new file mode 100644 index 000000000..c65b1c06d --- /dev/null +++ b/packages/backend/src/jobManager.test.ts @@ -0,0 +1,114 @@ +import { describe, expect, test, vi } from 'vitest'; + +// The module under test creates a logger at import time; stub it so importing pure helpers +// has no side effects (mirrors repoIndexManager.test.ts). +vi.mock('@sourcebot/shared', () => ({ + createLogger: vi.fn(() => ({ + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + })), +})); + +// Mock the constants module directly so its env-derived cache-dir paths don't load. +vi.mock('./constants.js', () => ({ + WORKER_STOP_GRACEFUL_TIMEOUT_MS: 5000, +})); + +import { normalizeJobState, parseDuration, reconcile } from './jobManager.js'; + +describe('parseDuration', () => { + test.each([ + ['500ms', 500], + ['30s', 30_000], + ['5m', 300_000], + ['6h', 21_600_000], + ['1d', 86_400_000], + ])('parses %s', (input, expected) => { + expect(parseDuration(input)).toBe(expected); + }); + + test('trims surrounding whitespace', () => { + expect(parseDuration(' 10m ')).toBe(600_000); + }); + + test.each(['', '5', 'm', '5x', '1.5h', '-5m', '5 m'])('throws on malformed "%s"', (input) => { + expect(() => parseDuration(input)).toThrow(); + }); +}); + +describe('normalizeJobState', () => { + test.each([ + 'waiting', + 'active', + 'delayed', + 'completed', + 'failed', + 'paused', + ])('passes through "%s"', (state) => { + expect(normalizeJobState(state)).toBe(state); + }); + + test('collapses prioritized and waiting-children to waiting', () => { + expect(normalizeJobState('prioritized')).toBe('waiting'); + expect(normalizeJobState('waiting-children')).toBe('waiting'); + }); + + test('maps anything unrecognized to unknown', () => { + expect(normalizeJobState('something-else')).toBe('unknown'); + expect(normalizeJobState('unknown')).toBe('unknown'); + }); +}); + +describe('reconcile', () => { + test('produces a cron workload carrying the given name and schedule', () => { + const cron = reconcile({ + name: 'x-sweep', + schedule: { every: '5m' }, + target: 'x', + scan: async () => [], + }); + expect(cron.name).toBe('x-sweep'); + expect(cron.schedule).toEqual({ every: '5m' }); + }); + + test('handler triggers each scanned item into the target, in order', async () => { + const triggered: Array<{ workload: string; data: unknown }> = []; + const cron = reconcile({ + name: 'x-sweep', + schedule: { pattern: '*/5 * * * *' }, + target: 'x', + scan: async () => [{ id: 'a' }, { id: 'b' }], + }); + + await cron.handler({ + trigger: async (workload, data) => { + triggered.push({ workload, data }); + }, + }); + + expect(triggered).toEqual([ + { workload: 'x', data: { id: 'a' } }, + { workload: 'x', data: { id: 'b' } }, + ]); + }); + + test('handler triggers nothing when scan returns empty', async () => { + let calls = 0; + const cron = reconcile({ + name: 'empty-sweep', + schedule: { every: '1m' }, + target: 'x', + scan: async () => [], + }); + + await cron.handler({ + trigger: async () => { + calls += 1; + }, + }); + + expect(calls).toBe(0); + }); +}); diff --git a/packages/backend/src/jobManager.ts b/packages/backend/src/jobManager.ts new file mode 100644 index 000000000..cd7689ef5 --- /dev/null +++ b/packages/backend/src/jobManager.ts @@ -0,0 +1,309 @@ +import * as Sentry from "@sentry/node"; +import { createLogger } from "@sourcebot/shared"; +import { Job, Queue, Worker } from "bullmq"; +import { Redis } from "ioredis"; +import { WORKER_STOP_GRACEFUL_TIMEOUT_MS } from "./constants.js"; +import { JobProducer } from "./jobProducer.js"; +import { CronWorkload, JobDetail, JobManager, QueueCounts, Schedule, Workload } from "./types.js"; + +const LOG_TAG = 'job-manager'; +const logger = createLogger(LOG_TAG); + +const CRON_QUEUE_NAME = 'cron'; +const CRON_KEEP_COMPLETED = 50; +const CRON_KEEP_FAILED = 200; + +const DURATION_UNITS_MS: Record = { + ms: 1, + s: 1000, + m: 1000 * 60, + h: 1000 * 60 * 60, + d: 1000 * 60 * 60 * 24, +}; + +export const parseDuration = (value: string): number => { + const match = /^(\d+)(ms|s|m|h|d)$/.exec(value.trim()); + if (!match) { + throw new Error(`Invalid duration "${value}". Expected e.g. "500ms", "30s", "5m", "6h", "1d".`); + } + return Number(match[1]) * DURATION_UNITS_MS[match[2]]; +}; + +export const normalizeJobState = (state: string): JobDetail['state'] => { + switch (state) { + case 'waiting': + case 'active': + case 'delayed': + case 'completed': + case 'failed': + case 'paused': + return state; + case 'prioritized': + case 'waiting-children': + return 'waiting'; + default: + return 'unknown'; + } +}; + +const scheduleToRepeat = (schedule: Schedule) => + 'pattern' in schedule ? { pattern: schedule.pattern } : { every: parseDuration(schedule.every) }; + +export function reconcile(opts: { + name: string; + schedule: Schedule; + target: string; + scan: () => Promise; +}): CronWorkload { + return { + name: opts.name, + schedule: opts.schedule, + handler: async ({ trigger }) => { + const items = await opts.scan(); + for (const item of items) { + await trigger(opts.target, item); + } + }, + }; +} + +export class BullMQJobManager implements JobManager { + private readonly workloads = new Map>(); + private readonly cronWorkloads = new Map(); + private readonly workers = new Map(); + private readonly producer: JobProducer; + private cronQueue?: Queue; + private cronWorker?: Worker; + private readonly abortController = new AbortController(); + + constructor(private readonly connection: Redis) { + this.producer = new JobProducer(connection); + } + + register(workload: Workload): void { + const name = workload.spec.name; + if (this.workloads.has(name)) { + throw new Error(`Workload "${name}" is already registered`); + } + this.workloads.set(name, workload as unknown as Workload); + } + + registerCron(cron: CronWorkload): void { + if (this.cronWorkloads.has(cron.name)) { + throw new Error(`Cron workload "${cron.name}" is already registered`); + } + this.cronWorkloads.set(cron.name, cron); + } + + async start(): Promise { + if (this.workloads.size === 0 && this.cronWorkloads.size === 0) { + logger.debug('start() called with nothing registered; nothing to do'); + return; + } + + for (const workload of this.workloads.values()) { + this.startWorkload(workload); + } + + if (this.cronWorkloads.size > 0) { + this.cronQueue = new Queue(CRON_QUEUE_NAME, { connection: this.connection }); + this.cronWorker = new Worker( + CRON_QUEUE_NAME, + (job) => this.runCron(job.name), + { connection: this.connection, concurrency: 1 }, + ); + this.cronWorker.on('failed', (job, error) => { + logger.error(`Cron "${job?.name}" run failed: ${error.message}`); + Sentry.captureException(error); + }); + this.cronWorker.on('error', (error) => { + logger.error('Cron worker error:', error); + }); + + for (const cron of this.cronWorkloads.values()) { + await this.cronQueue.upsertJobScheduler( + `cron:${cron.name}`, + scheduleToRepeat(cron.schedule), + { + name: cron.name, + opts: { + removeOnComplete: { count: CRON_KEEP_COMPLETED }, + removeOnFail: { count: CRON_KEEP_FAILED }, + }, + }, + ); + } + } + + logger.info( + `Started ${this.workloads.size} workload(s) [${[...this.workloads.keys()].join(', ') || '—'}] ` + + `and ${this.cronWorkloads.size} cron workload(s) [${[...this.cronWorkloads.keys()].join(', ') || '—'}]`, + ); + } + + async trigger(workloadName: string, data: T): Promise { + const workload = this.workloads.get(workloadName); + if (!workload) { + throw new Error(`Cannot trigger unknown workload "${workloadName}"`); + } + await this.producer.enqueue(workload.spec, data); + } + + async status(workloadName: string): Promise { + this.requireRegistered(workloadName); + const counts = await this.producer.queue(workloadName).getJobCounts( + 'waiting', + 'active', + 'delayed', + 'completed', + 'failed', + 'paused', + 'prioritized', + 'waiting-children', + ); + return { + waiting: counts.waiting ?? 0, + active: counts.active ?? 0, + delayed: counts.delayed ?? 0, + completed: counts.completed ?? 0, + failed: counts.failed ?? 0, + paused: counts.paused ?? 0, + prioritized: counts.prioritized ?? 0, + 'waiting-children': counts['waiting-children'] ?? 0, + }; + } + + async jobDetail(workloadName: string, jobId: string): Promise { + this.requireRegistered(workloadName); + const queue = this.producer.queue(workloadName); + const job = await queue.getJob(jobId); + if (!job) { + return null; + } + + const [state, jobLogs] = await Promise.all([ + job.getState(), + queue.getJobLogs(jobId), + ]); + + const enqueuedAt = job.timestamp; + const startedAt = job.processedOn ?? null; + const finishedAt = job.finishedOn ?? null; + + return { + id: job.id ?? jobId, + name: job.name, + state: normalizeJobState(state), + data: job.data, + attemptsMade: job.attemptsMade, + maxAttempts: job.opts.attempts ?? 1, + result: job.returnvalue ?? null, + failedReason: job.failedReason ?? null, + stacktrace: job.stacktrace ?? [], + logs: jobLogs.logs, + enqueuedAt, + startedAt, + finishedAt, + waitMs: startedAt !== null ? startedAt - enqueuedAt : null, + runMs: startedAt !== null && finishedAt !== null ? finishedAt - startedAt : null, + }; + } + + async stop(): Promise { + this.abortController.abort(); + + const workers = [...this.workers.values()]; + if (this.cronWorker) { + workers.push(this.cronWorker); + } + await Promise.all(workers.map((worker) => + Promise.race([ + worker.close(), + new Promise((resolve) => setTimeout(resolve, WORKER_STOP_GRACEFUL_TIMEOUT_MS)), + ]), + )); + + if (this.cronQueue) { + await this.cronQueue.close(); + } + await this.producer.close(); + + logger.info('Job manager stopped'); + } + + private startWorkload(workload: Workload): void { + const { spec, concurrency, rateLimit } = workload; + + this.producer.queue(spec.name); + + const worker = new Worker( + spec.name, + (job) => workload.process({ + data: job.data, + jobId: job.id ?? '', + attemptsMade: job.attemptsMade, + maxAttempts: job.opts.attempts ?? 1, + signal: this.abortController.signal, + log: async (message) => { await job.log(message); }, + updateProgress: (progress) => job.updateProgress(progress), + }), + { + connection: this.connection, + concurrency, + maxStalledCount: 1, + ...(rateLimit + ? { limiter: { max: rateLimit.max, duration: parseDuration(rateLimit.per) } } + : {}), + }, + ); + + worker.on('failed', (job, error) => { + void this.onWorkloadJobFailed(workload, job, error); + }); + worker.on('error', (error) => { + logger.error(`Worker "${spec.name}" error:`, error); + }); + + this.workers.set(spec.name, worker); + } + + private async runCron(cronName: string): Promise { + const cron = this.cronWorkloads.get(cronName); + if (!cron) { + logger.warn(`Cron fired for unknown workload "${cronName}"; skipping`); + return; + } + await cron.handler({ + trigger: (workload, data) => this.trigger(workload, data), + }); + } + + private async onWorkloadJobFailed( + workload: Workload, + job: Job | undefined, + error: Error, + ): Promise { + if (!job) { + return; + } + const maxAttempts = job.opts.attempts ?? 1; + const isTerminal = job.attemptsMade >= maxAttempts; + if (!isTerminal) { + logger.warn(`Workload "${workload.spec.name}" job ${job.id} failed attempt ${job.attemptsMade}/${maxAttempts}; will retry: ${error.message}`); + return; + } + logger.error(`Workload "${workload.spec.name}" job ${job.id} failed terminally after ${job.attemptsMade} attempt(s): ${error.message}`); + try { + await workload.onTerminalFailure?.(job.data, error); + } catch (hookError) { + Sentry.captureException(hookError); + logger.error(`onTerminalFailure for workload "${workload.spec.name}" threw:`, hookError); + } + } + + private requireRegistered(workloadName: string): void { + if (!this.workloads.has(workloadName)) { + throw new Error(`Workload "${workloadName}" is not registered`); + } + } +} diff --git a/packages/backend/src/jobProducer.ts b/packages/backend/src/jobProducer.ts new file mode 100644 index 000000000..f856d0beb --- /dev/null +++ b/packages/backend/src/jobProducer.ts @@ -0,0 +1,32 @@ +import { Queue } from "bullmq"; +import { Redis } from "ioredis"; +import { QueueSpec } from "./types.js"; + +export class JobProducer { + private readonly queues = new Map(); + + constructor(private readonly connection: Redis) {} + + queue(name: string): Queue { + let queue = this.queues.get(name); + if (!queue) { + queue = new Queue(name, { connection: this.connection }); + this.queues.set(name, queue); + } + return queue; + } + + async enqueue(spec: QueueSpec, data: T): Promise { + await this.queue(spec.name).add(spec.name, data, { + deduplication: { id: spec.dedupKey(data) }, + attempts: spec.jobOptions.attempts, + backoff: { type: spec.jobOptions.backoff.type, delay: spec.jobOptions.backoff.delayMs }, + removeOnComplete: { count: spec.jobOptions.keep.completed }, + removeOnFail: { count: spec.jobOptions.keep.failed }, + }); + } + + async close(): Promise { + await Promise.all([...this.queues.values()].map((queue) => queue.close())); + } +} diff --git a/packages/backend/src/types.ts b/packages/backend/src/types.ts index 8803b48b9..b14d5840b 100644 --- a/packages/backend/src/types.ts +++ b/packages/backend/src/types.ts @@ -24,4 +24,88 @@ export type RepoAuthCredentials = { * credentials for this repo. */ connectionConfig?: ConnectionConfig; -} \ No newline at end of file +} + +export interface ProcessContext { + data: TData; + jobId: string; + attemptsMade: number; + maxAttempts: number; + signal: AbortSignal; + log(message: string): Promise; + updateProgress(progress: number | object): Promise; +} + +export interface QueueSpec { + name: string; + dedupKey(data: TData): string; + jobOptions: { + attempts: number; + backoff: { type: 'fixed' | 'exponential'; delayMs: number }; + keep: { completed: number; failed: number }; + }; +} + +export interface Workload { + spec: QueueSpec; + concurrency: number; + rateLimit?: { max: number; per: string }; + process(ctx: ProcessContext): Promise; + onTerminalFailure?(data: TData, err: Error): Promise; +} + + +export type Schedule = { every: string } | { pattern: string }; + +export interface JobManager { + register(w: Workload): void; + registerCron(cron: CronWorkload): void; + + start(): Promise; + stop(): Promise; + + trigger(workload: string, data: T): Promise; + + status(workload: string): Promise; + jobDetail(workload: string, jobId: string): Promise; +} + +export interface CronWorkload { + name: string; + schedule: Schedule; + handler(ctx: CronContext): Promise; +} + +export interface CronContext { + trigger(workload: string, data: T): Promise; +} + + +export interface QueueCounts { + waiting: number; + active: number; + delayed: number; + completed: number; + failed: number; + paused: number; + prioritized?: number; + 'waiting-children'?: number; +} + +export interface JobDetail { + id: string; + name: string; + state: 'waiting' | 'active' | 'delayed' | 'completed' | 'failed' | 'paused' | 'unknown'; + data: TData; + attemptsMade: number; + maxAttempts: number; + result?: TResult | null; + failedReason?: string | null; + stacktrace?: string[]; + logs: string[]; + enqueuedAt: number; + startedAt: number | null; + finishedAt: number | null; + waitMs?: number | null; + runMs?: number | null; +}