Skip to content

Commit de5cf1b

Browse files
feat(pgxpool): acquire ping timeout
this commit adds ping timeout in acquire loop for when the connection ping results in timeout instead of an error
1 parent 61d3c96 commit de5cf1b

2 files changed

Lines changed: 67 additions & 3 deletions

File tree

pgxpool/pool.go

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -97,6 +97,7 @@ type Pool struct {
9797
maxConnLifetimeJitter time.Duration
9898
maxConnIdleTime time.Duration
9999
healthCheckPeriod time.Duration
100+
pingTimeout time.Duration
100101

101102
healthCheckChan chan struct{}
102103

@@ -166,6 +167,10 @@ type Config struct {
166167
// MaxConnIdleTime is the duration after which an idle connection will be automatically closed by the health check.
167168
MaxConnIdleTime time.Duration
168169

170+
// PingTimeout is the maximum amount of time to wait for a connection to pong before considering it as unhealthy and
171+
// destroying it. If zero, the default is no timeout.
172+
PingTimeout time.Duration
173+
169174
// MaxConns is the maximum size of the pool. The default is the greater of 4 or runtime.NumCPU().
170175
MaxConns int32
171176

@@ -238,6 +243,7 @@ func NewWithConfig(ctx context.Context, config *Config) (*Pool, error) {
238243
maxConnLifetime: config.MaxConnLifetime,
239244
maxConnLifetimeJitter: config.MaxConnLifetimeJitter,
240245
maxConnIdleTime: config.MaxConnIdleTime,
246+
pingTimeout: config.PingTimeout,
241247
healthCheckPeriod: config.HealthCheckPeriod,
242248
healthCheckChan: make(chan struct{}, 1),
243249
closeChan: make(chan struct{}),
@@ -601,7 +607,14 @@ func (p *Pool) Acquire(ctx context.Context) (c *Conn, err error) {
601607

602608
shouldPingParams := ShouldPingParams{Conn: cr.conn, IdleDuration: res.IdleDuration()}
603609
if p.shouldPing(ctx, shouldPingParams) {
604-
err := cr.conn.Ping(ctx)
610+
pingCtx := ctx
611+
if p.pingTimeout > 0 {
612+
var cancel context.CancelFunc
613+
pingCtx, cancel = context.WithTimeout(ctx, p.pingTimeout)
614+
defer cancel()
615+
}
616+
617+
err := cr.conn.Ping(pingCtx)
605618
if err != nil {
606619
res.Destroy()
607620
continue

pgxpool/pool_test.go

Lines changed: 53 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,11 +10,12 @@ import (
1010
"testing"
1111
"time"
1212

13+
"github.com/stretchr/testify/assert"
14+
"github.com/stretchr/testify/require"
15+
1316
"github.com/jackc/pgx/v5"
1417
"github.com/jackc/pgx/v5/pgxpool"
1518
"github.com/jackc/pgx/v5/pgxtest"
16-
"github.com/stretchr/testify/assert"
17-
"github.com/stretchr/testify/require"
1819
)
1920

2021
func TestConnect(t *testing.T) {
@@ -1281,3 +1282,53 @@ func TestPoolSendBatchBatchCloseTwice(t *testing.T) {
12811282
assert.NoError(t, err)
12821283
}
12831284
}
1285+
1286+
func TestPoolAcquirePingTimeout(t *testing.T) {
1287+
t.Parallel()
1288+
1289+
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
1290+
defer cancel()
1291+
1292+
config, err := pgxpool.ParseConfig(os.Getenv("PGX_TEST_DATABASE"))
1293+
require.NoError(t, err)
1294+
1295+
// Set a very short ping timeout to force timeout during ping and destruction of the connection
1296+
config.PingTimeout = 1 * time.Nanosecond
1297+
1298+
var conID *uint32
1299+
// Only ping the connection with the original PID to force creation of a new connection
1300+
config.ShouldPing = func(_ context.Context, params pgxpool.ShouldPingParams) bool {
1301+
if conID != nil && params.Conn.PgConn().PID() == *conID {
1302+
return true
1303+
}
1304+
1305+
return false
1306+
}
1307+
// Limit to a single connection to ensure the same connection is reused
1308+
config.MinConns = 1
1309+
config.MaxConns = 1
1310+
1311+
pool, err := pgxpool.NewWithConfig(ctx, config)
1312+
require.NoError(t, err)
1313+
defer pool.Close()
1314+
1315+
c, err := pool.Acquire(ctx)
1316+
require.NoError(t, err)
1317+
require.EqualValues(t, 1, pool.Stat().TotalConns())
1318+
originalPID := c.Conn().PgConn().PID()
1319+
conID = &originalPID
1320+
1321+
c.Release()
1322+
require.EqualValues(t, 1, pool.Stat().TotalConns())
1323+
1324+
c, err = pool.Acquire(ctx)
1325+
require.NoError(t, err)
1326+
require.EqualValues(t, 1, pool.Stat().TotalConns())
1327+
newPID := c.Conn().PgConn().PID()
1328+
1329+
c.Release()
1330+
require.EqualValues(t, 1, pool.Stat().TotalConns())
1331+
1332+
assert.NotEqualValues(t, originalPID, newPID,
1333+
"Expected new connection due to ping timeout, but got same connection")
1334+
}

0 commit comments

Comments
 (0)