runner_manager_platform/
service_request.rs

1//! Asking the running service to do something this process cannot.
2//!
3//! The service shares its state directory with the account that installed
4//! it, so a request left there is a channel both sides can use with no
5//! privilege and no socket. The asking command leaves a request under an
6//! identifier it chose; the service takes the request by deleting it, which is
7//! how the command knows somebody is listening; and the service answers under
8//! the same identifier, which is how the command knows the answer is to this
9//! request.
10//!
11//! [`GITHUB_DISCOVERY`] gives every request its own pair of files under
12//! `service-requests/`, so commands asking at once do not displace each other.
13//! [`CACHE_PRUNE`] keeps the single pair of files it was introduced with, which
14//! a service from that release still reads.
15
16use std::fs;
17use std::io;
18use std::path::{Path, PathBuf};
19use std::time::{Duration, Instant};
20
21use serde::de::DeserializeOwned;
22use serde::{Deserialize, Serialize};
23
24/// How often the service looks for a request.
25pub const POLL: Duration = Duration::from_secs(2);
26
27/// The directory under the state directory per-request channels use.
28const REQUESTS_DIR: &str = "service-requests";
29
30/// One kind of request and its answers.
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32pub struct Channel {
33    slots: Slots,
34}
35
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37enum Slots {
38    /// One request file and one result file, shared by every asker.
39    One {
40        request: &'static str,
41        result: &'static str,
42    },
43    /// `<prefix>.<id>.request` and `<prefix>.<id>.result` per request.
44    PerRequest(&'static str),
45}
46
47/// `host cache prune` for a cache root this account cannot write: the root of
48/// a service running as another account (LocalSystem on Windows, root
49/// elsewhere).
50pub const CACHE_PRUNE: Channel = Channel {
51    slots: Slots::One {
52        request: crate::dependency_cache::PRUNE_REQUEST_FILE,
53        result: crate::dependency_cache::PRUNE_RESULT_FILE,
54    },
55};
56
57/// What the stored GitHub credential reaches, asked of the service by a
58/// command that cannot read the credential itself.
59///
60/// The reason is a login-mode Mac reached over SSH: the credential is in the
61/// login keychain, which an SSH session cannot unlock, while the service runs
62/// in the desktop session and reads it every poll. `repo add`, `org add` and
63/// `auth status` used to fail there with the keychain's error. The answer
64/// holds what GitHub said about the installations, never the credential.
65pub const GITHUB_DISCOVERY: Channel = Channel {
66    slots: Slots::PerRequest("github-discovery"),
67};
68
69/// How long a command waits on the service, and how often it looks.
70#[derive(Debug, Clone, Copy, PartialEq, Eq)]
71pub struct Wait {
72    /// For the service to take the request: several of its polls, so a
73    /// running service always answers inside it.
74    pub taken: Duration,
75    /// For the answer, once the service has taken the request.
76    pub finished: Duration,
77    pub poll: Duration,
78}
79
80impl Wait {
81    /// What a command waiting on a GitHub request uses.
82    pub const GITHUB: Self = Self {
83        taken: Duration::from_secs(4 * POLL.as_secs()),
84        finished: Duration::from_secs(90),
85        poll: Duration::from_millis(250),
86    };
87}
88
89/// Why a command got no answer.
90#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
91pub enum AskError {
92    #[error("the request could not be left for the service: {0}")]
93    NotSent(String),
94    #[error("no running service took the request within {0} seconds")]
95    NotTaken(u64),
96    #[error("the service took the request but gave no answer within {0} seconds")]
97    NoAnswer(u64),
98    #[error("the service could not answer: {0}")]
99    Refused(String),
100}
101
102#[derive(Debug, Serialize, Deserialize)]
103struct Answer<T> {
104    request: String,
105    outcome: Result<T, String>,
106}
107
108/// The longest request identifier taken, and the characters it may hold. An
109/// identifier is chosen by the asking command (a timestamp and a process id)
110/// and copied into the answer, so the service takes nothing else.
111const MAX_ID: usize = 64;
112
113fn valid_id(id: &str) -> bool {
114    !id.is_empty()
115        && id.len() <= MAX_ID
116        && id
117            .chars()
118            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '.' | '_'))
119}
120
121/// An answer nobody collected is removed after this long.
122const UNCOLLECTED: Duration = Duration::from_secs(10 * 60);
123
124/// Reads a single-slot request without following a link and without reading
125/// more than an identifier's worth.
126///
127/// The service may run as root or LocalSystem while the state directory
128/// belongs to the account that installed it, so the request path is under that
129/// account's control. Following a link there would have the service read a
130/// file of the link's choosing, and copy it into an answer that account can
131/// read; a FIFO or `/dev/zero` would hold the service's loop.
132fn read_request(path: &Path) -> Option<String> {
133    use std::io::Read as _;
134    let mut options = fs::OpenOptions::new();
135    options.read(true);
136    #[cfg(unix)]
137    std::os::unix::fs::OpenOptionsExt::custom_flags(
138        &mut options,
139        libc::O_NOFOLLOW | libc::O_NONBLOCK,
140    );
141    #[cfg(windows)]
142    std::os::windows::fs::OpenOptionsExt::custom_flags(
143        &mut options,
144        // FILE_FLAG_OPEN_REPARSE_POINT: open a link as itself.
145        0x0020_0000,
146    );
147    let file = options.open(path).ok()?;
148    if !file.metadata().ok()?.is_file() {
149        return None;
150    }
151    let mut text = String::new();
152    file.take(MAX_ID as u64 + 2)
153        .read_to_string(&mut text)
154        .ok()?;
155    let id = text.trim();
156    valid_id(id).then(|| id.to_owned())
157}
158
159/// The answer file is the service's to write and the asking account's to read,
160/// whichever account the service runs as. It holds no secret.
161fn write_readable_by_the_asker(path: &Path, bytes: &[u8]) -> io::Result<()> {
162    use std::io::Write as _;
163    let parent = path.parent().unwrap_or_else(|| Path::new("."));
164    fs::create_dir_all(parent)?;
165    let mut temporary = tempfile::NamedTempFile::new_in(parent)?;
166    temporary.write_all(bytes)?;
167    temporary.as_file().sync_all()?;
168    // On the file before it appears under its name: the asker removes an
169    // answer as soon as it has read it, so a mode set afterwards can find
170    // nothing there.
171    #[cfg(unix)]
172    {
173        use std::os::unix::fs::PermissionsExt as _;
174        temporary
175            .as_file()
176            .set_permissions(fs::Permissions::from_mode(0o644))?;
177    }
178    temporary
179        .persist(path)
180        .map(|_| ())
181        .map_err(|error| error.error)
182}
183
184impl Channel {
185    /// Where a single-slot channel's request lives.
186    #[must_use]
187    pub fn request_path(&self, state_dir: &Path) -> PathBuf {
188        match self.slots {
189            Slots::One { request, .. } => state_dir.join(request),
190            Slots::PerRequest(prefix) => state_dir.join(REQUESTS_DIR).join(prefix),
191        }
192    }
193
194    fn request_file(&self, state_dir: &Path, id: &str) -> PathBuf {
195        match self.slots {
196            Slots::One { request, .. } => state_dir.join(request),
197            Slots::PerRequest(prefix) => state_dir
198                .join(REQUESTS_DIR)
199                .join(format!("{prefix}.{id}.request")),
200        }
201    }
202
203    fn result_file(&self, state_dir: &Path, id: &str) -> PathBuf {
204        match self.slots {
205            Slots::One { result, .. } => state_dir.join(result),
206            Slots::PerRequest(prefix) => state_dir
207                .join(REQUESTS_DIR)
208                .join(format!("{prefix}.{id}.result")),
209        }
210    }
211
212    /// Takes the pending requests and returns their identifiers. Taking one is
213    /// deleting it, so a request is answered once.
214    #[must_use]
215    pub fn take(&self, state_dir: &Path) -> Vec<String> {
216        match self.slots {
217            Slots::One { request, .. } => {
218                let path = state_dir.join(request);
219                if fs::symlink_metadata(&path).is_err() {
220                    return Vec::new();
221                }
222                let id = read_request(&path);
223                // Removed whatever it held: an unreadable or malformed request
224                // is not going to become a good one.
225                if fs::remove_file(&path).is_err() {
226                    return Vec::new();
227                }
228                id.into_iter().collect()
229            }
230            Slots::PerRequest(prefix) => {
231                let Ok(entries) = fs::read_dir(state_dir.join(REQUESTS_DIR)) else {
232                    return Vec::new();
233                };
234                let mut taken = Vec::new();
235                for entry in entries.flatten() {
236                    let name = entry.file_name();
237                    let Some(name) = name.to_str() else { continue };
238                    let Some(rest) = name.strip_prefix(prefix).and_then(|r| r.strip_prefix('.'))
239                    else {
240                        continue;
241                    };
242                    if let Some(id) = rest.strip_suffix(".request") {
243                        // The identifier is the file's name, so its contents
244                        // are never read, and a link is removed as a link.
245                        if fs::remove_file(entry.path()).is_ok() && valid_id(id) {
246                            taken.push(id.to_owned());
247                        }
248                    } else if rest.ends_with(".result")
249                        && entry
250                            .metadata()
251                            .and_then(|meta| meta.modified())
252                            .ok()
253                            .and_then(|at| at.elapsed().ok())
254                            .is_some_and(|age| age >= UNCOLLECTED)
255                    {
256                        let _ = fs::remove_file(entry.path());
257                    }
258                }
259                taken
260            }
261        }
262    }
263
264    /// The service's answer to `request`.
265    ///
266    /// # Errors
267    /// The answer could not be encoded or written.
268    pub fn answer<T: Serialize>(
269        &self,
270        state_dir: &Path,
271        request: &str,
272        outcome: Result<T, String>,
273    ) -> io::Result<()> {
274        let answer = Answer {
275            request: request.to_owned(),
276            outcome,
277        };
278        let json = serde_json::to_vec_pretty(&answer).map_err(io::Error::other)?;
279        let path = self.result_file(state_dir, request);
280        write_readable_by_the_asker(&path, &json)
281    }
282
283    /// The service's answer to `request`, once it has given one.
284    #[must_use]
285    pub fn read_answer<T: DeserializeOwned>(
286        &self,
287        state_dir: &Path,
288        request: &str,
289    ) -> Option<Result<T, String>> {
290        fs::read(self.result_file(state_dir, request))
291            .ok()
292            .and_then(|bytes| serde_json::from_slice::<Answer<T>>(&bytes).ok())
293            .filter(|answer| answer.request == request)
294            .map(|answer| answer.outcome)
295    }
296
297    /// Asks the service and waits for its answer.
298    ///
299    /// # Errors
300    /// [`AskError`], naming which half of the exchange did not happen.
301    pub fn ask<T: DeserializeOwned>(&self, state_dir: &Path, wait: Wait) -> Result<T, AskError> {
302        self.ask_with(state_dir, wait, || {})
303    }
304
305    /// [`Self::ask`], calling `sent` once the request is in place, so the
306    /// command can say it is waiting.
307    ///
308    /// # Errors
309    /// As [`Self::ask`].
310    pub fn ask_with<T: DeserializeOwned>(
311        &self,
312        state_dir: &Path,
313        wait: Wait,
314        sent: impl FnOnce(),
315    ) -> Result<T, AskError> {
316        // Unique within this process as well as across processes: two threads
317        // can read the same instant where the clock is coarse (macOS).
318        static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
319        let id = format!(
320            "{}-{}-{}",
321            chrono::Utc::now().format("%Y%m%dT%H%M%S%.9fZ"),
322            std::process::id(),
323            NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed)
324        );
325        let request = self.request_file(state_dir, &id);
326        crate::host_fitness::write_atomically(&request, id.as_bytes())
327            .map_err(|error| AskError::NotSent(format!("{}: {error}", request.display())))?;
328        sent();
329
330        let started = Instant::now();
331        let pending = |request: &Path| match self.slots {
332            Slots::One { .. } => read_request(request).as_deref() == Some(id.as_str()),
333            Slots::PerRequest(_) => fs::symlink_metadata(request).is_ok(),
334        };
335        while pending(&request) {
336            if started.elapsed() >= wait.taken {
337                // Withdrawn, so a service that starts later does not answer a
338                // command that is no longer waiting: only this command's own
339                // request, which in a single slot another command may have
340                // replaced. Gone already means it was taken after all.
341                match fs::remove_file(&request) {
342                    Err(error) if error.kind() == io::ErrorKind::NotFound => break,
343                    _ => return Err(AskError::NotTaken(wait.taken.as_secs())),
344                }
345            }
346            std::thread::sleep(wait.poll);
347        }
348
349        let taken = Instant::now();
350        loop {
351            if let Some(outcome) = self.read_answer(state_dir, &id) {
352                if matches!(self.slots, Slots::PerRequest(_)) {
353                    let _ = fs::remove_file(self.result_file(state_dir, &id));
354                }
355                return outcome.map_err(AskError::Refused);
356            }
357            if taken.elapsed() >= wait.finished {
358                return Err(AskError::NoAnswer(wait.finished.as_secs()));
359            }
360            std::thread::sleep(wait.poll);
361        }
362    }
363}
364
365#[cfg(test)]
366mod tests {
367    use super::*;
368
369    const QUICK: Wait = Wait {
370        taken: Duration::from_millis(600),
371        finished: Duration::from_secs(5),
372        poll: Duration::from_millis(20),
373    };
374
375    /// A stand-in service: takes one request and answers it with `outcome`.
376    fn service(
377        state_dir: PathBuf,
378        outcome: Result<u32, String>,
379    ) -> std::thread::JoinHandle<String> {
380        std::thread::spawn(move || {
381            loop {
382                if let Some(request) = GITHUB_DISCOVERY.take(&state_dir).pop() {
383                    GITHUB_DISCOVERY
384                        .answer(&state_dir, &request, outcome)
385                        .unwrap();
386                    return request;
387                }
388                std::thread::sleep(Duration::from_millis(10));
389            }
390        })
391    }
392
393    #[test]
394    fn a_running_service_takes_the_request_once_and_answers_it() {
395        let dir = tempfile::tempdir().unwrap();
396        let answering = service(dir.path().to_path_buf(), Ok(7));
397        assert_eq!(GITHUB_DISCOVERY.ask::<u32>(dir.path(), QUICK), Ok(7));
398        answering.join().unwrap();
399        assert!(
400            GITHUB_DISCOVERY.take(dir.path()).is_empty(),
401            "answered once"
402        );
403    }
404
405    #[test]
406    fn a_service_that_cannot_answer_says_why() {
407        let dir = tempfile::tempdir().unwrap();
408        let answering = service(dir.path().to_path_buf(), Err("no credential".into()));
409        assert_eq!(
410            GITHUB_DISCOVERY.ask::<u32>(dir.path(), QUICK),
411            Err(AskError::Refused("no credential".into()))
412        );
413        answering.join().unwrap();
414    }
415
416    #[test]
417    fn with_no_service_listening_the_request_is_withdrawn() {
418        let dir = tempfile::tempdir().unwrap();
419        assert!(matches!(
420            GITHUB_DISCOVERY.ask::<u32>(dir.path(), QUICK),
421            Err(AskError::NotTaken(_))
422        ));
423        assert!(
424            GITHUB_DISCOVERY.take(dir.path()).is_empty(),
425            "a later service must not answer a command that stopped waiting"
426        );
427    }
428
429    #[test]
430    fn an_answer_to_another_request_is_not_taken_for_this_one() {
431        let dir = tempfile::tempdir().unwrap();
432        GITHUB_DISCOVERY
433            .answer(dir.path(), "an-earlier-request", Ok(1_u32))
434            .unwrap();
435        let state_dir = dir.path().to_path_buf();
436        // Takes the request and never answers it.
437        let silent = std::thread::spawn(move || {
438            while GITHUB_DISCOVERY.take(&state_dir).is_empty() {
439                std::thread::sleep(Duration::from_millis(10));
440            }
441        });
442        let wait = Wait {
443            finished: Duration::from_millis(300),
444            ..QUICK
445        };
446        assert!(matches!(
447            GITHUB_DISCOVERY.ask::<u32>(dir.path(), wait),
448            Err(AskError::NoAnswer(_))
449        ));
450        silent.join().unwrap();
451    }
452
453    /// Two commands asking at once each get their own answer: neither
454    /// replaces the other's request.
455    #[test]
456    fn commands_asking_at_once_each_get_their_own_answer() {
457        let dir = tempfile::tempdir().unwrap();
458        let state_dir = dir.path().to_path_buf();
459        let service = std::thread::spawn(move || {
460            let mut answered = 0;
461            let deadline = Instant::now() + Duration::from_secs(10);
462            while answered < 2 && Instant::now() < deadline {
463                for request in GITHUB_DISCOVERY.take(&state_dir) {
464                    GITHUB_DISCOVERY
465                        .answer(&state_dir, &request, Ok(request.len()))
466                        .unwrap();
467                    answered += 1;
468                }
469                std::thread::sleep(Duration::from_millis(10));
470            }
471            answered
472        });
473        let asking =
474            |dir: PathBuf| std::thread::spawn(move || GITHUB_DISCOVERY.ask::<usize>(&dir, QUICK));
475        let first = asking(dir.path().to_path_buf());
476        let second = asking(dir.path().to_path_buf());
477        assert!(first.join().unwrap().is_ok());
478        assert!(second.join().unwrap().is_ok());
479        assert_eq!(service.join().unwrap(), 2);
480    }
481
482    /// The service never follows a link left at a request path, and never
483    /// takes an identifier it would not have chosen.
484    #[cfg(unix)]
485    #[test]
486    fn a_request_that_is_a_link_or_not_an_identifier_is_dropped_unread() {
487        let dir = tempfile::tempdir().unwrap();
488        let secret = dir.path().join("only-the-service-may-read");
489        fs::write(&secret, "the contents of somebody else's file").unwrap();
490        let request = CACHE_PRUNE.request_path(dir.path());
491        std::os::unix::fs::symlink(&secret, &request).unwrap();
492        assert!(CACHE_PRUNE.take(dir.path()).is_empty());
493        assert!(
494            fs::symlink_metadata(&request).is_err(),
495            "the link is removed"
496        );
497        assert!(secret.exists(), "and only the link");
498
499        fs::write(&request, "not an identifier\nat all").unwrap();
500        assert!(CACHE_PRUNE.take(dir.path()).is_empty());
501        fs::write(&request, "20261007T221210.123456789Z-4242\n").unwrap();
502        assert_eq!(
503            CACHE_PRUNE.take(dir.path()),
504            ["20261007T221210.123456789Z-4242"]
505        );
506    }
507}
runner_manager_platform/service_request.rs

runner_manager_platform/
service_request.rs
