diff --git a/README.md b/README.md index ada3039897..b06839745f 100644 --- a/README.md +++ b/README.md @@ -611,12 +611,19 @@ per-turn user and account concurrency slots, which are released between turns. ```yaml gateway: openai_ws: + # Total time to receive and decompress the first client message. + client_first_message_timeout_seconds: 30 # Close a client socket idle between completed turns; 0 disables this safeguard. ingress_inter_turn_idle_timeout_seconds: 300 # Distributed API-key limit for live client ingress sessions; 0 disables it. max_ingress_connections_per_api_key: 64 ``` +The first-message timeout is a total read deadline. Deployments that accept +large contexts or image-heavy requests over slower links can raise it to +120-300 seconds. It expires before HTTP bridge routing, so bridge mode does not +override this limit. + The connection cap is coordinated through Redis using a 60-second lease that is refreshed every 20 seconds. A process that cannot confirm a lease for a full lease lifetime closes its local WebSocket rather than continuing outside diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go index 857171d0ae..40947ef952 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -924,6 +924,9 @@ func (c *UserMessageQueueConfig) GetEffectiveMode() string { return "" } +// DefaultOpenAIWSClientFirstMessageTimeoutSeconds preserves the legacy ingress deadline. +const DefaultOpenAIWSClientFirstMessageTimeoutSeconds = 30 + // GatewayOpenAIWSConfig OpenAI Responses WebSocket 配置。 // 注意:默认全局开启;如需回滚可使用 force_http 或关闭 enabled。 type GatewayOpenAIWSConfig struct { @@ -931,6 +934,9 @@ type GatewayOpenAIWSConfig struct { ModeRouterV2Enabled bool `mapstructure:"mode_router_v2_enabled"` // IngressModeDefault: ingress 默认模式(off/ctx_pool/passthrough/http_bridge) IngressModeDefault string `mapstructure:"ingress_mode_default"` + // ClientFirstMessageTimeoutSeconds bounds the total time to read and decompress + // the first client response.create message after the WebSocket upgrade. + ClientFirstMessageTimeoutSeconds int `mapstructure:"client_first_message_timeout_seconds"` // IngressInterTurnIdleTimeoutSeconds bounds the time a client may remain idle // between completed ingress turns. Zero disables this protection. IngressInterTurnIdleTimeoutSeconds int `mapstructure:"ingress_inter_turn_idle_timeout_seconds"` @@ -1965,6 +1971,7 @@ func setDefaults() { viper.SetDefault("gateway.openai_ws.enabled", true) viper.SetDefault("gateway.openai_ws.mode_router_v2_enabled", false) viper.SetDefault("gateway.openai_ws.ingress_mode_default", "ctx_pool") + viper.SetDefault("gateway.openai_ws.client_first_message_timeout_seconds", DefaultOpenAIWSClientFirstMessageTimeoutSeconds) viper.SetDefault("gateway.openai_ws.ingress_inter_turn_idle_timeout_seconds", 300) viper.SetDefault("gateway.openai_ws.max_ingress_connections_per_api_key", 64) viper.SetDefault("gateway.openai_ws.oauth_enabled", true) @@ -2755,6 +2762,9 @@ func (c *Config) Validate() error { if c.Gateway.OpenAIWS.MaxConnsPerAccount <= 0 { return fmt.Errorf("gateway.openai_ws.max_conns_per_account must be positive") } + if c.Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds <= 0 { + return fmt.Errorf("gateway.openai_ws.client_first_message_timeout_seconds must be positive") + } if c.Gateway.OpenAIWS.IngressInterTurnIdleTimeoutSeconds < 0 { return fmt.Errorf("gateway.openai_ws.ingress_inter_turn_idle_timeout_seconds must be non-negative") } diff --git a/backend/internal/config/config_test.go b/backend/internal/config/config_test.go index 2a0e3a2580..9a2d725eed 100644 --- a/backend/internal/config/config_test.go +++ b/backend/internal/config/config_test.go @@ -227,6 +227,13 @@ func TestLoadDefaultOpenAIWSConfig(t *testing.T) { if cfg.Gateway.OpenAIWS.IngressModeDefault != "ctx_pool" { t.Fatalf("Gateway.OpenAIWS.IngressModeDefault = %q, want %q", cfg.Gateway.OpenAIWS.IngressModeDefault, "ctx_pool") } + if cfg.Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds != DefaultOpenAIWSClientFirstMessageTimeoutSeconds { + t.Fatalf( + "Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds = %d, want %d", + cfg.Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds, + DefaultOpenAIWSClientFirstMessageTimeoutSeconds, + ) + } if cfg.Gateway.OpenAIWS.IngressInterTurnIdleTimeoutSeconds != 300 { t.Fatalf("Gateway.OpenAIWS.IngressInterTurnIdleTimeoutSeconds = %d, want 300", cfg.Gateway.OpenAIWS.IngressInterTurnIdleTimeoutSeconds) } @@ -235,6 +242,15 @@ func TestLoadDefaultOpenAIWSConfig(t *testing.T) { } } +func TestLoadOpenAIWSClientFirstMessageTimeoutFromEnv(t *testing.T) { + resetViperWithJWTSecret(t) + t.Setenv("GATEWAY_OPENAI_WS_CLIENT_FIRST_MESSAGE_TIMEOUT_SECONDS", "120") + + cfg, err := Load() + require.NoError(t, err) + require.Equal(t, 120, cfg.Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds) +} + func TestLoadDefaultOpenAICompactModel(t *testing.T) { resetViperWithJWTSecret(t) @@ -1720,6 +1736,16 @@ func TestValidateConfig_OpenAIWSRules(t *testing.T) { mutate: func(c *Config) { c.Gateway.OpenAIWS.MaxConnsPerAccount = 0 }, wantErr: "gateway.openai_ws.max_conns_per_account", }, + { + name: "client_first_message_timeout_seconds 必须为正数", + mutate: func(c *Config) { c.Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds = 0 }, + wantErr: "gateway.openai_ws.client_first_message_timeout_seconds", + }, + { + name: "client_first_message_timeout_seconds 不能为负数", + mutate: func(c *Config) { c.Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds = -1 }, + wantErr: "gateway.openai_ws.client_first_message_timeout_seconds", + }, { name: "ingress_inter_turn_idle_timeout_seconds 不能为负数", mutate: func(c *Config) { c.Gateway.OpenAIWS.IngressInterTurnIdleTimeoutSeconds = -1 }, diff --git a/backend/internal/handler/openai_gateway_compact_log_test.go b/backend/internal/handler/openai_gateway_compact_log_test.go index e18509b4b3..16afd30c31 100644 --- a/backend/internal/handler/openai_gateway_compact_log_test.go +++ b/backend/internal/handler/openai_gateway_compact_log_test.go @@ -66,6 +66,20 @@ func (s *handlerInMemoryLogSink) ContainsFieldValue(field, substr string) bool { return false } +func (s *handlerInMemoryLogSink) FieldValueForMessage(message, field string) (any, bool) { + s.mu.Lock() + defer s.mu.Unlock() + for _, event := range s.events { + if event == nil || event.Message != message || event.Fields == nil { + continue + } + if value, ok := event.Fields[field]; ok { + return value, true + } + } + return nil, false +} + func captureHandlerStructuredLog(t *testing.T) (*handlerInMemoryLogSink, func()) { t.Helper() handlerStructuredLogCaptureMu.Lock() diff --git a/backend/internal/handler/openai_gateway_handler.go b/backend/internal/handler/openai_gateway_handler.go index cd18aa7a1b..83c4b4ac23 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -1403,7 +1403,8 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { c.Request = c.Request.WithContext(ctx) } - readCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + firstMessageTimeout := service.ResolveOpenAIWSClientFirstMessageTimeout(h.cfg) + readCtx, cancel := context.WithTimeout(ctx, firstMessageTimeout) msgType, firstMessage, err := wsConn.Read(readCtx) cancel() if err != nil { @@ -1418,7 +1419,7 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) { zap.String("client_ip", clientIP), zap.String("close_status", closeStatus), zap.String("close_reason", closeReason), - zap.Duration("read_timeout", 30*time.Second), + zap.Duration("read_timeout", firstMessageTimeout), ) closeOpenAIClientWS(wsConn, coderws.StatusPolicyViolation, "missing first response.create message") return diff --git a/backend/internal/handler/openai_gateway_handler_test.go b/backend/internal/handler/openai_gateway_handler_test.go index 21f5b4686c..d855f364a7 100644 --- a/backend/internal/handler/openai_gateway_handler_test.go +++ b/backend/internal/handler/openai_gateway_handler_test.go @@ -833,6 +833,40 @@ func TestOpenAIResponsesWebSocket_IngressCapacityRejected(t *testing.T) { require.Equal(t, coderws.StatusTryAgainLater, closeErr.Code) } +func TestOpenAIResponsesWebSocket_FirstMessageTimeoutUsesConfig(t *testing.T) { + gin.SetMode(gin.TestMode) + logSink, restore := captureHandlerStructuredLog(t) + defer restore() + + h := newOpenAIHandlerForPreviousResponseIDValidation(t, nil) + h.cfg = &config.Config{} + h.cfg.Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds = 1 + wsServer := newOpenAIWSHandlerTestServer(t, h, middleware.AuthSubject{UserID: 1, Concurrency: 1}) + defer wsServer.Close() + + dialCtx, cancelDial := context.WithTimeout(context.Background(), 3*time.Second) + clientConn, _, err := coderws.Dial(dialCtx, "ws"+strings.TrimPrefix(wsServer.URL, "http")+"/openai/v1/responses", nil) + cancelDial() + require.NoError(t, err) + defer func() { _ = clientConn.CloseNow() }() + + started := time.Now() + readCtx, cancelRead := context.WithTimeout(context.Background(), 5*time.Second) + _, _, err = clientConn.Read(readCtx) + cancelRead() + elapsed := time.Since(started) + + require.Error(t, err) + require.NotErrorIs(t, err, context.DeadlineExceeded) + require.GreaterOrEqual(t, elapsed, 500*time.Millisecond) + require.Less(t, elapsed, 4*time.Second) + require.Eventually(t, func() bool { + readTimeout, ok := logSink.FieldValueForMessage("openai.websocket_read_first_message_failed", "read_timeout") + return ok && readTimeout == time.Second && + logSink.ContainsMessageAtLevel("openai.websocket_read_first_message_failed", "warn") + }, time.Second, 10*time.Millisecond) +} + func TestOpenAIResponsesWebSocket_IngressLeaseReleasedOnEarlyReturn(t *testing.T) { gin.SetMode(gin.TestMode) cache := &concurrencyCacheMock{ diff --git a/backend/internal/service/openai_ws_http_bridge.go b/backend/internal/service/openai_ws_http_bridge.go index 77a5e59329..091b3592d1 100644 --- a/backend/internal/service/openai_ws_http_bridge.go +++ b/backend/internal/service/openai_ws_http_bridge.go @@ -22,6 +22,15 @@ const ( openAIWSHTTPBridgeErrorBodyLimitBytes = 64 * 1024 ) +// ResolveOpenAIWSClientFirstMessageTimeout returns the effective client ingress deadline. +func ResolveOpenAIWSClientFirstMessageTimeout(cfg *config.Config) time.Duration { + seconds := config.DefaultOpenAIWSClientFirstMessageTimeoutSeconds + if cfg != nil && cfg.Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds > 0 { + seconds = cfg.Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds + } + return time.Duration(seconds) * time.Second +} + func ResolveOpenAIWSClientReadLimitBytes(cfg *config.Config) int64 { if cfg == nil || cfg.Gateway.OpenAIWS.ClientReadLimitBytes <= 0 { return openAIWSClientReadLimitBytesDefault diff --git a/backend/internal/service/openai_ws_http_bridge_test.go b/backend/internal/service/openai_ws_http_bridge_test.go index b1bfb4d342..cbc11abcc2 100644 --- a/backend/internal/service/openai_ws_http_bridge_test.go +++ b/backend/internal/service/openai_ws_http_bridge_test.go @@ -19,6 +19,17 @@ import ( "github.com/tidwall/gjson" ) +func TestResolveOpenAIWSClientFirstMessageTimeout(t *testing.T) { + defaultTimeout := time.Duration(config.DefaultOpenAIWSClientFirstMessageTimeoutSeconds) * time.Second + require.Equal(t, defaultTimeout, ResolveOpenAIWSClientFirstMessageTimeout(nil)) + + cfg := &config.Config{} + require.Equal(t, defaultTimeout, ResolveOpenAIWSClientFirstMessageTimeout(cfg)) + + cfg.Gateway.OpenAIWS.ClientFirstMessageTimeoutSeconds = 120 + require.Equal(t, 120*time.Second, ResolveOpenAIWSClientFirstMessageTimeout(cfg)) +} + func TestPrepareOpenAIWSHTTPBridgeBodyStripsWSFields(t *testing.T) { body, err := prepareOpenAIWSHTTPBridgeBody([]byte(`{"type":"response.create","generate":true,"model":"gpt-5","stream":false,"previous_response_id":"resp_prev","input":"hi"}`)) require.NoError(t, err) diff --git a/deploy/config.example.yaml b/deploy/config.example.yaml index fdeccc00c3..7b98993c53 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -266,6 +266,8 @@ gateway: # ingress 默认模式:off|ctx_pool|passthrough|http_bridge(仅 mode_router_v2_enabled=true 生效) # 兼容旧值:shared/dedicated 会按 ctx_pool 处理。 ingress_mode_default: ctx_pool + # 完整读取并解压首条客户端消息的总超时(秒);大请求或慢链路可调高到 120-300。 + client_first_message_timeout_seconds: 30 # Close a client WebSocket that stays idle between completed turns (seconds). Set 0 to disable. ingress_inter_turn_idle_timeout_seconds: 300 # Limit live client WebSocket ingress sessions per API key across all instances. Set 0 to disable.