[Server] Add an opt-in per-session lock for concurrent requests

CHANGELOG.md

@@ -74,6 +74,10 @@ All notable changes to `mcp/sdk` will be documented in this file.
 * Fix a client request that timed out (elicitation, sampling) staying in the session's pending requests: `Protocol::checkResponse()` now expires it, answering with the `Request timed out` error, so the transports no longer implement the timeout themselves.
 * Fix a handshake-era Streamable HTTP stream pinning its PHP worker until the elicitation (or other client request) times out after the client disconnected: the idle stream now sends a keep-alive comment and stops once the connection is aborted.
 * [BC Break] Fix concurrent stdio tool calls that report progress, log, elicit or sample never finishing: `StdioTransport` keeps every suspended fiber and resumes each with the answer to its own request. A transport passes the suspended `\Fiber` as last argument to the pending-requests provider and the fiber yield handler.
+* Add an opt-in per-session lock, `Builder::setSessionLock()`, serializing the concurrent requests of one session so none of them overwrites what another saved: pending server requests, client answers, client info. `FileSessionLock` locks with `flock()` next to `FileSessionStore`, `SymfonyLockSessionLock` uses a symfony/lock store for sessions shared across machines; `SessionLockInterface` takes custom ones. Off by default. A request that cannot get the lock in time is answered with `503` and a JSON-RPC error; a failing lock backend with `503` and `The session lock is unavailable.`. `SessionLockTimeoutException`, a `SessionLockException`, tells a timeout from a backend failure. `SymfonyLockSessionLock` releases its locks from a shutdown function after a fatal error in a handler.
+* `Protocol::__construct()` takes an optional `SessionLockInterface $sessionLock` as its last argument.
+* `FileSessionStore::gc()` also removes the lock files of expired sessions a `FileSessionLock` in the same directory left behind, each only while no request holds it.
+* Resume a suspended handler with an error when the transport fails to store what it sends to the client next, instead of resuming it with `null`; `ClientGateway::notify()` then throws a `ClientException`.
 
 0.8.1
 -----

composer.json

@@ -38,7 +38,8 @@
   "suggest": {
     "firebase/php-jwt": "Required for the JwtTokenValidator.",
     "psr/cache-implementation": "Required for JwtTokenValidator::fromIssuer() to cache metadata and keys.",
-    "symfony/finder": "Required for file-based discovery."
+    "symfony/finder": "Required for file-based discovery.",
+    "symfony/lock": "Required for the SymfonyLockSessionLock."
   },
   "require-dev": {
     "ext-openssl": "*",
@@ -58,6 +59,7 @@
     "symfony/console": "^5.4 || ^6.4 || ^7.3 || ^8.0",
     "symfony/finder": "^5.4 || ^6.4 || ^7.3 || ^8.0",
     "symfony/http-client": "^5.4 || ^6.4 || ^7.3 || ^8.0",
+    "symfony/lock": "^5.4 || ^6.4 || ^7.3 || ^8.0",
     "symfony/process": "^5.4 || ^6.4 || ^7.3 || ^8.0"
   },
   "conflict": {

docs/run/server-builder.md

@@ -340,6 +340,7 @@ $server = Server::builder()
 | `setDiscovery()` | basePath, scanDirs?, excludeDirs?, cache?, namePatterns? | Configure attribute discovery |
 | `setDiscoverer()` | discoverer | Replace the class that scans for attributes (the discovery `cache` is then not used) |
 | `setSession()` | sessionStore?, sessionManager?, gcProbability?, gcDivisor? | Configure session management, see [Sessions](sessions.md) |
+| `setSessionLock()` | sessionLock | Serialize the concurrent requests of one session, see [Sessions](sessions.md#concurrent-requests) |
 | `setLogger()` | logger | Set PSR-3 logger |
 | `setContainer()` | container | Set PSR-11 container |
 | `setEventDispatcher()` | eventDispatcher | Set PSR-14 event dispatcher, see [Events](../advanced/events.md) |

docs/run/sessions.md

@@ -89,6 +89,78 @@ $server = Server::builder()
 > **Note**: When providing a custom `SessionManagerInterface` via the `$sessionManager` parameter,
 > the `gcProbability` and `gcDivisor` settings are ignored — you control GC behavior in your own implementation.
 
+## Concurrent Requests
+
+Over HTTP, two requests of one session can run at the same time on different workers: `tools/list` and
+`resources/list` sent together on connect, or parallel tool calls from an agent. Each request loads the
+session, changes it and writes it back whole, so the last write wins and the other request's changes are
+gone: a pending request to the client, the client's answer to one, the client info. Responses do not go
+through the session and are not affected.
+
+A session lock serializes the requests of one session. It is off by default; enable it with `setSessionLock()`
+and pair it with the store:
+
+```php
+use Mcp\Server\Session\FileSessionLock;
+use Mcp\Server\Session\FileSessionStore;
+
+// flock() on one file per session, for the workers of one machine
+$server = Server::builder()
+    ->setSession(new FileSessionStore(__DIR__ . '/sessions'))
+    ->setSessionLock(new FileSessionLock(__DIR__ . '/sessions'))
+    ->build();
+```
+
+Point `FileSessionLock` at the session directory: `FileSessionStore::gc()` then removes the lock files of
+expired sessions along with them. It never removes the file of a lock a request still holds.
+
+For a store shared across machines, use `SymfonyLockSessionLock` with any
+[symfony/lock](https://symfony.com/doc/current/components/lock.html) store (`composer require symfony/lock`):
+
+```php
+use Mcp\Server\Session\Psr16SessionStore;
+use Mcp\Server\Session\SymfonyLockSessionLock;
+use Symfony\Component\Cache\Adapter\RedisAdapter;
+use Symfony\Component\Cache\Psr16Cache;
+use Symfony\Component\Lock\LockFactory;
+use Symfony\Component\Lock\Store\RedisStore;
+
+$redis = RedisAdapter::createConnection('redis://localhost:6379');
+
+$server = Server::builder()
+    ->setSession(new Psr16SessionStore(new Psr16Cache(new RedisAdapter($redis, 'mcp_sessions'))))
+    ->setSessionLock(new SymfonyLockSessionLock(new LockFactory(new RedisStore($redis))))
+    ->build();
+```
+
+The lock is held from before a request reads its session until it saved it, so while its handlers run. It
+is released when a handler suspends to wait for the client (elicitation, sampling), so the answer's own
+request can be processed. The stream that waits for the answer takes the lock briefly on each poll, and only
+when it is free; otherwise it tries again on its next poll. Parallel tool calls of one session therefore run
+one after the other.
+
+A request that cannot get the lock within the timeout (30 seconds by default, the constructors' `$timeout`)
+is answered with HTTP `503` and a JSON-RPC error, instead of running on a session another request is about
+to overwrite. A lock backend that fails, e.g. a Redis that refuses connections, is answered with `503` too,
+but at once, with the error message `The session lock is unavailable.`, and logged as an error with its cause;
+a timeout is only logged as a warning. A handler that resumed after the client answered gets an error when the lock cannot be had to
+store its next request or notification: the `ClientGateway` method throws a `ClientException`, and the
+message does not reach the client.
+
+`SymfonyLockSessionLock` expires a held lock after `$ttl` (300 seconds by default), so a worker that dies
+holding one does not block the session for good. The lease is not refreshed while a handler runs, so `$ttl`
+must be longer than your longest handler runs; otherwise another request can take the lock before the first
+one saved the session. A fatal error in a handler (memory limit, `max_execution_time`) skips the normal
+release, so `SymfonyLockSessionLock` also releases its locks from a shutdown function; `FileSessionLock` needs
+none, as the operating system drops a `flock()` with the process.
+
+**Custom Session Locks:**
+
+Implement `SessionLockInterface` for another lock backend. `acquire()` blocks until the lock is held or
+throws `SessionLockTimeoutException` after the backend's timeout, and `SessionLockException` when the backend
+fails; `tryAcquire()` takes the lock only if it is free
+right now and returns `false` otherwise; `release()` gives it back.
+
 **Available Session Stores:**
 
 - `InMemorySessionStore`: Fast in-memory storage (default)

src/Exception/SessionLockException.php

@@ -0,0 +1,21 @@
+<?php
+
+/*
+ * This file is part of the official PHP MCP SDK.
+ *
+ * A collaboration between Symfony and the PHP Foundation.
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Mcp\Exception;
+
+/**
+ * A session lock could not be acquired or released.
+ *
+ * @author Vitalii Cherepanov <vbcherepanov@gmail.com>
+ */
+class SessionLockException extends \RuntimeException implements ExceptionInterface
+{
+}

src/Exception/SessionLockTimeoutException.php

@@ -0,0 +1,23 @@
+<?php
+
+/*
+ * This file is part of the official PHP MCP SDK.
+ *
+ * A collaboration between Symfony and the PHP Foundation.
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Mcp\Exception;
+
+/**
+ * Another request held the session lock for longer than the timeout.
+ *
+ * Unlike its parent, this means the lock backend works and the session is busy.
+ *
+ * @author Vitalii Cherepanov <vbcherepanov@gmail.com>
+ */
+final class SessionLockTimeoutException extends SessionLockException
+{
+}

src/Server/Builder.php

@@ -54,6 +54,7 @@
 use Mcp\Server\Resource\SessionSubscriptionManager;
 use Mcp\Server\Resource\SubscriptionManagerInterface;
 use Mcp\Server\Session\InMemorySessionStore;
+use Mcp\Server\Session\SessionLockInterface;
 use Mcp\Server\Session\SessionManager;
 use Mcp\Server\Session\SessionManagerInterface;
 use Mcp\Server\Session\SessionStoreInterface;
@@ -116,6 +117,8 @@ final class Builder
 
     private ?SessionStoreInterface $sessionStore = null;
 
+    private ?SessionLockInterface $sessionLock = null;
+
     private int $gcProbability = 1;
 
     private int $gcDivisor = 100;
@@ -637,6 +640,20 @@ public function setSession(
         return $this;
     }
 
+    /**
+     * Serializes the requests of one session, so two of them running at the same time
+     * cannot overwrite each other's changes to the session. Off by default.
+     *
+     * Pair the lock with the store: {@see Session\FileSessionLock} for {@see Session\FileSessionStore},
+     * {@see Session\SymfonyLockSessionLock} for a store shared across machines like {@see Session\Psr16SessionStore}.
+     */
+    public function setSessionLock(SessionLockInterface $sessionLock): self
+    {
+        $this->sessionLock = $sessionLock;
+
+        return $this;
+    }
+
     /**
      * @param string[] $scanDirs
      * @param string[] $excludeDirs
@@ -955,6 +972,7 @@ public function build(): Server
                 ? new InputRequiredShim($this->inputRequiredRounds, $this->inputRequiredTimeout, $parts['logger'])
                 : null,
             requestStateCodec: $this->requestStateCodec(),
+            sessionLock: $this->sessionLock,
         );
 
         $modernVersions = $this->modernVersions ?? ProtocolVersion::modernVersions();

src/Server/ClientGateway.php

@@ -90,10 +90,16 @@ public function __construct(
      *
      * This suspends the Fiber to let the transport flush the notification via SSE,
      * then immediately resumes execution.
+     *
+     * @throws ClientException if the notification could not be queued for the client
      */
     public function notify(Notification $notification): void
     {
-        \Fiber::suspend(new NotificationSuspension($notification, $this->session->getId()->toRfc4122()));
+        $result = \Fiber::suspend(new NotificationSuspension($notification, $this->session->getId()->toRfc4122()));
+
+        if ($result instanceof Error) {
+            throw new ClientException($result);
+        }
     }
 
     /**