aspero_engine/
lib.rs

1//! Routing, caching, merging and resilience over registered providers.
2//!
3//! The engine owns what providers deliberately don't: retries and backoff
4//! ([`retry`]), request pacing and the real HTTP client ([`transport`]), and
5//! choosing which provider answers what.
6//!
7//! - **Lookup** is interchangeable across providers: each ID goes to the
8//!   providers that accept its kind, in priority order, until one finds it.
9//!   A provider that doesn't know an ID or fails falls through to the next;
10//!   every attempt is reported.
11//! - **Search** is not interchangeable (rankings differ), so it runs against
12//!   one provider: the one named, or the first capable one. Filters that
13//!   provider doesn't declare are stripped and reported as ignored, never
14//!   silently dropped.
15//!
16//! - **Resolve / verify** turn a citation claim into scored candidates and a
17//!   verdict (see [`resolve`](crate::Engine::resolve)).
18//!
19//! - **Caching**: provider answers (records, "not found", search pages) are
20//!   cached with per-kind TTLs ([`CachePolicy`]) and served stale if a
21//!   provider fails; every answer says how it was [`Served`].
22//!
23//! - **Sharing**: one engine serves many callers at once. Lookups already
24//!   in flight are joined rather than repeated, and each caller sets its
25//!   own patience with [`CallOptions`]. Providers that can't answer (a
26//!   spent quota, a rejected key, repeated failures) are skipped until
27//!   they can ([`Engine::unavailable`]).
28//!
29//! Not yet: fan-out search.
30
31mod abstracts;
32mod check;
33mod enrich;
34mod health;
35mod inflight;
36mod inputs;
37mod lookup;
38mod options;
39mod registry;
40mod resolve;
41pub mod retry;
42mod search;
43mod standard;
44pub mod transport;
45mod verdict;
46mod versions;
47
48pub use abstracts::{Abstract, AbstractsReport};
49pub use check::{Checked, Finding, Severity, duplicates, findings};
50pub use registry::{Disagreement, disagreements};
51pub use resolve::{Candidate, ResolveReport, Sighting, Via};
52/// Re-exported for credentials in provider configs.
53pub use secrecy::SecretString;
54#[cfg(feature = "crossref")]
55pub use standard::CrossrefConfig;
56#[cfg(feature = "datacite")]
57pub use standard::DataciteConfig;
58#[cfg(feature = "doi")]
59pub use standard::DoiConfig;
60#[cfg(feature = "europepmc")]
61pub use standard::EuropePmcConfig;
62#[cfg(feature = "openalex")]
63pub use standard::OpenAlexConfig;
64#[cfg(feature = "openlibrary")]
65pub use standard::OpenLibraryConfig;
66pub use standard::{StandardConfig, standard};
67pub use verdict::{Verdict, Verification, describe_update};
68pub use versions::VersionsReport;
69
70use std::sync::Arc;
71use std::time::Duration;
72
73use aspero_core::{Error, ErrorCategory, Provider, ProviderId, ResiliencePolicy, Usage};
74pub use aspero_store::{Cache, Pacing, SqliteCache, Stats as CacheStats, StoreError};
75pub use inputs::{Extras, InputLookup, InputsReport, Looked};
76use jiff::Timestamp;
77pub use lookup::{Attempt, LookupReport, LookupResult, Outcome};
78pub use options::CallOptions;
79pub use search::SearchReport;
80
81use crate::health::Health;
82use crate::inflight::InFlight;
83use crate::retry::{Jitter, random_jitter, with_retries};
84
85/// A clock, injectable so cache expiry is testable.
86pub type Clock = fn() -> Timestamp;
87
88/// How long cached answers are trusted (DESIGN §5). Past these, a provider
89/// is asked again; if it then fails, the old answer is served as stale.
90#[derive(Debug, Clone, Copy, PartialEq, Eq)]
91#[non_exhaustive]
92pub struct CachePolicy {
93    pub records: Duration,
94    /// "Not found" answers: short, since works get registered.
95    pub misses: Duration,
96    pub searches: Duration,
97}
98
99impl Default for CachePolicy {
100    fn default() -> Self {
101        const DAY: u64 = 86_400;
102        CachePolicy {
103            records: Duration::from_secs(7 * DAY),
104            misses: Duration::from_secs(DAY),
105            searches: Duration::from_secs(DAY),
106        }
107    }
108}
109
110/// Where an answer came from.
111#[derive(Debug, Clone, Copy, PartialEq, Eq)]
112#[non_exhaustive]
113pub enum Served {
114    Live,
115    /// A cached answer within its TTL.
116    Cache,
117    /// An expired cached answer, served because the provider failed.
118    Stale,
119}
120
121impl Served {
122    pub fn as_str(self) -> &'static str {
123        match self {
124            Served::Live => "live",
125            Served::Cache => "cache",
126            Served::Stale => "stale",
127        }
128    }
129}
130
131/// Cheap to clone: clones, and views made with
132/// [`with_options`](Engine::with_options), share the cache, providers'
133/// pacing and requests in flight. Hold one per process.
134#[derive(Clone)]
135pub struct Engine {
136    /// In priority order.
137    entries: Vec<Entry>,
138    jitter: Jitter,
139    cache: Option<Arc<dyn Cache>>,
140    cache_policy: CachePolicy,
141    clock: Clock,
142    options: CallOptions,
143    inflight: Arc<InFlight>,
144    health: Arc<Health>,
145}
146
147#[derive(Clone)]
148struct Entry {
149    provider: Arc<dyn Provider>,
150    policy: ResiliencePolicy,
151}
152
153pub struct EngineBuilder {
154    entries: Vec<(i32, Entry)>,
155    jitter: Jitter,
156    cache: Option<Arc<dyn Cache>>,
157    cache_policy: CachePolicy,
158    clock: Clock,
159}
160
161impl EngineBuilder {
162    /// Registers a provider. Lower `priority` is tried first; ties keep
163    /// registration order.
164    pub fn provider(
165        mut self,
166        provider: Arc<dyn Provider>,
167        policy: ResiliencePolicy,
168        priority: i32,
169    ) -> Self {
170        self.entries.push((priority, Entry { provider, policy }));
171        self
172    }
173
174    /// Replaces the backoff jitter source (for deterministic tests).
175    pub fn jitter(mut self, jitter: Jitter) -> Self {
176        self.jitter = jitter;
177        self
178    }
179
180    /// Caches provider answers (and serves them stale if a provider fails).
181    pub fn cache(mut self, cache: Arc<dyn Cache>) -> Self {
182        self.cache = Some(cache);
183        self
184    }
185
186    pub fn cache_policy(mut self, policy: CachePolicy) -> Self {
187        self.cache_policy = policy;
188        self
189    }
190
191    /// Replaces the clock (for deterministic cache-expiry tests).
192    pub fn clock(mut self, clock: Clock) -> Self {
193        self.clock = clock;
194        self
195    }
196
197    pub fn build(mut self) -> Engine {
198        self.entries.sort_by_key(|(priority, _)| *priority);
199        Engine {
200            entries: self.entries.into_iter().map(|(_, e)| e).collect(),
201            jitter: self.jitter,
202            cache: self.cache,
203            cache_policy: self.cache_policy,
204            clock: self.clock,
205            options: CallOptions::default(),
206            inflight: Arc::default(),
207            health: Arc::default(),
208        }
209    }
210}
211
212#[derive(Debug, Clone, thiserror::Error)]
213#[non_exhaustive]
214pub enum EngineError {
215    #[error("no enabled provider is named {0:?}")]
216    UnknownProvider(String),
217    #[error("no enabled provider supports {0}")]
218    NoCapableProvider(&'static str),
219    #[error("{provider}: {error}")]
220    Provider {
221        provider: ProviderId,
222        #[source]
223        error: Error,
224    },
225}
226
227impl EngineError {
228    pub fn category(&self) -> ErrorCategory {
229        match self {
230            EngineError::UnknownProvider(_) => ErrorCategory::Usage,
231            EngineError::NoCapableProvider(_) => ErrorCategory::Unsupported,
232            EngineError::Provider { error, .. } => error.category(),
233        }
234    }
235}
236
237impl Engine {
238    pub fn builder() -> EngineBuilder {
239        EngineBuilder {
240            entries: Vec::new(),
241            jitter: random_jitter,
242            cache: None,
243            cache_policy: CachePolicy::default(),
244            clock: Timestamp::now,
245        }
246    }
247
248    /// This engine for a caller with different patience; it shares
249    /// everything else.
250    pub fn with_options(&self, options: CallOptions) -> Engine {
251        Engine {
252            options,
253            ..self.clone()
254        }
255    }
256
257    pub fn options(&self) -> CallOptions {
258        self.options
259    }
260
261    /// Providers refusing calls now, and why: a spent quota (until it
262    /// resets), rejected credentials, or an open circuit breaker.
263    pub fn unavailable(&self) -> Vec<(ProviderId, Error)> {
264        self.health.unavailable((self.clock)())
265    }
266
267    /// Runs one provider operation with retries, unless the provider is
268    /// refusing calls, and remembers how it went.
269    async fn call<T, F, Fut>(&self, entry: &Entry, op: F) -> Result<T, Error>
270    where
271        F: FnMut() -> Fut,
272        Fut: std::future::Future<Output = Result<T, Error>>,
273    {
274        let pass = self.health.admit(&entry.provider.id(), (self.clock)())?;
275        let policy = self.options.apply(&entry.policy);
276        let result = with_retries(&policy, self.jitter, op).await;
277        pass.record(&entry.policy, result.as_ref().err(), (self.clock)());
278        result
279    }
280
281    /// Registered providers, in priority order.
282    pub fn providers(&self) -> impl Iterator<Item = &Arc<dyn Provider>> {
283        self.entries.iter().map(|e| &e.provider)
284    }
285}
286
287fn add_usage(totals: &mut Vec<(ProviderId, Usage)>, provider: &ProviderId, usage: Usage) {
288    match totals.iter_mut().find(|(p, _)| p == provider) {
289        Some((_, total)) => *total = total.then(usage),
290        None => totals.push((provider.clone(), usage)),
291    }
292}
aspero_engine/lib.rs

aspero_engine/
lib.rs
