fix(adapters): bound from_mcp tools/list pagination (CON-14)
Replace rmcp's unbounded Peer::list_all_tools with a bounded walk over
list_tools: hard cap of 100 pages (MCP_MAX_TOOLS_LIST_PAGES) plus a 60 s
overall deadline (MCP_TOOLS_LIST_DEADLINE). Tripping either budget fails
loudly with AdapterError::DiscoveryFailed naming pages fetched and tools
accumulated — no silent truncation, no partial registration (import
fails closed, as before). A slow or hung page is cut off by a per-page
tokio timeout sized to the remaining budget.
Bounds are documented in the module doc. Integration test added: a
cycling-cursor server (next_cursor always Some("a")) terminates with
the clean budget error inside an outer 10 s guard; the existing 3-page
pagination test is unchanged and passes.
Verification: cargo test (304 pass), cargo test --features mcp,
cargo test --all-features (412 pass), clippy -D warnings both default
and --all-features --all-targets, cargo fmt --check, cargo doc --no-deps.
This commit is contained in:
@@ -20,6 +20,15 @@
|
|||||||
//! Repeated imports accumulate sessions — reconnecting importers should
|
//! Repeated imports accumulate sessions — reconnecting importers should
|
||||||
//! import once per process (or wait for a teardown handle in a later
|
//! import once per process (or wait for a teardown handle in a later
|
||||||
//! version).
|
//! version).
|
||||||
|
//!
|
||||||
|
//! Tool discovery is a bounded pagination loop over `tools/list` — never
|
||||||
|
//! rmcp's unbounded `Peer::list_all_tools`: at most `MCP_MAX_TOOLS_LIST_PAGES`
|
||||||
|
//! pages within `MCP_TOOLS_LIST_DEADLINE` overall. Trip either budget and
|
||||||
|
//! the import fails loudly with `AdapterError::DiscoveryFailed` naming the
|
||||||
|
//! budget (pages fetched, tools accumulated) — no silent truncation, no
|
||||||
|
//! partial registration (review-002 CON-14).
|
||||||
|
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use alkcall::client::{AdapterError, OperationAdapter};
|
use alkcall::client::{AdapterError, OperationAdapter};
|
||||||
use alkcall::core::types::{Capabilities, Secret};
|
use alkcall::core::types::{Capabilities, Secret};
|
||||||
@@ -33,7 +42,7 @@ use alkcall::registry::spec::{
|
|||||||
};
|
};
|
||||||
use rmcp::model::{
|
use rmcp::model::{
|
||||||
CallToolRequestParams, CallToolResult, ClientCapabilities, ClientInfo, Content, Implementation,
|
CallToolRequestParams, CallToolResult, ClientCapabilities, ClientInfo, Content, Implementation,
|
||||||
JsonObject, Tool,
|
JsonObject, PaginatedRequestParams, Tool,
|
||||||
};
|
};
|
||||||
use rmcp::service::RoleClient;
|
use rmcp::service::RoleClient;
|
||||||
use rmcp::transport::{
|
use rmcp::transport::{
|
||||||
@@ -45,6 +54,12 @@ use serde_json::{Map, Value};
|
|||||||
|
|
||||||
const MCP_CAPABILITY_KEY: &str = "mcp";
|
const MCP_CAPABILITY_KEY: &str = "mcp";
|
||||||
|
|
||||||
|
/// Upper bound on `tools/list` pages a single `import()` will fetch.
|
||||||
|
const MCP_MAX_TOOLS_LIST_PAGES: u32 = 100;
|
||||||
|
|
||||||
|
/// Overall time budget for the full `tools/list` pagination walk.
|
||||||
|
const MCP_TOOLS_LIST_DEADLINE: Duration = Duration::from_secs(60);
|
||||||
|
|
||||||
/// Declared error for transport-level `tools/call` failures (review-001
|
/// Declared error for transport-level `tools/call` failures (review-001
|
||||||
/// CON-11): the remote is unreachable, the transport closed, or the call
|
/// CON-11): the remote is unreachable, the transport closed, or the call
|
||||||
/// timed out. Payload is the array-of-content-blocks shape; retryable.
|
/// timed out. Payload is the array-of-content-blocks shape; retryable.
|
||||||
@@ -111,12 +126,7 @@ impl OperationAdapter for FromMCP {
|
|||||||
.map_err(|e| classify_init_error(&e))?;
|
.map_err(|e| classify_init_error(&e))?;
|
||||||
let peer: Peer<RoleClient> = running.peer().clone();
|
let peer: Peer<RoleClient> = running.peer().clone();
|
||||||
|
|
||||||
let tools = peer
|
let tools = list_all_tools_bounded(&peer).await?;
|
||||||
.list_all_tools()
|
|
||||||
.await
|
|
||||||
.map_err(|e| AdapterError::DiscoveryFailed {
|
|
||||||
message: format!("tools/list failed: {e}"),
|
|
||||||
})?;
|
|
||||||
|
|
||||||
let bundles = tools
|
let bundles = tools
|
||||||
.into_iter()
|
.into_iter()
|
||||||
@@ -128,6 +138,54 @@ impl OperationAdapter for FromMCP {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Bounded `tools/list` pagination walk (review-002 CON-14): the rmcp
|
||||||
|
/// equivalent loops `while cursor.is_some()` with no page cap and no
|
||||||
|
/// deadline, so a hostile/buggy server that never clears `next_cursor`
|
||||||
|
/// hangs `import()` while memory grows without bound. Stops after
|
||||||
|
/// [`MCP_MAX_TOOLS_LIST_PAGES`] pages or [`MCP_TOOLS_LIST_DEADLINE`]
|
||||||
|
/// elapses and fails loudly — naming pages fetched and tools accumulated —
|
||||||
|
/// rather than silently truncating.
|
||||||
|
async fn list_all_tools_bounded(peer: &Peer<RoleClient>) -> Result<Vec<Tool>, AdapterError> {
|
||||||
|
let started = Instant::now();
|
||||||
|
let mut tools = Vec::new();
|
||||||
|
let mut cursor = None;
|
||||||
|
for pages_fetched in 1..=MCP_MAX_TOOLS_LIST_PAGES {
|
||||||
|
let remaining = MCP_TOOLS_LIST_DEADLINE
|
||||||
|
.checked_sub(started.elapsed())
|
||||||
|
.ok_or_else(|| pagination_budget_error(pages_fetched, tools.len()))?;
|
||||||
|
let page = match tokio::time::timeout(
|
||||||
|
remaining,
|
||||||
|
peer.list_tools(Some(PaginatedRequestParams::default().with_cursor(cursor))),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
Ok(result) => result.map_err(|e| AdapterError::DiscoveryFailed {
|
||||||
|
message: format!("tools/list failed: {e}"),
|
||||||
|
})?,
|
||||||
|
Err(_) => return Err(pagination_budget_error(pages_fetched, tools.len())),
|
||||||
|
};
|
||||||
|
tools.extend(page.tools);
|
||||||
|
cursor = page.next_cursor;
|
||||||
|
if cursor.is_none() {
|
||||||
|
return Ok(tools);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(pagination_budget_error(
|
||||||
|
MCP_MAX_TOOLS_LIST_PAGES,
|
||||||
|
tools.len(),
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn pagination_budget_error(pages_fetched: u32, tools_accumulated: usize) -> AdapterError {
|
||||||
|
AdapterError::DiscoveryFailed {
|
||||||
|
message: format!(
|
||||||
|
"tools/list pagination exceeded budget (max {MCP_MAX_TOOLS_LIST_PAGES} pages or \
|
||||||
|
{MCP_TOOLS_LIST_DEADLINE:?} overall) after {pages_fetched} page(s) with \
|
||||||
|
{tools_accumulated} tool(s) accumulated; import failed without partial registration"
|
||||||
|
),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn classify_init_error(e: &rmcp::service::ClientInitializeError) -> AdapterError {
|
fn classify_init_error(e: &rmcp::service::ClientInitializeError) -> AdapterError {
|
||||||
use rmcp::service::ClientInitializeError as E;
|
use rmcp::service::ClientInitializeError as E;
|
||||||
match e {
|
match e {
|
||||||
|
|||||||
@@ -168,6 +168,37 @@ impl ServerHandler for PagingServer {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// A hostile paging `tools/list` server (review-002 CON-14): always
|
||||||
|
/// responds with the same single tool and `next_cursor: Some("a")` — the
|
||||||
|
/// cursor cycles and never clears. The importer must bound the walk and
|
||||||
|
/// fail with a clean `DiscoveryFailed` rather than hang forever.
|
||||||
|
struct CyclingCursorServer;
|
||||||
|
|
||||||
|
impl ServerHandler for CyclingCursorServer {
|
||||||
|
fn list_tools(
|
||||||
|
&self,
|
||||||
|
_request: Option<PaginatedRequestParams>,
|
||||||
|
_context: RequestContext<RoleServer>,
|
||||||
|
) -> impl std::future::Future<Output = Result<ListToolsResult, rmcp::ErrorData>>
|
||||||
|
+ rmcp::service::MaybeSendFuture
|
||||||
|
+ '_ {
|
||||||
|
let tools = vec![Tool::new_with_raw(
|
||||||
|
"hostile",
|
||||||
|
Some("served on every page, forever".into()),
|
||||||
|
Arc::new(serde_json::Map::new()),
|
||||||
|
)];
|
||||||
|
std::future::ready(Ok(ListToolsResult {
|
||||||
|
meta: None,
|
||||||
|
next_cursor: Some("a".to_string()),
|
||||||
|
tools,
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn get_info(&self) -> rmcp::model::ServerInfo {
|
||||||
|
rmcp::model::ServerInfo::default()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async fn spawn_server_for<S: ServerHandler + 'static>(
|
async fn spawn_server_for<S: ServerHandler + 'static>(
|
||||||
server: impl Fn() -> Result<S, std::io::Error> + Send + Sync + 'static,
|
server: impl Fn() -> Result<S, std::io::Error> + Send + Sync + 'static,
|
||||||
) -> (String, tokio::task::JoinHandle<()>) {
|
) -> (String, tokio::task::JoinHandle<()>) {
|
||||||
@@ -316,6 +347,39 @@ async fn import_follows_tools_list_pagination() {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn import_cycling_cursor_fails_bounded_with_discovery_failed() {
|
||||||
|
let (endpoint, _handle) = spawn_server_for(|| Ok(CyclingCursorServer)).await;
|
||||||
|
let adapter = FromMCP::new(endpoint, "x");
|
||||||
|
let started = Instant::now();
|
||||||
|
let result = tokio::time::timeout(Duration::from_secs(10), adapter.import()).await;
|
||||||
|
let elapsed = started.elapsed();
|
||||||
|
|
||||||
|
let outcome = result.expect("import must terminate, not hang (CON-14)");
|
||||||
|
match outcome {
|
||||||
|
Ok(_) => panic!("expected Err for a cycling-cursor server"),
|
||||||
|
Err(alkcall::client::AdapterError::DiscoveryFailed { message }) => {
|
||||||
|
assert!(
|
||||||
|
message.contains("pagination exceeded budget"),
|
||||||
|
"error must name the budget, got: {message}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
message.contains("page"),
|
||||||
|
"error must name pages fetched, got: {message}"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
message.contains("tool"),
|
||||||
|
"error must name tools accumulated, got: {message}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Err(other) => panic!("expected DiscoveryFailed, got {other}"),
|
||||||
|
}
|
||||||
|
assert!(
|
||||||
|
elapsed < Duration::from_secs(10),
|
||||||
|
"bounded walk must trip the page cap well inside the outer guard, took {elapsed:?}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn import_refuses_tool_name_containing_slash() {
|
async fn import_refuses_tool_name_containing_slash() {
|
||||||
struct SlashToolServer;
|
struct SlashToolServer;
|
||||||
|
|||||||
Reference in New Issue
Block a user