test: add comprehensive Curio installation and runtime fixtures

2026-09-09 17:14:52 +08:00
parent 0d38ef2959
commit 775d891288
38 changed files with 3609 additions and 4 deletions
@@ -0,0 +1,112 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+version = 4
+
+[[package]]
+name = "anyhow"
+version = "1.0.104"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470"
+
+[[package]]
+name = "itoa"
+version = "1.0.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
+
+[[package]]
+name = "memchr"
+version = "2.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
+
+[[package]]
+name = "proc-macro2"
+version = "1.0.107"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9"
+dependencies = [
+ "unicode-ident",
+]
+
+[[package]]
+name = "quote"
+version = "1.0.47"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001"
+dependencies = [
+ "proc-macro2",
+]
+
+[[package]]
+name = "serde"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
+dependencies = [
+ "serde_core",
+]
+
+[[package]]
+name = "serde_core"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48"
+dependencies = [
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_derive"
+version = "1.0.229"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "serde_json"
+version = "1.0.151"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
+dependencies = [
+ "itoa",
+ "memchr",
+ "serde",
+ "serde_core",
+ "zmij",
+]
+
+[[package]]
+name = "syn"
+version = "3.0.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "12df2e0110f65b775f769bb17ef989067a1d931b2eb822bd4346631eeada89f9"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "unicode-ident",
+]
+
+[[package]]
+name = "trove-qa-backend"
+version = "1.0.0"
+dependencies = [
+ "anyhow",
+ "serde_json",
+]
+
+[[package]]
+name = "unicode-ident"
+version = "1.0.24"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75"
+
+[[package]]
+name = "zmij"
+version = "1.0.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"
@@ -0,0 +1,9 @@
+[package]
+name = "trove-qa-backend"
+version = "1.0.0"
+edition = "2021"
+publish = false
+
+[dependencies]
+anyhow = "1"
+serde_json = "1"
@@ -0,0 +1,492 @@
+// Framing and request lifecycle adapted from Trove app/examples/backend-rust (Wire v1).
+// This diagnostic backend is test-only; fault modes require explicit manifest args.
+use anyhow::{anyhow, Context, Result};
+use serde_json::{json, Value};
+use std::{
+    collections::HashMap,
+    env, fs,
+    io::{Read, Write},
+    os::unix::net::UnixStream,
+    sync::{
+        atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
+        mpsc, Arc, Mutex,
+    },
+    thread,
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
+};
+
+const MAX_FRAME: usize = 8 * 1024 * 1024;
+const MAX_IN_FLIGHT: usize = 256;
+static ACTIVE_REQUESTS: AtomicUsize = AtomicUsize::new(0);
+fn active_count() -> usize {
+    ACTIVE_REQUESTS.load(Ordering::Relaxed)
+}
+
+type WireResult = std::result::Result<Value, Value>;
+type Pending = Arc<Mutex<HashMap<String, mpsc::Sender<Value>>>>;
+type Active = Arc<Mutex<HashMap<String, Arc<AtomicBool>>>>;
+type Subscriptions = Arc<Mutex<HashMap<String, Subscription>>>;
+struct Subscription {
+    task_id: String,
+    caller: Value,
+    seq: u64,
+}
+struct Queued {
+    bytes: Vec<u8>,
+    flushed: Option<mpsc::Sender<()>>,
+}
+#[derive(Clone)]
+struct Writer {
+    sender: mpsc::SyncSender<Queued>,
+    bytes: Arc<AtomicUsize>,
+}
+impl Writer {
+    fn new(mut socket: UnixStream) -> Self {
+        let (sender, receiver) = mpsc::sync_channel::<Queued>(512);
+        let bytes = Arc::new(AtomicUsize::new(0));
+        let budget = bytes.clone();
+        thread::spawn(move || {
+            for next in receiver {
+                let result = socket.write_all(&next.bytes).and_then(|_| socket.flush());
+                budget.fetch_sub(next.bytes.len(), Ordering::Relaxed);
+                if result.is_err() {
+                    break;
+                }
+                if let Some(flushed) = next.flushed {
+                    let _ = flushed.send(());
+                }
+            }
+        });
+        Self { sender, bytes }
+    }
+    fn send(&self, message: Value) -> Result<()> {
+        let payload = serde_json::to_vec(&message)?;
+        if payload.is_empty() || payload.len() > MAX_FRAME {
+            return Err(anyhow!("Frame exceeds 8 MiB"));
+        }
+        let size = payload.len() + 4;
+        if self
+            .bytes
+            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |used| {
+                (used + size <= 16 * 1024 * 1024).then_some(used + size)
+            })
+            .is_err()
+        {
+            return Err(anyhow!("Outbound byte queue is full"));
+        }
+        let mut bytes = (payload.len() as u32).to_be_bytes().to_vec();
+        bytes.extend(payload);
+        if let Err(error) = self.sender.try_send(Queued {
+            bytes,
+            flushed: None,
+        }) {
+            self.bytes.fetch_sub(size, Ordering::Relaxed);
+            return Err(anyhow!("Outbound queue unavailable: {error}"));
+        }
+        Ok(())
+    }
+    fn flush(&self) {
+        let (tx, rx) = mpsc::channel();
+        if self
+            .sender
+            .try_send(Queued {
+                bytes: vec![],
+                flushed: Some(tx),
+            })
+            .is_ok()
+        {
+            let _ = rx.recv_timeout(Duration::from_millis(500));
+        }
+    }
+    fn response(&self, id: Value, result: WireResult) {
+        let message = match result {
+            Ok(result) => json!({"v":1,"type":"response","id":id,"result":result}),
+            Err(error) => json!({"v":1,"type":"response","id":id,"error":error}),
+        };
+        let _ = self.send(message);
+    }
+}
+fn error(code: &str, message: impl ToString) -> Value {
+    json!({"code":code,"message":message.to_string(),"retryable":false})
+}
+fn key(id: &Value) -> String {
+    id.to_string()
+}
+fn read(reader: &mut UnixStream) -> Result<Value> {
+    let mut header = [0; 4];
+    reader.read_exact(&mut header)?;
+    let length = u32::from_be_bytes(header) as usize;
+    if length == 0 || length > MAX_FRAME {
+        return Err(anyhow!("Invalid frame length"));
+    }
+    let mut payload = vec![0; length];
+    reader.read_exact(&mut payload)?;
+    let message: Value = serde_json::from_slice(&payload)?;
+    if !message.is_object() || message["v"] != 1 {
+        return Err(anyhow!("Invalid message version"));
+    }
+    Ok(message)
+}
+fn now_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as u64
+}
+fn progress(writer: &Writer, subscriptions: &Subscriptions, context: &Value, amount: f64) {
+    for (id, sub) in subscriptions.lock().unwrap().iter_mut() {
+        if sub.caller == context["caller"] {
+            sub.seq += 1;
+            let _ = writer.send(json!({"v":1,"type":"event","subscriptionId":id,"event":"progress","seq":sub.seq,"data":{"taskId":sub.task_id,"progress":amount}}));
+        }
+    }
+}
+fn call_host(
+    writer: &Writer,
+    pending: &Pending,
+    ids: &AtomicU64,
+    service: &str,
+    method: &str,
+    params: Value,
+    context: &Value,
+    cancelled: &AtomicBool,
+) -> WireResult {
+    let id = format!("rust_{}", ids.fetch_add(1, Ordering::Relaxed));
+    let (tx, rx) = mpsc::channel();
+    {
+        let mut pending = pending.lock().unwrap();
+        if pending.len() >= MAX_IN_FLIGHT {
+            return Err(error(
+                "request.too_many_in_flight",
+                "Host request limit reached",
+            ));
+        }
+        pending.insert(id.clone(), tx);
+    }
+    let timeout_ms = context["deadlineUnixMs"]
+        .as_u64()
+        .unwrap_or(now_ms() + 30_000)
+        .saturating_sub(now_ms())
+        .min(30_000);
+    if let Err(problem) = writer.send(json!({"v":1,"type":"request","id":id,"service":service,"method":method,"params":params,"options":{"timeoutMs":timeout_ms},"context":context})) {
+        pending.lock().unwrap().remove(&id); return Err(error("transport.backpressure", problem));
+    }
+    let deadline = Instant::now() + Duration::from_millis(timeout_ms);
+    let result = loop {
+        if cancelled.load(Ordering::Relaxed) {
+            break Err(error("request.cancelled", "Request cancelled"));
+        }
+        if Instant::now() >= deadline {
+            break Err(error("request.timeout", "Host request timed out"));
+        }
+        match rx.recv_timeout(Duration::from_millis(10)) {
+            Ok(response) => {
+                break match response.get("error") {
+                    Some(error) => Err(error.clone()),
+                    None => Ok(response["result"].clone()),
+                }
+            }
+            Err(mpsc::RecvTimeoutError::Timeout) => {}
+            Err(_) => break Err(error("provider.disconnected", "Host disconnected")),
+        }
+    };
+    if pending.lock().unwrap().remove(&id).is_some() {
+        let _ = writer.send(json!({"v":1,"type":"cancel","id":id}));
+    }
+    result
+}
+fn request(
+    message: &Value,
+    writer: &Writer,
+    pending: &Pending,
+    ids: &AtomicU64,
+    subscriptions: &Subscriptions,
+    cancelled: &AtomicBool,
+) -> WireResult {
+    if !matches!(
+        message["service"].as_str(),
+        Some(service) if service == "@self" || service == format!("{}.probe", env::var("TROVE_CURIO_ID").unwrap_or_default())
+    ) {
+        return Err(error("service.not_found", "Unknown service"));
+    }
+    let context = &message["context"];
+    match message["method"].as_str() {
+        Some("echo") => return Ok(message["params"].clone()),
+        Some("info") => {
+            return Ok(json!({
+                "pid":std::process::id(), "curioId":env::var("TROVE_CURIO_ID").unwrap_or_default(),
+                "version":env::var("TROVE_CURIO_VERSION").unwrap_or_default(),
+                "label":env::var("QA_LABEL").unwrap_or_default(), "args":env::args().skip(1).collect::<Vec<_>>(),
+                "activeRequests":active_count(), "subscriptions":subscriptions.lock().unwrap().len()
+            }))
+        }
+        Some("delay") => {
+            let ms = message["params"]["ms"]
+                .as_u64()
+                .filter(|ms| *ms <= 10_000)
+                .ok_or_else(|| {
+                    error(
+                        "request.invalid_params",
+                        "ms must be an integer in 0..10000",
+                    )
+                })?;
+            let deadline = Instant::now() + Duration::from_millis(ms);
+            while Instant::now() < deadline {
+                if cancelled.load(Ordering::Relaxed) {
+                    return Err(error("request.cancelled", "Delay cancelled"));
+                }
+                thread::sleep(Duration::from_millis(5));
+            }
+            return Ok(message["params"]["value"].clone());
+        }
+        Some("fail") => {
+            return Err(
+                json!({"code":"qa.intentional_error","message":"Expected QA failure","retryable":false,"data":{"fixture":true}}),
+            )
+        }
+        Some("invalidResult") => return Ok(json!({"intentionally":"not the contract string"})),
+        Some("storageRoundtrip") => {
+            let key = message["params"]["key"]
+                .as_str()
+                .filter(|key| key.starts_with("qa."))
+                .ok_or_else(|| error("request.invalid_params", "key must start with qa."))?;
+            call_host(
+                writer,
+                pending,
+                ids,
+                "host.storage",
+                "set",
+                message["params"].clone(),
+                context,
+                cancelled,
+            )?;
+            return call_host(
+                writer,
+                pending,
+                ids,
+                "host.storage",
+                "get",
+                json!({"key":key}),
+                context,
+                cancelled,
+            );
+        }
+        Some("crash") if env::args().any(|arg| arg == "--allow-crash") => {
+            eprintln!("qa: deliberate exit 42");
+            std::process::exit(42);
+        }
+        _ => {}
+    }
+    let text = match message["method"].as_str() {
+        Some("uppercase") => message["params"]["text"]
+            .as_str()
+            .ok_or_else(|| error("request.invalid_params", "text must be a string"))?
+            .to_owned(),
+        _ => return Err(error("service.method_not_found", "Unknown method")),
+    };
+    progress(writer, subscriptions, context, 0.0);
+    let total = text.chars().count();
+    let mut output = String::new();
+    for (index, character) in text.chars().enumerate() {
+        if cancelled.load(Ordering::Relaxed) {
+            return Err(error("request.cancelled", "Request cancelled"));
+        }
+        output.extend(character.to_uppercase());
+        if index % 4096 == 0 {
+            progress(
+                writer,
+                subscriptions,
+                context,
+                index as f64 / total.max(1) as f64,
+            );
+            thread::yield_now();
+        }
+    }
+    if cancelled.load(Ordering::Relaxed) {
+        return Err(error("request.cancelled", "Request cancelled"));
+    }
+    progress(writer, subscriptions, context, 1.0);
+    Ok(json!({"text":output}))
+}
+fn session() -> Result<(String, String)> {
+    if let Ok(path) = env::var("TROVE_DEV_SESSION_FILE") {
+        let value: Value = serde_json::from_slice(&fs::read(path)?)?;
+        return Ok((
+            value["socketPath"]
+                .as_str()
+                .context("missing socketPath")?
+                .into(),
+            value["sessionToken"]
+                .as_str()
+                .context("missing sessionToken")?
+                .into(),
+        ));
+    }
+    Ok((
+        env::var("TROVE_SOCKET_PATH")?,
+        env::var("TROVE_SESSION_TOKEN")?,
+    ))
+}
+fn main() -> Result<()> {
+    let (path, mut token) = session()?;
+    if env::args().any(|arg| arg == "--bad-token") {
+        token = "qa-deliberately-invalid-token".into();
+    }
+    println!("qa: backend started pid={}", std::process::id());
+    eprintln!("qa: label={}", env::var("QA_LABEL").unwrap_or_default());
+    let mut reader = UnixStream::connect(path)?;
+    let writer = Writer::new(reader.try_clone()?);
+    let pending: Pending = Arc::new(Mutex::new(HashMap::new()));
+    let active: Active = Arc::new(Mutex::new(HashMap::new()));
+    let subscriptions: Subscriptions = Arc::new(Mutex::new(HashMap::new()));
+    let ids = Arc::new(AtomicU64::new(1));
+    writer.send(json!({"v":1,"type":"hello","client":"backend","sessionToken":token,"protocol":{"min":1,"max":1},"features":["call","cancel","subscribe"],"implementation":{"name":"trove-qa-backend","version":"1.0.0","language":"rust"}}))?;
+    let welcome = read(&mut reader)?;
+    if welcome["type"] != "welcome" || welcome["protocol"] != 1 {
+        return Err(anyhow!("Expected welcome"));
+    }
+    if env::args().any(|arg| arg == "--no-ready") {
+        // Stay connected but never become ready; the Host must terminate this process.
+        loop {
+            thread::sleep(Duration::from_secs(1));
+        }
+    }
+    writer.send(json!({"v":1,"type":"ready"}))?;
+    while let Ok(message) = read(&mut reader) {
+        let id = message["id"].clone();
+        match message["type"].as_str() {
+            Some("request") => {
+                let cancelled = Arc::new(AtomicBool::new(false));
+                {
+                    let mut calls = active.lock().unwrap();
+                    if calls.contains_key(&key(&id)) {
+                        writer.send(json!({"v":1,"type":"protocolError","error":error("protocol.duplicate_id","Duplicate active request ID")}))?;
+                        continue;
+                    }
+                    if calls.len() >= MAX_IN_FLIGHT {
+                        writer.response(
+                            id,
+                            Err(error(
+                                "request.too_many_in_flight",
+                                "Backend request limit reached",
+                            )),
+                        );
+                        continue;
+                    }
+                    calls.insert(key(&id), cancelled.clone());
+                }
+                let (writer, pending, ids, subscriptions, active) = (
+                    writer.clone(),
+                    pending.clone(),
+                    ids.clone(),
+                    subscriptions.clone(),
+                    active.clone(),
+                );
+                ACTIVE_REQUESTS.fetch_add(1, Ordering::Relaxed);
+                thread::spawn(move || {
+                    let result = request(
+                        &message,
+                        &writer,
+                        &pending,
+                        &ids,
+                        &subscriptions,
+                        &cancelled,
+                    );
+                    active.lock().unwrap().remove(&key(&id));
+                    ACTIVE_REQUESTS.fetch_sub(1, Ordering::Relaxed);
+                    writer.response(id, result);
+                });
+            }
+            Some("response") => {
+                if let Some(id) = id.as_str() {
+                    if let Some(tx) = pending.lock().unwrap().remove(id) {
+                        let _ = tx.send(message);
+                    }
+                }
+            }
+            Some("cancel") => {
+                if let Some(cancelled) = active.lock().unwrap().get(&key(&id)) {
+                    cancelled.store(true, Ordering::Relaxed);
+                }
+            }
+            Some("subscribe") => {
+                if !matches!(
+                    message["service"].as_str(),
+                    Some(service) if service == "@self" || service == format!("{}.probe", env::var("TROVE_CURIO_ID").unwrap_or_default())
+                ) || message["topic"] != "progress"
+                {
+                    writer.response(
+                        id,
+                        Err(error("service.method_not_found", "Unknown event topic")),
+                    );
+                    continue;
+                }
+                let Some(task_id) = message["params"]["taskId"]
+                    .as_str()
+                    .filter(|id| !id.is_empty())
+                else {
+                    writer.response(
+                        id,
+                        Err(error(
+                            "request.invalid_params",
+                            "taskId must be a nonempty string",
+                        )),
+                    );
+                    continue;
+                };
+                let mut subs = subscriptions.lock().unwrap();
+                if subs.len() >= MAX_IN_FLIGHT {
+                    writer.response(
+                        id,
+                        Err(error(
+                            "request.too_many_in_flight",
+                            "Subscription limit reached",
+                        )),
+                    );
+                    continue;
+                }
+                let sub_id = format!("rust_sub_{}", ids.fetch_add(1, Ordering::Relaxed));
+                subs.insert(
+                    sub_id.clone(),
+                    Subscription {
+                        task_id: task_id.into(),
+                        caller: message["context"]["caller"].clone(),
+                        seq: 0,
+                    },
+                );
+                writer.response(id, Ok(json!({"subscriptionId":sub_id})));
+            }
+            Some("unsubscribe") => {
+                let closed = message["subscriptionId"]
+                    .as_str()
+                    .is_some_and(|id| subscriptions.lock().unwrap().remove(id).is_some());
+                writer.response(id, Ok(json!({"closed":closed})));
+            }
+            Some("shutdown") => {
+                if env::args().any(|arg| arg == "--ignore-shutdown") {
+                    eprintln!("qa: deliberately ignoring shutdown");
+                    loop {
+                        thread::sleep(Duration::from_secs(1));
+                    }
+                }
+                eprintln!("qa: graceful shutdown");
+                writer.send(json!({"v":1,"type":"goodbye"}))?;
+                writer.flush();
+                // Keep the read half alive while the Host finishes its side of shutdown.
+                // The grace bound also allows exit if the Host never sends EOF.
+                reader.set_read_timeout(Some(Duration::from_millis(250)))?;
+                let mut eof = [0u8; 1];
+                let _ = reader.read(&mut eof);
+                break;
+            }
+            Some("protocolError") => break,
+            _ => {}
+        }
+    }
+    for cancelled in active.lock().unwrap().values() {
+        cancelled.store(true, Ordering::Relaxed);
+    }
+    subscriptions.lock().unwrap().clear();
+    pending.lock().unwrap().clear();
+    Ok(())
+}