From c19b861737219d9f0cb3fc691c4467301a93688e Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Wed, 30 Sep 2026 13:00:01 +0100 Subject: [PATCH] fix(store-inmemory): deliver events appended before a new subscription runs The in-memory store's subscribe and subscribeAll built their streams on Stream.fromPubSub, which only subscribes to the PubSub once the stream runs. An event appended after subscribe returned but before the stream started was published to nobody. Per-stream subscribe also took its history snapshot at subscribe time, so an event appended in that gap was in neither the snapshot nor the live feed. A live stream now takes its snapshot and subscribes to the PubSub when it runs, both under the lock that append holds while it publishes. The snapshot starts from the position recorded when subscribe was called, so nothing is lost or delivered twice, and a stream that is never run holds no subscription. The new tests subscribe, append, and only then run the stream. The fork-based tests added to the Postgres store pass against the old in-memory code, because the forked fiber subscribes before the append publishes. --- .changeset/inmemory-subscribe-gap.md | 7 ++ .../src/lib/InMemoryStore.ts | 92 +++++++++++++------ .../src/lib/inMemory.spec.ts | 70 ++++++++++++-- 3 files changed, 135 insertions(+), 34 deletions(-) create mode 100644 .changeset/inmemory-subscribe-gap.md diff --git a/.changeset/inmemory-subscribe-gap.md b/.changeset/inmemory-subscribe-gap.md new file mode 100644 index 00000000..91436361 --- /dev/null +++ b/.changeset/inmemory-subscribe-gap.md @@ -0,0 +1,7 @@ +--- +'@codeforbreakfast/eventsourcing-store-inmemory': patch +--- + +Fix lost events on new subscriptions. An event appended after `subscribe` or `subscribeAll` returned, but before the returned stream started running, never reached the subscriber. A subscription now delivers every event appended after the call returns, whenever you start running its stream. + +`InMemoryStore.get`, `getAll` and `getAllLiveOnly` get the same fix. diff --git a/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts b/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts index d6034181..6b230309 100644 --- a/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts +++ b/packages/eventsourcing-store-inmemory/src/lib/InMemoryStore.ts @@ -1,4 +1,14 @@ -import { Chunk, Effect, HashMap, Option, PubSub, Stream, SynchronizedRef, pipe } from 'effect'; +import { + Chunk, + Effect, + HashMap, + Option, + PubSub, + Queue, + Stream, + SynchronizedRef, + pipe, +} from 'effect'; import { EventStreamId, EventStreamPosition, @@ -214,8 +224,40 @@ export interface InMemoryStore { >; } -const createLiveEventStream = (eventStream: EventStream): Stream.Stream => - pipe(eventStream.events, Stream.fromChunk, Stream.concat(Stream.fromPubSub(eventStream.pubsub))); +const snapshotThenLive = + (eventStream: EventStream, offset: number) => + (liveEvents: Queue.Dequeue): Stream.Stream => + pipe( + eventStream.events, + Chunk.drop(offset), + Stream.fromChunk, + Stream.concat(Stream.fromQueue(liveEvents)) + ); + +const subscribeFromOffset = + (offset: number) => + (eventStream: EventStream) => + pipe(eventStream.pubsub, PubSub.subscribe, Effect.map(snapshotThenLive(eventStream, offset))); + +// Append publishes while it holds this lock, so taking the snapshot and subscribing under the +// same lock leaves no gap between them. Doing both when the stream runs, not when subscribe +// returns, catches events appended in the meantime without holding a subscription for a stream +// that is never run. +const liveStreamFromOffset = ( + value: SynchronizedRef.SynchronizedRef>, + selectEventStream: (current: Value) => Effect.Effect, never, never>, + offset: number +): Stream.Stream => + Stream.unwrapScoped( + SynchronizedRef.modifyEffect(value, (current) => + pipe( + current, + selectEventStream, + Effect.flatMap(subscribeFromOffset(offset)), + Effect.map((stream) => [stream, current] as const) + ) + ) + ); const createHistoricalEventStream = ( eventStream: EventStream @@ -237,12 +279,6 @@ const findEventStreamOrDie = ( }) ); -const getEventStreamAndCreateLive = ( - eventStreamsById: HashMap.HashMap>, - streamId: EventStreamId -): Effect.Effect, never, never> => - pipe(findEventStreamOrDie(eventStreamsById, streamId), Effect.map(createLiveEventStream)); - const getEventStreamAndCreateHistorical = ( eventStreamsById: HashMap.HashMap>, streamId: EventStreamId @@ -266,9 +302,13 @@ const getLiveStreamForId = (value: SynchronizedRef.SynchronizedRef>) => pipe( value, - SynchronizedRef.updateAndGetEffect(ensureEventStream(streamId)), - Effect.flatMap(({ eventStreamsById }) => - getEventStreamAndCreateLive(eventStreamsById, streamId) + SynchronizedRef.updateEffect(ensureEventStream(streamId)), + Effect.as( + liveStreamFromOffset( + value, + ({ eventStreamsById }) => findEventStreamOrDie(eventStreamsById, streamId), + 0 + ) ) ); @@ -283,14 +323,12 @@ const getHistoricalStreamForId = ) ); +const selectAllEventsStream = ({ allEventsStream }: Value) => Effect.succeed(allEventsStream); + const getAllEventsStream = ( value: SynchronizedRef.SynchronizedRef> ): Effect.Effect, never, never>, never, never> => - pipe( - value, - SynchronizedRef.get, - Effect.map(({ allEventsStream }) => createLiveEventStream(allEventsStream)) - ); + Effect.succeed(liveStreamFromOffset(value, selectAllEventsStream, 0)); const getAllEventsLiveOnlyStream = ( value: SynchronizedRef.SynchronizedRef> @@ -298,7 +336,9 @@ const getAllEventsLiveOnlyStream = ( pipe( value, SynchronizedRef.get, - Effect.map(({ allEventsStream }) => Stream.fromPubSub(allEventsStream.pubsub)) + Effect.map(({ allEventsStream }) => + liveStreamFromOffset(value, selectAllEventsStream, allEventsStream.events.length) + ) ); const appendForStore = @@ -326,13 +366,11 @@ export const make = () => allEventsStream, }) ), - Effect.map( - (value: SynchronizedRef.SynchronizedRef>): InMemoryStore => ({ - append: appendForStore(value), - get: getForStore(value), - getHistorical: getHistoricalForStore(value), - getAll: () => getAllEventsStream(value), - getAllLiveOnly: () => getAllEventsLiveOnlyStream(value), - }) - ) + Effect.map((value: SynchronizedRef.SynchronizedRef>): InMemoryStore => ({ + append: appendForStore(value), + get: getForStore(value), + getHistorical: getHistoricalForStore(value), + getAll: () => getAllEventsStream(value), + getAllLiveOnly: () => getAllEventsLiveOnlyStream(value), + })) ); diff --git a/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts b/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts index 0d5fbc24..513993aa 100644 --- a/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts +++ b/packages/eventsourcing-store-inmemory/src/lib/inMemory.spec.ts @@ -1,10 +1,12 @@ -import { Effect, Layer, Schema, pipe } from 'effect'; +import { Chunk, Effect, Layer, Schema, Stream, pipe } from 'effect'; +import { describe, expect, it } from 'bun:test'; import { silentLogger } from '@codeforbreakfast/bun-test-effect'; import { runEventStoreTestSuite, FooEventStore, encodedEventStore, EventStore, + EventStreamId, } from '@codeforbreakfast/eventsourcing-store'; import { subscribeAllContract } from '@codeforbreakfast/eventsourcing-testing-contracts'; import { makeInMemoryEventStore } from './index'; @@ -35,10 +37,64 @@ class StringEventStore extends Effect.Tag('StringEventStore')< EventStore >() {} -subscribeAllContract( - 'InMemoryEventStore', - Layer.provide( - Layer.effect(StringEventStore, Effect.flatMap(make(), makeInMemoryEventStore)), - silentLogger - ) +const StringEventStoreLayer = Layer.provide( + Layer.effect(StringEventStore, Effect.flatMap(make(), makeInMemoryEventStore)), + silentLogger ); + +subscribeAllContract('InMemoryEventStore', StringEventStoreLayer); + +const decodeStreamId = Schema.decodeSync(EventStreamId); +const streamId = decodeStreamId('immediate-stream'); + +const appendEvent = (store: EventStore) => + pipe( + ['event-after-subscribe'], + Stream.fromIterable, + Stream.run(store.append({ streamId, eventNumber: 0 })) + ); + +const collectFirst = (stream: Stream.Stream) => + pipe(stream, Stream.take(1), Stream.runCollect); + +const appendThenCollectFirst = + (store: EventStore) => + (stream: Stream.Stream) => + pipe(store, appendEvent, Effect.andThen(collectFirst(stream)), Effect.timeout('1 second')); + +const runPerStreamTest = (store: EventStore) => + pipe( + { streamId, eventNumber: 0 }, + store.subscribe, + Effect.flatMap(appendThenCollectFirst(store)), + Effect.map((events) => { + expect(Array.from(events)).toEqual(['event-after-subscribe']); + }) + ); + +const runAllEventsTest = (store: EventStore) => + pipe( + store.subscribeAll(), + Effect.flatMap(appendThenCollectFirst(store)), + Effect.map((events) => { + expect(Array.from(Chunk.map(events, (e) => e.event))).toEqual(['event-after-subscribe']); + }) + ); + +describe('Subscriptions are live as soon as they return', () => { + it('should deliver an event appended after subscribe returns but before the stream runs', () => + pipe( + StringEventStore, + Effect.flatMap(runPerStreamTest), + Effect.provide(StringEventStoreLayer), + Effect.runPromise + )); + + it('should deliver an event appended after subscribeAll returns but before the stream runs', () => + pipe( + StringEventStore, + Effect.flatMap(runAllEventsTest), + Effect.provide(StringEventStoreLayer), + Effect.runPromise + )); +});