Service protocol V8

docs/sdk-integration.md

@@ -159,6 +159,7 @@ loop {
         AnyCompleted             => { /* loop again — at least one handle is ready */ }
         ExecuteRun(handle)       => execute_run_closure(handle),
         CancelSignalReceived     => return cancelled(),
+        EphemeralNotificationReady(id) => complete_ephemeral(id, vm.take_ephemeral_notification(id)?),
         WaitingExternalProgress  => { flush_output(); await_external_progress(); }
     }
 }
@@ -169,7 +170,7 @@ The `future` passed to `do_progress` is a tree of handles combined with promise-
 tree is reused as the `SuspensionMessage` payload if the VM decides to suspend — so combinators are
 first-class, not a client-side concern.
 
-The four responses:
+The five responses:
 
 - **`AnyCompleted`** — at least one leaf moved. The caller drains the leaves that completed via
   `take_notification` and re-enters the loop if the user-facing combinator still has unresolved
@@ -185,6 +186,10 @@ The four responses:
   [`ctx.run`](#7--ctxrun).
 - **`CancelSignalReceived`** — only produced when implicit cancellation is enabled. The VM has
   observed a `CANCEL` signal; the SDK should propagate cancellation to the user-facing future.
+- **`EphemeralNotificationReady(id)`** — the notification of an ephemeral command (e.g.
+  `ephemeral_state_get`) is ready. It takes precedence over every other response. The SDK takes it
+  with `take_ephemeral_notification(id)`, completes the SDK-side promise associated with `id`, and
+  re-enters the loop. See [Ephemeral commands](#ephemeral-commands).
 
 When the SDK uses an eager input pump (see [§3](#3--input-and-output)), the two external sources of
 progress — fresh input bytes and a freshly-proposed run completion — both need to wake the await
@@ -217,6 +222,30 @@ The VM combines the SDK-supplied retry policy with the runtime's default to deci
 retryable failures (retry with delay, pause, or fail). The SDK does not implement the retry decision
 itself.
 
+### Ephemeral commands
+
+Ephemeral commands (protocol V8+) are commands that are neither recorded in the journal, nor replayed,
+as opposed to journaled commands. Their results are delivered as ephemeral notifications, which are
+not recorded either.
+
+`ephemeral_state_get(key)` reads a state value without recording it in the journal, and returns an
+ephemeral completion id. Ephemeral completion ids are scoped to the attempt and independent of notification
+handles. The SDK keeps a map from ephemeral completion id to a plain (non-combinable) SDK promise, and only
+the progress loop completes those promises, via `EphemeralNotificationReady`.
+
+Because the value is not recorded, the SDK should use it only inside a run closure, e.g. a
+`ctx.get(key, projection)` API can be implemented as
+`ctx.run("project:<key>", async () => projection(await ephemeralStateGet(key)))`.
+
+The VM answers locally when it knows the value (eager state, or state modified by the invocation),
+otherwise it writes a `GetStateEphemeralCommandMessage` to the output. In both cases the notification surfaces only
+through `do_progress`, so after calling `ephemeral_state_get` the SDK must make sure the progress loop
+wakes up: signal the external progress channel, so the loop polls the VM again, and flushes the output
+before waiting again.
+
+If the input stream closes while an ephemeral command is in flight, `do_progress` fails with a retryable error
+rather than suspending: the notification can't arrive anymore, and it can't be recovered on resume because it is not recorded.
+
 ## 8 — Terminating the invocation
 
 When user code completes, the SDK commits the result via `sys_write_output` (with the success or

docs/service-invocation-protocol.md

@@ -132,7 +132,9 @@ The 16-bit type field is partitioned into namespaces:
 
 | Range           | Namespace                                                                |
 |-----------------|--------------------------------------------------------------------------|
-| `0x0000–0x03FF` | Control frames (Start, Suspension, End, Error, …)                        |
+| `0x0000–0x01FF` | Control frames (Start, Suspension, End, Error, …)                        |
+| `0x0200–0x02FF` | Ephemeral commands (see [Ephemeral commands](#ephemeral-commands))       |
+| `0x0300–0x03FF` | Ephemeral notifications, paired with their command by the low byte       |
 | `0x0400–0x7FFF` | Commands                                                                 |
 | `0x8000–0xFBFF` | Notifications                                                            |
 | `0xFC00–0xFFFF` | Custom commands (see [Custom commands](#custom-commands))                |
@@ -394,6 +396,33 @@ When user code reads key `k`:
 of `state_map` so subsequent reads within the same invocation see the new value.
 `GetEagerStateKeysCommand` follows the same pattern for the set of known keys.
 
+## Ephemeral commands
+
+Since V8, the SDK can issue **ephemeral commands**, as opposed to journaled commands: neither the
+command nor its result (the ephemeral notification) is stored, and neither is replayed. On the wire they
+are control frames, an ephemeral command and its ephemeral notification, correlated by an `ephemeral_completion_id`
+that is scoped to the current attempt (starting from 1) and independent of completion ids.
+
+Because their result is not recorded, the SDK MUST only use it in a context where its effect is recorded
+otherwise, typically within a `ctx.run` closure whose result is then journaled.
+
+| Ephemeral command                 | Ephemeral notification                 | Description                                                  |
+|-----------------------------------|----------------------------------------|--------------------------------------------------------------|
+| `GetStateEphemeralCommandMessage` | `GetStateEphemeralNotificationMessage` | Read the current value of a state key (`void` if not found). |
+
+Rules:
+
+- The SDK sends ephemeral commands only in the PROCESSING phase.
+- The runtime can answer only while the request stream is open, so ephemeral commands are not
+  available in request/response mode. The SDK MUST NOT wait for an ephemeral notification once the input stream is
+  closed. Suspending does not help either, as the notification is not recorded, so the SDK fails the attempt with a
+  retryable error.
+- The runtime reads the latest committed state; it doesn't guarantee read-your-writes for state commands
+  the SDK sent in the current attempt that haven't been applied yet. The SDK serves keys it knows about
+  locally, without sending `GetStateEphemeralCommandMessage`: keys present in the eager state, and keys
+  set or cleared by the invocation itself (see [Eager state](#eager-state)). This also makes ephemeral
+  state reads usable in request/response mode, as long as the eager state covers the key.
+
 ## Custom commands
 
 The SDK MAY register custom commands in the `0xFC00–0xFFFF` range. The runtime treats them as opaque

service-protocol/dev/restate/service/protocol.proto

@@ -47,6 +47,9 @@ enum ServiceProtocolVersion {
   // * Semantic changes to Run proposal response, introduced ProposeRunCompletionAckMessage
   // * ErrorMessage.behavior to customize retry behavior
   V7 = 7;
+  // Added:
+  // * Ephemeral commands: GetStateEphemeralCommandMessage and GetStateEphemeralNotificationMessage
+  V8 = 8;
 }
 
 // --- Core frames ---
@@ -263,6 +266,40 @@ message ProposeRunCompletionAckMessage {
   uint32 completion_id = 1;
 }
 
+// --- Ephemeral commands and Notifications ---
+
+// Ephemeral commands, as opposed to journaled commands, are not part of the journal:
+// neither the ephemeral command, nor its ephemeral notification are stored, nor replayed.
+// They don't count as journal entries, and they don't advance the command or notification indexes.
+//
+// Ephemeral commands use the message types 0x0200 - 0x02FF, ephemeral notifications 0x0300 - 0x03FF.
+//
+// The server can send ephemeral notifications only while the request stream is open,
+// so ephemeral commands are not available in request/response mode.
+// If the request stream gets closed while an ephemeral command is in-flight,
+// the SDK cannot suspend waiting for its notification: it fails the attempt instead.
+//
+// The SDK ignores ephemeral notifications for unknown ephemeral commands,
+// or for ephemeral commands sent in another attempt.
+
+// Ephemeral command to get state.
+//
+// Type: 0x0200 + 0
+message GetStateEphemeralCommandMessage {
+  uint32 ephemeral_completion_id = 1;
+  bytes key = 2;
+}
+
+// Notification of GetStateEphemeralCommandMessage.
+// Type: 0x0300 + 0
+message GetStateEphemeralNotificationMessage {
+  uint32 ephemeral_completion_id = 1;
+  oneof result {
+    Void void = 4;
+    Value value = 5;
+  }
+}
+
 // --- Commands and Notifications ---
 
 // The Journal is modelled as commands and notifications.

src/fmt.rs

@@ -35,7 +35,9 @@ macro_rules! delegate_to_formatter {
     };
 }
 
-delegate_to_formatter!(display_closed_error(f: &mut fmt::Formatter<'_>, event: &str) -> fmt::Result);
+delegate_to_formatter!(
+    display_closed_error(f: &mut fmt::Formatter<'_>, event: &str) -> fmt::Result
+);
 
 pub(crate) struct DiffFormatter<'a, 'b> {
     fmt: &'a mut fmt::Formatter<'b>,

src/lib.rs

@@ -172,6 +172,27 @@ impl From<NotificationHandle> for u32 {
     }
 }
 
+/// Ephemeral completion id, identifying an ephemeral command and the ephemeral notification completing it, e.g. [VM::ephemeral_state_get].
+///
+/// Ephemeral commands are neither recorded in the journal, nor replayed.
+/// Their result is delivered as an ephemeral notification, see [AwaitResponse::EphemeralNotificationReady].
+///
+/// Ids are scoped to the current attempt and are unrelated to [NotificationHandle]s.
+#[derive(Debug, Hash, Clone, Copy, Eq, PartialEq, Ord, PartialOrd)]
+pub struct EphemeralCompletionId(u32);
+
+impl From<u32> for EphemeralCompletionId {
+    fn from(value: u32) -> Self {
+        EphemeralCompletionId(value)
+    }
+}
+
+impl From<EphemeralCompletionId> for u32 {
+    fn from(value: EphemeralCompletionId) -> Self {
+        value.0
+    }
+}
+
 #[derive(Debug, Clone, Copy, Eq, PartialEq)]
 pub struct CallHandle {
     pub invocation_id_notification_handle: NotificationHandle,
@@ -403,6 +424,12 @@ pub enum AwaitResponse {
     ExecuteRun(NotificationHandle),
     /// Returned only when [ImplicitCancellationOption::Enabled].
     CancelSignalReceived,
+    /// The notification of an ephemeral command is ready, see [VM::ephemeral_state_get].
+    ///
+    /// This takes precedence over any other response.
+    /// The SDK should take the notification using [VM::take_ephemeral_notification], complete the related SDK promise,
+    /// and then call [VM::do_await] again.
+    EphemeralNotificationReady(EphemeralCompletionId),
 }
 
 #[derive(Debug, Copy, Clone, Eq, PartialEq, strum::EnumIs)]
@@ -471,6 +498,26 @@ pub trait VM: Sized {
 
     fn sys_state_get_keys(&mut self) -> VMResult<NotificationHandle>;
 
+    /// Ephemeral command to get the state value of the given key: neither the command nor its notification are recorded in the journal.
+    ///
+    /// The value is answered locally when known (eager state, or state modified by this invocation),
+    /// otherwise it's requested to the runtime. In both cases, the result is made available through [VM::do_await],
+    /// returning [AwaitResponse::EphemeralNotificationReady] with the returned id. Then the SDK should take the value with [VM::take_ephemeral_notification].
+    ///
+    /// Because the result is not recorded, the SDK should use it only in contexts where its usage is recorded otherwise, e.g. within a `ctx.run`.
+    ///
+    /// This operation is allowed only in the processing phase, and requires protocol V8.
+    fn ephemeral_state_get(&mut self, key: String) -> VMResult<EphemeralCompletionId>;
+
+    /// Take the notification of an ephemeral command, see [VM::ephemeral_state_get].
+    ///
+    /// Returns [Value::Void] when the state is empty, [Value::Success] otherwise.
+    /// Returns `None` if the notification is not ready yet, or it was already taken.
+    fn take_ephemeral_notification(
+        &mut self,
+        completion_id: EphemeralCompletionId,
+    ) -> VMResult<Option<Value>>;
+
     fn sys_state_set(&mut self, key: String, value: Bytes, options: PayloadOptions)
         -> VMResult<()>;
 

src/service_protocol/generated/dev.restate.service.protocol.rs

@@ -201,6 +201,39 @@ pub struct ProposeRunCompletionAckMessage {
     #[prost(uint32, tag = "1")]
     pub completion_id: u32,
 }
+/// Ephemeral command to get state.
+///
+/// Type: 0x0200 + 0
+#[allow(dead_code)]
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct GetStateEphemeralCommandMessage {
+    #[prost(uint32, tag = "1")]
+    pub ephemeral_completion_id: u32,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub key: ::prost::bytes::Bytes,
+}
+/// Notification of GetStateEphemeralCommandMessage.
+/// Type: 0x0300 + 0
+#[allow(dead_code)]
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct GetStateEphemeralNotificationMessage {
+    #[prost(uint32, tag = "1")]
+    pub ephemeral_completion_id: u32,
+    #[prost(oneof = "get_state_ephemeral_notification_message::Result", tags = "4, 5")]
+    pub result: ::core::option::Option<get_state_ephemeral_notification_message::Result>,
+}
+/// Nested message and enum types in `GetStateEphemeralNotificationMessage`.
+pub mod get_state_ephemeral_notification_message {
+    #[allow(dead_code)]
+    #[allow(clippy::enum_variant_names)]
+    #[derive(Clone, PartialEq, Eq, Hash, ::prost::Oneof)]
+    pub enum Result {
+        #[prost(message, tag = "4")]
+        Void(super::Void),
+        #[prost(message, tag = "5")]
+        Value(super::Value),
+    }
+}
 /// A notification message follows the following duck-type:
 ///
 #[allow(dead_code)]
@@ -1046,6 +1079,9 @@ pub enum ServiceProtocolVersion {
     /// * Semantic changes to Run proposal response, introduced ProposeRunCompletionAckMessage
     /// * ErrorMessage.behavior to customize retry behavior
     V7 = 7,
+    /// Added:
+    /// * Ephemeral commands: GetStateEphemeralCommandMessage and GetStateEphemeralNotificationMessage
+    V8 = 8,
 }
 impl ServiceProtocolVersion {
     /// String value of the enum field names used in the ProtoBuf definition.
@@ -1062,6 +1098,7 @@ impl ServiceProtocolVersion {
             Self::V5 => "V5",
             Self::V6 => "V6",
             Self::V7 => "V7",
+            Self::V8 => "V8",
         }
     }
     /// Creates an enum from field names used in the ProtoBuf definition.
@@ -1075,6 +1112,7 @@ impl ServiceProtocolVersion {
             "V5" => Some(Self::V5),
             "V6" => Some(Self::V6),
             "V7" => Some(Self::V7),
+            "V8" => Some(Self::V8),
             _ => None,
         }
     }

src/service_protocol/header.rs

@@ -77,6 +77,8 @@ gen_message_type_enum!(
     ProposeRunCompletion = 0x0005,
     AwaitingOn = 0x0006,
     ProposeRunCompletionAck = 0x0007,
+    GetStateEphemeralCommand = 0x0200,
+    GetStateEphemeralNotification = 0x0300,
     InputCommand = 0x0400,
     OutputCommand = 0x0401,
     GetLazyStateCommand = 0x0402,
@@ -253,6 +255,35 @@ mod tests {
         22
     );
 
+    roundtrip_test!(
+        get_state_ephemeral_command,
+        MessageHeader::new(GetStateEphemeralCommand, 10),
+        GetStateEphemeralCommand,
+        10
+    );
+
+    roundtrip_test!(
+        get_state_ephemeral_notification,
+        MessageHeader::new(GetStateEphemeralNotification, 22),
+        GetStateEphemeralNotification,
+        22
+    );
+
+    #[test]
+    fn ephemeral_message_types_ranges() {
+        // Ephemeral commands live in 0x0200..0x0300, their notifications in 0x0300..0x0400,
+        // paired by the low byte. They're neither journal commands nor notifications.
+        for (ty, id) in [
+            (GetStateEphemeralCommand, 0x0200u16),
+            (GetStateEphemeralNotification, 0x0300u16),
+        ] {
+            assert_eq!(u16::from(ty), id);
+            assert_eq!(MessageType::try_from(id).unwrap(), ty);
+            assert!(!ty.is_command());
+            assert!(!ty.is_notification());
+        }
+    }
+
     roundtrip_test!(
         custom_entry,
         MessageHeader::new(CustomEntry(0xFC01), 10341),

src/service_protocol/messages.rs

@@ -89,6 +89,8 @@ impl_message_traits!(End: core);
 impl_message_traits!(ProposeRunCompletion: message);
 impl_message_traits!(AwaitingOn: core);
 impl_message_traits!(ProposeRunCompletionAck: core);
+impl_message_traits!(GetStateEphemeralCommand: core);
+impl_message_traits!(GetStateEphemeralNotification: core);
 
 impl RestateEncodableMessage for ProposeRunCompletionMessage {
     fn encode(&self, service_protocol_version: Version) -> Bytes {

src/service_protocol/version.rs

@@ -10,6 +10,7 @@ pub enum Version {
     V5 = 5,
     V6 = 6,
     V7 = 7,
+    V8 = 8,
 }
 
 const CONTENT_TYPE_V1: &str = "application/vnd.restate.invocation.v1";
@@ -19,6 +20,7 @@ const CONTENT_TYPE_V4: &str = "application/vnd.restate.invocation.v4";
 const CONTENT_TYPE_V5: &str = "application/vnd.restate.invocation.v5";
 const CONTENT_TYPE_V6: &str = "application/vnd.restate.invocation.v6";
 const CONTENT_TYPE_V7: &str = "application/vnd.restate.invocation.v7";
+const CONTENT_TYPE_V8: &str = "application/vnd.restate.invocation.v8";
 
 impl Version {
     pub const fn content_type(&self) -> &'static str {
@@ -30,6 +32,7 @@ impl Version {
             Version::V5 => CONTENT_TYPE_V5,
             Version::V6 => CONTENT_TYPE_V6,
             Version::V7 => CONTENT_TYPE_V7,
+            Version::V8 => CONTENT_TYPE_V8,
         }
     }
 
@@ -38,7 +41,7 @@ impl Version {
     }
 
     pub const fn maximum_supported_version() -> Self {
-        Version::V7
+        Version::V8
     }
 }
 
@@ -69,6 +72,7 @@ impl FromStr for Version {
             CONTENT_TYPE_V5 => Ok(Version::V5),
             CONTENT_TYPE_V6 => Ok(Version::V6),
             CONTENT_TYPE_V7 => Ok(Version::V7),
+            CONTENT_TYPE_V8 => Ok(Version::V8),
             s if s.starts_with("application/vnd.restate.invocation.") => {
                 Err(ContentTypeError::RestateContentType(s.to_owned()))
             }