[*] set lock_timeout via connection RuntimeParams instead of per-query transaction

Reduces metric queries from 4 roundtrips (BEGIN/SET/query/COMMIT) to 1 by
setting lock_timeout at connection level via pgx RuntimeParams.

- Add --conn-lock-timeout option (default: 100ms, env: PW_CONN_LOCK_TIMEOUT)
- Set lock_timeout in RuntimeParams for PostgreSQL sources on connect
- Simplify QueryMeasurements by removing transaction wrapper
- Set to "0" to disable lock_timeout

Ref: https://brandur.org/fragments/postgres-parameters

Author: claude

internal/reaper/database.go

@@ -8,7 +8,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/cybertec-postgresql/pgwatch/v3/internal/db"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/log"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/metrics"
 	"github.com/cybertec-postgresql/pgwatch/v3/internal/sinks"
@@ -17,30 +16,16 @@ import (
 )
 
 func QueryMeasurements(ctx context.Context, md *sources.SourceConn, sql string, args ...any) (metrics.Measurements, error) {
-	var conn db.PgxIface
-	var err error
-	var tx pgx.Tx
 	if strings.TrimSpace(sql) == "" {
 		return nil, errors.New("empty SQL")
 	}
 
-	conn = md.Conn
-	if md.IsPostgresSource() {
-		// we don't want transaction for non-postgres sources, e.g. pgbouncer
-		if tx, err = conn.Begin(ctx); err != nil {
-			return nil, err
-		}
-		defer func() { _ = tx.Commit(ctx) }()
-		_, err = tx.Exec(ctx, "SET LOCAL lock_timeout TO '100ms'")
-		if err != nil {
-			return nil, err
-		}
-		conn = tx
-	} else {
-		// we want simple protocol for non-postgres connections, e.g. pgpool
+	// For non-postgres connections (e.g. pgbouncer, pgpool), use simple protocol
+	if !md.IsPostgresSource() {
 		args = append([]any{pgx.QueryExecModeSimpleProtocol}, args...)
 	}
-	rows, err := conn.Query(ctx, sql, args...)
+	// lock_timeout is set at connection level via RuntimeParams, no need for transaction wrapper
+	rows, err := md.Conn.Query(ctx, sql, args...)
 	if err == nil {
 		return pgx.CollectRows(rows, metrics.RowToMeasurement)
 	}

internal/sources/cmdopts.go

@@ -9,4 +9,5 @@ type CmdOpts struct {
 	MaxParallelConnectionsPerDb  int      `long:"max-parallel-connections-per-db" mapstructure:"max-parallel-connections-per-db" description:"Max parallel metric fetches per DB. Note the multiplication effect on multi-DB instances" env:"PW_MAX_PARALLEL_CONNECTIONS_PER_DB" default:"4"`
 	TryCreateListedExtsIfMissing string   `long:"try-create-listed-exts-if-missing" mapstructure:"try-create-listed-exts-if-missing" description:"Try creating the listed extensions (comma sep.) on first connect for all monitored DBs when missing. Main usage - pg_stat_statements" env:"PW_TRY_CREATE_LISTED_EXTS_IF_MISSING" default:""`
 	CreateHelpers                bool     `long:"create-helpers" mapstructure:"create-helpers" description:"Create helper database objects from metric definitions" env:"PW_CREATE_HELPERS"`
+	ConnLockTimeout              string   `long:"conn-lock-timeout" mapstructure:"conn-lock-timeout" description:"PostgreSQL lock_timeout for metric query connections. Set to 0 to disable" env:"PW_CONN_LOCK_TIMEOUT" default:"100ms"`
 }

internal/sources/conn.go

@@ -89,6 +89,11 @@ func (md *SourceConn) Connect(ctx context.Context, opts CmdOpts) (err error) {
 		if opts.MaxParallelConnectionsPerDb > 0 {
 			md.ConnConfig.MaxConns = int32(opts.MaxParallelConnectionsPerDb)
 		}
+		// Set lock_timeout at connection level for PostgreSQL sources to avoid
+		// wrapping every query in a transaction with SET LOCAL lock_timeout
+		if md.IsPostgresSource() && opts.ConnLockTimeout != "" && opts.ConnLockTimeout != "0" {
+			md.ConnConfig.ConnConfig.RuntimeParams["lock_timeout"] = opts.ConnLockTimeout
+		}
 		md.Conn, err = NewConnWithConfig(ctx, md.ConnConfig)
 		if err != nil {
 			return err