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
//! What:
//! Reaches the commit service: connects to its address, starts it through its unit when nothing answers, and waits for a
//! service of the current release.
//! Why:
//! Submitting a request or a withdrawal begins by reaching a service that speaks this release; the steps have their own file
//! so the submit loop stays readable.
//!
//! In TS you'd write (pseudocode):
//! ```ts
//! const reader = await reachCurrent(config, notices, announced, trace, serviceSeen);
//! ```
use crate::service_client_settings_and_watches::{Activation, CONNECT_PAUSE, ClientConfig, START_NOTICE_GRACE};
use crate::service_client_listen_and_ask_to_retire::{Heard, connect, listen};
use crate::commit_timing::ReachTrace;
use crate::service_messages::{ClientFailure, STARTING_NOTICE, UPDATING_NOTICE};
use crate::service_protocol::{
PROTOCOL_VERSION, RELEASE, ServiceMessage, encode_retire,
};
use std::io::{BufReader, Write};
use std::os::unix::net::UnixStream;
use std::time::{Duration, Instant};
/// What:
/// Connects, waiting for a start the operating system performs, and returns the stream with the service's version.
/// Why:
/// With a manager to ask, the wait has no clock: it ends when the service greets, or when the manager has said twice in a
/// row that the unit failed. Where the manager cannot be asked it ends at the start wait, as it always did. Once a service
/// has greeted during this command (`service_seen`), a service that then greets nothing within the silence bound is the
/// one that stopped, and that ends the wait.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// const { stream, version } = await reach(config, notices, serviceSeen)
/// ```
pub(crate) fn reach(
config: &ClientConfig,
notices: &mut dyn Write,
announced: &mut bool,
trace: &mut ReachTrace,
service_seen: bool,
) -> Result<(BufReader<UnixStream>, u32, Option<String>), ClientFailure> {
let started: Instant = Instant::now();
let mut refused_here: bool = false;
let mut last_asked: Instant = started;
let mut failed_answers: u32 = 0;
let mut cannot_ask: bool = false;
let mut last_start_request: Option<Instant> = None;
loop {
trace.tries += 1;
match connect(&config.address) {
Ok(stream) => {
let patience: Duration = match (&config.start_watch, service_seen) {
(_, true) => config.heartbeat_silence,
(Some(watch), false) => watch.every,
(None, false) => config.start_wait,
};
let _timeout = stream.set_read_timeout(Some(patience));
let mut reader: BufReader<UnixStream> = BufReader::new(stream);
match listen(&mut reader) {
Heard::Message(ServiceMessage::Welcome { version, release }) => {
let _widened = reader.get_ref().set_read_timeout(Some(config.heartbeat_silence));
trace.saw_release(release.as_deref());
if refused_here {
trace.start_waited += started.elapsed();
}
return Ok((reader, version, release));
}
// A service that greeted before and now greets nothing is the one that stopped.
Heard::Silent if service_seen => return Err(ClientFailure::ServiceSilent),
// Connected, but nothing greeted: the service is still starting or is going away.
Heard::Message(_) | Heard::Closed | Heard::Silent => {
trace.refused += 1;
refused_here = true;
}
}
}
Err(_) => match config.activation {
Activation::Unsupported(platform) => return Err(ClientFailure::NoActivation(platform)),
Activation::Managed => {
trace.refused += 1;
refused_here = true;
}
},
}
match &config.start_watch {
Some(watch) => {
if last_asked.elapsed() >= watch.every {
last_asked = Instant::now();
let answer: Option<String> = (watch.state)(watch.bus.as_path(), watch.unit.as_str());
// A unit still `inactive` after the start wait, with the client's connection waiting on its socket, is one the
// manager did not start; it is asked to, again after each further start wait, and the client goes on waiting.
// A unit that is `activating` is being started and is left alone.
if answer.as_deref() == Some("inactive")
&& started.elapsed() >= config.start_wait
&& match last_start_request {
Some(at) => at.elapsed() >= config.start_wait,
None => true,
}
{
let _taken = (watch.start)(watch.bus.as_path(), watch.unit.as_str());
last_start_request = Some(Instant::now());
}
cannot_ask = answer.is_none();
failed_answers = if answer.as_deref() == Some("failed") { failed_answers + 1 } else { 0 };
}
// One answer of "failed" can be the unit's last run, before the connection made the manager start it again. A start the
// manager reports as under way is waited for until the start ceiling and no longer (owner, 2026-11-03: "nothing
// should ever wait forever or indefinitely").
if failed_answers >= 2 || (cannot_ask && started.elapsed() >= config.start_wait) || started.elapsed() >= config.start_ceiling {
return Err(ClientFailure::DidNotStart);
}
}
None => {
if started.elapsed() >= config.start_wait {
return Err(ClientFailure::DidNotStart);
}
}
}
if !*announced && started.elapsed() >= START_NOTICE_GRACE {
*announced = true;
let _written = writeln!(notices, "{STARTING_NOTICE}");
}
std::thread::sleep(CONNECT_PAUSE);
}
}
/// What:
/// Reaches a service that speaks this wrapper's protocol and was started from this wrapper's release, asking any other to
/// retire first.
/// Why:
/// The manager keeps its socket across an upgrade or a downgrade, so a service from the other release may answer: one
/// whose program was since removed, or one that behaves as the other release did. It is asked to finish its batch and
/// exit, and the next connection makes the manager start the program the rewritten unit names. The asking is bounded by
/// the start wait from the first ask (the same wait the manager's start gets), because the retiring service still accepts connections until it has exited, and
/// each connection it accepts in that time is asked again. A service of a newer protocol cannot be asked (it refuses a
/// message it cannot read as its own), so that ends the command.
///
/// In TS you'd write (pseudocode):
/// ```ts
/// const reader = await reachCurrent(config, notices, announced)
/// ```
pub(crate) fn reach_current(config: &ClientConfig, notices: &mut dyn Write, announced: &mut bool, trace: &mut ReachTrace, service_seen: bool) -> Result<BufReader<UnixStream>, ClientFailure> {
let began: Instant = Instant::now();
let reached = reach_current_untimed(config, notices, announced, trace, service_seen);
trace.elapsed += began.elapsed();
return reached;
}
/// `reach_current` without the clock around it.
pub(crate) fn reach_current_untimed(config: &ClientConfig, notices: &mut dyn Write, announced: &mut bool, trace: &mut ReachTrace, service_seen: bool) -> Result<BufReader<UnixStream>, ClientFailure> {
let mut give_up_at: Option<Instant> = None;
loop {
let (mut reader, version, release) = reach(config, notices, announced, trace, service_seen)?;
if version == PROTOCOL_VERSION && release.as_deref() == Some(RELEASE) {
trace.answered_by = release;
return Ok(reader);
}
if version > PROTOCOL_VERSION {
return Err(ClientFailure::NoCommonVersion);
}
// The give-up time is fixed the first time an other-release service is met, and the person is told then.
let deadline: Instant = match give_up_at {
Some(fixed) => fixed,
None => {
let _written = writeln!(notices, "{UPDATING_NOTICE}");
let fresh: Instant = Instant::now() + config.start_wait;
give_up_at = Some(fresh);
fresh
}
};
if Instant::now() >= deadline {
return Err(ClientFailure::OtherRelease);
}
// Written in the version the service announced: a service of an earlier release reads a retire in no other.
let mut retire: String = encode_retire(version);
retire.push('\n');
let _sent = reader.get_mut().write_all(retire.as_bytes());
trace.retires_sent += 1;
std::thread::sleep(CONNECT_PAUSE * 4);
}
}