diff --git a/apps/api/docs/swagger.json b/apps/api/docs/swagger.json index 4a77aad..bc5222c 100644 --- a/apps/api/docs/swagger.json +++ b/apps/api/docs/swagger.json @@ -5208,6 +5208,12 @@ "schema": { "$ref": "#/definitions/httpapi.ErrorEnvelope" } + }, + "503": { + "description": "Service Unavailable", + "schema": { + "$ref": "#/definitions/httpapi.ErrorEnvelope" + } } } } diff --git a/apps/api/docs/swagger.yaml b/apps/api/docs/swagger.yaml index cb7e255..b7271b8 100644 --- a/apps/api/docs/swagger.yaml +++ b/apps/api/docs/swagger.yaml @@ -6441,6 +6441,10 @@ paths: description: Internal Server Error schema: $ref: '#/definitions/httpapi.ErrorEnvelope' + "503": + description: Service Unavailable + schema: + $ref: '#/definitions/httpapi.ErrorEnvelope' summary: 本地登录 tags: - auth diff --git a/apps/api/internal/httpapi/availability_timeout_integration_test.go b/apps/api/internal/httpapi/availability_timeout_integration_test.go new file mode 100644 index 0000000..964627c --- /dev/null +++ b/apps/api/internal/httpapi/availability_timeout_integration_test.go @@ -0,0 +1,111 @@ +package httpapi + +import ( + "bytes" + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "os" + "strings" + "testing" + "time" + + "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" +) + +func TestReadyReturnsPostgresUnavailableWithinTwoSeconds(t *testing.T) { + db := newExhaustedPostgresStore(t) + server := &Server{store: db, logger: slog.New(slog.NewJSONHandler(io.Discard, nil))} + requestContext, cancel := context.WithTimeout(context.Background(), 4*time.Second) + defer cancel() + request := httptest.NewRequest(http.MethodGet, "/readyz", nil).WithContext(requestContext) + recorder := httptest.NewRecorder() + + startedAt := time.Now() + server.ready(recorder, request) + elapsed := time.Since(startedAt) + + assertUnavailableResponse(t, recorder, "POSTGRES_UNAVAILABLE", "postgres unavailable") + if elapsed > 3*time.Second { + t.Fatalf("readiness timeout took %s, want no more than 3s", elapsed) + } +} + +func TestLoginReturnsAuthStoreUnavailableWithinFiveSeconds(t *testing.T) { + db := newExhaustedPostgresStore(t) + var logs bytes.Buffer + server := &Server{store: db, logger: slog.New(slog.NewJSONHandler(&logs, nil))} + requestContext, cancel := context.WithTimeout(context.Background(), 7*time.Second) + defer cancel() + request := httptest.NewRequest(http.MethodPost, "/api/v1/auth/login", strings.NewReader(`{"account":"timeout-test-account","password":"timeout-test-password"}`)).WithContext(requestContext) + recorder := httptest.NewRecorder() + + startedAt := time.Now() + server.login(recorder, request) + elapsed := time.Since(startedAt) + + assertUnavailableResponse(t, recorder, "AUTH_STORE_UNAVAILABLE", "authentication service temporarily unavailable") + if elapsed > 6*time.Second { + t.Fatalf("login timeout took %s, want no more than 6s", elapsed) + } + logOutput := logs.String() + for _, field := range []string{"postgres_pool_max_connections", "postgres_pool_acquired_connections", "postgres_pool_idle_connections", "postgres_pool_empty_acquire_count", "postgres_pool_canceled_acquire_count"} { + if !strings.Contains(logOutput, field) { + t.Fatalf("login failure log did not include %q: %s", field, logOutput) + } + } + if strings.Contains(logOutput, "timeout-test-account") || strings.Contains(logOutput, "timeout-test-password") { + t.Fatalf("login failure log exposed credentials: %s", logOutput) + } +} + +func newExhaustedPostgresStore(t *testing.T) *store.Store { + t.Helper() + databaseURL := strings.TrimSpace(os.Getenv("AI_GATEWAY_TEST_DATABASE_URL")) + if databaseURL == "" { + t.Skip("set AI_GATEWAY_TEST_DATABASE_URL to run PostgreSQL availability timeout tests") + } + parsed, err := url.Parse(databaseURL) + if err != nil { + t.Fatalf("parse test database URL: %v", err) + } + query := parsed.Query() + query.Set("pool_max_conns", "1") + query.Set("pool_min_conns", "0") + parsed.RawQuery = query.Encode() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + db, err := store.Connect(ctx, parsed.String()) + if err != nil { + t.Fatalf("connect timeout test store: %v", err) + } + t.Cleanup(db.Close) + connection, err := db.Pool().Acquire(ctx) + if err != nil { + t.Fatalf("exhaust timeout test pool: %v", err) + } + t.Cleanup(connection.Release) + return db +} + +func assertUnavailableResponse(t *testing.T, recorder *httptest.ResponseRecorder, expectedCode, expectedMessage string) { + t.Helper() + if recorder.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want 503; body=%s", recorder.Code, recorder.Body.String()) + } + var envelope ErrorEnvelope + if err := json.Unmarshal(recorder.Body.Bytes(), &envelope); err != nil { + t.Fatalf("decode unavailable response: %v", err) + } + if envelope.Error.Code != expectedCode { + t.Fatalf("error code = %q, want %q; body=%s", envelope.Error.Code, expectedCode, recorder.Body.String()) + } + if envelope.Error.Message != expectedMessage { + t.Fatalf("error message = %q, want %q; body=%s", envelope.Error.Message, expectedMessage, recorder.Body.String()) + } +} diff --git a/apps/api/internal/httpapi/handlers.go b/apps/api/internal/httpapi/handlers.go index 4d510f3..0e29b08 100644 --- a/apps/api/internal/httpapi/handlers.go +++ b/apps/api/internal/httpapi/handlers.go @@ -17,6 +17,14 @@ import ( "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) +const ( + postgresReadinessTimeout = 2 * time.Second + localLoginStoreTimeout = 5 * time.Second + errorCodePostgresDown = "POSTGRES_UNAVAILABLE" + errorCodeAuthStoreDown = "AUTH_STORE_UNAVAILABLE" + authStoreUnavailableError = "authentication service temporarily unavailable" +) + // health godoc // @Summary 健康检查 // @Description 返回服务进程、运行环境和身份模式，供负载均衡或人工排障使用。 @@ -42,8 +50,11 @@ func (s *Server) health(w http.ResponseWriter, r *http.Request) { // @Failure 503 {object} ErrorEnvelope // @Router /readyz [get] func (s *Server) ready(w http.ResponseWriter, r *http.Request) { - if err := s.store.Ping(r.Context()); err != nil { - writeError(w, http.StatusServiceUnavailable, "postgres unavailable") + ctx, cancel := context.WithTimeout(r.Context(), postgresReadinessTimeout) + defer cancel() + if err := s.store.Ping(ctx); err != nil { + s.logPostgresUnavailable("postgres readiness check failed") + writeError(w, http.StatusServiceUnavailable, "postgres unavailable", errorCodePostgresDown) return } writeJSON(w, http.StatusOK, map[string]any{"ok": true}) @@ -121,6 +132,7 @@ func (s *Server) register(w http.ResponseWriter, r *http.Request) { // @Failure 401 {object} ErrorEnvelope // @Failure 403 {object} ErrorEnvelope // @Failure 500 {object} ErrorEnvelope +// @Failure 503 {object} ErrorEnvelope // @Router /api/v1/auth/login [post] func (s *Server) login(w http.ResponseWriter, r *http.Request) { var input store.LocalLoginInput @@ -128,12 +140,19 @@ func (s *Server) login(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "invalid json body") return } - user, err := s.store.AuthenticateLocalUser(r.Context(), input) + ctx, cancel := context.WithTimeout(r.Context(), localLoginStoreTimeout) + defer cancel() + user, err := s.store.AuthenticateLocalUser(ctx, input) if err != nil { if errors.Is(err, store.ErrInvalidCredentials) { writeError(w, http.StatusUnauthorized, "invalid account or password") return } + if store.IsPostgresUnavailable(err) { + s.logPostgresUnavailable("login authentication store unavailable") + writeError(w, http.StatusServiceUnavailable, authStoreUnavailableError, errorCodeAuthStoreDown) + return + } s.logger.Error("login local user failed", "error", err) writeError(w, http.StatusInternalServerError, "login failed") return @@ -145,6 +164,22 @@ func (s *Server) login(w http.ResponseWriter, r *http.Request) { s.writeAuthResponse(w, http.StatusOK, user) } +func (s *Server) logPostgresUnavailable(message string) { + if s.logger == nil || s.store == nil || s.store.Pool() == nil { + return + } + statistics := s.store.Pool().Stat() + s.logger.Error(message, + "error_category", "postgres_unavailable", + "postgres_pool_max_connections", statistics.MaxConns(), + "postgres_pool_total_connections", statistics.TotalConns(), + "postgres_pool_acquired_connections", statistics.AcquiredConns(), + "postgres_pool_idle_connections", statistics.IdleConns(), + "postgres_pool_empty_acquire_count", statistics.EmptyAcquireCount(), + "postgres_pool_canceled_acquire_count", statistics.CanceledAcquireCount(), + ) +} + func (s *Server) localIdentityEnabled() bool { mode := strings.ToLower(strings.TrimSpace(s.cfg.IdentityMode)) return mode == "" || mode == "standalone" || mode == "hybrid" diff --git a/apps/api/internal/store/postgres.go b/apps/api/internal/store/postgres.go index 08c5c21..ee17719 100644 --- a/apps/api/internal/store/postgres.go +++ b/apps/api/internal/store/postgres.go @@ -7,6 +7,7 @@ import ( "encoding/base64" "encoding/json" "errors" + "net" "strings" "time" "unicode" @@ -22,6 +23,11 @@ type Store struct { pool *pgxpool.Pool } +const ( + postgresApplicationName = "easyai-ai-gateway" + postgresConnectTimeout = 5 * time.Second +) + func defaultAPIKeyScopes() []string { return []string{"chat", "embedding", "rerank", "image", "video", "music", "audio", "voice_clone"} } @@ -67,7 +73,11 @@ var ( ) func Connect(ctx context.Context, databaseURL string) (*Store, error) { - pool, err := pgxpool.New(ctx, databaseURL) + config, err := postgresPoolConfig(databaseURL) + if err != nil { + return nil, err + } + pool, err := pgxpool.NewWithConfig(ctx, config) if err != nil { return nil, err } @@ -78,6 +88,38 @@ func Connect(ctx context.Context, databaseURL string) (*Store, error) { return &Store{pool: pool}, nil } +func postgresPoolConfig(databaseURL string) (*pgxpool.Config, error) { + config, err := pgxpool.ParseConfig(databaseURL) + if err != nil { + return nil, err + } + config.ConnConfig.ConnectTimeout = postgresConnectTimeout + config.ConnConfig.RuntimeParams["application_name"] = postgresApplicationName + return config, nil +} + +func IsPostgresUnavailable(err error) bool { + if err == nil { + return false + } + if errors.Is(err, context.DeadlineExceeded) { + return true + } + var connectError *pgconn.ConnectError + if errors.As(err, &connectError) { + return true + } + var networkError net.Error + if errors.As(err, &networkError) { + return true + } + var postgresError *pgconn.PgError + if errors.As(err, &postgresError) { + return strings.HasPrefix(postgresError.Code, "08") || postgresError.Code == "53300" || strings.HasPrefix(postgresError.Code, "57P0") + } + return pgconn.SafeToRetry(err) +} + func (s *Store) Close() { s.pool.Close() } diff --git a/apps/api/internal/store/postgres_config_test.go b/apps/api/internal/store/postgres_config_test.go new file mode 100644 index 0000000..622e055 --- /dev/null +++ b/apps/api/internal/store/postgres_config_test.go @@ -0,0 +1,49 @@ +package store + +import ( + "context" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" +) + +func TestPostgresPoolConfigSetsDiagnosticAndConnectTimeout(t *testing.T) { + config, err := postgresPoolConfig("postgresql://gateway:password@localhost:5432/gateway?sslmode=disable") + if err != nil { + t.Fatalf("parse PostgreSQL pool config: %v", err) + } + if config.ConnConfig.ConnectTimeout != 5*time.Second { + t.Fatalf("connect timeout = %s, want 5s", config.ConnConfig.ConnectTimeout) + } + if applicationName := config.ConnConfig.RuntimeParams["application_name"]; applicationName != "easyai-ai-gateway" { + t.Fatalf("application_name = %q, want easyai-ai-gateway", applicationName) + } +} + +func TestPostgresPoolConfigRejectsMalformedURL(t *testing.T) { + if _, err := postgresPoolConfig("://malformed"); err == nil { + t.Fatal("expected malformed PostgreSQL URL to fail") + } +} + +func TestIsPostgresUnavailableClassifiesConnectivityFailures(t *testing.T) { + for _, testCase := range []struct { + name string + err error + }{ + {name: "deadline", err: context.DeadlineExceeded}, + {name: "connection exception", err: &pgconn.PgError{Code: "08006"}}, + {name: "too many connections", err: &pgconn.PgError{Code: "53300"}}, + {name: "cannot connect now", err: &pgconn.PgError{Code: "57P03"}}, + } { + t.Run(testCase.name, func(t *testing.T) { + if !IsPostgresUnavailable(testCase.err) { + t.Fatalf("error %v was not classified as PostgreSQL unavailable", testCase.err) + } + }) + } + if IsPostgresUnavailable(&pgconn.PgError{Code: "42601"}) { + t.Fatal("SQL syntax error was incorrectly classified as PostgreSQL unavailable") + } +} diff --git a/apps/web/src/api.test.ts b/apps/web/src/api.test.ts index bdf8210..470cc3b 100644 --- a/apps/web/src/api.test.ts +++ b/apps/web/src/api.test.ts @@ -9,12 +9,39 @@ import { getAPITask, getCurrentUser, getOpsManagementSkillMetadata, + loginLocalAccount, OIDC_BROWSER_SESSION_CREDENTIAL, startIdentityPairing, retireIdentityPairingSecurityEventConflict, validateIdentityRevision, } from './api'; +describe('local login transport', () => { + afterEach(() => { + vi.useRealTimers(); + vi.unstubAllGlobals(); + }); + + it('aborts after ten seconds and returns a stable login timeout message', async () => { + vi.useFakeTimers(); + const fetchMock = vi.fn((_url: string, init?: RequestInit) => new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => reject(new DOMException('aborted', 'AbortError'))); + })); + vi.stubGlobal('fetch', fetchMock); + + const login = loginLocalAccount({ account: 'timeout-test-account', password: 'timeout-test-password' }); + const rejection = expect(login).rejects.toMatchObject({ + message: '登录请求超时，请稍后重试', + }); + await vi.advanceTimersByTimeAsync(10_000); + await rejection; + + const [, init] = fetchMock.mock.calls[0] as [string, RequestInit]; + expect(init.signal).toBeInstanceOf(AbortSignal); + expect(init.signal?.aborted).toBe(true); + }); +}); + describe('Gateway provisioning errors', () => { const cases = [ ['GATEWAY_USER_NOT_PROVISIONED', '该账号尚未开通 EasyAI Gateway'], diff --git a/apps/web/src/api.ts b/apps/web/src/api.ts index 834b372..aa79f36 100644 --- a/apps/web/src/api.ts +++ b/apps/web/src/api.ts @@ -121,6 +121,7 @@ export async function loginLocalAccount(input: { account: string; password: stri auth: false, body: input, method: 'POST', + timeoutMs: 10_000, }); } @@ -1206,7 +1207,7 @@ export async function deleteFileStorageChannel(token: string, channelId: string) async function request( path: string, - options: { token?: string; auth?: boolean; method?: string; body?: unknown; headers?: Record; signal?: AbortSignal } = {}, + options: { token?: string; auth?: boolean; method?: string; body?: unknown; headers?: Record; signal?: AbortSignal; timeoutMs?: number } = {}, ): Promise { const headers: Record = { ...(options.headers ?? {}) }; if (options.auth !== false && options.token && options.token !== OIDC_BROWSER_SESSION_CREDENTIAL) { @@ -1215,21 +1216,35 @@ async function request( if (options.body !== undefined) { headers['Content-Type'] = 'application/json'; } - const response = await fetch(`${API_BASE}${path}`, { - method: options.method ?? 'GET', - headers, - body: options.body === undefined ? undefined : JSON.stringify(options.body), - credentials: 'include', - signal: options.signal, - }); - if (!response.ok) { - const body = await response.text(); - throw new GatewayApiError(parseErrorDetails(body, response.status, `Request failed: ${response.status}`)); + const controller = options.timeoutMs ? new AbortController() : undefined; + const timeout = controller ? globalThis.setTimeout(() => controller.abort(), options.timeoutMs) : undefined; + const signal = controller && options.signal + ? AbortSignal.any([controller.signal, options.signal]) + : controller?.signal ?? options.signal; + try { + const response = await fetch(`${API_BASE}${path}`, { + method: options.method ?? 'GET', + headers, + body: options.body === undefined ? undefined : JSON.stringify(options.body), + credentials: 'include', + signal, + }); + if (!response.ok) { + const body = await response.text(); + throw new GatewayApiError(parseErrorDetails(body, response.status, `Request failed: ${response.status}`)); + } + if (response.status === 204) { + return undefined as T; + } + return response.json() as Promise; + } catch (error) { + if (controller?.signal.aborted) { + throw new GatewayApiError('登录请求超时，请稍后重试'); + } + throw error; + } finally { + if (timeout !== undefined) globalThis.clearTimeout(timeout); } - if (response.status === 204) { - return undefined as T; - } - return response.json() as Promise; } function authorizationHeader(token: string): Record { diff --git a/docker/nginx.conf b/docker/nginx.conf index 510bd73..a7ee56a 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -33,6 +33,20 @@ server { return 404; } + location = /gateway-api/api/v1/auth/login { + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_set_header Connection ""; + proxy_connect_timeout 3s; + proxy_read_timeout 15s; + proxy_send_timeout 15s; + proxy_redirect off; + proxy_pass http://api:8088/api/v1/auth/login; + } + location /gateway-api/ { proxy_http_version 1.1; proxy_set_header Host $host;