Merge branch 'wt/review-002-mcp-batch-cap'
This commit is contained in:
+80
-3
@@ -46,7 +46,7 @@ use rmcp::transport::{
|
|||||||
};
|
};
|
||||||
use serde_json::{Map, Value};
|
use serde_json::{Map, Value};
|
||||||
|
|
||||||
use crate::gateway::GatewayDispatch;
|
use crate::gateway::{GatewayDispatch, MAX_BATCH_OPERATIONS};
|
||||||
|
|
||||||
const TOOL_SEARCH: &str = "search";
|
const TOOL_SEARCH: &str = "search";
|
||||||
const TOOL_SCHEMA: &str = "schema";
|
const TOOL_SCHEMA: &str = "schema";
|
||||||
@@ -112,7 +112,8 @@ fn batch_input_schema() -> Value {
|
|||||||
},
|
},
|
||||||
"required": ["operation"]
|
"required": ["operation"]
|
||||||
},
|
},
|
||||||
"description": "The operations to invoke in this batch."
|
"maxItems": MAX_BATCH_OPERATIONS,
|
||||||
|
"description": "The operations to invoke in this batch. At most 100 operations per batch.",
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
"required": ["calls"]
|
"required": ["calls"]
|
||||||
@@ -223,6 +224,11 @@ impl ToMcpGateway {
|
|||||||
));
|
));
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
if calls.len() > MAX_BATCH_OPERATIONS {
|
||||||
|
return call_error_to_structured_error(CallError::invalid_input(format!(
|
||||||
|
"batch exceeds the maximum of {MAX_BATCH_OPERATIONS} operations"
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
|
||||||
let mut results: Vec<Value> = Vec::with_capacity(calls.len());
|
let mut results: Vec<Value> = Vec::with_capacity(calls.len());
|
||||||
for call in calls {
|
for call in calls {
|
||||||
@@ -412,7 +418,7 @@ pub(crate) fn gateway_tools() -> Vec<Tool> {
|
|||||||
Tool::new(
|
Tool::new(
|
||||||
Cow::Borrowed(TOOL_BATCH),
|
Cow::Borrowed(TOOL_BATCH),
|
||||||
Cow::Borrowed(
|
Cow::Borrowed(
|
||||||
"Invoke multiple operations in one tool call. Returns {\"results\": [...]} where each entry is {\"isError\": false, \"output\": ...} on success or {\"isError\": true, \"error\": {code, message, retryable}} on failure.",
|
"Invoke multiple operations in one tool call, executed serially in order (each bounded by the 30 s deadline). Returns {\"results\": [...]} where each entry is {\"isError\": false, \"output\": ...} on success or {\"isError\": true, \"error\": {code, message, retryable}} on failure. At most 100 operations per batch; larger batches are rejected with INVALID_INPUT without dispatching.",
|
||||||
),
|
),
|
||||||
value_to_object(batch_input_schema()),
|
value_to_object(batch_input_schema()),
|
||||||
),
|
),
|
||||||
@@ -1171,6 +1177,77 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn batch_over_cap_returns_invalid_input_without_dispatching() {
|
||||||
|
let registry = full_registry_with_ops(vec![(
|
||||||
|
"echo/run".to_string(),
|
||||||
|
OperationType::Query,
|
||||||
|
AccessControl::default(),
|
||||||
|
)]);
|
||||||
|
let calls: Vec<Value> = (0..MAX_BATCH_OPERATIONS + 1)
|
||||||
|
.map(|i| serde_json::json!({ "operation": "echo/run", "input": { "n": i } }))
|
||||||
|
.collect();
|
||||||
|
let mut args = Map::new();
|
||||||
|
args.insert("calls".to_string(), Value::Array(calls));
|
||||||
|
let dispatch_spine = dispatch(registry, provider());
|
||||||
|
let gateway = ToMcpGateway::new(Arc::clone(&dispatch_spine));
|
||||||
|
|
||||||
|
let result = invoke_tool(&gateway, "batch", Some(args), None).await;
|
||||||
|
assert_eq!(result.is_error, Some(true));
|
||||||
|
let structured = result.structured_content.expect("structured error present");
|
||||||
|
assert_eq!(
|
||||||
|
structured.get("code"),
|
||||||
|
Some(&Value::String("INVALID_INPUT".to_string()))
|
||||||
|
);
|
||||||
|
assert_eq!(structured.get("retryable"), Some(&Value::Bool(false)));
|
||||||
|
let message = structured
|
||||||
|
.get("message")
|
||||||
|
.and_then(Value::as_str)
|
||||||
|
.expect("message");
|
||||||
|
assert!(
|
||||||
|
message.contains("maximum of 100"),
|
||||||
|
"the cap message must state the limit: {message}"
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
dispatch_spine.invoke_count(),
|
||||||
|
0,
|
||||||
|
"no operation may be dispatched when the batch is over cap"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn batch_at_cap_is_accepted() {
|
||||||
|
let registry = full_registry_with_ops(vec![(
|
||||||
|
"echo/run".to_string(),
|
||||||
|
OperationType::Query,
|
||||||
|
AccessControl::default(),
|
||||||
|
)]);
|
||||||
|
let calls: Vec<Value> = (0..MAX_BATCH_OPERATIONS)
|
||||||
|
.map(|i| serde_json::json!({ "operation": "echo/run", "input": { "n": i } }))
|
||||||
|
.collect();
|
||||||
|
let mut args = Map::new();
|
||||||
|
args.insert("calls".to_string(), Value::Array(calls));
|
||||||
|
let dispatch_spine = dispatch(registry, provider());
|
||||||
|
let gateway = ToMcpGateway::new(Arc::clone(&dispatch_spine));
|
||||||
|
|
||||||
|
let result = invoke_tool(&gateway, "batch", Some(args), None).await;
|
||||||
|
assert_eq!(result.is_error, Some(false));
|
||||||
|
let structured = result.structured_content.expect("structured present");
|
||||||
|
let results = structured
|
||||||
|
.get("results")
|
||||||
|
.and_then(Value::as_array)
|
||||||
|
.expect("results array");
|
||||||
|
assert_eq!(results.len(), MAX_BATCH_OPERATIONS);
|
||||||
|
for entry in results {
|
||||||
|
assert_eq!(entry.get("isError"), Some(&Value::Bool(false)));
|
||||||
|
}
|
||||||
|
assert_eq!(
|
||||||
|
dispatch_spine.invoke_count(),
|
||||||
|
MAX_BATCH_OPERATIONS,
|
||||||
|
"every entry in an at-cap batch is dispatched"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn batch_returns_object_with_result_entries() {
|
async fn batch_returns_object_with_result_entries() {
|
||||||
let registry = full_registry_with_ops(vec![(
|
let registry = full_registry_with_ops(vec![(
|
||||||
|
|||||||
@@ -60,6 +60,7 @@ use alkcall::registry::registration::OperationRegistry;
|
|||||||
use alkcall::registry::spec::ErrorDefinition;
|
use alkcall::registry::spec::ErrorDefinition;
|
||||||
|
|
||||||
use super::openapi_spec::OpenAPISpec;
|
use super::openapi_spec::OpenAPISpec;
|
||||||
|
use crate::gateway::MAX_BATCH_OPERATIONS;
|
||||||
|
|
||||||
const GATEWAY_VERSION: &str = "1.2.0";
|
const GATEWAY_VERSION: &str = "1.2.0";
|
||||||
const GATEWAY_TITLE: &str = "alk gateway";
|
const GATEWAY_TITLE: &str = "alk gateway";
|
||||||
@@ -93,8 +94,6 @@ const SCHEME_BEARER: &str = "bearerAuth";
|
|||||||
|
|
||||||
const RESPONSE_REF: &str = "#/components/schemas/";
|
const RESPONSE_REF: &str = "#/components/schemas/";
|
||||||
|
|
||||||
const MAX_BATCH_OPERATIONS: usize = 100;
|
|
||||||
|
|
||||||
/// Project the registry into the fixed 6-endpoint gateway doc (ADR-042).
|
/// Project the registry into the fixed 6-endpoint gateway doc (ADR-042).
|
||||||
///
|
///
|
||||||
/// Returns [`AdapterError::SchemaParse`] if the generated doc does not
|
/// Returns [`AdapterError::SchemaParse`] if the generated doc does not
|
||||||
|
|||||||
@@ -30,6 +30,7 @@
|
|||||||
//! fixed window).
|
//! fixed window).
|
||||||
|
|
||||||
use std::collections::HashMap;
|
use std::collections::HashMap;
|
||||||
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
@@ -50,6 +51,7 @@ const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
|
|||||||
pub struct GatewayDispatch {
|
pub struct GatewayDispatch {
|
||||||
registry: Arc<OperationRegistry>,
|
registry: Arc<OperationRegistry>,
|
||||||
identity_provider: Arc<dyn IdentityProvider>,
|
identity_provider: Arc<dyn IdentityProvider>,
|
||||||
|
invoke_count: AtomicUsize,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl GatewayDispatch {
|
impl GatewayDispatch {
|
||||||
@@ -62,6 +64,7 @@ impl GatewayDispatch {
|
|||||||
Self {
|
Self {
|
||||||
registry,
|
registry,
|
||||||
identity_provider,
|
identity_provider,
|
||||||
|
invoke_count: AtomicUsize::new(0),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -75,6 +78,14 @@ impl GatewayDispatch {
|
|||||||
&self.identity_provider
|
&self.identity_provider
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// How many [`GatewayDispatch::invoke`] calls this spine has
|
||||||
|
/// served. A test-spy accessor: the over-cap batch tests assert it
|
||||||
|
/// stays at zero to prove no dispatch happened before the cap
|
||||||
|
/// rejection (review-002 PRJ-22).
|
||||||
|
pub fn invoke_count(&self) -> usize {
|
||||||
|
self.invoke_count.load(Ordering::Relaxed)
|
||||||
|
}
|
||||||
|
|
||||||
/// Resolve a bearer token to an identity (the auth-middleware hook).
|
/// Resolve a bearer token to an identity (the auth-middleware hook).
|
||||||
pub fn resolve_bearer(&self, token: &AuthToken) -> Option<Identity> {
|
pub fn resolve_bearer(&self, token: &AuthToken) -> Option<Identity> {
|
||||||
self.identity_provider.resolve_from_token(token)
|
self.identity_provider.resolve_from_token(token)
|
||||||
@@ -88,6 +99,7 @@ impl GatewayDispatch {
|
|||||||
op: &str,
|
op: &str,
|
||||||
input: Value,
|
input: Value,
|
||||||
) -> ResponseEnvelope {
|
) -> ResponseEnvelope {
|
||||||
|
self.invoke_count.fetch_add(1, Ordering::Relaxed);
|
||||||
let operation_name = strip_leading_slash(op).to_string();
|
let operation_name = strip_leading_slash(op).to_string();
|
||||||
let request_id = uuid::Uuid::new_v4().to_string();
|
let request_id = uuid::Uuid::new_v4().to_string();
|
||||||
let context = self.build_root_context(&request_id, &operation_name, identity);
|
let context = self.build_root_context(&request_id, &operation_name, identity);
|
||||||
|
|||||||
@@ -5,3 +5,10 @@ pub(crate) mod schema_cache;
|
|||||||
|
|
||||||
pub use dispatch::GatewayDispatch;
|
pub use dispatch::GatewayDispatch;
|
||||||
pub use routes::CallRequest;
|
pub use routes::CallRequest;
|
||||||
|
|
||||||
|
/// The maximum number of operations accepted in a single batch,
|
||||||
|
/// enforced identically by the HTTP `POST /batch` endpoint, the
|
||||||
|
/// `to_openapi` gateway-spec projection, and the MCP `batch` tool
|
||||||
|
/// (review-002 PRJ-22). One shared constant so the surfaces cannot
|
||||||
|
/// drift.
|
||||||
|
pub const MAX_BATCH_OPERATIONS: usize = 100;
|
||||||
|
|||||||
@@ -45,12 +45,12 @@ use serde_json::{json, Value};
|
|||||||
use super::dispatch::GatewayDispatch;
|
use super::dispatch::GatewayDispatch;
|
||||||
use super::error::call_error_to_http_response_with_identity;
|
use super::error::call_error_to_http_response_with_identity;
|
||||||
use super::schema_cache::{CompileFailed, PublishSchemaCache};
|
use super::schema_cache::{CompileFailed, PublishSchemaCache};
|
||||||
|
use super::MAX_BATCH_OPERATIONS;
|
||||||
use crate::server::auth::ResolvedIdentity;
|
use crate::server::auth::ResolvedIdentity;
|
||||||
use crate::server::state::RouterState;
|
use crate::server::state::RouterState;
|
||||||
|
|
||||||
const SERVICES_LIST: &str = "services/list";
|
const SERVICES_LIST: &str = "services/list";
|
||||||
const SERVICES_SCHEMA: &str = "services/schema";
|
const SERVICES_SCHEMA: &str = "services/schema";
|
||||||
const MAX_BATCH_OPERATIONS: usize = 100;
|
|
||||||
const MAX_PUBLISH_LINE_BYTES: usize = 2 * 1024 * 1024;
|
const MAX_PUBLISH_LINE_BYTES: usize = 2 * 1024 * 1024;
|
||||||
|
|
||||||
/// The explicit request-body limit for the whole gateway router
|
/// The explicit request-body limit for the whole gateway router
|
||||||
|
|||||||
Reference in New Issue
Block a user