Skip to content
Snippets Groups Projects
Commit 33189bb5 authored by Georges Racinet's avatar Georges Racinet
Browse files

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.
parent a273ee46
No related branches found
No related tags found
2 merge requests!151heptapod#743: making 0.36 the new stable,!144RHGitaly: implement CommitService.ListCommitsByOid
......@@ -18,6 +18,7 @@
pub mod config;
pub mod repository;
pub mod service;
pub mod streaming;
pub use config::*;
pub use gitaly::*;
......@@ -3,4 +3,6 @@
// 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;
......@@ -6,2 +8,3 @@
use std::path::PathBuf;
use std::sync::Arc;
......@@ -7,3 +10,5 @@
use tonic::Status;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::{Response, Status};
......@@ -9,5 +14,6 @@
use hg::repo::RepoError;
use hg::config::Config as CoreConfig;
use hg::repo::{Repo, RepoError};
use super::config::Config;
use super::gitaly::Repository;
......@@ -11,6 +17,7 @@
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 {
......
// 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)
});
}
}
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment