Fix request registration before synchronous batch delays

.changeset/request-zero-delay.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+Fix request resolvers with synchronously completing delays, including `RequestResolver.setDelay(0)` under `TestClock`, running an empty batch and failing instead of resolving the request.

packages/effect/src/internal/request.ts

@@ -157,7 +157,11 @@ const addEntry = <A extends Request.Any>(
         entrySet: new Set(),
         entries: new Set(),
         delayEffect: effect.flatMap(
-          effect.suspend(() => newBatch.resolver.delay),
+          effect.suspend(() => {
+            // Delay callbacks can enqueue requests before runForkWith returns.
+            newBatch.fiber = effect.getCurrentFiber()!
+            return newBatch.resolver.delay
+          }),
           (_) => runBatch(newBatch)
         ) as Effect<void>,
         run: effect.onExit(
@@ -190,15 +194,21 @@ const addEntry = <A extends Request.Any>(
       batch = newBatch
     }
     batchMap.set(key, batch)
-    batch.fiber = effect.runForkWith(fiber.context)(batch.delayEffect, { scheduler: fiber.cache.scheduler })
+    batch.entrySet.add(entry)
+    batch.entries.add(entry)
+    const delayFiber = effect.runForkWith(fiber.context)(batch.delayEffect, { scheduler: fiber.cache.scheduler })
+    // A synchronous delay may have already run and recycled this batch.
+    if (completed || batchMap.get(key) !== batch) return entry
+    batch.fiber = delayFiber
+  } else {
+    batch.entrySet.add(entry)
+    batch.entries.add(entry)
   }
 
-  batch.entrySet.add(entry)
-  batch.entries.add(entry)
   if (batch.resolver.collectWhile(batch.entries)) return entry
 
   batch.fiber!.interruptUnsafe(fiber.id)
-  batch.fiber = effect.runForkWith(fiber.context)(runBatch(batch), { scheduler: fiber.cache.scheduler })
+  effect.runForkWith(fiber.context)(runBatch(batch), { scheduler: fiber.cache.scheduler })
   return entry
 }
 

packages/effect/test/Request.test.ts

@@ -794,4 +794,87 @@ describe("Request", { concurrent: false }, () => {
       assert.strictEqual(delayServiceValue, "provided")
     })
   )
+
+  it.effect("collectWhile preserves synchronous delay side effects across pooled batches", () =>
+    Effect.gen(function*() {
+      const events: Array<string> = []
+      const resolver = Resolver.makeWith<GetRequestService>({
+        batchKey: () => undefined,
+        delay: Effect.andThen(Effect.sync(() => events.push("delay")), Effect.never).pipe(
+          Effect.onInterrupt(() => Effect.sync(() => events.push("interrupt")))
+        ),
+        collectWhile(entries) {
+          events.push(`collect:${entries.size}`)
+          return false
+        },
+        runAll: (entries) =>
+          Effect.sync(() => {
+            events.push(`run:${entries.length}`)
+            for (const entry of entries) entry.completeUnsafe(Exit.succeed("ok"))
+          })
+      })
+
+      // Repeat after synchronous completion to exercise reuse of a recycled batch.
+      for (let i = 0; i < 2; i++) {
+        assert.strictEqual(yield* Effect.request(GetRequestService(), resolver), "ok")
+      }
+      assert.deepStrictEqual(events, [
+        "delay",
+        "collect:1",
+        "interrupt",
+        "run:1",
+        "delay",
+        "collect:1",
+        "interrupt",
+        "run:1"
+      ])
+    }))
+
+  it.effect("synchronous delay callbacks can enqueue requests that fill a batch", () =>
+    Effect.gen(function*() {
+      const context = yield* Effect.context<never>()
+      const batches: Array<number> = []
+      const results: Array<Request.Result<GetRequestService>> = []
+      const resolver = Resolver.make<GetRequestService>((entries) =>
+        Effect.sync(() => {
+          batches.push(entries.length)
+          for (const entry of entries) entry.completeUnsafe(Exit.succeed("ok"))
+        })
+      ).pipe(
+        Resolver.batchN(2),
+        Resolver.setDelayEffect(Effect.andThen(
+          Effect.sync(() => {
+            Effect.requestUnsafe(GetRequestService(), { resolver, context, onExit: (exit) => results.push(exit) })
+          }),
+          Effect.never
+        ))
+      )
+
+      for (let i = 0; i < 2; i++) {
+        assert.strictEqual(yield* Effect.request(GetRequestService(), resolver), "ok")
+      }
+      assert.deepStrictEqual(batches, [2, 2])
+      assert.deepStrictEqual(results, [Exit.succeed("ok"), Exit.succeed("ok")])
+    }))
+
+  for (const [clock, test] of [["live clock", it.live], ["TestClock", it.effect]] as const) {
+    test(`setDelay(0) resolves a non-empty grouped batch with ${clock}`, () =>
+      Effect.gen(function*() {
+        const batches: Array<Array<number>> = []
+        const resolver = Resolver.makeGrouped<GetNameById, string>({
+          key: () => "all",
+          resolver: (entries) =>
+            Effect.sync(() => {
+              batches.push(entries.map(({ request }) => request.id))
+              for (const entry of entries) entry.completeUnsafe(Exit.succeed(String(entry.request.id)))
+            })
+        }).pipe(Resolver.setDelay(0), Resolver.batchN(2))
+
+        // Resume after a microtask, as in callers awaiting external work before requesting.
+        yield* Effect.promise(() => Promise.resolve())
+        const exit = yield* Effect.exit(Effect.request(new GetNameById({ id: 1 }), resolver))
+
+        assert.deepStrictEqual({ exit, batches }, { exit: Exit.succeed("1"), batches: [[1]] })
+      }))
+  }
 })