From 6dedc5973c28fd747bb1425e22c490edd2264a84 Mon Sep 17 00:00:00 2001 From: Catlaxy AI Date: Sun, 11 Oct 2026 22:04:27 +0800 Subject: [PATCH] =?UTF-8?q?fix(database):=20=E6=95=B0=E6=8D=AE=E5=BA=93?= =?UTF-8?q?=E8=BF=9E=E6=8E=A5=20-=20=E6=B6=88=E8=B4=B9=20libpq=20keepalive?= =?UTF-8?q?=20=E5=8F=82=E6=95=B0=E5=B9=B6=E9=BB=98=E8=AE=A4=E9=99=90?= =?UTF-8?q?=E5=88=B6=E9=97=B2=E7=BD=AE=E4=BA=8B=E5=8A=A1=20(#737)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pgx 不识别 DSN 中 libpq 的 keepalives* 参数，会作为服务端运行参数发送导致连接被拒；改为在建连前消费并应用到 TCP 拨号器。 未显式配置时为每条连接默认设置 idle_in_transaction_session_timeout=5min，避免卡住的事务长期持锁；DSN 显式配置时以 DSN 为准。 建连钩子先复制共享的 RuntimeParams，避免多条连接之间累积修改。 Co-authored-by: Claude Opus 5.5 --- backend/go.mod | 2 +- backend/internal/database/database.go | 6 +- .../internal/database/postgres_connection.go | 86 ++++++++++ .../postgres_connection_linux_test.go | 81 +++++++++ .../database/postgres_connection_test.go | 161 ++++++++++++++++++ 5 files changed, 334 insertions(+), 2 deletions(-) create mode 100644 backend/internal/database/postgres_connection.go create mode 100644 backend/internal/database/postgres_connection_linux_test.go create mode 100644 backend/internal/database/postgres_connection_test.go diff --git a/backend/go.mod b/backend/go.mod index 0ac499a33..08a449aad 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -7,6 +7,7 @@ require ( github.com/gabriel-vasile/mimetype v1.4.13 github.com/gin-gonic/gin v1.12.0 github.com/google/uuid v1.5.0 + github.com/jackc/pgx/v5 v5.10.0 github.com/qiniu/go-sdk/v7 v7.27.0 github.com/redis/go-redis/v9 v9.22.0 github.com/tencentyun/cos-go-sdk-v5 v0.7.75 @@ -44,7 +45,6 @@ require ( github.com/google/go-querystring v1.1.0 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect - github.com/jackc/pgx/v5 v5.10.0 // indirect github.com/jackc/puddle/v2 v2.2.2 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect diff --git a/backend/internal/database/database.go b/backend/internal/database/database.go index 827176046..d465cc5d8 100644 --- a/backend/internal/database/database.go +++ b/backend/internal/database/database.go @@ -8,6 +8,7 @@ import ( "yingce/backend/internal/logging" + "github.com/jackc/pgx/v5/stdlib" "gorm.io/driver/postgres" "gorm.io/driver/sqlite" "gorm.io/gorm" @@ -39,7 +40,10 @@ func Open(config Config) (*gorm.DB, error) { if dsn == "" { return nil, errors.New("PostgreSQL 模式必须配置 DATABASE_URL") } - return gorm.Open(postgres.Open(dsn), &gorm.Config{Logger: logging.GormLogger()}) + return gorm.Open(postgres.New(postgres.Config{ + DSN: dsn, + OptionOpenDB: []stdlib.OptionOpenDB{stdlib.OptionBeforeConnect(configurePostgresConnection)}, + }), &gorm.Config{Logger: logging.GormLogger()}) default: return nil, fmt.Errorf("不支持的数据库驱动：%s", driver) } diff --git a/backend/internal/database/postgres_connection.go b/backend/internal/database/postgres_connection.go new file mode 100644 index 000000000..087051cb8 --- /dev/null +++ b/backend/internal/database/postgres_connection.go @@ -0,0 +1,86 @@ +package database + +import ( + "context" + "fmt" + "maps" + "net" + "strconv" + "strings" + "time" + + "github.com/jackc/pgx/v5" +) + +// defaultIdleInTransactionSessionTimeout 限制事务内等待下一条命令的时间， +// 避免卡住的 idle in transaction 会话长期持有锁。DSN 显式配置时以 DSN 为准。 +const defaultIdleInTransactionSessionTimeout = "5min" + +var libpqKeepaliveParams = []string{"keepalives", "keepalives_idle", "keepalives_interval", "keepalives_count"} + +// configurePostgresConnection 在每次建立物理连接前调整 pgx 连接配置。 +func configurePostgresConnection(_ context.Context, config *pgx.ConnConfig) error { + // stdlib 传入的是浅拷贝，RuntimeParams 与连接器共享，修改前必须复制，否则会跨连接累积。 + config.RuntimeParams = maps.Clone(config.RuntimeParams) + if config.RuntimeParams == nil { + config.RuntimeParams = make(map[string]string) + } + if err := applyLibpqKeepalives(config); err != nil { + return err + } + if _, exists := config.RuntimeParams["idle_in_transaction_session_timeout"]; !exists { + // PostgreSQL 按顺序处理 options，DSN 原有 options 放在后面，同名设置以 DSN 为准。 + options := "-c idle_in_transaction_session_timeout=" + defaultIdleInTransactionSessionTimeout + if existing := strings.TrimSpace(config.RuntimeParams["options"]); existing != "" { + options += " " + existing + } + config.RuntimeParams["options"] = options + } + return nil +} + +// applyLibpqKeepalives 消费 libpq 的 keepalives* 参数并应用到 TCP 拨号器。 +// pgx 不识别这些参数，会把它们当作服务端运行参数发送，导致 PostgreSQL 拒绝连接。 +// 未配置任何 keepalives* 参数时保持 pgx 默认拨号器（Go 默认 keepalive）不变。 +func applyLibpqKeepalives(config *pgx.ConnConfig) error { + values := make(map[string]int) + for _, key := range libpqKeepaliveParams { + raw, ok := config.RuntimeParams[key] + if !ok { + continue + } + delete(config.RuntimeParams, key) + value, err := strconv.Atoi(strings.TrimSpace(raw)) + if err != nil || value < 0 || (key == "keepalives" && value > 1) { + return fmt.Errorf("invalid PostgreSQL connection parameter %s=%q", key, raw) + } + values[key] = value + } + if len(values) == 0 { + return nil + } + dialer := &net.Dialer{Timeout: config.ConnectTimeout} + if enabled, ok := values["keepalives"]; ok && enabled == 0 { + dialer.KeepAlive = -1 + } else { + // 未配置的项使用 Go 默认值（0）；libpq 的 0 表示操作系统默认值，对应 Go 的 -1。 + setting := func(key string) int { + value, ok := values[key] + if !ok { + return 0 + } + if value == 0 { + return -1 + } + return value + } + dialer.KeepAliveConfig = net.KeepAliveConfig{ + Enable: true, + Idle: time.Duration(setting("keepalives_idle")) * time.Second, + Interval: time.Duration(setting("keepalives_interval")) * time.Second, + Count: setting("keepalives_count"), + } + } + config.DialFunc = dialer.DialContext + return nil +} diff --git a/backend/internal/database/postgres_connection_linux_test.go b/backend/internal/database/postgres_connection_linux_test.go new file mode 100644 index 000000000..e7bad923d --- /dev/null +++ b/backend/internal/database/postgres_connection_linux_test.go @@ -0,0 +1,81 @@ +package database + +import ( + "net" + "os" + "strconv" + "sync" + "syscall" + "testing" +) + +// clientSocketKeepalive 在当前进程中按端口找到客户端 socket，读取其 keepalive 选项。 +func clientSocketKeepalive(port int) ([4]int, bool) { + entries, err := os.ReadDir("/proc/self/fd") + if err != nil { + return [4]int{}, false + } + for _, entry := range entries { + fd, err := strconv.Atoi(entry.Name()) + if err != nil { + continue + } + addr, err := syscall.Getsockname(fd) + inet, ok := addr.(*syscall.SockaddrInet4) + if err != nil || !ok || inet.Port != port { + continue + } + var values [4]int + for i, option := range [][2]int{ + {syscall.SOL_SOCKET, syscall.SO_KEEPALIVE}, + {syscall.IPPROTO_TCP, syscall.TCP_KEEPIDLE}, + {syscall.IPPROTO_TCP, syscall.TCP_KEEPINTVL}, + {syscall.IPPROTO_TCP, syscall.TCP_KEEPCNT}, + } { + if values[i], err = syscall.GetsockoptInt(fd, option[0], option[1]); err != nil { + return [4]int{}, false + } + } + return values, true + } + return [4]int{}, false +} + +func TestPostgresOpenAppliesLibpqKeepaliveToSocket(t *testing.T) { + for _, tc := range []struct { + name, params string + want [4]int + }{ + {"explicit", "keepalives_idle=90 keepalives_interval=20 keepalives_count=6", [4]int{1, 90, 20, 6}}, + {"partial keeps Go defaults", "keepalives_idle=45", [4]int{1, 45, 15, 9}}, + {"no params keeps Go defaults", "", [4]int{1, 15, 15, 9}}, + {"disabled", "keepalives=0", [4]int{0, -1, -1, -1}}, + } { + t.Run(tc.name, func(t *testing.T) { + server := newFakePostgresServer(t) + var mu sync.Mutex + var observed [][4]int + server.onStartup = func(conn net.Conn) { + // 已收到启动消息，说明客户端拨号已完成且 socket 选项已设置。 + values, ok := clientSocketKeepalive(conn.RemoteAddr().(*net.TCPAddr).Port) + if !ok { + values = [4]int{-2, -2, -2, -2} + } + mu.Lock() + observed = append(observed, values) + mu.Unlock() + } + openFakePostgres(t, server, tc.params, 0) + mu.Lock() + defer mu.Unlock() + for _, values := range observed { + if values[0] == -2 { + t.Fatal("client socket not found") + } + if values[0] != tc.want[0] || (tc.want[0] == 1 && values != tc.want) { + t.Fatalf("keepalive (enabled, idle, interval, count) = %v, want %v", values, tc.want) + } + } + }) + } +} diff --git a/backend/internal/database/postgres_connection_test.go b/backend/internal/database/postgres_connection_test.go new file mode 100644 index 000000000..1d9add394 --- /dev/null +++ b/backend/internal/database/postgres_connection_test.go @@ -0,0 +1,161 @@ +package database + +import ( + "bytes" + "encoding/binary" + "fmt" + "io" + "net" + "strings" + "sync" + "testing" + "time" +) + +// fakePostgresServer 只实现启动握手和简单查询，用于检查客户端发送的启动参数。 +type fakePostgresServer struct { + listener net.Listener + onStartup func(conn net.Conn) + mu sync.Mutex + startups []map[string]string +} + +func newFakePostgresServer(t *testing.T) *fakePostgresServer { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := &fakePostgresServer{listener: listener} + t.Cleanup(func() { _ = listener.Close() }) + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + go server.serve(conn) + } + }() + return server +} + +func (s *fakePostgresServer) port() int { + return s.listener.Addr().(*net.TCPAddr).Port +} + +func (s *fakePostgresServer) recorded() []map[string]string { + s.mu.Lock() + defer s.mu.Unlock() + return append([]map[string]string(nil), s.startups...) +} + +func (s *fakePostgresServer) serve(conn net.Conn) { + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) + var size uint32 + if err := binary.Read(conn, binary.BigEndian, &size); err != nil || size < 8 { + return + } + body := make([]byte, size-4) + if _, err := io.ReadFull(conn, body); err != nil { + return + } + params := map[string]string{} + fields := bytes.Split(body[4:], []byte{0}) + for i := 0; i+1 < len(fields) && len(fields[i]) > 0; i += 2 { + params[string(fields[i])] = string(fields[i+1]) + } + s.mu.Lock() + s.startups = append(s.startups, params) + s.mu.Unlock() + if s.onStartup != nil { + s.onStartup(conn) + } + // AuthenticationOk + ReadyForQuery + _, _ = conn.Write([]byte{'R', 0, 0, 0, 8, 0, 0, 0, 0, 'Z', 0, 0, 0, 5, 'I'}) + header := make([]byte, 5) + for { + if _, err := io.ReadFull(conn, header); err != nil { + return + } + payload := make([]byte, binary.BigEndian.Uint32(header[1:])-4) + if _, err := io.ReadFull(conn, payload); err != nil { + return + } + switch header[0] { + case 'Q': // EmptyQueryResponse + ReadyForQuery + _, _ = conn.Write([]byte{'I', 0, 0, 0, 4, 'Z', 0, 0, 0, 5, 'I'}) + case 'X': + return + } + } +} + +func openFakePostgres(t *testing.T, server *fakePostgresServer, params string, pings int) []map[string]string { + t.Helper() + dsn := fmt.Sprintf("host=127.0.0.1 port=%d user=test dbname=test sslmode=disable %s", server.port(), params) + db, err := Open(Config{Driver: "postgres", DSN: dsn}) + if err != nil { + t.Fatalf("open: %v", err) + } + sqlDB, err := db.DB() + if err != nil { + t.Fatal(err) + } + defer sqlDB.Close() + // 不保留空闲连接，每次 Ping 都会建立新的物理连接。 + sqlDB.SetMaxIdleConns(0) + for range pings { + if err := sqlDB.Ping(); err != nil { + t.Fatalf("ping: %v", err) + } + } + startups := server.recorded() + if len(startups) < pings+1 { + t.Fatalf("startup count = %d, want at least %d", len(startups), pings+1) + } + return startups +} + +func TestPostgresOpenConsumesLibpqKeepaliveParams(t *testing.T) { + server := newFakePostgresServer(t) + startups := openFakePostgres(t, server, "keepalives=1 keepalives_idle=90 keepalives_interval=20 keepalives_count=6", 0) + for _, params := range startups { + for key := range params { + if strings.HasPrefix(key, "keepalives") { + t.Fatalf("libpq parameter %s sent to server: %v", key, params) + } + } + } +} + +func TestPostgresOpenRejectsInvalidKeepaliveParam(t *testing.T) { + server := newFakePostgresServer(t) + dsn := fmt.Sprintf("host=127.0.0.1 port=%d user=test dbname=test sslmode=disable keepalives_idle=abc", server.port()) + if _, err := Open(Config{Driver: "postgres", DSN: dsn}); err == nil || !strings.Contains(err.Error(), "keepalives_idle") { + t.Fatalf("expected invalid keepalives_idle error, got %v", err) + } +} + +func TestPostgresOpenSetsIdleInTransactionTimeoutPerConnection(t *testing.T) { + server := newFakePostgresServer(t) + startups := openFakePostgres(t, server, "options='-c statement_timeout=4s'", 3) + want := "-c idle_in_transaction_session_timeout=5min -c statement_timeout=4s" + for i, params := range startups { + // 每条连接都应得到相同的 options，不能因共享 RuntimeParams 而跨连接累积。 + if params["options"] != want { + t.Fatalf("connection %d options = %q, want %q", i, params["options"], want) + } + } +} + +func TestPostgresOpenKeepsExplicitIdleInTransactionTimeout(t *testing.T) { + server := newFakePostgresServer(t) + startups := openFakePostgres(t, server, "idle_in_transaction_session_timeout=0 options='-c statement_timeout=4s'", 1) + for i, params := range startups { + if params["idle_in_transaction_session_timeout"] != "0" || params["options"] != "-c statement_timeout=4s" { + t.Fatalf("connection %d overrode explicit timeout: %v", i, params) + } + } +}