Skip to content

Commit 7777c90

Browse files
KSemenenkoKSemenenko
committed
fix: join CQRS producers after cancellation callback failures
1 parent bc3119f commit 7777c90

8 files changed

Lines changed: 398 additions & 4 deletions

File tree

‎Directory.Build.props‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,8 +27,8 @@
2727
<RepositoryUrl>https://github.com/managedcode/Communication</RepositoryUrl>
2828
<PackageProjectUrl>https://github.com/managedcode/Communication</PackageProjectUrl>
2929
<Product>Managed Code - Communication</Product>
30-
<Version>10.2.10</Version>
31-
<PackageVersion>10.2.10</PackageVersion>
30+
<Version>10.2.11</Version>
31+
<PackageVersion>10.2.11</PackageVersion>
3232

3333
</PropertyGroup>
3434
<PropertyGroup Condition="'$(GITHUB_ACTIONS)' == 'true'">

‎ManagedCode.Communication.Tests/CQRS/CqrsStreamCancellationJoinTests.cs‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
using System.Threading.Tasks;
2+
3+
namespace ManagedCode.Communication.Tests.CQRS;
4+
5+
public sealed class CqrsStreamCancellationJoinTests
6+
{
7+
[Test]
8+
public Task Create_EarlyDisposalJoinsProducerAfterCancellationCallbackFailure()
9+
=> CqrsStreamCancellationJoinScenario.AssertFaultingCallbackJoinAsync();
10+
11+
[Test]
12+
public Task Create_OrdinaryEarlyDisposalStillJoinsProducer()
13+
=> CqrsStreamCancellationJoinScenario.AssertOrdinaryCancellationJoinAsync();
14+
}

‎ManagedCode.Communication.Tests/CQRS/Infrastructure/CqrsStreamCancellationJoinCleanup.cs‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Runtime.ExceptionServices;
4+
using System.Threading.Tasks;
5+
6+
namespace ManagedCode.Communication.Tests.CQRS;
7+
8+
internal static class CqrsStreamCancellationJoinCleanup
9+
{
10+
private const string AggregateFailureMessage = "The CQRS stream operation and cleanup both failed.";
11+
12+
internal static List<Exception> Leaves(Exception failure)
13+
{
14+
var leaves = new List<Exception>();
15+
AddLeaves(failure, leaves);
16+
return leaves;
17+
}
18+
19+
internal static async Task ObserveTaskAsync(Task task, List<Exception> failures)
20+
{
21+
await task.ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
22+
if (task.Exception is { } aggregate)
23+
{
24+
failures.AddRange(aggregate.InnerExceptions);
25+
}
26+
else if (task.IsCanceled)
27+
{
28+
failures.Add(new TaskCanceledException(task));
29+
}
30+
}
31+
32+
internal static void ThrowFailures(Exception? primary, IReadOnlyList<Exception> cleanup)
33+
{
34+
if (primary is null && cleanup.Count == 0)
35+
{
36+
return;
37+
}
38+
if (primary is not null && cleanup.Count == 0)
39+
{
40+
ExceptionDispatchInfo.Capture(primary).Throw();
41+
}
42+
if (primary is null && cleanup.Count == 1)
43+
{
44+
ExceptionDispatchInfo.Capture(cleanup[0]).Throw();
45+
}
46+
47+
var failures = new List<Exception>(cleanup.Count + (primary is null ? 0 : 1));
48+
if (primary is not null)
49+
{
50+
failures.Add(primary);
51+
}
52+
failures.AddRange(cleanup);
53+
throw new AggregateException(AggregateFailureMessage, failures);
54+
}
55+
56+
private static void AddLeaves(Exception exception, List<Exception> leaves)
57+
{
58+
if (exception is AggregateException aggregate)
59+
{
60+
foreach (var inner in aggregate.InnerExceptions)
61+
{
62+
AddLeaves(inner, leaves);
63+
}
64+
}
65+
else
66+
{
67+
leaves.Add(exception);
68+
}
69+
}
70+
}

‎ManagedCode.Communication.Tests/CQRS/Infrastructure/CqrsStreamCancellationJoinProbe.cs‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
using System;
2+
using System.Threading;
3+
using System.Threading.Tasks;
4+
using ManagedCode.Communication.CQRS;
5+
6+
namespace ManagedCode.Communication.Tests.CQRS;
7+
8+
internal sealed class CqrsStreamCancellationJoinProbe
9+
{
10+
private const string StartedState = "started";
11+
private const string FinalState = "complete";
12+
private const string CallbackFailureMessage = "The registered cancellation callback failed.";
13+
private readonly TaskCompletionSource release = new(TaskCreationOptions.RunContinuationsAsynchronously);
14+
private readonly TaskCompletionSource handlerEntered = new(TaskCreationOptions.RunContinuationsAsynchronously);
15+
private readonly TaskCompletionSource callbackRegistered = new(TaskCreationOptions.RunContinuationsAsynchronously);
16+
private readonly TaskCompletionSource callbackEntered = new(TaskCreationOptions.RunContinuationsAsynchronously);
17+
private readonly TaskCompletionSource finallyEntered = new(TaskCreationOptions.RunContinuationsAsynchronously);
18+
private readonly TaskCompletionSource handlerSettled = new(TaskCreationOptions.RunContinuationsAsynchronously);
19+
20+
internal Task HandlerEntered => handlerEntered.Task;
21+
internal Task CallbackRegistered => callbackRegistered.Task;
22+
internal Task CallbackEntered => callbackEntered.Task;
23+
internal Task FinallyEntered => finallyEntered.Task;
24+
internal Task HandlerSettled => handlerSettled.Task;
25+
internal InvalidOperationException CallbackFailure { get; } = new(CallbackFailureMessage);
26+
27+
internal async ValueTask<Result<FinalResult>> HandleAsync(ICqrsStreamWriter<ProgressUpdate, FinalResult> writer)
28+
{
29+
handlerEntered.TrySetResult();
30+
try
31+
{
32+
await writer.StartedAsync(new ProgressUpdate(StartedState));
33+
using var registration = writer.CancellationToken.Register(
34+
static state => ((CqrsStreamCancellationJoinProbe)state!).ThrowFromCallback(), this);
35+
callbackRegistered.TrySetResult();
36+
await Task.Delay(Timeout.InfiniteTimeSpan, writer.CancellationToken).ConfigureAwait(true);
37+
return Result<FinalResult>.Succeed(new FinalResult(FinalState));
38+
}
39+
finally
40+
{
41+
finallyEntered.TrySetResult();
42+
await release.Task.ConfigureAwait(true);
43+
handlerSettled.TrySetResult();
44+
}
45+
}
46+
47+
internal void ReleaseHandler() => release.TrySetResult();
48+
49+
private void ThrowFromCallback()
50+
{
51+
callbackEntered.TrySetResult();
52+
throw CallbackFailure;
53+
}
54+
}

‎ManagedCode.Communication.Tests/CQRS/Infrastructure/CqrsStreamCancellationJoinScenario.cs‎

Lines changed: 156 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,156 @@
1+
using System;
2+
using System.Collections.Generic;
3+
using System.Threading;
4+
using System.Threading.Tasks;
5+
using ManagedCode.Communication.CQRS;
6+
using Shouldly;
7+
8+
namespace ManagedCode.Communication.Tests.CQRS;
9+
10+
internal static class CqrsStreamCancellationJoinScenario
11+
{
12+
private const string StartedState = "started";
13+
private const string FinalState = "complete";
14+
private const int OperationTimeoutSeconds = 10;
15+
private const int PrematureCompletionObservationMilliseconds = 250;
16+
17+
internal static async Task AssertFaultingCallbackJoinAsync()
18+
{
19+
var probe = new CqrsStreamCancellationJoinProbe();
20+
var enumerator = CqrsStream.Create<ProgressUpdate, FinalResult>(probe.HandleAsync).GetAsyncEnumerator();
21+
await RunFaultingCallbackJoinAsync(probe, enumerator);
22+
}
23+
24+
internal static async Task AssertOrdinaryCancellationJoinAsync()
25+
{
26+
using var cancellation = new CancellationTokenSource();
27+
var handlerSettled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
28+
var enumerator = CqrsStream.Create<ProgressUpdate, FinalResult>(async writer =>
29+
{
30+
try
31+
{
32+
await writer.StartedAsync(new ProgressUpdate(StartedState));
33+
await Task.Delay(Timeout.InfiniteTimeSpan, writer.CancellationToken).ConfigureAwait(true);
34+
return Result<FinalResult>.Succeed(new FinalResult(FinalState));
35+
}
36+
finally
37+
{
38+
handlerSettled.TrySetResult();
39+
}
40+
}).GetAsyncEnumerator(cancellation.Token);
41+
await RunOrdinaryCancellationJoinAsync(enumerator, handlerSettled.Task, cancellation);
42+
}
43+
44+
private static async Task RunFaultingCallbackJoinAsync(
45+
CqrsStreamCancellationJoinProbe probe, IAsyncEnumerator<Chunk> enumerator)
46+
{
47+
Task<bool>? firstMove = null;
48+
Task? disposal = null;
49+
Exception? primaryFailure = null;
50+
var cleanupFailures = new List<Exception>();
51+
var callbackFailureObserved = false;
52+
try
53+
{
54+
firstMove = enumerator.MoveNextAsync().AsTask();
55+
await probe.HandlerEntered.WaitAsync(TimeSpan.FromSeconds(OperationTimeoutSeconds));
56+
(await firstMove.WaitAsync(TimeSpan.FromSeconds(OperationTimeoutSeconds))).ShouldBeTrue();
57+
enumerator.Current.Kind.ShouldBe(CqrsStreamChunkKind.Started);
58+
await probe.CallbackRegistered.WaitAsync(TimeSpan.FromSeconds(OperationTimeoutSeconds));
59+
disposal = DisposeEnumeratorAsync(enumerator);
60+
await probe.CallbackEntered.WaitAsync(TimeSpan.FromSeconds(OperationTimeoutSeconds));
61+
await probe.FinallyEntered.WaitAsync(TimeSpan.FromSeconds(OperationTimeoutSeconds));
62+
await Task.WhenAny(disposal, Task.Delay(TimeSpan.FromMilliseconds(
63+
PrematureCompletionObservationMilliseconds)));
64+
disposal.IsCompleted.ShouldBeFalse();
65+
probe.ReleaseHandler();
66+
await probe.HandlerSettled.WaitAsync(TimeSpan.FromSeconds(OperationTimeoutSeconds));
67+
await AssertOnlyCallbackFailureAsync(disposal, probe.CallbackFailure);
68+
callbackFailureObserved = true;
69+
await AssertHealthyFollowUpAsync();
70+
}
71+
catch (Exception exception)
72+
{
73+
primaryFailure = exception;
74+
}
75+
finally
76+
{
77+
probe.ReleaseHandler();
78+
disposal ??= DisposeEnumeratorAsync(enumerator);
79+
if (!callbackFailureObserved)
80+
{
81+
await CqrsStreamCancellationJoinCleanup.ObserveTaskAsync(disposal, cleanupFailures);
82+
}
83+
if (firstMove is not null)
84+
{
85+
await CqrsStreamCancellationJoinCleanup.ObserveTaskAsync(firstMove, cleanupFailures);
86+
}
87+
await probe.HandlerSettled;
88+
}
89+
CqrsStreamCancellationJoinCleanup.ThrowFailures(primaryFailure, cleanupFailures);
90+
}
91+
92+
private static async Task RunOrdinaryCancellationJoinAsync(
93+
IAsyncEnumerator<Chunk> enumerator, Task handlerSettled, CancellationTokenSource cancellation)
94+
{
95+
Task<bool>? firstMove = null;
96+
Task? disposal = null;
97+
Exception? primaryFailure = null;
98+
var cleanupFailures = new List<Exception>();
99+
try
100+
{
101+
firstMove = enumerator.MoveNextAsync().AsTask();
102+
(await firstMove.WaitAsync(TimeSpan.FromSeconds(OperationTimeoutSeconds))).ShouldBeTrue();
103+
enumerator.Current.Kind.ShouldBe(CqrsStreamChunkKind.Started);
104+
disposal = DisposeEnumeratorAsync(enumerator);
105+
await handlerSettled.WaitAsync(TimeSpan.FromSeconds(OperationTimeoutSeconds));
106+
await disposal.WaitAsync(TimeSpan.FromSeconds(OperationTimeoutSeconds));
107+
}
108+
catch (Exception exception)
109+
{
110+
primaryFailure = exception;
111+
}
112+
finally
113+
{
114+
await CqrsStreamCancellationJoinCleanup.ObserveTaskAsync(
115+
cancellation.CancelAsync(), cleanupFailures);
116+
disposal ??= DisposeEnumeratorAsync(enumerator);
117+
await CqrsStreamCancellationJoinCleanup.ObserveTaskAsync(disposal, cleanupFailures);
118+
if (firstMove is not null)
119+
{
120+
await CqrsStreamCancellationJoinCleanup.ObserveTaskAsync(firstMove, cleanupFailures);
121+
}
122+
await handlerSettled;
123+
}
124+
CqrsStreamCancellationJoinCleanup.ThrowFailures(primaryFailure, cleanupFailures);
125+
}
126+
127+
private static async Task AssertOnlyCallbackFailureAsync(Task disposal, Exception expected)
128+
{
129+
var failure = await Should.ThrowAsync<Exception>(
130+
() => disposal.WaitAsync(TimeSpan.FromSeconds(OperationTimeoutSeconds)));
131+
var leaves = CqrsStreamCancellationJoinCleanup.Leaves(failure!);
132+
leaves.Count.ShouldBe(1);
133+
leaves[0].ShouldBeSameAs(expected);
134+
}
135+
136+
private static async Task AssertHealthyFollowUpAsync()
137+
{
138+
var chunks = new List<Chunk>();
139+
await foreach (var chunk in CqrsStream.Create<ProgressUpdate, FinalResult>(async writer =>
140+
{
141+
await writer.StartedAsync(new ProgressUpdate(StartedState));
142+
return Result<FinalResult>.Succeed(new FinalResult(FinalState));
143+
}))
144+
{
145+
chunks.Add(chunk);
146+
}
147+
chunks.ConvertAll(chunk => chunk.Kind).ShouldBe(
148+
[CqrsStreamChunkKind.Started, CqrsStreamChunkKind.Completed]);
149+
}
150+
151+
private static async Task DisposeEnumeratorAsync(IAsyncEnumerator<Chunk> enumerator)
152+
{
153+
await enumerator.DisposeAsync().ConfigureAwait(false);
154+
}
155+
156+
}

‎ManagedCode.Communication/Cqrs/CqrsStream.cs‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,8 @@ private static async IAsyncEnumerable<CqrsStreamChunk<TProgress, TResult>> Creat
118118
var writer = new CqrsStreamWriter<TProgress, TResult>(channel.Writer, token);
119119
var producer = ProduceAsync(handler, writer, channel, token);
120120

121+
Exception? iterationFailure = null;
122+
121123
try
122124
{
123125
while (true)
@@ -140,14 +142,19 @@ private static async IAsyncEnumerable<CqrsStreamChunk<TProgress, TResult>> Creat
140142
{
141143
break;
142144
}
145+
catch (Exception error)
146+
{
147+
iterationFailure = error;
148+
break;
149+
}
143150

144151
yield return chunk;
145152
}
146153
}
147154
finally
148155
{
149-
await streamCancellation.CancelAsync().ConfigureAwait(true);
150-
await producer.ConfigureAwait(true);
156+
await CqrsStreamProducerSettlement.SettleAsync(streamCancellation, producer, iterationFailure)
157+
.ConfigureAwait(true);
151158
}
152159
}
153160

0 commit comments

Comments
 (0)