git_policy_cli/
service_client.rs

1//! What:
2//!  The wrapper's side of a commit through the resident service: connect, wait honestly while the operating system
3//!  starts the service, send the request, show the queue position, survive a service that dies mid-request, and turn
4//!  every failure into the plain text of `service_messages.rs`.
5//! Why:
6//!  A failure the system can resolve is resolved here without the user: a service that is starting is waited for, one
7//!  that crashed is reconnected to with the same request identity, one that is older than this wrapper is asked to retire
8//!  and is reached again once the manager has started the new one. Only what cannot be resolved ends the command.
9//!
10//! In TS you'd write (pseudocode):
11//! ```ts
12//! const result = await submit(config, request, stderr); // landed, or prepare again, or a ClientFailure
13//! ```
14
15use super::service_messages::{ClientFailure, Platform, STARTING_NOTICE, UPDATING_NOTICE, position_notice};
16use super::service_protocol::{
17    ClientMessage, CommitRequest, FailureCode, PROTOCOL_VERSION, ServiceMessage, decode_service, encode_client,
18};
19use super::service_server::Bind;
20use std::io::{BufRead, BufReader, Write};
21use std::os::unix::net::UnixStream;
22use std::time::{Duration, Instant};
23
24/// How long the client waits for the operating system to start the service.
25pub const START_WAIT: Duration = Duration::from_secs(10);
26
27/// How long a start takes before the client says it is starting the service.
28pub const START_NOTICE_GRACE: Duration = Duration::from_secs(1);
29
30/// How long a commit waits in the queue before the client gives up and says so.
31pub const COMMIT_WAIT: Duration = Duration::from_secs(10 * 60);
32
33/// How many times a request is sent again after the connection broke before the client says it cannot be sure.
34pub const RECONNECTS: usize = 5;
35
36/// How many times the client asks an older service to retire.
37pub const REPLACEMENTS: usize = 2;
38
39/// The pause between connection tries while the service starts.
40const CONNECT_PAUSE: Duration = Duration::from_millis(25);
41
42/// What the client may rely on the operating system to do.
43#[derive(Clone, Copy, Debug, Eq, PartialEq)]
44pub enum Activation {
45    /// The manager owns the socket and starts the service on demand.
46    Managed,
47    /// This platform cannot start the service on demand.
48    Unsupported(Platform),
49}
50
51/// What the client needs to reach the service.
52#[derive(Clone, Debug)]
53pub struct ClientConfig {
54    /// Where the service listens.
55    pub address: Bind,
56    /// What the operating system can do about a service that is not running.
57    pub activation: Activation,
58    /// Longest wait for a start.
59    pub start_wait: Duration,
60    /// Longest wait in the queue.
61    pub commit_wait: Duration,
62}
63
64/// What a request came to.
65#[derive(Clone, Debug, Eq, PartialEq)]
66pub enum Submitted {
67    /// The commit is on the branch.
68    Landed { commit: String, already: bool },
69    /// A named path changed on the branch: prepare again from this base and submit again with the same identity.
70    PrepareAgain { base: String },
71}
72
73fn connect(address: &Bind) -> std::io::Result<UnixStream> {
74    return match address {
75        #[cfg(target_os = "linux")]
76        Bind::Abstract(name) => {
77            use std::os::linux::net::SocketAddrExt;
78            UnixStream::connect_addr(&std::os::unix::net::SocketAddr::from_abstract_name(name.as_bytes())?)
79        }
80        #[cfg(not(target_os = "linux"))]
81        Bind::Abstract(_) => Err(std::io::Error::other("abstract sockets need Linux")),
82        Bind::Path(path) => UnixStream::connect(path),
83        Bind::Inherited => Err(std::io::Error::other("a client does not inherit the socket")),
84    };
85}
86
87fn read_message(reader: &mut BufReader<UnixStream>) -> Option<ServiceMessage> {
88    let mut line: String = String::new();
89    return match reader.read_line(&mut line) {
90        Ok(read) if read > 0 => decode_service(line.trim_end_matches('\n')).ok(),
91        _ => None,
92    };
93}
94
95/// What:
96///  Connects, waiting for a start the operating system performs, and returns the stream with the service's version.
97/// Why:
98///   The wait is bounded and honest: one notice after the grace, then silence until the bound.
99///
100/// In TS you'd write (pseudocode):
101/// ```ts
102/// const { stream, version } = await reach(config, notices)
103/// ```
104fn reach(config: &ClientConfig, notices: &mut dyn Write, announced: &mut bool) -> Result<(BufReader<UnixStream>, u32), ClientFailure> {
105    let started: Instant = Instant::now();
106    loop {
107        match connect(&config.address) {
108            Ok(stream) => {
109                let _timeout = stream.set_read_timeout(Some(config.start_wait));
110                let mut reader: BufReader<UnixStream> = BufReader::new(stream);
111                if let Some(ServiceMessage::Welcome { version }) = read_message(&mut reader) {
112                    let _widened = reader.get_ref().set_read_timeout(Some(config.commit_wait));
113                    return Ok((reader, version));
114                }
115            }
116            Err(_) => match config.activation {
117                Activation::Unsupported(platform) => return Err(ClientFailure::NoActivation(platform)),
118                Activation::Managed => {}
119            },
120        }
121        if started.elapsed() >= config.start_wait {
122            return Err(ClientFailure::DidNotStart);
123        }
124        if !*announced && started.elapsed() >= START_NOTICE_GRACE {
125            *announced = true;
126            let _written = writeln!(notices, "{STARTING_NOTICE}");
127        }
128        std::thread::sleep(CONNECT_PAUSE);
129    }
130}
131
132/// The user-facing failure for a code the service gave.
133pub fn failure_for(code: FailureCode, request: &CommitRequest) -> ClientFailure {
134    let repository: String = request.repository.display().to_string();
135    return match code {
136        FailureCode::RepositoryFull => ClientFailure::RepositoryFull(repository),
137        FailureCode::RepositoryReadOnly => ClientFailure::RepositoryReadOnly(repository),
138        FailureCode::IndexInUse => ClientFailure::IndexInUse(String::from("Another program")),
139        FailureCode::Rejected => ClientFailure::Refused,
140        FailureCode::BranchKeepsChanging => ClientFailure::BranchKeepsChanging {
141            branch: request.reference.trim_start_matches("refs/heads/").to_string(),
142            times: super::service_land::LANDING_ATTEMPTS,
143        },
144        FailureCode::Unavailable => ClientFailure::Unavailable,
145    };
146}
147
148/// What:
149///  Sends a commit request and waits until it is answered.
150/// Why:
151///   The same identity goes with every send, so a service that landed the commit and died before answering reports it
152///  as already landed, and nothing lands twice.
153///
154/// In TS you'd write (pseudocode):
155/// ```ts
156/// await submit(config, request, notices)
157/// ```
158pub fn submit(config: &ClientConfig, request: &CommitRequest, notices: &mut dyn Write) -> Result<Submitted, ClientFailure> {
159    let line: String = encode_client(&ClientMessage::Commit(Box::new(request.clone())));
160    let waiting_since: Instant = Instant::now();
161    let mut announced: bool = false;
162    let mut replacements: usize = 0;
163    let mut breaks: usize = 0;
164    let mut last_position: Option<usize> = None;
165    loop {
166        if waiting_since.elapsed() >= config.commit_wait {
167            return Err(ClientFailure::WaitedTooLong { minutes: config.commit_wait.as_secs() / 60 });
168        }
169        let (mut reader, version) = reach(config, notices, &mut announced)?;
170        if version != PROTOCOL_VERSION {
171            // An older service is asked to finish and exit; the manager starts the current one for the next connection.
172            if version > PROTOCOL_VERSION || replacements >= REPLACEMENTS {
173                return Err(ClientFailure::NoCommonVersion);
174            }
175            replacements += 1;
176            let _written = writeln!(notices, "{UPDATING_NOTICE}");
177            let mut retire: String = encode_client(&ClientMessage::Retire);
178            retire.push('\n');
179            let _sent = reader.get_mut().write_all(retire.as_bytes());
180            std::thread::sleep(CONNECT_PAUSE * 4);
181            continue;
182        }
183        let mut payload: String = line.clone();
184        payload.push('\n');
185        if reader.get_mut().write_all(payload.as_bytes()).is_err() {
186            breaks += 1;
187            if breaks > RECONNECTS {
188                return Err(ClientFailure::MayNotHaveLanded);
189            }
190            continue;
191        }
192        loop {
193            match read_message(&mut reader) {
194                Some(ServiceMessage::Position { number, of }) => {
195                    if last_position != Some(number) {
196                        last_position = Some(number);
197                        let _written = writeln!(notices, "{}", position_notice(number, of));
198                    }
199                }
200                Some(ServiceMessage::Landed { commit, already, .. }) => return Ok(Submitted::Landed { commit, already }),
201                Some(ServiceMessage::RetryPrepare { base, .. }) => return Ok(Submitted::PrepareAgain { base }),
202                Some(ServiceMessage::Failed { code, .. }) => return Err(failure_for(code, request)),
203                Some(ServiceMessage::Welcome { .. }) => {}
204                None => break,
205            }
206        }
207        // The connection broke before an answer: the service died or was replaced. Send the same request again.
208        breaks += 1;
209        if breaks > RECONNECTS {
210            return Err(ClientFailure::MayNotHaveLanded);
211        }
212        std::thread::sleep(CONNECT_PAUSE * u32::try_from(breaks).unwrap_or(1));
213    }
214}
215
216/// What:
217///  Tells the service that work items were written for a repository, starting it through the operating system when it
218///  is not running, and returns once the service has taken the message.
219/// Why:
220///   The wrapper never starts the service or runs the work itself; the message is the only thing the service needs to
221///  look at the repository's work directory. The connection closes when the service has read the line, which is the
222///  confirmation that it will act.
223///
224/// In TS you'd write (pseudocode):
225/// ```ts
226/// await wake(config, commonDir)
227/// ```
228pub fn wake(config: &ClientConfig, common_dir: &std::path::Path, notices: &mut dyn Write) -> Result<(), ClientFailure> {
229    let mut line: String = encode_client(&ClientMessage::Work { common_dir: common_dir.to_path_buf() });
230    line.push('\n');
231    let mut announced: bool = false;
232    let mut replacements: usize = 0;
233    loop {
234        let (mut reader, version) = reach(config, notices, &mut announced)?;
235        if version != PROTOCOL_VERSION {
236            if version > PROTOCOL_VERSION || replacements >= REPLACEMENTS {
237                return Err(ClientFailure::NoCommonVersion);
238            }
239            replacements += 1;
240            let mut retire: String = encode_client(&ClientMessage::Retire);
241            retire.push('\n');
242            let _sent = reader.get_mut().write_all(retire.as_bytes());
243            std::thread::sleep(CONNECT_PAUSE * 4);
244            continue;
245        }
246        if reader.get_mut().write_all(line.as_bytes()).is_err() {
247            return Err(ClientFailure::Unavailable);
248        }
249        // The service closes the connection after it has read the line.
250        while read_message(&mut reader).is_some() {}
251        return Ok(());
252    }
253}
git_policy_cli/service_client.rs

git_policy_cli/
service_client.rs
