memoryd: a Honcho client per workspace, and its workspace, peers and sessions made sure of

Co-Authored-By: ornith-1.5-35b-a3b <agent@wntrmute.dev>
2026-09-28 12:13:48 -07:00
parent 28df2a0e3b
commit aabfbc2f0a
39 changed files with 2324 additions and 0 deletions
@@ -0,0 +1,260 @@
+//! A fake Honcho for tests, on 127.0.0.1 and a port of its own: it answers each request from the
+//! responses recorded from rift (`tests/fixtures/honcho/`), by method and path, and records every
+//! request, body included. It can be set to answer every request with a status instead, or to
+//! answer nothing. Stopped, and its held connections released, when dropped; every wait has a
+//! deadline. Do not edit.
+
+#![allow(dead_code)] // each test file uses a different part of this module
+
+use std::collections::HashSet;
+use std::io::{Read, Write};
+use std::net::{TcpListener, TcpStream};
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
+
+use serde_json::Value;
+
+/// One request as the fake read it.
+#[derive(Debug, Clone, PartialEq)]
+pub struct Request {
+    pub method: String,
+    pub path: String,
+    /// Header names in lower case.
+    pub headers: Vec<(String, String)>,
+    pub body: Option<Value>,
+}
+
+impl Request {
+    pub fn header(&self, name: &str) -> Option<&str> {
+        self.headers
+            .iter()
+            .find(|(k, _)| k == name)
+            .map(|(_, v)| v.as_str())
+    }
+}
+
+/// How the fake answers.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Mode {
+    /// From the fixtures.
+    Fixtures,
+    /// Every request with this status. A 422 echoes the request's body, as Honcho does.
+    Status(u16),
+    /// Read the request, then answer nothing until the fake is dropped (at most 30 s).
+    Hang,
+}
+
+struct Shared {
+    mode: Mutex<Mode>,
+    requests: Mutex<Vec<Request>>,
+    /// What was made, by method and path and id: a second create answers 200, a second delete
+    /// 404.
+    made: Mutex<HashSet<String>>,
+    stop: AtomicBool,
+}
+
+pub struct FakeHoncho {
+    pub url: String,
+    port: u16,
+    shared: Arc<Shared>,
+    accept: Option<JoinHandle<()>>,
+}
+
+/// A recorded fixture: its response's status and body.
+fn fixture(name: &str) -> (u16, String) {
+    let path = PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("tests/fixtures/honcho")
+        .join(name);
+    let text = std::fs::read_to_string(&path).unwrap();
+    let value: Value = serde_json::from_str(&text).unwrap();
+    let status = u16::try_from(value["response"]["status"].as_u64().unwrap()).unwrap();
+    let body = match &value["response"]["body"] {
+        Value::String(s) if s.is_empty() => String::new(),
+        other => other.to_string(),
+    };
+    (status, body)
+}
+
+impl FakeHoncho {
+    pub fn start() -> FakeHoncho {
+        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+        let port = listener.local_addr().unwrap().port();
+        let shared = Arc::new(Shared {
+            mode: Mutex::new(Mode::Fixtures),
+            requests: Mutex::new(Vec::new()),
+            made: Mutex::new(HashSet::new()),
+            stop: AtomicBool::new(false),
+        });
+        let for_accept = Arc::clone(&shared);
+        let accept = std::thread::spawn(move || {
+            for stream in listener.incoming() {
+                if for_accept.stop.load(Ordering::SeqCst) {
+                    return;
+                }
+                let Ok(stream) = stream else { return };
+                let shared = Arc::clone(&for_accept);
+                std::thread::spawn(move || serve(stream, &shared));
+            }
+        });
+        FakeHoncho {
+            url: format!("http://127.0.0.1:{port}"),
+            port,
+            shared,
+            accept: Some(accept),
+        }
+    }
+
+    pub fn set(&self, mode: Mode) {
+        *self.shared.mode.lock().unwrap() = mode;
+    }
+
+    /// Every request read so far, in order.
+    pub fn requests(&self) -> Vec<Request> {
+        self.shared.requests.lock().unwrap().clone()
+    }
+
+    /// Forget the requests read so far (not what was made).
+    pub fn clear(&self) {
+        self.shared.requests.lock().unwrap().clear();
+    }
+
+    /// Waits up to 10 s until at least `n` requests were read; true if they were.
+    pub fn wait_for(&self, n: usize) -> bool {
+        let deadline = Instant::now() + Duration::from_secs(10);
+        while Instant::now() < deadline {
+            if self.requests().len() >= n {
+                return true;
+            }
+            std::thread::sleep(Duration::from_millis(10));
+        }
+        false
+    }
+}
+
+impl Drop for FakeHoncho {
+    fn drop(&mut self) {
+        self.shared.stop.store(true, Ordering::SeqCst);
+        let _ = TcpStream::connect(("127.0.0.1", self.port));
+        if let Some(accept) = self.accept.take() {
+            let _ = accept.join();
+        }
+    }
+}
+
+/// Reads one request: the head up to its blank line, then `Content-Length` bytes of body.
+fn read_request(stream: &mut TcpStream) -> Option<Request> {
+    let mut buf = Vec::new();
+    let mut byte = [0u8; 1];
+    while !buf.ends_with(b"\r\n\r\n") {
+        if stream.read(&mut byte).ok()? == 0 {
+            return None;
+        }
+        buf.push(byte[0]);
+    }
+    let head = String::from_utf8(buf).ok()?;
+    let mut lines = head.split("\r\n");
+    let mut first = lines.next()?.split(' ');
+    let (method, path) = (first.next()?.to_string(), first.next()?.to_string());
+    let headers: Vec<(String, String)> = lines
+        .filter_map(|l| l.split_once(": "))
+        .map(|(k, v)| (k.to_ascii_lowercase(), v.to_string()))
+        .collect();
+    let length: usize = headers
+        .iter()
+        .find(|(k, _)| k == "content-length")
+        .map_or(Some(0), |(_, v)| v.parse().ok())?;
+    let mut body = vec![0u8; length];
+    stream.read_exact(&mut body).ok()?;
+    let body = if length == 0 {
+        None
+    } else {
+        serde_json::from_slice(&body).ok()
+    };
+    Some(Request {
+        method,
+        path,
+        headers,
+        body,
+    })
+}
+
+/// The fixture that answers `request`, by its method and the shape of its path; the second of a
+/// pair when the same thing was made (or deleted) before.
+fn answer(request: &Request, made: &Mutex<HashSet<String>>) -> (u16, String) {
+    let parts: Vec<&str> = request.path.trim_start_matches('/').split('/').collect();
+    let id = request
+        .body
+        .as_ref()
+        .and_then(|b| b["id"].as_str())
+        .unwrap_or_default()
+        .to_string();
+    let again = !made
+        .lock()
+        .unwrap()
+        .insert(format!("{} {} {id}", request.method, request.path));
+    let configured = request
+        .body
+        .as_ref()
+        .is_some_and(|b| b.get("configuration").is_some());
+    match (request.method.as_str(), parts.as_slice()) {
+        ("POST", ["v3", "workspaces"]) if again => fixture("02-workspace-create-again.json"),
+        ("POST", ["v3", "workspaces"]) => fixture("01-workspace-create.json"),
+        ("POST", ["v3", "workspaces", _, "peers"]) if again => {
+            fixture("05-peer-create-boxmaker0-again.json")
+        }
+        ("POST", ["v3", "workspaces", _, "peers"]) if configured => {
+            fixture("04-peer-create-boxmaker0.json")
+        }
+        ("POST", ["v3", "workspaces", _, "peers"]) => fixture("03-peer-create-kyle.json"),
+        ("POST", ["v3", "workspaces", _, "sessions"]) if again => {
+            fixture("07-session-create-again.json")
+        }
+        ("POST", ["v3", "workspaces", _, "sessions"]) => fixture("06-session-create.json"),
+        ("POST", ["v3", "workspaces", _, "sessions", _, "messages"]) => {
+            fixture("08-messages-add-batch.json")
+        }
+        ("DELETE", ["v3", "workspaces", _, "conclusions", _]) if again => {
+            fixture("16-conclusion-delete-again.json")
+        }
+        ("DELETE", ["v3", "workspaces", _, "conclusions", _]) => {
+            fixture("15-conclusion-delete.json")
+        }
+        _ => (404, r#"{"detail":"Not Found"}"#.to_string()),
+    }
+}
+
+fn serve(mut stream: TcpStream, shared: &Shared) {
+    let _ = stream.set_read_timeout(Some(Duration::from_secs(10)));
+    let Some(request) = read_request(&mut stream) else {
+        return;
+    };
+    shared.requests.lock().unwrap().push(request.clone());
+    let mode = *shared.mode.lock().unwrap();
+    let (status, body) = match mode {
+        Mode::Fixtures => answer(&request, &shared.made),
+        Mode::Status(422) => {
+            let input = request.body.clone().unwrap_or(Value::Null);
+            (
+                422,
+                serde_json::json!({ "detail": [{ "input": input }] }).to_string(),
+            )
+        }
+        Mode::Status(status) => (status, r#"{"detail":"refused"}"#.to_string()),
+        Mode::Hang => {
+            let deadline = Instant::now() + Duration::from_secs(30);
+            while !shared.stop.load(Ordering::SeqCst) && Instant::now() < deadline {
+                std::thread::sleep(Duration::from_millis(10));
+            }
+            return;
+        }
+    };
+    let head = format!(
+        "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+        body.len()
+    );
+    let _ = stream.write_all(head.as_bytes());
+    let _ = stream.write_all(body.as_bytes());
+}