Merge pull request #2159 from binwiederhier/pg-tcp-timeout

Give up on a database connection after 30s without an acknowledgment

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
+		},
+	}
+}

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())
+}

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{}
+}

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 == "" {

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)