# HG changeset patch # User Georges Racinet <georges.racinet@octobus.net> # Date 1681378712 -7200 # Thu Apr 13 11:38:32 2023 +0200 # Node ID 33189bb59352a0b658cdbdb36917911262cdd399 # Parent a273ee467f4927a582d4e64af0f7ec34fc072146 RHGitaly: helpers for server streaming gRPC methods involving repositories All future implementations of streaming methods with repositories will follow the same pattern, hence we provide a generic helper right away, taking care notably of repetitive error treatment (sending errors etc.). Detailed explanations of the constraints and how it will work are provided in the doc-comment of the `load_repo_and_stream` function. Concrete gRPC method implementations will consist mostly of calling it with the closure doing the actual work on the loaded repository. diff --git a/rust/rhgitaly/src/lib.rs b/rust/rhgitaly/src/lib.rs --- a/rust/rhgitaly/src/lib.rs +++ b/rust/rhgitaly/src/lib.rs @@ -18,6 +18,7 @@ pub mod config; pub mod repository; pub mod service; +pub mod streaming; pub use config::*; pub use gitaly::*; diff --git a/rust/rhgitaly/src/repository.rs b/rust/rhgitaly/src/repository.rs --- a/rust/rhgitaly/src/repository.rs +++ b/rust/rhgitaly/src/repository.rs @@ -3,14 +3,21 @@ // 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; +use std::marker::Send; use std::path::PathBuf; +use std::sync::Arc; -use tonic::Status; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; +use tonic::{Response, Status}; -use hg::repo::RepoError; +use hg::config::Config as CoreConfig; +use hg::repo::{Repo, RepoError}; use super::config::Config; use super::gitaly::Repository; +use super::streaming::BlockingResponseSender; /// Represent errors that are due to a wrong repository specification. /// @@ -78,6 +85,67 @@ Ok(res) } +pub fn load_repo(config: &Config, opt_repo: Option<&Repository>) -> Result<Repo, RepoLoadError> { + let repo = opt_repo.ok_or(RepoSpecError::MissingSpecification)?; + // TODO load core config once and for all and store it in our Config. + // but do we really want to have all this unrelated stuff around? + let core_config = + CoreConfig::load_non_repo().expect("Should have been able to read Mercurial config"); + // TODO better to use Repo::new_at_path, but it is private + // (Repo::find does verifications that are useless to us. + // At least it does not try to climb up when passed an explicit path) + Ok(Repo::find(&core_config, Some(repo_path(config, repo)?))?) +} + +/// Trait for requests with a repository field +/// +/// It provides the needed uniformity for methods such as [`load_repo_and_stream`] +pub trait RequestWithRepo: Send + 'static { + /// Grab a reference to the [`Repository`] field from the request. + /// + /// Like all submessages, the repository is optional. + fn repository_ref(&self) -> Option<&Repository>; +} + +/// Load a repository and initiate streaming responses +/// +/// This setups the `mpsc` channel expected by Tonic and spawns a blocking task (typically run +/// in a separate thread) loads the repository, and passes over the repository and the transmitting +/// end of the channel to the caller supplied closure. +/// +/// The `and_then` closure must perform its streaming by sending `Result<Resp, Status>` values +/// on the channel, using the provided [`BlockingResponseSender`]. +/// +/// If the repository loading fails, the appropriate gRPC error response is sent over +/// or logged if sending is not possible. +/// +/// Because Gitaly's error responses are not uniform, and we want to match them closely, +/// ethe caller has to supply a callable for conversion of [`RepoSpecError`] to the appropriate +/// [`Status`]. The [`default_repo_spec_error_status`] function can be used in the majority of +/// cases and serve as an example for other cases. +pub fn load_repo_and_stream<Req: RequestWithRepo, Resp: fmt::Debug + Send + 'static>( + config: Arc<Config>, + request: Req, + repo_spec_error_status: impl Fn(RepoSpecError) -> Status + Send + 'static, + and_then: impl FnOnce(Req, Repo, BlockingResponseSender<Resp>) + Send + 'static, +) -> Result<Response<ReceiverStream<Result<Resp, Status>>>, Status> { + // no point having channel capacity for several messages, since `blocking_send` is the only + // way to use it. + let (tx, rx) = mpsc::channel(1); + tokio::task::spawn_blocking(move || { + let btx: BlockingResponseSender<Resp> = tx.into(); + match load_repo(&config, request.repository_ref()) { + Err(RepoLoadError::SpecError(e)) => btx.send(Err(repo_spec_error_status(e))), + Err(RepoLoadError::LoadError(e)) => btx.send(Err(Status::internal(format!( + "Error loading repository: {:?}", + e + )))), + Ok(repo) => and_then(request, repo, btx), + } + }); + Ok(Response::new(ReceiverStream::new(rx))) +} + #[cfg(test)] mod tests { diff --git a/rust/rhgitaly/src/streaming.rs b/rust/rhgitaly/src/streaming.rs new file mode 100644 --- /dev/null +++ b/rust/rhgitaly/src/streaming.rs @@ -0,0 +1,41 @@ +// 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 tokio::sync::mpsc::Sender; +use tonic::Status; +use tracing::error; + +/// A specialization of [`Sender`] allowing only blocking sends and logging sending errors +/// +/// Server streaming methods involving a repository typically spawn a new thread with +/// [`tokio::task::spawn_blocking`] because the primitives in `hg-core` are blocking. From this +/// thread, the only possible output is to use the [`Sender`] end of a multiple-producer, +/// single-consumer (mpsc) channel, with its `blocking_send` method. +/// +/// This is currently the recommended way to send from synchronous to asynchronous code. +/// The only error condition to handle on these sends happens if the receiving side (which takes +/// care of sending the response to the client over the wire) is closing down or closed (e.g. for +/// loss of connectivity?). Since the channel is the only way for the task to report back +/// anything to the client, the only thing that can be done about such errors is to log them. +/// +/// This struct also has the advantage to enforce that code from Tokio blocking tasks should never +/// use the (async) `send` method of the underlying mpsc Sender. +#[derive(Debug, derive_more::From)] +pub struct BlockingResponseSender<Resp>(Sender<Result<Resp, Status>>); + +impl<Resp> BlockingResponseSender<Resp> { + pub fn send(&self, res: Result<Resp, Status>) { + self.0.blocking_send(res).unwrap_or_else(|send_error| { + let msg = match send_error.0 { + Ok(_) => "an ordinary response".into(), + Err(status) => format!("an error response: {}", status), + }; + // logging at error level because when we implement graceful + // shutdown, we will try and avoid to write on channels that are + // closing down. + error!("Channel closing down or closed while streaming {}", msg) + }); + } +}