Clean up the SubscriptionValidator abstraction (#3866)

This is not a useful abstraction these days, it was introduced long ago when we didn't have Configuration::pinned. (#3866)

Author: slinkydeveloper

crates/admin/src/rest_api/deployments.rs

@@ -56,8 +56,8 @@ use serde::Deserialize;
         from_type = "MetaApiError",
     )
 )]
-pub async fn create_deployment<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn create_deployment<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Extension(version): Extension<AdminApiVersion>,
     #[request_body(required = true)] Json(payload): Json<RegisterDeploymentRequest>,
 ) -> Result<impl IntoResponse, MetaApiError> {
@@ -181,8 +181,8 @@ pub async fn create_deployment<V, IC>(
         schema = "std::string::String"
     ))
 )]
-pub async fn get_deployment<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn get_deployment<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(deployment_id): Path<DeploymentId>,
 ) -> Result<Json<DetailedDeploymentResponse>, MetaApiError> {
     let (deployment, services) = state
@@ -200,8 +200,8 @@ pub async fn get_deployment<V, IC>(
     operation_id = "list_deployments",
     tags = "deployment"
 )]
-pub async fn list_deployments<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn list_deployments<IC>(
+    State(state): State<AdminServiceState<IC>>,
 ) -> Json<ListDeploymentsResponse> {
     let deployments = state
         .schema_registry
@@ -254,8 +254,8 @@ pub struct DeleteDeploymentParams {
         from_type = "MetaApiError",
     )
 )]
-pub async fn delete_deployment<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn delete_deployment<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(deployment_id): Path<DeploymentId>,
     Query(DeleteDeploymentParams { force }): Query<DeleteDeploymentParams>,
 ) -> Result<StatusCode, MetaApiError> {
@@ -286,8 +286,8 @@ pub async fn delete_deployment<V, IC>(
         schema = "std::string::String"
     ))
 )]
-pub async fn update_deployment<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn update_deployment<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Extension(version): Extension<AdminApiVersion>,
     method: Method,
     Path(deployment_id): Path<DeploymentId>,

crates/admin/src/rest_api/handlers.rs

@@ -29,8 +29,8 @@ use restate_types::schema::service::HandlerMetadata;
         schema = "std::string::String"
     ))
 )]
-pub async fn list_service_handlers<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn list_service_handlers<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(service_name): Path<String>,
 ) -> Result<Json<ListServiceHandlersResponse>, MetaApiError> {
     match state.schema_registry.list_service_handlers(&service_name) {
@@ -58,8 +58,8 @@ pub async fn list_service_handlers<V, IC>(
         )
     )
 )]
-pub async fn get_service_handler<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn get_service_handler<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path((service_name, handler_name)): Path<(String, String)>,
 ) -> Result<Json<HandlerMetadata>, MetaApiError> {
     match state

crates/admin/src/rest_api/invocations.rs

@@ -80,8 +80,8 @@ pub struct DeleteInvocationParams {
         from_type = "MetaApiError",
     )
 )]
-pub async fn delete_invocation<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn delete_invocation<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(invocation_id): Path<String>,
     Query(DeleteInvocationParams { mode }): Query<DeleteInvocationParams>,
 ) -> Result<StatusCode, MetaApiError> {
@@ -139,8 +139,8 @@ generate_meta_api_error!(KillInvocationError: [InvocationNotFoundError, Invocati
         schema = "std::string::String"
     ))
 )]
-pub async fn kill_invocation<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn kill_invocation<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(invocation_id): Path<String>,
 ) -> Result<(), KillInvocationError>
 where
@@ -199,8 +199,8 @@ generate_meta_api_error!(CancelInvocationError: [InvocationNotFoundError, Invoca
         from_type = "CancelInvocationError",
     )
 )]
-pub async fn cancel_invocation<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn cancel_invocation<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(invocation_id): Path<String>,
 ) -> Result<StatusCode, CancelInvocationError>
 where
@@ -241,8 +241,8 @@ generate_meta_api_error!(PurgeInvocationError: [InvocationNotFoundError, Invocat
         schema = "std::string::String"
     ))
 )]
-pub async fn purge_invocation<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn purge_invocation<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(invocation_id): Path<String>,
 ) -> Result<(), PurgeInvocationError>
 where
@@ -284,8 +284,8 @@ generate_meta_api_error!(PurgeJournalError: [InvocationNotFoundError, Invocation
         schema = "std::string::String"
     ))
 )]
-pub async fn purge_journal<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn purge_journal<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(invocation_id): Path<String>,
 ) -> Result<(), PurgeJournalError>
 where
@@ -398,8 +398,8 @@ generate_meta_api_error!(RestartInvocationError: [
         ),
     )
 )]
-pub async fn restart_as_new_invocation<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn restart_as_new_invocation<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(invocation_id): Path<String>,
     Query(RestartAsNewInvocationQueryParams { from, deployment }): Query<
         RestartAsNewInvocationQueryParams,
@@ -510,8 +510,8 @@ generate_meta_api_error!(ResumeInvocationError: [
         )
     )
 )]
-pub async fn resume_invocation<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn resume_invocation<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(invocation_id): Path<String>,
     Query(ResumeInvocationQueryParams { deployment }): Query<ResumeInvocationQueryParams>,
 ) -> Result<(), ResumeInvocationError>

crates/admin/src/rest_api/mod.rs

@@ -25,16 +25,14 @@ use okapi_operation::okapi::openapi3::{ExternalDocs, Tag};
 use okapi_operation::*;
 use restate_types::identifiers::PartitionKey;
 use restate_types::invocation::client::InvocationClient;
-use restate_types::schema::subscriptions::SubscriptionValidator;
 use restate_wal_protocol::{Destination, Header, Source};
 
 use crate::state::AdminServiceState;
 
 pub use version::{MAX_ADMIN_API_VERSION, MIN_ADMIN_API_VERSION};
 
-pub fn create_router<V, IC>(state: AdminServiceState<V, IC>) -> axum::Router<()>
+pub fn create_router<IC>(state: AdminServiceState<IC>) -> axum::Router<()>
 where
-    V: SubscriptionValidator + Send + Sync + Clone + 'static,
     IC: InvocationClient + Send + Sync + Clone + 'static,
 {
     let mut router = axum_integration::Router::new()

crates/admin/src/rest_api/services.rs

@@ -35,8 +35,8 @@ use tracing::{debug, warn};
     operation_id = "list_services",
     tags = "service"
 )]
-pub async fn list_services<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn list_services<IC>(
+    State(state): State<AdminServiceState<IC>>,
 ) -> Result<Json<ListServicesResponse>, MetaApiError> {
     let services = state.schema_registry.list_services();
 
@@ -55,8 +55,8 @@ pub async fn list_services<V, IC>(
         schema = "std::string::String"
     ))
 )]
-pub async fn get_service<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn get_service<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(service_name): Path<String>,
 ) -> Result<Json<ServiceMetadata>, MetaApiError> {
     state
@@ -87,8 +87,8 @@ pub async fn get_service<V, IC>(
         from_type = "MetaApiError",
     )
 )]
-pub async fn get_service_openapi<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn get_service_openapi<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(service_name): Path<String>,
 ) -> Result<Json<serde_json::Value>, MetaApiError> {
     // TODO return correct vnd type
@@ -112,8 +112,8 @@ pub async fn get_service_openapi<V, IC>(
         schema = "std::string::String"
     ))
 )]
-pub async fn modify_service<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn modify_service<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(service_name): Path<String>,
     #[request_body(required = true)] Json(ModifyServiceRequest {
         public,
@@ -183,8 +183,8 @@ pub async fn modify_service<V, IC>(
         from_type = "MetaApiError",
     )
 )]
-pub async fn modify_service_state<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn modify_service_state<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(service_name): Path<String>,
     #[request_body(required = true)] Json(ModifyServiceStateRequest {
         version,

crates/admin/src/rest_api/subscriptions.rs

@@ -12,7 +12,7 @@ use super::error::*;
 use crate::state::AdminServiceState;
 
 use restate_admin_rest_model::subscriptions::*;
-use restate_types::schema::subscriptions::{ListSubscriptionFilter, SubscriptionValidator};
+use restate_types::schema::subscriptions::ListSubscriptionFilter;
 
 use axum::extract::Query;
 use axum::extract::{Path, State};
@@ -41,8 +41,8 @@ use restate_types::identifiers::SubscriptionId;
         from_type = "MetaApiError",
     )
 )]
-pub async fn create_subscription<V: SubscriptionValidator, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn create_subscription<IC>(
+    State(state): State<AdminServiceState<IC>>,
     #[request_body(required = true)] Json(payload): Json<CreateSubscriptionRequest>,
 ) -> Result<impl axum::response::IntoResponse, MetaApiError> {
     let subscription = state
@@ -73,8 +73,8 @@ pub async fn create_subscription<V: SubscriptionValidator, IC>(
         schema = "std::string::String"
     ))
 )]
-pub async fn get_subscription<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn get_subscription<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(subscription_id): Path<SubscriptionId>,
 ) -> Result<Json<SubscriptionResponse>, MetaApiError> {
     let subscription = state
@@ -110,8 +110,8 @@ pub async fn get_subscription<V, IC>(
         )
     )
 )]
-pub async fn list_subscriptions<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn list_subscriptions<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Query(ListSubscriptionsParams { sink, source }): Query<ListSubscriptionsParams>,
 ) -> Json<ListSubscriptionsResponse> {
     let filters = match (sink, source) {
@@ -158,8 +158,8 @@ pub async fn list_subscriptions<V, IC>(
         from_type = "MetaApiError",
     )
 )]
-pub async fn delete_subscription<V, IC>(
-    State(state): State<AdminServiceState<V, IC>>,
+pub async fn delete_subscription<IC>(
+    State(state): State<AdminServiceState<IC>>,
     Path(subscription_id): Path<SubscriptionId>,
 ) -> Result<StatusCode, MetaApiError> {
     state

crates/admin/src/schema_registry/mod.rs

@@ -31,7 +31,7 @@ use restate_types::schema::deployment::{
 };
 use restate_types::schema::service::{HandlerMetadata, ServiceMetadata, ServiceMetadataResolver};
 use restate_types::schema::subscriptions::{
-    ListSubscriptionFilter, Subscription, SubscriptionResolver, SubscriptionValidator,
+    ListSubscriptionFilter, Subscription, SubscriptionResolver,
 };
 pub(crate) use restate_types::schema::updater::RegisterDeploymentResult;
 use restate_types::schema::{Schema, updater};
@@ -119,25 +119,22 @@ pub(crate) enum UpdateDeploymentAddress {
 
 /// This is the business logic driving the Admin API schema related endpoints.
 #[derive(Clone)]
-pub struct SchemaRegistry<V> {
+pub struct SchemaRegistry {
     metadata_writer: MetadataWriter,
     service_discovery: ServiceDiscovery,
     telemetry_http_client: Option<HttpClient>,
-    subscription_validator: V,
 }
 
-impl<V> SchemaRegistry<V> {
+impl SchemaRegistry {
     pub fn new(
         metadata_writer: MetadataWriter,
         service_discovery: ServiceDiscovery,
         telemetry_http_client: Option<HttpClient>,
-        subscription_validator: V,
     ) -> Self {
         Self {
             metadata_writer,
             service_discovery,
             telemetry_http_client,
-            subscription_validator,
         }
     }
 
@@ -613,12 +610,7 @@ impl<V> SchemaRegistry<V> {
     pub fn list_subscriptions(&self, filters: &[ListSubscriptionFilter]) -> Vec<Subscription> {
         Metadata::with_current(|m| m.schema()).list_subscriptions(filters)
     }
-}
 
-impl<V> SchemaRegistry<V>
-where
-    V: SubscriptionValidator,
-{
     pub(crate) async fn create_subscription(
         &self,
         source: Uri,
@@ -641,7 +633,6 @@ where
                             source.clone(),
                             sink.clone(),
                             options.clone(),
-                            &self.subscription_validator,
                         )
                     },
                 )?;

crates/admin/src/service.rs

@@ -24,7 +24,6 @@ use restate_types::config::AdminOptions;
 use restate_types::invocation::client::InvocationClient;
 use restate_types::live::LiveLoad;
 use restate_types::net::BindAddress;
-use restate_types::schema::subscriptions::SubscriptionValidator;
 use std::time::Duration;
 use tower::ServiceBuilder;
 use tower_http::classify::ServerErrorsFailureClass;
@@ -35,26 +34,24 @@ use tracing::{Span, debug, error, info, info_span};
 #[error("could not create the service client: {0}")]
 pub struct BuildError(#[from] restate_service_client::BuildError);
 
-pub struct AdminService<V, IC> {
+pub struct AdminService<IC> {
     bifrost: Bifrost,
-    schema_registry: SchemaRegistry<V>,
+    schema_registry: SchemaRegistry,
     invocation_client: IC,
     #[cfg(feature = "storage-query")]
     query_context: Option<restate_storage_query_datafusion::context::QueryContext>,
     #[cfg(feature = "metadata-api")]
     metadata_writer: MetadataWriter,
 }
 
-impl<V, IC> AdminService<V, IC>
+impl<IC> AdminService<IC>
 where
-    V: SubscriptionValidator + Send + Sync + Clone + 'static,
     IC: InvocationClient + Send + Sync + Clone + 'static,
 {
     pub fn new(
         metadata_writer: MetadataWriter,
         bifrost: Bifrost,
         invocation_client: IC,
-        subscription_validator: V,
         service_discovery: ServiceDiscovery,
         telemetry_http_client: Option<HttpClient>,
     ) -> Self {
@@ -66,7 +63,6 @@ where
                 metadata_writer,
                 service_discovery,
                 telemetry_http_client,
-                subscription_validator,
             ),
             invocation_client,
             #[cfg(feature = "storage-query")]