snix_castore/proto/
grpc_blobservice_wrapper.rs

1use crate::{B3Digest, blobservice::BlobService};
2use core::pin::pin;
3use futures::{TryFutureExt, stream::BoxStream};
4use std::{
5    collections::VecDeque,
6    ops::{Deref, DerefMut},
7};
8use tokio_stream::StreamExt;
9use tokio_util::io::ReaderStream;
10use tonic::{Request, Response, Status, Streaming, async_trait};
11use tracing::{Span, instrument, warn};
12
13pub struct GRPCBlobServiceWrapper<T> {
14    blob_service: T,
15}
16
17impl<T> GRPCBlobServiceWrapper<T> {
18    pub fn new(blob_service: T) -> Self {
19        Self { blob_service }
20    }
21}
22
23// This is necessary because bytes::BytesMut comes up with
24// a default 64 bytes capacity that cannot be changed
25// easily if you assume a bytes::BufMut trait implementation
26// Therefore, we override the Default implementation here
27// TODO(raitobezarius?): upstream me properly
28struct BytesMutWithDefaultCapacity<const N: usize> {
29    inner: bytes::BytesMut,
30}
31
32impl<const N: usize> Deref for BytesMutWithDefaultCapacity<N> {
33    type Target = bytes::BytesMut;
34    fn deref(&self) -> &Self::Target {
35        &self.inner
36    }
37}
38
39impl<const N: usize> DerefMut for BytesMutWithDefaultCapacity<N> {
40    fn deref_mut(&mut self) -> &mut Self::Target {
41        &mut self.inner
42    }
43}
44
45impl<const N: usize> Default for BytesMutWithDefaultCapacity<N> {
46    fn default() -> Self {
47        BytesMutWithDefaultCapacity {
48            inner: bytes::BytesMut::with_capacity(N),
49        }
50    }
51}
52
53impl<const N: usize> bytes::Buf for BytesMutWithDefaultCapacity<N> {
54    fn remaining(&self) -> usize {
55        self.inner.remaining()
56    }
57
58    fn chunk(&self) -> &[u8] {
59        self.inner.chunk()
60    }
61
62    fn advance(&mut self, cnt: usize) {
63        self.inner.advance(cnt);
64    }
65}
66
67unsafe impl<const N: usize> bytes::BufMut for BytesMutWithDefaultCapacity<N> {
68    fn remaining_mut(&self) -> usize {
69        self.inner.remaining_mut()
70    }
71
72    unsafe fn advance_mut(&mut self, cnt: usize) {
73        unsafe {
74            self.inner.advance_mut(cnt);
75        }
76    }
77
78    fn chunk_mut(&mut self) -> &mut bytes::buf::UninitSlice {
79        self.inner.chunk_mut()
80    }
81}
82
83#[async_trait]
84impl<T> super::blob_service_server::BlobService for GRPCBlobServiceWrapper<T>
85where
86    T: Deref<Target = dyn BlobService> + Send + Sync + 'static,
87{
88    // https://github.com/tokio-rs/tokio/issues/2723#issuecomment-1534723933
89    type ReadStream = BoxStream<'static, Result<super::BlobChunk, Status>>;
90
91    #[instrument(skip_all)]
92    async fn stat(
93        &self,
94        request: Request<super::StatBlobRequest>,
95    ) -> Result<Response<super::StatBlobResponse>, Status> {
96        let rq = request.into_inner();
97        let req_digest: B3Digest = rq
98            .digest
99            .try_into()
100            .map_err(|_e| Status::invalid_argument("invalid digest length"))?;
101
102        let span = Span::current();
103        span.record("blob.digest", req_digest.to_string());
104
105        match self.blob_service.chunks(&req_digest).await {
106            Ok(None) => Err(Status::not_found(format!("blob {} not found", &req_digest))),
107            Ok(Some(chunk_metas)) => Ok(Response::new(super::StatBlobResponse {
108                chunks: chunk_metas,
109                ..Default::default()
110            })),
111            Err(e) => {
112                warn!(err=%e, "failed to request chunks");
113                Err(e.into())
114            }
115        }
116    }
117
118    #[instrument(skip_all)]
119    async fn read(
120        &self,
121        request: Request<super::ReadBlobRequest>,
122    ) -> Result<Response<Self::ReadStream>, Status> {
123        let rq = request.into_inner();
124
125        let req_digest: B3Digest = rq
126            .digest
127            .try_into()
128            .map_err(|_e| Status::invalid_argument("invalid digest length"))?;
129
130        let span = Span::current();
131        span.record("blob.digest", req_digest.to_string());
132
133        match self.blob_service.open_read(&req_digest).await {
134            Ok(Some(r)) => {
135                let chunks_stream =
136                    ReaderStream::new(r).map(|chunk| Ok(super::BlobChunk { data: chunk? }));
137                Ok(Response::new(Box::pin(chunks_stream)))
138            }
139            Ok(None) => Err(Status::not_found(format!("blob {} not found", &req_digest))),
140            Err(e) => {
141                warn!(err=%e, "failed to call open_read");
142                Err(e.into())
143            }
144        }
145    }
146
147    #[instrument(skip_all)]
148    async fn put(
149        &self,
150        request: Request<Streaming<super::BlobChunk>>,
151    ) -> Result<Response<super::PutBlobResponse>, Status> {
152        let req_inner = request.into_inner();
153
154        let data_stream = req_inner.map(|x| {
155            x.map(|x| VecDeque::from(x.data.to_vec()))
156                .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))
157        });
158
159        let mut data_reader = tokio_util::io::StreamReader::new(data_stream);
160
161        let mut blob_writer = pin!(self.blob_service.open_write().await);
162
163        tokio::io::copy(&mut data_reader, &mut blob_writer)
164            .await
165            .map_err(|e| {
166                warn!("error copying: {}", e);
167                Status::internal("error copying")
168            })?;
169
170        let digest = blob_writer
171            .close()
172            .map_err(|e| {
173                warn!("error closing stream: {}", e);
174                Status::internal("error closing stream")
175            })
176            .await?;
177
178        Ok(Response::new(super::PutBlobResponse {
179            digest: digest.into(),
180        }))
181    }
182}
snix_castore/proto/grpc_blobservice_wrapper.rs

snix_castore/proto/
grpc_blobservice_wrapper.rs
