Skip to content

feat(client): cancel an abandoned tools/call, prompts/get or resources/read #1344

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
DaleSeo merged 3 commits into from
Oct 16, 2026
Merged

feat(client): cancel an abandoned tools/call, prompts/get or resources/read #1344

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5a4326d
feat(client): cancel an abandoned tools/call, prompts/get or resource…
talhayme Oct 12, 2026
01bb0f0
fix(client): cancel any abandoned client request, per review
talhayme Oct 12, 2026
a64880f
fix(client): keep server/discover off cancel-on-drop; borrow the peer…
talhayme Oct 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions crates/rmcp/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -371,6 +371,11 @@ name = "test_trace_context"
required-features = ["server", "client"]
path = "tests/test_trace_context.rs"

[[test]]
name = "test_abandoned_request_cancel"
required-features = ["server", "client"]
path = "tests/test_abandoned_request_cancel.rs"

[[test]]
name = "test_mrtr_behavior"
required-features = ["server", "client"]
Expand Down
86 changes: 68 additions & 18 deletions crates/rmcp/src/service/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -448,6 +448,39 @@ impl Drop for PendingSubscriptionRequest {
}
}

/// Tells the server about a request whose caller stopped waiting for it.
///
/// Armed while the response is awaited. Dropping it armed, which happens when
/// the caller drops the future (a `tokio::time::timeout`, a lost `select!`
/// branch), sends a best-effort `notifications/cancelled` for the request, so
/// the server does not keep working on a result nobody will read.
struct AbandonedRequestGuard<'a> {
peer: &'a Peer<RoleClient>,
id: Option<RequestId>,
}

impl<'a> AbandonedRequestGuard<'a> {
const REASON: &'static str = "request abandoned by the caller";

fn new(peer: &'a Peer<RoleClient>, id: RequestId) -> Self {
Self { peer, id: Some(id) }
}

fn disarm(&mut self) {
self.id.take();
}
}

impl Drop for AbandonedRequestGuard<'_> {
fn drop(&mut self) {
let Some(id) = self.id.take() else {
return;
};
self.peer
.try_cancel_request(id, Some(Self::REASON.to_owned()));
}
}

impl Subscription {
/// Return the originating listen request ID.
pub fn id(&self) -> &RequestId {
Expand Down Expand Up @@ -1110,7 +1143,7 @@ macro_rules! method {
$(#[$meta])*
pub async fn $method(&self) -> Result<$Resp, ServiceError> {
let result = self
.send_request(ClientRequest::$Req($Req {
.send_request_cancel_on_drop(ClientRequest::$Req($Req {
method: Default::default(),
}))
.await?;
Expand All @@ -1124,7 +1157,7 @@ macro_rules! method {
$(#[$meta])*
pub async fn $method(&self, params: $Param) -> Result<$Resp, ServiceError> {
let result = self
.send_request(ClientRequest::$Req($Req {
.send_request_cancel_on_drop(ClientRequest::$Req($Req {
method: Default::default(),
params,
extensions: Default::default(),
Expand All @@ -1140,7 +1173,7 @@ macro_rules! method {
$(#[$meta])*
pub async fn $method(&self, params: Option<$Param>) -> Result<$Resp, ServiceError> {
let result = self
.send_request(ClientRequest::$Req($Req {
.send_request_cancel_on_drop(ClientRequest::$Req($Req {
method: Default::default(),
params,
extensions: Default::default(),
Expand All @@ -1156,7 +1189,7 @@ macro_rules! method {
$(#[$meta])*
pub async fn $method(&self, params: $Param) -> Result<(), ServiceError> {
let result = self
.send_request(ClientRequest::$Req($Req {
.send_request_cancel_on_drop(ClientRequest::$Req($Req {
method: Default::default(),
params,
extensions: Default::default(),
Expand Down Expand Up @@ -1374,14 +1407,29 @@ impl Peer<RoleClient> {
.await;
}

/// [`Peer::send_request`] that sends `notifications/cancelled` if dropped before
/// the response arrives. Never route `initialize` or `server/discover` through here.
async fn send_request_cancel_on_drop(
&self,
request: ClientRequest,
) -> Result<ServerResult, ServiceError> {
let handle = self
.send_request_with_option(request, PeerRequestOptions::no_options())
.await?;
let mut guard = AbandonedRequestGuard::new(self, handle.id.clone());
let result = handle.await_response().await;
guard.disarm();
result
}
Comment thread
DaleSeo marked this conversation as resolved.

/// Send one `tools/call` request and return either a final result or an MRTR
/// `InputRequiredResult` without driving any follow-up rounds.
pub async fn call_tool_once(
&self,
params: CallToolRequestParams,
) -> Result<CallToolResponse, ServiceError> {
let result = self
.send_request(ClientRequest::CallToolRequest(CallToolRequest {
.send_request_cancel_on_drop(ClientRequest::CallToolRequest(CallToolRequest {
method: Default::default(),
params,
extensions: Default::default(),
Expand All @@ -1401,7 +1449,7 @@ impl Peer<RoleClient> {
/// SEP-2663 `tasks/get`: poll the current state of a task.
pub async fn get_task(&self, params: GetTaskParams) -> Result<GetTaskResult, ServiceError> {
let result = self
.send_request(ClientRequest::GetTaskRequest(GetTaskRequest::new(params)))
.send_request_cancel_on_drop(ClientRequest::GetTaskRequest(GetTaskRequest::new(params)))
.await?;
match result {
ServerResult::GetTaskResult(result) => Ok(result),
Expand All @@ -1413,7 +1461,7 @@ impl Peer<RoleClient> {
/// input requests. The acknowledgement is eventually consistent.
pub async fn update_task(&self, params: UpdateTaskParams) -> Result<(), ServiceError> {
let result = self
.send_request(ClientRequest::UpdateTaskRequest(UpdateTaskRequest::new(
.send_request_cancel_on_drop(ClientRequest::UpdateTaskRequest(UpdateTaskRequest::new(
params,
)))
.await?;
Expand All @@ -1427,7 +1475,7 @@ impl Peer<RoleClient> {
/// is cooperative; the ack does not guarantee the task stops.
pub async fn cancel_task(&self, params: CancelTaskParams) -> Result<(), ServiceError> {
let result = self
.send_request(ClientRequest::CancelTaskRequest(CancelTaskRequest::new(
.send_request_cancel_on_drop(ClientRequest::CancelTaskRequest(CancelTaskRequest::new(
params,
)))
.await?;
Expand All @@ -1444,7 +1492,7 @@ impl Peer<RoleClient> {
params: GetPromptRequestParams,
) -> Result<GetPromptResponse, ServiceError> {
let result = self
.send_request(ClientRequest::GetPromptRequest(GetPromptRequest {
.send_request_cancel_on_drop(ClientRequest::GetPromptRequest(GetPromptRequest {
method: Default::default(),
params,
extensions: Default::default(),
Expand Down Expand Up @@ -1474,7 +1522,7 @@ impl Peer<RoleClient> {

let generation = self.capture_response_cache_generation().await;
let result = self
.send_request(ClientRequest::ReadResourceRequest(ReadResourceRequest {
.send_request_cancel_on_drop(ClientRequest::ReadResourceRequest(ReadResourceRequest {
method: Default::default(),
params,
extensions: Default::default(),
Expand Down Expand Up @@ -1547,7 +1595,7 @@ impl Peer<RoleClient> {
let generation = self.capture_response_cache_generation().await;
let uses_cursor = request_uses_cursor(&params);
let result = self
.send_request(ClientRequest::ListPromptsRequest(ListPromptsRequest {
.send_request_cancel_on_drop(ClientRequest::ListPromptsRequest(ListPromptsRequest {
method: Default::default(),
params,
extensions: Default::default(),
Expand Down Expand Up @@ -1597,11 +1645,13 @@ impl Peer<RoleClient> {
let generation = self.capture_response_cache_generation().await;
let uses_cursor = request_uses_cursor(&params);
let result = self
.send_request(ClientRequest::ListResourcesRequest(ListResourcesRequest {
method: Default::default(),
params,
extensions: Default::default(),
}))
.send_request_cancel_on_drop(ClientRequest::ListResourcesRequest(
ListResourcesRequest {
method: Default::default(),
params,
extensions: Default::default(),
},
))
.await;
let result = match result {
Ok(result) => result,
Expand Down Expand Up @@ -1648,7 +1698,7 @@ impl Peer<RoleClient> {
let generation = self.capture_response_cache_generation().await;
let uses_cursor = request_uses_cursor(&params);
let result = self
.send_request(ClientRequest::ListResourceTemplatesRequest(
.send_request_cancel_on_drop(ClientRequest::ListResourceTemplatesRequest(
ListResourceTemplatesRequest {
method: Default::default(),
params,
Expand Down Expand Up @@ -1710,7 +1760,7 @@ impl Peer<RoleClient> {
let generation = self.capture_response_cache_generation().await;
let uses_cursor = request_uses_cursor(&params);
let result = self
.send_request(ClientRequest::ListToolsRequest(ListToolsRequest {
.send_request_cancel_on_drop(ClientRequest::ListToolsRequest(ListToolsRequest {
method: Default::default(),
params,
extensions: Default::default(),
Expand Down
Loading
Loading