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 + )); +});