Add cancellation handling to prevent UnobservedTaskException in Watcher

src/KubernetesClient.Classic/CancellationExtensions.cs

@@ -0,0 +1,56 @@
+namespace k8s
+{
+    /// <summary>
+    /// Polyfills for the cancellable overloads which are only available on modern .NET,
+    /// so the shared sources can rely on them unconditionally.
+    /// </summary>
+    internal static class CancellationExtensions
+    {
+        /// <summary>
+        /// Polyfill of <c>Task&lt;TResult&gt;.WaitAsync(CancellationToken)</c>.
+        /// </summary>
+        public static Task<TResult> WaitAsync<TResult>(this Task<TResult> task, CancellationToken cancellationToken)
+        {
+            if (task == null)
+            {
+                throw new ArgumentNullException(nameof(task));
+            }
+
+            if (task.IsCompleted || !cancellationToken.CanBeCanceled)
+            {
+                return task;
+            }
+
+            if (cancellationToken.IsCancellationRequested)
+            {
+                return Task.FromCanceled<TResult>(cancellationToken);
+            }
+
+            // Observe any exception from the original task to prevent an
+            // UnobservedTaskException when the continuation below is cancelled
+            // before the original task faults (e.g. the transport tears down the
+            // connection after cancellation).
+            _ = task.ContinueWith(
+                static t => { _ = t.Exception; },
+                CancellationToken.None,
+                TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously,
+                TaskScheduler.Default);
+
+            // here to pass cancellationToken into task
+            return task.ContinueWith(t => t.GetAwaiter().GetResult(), cancellationToken);
+        }
+
+        /// <summary>
+        /// Polyfill of <c>TextReader.ReadLineAsync(CancellationToken)</c>.
+        /// </summary>
+        public static Task<string> ReadLineAsync(this TextReader reader, CancellationToken cancellationToken)
+        {
+            if (reader == null)
+            {
+                throw new ArgumentNullException(nameof(reader));
+            }
+
+            return reader.ReadLineAsync().WaitAsync(cancellationToken);
+        }
+    }
+}

src/KubernetesClient.Classic/KubernetesClient.Classic.csproj

@@ -107,7 +107,6 @@
 
     <Compile Include="..\KubernetesClient\Watcher.cs" />
     <Compile Include="..\KubernetesClient\WatcherExt.cs" />
-    <Compile Include="..\KubernetesClient\LineSeparatedHttpContent.cs" />
 
     <Compile Include="..\KubernetesClient\Exceptions\KubeConfigException.cs" />
     <Compile Include="..\KubernetesClient\Exceptions\KubernetesClientException.cs" />

src/KubernetesClient.Classic/LineSeparatedHttpContent.cs

@@ -0,0 +1,211 @@
+using System.Net;
+using System.Net.Http;
+
+namespace k8s
+{
+    internal sealed class LineSeparatedHttpContent : HttpContent
+    {
+        private readonly HttpContent _originContent;
+        private readonly CancellationToken _cancellationToken;
+        private Stream _originStream;
+
+        public LineSeparatedHttpContent(HttpContent originContent, CancellationToken cancellationToken)
+        {
+            _originContent = originContent;
+            _cancellationToken = cancellationToken;
+        }
+
+        public TextReader StreamReader { get; private set; }
+
+        protected override async Task SerializeToStreamAsync(Stream stream, TransportContext context)
+        {
+            _originStream = await _originContent.ReadAsStreamAsync().ConfigureAwait(false);
+
+            var reader = new PeekableStreamReader(new CancelableStream(_originStream, _cancellationToken));
+            StreamReader = reader;
+
+            var firstLine = await reader.PeekLineAsync().ConfigureAwait(false);
+
+            var writer = new StreamWriter(stream);
+
+            await writer.WriteAsync(firstLine).ConfigureAwait(false);
+            await writer.FlushAsync().ConfigureAwait(false);
+        }
+
+        protected override bool TryComputeLength(out long length)
+        {
+            length = 0;
+            return false;
+        }
+
+        internal sealed class CancelableStream : Stream
+        {
+            private readonly Stream _innerStream;
+            private readonly CancellationToken _cancellationToken;
+
+            public CancelableStream(Stream innerStream, CancellationToken cancellationToken)
+            {
+                _innerStream = innerStream;
+                _cancellationToken = cancellationToken;
+            }
+
+            public override void Flush() =>
+                _innerStream.FlushAsync(_cancellationToken).GetAwaiter().GetResult();
+
+            public override async Task FlushAsync(CancellationToken cancellationToken)
+            {
+                using (var cancellationTokenSource = CreateCancellationTokenSource(cancellationToken))
+                {
+                    await _innerStream.FlushAsync(cancellationTokenSource.Token).ConfigureAwait(false);
+                }
+            }
+
+            public override int Read(byte[] buffer, int offset, int count) =>
+                _innerStream.ReadAsync(buffer, offset, count, _cancellationToken).GetAwaiter().GetResult();
+
+            public override async Task<int> ReadAsync(byte[] buffer, int offset, int count,
+                CancellationToken cancellationToken)
+            {
+                using (var cancellationTokenSource = CreateCancellationTokenSource(cancellationToken))
+                {
+                    return await _innerStream.ReadAsync(buffer, offset, count, cancellationTokenSource.Token)
+                        .ConfigureAwait(false);
+                }
+            }
+
+            public override long Seek(long offset, SeekOrigin origin) => _innerStream.Seek(offset, origin);
+
+            public override void SetLength(long value) => _innerStream.SetLength(value);
+
+            public override void Write(byte[] buffer, int offset, int count) =>
+                _innerStream.WriteAsync(buffer, offset, count, _cancellationToken).GetAwaiter().GetResult();
+
+            public override async Task WriteAsync(byte[] buffer, int offset, int count,
+                CancellationToken cancellationToken)
+            {
+                using (var cancellationTokenSource = CreateCancellationTokenSource(cancellationToken))
+                {
+                    await _innerStream.WriteAsync(buffer, offset, count, cancellationTokenSource.Token)
+                        .ConfigureAwait(false);
+                }
+            }
+
+            public override bool CanRead => _innerStream.CanRead;
+
+            public override bool CanSeek => _innerStream.CanSeek;
+
+            public override bool CanWrite => _innerStream.CanWrite;
+
+            public override long Length => _innerStream.Length;
+
+            public override long Position
+            {
+                get => _innerStream.Position;
+                set => _innerStream.Position = value;
+            }
+
+            protected override void Dispose(bool disposing)
+            {
+                if (disposing)
+                {
+                    _innerStream.Dispose();
+                }
+
+                base.Dispose(disposing);
+            }
+
+            private LinkedCancellationTokenSource CreateCancellationTokenSource(CancellationToken userCancellationToken)
+            {
+                return new LinkedCancellationTokenSource(_cancellationToken, userCancellationToken);
+            }
+
+            private readonly struct LinkedCancellationTokenSource : IDisposable
+            {
+                private readonly CancellationTokenSource _cancellationTokenSource;
+
+                public LinkedCancellationTokenSource(CancellationToken token1, CancellationToken token2)
+                {
+                    if (token1.CanBeCanceled && token2.CanBeCanceled)
+                    {
+                        _cancellationTokenSource = CancellationTokenSource.CreateLinkedTokenSource(token1, token2);
+                        Token = _cancellationTokenSource.Token;
+                    }
+                    else
+                    {
+                        _cancellationTokenSource = null;
+                        Token = token1.CanBeCanceled ? token1 : token2;
+                    }
+                }
+
+                public CancellationToken Token { get; }
+
+                public void Dispose()
+                {
+                    _cancellationTokenSource?.Dispose();
+                }
+            }
+        }
+
+        internal sealed class PeekableStreamReader : TextReader
+        {
+            private readonly Queue<string> _buffer;
+            private readonly StreamReader _inner;
+
+            public PeekableStreamReader(Stream stream)
+            {
+                _buffer = new Queue<string>();
+                _inner = new StreamReader(stream);
+            }
+
+            public override string ReadLine() => throw new NotImplementedException();
+
+            public override Task<string> ReadLineAsync()
+            {
+                if (_buffer.Count > 0)
+                {
+                    return Task.FromResult(_buffer.Dequeue());
+                }
+
+                return _inner.ReadLineAsync();
+            }
+
+            public async Task<string> PeekLineAsync()
+            {
+                var line = await ReadLineAsync().ConfigureAwait(false);
+                if (line == null)
+                {
+                    throw new EndOfStreamException();
+                }
+
+                _buffer.Enqueue(line);
+                return line;
+            }
+
+            public override int Read() => throw new NotImplementedException();
+
+            public override int Read(char[] buffer, int index, int count) => throw new NotImplementedException();
+
+            public override Task<int> ReadAsync(char[] buffer, int index, int count) =>
+                throw new NotImplementedException();
+
+            public override int ReadBlock(char[] buffer, int index, int count) => throw new NotImplementedException();
+
+            public override Task<int> ReadBlockAsync(char[] buffer, int index, int count) =>
+                throw new NotImplementedException();
+
+            public override string ReadToEnd() => throw new NotImplementedException();
+
+            public override Task<string> ReadToEndAsync() => throw new NotImplementedException();
+
+            protected override void Dispose(bool disposing)
+            {
+                if (disposing)
+                {
+                    _inner.Dispose();
+                }
+
+                base.Dispose(disposing);
+            }
+        }
+    }
+}

src/KubernetesClient/Watcher.cs

@@ -158,23 +158,12 @@ private async Task WatcherLoop(CancellationToken cancellationToken)
             Action<Exception> onError = null,
             [EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
-            Task<TR> AttachCancellationToken<TR>(Task<TR> task)
-            {
-                if (!task.IsCompleted)
-                {
-                    // here to pass cancellationToken into task
-                    return task.ContinueWith(t => t.GetAwaiter().GetResult(), cancellationToken);
-                }
-
-                return task;
-            }
-
-            using var streamReader = await AttachCancellationToken(streamReaderCreator()).ConfigureAwait(false);
+            using var streamReader = await streamReaderCreator().WaitAsync(cancellationToken).ConfigureAwait(false);
 
             for (; ; )
             {
                 // ReadLineAsync will return null when we've reached the end of the stream.
-                var line = await AttachCancellationToken(streamReader.ReadLineAsync()).ConfigureAwait(false);
+                var line = await streamReader.ReadLineAsync(cancellationToken).ConfigureAwait(false);
 
                 cancellationToken.ThrowIfCancellationRequested();
 

tests/KubernetesClient.Tests/WatchTests.cs

@@ -1028,5 +1028,70 @@ public async Task AsyncEnumerableWatchErrorHandling()
                 Assert.True(watchCompleted.IsSet);
             }
         }
+
+        [Fact]
+        public async Task CancellationDoesNotLeaveUnobservedTaskException()
+        {
+            // Regression test for https://github.com/kubernetes-client/csharp/issues/1813
+            // When the cancellation token is cancelled while a read is in flight and the
+            // underlying task subsequently faults (e.g. transport-level IOException after the
+            // connection is torn down), the faulting task must be observed so that no
+            // TaskScheduler.UnobservedTaskException is raised when it is finalized.
+            var unobservedExceptions = new List<Exception>();
+            void Handler(object sender, UnobservedTaskExceptionEventArgs e)
+            {
+                unobservedExceptions.Add(e.Exception);
+            }
+
+            TaskScheduler.UnobservedTaskException += Handler;
+            try
+            {
+                // Run the cancellation scenario in a separate, non-inlined method so that all
+                // references to the orphaned task go out of scope before we force a collection.
+                await RunCancelledWatchAsync().ConfigureAwait(true);
+
+                // Force the orphaned task to be finalized; without observing its exception this
+                // would raise TaskScheduler.UnobservedTaskException.
+                for (var i = 0; i < 5; i++)
+                {
+                    GC.Collect();
+                    GC.WaitForPendingFinalizers();
+                    await Task.Delay(50).ConfigureAwait(true);
+                }
+
+                Assert.Empty(unobservedExceptions);
+            }
+            finally
+            {
+                TaskScheduler.UnobservedTaskException -= Handler;
+            }
+        }
+
+        [System.Runtime.CompilerServices.MethodImpl(System.Runtime.CompilerServices.MethodImplOptions.NoInlining)]
+        private static async Task RunCancelledWatchAsync()
+        {
+            using var cts = new CancellationTokenSource();
+            var faultReader = new TaskCompletionSource<TextReader>(TaskCreationOptions.RunContinuationsAsynchronously);
+
+            Func<Task<TextReader>> streamReaderCreator = () => faultReader.Task;
+
+            var enumerator = Watcher<V1Pod>
+                .CreateWatchEventEnumerator(streamReaderCreator, onError: null, cancellationToken: cts.Token)
+                .GetAsyncEnumerator(cts.Token);
+
+            // Start the enumeration; this awaits the (not yet completed) creator task.
+            var moveNext = enumerator.MoveNextAsync();
+
+            // Cancel before the creator task completes. The cancellation-aware continuation
+            // is cancelled, but the original creator task is still pending.
+            cts.Cancel();
+
+            await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await moveNext.ConfigureAwait(true)).ConfigureAwait(true);
+
+            await enumerator.DisposeAsync().ConfigureAwait(true);
+
+            // Now fault the original creator task, mimicking the transport tear-down.
+            faultReader.SetException(new IOException("The request was aborted."));
+        }
     }
 }