tessera/
resolver.rs

1//! Concurrent presigned-URL resolution.
2//!
3//! Each `(asset, coin, month)` partition needs its own short-lived presigned
4//! URL, minted via the download endpoint. For multi-partition reads we resolve
5//! them concurrently — a scoped thread pool for the sync client,
6//! `futures::try_join_all` for async.
7
8use std::sync::atomic::{AtomicUsize, Ordering};
9use std::sync::{Mutex, PoisonError};
10
11use futures::future::try_join_all;
12
13use crate::error::TesseraError;
14use crate::models::PartitionRef;
15use crate::readers::ResolvedPartition;
16
17/// Cap on worker threads / concurrent URL fetches.
18pub const MAX_WORKERS: usize = 8;
19
20/// Resolve presigned URLs for `refs` concurrently, preserving order.
21///
22/// A single ref is fetched inline; multiple refs are fanned out over at most
23/// [`MAX_WORKERS`] worker threads with short-circuit on the first error.
24///
25/// # Errors
26///
27/// Returns the first fetch error, if any.
28/// # Panics
29///
30/// Panics if internal bookkeeping is inconsistent (a slot is unresolved
31/// although no error was recorded) — unreachable by construction.
32pub fn resolve_sync(
33    fetch_url: impl Fn(&PartitionRef) -> Result<String, TesseraError> + Sync,
34    refs: &[PartitionRef],
35) -> Result<Vec<ResolvedPartition>, TesseraError> {
36    if let [partition] = refs {
37        return Ok(vec![(partition.clone(), fetch_url(partition)?)]);
38    }
39
40    let workers = MAX_WORKERS.min(refs.len());
41    let next_index = AtomicUsize::new(0);
42    let urls: Mutex<Vec<Option<String>>> = Mutex::new(vec![None; refs.len()]);
43    let first_error: Mutex<Option<TesseraError>> = Mutex::new(None);
44
45    std::thread::scope(|scope| {
46        for _ in 0..workers {
47            scope.spawn(|| {
48                loop {
49                    if first_error
50                        .lock()
51                        .unwrap_or_else(PoisonError::into_inner)
52                        .is_some()
53                    {
54                        break;
55                    }
56                    let index = next_index.fetch_add(1, Ordering::Relaxed);
57                    if index >= refs.len() {
58                        break;
59                    }
60                    match fetch_url(&refs[index]) {
61                        Ok(url) => {
62                            urls.lock().unwrap_or_else(PoisonError::into_inner)[index] = Some(url);
63                        }
64                        Err(err) => {
65                            let mut guard =
66                                first_error.lock().unwrap_or_else(PoisonError::into_inner);
67                            if guard.is_none() {
68                                *guard = Some(err);
69                            }
70                            break;
71                        }
72                    }
73                }
74            });
75        }
76    });
77
78    if let Some(err) = first_error
79        .into_inner()
80        .unwrap_or_else(PoisonError::into_inner)
81    {
82        return Err(err);
83    }
84    let resolved = urls.into_inner().unwrap_or_else(PoisonError::into_inner);
85    Ok(refs
86        .iter()
87        .zip(resolved)
88        .map(|(partition, url)| {
89            (
90                partition.clone(),
91                url.expect("resolved when no error recorded"),
92            )
93        })
94        .collect())
95}
96
97/// Resolve presigned URLs for `refs` concurrently, preserving order.
98///
99/// # Errors
100///
101/// Returns the first fetch error, if any.
102pub async fn resolve_async<F, Fut>(
103    fetch_url: F,
104    refs: &[PartitionRef],
105) -> Result<Vec<ResolvedPartition>, TesseraError>
106where
107    F: Fn(PartitionRef) -> Fut,
108    Fut: Future<Output = Result<String, TesseraError>>,
109{
110    let urls = try_join_all(refs.iter().map(|partition| fetch_url(partition.clone()))).await?;
111    Ok(refs.iter().cloned().zip(urls).collect())
112}
tessera/resolver.rs

tessera/
resolver.rs
