saddle_service/
registry.rs

1use std::{
2    any::{Any, TypeId},
3    collections::HashSet,
4    sync::Arc,
5};
6
7use saddle_core::{CallContext, ErrorKind, ModuleId, OperationId, Result, SaddleError, ServiceId};
8use saddle_observability::{CallKind, Observer};
9
10use crate::{Service, ServiceHandler};
11
12#[derive(Clone, Debug, Eq, PartialEq)]
13pub struct ServiceDescriptor {
14    module: ModuleId,
15    service: ServiceId,
16    operation: OperationId,
17}
18
19impl ServiceDescriptor {
20    pub fn new(
21        module: impl Into<ModuleId>,
22        service: impl Into<ServiceId>,
23        operation: impl Into<OperationId>,
24    ) -> Self {
25        Self {
26            module: module.into(),
27            service: service.into(),
28            operation: operation.into(),
29        }
30    }
31
32    pub fn module(&self) -> &ModuleId {
33        &self.module
34    }
35
36    pub fn service(&self) -> &ServiceId {
37        &self.service
38    }
39
40    pub fn operation(&self) -> &OperationId {
41        &self.operation
42    }
43
44    fn identity(&self) -> ServiceIdentity {
45        ServiceIdentity {
46            module: self.module.to_string(),
47            service: self.service.to_string(),
48            operation: self.operation.to_string(),
49        }
50    }
51}
52
53#[derive(Clone, Debug, Eq, Hash, PartialEq)]
54struct ServiceIdentity {
55    module: String,
56    service: String,
57    operation: String,
58}
59
60struct Entry {
61    contract: TypeId,
62    descriptor: ServiceDescriptor,
63    implementation: Arc<dyn Any + Send + Sync>,
64}
65
66struct TypedImplementation<S: Service> {
67    handler: Arc<dyn ServiceHandler<S>>,
68}
69
70type Factory = Box<dyn FnOnce(&ServiceResolver<'_>) -> Result<Arc<dyn Any + Send + Sync>> + Send>;
71
72struct Definition {
73    contract: TypeId,
74    descriptor: ServiceDescriptor,
75    factory: Factory,
76}
77
78/// Builds an immutable process-local registry in dependency order.
79///
80/// [`register_with`](Self::register_with) is the standard dependency injection
81/// path. Its resolver only sees already constructed Services, so missing,
82/// reversed, and cyclic dependencies fail during `build`, never on a request.
83pub struct ServiceRegistryBuilder {
84    definitions: Vec<Definition>,
85    contracts: HashSet<TypeId>,
86    identities: HashSet<ServiceIdentity>,
87}
88
89impl ServiceRegistryBuilder {
90    pub fn new() -> Self {
91        Self {
92            definitions: Vec::new(),
93            contracts: HashSet::new(),
94            identities: HashSet::new(),
95        }
96    }
97
98    pub fn register<S, H>(&mut self, descriptor: ServiceDescriptor, handler: H) -> Result<()>
99    where
100        S: Service,
101        H: ServiceHandler<S>,
102    {
103        self.register_with::<S, H, _>(descriptor, move |_| Ok(handler))
104    }
105
106    /// Registers a factory whose actual client resolutions define dependencies.
107    pub fn register_with<S, H, F>(
108        &mut self,
109        descriptor: ServiceDescriptor,
110        factory: F,
111    ) -> Result<()>
112    where
113        S: Service,
114        H: ServiceHandler<S>,
115        F: FnOnce(&ServiceResolver<'_>) -> Result<H> + Send + 'static,
116    {
117        let contract = TypeId::of::<S>();
118        if !self.contracts.insert(contract) {
119            return Err(registration_error(
120                "service.duplicate_contract",
121                "the Service contract is already registered",
122            ));
123        }
124        if !self.identities.insert(descriptor.identity()) {
125            self.contracts.remove(&contract);
126            return Err(registration_error(
127                "service.duplicate_identity",
128                "the Service identity is already registered",
129            ));
130        }
131
132        self.definitions.push(Definition {
133            contract,
134            descriptor,
135            factory: Box::new(move |resolver| {
136                let handler = factory(resolver)?;
137                Ok(Arc::new(TypedImplementation::<S> {
138                    handler: Arc::new(handler),
139                }))
140            }),
141        });
142        Ok(())
143    }
144
145    /// Constructs Services in registration order and freezes the registry.
146    pub fn build(self, observer: Observer) -> Result<ServiceRegistry> {
147        let mut entries = Vec::with_capacity(self.definitions.len());
148        for definition in self.definitions {
149            let implementation = {
150                let resolver = ServiceResolver {
151                    entries: &entries,
152                    observer: &observer,
153                };
154                (definition.factory)(&resolver)?
155            };
156            entries.push(Entry {
157                contract: definition.contract,
158                descriptor: definition.descriptor,
159                implementation,
160            });
161        }
162        Ok(ServiceRegistry {
163            inner: Arc::new(RegistryInner { entries }),
164            observer,
165        })
166    }
167}
168
169impl Default for ServiceRegistryBuilder {
170    fn default() -> Self {
171        Self::new()
172    }
173}
174
175/// Dependency resolver available only while a Service factory is constructed.
176pub struct ServiceResolver<'a> {
177    entries: &'a [Entry],
178    observer: &'a Observer,
179}
180
181impl ServiceResolver<'_> {
182    pub fn client<S>(&self) -> Result<ServiceClient<S>>
183    where
184        S: Service,
185    {
186        client_from_entries(self.entries, self.observer.clone())
187    }
188}
189
190struct RegistryInner {
191    entries: Vec<Entry>,
192}
193
194#[derive(Clone)]
195pub struct ServiceRegistry {
196    inner: Arc<RegistryInner>,
197    observer: Observer,
198}
199
200impl ServiceRegistry {
201    pub fn client<S>(&self) -> Result<ServiceClient<S>>
202    where
203        S: Service,
204    {
205        client_from_entries(&self.inner.entries, self.observer.clone())
206    }
207
208    pub(crate) fn observer(&self) -> Observer {
209        self.observer.clone()
210    }
211}
212
213fn client_from_entries<S>(entries: &[Entry], observer: Observer) -> Result<ServiceClient<S>>
214where
215    S: Service,
216{
217    let entry = entries
218        .iter()
219        .find(|entry| entry.contract == TypeId::of::<S>())
220        .ok_or_else(missing_dependency)?;
221    let implementation = Arc::clone(&entry.implementation)
222        .downcast::<TypedImplementation<S>>()
223        .map_err(|_| {
224            SaddleError::new(
225                ErrorKind::Internal,
226                "service.registry_corrupt",
227                "the Service registry contains an invalid implementation",
228            )
229        })?;
230    Ok(ServiceClient {
231        descriptor: entry.descriptor.clone(),
232        implementation: Arc::clone(&implementation.handler),
233        observer,
234    })
235}
236
237pub struct ServiceClient<S: Service> {
238    descriptor: ServiceDescriptor,
239    implementation: Arc<dyn ServiceHandler<S>>,
240    observer: Observer,
241}
242
243impl<S: Service> Clone for ServiceClient<S> {
244    fn clone(&self) -> Self {
245        Self {
246            descriptor: self.descriptor.clone(),
247            implementation: Arc::clone(&self.implementation),
248            observer: self.observer.clone(),
249        }
250    }
251}
252
253impl<S: Service> ServiceClient<S> {
254    pub fn descriptor(&self) -> &ServiceDescriptor {
255        &self.descriptor
256    }
257
258    pub async fn call(&self, parent: &CallContext, request: S::Request) -> Result<S::Response> {
259        let call = self.observer.start_child_call(
260            parent,
261            CallKind::Service,
262            self.descriptor.module().clone(),
263            self.descriptor.service().clone(),
264            self.descriptor.operation().clone(),
265        );
266        let result = self.implementation.call(call.context(), request).await;
267        match &result {
268            Ok(_) => call.succeed(),
269            Err(error) => call.fail(error),
270        }
271        result
272    }
273}
274
275fn missing_dependency() -> SaddleError {
276    SaddleError::new(
277        ErrorKind::InvalidArgument,
278        "service.missing_dependency",
279        "a Service factory dependency is not registered before its dependent Service",
280    )
281}
282
283fn registration_error(code: &'static str, message: &'static str) -> SaddleError {
284    SaddleError::new(ErrorKind::Conflict, code, message)
285}
286
287#[cfg(test)]
288pub(crate) mod tests {
289    use std::io;
290
291    use saddle_core::{ApplicationId, SpanId, TraceId};
292    use saddle_observability::ObserverConfig;
293
294    use super::*;
295
296    pub(crate) struct Echo;
297    impl Service for Echo {
298        type Request = String;
299        type Response = String;
300    }
301
302    pub(crate) struct EchoImpl;
303    impl ServiceHandler<Echo> for EchoImpl {
304        fn call<'a>(
305            &'a self,
306            _context: &'a CallContext,
307            request: String,
308        ) -> crate::ServiceFuture<'a, String> {
309            Box::pin(async move { Ok(request) })
310        }
311    }
312
313    struct Prefix;
314    impl Service for Prefix {
315        type Request = String;
316        type Response = String;
317    }
318
319    struct PrefixImpl {
320        echo: ServiceClient<Echo>,
321    }
322
323    impl ServiceHandler<Prefix> for PrefixImpl {
324        fn call<'a>(
325            &'a self,
326            context: &'a CallContext,
327            request: String,
328        ) -> crate::ServiceFuture<'a, String> {
329            Box::pin(async move {
330                let value = self.echo.call(context, request).await?;
331                Ok(format!("prefix:{value}"))
332            })
333        }
334    }
335
336    pub(crate) fn observer() -> Observer {
337        Observer::with_writer(ObserverConfig::default(), io::sink()).unwrap()
338    }
339
340    pub(crate) fn context() -> CallContext {
341        CallContext::new(
342            ApplicationId::from("app"),
343            ModuleId::from("caller"),
344            ServiceId::from("caller"),
345            OperationId::from("call"),
346            TraceId::from_u128(1),
347            SpanId::from_u64(1),
348        )
349    }
350
351    #[test]
352    fn rejects_contract_and_identity_conflicts() {
353        let mut builder = ServiceRegistryBuilder::new();
354        let descriptor = ServiceDescriptor::new("test", "echo", "call");
355        builder
356            .register::<Echo, _>(descriptor.clone(), EchoImpl)
357            .unwrap();
358        assert_eq!(
359            builder
360                .register::<Echo, _>(descriptor, EchoImpl)
361                .unwrap_err()
362                .code(),
363            "service.duplicate_contract"
364        );
365    }
366
367    #[test]
368    fn actual_factory_resolution_rejects_missing_dependency() {
369        let mut builder = ServiceRegistryBuilder::new();
370        builder
371            .register_with::<Prefix, PrefixImpl, _>(
372                ServiceDescriptor::new("test", "prefix", "call"),
373                |resolver| {
374                    Ok(PrefixImpl {
375                        echo: resolver.client::<Echo>()?,
376                    })
377                },
378            )
379            .unwrap();
380        assert_eq!(
381            builder.build(observer()).err().unwrap().code(),
382            "service.missing_dependency"
383        );
384    }
385
386    #[tokio::test]
387    async fn factory_injects_typed_client_for_positive_service_chain() {
388        let observer = observer();
389        let mut builder = ServiceRegistryBuilder::new();
390        builder
391            .register::<Echo, _>(ServiceDescriptor::new("test", "echo", "call"), EchoImpl)
392            .unwrap();
393        builder
394            .register_with::<Prefix, PrefixImpl, _>(
395                ServiceDescriptor::new("test", "prefix", "call"),
396                |resolver| {
397                    Ok(PrefixImpl {
398                        echo: resolver.client::<Echo>()?,
399                    })
400                },
401            )
402            .unwrap();
403        let registry = builder.build(observer).unwrap();
404        let client = registry.client::<Prefix>().unwrap();
405        assert_eq!(
406            client.call(&context(), "hello".to_owned()).await.unwrap(),
407            "prefix:hello"
408        );
409    }
410}
saddle_service/registry.rs

saddle_service/
registry.rs
