pixelpass/src/common/bandwidth.rs

//! One-shot upstream bandwidth measurement against Cloudflare's open
//! speed-test endpoint. POST a fixed payload, time it, derive Mbps.
//!
//! Run via `tokio::task::spawn_blocking` from async contexts — ureq is a
//! blocking client and we don't want to wedge the tokio runtime during
//! the test.

use anyhow::{Context, Result};
use std::time::{Duration, Instant};

const ENDPOINT: &str = "https://speed.cloudflare.com/__up";
const PAYLOAD_BYTES: usize = 5 * 1024 * 1024; // 5 MiB
const HTTP_TIMEOUT: Duration = Duration::from_secs(30);
/// Multiplier applied to the raw measurement. TCP slow-start, ramp-up, and
/// real-world contention all mean a one-shot upstream test slightly
/// overestimates sustainable throughput; clamp to 80% for headroom.
const SAFETY_FACTOR: f64 = 0.80;

/// Result of a successful measurement.
#[derive(Debug, Clone)]
pub struct Measurement {
    /// Raw measured throughput in megabits per second.
    pub raw_mbps: f64,
    /// `raw_mbps * SAFETY_FACTOR` — the value to use when sizing things.
    pub safe_mbps: f64,
    /// How long the upload took.
    pub elapsed: Duration,
}

/// Blocking upload-speed test. Call from a `spawn_blocking` task.
pub fn measure_upstream_blocking() -> Result<Measurement> {
    let payload = vec![0u8; PAYLOAD_BYTES];
    let agent = ureq::Agent::config_builder()
        .timeout_global(Some(HTTP_TIMEOUT))
        .build()
        .new_agent();

    let start = Instant::now();
    let response = agent
        .post(ENDPOINT)
        .content_type("application/octet-stream")
        .send(&payload[..])
        .context("upload request to Cloudflare failed")?;
    let elapsed = start.elapsed();

    let status = response.status();
    if !status.is_success() {
        anyhow::bail!("Cloudflare returned HTTP {status}");
    }

    let bits = (PAYLOAD_BYTES as f64) * 8.0;
    let seconds = elapsed.as_secs_f64().max(0.001);
    let raw_mbps = bits / seconds / 1_000_000.0;
    let safe_mbps = raw_mbps * SAFETY_FACTOR;

    Ok(Measurement {
        raw_mbps,
        safe_mbps,
        elapsed,
    })
}

/// Convert a safe-upstream Mbps figure plus the host's per-viewer bitrate
/// (kbps for video, ignoring audio + protocol overhead which we account for
/// via SAFETY_FACTOR) into a recommended viewer count. Floors to at least 1.
pub fn recommended_max_viewers(safe_mbps: f64, bitrate_kbps: u32) -> u32 {
    let per_viewer_mbps = (bitrate_kbps as f64) / 1000.0;
    // Guard non-finite / non-positive inputs (only reachable from a corrupted
    // config): a NaN safe_mbps would cast to 0 and an infinite one to u32::MAX,
    // both of which break the "at least 1" contract.
    if !safe_mbps.is_finite() || safe_mbps <= 0.0 || per_viewer_mbps <= 0.0 {
        return 1;
    }
    let n = (safe_mbps / per_viewer_mbps).floor();
    if n < 1.0 { 1 } else { n as u32 }
}

#[cfg(test)]
mod tests {
    use super::recommended_max_viewers;

    #[test]
    fn divides_bandwidth_by_per_viewer_bitrate() {
        // 8 Mbps safe / 2 Mbps each = 4 viewers.
        assert_eq!(recommended_max_viewers(8.0, 2000), 4);
        // Floors the fractional part: 7.9 / 2 = 3.95 -> 3.
        assert_eq!(recommended_max_viewers(7.9, 2000), 3);
    }

    #[test]
    fn floors_to_at_least_one() {
        // Not even enough for one viewer still allows one (best effort).
        assert_eq!(recommended_max_viewers(0.5, 2000), 1);
        // Zero / unknown bitrate can't size a budget; floor to one.
        assert_eq!(recommended_max_viewers(8.0, 0), 1);
    }

    #[test]
    fn degenerate_inputs_floor_to_one() {
        // A corrupted config must not yield 0 (NaN) or u32::MAX (Inf).
        assert_eq!(recommended_max_viewers(f64::NAN, 2000), 1);
        assert_eq!(recommended_max_viewers(f64::INFINITY, 2000), 1);
        assert_eq!(recommended_max_viewers(-5.0, 2000), 1);
    }
}