infra: drive rpc interruptibility from request type

Author: patroza

.changeset/curly-beds-sing.md

@@ -0,0 +1,5 @@
+---
+"@effect-app/infra": patch
+---
+
+Use request-type RPC annotations to drive interruptibility and retry behavior, apply command uninterruptibility through the router wrapper path, and add an in-memory E2E test covering command vs query interruption behavior.

packages/infra/src/api/routing.ts

@@ -11,14 +11,20 @@ import { typedKeysOf, typedValuesOf } from "effect-app/utils"
 import { type Yieldable } from "effect/Effect"
 import { Rpc, RpcGroup, type RpcSerialization, RpcServer } from "effect/unstable/rpc"
 import { type LayerUtils } from "./layerUtils.js"
-import { type RouterMiddleware } from "./routing/middleware.js"
+import { RequestType as RequestTypeAnnotation, type RouterMiddleware } from "./routing/middleware.js"
 
 export * from "./routing/middleware.js"
 
+export const applyRequestTypeInterruptibility = <A, E, R>(
+  requestType: "command" | "query",
+  effect: Effect.Effect<A, E, R>
+) => requestType === "command" ? Rpc.uninterruptible(effect) : effect
+
 // it's the result of extending S.Req setting success, config
 // it's a schema plus some metadata
 export type AnyRequestModule = S.Top & {
   _tag: string // unique identifier for the request module
+  type: "command" | "query"
   config: any // ?
   success: S.Top // validates the success response
   error: S.Top // validates the failure response
@@ -386,12 +392,15 @@ export const makeRouter = <
                   static success = S.toEncoded(resource.success)
                 } as any
                 : resource,
-              (payload: any, headers: any) =>
-                (handler.handler(payload, headers) as Effect.Effect<unknown, unknown, unknown>).pipe(
+              (payload: any, headers: any) => {
+                const effect = (handler.handler(payload, headers) as Effect.Effect<unknown, unknown, unknown>).pipe(
                   Effect.withSpan(`Request.${meta.moduleName}.${resource._tag}`, {}, {
                     captureStackTrace: () => handler.stack // capturing the handler stack is the main reason why we are doing the span here
                   })
                 )
+
+                return applyRequestTypeInterruptibility(resource.type, effect)
+              }
             ] as const
             return acc
           }, {} as any) as {
@@ -418,6 +427,7 @@ export const makeRouter = <
                 return Rpc
                   .make(resource._tag, { payload: resource, success: resource.success, error: resource.error })
                   .annotate(middleware.requestContext, resource.config ?? {})
+                  .annotate(RequestTypeAnnotation, resource.type)
               })
             )
             .prefix(`${meta.moduleName}.`)

packages/infra/src/api/routing/middleware/middleware.ts

@@ -3,11 +3,11 @@ import { Cause, Config, Effect, Layer, Schema } from "effect"
 import { ConfigureInterruptibilityMiddleware, DevMode, DevModeMiddleware, LoggerMiddleware, RequestCacheMiddleware } from "effect-app/middleware"
 import { RpcContextMap, type RpcMiddleware } from "effect-app/rpc"
 import { pretty } from "effect-app/utils"
+import * as Context from "effect/Context"
 import { type Rpc } from "effect/unstable/rpc"
 import { logError, reportError } from "../../../errorReporter.js"
 import { InfraLogger } from "../../../logger.js"
 import { WithNsTransaction } from "../../../Store/SQL.js"
-import { determineMethod, isCommand } from "../utils.js"
 
 const logRequestError = logError("Request")
 const reportRequestError = reportError("Request")
@@ -29,22 +29,20 @@ export const RequestCacheMiddlewareLive = Layer.succeed(
 const isOptimisticConcurrencyException = (input: unknown) =>
   typeof input === "object" && input !== null && "_tag" in input && input._tag === "OptimisticConcurrencyException"
 
+export const RequestType = Context.Reference<"command" | "query">(
+  "@effect-app/infra/api/routing/RequestType",
+  { defaultValue: () => "query" }
+)
+
 export const ConfigureInterruptibilityMiddlewareLive = Layer.effect(
   ConfigureInterruptibilityMiddleware,
   Effect.gen(function*() {
-    const cache = new Map()
-    const getCached = (key: string, schema: Schema.Top) => {
-      const existing = cache.get(key)
-      if (existing) return existing
-      const n = determineMethod(key, schema)
-      cache.set(key, n)
-      return n
-    }
     return (effect, { rpc }) => {
-      const method = getCached(rpc._tag, rpc.payloadSchema)
+      const requestType = Context.get(rpc.annotations, RequestType)
+      const isCommand = requestType === "command"
 
-      effect = isCommand(method)
-        ? Effect.retry(Effect.uninterruptible(effect), { times: 1, while: isOptimisticConcurrencyException })
+      effect = isCommand
+        ? Effect.retry(effect, { times: 1, while: isOptimisticConcurrencyException })
         : Effect.interruptible(effect)
 
       return effect

packages/infra/test/routing-interruptibility.test.ts

@@ -0,0 +1,63 @@
+import { describe, expect, it } from "@effect/vitest"
+import { Effect, Fiber, Layer, Ref } from "effect"
+import { S } from "effect-app"
+import { ConfigureInterruptibilityMiddleware } from "effect-app/middleware"
+import { Rpc, RpcGroup, RpcTest } from "effect/unstable/rpc"
+import { applyRequestTypeInterruptibility } from "../src/api/routing.js"
+import { ConfigureInterruptibilityMiddlewareLive, RequestType } from "../src/api/routing/middleware.js"
+
+const InterruptibilityRpcs = RpcGroup.make(
+  Rpc
+    .make("doCommand", { success: S.Void })
+    .annotate(RequestType, "command")
+    .middleware(ConfigureInterruptibilityMiddleware),
+  Rpc
+    .make("doQuery", { success: S.Void })
+    .annotate(RequestType, "query")
+    .middleware(ConfigureInterruptibilityMiddleware)
+)
+
+const makeImplLayer = (commandDone: Ref.Ref<boolean>, queryDone: Ref.Ref<boolean>) =>
+  InterruptibilityRpcs.toLayer({
+    doCommand: () =>
+      applyRequestTypeInterruptibility(
+        "command",
+        Effect.sleep("120 millis").pipe(Effect.andThen(Ref.set(commandDone, true)))
+      ),
+    doQuery: () =>
+      applyRequestTypeInterruptibility(
+        "query",
+        Effect.sleep("120 millis").pipe(Effect.andThen(Ref.set(queryDone, true)))
+      )
+  })
+
+describe("routing interruptibility", () => {
+  it.live(
+    "e2e: command continues after client interrupt, query does not",
+    () =>
+      Effect.gen(function*() {
+        const commandDone = yield* Ref.make(false)
+        const queryDone = yield* Ref.make(false)
+
+        const client = yield* RpcTest
+          .makeClient(InterruptibilityRpcs)
+          .pipe(
+            Effect.provide(
+              Layer.mergeAll(makeImplLayer(commandDone, queryDone), ConfigureInterruptibilityMiddlewareLive)
+            )
+          )
+
+        const commandFiber = yield* Effect.forkDetach(client.doCommand())
+        yield* Effect.sleep("20 millis")
+        yield* Fiber.interrupt(commandFiber)
+        yield* Effect.sleep("180 millis")
+        expect(yield* Ref.get(commandDone)).toBe(true)
+
+        const queryFiber = yield* Effect.forkDetach(client.doQuery())
+        yield* Effect.sleep("20 millis")
+        yield* Fiber.interrupt(queryFiber)
+        yield* Effect.sleep("180 millis")
+        expect(yield* Ref.get(queryDone)).toBe(false)
+      })
+  )
+})