Skip to content

Commit ce3a57a

Browse files
talhaymeclaude
talhayme
and
claude
authored
feat(client): cancel an abandoned tools/call, prompts/get or resources/read (#1344)
* feat(client): cancel an abandoned tools/call, prompts/get or resources/read A caller that drops `call_tool` (a `tokio::time::timeout`, a lost `select!` branch) abandoned the request locally while the server kept working on it. Send a best-effort `notifications/cancelled` for the request in flight when that happens. A drop guard, modelled on `PendingSubscriptionRequest`, is armed while the response is awaited and disarmed as soon as it arrives, so dropping the future while the client handler fulfils MRTR input requests cancels nothing. Closes #1338 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(client): cancel any abandoned client request, per review Every request the client peer sends now goes through the drop guard, not only tools/call, prompts/get and resources/read. `initialize` is not sent through `Peer`, and the helper is client-only: a client must never cancel `initialize`, and from 2026-07-28 a server may send `notifications/cancelled` only to end a subscriptions/listen stream. The guard no longer unregisters a subscription that these requests never register, and the test helper fails when the call never reaches the point it is to be abandoned at instead of dropping the call anyway. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(client): keep server/discover off cancel-on-drop; borrow the peer in the guard Per review: server/discover is part of the handshake like initialize, so it goes back to plain send_request. AbandonedRequestGuard borrows the peer instead of cloning it, and the helper's doc keeps only the constraint a future editor must not break. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 98a5478 commit ce3a57a

3 files changed

Lines changed: 459 additions & 18 deletions

File tree

‎crates/rmcp/Cargo.toml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -384,6 +384,11 @@ name = "test_trace_context"
384384
required-features = ["server", "client"]
385385
path = "tests/test_trace_context.rs"
386386

387+
[[test]]
388+
name = "test_abandoned_request_cancel"
389+
required-features = ["server", "client"]
390+
path = "tests/test_abandoned_request_cancel.rs"
391+
387392
[[test]]
388393
name = "test_mrtr_behavior"
389394
required-features = ["server", "client"]

‎crates/rmcp/src/service/client.rs‎

Lines changed: 68 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -448,6 +448,39 @@ impl Drop for PendingSubscriptionRequest {
448448
}
449449
}
450450

451+
/// Tells the server about a request whose caller stopped waiting for it.
452+
///
453+
/// Armed while the response is awaited. Dropping it armed, which happens when
454+
/// the caller drops the future (a `tokio::time::timeout`, a lost `select!`
455+
/// branch), sends a best-effort `notifications/cancelled` for the request, so
456+
/// the server does not keep working on a result nobody will read.
457+
struct AbandonedRequestGuard<'a> {
458+
peer: &'a Peer<RoleClient>,
459+
id: Option<RequestId>,
460+
}
461+
462+
impl<'a> AbandonedRequestGuard<'a> {
463+
const REASON: &'static str = "request abandoned by the caller";
464+
465+
fn new(peer: &'a Peer<RoleClient>, id: RequestId) -> Self {
466+
Self { peer, id: Some(id) }
467+
}
468+
469+
fn disarm(&mut self) {
470+
self.id.take();
471+
}
472+
}
473+
474+
impl Drop for AbandonedRequestGuard<'_> {
475+
fn drop(&mut self) {
476+
let Some(id) = self.id.take() else {
477+
return;
478+
};
479+
self.peer
480+
.try_cancel_request(id, Some(Self::REASON.to_owned()));
481+
}
482+
}
483+
451484
impl Subscription {
452485
/// Return the originating listen request ID.
453486
pub fn id(&self) -> &RequestId {
@@ -1110,7 +1143,7 @@ macro_rules! method {
11101143
$(#[$meta])*
11111144
pub async fn $method(&self) -> Result<$Resp, ServiceError> {
11121145
let result = self
1113-
.send_request(ClientRequest::$Req($Req {
1146+
.send_request_cancel_on_drop(ClientRequest::$Req($Req {
11141147
method: Default::default(),
11151148
}))
11161149
.await?;
@@ -1124,7 +1157,7 @@ macro_rules! method {
11241157
$(#[$meta])*
11251158
pub async fn $method(&self, params: $Param) -> Result<$Resp, ServiceError> {
11261159
let result = self
1127-
.send_request(ClientRequest::$Req($Req {
1160+
.send_request_cancel_on_drop(ClientRequest::$Req($Req {
11281161
method: Default::default(),
11291162
params,
11301163
extensions: Default::default(),
@@ -1140,7 +1173,7 @@ macro_rules! method {
11401173
$(#[$meta])*
11411174
pub async fn $method(&self, params: Option<$Param>) -> Result<$Resp, ServiceError> {
11421175
let result = self
1143-
.send_request(ClientRequest::$Req($Req {
1176+
.send_request_cancel_on_drop(ClientRequest::$Req($Req {
11441177
method: Default::default(),
11451178
params,
11461179
extensions: Default::default(),
@@ -1156,7 +1189,7 @@ macro_rules! method {
11561189
$(#[$meta])*
11571190
pub async fn $method(&self, params: $Param) -> Result<(), ServiceError> {
11581191
let result = self
1159-
.send_request(ClientRequest::$Req($Req {
1192+
.send_request_cancel_on_drop(ClientRequest::$Req($Req {
11601193
method: Default::default(),
11611194
params,
11621195
extensions: Default::default(),
@@ -1374,14 +1407,29 @@ impl Peer<RoleClient> {
13741407
.await;
13751408
}
13761409

1410+
/// [`Peer::send_request`] that sends `notifications/cancelled` if dropped before
1411+
/// the response arrives. Never route `initialize` or `server/discover` through here.
1412+
async fn send_request_cancel_on_drop(
1413+
&self,
1414+
request: ClientRequest,
1415+
) -> Result<ServerResult, ServiceError> {
1416+
let handle = self
1417+
.send_request_with_option(request, PeerRequestOptions::no_options())
1418+
.await?;
1419+
let mut guard = AbandonedRequestGuard::new(self, handle.id.clone());
1420+
let result = handle.await_response().await;
1421+
guard.disarm();
1422+
result
1423+
}
1424+
13771425
/// Send one `tools/call` request and return either a final result or an MRTR
13781426
/// `InputRequiredResult` without driving any follow-up rounds.
13791427
pub async fn call_tool_once(
13801428
&self,
13811429
params: CallToolRequestParams,
13821430
) -> Result<CallToolResponse, ServiceError> {
13831431
let result = self
1384-
.send_request(ClientRequest::CallToolRequest(CallToolRequest {
1432+
.send_request_cancel_on_drop(ClientRequest::CallToolRequest(CallToolRequest {
13851433
method: Default::default(),
13861434
params,
13871435
extensions: Default::default(),
@@ -1401,7 +1449,7 @@ impl Peer<RoleClient> {
14011449
/// SEP-2663 `tasks/get`: poll the current state of a task.
14021450
pub async fn get_task(&self, params: GetTaskParams) -> Result<GetTaskResult, ServiceError> {
14031451
let result = self
1404-
.send_request(ClientRequest::GetTaskRequest(GetTaskRequest::new(params)))
1452+
.send_request_cancel_on_drop(ClientRequest::GetTaskRequest(GetTaskRequest::new(params)))
14051453
.await?;
14061454
match result {
14071455
ServerResult::GetTaskResult(result) => Ok(result),
@@ -1413,7 +1461,7 @@ impl Peer<RoleClient> {
14131461
/// input requests. The acknowledgement is eventually consistent.
14141462
pub async fn update_task(&self, params: UpdateTaskParams) -> Result<(), ServiceError> {
14151463
let result = self
1416-
.send_request(ClientRequest::UpdateTaskRequest(UpdateTaskRequest::new(
1464+
.send_request_cancel_on_drop(ClientRequest::UpdateTaskRequest(UpdateTaskRequest::new(
14171465
params,
14181466
)))
14191467
.await?;
@@ -1427,7 +1475,7 @@ impl Peer<RoleClient> {
14271475
/// is cooperative; the ack does not guarantee the task stops.
14281476
pub async fn cancel_task(&self, params: CancelTaskParams) -> Result<(), ServiceError> {
14291477
let result = self
1430-
.send_request(ClientRequest::CancelTaskRequest(CancelTaskRequest::new(
1478+
.send_request_cancel_on_drop(ClientRequest::CancelTaskRequest(CancelTaskRequest::new(
14311479
params,
14321480
)))
14331481
.await?;
@@ -1444,7 +1492,7 @@ impl Peer<RoleClient> {
14441492
params: GetPromptRequestParams,
14451493
) -> Result<GetPromptResponse, ServiceError> {
14461494
let result = self
1447-
.send_request(ClientRequest::GetPromptRequest(GetPromptRequest {
1495+
.send_request_cancel_on_drop(ClientRequest::GetPromptRequest(GetPromptRequest {
14481496
method: Default::default(),
14491497
params,
14501498
extensions: Default::default(),
@@ -1474,7 +1522,7 @@ impl Peer<RoleClient> {
14741522

14751523
let generation = self.capture_response_cache_generation().await;
14761524
let result = self
1477-
.send_request(ClientRequest::ReadResourceRequest(ReadResourceRequest {
1525+
.send_request_cancel_on_drop(ClientRequest::ReadResourceRequest(ReadResourceRequest {
14781526
method: Default::default(),
14791527
params,
14801528
extensions: Default::default(),
@@ -1547,7 +1595,7 @@ impl Peer<RoleClient> {
15471595
let generation = self.capture_response_cache_generation().await;
15481596
let uses_cursor = request_uses_cursor(&params);
15491597
let result = self
1550-
.send_request(ClientRequest::ListPromptsRequest(ListPromptsRequest {
1598+
.send_request_cancel_on_drop(ClientRequest::ListPromptsRequest(ListPromptsRequest {
15511599
method: Default::default(),
15521600
params,
15531601
extensions: Default::default(),
@@ -1597,11 +1645,13 @@ impl Peer<RoleClient> {
15971645
let generation = self.capture_response_cache_generation().await;
15981646
let uses_cursor = request_uses_cursor(&params);
15991647
let result = self
1600-
.send_request(ClientRequest::ListResourcesRequest(ListResourcesRequest {
1601-
method: Default::default(),
1602-
params,
1603-
extensions: Default::default(),
1604-
}))
1648+
.send_request_cancel_on_drop(ClientRequest::ListResourcesRequest(
1649+
ListResourcesRequest {
1650+
method: Default::default(),
1651+
params,
1652+
extensions: Default::default(),
1653+
},
1654+
))
16051655
.await;
16061656
let result = match result {
16071657
Ok(result) => result,
@@ -1648,7 +1698,7 @@ impl Peer<RoleClient> {
16481698
let generation = self.capture_response_cache_generation().await;
16491699
let uses_cursor = request_uses_cursor(&params);
16501700
let result = self
1651-
.send_request(ClientRequest::ListResourceTemplatesRequest(
1701+
.send_request_cancel_on_drop(ClientRequest::ListResourceTemplatesRequest(
16521702
ListResourceTemplatesRequest {
16531703
method: Default::default(),
16541704
params,
@@ -1710,7 +1760,7 @@ impl Peer<RoleClient> {
17101760
let generation = self.capture_response_cache_generation().await;
17111761
let uses_cursor = request_uses_cursor(&params);
17121762
let result = self
1713-
.send_request(ClientRequest::ListToolsRequest(ListToolsRequest {
1763+
.send_request_cancel_on_drop(ClientRequest::ListToolsRequest(ListToolsRequest {
17141764
method: Default::default(),
17151765
params,
17161766
extensions: Default::default(),

0 commit comments

Comments
 (0)