Have adapters write a large read to a grader socket, not a checksum.

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.
 

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.

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,

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<Value> {
+/// 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<SocketAddr>) -> Result<Value> {
     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<Fingerprint>)> {
+    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<std::io::Result<Fingerprint>>) {
+    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,

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}"
+        );
     }
 }
 

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<u64>) {
     }
 }
 
-/// 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<dyn Write> = 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))
         }