harmos_sdk/operations/
registration.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::any::Any;
18use std::future::Future;
19use std::pin::Pin;
20
21// Third Party Imports
22
23// Local Imports
24use super::{
25    Carrier, Contract, Executable, ExecutionContext, OperationDeclaration, OperationDefinition, Resource, ResourceSet, ServingError,
26    WireDeclaration, WireTable,
27};
28use crate::services::Service;
29use crate::{Message, ServicePolicy};
30
31/* === Constants === */
32
33/* === Macros === */
34
35/* === Type Aliases === */
36
37pub type ServiceFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T, ServingError>> + Send + 'a>>;
38type Install = fn(&mut WireTable, ResourceSet, Contract) -> Result<(), ServingError>;
39type InitializeService = fn(ExecutionContext) -> ServiceFuture<'static, Box<dyn ServiceInstance>>;
40type PrepareNative = fn(ResourceSet, Box<dyn Any + Send>, Box<dyn Any + Send>) -> Result<Box<dyn NativeInstance>, ServingError>;
41type PrepareEncoded = fn(ResourceSet, &[u8]) -> Result<PreparedNative, ServingError>;
42
43/* === Structs === */
44
45#[derive(Clone, Copy)]
46pub struct Catalog {
47    pub name: &'static str,
48    pub declarations: &'static [crate::catalog::Node<WireDeclaration>],
49    pub callees: &'static [crate::catalog::Node<&'static [WireDeclaration]>],
50    identity: fn() -> std::any::TypeId,
51    factory: fn() -> RegistrationBatch,
52}
53
54#[derive(Clone, Default)]
55pub struct RegistrationBatch {
56    pub operations: Vec<OperationFactory>,
57    pub services: Vec<ServiceFactory>,
58}
59
60pub struct PreparedNative {
61    #[cfg(feature = "streaming")]
62    pub streams: Option<super::WireStreams>,
63
64    pub instance: Box<dyn NativeInstance>,
65    pub grant: Option<crate::plugin::Grant>,
66    pub key: Option<super::CacheKey>,
67}
68
69#[derive(Clone)]
70pub struct OperationFactory {
71    pub definition: OperationDefinition,
72    pub operations: &'static [WireDeclaration],
73    declared: fn() -> OperationDefinition,
74    builtin: bool,
75    resources: &'static [Resource],
76    contract: fn(&harmos_domain::config::SchemaSettings) -> Result<Contract, super::ContractError>,
77    install: Install,
78    native: PrepareNative,
79    endpoints_valid: fn() -> bool,
80    native_types: fn() -> (std::any::TypeId, std::any::TypeId, std::any::TypeId),
81    encoded: PrepareEncoded,
82
83    #[cfg(all(feature = "grpc", feature = "streaming"))]
84    grpc: fn(ResourceSet, &[u8], super::WireStreamLimits) -> Result<PreparedNative, ServingError>,
85
86    #[cfg(all(feature = "grpc", feature = "streaming"))]
87    grpc_memory: fn(&[u8]) -> Result<usize, crate::MessageError>,
88
89    decoding_memory: fn(&[u8]) -> Result<usize, crate::MessageError>,
90}
91
92#[derive(Clone)]
93pub struct ServiceFactory {
94    pub local_id: String,
95    pub policy: ServicePolicy,
96    initialize: InitializeService,
97}
98
99struct InitializedService<S: Service> {
100    service: S,
101}
102
103struct NativeOperation<O: Carrier> {
104    dependencies: ResourceSet,
105    operation: Option<O>,
106    request: <O::Declaration as OperationDeclaration>::Request,
107    endpoints: O::StreamEndpoints,
108    stopped: bool,
109    encoded: bool,
110
111    #[cfg(all(feature = "grpc", feature = "streaming"))]
112    protobuf: bool,
113
114    #[cfg(feature = "streaming")]
115    streams: Option<super::WireStreams>,
116
117    #[cfg(feature = "streaming")]
118    epoch: u64,
119}
120
121/* === Enums === */
122
123/* === Traits === */
124
125pub trait OperationCatalog: 'static {
126    const CATALOG: Catalog;
127
128    #[doc(hidden)]
129    fn batch() -> RegistrationBatch;
130}
131
132pub trait ServiceInstance: Send {
133    fn run(&mut self, context: ExecutionContext) -> ServiceFuture<'_, ()>;
134    fn stop(&mut self) -> ServiceFuture<'_, ()>;
135    fn cleanup(&mut self) -> ServiceFuture<'_, ()>;
136}
137
138pub trait NativeInstance: Send {
139    fn classify(&self, error: &(dyn std::error::Error + Send + Sync + 'static)) -> crate::operations::FailureKind;
140    fn initialize(&mut self, context: ExecutionContext) -> ServiceFuture<'_, ()>;
141    fn run(&mut self, context: ExecutionContext) -> ServiceFuture<'_, super::response::ErasedResponse>;
142    fn stop(&mut self) -> ServiceFuture<'_, ()>;
143    fn cleanup(&mut self) -> ServiceFuture<'_, ()>;
144    fn transferred(&self) -> bool;
145}
146
147/* === Implementations === */
148
149impl Catalog {
150    pub const fn of<C: OperationCatalog>(
151        name: &'static str, declarations: &'static [crate::catalog::Node<WireDeclaration>],
152        callees: &'static [crate::catalog::Node<&'static [WireDeclaration]>],
153    ) -> Self {
154        Self {
155            name,
156            declarations,
157            callees,
158            identity: std::any::TypeId::of::<C>,
159            factory: C::batch,
160        }
161    }
162
163    #[doc(hidden)]
164    pub fn identity(&self) -> std::any::TypeId {
165        (self.identity)()
166    }
167
168    #[doc(hidden)]
169    pub fn batch(&self) -> RegistrationBatch {
170        (self.factory)()
171    }
172}
173
174impl RegistrationBatch {
175    pub fn new() -> Self {
176        Self::default()
177    }
178
179    pub fn operation<O: Executable>(self) -> Self {
180        self.with::<O, O::Carrier>(O::RESOURCES)
181    }
182
183    #[doc(hidden)]
184    pub fn carrier<D, O>(self) -> Self
185    where
186        D: OperationDeclaration,
187        O: Carrier<Declaration = D>,
188    {
189        self.with::<D, O>(&[])
190    }
191
192    fn with<D, O>(mut self, resources: &'static [Resource]) -> Self
193    where
194        D: OperationDeclaration,
195        O: Carrier<Declaration = D>,
196    {
197        self.operations.push(OperationFactory {
198            resources,
199            operations: D::OPERATIONS,
200            definition: D::definition(),
201            declared: D::definition,
202            builtin: std::any::TypeId::of::<D>() == std::any::TypeId::of::<crate::observers::Control>(),
203            contract: |settings| {
204                if std::any::TypeId::of::<D>() == std::any::TypeId::of::<crate::observers::Control>() {
205                    return crate::observers::contract();
206                }
207                WireDeclaration::of::<D>().contract_with_settings(settings)
208            },
209            install: WireTable::install::<D, O>,
210            native_types: || {
211                (
212                    std::any::TypeId::of::<<O::Declaration as OperationDeclaration>::Request>(),
213                    std::any::TypeId::of::<<O::Declaration as OperationDeclaration>::Response>(),
214                    std::any::TypeId::of::<O::StreamEndpoints>(),
215                )
216            },
217            endpoints_valid: || {
218                use super::EndpointSet;
219                O::StreamEndpoints::item_types() == super::item_types::<D>()
220            },
221            native: |dependencies, request, endpoints| {
222                use super::EndpointSet;
223                if O::StreamEndpoints::item_types() != super::item_types::<D>() {
224                    return Err(ServingError::Catalog);
225                }
226                let request = request.downcast::<<O::Declaration as OperationDeclaration>::Request>().map_err(|_| ServingError::Catalog)?;
227                let endpoints = endpoints.downcast::<O::StreamEndpoints>().map_err(|_| ServingError::Catalog)?;
228                Ok(Box::new(NativeOperation::<O> {
229                    dependencies,
230                    operation: None,
231                    request: *request,
232                    endpoints: *endpoints,
233                    stopped: false,
234                    encoded: false,
235                    #[cfg(all(feature = "grpc", feature = "streaming"))]
236                    protobuf: false,
237                    #[cfg(feature = "streaming")]
238                    streams: None,
239                    #[cfg(feature = "streaming")]
240                    epoch: 0,
241                }))
242            },
243            encoded: |dependencies, payload| {
244                use super::EndpointSet;
245                let endpoints = O::StreamEndpoints::empty().ok_or(ServingError::StreamUnsupported)?;
246                let request =
247                    <<O::Declaration as OperationDeclaration>::Request as crate::Message>::decode(payload).map_err(ServingError::Codec)?;
248                let grant = D::grant(&request);
249                let key = D::key(&request);
250                let instance = NativeOperation::<O> {
251                    dependencies,
252                    operation: None,
253                    request,
254                    endpoints,
255                    stopped: false,
256                    encoded: true,
257                    #[cfg(all(feature = "grpc", feature = "streaming"))]
258                    protobuf: false,
259                    #[cfg(feature = "streaming")]
260                    streams: None,
261                    #[cfg(feature = "streaming")]
262                    epoch: 0,
263                };
264                Ok(PreparedNative {
265                    instance: Box::new(instance),
266                    grant,
267                    key,
268                    #[cfg(feature = "streaming")]
269                    streams: None,
270                })
271            },
272            #[cfg(all(feature = "grpc", feature = "streaming"))]
273            grpc: |dependencies, payload, limits| {
274                use super::EndpointSet;
275                let request =
276                    crate::grpc::decode::<<O::Declaration as OperationDeclaration>::Request>(crate::grpc::DecodeInput::new(payload))
277                        .map_err(ServingError::Codec)?;
278                let (endpoints, streams) = match O::StreamEndpoints::empty() {
279                    Some(endpoints) => (endpoints, None),
280                    None => {
281                        let (endpoints, streams) = O::StreamEndpoints::bind_wire(limits)?;
282                        (endpoints, Some(streams.protobuf(super::serving::protobuf::Codec::of::<D>())?))
283                    }
284                };
285                let grant = D::grant(&request);
286                let key = D::key(&request);
287                let instance = NativeOperation::<O> {
288                    dependencies,
289                    operation: None,
290                    request,
291                    endpoints,
292                    stopped: false,
293                    encoded: true,
294                    protobuf: true,
295                    streams: streams.clone(),
296                    epoch: 0,
297                };
298                Ok(PreparedNative {
299                    instance: Box::new(instance),
300                    grant,
301                    key,
302                    streams,
303                })
304            },
305            #[cfg(all(feature = "grpc", feature = "streaming"))]
306            grpc_memory: |payload| {
307                crate::grpc::prepare::<<O::Declaration as OperationDeclaration>::Request>(crate::grpc::DecodeInput::new(payload))
308                    .map(|prepared| prepared.memory())
309            },
310            decoding_memory: <D::Request as crate::Message>::decoding_memory,
311        });
312        self
313    }
314
315    pub fn service<S: Service>(mut self, local_id: &str, policy: ServicePolicy) -> Self {
316        self.services.push(ServiceFactory {
317            local_id: local_id.to_owned(),
318            policy,
319            initialize: |context| {
320                Box::pin(async move {
321                    let service = S::initialize(context).await.map_err(service_error)?;
322                    Ok(Box::new(InitializedService { service }) as Box<dyn ServiceInstance>)
323                })
324            },
325        });
326        self
327    }
328}
329
330impl OperationFactory {
331    pub fn accepts_target(&self, target: &super::Target) -> bool {
332        (self.native_types)() == ((target.request)(), (target.response)(), std::any::TypeId::of::<()>())
333    }
334
335    pub fn accepts_native<D: OperationDeclaration, P: 'static>(&self) -> bool {
336        (self.native_types)() == (std::any::TypeId::of::<D::Request>(), std::any::TypeId::of::<D::Response>(), std::any::TypeId::of::<P>())
337    }
338
339    // Every Declared Resource Comes From The Snapshot Or The Carrier's Execution Context.
340    pub fn verify_resources(&self, resources: &ResourceSet, provided: &[std::any::TypeId]) -> Result<(), ServingError> {
341        let missing = self.resources.iter().find(|resource| !resources.contains(resource.id()) && !provided.contains(&resource.id()));
342        match missing {
343            Some(resource) => Err(ServingError::Resource {
344                operation: self.definition.id.to_string(),
345                resource: resource.name(),
346            }),
347            None => Ok(()),
348        }
349    }
350
351    pub fn validate(&self) -> Result<(), ServingError> {
352        if !self.builtin {
353            self.definition.validate().map_err(ServingError::from)?;
354        }
355        if self.definition != (self.declared)() {
356            return Err(ServingError::Catalog);
357        }
358        if !(self.endpoints_valid)() {
359            return Err(ServingError::EndpointTypes);
360        }
361        Ok(())
362    }
363
364    #[doc(hidden)]
365    pub fn missing_resources(&self, resources: &ResourceSet, provided: &[std::any::TypeId]) -> Vec<super::StartupError> {
366        self.resources
367            .iter()
368            .filter(|resource| !resources.contains(resource.id()) && !provided.contains(&resource.id()))
369            .map(|resource| super::StartupError::MissingResource {
370                operation: self.definition.id.to_string(),
371                resource: resource.name(),
372            })
373            .collect()
374    }
375
376    pub fn decoding_memory(&self, payload: &[u8]) -> Result<usize, crate::MessageError> {
377        (self.decoding_memory)(payload)
378    }
379
380    pub fn prepare_encoded(&self, resources: ResourceSet, payload: &[u8]) -> Result<PreparedNative, ServingError> {
381        (self.encoded)(resources, payload)
382    }
383
384    #[cfg(all(feature = "grpc", feature = "streaming"))]
385    pub fn prepare_grpc(
386        &self, resources: ResourceSet, payload: &[u8], limits: super::WireStreamLimits,
387    ) -> Result<PreparedNative, ServingError> {
388        (self.grpc)(resources, payload, limits)
389    }
390    #[cfg(all(feature = "grpc", feature = "streaming"))]
391    pub fn grpc_decoding_memory(&self, payload: &[u8]) -> Result<usize, crate::MessageError> {
392        (self.grpc_memory)(payload)
393    }
394
395    pub fn prepare_native(
396        &self, resources: ResourceSet, request: Box<dyn Any + Send>, endpoints: Box<dyn Any + Send>,
397    ) -> Result<Box<dyn NativeInstance>, ServingError> {
398        (self.native)(resources, request, endpoints)
399    }
400
401    pub fn contract(&self) -> Result<Contract, super::ContractError> {
402        self.contract_with_settings(&harmos_domain::config::SchemaSettings::default())
403    }
404
405    pub fn contract_with_settings(&self, settings: &harmos_domain::config::SchemaSettings) -> Result<Contract, super::ContractError> {
406        (self.contract)(settings)
407    }
408
409    pub fn install(self, table: &mut WireTable) -> Result<(), ServingError> {
410        self.validate()?;
411        self.verify_resources(table.owner_resources(), table.provided())?;
412        let contract = self.contract()?;
413        let resources = table.owner_resources().clone();
414        (self.install)(table, resources, contract)
415    }
416}
417
418impl<O: Carrier> NativeInstance for NativeOperation<O> {
419    fn classify(&self, error: &(dyn std::error::Error + Send + Sync + 'static)) -> crate::operations::FailureKind {
420        error.downcast_ref::<O::Error>().map_or(crate::operations::FailureKind::Permanent, O::classify)
421    }
422
423    fn initialize(&mut self, context: ExecutionContext) -> ServiceFuture<'_, ()> {
424        Box::pin(async move {
425            if self.operation.is_none() {
426                let context = context.with_resources(self.dependencies.clone());
427                self.operation = Some(O::bind(context).await.map_err(operation_error::<O>)?);
428            }
429            Ok(())
430        })
431    }
432
433    fn run(&mut self, context: ExecutionContext) -> ServiceFuture<'_, super::response::ErasedResponse> {
434        Box::pin(async move {
435            let operation = self.operation.as_mut().ok_or(ServingError::Uninitialized)?;
436            let context = context.with_resources(self.dependencies.clone());
437            #[cfg(feature = "streaming")]
438            let attempt = if let Some(streams) = &self.streams {
439                self.epoch = self.epoch.checked_add(1).ok_or(ServingError::StreamPhase)?;
440                Some(streams.start(self.epoch)?)
441            } else {
442                None
443            };
444            let response = operation.attempt(context, &mut self.request, &mut self.endpoints).await.map_err(operation_error::<O>)?;
445            #[cfg(feature = "streaming")]
446            if let Some(attempt) = attempt {
447                attempt.end();
448            }
449            #[cfg(all(feature = "grpc", feature = "streaming"))]
450            if self.protobuf {
451                let mut bytes = Vec::new();
452                crate::grpc::encode(&*response, &mut bytes).map_err(ServingError::Codec)?;
453                return Ok(super::Response::new(bytes).into_erased());
454            }
455            if self.encoded {
456                return Ok(super::Response::new(response.encode()).into_erased());
457            }
458            Ok(response.into_erased())
459        })
460    }
461
462    fn stop(&mut self) -> ServiceFuture<'_, ()> {
463        Box::pin(async move {
464            if self.stopped {
465                return Ok(());
466            }
467            if let Some(operation) = self.operation.as_mut() {
468                operation.stop().await.map_err(operation_error::<O>)?;
469            }
470            self.stopped = true;
471            Ok(())
472        })
473    }
474
475    fn cleanup(&mut self) -> ServiceFuture<'_, ()> {
476        Box::pin(async move {
477            if let Some(operation) = self.operation.as_mut() {
478                operation.cleanup().await.map_err(operation_error::<O>)?;
479            }
480            self.operation = None;
481            Ok(())
482        })
483    }
484
485    fn transferred(&self) -> bool {
486        use super::EndpointSet;
487        self.endpoints.transferred() || self.operation.as_ref().is_some_and(Carrier::transferred)
488    }
489}
490
491impl ServiceFactory {
492    pub fn initialize(&self, context: ExecutionContext) -> ServiceFuture<'static, Box<dyn ServiceInstance>> {
493        (self.initialize)(context)
494    }
495}
496
497impl<S: Service> ServiceInstance for InitializedService<S> {
498    fn run(&mut self, context: ExecutionContext) -> ServiceFuture<'_, ()> {
499        Box::pin(async move { self.service.run(context).await.map_err(service_error) })
500    }
501    fn stop(&mut self) -> ServiceFuture<'_, ()> {
502        Box::pin(async move { self.service.stop().await.map_err(service_error) })
503    }
504    fn cleanup(&mut self) -> ServiceFuture<'_, ()> {
505        Box::pin(async move { self.service.cleanup().await.map_err(service_error) })
506    }
507}
508
509/* === Public Functions === */
510
511/* === Private Functions === */
512
513fn service_error(error: impl std::error::Error + 'static) -> ServingError {
514    ServingError::Handler(crate::Failure::originate(crate::operations::FailureKind::Permanent, &error, crate::failure::Origin::Plugin))
515}
516
517fn operation_error<O: Carrier>(error: O::Error) -> ServingError {
518    ServingError::Native(std::sync::Arc::new(error))
519}
520
521/* === Tests === */
522
523#[cfg(test)]
524mod tests {
525    use super::*;
526    use std::sync::{
527        Arc,
528        atomic::{AtomicUsize, Ordering},
529    };
530
531    struct Worker(Arc<AtomicUsize>);
532
533    struct ReadCatalog;
534
535    impl OperationCatalog for ReadCatalog {
536        const CATALOG: Catalog = Catalog::of::<Self>("ReadCatalog", &[], &[]);
537        fn batch() -> RegistrationBatch {
538            RegistrationBatch::new().operation::<Read>()
539        }
540    }
541
542    #[crate::operation(harmos_path = crate, resources = [u64], id = "fixture.read", response = u64,
543        policy = crate::OperationPolicy {
544            timeout: std::time::Duration::from_secs(1),
545            attempt_timeout: std::time::Duration::from_secs(1),
546            retries: crate::RetryPolicy::Never,
547        },
548        key = |_: &Read| Some(super::super::CacheKey::new(vec![42]).unwrap()),
549    )]
550    struct Read;
551
552    impl crate::Operation for Read {
553        type Error = std::convert::Infallible;
554
555        async fn run(self, ctx: crate::Context<'_, Self>) -> Result<crate::Response<u64>, Self::Error> {
556            Ok(crate::Response::from_owner(ctx.resource::<u64>()))
557        }
558    }
559
560    fn ready<T>(mut future: ServiceFuture<'_, T>) -> T {
561        let mut task = std::task::Context::from_waker(std::task::Waker::noop());
562        match future.as_mut().poll(&mut task) {
563            std::task::Poll::Ready(Ok(value)) => value,
564            _ => panic!("synchronous fixture must complete successfully"),
565        }
566    }
567
568    #[test]
569    fn carrier_assignments_are_immutable() {
570        use crate::operations::{AssignedOperation, OperationAssignments};
571        let first = AssignedOperation {
572            operation: format!("fixture.{}", Read::ID),
573            fingerprint: Read::definition().fingerprint.0,
574            generation: 1,
575        };
576        let assignments = OperationAssignments::default();
577        let bytes = AssignedOperation::encode(std::slice::from_ref(&first)).unwrap();
578        assignments.install(&bytes).unwrap();
579        assignments.clone().install(&bytes).unwrap();
580        assert_eq!(assignments.resolve::<Read>().unwrap(), &first);
581        let mut replacement = first.clone();
582        replacement.generation = 2;
583        assert!(assignments.install(&AssignedOperation::encode(&[replacement]).unwrap()).is_err());
584        assert_eq!(assignments.resolve::<Read>().unwrap().generation, 1);
585        let mut wrong = first.clone();
586        wrong.fingerprint[0] ^= 1;
587        let incompatible = OperationAssignments::default();
588        incompatible.install(&AssignedOperation::encode(&[wrong]).unwrap()).unwrap();
589        assert!(incompatible.resolve::<Read>().is_err());
590        let duplicate = AssignedOperation {
591            operation: format!("other.{}", Read::ID),
592            ..first.clone()
593        };
594        assert!(OperationAssignments::default().install(&AssignedOperation::encode(&[first, duplicate]).unwrap()).is_err());
595    }
596
597    #[test]
598    fn native_response_retains_owner_across_retries_and_cleanup() {
599        let value = Arc::new(42u64);
600        let retained = Arc::downgrade(&value);
601        let mut dependencies = ResourceSet::new();
602        dependencies.insert(value).unwrap();
603        let batch = RegistrationBatch::new().operation::<Read>();
604        let factory = &batch.operations[0];
605        assert!(factory.prepare_native(dependencies.clone(), Box::new(7u64), Box::new(())).is_err());
606        let mut invocation = factory.prepare_native(dependencies, Box::new(Read), Box::new(())).unwrap();
607        let context = ExecutionContext::new(ResourceSet::new(), super::super::CallContext::default());
608
609        ready(invocation.initialize(context.clone()));
610        let first = ready(invocation.run(context.clone())).restore::<u64>().unwrap();
611        let second = ready(invocation.run(context)).restore::<u64>().unwrap();
612        assert!(first.ptr_eq(&second));
613        assert_eq!(*first, 42);
614        ready(invocation.stop());
615        ready(invocation.cleanup());
616
617        drop(invocation);
618        drop(batch);
619        assert!(retained.upgrade().is_some());
620        drop(first);
621        drop(second);
622        assert!(retained.upgrade().is_none());
623    }
624
625    #[test]
626    fn encoded_factory_preserves_request_cache_key_and_response() {
627        use crate::Message;
628
629        let mut dependencies = ResourceSet::new();
630        dependencies.insert(Arc::new(42u64)).unwrap();
631        let batch = RegistrationBatch::new().operation::<Read>();
632        let mut prepared = batch.operations[0].prepare_encoded(dependencies, &Read.encode()).unwrap();
633        assert_eq!(prepared.key, Read::key(&Read));
634        let context = ExecutionContext::new(ResourceSet::new(), super::super::CallContext::default());
635        ready(prepared.instance.initialize(context.clone()));
636        let response = ready(prepared.instance.run(context)).restore::<Vec<u8>>().unwrap();
637        assert_eq!(u64::decode(&response).unwrap(), 42);
638        ready(prepared.instance.stop());
639        ready(prepared.instance.cleanup());
640    }
641
642    impl Service for Worker {
643        type Error = crate::ResourceError;
644
645        async fn initialize(context: ExecutionContext) -> Result<Self, Self::Error> {
646            let counter = context.resource::<AtomicUsize>()?;
647            counter.fetch_add(1, Ordering::SeqCst);
648            Ok(Self(counter))
649        }
650
651        async fn run(&mut self, _: ExecutionContext) -> Result<(), Self::Error> {
652            self.0.fetch_add(1, Ordering::SeqCst);
653            Ok(())
654        }
655    }
656
657    #[test]
658    fn factory_rejects_a_substituted_declaration() {
659        let mut batch = RegistrationBatch::new().operation::<Read>();
660        batch.operations[0].definition.id = crate::Identifier::new("different").unwrap();
661        assert!(matches!(batch.operations[0].validate(), Err(ServingError::Catalog)));
662    }
663
664    #[test]
665    fn wire_identity_fences_replacement_before_decoding() {
666        use crate::operations::{ServingError, WireTable};
667        use crate::{Message, QualifiedId};
668        let batch = RegistrationBatch::new().operation::<Read>();
669        let fingerprint = crate::OperationDefinition::of::<Read>().unwrap().fingerprint.0;
670        let mut table = WireTable::new("fixture").unwrap();
671        table.resource(42u64);
672        table.capacity(std::num::NonZeroUsize::MIN);
673        table.register(batch.clone()).unwrap();
674        table.validate_binding().unwrap();
675        assert!(matches!(table.prepare(1, "spoof.fixture.read", fingerprint, 1, Read.encode()), Err(ServingError::Unknown { .. })));
676        assert!(matches!(table.prepare(1, "fixture.fixture.read", [0; 32], 1, vec![255]), Err(ServingError::Identity { .. })));
677        assert!(matches!(table.prepare(1, "fixture.fixture.read", fingerprint, 0, vec![255]), Err(ServingError::Identity { .. })));
678        table.prepare(1, "fixture.fixture.read", fingerprint, 7, Read.encode()).unwrap();
679        ready(Box::pin(table.unregister(&QualifiedId::qualify("fixture", "fixture.read").unwrap())));
680        table.register(batch).unwrap();
681        assert!(matches!(table.prepare(2, "fixture.fixture.read", fingerprint, 7, vec![255]), Err(ServingError::Identity { .. })));
682        table.prepare(2, "fixture.fixture.read", fingerprint, 8, Read.encode()).unwrap();
683        assert!(matches!(table.prepare(3, "fixture.fixture.read", fingerprint, 9, vec![255]), Err(ServingError::Identity { .. })));
684        ready(Box::pin(table.settle()));
685    }
686
687    #[test]
688    fn service_factory_reconstructs_from_its_snapshot() {
689        let counter = Arc::new(AtomicUsize::new(0));
690        let mut dependencies = ResourceSet::new();
691        dependencies.insert(counter.clone()).unwrap();
692        let policy = ServicePolicy {
693            stop_timeout: std::time::Duration::from_secs(1),
694            restart: crate::RestartPolicy::Never,
695        };
696        let batch = RegistrationBatch::new().service::<Worker>("worker", policy);
697        let factory = &batch.services[0];
698        let context = ExecutionContext::new(dependencies, super::super::CallContext::default());
699        let mut first = factory.initialize(context.clone());
700        let mut second = factory.initialize(context);
701        let mut task = std::task::Context::from_waker(std::task::Waker::noop());
702        assert!(matches!(first.as_mut().poll(&mut task), std::task::Poll::Ready(Ok(_))));
703        assert!(matches!(second.as_mut().poll(&mut task), std::task::Poll::Ready(Ok(_))));
704        assert_eq!(counter.load(Ordering::SeqCst), 2);
705    }
706
707    #[test]
708    fn binding_errors_aggregate() {
709        use crate::StartupError;
710        let mut table = WireTable::new("fixture").unwrap();
711        table.resource(7u32).resource(8u32);
712        table.catalog(&ReadCatalog::CATALOG).catalog(&ReadCatalog::CATALOG);
713        assert!(table.definitions().is_empty());
714
715        let Err(ServingError::Startup(error)) = table.validate_binding() else {
716            panic!("invalid binding must be refused");
717        };
718        let StartupError::Multiple { errors } = *error else {
719            panic!("independent binding errors must be aggregated");
720        };
721        assert_eq!(errors.len(), 3);
722        assert!(errors.iter().any(|error| matches!(error, StartupError::DuplicateResource { .. })));
723        assert!(errors.iter().any(|error| matches!(error, StartupError::DuplicateCatalog { .. })));
724        assert!(errors.iter().any(|error| matches!(error, StartupError::MissingResource { .. })));
725        assert!(table.definitions().is_empty());
726        assert_eq!(*table.context().resource::<u32>().unwrap(), 7);
727    }
728
729    #[test]
730    fn frozen_dynamic_resources() {
731        let mut table = WireTable::new("fixture").unwrap();
732        let owner = Arc::new(41u32);
733        table.resource::<u32>(owner.clone());
734        table.validate_binding().unwrap();
735        assert!(Arc::ptr_eq(&owner, &table.context().resource::<u32>().unwrap()));
736
737        let error = table.register(ReadCatalog::batch()).unwrap_err();
738        assert!(matches!(error, ServingError::Resource { .. }));
739        table.resource(42u64);
740        assert!(table.validate_binding().is_err());
741        assert!(table.context().resource::<u64>().is_err());
742        assert!(table.definitions().is_empty());
743    }
744
745    #[test]
746    fn resource_order_independence() {
747        let mut table = WireTable::new("fixture").unwrap();
748        table.catalog(&ReadCatalog::CATALOG).resource(42u64);
749        assert!(table.definitions().is_empty());
750        table.validate_binding().unwrap();
751        assert_eq!(table.definitions().len(), 1);
752        assert_eq!(*table.context().resource::<u64>().unwrap(), 42);
753    }
754}
harmos_sdk/operations/registration.rs

harmos_sdk/operations/
registration.rs
