PubSub replay window pins the replay list — SubscriptionRef leaks every published value

[ibarakaiev]

What version of Effect is running?

4.0.0-beta.101 (still present in 4.0.0-beta.102 — ReplayWindowImpl is identical). The underlying bug is also latent in v3 (verified on 3.19.8, same ReplayWindowImpl), but there it only affects PubSubs created explicitly with { replay: n }. v4 made it universal for SubscriptionRef users because SubscriptionRef.make now builds on PubSub.unbounded({ replay: 1 }), whereas v3's SubscriptionRef used a plain PubSub.unbounded() and prepended the current value with Stream.concat — v3 SubscriptionRef is unaffected.

What steps can reproduce the bug?

Each subscription to a replay-enabled PubSub creates a ReplayWindowImpl that captures buffer.head at subscribe time and only advances it while draining the replay. ReplayBuffer.offer appends every published value to the same (append-only) list, so once the replay is drained the window's stale node pins the list and transitively retains every value published after the drain, for the life of the subscriber (a subscriber that never drains pins from subscribe time instead). One long-lived changes subscriber → unbounded heap growth, even though the subscriber consumes everything promptly.

Run with node --expose-gc; heap grows ~400 KB/s (measured on Node 20/22, macOS). Remove the subscriber (or apply the fix below) and it is flat. The same repro against effect 3.19.8 (using ref.changes / Effect.forkDaemon) is flat, and an explicit PubSub.unbounded({ replay: 1 }) + one draining subscriber leaks identically on 3.19.8:

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

const fat = (i: number) => ({
  seq: i,
  units: Array.from({ length: 40 }, (_, k) => ({ id: `u${k}`, state: { seq: i } }))
})

const program = Effect.gen(function* () {
  const ref = yield* SubscriptionRef.make(fat(0))
  // one healthy, constantly-draining subscriber is enough
  yield* Effect.forkDetach(SubscriptionRef.changes(ref).pipe(Stream.runForEach(() => Effect.void)))
  yield* Effect.forkDetach(Effect.gen(function* () {
    let i = 0
    while (true) {
      yield* SubscriptionRef.set(ref, fat(++i))
      yield* Effect.sleep(Duration.millis(10))
    }
  }))
  yield* Effect.repeat(Effect.sync(() => {
    ;(globalThis as any).gc?.()
    console.log(`heap=${(process.memoryUsage().heapUsed / 1048576).toFixed(1)}MB`)
  }), Schedule.spaced(Duration.seconds(5)))
})

Effect.runPromise(program)

Heap-snapshot retainer chain for a leaked value:

value ← {value,next} replay node ← (.next × N) ← ReplayWindowImpl.head ← SubscriptionImpl.replayWindow

We hit this in production the day we deployed our v3→v4 migration: a backend publishing device state into SubscriptionRefs at telemetry rate leaked ~28 MB/min until it hit --max-old-space-size. The same workload on v3 was flat.

What is the expected behavior?

Consumed (or slid-out) replay values become collectable; a subscription that stays open indefinitely retains at most the replay window's own values.

Suggested fix

The window doesn't need to store a node reference at all — derive the cursor as buffer.head advanced by the number of values already taken (walk bounded by replay capacity, i.e. 1 for SubscriptionRef). Slid-out values are skipped automatically because buffer.head itself advances, which also removes the fastForward bookkeeping:

class ReplayWindowImpl<A> implements PubSub.ReplayWindow<A> {
  taken = 0
  remaining: number
  readonly buffer: ReplayBuffer<A>

  constructor(buffer: ReplayBuffer<A>) {
    this.buffer = buffer
    this.remaining = buffer.size
  }
  take(): A | undefined {
    if (this.remaining === 0) return undefined
    let node = this.buffer.head
    for (let i = 0; i < this.taken; i++) node = node.next!
    this.taken++
    this.remaining--
    return node.value as A
  }
  takeN(n: number): Array<A> {
    const len = Math.min(n, this.remaining)
    const items = new Array(len)
    for (let i = 0; i < len; i++) items[i] = this.take()
    return items
  }
  takeAll(): Array<A> {
    return this.takeN(this.remaining)
  }
}

We are running exactly this as a pnpm patch in production: the leaking repro goes flat, a late first take on an idle subscription still delivers the latest value (matching the old fastForward behavior — values that slid out are skipped), and closed idle subscriptions release their queue backlog as before.

Platform

• Node v20 / v22, macOS + Linux

[pullfrog]

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