1import { assert, describe, it } from "@effect/vitest"
2import { Cause, Clock, Duration, Effect, Exit, Fiber, Latch, Layer, Option, Schema } from "effect"
3import { ClusterSchema, ClusterWorkflowEngine, Entity, EntityId, Sharding } from "effect/cluster"
4import { PersistedQueue } from "effect/persistence"
5import { Rpc } from "effect/rpc"
6import { Activity, DurableClock, DurableDeferred, DurableQueue, Workflow, WorkflowEngine } from "effect/workflow"
7import { type Backend, type ClusterRunner, make } from "./harness.ts"
8
9const EndToEndWorkflow = Workflow.make("ClusterIntegrationEndToEnd", {
10 payload: {
11 id: Schema.String,
12 value: Schema.Number
13 },
14 success: Schema.Number,
15 idempotencyKey: ({ id }) => id
16})
17
18let endToEndGate = Latch.makeUnsafe(true)
19let endToEndEntered = Latch.makeUnsafe()
20const endToEndRuns = new Map<string, number>()
21
22const EndToEndWorkflowLayer = EndToEndWorkflow.toLayer(({ id, value }) =>
23 Activity.make({
24 name: "EndToEnd",
25 success: Schema.Number,
26 execute: Effect.gen(function*() {
27 endToEndRuns.set(id, (endToEndRuns.get(id) ?? 0) + 1)
28 endToEndEntered.openUnsafe()
29 yield* endToEndGate.await
30 return value + 1
31 })
32 })
33)
34
35const ReplayGate = DurableDeferred.make("ClusterIntegrationReplayGate", {
36 success: Schema.String
37})
38
39const ReplayWorkflow = Workflow.make("ClusterIntegrationReplay", {
40 payload: { id: Schema.String },
41 success: Schema.String,
42 idempotencyKey: ({ id }) => id
43})
44
45const replayRuns = new Map<string, number>()
46
47const ReplayWorkflowLayer = ReplayWorkflow.toLayer(Effect.fnUntraced(function*({ id }) {
48 yield* Activity.make({
49 name: "BeforeSuspension",
50 execute: Effect.sync(() => replayRuns.set(id, (replayRuns.get(id) ?? 0) + 1))
51 })
52 return yield* DurableDeferred.await(ReplayGate)
53}))
54
55const RestartGate = DurableDeferred.make("ClusterIntegrationRestartGate", {
56 success: Schema.String
57})
58
59const RestartWorkflow = Workflow.make("ClusterIntegrationRestart", {
60 payload: { id: Schema.String },
61 success: Schema.String,
62 idempotencyKey: ({ id }) => id
63})
64
65const RestartWorkflowLayer = RestartWorkflow.toLayer(() => DurableDeferred.await(RestartGate))
66
67const RaceReplayGate = DurableDeferred.make("ClusterIntegrationRaceReplayGate", {
68 success: Schema.String
69})
70
71const RaceReplayTail = DurableDeferred.make("ClusterIntegrationRaceReplayTail", {
72 success: Schema.String
73})
74
75const RaceReplayWorkflow = Workflow.make("ClusterIntegrationRaceReplay", {
76 payload: { id: Schema.String },
77 success: Schema.String,
78 idempotencyKey: ({ id }) => id
79})
80
81const raceReplayEnters = new Map<string, number>()
82
83const RaceReplayWorkflowLayer = RaceReplayWorkflow.toLayer(({ id }) =>
84 Effect.gen(function*() {
85 const winner = yield* DurableDeferred.raceAll({
86 name: "ClusterIntegrationRaceReplay",
87 success: Schema.String,
88 error: Schema.Never,
89 effects: [
90 DurableDeferred.await(RaceReplayGate),
91 Effect.sync(() => raceReplayEnters.set(id, (raceReplayEnters.get(id) ?? 0) + 1)).pipe(
92 Effect.andThen(Effect.never)
93 )
94 ]
95 })
96 const tail = yield* DurableDeferred.await(RaceReplayTail)
97 return `${winner}:${tail}`
98 })
99)
100
101const CompensationGate = DurableDeferred.make("ClusterIntegrationCompensationGate")
102
103class CompensationError extends Schema.Error<CompensationError>("ClusterIntegrationCompensationError")({
104 _tag: Schema.tag("ClusterIntegrationCompensationError"),
105 id: Schema.String
106}) {}
107
108const CompensationWorkflow = Workflow.make("ClusterIntegrationCompensation", {
109 payload: { id: Schema.String },
110 error: CompensationError,
111 idempotencyKey: ({ id }) => id
112})
113
114const compensationRuns = new Map<string, number>()
115
116const CompensationWorkflowLayer = CompensationWorkflow.toLayer(Effect.fnUntraced(function*({ id }) {
117 yield* Activity.make({
118 name: "ClusterIntegrationCompensationRegistered",
119 success: Schema.String,
120 execute: Effect.succeed(id)
121 }).pipe(
122 CompensationWorkflow.withCompensation(() =>
123 Effect.sync(() => compensationRuns.set(id, (compensationRuns.get(id) ?? 0) + 1))
124 )
125 )
126 yield* DurableDeferred.await(CompensationGate)
127 return yield* new CompensationError({ id })
128}))
129
130const SuspendFailureGate = DurableDeferred.make("ClusterIntegrationSuspendFailureGate")
131
132const SuspendFailureWorkflow = Workflow.make("ClusterIntegrationSuspendFailure", {
133 payload: { id: Schema.String },
134 idempotencyKey: ({ id }) => id
135}).annotate(Workflow.SuspendOnFailure, true)
136
137const suspendFailures = new Set<string>()
138
139const SuspendFailureWorkflowLayer = SuspendFailureWorkflow.toLayer(Effect.fnUntraced(function*({ id }) {
140 yield* DurableDeferred.await(SuspendFailureGate)
141 return yield* Activity.make({
142 name: "ClusterIntegrationSuspendFailure",
143 execute: Effect.sync(() => suspendFailures.add(id)).pipe(
144 Effect.andThen(Effect.die("suspend after owner loss"))
145 )
146 })
147}))
148
149const RaceBoundaryGateA = DurableDeferred.make("ClusterIntegrationRaceBoundaryGateA", {
150 success: Schema.String
151})
152
153const RaceBoundaryGateB = DurableDeferred.make("ClusterIntegrationRaceBoundaryGateB", {
154 success: Schema.String
155})
156
157const RaceBoundaryWorkflow = Workflow.make("ClusterIntegrationRaceBoundary", {
158 payload: { id: Schema.String },
159 success: Schema.String,
160 idempotencyKey: ({ id }) => id
161})
162
163let raceBoundaryCommitStarted = Latch.makeUnsafe()
164let raceBoundaryCommit = Latch.makeUnsafe()
165const raceBoundaryRuns = new Map<string, number>()
166
167const RaceBoundaryWorkflowLayer = RaceBoundaryWorkflow.toLayer(Effect.fnUntraced(function*({ id }) {
168 raceBoundaryRuns.set(id, (raceBoundaryRuns.get(id) ?? 0) + 1)
169 return yield* DurableDeferred.raceAll({
170 name: "ClusterIntegrationRaceBoundary",
171 success: Schema.String,
172 error: Schema.Never,
173 effects: [
174 DurableDeferred.await(RaceBoundaryGateA),
175 DurableDeferred.await(RaceBoundaryGateB)
176 ]
177 }).pipe(
178 Effect.ensuring(
179 Effect.sync(() => raceBoundaryCommitStarted.openUnsafe()).pipe(
180 Effect.andThen(raceBoundaryCommit.await)
181 )
182 )
183 )
184}))
185
186class RetryError extends Schema.Error<RetryError>("ClusterIntegrationRetryError")({
187 _tag: Schema.tag("ClusterIntegrationRetryError"),
188 attempt: Schema.Number
189}) {}
190
191const RetryWorkflow = Workflow.make("ClusterIntegrationRetry", {
192 payload: {
193 id: Schema.String,
194 succeed: Schema.Boolean
195 },
196 success: Schema.Number,
197 error: RetryError,
198 idempotencyKey: ({ id }) => id
199})
200
201const retryAttempts = new Map<string, Array<number>>()
202
203const RetryWorkflowLayer = RetryWorkflow.toLayer(({ id, succeed }) =>
204 Activity.make({
205 name: "Retry",
206 success: Schema.Number,
207 error: RetryError,
208 execute: Effect.gen(function*() {
209 const attempt = yield* Activity.CurrentAttempt
210 const attempts = retryAttempts.get(id) ?? []
211 attempts.push(attempt)
212 retryAttempts.set(id, attempts)
213 if (succeed && attempt === 3) return attempt
214 return yield* new RetryError({ attempt })
215 })
216 }).pipe(Activity.retry({ times: 2 }))
217)
218
219const ClockWorkflow = Workflow.make("ClusterIntegrationClock", {
220 payload: { id: Schema.String },
221 success: Schema.Number,
222 idempotencyKey: ({ id }) => id
223})
224
225const ClockWorkflowLayer = ClockWorkflow.toLayer(() =>
226 Effect.gen(function*() {
227 yield* DurableClock.sleep({
228 name: "RestartSleep",
229 duration: "1 second",
230 inMemoryThreshold: Duration.zero
231 })
232 return yield* Clock.currentTimeMillis
233 })
234)
235
236const Queue = DurableQueue.make({
237 name: "ClusterIntegrationQueue",
238 payload: { id: Schema.String },
239 success: Schema.String,
240 idempotencyKey: ({ id }) => id
241})
242
243const QueueWorkflow = Workflow.make("ClusterIntegrationQueue", {
244 payload: { id: Schema.String },
245 success: Schema.String,
246 idempotencyKey: ({ id }) => id
247})
248
249const QueueWorkflowLayer = QueueWorkflow.toLayer(({ id }) => DurableQueue.process(Queue, { id }))
250const queueRuns = new Map<string, number>()
251let queueWorkerGate = Latch.makeUnsafe()
252
253const QueueWorkerLayer = Layer.effectDiscard(
254 Effect.forkScoped(
255 Effect.suspend(() => queueWorkerGate.await).pipe(
256 Effect.andThen(DurableQueue.makeWorker(Queue, ({ id }) =>
257 Effect.sync(() => {
258 queueRuns.set(id, (queueRuns.get(id) ?? 0) + 1)
259 return `processed:${id}`
260 })))
261 )
262 )
263)
264
265const InterruptGate = DurableDeferred.make("ClusterIntegrationInterruptGate")
266
267const InterruptWorkflow = Workflow.make("ClusterIntegrationInterrupt", {
268 payload: { id: Schema.String },
269 idempotencyKey: ({ id }) => id
270})
271
272const InterruptWorkflowLayer = InterruptWorkflow.toLayer(() => DurableDeferred.await(InterruptGate))
273
274const ShutdownActivityWorkflow = Workflow.make("ClusterIntegrationShutdownActivity", {
275 payload: { id: Schema.String },
276 success: Schema.String,
277 idempotencyKey: ({ id }) => id
278})
279
280const ShutdownSuspendActivityWorkflow = Workflow.make("ClusterIntegrationShutdownSuspendActivity", {
281 payload: { id: Schema.String },
282 success: Schema.String,
283 idempotencyKey: ({ id }) => id
284}).annotate(Workflow.SuspendOnFailure, true)
285
286const activityHandoffState = {
287 ready: Latch.makeUnsafe(),
288 start: Latch.makeUnsafe(),
289 faultArmed: false,
290 persisted: Latch.makeUnsafe(),
291 faultRelease: Latch.makeUnsafe(true),
292 runs: new Map<string, number>(),
293 compensations: new Set<string>(),
294 resourceEvents: new Map<string, Array<"acquire" | "release">>()
295}
296
297const abandon = Effect.interruptible(Effect.callback<never>(() => {
298 const fiber = Fiber.getCurrent()!
299 fiber.interruptUnsafe(fiber.id, ClusterSchema.Abandon.annotation)
300}))
301
302const resetActivityHandoffState = (id: string, options?: { readonly faultReleaseOpen?: boolean }) => {
303 activityHandoffState.ready = Latch.makeUnsafe()
304 activityHandoffState.start = Latch.makeUnsafe()
305 activityHandoffState.faultArmed = false
306 activityHandoffState.persisted = Latch.makeUnsafe()
307 activityHandoffState.faultRelease = Latch.makeUnsafe(options?.faultReleaseOpen ?? true)
308 activityHandoffState.runs.delete(id)
309 activityHandoffState.compensations.delete(id)
310 activityHandoffState.resourceEvents.set(id, [])
311}
312
313const ShutdownActivityWorkflowLayer = ShutdownActivityWorkflow.toLayer(({ id }) =>
314 Effect.gen(function*() {
315 yield* Effect.succeed(id).pipe(
316 ShutdownActivityWorkflow.withCompensation(() => Effect.sync(() => activityHandoffState.compensations.add(id)))
317 )
318 yield* Effect.acquireRelease(
319 Effect.sync(() => activityHandoffState.resourceEvents.get(id)!.push("acquire")),
320 () => Effect.sync(() => activityHandoffState.resourceEvents.get(id)!.push("release"))
321 ).pipe(Workflow.provideScope)
322 activityHandoffState.ready.openUnsafe()
323 yield* activityHandoffState.start.await
324 return yield* Activity.make({
325 name: "ShutdownActivity",
326 success: Schema.String,
327 execute: Effect.sync(() => {
328 activityHandoffState.runs.set(id, (activityHandoffState.runs.get(id) ?? 0) + 1)
329 return `completed:${id}`
330 })
331 })
332 })
333)
334
335const ShutdownSuspendActivityWorkflowLayer = ShutdownSuspendActivityWorkflow.toLayer(({ id }) =>
336 Effect.gen(function*() {
337 yield* Effect.succeed(id).pipe(
338 ShutdownSuspendActivityWorkflow.withCompensation(() =>
339 Effect.sync(() => activityHandoffState.compensations.add(id))
340 )
341 )
342 yield* Effect.acquireRelease(
343 Effect.sync(() => activityHandoffState.resourceEvents.get(id)!.push("acquire")),
344 () => Effect.sync(() => activityHandoffState.resourceEvents.get(id)!.push("release"))
345 ).pipe(Workflow.provideScope)
346 activityHandoffState.ready.openUnsafe()
347 yield* activityHandoffState.start.await
348 return yield* Activity.make({
349 name: "ShutdownActivity",
350 success: Schema.String,
351 execute: Effect.sync(() => {
352 activityHandoffState.runs.set(id, (activityHandoffState.runs.get(id) ?? 0) + 1)
353 return `completed:${id}`
354 })
355 })
356 })
357)
358
359const ActivityHandoffShardingLayer = Layer.effect(
360 Sharding.Sharding,
361 Effect.gen(function*() {
362 const sharding = yield* Sharding.Sharding
363 return Sharding.Sharding.of({
364 ...sharding,
365 makeClient: (entity) =>
366 // The wrapped factory returns the same dynamic RPC client shape, but
367 // its generic relationship to the entity protocol is no longer visible.
368 Effect.map(sharding.makeClient(entity), (makeClient) => (entityId: string) => {
369 // RPC clients expose their protocol methods dynamically, so the
370 // Proxy type cannot retain this client's generated method shape.
371 return new Proxy(makeClient(entityId), {
372 get(target, property, receiver) {
373 if (property !== "activity") return Reflect.get(target, property, receiver)
374 const activity = Reflect.get(target, property, receiver) as (
375 payload: { readonly name: string },
376 options?: object
377 ) => Effect.Effect<unknown, unknown, unknown>
378 return (payload: { readonly name: string }, options?: object) => {
379 if (!activityHandoffState.faultArmed) {
380 return activity(payload, options)
381 }
382 activityHandoffState.faultArmed = false
383 // Mirror sendOutgoing's persisted abandon branch during a
384 // runner shutdown: accept the durable request, then interrupt
385 // the caller because this runner can never observe the reply.
386 return activity(payload, { ...options, discard: true }).pipe(
387 Effect.tap(() => activityHandoffState.persisted.open),
388 Effect.andThen(activityHandoffState.faultRelease.await),
389 Effect.andThen(abandon)
390 )
391 }
392 }
393 }) as any
394 }) as any
395 })
396 })
397)
398
399const ShutdownActivityEntities = Layer.mergeAll(
400 ShutdownActivityWorkflowLayer,
401 ShutdownSuspendActivityWorkflowLayer
402).pipe(
403 Layer.provide(Layer.effect(
404 WorkflowEngine.WorkflowEngine,
405 ClusterWorkflowEngine.make
406 )),
407 Layer.provide(ActivityHandoffShardingLayer),
408 Layer.orDie
409)
410
411const CompleteDeferred = Rpc.make("CompleteDeferred", {
412 payload: {
413 token: DurableDeferred.Token,
414 value: Schema.String
415 },
416 success: Schema.String
417})
418
419const CompleteRaceBoundaryDeferred = Rpc.make("CompleteRaceBoundaryDeferred", {
420 payload: {
421 token: DurableDeferred.Token,
422 value: Schema.String
423 },
424 success: Schema.String
425})
426
427const DeferredControl = Entity.make("ClusterIntegrationDeferredControl", [
428 CompleteDeferred,
429 CompleteRaceBoundaryDeferred
430])
431
432const DeferredControlLayer = DeferredControl.toLayer(Effect.gen(function*() {
433 const runner = yield* Entity.CurrentRunnerAddress
434 return {
435 CompleteDeferred: ({ payload }) =>
436 DurableDeferred.succeed(RestartGate, payload).pipe(
437 Effect.as(`${runner.host}:${runner.port}`)
438 ),
439 CompleteRaceBoundaryDeferred: ({ payload }) =>
440 DurableDeferred.succeed(RaceBoundaryGateB, payload).pipe(
441 Effect.as(`${runner.host}:${runner.port}`)
442 )
443 }
444}))
445
446const Workflows = Layer.mergeAll(
447 EndToEndWorkflowLayer,
448 ReplayWorkflowLayer,
449 RestartWorkflowLayer,
450 RaceReplayWorkflowLayer,
451 CompensationWorkflowLayer,
452 SuspendFailureWorkflowLayer,
453 RaceBoundaryWorkflowLayer,
454 RetryWorkflowLayer,
455 ClockWorkflowLayer,
456 QueueWorkflowLayer,
457 QueueWorkerLayer,
458 InterruptWorkflowLayer,
459 DeferredControlLayer
460)
461
462const entities = ({ prefix }: { readonly prefix: string }) => {
463 const queue = PersistedQueue.layer.pipe(
464 Layer.provideMerge(PersistedQueue.layerStoreSql({
465 tableName: `${prefix}_workflow_queue`,
466 pollInterval: 100,
467 lockRefreshInterval: 500,
468 lockExpiration: 1_750
469 }))
470 )
471 return Workflows.pipe(
472 Layer.provide(queue),
473 Layer.provide(ClusterWorkflowEngine.layer),
474 Layer.orDie
475 )
476}
477
478type Cluster = Effect.Success<ReturnType<typeof make>>
479
480const withWorkflow = <A, E, R>(
481 cluster: Cluster,
482 effect: Effect.Effect<A, E, R | WorkflowEngine.WorkflowEngine>
483) => Effect.provideService(effect, WorkflowEngine.WorkflowEngine, cluster.workflowEngine)
484
485const waitForSuspended = Effect.fnUntraced(function*<
486 Name extends string,
487 Payload extends Workflow.AnyStructSchema,
488 Success extends Schema.Top,
489 Error extends Schema.Top
490>(
491 cluster: Cluster,
492 workflow: Workflow.Workflow<Name, Payload, Success, Error>,
493 executionId: string
494) {
495 yield* cluster.waitUntil(
496 `${workflow._tag}/${executionId} did not suspend`,
497 Effect.map(workflow.poll(executionId), (result) => Option.isSome(result) && result.value._tag === "Suspended")
498 )
499})
500
501const waitForComplete = Effect.fnUntraced(function*<
502 Name extends string,
503 Payload extends Workflow.AnyStructSchema,
504 Success extends Schema.Top,
505 Error extends Schema.Top
506>(
507 cluster: Cluster,
508 workflow: Workflow.Workflow<Name, Payload, Success, Error>,
509 executionId: string
510) {
511 let complete: Workflow.Complete<Success["Type"], Error["Type"]> | undefined
512 yield* cluster.waitUntil(
513 `${workflow._tag}/${executionId} did not complete`,
514 Effect.map(workflow.poll(executionId), (result) => {
515 if (Option.isNone(result) || result.value._tag !== "Complete") return false
516 complete = result.value
517 return true
518 })
519 )
520 return complete!
521})
522
523const restart = Effect.fnUntraced(function*(cluster: Cluster) {
524 const running = cluster.runners.filter((runner) => runner.state() === "running")
525 yield* Effect.forEach(running, cluster.kill, { concurrency: "unbounded", discard: true })
526 const replacements = yield* cluster.start(3)
527 yield* cluster.waitForStableAssignments()
528 return replacements
529})
530
531const workflowOwner = (cluster: Cluster, executionId: string) => {
532 const shard = cluster.clientSharding.getShardId(EntityId.make(executionId), "default")
533 return cluster.ownersOfShard(shard)[0]
534}
535
536const findControlOnAnotherRunner = Effect.fnUntraced(function*(
537 cluster: Cluster,
538 workflowRunner: ClusterRunner
539) {
540 for (let index = 0; index < 2_000; index++) {
541 const id = `control-${index}`
542 const owner = yield* cluster.ownerOfEntity(DeferredControl, id)
543 if (owner !== undefined && owner !== workflowRunner) return [id, owner] as const
544 }
545 return yield* Effect.die("Could not route deferred control to another runner")
546})
547
548describe("cluster workflow integration", () => {
549 for (const backend of ["pg", "mysql"] satisfies ReadonlyArray<Backend>) {
550 it.live(`${backend}: executes end to end and deduplicates concurrent callers`, () =>
551 Effect.gen(function*() {
552 const id = `${backend}-end-to-end`
553 endToEndGate = Latch.makeUnsafe()
554 endToEndEntered = Latch.makeUnsafe()
555 const cluster = yield* make({ backend, entities })
556 yield* cluster.start(3)
557 yield* cluster.waitForStableAssignments()
558
559 const first = yield* withWorkflow(cluster, EndToEndWorkflow.execute({ id, value: 41 })).pipe(
560 Effect.forkChild({ startImmediately: true })
561 )
562 yield* cluster.waitUntil("The end-to-end activity did not start", Effect.as(endToEndEntered.await, true))
563 const second = yield* withWorkflow(cluster, EndToEndWorkflow.execute({ id, value: 41 })).pipe(
564 Effect.forkChild({ startImmediately: true })
565 )
566 endToEndGate.openUnsafe()
567
568 assert.strictEqual(yield* Fiber.join(first), 42)
569 assert.strictEqual(yield* Fiber.join(second), 42)
570 assert.strictEqual(endToEndRuns.get(id), 1)
571 }))
572
573 it.live(`${backend}: replays completed activities after the owner dies`, () =>
574 Effect.gen(function*() {
575 const id = `${backend}-replay`
576 const cluster = yield* make({ backend, entities })
577 yield* cluster.start(3)
578 yield* cluster.waitForStableAssignments()
579 const executionId = yield* withWorkflow(cluster, ReplayWorkflow.execute({ id }, { discard: true }))
580 yield* waitForSuspended(cluster, ReplayWorkflow, executionId)
581 assert.strictEqual(replayRuns.get(id), 1)
582
583 const owner = workflowOwner(cluster, executionId)
584 assert.isDefined(owner)
585 yield* cluster.kill(owner!)
586 yield* cluster.waitForStableAssignments()
587 const token = DurableDeferred.tokenFromExecutionId(ReplayGate, { workflow: ReplayWorkflow, executionId })
588 yield* withWorkflow(cluster, DurableDeferred.succeed(ReplayGate, { token, value: "resumed" }))
589 const result = yield* waitForComplete(cluster, ReplayWorkflow, executionId)
590
591 assert.deepStrictEqual(result.exit, Exit.succeed("resumed"))
592 assert.strictEqual(replayRuns.get(id), 1)
593 }))
594
595 it.live(`${backend}: resumes deferred workflows across a whole-cluster restart from another runner`, () =>
596 Effect.gen(function*() {
597 const id = `${backend}-restart`
598 const cluster = yield* make({ backend, entities })
599 yield* cluster.start(3)
600 yield* cluster.waitForStableAssignments()
601 const executionId = yield* withWorkflow(cluster, RestartWorkflow.execute({ id }, { discard: true }))
602 yield* waitForSuspended(cluster, RestartWorkflow, executionId)
603 yield* restart(cluster)
604
605 const owner = workflowOwner(cluster, executionId)
606 assert.isDefined(owner)
607 const [controlId, controlOwner] = yield* findControlOnAnotherRunner(cluster, owner!)
608 const control = yield* cluster.getClient(DeferredControl)
609 const token = DurableDeferred.tokenFromExecutionId(RestartGate, { workflow: RestartWorkflow, executionId })
610 const completedBy = yield* control(controlId).CompleteDeferred({ token, value: "after-restart" })
611 assert.strictEqual(completedBy, `${controlOwner.address.host}:${controlOwner.address.port}`)
612
613 const result = yield* waitForComplete(cluster, RestartWorkflow, executionId)
614 assert.deepStrictEqual(result.exit, Exit.succeed("after-restart"))
615 }))
616
617 it.live(`${backend}: wakes a DurableDeferred raceAll and replays the winner after owner death`, () =>
618 Effect.gen(function*() {
619 const id = `${backend}-race-replay`
620 const cluster = yield* make({ backend, entities })
621 yield* cluster.start(3)
622 yield* cluster.waitForStableAssignments()
623 const executionId = yield* withWorkflow(cluster, RaceReplayWorkflow.execute({ id }, { discard: true }))
624 yield* cluster.waitUntil(
625 "The durable race did not start",
626 Effect.sync(() => (raceReplayEnters.get(id) ?? 0) > 0)
627 )
628
629 const raceToken = DurableDeferred.tokenFromExecutionId(RaceReplayGate, {
630 workflow: RaceReplayWorkflow,
631 executionId
632 })
633 yield* withWorkflow(
634 cluster,
635 DurableDeferred.succeed(RaceReplayGate, {
636 token: raceToken,
637 value: "winner"
638 })
639 )
640 yield* waitForSuspended(cluster, RaceReplayWorkflow, executionId)
641 const entriesBeforeKill = raceReplayEnters.get(id)
642
643 const owner = workflowOwner(cluster, executionId)
644 assert.isDefined(owner)
645 yield* cluster.kill(owner!)
646 yield* cluster.waitForStableAssignments()
647
648 const tailToken = DurableDeferred.tokenFromExecutionId(RaceReplayTail, {
649 workflow: RaceReplayWorkflow,
650 executionId
651 })
652 yield* withWorkflow(
653 cluster,
654 DurableDeferred.succeed(RaceReplayTail, {
655 token: tailToken,
656 value: "after-owner-death"
657 })
658 )
659 const result = yield* waitForComplete(cluster, RaceReplayWorkflow, executionId)
660
661 assert.deepStrictEqual(result.exit, Exit.succeed("winner:after-owner-death"))
662 assert.strictEqual(raceReplayEnters.get(id), entriesBeforeKill)
663 }))
664
665 it.live(`${backend}: runs compensation and SuspendOnFailure after the owner is lost`, () =>
666 Effect.gen(function*() {
667 const compensationId = `${backend}-compensation-owner-loss`
668 const suspendId = `${backend}-suspend-owner-loss`
669 const cluster = yield* make({ backend, entities })
670 yield* cluster.start(3)
671 yield* cluster.waitForStableAssignments()
672 const compensationExecutionId = yield* withWorkflow(
673 cluster,
674 CompensationWorkflow.execute({ id: compensationId }, { discard: true })
675 )
676 const suspendExecutionId = yield* withWorkflow(
677 cluster,
678 SuspendFailureWorkflow.execute({ id: suspendId }, { discard: true })
679 )
680 yield* waitForSuspended(cluster, CompensationWorkflow, compensationExecutionId)
681 yield* waitForSuspended(cluster, SuspendFailureWorkflow, suspendExecutionId)
682
683 const compensationOwner = workflowOwner(cluster, compensationExecutionId)
684 const suspendOwner = workflowOwner(cluster, suspendExecutionId)
685 assert.isDefined(compensationOwner)
686 assert.isDefined(suspendOwner)
687 yield* Effect.forEach(
688 new Set([compensationOwner!, suspendOwner!]),
689 cluster.kill,
690 { discard: true }
691 )
692 yield* cluster.waitForStableAssignments()
693
694 const compensationToken = DurableDeferred.tokenFromExecutionId(CompensationGate, {
695 workflow: CompensationWorkflow,
696 executionId: compensationExecutionId
697 })
698 yield* withWorkflow(
699 cluster,
700 DurableDeferred.succeed(CompensationGate, {
701 token: compensationToken,
702 value: undefined
703 })
704 )
705 const compensationResult = yield* waitForComplete(
706 cluster,
707 CompensationWorkflow,
708 compensationExecutionId
709 )
710 assert(Exit.isFailure(compensationResult.exit))
711 const failure = Cause.findErrorOption(compensationResult.exit.cause)
712 assert(Option.isSome(failure))
713 assert.strictEqual(failure.value._tag, "ClusterIntegrationCompensationError")
714 assert.strictEqual(compensationRuns.get(compensationId), 1)
715
716 const suspendToken = DurableDeferred.tokenFromExecutionId(SuspendFailureGate, {
717 workflow: SuspendFailureWorkflow,
718 executionId: suspendExecutionId
719 })
720 yield* withWorkflow(
721 cluster,
722 DurableDeferred.succeed(SuspendFailureGate, {
723 token: suspendToken,
724 value: undefined
725 })
726 )
727 yield* cluster.waitUntil(
728 "SuspendOnFailure did not run after the workflow owner was lost",
729 Effect.sync(() => suspendFailures.has(suspendId))
730 )
731 yield* waitForSuspended(cluster, SuspendFailureWorkflow, suspendExecutionId)
732 }))
733
734 it.live(`${backend}: accepts a late durable-race completion across the suspend commit`, () =>
735 Effect.gen(function*() {
736 const id = `${backend}-race-suspend-commit`
737 raceBoundaryCommitStarted = Latch.makeUnsafe()
738 raceBoundaryCommit = Latch.makeUnsafe()
739 const cluster = yield* make({ backend, entities })
740 yield* cluster.start(3)
741 yield* cluster.waitForStableAssignments()
742 const executionId = yield* withWorkflow(
743 cluster,
744 RaceBoundaryWorkflow.execute({ id }, { discard: true })
745 )
746 yield* cluster.waitUntil(
747 "The durable race did not reach its suspend commit",
748 Effect.as(raceBoundaryCommitStarted.await, true)
749 )
750
751 const owner = workflowOwner(cluster, executionId)
752 assert.isDefined(owner)
753 const [controlId, controlOwner] = yield* findControlOnAnotherRunner(cluster, owner!)
754 const control = yield* cluster.getClient(DeferredControl)
755 const token = DurableDeferred.tokenFromExecutionId(RaceBoundaryGateB, {
756 workflow: RaceBoundaryWorkflow,
757 executionId
758 })
759 const completedBy = yield* control(controlId).CompleteRaceBoundaryDeferred({
760 token,
761 value: "late-winner"
762 })
763 assert.strictEqual(completedBy, `${controlOwner.address.host}:${controlOwner.address.port}`)
764
765 const killFiber = yield* cluster.kill(owner!).pipe(Effect.forkChild({ startImmediately: true }))
766 // The uninterruptible ensuring finalizer holds Scope.close here until
767 // the commit latch opens, keeping the owner blocked mid-commit.
768 raceBoundaryCommit.openUnsafe()
769 yield* Fiber.join(killFiber)
770 yield* cluster.waitForStableAssignments()
771
772 const result = yield* waitForComplete(cluster, RaceBoundaryWorkflow, executionId)
773 assert.deepStrictEqual(result.exit, Exit.succeed("late-winner"))
774 assert.isAtLeast(raceBoundaryRuns.get(id) ?? 0, 2)
775 }))
776
777 it.live(`${backend}: applies activity retry policy and preserves the exhausted error`, () =>
778 Effect.gen(function*() {
779 const cluster = yield* make({ backend, entities })
780 yield* cluster.start(3)
781 yield* cluster.waitForStableAssignments()
782 const successId = `${backend}-retry-success`
783 const failureId = `${backend}-retry-failure`
784
785 assert.strictEqual(
786 yield* withWorkflow(cluster, RetryWorkflow.execute({ id: successId, succeed: true })),
787 3
788 )
789 const error = yield* withWorkflow(
790 cluster,
791 RetryWorkflow.execute({ id: failureId, succeed: false })
792 ).pipe(Effect.flip)
793
794 assert.deepStrictEqual(retryAttempts.get(successId), [1, 2, 3])
795 assert.deepStrictEqual(retryAttempts.get(failureId), [1, 2, 3])
796 assert.strictEqual(error._tag, "ClusterIntegrationRetryError")
797 assert.strictEqual(error.attempt, 3)
798 }))
799
800 it.live(`${backend}: wakes a durable clock after a whole-cluster restart`, () =>
801 Effect.gen(function*() {
802 const cluster = yield* make({ backend, entities })
803 yield* cluster.start(3)
804 yield* cluster.waitForStableAssignments()
805 const started = yield* Clock.currentTimeMillis
806 const executionId = yield* withWorkflow(
807 cluster,
808 ClockWorkflow.execute({ id: `${backend}-clock` }, { discard: true })
809 )
810 yield* waitForSuspended(cluster, ClockWorkflow, executionId)
811 yield* restart(cluster)
812
813 const result = yield* waitForComplete(cluster, ClockWorkflow, executionId)
814 assert(Exit.isSuccess(result.exit))
815 assert.isAtLeast(result.exit.value - started, 900)
816 }))
817
818 it.live(`${backend}: persists queued work across restart and consumes it once`, () =>
819 Effect.gen(function*() {
820 const id = `${backend}-queue`
821 queueWorkerGate = Latch.makeUnsafe()
822 const cluster = yield* make({ backend, entities })
823 yield* cluster.start(3)
824 yield* cluster.waitForStableAssignments()
825 const executionId = yield* withWorkflow(cluster, QueueWorkflow.execute({ id }, { discard: true }))
826 yield* waitForSuspended(cluster, QueueWorkflow, executionId)
827 yield* restart(cluster)
828 queueWorkerGate.openUnsafe()
829
830 const result = yield* waitForComplete(cluster, QueueWorkflow, executionId)
831 assert.deepStrictEqual(result.exit, Exit.succeed(`processed:${id}`))
832 assert.strictEqual(queueRuns.get(id), 1)
833 }))
834
835 it.live(`${backend}: persists interruption across a whole-cluster restart`, () =>
836 Effect.gen(function*() {
837 const cluster = yield* make({ backend, entities })
838 yield* cluster.start(3)
839 yield* cluster.waitForStableAssignments()
840 const executionId = yield* withWorkflow(
841 cluster,
842 InterruptWorkflow.execute({ id: `${backend}-interrupt` }, { discard: true })
843 )
844 yield* waitForSuspended(cluster, InterruptWorkflow, executionId)
845 yield* withWorkflow(cluster, InterruptWorkflow.interrupt(executionId))
846 yield* restart(cluster)
847
848 const result = yield* waitForComplete(cluster, InterruptWorkflow, executionId)
849 assert(Exit.isFailure(result.exit))
850 assert.isTrue(Exit.hasInterrupts(result.exit))
851 assert.isTrue(Cause.hasInterrupts(result.exit.cause))
852 }))
853
854 it.live(`${backend}: replays a persisted activity handoff without leaking workflow resources`, () =>
855 Effect.gen(function*() {
856 const cluster = yield* make({ backend, entities: ShutdownActivityEntities })
857 yield* cluster.start(3)
858 yield* cluster.waitForStableAssignments()
859 for (
860 const [suffix, workflow] of [
861 ["default", ShutdownActivityWorkflow],
862 ["suspend-on-failure", ShutdownSuspendActivityWorkflow]
863 ] as const
864 ) {
865 const id = `${backend}-shutdown-activity-${suffix}`
866 resetActivityHandoffState(id)
867 const executionId = yield* withWorkflow(
868 cluster,
869 workflow.execute({ id }, { discard: true })
870 )
871 yield* cluster.waitUntil(
872 "The shutdown activity workflow did not start",
873 Effect.as(activityHandoffState.ready.await, true)
874 )
875
876 activityHandoffState.faultArmed = true
877 activityHandoffState.start.openUnsafe()
878
879 const result = yield* waitForComplete(cluster, workflow, executionId)
880 assert.isTrue(activityHandoffState.persisted.isOpen())
881 assert.deepStrictEqual(result.exit, Exit.succeed(`completed:${id}`))
882 assert.strictEqual(activityHandoffState.runs.get(id), 1)
883 assert.isFalse(activityHandoffState.compensations.has(id))
884 assert.deepStrictEqual(activityHandoffState.resourceEvents.get(id), [
885 "acquire",
886 "release",
887 "acquire",
888 "release"
889 ])
890 }
891 }))
892
893 it.live(`${backend}: preserves a safe interrupt during a persisted activity handoff`, () =>
894 Effect.gen(function*() {
895 const id = `${backend}-shutdown-activity-interrupt`
896 resetActivityHandoffState(id, { faultReleaseOpen: false })
897 const cluster = yield* make({ backend, entities: ShutdownActivityEntities })
898 yield* cluster.start(3)
899 yield* cluster.waitForStableAssignments()
900 const executionId = yield* withWorkflow(
901 cluster,
902 ShutdownActivityWorkflow.execute({ id }, { discard: true })
903 )
904 yield* cluster.waitUntil(
905 "The shutdown activity workflow did not start",
906 Effect.as(activityHandoffState.ready.await, true)
907 )
908
909 activityHandoffState.faultArmed = true
910 yield* withWorkflow(cluster, ShutdownActivityWorkflow.interrupt(executionId))
911 activityHandoffState.start.openUnsafe()
912 yield* cluster.waitUntil(
913 "The shutdown activity request was not persisted",
914 Effect.as(activityHandoffState.persisted.await, true)
915 )
916 yield* cluster.waitUntil(
917 "The shutdown activity request was not processed",
918 Effect.sync(() => activityHandoffState.runs.get(id) === 1)
919 )
920 activityHandoffState.faultRelease.openUnsafe()
921
922 const result = yield* waitForComplete(cluster, ShutdownActivityWorkflow, executionId)
923 assert.isTrue(Exit.isFailure(result.exit))
924 assert.isTrue(Exit.hasInterrupts(result.exit))
925 assert.isTrue(activityHandoffState.compensations.has(id))
926 }).pipe(
927 Effect.ensuring(Effect.sync(() => {
928 activityHandoffState.faultRelease.openUnsafe()
929 }))
930 ))
931
932 it.live(`${backend}: recovers a persisted activity across an actual owner handoff`, () =>
933 Effect.gen(function*() {
934 const id = `${backend}-shutdown-activity-handoff`
935 resetActivityHandoffState(id)
936 const cluster = yield* make({
937 backend,
938 config: { entityTerminationTimeout: 100 },
939 entities: ShutdownActivityWorkflowLayer.pipe(
940 Layer.provide(ClusterWorkflowEngine.layer),
941 Layer.orDie
942 )
943 })
944 yield* cluster.start(3)
945 yield* cluster.waitForStableAssignments()
946 const executionId = yield* withWorkflow(
947 cluster,
948 ShutdownActivityWorkflow.execute({ id }, { discard: true })
949 )
950 yield* cluster.waitUntil(
951 "The shutdown activity workflow did not start",
952 Effect.as(activityHandoffState.ready.await, true)
953 )
954
955 const owner = workflowOwner(cluster, executionId)
956 assert.isDefined(owner)
957 const stopping = yield* cluster.stop(owner!).pipe(Effect.forkChild({ startImmediately: true }))
958 activityHandoffState.start.openUnsafe()
959 yield* cluster.waitUntil(
960 "The shutdown activity workflow was not handed to another runner",
961 Effect.sync(() => {
962 const nextOwner = workflowOwner(cluster, executionId)
963 return nextOwner !== undefined && nextOwner !== owner
964 })
965 )
966
967 const result = yield* waitForComplete(cluster, ShutdownActivityWorkflow, executionId)
968 yield* Fiber.join(stopping)
969 assert.deepStrictEqual(result.exit, Exit.succeed(`completed:${id}`))
970 assert.strictEqual(activityHandoffState.runs.get(id), 1)
971 assert.strictEqual((yield* cluster.messageCounts()).unprocessed, 0)
972 }))
973 }
974})