diff --git a/db/pg/dialer_linux.go b/db/pg/dialer_linux.go new file mode 100644 index 00000000..f6adbda3 --- /dev/null +++ b/db/pg/dialer_linux.go @@ -0,0 +1,31 @@ +package pg + +import ( + "net" + "strings" + "syscall" + + "golang.org/x/sys/unix" +) + +// newDialer returns the dialer for database connections. It sets TCP_USER_TIMEOUT, so a +// connection whose query is never acknowledged (the node went away in a resize or failover) +// fails after tcpUserTimeout instead of after the kernel's ~15 minutes of retransmissions. +// It only bounds unacknowledged data, not a slow query on a healthy connection. +func newDialer() *net.Dialer { + return &net.Dialer{ + Control: func(network, address string, c syscall.RawConn) error { + if !strings.HasPrefix(network, "tcp") { + return nil // Unix sockets (host=/var/run/postgresql) have no TCP_USER_TIMEOUT + } + var sockErr error + err := c.Control(func(fd uintptr) { + sockErr = unix.SetsockoptInt(int(fd), syscall.IPPROTO_TCP, unix.TCP_USER_TIMEOUT, int(tcpUserTimeout.Milliseconds())) + }) + if err != nil { + return err + } + return sockErr + }, + } +} diff --git a/db/pg/dialer_linux_test.go b/db/pg/dialer_linux_test.go new file mode 100644 index 00000000..07aa36e5 --- /dev/null +++ b/db/pg/dialer_linux_test.go @@ -0,0 +1,70 @@ +package pg + +import ( + "context" + "net" + "path/filepath" + "syscall" + "testing" + + "github.com/stretchr/testify/require" + "golang.org/x/sys/unix" +) + +func TestNewDialer_SetsTCPUserTimeout(t *testing.T) { + // A query sent to a database node that went away (a resize or failover moved it) is never + // acknowledged; without TCP_USER_TIMEOUT, the kernel retransmits for ~15 minutes before the + // connection fails, and the request holds its pool slot all that time + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.Nil(t, err) + defer listener.Close() + conn, err := newDialer().DialContext(context.Background(), "tcp", listener.Addr().String()) + require.Nil(t, err) + defer conn.Close() + raw, err := conn.(*net.TCPConn).SyscallConn() + require.Nil(t, err) + var timeout int + var sockErr error + err = raw.Control(func(fd uintptr) { + timeout, sockErr = unix.GetsockoptInt(int(fd), syscall.IPPROTO_TCP, unix.TCP_USER_TIMEOUT) + }) + require.Nil(t, err) + require.Nil(t, sockErr) + require.Equal(t, int(tcpUserTimeout.Milliseconds()), timeout) +} + +func TestParseConnConfig_DialsWithTCPUserTimeout(t *testing.T) { + // The pool's connections are opened with newDialer, so they get TCP_USER_TIMEOUT too + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.Nil(t, err) + defer listener.Close() + config, err := parseConnConfig("postgres://user:pass@" + listener.Addr().String() + "/db") + require.Nil(t, err) + conn, err := config.DialFunc(context.Background(), "tcp", listener.Addr().String()) + require.Nil(t, err) + defer conn.Close() + raw, err := conn.(*net.TCPConn).SyscallConn() + require.Nil(t, err) + var timeout int + var sockErr error + err = raw.Control(func(fd uintptr) { + timeout, sockErr = unix.GetsockoptInt(int(fd), syscall.IPPROTO_TCP, unix.TCP_USER_TIMEOUT) + }) + require.Nil(t, err) + require.Nil(t, sockErr) + require.Equal(t, int(tcpUserTimeout.Milliseconds()), timeout) +} + +func TestParseConnConfig_DialsUnixSocket(t *testing.T) { + // TCP_USER_TIMEOUT does not exist for Unix sockets (host=/var/run/postgresql), so the dialer + // must not set it there, or every connection fails with "operation not supported" + path := filepath.Join(t.TempDir(), ".s.PGSQL.5432") + listener, err := net.Listen("unix", path) + require.Nil(t, err) + defer listener.Close() + config, err := parseConnConfig("postgres://user:pass@/db?host=" + filepath.Dir(path)) + require.Nil(t, err) + conn, err := config.DialFunc(context.Background(), "unix", path) + require.Nil(t, err) + require.Nil(t, conn.Close()) +} diff --git a/db/pg/dialer_other.go b/db/pg/dialer_other.go new file mode 100644 index 00000000..e2d4adad --- /dev/null +++ b/db/pg/dialer_other.go @@ -0,0 +1,13 @@ +//go:build !linux + +package pg + +import ( + "net" +) + +// newDialer returns the dialer for database connections. TCP_USER_TIMEOUT is Linux only, so +// elsewhere a connection to a vanished node fails only after the kernel's retransmissions. +func newDialer() *net.Dialer { + return &net.Dialer{} +} diff --git a/db/pg/pg.go b/db/pg/pg.go index e331a78e..e42e0d5b 100644 --- a/db/pg/pg.go +++ b/db/pg/pg.go @@ -1,18 +1,24 @@ package pg import ( - "database/sql" "fmt" "net/url" "strconv" "strings" "time" - _ "github.com/jackc/pgx/v5/stdlib" // PostgreSQL driver + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/stdlib" "heckel.io/ntfy/v2/db" ) +const ( + // tcpUserTimeout is how long a database connection may have data in flight without an + // acknowledgment before the kernel gives up on it (Linux only, see newDialer) + tcpUserTimeout = 30 * time.Second +) + // Advisory lock keys. PostgreSQL advisory locks share one database-wide key space, so every // ntfy key is defined here, following the "ntfy"+2586+letter scheme const ( @@ -72,10 +78,11 @@ func open(dsn string) (*db.Host, error) { return nil, err } u.RawQuery = q.Encode() - d, err := sql.Open("pgx", u.String()) + config, err := parseConnConfig(u.String()) if err != nil { return nil, err } + d := stdlib.OpenDB(*config) d.SetMaxOpenConns(maxOpenConns) if maxIdleConns == 0 { // database/sql keeps only 2 idle connections by default, so every burst above that @@ -95,6 +102,16 @@ func open(dsn string) (*db.Host, error) { }, nil } +// parseConnConfig parses the driver part of the DSN, and dials with newDialer +func parseConnConfig(dsn string) (*pgx.ConnConfig, error) { + config, err := pgx.ParseConfig(dsn) + if err != nil { + return nil, err + } + config.DialFunc = newDialer().DialContext + return config, nil +} + func extractIntParam(q url.Values, key string, defaultValue int) (int, error) { s := q.Get(key) if s == "" { diff --git a/docs/releases.md b/docs/releases.md index a669df0e..7a05d6bf 100644 --- a/docs/releases.md +++ b/docs/releases.md @@ -2359,6 +2359,7 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release * `ntfy cluster status` lists the node itself as the first row of the table (marked "this node", and "leader" if it leads) instead of a line above it ([#2153](https://github.com/binwiederhier/ntfy/pull/2153)) * The cluster and admin socket options lost their `experimental-` prefix, e.g. `cluster-listen` instead of `experimental-cluster-listen` (and `NTFY_CLUSTER_LISTEN`); the old names still work ([#2154](https://github.com/binwiederhier/ntfy/pull/2154)) * Send read-only queries to the least busy read replica instead of round-robin, so a replica that stalls stops getting new reads; the new `ntfy_db_ping_seconds` metric shows how long each database takes to answer its health check ([#2156](https://github.com/binwiederhier/ntfy/pull/2156)) +* PostgreSQL connections give up after 30 seconds without an answer from the database server (`TCP_USER_TIMEOUT`, Linux only), instead of hanging for ~15 minutes when a node goes away during a resize or failover ([#2159](https://github.com/binwiederhier/ntfy/pull/2159)) ### ntfy iOS app v1.8.0 (UNRELEASED)