harmos_sdk/operations/serving/
owned.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
17
18// Third Party Imports
19
20// Local Imports
21use super::{Answer, Finalization, Finished, Prepare, Prepared, Route, ServingError, WireTable, guarded};
22use crate::Message;
23use crate::QualifiedId;
24use crate::operations::{Carrier, Dependencies, EndpointSet, ExecutionContext, OperationDeclaration};
25
26/* === Constants === */
27
28/* === Macros === */
29
30/* === Type Aliases === */
31
32/* === Structs === */
33
34struct State<T: Carrier> {
35    physical: Option<Physical<T>>,
36
37    #[cfg(feature = "streaming")]
38    streams: Option<super::WireStreams>,
39
40    initialization: Initialization,
41    closing: bool,
42    finalization: Option<Finalization<Physical<T>>>,
43
44    #[cfg(feature = "grpc")]
45    protobuf: Option<super::protobuf::Codec>,
46
47    // The grant the decoded request needs; the host checks it before any attempt.
48    grant: Option<crate::plugin::Grant>,
49}
50
51struct Physical<T: Carrier> {
52    dependencies: Dependencies,
53    operation: Option<T>,
54    request: <T::Declaration as OperationDeclaration>::Request,
55    endpoints: T::StreamEndpoints,
56    stopped: bool,
57}
58
59struct StreamEndpoints<E> {
60    endpoints: E,
61
62    #[cfg(feature = "streaming")]
63    streams: Option<super::WireStreams>,
64}
65
66/* === Enums === */
67
68enum Initialization {
69    Pending,
70    Started,
71    Ready,
72}
73
74/* === Traits === */
75
76/* === Implementations === */
77
78impl WireTable {
79    pub(crate) fn install<D, T>(&mut self, dependencies: Dependencies, contract: crate::operations::Contract) -> Result<(), ServingError>
80    where
81        D: OperationDeclaration,
82        T: Carrier<Declaration = D>,
83    {
84        #[cfg(feature = "grpc")]
85        let protobuf = D::GRPC.is_some().then(super::protobuf::Codec::of::<D>);
86        #[cfg(feature = "streaming")]
87        let limits = self.stream_limits;
88        #[cfg(feature = "streaming")]
89        if T::StreamEndpoints::item_types() != (None, None) && limits.is_none() {
90            return Err(ServingError::StreamLimits);
91        }
92        #[cfg(not(feature = "streaming"))]
93        if T::StreamEndpoints::item_types() != (None, None) {
94            return Err(ServingError::StreamUnsupported);
95        }
96        let observe = std::any::TypeId::of::<D>() == std::any::TypeId::of::<crate::observers::Control>();
97        self.bind_owned_contract::<D, T>(
98            dependencies,
99            contract,
100            false,
101            observe,
102            move || {
103                if let Some(endpoints) = T::StreamEndpoints::empty() {
104                    return Ok(StreamEndpoints {
105                        endpoints,
106                        #[cfg(feature = "streaming")]
107                        streams: None,
108                    });
109                }
110                #[cfg(feature = "streaming")]
111                {
112                    let (endpoints, streams) = T::StreamEndpoints::bind_wire(limits.ok_or(ServingError::StreamLimits)?)?;
113                    Ok(StreamEndpoints { endpoints, streams: Some(streams) })
114                }
115                #[cfg(not(feature = "streaming"))]
116                Err(ServingError::StreamUnsupported)
117            },
118            #[cfg(feature = "grpc")]
119            protobuf,
120        )?;
121        if observe {
122            let id = QualifiedId::qualify(&self.owner, D::ID).map_err(|_| ServingError::Catalog)?;
123            self.handlers.get_mut(id.as_str()).expect("installed observe control").resident = true;
124        }
125        Ok(())
126    }
127
128    pub fn install_lifecycle<D, T>(
129        &mut self, dependencies: Dependencies, descriptor: harmos_domain::model::operations::lifecycle::Descriptor,
130    ) -> Result<(), ServingError>
131    where
132        D: crate::OperationDeclaration,
133        T: Carrier<Declaration = D, StreamEndpoints = ()>,
134    {
135        if self.closed || self.ordinary_closed {
136            return Err(ServingError::Closing);
137        }
138        if descriptor.action().id() != D::ID
139            || descriptor.contract().fingerprint()
140                != (harmos_domain::operations::Entry::of::<D>().fingerprint)().map_err(|_| ServingError::Catalog)?
141        {
142            return Err(ServingError::Catalog);
143        }
144        self.bind_owned_contract::<D, T>(
145            dependencies,
146            descriptor.into(),
147            true,
148            true,
149            || {
150                Ok(StreamEndpoints {
151                    endpoints: (),
152                    #[cfg(feature = "streaming")]
153                    streams: None,
154                })
155            },
156            #[cfg(feature = "grpc")]
157            None,
158        )
159    }
160
161    fn bind_owned_contract<D, T>(
162        &mut self, dependencies: Dependencies, contract: crate::operations::Contract, lifecycle: bool, builtin: bool,
163        endpoints: impl Fn() -> Result<StreamEndpoints<T::StreamEndpoints>, ServingError> + Send + Sync + 'static,
164        #[cfg(feature = "grpc")] protobuf: Option<super::protobuf::Codec>,
165    ) -> Result<(), ServingError>
166    where
167        D: OperationDeclaration,
168        T: Carrier<Declaration = D>,
169    {
170        let qualified = QualifiedId::qualify(&self.owner, D::ID).map_err(|_| ServingError::Catalog)?;
171        let id = qualified.as_str();
172        let (input, output) = T::StreamEndpoints::item_types();
173        if (input, output) != crate::operations::item_types::<D>()
174            || input.is_some() != contract.input().is_some()
175            || output.is_some() != contract.output().is_some()
176        {
177            return Err(ServingError::Catalog);
178        }
179        if self.handlers.contains_key(id) {
180            return Err(crate::Error::DuplicateOperation { id: id.to_owned() }.into());
181        }
182        let marked = contract.grpc().is_some();
183        #[cfg(not(feature = "grpc"))]
184        if marked {
185            return Err(ServingError::Codec(crate::MessageError::borrowed("gRPC operation requires provider protobuf support")));
186        }
187        #[cfg(feature = "grpc")]
188        if marked && protobuf.is_none() {
189            return Err(ServingError::Codec(crate::MessageError::borrowed("gRPC operation has no provider protobuf projection")));
190        }
191        let prepare: Prepare = Box::new(move |payload, grpc| {
192            if grpc && !marked {
193                return Err(ServingError::Codec(crate::MessageError::borrowed("operation has no gRPC contract")));
194            }
195            #[cfg(not(feature = "grpc"))]
196            if grpc {
197                return Err(ServingError::Codec(crate::MessageError::borrowed("provider protobuf codec is disabled")));
198            }
199            #[cfg(feature = "grpc")]
200            let selected = if grpc {
201                Some(protobuf.ok_or_else(|| ServingError::Codec(crate::MessageError::borrowed("provider has no protobuf codec")))?)
202            } else {
203                None
204            };
205            #[cfg(feature = "grpc")]
206            let request = if let Some(codec) = selected {
207                let mut request = None::<<T::Declaration as OperationDeclaration>::Request>;
208                (codec.decode)(&payload, &mut request).map_err(ServingError::Codec)?;
209                request.ok_or(ServingError::Catalog)?
210            } else {
211                <<T::Declaration as OperationDeclaration>::Request as crate::Message>::decode(&payload).map_err(ServingError::Codec)?
212            };
213            #[cfg(not(feature = "grpc"))]
214            let request =
215                <<T::Declaration as OperationDeclaration>::Request as crate::Message>::decode(&payload).map_err(ServingError::Codec)?;
216            let grant = <D as OperationDeclaration>::grant(&request);
217            let endpoints = endpoints()?;
218            #[cfg(all(feature = "grpc", feature = "streaming"))]
219            let endpoints = StreamEndpoints {
220                endpoints: endpoints.endpoints,
221                streams: match (endpoints.streams, selected) {
222                    (Some(streams), Some(codec)) => Some(streams.protobuf(codec)?),
223                    (streams, _) => streams,
224                },
225            };
226            let state = State::<T> {
227                #[cfg(feature = "streaming")]
228                streams: endpoints.streams,
229                physical: Some(Physical {
230                    dependencies: dependencies.clone(),
231                    operation: None,
232                    request,
233                    endpoints: endpoints.endpoints,
234                    stopped: false,
235                }),
236                initialization: Initialization::Pending,
237                closing: false,
238                finalization: None,
239                #[cfg(feature = "grpc")]
240                protobuf: selected,
241                grant,
242            };
243            Ok(Box::new(state))
244        });
245        let handler = Route {
246            callees: D::OPERATIONS,
247            generation: None,
248            lifecycle,
249            builtin,
250            resident: false,
251            closing: false,
252            definition: contract.operation().clone(),
253            contract: Some(contract),
254            prepare,
255        };
256        self.handlers.insert(id.to_owned(), handler);
257        Ok(())
258    }
259}
260
261impl<T: Carrier> Prepared for State<T> {
262    #[cfg(feature = "streaming")]
263    fn streams(&self) -> Option<super::WireStreams> {
264        self.streams.clone()
265    }
266
267    fn grant(&self) -> Option<crate::plugin::Grant> {
268        self.grant.clone()
269    }
270
271    fn bind<'a>(&'a mut self, context: &'a ExecutionContext) -> Finished<'a> {
272        Box::pin(async move {
273            if self.closing {
274                return Err(ServingError::Closing);
275            }
276            match self.initialization {
277                Initialization::Ready => return Ok(()),
278                Initialization::Started => return Err(ServingError::InitializationStarted),
279                Initialization::Pending => self.initialization = Initialization::Started,
280            }
281            let physical = self.physical.as_mut().expect("original operation before cleanup");
282            let context = context.for_registration(physical.dependencies.clone());
283            let operation = guarded(async { T::bind(context).await.map_err(operation_error::<T>) }).await?;
284            physical.operation = Some(operation);
285            self.initialization = Initialization::Ready;
286            Ok(())
287        })
288    }
289
290    fn attempt<'a>(&'a mut self, context: &'a ExecutionContext) -> Answer<'a> {
291        Box::pin(async move {
292            if self.closing {
293                return Err(ServingError::Closing);
294            }
295            if !matches!(self.initialization, Initialization::Ready) {
296                return Err(ServingError::Uninitialized);
297            }
298            let physical = self.physical.as_mut().expect("original initialized operation");
299            guarded(async {
300                let operation = physical.operation.as_mut().expect("initialized operation");
301                let response = operation
302                    .attempt(context.for_registration(physical.dependencies.clone()), &mut physical.request, &mut physical.endpoints)
303                    .await
304                    .map_err(operation_error::<T>)?;
305                #[cfg(feature = "grpc")]
306                if let Some(codec) = self.protobuf {
307                    return (codec.encode)(&*response).map_err(ServingError::Codec);
308                }
309                Ok(response.encode())
310            })
311            .await
312        })
313    }
314
315    fn finalize(&mut self) -> Finished<'_> {
316        self.closing = true;
317        if self.finalization.is_none() {
318            let mut physical = self.physical.take().expect("original operation for cleanup");
319            let future = async move {
320                let result = guarded(async {
321                    // An Operation That Never Initialized Owns Nothing To Stop
322                    let Some(operation) = physical.operation.as_mut() else {
323                        return Ok(());
324                    };
325                    if !physical.stopped {
326                        operation.stop().await.map_err(operation_error::<T>)?;
327                        physical.stopped = true;
328                    }
329                    operation.cleanup().await.map_err(operation_error::<T>)
330                })
331                .await;
332                (physical, result)
333            };
334            self.finalization = Some(Box::pin(future));
335        }
336        Box::pin(async move {
337            let (physical, result) = self.finalization.as_mut().expect("original cleanup future").await;
338            self.physical = Some(physical);
339            self.finalization = None;
340            result
341        })
342    }
343}
344
345/* === Public Functions === */
346
347/* === Private Functions === */
348
349// The handler's error becomes the failure's cause here, in the plugin process
350// where it originated; every later carrier forwards the same value.
351fn operation_error<T: Carrier>(error: T::Error) -> ServingError {
352    let failure = crate::Failure::originate(T::classify(&error), &error, crate::failure::Origin::Plugin);
353    ServingError::Handler(failure)
354}
355
356/* === Tests === */
harmos_sdk/operations/serving/owned.rs

harmos_sdk/operations/serving/
owned.rs
