diff --git a/docs/operations.md b/docs/operations.md index 8e45565..06e3e79 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -794,6 +794,7 @@ policy = "deny" - `oauth = true` on a `url` server enables `oauth login` (optional `scopes`, `client_name`, or `client_id` and `client_secret` for a client registered ahead of time, and `authorization_params`). - Rarely used stdio servers: `lazy = true` starts the process on the first call and stops it after `idle_secs` (default 600) without calls; `/healthz` shows it as `idle` meanwhile and its tools stay searchable. A failed start fails that call only, and the next call tries again. +- Parallel calls: the gateway sends a server at most `max_concurrent` calls at once (default 8), and the rest of a script's `asyncio.gather` waits its turn, so a burst of calls does not trip the server's rate limit. A call that gets no turn within `tool_timeout_secs` fails with a `TimeoutError` without being sent, and nothing is retried. A spoke calling one of the hub's servers waits out the hub's queue too, using the longer of its own `tool_timeout_secs` and the hub's. The setting takes 1 to 4096. When a tool's error reads as a rate limit (`rate limit`, `ratelimit` or `too many requests`), the script's error names this setting and the file to change it in, which is the included file for a server kept in one, so the agent can ask for it to be lowered, e.g. `max_concurrent = 4` under `[servers.linear]`. Builtin servers, such as `secrets`, answer inside the daemon and are not capped, so a script's finds still reach the owner as one prompt. - Keys stay in 1Password: a `headers`, `env` or `url` value can hold `{{ op://vault/item/field }}` references, as in `op inject`, e.g. `headers = { authorization = "Bearer {{ op://Dev/Context7/credential }}" }`. The daemon runs `op read` the first time the server connects and keeps the value in memory until it exits. A clean stop (`service restart`, `update`, a reboot) hands the values to the next daemon sealed, so it asks nothing again; to pick up a rotated key, run `codemode-mcp service restart --fresh`. With the 1Password app integration the Mac asks once when the daemon starts (a locked or unanswered prompt shows `secret_unavailable` in `/healthz` and retries every 10 minutes). `[secrets] op` names the `op` binary (default `op` on `PATH`). A `url` that carries the key, as Zoho's MCP URLs do, should embed the reference where the key sits: `url = "https://-.zohomcp.com/mcp/{{ op://Personal/Zoho MCP - InfraOps/key }}/message"`. References in a `url` may sit only in path segments and query names or values, never in the scheme, credentials, host, port or fragment, and the `url` needs a literal host; anything else is refused when the config loads. A key embedded in a `url` may hold only ASCII letters, digits and `- . _ ~ + / = : @ ! * , ;` (at least 16 bytes), which covers hex, base64, base64url, JWT-like and colon-separated keys; any other character, such as `\`, quotes, `<`, `>`, `^`, `{`, `}`, `|`, `[`, `]`, `#`, `?`, `%`, `&`, `$`, `(`, `)` or a space, refuses the server with a configuration error. It is encoded as data where it sits, so `/` or `+` in a key stays part of that one segment or query value. The secrets are exactly the values 1Password returns, in the encodings libraries print them in (percent-encoded in either case, JSON-escaped, HTML character references such as `+` or `+`, `+` read as a space), matched ignoring ASCII case, and any run of 8 or more of its bytes is redacted too, since rmcp cuts invalid responses to 256 bytes and can leave only part of a key; nothing else in the URL is guessed to be a key. A value read for an embedded reference that is itself an `http(s)` URL is refused with a configuration error naming the reference, and dropped here and from the phones' copies, so the next attempt asks again instead of replaying it. A `url` that is one whole reference (`url = "{{ op://Personal/Zoho MCP infraops/credential }}"`) is refused when the config loads, with or without OAuth, because codemode cannot tell which part of such a URL is the key; embed the reference instead, e.g. `url = "https://host/mcp/{{ op://Vault/Item/key }}"`. Log text is matched through nested escape layers (a client's `Debug` output escapes the JSON escapes of a response body), up to 4 passes. Only the configured template is ever shown (companion prompts name the reference, never the value), and the resolved URL is redacted from errors, `/healthz` and the daemon's logs. Whatever OAuth would store (credentials, pending state, the client registration) or print (the authorization link) is refused when it holds a key, and the link's `resource` parameter is left out only when it holds the key. - Per tool: `policy` (`allow`, `ask`, `deny`) and, for `ask`, `grant` (`session` or `per_call`). A `session` grant is given to the calling client, not to one MCP session: it covers every running session of that client name for 15 or 60 minutes. - The server is session-less: it issues no `mcp-session-id`, for any protocol version. Clients are told apart by the client name in the request `_meta`, the `initialize` handshake or the `user-agent` header. While `execute` or `wait` is waiting, it sends progress notifications every `limits.progress_interval_ms` (15000) to clients that sent a `progressToken`. diff --git a/src/config.rs b/src/config.rs index f6251b4..4e2217a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -147,6 +147,10 @@ pub struct Config { /// nothing can be written back #[serde(skip)] path: Option, + /// For servers set up across included files, the file that sets each + /// one's `max_concurrent`, or else the file holding its table + #[serde(skip)] + server_files: BTreeMap, } impl Default for Config { @@ -169,6 +173,7 @@ impl Default for Config { hub: None, servers: BTreeMap::new(), path: None, + server_files: BTreeMap::new(), } } } @@ -261,6 +266,15 @@ impl Config { self.path.as_deref() } + /// The file to change `servers..max_concurrent` in: the one + /// that sets it, or else the one holding the server's table + pub fn max_concurrent_file(&self, server: &str) -> Option<&Path> { + self.server_files + .get(server) + .map(PathBuf::as_path) + .or(self.path()) + } + /// Parses and validates configuration text; `path` is used in errors and /// to resolve relative `include` paths /// @@ -309,21 +323,35 @@ impl Config { } merged.add(table, path)?; - merged - .table - .try_into() - .map_err(|source| ConfigError::ParseIncluded { - files: format!( - "{} together with {}", - path.display(), - included_paths - .iter() - .map(|included| included.display().to_string()) - .collect::>() - .join(", ") - ), - source, - })? + let Merged { table, origins } = merged; + let mut config: Self = + table + .try_into() + .map_err(|source| ConfigError::ParseIncluded { + files: format!( + "{} together with {}", + path.display(), + included_paths + .iter() + .map(|included| included.display().to_string()) + .collect::>() + .join(", ") + ), + source, + })?; + // a value already set must be changed where it is, or the + // files would clash + for name in config.servers.keys() { + let key = ["servers", name, "max_concurrent"].map(str::to_owned); + if let Some(file) = (1..=key.len()) + .rev() + .find_map(|end| origins.get(&key[..end])) + { + config.server_files.insert(name.clone(), file.clone()); + } + } + + config } }; config.validate()?; @@ -693,6 +721,14 @@ fn default_idle_secs() -> u64 { 10 * 60 } +/// Highest `max_concurrent`; more is no limit in practice, and the +/// semaphore that enforces it panics past its own maximum +const MAX_CONCURRENT: usize = 4096; + +fn default_max_concurrent() -> usize { + 8 +} + fn default_op() -> String { "op".to_owned() } @@ -1254,6 +1290,11 @@ pub struct ServerConfig { /// Seconds a `lazy` server may go without calls before it is stopped #[serde(default = "default_idle_secs")] pub idle_secs: u64, + /// Calls the gateway sends this server at once; the rest wait their + /// turn, so a script's `asyncio.gather` cannot trip the server's rate + /// limit. Builtin servers answer inside the daemon and are not limited + #[serde(default = "default_max_concurrent")] + pub max_concurrent: usize, /// Endpoint of a Streamable HTTP server; may hold `{{ op://... }}` /// references for a URL that carries a key, and is then never shown /// resolved @@ -1485,6 +1526,11 @@ impl ServerConfig { } fn validate(&self, name: &str) -> Result<(), ConfigError> { + if !(1..=MAX_CONCURRENT).contains(&self.max_concurrent) { + return Err(ConfigError::Invalid(format!( + "servers.{name}.max_concurrent must be from 1 to {MAX_CONCURRENT}" + ))); + } for (tool, tool_override) in &self.tools { tool_override.validate(name, tool)?; } diff --git a/src/gateway.rs b/src/gateway.rs index 07da4e5..8aaf57b 100644 --- a/src/gateway.rs +++ b/src/gateway.rs @@ -119,7 +119,9 @@ impl ToolCallError { match self { Self::Unknown(_) => ExcType::NameError, Self::Denied(_) | Self::NotApproved { .. } => ExcType::PermissionError, - Self::Upstream(UpstreamError::TimedOut { .. }) => ExcType::TimeoutError, + Self::Upstream(UpstreamError::TimedOut { .. } | UpstreamError::Busy { .. }) => { + ExcType::TimeoutError + } Self::ResultTooLarge { .. } => ExcType::ValueError, Self::Hub(error) => match error.kind { HubErrorKind::Unknown => ExcType::NameError, @@ -140,7 +142,9 @@ impl ToolCallError { match self { Self::Unknown(_) => Outcome::NameError, Self::Denied(_) | Self::NotApproved { .. } => Outcome::Denied, - Self::Upstream(UpstreamError::TimedOut { .. }) => Outcome::Timeout, + Self::Upstream(UpstreamError::TimedOut { .. } | UpstreamError::Busy { .. }) => { + Outcome::Timeout + } Self::Hub(error) => match error.kind { HubErrorKind::Unknown => Outcome::NameError, HubErrorKind::Denied => Outcome::Denied, @@ -1017,8 +1021,19 @@ impl Gateway { approval: call.hub_request.clone(), sensitive: call.sensitive, }; + // the hub's own limit governs how long the call may queue and + // run there; an older hub does not say, so ours stands in + let timeout = self + .hub_view() + .servers + .iter() + .find(|server| server.name == call.server) + .and_then(|server| server.call_timeout_secs) + .map_or(self.config.tool_timeout(), |secs| { + Duration::from_secs(secs).max(self.config.tool_timeout()) + }); let answer = hub - .call(&forwarded, self.config.tool_timeout()) + .call(&forwarded, timeout) .await .map_err(|error| hub_unreachable(&error))?; let value = match answer { @@ -1042,7 +1057,7 @@ impl Gateway { .map_err(|error| match error { // the upstream's own error message can echo configured secrets UpstreamError::Service(rmcp::service::ServiceError::McpError(_)) => { - ToolCallError::ToolFailed(self.redactor.apply(&error.public_message())) + self.tool_failed(&call.server, self.redactor.apply(&error.public_message())) } error => ToolCallError::Upstream(error), })?; @@ -1183,6 +1198,10 @@ impl Gateway { .map(|server| server.aliases.clone()) .unwrap_or_default(), tools: export_tools(&self.config, snapshot), + call_timeout_secs: self + .upstreams + .get(&snapshot.name) + .map(|upstream| upstream.call_timeout().as_secs()), }) }) .collect(); @@ -1298,6 +1317,33 @@ fn omit_none_arguments(entry: &ToolEntry, mut arguments: Map) -> } impl Gateway { + /// A tool's error, with where to lower `max_concurrent` when it reads + /// as the server's rate limit, so the agent can say what to change + fn tool_failed(&self, server: &str, message: String) -> ToolCallError { + let limit = self + .config + .servers + .get(server) + .filter(|config| config.builtin.is_none()) + .map(|config| config.max_concurrent); + let Some(limit) = limit.filter(|_| looks_rate_limited(&message)) else { + return ToolCallError::ToolFailed(message); + }; + let config = self.config.max_concurrent_file(server).map_or_else( + || "the codemode-mcp config".to_owned(), + |path| format!("`{}`", path.display()), + ); + + // the path is the owner's own, but anything shown passes the + // redactor, which a value read from 1Password may share text with + ToolCallError::ToolFailed(self.redactor.apply(&format!( + "{message}\n\n`{server}` looks rate limited. codemode sends it at most {limit} calls \ + at once; make fewer calls at once, or ask the user to set a lower `max_concurrent` under \ + `[servers.{server}]` in {config} on the machine that runs `{server}`, then run \ + `codemode-mcp service restart`" + ))) + } + /// Converts an MCP tool result to the value a script receives /// /// Structured content wins. Without it, text that is a JSON object or @@ -1335,7 +1381,7 @@ impl Gateway { self.redactor.apply(&text.join("\n")) }; - return Err(ToolCallError::ToolFailed(message)); + return Err(self.tool_failed(&call.server, message)); } if matches!(call.output, OutputSpec::Text) { // a text tool that sends only structured content still gives the @@ -1361,6 +1407,22 @@ impl Gateway { } } +/// Whether an upstream's error reads as a rate limit, such as HTTP 429's +/// `Too Many Requests` +fn looks_rate_limited(message: &str) -> bool { + let message = message.to_ascii_lowercase(); + + [ + "rate limit", + "rate-limit", + "rate_limit", + "ratelimit", + "too many requests", + ] + .iter() + .any(|phrase| message.contains(phrase)) +} + /// Parses text blocks as one JSON object or array; scalars stay text because /// a tool answering `123` or `true` as prose is more likely than as data fn json_from_text(blocks: &[String]) -> Option { diff --git a/src/hub.rs b/src/hub.rs index c55b8ef..3e3d1af 100644 --- a/src/hub.rs +++ b/src/hub.rs @@ -93,6 +93,10 @@ pub struct HubServer { pub aliases: Vec, /// Its tools pub tools: Vec, + /// Seconds the hub lets a call of it wait for a turn under + /// `max_concurrent`, and then run; absent from older hubs + #[serde(default, skip_serializing_if = "Option::is_none")] + pub call_timeout_secs: Option, } /// Body of `POST /hub/v1/gate`: asks the hub whether a call may run now @@ -195,9 +199,10 @@ impl From<&ToolCallError> for HubError { let kind = match error { ToolCallError::Unknown(_) => HubErrorKind::Unknown, ToolCallError::Denied(_) | ToolCallError::NotApproved { .. } => HubErrorKind::Denied, - ToolCallError::Upstream(crate::upstream::UpstreamError::TimedOut { .. }) => { - HubErrorKind::TimedOut - } + ToolCallError::Upstream( + crate::upstream::UpstreamError::TimedOut { .. } + | crate::upstream::UpstreamError::Busy { .. }, + ) => HubErrorKind::TimedOut, ToolCallError::ResultTooLarge { .. } => HubErrorKind::TooLarge, ToolCallError::Hub(error) => error.kind, _ => HubErrorKind::Failed, diff --git a/src/hub/client.rs b/src/hub/client.rs index 6ecf970..490a48a 100644 --- a/src/hub/client.rs +++ b/src/hub/client.rs @@ -40,6 +40,8 @@ use crate::{ pub const ID_FILE: &str = "spoke.id"; /// How long a request that needs no person may take const QUICK: Duration = Duration::from_secs(30); +/// The longest call timeout a forwarded call is waited out for +const LONGEST_CALL: Duration = Duration::from_secs(24 * 60 * 60); /// How long a request that may wait for the owner may take: a long poll, /// or a read the owner approves on a device const PATIENT: Duration = Duration::from_secs(15 * 60); @@ -277,14 +279,21 @@ impl HubClient { self.post("/gate", call, QUICK).await } - /// Runs a call of the hub's tool there; `tool_timeout` is the spoke's - /// own limit, which the hub's may exceed + /// Runs a call of the hub's tool there; `tool_timeout` is the longer of + /// the spoke's limit and the one the hub lists for the server + /// + /// The hub may first queue the call behind the server's + /// `max_concurrent` for up to that timeout, so the wait covers that too pub async fn call( &self, call: &ForwardedCall, tool_timeout: Duration, ) -> Result { - self.post("/call", call, tool_timeout + QUICK).await + // the hub names its own timeout, so a huge one is held to a day + // rather than overflowing the deadline + let wait = tool_timeout.min(LONGEST_CALL).saturating_mul(2) + QUICK; + + self.post("/call", call, wait).await } /// The hub's verdict on request `id`, after a long poll; `None` while it diff --git a/src/upstream.rs b/src/upstream.rs index 4f764ca..fd6e406 100644 --- a/src/upstream.rs +++ b/src/upstream.rs @@ -32,7 +32,7 @@ use serde_json::{Map, Value}; use tokio::{ io::{AsyncBufReadExt, BufReader, DuplexStream}, process::Command, - sync::{Mutex, Notify, watch}, + sync::{Mutex, Notify, Semaphore, SemaphorePermit, watch}, task::JoinHandle, }; @@ -140,6 +140,20 @@ pub enum UpstreamError { /// Timeout that elapsed timeout: Duration, }, + /// Every `max_concurrent` slot stayed taken for the call's timeout; the + /// call was never sent, so failing it is safe + #[error( + "`{server}` had {limit} calls running and this one got no turn within {waited:?}; \ + it was not sent. Make fewer calls at once" + )] + Busy { + /// Server name + server: String, + /// Its `max_concurrent` + limit: usize, + /// How long the call waited + waited: Duration, + }, /// The server asked for something other than a tool result #[error("server asked for interactive input, which is not supported")] InputRequired, @@ -170,7 +184,9 @@ impl UpstreamError { Self::Header(_) | Self::Url | Self::UrlValue(_) => ErrorCategory::Configuration, Self::Secret(_) => ErrorCategory::Secret, Self::Initialize(_) => ErrorCategory::Initialize, - Self::ConnectTimeout(_) | Self::TimedOut { .. } => ErrorCategory::Timeout, + Self::ConnectTimeout(_) | Self::TimedOut { .. } | Self::Busy { .. } => { + ErrorCategory::Timeout + } Self::Service(_) | Self::InputRequired | Self::SetupFailed { .. } => { ErrorCategory::Protocol } @@ -188,6 +204,7 @@ impl UpstreamError { match self { Self::Unavailable { .. } | Self::TimedOut { .. } + | Self::Busy { .. } | Self::InputRequired | Self::SetupFailed { .. } | Self::NeedsLogin(_) => self.to_string(), @@ -346,6 +363,9 @@ pub struct Upstream { setup_epoch: AtomicU64, /// Serializes setup, so concurrent first calls run it once setup_turn: Mutex<()>, + /// Caps the calls in flight at `max_concurrent`; none for a builtin + /// server, whose finds must all reach the owner as one prompt + call_slots: Option, /// Logins for `oauth = true` servers oauth: Option>, /// Where the last tool list is kept for the next start @@ -439,6 +459,11 @@ impl Upstream { self.connected_peer().map(|_| ()) } + /// How long one call may wait for a turn, and then how long it may run + pub fn call_timeout(&self) -> Duration { + self.call_timeout + } + /// Whether the catalog offers this server's tools as callable pub fn callable(&self) -> bool { self.config.lazy || self.status() == UpstreamStatus::Connected @@ -554,6 +579,7 @@ impl Upstream { arguments: Map, ) -> Result { let _in_use = self.begin_use(); + let _slot = self.call_slot().await?; let (peer, epoch) = self.awake_peer().await?; self.ensure_setup(&peer, epoch, tool).await?; @@ -561,6 +587,27 @@ impl Upstream { .await } + /// Waits for one of the `max_concurrent` slots, so the server is never + /// sent more calls at once and nothing has to be retried + /// + /// The wait is bounded by the call's own timeout, so a call forwarded + /// from a spoke answers before the spoke stops waiting for it + async fn call_slot(&self) -> Result>, UpstreamError> { + let Some(slots) = &self.call_slots else { + return Ok(None); + }; + + match tokio::time::timeout(self.call_timeout, slots.acquire()).await { + // the semaphore is never closed + Ok(permit) => Ok(permit.ok()), + Err(_) => Err(UpstreamError::Busy { + server: self.name.clone(), + limit: self.config.max_concurrent, + waited: self.call_timeout, + }), + } + } + /// Adds the server's configured `arguments` that `tool` declares and the /// caller left out; the stubs hide them, so scripts never pass them fn with_fixed_arguments( @@ -1386,6 +1433,10 @@ impl Upstreams { refresh_turn: Mutex::new(()), setup_epoch: AtomicU64::new(0), setup_turn: Mutex::new(()), + call_slots: config + .builtin + .is_none() + .then(|| Semaphore::new(config.max_concurrent)), oauth: oauth.clone(), store: store.clone(), reconnect: watch::channel(0).0, diff --git a/tests/fixtures/fixture_server.rs b/tests/fixtures/fixture_server.rs index 0d140bc..58d7892 100644 --- a/tests/fixtures/fixture_server.rs +++ b/tests/fixtures/fixture_server.rs @@ -3,7 +3,7 @@ use std::{ sync::{ Arc, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicUsize, Ordering}, }, time::Duration, }; @@ -64,6 +64,28 @@ struct EchoParams { delay_ms: u64, } +#[derive(Deserialize, schemars::JsonSchema)] +struct OverlapParams { + /// How long to stay in flight + delay_ms: u64, +} + +#[derive(Serialize, schemars::JsonSchema)] +struct Overlap { + /// Most calls of this tool in flight at once so far + peak: usize, +} + +#[derive(Deserialize, schemars::JsonSchema)] +struct FailParams { + /// Text of the error to answer with + message: String, +} + +/// Calls of `overlap` running now, and the most seen at once +static IN_FLIGHT: AtomicUsize = AtomicUsize::new(0); +static PEAK: AtomicUsize = AtomicUsize::new(0); + #[derive(Deserialize, schemars::JsonSchema)] struct UseTokenParams { /// The token to use @@ -316,6 +338,26 @@ impl Fixture { }) } + #[tool(description = "Answer with an error carrying the message")] + fn fail( + &self, + Parameters(FailParams { message }): Parameters, + ) -> Result { + Err(message) + } + + #[tool(description = "Stay in flight for delay_ms, then report the most calls seen at once")] + async fn overlap(&self, Parameters(params): Parameters) -> Json { + let running = IN_FLIGHT.fetch_add(1, Ordering::SeqCst) + 1; + PEAK.fetch_max(running, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(params.delay_ms)).await; + IN_FLIGHT.fetch_sub(1, Ordering::SeqCst); + + Json(Overlap { + peak: PEAK.load(Ordering::SeqCst), + }) + } + #[tool(description = "The largest u64, which no i64 can hold")] fn wide(&self) -> Json { Json(Wide { n: u64::MAX }) diff --git a/tests/gateway/concurrency.rs b/tests/gateway/concurrency.rs new file mode 100644 index 0000000..8b25ffa --- /dev/null +++ b/tests/gateway/concurrency.rs @@ -0,0 +1,77 @@ +//! `max_concurrent`: calls past a server's cap wait their turn, and a rate +//! limit error says where to lower the cap + +use super::harness::{Harness, Setup, TestResult}; + +#[tokio::test(flavor = "multi_thread")] +async fn calls_past_the_cap_wait_and_rate_limits_name_the_setting() -> TestResult { + let fixture = env!("CARGO_BIN_EXE_codemode-fixture"); + let extra = format!("[servers.roomy]\ncommand = \"{fixture}\"\n"); + // a server kept in an included file is changed there, not in config.toml + let shared = std::env::temp_dir().join(format!( + "codemode-concurrency-{}-servers.toml", + std::process::id() + )); + std::fs::write( + &shared, + format!("[servers.shared]\ncommand = \"{fixture}\"\nmax_concurrent = 3\n"), + )?; + let top = format!("include = [\"{}\"]", shared.display()); + let Some(harness) = Harness::with(Setup { + server: "max_concurrent = 2", + extra: &extra, + top: &top, + ..Setup::default() + }) + .await? + else { + return Ok(()); + }; + + // every call still runs; none is refused for waiting + let body = harness + .execute_json( + "import asyncio\n\ + capped = await asyncio.gather(*[fixture_overlap(delay_ms=200) for _ in range(6)])\n\ + roomy = await asyncio.gather(*[roomy_overlap(delay_ms=200) for _ in range(12)])\n\ + [len(capped), max(r[\"peak\"] for r in capped), len(roomy), max(r[\"peak\"] for r in roomy)]", + ) + .await?; + assert_eq!(body["result"], serde_json::json!([6, 2, 12, 8]), "{body}"); + + let body = harness + .execute_json( + "messages: list[str] = []\n\ + for message in [\"429 Too Many Requests\", \"no such issue\"]:\n \ + try:\n await fixture_fail(message=message)\n \ + except RuntimeError as error:\n messages.append(str(error))\n\ + messages", + ) + .await?; + let limited = body["result"][0].as_str().unwrap_or_default(); + assert!( + limited.contains("at most 2 calls") + && limited.contains("`max_concurrent` under `[servers.fixture]`") + && limited.contains("config.toml"), + "{body}" + ); + let other = body["result"][1].as_str().unwrap_or_default(); + assert_eq!(other, "no such issue", "{body}"); + + let body = harness + .execute_json( + "message = \"\"\n\ + try:\n await shared_fail(message=\"Rate limit exceeded\")\n\ + except RuntimeError as error:\n message = str(error)\n\ + message", + ) + .await?; + let _ = std::fs::remove_file(&shared); + let limited = body["result"].as_str().unwrap_or_default(); + assert!( + limited.contains("at most 3 calls") && limited.contains(&format!("`{}`", shared.display())), + "{body}" + ); + + Ok(()) +} diff --git a/tests/gateway/main.rs b/tests/gateway/main.rs index 365b789..47efbf1 100644 --- a/tests/gateway/main.rs +++ b/tests/gateway/main.rs @@ -5,6 +5,7 @@ mod admin_gate; mod apple; mod approvals; mod companion; +mod concurrency; mod durable; mod harness; mod lazy;