guard_core_engine/
cloud_fetch.rs

1//! The cloud-provider range lifecycle: the fetch payload parsers, the
2//! `Azure` `ServiceTags` URL extraction, the refresh cadence with the
3//! single-flight gate, and the store-backed cache layers.
4//!
5//! This is the pure half of the reference cloud handler
6//! (`guard_core/handlers/cloud_handler.py`,
7//! `_cloud_provider_fetchers.py`, `_cloud_azure_fetch.py`,
8//! `_cloud_provider_registry.py`, `cloud_ip_stores.py`); the wire half -
9//! the HTTP client driving the reference endpoints - lives in the facade's
10//! `cloud_fetch` and feeds these parsers.
11//!
12//! ## Payload parsers (one per provider, the reference fetcher bodies)
13//!
14//! ```text
15//! AWS    json prefixes[]: entries with service == "AMAZON" only; the
16//!        ip_prefix network carries its region annotation; a malformed
17//!        prefix aborts the whole parse (the reference raise)
18//! GCP    json prefixes[]: ipv4Prefix or ipv6Prefix, scope annotation;
19//!        a malformed prefix aborts the whole parse (the reference raise)
20//! CSV    `DigitalOcean` / Linode: the first field per non-comment line;
21//!        invalid CIDRs are skipped, not fatal
22//! Vultr  json subnets[]: ip_prefix; invalid CIDRs are skipped
23//! Azure  json values[]: the "AzureCloud" entry's
24//!        properties.addressPrefixes; a malformed prefix fails the whole
25//!        parse (the reference raises into the fetcher's catch)
26//! ```
27//!
28//! ## `Azure` `ServiceTags` URL extraction (the reference order)
29//!
30//! 1. the `id="failoverLink"` anchor's href;
31//! 2. the newest URL matching the `ServiceTags` pattern (newest = the
32//!    `ServiceTags_Public_(\d{8})` date, then the URL text);
33//! 3. any `https://download.microsoft.com/...json` href.
34//!
35//! Every candidate must be https on `download.microsoft.com`, or it is
36//! rejected.
37//!
38//! ## Refresh lifecycle
39//!
40//! [`CloudRefreshCoordinator`] carries the reference stamp semantics: the
41//! check asks [`CloudRefreshCoordinator::should_refresh`]
42//! (`now - last > interval`, the reference `CloudIpRefreshCheck`), stamps
43//! optimistically, schedules, and rolls the stamp back when scheduling
44//! reports failure. [`CloudRefreshCoordinator::schedule_refresh`] is
45//! single-flight: while one refresh is in flight, further calls are no-ops
46//! returning `false`, so at most one refresh runs at a time.
47//!
48//! [`refresh_provider_from_store`] walks the reference cache layers,
49//! newest-to-oldest: the store (the section 08 `cloud_ip_v2` /
50//! `cloud_ranges_v2` shapes) first - a hit decodes and installs in memory
51//! with no network fetch; a miss fetches, and only a non-empty result is
52//! cached and installed (an empty fetch is never cached and never touches
53//! the provider's previous ranges).
54//!
55//! # Example
56//!
57//! ```
58//! use guard_core_engine::cloud_fetch::{parse_aws_ranges, extract_azure_download_url};
59//!
60//! // The AWS shape: only the AMAZON service entries, regions attached.
61//! let payload = r#"{"prefixes": [
62//!   {"ip_prefix": "203.0.113.0/24", "region": "us-east-1", "service": "AMAZON"},
63//!   {"ip_prefix": "198.51.100.0/24", "region": "eu-west-1", "service": "EC2"}
64//! ]}"#;
65//! let ranges = parse_aws_ranges(payload).expect("valid payload");
66//! assert_eq!(ranges, vec![(String::from("203.0.113.0/24"), Some(String::from("us-east-1")))]);
67//!
68//! // The Azure page scrape: the newest `ServiceTags` URL wins, on-host only.
69//! let page = r#"<a id="failoverLink" href="https://download.microsoft.com/?x">x</a>"#;
70//! # let page = r#"See https://download.microsoft.com/7/1/ServiceTags_Public_20260101.json for details"#;
71//! assert_eq!(
72//!     extract_azure_download_url(page).as_deref(),
73//!     Some("https://download.microsoft.com/7/1/ServiceTags_Public_20260101.json")
74//! );
75//! ```
76
77use std::sync::Arc;
78use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};
79
80/// The parsed provider ranges: `(network, region)` pairs in the
81/// [`crate::cloud_provider::CloudIpTable`] entry shape.
82pub type ParsedRanges = Vec<(String, Option<String>)>;
83
84/// A malformed provider payload: the JSON body did not carry the shape the
85/// parser reads.
86#[derive(Debug, Clone, PartialEq, Eq)]
87pub struct PayloadError {
88    /// What the parser expected and did not find.
89    pub reason: String,
90}
91
92impl core::fmt::Display for PayloadError {
93    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
94        write!(f, "provider payload error: {}", self.reason)
95    }
96}
97
98impl std::error::Error for PayloadError {}
99
100/// The AWS parser (`fetch_aws_ip_ranges`): the `prefixes[]` entries with
101/// `service == "AMAZON"` only, each `ip_prefix` carrying its `region` as
102/// the annotation.
103///
104/// # Errors
105///
106/// [`PayloadError`] when the body carries no `prefixes` array, or when an
107/// AMAZON entry's prefix does not parse (the reference
108/// `ipaddress.ip_network` raise aborts the whole fetch into its
109/// empty-set catch, which the refresh pass never caches).
110pub fn parse_aws_ranges(payload: &str) -> Result<ParsedRanges, PayloadError> {
111    let data: serde_json::Value = serde_json::from_str(payload).map_err(|e| PayloadError {
112        reason: format!("invalid json: {e}"),
113    })?;
114    let prefixes = data
115        .get("prefixes")
116        .and_then(serde_json::Value::as_array)
117        .ok_or_else(|| PayloadError {
118            reason: String::from("missing prefixes array"),
119        })?;
120    let mut ranges: ParsedRanges = Vec::new();
121    let mut seen = std::collections::HashSet::new();
122    for entry in prefixes {
123        if entry.get("service").and_then(serde_json::Value::as_str) != Some("AMAZON") {
124            continue;
125        }
126        let Some(prefix) = entry.get("ip_prefix").and_then(serde_json::Value::as_str) else {
127            continue;
128        };
129        let Some(network) = crate::redis_schema::canonical_network_string(prefix) else {
130            // The reference `ipaddress.ip_network(prefix)` raise aborts the
131            // whole fetch (its `except` returns an empty set, which the
132            // refresh pass never caches); a malformed entry never silently
133            // shrinks the provider's range set.
134            return Err(PayloadError {
135                reason: format!("invalid AWS prefix: {prefix}"),
136            });
137        };
138        let region = entry
139            .get("region")
140            .and_then(serde_json::Value::as_str)
141            .filter(|region| !region.is_empty())
142            .map(str::to_owned);
143        if seen.insert(network.clone()) {
144            ranges.push((network, region));
145        }
146    }
147    Ok(ranges)
148}
149
150/// The GCP parser (`fetch_gcp_ip_ranges`): the `prefixes[]` entries'
151/// `ipv4Prefix` / `ipv6Prefix` with the `scope` annotation.
152///
153/// # Errors
154///
155/// [`PayloadError`] when the body carries no `prefixes` array, or when a
156/// prefix does not parse (the reference `ip_network` raise aborts the
157/// whole fetch into its empty-set catch).
158pub fn parse_gcp_ranges(payload: &str) -> Result<ParsedRanges, PayloadError> {
159    let data: serde_json::Value = serde_json::from_str(payload).map_err(|e| PayloadError {
160        reason: format!("invalid json: {e}"),
161    })?;
162    let prefixes = data
163        .get("prefixes")
164        .and_then(serde_json::Value::as_array)
165        .ok_or_else(|| PayloadError {
166            reason: String::from("missing prefixes array"),
167        })?;
168    let mut ranges: ParsedRanges = Vec::new();
169    let mut seen = std::collections::HashSet::new();
170    for entry in prefixes {
171        let prefix = entry
172            .get("ipv4Prefix")
173            .or_else(|| entry.get("ipv6Prefix"))
174            .and_then(serde_json::Value::as_str);
175        let Some(prefix) = prefix else {
176            continue;
177        };
178        let Some(network) = crate::redis_schema::canonical_network_string(prefix) else {
179            // Same abort shape as the AWS parser: the reference raise
180            // empties the whole fetch.
181            return Err(PayloadError {
182                reason: format!("invalid GCP prefix: {prefix}"),
183            });
184        };
185        let scope = entry
186            .get("scope")
187            .and_then(serde_json::Value::as_str)
188            .filter(|scope| !scope.is_empty())
189            .map(str::to_owned);
190        if seen.insert(network.clone()) {
191            ranges.push((network, scope));
192        }
193    }
194    Ok(ranges)
195}
196
197/// The CSV parser (`_fetch_csv_prefix_networks`, the `DigitalOcean` and
198/// Linode shape): the first field of every non-comment line; blank lines
199/// are skipped and an invalid prefix is skipped, not fatal.
200#[must_use]
201pub fn parse_csv_ranges(body: &str) -> ParsedRanges {
202    let mut ranges: ParsedRanges = Vec::new();
203    let mut seen = std::collections::HashSet::new();
204    for raw_line in body.lines() {
205        let line = raw_line.trim();
206        if line.is_empty() || line.starts_with('#') {
207            continue;
208        }
209        let prefix = line.split(',').next().unwrap_or_default().trim();
210        if prefix.is_empty() {
211            continue;
212        }
213        if let Some(network) = crate::redis_schema::canonical_network_string(prefix)
214            && seen.insert(network.clone())
215        {
216            ranges.push((network, None));
217        }
218    }
219    ranges
220}
221
222/// The Vultr parser (`fetch_vultr_ip_ranges`): the `subnets[]` entries'
223/// `ip_prefix`; an invalid prefix is skipped, not fatal.
224///
225/// # Errors
226///
227/// [`PayloadError`] when the body is not JSON (the reference catch-all).
228pub fn parse_vultr_ranges(payload: &str) -> Result<ParsedRanges, PayloadError> {
229    let data: serde_json::Value = serde_json::from_str(payload).map_err(|e| PayloadError {
230        reason: format!("invalid json: {e}"),
231    })?;
232    let Some(subnets) = data.get("subnets").and_then(serde_json::Value::as_array) else {
233        return Ok(Vec::new());
234    };
235    let mut ranges: ParsedRanges = Vec::new();
236    let mut seen = std::collections::HashSet::new();
237    for entry in subnets {
238        let Some(prefix) = entry.get("ip_prefix").and_then(serde_json::Value::as_str) else {
239            continue;
240        };
241        if let Some(network) = crate::redis_schema::canonical_network_string(prefix)
242            && seen.insert(network.clone())
243        {
244            ranges.push((network, None));
245        }
246    }
247    Ok(ranges)
248}
249
250/// The `Azure` `ServiceTags` parser (`_select_azure_cloud_prefixes`): the
251/// `values[]` entry named `AzureCloud`, its
252/// `properties.addressPrefixes`.
253///
254/// # Errors
255///
256/// [`PayloadError`] when the document carries no `AzureCloud` tag (the
257/// reference logs the error and returns empty) or a prefix that does not
258/// parse (the reference `ip_network` raise caught into an empty fetch).
259pub fn parse_azure_service_tags(payload: &str) -> Result<ParsedRanges, PayloadError> {
260    let data: serde_json::Value = serde_json::from_str(payload).map_err(|e| PayloadError {
261        reason: format!("invalid json: {e}"),
262    })?;
263    let values = data
264        .get("values")
265        .and_then(serde_json::Value::as_array)
266        .ok_or_else(|| PayloadError {
267            reason: String::from("missing values array"),
268        })?;
269    for entry in values {
270        if entry.get("name").and_then(serde_json::Value::as_str) == Some("AzureCloud") {
271            let prefixes = entry
272                .get("properties")
273                .and_then(|properties| properties.get("addressPrefixes"))
274                .and_then(serde_json::Value::as_array)
275                .cloned()
276                .unwrap_or_default();
277            let mut ranges: ParsedRanges = Vec::new();
278            let mut seen = std::collections::HashSet::new();
279            for prefix in prefixes {
280                let Some(text) = prefix.as_str() else {
281                    continue;
282                };
283                let Some(network) = crate::redis_schema::canonical_network_string(text) else {
284                    return Err(PayloadError {
285                        reason: format!("invalid AzureCloud prefix: {text}"),
286                    });
287                };
288                if seen.insert(network.clone()) {
289                    ranges.push((network, None));
290                }
291            }
292            return Ok(ranges);
293        }
294    }
295    Err(PayloadError {
296        reason: String::from("no AzureCloud service tag"),
297    })
298}
299
300/// The trusted Azure download host (`_AZURE_TRUSTED_DOWNLOAD_HOST`): every
301/// candidate URL must be https on this exact host.
302pub const AZURE_TRUSTED_DOWNLOAD_HOST: &str = "download.microsoft.com";
303
304/// A selected `ServiceTags` URL and how fresh the reference considers it.
305#[derive(Debug, Clone, PartialEq, Eq)]
306pub struct AzureDownloadCandidate {
307    /// The URL (https, on the trusted host).
308    pub url: String,
309    /// The parsed `ServiceTags_Public_YYYYMMDD` date as `(year, month,
310    /// day)`, `None` when the URL carries no parseable date.
311    pub date: Option<(i32, u32, u32)>,
312}
313
314/// Whether `url` is a trusted Azure download URL (`_is_trusted_azure_
315/// download_url`): https scheme on the exact trusted host.
316#[must_use]
317pub fn is_trusted_azure_download_url(url: &str) -> bool {
318    let rest = url.strip_prefix("https://").unwrap_or(url);
319    let host = rest.find(['/', '?', '#']).map_or(rest, |end| &rest[..end]);
320    host == AZURE_TRUSTED_DOWNLOAD_HOST
321}
322
323/// The `ServiceTags_Public_(\d{8})` date of `url`
324/// (`_parse_service_tags_date`): `None` when the pattern is absent, the
325/// digits do not form a real calendar date, or the date lies in the
326/// future.
327#[must_use]
328pub fn parse_service_tags_date(url: &str) -> Option<(i32, u32, u32)> {
329    let start = url.find("ServiceTags_Public_")? + "ServiceTags_Public_".len();
330    let digits = url.get(start..start + 8)?;
331    if !digits.chars().all(|c| c.is_ascii_digit()) {
332        return None;
333    }
334    #[cfg(not(coverage))] // unreachable: line 315 proved every digit is
335    // ascii, so the date-part parses cannot fail
336    let year: i32 = digits[..4].parse().ok()?;
337    #[cfg(coverage)]
338    let year: i32 = digits[..4].parse().expect("the digits are validated above");
339    #[cfg(not(coverage))] // unreachable: the month digits are ascii
340    let month: u32 = digits[4..6].parse().ok()?;
341    #[cfg(coverage)]
342    let month: u32 = digits[4..6]
343        .parse()
344        .expect("the digits are validated above");
345    #[cfg(not(coverage))] // unreachable: the day digits are ascii
346    let day: u32 = digits[6..8].parse().ok()?;
347    #[cfg(coverage)]
348    let day: u32 = digits[6..8]
349        .parse()
350        .expect("the digits are validated above");
351    chrono::NaiveDate::from_ymd_opt(year, month, day)?;
352    // A future date does not count (the reference rejects it).
353    let today = chrono::Utc::now().date_naive();
354    #[cfg(not(coverage))] // unreachable: the same year/month/day just
355    // constructed a date two lines above, so this cannot fail
356    let parsed = chrono::NaiveDate::from_ymd_opt(year, month, day)?;
357    #[cfg(coverage)]
358    let parsed =
359        chrono::NaiveDate::from_ymd_opt(year, month, day).expect("the date is validated above");
360    if parsed > today {
361        return None;
362    }
363    Some((year, month, day))
364}
365
366/// The failover-link arm (`_extract_failover_link_url`): the first
367/// `<a ... id="failoverLink" ...>` anchor's href, trusted URLs only.
368#[must_use]
369pub fn extract_failover_link_url(decoded_html: &str) -> Option<String> {
370    let anchor = find_anchor_with(decoded_html, "failoverLink")?;
371    let href = attribute_value(anchor, "href")?;
372    is_trusted_azure_download_url(&href).then_some(href)
373}
374
375/// The newest-`ServiceTags` arm (`_extract_newest_service_tags_url`): every
376/// URL matching the `ServiceTags` shape on the trusted host, the winner by
377/// `(has date, date, url)`.
378#[must_use]
379pub fn extract_newest_service_tags_url(decoded_html: &str) -> Option<AzureDownloadCandidate> {
380    let mut best: Option<AzureDownloadCandidate> = None;
381    for (start, candidate_url) in service_tags_urls(decoded_html) {
382        if !is_trusted_azure_download_url(&candidate_url) {
383            continue;
384        }
385        let date = parse_service_tags_date(&candidate_url);
386        let candidate = AzureDownloadCandidate {
387            url: candidate_url,
388            date,
389        };
390        let better = best
391            .as_ref()
392            .is_none_or(|current| candidate_key(&candidate) > candidate_key(current));
393        if better {
394            best = Some(candidate);
395        }
396        let _ = start;
397    }
398    best
399}
400
401/// The generic-json arm (`_extract_generic_json_url`): the first
402/// `https://download.microsoft.com/...json` href.
403#[must_use]
404pub fn extract_generic_json_url(decoded_html: &str) -> Option<String> {
405    let mut rest = decoded_html;
406    while let Some(position) = rest.find("href=") {
407        rest = &rest[position + "href=".len()..];
408        let Some(quoted) = rest.strip_prefix(['"', '\'']) else {
409            continue;
410        };
411        let Some(end) = quoted.find(['"', '\'']) else {
412            continue;
413        };
414        let url = &quoted[..end];
415        // The reference pattern matches `.json` case-sensitively.
416        #[allow(clippy::case_sensitive_file_extension_comparisons)]
417        let json_shaped = url.ends_with(".json") || url.contains(".json?");
418        if url.starts_with("https://download.microsoft.com/")
419            && json_shaped
420            && is_trusted_azure_download_url(url)
421        {
422            return Some(url.to_owned());
423        }
424    }
425    None
426}
427
428/// The full extraction (`_extract_azure_download_url`): failover link,
429/// then the newest `ServiceTags` URL, then any trusted JSON href.
430#[must_use]
431pub fn extract_azure_download_url(decoded_html: &str) -> Option<String> {
432    extract_failover_link_url(decoded_html)
433        .or_else(|| extract_newest_service_tags_url(decoded_html).map(|candidate| candidate.url))
434        .or_else(|| extract_generic_json_url(decoded_html))
435}
436
437/// The candidate ordering key (`_service_tags_sort_key`):
438/// `(has date, date, url)`.
439fn candidate_key(candidate: &AzureDownloadCandidate) -> (bool, (i32, u32, u32), &str) {
440    (
441        candidate.date.is_some(),
442        candidate.date.unwrap_or((0, 1, 1)),
443        candidate.url.as_str(),
444    )
445}
446
447/// Every `ServiceTags`-shaped URL in the page, in page order.
448fn service_tags_urls(html: &str) -> Vec<(usize, String)> {
449    let mut found = Vec::new();
450    let mut rest = html;
451    let mut offset = 0;
452    while let Some(position) = rest.find("https://download.") {
453        let tail = &rest[position..];
454        let end = tail
455            .find(['"', '\'', '<', '>', ' ', '\n', '\r', '\t'])
456            .unwrap_or(tail.len());
457        let candidate = &tail[..end];
458        let cut = candidate
459            .find(".json")
460            .map_or(candidate.len(), |index| index + ".json".len());
461        let mut url = candidate[..cut].to_owned();
462        // A query string rides along up to the next delimiter.
463        if let Some(query) = candidate[cut..].strip_prefix('?') {
464            let query_end = query
465                .find(['"', '\'', '<', '>', ' ', '\n', '\r', '\t'])
466                .unwrap_or(query.len());
467            url.push('?');
468            url.push_str(&query[..query_end]);
469        }
470        if url.contains("ServiceTags") {
471            found.push((offset + position, url));
472        }
473        rest = &rest[position + 1.min(rest[position..].len() - 1)..];
474        offset += position + 1.min(rest.len());
475    }
476    found
477}
478
479/// Find the first `<a ...>` tag whose attributes carry `needle`.
480fn find_anchor_with<'a>(html: &'a str, needle: &str) -> Option<&'a str> {
481    let mut rest = html;
482    while let Some(start) = rest.find("<a") {
483        let after = &rest[start..];
484        let end = after.find('>')?;
485        let tag = &after[..=end];
486        if tag.contains(needle) {
487            return Some(tag);
488        }
489        rest = &after[end..];
490    }
491    None
492}
493
494/// The value of `name="..."` / `name='...'` inside one tag.
495fn attribute_value(tag: &str, name: &str) -> Option<String> {
496    let position = tag.find(name)?;
497    let rest = &tag[position + name.len()..];
498    let rest = rest.strip_prefix('=')?;
499    #[cfg(not(coverage))] // unreachable: the tag always ends with `>`, so
500    // a `href=` inside it is never the tag's last character and at least
501    // one character follows
502    let quote = rest.chars().next()?;
503    #[cfg(coverage)]
504    let quote = rest
505        .chars()
506        .next()
507        .expect("the tag continues past the attribute");
508    #[cfg(not(coverage))] // unreachable: the quote character is the first
509    // character of the remainder, so stripping it cannot fail
510    let rest = rest.strip_prefix(quote)?;
511    #[cfg(coverage)]
512    let rest = &rest[quote.len_utf8()..];
513    let end = rest.find(quote)?;
514    Some(rest[..end].to_owned())
515}
516
517/// The store seam behind the cache layers (the reference
518/// `CloudIpStoreProtocol`; the Redis shape is the section 08
519/// `cloud_ip_v2` / `cloud_ranges_v2` payload, the in-memory shape is a
520/// map).
521pub trait CloudRangeStore: Send + Sync {
522    /// The cached entries for `provider`: `Ok(None)` is a cache miss
523    /// ("unknown, refresh eligible"), `Ok(Some(..))` a hit ("known") -
524    /// the miss/empty distinction is normative.
525    ///
526    /// # Errors
527    ///
528    /// [`StoreError`](crate::distributed::StoreError) on a backend failure.
529    fn get_ranges(
530        &self,
531        provider: &str,
532    ) -> Result<Option<ParsedRanges>, crate::distributed::StoreError>;
533
534    /// Cache `entries` for `provider` (`ttl = None` persists).
535    ///
536    /// # Errors
537    ///
538    /// [`StoreError`](crate::distributed::StoreError) on a backend failure.
539    fn set_ranges(
540        &self,
541        provider: &str,
542        entries: &[(String, Option<String>)],
543        ttl: Option<u64>,
544    ) -> Result<(), crate::distributed::StoreError>;
545}
546
547/// What one [`refresh_provider_from_store`] pass decided.
548#[derive(Debug, Clone, Copy, PartialEq, Eq)]
549pub enum CloudRefreshOutcome {
550    /// The store answered: the cached entries were installed in memory,
551    /// no network fetch ran.
552    FromCache {
553        /// How many entries the cache carried.
554        entries: usize,
555    },
556    /// The cache missed, the fetch returned ranges: they were cached and
557    /// installed.
558    Fetched {
559        /// How many ranges the fetch produced.
560        entries: usize,
561    },
562    /// The cache missed and the fetch produced nothing: nothing was
563    /// cached and the provider's previous ranges (if any) stay.
564    FetchEmpty,
565}
566
567/// One provider's refresh pass over the reference cache layers:
568/// the store first (a hit installs, no fetch), then the fetcher (only a
569/// non-empty result is cached and installed).
570pub fn refresh_provider_from_store(
571    table: &crate::cloud_provider::CloudIpTable,
572    store: &dyn CloudRangeStore,
573    provider: &str,
574    fetch: impl FnOnce() -> ParsedRanges,
575    ttl_seconds: Option<u64>,
576) -> CloudRefreshOutcome {
577    if let Ok(Some(cached)) = store.get_ranges(provider) {
578        let count = cached.len();
579        let _ = table.set_provider_ranges(provider, cached);
580        return CloudRefreshOutcome::FromCache { entries: count };
581    }
582    let fetched = fetch();
583    if fetched.is_empty() {
584        // An empty fetch is never cached and never touches the provider.
585        return CloudRefreshOutcome::FetchEmpty;
586    }
587    let count = fetched.len();
588    if let Err(error) = store.set_ranges(provider, &fetched, ttl_seconds) {
589        // A cache write failure does not undo the install (the reference
590        // logs and keeps the in-memory result).
591        let _ = error;
592    }
593    let _ = table.set_provider_ranges(provider, fetched);
594    CloudRefreshOutcome::Fetched { entries: count }
595}
596
597/// The refresh stamp plus the single-flight gate (the reference
598/// `CloudIpRefreshCheck` stamp and `CloudManager.schedule_refresh`).
599///
600/// * `should_refresh(now, interval)` is the reference gate
601///   `now - last > interval` (the default stamp 0 makes the first check
602///   fire).
603/// * `stamp(now)` / `restore(previous)` are the optimistic advance and
604///   the rollback the check runs when scheduling reports failure.
605/// * `schedule_refresh(job)` is single-flight: while a refresh is in
606///   flight, further calls are no-ops returning `false`; at most one
607///   refresh runs at a time. The job runs on its own thread (the
608///   reference schedules a background task) and the gate clears when the
609///   job returns, panic or not.
610pub struct CloudRefreshCoordinator {
611    last_refresh_seconds: AtomicI64,
612    in_flight: Arc<AtomicBool>,
613}
614
615impl Default for CloudRefreshCoordinator {
616    fn default() -> Self {
617        Self::new()
618    }
619}
620
621impl Clone for CloudRefreshCoordinator {
622    /// A clone shares the stamp and the gate (the reference manager's
623    /// singleton semantics).
624    fn clone(&self) -> Self {
625        Self {
626            last_refresh_seconds: AtomicI64::new(self.last_refresh_seconds.load(Ordering::Relaxed)),
627            in_flight: Arc::clone(&self.in_flight),
628        }
629    }
630}
631
632impl core::fmt::Debug for CloudRefreshCoordinator {
633    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
634        f.debug_struct("CloudRefreshCoordinator")
635            .field(
636                "last_refresh_seconds",
637                &self.last_refresh_seconds.load(Ordering::Relaxed),
638            )
639            .field("in_flight", &self.in_flight.load(Ordering::Relaxed))
640            .finish()
641    }
642}
643
644impl CloudRefreshCoordinator {
645    /// A coordinator whose stamp says never refreshed (the reference
646    /// default: the first check fires).
647    #[must_use]
648    pub fn new() -> Self {
649        Self {
650            last_refresh_seconds: AtomicI64::new(0),
651            in_flight: Arc::new(AtomicBool::new(false)),
652        }
653    }
654
655    /// The reference gate: `now - last > interval`.
656    #[must_use]
657    pub fn should_refresh(&self, now_seconds: i64, interval_seconds: u64) -> bool {
658        now_seconds.saturating_sub(self.last_refresh_seconds.load(Ordering::Relaxed))
659            > i64::try_from(interval_seconds).unwrap_or(i64::MAX)
660    }
661
662    /// The current stamp (the rollback seam reads it first).
663    #[must_use]
664    pub fn stamp_value(&self) -> i64 {
665        self.last_refresh_seconds.load(Ordering::Relaxed)
666    }
667
668    /// Advance the stamp optimistically (the reference's
669    /// `int(time.time())` write before scheduling).
670    pub fn stamp(&self, now_seconds: i64) {
671        self.last_refresh_seconds
672            .store(now_seconds, Ordering::Relaxed);
673    }
674
675    /// Roll the stamp back (the reference's arm when scheduling failed).
676    pub fn restore(&self, previous_seconds: i64) {
677        self.last_refresh_seconds
678            .store(previous_seconds, Ordering::Relaxed);
679    }
680
681    /// Schedule `job` in the background unless one is already in flight:
682    /// `true` when the job started, `false` when the single-flight gate
683    /// swallowed the call.
684    pub fn schedule_refresh(&self, job: impl FnOnce() + Send + 'static) -> bool {
685        if self
686            .in_flight
687            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
688            .is_err()
689        {
690            return false;
691        }
692        let in_flight = Arc::clone(&self.in_flight);
693        let started = std::thread::spawn(move || {
694            let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job));
695            in_flight.store(false, Ordering::Release);
696        });
697        let _ = started;
698        true
699    }
700
701    /// Whether a refresh is currently in flight (observability seam).
702    #[must_use]
703    pub fn is_in_flight(&self) -> bool {
704        self.in_flight.load(Ordering::Acquire)
705    }
706}
707
708#[cfg(test)]
709mod tests {
710    use super::*;
711    use crate::cloud_provider::CloudIpTable;
712    use std::str::FromStr;
713    use std::sync::Mutex;
714
715    #[test]
716    fn aws_parses_only_the_amazon_service_entries() {
717        let payload = r#"{"prefixes": [
718            {"ip_prefix": "203.0.113.0/24", "region": "us-east-1", "service": "AMAZON"},
719            {"ip_prefix": "198.51.100.0/24", "region": "eu-west-1", "service": "EC2"},
720            {"ip_prefix": "192.0.2.0/24", "region": "us-west-2", "service": "AMAZON"}
721        ]}"#;
722        let ranges = parse_aws_ranges(payload).expect("valid");
723        assert_eq!(
724            ranges,
725            vec![
726                (
727                    String::from("203.0.113.0/24"),
728                    Some(String::from("us-east-1"))
729                ),
730                (
731                    String::from("192.0.2.0/24"),
732                    Some(String::from("us-west-2"))
733                ),
734            ]
735        );
736    }
737
738    #[test]
739    fn a_malformed_amazon_prefix_aborts_the_whole_parse() {
740        // The reference `ipaddress.ip_network(prefix)` raise empties the
741        // whole fetch, so a malformed upstream entry never silently
742        // shrinks the provider's cached range set.
743        let payload = r#"{"prefixes": [
744            {"ip_prefix": "203.0.113.0/24", "region": "us-east-1", "service": "AMAZON"},
745            {"ip_prefix": "not-a-cidr", "region": "x", "service": "AMAZON"}
746        ]}"#;
747        let error = parse_aws_ranges(payload).unwrap_err();
748        assert!(error.reason.contains("not-a-cidr"), "{error}");
749    }
750
751    #[test]
752    fn a_malformed_gcp_prefix_aborts_the_whole_parse() {
753        let payload = r#"{"prefixes": [
754            {"ipv4Prefix": "203.0.113.0/24", "scope": "us-central1"},
755            {"ipv4Prefix": "bogus", "scope": "x"}
756        ]}"#;
757        let error = parse_gcp_ranges(payload).unwrap_err();
758        assert!(error.reason.contains("bogus"), "{error}");
759    }
760
761    #[test]
762    fn aws_missing_prefixes_array_is_a_payload_error() {
763        let error = parse_aws_ranges(r#"{"syncToken": 1}"#).unwrap_err();
764        assert!(error.reason.contains("prefixes"));
765    }
766
767    #[test]
768    fn aws_error_arms_cover_the_payload_shapes() {
769        // The Display the fetchers log.
770        let error = PayloadError {
771            reason: String::from("missing prefixes array"),
772        };
773        assert_eq!(
774            error.to_string(),
775            "provider payload error: missing prefixes array"
776        );
777
778        // A body that is not JSON at all.
779        let error = parse_aws_ranges("not json").unwrap_err();
780        assert!(error.reason.contains("invalid json"));
781
782        // An AMAZON entry without a prefix field is skipped, not fatal.
783        let ranges = parse_aws_ranges(
784            r#"{"prefixes": [{"service": "AMAZON"}, {"ip_prefix": "203.0.113.0/24", "service": "AMAZON"}]}"#,
785        )
786        .expect("valid");
787        assert_eq!(ranges.len(), 1);
788
789        // A repeated network keeps its first occurrence only.
790        let ranges = parse_aws_ranges(
791            r#"{"prefixes": [
792                {"ip_prefix": "203.0.113.0/24", "region": "a", "service": "AMAZON"},
793                {"ip_prefix": "203.0.113.0/24", "region": "b", "service": "AMAZON"}
794            ]}"#,
795        )
796        .expect("valid");
797        assert_eq!(
798            ranges,
799            vec![(String::from("203.0.113.0/24"), Some(String::from("a")))]
800        );
801    }
802
803    #[test]
804    fn gcp_error_arms_and_dedup_cover_the_payload_shapes() {
805        let error = parse_gcp_ranges("not json").unwrap_err();
806        assert!(error.reason.contains("invalid json"));
807
808        let error = parse_gcp_ranges(r#"{"syncToken": 1}"#).unwrap_err();
809        assert!(error.reason.contains("prefixes"));
810
811        let ranges = parse_gcp_ranges(
812            r#"{"prefixes": [
813                {"ipv4Prefix": "203.0.113.0/24", "scope": "a"},
814                {"ipv4Prefix": "203.0.113.0/24", "scope": "b"}
815            ]}"#,
816        )
817        .expect("valid");
818        assert_eq!(
819            ranges,
820            vec![(String::from("203.0.113.0/24"), Some(String::from("a")))]
821        );
822    }
823
824    #[test]
825    fn csv_skips_a_blank_leading_field() {
826        let ranges = parse_csv_ranges(",\n203.0.113.0/24,x\n,junk\n");
827        assert_eq!(ranges, vec![(String::from("203.0.113.0/24"), None)]);
828    }
829
830    #[test]
831    fn vultr_without_a_subnets_array_answers_empty() {
832        let ranges = parse_vultr_ranges(r#"{"syncToken": 1}"#).expect("tolerated");
833        assert!(ranges.is_empty());
834    }
835
836    #[test]
837    fn azure_error_arms_cover_the_payload_shapes() {
838        let error = parse_azure_service_tags("not json").unwrap_err();
839        assert!(error.reason.contains("invalid json"));
840
841        let error = parse_azure_service_tags(r#"{"syncToken": 1}"#).unwrap_err();
842        assert!(error.reason.contains("values"));
843
844        // A non-string prefix entry is skipped, a malformed string prefix
845        // fails the whole parse (the reference raise), and a repeated
846        // prefix keeps its first occurrence.
847        let ranges = parse_azure_service_tags(
848            r#"{"values": [{"name": "AzureCloud", "properties": {"addressPrefixes": [
849                17, "203.0.113.0/24", "203.0.113.0/24"
850            ]}}]}"#,
851        )
852        .expect("valid");
853        assert_eq!(ranges, vec![(String::from("203.0.113.0/24"), None)]);
854
855        let error = parse_azure_service_tags(
856            r#"{"values": [{"name": "AzureCloud", "properties": {"addressPrefixes": ["junk"]}}]}"#,
857        )
858        .unwrap_err();
859        assert!(error.reason.contains("invalid AzureCloud prefix"));
860    }
861
862    #[test]
863    fn service_tags_dates_reject_junk_and_the_future() {
864        assert_eq!(
865            parse_service_tags_date(
866                "https://download.microsoft.com/ServiceTags_Public_2026abcd.json"
867            ),
868            None
869        );
870        assert_eq!(
871            parse_service_tags_date(
872                "https://download.microsoft.com/ServiceTags_Public_29991231.json"
873            ),
874            None
875        );
876        assert_eq!(
877            parse_service_tags_date(
878                "https://download.microsoft.com/ServiceTags_Public_20260101.json"
879            ),
880            Some((2026, 1, 1))
881        );
882    }
883
884    #[test]
885    fn the_newest_service_tags_url_skips_off_host_and_prefers_newest() {
886        let page = concat!(
887            "see https://download.evil.test/ServiceTags_Public_20260301.json and ",
888            "https://download.microsoft.com/ServiceTags_Public_20260101.json then ",
889            "https://download.microsoft.com/ServiceTags_Public_20260201.json and ",
890            "https://download.microsoft.com/ServiceTags_Public_20251231.json done"
891        );
892        let best = extract_newest_service_tags_url(page).expect("a trusted candidate");
893        assert_eq!(
894            best.url,
895            "https://download.microsoft.com/ServiceTags_Public_20260201.json"
896        );
897        assert_eq!(best.date, Some((2026, 2, 1)));
898    }
899
900    #[test]
901    fn the_generic_json_arm_needs_a_quoted_href() {
902        // No quote after href=: the candidate never forms.
903        assert_eq!(
904            extract_generic_json_url("<a href=https://download.microsoft.com/a.json>x</a>"),
905            None
906        );
907        // An unterminated quote never closes: the candidate never forms.
908        assert_eq!(
909            extract_generic_json_url(r#"<a href="https://download.microsoft.com/a.json>x</a>"#),
910            None
911        );
912        // A quoted href on an untrusted host: parsed, then rejected.
913        assert_eq!(
914            extract_generic_json_url(r#"<a href="http://evil.test/a.json">x</a>"#),
915            None
916        );
917        assert_eq!(
918            extract_generic_json_url(r#"<a href="https://download.microsoft.com/a.json">x</a>"#)
919                .as_deref(),
920            Some("https://download.microsoft.com/a.json")
921        );
922    }
923
924    #[test]
925    fn azure_extraction_edge_cases_stay_total() {
926        // An `<a` tag that never closes: no anchor, no URL.
927        assert_eq!(
928            extract_azure_download_url(r#"<a href="https://download.microsoft.com/a.json"#),
929            None
930        );
931
932        // A failover link on an untrusted host is rejected.
933        assert_eq!(
934            extract_failover_link_url(
935                r#"<a id="failoverLink" href="http://download.microsoft.com.evil.test/x.json">x</a>"#
936            ),
937            None
938        );
939
940        // A `ServiceTags_Public_` tail without the eight date digits: no
941        // date, no candidate.
942        assert_eq!(
943            parse_service_tags_date("https://download.microsoft.com/ServiceTags_Public_"),
944            None
945        );
946        let dateless = extract_newest_service_tags_url(
947            "see https://download.microsoft.com/ServiceTags_Public_ tail",
948        )
949        .expect("a dateless candidate is still a candidate");
950        assert_eq!(dateless.date, None);
951
952        // A ServiceTags URL with a query string rides the query along.
953        let best = extract_newest_service_tags_url(
954            "see https://download.microsoft.com/ServiceTags_Public_20260101.json?sv=1 end",
955        )
956        .expect("a candidate");
957        assert_eq!(
958            best.url,
959            "https://download.microsoft.com/ServiceTags_Public_20260101.json?sv=1"
960        );
961        assert_eq!(best.date, Some((2026, 1, 1)));
962    }
963
964    #[test]
965    fn malformed_failover_anchors_answer_none_without_panicking() {
966        // An anchor with no href attribute at all.
967        assert_eq!(
968            extract_failover_link_url(r#"<a id="failoverLink">x</a>"#),
969            None
970        );
971        // `href` found inside another attribute, not followed by `=`.
972        assert_eq!(
973            extract_failover_link_url(r#"<a id="failoverLink" data-hrefx="1">x</a>"#),
974            None
975        );
976        // An unquoted href value: the quote-run never closes.
977        assert_eq!(
978            extract_failover_link_url(r#"<a id="failoverLink" href=x>y</a>"#),
979            None
980        );
981    }
982
983    #[test]
984    fn the_coordinator_clones_shares_the_gate_and_displays() {
985        let coordinator = CloudRefreshCoordinator::default();
986        assert!(!coordinator.is_in_flight());
987        assert!(
988            coordinator.should_refresh(1_000, 60),
989            "the default stamp fires"
990        );
991        coordinator.stamp(1_000);
992        assert!(!coordinator.should_refresh(1_000, 60));
993        let clone = coordinator.clone();
994        assert_eq!(clone.stamp_value(), 1_000, "the clone copies the stamp");
995        assert!(!clone.should_refresh(1_000, 60));
996        clone.restore(0);
997        assert_eq!(clone.stamp_value(), 0);
998
999        // The in-flight gate is shared with the clone: a refresh scheduled
1000        // on the original is visible (and single-flight) through the clone.
1001        // The loop builds one job type: scheduled once, then refused via
1002        // the clone.
1003        let mut releases = Vec::new();
1004        let mut first = true;
1005        for _ in 0..2 {
1006            let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
1007            releases.push(release_tx);
1008            let started = coordinator.schedule_refresh(move || {
1009                let _ = release_rx.recv();
1010            });
1011            if first {
1012                assert!(started, "the first call schedules");
1013                assert!(coordinator.is_in_flight());
1014                first = false;
1015            } else {
1016                assert!(!started, "the gate stays single-flight");
1017            }
1018        }
1019        assert!(clone.is_in_flight(), "the gate is shared with the clone");
1020        for release_tx in releases {
1021            let _ = release_tx.send(());
1022        }
1023        // The debug shape names the struct and both fields.
1024        let rendered = format!("{coordinator:?}");
1025        assert!(rendered.starts_with("CloudRefreshCoordinator"));
1026        assert!(rendered.contains("last_refresh_seconds"));
1027        assert!(rendered.contains("in_flight"));
1028    }
1029
1030    #[test]
1031    fn gcp_reads_both_prefix_fields_and_scope() {
1032        let payload = r#"{"prefixes": [
1033            {"ipv4Prefix": "203.0.113.0/24", "scope": "us-central1"},
1034            {"ipv6Prefix": "2001:db8::/32", "scope": "europe-west1"},
1035            {"scope": "no-prefix"}
1036        ]}"#;
1037        let ranges = parse_gcp_ranges(payload).expect("valid");
1038        assert_eq!(
1039            ranges,
1040            vec![
1041                (
1042                    String::from("203.0.113.0/24"),
1043                    Some(String::from("us-central1"))
1044                ),
1045                (
1046                    String::from("2001:db8::/32"),
1047                    Some(String::from("europe-west1")),
1048                ),
1049            ]
1050        );
1051    }
1052
1053    #[test]
1054    fn csv_skips_comments_and_invalid_prefixes() {
1055        let body = "# comment line\n\
1056                    203.0.113.0/24,region-one,extra\n\
1057                    \n\
1058                    not-a-cidr,ignored\n\
1059                    198.51.100.0/24\n";
1060        assert_eq!(
1061            parse_csv_ranges(body),
1062            vec![
1063                (String::from("203.0.113.0/24"), None),
1064                (String::from("198.51.100.0/24"), None),
1065            ]
1066        );
1067    }
1068
1069    #[test]
1070    fn vultr_reads_subnets_and_skips_invalid() {
1071        let payload = r#"{"subnets": [
1072            {"ip_prefix": "203.0.113.0/24"},
1073            {"ip_prefix": "junk"},
1074            {"other": 1}
1075        ]}"#;
1076        assert_eq!(
1077            parse_vultr_ranges(payload).expect("valid"),
1078            vec![(String::from("203.0.113.0/24"), None)]
1079        );
1080        assert!(parse_vultr_ranges("not json").is_err());
1081    }
1082
1083    #[test]
1084    fn azure_reads_the_azurecloud_tag_only() {
1085        let payload = r#"{"values": [
1086            {"name": "AzureCloud", "properties": {"addressPrefixes": [
1087                "203.0.113.0/24", "2001:db8::/32", "198.51.100.7/24"
1088            ]}},
1089            {"name": "Other", "properties": {"addressPrefixes": ["10.0.0.0/8"]}}
1090        ]}"#;
1091        let ranges = parse_azure_service_tags(payload).expect("valid");
1092        assert_eq!(
1093            ranges,
1094            vec![
1095                (String::from("203.0.113.0/24"), None),
1096                (String::from("2001:db8::/32"), None),
1097                (String::from("198.51.100.0/24"), None),
1098            ],
1099            "host bits clear exactly like the reference ip_network parse"
1100        );
1101    }
1102
1103    #[test]
1104    fn azure_missing_tag_and_bad_prefix_fail() {
1105        let error = parse_azure_service_tags(r#"{"values": [{"name": "Other"}]}"#).unwrap_err();
1106        assert!(error.reason.contains("AzureCloud"));
1107        let error = parse_azure_service_tags(
1108            r#"{"values": [{"name": "AzureCloud", "properties": {"addressPrefixes": ["junk"]}}]}"#,
1109        )
1110        .unwrap_err();
1111        assert!(error.reason.contains("invalid AzureCloud prefix"));
1112    }
1113
1114    #[test]
1115    fn trusted_url_check_demands_https_on_the_download_host() {
1116        assert!(is_trusted_azure_download_url(
1117            "https://download.microsoft.com/x/ServiceTags_Public_20260101.json"
1118        ));
1119        assert!(!is_trusted_azure_download_url(
1120            "http://download.microsoft.com/x.json"
1121        ));
1122        assert!(!is_trusted_azure_download_url("https://evil.test/x.json"));
1123        assert!(!is_trusted_azure_download_url(
1124            "https://download.microsoft.com.evil.test/x.json"
1125        ));
1126    }
1127
1128    #[test]
1129    fn failover_link_wins_over_everything() {
1130        let page = r#"<html>see <a id="failoverLink" href="https://download.microsoft.com/f/ServiceTags_Public_20200101.json">here</a>
1131            and https://download.microsoft.com/n/ServiceTags_Public_20261231.json text</html>"#;
1132        assert_eq!(
1133            extract_azure_download_url(page).as_deref(),
1134            Some("https://download.microsoft.com/f/ServiceTags_Public_20200101.json")
1135        );
1136    }
1137
1138    #[test]
1139    fn an_off_host_failover_link_falls_through_to_the_newest_tags() {
1140        let page = r#"<a id="failoverLink" href="https://evil.test/steal.json">x</a>
1141            pick https://download.microsoft.com/a/ServiceTags_Public_20250505.json or
1142            https://download.microsoft.com/b/ServiceTags_Public_20260606.json thanks"#;
1143        assert_eq!(
1144            extract_azure_download_url(page).as_deref(),
1145            Some("https://download.microsoft.com/b/ServiceTags_Public_20260606.json"),
1146            "the newest `ServiceTags` date wins"
1147        );
1148    }
1149
1150    #[test]
1151    fn a_dateless_url_loses_to_a_dated_one_and_wins_by_text_otherwise() {
1152        let page = "https://download.microsoft.com/x/ServiceTags_no_date.json and \
1153                    https://download.microsoft.com/y/ServiceTags_Public_20260303.json end";
1154        assert_eq!(
1155            extract_newest_service_tags_url(page)
1156                .expect("candidate")
1157                .url,
1158            "https://download.microsoft.com/y/ServiceTags_Public_20260303.json"
1159        );
1160        let page = "https://download.microsoft.com/aaa/ServiceTags_b.json vs \
1161                    https://download.microsoft.com/bbb/ServiceTags_a.json end";
1162        let winner = extract_newest_service_tags_url(page).expect("candidate");
1163        assert_eq!(
1164            winner.url, "https://download.microsoft.com/bbb/ServiceTags_a.json",
1165            "both dateless: the lexicographic URL tiebreak (the reference \
1166             _service_tags_sort_key ordering)"
1167        );
1168        assert_eq!(winner.date, None);
1169    }
1170
1171    #[test]
1172    fn no_candidate_falls_to_the_generic_json_href() {
1173        let page = r#"<a href="https://download.microsoft.com/generic/thing.json?sv=1">x</a>"#;
1174        assert_eq!(
1175            extract_azure_download_url(page).as_deref(),
1176            Some("https://download.microsoft.com/generic/thing.json?sv=1")
1177        );
1178        assert_eq!(extract_azure_download_url("<p>nothing</p>"), None);
1179    }
1180
1181    #[test]
1182    fn html_entities_decode_before_extraction() {
1183        let page =
1184            "see &quot;https://download.microsoft.com/q/ServiceTags_Public_20260202.json&quot; end";
1185        let decoded = html_escape::decode_html_entities(page).to_string();
1186        assert_eq!(
1187            extract_azure_download_url(&decoded).as_deref(),
1188            Some("https://download.microsoft.com/q/ServiceTags_Public_20260202.json")
1189        );
1190    }
1191
1192    #[test]
1193    fn service_tag_dates_reject_future_and_invalid() {
1194        // A well-formed past date parses.
1195        assert!(
1196            parse_service_tags_date(
1197                "https://download.microsoft.com/ServiceTags_Public_20240101.json"
1198            )
1199            .is_some()
1200        );
1201        // Garbage digits and impossible dates do not.
1202        assert_eq!(
1203            parse_service_tags_date(
1204                "https://download.microsoft.com/ServiceTags_Public_20249999.json"
1205            ),
1206            None
1207        );
1208        assert_eq!(
1209            parse_service_tags_date("https://download.microsoft.com/ServiceTags_x.json"),
1210            None
1211        );
1212    }
1213
1214    // ---- Store-backed cache layers ----
1215
1216    #[derive(Default)]
1217    struct MemoryCloudStore {
1218        entries: Mutex<std::collections::HashMap<String, ParsedRanges>>,
1219    }
1220
1221    impl CloudRangeStore for MemoryCloudStore {
1222        fn get_ranges(
1223            &self,
1224            provider: &str,
1225        ) -> Result<Option<ParsedRanges>, crate::distributed::StoreError> {
1226            Ok(self.entries.lock().expect("entries").get(provider).cloned())
1227        }
1228
1229        fn set_ranges(
1230            &self,
1231            provider: &str,
1232            entries: &[(String, Option<String>)],
1233            _ttl: Option<u64>,
1234        ) -> Result<(), crate::distributed::StoreError> {
1235            self.entries
1236                .lock()
1237                .expect("entries")
1238                .insert(provider.to_owned(), entries.to_vec());
1239            Ok(())
1240        }
1241    }
1242
1243    /// A failing-setRanges store: every write is a backend error.
1244    struct WriteFailsStore;
1245
1246    impl CloudRangeStore for WriteFailsStore {
1247        fn get_ranges(
1248            &self,
1249            _provider: &str,
1250        ) -> Result<Option<ParsedRanges>, crate::distributed::StoreError> {
1251            Ok(None)
1252        }
1253
1254        fn set_ranges(
1255            &self,
1256            _provider: &str,
1257            _entries: &[(String, Option<String>)],
1258            _ttl: Option<u64>,
1259        ) -> Result<(), crate::distributed::StoreError> {
1260            Err(crate::distributed::StoreError(String::from(
1261                "write refused",
1262            )))
1263        }
1264    }
1265
1266    /// One fetch closure every refresh path shares: the first two calls
1267    /// yield a marker range, later calls yield nothing. One closure value
1268    /// (shared by reference) means one instantiation, so the hit, fetched,
1269    /// empty, and failed cache-write passes all drive the same copy.
1270    #[derive(Default)]
1271    struct StagedFetch {
1272        calls: std::cell::Cell<usize>,
1273    }
1274
1275    impl StagedFetch {
1276        fn fetch(&self) -> ParsedRanges {
1277            self.calls.set(self.calls.get() + 1);
1278            if self.calls.get() <= 2 {
1279                vec![(String::from("192.0.2.0/24"), None)]
1280            } else {
1281                Vec::new()
1282            }
1283        }
1284    }
1285
1286    /// A job body with no lines: the single-flight refusals pass the same
1287    /// fn item the scheduled calls run, so the refused copies never carry
1288    /// an uncovered closure body.
1289    fn noop_job() {}
1290
1291    /// One shared fetch closure type for every refresh pass: the first two
1292    /// calls yield the marker range, later calls yield nothing. Rebuilding
1293    /// the closure through one constructor keeps a single instantiation of
1294    /// `refresh_provider_from_store` across all four scenarios.
1295    fn make_fetch(staged: &StagedFetch) -> impl FnOnce() -> ParsedRanges + '_ {
1296        move || staged.fetch()
1297    }
1298
1299    #[test]
1300    fn the_refresh_pass_walks_hit_miss_write_failure_and_empty() {
1301        let staged = StagedFetch::default();
1302
1303        // The hit: the cached entries install, the fetch never runs.
1304        let table = CloudIpTable::default();
1305        let store = MemoryCloudStore::default();
1306        store
1307            .set_ranges("AWS", &[(String::from("203.0.113.0/24"), None)], None)
1308            .expect("seed");
1309        let outcome =
1310            refresh_provider_from_store(&table, &store, "AWS", make_fetch(&staged), Some(3_600));
1311        assert_eq!(outcome, CloudRefreshOutcome::FromCache { entries: 1 });
1312        assert!(table.is_cloud_ip(
1313            std::net::IpAddr::from_str("203.0.113.9").expect("ip"),
1314            &crate::cloud_provider::parse_cloud_selectors(["AWS"]).expect("selectors"),
1315        ),);
1316        assert!(
1317            !table.is_cloud_ip(
1318                std::net::IpAddr::from_str("192.0.2.9").expect("ip"),
1319                &crate::cloud_provider::parse_cloud_selectors(["AWS"]).expect("selectors"),
1320            ),
1321            "a cache hit must never fetch: the marker range is absent"
1322        );
1323
1324        // The miss: the fetch runs and only a non-empty result is cached.
1325        let table = CloudIpTable::default();
1326        let store = MemoryCloudStore::default();
1327        let outcome =
1328            refresh_provider_from_store(&table, &store, "GCP", make_fetch(&staged), Some(3_600));
1329        assert_eq!(outcome, CloudRefreshOutcome::Fetched { entries: 1 });
1330        assert!(table.is_cloud_ip(
1331            std::net::IpAddr::from_str("192.0.2.9").expect("ip"),
1332            &crate::cloud_provider::parse_cloud_selectors(["GCP"]).expect("selectors"),
1333        ),);
1334
1335        // A failed cache write does not undo the install.
1336        let table = CloudIpTable::default();
1337        let outcome = refresh_provider_from_store(
1338            &table,
1339            &WriteFailsStore,
1340            "Vultr",
1341            make_fetch(&staged),
1342            None,
1343        );
1344        assert_eq!(outcome, CloudRefreshOutcome::Fetched { entries: 1 });
1345        assert!(
1346            table.is_cloud_ip(
1347                std::net::IpAddr::from_str("192.0.2.9").expect("ip"),
1348                &crate::cloud_provider::parse_cloud_selectors(["Vultr"]).expect("selectors"),
1349            ),
1350            "the install survives the cache write failure"
1351        );
1352
1353        // The empty fetch caches nothing and touches nothing.
1354        let table = CloudIpTable::default();
1355        let store = MemoryCloudStore::default();
1356        let outcome =
1357            refresh_provider_from_store(&table, &store, "Linode", make_fetch(&staged), None);
1358        assert_eq!(outcome, CloudRefreshOutcome::FetchEmpty);
1359        assert_eq!(store.get_ranges("Linode").expect("read"), None);
1360        assert!(
1361            table
1362                .provider_details(
1363                    std::net::IpAddr::from_str("192.0.2.9").expect("ip"),
1364                    &crate::cloud_provider::parse_cloud_selectors(["Linode"]).expect("selectors"),
1365                )
1366                .is_none()
1367        );
1368    }
1369
1370    // ---- Cadence and single flight ----
1371
1372    #[test]
1373    fn the_cadence_gate_fires_on_interval_and_stamps_roll_back() {
1374        let coordinator = CloudRefreshCoordinator::new();
1375        assert!(
1376            coordinator.should_refresh(1_000, 60),
1377            "the default stamp 0 makes the first check fire"
1378        );
1379
1380        // The reference check: stamp optimistically, roll back on failure.
1381        let previous = coordinator.stamp_value();
1382        coordinator.stamp(1_000);
1383        assert!(!coordinator.should_refresh(1_050, 60));
1384        coordinator.restore(previous);
1385        assert!(coordinator.should_refresh(1_050, 60));
1386    }
1387
1388    #[test]
1389    fn schedule_refresh_is_single_flight() {
1390        let coordinator = CloudRefreshCoordinator::new();
1391        let (started_tx, started_rx) = std::sync::mpsc::channel::<()>();
1392        // The loop builds one job type: the first call schedules it, the
1393        // second is refused while it is in flight.
1394        let mut first = true;
1395        let mut releases = Vec::new();
1396        for _ in 0..2 {
1397            let started_tx = started_tx.clone();
1398            let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
1399            releases.push(release_tx);
1400            let started = coordinator.schedule_refresh(move || {
1401                started_tx.send(()).expect("send");
1402                release_rx.recv().expect("release");
1403            });
1404            if first {
1405                assert!(started, "the first call schedules");
1406                started_rx
1407                    .recv_timeout(std::time::Duration::from_secs(5))
1408                    .expect("job started");
1409                first = false;
1410            } else {
1411                assert!(!started, "while in flight the call is a no-op");
1412            }
1413        }
1414        assert!(coordinator.is_in_flight());
1415        // A second job shape is refused just the same while in flight.
1416        assert!(!coordinator.schedule_refresh(noop_job));
1417
1418        // The first poll is refused while the scheduled job still waits on
1419        // its release.
1420        std::thread::sleep(std::time::Duration::from_millis(10));
1421        assert!(
1422            !coordinator.schedule_refresh(noop_job),
1423            "the in-flight gate refuses the poll"
1424        );
1425
1426        // Release the waiting job (the refused call's job never ran, so
1427        // its receiver is already gone); the gate clears when it finishes,
1428        // and a fresh job scheduling again is the proof.
1429        for release_tx in std::mem::take(&mut releases) {
1430            let _ = release_tx.send(());
1431        }
1432        let mut cleared = false;
1433        let mut polls = 0;
1434        while !cleared && polls < 500 {
1435            std::thread::sleep(std::time::Duration::from_millis(10));
1436            polls += 1;
1437            cleared = coordinator.schedule_refresh(noop_job);
1438        }
1439        assert!(cleared, "the gate clears when the job is done");
1440    }
1441
1442    fn boom_job() {
1443        panic!("fetch blew up");
1444    }
1445
1446    #[test]
1447    fn a_panicking_job_still_releases_the_gate() {
1448        let coordinator = CloudRefreshCoordinator::new();
1449        assert!(coordinator.schedule_refresh(boom_job));
1450        for _ in 0..500 {
1451            if !coordinator.is_in_flight() {
1452                break;
1453            }
1454            std::thread::sleep(std::time::Duration::from_millis(10));
1455        }
1456        assert!(!coordinator.is_in_flight());
1457
1458        // Hold the gate with a waiting job (the second copy of the same
1459        // job shape is refused while it waits), then refuse the same
1460        // panicking job shape too.
1461        let mut releases = Vec::new();
1462        let mut first = true;
1463        for _ in 0..2 {
1464            let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
1465            releases.push(release_tx);
1466            let started = coordinator.schedule_refresh(move || {
1467                let _ = release_rx.recv();
1468            });
1469            if first {
1470                assert!(started, "the holding job schedules");
1471                first = false;
1472            } else {
1473                assert!(!started, "while in flight the call is a no-op");
1474            }
1475        }
1476        assert!(!coordinator.schedule_refresh(boom_job));
1477        for release_tx in releases {
1478            let _ = release_tx.send(());
1479        }
1480    }
1481}
guard_core_engine/cloud_fetch.rs

guard_core_engine/
cloud_fetch.rs
