fix(db): update prefix handling in module_prefix_snapshot_row

Modified the prefix column type in the module_prefix_snapshot_row table to TEXT, allowing for more flexible input. Adjusted related SQL queries and Go struct tags to ensure compatibility with JSON serialization. Cleaned up migration logic to handle prefix and community_id fields more robustly.

docs/db-diagnostics.md

@@ -0,0 +1,90 @@
+# Диагностика схемы PostgreSQL (EvoBGP)
+
+Runbook для оценки объёма БД и узких мест **перед** и **после** миграций оптимизации схемы. Выполнять на staging или production read-only сессией.
+
+## Подключение
+
+```bash
+psql "$EVOBGP_DATABASE_URL"
+```
+
+## 1. Размеры таблиц и индексов
+
+```sql
+SELECT relname,
+       pg_size_pretty(pg_total_relation_size(relid)) AS total,
+       pg_size_pretty(pg_relation_size(relid)) AS heap,
+       pg_size_pretty(pg_indexes_size(relid)) AS indexes
+FROM pg_catalog.pg_statio_user_tables
+ORDER BY pg_total_relation_size(relid) DESC;
+```
+
+**Ожидание:** лидеры — `revision_materialized_prefix`, `config_revision` (TOAST от preview), JSONB-кэши.
+
+## 2. Seq scan (горячие таблицы)
+
+```sql
+SELECT schemaname, relname, seq_scan, seq_tup_read, idx_scan
+FROM pg_stat_user_tables
+WHERE schemaname = 'public'
+ORDER BY seq_tup_read DESC;
+```
+
+Сброс статистики после деплоя: `SELECT pg_stat_reset();` (только осознанно, теряется baseline).
+
+## 3. Неиспользуемые индексы
+
+```sql
+SELECT indexrelname, idx_scan, pg_size_pretty(pg_relation_size(indexrelid)) AS size
+FROM pg_stat_user_indexes
+WHERE schemaname = 'public' AND idx_scan = 0
+ORDER BY pg_relation_size(indexrelid) DESC;
+```
+
+## 4. Дубликаты в materialized prefixes
+
+Перед UNIQUE `(revision_id, prefix, community_id, source)`:
+
+```sql
+SELECT revision_id, prefix, community_id, source, COUNT(*) AS n
+FROM revision_materialized_prefix
+GROUP BY 1, 2, 3, 4
+HAVING COUNT(*) > 1
+LIMIT 20;
+```
+
+## 5. Шаблон отчёта staging
+
+| Метрика | До | После | Дата |
+|---------|-----|-------|------|
+| `revision_materialized_prefix` total | | | |
+| `config_revision` total | | | |
+| `module_prefix_snapshot` total | | | |
+| `asn_prefix_cache` total | | | |
+| Top seq_scan table | | | |
+| Unused indexes (count) | | | |
+
+## 6. EXPLAIN для типовых запросов
+
+```sql
+-- Список префиксов ревизии (keyset)
+EXPLAIN (ANALYZE, BUFFERS)
+SELECT prefix::text, community_id::text, source
+FROM revision_materialized_prefix
+WHERE revision_id = '<revision-uuid>'::uuid
+ORDER BY id
+LIMIT 51;
+
+-- Diff added (anti-join)
+EXPLAIN (ANALYZE, BUFFERS)
+SELECT b.prefix::text
+FROM revision_materialized_prefix b
+LEFT JOIN revision_materialized_prefix a
+  ON a.revision_id = '<rev-a>'::uuid AND a.prefix = b.prefix
+WHERE b.revision_id = '<rev-b>'::uuid
+  AND a.prefix IS NULL
+ORDER BY b.prefix
+LIMIT 5001;
+```
+
+Цель: Index Scan / Bitmap Index Scan по `(revision_id, …)`, без Seq Scan на больших таблицах.

docs/production-checklist.md

@@ -0,0 +1,26 @@
+# Production checklist
+
+Краткий чеклист перед выводом EvoBGP в production (10+ клиентов, нестабильная сеть).
+
+## Обязательно
+
+- `EVOBGP_SEED_DEMO=0` — отключить demo-tenant и токен `Bearer dev`.
+- `EVOBGP_DEV_INSECURE` не задавать или `0` — не использовать lab-флаги в prod.
+- `EVOBGP_BUNDLE_SEED_HEX` — задать стабильный hex-ключ подписи бандлов; сохранить pubkey для нод.
+- PostgreSQL с TLS (`sslmode` не `disable`) при доступе вне private network.
+- `EVOBGP_CORS_ORIGINS` — явный whitelist origin веб-панели.
+- `EVOBGP_STALE_ON_UPSTREAM_ERROR=1` (по умолчанию) — stale snapshot при сбоях CDN/ASN/DoH.
+
+## Рекомендуется
+
+- `EVOBGP_JOB_MAX_CONCURRENT=16`, `EVOBGP_DB_MAX_CONNS=25`, `EVOBGP_COLLECT_CONCURRENCY=16` при росте tenants.
+- `EVOBGP_NODE_DISPATCH_INSECURE_TLS=0` — только валидный TLS к agent.
+- Ограничить `/metrics` сетевой политикой или reverse proxy.
+- Профиль `evobgp-all` или HA API + персистентная `job_audit` (PostgreSQL).
+- Мониторинг drift: `evobgp-deploy`, `last_applied_revision_id` vs published.
+
+## Не использовать в prod
+
+- `EVOBGP_CDN_ALLOW_PRIVATE=1` — только тесты/lab.
+- Plaintext `EVOBGP_API_KEYS` без ротации (break-glass — временно).
+- Ручное редактирование `evobgp_*.conf` на нодах без ревизии.

internal/agentserver/server.go

@@ -14,6 +14,8 @@ import (
 	"evobgp/internal/nodecli"
 )
 
+const upstreamErrorDetail = "upstream request failed"
+
 // Config holds evobgp-agent serve settings.
 type Config struct {
 	Listen          string
@@ -85,7 +87,7 @@ func (s *Server) handleBirdProtocols(w http.ResponseWriter, r *http.Request) {
 	out, err := birdfmt.ShowProtocols(ctx, sock, strings.TrimSpace(s.cfg.BirdcBin))
 	if err != nil {
 		log.Printf("agentserver: bird protocols: %v", err)
-		writeProblem(w, http.StatusBadGateway, err.Error())
+		writeProblem(w, http.StatusBadGateway, upstreamErrorDetail)
 		return
 	}
 	writeJSON(w, http.StatusOK, map[string]any{
@@ -126,7 +128,7 @@ func (s *Server) handleSync(w http.ResponseWriter, r *http.Request) {
 	})
 	if err != nil {
 		log.Printf("agentserver: sync: %v", err)
-		writeProblem(w, http.StatusBadGateway, err.Error())
+		writeProblem(w, http.StatusBadGateway, upstreamErrorDetail)
 		return
 	}
 	if s.cfg.OnSyncSuccess != nil {

internal/asnresolve/ripestat.go

@@ -13,6 +13,8 @@ import (
 	"strconv"
 	"strings"
 	"time"
+
+	"evobgp/internal/httpclient"
 )
 
 // DefaultRIPEStatURL is the RIPEstat announced-prefixes data call (no API key).
@@ -24,7 +26,7 @@ const DefaultASOverviewURL = "https://stat.ripe.net/data/as-overview/data.json"
 // AnnouncedPrefixes returns currently announced IPv4/IPv6 prefixes for the ASN (best-effort via RIPEstat).
 func AnnouncedPrefixes(ctx context.Context, hc *http.Client, asn int64) ([]netip.Prefix, error) {
 	if hc == nil {
-		hc = http.DefaultClient
+		hc = httpclient.New(httpclient.DefaultTimeout)
 	}
 	base := strings.TrimSpace(os.Getenv("EVOBGP_RIPESTAT_ANNOUNCED_PREFIXES_URL"))
 	if base == "" {
@@ -38,7 +40,7 @@ func AnnouncedPrefixes(ctx context.Context, hc *http.Client, asn int64) ([]netip
 	req.Header.Set("Accept", "application/json")
 	req.Header.Set("User-Agent", "evobgp-asnresolve/1.0")
 
-	resp, err := hc.Do(req)
+	resp, err := httpclient.DoWithBreaker(ctx, hc, req, 3)
 	if err != nil {
 		return nil, fmt.Errorf("ripestat fetch AS%d: %w", asn, err)
 	}
@@ -86,7 +88,7 @@ func AnnouncedPrefixes(ctx context.Context, hc *http.Client, asn int64) ([]netip
 // ASHolderName returns the holder / organization label for the ASN from RIPEstat as-overview (best-effort).
 func ASHolderName(ctx context.Context, hc *http.Client, asn int64) (string, error) {
 	if hc == nil {
-		hc = http.DefaultClient
+		hc = httpclient.New(httpclient.DefaultTimeout)
 	}
 	base := strings.TrimSpace(os.Getenv("EVOBGP_RIPESTAT_AS_OVERVIEW_URL"))
 	if base == "" {
@@ -100,7 +102,7 @@ func ASHolderName(ctx context.Context, hc *http.Client, asn int64) (string, erro
 	req.Header.Set("Accept", "application/json")
 	req.Header.Set("User-Agent", "evobgp-asnresolve/1.0")
 
-	resp, err := hc.Do(req)
+	resp, err := httpclient.DoWithBreaker(ctx, hc, req, 3)
 	if err != nil {
 		return "", fmt.Errorf("ripestat as-overview AS%d: %w", asn, err)
 	}

internal/httpapi/bootstrap.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"evobgp/internal/db"
+	"evobgp/internal/httpclient"
 	"evobgp/internal/jobs"
 	"evobgp/internal/observability"
 	"evobgp/internal/repository"
@@ -17,7 +18,7 @@ import (
 
 // NewCDNHTTPClient returns the shared HTTP client for CDN and preview fetches (PERF-02 / ERR-03).
 func NewCDNHTTPClient() *http.Client {
-	return &http.Client{Timeout: 45 * time.Second}
+	return httpclient.New(httpclient.DefaultTimeout)
 }
 
 // BootstrapWorkers opens the same store.Backend and jobs.Registry as New (without HTTP or bundle keys).
@@ -55,18 +56,40 @@ func BootstrapWorkers(ctx context.Context, opts Options) (store.Backend, *jobs.R
 	wk.Registry = reg
 	if pool != nil {
 		audit := repository.NewJobAuditWriter(pool)
-		reg.SetTerminalHook(func(j *jobs.Job) {
+		jobMeta := func(j *jobs.Job) map[string]any {
 			if j == nil {
-				return
+				return nil
 			}
 			st := j.Snapshot()
-			status, _ := st["status"].(string)
-			var errMsg *string
-			if e, ok := st["error"].(string); ok && e != "" {
-				errMsg = &e
-			}
-			audit.MarkTerminal(context.Background(), j.TenantID, j.ID, status, errMsg, time.Now().UTC())
-		})
+			meta, _ := st["meta"].(map[string]any)
+			return meta
+		}
+		reg.SetPersistHooks(
+			func(j *jobs.Job) {
+				if j == nil {
+					return
+				}
+				audit.UpsertQueued(context.Background(), j.TenantID, j.ID, j.Kind, j.IdempotencyKey, j.ModuleID, jobMeta(j))
+			},
+			func(j *jobs.Job) {
+				if j == nil {
+					return
+				}
+				audit.UpsertRunning(context.Background(), j.TenantID, j.ID, j.Kind, j.IdempotencyKey, jobMeta(j))
+			},
+			func(j *jobs.Job) {
+				if j == nil {
+					return
+				}
+				st := j.Snapshot()
+				status, _ := st["status"].(string)
+				var errMsg *string
+				if e, ok := st["error"].(string); ok && e != "" {
+					errMsg = &e
+				}
+				audit.MarkTerminal(context.Background(), j.TenantID, j.ID, status, errMsg, time.Now().UTC())
+			},
+		)
 	}
 	observability.RegisterStoreBackend(backend)
 	return backend, reg, pool, nil

internal/httpapi/routes.go

@@ -579,7 +579,7 @@ func (s *Server) handleGetRevision(w http.ResponseWriter, r *http.Request) {
 	if !s.requireAtLeast(w, a, "viewer") {
 		return
 	}
-	rev, err := s.store.GetRevision(a.TenantID, r.PathValue("revision_id"))
+	rev, err := s.store.GetRevisionSummary(a.TenantID, r.PathValue("revision_id"))
 	if err != nil {
 		writeProblem(w, http.StatusNotFound, "Not Found", "revision not found")
 		return

internal/httpapi/routes_crud.go

@@ -242,6 +242,14 @@ func (s *Server) handlePreviewCDNSource(w http.ResponseWriter, r *http.Request)
 		writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "url is required")
 		return
 	}
+	if _, err := pipeline.ValidateCDNURL(u); err != nil {
+		writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", invalidInputDetail)
+		return
+	}
+	if err := pipeline.ResolveCDNURLHost(r.Context(), u); err != nil {
+		writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", invalidInputDetail)
+		return
+	}
 	mod, err := s.store.GetModule(a.TenantID, r.PathValue("module_id"))
 	if err != nil {
 		writeStoreErr(w, err)
@@ -304,6 +312,16 @@ func (s *Server) handlePostCDNSource(w http.ResponseWriter, r *http.Request) {
 		writeProblem(w, http.StatusBadRequest, "Bad Request", "invalid json")
 		return
 	}
+	if body.URL != "" {
+		if _, err := pipeline.ValidateCDNURL(body.URL); err != nil {
+			writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", invalidInputDetail)
+			return
+		}
+		if err := pipeline.ResolveCDNURLHost(r.Context(), body.URL); err != nil {
+			writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", invalidInputDetail)
+			return
+		}
+	}
 	mid := r.PathValue("module_id")
 	x, err := s.store.CreateCDNSource(a.TenantID, mid, &body)
 	if err != nil {
@@ -324,6 +342,16 @@ func (s *Server) handlePatchCDNSource(w http.ResponseWriter, r *http.Request) {
 		writeProblem(w, http.StatusBadRequest, "Bad Request", "invalid json")
 		return
 	}
+	if body.URL != nil && strings.TrimSpace(*body.URL) != "" {
+		if _, err := pipeline.ValidateCDNURL(*body.URL); err != nil {
+			writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", invalidInputDetail)
+			return
+		}
+		if err := pipeline.ResolveCDNURLHost(r.Context(), *body.URL); err != nil {
+			writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", invalidInputDetail)
+			return
+		}
+	}
 	mid := r.PathValue("module_id")
 	x, err := s.store.UpdateCDNSource(a.TenantID, mid, r.PathValue("source_id"), &body)
 	if err != nil {

internal/httpapi/speakers.go

@@ -39,7 +39,10 @@ func speakerJSONFromStore(st store.Backend, sp *store.Speaker) map[string]any {
 	if strings.TrimSpace(sp.MetaJSON) != "" && sp.MetaJSON != "{}" {
 		var raw map[string]any
 		if json.Unmarshal([]byte(sp.MetaJSON), &raw) == nil {
-			m["meta_json"] = raw
+			delete(raw, "agent_secret")
+			if len(raw) > 0 {
+				m["meta_json"] = raw
+			}
 		}
 	}
 	if meta.AgentDomain != "" {

internal/httpapi/speakers_redact_test.go

@@ -0,0 +1,62 @@
+package httpapi
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestGetSpeaker_redactsAgentSecret(t *testing.T) {
+	srv, err := New(Options{InsecureDev: true, SeedDemo: true, BundleSeedHex: testBundleSeed})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer srv.Close()
+	tenant, _, _, _, demoSpk := srv.Store().DemoIDs()
+	mustSetTestAPIKeys(t, srv, "vwkey|"+tenant+"|viewer")
+
+	req := httptest.NewRequest(http.MethodGet, "/v1/speakers/"+demoSpk, nil)
+	req.Header.Set("Authorization", "Bearer vwkey")
+	rec := httptest.NewRecorder()
+	srv.Handler().ServeHTTP(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d body %s", rec.Code, rec.Body.String())
+	}
+	var out map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
+		t.Fatal(err)
+	}
+	if out["agent_secret"] != nil {
+		t.Fatalf("agent_secret must not appear at top level: %#v", out["agent_secret"])
+	}
+	meta, _ := out["meta_json"].(map[string]any)
+	if meta != nil {
+		if v, ok := meta["agent_secret"]; ok && v != nil && v != "" {
+			t.Fatalf("agent_secret must be redacted from meta_json: %#v", v)
+		}
+	}
+}
+
+func TestListSpeakers_redactsAgentSecret(t *testing.T) {
+	srv, err := New(Options{InsecureDev: true, SeedDemo: true, BundleSeedHex: testBundleSeed})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer srv.Close()
+	tenant, _, _, _, _ := srv.Store().DemoIDs()
+	mustSetTestAPIKeys(t, srv, "vwkey|"+tenant+"|viewer")
+
+	req := httptest.NewRequest(http.MethodGet, "/v1/speakers", nil)
+	req.Header.Set("Authorization", "Bearer vwkey")
+	rec := httptest.NewRecorder()
+	srv.Handler().ServeHTTP(rec, req)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d body %s", rec.Code, rec.Body.String())
+	}
+	body := rec.Body.String()
+	if strings.Contains(body, "agent_secret") {
+		t.Fatalf("list response must not contain agent_secret: %s", body)
+	}
+}

internal/httpclient/circuit.go

@@ -0,0 +1,55 @@
+package httpclient
+
+import (
+	"sync"
+	"time"
+)
+
+const (
+	defaultBreakerThreshold = 5
+	defaultBreakerCooldown  = 30 * time.Second
+)
+
+type hostBreaker struct {
+	mu        sync.Mutex
+	failures  int
+	openUntil time.Time
+}
+
+var hostBreakers sync.Map // string -> *hostBreaker
+
+func breakerForHost(host string) *hostBreaker {
+	if host == "" {
+		host = "_"
+	}
+	v, _ := hostBreakers.LoadOrStore(host, &hostBreaker{})
+	return v.(*hostBreaker)
+}
+
+func (b *hostBreaker) allow() bool {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return time.Now().After(b.openUntil)
+}
+
+func (b *hostBreaker) recordSuccess() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.failures = 0
+	b.openUntil = time.Time{}
+}
+
+func (b *hostBreaker) recordFailure() {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	b.failures++
+	if b.failures >= defaultBreakerThreshold {
+		b.openUntil = time.Now().Add(defaultBreakerCooldown)
+		b.failures = 0
+	}
+}
+
+// ResetHostBreakers clears all circuit breakers (tests only).
+func ResetHostBreakers() {
+	hostBreakers = sync.Map{}
+}

internal/httpclient/circuit_test.go

@@ -0,0 +1,34 @@
+package httpclient
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestDoWithBreaker_opensAfterFailures(t *testing.T) {
+	ResetHostBreakers()
+	var calls atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls.Add(1)
+		http.Error(w, "fail", http.StatusBadGateway)
+	}))
+	defer srv.Close()
+
+	hc := New(5 * time.Second)
+	for i := 0; i < defaultBreakerThreshold*3; i++ {
+		req, _ := http.NewRequest(http.MethodGet, srv.URL, nil)
+		_, _ = DoWithBreaker(context.Background(), hc, req, 1)
+	}
+	req, _ := http.NewRequest(http.MethodGet, srv.URL, nil)
+	_, err := DoWithBreaker(context.Background(), hc, req, 1)
+	if err == nil || err.Error() == "" {
+		t.Fatal("expected circuit open error")
+	}
+	if got := calls.Load(); got == 0 {
+		t.Fatal("expected at least one upstream call")
+	}
+}

internal/httpclient/httpclient.go

@@ -0,0 +1,87 @@
+// Package httpclient provides shared HTTP clients and retry helpers for outbound calls.
+package httpclient
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+const DefaultTimeout = 45 * time.Second
+
+// New returns an HTTP client with timeout and tuned idle connection pooling.
+func New(timeout time.Duration) *http.Client {
+	if timeout <= 0 {
+		timeout = DefaultTimeout
+	}
+	tr := http.DefaultTransport.(*http.Transport).Clone()
+	tr.MaxIdleConns = 100
+	tr.MaxIdleConnsPerHost = 10
+	return &http.Client{Timeout: timeout, Transport: tr}
+}
+
+// DoWithRetry executes hc.Do(req) up to maxAttempts times with linear backoff.
+func DoWithRetry(ctx context.Context, hc *http.Client, req *http.Request, maxAttempts int) (*http.Response, error) {
+	if maxAttempts <= 0 {
+		maxAttempts = 3
+	}
+	var lastErr error
+	for attempt := 0; attempt < maxAttempts; attempt++ {
+		if attempt > 0 {
+			wait := time.Duration(attempt) * 2 * time.Second
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case <-time.After(wait):
+			}
+			if req.GetBody != nil {
+				body, err := req.GetBody()
+				if err != nil {
+					return nil, err
+				}
+				req.Body = body
+			}
+		}
+		reqClone := req.Clone(ctx)
+		resp, err := hc.Do(reqClone)
+		if err != nil {
+			lastErr = err
+			continue
+		}
+		if resp.StatusCode >= 500 {
+			_, _ = io.Copy(io.Discard, resp.Body)
+			_ = resp.Body.Close()
+			lastErr = fmt.Errorf("httpclient: upstream %s", resp.Status)
+			continue
+		}
+		return resp, nil
+	}
+	if lastErr != nil {
+		return nil, lastErr
+	}
+	return nil, fmt.Errorf("httpclient: request failed after %d attempts", maxAttempts)
+}
+
+// DoWithBreaker applies per-host circuit breaking then retries transient failures.
+func DoWithBreaker(ctx context.Context, hc *http.Client, req *http.Request, maxAttempts int) (*http.Response, error) {
+	if req == nil || req.URL == nil {
+		return nil, fmt.Errorf("httpclient: nil request")
+	}
+	br := breakerForHost(req.URL.Hostname())
+	if !br.allow() {
+		return nil, fmt.Errorf("httpclient: circuit open for %s", req.URL.Hostname())
+	}
+	resp, err := DoWithRetry(ctx, hc, req, maxAttempts)
+	if err != nil {
+		br.recordFailure()
+		return nil, err
+	}
+	if resp.StatusCode >= 500 {
+		br.recordFailure()
+		return resp, nil
+	}
+	br.recordSuccess()
+	return resp, nil
+}

internal/httpclient/httpclient_test.go

@@ -0,0 +1,39 @@
+package httpclient
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func TestDoWithRetry_retriesOn500(t *testing.T) {
+	var calls int
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		if calls < 3 {
+			http.Error(w, "fail", http.StatusBadGateway)
+			return
+		}
+		w.WriteHeader(http.StatusOK)
+		_, _ = w.Write([]byte("ok"))
+	}))
+	defer srv.Close()
+
+	req, err := http.NewRequest(http.MethodGet, srv.URL, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := DoWithRetry(context.Background(), New(5*time.Second), req, 3)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("status %d", resp.StatusCode)
+	}
+	if calls != 3 {
+		t.Fatalf("want 3 calls, got %d", calls)
+	}
+}

internal/ingest/run.go

@@ -3,11 +3,11 @@ package ingest
 import (
 	"context"
 	"log"
-	"net/http"
 	"time"
 
 	"evobgp/internal/broker"
 	"evobgp/internal/config"
+	"evobgp/internal/httpclient"
 	"evobgp/internal/pipeline"
 	"evobgp/internal/store"
 )
@@ -17,6 +17,8 @@ type Deps struct {
 	Store store.Backend
 }
 
+var lastMaintenance time.Time
+
 // Run blocks until ctx is cancelled.
 func Run(ctx context.Context, deps *Deps) {
 	cfg := config.Load()
@@ -24,7 +26,7 @@ func Run(ctx context.Context, deps *Deps) {
 	if deps == nil || deps.Store == nil {
 		log.Fatalf("evobgp-ingest: missing store (pass ingest.Deps from BootstrapWorkers or evobgp-all)")
 	}
-	hc := &http.Client{Timeout: 45 * time.Second}
+	hc := httpclient.New(httpclient.DefaultTimeout)
 	t := time.NewTicker(60 * time.Second)
 	defer t.Stop()
 	log.Printf("evobgp-ingest: active (CDN conditional GET / ETag prefetch)")
@@ -34,7 +36,14 @@ func Run(ctx context.Context, deps *Deps) {
 			log.Printf("evobgp-ingest: stopped")
 			return
 		case <-t.C:
-			if err := pipeline.PrefetchCDNSourceETags(context.Background(), deps.Store, hc); err != nil {
+			if deps.Store != nil && time.Since(lastMaintenance) > time.Hour {
+				deps.Store.RunPeriodicMaintenance(ctx)
+				lastMaintenance = time.Now()
+			}
+			prefetchCtx, cancel := context.WithTimeout(ctx, 50*time.Second)
+			err := pipeline.PrefetchCDNSourceETags(prefetchCtx, deps.Store, hc)
+			cancel()
+			if err != nil {
 				log.Printf("evobgp-ingest: prefetch: %v", err)
 			}
 		}

internal/jobs/job.go

@@ -182,6 +182,8 @@ type Registry struct {
 	workerStart func(j *Job)
 	workerSem   chan struct{}
 	onTerminal  func(j *Job)
+	onEnqueued  func(j *Job)
+	onRunning   func(j *Job)
 }
 
 type idempoKey struct {
@@ -209,6 +211,44 @@ func (r *Registry) SetTerminalHook(fn func(j *Job)) {
 	r.onTerminal = fn
 }
 
+// SetPersistHooks registers best-effort callbacks for job lifecycle persistence.
+func (r *Registry) SetPersistHooks(onEnqueued, onRunning, onTerminal func(j *Job)) {
+	if r == nil {
+		return
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.onEnqueued = onEnqueued
+	r.onRunning = onRunning
+	if onTerminal != nil {
+		r.onTerminal = onTerminal
+	}
+}
+
+func (r *Registry) fireEnqueued(j *Job) {
+	if r == nil || j == nil {
+		return
+	}
+	r.mu.RLock()
+	fn := r.onEnqueued
+	r.mu.RUnlock()
+	if fn != nil {
+		fn(j)
+	}
+}
+
+func (r *Registry) fireRunning(j *Job) {
+	if r == nil || j == nil {
+		return
+	}
+	r.mu.RLock()
+	fn := r.onRunning
+	r.mu.RUnlock()
+	if fn != nil {
+		fn(j)
+	}
+}
+
 func (r *Registry) fireTerminal(j *Job) {
 	if r == nil || j == nil {
 		return
@@ -271,8 +311,6 @@ func (r *Registry) pruneTerminalIfOver(maxJobs int) {
 // Enqueue creates a job or returns an existing one for the same idempotency key.
 func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, moduleID *string, meta map[string]any) (*Job, bool, error) {
 	r.mu.Lock()
-	defer r.mu.Unlock()
-
 	maxJobs := registryMaxJobsFromEnv()
 	r.pruneTerminalIfOver(maxJobs)
 
@@ -281,6 +319,7 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module
 		if existing, ok := r.byIdempo[k]; ok {
 			st := existing.statusLocked()
 			if st == StatusQueued || st == StatusRunning {
+				r.mu.Unlock()
 				return existing, false, nil
 			}
 			delete(r.byIdempo, k)
@@ -302,8 +341,14 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module
 	}
 	r.byID[j.ID] = j
 	r.pruneTerminalIfOver(maxJobs)
+	enqueuedHook := r.onEnqueued
+	workerStart := r.workerStart
+	r.mu.Unlock()
 
-	if r.workerStart != nil {
+	if enqueuedHook != nil {
+		enqueuedHook(j)
+	}
+	if workerStart != nil {
 		go func() {
 			r.workerSem <- struct{}{}
 			active := len(r.workerSem)
@@ -313,7 +358,7 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module
 				<-r.workerSem
 				observability.RecordJobQueueDepth(len(r.workerSem), capacity)
 			}()
-			r.workerStart(j)
+			workerStart(j)
 		}()
 	}
 	return j, true, nil

internal/jobs/worker.go

@@ -13,6 +13,7 @@ import (
 
 	"evobgp/internal/birddeploy"
 	"evobgp/internal/birdfmt"
+	"evobgp/internal/httpclient"
 	"evobgp/internal/nodedispatch"
 	"evobgp/internal/observability"
 	"evobgp/internal/pipeline"
@@ -70,7 +71,7 @@ type revisionLogEntry struct {
 	Message        string   `json:"message"`
 }
 
-var defaultWorkerHTTP = &http.Client{Timeout: 45 * time.Second}
+var defaultWorkerHTTP = httpclient.New(httpclient.DefaultTimeout)
 
 func (w *Worker) httpClient() *http.Client {
 	if w != nil && w.HTTPClient != nil {
@@ -94,6 +95,9 @@ func (w *Worker) Process(j *Job) {
 		return
 	}
 	j.MarkRunning()
+	if w != nil && w.Registry != nil {
+		w.Registry.fireRunning(j)
+	}
 	if j.IsCancelRequested() {
 		j.MarkCancelled()
 		return

internal/nodecli/commands.go

@@ -16,6 +16,7 @@ import (
 
 	"evobgp/internal/birdfmt"
 	"evobgp/internal/bundle"
+	"evobgp/internal/httpclient"
 	"evobgp/internal/signing"
 )
 
@@ -55,6 +56,10 @@ func CmdPullBundle(args []string) int {
 	return 0
 }
 
+func nodeHTTPClient() *http.Client {
+	return httpclient.New(60 * time.Second)
+}
+
 func fetchLatestRevision(base, token, speaker string) (string, error) {
 	u := strings.TrimRight(base, "/") + "/v1/speakers/" + speaker + "/revisions/latest"
 	req, err := http.NewRequest(http.MethodGet, u, nil)
@@ -62,7 +67,9 @@ func fetchLatestRevision(base, token, speaker string) (string, error) {
 		return "", err
 	}
 	req.Header.Set("Authorization", "Bearer "+token)
-	resp, err := http.DefaultClient.Do(req)
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	resp, err := httpclient.DoWithRetry(ctx, nodeHTTPClient(), req, 3)
 	if err != nil {
 		return "", err
 	}
@@ -90,7 +97,9 @@ func fetchBundle(base, token, speaker, revision string) ([]byte, error) {
 		return nil, err
 	}
 	req.Header.Set("Authorization", "Bearer "+token)
-	resp, err := http.DefaultClient.Do(req)
+	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
+	defer cancel()
+	resp, err := httpclient.DoWithRetry(ctx, nodeHTTPClient(), req, 3)
 	if err != nil {
 		return nil, err
 	}

internal/pipeline/cdn_prefetch_test.go

@@ -11,6 +11,7 @@ import (
 )
 
 func TestCollectModulePrefixRows_CDNSendsIfNoneMatch(t *testing.T) {
+	t.Setenv("EVOBGP_CDN_ALLOW_PRIVATE", "1")
 	m := store.NewMemory()
 	m.SeedDemo()
 	tenant, _, _, _, _ := m.DemoIDs()
@@ -25,7 +26,7 @@ func TestCollectModulePrefixRows_CDNSendsIfNoneMatch(t *testing.T) {
 	}
 
 	var gotIfNoneMatch string
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		gotIfNoneMatch = strings.TrimSpace(r.Header.Get("If-None-Match"))
 		w.Header().Set("ETag", "etag-new")
 		_, _ = w.Write([]byte("198.51.100.0/24\n"))
@@ -54,6 +55,7 @@ func TestCollectModulePrefixRows_CDNSendsIfNoneMatch(t *testing.T) {
 }
 
 func TestCollectModulePrefixRows_CDN304UsesSnapshot(t *testing.T) {
+	t.Setenv("EVOBGP_CDN_ALLOW_PRIVATE", "1")
 	m := store.NewMemory()
 	m.SeedDemo()
 	tenant, _, _, _, _ := m.DemoIDs()
@@ -67,7 +69,7 @@ func TestCollectModulePrefixRows_CDN304UsesSnapshot(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusNotModified)
 	}))
 	defer srv.Close()
@@ -99,6 +101,7 @@ func TestCollectModulePrefixRows_CDN304UsesSnapshot(t *testing.T) {
 }
 
 func TestRefreshModuleIngest_CDN304UsesStoredSnapshot(t *testing.T) {
+	t.Setenv("EVOBGP_CDN_ALLOW_PRIVATE", "1")
 	m := store.NewMemory()
 	m.SeedDemo()
 	tenant, _, _, _, _ := m.DemoIDs()
@@ -112,7 +115,7 @@ func TestRefreshModuleIngest_CDN304UsesStoredSnapshot(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusNotModified)
 	}))
 	defer srv.Close()
@@ -140,6 +143,7 @@ func TestRefreshModuleIngest_CDN304UsesStoredSnapshot(t *testing.T) {
 }
 
 func TestCollectModulePrefixRows_CDN304RetriesWithoutETag(t *testing.T) {
+	t.Setenv("EVOBGP_CDN_ALLOW_PRIVATE", "1")
 	m := store.NewMemory()
 	m.SeedDemo()
 	tenant, _, _, _, _ := m.DemoIDs()
@@ -154,7 +158,7 @@ func TestCollectModulePrefixRows_CDN304RetriesWithoutETag(t *testing.T) {
 	}
 
 	var calls int
-	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		calls++
 		if calls == 1 {
 			if got := strings.TrimSpace(r.Header.Get("If-None-Match")); got != "etag-stable" {

internal/pipeline/cdn_snapshot.go

@@ -119,6 +119,12 @@ func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Cl
 	if u == "" {
 		return nil, nil
 	}
+	if _, err := ValidateCDNURL(u); err != nil {
+		return nil, err
+	}
+	if err := ResolveCDNURLHost(ctx, u); err != nil {
+		return nil, err
+	}
 	sourceKey := cdnSourceKey(src.ID)
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
 	if err != nil {
@@ -127,7 +133,7 @@ func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Cl
 	if etag := strings.TrimSpace(src.Etag); etag != "" {
 		req.Header.Set("If-None-Match", etag)
 	}
-	resp, err := hc.Do(req)
+	resp, err := upstreamHTTPDo(ctx, hc, req)
 	if err != nil {
 		return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
 	}
@@ -143,7 +149,7 @@ func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Cl
 		if err != nil {
 			return nil, err
 		}
-		resp, err = hc.Do(req2)
+		resp, err = upstreamHTTPDo(ctx, hc, req2)
 		if err != nil {
 			return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
 		}
@@ -189,6 +195,12 @@ func fetchCDNSourceRows(ctx context.Context, st store.Backend, hc *http.Client,
 	if u == "" {
 		return nil, nil
 	}
+	if _, err := ValidateCDNURL(u); err != nil {
+		return nil, err
+	}
+	if err := ResolveCDNURLHost(ctx, u); err != nil {
+		return nil, err
+	}
 	sourceKey := cdnSourceKey(src.ID)
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
 	if err != nil {
@@ -197,7 +209,7 @@ func fetchCDNSourceRows(ctx context.Context, st store.Backend, hc *http.Client,
 	if etag := strings.TrimSpace(src.Etag); etag != "" {
 		req.Header.Set("If-None-Match", etag)
 	}
-	resp, err := hc.Do(req)
+	resp, err := upstreamHTTPDo(ctx, hc, req)
 	if err != nil {
 		return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
 	}
@@ -212,7 +224,7 @@ func fetchCDNSourceRows(ctx context.Context, st store.Backend, hc *http.Client,
 		if err != nil {
 			return nil, err
 		}
-		resp, err = hc.Do(req2)
+		resp, err = upstreamHTTPDo(ctx, hc, req2)
 		if err != nil {
 			return nil, fmt.Errorf("cdn fetch %s: %w", u, err)
 		}

internal/pipeline/cdn_url.go

@@ -0,0 +1,116 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"net/netip"
+	"net/url"
+	"os"
+	"strings"
+	"time"
+)
+
+func isBlockedCDNIP(ip netip.Addr) bool {
+	if allowPrivateCDNURLs() {
+		return false
+	}
+	if !ip.IsValid() {
+		return true
+	}
+	return ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsMulticast() ||
+		ip.IsUnspecified() || ip == netip.MustParseAddr("169.254.169.254")
+}
+
+func allowPrivateCDNURLs() bool {
+	v := strings.TrimSpace(os.Getenv("EVOBGP_CDN_ALLOW_PRIVATE"))
+	return v == "1" || strings.EqualFold(v, "true")
+}
+
+func isBlockedCDNHostname(host string) bool {
+	if allowPrivateCDNURLs() {
+		return false
+	}
+	h := strings.ToLower(strings.TrimSpace(host))
+	if h == "" || h == "localhost" {
+		return true
+	}
+	if strings.HasSuffix(h, ".local") || strings.HasSuffix(h, ".internal") || strings.HasSuffix(h, ".localhost") {
+		return true
+	}
+	return false
+}
+
+// ValidateCDNURL checks CDN source URLs for SSRF-safe HTTPS endpoints (hostname only; no DNS resolve).
+func ValidateCDNURL(raw string) (string, error) {
+	raw = strings.TrimSpace(raw)
+	if raw == "" {
+		return "", fmt.Errorf("pipeline: cdn url is required")
+	}
+	u, err := url.Parse(raw)
+	if err != nil {
+		return "", fmt.Errorf("pipeline: cdn url invalid: %w", err)
+	}
+	if u.Scheme != "https" {
+		return "", fmt.Errorf("pipeline: cdn url must use https")
+	}
+	if u.User != nil {
+		return "", fmt.Errorf("pipeline: cdn url must not include credentials")
+	}
+	host := strings.TrimSpace(u.Hostname())
+	if host == "" {
+		return "", fmt.Errorf("pipeline: cdn url missing host")
+	}
+	if isBlockedCDNHostname(host) {
+		return "", fmt.Errorf("pipeline: cdn url blocked host")
+	}
+	if ip, err := netip.ParseAddr(host); err == nil {
+		if isBlockedCDNIP(ip) {
+			return "", fmt.Errorf("pipeline: cdn url blocked host")
+		}
+	}
+	return u.String(), nil
+}
+
+// ResolveCDNURLHost resolves a CDN hostname and rejects private/link-local targets (SSRF at fetch time).
+func ResolveCDNURLHost(ctx context.Context, raw string) error {
+	u, err := url.Parse(strings.TrimSpace(raw))
+	if err != nil {
+		return err
+	}
+	host := strings.TrimSpace(u.Hostname())
+	if host == "" {
+		return fmt.Errorf("pipeline: cdn url missing host")
+	}
+	if ip, err := netip.ParseAddr(host); err == nil {
+		if isBlockedCDNIP(ip) {
+			return fmt.Errorf("pipeline: cdn url blocked host")
+		}
+		return nil
+	}
+	if isBlockedCDNHostname(host) {
+		return fmt.Errorf("pipeline: cdn url blocked host")
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	resolveCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	ips, err := net.DefaultResolver.LookupIP(resolveCtx, "ip", host)
+	if err != nil {
+		return fmt.Errorf("pipeline: cdn url dns lookup: %w", err)
+	}
+	if len(ips) == 0 {
+		return fmt.Errorf("pipeline: cdn url dns lookup: no addresses")
+	}
+	for _, ip := range ips {
+		addr, ok := netip.AddrFromSlice(ip)
+		if !ok {
+			continue
+		}
+		if isBlockedCDNIP(addr) {
+			return fmt.Errorf("pipeline: cdn url resolves to blocked address")
+		}
+	}
+	return nil
+}

internal/pipeline/cdn_url_test.go

@@ -0,0 +1,34 @@
+package pipeline
+
+import "testing"
+
+func TestValidateCDNURL(t *testing.T) {
+	tests := []struct {
+		raw  string
+		ok   bool
+		want string
+	}{
+		{"https://cdn.example.com/prefixes.txt", true, "https://cdn.example.com/prefixes.txt"},
+		{"http://cdn.example.com/x", false, ""},
+		{"https://127.0.0.1/x", false, ""},
+		{"https://10.0.0.1/x", false, ""},
+		{"https://169.254.169.254/latest/meta-data", false, ""},
+		{"https://localhost/x", false, ""},
+		{"file:///etc/passwd", false, ""},
+		{"https://user:pass@cdn.example.com/x", false, ""},
+	}
+	for _, tc := range tests {
+		got, err := ValidateCDNURL(tc.raw)
+		if tc.ok && err != nil {
+			t.Errorf("%q: unexpected err %v", tc.raw, err)
+			continue
+		}
+		if !tc.ok && err == nil {
+			t.Errorf("%q: expected error", tc.raw)
+			continue
+		}
+		if tc.ok && got != tc.want {
+			t.Errorf("%q: got %q want %q", tc.raw, got, tc.want)
+		}
+	}
+}

internal/pipeline/collect_parallel.go

@@ -25,7 +25,7 @@ func prefixRowsForSource(rows []store.PrefixRow, sourceKey string) []store.Prefi
 	return out
 }
 
-func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module, list []*store.ASEntry) ([]store.PrefixRow, error) {
+func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, mod *store.Module, list []*store.ASEntry, priorSnapshot []store.PrefixRow) ([]store.PrefixRow, error) {
 	moduleID := mod.ID
 	legacy := strings.TrimSpace(os.Getenv("EVOBGP_ASN_RESOLVE")) == "0"
 	if legacy {
@@ -76,6 +76,41 @@ func collectASPrefixRows(ctx context.Context, st store.Backend, hc *http.Client,
 			}
 			pfxs, holder, err := resolveASNForEntry(ctx, st, hc, entry.ASN)
 			if err != nil {
+				if staleOnUpstreamError() {
+					if staleRows, staleHolder, ok := staleASNPrefixes(st, priorSnapshot, entry.ASN); ok {
+						logStaleUpstream("asn", fmt.Sprintf("AS%d: %v", entry.ASN, err))
+						src := fmt.Sprintf("as:%d", entry.ASN)
+						rows := append([]store.PrefixRow(nil), staleRows...)
+						for i := range rows {
+							rows[i].CommunityID = comm
+							rows[i].Source = src
+						}
+						results[idx] = entryResult{
+							rows:   rows,
+							metaID: entry.ID,
+							asn:    entry.ASN,
+							holder: staleHolder,
+							count:  int64(len(rows)),
+						}
+						return
+					}
+					if pfxs2, holder2, ok := asnCacheExpired(st, entry.ASN); ok {
+						logStaleUpstream("asn", fmt.Sprintf("AS%d expired cache: %v", entry.ASN, err))
+						src := fmt.Sprintf("as:%d", entry.ASN)
+						var rows []store.PrefixRow
+						for _, pfx := range pfxs2 {
+							rows = append(rows, store.PrefixRow{Prefix: pfx.String(), CommunityID: comm, Source: src})
+						}
+						results[idx] = entryResult{
+							rows:   rows,
+							metaID: entry.ID,
+							asn:    entry.ASN,
+							holder: holder2,
+							count:  int64(len(pfxs2)),
+						}
+						return
+					}
+				}
 				results[idx] = entryResult{err: fmt.Errorf("resolve AS%d: %w", entry.ASN, err)}
 				return
 			}
@@ -167,6 +202,13 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client
 			}
 			rows, err := fetchCDNSourceRows(ctx, st, hc, tenantID, moduleID, mod, src, priorSnapshot, now)
 			if err != nil {
+				if staleOnUpstreamError() {
+					if cached, ok := staleCDNPrefixes(st, tenantID, moduleID, priorSnapshot, src.ID); ok {
+						logStaleUpstream("cdn", fmt.Sprintf("source %s: %v", src.ID, err))
+						results[idx] = srcResult{rows: cached}
+						return
+					}
+				}
 				results[idx] = srcResult{err: err}
 				return
 			}
@@ -190,7 +232,7 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client
 	return out, nil
 }
 
-func collectDomainPrefixRows(ctx context.Context, hc *http.Client, mod *store.Module, profiles []*store.DohProfile, policy string, entries []*store.DomainEntry) ([]store.PrefixRow, error) {
+func collectDomainPrefixRows(ctx context.Context, hc *http.Client, mod *store.Module, profiles []*store.DohProfile, policy string, entries []*store.DomainEntry, priorSnapshot []store.PrefixRow) ([]store.PrefixRow, error) {
 	var validDom []*store.DomainEntry
 	for _, e := range entries {
 		if e != nil {
@@ -219,6 +261,19 @@ func collectDomainPrefixRows(ctx context.Context, hc *http.Client, mod *store.Mo
 			}
 			addrs, err := resolveDomainIPsWithPolicy(ctx, hc, profiles, policy, entry.FQDN)
 			if err != nil {
+				if staleOnUpstreamError() {
+					if cached, ok := staleDomainPrefixes(priorSnapshot, entry.FQDN); ok {
+						logStaleUpstream("domain", fmt.Sprintf("%q: %v", entry.FQDN, err))
+						rows := append([]store.PrefixRow(nil), cached...)
+						for i := range rows {
+							if rows[i].CommunityID == nil {
+								rows[i].CommunityID = comm
+							}
+						}
+						results[idx] = domResult{rows: rows}
+						return
+					}
+				}
 				results[idx] = domResult{err: fmt.Errorf("resolve domain %q: %w", entry.FQDN, err)}
 				return
 			}

internal/pipeline/collect_stale.go

@@ -0,0 +1,86 @@
+package pipeline
+
+import (
+	"fmt"
+	"log"
+	"net/netip"
+	"os"
+	"strings"
+
+	"evobgp/internal/store"
+)
+
+// staleOnUpstreamError reports whether ingest should keep last-known prefixes when an upstream fetch fails.
+// Enabled by default; set EVOBGP_STALE_ON_UPSTREAM_ERROR=0 to restore fail-fast behavior.
+func staleOnUpstreamError() bool {
+	v := strings.TrimSpace(os.Getenv("EVOBGP_STALE_ON_UPSTREAM_ERROR"))
+	if v == "" || v == "1" || strings.EqualFold(v, "true") {
+		return true
+	}
+	return false
+}
+
+func logStaleUpstream(kind, detail string) {
+	log.Printf("pipeline: stale upstream fallback (%s): %s", kind, detail)
+}
+
+func staleASNPrefixes(st store.Backend, priorSnapshot []store.PrefixRow, asn int64) ([]store.PrefixRow, string, bool) {
+	sourceKey := fmt.Sprintf("as:%d", asn)
+	if cached := prefixRowsForSource(priorSnapshot, sourceKey); len(cached) > 0 {
+		return cached, "", true
+	}
+	if st == nil {
+		return nil, "", false
+	}
+	ent, ok, err := st.GetASNPrefixCache(asn)
+	if err != nil || !ok || ent == nil || len(ent.Prefixes) == 0 {
+		return nil, "", false
+	}
+	var rows []store.PrefixRow
+	for _, p := range ent.Prefixes {
+		pfx, perr := netip.ParsePrefix(strings.TrimSpace(p))
+		if perr != nil {
+			continue
+		}
+		rows = append(rows, store.PrefixRow{Prefix: pfx.Masked().String(), Source: sourceKey})
+	}
+	if len(rows) == 0 {
+		return nil, "", false
+	}
+	return rows, ent.Holder, true
+}
+
+func staleDomainPrefixes(priorSnapshot []store.PrefixRow, fqdn string) ([]store.PrefixRow, bool) {
+	sourceKey := "domain:" + strings.TrimSpace(fqdn)
+	cached := prefixRowsForSource(priorSnapshot, sourceKey)
+	return cached, len(cached) > 0
+}
+
+func staleCDNPrefixes(st store.Backend, tenantID, moduleID string, priorSnapshot []store.PrefixRow, sourceID string) ([]store.PrefixRow, bool) {
+	sourceKey := cdnSourceKey(sourceID)
+	cached := cachedCDNPrefixRows(st, tenantID, moduleID, priorSnapshot, sourceKey)
+	return cached, len(cached) > 0
+}
+
+// asnCacheExpired returns cached ASN prefixes even past TTL (for stale fallback only).
+func asnCacheExpired(st store.Backend, asn int64) ([]netip.Prefix, string, bool) {
+	if st == nil {
+		return nil, "", false
+	}
+	ent, ok, err := st.GetASNPrefixCache(asn)
+	if err != nil || !ok || ent == nil || len(ent.Prefixes) == 0 {
+		return nil, "", false
+	}
+	out := make([]netip.Prefix, 0, len(ent.Prefixes))
+	for _, p := range ent.Prefixes {
+		pfx, perr := netip.ParsePrefix(strings.TrimSpace(p))
+		if perr != nil {
+			continue
+		}
+		out = append(out, pfx.Masked())
+	}
+	if len(out) == 0 {
+		return nil, "", false
+	}
+	return out, ent.Holder, true
+}

internal/pipeline/collect_stale_test.go

@@ -0,0 +1,72 @@
+package pipeline
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"evobgp/internal/store"
+)
+
+func TestCollectCDNPrefixRows_StaleOnFetchError(t *testing.T) {
+	t.Setenv("EVOBGP_STALE_ON_UPSTREAM_ERROR", "1")
+	t.Setenv("EVOBGP_CDN_ALLOW_PRIVATE", "1")
+
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "upstream down", http.StatusServiceUnavailable)
+	}))
+	defer srv.Close()
+
+	st := store.NewMemory()
+	st.SeedDemo()
+	tenant, _, _, _, _ := st.DemoIDs()
+	mod, err := st.CreateModule(tenant, &store.Module{Type: "CDN_CIDRS", Name: "cdn", Enabled: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.CreateCDNSource(tenant, mod.ID, &store.CDNSource{
+		ID: "s1", URL: srv.URL, SourceKind: "plain",
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	prior := []store.PrefixRow{
+		{Prefix: "203.0.113.0/24", Source: "cdn:s1"},
+	}
+	rows, err := collectCDNPrefixRows(context.Background(), st, srv.Client(), tenant, mod, []*store.CDNSource{{ID: "s1", URL: srv.URL, SourceKind: "plain"}}, prior)
+	if err != nil {
+		t.Fatalf("expected stale fallback, got err: %v", err)
+	}
+	if len(rows) != 1 || rows[0].Prefix != "203.0.113.0/24" {
+		t.Fatalf("unexpected rows: %+v", rows)
+	}
+}
+
+func TestCollectCDNPrefixRows_FailFastWhenNoStale(t *testing.T) {
+	t.Setenv("EVOBGP_STALE_ON_UPSTREAM_ERROR", "0")
+	t.Setenv("EVOBGP_CDN_ALLOW_PRIVATE", "1")
+
+	srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "upstream down", http.StatusServiceUnavailable)
+	}))
+	defer srv.Close()
+
+	st := store.NewMemory()
+	st.SeedDemo()
+	tenant, _, _, _, _ := st.DemoIDs()
+	mod, err := st.CreateModule(tenant, &store.Module{Type: "CDN_CIDRS", Name: "cdn", Enabled: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := st.CreateCDNSource(tenant, mod.ID, &store.CDNSource{
+		ID: "s1", URL: srv.URL, SourceKind: "plain",
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = collectCDNPrefixRows(context.Background(), st, srv.Client(), tenant, mod, []*store.CDNSource{{ID: "s1", URL: srv.URL, SourceKind: "plain"}}, nil)
+	if err == nil {
+		t.Fatal("expected error when stale disabled and no cache")
+	}
+}

internal/pipeline/prefetch.go

@@ -5,86 +5,116 @@ import (
 	"io"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
+	"evobgp/internal/httpclient"
 	"evobgp/internal/store"
 )
 
+type prefetchTask struct {
+	tenantID string
+	mod      *store.Module
+	src      *store.CDNSource
+}
+
 // PrefetchCDNSourceETags performs conditional GETs for CDN sources; on 200 parses CIDRs into module_prefix_snapshot.
 func PrefetchCDNSourceETags(ctx context.Context, st store.Backend, hc *http.Client) error {
 	if hc == nil {
-		hc = http.DefaultClient
+		hc = httpclient.New(httpclient.DefaultTimeout)
+	}
+	if ctx == nil {
+		ctx = context.Background()
 	}
 	tenants, err := st.ListTenantIDs()
 	if err != nil {
 		return err
 	}
-	now := time.Now().UTC()
+	var tasks []prefetchTask
 	for _, tid := range tenants {
 		for _, mod := range st.ListModules(tid) {
 			if mod == nil || !mod.Enabled || mod.Type != "CDN_CIDRS" {
 				continue
 			}
-			omod, err := st.GetModule(tid, mod.ID)
-			if err != nil {
-				continue
-			}
 			sources, err := st.ListCDNSources(tid, mod.ID)
 			if err != nil {
 				continue
 			}
-			var prior []store.PrefixRow
-			if snap, ok, _ := st.GetModulePrefixSnapshot(tid, mod.ID); ok && snap != nil {
-				prior = snap.Prefixes
-			}
 			for _, src := range sources {
-				if src == nil || strings.TrimSpace(src.URL) == "" {
-					continue
-				}
-				req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimSpace(src.URL), nil)
-				if err != nil {
-					continue
-				}
-				if etag := strings.TrimSpace(src.Etag); etag != "" {
-					req.Header.Set("If-None-Match", etag)
-				}
-				resp, err := hc.Do(req)
-				if err != nil {
-					continue
-				}
-				if resp.StatusCode == http.StatusNotModified {
-					_ = resp.Body.Close()
-					continue
-				}
-				if resp.StatusCode != http.StatusOK {
-					_, _ = io.Copy(io.Discard, resp.Body)
-					_ = resp.Body.Close()
-					continue
-				}
-				body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
-				_ = resp.Body.Close()
-				if err != nil {
-					continue
-				}
-				prefixStrs, err := parseCDNBody(string(body), src)
-				if err != nil {
-					continue
-				}
-				newEtag := strings.TrimSpace(resp.Header.Get("ETag"))
-				patch := &store.CDNSourcePatch{LastRefreshedAt: &now}
-				if newEtag != "" && newEtag != strings.TrimSpace(src.Etag) {
-					e := newEtag
-					patch.Etag = &e
-				}
-				_, _ = st.UpdateCDNSource(tid, mod.ID, src.ID, patch)
-				rows := cdnRowsFromParsed(omod, src, prefixStrs)
-				_ = mergeCDNSourceIntoModuleSnapshot(st, tid, omod, src.ID, rows)
-				_ = prior // prior may be stale after merge; refresh for next source in loop
-				if snap, ok, _ := st.GetModulePrefixSnapshot(tid, mod.ID); ok && snap != nil {
-					prior = snap.Prefixes
+				if src != nil && strings.TrimSpace(src.URL) != "" {
+					tasks = append(tasks, prefetchTask{tenantID: tid, mod: mod, src: src})
 				}
 			}
 		}
 	}
+	if len(tasks) == 0 {
+		return nil
+	}
+	sem := make(chan struct{}, collectConcurrency())
+	var wg sync.WaitGroup
+	for _, task := range tasks {
+		wg.Add(1)
+		go func(t prefetchTask) {
+			defer wg.Done()
+			sem <- struct{}{}
+			defer func() { <-sem }()
+			prefetchOneCDNSource(ctx, st, hc, t)
+		}(task)
+	}
+	wg.Wait()
 	return nil
 }
+
+func prefetchOneCDNSource(ctx context.Context, st store.Backend, hc *http.Client, t prefetchTask) {
+	now := time.Now().UTC()
+	tid, mod, src := t.tenantID, t.mod, t.src
+	u := strings.TrimSpace(src.URL)
+	if _, err := ValidateCDNURL(u); err != nil {
+		return
+	}
+	if err := ResolveCDNURLHost(ctx, u); err != nil {
+		return
+	}
+	omod, err := st.GetModule(tid, mod.ID)
+	if err != nil {
+		return
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
+	if err != nil {
+		return
+	}
+	if etag := strings.TrimSpace(src.Etag); etag != "" {
+		req.Header.Set("If-None-Match", etag)
+	}
+	resp, err := upstreamHTTPDo(ctx, hc, req)
+	if err != nil {
+		return
+	}
+	if resp.StatusCode == http.StatusNotModified {
+		_ = resp.Body.Close()
+		return
+	}
+	if resp.StatusCode != http.StatusOK {
+		_, _ = io.Copy(io.Discard, resp.Body)
+		_ = resp.Body.Close()
+		return
+	}
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
+	_ = resp.Body.Close()
+	if err != nil {
+		return
+	}
+	prefixStrs, err := parseCDNBody(string(body), src)
+	if err != nil {
+		return
+	}
+	newEtag := strings.TrimSpace(resp.Header.Get("ETag"))
+	patch := &store.CDNSourcePatch{LastRefreshedAt: &now}
+	if newEtag != "" && newEtag != strings.TrimSpace(src.Etag) {
+		e := newEtag
+		patch.Etag = &e
+	}
+	_, _ = st.UpdateCDNSource(tid, mod.ID, src.ID, patch)
+	rows := cdnRowsFromParsed(omod, src, prefixStrs)
+	_ = mergeCDNSourceIntoModuleSnapshot(st, tid, omod, src.ID, rows)
+}

internal/pipeline/refresh.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"evobgp/internal/birdfmt"
+	"evobgp/internal/httpclient"
 	"evobgp/internal/observability"
 	"evobgp/internal/store"
 
@@ -50,7 +51,7 @@ func MaterializedASPrefixKey(asn int64) string {
 // It does not create a new config revision.
 func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client, tenantID, moduleID string) error {
 	if hc == nil {
-		hc = http.DefaultClient
+		hc = httpclient.New(httpclient.DefaultTimeout)
 	}
 	start := time.Now()
 	mod, err := st.GetModule(tenantID, moduleID)
@@ -84,7 +85,7 @@ func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client,
 // If materialized prefixes are unchanged, returns latest revision id without creating a duplicate.
 func RenderTenantRevision(ctx context.Context, st store.Backend, hc *http.Client, tenantID, triggerModuleID string) (revisionID string, err error) {
 	if hc == nil {
-		hc = http.DefaultClient
+		hc = httpclient.New(httpclient.DefaultTimeout)
 	}
 	agg, err := aggregateTenantPrefixRowsAll(ctx, st, hc, tenantID)
 	if err != nil {
@@ -117,7 +118,7 @@ func RenderTenantRevision(ctx context.Context, st store.Backend, hc *http.Client
 func RenderTenantRevisionFromPrefixes(ctx context.Context, st store.Backend, hc *http.Client, tenantID, triggerModuleID string, rows []store.PrefixRow) (revisionID string, err error) {
 	_ = ctx
 	if hc == nil {
-		hc = http.DefaultClient
+		hc = httpclient.New(httpclient.DefaultTimeout)
 	}
 	agg := append([]store.PrefixRow(nil), rows...)
 	rawCount := len(agg)
@@ -176,7 +177,7 @@ func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Cli
 			return nil, err
 		}
 		sort.Slice(list, func(i, j int) bool { return list[i].ASN < list[j].ASN })
-		return collectASPrefixRows(ctx, st, hc, tenantID, mod, list)
+		return collectASPrefixRows(ctx, st, hc, tenantID, mod, list, priorSnapshot)
 	case "CDN_CIDRS":
 		sources, err := st.ListCDNSources(tenantID, moduleID)
 		if err != nil {
@@ -192,7 +193,7 @@ func collectModulePrefixRows(ctx context.Context, st store.Backend, hc *http.Cli
 		if err != nil {
 			return nil, err
 		}
-		return collectDomainPrefixRows(ctx, hc, mod, profiles, policy, entries)
+		return collectDomainPrefixRows(ctx, hc, mod, profiles, policy, entries, priorSnapshot)
 	default:
 		return nil, fmt.Errorf("pipeline: unknown module type %q", mod.Type)
 	}

internal/pipeline/tenant_refresh.go

@@ -7,13 +7,14 @@ import (
 	"strings"
 	"sync"
 
+	"evobgp/internal/httpclient"
 	"evobgp/internal/store"
 )
 
 // RefreshTenantModules ingests all listed modules in parallel and updates per-module snapshots.
 func RefreshTenantModules(ctx context.Context, st store.Backend, hc *http.Client, tenantID string, moduleIDs []string) error {
 	if hc == nil {
-		hc = http.DefaultClient
+		hc = httpclient.New(httpclient.DefaultTimeout)
 	}
 	var ids []string
 	seen := make(map[string]struct{})

internal/pipeline/upstream_http.go

@@ -0,0 +1,23 @@
+package pipeline
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"evobgp/internal/httpclient"
+)
+
+func upstreamHTTPDo(ctx context.Context, hc *http.Client, req *http.Request) (*http.Response, error) {
+	if hc == nil {
+		hc = httpclient.New(httpclient.DefaultTimeout)
+	}
+	resp, err := httpclient.DoWithBreaker(ctx, hc, req, 3)
+	if err != nil {
+		if req.URL != nil {
+			return nil, fmt.Errorf("cdn fetch %s: %w", req.URL.String(), err)
+		}
+		return nil, err
+	}
+	return resp, nil
+}

internal/repository/asn_cache.go

@@ -11,14 +11,22 @@ import (
 	"github.com/jackc/pgx/v5"
 )
 
+func asnCacheRowTableExists(ctx context.Context, q queryRower) bool {
+	var n int
+	err := q.QueryRow(ctx, `
+		SELECT 1 FROM information_schema.tables
+		WHERE table_schema = 'public' AND table_name = 'asn_prefix_cache_row'
+		LIMIT 1`).Scan(&n)
+	return err == nil
+}
+
 func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, bool, error) {
 	ctx := context.Background()
 	var holder string
 	var fetchedAt time.Time
-	var raw []byte
 	err := p.pool.QueryRow(ctx, `
-		SELECT holder, fetched_at, prefixes_json FROM asn_prefix_cache WHERE asn = $1`, asn).
-		Scan(&holder, &fetchedAt, &raw)
+		SELECT holder, fetched_at FROM asn_prefix_cache WHERE asn = $1`, asn).
+		Scan(&holder, &fetchedAt)
 	if err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
 			return nil, false, nil
@@ -26,8 +34,25 @@ func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, boo
 		return nil, false, err
 	}
 	var prefixes []string
-	if len(raw) > 0 {
-		_ = json.Unmarshal(raw, &prefixes)
+	if asnCacheRowTableExists(ctx, p.pool) {
+		rows, qerr := p.pool.Query(ctx, `
+			SELECT prefix::text FROM asn_prefix_cache_row WHERE asn = $1 ORDER BY prefix`, asn)
+		if qerr != nil {
+			return nil, false, qerr
+		}
+		defer rows.Close()
+		for rows.Next() {
+			var s string
+			if err := rows.Scan(&s); err != nil {
+				continue
+			}
+			prefixes = append(prefixes, s)
+		}
+	} else {
+		var raw []byte
+		if err := p.pool.QueryRow(ctx, `SELECT prefixes_json FROM asn_prefix_cache WHERE asn = $1`, asn).Scan(&raw); err == nil && len(raw) > 0 {
+			_ = json.Unmarshal(raw, &prefixes)
+		}
 	}
 	return &store.ASNPrefixCacheEntry{
 		ASN:       asn,
@@ -38,18 +63,40 @@ func (p *Postgres) GetASNPrefixCache(asn int64) (*store.ASNPrefixCacheEntry, boo
 }
 
 func (p *Postgres) SetASNPrefixCache(asn int64, holder string, prefixes []string) error {
-	raw, err := json.Marshal(prefixes)
+	ctx := context.Background()
+	tx, err := p.pool.Begin(ctx)
 	if err != nil {
 		return err
 	}
-	ctx := context.Background()
-	_, err = p.pool.Exec(ctx, `
-		INSERT INTO asn_prefix_cache (asn, holder, prefixes_json, fetched_at)
-		VALUES ($1, $2, $3::jsonb, now())
+	defer func() { _ = tx.Rollback(ctx) }()
+	_, err = tx.Exec(ctx, `
+		INSERT INTO asn_prefix_cache (asn, holder, fetched_at)
+		VALUES ($1, $2, now())
 		ON CONFLICT (asn) DO UPDATE SET
 			holder = EXCLUDED.holder,
-			prefixes_json = EXCLUDED.prefixes_json,
-			fetched_at = EXCLUDED.fetched_at`,
-		asn, holder, string(raw))
-	return err
+			fetched_at = EXCLUDED.fetched_at`, asn, holder)
+	if err != nil {
+		return err
+	}
+	if asnCacheRowTableExists(ctx, tx) {
+		if _, err := tx.Exec(ctx, `DELETE FROM asn_prefix_cache_row WHERE asn = $1`, asn); err != nil {
+			return err
+		}
+		for _, pfx := range prefixes {
+			if _, err := tx.Exec(ctx, `
+				INSERT INTO asn_prefix_cache_row (asn, prefix) VALUES ($1, $2::cidr)`, asn, pfx); err != nil {
+				return err
+			}
+		}
+	} else {
+		raw, err := json.Marshal(prefixes)
+		if err != nil {
+			return err
+		}
+		if _, err := tx.Exec(ctx, `
+			UPDATE asn_prefix_cache SET prefixes_json = $2::jsonb WHERE asn = $1`, asn, string(raw)); err != nil {
+			return err
+		}
+	}
+	return tx.Commit(ctx)
 }

internal/repository/ctx.go

@@ -0,0 +1,19 @@
+package repository
+
+import (
+	"context"
+	"time"
+)
+
+const defaultRepoTimeout = 60 * time.Second
+
+// boundedRepoCtx returns a context with default repository I/O timeout.
+func boundedRepoCtx(parent context.Context) (context.Context, context.CancelFunc) {
+	if parent == nil {
+		parent = context.Background()
+	}
+	if _, ok := parent.Deadline(); ok {
+		return parent, func() {}
+	}
+	return context.WithTimeout(parent, defaultRepoTimeout)
+}

internal/repository/job_audit.go

@@ -20,6 +20,28 @@ func NewJobAuditWriter(pool *pgxpool.Pool) *JobAuditWriter {
 	return &JobAuditWriter{pool: pool}
 }
 
+// UpsertQueued inserts a queued job row (best-effort).
+func (w *JobAuditWriter) UpsertQueued(ctx context.Context, tenantID, jobID, kind string, idempotencyKey *string, moduleID *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
+	}
+	var mod any
+	if moduleID != nil && *moduleID != "" {
+		mod = *moduleID
+	}
+	_, _ = w.pool.Exec(ctx, `
+		INSERT INTO job_audit (id, tenant_id, kind, status, idempotency_key, module_id, meta_json, created_at)
+		VALUES ($1::uuid, $2::uuid, $3, 'queued', $4, $5::uuid, $6::jsonb, now())
+		ON CONFLICT (tenant_id, idempotency_key) WHERE idempotency_key IS NOT NULL
+		DO UPDATE SET status='queued', meta_json=EXCLUDED.meta_json, module_id=EXCLUDED.module_id`,
+		jobID, tenantID, kind, idem, mod, metaJSON)
+}
+
 // 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 {
@@ -33,8 +55,7 @@ func (w *JobAuditWriter) UpsertRunning(ctx context.Context, tenantID, jobID, kin
 	_, _ = 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`,
+		ON CONFLICT (id) DO UPDATE SET status='running', started_at=COALESCE(job_audit.started_at, now()), meta_json=EXCLUDED.meta_json`,
 		jobID, tenantID, kind, idem, metaJSON)
 }
 

internal/repository/maintenance.go

@@ -0,0 +1,29 @@
+package repository
+
+import (
+	"context"
+	"time"
+)
+
+const (
+	jobAuditRetentionDays = 90
+	asnCacheRetentionDays = 7
+)
+
+// RunPeriodicMaintenance prunes stale job_audit and asn_prefix_cache rows (PostgreSQL).
+func (p *Postgres) RunPeriodicMaintenance(ctx context.Context) {
+	if p == nil || p.pool == nil {
+		return
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	jobCutoff := time.Now().UTC().Add(-time.Duration(jobAuditRetentionDays) * 24 * time.Hour)
+	_, _ = p.pool.Exec(ctx, `
+		DELETE FROM job_audit
+		WHERE created_at < $1
+		  AND status IN ('succeeded', 'failed', 'cancelled')`, jobCutoff)
+	asnCutoff := time.Now().UTC().Add(-time.Duration(asnCacheRetentionDays) * 24 * time.Hour)
+	_, _ = p.pool.Exec(ctx, `
+		DELETE FROM asn_prefix_cache WHERE fetched_at < $1`, asnCutoff)
+}

internal/repository/module_snapshot.go

@@ -12,16 +12,24 @@ import (
 	"github.com/jackc/pgx/v5"
 )
 
+func moduleSnapshotRowTableExists(ctx context.Context, q queryRower) bool {
+	var n int
+	err := q.QueryRow(ctx, `
+		SELECT 1 FROM information_schema.tables
+		WHERE table_schema = 'public' AND table_name = 'module_prefix_snapshot_row'
+		LIMIT 1`).Scan(&n)
+	return err == nil
+}
+
 func (p *Postgres) GetModulePrefixSnapshot(tenantID, moduleID string) (*store.ModulePrefixSnapshot, bool, error) {
 	ctx := context.Background()
 	var inputHash string
 	var collectedAt time.Time
-	var raw []byte
 	err := p.pool.QueryRow(ctx, `
-		SELECT input_hash, collected_at, prefixes_json
+		SELECT input_hash, collected_at
 		FROM module_prefix_snapshot
 		WHERE tenant_id = $1 AND module_id = $2`,
-		tenantID, moduleID).Scan(&inputHash, &collectedAt, &raw)
+		tenantID, moduleID).Scan(&inputHash, &collectedAt)
 	if err != nil {
 		if errors.Is(err, pgx.ErrNoRows) {
 			return nil, false, nil
@@ -29,9 +37,31 @@ func (p *Postgres) GetModulePrefixSnapshot(tenantID, moduleID string) (*store.Mo
 		return nil, false, err
 	}
 	var prefixes []store.PrefixRow
-	if len(raw) > 0 {
-		if err := json.Unmarshal(raw, &prefixes); err != nil {
-			return nil, false, err
+	if moduleSnapshotRowTableExists(ctx, p.pool) {
+		rows, qerr := p.pool.Query(ctx, `
+			SELECT prefix, community_id::text, source
+			FROM module_prefix_snapshot_row
+			WHERE tenant_id = $1::uuid AND module_id = $2::uuid
+			ORDER BY ord`, tenantID, moduleID)
+		if qerr != nil {
+			return nil, false, qerr
+		}
+		defer rows.Close()
+		for rows.Next() {
+			var pr store.PrefixRow
+			var comm *string
+			if err := rows.Scan(&pr.Prefix, &comm, &pr.Source); err != nil {
+				continue
+			}
+			pr.CommunityID = comm
+			prefixes = append(prefixes, pr)
+		}
+	} else {
+		var raw []byte
+		if err := p.pool.QueryRow(ctx, `
+			SELECT prefixes_json FROM module_prefix_snapshot
+			WHERE tenant_id = $1 AND module_id = $2`, tenantID, moduleID).Scan(&raw); err == nil && len(raw) > 0 {
+			_ = json.Unmarshal(raw, &prefixes)
 		}
 	}
 	return &store.ModulePrefixSnapshot{
@@ -45,20 +75,57 @@ func (p *Postgres) SetModulePrefixSnapshot(tenantID, moduleID, inputHash string,
 	if strings.TrimSpace(tenantID) == "" || strings.TrimSpace(moduleID) == "" || strings.TrimSpace(inputHash) == "" {
 		return store.ErrInvalidInput
 	}
-	raw, err := json.Marshal(prefixes)
+	ctx := context.Background()
+	tx, err := p.pool.Begin(ctx)
 	if err != nil {
 		return err
 	}
-	ctx := context.Background()
-	_, err = p.pool.Exec(ctx, `
-		INSERT INTO module_prefix_snapshot (tenant_id, module_id, input_hash, collected_at, prefixes_json)
-		VALUES ($1::uuid, $2::uuid, $3, now(), $4::jsonb)
+	defer func() { _ = tx.Rollback(ctx) }()
+	_, err = tx.Exec(ctx, `
+		INSERT INTO module_prefix_snapshot (tenant_id, module_id, input_hash, collected_at)
+		VALUES ($1::uuid, $2::uuid, $3, now())
 		ON CONFLICT (tenant_id, module_id) DO UPDATE SET
 			input_hash = EXCLUDED.input_hash,
-			collected_at = EXCLUDED.collected_at,
-			prefixes_json = EXCLUDED.prefixes_json`,
-		tenantID, moduleID, inputHash, string(raw))
-	return err
+			collected_at = EXCLUDED.collected_at`,
+		tenantID, moduleID, inputHash)
+	if err != nil {
+		return err
+	}
+	if moduleSnapshotRowTableExists(ctx, tx) {
+		if _, err := tx.Exec(ctx, `
+			DELETE FROM module_prefix_snapshot_row
+			WHERE tenant_id = $1::uuid AND module_id = $2::uuid`, tenantID, moduleID); err != nil {
+			return err
+		}
+		for i, pr := range prefixes {
+			var comm any
+			if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" {
+				comm = strings.TrimSpace(*pr.CommunityID)
+			}
+			src := pr.Source
+			if strings.TrimSpace(src) == "" {
+				src = "render"
+			}
+			if _, err := tx.Exec(ctx, `
+				INSERT INTO module_prefix_snapshot_row (tenant_id, module_id, ord, prefix, community_id, source)
+				VALUES ($1::uuid, $2::uuid, $3, $4, $5::uuid, $6)`,
+				tenantID, moduleID, i, strings.TrimSpace(pr.Prefix), comm, src); err != nil {
+				return err
+			}
+		}
+	} else {
+		raw, err := json.Marshal(prefixes)
+		if err != nil {
+			return err
+		}
+		if _, err := tx.Exec(ctx, `
+			UPDATE module_prefix_snapshot SET prefixes_json = $3::jsonb
+			WHERE tenant_id = $1::uuid AND module_id = $2::uuid`,
+			tenantID, moduleID, string(raw)); err != nil {
+			return err
+		}
+	}
+	return tx.Commit(ctx)
 }
 
 func (p *Postgres) DeleteModulePrefixSnapshot(tenantID, moduleID string) error {

internal/repository/postgres.go

@@ -659,7 +659,8 @@ func (p *Postgres) DeleteSpeaker(tenantID, id string) error {
 }
 
 func (p *Postgres) GetRevision(tenantID, revisionID string) (*store.Revision, error) {
-	ctx := context.Background()
+	ctx, cancel := boundedRepoCtx(context.Background())
+	defer cancel()
 	var r store.Revision
 	var mod *string
 	var parent *string
@@ -686,11 +687,38 @@ func (p *Postgres) GetRevision(tenantID, revisionID string) (*store.Revision, er
 	if mj.PreviewFragments == nil {
 		mj.PreviewFragments = map[string]string{}
 	}
-	r.PreviewFragments = mj.PreviewFragments
+	r.PreviewFragments = loadRevisionPreview(ctx, p.pool, revisionID, mj.PreviewFragments)
 	r.MaterializedPrefixCount = mj.MaterializedPrefixCount
 	return &r, nil
 }
 
+func (p *Postgres) GetRevisionSummary(tenantID, revisionID string) (*store.Revision, error) {
+	ctx, cancel := boundedRepoCtx(context.Background())
+	defer cancel()
+	var r store.Revision
+	var mod *string
+	var parent *string
+	var prefixCount int
+	err := p.pool.QueryRow(ctx, `
+		SELECT id::text, tenant_id::text, module_id::text, content_hash, parent_revision_id::text,
+			COALESCE((meta_json->>'materialized_prefix_count')::int, 0), created_at
+		FROM config_revision WHERE id=$1 AND tenant_id=$2`, revisionID, tenantID).Scan(
+		&r.ID, &r.TenantID, &mod, &r.ContentHash, &parent, &prefixCount, &r.CreatedAt)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil, store.ErrNotFound
+		}
+		return nil, err
+	}
+	if mod != nil {
+		r.ModuleID = *mod
+	}
+	r.ParentRevisionID = strOrNil(parent)
+	r.MaterializedPrefixCount = prefixCount
+	r.PreviewFragments = map[string]string{}
+	return &r, nil
+}
+
 func (p *Postgres) ListRevisions(tenantID, moduleID string, cursor string, limit int) ([]*store.Revision, string, bool) {
 	if limit <= 0 {
 		limit = 50
@@ -761,12 +789,7 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
 	if limit <= 0 {
 		limit = 50
 	}
-	off := 0
-	if cursor != "" {
-		if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
-			off = n
-		}
-	}
+	afterID, off, useOffset := store.ParsePrefixPageCursor(cursor)
 	ctx := context.Background()
 	var one int
 	if err := p.pool.QueryRow(ctx, `
@@ -777,34 +800,55 @@ func (p *Postgres) ListRevisionPrefixes(tenantID, revisionID string, cursor stri
 		}
 		return nil, "", false
 	}
-	rows, err := p.pool.Query(ctx, `
-		SELECT prefix::text, community_id::text, source FROM revision_materialized_prefix
-		WHERE revision_id=$1 ORDER BY id
-		LIMIT $2 OFFSET $3`, revisionID, limit+1, off)
+	if snapID, ok := p.revisionPrefixSnapshotID(ctx, revisionID); ok {
+		return p.listSnapshotPrefixes(ctx, snapID, cursor, limit)
+	}
+	var rows pgx.Rows
+	var err error
+	if useOffset {
+		rows, err = p.pool.Query(ctx, `
+			SELECT id, prefix::text, community_id::text, source FROM revision_materialized_prefix
+			WHERE revision_id=$1::uuid ORDER BY id
+			LIMIT $2 OFFSET $3`, revisionID, limit+1, off)
+	} else {
+		var afterArg any
+		if afterID != nil {
+			afterArg = *afterID
+		}
+		rows, err = p.pool.Query(ctx, `
+			SELECT id, prefix::text, community_id::text, source FROM revision_materialized_prefix
+			WHERE revision_id=$1::uuid AND ($2::bigint IS NULL OR id > $2::bigint)
+			ORDER BY id
+			LIMIT $3`, revisionID, afterArg, limit+1)
+	}
 	if err != nil {
 		return nil, "", false
 	}
 	defer rows.Close()
 	var all []store.PrefixRow
+	var ids []int64
 	for rows.Next() {
+		var rowID int64
 		var pr store.PrefixRow
 		var comm *string
-		if err := rows.Scan(&pr.Prefix, &comm, &pr.Source); err != nil {
+		if err := rows.Scan(&rowID, &pr.Prefix, &comm, &pr.Source); err != nil {
 			continue
 		}
 		pr.CommunityID = comm
+		ids = append(ids, rowID)
 		all = append(all, pr)
 	}
 	agentDebugNDJSON3214("B", "repository/postgres.go:ListRevisionPrefixes", "list_prefixes_fetched", map[string]any{
-		"rows": len(all), "limit": limit, "offset": off,
+		"rows": len(all), "limit": limit, "keyset": !useOffset,
 	})
 	more := len(all) > limit
 	if more {
 		all = all[:limit]
+		ids = ids[:limit]
 	}
 	next := ""
-	if more {
-		next = fmt.Sprintf("%d", off+limit)
+	if more && len(ids) > 0 {
+		next = store.FormatPrefixPageCursor(ids[len(ids)-1])
 	}
 	if len(all) == 0 {
 		return nil, "", false
@@ -820,26 +864,35 @@ func (p *Postgres) CreateRollbackRevision(tenantID, sourceRevisionID string) (st
 	ctx := context.Background()
 	newID := uuid.NewString()
 	parent := sourceRevisionID
-	meta, _ := json.Marshal(map[string]any{
-		"preview_fragments":         src.PreviewFragments,
-		"materialized_prefix_count": src.MaterializedPrefixCount,
-	})
+	meta, err := revisionMetaWithoutPreview(src.MaterializedPrefixCount)
+	if err != nil {
+		return "", err
+	}
 	var modArg any
 	if strings.TrimSpace(src.ModuleID) != "" {
 		modArg = src.ModuleID
 	}
-	_, err = p.pool.Exec(ctx, `
-		INSERT INTO config_revision (id, tenant_id, module_id, content_hash, parent_revision_id, meta_json)
-		VALUES ($1,$2,$3,$4,$5::uuid,$6::jsonb)`,
-		newID, tenantID, modArg, src.ContentHash+":rollback", parent, string(meta))
+	tx, err := p.pool.Begin(ctx)
 	if err != nil {
 		return "", err
 	}
-	// copy materialized prefixes
-	_, _ = p.pool.Exec(ctx, `
-		INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source, meta_json)
-		SELECT $1::uuid, prefix, community_id, source, meta_json FROM revision_materialized_prefix WHERE revision_id=$2::uuid`,
-		newID, sourceRevisionID)
+	defer func() { _ = tx.Rollback(ctx) }()
+	_, err = tx.Exec(ctx, `
+		INSERT INTO config_revision (id, tenant_id, module_id, content_hash, parent_revision_id, meta_json)
+		VALUES ($1,$2,$3,$4,$5::uuid,$6::jsonb)`,
+		newID, tenantID, modArg, src.ContentHash+":rollback", parent, meta)
+	if err != nil {
+		return "", err
+	}
+	if err := copyRevisionPreview(ctx, tx, newID, sourceRevisionID); err != nil {
+		return "", err
+	}
+	if err := p.copyRevisionPrefixSnapshotRef(ctx, tx, newID, sourceRevisionID); err != nil {
+		return "", err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return "", err
+	}
 	return newID, nil
 }
 
@@ -856,20 +909,18 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
 	var unchanged int
 	err := p.pool.QueryRow(ctx, `
 		SELECT COUNT(*)::int FROM (
-			SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$1::uuid
-			INTERSECT
-			SELECT prefix FROM revision_materialized_prefix WHERE revision_id=$2::uuid
+			SELECT b.prefix FROM (`+sqlRevisionPrefixes("$2")+`) b
+			INNER JOIN (`+sqlRevisionPrefixes("$1")+`) a ON a.prefix = b.prefix
 		) t`, aID, bID).Scan(&unchanged)
 	if err != nil {
 		return nil, err
 	}
-	// added: в B, нет в A; removed: в A, нет в B — без загрузки полных снапшотов в память.
 	rowsAdded, 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 LIMIT $3`, bID, aID, maxRevisionDiffRows+1)
+		SELECT b.prefix::text FROM (`+sqlRevisionPrefixes("$2")+`) b
+		LEFT JOIN (`+sqlRevisionPrefixes("$1")+`) a ON a.prefix = b.prefix
+		WHERE a.prefix IS NULL
+		ORDER BY b.prefix
+		LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
 	if err != nil {
 		return nil, err
 	}
@@ -888,11 +939,11 @@ func (p *Postgres) RevisionDiff(tenantID, aID, bID string) (map[string]any, erro
 	}
 	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 LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
+		SELECT a.prefix::text FROM (`+sqlRevisionPrefixes("$1")+`) a
+		LEFT JOIN (`+sqlRevisionPrefixes("$2")+`) b ON b.prefix = a.prefix
+		WHERE b.prefix IS NULL
+		ORDER BY a.prefix
+		LIMIT $3`, aID, bID, maxRevisionDiffRows+1)
 	if err != nil {
 		return nil, err
 	}
@@ -1028,10 +1079,7 @@ func (p *Postgres) CreateRenderRevision(revisionID, tenantID, moduleID string, p
 	if previewFragments == nil {
 		previewFragments = map[string]string{}
 	}
-	meta, err := json.Marshal(map[string]any{
-		"preview_fragments":         previewFragments,
-		"materialized_prefix_count": len(prefixes),
-	})
+	meta, err := revisionMetaWithoutPreview(len(prefixes))
 	if err != nil {
 		return err
 	}
@@ -1045,32 +1093,23 @@ func (p *Postgres) CreateRenderRevision(revisionID, tenantID, moduleID string, p
 	if parentRevisionID != nil && strings.TrimSpace(*parentRevisionID) != "" {
 		parent = strings.TrimSpace(*parentRevisionID)
 	}
+	revID := strings.TrimSpace(revisionID)
 	_, err = tx.Exec(ctx, `
 		INSERT INTO config_revision (id, tenant_id, module_id, content_hash, parent_revision_id, meta_json)
 		VALUES ($1::uuid, $2::uuid, $3::uuid, $4, $5::uuid, $6::jsonb)`,
-		strings.TrimSpace(revisionID), tenantID, moduleID, strings.TrimSpace(contentHash), parent, string(meta))
+		revID, tenantID, moduleID, strings.TrimSpace(contentHash), parent, meta)
 	if err != nil {
 		return err
 	}
-	if len(prefixes) > 0 {
-		_, err = tx.CopyFrom(ctx,
-			pgx.Identifier{"revision_materialized_prefix"},
-			[]string{"revision_id", "prefix", "community_id", "source"},
-			pgx.CopyFromSlice(len(prefixes), func(i int) ([]any, error) {
-				pr := prefixes[i]
-				var comm any
-				if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" {
-					comm = strings.TrimSpace(*pr.CommunityID)
-				}
-				src := pr.Source
-				if strings.TrimSpace(src) == "" {
-					src = "render"
-				}
-				return []any{strings.TrimSpace(revisionID), strings.TrimSpace(pr.Prefix), comm, src}, nil
-			}))
-		if err != nil {
-			return err
-		}
+	if err := insertRevisionPreview(ctx, tx, revID, previewFragments); err != nil {
+		return err
+	}
+	snapID, err := p.ensurePrefixSnapshot(ctx, tx, contentHash, prefixes)
+	if err != nil {
+		return err
+	}
+	if err := p.linkRevisionPrefixSnapshot(ctx, tx, revID, snapID); err != nil {
+		return err
 	}
 	if err := tx.Commit(ctx); err != nil {
 		return err

internal/repository/postgres_prefix_snapshot.go

@@ -0,0 +1,186 @@
+package repository
+
+import (
+	"context"
+	"errors"
+	"strings"
+
+	"evobgp/internal/store"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+)
+
+func prefixSnapshotTableExists(ctx context.Context, q queryRower) bool {
+	var n int
+	err := q.QueryRow(ctx, `
+		SELECT 1 FROM information_schema.tables
+		WHERE table_schema = 'public' AND table_name = 'prefix_snapshot'
+		LIMIT 1`).Scan(&n)
+	return err == nil
+}
+
+func normalizeSnapshotHash(contentHash string) string {
+	h := strings.TrimSpace(contentHash)
+	if strings.HasPrefix(h, "sha256:") {
+		h = strings.TrimPrefix(h, "sha256:")
+	}
+	if len(h) > 64 {
+		h = h[:64]
+	}
+	if len(h) < 64 {
+		h = h + strings.Repeat("0", 64-len(h))
+	}
+	return h
+}
+
+func (p *Postgres) revisionPrefixSnapshotID(ctx context.Context, revisionID string) (string, bool) {
+	if !prefixSnapshotTableExists(ctx, p.pool) {
+		return "", false
+	}
+	var snap *string
+	err := p.pool.QueryRow(ctx, `
+		SELECT prefix_snapshot_id::text FROM config_revision
+		WHERE id = $1::uuid AND prefix_snapshot_id IS NOT NULL`, revisionID).Scan(&snap)
+	if err != nil || snap == nil || strings.TrimSpace(*snap) == "" {
+		return "", false
+	}
+	return *snap, true
+}
+
+func (p *Postgres) ensurePrefixSnapshot(ctx context.Context, db execQuerier, contentHash string, prefixes []store.PrefixRow) (string, error) {
+	if !prefixSnapshotTableExists(ctx, db) {
+		return "", nil
+	}
+	hash := normalizeSnapshotHash(contentHash)
+	var existing string
+	err := db.QueryRow(ctx, `SELECT id::text FROM prefix_snapshot WHERE content_hash = $1`, hash).Scan(&existing)
+	if err == nil && existing != "" {
+		return existing, nil
+	}
+	if err != nil && !errors.Is(err, pgx.ErrNoRows) {
+		return "", err
+	}
+	snapID := uuid.NewString()
+	if _, err := db.Exec(ctx, `
+		INSERT INTO prefix_snapshot (id, content_hash) VALUES ($1::uuid, $2)
+		ON CONFLICT (content_hash) DO NOTHING`, snapID, hash); err != nil {
+		return "", err
+	}
+	if err := db.QueryRow(ctx, `SELECT id::text FROM prefix_snapshot WHERE content_hash = $1`, hash).Scan(&snapID); err != nil {
+		return "", err
+	}
+	var rowCount int
+	_ = db.QueryRow(ctx, `SELECT COUNT(*)::int FROM prefix_snapshot_row WHERE snapshot_id = $1::uuid`, snapID).Scan(&rowCount)
+	if rowCount > 0 {
+		return snapID, nil
+	}
+	for i, pr := range prefixes {
+		var comm any
+		if pr.CommunityID != nil && strings.TrimSpace(*pr.CommunityID) != "" {
+			comm = strings.TrimSpace(*pr.CommunityID)
+		}
+		src := pr.Source
+		if strings.TrimSpace(src) == "" {
+			src = "render"
+		}
+		if _, err := db.Exec(ctx, `
+			INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
+			VALUES ($1::uuid, $2, $3, $4::uuid, $5)`,
+			snapID, i, strings.TrimSpace(pr.Prefix), comm, src); err != nil {
+			return "", err
+		}
+	}
+	return snapID, nil
+}
+
+func (p *Postgres) listSnapshotPrefixes(ctx context.Context, snapshotID, cursor string, limit int) ([]store.PrefixRow, string, bool) {
+	afterOrd, off, useOffset := store.ParsePrefixPageCursor(cursor)
+	var rows pgx.Rows
+	var err error
+	if useOffset {
+		rows, err = p.pool.Query(ctx, `
+			SELECT ord, prefix::text, community_id::text, source
+			FROM prefix_snapshot_row
+			WHERE snapshot_id = $1::uuid
+			ORDER BY ord
+			LIMIT $2 OFFSET $3`, snapshotID, limit+1, off)
+	} else {
+		var afterArg any
+		if afterOrd != nil {
+			afterArg = int(*afterOrd)
+		}
+		rows, err = p.pool.Query(ctx, `
+			SELECT ord, prefix::text, community_id::text, source
+			FROM prefix_snapshot_row
+			WHERE snapshot_id = $1::uuid AND ($2::int IS NULL OR ord > $2::int)
+			ORDER BY ord
+			LIMIT $3`, snapshotID, afterArg, limit+1)
+	}
+	if err != nil {
+		return nil, "", false
+	}
+	defer rows.Close()
+	var all []store.PrefixRow
+	var ords []int64
+	for rows.Next() {
+		var ord int
+		var pr store.PrefixRow
+		var comm *string
+		if err := rows.Scan(&ord, &pr.Prefix, &comm, &pr.Source); err != nil {
+			continue
+		}
+		pr.CommunityID = comm
+		ords = append(ords, int64(ord))
+		all = append(all, pr)
+	}
+	more := len(all) > limit
+	if more {
+		all = all[:limit]
+		ords = ords[:limit]
+	}
+	next := ""
+	if more && len(ords) > 0 {
+		next = store.FormatPrefixPageCursor(ords[len(ords)-1])
+	}
+	if len(all) == 0 {
+		return nil, "", false
+	}
+	return all, next, more
+}
+
+func (p *Postgres) linkRevisionPrefixSnapshot(ctx context.Context, db execQuerier, revisionID, snapshotID string) error {
+	if snapshotID == "" || !prefixSnapshotTableExists(ctx, db) {
+		return nil
+	}
+	_, err := db.Exec(ctx, `
+		UPDATE config_revision SET prefix_snapshot_id = $2::uuid WHERE id = $1::uuid`,
+		revisionID, snapshotID)
+	return err
+}
+
+func (p *Postgres) copyRevisionPrefixSnapshotRef(ctx context.Context, db execQuerier, dstRevisionID, srcRevisionID string) error {
+	if !prefixSnapshotTableExists(ctx, db) {
+		return nil
+	}
+	_, err := db.Exec(ctx, `
+		UPDATE config_revision dst
+		SET prefix_snapshot_id = src.prefix_snapshot_id
+		FROM config_revision src
+		WHERE dst.id = $1::uuid AND src.id = $2::uuid AND src.prefix_snapshot_id IS NOT NULL`,
+		dstRevisionID, srcRevisionID)
+	return err
+}
+
+func sqlRevisionPrefixes(revParam string) string {
+	return `SELECT psr.prefix FROM config_revision cr
+JOIN prefix_snapshot_row psr ON psr.snapshot_id = cr.prefix_snapshot_id
+WHERE cr.id = ` + revParam + `::uuid AND cr.prefix_snapshot_id IS NOT NULL
+UNION ALL
+SELECT rmp.prefix FROM revision_materialized_prefix rmp
+WHERE rmp.revision_id = ` + revParam + `::uuid
+  AND NOT EXISTS (
+    SELECT 1 FROM config_revision cr2
+    WHERE cr2.id = ` + revParam + `::uuid AND cr2.prefix_snapshot_id IS NOT NULL
+  )`
+}

internal/repository/postgres_revision_preview.go

@@ -0,0 +1,90 @@
+package repository
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+func revisionPreviewTableExists(ctx context.Context, q queryRower) bool {
+	var n int
+	err := q.QueryRow(ctx, `
+		SELECT 1 FROM information_schema.tables
+		WHERE table_schema = 'public' AND table_name = 'config_revision_preview'
+		LIMIT 1`).Scan(&n)
+	return err == nil
+}
+
+type queryRower interface {
+	QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
+}
+
+func loadRevisionPreview(ctx context.Context, q queryRower, revisionID string, metaPreview map[string]string) map[string]string {
+	if revisionPreviewTableExists(ctx, q) {
+		var raw []byte
+		err := q.QueryRow(ctx, `
+			SELECT fragments FROM config_revision_preview WHERE revision_id = $1::uuid`,
+			revisionID).Scan(&raw)
+		if err == nil {
+			out := map[string]string{}
+			_ = json.Unmarshal(raw, &out)
+			if out == nil {
+				out = map[string]string{}
+			}
+			return out
+		}
+		if !errors.Is(err, pgx.ErrNoRows) {
+			return metaPreview
+		}
+	}
+	if metaPreview == nil {
+		return map[string]string{}
+	}
+	return metaPreview
+}
+
+type execQuerier interface {
+	queryRower
+	Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
+}
+
+func insertRevisionPreview(ctx context.Context, db execQuerier, revisionID string, preview map[string]string) error {
+	if !revisionPreviewTableExists(ctx, db) {
+		return nil
+	}
+	if preview == nil {
+		preview = map[string]string{}
+	}
+	raw, err := json.Marshal(preview)
+	if err != nil {
+		return err
+	}
+	_, err = db.Exec(ctx, `
+		INSERT INTO config_revision_preview (revision_id, fragments)
+		VALUES ($1::uuid, $2::jsonb)
+		ON CONFLICT (revision_id) DO UPDATE SET fragments = EXCLUDED.fragments`,
+		revisionID, string(raw))
+	return err
+}
+
+func copyRevisionPreview(ctx context.Context, db execQuerier, dstRevisionID, srcRevisionID string) error {
+	if !revisionPreviewTableExists(ctx, db) {
+		return nil
+	}
+	_, err := db.Exec(ctx, `
+		INSERT INTO config_revision_preview (revision_id, fragments)
+		SELECT $1::uuid, fragments FROM config_revision_preview WHERE revision_id = $2::uuid
+		ON CONFLICT (revision_id) DO UPDATE SET fragments = EXCLUDED.fragments`,
+		dstRevisionID, srcRevisionID)
+	return err
+}
+
+func revisionMetaWithoutPreview(materializedPrefixCount int) (string, error) {
+	raw, err := json.Marshal(map[string]any{
+		"materialized_prefix_count": materializedPrefixCount,
+	})
+	return string(raw), err
+}

internal/repository/postgres_seed.go

@@ -26,9 +26,8 @@ func (p *Postgres) seedDemo(ctx context.Context) error {
 	p1 := uuid.NewString()
 	p2 := uuid.NewString()
 
-	preview := map[string]any{
-		"preview_fragments": map[string]string{
-			"bird.conf": `# EvoBGP demo bundle
+	previewFrags := map[string]string{
+		"bird.conf": `# EvoBGP demo bundle
 router id 192.0.2.1;
 
 protocol device {
@@ -39,15 +38,12 @@ protocol direct {
     ipv6;
 }
 `,
-			"bird.d/evobgp_demo.conf": "# static demo fragment\n",
-		},
-		"materialized_prefix_count": 128,
+		"bird.d/evobgp_demo.conf": "# static demo fragment\n",
 	}
-	previewB, _ := json.Marshal(preview)
-	parentMeta, _ := json.Marshal(map[string]any{
-		"preview_fragments":         map[string]string{"bird.conf": "# parent revision\n"},
-		"materialized_prefix_count": 0,
-	})
+	previewMeta, _ := json.Marshal(map[string]any{"materialized_prefix_count": 128})
+	parentMeta, _ := json.Marshal(map[string]any{"materialized_prefix_count": 0})
+	parentPreview, _ := json.Marshal(map[string]string{"bird.conf": "# parent revision\n"})
+	previewFragsB, _ := json.Marshal(previewFrags)
 
 	tx, err := p.pool.Begin(ctx)
 	if err != nil {
@@ -77,13 +73,32 @@ protocol direct {
 		return err
 	}
 	if _, err := tx.Exec(ctx, `
-		INSERT INTO config_revision (id, tenant_id, content_hash, parent_revision_id, meta_json)
-		VALUES ($1,$2,'sha256:demo-rev-1',$3::uuid,$4::jsonb)`, rid, tid, parent, string(previewB)); err != nil {
+		INSERT INTO config_revision_preview (revision_id, fragments) VALUES ($1::uuid, $2::jsonb)`, parent, string(parentPreview)); err != nil {
 		return err
 	}
 	if _, err := tx.Exec(ctx, `
-		INSERT INTO revision_materialized_prefix (revision_id, prefix, community_id, source)
-		VALUES ($1::uuid,'203.0.113.0/24',$2::uuid,'demo'), ($1::uuid,'2001:db8::/32',$2::uuid,'demo')`, rid, cid); err != nil {
+		INSERT INTO config_revision (id, tenant_id, content_hash, parent_revision_id, meta_json)
+		VALUES ($1,$2,'sha256:demo-rev-1',$3::uuid,$4::jsonb)`, rid, tid, parent, string(previewMeta)); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(ctx, `
+		INSERT INTO config_revision_preview (revision_id, fragments) VALUES ($1::uuid, $2::jsonb)`, rid, string(previewFragsB)); err != nil {
+		return err
+	}
+	snapID := uuid.NewString()
+	demoHash := normalizeSnapshotHash("sha256:demo-rev-1")
+	if _, err := tx.Exec(ctx, `
+		INSERT INTO prefix_snapshot (id, content_hash) VALUES ($1::uuid, $2)`, snapID, demoHash); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(ctx, `
+		UPDATE config_revision SET prefix_snapshot_id = $2::uuid WHERE id = $1::uuid`, rid, snapID); err != nil {
+		return err
+	}
+	if _, err := tx.Exec(ctx, `
+		INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
+		VALUES ($1::uuid, 0, '203.0.113.0/24', $2::uuid, 'demo'),
+		       ($1::uuid, 1, '2001:db8::/32', $2::uuid, 'demo')`, snapID, cid); err != nil {
 		return err
 	}
 	if _, err := tx.Exec(ctx, `

internal/scheduler/run.go

@@ -14,6 +14,7 @@ import (
 
 	"evobgp/internal/broker"
 	"evobgp/internal/config"
+	"evobgp/internal/httpclient"
 	"evobgp/internal/jobs"
 	"evobgp/internal/pipeline"
 	"evobgp/internal/store"
@@ -108,6 +109,9 @@ func postTenantRefresh(ctx context.Context, deps *Deps, moduleIDs []string, idem
 	if err != nil {
 		return err
 	}
+	req.GetBody = func() (io.ReadCloser, error) {
+		return io.NopCloser(bytes.NewReader(body)), nil
+	}
 	req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(deps.APIToken))
 	req.Header.Set("Content-Type", "application/json")
 	if idempotencyKey != "" {
@@ -115,9 +119,9 @@ func postTenantRefresh(ctx context.Context, deps *Deps, moduleIDs []string, idem
 	}
 	hc := deps.HTTP
 	if hc == nil {
-		hc = http.DefaultClient
+		hc = httpclient.New(httpclient.DefaultTimeout)
 	}
-	resp, err := hc.Do(req)
+	resp, err := httpclient.DoWithRetry(ctx, hc, req, 3)
 	if err != nil {
 		return err
 	}

internal/store/backend.go

@@ -76,6 +76,8 @@ type Backend interface {
 	DeleteSpeaker(tenantID, id string) error
 
 	GetRevision(tenantID, revisionID string) (*Revision, error)
+	// GetRevisionSummary returns revision metadata without preview_fragments payloads.
+	GetRevisionSummary(tenantID, revisionID string) (*Revision, error)
 	ListRevisions(tenantID, moduleID string, cursor string, limit int) (items []*Revision, nextCursor string, hasMore bool)
 	ListRevisionPrefixes(tenantID, revisionID string, cursor string, limit int) (prefixes []PrefixRow, next string, more bool)
 	CreateRollbackRevision(tenantID, sourceRevisionID string) (newID string, err error)
@@ -110,6 +112,9 @@ type Backend interface {
 
 	// Ping verifies backend connectivity (no-op for in-memory).
 	Ping(ctx context.Context) error
+
+	// RunPeriodicMaintenance prunes stale DB rows (no-op for in-memory).
+	RunPeriodicMaintenance(ctx context.Context)
 }
 
 // ASNPrefixCacheEntry is a cached RIPEstat response for one ASN.
@@ -309,7 +314,7 @@ type SpeakerPatch struct {
 
 // PrefixRow is one materialized prefix for GET /revisions/.../prefixes.
 type PrefixRow struct {
-	Prefix      string
-	CommunityID *string
-	Source      string
+	Prefix      string  `json:"prefix"`
+	CommunityID *string `json:"community_id,omitempty"`
+	Source      string  `json:"source,omitempty"`
 }

internal/store/memory.go

@@ -305,6 +305,11 @@ func (m *Memory) Ping(ctx context.Context) error {
 	return nil
 }
 
+// RunPeriodicMaintenance is a no-op for the in-memory backend.
+func (m *Memory) RunPeriodicMaintenance(ctx context.Context) {
+	_ = ctx
+}
+
 // ListTenantIDs returns tenant ids sorted lexicographically.
 func (m *Memory) ListTenantIDs() ([]string, error) {
 	m.mu.RLock()
@@ -424,14 +429,19 @@ func (m *Memory) GetModule(tenantID, moduleID string) (*Module, error) {
 func (m *Memory) GetRevision(tenantID, revisionID string) (*Revision, error) {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
-	rev, ok := m.revisions[revisionID]
-	if !ok {
-		return nil, ErrNotFound
+	return m.getRevisionLocked(tenantID, revisionID)
+}
+
+func (m *Memory) GetRevisionSummary(tenantID, revisionID string) (*Revision, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	rev, err := m.getRevisionLocked(tenantID, revisionID)
+	if err != nil {
+		return nil, err
 	}
-	if rev.TenantID != tenantID {
-		return nil, ErrTenantScope
-	}
-	return rev, nil
+	cp := *rev
+	cp.PreviewFragments = nil
+	return &cp, nil
 }
 
 func (m *Memory) GetSpeaker(tenantID, speakerID string) (*Speaker, error) {

internal/store/memory_crud.go

@@ -1,8 +1,6 @@
 package store
 
 import (
-	"fmt"
-	"strconv"
 	"strings"
 	"time"
 
@@ -795,31 +793,37 @@ func (m *Memory) ListRevisionPrefixes(tenantID, revisionID string, cursor string
 	if limit <= 0 {
 		limit = 50
 	}
+	afterID, off, useOffset := ParsePrefixPageCursor(cursor)
 	m.mu.RLock()
 	defer m.mu.RUnlock()
 	if _, err := m.getRevisionLocked(tenantID, revisionID); err != nil {
 		return nil, "", false
 	}
-	all := m.revPrefixes[revisionID]
-	off := 0
-	if cursor != "" {
-		if n, err := strconv.Atoi(cursor); err == nil && n >= 0 {
-			off = n
-		}
+	allRows := m.revPrefixes[revisionID]
+	start := 0
+	if useOffset {
+		start = off
+	} else if afterID != nil {
+		start = int(*afterID) + 1
 	}
-	end := off + limit
-	next := ""
-	more := false
-	if end > len(all) {
-		end = len(all)
-	} else {
-		more = true
-		next = fmt.Sprintf("%d", end)
-	}
-	if off >= len(all) {
+	if start > len(allRows) {
 		return nil, "", false
 	}
-	return all[off:end], next, more
+	end := start + limit
+	next := ""
+	more := false
+	if end > len(allRows) {
+		end = len(allRows)
+	} else {
+		more = true
+		next = FormatPrefixPageCursor(int64(end - 1))
+	}
+	if start >= end {
+		return nil, "", false
+	}
+	out := make([]PrefixRow, end-start)
+	copy(out, allRows[start:end])
+	return out, next, more
 }
 
 func (m *Memory) ListGlobalSettings(tenantID string) (map[string]any, error) {

internal/store/pagination.go

@@ -0,0 +1,30 @@
+package store
+
+import (
+	"strconv"
+	"strings"
+)
+
+// ParsePrefixPageCursor decodes opaque cursors for revision prefix pagination.
+func ParsePrefixPageCursor(cursor string) (afterID *int64, offset int, useOffset bool) {
+	cursor = strings.TrimSpace(cursor)
+	if cursor == "" {
+		return nil, 0, false
+	}
+	if strings.HasPrefix(cursor, "o:") {
+		n, err := strconv.Atoi(strings.TrimPrefix(cursor, "o:"))
+		if err != nil || n < 0 {
+			return nil, 0, false
+		}
+		return nil, n, true
+	}
+	if n, err := strconv.ParseInt(cursor, 10, 64); err == nil && n >= 0 {
+		return &n, 0, false
+	}
+	return nil, 0, false
+}
+
+// FormatPrefixPageCursor encodes the keyset cursor (last row id or slice index).
+func FormatPrefixPageCursor(lastID int64) string {
+	return strconv.FormatInt(lastID, 10)
+}

internal/store/speaker_meta.go

@@ -72,7 +72,9 @@ func MergeSpeakerMetaJSON(existing string, patch SpeakerMeta) string {
 	if patch.LastDispatchAt != "" {
 		cur.LastDispatchAt = patch.LastDispatchAt
 	}
-	if patch.LastDispatchError != "" {
+	if patch.LastDispatchStatus == "ok" {
+		cur.LastDispatchError = ""
+	} else if patch.LastDispatchError != "" {
 		cur.LastDispatchError = patch.LastDispatchError
 	}
 	if patch.LastDispatchStatus != "" {

internal/store/speaker_meta_test.go

@@ -34,3 +34,24 @@ func TestAgentSyncURL(t *testing.T) {
 		t.Fatalf("got %q", u)
 	}
 }
+
+func TestMergeSpeakerMetaJSON_clearsDispatchErrorOnOk(t *testing.T) {
+	t.Parallel()
+	existing := store.SpeakerMetaJSON(store.SpeakerMeta{
+		LastDispatchError:  "HTTP 502: bundle 403",
+		LastDispatchStatus: "error",
+		SyncStatus:         "error",
+	})
+	merged := store.MergeSpeakerMetaJSON(existing, store.SpeakerMeta{
+		LastDispatchStatus: "ok",
+		SyncStatus:         "synced",
+		LastDispatchAt:     "2026-05-21T15:06:43Z",
+	})
+	m := store.ParseSpeakerMeta(merged)
+	if m.LastDispatchError != "" {
+		t.Fatalf("LastDispatchError should clear on ok dispatch, got %q", m.LastDispatchError)
+	}
+	if m.LastDispatchStatus != "ok" || m.SyncStatus != "synced" {
+		t.Fatalf("status: dispatch=%q sync=%q", m.LastDispatchStatus, m.SyncStatus)
+	}
+}

migrations/postgres/000014_query_indexes.down.sql

@@ -0,0 +1,6 @@
+DROP INDEX IF EXISTS idx_rev_mat_prefix_uniq;
+DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_cover;
+CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_id
+    ON revision_materialized_prefix (revision_id, id);
+DROP INDEX IF EXISTS idx_module_tenant_active_sort;
+DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_prefix;

migrations/postgres/000014_query_indexes.up.sql

@@ -0,0 +1,17 @@
+-- Query indexes: revision diff/list and module sort (H4 + M4).
+
+CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_prefix
+    ON revision_materialized_prefix (revision_id, prefix);
+
+CREATE INDEX IF NOT EXISTS idx_module_tenant_active_sort
+    ON module (tenant_id, priority, name)
+    WHERE deleted_at IS NULL;
+
+-- Covering index for prefix list pagination (PostgreSQL 11+ INCLUDE).
+DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_id;
+CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_cover
+    ON revision_materialized_prefix (revision_id, id)
+    INCLUDE (prefix, community_id, source);
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_rev_mat_prefix_uniq
+    ON revision_materialized_prefix (revision_id, prefix, community_id, source);

migrations/postgres/000015_revision_preview.down.sql

@@ -0,0 +1,6 @@
+UPDATE config_revision cr
+SET meta_json = cr.meta_json || jsonb_build_object('preview_fragments', COALESCE(p.fragments, '{}'::jsonb))
+FROM config_revision_preview p
+WHERE p.revision_id = cr.id;
+
+DROP TABLE IF EXISTS config_revision_preview;

migrations/postgres/000015_revision_preview.up.sql

@@ -0,0 +1,15 @@
+-- Split BIRD preview fragments out of config_revision.meta_json (H1).
+
+CREATE TABLE config_revision_preview (
+    revision_id   UUID PRIMARY KEY REFERENCES config_revision (id) ON DELETE CASCADE,
+    fragments     JSONB NOT NULL DEFAULT '{}'
+);
+
+INSERT INTO config_revision_preview (revision_id, fragments)
+SELECT id, COALESCE(meta_json->'preview_fragments', '{}'::jsonb)
+FROM config_revision
+WHERE meta_json ? 'preview_fragments';
+
+UPDATE config_revision
+SET meta_json = meta_json - 'preview_fragments'
+WHERE meta_json ? 'preview_fragments';

migrations/postgres/000016_prefix_snapshot.down.sql

@@ -0,0 +1,3 @@
+ALTER TABLE config_revision DROP COLUMN IF EXISTS prefix_snapshot_id;
+DROP TABLE IF EXISTS prefix_snapshot_row;
+DROP TABLE IF EXISTS prefix_snapshot;

migrations/postgres/000016_prefix_snapshot.up.sql

@@ -0,0 +1,22 @@
+-- Content-addressed prefix snapshots (H2 expand).
+
+CREATE TABLE prefix_snapshot (
+    id            UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    content_hash  CHAR(64) NOT NULL,
+    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CONSTRAINT prefix_snapshot_hash_uniq UNIQUE (content_hash)
+);
+
+CREATE TABLE prefix_snapshot_row (
+    snapshot_id   UUID NOT NULL REFERENCES prefix_snapshot (id) ON DELETE CASCADE,
+    ord           INTEGER NOT NULL,
+    prefix        TEXT NOT NULL,
+    community_id  UUID REFERENCES bgp_community (id) ON DELETE SET NULL,
+    source        TEXT NOT NULL DEFAULT '',
+    PRIMARY KEY (snapshot_id, ord)
+);
+
+CREATE INDEX idx_prefix_snapshot_row_snapshot ON prefix_snapshot_row (snapshot_id);
+
+ALTER TABLE config_revision
+    ADD COLUMN prefix_snapshot_id UUID REFERENCES prefix_snapshot (id) ON DELETE RESTRICT;

migrations/postgres/000017_prefix_snapshot_backfill.down.sql

@@ -0,0 +1,3 @@
+DELETE FROM prefix_snapshot_row;
+UPDATE config_revision SET prefix_snapshot_id = NULL WHERE prefix_snapshot_id IS NOT NULL;
+DELETE FROM prefix_snapshot;

migrations/postgres/000017_prefix_snapshot_backfill.up.sql

@@ -0,0 +1,35 @@
+-- Backfill prefix snapshots from revision_materialized_prefix.
+-- prefix_snapshot_row.prefix is TEXT (revision_materialized_prefix.prefix since 000003).
+
+ALTER TABLE prefix_snapshot_row
+    ALTER COLUMN prefix TYPE TEXT USING prefix::text;
+
+WITH new_snaps AS (
+    INSERT INTO prefix_snapshot (id, content_hash)
+    SELECT gen_random_uuid(),
+        substr(replace(cr.id::text, '-', '') || replace(cr.id::text, '-', ''), 1, 64)
+    FROM config_revision cr
+    WHERE cr.prefix_snapshot_id IS NULL
+      AND EXISTS (
+          SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = cr.id
+      )
+    RETURNING id, content_hash
+)
+UPDATE config_revision cr
+SET prefix_snapshot_id = ns.id
+FROM new_snaps ns
+WHERE cr.prefix_snapshot_id IS NULL
+  AND ns.content_hash = substr(replace(cr.id::text, '-', '') || replace(cr.id::text, '-', ''), 1, 64);
+
+INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
+SELECT cr.prefix_snapshot_id,
+    (row_number() OVER (PARTITION BY cr.id ORDER BY rmp.id) - 1)::int,
+    rmp.prefix,
+    rmp.community_id,
+    rmp.source
+FROM config_revision cr
+JOIN revision_materialized_prefix rmp ON rmp.revision_id = cr.id
+WHERE cr.prefix_snapshot_id IS NOT NULL
+  AND NOT EXISTS (
+      SELECT 1 FROM prefix_snapshot_row psr WHERE psr.snapshot_id = cr.prefix_snapshot_id
+  );

migrations/postgres/000018_prefix_snapshot_contract.down.sql

@@ -0,0 +1 @@
+COMMENT ON COLUMN config_revision.prefix_snapshot_id IS NULL;

migrations/postgres/000018_prefix_snapshot_contract.up.sql

@@ -0,0 +1,3 @@
+-- H2 contract marker: new revisions use prefix_snapshot_id only (enforced in application code).
+
+COMMENT ON COLUMN config_revision.prefix_snapshot_id IS 'Materialized prefixes; revision_materialized_prefix deprecated for new rows';

migrations/postgres/000019_module_snapshot_rows.down.sql

@@ -0,0 +1,16 @@
+ALTER TABLE module_prefix_snapshot ADD COLUMN prefixes_json JSONB NOT NULL DEFAULT '[]';
+
+UPDATE module_prefix_snapshot mps
+SET prefixes_json = COALESCE((
+    SELECT jsonb_agg(
+        jsonb_build_object(
+            'prefix', psr.prefix::text,
+            'community_id', psr.community_id,
+            'source', psr.source
+        ) ORDER BY psr.ord
+    )
+    FROM module_prefix_snapshot_row psr
+    WHERE psr.tenant_id = mps.tenant_id AND psr.module_id = mps.module_id
+), '[]'::jsonb);
+
+DROP TABLE IF EXISTS module_prefix_snapshot_row;

migrations/postgres/000019_module_snapshot_rows.up.sql

@@ -0,0 +1,46 @@
+CREATE TABLE module_prefix_snapshot_row (
+    tenant_id     UUID NOT NULL,
+    module_id     UUID NOT NULL,
+    ord           INTEGER NOT NULL,
+    prefix        TEXT NOT NULL,
+    community_id  UUID,
+    source        TEXT NOT NULL DEFAULT '',
+    PRIMARY KEY (tenant_id, module_id, ord),
+    FOREIGN KEY (tenant_id, module_id)
+        REFERENCES module_prefix_snapshot (tenant_id, module_id) ON DELETE CASCADE
+);
+
+-- prefixes_json from Go json.Marshal(PrefixRow) used "Prefix"/"CommunityID"/"Source" before json tags.
+INSERT INTO module_prefix_snapshot_row (tenant_id, module_id, ord, prefix, community_id, source)
+SELECT tenant_id,
+    module_id,
+    (row_number() OVER (PARTITION BY tenant_id, module_id ORDER BY ordinality) - 1)::int,
+    prefix,
+    NULLIF(community_id, '')::uuid,
+    COALESCE(source, '')
+FROM (
+    SELECT mps.tenant_id,
+        mps.module_id,
+        t.ordinality,
+        COALESCE(
+            NULLIF(trim(t.elem->>'prefix'), ''),
+            NULLIF(trim(t.elem->>'Prefix'), '')
+        ) AS prefix,
+        COALESCE(
+            NULLIF(trim(t.elem->>'community_id'), ''),
+            NULLIF(trim(t.elem->>'CommunityID'), '')
+        ) AS community_id,
+        COALESCE(
+            NULLIF(trim(t.elem->>'source'), ''),
+            NULLIF(trim(t.elem->>'Source'), ''),
+            ''
+        ) AS source
+    FROM module_prefix_snapshot mps
+    CROSS JOIN LATERAL jsonb_array_elements(mps.prefixes_json) WITH ORDINALITY AS t(elem, ordinality)
+    WHERE jsonb_typeof(mps.prefixes_json) = 'array'
+      AND jsonb_array_length(mps.prefixes_json) > 0
+) parsed
+WHERE parsed.prefix IS NOT NULL
+  AND trim(parsed.prefix) <> '';
+
+ALTER TABLE module_prefix_snapshot DROP COLUMN prefixes_json;

migrations/postgres/000020_asn_cache_rows.down.sql

@@ -0,0 +1,10 @@
+ALTER TABLE asn_prefix_cache ADD COLUMN prefixes_json JSONB NOT NULL DEFAULT '[]';
+
+UPDATE asn_prefix_cache apc
+SET prefixes_json = COALESCE((
+    SELECT jsonb_agg(apcr.prefix::text ORDER BY apcr.prefix::text)
+    FROM asn_prefix_cache_row apcr
+    WHERE apcr.asn = apc.asn
+), '[]'::jsonb);
+
+DROP TABLE IF EXISTS asn_prefix_cache_row;

migrations/postgres/000020_asn_cache_rows.up.sql

@@ -0,0 +1,14 @@
+CREATE TABLE asn_prefix_cache_row (
+    asn     BIGINT NOT NULL REFERENCES asn_prefix_cache (asn) ON DELETE CASCADE,
+    prefix  CIDR NOT NULL,
+    PRIMARY KEY (asn, prefix)
+);
+
+INSERT INTO asn_prefix_cache_row (asn, prefix)
+SELECT apc.asn, t.elem::cidr
+FROM asn_prefix_cache apc
+CROSS JOIN LATERAL jsonb_array_elements_text(apc.prefixes_json) AS t(elem)
+WHERE jsonb_typeof(apc.prefixes_json) = 'array'
+  AND jsonb_array_length(apc.prefixes_json) > 0;
+
+ALTER TABLE asn_prefix_cache DROP COLUMN prefixes_json;

migrations/postgres/000021_job_audit_brin.down.sql

@@ -0,0 +1 @@
+DROP INDEX IF EXISTS idx_job_audit_created_brin;

migrations/postgres/000021_job_audit_brin.up.sql

@@ -0,0 +1,2 @@
+CREATE INDEX IF NOT EXISTS idx_job_audit_created_brin
+    ON job_audit USING BRIN (created_at);

migrations/postgres/000022_schema_cleanup.down.sql

@@ -0,0 +1,18 @@
+CREATE TABLE IF NOT EXISTS module_cdn_fetch_log (
+    id          BIGSERIAL PRIMARY KEY,
+    module_id   UUID NOT NULL REFERENCES module (id) ON DELETE CASCADE,
+    source_id   UUID REFERENCES module_cdn_source (id) ON DELETE SET NULL,
+    http_status INTEGER,
+    bytes       BIGINT,
+    error       TEXT,
+    fetched_at  TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+CREATE INDEX IF NOT EXISTS idx_module_cdn_fetch_log_module ON module_cdn_fetch_log (module_id, fetched_at DESC);
+
+ALTER TABLE revision_materialized_prefix ADD COLUMN IF NOT EXISTS meta_json JSONB NOT NULL DEFAULT '{}';
+ALTER TABLE module_domain_entry ADD COLUMN IF NOT EXISTS resolve_meta JSONB NOT NULL DEFAULT '{}';
+
+CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_revision ON revision_materialized_prefix (revision_id);
+CREATE INDEX IF NOT EXISTS idx_module_tenant ON module (tenant_id);
+CREATE INDEX IF NOT EXISTS idx_module_prefix_snapshot_collected ON module_prefix_snapshot (collected_at);
+CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_value ON revision_materialized_prefix (prefix);

migrations/postgres/000022_schema_cleanup.up.sql

@@ -0,0 +1,8 @@
+ALTER TABLE revision_materialized_prefix DROP COLUMN IF EXISTS meta_json;
+ALTER TABLE module_domain_entry DROP COLUMN IF EXISTS resolve_meta;
+DROP TABLE IF EXISTS module_cdn_fetch_log;
+
+DROP INDEX IF EXISTS idx_rev_mat_prefix_revision;
+DROP INDEX IF EXISTS idx_module_tenant;
+DROP INDEX IF EXISTS idx_module_prefix_snapshot_collected;
+DROP INDEX IF EXISTS idx_rev_mat_prefix_value;

migrations/sqlite/000014_query_indexes.down.sql

@@ -0,0 +1,3 @@
+DROP INDEX IF EXISTS idx_rev_mat_prefix_uniq;
+DROP INDEX IF EXISTS idx_module_tenant_active_sort;
+DROP INDEX IF EXISTS idx_rev_mat_prefix_rev_prefix;

migrations/sqlite/000014_query_indexes.up.sql

@@ -0,0 +1,9 @@
+CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_rev_prefix
+    ON revision_materialized_prefix (revision_id, prefix);
+
+CREATE INDEX IF NOT EXISTS idx_module_tenant_active_sort
+    ON module (tenant_id, priority, name)
+    WHERE deleted_at IS NULL;
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_rev_mat_prefix_uniq
+    ON revision_materialized_prefix (revision_id, prefix, community_id, source);

migrations/sqlite/000015_revision_preview.down.sql

@@ -0,0 +1,9 @@
+UPDATE config_revision
+SET meta_json = json_set(
+    meta_json,
+    '$.preview_fragments',
+    json(COALESCE((SELECT fragments FROM config_revision_preview p WHERE p.revision_id = config_revision.id), '{}'))
+)
+WHERE id IN (SELECT revision_id FROM config_revision_preview);
+
+DROP TABLE IF EXISTS config_revision_preview;

migrations/sqlite/000015_revision_preview.up.sql

@@ -0,0 +1,13 @@
+CREATE TABLE config_revision_preview (
+    revision_id   TEXT NOT NULL PRIMARY KEY REFERENCES config_revision (id) ON DELETE CASCADE,
+    fragments     TEXT NOT NULL DEFAULT '{}'
+);
+
+INSERT INTO config_revision_preview (revision_id, fragments)
+SELECT id, json(COALESCE(json_extract(meta_json, '$.preview_fragments'), '{}'))
+FROM config_revision
+WHERE json_extract(meta_json, '$.preview_fragments') IS NOT NULL;
+
+UPDATE config_revision
+SET meta_json = json_remove(meta_json, '$.preview_fragments')
+WHERE json_extract(meta_json, '$.preview_fragments') IS NOT NULL;

migrations/sqlite/000016_prefix_snapshot.down.sql

@@ -0,0 +1,3 @@
+ALTER TABLE config_revision DROP COLUMN prefix_snapshot_id;
+DROP TABLE IF EXISTS prefix_snapshot_row;
+DROP TABLE IF EXISTS prefix_snapshot;

migrations/sqlite/000016_prefix_snapshot.up.sql

@@ -0,0 +1,18 @@
+CREATE TABLE prefix_snapshot (
+    id            TEXT NOT NULL PRIMARY KEY,
+    content_hash  TEXT NOT NULL UNIQUE,
+    created_at    TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+
+CREATE TABLE prefix_snapshot_row (
+    snapshot_id   TEXT NOT NULL REFERENCES prefix_snapshot (id) ON DELETE CASCADE,
+    ord           INTEGER NOT NULL,
+    prefix        TEXT NOT NULL,
+    community_id  TEXT,
+    source        TEXT NOT NULL DEFAULT '',
+    PRIMARY KEY (snapshot_id, ord)
+);
+
+CREATE INDEX idx_prefix_snapshot_row_snapshot ON prefix_snapshot_row (snapshot_id);
+
+ALTER TABLE config_revision ADD COLUMN prefix_snapshot_id TEXT REFERENCES prefix_snapshot (id) ON DELETE RESTRICT;

migrations/sqlite/000017_prefix_snapshot_backfill.down.sql

@@ -0,0 +1,3 @@
+DELETE FROM prefix_snapshot_row;
+UPDATE config_revision SET prefix_snapshot_id = NULL;
+DELETE FROM prefix_snapshot;

migrations/sqlite/000017_prefix_snapshot_backfill.up.sql

@@ -0,0 +1,30 @@
+-- SQLite backfill: one snapshot per revision with materialized prefixes.
+
+INSERT INTO prefix_snapshot (id, content_hash)
+SELECT lower(hex(randomblob(16))),
+    substr(replace(cr.id, '-', '') || replace(cr.id, '-', ''), 1, 64)
+FROM config_revision cr
+WHERE cr.prefix_snapshot_id IS NULL
+  AND EXISTS (SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = cr.id);
+
+UPDATE config_revision
+SET prefix_snapshot_id = (
+    SELECT ps.id FROM prefix_snapshot ps
+    WHERE ps.content_hash = substr(replace(config_revision.id, '-', '') || replace(config_revision.id, '-', ''), 1, 64)
+)
+WHERE prefix_snapshot_id IS NULL
+  AND EXISTS (SELECT 1 FROM revision_materialized_prefix rmp WHERE rmp.revision_id = config_revision.id);
+
+INSERT INTO prefix_snapshot_row (snapshot_id, ord, prefix, community_id, source)
+SELECT cr.prefix_snapshot_id,
+    (SELECT COUNT(*) FROM revision_materialized_prefix r2
+     WHERE r2.revision_id = cr.id AND r2.id <= rmp.id) - 1,
+    rmp.prefix,
+    rmp.community_id,
+    rmp.source
+FROM config_revision cr
+JOIN revision_materialized_prefix rmp ON rmp.revision_id = cr.id
+WHERE cr.prefix_snapshot_id IS NOT NULL
+  AND NOT EXISTS (
+      SELECT 1 FROM prefix_snapshot_row psr WHERE psr.snapshot_id = cr.prefix_snapshot_id
+  );

migrations/sqlite/000018_prefix_snapshot_contract.up.sql

@@ -0,0 +1 @@
+-- H2 contract marker (application stops writing revision_materialized_prefix for new revisions).

migrations/sqlite/000019_module_snapshot_rows.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE module_prefix_snapshot ADD COLUMN prefixes_json TEXT NOT NULL DEFAULT '[]';
+DROP TABLE IF EXISTS module_prefix_snapshot_row;

migrations/sqlite/000019_module_snapshot_rows.up.sql

@@ -0,0 +1,13 @@
+CREATE TABLE module_prefix_snapshot_row (
+    tenant_id     TEXT NOT NULL,
+    module_id     TEXT NOT NULL,
+    ord           INTEGER NOT NULL,
+    prefix        TEXT NOT NULL,
+    community_id  TEXT,
+    source        TEXT NOT NULL DEFAULT '',
+    PRIMARY KEY (tenant_id, module_id, ord),
+    FOREIGN KEY (tenant_id, module_id)
+        REFERENCES module_prefix_snapshot (tenant_id, module_id) ON DELETE CASCADE
+);
+
+ALTER TABLE module_prefix_snapshot DROP COLUMN prefixes_json;

migrations/sqlite/000020_asn_cache_rows.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE asn_prefix_cache ADD COLUMN prefixes_json TEXT NOT NULL DEFAULT '[]';
+DROP TABLE IF EXISTS asn_prefix_cache_row;

migrations/sqlite/000020_asn_cache_rows.up.sql

@@ -0,0 +1,7 @@
+CREATE TABLE asn_prefix_cache_row (
+    asn     INTEGER NOT NULL REFERENCES asn_prefix_cache (asn) ON DELETE CASCADE,
+    prefix  TEXT NOT NULL,
+    PRIMARY KEY (asn, prefix)
+);
+
+ALTER TABLE asn_prefix_cache DROP COLUMN prefixes_json;

migrations/sqlite/000021_job_audit_brin.down.sql

@@ -0,0 +1 @@
+DROP INDEX IF EXISTS idx_job_audit_created_brin;

migrations/sqlite/000021_job_audit_brin.up.sql

@@ -0,0 +1 @@
+CREATE INDEX IF NOT EXISTS idx_job_audit_created_brin ON job_audit (created_at);

migrations/sqlite/000022_schema_cleanup.down.sql

@@ -0,0 +1,18 @@
+ALTER TABLE revision_materialized_prefix ADD COLUMN meta_json TEXT NOT NULL DEFAULT '{}';
+ALTER TABLE module_domain_entry ADD COLUMN resolve_meta TEXT NOT NULL DEFAULT '{}';
+
+CREATE TABLE IF NOT EXISTS module_cdn_fetch_log (
+    id          INTEGER PRIMARY KEY AUTOINCREMENT,
+    module_id   TEXT NOT NULL REFERENCES module (id) ON DELETE CASCADE,
+    source_id   TEXT REFERENCES module_cdn_source (id) ON DELETE SET NULL,
+    http_status INTEGER,
+    bytes       INTEGER,
+    error       TEXT,
+    fetched_at  TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
+);
+CREATE INDEX IF NOT EXISTS idx_module_cdn_fetch_log_module ON module_cdn_fetch_log (module_id, fetched_at DESC);
+
+CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_revision ON revision_materialized_prefix (revision_id);
+CREATE INDEX IF NOT EXISTS idx_module_tenant ON module (tenant_id);
+CREATE INDEX IF NOT EXISTS idx_module_prefix_snapshot_collected ON module_prefix_snapshot (collected_at);
+CREATE INDEX IF NOT EXISTS idx_rev_mat_prefix_value ON revision_materialized_prefix (prefix);

migrations/sqlite/000022_schema_cleanup.up.sql

@@ -0,0 +1,8 @@
+ALTER TABLE revision_materialized_prefix DROP COLUMN meta_json;
+ALTER TABLE module_domain_entry DROP COLUMN resolve_meta;
+DROP TABLE IF EXISTS module_cdn_fetch_log;
+
+DROP INDEX IF EXISTS idx_rev_mat_prefix_revision;
+DROP INDEX IF EXISTS idx_module_tenant;
+DROP INDEX IF EXISTS idx_module_prefix_snapshot_collected;
+DROP INDEX IF EXISTS idx_rev_mat_prefix_value;

web/src/lib/network/network-metrics.ts

@@ -277,6 +277,12 @@ export function formatSpeakerError(raw: string | null | undefined): FormattedSpe
 }
 
 export function speakerDispatchError(s: SpeakerRow): FormattedSpeakerError | null {
+	const liveRev = s.live?.agent_last_applied_revision_id?.trim();
+	const pub = s.published_revision_id?.trim();
+	// CP meta can keep a stale dispatch error after a later successful agent sync.
+	if (s.live?.agent_ok && liveRev && pub && liveRev === pub) {
+		return null;
+	}
 	return formatSpeakerError(s.last_dispatch_error);
 }