Hacktoberfest 2026: los issues que los mantenedores marcaron para octubre, abiertos y aptos para principiantes. Explorar issues de Hacktoberfest

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

Cerrado
  • ¿Algo no está bien?
#9,987 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
3/5
Tiempo estimado
1-2 días
Aptitud para principiantes
75/100
Tipo de issue
Error
Claridad
Bien especificado
Estado de actividad
Activo
Stack tecnológico
typescript
Área
backend, performance

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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.

Lenguaje dominante
TypeScript
Estrellas
16.7k
Forks
808
Merge medio
11 h 33 min
PR fusionados (30 d)
855

Preparar el entorno

Este proyecto no incluye contenedor de desarrollo, Dockerfile ni guía de contribución, así que la configuración corre por tu cuenta: empieza por su README y consulta nuestra guía para la primera contribución para los pasos generales.

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de Effect-TS/effect

Todos los issues de Effect-TS/effect

Issues similares

Más issues de TypeScript

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.