Port to shared-core 7.0.1 and advertise service protocol v7

Bumps `restate-sdk-shared-core` from 0.10.0 to 7.0.1 and advertises
`maxProtocolVersion = 7` in discovery (min stays 5, matching shared-core's
`minimum_supported_version()..maximum_supported_version()` = V5..V7).

Adapts to the shared-core 7.0.1 API:
- `PayloadOptions::stable()` -> `stable_serialization()`
- `take_output()` now returns `Bytes` (the `TakeOutputResult` enum is gone),
  so the EOF/`UnexpectedOutputClosed` path is removed
- `do_progress(Vec<NotificationHandle>) -> DoProgressResponse` ->
  `do_await(UnresolvedFuture) -> AwaitResponse` (`Single` for single awaits,
  `FirstCompleted` for select)
- `sys_run` returns `RunHandle`, `sys_awakeable` returns `AwakeableHandle`
- `is_replaying()` -> `state().is_replaying()`
- `Target` gains `scope`/`limit_key`; `RetryPolicy` gains `on_max_attempts`

Also flush the output buffer when `do_await` returns `WaitingExternalProgress`:
on protocol v7 shared-core buffers an `AwaitingOnMessage` describing what the
invocation is suspending on, which must reach the runtime before we block on
input.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: slinkydeveloper

Cargo.toml

@@ -40,7 +40,7 @@ pin-project-lite = "0.2"
 rand = { version = "0.10", optional = true }
 regress = "0.10"
 restate-sdk-macros = { version = "0.10", path = "macros" }
-restate-sdk-shared-core = { version = "=0.10.0", features = ["request_identity", "sha2_random_seed", "http"] }
+restate-sdk-shared-core = { version = "=7.0.1", features = ["request_identity", "sha2_random_seed", "http"] }
 schemars = { version = "1.2", optional = true }
 serde = "1.0"
 serde_json = "1.0"

src/endpoint/context.rs

@@ -15,9 +15,9 @@ use futures::future::{BoxFuture, Either, Shared};
 use futures::{FutureExt, TryFutureExt};
 use pin_project_lite::pin_project;
 use restate_sdk_shared_core::{
-    CoreVM, DoProgressResponse, Error as CoreError, Header, NonEmptyValue, NotificationHandle,
-    PayloadOptions, RetryPolicy, RunExitResult, TakeOutputResult, Target, TerminalFailure, VM,
-    Value,
+    AwaitResponse, AwakeableHandle, CoreVM, Error as CoreError, Header, NonEmptyValue,
+    NotificationHandle, OnMaxAttempts, PayloadOptions, RetryPolicy, RunExitResult, RunHandle,
+    Target, TerminalFailure, UnresolvedFuture, VM, Value,
 };
 use std::borrow::Cow;
 use std::collections::HashMap;
@@ -67,10 +67,10 @@ impl ContextInternalInner {
     }
 
     pub(super) fn maybe_flip_span_replaying_field(&mut self) {
-        if !self.span_replaying_field_state && self.vm.is_replaying() {
+        if !self.span_replaying_field_state && self.vm.state().is_replaying() {
             tracing::Span::current().record("restate.sdk.is_replaying", true);
             self.span_replaying_field_state = true;
-        } else if self.span_replaying_field_state && !self.vm.is_replaying() {
+        } else if self.span_replaying_field_state && !self.vm.state().is_replaying() {
             tracing::Span::current().record("restate.sdk.is_replaying", false);
             self.span_replaying_field_state = false;
         }
@@ -131,20 +131,26 @@ impl From<RequestTarget> for Target {
                 handler,
                 key: None,
                 idempotency_key: None,
+                scope: None,
+                limit_key: None,
                 headers: vec![],
             },
             RequestTarget::Object { name, key, handler } => Target {
                 service: name,
                 handler,
                 key: Some(key),
                 idempotency_key: None,
+                scope: None,
+                limit_key: None,
                 headers: vec![],
             },
             RequestTarget::Workflow { name, key, handler } => Target {
                 service: name,
                 handler,
                 key: Some(key),
                 idempotency_key: None,
+                scope: None,
+                limit_key: None,
                 headers: vec![],
             },
         }
@@ -236,9 +242,10 @@ impl ContextInternal {
                     syscall: "input",
                     err: err.0.clone().into(),
                 };
-                let _ = inner_lock
-                    .vm
-                    .sys_write_output(NonEmptyValue::Failure(err.into()), PayloadOptions::stable());
+                let _ = inner_lock.vm.sys_write_output(
+                    NonEmptyValue::Failure(err.into()),
+                    PayloadOptions::stable_serialization(),
+                );
                 let _ = inner_lock.vm.sys_end();
                 // This causes the trap, plus logs the error
                 inner_lock.handler_state.mark_error(error_inner.into());
@@ -261,7 +268,7 @@ impl ContextInternal {
             inner_lock,
             inner_lock
                 .vm
-                .sys_state_get(key.to_owned(), PayloadOptions::stable())
+                .sys_state_get(key.to_owned(), PayloadOptions::stable_serialization())
         );
         inner_lock.maybe_flip_span_replaying_field();
 
@@ -307,9 +314,11 @@ impl ContextInternal {
         let mut inner_lock = must_lock!(self.inner);
         match t.serialize() {
             Ok(b) => {
-                let _ = inner_lock
-                    .vm
-                    .sys_state_set(key.to_owned(), b, PayloadOptions::stable());
+                let _ = inner_lock.vm.sys_state_set(
+                    key.to_owned(),
+                    b,
+                    PayloadOptions::stable_serialization(),
+                );
                 inner_lock.maybe_flip_span_replaying_field();
             }
             Err(e) => {
@@ -402,7 +411,7 @@ impl ContextInternal {
             .and_then(|input| {
                 inner_lock
                     .vm
-                    .sys_call(target, input, None, PayloadOptions::stable())
+                    .sys_call(target, input, None, PayloadOptions::stable_serialization())
                     .map_err(Into::into)
             });
 
@@ -501,7 +510,7 @@ impl ContextInternal {
                     + delay
             }),
             None,
-            PayloadOptions::stable(),
+            PayloadOptions::stable_serialization(),
         ) {
             Ok(h) => h,
             Err(e) => {
@@ -554,7 +563,7 @@ impl ContextInternal {
         inner_lock.maybe_flip_span_replaying_field();
 
         let (awakeable_id, handle) = match maybe_awakeable_id_and_handle {
-            Ok((s, handle)) => (s, handle),
+            Ok(AwakeableHandle { id, handle }) => (id, handle),
             Err(e) => {
                 inner_lock.fail(e.into());
                 return (
@@ -597,7 +606,7 @@ impl ContextInternal {
                 let _ = inner_lock.vm.sys_complete_awakeable(
                     id.to_owned(),
                     NonEmptyValue::Success(b),
-                    PayloadOptions::stable(),
+                    PayloadOptions::stable_serialization(),
                 );
             }
             Err(e) => {
@@ -610,7 +619,7 @@ impl ContextInternal {
         let _ = must_lock!(self.inner).vm.sys_complete_awakeable(
             id.to_owned(),
             NonEmptyValue::Failure(failure.into()),
-            PayloadOptions::stable(),
+            PayloadOptions::stable_serialization(),
         );
     }
 
@@ -679,7 +688,7 @@ impl ContextInternal {
                 let _ = inner_lock.vm.sys_complete_promise(
                     name.to_owned(),
                     NonEmptyValue::Success(b),
-                    PayloadOptions::stable(),
+                    PayloadOptions::stable_serialization(),
                 );
             }
             Err(e) => {
@@ -698,7 +707,7 @@ impl ContextInternal {
         let _ = must_lock!(self.inner).vm.sys_complete_promise(
             id.to_owned(),
             NonEmptyValue::Failure(failure.into()),
-            PayloadOptions::stable(),
+            PayloadOptions::stable_serialization(),
         );
     }
 
@@ -744,7 +753,7 @@ impl ContextInternal {
 
         let _ = inner_lock
             .vm
-            .sys_write_output(res_to_write, PayloadOptions::stable());
+            .sys_write_output(res_to_write, PayloadOptions::stable_serialization());
         inner_lock.maybe_flip_span_replaying_field();
     }
 
@@ -755,11 +764,8 @@ impl ContextInternal {
     pub(crate) fn consume_to_end(&self) {
         let mut inner_lock = must_lock!(self.inner);
 
-        let out = inner_lock.vm.take_output();
-        if let TakeOutputResult::Buffer(b) = out
-            && !b.is_empty()
-            && !inner_lock.write.send(b)
-        {
+        let b = inner_lock.vm.take_output();
+        if !b.is_empty() && !inner_lock.write.send(b) {
             // Nothing we can do anymore here
         }
     }
@@ -879,6 +885,7 @@ where
             max_interval: retry_policy.max_delay,
             max_attempts: retry_policy.max_attempts,
             max_duration: retry_policy.max_duration,
+            on_max_attempts: OnMaxAttempts::FailAsTerminal,
         };
         self
     }
@@ -911,14 +918,14 @@ where
                         .expect("Future should not be polled after returning Poll::Ready");
                     let mut inner_ctx = must_lock!(ctx);
 
-                    let handle = inner_ctx
+                    let RunHandle { handle, .. } = inner_ctx
                         .vm
                         .sys_run(this.name.to_owned())
                         .map_err(ErrorInner::from)?;
 
                     // Now we do progress once to check whether this closure should be executed or not.
-                    match inner_ctx.vm.do_progress(vec![handle]) {
-                        Ok(DoProgressResponse::ExecuteRun(handle_to_run)) => {
+                    match inner_ctx.vm.do_await(UnresolvedFuture::Single(handle)) {
+                        Ok(AwaitResponse::ExecuteRun(handle_to_run)) => {
                             // In case it returns ExecuteRun, it must be the handle we just gave it,
                             // and it means we need to execute the closure
                             assert_eq!(handle, handle_to_run);
@@ -931,7 +938,7 @@ where
                                 closure_fut: closure.run(),
                             });
                         }
-                        Ok(DoProgressResponse::CancelSignalReceived) => {
+                        Ok(AwaitResponse::CancelSignalReceived) => {
                             drop(inner_ctx);
                             // Got cancellation!
                             this.state.set(RunState::WaitingResultFut {

src/endpoint/futures/async_result_poll.rs

@@ -1,8 +1,8 @@
 use crate::endpoint::ErrorInner;
 use crate::endpoint::context::ContextInternalInner;
 use restate_sdk_shared_core::{
-    DoProgressResponse, Error as CoreError, NotificationHandle, TakeOutputResult, TerminalFailure,
-    VM, Value,
+    AwaitResponse, Error as CoreError, NotificationHandle, TerminalFailure, UnresolvedFuture, VM,
+    Value,
 };
 use std::future::Future;
 use std::pin::Pin;
@@ -55,21 +55,14 @@ impl Future for VmAsyncResultPollFuture {
                 AsyncResultPollState::Init { ctx, handle } => {
                     let mut inner_lock = must_lock!(ctx);
 
-                    // Let's consume some output to begin with
-                    let out = inner_lock.vm.take_output();
-                    match out {
-                        TakeOutputResult::Buffer(b) => {
-                            // Skip empty buffers: take_output returns an empty buffer when there's
-                            // nothing to send (e.g. while replaying completed entries). Sending it
-                            // would emit an empty HTTP/2 DATA frame per replayed await, which some
-                            // proxies (e.g. Envoy) reject when consecutive. See sdk-rust#114.
-                            if !b.is_empty() && !inner_lock.write.send(b) {
-                                return Poll::Ready(Err(ErrorInner::Suspended));
-                            }
-                        }
-                        TakeOutputResult::EOF => {
-                            return Poll::Ready(Err(ErrorInner::UnexpectedOutputClosed));
-                        }
+                    // Let's consume some output to begin with.
+                    // Skip empty buffers: take_output returns an empty buffer when there's
+                    // nothing to send (e.g. while replaying completed entries). Sending it
+                    // would emit an empty HTTP/2 DATA frame per replayed await, which some
+                    // proxies (e.g. Envoy) reject when consecutive. See sdk-rust#114.
+                    let b = inner_lock.vm.take_output();
+                    if !b.is_empty() && !inner_lock.write.send(b) {
+                        return Poll::Ready(Err(ErrorInner::Suspended));
                     }
 
                     // We can now start polling
@@ -106,22 +99,32 @@ impl Future for VmAsyncResultPollFuture {
                 AsyncResultPollState::PollProgress { ctx, handle } => {
                     let mut inner_lock = must_lock!(ctx);
 
-                    match inner_lock.vm.do_progress(vec![handle]) {
-                        Ok(DoProgressResponse::AnyCompleted) => {
+                    match inner_lock.vm.do_await(UnresolvedFuture::Single(handle)) {
+                        Ok(AwaitResponse::AnyCompleted) => {
                             // We're good, we got the response
                         }
-                        Ok(DoProgressResponse::ReadFromInput) => {
+                        Ok(AwaitResponse::WaitingExternalProgress {
+                            waiting_input: true,
+                            ..
+                        }) => {
+                            // do_await buffered an AwaitingOnMessage describing what we're
+                            // suspending on; flush it to the runtime before we block on input,
+                            // otherwise the runtime never learns what we're waiting for.
+                            let b = inner_lock.vm.take_output();
+                            if !b.is_empty() && !inner_lock.write.send(b) {
+                                return Poll::Ready(Err(ErrorInner::Suspended));
+                            }
                             drop(inner_lock);
                             self.state = Some(AsyncResultPollState::WaitingInput { ctx, handle });
                             continue;
                         }
-                        Ok(DoProgressResponse::ExecuteRun(_)) => {
-                            unimplemented!()
-                        }
-                        Ok(DoProgressResponse::WaitingPendingRun) => {
+                        Ok(AwaitResponse::WaitingExternalProgress { .. })
+                        | Ok(AwaitResponse::ExecuteRun(_)) => {
+                            // Waiting only on a run proposal is not expected on this await path.
+                            // These two are used by the shared-core to implement async run, which is not supported by the rust sdk currently.
                             unimplemented!()
                         }
-                        Ok(DoProgressResponse::CancelSignalReceived) => {
+                        Ok(AwaitResponse::CancelSignalReceived) => {
                             return Poll::Ready(Ok(Value::Failure(TerminalFailure {
                                 code: 409,
                                 message: "cancelled".to_string(),

src/endpoint/futures/select_poll.rs

@@ -2,8 +2,7 @@ use crate::endpoint::ErrorInner;
 use crate::endpoint::context::ContextInternalInner;
 use crate::errors::TerminalError;
 use restate_sdk_shared_core::{
-    DoProgressResponse, Error as CoreError, NotificationHandle, TakeOutputResult, TerminalFailure,
-    VM,
+    AwaitResponse, Error as CoreError, NotificationHandle, TerminalFailure, UnresolvedFuture, VM,
 };
 use std::future::Future;
 use std::pin::Pin;
@@ -56,21 +55,14 @@ impl Future for VmSelectAsyncResultPollFuture {
                 VmSelectAsyncResultPollState::Init { ctx, handles } => {
                     let mut inner_lock = must_lock!(ctx);
 
-                    // Let's consume some output to begin with
-                    let out = inner_lock.vm.take_output();
-                    match out {
-                        TakeOutputResult::Buffer(b) => {
-                            // Skip empty buffers: take_output returns an empty buffer when there's
-                            // nothing to send (e.g. while replaying completed entries). Sending it
-                            // would emit an empty HTTP/2 DATA frame per replayed await, which some
-                            // proxies (e.g. Envoy) reject when consecutive. See sdk-rust#114.
-                            if !b.is_empty() && !inner_lock.write.send(b) {
-                                return Poll::Ready(Err(ErrorInner::Suspended));
-                            }
-                        }
-                        TakeOutputResult::EOF => {
-                            return Poll::Ready(Err(ErrorInner::UnexpectedOutputClosed));
-                        }
+                    // Let's consume some output to begin with.
+                    // Skip empty buffers: take_output returns an empty buffer when there's
+                    // nothing to send (e.g. while replaying completed entries). Sending it
+                    // would emit an empty HTTP/2 DATA frame per replayed await, which some
+                    // proxies (e.g. Envoy) reject when consecutive. See sdk-rust#114.
+                    let b = inner_lock.vm.take_output();
+                    if !b.is_empty() && !inner_lock.write.send(b) {
+                        return Poll::Ready(Err(ErrorInner::Suspended));
                     }
 
                     // We can now start polling
@@ -108,23 +100,40 @@ impl Future for VmSelectAsyncResultPollFuture {
                 VmSelectAsyncResultPollState::PollProgress { ctx, handles } => {
                     let mut inner_lock = must_lock!(ctx);
 
-                    match inner_lock.vm.do_progress(handles.clone()) {
-                        Ok(DoProgressResponse::AnyCompleted) => {
+                    let unresolved_future = UnresolvedFuture::FirstCompleted(
+                        handles
+                            .iter()
+                            .copied()
+                            .map(UnresolvedFuture::Single)
+                            .collect(),
+                    );
+                    match inner_lock.vm.do_await(unresolved_future) {
+                        Ok(AwaitResponse::AnyCompleted) => {
                             // We're good, we got the response
                         }
-                        Ok(DoProgressResponse::ReadFromInput) => {
+                        Ok(AwaitResponse::WaitingExternalProgress {
+                            waiting_input: true,
+                            ..
+                        }) => {
+                            // do_await buffered an AwaitingOnMessage describing what we're
+                            // suspending on; flush it to the runtime before we block on input,
+                            // otherwise the runtime never learns what we're waiting for.
+                            let b = inner_lock.vm.take_output();
+                            if !b.is_empty() && !inner_lock.write.send(b) {
+                                return Poll::Ready(Err(ErrorInner::Suspended));
+                            }
                             drop(inner_lock);
                             self.state =
                                 Some(VmSelectAsyncResultPollState::WaitingInput { ctx, handles });
                             continue;
                         }
-                        Ok(DoProgressResponse::ExecuteRun(_)) => {
-                            unimplemented!()
-                        }
-                        Ok(DoProgressResponse::WaitingPendingRun) => {
+                        Ok(AwaitResponse::WaitingExternalProgress { .. })
+                        | Ok(AwaitResponse::ExecuteRun(_)) => {
+                            // Waiting only on a run proposal is not expected on this await path.
+                            // These two are used by the shared-core to implement async run, which is not supported by the rust sdk currently.
                             unimplemented!()
                         }
-                        Ok(DoProgressResponse::CancelSignalReceived) => {
+                        Ok(AwaitResponse::CancelSignalReceived) => {
                             return Poll::Ready(Ok(Err(TerminalFailure {
                                 code: 409,
                                 message: "cancelled".to_string(),