add WorkflowEngine interruptUnsafe (#1907)

Author: tim-smart

.changeset/public-deer-ring.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+add WorkflowEngine interruptUnsafe

packages/effect/src/unstable/cluster/ClusterWorkflowEngine.ts

@@ -30,6 +30,8 @@ import * as Entity from "./Entity.ts"
 import * as EntityAddress from "./EntityAddress.ts"
 import * as EntityId from "./EntityId.ts"
 import * as EntityType from "./EntityType.ts"
+import * as Envelope from "./Envelope.ts"
+import * as Message from "./Message.ts"
 import { MessageStorage } from "./MessageStorage.ts"
 import type { WithExitEncoded } from "./Reply.ts"
 import * as Reply from "./Reply.ts"
@@ -131,22 +133,30 @@ export const make = Effect.gen(function*() {
   })
   const clockClient = yield* ClockEntity.client
 
-  const requestIdFor = Effect.fnUntraced(function*(options: {
+  const entityAddressFor = (options: {
     readonly workflow: Workflow.Any
     readonly entityType: string
     readonly executionId: string
-    readonly tag: string
-    readonly id: string
-  }) {
+  }) => {
     const shardGroup = ServiceMap.get(options.workflow.annotations, ClusterSchema.ShardGroup)(
       options.executionId as EntityId.EntityId
     )
     const entityId = EntityId.make(options.executionId)
-    const address = EntityAddress.make({
+    return EntityAddress.make({
       entityType: EntityType.make(options.entityType),
       entityId,
       shardId: sharding.getShardId(entityId, shardGroup)
     })
+  }
+
+  const requestIdFor = Effect.fnUntraced(function*(options: {
+    readonly workflow: Workflow.Any
+    readonly entityType: string
+    readonly executionId: string
+    readonly tag: string
+    readonly id: string
+  }) {
+    const address = entityAddressFor(options)
     return yield* storage.requestIdForPrimaryKey({ address, tag: options.tag, id: options.id })
   })
 
@@ -253,6 +263,46 @@ export const make = Effect.gen(function*() {
     yield* sharding.reset(requestId.value)
   }, Effect.scoped)
 
+  const interrupt = Effect.fnUntraced(
+    function*(workflow: Workflow.Any, executionId: string) {
+      ensureEntity(workflow)
+      const requestId = yield* requestIdFor({
+        workflow,
+        entityType: `Workflow/${workflow.name}`,
+        executionId,
+        tag: "run",
+        id: ""
+      })
+      if (Option.isNone(requestId)) {
+        return Option.none()
+      }
+      const reply = yield* replyForRequestId(requestId.value)
+
+      const nonSuspendedReply = Option.filter(
+        reply,
+        (reply) => reply.exit._tag !== "Success" || reply.exit.value._tag !== "Suspended"
+      )
+      if (Option.isSome(nonSuspendedReply)) {
+        return Option.none()
+      }
+
+      yield* engine.deferredDone(InterruptSignal, {
+        workflowName: workflow.name,
+        executionId,
+        deferredName: InterruptSignal.name,
+        exit: Exit.void
+      })
+
+      return requestId
+    },
+    Effect.retry({
+      while: (e) => e._tag === "PersistenceError",
+      times: 3,
+      schedule: Schedule.exponential(250)
+    }),
+    Effect.orDie
+  )
+
   const engine = WorkflowEngine.makeUnsafe({
     register: (workflow, execute) =>
       Effect.suspend(() =>
@@ -386,39 +436,28 @@ export const make = Effect.gen(function*() {
       return Option.some(yield* exit)
     }, Effect.orDie),
 
-    interrupt: Effect.fnUntraced(
-      function*(workflow, executionId) {
-        ensureEntity(workflow)
-        const reply = yield* requestReply({
-          workflow,
-          entityType: `Workflow/${workflow.name}`,
-          executionId,
-          tag: "run",
-          id: ""
-        })
-
-        const nonSuspendedReply = Option.filter(
-          reply,
-          (reply) => reply.exit._tag !== "Success" || reply.exit.value._tag !== "Suspended"
-        )
-        if (Option.isSome(nonSuspendedReply)) {
-          return
-        }
-
-        yield* engine.deferredDone(InterruptSignal, {
-          workflowName: workflow.name,
-          executionId,
-          deferredName: InterruptSignal.name,
-          exit: Exit.void
-        })
-      },
-      Effect.retry({
-        while: (e) => e._tag === "PersistenceError",
-        times: 3,
-        schedule: Schedule.exponential(250)
-      }),
-      Effect.orDie
-    ),
+    interrupt: (workflow, executionId) => Effect.asVoid(interrupt(workflow, executionId)),
+    interruptUnsafe: Effect.fnUntraced(function*(workflow, executionId) {
+      const requestId = yield* interrupt(workflow, executionId)
+      if (Option.isNone(requestId)) return
+      const entity = ensureEntity(workflow)
+      const runRpc = entity.protocol.requests.get("run")!
+      yield* Effect.orDie(sharding.sendOutgoing(
+        new Message.OutgoingEnvelope({
+          rpc: runRpc,
+          envelope: new Envelope.Interrupt({
+            id: yield* sharding.getSnowflake,
+            address: entityAddressFor({
+              workflow,
+              entityType: `Workflow/${workflow.name}`,
+              executionId
+            }),
+            requestId: requestId.value
+          })
+        }),
+        false
+      ))
+    }),
 
     resume: (workflow, executionId) => ensureSuccess(resume(workflow, executionId)),
 

packages/effect/src/unstable/workflow/WorkflowEngine.ts

@@ -111,6 +111,15 @@ export class WorkflowEngine extends ServiceMap.Service<
       executionId: string
     ) => Effect.Effect<void>
 
+    /**
+     * Unsafely interrupt a registered workflow, potentially ignoring
+     * compensation finalizers and orphaning child workflows.
+     */
+    readonly interruptUnsafe: (
+      workflow: Workflow.Any,
+      executionId: string
+    ) => Effect.Effect<void>
+
     /**
      * Resume a registered workflow.
      */
@@ -282,6 +291,10 @@ export interface Encoded {
     workflow: Workflow.Any,
     executionId: string
   ) => Effect.Effect<void>
+  readonly interruptUnsafe: (
+    workflow: Workflow.Any,
+    executionId: string
+  ) => Effect.Effect<void>
   readonly resume: (
     workflow: Workflow.Any,
     executionId: string
@@ -416,6 +429,7 @@ export const makeUnsafe = (options: Encoded): WorkflowEngine["Service"] =>
     }),
     poll: options.poll,
     interrupt: options.interrupt,
+    interruptUnsafe: options.interruptUnsafe,
     resume: options.resume,
     activityExecute: Effect.fnUntraced(function*<
       Success extends Schema.Top,
@@ -617,6 +631,14 @@ export const layerMemory: Layer.Layer<WorkflowEngine> = Layer.effect(WorkflowEng
         state.instance.interrupted = true
         yield* resume(executionId)
       }),
+      interruptUnsafe: Effect.fnUntraced(function*(_workflow, executionId) {
+        const state = executions.get(executionId)
+        if (!state) return
+        state.instance.interrupted = true
+        if (state.fiber) {
+          yield* Fiber.interrupt(state.fiber)
+        }
+      }),
       resume(_workflow, executionId) {
         return resume(executionId)
       },