From 0751120cf0a760499a5b34e54072fb3436e23364 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 1 Nov 2026 06:27:33 -0600 Subject: [PATCH 1/3] Write raised events only to the stream they were raised to EventSlice.BuildOperations grouped RaisedEvents by stream and then handed every group all of them, so a slice raising to two streams wrote each event to both. It was worse than duplication: StreamAction.AddEvent stamps the stream identity onto the IEvent, and the instances were shared between the actions, so each later group re-tagged the events already queued for the earlier ones. The string identity branch had the same defect. The version numbering had a second one. A single stream slice that starts its stream numbers raised events from _events.Count, which is only true of the slice's own stream, but it was applied to every group. That is the shape Marten's docs recommend, a raise to this stream plus a raise to another, and it predicted the other stream's version and expected version from an unrelated event count. Only the slice's own stream is numbered now, and the store numbers the rest through QuickAppendEvents, as it already does for multi stream projections. The two branches differed only in their identity, so the append moves to one helper. A slice raising only to itself is unchanged. Co-Authored-By: Claude Opus 5.5 --- .../AggregationRunnerRaisedEventsTests.cs | 275 ++++++++++++++++++ src/JasperFx.Events/EventSlice.cs | 81 +++--- 2 files changed, 310 insertions(+), 46 deletions(-) create mode 100644 src/EventTests/Daemon/AggregationRunnerRaisedEventsTests.cs 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 From 2f8d7e828d09107577d947b013a947f22d7da1ea Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 1 Nov 2026 06:29:45 -0600 Subject: [PATCH 2/3] Stop IsValidNumber accepting what is not a number The group separator went into the pattern unescaped, so on de-DE, where it is ".", it matched any character and "1x234" was valid. Both parts of the pattern were optional, so "" and "-" were valid too, and $ matches before a trailing newline, so "123\n" was as well. Both separators are escaped now, the pattern has to find a digit, and it ends at \z. Everything that was a number before still is, including a bare fraction like ".5". This is fixed rather than deprecated even though nothing in the Critter Stack calls it. [Obsolete] would leave the wrong answers in place for any outside caller for the rest of 2.x, and there is no equivalent to send them to: decimal.TryParse accepts "1,00", which this rejects by design. Co-Authored-By: Claude Opus 5.5 --- src/CoreTests/StringExtensionsTester.cs | 49 +++++++++++++++++++++++++ src/JasperFx/Core/StringExtensions.cs | 17 ++++++--- 2 files changed, 61 insertions(+), 5 deletions(-) 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/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); } From 25832a864bf37dcab3ba96642ca1eb3a23ba9401 Mon Sep 17 00:00:00 2001 From: "Jeremy D. Miller" Date: Sun, 1 Nov 2026 06:31:04 -0600 Subject: [PATCH 3/3] Make ReadExpectedBufferAsync fail on a short read ReadBytesAsync hands back a zero-padded array when the stream ends early, so ReadExpectedBufferAsync compared the padding against what it expected. A stream holding [1] matched [1, 0], and an empty stream matched [0, 0]. It now reads with Stream.ReadExactlyAsync, and its existing catch turns the EndOfStreamException into false. ReadBytesAsync itself is left as it is on 2.x, with its contract documented and pinned by a test, because Wolverine depends on it. WireProtocol reads its length prefix through it and returns quietly on a zero length, which is how a connection that closes without sending anything, like a TCP health probe, is ignored. Throwing there, or returning a short array that BitConverter then rejects, would turn each of those into an error log and a write to a closed socket in every installed Wolverine that takes a newer JasperFx. Wolverine's acknowledgement reads discard this method's result, so they do not change. Co-Authored-By: Claude Opus 5.5 --- src/CoreTests/StreamExtensionsTests.cs | 83 ++++++++++++++++++++++++++ src/JasperFx/Core/StreamExtensions.cs | 9 ++- 2 files changed, 91 insertions(+), 1 deletion(-) 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/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)