From b1be0431931dbd9302883f37f732a12f98029d51 Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Thu, 8 Oct 2026 12:44:49 +0200 Subject: [PATCH] Service protocol V8: * Ephemeral commands and notifications * Ephemeral get state --- docs/sdk-integration.md | 31 +- docs/service-invocation-protocol.md | 31 +- .../dev/restate/service/protocol.proto | 37 + src/fmt.rs | 4 +- src/lib.rs | 47 ++ .../generated/dev.restate.service.protocol.rs | 38 + src/service_protocol/header.rs | 31 + src/service_protocol/messages.rs | 2 + src/service_protocol/version.rs | 6 +- src/tests/ephemeral_state_get.rs | 726 ++++++++++++++++++ src/tests/mod.rs | 1 + src/vm/ephemeral_commands.rs | 65 ++ src/vm/errors.rs | 30 + src/vm/mod.rs | 61 +- src/vm/transitions/async_results.rs | 12 +- src/vm/transitions/ephemeral.rs | 114 +++ src/vm/transitions/input.rs | 5 +- src/vm/transitions/mod.rs | 2 + 18 files changed, 1232 insertions(+), 11 deletions(-) create mode 100644 src/tests/ephemeral_state_get.rs create mode 100644 src/vm/ephemeral_commands.rs create mode 100644 src/vm/transitions/ephemeral.rs diff --git a/docs/sdk-integration.md b/docs/sdk-integration.md index 0614d6d..885328f 100644 --- a/docs/sdk-integration.md +++ b/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:", 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 diff --git a/docs/service-invocation-protocol.md b/docs/service-invocation-protocol.md index c7a86c6..adacd1a 100644 --- a/docs/service-invocation-protocol.md +++ b/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 diff --git a/service-protocol/dev/restate/service/protocol.proto b/service-protocol/dev/restate/service/protocol.proto index e56a146..230ef26 100644 --- a/service-protocol/dev/restate/service/protocol.proto +++ b/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. diff --git a/src/fmt.rs b/src/fmt.rs index 50a274a..845bd71 100644 --- a/src/fmt.rs +++ b/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>, diff --git a/src/lib.rs b/src/lib.rs index b32cc61..351ce1c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -172,6 +172,27 @@ impl From 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 for EphemeralCompletionId { + fn from(value: u32) -> Self { + EphemeralCompletionId(value) + } +} + +impl From 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; + /// 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; + + /// 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>; + fn sys_state_set(&mut self, key: String, value: Bytes, options: PayloadOptions) -> VMResult<()>; diff --git a/src/service_protocol/generated/dev.restate.service.protocol.rs b/src/service_protocol/generated/dev.restate.service.protocol.rs index 9573e91..f2ea62f 100644 --- a/src/service_protocol/generated/dev.restate.service.protocol.rs +++ b/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, +} +/// 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, } } diff --git a/src/service_protocol/header.rs b/src/service_protocol/header.rs index f78df1e..913ebff 100644 --- a/src/service_protocol/header.rs +++ b/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), diff --git a/src/service_protocol/messages.rs b/src/service_protocol/messages.rs index 8da4f90..eaf2f02 100644 --- a/src/service_protocol/messages.rs +++ b/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 { diff --git a/src/service_protocol/version.rs b/src/service_protocol/version.rs index 40bb7cc..7a1622e 100644 --- a/src/service_protocol/version.rs +++ b/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())) } diff --git a/src/tests/ephemeral_state_get.rs b/src/tests/ephemeral_state_get.rs new file mode 100644 index 0000000..85b2046 --- /dev/null +++ b/src/tests/ephemeral_state_get.rs @@ -0,0 +1,726 @@ +use super::*; + +use crate::service_protocol::messages::start_message::StateEntry; +use crate::service_protocol::messages::ErrorBehavior; +use crate::service_protocol::messages::{ + get_lazy_state_completion_notification_message, get_state_ephemeral_notification_message, + propose_run_completion_message, run_completion_notification_message, + ClearAllStateCommandMessage, ClearStateCommandMessage, EndMessage, GetLazyStateCommandMessage, + GetLazyStateCompletionNotificationMessage, GetStateEphemeralCommandMessage, + GetStateEphemeralNotificationMessage, ProposeRunCompletionAckMessage, + ProposeRunCompletionMessage, RunCommandMessage, RunCompletionNotificationMessage, + SetStateCommandMessage, +}; +use crate::vm::errors::codes; +use crate::vm::errors::{ + EphemeralStateGetWithClosedInput, EPHEMERAL_COMMAND_DURING_REPLAY, + INPUT_CLOSED_WHILE_WAITING_EPHEMERAL_NOTIFICATIONS, +}; + +fn start_with_state(partial_state: bool, state: &[(&'static str, &'static str)]) -> StartMessage { + StartMessage { + known_entries: 1, + partial_state, + state_map: state + .iter() + .map(|(k, v)| StateEntry { + key: Bytes::from_static(k.as_bytes()), + value: Bytes::from_static(v.as_bytes()), + }) + .collect(), + ..start_message(1) + } +} + +fn get_state_notification( + completion_id: u32, + value: Option<&'static str>, +) -> GetStateEphemeralNotificationMessage { + GetStateEphemeralNotificationMessage { + ephemeral_completion_id: completion_id, + result: Some(match value { + None => get_state_ephemeral_notification_message::Result::Void(Default::default()), + Some(v) => get_state_ephemeral_notification_message::Result::Value( + Bytes::from_static(v.as_bytes()).into(), + ), + }), + } +} + +/// Start a run, and assert the VM wants to execute it. +fn start_run(vm: &mut CoreVM) -> NotificationHandle { + let RunHandle { replayed, handle } = vm.sys_run("project:STATE".to_owned()).unwrap(); + assert!(!replayed); + assert_eq!( + vm.do_await(UnresolvedFuture::Single(handle)).unwrap(), + AwaitResponse::ExecuteRun(handle) + ); + handle +} + +/// Await the notification of the given ephemeral command, assuming it's the only ready one, and take it. +fn await_and_take( + vm: &mut CoreVM, + awaiting: NotificationHandle, + completion_id: EphemeralCompletionId, +) -> Value { + assert_eq!( + vm.do_await(UnresolvedFuture::Single(awaiting)).unwrap(), + AwaitResponse::EphemeralNotificationReady(completion_id) + ); + let value = vm + .take_ephemeral_notification(completion_id) + .unwrap() + .unwrap(); + // Can't take it twice + assert_eq!(vm.take_ephemeral_notification(completion_id).unwrap(), None); + value +} + +fn assert_run_command(output: &mut OutputIterator) { + assert_eq!( + output.next_decoded::().unwrap(), + RunCommandMessage { + result_completion_id: 1, + name: "project:STATE".to_owned(), + } + ); +} + +mod local { + use super::*; + use test_log::test; + + #[test] + fn complete_eager_state() { + let mut output = VMTestCase::new() + .input(start_with_state(false, &[("STATE", "Francesco")])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, _| { + vm.sys_input().unwrap(); + let run = start_run(vm); + + let id = vm.ephemeral_state_get("STATE".to_owned()).unwrap(); + assert_eq!(u32::from(id), 1); + assert_eq!( + await_and_take(vm, run, id), + Value::Success(Bytes::from_static(b"Francesco")) + ); + + // Complete state, missing key is void + let id = vm.ephemeral_state_get("OTHER".to_owned()).unwrap(); + assert_eq!(u32::from(id), 2); + assert_eq!(await_and_take(vm, run, id), Value::Void); + }); + + assert_run_command(&mut output); + // No GetStateEphemeralCommandMessage + assert_eq!(output.next(), None); + } + + #[test] + fn partial_eager_state_with_key() { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[("STATE", "Francesco")])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, _| { + vm.sys_input().unwrap(); + let run = start_run(vm); + + let id = vm.ephemeral_state_get("STATE".to_owned()).unwrap(); + assert_eq!( + await_and_take(vm, run, id), + Value::Success(Bytes::from_static(b"Francesco")) + ); + }); + + assert_run_command(&mut output); + assert_eq!(output.next(), None); + } + + #[test] + fn served_with_closed_input() { + // E.g. request/response mode, all good as long as the key is known + let mut output = VMTestCase::new() + .input(start_with_state(true, &[("STATE", "Francesco")])) + .input(input_entry_message(b"my-data")) + .run(|vm| { + vm.sys_input().unwrap(); + let run = start_run(vm); + + let id = vm.ephemeral_state_get("STATE".to_owned()).unwrap(); + assert_eq!( + await_and_take(vm, run, id), + Value::Success(Bytes::from_static(b"Francesco")) + ); + }); + + assert_run_command(&mut output); + assert_eq!(output.next(), None); + } + + #[test] + fn after_set_and_clear() { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[("B", "b")])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, _| { + vm.sys_input().unwrap(); + vm.sys_state_set( + "A".to_owned(), + Bytes::from_static(b"a"), + PayloadOptions::default(), + ) + .unwrap(); + vm.sys_state_clear("B".to_owned()).unwrap(); + let run = start_run(vm); + + let id = vm.ephemeral_state_get("A".to_owned()).unwrap(); + assert_eq!( + await_and_take(vm, run, id), + Value::Success(Bytes::from_static(b"a")) + ); + let id = vm.ephemeral_state_get("B".to_owned()).unwrap(); + assert_eq!(await_and_take(vm, run, id), Value::Void); + }); + + output.next_decoded::().unwrap(); + output.next_decoded::().unwrap(); + assert_run_command(&mut output); + assert_eq!(output.next(), None); + } + + #[test] + fn after_clear_all() { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, _| { + vm.sys_input().unwrap(); + vm.sys_state_clear_all().unwrap(); + let run = start_run(vm); + + let id = vm.ephemeral_state_get("UNKNOWN".to_owned()).unwrap(); + assert_eq!(await_and_take(vm, run, id), Value::Void); + }); + + output + .next_decoded::() + .unwrap(); + assert_run_command(&mut output); + assert_eq!(output.next(), None); + } + + #[test] + fn after_replayed_set_state() { + let mut output = VMTestCase::new() + .input(StartMessage { + known_entries: 2, + ..start_with_state(true, &[]) + }) + .input(input_entry_message(b"my-data")) + .input(SetStateCommandMessage { + key: Bytes::from_static(b"STATE"), + value: Some(Bytes::from_static(b"Francesco").into()), + ..Default::default() + }) + .run_without_closing_input(|vm, _| { + vm.sys_input().unwrap(); + vm.sys_state_set( + "STATE".to_owned(), + Bytes::from_static(b"Francesco"), + PayloadOptions::default(), + ) + .unwrap(); + let run = start_run(vm); + + let id = vm.ephemeral_state_get("STATE".to_owned()).unwrap(); + assert_eq!( + await_and_take(vm, run, id), + Value::Success(Bytes::from_static(b"Francesco")) + ); + }); + + assert_run_command(&mut output); + assert_eq!(output.next(), None); + } +} + +mod remote { + use super::*; + use test_log::test; + + #[test] + fn get_state_then_propose_projection() { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, encoder| { + vm.sys_input().unwrap(); + let run = start_run(vm); + + let id = vm.ephemeral_state_get("STATE".to_owned()).unwrap(); + assert_eq!(u32::from(id), 1); + + // Not ready yet + assert_eq!(vm.take_ephemeral_notification(id).unwrap(), None); + assert_eq!( + vm.do_await(UnresolvedFuture::Single(run)).unwrap(), + AwaitResponse::WaitingExternalProgress { + waiting_input: true, + waiting_run_proposal: true + } + ); + + vm.notify_input(encoder.encode(&get_state_notification(1, Some("Francesco")))); + let value = await_and_take(vm, run, id); + assert2::assert!(let Value::Success(s) = value); + + // Projection + vm.propose_run_completion( + run, + RunExitResult::Success(Bytes::from(s.len().to_string())), + RetryPolicy::default(), + ) + .unwrap(); + vm.notify_input( + encoder.encode(&ProposeRunCompletionAckMessage { completion_id: 1 }), + ); + assert_eq!( + vm.do_await(UnresolvedFuture::Single(run)).unwrap(), + AwaitResponse::AnyCompleted + ); + assert2::assert!(let Some(Value::Success(projected)) = vm.take_notification(run).unwrap()); + + vm.sys_write_output(NonEmptyValue::Success(projected), PayloadOptions::default()) + .unwrap(); + vm.sys_end().unwrap(); + }); + + assert_run_command(&mut output); + assert_eq!( + output + .next_decoded::() + .unwrap(), + GetStateEphemeralCommandMessage { + ephemeral_completion_id: 1, + key: Bytes::from_static(b"STATE"), + } + ); + // The projection result is the only thing recorded + assert_eq!( + output + .next_decoded::() + .unwrap(), + ProposeRunCompletionMessage { + result_completion_id: 1, + result: Some(propose_run_completion_message::Result::Value( + Bytes::from_static(b"9") + )), + } + ); + assert_that!( + output.next_decoded::().unwrap(), + is_output_with_success(b"9") + ); + output.next_decoded::().unwrap(); + assert_eq!(output.next(), None); + } + + #[test] + fn void_notification() { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, encoder| { + vm.sys_input().unwrap(); + let run = start_run(vm); + + let id = vm.ephemeral_state_get("STATE".to_owned()).unwrap(); + vm.notify_input(encoder.encode(&get_state_notification(1, None))); + assert_eq!(await_and_take(vm, run, id), Value::Void); + }); + + assert_run_command(&mut output); + output + .next_decoded::() + .unwrap(); + assert_eq!(output.next(), None); + } + + #[test] + fn takes_precedence_and_ids_are_independent_of_completion_ids() { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, encoder| { + vm.sys_input().unwrap(); + + let h1 = vm + .sys_state_get("A".to_owned(), PayloadOptions::default()) + .unwrap(); + let u1 = vm.ephemeral_state_get("B".to_owned()).unwrap(); + let h2 = vm + .sys_state_get("C".to_owned(), PayloadOptions::default()) + .unwrap(); + let u2 = vm.ephemeral_state_get("D".to_owned()).unwrap(); + assert_eq!(u32::from(u1), 1); + assert_eq!(u32::from(u2), 2); + + vm.notify_input(encoder.encode(&GetLazyStateCompletionNotificationMessage { + completion_id: 1, + result: Some( + get_lazy_state_completion_notification_message::Result::Value( + Bytes::from_static(b"a").into(), + ), + ), + })); + vm.notify_input(encoder.encode(&get_state_notification(2, Some("d")))); + vm.notify_input(encoder.encode(&get_state_notification(1, Some("b")))); + + // Both ephemeral notifications come first, in any order + let mut ready = vec![]; + for _ in 0..2 { + let AwaitResponse::EphemeralNotificationReady(id) = + vm.do_await(UnresolvedFuture::Single(h1)).unwrap() + else { + panic!("Expected EphemeralNotificationReady") + }; + let value = vm.take_ephemeral_notification(id).unwrap().unwrap(); + ready.push((u32::from(id), value)); + } + ready.sort_by_key(|(id, _)| *id); + assert_eq!( + ready, + vec![ + (1, Value::Success(Bytes::from_static(b"b"))), + (2, Value::Success(Bytes::from_static(b"d"))) + ] + ); + + // Then the notification + assert_eq!( + vm.do_await(UnresolvedFuture::Single(h1)).unwrap(), + AwaitResponse::AnyCompleted + ); + assert_eq!( + vm.take_notification(h1).unwrap(), + Some(Value::Success(Bytes::from_static(b"a"))) + ); + assert!(!vm.is_completed(h2)); + }); + + assert_eq!( + output + .next_decoded::() + .unwrap() + .result_completion_id, + 1 + ); + assert_eq!( + output + .next_decoded::() + .unwrap(), + GetStateEphemeralCommandMessage { + ephemeral_completion_id: 1, + key: Bytes::from_static(b"B"), + } + ); + assert_eq!( + output + .next_decoded::() + .unwrap() + .result_completion_id, + 2 + ); + assert_eq!( + output + .next_decoded::() + .unwrap(), + GetStateEphemeralCommandMessage { + ephemeral_completion_id: 2, + key: Bytes::from_static(b"D"), + } + ); + assert_eq!(output.next(), None); + } + + #[test] + fn does_not_affect_replay_of_the_run() { + let mut output = VMTestCase::new() + .input(StartMessage { + known_entries: 3, + ..start_with_state(true, &[]) + }) + .input(input_entry_message(b"my-data")) + .input(RunCommandMessage { + result_completion_id: 1, + name: "project:STATE".to_owned(), + }) + .input(RunCompletionNotificationMessage { + completion_id: 1, + result: Some(run_completion_notification_message::Result::Value( + Bytes::from_static(b"9").into(), + )), + }) + .run(|vm| { + vm.sys_input().unwrap(); + let RunHandle { replayed, handle } = + vm.sys_run("project:STATE".to_owned()).unwrap(); + assert!(replayed); + assert_eq!( + vm.do_await(UnresolvedFuture::Single(handle)).unwrap(), + AwaitResponse::AnyCompleted + ); + assert2::assert!(let Some(Value::Success(projected)) = vm.take_notification(handle).unwrap()); + vm.sys_write_output(NonEmptyValue::Success(projected), PayloadOptions::default()) + .unwrap(); + vm.sys_end().unwrap(); + }); + + assert_that!( + output.next_decoded::().unwrap(), + is_output_with_success(b"9") + ); + output.next_decoded::().unwrap(); + assert_eq!(output.next(), None); + } +} + +mod failures { + use super::*; + use test_log::test; + + fn expect_protocol_violation(notification: GetStateEphemeralNotificationMessage) { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, encoder| { + vm.sys_input().unwrap(); + let run = start_run(vm); + vm.ephemeral_state_get("STATE".to_owned()).unwrap(); + + vm.notify_input(encoder.encode(¬ification)); + assert_that!( + vm.do_await(UnresolvedFuture::Single(run)), + err(pat!(Error { + code: eq(codes::PROTOCOL_VIOLATION.code()) + })) + ); + }); + + assert_run_command(&mut output); + output + .next_decoded::() + .unwrap(); + assert_that!( + output.next_decoded::().unwrap(), + pat!(ErrorMessage { + code: eq(codes::PROTOCOL_VIOLATION.code() as u32), + }) + ); + assert_eq!(output.next(), None); + } + + #[test] + fn notification_with_unknown_id_is_ignored() { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, encoder| { + vm.sys_input().unwrap(); + let run = start_run(vm); + let id = vm.ephemeral_state_get("STATE".to_owned()).unwrap(); + + // Unknown id, ignored + vm.notify_input(encoder.encode(&get_state_notification(99, Some("Unknown")))); + assert_eq!( + vm.do_await(UnresolvedFuture::Single(run)).unwrap(), + AwaitResponse::WaitingExternalProgress { + waiting_input: true, + waiting_run_proposal: true + } + ); + + // The in-flight command still completes + vm.notify_input(encoder.encode(&get_state_notification(1, Some("Francesco")))); + assert_eq!( + await_and_take(vm, run, id), + Value::Success(Bytes::from_static(b"Francesco")) + ); + }); + + assert_run_command(&mut output); + output + .next_decoded::() + .unwrap(); + // No error message + assert_eq!(output.next(), None); + } + + #[test] + fn notification_without_result() { + expect_protocol_violation(GetStateEphemeralNotificationMessage { + ephemeral_completion_id: 1, + result: None, + }); + } + + #[test] + fn notification_after_closed_is_ignored() { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, encoder| { + vm.sys_input().unwrap(); + vm.ephemeral_state_get("STATE".to_owned()).unwrap(); + vm.sys_write_output( + NonEmptyValue::Success(Bytes::from_static(b"done")), + PayloadOptions::default(), + ) + .unwrap(); + vm.sys_end().unwrap(); + + vm.notify_input(encoder.encode(&get_state_notification(1, Some("Francesco")))); + + // Still closed, not failed + assert_that!(vm.is_ready_to_execute(), err(is_closed())); + }); + + output + .next_decoded::() + .unwrap(); + output.next_decoded::().unwrap(); + output.next_decoded::().unwrap(); + assert_eq!(output.next(), None); + } + + #[test] + fn input_already_closed_and_unknown_key() { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[])) + .input(input_entry_message(b"my-data")) + .run(|vm| { + vm.sys_input().unwrap(); + start_run(vm); + assert_that!( + vm.ephemeral_state_get("STATE".to_owned()), + err(eq_error( + EphemeralStateGetWithClosedInput::new("STATE".to_owned()).into() + )) + ); + }); + + assert_run_command(&mut output); + assert_that!( + output.next_decoded::().unwrap(), + error_message_as_error( + EphemeralStateGetWithClosedInput::new("STATE".to_owned()).into() + ) + ); + assert_eq!(output.next(), None); + } + + #[test] + fn input_closed_while_in_flight_errors_instead_of_suspending() { + let mut output = VMTestCase::new() + .input(start_with_state(true, &[])) + .input(input_entry_message(b"my-data")) + .run_without_closing_input(|vm, _| { + vm.sys_input().unwrap(); + let run = start_run(vm); + vm.ephemeral_state_get("STATE".to_owned()).unwrap(); + + vm.notify_input_closed(); + assert_that!( + vm.do_await(UnresolvedFuture::Single(run)), + err(eq_error(INPUT_CLOSED_WHILE_WAITING_EPHEMERAL_NOTIFICATIONS)) + ); + }); + + assert_run_command(&mut output); + output + .next_decoded::() + .unwrap(); + assert_that!( + output.next_decoded::().unwrap(), + error_message_as_error(INPUT_CLOSED_WHILE_WAITING_EPHEMERAL_NOTIFICATIONS) + ); + assert_eq!(output.next(), None); + } + + #[test] + fn unsupported_on_v7() { + let mut output = VMTestCase::with_version(Version::V7) + .input(start_with_state(false, &[("STATE", "Francesco")])) + .input(input_entry_message(b"my-data")) + .run(|vm| { + vm.sys_input().unwrap(); + assert_that!( + vm.ephemeral_state_get("STATE".to_owned()), + err(pat!(Error { + code: eq(codes::UNSUPPORTED_FEATURE.code()) + })) + ); + }); + + assert_that!( + output.next_decoded::().unwrap(), + pat!(ErrorMessage { + code: eq(codes::UNSUPPORTED_FEATURE.code() as u32), + }) + ); + assert_eq!(output.next(), None); + } + + #[test] + fn during_replay_follows_journal_mismatch_behavior() { + let mut output = VMTestCase::with_vm_options(VMOptions { + journal_mismatch_retry_behavior: JournalMismatchRetryBehavior::Pause, + ..VMOptions::default() + }) + .input(StartMessage { + known_entries: 2, + ..start_with_state(false, &[("STATE", "Francesco")]) + }) + .input(input_entry_message(b"my-data")) + .input(ClearAllStateCommandMessage::default()) + .run(|vm| { + vm.sys_input().unwrap(); + assert!(vm.ephemeral_state_get("STATE".to_owned()).is_err()); + }); + + assert_that!( + output.next_decoded::().unwrap(), + pat!(ErrorMessage { + code: eq(codes::JOURNAL_MISMATCH.code() as u32), + behavior: eq(i32::from(ErrorBehavior::Pause)), + }) + ); + assert_eq!(output.next(), None); + } + + #[test] + fn during_replay() { + let mut output = VMTestCase::new() + .input(StartMessage { + known_entries: 2, + ..start_with_state(false, &[("STATE", "Francesco")]) + }) + .input(input_entry_message(b"my-data")) + .input(ClearAllStateCommandMessage::default()) + .run(|vm| { + vm.sys_input().unwrap(); + assert_that!( + vm.ephemeral_state_get("STATE".to_owned()), + err(eq_error(EPHEMERAL_COMMAND_DURING_REPLAY)) + ); + }); + + assert_that!( + output.next_decoded::().unwrap(), + error_message_as_error(EPHEMERAL_COMMAND_DURING_REPLAY) + ); + assert_eq!(output.next(), None); + } +} diff --git a/src/tests/mod.rs b/src/tests/mod.rs index 212b217..44ee7db 100644 --- a/src/tests/mod.rs +++ b/src/tests/mod.rs @@ -1,5 +1,6 @@ mod async_result; mod calls; +mod ephemeral_state_get; mod failures; mod implicit_cancellation; mod input_output; diff --git a/src/vm/ephemeral_commands.rs b/src/vm/ephemeral_commands.rs new file mode 100644 index 0000000..cfbe45d --- /dev/null +++ b/src/vm/ephemeral_commands.rs @@ -0,0 +1,65 @@ +use crate::{EphemeralCompletionId, Value}; +use std::collections::HashMap; +use tracing::warn; + +/// Tracks ephemeral commands, that is commands that are neither recorded in the journal, nor replayed, +/// such as the ephemeral state get. Their results are delivered as ephemeral notifications. +/// +/// Ephemeral completion ids are scoped to the current attempt and are detached from journaled completion ids. +#[derive(Debug, Default)] +pub(crate) struct EphemeralCommands { + last_completion_id: u32, + /// `None` when the command is in-flight (sent to the runtime), `Some` when its notification is ready to be taken. + commands: HashMap>, +} + +impl EphemeralCommands { + pub(crate) fn next_completion_id(&mut self) -> u32 { + self.last_completion_id += 1; + self.last_completion_id + } + + pub(crate) fn register_in_flight(&mut self, completion_id: u32) { + self.commands.insert(completion_id, None); + } + + pub(crate) fn register_ready(&mut self, completion_id: u32, value: Value) { + self.commands.insert(completion_id, Some(value)); + } + + pub(crate) fn complete(&mut self, completion_id: u32, value: Value) { + match self.commands.get_mut(&completion_id) { + Some(slot) => { + *slot = Some(value); + } + _ => { + warn!( + "Received an ephemeral notification for the ephemeral completion id {completion_id}, but there is no in-flight ephemeral command with this id. Ignoring it." + ); + } + } + } + + pub(crate) fn any_ready(&self) -> Option { + self.commands + .iter() + .find(|(_, v)| v.is_some()) + .map(|(id, _)| EphemeralCompletionId::from(*id)) + } + + pub(crate) fn take_notification( + &mut self, + completion_id: EphemeralCompletionId, + ) -> Option { + let completion_id = u32::from(completion_id); + if matches!(self.commands.get(&completion_id), Some(Some(_))) { + self.commands.remove(&completion_id).flatten() + } else { + None + } + } + + pub(crate) fn has_in_flight(&self) -> bool { + self.commands.values().any(Option::is_none) + } +} diff --git a/src/vm/errors.rs b/src/vm/errors.rs index a9ab31a..7792eee 100644 --- a/src/vm/errors.rs +++ b/src/vm/errors.rs @@ -380,6 +380,35 @@ impl BadProposeRunCompletionAck { } } +pub const EMPTY_GET_STATE_EPHEMERAL_NOTIFICATION: Error = Error::new_const( + codes::PROTOCOL_VIOLATION, + "Unexpected empty result variant for get state ephemeral notification.", +); + +#[derive(Debug, thiserror::Error)] +#[error("Cannot get state '{key}' without recording it in the journal: the input stream is closed, thus Restate cannot answer the request. Getting state without recording it in the journal requires a bidirectional stream between Restate and the service (not available in request/response mode, e.g. AWS Lambda), unless the state value is available in the eager state sent by Restate.")] +pub struct EphemeralStateGetWithClosedInput { + key: String, +} + +impl EphemeralStateGetWithClosedInput { + pub fn new(key: String) -> Self { + Self { key } + } +} + +pub const INPUT_CLOSED_WHILE_WAITING_EPHEMERAL_NOTIFICATIONS: Error = Error::new_const( + codes::INTERNAL, + "The input stream was closed while waiting for the result of a get state not recorded in the journal. Getting state without recording it in the journal requires a bidirectional stream between Restate and the service (not available in request/response mode, e.g. AWS Lambda), unless the state value is available in the eager state sent by Restate.", +); + +/// Ephemeral commands should really only be executed within ctx.run, so this is a "proxy safeguard" for that, because ctx.run +/// are really only started when we're processing (just track AwaitResponse::ExecuteRun usages) +pub const EPHEMERAL_COMMAND_DURING_REPLAY: Error = Error::new_const( + codes::JOURNAL_MISMATCH, + "Cannot execute ephemeral commands (e.g. get state without recording it in the journal) during replay, as their result might differ from the original execution. Ephemeral commands can be used only while processing new journal entries, e.g. within a ctx.run closure. This usually means the code uses them outside a ctx.run closure, or the code was changed.", +); + #[derive(Debug, Clone, thiserror::Error)] #[error("The provided duration for '{0}' is out of bounds: {1:?}")] pub struct OutOfBoundsDuration( @@ -433,4 +462,5 @@ impl WithInvocationErrorCode impl_error_code!(BadEagerStateKeyError, INTERNAL); impl_error_code!(UnsupportedFeatureForNegotiatedVersion, UNSUPPORTED_FEATURE); impl_error_code!(BadProposeRunCompletionAck, PROTOCOL_VIOLATION); +impl_error_code!(EphemeralStateGetWithClosedInput, INTERNAL); impl_error_code!(OutOfBoundsDuration, INTERNAL); diff --git a/src/vm/mod.rs b/src/vm/mod.rs index a1f16bd..8cf26e7 100644 --- a/src/vm/mod.rs +++ b/src/vm/mod.rs @@ -16,17 +16,18 @@ use crate::vm::errors::{ use crate::vm::run_state::RunState; use crate::vm::transitions::*; use crate::{ - AttachInvocationTarget, AwaitResponse, AwakeableHandle, CallHandle, CommandRelationship, Error, - Header, ImplicitCancellationOption, Input, NonDeterministicChecksOption, NonEmptyValue, - NotificationHandle, PayloadOptions, ResponseHead, RetryPolicy, RunExitResult, RunHandle, - SendHandle, Target, TerminalFailure, UnresolvedFuture, VMOptions, VMResult, Value, - CANCEL_NOTIFICATION_HANDLE, + AttachInvocationTarget, AwaitResponse, AwakeableHandle, CallHandle, CommandRelationship, + EphemeralCompletionId, Error, Header, ImplicitCancellationOption, Input, + NonDeterministicChecksOption, NonEmptyValue, NotificationHandle, PayloadOptions, ResponseHead, + RetryPolicy, RunExitResult, RunHandle, SendHandle, Target, TerminalFailure, UnresolvedFuture, + VMOptions, VMResult, Value, CANCEL_NOTIFICATION_HANDLE, }; use async_results_state::AsyncResultsState; use base64::engine::{DecodePaddingMode, GeneralPurpose, GeneralPurposeConfig}; use base64::{alphabet, Engine}; use bytes::{Buf, BufMut, Bytes, BytesMut}; use context::{Context, EagerState, Output}; +use ephemeral_commands::EphemeralCommands; use std::borrow::Cow; use std::collections::{HashMap, VecDeque}; use std::mem::size_of; @@ -37,6 +38,7 @@ use tracing::{debug, enabled, instrument, Level}; mod async_results_state; mod context; +mod ephemeral_commands; pub(crate) mod errors; mod run_state; mod transitions; @@ -63,6 +65,7 @@ pub(crate) enum State { run_state: RunState, async_results: AsyncResultsState, eager_state: EagerState, + ephemeral_commands: EphemeralCommands, }, Closed, } @@ -89,6 +92,7 @@ impl State { run_state, async_results, eager_state, + ephemeral_commands: Default::default(), }, s => s, } @@ -470,6 +474,18 @@ impl super::VM for CoreVM { ret )] fn do_await(&mut self, unresolved_future: UnresolvedFuture) -> VMResult { + // Ephemeral notifications take precedence over anything else + if let Ok(State::Processing { + ephemeral_commands, .. + }) = &self.last_transition + { + if let Some(completion_id) = ephemeral_commands.any_ready() { + return Ok(AwaitResponse::EphemeralNotificationReady(completion_id)); + } + } + + // TODO(slinkydeveloper) this is getting complex over time, + // need to figure out how to fit this implicit cancellation and co back into the typestate pattern. if self.is_implicit_cancellation_enabled() { // We want the runtime to wake us up in case cancel notification comes in. let unresolved_future_with_cancellation = UnresolvedFuture::FirstCompleted(vec![ @@ -622,6 +638,41 @@ impl super::VM for CoreVM { self.do_transition(SysStateGetKeys) } + #[instrument( + level = "trace", + skip(self), + fields( + restate.invocation.id = self.debug_invocation_id(), + restate.protocol.state = self.debug_state(), + restate.journal.command_index = self.context.journal.command_index(), + restate.protocol.version = %self.context.negotiated_protocol_version + ), + ret + )] + fn ephemeral_state_get(&mut self, key: String) -> VMResult { + self.verify_feature_support("ephemeral state get", Version::V8)?; + invocation_debug_logs!(self, "Executing 'Ephemeral state get {key}'"); + self.do_transition(EphemeralStateGet(key)) + } + + #[instrument( + level = "trace", + skip(self), + fields( + restate.invocation.id = self.debug_invocation_id(), + restate.protocol.state = self.debug_state(), + restate.journal.command_index = self.context.journal.command_index(), + restate.protocol.version = %self.context.negotiated_protocol_version + ), + ret + )] + fn take_ephemeral_notification( + &mut self, + completion_id: EphemeralCompletionId, + ) -> VMResult> { + self.do_transition(TakeEphemeralNotification(completion_id)) + } + #[instrument( level = "trace", skip(self, value), diff --git a/src/vm/transitions/async_results.rs b/src/vm/transitions/async_results.rs index eb7a0d6..75701df 100644 --- a/src/vm/transitions/async_results.rs +++ b/src/vm/transitions/async_results.rs @@ -3,7 +3,9 @@ use crate::service_protocol::messages::AwaitingOnMessage; use crate::service_protocol::{MessageType, NotificationId, CANCEL_SIGNAL_ID}; use crate::vm::async_results_state::ResolveFutureResult; use crate::vm::context::Context; -use crate::vm::errors::UncompletedDoProgressDuringReplay; +use crate::vm::errors::{ + UncompletedDoProgressDuringReplay, INPUT_CLOSED_WHILE_WAITING_EPHEMERAL_NOTIFICATIONS, +}; use crate::vm::transitions::{HitSuspensionPoint, Transition, TransitionAndReturn}; use crate::vm::{awakeable_id_str, State}; use crate::{ @@ -105,6 +107,7 @@ impl TransitionAndReturn for State { State::Processing { ref mut async_results, ref mut run_state, + ref ephemeral_commands, .. } => { let ResolveFutureResult::WaitExternalInput(unresolved_future) = @@ -127,6 +130,13 @@ impl TransitionAndReturn for State { // Check suspension condition if context.input_is_closed { + // Some ephemeral command is waiting for its notification, which will never arrive now. + // Suspending is not an option, as the response is not recorded: + // the user code waiting for it (e.g. a run closure) would just hang forever. + if ephemeral_commands.has_in_flight() { + return Err(INPUT_CLOSED_WHILE_WAITING_EPHEMERAL_NOTIFICATIONS); + } + // Some run still executing; it's not time to suspend yet! if waiting_run_proposal { return Ok(( diff --git a/src/vm/transitions/ephemeral.rs b/src/vm/transitions/ephemeral.rs new file mode 100644 index 0000000..330adba --- /dev/null +++ b/src/vm/transitions/ephemeral.rs @@ -0,0 +1,114 @@ +use crate::service_protocol::messages::{ + get_state_ephemeral_notification_message, GetStateEphemeralCommandMessage, + GetStateEphemeralNotificationMessage, +}; +use crate::service_protocol::RawMessage; +use crate::vm::context::{Context, EagerGetState}; +use crate::vm::errors::{ + EphemeralStateGetWithClosedInput, EMPTY_GET_STATE_EPHEMERAL_NOTIFICATION, + EPHEMERAL_COMMAND_DURING_REPLAY, +}; +use crate::vm::transitions::{Transition, TransitionAndReturn}; +use crate::vm::State; +use crate::{EphemeralCompletionId, Error, Value}; +use bytes::Bytes; + +pub(crate) struct EphemeralStateGet(pub(crate) String); + +impl TransitionAndReturn for State { + type Output = EphemeralCompletionId; + + fn transition_and_return( + mut self, + context: &mut Context, + EphemeralStateGet(key): EphemeralStateGet, + ) -> Result<(Self, Self::Output), Error> { + match &mut self { + State::Processing { + eager_state, + ephemeral_commands, + .. + } => { + match eager_state.get(&key) { + EagerGetState::Value(v) => { + let completion_id = ephemeral_commands.next_completion_id(); + ephemeral_commands.register_ready(completion_id, Value::Success(v)); + Ok((self, completion_id.into())) + } + EagerGetState::Empty => { + let completion_id = ephemeral_commands.next_completion_id(); + ephemeral_commands.register_ready(completion_id, Value::Void); + Ok((self, completion_id.into())) + } + EagerGetState::Unknown => { + if context.input_is_closed { + // The runtime can't answer anymore. + return Err(EphemeralStateGetWithClosedInput::new(key).into()); + } + let completion_id = ephemeral_commands.next_completion_id(); + ephemeral_commands.register_in_flight(completion_id); + context.output.send(&GetStateEphemeralCommandMessage { + ephemeral_completion_id: completion_id, + key: Bytes::from(key), + }); + Ok((self, completion_id.into())) + } + } + } + State::Replaying { .. } => Err(EPHEMERAL_COMMAND_DURING_REPLAY), + s => Err(s.as_unexpected_state("SysEphemeralStateGet")), + } + } +} + +pub(crate) struct NewGetStateEphemeralNotificationMessage(pub(crate) RawMessage); + +impl Transition for State { + fn transition( + mut self, + _: &mut Context, + NewGetStateEphemeralNotificationMessage(msg): NewGetStateEphemeralNotificationMessage, + ) -> Result { + match &mut self { + State::Processing { + ephemeral_commands, .. + } => { + let msg = msg.decode_to::(0)?; + let value = match msg.result.ok_or(EMPTY_GET_STATE_EPHEMERAL_NOTIFICATION)? { + get_state_ephemeral_notification_message::Result::Void(_) => Value::Void, + get_state_ephemeral_notification_message::Result::Value(v) => { + Value::Success(v.content) + } + }; + ephemeral_commands.complete(msg.ephemeral_completion_id, value); + } + State::Closed => { + // Can ignore + } + s => return Err(s.as_unexpected_state("NewGetStateEphemeralNotificationMessage")), + }; + Ok(self) + } +} + +pub(crate) struct TakeEphemeralNotification(pub(crate) EphemeralCompletionId); + +impl TransitionAndReturn for State { + type Output = Option; + + fn transition_and_return( + mut self, + _: &mut Context, + TakeEphemeralNotification(completion_id): TakeEphemeralNotification, + ) -> Result<(Self, Self::Output), Error> { + match &mut self { + State::Processing { + ephemeral_commands, .. + } => { + let value = ephemeral_commands.take_notification(completion_id); + Ok((self, value)) + } + s => Err(s.as_unexpected_state("TakeEphemeralNotification")), + } + } +} diff --git a/src/vm/transitions/input.rs b/src/vm/transitions/input.rs index 4f8aacd..97863d0 100644 --- a/src/vm/transitions/input.rs +++ b/src/vm/transitions/input.rs @@ -5,7 +5,7 @@ use crate::vm::errors::{ BadEagerStateKeyError, INPUT_CLOSED_WHILE_WAITING_ENTRIES, KNOWN_ENTRIES_IS_ZERO, UNEXPECTED_INPUT_MESSAGE, }; -use crate::vm::transitions::Transition; +use crate::vm::transitions::{NewGetStateEphemeralNotificationMessage, Transition}; use crate::vm::{errors, State}; use crate::{Error, Version}; use bytes::Bytes; @@ -24,6 +24,9 @@ impl Transition for State { MessageType::ProposeRunCompletionAck => { self.transition(context, NewProposeRunCompletionAckMessage(msg)) } + MessageType::GetStateEphemeralNotification => { + self.transition(context, NewGetStateEphemeralNotificationMessage(msg)) + } _ => Err(UNEXPECTED_INPUT_MESSAGE)?, } } diff --git a/src/vm/transitions/mod.rs b/src/vm/transitions/mod.rs index 08ac8b1..c6ac8cf 100644 --- a/src/vm/transitions/mod.rs +++ b/src/vm/transitions/mod.rs @@ -1,4 +1,5 @@ mod async_results; +mod ephemeral; mod input; mod journal; mod terminal; @@ -8,6 +9,7 @@ use crate::vm::context::Context; use crate::vm::{errors, State}; use crate::{CoreVM, Error, JournalMismatchRetryBehavior, Version}; pub(crate) use async_results::*; +pub(crate) use ephemeral::*; pub(crate) use input::*; pub(crate) use journal::*; use std::mem;