feat(queue): replace PocketBase polling with Asynq + Redis

Introduce a Redis-backed Asynq task queue so the runner consumes TTS
jobs pushed by the backend instead of polling PocketBase.

- backend/internal/asynqqueue: Producer and Consumer wrappers
- backend/internal/runner: AsynqRunner mux, per-instance Prometheus
  registry (fixes duplicate-collector panic in tests), redisConnOpt
- backend/internal/config: REDIS_ADDR / REDIS_PASSWORD env vars
- backend/cmd/{backend,runner}/main.go: wire Redis when env set; fall
  back to legacy poll mode when unset
- Caddyfile: caddy-l4 TCP proxy for redis.libnovel.cc:6380 → homelab
- caddy/Dockerfile: add --with github.com/mholt/caddy-l4
- docker-compose.yml: Caddy exposes 6380, backend/runner get Redis env
- homelab/runner/docker-compose.yml: Redis sidecar, runner depends_on
- homelab/otel/grafana: Grafana dashboards (backend, catalogue, runner)
  and alerting rules / contact-points provisioning

backend/internal/runner/asynq_runner.go

@@ -0,0 +1,149 @@
+package runner
+
+// asynq_runner.go — Asynq-based task dispatch for the runner.
+//
+// When cfg.RedisAddr is set, Run() calls runAsynq() instead of runPoll().
+// The Asynq server replaces the polling loop: it listens on Redis for tasks
+// enqueued by the backend Producer and delivers them immediately.
+//
+// Handlers in this file decode Asynq job payloads and call the existing
+// runScrapeTask / runAudioTask methods, keeping all execution logic in one place.
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"github.com/hibiken/asynq"
+	asynqmetrics "github.com/hibiken/asynq/x/metrics"
+	"github.com/libnovel/backend/internal/asynqqueue"
+	"github.com/libnovel/backend/internal/domain"
+)
+
+// runAsynq starts an Asynq server that replaces the PocketBase poll loop.
+// It also starts the periodic catalogue refresh ticker.
+// Blocks until ctx is cancelled.
+func (r *Runner) runAsynq(ctx context.Context) error {
+	redisOpt, err := r.redisConnOpt()
+	if err != nil {
+		return fmt.Errorf("runner: parse redis addr: %w", err)
+	}
+
+	srv := asynq.NewServer(redisOpt, asynq.Config{
+		// Allocate concurrency slots for each task type.
+		// Total concurrency = scrape + audio slots.
+		Concurrency: r.cfg.MaxConcurrentScrape + r.cfg.MaxConcurrentAudio,
+		Queues: map[string]int{
+			asynqqueue.QueueDefault: 1,
+		},
+		// Let Asynq handle retries with exponential back-off.
+		RetryDelayFunc: asynq.DefaultRetryDelayFunc,
+		// Log errors from handlers via the existing structured logger.
+		ErrorHandler: asynq.ErrorHandlerFunc(func(_ context.Context, task *asynq.Task, err error) {
+			r.deps.Log.Error("runner: asynq task failed",
+				"type", task.Type(),
+				"err", err,
+			)
+		}),
+	})
+
+	mux := asynq.NewServeMux()
+	mux.HandleFunc(asynqqueue.TypeAudioGenerate, r.handleAudioTask)
+	mux.HandleFunc(asynqqueue.TypeScrapeBook, r.handleScrapeTask)
+	mux.HandleFunc(asynqqueue.TypeScrapeCatalogue, r.handleScrapeTask)
+
+	// Register Asynq queue metrics with the default Prometheus registry so
+	// the /metrics endpoint (metrics.go) can expose them.
+	inspector := asynq.NewInspector(redisOpt)
+	collector := asynqmetrics.NewQueueMetricsCollector(inspector)
+	if err := r.metricsRegistry.Register(collector); err != nil {
+		r.deps.Log.Warn("runner: could not register asynq prometheus collector", "err", err)
+	}
+
+	// Start the periodic catalogue refresh.
+	catalogueTick := time.NewTicker(r.cfg.CatalogueRefreshInterval)
+	defer catalogueTick.Stop()
+	if !r.cfg.SkipInitialCatalogueRefresh {
+		go r.runCatalogueRefresh(ctx)
+	} else {
+		r.deps.Log.Info("runner: skipping initial catalogue refresh (RUNNER_SKIP_INITIAL_CATALOGUE_REFRESH=true)")
+	}
+
+	r.deps.Log.Info("runner: asynq mode active", "redis_addr", r.cfg.RedisAddr)
+
+	// Run catalogue refresh ticker in the background.
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-catalogueTick.C:
+				go r.runCatalogueRefresh(ctx)
+			}
+		}
+	}()
+
+	// Start Asynq server (non-blocking).
+	if err := srv.Start(mux); err != nil {
+		return fmt.Errorf("runner: asynq server start: %w", err)
+	}
+
+	// Block until context is cancelled, then gracefully stop.
+	<-ctx.Done()
+	r.deps.Log.Info("runner: context cancelled, shutting down asynq server")
+	srv.Shutdown()
+	return nil
+}
+
+// redisConnOpt parses cfg.RedisAddr into an asynq.RedisConnOpt.
+// Supports full "redis://" / "rediss://" URLs and plain "host:port".
+func (r *Runner) redisConnOpt() (asynq.RedisConnOpt, error) {
+	addr := r.cfg.RedisAddr
+	// ParseRedisURI handles redis:// and rediss:// schemes.
+	if len(addr) > 7 && (addr[:8] == "redis://" || addr[:9] == "rediss://") {
+		return asynq.ParseRedisURI(addr)
+	}
+	// Plain "host:port" — use RedisClientOpt directly.
+	return asynq.RedisClientOpt{
+		Addr:     addr,
+		Password: r.cfg.RedisPassword,
+	}, nil
+}
+
+// handleScrapeTask is the Asynq handler for TypeScrapeBook and TypeScrapeCatalogue.
+func (r *Runner) handleScrapeTask(ctx context.Context, t *asynq.Task) error {
+	var p asynqqueue.ScrapePayload
+	if err := json.Unmarshal(t.Payload(), &p); err != nil {
+		return fmt.Errorf("unmarshal scrape payload: %w", err)
+	}
+	task := domain.ScrapeTask{
+		ID:          p.PBTaskID,
+		Kind:        p.Kind,
+		TargetURL:   p.TargetURL,
+		FromChapter: p.FromChapter,
+		ToChapter:   p.ToChapter,
+	}
+	r.tasksRunning.Add(1)
+	defer r.tasksRunning.Add(-1)
+	r.runScrapeTask(ctx, task)
+	return nil
+}
+
+// handleAudioTask is the Asynq handler for TypeAudioGenerate.
+func (r *Runner) handleAudioTask(ctx context.Context, t *asynq.Task) error {
+	var p asynqqueue.AudioPayload
+	if err := json.Unmarshal(t.Payload(), &p); err != nil {
+		return fmt.Errorf("unmarshal audio payload: %w", err)
+	}
+	task := domain.AudioTask{
+		ID:      p.PBTaskID,
+		Slug:    p.Slug,
+		Chapter: p.Chapter,
+		Voice:   p.Voice,
+	}
+	r.tasksRunning.Add(1)
+	defer r.tasksRunning.Add(-1)
+	r.runAudioTask(ctx, task)
+	return nil
+}

backend/internal/runner/metrics.go

@@ -1,21 +1,28 @@
 package runner
 
-// metrics.go — lightweight HTTP metrics endpoint for the runner.
+// metrics.go — Prometheus metrics HTTP endpoint for the runner.
 //
-// GET /metrics returns a JSON document with live task counters and uptime.
-// No external dependency (no Prometheus); plain net/http only.
+// GET /metrics returns a Prometheus text/plain scrape response.
+// Exposes:
+//   - Standard Go runtime metrics (via promhttp)
+//   - Runner task counters (tasks_running, tasks_completed, tasks_failed)
+//   - Asynq queue metrics (registered in asynq_runner.go when Redis is enabled)
+//
+// GET /health — simple liveness probe.
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"log/slog"
 	"net"
 	"net/http"
 	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
-// metricsServer serves GET /metrics for the runner process.
+// metricsServer serves GET /metrics and GET /health for the runner process.
 type metricsServer struct {
 	addr string
 	r    *Runner
@@ -23,21 +30,62 @@ type metricsServer struct {
 }
 
 func newMetricsServer(addr string, r *Runner, log *slog.Logger) *metricsServer {
-	return &metricsServer{addr: addr, r: r, log: log}
+	ms := &metricsServer{addr: addr, r: r, log: log}
+	ms.registerCollectors()
+	return ms
+}
+
+// registerCollectors registers runner-specific Prometheus collectors.
+// Called once at construction; Asynq queue collector is registered separately
+// in asynq_runner.go after the Redis connection is established.
+func (ms *metricsServer) registerCollectors() {
+	// Runner task gauges / counters backed by the atomic fields on Runner.
+	ms.r.metricsRegistry.MustRegister(prometheus.NewGaugeFunc(
+		prometheus.GaugeOpts{
+			Namespace: "runner",
+			Name:      "tasks_running",
+			Help:      "Number of tasks currently being processed.",
+		},
+		func() float64 { return float64(ms.r.tasksRunning.Load()) },
+	))
+	ms.r.metricsRegistry.MustRegister(prometheus.NewCounterFunc(
+		prometheus.CounterOpts{
+			Namespace: "runner",
+			Name:      "tasks_completed_total",
+			Help:      "Total number of tasks completed successfully since startup.",
+		},
+		func() float64 { return float64(ms.r.tasksCompleted.Load()) },
+	))
+	ms.r.metricsRegistry.MustRegister(prometheus.NewCounterFunc(
+		prometheus.CounterOpts{
+			Namespace: "runner",
+			Name:      "tasks_failed_total",
+			Help:      "Total number of tasks that ended in failure since startup.",
+		},
+		func() float64 { return float64(ms.r.tasksFailed.Load()) },
+	))
+	ms.r.metricsRegistry.MustRegister(prometheus.NewGaugeFunc(
+		prometheus.GaugeOpts{
+			Namespace: "runner",
+			Name:      "uptime_seconds",
+			Help:      "Seconds since the runner process started.",
+		},
+		func() float64 { return time.Since(ms.r.startedAt).Seconds() },
+	))
 }
 
 // ListenAndServe starts the HTTP server and blocks until ctx is cancelled or
 // a fatal listen error occurs.
 func (ms *metricsServer) ListenAndServe(ctx context.Context) error {
 	mux := http.NewServeMux()
-	mux.HandleFunc("GET /metrics", ms.handleMetrics)
+	mux.Handle("GET /metrics", promhttp.HandlerFor(ms.r.metricsRegistry, promhttp.HandlerOpts{}))
 	mux.HandleFunc("GET /health", ms.handleHealth)
 
 	srv := &http.Server{
 		Addr:         ms.addr,
 		Handler:      mux,
 		ReadTimeout:  5 * time.Second,
-		WriteTimeout: 5 * time.Second,
+		WriteTimeout: 10 * time.Second,
 		BaseContext:  func(_ net.Listener) context.Context { return ctx },
 	}
 
@@ -58,35 +106,8 @@ func (ms *metricsServer) ListenAndServe(ctx context.Context) error {
 	}
 }
 
-// handleMetrics handles GET /metrics.
-// Response shape (JSON):
-//
-//	{
-//	  "tasks_running":   N,
-//	  "tasks_completed": N,
-//	  "tasks_failed":    N,
-//	  "uptime_seconds":  N
-//	}
-func (ms *metricsServer) handleMetrics(w http.ResponseWriter, _ *http.Request) {
-	uptimeSec := int64(time.Since(ms.r.startedAt).Seconds())
-	metricsWriteJSON(w, 0, map[string]int64{
-		"tasks_running":   ms.r.tasksRunning.Load(),
-		"tasks_completed": ms.r.tasksCompleted.Load(),
-		"tasks_failed":    ms.r.tasksFailed.Load(),
-		"uptime_seconds":  uptimeSec,
-	})
-}
-
-// handleHealth handles GET /health — simple liveness probe for the metrics server.
+// handleHealth handles GET /health — simple liveness probe.
 func (ms *metricsServer) handleHealth(w http.ResponseWriter, _ *http.Request) {
-	metricsWriteJSON(w, 0, map[string]string{"status": "ok"})
-}
-
-// metricsWriteJSON writes v as a JSON response with the given status code.
-func metricsWriteJSON(w http.ResponseWriter, status int, v any) {
 	w.Header().Set("Content-Type", "application/json")
-	if status != 0 {
-		w.WriteHeader(status)
-	}
-	_ = json.NewEncoder(w).Encode(v)
+	_, _ = w.Write([]byte(`{"status":"ok"}`))
 }

backend/internal/runner/runner.go

@@ -34,6 +34,7 @@ import (
 	"github.com/libnovel/backend/internal/pockettts"
 	"github.com/libnovel/backend/internal/scraper"
 	"github.com/libnovel/backend/internal/taskqueue"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 // Config tunes the runner behaviour.
@@ -41,6 +42,7 @@ type Config struct {
 	// WorkerID uniquely identifies this runner instance in PocketBase records.
 	WorkerID string
 	// PollInterval is how often the runner checks for new tasks.
+	// Only used in PocketBase-polling mode (RedisAddr == "").
 	PollInterval time.Duration
 	// MaxConcurrentScrape limits simultaneous book-scrape goroutines.
 	MaxConcurrentScrape int
@@ -50,9 +52,11 @@ type Config struct {
 	OrchestratorWorkers int
 	// HeartbeatInterval is how often active tasks PATCH their heartbeat_at
 	// timestamp to signal they are still alive. Defaults to 30s when 0.
+	// Only used in PocketBase-polling mode.
 	HeartbeatInterval time.Duration
 	// StaleTaskThreshold is how old a heartbeat must be (or absent) before the
 	// task is considered orphaned and reset to pending. Defaults to 2m when 0.
+	// Only used in PocketBase-polling mode.
 	StaleTaskThreshold time.Duration
 	// CatalogueRefreshInterval is how often the runner walks the full catalogue,
 	// scrapes per-book metadata, downloads covers, and re-indexes everything in
@@ -66,6 +70,15 @@ type Config struct {
 	// MetricsAddr is the HTTP listen address for the /metrics endpoint.
 	// Defaults to ":9091". Set to "" to disable.
 	MetricsAddr string
+	// RedisAddr is the address of the Redis instance used for Asynq task
+	// dispatch. When set the runner switches from PocketBase-polling mode to
+	// Asynq ServeMux mode (immediate task delivery, no polling).
+	// Supports plain "host:port" or a full "rediss://..." URL.
+	// When empty the runner falls back to PocketBase polling.
+	RedisAddr string
+	// RedisPassword is the Redis AUTH password.
+	// Not required when RedisAddr is a full URL that includes credentials.
+	RedisPassword string
 }
 
 // Dependencies are the external services the runner depends on.
@@ -99,6 +112,8 @@ type Runner struct {
 	cfg  Config
 	deps Dependencies
 
+	metricsRegistry *prometheus.Registry
+
 	// Atomic task counters — read by /metrics without locking.
 	tasksRunning   atomic.Int64
 	tasksCompleted atomic.Int64
@@ -139,15 +154,18 @@ func New(cfg Config, deps Dependencies) *Runner {
 	if deps.SearchIndex == nil {
 		deps.SearchIndex = meili.NoopClient{}
 	}
-	return &Runner{cfg: cfg, deps: deps, startedAt: time.Now()}
+	return &Runner{cfg: cfg, deps: deps, startedAt: time.Now(), metricsRegistry: prometheus.NewRegistry()}
 }
 
-// Run starts the poll loop and the metrics HTTP server, blocking until ctx is
-// cancelled.
+// Run starts the worker loop and the metrics HTTP server, blocking until ctx
+// is cancelled.
+//
+// When cfg.RedisAddr is set the runner uses Asynq (immediate task delivery).
+// Otherwise it falls back to PocketBase polling (legacy mode).
 func (r *Runner) Run(ctx context.Context) error {
 	r.deps.Log.Info("runner: starting",
 		"worker_id", r.cfg.WorkerID,
-		"poll_interval", r.cfg.PollInterval,
+		"mode", r.mode(),
 		"max_scrape", r.cfg.MaxConcurrentScrape,
 		"max_audio", r.cfg.MaxConcurrentAudio,
 		"catalogue_refresh_interval", r.cfg.CatalogueRefreshInterval,
@@ -164,6 +182,23 @@ func (r *Runner) Run(ctx context.Context) error {
 		}()
 	}
 
+	if r.cfg.RedisAddr != "" {
+		return r.runAsynq(ctx)
+	}
+	return r.runPoll(ctx)
+}
+
+// mode returns a short string describing the active dispatch mode.
+func (r *Runner) mode() string {
+	if r.cfg.RedisAddr != "" {
+		return "asynq"
+	}
+	return "poll"
+}
+
+// runPoll is the legacy PocketBase-polling dispatch loop.
+// Used when cfg.RedisAddr is empty.
+func (r *Runner) runPoll(ctx context.Context) error {
 	scrapeSem := make(chan struct{}, r.cfg.MaxConcurrentScrape)
 	audioSem := make(chan struct{}, r.cfg.MaxConcurrentAudio)
 	var wg sync.WaitGroup
@@ -181,6 +216,8 @@ func (r *Runner) Run(ctx context.Context) error {
 		r.deps.Log.Info("runner: skipping initial catalogue refresh (RUNNER_SKIP_INITIAL_CATALOGUE_REFRESH=true)")
 	}
 
+	r.deps.Log.Info("runner: poll mode active", "poll_interval", r.cfg.PollInterval)
+
 	// Run one poll immediately on startup, then on each tick.
 	for {
 		r.poll(ctx, scrapeSem, audioSem, &wg)