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//! Not yet: fan-out search, the circuit breaker and the cross-process
24//! limiter.
25
26mod abstracts;
27mod check;
28mod resolve;
29pub mod retry;
30mod standard;
31pub mod transport;
32
33pub use abstracts::{Abstract, AbstractsReport};
34pub use check::{Checked, Finding, Severity, duplicates, findings};
35pub use resolve::{
36    Candidate, ResolveReport, Sighting, Verdict, Verification, Via, describe_update,
37};
38/// Re-exported for credentials in provider configs.
39pub use secrecy::SecretString;
40#[cfg(feature = "crossref")]
41pub use standard::CrossrefConfig;
42#[cfg(feature = "datacite")]
43pub use standard::DataciteConfig;
44#[cfg(feature = "doi")]
45pub use standard::DoiConfig;
46#[cfg(feature = "europepmc")]
47pub use standard::EuropePmcConfig;
48#[cfg(feature = "openalex")]
49pub use standard::OpenAlexConfig;
50#[cfg(feature = "openlibrary")]
51pub use standard::OpenLibraryConfig;
52pub use standard::{StandardConfig, standard};
53
54use std::sync::Arc;
55use std::time::Duration;
56
57use aspero_core::{
58    Error, ErrorCategory, FieldSet, Filter, Page, Provider, ProviderId, ResiliencePolicy,
59    SearchMode, SearchQuery, SourceRecord, Usage, WorkId,
60};
61pub use aspero_store::{Cache, SqliteCache, Stats as CacheStats, StoreError};
62use jiff::Timestamp;
63
64use crate::retry::{Jitter, random_jitter, with_retries};
65
66/// A clock, injectable so cache expiry is testable.
67pub type Clock = fn() -> Timestamp;
68
69/// How long cached answers are trusted (DESIGN §5). Past these, a provider
70/// is asked again; if it then fails, the old answer is served as stale.
71#[derive(Debug, Clone, Copy, PartialEq, Eq)]
72#[non_exhaustive]
73pub struct CachePolicy {
74    pub records: Duration,
75    /// "Not found" answers: short, since works get registered.
76    pub misses: Duration,
77    pub searches: Duration,
78}
79
80impl Default for CachePolicy {
81    fn default() -> Self {
82        const DAY: u64 = 86_400;
83        CachePolicy {
84            records: Duration::from_secs(7 * DAY),
85            misses: Duration::from_secs(DAY),
86            searches: Duration::from_secs(DAY),
87        }
88    }
89}
90
91/// Where an answer came from.
92#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93#[non_exhaustive]
94pub enum Served {
95    Live,
96    /// A cached answer within its TTL.
97    Cache,
98    /// An expired cached answer, served because the provider failed.
99    Stale,
100}
101
102impl Served {
103    pub fn as_str(self) -> &'static str {
104        match self {
105            Served::Live => "live",
106            Served::Cache => "cache",
107            Served::Stale => "stale",
108        }
109    }
110}
111
112pub struct Engine {
113    /// In priority order.
114    entries: Vec<Entry>,
115    jitter: Jitter,
116    cache: Option<Arc<dyn Cache>>,
117    cache_policy: CachePolicy,
118    clock: Clock,
119}
120
121struct Entry {
122    provider: Arc<dyn Provider>,
123    policy: ResiliencePolicy,
124}
125
126pub struct EngineBuilder {
127    entries: Vec<(i32, Entry)>,
128    jitter: Jitter,
129    cache: Option<Arc<dyn Cache>>,
130    cache_policy: CachePolicy,
131    clock: Clock,
132}
133
134impl EngineBuilder {
135    /// Registers a provider. Lower `priority` is tried first; ties keep
136    /// registration order.
137    pub fn provider(
138        mut self,
139        provider: Arc<dyn Provider>,
140        policy: ResiliencePolicy,
141        priority: i32,
142    ) -> Self {
143        self.entries.push((priority, Entry { provider, policy }));
144        self
145    }
146
147    /// Replaces the backoff jitter source (for deterministic tests).
148    pub fn jitter(mut self, jitter: Jitter) -> Self {
149        self.jitter = jitter;
150        self
151    }
152
153    /// Caches provider answers (and serves them stale if a provider fails).
154    pub fn cache(mut self, cache: Arc<dyn Cache>) -> Self {
155        self.cache = Some(cache);
156        self
157    }
158
159    pub fn cache_policy(mut self, policy: CachePolicy) -> Self {
160        self.cache_policy = policy;
161        self
162    }
163
164    /// Replaces the clock (for deterministic cache-expiry tests).
165    pub fn clock(mut self, clock: Clock) -> Self {
166        self.clock = clock;
167        self
168    }
169
170    pub fn build(mut self) -> Engine {
171        self.entries.sort_by_key(|(priority, _)| *priority);
172        Engine {
173            entries: self.entries.into_iter().map(|(_, e)| e).collect(),
174            jitter: self.jitter,
175            cache: self.cache,
176            cache_policy: self.cache_policy,
177            clock: self.clock,
178        }
179    }
180}
181
182/// The outcome of one provider's attempt at one ID.
183#[derive(Debug, Clone)]
184#[non_exhaustive]
185pub enum Outcome {
186    Found,
187    NotFound,
188    Failed(Error),
189}
190
191#[derive(Debug, Clone)]
192pub struct Attempt {
193    pub provider: ProviderId,
194    pub outcome: Outcome,
195    pub served: Served,
196}
197
198#[derive(Debug, Clone)]
199pub struct LookupResult {
200    /// As requested.
201    pub id: WorkId,
202    pub record: Option<SourceRecord>,
203    /// Every provider tried, in order. Empty: no enabled provider accepts
204    /// this kind of identifier.
205    pub attempts: Vec<Attempt>,
206}
207
208impl LookupResult {
209    /// No provider found it, and at least one could answer authoritatively
210    /// (rather than every attempt failing).
211    pub fn is_not_found(&self) -> bool {
212        self.record.is_none()
213            && self
214                .attempts
215                .iter()
216                .any(|a| matches!(a.outcome, Outcome::NotFound))
217    }
218}
219
220#[derive(Debug, Clone)]
221pub struct LookupReport {
222    /// One per requested ID, in request order.
223    pub results: Vec<LookupResult>,
224    /// Combined usage per provider that reported any.
225    pub usage: Vec<(ProviderId, Usage)>,
226}
227
228#[derive(Debug, Clone)]
229pub struct SearchReport {
230    pub provider: ProviderId,
231    pub page: Page,
232    /// Query filters the provider can't apply; they were not sent.
233    pub ignored_filters: Vec<Filter>,
234    pub usage: Option<Usage>,
235    pub served: Served,
236}
237
238#[derive(Debug, Clone, thiserror::Error)]
239#[non_exhaustive]
240pub enum EngineError {
241    #[error("no enabled provider is named {0:?}")]
242    UnknownProvider(String),
243    #[error("no enabled provider supports {0}")]
244    NoCapableProvider(&'static str),
245    #[error("{provider}: {error}")]
246    Provider {
247        provider: ProviderId,
248        #[source]
249        error: Error,
250    },
251}
252
253impl EngineError {
254    pub fn category(&self) -> ErrorCategory {
255        match self {
256            EngineError::UnknownProvider(_) => ErrorCategory::Usage,
257            EngineError::NoCapableProvider(_) => ErrorCategory::Unsupported,
258            EngineError::Provider { error, .. } => error.category(),
259        }
260    }
261}
262
263impl Engine {
264    pub fn builder() -> EngineBuilder {
265        EngineBuilder {
266            entries: Vec::new(),
267            jitter: random_jitter,
268            cache: None,
269            cache_policy: CachePolicy::default(),
270            clock: Timestamp::now,
271        }
272    }
273
274    /// Registered providers, in priority order.
275    pub fn providers(&self) -> impl Iterator<Item = &Arc<dyn Provider>> {
276        self.entries.iter().map(|e| &e.provider)
277    }
278
279    /// Looks up every ID, falling through providers by priority. Never fails
280    /// as a whole: per-ID outcomes are in the report.
281    pub async fn lookup(&self, ids: &[WorkId], fields: FieldSet) -> LookupReport {
282        let mut results: Vec<LookupResult> = ids
283            .iter()
284            .map(|id| LookupResult {
285                id: id.clone(),
286                record: None,
287                attempts: Vec::new(),
288            })
289            .collect();
290        let mut usage: Vec<(ProviderId, Usage)> = Vec::new();
291
292        for entry in &self.entries {
293            let caps = entry.provider.capabilities();
294            let eligible: Vec<usize> = (0..results.len())
295                .filter(|&i| results[i].record.is_none())
296                .filter(|&i| caps.lookup_kinds.contains(&results[i].id.kind()))
297                .collect();
298            let ids: Vec<WorkId> = eligible.iter().map(|&i| results[i].id.clone()).collect();
299            let answers = self.lookup_at(entry, &ids, fields, &mut usage).await;
300            for (&i, (attempt, record)) in eligible.iter().zip(answers) {
301                results[i].record = record;
302                results[i].attempts.push(attempt);
303            }
304        }
305        LookupReport { results, usage }
306    }
307
308    /// Asks one provider about `ids` (all of kinds it accepts): fresh cached
309    /// answers first, the rest in batches with retries, and stale cached
310    /// records for any that fail. One answer per ID, in order.
311    async fn lookup_at(
312        &self,
313        entry: &Entry,
314        ids: &[WorkId],
315        fields: FieldSet,
316        usage: &mut Vec<(ProviderId, Usage)>,
317    ) -> Vec<(Attempt, Option<SourceRecord>)> {
318        let provider = entry.provider.id();
319        let caps = entry.provider.capabilities();
320        let attempt = |outcome, served| Attempt {
321            provider: provider.clone(),
322            outcome,
323            served,
324        };
325        let mut answers: Vec<Option<(Attempt, Option<SourceRecord>)>> = vec![None; ids.len()];
326
327        // Fresh cached answers first (one cache read for the whole set).
328        let mut pending = Vec::new();
329        let mut from_cache = 0;
330        for (i, cached) in self
331            .cached_lookups(&provider, ids, fields)
332            .into_iter()
333            .enumerate()
334        {
335            answers[i] = match cached {
336                Some(Some(record)) => {
337                    from_cache += 1;
338                    Some((attempt(Outcome::Found, Served::Cache), Some(record)))
339                }
340                Some(None) => {
341                    from_cache += 1;
342                    Some((attempt(Outcome::NotFound, Served::Cache), None))
343                }
344                None => {
345                    pending.push(i);
346                    None
347                }
348            };
349        }
350
351        if from_cache > 0 {
352            tracing::info!(target: "aspero::cache", provider = %provider, ids = from_cache, "lookups served from cache");
353        }
354        let mut chunks = pending.chunks(caps.max_batch.max(1));
355        while let Some(chunk) = chunks.next() {
356            let batch: Vec<WorkId> = chunk.iter().map(|&i| ids[i].clone()).collect();
357            let answer = with_retries(&entry.policy, self.jitter, || {
358                entry.provider.lookup(&batch, fields)
359            })
360            .await;
361
362            let error = match answer {
363                Ok(resp) => {
364                    if let Some(u) = resp.usage {
365                        add_usage(usage, &provider, u);
366                    }
367                    // The contract is one answer per ID, in order; anything
368                    // else is a provider bug, not a miss.
369                    let in_order = resp.value.len() == batch.len()
370                        && resp.value.iter().zip(&batch).all(|(l, id)| &l.id == id);
371                    if in_order {
372                        let stored: Vec<_> = resp
373                            .value
374                            .iter()
375                            .map(|l| (&l.id, l.record.as_ref()))
376                            .collect();
377                        self.remember_lookups(&provider, &stored);
378                        for (&i, lookup) in chunk.iter().zip(resp.value) {
379                            let outcome = match lookup.record {
380                                Some(_) => Outcome::Found,
381                                None => Outcome::NotFound,
382                            };
383                            answers[i] = Some((attempt(outcome, Served::Live), lookup.record));
384                        }
385                        continue;
386                    }
387                    Error::decode(ContractViolation)
388                }
389                Err(error) => error,
390            };
391
392            // Out of budget or credentials: don't spend more calls on this
393            // provider in this lookup.
394            let give_up = matches!(
395                error.category(),
396                ErrorCategory::Throttled | ErrorCategory::Auth
397            );
398            let affected: Vec<usize> = if give_up {
399                chunk
400                    .iter()
401                    .chain(chunks.by_ref().flatten())
402                    .copied()
403                    .collect()
404            } else {
405                chunk.to_vec()
406            };
407            // Stale-on-error: an old answer beats none.
408            let affected_ids: Vec<WorkId> = affected.iter().map(|&i| ids[i].clone()).collect();
409            for (&i, stale) in affected
410                .iter()
411                .zip(self.stale_lookups(&provider, &affected_ids))
412            {
413                answers[i] = Some(match stale {
414                    Some(record) => (attempt(Outcome::Found, Served::Stale), Some(record)),
415                    None => (attempt(Outcome::Failed(error.clone()), Served::Live), None),
416                });
417            }
418        }
419        answers
420            .into_iter()
421            .map(|a| a.expect("every ID answered from cache, provider or failure"))
422            .collect()
423    }
424
425    /// Searches one provider: `source` if given, otherwise the first that
426    /// supports the query's mode.
427    pub async fn search(
428        &self,
429        source: Option<&str>,
430        query: &SearchQuery,
431    ) -> Result<SearchReport, EngineError> {
432        let citation = query.mode == SearchMode::Citation;
433        let entry = match source {
434            Some(name) => self
435                .entries
436                .iter()
437                .find(|e| e.provider.id().as_str() == name)
438                .ok_or_else(|| EngineError::UnknownProvider(name.to_owned()))?,
439            None => self
440                .entries
441                .iter()
442                .find(|e| {
443                    let caps = e.provider.capabilities();
444                    caps.search && (!citation || caps.citation_search)
445                })
446                .ok_or(EngineError::NoCapableProvider(if citation {
447                    "citation search"
448                } else {
449                    "search"
450                }))?,
451        };
452        let provider = entry.provider.id();
453        let caps = entry.provider.capabilities();
454        let refuse = |what| EngineError::Provider {
455            provider: provider.clone(),
456            error: Error::Unsupported(what),
457        };
458        if !caps.search {
459            return Err(refuse("search"));
460        }
461        if citation && !caps.citation_search {
462            return Err(refuse("citation search"));
463        }
464
465        let ignored_filters: Vec<Filter> = query
466            .filters()
467            .into_iter()
468            .filter(|f| !caps.search_filters.contains(f))
469            .collect();
470        let mut sent = query.clone();
471        for filter in &ignored_filters {
472            match filter {
473                Filter::Years => sent.years = None,
474                Filter::WorkType => sent.work_type = None,
475                Filter::OpenAccess => sent.open_access = None,
476                _ => {}
477            }
478        }
479
480        let report = |page, usage, served| SearchReport {
481            provider: provider.clone(),
482            page,
483            ignored_filters: ignored_filters.clone(),
484            usage,
485            served,
486        };
487        let cached = self.cached_search(&provider, &sent);
488        if let Some((page, true)) = cached {
489            tracing::info!(target: "aspero::cache", provider = %provider, text = %sent.text, "search served from cache");
490            return Ok(report(page, None, Served::Cache));
491        }
492        match with_retries(&entry.policy, self.jitter, || entry.provider.search(&sent)).await {
493            Ok(resp) => {
494                self.remember_search(&provider, &sent, &resp.value);
495                Ok(report(resp.value, resp.usage, Served::Live))
496            }
497            Err(error) => match cached {
498                Some((page, _)) => Ok(report(page, None, Served::Stale)),
499                None => Err(EngineError::Provider {
500                    provider: provider.clone(),
501                    error,
502                }),
503            },
504        }
505    }
506
507    // Cache access. Cache failures are never fatal: a broken cache behaves
508    // like an empty one, and failed writes are skipped.
509
510    /// For each ID, a fresh cached answer covering `fields`:
511    /// `Some(Some(record))`, `Some(None)` for a fresh "not found", or `None`
512    /// to ask the provider.
513    fn cached_lookups(
514        &self,
515        provider: &ProviderId,
516        ids: &[WorkId],
517        fields: FieldSet,
518    ) -> Vec<Option<Option<SourceRecord>>> {
519        let now = (self.clock)();
520        let policy = self.cache_policy;
521        self.cache_entries(provider, ids)
522            .into_iter()
523            .map(|entry| {
524                let entry = entry?;
525                match entry.value {
526                    Some(ref record)
527                        if entry.is_fresh(now, policy.records)
528                            && fields.is_subset(record.fields) =>
529                    {
530                        Some(entry.value)
531                    }
532                    None if entry.is_fresh(now, policy.misses) => Some(None),
533                    _ => None,
534                }
535            })
536            .collect()
537    }
538
539    /// For each ID, any cached record, however old.
540    fn stale_lookups(&self, provider: &ProviderId, ids: &[WorkId]) -> Vec<Option<SourceRecord>> {
541        self.cache_entries(provider, ids)
542            .into_iter()
543            .map(|entry| entry.and_then(|e| e.value))
544            .collect()
545    }
546
547    fn cache_entries(
548        &self,
549        provider: &ProviderId,
550        ids: &[WorkId],
551    ) -> Vec<Option<aspero_store::CachedLookup>> {
552        self.cache
553            .as_ref()
554            .and_then(|c| c.get_lookups(provider, ids).ok())
555            .unwrap_or_else(|| vec![None; ids.len()])
556    }
557
558    fn remember_lookups(
559        &self,
560        provider: &ProviderId,
561        answers: &[(&WorkId, Option<&SourceRecord>)],
562    ) {
563        if let Some(cache) = &self.cache {
564            let _ = cache.put_lookups(provider, answers, (self.clock)());
565        }
566    }
567
568    /// A cached page and whether it is still fresh.
569    fn cached_search(&self, provider: &ProviderId, query: &SearchQuery) -> Option<(Page, bool)> {
570        let entry = self.cache.as_ref()?.get_search(provider, query).ok()??;
571        let fresh = entry.is_fresh((self.clock)(), self.cache_policy.searches);
572        Some((entry.value, fresh))
573    }
574
575    fn remember_search(&self, provider: &ProviderId, query: &SearchQuery, page: &Page) {
576        if let Some(cache) = &self.cache {
577            let _ = cache.put_search(provider, query, page, (self.clock)());
578        }
579    }
580}
581
582fn add_usage(totals: &mut Vec<(ProviderId, Usage)>, provider: &ProviderId, usage: Usage) {
583    match totals.iter_mut().find(|(p, _)| p == provider) {
584        Some((_, total)) => *total = total.then(usage),
585        None => totals.push((provider.clone(), usage)),
586    }
587}
588
589#[derive(Debug, thiserror::Error)]
590#[error("provider answered a batch lookup out of order or incompletely")]
591struct ContractViolation;
aspero_engine/lib.rs

aspero_engine/
lib.rs
