diff --git a/internal/pipeline/cdn_prefetch_test.go b/internal/pipeline/cdn_prefetch_test.go index 6bcbfa9..9d20314 100644 --- a/internal/pipeline/cdn_prefetch_test.go +++ b/internal/pipeline/cdn_prefetch_test.go @@ -97,3 +97,96 @@ func TestCollectModulePrefixRows_CDN304UsesSnapshot(t *testing.T) { t.Fatalf("want cached prefix on 304, got %+v", collected) } } + +func TestRefreshModuleIngest_CDN304UsesStoredSnapshot(t *testing.T) { + m := store.NewMemory() + m.SeedDemo() + tenant, _, _, _, _ := m.DemoIDs() + + mod, err := m.CreateModule(tenant, &store.Module{ + Type: "CDN_CIDRS", + Name: "cdn-304-ingest", + Enabled: true, + }) + if err != nil { + t.Fatal(err) + } + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNotModified) + })) + defer srv.Close() + + etag := "etag-stable" + src, err := m.CreateCDNSource(tenant, mod.ID, &store.CDNSource{ + SourceKind: "txt", + URL: srv.URL, + Etag: etag, + }) + if err != nil { + t.Fatal(err) + } + prior := []store.PrefixRow{{ + Prefix: "203.0.113.0/24", + Source: cdnSourceKey(src.ID), + }} + if err := mergeCDNSourceIntoModuleSnapshot(m, tenant, mod, src.ID, prior); err != nil { + t.Fatal(err) + } + + if err := RefreshModuleIngest(context.Background(), m, srv.Client(), tenant, mod.ID); err != nil { + t.Fatal(err) + } +} + +func TestCollectModulePrefixRows_CDN304RetriesWithoutETag(t *testing.T) { + m := store.NewMemory() + m.SeedDemo() + tenant, _, _, _, _ := m.DemoIDs() + + mod, err := m.CreateModule(tenant, &store.Module{ + Type: "CDN_CIDRS", + Name: "cdn-304-retry", + Enabled: true, + }) + if err != nil { + t.Fatal(err) + } + + var calls int + srv := httptest.NewServer(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" { + t.Fatalf("first request want If-None-Match etag-stable, got %q", got) + } + w.WriteHeader(http.StatusNotModified) + return + } + if got := strings.TrimSpace(r.Header.Get("If-None-Match")); got != "" { + t.Fatalf("retry must omit If-None-Match, got %q", got) + } + w.Header().Set("ETag", "etag-stable") + _, _ = w.Write([]byte("198.51.100.0/24\n")) + })) + defer srv.Close() + + if _, err := m.CreateCDNSource(tenant, mod.ID, &store.CDNSource{ + SourceKind: "txt", + URL: srv.URL, + Etag: "etag-stable", + }); err != nil { + t.Fatal(err) + } + + collected, err := collectModulePrefixRows(context.Background(), m, srv.Client(), tenant, mod, nil) + if err != nil { + t.Fatal(err) + } + if calls != 2 { + t.Fatalf("want 2 HTTP calls (304 then 200), got %d", calls) + } + if len(collected) != 1 || collected[0].Prefix != "198.51.100.0/24" { + t.Fatalf("unexpected collected rows: %+v", collected) + } +} diff --git a/internal/pipeline/cdn_snapshot.go b/internal/pipeline/cdn_snapshot.go index 0f0e587..0f55792 100644 --- a/internal/pipeline/cdn_snapshot.go +++ b/internal/pipeline/cdn_snapshot.go @@ -15,6 +15,23 @@ func cdnSourceKey(sourceID string) string { return "cdn:" + strings.TrimSpace(sourceID) } +func cachedCDNPrefixRows(st store.Backend, tenantID, moduleID string, priorSnapshot []store.PrefixRow, sourceKey string) []store.PrefixRow { + if cached := prefixRowsForSource(priorSnapshot, sourceKey); len(cached) > 0 { + return cached + } + if st != nil { + if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, moduleID); ok && snap != nil { + if cached := prefixRowsForSource(snap.Prefixes, sourceKey); len(cached) > 0 { + return cached + } + } + if cached := latestCDNRowsBySource(st, tenantID)[sourceKey]; len(cached) > 0 { + return cached + } + } + return nil +} + func mergeSnapshotDropSource(rows []store.PrefixRow, sourceKey string) []store.PrefixRow { if len(rows) == 0 { return nil @@ -88,14 +105,29 @@ func applyCDNSourceHTTPResult(ctx context.Context, st store.Backend, hc *http.Cl if err != nil { return nil, fmt.Errorf("cdn fetch %s: %w", u, err) } - defer resp.Body.Close() if resp.StatusCode == http.StatusNotModified { - if cached := prefixRowsForSource(priorSnapshot, sourceKey); len(cached) > 0 { + if cached := cachedCDNPrefixRows(st, tenantID, moduleID, priorSnapshot, sourceKey); len(cached) > 0 { + _ = resp.Body.Close() return cached, nil } - return nil, fmt.Errorf("cdn url %s: 304 without cached prefixes", u) + // ETag is known but local snapshot is empty — force a full download. + _ = resp.Body.Close() + req2, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return nil, err + } + resp, err = hc.Do(req2) + if err != nil { + return nil, fmt.Errorf("cdn fetch %s: %w", u, err) + } + if resp.StatusCode == http.StatusNotModified { + _ = resp.Body.Close() + return nil, fmt.Errorf("cdn url %s: 304 without cached prefixes", u) + } } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { _, _ = io.Copy(io.Discard, resp.Body) return nil, fmt.Errorf("cdn url %s: %s", u, resp.Status) diff --git a/internal/pipeline/refresh.go b/internal/pipeline/refresh.go index 645d7af..6d63602 100644 --- a/internal/pipeline/refresh.go +++ b/internal/pipeline/refresh.go @@ -53,7 +53,11 @@ func RefreshModuleIngest(ctx context.Context, st store.Backend, hc *http.Client, return fmt.Errorf("module disabled") } - rows, err := collectModulePrefixRows(ctx, st, hc, tenantID, mod, nil) + var prior []store.PrefixRow + if snap, ok, _ := st.GetModulePrefixSnapshot(tenantID, moduleID); ok && snap != nil { + prior = snap.Prefixes + } + rows, err := collectModulePrefixRows(ctx, st, hc, tenantID, mod, prior) if err != nil { return err }