yah_qed/
waitfor.rs

1//! @arch:see(.yah/docs/working/W207-dashboard-e2e-in-qed.md)
2//!
3//! The `wait_for` health-gate primitive (W207 Gap #5). A [`StepKind::WaitFor`]
4//! step polls a single network endpoint until it answers, then advances — the
5//! gate that sits between a `background` sidecar (`yah-camp`, `vite preview`)
6//! and the step that talks to it, so the consumer never races a not-yet-bound
7//! port. Before this, the recipe hand-rolled
8//! `bash -c "until curl …/health; do sleep 1; done"`.
9//!
10//! [`StepKind::WaitFor`]: crate::types::StepKind::WaitFor
11//!
12//! Two probe shapes, both dependency-free (no HTTP client / TLS stack is pulled
13//! into qed — that matters for the musl-static build):
14//!
15//! - **http** — a plaintext HTTP/1.1 `GET` over a raw [`TcpStream`]. Healthy on
16//!   a 2xx/3xx status, or an exact match to `expect_status` when set. TLS is out
17//!   of scope for v1; an `https://` URL is rejected by the runner before we get
18//!   here.
19//! - **tcp** — a bare connect to `host:port`. Healthy the moment the port
20//!   accepts; no bytes are exchanged.
21//!
22//! The poll loop itself (deadline + interval + live `StepOutput` progress)
23//! lives in the runner, which holds the event sink; this module owns the pure
24//! URL parsing and the single-attempt probes so they unit-test without one.
25
26use std::time::Duration;
27use tokio::io::{AsyncReadExt, AsyncWriteExt};
28use tokio::net::TcpStream;
29use tokio::time::timeout;
30
31/// A parsed plaintext-HTTP target. v1 supports `http://` only — `https://` is
32/// rejected upstream (see [`crate::types::WaitForConfig::http_is_tls`]).
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct HttpTarget {
35    pub host: String,
36    pub port: u16,
37    /// Request target, always beginning with `/` (defaults to `/` when the URL
38    /// has no path).
39    pub path: String,
40}
41
42/// Split a plaintext-HTTP URL into `host` / `port` / `path`. Deliberately
43/// minimal — no query/fragment/userinfo handling beyond what a health-gate URL
44/// needs. Rejects an `https://` scheme (TLS is a v1 non-goal) and an empty host.
45pub fn parse_http_url(url: &str) -> Result<HttpTarget, String> {
46    let url = url.trim();
47    let rest = url
48        .strip_prefix("http://")
49        .ok_or_else(|| {
50            if url.starts_with("https://") {
51                format!("`{url}`: https health-gates are not supported in v1 (R513-F3) — use an http:// URL or a `tcp` target")
52            } else {
53                format!("`{url}`: wait-for `http` target must be an http:// URL")
54            }
55        })?;
56
57    // Split authority from path at the first '/'. The path keeps its leading
58    // slash; an absent path becomes "/".
59    let (authority, path) = match rest.find('/') {
60        Some(i) => (&rest[..i], rest[i..].to_string()),
61        None => (rest, "/".to_string()),
62    };
63    if authority.is_empty() {
64        return Err(format!("`{url}`: wait-for `http` target has an empty host"));
65    }
66
67    // Authority is host[:port]. Default port 80 for http.
68    let (host, port) = match authority.rsplit_once(':') {
69        Some((h, p)) => {
70            let port: u16 = p
71                .parse()
72                .map_err(|_| format!("`{url}`: invalid port `{p}`"))?;
73            (h.to_string(), port)
74        }
75        None => (authority.to_string(), 80),
76    };
77    if host.is_empty() {
78        return Err(format!("`{url}`: wait-for `http` target has an empty host"));
79    }
80
81    Ok(HttpTarget {
82        host,
83        port,
84        path: if path.is_empty() { "/".to_string() } else { path },
85    })
86}
87
88/// Is `status` acceptable? With `expect = Some(n)` only an exact `n` passes;
89/// otherwise any 2xx/3xx (the conventional "up and serving" band — a redirect
90/// from a health endpoint still means the server is listening).
91pub fn http_status_ok(status: u16, expect: Option<u16>) -> bool {
92    match expect {
93        Some(want) => status == want,
94        None => (200..400).contains(&status),
95    }
96}
97
98/// One HTTP `GET` attempt against `target`. `Ok(status)` on a complete response
99/// line; `Err(reason)` on connect / write / read failure or a malformed status
100/// line. `attempt_timeout` bounds the whole connect+request+response.
101pub async fn probe_http_once(
102    target: &HttpTarget,
103    attempt_timeout: Duration,
104) -> Result<u16, String> {
105    let fut = async {
106        let mut stream = TcpStream::connect((target.host.as_str(), target.port))
107            .await
108            .map_err(|e| format!("connect {}:{}: {e}", target.host, target.port))?;
109        let req = format!(
110            "GET {} HTTP/1.1\r\nHost: {}\r\nConnection: close\r\nUser-Agent: qed-waitfor\r\nAccept: */*\r\n\r\n",
111            target.path, target.host,
112        );
113        stream
114            .write_all(req.as_bytes())
115            .await
116            .map_err(|e| format!("write request: {e}"))?;
117
118        // We only need the status line. Read until we have the first CRLF (or
119        // the connection closes / a small cap is hit), then parse it.
120        let mut buf = Vec::with_capacity(256);
121        let mut chunk = [0u8; 256];
122        loop {
123            let n = stream
124                .read(&mut chunk)
125                .await
126                .map_err(|e| format!("read response: {e}"))?;
127            if n == 0 {
128                break;
129            }
130            buf.extend_from_slice(&chunk[..n]);
131            if buf.windows(2).any(|w| w == b"\r\n") || buf.len() > 8192 {
132                break;
133            }
134        }
135        parse_status_line(&buf)
136    };
137
138    timeout(attempt_timeout, fut)
139        .await
140        .map_err(|_| format!("no response within {}ms", attempt_timeout.as_millis()))?
141}
142
143/// Parse `HTTP/1.1 200 OK\r\n…` → `200`. Errors when the bytes don't look like
144/// an HTTP status line (e.g. a TLS handshake on a plaintext probe).
145fn parse_status_line(buf: &[u8]) -> Result<u16, String> {
146    let head = String::from_utf8_lossy(buf);
147    let first = head.lines().next().unwrap_or("").trim();
148    let mut parts = first.split_whitespace();
149    let version = parts.next().unwrap_or("");
150    if !version.starts_with("HTTP/") {
151        return Err(format!("not an HTTP response (got `{first}`)"));
152    }
153    let code = parts
154        .next()
155        .ok_or_else(|| format!("malformed status line `{first}`"))?;
156    code.parse::<u16>()
157        .map_err(|_| format!("malformed status code in `{first}`"))
158}
159
160/// One TCP connect attempt against `addr` (`host:port`). `Ok(())` the moment the
161/// port accepts; `Err(reason)` on connect failure or timeout. No bytes are
162/// exchanged — a successful connect is the whole signal.
163pub async fn probe_tcp_once(addr: &str, attempt_timeout: Duration) -> Result<(), String> {
164    let addr = addr.trim();
165    // Validate the shape up front so a `host` with no port fails clearly rather
166    // than as an opaque resolver error.
167    if addr.rsplit_once(':').is_none_or(|(_, p)| p.is_empty()) {
168        return Err(format!("`{addr}`: wait-for `tcp` target must be `host:port`"));
169    }
170    let fut = TcpStream::connect(addr);
171    match timeout(attempt_timeout, fut).await {
172        Ok(Ok(_stream)) => Ok(()),
173        Ok(Err(e)) => Err(format!("connect {addr}: {e}")),
174        Err(_) => Err(format!(
175            "connect {addr}: no response within {}ms",
176            attempt_timeout.as_millis()
177        )),
178    }
179}
180
181#[cfg(test)]
182mod tests {
183    use super::*;
184
185    #[test]
186    fn parses_http_url_with_port_and_path() {
187        let t = parse_http_url("http://localhost:3000/health").unwrap();
188        assert_eq!(t.host, "localhost");
189        assert_eq!(t.port, 3000);
190        assert_eq!(t.path, "/health");
191    }
192
193    #[test]
194    fn parses_http_url_defaults_port_and_path() {
195        let t = parse_http_url("http://example.com").unwrap();
196        assert_eq!(t.host, "example.com");
197        assert_eq!(t.port, 80);
198        assert_eq!(t.path, "/");
199    }
200
201    #[test]
202    fn rejects_https_with_pointed_message() {
203        let err = parse_http_url("https://localhost/health").unwrap_err();
204        assert!(err.contains("https"), "got: {err}");
205        assert!(err.contains("R513-F3"), "got: {err}");
206    }
207
208    #[test]
209    fn rejects_non_http_scheme() {
210        assert!(parse_http_url("localhost:3000/health").is_err());
211        assert!(parse_http_url("http:///health").is_err());
212    }
213
214    #[test]
215    fn rejects_bad_port() {
216        assert!(parse_http_url("http://localhost:notaport/").is_err());
217    }
218
219    #[test]
220    fn status_ok_band_and_exact() {
221        assert!(http_status_ok(200, None));
222        assert!(http_status_ok(204, None));
223        assert!(http_status_ok(302, None));
224        assert!(!http_status_ok(404, None));
225        assert!(!http_status_ok(500, None));
226        // exact
227        assert!(http_status_ok(204, Some(204)));
228        assert!(!http_status_ok(200, Some(204)));
229    }
230
231    #[test]
232    fn parses_status_line() {
233        assert_eq!(parse_status_line(b"HTTP/1.1 200 OK\r\n").unwrap(), 200);
234        assert_eq!(parse_status_line(b"HTTP/1.0 503 Service Unavailable\r\n").unwrap(), 503);
235        assert!(parse_status_line(b"garbage\r\n").is_err());
236        assert!(parse_status_line(b"\x16\x03\x01garbage").is_err());
237    }
238
239    #[tokio::test]
240    async fn tcp_probe_succeeds_against_a_live_listener() {
241        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
242        let addr = listener.local_addr().unwrap().to_string();
243        let res = probe_tcp_once(&addr, Duration::from_secs(1)).await;
244        assert!(res.is_ok(), "got: {res:?}");
245    }
246
247    #[tokio::test]
248    async fn tcp_probe_fails_against_a_dead_port() {
249        // Bind then drop to free a port nothing is listening on.
250        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
251        let addr = listener.local_addr().unwrap().to_string();
252        drop(listener);
253        let res = probe_tcp_once(&addr, Duration::from_millis(200)).await;
254        assert!(res.is_err(), "expected connect failure on a dead port");
255    }
256
257    #[tokio::test]
258    async fn tcp_probe_rejects_missing_port() {
259        assert!(probe_tcp_once("localhost", Duration::from_millis(100))
260            .await
261            .is_err());
262    }
263
264    #[tokio::test]
265    async fn http_probe_reads_status_from_a_canned_server() {
266        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
267        let addr = listener.local_addr().unwrap();
268        tokio::spawn(async move {
269            if let Ok((mut sock, _)) = listener.accept().await {
270                let mut scratch = [0u8; 1024];
271                let _ = sock.read(&mut scratch).await;
272                let _ = sock
273                    .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
274                    .await;
275            }
276        });
277        let target = HttpTarget {
278            host: addr.ip().to_string(),
279            port: addr.port(),
280            path: "/health".to_string(),
281        };
282        let status = probe_http_once(&target, Duration::from_secs(2)).await.unwrap();
283        assert_eq!(status, 200);
284    }
285
286    #[tokio::test]
287    async fn http_probe_surfaces_a_503() {
288        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
289        let addr = listener.local_addr().unwrap();
290        tokio::spawn(async move {
291            if let Ok((mut sock, _)) = listener.accept().await {
292                let mut scratch = [0u8; 1024];
293                let _ = sock.read(&mut scratch).await;
294                let _ = sock
295                    .write_all(b"HTTP/1.1 503 Service Unavailable\r\n\r\n")
296                    .await;
297            }
298        });
299        let target = HttpTarget {
300            host: addr.ip().to_string(),
301            port: addr.port(),
302            path: "/".to_string(),
303        };
304        let status = probe_http_once(&target, Duration::from_secs(2)).await.unwrap();
305        assert_eq!(status, 503);
306        assert!(!http_status_ok(status, None));
307    }
308}
yah_qed/waitfor.rs

yah_qed/
waitfor.rs
