hashtree_client/
store.rs

1use crate::ClientConfig;
2use anyhow::{bail, ensure, Context, 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        // Keep credentials, query strings and private URL paths out of diagnostics.
40        let origin = reqwest::Url::parse(server)
41            .map(|url| url.origin().ascii_serialization())
42            .unwrap_or_else(|_| "invalid source URL".into());
43        let context = || format!("Hashtree block {} from {origin}", hex::encode(hash));
44        match self.fetch_once(server, hash).await {
45            Err(error)
46                if error.downcast_ref::<reqwest::Error>().is_some_and(|error| {
47                    error.is_connect() || error.is_timeout() || error.is_body() || error.is_decode()
48                }) =>
49            {
50                // A partial response is never cached. Retry just this source once;
51                // the other hedged sources remain free to finish first.
52                tokio::time::sleep(Duration::from_millis(250)).await;
53                self.fetch_once(server, hash).await.with_context(context)
54            }
55            result => result.with_context(context),
56        }
57    }
58
59    async fn fetch_once(&self, server: &str, hash: &Hash) -> Result<Option<Vec<u8>>> {
60        let response = self
61            .http
62            .get(format!(
63                "{}/{}.bin",
64                server.trim_end_matches('/'),
65                hex::encode(hash)
66            ))
67            .send()
68            .await
69            .map_err(reqwest::Error::without_url)?;
70        if response.status() == reqwest::StatusCode::NOT_FOUND {
71            return Ok(None);
72        }
73        let response = response
74            .error_for_status()
75            .map_err(reqwest::Error::without_url)?;
76        ensure!(
77            response.status().is_success(),
78            "unexpected Hashtree HTTP status: {}",
79            response.status()
80        );
81        ensure!(
82            response
83                .content_length()
84                .is_none_or(|n| n <= MAX_BLOCK as u64),
85            "oversized Hashtree block"
86        );
87        let mut body = response.bytes_stream();
88        let mut bytes = Vec::new();
89        while let Some(part) = body.next().await {
90            let part = part.map_err(reqwest::Error::without_url)?;
91            ensure!(
92                bytes.len().saturating_add(part.len()) <= MAX_BLOCK,
93                "oversized Hashtree block"
94            );
95            bytes.extend_from_slice(&part);
96        }
97        verify(hash, &bytes)?;
98        Ok(Some(bytes))
99    }
100
101    async fn retrieve(&self, hash: &Hash) -> Result<Option<Vec<u8>>> {
102        if let Some(bytes) = self.cache.get(hash).await? {
103            verify(hash, &bytes)?;
104            return Ok(Some(bytes));
105        }
106        let mut servers: Vec<&str> = self.config.daemon_url.as_deref().into_iter().collect();
107        if !self.config.local_only {
108            servers.extend(self.config.read_servers.iter().map(String::as_str));
109        }
110        ensure!(
111            !servers.is_empty(),
112            "no Hashtree content sources configured"
113        );
114        let mut pending = FuturesUnordered::new();
115        for (i, server) in servers.into_iter().enumerate() {
116            pending.push(async move {
117                // Give the existing daemon first chance; slow peers do not block
118                // other sources or get misreported as a definitive absence.
119                tokio::time::sleep(Duration::from_millis((i as u64).min(8) * 150)).await;
120                self.fetch(server, hash).await
121            });
122        }
123        let mut errors = Vec::new();
124        while let Some(result) = pending.next().await {
125            match result {
126                Ok(Some(bytes)) => {
127                    self.cache.put(*hash, bytes.clone()).await?;
128                    return Ok(Some(bytes));
129                }
130                Ok(None) => {}
131                Err(error) => errors.push(format!("{error:#}")),
132            }
133        }
134        if !errors.is_empty() {
135            bail!("all Hashtree content sources failed: {}", errors.join("; "));
136        }
137        Ok(None)
138    }
139}
140
141#[async_trait]
142impl Store for BlobStore {
143    async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
144        self.retrieve(hash)
145            .await
146            .map_err(|e| StoreError::Other(format!("{e:#}")))
147    }
148    async fn put(&self, hash: Hash, bytes: Vec<u8>) -> Result<bool, StoreError> {
149        verify(&hash, &bytes).map_err(|e| StoreError::Other(e.to_string()))?;
150        self.cache.put(hash, bytes).await
151    }
152    async fn has(&self, hash: &Hash) -> Result<bool, StoreError> {
153        Ok(self.get(hash).await?.is_some())
154    }
155    async fn delete(&self, hash: &Hash) -> Result<bool, StoreError> {
156        self.cache.delete(hash).await
157    }
158}
hashtree_client/store.rs

hashtree_client/
store.rs
