From 8a9f1036c662a2352c9915ae6ae50076ab05e053 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 14:41:18 +1200 Subject: [PATCH 1/3] Fix PubSub replay value retention --- .changeset/fix-pubsub-replay-retention.md | 5 ++ packages/effect/src/PubSub.ts | 94 ++++++++++++++--------- packages/effect/test/PubSub.test.ts | 65 ++++++++++++++++ 3 files changed, 128 insertions(+), 36 deletions(-) create mode 100644 .changeset/fix-pubsub-replay-retention.md diff --git a/.changeset/fix-pubsub-replay-retention.md b/.changeset/fix-pubsub-replay-retention.md new file mode 100644 index 00000000000..632765fdcf7 --- /dev/null +++ b/.changeset/fix-pubsub-replay-retention.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Prevent replay-enabled PubSubs from retaining values beyond each subscription's replay window. diff --git a/packages/effect/src/PubSub.ts b/packages/effect/src/PubSub.ts index c7f0d62b133..66f93637380 100644 --- a/packages/effect/src/PubSub.ts +++ b/packages/effect/src/PubSub.ts @@ -1115,6 +1115,9 @@ const unsubscribe = (self: Subscription): Effect.Effect => Effect.sync(() => { self.subscribers.delete(self.subscription) self.subscription.unsubscribe() + if (self.replayWindow instanceof ReplayWindowImpl) { + self.replayWindow.close() + } self.strategy.onPubSubEmptySpaceUnsafe(self.pubsub, self.subscribers) }) ), @@ -1606,12 +1609,11 @@ class BoundedPubSubArb implements PubSub.Atomic { slide(): void { if (this.subscribersIndex !== this.publisherIndex) { const index = this.subscribersIndex % this.capacity + const value = this.array[index] this.array[index] = AbsentValue as unknown as A this.subscribers[index] = 0 this.subscribersIndex += 1 - } - if (this.replayBuffer) { - this.replayBuffer.slide() + this.replayBuffer?.slide(value) } } @@ -1799,12 +1801,11 @@ class BoundedPubSubPow2 implements PubSub.Atomic { slide(): void { if (this.subscribersIndex !== this.publisherIndex) { const index = this.subscribersIndex & this.mask + const value = this.array[index] this.array[index] = AbsentValue as unknown as A this.subscribers[index] = 0 this.subscribersIndex += 1 - } - if (this.replayBuffer) { - this.replayBuffer.slide() + this.replayBuffer?.slide(value) } } @@ -1975,11 +1976,10 @@ class BoundedPubSubSingle implements PubSub.Atomic { slide(): void { if (this.isFull()) { + const value = this.value this.subscribers = 0 this.value = AbsentValue as unknown as A - } - if (this.replayBuffer) { - this.replayBuffer.slide() + this.replayBuffer?.slide(value) } } @@ -2125,12 +2125,11 @@ class UnboundedPubSub implements PubSub.Atomic { slide(): void { if (this.publisherHead !== this.publisherTail) { + const value = this.publisherHead.next!.value as A this.publisherHead = this.publisherHead.next! this.publisherHead.value = AbsentValue this.subscribersIndex += 1 - } - if (this.replayBuffer) { - this.replayBuffer.slide() + this.replayBuffer?.slide(value) } } @@ -2709,6 +2708,7 @@ class ReplayBuffer { readonly capacity: number head: ReplayNode = { value: AbsentValue, next: null } tail: ReplayNode = this.head + readonly slideValues: Array = [] size = 0 index = 0 @@ -2716,7 +2716,8 @@ class ReplayBuffer { this.capacity = capacity } - slide() { + slide(value: A): void { + this.slideValues[this.index % this.capacity] = value this.index++ } offer(a: A): void { @@ -2740,48 +2741,69 @@ class ReplayBuffer { } class ReplayWindowImpl implements PubSub.ReplayWindow { - head: ReplayNode - index: number - remaining: number readonly buffer: ReplayBuffer + readonly values: Array + index = 0 + remaining: number + slideIndex: number constructor(buffer: ReplayBuffer) { this.buffer = buffer - this.index = buffer.index this.remaining = buffer.size - this.head = buffer.head - } - fastForward() { - while (this.index < this.buffer.index) { - this.head = this.head.next! - this.index++ + this.slideIndex = buffer.index + this.values = new Array(this.remaining) + let node = buffer.head + for (let i = 0; i < this.remaining; i++) { + this.values[i] = node.value as A + node = node.next! + } + } + close(): void { + this.values.length = 0 + this.remaining = 0 + } + sync(): void { + const slides = this.buffer.index - this.slideIndex + if (slides === 0 || this.remaining === 0) { + return + } + const count = Math.min(slides, this.remaining) + const start = this.buffer.index - count + if (slides >= this.remaining) { + this.values.fill(AbsentValue as unknown as A) + this.index = 0 + for (let i = 0; i < count; i++) { + this.values[i] = this.buffer.slideValues[(start + i) % this.buffer.capacity] + } + } else { + for (let i = 0; i < count; i++) { + this.index = (this.index + 1) % this.values.length + this.values[(this.index + this.remaining - 1) % this.values.length] = + this.buffer.slideValues[(start + i) % this.buffer.capacity] + } } + this.slideIndex = this.buffer.index } take(): A | undefined { if (this.remaining === 0) { return undefined - } else if (this.index < this.buffer.index) { - this.fastForward() } + this.sync() + const value = this.values[this.index] + this.values[this.index] = AbsentValue as unknown as A + this.index = (this.index + 1) % this.values.length this.remaining-- - const value = this.head.value - this.head = this.head.next! + if (this.remaining === 0) { + this.close() + } return value as A } takeN(n: number): Array { - if (this.remaining === 0) { - return [] - } else if (this.index < this.buffer.index) { - this.fastForward() - } const len = Math.min(n, this.remaining) const items = new Array(len) for (let i = 0; i < len; i++) { - const value = this.head.value as A - this.head = this.head.next! - items[i] = value + items[i] = this.take()! } - this.remaining -= len return items } takeAll(): Array { diff --git a/packages/effect/test/PubSub.test.ts b/packages/effect/test/PubSub.test.ts index bf1b79172d0..ba34487a543 100644 --- a/packages/effect/test/PubSub.test.ts +++ b/packages/effect/test/PubSub.test.ts @@ -2,6 +2,29 @@ import { assert, describe, it } from "@effect/vitest" import { Array, Effect, Exit, Fiber, Latch, PubSub, Stream } from "effect" import { pipe } from "effect/Function" +const retains = (root: object, target: object): boolean => { + const objects = [root] + const seen = new Set() + while (objects.length > 0) { + const current = objects.pop()! + if (current === target) { + return true + } + if (seen.has(current)) { + continue + } + seen.add(current) + for (const key of Reflect.ownKeys(current)) { + const descriptor = Object.getOwnPropertyDescriptor(current, key) + const value = descriptor && "value" in descriptor ? descriptor.value : undefined + if (typeof value === "object" && value !== null) { + objects.push(value) + } + } + } + return false +} + describe("PubSub", () => { it.effect("publishAll - capacity 2 (BoundedPubSubPow2)", () => { const messages = [1, 2] @@ -399,6 +422,48 @@ describe("PubSub", () => { })) describe("replay", () => { + it("does not retain values published after the replay window is drained", () => { + const pubsub = PubSub.makeAtomicUnbounded({ replay: 1 }) + pubsub.publish({}) + const replayWindow = pubsub.replayWindow() + replayWindow.take() + + const slidOut = {} + pubsub.publish(slidOut) + pubsub.publish({}) + + assert.isFalse(retains(replayWindow, slidOut)) + }) + + it("does not retain values published outside an undrained replay window", () => { + const pubsub = PubSub.makeAtomicUnbounded({ replay: 1 }) + const replayed = {} + pubsub.publish(replayed) + const replayWindow = pubsub.replayWindow() + + const slidOut = {} + pubsub.publish(slidOut) + pubsub.publish({}) + + assert.isFalse(retains(replayWindow, slidOut)) + assert.strictEqual(replayWindow.take(), replayed) + }) + + it("preserves replay order across multiple slides", () => { + const pubsub = PubSub.makeAtomicBounded({ capacity: 4, replay: 3 }) + pubsub.publishAll([1, 2, 3, 4, 5]) + const subscription = pubsub.subscribe() + const replayWindow = pubsub.replayWindow() + pubsub.publishAll([6, 7, 8, 9]) + for (const value of [10, 11, 12]) { + pubsub.slide() + pubsub.publish(value) + } + + assert.deepStrictEqual(replayWindow.takeAll(), [6, 7, 8]) + assert.deepStrictEqual(subscription.pollUpTo(Number.POSITIVE_INFINITY), [9, 10, 11, 12]) + }) + it.effect("unbounded", () => Effect.gen(function*() { const messages = [1, 2, 3, 4, 5] From 7f0d6ffc1996f36a8b76356501bb46c1838f8bb4 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 15:08:44 +1200 Subject: [PATCH 2/3] Preserve PubSub replay ordering --- packages/effect/src/PubSub.ts | 111 +++++++++++++++++----------- packages/effect/test/PubSub.test.ts | 60 +++++++++------ 2 files changed, 104 insertions(+), 67 deletions(-) diff --git a/packages/effect/src/PubSub.ts b/packages/effect/src/PubSub.ts index 66f93637380..22ef4a60831 100644 --- a/packages/effect/src/PubSub.ts +++ b/packages/effect/src/PubSub.ts @@ -139,6 +139,7 @@ export declare namespace PubSub { take(): A | undefined takeN(n: number): Array takeAll(): Array + close(): void readonly remaining: number } @@ -1115,9 +1116,7 @@ const unsubscribe = (self: Subscription): Effect.Effect => Effect.sync(() => { self.subscribers.delete(self.subscription) self.subscription.unsubscribe() - if (self.replayWindow instanceof ReplayWindowImpl) { - self.replayWindow.close() - } + self.replayWindow.close() self.strategy.onPubSubEmptySpaceUnsafe(self.pubsub, self.subscribers) }) ), @@ -1529,6 +1528,7 @@ const makeSubscriptionUnsafe = ( class BoundedPubSubArb implements PubSub.Atomic { array: Array + replayIndices: Array publisherIndex = 0 subscribers: Array subscriberCount = 0 @@ -1541,6 +1541,7 @@ class BoundedPubSubArb implements PubSub.Atomic { this.capacity = capacity this.replayBuffer = replayBuffer this.array = Array.from({ length: capacity }) + this.replayIndices = replayBuffer ? Array.from({ length: capacity }) : [] this.subscribers = Array.from({ length: capacity }) } @@ -1564,15 +1565,16 @@ class BoundedPubSubArb implements PubSub.Atomic { if (this.isFull()) { return false } + const replayIndex = this.replayBuffer?.offer(value) if (this.subscriberCount !== 0) { const index = this.publisherIndex % this.capacity this.array[index] = value + if (replayIndex !== undefined) { + this.replayIndices[index] = replayIndex + } this.subscribers[index] = this.subscriberCount this.publisherIndex += 1 } - if (this.replayBuffer) { - this.replayBuffer.offer(value) - } return true } @@ -1597,11 +1599,12 @@ class BoundedPubSubArb implements PubSub.Atomic { const a = chunk[iteratorIndex++] const index = this.publisherIndex % this.capacity this.array[index] = a + const replayIndex = this.replayBuffer?.offer(a) + if (replayIndex !== undefined) { + this.replayIndices[index] = replayIndex + } this.subscribers[index] = this.subscriberCount this.publisherIndex += 1 - if (this.replayBuffer) { - this.replayBuffer.offer(a) - } } return chunk.slice(iteratorIndex) } @@ -1613,7 +1616,7 @@ class BoundedPubSubArb implements PubSub.Atomic { this.array[index] = AbsentValue as unknown as A this.subscribers[index] = 0 this.subscribersIndex += 1 - this.replayBuffer?.slide(value) + this.replayBuffer?.slide(value, this.replayIndices[index]) } } @@ -1719,6 +1722,7 @@ class BoundedPubSubArbSubscription implements PubSub.BackingSubscripti class BoundedPubSubPow2 implements PubSub.Atomic { array: Array + replayIndices: Array mask: number publisherIndex = 0 subscribers: Array @@ -1732,6 +1736,7 @@ class BoundedPubSubPow2 implements PubSub.Atomic { this.capacity = capacity this.replayBuffer = replayBuffer this.array = Array.from({ length: capacity }) + this.replayIndices = replayBuffer ? Array.from({ length: capacity }) : [] this.mask = capacity - 1 this.subscribers = Array.from({ length: capacity }) } @@ -1756,15 +1761,16 @@ class BoundedPubSubPow2 implements PubSub.Atomic { if (this.isFull()) { return false } + const replayIndex = this.replayBuffer?.offer(value) if (this.subscriberCount !== 0) { const index = this.publisherIndex & this.mask this.array[index] = value + if (replayIndex !== undefined) { + this.replayIndices[index] = replayIndex + } this.subscribers[index] = this.subscriberCount this.publisherIndex += 1 } - if (this.replayBuffer) { - this.replayBuffer.offer(value) - } return true } @@ -1789,11 +1795,12 @@ class BoundedPubSubPow2 implements PubSub.Atomic { const elem = chunk[iteratorIndex++] const index = this.publisherIndex & this.mask this.array[index] = elem + const replayIndex = this.replayBuffer?.offer(elem) + if (replayIndex !== undefined) { + this.replayIndices[index] = replayIndex + } this.subscribers[index] = this.subscriberCount this.publisherIndex += 1 - if (this.replayBuffer) { - this.replayBuffer.offer(elem) - } } return chunk.slice(iteratorIndex) } @@ -1805,7 +1812,7 @@ class BoundedPubSubPow2 implements PubSub.Atomic { this.array[index] = AbsentValue as unknown as A this.subscribers[index] = 0 this.subscribersIndex += 1 - this.replayBuffer?.slide(value) + this.replayBuffer?.slide(value, this.replayIndices[index]) } } @@ -1913,6 +1920,7 @@ class BoundedPubSubSingle implements PubSub.Atomic { subscriberCount = 0 subscribers = 0 value: A = AbsentValue as unknown as A + replayIndex = 0 readonly capacity = 1 readonly replayBuffer: ReplayBuffer | undefined @@ -1945,14 +1953,15 @@ class BoundedPubSubSingle implements PubSub.Atomic { if (this.isFull()) { return false } + const replayIndex = this.replayBuffer?.offer(value) if (this.subscriberCount !== 0) { this.value = value + if (replayIndex !== undefined) { + this.replayIndex = replayIndex + } this.subscribers = this.subscriberCount this.publisherIndex += 1 } - if (this.replayBuffer) { - this.replayBuffer.offer(value) - } return true } @@ -1979,7 +1988,7 @@ class BoundedPubSubSingle implements PubSub.Atomic { const value = this.value this.subscribers = 0 this.value = AbsentValue as unknown as A - this.replayBuffer?.slide(value) + this.replayBuffer?.slide(value, this.replayIndex) } } @@ -2058,6 +2067,7 @@ class BoundedPubSubSingleSubscription implements PubSub.BackingSubscri interface Node { value: A | AbsentValue + replayIndex?: number subscribers: number next: Node | null } @@ -2096,19 +2106,21 @@ class UnboundedPubSub implements PubSub.Atomic { } publish(value: A): boolean { + const replayIndex = this.replayBuffer?.offer(value) const subscribers = this.publisherTail.subscribers if (subscribers !== 0) { - this.publisherTail.next = { + const node: Node = { value, subscribers, next: null } + if (replayIndex !== undefined) { + node.replayIndex = replayIndex + } + this.publisherTail.next = node this.publisherTail = this.publisherTail.next this.publisherIndex += 1 } - if (this.replayBuffer) { - this.replayBuffer.offer(value) - } return true } @@ -2125,11 +2137,12 @@ class UnboundedPubSub implements PubSub.Atomic { slide(): void { if (this.publisherHead !== this.publisherTail) { - const value = this.publisherHead.next!.value as A + const node = this.publisherHead.next! + const value = node.value as A this.publisherHead = this.publisherHead.next! this.publisherHead.value = AbsentValue this.subscribersIndex += 1 - this.replayBuffer?.slide(value) + this.replayBuffer?.slide(value, node.replayIndex!) } } @@ -2701,29 +2714,40 @@ const strategyCompleteSubscribersUnsafe = ( interface ReplayNode { value: A | AbsentValue + index: number next: ReplayNode | null } class ReplayBuffer { readonly capacity: number - head: ReplayNode = { value: AbsentValue, next: null } + head: ReplayNode = { value: AbsentValue, index: 0, next: null } tail: ReplayNode = this.head - readonly slideValues: Array = [] + readonly slideValues: Array<{ + readonly value: A + readonly index: number + }> = [] size = 0 index = 0 + publisherIndex = 0 constructor(capacity: number) { this.capacity = capacity } - slide(value: A): void { - this.slideValues[this.index % this.capacity] = value + slide(value: A, publisherIndex: number): void { + this.slideValues[this.index % this.capacity] = { + value, + index: publisherIndex + } this.index++ } - offer(a: A): void { + offer(a: A): number { + const index = this.publisherIndex++ this.tail.value = a + this.tail.index = index this.tail.next = { value: AbsentValue, + index: 0, next: null } this.tail = this.tail.next @@ -2732,6 +2756,7 @@ class ReplayBuffer { } else { this.size += 1 } + return index } offerAll(as: Iterable): void { for (const a of as) { @@ -2746,6 +2771,7 @@ class ReplayWindowImpl implements PubSub.ReplayWindow { index = 0 remaining: number slideIndex: number + newestIndex = -1 constructor(buffer: ReplayBuffer) { this.buffer = buffer @@ -2755,6 +2781,7 @@ class ReplayWindowImpl implements PubSub.ReplayWindow { let node = buffer.head for (let i = 0; i < this.remaining; i++) { this.values[i] = node.value as A + this.newestIndex = node.index node = node.next! } } @@ -2767,19 +2794,14 @@ class ReplayWindowImpl implements PubSub.ReplayWindow { if (slides === 0 || this.remaining === 0) { return } - const count = Math.min(slides, this.remaining) + const count = Math.min(slides, this.buffer.capacity) const start = this.buffer.index - count - if (slides >= this.remaining) { - this.values.fill(AbsentValue as unknown as A) - this.index = 0 - for (let i = 0; i < count; i++) { - this.values[i] = this.buffer.slideValues[(start + i) % this.buffer.capacity] - } - } else { - for (let i = 0; i < count; i++) { + for (let i = 0; i < count; i++) { + const entry = this.buffer.slideValues[(start + i) % this.buffer.capacity] + if (entry.index > this.newestIndex) { this.index = (this.index + 1) % this.values.length - this.values[(this.index + this.remaining - 1) % this.values.length] = - this.buffer.slideValues[(start + i) % this.buffer.capacity] + this.values[(this.index + this.remaining - 1) % this.values.length] = entry.value + this.newestIndex = entry.index } } this.slideIndex = this.buffer.index @@ -2815,5 +2837,6 @@ const emptyReplayWindow: PubSub.ReplayWindow = { remaining: 0, take: () => undefined, takeN: () => [], - takeAll: () => [] + takeAll: () => [], + close: () => void 0 } diff --git a/packages/effect/test/PubSub.test.ts b/packages/effect/test/PubSub.test.ts index ba34487a543..02972b61173 100644 --- a/packages/effect/test/PubSub.test.ts +++ b/packages/effect/test/PubSub.test.ts @@ -2,29 +2,6 @@ import { assert, describe, it } from "@effect/vitest" import { Array, Effect, Exit, Fiber, Latch, PubSub, Stream } from "effect" import { pipe } from "effect/Function" -const retains = (root: object, target: object): boolean => { - const objects = [root] - const seen = new Set() - while (objects.length > 0) { - const current = objects.pop()! - if (current === target) { - return true - } - if (seen.has(current)) { - continue - } - seen.add(current) - for (const key of Reflect.ownKeys(current)) { - const descriptor = Object.getOwnPropertyDescriptor(current, key) - const value = descriptor && "value" in descriptor ? descriptor.value : undefined - if (typeof value === "object" && value !== null) { - objects.push(value) - } - } - } - return false -} - describe("PubSub", () => { it.effect("publishAll - capacity 2 (BoundedPubSubPow2)", () => { const messages = [1, 2] @@ -539,6 +516,20 @@ describe("PubSub", () => { const sub3 = yield* PubSub.subscribe(pubsub) assert.deepStrictEqual(yield* PubSub.takeAll(sub3), [14, 15, 16]) })) + + it.effect("sliding preserves publish order with a lagging subscriber", () => + Effect.scoped( + Effect.gen(function*() { + const pubsub = yield* PubSub.sliding({ capacity: 4, replay: 3 }) + yield* PubSub.subscribe(pubsub) + yield* PubSub.publishAll(pubsub, [1, 2]) + const subscription = yield* PubSub.subscribe(pubsub) + yield* PubSub.publishAll(pubsub, [3, 4, 5]) + + const values = yield* PubSub.takeAll(subscription) + assert.isTrue(values.every((value, index) => index === 0 || values[index - 1] <= value)) + }) + )) }) it.effect("shutdown interrupts suspended subscribers", () => @@ -615,3 +606,26 @@ describe("PubSub", () => { }) )) }) + +const retains = (root: object, target: object): boolean => { + const objects = [root] + const seen = new Set() + while (objects.length > 0) { + const current = objects.pop()! + if (current === target) { + return true + } + if (seen.has(current)) { + continue + } + seen.add(current) + for (const key of Reflect.ownKeys(current)) { + const descriptor = Object.getOwnPropertyDescriptor(current, key) + const value = descriptor && "value" in descriptor ? descriptor.value : undefined + if (typeof value === "object" && value !== null) { + objects.push(value) + } + } + } + return false +} From 9837b66b9f254674b40a5449f33e03e00f59e570 Mon Sep 17 00:00:00 2001 From: Tim Smart Date: Fri, 31 Jul 2026 16:07:52 +1200 Subject: [PATCH 3/3] Avoid unbounded PubSub node shape transitions --- packages/effect/src/PubSub.ts | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/packages/effect/src/PubSub.ts b/packages/effect/src/PubSub.ts index 22ef4a60831..dc9ce053179 100644 --- a/packages/effect/src/PubSub.ts +++ b/packages/effect/src/PubSub.ts @@ -2067,7 +2067,7 @@ class BoundedPubSubSingleSubscription implements PubSub.BackingSubscri interface Node { value: A | AbsentValue - replayIndex?: number + replayIndex: number | undefined subscribers: number next: Node | null } @@ -2075,6 +2075,7 @@ interface Node { class UnboundedPubSub implements PubSub.Atomic { publisherHead: Node = { value: AbsentValue, + replayIndex: undefined, subscribers: 0, next: null } @@ -2111,12 +2112,10 @@ class UnboundedPubSub implements PubSub.Atomic { if (subscribers !== 0) { const node: Node = { value, + replayIndex, subscribers, next: null } - if (replayIndex !== undefined) { - node.replayIndex = replayIndex - } this.publisherTail.next = node this.publisherTail = this.publisherTail.next this.publisherIndex += 1