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

aspero_engine/
lib.rs
