fix(auth): 为登录链路增加有界超时
ci / verify (pull_request) Successful in 11m22s
ci / verify (pull_request) Successful in 11m22s
为 PostgreSQL 连接、就绪检查和本地登录设置分层超时,数据库不可用时返回稳定 503 错误码并记录无凭据的连接池统计。 前端登录在 10 秒后取消请求并兼容调用方 AbortSignal,Nginx 登录精确路由限制上游为 15 秒,同时更新 OpenAPI 和回归测试。
This commit is contained in:
@@ -5208,6 +5208,12 @@
|
||||
"schema": {
|
||||
"$ref": "#/definitions/httpapi.ErrorEnvelope"
|
||||
}
|
||||
},
|
||||
"503": {
|
||||
"description": "Service Unavailable",
|
||||
"schema": {
|
||||
"$ref": "#/definitions/httpapi.ErrorEnvelope"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
@@ -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"
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user