[Limiter] Add admin rule book endpoints and push to local cache

POST/PATCH/DELETE /rules write the cluster-global rule book through
read_modify_write. When a worker role runs in the same process, the
admin handler also pushes the new book into the local RuleBookCache
via a fire-and-forget observer threaded down from the node wiring,
shaving the metadata-store poll latency.

Also moves UpdateField<T> out of restate-limiter into restate-serde-util
since it's a generic RFC 7396 JSON Merge Patch combinator, not
rule-book-specific.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: tillrohrmann

Cargo.lock

@@ -28,8 +28,11 @@ restate-bifrost = { workspace = true, features = ["local-loglet", "replicated-lo
 restate-core = { workspace = true }
 restate-errors = { workspace = true }
 restate-ingestion-client = { workspace = true }
+restate-limiter = { workspace = true, features = ["rule-book"] }
 restate-metadata-store = { workspace = true }
+restate-util-string = { workspace = true }
 restate-metadata-providers = { workspace = true }
+restate-serde-util = { workspace = true }
 restate-service-client = { workspace = true }
 restate-service-protocol-v4 = { workspace = true, features = ["discovery", "serdes"] }
 restate-storage-query-datafusion = { workspace = true }
@@ -62,6 +65,7 @@ prost-dto = { workspace = true }
 rand = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
+serde_with = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tonic = { workspace = true, features = ["transport", "codegen", "gzip", "zstd"] }

crates/admin/Cargo.toml

@@ -17,7 +17,8 @@ use restate_core::{Metadata, my_node_id};
 use restate_types::config::Configuration;
 use restate_types::{NodeId, PlainNodeId};
 
-use crate::rest_api::error::{ErrorDescriptionResponse, GenericRestError};
+use crate::rest_api::ErrorDescriptionResponse;
+use crate::rest_api::error::GenericRestError;
 
 /// Cluster state endpoint
 #[utoipa::path(

crates/admin/src/rest_api/cluster_health.rs

@@ -8,15 +8,15 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
-use super::error::*;
-use crate::state::AdminServiceState;
 use std::time::SystemTime;
 
 use axum::extract::{Path, Query, State};
 use axum::http::{StatusCode, header};
 use axum::response::IntoResponse;
 use axum::{Extension, Json};
 use http::{Method, Uri};
+use serde::Deserialize;
+
 use restate_admin_rest_model::deployments::*;
 use restate_admin_rest_model::version::AdminApiVersion;
 use restate_errors::warn_it;
@@ -29,7 +29,10 @@ use restate_types::schema::registry::{
     Overwrite, TelemetryClient,
 };
 use restate_types::schema::service::ServiceMetadata;
-use serde::Deserialize;
+
+use super::error::*;
+use crate::rest_api::ErrorDescriptionResponse;
+use crate::state::AdminServiceState;
 
 /// Register deployment
 ///

crates/admin/src/rest_api/deployments.rs

@@ -8,16 +8,20 @@
 // the Business Source License, use of this software will be governed
 // by the Apache License, Version 2.0.
 
+use std::ops::RangeInclusive;
+
 use axum::Json;
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use codederror::{Code, CodedError};
+
 use restate_core::ShutdownError;
 use restate_types::identifiers::{DeploymentId, SubscriptionId};
 use restate_types::invocation::ServiceType;
 use restate_types::schema::registry::SchemaRegistryError;
-use serde::Serialize;
-use std::ops::RangeInclusive;
+
+use crate::rest_api::ErrorDescriptionResponse;
+
 // --- Few helpers to define Admin API errors.
 
 /// Macro to generate an Admin API Error enum with the given variants.
@@ -290,18 +294,6 @@ pub enum MetaApiError {
     DeprecatedPutDeployment,
 }
 
-/// # Error description response
-///
-/// Error details of the response
-#[derive(Debug, Serialize, utoipa::ToSchema)]
-pub(crate) struct ErrorDescriptionResponse {
-    message: String,
-    /// # Restate code
-    ///
-    /// Restate error code describing this error
-    restate_code: Option<&'static str>,
-}
-
 impl IntoResponse for MetaApiError {
     fn into_response(self) -> Response {
         let status_code = match &self {

crates/admin/src/rest_api/error.rs

@@ -18,11 +18,13 @@ mod health;
 mod invocations;
 mod kafka_clusters;
 mod query;
+mod rules;
 mod serdes;
 mod services;
 mod subscriptions;
 mod version;
 
+use serde::Serialize;
 use utoipa::OpenApi;
 use utoipa_axum::{router::OpenApiRouter, routes};
 
@@ -61,6 +63,7 @@ pub use version::{MAX_ADMIN_API_VERSION, MIN_ADMIN_API_VERSION};
         (name = "health", description = "Admin API health"),
         (name = "version", description = "API Version"),
         (name = "introspection", description = "System introspection"),
+        (name = "rule", description = "Limiter rule book management"),
     ),
     components(responses(
         error::meta_api_error::BadRequest,
@@ -122,6 +125,10 @@ where
             .routes(routes!(kafka_clusters::get_kafka_cluster))
             .routes(routes!(kafka_clusters::update_kafka_cluster))
             .routes(routes!(kafka_clusters::delete_kafka_cluster))
+            // Rule book endpoints
+            .routes(routes!(rules::create_rule))
+            .routes(routes!(rules::update_rule))
+            .routes(routes!(rules::delete_rule))
             // Query endpoint
             .routes(routes!(query::query))
     };
@@ -189,3 +196,15 @@ fn create_envelope_header(partition_key: PartitionKey) -> Header {
         },
     }
 }
+
+/// # Error description response
+///
+/// Error details of the response
+#[derive(Debug, Serialize, utoipa::ToSchema)]
+struct ErrorDescriptionResponse {
+    message: String,
+    /// # Restate code
+    ///
+    /// Restate error code describing this error
+    restate_code: Option<&'static str>,
+}