perf: incremental render scale, SQL pagination and observability

- BIRD filter chunking (500 CIDR); bounded job worker pool
- ListModulesPage SQL push-down; revision diff limit; batch revision prune
- DB pool tuning; Prometheus pipeline/job metrics
- Coalesce module_refresh via idempotency; GET /jobs/{id}/report
- JobAuditWriter foundation for job_audit persistence

Co-authored-by: Cursor <[email protected]>

internal/birdfmt/filter.go

@@ -9,6 +9,21 @@ import (
 
 const maxBGPASN = 4294967295
 
+const filterPrefixChunkSize = 500
+
+func writePrefixSetAcceptBlocks(b *strings.Builder, keys []string) {
+	for i := 0; i < len(keys); i += filterPrefixChunkSize {
+		end := i + filterPrefixChunkSize
+		if end > len(keys) {
+			end = len(keys)
+		}
+		chunk := keys[i:end]
+		b.WriteString("    if net ~ [ ")
+		b.WriteString(strings.Join(chunk, ", "))
+		b.WriteString(" ] then accept;\n")
+	}
+}
+
 func filterUniqueASNs(pathASNs []int64) []int64 {
 	seen := make(map[int64]struct{})
 	for _, a := range pathASNs {
@@ -52,9 +67,7 @@ func RenderExportFilterIPv4(filterName string, prefixes []netip.Prefix, pathASNs
 	b.WriteString(strings.TrimSpace(filterName))
 	b.WriteString(" {\n")
 	if len(keys) > 0 {
-		b.WriteString("    if net ~ [ ")
+		writePrefixSetAcceptBlocks(&b, keys)
-		b.WriteString(strings.Join(keys, ", "))
-		b.WriteString(" ] then accept;\n")
 	}
 	for _, asn := range asns {
 		fmt.Fprintf(&b, "    if bgp_path ~ [= * %d =] then accept;\n", asn)
@@ -95,9 +108,7 @@ func RenderExportFilterIPv6(filterName string, prefixes []netip.Prefix, pathASNs
 	b.WriteString(strings.TrimSpace(filterName))
 	b.WriteString(" {\n")
 	if len(keys) > 0 {
-		b.WriteString("    if net ~ [ ")
+		writePrefixSetAcceptBlocks(&b, keys)
-		b.WriteString(strings.Join(keys, ", "))
-		b.WriteString(" ] then accept;\n")
 	}
 	for _, asn := range asns {
 		fmt.Fprintf(&b, "    if bgp_path ~ [= * %d =] then accept;\n", asn)

internal/db/open.go

@@ -6,9 +6,12 @@ import (
 	"errors"
 	"fmt"
 	"io/fs"
+	"os"
 	"path"
 	"sort"
+	"strconv"
 	"strings"
+	"time"
 
 	"evobgp/migrations"
 
@@ -22,6 +25,17 @@ func OpenPostgresPool(ctx context.Context, dsn string) (*pgxpool.Pool, error) {
 	if err != nil {
 		return nil, err
 	}
+	if max := os.Getenv("EVOBGP_DB_MAX_CONNS"); max != "" {
+		if n, err := strconv.Atoi(strings.TrimSpace(max)); err == nil && n > 0 {
+			cfg.MaxConns = int32(n)
+		}
+	}
+	if min := os.Getenv("EVOBGP_DB_MIN_CONNS"); min != "" {
+		if n, err := strconv.Atoi(strings.TrimSpace(min)); err == nil && n >= 0 {
+			cfg.MinConns = int32(n)
+		}
+	}
+	cfg.MaxConnLifetime = 30 * time.Minute
 	pool, err := pgxpool.NewWithConfig(ctx, cfg)
 	if err != nil {
 		return nil, err

internal/httpapi/routes.go

@@ -69,6 +69,7 @@ func (s *Server) registerV1(m *http.ServeMux) {
 	m.HandleFunc("GET /bird/status", s.handleBirdStatus)
 	m.HandleFunc("GET /jobs", s.handleListJobs)
 	m.HandleFunc("GET /jobs/{job_id}", s.handleGetJob)
+	m.HandleFunc("GET /jobs/{job_id}/report", s.handleGetJobReport)
 	m.HandleFunc("POST /jobs/{job_id}/cancel", s.handleCancelJob)
 	m.HandleFunc("GET /speakers/{speaker_id}/revisions/latest", s.handleNodeLatestRevision)
 	m.HandleFunc("GET /speakers/{speaker_id}/bundle/{revision_id}", s.handleNodeBundle)
@@ -201,6 +202,22 @@ func (s *Server) handleListModules(w http.ResponseWriter, r *http.Request) {
 	}
 
 	filtered := make([]*store.Module, 0)
+	limit := parseListLimit(r)
+	cursor := r.URL.Query().Get("cursor")
+	if typeFilter == "" && enabledFilter == nil {
+		page, next, more := s.store.ListModulesPage(a.TenantID, cursor, limit)
+		for _, mod := range page {
+			filtered = append(filtered, mod)
+		}
+		items := make([]map[string]any, 0, len(filtered))
+		for _, mod := range filtered {
+			items = append(items, moduleJSON(mod))
+		}
+		writeJSON(w, http.StatusOK, map[string]any{
+			"items": items, "next_cursor": strPtrOrNull(next), "has_more": more,
+		})
+		return
+	}
 	for _, mod := range s.store.ListModules(a.TenantID) {
 		if typeFilter != "" && mod.Type != typeFilter {
 			continue
@@ -534,7 +551,8 @@ func (s *Server) enqueueModuleRefreshIfEnabled(tenantID, moduleID, trigger strin
 		return
 	}
 	mid := moduleID
-	_, _, _ = s.jobs.Enqueue(tenantID, jobs.KindModuleRefresh, nil, &mid, map[string]any{
+	key := "module_refresh:" + moduleID
+	_, _, _ = s.jobs.Enqueue(tenantID, jobs.KindModuleRefresh, &key, &mid, map[string]any{
 		"module_id": moduleID,
 		"trigger":   trigger,
 	})
@@ -601,6 +619,9 @@ func (s *Server) handleRevisionPreview(w http.ResponseWriter, r *http.Request) {
 	for k, v := range rev.PreviewFragments {
 		obj[k] = v
 	}
+	if expanded := pipeline.BuildExpandedBirdPreview(rev.PreviewFragments); expanded != "" {
+		obj[pipeline.AuxBirdFullExpandedKey()] = expanded
+	}
 	writeJSON(w, http.StatusOK, obj)
 }
 
@@ -855,6 +876,44 @@ func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, j.Snapshot())
 }
 
+func (s *Server) handleGetJobReport(w http.ResponseWriter, r *http.Request) {
+	a, ok := authFromContext(r.Context())
+	if !ok {
+		writeProblem(w, http.StatusUnauthorized, "Unauthorized", "missing auth")
+		return
+	}
+	if !s.requireAtLeast(w, a, "viewer") {
+		return
+	}
+	j, err := s.jobs.Get(a.TenantID, r.PathValue("job_id"))
+	if err != nil {
+		writeProblem(w, http.StatusNotFound, "Not Found", "job not found")
+		return
+	}
+	snap := j.Snapshot()
+	meta, _ := snap["meta"].(map[string]any)
+	out := map[string]any{
+		"job_id":    snap["job_id"],
+		"kind":      snap["kind"],
+		"status":    snap["status"],
+		"meta":      meta,
+		"error":     snap["error"],
+		"created_at": snap["created_at"],
+	}
+	if meta != nil {
+		if v, ok := meta["log_entries"]; ok {
+			out["log_entries"] = v
+		}
+		if v, ok := meta["log_total"]; ok {
+			out["log_total"] = v
+		}
+		if v, ok := meta["revision_id"]; ok {
+			out["revision_id"] = v
+		}
+	}
+	writeJSON(w, http.StatusOK, out)
+}
+
 func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) {
 	a, ok := authFromContext(r.Context())
 	if !ok {

internal/jobs/job.go

@@ -9,6 +9,8 @@ import (
 	"sync"
 	"time"
 
+	"evobgp/internal/observability"
+
 	"github.com/google/uuid"
 )
 
@@ -178,6 +180,7 @@ type Registry struct {
 	byID        map[string]*Job
 	byIdempo    map[idempoKey]*Job
 	workerStart func(j *Job)
+	workerSem   chan struct{}
 }
 
 type idempoKey struct {
@@ -186,13 +189,22 @@ type idempoKey struct {
 }
 
 func NewRegistry(workerStart func(j *Job)) *Registry {
+	maxWorkers := registryMaxConcurrentJobs()
 	return &Registry{
 		byID:        make(map[string]*Job),
 		byIdempo:    make(map[idempoKey]*Job),
 		workerStart: workerStart,
+		workerSem:   make(chan struct{}, maxWorkers),
 	}
 }
 
+func registryMaxConcurrentJobs() int {
+	if n, err := strconv.Atoi(strings.TrimSpace(os.Getenv("EVOBGP_JOB_MAX_CONCURRENT"))); err == nil && n > 0 {
+		return n
+	}
+	return 8
+}
+
 // pruneTerminalIfOver удаляет самые старые завершённые джобы (succeeded/failed/cancelled), пока len(byID) > maxJobs.
 func (r *Registry) pruneTerminalIfOver(maxJobs int) {
 	if r == nil || maxJobs <= 0 || len(r.byID) <= maxJobs {
@@ -244,7 +256,11 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module
 	if idempotencyKey != nil && *idempotencyKey != "" {
 		k := idempoKey{tenant: tenantID, key: *idempotencyKey}
 		if existing, ok := r.byIdempo[k]; ok {
-			return existing, false, nil
+			st := existing.statusLocked()
+			if st == StatusQueued || st == StatusRunning {
+				return existing, false, nil
+			}
+			delete(r.byIdempo, k)
 		}
 	}
 
@@ -265,7 +281,17 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module
 	r.pruneTerminalIfOver(maxJobs)
 
 	if r.workerStart != nil {
-		go r.workerStart(j)
+		go func() {
+			r.workerSem <- struct{}{}
+			active := len(r.workerSem)
+			capacity := cap(r.workerSem)
+			observability.RecordJobQueueDepth(active, capacity)
+			defer func() {
+				<-r.workerSem
+				observability.RecordJobQueueDepth(len(r.workerSem), capacity)
+			}()
+			r.workerStart(j)
+		}()
 	}
 	return j, true, nil
 }

internal/observability/metrics.go

@@ -80,6 +80,32 @@ var (
 		Help:      "Prefix row count after CIDR aggregation on tenant render.",
 		Buckets:   prometheus.ExponentialBuckets(1, 2, 16),
 	})
+
+	pipelineRefreshDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Name:      "pipeline_refresh_duration_seconds",
+		Help:      "Module refresh ingest duration by module type.",
+		Buckets:   prometheus.ExponentialBuckets(0.05, 2, 14),
+	}, []string{"module_type"})
+
+	renderPrefixCount = promauto.NewHistogram(prometheus.HistogramOpts{
+		Namespace: namespace,
+		Name:      "render_prefix_count",
+		Help:      "Materialized prefix count per tenant render.",
+		Buckets:   prometheus.ExponentialBuckets(10, 2, 16),
+	})
+
+	jobQueueActive = promauto.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Name:      "job_queue_active",
+		Help:      "Currently running in-process async jobs.",
+	})
+
+	jobQueueCapacity = promauto.NewGauge(prometheus.GaugeOpts{
+		Namespace: namespace,
+		Name:      "job_queue_capacity",
+		Help:      "Maximum concurrent in-process async jobs.",
+	})
 )
 
 var (
@@ -99,6 +125,27 @@ func RecordPrefixAggregation(rawCount, aggregatedCount int, duration time.Durati
 	prefixAggregationDuration.Observe(duration.Seconds())
 	prefixAggregationRawCount.Observe(float64(rawCount))
 	prefixAggregationAggregatedCount.Observe(float64(aggregatedCount))
+	renderPrefixCount.Observe(float64(aggregatedCount))
+}
+
+// RecordPipelineRefresh records module ingest duration.
+func RecordPipelineRefresh(moduleType string, duration time.Duration) {
+	if moduleType == "" {
+		moduleType = "unknown"
+	}
+	pipelineRefreshDuration.WithLabelValues(moduleType).Observe(duration.Seconds())
+}
+
+// RecordJobQueueDepth updates in-process job worker utilization gauges.
+func RecordJobQueueDepth(active, capacity int) {
+	if active < 0 {
+		active = 0
+	}
+	if capacity < 0 {
+		capacity = 0
+	}
+	jobQueueActive.Set(float64(active))
+	jobQueueCapacity.Set(float64(capacity))
 }
 
 // RecordJobTerminal increments jobs_finished_total for terminal statuses.

internal/repository/job_audit.go

@@ -0,0 +1,50 @@
+package repository
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// JobAuditWriter persists async job lifecycle rows to job_audit (optional cross-process queue foundation).
+type JobAuditWriter struct {
+	pool *pgxpool.Pool
+}
+
+func NewJobAuditWriter(pool *pgxpool.Pool) *JobAuditWriter {
+	if pool == nil {
+		return nil
+	}
+	return &JobAuditWriter{pool: pool}
+}
+
+// UpsertRunning inserts or updates a running job row (best-effort).
+func (w *JobAuditWriter) UpsertRunning(ctx context.Context, tenantID, jobID, kind string, idempotencyKey *string, meta map[string]any) {
+	if w == nil || w.pool == nil {
+		return
+	}
+	metaJSON, _ := json.Marshal(meta)
+	var idem any
+	if idempotencyKey != nil && *idempotencyKey != "" {
+		idem = *idempotencyKey
+	}
+	_, _ = w.pool.Exec(ctx, `
+		INSERT INTO job_audit (id, tenant_id, kind, status, idempotency_key, meta_json, created_at, started_at)
+		VALUES ($1::uuid, $2::uuid, $3, 'running', $4, $5::jsonb, now(), now())
+		ON CONFLICT (tenant_id, idempotency_key) WHERE idempotency_key IS NOT NULL
+		DO UPDATE SET status='running', started_at=now(), meta_json=EXCLUDED.meta_json`,
+		jobID, tenantID, kind, idem, metaJSON)
+}
+
+// MarkTerminal updates job_audit terminal state (best-effort).
+func (w *JobAuditWriter) MarkTerminal(ctx context.Context, tenantID, jobID, status string, errMsg *string, finishedAt time.Time) {
+	if w == nil || w.pool == nil {
+		return
+	}
+	_, _ = w.pool.Exec(ctx, `
+		UPDATE job_audit SET status=$3, error_message=$4, finished_at=$5
+		WHERE id=$1::uuid AND tenant_id=$2::uuid`,
+		jobID, tenantID, status, errMsg, finishedAt.UTC())
+}

internal/repository/postgres.go

@@ -162,6 +162,75 @@ func (p *Postgres) ListModules(tenantID string) []*store.Module {
 	return out
 }
 
+func (p *Postgres) ListModulesPage(tenantID, cursor string, limit int) ([]*store.Module, string, bool) {
+	if limit <= 0 {
+		limit = 50
+	}
+	off := 0
+	if cursor != "" {
+		if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
+			off = n
+		}
+	}
+	ctx := context.Background()
+	rows, err := p.pool.Query(ctx, `
+		SELECT id, type, name, enabled, priority, doh_profile_id::text, doh_resolver_policy,
+			refresh_interval_sec, cron_expr, default_community_id::text, last_refreshed_at
+		FROM module WHERE tenant_id = $1 AND deleted_at IS NULL
+		ORDER BY priority, name
+		LIMIT $2 OFFSET $3`, tenantID, limit+1, off)
+	if err != nil {
+		return nil, "", false
+	}
+	defer rows.Close()
+	var out []*store.Module
+	moduleByID := make(map[string]*store.Module)
+	for rows.Next() {
+		var m store.Module
+		m.TenantID = tenantID
+		var doh, dc, cron *string
+		var refresh *int32
+		var last *time.Time
+		if err := rows.Scan(&m.ID, &m.Type, &m.Name, &m.Enabled, &m.Priority, &doh, &m.DohResolverPolicy, &refresh, &cron, &dc, &last); err != nil {
+			continue
+		}
+		m.DohResolverPolicy = store.NormalizeDohResolverPolicy(m.DohResolverPolicy)
+		if refresh != nil {
+			m.RefreshIntervalSec = int(*refresh)
+		}
+		if cron != nil {
+			m.CronExpr = *cron
+		}
+		if doh != nil && *doh != "" {
+			m.DohProfileID = doh
+		}
+		if dc != nil && *dc != "" {
+			m.DefaultCommunityID = dc
+		}
+		if last != nil {
+			t := last.UTC()
+			m.LastRefreshedAt = &t
+		}
+		out = append(out, &m)
+		moduleByID[m.ID] = &m
+	}
+	if err := p.batchFillModuleDohFields(ctx, moduleByID); err != nil {
+		return nil, "", false
+	}
+	more := len(out) > limit
+	if more {
+		out = out[:limit]
+	}
+	next := ""
+	if more {
+		next = fmt.Sprintf("%d", off+limit)
+	}
+	if len(out) == 0 {
+		return nil, "", false
+	}
+	return out, next, more
+}
+
 func (p *Postgres) GetModule(tenantID, moduleID string) (*store.Module, error) {
 	ctx := context.Background()
 	var m store.Module
@@ -762,6 +831,8 @@ func (p *Postgres) CreateRollbackRevision(tenantID, sourceRevisionID string) (st
 	return newID, nil
 }
 
+const maxRevisionDiffRows = 5000
+
 func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, error) {
 	if _, err := p.GetRevision(tenantID, aID); err != nil {
 		return nil, err
@@ -786,7 +857,7 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
 			SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
 			EXCEPT
 			SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
-		) s ORDER BY 1`, bID, aID)
+		) s ORDER BY 1 LIMIT $3`, bID, aID, maxRevisionDiffRows+1)
 	if err != nil {
 		return nil, err
 	}
@@ -798,13 +869,18 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
 			continue
 		}
 		added = append(added, s)
+		if len(added) > maxRevisionDiffRows {
+			added = added[:maxRevisionDiffRows]
+			break
+		}
 	}
+	addedTruncated := len(added) >= maxRevisionDiffRows
 	rowsRem, err := p.pool.Query(ctx, `
 		SELECT prefix::text FROM (
 			SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
 			EXCEPT
 			SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
-		) s ORDER BY 1`, aID, bID)
+		) s ORDER BY 1 LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
 	if err != nil {
 		return nil, err
 	}
@@ -816,40 +892,56 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
 			continue
 		}
 		removed = append(removed, s)
+		if len(removed) > maxRevisionDiffRows {
+			removed = removed[:maxRevisionDiffRows]
+			break
+		}
 	}
 	return map[string]any{
 		"revision_a": aID,
 		"revision_b": bID,
 		"prefixes": map[string]any{
 			"added": added, "removed": removed, "unchanged_count": unchanged,
+			"truncated": addedTruncated || len(removed) >= maxRevisionDiffRows,
 		},
 	}, nil
 }
 
 func (p *Postgres) PruneRevisionsBefore(tenantID string, cutoff time.Time) (int, error) {
 	ctx := context.Background()
-	cmd, err := p.pool.Exec(ctx, `
+	total := 0
-		DELETE FROM config_revision AS cr
+	const batchSize = 50
-		WHERE cr.tenant_id = $1
+	for {
-		  AND cr.created_at < $2
+		cmd, err := p.pool.Exec(ctx, `
-		  AND cr.id <> (
+			DELETE FROM config_revision AS cr
-			SELECT id
+			WHERE cr.id IN (
-			FROM config_revision
+				SELECT id FROM config_revision
-			WHERE tenant_id = $1
+				WHERE tenant_id = $1
-			ORDER BY created_at DESC
+				  AND created_at < $2
-			LIMIT 1
+				  AND id <> (
-		  )
+					SELECT id FROM config_revision
-		  AND NOT EXISTS (
+					WHERE tenant_id = $1
-			SELECT 1
+					ORDER BY created_at DESC
-			FROM bgp_speaker AS sp
+					LIMIT 1
-			WHERE sp.tenant_id = $1
+				  )
-			  AND (sp.last_applied_revision_id = cr.id OR sp.published_revision_id = cr.id)
+				  AND NOT EXISTS (
-		  )`,
+					SELECT 1 FROM bgp_speaker AS sp
-		tenantID, cutoff.UTC())
+					WHERE sp.tenant_id = $1
-	if err != nil {
+					  AND (sp.last_applied_revision_id = config_revision.id OR sp.published_revision_id = config_revision.id)
-		return 0, err
+				  )
+				ORDER BY created_at ASC
+				LIMIT $3
+			)`, tenantID, cutoff.UTC(), batchSize)
+		if err != nil {
+			return total, err
+		}
+		n := int(cmd.RowsAffected())
+		total += n
+		if n < batchSize {
+			break
+		}
 	}
-	return int(cmd.RowsAffected()), nil
+	return total, nil
 }
 
 func (p *Postgres) SetLastAppliedRevision(tenantID, speakerID, revisionID string) error {

internal/store/memory.go

@@ -382,6 +382,11 @@ func (m *Memory) ListModules(tenantID string) []*Module {
 	return out
 }
 
+func (m *Memory) ListModulesPage(tenantID, cursor string, limit int) ([]*Module, string, bool) {
+	all := m.ListModules(tenantID)
+	return PaginateOffset(all, cursor, limit)
+}
+
 // ListPeers returns BGP peers for a tenant (sorted by name).
 func (m *Memory) ListPeers(tenantID string) []*BGPPeer {
 	m.mu.RLock()