Skip to content
Effect 4.0 is here. One ecosystem. Zero dependencies.

WorkflowEngine unstable

Defines workflow engine services and an in-memory implementation.

WorkflowEngine registers workflow handlers, runs executions, polls results, resumes suspended runs, executes activities, stores durable deferred results, and schedules durable clocks. WorkflowInstance holds the runtime state for one workflow run. The in-memory layer is useful for tests and local development.

7 exports Since v4.0.0 Source

Constructors

makeDeferredState

unstable
Since v4.0.0

Creates deferred state shared by workflow engines.

Signature

declare function makeDeferredState(): DeferredState

makeUnsafe

unstable
Since v4.0.0

Builds a typed WorkflowEngine service from a low-level encoded implementation.

When to use

Use when wiring a trusted low-level workflow engine implementation into the typed WorkflowEngine service.

Gotchas

The implementation must correctly persist, resume, and encode workflow state.

Signature

declare function makeUnsafe(options: Encoded): {
  readonly activityExecute: <Success extends Constraint, Error extends Constraint, R>(activity: Activity<Success, Error, R>, attempt: number) => Effect<Result<Success["Type"], Error["Type"]>, never, WorkflowInstance | R | Success["DecodingServices"] | Error["DecodingServices"]>;
  readonly deferredDone: <Success extends Constraint, Error extends Constraint>(deferred: DurableDeferred<Success, Error>, options: {
    readonly deferredName: string;
    readonly executionId: string;
    readonly exit: Exit<Success["Type"], Error["Type"]>;
    readonly workflowName: string;
  }) => Effect<void, never, Success["EncodingServices"] | Error["EncodingServices"]>;
  readonly deferredResult: <Success extends Constraint, Error extends Constraint>(deferred: DurableDeferred<Success, Error>) => Effect<Option<Exit<Success["Type"], Error["Type"]>>, never, WorkflowInstance>;
  readonly execute: <Name extends string, Payload extends AnyStructSchema, Success extends Top, Error extends Top, Discard extends boolean = false>(workflow: Workflow<Name, Payload, Success, Error>, options: {
    readonly detached?: boolean;
    readonly discard?: Discard;
    readonly executionId: string;
    readonly payload: Payload["Type"];
    readonly suspendedRetrySchedule?: Schedule<any, unknown, never, never>;
  }) => Effect<Discard extends true ? string : Success["Type"], Error["Type"], Payload["EncodingServices"] | Success["DecodingServices"] | Error["DecodingServices"]>;
  readonly interrupt: (workflow: Any, executionId: string) => Effect<void>;
  readonly interruptUnsafe: (workflow: Any, executionId: string) => Effect<void>;
  readonly poll: <Name extends string, Payload extends AnyStructSchema, Success extends Top, Error extends Top>(workflow: Workflow<Name, Payload, Success, Error>, executionId: string) => Effect<Option<Result<Success["Type"], Error["Type"]>>, never, Success["DecodingServices"] | Error["DecodingServices"]>;
  readonly register: <Name extends string, Payload extends AnyStructSchema, Success extends Top, Error extends Top, R>(workflow: Workflow<Name, Payload, Success, Error>, execute: (payload: Payload["Type"], executionId: string) => Effect<Success["Type"], Error["Type"], R>) => Effect<void, never, Scope | Exclude<R, Scope | WorkflowEngine | WorkflowInstance | Execution<Name>> | Payload["DecodingServices"] | Payload["EncodingServices"] | Success["DecodingServices"] | Success["EncodingServices"] | Error["DecodingServices"] | Error["EncodingServices"]>;
  readonly resume: (workflow: Any, executionId: string) => Effect<void>;
  readonly scheduleClock: (workflow: Any, options: {
    readonly clock: DurableClock;
    readonly executionId: string;
  }) => Effect<void>;
}

Layers

layerMemory

unstable
Since v4.0.0

Layer that provides an in-memory WorkflowEngine.

When to use

Use to run tests and local development workflows where durability is not needed.

Gotchas

This layer keeps state only in memory and is not suitable for production workflows that require durability.

Signature

declare const layerMemory: Layer.Layer<WorkflowEngine>

Models

DeferredState

interface unstable
Since v4.0.0

Tracks live workflow runs so deferred completions can wake them.

Signature

interface DeferredState {
  readonly deferredDone: (executionId: string, name: string) => Effect<void>;
  readonly trackRun: <A, E, R>(instance: {
    abandoned: boolean;
    readonly activityState: {
      count: number;
      readonly latch: Latch;
    };
    readonly awaitedDeferreds: Map<string, number>;
    cause: Cause<never> | undefined;
    readonly completedDeferreds: Set<string>;
    readonly executionId: string;
    failed: boolean;
    interrupted: boolean;
    readonly scope: Closeable;
    suspended: boolean;
    readonly workflow: Any;
  }, effect: Effect<A, E, R>) => Effect<A, E, Exclude<R, WorkflowInstance>>;
}

Services

Encoded

interface unstable
Since v4.0.0

Low-level workflow engine contract that works with encoded payloads and results before makeUnsafe adds typed schema decoding and encoding.

Signature

interface Encoded {
  readonly activityExecute: (activity: Any, attempt: number) => Effect<Result<unknown, unknown>, never, WorkflowInstance>;
  readonly awaitComplete?: (workflow: Any, executionId: string) => Effect<void>;
  readonly deferredDone: (options: {
    readonly deferredName: string;
    readonly executionId: string;
    readonly exit: Exit<unknown, unknown>;
    readonly workflowName: string;
  }) => Effect<void>;
  readonly deferredResult: (deferred: Any) => Effect<Option<Exit<unknown, unknown>>, never, WorkflowInstance>;
  readonly execute: <Discard extends boolean>(workflow: Any, options: {
    readonly detached?: boolean;
    readonly discard: Discard;
    readonly executionId: string;
    readonly parent?: {
      abandoned: boolean;
      readonly activityState: {
        count: number;
        readonly latch: Latch;
      };
      readonly awaitedDeferreds: Map<string, number>;
      cause: Cause<never> | undefined;
      readonly completedDeferreds: Set<string>;
      readonly executionId: string;
      failed: boolean;
      interrupted: boolean;
      readonly scope: Closeable;
      suspended: boolean;
      readonly workflow: Any;
    };
    readonly payload: object;
  }) => Effect<Discard extends true ? void : Result<unknown, unknown>>;
  readonly interrupt: (workflow: Any, executionId: string) => Effect<void>;
  readonly interruptUnsafe: (workflow: Any, executionId: string) => Effect<void>;
  readonly poll: (workflow: Any, executionId: string) => Effect<Option<Result<unknown, unknown>>>;
  readonly register: (workflow: Any, execute: (payload: object, executionId: string) => Effect<unknown, unknown, WorkflowEngine | WorkflowInstance>) => Effect<void, never, Scope>;
  readonly resume: (workflow: Any, executionId: string) => Effect<void>;
  readonly scheduleClock: (workflow: Any, options: {
    readonly clock: DurableClock;
    readonly executionId: string;
  }) => Effect<void>;
}

WorkflowEngine

unstable
Since v4.0.0

Service that represents workflow runtimes, responsible for registering and executing workflows and coordinating activities, durable deferreds, interrupts, resumes, and clocks.

Signature

declare class WorkflowEngine extends Shape<"effect/workflow/WorkflowEngine", {
  readonly activityExecute: <Success extends Constraint, Error extends Constraint, R>(activity: Activity<Success, Error, R>, attempt: number) => Effect<Result<Success["Type"], Error["Type"]>, never, WorkflowInstance | R | Success["DecodingServices"] | Error["DecodingServices"]>;
  readonly deferredDone: <Success extends Constraint, Error extends Constraint>(deferred: DurableDeferred<Success, Error>, options: {
    readonly deferredName: string;
    readonly executionId: string;
    readonly exit: Exit<Success["Type"], Error["Type"]>;
    readonly workflowName: string;
  }) => Effect<void, never, Success["EncodingServices"] | Error["EncodingServices"]>;
  readonly deferredResult: <Success extends Constraint, Error extends Constraint>(deferred: DurableDeferred<Success, Error>) => Effect<Option<Exit<Success["Type"], Error["Type"]>>, never, WorkflowInstance>;
  readonly execute: <Name extends string, Payload extends AnyStructSchema, Success extends Top, Error extends Top, Discard extends boolean = false>(workflow: Workflow<Name, Payload, Success, Error>, options: {
    readonly detached?: boolean;
    readonly discard?: Discard;
    readonly executionId: string;
    readonly payload: Payload["Type"];
    readonly suspendedRetrySchedule?: Schedule<any, unknown, never, never>;
  }) => Effect<Discard extends true ? string : Success["Type"], Error["Type"], Payload["EncodingServices"] | Success["DecodingServices"] | Error["DecodingServices"]>;
  readonly interrupt: (workflow: Any, executionId: string) => Effect<void>;
  readonly interruptUnsafe: (workflow: Any, executionId: string) => Effect<void>;
  readonly poll: <Name extends string, Payload extends AnyStructSchema, Success extends Top, Error extends Top>(workflow: Workflow<Name, Payload, Success, Error>, executionId: string) => Effect<Option<Result<Success["Type"], Error["Type"]>>, never, Success["DecodingServices"] | Error["DecodingServices"]>;
  readonly register: <Name extends string, Payload extends AnyStructSchema, Success extends Top, Error extends Top, R>(workflow: Workflow<Name, Payload, Success, Error>, execute: (payload: Payload["Type"], executionId: string) => Effect<Success["Type"], Error["Type"], R>) => Effect<void, never, Scope | Exclude<R, Scope | WorkflowEngine | WorkflowInstance | Execution<Name>> | Payload["DecodingServices"] | Payload["EncodingServices"] | Success["DecodingServices"] | Success["EncodingServices"] | Error["DecodingServices"] | Error["EncodingServices"]>;
  readonly resume: (workflow: Any, executionId: string) => Effect<void>;
  readonly scheduleClock: (workflow: Any, options: {
    readonly clock: DurableClock;
    readonly executionId: string;
  }) => Effect<void>;
}, this> {
  constructor(_: never);
}

WorkflowInstance

unstable
Since v4.0.0

Service that contains workflow runtime state for one execution.

When to use

Use to read or update workflow execution, suspension, interruption, lifetime, failure, and activity coordination state inside workflow engine internals.

Details

The service stores the execution ID, workflow definition, long-lived scope, suspension and interruption flags, the stored failure cause, and activity coordination state for a single workflow run.

Signature

declare class WorkflowInstance extends Shape<"effect/workflow/WorkflowEngine/WorkflowInstance", {
  abandoned: boolean;
  readonly activityState: {
    count: number;
    readonly latch: Latch;
  };
  readonly awaitedDeferreds: Map<string, number>;
  cause: Cause<never> | undefined;
  readonly completedDeferreds: Set<string>;
  readonly executionId: string;
  failed: boolean;
  interrupted: boolean;
  readonly scope: Closeable;
  suspended: boolean;
  readonly workflow: Any;
}, this> {
  constructor(_: never);
  static initial(workflow: Any, executionId: string, options?: {
    readonly awaitedDeferreds?: Map<string, number>;
    readonly scope?: Closeable;
  }): {
    abandoned: boolean;
    readonly activityState: {
      count: number;
      readonly latch: Latch;
    };
    readonly awaitedDeferreds: Map<string, number>;
    cause: Cause<never> | undefined;
    readonly completedDeferreds: Set<string>;
    readonly executionId: string;
    failed: boolean;
    interrupted: boolean;
    readonly scope: Closeable;
    suspended: boolean;
    readonly workflow: Any;
  };
}