From b291e337708d4d6f2a09fbe6ddd7b9dd9020f010 Mon Sep 17 00:00:00 2001 From: David Negstad Date: Tue, 13 Oct 2026 12:36:26 -0700 Subject: [PATCH 1/2] Keep DCP watches draining and close abandoned connections Own watch responses before reading events and close them on cancellation or periodic restart. Drain streams independently of resource handlers while preserving event order and existing lifecycle dispatch. Handle failed Status payloads and retry interrupted or transiently failed watches. Add behavioral watch regressions over HTTPS/HTTP2 with a shared test certificate and retain the five-minute reconnection safeguard. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: a4d11434-2c0b-4741-91ea-9eacc92ca29a --- src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs | 225 +++++ src/Aspire.Hosting/Dcp/DcpPipelineBuilder.cs | 13 +- src/Aspire.Hosting/Dcp/KubernetesService.cs | 84 +- .../Dcp/DcpApiServerCertificateFixture.cs | 26 + .../Dcp/KubernetesServiceTests.cs | 851 +++++++++++++++++- 5 files changed, 1124 insertions(+), 75 deletions(-) create mode 100644 tests/Aspire.Hosting.Tests/Dcp/DcpApiServerCertificateFixture.cs diff --git a/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs b/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs index 86849cb0194..452a052bab2 100644 --- a/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs +++ b/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs @@ -1,11 +1,16 @@ // Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT license. +using System.Buffers; +using System.IO.Pipelines; using System.Net; using System.Net.Http.Headers; +using System.Text; +using System.Text.Json; using Aspire.Hosting.Dcp.Model; using k8s; using k8s.Autorest; +using k8s.Models; namespace Aspire.Hosting.Dcp; @@ -24,6 +29,67 @@ public DcpKubernetesClient(KubernetesClientConfiguration config, params Delegati HttpClientTimeout = Timeout.InfiniteTimeSpan; } + /// + /// Opens a DCP resource watch without waiting for its first event. + /// This avoids KubernetesClient's setup-cancellation path that can leave an unread response open. + /// + /// The resource type returned by watch events. + /// The Kubernetes API group. + /// The resource API version. + /// The plural resource name, e.g. "executables". + /// The namespace, or null/empty for a cluster-scoped watch. + /// Closes the watch response on cancellation, even before enumeration starts. + /// Cancels connection establishment only; does not affect the returned stream. + /// + /// A single-use event stream, returned once response headers arrive. + /// Ending enumeration closes its HTTP response. + /// + public async Task> CreateWatchAsync( + string group, + string version, + string plural, + string? namespaceParameter, + CancellationToken connectionCancellationToken, + CancellationToken cancellationToken) + { + var collectionPath = string.IsNullOrEmpty(namespaceParameter) + ? $"apis/{group}/{version}/{plural}" + : $"apis/{group}/{version}/namespaces/{Uri.EscapeDataString(namespaceParameter)}/{plural}"; + using var request = new HttpRequestMessage(HttpMethod.Get, new Uri(BaseUri, $"{collectionPath}?watch=true")); + request.Version = HttpVersion.Version20; + if (Credentials is not null) + { + await Credentials.ProcessHttpRequestAsync(request, cancellationToken).ConfigureAwait(false); + } + + // KubernetesClient's list-and-watch adapter peeks the first event during setup and + // can abandon its reader when that wait is canceled. Own the raw response instead. + // https://github.com/microsoft/dcp/issues/286 + var response = await HttpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false); + try + { + cancellationToken.ThrowIfCancellationRequested(); + if (!response.IsSuccessStatusCode) + { + var content = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false); + throw new HttpOperationException($"Operation returned an invalid status code '{response.StatusCode}', response body {content}") + { + Request = new HttpRequestMessageWrapper(request, string.Empty), + Response = new HttpResponseMessageWrapper(response, content) + }; + } + + var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); + cancellationToken.ThrowIfCancellationRequested(); + return new WatchConnection(response, stream, connectionCancellationToken); + } + catch + { + response.Dispose(); + throw; + } + } + /// /// Asynchronously reads a sub-resource from a Kubernetes resource as a stream. /// @@ -140,6 +206,165 @@ public async Task PatchExecutionDocumentAsync( } } + private sealed class WatchConnection : IAsyncEnumerable<(WatchEventType, T)> + { + private readonly HttpResponseMessage _response; + private readonly PipeReader _reader; + private readonly CancellationToken _connectionToken; + private readonly CancellationTokenRegistration _cancellationRegistration; + private bool _firstLine = true; + + public WatchConnection(HttpResponseMessage response, Stream stream, CancellationToken cancellationToken) + { + _response = response; + _reader = PipeReader.Create(stream); + _connectionToken = cancellationToken; + // A canceled body read does not necessarily reset an HTTP/2 stream. Close the + // response even when enumeration is paused in the consumer or has not started. + _cancellationRegistration = cancellationToken.Register(static state => ((HttpResponseMessage)state!).Dispose(), response); + } + + public async IAsyncEnumerator<(WatchEventType, T)> GetAsyncEnumerator(CancellationToken cancellationToken = default) + { + using var response = _response; + using var cancellationRegistration = _cancellationRegistration; + using var consumerCancellation = cancellationToken.Register(static state => ((HttpResponseMessage)state!).Dispose(), response); + using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(_connectionToken, cancellationToken); + var receivedEvent = false; + + try + { + while (true) + { + string? line; + bool terminated; + try + { + (line, terminated) = await ReadEventLineAsync(linkedCts.Token).ConfigureAwait(false); + } + catch (Exception) when (linkedCts.IsCancellationRequested) + { + // Disposing the response can surface as IOException or ObjectDisposedException. + // Preserve the original token so the periodic-restart helper recognizes its deadline. + _connectionToken.ThrowIfCancellationRequested(); + cancellationToken.ThrowIfCancellationRequested(); + throw; + } + + _connectionToken.ThrowIfCancellationRequested(); + cancellationToken.ThrowIfCancellationRequested(); + if (line is null) + { + if (!receivedEvent) + { + // An empty response is a failed watch, not a successful short poll. + // Surface it to the retry pipeline rather than reconnecting in a tight loop. + throw new EndOfStreamException("DCP watch ended before returning an event."); + } + + yield break; + } + + if (DeserializeEvent(line, terminated) is { } watchEvent) + { + receivedEvent = true; + yield return watchEvent; + } + } + } + finally + { + await _reader.CompleteAsync().ConfigureAwait(false); + } + } + + private async ValueTask<(string? Line, bool Terminated)> ReadEventLineAsync(CancellationToken cancellationToken) + { + SequencePosition? searchedTo = null; + while (true) + { + var result = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false); + var buffer = result.Buffer; + var unsearched = searchedTo is { } position ? buffer.Slice(position) : buffer; + if (unsearched.PositionOf((byte)'\n') is { } newline) + { + var line = DecodeLine(buffer.Slice(0, newline)); + _reader.AdvanceTo(buffer.GetPosition(1, newline)); + + return (line, true); + } + if (result.IsCompleted) + { + var line = buffer.IsEmpty ? null : DecodeLine(buffer); + _reader.AdvanceTo(buffer.End); + + return (line, false); + } + + // No bytes are consumed until a complete record is found, so this position + // stays valid across reads and avoids rescanning the retained prefix. + searchedTo = buffer.IsEmpty ? null : buffer.End; + _reader.AdvanceTo(buffer.Start, buffer.End); + } + } + + private string DecodeLine(in ReadOnlySequence buffer) + { + var line = Encoding.UTF8.GetString(buffer); + if (_firstLine) + { + _firstLine = false; + if (line.Length > 0 && line[0] == '\uFEFF') + { + line = line[1..]; + } + } + + return line; + } + + private static (WatchEventType, T)? DeserializeEvent(string line, bool terminated) + { + // Kubernetes watches contain newline-delimited events: + // {"type":"MODIFIED","object":{"kind":"Executable","metadata":{...},...}} + // An ERROR can contain a Status instead; BOOKMARK objects contain only metadata. + // https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes + try + { + var envelope = KubernetesJson.Deserialize.WatchEvent>(line); + if (envelope?.Object is null) + { + throw new JsonException("DCP watch event does not contain an object."); + } + if (envelope.Object.Kind == "Status") + { + var status = KubernetesJson.Deserialize.WatchEvent>(line).Object; + if (status.Status == "Failure") + { + throw new KubernetesException(status); + } + if (status.Status != "Success") + { + throw new JsonException("DCP watch Status does not report Success or Failure."); + } + + // A successful Status is not a resource update and must not be deserialized as T. + return null; + } + + var watchEvent = KubernetesJson.Deserialize.WatchEvent>(line); + + return (watchEvent.Type, watchEvent.Object); + } + catch (JsonException ex) when (!terminated) + { + // EOF can leave a partial record such as {"type":"MODIFIED","object": + // Retry that interrupted stream, while complete malformed records remain errors. + throw new EndOfStreamException("DCP watch ended in an incomplete event.", ex); + } + } + } + private sealed class QueryBuilder { private readonly List _parameters = new List(); diff --git a/src/Aspire.Hosting/Dcp/DcpPipelineBuilder.cs b/src/Aspire.Hosting/Dcp/DcpPipelineBuilder.cs index e6939cc18c4..3471313bfd1 100644 --- a/src/Aspire.Hosting/Dcp/DcpPipelineBuilder.cs +++ b/src/Aspire.Hosting/Dcp/DcpPipelineBuilder.cs @@ -1,6 +1,8 @@ // Licensed to the .NET Foundation under one or more agreements. // The .NET Foundation licenses this file to you under the MIT license. +using System.Net; +using k8s; using Microsoft.Extensions.Logging; using Polly; using Polly.Retry; @@ -65,7 +67,16 @@ public static ResiliencePipeline BuildWatchResourcePipeline(ILogger logger) { var retryUntilCancelled = new RetryStrategyOptions() { - ShouldHandle = new PredicateBuilder().HandleInner(), + ShouldHandle = new PredicateBuilder() + .HandleInner() + .HandleInner(ex => ex.Status.Code is + (int)HttpStatusCode.Gone or + (int)HttpStatusCode.RequestTimeout or + (int)HttpStatusCode.TooManyRequests or + (int)HttpStatusCode.InternalServerError or + (int)HttpStatusCode.BadGateway or + (int)HttpStatusCode.ServiceUnavailable or + (int)HttpStatusCode.GatewayTimeout), BackoffType = DelayBackoffType.Exponential, MaxRetryAttempts = int.MaxValue, UseJitter = true, diff --git a/src/Aspire.Hosting/Dcp/KubernetesService.cs b/src/Aspire.Hosting/Dcp/KubernetesService.cs index 4978c4a582a..cce02e12874 100644 --- a/src/Aspire.Hosting/Dcp/KubernetesService.cs +++ b/src/Aspire.Hosting/Dcp/KubernetesService.cs @@ -7,6 +7,7 @@ using System.Net.Http.Headers; using System.Runtime.CompilerServices; using System.Text.Json; +using System.Threading.Channels; using Aspire.Hosting.Diagnostics; using Aspire.Hosting.Dcp.Model; using Aspire.Hosting.Utils; @@ -95,6 +96,8 @@ internal sealed class KubernetesService(ILogger logger, IOpti private ResiliencePipeline? _resiliencePipeline; private bool _disposed; + internal TimeSpan WatchRestartInterval { get; set; } = TimeSpan.FromMinutes(5); + public Task GetAsync(string name, string? namespaceParameter = null, CancellationToken cancellationToken = default) where T : CustomResource, IKubernetesStaticMetadata { @@ -302,8 +305,14 @@ public Task DeleteAsync(string name, string? namespaceParameter = null, Ca ObjectDisposedException.ThrowIf(_disposed, this); var resourceType = GetResourceFor(); - // WatchAsync can become unresponsive if running long enough - // We use a helper to periodically restart the inner watch enumerable + using var watchCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + var channel = Channel.CreateUnbounded<(WatchEventType, T)>(new UnboundedChannelOptions + { + SingleReader = true, + SingleWriter = true, + AllowSynchronousContinuations = false + }); + var innerWatchFactory = ((WatchEventType, T)? lastValue, CancellationToken restartCancellationToken) => { return ExecuteWithRetry( @@ -311,39 +320,58 @@ public Task DeleteAsync(string name, string? namespaceParameter = null, Ca T.ObjectKind, async (kubernetes, operationCancellationToken) => { - var responseTask = string.IsNullOrEmpty(namespaceParameter) - ? kubernetes.CustomObjects.ListClusterCustomObjectWithHttpMessagesAsync( - GroupVersion.Group, - GroupVersion.Version, - resourceType, - watch: true, - cancellationToken: operationCancellationToken) - : kubernetes.CustomObjects.ListNamespacedCustomObjectWithHttpMessagesAsync( - GroupVersion.Group, - GroupVersion.Version, - namespaceParameter, - resourceType, - watch: true, - cancellationToken: operationCancellationToken); - - // KubernetesClient's lazy WatchAsync does not await the HTTP response until enumeration starts. - await responseTask.ConfigureAwait(false); - - // TODO: KubernetesClient v18 marked WatchAsync extension method as obsolete. - // The new pattern uses Watcher directly, but requires significant refactoring. - // This API still works in v18.x and will be updated in a future change. -#pragma warning disable CS0618 // Type or member is obsolete - return responseTask.WatchAsync(onError: null, restartCancellationToken); -#pragma warning restore CS0618 // Type or member is obsolete + // The per-attempt budget covers response headers, not the lifetime of the + // stream. Its owner uses the restart token after the attempt has completed. + return await kubernetes.CreateWatchAsync( + GroupVersion.Group, GroupVersion.Version, resourceType, + namespaceParameter, restartCancellationToken, + operationCancellationToken).ConfigureAwait(false); }, RetryOnConnectivityAndConflictErrors, restartCancellationToken, timeoutEachAttempt: true); }; - await foreach (var item in PeriodicRestartAsyncEnumerable.CreateAsync(innerWatchFactory, restartInterval: TimeSpan.FromMinutes(5), cancellationToken: cancellationToken).ConfigureAwait(false)) + // Drain DCP independently of resource handlers. Waiting on the resource watcher's output + // semaphore, lifecycle callbacks, or synchronous log subscribers must not leave HTTP/2 + // flow control blocking DCP mutations. Keep every event, including intermediate states. + var watchTask = Task.Run(async () => + { + Exception? error = null; + try + { + await foreach (var item in PeriodicRestartAsyncEnumerable.CreateAsync( + innerWatchFactory, WatchRestartInterval, watchCts.Token).ConfigureAwait(false)) + { + channel.Writer.TryWrite(item); + } + } + catch (OperationCanceledException) when (watchCts.IsCancellationRequested) + { + // The consumer canceled or disposed its enumeration. + } + catch (Exception ex) + { + error = ex; + } + finally + { + channel.Writer.TryComplete(error); + } + }, CancellationToken.None); + + try + { + await foreach (var item in channel.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + cancellationToken.ThrowIfCancellationRequested(); + yield return item; + } + } + finally { - yield return item; + await watchCts.CancelAsync().ConfigureAwait(false); + await watchTask.ConfigureAwait(false); } } diff --git a/tests/Aspire.Hosting.Tests/Dcp/DcpApiServerCertificateFixture.cs b/tests/Aspire.Hosting.Tests/Dcp/DcpApiServerCertificateFixture.cs new file mode 100644 index 00000000000..4f65f079a50 --- /dev/null +++ b/tests/Aspire.Hosting.Tests/Dcp/DcpApiServerCertificateFixture.cs @@ -0,0 +1,26 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. + +using System.Security.Cryptography; +using System.Security.Cryptography.X509Certificates; + +namespace Aspire.Hosting.Tests.Dcp; + +public sealed class DcpApiServerCertificateFixture : IDisposable +{ + internal X509Certificate2 Certificate { get; } = CreateCertificate(); + + public void Dispose() + { + Certificate.Dispose(); + } + + private static X509Certificate2 CreateCertificate() + { + using var rsa = RSA.Create(2048); + var request = new CertificateRequest("CN=localhost", rsa, HashAlgorithmName.SHA256, RSASignaturePadding.Pkcs1); + using var certificate = request.CreateSelfSigned(DateTimeOffset.UtcNow.AddDays(-1), DateTimeOffset.UtcNow.AddDays(1)); + // Schannel requires an imported key rather than the ephemeral key from CreateSelfSigned. + return X509CertificateLoader.LoadPkcs12(certificate.Export(X509ContentType.Pfx), password: null); + } +} diff --git a/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs b/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs index 52755b909c2..fcd7390f65e 100644 --- a/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs +++ b/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs @@ -4,6 +4,10 @@ using System.Collections.Concurrent; using System.Globalization; using System.Net; +using System.Security.Cryptography.X509Certificates; +using System.Text; +using System.Text.Json; +using System.Threading.Channels; using Aspire.Hosting.Dcp; using Aspire.Hosting.Dcp.Model; using k8s; @@ -12,6 +16,8 @@ using Microsoft.AspNetCore.Builder; using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.InternalTesting; +using Microsoft.AspNetCore.Server.Kestrel.Core; using Microsoft.Extensions.Configuration; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; @@ -20,8 +26,744 @@ namespace Aspire.Hosting.Tests.Dcp; -public class KubernetesServiceTests +public class KubernetesServiceTests(DcpApiServerCertificateFixture certificateFixture) : IClassFixture { + [Theory] + [InlineData(null)] + [InlineData("test-namespace")] + public async Task WatchAsync_CancellationBeforeFirstEventClosesHttp2Stream(string? namespaceParameter) + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + using var watchCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var responseStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var requests = new ConcurrentQueue<(string Protocol, string Path, string Authorization)>(); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + requests.Enqueue((context.Request.Protocol, context.Request.Path, context.Request.Headers.Authorization.ToString())); + context.Response.ContentType = "application/json"; + await context.Response.StartAsync(context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + responseStarted.TrySetResult(); + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + } + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + + await using var enumerator = service.WatchAsync(namespaceParameter, watchCts.Token).GetAsyncEnumerator(); + var moveNext = enumerator.MoveNextAsync().AsTask(); + try + { + await responseStarted.Task.DefaultTimeout(); + Assert.False(moveNext.IsCompleted); + + watchCts.Cancel(); + await Assert.ThrowsAnyAsync(() => moveNext).DefaultTimeout(); + await requestCanceled.Task.DefaultTimeout(); + + var collectionPath = namespaceParameter is null + ? "/apis/usvc-dev.developer.microsoft.com/v1/containers" + : $"/apis/usvc-dev.developer.microsoft.com/v1/namespaces/{namespaceParameter}/containers"; + Assert.Equal([("HTTP/2", collectionPath, "Bearer dcp-test-token")], requests.ToArray()); + } + finally + { + watchCts.Cancel(); + try + { + await moveNext.DefaultTimeout(); + } + catch (OperationCanceledException) when (watchCts.IsCancellationRequested) + { + // Expected when closing a pending watch read. + } + } + } + + [Theory] + [InlineData(false, false, false)] + [InlineData(true, false, false)] + [InlineData(true, true, false)] + [InlineData(true, true, true)] + public async Task WatchAsync_RestartsAndClosesHttp2StreamWithoutConsumerProgress(bool sendHeaders, bool sendFirstEvent, bool blockFirstRequest) + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + service.WatchRestartInterval = TimeSpan.FromSeconds(1); + var requestsStarted = Channel.CreateUnbounded(); + var requestCancellations = new ConcurrentDictionary(StringComparer.Ordinal); + var requestCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + var requestNumber = Interlocked.Increment(ref requestCount); + var resourceName = $"frontend-{requestNumber}"; + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + requestCancellations[resourceName] = requestCanceled; + requestsStarted.Writer.TryWrite(resourceName); + Assert.Equal("HTTP/2", context.Request.Protocol); + try + { + if (blockFirstRequest && requestNumber == 1) + { + // Force setup to expire so the first observed event belongs to a later generation. + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + context.Response.ContentType = "application/json"; + if (sendFirstEvent) + { + await WriteWatchEventAsync(context, Container.Create(resourceName, "test-image")); + } + else if (sendHeaders) + { + await context.Response.StartAsync(context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + } + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + } + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + + await using var enumerator = service.WatchAsync(cancellationToken: cts.Token).GetAsyncEnumerator(); + var moveNext = enumerator.MoveNextAsync().AsTask(); + try + { + string watchedResourceName; + if (sendFirstEvent) + { + Assert.True(await moveNext.DefaultTimeout()); + watchedResourceName = enumerator.Current.Item2.Metadata.Name; + string startedResourceName; + do + { + startedResourceName = await requestsStarted.Reader.ReadAsync(cts.Token).DefaultTimeout(); + } + while (startedResourceName != watchedResourceName); + } + else + { + watchedResourceName = await requestsStarted.Reader.ReadAsync(cts.Token).DefaultTimeout(); + } + if (blockFirstRequest) + { + await requestCancellations["frontend-1"].Task.DefaultTimeout(); + Assert.NotEqual("frontend-1", watchedResourceName); + } + + await requestCancellations[watchedResourceName].Task.DefaultTimeout(); + var nextResourceName = await requestsStarted.Reader.ReadAsync(cts.Token).DefaultTimeout(); + Assert.NotEqual(watchedResourceName, nextResourceName); + if (!sendFirstEvent) + { + Assert.False(moveNext.IsCompleted); + } + } + finally + { + if (!moveNext.IsCompleted) + { + cts.Cancel(); + } + try + { + await moveNext.DefaultTimeout(); + } + catch (OperationCanceledException) when (cts.IsCancellationRequested) + { + // Expected when closing a pending watch read. + } + } + } + + [Fact] + public async Task WatchAsync_IdleHttp2StreamOutlivesConnectionDeadlineAndOtherApiTimeouts() + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + using var watchCts = CancellationTokenSource.CreateLinkedTokenSource(cts.Token); + var (service, kubeconfigPath, fileSystem) = CreateService( + kubernetesApiTimeout: TimeSpan.FromMilliseconds(500), + kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(4.5)); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var responseStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + if (context.Request.Query["watch"] == "true") + { + context.Response.ContentType = "application/json"; + await context.Response.StartAsync(context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + responseStarted.TrySetResult(); + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + } + } + else + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + await using var enumerator = service.WatchAsync(cancellationToken: watchCts.Token).GetAsyncEnumerator(); + var moveNext = enumerator.MoveNextAsync().AsTask(); + try + { + await responseStarted.Task.DefaultTimeout(); + await Assert.ThrowsAsync(() => service.ListAsync(cancellationToken: cts.Token)) + .DefaultTimeout(TestConstants.LongTimeoutDuration); + Assert.False(moveNext.IsCompleted); + Assert.False(requestCanceled.Task.IsCompleted); + watchCts.Cancel(); + await Assert.ThrowsAnyAsync(() => moveNext).DefaultTimeout(); + await requestCanceled.Task.DefaultTimeout(); + } + finally + { + watchCts.Cancel(); + try + { + await moveNext.DefaultTimeout(); + } + catch (OperationCanceledException) when (watchCts.IsCancellationRequested) + { + // Expected when closing a pending watch read. + } + } + } + + [Fact] + public async Task WatchAsync_DrainsHttp2StreamAndPreservesEventOrderWhileConsumerPaused() + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var sendRemainingEvents = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var eventsSent = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + const int eventCount = 65; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + Assert.Equal("HTTP/2", context.Request.Protocol); + var container = Container.Create("frontend", "test-image"); + container.Metadata.ResourceVersion = "0"; + await WriteWatchEventAsync(context, container); + await sendRemainingEvents.Task.WaitAsync(context.RequestAborted); + // Exceed the HTTP/2 receive window: without independent reads, these writes + // cannot finish while the consumer is paused on the first event. + container.Metadata.Annotations = new Dictionary { ["padding"] = new('x', 65536) }; + for (var i = 1; i < eventCount; i++) + { + container.Metadata.ResourceVersion = i.ToString(CultureInfo.InvariantCulture); + await WriteWatchEventAsync(context, container); + } + eventsSent.TrySetResult(); + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + } + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + + await using var enumerator = service.WatchAsync(cancellationToken: cts.Token).GetAsyncEnumerator(); + Task? moveNext = null; + try + { + moveNext = enumerator.MoveNextAsync().AsTask(); + Assert.True(await moveNext.DefaultTimeout()); + sendRemainingEvents.TrySetResult(); + await eventsSent.Task.DefaultTimeout(TestConstants.LongTimeoutDuration); + var versions = new List { enumerator.Current.Item2.Metadata.ResourceVersion }; + while (versions.Count < eventCount) + { + moveNext = enumerator.MoveNextAsync().AsTask(); + Assert.True(await moveNext.DefaultTimeout()); + versions.Add(enumerator.Current.Item2.Metadata.ResourceVersion); + } + Assert.Equal(Enumerable.Range(0, eventCount).Select(i => i.ToString(CultureInfo.InvariantCulture)), versions); + } + finally + { + cts.Cancel(); + if (moveNext is { IsCompleted: false }) + { + await Assert.ThrowsAnyAsync(() => moveNext).DefaultTimeout(); + } + } + await requestCanceled.Task.DefaultTimeout(); + } + + [Theory] + [InlineData("ERROR")] + [InlineData("MODIFIED")] + public async Task WatchAsync_PropagatesStatusErrors(string eventType) + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + await using var server = await StartServerAsync(cts.Token, requestHandler: context => + context.Response.WriteAsync( + $$$"""{"type":"{{{eventType}}}","object":{"kind":"Status","status":"Failure","reason":"Expired","code":410}}""" + "\n", + context.RequestAborted)).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + + await using var enumerator = service.WatchAsync(cancellationToken: cts.Token).GetAsyncEnumerator(); + var moveNext = enumerator.MoveNextAsync().AsTask(); + try + { + var error = await Assert.ThrowsAsync(() => moveNext).DefaultTimeout(); + Assert.Equal("Failure", error.Status.Status); + Assert.Equal(410, error.Status.Code); + } + finally + { + cts.Cancel(); + if (!moveNext.IsCompleted) + { + await Assert.ThrowsAnyAsync(() => moveNext).DefaultTimeout(); + } + } + } + + [Theory] + [InlineData("ADDED")] + [InlineData("MODIFIED")] + [InlineData("ERROR")] + public async Task WatchAsync_SuccessfulStatusDoesNotProduceResourceUpdateOrTerminateWatch(string eventType) + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var statusSent = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var sendResource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var requestCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + Interlocked.Increment(ref requestCount); + await context.Response.WriteAsync( + $$$"""{"type":"{{{eventType}}}","object":{"kind":"Status","status":"Success","code":200}}""" + "\n", + context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + statusSent.TrySetResult(); + try + { + await sendResource.Task.WaitAsync(context.RequestAborted); + await WriteWatchEventAsync(context, Container.Create("frontend", "test-image")); + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + } + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + + await using var enumerator = service.WatchAsync(cancellationToken: cts.Token).GetAsyncEnumerator(); + var moveNext = enumerator.MoveNextAsync().AsTask(); + try + { + await statusSent.Task.DefaultTimeout(); + sendResource.TrySetResult(); + Assert.True(await moveNext.DefaultTimeout()); + Assert.Equal(WatchEventType.Modified, enumerator.Current.Item1); + Assert.Equal("frontend", enumerator.Current.Item2.Metadata.Name); + Assert.Equal(1, Volatile.Read(ref requestCount)); + } + finally + { + cts.Cancel(); + if (!moveNext.IsCompleted) + { + await Assert.ThrowsAnyAsync(() => moveNext).DefaultTimeout(); + } + } + await requestCanceled.Task.DefaultTimeout(); + } + + [Theory] + [InlineData("null")] + [InlineData("""{"type":"MODIFIED","object":null}""")] + [InlineData("{invalid-json")] + [InlineData("""{"type":"ERROR","object":{"kind":"Status","code":500}}""")] + [InlineData("""{"type":"MODIFIED","object":{"kind":"Status","status":"Unknown"}}""")] + public async Task WatchAsync_PropagatesMalformedEvents(string line) + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + await using var server = await StartServerAsync(cts.Token, requestHandler: context => + context.Response.WriteAsync(line + "\n")).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + + await using var enumerator = service.WatchAsync(cancellationToken: cts.Token).GetAsyncEnumerator(); + var moveNext = enumerator.MoveNextAsync().AsTask(); + try + { + await Assert.ThrowsAnyAsync(() => moveNext).DefaultTimeout(); + } + finally + { + cts.Cancel(); + if (!moveNext.IsCompleted) + { + await Assert.ThrowsAnyAsync(() => moveNext).DefaultTimeout(); + } + } + } + + [Fact] + public async Task WatchResourceRetryPipeline_ReconnectsAfterTruncatedEvent() + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requestCount = 0; + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + var requestNumber = Interlocked.Increment(ref requestCount); + var container = Container.Create("frontend", "test-image"); + container.Metadata.ResourceVersion = requestNumber.ToString(CultureInfo.InvariantCulture); + await WriteWatchEventAsync(context, container); + if (requestNumber == 1) + { + await context.Response.WriteAsync("""{"type":"MODIFIED","object":""", context.RequestAborted); + return; + } + + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + } + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + var pipeline = DcpPipelineBuilder.BuildWatchResourcePipeline(NullLogger.Instance); + var versions = new List(); + + await pipeline.ExecuteAsync(async token => + { + await foreach (var (_, container) in service.WatchAsync(cancellationToken: token)) + { + versions.Add(container.Metadata.ResourceVersion); + if (versions.Count == 2) + { + return; + } + } + }, cts.Token).DefaultTimeout(TestConstants.LongTimeoutDuration); + + Assert.Equal(["1", "2"], versions); + Assert.Equal(2, Volatile.Read(ref requestCount)); + await requestCanceled.Task.DefaultTimeout(); + } + + [Fact] + public async Task WatchResourceRetryPipeline_ReconnectsAfterEmptyResponses() + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requestCount = 0; + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + if (Interlocked.Increment(ref requestCount) <= 2) + { + await context.Response.StartAsync(context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + return; + } + + await WriteWatchEventAsync(context, Container.Create("frontend", "test-image")); + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + } + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + var pipeline = DcpPipelineBuilder.BuildWatchResourcePipeline(NullLogger.Instance); + Container? observed = null; + + await pipeline.ExecuteAsync(async token => + { + await foreach (var (_, container) in service.WatchAsync(cancellationToken: token)) + { + observed = container; + return; + } + }, cts.Token).DefaultTimeout(TestConstants.LongTimeoutDuration); + + Assert.Equal("frontend", observed?.Metadata.Name); + Assert.Equal(3, Volatile.Read(ref requestCount)); + await requestCanceled.Task.DefaultTimeout(); + } + + [Fact] + public async Task WatchAsync_AcceptsCompleteFinalEventWithoutNewline() + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requestCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + if (Interlocked.Increment(ref requestCount) == 1) + { + var container = Container.Create("frontend", "test-image"); + await context.Response.WriteAsync( + $$"""{"type":"MODIFIED","object":{{KubernetesJson.Serialize(container)}}}""", context.RequestAborted); + return; + } + + await context.Response.StartAsync(context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + + await using var enumerator = service.WatchAsync(cancellationToken: cts.Token).GetAsyncEnumerator(); + var moveNext = enumerator.MoveNextAsync().AsTask(); + try + { + Assert.True(await moveNext.DefaultTimeout()); + Assert.Equal(WatchEventType.Modified, enumerator.Current.Item1); + Assert.Equal("frontend", enumerator.Current.Item2.Metadata.Name); + } + finally + { + cts.Cancel(); + if (!moveNext.IsCompleted) + { + await Assert.ThrowsAnyAsync(() => moveNext).DefaultTimeout(); + } + } + } + + [Theory] + [InlineData("\n", 1200)] + [InlineData("\r\n", 1200)] + [InlineData("\n", 1048576)] + [InlineData("\r\n", 1048576)] + public async Task WatchAsync_ReadsSegmentedUtf8EventWithoutWaitingForMoreData(string newline, int paddingLength) + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var firstPartSent = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var sendRemaining = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var message = new string('x', paddingLength) + "\u2603"; + var container = Container.Create("frontend", "test-image"); + container.Metadata.Annotations = new Dictionary { ["message"] = message }; + var line = ($$"""{"type":"MODIFIED","object":{{KubernetesJson.Serialize(container)}}}""") + .Replace("\\u2603", "\u2603", StringComparison.OrdinalIgnoreCase) + newline; + var bytes = Encoding.UTF8.GetPreamble().Concat(Encoding.UTF8.GetBytes(line)).ToArray(); + // Split a multibyte character across writes. Once the record ends, no more data + // arrives: the reader must return the event without waiting to fill its buffer. + var split = Array.IndexOf(bytes, (byte)0xe2) + 1; + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + context.Response.ContentType = "application/json"; + await context.Response.Body.WriteAsync(bytes.AsMemory(0, split), context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + firstPartSent.TrySetResult(); + await sendRemaining.Task.WaitAsync(context.RequestAborted); + await context.Response.Body.WriteAsync(bytes.AsMemory(split), context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + await using var enumerator = service.WatchAsync(cancellationToken: cts.Token).GetAsyncEnumerator(); + var moveNext = enumerator.MoveNextAsync().AsTask(); + try + { + await firstPartSent.Task.DefaultTimeout(); + Assert.False(moveNext.IsCompleted); + sendRemaining.TrySetResult(); + Assert.True(await moveNext.DefaultTimeout()); + Assert.Equal(message, enumerator.Current.Item2.Metadata.Annotations["message"]); + } + finally + { + sendRemaining.TrySetResult(); + if (!moveNext.IsCompleted) + { + cts.Cancel(); + } + try + { + await moveNext.DefaultTimeout(); + } + catch (OperationCanceledException) when (cts.IsCancellationRequested) + { + // Expected when closing a pending watch read. + } + } + } + + [Theory] + [InlineData(410, "Expired")] + [InlineData(408, "Timeout")] + [InlineData(429, "TooManyRequests")] + [InlineData(500, "InternalError")] + [InlineData(502, "BadGateway")] + [InlineData(503, "ServiceUnavailable")] + [InlineData(504, "Timeout")] + public async Task WatchResourceRetryPipeline_ReconnectsAfterRetryableStatus(int statusCode, string reason) + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requestCount = 0; + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => + { + var requestNumber = Interlocked.Increment(ref requestCount); + var container = Container.Create("frontend", "test-image"); + container.Metadata.ResourceVersion = requestNumber.ToString(CultureInfo.InvariantCulture); + await WriteWatchEventAsync(context, container); + if (requestNumber == 1) + { + await context.Response.WriteAsync( + $$$"""{"type":"ERROR","object":{"kind":"Status","status":"Failure","reason":"{{{reason}}}","code":{{{statusCode}}}}}""" + "\n", + context.RequestAborted); + return; + } + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + } + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + var pipeline = DcpPipelineBuilder.BuildWatchResourcePipeline(NullLogger.Instance); + var versions = new List(); + + await pipeline.ExecuteAsync(async token => + { + await foreach (var (_, container) in service.WatchAsync(cancellationToken: token)) + { + versions.Add(container.Metadata.ResourceVersion); + if (versions.Count == 2) + { + return; + } + } + }, cts.Token).DefaultTimeout(TestConstants.LongTimeoutDuration); + + Assert.Equal(["1", "2"], versions); + Assert.Equal(2, Volatile.Read(ref requestCount)); + await requestCanceled.Task.DefaultTimeout(); + } + + [Theory] + [InlineData(400, "BadRequest")] + [InlineData(401, "Unauthorized")] + [InlineData(403, "Forbidden")] + [InlineData(404, "NotFound")] + public async Task WatchResourceRetryPipeline_DoesNotRetryPermanentStatus(int statusCode, string reason) + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requestCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: context => + { + Interlocked.Increment(ref requestCount); + return context.Response.WriteAsync( + $$$"""{"type":"ERROR","object":{"kind":"Status","status":"Failure","reason":"{{{reason}}}","code":{{{statusCode}}}}}""" + "\n", + context.RequestAborted); + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + var pipeline = DcpPipelineBuilder.BuildWatchResourcePipeline(NullLogger.Instance); + + var error = await Assert.ThrowsAsync(() => pipeline.ExecuteAsync(async token => + { + await foreach (var _ in service.WatchAsync(cancellationToken: token)) + { + } + }, cts.Token).AsTask()).DefaultTimeout(); + + Assert.Equal(statusCode, error.Status.Code); + Assert.Equal(1, Volatile.Read(ref requestCount)); + } + + private static async Task WriteWatchEventAsync(HttpContext context, Container container) + { + if (!context.Response.HasStarted) + { + context.Response.ContentType = "application/json"; + } + await context.Response.WriteAsync( + $$"""{"type":"MODIFIED","object":{{KubernetesJson.Serialize(container)}}}""" + "\n", context.RequestAborted); + await context.Response.Body.FlushAsync(context.RequestAborted); + } + + private Task StartServerAsync( + CancellationToken cancellationToken = default, + int successfulRequestNumber = 1, + Func? requestHandler = null) + { + return TestDcpApiServer.StartAsync(certificateFixture.Certificate, cancellationToken, successfulRequestNumber, requestHandler); + } + [Fact] public async Task CreateAsync_DoesNotReadOrRetrySuccessfulCreate() { @@ -30,7 +772,7 @@ public async Task CreateAsync_DoesNotReadOrRetrySuccessfulCreate() using var disposableFileSystem = fileSystem; using var disposableService = service; var requests = new ConcurrentQueue(); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); var created = await ReadContainerAsync(context); @@ -50,7 +792,7 @@ public async Task DcpKubernetesClient_UsesCallerDeadlineInsteadOfSdkTimeout() { using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(20)); var requests = new ConcurrentQueue(); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); var created = await ReadContainerAsync(context); @@ -60,7 +802,8 @@ public async Task DcpKubernetesClient_UsesCallerDeadlineInsteadOfSdkTimeout() }); using var client = new DcpKubernetesClient(new KubernetesClientConfiguration { - Host = $"http://127.0.0.1:{server.Port}", + Host = $"https://127.0.0.1:{server.Port}", + SkipTlsVerify = true, HttpClientTimeout = TimeSpan.FromMilliseconds(100), }); @@ -87,7 +830,7 @@ public async Task CreateAsync_ReconcilesGatewayTimeoutAfterCommit(string? namesp using var disposableService = service; var requests = new ConcurrentQueue(); Container? stored = null; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue($"{context.Request.Method} {context.Request.Path}"); if (HttpMethods.IsPost(context.Request.Method)) @@ -130,7 +873,7 @@ public async Task CreateAsync_RetriesTransientFailureAfterConfirmingObjectAbsent var requests = new ConcurrentQueue(); var postedObjects = new ConcurrentQueue(); var postCount = 0; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsGet(context.Request.Method)) @@ -183,7 +926,7 @@ public async Task CreateAsync_ReconcilesAlreadyExistsWhenOriginalCreateCommitsAf var requests = new ConcurrentQueue(); Container? original = null; var postCount = 0; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsPost(context.Request.Method)) @@ -236,7 +979,7 @@ public async Task CreateAsync_RetriesCreateAfterTransientConfirmationFailure(int Container? stored = null; var postCount = 0; var getCount = 0; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsPost(context.Request.Method)) @@ -283,7 +1026,7 @@ public async Task CreateAsync_DoesNotConfirmOtherConflictsAfterAmbiguousCreate(s using var disposableService = service; var requests = new ConcurrentQueue(); var postCount = 0; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsGet(context.Request.Method)) @@ -325,7 +1068,7 @@ public async Task CreateAsync_DoesNotRetryPermanentFailureOrInitialConflict(int using var disposableFileSystem = fileSystem; using var disposableService = service; var requests = new ConcurrentQueue(); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); await WriteStatusAsync(context, statusCode, statusCode == 409 ? "AlreadyExists" : "PermanentFailure"); @@ -349,7 +1092,7 @@ public async Task CreateAsync_ReconcilesExistingObjectByName(bool hasAnnotations using var disposableFileSystem = fileSystem; using var disposableService = service; var requests = new ConcurrentQueue(); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsPost(context.Request.Method)) @@ -396,7 +1139,7 @@ public async Task CreateAsync_RecoversAfterAttemptTimeout(bool committedBeforeTi var canceledRequest = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); Container? stored = null; var postCount = 0; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { if (context.Request.Path.Value!.EndsWith("/containers", StringComparison.Ordinal) && HttpMethods.IsGet(context.Request.Method)) @@ -463,7 +1206,7 @@ public async Task CreateAsync_ReconcilesLostResponseAfterCommit() using var disposableService = service; var requests = new ConcurrentQueue(); Container? stored = null; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsPost(context.Request.Method)) @@ -497,7 +1240,7 @@ public async Task CreateAsync_RetriesCreateAfterConfirmationTimeout() Container? stored = null; var postCount = 0; var getCount = 0; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { if (HttpMethods.IsGet(context.Request.Method) && context.Request.Path.Value!.EndsWith("/containers", StringComparison.Ordinal)) @@ -561,7 +1304,7 @@ public async Task CreateAsync_RetriesWhenAlreadyExistsCannotBeConfirmed() using var disposableService = service; var requests = new ConcurrentQueue(); var postCount = 0; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsGet(context.Request.Method)) @@ -605,7 +1348,7 @@ public async Task CreateAsync_RecoveryBudgetIncludesServerRequestedRetryDelay(bo using var disposableFileSystem = fileSystem; using var disposableService = service; var requests = new ConcurrentQueue(); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsGet(context.Request.Method)) @@ -640,7 +1383,7 @@ public async Task CreateAsync_RecoveryBudgetIncludesConfirmationRetryDelay(bool using var disposableFileSystem = fileSystem; using var disposableService = service; var requests = new ConcurrentQueue(); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsPost(context.Request.Method)) @@ -674,7 +1417,7 @@ public async Task CreateAsync_ConfirmationTimeCountsTowardRetryAfter(bool useHtt using var disposableService = service; var requests = new ConcurrentQueue(); var postCount = 0; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { if (HttpMethods.IsGet(context.Request.Method) && context.Request.Path.Value!.EndsWith("/containers", StringComparison.Ordinal)) @@ -727,7 +1470,7 @@ public async Task CreateAsync_InitializationTimeoutDoesNotConfirmOrAdoptExisting using var disposableFileSystem = fileSystem; using var disposableService = service; var requests = new ConcurrentQueue(); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsGet(context.Request.Method)) @@ -767,7 +1510,7 @@ public async Task CreateAsync_RecoveryBudgetCancelsStalledReconciliation() using var disposableService = service; var requests = new ConcurrentQueue(); var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsPost(context.Request.Method)) @@ -804,7 +1547,7 @@ public async Task CreateAsync_DoesNotRetryPermanentReconciliationFailure() using var disposableFileSystem = fileSystem; using var disposableService = service; var requests = new ConcurrentQueue(); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); await WriteStatusAsync(context, @@ -831,7 +1574,7 @@ public async Task CreateAsync_CallerCancellationDoesNotReconcileOrRetry() var requests = new ConcurrentQueue(); var requestArrived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); requestArrived.TrySetResult(); @@ -874,7 +1617,7 @@ public async Task ExecuteWithRetry_EstablishesConnection_WhenKubeconfigInitially await Task.Delay(300, cts.Token); - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); WriteKubeconfig(kubeconfigPath, server.Port); var result = await listTask; @@ -897,7 +1640,7 @@ public async Task ExecuteWithRetry_UsesInitializationTimeout_WhenKubeconfigAppea // the initialization budget rather than the shorter steady-state budget. await Task.Delay(TimeSpan.FromSeconds(2), cts.Token); - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); WriteKubeconfig(kubeconfigPath, server.Port); var result = await listTask; @@ -913,7 +1656,7 @@ public async Task ExecuteWithRetry_InitializationBudgetIncludesApiTimeoutPlusAdd kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(1)); using var disposableFileSystem = fileSystem; using var disposableService = service; - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); server.DelayResponses(TimeSpan.FromSeconds(1.5)); WriteKubeconfig(kubeconfigPath, server.Port); @@ -932,7 +1675,7 @@ public async Task ExecuteWithRetry_ZeroInitializationAllowanceUsesApiBudget() kubernetesInitializationAdditionalTimeout: TimeSpan.Zero); using var disposableFileSystem = fileSystem; using var disposableService = service; - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); server.DelayResponses(TimeSpan.FromSeconds(2)); WriteKubeconfig(kubeconfigPath, server.Port); @@ -950,7 +1693,7 @@ public async Task ExecuteWithRetry_ClientSetupAndApiRequestShareInitializationBu kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(3)); using var disposableFileSystem = fileSystem; using var disposableService = service; - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); server.DelayResponses(TimeSpan.FromSeconds(3)); var listTask = service.ListAsync(cancellationToken: cts.Token); @@ -975,7 +1718,7 @@ public async Task CreateAsync_ConfirmsTimedOutWrite_WhenSetupConsumesInitializat var requests = new ConcurrentQueue(); var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); Container? stored = null; - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { requests.Enqueue(context.Request.Method); if (HttpMethods.IsGet(context.Request.Method)) @@ -1023,7 +1766,7 @@ public async Task ExecuteWithRetry_UsesInitializationTimeout_UntilFirstApiOperat // The fourth API request succeeds after exponential retry delays of 100, 200, and 400 milliseconds. // It is reachable under the initialization budget but not the 500-millisecond steady-state budget. - await using var server = await TestDcpApiServer.StartAsync(cts.Token, successfulRequestNumber: 4); + await using var server = await StartServerAsync(cts.Token, successfulRequestNumber: 4); var listTask = service.ListAsync(cancellationToken: cts.Token); await Task.Delay(TimeSpan.FromSeconds(1), cts.Token); @@ -1044,7 +1787,7 @@ public async Task ExecuteWithRetry_UsesApiRetryDuration_AfterFirstApiOperationSu using var disposableFileSystem = fileSystem; using var disposableService = service; - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); WriteKubeconfig(kubeconfigPath, server.Port); var result = await service.ListAsync(cancellationToken: cts.Token); @@ -1070,7 +1813,7 @@ public async Task ExecuteWithRetry_CancelsApiRequest_WhenRetryDurationExpires(st using var disposableFileSystem = fileSystem; using var disposableService = service; - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); WriteKubeconfig(kubeconfigPath, server.Port); Assert.Empty(await service.ListAsync(cancellationToken: cts.Token)); @@ -1099,7 +1842,7 @@ public async Task ExecuteWithRetry_ReadinessDoesNotShortenInFlightOperation() using var disposableService = service; var requestArrived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { if (context.Request.Path.Value!.EndsWith("/readiness-probe", StringComparison.Ordinal)) { @@ -1145,7 +1888,7 @@ public async Task WatchAsync_DoesNotMarkApiReady_WhileHttpResponseIsPending() using var disposableFileSystem = fileSystem; using var disposableService = service; - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); server.BlockWatchResponses(); WriteKubeconfig(kubeconfigPath, server.Port); @@ -1177,7 +1920,7 @@ public async Task WatchAsync_ReconnectUsesApiTimeout_AfterAnotherOperationConfir kubernetesInitializationAdditionalTimeout: TimeSpan.FromSeconds(2.75)); using var disposableFileSystem = fileSystem; using var disposableService = service; - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); server.BlockWatchResponses(); WriteKubeconfig(kubeconfigPath, server.Port); @@ -1213,7 +1956,7 @@ public async Task WatchAsync_ConnectedStreamOutlivesApiTimeout() using var disposableFileSystem = fileSystem; using var disposableService = service; var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { var container = Container.Create("frontend", "test-image"); context.Response.ContentType = "application/json"; @@ -1262,7 +2005,7 @@ public async Task GetLogStreamAsync_ConnectedStreamOutlivesApiTimeout() using var disposableFileSystem = fileSystem; using var disposableService = service; var continueStream = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - await using var server = await TestDcpApiServer.StartAsync(cts.Token, requestHandler: async context => + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { context.Response.ContentType = "text/plain"; await context.Response.WriteAsync("first\n", context.RequestAborted); @@ -1299,7 +2042,7 @@ public async Task WatchAsync_RemainsActiveBeyondApiRetryTimeout() using var disposableFileSystem = fileSystem; using var disposableService = service; - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); server.BlockWatchResponses(); WriteKubeconfig(kubeconfigPath, server.Port); @@ -1335,7 +2078,7 @@ public async Task ExecuteWithRetry_EstablishesConnection_WhenKubeconfigInitially // Give the read pipeline time to observe and retry the partial file before we finish writing it. await Task.Delay(300, cts.Token); - await using var server = await TestDcpApiServer.StartAsync(cts.Token); + await using var server = await StartServerAsync(cts.Token); // Finish the write by appending the remainder onto the same file. In-flight DCP calls will now succeed. CompleteKubeconfig(kubeconfigPath, server.Port); @@ -1405,15 +2148,15 @@ private static Task WriteStatusAsync(HttpContext context, int statusCode, string private static void WriteKubeconfig(string path, int port) { - // Minimal kubeconfig pointing at a plain-HTTP loopback endpoint with no auth, which is all the - // DcpKubernetesClient needs to issue custom-object requests against the fake server. + // The fixture's self-signed certificate needs no machine trust installation. var content = string.Format(CultureInfo.InvariantCulture, """ apiVersion: v1 kind: Config clusters: - name: dcp cluster: - server: http://127.0.0.1:{0} + server: https://127.0.0.1:{0} + insecure-skip-tls-verify: true contexts: - name: dcp context: @@ -1445,7 +2188,8 @@ private static void WritePartialKubeconfig(string path) clusters: - name: dcp cluster: - server: "http://127.0.0.1: + insecure-skip-tls-verify: true + server: "https://127.0.0.1: """); } @@ -1453,7 +2197,7 @@ private static void CompleteKubeconfig(string path, int port) { // Append (do not rewrite) the remainder of the kubeconfig that WritePartialKubeconfig left unfinished. // The remainder begins by closing the open server scalar ({port}") so the combined file parses as a - // valid kubeconfig pointing at the loopback fake server with no auth. + // valid kubeconfig pointing at the HTTPS loopback fake server. File.AppendAllText(path, string.Format(CultureInfo.InvariantCulture, """ {0}" contexts: @@ -1588,6 +2332,7 @@ public Task WaitForRequestCancellationAsync(CancellationToken cancellationToken) } public static async Task StartAsync( + X509Certificate2 certificate, CancellationToken cancellationToken = default, int successfulRequestNumber = 1, Func? requestHandler = null) @@ -1597,7 +2342,11 @@ public static async Task StartAsync( builder.Logging.ClearProviders(); // Port 0 lets the OS pick a free port that Kestrel binds and holds. After StartAsync the addresses // feature (exposed via app.Urls) is rewritten with the resolved address, so we can read the real port. - builder.WebHost.UseUrls("http://127.0.0.1:0"); + builder.WebHost.ConfigureKestrel(options => options.Listen(IPAddress.Loopback, 0, endpoint => + { + endpoint.Protocols = HttpProtocols.Http2; + endpoint.UseHttps(certificate); + })); var app = builder.Build(); var responseState = new ResponseState @@ -1607,6 +2356,8 @@ public static async Task StartAsync( app.Run(async context => { + Assert.True(context.Request.IsHttps); + Assert.Equal("HTTP/2", context.Request.Protocol); if (requestHandler is not null) { await requestHandler(context); @@ -1655,9 +2406,17 @@ public static async Task StartAsync( await context.Response.WriteAsync("""{"apiVersion":"usvc-dev.developer.microsoft.com/v1","kind":"ContainerList","items":[]}"""); }); - await app.StartAsync(cancellationToken).ConfigureAwait(false); + try + { + await app.StartAsync(cancellationToken).ConfigureAwait(false); + } + catch + { + await app.DisposeAsync().ConfigureAwait(false); + throw; + } - // e.g. "http://127.0.0.1:54321" -> 54321 + // e.g. "https://127.0.0.1:54321" -> 54321 var address = app.Urls.First(); var port = new Uri(address).Port; From e92b3d4bc91ec3452203aac5116a0593593c064c Mon Sep 17 00:00:00 2001 From: David Negstad Date: Tue, 13 Oct 2026 13:17:35 -0700 Subject: [PATCH 2/2] Fix watch final-record classification and HTTP status retries Distinguish incomplete JSON from complete malformed final records so missing newlines do not turn permanent parsing failures into retries. Apply one watch status policy to HTTP response failures and in-stream Status failures. Cover both response forms for every status code without duplicating theory rows, and add malformed final-record and partial JSON recovery scenarios. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: a4d11434-2c0b-4741-91ea-9eacc92ca29a --- src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs | 56 +++--- src/Aspire.Hosting/Dcp/DcpPipelineBuilder.cs | 20 +- .../Dcp/KubernetesServiceTests.cs | 184 +++++++++++------- 3 files changed, 157 insertions(+), 103 deletions(-) diff --git a/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs b/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs index 452a052bab2..8828ed5ef52 100644 --- a/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs +++ b/src/Aspire.Hosting/Dcp/DcpKubernetesClient.cs @@ -329,39 +329,51 @@ private static (WatchEventType, T)? DeserializeEvent(string line, bool terminate // {"type":"MODIFIED","object":{"kind":"Executable","metadata":{...},...}} // An ERROR can contain a Status instead; BOOKMARK objects contain only metadata. // https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes - try + if (!terminated) { - var envelope = KubernetesJson.Deserialize.WatchEvent>(line); - if (envelope?.Object is null) + // A missing newline does not make a complete payload such as null truncated. + // Validate syntax separately so resource shape errors remain non-retryable. + try { - throw new JsonException("DCP watch event does not contain an object."); + using var document = JsonDocument.Parse(line); } - if (envelope.Object.Kind == "Status") + catch (JsonException ex) { - var status = KubernetesJson.Deserialize.WatchEvent>(line).Object; - if (status.Status == "Failure") - { - throw new KubernetesException(status); - } - if (status.Status != "Success") + var reader = new Utf8JsonReader(Encoding.UTF8.GetBytes(line), isFinalBlock: false, state: default); + if (!JsonDocument.TryParseValue(ref reader, out var document)) { - throw new JsonException("DCP watch Status does not report Success or Failure."); + // EOF can leave a partial record such as {"type":"MODIFIED","object": + throw new EndOfStreamException("DCP watch ended in an incomplete event.", ex); } - - // A successful Status is not a resource update and must not be deserialized as T. - return null; + document.Dispose(); + throw; } + } - var watchEvent = KubernetesJson.Deserialize.WatchEvent>(line); - - return (watchEvent.Type, watchEvent.Object); + var envelope = KubernetesJson.Deserialize.WatchEvent>(line); + if (envelope?.Object is null) + { + throw new JsonException("DCP watch event does not contain an object."); } - catch (JsonException ex) when (!terminated) + if (envelope.Object.Kind == "Status") { - // EOF can leave a partial record such as {"type":"MODIFIED","object": - // Retry that interrupted stream, while complete malformed records remain errors. - throw new EndOfStreamException("DCP watch ended in an incomplete event.", ex); + var status = KubernetesJson.Deserialize.WatchEvent>(line).Object; + if (status.Status == "Failure") + { + throw new KubernetesException(status); + } + if (status.Status != "Success") + { + throw new JsonException("DCP watch Status does not report Success or Failure."); + } + + // A successful Status is not a resource update and must not be deserialized as T. + return null; } + + var watchEvent = KubernetesJson.Deserialize.WatchEvent>(line); + + return (watchEvent.Type, watchEvent.Object); } } diff --git a/src/Aspire.Hosting/Dcp/DcpPipelineBuilder.cs b/src/Aspire.Hosting/Dcp/DcpPipelineBuilder.cs index 3471313bfd1..f04428b586c 100644 --- a/src/Aspire.Hosting/Dcp/DcpPipelineBuilder.cs +++ b/src/Aspire.Hosting/Dcp/DcpPipelineBuilder.cs @@ -3,6 +3,7 @@ using System.Net; using k8s; +using k8s.Autorest; using Microsoft.Extensions.Logging; using Polly; using Polly.Retry; @@ -69,14 +70,8 @@ public static ResiliencePipeline BuildWatchResourcePipeline(ILogger logger) { ShouldHandle = new PredicateBuilder() .HandleInner() - .HandleInner(ex => ex.Status.Code is - (int)HttpStatusCode.Gone or - (int)HttpStatusCode.RequestTimeout or - (int)HttpStatusCode.TooManyRequests or - (int)HttpStatusCode.InternalServerError or - (int)HttpStatusCode.BadGateway or - (int)HttpStatusCode.ServiceUnavailable or - (int)HttpStatusCode.GatewayTimeout), + .HandleInner(ex => IsRetryableWatchStatusCode(ex.Status.Code)) + .HandleInner(ex => IsRetryableWatchStatusCode((int)ex.Response.StatusCode)), BackoffType = DelayBackoffType.Exponential, MaxRetryAttempts = int.MaxValue, UseJitter = true, @@ -96,4 +91,13 @@ public static ResiliencePipeline BuildWatchResourcePipeline(ILogger logger) var pipeline = new ResiliencePipelineBuilder().AddRetry(retryUntilCancelled).Build(); return pipeline; } + + private static bool IsRetryableWatchStatusCode(int? statusCode) => statusCode is + (int)HttpStatusCode.Gone or + (int)HttpStatusCode.RequestTimeout or + (int)HttpStatusCode.TooManyRequests or + (int)HttpStatusCode.InternalServerError or + (int)HttpStatusCode.BadGateway or + (int)HttpStatusCode.ServiceUnavailable or + (int)HttpStatusCode.GatewayTimeout; } diff --git a/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs b/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs index fcd7390f65e..a4de1b05ba9 100644 --- a/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs +++ b/tests/Aspire.Hosting.Tests/Dcp/KubernetesServiceTests.cs @@ -414,12 +414,21 @@ await context.Response.WriteAsync( } [Theory] - [InlineData("null")] - [InlineData("""{"type":"MODIFIED","object":null}""")] - [InlineData("{invalid-json")] - [InlineData("""{"type":"ERROR","object":{"kind":"Status","code":500}}""")] - [InlineData("""{"type":"MODIFIED","object":{"kind":"Status","status":"Unknown"}}""")] - public async Task WatchAsync_PropagatesMalformedEvents(string line) + [InlineData("null", true)] + [InlineData("null", false)] + [InlineData("""{"type":"MODIFIED","object":null}""", true)] + [InlineData("""{"type":"MODIFIED","object":null}""", false)] + [InlineData("{invalid-json", true)] + [InlineData("{invalid-json", false)] + [InlineData("""{"type":"ERROR","object":{"kind":"Status","code":500}}""", true)] + [InlineData("""{"type":"ERROR","object":{"kind":"Status","code":500}}""", false)] + [InlineData("""{"type":"MODIFIED","object":{"kind":"Status","status":"Unknown"}}""", true)] + [InlineData("""{"type":"MODIFIED","object":{"kind":"Status","status":"Unknown"}}""", false)] + [InlineData("42", false)] + [InlineData("true", false)] + [InlineData("[]", false)] + [InlineData("{}", false)] + public async Task WatchAsync_PropagatesMalformedEvents(string line, bool appendNewline) { using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); @@ -427,7 +436,7 @@ public async Task WatchAsync_PropagatesMalformedEvents(string line) using var disposableFileSystem = fileSystem; using var disposableService = service; await using var server = await StartServerAsync(cts.Token, requestHandler: context => - context.Response.WriteAsync(line + "\n")).DefaultTimeout(); + context.Response.WriteAsync(line + (appendNewline ? "\n" : string.Empty), context.RequestAborted)).DefaultTimeout(); WriteKubeconfig(kubeconfigPath, server.Port); await using var enumerator = service.WatchAsync(cancellationToken: cts.Token).GetAsyncEnumerator(); @@ -446,8 +455,11 @@ public async Task WatchAsync_PropagatesMalformedEvents(string line) } } - [Fact] - public async Task WatchResourceRetryPipeline_ReconnectsAfterTruncatedEvent() + [Theory] + [InlineData("""{"type":"MODIFIED","object":""")] + [InlineData("""{"type":"MODIFIED","object":{"kind":"Container","metadata":{"name":"frontend"}}""")] + [InlineData("""{"type":"MODIFIED","object":{"kind":"Container","metadata":{"name":"front""")] + public async Task WatchResourceRetryPipeline_ReconnectsAfterTruncatedEvent(string partialEvent) { using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); @@ -464,7 +476,7 @@ public async Task WatchResourceRetryPipeline_ReconnectsAfterTruncatedEvent() await WriteWatchEventAsync(context, container); if (requestNumber == 1) { - await context.Response.WriteAsync("""{"type":"MODIFIED","object":""", context.RequestAborted); + await context.Response.WriteAsync(partialEvent, context.RequestAborted); return; } @@ -661,54 +673,63 @@ public async Task WatchAsync_ReadsSegmentedUtf8EventWithoutWaitingForMoreData(st [InlineData(504, "Timeout")] public async Task WatchResourceRetryPipeline_ReconnectsAfterRetryableStatus(int statusCode, string reason) { - using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); - using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); - var (service, kubeconfigPath, fileSystem) = CreateService(); - using var disposableFileSystem = fileSystem; - using var disposableService = service; - var requestCount = 0; - var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - await using var server = await StartServerAsync(cts.Token, requestHandler: async context => - { - var requestNumber = Interlocked.Increment(ref requestCount); - var container = Container.Create("frontend", "test-image"); - container.Metadata.ResourceVersion = requestNumber.ToString(CultureInfo.InvariantCulture); - await WriteWatchEventAsync(context, container); - if (requestNumber == 1) - { - await context.Response.WriteAsync( - $$$"""{"type":"ERROR","object":{"kind":"Status","status":"Failure","reason":"{{{reason}}}","code":{{{statusCode}}}}}""" + "\n", - context.RequestAborted); - return; - } - try - { - await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); - } - catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + foreach (var httpResponse in new[] { false, true }) + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requestCount = 0; + var requestCanceled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var server = await StartServerAsync(cts.Token, requestHandler: async context => { - requestCanceled.TrySetResult(); - } - }).DefaultTimeout(); - WriteKubeconfig(kubeconfigPath, server.Port); - var pipeline = DcpPipelineBuilder.BuildWatchResourcePipeline(NullLogger.Instance); - var versions = new List(); + var requestNumber = Interlocked.Increment(ref requestCount); + if (requestNumber == 1 && httpResponse) + { + await WriteStatusAsync(context, statusCode, reason); + return; + } + var container = Container.Create("frontend", "test-image"); + container.Metadata.ResourceVersion = requestNumber.ToString(CultureInfo.InvariantCulture); + await WriteWatchEventAsync(context, container); + if (requestNumber == 1) + { + await context.Response.WriteAsync( + $$$"""{"type":"ERROR","object":{"kind":"Status","status":"Failure","reason":"{{{reason}}}","code":{{{statusCode}}}}}""" + "\n", + context.RequestAborted); + return; + } + try + { + await Task.Delay(Timeout.InfiniteTimeSpan, context.RequestAborted); + } + catch (OperationCanceledException) when (context.RequestAborted.IsCancellationRequested) + { + requestCanceled.TrySetResult(); + } + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + var pipeline = DcpPipelineBuilder.BuildWatchResourcePipeline(NullLogger.Instance); + var versions = new List(); + string[] expectedVersions = httpResponse ? ["2"] : ["1", "2"]; - await pipeline.ExecuteAsync(async token => - { - await foreach (var (_, container) in service.WatchAsync(cancellationToken: token)) + await pipeline.ExecuteAsync(async token => { - versions.Add(container.Metadata.ResourceVersion); - if (versions.Count == 2) + await foreach (var (_, container) in service.WatchAsync(cancellationToken: token)) { - return; + versions.Add(container.Metadata.ResourceVersion); + if (versions.Count == expectedVersions.Length) + { + return; + } } - } - }, cts.Token).DefaultTimeout(TestConstants.LongTimeoutDuration); + }, cts.Token).DefaultTimeout(TestConstants.LongTimeoutDuration); - Assert.Equal(["1", "2"], versions); - Assert.Equal(2, Volatile.Read(ref requestCount)); - await requestCanceled.Task.DefaultTimeout(); + Assert.Equal(expectedVersions, versions); + Assert.Equal(2, Volatile.Read(ref requestCount)); + await requestCanceled.Task.DefaultTimeout(); + } } [Theory] @@ -718,31 +739,48 @@ await pipeline.ExecuteAsync(async token => [InlineData(404, "NotFound")] public async Task WatchResourceRetryPipeline_DoesNotRetryPermanentStatus(int statusCode, string reason) { - using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); - using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); - var (service, kubeconfigPath, fileSystem) = CreateService(); - using var disposableFileSystem = fileSystem; - using var disposableService = service; - var requestCount = 0; - await using var server = await StartServerAsync(cts.Token, requestHandler: context => - { - Interlocked.Increment(ref requestCount); - return context.Response.WriteAsync( - $$$"""{"type":"ERROR","object":{"kind":"Status","status":"Failure","reason":"{{{reason}}}","code":{{{statusCode}}}}}""" + "\n", - context.RequestAborted); - }).DefaultTimeout(); - WriteKubeconfig(kubeconfigPath, server.Port); - var pipeline = DcpPipelineBuilder.BuildWatchResourcePipeline(NullLogger.Instance); + foreach (var httpResponse in new[] { false, true }) + { + using var timeout = AsyncTestHelpers.CreateDefaultTimeoutTokenSource(TestConstants.LongTimeoutDuration); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, TestContext.Current.CancellationToken); + var (service, kubeconfigPath, fileSystem) = CreateService(); + using var disposableFileSystem = fileSystem; + using var disposableService = service; + var requestCount = 0; + await using var server = await StartServerAsync(cts.Token, requestHandler: context => + { + Interlocked.Increment(ref requestCount); + if (httpResponse) + { + return WriteStatusAsync(context, statusCode, reason); + } + return context.Response.WriteAsync( + $$$"""{"type":"ERROR","object":{"kind":"Status","status":"Failure","reason":"{{{reason}}}","code":{{{statusCode}}}}}""" + "\n", + context.RequestAborted); + }).DefaultTimeout(); + WriteKubeconfig(kubeconfigPath, server.Port); + var pipeline = DcpPipelineBuilder.BuildWatchResourcePipeline(NullLogger.Instance); - var error = await Assert.ThrowsAsync(() => pipeline.ExecuteAsync(async token => - { - await foreach (var _ in service.WatchAsync(cancellationToken: token)) + if (httpResponse) + { + var error = await Assert.ThrowsAsync(WatchAsync).DefaultTimeout(); + Assert.Equal(statusCode, (int)error.Response.StatusCode); + } + else { + var error = await Assert.ThrowsAsync(WatchAsync).DefaultTimeout(); + Assert.Equal(statusCode, error.Status.Code); } - }, cts.Token).AsTask()).DefaultTimeout(); - Assert.Equal(statusCode, error.Status.Code); - Assert.Equal(1, Volatile.Read(ref requestCount)); + Assert.Equal(1, Volatile.Read(ref requestCount)); + + Task WatchAsync() => pipeline.ExecuteAsync(async token => + { + await foreach (var _ in service.WatchAsync(cancellationToken: token)) + { + } + }, cts.Token).AsTask(); + } } private static async Task WriteWatchEventAsync(HttpContext context, Container container)