hashtree_network/
blob_router.rs

1//! Adaptive routing across opaque blob sources.
2
3use std::collections::{HashMap, HashSet};
4use std::future::Future;
5use std::panic::AssertUnwindSafe;
6use std::pin::Pin;
7use std::sync::Arc;
8use std::time::{Duration, Instant};
9
10use async_trait::async_trait;
11use futures::{stream::FuturesUnordered, FutureExt, StreamExt};
12use hashtree_core::{
13    BlobReply, BlobRequest, BlobRoute, BlobRouteContext, Hash, Store, StoreError, BLOB_DEFAULT_HTL,
14    BLOB_MAX_BYTES, BLOB_MAX_HTL,
15};
16use tokio::sync::{Mutex, RwLock};
17
18const OUTCOME_DECAY: f64 = 0.875;
19
20/// Stable, application-owned identity for one opaque route.
21#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
22pub struct BlobRouteIdentity(String);
23
24impl BlobRouteIdentity {
25    pub fn new(identity: impl Into<String>) -> Self {
26        Self(identity.into())
27    }
28
29    pub fn as_str(&self) -> &str {
30        &self.0
31    }
32}
33
34impl From<&str> for BlobRouteIdentity {
35    fn from(value: &str) -> Self {
36        Self::new(value)
37    }
38}
39
40impl From<String> for BlobRouteIdentity {
41    fn from(value: String) -> Self {
42        Self::new(value)
43    }
44}
45
46/// One route registered with the outer router.
47#[derive(Clone)]
48pub struct BlobRouteEntry {
49    pub identity: BlobRouteIdentity,
50    pub route: Arc<dyn BlobRoute>,
51}
52
53impl BlobRouteEntry {
54    pub fn new(identity: impl Into<BlobRouteIdentity>, route: Arc<dyn BlobRoute>) -> Self {
55        Self {
56            identity: identity.into(),
57            route,
58        }
59    }
60}
61
62/// Fixed bounds for one in-memory router.
63#[derive(Clone, Copy, Debug)]
64pub struct BlobRouterConfig {
65    pub request_timeout: Duration,
66    pub max_routes: usize,
67    pub max_route_attempts: usize,
68    pub route_attempt_budget: usize,
69    pub max_in_flight: usize,
70    pub hedge_delay: Duration,
71    pub initial_cooldown: Duration,
72    pub max_cooldown: Duration,
73    pub exploration_interval: u64,
74}
75
76impl Default for BlobRouterConfig {
77    fn default() -> Self {
78        Self {
79            request_timeout: Duration::from_secs(10),
80            max_routes: 32,
81            max_route_attempts: 32,
82            route_attempt_budget: 4,
83            max_in_flight: 2,
84            hedge_delay: Duration::from_millis(75),
85            initial_cooldown: Duration::from_millis(250),
86            max_cooldown: Duration::from_secs(10),
87            exploration_interval: 16,
88        }
89    }
90}
91
92#[derive(Clone, Debug, Default)]
93struct RouteOutcomes {
94    successes: f64,
95    failures: f64,
96    timeouts: f64,
97    successful_latency_ms: Option<f64>,
98    consecutive_failures: u32,
99    cooldown_until: Option<Instant>,
100    last_attempt: Option<Instant>,
101}
102
103impl RouteOutcomes {
104    fn decay(&mut self) {
105        self.successes *= OUTCOME_DECAY;
106        self.failures *= OUTCOME_DECAY;
107        self.timeouts *= OUTCOME_DECAY;
108    }
109
110    fn reliability(&self) -> f64 {
111        (self.successes + 1.0) / (self.successes + self.failures + self.timeouts + 2.0)
112    }
113
114    fn score(&self) -> f64 {
115        let latency = self
116            .successful_latency_ms
117            .map(|latency| 50.0 / (latency + 50.0))
118            .unwrap_or(0.5);
119        0.7 * self.reliability() + 0.3 * latency
120    }
121}
122
123#[derive(Default)]
124struct RouterState {
125    searches: u64,
126    outcomes: HashMap<BlobRouteIdentity, RouteOutcomes>,
127}
128
129/// Read-only view of bounded, decaying route outcomes.
130#[derive(Clone, Debug)]
131pub struct BlobRouteOutcomeSnapshot {
132    pub successful_weight: f64,
133    pub failure_weight: f64,
134    pub timeout_weight: f64,
135    pub successful_latency_ms: Option<f64>,
136    pub cooling_down: bool,
137}
138
139enum AttemptOutcome {
140    Data {
141        identity: BlobRouteIdentity,
142        data: Vec<u8>,
143        elapsed: Duration,
144    },
145    NoResult {
146        identity: BlobRouteIdentity,
147    },
148    Failure {
149        identity: BlobRouteIdentity,
150        message: String,
151    },
152}
153
154type AttemptFuture = Pin<Box<dyn Future<Output = AttemptOutcome> + Send>>;
155
156fn launch_attempt(
157    ordered: &[BlobRouteEntry],
158    next_route: &mut usize,
159    attempted: &mut usize,
160    pending: &mut FuturesUnordered<AttemptFuture>,
161    pending_identities: &mut HashSet<BlobRouteIdentity>,
162    request: BlobRequest,
163    route_context: BlobRouteContext,
164) {
165    let entry = ordered[*next_route].clone();
166    *next_route += 1;
167    *attempted += 1;
168    pending_identities.insert(entry.identity.clone());
169    pending.push(
170        async move {
171            let started = Instant::now();
172            let result = AssertUnwindSafe(entry.route.route_with_context(request, route_context))
173                .catch_unwind()
174                .await;
175            match result {
176                Ok(Ok(BlobReply::Data(data))) => AttemptOutcome::Data {
177                    identity: entry.identity,
178                    data,
179                    elapsed: started.elapsed(),
180                },
181                Ok(Ok(BlobReply::NoResult)) => AttemptOutcome::NoResult {
182                    identity: entry.identity,
183                },
184                Ok(Err(error)) => AttemptOutcome::Failure {
185                    identity: entry.identity,
186                    message: error.to_string(),
187                },
188                Err(_) => AttemptOutcome::Failure {
189                    identity: entry.identity,
190                    message: "route panicked".to_string(),
191                },
192            }
193        }
194        .boxed(),
195    );
196}
197
198/// One adaptive read router. Routes remain opaque and own any peer sets behind
199/// them; the router only orders route identities and bounds route attempts.
200pub struct BlobRouter {
201    routes: RwLock<Vec<BlobRouteEntry>>,
202    state: Mutex<RouterState>,
203    cache: Option<Arc<dyn Store>>,
204    config: BlobRouterConfig,
205}
206
207impl BlobRouter {
208    pub fn new(
209        routes: Vec<BlobRouteEntry>,
210        cache: Option<Arc<dyn Store>>,
211        config: BlobRouterConfig,
212    ) -> Result<Self, StoreError> {
213        validate_routes(&routes, config.max_routes)?;
214        let outcomes = routes
215            .iter()
216            .map(|entry| (entry.identity.clone(), RouteOutcomes::default()))
217            .collect();
218        Ok(Self {
219            routes: RwLock::new(routes),
220            state: Mutex::new(RouterState {
221                searches: 0,
222                outcomes,
223            }),
224            cache,
225            config,
226        })
227    }
228
229    /// Replace the route set. Outcome state survives only when both identity
230    /// and the exact route allocation are unchanged; provider replacement is
231    /// therefore immediately eligible for recovery.
232    pub async fn set_routes(&self, routes: Vec<BlobRouteEntry>) -> Result<(), StoreError> {
233        validate_routes(&routes, self.config.max_routes)?;
234        let mut current = self.routes.write().await;
235        let old_routes: HashMap<_, _> = current
236            .iter()
237            .map(|entry| (entry.identity.clone(), Arc::clone(&entry.route)))
238            .collect();
239        let mut state = self.state.lock().await;
240        state.outcomes.retain(|identity, _| {
241            routes.iter().any(|entry| {
242                &entry.identity == identity
243                    && old_routes
244                        .get(identity)
245                        .is_some_and(|old| Arc::ptr_eq(old, &entry.route))
246            })
247        });
248        for entry in &routes {
249            state.outcomes.entry(entry.identity.clone()).or_default();
250        }
251        *current = routes;
252        Ok(())
253    }
254
255    pub async fn route_count(&self) -> usize {
256        self.routes.read().await.len()
257    }
258
259    pub async fn outcomes(&self) -> HashMap<BlobRouteIdentity, BlobRouteOutcomeSnapshot> {
260        let now = Instant::now();
261        self.state
262            .lock()
263            .await
264            .outcomes
265            .iter()
266            .map(|(identity, outcome)| {
267                (
268                    identity.clone(),
269                    BlobRouteOutcomeSnapshot {
270                        successful_weight: outcome.successes,
271                        failure_weight: outcome.failures,
272                        timeout_weight: outcome.timeouts,
273                        successful_latency_ms: outcome.successful_latency_ms,
274                        cooling_down: outcome.cooldown_until.is_some_and(|until| until > now),
275                    },
276                )
277            })
278            .collect()
279    }
280
281    pub async fn get(
282        &self,
283        hash: &Hash,
284        preferred: Option<&[BlobRouteIdentity]>,
285    ) -> Result<Option<Vec<u8>>, StoreError> {
286        self.get_request(
287            BlobRequest {
288                hash: *hash,
289                htl: BLOB_DEFAULT_HTL,
290            },
291            preferred,
292            None,
293        )
294        .await
295    }
296
297    pub async fn get_request(
298        &self,
299        request: BlobRequest,
300        preferred: Option<&[BlobRouteIdentity]>,
301        outer_context: Option<BlobRouteContext>,
302    ) -> Result<Option<Vec<u8>>, StoreError> {
303        if request.htl > BLOB_MAX_HTL {
304            return Err(StoreError::Other(format!(
305                "Hashtree blob HTL {} exceeds the maximum of {BLOB_MAX_HTL}",
306                request.htl
307            )));
308        }
309
310        let now = Instant::now();
311        let own_deadline = now + self.config.request_timeout;
312        let deadline = outer_context
313            .map(|context| context.deadline.min(own_deadline))
314            .unwrap_or(own_deadline);
315        if deadline <= now {
316            return Err(StoreError::Other(
317                "blob retrieval deadline expired before the search started".to_string(),
318            ));
319        }
320        let outer_budget = outer_context
321            .map(|context| context.attempt_budget)
322            .unwrap_or(self.config.max_route_attempts);
323        let max_attempts = self.config.max_route_attempts.min(outer_budget);
324        if max_attempts == 0 {
325            return Err(StoreError::Other(
326                "blob retrieval has no route attempt budget".to_string(),
327            ));
328        }
329
330        let ordered = self.ordered_routes(preferred).await;
331        if ordered.is_empty() {
332            return Ok(None);
333        }
334        let max_attempts = max_attempts.min(ordered.len());
335        let max_in_flight = self.config.max_in_flight.max(1).min(max_attempts);
336        let route_context = BlobRouteContext {
337            deadline,
338            attempt_budget: self.config.route_attempt_budget.min(outer_budget).max(1),
339        };
340
341        let mut pending: FuturesUnordered<AttemptFuture> = FuturesUnordered::new();
342        let mut pending_identities = HashSet::new();
343        let mut next_route = 0usize;
344        let mut attempted = 0usize;
345        let mut failures = Vec::new();
346        launch_attempt(
347            &ordered,
348            &mut next_route,
349            &mut attempted,
350            &mut pending,
351            &mut pending_identities,
352            request,
353            route_context,
354        );
355        let mut next_hedge = Instant::now() + self.config.hedge_delay;
356        let deadline_sleep = tokio::time::sleep_until(tokio::time::Instant::from_std(deadline));
357        tokio::pin!(deadline_sleep);
358
359        loop {
360            if pending.is_empty() {
361                if attempted < max_attempts {
362                    launch_attempt(
363                        &ordered,
364                        &mut next_route,
365                        &mut attempted,
366                        &mut pending,
367                        &mut pending_identities,
368                        request,
369                        route_context,
370                    );
371                    next_hedge = Instant::now() + self.config.hedge_delay;
372                } else {
373                    break;
374                }
375            }
376
377            let hedge_at = tokio::time::Instant::from_std(next_hedge.min(deadline));
378            tokio::select! {
379                _ = &mut deadline_sleep => {
380                    for identity in pending_identities.drain() {
381                        self.record_timeout(&identity).await;
382                    }
383                    return Err(StoreError::Other(
384                        "blob retrieval deadline expired before the search completed".to_string(),
385                    ));
386                }
387                outcome = pending.next() => {
388                    let Some(outcome) = outcome else { continue };
389                    let identity = match &outcome {
390                        AttemptOutcome::Data { identity, .. }
391                        | AttemptOutcome::NoResult { identity }
392                        | AttemptOutcome::Failure { identity, .. } => identity.clone(),
393                    };
394                    pending_identities.remove(&identity);
395                    match outcome {
396                        AttemptOutcome::Data { identity, data, elapsed } => {
397                            if data.len() > BLOB_MAX_BYTES {
398                                let message = format!(
399                                    "route {} returned {} bytes, exceeding the {BLOB_MAX_BYTES}-byte limit",
400                                    identity.as_str(),
401                                    data.len()
402                                );
403                                self.record_failure(&identity).await;
404                                failures.push(message);
405                            } else if hashtree_core::sha256(&data) != request.hash {
406                                let message = format!(
407                                    "route {} returned content with the wrong hash",
408                                    identity.as_str()
409                                );
410                                self.record_failure(&identity).await;
411                                failures.push(message);
412                            } else {
413                                self.record_success(&identity, elapsed).await;
414                                if let Some(cache) = &self.cache {
415                                    let _ = cache.put(request.hash, data.clone()).await;
416                                }
417                                return Ok(Some(data));
418                            }
419                        }
420                        AttemptOutcome::NoResult { identity } => {
421                            self.record_no_result(&identity).await;
422                        }
423                        AttemptOutcome::Failure { identity, message } => {
424                            self.record_failure(&identity).await;
425                            failures.push(format!("route {}: {message}", identity.as_str()));
426                        }
427                    }
428                    if attempted < max_attempts && pending.len() < max_in_flight {
429                        launch_attempt(
430                            &ordered,
431                            &mut next_route,
432                            &mut attempted,
433                            &mut pending,
434                            &mut pending_identities,
435                            request,
436                            route_context,
437                        );
438                        next_hedge = Instant::now() + self.config.hedge_delay;
439                    }
440                }
441                _ = tokio::time::sleep_until(hedge_at), if attempted < max_attempts && pending.len() < max_in_flight => {
442                    launch_attempt(
443                        &ordered,
444                        &mut next_route,
445                        &mut attempted,
446                        &mut pending,
447                        &mut pending_identities,
448                        request,
449                        route_context,
450                    );
451                    next_hedge = Instant::now() + self.config.hedge_delay;
452                }
453            }
454        }
455
456        if attempted < ordered.len() {
457            return Err(StoreError::Other(format!(
458                "blob retrieval exhausted its bounded route attempt budget ({attempted}/{})",
459                ordered.len()
460            )));
461        }
462        if let Some(message) = failures.into_iter().next() {
463            return Err(StoreError::Other(format!(
464                "blob retrieval was incomplete: {message}"
465            )));
466        }
467        Ok(None)
468    }
469
470    async fn ordered_routes(&self, preferred: Option<&[BlobRouteIdentity]>) -> Vec<BlobRouteEntry> {
471        let routes = self.routes.read().await.clone();
472        let preferred_order: HashMap<_, _> = preferred
473            .unwrap_or_default()
474            .iter()
475            .enumerate()
476            .map(|(index, identity)| (identity.clone(), index))
477            .collect();
478        let now = Instant::now();
479        let mut state = self.state.lock().await;
480        state.searches = state.searches.wrapping_add(1);
481        let explore = self.config.exploration_interval > 0
482            && state
483                .searches
484                .is_multiple_of(self.config.exploration_interval);
485        let mut ordered = routes;
486        ordered.sort_by(|left, right| {
487            match (
488                preferred_order.get(&left.identity),
489                preferred_order.get(&right.identity),
490            ) {
491                (Some(left), Some(right)) => return left.cmp(right),
492                (Some(_), None) => return std::cmp::Ordering::Less,
493                (None, Some(_)) => return std::cmp::Ordering::Greater,
494                (None, None) => {}
495            }
496            let left_stats = state
497                .outcomes
498                .get(&left.identity)
499                .cloned()
500                .unwrap_or_default();
501            let right_stats = state
502                .outcomes
503                .get(&right.identity)
504                .cloned()
505                .unwrap_or_default();
506            if explore {
507                return left_stats
508                    .last_attempt
509                    .cmp(&right_stats.last_attempt)
510                    .then_with(|| left.identity.cmp(&right.identity));
511            }
512            let left_cooling = left_stats.cooldown_until.is_some_and(|until| until > now);
513            let right_cooling = right_stats.cooldown_until.is_some_and(|until| until > now);
514            left_cooling
515                .cmp(&right_cooling)
516                .then_with(|| {
517                    right_stats
518                        .score()
519                        .partial_cmp(&left_stats.score())
520                        .unwrap_or(std::cmp::Ordering::Equal)
521                })
522                .then_with(|| left_stats.last_attempt.cmp(&right_stats.last_attempt))
523                .then_with(|| left.identity.cmp(&right.identity))
524        });
525        ordered
526    }
527
528    async fn record_success(&self, identity: &BlobRouteIdentity, elapsed: Duration) {
529        let mut state = self.state.lock().await;
530        let outcome = state.outcomes.entry(identity.clone()).or_default();
531        outcome.decay();
532        outcome.successes += 1.0;
533        let elapsed_ms = elapsed.as_secs_f64() * 1_000.0;
534        outcome.successful_latency_ms = Some(
535            outcome
536                .successful_latency_ms
537                .map(|old| 0.875 * old + 0.125 * elapsed_ms)
538                .unwrap_or(elapsed_ms),
539        );
540        outcome.consecutive_failures = 0;
541        outcome.cooldown_until = None;
542        outcome.last_attempt = Some(Instant::now());
543    }
544
545    async fn record_no_result(&self, identity: &BlobRouteIdentity) {
546        let mut state = self.state.lock().await;
547        let outcome = state.outcomes.entry(identity.clone()).or_default();
548        outcome.decay();
549        outcome.consecutive_failures = 0;
550        outcome.cooldown_until = None;
551        outcome.last_attempt = Some(Instant::now());
552    }
553
554    async fn record_failure(&self, identity: &BlobRouteIdentity) {
555        let mut state = self.state.lock().await;
556        let outcome = state.outcomes.entry(identity.clone()).or_default();
557        outcome.decay();
558        outcome.failures += 1.0;
559        self.apply_cooldown(outcome);
560    }
561
562    async fn record_timeout(&self, identity: &BlobRouteIdentity) {
563        let mut state = self.state.lock().await;
564        let outcome = state.outcomes.entry(identity.clone()).or_default();
565        outcome.decay();
566        outcome.timeouts += 1.0;
567        self.apply_cooldown(outcome);
568    }
569
570    fn apply_cooldown(&self, outcome: &mut RouteOutcomes) {
571        outcome.consecutive_failures = outcome.consecutive_failures.saturating_add(1);
572        let multiplier =
573            2u32.saturating_pow(outcome.consecutive_failures.saturating_sub(1).min(16));
574        let cooldown = self
575            .config
576            .initial_cooldown
577            .saturating_mul(multiplier)
578            .min(self.config.max_cooldown);
579        let now = Instant::now();
580        outcome.cooldown_until = Some(now + cooldown);
581        outcome.last_attempt = Some(now);
582    }
583}
584
585#[async_trait]
586impl BlobRoute for BlobRouter {
587    async fn route(&self, request: BlobRequest) -> Result<BlobReply, StoreError> {
588        Ok(match self.get_request(request, None, None).await? {
589            Some(data) => BlobReply::Data(data),
590            None => BlobReply::NoResult,
591        })
592    }
593
594    async fn route_with_context(
595        &self,
596        request: BlobRequest,
597        context: BlobRouteContext,
598    ) -> Result<BlobReply, StoreError> {
599        Ok(
600            match self.get_request(request, None, Some(context)).await? {
601                Some(data) => BlobReply::Data(data),
602                None => BlobReply::NoResult,
603            },
604        )
605    }
606}
607
608fn validate_routes(routes: &[BlobRouteEntry], max_routes: usize) -> Result<(), StoreError> {
609    if routes.len() > max_routes {
610        return Err(StoreError::Other(format!(
611            "blob router has {} routes, exceeding its bound of {max_routes}",
612            routes.len()
613        )));
614    }
615    let mut identities = HashSet::with_capacity(routes.len());
616    for entry in routes {
617        if entry.identity.as_str().is_empty() {
618            return Err(StoreError::Other(
619                "blob route identity must not be empty".to_string(),
620            ));
621        }
622        if !identities.insert(entry.identity.clone()) {
623            return Err(StoreError::Other(format!(
624                "duplicate blob route identity {}",
625                entry.identity.as_str()
626            )));
627        }
628    }
629    Ok(())
630}
hashtree_network/blob_router.rs

hashtree_network/
blob_router.rs
