stow-cli 0.5.0

CLI that routes rustc invocations through the stow public artifact cache
Documentation 
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

//! The facade side: one connection, one question, one answer.

use std::ffi::OsString;

use tokio::io::{AsyncRead, AsyncWrite};

use super::Endpoint;
use super::protocol::{Answer, Compiled, Plan, Request, read_frame, write_frame};

/// What the supervisor decided for this invocation.
#[derive(Debug)]
pub enum Decision {
    /// The outputs are already in place; exit 0 without running rustc.
    Served,
    /// Run the real rustc, then call [`report`] with its outcome.
    Compile(Ticket),
}

/// The supervisor's handle on a compile it asked the facade to run.
#[derive(Debug)]
pub struct Ticket {
    ticket: u64,
}

/// Connection to the supervisor, kept open across the plan and the report
/// so the compile's outcome reaches the same conversation.
pub struct Connection {
    stream: Stream,
    token: String,
}

enum Stream {
    #[cfg(unix)]
    Unix(tokio::net::UnixStream),
    Loopback(tokio::net::TcpStream),
}

impl Connection {
    /// Connect to `endpoint`.
    ///
    /// # Errors
    ///
    /// Any connection failure. The caller fails the build: an endpoint in
    /// the environment that cannot be reached means the supervisor died,
    /// and a build that quietly stops using the cache is the failure that
    /// hides.
    pub async fn open(endpoint: &Endpoint, token: String) -> Result<Self, String> {
        let stream = match endpoint {
            #[cfg(unix)]
            Endpoint::Unix(path) => tokio::net::UnixStream::connect(path)
                .await
                .map(Stream::Unix)
                .map_err(|error| format!("connect to supervisor at {}: {error}", path.display()))?,
            Endpoint::Loopback(port) => tokio::net::TcpStream::connect(("127.0.0.1", *port))
                .await
                .map(Stream::Loopback)
                .map_err(|error| format!("connect to supervisor on port {port}: {error}"))?,
        };
        Ok(Self { stream, token })
    }

    /// Ask what to do with one rustc invocation.
    ///
    /// # Errors
    ///
    /// Transport failures, and a supervisor that answers with a failure of
    /// its own.
    pub async fn plan(
        &mut self,
        executable: &std::ffi::OsStr,
        args: &[OsString],
    ) -> Result<Decision, String> {
        let plan = Plan::new(self.token.clone(), executable, args);
        match self.exchange(&Request::Plan(plan)).await? {
            Answer::Served => Ok(Decision::Served),
            Answer::Compile { ticket } => Ok(Decision::Compile(Ticket { ticket })),
            Answer::Recorded => Err("supervisor answered a plan with a report ack".to_owned()),
            Answer::Failed { message } => Err(message),
        }
    }

    /// Report the compile the supervisor asked for.
    ///
    /// # Errors
    ///
    /// Transport failures, and a supervisor that answers with a failure of
    /// its own.
    pub async fn report(&mut self, ticket: &Ticket, success: bool) -> Result<(), String> {
        let report = Compiled {
            token: self.token.clone(),
            ticket: ticket.ticket,
            success,
        };
        match self.exchange(&Request::Compiled(report)).await? {
            Answer::Recorded => Ok(()),
            Answer::Failed { message } => Err(message),
            Answer::Served | Answer::Compile { .. } => {
                Err("supervisor answered a report with a plan answer".to_owned())
            }
        }
    }

    async fn exchange(&mut self, request: &Request) -> Result<Answer, String> {
        match &mut self.stream {
            #[cfg(unix)]
            Stream::Unix(stream) => exchange_on(stream, request).await,
            Stream::Loopback(stream) => exchange_on(stream, request).await,
        }
    }
}

async fn exchange_on<S>(stream: &mut S, request: &Request) -> Result<Answer, String>
where
    S: AsyncRead + AsyncWrite + Unpin + Send,
{
    write_frame(stream, request).await?;
    read_frame(stream)
        .await?
        .ok_or_else(|| "supervisor closed the connection without answering".to_owned())
}