Add changes for the e2e tests

Author: slinkydeveloper

Cargo.toml

@@ -23,7 +23,7 @@ pin-project-lite = "0.2"
 rand = { version = "0.8.5", optional = true }
 regress = "0.10"
 restate-sdk-macros = { version = "0.3.0", path = "macros" }
-restate-sdk-shared-core = "0.1.0"
+restate-sdk-shared-core = { git = "https://github.com/restatedev/sdk-shared-core.git", branch = "issues/cancel-invocation-get-invocation-id" }
 serde = "1.0"
 serde_json = "1.0"
 thiserror = "1.0.63"

test-services/Dockerfile

@@ -7,5 +7,6 @@ RUN cargo build -p test-services
 RUN cp ./target/debug/test-services /bin/server
 
 ENV RUST_LOG="debug,restate_shared_core=trace"
+ENV RUST_BACKTRACE=1
 
 CMD ["/bin/server"]

test-services/src/proxy.rs

@@ -46,7 +46,7 @@ pub(crate) trait Proxy {
     #[name = "call"]
     async fn call(req: Json<ProxyRequest>) -> HandlerResult<Json<Vec<u8>>>;
     #[name = "oneWayCall"]
-    async fn one_way_call(req: Json<ProxyRequest>) -> HandlerResult<()>;
+    async fn one_way_call(req: Json<ProxyRequest>) -> HandlerResult<String>;
     #[name = "manyCalls"]
     async fn many_calls(req: Json<Vec<ManyCallRequest>>) -> HandlerResult<()>;
 }
@@ -70,16 +70,16 @@ impl Proxy for ProxyImpl {
         &self,
         ctx: Context<'_>,
         Json(req): Json<ProxyRequest>,
-    ) -> HandlerResult<()> {
+    ) -> HandlerResult<String> {
         let request = ctx.request::<_, ()>(req.to_target(), req.message);
 
-        if let Some(delay_millis) = req.delay_millis {
-            request.send_with_delay(Duration::from_millis(delay_millis));
+        let invocation_id = if let Some(delay_millis) = req.delay_millis {
+            request.send_with_delay(Duration::from_millis(delay_millis)).invocation_id().await?
         } else {
-            request.send();
-        }
+            request.send().invocation_id().await?
+        };
 
-        Ok(())
+        Ok(invocation_id)
     }
 
     async fn many_calls(

test-services/src/test_utils_service.rs

@@ -5,6 +5,7 @@ use futures::FutureExt;
 use restate_sdk::prelude::*;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use std::convert::Infallible;
 use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
@@ -62,6 +63,8 @@ pub(crate) trait TestUtilsService {
     async fn count_executed_side_effects(increments: u32) -> HandlerResult<u32>;
     #[name = "getEnvVariable"]
     async fn get_env_variable(env: String) -> HandlerResult<String>;
+    #[name = "cancelInvocation"]
+    async fn cancel_invocation(invocation_id: String) -> Result<(), Infallible>;
     #[name = "interpretCommands"]
     async fn interpret_commands(req: Json<InterpretRequest>) -> HandlerResult<()>;
 }
@@ -155,6 +158,12 @@ impl TestUtilsService for TestUtilsServiceImpl {
         Ok(std::env::var(env).ok().unwrap_or_default())
     }
 
+    async fn cancel_invocation(&self, ctx: Context<'_>, invocation_id: String) -> Result<(), Infallible> {
+        ctx.invocation_handle(invocation_id)
+            .cancel();
+        Ok(())
+    }
+
     async fn interpret_commands(
         &self,
         context: Context<'_>,