Skip to content

Commit b718f63

Browse files
NickSeagullNickSeagull
authored
fix: Event Store Tests for Large Event Counts (>100 events) (#224)
* . * .
1 parent 1c5b4f5 commit b718f63

10 files changed

Lines changed: 164 additions & 75 deletions

File tree

‎core/core/Array.hs‎

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ module Array (
4949
partitionBy,
5050
splitFirst,
5151
any,
52+
chunksOf,
5253

5354
-- * Compatibility
5455
unwrap,
@@ -658,4 +659,33 @@ getJusts self = do
658659
|> Maybe.map wrap
659660
|> Maybe.withDefault empty
660661
self
661-
|> flatMap maybeItemToArray
662+
|> flatMap maybeItemToArray
663+
664+
665+
-- | Split an array into chunks of a given size.
666+
-- The last chunk may contain fewer elements if the array length is not evenly divisible.
667+
--
668+
-- >>> chunksOf 2 (fromLinkedList [1,2,3,4,5] :: Array Int)
669+
-- Array [Array [1,2],Array [3,4],Array [5]]
670+
-- >>> chunksOf 3 (fromLinkedList [1,2,3,4,5,6] :: Array Int)
671+
-- Array [Array [1,2,3],Array [4,5,6]]
672+
-- >>> chunksOf 1 (fromLinkedList [1,2,3] :: Array Int)
673+
-- Array [Array [1],Array [2],Array [3]]
674+
-- >>> chunksOf 5 (fromLinkedList [1,2] :: Array Int)
675+
-- Array [Array [1,2]]
676+
-- >>> chunksOf 2 (fromLinkedList [] :: Array Int)
677+
-- Array []
678+
chunksOf :: Int -> Array a -> Array (Array a)
679+
chunksOf n (Array vector) = do
680+
let chunks =
681+
Data.Vector.toList vector
682+
|> chunkList n
683+
|> LinkedList.map Data.Vector.fromList
684+
|> Data.Vector.fromList
685+
Array (Data.Vector.map Array chunks)
686+
where
687+
chunkList :: Int -> [a] -> [[a]]
688+
chunkList _ [] = []
689+
chunkList size xs = do
690+
let (chunk, rest) = Prelude.splitAt size xs
691+
chunk : chunkList size rest

‎core/service/Service/EventStore/Postgres/Internal.hs‎

Lines changed: 23 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -520,7 +520,12 @@ performReadAllStreamEvents
520520
case evt of
521521
Ok goodEvent -> do
522522
stream |> Stream.writeItem goodEvent
523-
positionRef |> Var.set record.globalPosition
523+
-- Update position for next batch
524+
-- For forward: increment by 1 (since we use >= comparison)
525+
-- For backward: decrement by 1 (since we use <= comparison for global position)
526+
case readDirection of
527+
Just Backwards -> positionRef |> Var.set (record.globalPosition - 1)
528+
_ -> positionRef |> Var.set (record.globalPosition + 1)
524529
Err err -> do
525530
let entityName = record.entityName
526531
let streamId = record.inlinedStreamId
@@ -535,7 +540,12 @@ performReadAllStreamEvents
535540
}
536541
|> ToxicAllEvent
537542
stream |> Stream.writeItem toxicEvt
538-
positionRef |> Var.set record.globalPosition
543+
-- Update position for next batch
544+
-- For forward: increment by 1 (since we use >= comparison)
545+
-- For backward: decrement by 1 (since we use <= comparison for global position)
546+
case readDirection of
547+
Just Backwards -> positionRef |> Var.set (record.globalPosition - 1)
548+
_ -> positionRef |> Var.set (record.globalPosition + 1)
539549
Var.decrement remainingLimitRef
540550

541551
Task.when (Array.length records < batchSize) do
@@ -795,7 +805,11 @@ performReadStreamEvents
795805
Ok goodEvent -> do
796806
stream |> Stream.writeItem goodEvent
797807
-- Update position for next batch
798-
positionRef |> Var.set record.localPosition
808+
-- For forward: increment by 1 (since we use >= comparison)
809+
-- For backward: keep same position (since we use < comparison which already excludes it)
810+
case readDirection of
811+
Just Backwards -> positionRef |> Var.set record.localPosition
812+
_ -> positionRef |> Var.set (record.localPosition + 1)
799813
Err err -> do
800814
let locator = [fmt|#{EntityName.toText entityName}#{StreamId.toText streamId}|]
801815
let toxicEvt =
@@ -808,7 +822,12 @@ performReadStreamEvents
808822
}
809823
|> ToxicStreamEvent
810824
stream |> Stream.writeItem toxicEvt
811-
positionRef |> Var.set record.localPosition
825+
-- Update position for next batch
826+
-- For forward: increment by 1 (since we use >= comparison)
827+
-- For backward: keep same position (since we use < comparison which already excludes it)
828+
case readDirection of
829+
Just Backwards -> positionRef |> Var.set record.localPosition
830+
_ -> positionRef |> Var.set (record.localPosition + 1)
812831
Var.decrement remainingLimitRef
813832

814833
-- Check if we got fewer records than batch size (means we're at the end)

‎core/service/Service/EventStore/Postgres/Internal/Sessions.hs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -352,7 +352,7 @@ selectStreamEventBatch positionRef (EntityName entityName) (StreamId streamIdTex
352352
SELECT EventId, GlobalPosition, LocalPosition, InlinedStreamId, Entity, EventData, Metadata
353353
FROM Events
354354
WHERE Entity = '#{entityName}' AND InlinedStreamId = '#{streamIdText}'#{positionFilter}
355-
ORDER BY GlobalPosition #{direction}
355+
ORDER BY LocalPosition #{direction}
356356
LIMIT #{batchSize}
357357
|]
358358
let encoder = Encoders.noParams

‎core/testlib/Test/Service/EventStore/IndividualStreamOrdering/Context.hs‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,7 @@ data Context = Context
1919
entityName :: Event.EntityName,
2020
streamId :: Event.StreamId,
2121
store :: EventStore MyEvent,
22-
payload :: Event.InsertionPayload MyEvent,
23-
position :: Event.StreamPosition,
22+
allInsertions :: Array (Event.Insertion MyEvent),
2423
positions :: Array Event.StreamPosition
2524
}
2625

@@ -30,16 +29,17 @@ initialize newStore eventCountNumber = do
3029
store <- newStore
3130
streamId <- StreamId.new
3231
entityName <- Uuid.generate |> Task.map (toText .> Event.EntityName)
33-
insertions <-
32+
allInsertions <-
3433
Array.fromLinkedList [0 .. eventCountNumber - 1]
3534
|> Task.mapArray
3635
newInsertion
37-
let payload = Event.InsertionPayload {streamId, entityName, insertionType = Event.AnyStreamState, insertions}
3836

39-
insertResult <- payload |> store.insert |> Task.mapError toText
40-
let position = insertResult.localPosition
41-
let positions = Array.fromLinkedList [0 .. eventCountNumber - 1] |> Array.map (fromIntegral .> Event.StreamPosition)
37+
-- Insert events in chunks of 100 (batch size limit)
38+
allInsertions |> Array.chunksOf 100 |> Task.forEach \chunk -> do
39+
let payload = Event.InsertionPayload {streamId, entityName, insertionType = Event.AnyStreamState, insertions = chunk}
40+
payload |> store.insert |> Task.mapError toText |> discard
4241

42+
let positions = Array.fromLinkedList [0 .. eventCountNumber - 1] |> Array.map (fromIntegral .> Event.StreamPosition)
4343
let eventCount = fromIntegral eventCountNumber
4444

45-
return Context {eventCount, streamId, store, payload, position, positions, entityName}
45+
return Context {eventCount, streamId, store, allInsertions, positions, entityName}

‎core/testlib/Test/Service/EventStore/IndividualStreamOrdering/Spec.hs‎

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@ spec newStore = do
2323
whenEnvVar "TEST_EVENT_COUNT" do
2424
specWithCount newStore 100
2525
specWithCount newStore 1000
26+
27+
whenEnvVar "TEST_EVENT_COUNT_HIGH" do
2628
specWithCount newStore 10000
2729
specWithCount newStore 100000
2830
specWithCount newStore 1000000
@@ -74,19 +76,15 @@ specWithCount newStore eventCount = do
7476
-- since global positions are assigned dynamically based on append order
7577
Array.length events |> fromIntegral |> shouldBe context.eventCount
7678

77-
let payloadWithInsertion =
78-
context.payload.insertions
79-
|> Array.map (\i -> (context.payload, i))
80-
81-
-- Pair up actual events with expected events for validation
82-
let eventPairs = payloadWithInsertion |> Array.zip events
79+
-- Pair up actual events with expected insertions for validation
80+
let eventPairs = context.allInsertions |> Array.zip events
8381

8482
-- Each event should have the correct core properties
85-
eventPairs |> Task.forEach \((payload, insertion), event) -> do
83+
eventPairs |> Task.forEach \(insertion, event) -> do
8684
-- Validate event ID, entity ID, stream ID, and local position
8785
event.metadata.eventId |> shouldBe insertion.id
88-
event.entityName |> shouldBe payload.entityName
89-
event.streamId |> shouldBe payload.streamId
86+
event.entityName |> shouldBe context.entityName
87+
event.streamId |> shouldBe context.streamId
9088
(event.metadata.localPosition)
9189
|> shouldBe insertion.metadata.localPosition
9290

‎core/testlib/Test/Service/EventStore/ReadAllBackwardsFromEnd/Context.hs‎

Lines changed: 49 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -35,38 +35,62 @@ initialize newStore eventCount = do
3535
entity2IdText <- Uuid.generate |> Task.map toText
3636
let entity2Id = Event.EntityName entity2IdText
3737

38-
let generateEvents :: Event.EntityName -> Task Text (Array (Event.InsertionPayload MyEvent))
39-
generateEvents entityName = do
40-
Array.fromLinkedList [0 .. eventCount - 1] |> Task.mapArray \index -> do
41-
insertions <- Array.fromLinkedList [index] |> Task.mapArray newInsertion
42-
Event.InsertionPayload
43-
{ streamId,
44-
entityName,
45-
insertionType = Event.AnyStreamState,
46-
insertions
47-
}
48-
|> Task.yield
38+
-- Generate all insertions for both entities
39+
entity1Insertions <- Array.fromLinkedList [0 .. eventCount - 1] |> Task.mapArray newInsertion
40+
entity2Insertions <- Array.fromLinkedList [0 .. eventCount - 1] |> Task.mapArray newInsertion
4941

50-
entity1Events <- generateEvents entity1Id
51-
entity2Events <- generateEvents entity2Id
42+
-- Insert entity1 events in chunks of 100 (batch size limit) and track results
43+
entity1Results <-
44+
entity1Insertions
45+
|> Array.chunksOf 100
46+
|> Task.mapArray
47+
( \chunk -> do
48+
let payload = Event.InsertionPayload {streamId, entityName = entity1Id, insertionType = Event.AnyStreamState, insertions = chunk}
49+
payload |> store.insert |> Task.mapError toText
50+
)
5251

53-
entity1Inserted <-
54-
entity1Events
55-
|> Task.mapArray (\payload -> payload |> store.insert)
56-
|> Task.mapError toText
57-
58-
entity2Inserted <-
59-
entity2Events
60-
|> Task.mapArray (\payload -> payload |> store.insert)
61-
|> Task.mapError toText
52+
-- Insert entity2 events in chunks of 100 (batch size limit) and track results
53+
entity2Results <-
54+
entity2Insertions
55+
|> Array.chunksOf 100
56+
|> Task.mapArray
57+
( \chunk -> do
58+
let payload = Event.InsertionPayload {streamId, entityName = entity2Id, insertionType = Event.AnyStreamState, insertions = chunk}
59+
payload |> store.insert |> Task.mapError toText
60+
)
6261

6362
let maxGlobalPosition =
64-
entity1Inserted
65-
|> Array.append entity2Inserted
66-
|> Array.map (\event -> event.globalPosition)
63+
entity1Results
64+
|> Array.append entity2Results
65+
|> Array.map (\result -> result.globalPosition)
6766
|> Array.maximum
6867
|> Maybe.withDefault (Event.StreamPosition 0)
6968

69+
-- Create payload wrappers for compatibility (not used in tests, but kept for API consistency)
70+
let entity1Events =
71+
entity1Insertions
72+
|> Array.map
73+
( \insertion ->
74+
Event.InsertionPayload
75+
{ streamId,
76+
entityName = entity1Id,
77+
insertionType = Event.AnyStreamState,
78+
insertions = Array.wrap insertion
79+
}
80+
)
81+
82+
let entity2Events =
83+
entity2Insertions
84+
|> Array.map
85+
( \insertion ->
86+
Event.InsertionPayload
87+
{ streamId,
88+
entityName = entity2Id,
89+
insertionType = Event.AnyStreamState,
90+
insertions = Array.wrap insertion
91+
}
92+
)
93+
7094
return
7195
Context
7296
{ eventCount,

‎core/testlib/Test/Service/EventStore/ReadAllBackwardsFromEnd/Spec.hs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ spec newStore = do
2525
whenEnvVar "TEST_EVENT_COUNT" do
2626
specWithCount newStore 100
2727
specWithCount newStore 1000
28+
29+
whenEnvVar "TEST_EVENT_COUNT_HIGH" do
2830
specWithCount newStore 10000
2931
specWithCount newStore 100000
3032
specWithCount newStore 1000000

‎core/testlib/Test/Service/EventStore/ReadAllForwardsFromStart/Context.hs‎

Lines changed: 40 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -34,38 +34,52 @@ initialize newStore eventCount = do
3434
entity2IdText <- Uuid.generate |> Task.map toText
3535
let entity2Id = Event.EntityName entity2IdText
3636

37-
let generateEvents :: Event.EntityName -> Task Text (Array (Event.InsertionPayload MyEvent))
38-
generateEvents entityName = do
39-
Array.fromLinkedList [0 .. eventCount - 1] |> Task.mapArray \index -> do
40-
insertions <- Array.fromLinkedList [index] |> Task.mapArray newInsertion
41-
Event.InsertionPayload
42-
{ streamId,
43-
entityName,
44-
insertionType = Event.AnyStreamState,
45-
insertions
46-
}
47-
|> Task.yield
37+
-- Generate all insertions for both entities
38+
entity1Insertions <- Array.fromLinkedList [0 .. eventCount - 1] |> Task.mapArray newInsertion
39+
entity2Insertions <- Array.fromLinkedList [0 .. eventCount - 1] |> Task.mapArray newInsertion
4840

49-
entity1Events <- generateEvents entity1Id
50-
entity2Events <- generateEvents entity2Id
41+
-- Insert entity1 events in chunks of 100 (batch size limit)
42+
entity1Insertions |> Array.chunksOf 100 |> Task.forEach \chunk -> do
43+
let payload = Event.InsertionPayload {streamId, entityName = entity1Id, insertionType = Event.AnyStreamState, insertions = chunk}
44+
payload |> store.insert |> Task.mapError toText |> discard
5145

52-
entity1Inserted <-
53-
entity1Events
54-
|> Task.mapArray (\payload -> payload |> store.insert)
55-
|> Task.mapError toText
46+
-- Insert entity2 events in chunks of 100 (batch size limit)
47+
entity2Insertions |> Array.chunksOf 100 |> Task.forEach \chunk -> do
48+
let payload = Event.InsertionPayload {streamId, entityName = entity2Id, insertionType = Event.AnyStreamState, insertions = chunk}
49+
payload |> store.insert |> Task.mapError toText |> discard
5650

57-
let entity1Positions =
58-
entity1Inserted
59-
|> Array.map (\result -> result.localPosition)
51+
-- Create payload wrappers for compatibility (not used in tests, but kept for API consistency)
52+
let entity1Events =
53+
entity1Insertions
54+
|> Array.map
55+
( \insertion ->
56+
Event.InsertionPayload
57+
{ streamId,
58+
entityName = entity1Id,
59+
insertionType = Event.AnyStreamState,
60+
insertions = Array.wrap insertion
61+
}
62+
)
63+
64+
let entity2Events =
65+
entity2Insertions
66+
|> Array.map
67+
( \insertion ->
68+
Event.InsertionPayload
69+
{ streamId,
70+
entityName = entity2Id,
71+
insertionType = Event.AnyStreamState,
72+
insertions = Array.wrap insertion
73+
}
74+
)
6075

61-
entity2Inserted <-
62-
entity2Events
63-
|> Task.mapArray (\payload -> payload |> store.insert)
64-
|> Task.mapError toText
76+
let entity1Positions =
77+
Array.fromLinkedList [0 .. eventCount - 1]
78+
|> Array.map (fromIntegral .> Event.StreamPosition)
6579

6680
let entity2Positions =
67-
entity2Inserted
68-
|> Array.map (\result -> result.localPosition)
81+
Array.fromLinkedList [0 .. eventCount - 1]
82+
|> Array.map (fromIntegral .> Event.StreamPosition)
6983

7084
let allEvents = entity1Events |> Array.append entity2Events
7185
let allPositions = entity1Positions |> Array.append entity2Positions

‎core/testlib/Test/Service/EventStore/ReadAllForwardsFromStart/Spec.hs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ spec newStore = do
2525
whenEnvVar "TEST_EVENT_COUNT" do
2626
specWithCount newStore 100
2727
specWithCount newStore 1000
28+
29+
whenEnvVar "TEST_EVENT_COUNT_HIGH" do
2830
specWithCount newStore 10000
2931
specWithCount newStore 100000
3032
specWithCount newStore 1000000

‎core/testlib/Test/Service/EventStore/Subscriptions/Spec.hs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,7 @@ spec newStore = do
5050
|> discard
5151

5252
-- Wait briefly for async notifications to complete
53-
AsyncTask.sleep 10 |> Task.mapError (\_ -> "timeout")
53+
AsyncTask.sleep 100 |> Task.mapError (\_ -> "timeout")
5454

5555
-- Check that we received the events
5656
received <- ConcurrentVar.get receivedEvents

0 commit comments

Comments
 (0)