[WIP] Service Protocol V4

.github/workflows/integration.yaml

@@ -105,7 +105,7 @@ jobs:
           cache-to: type=gha,mode=max,scope=${{ github.workflow }}
 
       - name: Run test tool
-        uses: restatedev/sdk-test-suite@v2.4
+        uses: restatedev/sdk-test-suite@v3.0
         with:
           restateContainerImage: ${{ inputs.restateCommit != '' && 'localhost/restatedev/restate-commit-download:latest' || (inputs.restateImage != '' && inputs.restateImage || 'ghcr.io/restatedev/restate:main') }}
           serviceContainerImage: "restatedev/rust-test-services"

Cargo.toml

@@ -12,6 +12,7 @@ default = ["http_server", "rand", "uuid"]
 hyper = ["dep:hyper", "http-body-util", "restate-sdk-shared-core/http"]
 http_server = ["hyper", "hyper/server", "hyper/http2", "hyper-util", "tokio/net", "tokio/signal", "tokio/macros"]
 
+
 [dependencies]
 bytes = "1.6.1"
 futures = "0.3"
@@ -23,8 +24,7 @@ pin-project-lite = "0.2"
 rand = { version = "0.8.5", optional = true }
 regress = "0.10"
 restate-sdk-macros = { version = "0.3.2", path = "macros" }
-restate-sdk-shared-core = "0.1.0"
-sha2 = "=0.11.0-pre.3"
+restate-sdk-shared-core = { git = "https://github.com/restatedev/sdk-shared-core.git", branch = "main", features = ["request_identity", "sha2_random_seed", "http"] }
 serde = "1.0"
 serde_json = "1.0"
 thiserror = "1.0.63"

README.md

@@ -121,10 +121,11 @@ The Rust SDK is currently in active development, and might break across releases
 
 The compatibility with Restate is described in the following table:
 
-| Restate Server\sdk-rust | 0.0/0.1/0.2 | 0.3 |
-|-------------------------|-------------|-----|
-| 1.0                     | ✅           | ❌   |
-| 1.1                     | ✅           | ✅   |
+| Restate Server\sdk-rust | 0.0/0.1/0.2 | 0.3 | 0.4 |
+|-------------------------|-------------|-----|-----|
+| 1.0                     | ✅           | ❌   | ❌   |
+| 1.1                     | ✅           | ✅   | ❌   |
+| 1.2                     | ✅           | ✅   | ✅   |
 
 ## Contributing
 

examples/cron.rs

@@ -77,7 +77,7 @@ impl PeriodicTaskImpl {
             .object_client::<PeriodicTaskClient>(context.key())
             .run()
             // And send with a delay
-            .send_with_delay(Duration::from_secs(10));
+            .send_after(Duration::from_secs(10));
     }
 }
 

src/context/mod.rs

@@ -8,7 +8,7 @@ use std::time::Duration;
 
 mod request;
 mod run;
-pub use request::{Request, RequestTarget};
+pub use request::{CallFuture, InvocationHandle, Request, RequestTarget};
 pub use run::{RunClosure, RunFuture, RunRetryPolicy};
 
 pub type HeaderMap = http::HeaderMap<String>;
@@ -370,20 +370,20 @@ impl<'ctx, CTX: private::SealedContext<'ctx>> ContextTimers<'ctx> for CTX {}
 ///     // To a Service:
 ///     ctx.service_client::<MyServiceClient>()
 ///         .my_handler(String::from("Hi!"))
-///         .send_with_delay(Duration::from_millis(5000));
+///         .send_after(Duration::from_millis(5000));
 ///
 ///     // To a Virtual Object:
 ///     ctx.object_client::<MyVirtualObjectClient>("Mary")
 ///         .my_handler(String::from("Hi!"))
-///         .send_with_delay(Duration::from_millis(5000));
+///         .send_after(Duration::from_millis(5000));
 ///
 ///     // To a Workflow:
 ///     ctx.workflow_client::<MyWorkflowClient>("my-workflow-id")
 ///         .run(String::from("Hi!"))
-///         .send_with_delay(Duration::from_millis(5000));
+///         .send_after(Duration::from_millis(5000));
 ///     ctx.workflow_client::<MyWorkflowClient>("my-workflow-id")
 ///         .interact_with_workflow()
-///         .send_with_delay(Duration::from_millis(5000));
+///         .send_after(Duration::from_millis(5000));
 ///  #    Ok(())
 ///  # }
 /// ```
@@ -433,6 +433,11 @@ pub trait ContextClient<'ctx>: private::SealedContext<'ctx> {
         Request::new(self.inner_context(), request_target, req)
     }
 
+    /// Create an [`InvocationHandle`] from an invocation id.
+    fn invocation_handle(&self, invocation_id: String) -> impl InvocationHandle + 'ctx {
+        self.inner_context().invocation_handle(invocation_id)
+    }
+
     /// Create a service client. The service client is generated by the [`restate_sdk_macros::service`] macro with the same name of the trait suffixed with `Client`.
     ///
     /// ```rust,no_run
@@ -454,7 +459,7 @@ pub trait ContextClient<'ctx>: private::SealedContext<'ctx> {
     /// client.handle().send();
     ///
     /// // Schedule the request to be executed later
-    /// client.handle().send_with_delay(Duration::from_secs(60));
+    /// client.handle().send_after(Duration::from_secs(60));
     /// # }
     /// ```
     fn service_client<C>(&self) -> C
@@ -485,7 +490,7 @@ pub trait ContextClient<'ctx>: private::SealedContext<'ctx> {
     /// client.handle().send();
     ///
     /// // Schedule the request to be executed later
-    /// client.handle().send_with_delay(Duration::from_secs(60));
+    /// client.handle().send_after(Duration::from_secs(60));
     /// # }
     /// ```
     fn object_client<C>(&self, key: impl Into<String>) -> C
@@ -516,7 +521,7 @@ pub trait ContextClient<'ctx>: private::SealedContext<'ctx> {
     /// client.handle().send();
     ///
     /// // Schedule the request to be executed later
-    /// client.handle().send_with_delay(Duration::from_secs(60));
+    /// client.handle().send_after(Duration::from_secs(60));
     /// # }
     /// ```
     fn workflow_client<C>(&self, key: impl Into<String>) -> C
@@ -627,7 +632,7 @@ pub trait ContextAwakeables<'ctx>: private::SealedContext<'ctx> {
         &self,
     ) -> (
         String,
-        impl Future<Output = Result<T, TerminalError>> + Send + Sync + 'ctx,
+        impl Future<Output = Result<T, TerminalError>> + Send + 'ctx,
     ) {
         self.inner_context().awakeable()
     }

src/context/request.rs

@@ -72,6 +72,7 @@ impl fmt::Display for RequestTarget {
 pub struct Request<'a, Req, Res = ()> {
     ctx: &'a ContextInternal,
     request_target: RequestTarget,
+    idempotency_key: Option<String>,
     req: Req,
     res: PhantomData<Res>,
 }
@@ -81,33 +82,54 @@ impl<'a, Req, Res> Request<'a, Req, Res> {
         Self {
             ctx,
             request_target,
+            idempotency_key: None,
             req,
             res: PhantomData,
         }
     }
 
+    /// Add idempotency key to the request
+    pub fn idempotency_key(mut self, idempotency_key: impl Into<String>) -> Self {
+        self.idempotency_key = Some(idempotency_key.into());
+        self
+    }
+
     /// Call a service. This returns a future encapsulating the response.
-    pub fn call(self) -> impl Future<Output = Result<Res, TerminalError>> + Send
+    pub fn call(self) -> impl CallFuture<Result<Res, TerminalError>> + Send
     where
         Req: Serialize + 'static,
         Res: Deserialize + 'static,
     {
-        self.ctx.call(self.request_target, self.req)
+        self.ctx
+            .call(self.request_target, self.idempotency_key, self.req)
     }
 
     /// Send the request to the service, without waiting for the response.
-    pub fn send(self)
+    pub fn send(self) -> impl InvocationHandle
     where
         Req: Serialize + 'static,
     {
-        self.ctx.send(self.request_target, self.req, None)
+        self.ctx
+            .send(self.request_target, self.idempotency_key, self.req, None)
     }
 
     /// Schedule the request to the service, without waiting for the response.
-    pub fn send_with_delay(self, duration: Duration)
+    pub fn send_after(self, delay: Duration) -> impl InvocationHandle
     where
         Req: Serialize + 'static,
     {
-        self.ctx.send(self.request_target, self.req, Some(duration))
+        self.ctx.send(
+            self.request_target,
+            self.idempotency_key,
+            self.req,
+            Some(delay),
+        )
     }
 }
+
+pub trait InvocationHandle {
+    fn invocation_id(&self) -> impl Future<Output = Result<String, TerminalError>> + Send;
+    fn cancel(&self) -> impl Future<Output = Result<(), TerminalError>> + Send;
+}
+
+pub trait CallFuture<O>: Future<Output = O> + InvocationHandle {}