Function activity

packages/cluster/src/ClusterWorkflowEngine.ts:334–381

(request: Entity.Request<any>)

              },

              activity(request: Entity.Request<any>) {
                const activityId = `${executionId}/${request.payload.name}`
                const instance = WorkflowInstance.initial(workflow, executionId)
                interruptedActivities.delete(activityId)
                return Effect.gen(function*() {
                  let entry = activities.get(activityId)
                  while (!entry) {
                    const latch = Effect.unsafeMakeLatch()
                    activityLatches.set(activityId, latch)
                    yield* latch.await
                    entry = activities.get(activityId)
                  }
                  const contextMap = new Map(entry.runtime.context.unsafeMap)
                  contextMap.set(Activity.CurrentAttempt.key, request.payload.attempt)
                  contextMap.set(WorkflowInstance.key, instance)
                  const runtime = Runtime.make({
                    context: Context.unsafeMake(contextMap),
                    fiberRefs: entry.runtime.fiberRefs,
                    runtimeFlags: Runtime.defaultRuntimeFlags
                  })
                  return yield* entry.activity.executeEncoded.pipe(
                    Effect.provide(runtime)
                  )
                }).pipe(
                  Workflow.intoResult,
                  Effect.catchAllCause((cause) => {
                    const interruptors = Cause.interruptors(cause)
                    // we only want to store interrupts as suspends when the
                    // client requested it
                    const ids = Array.from(interruptors, (id) => Array.from(FiberId.ids(id))).flat()
                    const suspend = ids.includes(RpcServer.fiberIdClientInterrupt.id)
                    if (suspend) {
                      interruptedActivities.add(activityId)
                      return Effect.succeed(new Workflow.Suspended())
                    }
                    return Effect.failCause(cause)
                  }),
                  Effect.provideService(WorkflowInstance, instance),
                  Effect.provideService(Activity.CurrentAttempt, request.payload.attempt),
                  Effect.ensuring(Effect.sync(() => {
                    activities.delete(activityId)
                  })),
                  Rpc.wrap({
                    fork: true,
                    uninterruptible: true
                  })
                )
              },

              deferred: Effect.fnUntraced(function*(request: Entity.Request<any>) {
                yield* ensureSuccess(resume(workflow, executionId))