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                "{}/{}.bin",
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.status().is_success(),
54            "unexpected Hashtree HTTP status: {}",
55            response.status()
56        );
57        ensure!(
58            response
59                .content_length()
60                .is_none_or(|n| n <= MAX_BLOCK as u64),
61            "oversized Hashtree block"
62        );
63        let mut body = response.bytes_stream();
64        let mut bytes = Vec::new();
65        while let Some(part) = body.next().await {
66            let part = part?;
67            ensure!(
68                bytes.len().saturating_add(part.len()) <= MAX_BLOCK,
69                "oversized Hashtree block"
70            );
71            bytes.extend_from_slice(&part);
72        }
73        verify(hash, &bytes)?;
74        Ok(Some(bytes))
75    }
76
77    async fn retrieve(&self, hash: &Hash) -> Result<Option<Vec<u8>>> {
78        if let Some(bytes) = self.cache.get(hash).await? {
79            verify(hash, &bytes)?;
80            return Ok(Some(bytes));
81        }
82        let mut servers: Vec<&str> = self.config.daemon_url.as_deref().into_iter().collect();
83        if !self.config.local_only {
84            servers.extend(self.config.read_servers.iter().map(String::as_str));
85        }
86        ensure!(
87            !servers.is_empty(),
88            "no Hashtree content sources configured"
89        );
90        let mut pending = FuturesUnordered::new();
91        for (i, server) in servers.into_iter().enumerate() {
92            pending.push(async move {
93                // Give the existing daemon first chance; slow peers do not block
94                // other sources or get misreported as a definitive absence.
95                tokio::time::sleep(Duration::from_millis((i as u64).min(8) * 150)).await;
96                self.fetch(server, hash).await
97            });
98        }
99        let mut error = None;
100        while let Some(result) = pending.next().await {
101            match result {
102                Ok(Some(bytes)) => {
103                    self.cache.put(*hash, bytes.clone()).await?;
104                    return Ok(Some(bytes));
105                }
106                Ok(None) => {}
107                Err(e) => error = Some(e),
108            }
109        }
110        if let Some(e) = error {
111            return Err(e);
112        }
113        Ok(None)
114    }
115}
116
117#[async_trait]
118impl Store for BlobStore {
119    async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
120        self.retrieve(hash)
121            .await
122            .map_err(|e| StoreError::Other(e.to_string()))
123    }
124    async fn put(&self, hash: Hash, bytes: Vec<u8>) -> Result<bool, StoreError> {
125        verify(&hash, &bytes).map_err(|e| StoreError::Other(e.to_string()))?;
126        self.cache.put(hash, bytes).await
127    }
128    async fn has(&self, hash: &Hash) -> Result<bool, StoreError> {
129        Ok(self.get(hash).await?.is_some())
130    }
131    async fn delete(&self, hash: &Hash) -> Result<bool, StoreError> {
132        self.cache.delete(hash).await
133    }
134}
hashtree_client/store.rs

hashtree_client/
store.rs
