From e395e8f9f9c23509a864e605df2455f113dcba76 Mon Sep 17 00:00:00 2001 From: Boshi LIAN Date: Wed, 1 Jul 2026 14:35:35 -0700 Subject: [PATCH 1/4] Add cancellation handling to prevent UnobservedTaskException in Watcher --- src/KubernetesClient/Watcher.cs | 14 +++++ tests/KubernetesClient.Tests/WatchTests.cs | 65 ++++++++++++++++++++++ 2 files changed, 79 insertions(+) diff --git a/src/KubernetesClient/Watcher.cs b/src/KubernetesClient/Watcher.cs index 23868d4e0..c475908a6 100644 --- a/src/KubernetesClient/Watcher.cs +++ b/src/KubernetesClient/Watcher.cs @@ -162,6 +162,16 @@ Task AttachCancellationToken(Task task) { if (!task.IsCompleted) { + // 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); } @@ -174,7 +184,11 @@ Task AttachCancellationToken(Task task) for (; ; ) { // ReadLineAsync will return null when we've reached the end of the stream. +#if NET7_0_OR_GREATER + var line = await streamReader.ReadLineAsync(cancellationToken).ConfigureAwait(false); +#else var line = await AttachCancellationToken(streamReader.ReadLineAsync()).ConfigureAwait(false); +#endif 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.")); + } } } From 799523b3b5546e12d8ac669cfd683789f1c8cdc4 Mon Sep 17 00:00:00 2001 From: Boshi LIAN Date: Sat, 25 Jul 2026 01:50:41 -0700 Subject: [PATCH 2/4] Refactor LineSeparatedHttpContent and add TextReaderExtensions for async reading with cancellation support --- .../KubernetesClient.Classic.csproj | 1 - .../LineSeparatedHttpContent.cs | 211 ++++++++++++++++++ .../TextReaderExtensions.cs | 42 ++++ .../LineSeparatedHttpContent.cs | 15 ++ src/KubernetesClient/Watcher.cs | 4 - 5 files changed, 268 insertions(+), 5 deletions(-) create mode 100644 src/KubernetesClient.Classic/LineSeparatedHttpContent.cs create mode 100644 src/KubernetesClient.Classic/TextReaderExtensions.cs 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.Classic/TextReaderExtensions.cs b/src/KubernetesClient.Classic/TextReaderExtensions.cs new file mode 100644 index 000000000..f6539b144 --- /dev/null +++ b/src/KubernetesClient.Classic/TextReaderExtensions.cs @@ -0,0 +1,42 @@ +namespace k8s +{ + /// + /// Provides the ReadLineAsync(CancellationToken) overload which is only available + /// on .NET 7.0 or greater, so shared sources can rely on it unconditionally. + /// + internal static class TextReaderExtensions + { + public static Task ReadLineAsync(this TextReader reader, CancellationToken cancellationToken) + { + if (reader == null) + { + throw new ArgumentNullException(nameof(reader)); + } + + if (cancellationToken.IsCancellationRequested) + { + return Task.FromCanceled(cancellationToken); + } + + var task = reader.ReadLineAsync(); + + if (task.IsCompleted || !cancellationToken.CanBeCanceled) + { + return task; + } + + // 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); + } + } +} diff --git a/src/KubernetesClient/LineSeparatedHttpContent.cs b/src/KubernetesClient/LineSeparatedHttpContent.cs index 9206fe4ff..89edb6287 100644 --- a/src/KubernetesClient/LineSeparatedHttpContent.cs +++ b/src/KubernetesClient/LineSeparatedHttpContent.cs @@ -169,6 +169,21 @@ public override Task ReadLineAsync() return _inner.ReadLineAsync(); } + /// + /// The base implementation of this overload does not route + /// through , which would both bypass the peeked line + /// buffer and hit the unsupported character based read members. + /// + public override ValueTask ReadLineAsync(CancellationToken cancellationToken) + { + if (_buffer.Count > 0) + { + return new ValueTask(_buffer.Dequeue()); + } + + return _inner.ReadLineAsync(cancellationToken); + } + public async Task PeekLineAsync() { var line = await ReadLineAsync().ConfigureAwait(false); diff --git a/src/KubernetesClient/Watcher.cs b/src/KubernetesClient/Watcher.cs index c475908a6..217d354c1 100644 --- a/src/KubernetesClient/Watcher.cs +++ b/src/KubernetesClient/Watcher.cs @@ -184,11 +184,7 @@ Task AttachCancellationToken(Task task) for (; ; ) { // ReadLineAsync will return null when we've reached the end of the stream. -#if NET7_0_OR_GREATER var line = await streamReader.ReadLineAsync(cancellationToken).ConfigureAwait(false); -#else - var line = await AttachCancellationToken(streamReader.ReadLineAsync()).ConfigureAwait(false); -#endif cancellationToken.ThrowIfCancellationRequested(); From d0e21c5fb2298c2edfb4e38ce2c7ce35b277ef87 Mon Sep 17 00:00:00 2001 From: Boshi LIAN Date: Sat, 25 Jul 2026 02:01:45 -0700 Subject: [PATCH 3/4] Add CancellationExtensions for async task handling with cancellation support --- ...xtensions.cs => CancellationExtensions.cs} | 38 +++++++++++++------ src/KubernetesClient/Watcher.cs | 23 +---------- 2 files changed, 27 insertions(+), 34 deletions(-) rename src/KubernetesClient.Classic/{TextReaderExtensions.cs => CancellationExtensions.cs} (59%) diff --git a/src/KubernetesClient.Classic/TextReaderExtensions.cs b/src/KubernetesClient.Classic/CancellationExtensions.cs similarity index 59% rename from src/KubernetesClient.Classic/TextReaderExtensions.cs rename to src/KubernetesClient.Classic/CancellationExtensions.cs index f6539b144..d7f60cca6 100644 --- a/src/KubernetesClient.Classic/TextReaderExtensions.cs +++ b/src/KubernetesClient.Classic/CancellationExtensions.cs @@ -1,28 +1,29 @@ namespace k8s { /// - /// Provides the ReadLineAsync(CancellationToken) overload which is only available - /// on .NET 7.0 or greater, so shared sources can rely on it unconditionally. + /// Polyfills for the cancellable overloads which are only available on modern .NET, + /// so the shared sources can rely on them unconditionally. /// - internal static class TextReaderExtensions + internal static class CancellationExtensions { - public static Task ReadLineAsync(this TextReader reader, CancellationToken cancellationToken) + /// + /// Polyfill of Task<TResult>.WaitAsync(CancellationToken). + /// + public static Task WaitAsync(this Task task, CancellationToken cancellationToken) { - if (reader == null) + if (task == null) { - throw new ArgumentNullException(nameof(reader)); + throw new ArgumentNullException(nameof(task)); } - if (cancellationToken.IsCancellationRequested) + if (task.IsCompleted || !cancellationToken.CanBeCanceled) { - return Task.FromCanceled(cancellationToken); + return task; } - var task = reader.ReadLineAsync(); - - if (task.IsCompleted || !cancellationToken.CanBeCanceled) + if (cancellationToken.IsCancellationRequested) { - return task; + return Task.FromCanceled(cancellationToken); } // Observe any exception from the original task to prevent an @@ -38,5 +39,18 @@ public static Task ReadLineAsync(this TextReader reader, CancellationTok // 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/Watcher.cs b/src/KubernetesClient/Watcher.cs index 217d354c1..1008c640a 100644 --- a/src/KubernetesClient/Watcher.cs +++ b/src/KubernetesClient/Watcher.cs @@ -158,28 +158,7 @@ private async Task WatcherLoop(CancellationToken cancellationToken) Action onError = null, [EnumeratorCancellation] CancellationToken cancellationToken = default) { - Task AttachCancellationToken(Task task) - { - if (!task.IsCompleted) - { - // 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); - } - - return task; - } - - using var streamReader = await AttachCancellationToken(streamReaderCreator()).ConfigureAwait(false); + using var streamReader = await streamReaderCreator().WaitAsync(cancellationToken).ConfigureAwait(false); for (; ; ) { From 5c95331503a0484001e46537ae103a3a7def83bf Mon Sep 17 00:00:00 2001 From: Boshi LIAN Date: Sat, 25 Jul 2026 02:32:44 -0700 Subject: [PATCH 4/4] Remove redundant ReadLineAsync overload in PeekableStreamReader --- src/KubernetesClient/LineSeparatedHttpContent.cs | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/src/KubernetesClient/LineSeparatedHttpContent.cs b/src/KubernetesClient/LineSeparatedHttpContent.cs index 89edb6287..9206fe4ff 100644 --- a/src/KubernetesClient/LineSeparatedHttpContent.cs +++ b/src/KubernetesClient/LineSeparatedHttpContent.cs @@ -169,21 +169,6 @@ public override Task ReadLineAsync() return _inner.ReadLineAsync(); } - /// - /// The base implementation of this overload does not route - /// through , which would both bypass the peeked line - /// buffer and hit the unsupported character based read members. - /// - public override ValueTask ReadLineAsync(CancellationToken cancellationToken) - { - if (_buffer.Count > 0) - { - return new ValueTask(_buffer.Dequeue()); - } - - return _inner.ReadLineAsync(cancellationToken); - } - public async Task PeekLineAsync() { var line = await ReadLineAsync().ConfigureAwait(false);