Skip to content

Commit 112f6ac

Browse files
KSemenenkoKSemenenko
committed
Preserve Orleans scheduler when consuming CQRS streams
1 parent 7d43747 commit 112f6ac

8 files changed

Lines changed: 72 additions & 15 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.5</Version>
31-
<PackageVersion>10.2.5</PackageVersion>
30+
<Version>10.2.6</Version>
31+
<PackageVersion>10.2.6</PackageVersion>
3232

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

‎ManagedCode.Communication.Tests/Orleans/CqrsOrleansIntegrationTests.cs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,15 @@ public CqrsOrleansIntegrationTests(OrleansClusterFixture fixture)
2828
_grainFactory = fixture.Cluster.GrainFactory;
2929
}
3030

31+
[Test]
32+
[Arguments(false)]
33+
[Arguments(true)]
34+
public async Task StreamConsumptionPreservesGrainScheduler(bool collectOutcome)
35+
{
36+
var grain = _grainFactory.GetGrain<ICqrsProbeGrain>(Guid.NewGuid());
37+
(await grain.DrainOnGrainSchedulerAsync(collectOutcome)).ShouldBeTrue();
38+
}
39+
3140
[Test]
3241
public async Task ACompletedChunkRoundTripsThroughAGrain()
3342
{

‎ManagedCode.Communication.Tests/Orleans/Grains/CqrsProbeGrain.cs‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,4 +27,36 @@ public async IAsyncEnumerable<CqrsStreamChunk<OrleansProgress, OrleansReport>> S
2727
yield return CqrsStreamChunk<OrleansProgress, OrleansReport>.Completed(
2828
new OrleansReport("done"), sequence: 3);
2929
}
30+
public async Task<bool> DrainOnGrainSchedulerAsync(bool collectOutcome)
31+
{
32+
var scheduler = TaskScheduler.Current;
33+
var preserved = scheduler != TaskScheduler.Default;
34+
var stream = CheckContextAsync();
35+
if (collectOutcome)
36+
{
37+
await stream.ToOutcomeAsync();
38+
}
39+
else
40+
{
41+
await stream.ToResultAsync(async (_, _) =>
42+
{
43+
preserved &= TaskScheduler.Current == scheduler;
44+
await Task.Delay(10);
45+
preserved &= TaskScheduler.Current == scheduler;
46+
});
47+
}
48+
return preserved;
49+
50+
async IAsyncEnumerable<CqrsStreamChunk<int, bool>> CheckContextAsync()
51+
{
52+
for (var index = 0; index < 3; index++)
53+
{
54+
preserved &= TaskScheduler.Current == scheduler;
55+
await Task.Delay(10);
56+
preserved &= TaskScheduler.Current == scheduler;
57+
yield return CqrsStreamChunk<int, bool>.Progress(index);
58+
}
59+
yield return CqrsStreamChunk<int, bool>.Completed(true);
60+
}
61+
}
3062
}

‎ManagedCode.Communication.Tests/Orleans/Grains/ICqrsProbeGrain.cs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,4 +22,6 @@ Task<CqrsStreamChunk<OrleansProgress, OrleansReport>> EchoChunkAsync(
2222
CqrsStreamChunk<OrleansProgress, OrleansReport> chunk);
2323

2424
IAsyncEnumerable<CqrsStreamChunk<OrleansProgress, OrleansReport>> StreamAsync();
25+
26+
Task<bool> DrainOnGrainSchedulerAsync(bool collectOutcome);
2527
}

‎ManagedCode.Communication/Cqrs/CqrsStream.cs‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ public static IAsyncEnumerable<CqrsStreamChunk<TProgress, TResult>> Create<TProg
6363
ArgumentNullException.ThrowIfNull(handler);
6464

6565
return CreateCore<TProgress, TResult>(
66-
async writer => Result<TResult>.Succeed(await handler(writer).ConfigureAwait(false)),
66+
async writer => Result<TResult>.Succeed(await handler(writer).ConfigureAwait(true)),
6767
cancellationToken);
6868
}
6969

@@ -124,7 +124,7 @@ private static async IAsyncEnumerable<CqrsStreamChunk<TProgress, TResult>> Creat
124124

125125
try
126126
{
127-
if (!await channel.Reader.WaitToReadAsync(token).ConfigureAwait(false))
127+
if (!await channel.Reader.WaitToReadAsync(token).ConfigureAwait(true))
128128
{
129129
break;
130130
}
@@ -144,8 +144,8 @@ private static async IAsyncEnumerable<CqrsStreamChunk<TProgress, TResult>> Creat
144144
}
145145
finally
146146
{
147-
await streamCancellation.CancelAsync().ConfigureAwait(false);
148-
await producer.ConfigureAwait(false);
147+
await streamCancellation.CancelAsync().ConfigureAwait(true);
148+
await producer.ConfigureAwait(true);
149149
}
150150
}
151151

@@ -160,13 +160,13 @@ private static async Task ProduceAsync<TProgress, TResult>(
160160

161161
try
162162
{
163-
var final = await handler(writer).ConfigureAwait(false);
163+
var final = await handler(writer).ConfigureAwait(true);
164164

165165
var terminal = final.IsSuccess
166166
? CqrsStreamChunk<TProgress, TResult>.Completed(final, sequence: writer.NextSequence())
167167
: CqrsStreamChunk<TProgress, TResult>.Failed(final, sequence: writer.NextSequence());
168168

169-
await channel.Writer.WriteAsync(terminal, cancellationToken).ConfigureAwait(false);
169+
await channel.Writer.WriteAsync(terminal, cancellationToken).ConfigureAwait(true);
170170
}
171171
catch (OperationCanceledException)
172172
{
@@ -179,7 +179,7 @@ private static async Task ProduceAsync<TProgress, TResult>(
179179
await channel.Writer.WriteAsync(
180180
CqrsStreamChunk<TProgress, TResult>.FromException(exception, sequence: writer.NextSequence()),
181181
cancellationToken)
182-
.ConfigureAwait(false);
182+
.ConfigureAwait(true);
183183
}
184184
catch (OperationCanceledException)
185185
{

‎ManagedCode.Communication/Cqrs/CqrsStreamExtensions.cs‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -194,7 +194,7 @@ public static async Task<CqrsStreamOutcome<TProgress, TResult>> ToOutcomeAsync<T
194194

195195
var guarded = Guarded(stream, cancellationToken);
196196

197-
await foreach (var chunk in guarded.WithCancellation(cancellationToken).ConfigureAwait(false))
197+
await foreach (var chunk in guarded.WithCancellation(cancellationToken).ConfigureAwait(true))
198198
{
199199
chunks.Add(chunk);
200200

@@ -239,7 +239,7 @@ public static async Task<List<CqrsStreamChunk<TProgress, TResult>>> ToChunkListA
239239

240240
var chunks = new List<CqrsStreamChunk<TProgress, TResult>>();
241241

242-
await foreach (var chunk in stream.WithCancellation(cancellationToken).ConfigureAwait(false))
242+
await foreach (var chunk in stream.WithCancellation(cancellationToken).ConfigureAwait(true))
243243
{
244244
chunks.Add(chunk);
245245
}
@@ -288,15 +288,15 @@ private static async Task<Result<TResult>> ToResultAsync<TProgress, TResult>(
288288

289289
var guarded = Guarded(stream, cancellationToken);
290290

291-
await foreach (var chunk in guarded.WithCancellation(cancellationToken).ConfigureAwait(false))
291+
await foreach (var chunk in guarded.WithCancellation(cancellationToken).ConfigureAwait(true))
292292
{
293293
if ((onProgress is not null || onProgressAsync is not null) && chunk.TryGetProgress(out var payload))
294294
{
295295
onProgress?.Invoke(payload);
296296

297297
if (onProgressAsync is not null)
298298
{
299-
await onProgressAsync(payload, cancellationToken).ConfigureAwait(false);
299+
await onProgressAsync(payload, cancellationToken).ConfigureAwait(true);
300300
}
301301
}
302302

‎ManagedCode.Communication/Cqrs/CqrsStreamNormalizer.cs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ public static async IAsyncEnumerable<CqrsStreamChunk<TProgress, TResult>> Normal
1919
bool ensureTerminalChunk,
2020
[EnumeratorCancellation] CancellationToken cancellationToken = default)
2121
{
22+
// Enumeration and disposal must retain the caller scheduler, including Orleans grain identity.
2223
var enumerator = source.GetAsyncEnumerator(cancellationToken);
2324
var sequence = 0L;
2425
var sawTerminal = false;
@@ -32,7 +33,7 @@ public static async IAsyncEnumerable<CqrsStreamChunk<TProgress, TResult>> Normal
3233

3334
try
3435
{
35-
moved = await enumerator.MoveNextAsync().ConfigureAwait(false);
36+
moved = await enumerator.MoveNextAsync().ConfigureAwait(true);
3637
}
3738
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
3839
{
@@ -75,7 +76,7 @@ public static async IAsyncEnumerable<CqrsStreamChunk<TProgress, TResult>> Normal
7576
}
7677
finally
7778
{
78-
await enumerator.DisposeAsync().ConfigureAwait(false);
79+
await enumerator.DisposeAsync().ConfigureAwait(true);
7980
}
8081

8182
if (faultChunk is not null)

‎README.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2198,3 +2198,16 @@ This project is licensed under the MIT License - see the [LICENSE](LICENSE) file
21982198
`await stream.ToResultAsync(cancellationToken)` throws `OperationCanceledException` when the caller
21992199
cancels while the source finishes, including sources that end normally without a terminal chunk.
22002200
Cancellation is not converted into an incomplete-stream failure; callers can retain resumable state.
2201+
2202+
### CQRS streams inside Orleans grains
2203+
2204+
CQRS stream creation, normalization, `ToResultAsync`, and `ToOutcomeAsync` preserve the caller's scheduler and synchronization context. This keeps subsequent stream reads, disposal, and progress callbacks on the Orleans grain scheduler, preserving the calling grain identity across batches.
2205+
2206+
```csharp
2207+
var result = await otherGrain.StreamAsync().ToResultAsync(
2208+
async (progress, cancellationToken) =>
2209+
{
2210+
// Runs on this grain's scheduler; grain-owned state remains safe to update.
2211+
await SaveProgressAsync(progress, cancellationToken);
2212+
});
2213+
```

0 commit comments

Comments
 (0)