Repository: jackc/pgx
Pull request #1996 (2024-04-29 14:49:39)

make QueuedQuery.Fn property public, closes #1878
This commit fixes the overlook of the #1886 where `QueuedQuery.SQL` and `QueuedQuery.Arguments` properties were exposed.

And if for `SQL` and `Arguments` we had an opportunity to use reflection, for `QueuedQuery.fn` it's impossible.

One more benefit exposing `QueuedQuery.Fn` for developers is the ability to execute batched functions without knowing the kind of a queued query and mandatory calling `BatchResults.Close()`, e.g.

```go
batch := &pgx.Batch{}
batch.Queue("select func42($1)", 42).QueryRow(...)
batch.Queue("select * from foo").Query(...)
batch.Queue("insert").Exec(...)
...
batch.Queue("truncate foo").Exec(...)

br = conn.SendBatch(ctx, batch)
// check first two queued queries
err := batch.QueuedQueries[0].Fn()
err = errors.Join(err, batch.QueuedQueries[1].Fn())
// let BatchResults execute all that left and close 
err = errors.Join(err, br.Close())
if err != nil {
	fmt.Printf("SendBatch error: %v", err)
	return
}
```

pashagolub: When this PR will be merged, one will be able to test Batches with pgxmock:

```go
func TestBatch(t *testing.T) {
	t.Parallel()
	mock, _ := NewConn()
	a := assert.New(t)

	// define our expectations
	eb := mock.ExpectBatch()
	eb.ExpectQuery("select").WillReturnRows(NewRows([]string{"sum"}).AddRow(2))
	eb.ExpectExec("update").WithArgs(true, 1).WillReturnResult(NewResult("UPDATE", 1))

	// run the test
	batch := &pgx.Batch{}
	batch.Queue("select 1 + 1").QueryRow(func(row pgx.Row) error {
		var n int32
		return row.Scan(&n)
	})
	batch.Queue("update users set active = $1 where id = $2", true, 1).Exec(func(ct pgconn.CommandTag) (err error) {
		if ct.RowsAffected() != 1 {
			err = errors.New("expected 1 row to be affected")
		}
		return
	})

	err := mock.SendBatch(ctx, batch).Close()
	a.NoError(err)
	a.NoError(mock.ExpectationsWereMet())
}
```

Patch:
diff --git a/batch.go b/batch.go
index b9b46d1d..3540f57f 100644
--- a/batch.go
+++ b/batch.go
@@ -12,7 +12,7 @@ import (
 type QueuedQuery struct {
 	SQL       string
 	Arguments []any
-	fn        batchItemFunc
+	Fn        batchItemFunc
 	sd        *pgconn.StatementDescription
 }
 
@@ -20,7 +20,7 @@ type batchItemFunc func(br BatchResults) error
 
 // Query sets fn to be called when the response to qq is received.
 func (qq *QueuedQuery) Query(fn func(rows Rows) error) {
-	qq.fn = func(br BatchResults) error {
+	qq.Fn = func(br BatchResults) error {
 		rows, _ := br.Query()
 		defer rows.Close()
 
@@ -36,7 +36,7 @@ func (qq *QueuedQuery) Query(fn func(rows Rows) error) {
 
 // Query sets fn to be called when the response to qq is received.
 func (qq *QueuedQuery) QueryRow(fn func(row Row) error) {
-	qq.fn = func(br BatchResults) error {
+	qq.Fn = func(br BatchResults) error {
 		row := br.QueryRow()
 		return fn(row)
 	}
@@ -44,7 +44,7 @@ func (qq *QueuedQuery) QueryRow(fn func(row Row) error) {
 
 // Exec sets fn to be called when the response to qq is received.
 func (qq *QueuedQuery) Exec(fn func(ct pgconn.CommandTag) error) {
-	qq.fn = func(br BatchResults) error {
+	qq.Fn = func(br BatchResults) error {
 		ct, err := br.Exec()
 		if err != nil {
 			return err
@@ -228,8 +228,8 @@ func (br *batchResults) Close() error {
 
 	// Read and run fn for all remaining items
 	for br.err == nil && !br.closed && br.b != nil && br.qqIdx < len(br.b.QueuedQueries) {
-		if br.b.QueuedQueries[br.qqIdx].fn != nil {
-			err := br.b.QueuedQueries[br.qqIdx].fn(br)
+		if br.b.QueuedQueries[br.qqIdx].Fn != nil {
+			err := br.b.QueuedQueries[br.qqIdx].Fn(br)
 			if err != nil {
 				br.err = err
 			}
@@ -397,8 +397,8 @@ func (br *pipelineBatchResults) Close() error {
 
 	// Read and run fn for all remaining items
 	for br.err == nil && !br.closed && br.b != nil && br.qqIdx < len(br.b.QueuedQueries) {
-		if br.b.QueuedQueries[br.qqIdx].fn != nil {
-			err := br.b.QueuedQueries[br.qqIdx].fn(br)
+		if br.b.QueuedQueries[br.qqIdx].Fn != nil {
+			err := br.b.QueuedQueries[br.qqIdx].Fn(br)
 			if err != nil {
 				br.err = err
 			}
diff --git a/doc.go b/doc.go
index db99fc4c..bc0391dd 100644
--- a/doc.go
+++ b/doc.go
@@ -11,9 +11,10 @@ The primary way of establishing a connection is with [pgx.Connect]:
 
     conn, err := pgx.Connect(context.Background(), os.Getenv("DATABASE_URL"))
 
-The database connection string can be in URL or DSN format. Both PostgreSQL settings and pgx settings can be specified
-here. In addition, a config struct can be created by [ParseConfig] and modified before establishing the connection with
-[ConnectConfig] to configure settings such as tracing that cannot be configured with a connection string.
+The database connection string can be in URL or key/value format. Both PostgreSQL settings and pgx settings can be
+specified here. In addition, a config struct can be created by [ParseConfig] and modified before establishing the
+connection with [ConnectConfig] to configure settings such as tracing that cannot be configured with a connection
+string.
 
 Connection Pool
 
@@ -23,8 +24,8 @@ github.com/jackc/pgx/v5/pgxpool for a concurrency safe connection pool.
 Query Interface
 
 pgx implements Query in the familiar database/sql style. However, pgx provides generic functions such as CollectRows and
-ForEachRow that are a simpler and safer way of processing rows than manually calling rows.Next(), rows.Scan, and
-rows.Err().
+ForEachRow that are a simpler and safer way of processing rows than manually calling defer rows.Close(), rows.Next(),
+rows.Scan, and rows.Err().
 
 CollectRows can be used collect all returned rows into a slice.
 
diff --git a/pgconn/config.go b/pgconn/config.go
index 13def1e8..7d1c6fba 100644
--- a/pgconn/config.go
+++ b/pgconn/config.go
@@ -19,6 +19,7 @@ import (
 
 	"github.com/jackc/pgpassfile"
 	"github.com/jackc/pgservicefile"
+	"github.com/jackc/pgx/v5/pgconn/ctxwatch"
 	"github.com/jackc/pgx/v5/pgproto3"
 )
 
@@ -39,7 +40,12 @@ type Config struct {
 	DialFunc       DialFunc   // e.g. net.Dialer.DialContext
 	LookupFunc     LookupFunc // e.g. net.Resolver.LookupHost
 	BuildFrontend  BuildFrontendFunc
-	RuntimeParams  map[string]string // Run-time parameters to set on connection as session default values (e.g. search_path or application_name)
+
+	// BuildContextWatcherHandler is called to create a ContextWatcherHandler for a connection. The handler is called
+	// when a context passed to a PgConn method is canceled.
+	BuildContextWatcherHandler func(*PgConn) ctxwatch.Handler
+
+	RuntimeParams map[string]string // Run-time parameters to set on connection as session default values (e.g. search_path or application_name)
 
 	KerberosSrvName string
 	KerberosSpn     string
@@ -112,6 +118,14 @@ type FallbackConfig struct {
 	TLSConfig *tls.Config // nil disables TLS
 }
 
+// connectOneConfig is the configuration for a single attempt to connect to a single host.
+type connectOneConfig struct {
+	network          string
+	address          string
+	originalHostname string      // original hostname before resolving
+	tlsConfig        *tls.Config // nil disables TLS
+}
+
 // isAbsolutePath checks if the provided value is an absolute path either
 // beginning with a forward slash (as on Linux-based systems) or with a capital
 // letter A-Z followed by a colon and a backslash, e.g., "C:\", (as on Windows).
@@ -146,11 +160,11 @@ func NetworkAddress(host string, port uint16) (network, address string) {
 
 // ParseConfig builds a *Config from connString with similar behavior to the PostgreSQL standard C library libpq. It
 // uses the same defaults as libpq (e.g. port=5432) and understands most PG* environment variables. ParseConfig closely
-// matches the parsing behavior of libpq. connString may either be in URL format or keyword = value format (DSN style).
-// See https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING for details. connString also may be
-// empty to only read from the environment. If a password is not supplied it will attempt to read the .pgpass file.
+// matches the parsing behavior of libpq. connString may either be in URL format or keyword = value format. See
+// https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING for details. connString also may be empty
+// to only read from the environment. If a password is not supplied it will attempt to read the .pgpass file.
 //
-//	# Example DSN
+//	# Example Keyword/Value
 //	user=jack password=secret host=pg.example.com port=5432 dbname=mydb sslmode=verify-ca
 //
 //	# Example URL
@@ -169,7 +183,7 @@ func NetworkAddress(host string, port uint16) (network, address string) {
 //	postgres://jack:secret@foo.example.com:5432,bar.example.com:5432/mydb
 //
 // ParseConfig currently recognizes the following environment variable and their parameter key word equivalents passed
-// via database URL or DSN:
+// via database URL or keyword/value:
 //
 //	PGHOST
 //	PGPORT
@@ -233,16 +247,16 @@ func ParseConfigWithOptions(connString string, options ParseConfigOptions) (*Con
 	connStringSettings := make(map[string]string)
 	if connString != "" {
 		var err error
-		// connString may be a database URL or a DSN
+		// connString may be a database URL or in PostgreSQL keyword/value format
 		if strings.HasPrefix(connString, "postgres://") || strings.HasPrefix(connString, "postgresql://") {
 			connStringSettings, err = parseURLSettings(connString)
 			if err != nil {
 				return nil, &ParseConfigError{ConnString: connString, msg: "failed to parse as URL", err: err}
 			}
 		} else {
-			connStringSettings, err = parseDSNSettings(connString)
+			connStringSettings, err = parseKeywordValueSettings(connString)
 			if err != nil {
-				return nil, &ParseConfigError{ConnString: connString, msg: "failed to parse as DSN", err: err}
+				return nil, &ParseConfigError{ConnString: connString, msg: "failed to parse as keyword/value", err: err}
 			}
 		}
 	}
@@ -266,6 +280,9 @@ func ParseConfigWithOptions(connString string, options ParseConfigOptions) (*Con
 		BuildFrontend: func(r io.Reader, w io.Writer) *pgproto3.Frontend {
 			return pgproto3.NewFrontend(r, w)
 		},
+		BuildContextWatcherHandler: func(pgConn *PgConn) ctxwatch.Handler {
+			return &DeadlineContextWatcherHandler{Conn: pgConn.conn}
+		},
 		OnPgError: func(_ *PgConn, pgErr *PgError) bool {
 			// we want to automatically close any fatal errors
 			if strings.EqualFold(pgErr.Severity, "FATAL") {
@@ -517,7 +534,7 @@ func isIPOnly(host string) bool {
 
 var asciiSpace = [256]uint8{'\t': 1, '\n': 1, '\v': 1, '\f': 1, '\r': 1, ' ': 1}
 
-func parseDSNSettings(s string) (map[string]string, error) {
+func parseKeywordValueSettings(s string) (map[string]string, error) {
 	settings := make(map[string]string)
 
 	nameMap := map[string]string{
@@ -528,7 +545,7 @@ func parseDSNSettings(s string) (map[string]string, error) {
 		var key, val string
 		eqIdx := strings.IndexRune(s, '=')
 		if eqIdx < 0 {
-			return nil, errors.New("invalid dsn")
+			return nil, errors.New("invalid keyword/value")
 		}
 
 		key = strings.Trim(s[:eqIdx], " \t\n\r\v\f")
@@ -580,7 +597,7 @@ func parseDSNSettings(s string) (map[string]string, error) {
 		}
 
 		if key == "" {
-			return nil, errors.New("invalid dsn")
+			return nil, errors.New("invalid keyword/value")
 		}
 
 		settings[key] = val
diff --git a/pgconn/internal/ctxwatch/context_watcher.go b/pgconn/ctxwatch/context_watcher.go
similarity index 71%
rename from pgconn/internal/ctxwatch/context_watcher.go
rename to pgconn/ctxwatch/context_watcher.go
index b39cb3ee..db8884eb 100644
--- a/pgconn/internal/ctxwatch/context_watcher.go
+++ b/pgconn/ctxwatch/context_watcher.go
@@ -8,9 +8,8 @@ import (
 // ContextWatcher watches a context and performs an action when the context is canceled. It can watch one context at a
 // time.
 type ContextWatcher struct {
-	onCancel             func()
-	onUnwatchAfterCancel func()
-	unwatchChan          chan struct{}
+	handler     Handler
+	unwatchChan chan struct{}
 
 	lock              sync.Mutex
 	watchInProgress   bool
@@ -20,11 +19,10 @@ type ContextWatcher struct {
 // NewContextWatcher returns a ContextWatcher. onCancel will be called when a watched context is canceled.
 // OnUnwatchAfterCancel will be called when Unwatch is called and the watched context had already been canceled and
 // onCancel called.
-func NewContextWatcher(onCancel func(), onUnwatchAfterCancel func()) *ContextWatcher {
+func NewContextWatcher(handler Handler) *ContextWatcher {
 	cw := &ContextWatcher{
-		onCancel:             onCancel,
-		onUnwatchAfterCancel: onUnwatchAfterCancel,
-		unwatchChan:          make(chan struct{}),
+		handler:     handler,
+		unwatchChan: make(chan struct{}),
 	}
 
 	return cw
@@ -46,7 +44,7 @@ func (cw *ContextWatcher) Watch(ctx context.Context) {
 		go func() {
 			select {
 			case <-ctx.Done():
-				cw.onCancel()
+				cw.handler.HandleCancel(ctx)
 				cw.onCancelWasCalled = true
 				<-cw.unwatchChan
 			case <-cw.unwatchChan:
@@ -66,8 +64,17 @@ func (cw *ContextWatcher) Unwatch() {
 	if cw.watchInProgress {
 		cw.unwatchChan <- struct{}{}
 		if cw.onCancelWasCalled {
-			cw.onUnwatchAfterCancel()
+			cw.handler.HandleUnwatchAfterCancel()
 		}
 		cw.watchInProgress = false
 	}
 }
+
+type Handler interface {
+	// HandleCancel is called when the context that a ContextWatcher is currently watching is canceled. canceledCtx is the
+	// context that was canceled.
+	HandleCancel(canceledCtx context.Context)
+
+	// HandleUnwatchAfterCancel is called when a ContextWatcher that called HandleCancel on this Handler is unwatched.
+	HandleUnwatchAfterCancel()
+}
diff --git a/pgconn/doc.go b/pgconn/doc.go
index e3242cf4..70137501 100644
--- a/pgconn/doc.go
+++ b/pgconn/doc.go
@@ -5,8 +5,8 @@ nearly the same level is the C library libpq.
 
 Establishing a Connection
 
-Use Connect to establish a connection. It accepts a connection string in URL or DSN and will read the environment for
-libpq style environment variables.
+Use Connect to establish a connection. It accepts a connection string in URL or keyword/value format and will read the
+environment for libpq style environment variables.
 
 Executing a Query
 
@@ -20,13 +20,17 @@ result. The ReadAll method reads all query results into memory.
 
 Pipeline Mode
 
-Pipeline mode allows sending queries without having read the results of previously sent queries. It allows
-control of exactly how many and when network round trips occur.
+Pipeline mode allows sending queries without having read the results of previously sent queries. It allows control of
+exactly how many and when network round trips occur.
 
 Context Support
 
-All potentially blocking operations take a context.Context. If a context is canceled while the method is in progress the
-method immediately returns. In most circumstances, this will close the underlying connection.
+All potentially blocking operations take a context.Context. The default behavior when a context is canceled is for the
+method to immediately return. In most circumstances, this will also close the underlying connection. This behavior can
+be customized by using BuildContextWatcherHandler on the Config to create a ctxwatch.Handler with different behavior.
+This can be especially useful when queries that are frequently canceled and the overhead of creating new connections is
+a problem. DeadlineContextWatcherHandler and CancelRequestContextWatcherHandler can be used to introduce a delay before
+interrupting the query in such a way as to close the connection.
 
 The CancelRequest method may be used to request the PostgreSQL server cancel an in-progress query without forcing the
 client to abort.
diff --git a/pgconn/errors.go b/pgconn/errors.go
index 6790b9b6..ec4a6d47 100644
--- a/pgconn/errors.go
+++ b/pgconn/errors.go
@@ -12,8 +12,9 @@ import (
 
 // SafeToRetry checks if the err is guaranteed to have occurred before sending any data to the server.
 func SafeToRetry(err error) bool {
-	if e, ok := err.(interface{ SafeToRetry() bool }); ok {
-		return e.SafeToRetry()
+	var retryableErr interface{ SafeToRetry() bool }
+	if errors.As(err, &retryableErr) {
+		return retryableErr.SafeToRetry()
 	}
 	return false
 }
@@ -61,23 +62,37 @@ func (pe *PgError) SQLState() string {
 // ConnectError is the error returned when a connection attempt fails.
 type ConnectError struct {
 	Config *Config // The configuration that was used in the connection attempt.
-	msg    string
 	err    error
 }
 
 func (e *ConnectError) Error() string {
-	sb := &strings.Builder{}
-	fmt.Fprintf(sb, "failed to connect to `host=%s user=%s database=%s`: %s", e.Config.Host, e.Config.User, e.Config.Database, e.msg)
-	if e.err != nil {
-		fmt.Fprintf(sb, " (%s)", e.err.Error())
+	prefix := fmt.Sprintf("failed to connect to `user=%s database=%s`:", e.Config.User, e.Config.Database)
+	details := e.err.Error()
+	if strings.Contains(details, "\n") {
+		return prefix + "\n\t" + strings.ReplaceAll(details, "\n", "\n\t")
+	} else {
+		return prefix + " " + details
 	}
-	return sb.String()
 }
 
 func (e *ConnectError) Unwrap() error {
 	return e.err
 }
 
+type perDialConnectError struct {
+	address          string
+	originalHostname string
+	err              error
+}
+
+func (e *perDialConnectError) Error() string {
+	return fmt.Sprintf("%s (%s): %s", e.address, e.originalHostname, e.err.Error())
+}
+
+func (e *perDialConnectError) Unwrap() error {
+	return e.err
+}
+
 type connLockError struct {
 	status string
 }
@@ -196,10 +211,10 @@ func redactPW(connString string) string {
 			return redactURL(u)
 		}
 	}
-	quotedDSN := regexp.MustCompile(`password='[^']*'`)
-	connString = quotedDSN.ReplaceAllLiteralString(connString, "password=xxxxx")
-	plainDSN := regexp.MustCompile(`password=[^ ]*`)
-	connString = plainDSN.ReplaceAllLiteralString(connString, "password=xxxxx")
+	quotedKV := regexp.MustCompile(`password='[^']*'`)
+	connString = quotedKV.ReplaceAllLiteralString(connString, "password=xxxxx")
+	plainKV := regexp.MustCompile(`password=[^ ]*`)
+	connString = plainKV.ReplaceAllLiteralString(connString, "password=xxxxx")
 	brokenURL := regexp.MustCompile(`:[^:@]+?@`)
 	connString = brokenURL.ReplaceAllLiteralString(connString, ":xxxxxx@")
 	return connString
diff --git a/pgconn/pgconn.go b/pgconn/pgconn.go
index 170f18db..7efb522a 100644
--- a/pgconn/pgconn.go
+++ b/pgconn/pgconn.go
@@ -18,8 +18,8 @@ import (
 
 	"github.com/jackc/pgx/v5/internal/iobufpool"
 	"github.com/jackc/pgx/v5/internal/pgio"
+	"github.com/jackc/pgx/v5/pgconn/ctxwatch"
 	"github.com/jackc/pgx/v5/pgconn/internal/bgreader"
-	"github.com/jackc/pgx/v5/pgconn/internal/ctxwatch"
 	"github.com/jackc/pgx/v5/pgproto3"
 )
 
@@ -82,6 +82,8 @@ type PgConn struct {
 	slowWriteTimer    *time.Timer
 	bgReaderStarted   chan struct{}
 
+	customData map[string]any
+
 	config *Config
 
 	status byte // One of connStatus* constants
@@ -103,8 +105,9 @@ type PgConn struct {
 	cleanupDone chan struct{}
 }
 
-// Connect establishes a connection to a PostgreSQL server using the environment and connString (in URL or DSN format)
-// to provide configuration. See documentation for [ParseConfig] for details. ctx can be used to cancel a connect attempt.
+// Connect establishes a connection to a PostgreSQL server using the environment and connString (in URL or keyword/value
+// format) to provide configuration. See documentation for [ParseConfig] for details. ctx can be used to cancel a
+// connect attempt.
 func Connect(ctx context.Context, connString string) (*PgConn, error) {
 	config, err := ParseConfig(connString)
 	if err != nil {
@@ -114,9 +117,9 @@ func Connect(ctx context.Context, connString string) (*PgConn, error) {
 	return ConnectConfig(ctx, config)
 }
 
-// Connect establishes a connection to a PostgreSQL server using the environment and connString (in URL or DSN format)
-// and ParseConfigOptions to provide additional configuration. See documentation for [ParseConfig] for details. ctx can be
-// used to cancel a connect attempt.
+// Connect establishes a connection to a PostgreSQL server using the environment and connString (in URL or keyword/value
+// format) and ParseConfigOptions to provide additional configuration. See documentation for [ParseConfig] for details.
+// ctx can be used to cancel a connect attempt.
 func ConnectWithOptions(ctx context.Context, connString string, parseConfigOptions ParseConfigOptions) (*PgConn, error) {
 	config, err := ParseConfigWithOptions(connString, parseConfigOptions)
 	if err != nil {
@@ -131,113 +134,77 @@ func ConnectWithOptions(ctx context.Context, connString string, parseConfigOptio
 //
 // If config.Fallbacks are present they will sequentially be tried in case of error establishing network connection. An
 // authentication error will terminate the chain of attempts (like libpq:
-// https://www.postgresql.org/docs/11/libpq-connect.html#LIBPQ-MULTIPLE-HOSTS) and be returned as the error. Otherwise,
-// if all attempts fail the last error is returned.
-func ConnectConfig(octx context.Context, config *Config) (pgConn *PgConn, err error) {
+// https://www.postgresql.org/docs/11/libpq-connect.html#LIBPQ-MULTIPLE-HOSTS) and be returned as the error.
+func ConnectConfig(ctx context.Context, config *Config) (*PgConn, error) {
 	// Default values are set in ParseConfig. Enforce initial creation by ParseConfig rather than setting defaults from
 	// zero values.
 	if !config.createdByParseConfig {
 		panic("config must be created by ParseConfig")
 	}
 
-	// Simplify usage by treating primary config and fallbacks the same.
-	fallbackConfigs := []*FallbackConfig{
-		{
-			Host:      config.Host,
-			Port:      config.Port,
-			TLSConfig: config.TLSConfig,
-		},
-	}
-	fallbackConfigs = append(fallbackConfigs, config.Fallbacks...)
-	ctx := octx
-	fallbackConfigs, err = expandWithIPs(ctx, config.LookupFunc, fallbackConfigs)
-	if err != nil {
-		return nil, &ConnectError{Config: config, msg: "hostname resolving error", err: err}
-	}
+	var allErrors []error
 
-	if len(fallbackConfigs) == 0 {
-		return nil, &ConnectError{Config: config, msg: "hostname resolving error", err: errors.New("ip addr wasn't found")}
-	}
-
-	foundBestServer := false
-	var fallbackConfig *FallbackConfig
-	for i, fc := range fallbackConfigs {
-		// ConnectTimeout restricts the whole connection process.
-		if config.ConnectTimeout != 0 {
-			// create new context first time or when previous host was different
-			if i == 0 || (fallbackConfigs[i].Host != fallbackConfigs[i-1].Host) {
-				var cancel context.CancelFunc
-				ctx, cancel = context.WithTimeout(octx, config.ConnectTimeout)
-				defer cancel()
-			}
-		} else {
-			ctx = octx
-		}
-		pgConn, err = connect(ctx, config, fc, false)
-		if err == nil {
-			foundBestServer = true
-			break
-		} else if pgerr, ok := err.(*PgError); ok {
-			err = &ConnectError{Config: config, msg: "server error", err: pgerr}
-			const ERRCODE_INVALID_PASSWORD = "28P01"                    // wrong password
-			const ERRCODE_INVALID_AUTHORIZATION_SPECIFICATION = "28000" // wrong password or bad pg_hba.conf settings
-			const ERRCODE_INVALID_CATALOG_NAME = "3D000"                // db does not exist
-			const ERRCODE_INSUFFICIENT_PRIVILEGE = "42501"              // missing connect privilege
-			if pgerr.Code == ERRCODE_INVALID_PASSWORD ||
-				pgerr.Code == ERRCODE_INVALID_AUTHORIZATION_SPECIFICATION && fc.TLSConfig != nil ||
-				pgerr.Code == ERRCODE_INVALID_CATALOG_NAME ||
-				pgerr.Code == ERRCODE_INSUFFICIENT_PRIVILEGE {
-				break
-			}
-		} else if cerr, ok := err.(*ConnectError); ok {
-			if _, ok := cerr.err.(*NotPreferredError); ok {
-				fallbackConfig = fc
-			}
-		}
+	connectConfigs, errs := buildConnectOneConfigs(ctx, config)
+	if len(errs) > 0 {
+		allErrors = append(allErrors, errs...)
 	}
 
-	if !foundBestServer && fallbackConfig != nil {
-		pgConn, err = connect(ctx, config, fallbackConfig, true)
-		if pgerr, ok := err.(*PgError); ok {
-			err = &ConnectError{Config: config, msg: "server error", err: pgerr}
-		}
+	if len(connectConfigs) == 0 {
+		return nil, &ConnectError{Config: config, err: fmt.Errorf("hostname resolving error: %w", errors.Join(allErrors...))}
 	}
 
-	if err != nil {
-		return nil, err // no need to wrap in connectError because it will already be wrapped in all cases except PgError
+	pgConn, errs := connectPreferred(ctx, config, connectConfigs)
+	if len(errs) > 0 {
+		allErrors = append(allErrors, errs...)
+		return nil, &ConnectError{Config: config, err: errors.Join(allErrors...)}
 	}
 
 	if config.AfterConnect != nil {
 		err := config.AfterConnect(ctx, pgConn)
 		if err != nil {
 			pgConn.conn.Close()
-			return nil, &ConnectError{Config: config, msg: "AfterConnect error", err: err}
+			return nil, &ConnectError{Config: config, err: fmt.Errorf("AfterConnect error: %w", err)}
 		}
 	}
 
 	return pgConn, nil
 }
 
-func expandWithIPs(ctx context.Context, lookupFn LookupFunc, fallbacks []*FallbackConfig) ([]*FallbackConfig, error) {
-	var configs []*FallbackConfig
+// buildConnectOneConfigs resolves hostnames and builds a list of connectOneConfigs to try connecting to. It returns a
+// slice of successfully resolved connectOneConfigs and a slice of errors. It is possible for both slices to contain
+// values if some hosts were successfully resolved and others were not.
+func buildConnectOneConfigs(ctx context.Context, config *Config) ([]*connectOneConfig, []error) {
+	// Simplify usage by treating primary config and fallbacks the same.
+	fallbackConfigs := []*FallbackConfig{
+		{
+			Host:      config.Host,
+			Port:      config.Port,
+			TLSConfig: config.TLSConfig,
+		},
+	}
+	fallbackConfigs = append(fallbackConfigs, config.Fallbacks...)
+
+	var configs []*connectOneConfig
 
-	var lookupErrors []error
+	var allErrors []error
 
-	for _, fb := range fallbacks {
+	for _, fb := range fallbackConfigs {
 		// skip resolve for unix sockets
 		if isAbsolutePath(fb.Host) {
-			configs = append(configs, &FallbackConfig{
-				Host:      fb.Host,
-				Port:      fb.Port,
-				TLSConfig: fb.TLSConfig,
+			network, address := NetworkAddress(fb.Host, fb.Port)
+			configs = append(configs, &connectOneConfig{
+				network:          network,
+				address:          address,
+				originalHostname: fb.Host,
+				tlsConfig:        fb.TLSConfig,
 			})
 
 			continue
 		}
 
-		ips, err := lookupFn(ctx, fb.Host)
+		ips, err := config.LookupFunc(ctx, fb.Host)
 		if err != nil {
-			lookupErrors = append(lookupErrors, err)
+			allErrors = append(allErrors, err)
 			continue
 		}
 
@@ -246,63 +213,126 @@ func expandWithIPs(ctx context.Context, lookupFn LookupFunc, fallbacks []*Fallba
 			if err == nil {
 				port, err := strconv.ParseUint(splitPort, 10, 16)
 				if err != nil {
-					return nil, fmt.Errorf("error parsing port (%s) from lookup: %w", splitPort, err)
+					return nil, []error{fmt.Errorf("error parsing port (%s) from lookup: %w", splitPort, err)}
 				}
-				configs = append(configs, &FallbackConfig{
-					Host:      splitIP,
-					Port:      uint16(port),
-					TLSConfig: fb.TLSConfig,
+				network, address := NetworkAddress(splitIP, uint16(port))
+				configs = append(configs, &connectOneConfig{
+					network:          network,
+					address:          address,
+					originalHostname: fb.Host,
+					tlsConfig:        fb.TLSConfig,
 				})
 			} else {
-				configs = append(configs, &FallbackConfig{
-					Host:      ip,
-					Port:      fb.Port,
-					TLSConfig: fb.TLSConfig,
+				network, address := NetworkAddress(ip, fb.Port)
+				configs = append(configs, &connectOneConfig{
+					network:          network,
+					address:          address,
+					originalHostname: fb.Host,
+					tlsConfig:        fb.TLSConfig,
 				})
 			}
 		}
 	}
 
-	// See https://github.com/jackc/pgx/issues/1464. When Go 1.20 can be used in pgx consider using errors.Join so all
-	// errors are reported.
-	if len(configs) == 0 && len(lookupErrors) > 0 {
-		return nil, lookupErrors[0]
+	return configs, allErrors
+}
+
+// connectPreferred attempts to connect to the preferred host from connectOneConfigs. The connections are attempted in
+// order. If a connection is successful it is returned. If no connection is successful then all errors are returned. If
+// a connection attempt returns a [NotPreferredError], then that host will be used if no other hosts are successful.
+func connectPreferred(ctx context.Context, config *Config, connectOneConfigs []*connectOneConfig) (*PgConn, []error) {
+	octx := ctx
+	var allErrors []error
+
+	var fallbackConnectOneConfig *connectOneConfig
+	for i, c := range connectOneConfigs {
+		// ConnectTimeout restricts the whole connection process.
+		if config.ConnectTimeout != 0 {
+			// create new context first time or when previous host was different
+			if i == 0 || (connectOneConfigs[i].address != connectOneConfigs[i-1].address) {
+				var cancel context.CancelFunc
+				ctx, cancel = context.WithTimeout(octx, config.ConnectTimeout)
+				defer cancel()
+			}
+		} else {
+			ctx = octx
+		}
+
+		pgConn, err := connectOne(ctx, config, c, false)
+		if pgConn != nil {
+			return pgConn, nil
+		}
+
+		allErrors = append(allErrors, err)
+
+		var pgErr *PgError
+		if errors.As(err, &pgErr) {
+			const ERRCODE_INVALID_PASSWORD = "28P01"                    // wrong password
+			const ERRCODE_INVALID_AUTHORIZATION_SPECIFICATION = "28000" // wrong password or bad pg_hba.conf settings
+			const ERRCODE_INVALID_CATALOG_NAME = "3D000"                // db does not exist
+			const ERRCODE_INSUFFICIENT_PRIVILEGE = "42501"              // missing connect privilege
+			if pgErr.Code == ERRCODE_INVALID_PASSWORD ||
+				pgErr.Code == ERRCODE_INVALID_AUTHORIZATION_SPECIFICATION && c.tlsConfig != nil ||
+				pgErr.Code == ERRCODE_INVALID_CATALOG_NAME ||
+				pgErr.Code == ERRCODE_INSUFFICIENT_PRIVILEGE {
+				return nil, allErrors
+			}
+		}
+
+		var npErr *NotPreferredError
+		if errors.As(err, &npErr) {
+			fallbackConnectOneConfig = c
+		}
+	}
+
+	if fallbackConnectOneConfig != nil {
+		pgConn, err := connectOne(ctx, config, fallbackConnectOneConfig, true)
+		if err == nil {
+			return pgConn, nil
+		}
+		allErrors = append(allErrors, err)
 	}
 
-	return configs, nil
+	return nil, allErrors
 }
 
-func connect(ctx context.Context, config *Config, fallbackConfig *FallbackConfig,
+// connectOne makes one connection attempt to a single host.
+func connectOne(ctx context.Context, config *Config, connectConfig *connectOneConfig,
 	ignoreNotPreferredErr bool,
 ) (*PgConn, error) {
 	pgConn := new(PgConn)
 	pgConn.config = config
 	pgConn.cleanupDone = make(chan struct{})
+	pgConn.customData = make(map[string]any)
 
 	var err error
-	network, address := NetworkAddress(fallbackConfig.Host, fallbackConfig.Port)
-	netConn, err := config.DialFunc(ctx, network, address)
-	if err != nil {
-		return nil, &ConnectError{Config: config, msg: "dial error", err: normalizeTimeoutError(ctx, err)}
+
+	newPerDialConnectError := func(msg string, err error) *perDialConnectError {
+		err = normalizeTimeoutError(ctx, err)
+		e := &perDialConnectError{address: connectConfig.address, originalHostname: connectConfig.originalHostname, err: fmt.Errorf("%s: %w", msg, err)}
+		return e
 	}
 
-	pgConn.conn = netConn
-	pgConn.contextWatcher = newContextWatcher(netConn)
-	pgConn.contextWatcher.Watch(ctx)
+	pgConn.conn, err = config.DialFunc(ctx, connectConfig.network, connectConfig.address)
+	if err != nil {
+		return nil, newPerDialConnectError("dial error", err)
+	}
 
-	if fallbackConfig.TLSConfig != nil {
-		nbTLSConn, err := startTLS(netConn, fallbackConfig.TLSConfig)
+	if connectConfig.tlsConfig != nil {
+		pgConn.contextWatcher = ctxwatch.NewContextWatcher(&DeadlineContextWatcherHandler{Conn: pgConn.conn})
+		pgConn.contextWatcher.Watch(ctx)
+		tlsConn, err := startTLS(pgConn.conn, connectConfig.tlsConfig)
 		pgConn.contextWatcher.Unwatch() // Always unwatch `netConn` after TLS.
 		if err != nil {
-			netConn.Close()
-			return nil, &ConnectError{Config: config, msg: "tls error", err: normalizeTimeoutError(ctx, err)}
+			pgConn.conn.Close()
+			return nil, newPerDialConnectError("tls error", err)
 		}
 
-		pgConn.conn = nbTLSConn
-		pgConn.contextWatcher = newContextWatcher(nbTLSConn)
-		pgConn.contextWatcher.Watch(ctx)
+		pgConn.conn = tlsConn
 	}
 
+	pgConn.contextWatcher = ctxwatch.NewContextWatcher(config.BuildContextWatcherHandler(pgConn))
+	pgConn.contextWatcher.Watch(ctx)
 	defer pgConn.contextWatcher.Unwatch()
 
 	pgConn.parameterStatuses = make(map[string]string)
@@ -336,7 +366,7 @@ func connect(ctx context.Context, config *Config, fallbackConfig *FallbackConfig
 	pgConn.frontend.Send(&startupMsg)
 	if err := pgConn.flushWithPotentialWriteReadDeadlock(); err != nil {
 		pgConn.conn.Close()
-		return nil, &ConnectError{Config: config, msg: "failed to write startup message", err: normalizeTimeoutError(ctx, err)}
+		return nil, newPerDialConnectError("failed to write startup message", err)
 	}
 
 	for {
@@ -344,9 +374,9 @@ func connect(ctx context.Context, config *Config, fallbackConfig *FallbackConfig
 		if err != nil {
 			pgConn.conn.Close()
 			if err, ok := err.(*PgError); ok {
-				return nil, err
+				return nil, newPerDialConnectError("server error", err)
 			}
-			return nil, &ConnectError{Config: config, msg: "failed to receive message", err: normalizeTimeoutError(ctx, err)}
+			return nil, newPerDialConnectError("failed to receive message", err)
 		}
 
 		switch msg := msg.(type) {
@@ -359,26 +389,26 @@ func connect(ctx context.Context, config *Config, fallbackConfig *FallbackConfig
 			err = pgConn.txPasswordMessage(pgConn.config.Password)
 			if err != nil {
 				pgConn.conn.Close()
-				return nil, &ConnectError{Config: config, msg: "failed to write password message", err: err}
+				return nil, newPerDialConnectError("failed to write password message", err)
 			}
 		case *pgproto3.AuthenticationMD5Password:
 			digestedPassword := "md5" + hexMD5(hexMD5(pgConn.config.Password+pgConn.config.User)+string(msg.Salt[:]))
 			err = pgConn.txPasswordMessage(digestedPassword)
 			if err != nil {
 				pgConn.conn.Close()
-				return nil, &ConnectError{Config: config, msg: "failed to write password message", err: err}
+				return nil, newPerDialConnectError("failed to write password message", err)
 			}
 		case *pgproto3.AuthenticationSASL:
 			err = pgConn.scramAuth(msg.AuthMechanisms)
 			if err != nil {
 				pgConn.conn.Close()
-				return nil, &ConnectError{Config: config, msg: "failed SASL auth", err: err}
+				return nil, newPerDialConnectError("failed SASL auth", err)
 			}
 		case *pgproto3.AuthenticationGSS:
 			err = pgConn.gssAuth()
 			if err != nil {
 				pgConn.conn.Close()
-				return nil, &ConnectError{Config: config, msg: "failed GSS auth", err: err}
+				return nil, newPerDialConnectError("failed GSS auth", err)
 			}
 		case *pgproto3.ReadyForQuery:
 			pgConn.status = connStatusIdle
@@ -396,7 +426,7 @@ func connect(ctx context.Context, config *Config, fallbackConfig *FallbackConfig
 						return pgConn, nil
 					}
 					pgConn.conn.Close()
-					return nil, &ConnectError{Config: config, msg: "ValidateConnect failed", err: err}
+					return nil, newPerDialConnectError("ValidateConnect failed", err)
 				}
 			}
 			return pgConn, nil
@@ -404,21 +434,14 @@ func connect(ctx context.Context, config *Config, fallbackConfig *FallbackConfig
 			// handled by ReceiveMessage
 		case *pgproto3.ErrorResponse:
 			pgConn.conn.Close()
-			return nil, ErrorResponseToPgError(msg)
+			return nil, newPerDialConnectError("server error", ErrorResponseToPgError(msg))
 		default:
 			pgConn.conn.Close()
-			return nil, &ConnectError{Config: config, msg: "received unexpected message", err: err}
+			return nil, newPerDialConnectError("received unexpected message", err)
 		}
 	}
 }
 
-func newContextWatcher(conn net.Conn) *ctxwatch.ContextWatcher {
-	return ctxwatch.NewContextWatcher(
-		func() { conn.SetDeadline(time.Date(1, 1, 1, 1, 1, 1, 1, time.UTC)) },
-		func() { conn.SetDeadline(time.Time{}) },
-	)
-}
-
 func startTLS(conn net.Conn, tlsConfig *tls.Config) (net.Conn, error) {
 	err := binary.Write(conn, binary.BigEndian, []int32{8, 80877103})
 	if err != nil {
@@ -988,10 +1011,7 @@ func (pgConn *PgConn) CancelRequest(ctx context.Context) error {
 	defer cancelConn.Close()
 
 	if ctx != context.Background() {
-		contextWatcher := ctxwatch.NewContextWatcher(
-			func() { cancelConn.SetDeadline(time.Date(1, 1, 1, 1, 1, 1, 1, time.UTC)) },
-			func() { cancelConn.SetDeadline(time.Time{}) },
-		)
+		contextWatcher := ctxwatch.NewContextWatcher(&DeadlineContextWatcherHandler{Conn: cancelConn})
 		contextWatcher.Watch(ctx)
 		defer contextWatcher.Unwatch()
 	}
@@ -1524,8 +1544,10 @@ func (rr *ResultReader) Read() *Result {
 		values := rr.Values()
 		row := make([][]byte, len(values))
 		for i := range row {
-			row[i] = make([]byte, len(values[i]))
-			copy(row[i], values[i])
+			if values[i] != nil {
+				row[i] = make([]byte, len(values[i]))
+				copy(row[i], values[i])
+			}
 		}
 		br.Rows = append(br.Rows, row)
 	}
@@ -1880,6 +1902,11 @@ func (pgConn *PgConn) SyncConn(ctx context.Context) error {
 	return errors.New("SyncConn: conn never synchronized")
 }
 
+// CustomData returns a map that can be used to associate custom data with the connection.
+func (pgConn *PgConn) CustomData() map[string]any {
+	return pgConn.customData
+}
+
 // HijackedConn is the result of hijacking a connection.
 //
 // Due to the necessary exposure of internal implementation details, it is not covered by the semantic versioning
@@ -1892,6 +1919,7 @@ type HijackedConn struct {
 	TxStatus          byte
 	Frontend          *pgproto3.Frontend
 	Config            *Config
+	CustomData        map[string]any
 }
 
 // Hijack extracts the internal connection data. pgConn must be in an idle state. SyncConn should be called immediately
@@ -1914,6 +1942,7 @@ func (pgConn *PgConn) Hijack() (*HijackedConn, error) {
 		TxStatus:          pgConn.txStatus,
 		Frontend:          pgConn.frontend,
 		Config:            pgConn.config,
+		CustomData:        pgConn.customData,
 	}, nil
 }
 
@@ -1933,13 +1962,14 @@ func Construct(hc *HijackedConn) (*PgConn, error) {
 		txStatus:          hc.TxStatus,
 		frontend:          hc.Frontend,
 		config:            hc.Config,
+		customData:        hc.CustomData,
 
 		status: connStatusIdle,
 
 		cleanupDone: make(chan struct{}),
 	}
 
-	pgConn.contextWatcher = newContextWatcher(pgConn.conn)
+	pgConn.contextWatcher = ctxwatch.NewContextWatcher(hc.Config.BuildContextWatcherHandler(pgConn))
 	pgConn.bgReader = bgreader.New(pgConn.conn)
 	pgConn.slowWriteTimer = time.AfterFunc(time.Duration(math.MaxInt64),
 		func() {
@@ -2246,3 +2276,71 @@ func (p *Pipeline) Close() error {
 
 	return p.err
 }
+
+// DeadlineContextWatcherHandler handles canceled contexts by setting a deadline on a net.Conn.
+type DeadlineContextWatcherHandler struct {
+	Conn net.Conn
+
+	// DeadlineDelay is the delay to set on the deadline set on net.Conn when the context is canceled.
+	DeadlineDelay time.Duration
+}
+
+func (h *DeadlineContextWatcherHandler) HandleCancel(ctx context.Context) {
+	h.Conn.SetDeadline(time.Now().Add(h.DeadlineDelay))
+}
+
+func (h *DeadlineContextWatcherHandler) HandleUnwatchAfterCancel() {
+	h.Conn.SetDeadline(time.Time{})
+}
+
+// CancelRequestContextWatcherHandler handles canceled contexts by sending a cancel request to the server. It also sets
+// a deadline on a net.Conn as a fallback.
+type CancelRequestContextWatcherHandler struct {
+	Conn *PgConn
+
+	// CancelRequestDelay is the delay before sending the cancel request to the server.
+	CancelRequestDelay time.Duration
+
+	// DeadlineDelay is the delay to set on the deadline set on net.Conn when the context is canceled.
+	DeadlineDelay time.Duration
+
+	cancelFinishedChan             chan struct{}
+	handleUnwatchAfterCancelCalled func()
+}
+
+func (h *CancelRequestContextWatcherHandler) HandleCancel(context.Context) {
+	h.cancelFinishedChan = make(chan struct{})
+	var handleUnwatchedAfterCancelCalledCtx context.Context
+	handleUnwatchedAfterCancelCalledCtx, h.handleUnwatchAfterCancelCalled = context.WithCancel(context.Background())
+
+	deadline := time.Now().Add(h.DeadlineDelay)
+	h.Conn.conn.SetDeadline(deadline)
+
+	go func() {
+		defer close(h.cancelFinishedChan)
+
+		select {
+		case <-handleUnwatchedAfterCancelCalledCtx.Done():
+			return
+		case <-time.After(h.CancelRequestDelay):
+		}
+
+		cancelRequestCtx, cancel := context.WithDeadline(handleUnwatchedAfterCancelCalledCtx, deadline)
+		defer cancel()
+		h.Conn.CancelRequest(cancelRequestCtx)
+
+		// CancelRequest is inherently racy. Even though the cancel request has been received by the server at this point,
+		// it hasn't necessarily been delivered to the other connection. If we immediately return and the connection is
+		// immediately used then it is possible the CancelRequest will actually cancel our next query. The
+		// TestCancelRequestContextWatcherHandler Stress test can produce this error without the sleep below. The sleep time
+		// is arbitrary, but should be sufficient to prevent this error case.
+		time.Sleep(100 * time.Millisecond)
+	}()
+}
+
+func (h *CancelRequestContextWatcherHandler) HandleUnwatchAfterCancel() {
+	h.handleUnwatchAfterCancelCalled()
+	<-h.cancelFinishedChan
+
+	h.Conn.conn.SetDeadline(time.Time{})
+}
diff --git a/pgtype/interval.go b/pgtype/interval.go
index 21820938..06703d4d 100644
--- a/pgtype/interval.go
+++ b/pgtype/interval.go
@@ -132,22 +132,31 @@ func (encodePlanIntervalCodecText) Encode(value any, buf []byte) (newBuf []byte,
 
 	if interval.Days != 0 {
 		buf = append(buf, strconv.FormatInt(int64(interval.Days), 10)...)
-		buf = append(buf, " day "...)
+		buf = append(buf, " day"...)
 	}
 
-	absMicroseconds := interval.Microseconds
-	if absMicroseconds < 0 {
-		absMicroseconds = -absMicroseconds
-		buf = append(buf, '-')
-	}
+	if interval.Microseconds != 0 {
+		buf = append(buf, " "...)
+
+		absMicroseconds := interval.Microseconds
+		if absMicroseconds < 0 {
+			absMicroseconds = -absMicroseconds
+			buf = append(buf, '-')
+		}
+
+		hours := absMicroseconds / microsecondsPerHour
+		minutes := (absMicroseconds % microsecondsPerHour) / microsecondsPerMinute
+		seconds := (absMicroseconds % microsecondsPerMinute) / microsecondsPerSecond
 
-	hours := absMicroseconds / microsecondsPerHour
-	minutes := (absMicroseconds % microsecondsPerHour) / microsecondsPerMinute
-	seconds := (absMicroseconds % microsecondsPerMinute) / microsecondsPerSecond
-	microseconds := absMicroseconds % microsecondsPerSecond
+		timeStr := fmt.Sprintf("%02d:%02d:%02d", hours, minutes, seconds)
+		buf = append(buf, timeStr...)
+
+		microseconds := absMicroseconds % microsecondsPerSecond
+		if microseconds != 0 {
+			buf = append(buf, fmt.Sprintf(".%06d", microseconds)...)
+		}
+	}
 
-	timeStr := fmt.Sprintf("%02d:%02d:%02d.%06d", hours, minutes, seconds, microseconds)
-	buf = append(buf, timeStr...)
 	return buf, nil
 }
 
diff --git a/pgtype/pgtype_default.go b/pgtype/pgtype_default.go
index d56d95bd..38093ef4 100644
--- a/pgtype/pgtype_default.go
+++ b/pgtype/pgtype_default.go
@@ -82,8 +82,8 @@ func initDefaultMap() {
 	defaultMap.RegisterType(&Type{Name: "text", OID: TextOID, Codec: TextCodec{}})
 	defaultMap.RegisterType(&Type{Name: "tid", OID: TIDOID, Codec: TIDCodec{}})
 	defaultMap.RegisterType(&Type{Name: "time", OID: TimeOID, Codec: TimeCodec{}})
-	defaultMap.RegisterType(&Type{Name: "timestamp", OID: TimestampOID, Codec: TimestampCodec{}})
-	defaultMap.RegisterType(&Type{Name: "timestamptz", OID: TimestamptzOID, Codec: TimestamptzCodec{}})
+	defaultMap.RegisterType(&Type{Name: "timestamp", OID: TimestampOID, Codec: &TimestampCodec{}})
+	defaultMap.RegisterType(&Type{Name: "timestamptz", OID: TimestamptzOID, Codec: &TimestamptzCodec{}})
 	defaultMap.RegisterType(&Type{Name: "unknown", OID: UnknownOID, Codec: TextCodec{}})
 	defaultMap.RegisterType(&Type{Name: "uuid", OID: UUIDOID, Codec: UUIDCodec{}})
 	defaultMap.RegisterType(&Type{Name: "varbit", OID: VarbitOID, Codec: BitsCodec{}})
diff --git a/pgtype/time.go b/pgtype/time.go
index a3d0ab1a..61a3abdf 100644
--- a/pgtype/time.go
+++ b/pgtype/time.go
@@ -141,6 +141,8 @@ func (TimeCodec) PlanScan(m *Map, oid uint32, format int16, target any) ScanPlan
 		switch target.(type) {
 		case TimeScanner:
 			return scanPlanBinaryTimeToTimeScanner{}
+		case TextScanner:
+			return scanPlanBinaryTimeToTextScanner{}
 		}
 	case TextFormatCode:
 		switch target.(type) {
@@ -170,6 +172,34 @@ func (scanPlanBinaryTimeToTimeScanner) Scan(src []byte, dst any) error {
 	return scanner.ScanTime(Time{Microseconds: usec, Valid: true})
 }
 
+type scanPlanBinaryTimeToTextScanner struct{}
+
+func (scanPlanBinaryTimeToTextScanner) Scan(src []byte, dst any) error {
+	ts, ok := (dst).(TextScanner)
+	if !ok {
+		return ErrScanTargetTypeChanged
+	}
+
+	if src == nil {
+		return ts.ScanText(Text{})
+	}
+
+	if len(src) != 8 {
+		return fmt.Errorf("invalid length for time: %v", len(src))
+	}
+
+	usec := int64(binary.BigEndian.Uint64(src))
+
+	tim := Time{Microseconds: usec, Valid: true}
+
+	buf, err := TimeCodec{}.PlanEncode(nil, 0, TextFormatCode, tim).Encode(tim, nil)
+	if err != nil {
+		return err
+	}
+
+	return ts.ScanText(Text{String: string(buf), Valid: true})
+}
+
 type scanPlanTextAnyToTimeScanner struct{}
 
 func (scanPlanTextAnyToTimeScanner) Scan(src []byte, dst any) error {
diff --git a/pgtype/timestamp.go b/pgtype/timestamp.go
index 35d73956..677a2c6e 100644
--- a/pgtype/timestamp.go
+++ b/pgtype/timestamp.go
@@ -46,7 +46,7 @@ func (ts *Timestamp) Scan(src any) error {
 
 	switch src := src.(type) {
 	case string:
-		return scanPlanTextTimestampToTimestampScanner{}.Scan([]byte(src), ts)
+		return (&scanPlanTextTimestampToTimestampScanner{}).Scan([]byte(src), ts)
 	case time.Time:
 		*ts = Timestamp{Time: src, Valid: true}
 		return nil
@@ -116,17 +116,21 @@ func (ts *Timestamp) UnmarshalJSON(b []byte) error {
 	return nil
 }
 
-type TimestampCodec struct{}
+type TimestampCodec struct {
+	// ScanLocation is the location that the time is assumed to be in for scanning. This is different from
+	// TimestamptzCodec.ScanLocation in that this setting does change the instant in time that the timestamp represents.
+	ScanLocation *time.Location
+}
 
-func (TimestampCodec) FormatSupported(format int16) bool {
+func (*TimestampCodec) FormatSupported(format int16) bool {
 	return format == TextFormatCode || format == BinaryFormatCode
 }
 
-func (TimestampCodec) PreferredFormat() int16 {
+func (*TimestampCodec) PreferredFormat() int16 {
 	return BinaryFormatCode
 }
 
-func (TimestampCodec) PlanEncode(m *Map, oid uint32, format int16, value any) EncodePlan {
+func (*TimestampCodec) PlanEncode(m *Map, oid uint32, format int16, value any) EncodePlan {
 	if _, ok := value.(TimestampValuer); !ok {
 		return nil
 	}
@@ -220,27 +224,27 @@ func discardTimeZone(t time.Time) time.Time {
 	return t
 }
 
-func (TimestampCodec) PlanScan(m *Map, oid uint32, format int16, target any) ScanPlan {
+func (c *TimestampCodec) PlanScan(m *Map, oid uint32, format int16, target any) ScanPlan {
 
 	switch format {
 	case BinaryFormatCode:
 		switch target.(type) {
 		case TimestampScanner:
-			return scanPlanBinaryTimestampToTimestampScanner{}
+			return &scanPlanBinaryTimestampToTimestampScanner{location: c.ScanLocation}
 		}
 	case TextFormatCode:
 		switch target.(type) {
 		case TimestampScanner:
-			return scanPlanTextTimestampToTimestampScanner{}
+			return &scanPlanTextTimestampToTimestampScanner{location: c.ScanLocation}
 		}
 	}
 
 	return nil
 }
 
-type scanPlanBinaryTimestampToTimestampScanner struct{}
+type scanPlanBinaryTimestampToTimestampScanner struct{ location *time.Location }
 
-func (scanPlanBinaryTimestampToTimestampScanner) Scan(src []byte, dst any) error {
+func (plan *scanPlanBinaryTimestampToTimestampScanner) Scan(src []byte, dst any) error {
 	scanner := (dst).(TimestampScanner)
 
 	if src == nil {
@@ -264,15 +268,18 @@ func (scanPlanBinaryTimestampToTimestampScanner) Scan(src []byte, dst any) error
 			microsecFromUnixEpochToY2K/1000000+microsecSinceY2K/1000000,
 			(microsecFromUnixEpochToY2K%1000000*1000)+(microsecSinceY2K%1000000*1000),
 		).UTC()
+		if plan.location != nil {
+			tim = time.Date(tim.Year(), tim.Month(), tim.Day(), tim.Hour(), tim.Minute(), tim.Second(), tim.Nanosecond(), plan.location)
+		}
 		ts = Timestamp{Time: tim, Valid: true}
 	}
 
 	return scanner.ScanTimestamp(ts)
 }
 
-type scanPlanTextTimestampToTimestampScanner struct{}
+type scanPlanTextTimestampToTimestampScanner struct{ location *time.Location }
 
-func (scanPlanTextTimestampToTimestampScanner) Scan(src []byte, dst any) error {
+func (plan *scanPlanTextTimestampToTimestampScanner) Scan(src []byte, dst any) error {
 	scanner := (dst).(TimestampScanner)
 
 	if src == nil {
@@ -302,13 +309,17 @@ func (scanPlanTextTimestampToTimestampScanner) Scan(src []byte, dst any) error {
 			tim = time.Date(year, tim.Month(), tim.Day(), tim.Hour(), tim.Minute(), tim.Second(), tim.Nanosecond(), tim.Location())
 		}
 
+		if plan.location != nil {
+			tim = time.Date(tim.Year(), tim.Month(), tim.Day(), tim.Hour(), tim.Minute(), tim.Second(), tim.Nanosecond(), plan.location)
+		}
+
 		ts = Timestamp{Time: tim, Valid: true}
 	}
 
 	return scanner.ScanTimestamp(ts)
 }
 
-func (c TimestampCodec) DecodeDatabaseSQLValue(m *Map, oid uint32, format int16, src []byte) (driver.Value, error) {
+func (c *TimestampCodec) DecodeDatabaseSQLValue(m *Map, oid uint32, format int16, src []byte) (driver.Value, error) {
 	if src == nil {
 		return nil, nil
 	}
@@ -326,7 +337,7 @@ func (c TimestampCodec) DecodeDatabaseSQLValue(m *Map, oid uint32, format int16,
 	return ts.Time, nil
 }
 
-func (c TimestampCodec) DecodeValue(m *Map, oid uint32, format int16, src []byte) (any, error) {
+func (c *TimestampCodec) DecodeValue(m *Map, oid uint32, format int16, src []byte) (any, error) {
 	if src == nil {
 		return nil, nil
 	}
diff --git a/pgtype/timestamptz.go b/pgtype/timestamptz.go
index f568fe30..7efbcffd 100644
--- a/pgtype/timestamptz.go
+++ b/pgtype/timestamptz.go
@@ -54,7 +54,7 @@ func (tstz *Timestamptz) Scan(src any) error {
 
 	switch src := src.(type) {
 	case string:
-		return scanPlanTextTimestamptzToTimestamptzScanner{}.Scan([]byte(src), tstz)
+		return (&scanPlanTextTimestamptzToTimestamptzScanner{}).Scan([]byte(src), tstz)
 	case time.Time:
 		*tstz = Timestamptz{Time: src, Valid: true}
 		return nil
@@ -124,17 +124,21 @@ func (tstz *Timestamptz) UnmarshalJSON(b []byte) error {
 	return nil
 }
 
-type TimestamptzCodec struct{}
+type TimestamptzCodec struct {
+	// ScanLocation is the location to return scanned timestamptz values in. This does not change the instant in time that
+	// the timestamptz represents.
+	ScanLocation *time.Location
+}
 
-func (TimestamptzCodec) FormatSupported(format int16) bool {
+func (*TimestamptzCodec) FormatSupported(format int16) bool {
 	return format == TextFormatCode || format == BinaryFormatCode
 }
 
-func (TimestamptzCodec) PreferredFormat() int16 {
+func (*TimestamptzCodec) PreferredFormat() int16 {
 	return BinaryFormatCode
 }
 
-func (TimestamptzCodec) PlanEncode(m *Map, oid uint32, format int16, value any) EncodePlan {
+func (*TimestamptzCodec) PlanEncode(m *Map, oid uint32, format int16, value any) EncodePlan {
 	if _, ok := value.(TimestamptzValuer); !ok {
 		return nil
 	}
@@ -220,27 +224,27 @@ func (encodePlanTimestamptzCodecText) Encode(value any, buf []byte) (newBuf []by
 	return buf, nil
 }
 
-func (TimestamptzCodec) PlanScan(m *Map, oid uint32, format int16, target any) ScanPlan {
+func (c *TimestamptzCodec) PlanScan(m *Map, oid uint32, format int16, target any) ScanPlan {
 
 	switch format {
 	case BinaryFormatCode:
 		switch target.(type) {
 		case TimestamptzScanner:
-			return scanPlanBinaryTimestamptzToTimestamptzScanner{}
+			return &scanPlanBinaryTimestamptzToTimestamptzScanner{location: c.ScanLocation}
 		}
 	case TextFormatCode:
 		switch target.(type) {
 		case TimestamptzScanner:
-			return scanPlanTextTimestamptzToTimestamptzScanner{}
+			return &scanPlanTextTimestamptzToTimestamptzScanner{location: c.ScanLocation}
 		}
 	}
 
 	return nil
 }
 
-type scanPlanBinaryTimestamptzToTimestamptzScanner struct{}
+type scanPlanBinaryTimestamptzToTimestamptzScanner struct{ location *time.Location }
 
-func (scanPlanBinaryTimestamptzToTimestamptzScanner) Scan(src []byte, dst any) error {
+func (plan *scanPlanBinaryTimestamptzToTimestamptzScanner) Scan(src []byte, dst any) error {
 	scanner := (dst).(TimestamptzScanner)
 
 	if src == nil {
@@ -264,15 +268,18 @@ func (scanPlanBinaryTimestamptzToTimestamptzScanner) Scan(src []byte, dst any) e
 			microsecFromUnixEpochToY2K/1000000+microsecSinceY2K/1000000,
 			(microsecFromUnixEpochToY2K%1000000*1000)+(microsecSinceY2K%1000000*1000),
 		)
+		if plan.location != nil {
+			tim = tim.In(plan.location)
+		}
 		tstz = Timestamptz{Time: tim, Valid: true}
 	}
 
 	return scanner.ScanTimestamptz(tstz)
 }
 
-type scanPlanTextTimestamptzToTimestamptzScanner struct{}
+type scanPlanTextTimestamptzToTimestamptzScanner struct{ location *time.Location }
 
-func (scanPlanTextTimestamptzToTimestamptzScanner) Scan(src []byte, dst any) error {
+func (plan *scanPlanTextTimestamptzToTimestamptzScanner) Scan(src []byte, dst any) error {
 	scanner := (dst).(TimestamptzScanner)
 
 	if src == nil {
@@ -312,13 +319,17 @@ func (scanPlanTextTimestamptzToTimestamptzScanner) Scan(src []byte, dst any) err
 			tim = time.Date(year, tim.Month(), tim.Day(), tim.Hour(), tim.Minute(), tim.Second(), tim.Nanosecond(), tim.Location())
 		}
 
+		if plan.location != nil {
+			tim = tim.In(plan.location)
+		}
+
 		tstz = Timestamptz{Time: tim, Valid: true}
 	}
 
 	return scanner.ScanTimestamptz(tstz)
 }
 
-func (c TimestamptzCodec) DecodeDatabaseSQLValue(m *Map, oid uint32, format int16, src []byte) (driver.Value, error) {
+func (c *TimestamptzCodec) DecodeDatabaseSQLValue(m *Map, oid uint32, format int16, src []byte) (driver.Value, error) {
 	if src == nil {
 		return nil, nil
 	}
@@ -336,7 +347,7 @@ func (c TimestamptzCodec) DecodeDatabaseSQLValue(m *Map, oid uint32, format int1
 	return tstz.Time, nil
 }
 
-func (c TimestamptzCodec) DecodeValue(m *Map, oid uint32, format int16, src []byte) (any, error) {
+func (c *TimestamptzCodec) DecodeValue(m *Map, oid uint32, format int16, src []byte) (any, error) {
 	if src == nil {
 		return nil, nil
 	}
diff --git a/pgxpool/doc.go b/pgxpool/doc.go
index 06cc63d5..099443bc 100644
--- a/pgxpool/doc.go
+++ b/pgxpool/doc.go
@@ -8,7 +8,7 @@ The primary way of creating a pool is with [pgxpool.New]:
 
     pool, err := pgxpool.New(context.Background(), os.Getenv("DATABASE_URL"))
 
-The database connection string can be in URL or DSN format. PostgreSQL settings, pgx settings, and pool settings can be
+The database connection string can be in URL or keyword/value format. PostgreSQL settings, pgx settings, and pool settings can be
 specified here. In addition, a config struct can be created by [ParseConfig].
 
     config, err := pgxpool.ParseConfig(os.Getenv("DATABASE_URL"))
diff --git a/pgxpool/pool.go b/pgxpool/pool.go
index 9f74805e..6998e7e8 100644
--- a/pgxpool/pool.go
+++ b/pgxpool/pool.go
@@ -279,7 +279,7 @@ func NewWithConfig(ctx context.Context, config *Config) (*Pool, error) {
 //
 // See Config for definitions of these arguments.
 //
-//	# Example DSN
+//	# Example Keyword/Value
 //	user=jack password=secret host=pg.example.com port=5432 dbname=mydb sslmode=verify-ca pool_max_conns=10
 //
 //	# Example URL
diff --git a/rows.go b/rows.go
index 78723d6c..d4f7a901 100644
--- a/rows.go
+++ b/rows.go
@@ -419,6 +419,8 @@ type CollectableRow interface {
 type RowToFunc[T any] func(row CollectableRow) (T, error)
 
 // AppendRows iterates through rows, calling fn for each row, and appending the results into a slice of T.
+//
+// This function closes the rows automatically on return.
 func AppendRows[T any, S ~[]T](slice S, rows Rows, fn RowToFunc[T]) (S, error) {
 	defer rows.Close()
 
@@ -438,12 +440,16 @@ func AppendRows[T any, S ~[]T](slice S, rows Rows, fn RowToFunc[T]) (S, error) {
 }
 
 // CollectRows iterates through rows, calling fn for each row, and collecting the results into a slice of T.
+//
+// This function closes the rows automatically on return.
 func CollectRows[T any](rows Rows, fn RowToFunc[T]) ([]T, error) {
 	return AppendRows([]T{}, rows, fn)
 }
 
 // CollectOneRow calls fn for the first row in rows and returns the result. If no rows are found returns an error where errors.Is(ErrNoRows) is true.
 // CollectOneRow is to CollectRows as QueryRow is to Query.
+//
+// This function closes the rows automatically on return.
 func CollectOneRow[T any](rows Rows, fn RowToFunc[T]) (T, error) {
 	defer rows.Close()
 
@@ -469,6 +475,8 @@ func CollectOneRow[T any](rows Rows, fn RowToFunc[T]) (T, error) {
 // CollectExactlyOneRow calls fn for the first row in rows and returns the result.
 //   - If no rows are found returns an error where errors.Is(ErrNoRows) is true.
 //   - If more than 1 row is found returns an error where errors.Is(ErrTooManyRows) is true.
+//
+// This function closes the rows automatically on return.
 func CollectExactlyOneRow[T any](rows Rows, fn RowToFunc[T]) (T, error) {
 	defer rows.Close()
 
diff --git a/stdlib/sql.go b/stdlib/sql.go
index 3d65e23a..29cd3fbb 100644
--- a/stdlib/sql.go
+++ b/stdlib/sql.go
@@ -7,7 +7,7 @@
 //	  return err
 //	}
 //
-// Or from a DSN string.
+// Or from a keyword/value string.
 //
 //	db, err := sql.Open("pgx", "user=postgres password=secret host=localhost port=5432 database=pgx_test sslmode=disable")
 //	if err != nil {


Test patch:
diff --git a/pgconn/config_test.go b/pgconn/config_test.go
index 99cc6e1a..7f92bd53 100644
--- a/pgconn/config_test.go
+++ b/pgconn/config_test.go
@@ -336,7 +336,7 @@ func TestParseConfig(t *testing.T) {
 			},
 		},
 		{
-			name:       "DSN everything",
+			name:       "Key/value everything",
 			connString: "user=jack password=secret host=localhost port=5432 dbname=mydb sslmode=disable application_name=pgxtest search_path=myschema connect_timeout=5",
 			config: &pgconn.Config{
 				User:           "jack",
@@ -353,7 +353,7 @@ func TestParseConfig(t *testing.T) {
 			},
 		},
 		{
-			name:       "DSN with escaped single quote",
+			name:       "Key/value with escaped single quote",
 			connString: "user=jack\\'s password=secret host=localhost port=5432 dbname=mydb sslmode=disable",
 			config: &pgconn.Config{
 				User:          "jack's",
@@ -366,7 +366,7 @@ func TestParseConfig(t *testing.T) {
 			},
 		},
 		{
-			name:       "DSN with escaped backslash",
+			name:       "Key/value with escaped backslash",
 			connString: "user=jack password=sooper\\\\secret host=localhost port=5432 dbname=mydb sslmode=disable",
 			config: &pgconn.Config{
 				User:          "jack",
@@ -379,7 +379,7 @@ func TestParseConfig(t *testing.T) {
 			},
 		},
 		{
-			name:       "DSN with single quoted values",
+			name:       "Key/value with single quoted values",
 			connString: "user='jack' host='localhost' dbname='mydb' sslmode='disable'",
 			config: &pgconn.Config{
 				User:          "jack",
@@ -391,7 +391,7 @@ func TestParseConfig(t *testing.T) {
 			},
 		},
 		{
-			name:       "DSN with single quoted value with escaped single quote",
+			name:       "Key/value with single quoted value with escaped single quote",
 			connString: "user='jack\\'s' host='localhost' dbname='mydb' sslmode='disable'",
 			config: &pgconn.Config{
 				User:          "jack's",
@@ -403,7 +403,7 @@ func TestParseConfig(t *testing.T) {
 			},
 		},
 		{
-			name:       "DSN with empty single quoted value",
+			name:       "Key/value with empty single quoted value",
 			connString: "user='jack' password='' host='localhost' dbname='mydb' sslmode='disable'",
 			config: &pgconn.Config{
 				User:          "jack",
@@ -415,7 +415,7 @@ func TestParseConfig(t *testing.T) {
 			},
 		},
 		{
-			name:       "DSN with space between key and value",
+			name:       "Key/value with space between key and value",
 			connString: "user = 'jack' password = '' host = 'localhost' dbname = 'mydb' sslmode='disable'",
 			config: &pgconn.Config{
 				User:          "jack",
@@ -491,7 +491,7 @@ func TestParseConfig(t *testing.T) {
 			},
 		},
 		{
-			name:       "DSN multiple hosts one port",
+			name:       "Key/value multiple hosts one port",
 			connString: "user=jack password=secret host=foo,bar,baz port=5432 dbname=mydb sslmode=disable",
 			config: &pgconn.Config{
 				User:          "jack",
@@ -516,7 +516,7 @@ func TestParseConfig(t *testing.T) {
 			},
 		},
 		{
-			name:       "DSN multiple hosts multiple ports",
+			name:       "Key/value multiple hosts multiple ports",
 			connString: "user=jack password=secret host=foo,bar,baz port=1,2,3 dbname=mydb sslmode=disable",
 			config: &pgconn.Config{
 				User:          "jack",
@@ -772,18 +772,18 @@ func TestParseConfig(t *testing.T) {
 }
 
 // https://github.com/jackc/pgconn/issues/47
-func TestParseConfigDSNWithTrailingEmptyEqualDoesNotPanic(t *testing.T) {
+func TestParseConfigKVWithTrailingEmptyEqualDoesNotPanic(t *testing.T) {
 	_, err := pgconn.ParseConfig("host= user= password= port= database=")
 	require.NoError(t, err)
 }
 
-func TestParseConfigDSNLeadingEqual(t *testing.T) {
+func TestParseConfigKVLeadingEqual(t *testing.T) {
 	_, err := pgconn.ParseConfig("= user=jack")
 	require.Error(t, err)
 }
 
 // https://github.com/jackc/pgconn/issues/49
-func TestParseConfigDSNTrailingBackslash(t *testing.T) {
+func TestParseConfigKVTrailingBackslash(t *testing.T) {
 	_, err := pgconn.ParseConfig(`x=x\`)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "invalid backslash")
diff --git a/pgconn/internal/ctxwatch/context_watcher_test.go b/pgconn/ctxwatch/context_watcher_test.go
similarity index 66%
rename from pgconn/internal/ctxwatch/context_watcher_test.go
rename to pgconn/ctxwatch/context_watcher_test.go
index d62b97d3..302aabe3 100644
--- a/pgconn/internal/ctxwatch/context_watcher_test.go
+++ b/pgconn/ctxwatch/context_watcher_test.go
@@ -6,17 +6,32 @@ import (
 	"testing"
 	"time"
 
-	"github.com/jackc/pgx/v5/pgconn/internal/ctxwatch"
+	"github.com/jackc/pgx/v5/pgconn/ctxwatch"
 	"github.com/stretchr/testify/require"
 )
 
+type testHandler struct {
+	handleCancel             func(context.Context)
+	handleUnwatchAfterCancel func()
+}
+
+func (h *testHandler) HandleCancel(ctx context.Context) {
+	h.handleCancel(ctx)
+}
+
+func (h *testHandler) HandleUnwatchAfterCancel() {
+	h.handleUnwatchAfterCancel()
+}
+
 func TestContextWatcherContextCancelled(t *testing.T) {
 	canceledChan := make(chan struct{})
 	cleanupCalled := false
-	cw := ctxwatch.NewContextWatcher(func() {
-		canceledChan <- struct{}{}
-	}, func() {
-		cleanupCalled = true
+	cw := ctxwatch.NewContextWatcher(&testHandler{
+		handleCancel: func(context.Context) {
+			canceledChan <- struct{}{}
+		}, handleUnwatchAfterCancel: func() {
+			cleanupCalled = true
+		},
 	})
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -35,10 +50,12 @@ func TestContextWatcherContextCancelled(t *testing.T) {
 }
 
 func TestContextWatcherUnwatchdBeforeContextCancelled(t *testing.T) {
-	cw := ctxwatch.NewContextWatcher(func() {
-		t.Error("cancel func should not have been called")
-	}, func() {
-		t.Error("cleanup func should not have been called")
+	cw := ctxwatch.NewContextWatcher(&testHandler{
+		handleCancel: func(context.Context) {
+			t.Error("cancel func should not have been called")
+		}, handleUnwatchAfterCancel: func() {
+			t.Error("cleanup func should not have been called")
+		},
 	})
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -48,7 +65,7 @@ func TestContextWatcherUnwatchdBeforeContextCancelled(t *testing.T) {
 }
 
 func TestContextWatcherMultipleWatchPanics(t *testing.T) {
-	cw := ctxwatch.NewContextWatcher(func() {}, func() {})
+	cw := ctxwatch.NewContextWatcher(&testHandler{handleCancel: func(context.Context) {}, handleUnwatchAfterCancel: func() {}})
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -61,7 +78,7 @@ func TestContextWatcherMultipleWatchPanics(t *testing.T) {
 }
 
 func TestContextWatcherUnwatchWhenNotWatchingIsSafe(t *testing.T) {
-	cw := ctxwatch.NewContextWatcher(func() {}, func() {})
+	cw := ctxwatch.NewContextWatcher(&testHandler{handleCancel: func(context.Context) {}, handleUnwatchAfterCancel: func() {}})
 	cw.Unwatch() // unwatch when not / never watching
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -72,7 +89,7 @@ func TestContextWatcherUnwatchWhenNotWatchingIsSafe(t *testing.T) {
 }
 
 func TestContextWatcherUnwatchIsConcurrencySafe(t *testing.T) {
-	cw := ctxwatch.NewContextWatcher(func() {}, func() {})
+	cw := ctxwatch.NewContextWatcher(&testHandler{handleCancel: func(context.Context) {}, handleUnwatchAfterCancel: func() {}})
 
 	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
 	defer cancel()
@@ -88,10 +105,12 @@ func TestContextWatcherStress(t *testing.T) {
 	var cancelFuncCalls int64
 	var cleanupFuncCalls int64
 
-	cw := ctxwatch.NewContextWatcher(func() {
-		atomic.AddInt64(&cancelFuncCalls, 1)
-	}, func() {
-		atomic.AddInt64(&cleanupFuncCalls, 1)
+	cw := ctxwatch.NewContextWatcher(&testHandler{
+		handleCancel: func(context.Context) {
+			atomic.AddInt64(&cancelFuncCalls, 1)
+		}, handleUnwatchAfterCancel: func() {
+			atomic.AddInt64(&cleanupFuncCalls, 1)
+		},
 	})
 
 	cycleCount := 100000
@@ -134,7 +153,7 @@ func TestContextWatcherStress(t *testing.T) {
 }
 
 func BenchmarkContextWatcherUncancellable(b *testing.B) {
-	cw := ctxwatch.NewContextWatcher(func() {}, func() {})
+	cw := ctxwatch.NewContextWatcher(&testHandler{handleCancel: func(context.Context) {}, handleUnwatchAfterCancel: func() {}})
 
 	for i := 0; i < b.N; i++ {
 		cw.Watch(context.Background())
@@ -143,7 +162,7 @@ func BenchmarkContextWatcherUncancellable(b *testing.B) {
 }
 
 func BenchmarkContextWatcherCancelled(b *testing.B) {
-	cw := ctxwatch.NewContextWatcher(func() {}, func() {})
+	cw := ctxwatch.NewContextWatcher(&testHandler{handleCancel: func(context.Context) {}, handleUnwatchAfterCancel: func() {}})
 
 	for i := 0; i < b.N; i++ {
 		ctx, cancel := context.WithCancel(context.Background())
@@ -154,7 +173,7 @@ func BenchmarkContextWatcherCancelled(b *testing.B) {
 }
 
 func BenchmarkContextWatcherCancellable(b *testing.B) {
-	cw := ctxwatch.NewContextWatcher(func() {}, func() {})
+	cw := ctxwatch.NewContextWatcher(&testHandler{handleCancel: func(context.Context) {}, handleUnwatchAfterCancel: func() {}})
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
diff --git a/pgconn/errors_test.go b/pgconn/errors_test.go
index 171445cc..bbbfeb3c 100644
--- a/pgconn/errors_test.go
+++ b/pgconn/errors_test.go
@@ -19,12 +19,12 @@ func TestConfigError(t *testing.T) {
 			expectedMsg: "cannot parse `postgresql://foo:xxxxx@host`: msg",
 		},
 		{
-			name:        "dsn with password unquoted",
+			name:        "keyword/value with password unquoted",
 			err:         pgconn.NewParseConfigError("host=host password=password user=user", "msg", nil),
 			expectedMsg: "cannot parse `host=host password=xxxxx user=user`: msg",
 		},
 		{
-			name:        "dsn with password quoted",
+			name:        "keyword/value with password quoted",
 			err:         pgconn.NewParseConfigError("host=host password='pass word' user=user", "msg", nil),
 			expectedMsg: "cannot parse `host=host password=xxxxx user=user`: msg",
 		},
diff --git a/pgconn/pgconn_test.go b/pgconn/pgconn_test.go
index 604aa9e6..f9e3b4e1 100644
--- a/pgconn/pgconn_test.go
+++ b/pgconn/pgconn_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/jackc/pgx/v5/internal/pgio"
 	"github.com/jackc/pgx/v5/internal/pgmock"
 	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgconn/ctxwatch"
 	"github.com/jackc/pgx/v5/pgproto3"
 	"github.com/jackc/pgx/v5/pgtype"
 )
@@ -356,10 +357,8 @@ func TestConnectInvalidUser(t *testing.T) {
 
 	_, err = pgconn.ConnectConfig(ctx, config)
 	require.Error(t, err)
-	pgErr, ok := errors.Unwrap(err).(*pgconn.PgError)
-	if !ok {
-		t.Fatalf("Expected to receive a wrapped PgError, instead received: %v", err)
-	}
+	var pgErr *pgconn.PgError
+	require.ErrorAs(t, err, &pgErr)
 	if pgErr.Code != "28000" && pgErr.Code != "28P01" {
 		t.Fatalf("Expected to receive a PgError with code 28000 or 28P01, instead received: %v", pgErr)
 	}
@@ -528,6 +527,22 @@ func TestConnectWithFallback(t *testing.T) {
 	closeConn(t, conn)
 }
 
+func TestConnectFailsWithResolveFailureAndFailedConnectionAttempts(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	conn, err := pgconn.Connect(ctx, "host=localhost,127.0.0.1,foo.invalid port=1,2,3 sslmode=disable")
+	require.Error(t, err)
+	require.Nil(t, conn)
+
+	require.ErrorContains(t, err, "lookup foo.invalid")
+	// Not testing the entire string as depending on IPv4 or IPv6 support localhost may resolve to 127.0.0.1 or ::1.
+	require.ErrorContains(t, err, ":1 (localhost): dial error:")
+	require.ErrorContains(t, err, ":2 (127.0.0.1): dial error:")
+}
+
 func TestConnectWithValidateConnect(t *testing.T) {
 	t.Parallel()
 
@@ -1175,6 +1190,24 @@ func TestResultReaderValuesHaveSameCapacityAsLength(t *testing.T) {
 	ensureConnValid(t, pgConn)
 }
 
+// https://github.com/jackc/pgx/issues/1987
+func TestResultReaderReadNil(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+
+	pgConn, err := pgconn.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
+	require.NoError(t, err)
+	defer closeConn(t, pgConn)
+
+	result := pgConn.ExecParams(ctx, "select null::text", nil, nil, nil, nil).Read()
+	require.NoError(t, result.Err)
+	require.Nil(t, result.Rows[0][0])
+
+	ensureConnValid(t, pgConn)
+}
+
 func TestConnExecPrepared(t *testing.T) {
 	t.Parallel()
 
@@ -3184,6 +3217,22 @@ func TestConnOnPgError(t *testing.T) {
 	assert.True(t, pgConn.IsClosed())
 }
 
+func TestConnCustomData(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+
+	pgConn, err := pgconn.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
+	require.NoError(t, err)
+	defer closeConn(t, pgConn)
+
+	pgConn.CustomData()["foo"] = "bar"
+	assert.Equal(t, "bar", pgConn.CustomData()["foo"])
+
+	ensureConnValid(t, pgConn)
+}
+
 func Example() {
 	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
 	defer cancel()
@@ -3480,3 +3529,161 @@ func mustEncode(buf []byte, err error) []byte {
 	}
 	return buf
 }
+
+func TestDeadlineContextWatcherHandler(t *testing.T) {
+	t.Parallel()
+
+	t.Run("DeadlineExceeded with zero DeadlineDelay", func(t *testing.T) {
+		config, err := pgconn.ParseConfig(os.Getenv("PGX_TEST_DATABASE"))
+		require.NoError(t, err)
+		config.BuildContextWatcherHandler = func(conn *pgconn.PgConn) ctxwatch.Handler {
+			return &pgconn.DeadlineContextWatcherHandler{Conn: conn.Conn()}
+		}
+		config.ConnectTimeout = 5 * time.Second
+
+		pgConn, err := pgconn.ConnectConfig(context.Background(), config)
+		require.NoError(t, err)
+		defer closeConn(t, pgConn)
+
+		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel()
+
+		_, err = pgConn.Exec(ctx, "select 1, pg_sleep(1)").ReadAll()
+		require.Error(t, err)
+		require.ErrorIs(t, err, context.DeadlineExceeded)
+		require.True(t, pgConn.IsClosed())
+	})
+
+	t.Run("DeadlineExceeded with DeadlineDelay", func(t *testing.T) {
+		config, err := pgconn.ParseConfig(os.Getenv("PGX_TEST_DATABASE"))
+		require.NoError(t, err)
+		config.BuildContextWatcherHandler = func(conn *pgconn.PgConn) ctxwatch.Handler {
+			return &pgconn.DeadlineContextWatcherHandler{Conn: conn.Conn(), DeadlineDelay: 500 * time.Millisecond}
+		}
+		config.ConnectTimeout = 5 * time.Second
+
+		pgConn, err := pgconn.ConnectConfig(context.Background(), config)
+		require.NoError(t, err)
+		defer closeConn(t, pgConn)
+
+		ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+		defer cancel()
+
+		_, err = pgConn.Exec(ctx, "select 1, pg_sleep(0.250)").ReadAll()
+		require.NoError(t, err)
+
+		ensureConnValid(t, pgConn)
+	})
+}
+
+func TestCancelRequestContextWatcherHandler(t *testing.T) {
+	t.Parallel()
+
+	t.Run("DeadlineExceeded cancels request after CancelRequestDelay", func(t *testing.T) {
+		config, err := pgconn.ParseConfig(os.Getenv("PGX_TEST_DATABASE"))
+		require.NoError(t, err)
+		config.BuildContextWatcherHandler = func(conn *pgconn.PgConn) ctxwatch.Handler {
+			return &pgconn.CancelRequestContextWatcherHandler{
+				Conn:               conn,
+				CancelRequestDelay: 250 * time.Millisecond,
+				DeadlineDelay:      5000 * time.Millisecond,
+			}
+		}
+		config.ConnectTimeout = 5 * time.Second
+
+		pgConn, err := pgconn.ConnectConfig(context.Background(), config)
+		require.NoError(t, err)
+		defer closeConn(t, pgConn)
+
+		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel()
+
+		_, err = pgConn.Exec(ctx, "select 1, pg_sleep(3)").ReadAll()
+		require.Error(t, err)
+		var pgErr *pgconn.PgError
+		require.ErrorAs(t, err, &pgErr)
+
+		ensureConnValid(t, pgConn)
+	})
+
+	t.Run("DeadlineExceeded - do not send cancel request when query finishes in grace period", func(t *testing.T) {
+		config, err := pgconn.ParseConfig(os.Getenv("PGX_TEST_DATABASE"))
+		require.NoError(t, err)
+		config.BuildContextWatcherHandler = func(conn *pgconn.PgConn) ctxwatch.Handler {
+			return &pgconn.CancelRequestContextWatcherHandler{
+				Conn:               conn,
+				CancelRequestDelay: 1000 * time.Millisecond,
+				DeadlineDelay:      5000 * time.Millisecond,
+			}
+		}
+		config.ConnectTimeout = 5 * time.Second
+
+		pgConn, err := pgconn.ConnectConfig(context.Background(), config)
+		require.NoError(t, err)
+		defer closeConn(t, pgConn)
+
+		ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+		defer cancel()
+
+		_, err = pgConn.Exec(ctx, "select 1, pg_sleep(0.250)").ReadAll()
+		require.NoError(t, err)
+
+		ensureConnValid(t, pgConn)
+	})
+
+	t.Run("DeadlineExceeded sets conn deadline with DeadlineDelay", func(t *testing.T) {
+		config, err := pgconn.ParseConfig(os.Getenv("PGX_TEST_DATABASE"))
+		require.NoError(t, err)
+		config.BuildContextWatcherHandler = func(conn *pgconn.PgConn) ctxwatch.Handler {
+			return &pgconn.CancelRequestContextWatcherHandler{
+				Conn:               conn,
+				CancelRequestDelay: 5000 * time.Millisecond, // purposely setting this higher than DeadlineDelay to ensure the cancel request never happens.
+				DeadlineDelay:      250 * time.Millisecond,
+			}
+		}
+		config.ConnectTimeout = 5 * time.Second
+
+		pgConn, err := pgconn.ConnectConfig(context.Background(), config)
+		require.NoError(t, err)
+		defer closeConn(t, pgConn)
+
+		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel()
+
+		_, err = pgConn.Exec(ctx, "select 1, pg_sleep(1)").ReadAll()
+		require.Error(t, err)
+		require.ErrorIs(t, err, context.DeadlineExceeded)
+		require.True(t, pgConn.IsClosed())
+	})
+
+	for i := 0; i < 10; i++ {
+		t.Run(fmt.Sprintf("Stress %d", i), func(t *testing.T) {
+			t.Parallel()
+
+			config, err := pgconn.ParseConfig(os.Getenv("PGX_TEST_DATABASE"))
+			require.NoError(t, err)
+			config.BuildContextWatcherHandler = func(conn *pgconn.PgConn) ctxwatch.Handler {
+				return &pgconn.CancelRequestContextWatcherHandler{
+					Conn:               conn,
+					CancelRequestDelay: 5 * time.Millisecond,
+					DeadlineDelay:      1000 * time.Millisecond,
+				}
+			}
+			config.ConnectTimeout = 5 * time.Second
+
+			pgConn, err := pgconn.ConnectConfig(context.Background(), config)
+			require.NoError(t, err)
+			defer closeConn(t, pgConn)
+
+			for i := 0; i < 20; i++ {
+				func() {
+					ctx, cancel := context.WithTimeout(context.Background(), 4*time.Millisecond)
+					defer cancel()
+					pgConn.Exec(ctx, "select 1, pg_sleep(0.010)").ReadAll()
+					time.Sleep(100 * time.Millisecond) // ensure a cancel request that was a little late doesn't interrupt ensureConnValid.
+					ensureConnValid(t, pgConn)
+				}()
+			}
+		})
+	}
+}
diff --git a/pgtype/timestamp_test.go b/pgtype/timestamp_test.go
index 24f229d5..31b3ad82 100644
--- a/pgtype/timestamp_test.go
+++ b/pgtype/timestamp_test.go
@@ -38,6 +38,42 @@ func TestTimestampCodec(t *testing.T) {
 	})
 }
 
+func TestTimestampCodecWithScanLocationUTC(t *testing.T) {
+	skipCockroachDB(t, "Server does not support infinite timestamps (see https://github.com/cockroachdb/cockroach/issues/41564)")
+
+	connTestRunner := defaultConnTestRunner
+	connTestRunner.AfterConnect = func(ctx context.Context, t testing.TB, conn *pgx.Conn) {
+		conn.TypeMap().RegisterType(&pgtype.Type{
+			Name:  "timestamp",
+			OID:   pgtype.TimestampOID,
+			Codec: &pgtype.TimestampCodec{ScanLocation: time.UTC},
+		})
+	}
+
+	pgxtest.RunValueRoundTripTests(context.Background(), t, connTestRunner, nil, "timestamp", []pgxtest.ValueRoundTripTest{
+		// Have to use pgtype.Timestamp instead of time.Time as source because otherwise the simple and exec query exec
+		// modes will encode the time for timestamptz. That is, they will convert it from local time zone.
+		{pgtype.Timestamp{Time: time.Date(2000, 1, 1, 0, 0, 0, 0, time.Local), Valid: true}, new(time.Time), isExpectedEq(time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC))},
+	})
+}
+
+func TestTimestampCodecWithScanLocationLocal(t *testing.T) {
+	skipCockroachDB(t, "Server does not support infinite timestamps (see https://github.com/cockroachdb/cockroach/issues/41564)")
+
+	connTestRunner := defaultConnTestRunner
+	connTestRunner.AfterConnect = func(ctx context.Context, t testing.TB, conn *pgx.Conn) {
+		conn.TypeMap().RegisterType(&pgtype.Type{
+			Name:  "timestamp",
+			OID:   pgtype.TimestampOID,
+			Codec: &pgtype.TimestampCodec{ScanLocation: time.Local},
+		})
+	}
+
+	pgxtest.RunValueRoundTripTests(context.Background(), t, connTestRunner, nil, "timestamp", []pgxtest.ValueRoundTripTest{
+		{time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), new(time.Time), isExpectedEq(time.Date(2000, 1, 1, 0, 0, 0, 0, time.Local))},
+	})
+}
+
 // https://github.com/jackc/pgx/v4/pgtype/pull/128
 func TestTimestampTranscodeBigTimeBinary(t *testing.T) {
 	defaultConnTestRunner.RunTest(context.Background(), t, func(ctx context.Context, t testing.TB, conn *pgx.Conn) {
diff --git a/pgtype/timestamptz_test.go b/pgtype/timestamptz_test.go
index 0486ecdb..57248195 100644
--- a/pgtype/timestamptz_test.go
+++ b/pgtype/timestamptz_test.go
@@ -38,6 +38,40 @@ func TestTimestamptzCodec(t *testing.T) {
 	})
 }
 
+func TestTimestamptzCodecWithLocationUTC(t *testing.T) {
+	skipCockroachDB(t, "Server does not support infinite timestamps (see https://github.com/cockroachdb/cockroach/issues/41564)")
+
+	connTestRunner := defaultConnTestRunner
+	connTestRunner.AfterConnect = func(ctx context.Context, t testing.TB, conn *pgx.Conn) {
+		conn.TypeMap().RegisterType(&pgtype.Type{
+			Name:  "timestamptz",
+			OID:   pgtype.TimestamptzOID,
+			Codec: &pgtype.TimestamptzCodec{ScanLocation: time.UTC},
+		})
+	}
+
+	pgxtest.RunValueRoundTripTests(context.Background(), t, connTestRunner, nil, "timestamptz", []pgxtest.ValueRoundTripTest{
+		{time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC), new(time.Time), isExpectedEq(time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC))},
+	})
+}
+
+func TestTimestamptzCodecWithLocationLocal(t *testing.T) {
+	skipCockroachDB(t, "Server does not support infinite timestamps (see https://github.com/cockroachdb/cockroach/issues/41564)")
+
+	connTestRunner := defaultConnTestRunner
+	connTestRunner.AfterConnect = func(ctx context.Context, t testing.TB, conn *pgx.Conn) {
+		conn.TypeMap().RegisterType(&pgtype.Type{
+			Name:  "timestamptz",
+			OID:   pgtype.TimestamptzOID,
+			Codec: &pgtype.TimestamptzCodec{ScanLocation: time.Local},
+		})
+	}
+
+	pgxtest.RunValueRoundTripTests(context.Background(), t, connTestRunner, nil, "timestamptz", []pgxtest.ValueRoundTripTest{
+		{time.Date(2000, 1, 1, 0, 0, 0, 0, time.Local), new(time.Time), isExpectedEq(time.Date(2000, 1, 1, 0, 0, 0, 0, time.Local))},
+	})
+}
+
 // https://github.com/jackc/pgx/v4/pgtype/pull/128
 func TestTimestamptzTranscodeBigTimeBinary(t *testing.T) {
 	defaultConnTestRunner.RunTest(context.Background(), t, func(ctx context.Context, t testing.TB, conn *pgx.Conn) {