Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

PubSub replay window retains every message published after a subscription (Stream.share({ replay }) memory leak)

Aperta
  • Qualcosa non va?
#9,987 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
3/5
Tempo stimato
1-2 giorni
Idoneità per principianti
75/100
Tipo di issue
Bug
Chiarezza
Specificata chiaramente
Stato di attività
Attiva
Stack tecnologico
typescript
Ambito
backend, performance

Direzione di ricerca

Start in src/internal/pubsub.ts and trace ReplayWindowImpl's constructor, take(), and takeN() against the supplied reproduction. Check how each path releases its reference when the replay window is exhausted, then run the reproduction with garbage collection enabled. Done means subscribers still receive all values and late subscribers still receive replayed values, while consumed values are no longer retained.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

A subscription to a PubSub created with replay keeps a reference to a node of the replay buffer’s linked list for as long as the subscription lives. Every message published after that node stays reachable through next, so a long-lived subscriber leaks everything that is published, even though it consumes each message.

Minimal reproduction

[email protected] (also 3.21.4), Node 22/24, run with node --expose-gc repro.mjs:

`import { Effect, Stream, SubscriptionRef } from "effect"

const program = Effect.scoped(Effect.gen(function* () {
const ref = yield* SubscriptionRef.make(new Uint8Array(100_000))
const shared = yield* ref.changes.pipe(Stream.share({ capacity: "unbounded", replay: 1 }))

// A long-lived subscriber that keeps up with every value.
let seen = 0
yield* shared.pipe(Stream.runForEach(() => Effect.sync(() => { seen++ })), Effect.fork)
yield* Effect.sleep("20 millis")

for (let i = 0; i < 2000; i++) {
yield* SubscriptionRef.set(ref, new Uint8Array(100_000)) // 100 KB each
if (i % 100 === 0) yield* Effect.yieldNow()
}
yield* Effect.sleep("100 millis")
globalThis.gc()
const mib = Math.round(process.memoryUsage().arrayBuffers / 1048576)
console.log(subscriber saw ${seen} values; arrayBuffers retained: ${mib} MiB)
}))

Effect.runPromise(program)`

Output:

subscriber saw 2001 values; arrayBuffers retained: 191 MiB

All 2000 values are still alive after GC. With the change below the same script retains about 10 MiB.

Cause

ReplayWindowImpl in src/internal/pubsub.ts:

constructor(buffer) { this.buffer = buffer this.index = buffer.index this.remaining = buffer.size this.head = buffer.head // (1) } take() { if (this.remaining === 0) return undefined ... this.remaining-- const value = this.head.value this.head = this.head.next // (2) return value }

head is only read while remaining > 0, and remaining never grows. Once it reaches 0, or if it starts at 0 because the buffer was empty when the subscription was made, head is never read again but is never cleared either: (1) or (2) leave it pointing at the buffer’s current tail. The replay buffer keeps appending to that list, so the subscription retains every later message.

The repro hits case (1). The first subscriber of Stream.share subscribes before anything has been published. A late subscriber that drains its replay through (2) or takeN leaks the same way. We checked this separately: with only the constructor fixed, a late subscriber still retains the full 191 MiB, and with all three changes it retains 9 MiB.

A heap snapshot of the affected process shows the retaining path:

FiberRuntime → … → SubscriptionImpl → replayWindow (ReplayWindowImpl) → head → next → next → … (thousands) → value (TakeImpl) → exit → … → <the published value>

Suggested fix

Drop the reference as soon as the window is exhausted:

` constructor(buffer) {
this.buffer = buffer
this.index = buffer.index
this.remaining = buffer.size

  • this.head = buffer.head
  • this.head = this.remaining === 0 ? undefined : buffer.head
    }
    take() {
    ...
    this.remaining--
    const value = this.head.value
  • this.head = this.head.next
  • this.head = this.remaining === 0 ? undefined : this.head.next
    return value
    }
    takeN(n) {
    ...
    this.remaining -= len
  • if (this.remaining === 0) this.head = undefined
    return Chunk.unsafeFromArray(items)
    }`

We run this as a patch in production. Every value still reaches the subscriber, and a late subscriber still gets the replayed value.

Impact

Any Stream.share / Stream.broadcast* or PubSub with replay that has a long-lived subscriber. We found it through the Midnight wallet SDK, which shares wallet state this way. A service that stays synced leaks every wallet state it receives, about 55 MB/h, until it runs out of memory.

Versions

Reproduced: [email protected], [email protected].
[email protected] rewrote ReplayWindowImpl to copy the values into an array instead of holding a node, so it is probably not affected. We have not tested it.

Lingua principale
TypeScript
Stelle
16.7k
Fork
808
Merge medio
10h 49m
PR unite (30g)
846

Preparare l'ambiente

Questo progetto non fornisce container di sviluppo, Dockerfile né guida per i contributori, quindi l'ambiente è a tuo carico: parti dal suo README e consulta la nostra guida al primo contributo per i passaggi generali.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di Effect-TS/effect

Tutte le issue di Effect-TS/effect

Issue simili

Altre issue su TypeScript

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.