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
//! What:
//! The repositories the service must look at again, and when: those with a claim that another instance holds and that may run
//! out its lease while this service sits idle.
//! Why:
//! A restarted service that finds the claims of the instance it replaces cannot take them before their lease is over, and
//! nothing wakes it when the lease is: the next commit might be hours away, and the work would sit claimed by a holder that
//! died. The work loop asks this schedule how long to wait for a wake before it looks again.
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! const wait = schedule.nextWait(now); const due = schedule.takeDue(now); schedule.replace(ran, retries, now);
//! ```
use std::path::PathBuf;
use std::time::{Duration, Instant};
/// Added to a wait so that the lease is over when the service looks, not exactly at its edge.
pub const RETRY_SLACK: Duration = Duration::from_secs(1);
/// What:
/// The repositories to look at again, each with the time to do it.
/// Why:
/// One entry per repository, replaced by the newest answer of a round that looked at it.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// type RetrySchedule = Map<string, number>;
/// ```
#[derive(Debug, Default)]
pub struct RetrySchedule {
/// The repositories and when each is due.
due: Vec<(PathBuf, Instant)>,
}
/// Methods over the schedule.
impl RetrySchedule {
/// Whether nothing is waiting to be looked at again.
pub fn is_empty(&self) -> bool {
return self.due.is_empty();
}
/// How long until the first entry is due (nothing at all when the schedule is empty, zero for one already due).
pub fn next_wait(&self, now: Instant) -> Option<Duration> {
return self.due.iter().map(|(_, when)| { return when.saturating_duration_since(now); }).min();
}
/// Removes and returns the repositories that are due at `now`.
pub fn take_due(&mut self, now: Instant) -> Vec<PathBuf> {
let (due, waiting): (Vec<(PathBuf, Instant)>, Vec<(PathBuf, Instant)>) = std::mem::take(&mut self.due).into_iter().partition(|(_, when)| { return *when <= now; });
self.due = waiting;
return due.into_iter().map(|(directory, _)| { return directory; }).collect();
}
/// Forgets what was scheduled for the repositories a round just ran, and schedules what that round asked for.
pub fn replace(&mut self, ran: &[PathBuf], retries: Vec<(PathBuf, Duration)>, now: Instant) {
self.due.retain(|(directory, _)| { return !ran.contains(directory); });
for (directory, wait) in retries {
self.due.push((directory, now + wait + RETRY_SLACK));
}
}
}
/// Schedule controls stay out of the release executable.
#[cfg(test)]
#[path = "service_retry_tests.rs"]
mod tests;