Fix aggregateWithin continuation growth

.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.

packages/effect/src/Stream.ts

@@ -8427,13 +8427,13 @@ export const aggregateWithin: {
     let leftover: Arr.NonEmptyReadonlyArray<A2> | undefined
     let sinkHasInput = false
     const step = yield* Schedule.toStepWithSleep(schedule)
-    const stepToBuffer = Effect.suspend(function loop(): Pull.Pull<never, E3, void, R3> {
-      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<void, E3, C | void, R3> {
+      return Effect.flatMap(step(lastOutput), () => !sinkHasInput ? loop() : Queue.offer(buffer, scheduleStep))
     })
+    const stepToBuffer: Pull.Pull<never, E3, void, R3> = stepLoop.pipe(
+      Effect.flatMap(() => Effect.never),
+      Pull.catchDone(() => Cause.done())
+    )
 
     // buffer -> sink
     const pullFromBuffer: Pull.Pull<

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<number> = []
+        const schedule = Schedule.spaced("10 millis").pipe(
+          Schedule.tap(() =>
+            Effect.withFiber((fiber) =>
+              Effect.sync(() => {
+                continuationCounts.push(
+                  (fiber as unknown as { readonly _stack: ReadonlyArray<unknown> })._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(