asched_core/routine/
ipc.rs

1use super::{
2    Capabilities, FireOutcome, Routine, RoutineError, RoutineErrorKind, RunRecord, PROTOCOL_VERSION,
3};
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6use std::io::{BufRead, BufReader, Write};
7use std::os::unix::net::UnixStream;
8use std::path::{Path, PathBuf};
9use std::time::Duration;
10
11const MAX_RESPONSE_FRAME_BYTES: usize = 64 * 1024 * 1024;
12
13#[derive(Debug, Clone, Serialize, Deserialize)]
14pub struct Request {
15    pub protocol: u32,
16    pub project: PathBuf,
17    pub action: Action,
18}
19
20impl Request {
21    pub fn new(project: PathBuf, action: Action) -> Self {
22        Self {
23            protocol: PROTOCOL_VERSION,
24            project,
25            action,
26        }
27    }
28}
29
30#[derive(Debug, Clone, Serialize, Deserialize)]
31#[serde(tag = "action", rename_all = "snake_case")]
32pub enum Action {
33    List,
34    Show {
35        name: String,
36    },
37    Add {
38        revision: u64,
39        routine: Routine,
40    },
41    Edit {
42        revision: u64,
43        old_name: String,
44        routine: Routine,
45    },
46    Delete {
47        revision: u64,
48        name: String,
49    },
50    SetEnabled {
51        revision: u64,
52        name: String,
53        enabled: bool,
54    },
55    Run {
56        name: String,
57    },
58    Fire {
59        kind: String,
60        payload: Value,
61        event_id: String,
62    },
63    Cancel {
64        name: String,
65    },
66    Logs {
67        name: String,
68    },
69    Status,
70    Shutdown,
71}
72
73#[derive(Debug, Clone, Serialize, Deserialize)]
74pub struct RoutineView {
75    pub routine: Routine,
76    pub capabilities: Capabilities,
77    #[serde(default)]
78    pub next_run_epoch: Option<i64>,
79    pub latest_run: Option<RunRecord>,
80    #[serde(default)]
81    pub recent_runs: Vec<RunRecord>,
82}
83
84#[derive(Debug, Clone, Serialize, Deserialize)]
85#[serde(tag = "result", rename_all = "snake_case")]
86pub enum Response {
87    Routines {
88        revision: u64,
89        routines: Vec<RoutineView>,
90    },
91    Routine {
92        revision: u64,
93        routine: Box<RoutineView>,
94    },
95    Runs {
96        runs: Vec<RunRecord>,
97    },
98    Fire {
99        outcome: FireOutcome,
100    },
101    Daemon {
102        protocol: u32,
103        pid: u32,
104    },
105    Ok {
106        revision: Option<u64>,
107    },
108    Error {
109        kind: RoutineErrorKind,
110        message: String,
111    },
112}
113
114impl Response {
115    pub fn error(error: RoutineError) -> Self {
116        let kind = error.kind();
117        Self::Error {
118            kind,
119            message: error.to_string(),
120        }
121    }
122
123    pub fn into_result(self) -> Result<Self, RoutineError> {
124        match self {
125            Self::Error { kind, message } => Err(RoutineError::RemoteDaemon { kind, message }),
126            other => Ok(other),
127        }
128    }
129}
130
131pub fn send(socket: &Path, request: &Request) -> Result<Response, RoutineError> {
132    send_inner(socket, request, Duration::from_secs(30), false)
133}
134
135pub(crate) fn send_with_timeout(
136    socket: &Path,
137    request: &Request,
138    timeout: Duration,
139) -> Result<Response, RoutineError> {
140    send_inner(socket, request, timeout, true)
141}
142
143fn send_inner(
144    socket: &Path,
145    request: &Request,
146    timeout: Duration,
147    timeout_is_unavailable: bool,
148) -> Result<Response, RoutineError> {
149    let mut stream = UnixStream::connect(socket)
150        .map_err(|e| RoutineError::Unavailable(format!("{}: {e}", socket.display())))?;
151    stream.set_read_timeout(Some(timeout))?;
152    let mut data = serde_json::to_vec(request).map_err(|e| RoutineError::Corrupt(e.to_string()))?;
153    data.push(b'\n');
154    stream.write_all(&data)?;
155    stream.shutdown(std::net::Shutdown::Write)?;
156    let frame =
157        read_response_frame(
158            BufReader::new(stream),
159            MAX_RESPONSE_FRAME_BYTES,
160            |error| match error.kind() {
161                std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock
162                    if timeout_is_unavailable =>
163                {
164                    RoutineError::Unavailable("daemon response timed out".into())
165                }
166                _ => error.into(),
167            },
168        )?;
169    serde_json::from_slice(&frame)
170        .map_err(|e| RoutineError::Corrupt(format!("invalid daemon response: {e}")))
171}
172
173fn read_response_frame(
174    reader: impl BufRead,
175    max_bytes: usize,
176    map_io: impl FnOnce(std::io::Error) -> RoutineError,
177) -> Result<Vec<u8>, RoutineError> {
178    let mut frame = Vec::new();
179    reader
180        .take((max_bytes + 2) as u64)
181        .read_until(b'\n', &mut frame)
182        .map_err(map_io)?;
183    if frame.is_empty() {
184        return Err(RoutineError::Unavailable("daemon closed connection".into()));
185    }
186    if !frame.ends_with(b"\n") {
187        return Err(RoutineError::Corrupt(
188            "daemon response frame must end with a newline".into(),
189        ));
190    }
191    if frame.len() > max_bytes + 1 {
192        return Err(RoutineError::Corrupt(format!(
193            "daemon response frame exceeds {max_bytes} bytes"
194        )));
195    }
196    frame.pop();
197    Ok(frame)
198}
199
200#[cfg(test)]
201mod tests {
202    use super::*;
203
204    #[test]
205    fn protocol_v1_error_kinds_keep_the_existing_wire_strings() {
206        let cases = [
207            (RoutineErrorKind::Validation, "validation"),
208            (RoutineErrorKind::Duplicate, "duplicate"),
209            (RoutineErrorKind::NotFound, "not_found"),
210            (RoutineErrorKind::Conflict, "conflict"),
211            (RoutineErrorKind::ProjectCollision, "project_collision"),
212            (RoutineErrorKind::AlreadyRunning, "already_running"),
213            (RoutineErrorKind::ProtocolMismatch, "protocol_mismatch"),
214            (RoutineErrorKind::Unavailable, "unavailable"),
215            (RoutineErrorKind::Io, "io"),
216            (RoutineErrorKind::Corrupt, "corrupt"),
217        ];
218        for (kind, wire) in cases {
219            let response = Response::Error {
220                kind,
221                message: "detail".into(),
222            };
223            let json = serde_json::to_string(&response).unwrap();
224            assert!(json.contains(&format!(r#""kind":"{wire}""#)));
225            let decoded: Response = serde_json::from_str(&json).unwrap();
226            assert!(matches!(decoded, Response::Error { kind: decoded, .. } if decoded == kind));
227        }
228    }
229
230    #[test]
231    fn daemon_error_category_and_message_cross_the_client_boundary() {
232        let response: Response = serde_json::from_str(
233            r#"{"result":"error","kind":"conflict","message":"stale revision"}"#,
234        )
235        .unwrap();
236        assert!(matches!(
237            response.into_result(),
238            Err(RoutineError::RemoteDaemon {
239                kind: RoutineErrorKind::Conflict,
240                message,
241            }) if message == "stale revision"
242        ));
243    }
244
245    #[test]
246    fn unknown_error_kind_is_rejected_as_an_invalid_closed_domain() {
247        let result = serde_json::from_str::<Response>(
248            r#"{"result":"error","kind":"future_kind","message":"detail"}"#,
249        );
250        assert!(result.is_err());
251    }
252}
253
254#[cfg(test)]
255#[path = "ipc_contract_tests.rs"]
256mod ipc_contract_tests;
asched_core/routine/ipc.rs

asched_core/routine/
ipc.rs
