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 budget;
33mod check;
34mod enrich;
35mod health;
36mod inflight;
37mod inputs;
38mod lookup;
39mod options;
40mod registry;
41mod resolve;
42pub mod retry;
43mod search;
44mod standard;
45pub mod transport;
46mod verdict;
47mod versions;
48
49pub use abstracts::{Abstract, AbstractsReport};
50pub use budget::Budget;
51pub use check::{Checked, Finding, Severity, duplicates, findings};
52pub use registry::{Disagreement, disagreements};
53pub use resolve::{Candidate, ResolveReport, Sighting, Via};
54/// Re-exported for credentials in provider configs.
55pub use secrecy::SecretString;
56#[cfg(feature = "crossref")]
57pub use standard::CrossrefConfig;
58#[cfg(feature = "datacite")]
59pub use standard::DataciteConfig;
60#[cfg(feature = "doi")]
61pub use standard::DoiConfig;
62#[cfg(feature = "europepmc")]
63pub use standard::EuropePmcConfig;
64#[cfg(feature = "openalex")]
65pub use standard::OpenAlexConfig;
66#[cfg(feature = "openlibrary")]
67pub use standard::OpenLibraryConfig;
68pub use standard::{StandardConfig, standard};
69pub use verdict::{CitedField, Difference, Reason, Verdict, Verification, describe_update};
70pub use versions::VersionsReport;
71
72use std::sync::Arc;
73use std::time::Duration;
74
75use aspero_core::{
76    Error, ErrorCategory, ErrorKind, Provider, ProviderId, ResiliencePolicy, Response, Usage,
77};
78pub use aspero_store::{Cache, Pacing, SqliteCache, Stats as CacheStats, StoreError};
79pub use inputs::{Extras, InputLookup, InputsReport, Looked};
80use jiff::Timestamp;
81pub use lookup::{Attempt, LookupReport, LookupResult, Outcome};
82pub use options::CallOptions;
83pub use search::SearchReport;
84
85use crate::budget::Budgets;
86use crate::health::Health;
87use crate::inflight::InFlight;
88use crate::retry::{Jitter, random_jitter, with_retries};
89
90/// A clock, injectable so cache expiry is testable.
91pub type Clock = fn() -> Timestamp;
92
93/// How long cached answers are trusted (DESIGN §5). Past these, a provider
94/// is asked again; if it then fails, the old answer is served as stale.
95#[derive(Debug, Clone, Copy, PartialEq, Eq)]
96#[non_exhaustive]
97pub struct CachePolicy {
98    pub records: Duration,
99    /// "Not found" answers: short, since works get registered.
100    pub misses: Duration,
101    pub searches: Duration,
102}
103
104impl Default for CachePolicy {
105    fn default() -> Self {
106        const DAY: u64 = 86_400;
107        CachePolicy {
108            records: Duration::from_secs(7 * DAY),
109            misses: Duration::from_secs(DAY),
110            searches: Duration::from_secs(DAY),
111        }
112    }
113}
114
115/// Where an answer came from.
116#[derive(Debug, Clone, Copy, PartialEq, Eq)]
117#[non_exhaustive]
118pub enum Served {
119    Live,
120    /// A cached answer within its TTL.
121    Cache,
122    /// An expired cached answer, served because the provider failed.
123    Stale,
124}
125
126impl Served {
127    pub fn as_str(self) -> &'static str {
128        match self {
129            Served::Live => "live",
130            Served::Cache => "cache",
131            Served::Stale => "stale",
132        }
133    }
134}
135
136/// Cheap to clone: clones, and views made with
137/// [`with_options`](Engine::with_options), share the cache, providers'
138/// pacing and requests in flight. Hold one per process.
139#[derive(Clone)]
140pub struct Engine {
141    /// In priority order.
142    entries: Vec<Entry>,
143    jitter: Jitter,
144    cache: Option<Arc<dyn Cache>>,
145    cache_policy: CachePolicy,
146    clock: Clock,
147    options: CallOptions,
148    inflight: Arc<InFlight>,
149    health: Arc<Health>,
150    budgets: Arc<Budgets>,
151}
152
153#[derive(Clone)]
154struct Entry {
155    provider: Arc<dyn Provider>,
156    policy: ResiliencePolicy,
157}
158
159pub struct EngineBuilder {
160    entries: Vec<(i32, Entry)>,
161    jitter: Jitter,
162    cache: Option<Arc<dyn Cache>>,
163    cache_policy: CachePolicy,
164    clock: Clock,
165}
166
167impl EngineBuilder {
168    /// Registers a provider. Lower `priority` is tried first; ties keep
169    /// registration order.
170    pub fn provider(
171        mut self,
172        provider: Arc<dyn Provider>,
173        policy: ResiliencePolicy,
174        priority: i32,
175    ) -> Self {
176        self.entries.push((priority, Entry { provider, policy }));
177        self
178    }
179
180    /// Replaces the backoff jitter source (for deterministic tests).
181    pub fn jitter(mut self, jitter: Jitter) -> Self {
182        self.jitter = jitter;
183        self
184    }
185
186    /// Caches provider answers (and serves them stale if a provider fails).
187    pub fn cache(mut self, cache: Arc<dyn Cache>) -> Self {
188        self.cache = Some(cache);
189        self
190    }
191
192    pub fn cache_policy(mut self, policy: CachePolicy) -> Self {
193        self.cache_policy = policy;
194        self
195    }
196
197    /// Replaces the clock (for deterministic cache-expiry tests).
198    pub fn clock(mut self, clock: Clock) -> Self {
199        self.clock = clock;
200        self
201    }
202
203    pub fn build(mut self) -> Engine {
204        self.entries.sort_by_key(|(priority, _)| *priority);
205        Engine {
206            entries: self.entries.into_iter().map(|(_, e)| e).collect(),
207            jitter: self.jitter,
208            cache: self.cache,
209            cache_policy: self.cache_policy,
210            clock: self.clock,
211            options: CallOptions::default(),
212            inflight: Arc::default(),
213            health: Arc::default(),
214            budgets: Arc::default(),
215        }
216    }
217}
218
219#[derive(Debug, Clone, thiserror::Error)]
220#[non_exhaustive]
221pub enum EngineError {
222    #[error("no enabled provider is named {0:?}")]
223    UnknownProvider(String),
224    #[error("no enabled provider supports {0}")]
225    NoCapableProvider(&'static str),
226    #[error("{provider}: {error}")]
227    Provider {
228        provider: ProviderId,
229        #[source]
230        error: Error,
231    },
232}
233
234impl EngineError {
235    pub fn category(&self) -> ErrorCategory {
236        match self {
237            EngineError::UnknownProvider(_) => ErrorCategory::Usage,
238            EngineError::NoCapableProvider(_) => ErrorCategory::Unsupported,
239            EngineError::Provider { error, .. } => error.category(),
240        }
241    }
242
243    pub fn kind(&self) -> ErrorKind {
244        match self {
245            EngineError::UnknownProvider(_) => ErrorKind::BadRequest,
246            EngineError::NoCapableProvider(_) => ErrorKind::Unsupported,
247            EngineError::Provider { error, .. } => error.kind(),
248        }
249    }
250}
251
252impl Engine {
253    pub fn builder() -> EngineBuilder {
254        EngineBuilder {
255            entries: Vec::new(),
256            jitter: random_jitter,
257            cache: None,
258            cache_policy: CachePolicy::default(),
259            clock: Timestamp::now,
260        }
261    }
262
263    /// This engine for a caller with different patience; it shares
264    /// everything else.
265    pub fn with_options(&self, options: CallOptions) -> Engine {
266        Engine {
267            options,
268            ..self.clone()
269        }
270    }
271
272    pub fn options(&self) -> CallOptions {
273        self.options
274    }
275
276    /// Providers refusing calls now, and why: a spent quota (until it
277    /// resets), rejected credentials, or an open circuit breaker.
278    pub fn unavailable(&self) -> Vec<(ProviderId, Error)> {
279        self.health.unavailable((self.clock)())
280    }
281
282    /// Each provider's budget as its last response reported it: what's
283    /// left without spending any. Only providers with budgets (OpenAlex)
284    /// report one, and only once called.
285    pub fn budgets(&self) -> Vec<(ProviderId, Budget)> {
286        self.budgets.all()
287    }
288
289    /// Runs one provider operation with retries, unless the provider is
290    /// refusing calls, and remembers how it went.
291    async fn call<T, F, Fut>(&self, entry: &Entry, op: F) -> Result<Response<T>, Error>
292    where
293        F: FnMut() -> Fut,
294        Fut: std::future::Future<Output = Result<Response<T>, Error>>,
295    {
296        let provider = entry.provider.id();
297        let pass = self.health.admit(&provider, (self.clock)())?;
298        let policy = self.options.apply(&entry.policy);
299        let result = with_retries(&policy, self.jitter, op).await;
300        let now = (self.clock)();
301        pass.record(&entry.policy, result.as_ref().err(), now);
302        let usage = result.as_ref().ok().and_then(|r| r.usage.as_ref());
303        self.budgets
304            .note(&provider, usage, result.as_ref().err(), now);
305        result
306    }
307
308    /// Registered providers, in priority order.
309    pub fn providers(&self) -> impl Iterator<Item = &Arc<dyn Provider>> {
310        self.entries.iter().map(|e| &e.provider)
311    }
312}
313
314fn add_usage(totals: &mut Vec<(ProviderId, Usage)>, provider: &ProviderId, usage: Usage) {
315    match totals.iter_mut().find(|(p, _)| p == provider) {
316        Some((_, total)) => *total = total.then(usage),
317        None => totals.push((provider.clone(), usage)),
318    }
319}
aspero_engine/lib.rs

aspero_engine/
lib.rs
