1/**
2 * The cluster workflow engine runs durable workflows on top of cluster sharding
3 * and message storage. It adapts `WorkflowEngine.WorkflowEngine` so workflow
4 * executions, activities, deferred completions, resumes, interrupts, and durable
5 * clock wakeups are represented as persisted cluster entity messages.
6 *
7 * @stability unstable
8 * @since 4.0.0
9 */
10import * as Context from "../Context.ts"
11import * as DateTime from "../DateTime.ts"
12import * as Deferred from "../Deferred.ts"
13import * as Duration from "../Duration.ts"
14import * as Effect from "../Effect.ts"
15import * as Exit from "../Exit.ts"
16import * as Fiber from "../Fiber.ts"
17import * as Headers from "../http/Headers.ts"
18import * as Latch from "../Latch.ts"
19import * as Layer from "../Layer.ts"
20import * as Option from "../Option.ts"
21import * as PrimaryKey from "../PrimaryKey.ts"
22import * as RcMap from "../RcMap.ts"
23import type * as Record from "../Record.ts"
24import * as Rpc from "../rpc/Rpc.ts"
25import { ClientAbort } from "../rpc/RpcSchema.ts"
26import * as Schedule from "../Schedule.ts"
27import * as Schema from "../Schema.ts"
28import type * as Scope from "../Scope.ts"
29import * as Semaphore from "../Semaphore.ts"
30import * as Activity from "../workflow/Activity.ts"
31import * as DurableClock from "../workflow/DurableClock.ts"
32import * as DurableDeferred from "../workflow/DurableDeferred.ts"
33import * as Workflow from "../workflow/Workflow.ts"
34import * as WorkflowEngine from "../workflow/WorkflowEngine.ts"
35import * as ClusterSchema from "./ClusterSchema.ts"
36import * as DeliverAt from "./DeliverAt.ts"
37import * as Entity from "./Entity.ts"
38import * as EntityAddress from "./EntityAddress.ts"
39import * as EntityId from "./EntityId.ts"
40import * as EntityType from "./EntityType.ts"
41import * as Envelope from "./Envelope.ts"
42import * as ClusterAbandon from "./internal/clusterAbandon.ts"
43import { CurrentActivationScope } from "./internal/entityActivation.ts"
44import * as Message from "./Message.ts"
45import { MessageStorage } from "./MessageStorage.ts"
46import type { WithExitEncoded } from "./Reply.ts"
47import * as Reply from "./Reply.ts"
48import * as Sharding from "./Sharding.ts"
49import * as Snowflake from "./Snowflake.ts"
50
51/**
52 * Creates a `WorkflowEngine` implementation backed by cluster sharding and
53 * message storage.
54 *
55 * **Details**
56 *
57 * Workflow executions, activities, deferred completions, resumes, interrupts,
58 * and durable clock wakeups are coordinated through persisted cluster entities.
59 *
60 * @stability unstable
61 * @category constructors
62 * @since 4.0.0
63 */
64export const make = Effect.gen(function*() {
65 const sharding = yield* Sharding.Sharding
66 const storage = yield* MessageStorage
67
68 const workflows = new Map<string, Workflow.Any>()
69 const entities = new Map<
70 string,
71 Entity.Entity<
72 string,
73 | Rpc.Rpc<
74 "run",
75 Schema.Struct<
76 Record<
77 typeof payloadParentKey,
78 Schema.optional<
79 Schema.Struct<{
80 workflowName: typeof Schema.String
81 executionId: typeof Schema.String
82 }>
83 >
84 >
85 >,
86 Schema.declare<Workflow.Result<any, any>>
87 >
88 | Rpc.Rpc<"deferred", Schema.Struct<{ name: typeof Schema.String; exit: typeof ExitUnknown }>, typeof ExitUnknown>
89 | Rpc.Rpc<
90 "activity",
91 Schema.Struct<
92 { name: typeof Schema.String; attempt: typeof Schema.Int; withTransaction: typeof Schema.Boolean }
93 >,
94 Schema.declare<Workflow.Result<any, any>>
95 >
96 | typeof ResumeRpc
97 >
98 >()
99 const partialEntities = new Map<
100 string,
101 Entity.Entity<
102 string,
103 | Rpc.Rpc<"deferred", Schema.Struct<{ name: typeof Schema.String; exit: typeof ExitUnknown }>, typeof ExitUnknown>
104 | Rpc.Rpc<
105 "activity",
106 Schema.Struct<{ name: typeof Schema.String; attempt: typeof Schema.Int }>,
107 Schema.declare<Workflow.Result<any, any>>
108 >
109 | typeof ResumeRpc
110 >
111 >()
112 const ensureEntity = (workflow: Workflow.Any) => {
113 let entity = entities.get(workflow._tag)
114 if (!entity) {
115 entity = makeWorkflowEntity(workflow) as any
116 workflows.set(workflow._tag, workflow)
117 entities.set(workflow._tag, entity as any)
118 }
119 return entity!
120 }
121 const ensurePartialEntity = (workflowName: string) => {
122 let entity = partialEntities.get(workflowName)
123 if (!entity) {
124 entity = makePartialWorkflowEntity(workflowName) as any
125 partialEntities.set(workflowName, entity as any)
126 }
127 return entity!
128 }
129
130 const activities = new Map<string, {
131 readonly activity: Activity.Any
132 readonly context: Context.Context<any>
133 }>()
134 const interruptedActivities = new Set<string>()
135 const activityLatches = new Map<string, Latch.Latch>()
136 const deferredState = WorkflowEngine.makeDeferredState()
137 // Rebuilds share an activation; overlapping activations have separate results.
138 // Weak keys do not retain scopes, but contexts captured by clients, activities,
139 // or finalizers can keep a closed scope (and its results) reachable until later.
140 const pendingByActivation = new WeakMap<Scope.Scope, {
141 readonly workflowName: string
142 readonly executionId: string
143 readonly results: Map<string, Exit.Exit<unknown, unknown>>
144 }>()
145 const clients = yield* RcMap.make({
146 lookup: Effect.fnUntraced(function*(workflowName: string) {
147 const entity = entities.get(workflowName)
148 if (!entity) {
149 return yield* Effect.die(`Workflow ${workflowName} not registered`)
150 }
151 yield* RcMap.invalidate(clientsPartial, workflowName)
152 return yield* entity.client
153 }),
154 idleTimeToLive: "5 minutes"
155 })
156 const clientsPartial = yield* RcMap.make({
157 lookup: Effect.fnUntraced(function*(workflowName: string) {
158 const entity = entities.get(workflowName) ?? ensurePartialEntity(workflowName)
159 return yield* entity.client
160 }),
161 idleTimeToLive: "5 minutes"
162 })
163 const entityAddressFor = (options: {
164 readonly workflow: Workflow.Any
165 readonly entityType: string
166 readonly executionId: string
167 }) => {
168 const shardGroup = Context.get(options.workflow.annotations, ClusterSchema.ShardGroup)(
169 options.executionId as EntityId.EntityId
170 )
171 const entityId = EntityId.make(options.executionId)
172 return EntityAddress.make({
173 entityType: EntityType.make(options.entityType),
174 entityId,
175 shardId: sharding.getShardId(entityId, shardGroup)
176 })
177 }
178
179 const sendDiscard = Effect.fnUntraced(function*(options: {
180 readonly rpc: Rpc.AnyWithProps
181 readonly address: EntityAddress.EntityAddress
182 readonly payload: unknown
183 }) {
184 const payload = (options.rpc.payloadSchema as any).make(options.payload)
185 const span = yield* Effect.orDie(Effect.currentSpan)
186 const envelope = Envelope.makeRequest<any>({
187 requestId: yield* sharding.getSnowflake,
188 address: options.address,
189 tag: options.rpc._tag as any,
190 payload,
191 headers: Headers.empty,
192 traceId: span.traceId,
193 spanId: span.spanId,
194 sampled: span.sampled
195 })
196 yield* sharding.sendOutgoing(
197 new Message.OutgoingRequest({
198 envelope,
199 context: Context.empty() as Context.Context<any>,
200 lastReceivedReply: Option.none(),
201 rpc: options.rpc,
202 respond: () => Effect.void,
203 annotations: Context.get(options.rpc.annotations, ClusterSchema.Dynamic)(
204 options.rpc.annotations,
205 envelope as any
206 )
207 }),
208 true
209 )
210 })
211
212 const requestIdFor = Effect.fnUntraced(function*(options: {
213 readonly workflow: Workflow.Any
214 readonly entityType: string
215 readonly executionId: string
216 readonly tag: string
217 readonly id: string
218 }) {
219 const address = entityAddressFor(options)
220 return yield* storage.requestIdForPrimaryKey({ address, tag: options.tag, id: options.id })
221 })
222
223 const replyForRequestId = Effect.fnUntraced(function*(requestId: Snowflake.Snowflake) {
224 const replies = yield* storage.repliesForUnfiltered([requestId])
225 const last = replies[replies.length - 1]
226 if (last && last._tag === "WithExit") {
227 return Option.some(last as WithExitEncoded<Workflow.ResultEncoded<any, any>>)
228 }
229 return Option.none<WithExitEncoded<Workflow.ResultEncoded<any, any>>>()
230 })
231
232 const requestReply = Effect.fnUntraced(function*(options: {
233 readonly workflow: Workflow.Any
234 readonly entityType: string
235 readonly executionId: string
236 readonly tag: string
237 readonly id: string
238 }) {
239 const requestId = yield* requestIdFor(options)
240 if (Option.isNone(requestId)) {
241 return Option.none<WithExitEncoded<Workflow.ResultEncoded<any, any>>>()
242 }
243 return yield* replyForRequestId(requestId.value)
244 })
245
246 const resetActivityAttempt = Effect.fnUntraced(
247 function*(options: {
248 readonly workflow: Workflow.Any
249 readonly executionId: string
250 readonly activity: Activity.Any
251 readonly attempt: number
252 }) {
253 const requestId = yield* requestIdFor({
254 workflow: options.workflow,
255 entityType: `Workflow/${options.workflow._tag}`,
256 executionId: options.executionId,
257 tag: "activity",
258 id: activityPrimaryKey(options.activity.name, options.attempt)
259 })
260 if (Option.isNone(requestId)) return
261 yield* sharding.reset(requestId.value)
262 },
263 Effect.retry({
264 times: 3,
265 schedule: Schedule.exponential(250)
266 }),
267 Effect.orDie
268 )
269
270 const clearClock = Effect.fnUntraced(function*(options: {
271 readonly workflow: Workflow.Any
272 readonly executionId: string
273 }) {
274 const shardGroup = Context.get(options.workflow.annotations, ClusterSchema.ShardGroup)(
275 options.executionId as EntityId.EntityId
276 )
277 const entityId = EntityId.make(options.executionId)
278 const shardId = sharding.getShardId(entityId, shardGroup)
279 const clockAddress = EntityAddress.make({
280 entityType: ClockEntity.type,
281 entityId,
282 shardId
283 })
284 yield* storage.clearAddress(clockAddress)
285 })
286
287 const resume = Effect.fnUntraced(function*(workflow: Workflow.Any, executionId: string) {
288 const maybeReply = yield* requestReply({
289 workflow,
290 entityType: `Workflow/${workflow._tag}`,
291 executionId,
292 tag: "run",
293 id: ""
294 })
295
296 const maybeSuspended = Option.filter(
297 maybeReply,
298 (reply) => reply.exit._tag === "Success" && reply.exit.value._tag === "Suspended"
299 )
300
301 if (Option.isNone(maybeSuspended)) return
302 if (
303 !(yield* sharding.reset(
304 Snowflake.Snowflake(maybeSuspended.value.requestId),
305 { expectedReplyId: Snowflake.Snowflake(maybeSuspended.value.id) }
306 ))
307 ) {
308 return yield* Effect.fail(new Error("Failed to reset workflow run"))
309 }
310 yield* sharding.pollStorage
311 })
312
313 const sendResumeParent = Effect.fnUntraced(function*(options: {
314 readonly workflowName: string
315 readonly executionId: string
316 }, childExecutionId: string) {
317 const requestId = yield* requestIdFor({
318 workflow: workflows.get(options.workflowName)!,
319 entityType: `Workflow/${options.workflowName}`,
320 executionId: options.executionId,
321 tag: "resume",
322 id: childExecutionId
323 })
324 if (Option.isNone(requestId)) {
325 const client = (yield* RcMap.get(clientsPartial, options.workflowName))(options.executionId)
326 return yield* client.resume({ childExecutionId }, { discard: true })
327 }
328 const reply = yield* replyForRequestId(requestId.value)
329 if (Option.isNone(reply)) return
330 yield* sharding.reset(requestId.value)
331 }, Effect.scoped)
332
333 // A child or deferred may complete while a run is still unwinding. Subscribe
334 // to the run's reply before reading storage so the wake cannot fall between
335 // the run's last read and its suspended reply.
336 const waitForRunReply = Effect.fnUntraced(function*(workflow: Workflow.Any, request: Entity.Request<any>) {
337 const waiter = yield* storage.registerReplyHandler(
338 new Message.OutgoingRequest<any>({
339 envelope: request,
340 context: Context.empty() as Context.Context<any>,
341 rpc: ensureEntity(workflow).protocol.requests.get("run")!,
342 lastReceivedReply: Option.none(),
343 respond: () => Effect.void,
344 annotations: Context.empty()
345 })
346 ).pipe(Effect.forkScoped({ startImmediately: true }))
347 const reply = yield* replyForRequestId(request.requestId)
348 if (Option.isSome(reply)) return
349 yield* Fiber.join(waiter).pipe(
350 Effect.catchTag("EntityNotAssignedToRunner", () => ClusterAbandon.interrupt),
351 Effect.catchCause((cause) => ClusterAbandon.isCause(cause) ? ClusterAbandon.interrupt : Effect.failCause(cause))
352 )
353 }, Effect.scoped)
354
355 const interrupt = Effect.fnUntraced(
356 function*(workflow: Workflow.Any, executionId: string) {
357 ensureEntity(workflow)
358 const requestId = yield* requestIdFor({
359 workflow,
360 entityType: `Workflow/${workflow._tag}`,
361 executionId,
362 tag: "run",
363 id: ""
364 })
365 if (Option.isNone(requestId)) {
366 return Option.none()
367 }
368 const reply = yield* replyForRequestId(requestId.value)
369
370 const nonSuspendedReply = Option.filter(
371 reply,
372 (reply) => reply.exit._tag !== "Success" || reply.exit.value._tag !== "Suspended"
373 )
374 if (Option.isSome(nonSuspendedReply)) {
375 return Option.none()
376 }
377
378 yield* engine.deferredDone(InterruptSignal, {
379 workflowName: workflow._tag,
380 executionId,
381 deferredName: InterruptSignal.name,
382 exit: Exit.void
383 })
384
385 return requestId
386 },
387 Effect.retry({
388 while: (e) => e._tag === "PersistenceError",
389 times: 3,
390 schedule: Schedule.exponential(250)
391 }),
392 Effect.orDie
393 )
394
395 const engine = WorkflowEngine.makeUnsafe({
396 register: (workflow, execute) =>
397 Effect.suspend(() => {
398 const existing = workflows.get(workflow._tag)
399 const registration = sharding.registerEntity(
400 ensureEntity(workflow),
401 Effect.gen(function*() {
402 const address = yield* Entity.CurrentAddress
403 const executionId = address.entityId
404 const activation = yield* CurrentActivationScope
405 let pending = pendingByActivation.get(activation)
406 if (!pending) {
407 pending = { workflowName: workflow._tag, executionId, results: new Map() }
408 pendingByActivation.set(activation, pending)
409 }
410 // Retain awaited names through reply persistence, including suspension.
411 // Replays reuse the request id but track a fresh set of awaits.
412 let currentRun: {
413 readonly request: Entity.Request<any>
414 readonly instance: WorkflowEngine.WorkflowInstance["Service"]
415 } | undefined
416 // Concurrent wakes share one wait for the current run to publish its reply.
417 const resumeGate = Semaphore.makeUnsafe(1)
418 const resumeCurrentRun = Effect.suspend(() =>
419 resumeGate.withPermitsIfAvailable(1)(
420 currentRun ? waitForRunReply(workflow, currentRun.request) : Effect.void
421 )
422 ).pipe(
423 // Release the gate before reset can start another run, so later
424 // completions can wake that replay.
425 Effect.flatMap((waited) => Option.isSome(waited) ? resume(workflow, executionId) : Effect.void),
426 ensureSuccess
427 )
428 return {
429 run: (request: Entity.Request<any>) => {
430 const instance = WorkflowEngine.WorkflowInstance.initial(workflow, executionId)
431 currentRun = { request, instance }
432 const parent = (request.payload as any)[payloadParentKey] as
433 | { workflowName: string; executionId: string }
434 | undefined
435 return execute(workflow.payloadSchema.make(request.payload) as object, executionId).pipe(
436 Effect.onExit((exit) => {
437 const suspendOnFailure = Context.get(workflow.annotations, Workflow.SuspendOnFailure)
438 // An annotated cluster abandonment means this owner can no
439 // longer finish the run. Do not resume the parent, but
440 // still allow a pending durable interrupt to take effect.
441 instance.abandoned ||= exit._tag === "Failure" && ClusterAbandon.isCause(exit.cause)
442 if (instance.abandoned) {
443 instance.suspended = false
444 }
445 if (
446 !instance.suspended &&
447 !instance.abandoned &&
448 !(suspendOnFailure && exit._tag === "Failure")
449 ) {
450 return parent ? ensureSuccess(sendResumeParent(parent, executionId)) : Effect.void
451 }
452 return engine.deferredResult(InterruptSignal).pipe(
453 Effect.flatMap((maybeExit) => {
454 if (Option.isNone(maybeExit)) {
455 return Effect.void
456 }
457 instance.suspended = false
458 instance.interrupted = true
459 return Effect.andThen(
460 Effect.ignore(clearClock({ workflow, executionId })),
461 Effect.withFiber<void>((fiber) => Effect.interruptible(Fiber.interrupt(fiber)))
462 )
463 }),
464 Effect.orDie
465 )
466 }),
467 Workflow.intoResult,
468 (effect) => deferredState.trackRun(instance, effect),
469 // Request middleware and local callers may carry another entity's context.
470 Effect.provideService(CurrentActivationScope, activation)
471 ) as any
472 },
473
474 activity(request: Entity.Request<any>) {
475 const payload = request.payload as { name: string; attempt: number }
476 const activityId = `${executionId}/${payload.name}`
477 const instance = WorkflowEngine.WorkflowInstance.initial(workflow, executionId)
478 interruptedActivities.delete(activityId)
479 return Effect.gen(function*() {
480 let entry = activities.get(activityId)
481 while (!entry) {
482 const latch = Latch.makeUnsafe()
483 activityLatches.set(activityId, latch)
484 yield* latch.await
485 entry = activities.get(activityId)
486 }
487 const context = entry.context.pipe(
488 Context.add(WorkflowEngine.WorkflowInstance, instance),
489 Context.add(CurrentActivationScope, activation),
490 Context.add(Activity.CurrentAttempt, payload.attempt)
491 )
492 return yield* entry.activity.executeEncoded.pipe(
493 Effect.provideContext(context)
494 )
495 }).pipe(
496 Workflow.intoResult,
497 Effect.catchCause((cause) => {
498 // we only want to store interrupts as suspends when the
499 // client requested it
500 const suspend = cause.reasons.some((f) =>
501 f._tag === "Interrupt" && f.annotations.has(ClientAbort.key)
502 )
503 if (suspend) {
504 interruptedActivities.add(activityId)
505 return Effect.succeed(new Workflow.Suspended({}))
506 }
507 return Effect.failCause(cause)
508 }),
509 Effect.provideService(WorkflowEngine.WorkflowInstance, instance),
510 Effect.provideService(Activity.CurrentAttempt, payload.attempt),
511 Effect.ensuring(Effect.sync(() => {
512 activities.delete(activityId)
513 })),
514 Rpc.wrap({ fork: true, uninterruptible: true })
515 )
516 },
517
518 deferred: (request: Entity.Request<any>) => {
519 const payload = request.payload as any
520 pending.results.set(payload.name, payload.exit)
521 // Await registers before reading the cache. If this run has not
522 // registered yet, a later await will see the result just stored.
523 // Check when the wake runs and skip preemption too: interrupting
524 // a later await would require waiting for its reply before reset.
525 // A run completing its own deferred also reads it from the cache
526 // when it awaits later, so that completion never wakes the run.
527 const wake = Effect.suspend(() =>
528 currentRun && !currentRun.instance.awaitedDeferreds.has(payload.name)
529 ? currentRun.instance.completedDeferreds.has(payload.name)
530 ? Effect.void
531 : ensureSuccess(resume(workflow, executionId))
532 : deferredState.deferredDone(executionId, payload.name).pipe(Effect.andThen(resumeCurrentRun))
533 )
534 // An asynchronous reply releases the RPC concurrency permit while
535 // the wake waits; the pending request still occupies mailbox capacity.
536 // The activation owns the wake so it survives handler rebuilds,
537 // but activation shutdown still interrupts it for durable redelivery.
538 const reply = Deferred.makeUnsafe<Exit.Exit<unknown, unknown>>()
539 return wake.pipe(
540 Effect.as(payload.exit),
541 Deferred.into(reply),
542 Effect.forkIn(activation, { startImmediately: true }),
543 Effect.as(reply)
544 )
545 },
546
547 resume: () => resumeCurrentRun.pipe(Rpc.wrap({ fork: true, uninterruptible: true }))
548 }
549 }),
550 // Alongside the run, let deferred handlers briefly acquire a permit to
551 // fork their wake and return an asynchronous reply.
552 { concurrency: 2, maxIdleTime: entityMaxIdleTime }
553 ) as Effect.Effect<void, never, Scope.Scope>
554 if (existing === undefined || existing === workflow) return registration
555 return Effect.logWarning(
556 `Workflow "${workflow._tag}" is already registered with payload shape ${
557 payloadShape(
558 existing
559 )
560 }; ignoring duplicate definition with payload shape ${payloadShape(workflow)}`
561 ).pipe(Effect.andThen(registration))
562 }),
563
564 execute: (workflow, { discard, executionId, parent, payload }) => {
565 ensureEntity(workflow)
566 return RcMap.get(clients, workflow._tag).pipe(
567 Effect.flatMap((make) =>
568 make(executionId).run(
569 parent ?
570 {
571 ...payload,
572 [payloadParentKey]: { workflowName: parent.workflow._tag, executionId: parent.executionId }
573 } :
574 payload,
575 { discard }
576 )
577 ),
578 Effect.orDie,
579 Effect.scoped
580 )
581 },
582
583 poll: Effect.fnUntraced(function*(workflow, executionId) {
584 const entity = ensureEntity(workflow)
585 const exitSchema = Schema.toCodecJson(Rpc.exitSchema(entity.protocol.requests.get("run")!))
586 const reply = yield* requestReply({
587 workflow,
588 entityType: `Workflow/${workflow._tag}`,
589 executionId,
590 tag: "run",
591 id: ""
592 })
593 if (Option.isNone(reply)) return Option.none()
594 const exit = yield* (Schema.decodeUnknownEffect(exitSchema)(reply.value.exit) as Effect.Effect<
595 Exit.Exit<any, any>,
596 Schema.SchemaError
597 >)
598 return Option.some(yield* exit)
599 }, Effect.orDie),
600
601 interrupt: (workflow, executionId) => Effect.asVoid(interrupt(workflow, executionId)),
602 interruptUnsafe: Effect.fnUntraced(function*(workflow, executionId) {
603 const requestId = yield* interrupt(workflow, executionId)
604 if (Option.isNone(requestId)) return
605 const entity = ensureEntity(workflow)
606 const runRpc = entity.protocol.requests.get("run")!
607 yield* Effect.orDie(sharding.sendOutgoing(
608 new Message.OutgoingEnvelope({
609 rpc: runRpc,
610 envelope: new Envelope.Interrupt({
611 id: yield* sharding.getSnowflake,
612 address: entityAddressFor({
613 workflow,
614 entityType: `Workflow/${workflow._tag}`,
615 executionId
616 }),
617 requestId: requestId.value
618 })
619 }),
620 false
621 ))
622 }),
623
624 resume: (workflow, executionId) => ensureSuccess(resume(workflow, executionId)),
625
626 activityExecute: Effect.fnUntraced(
627 function*(activity, attempt) {
628 const services = yield* Effect.context<WorkflowEngine.WorkflowInstance>()
629 const instance = Context.get(services, WorkflowEngine.WorkflowInstance)
630 yield* Effect.annotateCurrentSpan("executionId", instance.executionId)
631 const activityId = `${instance.executionId}/${activity.name}`
632 const client = (yield* RcMap.get(clientsPartial, instance.workflow._tag))(instance.executionId)
633 while (true) {
634 if (!activities.has(activityId)) {
635 activities.set(activityId, { activity, context: services })
636 const latch = activityLatches.get(activityId)
637 if (latch) {
638 yield* latch.open
639 activityLatches.delete(activityId)
640 }
641 }
642 const result = yield* Effect.orDie(
643 client.activity({
644 name: activity.name,
645 attempt,
646 withTransaction: Context.get(activity.annotations, ClusterSchema.WithTransaction)
647 })
648 )
649 // If the activity has suspended and did not execute, we need to resume
650 // it by resetting the attempt and re-executing.
651 if (result._tag === "Suspended" && (activities.has(activityId) || interruptedActivities.has(activityId))) {
652 yield* resetActivityAttempt({
653 workflow: instance.workflow,
654 executionId: instance.executionId,
655 activity,
656 attempt
657 })
658 continue
659 }
660 activities.delete(activityId)
661 return result
662 }
663 },
664 Effect.scoped
665 ),
666
667 deferredResult: (deferred) =>
668 WorkflowEngine.WorkflowInstance.pipe(
669 Effect.flatMap(Effect.fnUntraced(function*(instance) {
670 const activation = yield* Effect.serviceOption(CurrentActivationScope)
671 const pending = Option.isSome(activation) ? pendingByActivation.get(activation.value) : undefined
672 // Captured contexts can cross workflow boundaries. Never use another
673 // execution's completion, even if its deferred has the same name.
674 const exit = pending?.workflowName === instance.workflow._tag && pending.executionId === instance.executionId
675 ? pending.results.get(deferred.name)
676 : undefined
677 if (exit) {
678 return Option.some(exit)
679 }
680 return yield* requestReply({
681 workflow: instance.workflow,
682 entityType: `Workflow/${instance.workflow._tag}`,
683 executionId: instance.executionId,
684 tag: "deferred",
685 id: deferred.name
686 }).pipe(
687 Effect.map((reply) => {
688 if (Option.isNone(reply)) {
689 return Option.none<Exit.Exit<unknown, unknown>>()
690 }
691 const decoded = decodeDeferredWithExit(reply.value as any)
692 return Option.some(
693 decoded.exit._tag === "Success"
694 ? decoded.exit.value
695 : decoded.exit
696 )
697 })
698 )
699 })),
700 Effect.retry({
701 while: (e) => e._tag === "PersistenceError",
702 times: 3,
703 schedule: Schedule.exponential(250)
704 }),
705 Effect.orDie
706 ),
707
708 deferredDone: Effect.fnUntraced(
709 function*({ deferredName, executionId, exit, workflowName }) {
710 const instance = yield* Effect.serviceOption(WorkflowEngine.WorkflowInstance)
711 // Activities run with their own instance, so their completions still
712 // wake a run that suspended before reaching the await. An interrupt
713 // must wake the run even when it signals itself.
714 if (
715 Option.isSome(instance) && instance.value.executionId === executionId &&
716 deferredName !== InterruptSignal.name
717 ) {
718 instance.value.completedDeferreds.add(deferredName)
719 }
720 const workflow = workflows.get(workflowName)
721 if (workflow) {
722 return yield* Effect.orDie(sendDiscard({
723 rpc: DeferredRpc,
724 address: entityAddressFor({
725 workflow,
726 entityType: `Workflow/${workflowName}`,
727 executionId
728 }),
729 payload: {
730 name: deferredName,
731 exit
732 }
733 }))
734 }
735 const client = yield* RcMap.get(clientsPartial, workflowName)
736 return yield* Effect.orDie(
737 client(executionId).deferred({
738 name: deferredName,
739 exit
740 }, { discard: true })
741 )
742 },
743 Effect.scoped
744 ),
745
746 scheduleClock(workflow, options) {
747 return DateTime.now.pipe(
748 Effect.flatMap((now) =>
749 sendDiscard({
750 rpc: ClockRpc,
751 address: entityAddressFor({
752 workflow,
753 entityType: ClockEntity.type,
754 executionId: options.executionId
755 }),
756 payload: {
757 name: options.clock.name,
758 workflowName: workflow._tag,
759 wakeUp: DateTime.mapEpochMillis(
760 DateTime.addDuration(now, options.clock.duration),
761 Math.ceil
762 )
763 }
764 })
765 ),
766 Effect.orDie
767 )
768 }
769 })
770
771 return engine
772})
773
774const retryPolicy = Schedule.min([
775 Schedule.exponential(200, 1.5),
776 Schedule.spaced("1 minute")
777])
778
779const ensureSuccess = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
780 effect.pipe(
781 Effect.sandbox,
782 Effect.retry(retryPolicy),
783 Effect.orDie
784 )
785
786const AnyOrVoid = Schema.Union([Schema.Undefined, Schema.Any])
787const ExitUnknown = Schema.Exit(AnyOrVoid, AnyOrVoid, Schema.Any)
788
789const ActivityRpc = Rpc.make("activity", {
790 payload: {
791 name: Schema.String,
792 attempt: Schema.Int,
793 withTransaction: Schema.Boolean.pipe(
794 Schema.withDecodingDefault(Effect.succeed(false))
795 )
796 },
797 primaryKey: ({ attempt, name }) => activityPrimaryKey(name, attempt),
798 success: Workflow.Result({
799 success: AnyOrVoid,
800 error: AnyOrVoid
801 })
802})
803 .annotate(ClusterSchema.Persisted, true)
804 .annotate(
805 ClusterSchema.Dynamic,
806 (annotations, request) =>
807 (request.payload as any).withTransaction
808 ? Context.add(annotations, ClusterSchema.WithTransaction, true)
809 : annotations
810 )
811
812const DeferredRpc = Rpc.make("deferred", {
813 payload: {
814 name: Schema.String,
815 exit: ExitUnknown
816 },
817 primaryKey: ({ name }) => name,
818 success: ExitUnknown
819})
820 .annotate(ClusterSchema.Persisted, true)
821 .annotate(ClusterSchema.Uninterruptible, true)
822
823const decodeDeferredWithExit = Schema.decodeSync(Schema.toCodecJson(Reply.WithExit.schema(DeferredRpc)))
824
825const ResumeRpc = Rpc.make("resume", {
826 // Older persisted resume envelopes have an empty payload and use the empty key.
827 payload: { childExecutionId: Schema.optional(Schema.String) },
828 // Different children must not share an in-flight wake: the parent may have
829 // already started another run when the next child finishes.
830 primaryKey: ({ childExecutionId }) => childExecutionId ?? ""
831})
832 .annotate(ClusterSchema.Persisted, true)
833 .annotate(ClusterSchema.Uninterruptible, "server")
834
835const payloadParentKey = "~effect/cluster/ClusterWorkflowEngine/payloadParentKey"
836
837// Workflow state is durable, so an idle entity (completed or suspended) can
838// be released quickly and is rebuilt from storage when the next message
839// arrives. This keeps completed and suspended executions from holding
840// `maxResidentEntities` slots for the full `entityMaxIdleTime`.
841const entityMaxIdleTime = Duration.seconds(10)
842
843const makeWorkflowEntity = (workflow: Workflow.Any) =>
844 Entity.make(`Workflow/${workflow._tag}`, [
845 Rpc.make("run", {
846 payload: {
847 ...workflow.payloadSchema.fields,
848 [payloadParentKey]: Schema.optional(Schema.Struct({
849 workflowName: Schema.String,
850 executionId: Schema.String
851 }))
852 },
853 primaryKey: () => "",
854 success: Workflow.Result({
855 success: workflow.successSchema,
856 error: workflow.errorSchema
857 })
858 })
859 .annotate(ClusterSchema.Persisted, true)
860 .annotate(ClusterSchema.Uninterruptible, true),
861
862 DeferredRpc,
863 ResumeRpc,
864 ActivityRpc
865 ]).annotateMerge(workflow.annotations)
866
867const makePartialWorkflowEntity = (workflowName: string) =>
868 Entity.make(`Workflow/${workflowName}`, [
869 DeferredRpc,
870 ResumeRpc,
871 ActivityRpc
872 ])
873
874const activityPrimaryKey = (activity: string, attempt: number) => `${activity}/${attempt}`
875
876class ClockPayload extends Schema.Class<ClockPayload>(`Workflow/DurableClock/Run`)({
877 name: Schema.String,
878 workflowName: Schema.String,
879 wakeUp: Schema.DateTimeUtcFromMillis
880}) {
881 [PrimaryKey.symbol]() {
882 return this.name
883 }
884 [DeliverAt.symbol]() {
885 return this.wakeUp
886 }
887}
888
889const ClockRpc = Rpc.make("run", { payload: ClockPayload })
890 .annotate(ClusterSchema.Persisted, true)
891 .annotate(ClusterSchema.Uninterruptible, true)
892
893const ClockEntity = Entity.make("Workflow/-/DurableClock", [
894 ClockRpc
895])
896
897const ClockEntityLayer = ClockEntity.toLayer(
898 Effect.gen(function*() {
899 const engine = yield* WorkflowEngine.WorkflowEngine
900 const address = yield* Entity.CurrentAddress
901 const executionId = address.entityId
902 return {
903 run(request) {
904 const deferred = DurableClock.make({ name: request.payload.name, duration: Duration.zero }).deferred
905 return ensureSuccess(engine.deferredDone(deferred, {
906 workflowName: request.payload.workflowName,
907 executionId,
908 deferredName: deferred.name,
909 exit: Exit.void
910 }))
911 }
912 }
913 }),
914 { maxIdleTime: entityMaxIdleTime }
915)
916
917const InterruptSignal = DurableDeferred.make("Workflow/InterruptSignal")
918
919/**
920 * Layer that provides `WorkflowEngine.WorkflowEngine` using the cluster workflow
921 * engine implementation.
922 *
923 * **Details**
924 *
925 * It requires cluster sharding and message storage, and also registers the
926 * durable clock entity used for workflow wakeups.
927 *
928 * @stability unstable
929 * @category layers
930 * @since 4.0.0
931 */
932export const layer: Layer.Layer<
933 WorkflowEngine.WorkflowEngine,
934 never,
935 Sharding.Sharding | MessageStorage
936> = ClockEntityLayer.pipe(
937 Layer.provideMerge(Layer.effect(WorkflowEngine.WorkflowEngine)(make))
938)
939
940const payloadShape = (workflow: Workflow.Any): string => {
941 try {
942 const { definitions, schema } = Schema.toJsonSchemaDocument(workflow.payloadSchema)
943 return JSON.stringify(Object.keys(definitions).length === 0 ? schema : { ...schema, $defs: definitions })
944 } catch {
945 return "<unavailable>"
946 }
947}