diff --git a/REFERENCE.md b/REFERENCE.md index 652dc26..427c3f4 100644 --- a/REFERENCE.md +++ b/REFERENCE.md @@ -37,7 +37,7 @@ An `s3.sign` header given as a list is sent once per value, in order. Its `paylo A read asks for a checksum with `checksum_mode` on S3, or `range_checksum` on an Azure range. It reports the checksum it checked as `checksum`: `{"algorithm", "value_base64"}`, with S3's `type`, such as `full_object`. -A large body is not held but generated: a pattern repeated to a length, `{"encoding": "repeat", "data": {"pattern_base64", "length"}}`. A call names one in `body`, which the adapter streams into its client without holding it. One of at most 1 MiB arrives as `body_base64` instead. A call with `body_result: "fingerprint"` reports the body it read as `body_length` and `body_crc64nvme_base64`, the big-endian CRC64NVME in base64, in place of `body_base64`. The rule `generated_body` checks such a fingerprint at its pointer: at the request's root, where every request has one, or at `/value`. Only `large.json` holds such bodies, and an adapter's time there grows with the bytes a case moves. An `aws-chunked` request also gets `decoded_body_length` and `decoded_body_crc64nvme_base64` for its payload, which `generated_body` checks in its place, and `aws_chunked` with its chunk count and trailers. An exchange with `repeats` answers as many requests as match it. A response with `serves_ranges` answers a ranged GET with that window of its body. +A large body is not held but generated: a pattern repeated to a length, `{"encoding": "repeat", "data": {"pattern_base64", "length"}}`. A call names one in `body`, which the adapter streams into its client without holding it. One of at most 1 MiB arrives as `body_base64` instead. A large read names `body_sink`, a loopback address such as `127.0.0.1:40123`. The adapter connects to it, copies the body it read into the connection, closes it, and reports its result without a body. The grader reads the body to its end and adds its `body_length` and `body_crc64nvme_base64`, the big-endian CRC64NVME in base64, to the result's `value`. The rule `generated_body` checks such a fingerprint at its pointer: at the request's root, where every request has one, or at `/value`. Only `large.json` holds such bodies, and an adapter's time there grows with the bytes a case moves. An `aws-chunked` request also gets `decoded_body_length` and `decoded_body_crc64nvme_base64` for its payload, which `generated_body` checks in its place, and `aws_chunked` with its chunk count and trailers. An exchange with `repeats` answers as many requests as match it. A response with `serves_ranges` answers a ranged GET with that window of its body. `source_account_url` names another Azure storage account that holds a copy source, by its blob service URL. The source is in the container of the endpoint's name, unless `source_container` names another. diff --git a/adapters/borink/Cargo.lock b/adapters/borink/Cargo.lock index b9cf57a..a47d909 100644 --- a/adapters/borink/Cargo.lock +++ b/adapters/borink/Cargo.lock @@ -33,7 +33,7 @@ dependencies = [ [[package]] name = "borink-object-storage-crypto" version = "0.0.3" -source = "git+https://github.com/borink-org/object-storage.git?rev=3439c0d7eafbea8d0a54d8e0e2104baa9201d172#3439c0d7eafbea8d0a54d8e0e2104baa9201d172" +source = "git+https://github.com/borink-org/object-storage.git?rev=55bde1e8a893494efb1c00c01b841fc7d2a7dea6#55bde1e8a893494efb1c00c01b841fc7d2a7dea6" dependencies = [ "borink-object-storage-proto", "hmac", @@ -45,12 +45,12 @@ dependencies = [ [[package]] name = "borink-object-storage-proto" version = "0.0.3" -source = "git+https://github.com/borink-org/object-storage.git?rev=3439c0d7eafbea8d0a54d8e0e2104baa9201d172#3439c0d7eafbea8d0a54d8e0e2104baa9201d172" +source = "git+https://github.com/borink-org/object-storage.git?rev=55bde1e8a893494efb1c00c01b841fc7d2a7dea6#55bde1e8a893494efb1c00c01b841fc7d2a7dea6" [[package]] name = "borink-object-tests-adapter" version = "0.0.0" -source = "git+https://github.com/borink-org/object-storage.git?rev=3439c0d7eafbea8d0a54d8e0e2104baa9201d172#3439c0d7eafbea8d0a54d8e0e2104baa9201d172" +source = "git+https://github.com/borink-org/object-storage.git?rev=55bde1e8a893494efb1c00c01b841fc7d2a7dea6#55bde1e8a893494efb1c00c01b841fc7d2a7dea6" dependencies = [ "base64 0.22.1", "borink-object-storage-crypto", diff --git a/adapters/borink/Cargo.toml b/adapters/borink/Cargo.toml index a20d888..b2d2b3b 100644 --- a/adapters/borink/Cargo.toml +++ b/adapters/borink/Cargo.toml @@ -11,7 +11,7 @@ publish = false [dependencies] # The adapter itself lives in borink-org/object-storage, beside the crates whose # API it calls. This is the commit whose adapter speaks this revision's protocol. -borink-object-tests-adapter = { git = "https://github.com/borink-org/object-storage.git", rev = "3439c0d7eafbea8d0a54d8e0e2104baa9201d172" } +borink-object-tests-adapter = { git = "https://github.com/borink-org/object-storage.git", rev = "55bde1e8a893494efb1c00c01b841fc7d2a7dea6" } [features] # HTTPS, which only the live probes need. Offline grading uses plain HTTP on loopback. diff --git a/src/model.rs b/src/model.rs index 1108543..7e4bc6e 100644 --- a/src/model.rs +++ b/src/model.rs @@ -289,7 +289,7 @@ pub enum Rule { /// A body generated from a pattern: the `body_length` and `body_crc64nvme_base64` at the /// check's pointer must be its fingerprint. At the request's root, it checks the request body; - /// at `/value`, the body a read reported. + /// at `/value`, the body a read wrote to its body sink. GeneratedBody { pattern_base64: String, length: u64, diff --git a/src/runner.rs b/src/runner.rs index 316d31a..1120439 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -1,11 +1,15 @@ //! Runs one adapter process per case and serves the case's loopback endpoint. -use crate::{Result, Session, generated::GeneratedBody, model::*}; +use crate::{ + Result, Session, + generated::{Fingerprint, FingerprintStream, GeneratedBody}, + model::*, +}; use base64::{Engine, engine::general_purpose::STANDARD}; use serde_json::{Value, json}; use std::{ collections::BTreeMap, io::{Read, Write}, - net::TcpListener, + net::{SocketAddr, TcpListener, TcpStream}, process::{Command, Stdio}, sync::{ Mutex, @@ -29,9 +33,15 @@ const INLINE_GENERATED_BYTES: u64 = 1 << 20; const ADAPTER_SECONDS: u64 = 30; const GENERATED_BYTES_PER_SECOND: u64 = 64 << 20; -/// Returns the call an adapter receives: a small generated body comes as its bytes. -fn adapter_call(call: &Value) -> Result { +/// Returns the call an adapter receives: a small generated body comes as its bytes, and a read +/// with `body_result: "fingerprint"` names the `body_sink` that its body goes to. +fn adapter_call(call: &Value, body_sink: Option) -> Result { let mut call = call.clone(); + if let Some(address) = body_sink { + let object = call.as_object_mut().ok_or("call")?; + object.remove("body_result"); + object.insert("body_sink".into(), json!(address.to_string())); + } if call["body"]["encoding"] == "repeat" { let generated: GeneratedBody = serde_json::from_value(call["body"]["data"].clone())?; if generated.length <= INLINE_GENERATED_BYTES { @@ -69,6 +79,49 @@ fn generated_byte_count(case: &Case) -> u64 { call_bytes + exchange_bytes } +/// Reads the bodies that an adapter writes to its body sink, one connection each, until the +/// adapter is done. Returns each connection's peer and the fingerprint of what it sent. +fn read_body_sink( + listener: &TcpListener, + stop: &AtomicBool, + timeout: Duration, +) -> Vec<(SocketAddr, std::io::Result)> { + let mut bodies = Vec::new(); + let mut buffer = vec![0; 1 << 20]; + while let Ok((mut stream, peer)) = listener.accept() { + let mut fingerprint = FingerprintStream::default(); + let read = stream.set_read_timeout(Some(timeout)).and_then(|()| { + loop { + match stream.read(&mut buffer)? { + 0 => return Ok(fingerprint.finish()), + count => fingerprint.update(&buffer[..count]), + } + } + }); + bodies.push((peer, read)); + if stop.load(Ordering::Acquire) { + break; + } + } + bodies +} + +/// Puts the fingerprint of the body that an adapter wrote to its body sink into its result's +/// `value`. A result that is not `ok`, or a sink that got no body or more than one, gets none, so +/// a `generated_body` check there fails. +fn insert_body_sink_fingerprint(result: &mut Value, bodies: Vec>) { + if result["outcome"] != "ok" { + return; + } + let [Ok(fingerprint)] = bodies.as_slice() else { + return; + }; + if !result["value"].is_object() { + result["value"] = json!({}); + } + fingerprint.insert_into(&mut result["value"]); +} + // The proxy variables that HTTP clients read, such as ureq and the AWS CRT. const PROXY_VARIABLES: &[&str] = &[ "ALL_PROXY", @@ -199,18 +252,31 @@ pub fn grade_case( } else { None }; + // A large read goes to a loopback socket, which the grader fingerprints as it arrives. + let body_sink = if case.call["body_result"] == "fingerprint" { + Some(TcpListener::bind("127.0.0.1:0")?) + } else { + None + }; + let body_sink_address = body_sink + .as_ref() + .map(TcpListener::local_addr) + .transpose()?; let message = json!({ "version": 1, "mode": if live_endpoint.is_some() {"live"} else {"offline"}, "provider": profile.provider, "endpoint": endpoint, - "call": adapter_call(&case.call)?, + "call": adapter_call(&case.call, body_sink_address)?, }); let adapter_time = Duration::from_secs( ADAPTER_SECONDS + generated_byte_count(case) / GENERATED_BYTES_PER_SECOND, ); let stop = AtomicBool::new(false); - let result = thread::scope(|scope| { + let (result, sink_bodies) = thread::scope(|scope| { + let sink_reader = body_sink + .as_ref() + .map(|listener| scope.spawn(|| read_body_sink(listener, &stop, adapter_time))); if let Some(server) = &server { for _ in 0..HTTP_WORKERS { scope.spawn(|| { @@ -243,14 +309,31 @@ pub fn grade_case( && let Ok(address) = server.local_addr() { for _ in 0..HTTP_WORKERS { - let _ = std::net::TcpStream::connect(address); + let _ = TcpStream::connect(address); } } - result + // The sink's reader blocks in accept too. The connection that wakes it is not a body. + let mut sink_bodies = Vec::new(); + if let (Some(reader), Some(address)) = (sink_reader, body_sink_address) { + let waker = TcpStream::connect(address) + .and_then(|stream| stream.local_addr()) + .ok(); + for (peer, body) in reader.join().unwrap() { + if Some(peer) != waker { + sink_bodies.push(body); + } + } + } + (result, sink_bodies) }); let session = session.into_inner().unwrap(); Ok(match result { - Ok(result) => session.finish(&result, SystemTime::now()), + Ok(mut result) => { + if body_sink_address.is_some() { + insert_body_sink_fingerprint(&mut result, sink_bodies); + } + session.finish(&result, SystemTime::now()) + } Err(error) => { json!({ "id": case.id, diff --git a/tests/process.rs b/tests/process.rs index 7bc0b4c..8b7273f 100644 --- a/tests/process.rs +++ b/tests/process.rs @@ -92,7 +92,7 @@ fn generated_round_trip_case(length: u64) -> (Case, Profile) { "lane": "core", "purpose": "Send and read back a generated body", "sources": ["tests/process.rs"], - "call": {"op": "put", "key": "object", "body": {"encoding": "repeat", "data": generated}}, + "call": {"op": "put", "key": "object", "body": {"encoding": "repeat", "data": generated}, "body_result": "fingerprint"}, "exchanges": [ {"alternatives": [{ "request": [ @@ -152,6 +152,15 @@ fn generated_bodies_cross_the_grader_without_being_held() { .contains("not the generated"), "{report}" ); + + let report = + runner::grade_case(&case, &profile, &adapter_command("generated-no-sink"), None) + .unwrap(); + assert_eq!( + report["verdict"], + json!("wrong"), + "{length} bytes: {report}" + ); } } diff --git a/tests/support/adapter.rs b/tests/support/adapter.rs index 479dd80..34ad802 100644 --- a/tests/support/adapter.rs +++ b/tests/support/adapter.rs @@ -1,6 +1,6 @@ //! Native process fixture for the grader tests. use base64::{Engine, engine::general_purpose::STANDARD}; -use object_tests::generated::{FingerprintStream, GeneratedBody}; +use object_tests::generated::GeneratedBody; use serde_json::{Value, json}; use std::{ io::{self, BufRead, BufReader, Read, Write}, @@ -34,8 +34,8 @@ fn read_response_head(reader: &mut impl BufRead) -> (u16, u64, Option) { } } -/// Sends a GET and feeds the body it gets into `fingerprint`; returns the object's total size. -fn read_into(address: &str, range: Option<(u64, u64)>, fingerprint: &mut FingerprintStream) -> u64 { +/// Sends a GET and copies the body it gets into `sink`; returns the object's total size. +fn read_into(address: &str, range: Option<(u64, u64)>, sink: &mut impl Write) -> u64 { let mut stream = TcpStream::connect(address).unwrap(); let range_header = range.map_or(String::new(), |(start, end)| { format!("Range: bytes={start}-{end}\r\n") @@ -51,7 +51,7 @@ fn read_into(address: &str, range: Option<(u64, u64)>, fingerprint: &mut Fingerp while remaining > 0 { let buffer = reader.fill_buf().unwrap(); let take = buffer.len().min(remaining as usize); - fingerprint.update(&buffer[..take]); + sink.write_all(&buffer[..take]).unwrap(); reader.consume(take); remaining -= take as u64; } @@ -68,10 +68,12 @@ enum Transfer { Ranged, /// Sent `aws-chunked`, in chunks of 1 MiB with a trailing checksum. AwsChunked, + /// Read back, but not written to the body sink. + NoSink, } -/// PUTs the call's body, generated or held, then GETs the object back and reports the -/// fingerprint of what it read. +/// PUTs the call's body, generated or held, then GETs the object back and writes what it read to +/// the call's `body_sink`. fn put_and_get_generated(input: &Value, transfer: Transfer) -> Value { let corrupt = transfer == Transfer::Corrupt; let address = input["endpoint"]["url"] @@ -143,25 +145,24 @@ fn put_and_get_generated(input: &Value, transfer: Transfer) -> Value { return json!({"outcome": "error", "kind": "other", "status": status}); } - let mut fingerprint = FingerprintStream::default(); + let mut sink: Box = if transfer == Transfer::NoSink { + Box::new(io::sink()) + } else { + Box::new(TcpStream::connect(call["body_sink"].as_str().unwrap()).unwrap()) + }; if transfer == Transfer::Ranged { let window = 7 << 20; - let total = read_into(&address, Some((0, window - 1)), &mut fingerprint); + let total = read_into(&address, Some((0, window - 1)), &mut sink); let mut start = window; while start < total { - read_into( - &address, - Some((start, start + window - 1)), - &mut fingerprint, - ); + read_into(&address, Some((start, start + window - 1)), &mut sink); start += window; } } else { - read_into(&address, None, &mut fingerprint); + read_into(&address, None, &mut sink); } - let mut value = json!({}); - fingerprint.finish().insert_into(&mut value); - json!({"outcome": "ok", "value": value}) + drop(sink); + json!({"outcome": "ok", "value": {}}) } fn perform_http_request(input: &Value) -> Value { @@ -244,6 +245,7 @@ fn main() -> ExitCode { "generated" => println!("{}", put_and_get_generated(&input, Transfer::Plain)), "generated-corrupt" => println!("{}", put_and_get_generated(&input, Transfer::Corrupt)), "generated-ranged" => println!("{}", put_and_get_generated(&input, Transfer::Ranged)), + "generated-no-sink" => println!("{}", put_and_get_generated(&input, Transfer::NoSink)), "generated-aws-chunked" => { println!("{}", put_and_get_generated(&input, Transfer::AwsChunked)) }