1/**
2 * Defines workflow engine services and an in-memory implementation.
3 *
4 * `WorkflowEngine` registers workflow handlers, runs executions, polls results,
5 * resumes suspended runs, executes activities, stores durable deferred results,
6 * and schedules durable clocks. `WorkflowInstance` holds the runtime state for
7 * one workflow run. The in-memory layer is useful for tests and local
8 * development.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Cause from "../Cause.ts"
14import * as Context from "../Context.ts"
15import * as Effect from "../Effect.ts"
16import * as Exit from "../Exit.ts"
17import * as Fiber from "../Fiber.ts"
18import * as FiberMap from "../FiberMap.ts"
19import * as Latch from "../Latch.ts"
20import * as Layer from "../Layer.ts"
21import * as Option from "../Option.ts"
22import * as Schedule from "../Schedule.ts"
23import * as Schema from "../Schema.ts"
24import * as Scope from "../Scope.ts"
25import type * as Activity from "./Activity.ts"
26import type { DurableClock } from "./DurableClock.ts"
27import type * as DurableDeferred from "./DurableDeferred.ts"
28import * as Workflow from "./Workflow.ts"
29
30/**
31 * Service that represents workflow runtimes, responsible for registering and
32 * executing workflows and coordinating activities, durable deferreds,
33 * interrupts, resumes, and clocks.
34 *
35 * @stability unstable
36 * @category services
37 * @since 4.0.0
38 */
39export class WorkflowEngine extends Context.Service<
40 WorkflowEngine,
41 {
42 /**
43 * Register a workflow with the engine.
44 */
45 readonly register: <
46 Name extends string,
47 Payload extends Workflow.AnyStructSchema,
48 Success extends Schema.Top,
49 Error extends Schema.Top,
50 R
51 >(
52 workflow: Workflow.Workflow<Name, Payload, Success, Error>,
53 execute: (
54 payload: Payload["Type"],
55 executionId: string
56 ) => Effect.Effect<Success["Type"], Error["Type"], R>
57 ) => Effect.Effect<
58 void,
59 never,
60 | Scope.Scope
61 | Exclude<
62 R,
63 | WorkflowEngine
64 | WorkflowInstance
65 | Workflow.Execution<Name>
66 | Scope.Scope
67 >
68 | Payload["DecodingServices"]
69 | Payload["EncodingServices"]
70 | Success["DecodingServices"]
71 | Success["EncodingServices"]
72 | Error["DecodingServices"]
73 | Error["EncodingServices"]
74 >
75
76 /**
77 * Execute a registered workflow.
78 */
79 readonly execute: <
80 Name extends string,
81 Payload extends Workflow.AnyStructSchema,
82 Success extends Schema.Top,
83 Error extends Schema.Top,
84 const Discard extends boolean = false
85 >(
86 workflow: Workflow.Workflow<Name, Payload, Success, Error>,
87 options: {
88 readonly executionId: string
89 readonly payload: Payload["Type"]
90 readonly discard?: Discard | undefined
91 readonly suspendedRetrySchedule?:
92 | Schedule.Schedule<any, unknown>
93 | undefined
94 }
95 ) => Effect.Effect<
96 Discard extends true ? string : Success["Type"],
97 Error["Type"],
98 | Payload["EncodingServices"]
99 | Success["DecodingServices"]
100 | Error["DecodingServices"]
101 >
102
103 /**
104 * Poll the current status of a registered workflow execution.
105 */
106 readonly poll: <
107 Name extends string,
108 Payload extends Workflow.AnyStructSchema,
109 Success extends Schema.Top,
110 Error extends Schema.Top
111 >(
112 workflow: Workflow.Workflow<Name, Payload, Success, Error>,
113 executionId: string
114 ) => Effect.Effect<
115 Option.Option<Workflow.Result<Success["Type"], Error["Type"]>>,
116 never,
117 Success["DecodingServices"] | Error["DecodingServices"]
118 >
119
120 /**
121 * Interrupt a registered workflow.
122 */
123 readonly interrupt: (
124 workflow: Workflow.Any,
125 executionId: string
126 ) => Effect.Effect<void>
127
128 /**
129 * Interrupts a registered workflow unsafely, potentially ignoring
130 * compensation finalizers and orphaning child workflows.
131 */
132 readonly interruptUnsafe: (
133 workflow: Workflow.Any,
134 executionId: string
135 ) => Effect.Effect<void>
136
137 /**
138 * Resume a registered workflow.
139 */
140 readonly resume: (
141 workflow: Workflow.Any,
142 executionId: string
143 ) => Effect.Effect<void>
144
145 /**
146 * Execute an activity from a workflow.
147 */
148 readonly activityExecute: <
149 Success extends Schema.Constraint,
150 Error extends Schema.Constraint,
151 R
152 >(
153 activity: Activity.Activity<Success, Error, R>,
154 attempt: number
155 ) => Effect.Effect<
156 Workflow.Result<Success["Type"], Error["Type"]>,
157 never,
158 | Success["DecodingServices"]
159 | Error["DecodingServices"]
160 | R
161 | WorkflowInstance
162 >
163
164 /**
165 * Try to retrieve the result of an DurableDeferred
166 */
167 readonly deferredResult: <
168 Success extends Schema.Constraint,
169 Error extends Schema.Constraint
170 >(
171 deferred: DurableDeferred.DurableDeferred<Success, Error>
172 ) => Effect.Effect<
173 Option.Option<Exit.Exit<Success["Type"], Error["Type"]>>,
174 never,
175 WorkflowInstance
176 >
177
178 /**
179 * Set the result of a DurableDeferred, and then resume any waiting
180 * workflows.
181 */
182 readonly deferredDone: <
183 Success extends Schema.Constraint,
184 Error extends Schema.Constraint
185 >(
186 deferred: DurableDeferred.DurableDeferred<Success, Error>,
187 options: {
188 readonly workflowName: string
189 readonly executionId: string
190 readonly deferredName: string
191 readonly exit: Exit.Exit<Success["Type"], Error["Type"]>
192 }
193 ) => Effect.Effect<
194 void,
195 never,
196 Success["EncodingServices"] | Error["EncodingServices"]
197 >
198
199 /**
200 * Schedule a wake up for a DurableClock
201 */
202 readonly scheduleClock: (
203 workflow: Workflow.Any,
204 options: {
205 readonly executionId: string
206 readonly clock: DurableClock
207 }
208 ) => Effect.Effect<void>
209 }
210>()("effect/workflow/WorkflowEngine") {}
211
212/**
213 * Service that contains workflow runtime state for one execution.
214 *
215 * **When to use**
216 *
217 * Use to read or update workflow execution, suspension, interruption,
218 * lifetime, failure, and activity coordination state inside workflow engine
219 * internals.
220 *
221 * **Details**
222 *
223 * The service stores the execution ID, workflow definition, long-lived scope,
224 * suspension and interruption flags, the stored failure cause, and activity
225 * coordination state for a single workflow run.
226 *
227 * @stability unstable
228 * @category services
229 * @since 4.0.0
230 */
231export class WorkflowInstance extends Context.Service<
232 WorkflowInstance,
233 {
234 /**
235 * The workflow execution ID.
236 */
237 readonly executionId: string
238
239 /**
240 * The workflow definition.
241 */
242 readonly workflow: Workflow.Any
243
244 /**
245 * A scope that represents the lifetime of the workflow.
246 *
247 * It is only closed when the workflow is completed.
248 */
249 readonly scope: Scope.Closeable
250
251 /**
252 * Whether the workflow has requested to be suspended.
253 */
254 suspended: boolean
255
256 /**
257 * Whether the workflow has requested to be interrupted.
258 */
259 interrupted: boolean
260
261 /**
262 * Whether the current workflow run has been abandoned for replay. When
263 * `true`, callbacks registered with `Workflow.addFinalizer` are skipped as
264 * the owner-local scope closes.
265 */
266 abandoned: boolean
267
268 /**
269 * When SuspendOnFailure is triggered, the cause of the failure is stored
270 * here.
271 */
272 cause: Cause.Cause<never> | undefined
273
274 /** Deferred names this run parked on; their completions preempt the run. */
275 readonly awaitedDeferreds: Set<string>
276
277 /** Deferred names this run completed itself; their completions need not wake it. */
278 readonly completedDeferreds: Set<string>
279
280 readonly activityState: {
281 count: number
282 readonly latch: Latch.Latch
283 }
284 }
285>()("effect/workflow/WorkflowEngine/WorkflowInstance") {
286 static initial(
287 workflow: Workflow.Any,
288 executionId: string,
289 scope = Scope.makeUnsafe()
290 ): WorkflowInstance["Service"] {
291 return WorkflowInstance.of({
292 executionId,
293 workflow,
294 scope,
295 suspended: false,
296 interrupted: false,
297 abandoned: false,
298 cause: undefined,
299 awaitedDeferreds: new Set(),
300 completedDeferreds: new Set(),
301 activityState: {
302 count: 0,
303 latch: Latch.makeUnsafe()
304 }
305 })
306 }
307}
308
309/**
310 * Tracks live workflow runs so deferred completions can wake them.
311 *
312 * @stability unstable
313 * @category models
314 * @since 4.0.0
315 */
316export interface DeferredState {
317 /** Tracks and provides a run until it completes or suspends. */
318 readonly trackRun: <A, E, R>(
319 instance: WorkflowInstance["Service"],
320 effect: Effect.Effect<A, E, R>
321 ) => Effect.Effect<A, E, Exclude<R, WorkflowInstance>>
322
323 /** Preempts a run parked on the completed deferred. */
324 readonly deferredDone: (
325 executionId: string,
326 name: string
327 ) => Effect.Effect<void>
328}
329
330/**
331 * Creates deferred state shared by workflow engines.
332 *
333 * @stability unstable
334 * @category constructors
335 * @since 4.0.0
336 */
337export const makeDeferredState = (): DeferredState => {
338 const running = new Map<string, {
339 readonly instance: WorkflowInstance["Service"]
340 readonly fiber: Fiber.Fiber<unknown, unknown>
341 }>()
342 return {
343 trackRun: (instance, effect) =>
344 Effect.withFiber((fiber) => {
345 const run = { instance, fiber: fiber as Fiber.Fiber<unknown, unknown> }
346 running.set(instance.executionId, run)
347 return Effect.ensuring(
348 Effect.provideService(effect, WorkflowInstance, instance),
349 Effect.sync(() => {
350 if (running.get(instance.executionId) !== run) return
351 running.delete(instance.executionId)
352 })
353 )
354 }),
355 deferredDone: (executionId, name) =>
356 Effect.withFiber((current) => {
357 const run = running.get(executionId)
358 if (!run) return Effect.void
359 if (
360 run.fiber === current ||
361 run.fiber.pollUnsafe() ||
362 !run.instance.awaitedDeferreds.has(name)
363 ) {
364 return Effect.void
365 }
366 // The engine stores the completion before waking this run. Replay
367 // observes it after the interrupted run reports suspension.
368 run.instance.suspended = true
369 return Fiber.interrupt(run.fiber)
370 })
371 }
372}
373
374/**
375 * Low-level workflow engine contract that works with encoded payloads and
376 * results before `makeUnsafe` adds typed schema decoding and encoding.
377 *
378 * @stability unstable
379 * @category services
380 * @since 4.0.0
381 */
382export interface Encoded {
383 readonly register: (
384 workflow: Workflow.Any,
385 execute: (
386 payload: object,
387 executionId: string
388 ) => Effect.Effect<unknown, unknown, WorkflowInstance | WorkflowEngine>
389 ) => Effect.Effect<void, never, Scope.Scope>
390 readonly execute: <const Discard extends boolean>(
391 workflow: Workflow.Any,
392 options: {
393 readonly executionId: string
394 readonly payload: object
395 readonly discard: Discard
396 readonly parent?: WorkflowInstance["Service"] | undefined
397 }
398 ) => Effect.Effect<
399 Discard extends true ? void : Workflow.Result<unknown, unknown>
400 >
401 readonly poll: (
402 workflow: Workflow.Any,
403 executionId: string
404 ) => Effect.Effect<Option.Option<Workflow.Result<unknown, unknown>>>
405 readonly interrupt: (
406 workflow: Workflow.Any,
407 executionId: string
408 ) => Effect.Effect<void>
409 readonly interruptUnsafe: (
410 workflow: Workflow.Any,
411 executionId: string
412 ) => Effect.Effect<void>
413 readonly resume: (
414 workflow: Workflow.Any,
415 executionId: string
416 ) => Effect.Effect<void>
417 readonly activityExecute: (
418 activity: Activity.Any,
419 attempt: number
420 ) => Effect.Effect<
421 Workflow.Result<unknown, unknown>,
422 never,
423 WorkflowInstance
424 >
425 readonly deferredResult: (
426 deferred: DurableDeferred.Any
427 ) => Effect.Effect<
428 Option.Option<Exit.Exit<unknown, unknown>>,
429 never,
430 WorkflowInstance
431 >
432 readonly deferredDone: (options: {
433 readonly workflowName: string
434 readonly executionId: string
435 readonly deferredName: string
436 readonly exit: Exit.Exit<unknown, unknown>
437 }) => Effect.Effect<void>
438 readonly scheduleClock: (
439 workflow: Workflow.Any,
440 options: {
441 readonly executionId: string
442 readonly clock: DurableClock
443 }
444 ) => Effect.Effect<void>
445}
446
447/**
448 * Builds a typed `WorkflowEngine` service from a low-level encoded
449 * implementation.
450 *
451 * **When to use**
452 *
453 * Use when wiring a trusted low-level workflow engine implementation into the
454 * typed `WorkflowEngine` service.
455 *
456 * **Gotchas**
457 *
458 * The implementation must correctly persist, resume, and encode workflow state.
459 *
460 * @stability unstable
461 * @category constructors
462 * @since 4.0.0
463 */
464export const makeUnsafe = (options: Encoded): WorkflowEngine["Service"] =>
465 WorkflowEngine.of({
466 register: Effect.fnUntraced(function*(workflow, execute) {
467 const services = yield* Effect.context<WorkflowEngine>()
468 yield* options.register(workflow, (payload, executionId) =>
469 Effect.suspend(() =>
470 execute(payload, executionId)
471 ).pipe(
472 Effect.updateContext(
473 (input) => Context.merge(services, input) as Context.Context<any>
474 )
475 ))
476 }),
477 execute: Effect.fnUntraced(function*<
478 Name extends string,
479 Payload extends Workflow.AnyStructSchema,
480 Success extends Schema.Top,
481 Error extends Schema.Top,
482 const Discard extends boolean = false
483 >(
484 self: Workflow.Workflow<Name, Payload, Success, Error>,
485 opts: {
486 readonly executionId: string
487 readonly payload: Payload["Type"]
488 readonly discard?: Discard | undefined
489 readonly suspendedRetrySchedule?:
490 | Schedule.Schedule<any, unknown>
491 | undefined
492 }
493 ) {
494 const payload = opts.payload
495 const executionId = opts.executionId
496 const suspendedRetrySchedule = opts.suspendedRetrySchedule ?? defaultRetrySchedule
497 yield* Effect.annotateCurrentSpan({ executionId })
498 let result = Option.none<Workflow.Result<Success["Type"], Error["Type"]>>()
499
500 // link interruption with parent workflow
501 const parentInstance = yield* Effect.serviceOption(WorkflowInstance)
502 if (Option.isSome(parentInstance)) {
503 const instance = parentInstance.value
504 yield* Effect.addFinalizer(() => {
505 if (!instance.interrupted || (Option.isSome(result) && result.value._tag === "Complete")) {
506 return Effect.void
507 }
508 return options.interrupt(self, executionId)
509 })
510 }
511 const run = options.execute(self, {
512 executionId,
513 payload: payload as object,
514 discard: opts.discard ?? false,
515 parent: Option.getOrUndefined(parentInstance)
516 }) as Effect.Effect<Workflow.Result<Success["Type"], Error["Type"]>>
517
518 if (opts.discard) {
519 yield* run
520 return executionId
521 }
522
523 if (Option.isSome(parentInstance)) {
524 const wrapped = yield* Workflow.wrapActivityResult(
525 run,
526 (result) => result._tag === "Suspended"
527 )
528 result = Option.some(wrapped)
529 if (wrapped._tag === "Suspended") {
530 return yield* Workflow.suspend(parentInstance.value)
531 }
532 return yield* wrapped.exit
533 }
534
535 let sleep: Effect.Effect<any> | undefined
536 while (true) {
537 const wrapped = yield* run
538 result = Option.some(wrapped)
539 if (wrapped._tag === "Complete") {
540 return yield* wrapped.exit as Exit.Exit<any>
541 }
542 sleep ??= (yield* Schedule.toStepWithSleep(suspendedRetrySchedule))(
543 void 0
544 ).pipe(
545 Effect.catch(() =>
546 Effect.die(
547 `${self._tag}.execute: suspendedRetrySchedule exhausted`
548 )
549 )
550 )
551 yield* sleep
552 }
553 }),
554 poll: options.poll,
555 interrupt: options.interrupt,
556 interruptUnsafe: options.interruptUnsafe,
557 resume: options.resume,
558 activityExecute: Effect.fnUntraced(function*<
559 Success extends Schema.Constraint,
560 Error extends Schema.Constraint,
561 R
562 >(activity: Activity.Activity<Success, Error, R>, attempt: number) {
563 const result = yield* options.activityExecute(activity, attempt)
564 if (result._tag === "Suspended") {
565 return result
566 }
567 const exit = yield* Effect.orDie(
568 Schema.decodeEffect(activity.exitSchemaPartial)(toJsonExit(result.exit))
569 )
570 return new Workflow.Complete({ exit })
571 }),
572 deferredResult: Effect.fnUntraced(
573 function*<Success extends Schema.Constraint, Error extends Schema.Constraint>(
574 deferred: DurableDeferred.DurableDeferred<Success, Error>
575 ) {
576 const instance = yield* WorkflowInstance
577 yield* Effect.annotateCurrentSpan({
578 executionId: instance.executionId
579 })
580 const exit = yield* options.deferredResult(deferred)
581 if (Option.isNone(exit)) {
582 return Option.none()
583 }
584 return Option.some(
585 yield* Effect.orDie(
586 Schema.decodeEffect(deferred.exitSchema)(toJsonExit(exit.value))
587 ) as Effect.Effect<Exit.Exit<Success["Type"], Error["Type"]>>
588 )
589 },
590 Effect.withSpan(
591 "WorkflowEngine.deferredResult",
592 (deferred) => ({
593 attributes: { name: deferred.name }
594 }),
595 { captureStackTrace: false }
596 )
597 ),
598 deferredDone: Effect.fnUntraced(
599 function*<Success extends Schema.Constraint, Error extends Schema.Constraint>(
600 deferred: DurableDeferred.DurableDeferred<Success, Error>,
601 opts: {
602 readonly workflowName: string
603 readonly executionId: string
604 readonly deferredName: string
605 readonly exit: Exit.Exit<Success["Type"], Error["Type"]>
606 }
607 ) {
608 return yield* options.deferredDone({
609 workflowName: opts.workflowName,
610 executionId: opts.executionId,
611 deferredName: opts.deferredName,
612 exit: yield* Schema.encodeEffect(deferred.exitSchema)(
613 opts.exit
614 ) as Effect.Effect<Exit.Exit<unknown, unknown>>
615 })
616 },
617 Effect.withSpan(
618 "WorkflowEngine.deferredDone",
619 (_, { deferredName, executionId }) => ({
620 attributes: { name: deferredName, executionId }
621 }),
622 { captureStackTrace: false }
623 )
624 ),
625 scheduleClock: (workflow, opts) =>
626 options.scheduleClock(workflow, opts).pipe(
627 Effect.withSpan(
628 "WorkflowEngine.scheduleClock",
629 {
630 attributes: {
631 executionId: opts.executionId,
632 name: opts.clock.name
633 }
634 },
635 {
636 captureStackTrace: false
637 }
638 )
639 )
640 })
641
642const defaultRetrySchedule = Schedule.min([
643 Schedule.exponential(200, 1.5),
644 Schedule.spaced(30000)
645])
646
647/**
648 * Layer that provides an in-memory `WorkflowEngine`.
649 *
650 * **When to use**
651 *
652 * Use to run tests and local development workflows where durability is not
653 * needed.
654 *
655 * **Gotchas**
656 *
657 * This layer keeps state only in memory and is not suitable for production
658 * workflows that require durability.
659 *
660 * @stability unstable
661 * @category layers
662 * @since 4.0.0
663 */
664export const layerMemory: Layer.Layer<WorkflowEngine> = Layer.effect(WorkflowEngine)(
665 Effect.gen(function*() {
666 const scope = yield* Effect.scope
667
668 const workflows = new Map<string, {
669 readonly workflow: Workflow.Any
670 readonly execute: (
671 payload: object,
672 executionId: string
673 ) => Effect.Effect<unknown, unknown, WorkflowInstance | WorkflowEngine>
674 readonly scope: Scope.Scope
675 }>()
676
677 type ExecutionState = {
678 readonly payload: object
679 readonly execute: (
680 payload: object,
681 executionId: string
682 ) => Effect.Effect<unknown, unknown, WorkflowInstance | WorkflowEngine>
683 readonly parent: string | undefined
684 instance: WorkflowInstance["Service"]
685 interrupted: boolean
686 resumeRequested: boolean
687 fiber: Fiber.Fiber<Workflow.Result<unknown, unknown>> | undefined
688 }
689 const executions = new Map<string, ExecutionState>()
690
691 type ActivityState = {
692 exit: Exit.Exit<Workflow.Result<unknown, unknown>> | undefined
693 }
694 const activities = new Map<string, ActivityState>()
695
696 const deferredState = makeDeferredState()
697
698 const resume = Effect.fnUntraced(function*(executionId: string): Effect.fn.Return<void> {
699 const state = executions.get(executionId)
700 if (!state) return
701 const exit = state.fiber?.pollUnsafe()
702 if (exit && exit._tag === "Success" && exit.value._tag === "Complete") {
703 return
704 } else if (state.fiber && !exit) {
705 // A child can finish while the parent is still releasing its activity.
706 // Preserve the wake until that run has published its suspended result.
707 if (!state.resumeRequested) {
708 state.resumeRequested = true
709 yield* Fiber.await(state.fiber).pipe(
710 Effect.flatMap(() => {
711 state.resumeRequested = false
712 return resume(executionId)
713 }),
714 Effect.forkIn(scope)
715 )
716 }
717 return
718 }
719
720 const entry = workflows.get(state.instance.workflow._tag)!
721 const instance = WorkflowInstance.initial(
722 state.instance.workflow,
723 state.instance.executionId,
724 state.instance.scope
725 )
726 state.instance = instance
727 state.fiber = yield* state.execute(state.payload, state.instance.executionId).pipe(
728 Effect.onExit(() => {
729 if (!state.interrupted) {
730 return Effect.void
731 }
732 instance.interrupted = true
733 instance.suspended = false
734 return Effect.withFiber((fiber) => Effect.interruptible(Fiber.interrupt(fiber)))
735 }),
736 Workflow.intoResult,
737 Effect.provideService(WorkflowEngine, engine),
738 (effect) => deferredState.trackRun(instance, effect),
739 Effect.tap((result) => {
740 if (!state.parent || result._tag !== "Complete") {
741 return Effect.void
742 }
743 return Effect.forkIn(resume(state.parent), scope)
744 }),
745 Effect.forkIn(entry.scope)
746 )
747 })
748
749 const deferredResults = new Map<string, Exit.Exit<any, any>>()
750
751 const clocks = yield* FiberMap.make<string>()
752
753 const engine = makeUnsafe({
754 register: Effect.fnUntraced(function*(workflow, execute) {
755 workflows.set(workflow._tag, {
756 workflow,
757 execute,
758 scope: yield* Effect.scope
759 })
760 }),
761 execute: Effect.fnUntraced(function*(workflow, options) {
762 const entry = workflows.get(workflow._tag)
763 if (!entry) {
764 return yield* Effect.orDie(Effect.fail(`Workflow ${workflow._tag} is not registered`))
765 }
766
767 let state = executions.get(options.executionId)
768 if (!state) {
769 state = {
770 payload: options.payload,
771 execute: entry.execute,
772 instance: WorkflowInstance.initial(workflow, options.executionId),
773 interrupted: false,
774 resumeRequested: false,
775 fiber: undefined,
776 parent: options.parent?.executionId
777 }
778 executions.set(options.executionId, state)
779 yield* resume(options.executionId)
780 }
781 if (options.discard) return
782 // Capture together so a wake that swaps in a replay cannot desync them.
783 const instance = state.instance
784 const exit = yield* Fiber.await(state.fiber!)
785 if (Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause) && instance.suspended) {
786 // A completion preempted the run; the caller retries into the replay.
787 return new Workflow.Suspended({}) as any
788 }
789 return (yield* exit) as any
790 }),
791 interrupt: Effect.fnUntraced(function*(_workflow, executionId) {
792 const state = executions.get(executionId)
793 if (!state) return
794 state.interrupted = true
795 yield* resume(executionId)
796 }),
797 interruptUnsafe: Effect.fnUntraced(function*(_workflow, executionId) {
798 const state = executions.get(executionId)
799 if (!state) return
800 state.interrupted = true
801 if (state.fiber) {
802 yield* Fiber.interrupt(state.fiber)
803 }
804 }),
805 resume(_workflow, executionId) {
806 return resume(executionId)
807 },
808 activityExecute: Effect.fnUntraced(function*(activity, attempt) {
809 const instance = yield* WorkflowInstance
810 const activityId = `${instance.executionId}/${activity.name}/${attempt}`
811 let state = activities.get(activityId)
812 if (state) {
813 const exit = state.exit
814 if (exit && exit._tag === "Success" && exit.value._tag === "Suspended") {
815 state.exit = undefined
816 } else if (exit) {
817 return yield* exit
818 }
819 } else {
820 state = { exit: undefined }
821 activities.set(activityId, state)
822 }
823 const activityInstance = WorkflowInstance.initial(instance.workflow, instance.executionId)
824 activityInstance.interrupted = instance.interrupted
825 return yield* activity.executeEncoded.pipe(
826 Workflow.intoResult,
827 Effect.provideService(WorkflowInstance, activityInstance),
828 Effect.onExit((exit) => {
829 state.exit = exit
830 return Effect.void
831 })
832 )
833 }),
834 poll: (_workflow, executionId) =>
835 Effect.suspend(() => {
836 const state = executions.get(executionId)
837 if (!state) {
838 return Effect.succeedNone
839 }
840 const exit = state.fiber?.pollUnsafe()
841 if (!exit) {
842 return Effect.succeedNone
843 }
844 return exit._tag === "Success"
845 ? Effect.succeedSome(exit.value)
846 : Effect.die(exit.cause)
847 }),
848 deferredResult: Effect.fnUntraced(function*(deferred) {
849 const instance = yield* WorkflowInstance
850 const id = `${instance.executionId}/${deferred.name}`
851 return Option.fromNullishOr(deferredResults.get(id))
852 }),
853 deferredDone: (options) =>
854 Effect.suspend(() => {
855 const id = `${options.executionId}/${options.deferredName}`
856 if (deferredResults.has(id)) return Effect.void
857 deferredResults.set(id, options.exit)
858 const wake = Effect.andThen(
859 deferredState.deferredDone(options.executionId, options.deferredName),
860 resume(options.executionId)
861 )
862 return Effect.flatMap(Effect.serviceOption(WorkflowInstance), (instance) =>
863 // A workflow finalizer cannot wait for its own run's cleanup.
864 // The engine scope owns the wake so cleanup still precedes replay.
865 Option.isSome(instance) && instance.value.executionId === options.executionId
866 ? wake.pipe(Effect.forkIn(scope), Effect.asVoid)
867 : wake)
868 }),
869 scheduleClock: (workflow, options) =>
870 engine.deferredDone(options.clock.deferred, {
871 workflowName: workflow._tag,
872 executionId: options.executionId,
873 deferredName: options.clock.deferred.name,
874 exit: Exit.void
875 }).pipe(
876 Effect.delay(options.clock.duration),
877 FiberMap.run(clocks, `${options.executionId}/${options.clock.name}`, { onlyIfMissing: true }),
878 Effect.asVoid
879 )
880 })
881
882 return engine
883 })
884)
885
886const toJsonExit = Exit.map((value: any) => value ?? null)