restate_sdk/ingress/
mod.rs

1//! # Ingress client
2//!
3//! Ingress clients invoke Restate handlers from outside a Restate service. The impl-block service,
4//! virtual-object, and workflow macros generate a typed `<Type>IngressClient` for their handlers.
5//! The ingress client requires Restate 1.7 or newer.
6//!
7//! ## Using reqwest
8//!
9//! Enable the `reqwest-client` feature to use [`ReqwestClient`]:
10//!
11//! ```rust,no_run
12//! # use restate_sdk::prelude::*;
13//! # struct Greeter;
14//! # #[restate_sdk::service]
15//! # impl Greeter {
16//! #     #[handler]
17//! #     async fn greet(&self, _ctx: Context<'_>, name: String) -> HandlerResult<String> {
18//! #         Ok(format!("Hello, {name}!"))
19//! #     }
20//! # }
21//! # #[cfg(feature = "reqwest-client")]
22//! # async fn call_greeter() -> Result<(), Box<dyn std::error::Error>> {
23//! use restate_sdk::ingress::ReqwestClient;
24//!
25//! let client = ReqwestClient::connect("http://localhost:8080".parse()?)?;
26//! let greeter = GreeterIngressClient::from_client(client);
27//! let greeting = greeter
28//!     .greet("Ada".to_owned())
29//!     .call()
30//!     .await?
31//!     .into_body()?;
32//!
33//! println!("{greeting}");
34//! # Ok(())
35//! # }
36//! ```
37//!
38//! ## Using another HTTP client
39//!
40//! Implement [`RequestExecutor`] for your HTTP client and pass it to [`Client::new`].
41
42use std::error::Error;
43use std::future::Future;
44use std::marker::PhantomData;
45use std::sync::Arc;
46use std::time::Duration;
47
48pub use crate::context::RequestTarget;
49use crate::serde::{Deserialize, PayloadMetadata, Serialize};
50use bytes::Bytes;
51use http::header::{AUTHORIZATION, CONTENT_TYPE, InvalidHeaderValue};
52use http::uri::PathAndQuery;
53use http::{
54    HeaderMap, HeaderName, HeaderValue, Method, Request as HttpRequest, Response as HttpResponse,
55    Response, StatusCode, Uri,
56};
57use percent_encoding::{AsciiSet, NON_ALPHANUMERIC, utf8_percent_encode};
58
59const IDEMPOTENCY_KEY: HeaderName = HeaderName::from_static("idempotency-key");
60const X_RESTATE_ID: HeaderName = HeaderName::from_static("x-restate-id");
61const X_RESTATE_LIMIT_KEY: HeaderName = HeaderName::from_static("x-restate-limit-key");
62const OUTPUT_NOT_READY: u16 = 470;
63
64const PATH_SEGMENT_ENCODE_SET: &AsciiSet = &NON_ALPHANUMERIC
65    .remove(b'-')
66    .remove(b'.')
67    .remove(b'_')
68    .remove(b'~');
69
70type BoxError = Box<dyn Error + Send + Sync + 'static>;
71
72/// Executes one fully buffered HTTP request/response exchange.
73///
74/// Implementations must return the complete response body. Streaming and transport-native body
75/// types intentionally do not cross this boundary.
76pub trait RequestExecutor: Send + Sync + 'static {
77    type Error: Error + Send + Sync + 'static;
78
79    fn execute(
80        &self,
81        request: HttpRequest<Bytes>,
82    ) -> impl Future<Output = Result<HttpResponse<Bytes>, Self::Error>> + Send;
83}
84
85/// Error returned when constructing an ingress client.
86#[derive(Debug, thiserror::Error)]
87pub enum ClientBuildError {
88    #[error("base URI must include a scheme and authority")]
89    RelativeBaseUri,
90    #[error("base URI must not include a query")]
91    BaseUriHasQuery,
92    #[error("Restate auth token is not valid as an HTTP header value")]
93    InvalidAuthToken {
94        #[source]
95        source: InvalidHeaderValue,
96    },
97}
98
99/// Error produced while constructing, executing, or decoding an ingress request.
100#[derive(Debug, thiserror::Error)]
101pub enum ClientError {
102    #[error("ingress transport failed: {source}")]
103    Transport {
104        #[source]
105        source: BoxError,
106    },
107    #[error("failed to construct ingress request: {source}")]
108    Request {
109        #[source]
110        source: BoxError,
111    },
112    #[error("failed to serialize ingress request: {source}")]
113    Serialization {
114        #[source]
115        source: BoxError,
116    },
117    #[error("invalid ingress request: {message}")]
118    InvalidRequest { message: String },
119    #[error("malformed ingress response: {message}")]
120    Protocol {
121        message: String,
122        response: Box<HttpResponse<Bytes>>,
123    },
124    #[error("ingress returned HTTP status {status}")]
125    Status {
126        status: StatusCode,
127        response: Box<HttpResponse<Bytes>>,
128    },
129    #[error("failed to decode ingress response payload: {source}")]
130    PayloadDecode {
131        #[source]
132        source: BoxError,
133        response: Box<HttpResponse<Bytes>>,
134    },
135}
136
137impl ClientError {
138    /// Returns the original buffered HTTP response when the failure happened after transport.
139    pub fn response(&self) -> Option<&HttpResponse<Bytes>> {
140        match self {
141            Self::Protocol { response, .. }
142            | Self::Status { response, .. }
143            | Self::PayloadDecode { response, .. } => Some(response),
144            Self::Transport { .. }
145            | Self::Request { .. }
146            | Self::Serialization { .. }
147            | Self::InvalidRequest { .. } => None,
148        }
149    }
150
151    /// Consumes the error and returns its original buffered HTTP response, if any.
152    pub fn into_response(self) -> Option<HttpResponse<Bytes>> {
153        match self {
154            Self::Protocol { response, .. }
155            | Self::Status { response, .. }
156            | Self::PayloadDecode { response, .. } => Some(*response),
157            Self::Transport { .. }
158            | Self::Request { .. }
159            | Self::Serialization { .. }
160            | Self::InvalidRequest { .. } => None,
161        }
162    }
163}
164
165struct ClientState<E> {
166    scheme: http::uri::Scheme,
167    authority: http::uri::Authority,
168    base_path: String,
169    executor: E,
170}
171
172/// A transport-neutral Restate ingress client.
173pub struct Client<E> {
174    state: Arc<ClientState<E>>,
175    default_headers: Arc<HeaderMap>,
176}
177
178impl<E> Clone for Client<E> {
179    fn clone(&self) -> Self {
180        Self {
181            state: Arc::clone(&self.state),
182            default_headers: Arc::clone(&self.default_headers),
183        }
184    }
185}
186
187impl<E: RequestExecutor> Client<E> {
188    /// Creates a client with no default headers.
189    pub fn new(base_uri: Uri, executor: E) -> Result<Self, ClientBuildError> {
190        Self::builder(base_uri, executor).build()
191    }
192
193    /// Creates a client builder.
194    pub fn builder(base_uri: Uri, executor: E) -> ClientBuilder<E> {
195        ClientBuilder {
196            base_uri,
197            executor,
198            default_headers: HeaderMap::new(),
199        }
200    }
201
202    /// Adds a raw Restate Cloud authentication token to every request.
203    ///
204    /// This method prepends `Bearer ` and installs the resulting `Authorization` header as a
205    /// default. An `Authorization` header set on an individual request overrides it.
206    ///
207    /// # Errors
208    ///
209    /// Returns [`ClientBuildError::InvalidAuthToken`] when the token cannot be represented in an
210    /// HTTP header.
211    pub fn with_restate_auth_token(
212        mut self,
213        token: impl AsRef<str>,
214    ) -> Result<Self, ClientBuildError> {
215        let mut value = HeaderValue::from_str(&format!("Bearer {}", token.as_ref()))
216            .map_err(|source| ClientBuildError::InvalidAuthToken { source })?;
217        value.set_sensitive(true);
218        Arc::make_mut(&mut self.default_headers).insert(AUTHORIZATION, value);
219        Ok(self)
220    }
221
222    /// Creates a typed request for a service handler.
223    pub fn request<Req, Res>(&self, target: RequestTarget, input: Req) -> Request<E, Req, Res> {
224        Request {
225            client: self.clone(),
226            target,
227            input,
228            headers: HeaderMap::new(),
229            idempotency_key: None,
230            scope: None,
231            limit_key: None,
232            response: PhantomData,
233        }
234    }
235
236    /// Creates a typed handle for a persisted or externally obtained invocation ID.
237    pub fn invocation_handle<Res>(
238        &self,
239        invocation_id: impl Into<InvocationId>,
240    ) -> InvocationHandle<E, Res> {
241        InvocationHandle {
242            client: self.clone(),
243            invocation_id: Arc::new(invocation_id.into()),
244            response: PhantomData,
245        }
246    }
247
248    /// Looks up the invocation handle for a workflow by name and key.
249    ///
250    /// `scope` optionally restricts the lookup to a Restate Cloud scope; pass `None` outside Cloud.
251    pub async fn lookup_workflow<Res>(
252        &self,
253        workflow_name: impl Into<String>,
254        workflow_key: impl Into<String>,
255        scope: Option<String>,
256    ) -> Result<InvocationHandle<E, Res>, ClientError> {
257        let body = serde_json::to_vec(&LookupRequest::Workflow {
258            workflow_name: workflow_name.into(),
259            workflow_key: workflow_key.into(),
260            scope,
261        })
262        .map_err(|source| ClientError::Serialization {
263            source: Box::new(source),
264        })?;
265
266        let response = self
267            .post_json("/restate/lookup".to_owned(), body.into())
268            .await?;
269        if response.status() != StatusCode::OK {
270            return Err(status_error(response));
271        }
272
273        let lookup: LookupResponse = match serde_json::from_slice(response.body()) {
274            Ok(lookup) => lookup,
275            Err(source) => {
276                return Err(ClientError::PayloadDecode {
277                    source: Box::new(source),
278                    response: Box::new(response),
279                });
280            }
281        };
282
283        Ok(self.invocation_handle(lookup.invocation_id))
284    }
285
286    fn uri(&self, path_and_query: String) -> Result<Uri, ClientError> {
287        let path_and_query = PathAndQuery::try_from(path_and_query).map_err(request_error)?;
288        Uri::builder()
289            .scheme(self.state.scheme.clone())
290            .authority(self.state.authority.clone())
291            .path_and_query(path_and_query)
292            .build()
293            .map_err(request_error)
294    }
295
296    fn path(&self, suffix: &str) -> String {
297        format!("{}{suffix}", self.state.base_path)
298    }
299
300    async fn execute(
301        &self,
302        request: HttpRequest<Bytes>,
303    ) -> Result<HttpResponse<Bytes>, ClientError> {
304        self.state
305            .executor
306            .execute(request)
307            .await
308            .map_err(|source| ClientError::Transport {
309                source: Box::new(source),
310            })
311    }
312
313    async fn get(&self, path: String) -> Result<HttpResponse<Bytes>, ClientError> {
314        let mut request = HttpRequest::builder()
315            .method(Method::GET)
316            .uri(self.uri(self.path(&path))?)
317            .body(Bytes::new())
318            .map_err(request_error)?;
319        *request.headers_mut() = self.default_headers.as_ref().clone();
320        self.execute(request).await
321    }
322
323    async fn post_json(
324        &self,
325        path: String,
326        body: Bytes,
327    ) -> Result<HttpResponse<Bytes>, ClientError> {
328        let mut request = HttpRequest::builder()
329            .method(Method::POST)
330            .uri(self.uri(self.path(&path))?)
331            .body(body)
332            .map_err(request_error)?;
333        let headers = request.headers_mut();
334        *headers = self.default_headers.as_ref().clone();
335        headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
336        self.execute(request).await
337    }
338}
339
340/// Builder for [`Client`].
341pub struct ClientBuilder<E> {
342    base_uri: Uri,
343    executor: E,
344    default_headers: HeaderMap,
345}
346
347impl<E: RequestExecutor> ClientBuilder<E> {
348    /// Adds or replaces a header included in every request.
349    pub fn default_header(mut self, name: HeaderName, value: HeaderValue) -> Self {
350        self.default_headers.insert(name, value);
351        self
352    }
353
354    /// Validates the base URI and builds the client.
355    pub fn build(self) -> Result<Client<E>, ClientBuildError> {
356        let Some(scheme) = self.base_uri.scheme().cloned() else {
357            return Err(ClientBuildError::RelativeBaseUri);
358        };
359        let Some(authority) = self.base_uri.authority().cloned() else {
360            return Err(ClientBuildError::RelativeBaseUri);
361        };
362        if self.base_uri.query().is_some() {
363            return Err(ClientBuildError::BaseUriHasQuery);
364        }
365
366        let path = self.base_uri.path();
367        let base_path = if path == "/" {
368            String::new()
369        } else {
370            path.trim_end_matches('/').to_owned()
371        };
372
373        Ok(Client {
374            state: Arc::new(ClientState {
375                scheme,
376                authority,
377                base_path,
378                executor: self.executor,
379            }),
380            default_headers: Arc::new(self.default_headers),
381        })
382    }
383}
384
385fn encoded_target_path(target: &RequestTarget) -> String {
386    match target {
387        RequestTarget::Service { name, handler } => {
388            format!("{}/{}", encode_segment(name), encode_segment(handler))
389        }
390        RequestTarget::Object { name, key, handler }
391        | RequestTarget::Workflow { name, key, handler } => format!(
392            "{}/{}/{}",
393            encode_segment(name),
394            encode_segment(key),
395            encode_segment(handler)
396        ),
397    }
398}
399
400/// An owned typed ingress request builder.
401pub struct Request<E, Req, Res = ()> {
402    client: Client<E>,
403    target: RequestTarget,
404    input: Req,
405    headers: HeaderMap,
406    idempotency_key: Option<String>,
407    scope: Option<String>,
408    limit_key: Option<String>,
409    response: PhantomData<fn() -> Res>,
410}
411
412impl<E, Req, Res> Request<E, Req, Res> {
413    /// Adds or replaces a request header.
414    pub fn header(mut self, name: HeaderName, value: HeaderValue) -> Self {
415        self.headers.insert(name, value);
416        self
417    }
418
419    /// Overrides the outgoing content type.
420    pub fn content_type(mut self, value: HeaderValue) -> Self {
421        self.headers.insert(CONTENT_TYPE, value);
422        self
423    }
424
425    pub fn idempotency_key(mut self, key: impl Into<String>) -> Self {
426        self.idempotency_key = Some(key.into());
427        self
428    }
429
430    pub fn scope(mut self, scope: impl Into<String>) -> Self {
431        self.scope = Some(scope.into());
432        self
433    }
434
435    pub fn limit_key(mut self, key: impl Into<String>) -> Self {
436        self.limit_key = Some(key.into());
437        self
438    }
439
440    async fn call_with_metadata(
441        self,
442        metadata: Option<RequestMetadata>,
443    ) -> Result<CallResponse<E, Res>, ClientError>
444    where
445        E: RequestExecutor,
446        Req: Serialize,
447    {
448        let (client, request) = self.into_http_request(InvokeType::Call, None, metadata)?;
449        let response = client.execute(request).await?;
450        let invocation_id = match invocation_id_header(&response) {
451            Ok(invocation_id) => invocation_id,
452            Err(message) => {
453                return Err(ClientError::Protocol {
454                    message,
455                    response: Box::new(response),
456                });
457            }
458        };
459        let handle = client.invocation_handle(invocation_id);
460        Ok(CallResponse { response, handle })
461    }
462
463    async fn send_with_delay_and_metadata(
464        self,
465        delay: Option<Duration>,
466        metadata: Option<RequestMetadata>,
467    ) -> Result<SendResponse<E, Res>, ClientError>
468    where
469        E: RequestExecutor,
470        Req: Serialize,
471    {
472        let (client, request) = self.into_http_request(InvokeType::Send, delay, metadata)?;
473        let response = client.execute(request).await?;
474        if response.status() != StatusCode::ACCEPTED {
475            return Err(status_error(response));
476        }
477
478        let invocation_id = match invocation_id_header(&response) {
479            Ok(invocation_id) => invocation_id,
480            Err(message) => {
481                return Err(ClientError::Protocol {
482                    message,
483                    response: Box::new(response),
484                });
485            }
486        };
487        let acknowledgement: SendAcknowledgement = match serde_json::from_slice(response.body()) {
488            Ok(acknowledgement) => acknowledgement,
489            Err(source) => {
490                return Err(ClientError::Protocol {
491                    message: format!("invalid send acknowledgement: {source}"),
492                    response: Box::new(response),
493                });
494            }
495        };
496        if acknowledgement.invocation_id.is_empty() {
497            return Err(ClientError::Protocol {
498                message: "send acknowledgement contains an empty invocationId".to_owned(),
499                response: Box::new(response),
500            });
501        }
502
503        Ok(SendResponse {
504            handle: client.invocation_handle(invocation_id),
505            status: acknowledgement.status.into(),
506        })
507    }
508
509    fn into_http_request(
510        self,
511        invoke_type: InvokeType,
512        delay: Option<Duration>,
513        metadata: Option<RequestMetadata>,
514    ) -> Result<(Client<E>, HttpRequest<Bytes>), ClientError>
515    where
516        E: RequestExecutor,
517        Req: Serialize,
518    {
519        let Self {
520            client,
521            target,
522            input,
523            headers,
524            idempotency_key,
525            scope,
526            limit_key,
527            response: _,
528        } = self;
529
530        if limit_key.as_deref().is_some_and(|key| !key.is_empty()) && scope.is_none() {
531            return Err(ClientError::InvalidRequest {
532                message: "a limit key requires a scope".to_owned(),
533            });
534        }
535        if let Some(limit_key) = limit_key.as_deref() {
536            validate_limit_key(limit_key)?;
537        }
538
539        let body = input
540            .serialize()
541            .map_err(|source| ClientError::Serialization {
542                source: Box::new(source),
543            })?;
544        let mut merged_headers = client.default_headers.as_ref().clone();
545        for (name, value) in headers {
546            if let Some(name) = name {
547                merged_headers.insert(name, value);
548            }
549        }
550        if let Some(key) = idempotency_key {
551            let value = HeaderValue::try_from(key).map_err(request_error)?;
552            merged_headers.insert(IDEMPOTENCY_KEY, value);
553        }
554        if let Some(key) = limit_key {
555            let value = HeaderValue::try_from(key).map_err(request_error)?;
556            merged_headers.insert(X_RESTATE_LIMIT_KEY, value);
557        }
558
559        if let Some(metadata) = metadata {
560            let should_set_content_type = !body.is_empty()
561                || metadata.set_content_type_if_empty
562                || metadata.input_is_required;
563            if should_set_content_type && !merged_headers.contains_key(CONTENT_TYPE) {
564                let value = HeaderValue::try_from(metadata.content_type).map_err(request_error)?;
565                merged_headers.insert(CONTENT_TYPE, value);
566            }
567        }
568
569        let operation = match invoke_type {
570            InvokeType::Call => "call",
571            InvokeType::Send => "send",
572        };
573        let prefix = match scope {
574            Some(scope) => format!("/restate/scope/{}/{operation}", encode_segment(&scope)),
575            None => format!("/restate/{operation}"),
576        };
577        let mut path_and_query = client.path(&format!("{prefix}/{}", encoded_target_path(&target)));
578        if let Some(delay) = delay {
579            path_and_query.push_str("?delay=");
580            path_and_query.push_str(&format_iso8601_duration(delay));
581        }
582
583        let mut request = HttpRequest::builder()
584            .method(Method::POST)
585            .uri(client.uri(path_and_query)?)
586            .body(body)
587            .map_err(request_error)?;
588        *request.headers_mut() = merged_headers;
589        Ok((client, request))
590    }
591}
592
593struct RequestMetadata {
594    content_type: &'static str,
595    set_content_type_if_empty: bool,
596    input_is_required: bool,
597}
598
599impl RequestMetadata {
600    fn for_payload<T: PayloadMetadata>() -> Self {
601        let output = T::output_metadata();
602        Self {
603            content_type: output.content_type,
604            set_content_type_if_empty: output.set_content_type_if_empty,
605            input_is_required: T::input_metadata().is_required,
606        }
607    }
608}
609
610impl<E, Req, Res> Request<E, Req, Res>
611where
612    Req: Serialize + PayloadMetadata,
613{
614    /// Invokes a handler and waits for its completed response.
615    pub async fn call(self) -> Result<CallResponse<E, Res>, ClientError>
616    where
617        E: RequestExecutor,
618    {
619        self.call_with_metadata(Some(RequestMetadata::for_payload::<Req>()))
620            .await
621    }
622
623    /// Submits a one-way invocation.
624    pub async fn send(self) -> Result<SendResponse<E, Res>, ClientError>
625    where
626        E: RequestExecutor,
627    {
628        self.send_with_delay_and_metadata(None, Some(RequestMetadata::for_payload::<Req>()))
629            .await
630    }
631
632    /// Submits a one-way invocation to execute after `delay`.
633    pub async fn send_after(self, delay: Duration) -> Result<SendResponse<E, Res>, ClientError>
634    where
635        E: RequestExecutor,
636    {
637        self.send_with_delay_and_metadata(Some(delay), Some(RequestMetadata::for_payload::<Req>()))
638            .await
639    }
640}
641
642// This impl is deliberately disjoint from the metadata-bearing impl above: `()` does not
643// implement `PayloadMetadata`, and downstream crates cannot add that impl under Rust's orphan
644// rules. Unit remains the generated client's no-input marker without changing global metadata or
645// discovery behavior.
646impl<E, Res> Request<E, (), Res> {
647    /// Invokes a no-input handler and waits for its completed response.
648    pub async fn call(self) -> Result<CallResponse<E, Res>, ClientError>
649    where
650        E: RequestExecutor,
651    {
652        self.call_with_metadata(None).await
653    }
654
655    /// Submits a one-way no-input invocation.
656    pub async fn send(self) -> Result<SendResponse<E, Res>, ClientError>
657    where
658        E: RequestExecutor,
659    {
660        self.send_with_delay_and_metadata(None, None).await
661    }
662
663    /// Submits a one-way no-input invocation to execute after `delay`.
664    pub async fn send_after(self, delay: Duration) -> Result<SendResponse<E, Res>, ClientError>
665    where
666        E: RequestExecutor,
667    {
668        self.send_with_delay_and_metadata(Some(delay), None).await
669    }
670}
671
672#[derive(Clone, Copy)]
673enum InvokeType {
674    Call,
675    Send,
676}
677
678/// A Restate invocation identifier.
679pub type InvocationId = String;
680
681/// A completed call response and its invocation handle.
682#[must_use = "call responses must be consumed with into_body or used for their invocation handle"]
683pub struct CallResponse<E, Res> {
684    response: HttpResponse<Bytes>,
685    handle: InvocationHandle<E, Res>,
686}
687
688impl<E, Res> CallResponse<E, Res> {
689    /// Returns an invocation handle for this request.
690    pub fn invocation_handle(&self) -> InvocationHandle<E, Res> {
691        self.handle.clone()
692    }
693
694    pub fn status(&self) -> StatusCode {
695        self.response.status()
696    }
697
698    pub fn headers(&self) -> &HeaderMap<HeaderValue> {
699        self.response.headers()
700    }
701
702    /// Get the http response
703    pub fn into_http_response(self) -> Result<HttpResponse<Res>, ClientError>
704    where
705        Res: Deserialize,
706    {
707        decode_response(self.response)
708    }
709
710    /// Get the http response with the raw payload
711    pub fn into_raw_http_response(self) -> HttpResponse<Bytes> {
712        self.response
713    }
714
715    /// Checks if response is 200 OK, then returns the response body.
716    pub fn into_body(self) -> Result<Res, ClientError>
717    where
718        Res: Deserialize,
719    {
720        if self.response.status() != StatusCode::OK {
721            return Err(status_error(self.response));
722        }
723        decode_body(self.response)
724    }
725}
726
727/// The result of submitting a one-way invocation.
728#[derive(Clone, Copy, Debug, PartialEq, Eq)]
729pub enum SendStatus {
730    Accepted,
731    PreviouslyAccepted,
732}
733
734#[derive(serde::Serialize)]
735#[serde(
736    tag = "target",
737    rename_all = "camelCase",
738    rename_all_fields = "camelCase"
739)]
740enum LookupRequest {
741    Workflow {
742        workflow_name: String,
743        workflow_key: String,
744        #[serde(skip_serializing_if = "Option::is_none")]
745        scope: Option<String>,
746    },
747}
748
749#[derive(serde::Deserialize)]
750#[serde(rename_all = "camelCase")]
751struct LookupResponse {
752    invocation_id: InvocationId,
753}
754
755#[derive(serde::Deserialize)]
756#[serde(rename_all = "camelCase")]
757struct SendAcknowledgement {
758    invocation_id: String,
759    status: SendStatusWire,
760}
761
762#[derive(serde::Deserialize)]
763enum SendStatusWire {
764    Accepted,
765    PreviouslyAccepted,
766}
767
768impl From<SendStatusWire> for SendStatus {
769    fn from(value: SendStatusWire) -> Self {
770        match value {
771            SendStatusWire::Accepted => Self::Accepted,
772            SendStatusWire::PreviouslyAccepted => Self::PreviouslyAccepted,
773        }
774    }
775}
776
777/// A successful one-way invocation acknowledgement.
778pub struct SendResponse<E, Res> {
779    handle: InvocationHandle<E, Res>,
780    status: SendStatus,
781}
782
783impl<E, Res> SendResponse<E, Res> {
784    /// Returns the invocation handle to interact with this invocation.
785    pub fn invocation_handle(&self) -> InvocationHandle<E, Res> {
786        self.handle.clone()
787    }
788
789    pub fn send_status(&self) -> SendStatus {
790        self.status
791    }
792}
793
794/// A typed handle to a Restate invocation.
795pub struct InvocationHandle<E, Res> {
796    client: Client<E>,
797    invocation_id: Arc<InvocationId>,
798    response: PhantomData<fn() -> Res>,
799}
800
801impl<E, Res> Clone for InvocationHandle<E, Res> {
802    fn clone(&self) -> Self {
803        Self {
804            client: self.client.clone(),
805            invocation_id: self.invocation_id.clone(),
806            response: PhantomData,
807        }
808    }
809}
810
811impl<E, Res> InvocationHandle<E, Res>
812where
813    E: RequestExecutor,
814    Res: Deserialize,
815{
816    pub fn invocation_id(&self) -> &InvocationId {
817        self.invocation_id.as_ref()
818    }
819
820    /// Waits for the invocation to finish and decodes its output.
821    pub async fn attach(&self) -> Result<HttpResponse<Res>, ClientError> {
822        let response = self
823            .client
824            .get(format!(
825                "/restate/attach/{}",
826                encode_segment(self.invocation_id.as_str())
827            ))
828            .await?;
829        if response.status() != StatusCode::OK {
830            return Err(status_error(response));
831        }
832        decode_response(response)
833    }
834
835    /// Peeks at the invocation output without waiting for completion.
836    pub async fn output(&self) -> Result<HttpResponse<Output<Res>>, ClientError> {
837        let response = self
838            .client
839            .get(format!(
840                "/restate/output/{}",
841                encode_segment(self.invocation_id.as_str())
842            ))
843            .await?;
844        match response.status() {
845            StatusCode::OK => decode_response(response).map(|response| response.map(Output::Ready)),
846            status if status.as_u16() == OUTPUT_NOT_READY => Ok(response.map(|_| Output::NotReady)),
847            _ => Err(status_error(response)),
848        }
849    }
850}
851
852/// Output state returned by [`InvocationHandle::output`].
853#[derive(Clone, Debug, PartialEq, Eq)]
854pub enum Output<T> {
855    Ready(T),
856    NotReady,
857}
858
859fn encode_segment(segment: &str) -> String {
860    utf8_percent_encode(segment, PATH_SEGMENT_ENCODE_SET).to_string()
861}
862
863fn format_iso8601_duration(duration: Duration) -> String {
864    if duration.subsec_nanos() == 0 {
865        return format!("PT{}S", duration.as_secs());
866    }
867    let fraction = format!("{:09}", duration.subsec_nanos());
868    format!(
869        "PT{}.{fraction}S",
870        duration.as_secs(),
871        fraction = fraction.trim_end_matches('0')
872    )
873}
874
875fn validate_limit_key(key: &str) -> Result<(), ClientError> {
876    if key.is_empty() {
877        return Ok(());
878    }
879    let trimmed = key.strip_suffix('/').unwrap_or(key);
880    let components: Vec<_> = trimmed.split('/').collect();
881    if components.len() > 2 {
882        return invalid_request("a limit key can contain at most two components");
883    }
884    for component in components {
885        if component.is_empty() {
886            return invalid_request("limit key components cannot be empty");
887        }
888        if component.len() > 36 {
889            return invalid_request("limit key components cannot exceed 36 bytes");
890        }
891        if !component
892            .bytes()
893            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'.' | b'-'))
894        {
895            return invalid_request(
896                "limit key components may contain only ASCII letters, digits, '_', '.', and '-'",
897            );
898        }
899    }
900    Ok(())
901}
902
903fn invalid_request<T>(message: impl Into<String>) -> Result<T, ClientError> {
904    Err(ClientError::InvalidRequest {
905        message: message.into(),
906    })
907}
908
909fn invocation_id_header(response: &HttpResponse<Bytes>) -> Result<InvocationId, String> {
910    let value = response
911        .headers()
912        .get(&X_RESTATE_ID)
913        .ok_or_else(|| "response is missing x-restate-id".to_owned())?;
914    let value = value
915        .to_str()
916        .map_err(|_| "response has a non-ASCII x-restate-id".to_owned())?;
917    if value.is_empty() {
918        return Err("response has an empty x-restate-id".to_owned());
919    }
920    Ok(InvocationId::from(value))
921}
922
923fn request_error(source: impl Error + Send + Sync + 'static) -> ClientError {
924    ClientError::Request {
925        source: Box::new(source),
926    }
927}
928
929fn status_error(response: HttpResponse<Bytes>) -> ClientError {
930    ClientError::Status {
931        status: response.status(),
932        response: Box::new(response),
933    }
934}
935
936fn decode_body<T: Deserialize>(response: HttpResponse<Bytes>) -> Result<T, ClientError> {
937    let mut body = response.body().clone();
938    T::deserialize(&mut body).map_err(|source| ClientError::PayloadDecode {
939        source: Box::new(source),
940        response: Box::new(response),
941    })
942}
943
944fn decode_response<T: Deserialize>(
945    response: HttpResponse<Bytes>,
946) -> Result<HttpResponse<T>, ClientError> {
947    let (parts, mut body) = response.into_parts();
948    match T::deserialize(&mut body) {
949        Ok(body) => Ok(HttpResponse::from_parts(parts, body)),
950        Err(source) => Err(ClientError::PayloadDecode {
951            source: Box::new(source),
952            response: Box::new(Response::from_parts(parts, body)),
953        }),
954    }
955}
956
957#[cfg(feature = "reqwest-client")]
958impl RequestExecutor for reqwest::Client {
959    type Error = reqwest::Error;
960
961    async fn execute(
962        &self,
963        request: HttpRequest<Bytes>,
964    ) -> Result<HttpResponse<Bytes>, Self::Error> {
965        let request: reqwest::Request = request.try_into()?;
966        let response = self.execute(request).await?;
967        let status = response.status();
968        let version = response.version();
969        let headers = response.headers().clone();
970        let body = response.bytes().await?;
971
972        let mut response = HttpResponse::new(body);
973        *response.status_mut() = status;
974        *response.version_mut() = version;
975        *response.headers_mut() = headers;
976        Ok(response)
977    }
978}
979
980/// Reqwest-backed ingress client, available with the `reqwest-client` feature.
981#[cfg(feature = "reqwest-client")]
982pub type ReqwestClient = Client<reqwest::Client>;
983
984#[cfg(feature = "reqwest-client")]
985impl Client<reqwest::Client> {
986    /// Connects using reqwest's default client configuration.
987    pub fn connect(base_uri: Uri) -> Result<Self, ClientBuildError> {
988        Self::new(base_uri, reqwest::Client::new())
989    }
990}
restate_sdk/ingress/mod.rs

restate_sdk/ingress/
mod.rs
