1/**
2 * Defines typed durable workflows.
3 *
4 * A `Workflow` has a stable tag, schemas for payload, success, and failure, and
5 * an idempotency key used to derive execution ids. Workflow definitions can be
6 * executed, discarded, polled, interrupted, resumed, and registered with a
7 * handler layer. This module also includes workflow result types, compensation
8 * and cleanup helpers, suspension support, and settings for defect capture or
9 * failure suspension.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Arr from "../Array.ts"
15import * as Cause from "../Cause.ts"
16import * as Context from "../Context.ts"
17import * as Data from "../Data.ts"
18import * as Effect from "../Effect.ts"
19import * as Exit from "../Exit.ts"
20import * as Fiber from "../Fiber.ts"
21import * as Filter from "../Filter.ts"
22import { constFalse, constTrue, dual, identity } from "../Function.ts"
23import * as Layer from "../Layer.ts"
24import type * as Option from "../Option.ts"
25import * as Predicate from "../Predicate.ts"
26import type { ExitEncoded } from "../rpc/RpcMessage.ts"
27import type * as Schedule from "../Schedule.ts"
28import * as Schema from "../Schema.ts"
29import * as SchemaIssue from "../SchemaIssue.ts"
30import * as SchemaParser from "../SchemaParser.ts"
31import * as Tranformation from "../SchemaTransformation.ts"
32import * as Scope from "../Scope.ts"
33import { makeHashDigest } from "./internal/crypto.ts"
34import type { WorkflowEngine, WorkflowInstance } from "./WorkflowEngine.ts"
35
36const TypeId = "~effect/workflow/Workflow"
37
38/**
39 * Durable workflow definition with typed payload, success, and error schemas
40 * plus operations for execution, polling, interruption, resumption, and
41 * registration.
42 *
43 * @stability unstable
44 * @category models
45 * @since 4.0.0
46 */
47export interface Workflow<
48 Tag extends string,
49 Payload extends AnyStructSchema,
50 Success extends Schema.Top,
51 Error extends Schema.Top
52> {
53 new(_: never): {}
54
55 readonly [TypeId]: typeof TypeId
56 readonly _tag: Tag
57 readonly payloadSchema: Payload
58 readonly successSchema: Success
59 readonly errorSchema: Error
60 readonly annotations: Context.Context<never>
61 readonly idempotencyKey: (payload: Payload["Type"]) => string
62 readonly suspendedRetrySchedule?: Schedule.Schedule<any, unknown> | undefined
63
64 /**
65 * Add an annotation to the workflow.
66 */
67 annotate<I, S>(
68 key: Context.Key<I, S>,
69 value: S
70 ): Workflow<Tag, Payload, Success, Error>
71
72 /**
73 * Merge multiple annotations into the workflow.
74 */
75 annotateMerge<I>(
76 annotations: Context.Context<I>
77 ): Workflow<Tag, Payload, Success, Error>
78
79 /**
80 * Execute the workflow with the given payload.
81 */
82 readonly execute: <const Discard extends boolean = false>(
83 payload: Payload["~type.make.in"],
84 options?: {
85 readonly discard?: Discard
86 }
87 ) => Effect.Effect<
88 Discard extends true ? string : Success["Type"],
89 Discard extends true ? never : Error["Type"],
90 | WorkflowEngine
91 | Payload["EncodingServices"]
92 | Success["DecodingServices"]
93 | Error["DecodingServices"]
94 >
95
96 /**
97 * Poll the current status of a workflow execution.
98 */
99 readonly poll: (
100 executionId: string
101 ) => Effect.Effect<
102 Option.Option<Result<Success["Type"], Error["Type"]>>,
103 never,
104 WorkflowEngine | Success["DecodingServices"] | Error["DecodingServices"]
105 >
106
107 /**
108 * Interrupt a workflow execution for the given execution ID.
109 */
110 readonly interrupt: (
111 executionId: string
112 ) => Effect.Effect<void, never, WorkflowEngine>
113
114 /**
115 * Manually resume a workflow execution for the given execution ID.
116 */
117 readonly resume: (
118 executionId: string
119 ) => Effect.Effect<void, never, WorkflowEngine>
120
121 /**
122 * Create a layer that registers the workflow and provides an effect to
123 * execute it.
124 */
125 readonly toLayer: <R>(
126 execute: (
127 payload: Payload["Type"],
128 executionId: string
129 ) => Effect.Effect<Success["Type"], Error["Type"], R>
130 ) => Layer.Layer<
131 never,
132 never,
133 | WorkflowEngine
134 | Exclude<
135 R,
136 WorkflowEngine | WorkflowInstance | Execution<Tag> | Scope.Scope
137 >
138 | Payload["DecodingServices"]
139 | Payload["EncodingServices"]
140 | Success["DecodingServices"]
141 | Success["EncodingServices"]
142 | Error["DecodingServices"]
143 | Error["EncodingServices"]
144 >
145
146 /**
147 * For the given payload, compute the deterministic execution ID.
148 */
149 readonly executionId: (
150 payload: Payload["~type.make.in"]
151 ) => Effect.Effect<string>
152
153 /**
154 * Add compensation logic to an effect inside a Workflow.
155 *
156 * **Details**
157 *
158 * The compensation finalizer is called if the entire workflow fails, allowing you to perform cleanup or other actions based on the success value and the cause of the workflow failure.
159 *
160 * **Gotchas**
161 *
162 * Compensation finalizers are only registered for top-level effects in the workflow and do not work for nested activities.
163 */
164 readonly withCompensation: {
165 <A, R2>(
166 compensation: (
167 value: A,
168 cause: Cause.Cause<Error["Type"]>
169 ) => Effect.Effect<void, never, R2>
170 ): <E, R>(
171 effect: Effect.Effect<A, E, R>
172 ) => Effect.Effect<
173 A,
174 E,
175 R | R2 | WorkflowInstance | Execution<Tag> | Scope.Scope
176 >
177 <A, E, R, R2>(
178 effect: Effect.Effect<A, E, R>,
179 compensation: (
180 value: A,
181 cause: Cause.Cause<Error["Type"]>
182 ) => Effect.Effect<void, never, R2>
183 ): Effect.Effect<
184 A,
185 E,
186 R | R2 | WorkflowInstance | Execution<Tag> | Scope.Scope
187 >
188 }
189}
190
191/**
192 * Schema constraint for workflow payload schemas that expose struct fields.
193 *
194 * @stability unstable
195 * @category schemas
196 * @since 4.0.0
197 */
198export interface AnyStructSchema extends Schema.Top {
199 readonly fields: Schema.Struct.Fields
200}
201
202/**
203 * Type-level marker for services associated with a specific workflow
204 * execution tag.
205 *
206 * @stability unstable
207 * @category utility types
208 * @since 4.0.0
209 */
210export interface Execution<Tag extends string> {
211 readonly _: unique symbol
212 readonly _tag: Tag
213}
214
215/**
216 * Type-erased workflow shape for APIs that operate on workflows without
217 * preserving their specific payload, success, or error types.
218 *
219 * @stability unstable
220 * @category models
221 * @since 4.0.0
222 */
223export interface Any {
224 new(_: never): {}
225
226 readonly [TypeId]: typeof TypeId
227 readonly _tag: string
228 readonly executionId: (payload: any) => Effect.Effect<string>
229 readonly payloadSchema: AnyStructSchema
230 readonly successSchema: Schema.Top
231 readonly errorSchema: Schema.Top
232 readonly annotations: Context.Context<never>
233 readonly idempotencyKey: (payload: any) => string
234 readonly suspendedRetrySchedule?: Schedule.Schedule<any, unknown> | undefined
235}
236
237/**
238 * Type-erased workflow shape that also exposes executable operations needed by
239 * workflow proxy and engine helpers.
240 *
241 * @stability unstable
242 * @category models
243 * @since 4.0.0
244 */
245export interface AnyWithProps extends Any {
246 readonly payloadSchema: AnyStructSchema
247 readonly successSchema: Schema.Top
248 readonly errorSchema: Schema.Top
249 readonly execute: (
250 payload: any,
251 options?: { readonly discard?: boolean }
252 ) => Effect.Effect<any, any, any>
253 readonly resume: (
254 executionId: string
255 ) => Effect.Effect<void, never, WorkflowEngine>
256}
257
258/**
259 * Extracts the payload schema from a `Workflow`.
260 *
261 * @stability unstable
262 * @category utility types
263 * @since 4.0.0
264 */
265export type PayloadSchema<W> = W extends Workflow<
266 infer _Name,
267 infer _Payload,
268 infer _Success,
269 infer _Error
270> ? _Payload
271 : never
272
273/**
274 * Computes the schema services required by clients that execute or poll
275 * workflows.
276 *
277 * @stability unstable
278 * @category utility types
279 * @since 4.0.0
280 */
281export type RequirementsClient<Workflows extends Any> = Workflows extends Workflow<
282 infer _Name,
283 infer _Payload,
284 infer _Success,
285 infer _Error
286> ?
287 | _Payload["EncodingServices"]
288 | _Success["DecodingServices"]
289 | _Error["DecodingServices"]
290 : never
291
292/**
293 * Computes the schema services required by handlers that decode workflow
294 * payloads and encode workflow results.
295 *
296 * @stability unstable
297 * @category utility types
298 * @since 4.0.0
299 */
300export type RequirementsHandler<Workflows extends Any> = Workflows extends Workflow<
301 infer _Name,
302 infer _Payload,
303 infer _Success,
304 infer _Error
305> ?
306 | _Payload["DecodingServices"]
307 | _Payload["EncodingServices"]
308 | _Success["DecodingServices"]
309 | _Success["EncodingServices"]
310 | _Error["DecodingServices"]
311 | _Error["EncodingServices"]
312 : never
313
314const EngineTag = Context.Service<WorkflowEngine, WorkflowEngine["Service"]>(
315 "effect/workflow/WorkflowEngine" satisfies typeof WorkflowEngine.key
316)
317
318const InstanceTag = Context.Service<
319 WorkflowInstance,
320 WorkflowInstance["Service"]
321>(
322 "effect/workflow/WorkflowEngine/WorkflowInstance" satisfies typeof WorkflowInstance.key
323)
324
325const makeExecutionIdFromPayload = (self: AnyWithProps, payload: unknown) =>
326 makeHashDigest(`${self._tag.length}:${self._tag}:${self.idempotencyKey(payload)}`)
327
328const Proto = {
329 [TypeId]: TypeId,
330 annotate(this: AnyWithProps, tag: Context.Key<any, any>, value: any) {
331 return makeProto({
332 _tag: this._tag,
333 payloadSchema: this.payloadSchema,
334 successSchema: this.successSchema,
335 errorSchema: this.errorSchema,
336 annotations: Context.add(this.annotations, tag, value),
337 idempotencyKey: this.idempotencyKey,
338 suspendedRetrySchedule: this.suspendedRetrySchedule
339 })
340 },
341 annotateMerge(this: AnyWithProps, context: Context.Context<any>) {
342 return makeProto({
343 _tag: this._tag,
344 payloadSchema: this.payloadSchema,
345 successSchema: this.successSchema,
346 errorSchema: this.errorSchema,
347 annotations: Context.merge(this.annotations, context),
348 idempotencyKey: this.idempotencyKey,
349 suspendedRetrySchedule: this.suspendedRetrySchedule
350 })
351 },
352 execute<const Discard extends boolean = false>(
353 this: AnyWithProps,
354 fields: any,
355 opts?: { readonly discard?: Discard } | undefined
356 ) {
357 return Effect.suspend(() => {
358 const payload = this.payloadSchema.make(fields)
359 const run = Effect.flatMap(
360 EngineTag,
361 (engine) =>
362 Effect.flatMap(makeExecutionIdFromPayload(this, payload), (executionId) =>
363 Effect.andThen(
364 Effect.annotateCurrentSpan({ executionId }),
365 engine.execute(this as any, {
366 executionId,
367 payload,
368 discard: opts?.discard,
369 suspendedRetrySchedule: this.suspendedRetrySchedule
370 })
371 ))
372 )
373 // Register with the parent before the execution id is computed, so a
374 // sibling that suspends first waits for this child to be dispatched.
375 return opts?.discard ? run : withPendingActivity(run)
376 }).pipe(
377 Effect.withSpan(
378 `${this._tag}.execute`,
379 {},
380 { captureStackTrace: false }
381 )
382 ) as any
383 },
384 poll(this: Workflow<string, AnyStructSchema, Schema.Top, Schema.Top>, executionId: string) {
385 return Effect.flatMap(EngineTag, (engine) => engine.poll(this, executionId)).pipe(
386 Effect.withSpan(`${this._tag}.poll`, { attributes: { executionId } }, { captureStackTrace: false })
387 )
388 },
389 interrupt(this: AnyWithProps, executionId: string) {
390 return Effect.flatMap(EngineTag, (engine) => engine.interrupt(this, executionId)).pipe(
391 Effect.withSpan(`${this._tag}.interrupt`, { attributes: { executionId } }, { captureStackTrace: false })
392 )
393 },
394 resume(this: Workflow<string, AnyStructSchema, Schema.Top, Schema.Top>, executionId: string) {
395 return Effect.flatMap(EngineTag, (engine) => engine.resume(this, executionId)).pipe(
396 Effect.withSpan(`${this._tag}.resume`, { attributes: { executionId } }, { captureStackTrace: false })
397 )
398 },
399 toLayer(this: Workflow<string, AnyStructSchema, Schema.Top, Schema.Top>, execute: any) {
400 return Layer.effectDiscard(
401 Effect.flatMap(EngineTag, (engine) => engine.register(this, execute))
402 )
403 },
404 executionId(this: AnyWithProps, payload: any) {
405 return Effect.flatMap(
406 Effect.orDie(this.payloadSchema.makeEffect(payload)),
407 (payload) => makeExecutionIdFromPayload(this, payload)
408 )
409 },
410 withCompensation: ((...args: ReadonlyArray<any>) => (withCompensation as any)(...args))
411}
412
413const makeProto = <
414 const Tag extends string,
415 Payload extends AnyStructSchema,
416 Success extends Schema.Top,
417 Error extends Schema.Top
418>(options: {
419 readonly _tag: Tag
420 readonly payloadSchema: Payload
421 readonly successSchema: Success
422 readonly errorSchema: Error
423 readonly annotations: Context.Context<never>
424 readonly idempotencyKey: (payload: Payload["Type"]) => string
425 readonly suspendedRetrySchedule?: Schedule.Schedule<any, unknown> | undefined
426}): Workflow<Tag, Payload, Success, Error> => {
427 function Workflow() {}
428 Object.setPrototypeOf(Workflow, Proto)
429 Object.assign(Workflow, options)
430 return Workflow as any
431}
432
433/**
434 * Creates a durable workflow definition with schemas, annotations, and
435 * deterministic execution IDs derived from the workflow tag and idempotency
436 * key.
437 *
438 * @stability unstable
439 * @category constructors
440 * @since 4.0.0
441 */
442export const make = <
443 const Tag extends string,
444 Payload extends Schema.Struct.Fields | AnyStructSchema,
445 Success extends Schema.Top = Schema.Void,
446 Error extends Schema.Top = Schema.Never
447>(tag: Tag, options: {
448 readonly payload: Payload
449 readonly idempotencyKey: (
450 payload: Payload extends Schema.Struct.Fields ? Schema.Struct.Type<Payload>
451 : Payload["Type"]
452 ) => string
453 readonly success?: Success
454 readonly error?: Error
455 readonly suspendedRetrySchedule?: Schedule.Schedule<any, unknown> | undefined
456 readonly annotations?: Context.Context<never>
457}): Workflow<
458 Tag,
459 Payload extends Schema.Struct.Fields ? Schema.Struct<Payload> : Payload,
460 Success,
461 Error
462> =>
463 makeProto<Tag, Payload extends Schema.Struct.Fields ? Schema.Struct<Payload> : Payload, Success, Error>({
464 _tag: tag,
465 payloadSchema: (Schema.isSchema(options.payload)
466 ? options.payload
467 : Schema.Struct(options.payload as any)) as Payload extends Schema.Struct.Fields ? Schema.Struct<Payload>
468 : Payload,
469 successSchema: options.success ?? (Schema.Void as any),
470 errorSchema: options.error ?? (Schema.Never as any),
471 annotations: options.annotations ?? Context.empty(),
472 idempotencyKey: options.idempotencyKey as any,
473 suspendedRetrySchedule: options.suspendedRetrySchedule
474 })
475
476const ResultTypeId = "~effect/workflow/Workflow/Result"
477
478/**
479 * Returns `true` when a value is a workflow `Result`.
480 *
481 * @stability unstable
482 * @category guards
483 * @since 4.0.0
484 */
485export const isResult = <A = unknown, E = unknown>(
486 u: unknown
487): u is Result<A, E> => Predicate.hasProperty(u, ResultTypeId)
488
489/**
490 * Result of a workflow execution, either a completed exit or a suspended
491 * workflow state.
492 *
493 * @stability unstable
494 * @category models
495 * @since 4.0.0
496 */
497export type Result<A, E> = Complete<A, E> | Suspended
498
499/**
500 * Encoded representation of a workflow `Result`.
501 *
502 * @stability unstable
503 * @category models
504 * @since 4.0.0
505 */
506export type ResultEncoded<A, E> =
507 | CompleteEncoded<A, E>
508 | typeof Suspended.Encoded
509
510/**
511 * Encoded representation of a completed workflow result containing an encoded
512 * `Exit`.
513 *
514 * @stability unstable
515 * @category models
516 * @since 4.0.0
517 */
518export interface CompleteEncoded<A, E> {
519 readonly _tag: "Complete"
520 readonly exit: ExitEncoded<A, E>
521}
522
523/**
524 * Schema constructor for `Complete` workflow results using the supplied
525 * success and error schemas.
526 *
527 * @stability unstable
528 * @category schemas
529 * @since 4.0.0
530 */
531export interface CompleteSchema<
532 Success extends Schema.Constraint,
533 Error extends Schema.Constraint
534> extends
535 Schema.declareConstructor<
536 Complete<Success["Type"], Error["Type"]>,
537 Complete<Success["Encoded"], Error["Encoded"]>,
538 readonly [Schema.Exit<Success, Error, Schema.Defect>]
539 >
540{
541 readonly success: Success
542 readonly error: Error
543}
544
545/**
546 * Represents a completed workflow execution with its success or failure `Exit`.
547 *
548 * @stability unstable
549 * @category models
550 * @since 4.0.0
551 */
552export class Complete<A, E> extends Data.TaggedClass("Complete")<{
553 readonly exit: Exit.Exit<A, E>
554}> {
555 /**
556 * Marks this value as a workflow result for runtime guards.
557 *
558 * @stability unstable
559 * @since 4.0.0
560 */
561 readonly [ResultTypeId] = ResultTypeId
562
563 /**
564 * Builds the schema for completed workflow results from success and error schemas.
565 *
566 * @stability unstable
567 * @since 4.0.0
568 */
569 static Schema<Success extends Schema.Constraint, Error extends Schema.Constraint>(options: {
570 readonly success: Success
571 readonly error: Error
572 }): CompleteSchema<Success, Error> {
573 // TODO: extract to a helper function
574 const schema = Schema.declareConstructor<
575 Complete<Success["Type"], Error["Type"]>,
576 Complete<Success["Encoded"], Error["Encoded"]>
577 >()(
578 [Schema.Exit(options.success, options.error, Schema.Defect())],
579 ([exit]) => (input, ast, options) => {
580 if (!(isResult(input) && input._tag === "Complete")) {
581 return Effect.fail(new SchemaIssue.InvalidType(ast, input, options))
582 }
583 return Effect.mapBothEager(
584 SchemaParser.decodeEffect(exit)(input.exit, options),
585 {
586 onSuccess: (exit) => new Complete({ exit }),
587 onFailure: (issue) => SchemaIssue.makeCompositeAtKey(ast, "exit", issue, input, options)
588 }
589 )
590 },
591 {
592 expected: "Workflow.Complete",
593 toCodecJson: ([exit]) =>
594 Schema.link<Complete<Success["Encoded"], Error["Encoded"]>>()(
595 Schema.Struct({
596 _tag: Schema.tag("Complete"),
597 exit
598 }),
599 Tranformation.transform({
600 decode: (encoded) => new Complete({ exit: encoded.exit }),
601 encode: (result) => (({
602 _tag: "Complete",
603 exit: result.exit
604 }) as const)
605 })
606 )
607 }
608 )
609 return Schema.make(schema.ast, {
610 success: options.success,
611 error: options.error
612 })
613 }
614}
615
616/**
617 * Represents a suspended workflow execution, optionally carrying the cause that
618 * triggered suspension.
619 *
620 * @stability unstable
621 * @category schemas
622 * @since 4.0.0
623 */
624export class Suspended extends Schema.Class<Suspended>(
625 "effect/workflow/Workflow/Suspended"
626)({
627 _tag: Schema.tag("Suspended"),
628 cause: Schema.optional(Schema.Cause(Schema.Never, Schema.Defect()))
629}) {
630 /**
631 * Marks this value as a workflow result for runtime guards.
632 *
633 * @stability unstable
634 * @since 4.0.0
635 */
636 readonly [ResultTypeId] = ResultTypeId
637}
638
639/**
640 * Creates a schema for workflow results using the supplied success and error
641 * schemas.
642 *
643 * @stability unstable
644 * @category schemas
645 * @since 4.0.0
646 */
647export const Result = <
648 Success extends Schema.Constraint,
649 Error extends Schema.Constraint
650>(options: {
651 readonly success: Success
652 readonly error: Error
653}) => Schema.Union([Complete.Schema(options), Suspended])
654
655const AnyOrVoid = Schema.Union([Schema.Any, Schema.Void])
656
657/**
658 * Schema for encoded workflow results with generic success and error payloads.
659 *
660 * @stability unstable
661 * @category schemas
662 * @since 4.0.0
663 */
664export const ResultEncoded: Schema.Codec<ResultEncoded<any, any>> = Schema.toEncoded(
665 Schema.toCodecJson(
666 Result({
667 success: AnyOrVoid,
668 error: AnyOrVoid
669 })
670 )
671) as any
672
673/**
674 * Runs an effect as a workflow execution and converts its outcome into a
675 * `Result`, handling suspension, defect capture, interruption, and workflow
676 * scope finalization.
677 *
678 * @stability unstable
679 * @category converting
680 * @since 4.0.0
681 */
682export const intoResult = <A, E, R>(
683 effect: Effect.Effect<A, E, R>
684): Effect.Effect<
685 Result<A, E>,
686 never,
687 Exclude<R, Scope.Scope> | WorkflowInstance
688> =>
689 Effect.contextWith((context: Context.Context<WorkflowInstance>) => {
690 const instance = Context.get(context, InstanceTag)
691 const captureDefects = Context.get(instance.workflow.annotations, CaptureDefects)
692 const suspendOnFailure = Context.get(instance.workflow.annotations, SuspendOnFailure)
693 return effect.pipe(
694 // so we can use external interruption to suspend the workflow
695 Effect.forkChild({ startImmediately: true }),
696 Effect.flatMap((fiber) => Effect.onInterrupt(Fiber.join(fiber), () => Fiber.interrupt(fiber))),
697 Effect.interruptible,
698 suspendOnFailure
699 ? Effect.catchCause((cause) => {
700 if (instance.abandoned) {
701 return Effect.failCause(cause)
702 }
703 instance.suspended = true
704 if (!Cause.hasInterruptsOnly(cause)) {
705 instance.cause = Cause.die(Cause.squash(cause))
706 }
707 return Effect.interrupt
708 })
709 : identity,
710 Effect.scoped,
711 Effect.matchCauseEffect({
712 onSuccess: (value) => Effect.succeed(new Complete({ exit: Exit.succeed(value) })),
713 onFailure: (cause): Effect.Effect<Result<A, E>> => {
714 const [interrupts, reasons] = Arr.partition(
715 cause.reasons,
716 Filter.fromPredicate(Cause.isInterruptReason)
717 )
718 const hasInterruptsOnly = interrupts.length === cause.reasons.length
719 const filtered = reasons.length === 0 ? cause : Cause.fromReasons(reasons)
720 const abandoned = instance.abandoned && !instance.interrupted && hasInterruptsOnly
721 instance.abandoned = abandoned
722 if (instance.suspended && hasInterruptsOnly && !abandoned) {
723 return Effect.succeed(new Suspended({ cause: instance.cause }))
724 }
725 if (!instance.interrupted && hasInterruptsOnly) {
726 // External interrupts exit without a result. When the cluster has
727 // marked this attempt abandoned, owner-local resources are
728 // released without running durable workflow finalizers so the run
729 // can replay elsewhere.
730 return Effect.failCause(filtered as Cause.Cause<never>)
731 }
732 return !captureDefects && Cause.hasDies(cause)
733 ? Effect.failCause(filtered as Cause.Cause<never>)
734 : Effect.succeed(new Complete({ exit: Exit.failCause(filtered) }))
735 }
736 }),
737 (eff) =>
738 Effect.onExitPrimitive(eff, (exit) => {
739 if (Exit.isFailure(exit)) {
740 return Scope.close(instance.scope, exit)
741 } else if (exit.value._tag === "Complete") {
742 return Scope.close(instance.scope, exit.value.exit)
743 }
744 return Effect.void
745 }, true),
746 Effect.uninterruptible
747 )
748 })
749
750/**
751 * Wraps an activity-like effect so workflow suspension waits for currently
752 * running activities to finish or suspend.
753 *
754 * @stability unstable
755 * @category resource management
756 * @since 4.0.0
757 */
758export const wrapActivityResult = <A, E, R>(
759 effect: Effect.Effect<A, E, R>,
760 isSuspend: (value: A) => boolean
761): Effect.Effect<A, E, R | WorkflowInstance> =>
762 Effect.contextWith((context: Context.Context<WorkflowInstance>) => {
763 const instance = Context.get(context, InstanceTag)
764 return Effect.acquireUseRelease(
765 Effect.sync(() => adoptActivityUnsafe(context, instance) ?? registerActivityUnsafe(instance)),
766 () => effect,
767 (registration, exit) => {
768 releaseActivityUnsafe(registration)
769 const isSuspended = Exit.isSuccess(exit) && isSuspend(exit.value)
770 if (
771 Exit.isSuccess(exit) &&
772 isResult(exit.value) &&
773 exit.value._tag === "Suspended" &&
774 exit.value.cause
775 ) {
776 instance.cause = instance.cause
777 ? Cause.combine(instance.cause, exit.value.cause)
778 : exit.value.cause
779 }
780 return isSuspended && instance.activityState.count > 0
781 ? waitForZero(instance)
782 : Effect.void
783 }
784 )
785 })
786
787interface ActivityRegistration {
788 readonly instance: WorkflowInstance["Service"]
789 state: "pending" | "adopted" | "released"
790}
791
792const PendingActivityRegistration = Context.Service<ActivityRegistration>(
793 "effect/workflow/Workflow/PendingActivityRegistration"
794)
795
796const registerActivityUnsafe = (instance: WorkflowInstance["Service"]): ActivityRegistration => {
797 const state = instance.activityState
798 if (state.count === 0) state.latch.closeUnsafe()
799 state.count++
800 return { instance, state: "pending" }
801}
802
803const adoptActivityUnsafe = (
804 context: Context.Context<never>,
805 instance: WorkflowInstance["Service"]
806): ActivityRegistration | undefined => {
807 const pending = Context.getOrUndefined(context, PendingActivityRegistration)
808 if (!pending || pending.instance !== instance) {
809 return undefined
810 }
811 if (pending.state !== "pending") {
812 throw new Error("Workflow.wrapActivityResult: pending child registration has already been consumed")
813 }
814 pending.state = "adopted"
815 return pending
816}
817
818const releaseActivityUnsafe = (registration: ActivityRegistration) => {
819 if (registration.state === "released") return
820 registration.state = "released"
821 const state = registration.instance.activityState
822 state.count--
823 if (state.count === 0) state.latch.openUnsafe()
824}
825
826/**
827 * Registers a pending child workflow execution with the current workflow
828 * instance before any asynchronous work runs. `wrapActivityResult` adopts the
829 * registration once the execution is dispatched.
830 */
831const withPendingActivity = <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
832 Effect.contextWith((context: Context.Context<never>) => {
833 const instance = Context.getOrUndefined(context, InstanceTag)
834 if (!instance) return effect
835 return Effect.acquireUseRelease(
836 Effect.sync(() => registerActivityUnsafe(instance)),
837 (registration) => Effect.provideService(effect, PendingActivityRegistration, registration),
838 (registration) => Effect.sync(() => releaseActivityUnsafe(registration))
839 )
840 })
841
842const waitForZero = Effect.fnUntraced(function*(instance: WorkflowInstance["Service"]) {
843 const state = instance.activityState
844 while (true) {
845 if (state.count > 0) {
846 yield* state.latch.await
847 yield* Effect.yieldNow
848 continue
849 }
850 yield* Effect.yieldNow
851 if (state.count === 0) return
852 }
853})
854
855/**
856 * Accesses the workflow scope, which is only closed when the workflow execution fully completes.
857 *
858 * @stability unstable
859 * @category resource management
860 * @since 4.0.0
861 */
862export const scope: Effect.Effect<
863 Scope.Scope,
864 never,
865 WorkflowInstance
866> = Effect.map(
867 InstanceTag,
868 (instance) => instance.scope as Scope.Scope
869)
870
871/**
872 * Provides the workflow scope to the given effect, and closes the scope only when the workflow execution fully completes.
873 *
874 * @stability unstable
875 * @category resource management
876 * @since 4.0.0
877 */
878export const provideScope = <A, E, R>(
879 effect: Effect.Effect<A, E, R>
880): Effect.Effect<A, E, Exclude<R, Scope.Scope> | WorkflowInstance> =>
881 Effect.flatMap(scope, (scope) => Scope.provide(effect, scope))
882
883/**
884 * Adds an exit finalizer to the current workflow scope, preserving the
885 * services available when the finalizer is registered.
886 *
887 * **Details**
888 *
889 * Body-level `Effect.onExit` finalizers cannot observe a deposited workflow
890 * interrupt. Use this function for terminal-state work that must observe it.
891 * Finalizers are skipped when a cluster owner abandons a run attempt for
892 * replay; owner-local scope finalizers registered with {@link provideScope}
893 * still run.
894 *
895 * @stability unstable
896 * @category resource management
897 * @since 4.0.0
898 */
899export const addFinalizer: <R>(
900 f: (exit: Exit.Exit<unknown, unknown>) => Effect.Effect<void, never, R>
901) => Effect.Effect<
902 void,
903 never,
904 WorkflowInstance | R
905> = Effect.fnUntraced(function*<R>(
906 f: (exit: Exit.Exit<unknown, unknown>) => Effect.Effect<void, never, R>
907) {
908 const instance = yield* InstanceTag
909 const services = yield* Effect.context<R>()
910 yield* Scope.addFinalizerExit(
911 instance.scope,
912 (exit) => instance.abandoned ? Effect.void : Effect.provideContext(f(exit), services)
913 )
914})
915
916/**
917 * Adds compensation logic to an effect inside a Workflow.
918 *
919 * **When to use**
920 *
921 * Use when a top-level workflow step needs compensating cleanup if the overall
922 * workflow later fails after the step succeeds.
923 *
924 * **Details**
925 *
926 * The compensation finalizer is called if the entire workflow fails, allowing you to perform cleanup or other actions based on the success value and the cause of the workflow failure.
927 *
928 * **Gotchas**
929 *
930 * Compensation finalizers are only registered for top-level effects in the workflow and do not work for nested activities.
931 *
932 * @stability unstable
933 * @category compensation
934 * @since 4.0.0
935 */
936export const withCompensation: {
937 <A, R2>(
938 compensation: (value: A, cause: Cause.Cause<unknown>) => Effect.Effect<void, never, R2>
939 ): <E, R>(
940 effect: Effect.Effect<A, E, R>
941 ) => Effect.Effect<A, E, R | R2 | WorkflowInstance | Scope.Scope>
942 <A, E, R, R2>(
943 effect: Effect.Effect<A, E, R>,
944 compensation: (value: A, cause: Cause.Cause<unknown>) => Effect.Effect<void, never, R2>
945 ): Effect.Effect<A, E, R | R2 | WorkflowInstance | Scope.Scope>
946} = dual(2, <A, E, R, R2>(
947 effect: Effect.Effect<A, E, R>,
948 compensation: (value: A, cause: Cause.Cause<unknown>) => Effect.Effect<void, never, R2>
949): Effect.Effect<A, E, R | R2 | WorkflowInstance | Scope.Scope> =>
950 Effect.uninterruptibleMask((restore) =>
951 Effect.tap(
952 restore(effect),
953 (value) => addFinalizer((exit) => Exit.isSuccess(exit) ? Effect.void : compensation(value, exit.cause))
954 )
955 ))
956
957/**
958 * Marks a workflow instance as suspended and interrupts the current fiber to
959 * stop execution until it is resumed.
960 *
961 * @stability unstable
962 * @category interruption
963 * @since 4.0.0
964 */
965export const suspend = (instance: WorkflowInstance["Service"]): Effect.Effect<never> =>
966 Effect.interruptible(Effect.callback<never>(() => {
967 instance.suspended = true
968 const fiber = Fiber.getCurrent()!
969 fiber.interruptUnsafe(fiber.id)
970 }))
971
972/**
973 * Captures defects for a workflow and includes them in the result of the workflow or its activities.
974 *
975 * **Details**
976 *
977 * By default, this annotation is set to `true`, meaning defects are captured.
978 *
979 * @stability unstable
980 * @category services
981 * @since 4.0.0
982 */
983export const CaptureDefects = Context.Reference<boolean>(
984 "effect/workflow/Workflow/CaptureDefects",
985 {
986 defaultValue: constTrue
987 }
988)
989
990/**
991 * Marks a workflow to suspend when it encounters any error.
992 *
993 * **Details**
994 *
995 * The suspended execution can later be resumed with the workflow's `resume` method, for example `MyWorkflow.resume(executionId)`.
996 *
997 * @stability unstable
998 * @category services
999 * @since 4.0.0
1000 */
1001export const SuspendOnFailure = Context.Reference<boolean>(
1002 "effect/workflow/Workflow/SuspendOnFailure",
1003 {
1004 defaultValue: constFalse
1005 }
1006)