Run durably with Workflow SDK

An Execution Adapter is the runtime strategy that executes Migration Definitions inline or starts them through a durable workflow provider. The Workflow SDK Execution Adapter uses durable steps when a Migration Run must outlive an HTTP request or worker. Definitions retain their Source, processing style, Migration Store, and any Tracking Record Contract or Rollback Pipeline they declare.

The boundary between the starting host and the worker is a Migration Execution Envelope: a serializable execution request containing the Migration Run ID, Migration Definition Registry ID, request, and diagnostic planned order. The Workflow SDK Migration Execution Envelope also carries the complete acquired Migration Definition Lock set transferred to the provider-owned execution.

Separate the responsibilities

Starting through the Migrate Server resolves the requested selection against its authoritative registry. The adapter acquires locks and records the Migration Run before handing the envelope to Workflow SDK. The worker resolves the registry ID and re-plans before doing work. Planned order in the envelope is diagnostic metadata, not a frozen executable plan; functions, Layers, and database connections remain in their own runtime.

The Migration Store stays authoritative for migration progress. Workflow history explains which steps ran; it does not replace Migration Item State or tracking. Normal execution still discovers eligible Source Items in source order. Failed-item retries and explicit identity targeting remain separate modes. Migration Run Observation is also separate: a client can stop observing while the durable provider continues execution.

Use the repository integration

This recipe follows the complete Workflow SDK example application. Its concrete workload is a PostgreSQL catalog of authors and books, with a SQL Migration Store and a Next.js Migrate Server. The same host and step boundaries apply to other sources and destinations. Read Defining migrations and Migration stores before adapting the integration.

Run the example as a complete monorepo checkout. The code blocks below show files from that application; their adjacent imports, workspace packages, database setup, and Next.js configuration are supplied by the checkout. Use this map when adapting it:

To bring this into your own app, replace catalogRegistry with your self-contained definitions, retain a stable registry ID and shared durable store, and wire your own authorization. Keep the workflow, step, and host modules separate as shown.

Start the complete example locally

From a checkout of this monorepo, use Node.js 24 or later, pnpm, and Docker:

pnpm install
cd examples/workflow-sdk
cp .env.example .env.local
pnpm db:up
pnpm db:setup
pnpm dev

The example's environment file points DATABASE_URL to its local PostgreSQL container on port 55433 and supplies development tokens. Database setup installs the migration-store schema and seeds the catalog. Use this dedicated demo database; the optional db:setup -- --reset command drops and recreates demo data.

Open http://127.0.0.1:3100. In a second terminal at the repository root, connect the TUI:

MIGRATE_SERVER_TOKEN=local-migrate-secret \
  pnpm --filter @migrate-sdk/tui dev -- \
  --server http://127.0.0.1:3100/api/migrate

Start authors or the catalog group. Inspect the workflow from examples/workflow-sdk with:

pnpm exec workflow inspect runs
pnpm exec workflow web

The example uses workflow 4.x with withWorkflow in next.config.ts. That integration transforms "use workflow" and "use step" functions. If adapting it to another Next.js application, follow the official Next.js setup, including excluding Workflow's internal routes from application proxy middleware. For Turborepo caching, include generated app/.well-known/workflow/** or src/app/.well-known/workflow/** routes among build outputs.

Keep the workflow entry point small

Import orchestration helpers from @migrate-sdk/workflow-sdk/workflow inside the workflow module. That entry point is safe for Workflow's deterministic runtime. Keep Effect execution, database clients, migration definitions, and server imports in the step module.

This is the complete workflow entry point used by the example:

import {
  runMigrationExecutionWorkflow,
  runMigrationRollbackWorkflow,
  type WorkflowSdkMigrationRollbackEnvelope,
  type WorkflowSdkMigrationRollbackSummary,
  type WorkflowSdkMigrationRunEnvelope,
  type WorkflowSdkMigrationRunSummary,
} from "@migrate-sdk/workflow-sdk/workflow";
import {
  beginMigrationRunStep,
  cancelMigrationRunStep,
  completeMigrationRunStep,
  executeMigrationRollbackStep,
  executeMigrationRunCursorWindowStep,
  executeMigrationRunRollbackOrphansPageStep,
  failMigrationRunStep,
} from "./workflow-steps";

export async function catalogMigrationWorkflow(
  envelope:
    | WorkflowSdkMigrationRunEnvelope
    | WorkflowSdkMigrationRollbackEnvelope
): Promise<
  WorkflowSdkMigrationRunSummary | WorkflowSdkMigrationRollbackSummary
> {
  "use workflow";

  if (envelope.kind === "rollback") {
    return await runMigrationRollbackWorkflow(envelope, {
      execute: executeMigrationRollbackStep,
    });
  }

  return await runMigrationExecutionWorkflow(envelope, {
    begin: beginMigrationRunStep,
    cancel: cancelMigrationRunStep,
    complete: completeMigrationRunStep,
    executeCursorWindow: executeMigrationRunCursorWindowStep,
    executeRollbackOrphansPage: executeMigrationRunRollbackOrphansPageStep,
    fail: failMigrationRunStep,
  });
}

Forward execution has a begin step, repeated cursor-window steps, optional orphan-cleanup pages, and terminal completion, cancellation, or failure steps. The helper carries cumulative progress between steps. The rollback branch delegates to the separate rollback helper and its execution step.

Do not replace this with one "use step" around the entire forward migration. The Source Cursor Window boundary lets the adapter coordinate durable progress across smaller units of work. One window can contain several Source Items; it is not one provider step per item.

Rebuild the runtime inside steps

Every step needs access to the same registry ID, compatible definitions, and durable store. A process-local in-memory store loses that relationship when a different worker executes the next step.

The example supplies a PostgreSQL store:

import { PgClient } from "@effect/sql-pg";
import { Config } from "effect";
import { SqlMigrationStore } from "migrate-sdk/stores/sql";

export const migrationStoreTablePrefix = "migrate_workflow_sdk_example";

export const PostgresLive = PgClient.layerConfig({
  url: Config.Redacted("DATABASE_URL"),
});

export const migrationStore = SqlMigrationStore.layerFromClient(PostgresLive, {
  tablePrefix: migrationStoreTablePrefix,
});

Its step runtime combines four layers:

• migrationStore, explicitly available for finalization and lock release.
• MigrationDefinitionRegistryCatalog.layer, which resolves the envelope's registry ID to catalogRegistry.
• MigrationRunStepExecutor.defaultLayer, which executes forward cursor windows.
• MigrationRollbackExecutor.layer, which executes rollback.

The definitions supply their source and destination dependencies. The server and step workers must use the same database, table prefix, and compatible registry. Supplying the original store explicitly also lets finalization find durable ownership if a definition is removed or the registry changes during deployment; an ambiguous or missing store cannot safely be guessed.

The full step module below includes imports, runtime composition, telemetry, and every lifecycle function. runEffect turns an Effect into a Promise at this runtime boundary. Workflow and migration run IDs are attached to spans so separate step traces can be correlated.

import {
  beginMigrationRunExecutionEnvelope,
  cancelMigrationRunExecutionEnvelope,
  completeMigrationRunExecutionEnvelope,
  disableWorkflowStepRetries,
  executeMigrationRollbackExecutionEnvelope,
  executeMigrationRunCursorWindow,
  executeMigrationRunRollbackOrphansPage,
  failMigrationRunExecutionEnvelope,
} from "@migrate-sdk/workflow-sdk/steps";
import type {
  WorkflowSdkMigrationRollbackEnvelope,
  WorkflowSdkMigrationRollbackSummary,
  WorkflowSdkMigrationRunCursorWindowResult,
  WorkflowSdkMigrationRunCursorWindowState,
  WorkflowSdkMigrationRunEnvelope,
  WorkflowSdkMigrationRunRollbackOrphansPageResult,
  WorkflowSdkMigrationRunRollbackOrphansState,
  WorkflowSdkMigrationRunSummary,
} from "@migrate-sdk/workflow-sdk/workflow";
import { Effect, Layer } from "effect";
import {
  MigrationDefinitionRegistryCatalog,
  MigrationRollbackExecutor,
  MigrationRunStepExecutor,
} from "migrate-sdk/core";
import { migrationTelemetryLayer } from "migrate-sdk/telemetry";
import { getStepMetadata, getWorkflowMetadata } from "workflow";
import { catalogRegistry } from "./catalog";
import { migrationStore } from "./database";

const RuntimeLive = Layer.mergeAll(
  migrationStore,
  MigrationDefinitionRegistryCatalog.layer({
    registries: [catalogRegistry],
  }),
  MigrationRollbackExecutor.layer,
  MigrationRunStepExecutor.defaultLayer
);

const runEffect = <A, E>(
  effect: Effect.Effect<
    A,
    E,
    | MigrationDefinitionRegistryCatalog
    | MigrationRollbackExecutor
    | MigrationRunStepExecutor
  >,
  name: string,
  envelope:
    | WorkflowSdkMigrationRunEnvelope
    | WorkflowSdkMigrationRollbackEnvelope,
  definitionId?: string
) => {
  const step = getStepMetadata();
  const workflow = getWorkflowMetadata();
  return Effect.runPromise(
    effect.pipe(
      Effect.provide(RuntimeLive),
      Effect.withSpan(`workflow.step.${name}`, {
        attributes: {
          "migration.run.id": envelope.runId,
          ...(definitionId === undefined
            ? {}
            : { "migration.definition.id": definitionId }),
          "workflow.run.id": workflow.workflowRunId,
          "workflow.name": workflow.workflowName,
          "workflow.step.id": step.stepId,
          "workflow.step.name": step.stepName,
          "workflow.step.attempt": step.attempt,
        },
      }),
      Effect.provide(migrationTelemetryLayer)
    )
  );
};

export async function beginMigrationRunStep(
  envelope: WorkflowSdkMigrationRunEnvelope
): Promise<{ readonly rollbackOrphans: boolean }> {
  "use step";

  return await runEffect(
    beginMigrationRunExecutionEnvelope(envelope),
    "begin",
    envelope
  );
}

export async function executeMigrationRunCursorWindowStep(input: {
  readonly definitionId: WorkflowSdkMigrationRunEnvelope["executionDefinitionIds"][number];
  readonly envelope: WorkflowSdkMigrationRunEnvelope;
  readonly runId: WorkflowSdkMigrationRunEnvelope["runId"];
  readonly state: WorkflowSdkMigrationRunCursorWindowState;
}): Promise<WorkflowSdkMigrationRunCursorWindowResult> {
  "use step";

  return await runEffect(
    executeMigrationRunCursorWindow({
      definitionId: input.definitionId,
      envelope: input.envelope,
      runId: input.runId,
      state: input.state,
    }),
    "cursor-window",
    input.envelope,
    input.definitionId
  );
}

export async function executeMigrationRunRollbackOrphansPageStep(input: {
  readonly definitionId: WorkflowSdkMigrationRunEnvelope["executionDefinitionIds"][number];
  readonly envelope: WorkflowSdkMigrationRunEnvelope;
  readonly runId: WorkflowSdkMigrationRunEnvelope["runId"];
  readonly state: WorkflowSdkMigrationRunRollbackOrphansState;
}): Promise<WorkflowSdkMigrationRunRollbackOrphansPageResult> {
  "use step";

  return await runEffect(
    executeMigrationRunRollbackOrphansPage({
      definitionId: input.definitionId,
      envelope: input.envelope,
      runId: input.runId,
      state: input.state,
    }),
    "rollback-orphans-page",
    input.envelope,
    input.definitionId
  );
}

export async function completeMigrationRunStep(input: {
  readonly definitions: WorkflowSdkMigrationRunSummary["definitions"];
  readonly envelope: WorkflowSdkMigrationRunEnvelope;
}): Promise<WorkflowSdkMigrationRunSummary> {
  "use step";

  return await runEffect(
    completeMigrationRunExecutionEnvelope({
      definitions: input.definitions,
      envelope: input.envelope,
    }),
    "complete",
    input.envelope
  );
}

export async function cancelMigrationRunStep(input: {
  readonly definitions: WorkflowSdkMigrationRunSummary["definitions"];
  readonly envelope: WorkflowSdkMigrationRunEnvelope;
}): Promise<WorkflowSdkMigrationRunSummary> {
  "use step";

  return await runEffect(
    cancelMigrationRunExecutionEnvelope({
      definitions: input.definitions,
      envelope: input.envelope,
    }),
    "cancel",
    input.envelope
  );
}

export async function failMigrationRunStep(input: {
  readonly definitions: WorkflowSdkMigrationRunSummary["definitions"];
  readonly envelope: WorkflowSdkMigrationRunEnvelope;
  readonly error: unknown;
  readonly failedDefinitionId?: WorkflowSdkMigrationRunEnvelope["executionDefinitionIds"][number];
}): Promise<void> {
  "use step";

  return await runEffect(
    failMigrationRunExecutionEnvelope({
      definitions: input.definitions,
      envelope: input.envelope,
      error: input.error,
      ...(input.failedDefinitionId === undefined
        ? {}
        : { failedDefinitionId: input.failedDefinitionId }),
    }),
    "fail",
    input.envelope,
    input.failedDefinitionId
  );
}

export async function executeMigrationRollbackStep(
  envelope: WorkflowSdkMigrationRollbackEnvelope
): Promise<WorkflowSdkMigrationRollbackSummary> {
  "use step";

  return await runEffect(
    executeMigrationRollbackExecutionEnvelope(envelope),
    "rollback",
    envelope
  );
}

disableWorkflowStepRetries(executeMigrationRunCursorWindowStep);
disableWorkflowStepRetries(executeMigrationRunRollbackOrphansPageStep);

Choose retries at the correct boundary

Workflow SDK normally retries thrown step errors; maxRetries = 0 disables that automatic retry for a function. See its retry contract.

The example deliberately applies disableWorkflowStepRetries to both the cursor-window step and the orphan-cleanup page step. Keep those calls when adapting the module.

Those steps save item outcomes and cursor or cleanup progress in the Migration Store. That commit is separate from Workflow SDK saving the step's return value. If the return value is lost after migration progress is committed, automatically invoking the same step with old workflow state can consume work against an already-advanced checkpoint.

Lifecycle and finalization steps remain retryable so a transient failure does not prevent durable completion, cancellation, failure recording, or lock release. Do not apply a blanket no-retry policy to every step.

There are still three different recovery mechanisms:

Disabling automatic cursor-step retries does not provide exactly-once destination writes. A worker can die after a destination accepts a write but before the Migration Item State is saved. Use an upsert keyed by the stable Source Identity or a destination-supported idempotency key. For multi-request work, save useful operation receipts in the Destination Journal so recovery can reason about partial effects. See Tracking and Retries.

Give the server the durable adapter

The server uses WorkflowSdkMigrationExecutable.layer instead of inline execution. It starts the configured workflow through WorkflowSdkClient.layer:

import {
  WorkflowSdkClient,
  WorkflowSdkMigrationExecutable,
} from "@migrate-sdk/workflow-sdk";
import { Layer } from "effect";
import {
  MigrateServerHttp,
  RegistryMigrateServer,
} from "migrate-sdk/server/http";
import { catalogRegistry } from "../migrations/catalog";
import { catalogMigrationWorkflow } from "../migrations/catalog-migration.workflow";

const executableLayer = WorkflowSdkMigrationExecutable.layer({
  workflow: catalogMigrationWorkflow,
}).pipe(Layer.provide(WorkflowSdkClient.layer));

const environmentLabel =
  process.env.VERCEL_PROJECT_PRODUCTION_URL === undefined
    ? "Local Workflow dev server"
    : `https://${process.env.VERCEL_PROJECT_PRODUCTION_URL}/`;

export const migrateServerLayer = RegistryMigrateServer.layer({
  environment: {
    id: process.env.VERCEL_ENV ?? "local",
    label: environmentLabel,
  },
  registry: catalogRegistry,
}).pipe(Layer.provide(executableLayer));

export const migrateServerHttpLayer = MigrateServerHttp.layer.pipe(
  Layer.provide(migrateServerLayer)
);

The example mounts this layer at POST /api/migrate using MigrateServerHttp.toWebHandler and application-owned authorization middleware. Clients use that protocol endpoint; they do not invoke your workflow function directly or construct lock envelopes themselves.

import { Layer } from "effect";
import { MigrateServerHttp } from "migrate-sdk/server/http";
import { migrateServerAuthorizationMiddleware } from "@/server/auth";
import { migrateServerHttpLayer } from "@/server/migrate-server";
import { MigrateServerAccess } from "@/server/migrate-server-access";

const migrateServerHttp = MigrateServerHttp.toWebHandler(
  Layer.merge(migrateServerHttpLayer, MigrateServerAccess.layer),
  migrateServerAuthorizationMiddleware
);

export const runtime = "nodejs";
export const maxDuration = 60;

export const POST = (request: Request) => migrateServerHttp.handler(request);

import { Effect } from "effect";
import {
  HttpMiddleware,
  HttpServerRequest,
  HttpServerResponse,
} from "effect/http";
import { DemoSetupAccess } from "./demo-setup-access";
import { MigrateServerAccess } from "./migrate-server-access";

const isAuthorizedBearerHeader = (
  authorization: string | undefined,
  token: string | undefined
): boolean =>
  token !== undefined &&
  token.length > 0 &&
  authorization === `Bearer ${token}`;

export const isAuthorizedDemoSetupRequest = (
  request: Request
): Effect.Effect<boolean, never, DemoSetupAccess> =>
  DemoSetupAccess.pipe(
    Effect.map(({ token }) =>
      isAuthorizedBearerHeader(
        request.headers.get("authorization") ?? undefined,
        token
      )
    )
  );

export const isAuthorizedCronRequest = (request: Request): boolean =>
  isAuthorizedBearerHeader(
    request.headers.get("authorization") ?? undefined,
    process.env.CRON_SECRET
  );

export const migrateServerAuthorizationMiddleware = HttpMiddleware.make(
  (httpApp) =>
    Effect.gen(function* () {
      const request = yield* HttpServerRequest.HttpServerRequest;
      const { token } = yield* MigrateServerAccess;
      if (!isAuthorizedBearerHeader(request.headers.authorization, token)) {
        return HttpServerResponse.text("Unauthorized", { status: 401 });
      }
      return yield* httpApp;
    })
);

import { Config, Context, Effect, Layer } from "effect";

export interface MigrateServerAccessValue {
  readonly token: string;
}

export function makeMigrateServerAccess(
  token: string
): MigrateServerAccessValue {
  return { token: token.trim() };
}

export class MigrateServerAccess extends Context.Service<
  MigrateServerAccess,
  MigrateServerAccessValue
>()("@migrate-sdk/examples/workflow-sdk/MigrateServerAccess") {
  static readonly layer = Layer.effect(
    MigrateServerAccess,
    Config.String("MIGRATE_SERVER_TOKEN").pipe(
      Config.withDefault(""),
      Effect.map(makeMigrateServerAccess)
    )
  );
}

import { Config, Context, Effect, Layer } from "effect";

export interface DemoSetupAccessValue {
  readonly token: string;
}

export function makeDemoSetupAccess(token: string): DemoSetupAccessValue {
  return { token: token.trim() };
}

export class DemoSetupAccess extends Context.Service<
  DemoSetupAccess,
  DemoSetupAccessValue
>()("@migrate-sdk/examples/workflow-sdk/DemoSetupAccess") {
  static readonly layer = Layer.effect(
    DemoSetupAccess,
    Config.String("DEMO_SETUP_TOKEN").pipe(
      Config.withDefault(""),
      Effect.map(makeDemoSetupAccess)
    )
  );
}

The HTTP request can finish after the workflow has accepted the run. Observation reconnects separately using the Migration Run ID. A scheduled job can use the same server service to prepare and start an operation; the example's /api/cron/import route does this for the catalog group, preserving the same planning and lock rules as a TUI request.

Stop and recover

Use the connected TUI's safe-stop action or address the same server from the CLI. From examples/workflow-sdk, with MIGRATE_SERVER_TOKEN set to your server's token:

pnpm exec migrate --server http://127.0.0.1:3100/api/migrate runs list
pnpm exec migrate --server http://127.0.0.1:3100/api/migrate runs stop RUN_ID
pnpm exec migrate --server http://127.0.0.1:3100/api/migrate runs observe RUN_ID

Replace RUN_ID with the active Migration Run ID. The server records a cancellation request in the Migration Store; execution stops scheduling new work, drains already-started work, and finalizes cancellation. Keep observing until the run is terminal. Closing the terminal only detaches observation.

Avoid directly terminating the provider workflow as your normal stop mechanism: it can bypass migration cleanup and leave locks behind. If that happens, establish that the provider can no longer run before following lock recovery.

If a cursor step fails, inspect durable item state and the workflow error before starting another migration run. The new run uses the saved cursor and eligibility rules. Failed-item retry selects saved failures; it cannot recover a write whose result was never durably recorded without the destination's own replay protection.

Deploy the integration

For the repository example, create a Vercel project rooted at examples/workflow-sdk. Set DATABASE_URL to a reachable PostgreSQL database and configure MIGRATE_SERVER_TOKEN, DEMO_SETUP_TOKEN, and CRON_SECRET for that environment. Enable Vercel system environment variables so Workflow can identify its deployment. The official deployment guide explains runtime selection and alternative Workflow Worlds.

Initialize the dedicated demo database through the example's authenticated /api/demo/setup endpoint or the setup script against that database. The demo's setup and reset endpoints exist for its fixture data; they are not a general production database migration strategy.

The example intentionally displays its Migrate Server token in the browser to let visitors connect. When adapting it for a private environment, remove that public token display and keep authorization in your application or access gateway. Use a real secret manager for client credentials.

Keep source, destination, and store credentials on the deployed server and workers. Confirm that every worker resolves the same registry and store, and exercise a limited run, reconnect, failed-item retry, and cooperative stop in a staging environment before relying on unattended operation. A local type check or docs build does not verify your database or provider deployment.