Skip to content

Disable Protocol < V4 in invoker #5259

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
slinkydeveloper merged 3 commits into from
Sep 10, 2026
Merged

Disable Protocol < V4 in invoker #5259

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0d4ec7d
Auto-fail invocations using service protocol <= 3
slinkydeveloper Aug 28, 2026
162111a
Remove service protocol runner <= v3, and cascade remove all the othe…
slinkydeveloper Aug 28, 2026
6b19758
Release notes
slinkydeveloper Sep 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 3 additions & 44 deletions crates/invoker-impl/src/error.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -60,21 +58,12 @@ pub(crate) enum InvokerError {
UnexpectedContentType(Option<HeaderValue>, 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,
Expand All @@ -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>, EntryIndex),
#[error("malformed ProposeRunCompletionMessage, missing result field")]
#[code(restate_errors::RT0012)]
MalformedProposeRunCompletion,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -312,6 +293,7 @@ impl InvokerError {
RequestedErrorBehavior::retry(*retry_after)
}
InvokerError::MaxFutureDepthReached { .. } => RequestedErrorBehavior::Pause,
InvokerError::DeploymentDeprecated { .. } => RequestedErrorBehavior::Fail,
_ => RequestedErrorBehavior::Retry,
}
}
Expand All @@ -320,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())
}
Expand Down Expand Up @@ -418,16 +387,6 @@ pub(crate) struct SdkInvocationError {
pub(crate) error: Box<InvocationError>,
}

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 {
Expand Down
88 changes: 24 additions & 64 deletions crates/invoker-impl/src/invocation_task/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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, JournalKind,
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.
Expand Down Expand Up @@ -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<EnrichedRawEntry>,
/// 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,
Expand All @@ -196,7 +182,6 @@ pub(super) enum InvocationTaskOutputInner {
unresolved_future: UnresolvedFuture,
},
Closed,
Suspended(HashSet<EntryIndex>),
SuspendedV2(HashSet<NotificationId>),
SuspendedV3(UnresolvedFuture),
Failed(InvokerError, LocalMemoryPool),
Expand Down Expand Up @@ -254,7 +239,7 @@ fn new_invoker_body(
}

/// Represents an open invocation stream
pub(super) struct InvocationTask<EE, DMR> {
pub(super) struct InvocationTask<DMR> {
// Shared client
client: ServiceClient,

Expand All @@ -273,7 +258,6 @@ pub(super) struct InvocationTask<EE, DMR> {
max_awaited_future_depth: usize,

// Invoker tx/rx
entry_enricher: EE,
schemas: Live<DMR>,
invoker_tx: mpsc::UnboundedSender<InvocationTaskOutput>,
invoker_rx: mpsc::UnboundedReceiver<Notification>,
Expand All @@ -288,7 +272,6 @@ pub(super) struct InvocationTask<EE, DMR> {
enum TerminalLoopState<T> {
Continue(T),
Closed,
Suspended(HashSet<EntryIndex>),
SuspendedV2(HashSet<NotificationId>),
SuspendedV3(UnresolvedFuture),
Failed(InvokerError),
Expand All @@ -302,7 +285,7 @@ impl<T> TerminalLoopState<T> {
}

fn is_suspend(&self) -> bool {
matches!(self, Self::Suspended(_) | Self::SuspendedV2(_))
matches!(self, Self::SuspendedV2(_) | Self::SuspendedV3(_))
}
}

Expand All @@ -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),
Expand All @@ -337,9 +319,8 @@ macro_rules! shortcircuit {
};
}

impl<EE, Schemas> InvocationTask<EE, Schemas>
impl<Schemas> InvocationTask<Schemas>
where
EE: EntryEnricher,
Schemas: DeploymentResolver + InvocationTargetResolver,
{
#[allow(clippy::too_many_arguments)]
Expand All @@ -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<Schemas>,
invoker_tx: mpsc::UnboundedSender<InvocationTaskOutput>,
invoker_rx: mpsc::UnboundedReceiver<Notification>,
Expand All @@ -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,
Expand Down Expand Up @@ -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) => {
Expand Down Expand Up @@ -547,9 +525,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(),
Expand Down Expand Up @@ -579,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<EE, Schemas> InvocationTask<EE, Schemas> {
impl<Schemas> InvocationTask<Schemas> {
/// 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 {
Expand Down
Loading
Loading