kithara-net 0.0.1-alpha4

Streaming HTTP for audio: retry, timeout, byte ranges.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
use std::{cmp::min, collections::HashMap, fmt};

use bitflags::bitflags;
use bon::Builder;
use kithara_bufpool::BytePool;
use kithara_platform::{time::Duration, traits::FromWithParams};

bitflags! {
    /// HTTP `Accept-Encoding` algorithms the client advertises and is
    /// willing to decode. Reqwest auto-adds the corresponding
    /// `Accept-Encoding` header for every algorithm whose flag is set;
    /// the rest are disabled via `ClientBuilder::no_*` so the wire
    /// header stays in lockstep with this set.
    ///
    /// Subset selection matters when talking to anti-bot WAFs that
    /// fingerprint clients by their exact `Accept-Encoding` value.
    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
    pub struct Compression: u8 {
        const GZIP    = 1 << 0;
        const DEFLATE = 1 << 1;
        const BROTLI  = 1 << 2;
        const ZSTD    = 1 << 3;
    }
}

/// TLS+HTTP fingerprint the native `client-wreq` backend impersonates. The
/// DRM keyserver sits behind an anti-bot WAF that fingerprints the TLS
/// `ClientHello` (JA3) and 418-rejects non-browser stacks; presenting a real
/// browser fingerprint via `wreq` is what gets a 200. Default `Safari` matches
/// iOS `URLSession`; Android selects a different preset. Inert under the
/// `client-reqwest` backend and on wasm32 (no emulation; the browser fetch
/// already carries a real fingerprint).
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
#[non_exhaustive]
pub enum ImpersonatePreset {
    #[default]
    Safari,
    Chrome,
}

#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Headers {
    inner: HashMap<String, String>,
}

impl Headers {
    #[must_use]
    // ast-grep-ignore: style.prefer-default-derive
    pub fn new() -> Self {
        Self::default()
    }

    pub fn get(&self, key: &str) -> Option<&str> {
        self.inner.get(key).map(String::as_str)
    }

    pub fn insert<K: Into<String>, V: Into<String>>(&mut self, key: K, value: V) {
        self.inner.insert(key.into(), value.into());
    }

    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }

    pub fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
        self.inner.iter().map(|(k, v)| (k.as_str(), v.as_str()))
    }
}

impl From<HashMap<String, String>> for Headers {
    fn from(map: HashMap<String, String>) -> Self {
        Self { inner: map }
    }
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RangeSpec {
    pub end: Option<u64>,
    pub start: u64,
}

impl RangeSpec {
    #[must_use]
    pub fn new(start: u64, end: Option<u64>) -> Self {
        Self { end, start }
    }
}

impl fmt::Display for RangeSpec {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self.end {
            Some(end) => write!(f, "bytes={}-{}", self.start, end),
            None => write!(f, "bytes={}-", self.start),
        }
    }
}

#[derive(Clone, Debug, Builder)]
#[non_exhaustive]
pub struct RetryPolicy {
    #[builder(default = Duration::from_millis(100))]
    pub base_delay: Duration,
    #[builder(default = Duration::from_secs(5))]
    pub max_delay: Duration,
    #[builder(default = 3)]
    pub max_retries: u32,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self::builder().build()
    }
}

impl RetryPolicy {
    #[must_use]
    pub fn new(max_retries: u32, base_delay: Duration, max_delay: Duration) -> Self {
        Self {
            base_delay,
            max_delay,
            max_retries,
        }
    }

    #[must_use]
    pub fn delay_for_attempt(&self, attempt: u32) -> Duration {
        const BACKOFF_BASE: u32 = 2;

        if attempt == 0 {
            return Duration::ZERO;
        }
        let exponential_delay = self.base_delay * BACKOFF_BASE.pow(attempt.saturating_sub(1));
        min(exponential_delay, self.max_delay)
    }
}

#[derive(Clone, Debug, Builder)]
#[non_exhaustive]
pub struct NetOptions {
    /// `Accept-Encoding` algorithms the client offers and auto-decodes.
    /// Defaults to all four (`gzip | deflate | brotli | zstd`); narrow it
    /// when an upstream rejects the full set (anti-bot WAFs that
    /// fingerprint on the exact `Accept-Encoding` string are a common
    /// reason).
    #[builder(default = Compression::all())]
    pub compression: Compression,
    /// Maximum allowed inactivity between consecutive read operations.
    /// Maps to [`reqwest::ClientBuilder::read_timeout`] (documented as
    /// "The timeout applies to each read operation, and resets after a
    /// successful read") and also drives the Downloader-layer
    /// `BodyStream` chunk-inactivity guard for the same semantics one
    /// layer down.
    ///
    /// Protects against zombie connections that send headers but then
    /// stop streaming bytes. Does **not** cap the total request
    /// lifetime; a legitimately slow stream that keeps delivering
    /// chunks (even one byte every few seconds) is not aborted.
    /// Default 30s is sized to absorb realistic mobile-network stalls
    /// (TCP retransmits, captive-portal warm-up, server-side TTFB
    /// spikes) without aborting valid slow streams — the player's
    /// contract is "wait for the segment, regardless of connection
    /// speed", and a 10s cap raced real fixtures.
    #[builder(default = Duration::from_secs(30))]
    pub inactivity_timeout: Duration,
    /// Hard cap on total request lifetime. Maps to
    /// [`reqwest::RequestBuilder::timeout`]. `None` lets streaming
    /// downloads run indefinitely as long as `inactivity_timeout` is
    /// satisfied — required for the player to honour "wait for the
    /// segment, regardless of connection speed". Default `Some(2 min)`
    /// keeps a safety net against pathological cases (server stuck in
    /// mid-body without ever closing) while not racing realistic
    /// slow-network seeks.
    pub total_timeout: Option<Duration>,
    #[builder(default)]
    pub retry_policy: RetryPolicy,
    /// Accept invalid TLS certificates (self-signed, expired, wrong hostname).
    /// **Security risk** — use only for local development and test servers.
    #[builder(default)]
    pub is_insecure: bool,
    /// Max idle connections per host. Enables HTTP keep-alive connection
    /// reuse, reducing `TIME_WAIT` accumulation under high request volume.
    /// Set to 0 to disable pooling.
    #[builder(default = 8)]
    pub pool_max_idle_per_host: usize,
    /// Apple `NSURLSession` streaming body queue capacity, measured in
    /// delivered data chunks waiting for Rust consumption. Set to 0 to disable
    /// `URLSession` task suspension for queued body chunks.
    #[builder(default = 32)]
    pub body_queue_capacity: usize,
    /// Queue length at or below which a suspended Apple streaming task resumes.
    /// Values greater than or equal to [`Self::body_queue_capacity`] are valid:
    /// they resume as soon as the consumer drains one chunk.
    #[builder(default = 16)]
    pub body_queue_resume_at: usize,
    /// Shared byte buffer pool used by backends that must copy platform-owned
    /// response buffers before handing bytes to Rust consumers.
    pub byte_pool: Option<BytePool>,
    /// Browser TLS+HTTP2 fingerprint the native `client-wreq` backend
    /// impersonates. Defaults to `Safari`. Ignored by the `client-reqwest`
    /// backend and on wasm32 (no emulation there).
    #[builder(default)]
    pub impersonate: ImpersonatePreset,
}

impl Default for NetOptions {
    fn default() -> Self {
        Self::builder()
            .total_timeout(Duration::from_secs(120))
            .build()
    }
}

impl FromWithParams<Self, Option<BytePool>> for NetOptions {
    fn build(options: Self, byte_pool: Option<BytePool>) -> Self {
        Self::builder()
            .compression(options.compression)
            .inactivity_timeout(options.inactivity_timeout)
            .maybe_total_timeout(options.total_timeout)
            .retry_policy(options.retry_policy)
            .is_insecure(options.is_insecure)
            .pool_max_idle_per_host(options.pool_max_idle_per_host)
            .body_queue_capacity(options.body_queue_capacity)
            .body_queue_resume_at(options.body_queue_resume_at)
            .maybe_byte_pool(options.byte_pool.or(byte_pool))
            .impersonate(options.impersonate)
            .build()
    }
}

#[cfg(test)]
mod tests {
    mod kithara {
        pub(crate) use kithara_test_macros::test;
    }

    use super::*;

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case::empty_headers(Headers::new(), true)]
    #[case::headers_with_values({
        let mut h = Headers::new();
        h.insert("key1", "value1");
        h.insert("key2", "value2");
        h
    }, false)]
    async fn test_headers_is_empty(#[case] headers: Headers, #[case] expected_empty: bool) {
        assert_eq!(headers.is_empty(), expected_empty);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case::simple_key("key1", "value1")]
    #[case::content_type("Content-Type", "application/json")]
    #[case::custom_header("X-Custom-Header", "custom-value")]
    async fn test_headers_insert_and_get(#[case] key: &str, #[case] value: &str) {
        let mut headers = Headers::new();
        headers.insert(key, value);

        assert_eq!(headers.get(key), Some(value));
        assert_eq!(headers.get("non-existent"), None);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    async fn test_headers_iter() {
        let mut headers = Headers::new();
        headers.insert("key1", "value1");
        headers.insert("key2", "value2");
        headers.insert("key3", "value3");

        let mut iterated = HashMap::new();
        for (k, v) in headers.iter() {
            iterated.insert(k.to_string(), v.to_string());
        }

        assert_eq!(iterated.len(), 3);
        assert_eq!(iterated.get("key1"), Some(&"value1".to_string()));
        assert_eq!(iterated.get("key2"), Some(&"value2".to_string()));
        assert_eq!(iterated.get("key3"), Some(&"value3".to_string()));
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    async fn test_headers_from_hashmap() {
        let mut map = HashMap::new();
        map.insert("key1".to_string(), "value1".to_string());
        map.insert("key2".to_string(), "value2".to_string());

        let headers: Headers = map.into();

        assert!(!headers.is_empty());
        assert_eq!(headers.get("key1"), Some("value1"));
        assert_eq!(headers.get("key2"), Some("value2"));
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    async fn test_headers_default() {
        let headers = Headers::default();
        assert!(headers.is_empty());
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case::full_range(0, Some(100), "bytes=0-100")]
    #[case::open_ended(50, None, "bytes=50-")]
    #[case::single_byte(10, Some(10), "bytes=10-10")]
    #[case::zero_length(0, Some(0), "bytes=0-0")]
    async fn test_range_spec_to_header_value(
        #[case] start: u64,
        #[case] end: Option<u64>,
        #[case] expected_header: &str,
    ) {
        let range = RangeSpec::new(start, end);
        assert_eq!(range.to_string(), expected_header);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case::equal_ranges(RangeSpec::new(0, Some(100)), RangeSpec::new(0, Some(100)), true)]
    #[case::different_starts(RangeSpec::new(0, Some(100)), RangeSpec::new(1, Some(100)), false)]
    #[case::different_ends(RangeSpec::new(0, Some(100)), RangeSpec::new(0, Some(99)), false)]
    #[case::one_open_ended(RangeSpec::new(0, None), RangeSpec::new(0, None), true)]
    #[case::mixed_ends(RangeSpec::new(0, Some(100)), RangeSpec::new(0, None), false)]
    async fn test_range_spec_partial_eq(
        #[case] range1: RangeSpec,
        #[case] range2: RangeSpec,
        #[case] expected_equal: bool,
    ) {
        assert_eq!(range1 == range2, expected_equal);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    async fn test_range_spec_debug() {
        let range = RangeSpec::new(10, Some(20));
        let debug_output = format!("{:?}", range);
        assert!(debug_output.contains("RangeSpec"));
        assert!(debug_output.contains("start: 10"));
        assert!(debug_output.contains("end: Some(20)"));
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    async fn test_range_spec_clone() {
        let range1 = RangeSpec::new(10, Some(20));
        let range2 = range1.clone();

        assert_eq!(range1, range2);
        assert_eq!(range1.start, range2.start);
        assert_eq!(range1.end, range2.end);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    async fn test_retry_policy_default() {
        let policy = RetryPolicy::default();

        assert_eq!(policy.max_retries, 3);
        assert_eq!(policy.base_delay, Duration::from_millis(100));
        assert_eq!(policy.max_delay, Duration::from_secs(5));
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case(1, Duration::from_millis(50), Duration::from_secs(1))]
    #[case(5, Duration::from_millis(100), Duration::from_secs(2))]
    #[case(10, Duration::from_millis(200), Duration::from_secs(10))]
    async fn test_retry_policy_new(
        #[case] max_retries: u32,
        #[case] base_delay: Duration,
        #[case] max_delay: Duration,
    ) {
        let policy = RetryPolicy::new(max_retries, base_delay, max_delay);

        assert_eq!(policy.max_retries, max_retries);
        assert_eq!(policy.base_delay, base_delay);
        assert_eq!(policy.max_delay, max_delay);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case(0, Duration::ZERO)]
    #[case(1, Duration::from_millis(100))]
    #[case(2, Duration::from_millis(200))]
    #[case(3, Duration::from_millis(400))]
    #[case(4, Duration::from_millis(800))]
    #[case(5, Duration::from_millis(1600))]
    #[case(10, Duration::from_secs(5))]
    #[case(20, Duration::from_secs(5))]
    async fn test_retry_policy_delay_for_attempt_default(
        #[case] attempt: u32,
        #[case] expected_delay: Duration,
    ) {
        let policy = RetryPolicy::default();
        let delay = policy.delay_for_attempt(attempt);

        assert_eq!(delay, expected_delay);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case(
        1,
        Duration::from_millis(50),
        Duration::from_millis(200),
        0,
        Duration::ZERO
    )]
    #[case(
        1,
        Duration::from_millis(50),
        Duration::from_millis(200),
        1,
        Duration::from_millis(50)
    )]
    #[case(
        1,
        Duration::from_millis(50),
        Duration::from_millis(200),
        2,
        Duration::from_millis(100)
    )]
    #[case(
        1,
        Duration::from_millis(50),
        Duration::from_millis(200),
        3,
        Duration::from_millis(200)
    )]
    #[case(
        1,
        Duration::from_millis(50),
        Duration::from_millis(200),
        4,
        Duration::from_millis(200)
    )]
    async fn test_retry_policy_delay_for_attempt_custom(
        #[case] max_retries: u32,
        #[case] base_delay: Duration,
        #[case] max_delay: Duration,
        #[case] attempt: u32,
        #[case] expected_delay: Duration,
    ) {
        let policy = RetryPolicy::new(max_retries, base_delay, max_delay);
        let delay = policy.delay_for_attempt(attempt);

        assert_eq!(delay, expected_delay);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    async fn test_retry_policy_debug() {
        let policy = RetryPolicy::default();
        let debug_output = format!("{:?}", policy);

        assert!(debug_output.contains("RetryPolicy"));
        assert!(debug_output.contains("max_retries: 3"));
        assert!(debug_output.contains("base_delay"));
        assert!(debug_output.contains("max_delay"));
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    async fn test_retry_policy_clone() {
        let policy1 = RetryPolicy::new(5, Duration::from_millis(100), Duration::from_secs(2));
        let policy2 = policy1.clone();

        assert_eq!(policy1.max_retries, policy2.max_retries);
        assert_eq!(policy1.base_delay, policy2.base_delay);
        assert_eq!(policy1.max_delay, policy2.max_delay);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case::start_equals_end(10, Some(10), "bytes=10-10")]
    #[case::start_greater_than_end(20, Some(10), "bytes=20-10")]
    #[case::max_values(u64::MAX, Some(u64::MAX), &format!("bytes={}-{}", u64::MAX, u64::MAX))]
    async fn test_range_spec_edge_cases(
        #[case] start: u64,
        #[case] end: Option<u64>,
        #[case] expected_header: &str,
    ) {
        let range = RangeSpec::new(start, end);
        assert_eq!(range.to_string(), expected_header);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case::zero_max_retries(0, Duration::from_millis(100), Duration::from_secs(1))]
    #[case::large_max_retries(100, Duration::from_millis(10), Duration::from_secs(10))]
    #[case::zero_base_delay(3, Duration::ZERO, Duration::from_secs(1))]
    #[case::zero_max_delay(3, Duration::from_millis(100), Duration::ZERO)]
    async fn test_retry_policy_edge_cases(
        #[case] max_retries: u32,
        #[case] base_delay: Duration,
        #[case] max_delay: Duration,
    ) {
        let policy = RetryPolicy::new(max_retries, base_delay, max_delay);

        for attempt in 0..=5 {
            let delay = policy.delay_for_attempt(attempt);

            assert!(delay >= Duration::ZERO);

            assert!(delay <= max_delay);

            if base_delay == Duration::ZERO {
                assert_eq!(delay, Duration::ZERO);
            }
        }
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case(10)]
    #[case(20)]
    async fn test_retry_policy_large_attempts(#[case] attempt: u32) {
        let policy = RetryPolicy::default();

        let delay = policy.delay_for_attempt(attempt);

        assert!(delay <= policy.max_delay);
        assert!(delay >= Duration::ZERO);
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    #[case::with_spaces("X-Custom Header", "value with spaces")]
    #[case::with_unicode("X-Emoji", "🎉")]
    #[case::with_special_chars("X-Special", "a\tb\nc")]
    #[case::empty_value("X-Empty", "")]
    async fn test_headers_special_characters(#[case] key: &str, #[case] value: &str) {
        let mut headers = Headers::new();
        headers.insert(key, value);

        assert_eq!(headers.get(key), Some(value));
    }

    #[kithara::test(tokio, timeout(Duration::from_secs(5)))]
    async fn test_headers_case_sensitive() {
        let mut headers = Headers::new();
        headers.insert("Content-Type", "application/json");
        headers.insert("content-type", "text/plain");

        assert_eq!(headers.get("Content-Type"), Some("application/json"));
        assert_eq!(headers.get("content-type"), Some("text/plain"));
        assert_ne!(headers.get("Content-Type"), headers.get("content-type"));
    }
}