From 777d921c8b9cb46ee82b43a06d0a68219191847c Mon Sep 17 00:00:00 2001 From: Alkin Tezuysal Date: Mon, 2 Nov 2026 10:01:59 +0000 Subject: [PATCH] feat(mcp): component_mcp serves calls concurrently, with session limits - mcp_http.threads (read-only, default 4): I/O workers; each serves one call at a time (the spike had one). Router assigns connections to the workers in turn. - mcp_http.max_sessions_per_account (read-only, default 4): caps an account's concurrent endpoint sessions (error 1203), since the server's max_user_connections does not apply to command-service sessions. - Sessions are subject to max_connections (MYSQL_COMMAND_BYPASS_MAX_CONNECTIONS = false, 26.10) instead of bypassing it (error 1040 when full). test-mcp-endpoint.sh: 4 x SLEEP(2) overlap in under 4 s, cap 2 lets 2 of 4 through and refuses 2, max_connections = 1 with one idle client refuses the call. 40 checks; with the bypass left on, the max_connections check fails. Part of #256 (RFC 0001 Phase 5). Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 1 + docs/MCP_COMPONENTS.md | 8 ++-- scripts/test-mcp-endpoint.sh | 32 +++++++++++++ src/mcp/endpoint/component_mcp.cc | 77 ++++++++++++++++++++++++++++++- 4 files changed, 113 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fe7b149c..af550bca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- **`component_mcp` serves calls concurrently, with session limits** (RFC 0001 Phase 5, [#256](https://github.com/askdba/myvector/issues/256)). `mcp_http.threads` (default 4) I/O workers each serve one call at a time; the spike had one. `mcp_http.max_sessions_per_account` (default 4) caps an account's concurrent endpoint sessions (error 1203), since `max_user_connections` does not apply to them, and the sessions are now subject to `max_connections` (error 1040) instead of bypassing it. `scripts/test-mcp-endpoint.sh` checks overlap of 4 calls, the cap and `max_connections`. - **`component_mcp` (RFC 0001 Phase 4 spike, MySQL 26.10).** An MCP Streamable HTTP endpoint inside `mysqld` on MySQL Router's HTTP server (the libraries `component_mysql_rest_service` links), built in the server tree with `MYVECTOR_BUILD_MCP_ENDPOINT=ON ./scripts/build-component-26.10-docker.sh`. It is not for production. - **Endpoint:** loopback, plain HTTP, one bearer token (`mcp_http.token_sha256`) mapped to one account (`mcp_http.account`). - **Tools:** `list_databases` (tool pack) and `run_query` (one statement). diff --git a/docs/MCP_COMPONENTS.md b/docs/MCP_COMPONENTS.md index 4b112f96..245a21d8 100644 --- a/docs/MCP_COMPONENTS.md +++ b/docs/MCP_COMPONENTS.md @@ -192,7 +192,7 @@ An MCP **Streamable HTTP** endpoint (JSON-RPC 2.0 over `POST /mcp`) inside `mysq - **Tools:** `list_databases` (through the `mysql_mcp` tool pack) and `run_query` (one statement). - **Results:** `run_query` returns at most 1000 rows and 4 MB of serialized rows (escapes counted), as `{"rows": [{"col": value, ...}], "truncated": true|false}`. A value is a JSON string, `null`, or `{"base64": "..."}` when its bytes are not valid UTF-8 (binary data). The session's `sql_select_limit` stops the server at 1001 rows for statements without their own `LIMIT`. - **Known spike limit:** the command service's default consumer collects every row while a statement runs, so a statement with a large explicit `LIMIT` is buffered in full in `mysqld` before the reply is cut. Bounding that needs a custom row consumer (`mysql_command_consumer` services), which is Phase 5 work. -- **Missing:** the `account_locked` check, a per-account session cap and column masking. All are in Phase 5. +- **Missing:** the `account_locked` check and column masking. Both are in Phase 5. ### Build and install @@ -223,6 +223,8 @@ INSTALL COMPONENT 'file://component_mcp' | Variable | Meaning | |---|---| | `mcp_http.port` | TCP port on `127.0.0.1` (read-only, default 33480) | +| `mcp_http.threads` | I/O worker threads (read-only, default 4, 1 to 64). Each serves one call at a time, so this many calls run at once; connections are assigned to the workers in turn | +| `mcp_http.max_sessions_per_account` | Concurrent endpoint sessions per account (read-only, default 4, 1 to 1024). A call over the cap fails with error 1203. The server's `max_user_connections` does not apply to these sessions | | `mcp_http.account` | The account (`user@host`) every request runs as (read-only) | | `mcp_http.token_sha256` | Hex SHA-256 of the bearer token (read-only, not persisted, sensitive: readable only with `SENSITIVE_VARIABLES_OBSERVER`). Requests without the matching `Authorization: Bearer ` get 401 | @@ -240,7 +242,7 @@ So an account with `MCP_USER` is read-only through the endpoint, and its schema Facts recorded for Phase 5: - **Locked accounts:** impersonation skips the login checks, so a locked account (`ACCOUNT LOCK`) can be impersonated. The endpoint must check `account_locked` itself (26.10 has the lookup). -- **Connection limits:** `max_user_connections` is not enforced for these sessions, so the endpoint needs its own per-account cap. They do count against `max_connections`. +- **Connection limits:** `max_user_connections` is not enforced for these sessions, so the endpoint has its own per-account cap (`mcp_http.max_sessions_per_account`, Phase 5). They count against `max_connections`, and by default the command service lets them exceed it; the endpoint turns that off (`MYSQL_COMMAND_BYPASS_MAX_CONNECTIONS`, 26.10), so a call fails with error 1040 when the server is full. - **Auth backends:** Router's `http_auth_realm` / `http_auth_backend` support Basic auth only, so the endpoint checks bearer tokens itself (SHA-256, constant-time compare). ### Spike measurements (MySQL 26.10.0, arm64, 16 cores) @@ -249,7 +251,7 @@ Facts recorded for Phase 5: |---|---|---| | Build of the one target, fresh tree (after a 36 s configure) | 95 s | 265 s | | `.so` size | 3.1 MB | 21 MB | -| Threads added to `mysqld` (one I/O worker) | 3, released on uninstall | — | +| Threads added to `mysqld` (one I/O worker, the spike) | 3, released on uninstall | — | ### Tests diff --git a/scripts/test-mcp-endpoint.sh b/scripts/test-mcp-endpoint.sh index db9b1049..22889708 100755 --- a/scripts/test-mcp-endpoint.sh +++ b/scripts/test-mcp-endpoint.sh @@ -25,6 +25,15 @@ sql() { docker exec -i "$C" mysql -h127.0.0.1 -uroot -ptestpass -N -e "$1" 2>&1 rpc() { docker run --rm --network "container:$C" "$CURL_IMAGE" -s -m 20 -w '\n%{http_code}' -X POST \ http://127.0.0.1:33480/mcp -H 'Content-Type: application/json' ${1:+-H "Authorization: Bearer $1"} -d "$2"; } call() { rpc tok-123 "{\"jsonrpc\":\"2.0\",\"id\":9,\"method\":\"tools/call\",\"params\":{\"name\":\"$1\",\"arguments\":$2}}"; } +# par N SQL: N concurrent run_query calls from one sidecar; prints the replies (they can +# interleave, so count matches with grep -o), then the wall time in seconds as "elapsed=S". +par() { docker run --rm --network "container:$C" --entrypoint sh "$CURL_IMAGE" -c ' + n=$0; body=$1; start=$(date +%s); i=0 + while [ "$i" -lt "$n" ]; do + curl -s -m 30 -w "\n" -X POST http://127.0.0.1:33480/mcp -H "Content-Type: application/json" \ + -H "Authorization: Bearer tok-123" -d "$body" & i=$((i + 1)) + done; wait; echo "elapsed=$(( $(date +%s) - start ))"' "$1" \ + "{\"jsonrpc\":\"2.0\",\"id\":9,\"method\":\"tools/call\",\"params\":{\"name\":\"run_query\",\"arguments\":{\"sql\":\"$2\"}}}"; } alive() { docker ps -q -f "name=$C" | grep -q . && echo up || echo DOWN; } echo "== $IMAGE" @@ -68,6 +77,29 @@ expect "nothing written" "$(sql "SELECT GROUP_CONCAT(i) FROM d1.t")" "^1$" expect "guard counted the endpoint's statements" "$(sql "SHOW GLOBAL STATUS LIKE 'mcp.statements_denied'")" "mcp.statements_denied [1-9]" expect "no MCP session left" "$(sql "SHOW GLOBAL STATUS LIKE 'mcp.sessions_connected'")" " 0$" +# Concurrency: mcp_http.threads (default 4) I/O workers serve calls at once. +expect "threads and session cap defaults" "$(sql "SELECT @@mcp_http.threads, @@mcp_http.max_sessions_per_account")" "^4 4$" +four=$(par 4 "SELECT SLEEP(2) AS s") +expect "4 concurrent calls all succeed" "$(printf '%s\n' "$four" | grep -o '"isError":false' | wc -l | tr -d ' ')" "^4$" +expect "4 concurrent calls overlap (4 x 2 s in under 4 s)" "$(printf '%s\n' "$four" | grep elapsed=)" "^elapsed=[0-3]$" +# max_user_connections does not apply to these sessions; mcp_http.max_sessions_per_account does. +sql "UNINSTALL COMPONENT 'file://component_mcp'" >/dev/null +expect "reinstall with a session cap of 2" "$(sql "INSTALL COMPONENT 'file://component_mcp' SET GLOBAL mcp_http.port = 33480, GLOBAL mcp_http.max_sessions_per_account = 2, GLOBAL mcp_http.account = 'mcp@%', GLOBAL mcp_http.token_sha256 = '$TOKEN_HEX'")" "^$" +capped=$(par 4 "SELECT SLEEP(2) AS s") +expect "cap 2: two of four concurrent calls succeed" "$(printf '%s\n' "$capped" | grep -o '"isError":false' | wc -l | tr -d ' ')" "^2$" +expect "cap 2: the others are refused (1203)" "$(printf '%s\n' "$capped" | grep -o "Error 1203: User mcp already has 2 active endpoint sessions (mcp_http.max_sessions_per_account)" | wc -l | tr -d ' ')" "^2$" +expect "cap 2: a session slot is free again afterwards" "$(call run_query '{"sql":"SELECT 1 AS one"}')" '"isError":false' +# The sessions count against max_connections and, unlike the command service's default, are +# subject to it: one idle client fills max_connections = 1. +sql "SET GLOBAL max_connections = 1" +sql "SELECT SLEEP(4)" >/dev/null & +sleep 1.5 +expect "max_connections applies to endpoint sessions" "$(call run_query '{"sql":"SELECT 1 AS one"}')" 'Error 1040: Too many connections.*"isError":true' +wait +sql "SET GLOBAL max_connections = 151" +expect "a call works again below max_connections" "$(call run_query '{"sql":"SELECT 1 AS one"}')" '"isError":false' +expect "no MCP session left after concurrent calls" "$(sql "SHOW GLOBAL STATUS LIKE 'mcp.sessions_connected'")" " 0$" + # An account without MCP_USER is not treated as an MCP session (the token maps to app). sql "UNINSTALL COMPONENT 'file://component_mcp'" >/dev/null expect "reinstall mapped to app" "$(sql "INSTALL COMPONENT 'file://component_mcp' SET GLOBAL mcp_http.port = 33480, GLOBAL mcp_http.account = 'app@%', GLOBAL mcp_http.token_sha256 = '$TOKEN_HEX'")" "^$" diff --git a/src/mcp/endpoint/component_mcp.cc b/src/mcp/endpoint/component_mcp.cc index 1854d2da..41998672 100644 --- a/src/mcp/endpoint/component_mcp.cc +++ b/src/mcp/endpoint/component_mcp.cc @@ -46,7 +46,9 @@ #include #include +#include #include +#include #include #include #include @@ -88,9 +90,12 @@ namespace { // ----- settings ------------------------------------------------------------------------- unsigned long opt_port = 33480; +unsigned long opt_threads = 4; // I/O workers: calls served at once +unsigned long opt_max_sessions_per_account = 4; // concurrent calls per account char *opt_account = nullptr; // user@host the token maps to char *opt_token_sha256 = nullptr; // hex SHA-256 of the bearer token -bool port_registered = false, account_registered = false, token_registered = false; +bool port_registered = false, threads_registered = false, max_sessions_registered = false, + account_registered = false, token_registered = false; bool register_sysvars() { INTEGRAL_CHECK_ARG(ulong) port_arg; @@ -103,6 +108,33 @@ bool register_sysvars() { "TCP port of the MCP endpoint (loopback only)", nullptr, nullptr, &port_arg, &opt_port)) return true; port_registered = true; + { + INTEGRAL_CHECK_ARG(ulong) threads_arg; + threads_arg.def_val = 4; + threads_arg.min_val = 1; + threads_arg.max_val = 64; + threads_arg.blk_sz = 0; + if (mysql_service_component_sys_variable_register->register_variable( + "mcp_http", "threads", PLUGIN_VAR_LONG | PLUGIN_VAR_UNSIGNED | PLUGIN_VAR_READONLY, + "I/O worker threads; each serves one call at a time", nullptr, nullptr, &threads_arg, + &opt_threads)) + return true; + threads_registered = true; + } + { + INTEGRAL_CHECK_ARG(ulong) max_sessions_arg; + max_sessions_arg.def_val = 4; + max_sessions_arg.min_val = 1; + max_sessions_arg.max_val = 1024; + max_sessions_arg.blk_sz = 0; + if (mysql_service_component_sys_variable_register->register_variable( + "mcp_http", "max_sessions_per_account", + PLUGIN_VAR_LONG | PLUGIN_VAR_UNSIGNED | PLUGIN_VAR_READONLY, + "Concurrent endpoint sessions per account (max_user_connections does not apply)", + nullptr, nullptr, &max_sessions_arg, &opt_max_sessions_per_account)) + return true; + max_sessions_registered = true; + } { STR_CHECK_ARG(str) account_arg; account_arg.def_val = nullptr; @@ -140,6 +172,8 @@ bool unregister_sysvars() { }; drop(token_registered, "token_sha256"); drop(account_registered, "account"); + drop(max_sessions_registered, "max_sessions_per_account"); + drop(threads_registered, "threads"); drop(port_registered, "port"); return failed; } @@ -266,6 +300,34 @@ void read_error(MYSQL_H h, SqlResult *r) { r->error = msg != nullptr ? msg : ""; } +// Per-account count of open endpoint sessions. The server does not apply max_user_connections +// to command-service sessions, so mcp_http.max_sessions_per_account caps them here. +class SessionSlot { + public: + explicit SessionSlot(const std::string &account) : account_(account) { + std::lock_guard lock(mutex_); + unsigned long &n = open_[account_]; + if (n >= opt_max_sessions_per_account) return; + ++n; + held_ = true; + } + ~SessionSlot() { + if (!held_) return; + std::lock_guard lock(mutex_); + auto it = open_.find(account_); + if (--it->second == 0) open_.erase(it); + } + SessionSlot(const SessionSlot &) = delete; + SessionSlot &operator=(const SessionSlot &) = delete; + bool held() const { return held_; } + + private: + static inline std::mutex mutex_; + static inline std::map open_; + std::string account_; + bool held_ = false; +}; + // Runs one statement as account (user@host) and returns its first result set. SqlResult run_as(const std::string &account, const std::string &sql) { SqlResult r; @@ -275,6 +337,13 @@ SqlResult run_as(const std::string &account, const std::string &sql) { return r; } const std::string user = account.substr(0, at), host = account.substr(at + 1); + SessionSlot slot(account); + if (!slot.held()) { + r.sql_errno = ER_TOO_MANY_USER_CONNECTIONS; + r.error = "User " + user + " already has " + std::to_string(opt_max_sessions_per_account) + + " active endpoint sessions (mcp_http.max_sessions_per_account)"; + return r; + } MYSQL_H h = nullptr; if (mysql_service_mysql_command_factory->init(&h) || h == nullptr) { r.error = "cannot create a session"; @@ -284,10 +353,14 @@ SqlResult run_as(const std::string &account, const std::string &sql) { // Client protocol capabilities (mysql_com.h, which components cannot include). constexpr uint32_t kClientMultiResults = 1UL << 17, kClientPsMultiResults = 1UL << 18; const uint32_t client_flags = kClientMultiResults | kClientPsMultiResults; + // Unlike the command service's default, the sessions are subject to max_connections. + static constexpr bool kBypassMaxConnections = false; if (mysql_service_mysql_command_options->set(h, MYSQL_COMMAND_PROTOCOL, "local") || mysql_service_mysql_command_options->set(h, MYSQL_COMMAND_CLIENT_FLAGS, &client_flags) || mysql_service_mysql_command_options->set(h, MYSQL_COMMAND_USER_NAME, user.c_str()) || mysql_service_mysql_command_options->set(h, MYSQL_COMMAND_HOST_NAME, host.c_str()) || + mysql_service_mysql_command_options->set(h, MYSQL_COMMAND_BYPASS_MAX_CONNECTIONS, + &kBypassMaxConnections) || mysql_service_mysql_command_factory->connect(h)) { read_error(h, &r); if (r.error.empty()) r.error = "cannot open a session for " + account; @@ -537,7 +610,7 @@ bool start_server() { // HTTP work runs on the io threads, which run SQL through the command services. io.set_thread_lifecycle({[](size_t) { mysql_service_mysql_command_thread->init(); }, [](size_t) { mysql_service_mysql_command_thread->end(); }}); - if (!io.init(1, IoBackend::preferred())) return true; + if (!io.init(opt_threads, IoBackend::preferred())) return true; io_initialized = true; try { server = std::make_shared(