offline-protocol 0.30.0

//! The HTTPS uploader: one batch at a time, oldest first, with idempotent
//! retry.
//!
//! Every batch carries a `batch_id` that is reused on every resend, and the
//! ingest deduplicates on it, so a batch may be sent more than once and can
//! never be counted more than once. That is what lets the queue be
//! at-least-once: a crash between a send and its acknowledgement replays the
//! batch, and the replay is answered `200 {deduplicated: true}`.
//!
//! # What each answer means
//!
//! | Answer | Action |
//! |---|---|
//! | 2xx | The ingest owns the batch; delete it, reset the backoff |
//! | 3xx | Never followed, so it arrives as an answer; drop it, keep going |
//! | 413 | The batch can never fit; drop it, keep going |
//! | 403 with `error: telemetry_disabled` | The application's telemetry toggle is off in the developer portal; drop the batch and everything queued behind it, record the error, stop until the next `enable_telemetry`, and have the pipe discard what it collects meanwhile |
//! | any other 401, 403 | The key or the app id is wrong; keep the batch, record the error, stop until the next `enable_telemetry` |
//! | 429 | Wait `Retry-After` when given, else back off |
//! | 408, 5xx, no answer | Back off: 1 s doubling to 15 min, plus up to a second of jitter |
//! | any other 4xx | The bytes will never be accepted; drop, keep going |
//!
//! A queued batch older than six days is dropped unsent: the ingest
//! deduplicates for seven, and a batch replayed past that window would be
//! ingested twice.
//!
//! The uploader never takes the emit path's lock and holds the store only
//! while it is the only thread that could touch it.

use std::time::{Duration, Instant};

use crate::telemetry::pipe::store::BatchStore;

/// Where every batch goes. Fixed at build time; see `build.rs`.
pub(crate) const TELEMETRY_INGEST_URL: &str = env!("OFFLINE_TELEMETRY_ENDPOINT");
/// The SDK version stamped on every batch.
pub(crate) const SDK_VERSION: &str = env!("CARGO_PKG_VERSION");

/// Backoff floor on a retryable failure.
pub(crate) const BACKOFF_BASE_MS: i64 = 1_000;
/// Backoff ceiling.
pub(crate) const BACKOFF_MAX_MS: i64 = 15 * 60_000;
/// Additive jitter is drawn from `[0, BACKOFF_JITTER_MS)`.
pub(crate) const BACKOFF_JITTER_MS: u64 = 1_000;
/// A batch older than this is dropped rather than sent.
pub(crate) const MAX_BATCH_AGE_MS: i64 = 6 * 24 * 60 * 60_000;
/// Non-final flushes are deferred below this battery level unless charging.
pub(crate) const BATTERY_DEFER_BELOW: u8 = 15;

pub(crate) const HEADER_APP_ID: &str = "X-Mesh-Analytics-App-Id";
pub(crate) const HEADER_IDEMPOTENCY: &str = "Idempotency-Key";
/// The `error` code the ingest answers a 403 with while the application's
/// telemetry toggle is off in the developer portal. Every other 403 carries
/// `forbidden`, and is a bad key or a key for another app.
pub(crate) const ERROR_TELEMETRY_DISABLED: &str = "telemetry_disabled";

/// One HTTP request, as the client sees it.
pub(crate) struct Request<'a> {
    pub(crate) url: &'a str,
    pub(crate) headers: Vec<(&'static str, &'a str)>,
    pub(crate) body: &'a str,
}

/// One HTTP answer, reduced to what the uploader classifies.
#[derive(Debug, Clone)]
pub(crate) struct Response {
    pub(crate) status: u16,
    pub(crate) retry_after: Option<String>,
    pub(crate) body: String,
}

/// The transport the uploader posts through. Implemented by `UreqClient` and
/// by the test doubles.
pub(crate) trait HttpClient: Send {
    /// Posts one request. `Err` is a transport failure: no answer arrived.
    fn post(&mut self, request: &Request<'_>) -> Result<Response, String>;
}

/// Outcome of one send attempt, after classification.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Outcome {
    /// Accepted (or a deduplicated replay): `accepted` events counted, or
    /// the whole batch when the answer carried no count.
    Ok { accepted: Option<u64> },
    /// Retry later; `after_ms` is an explicit server-given delay.
    Retry { after_ms: Option<i64> },
    /// Never resend these bytes.
    Drop,
    /// Never resend anything until re-enabled.
    AuthHalt,
    /// The developer portal has hosted telemetry switched off for this
    /// application (403 with `error: telemetry_disabled`). Drop the batch and
    /// everything queued, and never resend anything until re-enabled. Unlike
    /// a bad key this is not a fault the developer fixes and re-enables
    /// through, and holding the events would resend the off period once the
    /// toggle came back, which is the backfill the toggle promises not to make.
    Disabled,
}

/// What one drain did, for the caller's stats.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub(crate) struct DrainReport {
    pub(crate) sent_events: u64,
    pub(crate) accepted_events: u64,
    pub(crate) dropped_events: u64,
    pub(crate) batches_sent: u32,
    pub(crate) halted: bool,
    /// The halt was a `telemetry_disabled`: the queue was dropped, and the
    /// pipe should discard what it collects from here on.
    pub(crate) disabled: bool,
}

pub(crate) struct Uploader {
    client: Box<dyn HttpClient>,
    authorization: String,
    app_id: String,
    user_agent: String,
    retry_attempt: u32,
    /// Epoch ms before which sends are suppressed.
    next_retry_at_ms: Option<i64>,
    auth_halted: bool,
    last_error: Option<String>,
    jitter: Box<dyn FnMut() -> u64 + Send>,
}

impl Uploader {
    pub(crate) fn new(
        client: Box<dyn HttpClient>,
        api_key: &str,
        app_id: &str,
        user_agent: String,
        jitter: Box<dyn FnMut() -> u64 + Send>,
    ) -> Self {
        Self {
            client,
            authorization: format!("Bearer {api_key}"),
            app_id: app_id.to_string(),
            user_agent,
            retry_attempt: 0,
            next_retry_at_ms: None,
            auth_halted: false,
            last_error: None,
            jitter,
        }
    }

    /// Whether a 401 or 403 stopped the uploader.
    #[cfg(test)]
    pub(crate) fn is_halted(&self) -> bool {
        self.auth_halted
    }

    /// The moment the backoff lifts, if one is in force.
    pub(crate) fn next_retry_at_ms(&self) -> Option<i64> {
        self.next_retry_at_ms
    }

    pub(crate) fn last_error(&self) -> Option<&str> {
        self.last_error.as_deref()
    }

    /// Sends queued batches oldest first until one must be retried, the
    /// queue empties, `max_batches` were sent, or `deadline` passes.
    pub(crate) fn drain(
        &mut self,
        store: &mut BatchStore,
        now_ms: i64,
        max_batches: u32,
        deadline: Option<Instant>,
    ) -> DrainReport {
        let mut report = DrainReport::default();
        loop {
            if self.auth_halted {
                report.halted = true;
                break;
            }
            if self.next_retry_at_ms.is_some_and(|at| now_ms < at) {
                break;
            }
            if report.batches_sent >= max_batches {
                break;
            }
            if deadline.is_some_and(|d| Instant::now() >= d) {
                break;
            }
            let Some(head) = store.front() else {
                break;
            };
            if now_ms - head.created_ms > MAX_BATCH_AGE_MS {
                let count = u64::from(head.event_count);
                store.pop_front();
                report.dropped_events += count;
                continue;
            }
            let count = u64::from(head.event_count);
            let batch_id = head.batch_id.to_string();
            let outcome = self.send(&batch_id, &head.body, now_ms);
            tracing::debug!(
                target: "offline_protocol::telemetry::pipe",
                batch_id = %batch_id,
                events = count,
                outcome = ?outcome,
                "telemetry batch sent"
            );
            match outcome {
                Outcome::Ok { accepted } => {
                    store.pop_front();
                    report.sent_events += count;
                    report.accepted_events += accepted.unwrap_or(count);
                    report.batches_sent += 1;
                    // Cleared only here. `last_error` is what the stats
                    // surface as "the most recent send failure", and an
                    // accepted batch is the one event that makes an earlier
                    // one no longer current. A drop also resets the backoff
                    // but keeps its error, because the drop is the error.
                    self.last_error = None;
                    self.reset_backoff();
                }
                Outcome::Drop => {
                    store.pop_front();
                    report.dropped_events += count;
                    self.reset_backoff();
                }
                Outcome::AuthHalt => {
                    // The batch stays queued. A 401 or 403 is a configuration
                    // fault the developer fixes and re-enables through, not
                    // bytes the ingest will never take, and the halt already
                    // stops every later send: dropping the head would lose one
                    // batch and buy nothing. The caps still bound the queue if
                    // the key is never fixed.
                    self.auth_halted = true;
                    report.halted = true;
                    break;
                }
                Outcome::Disabled => {
                    // The toggle is off in the portal. Nothing queued will be
                    // wanted when it comes back on, since resending it then
                    // is exactly the backfill the toggle promises not to
                    // make, so the whole queue goes, not just the head, and
                    // the halt stops every later send until the next enable.
                    while let Some(batch) = store.pop_front() {
                        report.dropped_events += u64::from(batch.event_count);
                    }
                    self.reset_backoff();
                    self.auth_halted = true;
                    report.halted = true;
                    report.disabled = true;
                    break;
                }
                Outcome::Retry { after_ms } => {
                    self.schedule_backoff(now_ms, after_ms);
                    break;
                }
            }
        }
        store.flush_index();
        report
    }

    fn send(&mut self, batch_id: &str, body: &str, now_ms: i64) -> Outcome {
        let request = Request {
            url: TELEMETRY_INGEST_URL,
            headers: vec![
                ("Authorization", self.authorization.as_str()),
                (HEADER_APP_ID, self.app_id.as_str()),
                ("Content-Type", "application/json"),
                (HEADER_IDEMPOTENCY, batch_id),
                ("User-Agent", self.user_agent.as_str()),
            ],
            body,
        };
        let response = match self.client.post(&request) {
            Ok(response) => response,
            Err(err) => {
                self.last_error = Some(err);
                return Outcome::Retry { after_ms: None };
            }
        };
        classify_response(&response, now_ms, &mut self.last_error)
    }

    fn reset_backoff(&mut self) {
        self.retry_attempt = 0;
        self.next_retry_at_ms = None;
    }

    fn schedule_backoff(&mut self, now_ms: i64, after_ms: Option<i64>) {
        let delay = match after_ms {
            Some(after) => after.clamp(0, BACKOFF_MAX_MS),
            None => {
                let exponent = self.retry_attempt.min(20);
                let capped = (BACKOFF_BASE_MS << exponent).min(BACKOFF_MAX_MS);
                let jitter = ((self.jitter)() % BACKOFF_JITTER_MS) as i64;
                capped + jitter
            }
        };
        self.retry_attempt = self.retry_attempt.saturating_add(1);
        self.next_retry_at_ms = Some(now_ms + delay);
    }
}

/// Classifies one answer. Public to the module so the status matrix can be
/// tested without a client.
///
/// `now_ms` is the pipe's clock rather than the wall clock, because a
/// `Retry-After` given as an HTTP date is a delay only relative to the same
/// clock the backoff is scheduled against.
pub(crate) fn classify_response(
    response: &Response,
    now_ms: i64,
    last_error: &mut Option<String>,
) -> Outcome {
    let status = response.status;
    if (200..300).contains(&status) {
        return Outcome::Ok {
            accepted: accepted_count(&response.body),
        };
    }
    *last_error = Some(format!("ingest responded {status}"));
    match status {
        // Only a 403, and only with the code. A body that does not parse or
        // names any other code is the ordinary halt, so an ingest that
        // predates the code, or answers `forbidden`, keeps the queue as before.
        403 if has_error_code(&response.body, ERROR_TELEMETRY_DISABLED) => {
            *last_error = Some(format!("ingest responded 403 ({ERROR_TELEMETRY_DISABLED})"));
            Outcome::Disabled
        }
        401 | 403 => Outcome::AuthHalt,
        408 | 500..=599 => Outcome::Retry { after_ms: None },
        429 => Outcome::Retry {
            after_ms: response
                .retry_after
                .as_deref()
                .and_then(|value| parse_retry_after(value, now_ms)),
        },
        // 413 included: the ingest's body limit is far above the largest
        // batch the pipe cuts, so an oversized batch is a pathological
        // event, and splitting it would only send the pathology twice.
        //
        // A 3xx lands here too. The agent follows no redirect (see
        // `UreqClient::with_timeouts`), so a redirect arrives as an answer,
        // and resending to the endpoint fixed at build time would only draw
        // the same answer again.
        _ => Outcome::Drop,
    }
}

/// Whether an ingest error body, `{"error": "...", "message": "..."}`, names
/// `code`. A body that is not JSON, or has no string `error`, names nothing.
fn has_error_code(body: &str, code: &str) -> bool {
    serde_json::from_str::<serde_json::Value>(body)
        .is_ok_and(|value| value.get("error").and_then(serde_json::Value::as_str) == Some(code))
}

/// The number of events the ingest counted for a 2xx answer.
///
/// A 202 body carries `accepted`; a 200 replay carries `deduplicated: true`
/// and counted nothing new. A body that is neither is `None`, which the
/// caller reads as the batch having been accepted whole, since that is what
/// a bare 2xx means.
fn accepted_count(body: &str) -> Option<u64> {
    let value = serde_json::from_str::<serde_json::Value>(body).ok()?;
    if let Some(accepted) = value.get("accepted").and_then(|v| v.as_u64()) {
        return Some(accepted);
    }
    if value.get("deduplicated").and_then(|v| v.as_bool()) == Some(true) {
        return Some(0);
    }
    None
}

/// Parses a `Retry-After` value, seconds or an HTTP date, into a delay.
pub(crate) fn parse_retry_after(value: &str, now_ms: i64) -> Option<i64> {
    let value = value.trim();
    if let Ok(seconds) = value.parse::<u64>() {
        return Some((seconds as i64).saturating_mul(1000));
    }
    let date = chrono::DateTime::parse_from_rfc2822(value).ok()?;
    Some((date.timestamp_millis() - now_ms).max(0))
}

/// Whether a non-final flush should wait for a charger.
pub(crate) fn battery_defers(battery_level: Option<u8>, is_charging: bool) -> bool {
    matches!(battery_level, Some(level) if level < BATTERY_DEFER_BELOW) && !is_charging
}

/// The `User-Agent` every request carries.
pub(crate) fn user_agent(os_token: &str, os_major: u16) -> String {
    format!("offline-protocol-sdk/{SDK_VERSION} ({os_token}/{os_major})")
}

/// The roots the uploader trusts, as PEM: the server roots of the
/// certificate authorities the ingest's edge can be issued from, and no
/// others. See the file's header for the list and why.
const INGEST_ROOTS_PEM: &[u8] = include_bytes!("ingest-roots.pem");

/// [`INGEST_ROOTS_PEM`] as the agent's root store. The file is a constant the
/// tests parse in full, so nothing is dropped here in practice; were a root
/// ever unreadable, the upload would fail validation and back off rather than
/// take the uploader thread down.
fn ingest_roots() -> ureq::tls::RootCerts {
    let roots: Vec<_> = ureq::tls::parse_pem(INGEST_ROOTS_PEM)
        .filter_map(|item| match item {
            Ok(ureq::tls::PemItem::Certificate(cert)) => Some(cert),
            _ => None,
        })
        .collect();
    ureq::tls::RootCerts::new_with_certs(&roots)
}

/// The production client: TLS through rustls with the ingest's pinned roots,
/// no redirect followed, an HTTP CONNECT proxy when the environment names
/// one, a connection pool that outlives the flush cadence, and bounded
/// timeouts so a stalled socket cannot hold the uploader thread past its
/// deadline.
pub(crate) struct UreqClient {
    agent: ureq::Agent,
}

impl UreqClient {
    /// Global request timeout: a stalled socket cannot hold the uploader
    /// thread past this.
    pub(crate) const GLOBAL_TIMEOUT: Duration = Duration::from_secs(15);
    /// Connect timeout.
    pub(crate) const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
    /// How long an idle connection is kept. Above the flush cadence, so a
    /// flush every 30 s reuses the TLS session rather than handshaking.
    pub(crate) const IDLE_AGE: Duration = Duration::from_secs(60);

    pub(crate) fn new() -> Self {
        Self::with_timeouts(Self::GLOBAL_TIMEOUT, Self::CONNECT_TIMEOUT)
    }

    /// Every setting the egress properties in `docs/telemetry.md` rest on is
    /// named here rather than inherited, because a ureq upgrade that moved a
    /// default would otherwise change how the SDK sends with no diff in this
    /// crate. Each one is pinned by a test below.
    pub(crate) fn with_timeouts(global: Duration, connect: Duration) -> Self {
        let config = ureq::Agent::config_builder()
            .http_status_as_error(false)
            // An ordinary HTTP CONNECT proxy when the process environment
            // names one (`ALL_PROXY`, `HTTPS_PROXY` or `HTTP_PROXY`, with
            // `NO_PROXY` exempting hosts), read here, which is when telemetry
            // is enabled. The tunnel carries a TLS session that terminates at
            // the ingest, so the proxy forwards the payload without reading
            // it.
            .proxy(ureq::Proxy::try_from_env())
            // No redirect is followed. Followed, a 301, 302 or 303 turned the
            // POST into a GET at the new location, a 2xx there was classified
            // as success, and the batch was deleted unsent and counted as
            // accepted. At zero every 3xx comes back as an answer, and
            // `classify_response` drops it.
            .max_redirects(0)
            // Only the roots that can issue the ingest's certificate, compiled
            // into the SDK, never the device's trust store: an interception
            // certificate installed only on the device fails validation, and
            // so does one from any other authority in the Mozilla set,
            // including one later found compromised. The crypto provider is
            // named too, rather than taken from a process default another
            // library may have installed.
            .tls_config(
                ureq::tls::TlsConfig::builder()
                    .root_certs(ingest_roots())
                    .unversioned_rustls_crypto_provider(std::sync::Arc::new(
                        rustls::crypto::ring::default_provider(),
                    ))
                    .disable_verification(false)
                    .build(),
            )
            .timeout_global(Some(global))
            .timeout_connect(Some(connect))
            .max_idle_age(Self::IDLE_AGE)
            .max_idle_connections_per_host(1)
            .build();
        Self {
            agent: ureq::Agent::new_with_config(config),
        }
    }
}

impl HttpClient for UreqClient {
    fn post(&mut self, request: &Request<'_>) -> Result<Response, String> {
        let mut builder = self.agent.post(request.url);
        for (name, value) in &request.headers {
            builder = builder.header(*name, *value);
        }
        let mut response = builder
            .send(request.body.as_bytes())
            .map_err(|err| err.to_string())?;
        let status = response.status().as_u16();
        let retry_after = response
            .headers()
            .get("retry-after")
            .and_then(|v| v.to_str().ok())
            .map(str::to_string);
        let body = response
            .body_mut()
            .with_config()
            .limit(64 * 1024)
            .read_to_string()
            .unwrap_or_default();
        Ok(Response {
            status,
            retry_after,
            body,
        })
    }
}

/// Uploader cadence knobs the worker reads.
pub(crate) const MAX_BATCHES_PER_WAKE: u32 = 8;

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::telemetry::pipe::store::tests::{batch, test_store};
    use crate::telemetry::pipe::store::Backend;
    use std::collections::VecDeque;
    use std::sync::{Arc, Mutex};

    /// A client that answers from a script and records every request.
    #[derive(Default)]
    pub(crate) struct ScriptedClient {
        pub(crate) answers: VecDeque<Result<Response, String>>,
        pub(crate) requests: Arc<Mutex<Vec<(Vec<(String, String)>, String)>>>,
    }

    impl HttpClient for ScriptedClient {
        fn post(&mut self, request: &Request<'_>) -> Result<Response, String> {
            self.requests.lock().unwrap().push((
                request
                    .headers
                    .iter()
                    .map(|(k, v)| (k.to_string(), v.to_string()))
                    .collect(),
                request.body.to_string(),
            ));
            self.answers
                .pop_front()
                .unwrap_or_else(|| Err("no scripted answer".into()))
        }
    }

    pub(crate) fn answer(status: u16, body: &str) -> Result<Response, String> {
        Ok(Response {
            status,
            retry_after: None,
            body: body.into(),
        })
    }

    fn uploader(
        answers: Vec<Result<Response, String>>,
    ) -> (Uploader, Arc<Mutex<Vec<(Vec<(String, String)>, String)>>>) {
        let client = ScriptedClient {
            answers: answers.into(),
            requests: Arc::default(),
        };
        let requests = client.requests.clone();
        (
            Uploader::new(
                Box::new(client),
                "key",
                "app_1",
                user_agent("ios", 18),
                Box::new(|| 0),
            ),
            requests,
        )
    }

    #[test]
    fn every_request_carries_the_five_headers() {
        let (mut up, requests) = uploader(vec![answer(202, r#"{"accepted":1,"rejected":0}"#)]);
        let mut store = test_store(Backend::Memory);
        store.push(batch(1, 0));
        up.drain(&mut store, 0, 8, None);
        let (headers, _) = &requests.lock().unwrap()[0];
        let names: Vec<&str> = headers.iter().map(|(k, _)| k.as_str()).collect();
        assert_eq!(
            names,
            [
                "Authorization",
                HEADER_APP_ID,
                "Content-Type",
                HEADER_IDEMPOTENCY,
                "User-Agent"
            ]
        );
        assert_eq!(headers[0].1, "Bearer key");
        assert_eq!(headers[1].1, "app_1");
        assert!(headers[4].1.starts_with("offline-protocol-sdk/"));
        assert!(headers[4].1.ends_with("(ios/18)"));
    }

    #[test]
    fn a_2xx_deletes_and_counts_sent_and_accepted_separately() {
        let (mut up, _) = uploader(vec![
            answer(202, r#"{"accepted":2,"rejected":1,"rejected_reasons":[]}"#),
            answer(200, r#"{"deduplicated":true}"#),
            answer(204, ""),
        ]);
        let mut store = test_store(Backend::Memory);
        store.push(batch(3, 0));
        store.push(batch(4, 0));
        store.push(batch(5, 0));
        let report = up.drain(&mut store, 0, 8, None);
        assert!(store.is_empty());
        assert_eq!(report.sent_events, 12);
        // 2 from the 202 body, 0 for the replay, 5 for a bare 2xx.
        assert_eq!(report.accepted_events, 7);
    }

    #[test]
    fn a_4xx_drops_and_continues_and_401_403_halt() {
        for status in [400, 413, 422] {
            let (mut up, _) = uploader(vec![answer(status, ""), answer(202, r#"{"accepted":1}"#)]);
            let mut store = test_store(Backend::Memory);
            store.push(batch(1, 0));
            store.push(batch(1, 0));
            let report = up.drain(&mut store, 0, 8, None);
            assert!(store.is_empty(), "{status}");
            assert_eq!(report.dropped_events, 1, "{status}");
            assert_eq!(report.sent_events, 1, "{status}");
            assert!(up.next_retry_at_ms().is_none(), "{status} resets backoff");
        }
        for status in [401, 403] {
            let (mut up, requests) = uploader(vec![answer(status, ""), answer(202, "")]);
            let mut store = test_store(Backend::Memory);
            store.push(batch(1, 0));
            store.push(batch(1, 0));
            let report = up.drain(&mut store, 0, 8, None);
            assert!(report.halted);
            assert!(up.is_halted());
            assert_eq!(
                report.dropped_events, 0,
                "{status}: a rejected key loses nothing; the queue waits for a good one"
            );
            assert_eq!(store.len(), 2, "{status}: the refused batch is kept too");
            assert_eq!(
                requests.lock().unwrap().len(),
                1,
                "{status}: nothing more is sent"
            );
            assert_eq!(
                up.last_error(),
                Some(format!("ingest responded {status}").as_str())
            );
            let again = up.drain(&mut store, 1_000_000, 8, None);
            assert!(again.halted);
            assert_eq!(requests.lock().unwrap().len(), 1);
        }
    }

    #[test]
    fn a_403_telemetry_disabled_drops_the_whole_queue_and_halts() {
        let body = r#"{"error":"telemetry_disabled","message":"forbidden: mesh telemetry is disabled for this application in the developer portal"}"#;
        let (mut up, requests) = uploader(vec![answer(403, body), answer(202, "")]);
        let mut store = test_store(Backend::Memory);
        store.push(batch(1, 0));
        store.push(batch(4, 0));
        let report = up.drain(&mut store, 0, 8, None);
        assert!(report.halted);
        assert!(report.disabled);
        assert!(up.is_halted());
        assert!(
            store.is_empty(),
            "the refused batch and the one queued behind it are both gone"
        );
        assert_eq!(report.dropped_events, 5);
        assert_eq!(requests.lock().unwrap().len(), 1, "nothing more is sent");
        assert!(up.next_retry_at_ms().is_none());
        assert_eq!(
            up.last_error(),
            Some("ingest responded 403 (telemetry_disabled)")
        );
        // Still halted on a later wake, whatever has been queued since.
        store.push(batch(1, 0));
        let again = up.drain(&mut store, 1_000_000, 8, None);
        assert!(again.halted);
        assert_eq!(requests.lock().unwrap().len(), 1);
    }

    #[test]
    fn only_a_403_carrying_the_telemetry_disabled_code_disables() {
        let mut last = None;
        let response = |status: u16, body: &str| Response {
            status,
            retry_after: None,
            body: body.into(),
        };
        let disabled = r#"{"error":"telemetry_disabled","message":"x"}"#;
        assert_eq!(
            classify_response(&response(403, disabled), 0, &mut last),
            Outcome::Disabled
        );
        assert_eq!(
            last.as_deref(),
            Some("ingest responded 403 (telemetry_disabled)")
        );
        // The code on any other status is not the portal toggle.
        assert_eq!(
            classify_response(&response(401, disabled), 0, &mut last),
            Outcome::AuthHalt
        );
        assert_eq!(last.as_deref(), Some("ingest responded 401"));
        // A 403 with the shared code, no body, or junk is the ordinary halt:
        // an ingest that predates the code keeps the queue as it always did.
        for body in [
            r#"{"error":"forbidden","message":"forbidden: api key revoked"}"#,
            "",
            "not json",
            r#"{"message":"no code"}"#,
        ] {
            assert_eq!(
                classify_response(&response(403, body), 0, &mut last),
                Outcome::AuthHalt,
                "{body:?}"
            );
            assert_eq!(last.as_deref(), Some("ingest responded 403"), "{body:?}");
        }
    }

    #[test]
    fn retryable_answers_back_off_exponentially_with_jitter_to_a_15_minute_cap() {
        let client = ScriptedClient {
            answers: (0..12).map(|_| answer(500, "")).collect(),
            requests: Arc::default(),
        };
        let mut up = Uploader::new(
            Box::new(client),
            "key",
            "app",
            String::new(),
            Box::new(|| 7),
        );
        let mut store = test_store(Backend::Memory);
        store.push(batch(1, 0));
        let mut now = 0;
        let mut delays = Vec::new();
        for _ in 0..12 {
            up.drain(&mut store, now, 8, None);
            let at = up.next_retry_at_ms().expect("backoff armed");
            delays.push(at - now);
            now = at;
        }
        assert_eq!(delays[0], 1_007);
        assert_eq!(delays[1], 2_007);
        assert_eq!(delays[2], 4_007);
        assert_eq!(delays[9], 512_007);
        assert_eq!(delays[10], BACKOFF_MAX_MS + 7);
        assert_eq!(delays[11], BACKOFF_MAX_MS + 7);
        assert_eq!(store.len(), 1, "the batch is kept for the next try");
    }

    #[test]
    fn a_408_and_a_network_error_are_retryable_and_the_backoff_window_is_honoured() {
        let (mut up, requests) = uploader(vec![answer(408, ""), Err("connection refused".into())]);
        let mut store = test_store(Backend::Memory);
        store.push(batch(1, 0));
        up.drain(&mut store, 0, 8, None);
        assert_eq!(requests.lock().unwrap().len(), 1);
        up.drain(&mut store, 10, 8, None);
        assert_eq!(
            requests.lock().unwrap().len(),
            1,
            "inside the backoff window"
        );
        let at = up.next_retry_at_ms().expect("armed");
        up.drain(&mut store, at, 8, None);
        assert_eq!(requests.lock().unwrap().len(), 2);
        assert_eq!(up.last_error(), Some("connection refused"));
    }

    #[test]
    fn a_429_honours_retry_after_in_seconds_and_as_a_date_and_backs_off_without_it() {
        let mut last = None;
        let seconds = Response {
            status: 429,
            retry_after: Some("120".into()),
            body: String::new(),
        };
        assert_eq!(
            classify_response(&seconds, 0, &mut last),
            Outcome::Retry {
                after_ms: Some(120_000)
            }
        );
        let bare = Response {
            status: 429,
            retry_after: None,
            body: String::new(),
        };
        assert_eq!(
            classify_response(&bare, 0, &mut last),
            Outcome::Retry { after_ms: None }
        );
        let now = chrono::Utc::now().timestamp_millis();
        let date = chrono::DateTime::from_timestamp_millis(now + 90_000)
            .expect("valid")
            .to_rfc2822();
        assert!(matches!(
            parse_retry_after(&date, now),
            Some(ms) if (89_000..=90_000).contains(&ms)
        ));
        assert_eq!(parse_retry_after("soon", now), None);
    }

    #[test]
    fn a_batch_past_the_ttl_is_dropped_unsent() {
        let (mut up, requests) = uploader(vec![answer(202, "")]);
        let mut store = test_store(Backend::Memory);
        store.push(batch(3, 0));
        store.push(batch(1, MAX_BATCH_AGE_MS));
        let report = up.drain(&mut store, MAX_BATCH_AGE_MS + 1, 8, None);
        assert_eq!(report.dropped_events, 3);
        assert_eq!(report.sent_events, 1);
        assert_eq!(requests.lock().unwrap().len(), 1);
    }

    #[test]
    fn a_drain_sends_at_most_max_batches_per_wake() {
        let (mut up, requests) = uploader((0..10).map(|_| answer(202, "")).collect());
        let mut store = test_store(Backend::Memory);
        for _ in 0..10 {
            store.push(batch(1, 0));
        }
        up.drain(&mut store, 0, 8, None);
        assert_eq!(requests.lock().unwrap().len(), 8);
        assert_eq!(store.len(), 2);
    }

    /// The real client against a local server: the headers and body arrive
    /// as sent, and the status and `Retry-After` come back as answered.
    #[test]
    fn the_ureq_client_posts_headers_and_body_and_reads_status_and_retry_after() {
        let server = tiny_http::Server::http("127.0.0.1:0").expect("local server");
        let addr = server.server_addr().to_ip().expect("tcp listener");
        let url = format!("http://{addr}/v1/events");
        let served = std::thread::spawn(move || {
            let mut seen = Vec::new();
            for n in 0..3 {
                let mut request = server.recv().expect("request");
                let mut body = String::new();
                request
                    .as_reader()
                    .read_to_string(&mut body)
                    .expect("body reads");
                let headers: Vec<(String, String)> = request
                    .headers()
                    .iter()
                    .map(|h| (h.field.as_str().to_string(), h.value.as_str().to_string()))
                    .collect();
                seen.push((headers, body));
                let response = match n {
                    0 => tiny_http::Response::from_string(r#"{"accepted":1,"rejected":0}"#)
                        .with_status_code(202),
                    1 => tiny_http::Response::from_string("")
                        .with_status_code(429)
                        .with_header(
                            tiny_http::Header::from_bytes(&b"Retry-After"[..], &b"7"[..])
                                .expect("header"),
                        ),
                    _ => tiny_http::Response::from_string("").with_status_code(500),
                };
                request.respond(response).expect("respond");
            }
            seen
        });

        let mut client = UreqClient::new();
        let post = |client: &mut UreqClient| {
            client.post(&Request {
                url: &url,
                headers: vec![
                    ("Authorization", "Bearer mp_key"),
                    (HEADER_APP_ID, "app_1"),
                    ("Content-Type", "application/json"),
                    (HEADER_IDEMPOTENCY, "b-1"),
                    ("User-Agent", "offline-protocol-sdk/test (linux/0)"),
                ],
                body: r#"{"events":[]}"#,
            })
        };
        let first = post(&mut client).expect("answered");
        assert_eq!(first.status, 202);
        assert!(first.body.contains("accepted"));
        let second = post(&mut client).expect("answered");
        assert_eq!(second.status, 429);
        assert_eq!(second.retry_after.as_deref(), Some("7"));
        let third = post(&mut client).expect("answered");
        assert_eq!(third.status, 500);

        let seen = served.join().expect("server thread");
        let (headers, body) = &seen[0];
        assert_eq!(body, r#"{"events":[]}"#);
        let get = |name: &str| {
            headers
                .iter()
                .find(|(k, _)| k.eq_ignore_ascii_case(name))
                .map(|(_, v)| v.as_str())
        };
        assert_eq!(get("authorization"), Some("Bearer mp_key"));
        assert_eq!(get("x-mesh-analytics-app-id"), Some("app_1"));
        assert_eq!(get("content-type"), Some("application/json"));
        assert_eq!(get("idempotency-key"), Some("b-1"));
        assert_eq!(
            get("user-agent"),
            Some("offline-protocol-sdk/test (linux/0)")
        );
    }

    /// A server that accepts the connection and never answers is a transport
    /// error at the global timeout, not a hang.
    #[test]
    fn the_ureq_client_gives_up_on_a_server_that_never_answers() {
        let server = tiny_http::Server::http("127.0.0.1:0").expect("local server");
        let addr = server.server_addr().to_ip().expect("tcp listener");
        let url = format!("http://{addr}/v1/events");
        let hold = std::thread::spawn(move || {
            let request = server.recv().expect("request");
            std::thread::sleep(Duration::from_millis(1_500));
            drop(request);
        });
        let mut client =
            UreqClient::with_timeouts(Duration::from_millis(400), Duration::from_millis(400));
        let started = Instant::now();
        let result = client.post(&Request {
            url: &url,
            headers: Vec::new(),
            body: "{}",
        });
        assert!(result.is_err(), "{result:?}");
        assert!(
            started.elapsed() < Duration::from_millis(1_400),
            "{:?}",
            started.elapsed()
        );
        hold.join().expect("server thread");
    }

    /// Refused outright: a connection error, not a status.
    #[test]
    fn the_ureq_client_reports_a_refused_connection_as_a_transport_error() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        let addr = listener.local_addr().expect("addr");
        drop(listener);
        let url = format!("http://{addr}/v1/events");
        let mut client = UreqClient::with_timeouts(Duration::from_secs(2), Duration::from_secs(2));
        let result = client.post(&Request {
            url: &url,
            headers: Vec::new(),
            body: "{}",
        });
        assert!(result.is_err());
    }

    #[test]
    fn a_3xx_drops_the_batch_and_continues() {
        for status in [300, 301, 302, 303, 307, 308] {
            let (mut up, _) = uploader(vec![answer(status, ""), answer(202, r#"{"accepted":1}"#)]);
            let mut store = test_store(Backend::Memory);
            store.push(batch(1, 0));
            store.push(batch(1, 0));
            let report = up.drain(&mut store, 0, 8, None);
            assert!(store.is_empty(), "{status}");
            assert_eq!(report.dropped_events, 1, "{status}");
            assert_eq!(report.sent_events, 1, "{status}");
            assert_eq!(
                report.accepted_events, 1,
                "{status}: a redirect is never counted as accepted"
            );
            assert!(!report.halted, "{status}");
        }
    }

    /// A redirect comes back as an answer rather than being followed, so it
    /// reaches the status matrix instead of a location the build never named.
    #[test]
    fn the_ureq_client_returns_a_redirect_rather_than_following_it() {
        let server = tiny_http::Server::http("127.0.0.1:0").expect("local server");
        let addr = server.server_addr().to_ip().expect("tcp listener");
        let url = format!("http://{addr}/v1/events");
        let location = format!("http://{addr}/moved");
        let served = std::thread::spawn(move || {
            let mut paths = Vec::new();
            for status in [301, 308] {
                let request = server.recv().expect("request");
                paths.push(request.url().to_string());
                let response = tiny_http::Response::from_string("")
                    .with_status_code(status)
                    .with_header(
                        tiny_http::Header::from_bytes(&b"Location"[..], location.as_bytes())
                            .expect("header"),
                    );
                request.respond(response).expect("respond");
            }
            // A followed redirect would arrive here as one more request.
            if let Some(request) = server
                .recv_timeout(Duration::from_millis(500))
                .expect("server")
            {
                paths.push(request.url().to_string());
            }
            paths
        });

        let mut client = UreqClient::new();
        for expected in [301, 308] {
            let response = client
                .post(&Request {
                    url: &url,
                    headers: vec![("Content-Type", "application/json")],
                    body: r#"{"events":[]}"#,
                })
                .expect("answered");
            assert_eq!(response.status, expected);
        }
        assert_eq!(
            served.join().expect("server thread"),
            ["/v1/events", "/v1/events"],
            "nothing followed the Location header"
        );
    }

    /// Set only on the child process the proxy test spawns, so the child
    /// half is a no-op in an ordinary run and never posts anywhere.
    const PROXY_PROBE_ENV: &str = "OFFLINE_TELEMETRY_PROXY_PROBE";

    /// Every variable ureq reads a proxy from, in both cases. A child process
    /// has all of them cleared before the one under test is set: ureq takes
    /// the first of these it can parse, a developer's shell may export any of
    /// them, and on Windows the names are case-insensitive, so clearing
    /// `https_proxy` after setting `HTTPS_PROXY` would clear the stand-in.
    const PROXY_VARS: [&str; 8] = [
        "ALL_PROXY",
        "all_proxy",
        "HTTPS_PROXY",
        "https_proxy",
        "HTTP_PROXY",
        "http_proxy",
        "NO_PROXY",
        "no_proxy",
    ];

    /// Set only on the child processes the proxy configuration test spawns,
    /// to the `host:port` the child's agent must be configured with, or
    /// `none`.
    const PROXY_CONFIG_ENV: &str = "OFFLINE_TELEMETRY_PROXY_CONFIG";

    /// The production constructor tunnels through the CONNECT proxy the
    /// environment names, and TLS runs through that tunnel to the
    /// destination rather than terminating at the proxy.
    ///
    /// The proxy variable is set on a child process that runs only
    /// `proxy_probe_child`, never on this one: the environment is
    /// process-wide, and every other `UreqClient` test in the binary would
    /// route through the stand-in while it was set.
    #[test]
    fn the_ureq_client_tunnels_through_the_connect_proxy_the_environment_names() {
        use std::io::{Read, Write};

        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        listener
            .set_nonblocking(true)
            .expect("nonblocking listener");
        let proxy = format!("http://{}", listener.local_addr().expect("addr"));
        let (_, module) = module_path!()
            .split_once("::")
            .expect("crate-qualified path");
        let mut command = std::process::Command::new(std::env::current_exe().expect("test binary"));
        command.args(["--exact", &format!("{module}::proxy_probe_child")]);
        for name in PROXY_VARS {
            command.env_remove(name);
        }
        let mut child = command
            .env("HTTPS_PROXY", &proxy)
            .env(PROXY_PROBE_ENV, "1")
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .spawn()
            .expect("child test process");

        let deadline = Instant::now() + Duration::from_secs(30);
        let mut tunnel = loop {
            match listener.accept() {
                Ok((stream, _)) => break stream,
                Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
                    if let Some(status) = child.try_wait().expect("child status") {
                        panic!("the client never reached the proxy (child exited: {status})");
                    }
                    if Instant::now() >= deadline {
                        let _ = child.kill();
                        panic!("the client never reached the proxy");
                    }
                    std::thread::sleep(Duration::from_millis(10));
                }
                Err(err) => panic!("accept: {err}"),
            }
        };
        tunnel.set_nonblocking(false).expect("blocking tunnel");
        tunnel
            .set_read_timeout(Some(Duration::from_secs(10)))
            .expect("read timeout");

        let mut head = Vec::new();
        let mut byte = [0u8; 1];
        while !head.ends_with(b"\r\n\r\n") {
            tunnel.read_exact(&mut byte).expect("CONNECT request");
            head.push(byte[0]);
        }
        let head = String::from_utf8(head).expect("ASCII request head");
        assert!(
            head.starts_with("CONNECT ingest.invalid:443 HTTP/1.1\r\n"),
            "{head:?}"
        );
        tunnel
            .write_all(b"HTTP/1.1 200 Connection established\r\n\r\n")
            .expect("open the tunnel");

        // What follows is a TLS handshake naming the destination, not an
        // HTTP request: the proxy forwards the session and reads none of it.
        let mut record = [0u8; 5];
        tunnel.read_exact(&mut record).expect("TLS record header");
        assert_eq!(record[0], 0x16, "a TLS handshake record, not plaintext");
        let mut hello = vec![0u8; usize::from(u16::from_be_bytes([record[3], record[4]]))];
        tunnel.read_exact(&mut hello).expect("ClientHello");
        let sni = b"ingest.invalid";
        assert!(
            hello.windows(sni.len()).any(|w| w == sni),
            "the ClientHello names the destination"
        );

        drop(tunnel);
        let status = child.wait().expect("child exits");
        assert!(status.success(), "child test failed: {status}");
    }

    /// The child half of the proxy test: one post through the production
    /// constructor, to a host that cannot resolve, so a client that skipped
    /// the proxy fails at DNS instead of reaching anything real.
    #[test]
    fn proxy_probe_child() {
        if std::env::var_os(PROXY_PROBE_ENV).is_none() {
            return;
        }
        let _ = UreqClient::new().post(&Request {
            url: "https://ingest.invalid/v1/events",
            headers: Vec::new(),
            body: "{}",
        });
    }

    /// The agent takes its proxy from the environment when it is built, in
    /// the order `docs/telemetry.md` states, exempts what `NO_PROXY` names,
    /// and has none when nothing is set.
    ///
    /// Asserted on the agent's configuration, in child processes so that no
    /// variable reaches this one. The tunnel test above proves a configured
    /// proxy carries the request; this pins which proxy gets configured.
    #[test]
    fn the_agent_takes_its_proxy_from_the_environment_when_it_is_built() {
        let cases: [(&[(&str, &str)], &str); 3] = [
            (
                &[
                    ("HTTPS_PROXY", "http://127.0.0.1:3128"),
                    ("NO_PROXY", "skip.invalid"),
                ],
                "127.0.0.1:3128",
            ),
            (
                &[
                    ("HTTP_PROXY", "http://127.0.0.3:8080"),
                    ("HTTPS_PROXY", "http://127.0.0.1:3128"),
                    ("ALL_PROXY", "http://127.0.0.2:8001"),
                ],
                "127.0.0.2:8001",
            ),
            (&[], "none"),
        ];
        let (_, module) = module_path!()
            .split_once("::")
            .expect("crate-qualified path");
        for (vars, expected) in cases {
            let mut command =
                std::process::Command::new(std::env::current_exe().expect("test binary"));
            command.args(["--exact", &format!("{module}::proxy_config_child")]);
            for name in PROXY_VARS {
                command.env_remove(name);
            }
            for (name, value) in vars {
                command.env(name, value);
            }
            let output = command
                .env(PROXY_CONFIG_ENV, expected)
                .output()
                .expect("child test process");
            assert!(
                output.status.success(),
                "expected {expected}: {}{}",
                String::from_utf8_lossy(&output.stdout),
                String::from_utf8_lossy(&output.stderr)
            );
        }
    }

    /// The child half of the proxy configuration test: builds the production
    /// client and checks its proxy against [`PROXY_CONFIG_ENV`].
    #[test]
    fn proxy_config_child() {
        let Some(expected) = std::env::var_os(PROXY_CONFIG_ENV) else {
            return;
        };
        let expected = expected.to_str().expect("UTF-8");
        let client = UreqClient::new();
        let proxy = client.agent.config().proxy();
        if expected == "none" {
            assert!(
                proxy.is_none(),
                "{proxy:?} with no proxy in the environment"
            );
            return;
        }
        let proxy = proxy.expect("the environment's proxy was configured");
        assert_eq!(format!("{}:{}", proxy.host(), proxy.port()), expected);
        assert_eq!(proxy.protocol(), ureq::ProxyProtocol::Http);
        assert!(proxy.is_from_env());
        let ingest: ureq::http::Uri = "https://ingest.invalid/v1/events".parse().expect("URI");
        assert!(!proxy.is_no_proxy(&ingest));
        if std::env::var_os("NO_PROXY").is_some() {
            let exempt: ureq::http::Uri = "https://skip.invalid/v1/events".parse().expect("URI");
            assert!(proxy.is_no_proxy(&exempt), "NO_PROXY was not honoured");
        }
    }

    /// A certificate nothing in the pinned roots issued is refused, as an
    /// interception proxy's is, even when it names the host that was dialled.
    ///
    /// The server stands in for a TLS-intercepting proxy whose root a user
    /// installed in the device trust store: its certificate is valid for
    /// `127.0.0.1` and self-issued. The client must abort the handshake, so
    /// no request byte reaches the server. The accepting half would need a
    /// certificate a pinned root issued, so it is pinned on the agent's
    /// configuration instead, by the test below.
    #[test]
    fn a_certificate_the_pinned_roots_do_not_trust_is_refused() {
        use rustls::pki_types::pem::PemObject;
        use rustls::pki_types::{CertificateDer, PrivateKeyDer};

        const CERT: &[u8] = include_bytes!(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/tests/fixtures/untrusted-interception-cert.pem"
        ));
        const KEY: &[u8] = include_bytes!(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/tests/fixtures/untrusted-interception-key.pem"
        ));

        let config = rustls::ServerConfig::builder_with_provider(std::sync::Arc::new(
            rustls::crypto::ring::default_provider(),
        ))
        .with_safe_default_protocol_versions()
        .expect("protocol versions")
        .with_no_client_auth()
        .with_single_cert(
            vec![CertificateDer::from_pem_slice(CERT).expect("certificate")],
            PrivateKeyDer::from_pem_slice(KEY).expect("private key"),
        )
        .expect("server config");
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        let url = format!("https://{}/v1/events", listener.local_addr().expect("addr"));
        let server = std::thread::spawn(move || {
            let (mut tcp, _) = listener.accept().expect("accept");
            tcp.set_read_timeout(Some(Duration::from_secs(10)))
                .expect("read timeout");
            let mut tls =
                rustls::ServerConnection::new(std::sync::Arc::new(config)).expect("connection");
            while tls.is_handshaking() {
                if let Err(err) = tls.complete_io(&mut tcp) {
                    return Err(err.to_string());
                }
            }
            Ok(())
        });

        let refused = UreqClient::new()
            .post(&Request {
                url: &url,
                headers: vec![("Content-Type", "application/json")],
                body: r#"{"events":[]}"#,
            })
            .expect_err("a certificate the pinned roots do not trust was accepted");
        assert!(
            refused.contains("invalid peer certificate") && refused.contains("UnknownIssuer"),
            "{refused}"
        );
        assert!(
            server.join().expect("server thread").is_err(),
            "the server finished a handshake the client should have aborted"
        );
    }

    /// SHA-256 of each root in `ingest-roots.pem`, in file order: ISRG Root
    /// X1, X2, YE and YR (Let's Encrypt); GTS Root R1 to R4 (Google Trust
    /// Services);
    /// SSL.com TLS RSA and ECC Root CA 2022, SSL.com Root Certification
    /// Authority RSA and ECC; Sectigo Public Server Authentication Root R46
    /// and E46, USERTrust RSA and ECC Certification Authority (Sectigo).
    const INGEST_ROOT_FINGERPRINTS: [&str; 16] = [
        "96bcec06264976f37460779acf28c5a7cfe8a3c0aae11a8ffcee05c0bddf08c6",
        "69729b8e15a86efc177a57afb7171dfc64add28c2fca8cf1507e34453ccb1470",
        "e14ffcad5b0025731006caa43a121a22d8e9700f4fb9cf852f02a708aa5d5666",
        "e57b7e6f150c419102e8d5c055729ff967b9d1a829bf00cec89ca604ebf4a86f",
        "d947432abde7b7fa90fc2e6b59101b1280e0e1c7e4e40fa3c6887fff57a7f4cf",
        "8d25cd97229dbf70356bda4eb3cc734031e24cf00fafcfd32dc76eb5841c7ea8",
        "34d8a73ee208d9bcdb0d956520934b4e40e69482596e8b6f73c8426b010a6f48",
        "349dfa4058c5e263123b398ae795573c4e1313c83fe68f93556cd5e8031b3c7d",
        "8faf7d2e2cb4709bb8e0b33666bf75a5dd45b5de480f8ea8d4bfe6bebc17f2ed",
        "c32ffd9f46f936d16c3673990959434b9ad60aafbb9e7cf33654f144cc1ba143",
        "85666a562ee0be5ce925c1d8890a6f76a87ec16d4d7d5f29ea7419cf20123b69",
        "3417bb06cc6007da1b961c920b8ab4ce3fad820e4aa30b9acbc4a74ebdcebc65",
        "7bb647a62aeeac88bf257aa522d01ffea395e0ab45c73f93f65654ec38f25a06",
        "c90f26f0fb1b4018b22227519b5ca2b53e2ca5b3be5cf18efe1bef47380c5383",
        "e793c9b02fd8aa13e21c31228accb08119643b749c898964b1746d46c3d4cbd2",
        "4ff460d54b9c86dabfbcfc5712e0400d2bed3fbc4d4fbdaa86e06adcd2a9ad7a",
    ];

    /// The setting itself is pinned: exactly the sixteen roots in
    /// `ingest-roots.pem`, byte for byte, with verification on and the crypto
    /// provider named. The behaviour is pinned by the captured-chain test
    /// below.
    #[test]
    fn the_agent_validates_against_the_pinned_ingest_roots_only() {
        use sha2::Digest;

        let client = UreqClient::new();
        let tls = client.agent.config().tls_config();
        let ureq::tls::RootCerts::Specific(roots) = tls.root_certs() else {
            panic!("the agent trusts a root set other than the pinned one");
        };
        let fingerprints: Vec<String> = roots
            .iter()
            .map(|cert| hex::encode(sha2::Sha256::digest(cert.der())))
            .collect();
        assert_eq!(fingerprints, INGEST_ROOT_FINGERPRINTS);
        assert!(!tls.disable_verification());
        assert!(tls.unversioned_rustls_crypto_provider().is_some());
    }

    /// Every PEM block in the file is a certificate the agent takes, so a
    /// malformed block cannot silently shrink the set the test above pins.
    #[test]
    fn every_block_in_the_roots_file_is_a_certificate() {
        let blocks = std::str::from_utf8(INGEST_ROOTS_PEM)
            .expect("ASCII")
            .matches("-----BEGIN CERTIFICATE-----")
            .count();
        let parsed = ureq::tls::parse_pem(INGEST_ROOTS_PEM)
            .map(|item| item.expect("a well-formed PEM block"))
            .filter(|item| matches!(item, ureq::tls::PemItem::Certificate(_)))
            .count();
        assert_eq!((blocks, parsed), (16, 16));
    }

    /// The pinned roots as DER, the form rustls takes them in.
    fn pinned_roots_der() -> Vec<rustls::pki_types::CertificateDer<'static>> {
        ureq::tls::parse_pem(INGEST_ROOTS_PEM)
            .filter_map(|item| match item {
                Ok(ureq::tls::PemItem::Certificate(cert)) => {
                    Some(rustls::pki_types::CertificateDer::from(cert.der().to_vec()))
                }
                _ => None,
            })
            .collect()
    }

    /// Every pinned root becomes a rustls trust anchor. ureq hands the set to
    /// `RootCertStore::add_parsable_certificates`, which skips a certificate
    /// it cannot use and says so only in a debug log, so the configuration
    /// test above could pass with an anchor rustls never loaded.
    #[test]
    fn every_pinned_root_becomes_a_rustls_trust_anchor() {
        let mut store = rustls::RootCertStore::empty();
        let (added, ignored) = store.add_parsable_certificates(pinned_roots_der());
        assert_eq!((added, ignored), (16, 0));
    }

    /// The ingest's real chain, captured on 2026-10-21 (the leaf, Let's
    /// Encrypt YE1, ISRG Root YE cross-signed by X2, and X2 cross-signed by
    /// X1), verified offline against the pinned roots at a fixed time inside
    /// the leaf's validity. It is checked twice: as served, and cut at Root
    /// YE, the shorter chain Let's Encrypt also publishes. The second is why
    /// ISRG Root YE is pinned. Without it that chain has no anchor, and an
    /// edge that stopped sending the cross-signs would silence telemetry on
    /// every shipped SDK.
    #[test]
    fn the_captured_ingest_chain_validates_as_served_and_cut_at_root_ye() {
        use rustls::client::danger::ServerCertVerifier;
        use rustls::pki_types::pem::PemObject;
        use rustls::pki_types::{CertificateDer, ServerName, UnixTime};

        const CHAIN: &[u8] = include_bytes!(concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/tests/fixtures/ingest-chain-2026-10.pem"
        ));
        let chain: Vec<CertificateDer<'static>> = CertificateDer::pem_slice_iter(CHAIN)
            .collect::<Result<_, _>>()
            .expect("the chain fixture parses");
        let [leaf, ye1, ye_by_x2, x2_by_x1] = chain.as_slice() else {
            panic!(
                "the chain fixture holds {} certificates, not 4",
                chain.len()
            );
        };

        let mut roots = rustls::RootCertStore::empty();
        roots.add_parsable_certificates(pinned_roots_der());
        let verifier = rustls::client::WebPkiServerVerifier::builder_with_provider(
            std::sync::Arc::new(roots),
            std::sync::Arc::new(rustls::crypto::ring::default_provider()),
        )
        .build()
        .expect("a verifier over the pinned roots");
        // 2026-10-21T00:00:00Z, inside every certificate's validity.
        let now = UnixTime::since_unix_epoch(Duration::from_secs(1_792_540_800));
        let host = ServerName::try_from("analytics.offlineprotocol.com").expect("a DNS name");

        let served = vec![ye1.clone(), ye_by_x2.clone(), x2_by_x1.clone()];
        let cut_at_root_ye = vec![ye1.clone()];
        for (shape, intermediates) in [("as served", served), ("cut at Root YE", cut_at_root_ye)] {
            verifier
                .verify_server_cert(leaf, &intermediates, &host, &[], now)
                .unwrap_or_else(|err| panic!("the chain {shape} was refused: {err}"));
        }
    }

    /// The real ingest's chain validates against the pinned roots. Needs the
    /// network, so it runs only on request:
    /// `cargo test -p offline-protocol -- --ignored the_ingest_chain`. The
    /// `Ingest Chain` workflow runs it daily.
    /// Any HTTP answer passes (an unauthenticated batch is refused with a
    /// 4xx); only a TLS or connection failure fails it.
    #[test]
    #[ignore = "needs the network"]
    fn the_ingest_chain_validates_against_the_pinned_roots() {
        let answer = UreqClient::new().post(&Request {
            url: TELEMETRY_INGEST_URL,
            headers: vec![("Content-Type", "application/json")],
            body: r#"{"events":[]}"#,
        });
        assert!(answer.is_ok(), "{answer:?}");
    }

    #[test]
    fn battery_deferral_applies_below_fifteen_percent_unless_charging() {
        assert!(battery_defers(Some(14), false));
        assert!(!battery_defers(Some(14), true));
        assert!(!battery_defers(Some(15), false));
        assert!(!battery_defers(None, false));
    }
}