diff --git a/packages/sideffect/benchmarks/workflow-overhead.bench.ts b/packages/sideffect/benchmarks/workflow-overhead.bench.ts new file mode 100644 index 0000000..4e3952c --- /dev/null +++ b/packages/sideffect/benchmarks/workflow-overhead.bench.ts @@ -0,0 +1,286 @@ +import { performance } from "node:perf_hooks"; + +import { Effect } from "effect"; + +import { Schema, Step, Workflow, WorkflowEngine } from "../src/index.ts"; +import type { NativeWorkflowStep, WorkflowLayer } from "../src/types.ts"; +import type { WorkflowStepContext, WorkflowStepEvent } from "cloudflare:workers"; + +interface Payload { + readonly id: string; + readonly count: number; +} + +interface BenchResult { + readonly name: string; + readonly steps: number; + readonly samplesMs: ReadonlyArray; + readonly medianMs: number; + readonly minMs: number; + readonly maxMs: number; + readonly deltaFromBaselineMs: number; + readonly microsecondsPerStep: number; +} + +interface BenchCase { + readonly name: string; + readonly makeLayer: (steps: number) => WorkflowLayer; +} + +interface Options { + readonly steps: ReadonlyArray; + readonly samples: number; +} + +const payload: Payload = { id: "img_123", count: 1 }; +const payloadSchema = Schema.Struct({ id: Schema.String, count: Schema.Number }); +const workflow = Workflow.make({ name: "bench-workflow", payload: payloadSchema }); +const nativeStep = fakeNativeStep(); + +const directStep = Step.make("direct step", { + payload: payloadSchema, + result: payloadSchema, + run: (input: Payload) => input, +}); + +const promiseStep = Step.make("promise step", { + payload: payloadSchema, + result: payloadSchema, + run: (input: Payload) => Promise.resolve(input), +}); + +const effectSucceedStep = Step.make("effect succeed step", { + payload: payloadSchema, + result: payloadSchema, + run: (input: Payload) => Effect.succeed(input), +}); + +const untraced = Effect.fnUntraced(function* (input: Payload) { + return yield* Effect.succeed(input); +}); + +const effectFnUntracedStep = Step.make("effect fn untraced step", { + payload: payloadSchema, + result: payloadSchema, + run: (input: Payload) => untraced(input), +}); + +const traced = Effect.fn("benchmark.tracedStep")((input: Payload) => Effect.succeed(input)); + +const effectFnTracedStep = Step.make("effect fn traced step", { + payload: payloadSchema, + result: payloadSchema, + run: (input: Payload) => traced(input), +}); + +const cases: ReadonlyArray = [ + { + name: "sideffect-direct-struct", + makeLayer: (steps) => workflow.toLayer((_workflow, step) => runSteps(steps, step, directStep)), + }, + { + name: "sideffect-promise-step", + makeLayer: (steps) => workflow.toLayer((_workflow, step) => runSteps(steps, step, promiseStep)), + }, + { + name: "sideffect-effect-succeed-step", + makeLayer: (steps) => + workflow.toLayer((_workflow, step) => runSteps(steps, step, effectSucceedStep)), + }, + { + name: "sideffect-effect-fn-untraced-step", + makeLayer: (steps) => + workflow.toLayer((_workflow, step) => runSteps(steps, step, effectFnUntracedStep)), + }, + { + name: "sideffect-effect-fn-traced-step", + makeLayer: (steps) => + workflow.toLayer((_workflow, step) => runSteps(steps, step, effectFnTracedStep)), + }, + { + name: "sideffect-effect-workflow", + makeLayer: (steps) => + workflow.toLayer((_workflow, step) => + Effect.tryPromise({ + try: () => runSteps(steps, step, directStep), + catch: (cause) => cause, + }), + ), + }, + { + name: "sideffect-effect-per-step-wrapper", + makeLayer: (steps) => + workflow.toLayer(async (_workflow, step) => { + let current = payload; + for (let index = 0; index < steps; index++) { + current = await Effect.runPromise( + Effect.tryPromise({ + try: () => step.do(directStep, current), + catch: (cause) => cause, + }), + ); + } + return current; + }), + }, +]; + +const options = parseOptions(process.argv.slice(2)); +const results = await runBenchmarks(options); + +printMarkdown(results); +console.log("\nJSON:"); +console.log(JSON.stringify(results, null, 2)); + +async function runBenchmarks(options: Options): Promise> { + const results: Array = []; + + for (const steps of options.steps) { + const baseline = await runNativeBaseline(steps, options.samples); + results.push(baseline); + + for (const benchCase of cases) { + results.push(await runCase(benchCase, steps, options.samples, baseline.medianMs)); + } + } + + return results; +} + +async function runNativeBaseline(steps: number, samples: number): Promise { + await runNativeSteps(steps); + const samplesMs = await sample(samples, () => runNativeSteps(steps)); + return summarize("native-fake-baseline", steps, samplesMs, 0); +} + +async function runCase( + benchCase: BenchCase, + steps: number, + samples: number, + baselineMs: number, +): Promise { + const layer = benchCase.makeLayer(steps); + await runLayer(layer); + const samplesMs = await sample(samples, () => runLayer(layer)); + return summarize(benchCase.name, steps, samplesMs, baselineMs); +} + +async function sample( + samples: number, + run: () => Promise, +): Promise> { + const durations: Array = []; + for (let index = 0; index < samples; index++) { + const start = performance.now(); + await run(); + durations.push(performance.now() - start); + } + return durations; +} + +function summarize( + name: string, + steps: number, + samplesMs: ReadonlyArray, + baselineMs: number, +): BenchResult { + const sorted = [...samplesMs].sort((left, right) => left - right); + const medianMs = sorted[Math.floor(sorted.length / 2)] ?? 0; + return { + name, + steps, + samplesMs, + medianMs, + minMs: sorted[0] ?? 0, + maxMs: sorted.at(-1) ?? 0, + deltaFromBaselineMs: medianMs - baselineMs, + microsecondsPerStep: (medianMs * 1000) / steps, + }; +} + +async function runLayer(layer: WorkflowLayer): Promise { + return WorkflowEngine.run(layer, { + env: {}, + ctx: {}, + event: { + payload, + timestamp: new Date(0), + instanceId: "benchmark-instance", + workflowName: "bench-workflow", + }, + step: nativeStep, + }); +} + +async function runNativeSteps(steps: number): Promise { + let current = payload; + for (let index = 0; index < steps; index++) { + current = await nativeStep.do("native step", () => current); + } + return current; +} + +async function runSteps( + steps: number, + step: Parameters["run"]>[1], + stepDefinition: typeof directStep, +): Promise { + let current = payload; + for (let index = 0; index < steps; index++) { + current = await step.do(stepDefinition, current); + } + return current; +} + +function fakeNativeStep(): NativeWorkflowStep { + return { + do: (async (...args: Array) => { + const callback = args.find((arg) => typeof arg === "function") as ( + context: WorkflowStepContext, + ) => Payload | Promise; + return callback({ step: { name: String(args[0]), count: 1 }, attempt: 1, config: {} }); + }) as NativeWorkflowStep["do"], + async sleep() {}, + async sleepUntil() {}, + waitForEvent: (async ( + _name: string, + eventOptions: { readonly type: string; readonly timeout?: string | number }, + ): Promise> => ({ + payload: undefined as A, + timestamp: new Date(0), + type: eventOptions.type, + })) as NativeWorkflowStep["waitForEvent"], + }; +} + +function parseOptions(args: ReadonlyArray): Options { + const steps = valueFor(args, "--steps") + ?.split(",") + .map((value) => Number(value.trim())) + .filter((value) => Number.isInteger(value) && value > 0) ?? [1_000, 10_000]; + const samples = Number(valueFor(args, "--samples") ?? "5"); + + return { + steps, + samples: Number.isInteger(samples) && samples > 0 ? samples : 5, + }; +} + +function valueFor(args: ReadonlyArray, name: string): string | undefined { + const prefix = `${name}=`; + return args.find((arg) => arg.startsWith(prefix))?.slice(prefix.length); +} + +function printMarkdown(results: ReadonlyArray): void { + console.log("| steps | case | median ms | delta ms | us/step | min ms | max ms |"); + console.log("|---:|---|---:|---:|---:|---:|---:|"); + for (const result of results) { + console.log( + `| ${result.steps} | ${result.name} | ${format(result.medianMs)} | ${format(result.deltaFromBaselineMs)} | ${format(result.microsecondsPerStep)} | ${format(result.minMs)} | ${format(result.maxMs)} |`, + ); + } +} + +function format(value: number): string { + return value.toFixed(3); +} diff --git a/packages/sideffect/package.json b/packages/sideffect/package.json index e867444..7cf5252 100644 --- a/packages/sideffect/package.json +++ b/packages/sideffect/package.json @@ -27,6 +27,7 @@ }, "scripts": { "build": "vp pack", + "bench:workflow-overhead": "bun benchmarks/workflow-overhead.bench.ts", "dev": "vp pack --watch", "test": "vp test", "check": "vp check", diff --git a/packages/sideffect/src/runtime.ts b/packages/sideffect/src/runtime.ts index de025f3..4b00237 100644 --- a/packages/sideffect/src/runtime.ts +++ b/packages/sideffect/src/runtime.ts @@ -4,6 +4,8 @@ import { makeNonRetryableError } from "./errors.ts"; import type { NonRetryableErrorConstructor } from "./errors.ts"; import type { MaybeEffect } from "./types.ts"; +const schemaDecoders = new WeakMap unknown>(); + /** @internal Runtime options used when converting expected failures. */ export interface RuntimeErrorOptions { /** Optional native Cloudflare `NonRetryableError` constructor. */ @@ -28,7 +30,7 @@ export function decodeWithSchema( options: RuntimeErrorOptions = {}, ): A { try { - return Schema.decodeUnknownSync(schema as never)(value) as A; + return decoderFor(schema)(value) as A; } catch (cause) { throw makeNonRetryableError( `${operation} failed because the value did not match its schema. The workflow will not retry this invalid input. Cause: ${String(cause)}`, @@ -36,3 +38,14 @@ export function decodeWithSchema( ); } } + +function decoderFor(schema: Schema.Schema): (value: unknown) => A { + const cached = schemaDecoders.get(schema); + if (cached) { + return cached as (value: unknown) => A; + } + + const decoder = Schema.decodeUnknownSync(schema as never) as (value: unknown) => A; + schemaDecoders.set(schema, decoder); + return decoder; +} diff --git a/packages/sideffect/tests/index.test.ts b/packages/sideffect/tests/index.test.ts index 5613092..89b0963 100644 --- a/packages/sideffect/tests/index.test.ts +++ b/packages/sideffect/tests/index.test.ts @@ -299,6 +299,26 @@ test("workflow engine accepts Effect workflow bodies and preserves catchTag", as ).rejects.toBeInstanceOf(NonRetryableError); }); +test("workflow engine accepts Effect step bodies", async () => { + const effectStep = Step.make("effect step", { + payload: Schema.Struct({ imageId: Schema.String }), + result: Schema.Struct({ id: Schema.String }), + run: ({ imageId }) => Effect.succeed({ id: imageId }), + }); + const layer = imageWorkflow.toLayer(async (workflow, step) => { + return step.do(effectStep, { imageId: workflow.payload.imageId }); + }); + + await expect( + WorkflowEngine.run(layer, { + env: {}, + ctx: {}, + event: fakeWorkflowEvent({ imageId: "img_123" }), + step: fakeNativeStep(), + }), + ).resolves.toEqual({ id: "img_123" }); +}); + test("step.do forwards Cloudflare retry and timeout options", async () => { const calls: Array = []; const layer = imageWorkflow.toLayer(async (workflow, step) => { @@ -509,6 +529,42 @@ test("rollback handlers receive context and forward rollback config", async () = ]); }); +test("rollback handlers may return Effects", async () => { + const rollbacks: Array = []; + const stepWithEffectRollback = Step.make("step with effect rollback", { + payload: Schema.String, + result: Schema.String, + run: (value) => value, + }).pipe( + Rollback.with((result) => + Effect.sync(() => { + rollbacks.push(`rolled back:${result}`); + }), + ), + ); + const layer = imageWorkflow.toLayer(async (_workflow, step) => { + return step.do(stepWithEffectRollback, "value"); + }); + const calls: Array = []; + + await WorkflowEngine.run(layer, { + env: {}, + ctx: {}, + event: fakeWorkflowEvent({ imageId: "img_123" }), + step: fakeNativeStep(calls), + }); + + const rollbackOptions = (calls[0] as Array)[1] as { + readonly rollback: (context: { + readonly output: string; + readonly error: Error; + }) => Promise; + }; + await rollbackOptions.rollback({ output: "value", error: new Error("rollback") }); + + expect(rollbacks).toEqual(["rolled back:value"]); +}); + test("sleep, sleepUntil, and waitForEvent delegate to native WorkflowStep", async () => { const calls: Array = []; const date = new Date(0);