feat(node): Add Crons monitoring for graphile-worker crontab jobs

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',

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

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 })],
+});

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();

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 } },
+      );
+    });
+  }
+});

packages/astro/src/index.server.ts

@@ -62,6 +62,7 @@ export {
   getTraceMetaTags,
   httpHeadersToSpanAttributes,
   winterCGHeadersToDict,
+  graphileWorkerIntegration,
   graphqlIntegration,
   hapiIntegration,
   httpIntegration,

packages/aws-serverless/src/index.ts

@@ -116,6 +116,7 @@ export {
   firebaseIntegration,
   fsIntegration,
   genericPoolIntegration,
+  graphileWorkerIntegration,
   graphqlIntegration,
   knexIntegration,
   kafkaIntegration,

packages/bun/src/index.ts

@@ -138,6 +138,7 @@ export {
   // oxlint-disable-next-line typescript/no-deprecated
   setupKoaErrorHandler,
   genericPoolIntegration,
+  graphileWorkerIntegration,
   graphqlIntegration,
   knexIntegration,
   kafkaIntegration,

packages/elysia/src/index.ts

@@ -117,6 +117,7 @@ export {
   // oxlint-disable-next-line typescript/no-deprecated
   setupKoaErrorHandler,
   genericPoolIntegration,
+  graphileWorkerIntegration,
   graphqlIntegration,
   knexIntegration,
   kafkaIntegration,

packages/google-cloud-serverless/src/index.ts

@@ -116,6 +116,7 @@ export {
   fastifyIntegration,
   firebaseIntegration,
   genericPoolIntegration,
+  graphileWorkerIntegration,
   graphqlIntegration,
   knexIntegration,
   kafkaIntegration,

packages/node/src/index.ts

@@ -12,6 +12,7 @@ export {
   firebaseIntegration,
   genericPoolIntegration,
   googleGenAIIntegration,
+  graphileWorkerIntegration,
   graphqlIntegration,
   groqIntegration,
   hapiIntegration,

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';

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<string | RegExp>;
+}
+
+interface CronItem {
+  identifier: string;
+  task: string;
+  payload?: Record<string, unknown> | 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<object>();
+
+  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<Job, { slug: string; checkInId: string; start: number }>();
+
+  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;
+}

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,

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'] },