fix(pipeline): harden upstream resilience and production shutdown

DoH через DoWithRetry; CDN preview через UpstreamHTTPDo; частичный fail CDN (EVOBGP_CDN_PARTIAL_OK); безопасный доступ к Job.Meta; drain jobs при SIGTERM; ValidateProductionEnforce при EVOBGP_PRODUCTION=1.

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

cmd/evobgp-all/main.go

@@ -29,6 +29,9 @@ func main() {
 		os.Exit(dbcli.Run(os.Args[2:]))
 	}
 	cfg := config.Load()
+	if err := config.ValidateProductionEnforce(); err != nil {
+		log.Fatal(err)
+	}
 	opts := httpapi.Options{
 		APIKeys:                 os.Getenv("EVOBGP_API_KEYS"),
 		DatabaseURL:             cfg.DatabaseURL,
@@ -86,6 +89,13 @@ func main() {
 	<-ctx.Done()
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
 	defer cancel()
+	n := srv.Jobs().RequestCancelAll()
+	if n > 0 {
+		log.Printf("draining %d job(s)…", n)
+		if err := srv.Jobs().Drain(shutdownCtx); err != nil {
+			log.Printf("job drain: %v", err)
+		}
+	}
 	if err := httpSrv.Shutdown(shutdownCtx); err != nil {
 		log.Printf("HTTP shutdown: %v", err)
 	}

cmd/evobgp-api/main.go

@@ -24,6 +24,9 @@ func main() {
 		os.Exit(dbcli.Run(os.Args[2:]))
 	}
 	cfg := config.Load()
+	if err := config.ValidateProductionEnforce(); err != nil {
+		log.Fatal(err)
+	}
 	seedDemo := os.Getenv("EVOBGP_SEED_DEMO") != "0"
 	opts := httpapi.Options{
 		APIKeys:            os.Getenv("EVOBGP_API_KEYS"),
@@ -78,6 +81,13 @@ func main() {
 	<-ctx.Done()
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
 	defer cancel()
+	n := srv.Jobs().RequestCancelAll()
+	if n > 0 {
+		log.Printf("draining %d job(s)…", n)
+		if err := srv.Jobs().Drain(shutdownCtx); err != nil {
+			log.Printf("job drain: %v", err)
+		}
+	}
 	if err := httpSrv.Shutdown(shutdownCtx); err != nil {
 		log.Printf("HTTP shutdown: %v", err)
 	}

docs/production-checklist.md

@@ -4,12 +4,14 @@
 
 ## Обязательно
 
+- `EVOBGP_PRODUCTION=1` (или `EVOBGP_ENV=production`) — при старте API/`evobgp-all` процесс **откажется** стартовать, если нарушены жёсткие требования ниже (`config.ValidateProductionEnforce`).
 - `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_CDN_PARTIAL_OK=1` — при сбое одного CDN source без stale cache продолжать refresh остальных (иначе fail модуля).
 
 ## Рекомендуется
 

internal/config/production.go

@@ -0,0 +1,38 @@
+package config
+
+import (
+	"fmt"
+	"os"
+	"strings"
+)
+
+// ProductionMode reports whether EVOBGP_PRODUCTION / EVOBGP_ENV=production is set.
+func ProductionMode() bool {
+	if v := strings.TrimSpace(os.Getenv("EVOBGP_PRODUCTION")); v == "1" || strings.EqualFold(v, "true") {
+		return true
+	}
+	env := strings.ToLower(strings.TrimSpace(os.Getenv("EVOBGP_ENV")))
+	return env == "production" || env == "prod"
+}
+
+// ValidateProductionEnforce enforces docs/production-checklist.md hard requirements
+// when ProductionMode() is true. Returns an error that should abort process start.
+func ValidateProductionEnforce() error {
+	if !ProductionMode() {
+		return nil
+	}
+	seedDemo := os.Getenv("EVOBGP_SEED_DEMO")
+	if seedDemo != "0" {
+		return fmt.Errorf("config: production requires EVOBGP_SEED_DEMO=0 (got %q)", seedDemo)
+	}
+	if strings.TrimSpace(os.Getenv("EVOBGP_DEV_INSECURE")) == "1" {
+		return fmt.Errorf("config: production forbids EVOBGP_DEV_INSECURE=1")
+	}
+	if strings.TrimSpace(os.Getenv("EVOBGP_BUNDLE_SEED_HEX")) == "" {
+		return fmt.Errorf("config: production requires EVOBGP_BUNDLE_SEED_HEX")
+	}
+	if strings.TrimSpace(os.Getenv("EVOBGP_CDN_ALLOW_PRIVATE")) == "1" {
+		return fmt.Errorf("config: production forbids EVOBGP_CDN_ALLOW_PRIVATE=1")
+	}
+	return nil
+}

internal/config/production_test.go

@@ -0,0 +1,31 @@
+package config
+
+import "testing"
+
+func TestValidateProductionEnforce(t *testing.T) {
+	t.Setenv("EVOBGP_PRODUCTION", "")
+	t.Setenv("EVOBGP_ENV", "")
+	if err := ValidateProductionEnforce(); err != nil {
+		t.Fatalf("non-production: %v", err)
+	}
+
+	t.Setenv("EVOBGP_PRODUCTION", "1")
+	t.Setenv("EVOBGP_SEED_DEMO", "1")
+	t.Setenv("EVOBGP_BUNDLE_SEED_HEX", "abcd")
+	t.Setenv("EVOBGP_DEV_INSECURE", "")
+	t.Setenv("EVOBGP_CDN_ALLOW_PRIVATE", "")
+	if err := ValidateProductionEnforce(); err == nil {
+		t.Fatal("expected error for SEED_DEMO!=0")
+	}
+
+	t.Setenv("EVOBGP_SEED_DEMO", "0")
+	t.Setenv("EVOBGP_BUNDLE_SEED_HEX", "")
+	if err := ValidateProductionEnforce(); err == nil {
+		t.Fatal("expected error for missing BUNDLE_SEED_HEX")
+	}
+
+	t.Setenv("EVOBGP_BUNDLE_SEED_HEX", "bd8fbcd31545aacfdd228203beca8e945ab9a752f2ce5624cf42d9f316389a9d")
+	if err := ValidateProductionEnforce(); err != nil {
+		t.Fatalf("valid production: %v", err)
+	}
+}

internal/httpapi/routes_crud.go

@@ -314,7 +314,7 @@ func (s *Server) handlePreviewCDNSource(w http.ResponseWriter, r *http.Request)
 		writeProblem(w, http.StatusUnprocessableEntity, "Unprocessable Entity", "invalid url")
 		return
 	}
-	resp, err := s.cdnHTTP.Do(req)
+	resp, err := pipeline.UpstreamHTTPDo(r.Context(), s.cdnHTTP, req)
 	if err != nil {
 		writeBadGateway(w, "cdn preview fetch", err)
 		return

internal/jobs/job.go

@@ -1,6 +1,7 @@
 package jobs
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"sort"
@@ -111,6 +112,42 @@ func (j *Job) mergeMeta(kv map[string]any) {
 	}
 }
 
+// metaString returns a string meta field under lock (safe vs concurrent mergeMeta).
+func (j *Job) metaString(key string) string {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	if j.Meta == nil {
+		return ""
+	}
+	s, _ := j.Meta[key].(string)
+	return s
+}
+
+// metaBool returns a bool meta field under lock.
+func (j *Job) metaBool(key string) bool {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	if j.Meta == nil {
+		return false
+	}
+	b, _ := j.Meta[key].(bool)
+	return b
+}
+
+// metaCopy returns a shallow copy of job meta under lock.
+func (j *Job) metaCopy() map[string]any {
+	j.mu.Lock()
+	defer j.mu.Unlock()
+	if j.Meta == nil {
+		return nil
+	}
+	out := make(map[string]any, len(j.Meta))
+	for k, v := range j.Meta {
+		out[k] = v
+	}
+	return out
+}
+
 // statusLocked is used by the worker defer for metrics (any stable terminal or in-flight status).
 func (j *Job) statusLocked() string {
 	j.mu.Lock()
@@ -528,3 +565,65 @@ func (r *Registry) RequestCancel(tenantID, jobID string) (*Job, error) {
 	j.RequestCancel()
 	return j, nil
 }
+
+// RequestCancelAll requests cancellation of all non-terminal jobs (all tenants).
+func (r *Registry) RequestCancelAll() int {
+	if r == nil {
+		return 0
+	}
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	n := 0
+	for _, j := range r.byID {
+		if j == nil {
+			continue
+		}
+		st := j.statusLocked()
+		if st == StatusSucceeded || st == StatusFailed || st == StatusCancelled {
+			continue
+		}
+		j.RequestCancel()
+		n++
+	}
+	return n
+}
+
+// ActiveCount returns the number of queued or running jobs.
+func (r *Registry) ActiveCount() int {
+	if r == nil {
+		return 0
+	}
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	n := 0
+	for _, j := range r.byID {
+		if j == nil {
+			continue
+		}
+		st := j.statusLocked()
+		if st == StatusQueued || st == StatusRunning {
+			n++
+		}
+	}
+	return n
+}
+
+// Drain waits until no queued/running jobs remain or ctx is done.
+// Call RequestCancelAll first for a cooperative shutdown.
+func (r *Registry) Drain(ctx context.Context) error {
+	if r == nil {
+		return nil
+	}
+	ticker := time.NewTicker(50 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		if r.ActiveCount() == 0 {
+			return nil
+		}
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+	}
+}

internal/jobs/maintenance_worker.go

@@ -21,14 +21,13 @@ func (w *Worker) runMaintenancePolicy(j *Job) {
 		j.Fail("postgresql not configured")
 		return
 	}
-	policyID, _ := j.Meta["policy_id"].(string)
+	policyID := strings.TrimSpace(j.metaString("policy_id"))
-	policyID = strings.TrimSpace(policyID)
 	if policyID == "" {
 		j.Fail("missing policy_id in job meta")
 		return
 	}
-	dryRun, _ := j.Meta["dry_run"].(bool)
+	dryRun := j.metaBool("dry_run")
-	actor, _ := j.Meta["actor_prefix"].(string)
+	actor := j.metaString("actor_prefix")
 	ctx, cancel := j.workContext()
 	defer cancel()
 

internal/jobs/postgres_worker.go

@@ -93,9 +93,9 @@ func (w *Worker) runPostgresMaint(j *Job, kind string) {
 		j.Fail("postgresql not configured")
 		return
 	}
-	table, _ := j.Meta["table"].(string)
+	table := j.metaString("table")
-	dryRun, _ := j.Meta["dry_run"].(bool)
+	dryRun := j.metaBool("dry_run")
-	actor, _ := j.Meta["actor_prefix"].(string)
+	actor := j.metaString("actor_prefix")
 	ctx, cancel := j.workContext()
 	defer cancel()
 	auditID, _ := pgmonitor.InsertMaintenanceAudit(ctx, w.PgPool, j.TenantID, actor, kind, table, dryRun)

internal/jobs/worker.go

@@ -291,8 +291,8 @@ func (w *Worker) runModuleRefresh(j *Job) {
 		}
 	}()
 
-	mid, _ := j.Meta["module_id"].(string)
+	mid := strings.TrimSpace(j.metaString("module_id"))
-	if strings.TrimSpace(mid) == "" {
+	if mid == "" {
 		j.Fail("missing module_id in job meta")
 		return
 	}
@@ -325,7 +325,7 @@ func (w *Worker) runTenantRefresh(j *Job) {
 		}
 	}()
 
-	moduleIDs := moduleIDsFromJobMeta(j.Meta)
+	moduleIDs := moduleIDsFromJobMeta(j.metaCopy())
 	if len(moduleIDs) == 0 {
 		j.Fail("missing module_ids in job meta")
 		return
@@ -449,8 +449,9 @@ func (w *Worker) enqueueDeployAllSpeakers(j *Job, tenantID, revID string) {
 }
 
 func (w *Worker) runDeployApply(j *Job) {
-	revID, _ := j.Meta["revision_id"].(string)
+	revID := strings.TrimSpace(j.metaString("revision_id"))
-	spk, hasSpeaker := j.Meta["speaker_id"].(string)
+	spk := strings.TrimSpace(j.metaString("speaker_id"))
+	hasSpeaker := spk != ""
 	if revID == "" {
 		j.Fail("missing revision_id in job meta")
 		return
@@ -571,7 +572,7 @@ func (w *Worker) runDeployApply(j *Job) {
 }
 
 func (w *Worker) runRollback(j *Job) {
-	src, _ := j.Meta["source_revision_id"].(string)
+	src := strings.TrimSpace(j.metaString("source_revision_id"))
 	if src == "" {
 		j.Fail("missing source_revision_id in job meta")
 		return

internal/pipeline/collect_parallel.go

@@ -3,6 +3,7 @@ package pipeline
 import (
 	"context"
 	"fmt"
+	"log"
 	"net/http"
 	"os"
 	"strings"
@@ -218,12 +219,21 @@ func collectCDNPrefixRows(ctx context.Context, st store.Backend, hc *http.Client
 	wg.Wait()
 
 	var out []store.PrefixRow
+	var skipped int
 	for _, r := range results {
 		if r.err != nil {
+			if cdnPartialOK() {
+				log.Printf("pipeline: CDN partial skip source error: %v", r.err)
+				skipped++
+				continue
+			}
 			return nil, r.err
 		}
 		out = append(out, r.rows...)
 	}
+	if skipped > 0 && len(out) == 0 && len(valid) > 0 {
+		return nil, fmt.Errorf("cdn: all %d source(s) failed (partial ok)", len(valid))
+	}
 	if len(valid) > 0 {
 		if err := mergeAllCDNSourcesIntoModuleSnapshot(st, tenantID, mod, priorSnapshot, out); err != nil {
 			return nil, err

internal/pipeline/collect_stale.go

@@ -20,6 +20,13 @@ func staleOnUpstreamError() bool {
 	return false
 }
 
+// cdnPartialOK reports whether a failed CDN source without stale cache should be skipped
+// instead of failing the whole module refresh. Opt-in: EVOBGP_CDN_PARTIAL_OK=1.
+func cdnPartialOK() bool {
+	v := strings.TrimSpace(os.Getenv("EVOBGP_CDN_PARTIAL_OK"))
+	return v == "1" || strings.EqualFold(v, "true")
+}
+
 func logStaleUpstream(kind, detail string) {
 	log.Printf("pipeline: stale upstream fallback (%s): %s", kind, detail)
 }

internal/pipeline/refresh.go

@@ -282,7 +282,7 @@ func resolveDomainWithDOHMessage(ctx context.Context, hc *http.Client, baseURL,
 		return nil, err
 	}
 	req.Header.Set("Accept", "application/dns-message")
-	resp, err := hc.Do(req)
+	resp, err := httpclient.DoWithRetry(ctx, hc, req, 3)
 	if err != nil {
 		return nil, err
 	}
@@ -354,7 +354,7 @@ func resolveDomainWithDOHJSON(ctx context.Context, hc *http.Client, baseURL, hos
 	}
 	req.Header.Set("Accept", "application/dns-json")
 
-	resp, err := hc.Do(req)
+	resp, err := httpclient.DoWithRetry(ctx, hc, req, 3)
 	if err != nil {
 		return nil, err
 	}

internal/pipeline/upstream_http.go

@@ -8,6 +8,11 @@ import (
 	"evobgp/internal/httpclient"
 )
 
+// UpstreamHTTPDo performs an outbound GET/POST with circuit breaker + retries (CDN, previews).
+func UpstreamHTTPDo(ctx context.Context, hc *http.Client, req *http.Request) (*http.Response, error) {
+	return upstreamHTTPDo(ctx, hc, req)
+}
+
 func upstreamHTTPDo(ctx context.Context, hc *http.Client, req *http.Request) (*http.Response, error) {
 	if hc == nil {
 		hc = httpclient.New(httpclient.DefaultTimeout)