Don't allow requests deployments using old protocol anymore (#3877)

* Don't allow requests deployments using old protocol anymore

* Apply suggestions from code review

Co-authored-by: Till Rohrmann <[email protected]>

---------

Co-authored-by: Till Rohrmann <[email protected]>

Author: slinkydeveloper

crates/errors/src/error_codes/RT0020.md

@@ -0,0 +1,4 @@
+## RT0020
+
+The requested service is exposed by a deployment using a deprecated service protocol version. New requests won't be accepted.
+Upgrade the SDK in your service code and register it as a new deployment to continue accepting requests to the requested service.

crates/errors/src/lib.rs

@@ -36,7 +36,7 @@ mod helper;
 
 declare_restate_error_codes!(
     RT0001, RT0002, RT0003, RT0004, RT0005, RT0006, RT0007, RT0009, RT0010, RT0011, RT0012, RT0013,
-    RT0014, RT0015, RT0016, RT0017, RT0018, RT0019, META0003, META0004, META0005, META0006,
+    RT0014, RT0015, RT0016, RT0017, RT0018, RT0019, RT0020, META0003, META0004, META0005, META0006,
     META0009, META0010, META0011, META0012, META0013, META0014, META0015, META0016, META0017
 );
 

crates/ingress-http/src/handler/error.rs

@@ -14,6 +14,7 @@ use crate::RequestDispatcherError;
 use bytes::Bytes;
 use http::{Response, StatusCode, header};
 use restate_types::errors::{IdDecodeError, InvocationError};
+use restate_types::identifiers::DeploymentId;
 use restate_types::schema::invocation_target::InputValidationError;
 use serde::Serialize;
 use std::string;
@@ -28,6 +29,10 @@ pub(crate) enum HandlerError {
         "the service '{0}' exists, but the handler '{1}' was not found, check that the handler exists in the latest registered service version."
     )]
     ServiceHandlerNotFound(String, String),
+    #[error(
+        "the service {0} is exposed by the deprecated deployment {1}. Upgrade the SDK used by {0}."
+    )]
+    DeploymentDeprecated(String, DeploymentId),
     #[error("invocation not found")]
     InvocationNotFound,
     #[error(
@@ -128,7 +133,8 @@ impl HandlerError {
             | HandlerError::BadWorkflowPath
             | HandlerError::InputValidation(_)
             | HandlerError::UnsupportedIdempotencyKey
-            | HandlerError::UnsupportedGetOutput => StatusCode::BAD_REQUEST,
+            | HandlerError::UnsupportedGetOutput
+            | HandlerError::DeploymentDeprecated(_, _) => StatusCode::BAD_REQUEST,
             HandlerError::DispatcherError(_) => {
                 // TODO add more distinctions between different dispatcher errors (unavailable, etc)
                 StatusCode::INTERNAL_SERVER_ERROR

crates/ingress-http/src/handler/service_handler.rs

@@ -34,7 +34,7 @@ use restate_types::invocation::{
     SpanRelation, WorkflowHandlerType,
 };
 use restate_types::schema::invocation_target::{
-    InvocationTargetMetadata, InvocationTargetResolver,
+    DeploymentStatus, InvocationTargetMetadata, InvocationTargetResolver,
 };
 use restate_types::time::MillisSinceEpoch;
 
@@ -101,6 +101,9 @@ where
                 handler_name.clone(),
             ));
         };
+        if let DeploymentStatus::Deprecated(dp_id) = invocation_target_meta.deployment_status {
+            return Err(HandlerError::DeploymentDeprecated(service_name, dp_id));
+        }
 
         // Check if Idempotency-Key is available
         let idempotency_key = parse_idempotency(req.headers())?;

crates/ingress-kafka/src/dispatcher.rs

@@ -8,13 +8,13 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-use std::borrow::Borrow;
-use std::sync::Arc;
-
+use anyhow::bail;
 use bytes::Bytes;
 use opentelemetry::propagation::{Extractor, TextMapPropagator};
 use opentelemetry::trace::{Span, SpanContext, TraceContextExt};
 use opentelemetry_sdk::propagation::TraceContextPropagator;
+use std::borrow::Borrow;
+use std::sync::Arc;
 use tracing::debug;
 
 use restate_bifrost::Bifrost;
@@ -25,7 +25,7 @@ use restate_types::live;
 use restate_types::message::MessageIndex;
 use restate_types::partition_table::PartitionTableError;
 use restate_types::schema::Schema;
-use restate_types::schema::invocation_target::InvocationTargetResolver;
+use restate_types::schema::invocation_target::{DeploymentStatus, InvocationTargetResolver};
 use restate_types::schema::subscriptions::{EventInvocationTargetTemplate, Sink, Subscription};
 use restate_wal_protocol::{Command, Destination, Envelope, Header, Source};
 
@@ -102,13 +102,21 @@ impl KafkaIngressEvent {
         };
 
         // Compute the retention values
-        let invocation_retention = schema
+        let target = schema
             .resolve_latest_invocation_target(
                 invocation_target.service_name(),
                 invocation_target.handler_name(),
             )
-            .ok_or_else(|| anyhow::anyhow!("Service and handler are not registered"))?
-            .compute_retention(false);
+            .ok_or_else(|| anyhow::anyhow!("Service and handler are not registered"))?;
+
+        if let DeploymentStatus::Deprecated(dp_id) = target.deployment_status {
+            bail!(
+                "the service {} is exposed by the deprecated deployment {dp_id}, please upgrade the SDK.",
+                invocation_target.service_name()
+            )
+        }
+
+        let invocation_retention = target.compute_retention(false);
 
         // Time to generate invocation id
         let invocation_id = InvocationId::generate(&invocation_target, None);

crates/invoker-impl/src/error.rs

@@ -293,6 +293,9 @@ pub(crate) enum CommandPreconditionError {
     NoStateOperations,
     #[error("unsupported entry type, this handler type cannot write state")]
     NoWriteStateOperations,
+    #[error("the service {0} is exposed by the deprecated deployment {1}.")]
+    #[code(restate_errors::RT0020)]
+    DeploymentDeprecated(String, DeploymentId),
 }
 
 #[derive(Debug)]

crates/invoker-impl/src/invocation_task/service_protocol_runner_v4.rs

@@ -50,7 +50,7 @@ use restate_types::journal_v2::{
     CommandIndex, CommandType, Entry, EntryType, NotificationId, RunCompletion, RunResult, SignalId,
 };
 use restate_types::schema::deployment::{Deployment, DeploymentType, ProtocolType};
-use restate_types::schema::invocation_target::InvocationTargetResolver;
+use restate_types::schema::invocation_target::{DeploymentStatus, InvocationTargetResolver};
 use restate_types::service_protocol::ServiceProtocolVersion;
 
 use crate::Notification;
@@ -1027,6 +1027,13 @@ fn resolve_call_request(
             )
         })?;
 
+    if let DeploymentStatus::Deprecated(dp_id) = meta.deployment_status {
+        return Err(CommandPreconditionError::DeploymentDeprecated(
+            request.service_name.to_string(),
+            dp_id,
+        ));
+    }
+
     if !request.key.is_empty() && !meta.target_ty.is_keyed() {
         return Err(CommandPreconditionError::UnexpectedKey(
             request.service_name.to_string(),

crates/types/src/schema/invocation_target.rs

@@ -72,6 +72,14 @@ pub enum OnMaxAttempts {
     Kill,
 }
 
+#[derive(Debug, Clone, Default)]
+pub enum DeploymentStatus {
+    #[default]
+    Enabled,
+    /// Deployment is disabled, and new requests should not be accepted.
+    Deprecated(DeploymentId),
+}
+
 #[derive(Debug, Clone)]
 pub struct InvocationTargetMetadata {
     pub public: bool,
@@ -84,6 +92,8 @@ pub struct InvocationTargetMetadata {
     pub target_ty: InvocationTargetType,
     pub input_rules: InputRules,
     pub output_rules: OutputRules,
+
+    pub deployment_status: DeploymentStatus,
 }
 
 impl InvocationTargetMetadata {
@@ -577,6 +587,7 @@ pub mod test_util {
                 target_ty: invocation_target_type,
                 input_rules: Default::default(),
                 output_rules: Default::default(),
+                deployment_status: DeploymentStatus::Enabled,
             }
         }
     }

crates/types/src/schema/metadata/mod.rs

@@ -31,16 +31,17 @@ use crate::net::metadata::{MetadataContainer, MetadataKind};
 use crate::retries::{RetryIter, RetryPolicy};
 use crate::schema::deployment::{DeploymentResolver, DeploymentType};
 use crate::schema::invocation_target::{
-    DEFAULT_IDEMPOTENCY_RETENTION, DEFAULT_WORKFLOW_COMPLETION_RETENTION, InputRules,
-    InvocationAttemptOptions, InvocationTargetMetadata, InvocationTargetResolver, OnMaxAttempts,
-    OutputRules,
+    DEFAULT_IDEMPOTENCY_RETENTION, DEFAULT_WORKFLOW_COMPLETION_RETENTION, DeploymentStatus,
+    InputRules, InvocationAttemptOptions, InvocationTargetMetadata, InvocationTargetResolver,
+    OnMaxAttempts, OutputRules,
 };
 use crate::schema::metadata::openapi::ServiceOpenAPI;
 use crate::schema::service::{
     HandlerRetryPolicyMetadata, ServiceMetadataResolver, ServiceRetryPolicyMetadata,
 };
 use crate::schema::subscriptions::{ListSubscriptionFilter, Subscription, SubscriptionResolver};
 use crate::schema::{deployment, service};
+use crate::service_protocol::ServiceProtocolVersion;
 use crate::time::MillisSinceEpoch;
 use crate::{Version, Versioned, identifiers};
 
@@ -597,7 +598,8 @@ impl InvocationTargetResolver for Schema {
         let handler_name = handler_name.as_ref();
 
         let ActiveServiceRevision {
-            service_revision, ..
+            service_revision,
+            deployment_id,
         } = self.active_service_revisions.get(service_name)?;
         let handler = service_revision.handlers.get(handler_name)?;
 
@@ -627,13 +629,32 @@ impl InvocationTargetResolver for Schema {
             )
             .unwrap_or(Duration::ZERO);
 
+        let deployment_status = self
+            .deployments
+            .get(deployment_id)
+            .map(|dp| {
+                if ServiceProtocolVersion::is_acceptable_for_new_invocations(
+                    *dp.supported_protocol_versions.start(),
+                    *dp.supported_protocol_versions.end(),
+                ) {
+                    DeploymentStatus::Enabled
+                } else {
+                    DeploymentStatus::Deprecated(dp.id)
+                }
+            })
+            // It should never happen that the deployment doesn't exist,
+            // this is an invalid schema registry otherwise.
+            // But let's not panic yet, this will fail later on.
+            .unwrap_or_default();
+
         Some(InvocationTargetMetadata {
             public: handler.public.unwrap_or(service_revision.public),
             completion_retention,
             journal_retention,
             target_ty: handler.target_ty,
             input_rules: handler.input_rules.clone(),
             output_rules: handler.output_rules.clone(),
+            deployment_status,
         })
     }
 

crates/types/src/service_protocol.rs

@@ -37,6 +37,10 @@ impl ServiceProtocolVersion {
             && max_version >= i32::from(MIN_DISCOVERABLE_SERVICE_PROTOCOL_VERSION)
     }
 
+    pub fn is_acceptable_for_new_invocations(min_version: i32, max_version: i32) -> bool {
+        Self::is_acceptable_for_discovery(min_version, max_version)
+    }
+
     pub fn is_supported_for_inflight_invocation(&self) -> bool {
         MIN_INFLIGHT_SERVICE_PROTOCOL_VERSION <= *self
             && *self <= MAX_INFLIGHT_SERVICE_PROTOCOL_VERSION