Merge pull request #26 from 239573049/feature/connection

Feature/connection

docker-compose.yml

@@ -1,6 +1,6 @@
 services:
   fast-gateway.service:
-    image: aidotnet/fast-gateway
+    image: registry.cn-shenzhen.aliyuncs.com/token-ai/fast-gateway
     container_name: fast-gateway
     restart: always
     build:

src/Core/Entities/Server.cs

@@ -71,4 +71,19 @@ public sealed class Server
     /// 请求超时时间（单位：秒）。默认900秒（15分钟）
     /// </summary>
     public int Timeout { get; set; } = 900;
+
+    /// <summary>
+    /// 是否启用请求级故障转移
+    /// </summary>
+    public bool EnableRequestFailover { get; set; }
+
+    /// <summary>
+    /// 请求级故障转移的连接超时（毫秒）
+    /// </summary>
+    public int FailoverConnectTimeoutMs { get; set; } = 150;
+
+    /// <summary>
+    /// 单个请求故障转移总预算（毫秒）
+    /// </summary>
+    public int FailoverBudgetMs { get; set; } = 500;
 }

src/FastGateway/Dto/ServerDto.cs

@@ -70,4 +70,19 @@ public class ServerDto
     ///     请求超时时间（单位：秒）。默认900秒（15分钟）
     /// </summary>
     public int Timeout { get; set; } = 900;
+
+    /// <summary>
+    ///     是否启用请求级故障转移
+    /// </summary>
+    public bool EnableRequestFailover { get; set; }
+
+    /// <summary>
+    ///     请求级故障转移的连接超时（毫秒）
+    /// </summary>
+    public int FailoverConnectTimeoutMs { get; set; } = 150;
+
+    /// <summary>
+    ///     单个请求故障转移总预算（毫秒）
+    /// </summary>
+    public int FailoverBudgetMs { get; set; } = 500;
 }

src/FastGateway/Gateway/FastGatewayForwarderHttpClientFactory.cs

@@ -1,25 +1,48 @@
-using System.Diagnostics;
+using FastGateway.Tunnels;
+using System.Diagnostics;
 using System.Net;
 using Yarp.ReverseProxy.Forwarder;
 
-namespace FastGateway.Gateway
-{
-    public class FastGatewayForwarderHttpClientFactory : IForwarderHttpClientFactory
-    {
-        public HttpMessageInvoker CreateClient(ForwarderHttpClientContext context)
-        {
-            var handler = new SocketsHttpHandler
-            {
-                UseProxy = false,
-                AllowAutoRedirect = false,
-                AutomaticDecompression = DecompressionMethods.None | DecompressionMethods.GZip | DecompressionMethods.Deflate | DecompressionMethods.Brotli,
-                UseCookies = false,
-                EnableMultipleHttp2Connections = true,
-                ActivityHeadersPropagator = new ReverseProxyPropagator(DistributedContextPropagator.Current),
-                ConnectTimeout = TimeSpan.FromSeconds(600),
-            };
+namespace FastGateway.Gateway;
 
-            return new HttpMessageInvoker(handler, disposeHandler: true);
+internal sealed class FastGatewayForwarderHttpClientFactory(
+    TunnelClientFactory tunnelClientFactory,
+    StandardForwarderHttpClientFactory standardForwarderHttpClientFactory)
+    : IForwarderHttpClientFactory
+{
+    private const string ClientModeMetadataKey = "FastGateway.ClientMode";
+    private const string TunnelClientMode = "Tunnel";
+
+    public HttpMessageInvoker CreateClient(ForwarderHttpClientContext context)
+    {
+        if (context.NewMetadata is not null &&
+            context.NewMetadata.TryGetValue(ClientModeMetadataKey, out var clientMode) &&
+            string.Equals(clientMode, TunnelClientMode, StringComparison.OrdinalIgnoreCase))
+        {
+            return tunnelClientFactory.CreateClient(context);
         }
+
+        return standardForwarderHttpClientFactory.CreateClient(context);
+    }
+}
+
+public sealed class StandardForwarderHttpClientFactory : ForwarderHttpClientFactory
+{
+    protected override void ConfigureHandler(ForwarderHttpClientContext context, SocketsHttpHandler handler)
+    {
+        handler.UseProxy = false;
+        handler.AllowAutoRedirect = false;
+        handler.AutomaticDecompression = DecompressionMethods.None | DecompressionMethods.GZip | DecompressionMethods.Deflate | DecompressionMethods.Brotli;
+        handler.UseCookies = false;
+        handler.ActivityHeadersPropagator = new ReverseProxyPropagator(DistributedContextPropagator.Current);
+        handler.ConnectTimeout = TimeSpan.FromSeconds(1);
+        handler.PooledConnectionLifetime = TimeSpan.FromMinutes(10);
+        handler.PooledConnectionIdleTimeout = TimeSpan.FromMinutes(2);
+        handler.ResponseDrainTimeout = TimeSpan.FromSeconds(10);
+        handler.EnableMultipleHttp2Connections = true;
+        handler.EnableMultipleHttp3Connections = false;
+        handler.MaxConnectionsPerServer = context.NewConfig?.MaxConnectionsPerServer ?? 1024;
+
+        base.ConfigureHandler(context, handler);
     }
 }

src/FastGateway/Gateway/Gateway.cs

@@ -8,6 +8,7 @@ using FastGateway.Services;
 using FastGateway.Tunnels;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Http.Features;
+using Microsoft.AspNetCore.Http.Timeouts;
 using Microsoft.AspNetCore.Server.Kestrel.Core;
 using Microsoft.AspNetCore.Server.Kestrel.Https;
 using Microsoft.AspNetCore.WebSockets;
@@ -20,6 +21,7 @@ using System.Text;
 using Yarp.ReverseProxy;
 using Yarp.ReverseProxy.Configuration;
 using Yarp.ReverseProxy.Health;
+using Yarp.ReverseProxy.LoadBalancing;
 using Yarp.ReverseProxy.Model;
 using Yarp.ReverseProxy.Forwarder;
 using Yarp.ReverseProxy.Transforms;
@@ -33,6 +35,9 @@ public static class Gateway
 {
     private const string Root = "Root";
     private const string GatewayVersionHeader = "X-FastGateway-Version";
+    private const string ClientModeMetadataKey = "FastGateway.ClientMode";
+    private const string TunnelClientMode = "Tunnel";
+    private const string StandardClientMode = "Standard";
     private static readonly ConcurrentDictionary<string, WebApplication> GatewayWebApplications = new();
 
     private static readonly DestinationConfig StaticProxyDestination = new() { Address = "http://127.0.0.1" };
@@ -381,6 +386,10 @@ public static class Gateway
             builder.Services.AddRateLimitService(rateLimits);
 
             builder.Services.AddTunnel();
+            builder.Services.AddSingleton<StandardForwarderHttpClientFactory>();
+            builder.Services.AddSingleton<FastGatewayForwarderHttpClientFactory>();
+            builder.Services.AddSingleton<IForwarderHttpClientFactory>(s => s.GetRequiredService<FastGatewayForwarderHttpClientFactory>());
+            builder.Services.AddSingleton<ConfigurationService>();
 
             if (server.StaticCompress)
                 builder.Services.AddResponseCompression();
@@ -481,12 +490,26 @@ public static class Gateway
             app.UseInitGatewayMiddleware();
 
             app.UseRequestTimeouts();
+            app.Use(async (context, next) =>
+            {
+                if (IsSseRequest(context.Request))
+                {
+                    context.Features.Get<IHttpRequestTimeoutFeature>()?.DisableTimeout();
+                    context.Response.Headers.CacheControl = "no-cache";
+                    context.Response.Headers["X-Accel-Buffering"] = "no";
+                }
+
+                await next(context);
+            });
 
             app.UseRateLimitMiddleware(rateLimits);
 
             // 黑名单默认启用（安全防护），白名单按服务开关控制
             app.UseBlacklistMiddleware(blacklistAndWhitelists, enableBlacklist: true, enableWhitelist: server.EnableWhitelist);
 
+            app.UseClusterRequestFailover(server.Id, gatewayVersion);
+            app.UseProxyErrorResponse(gatewayVersion);
+
             app.UseAbnormalIpMonitoring(server.Id);
 
             GatewayWebApplications.TryAdd(server.Id, app);
@@ -576,6 +599,50 @@ public static class Gateway
         return app;
     }
 
+    private static HttpClientConfig CreateHttpClientConfig()
+    {
+        return new HttpClientConfig
+        {
+            MaxConnectionsPerServer = 1024,
+            EnableMultipleHttp2Connections = true
+        };
+    }
+
+    private static ForwarderRequestConfig CreateHttpRequestConfig(Server server)
+    {
+        var timeoutSeconds = server.Timeout > 0 ? server.Timeout : 900;
+        if (timeoutSeconds < 600) timeoutSeconds = 600;
+
+        return new ForwarderRequestConfig
+        {
+            ActivityTimeout = TimeSpan.FromSeconds(timeoutSeconds),
+            AllowResponseBuffering = false
+        };
+    }
+
+    private static Dictionary<string, string> CreateClusterMetadata(string? service)
+    {
+        return new Dictionary<string, string>(1)
+        {
+            {
+                ClientModeMetadataKey,
+                IsTunnelService(service) ? TunnelClientMode : StandardClientMode
+            }
+        };
+    }
+
+    private static bool IsTunnelService(string? service)
+    {
+        if (string.IsNullOrWhiteSpace(service)) return false;
+        return service.StartsWith("http://node_", StringComparison.OrdinalIgnoreCase)
+               || service.StartsWith("https://node_", StringComparison.OrdinalIgnoreCase);
+    }
+
+    private static bool IsSseRequest(HttpRequest request)
+    {
+        return request.Headers.Accept.ToString().Contains("text/event-stream", StringComparison.OrdinalIgnoreCase);
+    }
+
     private static (IReadOnlyList<RouteConfig> routes, IReadOnlyList<ClusterConfig> clusters) BuildConfig(
         DomainName[] domainNames, Server server)
     {
@@ -644,7 +711,10 @@ public static class Gateway
                             config
                         }
                     },
-                    HealthCheck = CreateHealthCheckConfig(domainName)
+                    HealthCheck = CreateHealthCheckConfig(domainName),
+                    HttpClient = CreateHttpClientConfig(),
+                    HttpRequest = CreateHttpRequestConfig(server),
+                    Metadata = CreateClusterMetadata(domainName.Service)
                 };
 
                 clusters.Add(cluster);
@@ -661,7 +731,11 @@ public static class Gateway
                 {
                     ClusterId = domainName.Id,
                     Destinations = destinations,
-                    HealthCheck = CreateHealthCheckConfig(domainName)
+                    LoadBalancingPolicy = LoadBalancingPolicies.LeastRequests,
+                    HealthCheck = CreateHealthCheckConfig(domainName),
+                    HttpClient = CreateHttpClientConfig(),
+                    HttpRequest = CreateHttpRequestConfig(server),
+                    Metadata = CreateClusterMetadata(null)
                 };
 
                 clusters.Add(cluster);

src/FastGateway/Middleware/ClusterRequestFailoverMiddleware.cs

@@ -0,0 +1,392 @@
+using Core.Entities;
+using Core.Entities.Core;
+using FastGateway.Services;
+using Microsoft.AspNetCore.Http;
+using System.Collections.Concurrent;
+using System.Diagnostics;
+using System.Net;
+using System.Net.Http;
+using System.Net.Sockets;
+using Yarp.ReverseProxy;
+using Yarp.ReverseProxy.Forwarder;
+using Yarp.ReverseProxy.Health;
+using Yarp.ReverseProxy.Model;
+
+namespace FastGateway.Middleware;
+
+public sealed class ClusterRequestFailoverMiddleware
+{
+    private static readonly string[] AllowedMethods = [HttpMethods.Get, HttpMethods.Head, HttpMethods.Options];
+    private static readonly ConcurrentDictionary<int, HttpMessageInvoker> Clients = new();
+
+    private readonly RequestDelegate _next;
+    private readonly string _serverId;
+    private readonly string _gatewayVersion;
+    private readonly ILogger<ClusterRequestFailoverMiddleware> _logger;
+
+    public ClusterRequestFailoverMiddleware(
+        RequestDelegate next,
+        string serverId,
+        string gatewayVersion,
+        ILogger<ClusterRequestFailoverMiddleware> logger)
+    {
+        _next = next;
+        _serverId = serverId;
+        _gatewayVersion = gatewayVersion;
+        _logger = logger;
+    }
+
+    public async Task InvokeAsync(
+        HttpContext context,
+        ConfigurationService configurationService,
+        IProxyStateLookup proxyStateLookup,
+        IHttpForwarder httpForwarder,
+        IDestinationHealthUpdater destinationHealthUpdater)
+    {
+        if (!ShouldHandleRequest(context))
+        {
+            await _next(context);
+            return;
+        }
+
+        var server = configurationService.GetServer(_serverId);
+        if (server is null || !server.EnableRequestFailover)
+        {
+            await _next(context);
+            return;
+        }
+
+        var domainName = FindMatchedDomain(configurationService.GetDomainNamesByServerId(_serverId), context);
+        if (domainName is null)
+        {
+            await _next(context);
+            return;
+        }
+
+        if (!proxyStateLookup.TryGetCluster(domainName.Id, out var cluster) || cluster is null)
+        {
+            await _next(context);
+            return;
+        }
+
+        var candidates = GetCandidateDestinations(cluster).ToArray();
+        if (candidates.Length <= 1)
+        {
+            await _next(context);
+            return;
+        }
+
+        var connectTimeoutMs = server.FailoverConnectTimeoutMs > 0 ? server.FailoverConnectTimeoutMs : 150;
+        var budgetMs = server.FailoverBudgetMs >= connectTimeoutMs ? server.FailoverBudgetMs : 500;
+        var requestTimeoutSeconds = server.Timeout > 0 ? server.Timeout : 900;
+        var reactivationPeriod = GetReactivationPeriod(domainName);
+        var attemptedDestinationIds = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
+        var stopwatch = Stopwatch.StartNew();
+        ForwarderError lastError = ForwarderError.None;
+        Exception? lastException = null;
+
+        foreach (var destination in Shuffle(candidates))
+        {
+            if (!attemptedDestinationIds.Add(destination.DestinationId))
+            {
+                continue;
+            }
+
+            if (stopwatch.ElapsedMilliseconds > budgetMs && attemptedDestinationIds.Count > 1)
+            {
+                break;
+            }
+
+            context.Features.Set<IForwarderErrorFeature?>(null);
+            var requestConfig = new ForwarderRequestConfig
+            {
+                ActivityTimeout = TimeSpan.FromSeconds(requestTimeoutSeconds)
+            };
+            var transformer = new ClusterFailoverTransformer(domainName, server, _gatewayVersion);
+            var httpClient = GetOrCreateClient(connectTimeoutMs);
+            var destinationPrefix = destination.Model.Config.Address;
+
+            var error = await httpForwarder.SendAsync(
+                context,
+                destinationPrefix,
+                httpClient,
+                requestConfig,
+                transformer,
+                context.RequestAborted);
+
+            if (error == ForwarderError.None)
+            {
+                if (attemptedDestinationIds.Count > 1)
+                {
+                    _logger.LogInformation(
+                        "集群请求故障转移成功 ClusterId={ClusterId} DestinationId={DestinationId} Attempts={Attempts} ElapsedMs={ElapsedMs}",
+                        cluster.ClusterId,
+                        destination.DestinationId,
+                        attemptedDestinationIds.Count,
+                        stopwatch.ElapsedMilliseconds);
+                }
+                return;
+            }
+
+            var errorFeature = context.Features.Get<IForwarderErrorFeature>();
+            lastError = error;
+            lastException = errorFeature?.Exception;
+
+            if (IsRetriableTransportError(error, lastException) && !context.Response.HasStarted)
+            {
+                destinationHealthUpdater.SetPassive(cluster, destination, DestinationHealth.Unhealthy, reactivationPeriod);
+                _logger.LogWarning(
+                    lastException,
+                    "集群请求故障转移，目标切换 ClusterId={ClusterId} DestinationId={DestinationId} Error={Error} Attempt={Attempt} ElapsedMs={ElapsedMs}",
+                    cluster.ClusterId,
+                    destination.DestinationId,
+                    error,
+                    attemptedDestinationIds.Count,
+                    stopwatch.ElapsedMilliseconds);
+                continue;
+            }
+
+            return;
+        }
+
+        if (!context.Response.HasStarted)
+        {
+            context.Response.StatusCode = StatusCodes.Status504GatewayTimeout;
+            context.Response.Headers["Server"] = "FastGateway";
+            context.Response.Headers["X-FastGateway-Version"] = _gatewayVersion;
+            await context.Response.WriteAsJsonAsync(new
+            {
+                Code = StatusCodes.Status504GatewayTimeout,
+                Message = "没有可用的健康上游节点或请求级故障转移预算已耗尽",
+                Error = lastError.ToString(),
+                Detail = lastException?.Message
+            });
+        }
+    }
+
+    private static bool ShouldHandleRequest(HttpContext context)
+    {
+        if (!AllowedMethods.Contains(context.Request.Method, StringComparer.OrdinalIgnoreCase))
+        {
+            return false;
+        }
+
+        if (context.WebSockets.IsWebSocketRequest)
+        {
+            return false;
+        }
+
+        if (context.Request.Headers.Connection == "Upgrade")
+        {
+            return false;
+        }
+
+        if (context.Request.Headers.Accept.ToString().Contains("text/event-stream", StringComparison.OrdinalIgnoreCase))
+        {
+            return false;
+        }
+
+        return true;
+    }
+
+    private static DomainName? FindMatchedDomain(IEnumerable<DomainName> domainNames, HttpContext context)
+    {
+        var requestPath = context.Request.Path;
+        var requestHost = context.Request.Host.Host;
+
+        return domainNames
+            .Where(x => x is { Enable: true, ServiceType: ServiceType.ServiceCluster })
+            .Where(x => MatchHost(x, requestHost) && MatchPath(x, requestPath))
+            .OrderByDescending(GetHostPriority)
+            .ThenByDescending(x => NormalizeRoutePath(x.Path).Length)
+            .FirstOrDefault();
+    }
+
+    private static int GetHostPriority(DomainName domainName)
+    {
+        if (domainName.Domains is not { Length: > 0 })
+        {
+            return 0;
+        }
+
+        return domainName.Domains.Any(x => !x.Contains('*')) ? 2 : 1;
+    }
+
+    private static bool MatchHost(DomainName domainName, string requestHost)
+    {
+        if (domainName.Domains is not { Length: > 0 })
+        {
+            return true;
+        }
+
+        foreach (var hostPattern in domainName.Domains)
+        {
+            if (string.IsNullOrWhiteSpace(hostPattern))
+            {
+                continue;
+            }
+
+            if (hostPattern == "*")
+            {
+                return true;
+            }
+
+            if (string.Equals(hostPattern, requestHost, StringComparison.OrdinalIgnoreCase))
+            {
+                return true;
+            }
+
+            if (hostPattern.StartsWith("*.", StringComparison.Ordinal) &&
+                requestHost.EndsWith(hostPattern[1..], StringComparison.OrdinalIgnoreCase))
+            {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    private static bool MatchPath(DomainName domainName, PathString requestPath)
+    {
+        var routePath = NormalizeRoutePath(domainName.Path);
+        if (routePath == "/")
+        {
+            return true;
+        }
+
+        return requestPath.StartsWithSegments(routePath, out _);
+    }
+
+    private static string NormalizeRoutePath(string? path)
+    {
+        if (string.IsNullOrWhiteSpace(path) || path == "/")
+        {
+            return "/";
+        }
+
+        return "/" + path.Trim().Trim('/');
+    }
+
+    private static IEnumerable<DestinationState> GetCandidateDestinations(ClusterState cluster)
+    {
+        return cluster.Destinations.Values
+            .Where(x => GetEffectiveHealth(x.Health) != DestinationHealth.Unhealthy);
+    }
+
+    private static DestinationHealth GetEffectiveHealth(DestinationHealthState healthState)
+    {
+        if (healthState.Active == DestinationHealth.Unhealthy || healthState.Passive == DestinationHealth.Unhealthy)
+        {
+            return DestinationHealth.Unhealthy;
+        }
+
+        if (healthState.Active == DestinationHealth.Unknown || healthState.Passive == DestinationHealth.Unknown)
+        {
+            return DestinationHealth.Unknown;
+        }
+
+        return DestinationHealth.Healthy;
+    }
+
+    private static IEnumerable<DestinationState> Shuffle(IReadOnlyCollection<DestinationState> destinations)
+    {
+        return destinations.OrderBy(_ => Random.Shared.Next());
+    }
+
+    private static TimeSpan GetReactivationPeriod(DomainName domainName)
+    {
+        var seconds = domainName.HealthCheckIntervalSeconds;
+        if (seconds <= 0)
+        {
+            seconds = 10;
+        }
+
+        return TimeSpan.FromSeconds(seconds);
+    }
+
+    private static bool IsRetriableTransportError(ForwarderError error, Exception? exception)
+    {
+        if (exception is HttpRequestException or SocketException or TaskCanceledException or TimeoutException)
+        {
+            return true;
+        }
+
+        return error is ForwarderError.Request
+            or ForwarderError.RequestTimedOut
+            or ForwarderError.ResponseHeaders
+            or ForwarderError.RequestCreation;
+    }
+
+    private static HttpMessageInvoker GetOrCreateClient(int connectTimeoutMs)
+    {
+        return Clients.GetOrAdd(connectTimeoutMs, static timeout =>
+        {
+            var handler = new SocketsHttpHandler
+            {
+                UseProxy = false,
+                AllowAutoRedirect = false,
+                AutomaticDecompression = DecompressionMethods.None | DecompressionMethods.GZip | DecompressionMethods.Deflate | DecompressionMethods.Brotli,
+                UseCookies = false,
+                EnableMultipleHttp2Connections = true,
+                ActivityHeadersPropagator = new ReverseProxyPropagator(DistributedContextPropagator.Current),
+                ConnectTimeout = TimeSpan.FromMilliseconds(timeout),
+                PooledConnectionIdleTimeout = TimeSpan.FromMinutes(1),
+                ResponseDrainTimeout = TimeSpan.FromSeconds(30)
+            };
+
+            return new HttpMessageInvoker(handler, disposeHandler: true);
+        });
+    }
+
+    private sealed class ClusterFailoverTransformer(DomainName domainName, Server server, string gatewayVersion)
+        : HttpTransformer
+    {
+        public override async ValueTask TransformRequestAsync(
+            HttpContext httpContext,
+            HttpRequestMessage proxyRequest,
+            string destinationPrefix,
+            CancellationToken cancellationToken)
+        {
+            await base.TransformRequestAsync(httpContext, proxyRequest, destinationPrefix, cancellationToken);
+
+            var routePath = NormalizeRoutePath(domainName.Path);
+            var forwardPath = httpContext.Request.Path;
+            if (routePath != "/" && httpContext.Request.Path.StartsWithSegments(routePath, out var remaining))
+            {
+                forwardPath = remaining.HasValue ? remaining : new PathString("/");
+            }
+
+            proxyRequest.RequestUri = RequestUtilities.MakeDestinationAddress(destinationPrefix, forwardPath, httpContext.Request.QueryString);
+
+            if (domainName.Domains.Any(x => x.Contains('*')) || server.CopyRequestHost)
+            {
+                proxyRequest.Headers.Host = httpContext.Request.Host.Value;
+            }
+        }
+
+        public override async ValueTask<bool> TransformResponseAsync(
+            HttpContext httpContext,
+            HttpResponseMessage? proxyResponse,
+            CancellationToken cancellationToken)
+        {
+            var shouldCopy = await base.TransformResponseAsync(httpContext, proxyResponse, cancellationToken);
+            if (!shouldCopy)
+            {
+                return false;
+            }
+
+            httpContext.Response.Headers.Remove("Server");
+            httpContext.Response.Headers["Server"] = "FastGateway";
+            httpContext.Response.Headers["X-FastGateway-Version"] = gatewayVersion;
+            return true;
+        }
+    }
+}
+
+public static class ClusterRequestFailoverMiddlewareExtensions
+{
+    public static IApplicationBuilder UseClusterRequestFailover(this IApplicationBuilder app, string serverId, string gatewayVersion)
+    {
+        return app.UseMiddleware<ClusterRequestFailoverMiddleware>(serverId, gatewayVersion);
+    }
+}

src/FastGateway/Middleware/ProxyErrorResponseMiddleware.cs

@@ -0,0 +1,96 @@
+using Microsoft.AspNetCore.Http;
+using Yarp.ReverseProxy.Forwarder;
+
+namespace FastGateway.Middleware;
+
+public sealed class ProxyErrorResponseMiddleware
+{
+    private readonly RequestDelegate _next;
+    private readonly string _gatewayVersion;
+
+    public ProxyErrorResponseMiddleware(RequestDelegate next, string gatewayVersion)
+    {
+        _next = next;
+        _gatewayVersion = gatewayVersion;
+    }
+
+    public async Task InvokeAsync(HttpContext context)
+    {
+        await _next(context);
+
+        if (context.Response.HasStarted)
+        {
+            return;
+        }
+
+        if (context.Request.Path.StartsWithSegments("/internal/gateway", StringComparison.OrdinalIgnoreCase))
+        {
+            return;
+        }
+
+        var statusCode = context.Response.StatusCode;
+        if (statusCode is not (StatusCodes.Status502BadGateway or StatusCodes.Status503ServiceUnavailable or StatusCodes.Status504GatewayTimeout))
+        {
+            return;
+        }
+
+        var errorFeature = context.Features.Get<IForwarderErrorFeature>();
+        var error = errorFeature?.Error ?? ForwarderError.None;
+        var exception = errorFeature?.Exception;
+
+        var payload = new
+        {
+            Code = statusCode,
+            Message = GetMessage(statusCode, error),
+            Error = error == ForwarderError.None ? GetFallbackError(statusCode) : error.ToString(),
+            Detail = exception?.Message,
+            RequestId = context.TraceIdentifier,
+            Timestamp = DateTimeOffset.UtcNow,
+            Path = context.Request.Path.Value
+        };
+
+        context.Response.Clear();
+        context.Response.StatusCode = statusCode;
+        context.Response.ContentType = "application/json; charset=utf-8";
+        context.Response.Headers["Server"] = "FastGateway";
+        context.Response.Headers["X-FastGateway-Version"] = _gatewayVersion;
+        await context.Response.WriteAsJsonAsync(payload);
+    }
+
+    private static string GetMessage(int statusCode, ForwarderError error)
+    {
+        return statusCode switch
+        {
+            StatusCodes.Status503ServiceUnavailable when error == ForwarderError.NoAvailableDestinations
+                => "当前没有可用的上游服务节点",
+            StatusCodes.Status503ServiceUnavailable
+                => "上游服务暂时不可用",
+            StatusCodes.Status504GatewayTimeout when error == ForwarderError.RequestTimedOut
+                => "上游服务响应超时",
+            StatusCodes.Status504GatewayTimeout
+                => "网关转发请求超时",
+            StatusCodes.Status502BadGateway
+                => "网关转发请求失败",
+            _ => "代理请求失败"
+        };
+    }
+
+    private static string GetFallbackError(int statusCode)
+    {
+        return statusCode switch
+        {
+            StatusCodes.Status503ServiceUnavailable => "ServiceUnavailable",
+            StatusCodes.Status504GatewayTimeout => "GatewayTimeout",
+            StatusCodes.Status502BadGateway => "BadGateway",
+            _ => "ProxyError"
+        };
+    }
+}
+
+public static class ProxyErrorResponseMiddlewareExtensions
+{
+    public static IApplicationBuilder UseProxyErrorResponse(this IApplicationBuilder app, string gatewayVersion)
+    {
+        return app.UseMiddleware<ProxyErrorResponseMiddleware>(gatewayVersion);
+    }
+}

src/FastGateway/Services/ServerService.cs

@@ -41,7 +41,12 @@ public static class ServerService
                     EnableBlacklist = x.EnableBlacklist,
                     EnableTunnel = x.EnableTunnel,
                     EnableWhitelist = x.EnableWhitelist,
-                    Description = x.Description
+                    Description = x.Description,
+                    MaxRequestBodySize = x.MaxRequestBodySize,
+                    Timeout = x.Timeout,
+                    EnableRequestFailover = x.EnableRequestFailover,
+                    FailoverConnectTimeoutMs = x.FailoverConnectTimeoutMs,
+                    FailoverBudgetMs = x.FailoverBudgetMs
                 });
             })
             .WithDescription("获取服务列表")

src/FastGateway/Tunnels/AgentClientConnection.cs

@@ -115,7 +115,8 @@ public sealed partial class AgentClientConnection : IAsyncDisposable
             switch (text)
             {
                 case null:
-                    break;
+                    Log.LogClosed(_logger, ClientId);
+                    return;
 
                 case Ping:
                     Log.LogRecvPing(_logger, ClientId);

src/TunnelClient/Worker.cs

@@ -30,22 +30,21 @@ public class Worker : BackgroundService
         var serverClient = new ServerClient(monitorServer, tunnel,
             _services.GetRequiredService<ILogger<ServerClient>>());
 
-        await monitorServer.RegisterNodeAsync(tunnel, stoppingToken);
-
         while (!stoppingToken.IsCancellationRequested)
         {
-            await MonitorServerAsync(serverClient, tunnel, stoppingToken);
+            await MonitorServerAsync(monitorServer, serverClient, tunnel, stoppingToken);
             _logger.LogInformation("尝试重新连接到服务器...");
             await Task.Delay(tunnel.ReconnectInterval, stoppingToken);
             _logger.LogInformation("重新连接到服务器中...");
         }
     }
 
-    private async Task MonitorServerAsync(ServerClient serverClient,
+    private async Task MonitorServerAsync(MonitorServer monitorServer, ServerClient serverClient,
         Tunnel tunnel, CancellationToken stoppingToken)
     {
         try
         {
+            await monitorServer.RegisterNodeAsync(tunnel, stoppingToken);
             await serverClient.TransportCoreAsync(tunnel, stoppingToken);
         }
         catch (UnauthorizedAccessException e)

web/src/pages/server/features/CreateServer.tsx

@@ -49,6 +49,9 @@ export default function CreateServer({
             copyRequestHost: true,
             maxRequestBodySize: null,
             timeout: 900,
+            enableRequestFailover: false,
+            failoverConnectTimeoutMs: 150,
+            failoverBudgetMs: 500,
         }),
         []
     );
@@ -84,6 +87,19 @@ export default function CreateServer({
             return;
         }
 
+        if (value.enableRequestFailover) {
+            if (value.failoverConnectTimeoutMs <= 0) {
+                toast.error("故障转移连接超时必须大于 0");
+                setTab("limits");
+                return;
+            }
+            if (value.failoverBudgetMs < value.failoverConnectTimeoutMs) {
+                toast.error("故障转移总预算不能小于连接超时");
+                setTab("limits");
+                return;
+            }
+        }
+
         if (value.redirectHttps && !value.isHttps) {
             toast.error("启用 HTTPS 重定向时必须同时启用 HTTPS");
             setTab("features");
@@ -291,6 +307,50 @@ export default function CreateServer({
                                     />
                                 </div>
                             </div>
+
+                            <div className="grid gap-4 md:grid-cols-2">
+                                <div className="space-y-2">
+                                    <FieldLabel
+                                        htmlFor="create-server-failover-connect-timeout"
+                                        label="故障转移连接超时(ms)"
+                                        tooltip="仅在启用请求级故障转移后生效，用于快速判定单个上游节点连接失败。"
+                                    />
+                                    <Input
+                                        id="create-server-failover-connect-timeout"
+                                        type="number"
+                                        value={value.failoverConnectTimeoutMs}
+                                        onChange={(e) =>
+                                            setValue((prev) => ({
+                                                ...prev,
+                                                failoverConnectTimeoutMs: Number(e.target.value),
+                                            }))
+                                        }
+                                        placeholder="例如：150"
+                                        disabled={!value.enableRequestFailover}
+                                    />
+                                </div>
+
+                                <div className="space-y-2">
+                                    <FieldLabel
+                                        htmlFor="create-server-failover-budget"
+                                        label="故障转移总预算(ms)"
+                                        tooltip="单个请求在多个健康节点之间切换的总耗时预算。"
+                                    />
+                                    <Input
+                                        id="create-server-failover-budget"
+                                        type="number"
+                                        value={value.failoverBudgetMs}
+                                        onChange={(e) =>
+                                            setValue((prev) => ({
+                                                ...prev,
+                                                failoverBudgetMs: Number(e.target.value),
+                                            }))
+                                        }
+                                        placeholder="例如：500"
+                                        disabled={!value.enableRequestFailover}
+                                    />
+                                </div>
+                            </div>
                         </TabsContent>
 
                         <TabsContent value="features" className="space-y-4">
@@ -364,6 +424,18 @@ export default function CreateServer({
                                         }))
                                     }
                                 />
+                                <SettingSwitch
+                                    id="create-server-request-failover"
+                                    label="请求级故障转移"
+                                    description="仅对 GET/HEAD/OPTIONS 生效，网络层失败时尝试切换到其他健康节点。"
+                                    checked={value.enableRequestFailover}
+                                    onCheckedChange={(next) =>
+                                        setValue((prev) => ({
+                                            ...prev,
+                                            enableRequestFailover: next,
+                                        }))
+                                    }
+                                />
                                 <SettingSwitch
                                     id="create-server-https"
                                     label="启用 HTTPS"

web/src/pages/server/features/UpdateServer.tsx

@@ -51,6 +51,9 @@ export default function UpdateServer({
             copyRequestHost: true,
             maxRequestBodySize: null,
             timeout: 900,
+            enableRequestFailover: false,
+            failoverConnectTimeoutMs: 150,
+            failoverBudgetMs: 500,
         }),
         []
     );
@@ -86,6 +89,19 @@ export default function UpdateServer({
             return;
         }
 
+        if (value.enableRequestFailover) {
+            if (value.failoverConnectTimeoutMs <= 0) {
+                toast.error("故障转移连接超时必须大于 0");
+                setTab("limits");
+                return;
+            }
+            if (value.failoverBudgetMs < value.failoverConnectTimeoutMs) {
+                toast.error("故障转移总预算不能小于连接超时");
+                setTab("limits");
+                return;
+            }
+        }
+
         if (value.redirectHttps && !value.isHttps) {
             toast.error("启用 HTTPS 重定向时必须同时启用 HTTPS");
             setTab("features");
@@ -298,6 +314,50 @@ export default function UpdateServer({
                                     />
                                 </div>
                             </div>
+
+                            <div className="grid gap-4 md:grid-cols-2">
+                                <div className="space-y-2">
+                                    <FieldLabel
+                                        htmlFor="update-server-failover-connect-timeout"
+                                        label="故障转移连接超时(ms)"
+                                        tooltip="仅在启用请求级故障转移后生效，用于快速判定单个上游节点连接失败。"
+                                    />
+                                    <Input
+                                        id="update-server-failover-connect-timeout"
+                                        type="number"
+                                        value={value.failoverConnectTimeoutMs}
+                                        onChange={(e) =>
+                                            setValue((prev) => ({
+                                                ...prev,
+                                                failoverConnectTimeoutMs: Number(e.target.value),
+                                            }))
+                                        }
+                                        placeholder="例如：150"
+                                        disabled={!value.enableRequestFailover}
+                                    />
+                                </div>
+
+                                <div className="space-y-2">
+                                    <FieldLabel
+                                        htmlFor="update-server-failover-budget"
+                                        label="故障转移总预算(ms)"
+                                        tooltip="单个请求在多个健康节点之间切换的总耗时预算。"
+                                    />
+                                    <Input
+                                        id="update-server-failover-budget"
+                                        type="number"
+                                        value={value.failoverBudgetMs}
+                                        onChange={(e) =>
+                                            setValue((prev) => ({
+                                                ...prev,
+                                                failoverBudgetMs: Number(e.target.value),
+                                            }))
+                                        }
+                                        placeholder="例如：500"
+                                        disabled={!value.enableRequestFailover}
+                                    />
+                                </div>
+                            </div>
                         </TabsContent>
 
                         <TabsContent value="features" className="space-y-4">
@@ -371,6 +431,18 @@ export default function UpdateServer({
                                         }))
                                     }
                                 />
+                                <SettingSwitch
+                                    id="update-server-request-failover"
+                                    label="请求级故障转移"
+                                    description="仅对 GET/HEAD/OPTIONS 生效，网络层失败时尝试切换到其他健康节点。"
+                                    checked={value.enableRequestFailover}
+                                    onCheckedChange={(next) =>
+                                        setValue((prev) => ({
+                                            ...prev,
+                                            enableRequestFailover: next,
+                                        }))
+                                    }
+                                />
                                 <SettingSwitch
                                     id="update-server-https"
                                     label="启用 HTTPS"

web/src/types/index.ts

@@ -14,6 +14,9 @@ export interface Server {
     copyRequestHost: boolean;
     maxRequestBodySize: number | null;
     timeout: number;
+    enableRequestFailover: boolean;
+    failoverConnectTimeoutMs: number;
+    failoverBudgetMs: number;
 }