From 0d4ec7d7db2e4074ea9c686c4cf4a0cde20307a0 Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Fri, 28 Aug 2026 15:35:19 +0200 Subject: [PATCH 1/3] Auto-fail invocations using service protocol <= 3 --- crates/invoker-impl/src/error.rs | 1 + crates/invoker-impl/src/invocation_task/mod.rs | 6 ++---- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/crates/invoker-impl/src/error.rs b/crates/invoker-impl/src/error.rs index 8335791f54..09c4d15f29 100644 --- a/crates/invoker-impl/src/error.rs +++ b/crates/invoker-impl/src/error.rs @@ -312,6 +312,7 @@ impl InvokerError { RequestedErrorBehavior::retry(*retry_after) } InvokerError::MaxFutureDepthReached { .. } => RequestedErrorBehavior::Pause, + InvokerError::DeploymentDeprecated { .. } => RequestedErrorBehavior::Fail, _ => RequestedErrorBehavior::Retry, } } diff --git a/crates/invoker-impl/src/invocation_task/mod.rs b/crates/invoker-impl/src/invocation_task/mod.rs index 7d2a076273..6a72104abc 100644 --- a/crates/invoker-impl/src/invocation_task/mod.rs +++ b/crates/invoker-impl/src/invocation_task/mod.rs @@ -48,7 +48,7 @@ use restate_types::service_protocol::ServiceProtocolVersion; use restate_util_bytecount::{ByteCount, NonZeroByteCount}; use restate_util_string::ReString; use restate_worker_api::invoker::invocation_reader::{ - EagerState, InvocationReader, InvocationReaderTransaction, JournalKind, + EagerState, InvocationReader, InvocationReaderTransaction, }; use restate_worker_api::invoker::{EntryEnricher, InvocationReaderError}; @@ -547,9 +547,7 @@ where self.abort_timeout = abort_timeout; } - if chosen_service_protocol_version < ServiceProtocolVersion::V4 - && journal_metadata.journal_kind == JournalKind::V2 - { + if chosen_service_protocol_version < ServiceProtocolVersion::V4 { // We don't support migrating from journal v2 to journal v1! shortcircuit!(Err(InvokerError::DeploymentDeprecated( self.invocation_target.service_name().to_string(), From 162111aa139a7a98cfeb0c5446bb429ee886d785 Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Fri, 28 Aug 2026 15:52:11 +0200 Subject: [PATCH 2/3] Remove service protocol runner <= v3, and cascade remove all the other code we don't need anymore. --- crates/invoker-impl/src/error.rs | 46 +- .../invoker-impl/src/invocation_task/mod.rs | 82 +- .../service_protocol_runner.rs | 789 ------------------ .../service_protocol_runner_v4.rs | 21 +- crates/invoker-impl/src/lib.rs | 243 +----- crates/invoker-impl/src/quota.rs | 10 - crates/invoker-impl/src/test_util.rs | 99 +-- crates/service-protocol/src/lib.rs | 4 - .../service-protocol/src/message/encoding.rs | 495 ----------- crates/service-protocol/src/message/header.rs | 545 ------------ crates/service-protocol/src/message/mod.rs | 124 --- .../worker-api/src/invoker/entry_enricher.rs | 23 - crates/worker-api/src/invoker/mod.rs | 2 - crates/worker/src/invoker_integration.rs | 406 --------- crates/worker/src/lib.rs | 1 - crates/worker/src/partition/leadership/mod.rs | 32 +- 16 files changed, 59 insertions(+), 2863 deletions(-) delete mode 100644 crates/invoker-impl/src/invocation_task/service_protocol_runner.rs delete mode 100644 crates/service-protocol/src/message/encoding.rs delete mode 100644 crates/service-protocol/src/message/header.rs delete mode 100644 crates/service-protocol/src/message/mod.rs delete mode 100644 crates/worker-api/src/invoker/entry_enricher.rs delete mode 100644 crates/worker/src/invoker_integration.rs diff --git a/crates/invoker-impl/src/error.rs b/crates/invoker-impl/src/error.rs index 09c4d15f29..71bb700c86 100644 --- a/crates/invoker-impl/src/error.rs +++ b/crates/invoker-impl/src/error.rs @@ -8,7 +8,6 @@ // the Business Source License, use of this software will be governed // by the Apache License, Version 2.0. -use std::collections::HashSet; use std::error::Error as StdError; use std::fmt; use std::ops::RangeInclusive; @@ -18,7 +17,6 @@ use http::{HeaderName, HeaderValue}; use restate_memory::OutOfMemoryKind; use restate_service_client::ServiceClientError; -use restate_service_protocol::message::{EncodingError, MessageType}; use restate_types::errors::{IdDecodeError, InvocationError, InvocationErrorCode, codes}; use restate_types::identifiers::DeploymentId; use restate_types::journal::raw::RawEntryCodecError; @@ -60,21 +58,12 @@ pub(crate) enum InvokerError { UnexpectedContentType(Option, HeaderValue), #[error("received unexpected message: {0:?}")] #[code(restate_errors::RT0012)] - UnexpectedMessage(MessageType), - #[error("received unexpected message: {0:?}")] - #[code(restate_errors::RT0012)] UnexpectedMessageV4(restate_service_protocol_v4::message_codec::MessageType), #[error("message encoding error: {0}")] - Encoding( - #[from] - #[code] - EncodingError, - ), - #[error("message encoding error: {0}")] #[code(restate_errors::RT0012)] - EncodingV2(#[from] journal_v2::encoding::DecodingError), + Encoding(#[from] journal_v2::encoding::DecodingError), #[error("message encoding error: {0}")] - EncoderV2( + Encoder( #[from] #[code] restate_service_protocol_v4::message_codec::EncodingError, @@ -99,11 +88,6 @@ pub(crate) enum InvokerError { #[error("got empty AwaitingOnMessage")] #[code(restate_errors::RT0012)] EmptyAwaitingOnMessage, - #[error( - "got bad SuspensionMessage, suspending on journal indexes {0:?}, but journal length is {1}" - )] - #[code(restate_errors::RT0012)] - BadSuspensionMessage(HashSet, EntryIndex), #[error("malformed ProposeRunCompletionMessage, missing result field")] #[code(restate_errors::RT0012)] MalformedProposeRunCompletion, @@ -136,9 +120,6 @@ pub(crate) enum InvokerError { #[code(restate_errors::RT0001)] AbortTimeoutFired(FriendlyDuration), - #[error("cannot process entry {1} (index {0}) because of a failed precondition: {2}")] - #[code(restate_errors::RT0017)] - EntryEnrichment(EntryIndex, EntryType, #[source] InvocationError), #[error("cannot process command {1} (command index {0}) because of a failed precondition: {2}")] CommandPrecondition( CommandIndex, @@ -321,19 +302,6 @@ impl InvokerError { match self { InvokerError::Sdk(sdk_error) => *sdk_error.error, InvokerError::SdkV2(sdk_error) => *sdk_error.error, - InvokerError::EntryEnrichment(entry_index, entry_type, e) => { - let msg = format!( - "Error when processing entry {} of type {}: {}", - entry_index, - entry_type, - e.message() - ); - let mut err = InvocationError::new(e.code(), msg); - if let Some(desc) = e.into_stacktrace() { - err = err.with_stacktrace(desc); - } - err - } e @ InvokerError::BadNegotiatedServiceProtocolVersion(_) => { InvocationError::new(codes::UNSUPPORTED_MEDIA_TYPE, e.to_string()) } @@ -419,16 +387,6 @@ pub(crate) struct SdkInvocationError { pub(crate) error: Box, } -impl SdkInvocationError { - pub(crate) fn unknown() -> Self { - Self { - related_entry: None, - next_retry_interval_override: None, - error: Default::default(), - } - } -} - impl fmt::Display for SdkInvocationError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { if self.error.code() == codes::JOURNAL_MISMATCH { diff --git a/crates/invoker-impl/src/invocation_task/mod.rs b/crates/invoker-impl/src/invocation_task/mod.rs index 6a72104abc..c923203b47 100644 --- a/crates/invoker-impl/src/invocation_task/mod.rs +++ b/crates/invoker-impl/src/invocation_task/mod.rs @@ -9,7 +9,6 @@ // by the Apache License, Version 2.0. mod retry_after; -mod service_protocol_runner; mod service_protocol_runner_v4; use std::collections::HashSet; @@ -37,8 +36,6 @@ use restate_types::LimitKey; use restate_types::deployment::PinnedDeployment; use restate_types::identifiers::InvocationId; use restate_types::invocation::{FencingToken, InvocationTarget}; -use restate_types::journal::EntryIndex; -use restate_types::journal::enriched::EnrichedRawEntry; use restate_types::journal_v2::raw::RawNotification; use restate_types::journal_v2::{self, CommandIndex, NotificationId, UnresolvedFuture}; use restate_types::live::Live; @@ -47,15 +44,14 @@ use restate_types::schema::invocation_target::InvocationTargetResolver; use restate_types::service_protocol::ServiceProtocolVersion; use restate_util_bytecount::{ByteCount, NonZeroByteCount}; use restate_util_string::ReString; +use restate_worker_api::invoker::InvocationReaderError; use restate_worker_api::invoker::invocation_reader::{ EagerState, InvocationReader, InvocationReaderTransaction, }; -use restate_worker_api::invoker::{EntryEnricher, InvocationReaderError}; use super::Notification; use crate::TokenBucket; use crate::error::{InvocationMemoryExhausted, InvokerError}; -use crate::invocation_task::service_protocol_runner::ServiceProtocolRunner; use crate::metric_definitions::{INVOKER_EAGER_STATE_TRUNCATED, INVOKER_TASK_DURATION}; // Clippy false positive, might be caused by Bytes contained within HeaderValue. @@ -163,16 +159,6 @@ pub(super) enum InvocationTaskOutputInner { // `has_changed` indicates if we believe this is a freshly selected endpoint or not. PinnedDeployment(PinnedDeployment, /* has_changed: */ bool), ServerHeaderReceived(String), - NewEntry { - entry_index: EntryIndex, - entry: Box, - /// If true, the SDK requested to be notified when the entry is correctly stored. - /// - /// When reading the entry from the storage this flag will always be false, as we never need to send acks for entries sent during a journal replay. - /// - /// See https://github.com/restatedev/service-protocol/blob/main/service-invocation-protocol.md#acknowledgment-of-stored-entries - requires_ack: bool, - }, NewCommand { command_index: CommandIndex, command: journal_v2::raw::RawCommand, @@ -196,7 +182,6 @@ pub(super) enum InvocationTaskOutputInner { unresolved_future: UnresolvedFuture, }, Closed, - Suspended(HashSet), SuspendedV2(HashSet), SuspendedV3(UnresolvedFuture), Failed(InvokerError, LocalMemoryPool), @@ -254,7 +239,7 @@ fn new_invoker_body( } /// Represents an open invocation stream -pub(super) struct InvocationTask { +pub(super) struct InvocationTask { // Shared client client: ServiceClient, @@ -273,7 +258,6 @@ pub(super) struct InvocationTask { max_awaited_future_depth: usize, // Invoker tx/rx - entry_enricher: EE, schemas: Live, invoker_tx: mpsc::UnboundedSender, invoker_rx: mpsc::UnboundedReceiver, @@ -288,7 +272,6 @@ pub(super) struct InvocationTask { enum TerminalLoopState { Continue(T), Closed, - Suspended(HashSet), SuspendedV2(HashSet), SuspendedV3(UnresolvedFuture), Failed(InvokerError), @@ -302,7 +285,7 @@ impl TerminalLoopState { } fn is_suspend(&self) -> bool { - matches!(self, Self::Suspended(_) | Self::SuspendedV2(_)) + matches!(self, Self::SuspendedV2(_) | Self::SuspendedV3(_)) } } @@ -328,7 +311,6 @@ macro_rules! shortcircuit { match TerminalLoopState::from($value) { TerminalLoopState::Continue(v) => v, TerminalLoopState::Closed => return TerminalLoopState::Closed, - TerminalLoopState::Suspended(v) => return TerminalLoopState::Suspended(v), TerminalLoopState::SuspendedV2(v) => return TerminalLoopState::SuspendedV2(v), TerminalLoopState::SuspendedV3(v) => return TerminalLoopState::SuspendedV3(v), TerminalLoopState::ShouldYield(oom) => return TerminalLoopState::ShouldYield(oom), @@ -337,9 +319,8 @@ macro_rules! shortcircuit { }; } -impl InvocationTask +impl InvocationTask where - EE: EntryEnricher, Schemas: DeploymentResolver + InvocationTargetResolver, { #[allow(clippy::too_many_arguments)] @@ -354,7 +335,6 @@ where message_size_warning: NonZeroUsize, message_size_limit: NonZeroUsize, retry_count_since_last_stored_entry: u32, - entry_enricher: EE, deployment_metadata_resolver: Live, invoker_tx: mpsc::UnboundedSender, invoker_rx: mpsc::UnboundedReceiver, @@ -372,7 +352,6 @@ where inactivity_timeout: default_inactivity_timeout, abort_timeout: default_abort_timeout, eager_state_size_limit, - entry_enricher, schemas: deployment_metadata_resolver, invoker_tx, invoker_rx, @@ -420,7 +399,6 @@ where unreachable!("This is not supposed to happen. This is a runtime bug") } TerminalLoopState::Closed => InvocationTaskOutputInner::Closed, - TerminalLoopState::Suspended(v) => InvocationTaskOutputInner::Suspended(v), TerminalLoopState::SuspendedV2(v) => InvocationTaskOutputInner::SuspendedV2(v), TerminalLoopState::SuspendedV3(v) => InvocationTaskOutputInner::SuspendedV3(v), TerminalLoopState::Failed(e) => { @@ -577,43 +555,27 @@ where deployment_changed, )); - if chosen_service_protocol_version <= ServiceProtocolVersion::V3 { - // Protocol runner for service protocol <= v3 - let service_protocol_runner = - ServiceProtocolRunner::new(self, chosen_service_protocol_version); - service_protocol_runner - .run( - txn, - journal_metadata, - keyed_service_id, - deployment, - reader_for_bidi, - invocation_budget, - ) - .await - } else { - // Protocol runner for service protocol v4+ - let service_protocol_runner = service_protocol_runner_v4::ServiceProtocolRunner::new( - self, - chosen_service_protocol_version, - &deployment.ty, - self.max_awaited_future_depth, - ); - service_protocol_runner - .run( - txn, - journal_metadata, - keyed_service_id, - deployment, - reader_for_bidi, - invocation_budget, - ) - .await - } + // Protocol runner for service protocol v4+ + let service_protocol_runner = service_protocol_runner_v4::ServiceProtocolRunner::new( + self, + chosen_service_protocol_version, + &deployment.ty, + self.max_awaited_future_depth, + ); + service_protocol_runner + .run( + txn, + journal_metadata, + keyed_service_id, + deployment, + reader_for_bidi, + invocation_budget, + ) + .await } } -impl InvocationTask { +impl InvocationTask { /// Send a non-terminal message to the invoker main loop. pub(crate) fn send_invoker_tx(&self, invocation_task_output_inner: InvocationTaskOutputInner) { let _ = self.invoker_tx.send(InvocationTaskOutput { diff --git a/crates/invoker-impl/src/invocation_task/service_protocol_runner.rs b/crates/invoker-impl/src/invocation_task/service_protocol_runner.rs deleted file mode 100644 index 6c5d34430a..0000000000 --- a/crates/invoker-impl/src/invocation_task/service_protocol_runner.rs +++ /dev/null @@ -1,789 +0,0 @@ -// Copyright (c) 2023 - 2026 Restate Software, Inc., Restate GmbH. -// All rights reserved. -// -// Use of this software is governed by the Business Source License -// included in the LICENSE file. -// -// As of the Change Date specified in that file, in accordance with -// the Business Source License, use of this software will be governed -// by the Apache License, Version 2.0. - -use std::collections::HashSet; -use std::time::Duration; - -use bytes::Bytes; -use bytestring::ByteString; -use futures::{Stream, StreamExt}; -use http::uri::PathAndQuery; -use http::{HeaderMap, HeaderName, HeaderValue, StatusCode}; -use opentelemetry::trace::TraceFlags; -use prost::Message; -use tokio::sync::mpsc; -use tracing::{debug, trace, warn}; - -use restate_errors::warn_it; -use restate_memory::{LocalMemoryLease, LocalMemoryPool, PinnableMemoryStream}; -use restate_service_client::{Method, Parts, Request}; -use restate_service_protocol::codec::ProtobufRawEntryCodec; -use restate_service_protocol::message::{ - Decoder, Encoder, MessageHeader, MessageType, ProtocolMessage, StateEntry, -}; -use restate_service_protocol_v4::entry_codec::ServiceProtocolV4Codec; -use restate_types::errors::InvocationError; -use restate_types::identifiers::ServiceId; -use restate_types::identifiers::{EntryIndex, InvocationId}; -use restate_types::invocation::ServiceInvocationSpanContext; -use restate_types::journal::raw::RawEntryCodec; -use restate_types::journal::{Completion, CompletionResult, EntryType}; -use restate_types::journal_v2; -use restate_types::journal_v2::EntryMetadata; -use restate_types::schema::deployment::{Deployment, ProtocolType}; -use restate_types::service_protocol::ServiceProtocolVersion; -use restate_worker_api::invoker::invocation_reader::{ - EagerState, InvocationReader, InvocationReaderError, InvocationReaderTransaction, JournalEntry, - JournalKind, -}; -use restate_worker_api::invoker::{EntryEnricher, JournalMetadata}; - -use crate::Notification; -use crate::error::{InvocationErrorRelatedEntry, InvokerError, SdkInvocationError}; -use crate::invocation_task::{ - InvocationTask, InvocationTaskOutputInner, InvokerBodySender, InvokerBodyType, ResponseChunk, - ResponseStream, TerminalLoopState, X_RESTATE_SERVER, collect_eager_state, - invocation_id_to_header_value, leased_frame, new_invoker_body, - service_protocol_version_to_header_value, -}; - -/// Provides the value of the invocation id -const INVOCATION_ID_HEADER_NAME: HeaderName = HeaderName::from_static("x-restate-invocation-id"); - -const GATEWAY_ERRORS_CODES: [http::StatusCode; 3] = [ - http::StatusCode::BAD_GATEWAY, - http::StatusCode::SERVICE_UNAVAILABLE, - http::StatusCode::GATEWAY_TIMEOUT, -]; - -/// Runs the interaction between the server and the service endpoint. -pub struct ServiceProtocolRunner<'a, EE, DMR> { - invocation_task: &'a mut InvocationTask, - - service_protocol_version: ServiceProtocolVersion, - - // Encoder/Decoder - encoder: Encoder, - decoder: Decoder, - - // task state - next_journal_index: EntryIndex, -} - -impl<'a, EE, DMR> ServiceProtocolRunner<'a, EE, DMR> -where - EE: EntryEnricher, -{ - pub fn new( - invocation_task: &'a mut InvocationTask, - service_protocol_version: ServiceProtocolVersion, - ) -> Self { - let encoder = Encoder::new(service_protocol_version); - let decoder = Decoder::new( - service_protocol_version, - invocation_task.message_size_warning, - invocation_task.message_size_limit, - ); - - Self { - invocation_task, - service_protocol_version, - encoder, - decoder, - next_journal_index: 0, - } - } - - /// How often to release excess outbound budget capacity during the bidi-stream phase. - const BUDGET_RELEASE_INTERVAL: Duration = Duration::from_secs(5); - - /// Run the service protocol interaction. - /// - /// # Arguments - /// * `keyed_service_id` - If `Some`, eager state loading is enabled and we'll read/send - /// state for this service upfront. If `None`, lazy state is used (either because this - /// isn't a keyed service, or lazy state is enabled, or eager state is disabled). - pub async fn run( - mut self, - txn: Txn, - journal_metadata: JournalMetadata, - keyed_service_id: Option, - deployment: Deployment, - invocation_reader: IR, - outbound_budget: &mut LocalMemoryPool, - ) -> TerminalLoopState<()> - where - Txn: InvocationReaderTransaction, - IR: InvocationReader, - { - // Figure out the protocol type. Force RequestResponse if inactivity_timeout is zero - let protocol_type = if self.invocation_task.inactivity_timeout.is_zero() { - ProtocolType::RequestResponse - } else { - deployment.ty.protocol_type() - }; - - // Close the invoker_rx in case it's request response, this avoids further buffering of messages in this channel. - if protocol_type == ProtocolType::RequestResponse { - self.invocation_task.invoker_rx.close(); - } - - let path: PathAndQuery = format!( - "/invoke/{}/{}", - self.invocation_task.invocation_target.service_name(), - self.invocation_task.invocation_target.handler_name() - ) - .try_into() - .expect("must be able to build a valid invocation path"); - - let journal_size = journal_metadata.length; - - debug!( - restate.invocation.id = %self.invocation_task.invocation_id, - deployment.address = %deployment.address_display(), - deployment.service_protocol_version = %self.service_protocol_version.as_repr(), - path = %path, - "Executing invocation at deployment" - ); - - // Create an arc of the parent SpanContext. - // We send this with every journal entry to correctly link new spans generated from journal entries. - let service_invocation_span_context = journal_metadata.span_context; - - // Prepare the request - let (mut http_stream_tx, request) = Self::prepare_request( - path, - deployment, - self.service_protocol_version, - &self.invocation_task.invocation_id, - &service_invocation_span_context, - self.invocation_task.invocation_target.key(), - ); - - // Initialize the response stream state - let mut http_stream_rx = std::pin::pin!(ResponseStream::new( - self.invocation_task.client.call(request) - )); - - // === Replay phase (transaction alive) === - { - // Read state if needed (state is collected for the START message). - // LocalMemoryPool-gated: each state entry acquires a lease from the outbound - // budget. The per-entry leases are merged into a single lease that - // accompanies the start message frame. - let state = if let Some(ref service_id) = keyed_service_id { - Some(crate::shortcircuit!( - txn.read_state_budgeted(service_id, outbound_budget) - .map_err(InvokerError::from_state_reader) - )) - } else { - None - }; - - // Send start message with state (leases are merged inside write_start) - crate::shortcircuit!( - self.write_start( - &mut http_stream_tx, - journal_size, - state, - self.invocation_task.retry_count_since_last_stored_entry, - journal_metadata.last_modification_date.elapsed() - ) - .await - ); - - // Read journal stream from storage and execute the replay. - // LocalMemoryPool-gated: each entry acquires a lease before it's sent. - let journal_stream = crate::shortcircuit!( - txn.read_journal_budgeted( - &self.invocation_task.invocation_id, - journal_size, - journal_metadata.journal_kind, - outbound_budget, - ) - .map_err(InvokerError::from_journal_reader) - ); - crate::shortcircuit!( - self.replay_loop(&mut http_stream_tx, &mut http_stream_rx, journal_stream) - .await - ); - } - // === End replay phase - streams dropped, transaction can be dropped === - - // Transaction dropped - RocksDB snapshot released! - drop(txn); - - // Check all the entries have been replayed - debug_assert_eq!(self.next_journal_index, journal_size); - - // Release excess local capacity accumulated during replay back to the - // global pool before entering the bidi stream phase. - outbound_budget.release_excess(); - - // If we have the invoker_rx and the protocol type is bidi stream, - // then we can use the bidi_stream loop reading the invoker_rx and the http_stream_rx - if protocol_type == ProtocolType::BidiStream { - trace!("Protocol is in bidi stream mode, will now start the send/receive loop"); - crate::shortcircuit!( - self.bidi_stream_loop( - &service_invocation_span_context, - http_stream_tx, - &mut http_stream_rx, - invocation_reader, - outbound_budget, - ) - .await - ); - } else { - trace!("Protocol is in bidi stream mode, will now drop the sender side of the request"); - // Drop the http_stream_tx. - // This is required in HTTP/1.1 to let the deployment send the headers back - drop(http_stream_tx) - } - - // We don't have the invoker_rx, so we simply consume the response - trace!("Sender side of the request has been dropped, now processing the response"); - let result = self - .response_stream_loop(&service_invocation_span_context, &mut http_stream_rx) - .await; - - // Sanity check of the stream decoder - if self.decoder.has_remaining() { - warn_it!( - InvokerError::WriteAfterEndOfStream, - "The read buffer is non empty after the stream has been closed." - ); - } - - result - } - - fn prepare_request( - path: PathAndQuery, - deployment: Deployment, - service_protocol_version: ServiceProtocolVersion, - invocation_id: &InvocationId, - parent_span_context: &ServiceInvocationSpanContext, - service_key: Option<&ByteString>, - ) -> (InvokerBodySender, Request) { - // Use an unbounded channel: backpressure is provided by the memory budget - // (each frame's Bytes embeds a LocalMemoryLease via from_owner) rather than - // channel capacity. - let (http_stream_tx, http_stream_rx) = mpsc::unbounded_channel(); - let request_body = new_invoker_body(http_stream_rx); - - let service_protocol_header_value = - service_protocol_version_to_header_value(service_protocol_version); - - let invocation_id_header_value = invocation_id_to_header_value(invocation_id); - - let mut headers = HeaderMap::from_iter([ - ( - http::header::CONTENT_TYPE, - service_protocol_header_value.clone(), - ), - (http::header::ACCEPT, service_protocol_header_value), - (INVOCATION_ID_HEADER_NAME, invocation_id_header_value), - ]); - - // Inject OpenTelemetry context into the headers - // The parent span as seen by the SDK will be the service invocation span context - // which is emitted at INFO level representing the invocation, *not* the DEBUG level - // `invoker_invocation_task` which wraps this code. This is so that headers will be sent - // when in INFO level, not just in DEBUG level. - { - let span_context = parent_span_context.span_context(); - if span_context.is_valid() { - const SUPPORTED_VERSION: u8 = 0; - let header_value = format!( - "{:02x}-{}-{}-{:02x}", - SUPPORTED_VERSION, - span_context.trace_id(), - span_context.span_id(), - span_context.trace_flags() & TraceFlags::SAMPLED - ); - if let Ok(header_value) = HeaderValue::try_from(header_value) { - headers.insert("traceparent", header_value); - } - if let Ok(tracestate) = - HeaderValue::from_str(span_context.trace_state().header().as_ref()) - { - headers.insert("tracestate", tracestate); - } - } - } - - let mut request_parts = Parts::from_deployment(deployment, Method::Post, path, headers); - if let Some(service_key) = service_key { - request_parts = request_parts.with_request_identity_sub_field(service_key.clone()); - } - - (http_stream_tx, Request::new(request_parts, request_body)) - } - - // --- Loops - - /// This loop concurrently pushes journal entries and waits for the response headers and end of replay. - async fn replay_loop( - &mut self, - http_stream_tx: &mut InvokerBodySender, - http_stream_rx: &mut S, - journal_stream: JournalStream, - ) -> TerminalLoopState<()> - where - JournalStream: Stream> + Unpin, - S: Stream> + Unpin, - E: InvocationReaderError, - { - let mut journal_stream = journal_stream.fuse(); - let mut got_headers = false; - loop { - tokio::select! { - got_headers_res = http_stream_rx.next(), if !got_headers => { - got_headers = true; - // The reason we want to poll headers in this function is - // to exit early in case an error is returned during replays. - match crate::shortcircuit!(got_headers_res.transpose()) { - None => { - return TerminalLoopState::Failed(InvokerError::Sdk(SdkInvocationError::unknown())); - } - Some(ResponseChunk::Parts(headers)) => { - crate::shortcircuit!(self.handle_response_headers(headers)); - } - Some(ResponseChunk::Data(_)) => { - panic!("Unexpected poll after the headers have been resolved already") - } - }; - - }, - opt_je = journal_stream.next() => { - match opt_je { - Some(Ok((JournalEntry::JournalV1(je), lease))) => { - crate::shortcircuit!(self.write_with_lease(http_stream_tx, ProtocolMessage::UnparsedEntry(je), Some(lease))); - self.next_journal_index += 1; - }, - Some(Ok((JournalEntry::JournalV2(re), lease))) => { - if re.ty() == journal_v2::EntryType::Command(journal_v2::CommandType::Input) { - let input_entry = crate::shortcircuit!(re.decode::()); - crate::shortcircuit!(self.write_with_lease(http_stream_tx, ProtocolMessage::UnparsedEntry( - ProtobufRawEntryCodec::serialize_as_input_entry( - input_entry.headers, - input_entry.payload - ).erase_enrichment() - ), Some(lease))); - self.next_journal_index += 1; - } else { - panic!("This is unexpected, when an entry is stored with journal v2, only input entry is allowed!") - } - } - Some(Ok((JournalEntry::JournalV1Completion(_), _))) => { - // During replay, a JournalV1Completion means the completion - // arrived before the entry itself. This entry cannot be replayed - // to the SDK because we don't have the original entry bytes. - // This should not happen in normal operation since entries are - // always stored before completions during replay. - panic!("Unexpected JournalV1Completion during replay: completion arrived before entry was stored") - } - Some(Err(e)) => { - return TerminalLoopState::from( - Err::<(), _>(InvokerError::from_journal_reader(e)), - ); - } - None => { - // No need to wait for the headers to continue - trace!("Finished to replay the journal"); - return TerminalLoopState::Continue(()) - } - } - } - } - } - } - - /// This loop concurrently reads the http response stream and journal completions from the invoker. - async fn bidi_stream_loop( - &mut self, - parent_span_context: &ServiceInvocationSpanContext, - mut http_stream_tx: InvokerBodySender, - http_stream_rx: &mut S, - mut invocation_reader: IR, - outbound_budget: &mut LocalMemoryPool, - ) -> TerminalLoopState<()> - where - S: Stream> + Unpin, - IR: InvocationReader, - { - let mut release_interval = tokio::time::interval(Self::BUDGET_RELEASE_INTERVAL); - release_interval.tick().await; // consume initial immediate tick - loop { - tokio::select! { - opt_completion = self.invocation_task.invoker_rx.recv() => { - match opt_completion { - Some(Notification::Completion(entry_index)) => { - trace!(restate.journal.index = entry_index, "Reading completion from storage"); - let (completion, lease) = crate::shortcircuit!( - read_completion_from_storage_budgeted( - &mut invocation_reader, - &self.invocation_task.invocation_id, - entry_index, - outbound_budget, - ).await - ); - trace!("Sending the completion to the wire"); - crate::shortcircuit!(self.write_with_lease(&mut http_stream_tx, completion.into(), Some(lease))); - }, - Some(Notification::CommandAck(entry_index)) => { - trace!("Sending the ack to the wire"); - crate::shortcircuit!(self.write(&mut http_stream_tx, ProtocolMessage::new_entry_ack(entry_index))); - }, - Some(Notification::Entry { .. }) | Some(Notification::ProposeRunCompletionAck(_)) => { - panic!("We don't expect to receive journal_v2 entries, this is an invoker bug.") - }, - None => { - // Completion channel is closed, - // the invoker main loop won't send completions anymore. - // Response stream might still be open though. - return TerminalLoopState::Continue(()) - }, - } - }, - chunk = http_stream_rx.next() => { - match crate::shortcircuit!(chunk.transpose()) { - None => { - return TerminalLoopState::Failed(InvokerError::Sdk(SdkInvocationError::unknown())); - } - Some(ResponseChunk::Parts(parts)) => crate::shortcircuit!(self.handle_response_headers(parts)), - Some(ResponseChunk::Data(buf)) => crate::shortcircuit!(self.handle_read(parent_span_context, buf)), - } - }, - _ = release_interval.tick() => { - outbound_budget.release_excess(); - }, - _ = tokio::time::sleep(self.invocation_task.inactivity_timeout) => { - debug!("Inactivity detected, going to suspend invocation"); - // Just return. This will drop the invoker_rx and http_stream_tx, - // closing the request stream and the invoker input channel. - return TerminalLoopState::Continue(()) - }, - } - } - } - - async fn response_stream_loop( - &mut self, - parent_span_context: &ServiceInvocationSpanContext, - http_stream_rx: &mut S, - ) -> TerminalLoopState<()> - where - S: Stream> + Unpin, - { - loop { - tokio::select! { - chunk = http_stream_rx.next() => { - match crate::shortcircuit!(chunk.transpose()) { - None => { - return TerminalLoopState::Failed(InvokerError::Sdk(SdkInvocationError::unknown())); - } - Some(ResponseChunk::Parts(parts)) => crate::shortcircuit!(self.handle_response_headers(parts)), - Some(ResponseChunk::Data(buf)) => crate::shortcircuit!(self.handle_read(parent_span_context, buf)), - } - }, - _ = tokio::time::sleep(self.invocation_task.abort_timeout) => { - warn!("Inactivity detected, going to close invocation"); - return TerminalLoopState::Failed(InvokerError::AbortTimeoutFired(self.invocation_task.abort_timeout.into())) - }, - } - } - } - - // --- Read and write methods - - async fn write_start( - &mut self, - http_stream_tx: &mut InvokerBodySender, - journal_size: u32, - state: Option>, - retry_count_since_last_stored_entry: u32, - duration_since_last_stored_entry: Duration, - ) -> Result<(), InvokerError> - where - S: PinnableMemoryStream> + Send, - E: InvocationReaderError, - { - // Collect state entries with size limit - let (partial_state, state_map, state_lease) = collect_eager_state( - state, - self.invocation_task.eager_state_size_limit, - |(key, value)| StateEntry { key, value }, - ) - .await?; - - // Send the invoke frame with the merged state lease - self.write_with_lease( - http_stream_tx, - ProtocolMessage::new_start_message( - Bytes::copy_from_slice(&self.invocation_task.invocation_id.to_bytes()), - self.invocation_task.invocation_id.to_string(), - self.invocation_task - .invocation_target - .key() - .map(|bs| bs.as_bytes().clone()), - journal_size, - partial_state, - state_map, - retry_count_since_last_stored_entry, - duration_since_last_stored_entry, - ), - state_lease, - ) - } - - fn write( - &mut self, - http_stream_tx: &mut InvokerBodySender, - msg: ProtocolMessage, - ) -> Result<(), InvokerError> { - self.write_with_lease(http_stream_tx, msg, None) - } - - fn write_with_lease( - &mut self, - http_stream_tx: &mut InvokerBodySender, - msg: ProtocolMessage, - lease: Option, - ) -> Result<(), InvokerError> { - trace!(restate.protocol.message = ?msg, "Sending message"); - let buf = self.encoder.encode(msg); - - if http_stream_tx.send(Ok(leased_frame(buf, lease))).is_err() { - return Err(InvokerError::UnexpectedClosedRequestStream); - }; - Ok(()) - } - - fn handle_response_headers( - &mut self, - mut parts: http::response::Parts, - ) -> Result<(), InvokerError> { - // if service is running behind a gateway, the service can be down - // but we still get a response code from the gateway itself. In that - // case we still need to return the proper error - if GATEWAY_ERRORS_CODES.contains(&parts.status) { - return Err(InvokerError::ServiceUnavailable(parts.status)); - } - - // otherwise we return generic UnexpectedResponse - if !parts.status.is_success() { - // Decorate the error in case of UNSUPPORTED_MEDIA_TYPE, as it probably is the incompatible protocol version - if parts.status == StatusCode::UNSUPPORTED_MEDIA_TYPE { - return Err(InvokerError::BadNegotiatedServiceProtocolVersion( - self.service_protocol_version, - )); - } - - return Err(InvokerError::UnexpectedResponse(parts.status)); - } - - let content_type = parts.headers.remove(http::header::CONTENT_TYPE); - let expected_content_type = - service_protocol_version_to_header_value(self.service_protocol_version); - match content_type { - Some(ct) => - { - #[allow(clippy::borrow_interior_mutable_const)] - if ct != expected_content_type { - return Err(InvokerError::UnexpectedContentType( - Some(ct), - expected_content_type, - )); - } - } - None => { - return Err(InvokerError::UnexpectedContentType( - None, - expected_content_type, - )); - } - } - - if let Some(hv) = parts.headers.remove(X_RESTATE_SERVER) { - self.invocation_task - .send_invoker_tx(InvocationTaskOutputInner::ServerHeaderReceived( - hv.to_str() - .map_err(|e| InvokerError::BadHeader(X_RESTATE_SERVER, e))? - .to_owned(), - )) - } - - Ok(()) - } - - fn handle_read( - &mut self, - parent_span_context: &ServiceInvocationSpanContext, - buf: Bytes, - ) -> TerminalLoopState<()> { - self.decoder.push(buf); - - while let Some((frame_header, frame)) = crate::shortcircuit!(self.decoder.consume_next()) { - crate::shortcircuit!(self.handle_message(parent_span_context, frame_header, frame)); - } - - TerminalLoopState::Continue(()) - } - - fn handle_message( - &mut self, - parent_span_context: &ServiceInvocationSpanContext, - mh: MessageHeader, - message: ProtocolMessage, - ) -> TerminalLoopState<()> { - trace!(restate.protocol.message_header = ?mh, restate.protocol.message = ?message, "Received message"); - match message { - ProtocolMessage::Start { .. } => { - TerminalLoopState::Failed(InvokerError::UnexpectedMessage(MessageType::Start)) - } - ProtocolMessage::Completion(_) => { - TerminalLoopState::Failed(InvokerError::UnexpectedMessage(MessageType::Completion)) - } - ProtocolMessage::EntryAck(_) => { - TerminalLoopState::Failed(InvokerError::UnexpectedMessage(MessageType::EntryAck)) - } - ProtocolMessage::Suspension(suspension) => { - let suspension_indexes = HashSet::from_iter(suspension.entry_indexes); - // We currently don't support empty suspension_indexes set - if suspension_indexes.is_empty() { - return TerminalLoopState::Failed(InvokerError::EmptySuspensionMessage); - } - // Sanity check on the suspension indexes - if *suspension_indexes.iter().max().unwrap() >= self.next_journal_index { - return TerminalLoopState::Failed(InvokerError::BadSuspensionMessage( - suspension_indexes, - self.next_journal_index, - )); - } - TerminalLoopState::Suspended(suspension_indexes) - } - ProtocolMessage::Error(e) => { - TerminalLoopState::Failed(InvokerError::Sdk(SdkInvocationError { - related_entry: Some(InvocationErrorRelatedEntry { - related_entry_index: e.related_entry_index, - related_entry_name: e.related_entry_name.clone(), - related_entry_type: e - .related_entry_type - .and_then(|t| u16::try_from(t).ok()) - .and_then(|idx| MessageType::try_from(idx).ok()) - .and_then(|mt| EntryType::try_from(mt).ok()), - entry_was_committed: e - .related_entry_index - .is_some_and(|entry_idx| entry_idx < self.next_journal_index), - }), - next_retry_interval_override: e.next_retry_delay.map(Duration::from_millis), - error: InvocationError::from(e).into(), - })) - } - ProtocolMessage::End(_) => TerminalLoopState::Closed, - ProtocolMessage::UnparsedEntry(entry) => { - let entry_type = entry.header().as_entry_type(); - let enriched_entry = crate::shortcircuit!( - self.invocation_task - .entry_enricher - .enrich_entry( - entry, - &self.invocation_task.invocation_target, - parent_span_context - ) - .map_err(|e| InvokerError::EntryEnrichment( - self.next_journal_index, - entry_type, - e - )) - ); - self.invocation_task - .send_invoker_tx(InvocationTaskOutputInner::NewEntry { - entry_index: self.next_journal_index, - entry: enriched_entry.into(), - requires_ack: mh - .requires_ack() - .expect("All entry messages support requires_ack"), - }); - self.next_journal_index += 1; - TerminalLoopState::Continue(()) - } - } - } -} - -/// Reads a v1 completion from storage with budget tracking. -/// -/// Reads the entry and acquires a [`LocalMemoryLease`] for its serialized size from -/// the outbound budget. Only used by the v1-v3 protocol runner. -async fn read_completion_from_storage_budgeted( - invocation_reader: &mut IR, - invocation_id: &InvocationId, - entry_index: EntryIndex, - budget: &mut LocalMemoryPool, -) -> Result<(Completion, LocalMemoryLease), InvokerError> { - let (entry, lease) = invocation_reader - .read_journal_entry_budgeted(invocation_id, entry_index, JournalKind::V1, budget) - .await - .map_err(InvokerError::from_journal_reader)? - .ok_or_else(|| { - InvokerError::JournalReader(anyhow::anyhow!( - "journal entry {entry_index} not found for completion read" - )) - })?; - let completion = extract_completion(entry_index, entry)?; - Ok((completion, lease)) -} - -/// Extracts a [`Completion`] from a journal entry read from storage. -fn extract_completion( - entry_index: EntryIndex, - journal_entry: JournalEntry, -) -> Result { - use restate_types::service_protocol; - - match journal_entry { - JournalEntry::JournalV1(plain_raw_entry) => { - let extractor = service_protocol::CompletionResultExtractor::decode( - plain_raw_entry.serialized_entry().as_ref(), - ) - .map_err(|e| { - InvokerError::JournalReader(anyhow::anyhow!( - "failed to decode completion from entry {entry_index}: {e}" - )) - })?; - - let result = match extractor.result { - Some(service_protocol::completion_result_extractor::Result::Empty(_)) => { - CompletionResult::Empty - } - Some(service_protocol::completion_result_extractor::Result::Value(b)) => { - CompletionResult::Success(b) - } - Some(service_protocol::completion_result_extractor::Result::Failure(f)) => { - CompletionResult::Failure(f.code.into(), f.message.into()) - } - None => { - return Err(InvokerError::JournalReader(anyhow::anyhow!( - "journal entry {entry_index} has no completion result" - ))); - } - }; - - Ok(Completion::new(entry_index, result)) - } - JournalEntry::JournalV1Completion(result) => Ok(Completion::new(entry_index, result)), - JournalEntry::JournalV2(_) => { - panic!("v1-v3 protocol runner should not encounter JournalV2 entries") - } - } -} diff --git a/crates/invoker-impl/src/invocation_task/service_protocol_runner_v4.rs b/crates/invoker-impl/src/invocation_task/service_protocol_runner_v4.rs index 2873a77712..f016a3750e 100644 --- a/crates/invoker-impl/src/invocation_task/service_protocol_runner_v4.rs +++ b/crates/invoker-impl/src/invocation_task/service_protocol_runner_v4.rs @@ -93,8 +93,8 @@ const RATE_LIMITED_CODES: [StatusCode; 2] = [ ]; /// Runs the interaction between the server and the service endpoint. -pub struct ServiceProtocolRunner<'a, EE, Schemas> { - invocation_task: &'a mut InvocationTask, +pub struct ServiceProtocolRunner<'a, Schemas> { + invocation_task: &'a mut InvocationTask, service_protocol_version: ServiceProtocolVersion, @@ -109,12 +109,12 @@ pub struct ServiceProtocolRunner<'a, EE, Schemas> { max_awaited_future_depth: usize, } -impl<'a, EE, Schemas> ServiceProtocolRunner<'a, EE, Schemas> +impl<'a, Schemas> ServiceProtocolRunner<'a, Schemas> where Schemas: InvocationTargetResolver, { pub fn new( - invocation_task: &'a mut InvocationTask, + invocation_task: &'a mut InvocationTask, service_protocol_version: ServiceProtocolVersion, deployment_type: &DeploymentType, max_awaited_future_depth: usize, @@ -276,9 +276,8 @@ where TerminalLoopState::Closed => { attempt_span.set_status(Status::Ok); } - TerminalLoopState::Suspended(_) - | TerminalLoopState::SuspendedV2(_) - | TerminalLoopState::SuspendedV3(_) => { + + TerminalLoopState::SuspendedV2(_) | TerminalLoopState::SuspendedV3(_) => { attempt_span.add_event( restate_tracing_instrumentation::semconv::event::RESTATE_INVOCATION_LIFECYCLE_SUSPENDED, vec![], @@ -1049,7 +1048,7 @@ where // original bytes downstream to avoid a re-encode round trip. let parsed = crate::shortcircuit!( proto_lite::GetInvocationOutputCommandMessageLite::decode(cmd.as_ref()) - .map_err(|err| InvokerError::EncodingV2(GenericError::from(err).into())) + .map_err(|err| InvokerError::Encoding(GenericError::from(err).into())) ); if let Some(target) = parsed.target.as_ref() { shortcircuit!(Self::validate_target(target).map_err(|err| { @@ -1072,7 +1071,7 @@ where // See `Message::GetInvocationOutputCommand` above for why we decode-then-forward. let parsed = shortcircuit!( proto_lite::AttachInvocationCommandMessageLite::decode(cmd.as_ref()) - .map_err(|err| InvokerError::EncodingV2(GenericError::from(err).into())) + .map_err(|err| InvokerError::Encoding(GenericError::from(err).into())) ); if let Some(target) = parsed.target.as_ref() { shortcircuit!(Self::validate_target(target).map_err(|err| { @@ -1435,7 +1434,7 @@ where let unresolved_future: UnresolvedFuture = shortcircuit!( awaiting_on .try_into() - .map_err(|e| InvokerError::EncodingV2(GenericError::from(e).into())) + .map_err(|e| InvokerError::Encoding(GenericError::from(e).into())) ); self.invocation_task .send_invoker_tx(InvocationTaskOutputInner::AwaitingOn { unresolved_future }); @@ -1462,7 +1461,7 @@ where let future: UnresolvedFuture = shortcircuit!( awaiting_on .try_into() - .map_err(|e| InvokerError::EncodingV2(GenericError::from(e).into())) + .map_err(|e| InvokerError::Encoding(GenericError::from(e).into())) ); // We currently don't support empty future set diff --git a/crates/invoker-impl/src/lib.rs b/crates/invoker-impl/src/lib.rs index c21dcca586..ae96a93967 100644 --- a/crates/invoker-impl/src/lib.rs +++ b/crates/invoker-impl/src/lib.rs @@ -49,7 +49,6 @@ use restate_types::identifiers::PartitionId; use restate_types::identifiers::{DeploymentId, InvocationId, WithPartitionKey}; use restate_types::invocation::{FencingToken, InvocationTarget}; use restate_types::journal::EntryIndex; -use restate_types::journal::enriched::EnrichedRawEntry; use restate_types::journal_events::raw::RawEvent; use restate_types::journal_events::{Event, PausedEvent, TransientErrorEvent}; use restate_types::journal_v2::raw::{RawCommand, RawNotification}; @@ -65,7 +64,7 @@ use restate_util_time::DurationExt; use restate_worker_api::invoker::capacity::TokenBucket; use restate_worker_api::invoker::invocation_reader::InvocationReader; use restate_worker_api::invoker::{ - Effect, EffectKind, EntryEnricher, FencedEffect, InvocationStatusReport, YieldReason, + Effect, EffectKind, FencedEffect, InvocationStatusReport, YieldReason, }; use restate_worker_api::resources::ReservedResources; @@ -134,18 +133,16 @@ trait InvocationTaskRunner { ) -> AbortHandle; } -struct DefaultInvocationTaskRunner { +struct DefaultInvocationTaskRunner { client: ServiceClient, - entry_enricher: EE, schemas: Live, action_token_bucket: Option, allow_protocol_v7: bool, } -impl InvocationTaskRunner for DefaultInvocationTaskRunner +impl InvocationTaskRunner for DefaultInvocationTaskRunner where IR: InvocationReader + Clone + Send + Sync + 'static, - EE: EntryEnricher + Clone + Send + Sync + 'static, Schemas: DeploymentResolver + InvocationTargetResolver + Clone + Send + Sync + 'static, { fn start_invocation_task( @@ -178,7 +175,6 @@ where opts.message_size_warning.as_non_zero_usize(), opts.message_size_limit(), retry_count_since_last_stored_entry, - self.entry_enricher.clone(), self.schemas.clone(), invoker_tx, invoker_rx, @@ -227,7 +223,7 @@ impl From for u16 { } // -- Service implementation -pub struct Service { +pub struct Service { // Used for constructing the invoker sender and status reader input_tx: mpsc::UnboundedSender, status_tx: mpsc::UnboundedSender< @@ -237,12 +233,11 @@ pub struct Service { tmp_dir: PathBuf, // We have this level of indirection to hide the InvocationTaskRunner, // which is a rather internal thing we have only for mocking. - inner: - ServiceInner, Schemas, StorageReader>, + inner: ServiceInner, Schemas, StorageReader>, invocation_token_bucket: Option, } -impl Service { +impl Service { #[allow(clippy::too_many_arguments)] pub(crate) fn new( invoker_id: impl Into, @@ -252,14 +247,12 @@ impl Service, client: ServiceClient, - entry_enricher: TEntryEnricher, invocation_token_bucket: Option, action_token_bucket: Option, memory_pool: MemoryPool, - ) -> Service + ) -> Service where StorageReader: InvocationReader + Clone + Send + Sync + 'static, - TEntryEnricher: EntryEnricher, Schemas: DeploymentResolver + InvocationTargetResolver + Clone, { let invoker_id = invoker_id.into(); @@ -280,7 +273,6 @@ impl Service Service, service_client_options: &ServiceClientOptions, invoker_options: &InvokerOptions, - entry_enricher: TEntryEnricher, schemas: Live, invocation_token_bucket: Option, action_token_bucket: Option, memory_pool: MemoryPool, - ) -> Result, BuildError> + ) -> Result, BuildError> where StorageReader: InvocationReader + Clone + Send + Sync + 'static, - TEntryEnricher: EntryEnricher, Schemas: DeploymentResolver + InvocationTargetResolver + Clone, { metric_definitions::describe_metrics(); @@ -341,7 +331,6 @@ impl Service Service +impl Service where IR: InvocationReader + Clone + Send + Sync + 'static, - EE: EntryEnricher + Clone + Send + Sync + 'static, Schemas: DeploymentResolver + InvocationTargetResolver + Clone + Send + Sync + 'static, { pub fn handle(&self) -> InvokerHandle { @@ -578,14 +566,6 @@ where x_restate_server_header ) } - InvocationTaskOutputInner::NewEntry {entry_index, entry, requires_ack} => { - self.handle_new_entry( - invocation_id, - entry_index, - *entry, - requires_ack - ).await - }, InvocationTaskOutputInner::NewNotificationProposal { notification, requested_ack } => { self.handle_new_notification_proposal( invocation_id, @@ -605,9 +585,6 @@ where InvocationTaskOutputInner::Failed(e, returned_budget) => { self.handle_invocation_task_failed(invocation_id, e, returned_budget).await }, - InvocationTaskOutputInner::Suspended(indexes) => { - self.handle_invocation_task_suspended(invocation_id, indexes).await - } InvocationTaskOutputInner::NewCommand { command, command_index, requested_ack } => { self.handle_new_command( invocation_id, @@ -855,59 +832,6 @@ where }); } - #[instrument( - level = "trace", - skip_all, - fields( - restate.invocation.id = %invocation_id, - restate.journal.index = entry_index, - restate.journal.entry_type = ?entry.ty(), - ) - )] - async fn handle_new_entry( - &mut self, - invocation_id: InvocationId, - entry_index: EntryIndex, - entry: EnrichedRawEntry, - requires_ack: bool, - ) { - if let Some((output_tx, ism)) = self - .invocation_state_machine_manager - .resolve_invocation(&invocation_id) - { - ism.notify_new_command(entry_index, requires_ack); - trace!( - restate.invocation.target = %ism.invocation_target, - "Received a new entry. Invocation state: {:?}", - ism.invocation_state_debug() - ); - self.status_store.on_progress_made(&invocation_id); - if let Some(pinned_deployment) = ism.pinned_deployment_to_notify() { - let _ = output_tx - .send(fence( - ism.fencing_token, - Effect { - invocation_id, - kind: EffectKind::PinnedDeployment(pinned_deployment), - }, - )) - .await; - } - let _ = output_tx - .send(fence( - ism.fencing_token, - Effect { - invocation_id, - kind: EffectKind::JournalEntry { entry_index, entry }, - }, - )) - .await; - } else { - // If no state machine, this might be an entry for an aborted invocation. - trace!("No state machine found for given entry"); - } - } - #[instrument( level = "trace", skip_all, @@ -1125,74 +1049,6 @@ where } } - #[instrument( - level = "trace", - skip_all, - fields( - restate.invocation.id = %invocation_id, - ) - )] - async fn handle_invocation_task_suspended( - &mut self, - invocation_id: InvocationId, - entry_indexes: HashSet, - ) { - if let Some((sender, _, ism)) = self - .invocation_state_machine_manager - .remove_invocation(&invocation_id) - { - counter!( - INVOKER_INVOCATION_TASKS, - "status" => TASK_OP_SUSPENDED, - "partition_id" => self.invoker_id_label.clone() - ) - .increment(1); - self.status_store.on_end(&invocation_id); - - if ism.requested_pause { - // We should send pause instead - trace!( - restate.invocation.target = %ism.invocation_target, - "Pausing invocation after suspension" - ); - - let _ = sender - .send(fence( - ism.fencing_token, - Effect { - invocation_id, - kind: EffectKind::Paused { - paused_event: RawEvent::from(Event::Paused(PausedEvent { - last_failure: None, - })), - }, - }, - )) - .await; - } else { - trace!( - restate.invocation.target = %ism.invocation_target, - "Suspending invocation" - ); - - let _ = sender - .send(fence( - ism.fencing_token, - Effect { - invocation_id, - kind: EffectKind::Suspended { - waiting_for_completed_entries: entry_indexes, - }, - }, - )) - .await; - } - } else { - // If no state machine, this might be a result for an aborted invocation. - trace!("No state machine found for invocation task suspended signal"); - } - } - #[instrument( level = "trace", skip_all, @@ -1986,8 +1842,6 @@ mod tests { use restate_types::errors::{InvocationError, codes}; use restate_types::identifiers::ServiceRevision; use restate_types::invocation::ServiceType; - use restate_types::journal::enriched::EnrichedEntryHeader; - use restate_types::journal::raw::RawEntry; use restate_types::journal_events::EventType; use restate_types::journal_v2::{Command, Encoder, Entry, OutputCommand, OutputResult}; use restate_types::live::Constant; @@ -2285,7 +2139,6 @@ mod tests { AssumeRoleCacheMode::None, ) .unwrap(), - test_util::MockEntryEnricher, None, None, MemoryPool::unlimited(), @@ -2417,82 +2270,6 @@ mod tests { assert!(!service_inner.quota.is_slot_available()); } - #[test(restate_core::test)] - async fn reclaim_quota_after_abort() { - let invoker_options = InvokerOptionsBuilder::default() - .inactivity_timeout(FriendlyDuration::ZERO) - .abort_timeout(FriendlyDuration::ZERO) - .disable_eager_state(false) - .message_size_warning(NonZeroUsize::new(1024).unwrap().into()) - .message_size_limit(None) - .build() - .unwrap(); - let invocation_id = InvocationId::mock_random(); - - let (_, _status_tx, _effects_rx, mut service_inner) = ServiceInner::mock( - |invocation_id, - _service_id, - _storage_reader, - invoker_tx: mpsc::UnboundedSender, - _| { - let _ = invoker_tx.send(InvocationTaskOutput { - invocation_id, - fencing_token: 0, - inner: InvocationTaskOutputInner::NewEntry { - entry_index: 1, - entry: RawEntry::new(EnrichedEntryHeader::SetState {}, Bytes::default()) - .into(), - requires_ack: false, - }, - }); - pending() // Never ends - }, - MockSchemas( - // fixed amount of retries so that an invocation eventually completes with a failure - Some(RetryPolicy::fixed_delay(Duration::ZERO, Some(1))), - Some(OnMaxAttempts::Kill), - ), - Some(NonZeroUsize::new(2).unwrap()), - EmptyStorageReader, - ); - - // Invoke the service - let budget = service_inner.test_budget(); - service_inner.handle_invoke( - &invoker_options, - invocation_id, - 0, - InvocationTarget::mock_virtual_object(), - budget, - ); - - // We should receive the new entry here - let invoker_effect = service_inner.invocation_tasks_rx.recv().await.unwrap(); - assert_eq!(invoker_effect.invocation_id, invocation_id); - check!(let InvocationTaskOutputInner::NewEntry { .. } = invoker_effect.inner); - - // Check the quota - assert_eq!(service_inner.quota.available_slots(), 1); - - // Abort the invocation - service_inner.handle_abort_invocation(&invocation_id); - - // Check the quota - assert_eq!(service_inner.quota.available_slots(), 2); - - // Handle error coming after the abort (this should be noop) - service_inner - .handle_invocation_task_failed( - invocation_id, - InvokerError::EmptySuspensionMessage, /* any error is fine */ - service_inner.test_budget(), - ) - .await; - - // Check the quota, should not be changed - assert_eq!(service_inner.quota.available_slots(), 2); - } - #[test(restate_core::test(start_paused = true))] async fn notification_triggers_retry() { let invoker_options = InvokerOptionsBuilder::default() @@ -3206,7 +2983,7 @@ mod tests { // Simulate the invocation task suspending service_inner - .handle_invocation_task_suspended( + .handle_invocation_task_suspended_v2( invocation_id, HashSet::new(), // No pending entries ) diff --git a/crates/invoker-impl/src/quota.rs b/crates/invoker-impl/src/quota.rs index 5a6a16cdc6..5a97134652 100644 --- a/crates/invoker-impl/src/quota.rs +++ b/crates/invoker-impl/src/quota.rs @@ -135,16 +135,6 @@ impl InvokerConcurrencyQuota { } } } - - #[cfg(test)] - pub(super) fn available_slots(&self) -> usize { - match &self.inner { - InvokerConcurrencyQuotaInner::Unlimited => usize::MAX, - InvokerConcurrencyQuotaInner::Limited { slots, .. } => { - slots.available_slots.load(Ordering::Relaxed) - } - } - } } /// An acquired concurrency slot. diff --git a/crates/invoker-impl/src/test_util.rs b/crates/invoker-impl/src/test_util.rs index fbe7be92ea..f3e5e65456 100644 --- a/crates/invoker-impl/src/test_util.rs +++ b/crates/invoker-impl/src/test_util.rs @@ -15,13 +15,8 @@ use bytes::Bytes; use restate_errors::NotRunningError; use restate_memory::{IgnorePinnableMemoryStream, LocalMemoryLease, LocalMemoryPool}; use restate_types::LimitKey; -use restate_types::errors::InvocationError; -use restate_types::identifiers::{EntryIndex, InvocationId, InvocationUuid, ServiceId}; +use restate_types::identifiers::{EntryIndex, InvocationId, ServiceId}; use restate_types::invocation::{FencingToken, InvocationTarget, ServiceInvocationSpanContext}; -use restate_types::journal::enriched::{ - AwakeableEnrichmentResult, CallEnrichmentResult, EnrichedEntryHeader, EnrichedRawEntry, -}; -use restate_types::journal::raw::{PlainEntryHeader, PlainRawEntry, RawEntry}; use restate_types::journal_v2::CommandIndex; use restate_types::time::MillisSinceEpoch; use restate_types::vqueues::VQueueId; @@ -29,7 +24,7 @@ use restate_util_string::ReString; use restate_worker_api::invoker::invocation_reader::{ EagerState, InvocationReader, InvocationReaderTransaction, JournalEntry, JournalKind, }; -use restate_worker_api::invoker::{EntryEnricher, InvokerHandle, JournalMetadata}; +use restate_worker_api::invoker::{InvokerHandle, JournalMetadata}; use restate_worker_api::resources::ReservedResources; #[derive(Debug, Clone, Default)] @@ -196,93 +191,3 @@ impl InvokerHandle for MockInvokerHandle { Ok(()) } } - -#[derive(Debug, Default, Clone)] -pub struct MockEntryEnricher; - -impl EntryEnricher for MockEntryEnricher { - fn enrich_entry( - &mut self, - entry: PlainRawEntry, - _current_invocation_target: &InvocationTarget, - current_invocation_span_context: &ServiceInvocationSpanContext, - ) -> Result { - let (header, entry) = entry.into_inner(); - let enriched_header = match header { - PlainEntryHeader::Input {} => EnrichedEntryHeader::Input {}, - PlainEntryHeader::Output {} => EnrichedEntryHeader::Output {}, - PlainEntryHeader::GetState { is_completed } => { - EnrichedEntryHeader::GetState { is_completed } - } - PlainEntryHeader::SetState {} => EnrichedEntryHeader::SetState {}, - PlainEntryHeader::ClearState {} => EnrichedEntryHeader::ClearState {}, - PlainEntryHeader::GetStateKeys { is_completed } => { - EnrichedEntryHeader::GetStateKeys { is_completed } - } - PlainEntryHeader::ClearAllState {} => EnrichedEntryHeader::ClearAllState {}, - PlainEntryHeader::GetPromise { is_completed } => { - EnrichedEntryHeader::GetPromise { is_completed } - } - PlainEntryHeader::PeekPromise { is_completed } => { - EnrichedEntryHeader::PeekPromise { is_completed } - } - PlainEntryHeader::CompletePromise { is_completed } => { - EnrichedEntryHeader::CompletePromise { is_completed } - } - PlainEntryHeader::Sleep { is_completed } => EnrichedEntryHeader::Sleep { is_completed }, - PlainEntryHeader::Call { is_completed, .. } => { - if !is_completed { - EnrichedEntryHeader::Call { - is_completed, - enrichment_result: Some(CallEnrichmentResult { - invocation_id: InvocationId::mock_random(), - invocation_target: InvocationTarget::service("", ""), - completion_retention_time: None, - span_context: current_invocation_span_context.clone(), - }), - } - } else { - // No need to service resolution if the entry was completed by the service - EnrichedEntryHeader::Call { - is_completed, - enrichment_result: None, - } - } - } - PlainEntryHeader::OneWayCall { .. } => EnrichedEntryHeader::OneWayCall { - enrichment_result: CallEnrichmentResult { - invocation_id: InvocationId::mock_random(), - invocation_target: InvocationTarget::service("", ""), - completion_retention_time: None, - span_context: current_invocation_span_context.clone(), - }, - }, - PlainEntryHeader::Awakeable { is_completed } => { - EnrichedEntryHeader::Awakeable { is_completed } - } - PlainEntryHeader::CompleteAwakeable { .. } => EnrichedEntryHeader::CompleteAwakeable { - enrichment_result: AwakeableEnrichmentResult { - invocation_id: InvocationId::from_parts( - 0, - InvocationUuid::mock_generate(&InvocationTarget::mock_service()), - ), - entry_index: 1, - }, - }, - PlainEntryHeader::Run {} => EnrichedEntryHeader::Run {}, - PlainEntryHeader::Custom { code } => EnrichedEntryHeader::Custom { code }, - PlainEntryHeader::CancelInvocation => EnrichedEntryHeader::CancelInvocation, - PlainEntryHeader::GetCallInvocationId { is_completed } => { - EnrichedEntryHeader::GetCallInvocationId { is_completed } - } - PlainEntryHeader::AttachInvocation { is_completed } => { - EnrichedEntryHeader::AttachInvocation { is_completed } - } - PlainEntryHeader::GetInvocationOutput { is_completed } => { - EnrichedEntryHeader::GetInvocationOutput { is_completed } - } - }; - - Ok(RawEntry::new(enriched_header, entry)) - } -} diff --git a/crates/service-protocol/src/lib.rs b/crates/service-protocol/src/lib.rs index 6bc3f0b012..57467edaf8 100644 --- a/crates/service-protocol/src/lib.rs +++ b/crates/service-protocol/src/lib.rs @@ -11,9 +11,5 @@ //! This crate contains the code-generated structs of [service-protocol](https://github.com/restatedev/service-protocol) and the codec to use them. //! TODO(slinkydeveloper) get rid of this module when service-protocol version <= 3 gets dropped -pub const RESTATE_SERVICE_PROTOCOL_VERSION: u16 = 2; - #[cfg(feature = "codec")] pub mod codec; -#[cfg(feature = "message")] -pub mod message; diff --git a/crates/service-protocol/src/message/encoding.rs b/crates/service-protocol/src/message/encoding.rs deleted file mode 100644 index 92d57fa599..0000000000 --- a/crates/service-protocol/src/message/encoding.rs +++ /dev/null @@ -1,495 +0,0 @@ -// Copyright (c) 2023 - 2026 Restate Software, Inc., Restate GmbH. -// All rights reserved. -// -// Use of this software is governed by the Business Source License -// included in the LICENSE file. -// -// As of the Change Date specified in that file, in accordance with -// the Business Source License, use of this software will be governed -// by the Apache License, Version 2.0. - -use std::mem; -use std::num::NonZeroUsize; - -use bytes::{Buf, BufMut, Bytes, BytesMut}; -use bytes_utils::SegmentedBuf; -use tracing::warn; - -use restate_types::journal::raw::{PlainEntryHeader, RawEntry}; -use restate_types::service_protocol::ServiceProtocolVersion; -use restate_util_bytecount::ByteCount; - -use super::header::UnknownMessageType; -use super::*; - -#[derive(Debug, codederror::CodedError, thiserror::Error)] -#[code(restate_errors::RT0012)] -pub enum EncodingError { - #[error("cannot decode message type {0:?}. This looks like a bug of the SDK. Reason: {1:?}")] - DecodeMessage(MessageType, #[source] prost::DecodeError), - #[error(transparent)] - UnknownMessageType(#[from] UnknownMessageType), - #[error("hit message size limit: {0} >= {1}")] - #[code(restate_errors::RT0003)] - MessageSizeLimit(usize, NonZeroUsize), -} - -// --- Input message encoder - -// TODO: To reduce allocation overhead for small messages (completions, acks), we could -// re-introduce a small bounded arena (e.g. 4-8 KiB) that is reused across encode calls. -// The key constraint is that it must not grow unbounded — the previous arena retained the -// high-water-mark capacity (up to 32 MiB) for the entire invocation lifetime, wasting -// memory across thousands of concurrent long-lived invocations. See #4364. -pub struct Encoder; - -impl Encoder { - pub fn new(service_protocol_version: ServiceProtocolVersion) -> Self { - assert_ne!( - service_protocol_version, - ServiceProtocolVersion::Unspecified, - "A protocol version should be specified" - ); - Self - } - - /// Encodes a message to bytes. - /// - /// Each call allocates a right-sized buffer for the message. This avoids retaining a - /// high-water-mark arena that would hold memory for the lifetime of the encoder — which - /// matters when thousands of long-lived invocations each encoded one large message during - /// replay but only send small completions/acks afterwards. - // Todo: Once we merge thread-local buffer pools (https://github.com/restatedev/restate/pull/4366), - // we can consider passing in a reusable buffer. - pub fn encode(&mut self, msg: ProtocolMessage) -> Bytes { - let len = 8 + msg.encoded_len(); - let mut buf = BytesMut::with_capacity(len); - let header = generate_header(&msg); - buf.put_u64(header.into()); - encode_msg(&msg, &mut buf).expect( - "Encoding messages should be infallible, \ - this error indicates a bug in the invoker code. \ - Please contact the Restate developers.", - ); - buf.freeze() - } -} - -#[inline(always)] -fn generate_header(msg: &ProtocolMessage) -> MessageHeader { - let len: u32 = msg - .encoded_len() - .try_into() - .expect("Protocol messages can't be larger than u32"); - match msg { - ProtocolMessage::Start(_) => MessageHeader::new_start(len), - ProtocolMessage::Completion(_) => MessageHeader::new(MessageType::Completion, len), - ProtocolMessage::Suspension(_) => MessageHeader::new(MessageType::Suspension, len), - ProtocolMessage::Error(_) => MessageHeader::new(MessageType::Error, len), - ProtocolMessage::End(_) => MessageHeader::new(MessageType::End, len), - ProtocolMessage::EntryAck(_) => MessageHeader::new(MessageType::EntryAck, len), - ProtocolMessage::UnparsedEntry(entry) => { - let completed_flag = entry.header().is_completed(); - MessageHeader::new_entry_header( - raw_header_to_message_type(entry.header()), - completed_flag, - len, - ) - } - } -} - -#[inline(always)] -fn encode_msg(msg: &ProtocolMessage, buf: &mut impl BufMut) -> Result<(), prost::EncodeError> { - match msg { - ProtocolMessage::Start(m) => m.encode(buf), - ProtocolMessage::Completion(m) => m.encode(buf), - ProtocolMessage::Suspension(m) => m.encode(buf), - ProtocolMessage::Error(m) => m.encode(buf), - ProtocolMessage::End(m) => m.encode(buf), - ProtocolMessage::EntryAck(m) => m.encode(buf), - ProtocolMessage::UnparsedEntry(entry) => { - buf.put(entry.serialized_entry().clone()); - Ok(()) - } - } -} - -// --- Input message decoder - -/// Stateful decoder to decode [`ProtocolMessage`] -pub struct Decoder { - buf: SegmentedBuf, - state: DecoderState, - message_size_warning: NonZeroUsize, - message_size_limit: NonZeroUsize, -} - -impl Decoder { - pub fn new( - service_protocol_version: ServiceProtocolVersion, - message_size_warning: NonZeroUsize, - message_size_limit: NonZeroUsize, - ) -> Self { - assert_ne!( - service_protocol_version, - ServiceProtocolVersion::Unspecified, - "A protocol version should be specified" - ); - Self { - buf: SegmentedBuf::new(), - state: DecoderState::WaitingHeader, - message_size_warning, - message_size_limit, - } - } - - pub fn has_remaining(&self) -> bool { - self.buf.has_remaining() - } - - /// Concatenate a new chunk in the internal buffer. - pub fn push(&mut self, buf: Bytes) { - self.buf.push(buf) - } - - /// Try to consume the next message in the internal buffer. - pub fn consume_next( - &mut self, - ) -> Result, EncodingError> { - loop { - let remaining = self.buf.remaining(); - - if remaining < self.state.needs_bytes() { - return Ok(None); - } - - if let Some(res) = self.state.decode( - &mut self.buf, - self.message_size_warning, - self.message_size_limit, - )? { - return Ok(Some(res)); - } - } - } -} - -#[derive(Default)] -enum DecoderState { - #[default] - WaitingHeader, - WaitingPayload(MessageHeader), -} - -impl DecoderState { - fn needs_bytes(&self) -> usize { - match self { - DecoderState::WaitingHeader => 8, - DecoderState::WaitingPayload(h) => h.frame_length() as usize, - } - } - - fn decode( - &mut self, - mut buf: impl Buf, - message_size_warning: NonZeroUsize, - message_size_limit: NonZeroUsize, - ) -> Result, EncodingError> { - let mut res = None; - - *self = match mem::take(self) { - DecoderState::WaitingHeader => { - let header: MessageHeader = buf.get_u64().try_into()?; - let message_length = - usize::try_from(header.frame_length()).expect("u32 must convert into usize"); - - if message_length >= message_size_warning.get() { - warn!( - "Message size warning for '{:?}': {} >= {}. \ - Generating very large messages can make the system unstable if configured with too little memory. \ - You can increase the threshold to avoid this warning by changing the worker.invoker.message_size_warning config option", - header.message_type(), - ByteCount::from(message_length), - ByteCount::from(message_size_warning), - ); - } - if message_length >= message_size_limit.get() { - return Err(EncodingError::MessageSizeLimit( - message_length, - message_size_limit, - )); - } - - DecoderState::WaitingPayload(header) - } - DecoderState::WaitingPayload(h) => { - let msg = decode_protocol_message(&h, buf.take(h.frame_length() as usize)) - .map_err(|e| EncodingError::DecodeMessage(h.message_type(), e))?; - res = Some((h, msg)); - DecoderState::WaitingHeader - } - }; - - Ok(res) - } -} - -fn decode_protocol_message( - header: &MessageHeader, - mut buf: impl Buf, -) -> Result { - Ok(match header.message_type() { - MessageType::Start => ProtocolMessage::Start(service_protocol::StartMessage::decode(buf)?), - MessageType::Completion => { - ProtocolMessage::Completion(service_protocol::CompletionMessage::decode(buf)?) - } - MessageType::Suspension => { - ProtocolMessage::Suspension(service_protocol::SuspensionMessage::decode(buf)?) - } - MessageType::Error => ProtocolMessage::Error(service_protocol::ErrorMessage::decode(buf)?), - MessageType::End => ProtocolMessage::End(service_protocol::EndMessage::decode(buf)?), - MessageType::EntryAck => { - ProtocolMessage::EntryAck(service_protocol::EntryAckMessage::decode(buf)?) - } - _ => ProtocolMessage::UnparsedEntry(RawEntry::new( - message_header_to_raw_header(header), - // NOTE: This is a no-op copy if the Buf is instance of Bytes. - // In case of SegmentedBuf, this doesn't copy if the whole message is contained - // in a single Bytes instance. - buf.copy_to_bytes(buf.remaining()), - )), - }) -} - -macro_rules! expect_flag { - ($message_header:expr, $name:ident) => { - MessageHeader::$name($message_header) - .expect(concat!(stringify!($name), " flag being present")) - }; -} - -fn message_header_to_raw_header(message_header: &MessageHeader) -> PlainEntryHeader { - debug_assert!( - !matches!( - message_header.message_type(), - MessageType::Start - | MessageType::Completion - | MessageType::Suspension - | MessageType::EntryAck - | MessageType::Error - | MessageType::End - ), - "Message is not an entry type. This is a Restate bug. Please contact the developers." - ); - match message_header.message_type() { - MessageType::Start => unreachable!(), - MessageType::Completion => unreachable!(), - MessageType::Suspension => unreachable!(), - MessageType::Error => unreachable!(), - MessageType::End => unreachable!(), - MessageType::EntryAck => unreachable!(), - - MessageType::InputEntry => PlainEntryHeader::Input {}, - MessageType::OutputEntry => PlainEntryHeader::Output {}, - MessageType::GetStateEntry => PlainEntryHeader::GetState { - is_completed: expect_flag!(message_header, completed), - }, - MessageType::SetStateEntry => PlainEntryHeader::SetState {}, - MessageType::ClearStateEntry => PlainEntryHeader::ClearState {}, - MessageType::GetStateKeysEntry => PlainEntryHeader::GetStateKeys { - is_completed: expect_flag!(message_header, completed), - }, - MessageType::ClearAllStateEntry => PlainEntryHeader::ClearAllState {}, - MessageType::GetPromiseEntry => PlainEntryHeader::GetPromise { - is_completed: expect_flag!(message_header, completed), - }, - MessageType::PeekPromiseEntry => PlainEntryHeader::PeekPromise { - is_completed: expect_flag!(message_header, completed), - }, - MessageType::CompletePromiseEntry => PlainEntryHeader::CompletePromise { - is_completed: expect_flag!(message_header, completed), - }, - MessageType::SleepEntry => PlainEntryHeader::Sleep { - is_completed: expect_flag!(message_header, completed), - }, - MessageType::InvokeEntry => PlainEntryHeader::Call { - is_completed: expect_flag!(message_header, completed), - enrichment_result: None, - }, - MessageType::BackgroundInvokeEntry => PlainEntryHeader::OneWayCall { - enrichment_result: (), - }, - MessageType::AwakeableEntry => PlainEntryHeader::Awakeable { - is_completed: expect_flag!(message_header, completed), - }, - MessageType::CompleteAwakeableEntry => PlainEntryHeader::CompleteAwakeable { - enrichment_result: (), - }, - MessageType::SideEffectEntry => PlainEntryHeader::Run {}, - MessageType::CancelInvocationEntry => PlainEntryHeader::CancelInvocation {}, - MessageType::GetCallInvocationIdEntry => PlainEntryHeader::GetCallInvocationId { - is_completed: expect_flag!(message_header, completed), - }, - MessageType::AttachInvocationEntry => PlainEntryHeader::AttachInvocation { - is_completed: expect_flag!(message_header, completed), - }, - MessageType::GetInvocationOutputEntry => PlainEntryHeader::GetInvocationOutput { - is_completed: expect_flag!(message_header, completed), - }, - MessageType::CustomEntry(code) => PlainEntryHeader::Custom { code }, - } -} - -fn raw_header_to_message_type(entry_header: &PlainEntryHeader) -> MessageType { - match entry_header { - PlainEntryHeader::Input { .. } => MessageType::InputEntry, - PlainEntryHeader::Output { .. } => MessageType::OutputEntry, - PlainEntryHeader::GetState { .. } => MessageType::GetStateEntry, - PlainEntryHeader::SetState { .. } => MessageType::SetStateEntry, - PlainEntryHeader::ClearState { .. } => MessageType::ClearStateEntry, - PlainEntryHeader::GetStateKeys { .. } => MessageType::GetStateKeysEntry, - PlainEntryHeader::ClearAllState { .. } => MessageType::ClearAllStateEntry, - PlainEntryHeader::GetPromise { .. } => MessageType::GetPromiseEntry, - PlainEntryHeader::PeekPromise { .. } => MessageType::PeekPromiseEntry, - PlainEntryHeader::CompletePromise { .. } => MessageType::CompletePromiseEntry, - PlainEntryHeader::Sleep { .. } => MessageType::SleepEntry, - PlainEntryHeader::Call { .. } => MessageType::InvokeEntry, - PlainEntryHeader::OneWayCall { .. } => MessageType::BackgroundInvokeEntry, - PlainEntryHeader::Awakeable { .. } => MessageType::AwakeableEntry, - PlainEntryHeader::CompleteAwakeable { .. } => MessageType::CompleteAwakeableEntry, - PlainEntryHeader::Run { .. } => MessageType::SideEffectEntry, - PlainEntryHeader::CancelInvocation => MessageType::CancelInvocationEntry, - PlainEntryHeader::GetCallInvocationId { .. } => MessageType::GetCallInvocationIdEntry, - PlainEntryHeader::AttachInvocation { .. } => MessageType::AttachInvocationEntry, - PlainEntryHeader::GetInvocationOutput { .. } => MessageType::GetInvocationOutputEntry, - PlainEntryHeader::Custom { code, .. } => MessageType::CustomEntry(*code), - } -} - -#[cfg(test)] -mod tests { - use super::*; - - use crate::codec::ProtobufRawEntryCodec; - use restate_test_util::{assert, assert_eq, let_assert}; - use restate_types::journal::raw::RawEntryCodec; - - #[test] - fn fill_decoder_with_several_messages() { - let mut encoder = Encoder::new(ServiceProtocolVersion::V1); - let mut decoder = Decoder::new( - ServiceProtocolVersion::V1, - NonZeroUsize::MAX, - NonZeroUsize::MAX, - ); - - let expected_msg_0 = ProtocolMessage::new_start_message( - "key".into(), - "key".into(), - Some("key".into()), - 1, - true, - vec![], - 10, - Duration::ZERO, - ); - - let expected_msg_1: ProtocolMessage = ProtobufRawEntryCodec::serialize_as_input_entry( - vec![], - Bytes::from_static("input".as_bytes()), - ) - .erase_enrichment() - .into(); - let expected_msg_2: ProtocolMessage = Completion { - entry_index: 1, - result: CompletionResult::Empty, - } - .into(); - - decoder.push(encoder.encode(expected_msg_0.clone())); - decoder.push(encoder.encode(expected_msg_1.clone())); - decoder.push(encoder.encode(expected_msg_2.clone())); - - let (actual_msg_header_0, actual_msg_0) = decoder.consume_next().unwrap().unwrap(); - assert_eq!(actual_msg_header_0.message_type(), MessageType::Start); - assert_eq!(actual_msg_0, expected_msg_0); - - let (actual_msg_header_1, actual_msg_1) = decoder.consume_next().unwrap().unwrap(); - assert_eq!(actual_msg_header_1.message_type(), MessageType::InputEntry); - assert_eq!(actual_msg_header_1.completed(), None); - assert_eq!(actual_msg_1, expected_msg_1); - - let (actual_msg_header_2, actual_msg_2) = decoder.consume_next().unwrap().unwrap(); - assert_eq!(actual_msg_header_2.message_type(), MessageType::Completion); - assert_eq!(actual_msg_2, expected_msg_2); - - assert!(decoder.consume_next().unwrap().is_none()); - } - - #[test] - fn fill_decoder_with_partial_header() { - partial_decoding_test(4) - } - - #[test] - fn fill_decoder_with_partial_body() { - partial_decoding_test(10) - } - - fn partial_decoding_test(split_index: usize) { - let mut encoder = Encoder::new(ServiceProtocolVersion::V1); - let mut decoder = Decoder::new( - ServiceProtocolVersion::V1, - NonZeroUsize::MAX, - NonZeroUsize::MAX, - ); - - let expected_msg: ProtocolMessage = ProtobufRawEntryCodec::serialize_as_input_entry( - vec![], - Bytes::from_static("input".as_bytes()), - ) - .erase_enrichment() - .into(); - let expected_msg_encoded = encoder.encode(expected_msg.clone()); - - decoder.push(expected_msg_encoded.slice(0..split_index)); - assert!(decoder.consume_next().unwrap().is_none()); - - decoder.push(expected_msg_encoded.slice(split_index..)); - - let (actual_msg_header, actual_msg) = decoder.consume_next().unwrap().unwrap(); - assert_eq!(actual_msg_header.message_type(), MessageType::InputEntry); - assert_eq!(actual_msg_header.completed(), None); - assert_eq!(actual_msg, expected_msg); - - assert!(decoder.consume_next().unwrap().is_none()); - } - - #[test] - fn hit_message_size_limit() { - let mut decoder = Decoder::new( - ServiceProtocolVersion::V1, - NonZeroUsize::new((u8::MAX / 2) as usize).unwrap(), - NonZeroUsize::new(u8::MAX as usize).unwrap(), - ); - - let mut encoder = Encoder::new(ServiceProtocolVersion::V1); - let message = ProtocolMessage::from( - ProtobufRawEntryCodec::serialize_as_input_entry( - vec![], - (0..=u8::MAX).collect::>().into(), - ) - .erase_enrichment(), - ); - let expected_msg_size = message.encoded_len(); - let msg = encoder.encode(message); - - decoder.push(msg); - let_assert!( - EncodingError::MessageSizeLimit(msg_size, limit) = decoder.consume_next().unwrap_err() - ); - assert_eq!(msg_size, expected_msg_size); - assert_eq!(limit, NonZeroUsize::new(u8::MAX as usize).unwrap()) - } -} diff --git a/crates/service-protocol/src/message/header.rs b/crates/service-protocol/src/message/header.rs deleted file mode 100644 index 7f66da97a3..0000000000 --- a/crates/service-protocol/src/message/header.rs +++ /dev/null @@ -1,545 +0,0 @@ -// Copyright (c) 2023 - 2026 Restate Software, Inc., Restate GmbH. -// All rights reserved. -// -// Use of this software is governed by the Business Source License -// included in the LICENSE file. -// -// As of the Change Date specified in that file, in accordance with -// the Business Source License, use of this software will be governed -// by the Apache License, Version 2.0. - -use restate_types::journal::EntryType; - -const CUSTOM_MESSAGE_MASK: u16 = 0xFC00; -const COMPLETED_MASK: u64 = 0x0001_0000_0000; -const REQUIRES_ACK_MASK: u64 = 0x8000_0000_0000; - -type MessageTypeId = u16; - -#[derive(Copy, Clone, Debug, PartialEq, Eq)] -pub enum MessageKind { - Core, - IO, - State, - Syscall, - CustomEntry, -} - -#[derive(Copy, Clone, Debug, PartialEq, Eq)] -pub enum MessageType { - Start, - Completion, - Suspension, - Error, - End, - EntryAck, - InputEntry, - OutputEntry, - GetStateEntry, - SetStateEntry, - ClearStateEntry, - GetStateKeysEntry, - ClearAllStateEntry, - SleepEntry, - InvokeEntry, - BackgroundInvokeEntry, - AwakeableEntry, - CompleteAwakeableEntry, - SideEffectEntry, - GetPromiseEntry, - PeekPromiseEntry, - CompletePromiseEntry, - CancelInvocationEntry, - GetCallInvocationIdEntry, - AttachInvocationEntry, - GetInvocationOutputEntry, - CustomEntry(u16), -} - -impl MessageType { - fn kind(&self) -> MessageKind { - match self { - MessageType::Start => MessageKind::Core, - MessageType::Completion => MessageKind::Core, - MessageType::Suspension => MessageKind::Core, - MessageType::Error => MessageKind::Core, - MessageType::End => MessageKind::Core, - MessageType::EntryAck => MessageKind::Core, - MessageType::InputEntry => MessageKind::IO, - MessageType::OutputEntry => MessageKind::IO, - MessageType::GetStateEntry => MessageKind::State, - MessageType::SetStateEntry => MessageKind::State, - MessageType::ClearStateEntry => MessageKind::State, - MessageType::GetStateKeysEntry => MessageKind::State, - MessageType::ClearAllStateEntry => MessageKind::State, - MessageType::SleepEntry => MessageKind::Syscall, - MessageType::InvokeEntry => MessageKind::Syscall, - MessageType::BackgroundInvokeEntry => MessageKind::Syscall, - MessageType::AwakeableEntry => MessageKind::Syscall, - MessageType::CompleteAwakeableEntry => MessageKind::Syscall, - MessageType::SideEffectEntry => MessageKind::Syscall, - MessageType::GetPromiseEntry => MessageKind::State, - MessageType::PeekPromiseEntry => MessageKind::State, - MessageType::CompletePromiseEntry => MessageKind::State, - MessageType::CancelInvocationEntry => MessageKind::Syscall, - MessageType::GetCallInvocationIdEntry => MessageKind::Syscall, - MessageType::AttachInvocationEntry => MessageKind::Syscall, - MessageType::GetInvocationOutputEntry => MessageKind::Syscall, - MessageType::CustomEntry(_) => MessageKind::CustomEntry, - } - } - - fn has_completed_flag(&self) -> bool { - matches!( - self, - MessageType::GetStateEntry - | MessageType::GetStateKeysEntry - | MessageType::SleepEntry - | MessageType::InvokeEntry - | MessageType::AwakeableEntry - | MessageType::GetPromiseEntry - | MessageType::PeekPromiseEntry - | MessageType::CompletePromiseEntry - | MessageType::GetCallInvocationIdEntry - | MessageType::AttachInvocationEntry - | MessageType::GetInvocationOutputEntry - ) - } - - fn has_requires_ack_flag(&self) -> bool { - matches!( - self.kind(), - MessageKind::State | MessageKind::IO | MessageKind::Syscall | MessageKind::CustomEntry - ) - } -} - -const START_MESSAGE_TYPE: u16 = 0x0000; -const COMPLETION_MESSAGE_TYPE: u16 = 0x0001; -const SUSPENSION_MESSAGE_TYPE: u16 = 0x0002; -const ERROR_MESSAGE_TYPE: u16 = 0x0003; -const ENTRY_ACK_MESSAGE_TYPE: u16 = 0x0004; -const END_MESSAGE_TYPE: u16 = 0x0005; -const INPUT_ENTRY_MESSAGE_TYPE: u16 = 0x0400; -const OUTPUT_ENTRY_MESSAGE_TYPE: u16 = 0x0401; -const GET_STATE_ENTRY_MESSAGE_TYPE: u16 = 0x0800; -const SET_STATE_ENTRY_MESSAGE_TYPE: u16 = 0x0801; -const CLEAR_STATE_ENTRY_MESSAGE_TYPE: u16 = 0x0802; -const CLEAR_ALL_STATE_ENTRY_MESSAGE_TYPE: u16 = 0x0803; -const GET_STATE_KEYS_ENTRY_MESSAGE_TYPE: u16 = 0x0804; -const GET_PROMISE_ENTRY_MESSAGE_TYPE: u16 = 0x0808; -const PEEK_PROMISE_ENTRY_MESSAGE_TYPE: u16 = 0x0809; -const COMPLETE_PROMISE_ENTRY_MESSAGE_TYPE: u16 = 0x080A; -const SLEEP_ENTRY_MESSAGE_TYPE: u16 = 0x0C00; -const INVOKE_ENTRY_MESSAGE_TYPE: u16 = 0x0C01; -const BACKGROUND_INVOKE_ENTRY_MESSAGE_TYPE: u16 = 0x0C02; -const AWAKEABLE_ENTRY_MESSAGE_TYPE: u16 = 0x0C03; -const COMPLETE_AWAKEABLE_ENTRY_MESSAGE_TYPE: u16 = 0x0C04; -const SIDE_EFFECT_ENTRY_MESSAGE_TYPE: u16 = 0x0C05; -const CANCEL_INVOCATION_ENTRY_MESSAGE_TYPE: u16 = 0x0C06; -const GET_CALL_INVOCATION_ID_ENTRY_MESSAGE_TYPE: u16 = 0x0C07; -const ATTACH_INVOCATION_ENTRY_MESSAGE_TYPE: u16 = 0x0C08; -const GET_INVOCATION_OUTPUT_ENTRY_MESSAGE_TYPE: u16 = 0x0C09; - -impl From for MessageTypeId { - fn from(mt: MessageType) -> Self { - match mt { - MessageType::Start => START_MESSAGE_TYPE, - MessageType::Completion => COMPLETION_MESSAGE_TYPE, - MessageType::Suspension => SUSPENSION_MESSAGE_TYPE, - MessageType::Error => ERROR_MESSAGE_TYPE, - MessageType::End => END_MESSAGE_TYPE, - MessageType::EntryAck => ENTRY_ACK_MESSAGE_TYPE, - MessageType::InputEntry => INPUT_ENTRY_MESSAGE_TYPE, - MessageType::OutputEntry => OUTPUT_ENTRY_MESSAGE_TYPE, - MessageType::GetStateEntry => GET_STATE_ENTRY_MESSAGE_TYPE, - MessageType::SetStateEntry => SET_STATE_ENTRY_MESSAGE_TYPE, - MessageType::ClearStateEntry => CLEAR_STATE_ENTRY_MESSAGE_TYPE, - MessageType::ClearAllStateEntry => CLEAR_ALL_STATE_ENTRY_MESSAGE_TYPE, - MessageType::GetStateKeysEntry => GET_STATE_KEYS_ENTRY_MESSAGE_TYPE, - MessageType::SleepEntry => SLEEP_ENTRY_MESSAGE_TYPE, - MessageType::InvokeEntry => INVOKE_ENTRY_MESSAGE_TYPE, - MessageType::BackgroundInvokeEntry => BACKGROUND_INVOKE_ENTRY_MESSAGE_TYPE, - MessageType::AwakeableEntry => AWAKEABLE_ENTRY_MESSAGE_TYPE, - MessageType::CompleteAwakeableEntry => COMPLETE_AWAKEABLE_ENTRY_MESSAGE_TYPE, - MessageType::SideEffectEntry => SIDE_EFFECT_ENTRY_MESSAGE_TYPE, - MessageType::GetPromiseEntry => GET_PROMISE_ENTRY_MESSAGE_TYPE, - MessageType::PeekPromiseEntry => PEEK_PROMISE_ENTRY_MESSAGE_TYPE, - MessageType::CompletePromiseEntry => COMPLETE_PROMISE_ENTRY_MESSAGE_TYPE, - MessageType::CancelInvocationEntry => CANCEL_INVOCATION_ENTRY_MESSAGE_TYPE, - MessageType::GetCallInvocationIdEntry => GET_CALL_INVOCATION_ID_ENTRY_MESSAGE_TYPE, - MessageType::AttachInvocationEntry => ATTACH_INVOCATION_ENTRY_MESSAGE_TYPE, - MessageType::GetInvocationOutputEntry => GET_INVOCATION_OUTPUT_ENTRY_MESSAGE_TYPE, - MessageType::CustomEntry(id) => id, - } - } -} - -#[derive(Debug, thiserror::Error)] -#[error("unknown message code {0:#x}")] -pub struct UnknownMessageType(u16); - -impl TryFrom for MessageType { - type Error = UnknownMessageType; - - fn try_from(value: MessageTypeId) -> Result { - match value { - START_MESSAGE_TYPE => Ok(MessageType::Start), - COMPLETION_MESSAGE_TYPE => Ok(MessageType::Completion), - SUSPENSION_MESSAGE_TYPE => Ok(MessageType::Suspension), - ERROR_MESSAGE_TYPE => Ok(MessageType::Error), - END_MESSAGE_TYPE => Ok(MessageType::End), - ENTRY_ACK_MESSAGE_TYPE => Ok(MessageType::EntryAck), - INPUT_ENTRY_MESSAGE_TYPE => Ok(MessageType::InputEntry), - OUTPUT_ENTRY_MESSAGE_TYPE => Ok(MessageType::OutputEntry), - GET_STATE_ENTRY_MESSAGE_TYPE => Ok(MessageType::GetStateEntry), - SET_STATE_ENTRY_MESSAGE_TYPE => Ok(MessageType::SetStateEntry), - CLEAR_STATE_ENTRY_MESSAGE_TYPE => Ok(MessageType::ClearStateEntry), - GET_STATE_KEYS_ENTRY_MESSAGE_TYPE => Ok(MessageType::GetStateKeysEntry), - CLEAR_ALL_STATE_ENTRY_MESSAGE_TYPE => Ok(MessageType::ClearAllStateEntry), - SLEEP_ENTRY_MESSAGE_TYPE => Ok(MessageType::SleepEntry), - INVOKE_ENTRY_MESSAGE_TYPE => Ok(MessageType::InvokeEntry), - BACKGROUND_INVOKE_ENTRY_MESSAGE_TYPE => Ok(MessageType::BackgroundInvokeEntry), - AWAKEABLE_ENTRY_MESSAGE_TYPE => Ok(MessageType::AwakeableEntry), - COMPLETE_AWAKEABLE_ENTRY_MESSAGE_TYPE => Ok(MessageType::CompleteAwakeableEntry), - GET_PROMISE_ENTRY_MESSAGE_TYPE => Ok(MessageType::GetPromiseEntry), - PEEK_PROMISE_ENTRY_MESSAGE_TYPE => Ok(MessageType::PeekPromiseEntry), - COMPLETE_PROMISE_ENTRY_MESSAGE_TYPE => Ok(MessageType::CompletePromiseEntry), - SIDE_EFFECT_ENTRY_MESSAGE_TYPE => Ok(MessageType::SideEffectEntry), - CANCEL_INVOCATION_ENTRY_MESSAGE_TYPE => Ok(MessageType::CancelInvocationEntry), - GET_CALL_INVOCATION_ID_ENTRY_MESSAGE_TYPE => Ok(MessageType::GetCallInvocationIdEntry), - ATTACH_INVOCATION_ENTRY_MESSAGE_TYPE => Ok(MessageType::AttachInvocationEntry), - GET_INVOCATION_OUTPUT_ENTRY_MESSAGE_TYPE => Ok(MessageType::GetInvocationOutputEntry), - v if ((v & CUSTOM_MESSAGE_MASK) != 0) => Ok(MessageType::CustomEntry(v)), - v => Err(UnknownMessageType(v)), - } - } -} - -impl TryFrom for EntryType { - type Error = MessageType; - - fn try_from(value: MessageType) -> Result { - match value { - MessageType::InputEntry => Ok(EntryType::Input), - MessageType::OutputEntry => Ok(EntryType::Output), - MessageType::GetStateEntry => Ok(EntryType::GetState), - MessageType::SetStateEntry => Ok(EntryType::SetState), - MessageType::ClearStateEntry => Ok(EntryType::ClearState), - MessageType::GetStateKeysEntry => Ok(EntryType::GetStateKeys), - MessageType::ClearAllStateEntry => Ok(EntryType::ClearAllState), - MessageType::SleepEntry => Ok(EntryType::Sleep), - MessageType::InvokeEntry => Ok(EntryType::Call), - MessageType::BackgroundInvokeEntry => Ok(EntryType::OneWayCall), - MessageType::AwakeableEntry => Ok(EntryType::Awakeable), - MessageType::CompleteAwakeableEntry => Ok(EntryType::CompleteAwakeable), - MessageType::SideEffectEntry => Ok(EntryType::Run), - MessageType::GetPromiseEntry => Ok(EntryType::GetPromise), - MessageType::PeekPromiseEntry => Ok(EntryType::PeekPromise), - MessageType::CompletePromiseEntry => Ok(EntryType::CompletePromise), - MessageType::CancelInvocationEntry => Ok(EntryType::CancelInvocation), - MessageType::GetCallInvocationIdEntry => Ok(EntryType::GetCallInvocationId), - MessageType::AttachInvocationEntry => Ok(EntryType::AttachInvocation), - MessageType::GetInvocationOutputEntry => Ok(EntryType::GetInvocationOutput), - MessageType::CustomEntry(_) => Ok(EntryType::Custom), - MessageType::Start - | MessageType::Completion - | MessageType::Suspension - | MessageType::Error - | MessageType::End - | MessageType::EntryAck => Err(value), - } - } -} - -#[derive(Clone, Debug, PartialEq, Eq)] -pub struct MessageHeader { - ty: MessageType, - length: u32, - - // --- Flags - /// Only `CompletableEntries` have completed flag. See [`MessageType#allows_completed_flag`]. - completed_flag: Option, - /// All Entry messages may have requires ack flag. - requires_ack_flag: Option, -} - -impl MessageHeader { - #[inline] - pub fn new(ty: MessageType, length: u32) -> Self { - Self::_new(ty, None, None, length) - } - - #[inline] - pub fn new_start(length: u32) -> Self { - Self::_new(MessageType::Start, None, None, length) - } - - #[inline] - pub(super) fn new_entry_header( - ty: MessageType, - completed_flag: Option, - length: u32, - ) -> Self { - debug_assert!(completed_flag.is_some() == ty.has_completed_flag()); - - MessageHeader { - ty, - length, - completed_flag, - // It is always false when sending entries from the runtime - requires_ack_flag: Some(false), - } - } - - #[inline] - fn _new( - ty: MessageType, - completed_flag: Option, - requires_ack_flag: Option, - length: u32, - ) -> Self { - MessageHeader { - ty, - length, - completed_flag, - requires_ack_flag, - } - } - - #[inline] - pub fn message_kind(&self) -> MessageKind { - self.ty.kind() - } - - #[inline] - pub fn message_type(&self) -> MessageType { - self.ty - } - - #[inline] - pub fn completed(&self) -> Option { - self.completed_flag - } - - #[inline] - pub fn requires_ack(&self) -> Option { - self.requires_ack_flag - } - - #[inline] - pub fn frame_length(&self) -> u32 { - self.length - } -} - -macro_rules! read_flag_if { - ($cond:expr, $value:expr, $mask:expr) => { - if $cond { - Some(($value & $mask) != 0) - } else { - None - } - }; -} - -impl TryFrom for MessageHeader { - type Error = UnknownMessageType; - - /// Deserialize the protocol header. - /// See https://github.com/restatedev/service-protocol/blob/main/service-invocation-protocol.md#message-header - fn try_from(value: u64) -> Result { - let ty_code = (value >> 48) as u16; - let ty: MessageType = ty_code.try_into()?; - - let completed_flag = read_flag_if!(ty.has_completed_flag(), value, COMPLETED_MASK); - let requires_ack_flag = read_flag_if!(ty.has_requires_ack_flag(), value, REQUIRES_ACK_MASK); - let length = value as u32; - - Ok(MessageHeader::_new( - ty, - completed_flag, - requires_ack_flag, - length, - )) - } -} - -macro_rules! write_flag { - ($flag:expr, $value:expr, $mask:expr) => { - if let Some(true) = $flag { - *$value |= $mask; - } - }; -} - -impl From for u64 { - /// Serialize the protocol header. - /// See https://github.com/restatedev/service-protocol/blob/main/service-invocation-protocol.md#message-header - fn from(message_header: MessageHeader) -> Self { - let mut res = - ((u16::from(message_header.ty) as u64) << 48) | (message_header.length as u64); - - write_flag!(message_header.completed_flag, &mut res, COMPLETED_MASK); - write_flag!( - message_header.requires_ack_flag, - &mut res, - REQUIRES_ACK_MASK - ); - - res - } -} - -#[cfg(test)] -mod tests { - - use super::{MessageKind::*, MessageType::*, *}; - - impl MessageHeader { - fn new_completable_entry(ty: MessageType, completed: bool, length: u32) -> Self { - Self::new_entry_header(ty, Some(completed), length) - } - } - - macro_rules! roundtrip_test { - ($test_name:ident, $header:expr, $ty:expr, $kind:expr, $len:expr) => { - roundtrip_test!($test_name, $header, $ty, $kind, $len, None, None, None); - }; - ($test_name:ident, $header:expr, $ty:expr, $kind:expr, $len:expr, version: $protocol_version:expr) => { - roundtrip_test!( - $test_name, - $header, - $ty, - $kind, - $len, - None, - Some($protocol_version), - None - ); - }; - ($test_name:ident, $header:expr, $ty:expr, $kind:expr, $len:expr, completed: $completed:expr) => { - roundtrip_test!( - $test_name, - $header, - $ty, - $kind, - $len, - Some($completed), - None, - None - ); - }; - ($test_name:ident, $header:expr, $ty:expr, $kind:expr, $len:expr, requires_ack: $requires_ack:expr) => { - roundtrip_test!( - $test_name, - $header, - $ty, - $kind, - $len, - None, - None, - Some($requires_ack) - ); - }; - ($test_name:ident, $header:expr, $ty:expr, $kind:expr, $len:expr, requires_ack: $requires_ack:expr, completed: $completed:expr) => { - roundtrip_test!( - $test_name, - $header, - $ty, - $kind, - $len, - Some($completed), - None, - Some($requires_ack) - ); - }; - ($test_name:ident, $header:expr, $ty:expr, $kind:expr, $len:expr, $completed:expr, $protocol_version:expr, $requires_ack:expr) => { - #[test] - fn $test_name() { - let serialized: u64 = $header.into(); - let header: MessageHeader = serialized.try_into().unwrap(); - - assert_eq!(header.message_type(), $ty); - assert_eq!(header.message_kind(), $kind); - assert_eq!(header.completed(), $completed); - assert_eq!(header.requires_ack(), $requires_ack); - assert_eq!(header.frame_length(), $len); - } - }; - } - - roundtrip_test!( - start, - MessageHeader::new_start(25), - Start, - Core, - 25, - version: 1 - ); - - roundtrip_test!( - completion, - MessageHeader::new(Completion, 22), - Completion, - Core, - 22 - ); - - roundtrip_test!( - completed_get_state, - MessageHeader::new_completable_entry(GetStateEntry, true, 0), - GetStateEntry, - State, - 0, - requires_ack: false, - completed: true - ); - - roundtrip_test!( - not_completed_get_state, - MessageHeader::new_completable_entry(GetStateEntry, false, 0), - GetStateEntry, - State, - 0, - requires_ack: false, - completed: false - ); - - roundtrip_test!( - completed_get_state_with_len, - MessageHeader::new_completable_entry(GetStateEntry, true, 10341), - GetStateEntry, - State, - 10341, - requires_ack: false, - completed: true - ); - - roundtrip_test!( - set_state_with_requires_ack, - MessageHeader::_new(SetStateEntry, None, Some(true), 10341), - SetStateEntry, - State, - 10341, - requires_ack: true - ); - - roundtrip_test!( - custom_entry, - MessageHeader::new(MessageType::CustomEntry(0xFC00), 10341), - MessageType::CustomEntry(0xFC00), - MessageKind::CustomEntry, - 10341, - requires_ack: false - ); - - roundtrip_test!( - custom_entry_with_requires_ack, - MessageHeader::_new(MessageType::CustomEntry(0xFC00), None, Some(true), 10341), - MessageType::CustomEntry(0xFC00), - MessageKind::CustomEntry, - 10341, - requires_ack: true - ); -} diff --git a/crates/service-protocol/src/message/mod.rs b/crates/service-protocol/src/message/mod.rs deleted file mode 100644 index de583e1a93..0000000000 --- a/crates/service-protocol/src/message/mod.rs +++ /dev/null @@ -1,124 +0,0 @@ -// Copyright (c) 2023 - 2026 Restate Software, Inc., Restate GmbH. -// All rights reserved. -// -// Use of this software is governed by the Business Source License -// included in the LICENSE file. -// -// As of the Change Date specified in that file, in accordance with -// the Business Source License, use of this software will be governed -// by the Apache License, Version 2.0. - -//! Module containing definitions of Protocol messages, -//! including encoding and decoding of headers and message payloads. - -mod encoding; -mod header; - -use std::time::Duration; - -use bytes::Bytes; -use prost::Message; - -use restate_types::journal::CompletionResult; -use restate_types::journal::raw::PlainRawEntry; -use restate_types::journal::{Completion, EntryIndex}; -use restate_types::service_protocol; - -pub use encoding::{Decoder, Encoder, EncodingError}; -pub use header::{MessageHeader, MessageKind, MessageType}; -pub use restate_types::service_protocol::start_message::StateEntry; - -#[derive(Debug, Clone, PartialEq)] -pub enum ProtocolMessage { - // Core - Start(service_protocol::StartMessage), - Completion(service_protocol::CompletionMessage), - Suspension(service_protocol::SuspensionMessage), - Error(service_protocol::ErrorMessage), - End(service_protocol::EndMessage), - EntryAck(service_protocol::EntryAckMessage), - - // Entries are not parsed at this point - UnparsedEntry(PlainRawEntry), -} - -impl ProtocolMessage { - #[allow(clippy::too_many_arguments)] - pub fn new_start_message( - id: Bytes, - debug_id: String, - key: Option, - known_entries: u32, - partial_state: bool, - state_map: Vec, - retry_count_since_last_stored_entry: u32, - duration_since_last_stored_entry: Duration, - ) -> Self { - Self::Start(service_protocol::StartMessage { - id, - debug_id, - known_entries, - partial_state, - state_map, - key: key - .and_then(|b| String::from_utf8(b.to_vec()).ok()) - .unwrap_or_default(), - retry_count_since_last_stored_entry, - duration_since_last_stored_entry: duration_since_last_stored_entry.as_millis() as u64, - }) - } - - pub fn new_entry_ack(entry_index: EntryIndex) -> ProtocolMessage { - Self::EntryAck(service_protocol::EntryAckMessage { entry_index }) - } - - pub(crate) fn encoded_len(&self) -> usize { - match self { - ProtocolMessage::Start(m) => m.encoded_len(), - ProtocolMessage::Completion(m) => m.encoded_len(), - ProtocolMessage::Suspension(m) => m.encoded_len(), - ProtocolMessage::Error(m) => m.encoded_len(), - ProtocolMessage::End(m) => m.encoded_len(), - ProtocolMessage::EntryAck(m) => m.encoded_len(), - ProtocolMessage::UnparsedEntry(entry) => entry.serialized_entry().len(), - } - } -} - -impl From for ProtocolMessage { - fn from(completion: Completion) -> Self { - match completion.result { - CompletionResult::Empty => { - ProtocolMessage::Completion(service_protocol::CompletionMessage { - entry_index: completion.entry_index, - result: Some(service_protocol::completion_message::Result::Empty( - service_protocol::Empty {}, - )), - }) - } - CompletionResult::Success(b) => { - ProtocolMessage::Completion(service_protocol::CompletionMessage { - entry_index: completion.entry_index, - result: Some(service_protocol::completion_message::Result::Value(b)), - }) - } - CompletionResult::Failure(code, message) => { - ProtocolMessage::Completion(service_protocol::CompletionMessage { - entry_index: completion.entry_index, - result: Some(service_protocol::completion_message::Result::Failure( - service_protocol::Failure { - code: code.into(), - message: message.to_string(), - }, - )), - }) - } - } - } -} - -impl From for ProtocolMessage { - fn from(value: PlainRawEntry) -> Self { - Self::UnparsedEntry(value) - } -} diff --git a/crates/worker-api/src/invoker/entry_enricher.rs b/crates/worker-api/src/invoker/entry_enricher.rs deleted file mode 100644 index ad2214647a..0000000000 --- a/crates/worker-api/src/invoker/entry_enricher.rs +++ /dev/null @@ -1,23 +0,0 @@ -// Copyright (c) 2023 - 2026 Restate Software, Inc., Restate GmbH. -// All rights reserved. -// -// Use of this software is governed by the Business Source License -// included in the LICENSE file. -// -// As of the Change Date specified in that file, in accordance with -// the Business Source License, use of this software will be governed -// by the Apache License, Version 2.0. - -use restate_types::errors::InvocationError; -use restate_types::invocation::{InvocationTarget, ServiceInvocationSpanContext}; -use restate_types::journal::enriched::EnrichedRawEntry; -use restate_types::journal::raw::PlainRawEntry; - -pub trait EntryEnricher { - fn enrich_entry( - &mut self, - entry: PlainRawEntry, - current_invocation_target: &InvocationTarget, - current_invocation_span_context: &ServiceInvocationSpanContext, - ) -> Result; -} diff --git a/crates/worker-api/src/invoker/mod.rs b/crates/worker-api/src/invoker/mod.rs index 70356985b5..f4a28e850e 100644 --- a/crates/worker-api/src/invoker/mod.rs +++ b/crates/worker-api/src/invoker/mod.rs @@ -10,13 +10,11 @@ pub mod capacity; mod effects; -pub mod entry_enricher; mod handle; pub mod invocation_reader; pub mod status_handle; pub use effects::*; -pub use entry_enricher::EntryEnricher; pub use handle::*; pub use invocation_reader::{InvocationReaderError, JournalKind, JournalMetadata}; pub use restate_storage_api::vqueue_table::scheduler::YieldReason; diff --git a/crates/worker/src/invoker_integration.rs b/crates/worker/src/invoker_integration.rs deleted file mode 100644 index 8cec141022..0000000000 --- a/crates/worker/src/invoker_integration.rs +++ /dev/null @@ -1,406 +0,0 @@ -// Copyright (c) 2023 - 2026 Restate Software, Inc., Restate GmbH. -// All rights reserved. -// -// Use of this software is governed by the Business Source License -// included in the LICENSE file. -// -// As of the Change Date specified in that file, in accordance with -// the Business Source License, use of this software will be governed -// by the Apache License, Version 2.0. - -use std::marker::PhantomData; -use std::ops::Deref; -use std::str::FromStr; - -use anyhow::anyhow; -use assert2::let_assert; -use bytes::Bytes; -use bytestring::ByteString; - -use restate_types::errors::{InvocationError, codes}; -use restate_types::identifiers::{AwakeableIdentifier, ExternalSignalIdentifier, InvocationId}; -use restate_types::invocation::{ - InvocationTarget, InvocationTargetType, ServiceInvocationSpanContext, ServiceType, SpanRelation, -}; -use restate_types::journal::enriched::{ - AwakeableEnrichmentResult, CallEnrichmentResult, EnrichedEntryHeader, EnrichedRawEntry, -}; -use restate_types::journal::raw::{PlainEntryHeader, PlainRawEntry, RawEntry, RawEntryCodec}; -use restate_types::journal::{ - AttachInvocationEntry, AttachInvocationTarget, CancelInvocationEntry, CancelInvocationTarget, - CompleteAwakeableEntry, Entry, GetInvocationOutputEntry, InvokeEntry, OneWayCallEntry, -}; -use restate_types::journal::{EntryType, InvokeRequest}; -use restate_types::journal_v2::SignalId; -use restate_types::live::Live; -use restate_types::schema::invocation_target::{DeploymentStatus, InvocationTargetResolver}; - -#[derive(Clone)] -pub(super) struct EntryEnricher { - schemas: Live, - - _codec: PhantomData, -} - -impl EntryEnricher { - pub(super) fn new(schemas: Live) -> Self { - Self { - schemas, - _codec: Default::default(), - } - } -} - -impl EntryEnricher -where - Schemas: InvocationTargetResolver, - Codec: RawEntryCodec, -{ - fn resolve_service_invocation_target( - &mut self, - entry_type: EntryType, - serialized_entry: &Bytes, - request_extractor: impl Fn(Entry) -> InvokeRequest, - span_relation: SpanRelation, - ) -> Result { - let entry = Codec::deserialize(entry_type, serialized_entry.clone()) - .map_err(|e| InvocationError::internal(e.to_string()))?; - let request = request_extractor(entry); - - let meta = self - .schemas - .live_load() - .resolve_latest_invocation_target(&request.service_name, &request.handler_name) - .ok_or_else(|| { - InvocationError::service_handler_not_found( - &request.service_name, - &request.handler_name, - ) - })?; - if let DeploymentStatus::Deprecated(dp_id) = meta.deployment_status { - return Err(InvocationError::new( - codes::INTERNAL, - format!( - "The service {} is exposed by the deprecated deployment {dp_id}. Upgrade the SDK used by {} and register a new deployment.", - request.service_name, request.service_name - ), - )); - } - - let invocation_target = match meta.target_ty { - InvocationTargetType::Service => { - InvocationTarget::service(request.service_name, request.handler_name) - } - InvocationTargetType::VirtualObject(h_ty) => InvocationTarget::virtual_object( - request.service_name.clone(), - ByteString::try_from(request.key.clone().into_bytes()).map_err(|e| { - InvocationError::from(anyhow!( - "The request key is not a valid UTF-8 string: {e}" - )) - })?, - request.handler_name, - h_ty, - ), - InvocationTargetType::Workflow(h_ty) => InvocationTarget::workflow( - request.service_name.clone(), - ByteString::try_from(request.key.clone().into_bytes()).map_err(|e| { - InvocationError::from(anyhow!( - "The request key is not a valid UTF-8 string: {e}" - )) - })?, - request.handler_name, - h_ty, - ), - }; - - let idempotency_key = if let Some(idempotency_key) = &request.idempotency_key { - if idempotency_key.is_empty() { - return Err(InvocationError::from(anyhow!( - "The provided idempotency key is empty" - ))); - } - Some(idempotency_key.deref()) - } else { - None - }; - let invocation_id = InvocationId::generate(&invocation_target, idempotency_key); - - // Create the span context - let span_context = ServiceInvocationSpanContext::start(&invocation_id, span_relation); - - let completion_retention_duration = meta - .compute_retention(idempotency_key.is_some()) - .completion_retention; - - Ok(CallEnrichmentResult { - invocation_id, - invocation_target, - completion_retention_time: if completion_retention_duration.is_zero() { - None - } else { - Some(completion_retention_duration) - }, - span_context, - }) - } -} - -impl restate_worker_api::invoker::EntryEnricher for EntryEnricher -where - Schemas: InvocationTargetResolver, - Codec: RawEntryCodec, -{ - fn enrich_entry( - &mut self, - entry: PlainRawEntry, - current_invocation_target: &InvocationTarget, - current_invocation_span_context: &ServiceInvocationSpanContext, - ) -> Result { - let (header, serialized_entry) = entry.into_inner(); - - let enriched_header = match header { - PlainEntryHeader::Input {} => EnrichedEntryHeader::Input {}, - PlainEntryHeader::Output {} => EnrichedEntryHeader::Output {}, - PlainEntryHeader::GetState { is_completed } => { - can_read_state( - &header.as_entry_type(), - ¤t_invocation_target.invocation_target_ty(), - )?; - EnrichedEntryHeader::GetState { is_completed } - } - PlainEntryHeader::SetState {} => { - can_write_state( - &header.as_entry_type(), - ¤t_invocation_target.invocation_target_ty(), - )?; - EnrichedEntryHeader::SetState {} - } - PlainEntryHeader::ClearState {} => { - can_write_state( - &header.as_entry_type(), - ¤t_invocation_target.invocation_target_ty(), - )?; - EnrichedEntryHeader::ClearState {} - } - PlainEntryHeader::GetStateKeys { is_completed } => { - can_read_state( - &header.as_entry_type(), - ¤t_invocation_target.invocation_target_ty(), - )?; - EnrichedEntryHeader::GetStateKeys { is_completed } - } - PlainEntryHeader::ClearAllState => { - can_write_state( - &header.as_entry_type(), - ¤t_invocation_target.invocation_target_ty(), - )?; - EnrichedEntryHeader::ClearAllState {} - } - PlainEntryHeader::GetPromise { is_completed } => { - check_workflow_type( - &header.as_entry_type(), - ¤t_invocation_target.service_ty(), - )?; - EnrichedEntryHeader::GetPromise { is_completed } - } - PlainEntryHeader::PeekPromise { is_completed } => { - check_workflow_type( - &header.as_entry_type(), - ¤t_invocation_target.service_ty(), - )?; - EnrichedEntryHeader::PeekPromise { is_completed } - } - PlainEntryHeader::CompletePromise { is_completed } => { - check_workflow_type( - &header.as_entry_type(), - ¤t_invocation_target.service_ty(), - )?; - EnrichedEntryHeader::CompletePromise { is_completed } - } - PlainEntryHeader::Sleep { is_completed } => EnrichedEntryHeader::Sleep { is_completed }, - PlainEntryHeader::Call { is_completed, .. } => { - if !is_completed { - let enrichment_result = self.resolve_service_invocation_target( - header.as_entry_type(), - &serialized_entry, - |entry| { - let_assert!(Entry::Call(InvokeEntry { request, .. }) = entry); - request - }, - current_invocation_span_context.as_parent(), - )?; - - EnrichedEntryHeader::Call { - is_completed, - enrichment_result: Some(enrichment_result), - } - } else { - // No need to service resolution if the entry was completed by the deployment - EnrichedEntryHeader::Call { - is_completed, - enrichment_result: None, - } - } - } - PlainEntryHeader::OneWayCall { .. } => { - let enrichment_result = self.resolve_service_invocation_target( - header.as_entry_type(), - &serialized_entry, - |entry| { - let_assert!(Entry::OneWayCall(OneWayCallEntry { request, .. }) = entry); - request - }, - current_invocation_span_context.as_linked(), - )?; - - EnrichedEntryHeader::OneWayCall { enrichment_result } - } - PlainEntryHeader::Awakeable { is_completed } => { - EnrichedEntryHeader::Awakeable { is_completed } - } - PlainEntryHeader::CompleteAwakeable { .. } => { - let entry = - Codec::deserialize(EntryType::CompleteAwakeable, serialized_entry.clone()) - .map_err(|e| InvocationError::internal(e.to_string()))?; - let_assert!(Entry::CompleteAwakeable(CompleteAwakeableEntry { id, .. }) = entry); - - let (invocation_id, entry_index) = if let Ok(old_awk_id) = - AwakeableIdentifier::from_str(&id) - { - old_awk_id.into_inner() - } else if let Ok(new_awk_id) = ExternalSignalIdentifier::from_str(&id) { - let (invocation_id, signal_id) = new_awk_id.into_inner(); - if let SignalId::Index(idx) = signal_id { - (invocation_id, idx) - } else { - return Err(InvocationError::new( - codes::BAD_REQUEST, - "Unsupported awakeable signal identifier. Only signals with auto generated id can be completed using service protocol <= v3.".to_string(), - )); - } - } else { - return Err(InvocationError::new( - codes::BAD_REQUEST, - "Invalid awakeable identifier. The identifier doesn't start with `awk_1`, neither with `sign_1`".to_string(), - )); - }; - - EnrichedEntryHeader::CompleteAwakeable { - enrichment_result: AwakeableEnrichmentResult { - invocation_id, - entry_index, - }, - } - } - PlainEntryHeader::Run { .. } => EnrichedEntryHeader::Run {}, - PlainEntryHeader::CancelInvocation { .. } => { - // Validate the invocation id is valid - let entry = - Codec::deserialize(EntryType::CancelInvocation, serialized_entry.clone()) - .map_err(|e| InvocationError::internal(e.to_string()))?; - let_assert!(Entry::CancelInvocation(CancelInvocationEntry { target }) = entry); - if let CancelInvocationTarget::InvocationId(id) = target - && let Err(e) = id.parse::() - { - return Err(InvocationError::new( - codes::BAD_REQUEST, - format!("The given invocation id '{id}' to cancel is invalid: {e}"), - )); - } - - EnrichedEntryHeader::CancelInvocation {} - } - PlainEntryHeader::AttachInvocation { is_completed } => { - // Validate the invocation id is valid - let entry = - Codec::deserialize(EntryType::AttachInvocation, serialized_entry.clone()) - .map_err(|e| InvocationError::internal(e.to_string()))?; - let_assert!(Entry::AttachInvocation(AttachInvocationEntry { target, .. }) = entry); - if let AttachInvocationTarget::InvocationId(id) = target - && let Err(e) = id.parse::() - { - return Err(InvocationError::new( - codes::BAD_REQUEST, - format!("The given invocation id '{id}' to attach is invalid: {e}"), - )); - } - - EnrichedEntryHeader::AttachInvocation { is_completed } - } - PlainEntryHeader::GetInvocationOutput { is_completed } => { - // Validate the invocation id is valid - let entry = - Codec::deserialize(EntryType::GetInvocationOutput, serialized_entry.clone()) - .map_err(|e| InvocationError::internal(e.to_string()))?; - let_assert!( - Entry::GetInvocationOutput(GetInvocationOutputEntry { target, .. }) = entry - ); - if let AttachInvocationTarget::InvocationId(id) = target - && let Err(e) = id.parse::() - { - return Err(InvocationError::new( - codes::BAD_REQUEST, - format!("The given invocation id '{id}' to get output is invalid: {e}"), - )); - } - - EnrichedEntryHeader::GetInvocationOutput { is_completed } - } - PlainEntryHeader::GetCallInvocationId { is_completed } => { - EnrichedEntryHeader::GetCallInvocationId { is_completed } - } - PlainEntryHeader::Custom { code } => EnrichedEntryHeader::Custom { code }, - }; - - Ok(RawEntry::new(enriched_header, serialized_entry)) - } -} - -#[inline] -fn check_workflow_type( - entry_type: &EntryType, - service_type: &ServiceType, -) -> Result<(), InvocationError> { - if *service_type != ServiceType::Workflow { - return Err(InvocationError::new( - codes::BAD_REQUEST, - format!( - "The service type {service_type} does not support the entry type {entry_type}, only Workflow supports it" - ), - )); - } - Ok(()) -} - -#[inline] -fn can_read_state( - entry_type: &EntryType, - invocation_target_type: &InvocationTargetType, -) -> Result<(), InvocationError> { - if !invocation_target_type.can_read_state() { - return Err(InvocationError::new( - codes::BAD_REQUEST, - format!( - "The service/handler type {invocation_target_type} does not have state and, therefore, does not support the entry type {entry_type}" - ), - )); - } - Ok(()) -} - -#[inline] -fn can_write_state( - entry_type: &EntryType, - invocation_target_type: &InvocationTargetType, -) -> Result<(), InvocationError> { - can_read_state(entry_type, invocation_target_type)?; - if !invocation_target_type.can_write_state() { - return Err(InvocationError::new( - codes::BAD_REQUEST, - format!( - "The service/handler type {invocation_target_type} has no exclusive state access and, therefore, does not support the entry type {entry_type}" - ), - )); - } - Ok(()) -} diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 7c477a630b..10b93078c8 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -12,7 +12,6 @@ extern crate core; mod error; mod handle; -mod invoker_integration; mod metric_definitions; #[cfg(feature = "expose-internals")] pub mod partition; diff --git a/crates/worker/src/partition/leadership/mod.rs b/crates/worker/src/partition/leadership/mod.rs index 67416ea7bc..26ed50f733 100644 --- a/crates/worker/src/partition/leadership/mod.rs +++ b/crates/worker/src/partition/leadership/mod.rs @@ -35,7 +35,6 @@ use restate_invoker_impl::{ InvokerHandle as InvokerChannelServiceHandle, Service as InvokerService, }; use restate_partition_store::PartitionStore; -use restate_service_protocol::codec::ProtobufRawEntryCodec; use restate_storage_api::StorageError; use restate_storage_api::deduplication_table::EpochSequenceNumber; use restate_storage_api::invocation_status_table::{ @@ -78,7 +77,6 @@ use restate_worker_api::{ use self::durability_tracker::DurabilityTracker; use self::fencing::FencingTokens; use self::trim_queue::{HasTrimQueue, LogTrimmer}; -use crate::invoker_integration::EntryEnricher; use crate::partition::LeadershipInfo; use crate::partition::cleaner::{self, Cleaner}; use crate::partition::invoker_storage_reader::InvokerStorageReader; @@ -667,23 +665,19 @@ where let (invoker_tx, invoker_rx) = mpsc::channel(config.worker.internal_queue_length()); let invoker_rx = ReceiverStream::new(invoker_rx); - let invoker: InvokerService< - InvokerStorageReader, - EntryEnricher, - Schema, - > = InvokerService::from_options( - processor.partition_id(), - processor.key_range(), - InvokerStorageReader::new(partition_store.clone()), - invoker_tx, - &config.worker.invoker.service_client, - &config.worker.invoker, - EntryEnricher::new(schema.clone()), - schema, - node_ctx.invoker_capacity.invocation_token_bucket.clone(), - node_ctx.invoker_capacity.action_token_bucket.clone(), - node_ctx.invoker_capacity.memory_pool.clone(), - )?; + let invoker: InvokerService, Schema> = + InvokerService::from_options( + processor.partition_id(), + processor.key_range(), + InvokerStorageReader::new(partition_store.clone()), + invoker_tx, + &config.worker.invoker.service_client, + &config.worker.invoker, + schema, + node_ctx.invoker_capacity.invocation_token_bucket.clone(), + node_ctx.invoker_capacity.action_token_bucket.clone(), + node_ctx.invoker_capacity.memory_pool.clone(), + )?; let mut invoker_handle = invoker.handle(); From 6b19758a2c05e2a789b958dcca4757d3ef43881f Mon Sep 17 00:00:00 2001 From: slinkydeveloper Date: Thu, 10 Sep 2026 14:29:40 +0200 Subject: [PATCH 3/3] Release notes --- .../unreleased/drop-service-protocol-v3.md | 47 +++++++++++++++++++ 1 file changed, 47 insertions(+) create mode 100644 release-notes/unreleased/drop-service-protocol-v3.md diff --git a/release-notes/unreleased/drop-service-protocol-v3.md b/release-notes/unreleased/drop-service-protocol-v3.md new file mode 100644 index 0000000000..e96715362e --- /dev/null +++ b/release-notes/unreleased/drop-service-protocol-v3.md @@ -0,0 +1,47 @@ +# Release Notes: Drop support for service protocol <= v3 + +## Breaking Change + +### What Changed +Restate no longer runs invocations against deployments that use service protocol +version 3 or lower. When such an invocation is attempted, the invoker now fails it +immediately with `RT0020` (`service is exposed by the deprecated deployment +, please upgrade the SDK used by the service`) instead of retrying. + +### Why This Matters +Service protocol <= v3 relies on runner code that has been removed. Continuing to run +these invocations is no longer possible, so rather than retrying indefinitely they are +now failed fast and surfaced to you. + +### Impact on Users +- **New deployments**: Register services with an SDK that speaks service protocol v4 or + later. Older SDKs are rejected. +- **Existing deployments**: Any invocation still pinned to service protocol <= v3 will + **fail** after upgrade. The invocation is not retried; it terminates with an error. + +### Migration Guidance + +**Before upgrading**, check whether you still have invocations pinned to service +protocol v3 or lower. The `sys_invocation` table exposes the negotiated protocol +version in the `pinned_service_protocol_version` column: + +```sql +SELECT id, target, pinned_deployment_id, pinned_service_protocol_version +FROM sys_invocation +WHERE pinned_service_protocol_version <= 3; +``` + +> Note: `pinned_service_protocol_version` is only set after the first journal entry has +> been stored for an invocation, so newly created invocations that have not started yet +> will not appear here. + +If the query returns rows, drain or complete those invocations before upgrading, or +upgrade the SDK behind the affected deployment so new attempts negotiate protocol v4+. + +**After upgrading**, invocations that failed with this error can be re-run from the +beginning using **restart-as-new**, which starts a fresh invocation from scratch (the +new invocation records the original in the `restarted_from` column of `sys_invocation`). + +### Related Issues +- Auto-fail invocations using service protocol <= 3 +- Remove service protocol runner <= v3