snix_castore/blob_engine/chunking/
mod.rs

1use tonic::async_trait;
2use typed_builder::TypedBuilder;
3
4use crate::{
5    B3Digest,
6    blob_engine::BlobEngine,
7    blobstore::{self, BlobReader, BlobStore, BlobWriter},
8    chunkstore::{self, ChunkStore},
9};
10
11mod writer;
12use writer::BlobWriterWrapper;
13
14/// A BlobEngine that will first query [BlobStore]
15/// to retrieve more granular chunking,
16/// then use the [ChunkStore] to render the blob.
17#[derive(TypedBuilder)]
18pub struct ChunkingBlobEngine<BS, CS> {
19    /// [BlobStore] that's used.
20    blob_store: BS,
21
22    /// Used [ChunkStore].
23    chunk_store: CS,
24
25    /// How many chunks to fetch in parallel for each reader
26    #[builder(default = 32)]
27    fetch_concurrency: usize,
28
29    /// How many chunk writes to do in parallel for each writer
30    #[builder(default = 32)]
31    upload_concurrency: usize,
32
33    /// The average chunk size to use when chunking writes.
34    /// Uses half of that as minimum, and double as maximum.
35    #[builder(default=256 * 1024)]
36    avg_chunk_size: u32,
37}
38
39#[async_trait]
40impl<BS, CS> BlobEngine for ChunkingBlobEngine<BS, CS>
41where
42    BS: BlobStore,
43    CS: ChunkStore + Clone + 'static,
44{
45    #[tracing::instrument(skip_all, fields(blob_digest=%digest))]
46    async fn has(&self, digest: &B3Digest) -> Result<bool, super::Error> {
47        Ok(self
48            .blob_store
49            .has(digest)
50            .await
51            .map_err(Error::Blobstore)?)
52    }
53
54    #[tracing::instrument(skip_all, fields(blob_digest=%digest, blob.size_hint=size_hint))]
55    async fn open_read(
56        &self,
57        digest: &B3Digest,
58        size_hint: Option<u64>,
59    ) -> Result<Option<Box<dyn crate::blobstore::Reader + 'static>>, super::Error> {
60        // If the size_hint suggests it's smaller than min_size,
61        // check chunk_store first before falling back to blob_store.
62        // We still need to check there as chunking parameters might have changed.
63        if let Some(size_hint) = size_hint
64            && size_hint < (self.avg_chunk_size as u64) / 2
65            && let Some(chunk) = self
66                .chunk_store
67                .get(digest)
68                .await
69                .map_err(Error::Chunkstore)?
70        {
71            return Ok(Some(Box::new(BlobReader::<CS>::from_single_chunk(chunk))
72                as Box<dyn crate::blobstore::Reader>));
73        }
74        Ok(self
75            .blob_store
76            .get(digest)
77            .await
78            .map_err(Error::Blobstore)?
79            .map(|blob_meta| {
80                Box::new(BlobReader::from_blob_meta(
81                    blob_meta,
82                    self.chunk_store.clone(),
83                    self.fetch_concurrency,
84                )) as Box<dyn crate::blobstore::Reader>
85            }))
86    }
87
88    async fn open_write<'a>(&'a self) -> Box<dyn crate::blobstore::Writer + 'a> {
89        let blob_writer = BlobWriter::new(
90            self.chunk_store.clone(),
91            self.avg_chunk_size,
92            self.upload_concurrency,
93        );
94
95        Box::new(BlobWriterWrapper::new(blob_writer, &self.blob_store))
96            as Box<dyn crate::blobstore::Writer>
97    }
98}
99
100#[derive(thiserror::Error, Debug)]
101enum Error {
102    #[error("querying blobstore: {0}")]
103    Blobstore(blobstore::Error),
104    #[error("querying chunkstore: {0}")]
105    Chunkstore(chunkstore::Error),
106}
107
108impl From<Error> for super::Error {
109    fn from(value: Error) -> Self {
110        Self(value.into())
111    }
112}
snix_castore/blob_engine/chunking/mod.rs

snix_castore/blob_engine/chunking/
mod.rs
