From 7c8d0bdc99d0954da7ef04b408f82d48a7ce1870 Mon Sep 17 00:00:00 2001 From: Hein Date: Fri, 18 Sep 2026 23:08:31 +0200 Subject: [PATCH] feat: added application_name --- .github/workflows/integration.yml | 11 ++++++++++ README.md | 8 +++++++ cmd/broker/main.go | 3 ++- pkg/broker/adapter/postgres.go | 35 ++++++++++++++++++++++++++---- pkg/broker/broker.go | 5 +++-- pkg/broker/config/config.go | 7 ++++-- pkg/broker/install/install.go | 12 +++++++--- pkg/broker/worker/worker.go | 24 ++++++++++++++------ tests/integration/workflow_test.go | 2 +- 9 files changed, 87 insertions(+), 20 deletions(-) diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 0e8ed49..3e12e5d 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -35,6 +35,17 @@ jobs: go-version: '1.26' cache: true + - name: Check formatting + run: | + make fmt + git diff --exit-code -- '*.go' + + - name: Install golangci-lint + run: go install github.com/golangci/golangci-lint/cmd/golangci-lint@latest + + - name: Run lint + run: make lint + - name: Run all tests env: # act_runner runs this job in its own container alongside the diff --git a/README.md b/README.md index 3c01b9b..49fcc58 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,14 @@ A robust, event-driven job processing system for PostgreSQL that uses LISTEN/NOTIFY for real-time job execution. It supports multiple queues, priority-based scheduling, multi-tenant row-level security, and can be used both as a standalone service or as a Go library. +## Status +[![Integration Tests](https://git.warky.dev/wdevs/pgsql-broker/actions/workflows/integration.yml/badge.svg?branch=main)](https://git.warky.dev/wdevs/pgsql-broker/actions?workflow=integration.yml) + +[![Release](https://git.warky.dev/wdevs/pgsql-broker/actions/workflows/release.yml/badge.svg?branch=main)](https://git.warky.dev/wdevs/pgsql-broker/actions?workflow=release.yml) + +[![Build & Release Docker Image](https://git.warky.dev/wdevs/pgsql-broker/actions/workflows/docker-release.yml/badge.svg?branch=main)](https://git.warky.dev/wdevs/pgsql-broker/actions?workflow=docker-release.yml) + + ## Features - **Multi-Database Support**: Single broker process can manage multiple database connections diff --git a/cmd/broker/main.go b/cmd/broker/main.go index b81bbce..e92159e 100644 --- a/cmd/broker/main.go +++ b/cmd/broker/main.go @@ -227,7 +227,8 @@ func runInstall() error { } // Install/verify on all configured databases - for i, dbCfg := range cfg.Databases { + for i := range cfg.Databases { + dbCfg := &cfg.Databases[i] logger.Info("processing database", "index", i, "name", dbCfg.Name, "host", dbCfg.Host, "database", dbCfg.Database) // Create database adapter. With --with-roles, the config file's own diff --git a/pkg/broker/adapter/postgres.go b/pkg/broker/adapter/postgres.go index abed6cf..3ac3459 100644 --- a/pkg/broker/adapter/postgres.go +++ b/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 diff --git a/pkg/broker/broker.go b/pkg/broker/broker.go index fc6f3fc..1a51605 100644 --- a/pkg/broker/broker.go +++ b/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() diff --git a/pkg/broker/config/config.go b/pkg/broker/config/config.go index d1041df..5a901c3 100644 --- a/pkg/broker/config/config.go +++ b/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)), } } diff --git a/pkg/broker/install/install.go b/pkg/broker/install/install.go index 2297a91..e8b835a 100644 --- a/pkg/broker/install/install.go +++ b/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) } diff --git a/pkg/broker/worker/worker.go b/pkg/broker/worker/worker.go index dcec69a..fa53951 100644 --- a/pkg/broker/worker/worker.go +++ b/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) diff --git a/tests/integration/workflow_test.go b/tests/integration/workflow_test.go index 6905cfe..ef2ef43 100644 --- a/tests/integration/workflow_test.go +++ b/tests/integration/workflow_test.go @@ -94,7 +94,7 @@ func TestBrokerWorkflow(t *testing.T) { QueueTimerSec: 1, // Short interval for testing QueueBufferSize: 10, WorkerIdleTimeoutSec: 5, - NotifyRetrySeconds: 5 * time.Second, + NotifyRetryInterval: 5 * time.Second, EnableDebug: true, }, Logging: config.LoggingConfig{