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
//! When a command gives up (ADR 039 ยง Failure handling: reassign-only tail,
//! the stop policy).
//!
//! The clock measures time since the last verified byte anywhere in the
//! command, never time since the start. A command in a terminal has no limit:
//! the human watches the bar and presses Ctrl-C. A script gets
//! [`SCRIPT_GIVE_UP`](crate::stop::SCRIPT_GIVE_UP). `--give-up-after-secs`
//! overrides both.
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Mutex, PoisonError};
use tokio::time::{Duration, Instant};
use tokio_util::sync::CancellationToken;
/// The no-progress limit when stderr is not a terminal.
pub const SCRIPT_GIVE_UP: Duration = Duration::from_mins(10);
/// The instant of the command's last verified byte.
///
/// A [`ClockHold`] stops the clock: while any hold is alive the command counts
/// as making progress, for time spent waiting on the command's own consumer
/// rather than on a source.
#[derive(Debug)]
pub struct ProgressClock {
last: Mutex<Instant>,
holds: AtomicUsize,
}
impl ProgressClock {
/// A clock whose last progress is now.
#[must_use]
pub fn new() -> Self {
Self {
last: Mutex::new(Instant::now()),
holds: AtomicUsize::new(0),
}
}
/// Record a verified byte now.
pub fn tick(&self) {
*self.last.lock().unwrap_or_else(PoisonError::into_inner) = Instant::now();
}
/// The instant of the last verified byte, or now while a hold is alive.
#[must_use]
pub fn last(&self) -> Instant {
if self.holds.load(Ordering::Acquire) > 0 {
return Instant::now();
}
*self.last.lock().unwrap_or_else(PoisonError::into_inner)
}
/// Stop the clock until the returned hold drops. The drop counts as
/// progress, so the no-progress limit restarts from it.
#[must_use]
pub fn hold(&self) -> ClockHold<'_> {
self.holds.fetch_add(1, Ordering::AcqRel);
ClockHold(self)
}
}
/// Stops a [`ProgressClock`] while it lives. See [`ProgressClock::hold`].
#[derive(Debug)]
pub struct ClockHold<'a>(&'a ProgressClock);
impl Drop for ClockHold<'_> {
fn drop(&mut self) {
self.0.tick();
self.0.holds.fetch_sub(1, Ordering::AcqRel);
}
}
impl Default for ProgressClock {
fn default() -> Self {
Self::new()
}
}
/// When the command stops for lack of progress.
///
/// A bundle pull also halts it: once one entry ends the pull, every sibling
/// entry's fetch stops at its next wait or loop pass and lets its lane builds
/// drain, rather than being dropped in the middle of one. A recovery step in
/// progress runs to its outcome first. A clone shares the halt, which holds
/// for one pass of the pull.
#[derive(Debug, Clone)]
pub struct StopPolicy {
/// The no-progress limit. `None` waits until the caller interrupts.
pub give_up_after: Option<Duration>,
/// The command-wide progress clock.
pub clock: std::sync::Arc<ProgressClock>,
/// Cancelled once a sibling entry ends this pass of the pull. A clone
/// shares it; [`StopPolicy::next_pass`] starts a fresh one.
halt: CancellationToken,
}
impl StopPolicy {
/// The policy for a command whose stderr `is_terminal`, with the user's
/// optional `--give-up-after-secs` override.
#[must_use]
pub fn new(
is_terminal: bool,
give_up_after: Option<Duration>,
clock: std::sync::Arc<ProgressClock>,
) -> Self {
let give_up_after = give_up_after.or((!is_terminal).then_some(SCRIPT_GIVE_UP));
Self {
give_up_after,
clock,
halt: CancellationToken::new(),
}
}
/// The same policy, on the same clock, for the next pass of a pull: a
/// halt of this pass does not carry over.
#[must_use]
pub(crate) fn next_pass(&self) -> Self {
Self {
give_up_after: self.give_up_after,
clock: std::sync::Arc::clone(&self.clock),
halt: CancellationToken::new(),
}
}
/// Halt every fetch on this policy at its next wait or loop pass.
pub(crate) fn halt(&self) {
self.halt.cancel();
}
/// Whether this policy is halted.
pub(crate) fn is_halted(&self) -> bool {
self.halt.is_cancelled()
}
/// `Ok` until this policy is halted.
///
/// # Errors
///
/// [`Halted`] once it is.
pub(crate) fn ensure_running(&self) -> Result<(), Halted> {
if self.is_halted() {
Err(Halted)
} else {
Ok(())
}
}
/// Resolve once this policy is halted.
pub(crate) async fn halted(&self) -> Halted {
self.halt.cancelled().await;
Halted
}
/// Resolve once the command has made no progress for the limit. Never
/// resolves without a limit.
pub async fn expired(&self) -> GaveUp {
let Some(limit) = self.give_up_after else {
return std::future::pending().await;
};
loop {
let deadline = self.clock.last() + limit;
tokio::time::sleep_until(deadline).await;
if self.clock.last() + limit <= Instant::now() {
return GaveUp { idle: limit };
}
}
}
}
/// The command made no progress for `idle` and stopped. A rerun resumes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct GaveUp {
/// How long the command went without a verified byte.
pub idle: Duration,
}
impl std::fmt::Display for GaveUp {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let secs = self.idle.as_secs();
if secs >= 60 && secs.is_multiple_of(60) {
write!(f, "no progress for {}m; rerun to resume", secs / 60)
} else {
write!(f, "no progress for {secs}s; rerun to resume")
}
}
}
impl std::error::Error for GaveUp {}
/// A sibling entry ended the pull, so this entry's fetch stopped
/// ([`StopPolicy::halt`]). The pull ends with the sibling's error, so this
/// one reaches a caller only as the text a late lane build's escrow folds in.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct Halted;
impl std::fmt::Display for Halted {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("another entry ended the pull")
}
}
impl std::error::Error for Halted {}
#[cfg(test)]
mod tests;