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

aspero_engine/
lib.rs
