From c0e49d6ee2ae4b3ae65f8a6fbba3bad6f1748373 Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Wed, 30 Sep 2026 13:31:20 +0100 Subject: [PATCH 1/3] test(store-inmemory): show a slow subscriber stalling appends --- .../src/lib/inMemory.spec.ts | 40 +++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts b/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts index 513993aa..fc1ea0e2 100644 --- a/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts +++ b/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts @@ -98,3 +98,43 @@ describe('Subscriptions are live as soon as they return', () => { Effect.runPromise )); }); + +const lagBehindEvents = 1000; + +const appendOneEventAt = (store: EventStore) => (eventNumber: number) => + pipe( + ['event'], + Stream.fromIterable, + Stream.run(store.append({ streamId: decodeStreamId('lagging-stream'), eventNumber })) + ); + +const stallOnEveryEvent = () => Effect.never; + +const startSubscriberThatNeverDrains = (store: EventStore) => + pipe(store.subscribeAll(), Effect.flatMap(Stream.runForEach(stallOnEveryEvent)), Effect.fork); + +const appendManyEvents = (store: EventStore) => + Effect.forEach( + Array.from({ length: lagBehindEvents }, (_, i) => i), + appendOneEventAt(store), + { discard: true } + ); + +const runLaggingSubscriberTest = (store: EventStore) => + pipe( + startSubscriberThatNeverDrains(store), + Effect.andThen(Effect.yieldNow()), + Effect.andThen(appendManyEvents(store)), + Effect.timeout('2 seconds') + ); + +describe('A slow subscriber', () => { + it('should not stall appends however far it lags behind', () => + pipe( + StringEventStore, + Effect.flatMap(runLaggingSubscriberTest), + Effect.scoped, + Effect.provide(StringEventStoreLayer), + Effect.runPromise + )); +}); From 30b6f45705bbce43d495a11a6058a4fb9f381296 Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Wed, 30 Sep 2026 13:32:14 +0100 Subject: [PATCH 2/3] fix(store-inmemory): stop a slow subscriber stalling appends --- .changeset/inmemory-slow-subscriber.md | 7 +++++++ .../eventsourcing-store-inmemory/src/lib/InMemoryStore.ts | 3 +-- .../eventsourcing-store-inmemory/src/lib/inMemory.spec.ts | 3 ++- 3 files changed, 10 insertions(+), 3 deletions(-) create mode 100644 .changeset/inmemory-slow-subscriber.md diff --git a/.changeset/inmemory-slow-subscriber.md b/.changeset/inmemory-slow-subscriber.md new file mode 100644 index 00000000..f83f7904 --- /dev/null +++ b/.changeset/inmemory-slow-subscriber.md @@ -0,0 +1,7 @@ +--- +'@codeforbreakfast/eventsourcing-store-inmemory': patch +--- + +Fix appends stalling behind a slow subscriber. The store sized its subscription buffers with `2 ^ 8`, which is 10 in JavaScript, so a subscriber ten events behind blocked `append` while it held the store lock. A consumer that appended to a stream it was subscribed to could deadlock. + +Subscription buffers are now unbounded, so `append` never waits for a subscriber. A subscriber that never drains now grows memory instead of stalling appends. diff --git a/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts b/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts index 6b230309..a0423f1d 100644 --- a/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts +++ b/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts @@ -23,8 +23,7 @@ interface EventStream { const emptyStream = (): Effect.Effect, never, never> => pipe( - 2 ^ 8, - PubSub.bounded, + PubSub.unbounded(), Effect.map((pubsub) => ({ events: Chunk.empty(), pubsub, diff --git a/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts b/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts index fc1ea0e2..eb7bdc9c 100644 --- a/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts +++ b/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts @@ -122,7 +122,8 @@ const appendManyEvents = (store: EventStore) => const runLaggingSubscriberTest = (store: EventStore) => pipe( - startSubscriberThatNeverDrains(store), + store, + startSubscriberThatNeverDrains, Effect.andThen(Effect.yieldNow()), Effect.andThen(appendManyEvents(store)), Effect.timeout('2 seconds') From bcd776e49cf2f5d9a4c11052899dfa178e5d7444 Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Wed, 30 Sep 2026 17:28:37 +0100 Subject: [PATCH 3/3] fix(store-inmemory): restore the intended 256-event subscription buffer --- .changeset/inmemory-slow-subscriber.md | 4 ++-- .../eventsourcing-store-inmemory/src/lib/InMemoryStore.ts | 5 ++++- .../eventsourcing-store-inmemory/src/lib/inMemory.spec.ts | 4 ++-- 3 files changed, 8 insertions(+), 5 deletions(-) diff --git a/.changeset/inmemory-slow-subscriber.md b/.changeset/inmemory-slow-subscriber.md index f83f7904..29440816 100644 --- a/.changeset/inmemory-slow-subscriber.md +++ b/.changeset/inmemory-slow-subscriber.md @@ -2,6 +2,6 @@ '@codeforbreakfast/eventsourcing-store-inmemory': patch --- -Fix appends stalling behind a slow subscriber. The store sized its subscription buffers with `2 ^ 8`, which is 10 in JavaScript, so a subscriber ten events behind blocked `append` while it held the store lock. A consumer that appended to a stream it was subscribed to could deadlock. +Fix the subscription buffer size. The store meant to buffer 256 events per subscription but wrote `2 ^ 8`, which is 10 in JavaScript. A subscriber only ten events behind blocked `append`, and a consumer that appended to a stream it was subscribed to could deadlock. -Subscription buffers are now unbounded, so `append` never waits for a subscriber. A subscriber that never drains now grows memory instead of stalling appends. +The buffer is now 256. A subscriber that falls 256 events behind still blocks `append`. This is deliberate: the in-memory store is meant for tests and similar uses, where failing fast is better than growing memory without limit. diff --git a/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts b/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts index a0423f1d..10b1c211 100644 --- a/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts +++ b/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts @@ -21,9 +21,12 @@ interface EventStream { readonly pubsub: PubSub.PubSub; } +const subscriberBufferCapacity = 256; + const emptyStream = (): Effect.Effect, never, never> => pipe( - PubSub.unbounded(), + subscriberBufferCapacity, + PubSub.bounded, Effect.map((pubsub) => ({ events: Chunk.empty(), pubsub, diff --git a/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts b/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts index eb7bdc9c..eaa30000 100644 --- a/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts +++ b/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts @@ -99,7 +99,7 @@ describe('Subscriptions are live as soon as they return', () => { )); }); -const lagBehindEvents = 1000; +const lagBehindEvents = 100; const appendOneEventAt = (store: EventStore) => (eventNumber: number) => pipe( @@ -130,7 +130,7 @@ const runLaggingSubscriberTest = (store: EventStore) => ); describe('A slow subscriber', () => { - it('should not stall appends however far it lags behind', () => + it('should not stall appends while it lags behind by fewer events than the buffer holds', () => pipe( StringEventStore, Effect.flatMap(runLaggingSubscriberTest),