diff --git a/src/KubernetesClient.Classic/CancellationExtensions.cs b/src/KubernetesClient.Classic/CancellationExtensions.cs
new file mode 100644
index 000000000..d7f60cca6
--- /dev/null
+++ b/src/KubernetesClient.Classic/CancellationExtensions.cs
@@ -0,0 +1,56 @@
+namespace k8s
+{
+ ///
+ /// Polyfills for the cancellable overloads which are only available on modern .NET,
+ /// so the shared sources can rely on them unconditionally.
+ ///
+ internal static class CancellationExtensions
+ {
+ ///
+ /// Polyfill of Task<TResult>.WaitAsync(CancellationToken).
+ ///
+ public static Task WaitAsync(this Task task, CancellationToken cancellationToken)
+ {
+ if (task == null)
+ {
+ throw new ArgumentNullException(nameof(task));
+ }
+
+ if (task.IsCompleted || !cancellationToken.CanBeCanceled)
+ {
+ return task;
+ }
+
+ if (cancellationToken.IsCancellationRequested)
+ {
+ return Task.FromCanceled(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);
+ }
+
+ ///
+ /// Polyfill of TextReader.ReadLineAsync(CancellationToken).
+ ///
+ public static Task ReadLineAsync(this TextReader reader, CancellationToken cancellationToken)
+ {
+ if (reader == null)
+ {
+ throw new ArgumentNullException(nameof(reader));
+ }
+
+ return reader.ReadLineAsync().WaitAsync(cancellationToken);
+ }
+ }
+}
diff --git a/src/KubernetesClient.Classic/KubernetesClient.Classic.csproj b/src/KubernetesClient.Classic/KubernetesClient.Classic.csproj
index 902dc41dd..8980f5124 100644
--- a/src/KubernetesClient.Classic/KubernetesClient.Classic.csproj
+++ b/src/KubernetesClient.Classic/KubernetesClient.Classic.csproj
@@ -107,7 +107,6 @@
-
diff --git a/src/KubernetesClient.Classic/LineSeparatedHttpContent.cs b/src/KubernetesClient.Classic/LineSeparatedHttpContent.cs
new file mode 100644
index 000000000..9206fe4ff
--- /dev/null
+++ b/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 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 _buffer;
+ private readonly StreamReader _inner;
+
+ public PeekableStreamReader(Stream stream)
+ {
+ _buffer = new Queue();
+ _inner = new StreamReader(stream);
+ }
+
+ public override string ReadLine() => throw new NotImplementedException();
+
+ public override Task ReadLineAsync()
+ {
+ if (_buffer.Count > 0)
+ {
+ return Task.FromResult(_buffer.Dequeue());
+ }
+
+ return _inner.ReadLineAsync();
+ }
+
+ public async Task 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 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 ReadBlockAsync(char[] buffer, int index, int count) =>
+ throw new NotImplementedException();
+
+ public override string ReadToEnd() => throw new NotImplementedException();
+
+ public override Task ReadToEndAsync() => throw new NotImplementedException();
+
+ protected override void Dispose(bool disposing)
+ {
+ if (disposing)
+ {
+ _inner.Dispose();
+ }
+
+ base.Dispose(disposing);
+ }
+ }
+ }
+}
diff --git a/src/KubernetesClient/Watcher.cs b/src/KubernetesClient/Watcher.cs
index 23868d4e0..1008c640a 100644
--- a/src/KubernetesClient/Watcher.cs
+++ b/src/KubernetesClient/Watcher.cs
@@ -158,23 +158,12 @@ private async Task WatcherLoop(CancellationToken cancellationToken)
Action onError = null,
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
- Task AttachCancellationToken
(Task
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();
diff --git a/tests/KubernetesClient.Tests/WatchTests.cs b/tests/KubernetesClient.Tests/WatchTests.cs
index 53259d770..fe314b3c6 100644
--- a/tests/KubernetesClient.Tests/WatchTests.cs
+++ b/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();
+ 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(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ Func> streamReaderCreator = () => faultReader.Task;
+
+ var enumerator = Watcher
+ .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(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."));
+ }
}
}