diff --git a/cmd/generate-types/main.go b/cmd/generate-types/main.go index 743516d67..474536191 100644 --- a/cmd/generate-types/main.go +++ b/cmd/generate-types/main.go @@ -92,6 +92,28 @@ export interface HealthStatus { action?: HealthAction; } +`) + + // Activity status vocabulary - generated from internal/storage/activity_models.go + // (ValidActivityStatuses). Closed set: every consumer switches on it, so a new + // status has to be added in Go and regenerated here. + sb.WriteString(`// Activity status vocabulary - generated from internal/storage/activity_models.go +export const ActivityStatusSuccess = 'success' as const; +export const ActivityStatusError = 'error' as const; +export const ActivityStatusBlocked = 'blocked' as const; +/** Spec 093: shed by a concurrency limit before reaching the upstream. */ +export const ActivityStatusRejected = 'rejected' as const; +export type ActivityStatusValue = + | typeof ActivityStatusSuccess + | typeof ActivityStatusError + | typeof ActivityStatusBlocked + | typeof ActivityStatusRejected; + +/** Machine-readable cause of a spec-093 rejection (activity metadata rejection_reason). */ +export type RejectionReason = 'queue_full' | 'queue_timeout'; +/** Limiter tier that shed the call (activity metadata rejection_scope). */ +export type RejectionScope = 'server' | 'global'; + `) // Server types @@ -128,6 +150,13 @@ export interface HealthStatus { health?: HealthStatus; // Unified health status calculated by the backend trust_mode?: string; // Per-server approval trust mode (spec 086): 'auto' | 'scan' | 'manual'; raw configured value, absent when unset (effective default: manual) security_scan?: SecurityScanSummary; // Latest scan summary (spec 086); ABSENT when no scan has ever run + // Spec 093 (#955) per-server concurrency overrides. Tri-state: absent = + // inherit server_concurrency_defaults, 0 = disabled for this server, + // positive = override. Effective concurrency is additionally bounded by the + // proxy-wide (global) limiter. + max_concurrent_requests?: number; + queue_size?: number; + queue_timeout?: string; // Go duration string, e.g. '30s' } export interface SecurityScanSummary { diff --git a/cmd/mcpproxy/activity_cmd.go b/cmd/mcpproxy/activity_cmd.go index 18488063f..ed30dacfe 100644 --- a/cmd/mcpproxy/activity_cmd.go +++ b/cmd/mcpproxy/activity_cmd.go @@ -105,7 +105,9 @@ func (f *ActivityFilter) Validate() error { // Validate status if f.Status != "" { - validStatuses := []string{"success", "error", "blocked"} + // Spec 093 added "rejected" (shed by a concurrency limit) to the closed + // activity status vocabulary; the CLI filter must accept it. + validStatuses := []string{"success", "error", "blocked", "rejected"} valid := false for _, s := range validStatuses { if f.Status == s { @@ -735,7 +737,7 @@ func init() { activityListCmd.Flags().StringVarP(&activityType, "type", "t", "", "Filter by type (comma-separated for multiple): tool_call, system_start, system_stop, internal_tool_call, config_change, policy_decision, quarantine_change, server_change") activityListCmd.Flags().StringVarP(&activityServer, "server", "s", "", "Filter by server name") activityListCmd.Flags().StringVar(&activityTool, "tool", "", "Filter by tool name") - activityListCmd.Flags().StringVar(&activityStatus, "status", "", "Filter by status: success, error, blocked") + activityListCmd.Flags().StringVar(&activityStatus, "status", "", "Filter by status: success, error, blocked, rejected") activityListCmd.Flags().StringVar(&activitySessionID, "session", "", "Filter by session — a work session id (ws-...) or a raw MCP transport session id") activityListCmd.Flags().StringVar(&activityStartTime, "start-time", "", "Filter records after this time (RFC3339)") activityListCmd.Flags().StringVar(&activityEndTime, "end-time", "", "Filter records before this time (RFC3339)") @@ -772,7 +774,7 @@ func init() { activityExportCmd.Flags().StringVarP(&activityType, "type", "t", "", "Filter by type (comma-separated): tool_call, system_start, system_stop, internal_tool_call, config_change, policy_decision, quarantine_change, server_change") activityExportCmd.Flags().StringVarP(&activityServer, "server", "s", "", "Filter by server name") activityExportCmd.Flags().StringVar(&activityTool, "tool", "", "Filter by tool name") - activityExportCmd.Flags().StringVar(&activityStatus, "status", "", "Filter by status") + activityExportCmd.Flags().StringVar(&activityStatus, "status", "", "Filter by status: success, error, blocked, rejected") activityExportCmd.Flags().StringVar(&activitySessionID, "session", "", "Filter by session — a work session id (ws-...) or a raw MCP transport session id") activityExportCmd.Flags().StringVar(&activityStartTime, "start-time", "", "Filter after this time (RFC3339)") activityExportCmd.Flags().StringVar(&activityEndTime, "end-time", "", "Filter before this time (RFC3339)") @@ -1208,8 +1210,11 @@ func formatToolCallEvent(event map[string]interface{}, timestamp string) string if errMsg != "" { line += " " + errMsg } - if status == "blocked" { + switch status { + case "blocked": line += " BLOCKED" + case "rejected": + line += " REJECTED (concurrency limit)" } return line } @@ -1308,6 +1313,9 @@ func formatStatusIcon(status string) string { return "\u2717" // X case "blocked": return "\u2298" // circle with slash + case "rejected": + // Spec 093: shed by a concurrency limit — backpressure, not a failure. + return "\u23f8" // pause default: return "?" } diff --git a/docs/cli/activity-commands.md b/docs/cli/activity-commands.md index 1d7616e17..e8f17b8d6 100644 --- a/docs/cli/activity-commands.md +++ b/docs/cli/activity-commands.md @@ -57,7 +57,7 @@ mcpproxy activity list [flags] | `--type` | `-t` | | Filter by type (comma-separated for multiple): `tool_call`, `system_start`, `system_stop`, `internal_tool_call`, `config_change`, `policy_decision`, `quarantine_change`, `server_change`, `credential_broker` | | `--server` | `-s` | | Filter by server name | | `--tool` | | | Filter by tool name | -| `--status` | | | Filter by status: `success`, `error`, `blocked` | +| `--status` | | | Filter by status: `success`, `error`, `blocked`, `rejected` (shed by a concurrency limit, see [Concurrency Limits](../configuration/config-file.md#concurrency-limits--request-queueing)) | | `--intent-type` | | | Filter by intent operation type: `read`, `write`, `destructive` | | `--request-id` | | | Filter by HTTP request ID for log correlation | | `--no-icons` | | | Disable emoji icons in output (use text instead) | @@ -377,6 +377,7 @@ database:query 15 calls "success_count": 142, "error_count": 5, "blocked_count": 3, + "rejected_count": 0, "success_rate": 0.947, "top_servers": [ {"name": "github", "count": 75}, diff --git a/docs/configuration.md b/docs/configuration.md index 204f0dafb..778e9feba 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -180,6 +180,137 @@ the proxy. > the Web UI / macOS app is planned; for now set per-server overrides via the > Raw JSON editor or the REST API. +### Concurrency Limits & Request Queueing + +Multi-user or multi-agent deployments can overwhelm a fragile upstream (a +database-backed stdio server, a rate-limited API) with simultaneous tool calls. +MCPProxy can cap how many upstream tool calls run at once and park the excess in +a bounded FIFO queue, shedding predictably when that queue is full +(GitHub [#955](https://github.com/smart-mcp-proxy/mcpproxy-go/issues/955)). + +**Everything is off by default** — with no keys set, behavior is exactly as +before: no limiting, no queueing, no new errors. + +There are **three separately named scopes**, each carrying the same three +settings: + +| Scope | Where | What it caps | +|-------|-------|--------------| +| Global aggregate limiter | top-level `max_concurrent_requests` / `queue_size` / `queue_timeout` | All upstream tool calls across the whole proxy | +| Per-server default set | `server_concurrency_defaults` object | Blanket per-server values, inherited by every server that does not override them | +| Per-server override | the same three keys on an `mcpServers[]` entry | That one server | + +```json +{ + "max_concurrent_requests": 50, + "queue_size": 100, + "queue_timeout": "30s", + + "server_concurrency_defaults": { + "max_concurrent_requests": 5, + "queue_size": 10, + "queue_timeout": "30s" + }, + + "mcpServers": [ + { "name": "fragile-db", "command": "db-mcp", "max_concurrent_requests": 1, "queue_size": 2 }, + { "name": "fast-api", "url": "https://api.example.com/mcp", "max_concurrent_requests": 0 } + ] +} +``` + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `max_concurrent_requests` | integer | unset (off) | Maximum upstream tool calls running at once in this scope. `0` (or unset) = no limiter for this scope. | +| `queue_size` | integer | `0` | How many calls may wait for a slot. `0` = no pending capacity: a call arriving at the cap is shed immediately. | +| `queue_timeout` | duration | `"30s"` when a limiter is active | How long a call may wait in the queue before being shed. | + +**Tri-state per-server semantics.** Each per-server key is independently +tri-state: + +- **absent** — inherit the value from `server_concurrency_defaults`; +- **`0`** — disable that setting for this server. `max_concurrent_requests: 0` + opts the server out of per-server limiting entirely (even when the default set + configures one); `queue_size: 0` keeps the cap but removes the queue, so + excess calls are shed instantly; +- **positive** — override the default for this server. + +The global aggregate limiter is **never** an inheritance source for a server. +It applies *on top*: a server's effective concurrency is +**min(resolved per-server limit, global limit)**. Waiting for a per-server slot +does not consume global capacity — the per-server slot is taken first. + +**One deadline, not two.** `queue_timeout` is a total wait budget, not a +per-tier one: a call waiting for a per-server slot and then a global slot shares +a single absolute deadline (the smallest configured timeout among the active +scopes). Queue waiting never consumes the call's execution timeout — the +execution budget starts after admission. + +**Recommended starting point for stdio upstreams: `5`.** stdio does not mean +serial: the MCP stdio transport multiplexes by JSON-RPC id and common SDK +servers process calls through a small worker pool, so `5` mirrors typical +upstream capacity. Drop to `1` only for a server you know is single-threaded or +backed by a fragile store. + +**Shedding.** A shed call is reported as a readable, retry-friendly error rather +than a dropped connection: MCP tool calls get an error tool result, the REST +tool-call endpoint returns `429` with `Retry-After`, and the activity log +records the call with a `rejected` status carrying the reason (`queue_full` or +`queue_timeout`) and scope (`server` or `global`). + +**Hot reload.** All limits are hot-reloadable — edit the config file and the new +values govern subsequent admissions without a restart. Running calls are never +interrupted, but they keep counting against the new caps: after lowering a cap, +nothing new is admitted until occupancy drains below it. Raising a cap admits +waiting calls immediately; queued calls keep their original deadline. + +**Validation.** Negative values are rejected, as is a positive `queue_size` in a +scope whose `max_concurrent_requests` resolves to disabled (a queue in front of +no limiter can never admit anything). Errors name the offending scope and field. +The exception is the documented opt-out above: an explicit per-server +`max_concurrent_requests: 0` is valid even when the default set defines a queue. + +Only the **global aggregate** scope has environment overrides +(`MCPPROXY_MAX_CONCURRENT_REQUESTS`, `MCPPROXY_QUEUE_SIZE`, +`MCPPROXY_QUEUE_TIMEOUT`); the default set and per-server overrides are +file/API-configured. + +**Per-server limits over REST.** `POST /api/v1/servers` and +`PATCH /api/v1/servers/{name}` accept `max_concurrent_requests`, `queue_size` +and `queue_timeout` alongside the other per-server fields, and +`GET /api/v1/servers` echoes them back. All three keep tri-state semantics on +PATCH: omitting a key leaves the stored value alone, and an explicit `0` is the +documented opt-out — it is applied, not treated as "unset". + +```bash +curl -X PATCH http://127.0.0.1:8080/api/v1/servers/fragile-db \ + -H "X-API-Key: $MCPPROXY_API_KEY" -H 'Content-Type: application/json' \ + -d '{"max_concurrent_requests": 1, "queue_size": 2, "queue_timeout": "10s"}' +``` + +**What is limited, and what is not.** Limits apply to *upstream tool calls* from +every in-process origin — the `call_tool_*` variants, direct-routing mode, the +REST `POST /api/v1/tools/call` endpoint, sandboxed `code_execution` scripts and +activity replay. Local tool search, coalesced tool listings and health probes +are never throttled: they are lightweight and must not be able to queue behind a +saturated upstream. The separate-process CLI debug client is out of scope. + +**Observability.** Saturation is visible on the Prometheus surface: + +| Metric | Type | Labels | Meaning | +|--------|------|--------|---------| +| `mcpproxy_tool_calls_rejected_total` | counter | `server`, `reason`, `scope` | Calls shed by a limit. `reason` = `queue_full` \| `queue_timeout`; `scope` = `server` \| `global`. `server` is the call's target even for a global shed. | +| `mcpproxy_concurrency_active` | gauge | `scope`, `server` | Calls currently holding a slot. | +| `mcpproxy_concurrency_queue_depth` | gauge | `scope`, `server` | Calls currently waiting for a slot. | + +Gauges are sampled every 10s; the counter is exact — it is incremented at the +rejection itself, not derived from the internal event stream, so a burst of +sheds cannot lose increments. The same sheds also appear in the activity log +(`status: rejected`) — written on the same synchronous path, and exactly one row +per shed — including the ones from `code_execution` and replay: the rejection is +recorded at the limiter, below the MCP dispatch layer, so no origin can bypass +it and none can double-report it. + ### Debug & Development ```json @@ -237,6 +368,9 @@ the proxy. | `health_check_interval` | duration | No | Per-server override for the global [`health_check_interval`](#discovery--health-checks). `"0s"` disables the liveness probe for this server only. Range: `5s`–`1h`. Omit to inherit the global value. | | `tool_discovery_interval` | duration | No | Per-server override for the global [`tool_discovery_interval`](#discovery--health-checks). Overrides the global/default cadence for this server only; `"0s"` disables the periodic tool-discovery sweep for this server (connect-time and reactive `list_changed` discovery still run). Range: `30s`–`24h`. Omit to inherit the global value. | | `init_timeout` | duration | No | Per-server override for the global [`init_timeout`](#search--tool-limits) — the MCP `initialize` handshake deadline. Raise it for an upstream that warms up (caches/indexes data) before responding to `initialize` (e.g. `"120s"`, `"3m"`); without it such a server is killed mid-startup and, with `docker run --rm`, retries forever. Range: `1s`–`30m`. Omit to inherit the global value (30s default). Settable via the `upstream_servers` tool and `mcpproxy upstream patch --init-timeout`. | +| `max_concurrent_requests` | integer | No | Per-server cap on concurrently running upstream tool calls (see [Concurrency Limits & Request Queueing](#concurrency-limits--request-queueing)). Omit to inherit `server_concurrency_defaults`; `0` opts this server out of per-server limiting; positive = that cap. The global limiter still applies on top. | +| `queue_size` | integer | No | How many calls may wait for this server's slot. Omit to inherit the default set; `0` = no pending capacity (shed immediately at the cap). | +| `queue_timeout` | duration | No | How long a call may wait for this server's slot (e.g. `"10s"`). Omit to inherit the default set (30s when a limiter is active). | | `oauth` | object | No | OAuth configuration (see [OAuth Configuration](#oauth-configuration)) | | `isolation` | object | No | Per-server Docker isolation settings (see [Docker Isolation](#docker-isolation)) | | `enabled` | boolean | No | Enable/disable server (default: `true`) | @@ -1307,6 +1441,9 @@ Many configuration options can be overridden via environment variables: | `MCPPROXY_CERTS_DIR` | `tls.certs_dir` | Custom certificates directory | | `MCPPROXY_DATA` | `data_dir` | Override data directory | | `MCPPROXY_TOOL_RESPONSE_MODE` | `tool_response_mode` | `retrieve_tools` serialization: `full` (default) or `compact` | +| `MCPPROXY_MAX_CONCURRENT_REQUESTS` | `max_concurrent_requests` | Global aggregate cap on concurrent upstream tool calls (`0` disables it). See [Concurrency Limits](#concurrency-limits--request-queueing) | +| `MCPPROXY_QUEUE_SIZE` | `queue_size` | Global aggregate wait-queue length (`0` = shed at the cap) | +| `MCPPROXY_QUEUE_TIMEOUT` | `queue_timeout` | Global aggregate queue wait budget, e.g. `30s` | | `MCPPROXY_DISABLE_OAUTH` | - | Disable OAuth for testing | | `HEADLESS` | - | Run in headless mode | diff --git a/docs/configuration/config-file.md b/docs/configuration/config-file.md index 663fb90c2..ebbf3bcc7 100644 --- a/docs/configuration/config-file.md +++ b/docs/configuration/config-file.md @@ -135,6 +135,67 @@ They only widen in one direction — they cannot re-enable checking that the config disabled. See [Version Updates](/features/version-updates) for where updates are surfaced. +### Concurrency Limits & Request Queueing + +Caps how many upstream tool calls may run at once, so a burst cannot overwhelm a +fragile upstream. **Off by default** — with no keys set there is no limiting, no +queueing and no new errors. + +Three separately named scopes carry the same three settings: + +| Scope | Where | What it caps | +|-------|-------|--------------| +| Global aggregate | top-level `max_concurrent_requests` / `queue_size` / `queue_timeout` | All upstream tool calls across the whole proxy | +| Per-server defaults | `server_concurrency_defaults` object | Blanket per-server values, inherited by servers that do not override them | +| Per-server override | the same three keys on an `mcpServers[]` entry | That one server | + +```json +{ + "max_concurrent_requests": 50, + "queue_size": 100, + "queue_timeout": "30s", + + "server_concurrency_defaults": { + "max_concurrent_requests": 5, + "queue_size": 10 + }, + + "mcpServers": [ + { "name": "fragile-db", "command": "db-mcp", "max_concurrent_requests": 1, "queue_size": 2 }, + { "name": "fast-api", "url": "https://api.example.com/mcp", "max_concurrent_requests": 0 } + ] +} +``` + +| Option | Type | Default | Description | +|--------|------|---------|-------------| +| `max_concurrent_requests` | integer | unset (off) | Upstream tool calls allowed to run at once in this scope. `0` or unset = no limiter for this scope | +| `queue_size` | integer | `0` | How many calls may wait for a slot. `0` = shed immediately at the cap | +| `queue_timeout` | duration | `"30s"` when a limiter is active | How long a call may wait before being shed | + +**Tri-state per-server semantics.** Each per-server key is independent: **absent** +inherits from `server_concurrency_defaults`, **`0`** disables that setting for +this server (`max_concurrent_requests: 0` opts the server out of per-server +limiting entirely), and a **positive** value overrides the default. + +The global limiter is never an inheritance source — it applies on top, so a +server's effective concurrency is **min(per-server limit, global limit)**. +`queue_timeout` is one total wait budget across both tiers, not one per tier, +and queue waiting never eats into the call's execution timeout. + +**Shedding.** A shed call gets a readable, retry-friendly error: an error tool +result for MCP calls, HTTP 429 with `Retry-After` for the REST tool-call +endpoint, and an activity record with the `rejected` status carrying the reason +(`queue_full` or `queue_timeout`) and scope (`server` or `global`). All limits +are hot-reloadable. + +For stdio upstreams, start at `5` rather than `1`: the transport multiplexes and +most SDK servers use a small worker pool. + +Full reference — validation rules, metrics, and which origins are limited — +lives in [`docs/configuration.md`](https://github.com/smart-mcp-proxy/mcpproxy-go/blob/main/docs/configuration.md#concurrency-limits--request-queueing) +in the repository. + ### MCP Servers See [Upstream Servers](/configuration/upstream-servers) for detailed server configuration. diff --git a/docs/configuration/environment-variables.md b/docs/configuration/environment-variables.md index 6eece1208..90fb14e72 100644 --- a/docs/configuration/environment-variables.md +++ b/docs/configuration/environment-variables.md @@ -63,6 +63,18 @@ These variables control browser behavior for OAuth flows: | `CI` | CI environment detection (disables browser) | - | | `BROWSER` | Custom browser executable for OAuth | System default | +### Concurrency Limits + +These override the **global aggregate** limiter only — the per-server default +set and per-server overrides are file/API-configured. See +[Concurrency Limits & Request Queueing](./config-file.md#concurrency-limits--request-queueing). + +| Variable | Description | Default | +|----------|-------------|---------| +| `MCPPROXY_MAX_CONCURRENT_REQUESTS` | Proxy-wide cap on upstream tool calls running at once. `0` disables the global limiter | `0` (off) | +| `MCPPROXY_QUEUE_SIZE` | How many calls may wait for a global slot. `0` = shed immediately at the cap | `0` | +| `MCPPROXY_QUEUE_TIMEOUT` | How long a call may wait before being shed, e.g. `30s` | `30s` when the limiter is active | + ### Core Server Examples ```bash diff --git a/docs/research/request-concurrency-issue-955-2026-08-07.html b/docs/research/request-concurrency-issue-955-2026-08-07.html new file mode 100644 index 000000000..51fdc8050 --- /dev/null +++ b/docs/research/request-concurrency-issue-955-2026-08-07.html @@ -0,0 +1,295 @@ + + +Request queueing & concurrency limits — decision report (#955) + +
+

MCPProxy · Engineering decision report

+

Request queueing & per-upstream concurrency limits

+

+ Issue #955 — request queueing / concurrency limit for multi-user deployments + · 2026-08-07 · 21-agent research workflow; choke-point analysis verified against the code +

+ +
+

Recommendation

+

Option A — a two-tier semaphore limiter inside managed.Client.CallTool, per-server first, then global.

+

It is the only placement that covers every in-daemon upstream call path — the audit proved code_execution and activity replay bypass Manager.CallTool and hit the managed client directly. It needs no new dependency (golang.org/x/sync is already pinned), inherits FIFO fairness and ctx-aware queue_timeout from semaphore.Weighted, sidesteps the Manager.CallTool RLock stall hazard, and gets hot-reload propagation free from the existing SetGlobalConfig fan-out. Config: max_concurrent_requests, queue_size, queue_timeout — global + per-server overrides, 0 = unlimited (pure opt-in, zero behavior change on upgrade).

+
+ +

01The choke point (verified)

+

All four dispatch paths converge on one function — and two of them bypass the Manager entirely, which rules out any limiter placed above the managed client:

+ 
+call_tool_read|write|destructive  mcp.go:2112 ──┐
+legacy call_tool                  mcp.go:2536 ──┼── Manager.CallTool ──┐
+direct-routing mode           mcp_routing.go:216 ──┘   manager.go:1123      │
+REST /api/v1/tools/call → CallToolDirect mcp.go:5356 ──(same variants)──┤
+                                                                          ├──▶ managed.Client.CallTool
+code_execution     mcp_code_execution.go:453-464 ──── GetClient ─────────┤     client.go:640 — limiter here
+activity replay          runtime.go:1218-1228  ──── GetClient ─────────┘
+ + +

02Options

+
+ + + + + + + +
OptionCovers code_execution / replayEffortRisk
A · Two-tier semaphore in managed clientYes — only option that doesMLow-medium
B · Manager.CallTool + MCP dispatchNo — verified bypass holeSMedium
C · Inbound HTTP middleware onlyNo per-upstream limits at allSLow impl / high product
D · Bounded-queue dispatcher per upstreamYesLHigh
+ +
+
+

Option A — Two-tier semaphore limiter at the managed-client choke point

+
RECOMMENDEDEffort MRisk Low-medium
+
+

New internal/upstream/limiter package on x/sync/semaphore. Each limiter = two weighted semaphores: an admission semaphore sized max_concurrent + queue_size acquired with TryAcquire (failure = queue full → instant shed, Envoy max_pending_requests semantics), and a run semaphore sized max_concurrent acquired ctx-aware under queue_timeout. One limiter per server (registry keyed by name) + one global; acquired per-server-first inside managed.Client.CallTool.

+
+

Pros

    +
  • Covers 100% of in-daemon upstream traffic including the code_execution and replay bypass paths — the only placement that does.
    • +
  • No new dependency; semaphore.Weighted gives FIFO fairness, cancellable waits, and queue_timeout free.
    • +
  • Server edition works automatically (shared managed client per server bounds aggregate multi-user load).
    • +
  • Hot-reload free via existing SetGlobalConfig fan-out + per-server config atomics.
    • +
  • Avoids the Manager.CallTool RLock hazard (blocking there stalls AddServer/RemoveServer writers).
    • +
  • Registry keyed by server name is forward-compatible with Spec 074 per-(user,server) brokered pools.
    • +
  • Protects stdio naturally — the transport genuinely multiplexes, so waiters just block before the frame write.
    • +
+

Cons

    +
  • Touches a hot, subtle file (managed/client.go) — must not regress the state machine, ListTools coalescing, or reconnect paths.
    • +
  • Typed shed errors must propagate to mcp.go/httpapi for correct rendering (isError vs 429) — a small cross-layer contract.
    • +
  • Semaphores can’t resize → hot-reload needs the atomic-swap-with-bound-release-closures pattern (fiddly but unit-testable).
    • +
  • Doesn’t govern the separate-process CLI debug client — acceptable; document it.
    • +
+
+
+ +
+

Option B — Limiter in Manager.CallTool + MCP dispatch layer

+
Effort SRisk Medium
+

Global limit in the mcp.go call_tool dispatch; per-server limit in Manager.CallTool. No managed-client changes.

+
+

Pros

    +
  • Smaller upstream-layer diff; mcp.go already has activity/span/metrics seams in place.
    • +
  • Global sheds map cleanly to MCP tool errors with no error-type plumbing.
    • +
+

Cons

    +
  • Verified hole: code_execution and replay call the managed client directly — one script can still stampede a stdio upstream, defeating the issue’s core requirement.
    • +
  • Manager.CallTool holds m.mu.RLock for the whole call; a limiter blocking under it stalls AddServer/RemoveServer — fixing that replicates the drop-and-reacquire dance Option A avoids.
    • +
  • Two enforcement sites; future call paths must remember to route through them.
    • +
+
+
+ +
+

Option C — Inbound HTTP middleware only

+
Effort SHigh product risk
+

A chi middleware capping concurrent inbound requests on /mcp and /api/v1, returning 429 when saturated. No per-upstream awareness.

+
+

Pros

    +
  • Trivial; standard reverse-proxy pattern; clean 429 semantics on REST; protects the daemon’s own goroutine budget.
    • +
+

Cons

    +
  • Cannot express per-upstream limits — one slow stdio server still absorbs unbounded calls while fast servers get throttled: the opposite of what #955 asks.
    • +
  • Amplification blind spot: one code_execution request fans out to many upstream calls, invisible to inbound counting.
    • +
  • Transport-level sheds can abort agent client loops; internal callers (replay) never traverse the listener.
    • +
+
+
+ +
+

Option D — Explicit bounded-queue dispatcher (worker pool per upstream)

+
Effort LRisk High
+

Dedicated dispatcher goroutine + FIFO queue struct per upstream, requests as first-class objects with deadlines, identity, priority — mirroring mcp-go’s own stdio worker pool (5 workers / queue 100).

+
+

Pros

    +
  • Richest future capabilities: queue introspection in the Web UI, per-user fair queuing, priorities, cancellation by id.
    • +
+

Cons

    +
  • Substantially more code and state for behavior two semaphores already deliver (semaphore waiters are FIFO — fairness is equal today).
    • +
  • New failure modes: dispatcher leaks, queue-entry lifecycle bugs, shutdown ordering against the managed-client state machine.
    • +
  • Per-user fairness — the one thing this buys — is a server-edition concern with no current demand, interacting with unimplemented Spec 074. Speculative now.
    • +
+
+
+ +

03Shed semantics

+
+ + + + + + + +
SurfaceOn queue full / timeoutWhy
MCP tools/callisError:true tool result: “server '<name>' is at its concurrency limit (<n> running, <q> queued). Retry in a few seconds.”The MCP spec’s own canonical isError example is a rate-limit message; agent LLMs read it and back off. A JSON-RPC protocol error can abort some client loops.
REST APIHTTP 429 + Retry-Afternginx-documented API-correct status; Envoy/LiteLLM precedent.
Activity logNew status rejected (not generic error)Dashboards must separate saturation from upstream failure.
Metricsmcpproxy_tool_calls_rejected_total{server, reason="queue_full"|"queue_timeout"} + queue-depth gaugeEnvoy-style overflow counters; via the event-bus metrics bridge.
+

Stdio does not mean limit=1. mcp-go’s own stdio server runs tools/call through a 5-worker pool (queue 100); the client transport legally pipelines by JSON-RPC id; TS/Python SDKs process concurrently. Docs should recommend max_concurrent_requests: 5 for stdio upstreams, with 1 as the floor for fragile servers.

+ +

04Implementation plan

+
    +
  1. +

    Limiter package (pure, TDD-first)

    +
      +
    • internal/upstream/limiter/limiter.go + tests. Acquire(ctx, queueTimeout) (release func(), err): admission TryAcquire fail → ErrQueueFull; run Acquire under timeout fail → release admission, ErrQueueTimeout (wrap ctx.Err to distinguish caller-cancel). Release closure bound to this instance so hot-swaps can’t double-release. Zero-config limiter = no-op passthrough.
      • +
    • Registry: map[serverName]*atomic.Pointer[Limiter] + global pointer; Update() swaps atomically; in-flight releases drain the old instance harmlessly.
      • +
    +
    2. +
  2. +

    Config fields — follow the 4-point checklist exactly

    +
      +
    • Global max_concurrent_requests, queue_size, queue_timeout near CallToolTimeout (config.go:173); per-server tri-state pointers copying the HealthCheckInterval pattern (config.go:521-530). Defaults 0 / 0 / 30s. Validation: non-negative; queue_size>0 requires max>0.
      • +
    • Explicit DetectConfigChanges clause for the three global fields (config_hotreload.go:102-104 model) — per-server fields are already covered by the Servers DeepEqual.
      • +
    • Env overrides MCPPROXY_MAX_CONCURRENT_REQUESTS / _QUEUE_SIZE / _QUEUE_TIMEOUT (loader.go:638-690); then make swagger (CI verifies) and docs/configuration.md.
      • +
    +
    3. +
  3. +

    Enforcement at the choke point

    +
      +
    • Manager owns the Registry; builds per-server limiters in AddServer/ApplyConfig, global in SetGlobalConfig.
      • +
    • managed/client.go:640 CallTool: resolve effective values (per-server pointer → global default); acquire per-server, then global (a slow upstream’s queue can’t pin global slots); defer release() both; delegate to coreClient. Do not wrap ListTools or Ping. Reconnect-on-use runs before CallTool, so it never holds a slot.
      • +
    +
    4. +
  4. +

    Shed semantics + observability

    +
      +
    • Map ErrQueueFull/ErrQueueTimeout via errors.Is in mcp.go (isError tool result, early-error pattern at :2074-2076) and httpapi (429 + Retry-After).
      • +
    • Activity event status rejected; rejection counter + queue-depth gauge next to existing tool-call metrics (observability/metrics.go:114-129, bridge at observability_bridge.go:16-66). Follow-up: sustained saturation → degraded in the health calculator.
      • +
    +
    5. +
  5. +

    Tests + rollout

    +
      +
    • Unit: concurrent acquire/release, instant reject on full queue, queue-timeout, hot-swap under load, zero-config no-op. Hot-reload: extend global_config_hotreload_test.go.
      • +
    • E2E: slow stdio server with max=1/queue=1 — call 2 queues, call 3 sheds with isError; a code_execution case proving the bypass path is limited. go test -race, -tags server, ./scripts/test-api-e2e.sh.
      • +
    • Rollout: defaults 0/0 = pure opt-in, ships dark, no migration; hot-reload lets headless operators tune without restarts.
      • +
    +
    6. +
+ +

05Open decisions (maintainer input needed)

+
    +
  1. Defaults: keep 0 = unlimited everywhere (recommended, zero behavior change) vs an out-of-the-box stdio default (e.g. 5, mirroring mcp-go’s worker pool) that changes behavior on upgrade.
    2. +
  2. Protocol-level shed: is isError + REST 429 enough, or also emit a -32029 JSON-RPC error with data.retryAfter for non-tool methods (emerging convention; unverified how Claude Code/Cursor react mid-loop)?
    3. +
  3. Literal guarantee: should upstream tools/list and Ping ever count against a strict per-upstream limit for fragile single-threaded stdio servers (sketch excludes them)?
    4. +
  4. Activity vocabulary: new rejected status (touches Web UI filters, possibly telemetry schema) vs reusing error with a distinguishing code.
    5. +
  5. Per-user fairness (server edition): acceptable for v1 that one noisy user can occupy a FIFO queue, or reserve a (user,server) key shape in the Registry API now, ahead of Spec 074?
    6. +
  6. queue_timeout scope: per-server override for symmetry (as sketched) or global-only for a smaller config surface?
    7. +
  7. Health integration: saturation → degraded in v1 or follow-up?
    8. +
+ +

Method: multi-agent workflow (codebase choke-point audit + external prior-art research: Envoy, nginx, LiteLLM, MCP spec 2025-06-18, mcp-go v0.57.0 internals · 1.2M tokens). Companion report: macOS auto-updater for issue #957.

+
diff --git a/frontend/src/components/ActivityWidget.vue b/frontend/src/components/ActivityWidget.vue index 0b469d54a..11756f806 100644 --- a/frontend/src/components/ActivityWidget.vue +++ b/frontend/src/components/ActivityWidget.vue @@ -143,7 +143,9 @@ const getStatusBadgeClass = (status: string): string => { const statusClasses: Record = { 'success': 'badge-success', 'error': 'badge-error', - 'blocked': 'badge-warning' + 'blocked': 'badge-warning', + // Spec 093: shed by a concurrency limit (backpressure, not an upstream fault). + 'rejected': 'badge-info' } return statusClasses[status] || 'badge-ghost' } diff --git a/frontend/src/types/api.ts b/frontend/src/types/api.ts index 3f41539d8..3c317b2b6 100644 --- a/frontend/src/types/api.ts +++ b/frontend/src/types/api.ts @@ -511,6 +511,7 @@ export interface UsageToolStat { errors: number error_rate: number blocked: number + rejected: number // Spec 093: shed by a concurrency limit; never executed total_resp_bytes: number avg_resp_bytes: number | null // null when only legacy 0-byte calls exist total_req_bytes: number @@ -548,7 +549,7 @@ export interface UsageAggregateResponse { export type UsageWindow = '24h' | '7d' | 'all' export type UsageSort = 'calls' | 'resp_bytes' | 'error_rate' | 'p95' -export type UsageStatus = 'success' | 'error' | 'blocked' +export type UsageStatus = 'success' | 'error' | 'blocked' | 'rejected' export interface ToolCallRecord { id: string @@ -726,7 +727,17 @@ export type ActivityType = export type ActivitySource = 'mcp' | 'cli' | 'api' -export type ActivityStatus = 'success' | 'error' | 'blocked' +/** + * Closed activity-status vocabulary. 'rejected' (Spec 093) means the call was + * shed by a concurrency limit before it reached the upstream — proxy + * backpressure, not an upstream failure. + */ +export type ActivityStatus = 'success' | 'error' | 'blocked' | 'rejected' + +/** Spec 093: machine-readable cause of a rejection (activity metadata). */ +export type RejectionReason = 'queue_full' | 'queue_timeout' +/** Spec 093: which limiter tier shed the call (activity metadata). */ +export type RejectionScope = 'server' | 'global' export interface ActivityRecord { id: string @@ -780,6 +791,7 @@ export interface ActivitySummaryResponse { success_count: number error_count: number blocked_count: number + rejected_count: number top_servers?: ActivityTopServer[] top_tools?: ActivityTopTool[] start_time: string diff --git a/frontend/src/types/contracts.ts b/frontend/src/types/contracts.ts index 2d54819d3..da6b25662 100644 --- a/frontend/src/types/contracts.ts +++ b/frontend/src/types/contracts.ts @@ -47,6 +47,23 @@ export interface HealthStatus { action?: HealthAction; } +// Activity status vocabulary - generated from internal/storage/activity_models.go +export const ActivityStatusSuccess = 'success' as const; +export const ActivityStatusError = 'error' as const; +export const ActivityStatusBlocked = 'blocked' as const; +/** Spec 093: shed by a concurrency limit before reaching the upstream. */ +export const ActivityStatusRejected = 'rejected' as const; +export type ActivityStatusValue = + | typeof ActivityStatusSuccess + | typeof ActivityStatusError + | typeof ActivityStatusBlocked + | typeof ActivityStatusRejected; + +/** Machine-readable cause of a spec-093 rejection (activity metadata rejection_reason). */ +export type RejectionReason = 'queue_full' | 'queue_timeout'; +/** Limiter tier that shed the call (activity metadata rejection_scope). */ +export type RejectionScope = 'server' | 'global'; + export interface Server { id: string; name: string; @@ -80,6 +97,13 @@ export interface Server { health?: HealthStatus; // Unified health status calculated by the backend trust_mode?: string; // Per-server approval trust mode (spec 086): 'auto' | 'scan' | 'manual'; raw configured value, absent when unset (effective default: manual) security_scan?: SecurityScanSummary; // Latest scan summary (spec 086); ABSENT when no scan has ever run + // Spec 093 (#955) per-server concurrency overrides. Tri-state: absent = + // inherit server_concurrency_defaults, 0 = disabled for this server, + // positive = override. Effective concurrency is additionally bounded by the + // proxy-wide (global) limiter. + max_concurrent_requests?: number; + queue_size?: number; + queue_timeout?: string; // Go duration string, e.g. '30s' } export interface SecurityScanSummary { diff --git a/frontend/src/utils/activity.ts b/frontend/src/utils/activity.ts index 4b4613a9f..1f0e809d4 100644 --- a/frontend/src/utils/activity.ts +++ b/frontend/src/utils/activity.ts @@ -23,14 +23,18 @@ const typeIcons: Record = { const statusLabels: Record = { 'success': 'Success', 'error': 'Error', - 'blocked': 'Blocked' + 'blocked': 'Blocked', + // Spec 093: shed by a concurrency limit before reaching the upstream. + 'rejected': 'Rejected' } // Status badge CSS classes (DaisyUI) const statusClasses: Record = { 'success': 'badge-success', 'error': 'badge-error', - 'blocked': 'badge-warning' + 'blocked': 'badge-warning', + // Distinct from both error (upstream fault) and blocked (policy): backpressure. + 'rejected': 'badge-info' } // Intent operation type icons diff --git a/frontend/src/views/Activity.vue b/frontend/src/views/Activity.vue index d78eef939..c3c7754b3 100644 --- a/frontend/src/views/Activity.vue +++ b/frontend/src/views/Activity.vue @@ -72,6 +72,16 @@
Blocked
{{ summary.blocked_count }}
+ @@ -146,6 +156,7 @@ + @@ -1210,7 +1221,8 @@ const formatStatus = (status: string): string => { const statusLabels: Record = { 'success': 'Success', 'error': 'Error', - 'blocked': 'Blocked' + 'blocked': 'Blocked', + 'rejected': 'Rejected' } return statusLabels[status] || status } @@ -1219,7 +1231,9 @@ const getStatusBadgeClass = (status: string): string => { const statusClasses: Record = { 'success': 'badge-success', 'error': 'badge-error', - 'blocked': 'badge-warning' + 'blocked': 'badge-warning', + // Spec 093: shed by a concurrency limit (backpressure, not an upstream fault). + 'rejected': 'badge-info' } return statusClasses[status] || 'badge-ghost' } diff --git a/frontend/src/views/Usage.vue b/frontend/src/views/Usage.vue index f4d99425e..fe075f1b7 100644 --- a/frontend/src/views/Usage.vue +++ b/frontend/src/views/Usage.vue @@ -20,6 +20,7 @@ +