From 9a4d0ecca05342a31a648c4c10120b285d0040d8 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 16 Oct 2026 21:27:21 +0000 Subject: [PATCH 1/5] test(Request): reproduce zero-delay batch registration race --- packages/effect/test/Request.test.ts | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/packages/effect/test/Request.test.ts b/packages/effect/test/Request.test.ts index 05ca5b26fe4..0d5fee59063 100644 --- a/packages/effect/test/Request.test.ts +++ b/packages/effect/test/Request.test.ts @@ -794,4 +794,25 @@ describe("Request", { concurrent: false }, () => { assert.strictEqual(delayServiceValue, "provided") }) ) + + 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> = [] + const resolver = Resolver.makeGrouped({ + 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]] }) + })) + } }) From 4e032e39d4b84a9dbbcbc1114c547968a9599c54 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 16 Oct 2026 21:29:48 +0000 Subject: [PATCH 2/5] Fix request registration before synchronous batch delays --- .changeset/request-zero-delay.md | 5 +++++ packages/effect/src/internal/request.ts | 20 +++++++++++++++----- 2 files changed, 20 insertions(+), 5 deletions(-) create mode 100644 .changeset/request-zero-delay.md diff --git a/.changeset/request-zero-delay.md b/.changeset/request-zero-delay.md new file mode 100644 index 00000000000..25a91288b04 --- /dev/null +++ b/.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. diff --git a/packages/effect/src/internal/request.ts b/packages/effect/src/internal/request.ts index 61fe039ce20..0888b4ba41e 100644 --- a/packages/effect/src/internal/request.ts +++ b/packages/effect/src/internal/request.ts @@ -157,7 +157,11 @@ const addEntry = ( 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, run: effect.onExit( @@ -190,15 +194,21 @@ const addEntry = ( 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 } From 6fbfaf1b93e2775f7bb371311533699bc24d1ed9 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 16 Oct 2026 21:31:49 +0000 Subject: [PATCH 3/5] test(Request): cover synchronous delay reentrancy and pooled reuse --- packages/effect/test/Request.test.ts | 62 ++++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/packages/effect/test/Request.test.ts b/packages/effect/test/Request.test.ts index 0d5fee59063..e799c18a3c0 100644 --- a/packages/effect/test/Request.test.ts +++ b/packages/effect/test/Request.test.ts @@ -795,6 +795,68 @@ describe("Request", { concurrent: false }, () => { }) ) + it.effect("collectWhile preserves synchronous delay side effects across pooled batches", () => + Effect.gen(function*() { + const events: Array = [] + const resolver = Resolver.makeWith({ + 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() + const batches: Array = [] + const results: Array> = [] + const resolver = Resolver.make((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*() { From f7c9dc1a022dfd731464b1f8d370413ada16365e Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 16 Oct 2026 21:43:28 +0000 Subject: [PATCH 4/5] Simplify request batch registration and reword changeset --- .changeset/request-zero-delay.md | 2 +- packages/effect/src/internal/request.ts | 23 ++++++++++++----------- 2 files changed, 13 insertions(+), 12 deletions(-) diff --git a/.changeset/request-zero-delay.md b/.changeset/request-zero-delay.md index 25a91288b04..f939d808427 100644 --- a/.changeset/request-zero-delay.md +++ b/.changeset/request-zero-delay.md @@ -2,4 +2,4 @@ "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. +Fix `Effect.request` failing when a resolver's batch delay completes synchronously, such as `RequestResolver.setDelay(0)` under `TestClock`, by registering the request before the delay starts. diff --git a/packages/effect/src/internal/request.ts b/packages/effect/src/internal/request.ts index 0888b4ba41e..3c234b53f56 100644 --- a/packages/effect/src/internal/request.ts +++ b/packages/effect/src/internal/request.ts @@ -10,7 +10,7 @@ import type * as Request from "../Request.ts" import { makeEntry } from "../Request.ts" import type { RequestResolver } from "../RequestResolver.ts" import { Scheduler } from "../Scheduler.ts" -import { exitDie, isEffect } from "./core.ts" +import { exitDie, isEffect, withFiber } from "./core.ts" import * as effect from "./effect.ts" /** @internal */ @@ -103,7 +103,7 @@ interface Batch { readonly entries: Set> readonly delayEffect: Effect readonly run: Effect - fiber?: Fiber | undefined + fiber?: Fiber | undefined } const batchPool: Array = [] @@ -143,6 +143,7 @@ const addEntry = ( } const key = resolver.batchKey(entry) batch = batchMap.get(key) + const isNewBatch = batch === undefined if (!batch) { if (batchPool.length > 0) { batch = batchPool.pop()! @@ -157,9 +158,9 @@ const addEntry = ( entrySet: new Set(), entries: new Set(), delayEffect: effect.flatMap( - effect.suspend(() => { - // Delay callbacks can enqueue requests before runForkWith returns. - newBatch.fiber = effect.getCurrentFiber()! + withFiber((delayFiber) => { + // Reentrant delay callbacks may fill the batch and interrupt this fiber. + newBatch.fiber = delayFiber return newBatch.resolver.delay }), (_) => runBatch(newBatch) @@ -194,15 +195,15 @@ const addEntry = ( batch = newBatch } batchMap.set(key, batch) - batch.entrySet.add(entry) - batch.entries.add(entry) + } + + batch.entrySet.add(entry) + batch.entries.add(entry) + if (isNewBatch) { 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) } if (batch.resolver.collectWhile(batch.entries)) return entry @@ -226,7 +227,7 @@ const removeEntryUnsafe = ( if (!batch.entries.delete(entry)) return batch.entrySet.delete(entry) - let fiber: Fiber | undefined + let fiber: Fiber | undefined if (batch.entries.size === 0) { batchMap.delete(key) fiber = batch.fiber From fbc184ddd02e150ba0fb03e6ffed185bdf3e3b06 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Sun, 18 Oct 2026 23:42:11 +0000 Subject: [PATCH 5/5] test(Request): reduce synchronous delay coverage to one regression --- packages/effect/test/Request.test.ts | 88 ++++------------------------ 1 file changed, 11 insertions(+), 77 deletions(-) diff --git a/packages/effect/test/Request.test.ts b/packages/effect/test/Request.test.ts index e799c18a3c0..d4318cbfcbf 100644 --- a/packages/effect/test/Request.test.ts +++ b/packages/effect/test/Request.test.ts @@ -795,86 +795,20 @@ describe("Request", { concurrent: false }, () => { }) ) - it.effect("collectWhile preserves synchronous delay side effects across pooled batches", () => + it.effect("setDelay(0) registers the request before running the batch", () => Effect.gen(function*() { - const events: Array = [] - const resolver = Resolver.makeWith({ - 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) => + const batches: Array> = [] + const resolver = Resolver.makeGrouped({ + key: () => "all", + resolver: (entries) => Effect.sync(() => { - events.push(`run:${entries.length}`) - for (const entry of entries) entry.completeUnsafe(Exit.succeed("ok")) + batches.push(entries.map(({ request }) => request.id)) + for (const entry of entries) entry.completeUnsafe(Exit.succeed(String(entry.request.id))) }) - }) - - // 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() - const batches: Array = [] - const results: Array> = [] - const resolver = Resolver.make((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 - )) - ) + }).pipe(Resolver.setDelay(0), Resolver.batchN(2)) - 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")]) + yield* Effect.promise(() => Promise.resolve()) + assert.strictEqual(yield* Effect.request(new GetNameById({ id: 1 }), resolver), "1") + assert.deepStrictEqual(batches, [[1]]) })) - - 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> = [] - const resolver = Resolver.makeGrouped({ - 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]] }) - })) - } })