diff --git a/src/CoreTests/StreamExtensionsTests.cs b/src/CoreTests/StreamExtensionsTests.cs index b6f5c5de..ab1c0da2 100644 --- a/src/CoreTests/StreamExtensionsTests.cs +++ b/src/CoreTests/StreamExtensionsTests.cs @@ -65,4 +65,87 @@ public async Task read_all_bytes_async() actual.ShouldBe(bytes); } + + [Fact] + public async Task read_expected_buffer_matches_the_exact_bytes() + { + var stream = new MemoryStream([1, 2, 3]); + + (await stream.ReadExpectedBufferAsync([1, 2, 3])).ShouldBeTrue(); + } + + [Fact] + public async Task read_expected_buffer_rejects_different_bytes() + { + var stream = new MemoryStream([1, 2, 4]); + + (await stream.ReadExpectedBufferAsync([1, 2, 3])).ShouldBeFalse(); + } + + [Fact] + public async Task read_expected_buffer_only_consumes_what_it_expects() + { + var stream = new MemoryStream([1, 2, 3]); + + (await stream.ReadExpectedBufferAsync([1, 2])).ShouldBeTrue(); + + stream.Position.ShouldBe(2); + } + + [Fact] + public async Task read_expected_buffer_rejects_a_stream_that_ends_short_of_trailing_zeros() + { + var stream = new MemoryStream([1]); + + (await stream.ReadExpectedBufferAsync([1, 0])).ShouldBeFalse(); + } + + [Fact] + public async Task read_expected_buffer_rejects_an_empty_stream_when_expecting_zeros() + { + var stream = new MemoryStream(); + + (await stream.ReadExpectedBufferAsync([0, 0])).ShouldBeFalse(); + } + + [Fact] + public async Task read_expected_buffer_keeps_reading_across_partial_reads() + { + var stream = new OneByteAtATimeStream([1, 2, 3]); + + (await stream.ReadExpectedBufferAsync([1, 2, 3])).ShouldBeTrue(); + } + + [Fact] + public async Task read_bytes_keeps_reading_across_partial_reads() + { + var stream = new OneByteAtATimeStream([1, 2, 3]); + + (await stream.ReadBytesAsync(3)).ShouldBe(new byte[] { 1, 2, 3 }); + } + + // Pinned, not endorsed. Wolverine's TCP listener reads its length prefix with this and treats all zeros + // as nothing to do, which is how it ignores a connection that closes without sending anything, like a + // TCP health probe. Throwing here would turn each of those into an error + [Fact] + public async Task read_bytes_leaves_the_rest_of_the_buffer_zeroed_on_a_short_read() + { + var stream = new MemoryStream([1]); + + (await stream.ReadBytesAsync(3)).ShouldBe(new byte[] { 1, 0, 0 }); + } + + [Fact] + public async Task read_bytes_returns_zeros_from_an_empty_stream() + { + var stream = new MemoryStream(); + + (await stream.ReadBytesAsync(4)).ShouldBe(new byte[4]); + } + + private class OneByteAtATimeStream(byte[] bytes) : MemoryStream(bytes) + { + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + => base.ReadAsync(buffer[..Math.Min(1, buffer.Length)], cancellationToken); + } } \ No newline at end of file diff --git a/src/CoreTests/StringExtensionsTester.cs b/src/CoreTests/StringExtensionsTester.cs index f9d3f49e..8fae647c 100644 --- a/src/CoreTests/StringExtensionsTester.cs +++ b/src/CoreTests/StringExtensionsTester.cs @@ -132,6 +132,55 @@ public void numbers_should_be_invalid() foreach (var number in numbers) number.IsValidNumber(new CultureInfo("en-us")).ShouldBeFalse(); } + [Theory] + [InlineData("1x234")] + [InlineData("1 234")] + [InlineData("12x345x678")] + public void the_group_separator_is_not_a_wildcard(string number) + { + // The de-DE group separator is "." + number.IsValidNumber(new CultureInfo("de-DE")).ShouldBeFalse(); + } + + [Theory] + [InlineData("")] + [InlineData("-")] + public void a_number_has_to_have_a_digit(string number) + { + number.IsValidNumber(new CultureInfo("en-us")).ShouldBeFalse(); + } + + [Theory] + [InlineData("123\n")] + [InlineData("1,000.1\n")] + public void nothing_may_follow_the_number(string number) + { + number.IsValidNumber(new CultureInfo("en-us")).ShouldBeFalse(); + } + + [Theory] + [InlineData("0")] + [InlineData("-1")] + [InlineData(".5")] + [InlineData("-.5")] + [InlineData("-1,000.5")] + public void numbers_that_stay_valid(string number) + { + number.IsValidNumber(new CultureInfo("en-us")).ShouldBeTrue(); + } + + [Fact] + public void every_culture_accepts_a_number_written_with_its_own_separators() + { + foreach (var culture in CultureInfo.GetCultures(CultureTypes.AllCultures)) + { + var format = culture.NumberFormat; + var number = $"1{format.NumberGroupSeparator}234{format.NumberDecimalSeparator}5"; + + number.IsValidNumber(culture).ShouldBeTrue(culture.Name); + } + } + [Fact] public void to_bool() { diff --git a/src/EventTests/Daemon/AggregationRunnerRaisedEventsTests.cs b/src/EventTests/Daemon/AggregationRunnerRaisedEventsTests.cs new file mode 100644 index 00000000..c4a46fc4 --- /dev/null +++ b/src/EventTests/Daemon/AggregationRunnerRaisedEventsTests.cs @@ -0,0 +1,275 @@ +using EventTests.Projections; +using JasperFx.Core; +using JasperFx.Events; +using JasperFx.Events.Daemon; +using JasperFx.Events.Grouping; +using JasperFx.Events.Projections; +using Microsoft.Extensions.Logging.Abstractions; +using NSubstitute; +using Shouldly; + +namespace EventTests.Daemon; + +// Events raised from RaiseSideEffects, followed from the projection through the runner to what the store's +// batch is told to write. Assertions read the actions after every one has been built, because a store +// executes them at commit and sees whatever the event instances say by then. +public class AggregationRunnerRaisedEventsTests +{ + private readonly RecordingBatch theBatch = new(); + + private async Task applyAsync(EventSlice slice, AggregationScope scope, + StreamIdentity streamIdentity, Action> raiseSideEffects) where TId : notnull + { + var registry = Substitute.For(); + registry.StreamIdentity.Returns(streamIdentity); + registry.TimeProvider.Returns(TimeProvider.System); + registry.EventMappingFor(Arg.Any()).Returns(Substitute.For()); + + var store = Substitute.For>(); + store.Registry.Returns(registry); + + var projection = Substitute.For>(); + projection.Scope.Returns(scope); + projection.MatchesAnyDeleteType(Arg.Any>()).Returns(false); + + var snapshot = new User("Beast", "Hank McCoy"); + projection + .DetermineActionAsync(Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any>(), Arg.Any>(), + Arg.Any()) + .Returns(new ValueTask<(User?, ActionType)>((snapshot, ActionType.Store))); + projection + .TryApplyMetadata(Arg.Any>(), Arg.Any(), Arg.Any(), + Arg.Any>()) + .Returns(((IEvent?)null, snapshot)); + projection + .RaiseSideEffects(Arg.Any(), Arg.Any(), Arg.Any>()) + .Returns(call => + { + raiseSideEffects(call.Arg>()); + return ValueTask.CompletedTask; + }); + + var storage = Substitute.For>(); + storage.TenantId.Returns("foo"); + + var runner = new AggregationRunner( + store, + Substitute.For(), + projection, + SliceBehavior.None, + Substitute.For(), + NullLogger.Instance); + + await runner.ApplyChangesAsync(ShardExecutionMode.Continuous, theBatch, new FakeOperations(), slice, + storage, Substitute.For>(), CancellationToken.None); + } + + private static Event eventOnStream(Guid streamId, long version) + => new(new AEvent()) { StreamId = streamId, Version = version }; + + private static Event eventOnStream(string streamKey, long version) + => new(new AEvent()) { StreamKey = streamKey, Version = version }; + + [Fact] + public async Task a_multi_stream_slice_writes_each_raised_event_only_to_the_stream_it_was_raised_to() + { + var left = Guid.NewGuid(); + var right = Guid.NewGuid(); + var a = new AEvent(); + var b = new BEvent(); + var c = new CEvent(); + + var slice = new EventSlice(Guid.NewGuid(), "foo", [eventOnStream(Guid.NewGuid(), 4)]); + + await applyAsync(slice, AggregationScope.MultiStream, StreamIdentity.AsGuid, s => + { + s.AppendEvent(left, a); + s.AppendEvent(right, b); + s.AppendEvent(left, c); + }); + + theBatch.QuickAppends.Count.ShouldBe(2); + + var leftAction = theBatch.QuickAppends.Single(x => x.Id == left); + leftAction.Events.Select(x => x.Data).ShouldBe([a, c]); + leftAction.Events.ShouldAllBe(x => x.StreamId == left); + + var rightAction = theBatch.QuickAppends.Single(x => x.Id == right); + rightAction.Events.Select(x => x.Data).ShouldBe([b]); + rightAction.Events.ShouldAllBe(x => x.StreamId == right); + } + + [Fact] + public async Task a_multi_stream_slice_writes_each_raised_event_only_to_the_stream_it_was_raised_to_with_string_keys() + { + var a = new AEvent(); + var b = new BEvent(); + var c = new CEvent(); + + var slice = new EventSlice("slice", "foo", [eventOnStream("elsewhere", 4)]); + + await applyAsync(slice, AggregationScope.MultiStream, StreamIdentity.AsString, s => + { + s.AppendEvent("left", a); + s.AppendEvent("right", b); + s.AppendEvent("left", c); + }); + + theBatch.QuickAppends.Count.ShouldBe(2); + + var leftAction = theBatch.QuickAppends.Single(x => x.Key == "left"); + leftAction.Events.Select(x => x.Data).ShouldBe([a, c]); + leftAction.Events.ShouldAllBe(x => x.StreamKey == "left"); + + var rightAction = theBatch.QuickAppends.Single(x => x.Key == "right"); + rightAction.Events.Select(x => x.Data).ShouldBe([b]); + rightAction.Events.ShouldAllBe(x => x.StreamKey == "right"); + } + + [Fact] + public async Task a_new_single_stream_slice_numbers_only_its_own_stream() + { + // The shape the Marten docs recommend: one event back onto this stream, one onto another + var own = Guid.NewGuid(); + var other = Guid.NewGuid(); + var mine = new AEvent(); + var theirs = new BEvent(); + + var slice = new EventSlice(own, "foo", [eventOnStream(own, 1), eventOnStream(own, 2)]); + + await applyAsync(slice, AggregationScope.SingleStream, StreamIdentity.AsGuid, s => + { + s.AppendEvent(mine); + s.AppendEvent(other, theirs); + }); + + var versioned = theBatch.VersionedAppends.ShouldHaveSingleItem(); + versioned.Action.Id.ShouldBe(own); + versioned.Event.Data.ShouldBe(mine); + versioned.Version.ShouldBe(3); + + var updated = theBatch.VersionUpdates.ShouldHaveSingleItem(); + updated.Id.ShouldBe(own); + updated.Version.ShouldBe(3); + updated.ExpectedVersionOnServer.ShouldBe(2); + + // Nothing on this slice says what version the other stream is at, so the store has to work it out + var appended = theBatch.QuickAppends.ShouldHaveSingleItem(); + appended.Id.ShouldBe(other); + appended.ExpectedVersionOnServer.ShouldBeNull(); + appended.Events.Select(x => x.Data).ShouldBe([theirs]); + } + + [Fact] + public async Task a_new_single_stream_slice_numbers_only_its_own_stream_with_string_keys() + { + var mine = new AEvent(); + var theirs = new BEvent(); + + var slice = new EventSlice("own", "foo", [eventOnStream("own", 1), eventOnStream("own", 2)]); + + await applyAsync(slice, AggregationScope.SingleStream, StreamIdentity.AsString, s => + { + s.AppendEvent(mine); + s.AppendEvent("other", theirs); + }); + + var versioned = theBatch.VersionedAppends.ShouldHaveSingleItem(); + versioned.Action.Key.ShouldBe("own"); + versioned.Event.Data.ShouldBe(mine); + versioned.Version.ShouldBe(3); + + var updated = theBatch.VersionUpdates.ShouldHaveSingleItem(); + updated.Key.ShouldBe("own"); + updated.Version.ShouldBe(3); + updated.ExpectedVersionOnServer.ShouldBe(2); + + var appended = theBatch.QuickAppends.ShouldHaveSingleItem(); + appended.Key.ShouldBe("other"); + appended.ExpectedVersionOnServer.ShouldBeNull(); + appended.Events.Select(x => x.Data).ShouldBe([theirs]); + } + + [Fact] + public async Task a_new_single_stream_slice_raising_only_to_another_stream_does_not_assume_its_version() + { + var own = Guid.NewGuid(); + var other = Guid.NewGuid(); + + var slice = new EventSlice(own, "foo", [eventOnStream(own, 1), eventOnStream(own, 2)]); + + await applyAsync(slice, AggregationScope.SingleStream, StreamIdentity.AsGuid, + s => s.AppendEvent(other, new BEvent())); + + theBatch.VersionedAppends.ShouldBeEmpty(); + theBatch.VersionUpdates.ShouldBeEmpty(); + + var appended = theBatch.QuickAppends.ShouldHaveSingleItem(); + appended.Id.ShouldBe(other); + appended.ExpectedVersionOnServer.ShouldBeNull(); + } + + [Fact] + public async Task a_new_single_stream_slice_raising_to_itself_continues_its_version_sequence() + { + var own = Guid.NewGuid(); + var first = new AEvent(); + var second = new BEvent(); + + var slice = new EventSlice(own, "foo", [eventOnStream(own, 1), eventOnStream(own, 2)]); + + await applyAsync(slice, AggregationScope.SingleStream, StreamIdentity.AsGuid, s => + { + s.AppendEvent(first); + s.AppendEvent(second); + }); + + theBatch.VersionedAppends.Select(x => (x.Event.Data, x.Version)).ShouldBe([(first, 3L), (second, 4L)]); + theBatch.QuickAppends.ShouldBeEmpty(); + + var updated = theBatch.VersionUpdates.ShouldHaveSingleItem(); + updated.Version.ShouldBe(4); + updated.ExpectedVersionOnServer.ShouldBe(2); + } + + [Fact] + public async Task a_single_stream_slice_appending_to_an_existing_stream_leaves_the_version_to_the_store() + { + var own = Guid.NewGuid(); + + var slice = new EventSlice(own, "foo", [eventOnStream(own, 5)]); + + await applyAsync(slice, AggregationScope.SingleStream, StreamIdentity.AsGuid, + s => s.AppendEvent(new AEvent())); + + theBatch.VersionedAppends.ShouldBeEmpty(); + theBatch.VersionUpdates.ShouldBeEmpty(); + theBatch.QuickAppends.ShouldHaveSingleItem().Id.ShouldBe(own); + } + + private class RecordingBatch : IProjectionBatch + { + public List QuickAppends { get; } = new(); + + // The version is read as it arrives, since the event instance is free to change afterwards + public List<(StreamAction Action, IEvent Event, long Version)> VersionedAppends { get; } = new(); + + public List VersionUpdates { get; } = new(); + + public void QuickAppendEventWithVersion(StreamAction action, IEvent @event) + => VersionedAppends.Add((action, @event, @event.Version)); + + public void UpdateStreamVersion(StreamAction action) => VersionUpdates.Add(action); + + public void QuickAppendEvents(StreamAction action) => QuickAppends.Add(action); + + public Task ExecuteAsync(CancellationToken token) => Task.CompletedTask; + + public Task PublishMessageAsync(object message, string tenantId) => Task.CompletedTask; + + public ValueTask RecordProgress(EventRange range) => ValueTask.CompletedTask; + + public ValueTask DisposeAsync() => ValueTask.CompletedTask; + } +} diff --git a/src/JasperFx.Events/EventSlice.cs b/src/JasperFx.Events/EventSlice.cs index 3c3ebf49..c01e0ee0 100644 --- a/src/JasperFx.Events/EventSlice.cs +++ b/src/JasperFx.Events/EventSlice.cs @@ -309,29 +309,9 @@ public void BuildOperations( foreach (var group in groups) { - var action = StreamAction.Append(group.Key, RaisedEvents.ToArray()); - action.TenantId = TenantId; - - if (aggregationScope == AggregationScope.SingleStream && ActionType == StreamActionType.Start) - { - var version = _events.Count; - action.ExpectedVersionOnServer = version; - - foreach (var @event in RaisedEvents) - { - @event.Version = ++version; - storage.QuickAppendEventWithVersion(action, @event); - } - - action.Version = version; - - storage.UpdateStreamVersion(action); - } - else - { - action.TenantId = TenantId; - storage.QuickAppendEvents(action); - } + var action = StreamAction.Append(group.Key, group.ToArray()); + appendRaisedEvents(action, storage, aggregationScope, + group.Key == _events.FirstOrDefault()?.StreamId); } } else @@ -341,33 +321,42 @@ public void BuildOperations( foreach (var group in groups) { - var action = StreamAction.Append(group.Key, RaisedEvents.ToArray()); - action.TenantId = TenantId; - - if (aggregationScope == AggregationScope.SingleStream && ActionType == StreamActionType.Start) - { - var version = _events.Count; - action.ExpectedVersionOnServer = version; - - foreach (var @event in RaisedEvents) - { - @event.Version = ++version; - storage.QuickAppendEventWithVersion(action, @event); - } - - action.Version = version; - - storage.UpdateStreamVersion(action); - } - else - { - action.TenantId = TenantId; - storage.QuickAppendEvents(action); - } + var action = StreamAction.Append(group.Key, group.ToArray()); + appendRaisedEvents(action, storage, aggregationScope, + group.Key == _events.FirstOrDefault()?.StreamKey); } } } + private void appendRaisedEvents(StreamAction action, IProjectionBatch storage, + AggregationScope aggregationScope, bool isOwnStream) + { + action.TenantId = TenantId; + + // A slice that starts a stream holds that stream from its first event, so _events.Count is that + // stream's version. It says nothing about any other stream a projection raises events to, and + // those are left to the store to number + if (isOwnStream && aggregationScope == AggregationScope.SingleStream && ActionType == StreamActionType.Start) + { + var version = _events.Count; + action.ExpectedVersionOnServer = version; + + foreach (var @event in action.Events) + { + @event.Version = ++version; + storage.QuickAppendEventWithVersion(action, @event); + } + + action.Version = version; + + storage.UpdateStreamVersion(action); + } + else + { + storage.QuickAppendEvents(action); + } + } + /// /// Replace the supplied event within the slice with the new event data /// but with a copy of the original event's metadata. This is meant for diff --git a/src/JasperFx/Core/StreamExtensions.cs b/src/JasperFx/Core/StreamExtensions.cs index ca3d7897..22a35ca1 100644 --- a/src/JasperFx/Core/StreamExtensions.cs +++ b/src/JasperFx/Core/StreamExtensions.cs @@ -61,6 +61,12 @@ public static async Task ReadAllBytesAsync(this Stream stream) /// /// Read Byte array from stream /// + /// + /// The array is always long. If the stream ends first the remainder is + /// left zeroed and nothing says so, which makes a short read look like zeros. Use + /// when a short read + /// has to fail. + /// /// /// /// @@ -88,7 +94,8 @@ public static async Task ReadExpectedBufferAsync(this Stream stream, byte[ { try { - var bytes = await stream.ReadBytesAsync(expected.Length).ConfigureAwait(false); + var bytes = new byte[expected.Length]; + await stream.ReadExactlyAsync(bytes).ConfigureAwait(false); return expected.SequenceEqual(bytes); } catch (Exception) diff --git a/src/JasperFx/Core/StringExtensions.cs b/src/JasperFx/Core/StringExtensions.cs index 4fdf6a2d..f2e318e8 100644 --- a/src/JasperFx/Core/StringExtensions.cs +++ b/src/JasperFx/Core/StringExtensions.cs @@ -255,12 +255,19 @@ public static bool IsValidNumber(this string number) public static bool IsValidNumber(this string number, CultureInfo culture) { + var groupSeparator = Regex.Escape(culture.NumberFormat.NumberGroupSeparator); + var decimalSeparator = Regex.Escape(culture.NumberFormat.NumberDecimalSeparator); + + // A whole part with an optional fraction, or a bare fraction, so that there is always a digit. + // \z rather than $, which would allow a trailing newline var _validNumberPattern = - @"^-?(?:\d+|\d{1,3}(?:" - + culture.NumberFormat.NumberGroupSeparator + - @"\d{3})+)?(?:\" - + culture.NumberFormat.NumberDecimalSeparator + - @"\d+)?$"; + @"^-?(?:(?:\d+|\d{1,3}(?:" + + groupSeparator + + @"\d{3})+)(?:" + + decimalSeparator + + @"\d+)?|" + + decimalSeparator + + @"\d+)\z"; return new Regex(_validNumberPattern, RegexOptions.ECMAScript).IsMatch(number); }