fix(store-inmemory): restore the intended 256-event subscription buffer

.changeset/inmemory-slow-subscriber.md

@@ -0,0 +1,7 @@
+---
+'@codeforbreakfast/eventsourcing-store-inmemory': patch
+---
+
+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.
+
+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.

packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts

@@ -21,9 +21,11 @@ interface EventStream<V> {
   readonly pubsub: PubSub.PubSub<V>;
 }
 
+const subscriberBufferCapacity = 256;
+
 const emptyStream = <V>(): Effect.Effect<EventStream<V>, never, never> =>
   pipe(
-    2 ^ 8,
+    subscriberBufferCapacity,
     PubSub.bounded<V>,
     Effect.map((pubsub) => ({
       events: Chunk.empty<V>(),

packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts

@@ -98,3 +98,44 @@
       Effect.runPromise
     ));
 });
+
+const lagBehindEvents = 100;
+
+const appendOneEventAt = (store: EventStore<string>) => (eventNumber: number) =>
+  pipe(
+    ['event'],
+    Stream.fromIterable,
+    Stream.run(store.append({ streamId: decodeStreamId('lagging-stream'), eventNumber }))
+  );
+
+const stallOnEveryEvent = () => Effect.never;
+
+const startSubscriberThatNeverDrains = (store: EventStore<string>) =>
+  pipe(store.subscribeAll(), Effect.flatMap(Stream.runForEach(stallOnEveryEvent)), Effect.fork);
+
+const appendManyEvents = (store: EventStore<string>) =>
+  Effect.forEach(
+    Array.from({ length: lagBehindEvents }, (_, i) => i),
+    appendOneEventAt(store),
+    { discard: true }
+  );
+
+const runLaggingSubscriberTest = (store: EventStore<string>) =>
+  pipe(
+    store,
+    startSubscriberThatNeverDrains,
+    Effect.andThen(Effect.yieldNow()),
+    Effect.andThen(appendManyEvents(store)),
+    Effect.timeout('2 seconds')
+  );
+
+describe('A slow subscriber', () => {
+  it('should not stall appends while it lags behind by fewer events than the buffer holds', () =>
+    pipe(
+      StringEventStore,
+      Effect.flatMap(runLaggingSubscriberTest),
+      Effect.scoped,
+      Effect.provide(StringEventStoreLayer),
+      Effect.runPromise
+    ));
+});