Skip to main content

max / makenotwork

8.9 KB · 250 lines History Blame Raw
1 //! Reading each source on its own interval.
2 //!
3 //! One task per source, each reporting into a channel the UI drains. Sources
4 //! are independent on purpose: a daemon that hangs must not stop the others
5 //! from updating, because the tab that stops updating is exactly the one you
6 //! need to see.
7 //!
8 //! Two origins, one loop. An HTTP source is fetched and a file source is read,
9 //! and past that nothing downstream can tell them apart: both produce a
10 //! `Payload` or a short reason, on the same tick, into the same channel. A batch
11 //! producer is legible here because the UI already keeps the last payload across
12 //! a failed poll, so a sweep that finished at 03:20 still reads at 15:00.
13
14 use std::path::Path;
15 use std::time::Duration;
16
17 use chrono::Utc;
18 use ops_status::Payload;
19 use tokio::sync::mpsc;
20
21 use crate::config::{Origin, Source};
22
23 /// A poll result, tagged with which source it came from.
24 pub(crate) struct Update {
25 pub index: usize,
26 pub at: chrono::DateTime<Utc>,
27 pub result: Result<Payload, String>,
28 }
29
30 /// How long a single request may take before it counts as unreachable.
31 ///
32 /// Bounded well under any sane poll interval: a request that outlives its own
33 /// interval would stack tasks up behind it, and a source that is merely slow
34 /// should read as a problem rather than silently delaying every later poll.
35 const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
36
37 /// Spawn one polling task per source. Returns the receiving end.
38 pub(crate) fn spawn_all(sources: &[Source]) -> mpsc::Receiver<Update> {
39 // Capacity comfortably exceeds one in-flight update per source, so a
40 // momentarily busy UI loop cannot make a poller block or drop a result.
41 let (tx, rx) = mpsc::channel(sources.len().max(1) * 4);
42 for (index, source) in sources.iter().enumerate() {
43 tokio::spawn(poll_forever(index, source.clone(), tx.clone()));
44 }
45 rx
46 }
47
48 async fn poll_forever(index: usize, source: Source, tx: mpsc::Sender<Update>) {
49 let origin = source.origin();
50 // Built once per task, and only for an HTTP source: a file source has no
51 // transport to fail, so a broken TLS setup must not take its tab down with
52 // the others.
53 let client = if matches!(origin, Origin::Http(_)) {
54 match crate::tls::client(REQUEST_TIMEOUT) {
55 Ok(client) => Some(client),
56 Err(e) => {
57 let _ = tx
58 .send(Update {
59 index,
60 at: Utc::now(),
61 result: Err(format!("client: {e}")),
62 })
63 .await;
64 return;
65 }
66 }
67 } else {
68 None
69 };
70
71 let mut ticker = tokio::time::interval(source.poll_interval());
72 // A poll that overruns its interval must not cause a burst of catch-up
73 // requests at the next tick.
74 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
75
76 loop {
77 ticker.tick().await;
78 let result = match &origin {
79 // The token is read per request rather than captured once, so
80 // rotating it does not require restarting magicmirror.
81 Origin::Http(url) => match &client {
82 Some(client) => fetch(client, url, source.token().as_deref()).await,
83 None => Err("client: not built".into()),
84 },
85 Origin::File(path) => read(path).await,
86 };
87 if tx
88 .send(Update {
89 index,
90 at: Utc::now(),
91 result,
92 })
93 .await
94 .is_err()
95 {
96 return; // UI is gone
97 }
98 }
99 }
100
101 async fn fetch(
102 client: &reqwest::Client,
103 url: &str,
104 token: Option<&str>,
105 ) -> Result<Payload, String> {
106 let mut request = client.get(url);
107 if let Some(token) = token {
108 request = request.bearer_auth(token);
109 }
110
111 let response = request.send().await.map_err(short_error)?;
112 let status = response.status();
113 if !status.is_success() {
114 // 401 is worth its own words: it is the most likely misconfiguration
115 // and the least self-evident from a bare status code.
116 return Err(match status.as_u16() {
117 401 | 403 => "unauthorized (check the source's token_env)".into(),
118 code => format!("HTTP {code}"),
119 });
120 }
121
122 let body = response.bytes().await.map_err(short_error)?;
123 serde_json::from_slice(&body).map_err(|e| format!("bad payload: {e}"))
124 }
125
126 /// Read a payload a batch producer left on disk.
127 ///
128 /// Every way this fails is the same condition an unreachable daemon is in --
129 /// nothing current to show -- so it reports the same shape, and the UI keeps the
130 /// last good payload either way. A missing file is the ordinary case on a fresh
131 /// machine: the producer has not run yet.
132 async fn read(path: &Path) -> Result<Payload, String> {
133 let bytes = tokio::fs::read(path).await.map_err(short_io_error)?;
134 serde_json::from_slice(&bytes).map_err(|e| format!("bad payload: {e}"))
135 }
136
137 /// The rollup has one column for this, and an io error stringifies with the
138 /// whole path in it, which is already the tab's own configuration.
139 #[allow(clippy::needless_pass_by_value)]
140 fn short_io_error(e: std::io::Error) -> String {
141 match e.kind() {
142 std::io::ErrorKind::NotFound => "no payload written yet".into(),
143 std::io::ErrorKind::PermissionDenied => "permission denied".into(),
144 std::io::ErrorKind::IsADirectory => "path is a directory".into(),
145 kind => kind.to_string(),
146 }
147 }
148
149 /// reqwest errors stringify into a paragraph with a full URL chain. The rollup
150 /// has one column for this, so keep the part that says what went wrong.
151 // e is consumed into the short string.
152 #[allow(clippy::needless_pass_by_value)]
153 fn short_error(e: reqwest::Error) -> String {
154 if e.is_timeout() {
155 return "timed out".into();
156 }
157 if e.is_connect() {
158 return "connection refused".into();
159 }
160 if e.is_decode() {
161 return "bad response body".into();
162 }
163 let text = e.to_string();
164 text.split(':')
165 .next_back()
166 .unwrap_or(&text)
167 .trim()
168 .to_string()
169 }
170
171 #[cfg(test)]
172 mod tests {
173 use super::*;
174
175 #[test]
176 fn a_status_url_is_built_from_the_configured_base() {
177 let source: Source = toml::from_str(
178 r#"
179 name = "sando"
180 url = "http://fw13:8080"
181 "#,
182 )
183 .unwrap();
184 assert_eq!(
185 source.origin(),
186 Origin::Http("http://fw13:8080/status.json".into())
187 );
188 }
189
190 /// The batch producer's ordinary state on a fresh machine: configured,
191 /// pointed at the right place, and nothing has run yet. It has to read as a
192 /// source with nothing current rather than as a broken config.
193 #[tokio::test]
194 async fn a_payload_that_has_not_been_written_yet_is_a_short_reason() {
195 let dir = tempfile::tempdir().unwrap();
196 let err = read(&dir.path().join("status.json")).await.unwrap_err();
197 assert_eq!(err, "no payload written yet");
198 }
199
200 #[tokio::test]
201 async fn a_payload_on_disk_reads_the_same_as_one_off_the_wire() {
202 let dir = tempfile::tempdir().unwrap();
203 let path = dir.path().join("status.json");
204 let written = Payload::new("witchbroom", Utc::now());
205 std::fs::write(&path, serde_json::to_string(&written).unwrap()).unwrap();
206
207 let got = read(&path).await.unwrap();
208 assert_eq!(got.source, "witchbroom");
209 assert_eq!(got, written);
210 }
211
212 #[tokio::test]
213 async fn a_file_that_is_not_a_payload_says_so_rather_than_panicking() {
214 let dir = tempfile::tempdir().unwrap();
215 let path = dir.path().join("status.json");
216 std::fs::write(&path, b"half a write").unwrap();
217 assert!(read(&path).await.unwrap_err().starts_with("bad payload"));
218 }
219
220 // Opens a socket, which miri has no shim for. See the note in `exec`.
221 #[tokio::test]
222 #[cfg_attr(miri, ignore)]
223 async fn an_unreachable_host_reports_a_short_reason() {
224 let client = crate::tls::client(Duration::from_millis(300)).unwrap();
225 // Port 1 on loopback: nothing listens, so this refuses fast.
226 let err = fetch(&client, "http://127.0.0.1:1/status.json", None)
227 .await
228 .unwrap_err();
229 assert!(
230 err.len() < 60,
231 "the rollup has one column for this: {err:?}"
232 );
233 assert!(
234 !err.contains("http://"),
235 "no URL chain in the summary: {err:?}"
236 );
237 }
238
239 // Builds a reqwest client, which reaches for the same socket machinery.
240 #[tokio::test]
241 #[cfg_attr(miri, ignore)]
242 async fn a_body_that_is_not_a_payload_is_an_error_not_a_panic() {
243 let client = crate::tls::client(Duration::from_millis(300)).unwrap();
244 let bad: Result<Payload, String> =
245 serde_json::from_slice(b"not json").map_err(|e| format!("bad payload: {e}"));
246 assert!(bad.unwrap_err().starts_with("bad payload"));
247 drop(client);
248 }
249 }
250