Merge pull request 'feat: add Prometheus metrics and dashboard' (#6) from issue-4-prometheus-metrics into main

Reviewed-on: #6
Reviewed-by: Warky <2+warkanum@noreply@warky.dev>

.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/v2/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

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
@@ -275,6 +283,16 @@ docker run --rm -v $(pwd)/broker.yaml:/etc/pgsql-broker/broker.yaml:ro git.warky
 
 Config must be mounted at `/etc/pgsql-broker/broker.yaml` — no config is baked into the image.
 
+### Prebuilt image: `docker-compose.prebuilt.yml`
+
+Same stack as `docker-compose.yml` below, but pulls `git.warky.dev/wdevs/pgsql-broker:latest` instead of building from source — no local Go toolchain or Dockerfile needed.
+
+```bash
+cp broker.docker.example.yaml broker.docker.yaml   # set broker_runtime password
+cp .env.example .env                                # set POSTGRES_PASSWORD + 3 BROKER_*_PASSWORD
+docker-compose -f docker-compose.prebuilt.yml up -d
+```
+
 ### Building the image locally
 
 ```bash
@@ -365,8 +383,12 @@ Global settings applied to all database instances:
 | `worker_idle_timeout_sec` | Worker idle timeout | `10` |
 | `notify_retry_seconds` | NOTIFY retry interval | `30s` |
 | `enable_debug` | Enable debug logging | `false` |
-| `lease_seconds` | Job lease duration before reclaimable | - |
+| `lease_seconds` | Job lease duration before reclaimable | `60` |
-| `stale_job_recovery_sec` | Interval for reclaiming expired leases | - |
+| `stale_job_recovery_sec` | Interval for reclaiming expired leases | `30` |
+| `metrics_enabled` | Enable Prometheus and the embedded dashboard | `true` |
+| `metrics_host` | Metrics HTTP bind host | `0.0.0.0` |
+| `metrics_port` | Metrics HTTP bind port | `9469` |
+| `queue_depth_poll_sec` | Queue depth collection interval | `15` |
 
 ## Development
 

broker.example.yaml

@@ -41,6 +41,10 @@ broker:
   worker_idle_timeout_sec: 10  # Worker idle timeout
   notify_retry_seconds: 30s  # LISTEN/NOTIFY retry interval
   enable_debug: false  # Enable debug logging
+  metrics_enabled: true # Expose Prometheus and the embedded dashboard
+  metrics_host: 0.0.0.0
+  metrics_port: 9469
+  queue_depth_poll_sec: 15
 
 # Logging settings
 logging:

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

docker-compose.prebuilt.yml

@@ -0,0 +1,46 @@
+services:
+  postgres:
+    image: docker.io/library/postgres:16-alpine
+    environment:
+      POSTGRES_DB: broker
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env}
+    volumes:
+      - postgres-data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U postgres -d broker"]
+      interval: 2s
+      timeout: 3s
+      retries: 30
+    restart: unless-stopped
+
+  # One-shot: applies migrations and creates/rotates the least-privilege
+  # broker_admin/broker_runtime/broker_enqueue roles, then exits. The
+  # broker service below only starts once this completes successfully.
+  migrate:
+    image: git.warky.dev/wdevs/pgsql-broker:latest
+    depends_on:
+      postgres:
+        condition: service_healthy
+    volumes:
+      - ./broker.docker.yaml:/etc/pgsql-broker/broker.yaml:ro
+    environment:
+      PGUSER: postgres
+      PGPASSWORD: ${POSTGRES_PASSWORD:?set POSTGRES_PASSWORD in .env}
+      BROKER_ADMIN_PASSWORD: ${BROKER_ADMIN_PASSWORD:?set BROKER_ADMIN_PASSWORD in .env}
+      BROKER_RUNTIME_PASSWORD: ${BROKER_RUNTIME_PASSWORD:?set BROKER_RUNTIME_PASSWORD in .env}
+      BROKER_ENQUEUE_PASSWORD: ${BROKER_ENQUEUE_PASSWORD:?set BROKER_ENQUEUE_PASSWORD in .env}
+    command: ["install", "--with-roles"]
+    restart: "no"
+
+  broker:
+    image: git.warky.dev/wdevs/pgsql-broker:latest
+    depends_on:
+      migrate:
+        condition: service_completed_successfully
+    volumes:
+      - ./broker.docker.yaml:/etc/pgsql-broker/broker.yaml:ro
+    restart: unless-stopped
+
+volumes:
+  postgres-data:

go.mod

@@ -4,6 +4,7 @@ go 1.26.0
 
 require (
 	github.com/lib/pq v1.10.9
+	github.com/prometheus/client_golang v1.20.5
 	github.com/spf13/cobra v1.10.2
 	github.com/spf13/viper v1.21.0
 	github.com/stretchr/testify v1.11.1
@@ -11,12 +12,19 @@ require (
 )
 
 require (
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/fsnotify/fsnotify v1.9.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.4.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/klauspost/compress v1.17.9 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_model v0.6.1 // indirect
+	github.com/prometheus/common v0.55.0 // indirect
+	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/sagikazarmark/locafero v0.11.0 // indirect
 	github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect
 	github.com/spf13/afero v1.15.0 // indirect
@@ -26,5 +34,6 @@ require (
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/sys v0.48.0 // indirect
 	golang.org/x/text v0.28.0 // indirect
+	google.golang.org/protobuf v1.34.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,3 +1,7 @@
+github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
+github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -11,18 +15,32 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
+github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
+github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
+github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
 github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
+github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4=
 github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8=
+github.com/prometheus/client_golang v1.20.5 h1:cxppBPuYhUnsO6yo/aoRol4L7q7UFfdm+bR9r+8l63Y=
-github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
+github.com/prometheus/client_golang v1.20.5/go.mod h1:PIEt8X02hGcP8JWbeHyeZ53Y/jReSnHgO035n//V5WE=
+github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
+github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
+github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G1dc=
+github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8=
+github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
+github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
+github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/sagikazarmark/locafero v0.11.0 h1:1iurJgmM9G3PA/I+wWYIOw/5SyBtxapeHDcg+AAIFXc=
 github.com/sagikazarmark/locafero v0.11.0/go.mod h1:nVIGvgyzw595SUSUE6tvCp3YYTeHs15MvlmU87WwIik=
@@ -51,8 +69,10 @@ golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE=
 golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc=
 golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng=
 golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU=
+google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
+google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
-gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

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

@@ -4,9 +4,11 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/config"
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/metrics"
 )
 
 // Broker manages multiple database instances
@@ -17,6 +19,8 @@ type Broker struct {
 	instances []*DatabaseInstance
 	ctx       context.Context
 	cancel    context.CancelFunc
+	metrics   *metrics.Metrics
+	server    *metrics.Server
 	shutdown  bool
 	mu        sync.RWMutex
 }
@@ -33,6 +37,16 @@ func New(cfg *config.Config, logger adapter.Logger, version string) (*Broker, er
 		ctx:       ctx,
 		cancel:    cancel,
 	}
+	if cfg.Broker.MetricsEnabled {
+		broker.metrics = metrics.New()
+		broker.metrics.SetDatabaseCount(len(cfg.Databases))
+		addr := fmt.Sprintf("%s:%d", cfg.Broker.MetricsHost, cfg.Broker.MetricsPort)
+		server, err := metrics.NewServer(broker.metrics, addr, broker.logger)
+		if err != nil {
+			return nil, err
+		}
+		broker.server = server
+	}
 
 	return broker, nil
 }
@@ -42,14 +56,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, b.metrics)
 		if err != nil {
 			// Stop any already-started instances
 			b.stopInstances()
@@ -68,6 +83,9 @@ func (b *Broker) Start() error {
 	}
 
 	b.logger.Info("broker started successfully", "database_instances", len(b.instances))
+	if b.server != nil {
+		b.server.Start()
+	}
 	return nil
 }
 
@@ -88,6 +106,13 @@ func (b *Broker) Stop() error {
 
 	// Stop all instances
 	b.stopInstances()
+	if b.server != nil {
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		if err := b.server.Stop(ctx); err != nil {
+			b.logger.Error("failed to stop metrics server", "error", err)
+		}
+	}
 
 	b.logger.Info("broker stopped")
 	return nil

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,13 +48,23 @@ 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.
 	LeaseSeconds int `mapstructure:"lease_seconds"`
 	// StaleJobRecoverySec is the interval between broker_recover_stale_jobs sweeps.
 	StaleJobRecoverySec int `mapstructure:"stale_job_recovery_sec"`
+	// MetricsEnabled controls whether the embedded Prometheus metrics HTTP
+	// server (exposition endpoint + HTML dashboard) is started.
+	MetricsEnabled bool `mapstructure:"metrics_enabled"`
+	// MetricsHost is the bind address for the metrics server.
+	MetricsHost string `mapstructure:"metrics_host"`
+	// MetricsPort is the bind port for the metrics server.
+	MetricsPort int `mapstructure:"metrics_port"`
+	// QueueDepthPollSec is the interval between polls of pending job counts
+	// used to populate the broker_jobs_queued gauge.
+	QueueDepthPollSec int `mapstructure:"queue_depth_poll_sec"`
 }
 
 // LoggingConfig holds logging settings
@@ -119,6 +130,10 @@ func setDefaults(v *viper.Viper) {
 	v.SetDefault("broker.enable_debug", false)
 	v.SetDefault("broker.lease_seconds", 60)
 	v.SetDefault("broker.stale_job_recovery_sec", 30)
+	v.SetDefault("broker.metrics_enabled", true)
+	v.SetDefault("broker.metrics_host", "0.0.0.0")
+	v.SetDefault("broker.metrics_port", 9469)
+	v.SetDefault("broker.queue_depth_poll_sec", 15)
 
 	// Logging defaults
 	v.SetDefault("logging.level", "info")
@@ -132,7 +147,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 +211,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/database_instance.go

@@ -13,6 +13,7 @@ import (
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/config"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/install"
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/metrics"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/models"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/queue"
 )
@@ -36,6 +37,7 @@ type DatabaseInstance struct {
 	shutdownMu   sync.RWMutex
 	jobsHandled  int64
 	startTime    time.Time
+	metrics      *metrics.Metrics
 
 	// sessionConn holds the pg_try_advisory_lock acquired by
 	// registerInstance. The lock is scoped to this one physical connection,
@@ -45,12 +47,16 @@ type DatabaseInstance struct {
 }
 
 // NewDatabaseInstance creates a new database instance
-func NewDatabaseInstance(cfg *config.Config, dbCfg *config.DatabaseConfig, db adapter.DBAdapter, logger adapter.Logger, version string, parentCtx context.Context) (*DatabaseInstance, error) {
+func NewDatabaseInstance(cfg *config.Config, dbCfg *config.DatabaseConfig, db adapter.DBAdapter, logger adapter.Logger, version string, parentCtx context.Context, brokerMetrics ...*metrics.Metrics) (*DatabaseInstance, error) {
 	hostname, err := os.Hostname()
 	if err != nil {
 		hostname = "unknown"
 	}
 
+	var instanceMetrics *metrics.Metrics
+	if len(brokerMetrics) > 0 {
+		instanceMetrics = brokerMetrics[0]
+	}
 	instance := &DatabaseInstance{
 		Name:         fmt.Sprintf("%s-%s", cfg.Broker.Name, dbCfg.Name),
 		DatabaseName: dbCfg.Name,
@@ -64,6 +70,7 @@ func NewDatabaseInstance(cfg *config.Config, dbCfg *config.DatabaseConfig, db ad
 		queues:       make(map[int]*queue.Queue),
 		ctx:          parentCtx,
 		startTime:    time.Now(),
+		metrics:      instanceMetrics,
 	}
 
 	return instance, nil
@@ -109,9 +116,54 @@ func (i *DatabaseInstance) Start() error {
 	adapter.SupervisedGo(i.logger, "stale-job-recovery-routine", i.staleJobRecoveryRoutine)
 
 	i.logger.Info("database instance started successfully")
+	if i.metrics != nil {
+		adapter.SupervisedGo(i.logger, "metrics-queue-depth-routine", i.queueDepthRoutine)
+	}
 	return nil
 }
 
+// queueDepthRoutine periodically exports pending jobs grouped by queue.
+func (i *DatabaseInstance) queueDepthRoutine() {
+	interval := time.Duration(i.config.Broker.QueueDepthPollSec) * time.Second
+	if interval <= 0 {
+		interval = 15 * time.Second
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	i.updateQueueDepthMetrics()
+	for {
+		select {
+		case <-ticker.C:
+			i.updateQueueDepthMetrics()
+		case <-i.ctx.Done():
+			return
+		}
+	}
+}
+
+func (i *DatabaseInstance) updateQueueDepthMetrics() {
+	rows, err := i.db.Query(i.ctx, "SELECT job_queue, COUNT(*) FROM broker.broker_jobs WHERE complete_status = 0 GROUP BY job_queue")
+	if err != nil {
+		i.logger.Warn("failed to collect queue depth metrics", "error", err)
+		return
+	}
+	defer rows.Close()
+	for queueNumber := 1; queueNumber <= i.dbConfig.QueueCount; queueNumber++ {
+		i.metrics.SetJobsQueued(i.DatabaseName, queueNumber, 0)
+	}
+	for rows.Next() {
+		var queueNumber, count int
+		if err := rows.Scan(&queueNumber, &count); err != nil {
+			i.logger.Warn("failed to scan queue depth metric", "error", err)
+			return
+		}
+		i.metrics.SetJobsQueued(i.DatabaseName, queueNumber, float64(count))
+	}
+	if err := rows.Err(); err != nil {
+		i.logger.Warn("failed to read queue depth metrics", "error", err)
+	}
+}
+
 // ensureSchema checks the embedded migration set against the database and,
 // depending on dbConfig.AutoMigrate, either applies pending migrations or
 // fails startup fast rather than running against a stale/missing schema.
@@ -261,6 +313,8 @@ func (i *DatabaseInstance) startQueues() error {
 			FetchSize:    i.config.Broker.FetchQueryQueSize,
 			TenantID:     i.dbConfig.TenantID,
 			LeaseSeconds: leaseSeconds,
+			Metrics:      i.metrics,
+			DatabaseName: i.DatabaseName,
 		}
 
 		q := queue.New(queueCfg)
@@ -271,6 +325,7 @@ func (i *DatabaseInstance) startQueues() error {
 		i.queues[queueNum] = q
 		i.logger.Info("queue started", "number", queueNum)
 	}
+	i.metrics.SetQueueCount(i.DatabaseName, len(i.queues))
 
 	return nil
 }

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/metrics/metrics.go

@@ -0,0 +1,130 @@
+// Package metrics defines the broker's Prometheus collectors and a small
+// embedded HTTP server that exposes them, both as the standard /metrics
+// exposition endpoint and as a single-page HTML dashboard.
+package metrics
+
+import (
+	"strconv"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// Metrics holds every Prometheus collector the broker exposes. It is safe
+// for concurrent use, and a nil *Metrics is safe to call methods on (all
+// recording methods become no-ops), so callers that construct a Broker
+// without metrics enabled don't need to special-case it.
+type Metrics struct {
+	Registry *prometheus.Registry
+
+	jobsCompleted *prometheus.CounterVec
+	jobsFailed    *prometheus.CounterVec
+	jobsRequeued  *prometheus.CounterVec
+	jobDuration   *prometheus.HistogramVec
+	jobsQueued    *prometheus.GaugeVec
+	databaseCount prometheus.Gauge
+	queueCount    *prometheus.GaugeVec
+}
+
+// New creates a Metrics instance with a fresh (non-global) registry, so
+// multiple brokers can coexist in the same process -- e.g. in tests --
+// without colliding on prometheus.DefaultRegisterer.
+func New() *Metrics {
+	registry := prometheus.NewRegistry()
+
+	m := &Metrics{
+		Registry: registry,
+		jobsCompleted: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "broker_jobs_completed_total",
+			Help: "Total number of jobs that completed successfully.",
+		}, []string{"database", "job_group", "job_name"}),
+		jobsFailed: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "broker_jobs_failed_total",
+			Help: "Total number of jobs that were dead-lettered after exhausting retries.",
+		}, []string{"database", "job_group", "job_name"}),
+		jobsRequeued: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "broker_jobs_requeued_total",
+			Help: "Total number of job attempts that failed and were requeued for retry.",
+		}, []string{"database", "job_group", "job_name"}),
+		jobDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "broker_job_duration_seconds",
+			Help:    "Job execution duration in seconds, by group and name.",
+			Buckets: prometheus.DefBuckets,
+		}, []string{"database", "job_group", "job_name"}),
+		jobsQueued: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "broker_jobs_queued",
+			Help: "Current number of pending (not yet claimed) jobs, by database and queue.",
+		}, []string{"database", "queue"}),
+		databaseCount: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "broker_databases",
+			Help: "Number of database instances managed by this broker process.",
+		}),
+		queueCount: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "broker_queues",
+			Help: "Number of queues configured for a database instance.",
+		}, []string{"database"}),
+	}
+
+	registry.MustRegister(
+		m.jobsCompleted,
+		m.jobsFailed,
+		m.jobsRequeued,
+		m.jobDuration,
+		m.jobsQueued,
+		m.databaseCount,
+		m.queueCount,
+	)
+
+	return m
+}
+
+// RecordJobCompleted records a successfully completed job attempt.
+func (m *Metrics) RecordJobCompleted(database, group, name string, duration time.Duration) {
+	if m == nil {
+		return
+	}
+	m.jobsCompleted.WithLabelValues(database, group, name).Inc()
+	m.jobDuration.WithLabelValues(database, group, name).Observe(duration.Seconds())
+}
+
+// RecordJobFailed records a job attempt that was dead-lettered (attempts exhausted).
+func (m *Metrics) RecordJobFailed(database, group, name string, duration time.Duration) {
+	if m == nil {
+		return
+	}
+	m.jobsFailed.WithLabelValues(database, group, name).Inc()
+	m.jobDuration.WithLabelValues(database, group, name).Observe(duration.Seconds())
+}
+
+// RecordJobRequeued records a job attempt that failed but was requeued for retry.
+func (m *Metrics) RecordJobRequeued(database, group, name string, duration time.Duration) {
+	if m == nil {
+		return
+	}
+	m.jobsRequeued.WithLabelValues(database, group, name).Inc()
+	m.jobDuration.WithLabelValues(database, group, name).Observe(duration.Seconds())
+}
+
+// SetJobsQueued sets the current pending job count for a database/queue pair.
+func (m *Metrics) SetJobsQueued(database string, queue int, count float64) {
+	if m == nil {
+		return
+	}
+	m.jobsQueued.WithLabelValues(database, strconv.Itoa(queue)).Set(count)
+}
+
+// SetDatabaseCount sets the number of database instances managed by this process.
+func (m *Metrics) SetDatabaseCount(n int) {
+	if m == nil {
+		return
+	}
+	m.databaseCount.Set(float64(n))
+}
+
+// SetQueueCount sets the number of queues configured for a database instance.
+func (m *Metrics) SetQueueCount(database string, n int) {
+	if m == nil {
+		return
+	}
+	m.queueCount.WithLabelValues(database).Set(float64(n))
+}

pkg/broker/metrics/page.html

@@ -0,0 +1,215 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+<meta charset="utf-8">
+<meta name="viewport" content="width=device-width, initial-scale=1">
+<title>pgsql-broker metrics</title>
+<style>
+  :root {
+    color-scheme: dark;
+    --bg: #0f1115;
+    --panel: #161a22;
+    --border: #262b36;
+    --text: #e6e9ef;
+    --muted: #8b93a7;
+    --accent: #5aa8ff;
+    --good: #4caf7d;
+    --bad: #e5636b;
+  }
+  * { box-sizing: border-box; }
+  body {
+    margin: 0;
+    padding: 2rem;
+    background: var(--bg);
+    color: var(--text);
+    font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace;
+  }
+  h1 {
+    font-size: 1.25rem;
+    margin: 0 0 0.25rem;
+  }
+  #status {
+    color: var(--muted);
+    font-size: 0.8rem;
+    margin-bottom: 1.5rem;
+  }
+  .grid {
+    display: grid;
+    grid-template-columns: repeat(auto-fill, minmax(360px, 1fr));
+    gap: 1rem;
+  }
+  .card {
+    background: var(--panel);
+    border: 1px solid var(--border);
+    border-radius: 8px;
+    padding: 1rem;
+  }
+  .card h2 {
+    font-size: 0.9rem;
+    margin: 0 0 0.75rem;
+    color: var(--accent);
+    font-weight: 600;
+  }
+  .card .help {
+    color: var(--muted);
+    font-size: 0.75rem;
+    margin: -0.5rem 0 0.75rem;
+  }
+  table {
+    width: 100%;
+    border-collapse: collapse;
+    font-size: 0.8rem;
+  }
+  td, th {
+    text-align: left;
+    padding: 0.2rem 0.4rem 0.2rem 0;
+    border-bottom: 1px solid var(--border);
+  }
+  td.value {
+    text-align: right;
+    font-variant-numeric: tabular-nums;
+    white-space: nowrap;
+  }
+  td.labels {
+    color: var(--muted);
+  }
+  .completed .value { color: var(--good); }
+  .failed .value { color: var(--bad); }
+  .empty {
+    color: var(--muted);
+    font-size: 0.8rem;
+  }
+</style>
+</head>
+<body>
+  <h1>pgsql-broker metrics</h1>
+  <div id="status">loading&hellip;</div>
+  <div class="grid" id="grid"></div>
+
+<script>
+(function () {
+  "use strict";
+
+  // Metric name -> { help, className } for the cards we render, in order.
+  var METRICS = [
+    { name: "broker_jobs_completed_total", title: "Jobs completed", cls: "completed" },
+    { name: "broker_jobs_failed_total", title: "Jobs failed (dead-lettered)", cls: "failed" },
+    { name: "broker_jobs_requeued_total", title: "Jobs requeued (retries)", cls: "" },
+    { name: "broker_jobs_queued", title: "Jobs currently queued", cls: "" },
+    { name: "broker_job_duration_seconds_sum", title: "Job duration, total seconds by group/name", cls: "" },
+    { name: "broker_job_duration_seconds_count", title: "Job duration, sample count by group/name", cls: "" },
+    { name: "broker_databases", title: "Databases managed", cls: "" },
+    { name: "broker_queues", title: "Queues per database", cls: "" }
+  ];
+
+  // Parses Prometheus text exposition format into { name: [{labels, value}] }.
+  function parseMetrics(text) {
+    var byName = {};
+    var lines = text.split("\n");
+    for (var i = 0; i < lines.length; i++) {
+      var line = lines[i];
+      if (!line || line[0] === "#") continue;
+
+      var name, labels = {}, rest;
+      var braceIdx = line.indexOf("{");
+      var spaceIdx;
+      if (braceIdx !== -1) {
+        name = line.slice(0, braceIdx);
+        var closeIdx = line.indexOf("}", braceIdx);
+        if (closeIdx === -1) continue;
+        var labelStr = line.slice(braceIdx + 1, closeIdx);
+        var labelRe = /(\w+)="((?:[^"\\]|\\.)*)"/g;
+        var m;
+        while ((m = labelRe.exec(labelStr)) !== null) {
+          labels[m[1]] = m[2].replace(/\\"/g, '"').replace(/\\\\/g, "\\");
+        }
+        rest = line.slice(closeIdx + 1).trim();
+      } else {
+        spaceIdx = line.indexOf(" ");
+        if (spaceIdx === -1) continue;
+        name = line.slice(0, spaceIdx);
+        rest = line.slice(spaceIdx + 1).trim();
+      }
+
+      var value = parseFloat(rest.split(" ")[0]);
+      if (isNaN(value)) continue;
+
+      if (!byName[name]) byName[name] = [];
+      byName[name].push({ labels: labels, value: value });
+    }
+    return byName;
+  }
+
+  function formatLabels(labels) {
+    var keys = Object.keys(labels).sort();
+    return keys.map(function (k) { return k + "=" + labels[k]; }).join(", ");
+  }
+
+  function formatValue(v) {
+    if (Number.isInteger(v)) return String(v);
+    return v.toFixed(3);
+  }
+
+  function render(byName) {
+    var grid = document.getElementById("grid");
+    grid.innerHTML = "";
+
+    METRICS.forEach(function (spec) {
+      var series = byName[spec.name] || [];
+      var card = document.createElement("div");
+      card.className = "card " + spec.cls;
+
+      var h2 = document.createElement("h2");
+      h2.textContent = spec.title;
+      card.appendChild(h2);
+
+      if (series.length === 0) {
+        var empty = document.createElement("div");
+        empty.className = "empty";
+        empty.textContent = "no data yet";
+        card.appendChild(empty);
+      } else {
+        var table = document.createElement("table");
+        series.sort(function (a, b) { return b.value - a.value; });
+        series.forEach(function (s) {
+          var tr = document.createElement("tr");
+          var tdLabels = document.createElement("td");
+          tdLabels.className = "labels";
+          tdLabels.textContent = formatLabels(s.labels) || "(none)";
+          var tdValue = document.createElement("td");
+          tdValue.className = "value";
+          tdValue.textContent = formatValue(s.value);
+          tr.appendChild(tdLabels);
+          tr.appendChild(tdValue);
+          table.appendChild(tr);
+        });
+        card.appendChild(table);
+      }
+
+      grid.appendChild(card);
+    });
+  }
+
+  function refresh() {
+    fetch("metrics", { cache: "no-store" })
+      .then(function (resp) {
+        if (!resp.ok) throw new Error("HTTP " + resp.status);
+        return resp.text();
+      })
+      .then(function (text) {
+        render(parseMetrics(text));
+        document.getElementById("status").textContent =
+          "last updated " + new Date().toLocaleTimeString();
+      })
+      .catch(function (err) {
+        document.getElementById("status").textContent =
+          "failed to load metrics: " + err.message;
+      });
+  }
+
+  refresh();
+  setInterval(refresh, 5000);
+})();
+</script>
+</body>
+</html>

pkg/broker/metrics/server.go

@@ -0,0 +1,74 @@
+package metrics
+
+import (
+	"context"
+	_ "embed"
+	"fmt"
+	"net"
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
+)
+
+//go:embed page.html
+var dashboardHTML []byte
+
+// Server is the embedded HTTP server exposing /metrics (standard Prometheus
+// exposition format) and / (a single-page HTML dashboard that polls
+// /metrics).
+type Server struct {
+	httpServer *http.Server
+	listener   net.Listener
+	logger     adapter.Logger
+}
+
+// NewServer builds a Server bound to addr (e.g. "127.0.0.1:9469"). Binding
+// happens immediately so a port conflict is reported to the caller rather
+// than surfacing later in a background goroutine.
+func NewServer(m *Metrics, addr string, logger adapter.Logger) (*Server, error) {
+	listener, err := net.Listen("tcp", addr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to bind metrics server to %s: %w", addr, err)
+	}
+
+	mux := http.NewServeMux()
+	mux.Handle("/metrics", promhttp.HandlerFor(m.Registry, promhttp.HandlerOpts{}))
+	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/" {
+			http.NotFound(w, r)
+			return
+		}
+		w.Header().Set("Content-Type", "text/html; charset=utf-8")
+		_, _ = w.Write(dashboardHTML)
+	})
+
+	return &Server{
+		httpServer: &http.Server{Handler: mux},
+		listener:   listener,
+		logger:     logger.With("component", "metrics-server"),
+	}, nil
+}
+
+// Addr returns the actual bound address (useful when addr was given with a
+// ":0" port).
+func (s *Server) Addr() string {
+	return s.listener.Addr().String()
+}
+
+// Start serves in the background. It returns immediately; Serve errors
+// (other than a clean Shutdown) are logged.
+func (s *Server) Start() {
+	s.logger.Info("metrics server listening", "addr", s.Addr())
+	go func() {
+		if err := s.httpServer.Serve(s.listener); err != nil && err != http.ErrServerClosed {
+			s.logger.Error("metrics server stopped unexpectedly", "error", err)
+		}
+	}()
+}
+
+// Stop gracefully shuts down the server.
+func (s *Server) Stop(ctx context.Context) error {
+	return s.httpServer.Shutdown(ctx)
+}

pkg/broker/queue/queue.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/metrics"
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/worker"
 )
 
@@ -34,6 +35,8 @@ type Config struct {
 	FetchSize    int
 	TenantID     string
 	LeaseSeconds int
+	Metrics      *metrics.Metrics
+	DatabaseName string
 }
 
 // New creates a new queue manager
@@ -73,6 +76,8 @@ func (q *Queue) Start(cfg Config) error {
 			FetchSize:    cfg.FetchSize,
 			TenantID:     cfg.TenantID,
 			LeaseSeconds: cfg.LeaseSeconds,
+			Metrics:      cfg.Metrics,
+			DatabaseName: cfg.DatabaseName,
 		})
 
 		if err := w.Start(q.ctx); err != nil {

pkg/broker/worker/worker.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/adapter"
+	"git.warky.dev/wdevs/pgsql-broker/pkg/broker/metrics"
 )
 
 // Worker represents a single job processing worker
@@ -29,6 +30,8 @@ type Worker struct {
 	fetchSize    int
 	tenantID     string
 	leaseSeconds int
+	metrics      *metrics.Metrics
+	databaseName string
 }
 
 // Stats holds worker statistics
@@ -50,6 +53,8 @@ type Config struct {
 	FetchSize    int
 	TenantID     string
 	LeaseSeconds int
+	Metrics      *metrics.Metrics
+	DatabaseName string
 }
 
 // New creates a new worker
@@ -72,6 +77,8 @@ func New(cfg Config) *Worker {
 		fetchSize:    cfg.FetchSize,
 		tenantID:     cfg.TenantID,
 		leaseSeconds: leaseSeconds,
+		metrics:      cfg.Metrics,
+		databaseName: cfg.DatabaseName,
 	}
 }
 
@@ -204,26 +211,47 @@ 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
+			// No job found, rollback
-			return        // No more jobs
+			if rbErr := tx.Rollback(); rbErr != nil {
+				w.logger.Error("failed to rollback transaction", "error", rbErr)
+			}
+			return // No more jobs
+		}
+
+		jobName, jobGroup, err := w.fetchJobLabelsTx(ctx, tx, jobID)
+		if err != nil {
+			w.logger.Warn("failed to fetch job labels for metrics", "job_id", jobID, "error", err)
 		}
 
 		// Run the job
-		if err := w.runJobTx(ctx, tx, jobID, leaseToken); err != nil {
+		start := time.Now()
-			tx.Rollback() // Rollback on genuine infra failure
+		jobStatus, err := w.runJobTx(ctx, tx, jobID, leaseToken)
+		duration := time.Since(start)
+		if err == nil {
+			w.recordJobMetric(jobStatus, jobGroup, jobName, duration)
+		}
+		if err != nil {
+			// 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 +275,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)
@@ -277,7 +305,7 @@ func (w *Worker) fetchNextJobTx(ctx context.Context, tx adapter.DBTransaction) (
 // error (triggering a rollback of the claim) on a genuine infra failure --
 // job outcomes reported via p_job_status (requeued/completed/dead-lettered)
 // are always committed.
-func (w *Worker) runJobTx(ctx context.Context, tx adapter.DBTransaction, jobID int64, leaseToken string) error {
+func (w *Worker) runJobTx(ctx context.Context, tx adapter.DBTransaction, jobID int64, leaseToken string) (int, error) {
 	w.logger.Debug("running job", "job_id", jobID)
 
 	var retval int
@@ -290,15 +318,43 @@ func (w *Worker) runJobTx(ctx context.Context, tx adapter.DBTransaction, jobID i
 	).Scan(&retval, &errmsg, &jobStatus)
 
 	if err != nil {
-		return fmt.Errorf("query error: %w", err)
+		return 0, fmt.Errorf("query error: %w", err)
 	}
 
 	if retval > 0 {
-		return fmt.Errorf("broker_run error: %s", errmsg)
+		return 0, fmt.Errorf("broker_run error: %s", errmsg)
 	}
 
 	w.logger.Debug("job finished", "job_id", jobID, "job_status", jobStatus)
-	return nil
+	return jobStatus, nil
+}
+
+// fetchJobLabelsTx looks up the job_name/job_group of jobID for metric
+// labeling. Best-effort: callers log and continue on error rather than
+// failing the job over a metrics lookup.
+func (w *Worker) fetchJobLabelsTx(ctx context.Context, tx adapter.DBTransaction, jobID int64) (jobName, jobGroup string, err error) {
+	err = tx.QueryRow(ctx,
+		"SELECT job_name, job_group FROM broker.broker_jobs WHERE id_broker_jobs = $1",
+		jobID,
+	).Scan(&jobName, &jobGroup)
+	if err != nil {
+		return "", "", fmt.Errorf("query error: %w", err)
+	}
+	return jobName, jobGroup, nil
+}
+
+// recordJobMetric routes a finished job attempt to the appropriate
+// Prometheus counter/histogram based on the p_job_status broker_run
+// reported (0=requeued, 2=completed, 3=dead-lettered).
+func (w *Worker) recordJobMetric(jobStatus int, jobGroup, jobName string, duration time.Duration) {
+	switch jobStatus {
+	case 2:
+		w.metrics.RecordJobCompleted(w.databaseName, jobGroup, jobName, duration)
+	case 3:
+		w.metrics.RecordJobFailed(w.databaseName, jobGroup, jobName, duration)
+	case 0:
+		w.metrics.RecordJobRequeued(w.databaseName, jobGroup, jobName, duration)
+	}
 }
 
 // updateActivity updates the last activity timestamp

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{