diff --git a/apps/docs/docs/workers.mdx b/apps/docs/docs/workers.mdx index 62ac061a..9efe0371 100644 --- a/apps/docs/docs/workers.mdx +++ b/apps/docs/docs/workers.mdx @@ -188,6 +188,32 @@ bunx @openworkflow/cli worker start +## Logging + +Workers log operational errors, such as failed polls and heartbeats, to +`console` by default. Pass a `logger` with `debug`, `info`, `warn`, and `error` +methods to send them somewhere else, such as a structured logger: + +```ts +import pino from "pino"; + +const log = pino(); + +const ow = new OpenWorkflow({ + backend, + logger: { + debug: (message, ...args) => log.debug({ args }, message), + info: (message, ...args) => log.info({ args }, message), + warn: (message, ...args) => log.warn({ args }, message), + // workers pass the error after the message; `err` gets pino's serializer + error: (message, err, ...args) => log.error({ err, args }, message), + }, +}); +``` + +Workers created with `ow.newWorker()` use the client's logger. You can also +pass `logger` to `newWorker()` or to `new Worker()` directly. + ## Workflow Discovery The CLI automatically discovers workflows from directories specified in your diff --git a/packages/openworkflow/client/client.ts b/packages/openworkflow/client/client.ts index 8e5cb4cc..7f2c234b 100644 --- a/packages/openworkflow/client/client.ts +++ b/packages/openworkflow/client/client.ts @@ -1,6 +1,7 @@ import type { Backend, SendSignalResult } from "../core/backend.js"; import type { DurationString } from "../core/duration.js"; import type { JsonValue } from "../core/json.js"; +import type { Logger } from "../core/logger.js"; import type { StandardSchemaV1 } from "../core/standard-schema.js"; import { calculateDateFromDuration } from "../core/step-attempt.js"; import { @@ -44,6 +45,11 @@ type WorkflowRunInput = SchemaInput; */ export interface OpenWorkflowOptions { backend: Backend; + /** + * Default logger for workers created with `newWorker()`. + * @default console + */ + logger?: Logger | undefined; } /** @@ -51,10 +57,12 @@ export interface OpenWorkflowOptions { */ export class OpenWorkflow { private backend: Backend; + private logger: Logger | undefined; private registry = new WorkflowRegistry(); constructor(options: OpenWorkflowOptions) { this.backend = options.backend; + this.logger = options.logger; } /** @@ -62,16 +70,18 @@ export class OpenWorkflow { * @param options - Worker options * @param options.concurrency - Max concurrent workflow runs * @param options.leaseDurationMs - Workflow run lease duration + * @param options.logger - Logger, overriding the client's logger * @returns Worker instance */ newWorker( - options?: Pick, + options?: Pick, ): Worker { return new Worker({ backend: this.backend, workflows: this.registry.getAll(), concurrency: options?.concurrency, leaseDurationMs: options?.leaseDurationMs, + logger: options?.logger ?? this.logger, }); } diff --git a/packages/openworkflow/core/logger.ts b/packages/openworkflow/core/logger.ts new file mode 100644 index 00000000..443f6fdc --- /dev/null +++ b/packages/openworkflow/core/logger.ts @@ -0,0 +1,16 @@ +/** + * Receives operational log output from OpenWorkflow, such as worker loop and + * heartbeat errors. The method signatures match `console`, so `console` itself + * is a valid logger, and structured loggers can be adapted with a thin wrapper. + */ +export interface Logger { + debug(message: string, ...args: unknown[]): void; + info(message: string, ...args: unknown[]): void; + warn(message: string, ...args: unknown[]): void; + error(message: string, ...args: unknown[]): void; +} + +/** + * The default logger, which writes to the global `console`. + */ +export const consoleLogger: Logger = console; diff --git a/packages/openworkflow/index.ts b/packages/openworkflow/index.ts index fee98c7e..595274c2 100644 --- a/packages/openworkflow/index.ts +++ b/packages/openworkflow/index.ts @@ -3,6 +3,7 @@ export type { OpenWorkflowOptions } from "./client/client.js"; export { OpenWorkflow } from "./client/client.js"; // core +export type { Logger } from "./core/logger.js"; export type { RetryPolicy, Workflow } from "./core/workflow-definition.js"; export type { WorkflowRunMetadata } from "./core/workflow-function.js"; export { diff --git a/packages/openworkflow/worker/worker.test.ts b/packages/openworkflow/worker/worker.test.ts index e07bb845..7221bc7c 100644 --- a/packages/openworkflow/worker/worker.test.ts +++ b/packages/openworkflow/worker/worker.test.ts @@ -1701,6 +1701,175 @@ describe("Worker", () => { } }); + describe("logger", () => { + test("logs tick failures to console by default", async () => { + const error = new Error("boom"); + const consoleErrorSpy = vi.spyOn(console, "error").mockReturnValue(); + const worker = new Worker({ + backend: createStubBackend({ + claimWorkflowRun: vi.fn().mockRejectedValue(error), + }), + workflows: [], + }); + + try { + await worker.start(); + await worker.stop(); + + expect(consoleErrorSpy).toHaveBeenCalledWith( + "Worker tick failed:", + error, + ); + } finally { + consoleErrorSpy.mockRestore(); + } + }); + + test("logs tick failures to a custom logger", async () => { + const error = new Error("boom"); + const consoleErrorSpy = vi.spyOn(console, "error").mockReturnValue(); + const logger = createTestLogger(); + const worker = new Worker({ + backend: createStubBackend({ + claimWorkflowRun: vi.fn().mockRejectedValue(error), + }), + workflows: [], + logger, + }); + + try { + await worker.start(); + await worker.stop(); + + expect(logger.error).toHaveBeenCalledWith("Worker tick failed:", error); + expect(consoleErrorSpy).not.toHaveBeenCalled(); + } finally { + consoleErrorSpy.mockRestore(); + } + }); + + test("logs heartbeat failures to a custom logger", async () => { + vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] }); + + const backend = await createTestBackend(); + const error = new Error("heartbeat boom"); + vi.spyOn(backend, "extendWorkflowRunLease").mockRejectedValue(error); + const logger = createTestLogger(); + const client = new OpenWorkflow({ backend }); + const stepStarted = Promise.withResolvers(); + const releaseStep = Promise.withResolvers(); + + const workflow = client.defineWorkflow( + { name: "heartbeat-logger" }, + async ({ step }) => { + await step.run({ name: "wait" }, async () => { + stepStarted.resolve(null); + await releaseStep.promise; + }); + return "done"; + }, + ); + const worker = client.newWorker({ logger }); + + try { + const handle = await workflow.run(); + await worker.tick(); + await stepStarted.promise; + + await vi.advanceTimersByTimeAsync(15_000); + expect(logger.error).toHaveBeenCalledWith("Heartbeat failed:", error); + + releaseStep.resolve(null); + await expect(handle.result()).resolves.toBe("done"); + } finally { + releaseStep.resolve(null); + await worker.stop(); + vi.useRealTimers(); + } + }); + + test("logs critical execution failures to a custom logger", async () => { + const backend = await createTestBackend(); + const error = new Error("backend boom"); + vi.spyOn(backend, "failWorkflowRun").mockRejectedValue(error); + const logger = createTestLogger(); + const client = new OpenWorkflow({ backend }); + const workflow = client.defineWorkflow( + { name: "critical-execution-logger" }, + () => { + throw new Error("workflow boom"); + }, + ); + const worker = client.newWorker({ logger }); + const handle = await workflow.run(); + + await worker.tick(); + await expect(worker.stop()).resolves.toBeUndefined(); + + expect(logger.error).toHaveBeenCalledWith( + `Critical error during workflow execution for run ${handle.workflowRun.id}:`, + error, + ); + expect(logger.error.mock.calls[0]?.[1]).toBe(error); + }); + + test("keeps polling when the logger throws", async () => { + vi.useFakeTimers(); + + const claimWorkflowRun = vi.fn().mockRejectedValue(new Error("boom")); + const logger = createTestLogger(); + logger.error.mockImplementation(() => { + throw new Error("logger boom"); + }); + const worker = new Worker({ + backend: createStubBackend({ claimWorkflowRun }), + workflows: [], + logger, + }); + + try { + await worker.start(); + await vi.advanceTimersByTimeAsync(100); + + expect(claimWorkflowRun).toHaveBeenCalledTimes(2); + expect(logger.error).toHaveBeenCalledTimes(2); + } finally { + const stopPromise = worker.stop(); + await vi.runOnlyPendingTimersAsync(); + await stopPromise; + vi.useRealTimers(); + } + }); + + test("uses the client logger unless the worker overrides it", async () => { + const error = new Error("boom"); + const backend = createStubBackend({ + claimWorkflowRun: vi.fn().mockRejectedValue(error), + }); + const clientLogger = createTestLogger(); + const workerLogger = createTestLogger(); + const client = new OpenWorkflow({ backend, logger: clientLogger }); + + const defaultWorker = client.newWorker(); + await defaultWorker.start(); + await defaultWorker.stop(); + expect(clientLogger.error).toHaveBeenCalledWith( + "Worker tick failed:", + error, + ); + + clientLogger.error.mockClear(); + const overriddenWorker = client.newWorker({ logger: workerLogger }); + await overriddenWorker.start(); + await overriddenWorker.stop(); + expect(workerLogger.error).toHaveBeenCalledWith( + "Worker tick failed:", + error, + ); + expect(clientLogger.error).not.toHaveBeenCalled(); + }); + }); + test("respects custom retry policy from spec", async () => { const backend = await createTestBackend(); const client = new OpenWorkflow({ backend }); @@ -2288,6 +2457,10 @@ function retryDelayMs(run: WorkflowRun, fallbackStartMs: number): number { return run.availableAt.getTime() - fallbackStartMs; } +function createTestLogger() { + return { debug: vi.fn(), info: vi.fn(), warn: vi.fn(), error: vi.fn() }; +} + function sleep(ms: number): Promise { return new Promise((resolve) => { setTimeout(resolve, ms); diff --git a/packages/openworkflow/worker/worker.ts b/packages/openworkflow/worker/worker.ts index 3878a4bc..a72d272d 100644 --- a/packages/openworkflow/worker/worker.ts +++ b/packages/openworkflow/worker/worker.ts @@ -1,6 +1,7 @@ import type { Backend } from "../core/backend.js"; import { type BackoffPolicy, computeBackoffDelayMs } from "../core/backoff.js"; import { parseDuration } from "../core/duration.js"; +import { type Logger, consoleLogger } from "../core/logger.js"; import type { RetryPolicy, Workflow } from "../core/workflow-definition.js"; import { DEFAULT_WORKFLOW_RETRY_POLICY } from "../core/workflow-definition.js"; import { WorkflowRegistry } from "../core/workflow-registry.js"; @@ -45,6 +46,11 @@ export interface WorkerOptions { * @default 30000 (30 seconds) */ leaseDurationMs?: number | undefined; + /** + * Logger for worker errors, such as failed polls and heartbeats. + * @default console + */ + logger?: Logger | undefined; } /** @@ -55,6 +61,7 @@ export class Worker { private readonly backend: Backend; private readonly workerIds: string[]; private readonly leaseDurationMs: number; + private readonly logger: Logger; private readonly registry = new WorkflowRegistry(); private readonly activeExecutions = new Set(); private running = false; @@ -63,6 +70,7 @@ export class Worker { constructor(options: WorkerOptions) { this.backend = options.backend; + this.logger = options.logger ?? consoleLogger; for (const workflow of options.workflows) { this.registry.register(workflow); @@ -156,7 +164,7 @@ export class Worker { await sleep(getPollBackoffDelayMs(this.backoffAttempts)); } } catch (error) { - console.error("Worker tick failed:", error); + logError(this.logger, "Worker tick failed:", error); this.backoffAttempts += 1; await sleep(getPollBackoffDelayMs(this.backoffAttempts)); } @@ -205,6 +213,7 @@ export class Worker { workflowRun, workerId, leaseDurationMs: this.leaseDurationMs, + logger: this.logger, }); this.activeExecutions.add(execution); @@ -246,7 +255,8 @@ export class Worker { } catch (error) { // specifically for unexpected errors in the execution wrapper itself, not // for business logic errors (those are handled inside executeWorkflow) - console.error( + logError( + this.logger, `Critical error during workflow execution for run ${execution.workflowRun.id}:`, error, ); @@ -262,6 +272,7 @@ interface WorkflowExecutionOptions { workflowRun: WorkflowRun; workerId: string; leaseDurationMs: number; + logger: Logger; } /** @@ -273,6 +284,7 @@ class WorkflowExecution { workflowRun: WorkflowRun; workerId: string; private readonly leaseDurationMs: number; + private readonly logger: Logger; private heartbeatTimer: NodeJS.Timeout | null = null; constructor(options: WorkflowExecutionOptions) { @@ -280,6 +292,7 @@ class WorkflowExecution { this.workflowRun = options.workflowRun; this.workerId = options.workerId; this.leaseDurationMs = options.leaseDurationMs; + this.logger = options.logger; } /** @@ -302,7 +315,7 @@ class WorkflowExecution { leaseDurationMs, }) .catch((cause: unknown) => { - console.error("Heartbeat failed:", cause); + logError(this.logger, "Heartbeat failed:", cause); }); }), heartbeatIntervalMs, @@ -320,6 +333,20 @@ class WorkflowExecution { } } +/** + * Log an error without letting a failing logger stop the worker loop or + * surface as an unhandled rejection. + * @param logger - Logger to write to + * @param args - Arguments for `logger.error` + */ +function logError(logger: Logger, ...args: Parameters): void { + try { + logger.error(...args); + } catch { + // a failing logger must not take down the worker + } +} + /** * Sleep for a given duration. * @param ms - Milliseconds to sleep