Support for IAsyncEnumerable<T> returning methods in grain observers.

global.json

@@ -1,7 +1,7 @@
 {
   "sdk": {
     "rollForward": "major",
-    "version": "11.0.100-rc.2.26504.105"
+    "version": "11.0.100-rc.3.26522.148"
   },
   "test": {
     "runner": "Microsoft.Testing.Platform"

src/Orleans.Core.Abstractions/Exceptions/ObserverNotReachableException.cs

@@ -0,0 +1,41 @@
+namespace Orleans.Runtime;
+
+/// <summary>
+/// Indicates that an observer is no longer reachable.
+/// </summary>
+/// <remarks>
+/// Thrown when a call is made to an observer that is not (or is no longer) reachable,
+/// for example because it was deregistered, garbage collected, or its hosting extension was disposed.
+/// This exception is terminal: retrying the same call will not succeed!
+/// </remarks>
+[Serializable]
+[GenerateSerializer]
+#pragma warning disable RCS1194 // Implement exception constructors.
+public sealed class ObserverNotReachableException : OrleansException
+#pragma warning restore RCS1194 // Implement exception constructors.
+{
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ObserverNotReachableException"/> class.
+    /// </summary>
+    /// <param name="message">
+    /// The exception message.
+    /// </param>
+    internal ObserverNotReachableException(string message)
+        : base(message)
+    {
+    }
+
+    /// <summary>
+    /// Initializes a new instance of the <see cref="ObserverNotReachableException"/> class.
+    /// </summary>
+    /// <param name="message">
+    /// The message.
+    /// </param>
+    /// <param name="innerException">
+    /// The inner exception.
+    /// </param>
+    internal ObserverNotReachableException(string message, Exception innerException)
+        : base(message, innerException)
+    {
+    }
+}

src/Orleans.Core.Abstractions/Exceptions/OrleansMessageRejectionException.cs

@@ -57,4 +57,3 @@ protected OrleansMessageRejectionException(SerializationInfo info, StreamingCont
         }
     }
 }
-

src/Orleans.Core/Runtime/AsyncEnumerableGrainExtension.cs

@@ -117,6 +117,11 @@ bool MarkAndCheck(Guid requestId)
     /// <inheritdoc/>
     public ValueTask<(EnumerationResult Status, object? Value)> StartEnumeration<T>(Guid requestId, [Immutable] IAsyncEnumerableRequest<T> request, CancellationToken cancellationToken)
     {
+        if (request.MaxBatchSize <= 0)
+        {
+            return ThrowInvalidBatchSize(request);
+        }
+
         ref var entry = ref CollectionsMarshal.GetValueRefOrAddDefault(_enumerators, requestId, out bool exists);
         if (exists)
         {
@@ -130,10 +135,15 @@ bool MarkAndCheck(Guid requestId)
         entry.Enumerator = enumerator;
         entry.MaxBatchSize = request.MaxBatchSize;
         entry.CancellationTokenSource = cts;
-        Debug.Assert(entry.MaxBatchSize > 0, "Max batch size must be positive.");
         return MoveNextCore(ref entry, requestId, enumerator, cancellationToken);
 
-        static ValueTask<(EnumerationResult Status, object? Value)> ThrowAlreadyExists() => ValueTask.FromException<(EnumerationResult Status, object? Value)>(new InvalidOperationException("An enumerator with the same id already exists."));
+        static ValueTask<(EnumerationResult Status, object? Value)> ThrowAlreadyExists() =>
+            ValueTask.FromException<(EnumerationResult Status, object? Value)>(
+                new InvalidOperationException("An enumerator with the same id already exists."));
+
+        static ValueTask<(EnumerationResult Status, object? Value)> ThrowInvalidBatchSize(IAsyncEnumerableRequest<T> request) =>
+            ValueTask.FromException<(EnumerationResult Status, object? Value)>(
+                new ArgumentOutOfRangeException(nameof(request), request.MaxBatchSize, "Max batch size must be positive."));
     }
 
     /// <inheritdoc/>