test: add CDN prefetch tests for 304 responses and snapshot handling

- Introduced `TestRefreshModuleIngest_CDN304UsesStoredSnapshot` to verify that the module ingest correctly uses stored snapshots when receiving a 304 Not Modified response.
- Added `TestCollectModulePrefixRows_CDN304RetriesWithoutETag` to ensure that the system retries without the If-None-Match header after a 304 response, and correctly collects prefixes on subsequent successful requests.
- Updated `cachedCDNPrefixRows` function to improve handling of cached prefixes during CDN source processing.
- Modified `RefreshModuleIngest` to utilize prior snapshots when collecting module prefix rows.

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)