crates/lake-product/src/http.rs
//! A tiny HTTP/1.1 client and server over `std::net`: what the Lake boxes and the bench's installer speak.
//! Plain on purpose — a real product uses TLS and mTLS on these doors (see the integration
//! guide's auth model); the Lake shows the calls, not the transport.
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::time::Duration;
/// The deadline of every Lake call, both ways. A door without a deadline is a finding.
pub const TIMEOUT: Duration = Duration::from_secs(10);
/// One request as the server reads it.
#[derive(Clone, Debug, Default)]
pub struct Request {
pub method: String,
pub path: String,
pub headers: Vec<(String, String)>,
pub body: Vec<u8>,
}
impl Request {
pub fn header(&self, name: &str) -> Option<&str> {
self.headers
.iter()
.find(|(k, _)| k.eq_ignore_ascii_case(name))
.map(|(_, v)| v.as_str())
}
pub fn json(&self) -> serde_json::Value {
serde_json::from_slice(&self.body).unwrap_or(serde_json::Value::Null)
}
}
/// `(status, body)`.
pub type Response = (u16, String);
/// Serve `handle` on `listener`, one thread per connection, until the process ends.
pub fn serve(listener: TcpListener, handle: impl Fn(Request) -> Response + Send + Sync + 'static) {
let handle = std::sync::Arc::new(handle);
std::thread::spawn(move || {
for stream in listener.incoming().flatten() {
let handle = handle.clone();
std::thread::spawn(move || {
let _ = stream.set_read_timeout(Some(TIMEOUT));
let _ = stream.set_write_timeout(Some(TIMEOUT));
if let Ok(req) = read_request(&stream) {
let (status, body) = handle(req);
let mut s = stream;
let _ = write!(s, "HTTP/1.1 {status} X\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", body.len());
}
});
}
});
}
fn read_request(stream: &TcpStream) -> std::io::Result<Request> {
let mut r = BufReader::new(stream);
let mut line = String::new();
r.read_line(&mut line)?;
let mut parts = line.split_whitespace();
let method = parts.next().unwrap_or_default().to_string();
let path = parts.next().unwrap_or_default().to_string();
let mut headers = Vec::new();
let mut len = 0usize;
loop {
let mut h = String::new();
r.read_line(&mut h)?;
let h = h.trim_end();
if h.is_empty() {
break;
}
if let Some((k, v)) = h.split_once(':') {
let (k, v) = (k.trim().to_string(), v.trim().to_string());
if k.eq_ignore_ascii_case("content-length") {
len = v.parse().unwrap_or(0);
}
headers.push((k, v));
}
}
let mut body = vec![0u8; len];
r.read_exact(&mut body)?;
Ok(Request {
method,
path,
headers,
body,
})
}
/// One call: `method url` with a JSON body. `Err` names an unreachable door or a timeout.
pub fn call(addr: SocketAddr, method: &str, path: &str, body: &str) -> Result<Response, String> {
call_with(
addr,
method,
path,
&[("content-type".to_string(), "application/json".to_string())],
body,
)
}
/// [`call`] with the caller's headers (a vendor's signed callback carries its signature in one).
pub fn call_with(
addr: SocketAddr,
method: &str,
path: &str,
headers: &[(String, String)],
body: &str,
) -> Result<Response, String> {
let mut s = TcpStream::connect_timeout(&addr, TIMEOUT)
.map_err(|e| format!("{method} http://{addr}{path}: {e}"))?;
s.set_read_timeout(Some(TIMEOUT))
.map_err(|e| e.to_string())?;
s.set_write_timeout(Some(TIMEOUT))
.map_err(|e| e.to_string())?;
let mut head = format!("{method} {path} HTTP/1.1\r\nhost: {addr}\r\n");
for (k, v) in headers {
head.push_str(&format!("{k}: {v}\r\n"));
}
// One write: a door that reads only its request line then closes sees the whole request.
head.push_str(&format!(
"content-length: {}\r\nconnection: close\r\n\r\n{body}",
body.len()
));
s.write_all(head.as_bytes())
.map_err(|e| format!("{method} http://{addr}{path}: {e}"))?;
let mut raw = String::new();
s.read_to_string(&mut raw)
.map_err(|e| format!("{method} http://{addr}{path}: {e}"))?;
let status = raw
.split_whitespace()
.nth(1)
.and_then(|c| c.parse().ok())
.ok_or_else(|| format!("{method} http://{addr}{path}: no status line"))?;
let body = raw
.split_once("\r\n\r\n")
.map(|(_, b)| b.to_string())
.unwrap_or_default();
Ok((status, body))
}
/// [`call`] that wants a 2xx.
pub fn ok(addr: SocketAddr, method: &str, path: &str, body: &str) -> Result<String, String> {
match call(addr, method, path, body)? {
(200..=299, b) => Ok(b),
(s, b) => Err(format!("{method} {path} answered {s}: {b}")),
}
}
#[cfg(test)]
mod tests {
//! The echo product's plain HTTP (lane TEST-SWEEP-MONETIZE, 2026-10-24): the crate had no
//! unit test of its own; the bench exercised it only through the whole lifecycle.
use super::*;
#[test]
fn a_header_is_found_whatever_its_case_and_a_bad_body_is_null() {
let r = Request {
method: "POST".into(),
path: "/pair".into(),
headers: vec![("Content-Type".into(), "application/json".into())],
body: b"{\"a\": 1}".to_vec(),
};
assert_eq!(r.header("content-type"), Some("application/json"));
assert_eq!(r.header("CONTENT-TYPE"), Some("application/json"));
assert_eq!(r.header("x-echo-user"), None);
assert_eq!(r.json()["a"], 1);
let bad = Request {
body: b"not json".to_vec(),
..Default::default()
};
assert_eq!(bad.json(), serde_json::Value::Null);
}
#[test]
fn a_request_served_on_loopback_comes_back_with_its_method_path_headers_and_body() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
serve(listener, |req| {
let who = req.header("x-echo-user").unwrap_or("nobody").to_string();
(200, serde_json::json!({ "method": req.method, "path": req.path, "who": who, "echo": req.json() }).to_string())
});
let (status, body) = call(addr, "POST", "/ticket?x=1", "{\"tenant\":\"acme\"}").unwrap();
assert_eq!(status, 200);
let j: serde_json::Value = serde_json::from_str(&body).unwrap();
assert_eq!(j["method"], "POST");
assert_eq!(j["path"], "/ticket?x=1");
assert_eq!(j["who"], "nobody", "no user header was sent");
assert_eq!(j["echo"]["tenant"], "acme");
let (status, body) = call(addr, "GET", "/login", "").unwrap();
assert_eq!(
(
status,
serde_json::from_str::<serde_json::Value>(&body).unwrap()["echo"].clone()
),
(200, serde_json::Value::Null),
"an empty body is Null"
);
}
#[test]
fn ok_keeps_a_2xx_body_and_turns_anything_else_into_an_error_naming_the_call() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
serve(listener, |req| {
if req.path == "/first-login" {
(200, "{\"door\":\"open\"}".into())
} else {
(404, "{\"error\":\"no such door\"}".into())
}
});
assert_eq!(
ok(addr, "GET", "/first-login", "").unwrap(),
"{\"door\":\"open\"}"
);
let e = ok(addr, "GET", "/nowhere", "").unwrap_err();
assert!(
e.starts_with("GET /nowhere answered 404: ") && e.contains("no such door"),
"{e}"
);
}
#[test]
fn a_door_nobody_listens_on_is_an_error_naming_the_address() {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let addr = listener.local_addr().unwrap();
drop(listener);
let e = call(addr, "GET", "/login", "").unwrap_err();
assert!(e.starts_with(&format!("GET http://{addr}/login: ")), "{e}");
}
}