decdn_client/
stop.rs

1//! When a command gives up (ADR 039 § Failure handling: reassign-only tail,
2//! the stop policy).
3//!
4//! The clock measures time since the last verified byte anywhere in the
5//! command, never time since the start. A command in a terminal has no limit:
6//! the human watches the bar and presses Ctrl-C. A script gets
7//! [`SCRIPT_GIVE_UP`](crate::stop::SCRIPT_GIVE_UP). `--give-up-after-secs`
8//! overrides both.
9
10use std::sync::atomic::{AtomicUsize, Ordering};
11use std::sync::{Mutex, PoisonError};
12
13use tokio::time::{Duration, Instant};
14
15/// The no-progress limit when stderr is not a terminal.
16pub const SCRIPT_GIVE_UP: Duration = Duration::from_mins(10);
17
18/// The instant of the command's last verified byte.
19///
20/// A [`ClockHold`] stops the clock: while any hold is alive the command counts
21/// as making progress, for time spent waiting on the command's own consumer
22/// rather than on a source.
23#[derive(Debug)]
24pub struct ProgressClock {
25    last: Mutex<Instant>,
26    holds: AtomicUsize,
27}
28
29impl ProgressClock {
30    /// A clock whose last progress is now.
31    #[must_use]
32    pub fn new() -> Self {
33        Self {
34            last: Mutex::new(Instant::now()),
35            holds: AtomicUsize::new(0),
36        }
37    }
38
39    /// Record a verified byte now.
40    pub fn tick(&self) {
41        *self.last.lock().unwrap_or_else(PoisonError::into_inner) = Instant::now();
42    }
43
44    /// The instant of the last verified byte, or now while a hold is alive.
45    #[must_use]
46    pub fn last(&self) -> Instant {
47        if self.holds.load(Ordering::Acquire) > 0 {
48            return Instant::now();
49        }
50        *self.last.lock().unwrap_or_else(PoisonError::into_inner)
51    }
52
53    /// Stop the clock until the returned hold drops. The drop counts as
54    /// progress, so the no-progress limit restarts from it.
55    #[must_use]
56    pub fn hold(&self) -> ClockHold<'_> {
57        self.holds.fetch_add(1, Ordering::AcqRel);
58        ClockHold(self)
59    }
60}
61
62/// Stops a [`ProgressClock`] while it lives. See [`ProgressClock::hold`].
63#[derive(Debug)]
64pub struct ClockHold<'a>(&'a ProgressClock);
65
66impl Drop for ClockHold<'_> {
67    fn drop(&mut self) {
68        self.0.tick();
69        self.0.holds.fetch_sub(1, Ordering::AcqRel);
70    }
71}
72
73impl Default for ProgressClock {
74    fn default() -> Self {
75        Self::new()
76    }
77}
78
79/// When the command stops for lack of progress.
80#[derive(Debug, Clone)]
81pub struct StopPolicy {
82    /// The no-progress limit. `None` waits until the caller interrupts.
83    pub give_up_after: Option<Duration>,
84    /// The command-wide progress clock.
85    pub clock: std::sync::Arc<ProgressClock>,
86}
87
88impl StopPolicy {
89    /// The policy for a command whose stderr `is_terminal`, with the user's
90    /// optional `--give-up-after-secs` override.
91    #[must_use]
92    pub fn new(
93        is_terminal: bool,
94        give_up_after: Option<Duration>,
95        clock: std::sync::Arc<ProgressClock>,
96    ) -> Self {
97        let give_up_after = give_up_after.or((!is_terminal).then_some(SCRIPT_GIVE_UP));
98        Self {
99            give_up_after,
100            clock,
101        }
102    }
103
104    /// Resolve once the command has made no progress for the limit. Never
105    /// resolves without a limit.
106    pub async fn expired(&self) -> GaveUp {
107        let Some(limit) = self.give_up_after else {
108            return std::future::pending().await;
109        };
110        loop {
111            let deadline = self.clock.last() + limit;
112            tokio::time::sleep_until(deadline).await;
113            if self.clock.last() + limit <= Instant::now() {
114                return GaveUp { idle: limit };
115            }
116        }
117    }
118}
119
120/// The command made no progress for `idle` and stopped. A rerun resumes.
121#[derive(Debug, Clone, Copy, PartialEq, Eq)]
122pub struct GaveUp {
123    /// How long the command went without a verified byte.
124    pub idle: Duration,
125}
126
127impl std::fmt::Display for GaveUp {
128    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
129        let secs = self.idle.as_secs();
130        if secs >= 60 && secs.is_multiple_of(60) {
131            write!(f, "no progress for {}m; rerun to resume", secs / 60)
132        } else {
133            write!(f, "no progress for {secs}s; rerun to resume")
134        }
135    }
136}
137
138impl std::error::Error for GaveUp {}
139
140#[cfg(test)]
141mod tests;
decdn_client/stop.rs

decdn_client/
stop.rs
