1143 lines
42 KiB
Rust
1143 lines
42 KiB
Rust
//! Independent http POSTs may overlap. A POST that reports an expired session
|
|
//! is retried at most once.
|
|
#![cfg(not(feature = "local"))]
|
|
|
|
use std::{
|
|
collections::HashMap,
|
|
io,
|
|
sync::{
|
|
Arc, Mutex as StdMutex,
|
|
atomic::{AtomicBool, AtomicUsize, Ordering::SeqCst},
|
|
},
|
|
time::Duration,
|
|
};
|
|
|
|
use futures::{StreamExt, stream::BoxStream};
|
|
use http::{HeaderName, HeaderValue};
|
|
use rmcp::{
|
|
model::{
|
|
CallToolRequestParams, CancelledNotificationParam, ClientConfig, ClientJsonRpcMessage,
|
|
ClientRequest, DiscoverResult, InitializeRequestParams, ProtocolVersion, Request,
|
|
RequestId, RequestMetaObject, ServerJsonRpcMessage,
|
|
},
|
|
service::{
|
|
ClientLifecycleMode, PeerRequestOptions, RequestHandle, RoleClient, RunningService,
|
|
serve_client_with_lifecycle,
|
|
},
|
|
transport::streamable_http_client::{
|
|
StreamableHttpClient, StreamableHttpClientTransport, StreamableHttpClientTransportConfig,
|
|
StreamableHttpError, StreamableHttpPostResponse,
|
|
},
|
|
};
|
|
use serde_json::{Value, json};
|
|
use sse_stream::{Error as SseError, Sse};
|
|
use tokio::{
|
|
sync::{Mutex, mpsc, oneshot},
|
|
task::JoinHandle,
|
|
time::timeout,
|
|
};
|
|
use tokio_stream::wrappers::UnboundedReceiverStream;
|
|
use tokio_util::sync::CancellationToken;
|
|
|
|
const TEST_TIMEOUT: Duration = Duration::from_secs(5);
|
|
type PostResult = Result<StreamableHttpPostResponse, StreamableHttpError<io::Error>>;
|
|
type Call = JoinHandle<anyhow::Result<()>>;
|
|
type SseReceiver = mpsc::UnboundedReceiver<Result<Sse, SseError>>;
|
|
|
|
#[derive(Default)]
|
|
struct Counts {
|
|
initialized: AtomicUsize,
|
|
hold_reinitialization: AtomicBool,
|
|
manual_controls: AtomicBool,
|
|
reject_unmatched_cancellation: AtomicBool,
|
|
local_streams: StdMutex<HashMap<RequestId, CancellationToken>>,
|
|
local_cancellations: AtomicUsize,
|
|
deleted: AtomicUsize,
|
|
cancelled: AtomicUsize,
|
|
posted: AtomicUsize,
|
|
active: AtomicUsize,
|
|
peak: AtomicUsize,
|
|
}
|
|
|
|
struct ActivePost(Arc<Counts>);
|
|
|
|
impl Drop for ActivePost {
|
|
fn drop(&mut self) {
|
|
self.0.active.fetch_sub(1, SeqCst);
|
|
}
|
|
}
|
|
|
|
struct LocalStreamRegistration {
|
|
id: RequestId,
|
|
counts: Arc<Counts>,
|
|
dropped: Option<oneshot::Sender<()>>,
|
|
}
|
|
|
|
impl Drop for LocalStreamRegistration {
|
|
fn drop(&mut self) {
|
|
self.counts.local_streams.lock().unwrap().remove(&self.id);
|
|
let _ = self.dropped.take().unwrap().send(());
|
|
}
|
|
}
|
|
|
|
struct Posted {
|
|
id: Value,
|
|
name: String,
|
|
session: Option<Arc<str>>,
|
|
reply: oneshot::Sender<PostResult>,
|
|
returned: oneshot::Receiver<()>,
|
|
}
|
|
|
|
struct ControlPost {
|
|
message: Value,
|
|
session: Option<Arc<str>>,
|
|
reply: oneshot::Sender<PostResult>,
|
|
}
|
|
|
|
fn response(id: Value, result: Value) -> ServerJsonRpcMessage {
|
|
serde_json::from_value(json!({ "jsonrpc": "2.0", "id": id, "result": result }))
|
|
.expect("valid scripted response")
|
|
}
|
|
|
|
fn sse(message: Value) -> Result<Sse, SseError> {
|
|
Ok(Sse {
|
|
event: Some("message".into()),
|
|
data: Some(message.to_string()),
|
|
id: None,
|
|
retry: None,
|
|
})
|
|
}
|
|
|
|
async fn next_event<T>(receiver: &mut mpsc::UnboundedReceiver<T>) -> T {
|
|
timeout(TEST_TIMEOUT, receiver.recv())
|
|
.await
|
|
.expect("expected scripted event")
|
|
.expect("scripted client remains connected")
|
|
}
|
|
|
|
impl Posted {
|
|
fn result(&self) -> ServerJsonRpcMessage {
|
|
response(
|
|
self.id.clone(),
|
|
json!({ "content": [{ "type": "text", "text": self.name }] }),
|
|
)
|
|
}
|
|
|
|
fn finish(self, result: PostResult) {
|
|
self.reply.send(result).expect("POST is still waiting");
|
|
}
|
|
|
|
fn succeed(self) {
|
|
let result = StreamableHttpPostResponse::Json(self.result(), None);
|
|
self.finish(Ok(result));
|
|
}
|
|
|
|
fn expire(self) {
|
|
self.finish(Err(StreamableHttpError::SessionExpired));
|
|
}
|
|
|
|
async fn finish_and_wait(self, result: PostResult) -> anyhow::Result<()> {
|
|
let Self {
|
|
reply, returned, ..
|
|
} = self;
|
|
reply.send(result).expect("POST is still waiting");
|
|
timeout(TEST_TIMEOUT, returned).await??;
|
|
Ok(())
|
|
}
|
|
|
|
async fn expire_and_wait(self) -> anyhow::Result<()> {
|
|
self.finish_and_wait(Err(StreamableHttpError::SessionExpired))
|
|
.await
|
|
}
|
|
|
|
async fn start_sse(self) -> anyhow::Result<oneshot::Sender<()>> {
|
|
let message = serde_json::to_value(self.result()).unwrap();
|
|
let (release, released) = oneshot::channel();
|
|
let stream = futures::stream::once(async move {
|
|
released.await.expect("release the SSE response");
|
|
sse(message)
|
|
})
|
|
.boxed();
|
|
self.finish_and_wait(Ok(StreamableHttpPostResponse::Sse(stream, None)))
|
|
.await?;
|
|
Ok(release)
|
|
}
|
|
|
|
async fn start_local_sse(
|
|
self,
|
|
counts: Arc<Counts>,
|
|
) -> anyhow::Result<(CancellationToken, oneshot::Receiver<()>)> {
|
|
let id: RequestId = serde_json::from_value(self.id.clone())?;
|
|
let cancellation = CancellationToken::new();
|
|
let eof = cancellation.clone();
|
|
counts
|
|
.local_streams
|
|
.lock()
|
|
.unwrap()
|
|
.insert(id.clone(), cancellation.clone());
|
|
let (dropped, closed) = oneshot::channel();
|
|
let registration = LocalStreamRegistration {
|
|
id,
|
|
counts,
|
|
dropped: Some(dropped),
|
|
};
|
|
let (started, listening) = oneshot::channel();
|
|
let stream = futures::stream::once(async move {
|
|
let _registration = registration;
|
|
let _ = started.send(());
|
|
cancellation.cancelled().await;
|
|
None::<Result<Sse, SseError>>
|
|
})
|
|
.filter_map(futures::future::ready)
|
|
.boxed();
|
|
self.finish_and_wait(Ok(StreamableHttpPostResponse::Sse(stream, None)))
|
|
.await?;
|
|
timeout(TEST_TIMEOUT, listening).await??;
|
|
Ok((eof, closed))
|
|
}
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct ScriptedClient {
|
|
started: mpsc::UnboundedSender<Posted>,
|
|
controls: mpsc::UnboundedSender<ControlPost>,
|
|
incoming: Arc<Mutex<Option<SseReceiver>>>,
|
|
reinitializing: mpsc::UnboundedSender<oneshot::Sender<()>>,
|
|
counts: Arc<Counts>,
|
|
}
|
|
|
|
impl ScriptedClient {
|
|
async fn control_post(&self, message: Value, session: Option<Arc<str>>) -> PostResult {
|
|
if !self.counts.manual_controls.load(SeqCst) {
|
|
return Ok(StreamableHttpPostResponse::Accepted);
|
|
}
|
|
let (reply, result) = oneshot::channel();
|
|
self.controls
|
|
.send(ControlPost {
|
|
message,
|
|
session,
|
|
reply,
|
|
})
|
|
.expect("test remains connected");
|
|
result.await.expect("test answers the control POST")
|
|
}
|
|
}
|
|
|
|
impl StreamableHttpClient for ScriptedClient {
|
|
type Error = io::Error;
|
|
|
|
async fn post_message(
|
|
&self,
|
|
_uri: Arc<str>,
|
|
message: ClientJsonRpcMessage,
|
|
session: Option<Arc<str>>,
|
|
_auth_header: Option<String>,
|
|
_custom_headers: HashMap<HeaderName, HeaderValue>,
|
|
) -> PostResult {
|
|
let value = serde_json::to_value(message).unwrap();
|
|
match value["method"].as_str() {
|
|
Some("server/discover") => Ok(StreamableHttpPostResponse::Json(
|
|
response(
|
|
value["id"].clone(),
|
|
serde_json::to_value(DiscoverResult::new(
|
|
vec![ProtocolVersion::V_2026_07_28],
|
|
serde_json::from_value(json!({ "tools": {} })).unwrap(),
|
|
))
|
|
.unwrap(),
|
|
),
|
|
None,
|
|
)),
|
|
Some("initialize") => {
|
|
let generation = self.counts.initialized.fetch_add(1, SeqCst) + 1;
|
|
if generation > 1 && self.counts.hold_reinitialization.load(SeqCst) {
|
|
let (release, released) = oneshot::channel();
|
|
self.reinitializing
|
|
.send(release)
|
|
.expect("test remains connected");
|
|
released.await.expect("test releases reinitialization");
|
|
}
|
|
Ok(StreamableHttpPostResponse::Json(
|
|
response(
|
|
value["id"].clone(),
|
|
json!({
|
|
"protocolVersion": "2025-11-25",
|
|
"capabilities": { "tools": {} },
|
|
"serverInfo": { "name": "scripted", "version": "1" },
|
|
}),
|
|
),
|
|
Some(format!("session-{generation}")),
|
|
))
|
|
}
|
|
Some("notifications/initialized") => Ok(StreamableHttpPostResponse::Accepted),
|
|
Some("notifications/cancelled") => {
|
|
let id: RequestId =
|
|
serde_json::from_value(value["params"]["requestId"].clone()).unwrap();
|
|
let local = self.counts.local_streams.lock().unwrap().get(&id).cloned();
|
|
if let Some(cancellation) = local {
|
|
self.counts.local_cancellations.fetch_add(1, SeqCst);
|
|
cancellation.cancel();
|
|
return Ok(StreamableHttpPostResponse::Accepted);
|
|
}
|
|
self.counts.cancelled.fetch_add(1, SeqCst);
|
|
if self.counts.reject_unmatched_cancellation.load(SeqCst) {
|
|
return Err(StreamableHttpError::Client(io::Error::other(
|
|
"local event cancellation reached the http POST path",
|
|
)));
|
|
}
|
|
self.control_post(value, session).await
|
|
}
|
|
Some("tools/call") => {
|
|
self.counts.posted.fetch_add(1, SeqCst);
|
|
let active = self.counts.active.fetch_add(1, SeqCst) + 1;
|
|
self.counts.peak.fetch_max(active, SeqCst);
|
|
let _active = ActivePost(self.counts.clone());
|
|
let (reply, response) = oneshot::channel();
|
|
let (finished, returned) = oneshot::channel();
|
|
self.started
|
|
.send(Posted {
|
|
id: value["id"].clone(),
|
|
name: value["params"]["name"].as_str().unwrap().to_owned(),
|
|
session,
|
|
reply,
|
|
returned,
|
|
})
|
|
.expect("test remains connected");
|
|
let response = response.await.expect("test answers each POST");
|
|
let _ = finished.send(());
|
|
response
|
|
}
|
|
None if value.get("result").is_some() || value.get("error").is_some() => {
|
|
self.control_post(value, session).await
|
|
}
|
|
method => panic!("unexpected scripted method: {method:?}"),
|
|
}
|
|
}
|
|
|
|
async fn delete_session(
|
|
&self,
|
|
_uri: Arc<str>,
|
|
_session: Arc<str>,
|
|
_auth_header: Option<String>,
|
|
_custom_headers: HashMap<HeaderName, HeaderValue>,
|
|
) -> Result<(), StreamableHttpError<Self::Error>> {
|
|
assert_eq!(
|
|
self.counts.active.load(SeqCst),
|
|
0,
|
|
"POSTs must stop before deleting the session"
|
|
);
|
|
self.counts.deleted.fetch_add(1, SeqCst);
|
|
Ok(())
|
|
}
|
|
|
|
async fn get_stream(
|
|
&self,
|
|
_uri: Arc<str>,
|
|
_session: Option<Arc<str>>,
|
|
_last_event_id: Option<String>,
|
|
_auth_header: Option<String>,
|
|
_custom_headers: HashMap<HeaderName, HeaderValue>,
|
|
) -> Result<BoxStream<'static, Result<Sse, SseError>>, StreamableHttpError<Self::Error>> {
|
|
Ok(match self.incoming.lock().await.take() {
|
|
Some(incoming) => UnboundedReceiverStream::new(incoming).boxed(),
|
|
None => futures::stream::pending().boxed(),
|
|
})
|
|
}
|
|
}
|
|
|
|
struct Harness {
|
|
client: RunningService<RoleClient, ClientConfig>,
|
|
started: mpsc::UnboundedReceiver<Posted>,
|
|
controls: mpsc::UnboundedReceiver<ControlPost>,
|
|
incoming: mpsc::UnboundedSender<Result<Sse, SseError>>,
|
|
reinitializations: mpsc::UnboundedReceiver<oneshot::Sender<()>>,
|
|
counts: Arc<Counts>,
|
|
}
|
|
|
|
fn config() -> StreamableHttpClientTransportConfig {
|
|
StreamableHttpClientTransportConfig::with_uri("http://scripted/mcp")
|
|
}
|
|
|
|
fn transport_error(error: &anyhow::Error) -> &StreamableHttpError<io::Error> {
|
|
let service_error = error
|
|
.downcast_ref::<rmcp::ServiceError>()
|
|
.expect("expected a service error");
|
|
let rmcp::ServiceError::TransportSend(transport_error) = service_error else {
|
|
panic!("expected a transport error, got {service_error:?}");
|
|
};
|
|
transport_error
|
|
.error
|
|
.downcast_ref::<StreamableHttpError<io::Error>>()
|
|
.expect("expected a streamable http error")
|
|
}
|
|
|
|
fn assert_recovery_timeout(error: anyhow::Error) {
|
|
assert!(matches!(
|
|
transport_error(&error),
|
|
StreamableHttpError::SessionRecoveryTimeout
|
|
));
|
|
}
|
|
|
|
impl Harness {
|
|
async fn start(config: StreamableHttpClientTransportConfig) -> anyhow::Result<Self> {
|
|
Self::with_lifecycle(config, ClientLifecycleMode::Initialize).await
|
|
}
|
|
|
|
async fn with_lifecycle(
|
|
config: StreamableHttpClientTransportConfig,
|
|
lifecycle: ClientLifecycleMode,
|
|
) -> anyhow::Result<Self> {
|
|
let (started, requests) = mpsc::unbounded_channel();
|
|
let (control_tx, controls) = mpsc::unbounded_channel();
|
|
let (incoming, incoming_rx) = mpsc::unbounded_channel();
|
|
let (reinitializing, reinitializations) = mpsc::unbounded_channel();
|
|
let counts = Arc::new(Counts::default());
|
|
let transport = StreamableHttpClientTransport::with_client(
|
|
ScriptedClient {
|
|
started,
|
|
controls: control_tx,
|
|
incoming: Arc::new(Mutex::new(Some(incoming_rx))),
|
|
reinitializing,
|
|
counts: counts.clone(),
|
|
},
|
|
config,
|
|
);
|
|
let client =
|
|
serve_client_with_lifecycle(ClientConfig::default(), transport, lifecycle).await?;
|
|
Ok(Self {
|
|
client,
|
|
started: requests,
|
|
controls,
|
|
incoming,
|
|
reinitializations,
|
|
counts,
|
|
})
|
|
}
|
|
|
|
fn call(&self, name: impl Into<String>) -> Call {
|
|
let name = name.into();
|
|
let peer = self.client.peer().clone();
|
|
tokio::spawn(async move {
|
|
let result = peer
|
|
.call_tool(CallToolRequestParams::new(name.clone()))
|
|
.await?;
|
|
anyhow::ensure!(serde_json::to_value(result)?["content"][0]["text"] == name);
|
|
Ok(())
|
|
})
|
|
}
|
|
|
|
async fn cancellable(&self, name: &'static str) -> anyhow::Result<RequestHandle<RoleClient>> {
|
|
self.cancellable_with_options(name, PeerRequestOptions::no_options())
|
|
.await
|
|
}
|
|
|
|
async fn cancellable_with_options(
|
|
&self,
|
|
name: &'static str,
|
|
options: PeerRequestOptions,
|
|
) -> anyhow::Result<RequestHandle<RoleClient>> {
|
|
Ok(self
|
|
.client
|
|
.peer()
|
|
.send_cancellable_request(
|
|
ClientRequest::CallToolRequest(Request::new(CallToolRequestParams::new(name))),
|
|
options,
|
|
)
|
|
.await?)
|
|
}
|
|
|
|
fn notify_cancellation(&self, id: RequestId) -> JoinHandle<Result<(), rmcp::ServiceError>> {
|
|
let peer = self.client.peer().clone();
|
|
tokio::spawn(async move {
|
|
peer.notify_cancelled(CancelledNotificationParam::new(Some(id), None))
|
|
.await
|
|
})
|
|
}
|
|
|
|
async fn next(&mut self) -> Posted {
|
|
next_event(&mut self.started).await
|
|
}
|
|
|
|
async fn next_control(&mut self) -> ControlPost {
|
|
next_event(&mut self.controls).await
|
|
}
|
|
|
|
async fn exchange_ping(&mut self, id: &str) {
|
|
self.incoming
|
|
.send(sse(json!({ "jsonrpc": "2.0", "id": id, "method": "ping" })))
|
|
.expect("common SSE stream remains open");
|
|
let control = self.next_control().await;
|
|
assert_eq!(control.message["id"], id);
|
|
assert!(control.message["result"].is_object());
|
|
assert_eq!(control.session.as_deref(), Some("session-1"));
|
|
control
|
|
.reply
|
|
.send(Ok(StreamableHttpPostResponse::Accepted))
|
|
.expect("reply POST is still waiting");
|
|
}
|
|
|
|
async fn finish(
|
|
self,
|
|
calls: Vec<Call>,
|
|
posted: usize,
|
|
initialized: usize,
|
|
) -> anyhow::Result<()> {
|
|
for call in calls {
|
|
timeout(TEST_TIMEOUT, call).await???;
|
|
}
|
|
assert_eq!(self.counts.posted.load(SeqCst), posted);
|
|
assert_eq!(self.counts.initialized.load(SeqCst), initialized);
|
|
assert_eq!(self.counts.active.load(SeqCst), 0);
|
|
self.client.cancel().await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn json_limits_allow_overlap_and_preserve_response_ids() -> anyhow::Result<()> {
|
|
let mut zero = config();
|
|
zero.max_concurrent_requests = 0;
|
|
for (config, limit, total) in [
|
|
(config().max_concurrent_requests(2), 2, 5),
|
|
(config().max_concurrent_requests(1), 1, 3),
|
|
(zero, 1, 2),
|
|
(config(), 16, 17),
|
|
] {
|
|
let mut harness = Harness::start(config).await?;
|
|
let calls = (0..total)
|
|
.map(|index| harness.call(format!("request-{index}")))
|
|
.collect();
|
|
let mut pending = Vec::new();
|
|
for _ in 0..limit {
|
|
pending.push(harness.next().await);
|
|
}
|
|
assert_eq!(harness.counts.active.load(SeqCst), limit);
|
|
// Keep the oldest response blocked while newer requests finish first.
|
|
for _ in limit..total {
|
|
pending.pop().unwrap().succeed();
|
|
pending.push(harness.next().await);
|
|
}
|
|
for request in pending.into_iter().rev() {
|
|
request.succeed();
|
|
}
|
|
let counts = harness.counts.clone();
|
|
harness.finish(calls, total, 1).await?;
|
|
assert_eq!(counts.peak.load(SeqCst), limit);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn early_sse_response_releases_the_post_slot() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(1)).await?;
|
|
let first = harness.call("first");
|
|
let release = harness.next().await.start_sse().await?;
|
|
let second = harness.call("second");
|
|
harness.next().await.succeed();
|
|
timeout(TEST_TIMEOUT, second).await???;
|
|
assert!(!first.is_finished(), "the SSE response is still blocked");
|
|
release.send(()).unwrap();
|
|
assert_eq!(harness.counts.peak.load(SeqCst), 1);
|
|
harness.finish(vec![first], 2, 1).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_common_stream_response_finishes_the_request_before_its_post_returns()
|
|
-> anyhow::Result<()> {
|
|
let (stream, incoming) = mpsc::unbounded_channel();
|
|
for (late, orphan) in [
|
|
(Ok(StreamableHttpPostResponse::Accepted), None),
|
|
(Err(StreamableHttpError::SessionExpired), None),
|
|
(
|
|
Ok(StreamableHttpPostResponse::Sse(
|
|
UnboundedReceiverStream::new(incoming).boxed(),
|
|
None,
|
|
)),
|
|
Some(stream),
|
|
),
|
|
] {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(1)).await?;
|
|
let request = harness.cancellable("completed-on-common-stream").await?;
|
|
let post = harness.next().await;
|
|
harness
|
|
.incoming
|
|
.send(sse(serde_json::to_value(post.result())?))?;
|
|
timeout(TEST_TIMEOUT, request.await_response()).await??;
|
|
|
|
post.finish_and_wait(late).await?;
|
|
let next = harness.call("after-completed-request");
|
|
let post = harness.next().await;
|
|
assert_eq!(
|
|
post.name, "after-completed-request",
|
|
"do not retry completed work"
|
|
);
|
|
assert_eq!(post.session.as_deref(), Some("session-1"));
|
|
post.succeed();
|
|
timeout(TEST_TIMEOUT, next).await???;
|
|
if let Some(stream) = orphan {
|
|
timeout(Duration::from_secs(1), stream.closed())
|
|
.await
|
|
.expect("a late SSE response must not leave an orphan stream");
|
|
}
|
|
harness.finish(vec![], 2, 1).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn common_stream_response_keeps_numeric_and_string_ids_distinct() -> anyhow::Result<()> {
|
|
use rmcp::transport::Transport;
|
|
|
|
let (started, mut requests) = mpsc::unbounded_channel();
|
|
let (incoming, incoming_rx) = mpsc::unbounded_channel();
|
|
let counts = Arc::new(Counts::default());
|
|
let mut transport = StreamableHttpClientTransport::with_client(
|
|
ScriptedClient {
|
|
started,
|
|
controls: mpsc::unbounded_channel().0,
|
|
incoming: Arc::new(Mutex::new(Some(incoming_rx))),
|
|
reinitializing: mpsc::unbounded_channel().0,
|
|
counts: counts.clone(),
|
|
},
|
|
config(),
|
|
);
|
|
for message in [
|
|
json!({
|
|
"jsonrpc": "2.0",
|
|
"id": 0,
|
|
"method": "initialize",
|
|
"params": InitializeRequestParams::default()
|
|
}),
|
|
json!({ "jsonrpc": "2.0", "method": "notifications/initialized" }),
|
|
] {
|
|
timeout(
|
|
TEST_TIMEOUT,
|
|
transport.send(serde_json::from_value(message)?),
|
|
)
|
|
.await??;
|
|
}
|
|
assert!(timeout(TEST_TIMEOUT, transport.receive()).await?.is_some());
|
|
let call = |id, name| {
|
|
ClientJsonRpcMessage::request(
|
|
ClientRequest::CallToolRequest(Request::new(CallToolRequestParams::new(name))),
|
|
id,
|
|
)
|
|
};
|
|
|
|
let mut numeric_send = Box::pin(transport.send(call(RequestId::Number(7), "numeric")));
|
|
assert!(futures::poll!(numeric_send.as_mut()).is_pending());
|
|
let numeric_post = next_event(&mut requests).await;
|
|
let numeric_result = numeric_post.result();
|
|
let release_numeric = numeric_post.start_sse().await?;
|
|
timeout(TEST_TIMEOUT, numeric_send).await??;
|
|
|
|
// This exact string id is pending, but has no response stream registration.
|
|
let mut string_send = Box::pin(transport.send(call(RequestId::String("7".into()), "string")));
|
|
assert!(futures::poll!(string_send.as_mut()).is_pending());
|
|
let string_post = next_event(&mut requests).await;
|
|
let string_result = serde_json::to_value(string_post.result())?;
|
|
string_post.finish(Ok(StreamableHttpPostResponse::Accepted));
|
|
timeout(TEST_TIMEOUT, string_send).await??;
|
|
|
|
incoming.send(sse(string_result.clone()))?;
|
|
let received = timeout(TEST_TIMEOUT, transport.receive())
|
|
.await?
|
|
.expect("string-id response");
|
|
assert_eq!(serde_json::to_value(received)?, string_result);
|
|
|
|
release_numeric
|
|
.send(())
|
|
.expect("the distinct numeric-id stream must remain open");
|
|
let received = timeout(TEST_TIMEOUT, transport.receive())
|
|
.await?
|
|
.expect("numeric-id response");
|
|
assert_eq!(
|
|
serde_json::to_value(received)?,
|
|
serde_json::to_value(numeric_result)?
|
|
);
|
|
assert_eq!(counts.posted.load(SeqCst), 2);
|
|
assert_eq!(counts.cancelled.load(SeqCst), 0);
|
|
transport.close().await?;
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_common_stream_response_prevents_replay_during_session_recovery() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(2)).await?;
|
|
let blocking = harness.call("blocking-recovery");
|
|
let blocked = harness.next().await;
|
|
let request = harness.cancellable("completed-during-recovery").await?;
|
|
let expired = harness.next().await;
|
|
let final_response = sse(serde_json::to_value(expired.result())?);
|
|
expired.expire_and_wait().await?;
|
|
harness.incoming.send(final_response)?;
|
|
timeout(TEST_TIMEOUT, request.await_response()).await??;
|
|
blocked.succeed();
|
|
timeout(TEST_TIMEOUT, blocking).await???;
|
|
|
|
let next = harness.call("after-completed-request");
|
|
let post = harness.next().await;
|
|
assert_eq!(
|
|
post.name, "after-completed-request",
|
|
"do not retry completed work"
|
|
);
|
|
let initializations = harness.counts.initialized.load(SeqCst);
|
|
assert!((1..=2).contains(&initializations));
|
|
post.succeed();
|
|
timeout(TEST_TIMEOUT, next).await???;
|
|
harness.finish(vec![], 3, initializations).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn concurrent_session_expiry_shares_one_reinitialization() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(2)).await?;
|
|
let calls = vec![harness.call("first"), harness.call("second")];
|
|
// Hold both requests before releasing either expired response.
|
|
let first = harness.next().await;
|
|
let second = harness.next().await;
|
|
let mut originals = HashMap::new();
|
|
for request in [first, second] {
|
|
assert_eq!(request.session.as_deref(), Some("session-1"));
|
|
originals.insert(request.name.clone(), request.id.clone());
|
|
request.expire();
|
|
}
|
|
for _ in 0..2 {
|
|
let retry = harness.next().await;
|
|
assert_eq!(retry.session.as_deref(), Some("session-2"));
|
|
assert_eq!(originals.remove(&retry.name), Some(retry.id.clone()));
|
|
retry.succeed();
|
|
}
|
|
assert!(originals.is_empty());
|
|
harness.finish(calls, 4, 2).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cancellation_still_runs_while_recovery_waits_for_old_posts() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(3)).await?;
|
|
let hanging = harness.cancellable("hanging").await?;
|
|
let mut blocked = harness.next().await;
|
|
let expired = harness.call("expired");
|
|
harness.next().await.expire_and_wait().await?;
|
|
assert_eq!(harness.counts.initialized.load(SeqCst), 1);
|
|
assert_eq!(harness.counts.active.load(SeqCst), 1);
|
|
|
|
timeout(Duration::from_secs(1), hanging.cancel(None))
|
|
.await
|
|
.expect("cancellation must bypass the session recovery wait")?;
|
|
timeout(TEST_TIMEOUT, blocked.reply.closed()).await?;
|
|
let retry = harness.next().await;
|
|
assert_eq!(retry.name, "expired");
|
|
assert_eq!(retry.session.as_deref(), Some("session-2"));
|
|
retry.succeed();
|
|
harness.finish(vec![expired], 3, 2).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn server_replies_still_run_while_recovery_waits_for_old_posts() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(2)).await?;
|
|
harness.counts.manual_controls.store(true, SeqCst);
|
|
let waiting = harness.call("waiting-for-client");
|
|
let blocked = harness.next().await;
|
|
let expired = harness.call("expired");
|
|
harness.next().await.expire_and_wait().await?;
|
|
|
|
harness.exchange_ping("recovery-ping").await;
|
|
blocked.succeed();
|
|
let retry = harness.next().await;
|
|
assert_eq!(retry.name, "expired");
|
|
assert_eq!(retry.session.as_deref(), Some("session-2"));
|
|
retry.succeed();
|
|
harness.finish(vec![waiting, expired], 3, 2).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_version_barrier_allows_the_server_reply_it_needs() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(2)).await?;
|
|
harness.counts.manual_controls.store(true, SeqCst);
|
|
let mut meta = RequestMetaObject::new();
|
|
meta.set_protocol_version(ProtocolVersion::V_2025_06_18);
|
|
let barrier = harness
|
|
.cancellable_with_options("barrier", PeerRequestOptions::no_options().with_meta(meta))
|
|
.await?;
|
|
let blocked = harness.next().await;
|
|
let queued = harness.cancellable("after-barrier").await?;
|
|
|
|
harness.exchange_ping("barrier-ping").await;
|
|
assert_eq!(harness.counts.posted.load(SeqCst), 1);
|
|
blocked.succeed();
|
|
timeout(TEST_TIMEOUT, barrier.await_response()).await??;
|
|
let next = harness.next().await;
|
|
assert_eq!(next.name, "after-barrier");
|
|
next.succeed();
|
|
timeout(TEST_TIMEOUT, queued.await_response()).await??;
|
|
harness.finish(vec![], 2, 1).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_cancelled_version_barrier_does_not_block_unrelated_posts() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(2)).await?;
|
|
let blocking = harness.call("blocking");
|
|
let blocked = harness.next().await;
|
|
let mut meta = RequestMetaObject::new();
|
|
meta.set_protocol_version(ProtocolVersion::V_2025_06_18);
|
|
let barrier = harness
|
|
.cancellable_with_options(
|
|
"cancelled-barrier",
|
|
PeerRequestOptions::no_options().with_meta(meta),
|
|
)
|
|
.await?;
|
|
timeout(TEST_TIMEOUT, barrier.cancel(None)).await??;
|
|
|
|
let next = harness.call("after-cancelled-barrier");
|
|
let post = harness.next().await;
|
|
assert_eq!(post.name, "after-cancelled-barrier");
|
|
post.succeed();
|
|
timeout(TEST_TIMEOUT, next).await???;
|
|
assert!(!blocked.reply.is_closed(), "the older POST is still held");
|
|
blocked.succeed();
|
|
harness.finish(vec![blocking], 2, 1).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn dropping_a_parked_barrier_send_unblocks_the_ordinary_queue() -> anyhow::Result<()> {
|
|
use std::task::{Context, Wake, Waker};
|
|
|
|
use rmcp::transport::Transport;
|
|
|
|
struct WakeSignal(mpsc::UnboundedSender<()>);
|
|
impl Wake for WakeSignal {
|
|
fn wake(self: Arc<Self>) {
|
|
let _ = self.0.send(());
|
|
}
|
|
}
|
|
|
|
let (started, mut requests) = mpsc::unbounded_channel();
|
|
let counts = Arc::new(Counts::default());
|
|
let mut config = config().max_concurrent_requests(2);
|
|
config.channel_buffer_capacity = 1;
|
|
let mut transport = StreamableHttpClientTransport::with_client(
|
|
ScriptedClient {
|
|
started,
|
|
controls: mpsc::unbounded_channel().0,
|
|
incoming: Arc::new(Mutex::new(None)),
|
|
reinitializing: mpsc::unbounded_channel().0,
|
|
counts: counts.clone(),
|
|
},
|
|
config,
|
|
);
|
|
for message in [
|
|
json!({
|
|
"jsonrpc": "2.0",
|
|
"id": 0,
|
|
"method": "initialize",
|
|
"params": InitializeRequestParams::default()
|
|
}),
|
|
json!({ "jsonrpc": "2.0", "method": "notifications/initialized" }),
|
|
] {
|
|
timeout(
|
|
TEST_TIMEOUT,
|
|
transport.send(serde_json::from_value(message)?),
|
|
)
|
|
.await??;
|
|
}
|
|
assert!(timeout(TEST_TIMEOUT, transport.receive()).await?.is_some());
|
|
let call = |id, params| {
|
|
ClientJsonRpcMessage::request(
|
|
ClientRequest::CallToolRequest(Request::new(params)),
|
|
RequestId::Number(id),
|
|
)
|
|
};
|
|
let mut first = Box::pin(transport.send(call(1, CallToolRequestParams::new("held"))));
|
|
assert!(futures::poll!(first.as_mut()).is_pending());
|
|
let blocked = next_event(&mut requests).await;
|
|
|
|
let mut meta = RequestMetaObject::new();
|
|
meta.set_protocol_version(ProtocolVersion::V_2025_06_18);
|
|
let mut barrier_request = Request::new(CallToolRequestParams::new("abandoned-barrier"));
|
|
barrier_request.extensions.insert(meta);
|
|
let mut barrier = Box::pin(transport.send(ClientJsonRpcMessage::request(
|
|
ClientRequest::CallToolRequest(barrier_request),
|
|
RequestId::Number(2),
|
|
)));
|
|
assert!(futures::poll!(barrier.as_mut()).is_pending());
|
|
let mut next = Box::pin(transport.send(call(3, CallToolRequestParams::new("after-barrier"))));
|
|
let (wake, mut woke) = mpsc::unbounded_channel();
|
|
let waker = Waker::from(Arc::new(WakeSignal(wake)));
|
|
assert!(
|
|
next.as_mut()
|
|
.poll(&mut Context::from_waker(&waker))
|
|
.is_pending()
|
|
);
|
|
// The barrier fills the queue. The next send wakes only after the worker parks it.
|
|
next_event(&mut woke).await;
|
|
assert_eq!(counts.posted.load(SeqCst), 1, "the barrier must be parked");
|
|
drop(barrier);
|
|
assert!(futures::poll!(next.as_mut()).is_pending());
|
|
|
|
let post = timeout(Duration::from_secs(1), requests.recv())
|
|
.await
|
|
.expect("dropping the parked send must wake the worker")
|
|
.unwrap();
|
|
assert_eq!(post.name, "after-barrier");
|
|
post.succeed();
|
|
timeout(TEST_TIMEOUT, next).await??;
|
|
assert!(timeout(TEST_TIMEOUT, transport.receive()).await?.is_some());
|
|
assert!(!blocked.reply.is_closed(), "the first POST remains held");
|
|
blocked.succeed();
|
|
timeout(TEST_TIMEOUT, first).await??;
|
|
assert_eq!(counts.posted.load(SeqCst), 2);
|
|
assert_eq!(counts.cancelled.load(SeqCst), 0);
|
|
transport.close().await?;
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn recovery_deadline_drops_ambiguous_posts_without_retrying_them() -> anyhow::Result<()> {
|
|
assert_eq!(config().session_recovery_timeout, Duration::from_secs(5));
|
|
let mut harness = Harness::start(
|
|
config()
|
|
.max_concurrent_requests(3)
|
|
.session_recovery_timeout(Duration::from_millis(50)),
|
|
)
|
|
.await?;
|
|
let ambiguous = harness.call("possibly-applied");
|
|
let mut blocked = harness.next().await;
|
|
let expired = harness.call("expired");
|
|
let rejected = harness.next().await;
|
|
let rejected_id = rejected.id.clone();
|
|
rejected.expire_and_wait().await?;
|
|
|
|
assert_recovery_timeout(timeout(TEST_TIMEOUT, ambiguous).await??.unwrap_err());
|
|
timeout(TEST_TIMEOUT, blocked.reply.closed()).await?;
|
|
let retry = harness.next().await;
|
|
assert_eq!(retry.name, "expired");
|
|
assert_eq!(retry.id, rejected_id);
|
|
assert_eq!(retry.session.as_deref(), Some("session-2"));
|
|
retry.succeed();
|
|
harness.finish(vec![expired], 3, 2).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn reinitialization_has_its_own_deadline() -> anyhow::Result<()> {
|
|
let mut harness =
|
|
Harness::start(config().session_recovery_timeout(Duration::from_millis(50))).await?;
|
|
harness.counts.hold_reinitialization.store(true, SeqCst);
|
|
let expired = harness.call("expired");
|
|
harness.next().await.expire_and_wait().await?;
|
|
let mut reinitialization = next_event(&mut harness.reinitializations).await;
|
|
|
|
assert_recovery_timeout(timeout(TEST_TIMEOUT, expired).await??.unwrap_err());
|
|
timeout(TEST_TIMEOUT, reinitialization.closed()).await?;
|
|
harness.finish(vec![], 1, 2).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn an_expired_retry_is_not_retried_again() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(2)).await?;
|
|
let call = harness.call("expires-twice");
|
|
let first = harness.next().await;
|
|
let id = first.id.clone();
|
|
first.expire();
|
|
let retry = harness.next().await;
|
|
assert_eq!(retry.id, id);
|
|
assert_eq!(retry.session.as_deref(), Some("session-2"));
|
|
retry.expire();
|
|
let error = timeout(TEST_TIMEOUT, call).await??.unwrap_err();
|
|
assert!(error.to_string().contains("Session expired"));
|
|
harness.finish(vec![], 2, 2).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_lost_post_response_is_not_retried() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(2)).await?;
|
|
let call = harness.call("possibly-applied");
|
|
harness
|
|
.next()
|
|
.await
|
|
.finish(Err(StreamableHttpError::Client(io::Error::new(
|
|
io::ErrorKind::ConnectionReset,
|
|
"scripted response lost",
|
|
))));
|
|
let error = timeout(TEST_TIMEOUT, call).await??.unwrap_err();
|
|
assert!(error.to_string().contains("scripted response lost"));
|
|
harness.finish(vec![], 1, 1).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn cancellation_bypasses_queued_posts_at_capacity() -> anyhow::Result<()> {
|
|
for (lifecycle, legacy_notifications, initializations) in [
|
|
(ClientLifecycleMode::Initialize, 2, 1),
|
|
(
|
|
ClientLifecycleMode::Discover {
|
|
preferred_versions: vec![ProtocolVersion::V_2026_07_28],
|
|
},
|
|
0,
|
|
0,
|
|
),
|
|
] {
|
|
let mut harness =
|
|
Harness::with_lifecycle(config().max_concurrent_requests(1), lifecycle).await?;
|
|
let request = harness.cancellable("cancel-me").await?;
|
|
let mut blocked = harness.next().await;
|
|
let cancelled = harness.cancellable("never-send").await?;
|
|
timeout(TEST_TIMEOUT, cancelled.cancel(None)).await??;
|
|
let queued = harness.cancellable("queued").await?;
|
|
timeout(TEST_TIMEOUT, request.cancel(None)).await??;
|
|
timeout(TEST_TIMEOUT, blocked.reply.closed()).await?;
|
|
let next = harness.next().await;
|
|
assert_eq!(next.name, "queued");
|
|
next.succeed();
|
|
timeout(TEST_TIMEOUT, queued.await_response()).await??;
|
|
assert_eq!(harness.counts.cancelled.load(SeqCst), legacy_notifications);
|
|
harness.finish(vec![], 2, initializations).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn legacy_cancellation_reaches_the_adapter_before_its_local_stream_drops()
|
|
-> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(1)).await?;
|
|
harness.counts.manual_controls.store(true, SeqCst);
|
|
let stale = harness.notify_cancellation(RequestId::Number(999));
|
|
let held = harness.next_control().await;
|
|
harness.counts.manual_controls.store(false, SeqCst);
|
|
harness
|
|
.counts
|
|
.reject_unmatched_cancellation
|
|
.store(true, SeqCst);
|
|
|
|
let event = harness.cancellable("local-events").await?;
|
|
let (eof, dropped) = harness
|
|
.next()
|
|
.await
|
|
.start_local_sse(harness.counts.clone())
|
|
.await?;
|
|
let peer = harness.client.peer().clone();
|
|
let mut cancel = Box::pin(peer.notify_cancelled(CancelledNotificationParam::new(
|
|
Some(event.id.clone()),
|
|
None,
|
|
)));
|
|
assert!(futures::poll!(cancel.as_mut()).is_pending());
|
|
// The ordinary request follows the cancellation through the service queue.
|
|
let probe = harness.call("after-cancellation");
|
|
harness.next().await.succeed();
|
|
timeout(TEST_TIMEOUT, probe).await???;
|
|
|
|
// EOF is ready, but queued cancellation must pause reads until dispatch.
|
|
eof.cancel();
|
|
let probe = harness.call("after-eof");
|
|
harness.next().await.succeed();
|
|
timeout(TEST_TIMEOUT, probe).await???;
|
|
assert!(
|
|
harness
|
|
.counts
|
|
.local_streams
|
|
.lock()
|
|
.unwrap()
|
|
.contains_key(&event.id),
|
|
"the adapter's stream registration must survive queued cancellation"
|
|
);
|
|
assert!(!held.reply.is_closed(), "the control slot is still held");
|
|
|
|
held.reply
|
|
.send(Ok(StreamableHttpPostResponse::Accepted))
|
|
.unwrap();
|
|
timeout(TEST_TIMEOUT, stale).await???;
|
|
timeout(TEST_TIMEOUT, cancel).await??;
|
|
timeout(TEST_TIMEOUT, dropped).await??;
|
|
assert!(matches!(
|
|
timeout(TEST_TIMEOUT, event.await_response()).await?,
|
|
Err(rmcp::ServiceError::Cancelled { .. })
|
|
));
|
|
assert_eq!(harness.counts.local_cancellations.load(SeqCst), 1);
|
|
assert_eq!(harness.counts.cancelled.load(SeqCst), 1);
|
|
harness.finish(vec![], 3, 1).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn control_timeout_is_configurable_and_releases_its_slot() -> anyhow::Result<()> {
|
|
assert_eq!(config().control_request_timeout, Duration::from_secs(5));
|
|
let mut harness =
|
|
Harness::start(config().control_request_timeout(Duration::from_millis(50))).await?;
|
|
harness.counts.manual_controls.store(true, SeqCst);
|
|
let cancellation = harness.notify_cancellation(RequestId::Number(999));
|
|
let mut cancelled_post = harness.next_control().await;
|
|
harness.incoming.send(sse(json!({
|
|
"jsonrpc": "2.0", "id": "timed-reply", "method": "ping"
|
|
})))?;
|
|
|
|
let error = timeout(Duration::from_secs(1), cancellation)
|
|
.await??
|
|
.unwrap_err();
|
|
assert!(matches!(
|
|
transport_error(&error.into()),
|
|
StreamableHttpError::ControlRequestTimeout
|
|
));
|
|
timeout(Duration::from_secs(1), cancelled_post.reply.closed()).await?;
|
|
|
|
let mut reply_post = harness.next_control().await;
|
|
assert_eq!(reply_post.message["id"], "timed-reply");
|
|
assert!(reply_post.message["result"].is_object());
|
|
harness.counts.manual_controls.store(false, SeqCst);
|
|
let next = harness.notify_cancellation(RequestId::Number(1000));
|
|
timeout(Duration::from_secs(1), reply_post.reply.closed()).await?;
|
|
timeout(Duration::from_secs(1), next).await???;
|
|
assert_eq!(harness.counts.cancelled.load(SeqCst), 2);
|
|
harness.finish(vec![], 0, 1).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_hanging_legacy_control_does_not_delay_local_cancellation() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(1)).await?;
|
|
harness.counts.manual_controls.store(true, SeqCst);
|
|
let stale = harness.notify_cancellation(RequestId::Number(999));
|
|
let mut held = harness.next_control().await;
|
|
assert_eq!(held.message["params"]["requestId"], 999);
|
|
harness.counts.manual_controls.store(false, SeqCst);
|
|
|
|
let live = harness.cancellable("live-post").await?;
|
|
let mut blocked = harness.next().await;
|
|
let cancel_post = tokio::spawn(async move { live.cancel(None).await });
|
|
timeout(Duration::from_secs(1), blocked.reply.closed()).await?;
|
|
|
|
let streaming = harness.cancellable("live-stream").await?;
|
|
let mut stream = harness.next().await.start_sse().await?;
|
|
let cancel_stream = harness.notify_cancellation(streaming.id.clone());
|
|
assert!(
|
|
!held.reply.is_closed(),
|
|
"the old control POST is still held"
|
|
);
|
|
assert_eq!(harness.counts.cancelled.load(SeqCst), 1);
|
|
|
|
// The default control timeout is five seconds; give its watchdog headroom.
|
|
let error = timeout(Duration::from_secs(10), stale).await??.unwrap_err();
|
|
assert!(matches!(
|
|
transport_error(&error.into()),
|
|
StreamableHttpError::ControlRequestTimeout
|
|
));
|
|
timeout(TEST_TIMEOUT, held.reply.closed()).await?;
|
|
timeout(TEST_TIMEOUT, cancel_post).await???;
|
|
timeout(TEST_TIMEOUT, cancel_stream).await???;
|
|
timeout(TEST_TIMEOUT, stream.closed()).await?;
|
|
assert!(matches!(
|
|
timeout(TEST_TIMEOUT, streaming.await_response()).await?,
|
|
Err(rmcp::ServiceError::Cancelled { .. })
|
|
));
|
|
assert_eq!(harness.counts.cancelled.load(SeqCst), 3);
|
|
harness.finish(vec![], 2, 1).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn close_drops_blocked_posts_before_deleting_the_session() -> anyhow::Result<()> {
|
|
let mut harness = Harness::start(config().max_concurrent_requests(2)).await?;
|
|
let calls = [harness.call("first"), harness.call("second")];
|
|
let posts = [harness.next().await, harness.next().await];
|
|
timeout(TEST_TIMEOUT, harness.client.cancel()).await??;
|
|
assert!(posts.iter().all(|post| post.reply.is_closed()));
|
|
assert_eq!(harness.counts.active.load(SeqCst), 0);
|
|
assert_eq!(harness.counts.deleted.load(SeqCst), 1);
|
|
for call in calls {
|
|
assert!(timeout(TEST_TIMEOUT, call).await??.is_err());
|
|
}
|
|
Ok(())
|
|
}
|