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

Two batch endpoints write the cluster-global rule book through
read_modify_write, mirroring `RuleBook::apply_changes` so a request
either commits in full or leaves the book untouched:

  * `PUT /limits/rules` — body is a list of `UpsertRuleRequest`s.
    Each entry carries a fully-specified rule body plus an optional
    `Precondition` (internally-tagged enum, defaults to `none`):
      - `{ "type": "none" }` → unconditional upsert,
      - `{ "type": "matches", "version": v }` → reject unless the
        rule's current version is `v`,
      - `{ "type": "does_not_exist" }` → strict insert.
    Returns the post-batch state of every entry so callers chaining
    `Matches(v)` get the new versions without a second read.

  * `POST /limits/rules/bulk-delete` — body is a list of
    `{ pattern, expected_version: Option<u32> }`. Missing version is
    an unconditional, idempotent delete; present version maps to
    `Precondition::Matches(v)`. Returns the patterns this batch
    actually removed (idempotent no-ops are omitted).

`Precondition` is consumed directly by the request DTOs. To make that
possible, this commit also opts the limiter's `serde` and `schema`
features into deriving Serialize/Deserialize/ToSchema on
`Precondition`, and gives `restate_types::Version` a transparent
`utoipa::ToSchema` derive (`#[schema(value_type = u32)]`) under the
existing `utoipa-schema` feature, so the OpenAPI spec sees Version as
a plain integer.

When a worker role runs in the same process, both handlers also push
the freshly written book into the local `RuleBookCache` via a
fire-and-forget observer threaded down from the node wiring, shaving
the metadata-store poll latency.

Errors collapse to three variants — `PreconditionFailed` (409),
`CapExceeded` (422), and `MetadataStore` (500).

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

Author: tillrohrmann

Cargo.lock

@@ -9,14 +9,21 @@ publish = false
 
 [features]
 default = []
-schema = ["dep:utoipa", "restate-serde-util/utoipa-schema", "restate-types/utoipa-schema"]
+schema = [
+    "dep:utoipa",
+    "restate-limiter/schema",
+    "restate-serde-util/utoipa-schema",
+    "restate-types/utoipa-schema",
+]
 
 [dependencies]
 restate-workspace-hack = { workspace = true }
 
+restate-limiter = { workspace = true, features = ["rule-book", "serde"] }
 restate-types = { workspace = true }
 restate-serde-util = { workspace = true }
 restate-time-util = { workspace = true, features = ["serde_with"] }
+restate-util-string = { workspace = true }
 
 bytes = { workspace = true }
 derive_more = { workspace = true, features = ["try_from"] }

crates/admin-rest-model/Cargo.toml

@@ -13,6 +13,7 @@ pub mod handlers;
 pub mod invocations;
 pub mod kafka_clusters;
 pub mod query;
+pub mod rules;
 pub mod services;
 pub mod subscriptions;
 pub mod version;

crates/admin-rest-model/src/lib.rs

@@ -0,0 +1,97 @@
+// 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 serde::{Deserialize, Serialize};
+use serde_with::serde_as;
+
+use restate_limiter::{PersistedRule, Precondition, RulePattern, UserLimits};
+use restate_types::Version;
+use restate_util_string::ReString;
+
+/// One entry in the body of `PUT /limits/rules`.
+///
+/// Each entry carries a fully-specified rule body plus an optional
+/// [`Precondition`]. Omitting the `precondition` field defaults to
+/// `Precondition::None` (unconditional upsert).
+#[serde_as]
+#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
+#[derive(Debug, Deserialize)]
+pub struct UpsertRuleRequest {
+    /// The pattern that selects which scope/limit-key combinations the
+    /// rule applies to. Examples: `"*"`, `"scope1/*"`, `"scope1/foo/bar"`.
+    #[cfg_attr(feature = "schema", schema(value_type = String))]
+    #[serde_as(as = "serde_with::DisplayFromStr")]
+    pub pattern: RulePattern<ReString>,
+    #[serde(default)]
+    pub limits: UserLimits,
+    /// Free-form description shown in the rule book; not consulted at
+    /// runtime.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub reason: Option<String>,
+    /// Soft-tombstone toggle. `true` parks the rule (the runtime treats
+    /// it as absent) without removing it.
+    #[serde(default)]
+    pub disabled: bool,
+    /// Optimistic-concurrency guard. `{ "type": "matches", "version": v }`
+    /// requires the rule's current version to be `v`;
+    /// `{ "type": "does_not_exist" }` requires the rule to be absent
+    /// (strict insert); `{ "type": "none" }` (or omitted) is
+    /// unconditional.
+    #[serde(default)]
+    pub precondition: Precondition,
+}
+
+/// One entry in the body of `POST /limits/rules/bulk-delete`.
+#[serde_as]
+#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
+#[derive(Debug, Deserialize)]
+pub struct DeleteRuleRequest {
+    #[cfg_attr(feature = "schema", schema(value_type = String))]
+    #[serde_as(as = "serde_with::DisplayFromStr")]
+    pub pattern: RulePattern<ReString>,
+    /// Optimistic-concurrency match. Absent → unconditional delete
+    /// (idempotent: deleting an already-absent rule succeeds as a
+    /// no-op). Present → reject unless the rule's current version is
+    /// the supplied value.
+    #[cfg_attr(feature = "schema", schema(value_type = Option<u32>))]
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub expected_version: Option<Version>,
+}
+
+#[serde_as]
+#[cfg_attr(feature = "schema", derive(utoipa::ToSchema))]
+#[derive(Debug, Serialize)]
+pub struct RuleResponse {
+    #[cfg_attr(feature = "schema", schema(value_type = String))]
+    #[serde_as(as = "serde_with::DisplayFromStr")]
+    pub pattern: RulePattern<ReString>,
+    pub limits: UserLimits,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub reason: Option<String>,
+    pub disabled: bool,
+    /// Per-rule version: bumped on runtime-relevant changes.
+    #[cfg_attr(feature = "schema", schema(value_type = u32))]
+    pub version: Version,
+    /// Seconds since UNIX epoch.
+    pub last_modified_seconds_since_epoch: u64,
+}
+
+impl From<(RulePattern<ReString>, &PersistedRule)> for RuleResponse {
+    fn from((pattern, rule): (RulePattern<ReString>, &PersistedRule)) -> Self {
+        RuleResponse {
+            pattern,
+            limits: rule.limits.clone(),
+            reason: rule.reason.clone(),
+            disabled: rule.disabled,
+            version: rule.version,
+            last_modified_seconds_since_epoch: rule.last_modified.as_unix_seconds(),
+        }
+    }
+}

crates/admin-rest-model/src/rules.rs

@@ -28,8 +28,10 @@ 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-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 }
@@ -63,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,17 +8,21 @@
 // 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 restate_util_string::RestrictedValueError;
-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.
@@ -293,18 +297,6 @@ pub enum MetaApiError {
     BadScope(RestrictedValueError),
 }
 
-/// # 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 {
@@ -369,7 +361,8 @@ impl utoipa::IntoResponses for MetaApiError {
 
 pub mod meta_api_error {
     //! Those types are only used to generate the corresponding OpenAPI specification for error types
-    //! that are referenced by [`super::MetaApiError`] when calling [`utoipa::IntoResponses`].
+    //! that are referenced by [`crate::rest_api::error::MetaApiError`] and [`crate::rest_api::rules::RulesApiError`]
+    //! when calling [`utoipa::IntoResponses`].
     #![allow(dead_code)]
 
     /// Bad request
@@ -388,6 +381,10 @@ pub mod meta_api_error {
     #[derive(utoipa::ToResponse)]
     pub struct Conflict(super::ErrorDescriptionResponse);
 
+    /// Unprocessable entity
+    #[derive(utoipa::ToResponse)]
+    pub struct UnprocessableEntity(super::ErrorDescriptionResponse);
+
     /// Internal server error
     #[derive(utoipa::ToResponse)]
     pub struct InternalServerError(super::ErrorDescriptionResponse);

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,12 +63,14 @@ 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,
         error::meta_api_error::NotFound,
         error::meta_api_error::MethodNotAllowed,
         error::meta_api_error::Conflict,
+        error::meta_api_error::UnprocessableEntity,
         error::meta_api_error::InternalServerError))
 )]
 struct AdminApiDoc;
@@ -122,6 +126,9 @@ 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::upsert_rules))
+            .routes(routes!(rules::delete_rules))
             // 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>,
+}