feat: added application_name

pkg/broker/adapter/postgres.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -22,6 +23,10 @@ type PostgresConfig struct {
 	MaxIdleConns    int
 	ConnMaxLifetime time.Duration
 	ConnMaxIdleTime time.Duration
+	// ApplicationName identifies this instance's pool connections in
+	// pg_stat_activity (e.g. "PGSQL_BROKER_INSTANCE1"). The LISTEN
+	// connection appends "_LISTENER" to this value.
+	ApplicationName string
 }
 
 // PostgresAdapter implements DBAdapter for PostgreSQL
@@ -170,7 +175,7 @@ func (p *PostgresAdapter) Query(ctx context.Context, query string, args ...inter
 
 // Listen starts listening on a PostgreSQL notification channel
 func (p *PostgresAdapter) Listen(ctx context.Context, channel string, handler NotificationHandler) error {
-	connStr := p.buildConnectionString()
+	connStr := p.buildConnectionStringWithAppName(p.config.ApplicationName + "_LISTENER")
 
 	reportProblem := func(ev pq.ListenerEventType, err error) {
 		if err != nil {
@@ -211,7 +216,11 @@ func (p *PostgresAdapter) Listen(ctx context.Context, channel string, handler No
 				p.logger.Info("stopping listener", "channel", channel)
 				return
 			case <-time.After(90 * time.Second):
-				SafeGo(p.logger, "listener-ping-"+channel, func() { listener.Ping() })
+				SafeGo(p.logger, "listener-ping-"+channel, func() {
+					if err := listener.Ping(); err != nil {
+						p.logger.Error("listener ping failed", "channel", channel, "error", err)
+					}
+				})
 			}
 		}
 	})
@@ -232,24 +241,42 @@ func (p *PostgresAdapter) Unlisten(ctx context.Context, channel string) error {
 	return listener.Unlisten(channel)
 }
 
-// buildConnectionString builds a PostgreSQL connection string
+// buildConnectionString builds a PostgreSQL connection string for the
+// pooled connection, using the adapter's own application name.
 func (p *PostgresAdapter) buildConnectionString() string {
+	return p.buildConnectionStringWithAppName(p.config.ApplicationName)
+}
+
+// buildConnectionStringWithAppName builds a PostgreSQL connection string
+// with the given application_name, so pooled and LISTEN connections can be
+// told apart in pg_stat_activity.
+func (p *PostgresAdapter) buildConnectionStringWithAppName(appName string) string {
 	sslMode := p.config.SSLMode
 	if sslMode == "" {
 		sslMode = "disable"
 	}
 
 	return fmt.Sprintf(
-		"host=%s port=%d user=%s password=%s dbname=%s sslmode=%s options='-c search_path=broker,public'",
+		"host=%s port=%d user=%s password=%s dbname=%s sslmode=%s application_name=%s options='-c search_path=broker,public'",
 		p.config.Host,
 		p.config.Port,
 		p.config.User,
 		p.config.Password,
 		p.config.Database,
 		sslMode,
+		quoteDSNValue(appName),
 	)
 }
 
+// quoteDSNValue escapes a value for use in a libpq keyword/value connection
+// string, single-quoting it and backslash-escaping embedded backslashes and
+// quotes per the libpq connection string format.
+func quoteDSNValue(v string) string {
+	v = strings.ReplaceAll(v, `\`, `\\`)
+	v = strings.ReplaceAll(v, `'`, `\'`)
+	return "'" + v + "'"
+}
+
 // Conn returns a single physical connection pinned out of the pool, for
 // session-scoped operations (e.g. pg_try_advisory_lock) that must survive
 // across calls and must not be silently reaped or handed to another caller

pkg/broker/broker.go

@@ -42,14 +42,15 @@ func (b *Broker) Start() error {
 	b.logger.Info("starting broker", "database_count", len(b.config.Databases))
 
 	// Create and start an instance for each database
-	for i, dbCfg := range b.config.Databases {
+	for i := range b.config.Databases {
+		dbCfg := &b.config.Databases[i]
 		b.logger.Info("starting database instance", "name", dbCfg.Name, "host", dbCfg.Host, "database", dbCfg.Database)
 
 		// Create database adapter
 		dbAdapter := adapter.NewPostgresAdapter(dbCfg.ToPostgresConfig(), b.logger)
 
 		// Create database instance
-		instance, err := NewDatabaseInstance(b.config, &dbCfg, dbAdapter, b.logger, b.version, b.ctx)
+		instance, err := NewDatabaseInstance(b.config, dbCfg, dbAdapter, b.logger, b.version, b.ctx)
 		if err != nil {
 			// Stop any already-started instances
 			b.stopInstances()

pkg/broker/config/config.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"fmt"
+	"strings"
 	"time"
 
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
@@ -47,7 +48,7 @@ type BrokerConfig struct {
 	QueueTimerSec        int           `mapstructure:"queue_timer_sec"`
 	QueueBufferSize      int           `mapstructure:"queue_buffer_size"`
 	WorkerIdleTimeoutSec int           `mapstructure:"worker_idle_timeout_sec"`
-	NotifyRetrySeconds   time.Duration `mapstructure:"notify_retry_seconds"`
+	NotifyRetryInterval  time.Duration `mapstructure:"notify_retry_seconds"`
 	EnableDebug          bool          `mapstructure:"enable_debug"`
 	// LeaseSeconds is how long a claimed job's lease is valid for before
 	// broker_recover_stale_jobs considers it abandoned.
@@ -132,7 +133,8 @@ func validateConfig(config *Config) error {
 	}
 
 	// Validate each database configuration
-	for i, db := range config.Databases {
+	for i := range config.Databases {
+		db := &config.Databases[i]
 		if db.Name == "" {
 			return fmt.Errorf("database[%d]: name is required", i)
 		}
@@ -195,5 +197,6 @@ func (d *DatabaseConfig) ToPostgresConfig() adapter.PostgresConfig {
 		MaxIdleConns:    d.MaxIdleConns,
 		ConnMaxLifetime: d.ConnMaxLifetime,
 		ConnMaxIdleTime: d.ConnMaxIdleTime,
+		ApplicationName: fmt.Sprintf("PGSQL_BROKER_%s", strings.ToUpper(d.Name)),
 	}
 }

pkg/broker/install/install.go

@@ -205,7 +205,9 @@ func (i *Installer) ApplyMigrations(ctx context.Context) error {
 		}
 
 		if err := execStatements(ctx, tx, string(content)); err != nil {
-			tx.Rollback()
+			if rbErr := tx.Rollback(); rbErr != nil {
+				i.logger.Error("failed to rollback migration transaction", "version", m.version, "error", rbErr)
+			}
 			return fmt.Errorf("failed to apply migration %s: %w", m.name, err)
 		}
 
@@ -213,7 +215,9 @@ func (i *Installer) ApplyMigrations(ctx context.Context) error {
 			"INSERT INTO broker.broker_schema_migrations (version, name) VALUES ($1, $2)",
 			m.version, m.name,
 		); err != nil {
-			tx.Rollback()
+			if rbErr := tx.Rollback(); rbErr != nil {
+				i.logger.Error("failed to rollback migration transaction", "version", m.version, "error", rbErr)
+			}
 			return fmt.Errorf("failed to record migration %s: %w", m.name, err)
 		}
 
@@ -293,7 +297,9 @@ func (i *Installer) InstallRoles(ctx context.Context, passwords RolePasswords) e
 		rendered := replacer.Replace(string(content))
 
 		if err := execStatements(ctx, tx, rendered); err != nil {
-			tx.Rollback()
+			if rbErr := tx.Rollback(); rbErr != nil {
+				i.logger.Error("failed to rollback roles script transaction", "name", name, "error", rbErr)
+			}
 			return fmt.Errorf("failed to apply roles script %s: %w", name, err)
 		}
 

pkg/broker/worker/worker.go

@@ -204,26 +204,36 @@ func (w *Worker) processJobs(ctx context.Context) {
 		}
 
 		if err := w.setTenantTx(ctx, tx); err != nil {
-			tx.Rollback()
+			if rbErr := tx.Rollback(); rbErr != nil {
+				w.logger.Error("failed to rollback transaction", "error", rbErr)
+			}
 			w.logger.Error("failed to set tenant", "error", err)
 			return
 		}
 
 		jobID, leaseToken, err := w.fetchNextJobTx(ctx, tx)
 		if err != nil {
-			tx.Rollback()
+			if rbErr := tx.Rollback(); rbErr != nil {
+				w.logger.Error("failed to rollback transaction", "error", rbErr)
+			}
 			w.logger.Error("failed to fetch job", "error", err)
 			return
 		}
 
 		if jobID <= 0 {
-			tx.Rollback() // No job found, rollback
-			return        // No more jobs
+			// No job found, rollback
+			if rbErr := tx.Rollback(); rbErr != nil {
+				w.logger.Error("failed to rollback transaction", "error", rbErr)
+			}
+			return // No more jobs
 		}
 
 		// Run the job
 		if err := w.runJobTx(ctx, tx, jobID, leaseToken); err != nil {
-			tx.Rollback() // Rollback on genuine infra failure
+			// Rollback on genuine infra failure
+			if rbErr := tx.Rollback(); rbErr != nil {
+				w.logger.Error("failed to rollback transaction", "error", rbErr)
+			}
 			w.logger.Error("failed to run job", "job_id", jobID, "error", err)
 		} else {
 			if err := tx.Commit(); err != nil {
@@ -247,13 +257,13 @@ func (w *Worker) setTenantTx(ctx context.Context, tx adapter.DBTransaction) erro
 
 // fetchNextJobTx fetches the next job from the queue within a transaction,
 // claiming it with a lease that must be presented back to broker_run.
-func (w *Worker) fetchNextJobTx(ctx context.Context, tx adapter.DBTransaction) (int64, string, error) {
+func (w *Worker) fetchNextJobTx(ctx context.Context, tx adapter.DBTransaction) (jobID int64, leaseToken string, err error) {
 	var retval int
 	var errmsg string
 	var nullableJobID sql.NullInt64
 	var nullableLeaseToken sql.NullString
 
-	err := tx.QueryRow(ctx,
+	err = tx.QueryRow(ctx,
 		"SELECT p_retval, p_errmsg, p_job_id, p_lease_token FROM broker.broker_get($1, $2, $3)",
 		w.QueueNumber, w.InstanceID, w.leaseSeconds,
 	).Scan(&retval, &errmsg, &nullableJobID, &nullableLeaseToken)