So GES returns a checkpoint at the end of the stream, I'll add a test and implementation for the other providers
nCrunch is having issues building the CES tests project, runs ok in cmdline test runner
ok this is weird
GES client -> when I read to the end of the stream it returns a checkpoint and IsEndOfStream = true
when I read again from the given checkpoint, I get no events, IsEndOfStream = false and I get a new Checkpoint
hmmm HMMM
does that smell like a bug?
Not sure
Checkpoint will change(increase) for other reasons
that is global across all streams
yeah this is the ReadAllForward scenario
but IsEndOfStream = false when reading past the end of a stream...
but even if you get IsEndOfStream = false, when you iterate the result page that has zero items, what is the real problem
or rather, is there a real problem
maybe we could fudge the IsEndOfStream
just thinking out loud
sure
the problem is that when i get to the end of the stream, I still need a Checkpoint to persist
is the checkpoint null / empty?
it is for the sql providers
one sec, firing up VS
i've implemented it in the postgres one (max(ordinal) + 1)
GES does return a checkpoint
but when you read from that checkpoint
i would expect {no events, isEndOfStream = true, a checkpoint}
What I have been doing to assert compatibility is to write a test to see what happens
i.e. not really on docs
(possibly the same checkpoint, i don't care)
yeah i've added a test
so if the GES behaviour seems weird, should prob ping the GES docs/peoples
(I'm interested too btw).
yeah that's what I was thinking, just cloning EventStore now
The good news is that we've managed to move all our events into Cedar, and project them all using a CatchUpProjection, but now we're at the end of the stream and our checkpoint is null :)
that's no good !
danbarua/Cedar.EventStore@9c3df31
so close, yet so far!
Just opening CES now, I think I have a test to cover this.
it tests reading up to the end of the stream
as does GES client tests
it doesn't test trying to read beyond the end of the stream
Ok, will add that that does a second read.
Give me about hr, got a meeting
no prob
urgh, can't seem to build or test GES on my VM
TestFixtureSetUp failed in TestsInitFixture
Exception doesn't have a stacktrace
Exception doesn't have a stacktrace
I can imagine that's a bit of a beast to build...
yeah
i grabbed js1.dll from the release binaries :trollface:
btw I have another branch
append-other-scenarios that has some failing tests for other edge cases. Going to fix that up shortly.
[Test, Category("LongRunning")]
public void be_return_empty_slice_with_position_if_asked_to_read_from_the_last_position()
{
var all = new List<RecordedEvent>();
var position = Position.Start;
AllEventsSlice slice;
while (!(slice = _conn.ReadAllEventsForwardAsync(position, 1, false).Result).IsEndOfStream)
{
all.Add(slice.Events.Single().Event);
position = slice.NextPosition;
}
Assert.That(EventDataComparer.Equal(_testEvents, all.Skip(all.Count - _testEvents.Length).ToArray()));
slice = _conn.ReadAllEventsForwardAsync(position, 1, false).Result;
Assert.That(slice.IsEndOfStream);
Assert.That(slice.Events.Length == 0);
Assert.That(slice.NextPosition == slice.FromPosition);
}
yeah I merged in all the bits from
append-other-scenarios
i'm not actually sure what the behaviour should be ^ the above test is speculation
hmm
on a related note, how are you finding the test structure in the CES solution? I need to come up with a way of reusable tests across all implementations; best I could come up with.
it works well enough :trollface: tbh i haven't given it much thought as I'm focussing on getting-shit-done right now :)
also I'm kind of familiar with it as it's similar to NEventStore so nothing i haven't seen before
so the "bug" , are you doing ReadAll or ReadStream?
ReadAll
try
{
var checkpointValue = (await dbConnection.QueryAsync<string>("SELECT checkpoint FROM projections.checkpoint;")).FirstOrDefault();
var checkpoint = checkpointValue != null ? new Checkpoint(checkpointValue) : Checkpoint.Start;
var events = await eventStore.ReadAll(checkpoint, 512);
foreach (var streamEvent in events.StreamEvents)
{
//dispatch events
var eventType = Type.GetType(streamEvent.Type, true, true);
var body = JsonSerializer.DeserializeFromString(streamEvent.JsonData, eventType) as IDomainEvent;
var headers = JsonSerializer.DeserializeFromString<IDictionary>(streamEvent.JsonMetadata);
var envelope = EventEnvelope.Create(body);
envelope.CopyHeaders(headers);
messageBus.Publish(envelope);
}
var nextCheckpoint = events.IsEnd ? Checkpoint.End : events.NextCheckpoint;
await dbConnection.ExecuteAsync("UPDATE projections.checkpoint SET checkpoint = @checkpoint;", new { checkpoint = nextCheckpoint.ToString() });
tx.Commit();
}var nextCheckpoint = events.IsEnd ? Checkpoint.End : events.NextCheckpoint; <-- this is completely wrongbasically read all, persist checkpoint, go to sleep 500ms, read all from persisted checkpoint, etc
so if i get to the end of the stream, i should get a checkpoint back that will give me any events after that checkpoint
yeah does that look odd to you?
Not getting IsEnd = true, nor am I getting a null checkpoint. Hmm.
yeah
definitely not the crack i was smoking last night then
lol
going to put a loop in, see if it ever ends.
oh interesting
[Fact]
public async Task When_read_past_end_of_all()
{
using(var fixture = GetFixture())
{
using(var eventStore = await fixture.GetEventStore())
{
await eventStore.AppendToStream("stream-1", ExpectedVersion.NoStream, CreateNewStreamEvents(1, 2, 3));
bool isEnd = false;
int count = 0;
Checkpoint checkpoint = Checkpoint.Start;
while (!isEnd)
{
_testOutputHelper.WriteLine($"Loop {count}");
var streamEventsPage = await eventStore.ReadAll(checkpoint, 10);
_testOutputHelper.WriteLine($"FromCheckpoint = {streamEventsPage.FromCheckpoint}");
_testOutputHelper.WriteLine($"NextCheckpoint = {streamEventsPage.NextCheckpoint}");
_testOutputHelper.WriteLine($"IsEnd = {streamEventsPage.IsEnd}");
_testOutputHelper.WriteLine($"StreamEvents.Count = {streamEventsPage.StreamEvents.Count}");
_testOutputHelper.WriteLine("");
checkpoint = streamEventsPage.NextCheckpoint;
isEnd = streamEventsPage.IsEnd;
count++;
if(count > 100)
{
throw new Exception("omg wtf");
}
}
}
}
}
results in
Loop 0
FromCheckpoint = 0/0
NextCheckpoint = 1624/1624
IsEnd = False
StreamEvents.Count = 3
Loop 1
FromCheckpoint = 1624/1624
NextCheckpoint = 2362/2362
IsEnd = False
StreamEvents.Count = 0
Loop 2
FromCheckpoint = 2362/2362
NextCheckpoint = 2362/2362
IsEnd = True
StreamEvents.Count = 0
ok, tried to format that
but it does end
you can edit chats
I tried and failed
ok cool so we get the gist of it
GES might hand you some empty chunks but eventually you get to IsEndOfStream = true and NextCheckpoint == FromCheckpoint
your test passes for GES and Postgres
Am going to clean up this test and commit it
so i basically had to add
while (!allEventsPage.IsEnd)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
}
actually it's mostly fine as is
to my test, because CES may hand you an empty page from the end of the stream..
yes, Loops 0 had a page count 3, but loops 1 & 2 has a count of zero before loop 3 being IsEnd = true
so yeah end up a couple extra reads but no extra events.
but.. is there a race whereby, as a reader, you'll never reach the end?
The subscription API will be better for this anyway.
for streams that never end? sure
right now i'm just building a .exe to build out the read model
then poll for new events
in your code above, just ignore the IsEnd thing?
try
{
var checkpointValue = (await dbConnection.QueryAsync<string>("SELECT checkpoint FROM projections.checkpoint;")).FirstOrDefault();
var checkpoint = checkpointValue != null ? new Checkpoint(checkpointValue) : Checkpoint.Start;
var events = await eventStore.ReadAll(checkpoint, 512);
foreach (var streamEvent in events.StreamEvents)
{
//dispatch events
var eventType = Type.GetType(streamEvent.Type, true, true);
var body = JsonSerializer.DeserializeFromString(streamEvent.JsonData, eventType) as IDomainEvent;
var headers = JsonSerializer.DeserializeFromString<IDictionary>(streamEvent.JsonMetadata);
var envelope = EventEnvelope.Create(body);
envelope.CopyHeaders(headers);
messageBus.Publish(envelope);
}
await dbConnection.ExecuteAsync("UPDATE projections.checkpoint SET checkpoint = @checkpoint;", new { checkpoint = events.NextCheckpoint });
tx.Commit();
}
I need that to make the test pass for GES
yes that's it
i was testing for isEnd to figure out what the checkpoint would be
that was a red herring
the original code looked like that
yep
shall i add in the end-of-stream checkpoint for MSSQL/Sqlite or just focus on postgres in this branch and deal with that later?
Just focus on postgres pls
sure, keeps the merge clean
Yep
This test is failing for MsSql, going to fix it now.
here's the test for full functionality:
[Fact]
public async Task Read_forwards_to_the_end_should_return_a_valid_Checkpoint()
{
using (var fixture = GetFixture())
{
using (var eventStore = await fixture.GetEventStore())
{
await eventStore.AppendToStream("stream-1", ExpectedVersion.NoStream, CreateNewStreamEvents(1, 2, 3, 4, 5, 6));
// read to the end of the stream
var allEventsPage = await eventStore.ReadAll(Checkpoint.Start, 4);
while (!allEventsPage.IsEnd)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
}
allEventsPage.IsEnd.Should().BeTrue();
Checkpoint currentCheckpoint = allEventsPage.NextCheckpoint;
currentCheckpoint.Should().NotBeNull();
// read end of stream again, should be empty, should return same checkpoint
allEventsPage = await eventStore.ReadAll(currentCheckpoint, 10);
while (!allEventsPage.IsEnd)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
}
allEventsPage.StreamEvents.Should().BeEmpty();
allEventsPage.IsEnd.Should().BeTrue();
allEventsPage.NextCheckpoint.Should().NotBeNull();
currentCheckpoint = allEventsPage.NextCheckpoint;
// append some events then read again from the saved checkpoint, the next checkpoint should have moved
await eventStore.AppendToStream("stream-1", ExpectedVersion.Any, CreateNewStreamEvents(7, 8, 9));
allEventsPage = await eventStore.ReadAll(currentCheckpoint, 10);
while (!allEventsPage.IsEnd)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
}
allEventsPage.IsEnd.Should().BeTrue();
allEventsPage.NextCheckpoint.Should().NotBeNull();
allEventsPage.NextCheckpoint.Should().NotBe(currentCheckpoint.Value);
}
}
}while (!allEventsPage.IsEnd)
that'll go into infinite loop in a bad impl?
yep
right, I'll co-opt this, but will add a short circuit for that while loop
quick edit
tbh the while loop is copied from another test
ah wait you have a counter in there
yep
i see what that's for now thought that was for an assertion
i took it out to focus the assertions on the case in point
one min, will edit it and share
already done mine..
[Fact]
public async Task Read_forwards_to_the_end_should_return_a_valid_Checkpoint()
{
using (var fixture = GetFixture())
{
using (var eventStore = await fixture.GetEventStore())
{
await eventStore.AppendToStream("stream-1", ExpectedVersion.NoStream, CreateNewStreamEvents(1, 2, 3, 4, 5, 6));
// read to the end of the stream
var allEventsPage = await eventStore.ReadAll(Checkpoint.Start, 4);
int count = 0; //counter is used to short circuit bad implementations that never return IsEnd = true
while (!allEventsPage.IsEnd && count < 20)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
count++;
}
allEventsPage.IsEnd.Should().BeTrue();
Checkpoint currentCheckpoint = allEventsPage.NextCheckpoint;
currentCheckpoint.Should().NotBeNull();
// read end of stream again, should be empty, should return same checkpoint
allEventsPage = await eventStore.ReadAll(currentCheckpoint, 10);
count = 0;
while (!allEventsPage.IsEnd && count < 20)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
count++;
}
allEventsPage.StreamEvents.Should().BeEmpty();
allEventsPage.IsEnd.Should().BeTrue();
allEventsPage.NextCheckpoint.Should().NotBeNull();
currentCheckpoint = allEventsPage.NextCheckpoint;
// append some events then read again from the saved checkpoint, the next checkpoint should have moved
await eventStore.AppendToStream("stream-1", ExpectedVersion.Any, CreateNewStreamEvents(7, 8, 9));
allEventsPage = await eventStore.ReadAll(currentCheckpoint, 10);
count = 0;
while (!allEventsPage.IsEnd && count < 20)
{
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, 10);
count++;
}
allEventsPage.IsEnd.Should().BeTrue();
allEventsPage.NextCheckpoint.Should().NotBeNull();
allEventsPage.NextCheckpoint.Should().NotBe(currentCheckpoint.Value);
}
}
}
I'm adding this test to Cedar.EventsStore.Tests
I also want to console out some info.
_tesOutputHelper
prob split this into 2 or 3 tests, it's a bit on the large side.
defo
replay 1.3million events , make a cup of coffee
namespace Cedar.EventStore
{
using System.Threading.Tasks;
using FluentAssertions;
using Xunit;
public abstract partial class EventStoreAcceptanceTests
{
[Fact]
public async Task Can_read_to_end_of_allstream()
{
using(var fixture = GetFixture())
{
using(var eventStore = await fixture.GetEventStore())
{
await eventStore.AppendToStream("stream-1", ExpectedVersion.NoStream, CreateNewStreamEvents(1, 2, 3));
var lastAllEventPage = await ReadAllStreamToEnd(eventStore);
lastAllEventPage.StreamEvents.Should().BeEmpty();
lastAllEventPage.IsEnd.Should().BeTrue();
lastAllEventPage.NextCheckpoint.Should().Be(lastAllEventPage.FromCheckpoint);
}
}
}
[Fact]
public async Task When_read_to_end_of_allstream_then_append_event_then_can_read_to_end_of_allstream()
{
using (var fixture = GetFixture())
{
using (var eventStore = await fixture.GetEventStore())
{
await eventStore.AppendToStream("stream-1", ExpectedVersion.NoStream, CreateNewStreamEvents(1, 2, 3));
await ReadAllStreamToEnd(eventStore);
await eventStore.AppendToStream("stream-1", 2, CreateNewStreamEvents(1, 2, 3));
var lastAllEventPage = await ReadAllStreamToEnd(eventStore);
lastAllEventPage.StreamEvents.Should().BeEmpty();
lastAllEventPage.IsEnd.Should().BeTrue();
lastAllEventPage.NextCheckpoint.Should().Be(lastAllEventPage.NextCheckpoint);
}
}
}
private async Task<AllEventsPage> ReadAllStreamToEnd(IEventStore eventStore)
{
int pageSize = 4, count = 0;
var allEventsPage = await eventStore.ReadAll(Checkpoint.Start, pageSize);
LogAllEventsPage(allEventsPage);
while (!allEventsPage.IsEnd && count < 10)
{
_testOutputHelper.WriteLine($"Loop {count}");
allEventsPage = await eventStore.ReadAll(allEventsPage.NextCheckpoint, pageSize);
LogAllEventsPage(allEventsPage);
count++;
}
return allEventsPage;
}
private void LogAllEventsPage(AllEventsPage allEventsPage)
{
_testOutputHelper.WriteLine($"FromCheckpoint = {allEventsPage.FromCheckpoint}");
_testOutputHelper.WriteLine($"NextCheckpoint = {allEventsPage.NextCheckpoint}");
_testOutputHelper.WriteLine($"IsEnd = {allEventsPage.IsEnd}");
_testOutputHelper.WriteLine($"StreamEvents.Count = {allEventsPage.StreamEvents.Count}");
_testOutputHelper.WriteLine("");
}
}}
sweet, that's much tidier :+1:
committing
cool i'll merge it in
commiting to master
will merge it with append-other-scenarios
actually, I think I'll merge that branch in also
MSSql tests failing on master anyway.
Then I'll restart fixing them
ok on master
One failing GES test, will have a look aprés lunch
i changed
"data", "metadata" to "\"data\"", "\"metadata\"" since postgres checks for valid json on input
i've got When_read_to_end_of_allstream_then_append_then_can_read_to_end_of_allstream StreamEvents.Should().Be.Empty() failing
the failing test is because GES is treating the second append idempotently oop
actually, the idempotent area is one thing I don't know how I'm going to handle.
Added some idempotency tests. Passing on GES, failing on MSSql oc
oooh
does GES compare the binary payload?
don't think so, just the event ID
ah i see
i have cheated here
i used the NES commit id
so i could trace fuckups back to the source
i guess you could put a unique index on it and swallow the exception
basically if you give it an expected version and the event ids match whatever event Id are in that place, then all is ok
What's the use case for this?
i.e.
A) Append ExpectedVersion= 0, A,B,C,D,E,F
then
B) Append ExpectedVersion= 0, A,B,C,D,E,F
A) Append ExpectedVersion= 0, A,B,C,D,E,F
then
B) Append ExpectedVersion= 0, A,B,C,D,E,F
it's the same op
so no exceptions
ok A-F is now A-FFFFFFFFFFFFFF
i guess we're following GES so just go with whatever they do eh?
also these will not throw either
C) Append ExpectedVersion = 1; B,C,D,E,F
D) Append ExpectedVersion= 3; D,E
C) Append ExpectedVersion = 1; B,C,D,E,F
D) Append ExpectedVersion= 3; D,E
Sure, I understand that, I'm just wondering what kind of situations this is useful in?
Idempotency is a good property to have?
the fact that GES is not transactional
and is usable over HTTP I guess
ah right
or rather, doesn't support n-phase commit.
nor DTC
I'm imagining an HTTP operation that succeeds server side but client doesn't get response
Client thenretries
yep, now it makes sense
aka HTTP PUT
yes, totally makes sense now
replicating it on sql stores will be fun - :)
am open to better names on those test btw
so you could knock up some OWIN bits to build yourself an EventStore knockoff on a RDBMS backend
note could not should heh
heh
I might actually have to for our own reasons.
MSSql is borked.
I'll have to come back to this later. Lunch time!
np
oc, you could get postgres working on these test (everything on master). Would :heart: a review on them too.
yeah i have it all working
apart from one
streamEvents.Empty on end-of stream
also nCrunch stubbornly refusing to load GetEventStore.Tests
updating EventStore.Client in GES.Tests to 3.0.2 (inline with Cedar.ES.GES) seemed to sort it out
error message?
ah
it was the custom after build task
js1.dll not found
actually i suspect i might have commented that out to get it working too haha
(the post-build thingy)
Fffuuuu
that shit keeps happening to 2 projects and I don't know whay
why
UCS-2 Little Endian
but on only every alternate line?
GesEventStoreFixture.cs (18, 51): The type 'IEventStoreConnection' is defined in an assembly that is not referenced. You must add a reference to assembly 'EventStore.ClientAPI, Version=3.0.2.0, Culture=neutral, PublicKeyToken=null'.
cool
update-package eventstore.client
yeah see that...
hm nuget in vs2015 horribly borken
there's a surprise
fixed on master
right, lunch.