Function drive

packages/infra/src/WorkflowEngineSqlite.ts:502–567

(
    executionId: string,
    payload: object,
    parent: string | undefined,
    entry: Registered
  )

  // --- Drive logic ------------------------------------------------------

  const drive = (
    executionId: string,
    payload: object,
    parent: string | undefined,
    entry: Registered
  ): Effect.Effect<void> =>
    Effect.gen(function*() {
      let local = locals.get(executionId)
      if (local?.fiber) {
        const polled = local.fiber.pollUnsafe()
        const stillRunning = !polled
        const completedNotResume = polled && polled._tag === "Success" && polled.value._tag === "Complete"
        if (stillRunning || completedNotResume) return
      }

      const stateOpt = yield* readExec(executionId)
      if (Option.isNone(stateOpt) || stateOpt.value.status === "complete") return

      const claimed = yield* tryClaim(stateOpt.value)
      const state = Option.isSome(claimed) ? claimed.value : stateOpt.value

      const instance = WorkflowInstance.initial(entry.workflow, executionId)
      instance.interrupted = state.interrupted
      if (!local) {
        local = { instance, fiber: undefined, parent }
        locals.set(executionId, local)
      } else {
        local.instance = instance
      }

      const onComplete = Effect.fnUntraced(function*(result: Workflow.Result<unknown, unknown>) {
        const current = yield* readExec(executionId)
        if (Option.isNone(current) || current.value.status === "complete") return
        const isComplete = result._tag === "Complete"
        const completedResult = isComplete ? yield* encodeResult(entry.workflow, result) : undefined
        yield* replaceExec(current.value, {
          status: isComplete ? "complete" : current.value.status,
          suspended: result._tag === "Suspended",
          interrupted: instance.interrupted,
          completedResult,
          worker: isComplete ? undefined : current.value.worker,
          leaseExpiresAt: isComplete ? undefined : current.value.leaseExpiresAt
        })
          .pipe(Effect.catchTag("OptimisticConcurrencyException", () => Effect.void))
        if (parent && isComplete) {
          yield* Effect.forkIn(driveById(parent), scope)
        }
      })

      local.fiber = yield* entry.execute(payload, executionId).pipe(
        Effect.onExit(() => {
          if (!instance.interrupted) return Effect.void
          instance.suspended = false
          return Effect.withFiber((fiber) => Effect.interruptible(Fiber.interrupt(fiber)))
        }),
        Workflow.intoResult,
        Effect.provideService(WorkflowInstance, instance),
        Effect.provideService(WorkflowEngine, engine),