Stream.aggregateWithin / groupedWithin grows the fiber continuation chain on every schedule tick while idle

[ibarakaiev]

What version of Effect is running?

4.0.0-beta.101 (code unchanged in 4.0.0-beta.102)

What steps can reproduce the bug?

aggregateWithin's schedule loop re-enters itself through nested flatMap when the sink has received no input:

const stepToBuffer = Effect.suspend(function loop() {
  return step(lastOutput).pipe(
    Effect.flatMap(() => !sinkHasInput ? loop() : Queue.offer(buffer, scheduleStep)),
    Effect.flatMap(() => Effect.never),
    Pull.catchDone(() => Cause.done())
  )
})

Because the recursion happens inside the first flatMap, every iteration pushes the remaining two continuations (flatMap(() => Effect.never) and catchDone) onto the fiber's continuation chain and they never pop — they only unwind when input finally arrives and the window's raceFirst settles. A stream that receives no input therefore grows its fiber stack on every schedule tick, forever. The leak is proportional to idleness, not load: a telemetry batcher on a quiet source (Stream.fromQueue(q).pipe(Stream.groupedWithin(5000, "1 second"), …)) leaks continuously while doing nothing.

Repro (run with node --expose-gc; grows ~100 KB/s at this schedule rate, flat after the fix below):

import { Effect, Stream, Duration, Schedule } from "effect"

const program = Effect.gen(function* () {
  yield* Effect.forkDetach(
    Stream.never.pipe(
      Stream.groupedWithin(5000, Duration.millis(10)),
      Stream.runDrain
    )
  )
  yield* Effect.repeat(Effect.sync(() => {
    ;(globalThis as any).gc?.()
    console.log(`heap=${(process.memoryUsage().heapUsed / 1048576).toFixed(2)}MB`)
  }), Schedule.spaced(Duration.seconds(5)))
})

Effect.runPromise(program)

We found this after fixing #6804: an idle 1 Hz groupedWithin batch consumer retained ~3–4 MB/hour per idle stream. A 60-minute sampling heap profile attributed 55% of retained bytes to the loop body in Stream.js (aggregateWithin), with no application frames on the stack.

What is the expected behavior?

An idle aggregateWithin/groupedWithin stream runs in constant memory.

Suggested fix

Recurse from the innermost continuation only, so stack depth stays constant; wrap Effect.never and catchDone around the loop once:

const stepLoop = Effect.suspend(function loop(): Pull.Pull<void, E3, 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())
)

Semantics are unchanged (Cause.done from the schedule still reaches the single catchDone; size- and time-triggered flushes behave identically in our tests). We're running this as a pnpm patch: the idle repro goes flat and batching behavior is unchanged.

Platform

Node v20 / v22, macOS + Linux

[pullfrog]

  ^{｜ Build this ➔ ｜ Make a plan ➔}