feat(pipeline): use stale snapshot when upstream fetch fails

При ошибке CDN/ASN/DoH ingest использует последний снимок префиксов
(или просроченный ASN-кэш), если EVOBGP_STALE_ON_UPSTREAM_ERROR не равен 0.

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

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,70 @@
+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")
+
+	srv := httptest.NewServer(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")
+
+	srv := httptest.NewServer(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/refresh.go

@@ -176,7 +176,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 +192,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)
 	}