Fix concurrent Cloudflare entity handler builds (#7346)

Author: tim-smart

.changeset/cloudflare-entity-handler-cache.md

@@ -0,0 +1,5 @@
+---
+"@effect/platform-cloudflare": patch
+---
+
+Share an in-flight entity handler build between concurrent first requests.

packages/platform/cloudflare/src/internal/entityRuntime.ts

@@ -64,6 +64,7 @@ export const makeEntityRuntime = Effect.fnUntraced(function*(
   replyRegistry?: EntityReplyRegistry
 ) {
   let cached: CachedHandlers | undefined
+  let building: Deferred.Deferred<CachedHandlers> | undefined
   const metricContext = Context.merge(
     registration.context,
     Metric.CurrentMetricAttributes.context({ type: registration.entity.type })
@@ -84,8 +85,7 @@ export const makeEntityRuntime = Effect.fnUntraced(function*(
     )
   })
 
-  const getHandlers = Effect.fnUntraced(function*() {
-    if (cached !== undefined) return cached
+  const buildHandlers = Effect.fnUntraced(function*() {
     const scope = yield* Scope.make()
     let context = registration.context.pipe(
       Context.add(CurrentAddress, address),
@@ -99,11 +99,27 @@ export const makeEntityRuntime = Effect.fnUntraced(function*(
     if (replyRegistry !== undefined) {
       context = Context.add(context, CurrentReplyRegistry, replyRegistry)
     }
-    const handlers = yield* Effect.provideContext(registration.build, context)
+    const handlers = yield* Effect.provideContext(registration.build, context).pipe(
+      Effect.tapCause((cause) => Scope.close(scope, Exit.failCause(cause)))
+    )
     ClusterMetrics.entities.modifyUnsafe(BigInt(1), metricContext)
-    return cached = { handlers, context, scope }
+    return { handlers, context, scope }
   })
 
+  const getHandlers = (): Effect.Effect<CachedHandlers> =>
+    Effect.suspend(() => {
+      if (cached !== undefined) return Effect.succeed(cached)
+      if (building !== undefined) return Deferred.await(building)
+      const deferred = Deferred.makeUnsafe<CachedHandlers>()
+      building = deferred
+      return Effect.onExit(buildHandlers(), (exit) =>
+        Effect.sync(() => {
+          building = undefined
+          if (Exit.isSuccess(exit)) cached = exit.value
+          Deferred.doneUnsafe(deferred, exit)
+        }))
+    })
+
   const runWithDefectRetry = <A, E, R>(effect: Effect.Effect<A, E, R>) => {
     const policy = registration.options?.defectRetryPolicy
     if (policy === undefined) return Effect.exit(effect)

packages/platform/cloudflare/test/EntityRuntime.test.ts

@@ -1,7 +1,21 @@
 import type { EntityRegistration } from "@effect/platform-cloudflare/internal/entityRegistry"
 import { makeEntityRuntime } from "@effect/platform-cloudflare/internal/entityRuntime"
 import { assert, describe, it } from "@effect/vitest"
-import { Cause, Context, Effect, Exit, Metric, Option, Schedule, Schema, Stream, Tracer } from "effect"
+import {
+  Cause,
+  Context,
+  Deferred,
+  Effect,
+  Exit,
+  Fiber,
+  Metric,
+  Option,
+  Schedule,
+  Schema,
+  Scope,
+  Stream,
+  Tracer
+} from "effect"
 import { ClusterMetrics, Entity, EntityAddress, EntityId, EntityType, ShardId } from "effect/unstable/cluster"
 import { Rpc, RpcSchema } from "effect/unstable/rpc"
 
@@ -131,6 +145,72 @@ describe("EntityRuntime", () => {
       assert.deepStrictEqual(replies.map((reply) => reply.exit.value), ["pong", "pong"])
     }))
 
+  it.effect("shares an asynchronous handler build between concurrent first requests", () =>
+    Effect.gen(function*() {
+      const Concurrent = Entity.make("Concurrent", [
+        Rpc.make("Ping", { success: Schema.String })
+      ])
+      const concurrentAddress = new EntityAddress.EntityAddress({
+        shardId: ShardId.make("default", 1),
+        entityType: EntityType.make("Concurrent"),
+        entityId: EntityId.make("42")
+      })
+      const context = Context.empty()
+      const metricContext = Context.merge(
+        context,
+        Metric.CurrentMetricAttributes.context({ type: Concurrent.type })
+      )
+      const releaseBuild = Deferred.makeUnsafe<void>()
+      let builds = 0
+      let finalizers = 0
+      const registration: EntityRegistration = {
+        entity: Concurrent,
+        build: Effect.gen(function*() {
+          const scope = Option.getOrThrow(yield* Effect.serviceOption(Scope.Scope))
+          builds++
+          yield* Scope.addFinalizer(
+            scope,
+            Effect.sync(() => {
+              finalizers++
+            })
+          )
+          yield* Deferred.await(releaseBuild)
+          return Concurrent.of({ Ping: () => Effect.succeed("pong") })
+        }),
+        options: { concurrency: "unbounded" },
+        context
+      }
+      const runtime = yield* makeEntityRuntime(registration, concurrentAddress, () => "reply")
+      const first = yield* Effect.forkChild(
+        runtime.run({ ...request, address: concurrentAddress } as any, Option.none(), false, () => Effect.void)
+      )
+      const second = yield* Effect.forkChild(
+        runtime.run(
+          {
+            ...request,
+            requestId: "0198bd72-6a83-72f1-8d87-5e9b5cf1e003",
+            address: concurrentAddress
+          } as any,
+          Option.none(),
+          false,
+          () => Effect.void
+        )
+      )
+
+      yield* Effect.yieldNow
+      yield* Deferred.succeed(releaseBuild, undefined)
+      yield* Fiber.join(first)
+      yield* Fiber.join(second)
+
+      assert.strictEqual(builds, 1)
+      assert.strictEqual(ClusterMetrics.entities.valueUnsafe(metricContext).value, BigInt(1))
+      assert.strictEqual(finalizers, 0)
+
+      yield* runtime.invalidate()
+      assert.strictEqual(ClusterMetrics.entities.valueUnsafe(metricContext).value, BigInt(0))
+      assert.strictEqual(finalizers, 1)
+    }))
+
   it.effect("resumes ask stream sequence from lastSentChunk and ends with WithExit", () =>
     Effect.gen(function*() {
       const Streaming = Entity.make("User", [