hashtree_client/
store.rs

1use crate::ClientConfig;
2use anyhow::{ensure, Result};
3use async_trait::async_trait;
4use futures::{stream::FuturesUnordered, StreamExt};
5use hashtree_core::{Hash, Store, StoreError};
6use hashtree_fs::FsBlobStore;
7use sha2::{Digest, Sha256};
8use std::{path::Path, time::Duration};
9
10const MAX_BLOCK: usize = 4 * 1024 * 1024;
11
12pub struct BlobStore {
13    cache: FsBlobStore,
14    config: ClientConfig,
15    pub(crate) http: reqwest::Client,
16}
17
18fn verify(hash: &Hash, bytes: &[u8]) -> Result<()> {
19    ensure!(
20        bytes.len() <= MAX_BLOCK && Sha256::digest(bytes).as_slice() == hash,
21        "invalid Hashtree block hash or size"
22    );
23    Ok(())
24}
25
26impl BlobStore {
27    pub fn new(cache: &Path, config: ClientConfig) -> Result<Self> {
28        Ok(Self {
29            cache: FsBlobStore::new(cache)?,
30            http: reqwest::Client::builder()
31                .timeout(config.request_timeout)
32                .redirect(reqwest::redirect::Policy::none())
33                .build()?,
34            config,
35        })
36    }
37
38    async fn fetch(&self, server: &str, hash: &Hash) -> Result<Option<Vec<u8>>> {
39        let response = self
40            .http
41            .get(format!(
42                "{}/{}",
43                server.trim_end_matches('/'),
44                hex::encode(hash)
45            ))
46            .send()
47            .await?;
48        if response.status() == reqwest::StatusCode::NOT_FOUND {
49            return Ok(None);
50        }
51        let response = response.error_for_status()?;
52        ensure!(
53            response
54                .content_length()
55                .is_none_or(|n| n <= MAX_BLOCK as u64),
56            "oversized Hashtree block"
57        );
58        let mut body = response.bytes_stream();
59        let mut bytes = Vec::new();
60        while let Some(part) = body.next().await {
61            let part = part?;
62            ensure!(
63                bytes.len().saturating_add(part.len()) <= MAX_BLOCK,
64                "oversized Hashtree block"
65            );
66            bytes.extend_from_slice(&part);
67        }
68        verify(hash, &bytes)?;
69        Ok(Some(bytes))
70    }
71
72    async fn retrieve(&self, hash: &Hash) -> Result<Option<Vec<u8>>> {
73        if let Some(bytes) = self.cache.get(hash).await? {
74            verify(hash, &bytes)?;
75            return Ok(Some(bytes));
76        }
77        let mut servers: Vec<&str> = self.config.daemon_url.as_deref().into_iter().collect();
78        if !self.config.local_only {
79            servers.extend(self.config.read_servers.iter().map(String::as_str));
80        }
81        ensure!(
82            !servers.is_empty(),
83            "no Hashtree content sources configured"
84        );
85        let mut pending = FuturesUnordered::new();
86        for (i, server) in servers.into_iter().enumerate() {
87            pending.push(async move {
88                // Give the existing daemon first chance; slow peers do not block
89                // other sources or get misreported as a definitive absence.
90                tokio::time::sleep(Duration::from_millis((i as u64).min(8) * 150)).await;
91                self.fetch(server, hash).await
92            });
93        }
94        let mut error = None;
95        while let Some(result) = pending.next().await {
96            match result {
97                Ok(Some(bytes)) => {
98                    self.cache.put(*hash, bytes.clone()).await?;
99                    return Ok(Some(bytes));
100                }
101                Ok(None) => {}
102                Err(e) => error = Some(e),
103            }
104        }
105        if let Some(e) = error {
106            return Err(e);
107        }
108        Ok(None)
109    }
110}
111
112#[async_trait]
113impl Store for BlobStore {
114    async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
115        self.retrieve(hash)
116            .await
117            .map_err(|e| StoreError::Other(e.to_string()))
118    }
119    async fn put(&self, hash: Hash, bytes: Vec<u8>) -> Result<bool, StoreError> {
120        verify(&hash, &bytes).map_err(|e| StoreError::Other(e.to_string()))?;
121        self.cache.put(hash, bytes).await
122    }
123    async fn has(&self, hash: &Hash) -> Result<bool, StoreError> {
124        Ok(self.get(hash).await?.is_some())
125    }
126    async fn delete(&self, hash: &Hash) -> Result<bool, StoreError> {
127        self.cache.delete(hash).await
128    }
129}
hashtree_client/store.rs

hashtree_client/
store.rs
