diff --git a/dev-packages/bun-integration-tests/node-suites/excludes.ts b/dev-packages/bun-integration-tests/node-suites/excludes.ts index 65edcd311712..9813ed2262d8 100644 --- a/dev-packages/bun-integration-tests/node-suites/excludes.ts +++ b/dev-packages/bun-integration-tests/node-suites/excludes.ts @@ -62,6 +62,7 @@ const NO_FETCH_INSTRUMENTATION = [ // instrumentation creates no spans. Apps must be built with `@sentry/bun/plugin`. // See https://github.com/getsentry/sentry-javascript/issues/23882 export const NO_AUTO_INSTRUMENTATION = [ + 'suites/cron/graphile-worker/test.ts', 'suites/express/**', 'suites/fs-instrumentation/test.ts', 'suites/hono/test.ts', diff --git a/dev-packages/node-integration-tests/suites/cron/graphile-worker/docker-compose.yml b/dev-packages/node-integration-tests/suites/cron/graphile-worker/docker-compose.yml new file mode 100644 index 000000000000..3835e2258453 --- /dev/null +++ b/dev-packages/node-integration-tests/suites/cron/graphile-worker/docker-compose.yml @@ -0,0 +1,18 @@ +version: '3.9' + +services: + db: + image: postgres:16 + restart: always + ports: + - '5498:5432' + environment: + POSTGRES_USER: test + POSTGRES_PASSWORD: test + POSTGRES_DB: tests + healthcheck: + test: ['CMD-SHELL', 'pg_isready -U test -d tests'] + interval: 2s + timeout: 3s + retries: 30 + start_period: 5s diff --git a/dev-packages/node-integration-tests/suites/cron/graphile-worker/instrument.mjs b/dev-packages/node-integration-tests/suites/cron/graphile-worker/instrument.mjs new file mode 100644 index 000000000000..f4117ccb14b8 --- /dev/null +++ b/dev-packages/node-integration-tests/suites/cron/graphile-worker/instrument.mjs @@ -0,0 +1,9 @@ +import * as Sentry from '@sentry/node'; +import { loggingTransport } from '@sentry-internal/node-integration-tests'; + +Sentry.init({ + dsn: 'https://public@dsn.ingest.sentry.io/1337', + release: '1.0', + transport: loggingTransport, + integrations: [Sentry.graphileWorkerIntegration({ monitorCronJobs: true })], +}); diff --git a/dev-packages/node-integration-tests/suites/cron/graphile-worker/scenario.mjs b/dev-packages/node-integration-tests/suites/cron/graphile-worker/scenario.mjs new file mode 100644 index 000000000000..e12ceb643b7f --- /dev/null +++ b/dev-packages/node-integration-tests/suites/cron/graphile-worker/scenario.mjs @@ -0,0 +1,39 @@ +import { makeWorkerUtils, run } from 'graphile-worker'; + +const connectionString = 'postgres://test:test@localhost:5498/tests'; + +async function main() { + // Mark the crontab items as known two hours ago, so `?fill=2h` backfills their runs right away. + const utils = await makeWorkerUtils({ connectionString }); + await utils.migrate(); + await utils.withPgClient(client => + client.query( + `insert into graphile_worker._private_known_crontabs (identifier, known_since) + select unnest($1::text[]), now() - interval '2 hours' + on conflict (identifier) do update set known_since = excluded.known_since, last_execution = null`, + [['hourly-report', 'failing_task']], + ), + ); + await utils.release(); + + const runner = await run({ + connectionString, + concurrency: 1, + noHandleSignals: true, + pollInterval: 100, + crontab: ['0 * * * * report_task ?fill=2h&id=hourly-report', '30 * * * * failing_task ?fill=2h&max=1'].join('\n'), + taskList: { + report_task: async () => {}, + failing_task: async () => { + throw new Error('Cron job failed'); + }, + }, + }); + + setTimeout(async () => { + await runner.stop(); + process.exit(0); + }, 15_000); +} + +main(); diff --git a/dev-packages/node-integration-tests/suites/cron/graphile-worker/test.ts b/dev-packages/node-integration-tests/suites/cron/graphile-worker/test.ts new file mode 100644 index 000000000000..1dcd8ef4702c --- /dev/null +++ b/dev-packages/node-integration-tests/suites/cron/graphile-worker/test.ts @@ -0,0 +1,79 @@ +import { afterAll, describe, expect } from 'vitest'; +import { supports } from '../../../utils'; +import { cleanupChildProcesses, createEsmAndCjsTests, describeWithDockerCompose } from '../../../utils/runner'; + +describeWithDockerCompose('graphile-worker crontab monitoring', { workingDirectory: [__dirname] }, () => { + afterAll(() => { + cleanupChildProcesses(); + }); + + // 0.18 ships only ES modules. The CJS test `require()`s it, which the module hooks don't see on Node 20 or Deno. + for (const [version, runIf] of [ + ['0.16.6', true], + ['0.18.0', supports({ min: 22, runtimes: ['node'] })], + ] as const) { + describe.runIf(runIf)(`graphile-worker ${version}`, () => { + createEsmAndCjsTests( + __dirname, + 'scenario.mjs', + 'instrument.mjs', + (createRunner, test) => { + test('sends check-ins for crontab jobs', { timeout: 90_000 }, async () => { + await createRunner() + .unordered() + .expect({ + check_in: { + check_in_id: expect.any(String), + monitor_slug: 'hourly-report', + status: 'in_progress', + release: '1.0', + }, + }) + .expect({ + check_in: { + check_in_id: expect.any(String), + monitor_slug: 'hourly-report', + status: 'ok', + release: '1.0', + duration: expect.any(Number), + }, + }) + .expect({ + check_in: { + check_in_id: expect.any(String), + monitor_slug: 'failing_task', + status: 'in_progress', + release: '1.0', + }, + }) + .expect({ + check_in: { + check_in_id: expect.any(String), + monitor_slug: 'failing_task', + status: 'error', + release: '1.0', + duration: expect.any(Number), + }, + }) + .expect({ + event: { + exception: { + values: [ + { + type: 'Error', + value: 'Cron job failed', + mechanism: { type: 'auto.function.graphile_worker.cron', handled: false }, + }, + ], + }, + }, + }) + .start() + .completed(); + }); + }, + { additionalDependencies: { 'graphile-worker': version } }, + ); + }); + } +}); diff --git a/packages/astro/src/index.server.ts b/packages/astro/src/index.server.ts index dc77f7d5f6b9..9d0ccfa1007a 100644 --- a/packages/astro/src/index.server.ts +++ b/packages/astro/src/index.server.ts @@ -62,6 +62,7 @@ export { getTraceMetaTags, httpHeadersToSpanAttributes, winterCGHeadersToDict, + graphileWorkerIntegration, graphqlIntegration, hapiIntegration, httpIntegration, diff --git a/packages/aws-serverless/src/index.ts b/packages/aws-serverless/src/index.ts index cd302d5523f0..476fb7837eb0 100644 --- a/packages/aws-serverless/src/index.ts +++ b/packages/aws-serverless/src/index.ts @@ -116,6 +116,7 @@ export { firebaseIntegration, fsIntegration, genericPoolIntegration, + graphileWorkerIntegration, graphqlIntegration, knexIntegration, kafkaIntegration, diff --git a/packages/bun/src/index.ts b/packages/bun/src/index.ts index c0cc191e2850..4f1b2166b8c0 100644 --- a/packages/bun/src/index.ts +++ b/packages/bun/src/index.ts @@ -138,6 +138,7 @@ export { // oxlint-disable-next-line typescript/no-deprecated setupKoaErrorHandler, genericPoolIntegration, + graphileWorkerIntegration, graphqlIntegration, knexIntegration, kafkaIntegration, diff --git a/packages/elysia/src/index.ts b/packages/elysia/src/index.ts index 865ace1fc978..fdd79bdf0b3a 100644 --- a/packages/elysia/src/index.ts +++ b/packages/elysia/src/index.ts @@ -117,6 +117,7 @@ export { // oxlint-disable-next-line typescript/no-deprecated setupKoaErrorHandler, genericPoolIntegration, + graphileWorkerIntegration, graphqlIntegration, knexIntegration, kafkaIntegration, diff --git a/packages/google-cloud-serverless/src/index.ts b/packages/google-cloud-serverless/src/index.ts index 5bbf486e2342..a6497d7b9b66 100644 --- a/packages/google-cloud-serverless/src/index.ts +++ b/packages/google-cloud-serverless/src/index.ts @@ -116,6 +116,7 @@ export { fastifyIntegration, firebaseIntegration, genericPoolIntegration, + graphileWorkerIntegration, graphqlIntegration, knexIntegration, kafkaIntegration, diff --git a/packages/node/src/index.ts b/packages/node/src/index.ts index df5d309ee9e6..e869cc91cf41 100644 --- a/packages/node/src/index.ts +++ b/packages/node/src/index.ts @@ -12,6 +12,7 @@ export { firebaseIntegration, genericPoolIntegration, googleGenAIIntegration, + graphileWorkerIntegration, graphqlIntegration, groqIntegration, hapiIntegration, diff --git a/packages/server-utils/src/index.ts b/packages/server-utils/src/index.ts index 41eb3845cf1a..84d07bbb8b4a 100644 --- a/packages/server-utils/src/index.ts +++ b/packages/server-utils/src/index.ts @@ -39,6 +39,8 @@ export { awsIntegration } from './integrations/aws-sdk'; export { dataloaderIntegration } from './integrations/dataloader'; export { genericPoolIntegration } from './integrations/generic-pool'; export { googleGenAIIntegration } from './integrations/google-genai'; +export { graphileWorkerIntegration } from './integrations/graphile-worker'; +export type { GraphileWorkerOptions } from './integrations/graphile-worker'; export { graphqlIntegration } from './integrations/graphql'; export { hapiIntegration } from './integrations/hapi'; export { honoIntegration, honoMiddleware } from './integrations/hono'; diff --git a/packages/server-utils/src/integrations/graphile-worker.ts b/packages/server-utils/src/integrations/graphile-worker.ts new file mode 100644 index 000000000000..0767a70539a2 --- /dev/null +++ b/packages/server-utils/src/integrations/graphile-worker.ts @@ -0,0 +1,156 @@ +import * as diagnosticsChannel from '../utils/diagnosticsChannel'; +import type { IntegrationFn } from '@sentry/core'; +import { + captureCheckIn, + captureException, + debug, + defineIntegration, + isPlainObject, + stringMatchesSomePattern, + timestampInSeconds, +} from '@sentry/core'; +import { DEBUG_BUILD } from '../debug-build'; +import { CHANNELS } from '../orchestrion/channels'; +import { graphileWorkerModuleNames } from '../orchestrion/config/graphile-worker'; +import { invokeOrchestrionInstrumentation } from '../orchestrion/instrumentation'; + +const INTEGRATION_NAME = 'GraphileWorker' as const; + +export interface GraphileWorkerOptions { + /** + * Send check-ins for the jobs the crontab schedules. The monitor slug is the identifier of the + * crontab item (its `id` option, or else the task identifier). + * + * Defaults to `false`. + */ + monitorCronJobs?: boolean; + + /** + * Crontab item identifiers not to monitor. Strings must match the identifier exactly. + */ + excludeCronJobs?: Array; +} + +interface CronItem { + identifier: string; + task: string; + payload?: Record | null; + match: (digest: Record<'min' | 'hour' | 'date' | 'month' | 'dow', number>) => boolean; +} + +interface Job { + task_identifier?: string; + payload?: unknown; +} + +interface Events { + on(event: string, listener: (event: { job?: Job; error?: unknown }) => void): unknown; +} + +const _graphileWorkerIntegration = ((options: GraphileWorkerOptions = {}) => { + return { + name: INTEGRATION_NAME, + setup(client) { + if (options.monitorCronJobs) { + invokeOrchestrionInstrumentation(client, graphileWorkerModuleNames, instrumentGraphileWorker, [options], { + requiresTracingChannelBinding: false, + }); + } + }, + }; +}) satisfies IntegrationFn; + +/** + * Sends Crons check-ins for the jobs the `graphile-worker` crontab schedules. + * Requires the Sentry runtime hook or bundler plugin. + */ +export const graphileWorkerIntegration = defineIntegration(_graphileWorkerIntegration); + +function instrumentGraphileWorker(options: GraphileWorkerOptions): void { + const monitoredEvents = new WeakSet(); + + diagnosticsChannel.tracingChannel<{ arguments: unknown[] }>(CHANNELS.GRAPHILE_WORKER_RUN_CRON).subscribe({ + start(data) { + const [sharedOptions, items, requirements] = data.arguments as [ + { events?: Events }?, + unknown?, + { events?: Events }?, + ]; + const events = requirements?.events || sharedOptions?.events; + if (Array.isArray(items) && typeof events?.on === 'function' && !monitoredEvents.has(events)) { + monitoredEvents.add(events); + monitorGraphileCronJobs(events, items as CronItem[], options); + } + }, + end() {}, + asyncStart() {}, + asyncEnd() {}, + error() {}, + }); +} + +/** Sends check-ins for the jobs a runner's crontab schedules. Exported for tests. */ +export function monitorGraphileCronJobs(events: Events, items: CronItem[], options: GraphileWorkerOptions): void { + const runs = new WeakMap(); + + events.on('job:start', ({ job }) => { + try { + const item = job && findCronItem(items, job); + if (!job || !item || stringMatchesSomePattern(item.identifier, options.excludeCronJobs, true)) { + return; + } + const slug = item.identifier; + const checkInId = captureCheckIn({ monitorSlug: slug, status: 'in_progress' }); + runs.set(job, { slug, checkInId, start: timestampInSeconds() }); + } catch (e) { + DEBUG_BUILD && debug.error('[GraphileWorker] Failed to send the check-in:', e); + } + }); + + events.on('job:complete', ({ job, error }) => { + const run = job && runs.get(job); + if (!job || !run) { + return; + } + runs.delete(job); + try { + if (error) { + captureException(error, { mechanism: { handled: false, type: 'auto.function.graphile_worker.cron' } }); + } + captureCheckIn({ + monitorSlug: run.slug, + status: error ? 'error' : 'ok', + checkInId: run.checkInId, + duration: timestampInSeconds() - run.start, + }); + } catch (e) { + DEBUG_BUILD && debug.error('[GraphileWorker] Failed to send the check-in:', e); + } + }); +} + +// Cron jobs carry the task, the item's payload and `_cron.ts`, but not the item identifier. +function findCronItem(items: CronItem[], job: Job): CronItem | undefined { + if (!isPlainObject(job.payload)) { + return undefined; + } + const { _cron: cron, ...payload } = job.payload; + if (!isPlainObject(cron) || typeof cron.ts !== 'string') { + return undefined; + } + const ts = new Date(cron.ts); + const digest = { + min: ts.getUTCMinutes(), + hour: ts.getUTCHours(), + date: ts.getUTCDate(), + month: ts.getUTCMonth() + 1, + dow: ts.getUTCDay(), + }; + + let candidates = items.filter(item => item.task === job.task_identifier && item.match(digest)); + if (candidates.length > 1) { + const json = JSON.stringify(payload); + candidates = candidates.filter(item => JSON.stringify(item.payload || {}) === json); + } + return candidates.length === 1 ? candidates[0] : undefined; +} diff --git a/packages/server-utils/src/orchestrion/channels.ts b/packages/server-utils/src/orchestrion/channels.ts index b5919b9eb071..4e383e681097 100644 --- a/packages/server-utils/src/orchestrion/channels.ts +++ b/packages/server-utils/src/orchestrion/channels.ts @@ -6,6 +6,7 @@ import { expressChannels } from './config/express'; import { firebaseChannels } from './config/firebase'; import { genericPoolChannels } from './config/generic-pool'; import { googleGenAiChannels } from './config/google-genai'; +import { graphileWorkerChannels } from './config/graphile-worker'; import { graphqlChannels } from './config/graphql'; import { groqChannels } from './config/groq'; import { hapiChannels } from './config/hapi'; @@ -63,6 +64,7 @@ export const CHANNELS = { ...firebaseChannels, ...genericPoolChannels, ...googleGenAiChannels, + ...graphileWorkerChannels, ...graphqlChannels, ...groqChannels, ...hapiChannels, diff --git a/packages/server-utils/src/orchestrion/config/channel-integration-definitions.ts b/packages/server-utils/src/orchestrion/config/channel-integration-definitions.ts index d3583a3977ec..d53a37c8cc00 100644 --- a/packages/server-utils/src/orchestrion/config/channel-integration-definitions.ts +++ b/packages/server-utils/src/orchestrion/config/channel-integration-definitions.ts @@ -27,6 +27,7 @@ export const CHANNEL_INTEGRATION_DEFINITIONS = [ { exportName: 'knexIntegration', modules: ['knex'] }, { exportName: 'tediousIntegration', modules: ['tedious'] }, { exportName: 'genericPoolIntegration', modules: ['generic-pool'] }, + { exportName: 'graphileWorkerIntegration', modules: ['graphile-worker'] }, { exportName: 'lruMemoizerIntegration', modules: ['lru-memoizer'] }, { exportName: 'openAIIntegration', modules: ['openai'] }, { exportName: 'anthropicAIIntegration', modules: ['@anthropic-ai/sdk'] }, diff --git a/packages/server-utils/src/orchestrion/config/graphile-worker.ts b/packages/server-utils/src/orchestrion/config/graphile-worker.ts new file mode 100644 index 000000000000..99d3fc339c6e --- /dev/null +++ b/packages/server-utils/src/orchestrion/config/graphile-worker.ts @@ -0,0 +1,18 @@ +import type { InstrumentationConfig } from '../apmTypes'; +import { getModuleNames } from './module-names'; + +// `runCron(compiledSharedOptions, parsedCronItems, requirements)` starts the crontab scheduler of a +// runner. It is a `const` arrow in both the CJS (<0.18) and ESM (0.18) builds. +export const graphileWorkerConfig = [ + { + channelName: 'runCron', + module: { name: 'graphile-worker', versionRange: '>=0.16.0 <0.19', filePath: 'dist/cron.js' }, + functionQuery: { expressionName: 'runCron', kind: 'Sync' }, + }, +] satisfies InstrumentationConfig[]; + +export const graphileWorkerModuleNames = getModuleNames(graphileWorkerConfig); + +export const graphileWorkerChannels = { + GRAPHILE_WORKER_RUN_CRON: 'orchestrion:graphile-worker:runCron', +} as const; diff --git a/packages/server-utils/src/orchestrion/config/index.ts b/packages/server-utils/src/orchestrion/config/index.ts index 0cf0e363728f..793f76be43ca 100644 --- a/packages/server-utils/src/orchestrion/config/index.ts +++ b/packages/server-utils/src/orchestrion/config/index.ts @@ -10,6 +10,7 @@ import { expressConfig } from './express'; import { firebaseConfig } from './firebase'; import { genericPoolConfig } from './generic-pool'; import { googleGenAiConfig } from './google-genai'; +import { graphileWorkerConfig } from './graphile-worker'; import { graphqlConfig } from './graphql'; import { groqConfig } from './groq'; import { hapiConfig } from './hapi'; @@ -68,6 +69,7 @@ export const SENTRY_INSTRUMENTATIONS: InstrumentationConfig[] = [ ...firebaseConfig, ...genericPoolConfig, ...googleGenAiConfig, + ...graphileWorkerConfig, ...graphqlConfig, ...groqConfig, ...hapiConfig, diff --git a/packages/server-utils/test/integrations/graphile-worker.test.ts b/packages/server-utils/test/integrations/graphile-worker.test.ts new file mode 100644 index 000000000000..a118604d17b6 --- /dev/null +++ b/packages/server-utils/test/integrations/graphile-worker.test.ts @@ -0,0 +1,132 @@ +import { EventEmitter } from 'node:events'; +import * as SentryCore from '@sentry/core'; +import { setCurrentClient } from '@sentry/core'; +import type { MockInstance } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { monitorGraphileCronJobs } from '../../src/integrations/graphile-worker'; +import { getDefaultTestClientOptions, TestClient } from '../mocks/client'; + +type Digest = { min: number; hour: number; date: number; month: number; dow: number }; + +const ALL = (from: number, to: number): number[] => Array.from({ length: to - from + 1 }, (_, i) => from + i); + +// Mirrors a graphile-worker `ParsedCronItem` built from a crontab line. +function cronItem( + identifier: string, + task: string, + ranges: { minutes?: number[]; hours?: number[]; dates?: number[]; months?: number[]; dows?: number[] }, + payload: Record = {}, +) { + const parsedCronMatch = { + minutes: ALL(0, 59), + hours: ALL(0, 23), + dates: ALL(1, 31), + months: ALL(1, 12), + dows: ALL(0, 6), + ...ranges, + }; + const match = Object.assign( + (d: Digest) => + parsedCronMatch.minutes.includes(d.min) && + parsedCronMatch.hours.includes(d.hour) && + parsedCronMatch.months.includes(d.month), + { parsedCronMatch }, + ); + return { identifier, task, payload, match }; +} + +function cronJob(task: string, ts: string, payload: Record = {}) { + return { task_identifier: task, payload: { ...payload, _cron: { ts, backfilled: false } } }; +} + +describe('graphile-worker crontab monitoring', () => { + let captureCheckIn: MockInstance; + let captureException: MockInstance; + + beforeEach(() => { + const client = new TestClient(getDefaultTestClientOptions()); + setCurrentClient(client); + client.init(); + captureCheckIn = vi.spyOn(SentryCore, 'captureCheckIn').mockReturnValue('check-in-id'); + captureException = vi.spyOn(SentryCore, 'captureException').mockReturnValue('event-id'); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + function runJob(events: EventEmitter, job: object, error?: unknown): void { + events.emit('job:start', { job }); + events.emit('job:complete', { job, error }); + } + + it('sends check-ins using the item identifier as slug', () => { + const events = new EventEmitter(); + monitorGraphileCronJobs(events, [cronItem('hourly-report', 'report', { minutes: [0], dows: [1, 2, 3, 4, 5] })], {}); + + runJob(events, cronJob('report', '2026-10-16T09:00:00.000Z')); + + expect(captureCheckIn.mock.calls).toEqual([ + [{ monitorSlug: 'hourly-report', status: 'in_progress' }], + [{ monitorSlug: 'hourly-report', status: 'ok', checkInId: 'check-in-id', duration: expect.any(Number) }], + ]); + }); + + it('sends an error check-in and captures the error', () => { + const events = new EventEmitter(); + monitorGraphileCronJobs(events, [cronItem('report', 'report', { minutes: [0] })], {}); + const error = new Error('boom'); + + runJob(events, cronJob('report', '2026-10-16T09:00:00.000Z'), error); + + expect(captureCheckIn.mock.calls[1]?.[0]).toMatchObject({ monitorSlug: 'report', status: 'error' }); + expect(captureException).toHaveBeenCalledWith(error, { + mechanism: { handled: false, type: 'auto.function.graphile_worker.cron' }, + }); + }); + + it('tells items of the same task apart by schedule and payload', () => { + const events = new EventEmitter(); + monitorGraphileCronJobs( + events, + [ + cronItem('morning', 'send', { minutes: [0], hours: [8] }), + cronItem('evening-a', 'send', { minutes: [0], hours: [20] }, { list: 'a' }), + cronItem('evening-b', 'send', { minutes: [0], hours: [20] }, { list: 'b' }), + ], + {}, + ); + + runJob(events, cronJob('send', '2026-10-16T08:00:00.000Z')); + runJob(events, cronJob('send', '2026-10-16T20:00:00.000Z', { list: 'b' })); + + expect(captureCheckIn.mock.calls.map(([checkIn]) => checkIn.monitorSlug)).toEqual([ + 'morning', + 'morning', + 'evening-b', + 'evening-b', + ]); + }); + + it('skips jobs that were not scheduled by the crontab, ambiguous jobs and excluded items', () => { + const events = new EventEmitter(); + monitorGraphileCronJobs( + events, + [ + cronItem('a', 'same', { minutes: [0] }), + cronItem('b', 'same', { minutes: [0] }), + cronItem('excluded', 'excluded', { minutes: [0] }), + cronItem('other-excluded', 'other', { minutes: [0] }), + ], + { excludeCronJobs: ['excluded', /^other-/] }, + ); + + runJob(events, { task_identifier: 'same', payload: {} }); + runJob(events, cronJob('same', '2026-10-16T09:00:00.000Z')); + runJob(events, cronJob('excluded', '2026-10-16T09:00:00.000Z')); + runJob(events, cronJob('other', '2026-10-16T09:00:00.000Z')); + + expect(captureCheckIn).not.toHaveBeenCalled(); + expect(captureException).not.toHaveBeenCalled(); + }); +});