//! The service itself, started here, sent programs nested thousands deep in //! every place a client can put one: the job's own program and an intent's //! reference (`POST /v1/jobs`, `POST /v1/run`, a dry run), the MCP run tool, //! the checks, and an error-correction circuit. A stack overflow in any of //! them would take the whole process down, on a POST of a few kilobytes. Each //! is refused with the reader's code, and the service still answers //! `/v1/health` after each. //! //! And a request whose references would take long to read is refused //! within one budget, in bounded time, and counts toward its client's //! window as every refused request does. use serde_json::{Value, json}; use std::io::{Read, Write}; use std::net::TcpStream; use std::path::Path; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; mod common; use common::Running; /// One HTTP/1.1 exchange: the status, the headers and the body. fn http(port: u16, method: &str, path: &str, body: Option<&Value>) -> (u16, String, String) { let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(120))).unwrap(); let text = body.map(Value::to_string).unwrap_or_default(); write!(s, "{method} {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nAccept: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap(); let mut out = String::new(); let _ = s.read_to_string(&mut out); let status = out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0); let (head, body) = out.split_once("\r\n\r\n").map(|(h, b)| (h.to_string(), b.to_string())).unwrap_or_default(); (status, head, body) } fn json_of(port: u16, method: &str, path: &str, body: Option<&Value>) -> (u16, Value) { let (code, _, text) = http(port, method, path, body); (code, serde_json::from_str(&text).unwrap_or_else(|e| panic!("{method} {path}: {code}: {e}: {text}"))) } /// The service, its store at `dir`, `burst` submissions a window, once it /// answers: its process and the port it bound. fn start(dir: &Path, burst: u32) -> (Running, u16) { start_with(dir, burst, &[]) } /// [`start`], with `env` set beside. fn start_with(dir: &Path, burst: u32, env: &[(&str, &str)]) -> (Running, u16) { start_from(Path::new(env!("CARGO_BIN_EXE_wai-quantum-service")), dir, burst, env) } /// [`start_with`], the service started from the binary at `bin`. It binds a /// port the system chooses, and says which in its log, so no two services /// started at once (by these tests or anything else on the host) race for one. fn start_from(bin: &Path, dir: &Path, burst: u32, env: &[(&str, &str)]) -> (Running, u16) { let mut cmd = Command::new(bin); cmd.arg("serve"); launch(cmd, dir, burst, env) } /// [`start_with`], the service started by a shell that runs `limit` first /// (`ulimit -n 256`, say). #[cfg(unix)] fn start_limited(dir: &Path, limit: &str, env: &[(&str, &str)]) -> (Running, u16) { let mut cmd = Command::new("sh"); cmd.arg("-c").arg(format!("{limit} && exec \"$0\" serve")).arg(env!("CARGO_BIN_EXE_wai-quantum-service")); launch(cmd, dir, burst_of(env), env) } #[cfg(unix)] fn burst_of(env: &[(&str, &str)]) -> u32 { env.iter().find(|(k, _)| *k == "WAI_QSVC_CLIENT_BURST").and_then(|(_, v)| v.parse().ok()).unwrap_or(1000) } /// Every service log written under `dir`. With the connection tests, which hold /// the service to a low open-file limit (`ulimit`): Unix alone, as Windows sets a /// process no such limit (the service says so as it starts, and holds its /// connections under `WAI_QSVC_CONNECTIONS` alone). #[cfg(unix)] fn logs(dir: &Path) -> String { let mut out = String::new(); for e in std::fs::read_dir(dir).into_iter().flatten().flatten() { if e.file_name().to_string_lossy().starts_with("service-") { out += &std::fs::read_to_string(e.path()).unwrap_or_default(); } } out } /// `cmd`, the service, started over the store at `dir`. fn launch(mut cmd: Command, dir: &Path, burst: u32, env: &[(&str, &str)]) -> (Running, u16) { static STARTS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); let path = dir.join(format!("service-{}.log", STARTS.fetch_add(1, std::sync::atomic::Ordering::SeqCst))); cmd.envs(env.iter().copied()) .env("WAI_QSVC_ADDR", "127.0.0.1:0") .env("WAI_QSVC_STORE", dir.join("store")) .env("WAI_QSVC_KEY", dir.join("signing.key")) .env("WAI_QSVC_JOB_GB", "2") .env("WAI_QSVC_BUDGET_GB", "4") .env("WAI_QSVC_WORKERS", "1") .env("WAI_QSVC_ENERGY", "0") .env("WAI_QSVC_CLIENT_BURST", burst.to_string()) .stdout(Stdio::null()) .stderr(std::fs::File::create(&path).map(Stdio::from).unwrap_or_else(|_| Stdio::null())); // A binary this test process has just written (a copy to replace) can // be busy for a moment: a child another test forked between the write's // open and its close holds the file open for writing until it execs // (`ETXTBSY`, Linux), whatever close-on-exec says. Started again shortly. let tried = Instant::now(); let mut child = Running::new(loop { match cmd.spawn() { Err(e) if e.kind() == std::io::ErrorKind::ExecutableFileBusy && tried.elapsed() < Duration::from_secs(10) => std::thread::sleep(Duration::from_millis(50)), started => break started.expect("the service starts"), } }); let log = || std::fs::read_to_string(&path).unwrap_or_default(); let t = Instant::now(); let port = loop { if let Some(p) = log().lines().find_map(|l| l.split(" on 127.0.0.1:").nth(1)?.split(',').next()?.trim().parse::().ok()) { break p; } if let Ok(Some(status)) = child.try_wait() { panic!("the service ended as it started ({status}): {}", log()); } assert!(t.elapsed() < Duration::from_secs(120), "the service did not start in 120 s: {}", log()); std::thread::sleep(Duration::from_millis(50)); }; while TcpStream::connect(("127.0.0.1", port)).is_err() || http(port, "GET", "/v1/health", None).0 != 200 { if let Ok(Some(status)) = child.try_wait() { panic!("the service ended ({status}): {}", log()); } assert!(t.elapsed() < Duration::from_secs(120), "the service did not answer in 120 s: {}", log()); std::thread::sleep(Duration::from_millis(50)); } (child, port) } /// The service is up and answering, and its process has not ended. fn alive(port: u16, svc: &mut Child, after: &str) { assert!(svc.try_wait().unwrap().is_none(), "the service ended after {after}"); let (code, h) = json_of(port, "GET", "/v1/health", None); assert_eq!((code, h["ok"].as_bool()), (200, Some(true)), "after {after}: {h}"); } fn dir(name: &str) -> std::path::PathBuf { let d = std::env::temp_dir().join(format!("wai-qsvc-{name}-{}", std::process::id())); let _ = std::fs::remove_dir_all(&d); std::fs::create_dir_all(&d).unwrap(); d } const HEAD: &str = "OPENQASM 3.0;\nqubit[1] q;\nbit[1] c;\n"; /// `rz(((…0.5…))) q[0];`, `n` parentheses deep: about 2n bytes. fn parens(n: usize) -> String { format!("{HEAD}rz({}0.5{}) q[0];\nc = measure q;\n", "(".repeat(n), ")".repeat(n)) } #[test] fn programs_nested_thousands_deep_are_refused_and_the_service_runs_on() { let d = dir("depth"); let (mut svc, port) = start(&d, 1000); let deep = parens(5_000); let blocks = format!("{HEAD}{}x q[0];{}\nc = measure q;\n", "{ ".repeat(5_000), " }".repeat(5_000)); let power = format!("{HEAD}rz(1{}) q[0];\nc = measure q;\n", " ** 1".repeat(50_000)); let refused = |code: u16, body: &Value, what: &str| { assert!((400..500).contains(&code), "{what}: {code} {body}"); let text = body.to_string(); assert!(text.contains("Q006"), "{what}: {body}"); }; // The job's own program, through both submissions and a dry run. for (path, qasm, what) in [("/v1/jobs", &deep, "parentheses"), ("/v1/run", &blocks, "blocks"), ("/v1/jobs", &power, "a power")] { let (code, b) = json_of(port, "POST", path, Some(&json!({ "qasm": qasm, "shots": 10 }))); refused(code, &b, what); alive(port, &mut svc, what); } let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": deep, "shots": 10, "dry_run": true }))); assert_eq!(code, 200, "{b}"); assert_eq!(b["accepted"], false, "{b}"); refused(400, &b["refusal"], "a dry run"); alive(port, &mut svc, "a dry run"); // An intent's reference: `rz(((((…0…))))) q[0];`. let reference = format!("{HEAD}rz({}0{}) q[0];\n", "(".repeat(5_000), ")".repeat(5_000)); let bell = format!("{HEAD}h q[0];\nc = measure q;\n"); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": bell, "shots": 10, "intent": { "equivalent": reference } }))); refused(code, &b, "a reference"); alive(port, &mut svc, "a reference"); // A set's items, and a sweep's generator. let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "programs": [bell, deep], "shots": 10 }))); refused(code, &b, "a set"); alive(port, &mut svc, "a set"); let generator = format!("OPENQASM 3.0;\ninput int n;\nconst int a = {}n{};\nqubit[1] q;\nx q[0];\n", "(".repeat(5_000), ")".repeat(5_000)); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": generator, "sweep": { "n": [1, 2] }, "shots": 10 }))); refused(code, &b, "a sweep"); alive(port, &mut svc, "a sweep"); // The MCP run tool, and a run as a task. let call = |args: Value, task: bool| { let mut params = json!({ "name": "run_quantum_circuit", "arguments": args }); if task { params["task"] = json!({ "ttl": 60000 }); } json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": params }))) }; let (code, b) = call(json!({ "qasm": deep, "shots": 10 }), false); assert_eq!((code, b["result"]["isError"].as_bool()), (200, Some(true)), "{b}"); assert!(b.to_string().contains("Q006"), "{b}"); alive(port, &mut svc, "the MCP run tool"); let (code, b) = call(json!({ "qasm": bell, "shots": 10, "intent": { "equivalent": reference } }), true); assert_eq!(code, 200, "{b}"); assert!(b["error"].is_object() && b.to_string().contains("Q006"), "{b}"); alive(port, &mut svc, "an MCP task"); // The checks: each reads its programs, past its count, off the runtime. for (path, body) in [ ("/v1/critique", json!({ "qasm": deep })), ("/v1/equivalent", json!({ "a": deep, "b": bell })), ("/v1/compile", json!({ "qasm": deep })), ("/v1/estimate", json!({ "qasm": deep })), ("/v1/shortcuts", json!({ "qasm": blocks })), ] { let (code, b) = json_of(port, "POST", path, Some(&body)); assert!(code < 500, "{path}: {code} {b}"); assert!(b.to_string().contains("Q006"), "{path}: {b}"); alive(port, &mut svc, path); } // An error-correction circuit's blocks, nested fifty thousand deep (as // deep as a body of a megabyte holds). let circuit = format!("{}H 0\n{}", "REPEAT 1 {\n".repeat(50_000), "}\n".repeat(50_000)); let (code, b) = json_of(port, "POST", "/v1/qec/decode", Some(&json!({ "circuit": circuit, "shots": 10 }))); assert!((400..500).contains(&code) && b.to_string().contains("nested deeper"), "{code} {b}"); alive(port, &mut svc, "a circuit's blocks"); // Nested to what people write, a job still runs. let (code, b) = json_of(port, "POST", "/v1/run", Some(&json!({ "qasm": parens(30), "shots": 10, "seed": 1 }))); assert_eq!((code, b["status"].as_str()), (200, Some("done")), "{b}"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// References that each take long to read, many of them: refused within one /// budget for the whole request, each read once, in bounded time, as too /// large for the service, and counted toward the client's window, so a /// client cannot send such requests without end. #[test] fn a_request_with_heavy_references_is_refused_in_bounded_time_and_counted() { let d = dir("budget"); let (mut svc, port) = start(&d, 4); // Each reference loops five million times on reading, and each differs, // so none is read from another's reading. let heavy = |k: usize| format!("OPENQASM 3.0;\nqubit[1] q;\nint a = 0;\nfor int i in [0:4999999] {{ a += 1; }}\n// {k}\nx q[0];\n"); let job = json!({ "qasm": format!("{HEAD}x q[0];\n"), "shots": 10, "intent": (0..16).map(|k| json!({ "equivalent": heavy(k) })).collect::>(), }); let t = Instant::now(); let (code, _, body) = http(port, "POST", "/v1/jobs", Some(&job)); let took = t.elapsed(); assert_eq!(code, 422, "{body}"); assert!(body.contains("J031") && body.contains("steps"), "{body}"); // One budget is about as long as one program the reader takes at its // bound; sixteen references read three or four times each took minutes. assert!(took < Duration::from_secs(60), "{took:?}"); alive(port, &mut svc, "heavy references"); // Too many intents are refused before any is read. let many = json!({ "qasm": format!("{HEAD}x q[0];\n"), "intent": (0..17).map(|k| json!({ "equivalent": heavy(k) })).collect::>() }); let t = Instant::now(); let (code, _, body) = http(port, "POST", "/v1/jobs", Some(&many)); assert!(code == 400 && body.contains("J020") && body.contains("at most 16"), "{code} {body}"); assert!(t.elapsed() < Duration::from_secs(5), "{:?}", t.elapsed()); // Two refused requests and two deep ones count: the fourth is the // window's last, and the fifth is refused for the window. let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": parens(5_000), "shots": 10 }))); assert!(code == 400 && b.to_string().contains("Q006"), "{code} {b}"); let (code, head, _) = http(port, "POST", "/v1/jobs", Some(&json!({ "qasm": parens(5_000), "shots": 10 }))); assert_eq!(code, 400, "{head}"); assert!(head.to_ascii_lowercase().contains("ratelimit: \"submissions\";r=0"), "{head}"); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": format!("{HEAD}x q[0];\n"), "shots": 10 }))); assert!(code == 429 && b.to_string().contains("J030"), "{code} {b}"); alive(port, &mut svc, "the window"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// Memory, through every door that reads a program: a declaration of a /// hundred bytes that asks for 137 GB, an array of 10^10 booleans, a bit /// register of 4·10^9 bits, resets of a wide register in a loop. Each ended /// the process (an allocation that fails aborts it) or grew it without end. /// Now each is refused (`Q130`) and the service answers `/v1/health` after. /// A bit string wider than an integer is a refusal, not a 500; a run whose /// MCP arguments are not an object, a refusal, not a dropped connection. #[test] fn programs_that_would_hold_without_limit_are_refused_and_the_service_runs_on() { let d = dir("memory"); let (mut svc, port) = start(&d, 1000); let program = |body: &str| format!("OPENQASM 3.0;\nqubit[1] q;\n{body}x q[0];\n"); for body in [ "array[int[8], 4294967295] a;\n", "array[bool, 100000, 100000] a;\n", "bit[4000000000] c;\n", "qubit[60000] r;\nfor int i in [0:999999] reset r;\n", ] { let qasm = program(body); for (path, req) in [("/v1/jobs", json!({ "qasm": qasm, "dry_run": true })), ("/v1/jobs", json!({ "qasm": qasm, "shots": 10 })), ("/v1/critique", json!({ "qasm": qasm }))] { let t = Instant::now(); let (code, b) = json_of(port, "POST", path, Some(&req)); assert!(code < 500 && b.to_string().contains("Q130"), "{path} {body}: {code} {b}"); assert!(t.elapsed() < Duration::from_secs(60), "{path} {body}: {:?}", t.elapsed()); alive(port, &mut svc, body); } } let wide = program(&format!("int[32] x = \"{}\";\n", "1".repeat(200))); for (path, req) in [("/v1/jobs", json!({ "qasm": wide, "dry_run": true })), ("/v1/jobs", json!({ "qasm": wide, "shots": 10 })), ("/v1/critique", json!({ "qasm": wide }))] { let (code, b) = json_of(port, "POST", path, Some(&req)); assert!(code < 500 && b.to_string().contains("Q112"), "{path}: {code} {b}"); } for args in [json!([1, 2]), json!(5), json!("x")] { let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_circuit", "arguments": args } }))); assert_eq!(code, 200, "{b}"); assert!(b["result"]["isError"] == true || b["error"].is_object(), "{b}"); alive(port, &mut svc, "non-object arguments"); } svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// The reading child's own caps, what holds where a limit of the reader's /// might not: given 100 MiB, a set whose programs each read within the /// reader's limits but together hold more is refused for its memory, as is an error-correction circuit that unrolls past it; /// the service runs on. #[test] fn a_reading_past_its_memory_is_stopped_and_refused_and_the_service_runs_on() { let d = dir("readcap"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_READ_GB", "0.1"), ("WAI_QSVC_READ_SECS", "120")]); let big = |k: u32| format!("OPENQASM 3.0;\nqubit[2] q;\nfor int i in [0:999998] {{ x q[0]; x q[1]; }}\n// {k}\n"); let t = Instant::now(); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "programs": (0..2).map(big).collect::>(), "shots": 10 }))); assert!(code == 422 && b.to_string().contains("J031") && b.to_string().contains("MiB"), "{code} {b}"); assert!(t.elapsed() < Duration::from_secs(120), "{:?}", t.elapsed()); alive(port, &mut svc, "a set past the reading's memory"); let circuit = "H 0 1 2 3 4 5 6 7\nREPEAT 100000000 {\n X_ERROR(0.1) 0 1 2 3 4 5 6 7\n}\nM 0\nDETECTOR rec[-1]\n"; let t = Instant::now(); let (code, b) = json_of(port, "POST", "/v1/qec/decode", Some(&json!({ "circuit": circuit, "shots": 10 }))); assert!(code == 422 && b.to_string().contains("J031"), "{code} {b}"); assert!(t.elapsed() < Duration::from_secs(150), "{:?}", t.elapsed()); alive(port, &mut svc, "a circuit past the reading's memory"); svc.kill().unwrap(); svc.wait().unwrap(); // A sweep written out comes back into the service: held too, however // small its request, as that memory is the one every job shares. let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_READ_ANSWER_MIB", "1")]); let body = "x q;\n".repeat(500); let sweep = json!({ "qasm": format!("OPENQASM 3.0;\ninput int n;\nqubit q;\nfor int i in [1:n] {{\n{body}}}\n"), "sweep": { "n": (0..100).map(|k| 100 + k).collect::>() }, "shots": 1 }); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep)); assert!(code == 422 && b.to_string().contains("J031") && b.to_string().contains("gave more than"), "{code} {b}"); alive(port, &mut svc, "a sweep past what a reading gives"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A client told to try again shortly spends nothing by doing so: refused /// for its jobs at once, before anything is counted or read, its window /// stays whole however often it obeys `Retry-After`; and a job sent again /// under its key, its window spent, is still its job, given as OpenQASM or /// from another format. #[test] fn a_client_told_to_retry_spends_nothing_and_its_key_s_job_comes_back_past_the_window() { let d = dir("retry"); let (mut svc, port) = start_with(&d, 4, &[("WAI_QSVC_CLIENT_ACTIVE", "1")]); let left = |head: &str| head.to_ascii_lowercase().lines().find_map(|l| l.strip_prefix("ratelimit: \"submissions\";r=").map(|r| r.split(';').next().unwrap_or("").to_string())).unwrap_or_default(); let post = |path: &str, body: &Value, key: Option<&str>| { let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(120))).unwrap(); let text = body.to_string(); let k = key.map(|k| format!("Idempotency-Key: {k}\r\n")).unwrap_or_default(); write!(s, "POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\n{k}Content-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap(); let mut out = String::new(); let _ = s.read_to_string(&mut out); let code: u16 = out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0); let (head, body) = out.split_once("\r\n\r\n").map(|(h, b)| (h.to_string(), b.to_string())).unwrap_or_default(); (code, head, body) }; // A job that runs long, its answer lost: sent under a key, given from a // source in OpenQASM. let mut long = String::from("OPENQASM 3.0;\nqubit[24] q;\nh q;\n"); for k in 0..6000 { long += &format!("rx(0.{}) q[{}];\ncx q[{}], q[{}];\n", k % 9 + 1, k % 24, k % 24, (k + 1) % 24); } let job = json!({ "source": { "format": "qasm", "program": long }, "observables": ["Z0"] }); let (code, head, first) = post("/v1/jobs", &job, Some("lost-answer")); assert_eq!(code, 202, "{first}"); assert_eq!(left(&head), "3", "{head}"); let id = serde_json::from_str::(&first).unwrap()["id"].as_str().unwrap().to_string(); // Told twenty times to wait for it: nothing spent. let bell = json!({ "qasm": "OPENQASM 3.0; qubit q; h q;", "shots": 10 }); for _ in 0..20 { let (code, head, body) = post("/v1/jobs", &bell, None); assert!(code == 429 && body.contains("wait for one to finish"), "{code} {body}"); assert_eq!(left(&head), "3", "{head}"); // Sent again at that pace, what counts would never spend the window. assert!(head.to_ascii_lowercase().contains("retry-after: 150"), "{head}"); } // Its window spent by refusals that do count, for _ in 0..3 { let (code, _, body) = post("/v1/critique", &json!({ "qasm": "OPENQASM 3.0; qubit q; foo q;" }), None); assert_eq!(code, 200, "{body}"); } let (code, _, _) = post("/v1/critique", &json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" }), None); assert_eq!(code, 429); // its job, sent again under its key, still comes back to it. let (code, _, again) = post("/v1/jobs", &job, Some("lost-answer")); assert_eq!(code, 202, "{again}"); assert_eq!(serde_json::from_str::(&again).unwrap()["id"], id.as_str(), "{again}"); let _ = http(port, "DELETE", &format!("/v1/jobs/{id}"), None); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// An index in a few bytes of text would make what is as long as the index: /// a decoder's table for 10^14 detectors, in 79 bytes. Each way of naming one /// is refused for what it is, and the service runs on. #[test] fn an_index_past_what_a_model_holds_is_refused_and_the_service_runs_on() { let d = dir("indices"); let (mut svc, port) = start(&d, 1000); for body in [ json!({ "model": "shift_detectors 100000000000000\ndetector(1) D0\n", "detectors": [] }), json!({ "model": "shift_detectors 100000000000000\ndetector D0\n", "detectors": [] }), json!({ "model": "detector(1) D4294967294\n", "detectors": [] }), json!({ "circuit": "M 4000000000\nDETECTOR rec[-1]\n", "shots": 10 }), ] { let (code, b) = json_of(port, "POST", "/v1/qec/decode", Some(&body)); assert!((400..500).contains(&code) && b.to_string().contains("past the"), "{body}: {code} {b}"); alive(port, &mut svc, "an index"); } let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "qec_decode", "arguments": { "model": "shift_detectors 100000000000000\ndetector(1) D0\n", "detectors": [] } } }))); assert!(code == 200 && b.to_string().contains("past the"), "{code} {b}"); alive(port, &mut svc, "an index through MCP"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A program a reader takes seconds over, each differing, so none is read /// from another's reading. fn slow(k: usize) -> String { format!("OPENQASM 3.0;\nqubit[1] q;\nint a = 0;\nfor int i in [0:4999999] {{ a += 1; }}\n// {k}\nx q[0];\n") } /// `n` requests at once, each as `send` makes it: the seconds each took. fn at_once(n: usize, send: impl Fn(usize) + Sync) -> Vec { std::thread::scope(|t| { let handles: Vec<_> = (0..n) .map(|i| { let send = &send; t.spawn(move || { let began = Instant::now(); send(i); began.elapsed().as_secs_f64() }) }) .collect(); handles.into_iter().map(|h| h.join().unwrap()).collect() }) } /// How long `/v1/health` takes while `during` runs beside it, at its worst. fn health_during(port: u16, during: impl FnOnce() + Send) -> f64 { let done = std::sync::atomic::AtomicBool::new(false); std::thread::scope(|t| { let probe = t.spawn(|| { let mut worst = 0f64; while !done.load(std::sync::atomic::Ordering::SeqCst) { let began = Instant::now(); assert_eq!(http(port, "GET", "/v1/health", None).0, 200); worst = worst.max(began.elapsed().as_secs_f64()); std::thread::sleep(Duration::from_millis(20)); } worst }); // Set on the way out however `during` ends, so a failure in it fails here. struct Done<'a>(&'a std::sync::atomic::AtomicBool); impl Drop for Done<'_> { fn drop(&mut self) { self.0.store(true, std::sync::atomic::Ordering::SeqCst); } } { let _done = Done(&done); during(); } probe.join().unwrap() }) } /// A refused MCP task used to read its program on a runtime worker, with no /// budget and no count: four of them stalled every route. A task refused /// before anything was counted reads nothing, and spends nothing. #[test] fn a_refused_mcp_task_reads_nothing_and_holds_no_worker() { let d = dir("tasks"); let (mut svc, port) = start(&d, 1000); let worst = health_during(port, || { let took = at_once(4, |k| { let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_circuit", "task": {}, "arguments": { "qasm": slow(k), "idempotency_key": 5 } } }))); assert!(code == 200 && b["error"]["data"]["verdict"]["level"] == "refused", "{b}"); }); assert!(took.iter().all(|&t| t < 1.0), "{took:?}"); }); assert!(worst < 1.0, "health took {worst} s"); alive(port, &mut svc, "refused tasks"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// Every check counts, refused or not, before its bounds read anything; a /// set's receipt is read as a check is, as its sweep is written out again. #[test] fn a_check_and_a_set_s_receipt_count_before_they_read() { let d = dir("counted"); let (mut svc, port) = start(&d, 5); let left = |head: &str| head.to_ascii_lowercase().lines().find_map(|l| l.strip_prefix("ratelimit: \"submissions\";r=").map(|r| r.split(';').next().unwrap_or("").to_string())).unwrap_or_default(); let wide = format!("OPENQASM 3.0;\nqubit[22] q;\nh q;\n{}", "rz(0.3) q;\n".repeat(200)); let (code, head, body) = http(port, "POST", "/v1/equivalent", Some(&json!({ "a": wide, "b": wide }))); assert!(code == 422 && body.contains("J031"), "{code} {body}"); assert_eq!(left(&head), "4", "{head}"); let (code, head, body) = http(port, "POST", "/v1/critique", Some(&json!({ "qasm": format!("OPENQASM 3.0;\nqubit q;\nrz({}0.5{}) q;\n", "(".repeat(5_000), ")".repeat(5_000)) }))); assert!(code == 200 && body.contains("Q006"), "{code} {body}"); assert_eq!(left(&head), "3", "{head}"); let set = json!({ "request": { "qasm": "input int n;\nqubit q;\nx q;\n", "sweep": { "n": [1, 2] } }, "result": { "format": "wai.quantum.set/1", "count": 2, "items": [], "receipt": {} } }); let (code, head, body) = http(port, "POST", "/v1/verify", Some(&set)); assert!(code == 200 && body.contains("\"valid\":false"), "{code} {body}"); assert_eq!(left(&head), "2", "{head}"); let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "verify_quantum_receipt", "arguments": set } }))); assert!(code == 200 && b["result"]["structuredContent"]["valid"] == false, "{b}"); let (_, head, _) = http(port, "GET", "/v1", None); assert_eq!(left(&head), "1", "{head}"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// Reading runs off the request threads: eight clients' heavy requests at /// once leave `/v1/health` answering, as a job's reading used to stall it. /// A client has its share of the turns, told at once past it. #[test] fn reading_leaves_the_runtime_free_and_a_client_has_its_share() { let d = dir("runtime"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1")]); let from = |who: &str, path: &str, body: &Value| { let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(120))).unwrap(); let text = body.to_string(); write!(s, "POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nX-Forwarded-For: {who}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap(); let mut out = String::new(); let _ = s.read_to_string(&mut out); out }; let worst = health_during(port, || { at_once(8, |k| { let job = json!({ "qasm": format!("{HEAD}x q[0];\n"), "shots": 10, "intent": (0..16).map(|i| json!({ "equivalent": slow(100 * k + i) })).collect::>() }); let out = from(&format!("10.0.0.{k}"), "/v1/jobs", &job); assert!(out.contains("J031") || out.contains("J032"), "{out}"); }); }); assert!(worst < 1.0, "health took {worst} s"); // One client's parallel calls, as an agent makes them: served one after // another behind its share, every one answered. at_once(5, |_| { let out = from("10.0.1.1", "/v1/critique", &json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" })); assert!(out.starts_with("HTTP/1.1 200"), "{out}"); }); // Past its share, a lent turn and the requests it may have waiting, told // at once. let answers = std::sync::Mutex::new(Vec::new()); at_once(11, |k| { if k == 10 { std::thread::sleep(Duration::from_millis(300)); } let began = Instant::now(); let out = from("10.0.1.2", "/v1/critique", &json!({ "qasm": slow(1000 + k) })); answers.lock().unwrap().push((out.contains("J032"), began.elapsed().as_secs_f64())); }); let answers = answers.into_inner().unwrap(); assert!(answers.iter().any(|&(busy, took)| busy && took < 1.0), "{answers:?}"); alive(port, &mut svc, "the turns"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// Steps bound work, not time: a reading past its deadline is stopped, in its /// child, and refused, and the turn it held goes to the next. A client that /// hangs up ends its reading, and its turn comes back at once. #[test] fn a_reading_past_its_time_is_stopped_and_a_hang_up_ends_it() { let d = dir("deadline"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_READ_SECS", "1"), ("WAI_QSVC_CLIENT_READS", "1")]); let long = "OPENQASM 3.0;\nqubit q;\nint a = 0;\nfor int i in [0:9899999] { a += 1; }\nx q;\n"; let t = Instant::now(); let (code, b) = json_of(port, "POST", "/v1/critique", Some(&json!({ "qasm": long }))); assert!(code == 422 && b.to_string().contains("J031") && b.to_string().contains("1 s"), "{code} {b}"); assert!(t.elapsed() < Duration::from_secs(10), "{:?}", t.elapsed()); alive(port, &mut svc, "a reading past its time"); svc.kill().unwrap(); svc.wait().unwrap(); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_CLIENT_READS", "1")]); let text = json!({ "qasm": long }).to_string(); let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); write!(s, "POST /v1/critique HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{text}", text.len()).unwrap(); std::thread::sleep(Duration::from_millis(300)); drop(s); let began = Instant::now(); loop { let (code, _, body) = http(port, "POST", "/v1/critique", Some(&json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" }))); if code == 200 { break; } assert!(code == 503 && began.elapsed() < Duration::from_secs(2), "{code} {body} after {:?}", began.elapsed()); std::thread::sleep(Duration::from_millis(50)); } svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A check's answer reads within the check's budget: a repair of forty fixes /// of a program a read takes seconds over ran for minutes. Each pass of a /// repair is a new text, so it ends at the budget, refused. #[test] fn a_check_s_answer_is_held_to_its_budget() { let d = dir("answer"); let (mut svc, port) = start(&d, 1000); let program = format!("OPENQASM 3.0;\nqubit[1] q;\nbit[1] c;\nint x = 0;\nfor int i in [0:9899999] {{ x = i; }}\n{}", "measure q[0];\n".repeat(40)); let t = Instant::now(); let (code, b) = json_of(port, "POST", "/v1/repair", Some(&json!({ "qasm": program }))); assert!(code == 422 && b.to_string().contains("J031"), "{code} {b}"); assert!(t.elapsed() < Duration::from_secs(60), "{:?}", t.elapsed()); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A program within the reader's own bounds is taken here: its statements and /// its gates each near their bound, past what one count of both would allow. #[test] fn a_job_within_the_reader_s_bounds_reads_here() { let d = dir("own"); let (mut svc, port) = start(&d, 1000); let program = "OPENQASM 3.0;\nqubit q;\nint a = 0;\nfor int i in [0:8499999] { a += 1; }\nfor int i in [0:1399999] { x q; }\n"; let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": program, "shots": 1, "dry_run": true }))); assert!(code == 200 && b["accepted"] == true, "{code} {b}"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A client that stops sending holds its connection for no longer than the /// limits say: a header section left open is closed, a body left open is /// refused `408`, and the service answers beside them. #[test] fn a_stuck_connection_is_let_go() { let d = dir("stuck"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_HEADER_SECS", "1"), ("WAI_QSVC_BODY_SECS", "1")]); let open = |text: &str| { let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(10))).unwrap(); s.write_all(text.as_bytes()).unwrap(); s }; let mut header = open("POST /v1/jobs HTTP/1.1\r\nHost: 127.0.0.1\r\nX-Half: "); let mut body = open("POST /v1/jobs HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: 100\r\n\r\n{\"qasm\""); alive(port, &mut svc, "stuck connections"); let t = Instant::now(); let mut out = String::new(); let _ = header.read_to_string(&mut out); assert!(t.elapsed() < Duration::from_secs(5), "{:?}", t.elapsed()); out.clear(); let _ = body.read_to_string(&mut out); assert!(out.starts_with("HTTP/1.1 408"), "{out}"); assert!(t.elapsed() < Duration::from_secs(5), "{:?}", t.elapsed()); alive(port, &mut svc, "connections let go"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// `body` sent to `path` as the client `who` (behind the trusted proxy): /// the status and the whole answer. fn post_as(port: u16, who: &str, path: &str, body: &Value) -> (u16, String) { let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(150))).unwrap(); let text = body.to_string(); write!(s, "POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nX-Forwarded-For: {who}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap(); let mut out = String::new(); let _ = s.read_to_string(&mut out); (out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0), out) } /// An error-correction threshold at the service's bounds whose shots run for /// minutes: what one client used to hold every check turn with. fn heavy_threshold() -> Value { json!({ "family": "surface", "distances": [7, 7, 7, 7], "p": [0.04, 0.045, 0.05, 0.055], "shots": 12_500 }) } /// One client sending checks that each hold a turn for as long as a reading /// may take holds no more than its share of the checks: every other client's /// check is answered beside them, where it waited and was refused. #[test] fn one_client_cannot_hold_every_check_turn() { let d = dir("checkshare"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1"), ("WAI_QSVC_QEC_SECS", "60"), ("WAI_QSVC_READ_SECS", "40")]); std::thread::scope(|t| { for _ in 0..2 { t.spawn(|| post_as(port, "198.51.100.66", "/v1/qec/threshold", &heavy_threshold())); } std::thread::sleep(Duration::from_millis(1500)); for (path, body) in [("/v1/critique", json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" })), ("/v1/qec/code", json!({ "code": "surface:3", "certify": false }))] { let began = Instant::now(); let (code, out) = post_as(port, "203.0.113.7", path, &body); assert!(code == 200 && began.elapsed() < Duration::from_secs(5), "{path}: {code} after {:?}: {out}", began.elapsed()); } alive(port, &mut svc, "one client's checks"); svc.kill().unwrap(); }); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// An error-correction call at the bounds whose shots would run for minutes /// ends well inside the reading's time: refused as soon as its first shots /// say the rest would pass the call's time, with the command that runs it /// locally. One that fits is answered. #[test] fn a_qec_call_at_its_bounds_ends_well_inside_the_reading_s_time() { let d = dir("qectime"); let (mut svc, port) = start(&d, 1000); for (path, body) in [("/v1/qec/threshold", heavy_threshold()), ("/v1/qec/decode", json!({ "memory": { "distance": 7, "rounds": 7, "p": 0.05 }, "shots": 200_000 }))] { let began = Instant::now(); let (code, b) = json_of(port, "POST", path, Some(&body)); assert!(code == 422 && b.to_string().contains("J031") && b.to_string().contains("wai-quantum qec"), "{path}: {code} {b}"); assert!(began.elapsed() < Duration::from_secs(30), "{path}: {:?}", began.elapsed()); alive(port, &mut svc, path); } let (code, b) = json_of(port, "POST", "/v1/qec/threshold", Some(&json!({ "distances": [3], "p": [0.001], "shots": 1000 }))); assert_eq!(code, 200, "{b}"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// The service's own peak memory, KiB, where the platform reports it. fn peak_kib(svc: &Child) -> Option { let s = std::fs::read_to_string(format!("/proc/{}/status", svc.id())).ok()?; s.lines().find(|l| l.starts_with("VmHWM:"))?.split_whitespace().nth(1)?.parse().ok() } /// A reading's answer comes back into the service, whose memory every job /// shares: a sweep whose hundred items each copy a large field, a few hundred /// kilobytes sent, is refused for what its answer takes once read, before the /// service reads it, and the service's own memory stays small. A dry run /// takes back the plans and the points alone, and is held to what its /// submission's answer would take: it says the same sweep is refused. #[test] fn a_reading_s_answer_is_held_to_what_it_takes_once_read() { let d = dir("answer-size"); let (mut svc, port) = start(&d, 1000); let sweep = |pad: usize, dry: bool| json!({ "qasm": "OPENQASM 3.0;\ninput int n;\nqubit q;\nx q;\n", "sweep": { "n": (0..100).collect::>() }, "shots": 1, "dry_run": dry, "pad": vec![0; pad] }); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep(0, true))); assert!(code == 200 && b["accepted"] == true && b["items"].as_array().is_some_and(|i| i.len() == 100), "{code} {b}"); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep(100_000, true))); assert!(code == 200 && b["accepted"] == false && b["refusal"]["code"] == "J031" && b.to_string().contains("once read"), "{code} {b}"); // A job keeps its items: past what the service takes back, refused. let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep(100_000, false))); assert!(code == 422 && b.to_string().contains("J031") && b.to_string().contains("once read"), "{code} {b}"); if let Some(kib) = peak_kib(&svc) { assert!(kib < 400 << 10, "the service held {kib} KiB"); } alive(port, &mut svc, "a large answer"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// The binary replaced before the restart, by a rename (Linux, where a copy /// over a running binary is refused), a copy over it (macOS), or on Windows, /// which neither writes over nor renames onto a running binary, by moving the /// running one aside and putting the new one at its name: the running service /// goes on reading and running jobs with its own until it is restarted, where /// every reading failed. #[test] fn the_binary_can_be_replaced_before_the_restart() { let d = dir("replace"); let bin = d.join(format!("app/wai-quantum-service{}", std::env::consts::EXE_SUFFIX)); std::fs::create_dir_all(bin.parent().unwrap()).unwrap(); std::fs::copy(env!("CARGO_BIN_EXE_wai-quantum-service"), &bin).unwrap(); let (mut svc, port) = start_from(&bin, &d, 1000, &[]); let stub = d.join("app/stub"); std::fs::write(&stub, "#!/bin/sh\necho 'not this service' >&2\nexit 2\n").unwrap(); #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; std::fs::set_permissions(&stub, std::fs::Permissions::from_mode(0o755)).unwrap(); } if cfg!(target_os = "linux") { std::fs::rename(&stub, &bin).unwrap(); } else if cfg!(windows) { std::fs::rename(&bin, d.join("app/wai-quantum-service.old")).unwrap(); std::fs::rename(&stub, &bin).unwrap(); } else { std::fs::copy(&stub, &bin).unwrap(); } let (code, b) = json_of(port, "POST", "/v1/critique", Some(&json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" }))); assert_eq!(code, 200, "{b}"); let (code, b) = json_of(port, "POST", "/v1/run", Some(&json!({ "qasm": "OPENQASM 3.0; qubit q; bit c; h q; c = measure q;", "shots": 10 }))); assert!(code == 200 && b["status"] == "done", "{code} {b}"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// The service's memory now, KiB: resident and swapped on Linux, resident /// elsewhere (on Windows its working set, which PowerShell reads, as it has no /// `ps`). fn mem_kib(svc: &Child) -> u64 { if let Ok(s) = std::fs::read_to_string(format!("/proc/{}/status", svc.id())) { let kb = |f: &str| s.lines().find(|l| l.starts_with(f)).and_then(|l| l.split_whitespace().nth(1)?.parse::().ok()).unwrap_or(0); return kb("VmRSS:") + kb("VmSwap:"); } if cfg!(windows) { let o = Command::new("powershell").args(["-NoProfile", "-NonInteractive", "-Command", &format!("(Get-Process -Id {}).WorkingSet64", svc.id())]).output().unwrap(); let said = String::from_utf8_lossy(&o.stdout); return said.trim().parse::().unwrap_or_else(|_| panic!("PowerShell read no working set: {said} {}", String::from_utf8_lossy(&o.stderr))) / 1024; } let o = Command::new("ps").args(["-o", "rss=", "-p", &svc.id().to_string()]).output().unwrap(); String::from_utf8_lossy(&o.stdout).trim().parse().unwrap_or(0) } /// What ended jobs keep: thirty jobs from one client, each carrying a field /// of a few hundred thousand numbers (a parsed request took tens of MiB, kept /// for a day and reloaded on restart), keep their requests' text alone, and /// the service stays small, after a restart too. #[test] fn ended_jobs_keep_little_and_a_restart_keeps_it_little() { let d = dir("kept"); let (mut svc, port) = start(&d, 1000); let job = json!({ "qasm": "OPENQASM 3.0; qubit q; bit c; h q; c = measure q;", "shots": 8, "pad": vec![0; 400_000] }); let mut last = String::new(); for _ in 0..30 { let (code, b) = json_of(port, "POST", "/v1/run", Some(&job)); assert!(code == 200 && b["status"] == "done", "{code} {b}"); last = b["id"].as_str().unwrap().to_string(); } let kib = mem_kib(&svc); assert!(kib < 200 << 10, "after thirty ended jobs the service holds {kib} KiB"); svc.kill().unwrap(); svc.wait().unwrap(); let (mut svc, port) = start(&d, 1000); let (code, b) = json_of(port, "GET", &format!("/v1/jobs/{last}"), None); assert!(code == 200 && b["request"]["pad"].as_array().is_some_and(|p| p.len() == 400_000), "{code}"); let kib = mem_kib(&svc); assert!(kib < 200 << 10, "after a restart the service holds {kib} KiB"); alive(port, &mut svc, "the jobs kept"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A program that runs for minutes. fn long_job() -> Value { let mut long = String::from("OPENQASM 3.0;\nqubit[24] q;\nh q;\n"); for k in 0..6000 { long += &format!("rx(0.{}) q[{}];\ncx q[{}], q[{}];\n", k % 9 + 1, k % 24, k % 24, (k + 1) % 24); } json!({ "qasm": long, "observables": ["Z0"] }) } /// Every answer comes within the answer's time from the request's arrival: /// a run whose job is still running answers with the job to poll, by REST /// and by MCP, where it waited 25 s and 50 s and passed a proxy's wait. #[test] fn a_run_s_answer_comes_within_the_answer_s_time() { let d = dir("answer-time"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_ANSWER_SECS", "15")]); let began = Instant::now(); let (code, b) = json_of(port, "POST", "/v1/run", Some(&long_job())); assert!(code == 202 && b["id"].is_string() && began.elapsed() < Duration::from_secs(20), "{code} after {:?}", began.elapsed()); let first = b["id"].as_str().unwrap().to_string(); let began = Instant::now(); let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_circuit", "arguments": long_job() } }))); let sc = &b["result"]["structuredContent"]; assert!(code == 200 && sc["id"].is_string() && began.elapsed() < Duration::from_secs(20), "{code} after {:?}: {}", began.elapsed(), sc["status"]); assert!(sc["verdict"].is_object(), "{sc}"); for id in [first.as_str(), sc["id"].as_str().unwrap()] { let _ = http(port, "DELETE", &format!("/v1/jobs/{id}"), None); } alive(port, &mut svc, "runs answered in time"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// One client's quick calls pass its own slow, in-bounds check, on a turn /// lent beside it, and its burst of eight quick calls is answered in full, /// where they waited behind it and were refused. #[test] fn a_client_s_quick_calls_pass_its_own_slow_one_and_its_burst_is_answered() { let d = dir("ownqueue"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1"), ("WAI_QSVC_QEC_SECS", "60"), ("WAI_QSVC_READ_SECS", "40")]); let me = "198.51.100.23"; std::thread::scope(|t| { t.spawn(|| post_as(port, me, "/v1/qec/threshold", &heavy_threshold())); std::thread::sleep(Duration::from_millis(1500)); let began = Instant::now(); let (code, out) = post_as(port, me, "/v1/critique", &json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" })); // Well inside the slow call's 40 s, and inside the lent turn's 5 s. assert!(code == 200 && began.elapsed() < Duration::from_secs(5), "{code} after {:?}: {out}", began.elapsed()); let answers = at_once(8, |_| { let (code, out) = post_as(port, me, "/v1/critique", &json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" })); assert_eq!(code, 200, "{out}"); }); assert!(answers.iter().all(|&s| s < 10.0), "{answers:?}"); alive(port, &mut svc, "a client's own queue"); svc.kill().unwrap(); }); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// The caps stop a reading when no new process can be started: the kill is a /// signal the service sends itself. Linux, where a process's limits can be /// lowered while it runs (`prlimit`). #[cfg(target_os = "linux")] #[test] fn a_reading_is_stopped_when_no_process_can_be_started() { if Command::new("prlimit").arg("--version").output().is_err() { eprintln!("no prlimit here: not run"); return; } let d = dir("nproc"); // No time of its own for the call: only the reading's cap stops it. let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_READ_SECS", "4"), ("WAI_QSVC_QEC_SECS", "100000")]); let pid = svc.id().to_string(); let answer = std::thread::spawn(move || { let began = Instant::now(); let (code, b) = json_of(port, "POST", "/v1/qec/threshold", Some(&heavy_threshold())); (code, b, began.elapsed()) }); std::thread::sleep(Duration::from_millis(1500)); let ok = Command::new("prlimit").args(["--pid", &pid, "--nproc=1:1"]).status().unwrap().success(); assert!(ok, "prlimit"); let (code, b, took) = answer.join().unwrap(); assert!(code == 422 && b.to_string().contains("J031") && took < Duration::from_secs(15), "{code} after {took:?}: {b}"); assert!(svc.try_wait().unwrap().is_none(), "the service ended"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// This test process's own open-file limit raised to `n`, as far as its hard /// limit lets it: it holds hundreds of connections to the service. #[cfg(unix)] fn room_for(n: u64) { // SAFETY: getrlimit and setrlimit read and write the struct given. unsafe { let mut l = libc::rlimit { rlim_cur: 0, rlim_max: 0 }; if libc::getrlimit(libc::RLIMIT_NOFILE, &mut l) == 0 && l.rlim_cur < n as libc::rlim_t { l.rlim_cur = (n as libc::rlim_t).min(l.rlim_max); libc::setrlimit(libc::RLIMIT_NOFILE, &l); } } } /// How many connections this test process can hold open beside what it /// needs for itself and its health checks: `want`, or fewer where its own /// open-file limit is low (a whole suite run under `ulimit -n 256`). #[cfg(unix)] fn can_hold(want: usize) -> usize { // SAFETY: getrlimit writes the struct given. let soft = unsafe { let mut l = libc::rlimit { rlim_cur: 0, rlim_max: 0 }; if libc::getrlimit(libc::RLIMIT_NOFILE, &mut l) == 0 { l.rlim_cur as usize } else { 256 } }; // Under a low limit of its own the suite's tests share it, four at a time. if soft < 1024 { want.min(soft / 4) } else { want } } /// Whether this test process's hard open-file limit lets a child raise its /// soft one past 256. #[cfg(unix)] fn can_raise() -> bool { // SAFETY: getrlimit writes the struct given. unsafe { let mut l = libc::rlimit { rlim_cur: 0, rlim_max: 0 }; libc::getrlimit(libc::RLIMIT_NOFILE, &mut l) == 0 && l.rlim_max > 256 } } /// Whether a health check on a new connection is answered `200` within 3 s. With /// the connection tests, Unix alone ([`logs`]). #[cfg(unix)] fn healthy(port: u16) -> bool { let Ok(mut s) = TcpStream::connect_timeout(&std::net::SocketAddr::from(([127, 0, 0, 1], port)), Duration::from_secs(3)) else { return false }; let _ = s.set_read_timeout(Some(Duration::from_secs(3))); let mut out = String::new(); write!(s, "GET /v1/health HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n").is_ok() && s.read_to_string(&mut out).is_ok() && out.split_whitespace().nth(1) == Some("200") } /// Connections held open, from one client and from many, partway through a /// header section or a body, take no more of the service's open files than it /// has, and a health check is answered throughout: where its limit is 256 and /// cannot be raised (a launchd service's), and where only its soft limit is /// 256, which the service raises. Past a client's open requests, the rest /// are answered `503` at once, unread. #[cfg(unix)] #[test] fn held_connections_leave_room_and_health_answers() { room_for(8192); for (name, limit) in [("hard", "ulimit -Sn 256 && ulimit -Hn 256"), ("soft", "ulimit -Sn 256")] { let d = dir(&format!("files-{name}")); let (mut svc, port) = start_limited(&d, limit, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1")]); let answered = |what: &str| { let ok = (0..15).filter(|_| healthy(port)).count(); assert_eq!(ok, 15, "{name}: health answered {ok} of 15 {what}: {}", logs(&d)); }; // One client, 400 header sections that never end. let n = can_hold(400); let held: Vec = (0..n) .filter_map(|_| { let mut s = TcpStream::connect(("127.0.0.1", port)).ok()?; let _ = s.write_all(b"GET /v1/health HTTP/1.1\r\nHost: 127.0.0.1\r\n"); Some(s) }) .collect(); std::thread::sleep(Duration::from_secs(1)); answered("beside 400 header sections held open"); drop(held); // 400 clients, a body each that stops arriving. let held: Vec = (0..n) .filter_map(|i| { let mut s = TcpStream::connect(("127.0.0.1", port)).ok()?; let _ = write!(s, "POST /v1/run HTTP/1.1\r\nHost: 127.0.0.1\r\nX-Forwarded-For: 198.51.{}.{}\r\nContent-Type: application/json\r\nContent-Length: 1048576\r\n\r\n{{\"qasm\": \"", i / 200, i % 200 + 1); Some(s) }) .collect(); std::thread::sleep(Duration::from_millis(1500)); answered("beside 400 clients' bodies held open"); drop(held); // One client, 400 bodies that stop arriving: past its open requests, // refused at once. let mut held: Vec = (0..n) .filter_map(|_| { let mut s = TcpStream::connect(("127.0.0.1", port)).ok()?; let _ = write!(s, "POST /v1/run HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: 1048576\r\n\r\n{{\"qasm\": \""); Some(s) }) .collect(); answered("beside one client's 400 bodies held open"); let refused = held .iter_mut() .rev() .take(n.saturating_sub(40).min(50)) .map(|s| { let _ = s.set_read_timeout(Some(Duration::from_secs(2))); let mut out = [0u8; 512]; let n = s.read(&mut out).unwrap_or(0); let text = String::from_utf8_lossy(&out[..n]); n == 0 || (text.starts_with("HTTP/1.1 503") && text.contains("J032")) }) .filter(|refused| *refused) .count(); assert_eq!(refused, n.saturating_sub(40).min(50), "{name}: a client's requests past its open ones are refused or closed at once"); drop(held); assert!(!logs(&d).contains("Too many open files"), "{name}: {}", logs(&d)); // The limit, raised where it can be, and logged. if name == "soft" && can_raise() { assert!(logs(&d).contains("(raised from 256"), "{}", logs(&d)); } else { assert!(logs(&d).contains("open files: 256"), "{}", logs(&d)); } alive(port, &mut svc, "connections held open"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } } /// A sweep whose result is more than its client's ended jobs may keep, /// sent within every published limit, is read in full: by a poll, from /// `/v1/run` and from the MCP set tool. It is kept until it has been read, /// where it was dropped as it ended and the client was told no such job. #[test] fn a_result_larger_than_its_client_keeps_is_kept_until_it_is_read() { let d = dir("large-result"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_CLIENT_KEPT_MIB", "1")]); let sweep = |points: usize| json!({ "qasm": "OPENQASM 3.0;\ninput float theta;\nqubit[16] q;\nbit[16] c;\nh q;\nrz(theta) q[0];\nc = measure q;\n", "sweep": { "theta": (0..points).map(|i| 0.01 * (i + 1) as f64).collect::>() }, "shots": 20000, "seed": 7 }); let done = |id: &str| { let t = Instant::now(); loop { let (code, b) = json_of(port, "GET", &format!("/v1/jobs/{id}?wait=25"), None); assert_eq!(code, 200, "{id} after {:?}: {b}", t.elapsed()); if b["status"] == "done" { return b; } assert!(b["status"] == "queued" || b["status"] == "running", "{b}"); assert!(t.elapsed() < Duration::from_secs(120), "{id}"); } }; let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep(10))); assert_eq!(code, 202, "{b}"); let job = done(b["id"].as_str().unwrap()); assert_eq!(job["result"]["items"].as_array().map(Vec::len), Some(10)); let (code, b) = json_of(port, "POST", "/v1/run", Some(&sweep(10))); assert!(code == 200 || code == 202, "{code}: {b}"); let job = done(b["id"].as_str().unwrap()); assert_eq!(job["result"]["items"].as_array().map(Vec::len), Some(10)); let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_set", "arguments": sweep(10) } }))); let sc = &b["result"]["structuredContent"]; assert_eq!(code, 200, "{b}"); assert!(b["result"]["isError"] == false && sc["id"].is_string(), "{}", b["result"]["content"]); let job = done(sc["id"].as_str().unwrap()); assert_eq!(job["result"]["items"].as_array().map(Vec::len), Some(10)); alive(port, &mut svc, "large results"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A job's report is the same bytes on every route that serves it (a poll, /// its events, the MCP job tool) and after a restart: one RFC 8785 text, /// its receipt's keys in their order as well as its request's. #[test] fn a_job_reads_the_same_on_every_route_and_after_a_restart() { let d = dir("same-bytes"); let (mut svc, port) = start(&d, 1000); let request = json!({ "qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\n", "shots": 200, "seed": 9, "noise": { "depolarizing_1q": 0.000001, "readout": 1e-7, "depolarizing_2q": 0.0 }, "observables": ["Z0", "0.5 Z1"] }); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&request)); assert_eq!(code, 202, "{b}"); let id = b["id"].as_str().unwrap().to_string(); let get = || { let t = Instant::now(); loop { let (code, _, text) = http(port, "GET", &format!("/v1/jobs/{id}?wait=25"), None); assert_eq!(code, 200, "{text}"); if text.contains("\"status\":\"done\"") { return text; } assert!(t.elapsed() < Duration::from_secs(60), "{text}"); } }; let text = get(); let v: Value = serde_json::from_str(&text).unwrap(); assert_eq!(text.as_bytes(), wai_quantum::engine::receipt::canonical(&v), "a report is RFC 8785"); let (_, _, events) = http(port, "GET", &format!("/v1/jobs/{id}/events"), None); let event = events.lines().find_map(|l| l.strip_prefix("data: ").filter(|d| d.contains("\"result\""))).unwrap_or_else(|| panic!("{events}")); assert_eq!(event, text, "the job event"); let (_, _, mcp) = http(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "get_quantum_job", "arguments": { "id": id } } }))); // The job in the MCP answer is the report's own bytes, what MCP adds to // it (its verdict) after them. assert!(mcp.contains(&text[..text.len() - 1]), "the MCP job tool: {mcp}"); svc.kill().unwrap(); svc.wait().unwrap(); let (mut svc, port) = start(&d, 1000); let (code, _, again) = http(port, "GET", &format!("/v1/jobs/{id}"), None); assert_eq!(code, 200, "{again}"); assert_eq!(again, text, "after a restart"); alive(port, &mut svc, "a report on every route"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// An MCP message over 64 KiB from a client past its window is answered /// under its own id, as a tool result that says why; a resend under its key /// of a job it holds is answered with that job. #[test] fn a_large_mcp_message_past_the_window_is_answered_under_its_own_id() { let d = dir("mcp-large"); let (mut svc, port) = start(&d, 2); let big = format!("OPENQASM 3.0;\nqubit[2] q;\n{}", "h q[0];\n".repeat(9000)); let call = |id: u32, name: &str, args: Value| json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": id, "method": "tools/call", "params": { "name": name, "arguments": args } }))); let (code, first) = call(1, "run_quantum_circuit", json!({ "qasm": big, "shots": 10, "idempotency_key": "large-1" })); assert_eq!(code, 200, "{first}"); let job = first["result"]["structuredContent"]["id"].as_str().unwrap_or_else(|| panic!("{first}")).to_string(); let (code, b) = call(2, "critique_qasm", json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" })); assert_eq!(code, 200, "{b}"); let (code, b) = call(3, "critique_qasm", json!({ "qasm": big })); assert_eq!((code, b["id"].as_u64(), b["result"]["isError"].as_bool()), (200, Some(3), Some(true)), "{b}"); assert!(b["result"]["content"][0]["text"].as_str().is_some_and(|t| t.contains("J030")), "{b}"); let (code, b) = call(4, "run_quantum_circuit", json!({ "qasm": big, "shots": 10, "idempotency_key": "large-1" })); assert_eq!((code, b["id"].as_u64()), (200, Some(4)), "{b}"); assert_eq!(b["result"]["structuredContent"]["id"].as_str(), Some(job.as_str()), "{b}"); alive(port, &mut svc, "a large MCP message past the window"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// Past its window a client's body is not read, whatever key it carries /// where the client holds no job under that key; a receipt check has a /// window of its own, asked before its body is read. #[test] fn a_body_is_read_only_once_its_window_or_its_key_allows() { let d = dir("unread-key"); let (mut svc, port) = start_with(&d, 0, &[("WAI_QSVC_CLIENT_VERIFIES", "2")]); let raw = |path: &str, extra: &str, body: &str| { let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(30))).unwrap(); write!(s, "POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\n{extra}Content-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).unwrap(); let mut out = String::new(); let _ = s.read_to_string(&mut out); out.split_whitespace().nth(1).and_then(|c| c.parse::().ok()).unwrap_or(0) }; for path in ["/v1/jobs", "/v1/run"] { assert_eq!(raw(path, "Idempotency-Key: fresh-key\r\n", "{not json"), 429, "{path}: a fresh key past the window"); } // Every body gets a verdict (this one `unreadable`), within verify's own window. let codes: Vec = (0..3).map(|_| raw("/v1/verify", "", "{not json")).collect(); assert_eq!(codes, [200, 200, 429], "verify's own window"); alive(port, &mut svc, "bodies left unread"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// `tasks/result` holds its answer within the answer's time from arrival, /// as every other route does, where it held a fixed 50 s. #[test] fn a_task_s_result_is_held_within_the_answer_s_time() { let d = dir("task-wait"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_ANSWER_SECS", "10"), ("WAI_QSVC_MIN_FREE_PCT", "100")]); let program = format!("OPENQASM 3.0;\nqubit[25] q;\nh q;\n{}", (0..25).map(|i| format!("t q[{i}];\n")).collect::()); let (code, made) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_circuit", "task": {}, "arguments": { "qasm": program, "shots": 10 } } }))); assert_eq!(code, 200, "{made}"); let task = made["result"]["task"]["taskId"].as_str().unwrap_or_else(|| panic!("{made}")).to_string(); let began = Instant::now(); let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 2, "method": "tasks/result", "params": { "taskId": task } }))); let took = began.elapsed(); assert_eq!(code, 200, "{b}"); assert!(took < Duration::from_secs(12), "answered after {took:?}: {b}"); assert!(b["error"]["message"].as_str().is_some_and(|m| m.contains("still working")), "{b}"); let _ = http(port, "DELETE", &format!("/v1/jobs/{task}"), None); alive(port, &mut svc, "a task's result held"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A documented QEC call sent beside another check from the same client is /// answered: a lent turn moves onto the client's own as that comes free, and /// runs past its 5 s while no one waits, where it was stopped at 5 s. /// /// Of the two decodes sent at once, what is held is what the service counts of /// its lent turns (`wai_quantum_lent_turns_total`, [`lent_turns`]), by its own /// clock: a reading was still on its lent turn after the turn's 5 s (`kept`), /// none was stopped for anyone (`taken_back`), and both calls were answered. No /// time measured here is held to anything. The decode is sized to this host /// first ([`lent_decode`]), and a host's speed changes between that and the two /// at once (CI's runner ran one of them in 4.5 s that had taken 11.2 s alone a /// moment before, run 3966ee245): where no lent turn was kept past its 5 s the /// round checked nothing, whatever was answered, so the decode is sized again /// from the time it took and the two are sent again, [`LENT_ROUNDS`] times at /// most. So is a decode the service projects past the time a call may take /// here (`J031`), from that projection. Any other refusal, of a lent turn /// taken back above all (`503`, `J032`), fails the test in the round it is in. #[test] fn a_documented_qec_call_beside_another_check_from_its_client_is_answered() { let d = dir("lent-qec"); // The QEC, reading and answer times raised, so a slow or loaded host // runs the decode, however long it takes there, rather than refusing it. let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1"), ("WAI_QSVC_QEC_SECS", "100"), ("WAI_QSVC_READ_SECS", "100"), ("WAI_QSVC_ANSWER_SECS", "110")]); let me = "198.51.100.41"; let mut decode = lent_decode(port); let critique = json!({ "qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\n" }); for first in ["/v1/critique", "/v1/qec/decode"] { let mut round = 1; loop { let pair = [(first, if first == "/v1/critique" { &critique } else { &decode }), ("/v1/qec/decode", &decode)]; let before = lent_turns(port); let answers: Vec<(u16, String, f64)> = std::thread::scope(|t| { let calls: Vec<_> = pair .iter() .enumerate() .map(|(i, (path, body))| { t.spawn(move || { std::thread::sleep(Duration::from_millis(30 * i as u64)); let began = Instant::now(); let (code, out) = post_as(port, me, path, body); (code, out, began.elapsed().as_secs_f64()) }) }) .collect(); calls.into_iter().map(|c| c.join().unwrap()).collect() }); let after = lent_turns(port); let [lent, moved, kept, taken_back] = [0, 1, 2, 3].map(|i| after[i] - before[i]); let took: Vec = answers.iter().map(|a| format!("{} in {:.1} s", a.0, a.2)).collect(); eprintln!("{first} beside a decode of {} shots, rounds {}, round {round}: {took:?}; turns lent {lent}, moved {moved}, kept past {LENT_FOR_SECS} s {kept}, taken back {taken_back}", decode["shots"], decode["memory"]["rounds"]); // Sized again, and not judged: a decode the service will not start, as too long here now. let projected = answers.iter().filter(|a| a.0 == 422).find_map(|a| projected_secs(a.1.lines().last().unwrap_or(""))); let again = if let Some(secs) = projected { Some(secs) } else { for ((path, _), (code, out, took)) in pair.iter().zip(&answers) { assert_eq!(*code, 200, "{path} after {took:.1} s: {}", out.lines().last().unwrap_or("")); } assert_eq!(taken_back, 0, "a lent turn was taken back with no one waiting: {took:?}"); // Two decodes at once: one held the client's own turn and the other a lent one, // past that turn's 5 s (offload::LENT_FOR) by the service's count, or nothing // was checked: the lent stretch then lasted under 5 s, whatever the calls took. (first == "/v1/qec/decode" && kept == 0).then(|| answers.iter().map(|a| a.2).fold(LENT_FOR_SECS, f64::min)) }; let Some(secs) = again else { break }; let sized = resized(&decode, secs); assert!(round < LENT_ROUNDS && sized.is_some(), "no decode within the documented bounds held a lent turn past its {LENT_FOR_SECS} s here in {round} rounds, the last {decode}: {took:?}; turns lent {lent}, moved {moved}, kept {kept}"); decode = sized.unwrap(); round += 1; } } alive(port, &mut svc, "QEC calls beside others"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// How long a turn lent past a client's share lasts once others wait /// (`offload::LENT_FOR` in the service, which this test cannot import). const LENT_FOR_SECS: f64 = 5.0; /// How many times [`a_documented_qec_call_beside_another_check_from_its_client_is_answered`] /// sends a pair of calls, its decode sized again between them. const LENT_ROUNDS: usize = 4; /// What the service has counted of the turns lent in its checks pool /// (`wai_quantum_lent_turns_total{pool="checks"}` of `GET /metrics`), by event: /// `lent`, `moved`, `kept` and `taken_back` (`offload::LENT_EVENTS`). A reading /// is counted before its call is answered. fn lent_turns(port: u16) -> [u64; 4] { let (code, _, text) = http(port, "GET", "/metrics", None); assert_eq!(code, 200, "GET /metrics: {text}"); let f = wai_metrics::read::parse(&text).unwrap_or_else(|e| panic!("the exposition does not read: {e:?}\n{text}")); ["lent", "moved", "kept", "taken_back"].map(|event| wai_metrics::read::value(&f, "wai_quantum_lent_turns_total", &[("pool", "checks"), ("event", event)]).unwrap_or_else(|| panic!("no count of {event}: {text}")) as u64) } /// The time a refusal of a call as too slow here (`J031`) projects for it. fn projected_secs(refusal: &str) -> Option { refusal.contains("J031").then(|| refusal.split("would take about ").nth(1).and_then(|r| r.split(' ').next()).and_then(|s| s.parse::().ok()).filter(|s| *s > 0.0)).flatten() } /// The decode `d`, which took `secs`, sized to take about 12 s: its shots in /// proportion (at least 1000), and where the 200,000 a decode may have are too /// few, its rounds doubled (5, 9, 17, 21). None where it is already the longest /// the service documents. fn resized(d: &Value, secs: f64) -> Option { let (shots, rounds) = (d["shots"].as_u64().unwrap(), d["memory"]["rounds"].as_u64().unwrap()); let want = (shots as f64 * 12.0 / secs) as u64; let mut d = d.clone(); if secs > 12.0 || want <= 200_000 { d["shots"] = json!(want.clamp(1000, 200_000)); } else if rounds < 21 { d["shots"] = json!(200_000); d["memory"]["rounds"] = json!((rounds * 2 - 1).min(21)); } else if shots < 200_000 { d["shots"] = json!(200_000); } else { return None; } Some(d) } /// The decode [`a_documented_qec_call_beside_another_check_from_its_client_is_answered`] /// begins with, sized to this host: a memory experiment within the service's /// documented bounds (200,000 shots, 21 rounds) that takes 8 to 24 s alone here, /// about 12. Its time is measured alone, as another client, or read from a /// refusal's projection (`J031`), and it is sized again from that ([`resized`]), /// at most four times. /// /// The bounds are what it aims at, and nothing is held to them: a host whose /// speed moves between one measure and the next leaves it outside them, and the /// test sizes it again from the calls it judges. From 8 s, a decode on a lent /// turn runs past that turn's 5 s unless the host runs it 1.6 times faster in /// the burst than it did alone just before; to 24 s, two at once sharing one /// core fit the 100 s the test gives a call, with the host twice as slow again. /// At its fixed 200,000 shots it took 5.7 s alone on the Framework (barely past /// the lent 5 s, and less on a faster host), 22 s held to two CPUs beside six /// busy loops, and was projected at 101 s, and refused, on CI's loaded runner /// 7049b355 (run 28969). fn lent_decode(port: u16) -> Value { let alone = |d: &Value| -> f64 { let began = Instant::now(); let (code, out) = post_as(port, "203.0.113.97", "/v1/qec/decode", d); let secs = began.elapsed().as_secs_f64(); let last = out.lines().last().unwrap_or(""); let secs = match code { 200 => secs, 422 if last.contains("J031") => projected_secs(last).unwrap_or_else(|| panic!("a decode refused as too slow with no projected time: {last}")), _ => panic!("the decode alone: {code}: {last}"), }; eprintln!("the decode alone, {} shots, rounds {}: {code}, {secs:.1} s", d["shots"], d["memory"]["rounds"]); secs }; let mut d = json!({ "memory": { "distance": 5, "rounds": 5, "p": 0.005 }, "shots": 20_000, "seed": 1 }); let mut secs = alone(&d); for _ in 0..4 { if (8.0..=24.0).contains(&secs) { break; } let Some(sized) = resized(&d, secs) else { break }; d = sized; secs = alone(&d); } d } /// The program the round-5 verifier kept the service busy with: a layer of /// T gates and a CX ladder, thirty times over, one shot. It runs for tens of /// seconds (the verifier's 22 qubits took about 35 s on its host; 24 here, so /// a faster host still runs it past every wait below), so a `?wait=25` poll /// on it holds its whole wait. fn pump() -> Value { let n = 24; let mut q = format!("OPENQASM 3.0;\nqubit[{n}] q;\nbit[{n}] c;\n"); for _ in 0..30 { q += "t q;\n"; for i in 0..n - 1 { q += &format!("cx q[{i}], q[{}];\n", i + 1); } } q += "c = measure q;\n"; json!({ "qasm": q, "method": "auto", "shots": 1 }) } /// Whether a connection is still open, and what has been answered on it so /// far, read without waiting. With the connection tests, Unix alone ([`logs`]). #[cfg(unix)] fn so_far(s: &mut TcpStream) -> (bool, Vec) { let _ = s.set_nonblocking(true); let (mut got, mut b) = (Vec::new(), [0u8; 4096]); let open = loop { match s.read(&mut b) { Ok(0) => break false, Ok(n) => got.extend_from_slice(&b[..n]), Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break true, Err(_) => break false, } }; let _ = s.set_nonblocking(false); (open, got) } /// The round-5 verifier's flood, as its probe sent it: where the service's /// open-file limit is 256 and cannot be raised (192 connections), 210 long /// polls (`?wait=25`) on a running job, each from an address of its own /// (`10.0.k.m`), opened 3 ms apart, then health checked 25 times. They held /// every connection, none could be closed for room, and every health check /// was answered `503`. Long polls are held to a share of the connections, /// past it answered at once (`503`, with `Retry-After`), and four connections past /// the cap are kept for health checks: each is answered `200`. The same for /// event streams, which held their connections as long polls did. Where this /// test process holds fewer connections (a suite run under `ulimit -n 256`), /// the service's cap is lowered as far, so they still outnumber it. #[cfg(unix)] #[test] fn long_polls_and_streams_leave_a_health_check_answered() { room_for(8192); let n = can_hold(210); let cap = (n * 192 / 210).max(16).to_string(); let mut env = vec![("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1")]; if n < 210 { env.push(("WAI_QSVC_CONNECTIONS", cap.as_str())); } let d = dir("long-polls"); let (mut svc, port) = start_limited(&d, "ulimit -Sn 256 && ulimit -Hn 256", &env); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&pump())); assert_eq!(code, 202, "{b}"); let id = b["id"].as_str().unwrap().to_string(); for (what, path) in [("long polls", format!("/v1/jobs/{id}?wait=25")), ("event streams", format!("/v1/jobs/{id}/events"))] { let mut held: Vec = (0..n) .filter_map(|i| { let mut s = TcpStream::connect(("127.0.0.1", port)).ok()?; let _ = write!(s, "GET {path} HTTP/1.1\r\nHost: x\r\nX-Forwarded-For: 10.0.{}.{}\r\n\r\n", i % 8, i / 8 + 1); std::thread::sleep(Duration::from_millis(3)); Some(s) }) .collect(); std::thread::sleep(Duration::from_secs(3)); let ok = (0..25).filter(|_| healthy(port)).count(); // The flood was real: the job still runs, and a long poll is held // with nothing answered, a stream with its answer begun. let waiting = held .iter_mut() .map(so_far) .filter(|(open, got)| *open && if what == "long polls" { got.is_empty() } else { got.starts_with(b"HTTP/1.1 200") }) .count(); let sent = held.len(); drop(held); std::thread::sleep(Duration::from_millis(500)); let (_, now) = json_of(port, "GET", &format!("/v1/jobs/{id}"), None); assert!(now["status"] == "queued" || now["status"] == "running", "{what}: the job ended before the checks, so nothing was held for it: {now}"); assert!(waiting > 0, "{what}: none of {sent} held"); assert_eq!(ok, 25, "{what}: health answered {ok} of 25 beside {sent} {what}, {waiting} of them held: {}", logs(&d)); } let _ = http(port, "DELETE", &format!("/v1/jobs/{id}"), None); alive(port, &mut svc, "long polls and streams"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// [`post_as`] for the MCP endpoint: a tool call, and whether it was answered /// with no error. fn mcp_as(port: u16, who: &str, tool: &str, args: &Value) -> (bool, String) { let body = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": tool, "arguments": args } }); let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(150))).unwrap(); let text = body.to_string(); write!(s, "POST /mcp HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nAccept: application/json\r\nX-Forwarded-For: {who}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap(); let mut out = String::new(); let _ = s.read_to_string(&mut out); let answer: Value = out.split_once("\r\n\r\n").and_then(|(_, b)| serde_json::from_str(b).ok()).unwrap_or(Value::Null); (out.starts_with("HTTP/1.1 200") && answer["result"].is_object() && answer["result"]["isError"] != true, out) } /// The README's threshold (`qec threshold --distances 3,5,7 --p /// 0.002,0.004 --shots 20000`) made longer, as the round-5 verifier made it /// longer (30,000 shots on its host, where that took about 15 s), until it /// takes 11 s alone here, past the 10 s a request waits for a turn: its /// shots raised half again at a time to the 200,000 a threshold here takes /// in all, then its rounds a point raised from the distance to 9, 11, 13 (21 /// at most here), for as long as the service still takes it within the time /// an error-correction call may take. The longest call it took, and how /// long that took. /// /// On a loaded host the README's threshold alone comes near the 25 s a call /// takes by default, or past them: the service refuses a call it projects /// past its time (`J031`), from a pilot of the call's own shots. CI's Linux /// runner projected 28 s and refused it, and the test failed before it /// began. A call that took more than 14 s, or was refused, has its shots /// scaled to the 12 s its time (or the refusal's projection) gives, at least /// 1000, and is sent again, at most three times: a long check, past the 10 s /// a request waits for a turn. The tests that send it beside other calls /// start the service with that time raised ([`LONG_CHECKS`]), since two at /// once share the host it was measured on alone. fn slow_threshold(port: u16) -> (Value, f64) { let run = |t: &Value| -> Result { let began = Instant::now(); let (code, out) = post_as(port, "203.0.113.99", "/v1/qec/threshold", t); let secs = began.elapsed().as_secs_f64(); eprintln!("the threshold alone, {} shots, rounds {}: {code} in {secs:.1} s", t["shots"], t["rounds"]); match code { 200 => Ok(secs), 422 if out.contains("J031") => Err(out.lines().last().unwrap_or("").to_string()), _ => panic!("the threshold alone: {code}: {}", out.lines().last().unwrap_or("")), } }; let mut t = json!({ "distances": [3, 5, 7], "p": [0.002, 0.004], "shots": 20000 }); let mut first = run(&t); for _ in 0..3 { let secs = match &first { Ok(secs) if *secs <= 14.0 => break, Ok(secs) => *secs, Err(why) => { let projected = why.split("would take about ").nth(1).and_then(|r| r.split(' ').next()).and_then(|s| s.parse::().ok()).filter(|s| *s > 0.0); projected.unwrap_or_else(|| panic!("a refusal as too slow with no projected time: {why}")) } }; let shots = t["shots"].as_u64().unwrap_or(20000); let fewer = ((shots as f64 * 12.0 / secs) as u64).max(1000); if fewer >= shots { break; } t["shots"] = json!(fewer); first = run(&t); } let mut best = (t.clone(), first.unwrap_or_else(|why| panic!("the threshold was refused as too slow here, at {} shots: {why}", t["shots"]))); while best.1 < 11.0 { let shots = t["shots"].as_u64().unwrap_or(20000); let rounds = t["rounds"].as_u64().unwrap_or(7); if shots < 33_333 { t["shots"] = json!((shots * 3 / 2).min(33_333)); } else if rounds < 13 { t["rounds"] = json!(rounds + 2); } else { break; } match run(&t) { Ok(secs) => best = (t.clone(), secs), Err(_) => break, } } best } /// The README's decode (`--memory 5,5,0.003 --shots 20000 --decoder /// correlated`) as [`the_readme_s_burst_from_one_client_is_answered`] sends it, /// sized to this host. In the burst it holds the client's own turn while the /// threshold is lent one, and what the test holds is that the threshold moves /// onto that turn as the decode ends and gives the lent one back to the /// critiques. That needs the decode to end before a critique has waited 7 s /// (`offload::OWN_PATIENCE`): past that the lent turn is taken back from the /// threshold for them, which refuses it (`J032`, as the service should), and /// the move is never seen. At 20,000 shots it takes 1 to 5 s in the burst on a /// host as busy as two CPUs beside six busy loops, and took 9.1 s with the /// whole suite on three CPUs beside eighteen, where the threshold was refused. /// So it is measured alone first, as another client, and where it took more /// than 2 s its shots are scaled to the 1.5 s that time gives (at least 1000): /// with room, beside the threshold, for a host twice as slow again. fn burst_decode(port: u16) -> Value { let mut d = json!({ "memory": { "distance": 5, "rounds": 5, "p": 0.003 }, "shots": 20000, "decoder": "correlated" }); let began = Instant::now(); let (code, out) = post_as(port, "203.0.113.98", "/v1/qec/decode", &d); let secs = began.elapsed().as_secs_f64(); assert_eq!(code, 200, "the README's decode alone, after {secs:.1} s: {}", out.lines().last().unwrap_or("")); if secs > 2.0 { d["shots"] = json!(((20000.0 * 1.5 / secs) as u64).max(1000)); } eprintln!("the README's decode alone, 20000 shots: {secs:.1} s; sent with {}", d["shots"]); d } /// The service's own times raised for a test whose long checks are sized to /// this host alone ([`slow_threshold`]) and then sent two or more at once: the /// seconds an error-correction call's shots may take, a reading's and an /// answer's, as [`a_documented_qec_call_beside_another_check_from_its_client_is_answered`] /// raises them. At the defaults (25, 45 and 55 s) CI's loaded Linux runner /// sized the threshold to 30,000 shots, 17 to 20 s alone, and then, two sent at /// once, projected one or both at 27 to 35 s and refused them (`J031`): a /// refusal of a call for its own time, which these tests are not about. What /// they hold is how one client's turns are lent and given back, which these /// times do not set (`offload::WAIT`, `LENT_FOR` and `OWN_PATIENCE` do), so /// raising them changes nothing they hold. const LONG_CHECKS: [(&str, &str); 4] = [("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1"), ("WAI_QSVC_QEC_SECS", "100"), ("WAI_QSVC_READ_SECS", "100"), ("WAI_QSVC_ANSWER_SECS", "110")]; /// The round-5 verifier's burst, at the service's own limits on a client's /// turns: one client sends the README's decode (`--memory 5,5,0.003 --shots /// 20000 --decoder correlated`), a threshold, and six critiques of a Bell /// pair, 30 ms apart, over HTTP and then over MCP. The threshold, on a turn /// lent beside the decode, moved onto the client's own turn when the decode /// ended and kept the lent one, so no critique could have either: each /// waited out its 10 s and was refused (`J032`), or, with the threshold under /// 10 s, was answered only as it ended. A moved call gives its lent turn /// back: every call is answered, each critique well before the threshold. /// /// The threshold is sized to this host alone, and the burst shares the host /// with it, so the service runs with an error-correction call's time raised /// ([`LONG_CHECKS`]): at the default 25 s a loaded host refused the threshold /// (`J031`) as it ran beside the decode, and nothing was tested. Should a /// host still refuse it for its time, the threshold is measured again and /// the burst sent again, three times at most. The decode is sized to this /// host as well ([`burst_decode`]), so that it ends while the critiques may /// still wait. #[test] fn the_readme_s_burst_from_one_client_is_answered() { let d = dir("readme-burst"); let (mut svc, port) = start_with(&d, 1000, &LONG_CHECKS); let critique = json!({ "qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\n" }); for (route, me) in [("HTTP", "198.51.100.77"), ("MCP", "198.51.100.78")] { let mut tries = 0; let (answers, at, alone) = loop { tries += 1; let (threshold, alone) = slow_threshold(port); let decode = burst_decode(port); let mut calls: Vec<(&str, &str, &Value)> = vec![("/v1/qec/decode", "qec_decode", &decode), ("/v1/qec/threshold", "qec_threshold", &threshold)]; calls.extend(std::iter::repeat_n(("/v1/critique", "critique_qasm", &critique), 6)); let began = Instant::now(); let answers: Vec<(bool, String, f64)> = std::thread::scope(|t| { let each: Vec<_> = calls .iter() .enumerate() .map(|(i, (path, tool, body))| { let began = &began; t.spawn(move || { std::thread::sleep(Duration::from_millis(30 * i as u64)); let (ok, out) = if route == "HTTP" { let (code, out) = post_as(port, me, path, body); (code == 200, out) } else { mcp_as(port, me, tool, body) }; (ok, out, began.elapsed().as_secs_f64()) }) }) .collect(); each.into_iter().map(|c| c.join().unwrap()).collect() }); let at: Vec = calls.iter().zip(&answers).map(|((path, tool, _), (ok, _, s))| format!("{} {} at {s:.2} s", if route == "HTTP" { path } else { tool }, if *ok { "answered" } else { "refused" })).collect(); eprintln!("{route}, try {tries}, the threshold alone {alone:.1} s: {}", at.join(", ")); let (ok, out, _) = &answers[1]; if !ok && out.contains("J031") && out.contains("would take about") && tries < 3 { eprintln!("{route}: the threshold, measured beside the burst, was past the time a call may take, so nothing was tested; again"); continue; } break (answers, at, alone); }; for ((ok, out, _), name) in answers.iter().zip(&at) { assert!(*ok, "{route}: {name} (the threshold alone {alone:.1} s): {}\n{at:#?}", out.lines().last().unwrap_or("")); } let threshold_at = answers[1].2; for (_, _, s) in &answers[2..] { assert!(*s + 1.0 < threshold_at, "{route}: a critique answered at {s:.2} s, the threshold at {threshold_at:.2} s (alone {alone:.1} s): the critiques waited for it\n{at:#?}"); } } alive(port, &mut svc, "the README's burst"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A JSON-RPC id is echoed exactly, an integer past 2^53 included. #[test] fn an_mcp_id_is_echoed_exactly() { let d = dir("mcp-ids"); let (mut svc, port) = start(&d, 1000); for id in ["9007199254740993", "-9007199254740993", "9223372036854775807", "12345678901234567890", "1730000000123456789"] { let body: Value = serde_json::from_str(&format!("{{\"jsonrpc\":\"2.0\",\"id\":{id},\"method\":\"ping\"}}")).unwrap(); let (code, _, text) = http(port, "POST", "/mcp", Some(&body)); assert_eq!(code, 200, "{text}"); assert!(text.contains(&format!("\"id\":{id},")) || text.contains(&format!("\"id\":{id}}}")), "{id}: {text}"); } alive(port, &mut svc, "ids echoed"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A job's result is held to `WAI_QSVC_RESULT_MIB` before it runs, where a /// result past it was run and then failed: a sweep whose result's bound /// passes the cap is refused at once (`422`, `J031`), naming its bound and /// the cap, on every route that takes a job, and its dry run says so; the /// largest sweep under the cap is taken and ends done, its result within its /// bound; a pulse run past the cap is refused at once as well, by its /// schedule. At the defaults, the published set at its most, 100 items of /// 65,536 outcomes each (about 150 MB as served), is refused, and 100 /// points of 30,000 shots of 16 bits each are taken. #[test] fn a_result_past_what_a_job_may_give_is_refused_before_it_runs() { let sweep = |points: usize, shots: u64| json!({ "qasm": "OPENQASM 3.0;\ninput float theta;\nqubit[16] q;\nbit[16] c;\nh q;\nrz(theta) q[0];\nc = measure q;\n", "sweep": { "theta": (0..points).map(|i| 0.01 * (i + 1) as f64).collect::>() }, "shots": shots, "seed": 7 }); let dry = |port: u16, v: &Value| { let mut v = v.clone(); v["dry_run"] = json!(true); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&v)); assert_eq!(code, 200, "{b}"); b }; let d = dir("result-cap"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_RESULT_MIB", "1")]); let cap = 1u64 << 20; assert_eq!(dry(port, &sweep(1, 2000))["bounds"]["result_bytes"].as_u64(), Some(cap)); // The most points of 2,000 shots a point under the cap, by the dry runs' bounds. let bound = |n: usize| dry(port, &sweep(n, 2000))["result_bytes"].as_u64().unwrap(); let (one, two) = (bound(1), bound(2)); let mut points = ((cap - (2 * one - two)) / (two - one)) as usize; while bound(points) > cap { points -= 1; } while bound(points + 1) <= cap { points += 1; } let past = dry(port, &sweep(points + 1, 2000)); assert!(past["accepted"] == false && past["status"] == 422 && past["refusal"]["code"] == "J031", "{past}"); // Refused at once, on every route that takes a job. let mcp = |args: Value| json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_set", "arguments": args } }); for (path, body) in [("/v1/jobs", sweep(points + 1, 2000)), ("/v1/run", sweep(points + 1, 2000)), ("/mcp", mcp(sweep(points + 1, 2000)))] { let began = Instant::now(); let (code, b) = json_of(port, "POST", path, Some(&body)); let why = if path == "/mcp" { b["result"]["content"][0]["text"].as_str().unwrap_or("").to_string() } else { b["error"].as_str().unwrap_or("").to_string() }; assert!(if path == "/mcp" { code == 200 && b["result"]["isError"] == true } else { code == 422 }, "{path}: {code} {b}"); assert!(why.contains("J031") && why.contains("could take up to") && why.contains("the 1 MiB"), "{path}: {why}"); assert!(began.elapsed() < Duration::from_secs(10), "{path}: answered after {:?}", began.elapsed()); } // At the bound: taken, and done, its result within its bound. let at = bound(points); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep(points, 2000))); assert_eq!(code, 202, "{b}"); let id = b["id"].as_str().unwrap().to_string(); let t = Instant::now(); let job = loop { let (code, b) = json_of(port, "GET", &format!("/v1/jobs/{id}?wait=25"), None); assert_eq!(code, 200, "{b}"); if b["status"] != "queued" && b["status"] != "running" { break b; } assert!(t.elapsed() < Duration::from_secs(120)); }; assert_eq!(job["status"], "done", "{}", job["error"]); let served = wai_quantum::engine::receipt::canonical(&job["result"]).len() as u64; eprintln!("{points} points of 2,000 shots: bound {at} bytes, served {served}"); assert!(served <= at && at <= cap, "{points} points: served {served} bytes, bound {at}"); // A pulse run is bounded from its schedule as well: 100,000 bits // declared (in registers of 64, as the pulse reader takes them) and one // captured, each bit named and every outcome's key as wide (about 1.4 MB // as served), are refused at once, where the run had been taken and // failed only as it ended; 100 bits are taken. let pulse = |n: usize| json!({ "method": "pulse", "shots": 100, "qasm": format!("OPENQASM 3.0;\ndefcalgrammar \"openpulse\";\ncal {{\n extern port d0;\n extern port a0;\n frame q0 = newframe(d0, 5.0e9, 0);\n frame c0 = newframe(a0, 7.1e9, 0);\n}}\ndefcal x $0 {{ play(q0, gaussian(0.4, 64dt, 16dt)); }}\ndefcal measure $0 -> bit {{ return capture_v2(c0, 4dt); }}\n{}x $0;\nb{}[{}] = measure $0;\n", (0..n.div_ceil(64)).map(|r| format!("bit[{}] b{r};\n", 64.min(n - 64 * r))).collect::(), (n - 1) / 64, (n - 1) % 64) }); let began = Instant::now(); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&pulse(100_000))); if code == 202 { let id = b["id"].as_str().unwrap_or_default().to_string(); let job = loop { let (_, b) = json_of(port, "GET", &format!("/v1/jobs/{id}?wait=25"), None); if b["status"] != "queued" && b["status"] != "running" { break b; } assert!(began.elapsed() < Duration::from_secs(300)); }; panic!("a pulse run of 100,000 bits was taken, and then after {:?}: {} {}", began.elapsed(), job["status"], job["error"]); } let why = b["error"].as_str().unwrap_or_default(); assert!(code == 422 && why.contains("J031") && why.contains("could take up to") && why.contains("the 1 MiB"), "a pulse run past the cap: {code} {b}"); assert!(began.elapsed() < Duration::from_secs(10), "answered after {:?}", began.elapsed()); let small = dry(port, &pulse(100)); assert!(small["accepted"] == true && small["result_bytes"].as_u64().is_some_and(|b| b < cap), "{small}"); alive(port, &mut svc, "results held to their cap"); svc.kill().unwrap(); svc.wait().unwrap(); // The defaults. let (mut svc, port) = start(&d, 1000); let most = dry(port, &sweep(100, 1_000_000)); let bytes = most["result_bytes"].as_u64().unwrap_or(0); eprintln!("100 points of 65,536 outcomes: bound {bytes} bytes"); assert!(most["accepted"] == false && most["refusal"]["code"] == "J031" && (130_000_000..180_000_000).contains(&bytes), "{most}"); let taken = dry(port, &sweep(100, 30_000)); eprintln!("100 points of 30,000 shots: bound {} bytes", taken["result_bytes"]); assert!(taken["accepted"] == true, "{taken}"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A job's report once it has ended, polled. fn ended(port: u16, id: &str, within: Duration) -> Value { let t = Instant::now(); loop { let (code, b) = json_of(port, "GET", &format!("/v1/jobs/{id}?wait=5"), None); assert_eq!(code, 200, "{b}"); if b["status"] != "queued" && b["status"] != "running" { return b; } assert!(t.elapsed() < within, "{id} still {} after {within:?}", b["status"]); } } /// [`json_of`] as the client `who` (through a trusted proxy). fn json_as(port: u16, who: &str, path: &str, body: &Value) -> (u16, Value) { let (code, out) = post_as(port, who, path, body); let text = out.split_once("\r\n\r\n").map(|(_, b)| b).unwrap_or(""); (code, serde_json::from_str(text).unwrap_or(Value::Null)) } /// An MCP tool's answer: its result's `structuredContent`, and whether it /// was an error. fn mcp_tool(port: u16, tool: &str, args: &Value) -> (bool, Value) { let body = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": tool, "arguments": args } }); let (code, b) = json_of(port, "POST", "/mcp", Some(&body)); assert_eq!(code, 200, "{tool}: {b}"); (b["result"]["isError"] == true, b["result"]["structuredContent"].clone()) } /// A 16-qubit sweep: `points` angles, `shots` a point, 16 measured bits. fn sweep16(points: usize, shots: u64) -> Value { json!({ "qasm": "OPENQASM 3.0;\ninput float theta;\nqubit[16] q;\nbit[16] c;\nh q;\nrz(theta) q[0];\nc = measure q;\n", "sweep": { "theta": (0..points).map(|i| 0.01 * (i + 1) as f64).collect::>() }, "shots": shots, "seed": 7 }) } /// An MCP dry run says what the MCP run tools take: it plans the job as they /// run it, its verdict asked, so the most points of a sweep its dry run /// accepts under a cap of 1 MiB are taken by `run_quantum_set`, where the /// dry run planned the job without its verdict and the run then refused it /// at once (`J031`). #[test] fn an_mcp_dry_run_says_what_the_run_tools_take() { let d = dir("mcp-dry"); let (mut svc, port) = start_with(&d, 10_000, &[("WAI_QSVC_RESULT_MIB", "1")]); let accepted = |n: usize| -> bool { let (err, v) = mcp_tool(port, "dry_run_quantum_job", &sweep16(n, 2000)); assert!(!err, "{v}"); v["accepted"] == true }; let mut n = 1; while accepted(n + 1) { n += 1; assert!(n < 100); } let (err, v) = mcp_tool(port, "run_quantum_set", &sweep16(n, 2000)); let why = v["error"].as_str().unwrap_or("").to_string(); assert!(!(err && why.contains("J031")), "{n} points, accepted by the MCP dry run, refused by the run: {why}"); let id = v["id"].as_str().unwrap_or_default().to_string(); let job = if v["status"] == "done" { v } else { ended(port, &id, Duration::from_secs(120)) }; assert_eq!(job["status"], "done", "{}", job["error"]); let (err, v) = mcp_tool(port, "run_quantum_set", &sweep16(n + 1, 2000)); assert!(err && v["error"].as_str().is_some_and(|w| w.contains("J031")), "{} points: {v}", n + 1); alive(port, &mut svc, "MCP dry runs"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A dry run that says a request is taken is taken: a sweep whose items /// each carry a large shared field read, kept, past what the service takes /// back of one reading (`J031`) is refused by its dry run too, as the dry /// run's reading is held to what the submission's would take; one under it /// is taken by both. #[test] fn a_dry_run_that_says_taken_is_taken_by_its_submission() { let d = dir("dry-weight"); let (mut svc, port) = start(&d, 10_000); for pad in [12_000usize, 20_000] { let pads = vec!["1e20"; pad].join(","); let text = format!("{{\"qasm\":\"OPENQASM 3.0;\\ninput float theta;\\nqubit q;\\nbit c;\\nrx(theta) q;\\nc = measure q;\\n\",\"sweep\":{{\"theta\":[{}]}},\"shots\":10,\"pad\":[{pads}]}}", (0..100).map(|i| format!("0.{}", i + 1)).collect::>().join(",")); let send = |dry: bool| -> (u16, String) { let body = if dry { text.replacen("{", "{\"dry_run\":true,", 1) } else { text.clone() }; let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(120))).unwrap(); write!(s, "POST /v1/jobs HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).unwrap(); let mut out = String::new(); let _ = s.read_to_string(&mut out); (out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0), out.split_once("\r\n\r\n").map(|(_, b)| b.to_string()).unwrap_or_default()) }; let (code, dry) = send(true); assert_eq!(code, 200, "{dry}"); let dry: Value = serde_json::from_str(&dry).unwrap(); let (code, sent) = send(false); eprintln!("pad {pad}: dry run accepted {}, submission {code}", dry["accepted"]); assert_eq!(dry["accepted"] == true, code == 202, "pad {pad}: the dry run said {} ({}), the submission {code} {sent}", dry["accepted"], dry["refusal"]); if code == 202 { let id = serde_json::from_str::(&sent).unwrap()["id"].as_str().unwrap().to_string(); let _ = http(port, "DELETE", &format!("/v1/jobs/{id}"), None); } else { assert!(sent.contains("J031") && dry["refusal"]["code"] == "J031", "pad {pad}: {sent} / {dry}"); } } alive(port, &mut svc, "dry runs held to the submission's reading"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A set's bound counts each item's request as it is served: a number RFC /// 8785 writes longer than serde (`1e20`, 21 characters there) is counted at /// its served length, so the result of a sweep sharing thousands of them is /// within its dry run's `result_bytes`. #[test] fn a_set_s_bound_counts_its_items_requests_as_served() { let d = dir("set-numbers"); let (mut svc, port) = start(&d, 10_000); let pads = vec!["1e20"; 20_000].join(","); let text = format!("{{\"qasm\":\"OPENQASM 3.0;\\ninput float theta;\\nqubit q;\\nbit c;\\nrx(theta) q;\\nc = measure q;\\n\",\"sweep\":{{\"theta\":[0.1,0.2,0.3]}},\"shots\":10,\"pad\":[{pads}]}}"); let req: Value = serde_json::from_str(&text).unwrap(); let mut dry = req.clone(); dry["dry_run"] = json!(true); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&dry)); assert_eq!(code, 200, "{b}"); let bound = b["result_bytes"].as_u64().unwrap(); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&req)); assert_eq!(code, 202, "{b}"); let job = ended(port, b["id"].as_str().unwrap(), Duration::from_secs(120)); assert_eq!(job["status"], "done", "{}", job["error"]); let served = wai_quantum::engine::receipt::canonical(&job["result"]).len() as u64; eprintln!("3 points sharing 20,000 numbers: served {served} bytes, bound {bound}"); assert!(served <= bound, "served {served} bytes, past the bound of {bound}"); alive(port, &mut svc, "a set's numbers"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A GHZ state's sweep is bounded by the two outcomes the stabilizer method /// can give, not by a distinct outcome a shot: 100 copies of a 24-qubit GHZ /// state at 30,000 shots each, refused before at a bound of 89 MB for a /// result of about 250 KB, are taken and run, their result within its bound. #[test] fn an_honest_clifford_set_is_taken_by_the_outcomes_it_can_give() { let d = dir("ghz-set"); let (mut svc, port) = start(&d, 10_000); let mut ghz = "OPENQASM 3.0;\ninclude \"stdgates.inc\";\nqubit[24] q;\nbit[24] c;\nh q[0];\n".to_string(); for i in 1..24 { ghz += &format!("cx q[{}], q[{i}];\n", i - 1); } ghz += "c = measure q;\n"; let set = json!({ "programs": vec![ghz; 100], "shots": 30_000, "seed": 3 }); let mut dry = set.clone(); dry["dry_run"] = json!(true); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&dry)); assert_eq!(code, 200, "{b}"); assert!(b["accepted"] == true, "{}", b["refusal"]); let bound = b["result_bytes"].as_u64().unwrap(); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&set)); assert_eq!(code, 202, "{b}"); let job = ended(port, b["id"].as_str().unwrap(), Duration::from_secs(300)); assert_eq!(job["status"], "done", "{}", job["error"]); let served = wai_quantum::engine::receipt::canonical(&job["result"]).len() as u64; eprintln!("100 GHZ24 at 30,000 shots: served {served} bytes, bound {bound}"); assert!(served <= bound && bound < 8 << 20, "served {served}, bound {bound}"); alive(port, &mut svc, "a GHZ set"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A run with its verdict is bounded by what its critique can say, not by a /// remark on every gate: a loop of 124,700 gates with nothing to remark on is /// run by an MCP run tool and by `POST /v1/run` with `"verdict": true` (the /// Studio's request), where both were refused for a 64.7 MiB bound. A job /// past the cap is refused naming its largest part and what makes it /// smaller: a verdict's critique of thousands of gates that cancel, the /// reader's notes, or the counts. #[test] fn a_long_loop_s_verdict_is_taken_and_a_refusal_names_its_part() { let lp = "OPENQASM 3.0;\ninclude \"stdgates.inc\";\nqubit[8] q;\nbit[8] c;\nfor int s in [1:4300] {\n cx q[0], q[1]; rz(0.1) q[1]; cx q[0], q[1];\n cx q[1], q[2]; rz(0.2) q[2]; cx q[1], q[2];\n cx q[2], q[3]; rz(0.3) q[3]; cx q[2], q[3];\n cx q[3], q[4]; rz(0.4) q[4]; cx q[3], q[4];\n cx q[4], q[5]; rz(0.5) q[5]; cx q[4], q[5];\n cx q[5], q[6]; rz(0.6) q[6]; cx q[5], q[6];\n cx q[6], q[7]; rz(0.7) q[7]; cx q[6], q[7];\n rx(0.1) q[0]; rx(0.1) q[1]; rx(0.1) q[2]; rx(0.1) q[3]; rx(0.1) q[4]; rx(0.1) q[5]; rx(0.1) q[6]; rx(0.1) q[7];\n}\nc = measure q;\n"; let d = dir("loop-verdict"); let (mut svc, port) = start(&d, 10_000); let (err, v) = mcp_tool(port, "run_quantum_circuit", &json!({ "qasm": lp, "shots": 1000 })); assert!(!err, "the MCP run: {v}"); // Taken, and run to its end. A host loaded enough to run it past the // answer's time (55 s) answers with the job to poll (`202`), as it answers // any run, and the job is polled to its end: with the whole suite on three // CPUs beside eighteen busy loops it was, where `200` alone was asked. let (code, b) = json_of(port, "POST", "/v1/run", Some(&json!({ "qasm": lp, "shots": 1000, "verdict": true }))); assert!(code == 200 || code == 202, "{code} {b}"); let b = if code == 202 { ended(port, b["id"].as_str().unwrap(), Duration::from_secs(600)) } else { b }; assert!(b["status"] == "done", "{code} {}", b["error"]); svc.kill().unwrap(); svc.wait().unwrap(); // Past a cap of 1 MiB: each refusal names the part that passes it. let (mut svc, port) = start_with(&d, 10_000, &[("WAI_QSVC_RESULT_MIB", "1")]); let cancels = "OPENQASM 3.0;\nqubit q;\nbit c;\nfor int i in [1:10000] { h q; h q; }\nc = measure q;\n"; let noted = format!("OPENQASM 3.0;\nqubit q;\nbit c;\n{}h q;\nc = measure q;\n", (0..7000).map(|i| format!("pragma note{i} {}\n", "x".repeat(120))).collect::()); for (what, job, says) in [ ("a critique of 10,000 pairs", json!({ "qasm": cancels, "shots": 10, "verdict": true }), "its verdict"), ("7,000 notes", json!({ "qasm": noted, "shots": 10 }), "the reader's warnings and notes"), ("a sweep's counts", sweep16(40, 2000), "its counts"), ] { let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&job)); let why = b["error"].as_str().unwrap_or(""); eprintln!("{what}: {code} {why}"); assert!(code == 422 && why.contains("J031") && why.contains(says), "{what}: {code} {why}"); } alive(port, &mut svc, "verdicts bounded by their critique"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// The README's largest sweep, 100 points of 16 measured bits at 30,000 /// shots a point, is taken and ends done at the service's defaults: the set's /// worker keeps each item's result as text as it ends, where it kept every /// item's as a tree and passed the memory it was admitted with (on Linux /// every time, `the job used 693 MiB, over the 258 MiB`). #[test] fn the_readme_s_largest_sweep_is_taken_and_ends_done() { let d = dir("readme-sweep"); let (mut svc, port) = start(&d, 10_000); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep16(100, 30_000))); assert_eq!(code, 202, "{b}"); let job = ended(port, b["id"].as_str().unwrap(), Duration::from_secs(600)); assert_eq!(job["status"], "done", "{}", job["error"]); assert_eq!(job["result"]["items"].as_array().map(Vec::len), Some(100)); alive(port, &mut svc, "the README's sweep"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A client's quick calls are never stuck behind two of its own long checks: /// one client sends two thresholds that each run past the 10 s a request /// waits for a turn, then four critiques. The first threshold holds the /// client's own turn and the second a lent one; the lent turn goes back once /// a critique of its own has waited 7 s, so every critique is answered (the /// second threshold may be refused, `J032`, to send again), where every /// critique waited out its 10 s and was refused. /// /// The thresholds are sized to this host alone and run two at once, so the /// service runs with an error-correction call's time raised ([`LONG_CHECKS`]): /// at the default 25 s, CI's loaded Linux runner refused both for their time /// (`J031`, 28 and 30 s projected, f032d85d3) or the second (27 s, 66359317c; /// 35 s, 71df7608d), and the critiques were answered with nothing held. #[test] fn a_client_s_quick_calls_pass_two_of_its_own_long_checks() { let d = dir("two-long"); let (mut svc, port) = start_with(&d, 1000, &LONG_CHECKS); let me = "198.51.100.91"; let critique = json!({ "qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\n" }); let (threshold, alone) = slow_threshold(port); if alone < 11.0 { eprintln!("the threshold took {alone:.1} s alone, under the 10 s a request waits for a turn: nothing to test on this host"); } let began = Instant::now(); let answers: Vec<(String, u16, String, f64)> = std::thread::scope(|t| { let mut calls = Vec::new(); for (i, (name, path, body, after)) in [("threshold 1", "/v1/qec/threshold", &threshold, 0u64), ("threshold 2", "/v1/qec/threshold", &threshold, 30)].into_iter().chain((0..4).map(|k| ("critique", "/v1/critique", &critique, 4000 + 30 * k))).enumerate() { let began = &began; calls.push(t.spawn(move || { std::thread::sleep(Duration::from_millis(after)); let (code, out) = post_as(port, me, path, body); (format!("{name} #{i}"), code, out.lines().last().unwrap_or("").to_string(), began.elapsed().as_secs_f64()) })); } calls.into_iter().map(|c| c.join().unwrap()).collect() }); eprintln!("{answers:#?}"); for (name, code, out, at) in &answers { if name.starts_with("critique") || name.starts_with("threshold 1") { assert_eq!(*code, 200, "{name} at {at:.2} s: {out}"); } else { assert!(*code == 200 || (*code == 503 && out.contains("lent")), "{name} at {at:.2} s: {code} {out}"); } } alive(port, &mut svc, "two long checks and quick calls"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// Each JSON-RPC id is echoed as it was sent: a fraction or an exponent /// (`1.0`, `1e3`), and an integer past 64 bits, which read as a number and /// written again came back as `1`, `1000` and `1.2345678901234568e+29`. #[test] fn an_mcp_id_is_echoed_as_it_was_sent() { let d = dir("mcp-raw-ids"); let (mut svc, port) = start(&d, 1000); for id in ["1.0", "1e3", "1E3", "-0.0", "1.50", "18446744073709551616", "123456789012345678901234567890", "\"1.0\"", "7"] { for method in ["ping", "tools/list"] { let body = format!("{{\"jsonrpc\":\"2.0\",\"id\":{id},\"method\":\"{method}\"}}"); let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(60))).unwrap(); write!(s, "POST /mcp HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nAccept: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).unwrap(); let mut out = String::new(); let _ = s.read_to_string(&mut out); assert!(out.starts_with("HTTP/1.1 200"), "{id} {method}: {out}"); assert!(out.contains(&format!("\"id\":{id},")) || out.contains(&format!("\"id\":{id}}}")), "{method}: sent the id {id}, answered {}", out.split_once("\r\n\r\n").map(|(_, b)| b).unwrap_or("").chars().take(120).collect::()); } } alive(port, &mut svc, "ids echoed as sent"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// A long poll past its client's holds is not held, and says so: `429`, /// `J032`, `Retry-After` and the job's status, so a client polling in a /// loop waits, where it was answered `200` at once, unchanged, and a client /// that polled again at once spun. Another client's long poll is held. #[test] fn a_long_poll_past_its_client_s_holds_says_when_to_poll_again() { let d = dir("poll-again"); let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1")]); let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&pump())); assert_eq!(code, 202, "{b}"); let id = b["id"].as_str().unwrap().to_string(); let me = "198.51.100.93"; let streams: Vec = (0..4) .map(|_| { let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); write!(s, "GET /v1/jobs/{id}/events HTTP/1.1\r\nHost: x\r\nX-Forwarded-For: {me}\r\n\r\n").unwrap(); s }) .collect(); std::thread::sleep(Duration::from_millis(500)); let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(60))).unwrap(); let t = Instant::now(); write!(s, "GET /v1/jobs/{id}?wait=25 HTTP/1.1\r\nHost: x\r\nX-Forwarded-For: {me}\r\nConnection: close\r\n\r\n").unwrap(); let mut out = String::new(); let _ = s.read_to_string(&mut out); let took = t.elapsed(); let (head, body) = out.split_once("\r\n\r\n").unwrap_or_default(); let body: Value = serde_json::from_str(body).unwrap_or(Value::Null); assert!(head.starts_with("HTTP/1.1 429") && head.to_ascii_lowercase().contains("retry-after: 2"), "a fifth hold: {head}"); assert!(body["code"] == "J032" && (body["status"] == "running" || body["status"] == "queued") && took < Duration::from_secs(5), "{body} after {took:?}"); // Another client's long poll is held. let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap(); s.set_read_timeout(Some(Duration::from_secs(60))).unwrap(); let t = Instant::now(); write!(s, "GET /v1/jobs/{id}?wait=2 HTTP/1.1\r\nHost: x\r\nX-Forwarded-For: 198.51.100.94\r\nConnection: close\r\n\r\n").unwrap(); let mut out = String::new(); let _ = s.read_to_string(&mut out); assert!(out.starts_with("HTTP/1.1 200") && t.elapsed() >= Duration::from_millis(1900), "another client's poll: {:?} {}", t.elapsed(), out.lines().next().unwrap_or("")); drop(streams); let _ = http(port, "DELETE", &format!("/v1/jobs/{id}"), None); alive(port, &mut svc, "polls past the holds"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); } /// What ended jobs keep unread stays under `WAI_QSVC_UNREAD_MIB`: a job is /// taken only while every result not yet read, those of the jobs queued and /// running at their bounds among them, and its own bound stay under the cap. /// Thirty clients each send a job of about 0.8 MiB while one job holds the /// only worker; where each was taken because nothing had ended, so their /// results passed the 2 MiB cap six times over once they ended, those past /// the cap are refused, `503`, and what they all hold unread stays under it. #[test] fn ended_jobs_unread_stay_under_the_unread_cap() { let d = dir("unread-cap"); let env = [("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1"), ("WAI_QSVC_UNREAD_MIB", "2"), ("WAI_QSVC_KEPT_MIB", "1"), ("WAI_QSVC_CLIENT_KEPT_MIB", "1"), ("WAI_QSVC_RESULT_MIB", "1"), ("WAI_QSVC_CLIENT_UNREAD_MIB", "4"), ("WAI_QSVC_MAX_QUEUE", "40")]; let (mut svc, port) = start_with(&d, 1000, &env); let (code, b) = json_as(port, "198.51.100.200", "/v1/jobs", &pump()); assert_eq!(code, 202, "{b}"); let blocker = b["id"].as_str().unwrap().to_string(); let job = json!({ "qasm": "OPENQASM 3.0;\nqubit[15] q;\nbit[15] c;\nh q;\nc = measure q;\n", "shots": 30_000, "seed": 1 }); let mut dry = job.clone(); dry["dry_run"] = json!(true); let (_, b) = json_as(port, "198.51.100.199", "/v1/jobs", &dry); let bound = b["result_bytes"].as_u64().unwrap(); assert!(b["accepted"] == true && bound > 512 << 10 && bound < 1 << 20, "{b}"); let mut taken = Vec::new(); for k in 0..30 { let (code, b) = json_as(port, &format!("198.51.100.{}", k + 1), "/v1/jobs", &job); match code { 202 => taken.push(b["id"].as_str().unwrap().to_string()), 503 => assert!(b["error"].as_str().is_some_and(|w| w.contains("not yet read") && w.contains("J032")), "{b}"), _ => panic!("{code} {b}"), } } let _ = http(port, "DELETE", &format!("/v1/jobs/{blocker}"), None); // Each ended, none read until all have. let t = Instant::now(); loop { let (_, v) = json_of(port, "GET", "/v1", None); if v["load"]["queued"] == 0 && v["load"]["running"] == 0 { break; } assert!(t.elapsed() < Duration::from_secs(120), "{}", v["load"]); std::thread::sleep(Duration::from_millis(200)); } let held: u64 = taken.iter().map(|id| { let (_, j) = json_of(port, "GET", &format!("/v1/jobs/{id}"), None); assert_eq!(j["status"], "done", "{}", j["error"]); http(port, "GET", &format!("/v1/jobs/{id}"), None).2.len() as u64 }).sum(); eprintln!("{} of 30 taken, bound {bound} each; held unread at once: {held} bytes of jobs' reports", taken.len()); assert!(!taken.is_empty() && taken.len() < 30, "{} of 30 taken", taken.len()); assert!(taken.len() as u64 * bound <= 2 << 20, "{} jobs of {bound} bytes taken under a 2 MiB cap", taken.len()); alive(port, &mut svc, "the unread cap"); svc.kill().unwrap(); svc.wait().unwrap(); let _ = std::fs::remove_dir_all(&d); }