Repository: jackc/pgx
Pull request #1443 (2022-12-19 18:10:41)

Make MaxConnLifetimeJitter setting actually jitter (addresses #1437)
This change addresses the issue described in https://github.com/jackc/pgx/issues/1437 by caching the connection expiry time on the connection resource itself, rather than re-calculating it every time a connection is released.


Patch:
diff --git a/conn.go b/conn.go
index 4c8b59da..b6c1ab8f 100644
--- a/conn.go
+++ b/conn.go
@@ -1106,6 +1106,8 @@ func (c *Conn) sendBatchExtendedWithDescription(ctx context.Context, b *Batch, d
 	for _, bi := range b.queuedQueries {
 		err := c.eqb.Build(c.typeMap, bi.sd, bi.arguments)
 		if err != nil {
+			// we wrap the error so we the user can understand which query failed inside the batch
+			err = fmt.Errorf("error building query %s: %w", bi.query, err)
 			return &pipelineBatchResults{ctx: ctx, conn: c, err: err}
 		}
 
diff --git a/pgtype/builtin_wrappers.go b/pgtype/builtin_wrappers.go
index e34dd578..8bf367c1 100644
--- a/pgtype/builtin_wrappers.go
+++ b/pgtype/builtin_wrappers.go
@@ -910,3 +910,43 @@ func (a *anyMultiDimSliceArray) ScanIndexType() any {
 	}
 	return reflect.New(lowestSliceType.Elem()).Interface()
 }
+
+type anyArrayArrayReflect struct {
+	array reflect.Value
+}
+
+func (a anyArrayArrayReflect) Dimensions() []ArrayDimension {
+	return []ArrayDimension{{Length: int32(a.array.Len()), LowerBound: 1}}
+}
+
+func (a anyArrayArrayReflect) Index(i int) any {
+	return a.array.Index(i).Interface()
+}
+
+func (a anyArrayArrayReflect) IndexType() any {
+	return reflect.New(a.array.Type().Elem()).Elem().Interface()
+}
+
+func (a *anyArrayArrayReflect) SetDimensions(dimensions []ArrayDimension) error {
+	if dimensions == nil {
+		return fmt.Errorf("anyArrayArrayReflect: cannot scan NULL into %v", a.array.Type().String())
+	}
+
+	if len(dimensions) != 1 {
+		return fmt.Errorf("anyArrayArrayReflect: cannot scan multi-dimensional array into %v", a.array.Type().String())
+	}
+
+	if int(dimensions[0].Length) != a.array.Len() {
+		return fmt.Errorf("anyArrayArrayReflect: cannot scan array with length %v into %v", dimensions[0].Length, a.array.Type().String())
+	}
+
+	return nil
+}
+
+func (a *anyArrayArrayReflect) ScanIndex(i int) any {
+	return a.array.Index(i).Addr().Interface()
+}
+
+func (a *anyArrayArrayReflect) ScanIndexType() any {
+	return reflect.New(a.array.Type().Elem()).Interface()
+}
diff --git a/pgtype/pgtype.go b/pgtype/pgtype.go
index 2aa96e68..7cfa3d58 100644
--- a/pgtype/pgtype.go
+++ b/pgtype/pgtype.go
@@ -223,6 +223,7 @@ func NewMap() *Map {
 			TryWrapStructEncodePlan,
 			TryWrapSliceEncodePlan,
 			TryWrapMultiDimSliceEncodePlan,
+			TryWrapArrayEncodePlan,
 		},
 
 		TryWrapScanPlanFuncs: []TryWrapScanPlanFunc{
@@ -232,6 +233,7 @@ func NewMap() *Map {
 			TryWrapStructScanPlan,
 			TryWrapPtrSliceScanPlan,
 			TryWrapPtrMultiDimSliceScanPlan,
+			TryWrapPtrArrayScanPlan,
 		},
 	}
 
@@ -1139,6 +1141,31 @@ func (plan *wrapPtrMultiDimSliceScanPlan) Scan(src []byte, target any) error {
 	return plan.next.Scan(src, &anyMultiDimSliceArray{slice: reflect.ValueOf(target).Elem()})
 }
 
+// TryWrapPtrArrayScanPlan tries to wrap a pointer to a single dimension array.
+func TryWrapPtrArrayScanPlan(target any) (plan WrappedScanPlanNextSetter, nextValue any, ok bool) {
+	targetValue := reflect.ValueOf(target)
+	if targetValue.Kind() != reflect.Ptr {
+		return nil, nil, false
+	}
+
+	targetElemValue := targetValue.Elem()
+
+	if targetElemValue.Kind() == reflect.Array {
+		return &wrapPtrArrayReflectScanPlan{}, &anyArrayArrayReflect{array: targetElemValue}, true
+	}
+	return nil, nil, false
+}
+
+type wrapPtrArrayReflectScanPlan struct {
+	next ScanPlan
+}
+
+func (plan *wrapPtrArrayReflectScanPlan) SetNext(next ScanPlan) { plan.next = next }
+
+func (plan *wrapPtrArrayReflectScanPlan) Scan(src []byte, target any) error {
+	return plan.next.Scan(src, &anyArrayArrayReflect{array: reflect.ValueOf(target).Elem()})
+}
+
 // PlanScan prepares a plan to scan a value into target.
 func (m *Map) PlanScan(oid uint32, formatCode int16, target any) ScanPlan {
 	oidMemo := m.memoizedScanPlans[oid]
@@ -1941,6 +1968,35 @@ func (plan *wrapMultiDimSliceEncodePlan) Encode(value any, buf []byte) (newBuf [
 	return plan.next.Encode(&w, buf)
 }
 
+func TryWrapArrayEncodePlan(value any) (plan WrappedEncodePlanNextSetter, nextValue any, ok bool) {
+	if _, ok := value.(driver.Valuer); ok {
+		return nil, nil, false
+	}
+
+	if valueType := reflect.TypeOf(value); valueType != nil && valueType.Kind() == reflect.Array {
+		w := anyArrayArrayReflect{
+			array: reflect.ValueOf(value),
+		}
+		return &wrapArrayEncodeReflectPlan{}, w, true
+	}
+
+	return nil, nil, false
+}
+
+type wrapArrayEncodeReflectPlan struct {
+	next EncodePlan
+}
+
+func (plan *wrapArrayEncodeReflectPlan) SetNext(next EncodePlan) { plan.next = next }
+
+func (plan *wrapArrayEncodeReflectPlan) Encode(value any, buf []byte) (newBuf []byte, err error) {
+	w := anyArrayArrayReflect{
+		array: reflect.ValueOf(value),
+	}
+
+	return plan.next.Encode(w, buf)
+}
+
 func newEncodeError(value any, m *Map, oid uint32, formatCode int16, err error) error {
 	var format string
 	switch formatCode {
diff --git a/pgxpool/pool.go b/pgxpool/pool.go
index 236ba000..51f8afcc 100644
--- a/pgxpool/pool.go
+++ b/pgxpool/pool.go
@@ -22,10 +22,11 @@ var defaultMaxConnIdleTime = time.Minute * 30
 var defaultHealthCheckPeriod = time.Minute
 
 type connResource struct {
-	conn      *pgx.Conn
-	conns     []Conn
-	poolRows  []poolRow
-	poolRowss []poolRows
+	conn       *pgx.Conn
+	conns      []Conn
+	poolRows   []poolRow
+	poolRowss  []poolRows
+	maxAgeTime time.Time
 }
 
 func (cr *connResource) getConn(p *Pool, res *puddle.Resource[*connResource]) *Conn {
@@ -219,11 +220,15 @@ func NewWithConfig(ctx context.Context, config *Config) (*Pool, error) {
 					}
 				}
 
+				jitterSecs := rand.Float64() * config.MaxConnLifetimeJitter.Seconds()
+				maxAgeTime := time.Now().Add(config.MaxConnLifetime).Add(time.Duration(jitterSecs) * time.Second)
+
 				cr := &connResource{
-					conn:      conn,
-					conns:     make([]Conn, 64),
-					poolRows:  make([]poolRow, 64),
-					poolRowss: make([]poolRows, 64),
+					conn:       conn,
+					conns:      make([]Conn, 64),
+					poolRows:   make([]poolRow, 64),
+					poolRowss:  make([]poolRows, 64),
+					maxAgeTime: maxAgeTime,
 				}
 
 				return cr, nil
@@ -364,17 +369,7 @@ func (p *Pool) Close() {
 }
 
 func (p *Pool) isExpired(res *puddle.Resource[*connResource]) bool {
-	now := time.Now()
-	// Small optimization to avoid rand. If it's over lifetime AND jitter, immediately
-	// return true.
-	if now.Sub(res.CreationTime()) > p.maxConnLifetime+p.maxConnLifetimeJitter {
-		return true
-	}
-	if p.maxConnLifetimeJitter == 0 {
-		return false
-	}
-	jitterSecs := rand.Float64() * p.maxConnLifetimeJitter.Seconds()
-	return now.Sub(res.CreationTime()) > p.maxConnLifetime+(time.Duration(jitterSecs)*time.Second)
+	return time.Now().After(res.Value().maxAgeTime)
 }
 
 func (p *Pool) triggerHealthCheck() {


Test patch:
diff --git a/pgtype/array_codec_test.go b/pgtype/array_codec_test.go
index a558d0fc..ede104ac 100644
--- a/pgtype/array_codec_test.go
+++ b/pgtype/array_codec_test.go
@@ -126,6 +126,29 @@ func TestArrayCodecAnySlice(t *testing.T) {
 	})
 }
 
+// https://github.com/jackc/pgx/issues/1442
+func TestArrayCodecAnyArray(t *testing.T) {
+	defaultConnTestRunner.RunTest(context.Background(), t, func(ctx context.Context, t testing.TB, conn *pgx.Conn) {
+		type _point3 [3]float32
+
+		for i, tt := range []struct {
+			expected any
+		}{
+			{_point3{0, 0, 0}},
+			{_point3{1, 2, 3}},
+		} {
+			var actual _point3
+			err := conn.QueryRow(
+				ctx,
+				"select $1::float4[]",
+				tt.expected,
+			).Scan(&actual)
+			assert.NoErrorf(t, err, "%d", i)
+			assert.Equalf(t, tt.expected, actual, "%d", i)
+		}
+	})
+}
+
 // https://github.com/jackc/pgx/issues/1273#issuecomment-1218262703
 func TestArrayCodecSliceArgConversion(t *testing.T) {
 	defaultConnTestRunner.RunTest(context.Background(), t, func(ctx context.Context, t testing.TB, conn *pgx.Conn) {