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
17use std::sync::Arc;
18
19// Third Party Imports
20
21// Local Imports
22use super::{Answer, Finalization, Finished, Handler, Prepare, Prepared, ServingError, WireTable, guarded};
23use crate::Message;
24use crate::operations::{EndpointSet, Operation, OperationDeclaration, WireDeclaration};
25
26/* === Constants === */
27
28/* === Macros === */
29
30/* === Type Aliases === */
31
32/* === Structs === */
33
34struct State<T: Operation> {
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: Operation> {
52    resources: Arc<T::Resources>,
53    operation: Option<T>,
54    request: T::Declaration,
55    endpoints: T::Endpoints,
56    stopped: bool,
57}
58
59struct Endpoints<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<C: Send + Sync + 'static> WireTable<C> {
79    pub fn bind_operation<T>(&mut self, resources: Arc<T::Resources>) -> Result<(), ServingError>
80    where
81        T: for<'a> Operation<Context<'a> = &'a C, Endpoints = ()>,
82    {
83        if self.closed || self.ordinary_closed {
84            return Err(ServingError::Closing);
85        }
86        let contract = WireDeclaration::of::<T::Declaration>().contract()?;
87        if contract.input().is_some() || contract.output().is_some() {
88            return Err(ServingError::Streaming { operation: T::Declaration::ID.to_owned() });
89        }
90        #[cfg(feature = "grpc")]
91        let protobuf = T::Declaration::GRPC.is_some().then(super::protobuf::Codec::of::<T::Declaration>);
92        self.bind_owned_contract::<T>(
93            resources,
94            contract,
95            false,
96            false,
97            || {
98                Ok(Endpoints {
99                    endpoints: (),
100                    #[cfg(feature = "streaming")]
101                    streams: None,
102                })
103            },
104            #[cfg(feature = "grpc")]
105            protobuf,
106        )
107    }
108
109    // Bind a resident control: it runs until the host cancels it and is not
110    // charged against the ordinary invocation capacity.
111    pub fn bind_resident_operation<T>(&mut self, resources: Arc<T::Resources>) -> Result<(), ServingError>
112    where
113        T: for<'a> Operation<Context<'a> = &'a C, Endpoints = ()>,
114    {
115        let policy = T::Declaration::POLICY;
116        if policy.timeout != crate::operations::UNBOUNDED || policy.attempt_timeout != crate::operations::UNBOUNDED {
117            return Err(ServingError::Catalog);
118        }
119        self.bind_operation::<T>(resources)?;
120        if let Some(handler) = self.handlers.get_mut(T::Declaration::ID) {
121            handler.resident = true;
122        }
123        Ok(())
124    }
125
126    // Bind the built-in observe control a sidecar serves for its observers.
127    pub fn bind_observe_control<T>(&mut self, resources: Arc<T::Resources>) -> Result<(), ServingError>
128    where
129        T: for<'a> Operation<Context<'a> = &'a C, Endpoints = (), Declaration = crate::observers::Control>,
130    {
131        if self.closed || self.ordinary_closed {
132            return Err(ServingError::Closing);
133        }
134        self.bind_owned_contract::<T>(
135            resources,
136            crate::observers::contract()?,
137            false,
138            true,
139            || {
140                Ok(Endpoints {
141                    endpoints: (),
142                    #[cfg(feature = "streaming")]
143                    streams: None,
144                })
145            },
146            #[cfg(feature = "grpc")]
147            None,
148        )?;
149        if let Some(handler) = self.handlers.get_mut(crate::observers::Control::ID) {
150            handler.resident = true;
151        }
152        Ok(())
153    }
154
155    #[cfg(feature = "streaming")]
156    pub fn bind_streaming_operation<T>(&mut self, resources: Arc<T::Resources>, limits: super::WireStreamLimits) -> Result<(), ServingError>
157    where
158        T: for<'a> Operation<Context<'a> = &'a C>,
159        T::Endpoints: super::WireEndpoints,
160    {
161        use super::WireEndpoints;
162        if self.closed || self.ordinary_closed {
163            return Err(ServingError::Closing);
164        }
165        if T::Endpoints::item_types() == (None, None) {
166            return Err(ServingError::StreamUnsupported);
167        }
168        limits.validate()?;
169        let contract = WireDeclaration::of::<T::Declaration>().contract()?;
170        #[cfg(feature = "grpc")]
171        let protobuf = T::Declaration::GRPC.is_some().then(super::protobuf::Codec::of::<T::Declaration>);
172        self.bind_owned_contract::<T>(
173            resources,
174            contract,
175            false,
176            false,
177            move || {
178                let (endpoints, streams) = T::Endpoints::bind(limits)?;
179                Ok(Endpoints { endpoints, streams: Some(streams) })
180            },
181            #[cfg(feature = "grpc")]
182            protobuf,
183        )
184    }
185
186    #[cfg(feature = "streaming")]
187    pub(super) fn bind_catalog_operation<T>(&mut self, resources: Arc<T::Resources>) -> Result<(), ServingError>
188    where
189        T: for<'a> Operation<Context<'a> = &'a C>,
190        T::Endpoints: super::WireEndpoints,
191    {
192        use super::WireEndpoints;
193        let limits = self.stream_limits;
194        if T::Endpoints::item_types() != (None, None) && limits.is_none() {
195            return Err(ServingError::StreamLimits);
196        }
197        let contract = WireDeclaration::of::<T::Declaration>().contract()?;
198        #[cfg(feature = "grpc")]
199        let protobuf = T::Declaration::GRPC.is_some().then(super::protobuf::Codec::of::<T::Declaration>);
200        self.bind_owned_contract::<T>(
201            resources,
202            contract,
203            false,
204            false,
205            move || {
206                if let Some(endpoints) = T::Endpoints::empty() {
207                    return Ok(Endpoints { endpoints, streams: None });
208                }
209                let (endpoints, streams) = T::Endpoints::bind(limits.ok_or(ServingError::StreamLimits)?)?;
210                Ok(Endpoints { endpoints, streams: Some(streams) })
211            },
212            #[cfg(feature = "grpc")]
213            protobuf,
214        )
215    }
216
217    pub(crate) fn bind_owned_lifecycle<T>(
218        &mut self, resources: Arc<T::Resources>, descriptor: crate::operations::lifecycle::Descriptor,
219    ) -> Result<(), ServingError>
220    where
221        T: for<'a> Operation<Context<'a> = &'a C, Endpoints = ()>,
222        T::Declaration: crate::operations::lifecycle::SdkOperation,
223    {
224        use crate::operations::lifecycle::SdkOperation;
225        if self.closed || self.ordinary_closed {
226            return Err(ServingError::Closing);
227        }
228        if descriptor.action() != T::Declaration::ACTION {
229            return Err(ServingError::Catalog);
230        }
231        self.bind_owned_contract::<T>(
232            resources,
233            descriptor.contract().clone(),
234            true,
235            true,
236            || {
237                Ok(Endpoints {
238                    endpoints: (),
239                    #[cfg(feature = "streaming")]
240                    streams: None,
241                })
242            },
243            #[cfg(feature = "grpc")]
244            None,
245        )
246    }
247
248    fn bind_owned_contract<T>(
249        &mut self, resources: Arc<T::Resources>, contract: crate::operations::Contract, lifecycle: bool, builtin: bool,
250        endpoints: impl Fn() -> Result<Endpoints<T::Endpoints>, ServingError> + Send + Sync + 'static,
251        #[cfg(feature = "grpc")] protobuf: Option<super::protobuf::Codec>,
252    ) -> Result<(), ServingError>
253    where
254        T: for<'a> Operation<Context<'a> = &'a C>,
255    {
256        let id = T::Declaration::ID;
257        let (input, output) = T::Endpoints::item_types();
258        if (input, output) != crate::operations::item_types::<T::Declaration>()
259            || input.is_some() != contract.input().is_some()
260            || output.is_some() != contract.output().is_some()
261        {
262            return Err(ServingError::Catalog);
263        }
264        if self.handlers.contains_key(id) {
265            return Err(crate::Error::DuplicateOperation { id: id.to_owned() }.into());
266        }
267        let marked = contract.grpc().is_some();
268        #[cfg(not(feature = "grpc"))]
269        if marked {
270            return Err(ServingError::Codec(crate::MessageError::borrowed("gRPC operation requires provider protobuf support")));
271        }
272        #[cfg(feature = "grpc")]
273        if marked && protobuf.is_none() {
274            return Err(ServingError::Codec(crate::MessageError::borrowed("gRPC operation has no provider protobuf projection")));
275        }
276        let prepare: Prepare<C> = Box::new(move |payload, grpc| {
277            if grpc && !marked {
278                return Err(ServingError::Codec(crate::MessageError::borrowed("operation has no gRPC contract")));
279            }
280            #[cfg(not(feature = "grpc"))]
281            if grpc {
282                return Err(ServingError::Codec(crate::MessageError::borrowed("provider protobuf codec is disabled")));
283            }
284            #[cfg(feature = "grpc")]
285            let selected = if grpc {
286                Some(protobuf.ok_or_else(|| ServingError::Codec(crate::MessageError::borrowed("provider has no protobuf codec")))?)
287            } else {
288                None
289            };
290            #[cfg(feature = "grpc")]
291            let request = if let Some(codec) = selected {
292                let mut request = None::<T::Declaration>;
293                (codec.decode)(&payload, &mut request).map_err(ServingError::Codec)?;
294                request.ok_or(ServingError::Catalog)?
295            } else {
296                T::Declaration::decode(&payload).map_err(ServingError::Codec)?
297            };
298            #[cfg(not(feature = "grpc"))]
299            let request = T::Declaration::decode(&payload).map_err(ServingError::Codec)?;
300            let grant = <T::Declaration as OperationDeclaration>::grant(&request);
301            let endpoints = endpoints()?;
302            #[cfg(all(feature = "grpc", feature = "streaming"))]
303            let endpoints = Endpoints {
304                endpoints: endpoints.endpoints,
305                streams: match (endpoints.streams, selected) {
306                    (Some(streams), Some(codec)) => Some(streams.protobuf(codec)?),
307                    (streams, _) => streams,
308                },
309            };
310            let state = State::<T> {
311                #[cfg(feature = "streaming")]
312                streams: endpoints.streams,
313                physical: Some(Physical {
314                    resources: resources.clone(),
315                    operation: None,
316                    request,
317                    endpoints: endpoints.endpoints,
318                    stopped: false,
319                }),
320                initialization: Initialization::Pending,
321                closing: false,
322                finalization: None,
323                #[cfg(feature = "grpc")]
324                protobuf: selected,
325                grant,
326            };
327            Ok(Box::new(state))
328        });
329        let handler =
330            Handler { lifecycle, builtin, resident: false, definition: contract.operation().clone(), contract: Some(contract), prepare };
331        self.handlers.insert(id.to_owned(), handler);
332        Ok(())
333    }
334}
335
336impl<T, C> Prepared<C> for State<T>
337where
338    C: Send + Sync + 'static,
339    T: for<'a> Operation<Context<'a> = &'a C>,
340{
341    #[cfg(feature = "streaming")]
342    fn streams(&self) -> Option<super::WireStreams> {
343        self.streams.clone()
344    }
345
346    fn grant(&self) -> Option<crate::plugin::Grant> {
347        self.grant.clone()
348    }
349
350    fn initialize<'a>(&'a mut self, context: &'a C) -> Finished<'a> {
351        Box::pin(async move {
352            if self.closing {
353                return Err(ServingError::Closing);
354            }
355            match self.initialization {
356                Initialization::Ready => return Ok(()),
357                Initialization::Started => return Err(ServingError::InitializationStarted),
358                Initialization::Pending => self.initialization = Initialization::Started,
359            }
360            let physical = self.physical.as_mut().expect("original operation before cleanup");
361            let resources = Arc::clone(&physical.resources);
362            let operation = guarded(async { T::initialize(resources, context).await.map_err(operation_error::<T>) }).await?;
363            physical.operation = Some(operation);
364            self.initialization = Initialization::Ready;
365            Ok(())
366        })
367    }
368
369    fn attempt<'a>(&'a mut self, context: &'a C) -> Answer<'a> {
370        Box::pin(async move {
371            if self.closing {
372                return Err(ServingError::Closing);
373            }
374            if !matches!(self.initialization, Initialization::Ready) {
375                return Err(ServingError::Uninitialized);
376            }
377            let physical = self.physical.as_mut().expect("original initialized operation");
378            guarded(async {
379                let operation = physical.operation.as_mut().expect("initialized operation");
380                let response =
381                    operation.run(context, &mut physical.request, &mut physical.endpoints).await.map_err(operation_error::<T>)?;
382                #[cfg(feature = "grpc")]
383                if let Some(codec) = self.protobuf {
384                    return (codec.encode)(&*response).map_err(ServingError::Codec);
385                }
386                Ok(response.encode_to_vec())
387            })
388            .await
389        })
390    }
391
392    fn finalize(&mut self) -> Finished<'_> {
393        self.closing = true;
394        if self.finalization.is_none() {
395            let mut physical = self.physical.take().expect("original operation for cleanup");
396            let future = async move {
397                let result = guarded(async {
398                    // An Operation That Never Initialized Owns Nothing To Stop
399                    let Some(operation) = physical.operation.as_mut() else {
400                        return Ok(());
401                    };
402                    if !physical.stopped {
403                        operation.stop().await.map_err(operation_error::<T>)?;
404                        physical.stopped = true;
405                    }
406                    operation.cleanup().await.map_err(operation_error::<T>)
407                })
408                .await;
409                (physical, result)
410            };
411            self.finalization = Some(Box::pin(future));
412        }
413        Box::pin(async move {
414            let (physical, result) = self.finalization.as_mut().expect("original cleanup future").await;
415            self.physical = Some(physical);
416            self.finalization = None;
417            result
418        })
419    }
420}
421
422/* === Public Functions === */
423
424/* === Private Functions === */
425
426// The handler's error becomes the failure's cause here, in the plugin process
427// where it originated; every later carrier forwards the same value.
428fn operation_error<T: Operation>(error: T::Error) -> ServingError {
429    let failure = crate::Failure::originate(T::classify(&error), &error, crate::failure::Origin::Plugin);
430    ServingError::Handler(failure)
431}
432
433/* === Tests === */
harmos_sdk/operations/serving/owned.rs

harmos_sdk/operations/serving/
owned.rs
