# HG changeset patch # User Georges Racinet <georges.racinet@octobus.net> # Date 1690628939 -7200 # Sat Jul 29 13:08:59 2023 +0200 # Branch oldstable # Node ID 03355820dccfd6fbcc2f00b96e9a9475bc759afd # Parent 7e37f1565011491bad0a249697d957a39db21ba1 RHGitaly: BlobService.GetBlob This builds on the previously introduced repository file extraction and conversion to `GetBlobResponse`. It is important to emit at least one response, even if the resulting data is empty (usually because of `limit=0`). diff --git a/rust/rhgitaly/proto/blob.proto b/rust/rhgitaly/proto/blob.proto --- a/rust/rhgitaly/proto/blob.proto +++ b/rust/rhgitaly/proto/blob.proto @@ -7,6 +7,21 @@ option go_package = "gitlab.com/gitlab-org/gitaly/v15/proto/go/gitalypb"; +// BlobService is a service which provides RPCs to retrieve Git blobs from a +// specific repository. +service BlobService { + + // GetBlob returns the contents of a blob object referenced by its object + // ID. We use a stream to return a chunked arbitrarily large binary + // response + rpc GetBlob(GetBlobRequest) returns (stream GetBlobResponse) { + option (op_type) = { + op: ACCESSOR + }; + } +} + + // This comment is left unintentionally blank. message GetBlobRequest { // This comment is left unintentionally blank. diff --git a/rust/rhgitaly/src/main.rs b/rust/rhgitaly/src/main.rs --- a/rust/rhgitaly/src/main.rs +++ b/rust/rhgitaly/src/main.rs @@ -13,6 +13,7 @@ use tonic::transport::Server; use rhgitaly::config::Config; +use rhgitaly::service::blob::blob_server; use rhgitaly::service::commit::commit_server; use rhgitaly::service::r#ref::ref_server; use rhgitaly::service::repository::repository_server; @@ -63,6 +64,7 @@ let config = Arc::new(Config::from_env()); let server = Server::builder() .add_service(server_server()) + .add_service(blob_server(&config)) .add_service(commit_server(&config)) .add_service(ref_server(&config)) .add_service(repository_server(&config)); diff --git a/rust/rhgitaly/src/service/blob.rs b/rust/rhgitaly/src/service/blob.rs new file mode 100644 --- /dev/null +++ b/rust/rhgitaly/src/service/blob.rs @@ -0,0 +1,116 @@ +// Copyright 2023 Georges Racinet <georges.racinet@octobus.net> +// +// This software may be used and distributed according to the terms of the +// GNU General Public License version 2 or any later version. +// SPDX-License-Identifier: GPL-2.0-or-later +use std::fmt::Debug; +use std::sync::Arc; +use tonic::{metadata::Ascii, metadata::MetadataValue, Request, Response, Status}; + +use hg::revlog::RevlogError; + +use crate::config::Config; +use crate::gitaly::blob_service_server::{BlobService, BlobServiceServer}; +use crate::gitaly::{GetBlobRequest, GetBlobResponse, Repository}; +use crate::mercurial::{lookup_blob, ObjectMetadata}; +use crate::message::BlobResponseChunk; +use crate::metadata::correlation_id; +use crate::oid::extract_blob_oid; +use crate::repository::{default_repo_spec_error_status, load_repo_and_stream, RequestWithRepo}; +use crate::streaming::{BlockingResponseSender, ResultResponseStream, WRITE_BUFFER_SIZE}; + +use tokio_stream::wrappers::ReceiverStream; +use tracing::{info, instrument}; + +impl RequestWithRepo for GetBlobRequest { + fn repository_ref(&self) -> Option<&Repository> { + self.repository.as_ref() + } +} + +#[derive(Debug)] +pub struct BlobServiceImpl { + config: Arc<Config>, +} + +#[tonic::async_trait] +impl BlobService for BlobServiceImpl { + type GetBlobStream = ReceiverStream<Result<GetBlobResponse, Status>>; + + async fn get_blob( + &self, + request: Request<GetBlobRequest>, + ) -> Result<Response<Self::GetBlobStream>, Status> { + let (metadata, _ext, inner) = request.into_parts(); + + self.inner_get_blob(inner, correlation_id(&metadata)).await + } +} + +impl BlobServiceImpl { + #[instrument(name = "get_blob", skip(self))] + async fn inner_get_blob( + &self, + request: GetBlobRequest, + correlation_id: Option<&MetadataValue<Ascii>>, + ) -> ResultResponseStream<GetBlobResponse> { + info!("Processing"); + if request.oid.is_empty() { + return Err(Status::invalid_argument("empty Oid")); + } + + let (node, path) = extract_blob_oid(&request.oid).map_err(|e| { + Status::invalid_argument(format!("Error parsing Oid {:?}: {:?}", request.oid, e)) + })?; + + load_repo_and_stream( + self.config.clone(), + request, + default_repo_spec_error_status, + move |req, repo, tx| match lookup_blob(&repo, node.into(), &path) { + Ok(Some((mut data, metadata))) => { + if req.limit >= 0 { + data.truncate(req.limit as usize); + } + stream_blob(&tx, data, metadata); + } + Ok(None) => { + tx.send(Ok(GetBlobResponse::default())); + } + Err(RevlogError::InvalidRevision) => { + tx.send(Ok(GetBlobResponse::default())); + } + Err(e) => { + tx.send(Err(Status::internal(format!( + "Error looking up blob: {:?}", + e + )))); + } + }, + ) + } +} + +fn stream_blob<R: BlobResponseChunk>( + tx: &BlockingResponseSender<R>, + data: Vec<u8>, + metadata: ObjectMetadata, +) { + let mut chunks = data.chunks(*WRITE_BUFFER_SIZE); + // In case `data` is empty, the iterator will yield `None` immediately, + // but we still need to send a response with empty data and the metadata, + // as it is a common pattern for clients to use `limit=0` just to get metadata. + // One would hope not to have to retrieve the entire content in this case, but + // sadly this is needed to return the correct size. + tx.send(Ok(R::with_metadata(chunks.next().unwrap_or(&[]), metadata))); + for chunk in chunks { + tx.send(Ok(R::only_data(chunk))); + } +} + +/// Takes care of boilerplate that would instead be in the startup sequence. +pub fn blob_server(config: &Arc<Config>) -> BlobServiceServer<BlobServiceImpl> { + BlobServiceServer::new(BlobServiceImpl { + config: config.clone(), + }) +} diff --git a/rust/rhgitaly/src/service/mod.rs b/rust/rhgitaly/src/service/mod.rs --- a/rust/rhgitaly/src/service/mod.rs +++ b/rust/rhgitaly/src/service/mod.rs @@ -3,6 +3,7 @@ // This software may be used and distributed according to the terms of the // GNU General Public License version 2 or any later version. // SPDX-License-Identifier: GPL-2.0-or-later +pub mod blob; pub mod commit; pub mod r#ref; pub mod repository;