sioul_sync/
stalls.rs

1// SPDX-License-Identifier: GPL-3.0-or-later
2// Copyright © 2026 Aurélien Pierre
3
4//! The one place Sioul builds on ureq's transport, `ureq::unversioned`, which
5//! ureq keeps out of its semver promise: any ureq release may change it.
6//! Written for ureq 3.4 (3.4.2 in `Cargo.lock`, 9 October 2026). When an
7//! update breaks the build here, this file alone changes; what it does stays
8//! the same: each wait for the network bounded, not only each phase of a
9//! request.
10//!
11//! ureq's own limits count a whole phase of a request (all of an answer's
12//! body, for instance): an answer trickling in on a weak signal for longer
13//! than that is cut although it moves, and one that stops is waited for until
14//! the phase's end. Here each read and each write waits `stall` at most: an
15//! answer goes on as long as it moves, and one that stops is given up soon.
16//!
17//! A shared agent (`shared_agent`) keeps its connections for requests that
18//! ask different waits: each request says its own first, on its thread
19//! (`ask`), since a connection kept in ureq's pool serves the next request
20//! whichever opened it.
21
22use std::cell::Cell;
23use std::time::Duration;
24use ureq::unversioned::resolver::DefaultResolver;
25use ureq::unversioned::transport::{Buffers, ConnectionDetails, Connector, DefaultConnector, NextTimeout, Transport};
26
27/// An agent with ureq's `config`, each wait for the network `stall` at most;
28/// `patient`: the wait for an answer is not capped (a file sent whole, whose
29/// bytes may wait in the system's buffers on a slow uplink while the answer
30/// is awaited: bounded by the request's whole budget alone).
31pub(crate) fn agent(config: ureq::config::Config, stall: Duration, patient: bool) -> ureq::Agent {
32    ureq::Agent::with_parts(config, DefaultConnector::new().chain(Stalls { stall, patient, asked: false }), DefaultResolver::default())
33}
34
35/// An agent whose connections serve requests that ask different waits (the
36/// sharing's one agent per process, `remote`): each wait as the request
37/// running on this thread asked (`ask`); none asked yet on it, `stall`, not
38/// patient.
39pub(crate) fn shared_agent(config: ureq::config::Config, stall: Duration) -> ureq::Agent {
40    ureq::Agent::with_parts(config, DefaultConnector::new().chain(Stalls { stall, patient: false, asked: true }), DefaultResolver::default())
41}
42
43thread_local! {
44    /// What the last request run on this thread through a shared agent asked (`ask`).
45    static ASKED: Cell<Option<(Duration, bool)>> = const { Cell::new(None) };
46}
47
48/// The waits of the next requests this thread runs through a shared agent
49/// (`shared_agent`), their answers' bodies read included: each read and write
50/// `stall` at most, the wait for an answer not capped when `patient`. Said
51/// before each request; holds until the next is said. Other agents never read it.
52pub(crate) fn ask(stall: Duration, patient: bool) {
53    ASKED.set(Some((stall, patient)));
54}
55
56/// Each wait for the network, at most `stall`; `patient`: the waits for an
57/// answer are not capped (`agent`'s); `asked`: both as the request running on
58/// the thread asked (`shared_agent`).
59#[derive(Debug)]
60struct Stalls {
61    stall: Duration,
62    patient: bool,
63    asked: bool,
64}
65
66impl Connector<Box<dyn Transport>> for Stalls {
67    type Out = Stalling;
68
69    fn connect(&self, _: &ConnectionDetails, chained: Option<Box<dyn Transport>>) -> Result<Option<Stalling>, ureq::Error> {
70        Ok(chained.map(|transport| Stalling { transport, stall: self.stall, patient: self.patient, asked: self.asked }))
71    }
72}
73
74#[derive(Debug)]
75struct Stalling {
76    transport: Box<dyn Transport>,
77    stall: Duration,
78    patient: bool,
79    asked: bool,
80}
81
82impl Stalling {
83    /// Its stall and whether it is patient, now: as asked on this thread, for a shared agent's.
84    fn waits(&self) -> (Duration, bool) {
85        if self.asked { ASKED.get().unwrap_or((self.stall, self.patient)) } else { (self.stall, self.patient) }
86    }
87
88    fn within(timeout: NextTimeout, stall: Duration) -> NextTimeout {
89        NextTimeout { after: timeout.after.min(stall.into()), reason: timeout.reason }
90    }
91}
92
93impl Transport for Stalling {
94    fn buffers(&mut self) -> &mut dyn Buffers {
95        self.transport.buffers()
96    }
97
98    fn transmit_output(&mut self, amount: usize, timeout: NextTimeout) -> Result<(), ureq::Error> {
99        let timeout = Stalling::within(timeout, self.waits().0);
100        self.transport.transmit_output(amount, timeout)
101    }
102
103    fn await_input(&mut self, timeout: NextTimeout) -> Result<bool, ureq::Error> {
104        // An upload's answer, awaited while its last bytes drain from the
105        // system's buffers on a slow uplink: within the whole budget alone.
106        let (stall, patient) = self.waits();
107        let timeout = if patient { timeout } else { Stalling::within(timeout, stall) };
108        self.transport.await_input(timeout)
109    }
110
111    fn is_open(&mut self) -> bool {
112        self.transport.is_open()
113    }
114
115    fn is_tls(&self) -> bool {
116        self.transport.is_tls()
117    }
118}
sioul_sync/stalls.rs

sioul_sync/
stalls.rs
