Skip to main content

max / makenotwork

5.4 KB · 173 lines History Blame Raw
1 //! Fetching 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 use std::time::Duration;
9
10 use chrono::Utc;
11 use ops_status::Payload;
12 use tokio::sync::mpsc;
13
14 use crate::config::Source;
15
16 /// A poll result, tagged with which source it came from.
17 pub(crate) struct Update {
18 pub index: usize,
19 pub at: chrono::DateTime<Utc>,
20 pub result: Result<Payload, String>,
21 }
22
23 /// How long a single request may take before it counts as unreachable.
24 ///
25 /// Bounded well under any sane poll interval: a request that outlives its own
26 /// interval would stack tasks up behind it, and a source that is merely slow
27 /// should read as a problem rather than silently delaying every later poll.
28 const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
29
30 /// Spawn one polling task per source. Returns the receiving end.
31 pub(crate) fn spawn_all(sources: &[Source]) -> mpsc::Receiver<Update> {
32 // Capacity comfortably exceeds one in-flight update per source, so a
33 // momentarily busy UI loop cannot make a poller block or drop a result.
34 let (tx, rx) = mpsc::channel(sources.len().max(1) * 4);
35 for (index, source) in sources.iter().enumerate() {
36 tokio::spawn(poll_forever(index, source.clone(), tx.clone()));
37 }
38 rx
39 }
40
41 async fn poll_forever(index: usize, source: Source, tx: mpsc::Sender<Update>) {
42 let client = match reqwest::Client::builder().timeout(REQUEST_TIMEOUT).build() {
43 Ok(client) => client,
44 Err(e) => {
45 let _ = tx
46 .send(Update {
47 index,
48 at: Utc::now(),
49 result: Err(format!("client: {e}")),
50 })
51 .await;
52 return;
53 }
54 };
55
56 let url = source.status_url();
57 let mut ticker = tokio::time::interval(source.poll_interval());
58 // A poll that overruns its interval must not cause a burst of catch-up
59 // requests at the next tick.
60 ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
61
62 loop {
63 ticker.tick().await;
64 let result = fetch(&client, &url, source.token().as_deref()).await;
65 // The token is read per request rather than captured once, so rotating
66 // it does not require restarting the viewer.
67 if tx
68 .send(Update {
69 index,
70 at: Utc::now(),
71 result,
72 })
73 .await
74 .is_err()
75 {
76 return; // UI is gone
77 }
78 }
79 }
80
81 async fn fetch(
82 client: &reqwest::Client,
83 url: &str,
84 token: Option<&str>,
85 ) -> Result<Payload, String> {
86 let mut request = client.get(url);
87 if let Some(token) = token {
88 request = request.bearer_auth(token);
89 }
90
91 let response = request.send().await.map_err(short_error)?;
92 let status = response.status();
93 if !status.is_success() {
94 // 401 is worth its own words: it is the most likely misconfiguration
95 // and the least self-evident from a bare status code.
96 return Err(match status.as_u16() {
97 401 | 403 => "unauthorized (check the source's token_env)".into(),
98 code => format!("HTTP {code}"),
99 });
100 }
101
102 let body = response.bytes().await.map_err(short_error)?;
103 serde_json::from_slice(&body).map_err(|e| format!("bad payload: {e}"))
104 }
105
106 /// reqwest errors stringify into a paragraph with a full URL chain. The rollup
107 /// has one column for this, so keep the part that says what went wrong.
108 // e is consumed into the short string.
109 #[allow(clippy::needless_pass_by_value)]
110 fn short_error(e: reqwest::Error) -> String {
111 if e.is_timeout() {
112 return "timed out".into();
113 }
114 if e.is_connect() {
115 return "connection refused".into();
116 }
117 if e.is_decode() {
118 return "bad response body".into();
119 }
120 let text = e.to_string();
121 text.split(':')
122 .next_back()
123 .unwrap_or(&text)
124 .trim()
125 .to_string()
126 }
127
128 #[cfg(test)]
129 mod tests {
130 use super::*;
131
132 #[test]
133 fn a_status_url_is_built_from_the_configured_base() {
134 let source: Source = toml::from_str(
135 r#"
136 name = "sando"
137 url = "http://fw13:8080"
138 "#,
139 )
140 .unwrap();
141 assert_eq!(source.status_url(), "http://fw13:8080/status.json");
142 }
143
144 #[tokio::test]
145 async fn an_unreachable_host_reports_a_short_reason() {
146 let client = reqwest::Client::builder()
147 .timeout(Duration::from_millis(300))
148 .build()
149 .unwrap();
150 // Port 1 on loopback: nothing listens, so this refuses fast.
151 let err = fetch(&client, "http://127.0.0.1:1/status.json", None)
152 .await
153 .unwrap_err();
154 assert!(
155 err.len() < 60,
156 "the rollup has one column for this: {err:?}"
157 );
158 assert!(
159 !err.contains("http://"),
160 "no URL chain in the summary: {err:?}"
161 );
162 }
163
164 #[tokio::test]
165 async fn a_body_that_is_not_a_payload_is_an_error_not_a_panic() {
166 let client = reqwest::Client::new();
167 let bad: Result<Payload, String> =
168 serde_json::from_slice(b"not json").map_err(|e| format!("bad payload: {e}"));
169 assert!(bad.unwrap_err().starts_with("bad payload"));
170 drop(client);
171 }
172 }
173