// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
using System.Runtime.CompilerServices;
using Aspire.Hosting.ApplicationModel;
using Aspire.Hosting.Diagnostics;
using Aspire.Hosting.Pipelines;
using Aspire.Hosting.Utils;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
namespace Aspire.Hosting.Backchannel;
internal class AppHostRpcTarget(
ILogger<AppHostRpcTarget> logger,
ResourceNotificationService resourceNotificationService,
IServiceProvider serviceProvider,
ProfilingTelemetry profilingTelemetry,
PipelineActivityReporter activityReporter,
IHostApplicationLifetime lifetime,
DistributedApplicationOptions options,
AppHostStartupState startupState,
IInteractionFileUploadStore fileUploadStore,
IConfiguration configuration)
{
private readonly CancellationTokenSource _shutdownCts = new();
public async IAsyncEnumerable<BackchannelLogEntry> GetAppHostLogEntriesAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
// Complete the stream immediately if shutdown has already been requested.
if (_shutdownCts.IsCancellationRequested)
{
yield break;
}
// Create a linked token source that will be cancelled when shutdown is requested
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _shutdownCts.Token);
var linkedToken = linkedCts.Token;
var loggerProvider = serviceProvider.GetService<BackchannelLoggerProvider>();
if (loggerProvider is null)
{
yield break;
}
// Subscribe atomically: snapshot + channel for new entries, no gap
var (snapshot, subscriberId, channel) = loggerProvider.Subscribe();
try
{
// Replay buffered entries first so late-connecting clients see history
foreach (var entry in snapshot)
{
yield return entry;
}
// Stream live entries — uses a helper that swallows OperationCanceledException on cancellation
// instead of propagating it, since yield return cannot appear in a try/catch block.
await foreach (var entry in AsyncEnumerableUtils.ReadUntilCancelledAsync(channel.Reader.ReadAllAsync(linkedToken), linkedToken).ConfigureAwait(false))
{
yield return entry;
}
}
finally
{
loggerProvider.Unsubscribe(subscriberId);
}
}
public async IAsyncEnumerable<PublishingActivity> GetPublishingActivitiesAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
// Complete the stream immediately if shutdown has already been requested.
if (_shutdownCts.IsCancellationRequested)
{
yield break;
}
// Create a linked token source that will be cancelled when shutdown is requested
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _shutdownCts.Token);
var linkedToken = linkedCts.Token;
while (!linkedToken.IsCancellationRequested)
{
PublishingActivity? publishingActivity = null;
try
{
publishingActivity = await activityReporter.ActivityItemUpdated.Reader.ReadAsync(linkedToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (linkedToken.IsCancellationRequested)
{
// Expected when the stream is cancelled due to shutdown or client disconnect.
logger.LogDebug("Publishing activities stream cancelled.");
yield break;
}
// Terminate the stream if the publishing activity is null
if (publishingActivity == null)
{
yield break;
}
yield return publishingActivity;
}
}
public async IAsyncEnumerable<RpcResourceState> GetResourceStatesAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
// Complete the stream immediately if shutdown has already been requested.
if (_shutdownCts.IsCancellationRequested)
{
yield break;
}
// Create a linked token source that will be cancelled when shutdown is requested
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _shutdownCts.Token);
var linkedToken = linkedCts.Token;
var resourceEvents = resourceNotificationService.WatchAsync(linkedToken);
// Use a helper that swallows OperationCanceledException on cancellation instead of propagating it,
// since yield return cannot appear in a try/catch block.
await foreach (var resourceEvent in AsyncEnumerableUtils.ReadUntilCancelledAsync(resourceEvents, linkedToken).ConfigureAwait(false))
{
if (string.Equals(resourceEvent.Resource.Name, KnownResourceNames.AspireDashboard, StringComparisons.ResourceName))
{
// Skip the dashboard resource, as it is handled separately.
continue;
}
if (!resourceEvent.Resource.TryGetEndpoints(out var endpoints))
{
logger.LogTrace("Resource {ResourceName} does not have endpoints.", resourceEvent.Resource.Name);
endpoints = Enumerable.Empty<EndpointAnnotation>();
}
var endpointUris = endpoints
.Where(e => e.AllocatedEndpoint != null)
.Select(e => e.AllocatedEndpoint!.UriString)
.ToArray();
// Compute health status
var healthStatus = CustomResourceSnapshot.ComputeHealthStatus(resourceEvent.Snapshot.HealthReports, resourceEvent.Snapshot.State?.Text);
yield return new RpcResourceState
{
Resource = resourceEvent.Resource.Name,
Type = resourceEvent.Snapshot.ResourceType,
State = resourceEvent.Snapshot.State?.Text ?? "Unknown",
Endpoints = endpointUris,
Health = healthStatus?.ToString()
};
}
}
public Task RequestStopAsync(CancellationToken cancellationToken)
{
_ = cancellationToken;
// Cancel inflight streaming RPC calls before stopping the application
_shutdownCts.Cancel();
lifetime.StopApplication();
return Task.CompletedTask;
}
/// <summary>
/// Cancels inflight streaming RPC calls to allow graceful shutdown.
/// This should be called before stopping the application to prevent JSON-RPC errors on clients.
/// </summary>
public void CancelInflightRpcCalls()
{
_shutdownCts.Cancel();
}
public async Task<DashboardUrlsState> GetDashboardUrlsAsync(CancellationToken cancellationToken)
{
using var activity = profilingTelemetry.StartJsonRpcServerCall(nameof(GetDashboardUrlsAsync), streaming: false);
if (!options.DashboardEnabled)
{
logger.LogDebug("Dashboard URL requested but dashboard is disabled.");
activity.SetDashboardHealthy(false);
return new DashboardUrlsState { DashboardHealthy = false };
}
try
{
var urls = await DashboardUrlsHelper.GetDashboardUrlsAsync(serviceProvider, logger, cancellationToken).ConfigureAwait(false);
activity.SetDashboardHealthy(urls.DashboardHealthy);
return urls;
}
catch (Exception ex)
{
activity.SetError(ex);
throw;
}
}
public Task NotifyAppHostReadyAsync(CancellationToken cancellationToken)
{
_ = cancellationToken;
startupState.MarkReady();
return Task.CompletedTask;
}
#pragma warning disable CA1822
public Task<string[]> GetCapabilitiesAsync(CancellationToken cancellationToken)
{
using var activity = profilingTelemetry.StartJsonRpcServerCall(nameof(GetCapabilitiesAsync), streaming: false);
// The purpose of this API is to allow the CLI to determine what API surfaces
// the AppHost supports. In 9.2 we'll be saying that you need a 9.2 apphost,
// but the 9.3 CLI might actually support working with 9.2 apphosts. The idea
// is that when the backchannel is established the CLI will call this API
// and store the results. The "baseline.v0" capability is the bare minimum
// that we need as of CLI version 9.2-preview*.
//
// Some capabilities will be opt in. For example in 9.3 we might refine the
// publishing activities API to return more information, or add log streaming
// features. So that would add a new capability that the apphost can report
// on initial backchannel negotiation and the CLI can adapt its behavior around
// that. There may be scenarios where we need to break compatibility at which
// point we might increase the baseline version that the apphost reports.
//
// The ability to support a back channel at all is determined by the CLI by
// making sure that the apphost version is at least > 9.2.
_ = cancellationToken;
return Task.FromResult(new string[] {
"baseline.v2",
"pipeline-steps.v1",
"pipeline-steps.v2"
});
}
#pragma warning restore CA1822
public async Task CompletePromptResponseAsync(string promptId, PublishingPromptInputAnswer[] answers, CancellationToken cancellationToken = default)
{
await activityReporter.CompleteInteractionAsync(promptId, answers, updateResponse: false, cancellationToken).ConfigureAwait(false);
}
public async Task UpdatePromptResponseAsync(string promptId, PublishingPromptInputAnswer[] answers, CancellationToken cancellationToken = default)
{
await activityReporter.CompleteInteractionAsync(promptId, answers, updateResponse: true, cancellationToken).ConfigureAwait(false);
}
/// <summary>
/// Registers a local file in the upload store by copying it to a managed temp location.
/// Returns the file ID that can be used to reference the file in interaction responses.
/// </summary>
public async Task<UploadFileResponse> UploadFileAsync(UploadFileRequest request, CancellationToken cancellationToken = default)
{
var maxUploadSize = FileUploadHelpers.GetMaxFileUploadSize(configuration);
if (request.Data.Length > maxUploadSize)
{
throw new InvalidOperationException($"File '{request.FileName}' exceeds the maximum upload size of {maxUploadSize} bytes.");
}
if (request.InteractionId <= 0)
{
throw new InvalidOperationException("An interaction ID is required when uploading a file.");
}
if (string.IsNullOrEmpty(request.InputName))
{
throw new InvalidOperationException("An input name is required when uploading a file.");
}
var (fileId, filePath) = fileUploadStore.CreateEntry(request.FileName, request.InteractionId, request.InputName);
try
{
var destStream = new FileStream(filePath, FileMode.Create, FileAccess.Write, FileShare.None, bufferSize: 81920, useAsync: true);
await using (destStream.ConfigureAwait(false))
{
await destStream.WriteAsync(request.Data, cancellationToken).ConfigureAwait(false);
}
}
catch
{
fileUploadStore.RemoveEntry(request.InteractionId, fileId);
throw;
}
fileUploadStore.CompleteUpload(request.InteractionId, fileId);
return new UploadFileResponse { FileId = fileId };
}
public async Task<GetPipelineStepsResponse> GetPipelineStepsAsync(GetPipelineStepsRequest? request = null, CancellationToken cancellationToken = default)
{
using var activity = profilingTelemetry.StartJsonRpcServerCall(nameof(GetPipelineStepsAsync), streaming: false, request?.TraceContext);
logger.LogDebug("Resolving pipeline steps for list-steps request.");
#pragma warning disable ASPIREPIPELINES001
var pipeline = serviceProvider.GetRequiredService<IDistributedApplicationPipeline>() as DistributedApplicationPipeline
?? throw new InvalidOperationException("Pipeline is not a DistributedApplicationPipeline.");
var model = serviceProvider.GetRequiredService<DistributedApplicationModel>();
var executionContext = serviceProvider.GetRequiredService<DistributedApplicationExecutionContext>();
var pipelineContext = new PipelineContext(model, executionContext, serviceProvider, logger, cancellationToken);
var resolvedSteps = await pipeline.ResolveStepsAsync(pipelineContext).ConfigureAwait(false);
// If a target step is specified, filter to its transitive dependencies
if (!string.IsNullOrEmpty(request?.Step))
{
var stepsByName = resolvedSteps.ToDictionary(s => s.Name, StringComparer.Ordinal);
if (stepsByName.TryGetValue(request.Step, out var targetStep))
{
resolvedSteps = DistributedApplicationPipeline.ComputeTransitiveDependencies(targetStep, stepsByName);
}
else
{
var availableSteps = string.Join(", ", resolvedSteps.Select(s => $"'{s.Name}'"));
throw new InvalidOperationException(
$"Step '{request.Step}' not found in pipeline. Available steps: {availableSteps}");
}
}
var orderedSteps = DistributedApplicationPipeline.GetTopologicalOrder(resolvedSteps);
#pragma warning restore ASPIREPIPELINES001
return new GetPipelineStepsResponse
{
Steps = orderedSteps.Select(step => new PipelineStepInfo
{
Name = step.Name,
Description = step.Description,
DependsOn = [.. step.DependsOnSteps],
Tags = [.. step.Tags],
ResourceName = step.Resource?.Name
}).ToArray()
};
}
}