diff --git a/.changeset/fix-stream-aggregate-within-idle.md b/.changeset/fix-stream-aggregate-within-idle.md new file mode 100644 index 00000000000..96744190cf3 --- /dev/null +++ b/.changeset/fix-stream-aggregate-within-idle.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Fix `Stream.aggregateWithin` and `Stream.groupedWithin` retaining fiber continuations on every schedule tick while upstream is idle. diff --git a/packages/effect/src/Stream.ts b/packages/effect/src/Stream.ts index c3bfc2040c9..2ba4ab55865 100644 --- a/packages/effect/src/Stream.ts +++ b/packages/effect/src/Stream.ts @@ -8427,13 +8427,13 @@ export const aggregateWithin: { let leftover: Arr.NonEmptyReadonlyArray | undefined let sinkHasInput = false const step = yield* Schedule.toStepWithSleep(schedule) - const stepToBuffer = Effect.suspend(function loop(): Pull.Pull { - return step(lastOutput).pipe( - Effect.flatMap(() => !sinkHasInput ? loop() : Queue.offer(buffer, scheduleStep)), - Effect.flatMap(() => Effect.never), - Pull.catchDone(() => Cause.done()) - ) + const stepLoop = Effect.suspend(function loop(): Pull.Pull { + return Effect.flatMap(step(lastOutput), () => !sinkHasInput ? loop() : Queue.offer(buffer, scheduleStep)) }) + const stepToBuffer: Pull.Pull = stepLoop.pipe( + Effect.flatMap(() => Effect.never), + Pull.catchDone(() => Cause.done()) + ) // buffer -> sink const pullFromBuffer: Pull.Pull< diff --git a/packages/effect/test/Stream.test.ts b/packages/effect/test/Stream.test.ts index 7e2912701c5..b33be4d4c3d 100644 --- a/packages/effect/test/Stream.test.ts +++ b/packages/effect/test/Stream.test.ts @@ -2251,6 +2251,31 @@ describe("Stream", () => { }) describe("aggregateWithin", () => { + it.effect("does not grow the fiber continuation stack while upstream is idle", () => + Effect.gen(function*() { + const continuationCounts: Array = [] + const schedule = Schedule.spaced("10 millis").pipe( + Schedule.tap(() => + Effect.withFiber((fiber) => + Effect.sync(() => { + continuationCounts.push( + (fiber as unknown as { readonly _stack: ReadonlyArray })._stack.length + ) + }) + ) + ) + ) + const fiber = yield* Stream.never.pipe( + Stream.aggregateWithin(Sink.take(25), schedule), + Stream.runDrain, + Effect.forkChild({ startImmediately: true }) + ) + yield* TestClock.adjust("1 second") + assert.isAbove(continuationCounts.length, 1) + assert.strictEqual(continuationCounts.at(-1), continuationCounts[0]) + yield* Fiber.interrupt(fiber) + })) + it.effect("groupedWithin does not emit empty arrays when upstream is idle", () => Effect.gen(function*() { const fiber = yield* Stream.never.pipe(