nordisk/holger

server/lib/tests/a_grpc_upload_streams_into_the_backend.rs

//! stinky 14: the gRPC write door streams. `PutArtifact` carries the whole body
//! in one message; `PutArtifactStream` carries it in pieces, and the server
//! hands the pieces to the backend's `write_from_reader` as they arrive (a
//! backend with `Caps::streaming_uploads` writes them to disk as it reads, any
//! other collects them for its `put`). Over a real TCP gRPC port, with the
//! generated client.

use std::io::Write;

use holger_server_lib::grpc::holger_proto::repository_service_client::RepositoryServiceClient;
use holger_server_lib::grpc::holger_proto::{ArtifactId, FetchArtifactRequest, PutArtifactChunk};
use holger_server_lib::{read_ron_config, wire_holger, Holger};
use traits::test_support::hold_port;

fn config(data: &str, grpc: u16, http: u16, max_body: usize) -> String {
    format!(
        r#"(
    repositories: [
        (
            ron_name: "raw",
            ron_repo_type: "raw",
            ron_upstreams: [],
            ron_in: None,
            ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "default")),
            ron_data_dir: Some("{data}"),
        ),
    ],
    exposed_endpoints: [
        (ron_name: "default", ron_url: "127.0.0.1:{grpc}", ron_http_url: Some("127.0.0.1:{http}"), ron_max_body_bytes: Some({max_body})),
    ],
    storage_endpoints: [
        (ron_name: "storage", ron_storage_type: "file", ron_path: "{data}"),
    ],
)"#
    )
}

struct Running {
    data: tempfile::TempDir,
    _cfg: tempfile::NamedTempFile,
    grpc: String,
}

fn start(max_body: usize) -> Running {
    let (grpc, http) = (hold_port(), hold_port());
    let data = tempfile::tempdir().unwrap();
    let ron = config(&data.path().display().to_string(), grpc, http, max_body);
    let mut cfg = tempfile::NamedTempFile::new().unwrap();
    cfg.write_all(ron.as_bytes()).unwrap();
    let mut holger: Holger = read_ron_config(cfg.path()).unwrap();
    holger.instantiate_backends().unwrap();
    wire_holger(&mut holger).unwrap();
    holger.start().unwrap();
    Running {
        data,
        _cfg: cfg,
        grpc: format!("http://127.0.0.1:{grpc}"),
    }
}

async fn dial(url: &str) -> RepositoryServiceClient<tonic::transport::Channel> {
    for _ in 0..100 {
        if let Ok(c) = RepositoryServiceClient::connect(url.to_string()).await {
            return c;
        }
        tokio::time::sleep(std::time::Duration::from_millis(20)).await;
    }
    panic!("gRPC never came up at {url}");
}

fn id(name: &str) -> ArtifactId {
    ArtifactId {
        namespace: String::new(),
        name: name.into(),
        version: String::new(),
    }
}

/// The first piece names the artifact; the rest carry bytes only.
fn pieces(name: &str, body: &[u8], piece: usize) -> Vec<PutArtifactChunk> {
    body.chunks(piece)
        .enumerate()
        .map(|(i, c)| PutArtifactChunk {
            repository: if i == 0 { "raw".into() } else { String::new() },
            id: (i == 0).then(|| id(name)),
            data: c.to_vec().into(),
            async_job: false,
        })
        .collect()
}

#[tokio::test(flavor = "multi_thread")]
async fn a_streamed_grpc_upload_lands_whole_and_reads_back_byte_for_byte() {
    let run = start(64 * 1024 * 1024);
    let mut client = dial(&run.grpc).await;
    let body: Vec<u8> = (0..(5 * 1024 * 1024 + 17))
        .map(|i| (i % 251) as u8)
        .collect();
    let reply = client
        .put_artifact_stream(tokio_stream::iter(pieces("big.bin", &body, 256 * 1024)))
        .await
        .expect("the streamed upload is accepted")
        .into_inner();
    assert!(reply.success, "{reply:?}");
    assert_eq!(
        std::fs::read(run.data.path().join("big.bin")).unwrap(),
        body,
        "on disk, whole"
    );
    // Read back through the streamed read (`StreamArtifact`): the unary
    // `FetchArtifact` is the whole-buffer convenience and a 5 MiB body is over a
    // stock client's 4 MiB message limit, which is the point of streaming.
    let mut stream = client
        .stream_artifact(FetchArtifactRequest {
            repository: "raw".into(),
            id: Some(id("big.bin")),
        })
        .await
        .expect("read back")
        .into_inner();
    let mut back = Vec::new();
    while let Some(piece) = stream.message().await.expect("a piece") {
        back.extend_from_slice(&piece.data);
    }
    assert_eq!(back, body, "and served back byte for byte");
}

#[tokio::test(flavor = "multi_thread")]
async fn a_streamed_upload_over_the_body_cap_is_refused_and_leaves_nothing() {
    let run = start(1024 * 1024);
    let mut client = dial(&run.grpc).await;
    let body = vec![7u8; 3 * 1024 * 1024];
    let err = client
        .put_artifact_stream(tokio_stream::iter(pieces("huge.bin", &body, 128 * 1024)))
        .await
        .expect_err("over the cap");
    assert_eq!(err.code(), tonic::Code::ResourceExhausted, "{err:?}");
    assert!(
        !run.data.path().join("huge.bin").exists(),
        "a refused stream leaves no file behind"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn a_stream_that_names_no_artifact_is_refused_by_name() {
    let run = start(1024 * 1024);
    let mut client = dial(&run.grpc).await;
    let nameless = vec![PutArtifactChunk {
        repository: "raw".into(),
        id: None,
        data: b"x".to_vec().into(),
        async_job: false,
    }];
    let err = client
        .put_artifact_stream(tokio_stream::iter(nameless))
        .await
        .expect_err("no id");
    assert_eq!(err.code(), tonic::Code::InvalidArgument, "{err:?}");
}