lunaris_retrieve/
service.rs

1//! `tower::Service<Query, Response = Vec<Hit>>` adapter.
2//!
3//! The retriever IS a tower service per blueprint §8 — composing with
4//! `ServiceBuilder::new().rate_limit(..).timeout(..).retry(..).service(retriever)`
5//! gives the entire ecosystem of tower middleware for free.
6//!
7//! ## Construction
8//!
9//! ```no_run
10//! use std::sync::Arc;
11//! use std::time::Duration;
12//!
13//! use lunaris_core::{Embedder, KeywordPort, Scope, StoragePort};
14//! use lunaris_retrieve::{Query, RetrievalService, Retriever};
15//! use tower::{Service, ServiceBuilder, ServiceExt};
16//!
17//! # async fn demo(
18//! #     root: Arc<dyn Retriever>,
19//! #     embedder: Arc<dyn Embedder>,
20//! #     storage: Arc<dyn StoragePort>,
21//! #     keyword: Arc<dyn KeywordPort>,
22//! #     scope: Scope,
23//! # ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
24//! let svc = RetrievalService::new(root, embedder, storage, keyword, scope);
25//! let mut wrapped = ServiceBuilder::new()
26//!     .rate_limit(10, Duration::from_secs(1))
27//!     .timeout(Duration::from_secs(5))
28//!     .service(svc);
29//! let hits = wrapped.ready().await?.call(Query::text("foo")).await?;
30//! # Ok(()) }
31//! ```
32
33use std::future::Future;
34use std::pin::Pin;
35use std::sync::Arc;
36use std::task::{Context, Poll};
37
38use lunaris_core::{Embedder, KeywordPort, LunarisError, Scope, StoragePort};
39use tower::Service;
40
41use crate::hydrate::hydrate;
42use crate::operators::{QueryContext, Retriever};
43use crate::types::{Hit, Query};
44
45/// Boxed future for the tower::Service impl. We use a `BoxFuture` rather than
46/// a named future type because Service futures must be `'static + Send` and
47/// the inner retriever's future is opaque.
48type BoxFuture = Pin<Box<dyn Future<Output = Result<Vec<Hit>, LunarisError>> + Send + 'static>>;
49
50/// `tower::Service` adapter wrapping a built operator tree.
51///
52/// Cloneable — every clone shares the same `Arc<dyn Retriever>` so wrapping
53/// in a `Buffer` / `Limit` layer is cheap.
54///
55/// RFC 0001 Wave 2: every constructed service carries a [`Scope`] that
56/// scopes the per-call retrieval. `RetrievalService::new` now requires
57/// the scope explicitly; the v0.2 `Scope::dev()` default is gone.
58#[derive(Clone)]
59pub struct RetrievalService {
60    pub(crate) root: Arc<dyn Retriever>,
61    pub(crate) embedder: Arc<dyn Embedder>,
62    pub(crate) storage: Arc<dyn StoragePort>,
63    pub(crate) keyword: Arc<dyn KeywordPort>,
64    pub(crate) scope: Scope,
65}
66
67impl RetrievalService {
68    pub fn new(
69        root: Arc<dyn Retriever>,
70        embedder: Arc<dyn Embedder>,
71        storage: Arc<dyn StoragePort>,
72        keyword: Arc<dyn KeywordPort>,
73        scope: Scope,
74    ) -> Self {
75        Self { root, embedder, storage, keyword, scope }
76    }
77}
78
79impl Service<Query> for RetrievalService {
80    type Response = Vec<Hit>;
81    type Error = LunarisError;
82    type Future = BoxFuture;
83
84    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
85        // Always ready — backpressure happens inside the operator tree (vector
86        // search timeout / keyword search timeout / etc.). Tower's middleware
87        // (rate_limit / concurrency_limit) wraps THIS service for shared
88        // backpressure across callers.
89        Poll::Ready(Ok(()))
90    }
91
92    fn call(&mut self, q: Query) -> Self::Future {
93        let storage = self.storage.clone();
94        let embedder = self.embedder.clone();
95        let keyword = self.keyword.clone();
96        let root = self.root.clone();
97        let scope = self.scope.clone();
98        let as_of = q.as_of;
99
100        Box::pin(async move {
101            // P0 #1 Wave 2: scope is carried on the service itself — every
102            // call inherits the constructor-supplied partition key.
103            let ctx = QueryContext::new(q, scope, embedder, storage.clone(), keyword);
104            let raw = root.retrieve(&ctx).await?;
105            // Plan 04-04 B-9: RetrievalService callers don't have a verifier
106            // queue-depth check (only `Lunaris::recall_with_degraded_check`
107            // does), so initial_degraded is unconditionally false here.
108            hydrate(storage.as_ref(), &ctx.scope, raw, as_of, false).await
109        })
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use super::*;
116    use std::any::TypeId;
117
118    #[test]
119    fn retrieval_service_is_send_sync() {
120        fn assert_send_sync<T: Send + Sync>() {}
121        assert_send_sync::<RetrievalService>();
122    }
123
124    #[test]
125    fn future_type_is_static_send() {
126        // Sanity: future type is `'static + Send` per Service contract.
127        // TypeId comparison is a compile-time-ish proxy.
128        let _ = TypeId::of::<BoxFuture>();
129    }
130}
lunaris_retrieve/service.rs

lunaris_retrieve/
service.rs
