chore: add debug visibility on bad connection pool

clickhouse.go

@@ -36,6 +36,8 @@ var (
 	ErrAcquireConnNoAddress      = errors.New("clickhouse: no valid address supplied")
 	ErrServerUnexpectedData      = errors.New("code: 101, message: Unexpected packet Data received from client")
 	ErrConnectionClosed          = errors.New("clickhouse: connection is closed")
+
+	errConnMaxLifetimeExceeded = errors.New("clickhouse: connection max lifetime exceeded")
 )
 
 type OpError struct {
@@ -89,7 +91,8 @@ type nativeTransport interface {
 	exec(ctx context.Context, query string, args ...any) error
 	asyncInsert(ctx context.Context, query string, wait bool, args ...any) error
 	ping(context.Context) error
-	isBad() bool
+	// healthCheck reports why the connection is unusable; nil means healthy.
+	healthCheck() error
 	connID() int
 	connectedAtTime() time.Time
 	isReleased() bool
@@ -342,13 +345,14 @@ func (ch *clickhouse) acquire(ctx context.Context) (conn nativeTransport, err er
 	}
 
 	if err == nil && conn != nil {
-		if !conn.isBad() {
+		if badErr := conn.healthCheck(); badErr == nil {
 			conn.setReleased(false)
 			conn.getLogger().Debug("connection acquired from pool")
 			return conn, nil
+		} else {
+			conn.getLogger().Debug("closing bad connection from pool", slog.Any("reason", badErr))
+			conn.close()
 		}
-
-		conn.close()
 	}
 
 	if conn, err = ch.dial(ctx); err != nil {

clickhouse_std.go

@@ -145,7 +145,8 @@ func OpenDB(opt *Options) *sql.DB {
 }
 
 type stdConnect interface {
-	isBad() bool
+	// healthCheck reports why the connection is unusable; nil means healthy.
+	healthCheck() error
 	close() error
 	query(ctx context.Context, release nativeTransportRelease, query string, args ...any) (*rows, error)
 	exec(ctx context.Context, query string, args ...any) error
@@ -182,8 +183,8 @@ func (std *stdDriver) Open(dsn string) (_ driver.Conn, err error) {
 var _ driver.Driver = (*stdDriver)(nil)
 
 func (std *stdDriver) ResetSession(ctx context.Context) error {
-	if std.conn.isBad() {
-		std.logger.Debug("resetting session because connection is bad")
+	if err := std.conn.healthCheck(); err != nil {
+		std.logger.Debug("resetting session because connection is bad", slog.Any("reason", err))
 		return driver.ErrBadConn
 	}
 	return nil
@@ -192,8 +193,8 @@ func (std *stdDriver) ResetSession(ctx context.Context) error {
 var _ driver.SessionResetter = (*stdDriver)(nil)
 
 func (std *stdDriver) Ping(ctx context.Context) error {
-	if std.conn.isBad() {
-		std.logger.Debug("ping: connection is bad")
+	if err := std.conn.healthCheck(); err != nil {
+		std.logger.Debug("ping: connection is bad", slog.Any("reason", err))
 		return driver.ErrBadConn
 	}
 
@@ -203,17 +204,17 @@ func (std *stdDriver) Ping(ctx context.Context) error {
 var _ driver.Pinger = (*stdDriver)(nil)
 
 func (std *stdDriver) Begin() (driver.Tx, error) {
-	if std.conn.isBad() {
-		std.logger.Debug("begin: connection is bad")
+	if err := std.conn.healthCheck(); err != nil {
+		std.logger.Debug("begin: connection is bad", slog.Any("reason", err))
 		return nil, driver.ErrBadConn
 	}
 
 	return std, nil
 }
 
 func (std *stdDriver) BeginTx(ctx context.Context, opts driver.TxOptions) (driver.Tx, error) {
-	if std.conn.isBad() {
-		std.logger.Debug("begin tx: connection is bad")
+	if err := std.conn.healthCheck(); err != nil {
+		std.logger.Debug("begin tx: connection is bad", slog.Any("reason", err))
 		return nil, driver.ErrBadConn
 	}
 
@@ -252,8 +253,8 @@ func (std *stdDriver) CheckNamedValue(nv *driver.NamedValue) error { return nil
 var _ driver.NamedValueChecker = (*stdDriver)(nil)
 
 func (std *stdDriver) ExecContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Result, error) {
-	if std.conn.isBad() {
-		std.logger.Debug("exec context: connection is bad")
+	if err := std.conn.healthCheck(); err != nil {
+		std.logger.Debug("exec context: connection is bad", slog.Any("reason", err))
 		return nil, driver.ErrBadConn
 	}
 
@@ -276,8 +277,8 @@ func (std *stdDriver) ExecContext(ctx context.Context, query string, args []driv
 }
 
 func (std *stdDriver) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
-	if std.conn.isBad() {
-		std.logger.Debug("query context: connection is bad")
+	if err := std.conn.healthCheck(); err != nil {
+		std.logger.Debug("query context: connection is bad", slog.Any("reason", err))
 		return nil, driver.ErrBadConn
 	}
 
@@ -301,8 +302,8 @@ func (std *stdDriver) Prepare(query string) (driver.Stmt, error) {
 }
 
 func (std *stdDriver) PrepareContext(ctx context.Context, query string) (driver.Stmt, error) {
-	if std.conn.isBad() {
-		std.logger.Debug("prepare context: connection is bad")
+	if err := std.conn.healthCheck(); err != nil {
+		std.logger.Debug("prepare context: connection is bad", slog.Any("reason", err))
 		return nil, driver.ErrBadConn
 	}
 

clickhouse_test.go

@@ -3,9 +3,12 @@
 package clickhouse
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"log/slog"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -211,6 +214,61 @@ func TestAcquire_BadConnection(t *testing.T) {
 	}
 }
 
+// TestAcquire_BadConnectionLogsReason tests that discarding a bad pooled
+// connection logs why it was considered bad before dialing a new one.
+func TestAcquire_BadConnectionLogsReason(t *testing.T) {
+	var buf bytes.Buffer
+	logger := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug}))
+
+	var connID atomic.Int64
+	conn, err := Open(&Options{
+		Addr:            []string{"localhost:9000"},
+		DialTimeout:     time.Second,
+		MaxOpenConns:    5,
+		MaxIdleConns:    2,
+		ConnMaxLifetime: time.Hour,
+		DialStrategy: func(ctx context.Context, id int, opt *Options, dial Dial) (DialResult, error) {
+			nextID := int(connID.Add(1))
+			return DialResult{conn: &mockTransport{connectedAt: time.Now(), id: nextID, logger: logger}}, nil
+		},
+	})
+	if err != nil {
+		t.Fatalf("Open failed: %v", err)
+	}
+	defer conn.Close()
+
+	ch := conn.(*clickhouse)
+
+	// Acquire and release while healthy so the connection is pooled
+	conn1, err := ch.acquire(context.Background())
+	if err != nil {
+		t.Fatalf("first acquire failed: %v", err)
+	}
+	ch.release(conn1, nil)
+
+	// Mark it bad after pooling so the acquire path detects it
+	conn1.(*mockTransport).setBad(true)
+
+	conn2, err := ch.acquire(context.Background())
+	if err != nil {
+		t.Fatalf("second acquire failed: %v", err)
+	}
+	if conn1.connID() == conn2.connID() {
+		t.Error("expected a new connection after bad connection detected")
+	}
+
+	logs := buf.String()
+	if !strings.Contains(logs, "closing bad connection from pool") {
+		t.Errorf("expected log message about closing bad connection, got:\n%s", logs)
+	}
+	if !strings.Contains(logs, errMockConnBad.Error()) {
+		t.Errorf("expected log to contain the bad-connection reason, got:\n%s", logs)
+	}
+	if !strings.Contains(logs, "new connection established") {
+		t.Errorf("expected log about new connection, got:\n%s", logs)
+	}
+}
+
 // TestAcquire_MaxOpenConnsLimit tests that MaxOpenConns limit is respected
 func TestAcquire_MaxOpenConnsLimit(t *testing.T) {
 	maxOpen := 2

conn.go

@@ -183,20 +183,21 @@ func (c *connect) settings(querySettings Settings) []proto.Setting {
 	return settings
 }
 
-func (c *connect) isBad() bool {
+func (c *connect) healthCheck() error {
 	if c.isClosed() {
-		return true
+		return ErrConnectionClosed
 	}
 
-	if time.Since(c.connectedAt) >= c.opt.ConnMaxLifetime {
-		return true
+	if age := time.Since(c.connectedAt); age >= c.opt.ConnMaxLifetime {
+		return fmt.Errorf("%w: age %s exceeds max lifetime %s",
+			errConnMaxLifetimeExceeded, age.Round(time.Millisecond), c.opt.ConnMaxLifetime)
 	}
 
 	if err := c.connCheck(); err != nil {
-		return true
+		return fmt.Errorf("clickhouse: connection check failed: %w", err)
 	}
 
-	return false
+	return nil
 }
 
 func (c *connect) isReleased() bool {

conn_health_test.go

@@ -0,0 +1,53 @@
+package clickhouse
+
+import (
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func healthyMockConnect() *connect {
+	conn := createMockConnect(&mockNetConn{})
+	conn.connectedAt = time.Now()
+	conn.opt = &Options{ConnMaxLifetime: time.Hour}
+	return conn
+}
+
+func TestConnectHealthCheck_Healthy(t *testing.T) {
+	conn := healthyMockConnect()
+
+	assert.NoError(t, conn.healthCheck())
+}
+
+func TestConnectHealthCheck_Closed(t *testing.T) {
+	conn := healthyMockConnect()
+	conn.setClosed()
+
+	err := conn.healthCheck()
+	require.Error(t, err)
+	assert.ErrorIs(t, err, ErrConnectionClosed)
+}
+
+func TestConnectHealthCheck_LifetimeExceeded(t *testing.T) {
+	conn := healthyMockConnect()
+	conn.connectedAt = time.Now().Add(-2 * time.Hour)
+
+	err := conn.healthCheck()
+	require.Error(t, err)
+	assert.ErrorIs(t, err, errConnMaxLifetimeExceeded)
+	assert.Contains(t, err.Error(), "age")
+	assert.Contains(t, err.Error(), "max lifetime")
+}
+
+func TestHTTPConnectHealthCheck(t *testing.T) {
+	closed := &httpConnect{}
+	err := closed.healthCheck()
+	require.Error(t, err)
+	assert.ErrorIs(t, err, ErrConnectionClosed)
+
+	healthy := &httpConnect{client: &http.Client{}}
+	assert.NoError(t, healthy.healthCheck())
+}

conn_http.go

@@ -313,8 +313,11 @@ func (h *httpConnect) setReleased(released bool) {
 func (h *httpConnect) freeBuffer() {
 }
 
-func (h *httpConnect) isBad() bool {
-	return h.client == nil
+func (h *httpConnect) healthCheck() error {
+	if h.client == nil {
+		return ErrConnectionClosed
+	}
+	return nil
 }
 
 func (h *httpConnect) queryHello(ctx context.Context, release nativeTransportRelease) (proto.ServerHandshake, error) {

conn_pool.go

@@ -3,6 +3,7 @@ package clickhouse
 import (
 	"context"
 	"errors"
+	"log/slog"
 	"sync"
 	"time"
 
@@ -80,12 +81,20 @@ func (i *connPool) Get(ctx context.Context) (nativeTransport, error) {
 			return conn, nil
 		}
 
+		i.logExpired(conn, "closing expired connection from pool")
 		conn.close()
 	}
 }
 
 func (i *connPool) Put(conn nativeTransport) {
-	if i.isExpired(conn) || conn.isBad() {
+	if i.isExpired(conn) {
+		i.logExpired(conn, "connection not returned to pool: lifetime expired")
+		conn.close()
+		return
+	}
+
+	if err := conn.healthCheck(); err != nil {
+		conn.getLogger().Debug("connection not returned to pool: connection is bad", slog.Any("reason", err))
 		conn.close()
 		return
 	}
@@ -100,6 +109,7 @@ func (i *connPool) Put(conn nativeTransport) {
 	// Try to push the connection
 	if !i.conns.Push(conn) {
 		// Buffer is full, close the connection
+		conn.getLogger().Debug("connection not returned to pool: pool is full")
 		conn.close()
 	}
 }
@@ -166,6 +176,7 @@ func (i *connPool) drainPool() {
 	for conn := range i.conns.DeleteFunc(func(conn nativeTransport) bool {
 		return i.isExpired(conn)
 	}) {
+		i.logExpired(conn, "closing expired connection from pool")
 		conn.close()
 	}
 }
@@ -174,6 +185,13 @@ func (i *connPool) isExpired(conn nativeTransport) bool {
 	return !time.Now().Before(i.expires(conn))
 }
 
+func (i *connPool) logExpired(conn nativeTransport, msg string) {
+	conn.getLogger().Debug(msg,
+		slog.Duration("age", time.Since(conn.connectedAtTime())),
+		slog.Duration("max_lifetime", i.maxConnLifetime),
+	)
+}
+
 func (i *connPool) expires(conn nativeTransport) time.Time {
 	return conn.connectedAtTime().Add(i.maxConnLifetime)
 }