fix(agentserver): hide upstream errors from clients

Generic 502 detail как в httpapi; добавлен docs/production-checklist.md.

Co-authored-by: Cursor <cursoragent@cursor.com>

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"
 )
@@ -24,7 +24,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 +34,10 @@ 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 {
+			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/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/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
@@ -691,6 +692,33 @@ func (p *Postgres) GetRevision(tenantID, revisionID string) (*store.Revision, er
 	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

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)

internal/store/memory.go

@@ -424,14 +424,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/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)
+	}
+}

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);
 }