1import { assert, describe, it } from "@effect/vitest"
2import { Cause, Duration, Effect, Exit, Fiber, Latch, Layer, Option, Ref, Schema, Scope } from "effect"
3import { TestClock } from "effect/testing"
4import { Activity, DurableClock, DurableDeferred, Workflow, WorkflowEngine } from "effect/workflow"
5
6describe("deferred self-completion", () => {
7 for (const failure of [false, true]) {
8 it.live(failure ? "failure" : "success", () =>
9 Effect.gen(function*() {
10 const signal = DurableDeferred.make("signal", { success: Schema.String, error: Schema.String })
11 const read = yield* Latch.make()
12 const cleanup = yield* Latch.make()
13 const release = yield* Latch.make()
14 const events: Array<string> = []
15 let runs = 0
16 const workflow = Workflow.make("SelfCompletion", {
17 payload: {},
18 success: Schema.String,
19 error: Schema.String,
20 idempotencyKey: () => "one"
21 })
22 const layer = workflow.toLayer(() =>
23 Effect.gen(function*() {
24 const run = ++runs
25 events.push(`start-${run}`)
26 const engine = yield* WorkflowEngine.WorkflowEngine
27 return yield* DurableDeferred.raceAll({
28 name: "race",
29 success: Schema.String,
30 error: Schema.String,
31 effects: [
32 DurableDeferred.await(signal),
33 read.await.pipe(
34 Effect.andThen(Effect.yieldNow),
35 Effect.andThen(failure ? Effect.fail("boom") : Effect.succeed("ok")),
36 DurableDeferred.into(signal),
37 // Successful completion must preempt this producer and replay the run.
38 Effect.andThen(Effect.never),
39 Effect.onInterrupt(() =>
40 run === 1
41 ? Effect.gen(function*() {
42 events.push("cleanup-start")
43 yield* cleanup.open
44 yield* release.await
45 events.push("cleanup-end")
46 })
47 : Effect.void
48 )
49 )
50 ]
51 }).pipe(
52 Effect.provideService(WorkflowEngine.WorkflowEngine, {
53 ...engine,
54 deferredResult: (deferred) =>
55 engine.deferredResult(deferred).pipe(
56 Effect.tap(() => deferred.name === signal.name ? read.open : Effect.void)
57 )
58 }),
59 Effect.ensuring(Effect.sync(() => events.push(`end-${run}`)))
60 )
61 })
62 ).pipe(Layer.provideMerge(WorkflowEngine.layerMemory))
63
64 yield* Effect.gen(function*() {
65 const execution = yield* workflow.execute({}).pipe(Effect.exit, Effect.forkChild({ startImmediately: true }))
66 if (!failure) {
67 yield* cleanup.await
68 for (let i = 0; i < 20; i++) yield* Effect.yieldNow
69 assert.deepStrictEqual(events, ["start-1", "cleanup-start"], "replay must wait for cleanup")
70 yield* release.open
71 }
72 const result = yield* Fiber.join(execution)
73 if (failure) {
74 assert.ok(Exit.isFailure(result))
75 assert.strictEqual(result.cause.reasons.length, 1)
76 const reason = result.cause.reasons[0]
77 assert.ok(Cause.isFailReason(reason))
78 assert.strictEqual(reason.error, "boom")
79 } else {
80 assert.deepStrictEqual(result, Exit.succeed("ok"))
81 assert.deepStrictEqual(events, ["start-1", "cleanup-start", "cleanup-end", "end-1", "start-2", "end-2"])
82 }
83 }).pipe(Effect.provide(layer))
84 }), 5_000)
85 }
86})
87
88describe("WorkflowEngine", () => {
89 const IncrementWorkflow = Workflow.make("WorkflowEngine/IncrementWorkflow", {
90 payload: { value: Schema.Number },
91 success: Schema.Number,
92 idempotencyKey: ({ value }) => String(value)
93 })
94
95 const IncrementWorkflowLayer = IncrementWorkflow.toLayer(({ value }) => Effect.succeed(value + 1))
96
97 class ClassWorkflow extends Workflow.make("WorkflowEngine/ClassWorkflow", {
98 payload: { value: Schema.Number },
99 success: Schema.Number,
100 idempotencyKey: ({ value }) => String(value)
101 }) {}
102
103 const ClassWorkflowLayer = ClassWorkflow.toLayer(({ value }) => Effect.succeed(value + 1))
104
105 const DeferredRaceWorkflow = Workflow.make("WorkflowEngine/DeferredRaceWorkflow", {
106 payload: { id: Schema.String },
107 success: Schema.String,
108 idempotencyKey: ({ id }) => id
109 })
110
111 const DeferredRaceGate = DurableDeferred.make("WorkflowEngine/DeferredRaceGate", {
112 success: Schema.String
113 })
114
115 let deferredRaceRuns = 0
116 const DeferredRaceWorkflowLayer = DeferredRaceWorkflow.toLayer(() =>
117 Effect.suspend(() => {
118 deferredRaceRuns++
119 return DurableDeferred.raceAll({
120 name: "memory-deferred-race",
121 success: Schema.String,
122 error: Schema.Never,
123 effects: [
124 DurableDeferred.await(DeferredRaceGate),
125 Effect.sleep("10 seconds").pipe(Effect.as("activity"))
126 ]
127 })
128 })
129 )
130
131 it.effect("layer executes and polls workflows", () =>
132 Effect.gen(function*() {
133 const executionId = yield* IncrementWorkflow.execute({ value: 1 }, { discard: true })
134 const result = yield* IncrementWorkflow.execute({ value: 1 })
135 const polled = yield* IncrementWorkflow.poll(executionId)
136
137 assert.strictEqual(result, 2)
138 assert(Option.isSome(polled) && polled.value._tag === "Complete" && Exit.isSuccess(polled.value.exit))
139 assert.strictEqual(polled.value.exit.value, 2)
140 }).pipe(
141 Effect.provide(IncrementWorkflowLayer.pipe(
142 Layer.provideMerge(WorkflowEngine.layerMemory)
143 ))
144 ))
145
146 for (
147 const { childCount, concurrency, waves } of [
148 { childCount: 3, concurrency: "unbounded" as const, waves: [3] },
149 { childCount: 5, concurrency: 2, waves: [2, 4, 5] }
150 ]
151 ) {
152 it.effect(`layerMemory replays suspended activity fan-out with concurrency ${concurrency}`, () =>
153 Effect.gen(function*() {
154 const Parent = Workflow.make("WorkflowEngine/FanOutParent", {
155 payload: {},
156 success: Schema.Array(Schema.Number),
157 idempotencyKey: () => "parent"
158 })
159 const Child = Workflow.make("WorkflowEngine/FanOutChild", {
160 payload: { index: Schema.Number },
161 success: Schema.Number,
162 idempotencyKey: ({ index }) => String(index)
163 })
164 const started = new Set<number>()
165 let parentRuns = 0
166 let activityRuns = 0
167 let released = 0
168 const ParentLayer = Parent.toLayer(() =>
169 Effect.suspend(() => {
170 parentRuns++
171 return Activity.make({
172 name: "fan-out",
173 success: Schema.Array(Schema.Number),
174 execute: Effect.suspend(() => {
175 activityRuns++
176 return Effect.forEach(
177 Array.from({ length: childCount }, (_, index) => index),
178 (index) => Child.execute({ index }),
179 { concurrency }
180 )
181 }).pipe(Effect.ensuring(Effect.sync(() => {
182 released++
183 })))
184 })
185 })
186 )
187 const ChildLayer = Child.toLayer(Effect.fnUntraced(function*({ index }) {
188 started.add(index)
189 yield* DurableClock.sleep({ name: "child", duration: "2 seconds", inMemoryThreshold: Duration.zero })
190 return index
191 }))
192 yield* Effect.gen(function*() {
193 const executionId = yield* Parent.execute({}, { discard: true })
194 // Execution-ID digests use real promises, so wait for state rather than clock ticks.
195 for (const expected of waves) {
196 let result = yield* Parent.poll(executionId)
197 while (Option.isNone(result) || result.value._tag !== "Suspended" || started.size !== expected) {
198 yield* Effect.yieldNow
199 result = yield* Parent.poll(executionId)
200 }
201 assert.deepStrictEqual([...started].sort(), Array.from({ length: expected }, (_, index) => index))
202 assert.strictEqual(released, activityRuns)
203 assert.isAtMost(activityRuns, parentRuns)
204 if (expected === waves[0]) assert.strictEqual(parentRuns, 1)
205 yield* TestClock.adjust("2 seconds")
206 }
207 let result = yield* Parent.poll(executionId)
208 while (Option.isNone(result) || result.value._tag !== "Complete") {
209 yield* Effect.yieldNow
210 result = yield* Parent.poll(executionId)
211 }
212 assert.deepStrictEqual(
213 result.value,
214 new Workflow.Complete({ exit: Exit.succeed(Array.from({ length: childCount }, (_, index) => index)) })
215 )
216 assert.isAtLeast(parentRuns, waves.length + 1)
217 assert.isAtMost(activityRuns, parentRuns)
218 }).pipe(
219 Effect.provide(Layer.mergeAll(ParentLayer, ChildLayer).pipe(Layer.provideMerge(WorkflowEngine.layerMemory)))
220 )
221 }))
222 }
223
224 it.effect("layerMemory resumes when children complete during activity cleanup", () =>
225 Effect.gen(function*() {
226 const cleaningUp = yield* Latch.make()
227 const release = yield* Latch.make()
228 const Parent = Workflow.make("WorkflowEngine/CleanupParent", {
229 payload: {},
230 success: Schema.Array(Schema.Number),
231 idempotencyKey: () => "parent"
232 })
233 const Child = Workflow.make("WorkflowEngine/CleanupChild", {
234 payload: { index: Schema.Number },
235 success: Schema.Number,
236 idempotencyKey: ({ index }) => String(index)
237 })
238 const ParentLayer = Parent.toLayer(() =>
239 Activity.make({
240 name: "children",
241 success: Schema.Array(Schema.Number),
242 execute: Effect.forEach([0, 1], (index) => Child.execute({ index }), { concurrency: "unbounded" }).pipe(
243 Effect.ensuring(Effect.andThen(cleaningUp.open, release.await))
244 )
245 })
246 )
247 const ChildLayer = Child.toLayer(({ index }) =>
248 DurableClock.sleep({ name: "wait", duration: "2 seconds", inMemoryThreshold: Duration.zero }).pipe(
249 Effect.as(index)
250 )
251 )
252 yield* Effect.gen(function*() {
253 const executionId = yield* Parent.execute({}, { discard: true })
254 yield* cleaningUp.await
255 yield* TestClock.adjust("2 seconds")
256 yield* release.open
257 let result = yield* Parent.poll(executionId)
258 while (Option.isNone(result) || result.value._tag !== "Complete") {
259 yield* Effect.yieldNow
260 result = yield* Parent.poll(executionId)
261 }
262 assert.deepStrictEqual(
263 result.value,
264 new Workflow.Complete({ exit: Exit.succeed([0, 1]) })
265 )
266 }).pipe(
267 Effect.provide(Layer.mergeAll(ParentLayer, ChildLayer).pipe(Layer.provideMerge(WorkflowEngine.layerMemory)))
268 )
269 }))
270
271 for (const mode of ["failure", "interruption"] as const) {
272 it.effect(`layerMemory releases a pending child registration on ${mode}`, () =>
273 Effect.gen(function*() {
274 const computingId = yield* Latch.make()
275 const started = new Set<number>()
276 const Parent = Workflow.make(`WorkflowEngine/PendingParent/${mode}`, {
277 payload: {},
278 success: Schema.Array(Schema.Number),
279 idempotencyKey: () => "parent"
280 })
281 const Child = Workflow.make(`WorkflowEngine/PendingChild/${mode}`, {
282 payload: { index: Schema.Number },
283 success: Schema.Number,
284 idempotencyKey: ({ index }) => {
285 if (index === 1) {
286 if (mode === "failure") throw new Error("cannot compute the execution id")
287 computingId.openUnsafe()
288 }
289 return String(index)
290 }
291 })
292 const ParentLayer = Parent.toLayer(() =>
293 Activity.make({
294 name: "children",
295 success: Schema.Array(Schema.Number),
296 execute: Effect.forEach([0, 1], (index) => {
297 const execute = Child.execute({ index })
298 if (index === 0) return execute
299 return mode === "failure"
300 ? Effect.catchDefect(execute, () => Effect.succeed(-1))
301 : Effect.raceFirst(execute, Effect.as(computingId.await, -1))
302 }, { concurrency: "unbounded" })
303 })
304 )
305 const ChildLayer = Child.toLayer(Effect.fnUntraced(function*({ index }) {
306 started.add(index)
307 yield* DurableClock.sleep({ name: "wait", duration: "2 seconds", inMemoryThreshold: Duration.zero })
308 return index
309 }))
310 yield* Effect.gen(function*() {
311 const executionId = yield* Parent.execute({}, { discard: true })
312 let result = yield* Parent.poll(executionId)
313 while (Option.isNone(result) || result.value._tag !== "Suspended") {
314 yield* Effect.yieldNow
315 result = yield* Parent.poll(executionId)
316 }
317 assert.deepStrictEqual([...started], [0])
318 yield* TestClock.adjust("2 seconds")
319 result = yield* Parent.poll(executionId)
320 while (Option.isNone(result) || result.value._tag !== "Complete") {
321 yield* Effect.yieldNow
322 result = yield* Parent.poll(executionId)
323 }
324 assert.deepStrictEqual(
325 result.value,
326 new Workflow.Complete({ exit: Exit.succeed([0, -1]) })
327 )
328 }).pipe(
329 Effect.provide(Layer.mergeAll(ParentLayer, ChildLayer).pipe(Layer.provideMerge(WorkflowEngine.layerMemory)))
330 )
331 }))
332 }
333
334 it.effect("discard returns the deterministic execution ID", () =>
335 Effect.gen(function*() {
336 const executionId = yield* IncrementWorkflow.executionId({ value: 1 })
337 const discardedExecutionId = yield* IncrementWorkflow.execute({ value: 1 }, { discard: true })
338
339 assert.strictEqual(discardedExecutionId, executionId)
340 }).pipe(
341 Effect.provide(IncrementWorkflowLayer.pipe(
342 Layer.provideMerge(WorkflowEngine.layerMemory)
343 ))
344 ))
345
346 it.effect("distinguishes workflow tags and idempotency keys at string boundaries", () =>
347 Effect.gen(function*() {
348 const makeWorkflow = (tag: string) =>
349 Workflow.make(tag, {
350 payload: { key: Schema.String, value: Schema.Number },
351 idempotencyKey: ({ key }) => key
352 })
353
354 for (
355 const [leftTag, leftKey, rightTag, rightKey] of [
356 ["Order", "Pay-42", "Order-Pay", "42"],
357 ["Order", "Pay:42", "Order:Pay", "42"],
358 ["", "-", "-", ""],
359 ["A\"", "B\\", "A\"-B", "\\"]
360 ]
361 ) {
362 const left = makeWorkflow(leftTag)
363 const right = makeWorkflow(rightTag)
364 const leftId = yield* left.executionId({ key: leftKey, value: 1 })
365 const rightId = yield* right.executionId({ key: rightKey, value: 1 })
366
367 assert.notStrictEqual(leftId, rightId)
368 assert.strictEqual(yield* left.executionId({ key: leftKey, value: 1 }), leftId)
369 assert.strictEqual(yield* left.executionId({ key: leftKey, value: 2 }), leftId)
370 }
371 }))
372
373 it.effect("supports class extension", () =>
374 Effect.gen(function*() {
375 const result = yield* ClassWorkflow.execute({ value: 1 })
376
377 assert.strictEqual(ClassWorkflow._tag, "WorkflowEngine/ClassWorkflow")
378 assert.strictEqual(result, 2)
379 }).pipe(
380 Effect.provide(ClassWorkflowLayer.pipe(
381 Layer.provideMerge(WorkflowEngine.layerMemory)
382 ))
383 ))
384
385 it.effect("layerMemory wakes an active workflow when a durable deferred completes", () =>
386 Effect.gen(function*() {
387 const payload = { id: "memory-deferred-race" }
388 const executionId = yield* DeferredRaceWorkflow.executionId(payload)
389 const fiber = yield* DeferredRaceWorkflow.execute(payload).pipe(
390 Effect.forkChild({ startImmediately: true })
391 )
392
393 // Park the deferred branch before completing it.
394 yield* TestClock.adjust(1)
395 yield* TestClock.adjust(1)
396 const token = DurableDeferred.tokenFromExecutionId(DeferredRaceGate, {
397 workflow: DeferredRaceWorkflow,
398 executionId
399 })
400 yield* DurableDeferred.succeed(DeferredRaceGate, { token, value: "signal" })
401 // Require the wake to settle before the sleeper.
402 let polled = yield* DeferredRaceWorkflow.poll(executionId)
403 while (Option.isNone(polled) || polled.value._tag !== "Complete") {
404 yield* Effect.yieldNow
405 polled = yield* DeferredRaceWorkflow.poll(executionId)
406 }
407
408 // Let the caller's suspended-retry loop pick up a replayed result.
409 yield* TestClock.adjust("1 second")
410 assert.strictEqual(yield* Fiber.join(fiber), "signal")
411 // Usually the completion preempts the parked run (2 runs); under load
412 // it can land before the branch parks and is read directly (1 run).
413 assert(deferredRaceRuns === 1 || deferredRaceRuns === 2)
414 }).pipe(
415 Effect.provide(DeferredRaceWorkflowLayer.pipe(
416 Layer.provideMerge(WorkflowEngine.layerMemory)
417 ))
418 ))
419
420 it.effect("layerMemory propagates interruption when the engine is shut down", () =>
421 Effect.gen(function*() {
422 const Stuck = Workflow.make("WorkflowEngine/ShutdownWorkflow", {
423 payload: { id: Schema.String },
424 idempotencyKey: ({ id }) => id
425 })
426 const layer = Stuck.toLayer(() => Effect.never).pipe(
427 Layer.provideMerge(WorkflowEngine.layerMemory)
428 )
429 const scope = yield* Scope.make()
430 const context = yield* Scope.provide(Layer.build(layer), scope)
431 const fiber = yield* Stuck.execute({ id: "one" }).pipe(
432 Effect.provideContext(context),
433 Effect.forkChild({ startImmediately: true })
434 )
435 yield* Effect.yieldNow
436 yield* Effect.yieldNow
437
438 // Shutting down must interrupt the caller, not report a suspension.
439 yield* Scope.close(scope, Exit.void)
440 const exit = yield* Fiber.await(fiber)
441 assert(Exit.hasInterrupts(exit))
442 }))
443
444 it.effect("layerMemory runs compensations when the engine is shut down", () =>
445 Effect.gen(function*() {
446 const ready = Latch.makeUnsafe()
447 const compensated: Array<string> = []
448 const Stuck = Workflow.make("WorkflowEngine/ShutdownCompensation", {
449 payload: { id: Schema.String },
450 idempotencyKey: ({ id }) => id
451 })
452 const layer = Stuck.toLayer(() =>
453 Effect.gen(function*() {
454 yield* Effect.succeed("a").pipe(
455 Stuck.withCompensation((value) => Effect.sync(() => compensated.push(value)))
456 )
457 ready.openUnsafe()
458 return yield* Effect.never
459 })
460 ).pipe(
461 Layer.provideMerge(WorkflowEngine.layerMemory)
462 )
463 const scope = yield* Scope.make()
464 const context = yield* Scope.provide(Layer.build(layer), scope)
465 const fiber = yield* Stuck.execute({ id: "one" }).pipe(
466 Effect.provideContext(context),
467 Effect.forkChild({ startImmediately: true })
468 )
469 yield* ready.await
470
471 yield* Scope.close(scope, Exit.void)
472 const exit = yield* Fiber.await(fiber)
473
474 assert(Exit.hasInterrupts(exit))
475 assert.deepStrictEqual(compensated, ["a"])
476 }))
477
478 it.effect("layerMemory closes finalizers registered before suspension", () =>
479 Effect.gen(function*() {
480 const gate = DurableDeferred.make("WorkflowEngine/SuspendedScope/Gate")
481 const finalized: Array<number> = []
482 let runs = 0
483 const Suspends = Workflow.make("WorkflowEngine/SuspendedScope", {
484 payload: { id: Schema.String },
485 idempotencyKey: ({ id }) => id
486 })
487 const layer = Suspends.toLayer(() =>
488 Effect.gen(function*() {
489 const run = ++runs
490 yield* Workflow.addFinalizer(() => Effect.sync(() => finalized.push(run)))
491 yield* DurableDeferred.await(gate)
492 })
493 ).pipe(Layer.provideMerge(WorkflowEngine.layerMemory))
494
495 yield* Effect.gen(function*() {
496 const payload = { id: "one" }
497 const executionId = yield* Suspends.execute(payload, { discard: true })
498 let result = yield* Suspends.poll(executionId)
499 while (Option.isNone(result) || result.value._tag !== "Suspended") {
500 yield* Effect.yieldNow
501 result = yield* Suspends.poll(executionId)
502 }
503
504 const token = DurableDeferred.tokenFromExecutionId(gate, { workflow: Suspends, executionId })
505 yield* DurableDeferred.succeed(gate, { token, value: void 0 })
506 yield* Suspends.execute(payload)
507
508 assert.deepStrictEqual(finalized, [2, 1])
509 }).pipe(Effect.provide(layer))
510 }))
511
512 it.effect("layerMemory hides deposited interrupts from body finalizers", () =>
513 Effect.gen(function*() {
514 const gate = DurableDeferred.make("WorkflowEngine/InterruptBodyFinalizer/Gate")
515 const observed = yield* Ref.make<ReadonlyArray<boolean>>([])
516 const Suspends = Workflow.make("WorkflowEngine/InterruptBodyFinalizer", {
517 payload: { id: Schema.String },
518 idempotencyKey: ({ id }) => id
519 })
520 const layer = Suspends.toLayer(Effect.fnUntraced(function*() {
521 const instance = yield* WorkflowEngine.WorkflowInstance
522 return yield* DurableDeferred.await(gate).pipe(
523 Effect.onExit(() => Ref.update(observed, (values) => [...values, instance.interrupted]))
524 )
525 })).pipe(Layer.provideMerge(WorkflowEngine.layerMemory))
526
527 yield* Effect.gen(function*() {
528 const payload = { id: "one" }
529 const executionId = yield* Suspends.execute(payload, { discard: true })
530 let result = yield* Suspends.poll(executionId)
531 while (Option.isNone(result) || result.value._tag !== "Suspended") {
532 yield* Effect.yieldNow
533 result = yield* Suspends.poll(executionId)
534 }
535
536 yield* Suspends.interrupt(executionId)
537 while (Option.isNone(result) || result.value._tag !== "Complete") {
538 yield* Effect.yieldNow
539 result = yield* Suspends.poll(executionId)
540 }
541
542 assert.deepStrictEqual(yield* Ref.get(observed), [false, false])
543 }).pipe(Effect.provide(layer))
544 }))
545
546 it.effect("layerMemory exposes deposited interrupts to workflow finalizers", () =>
547 Effect.gen(function*() {
548 const gate = DurableDeferred.make("WorkflowEngine/InterruptWorkflowFinalizer/Gate")
549 const observed = yield* Ref.make<ReadonlyArray<boolean>>([])
550 const runs = yield* Ref.make(0)
551 const Suspends = Workflow.make("WorkflowEngine/InterruptWorkflowFinalizer", {
552 payload: { id: Schema.String },
553 idempotencyKey: ({ id }) => id
554 })
555 const layer = Suspends.toLayer(Effect.fnUntraced(function*() {
556 const run = yield* Ref.getAndUpdate(runs, (run) => run + 1)
557 const instance = yield* WorkflowEngine.WorkflowInstance
558 if (run === 1) {
559 yield* Workflow.addFinalizer(() => Ref.update(observed, (values) => [...values, instance.interrupted]))
560 }
561 return yield* DurableDeferred.await(gate)
562 })).pipe(Layer.provideMerge(WorkflowEngine.layerMemory))
563
564 yield* Effect.gen(function*() {
565 const payload = { id: "one" }
566 const executionId = yield* Suspends.execute(payload, { discard: true })
567 let result = yield* Suspends.poll(executionId)
568 while (Option.isNone(result) || result.value._tag !== "Suspended") {
569 yield* Effect.yieldNow
570 result = yield* Suspends.poll(executionId)
571 }
572
573 yield* Suspends.interrupt(executionId)
574 while (Option.isNone(result) || result.value._tag !== "Complete") {
575 yield* Effect.yieldNow
576 result = yield* Suspends.poll(executionId)
577 }
578
579 assert.deepStrictEqual(yield* Ref.get(observed), [true])
580 }).pipe(Effect.provide(layer))
581 }))
582
583 it.effect("layerMemory preserves interruptUnsafe across a suspended replay", () =>
584 Effect.gen(function*() {
585 const gate = DurableDeferred.make("WorkflowEngine/InterruptUnsafeReplay/Gate")
586 const Suspends = Workflow.make("WorkflowEngine/InterruptUnsafeReplay", {
587 payload: { id: Schema.String },
588 idempotencyKey: ({ id }) => id
589 })
590 const layer = Suspends.toLayer(() => DurableDeferred.await(gate)).pipe(
591 Layer.provideMerge(WorkflowEngine.layerMemory)
592 )
593
594 yield* Effect.gen(function*() {
595 const engine = yield* WorkflowEngine.WorkflowEngine
596 const payload = { id: "one" }
597 const executionId = yield* Suspends.execute(payload, { discard: true })
598 let result = yield* Suspends.poll(executionId)
599 while (Option.isNone(result) || result.value._tag !== "Suspended") {
600 yield* Effect.yieldNow
601 result = yield* Suspends.poll(executionId)
602 }
603
604 yield* engine.interruptUnsafe(Suspends, executionId)
605 const token = DurableDeferred.tokenFromExecutionId(gate, { workflow: Suspends, executionId })
606 yield* DurableDeferred.succeed(gate, { token, value: void 0 })
607 while (Option.isNone(result) || result.value._tag !== "Complete") {
608 yield* Effect.yieldNow
609 result = yield* Suspends.poll(executionId)
610 }
611
612 assert(Option.isSome(result) && result.value._tag === "Complete")
613 assert(Exit.hasInterrupts(result.value.exit))
614 }).pipe(Effect.provide(layer))
615 }))
616})