From 74e296703a1be763b7f5c2c01d55b42d498498ab Mon Sep 17 00:00:00 2001 From: Tiantianr <260026737+Tiantianr@users.noreply.github.com> Date: Tue, 14 Jul 2026 22:40:05 +0800 Subject: [PATCH] fix(openai-ws): make first-message timeout configurable Add a dedicated client first-message timeout while preserving the legacy 30-second default. Use the resolved value for both the WebSocket read deadline and structured timeout logs, and document tuning for large requests or slow links. Add configuration, validation, handler, and resolver regression coverage. Refs #4158 --- README.md | 7 ++++ backend/internal/config/config.go | 10 ++++++ backend/internal/config/config_test.go | 26 ++++++++++++++ .../openai_gateway_compact_log_test.go | 14 ++++++++ .../handler/openai_gateway_handler.go | 5 +-- .../handler/openai_gateway_handler_test.go | 34 +++++++++++++++++++ .../internal/service/openai_ws_http_bridge.go | 9 +++++ .../service/openai_ws_http_bridge_test.go | 11 ++++++ deploy/config.example.yaml | 2 ++ 9 files changed, 116 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 00f021ea2c..10b0475b6e 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 1262845cea..1d330b74de 100644 --- a/backend/internal/config/config.go +++ b/backend/internal/config/config.go @@ -919,6 +919,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 { @@ -926,6 +929,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"` @@ -1958,6 +1964,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) @@ -2740,6 +2747,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 2f9defb80e..1853a790c3 100644 --- a/backend/internal/config/config_test.go +++ b/backend/internal/config/config_test.go @@ -199,6 +199,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) } @@ -207,6 +214,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) @@ -1682,6 +1698,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 231363a8a3..5ce2d1d1d1 100644 --- a/backend/internal/handler/openai_gateway_handler.go +++ b/backend/internal/handler/openai_gateway_handler.go @@ -1320,7 +1320,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 { @@ -1335,7 +1336,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 e4b594c0b8..c31e3a305f 100644 --- a/backend/internal/handler/openai_gateway_handler_test.go +++ b/backend/internal/handler/openai_gateway_handler_test.go @@ -741,6 +741,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 dbc24c850a..721e6ac046 100644 --- a/backend/internal/service/openai_ws_http_bridge.go +++ b/backend/internal/service/openai_ws_http_bridge.go @@ -23,6 +23,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 4cf759497f..bff9b0f4e1 100644 --- a/deploy/config.example.yaml +++ b/deploy/config.example.yaml @@ -257,6 +257,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.