Hacktoberfest 2026：メンテナが10月に向けて印を付けた、オープンで初心者向けの issue。 Hacktoberfest の issue を見る

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

クローズ
  • 何か問題がありますか？
#9,987 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

まだ誰も着手していません。

評価

難易度
3/5
見積もり時間
1〜2日
初心者へのやさしさ
75/100
issue の種類
バグ
明瞭さ
明確に書かれている
活発さ
活発
技術スタック
typescript
領域
backend, performance

調査の方向性

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.

索引モデルが issue の本文から書いたものです。

説明

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.

主要言語
TypeScript
スター
16.7k
フォーク
808
平均マージ
11時間 11分
マージ済み PR（30日）
840

環境構築

このプロジェクトには開発コンテナ、Dockerfile、コントリビューションガイドがありません。まず README を読み、一般的な手順ははじめてのコントリビューションガイドを参照してください。

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

Effect-TS/effect のほかの issue

Effect-TS/effect の issue をすべて見る

似ている issue

TypeScript の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。