snix_castore/blobstore/
object_store.rs

1//! Object-store backend, also supporting the local filesystem.
2
3use std::sync::Arc;
4
5use data_encoding::HEXLOWER;
6use hashbrown::HashMap;
7use object_store::{ObjectStoreExt, PutPayload, path::Path};
8use prost::Message;
9use tonic::async_trait;
10use tracing::{Level, instrument};
11
12use crate::{
13    B3Digest,
14    blobstore::{BlobMeta, BlobStore},
15    chunkstore::{ChunkStore, object_store::ObjectStoreChunkStore},
16    composition::{CompositionContext, ServiceBuilder},
17    proto,
18};
19
20/// Stores BlobMeta in any backend supported by the [object_store] crate.
21///
22/// Each [BlobMeta] is stored as a [proto::BlobMeta] at
23/// `{base_path}/blob/b3/{HEXLOWER(digest[0..2])}/{HEXLOWER(digest)}`.
24///
25/// If old BlobMeta are present, it might also access
26/// `{base_path}/chunk/b3/{HEXLOWER(digest[0..2])}/{HEXLOWER(digest)}`.
27pub struct ObjectStoreBlobStore {
28    instance_name: String,
29    object_store: Arc<dyn object_store::ObjectStore>,
30    base_path: Path,
31}
32
33#[instrument(level=Level::TRACE, skip_all,fields(base_path=%base_path,blob.digest=%digest),ret(Display))]
34fn derive_blob_path(base_path: &Path, digest: &B3Digest) -> Path {
35    base_path
36        .clone()
37        .join("blobs")
38        .join("b3")
39        .join(HEXLOWER.encode(&digest[..2]))
40        .join(HEXLOWER.encode(&digest[..]))
41}
42
43#[async_trait]
44impl BlobStore for ObjectStoreBlobStore {
45    #[instrument(skip_all, err, fields(blob.digest=%digest, instance_name=%self.instance_name))]
46    async fn get(&self, digest: &B3Digest) -> Result<Option<BlobMeta>, super::Error> {
47        if digest == &B3Digest::EMPTY {
48            return Ok(Some(BlobMeta::from_digests_and_sizes([])));
49        }
50
51        let p = derive_blob_path(&self.base_path, digest);
52        match self.object_store.get(&p).await {
53            Ok(get_result) => {
54                let bytes = get_result
55                    .bytes()
56                    .await
57                    .map_err(|err| Error::object_store("reading payload", err))?;
58
59                // parse proto
60                let blob_meta_proto =
61                    proto::BlobMeta::decode(bytes).map_err(Error::DecodeBlobMetaProto)?;
62
63                let blob_meta = match BlobMeta::try_from(blob_meta_proto) {
64                    Ok(blob_meta) => blob_meta,
65                    Err(super::blob_meta::DecodeError::NoChunks) => {
66                        // synthesize BlobMeta for single-chunk blob
67                        self.gen_single_chunk_blob_meta(digest).await?
68
69                        // FUTUREWORK: write fixed BlobMeta, configurable?
70                    }
71                    Err(err) => Err(Error::DecodeBlobMeta(err))?,
72                };
73
74                Ok(Some(blob_meta))
75            }
76            Err(object_store::Error::NotFound { .. }) => Ok(None),
77            Err(err) => Err(Error::object_store("getting blob", err))?,
78        }
79    }
80
81    #[instrument(skip_all, err, fields(blob.digest=%digest, instance_name=%self.instance_name))]
82    async fn has(&self, digest: &B3Digest) -> Result<bool, super::Error> {
83        let p = derive_blob_path(&self.base_path, digest);
84        match self.object_store.head(&p).await {
85            Ok(_) => Ok(true),
86            Err(object_store::Error::NotFound { .. }) => Ok(false),
87            Err(err) => Err(Error::object_store("heading blob", err))?,
88        }
89    }
90
91    #[instrument(skip_all, err, fields(blob.digest=%digest, instance_name=%self.instance_name))]
92    async fn put(&self, digest: &B3Digest, blob_meta: BlobMeta) -> Result<(), super::Error> {
93        let p = derive_blob_path(&self.base_path, digest);
94
95        let serialized = proto::BlobMeta::from(blob_meta).encode_to_vec();
96        self.object_store
97            .put(&p, PutPayload::from_bytes(serialized.into()))
98            .await
99            .map_err(|err| Error::object_store("putting blob", err))?;
100
101        Ok(())
102    }
103}
104
105impl ObjectStoreBlobStore {
106    /// Create [BlobMeta] for a single chunk.
107    ///
108    /// In an earlier format, when [BlobMeta] described a single chunk,
109    /// the version serialized to disk was an empty file.
110    /// As we cannot immediately return a [BlobMeta] from that,
111    /// as we don't know its size.
112    ///
113    /// This fetches the chunk, assuming an ObjectStoreChunkStore at the same
114    /// backend and base path, and returns a [BlobMeta].
115    ///
116    /// If the chunk cannot be fetched from the chunk store, it returns None.
117    async fn gen_single_chunk_blob_meta(&self, digest: &B3Digest) -> Result<BlobMeta, Error> {
118        // retrieve the single chunk, by creating a [ObjectStoreChunkStore] on the fly.
119        let chunk_store = ObjectStoreChunkStore::from_parts(
120            format!("{}-chunkstore", self.instance_name),
121            self.object_store.clone(),
122            self.base_path.clone(),
123        );
124
125        let chunk = chunk_store
126            .get(digest)
127            .await
128            .map_err(Error::SingleChunkGetFailure)?
129            .ok_or(Error::SingleChunkNotFound)?;
130
131        Ok(BlobMeta::from_digests_and_sizes([(
132            *digest,
133            chunk.len() as u64,
134        )]))
135    }
136}
137
138/// Configuration for an [ObjectStoreBlobStore].
139#[derive(serde::Deserialize)]
140#[serde(deny_unknown_fields)]
141pub struct ObjectStoreBlobStoreConfig {
142    object_store_url: url::Url,
143    #[serde(default)]
144    object_store_options: HashMap<String, String>,
145}
146
147impl TryFrom<url::Url> for ObjectStoreBlobStoreConfig {
148    type Error = Box<dyn std::error::Error + Send + Sync>;
149    fn try_from(url: url::Url) -> Result<Self, Self::Error> {
150        Ok(ObjectStoreBlobStoreConfig {
151            object_store_url: crate::object_store::trim_objectstore_prefix(&url)?,
152            object_store_options: url
153                .query_pairs()
154                .into_iter()
155                .map(|(k, v)| (k.to_string(), v.to_string()))
156                .collect(),
157        })
158    }
159}
160
161#[async_trait]
162impl ServiceBuilder for ObjectStoreBlobStoreConfig {
163    type Output = dyn BlobStore;
164    async fn build<'a>(
165        &'a self,
166        instance_name: &str,
167        _context: &CompositionContext,
168    ) -> Result<Arc<Self::Output>, Box<dyn std::error::Error + Send + Sync>> {
169        let (object_store, base_path) = crate::object_store::setup_object_store(
170            &self.object_store_url,
171            &self.object_store_options,
172        )
173        .await?;
174
175        Ok(Arc::new(ObjectStoreBlobStore {
176            instance_name: instance_name.to_string(),
177            object_store: Arc::new(object_store),
178            base_path,
179        }))
180    }
181}
182
183/// Error returned from [ObjectStoreChunkStore].
184#[derive(thiserror::Error, Debug)]
185enum Error {
186    #[error("decoding BlobMeta proto: {0}")]
187    DecodeBlobMetaProto(#[from] prost::DecodeError),
188
189    #[error("decoding BlobMeta: {0}")]
190    DecodeBlobMeta(#[from] crate::blobstore::blob_meta::DecodeError),
191
192    #[error("{0} in object_store: {1})")]
193    ObjectStore(&'static str, object_store::Error),
194
195    #[error("unable to fetch single chunk: {0}")]
196    SingleChunkGetFailure(crate::chunkstore::Error),
197
198    #[error("unable to find single chunk")]
199    SingleChunkNotFound,
200}
201
202impl Error {
203    /// Wrap an [object_store::Error] with some context
204    fn object_store(message: &'static str, err: object_store::Error) -> Self {
205        Self::ObjectStore(message, err)
206    }
207}
208
209impl From<Error> for super::Error {
210    fn from(value: Error) -> Self {
211        super::Error(Box::new(value))
212    }
213}
214
215#[cfg(test)]
216mod test {
217    use std::sync::Arc;
218
219    use object_store::{ObjectStoreExt, PutPayload};
220
221    use super::ObjectStoreBlobStore;
222    use crate::{
223        blobstore::{BlobMeta, BlobStore},
224        chunkstore::{self, Chunk, ChunkStore},
225    };
226
227    #[tokio::test]
228    async fn single_chunk_blobs() {
229        let chunk = Chunk::from_static(b"test");
230        let exp_blob_meta =
231            BlobMeta::from_digests_and_sizes([(chunk.digest(), chunk.len() as u64)]);
232
233        let os = Arc::new(object_store::memory::InMemory::new());
234        let base_path: object_store::path::Path = "/".into();
235        let instance_name = "uut".to_string();
236
237        let chunk_store = chunkstore::object_store::ObjectStoreChunkStore::from_parts(
238            instance_name.clone(),
239            os.clone(),
240            base_path.clone(),
241        );
242
243        let blob_store = ObjectStoreBlobStore {
244            instance_name,
245            object_store: os.clone(),
246            base_path: base_path.clone(),
247        };
248
249        // Store the Chunk only, and persist the legacy 0-byte marker file where the BlobMeta should be.
250        chunk_store
251            .put(chunk.clone())
252            .await
253            .expect("putting chunk to succeed");
254        let blob_path = super::derive_blob_path(&base_path, &chunk.digest());
255        os.put(&blob_path, PutPayload::from_static(b""))
256            .await
257            .expect("put to succeed");
258
259        // Try to retrieve a BlobMeta, it should be synthesized.
260        let blob_meta = blob_store
261            .get(&chunk.digest())
262            .await
263            .expect("to succeed")
264            .expect("to be some");
265        assert_eq!(exp_blob_meta, blob_meta, "expect BlobMeta to be correct");
266    }
267}
snix_castore/blobstore/object_store.rs

snix_castore/blobstore/
object_store.rs
