IAsyncEnumerable: Fix cancellation propagation, bound operation times, use mark-and-sweep cleanup (#9387)

IAsyncEnumerable: bound polling, cancellation, and disposal time

Author: ReubenBond

src/Orleans.Core.Abstractions/Runtime/AsyncEnumerableRequest.cs

@@ -49,6 +49,11 @@ public enum EnumerationResult
     /// </summary>
     Error = 1 << 5,
 
+    /// <summary>
+    /// Enumeration was canceled.
+    /// </summary>
+    Canceled = 1 << 6,
+
     /// <summary>
     /// This result indicates that enumeration has completed and that no further results will be produced.
     /// </summary>
@@ -244,26 +249,26 @@ public async ValueTask<bool> MoveNextAsync()
         (EnumerationResult Status, object Value) result;
         while (true)
         {
-            if (_cancellationToken.IsCancellationRequested)
-            {
-                _current = default;
-                return false;
-            }
+            _cancellationToken.ThrowIfCancellationRequested();
 
             if (!_initialized)
             {
-                result = await _target.StartEnumeration(_requestId, _request);
+                result = await _target.StartEnumeration(_requestId, _request).AsTask().WaitAsync(_cancellationToken);
                 _initialized = true;
             }
             else
             {
-                result = await _target.MoveNext<T>(_requestId);
+                result = await _target.MoveNext<T>(_requestId).AsTask().WaitAsync(_cancellationToken);
             }
 
             if (result.Status is EnumerationResult.Error)
             {
                 ExceptionDispatchInfo.Capture((Exception)result.Value).Throw();
             }
+            else if (result.Status is EnumerationResult.Canceled)
+            {
+                throw new OperationCanceledException();
+            }
 
             if (result.Status is not EnumerationResult.Heartbeat)
             {
@@ -274,7 +279,7 @@ public async ValueTask<bool> MoveNextAsync()
         if (result.Status is EnumerationResult.MissingEnumeratorError)
         {
             throw new EnumerationAbortedException("Enumeration aborted: the remote target does not have a record of this enumerator."
-                + " This likely indicates that the remote grain was deactivated since enumeration begun.");
+                + " This likely indicates that the remote grain was deactivated since enumeration begun or that the enumerator was idle for longer than the expiration period.");
         }
 
         Debug.Assert((result.Status & (EnumerationResult.Element | EnumerationResult.Batch | EnumerationResult.Completed)) != 0);

src/Orleans.Core/Configuration/Options/MessagingOptions.cs

@@ -55,5 +55,10 @@ public TimeSpan ResponseTimeout
         /// </summary>
         /// <value>The maximum message body size is 100 MB by default.</value>
         public int MaxMessageBodySize { get; set; } = 100 * 1024 * 1024;
+
+        /// <summary>
+        /// Gets the response timeout underlying the <see cref="ResponseTimeout"/> property, without debugger checks.
+        /// </summary>
+        internal TimeSpan ConfiguredResponseTimeout => _responseTimeout;
     }
 }