diff --git a/CHANGELOG.md b/CHANGELOG.md index 64fb49a6..8ebbe7b8 100644 --- a/CHANGELOG.md +++ b/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 ----- diff --git a/composer.json b/composer.json index fd3ce3d0..be233eee 100644 --- a/composer.json +++ b/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": { diff --git a/docs/run/server-builder.md b/docs/run/server-builder.md index 9181820f..b1fb6a16 100644 --- a/docs/run/server-builder.md +++ b/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) | diff --git a/docs/run/sessions.md b/docs/run/sessions.md index 200869db..db3c3cf3 100644 --- a/docs/run/sessions.md +++ b/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) diff --git a/src/Exception/SessionLockException.php b/src/Exception/SessionLockException.php new file mode 100644 index 00000000..06e91af3 --- /dev/null +++ b/src/Exception/SessionLockException.php @@ -0,0 +1,21 @@ + + */ +class SessionLockException extends \RuntimeException implements ExceptionInterface +{ +} diff --git a/src/Exception/SessionLockTimeoutException.php b/src/Exception/SessionLockTimeoutException.php new file mode 100644 index 00000000..c8863068 --- /dev/null +++ b/src/Exception/SessionLockTimeoutException.php @@ -0,0 +1,23 @@ + + */ +final class SessionLockTimeoutException extends SessionLockException +{ +} diff --git a/src/Server/Builder.php b/src/Server/Builder.php index f1dd6d39..6eadb805 100644 --- a/src/Server/Builder.php +++ b/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(); diff --git a/src/Server/ClientGateway.php b/src/Server/ClientGateway.php index 4d8a144c..f37b3ef0 100644 --- a/src/Server/ClientGateway.php +++ b/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); + } } /** diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index 90f37bd7..584b3fa8 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -19,6 +19,8 @@ use Mcp\Event\ServerRequestEvent; use Mcp\Exception\InvalidInputMessageException; use Mcp\Exception\RuntimeException; +use Mcp\Exception\SessionLockException; +use Mcp\Exception\SessionLockTimeoutException; use Mcp\Exception\SessionStoreException; use Mcp\JsonRpc\MessageFactory; use Mcp\Schema\JsonRpc\Error; @@ -31,6 +33,7 @@ use Mcp\Server\Handler\Notification\NotificationHandlerInterface; use Mcp\Server\Handler\Request\RequestHandlerInterface; use Mcp\Server\Session\SessionInterface; +use Mcp\Server\Session\SessionLockInterface; use Mcp\Server\Session\SessionManagerInterface; use Mcp\Server\Stateless\InputContext; use Mcp\Server\Stateless\RequestStateCodec; @@ -76,6 +79,10 @@ class Protocol */ private const INTERNAL_ERROR_MESSAGE = 'Internal server error.'; + /** Answer to a request that could not get its session's lock in time, see {@see SessionLockInterface}. */ + private const SESSION_BUSY_MESSAGE = 'Another request of this session is still being processed.'; + private const SESSION_LOCK_UNAVAILABLE_MESSAGE = 'The session lock is unavailable.'; + /** * The client request each suspended fiber waits on. Pending requests are stored in the session, * which concurrent calls share, so a fiber must only be resumed with the answer to the one it sent. @@ -97,6 +104,7 @@ public function __construct( private readonly ?EventDispatcherInterface $eventDispatcher = null, private readonly ?InputRequiredShim $inputRequiredShim = null, private readonly ?RequestStateCodec $requestStateCodec = null, + private readonly ?SessionLockInterface $sessionLock = null, ) { $this->awaitedRequestIds = new \WeakMap(); } @@ -162,8 +170,10 @@ public function processInput(TransportInterface $transport, string $input, ?Uuid * Determines the id an unprocessable input has to be answered under. * * Returns null when the input carries no request at all, in which case it consists of - * notifications only and JSON-RPC forbids answering it. A batch resolves to the empty id - * because its failure cannot be attributed to one of its requests. + * notifications and client responses only, and JSON-RPC forbids answering it: an error + * under a response's id would read as the answer to the client's own request of that id. + * A batch resolves to the empty id because its failure cannot be attributed to one of + * its requests. */ private static function findResponseId(string $input): string|int|null { @@ -180,11 +190,11 @@ private static function findResponseId(string $input): string|int|null if (!array_is_list($data)) { $id = $data['id'] ?? null; - return \is_string($id) || \is_int($id) ? $id : null; + return isset($data['method']) && (\is_string($id) || \is_int($id)) ? $id : null; } foreach ($data as $message) { - if (\is_array($message) && isset($message['id'])) { + if (\is_array($message) && isset($message['id'], $message['method'])) { return ''; } } @@ -212,6 +222,42 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U return; } + // Taken before the session is read and held until it is saved, so no other request of + // the session reads a state this one is about to replace. A session created here for + // an `initialize` is not known to anyone else yet and needs no lock, and an unknown + // session id is refused by resolveSession() without one; the session is looked up + // again under the lock, as it may have ended while this request waited. + $takesLock = null !== $sessionId && null !== $this->sessionLock && $this->sessionManager->exists($sessionId); + + if ($takesLock) { + try { + $refusal = $this->acquireSessionLock($sessionId) ? null : self::SESSION_BUSY_MESSAGE; + } catch (SessionLockException) { + $refusal = self::SESSION_LOCK_UNAVAILABLE_MESSAGE; + } + + if (null !== $refusal) { + $this->sendResponse($transport, Error::forServerError($refusal, self::findResponseId($input)), null, ['status_code' => 503]); + + return; + } + } + + try { + $this->handleMessages($transport, $sessionId, $messages, $accessToken); + } finally { + if ($takesLock) { + $this->releaseSessionLock($sessionId); + } + } + } + + /** + * @param TransportInterface $transport + * @param array $messages + */ + private function handleMessages(TransportInterface $transport, ?Uuid $sessionId, array $messages, ?AccessToken $accessToken): void + { $session = $this->resolveSession($transport, $sessionId, $messages); if (null === $session) { return; @@ -570,16 +616,24 @@ private function queueOutgoing(Request|Notification $message, array $context, Se */ public function consumeOutgoingMessages(Uuid $sessionId): array { - $session = $this->sessionManager->createWithId($sessionId); - $queue = $session->get(self::SESSION_OUTGOING_QUEUE, []); - - // Saving an unchanged session would only overwrite what a concurrent request saved in the meantime. - if ([] !== $queue) { - $session->set(self::SESSION_OUTGOING_QUEUE, []); - $session->save(); + if (!$this->tryAcquireSessionLockForPoll($sessionId)) { + return []; } - return $queue; + try { + $session = $this->sessionManager->createWithId($sessionId); + $queue = $session->get(self::SESSION_OUTGOING_QUEUE, []); + + // Saving an unchanged session would only overwrite what a concurrent request saved in the meantime. + if ([] !== $queue) { + $session->set(self::SESSION_OUTGOING_QUEUE, []); + $session->save(); + } + + return $queue; + } finally { + $this->releaseSessionLock($sessionId); + } } /** @@ -594,23 +648,31 @@ public function consumeOutgoingMessages(Uuid $sessionId): array */ public function checkResponse(int $requestId, Uuid $sessionId): Response|Error|null { - $session = $this->sessionManager->createWithId($sessionId); - $responseData = $session->get(self::SESSION_RESPONSES.".{$requestId}"); + if (!$this->tryAcquireSessionLockForPoll($sessionId)) { + return null; + } - if (null === $responseData) { - $pending = $session->get(self::SESSION_PENDING_REQUESTS, [])[$requestId] ?? null; + try { + $session = $this->sessionManager->createWithId($sessionId); + $responseData = $session->get(self::SESSION_RESPONSES.".{$requestId}"); - return \is_array($pending) && $this->hasTimedOut($pending) ? $this->expireRequest($requestId, $sessionId) : null; - } + if (null === $responseData) { + $pending = $session->get(self::SESSION_PENDING_REQUESTS, [])[$requestId] ?? null; - $this->logger->debug('Found and consuming client response.', [ - 'request_id' => $requestId, - 'session_id' => $sessionId->toRfc4122(), - ]); + return \is_array($pending) && $this->hasTimedOut($pending) ? $this->expireRequest($requestId, $sessionId) : null; + } - $session->set(self::SESSION_RESPONSES.".{$requestId}", null); - $session->forget(self::SESSION_PENDING_REQUESTS.".{$requestId}"); - $session->save(); + $this->logger->debug('Found and consuming client response.', [ + 'request_id' => $requestId, + 'session_id' => $sessionId->toRfc4122(), + ]); + + $session->set(self::SESSION_RESPONSES.".{$requestId}", null); + $session->forget(self::SESSION_PENDING_REQUESTS.".{$requestId}"); + $session->save(); + } finally { + $this->releaseSessionLock($sessionId); + } try { if (isset($responseData['error'])) { @@ -700,17 +762,6 @@ public function handleFiberYield(mixed $yieldedValue, ?Uuid $sessionId): ?int return null; } - // A session destroyed while the request ran, e.g. by a DELETE, must not be saved back. - if (!$this->sessionManager->exists($sessionId)) { - $this->logger->warning('Fiber yielded for a session that no longer exists.', [ - 'session_id' => $sessionId->toRfc4122(), - ]); - - return null; - } - - $session = $this->sessionManager->createWithId($sessionId); - if ($yieldedValue->sessionId !== $sessionId->toRfc4122()) { $this->logger->warning('Fiber yielded payload with mismatched session ID.', [ 'payload_session_id' => $yieldedValue->sessionId, @@ -718,14 +769,33 @@ public function handleFiberYield(mixed $yieldedValue, ?Uuid $sessionId): ?int ]); } + // The message the fiber suspended on must be stored, so this waits for the lock; when it + // cannot be had, the exception makes the transport resume the fiber with an error. + $this->sessionLock?->acquire($sessionId); + try { - if ($yieldedValue instanceof RequestSuspension) { - return $this->sendRequest($yieldedValue->request, $yieldedValue->timeout, $session); + // A session destroyed while the request ran, e.g. by a DELETE, must not be saved back. + if (!$this->sessionManager->exists($sessionId)) { + $this->logger->warning('Fiber yielded for a session that no longer exists.', [ + 'session_id' => $sessionId->toRfc4122(), + ]); + + return null; } - $this->sendNotification($yieldedValue->notification, $session); + $session = $this->sessionManager->createWithId($sessionId); + + try { + if ($yieldedValue instanceof RequestSuspension) { + return $this->sendRequest($yieldedValue->request, $yieldedValue->timeout, $session); + } + + $this->sendNotification($yieldedValue->notification, $session); + } finally { + $session->save(); + } } finally { - $session->save(); + $this->releaseSessionLock($sessionId); } return null; @@ -901,7 +971,91 @@ private static function findRequestId(array $messages): string|int|null */ public function destroySession(Uuid $sessionId): void { - $this->sessionManager->destroy($sessionId); + // Destroyed even when a request of the session holds the lock past the timeout, or the + // lock is unavailable: the client ended the session, and Session::save() does not write + // back a session that was destroyed since it was read. An unknown session id has nothing to lock. + try { + $locked = null !== $this->sessionLock && $this->sessionManager->exists($sessionId) && $this->acquireSessionLock($sessionId); + } catch (SessionLockException) { + $locked = false; + } + + try { + $this->sessionManager->destroy($sessionId); + } finally { + if ($locked) { + $this->releaseSessionLock($sessionId); + } + } + $this->logger->info('Session destroyed.', ['session_id' => $sessionId->toRfc4122()]); } + + /** + * Returns false when another request holds the lock past the timeout, or right now when not waiting; + * true when it was acquired, or there is no lock. + * + * @param bool $wait false to only take a lock that is free right now + * + * @throws SessionLockException when the lock backend failed + */ + private function acquireSessionLock(Uuid $sessionId, bool $wait = true): bool + { + if (null === $this->sessionLock) { + return true; + } + + try { + if (!$wait) { + return $this->sessionLock->tryAcquire($sessionId); + } + + $this->sessionLock->acquire($sessionId); + } catch (SessionLockTimeoutException $e) { + $this->logger->warning(\sprintf('Failed to acquire session lock: %s', $e->getMessage()), [ + 'session_id' => $sessionId->toRfc4122(), + 'exception' => $e, + ]); + + return false; + } catch (SessionLockException $e) { + $this->logger->error(\sprintf('Session lock is unavailable: %s', $e->getMessage()), [ + 'session_id' => $sessionId->toRfc4122(), + 'exception' => $e, + ]); + + throw $e; + } + + return true; + } + + /** + * A stream polls again on its next turn, so neither a busy lock nor a failing backend is waited on here. + */ + private function tryAcquireSessionLockForPoll(Uuid $sessionId): bool + { + try { + return $this->acquireSessionLock($sessionId, wait: false); + } catch (SessionLockException) { + return false; + } + } + + private function releaseSessionLock(Uuid $sessionId): void + { + if (null === $this->sessionLock) { + return; + } + + try { + $this->sessionLock->release($sessionId); + } catch (SessionLockException $e) { + // The work is done; a stuck lock is logged for the operator and expires or times out for the next request. + $this->logger->error(\sprintf('Failed to release session lock: %s', $e->getMessage()), [ + 'session_id' => $sessionId->toRfc4122(), + 'exception' => $e, + ]); + } + } } diff --git a/src/Server/Session/FileSessionLock.php b/src/Server/Session/FileSessionLock.php new file mode 100644 index 00000000..38c04af7 --- /dev/null +++ b/src/Server/Session/FileSessionLock.php @@ -0,0 +1,176 @@ + + */ +final class FileSessionLock implements SessionLockInterface +{ + public const FILE_SUFFIX = '.lock'; + + private const RETRY_DELAY_MICROSECONDS = 10_000; + + /** @var array */ + private array $handles = []; + + /** + * @param float $timeout how long to wait for a lock held by another request, in seconds + */ + public function __construct( + private readonly string $directory, + private readonly float $timeout = 30.0, + private readonly LoggerInterface $logger = new NullLogger(), + ) { + if ($this->timeout < 0) { + throw new InvalidArgumentException('timeout must be greater than or equal to 0.'); + } + + if (!is_dir($this->directory) && !@mkdir($this->directory, 0775, true) && !is_dir($this->directory)) { + $this->logger->warning('Failed to create session lock directory.', [ + 'directory' => $this->directory, + 'error' => error_get_last()['message'] ?? 'unknown', + ]); + } + + if (!is_dir($this->directory) || !is_writable($this->directory)) { + throw new RuntimeException(\sprintf('Session lock directory "%s" is not writable.', $this->directory)); + } + } + + public function acquire(Uuid $sessionId): void + { + if (!$this->lock($sessionId, $this->timeout)) { + throw new SessionLockTimeoutException(\sprintf('Timed out after %.3fs waiting for the lock of session "%s".', $this->timeout, $sessionId->toRfc4122())); + } + } + + public function tryAcquire(Uuid $sessionId): bool + { + return $this->lock($sessionId, 0.0); + } + + public function release(Uuid $sessionId): void + { + $key = $sessionId->toRfc4122(); + + $handle = $this->handles[$key] ?? null; + if (null === $handle) { + throw new SessionLockException(\sprintf('The lock of session "%s" is not held.', $key)); + } + + unset($this->handles[$key]); + + $unlocked = flock($handle, \LOCK_UN); + fclose($handle); + + if (!$unlocked) { + throw new SessionLockException(\sprintf('Failed to unlock the lock file of session "%s".', $key)); + } + } + + /** + * @return bool false when another request still held the lock after $timeout seconds + */ + private function lock(Uuid $sessionId, float $timeout): bool + { + $key = $sessionId->toRfc4122(); + + if (isset($this->handles[$key])) { + throw new SessionLockException(\sprintf('The lock of session "%s" is already held.', $key)); + } + + $path = $this->pathFor($sessionId); + $deadline = hrtime(true) + (int) ($timeout * 1_000_000_000); + + while (true) { + $handle = @fopen($path, 'c'); + if (false === $handle) { + throw new SessionLockException(\sprintf('Failed to open the lock file of session "%s": %s', $key, error_get_last()['message'] ?? 'unknown error')); + } + + while (!flock($handle, \LOCK_EX | \LOCK_NB, $wouldBlock)) { + if (1 !== $wouldBlock) { + fclose($handle); + + throw new SessionLockException(\sprintf('Failed to lock the lock file of session "%s".', $key)); + } + + if (hrtime(true) >= $deadline) { + fclose($handle); + + return false; + } + + usleep(self::RETRY_DELAY_MICROSECONDS); + } + + // FileSessionStore::gc() unlinks a lock file only while it holds its lock. A worker + // that waited on that file holds a lock nobody else sees once it gets it, so it + // starts over on the file now at the path. + if (self::isFileAt($handle, $path)) { + break; + } + + flock($handle, \LOCK_UN); + fclose($handle); + } + + // flock() leaves the modification time alone; refreshing it keeps the + // file of a live session out of FileSessionStore::gc(). + @touch($path); + + $this->handles[$key] = $handle; + + return true; + } + + /** + * Whether the open file is the one at the path, and not an unlinked one. + * + * @param resource $handle + * + * @internal shared with {@see FileSessionStore::gc()} + */ + public static function isFileAt($handle, string $path): bool + { + $open = fstat($handle); + clearstatcache(true, $path); + $current = @stat($path); + + return false !== $open && false !== $current + && $open['dev'] === $current['dev'] && $open['ino'] === $current['ino']; + } + + private function pathFor(Uuid $sessionId): string + { + return $this->directory.\DIRECTORY_SEPARATOR.$sessionId->toRfc4122().self::FILE_SUFFIX; + } +} diff --git a/src/Server/Session/FileSessionStore.php b/src/Server/Session/FileSessionStore.php index 713f6e82..6aa50b43 100644 --- a/src/Server/Session/FileSessionStore.php +++ b/src/Server/Session/FileSessionStore.php @@ -144,7 +144,8 @@ public function destroy(Uuid $id): bool /** * Remove sessions older than the configured TTL, along with the temporary files - * of writes that never made it into place. + * of writes that never made it into place and the lock files a {@see FileSessionLock} + * in the same directory left behind. * Returns an array of deleted session IDs (UUID instances). */ public function gc(): array @@ -168,12 +169,19 @@ public function gc(): array continue; } + $path = $this->directory.\DIRECTORY_SEPARATOR.$entry; + + if ($this->isLockFile($entry)) { + $this->deleteExpiredLockFile($path, $now); + + continue; + } + $isSession = Uuid::isValid($entry); if (!$isSession && !$this->isTemporaryFile($entry)) { continue; } - $path = $this->directory.\DIRECTORY_SEPARATOR.$entry; if (!is_file($path)) { continue; } @@ -200,6 +208,45 @@ public function gc(): array return $deleted; } + /** + * Unlinks the file only while holding its lock: a request that holds the lock keeps its + * file however long it runs, and one waiting for it notices the file is gone once it gets + * the lock, see {@see FileSessionLock}. + */ + private function deleteExpiredLockFile(string $path, int $now): void + { + if (!is_file($path) || ($now - (@filemtime($path) ?: 0)) <= $this->ttl) { + return; + } + + $handle = @fopen($path, 'r'); + if (false === $handle) { + return; + } + + try { + if (!flock($handle, \LOCK_EX | \LOCK_NB)) { + return; + } + + // Checked again under the lock: the file may have been locked, touched and released, + // or replaced, since it was first looked at. + if (!FileSessionLock::isFileAt($handle, $path) || ($now - (@filemtime($path) ?: 0)) <= $this->ttl) { + return; + } + + if (!@unlink($path) && is_file($path)) { + $this->logger->warning('Failed to delete expired session lock file.', [ + 'path' => $path, + 'error' => error_get_last()['message'] ?? 'unknown', + ]); + } + } finally { + flock($handle, \LOCK_UN); + fclose($handle); + } + } + private function isTemporaryFile(string $entry): bool { if (1 !== preg_match('/^(?[^.]+)\\.[0-9a-f]{12}\\.tmp$/', $entry, $matches)) { @@ -209,6 +256,15 @@ private function isTemporaryFile(string $entry): bool return Uuid::isValid($matches['id']); } + private function isLockFile(string $entry): bool + { + if (!str_ends_with($entry, FileSessionLock::FILE_SUFFIX)) { + return false; + } + + return Uuid::isValid(substr($entry, 0, -\strlen(FileSessionLock::FILE_SUFFIX))); + } + private function pathFor(Uuid $id): string { return $this->directory.\DIRECTORY_SEPARATOR.$id->toRfc4122(); diff --git a/src/Server/Session/SessionLockInterface.php b/src/Server/Session/SessionLockInterface.php new file mode 100644 index 00000000..ceb150c0 --- /dev/null +++ b/src/Server/Session/SessionLockInterface.php @@ -0,0 +1,52 @@ + + */ +interface SessionLockInterface +{ + /** + * Blocks until the lock of the session is held, or gives up after the implementation's timeout. + * + * @throws SessionLockTimeoutException when another request still held the lock after the timeout + * @throws SessionLockException when the lock is already held by this instance, or the backend failed + */ + public function acquire(Uuid $sessionId): void; + + /** + * Takes the lock of the session only if no other request holds it right now, without waiting. + * + * @return bool false when another request holds the lock + * + * @throws SessionLockException when the lock is already held by this instance, or the backend failed + */ + public function tryAcquire(Uuid $sessionId): bool; + + /** + * @throws SessionLockException when the lock is not held by this instance, or releasing it failed + */ + public function release(Uuid $sessionId): void; +} diff --git a/src/Server/Session/SymfonyLockSessionLock.php b/src/Server/Session/SymfonyLockSessionLock.php new file mode 100644 index 00000000..e065d2dc --- /dev/null +++ b/src/Server/Session/SymfonyLockSessionLock.php @@ -0,0 +1,167 @@ + + */ +final class SymfonyLockSessionLock implements SessionLockInterface +{ + private const RETRY_DELAY_MICROSECONDS = 10_000; + + /** @var array */ + private array $locks = []; + + private bool $releasesOnShutdown = false; + + /** + * @param float $timeout how long to wait for a lock held by another request, in seconds + * @param float $ttl how long a held lock lives before the store expires it, in seconds; longer than the longest handler + * @param string $prefix prepended to the session id to form the lock resource name + */ + public function __construct( + private readonly LockFactory $lockFactory, + private readonly float $timeout = 30.0, + private readonly float $ttl = 300.0, + private readonly string $prefix = 'mcp-session-', + private readonly LoggerInterface $logger = new NullLogger(), + ) { + if ($this->timeout < 0) { + throw new InvalidArgumentException('timeout must be greater than or equal to 0.'); + } + + if ($this->ttl <= 0) { + throw new InvalidArgumentException('ttl must be greater than 0.'); + } + } + + public function acquire(Uuid $sessionId): void + { + if (!$this->lock($sessionId, $this->timeout)) { + throw new SessionLockTimeoutException(\sprintf('Timed out after %.3fs waiting for the lock of session "%s".', $this->timeout, $sessionId->toRfc4122())); + } + } + + public function tryAcquire(Uuid $sessionId): bool + { + return $this->lock($sessionId, 0.0); + } + + public function release(Uuid $sessionId): void + { + $key = $sessionId->toRfc4122(); + + $lock = $this->locks[$key] ?? null; + if (null === $lock) { + throw new SessionLockException(\sprintf('The lock of session "%s" is not held.', $key)); + } + + unset($this->locks[$key]); + + try { + $lock->release(); + } catch (LockExceptionInterface $e) { + throw new SessionLockException(\sprintf('Failed to release the lock of session "%s": %s', $key, self::describe($e)), 0, $e); + } + } + + /** + * Releases every lock this instance still holds; a lock that fails to release is logged and left to expire. + */ + public function releaseAll(): void + { + foreach ($this->locks as $key => $lock) { + unset($this->locks[$key]); + + try { + $lock->release(); + } catch (LockExceptionInterface $e) { + $this->logger->warning(\sprintf('Failed to release the lock of session "%s", it expires after its ttl: %s', $key, self::describe($e)), [ + 'session_id' => $key, + 'exception' => $e, + ]); + } + } + } + + /** + * @return bool false when another request still held the lock after $timeout seconds + */ + private function lock(Uuid $sessionId, float $timeout): bool + { + $key = $sessionId->toRfc4122(); + + if (isset($this->locks[$key])) { + throw new SessionLockException(\sprintf('The lock of session "%s" is already held.', $key)); + } + + $lock = $this->lockFactory->createLock($this->prefix.$key, $this->ttl); + $deadline = hrtime(true) + (int) ($timeout * 1_000_000_000); + + try { + while (!$lock->acquire()) { + if (hrtime(true) >= $deadline) { + return false; + } + + usleep(self::RETRY_DELAY_MICROSECONDS); + } + } catch (LockExceptionInterface $e) { + throw new SessionLockException(\sprintf('Failed to acquire the lock of session "%s": %s', $key, self::describe($e)), 0, $e); + } + + $this->locks[$key] = $lock; + + if (!$this->releasesOnShutdown) { + register_shutdown_function($this->releaseAll(...)); + $this->releasesOnShutdown = true; + } + + return true; + } + + /** + * The lock component wraps a failing store in its own exception with a generic message, so the cause, e.g. + * a refused connection, is added from the innermost exception. + */ + private static function describe(\Throwable $e): string + { + $cause = $e; + while (null !== $cause->getPrevious()) { + $cause = $cause->getPrevious(); + } + + return $cause === $e ? $e->getMessage() : \sprintf('%s (%s)', $e->getMessage(), $cause->getMessage()); + } +} diff --git a/src/Server/Transport/BaseTransport.php b/src/Server/Transport/BaseTransport.php index f80b1046..6acebbc9 100644 --- a/src/Server/Transport/BaseTransport.php +++ b/src/Server/Transport/BaseTransport.php @@ -37,6 +37,8 @@ abstract class BaseTransport implements TransportInterface { use ManagesTransportCallbacks; + private const YIELD_FAILED_MESSAGE = 'Failed to send the message to the client.'; + protected ?Uuid $sessionId = null; /** @@ -112,22 +114,34 @@ protected function checkForResponse(int $requestId, ?Uuid $sessionId): Response| } /** + * Hands what the session's fiber yielded to the yield handler. + * + * When the handler fails, the message the fiber suspended on never reaches the client, so + * the fiber is resumed with an error instead of being left waiting on nothing; whatever it + * yields next is handled the same way. + * * @param FiberSuspend|null $yielded * @param McpFiber $fiber the fiber that yielded */ protected function handleFiberYield(mixed $yielded, ?Uuid $sessionId, \Fiber $fiber): void { - if (null === $yielded || !\is_callable($this->fiberYieldHandler)) { - return; - } - - try { - ($this->fiberYieldHandler)($yielded, $sessionId, $fiber); - } catch (\Throwable $e) { - $this->logger->error('Fiber yield handler failed.', [ - 'exception' => $e, - 'sessionId' => $sessionId?->toRfc4122(), - ]); + while (null !== $yielded && \is_callable($this->fiberYieldHandler)) { + try { + ($this->fiberYieldHandler)($yielded, $sessionId, $fiber); + + return; + } catch (\Throwable $e) { + $this->logger->error('Fiber yield handler failed.', [ + 'exception' => $e, + 'sessionId' => $sessionId?->toRfc4122(), + ]); + } + + if (!$fiber->isSuspended()) { + return; + } + + $yielded = $fiber->resume(Error::forInternalError(self::YIELD_FAILED_MESSAGE)); } } diff --git a/tests/Unit/Server/BuilderTest.php b/tests/Unit/Server/BuilderTest.php index 0ca18360..fa101c40 100644 --- a/tests/Unit/Server/BuilderTest.php +++ b/tests/Unit/Server/BuilderTest.php @@ -35,9 +35,12 @@ use Mcp\Server\Stateless\StatelessProtocol; use Mcp\Server\Subscription\InMemoryNotificationBus; use Mcp\Server\Subscription\PublishingEventDispatcher; +use Mcp\Server\Transport\InMemoryTransport; use Mcp\Tests\Unit\Server\Extension\ThingExtension; use Mcp\Tests\Unit\Server\Extension\ThingListHandler; use Mcp\Tests\Unit\Server\Extension\ThingListRequest; +use Mcp\Tests\Unit\Server\Session\Fixture\RecordingSessionLock; +use Mcp\Tests\Unit\Server\Session\Fixture\RecordingSessionStore; use PHPUnit\Framework\Attributes\TestDox; use PHPUnit\Framework\TestCase; @@ -192,6 +195,52 @@ public function testBuildAdvertisesToolsForPreloadedCustomRegistry(): void $this->assertTrue($capabilities->tools); } + #[TestDox('setSessionLock() returns the builder for fluent chaining')] + public function testSetSessionLockReturnsSelf(): void + { + $log = []; + $builder = Server::builder(); + + $this->assertSame($builder, $builder->setSessionLock(new RecordingSessionLock($log))); + } + + #[TestDox('setSessionLock() wires the lock around every session access of the built server')] + public function testSetSessionLockIsWiredIntoTheProtocol(): void + { + $log = []; + $store = new RecordingSessionStore($log); + $lock = new RecordingSessionLock($log); + + $server = Server::builder() + ->setSession($store, gcProbability: 0) + ->setSessionLock($lock) + ->build(); + + $server->run(new InMemoryTransport([ + '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}', + '{"jsonrpc":"2.0","id":2,"method":"ping"}', + ])); + + $this->assertCount(11, $log, implode(', ', $log)); + $sessionId = substr($log[0], \strlen('read ')); + $this->assertSame([ + // initialize: the session is new, nobody else can know it yet + "read $sessionId", + "read $sessionId", + "write $sessionId", + // ping: locked around read and save + "acquire $sessionId", + "read $sessionId", + "read $sessionId", + "write $sessionId", + "release $sessionId", + // the transport ends the session when it is done + "acquire $sessionId", + "destroy $sessionId", + "release $sessionId", + ], $log); + } + #[TestDox('setLazyLoading() returns the builder for fluent chaining')] public function testSetLazyLoadingReturnsSelf(): void { diff --git a/tests/Unit/Server/ClientGatewayTest.php b/tests/Unit/Server/ClientGatewayTest.php index ed9a5e8b..2df0b11e 100644 --- a/tests/Unit/Server/ClientGatewayTest.php +++ b/tests/Unit/Server/ClientGatewayTest.php @@ -24,6 +24,7 @@ use Mcp\Schema\JsonRpc\Request; use Mcp\Schema\JsonRpc\Response; use Mcp\Schema\Notification\LoggingMessageNotification; +use Mcp\Schema\Notification\ProgressNotification; use Mcp\Schema\Request\ElicitRequest; use Mcp\Schema\Request\ListRootsRequest; use Mcp\Schema\Result\ElicitResult; @@ -238,6 +239,35 @@ private function logInFiber(?string $minimumLevel, LoggingLevel $level): ?Loggin return $suspend->notification; } + public function testNotifyReturnsWhenResumedWithNothing(): void + { + $session = $this->createMock(SessionInterface::class); + $session->method('getId')->willReturn(Uuid::v4()); + $gateway = new ClientGateway($session); + + $fiber = new \Fiber(static fn () => $gateway->notify(new ProgressNotification('p', 0.5))); + $this->assertInstanceOf(NotificationSuspension::class, $fiber->start()); + + $fiber->resume(); + + $this->assertTrue($fiber->isTerminated()); + } + + public function testNotifyThrowsClientExceptionWhenTheNotificationCouldNotBeQueued(): void + { + $session = $this->createMock(SessionInterface::class); + $session->method('getId')->willReturn(Uuid::v4()); + $gateway = new ClientGateway($session); + + $fiber = new \Fiber(static fn () => $gateway->notify(new ProgressNotification('p', 0.5))); + $fiber->start(); + + $this->expectException(ClientException::class); + $this->expectExceptionMessage('could not be queued'); + + $fiber->resume(Error::forInternalError('The notification could not be queued.')); + } + /** * @param array $capabilities the client capabilities the session reports */ diff --git a/tests/Unit/Server/ProtocolSessionLockTest.php b/tests/Unit/Server/ProtocolSessionLockTest.php new file mode 100644 index 00000000..b273816a --- /dev/null +++ b/tests/Unit/Server/ProtocolSessionLockTest.php @@ -0,0 +1,479 @@ + */ + private array $log = []; + + private RecordingSessionStore $store; + + protected function setUp(): void + { + $this->store = new RecordingSessionStore($this->log); + } + + #[TestDox('a request holds the lock from before the session is read until after it is saved')] + public function testRequestHoldsTheLockAroundReadAndSave(): void + { + $lock = new RecordingSessionLock($this->log); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession(); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId); + + $this->assertSame([ + "acquire $sessionId", + "read $sessionId", + // save() reads again to write back only the keys this request changed + "read $sessionId", + "write $sessionId", + "release $sessionId", + ], $this->log); + $this->assertSame([], $lock->heldSessions()); + $this->assertCount(1, $transport->sent); + $this->assertSame(1, json_decode($transport->sent[0]['message'], true)['id']); + } + + #[TestDox('the lock is released when resolving the session throws')] + public function testLockIsReleasedWhenTheSessionCannotBeResolved(): void + { + $lock = new RecordingSessionLock($this->log); + $sessionManager = $this->createMock(SessionManagerInterface::class); + // Found before the lock is taken, the store fails when the session is looked up again under it. + $lookups = 0; + $sessionManager->method('exists')->willReturnCallback(static function () use (&$lookups): bool { + if (++$lookups > 1) { + throw new \RuntimeException('store down'); + } + + return true; + }); + $protocol = new Protocol([new PingHandler()], [], MessageFactory::make(), $sessionManager, sessionLock: $lock); + $sessionId = Uuid::v4(); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId); + + $this->assertSame(["acquire $sessionId", "release $sessionId"], $this->log); + $this->assertSame([], $lock->heldSessions()); + $this->assertCount(1, $transport->sent); + $this->assertSame(Error::INTERNAL_ERROR, json_decode($transport->sent[0]['message'], true)['error']['code']); + } + + #[TestDox('a request for an unknown session takes no lock, so no lock is left behind for an id nobody uses')] + public function testUnknownSessionTakesNoLock(): void + { + $lock = new RecordingSessionLock($this->log); + $protocol = $this->protocol($lock); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1,"method":"ping"}', Uuid::v4()); + + $this->assertSame([], $this->log); + $this->assertCount(1, $transport->sent); + $this->assertSame(404, $transport->sent[0]['context']['status_code']); + } + + #[TestDox('a session that ends while a request waits for its lock is refused under the lock')] + public function testSessionEndedWhileWaitingForTheLockIsRefused(): void + { + $lock = new RecordingSessionLock($this->log); + $sessionManager = $this->createMock(SessionManagerInterface::class); + $sessionManager->method('exists')->willReturnOnConsecutiveCalls(true, false); + $sessionManager->expects($this->never())->method('createWithId'); + $protocol = new Protocol([new PingHandler()], [], MessageFactory::make(), $sessionManager, sessionLock: $lock); + $sessionId = Uuid::v4(); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId); + + $this->assertSame(["acquire $sessionId", "release $sessionId"], $this->log); + $this->assertCount(1, $transport->sent); + $this->assertSame(404, $transport->sent[0]['context']['status_code']); + } + + #[TestDox('a request without a session id, like initialize, takes no lock')] + public function testRequestWithoutSessionIdTakesNoLock(): void + { + $lock = new RecordingSessionLock($this->log); + $protocol = $this->protocol($lock); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1,"method":"ping"}', null); + + $this->assertSame([], $this->log); + $this->assertCount(1, $transport->sent); + $this->assertSame(400, $transport->sent[0]['context']['status_code']); + } + + #[TestDox('a request whose session is locked by another request is refused before the session is read')] + public function testBusySessionIsRefusedWithoutReadingIt(): void + { + $lock = new RecordingSessionLock($this->log, acquirable: false); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession(); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId); + + $this->assertSame([], $this->log); + $this->assertCount(1, $transport->sent); + $this->assertSame(503, $transport->sent[0]['context']['status_code']); + + $error = json_decode($transport->sent[0]['message'], true); + $this->assertSame(1, $error['id']); + $this->assertSame(Error::SERVER_ERROR, $error['error']['code']); + $this->assertSame('Another request of this session is still being processed.', $error['error']['message']); + } + + #[TestDox('a busy session is logged as a warning, not as an error')] + public function testBusySessionIsLoggedAsAWarning(): void + { + $logger = new RecordingLogger(); + $protocol = $this->protocol(new RecordingSessionLock($this->log, acquirable: false), $logger); + + $protocol->processInput(new RecordingTransport(), '{"jsonrpc":"2.0","id":1,"method":"ping"}', $this->existingSession()); + + $this->assertCount(1, $logger->messagesAt(LogLevel::WARNING)); + $this->assertStringContainsString('Timed out', $logger->messagesAt(LogLevel::WARNING)[0]); + $this->assertSame([], $logger->messagesAt(LogLevel::ERROR)); + } + + #[TestDox('a lock backend that is down is answered with 503 saying so, and logged as an error with its cause')] + public function testUnavailableLockIsRefusedAndLoggedWithItsCause(): void + { + $logger = new RecordingLogger(); + $protocol = $this->protocol(new RecordingSessionLock($this->log, failure: 'Connection refused'), $logger); + $sessionId = $this->existingSession(); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId); + + $this->assertSame([], $this->log); + $this->assertCount(1, $transport->sent); + $this->assertSame(503, $transport->sent[0]['context']['status_code']); + + $error = json_decode($transport->sent[0]['message'], true); + $this->assertSame(1, $error['id']); + $this->assertSame(Error::SERVER_ERROR, $error['error']['code']); + $this->assertSame('The session lock is unavailable.', $error['error']['message']); + + $errors = $logger->messagesAt(LogLevel::ERROR); + $this->assertCount(1, $errors); + $this->assertStringContainsString('Connection refused', $errors[0]); + $this->assertSame([], $logger->messagesAt(LogLevel::WARNING)); + } + + #[TestDox('a stream polling while the lock backend is down gets nothing yet, instead of an exception')] + public function testPollingWithAnUnavailableLockReturnsNothing(): void + { + $protocol = $this->protocol(new RecordingSessionLock($this->log, failure: 'Connection refused')); + $sessionId = $this->existingSession(); + + $this->assertNull($protocol->checkResponse(1, $sessionId)); + $this->assertSame([], $protocol->consumeOutgoingMessages($sessionId)); + $this->assertSame([], $this->log); + } + + #[TestDox('a session the client ended is destroyed even when the lock backend is down')] + public function testDestroyingWithAnUnavailableLockStillDestroys(): void + { + $protocol = $this->protocol(new RecordingSessionLock($this->log, failure: 'Connection refused')); + $sessionId = $this->existingSession(); + + $protocol->destroySession($sessionId); + + $this->assertSame(["destroy $sessionId"], $this->log); + $this->assertFalse($this->store->exists($sessionId)); + } + + #[TestDox('a refused batch is answered under the empty id, as a batch failure is')] + public function testRefusedBatchIsAnsweredUnderTheEmptyId(): void + { + $lock = new RecordingSessionLock($this->log, acquirable: false); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession(); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '[{"jsonrpc":"2.0","id":1,"method":"ping"},{"jsonrpc":"2.0","id":2,"method":"ping"}]', $sessionId); + + $this->assertCount(1, $transport->sent); + $this->assertSame(503, $transport->sent[0]['context']['status_code']); + $this->assertSame('', json_decode($transport->sent[0]['message'], true)['id']); + } + + #[TestDox('a client response refused for a busy session is not answered under its id, which is the client\'s own request id')] + public function testRefusedClientResponseIsNotAnsweredUnderItsId(): void + { + $lock = new RecordingSessionLock($this->log, acquirable: false); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession(); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1000,"result":{"action":"accept"}}', $sessionId); + + $this->assertCount(1, $transport->sent); + $this->assertSame(503, $transport->sent[0]['context']['status_code']); + $error = json_decode($transport->sent[0]['message'], true); + $this->assertNull($error['id']); + $this->assertSame(Error::SERVER_ERROR, $error['error']['code']); + } + + #[TestDox('a refused batch of client responses is not answered under the empty id either')] + public function testRefusedBatchOfClientResponsesIsNotAnsweredUnderTheEmptyId(): void + { + $lock = new RecordingSessionLock($this->log, acquirable: false); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession(); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '[{"jsonrpc":"2.0","id":1000,"result":{}},{"jsonrpc":"2.0","method":"notifications/initialized"}]', $sessionId); + + $this->assertCount(1, $transport->sent); + $this->assertNull(json_decode($transport->sent[0]['message'], true)['id']); + } + + #[TestDox('a client response whose processing fails is not answered at all, like a notification')] + public function testFailingClientResponseIsNotAnswered(): void + { + $sessionManager = $this->createMock(SessionManagerInterface::class); + $sessionManager->method('exists')->willThrowException(new \RuntimeException('store down')); + $protocol = new Protocol([new PingHandler()], [], MessageFactory::make(), $sessionManager, sessionLock: new RecordingSessionLock($this->log)); + $transport = new RecordingTransport(); + + $protocol->processInput($transport, '{"jsonrpc":"2.0","id":1000,"result":{}}', Uuid::v4()); + + $this->assertSame([], $transport->sent); + } + + #[TestDox('consuming the outgoing queue locks around its read and save, without waiting for the lock')] + public function testConsumingOutgoingMessagesLocks(): void + { + $lock = new RecordingSessionLock($this->log); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession([ + '_mcp.outgoing_queue' => [['message' => '{"jsonrpc":"2.0","method":"notifications/message"}', 'context' => ['type' => 'notification']]], + ]); + + $messages = $protocol->consumeOutgoingMessages($sessionId); + + $this->assertCount(1, $messages); + $this->assertSame([ + "try-acquire $sessionId", + "read $sessionId", + "read $sessionId", + "write $sessionId", + "release $sessionId", + ], $this->log); + } + + #[TestDox('consuming an empty outgoing queue still happens under the lock, and does not save')] + public function testConsumingAnEmptyQueueLocksWithoutSaving(): void + { + $lock = new RecordingSessionLock($this->log); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession(); + + $this->assertSame([], $protocol->consumeOutgoingMessages($sessionId)); + $this->assertSame(["try-acquire $sessionId", "read $sessionId", "release $sessionId"], $this->log); + } + + #[TestDox('consuming a client response locks around its read and save, without waiting for the lock')] + public function testCheckingForAResponseLocks(): void + { + $lock = new RecordingSessionLock($this->log); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession([ + '_mcp.pending_requests' => [7 => ['request_id' => 7, 'timeout' => 120, 'timestamp' => time()]], + '_mcp.responses' => [7 => ['jsonrpc' => '2.0', 'id' => 7, 'result' => ['ok' => true]]], + ]); + + $this->assertInstanceOf(Response::class, $protocol->checkResponse(7, $sessionId)); + $this->assertSame([ + "try-acquire $sessionId", + "read $sessionId", + "read $sessionId", + "write $sessionId", + "release $sessionId", + ], $this->log); + } + + #[TestDox('reading the pending requests takes no lock')] + public function testReadingPendingRequestsTakesNoLock(): void + { + $lock = new RecordingSessionLock($this->log); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession([ + '_mcp.pending_requests' => [7 => ['request_id' => 7, 'timeout' => 120, 'timestamp' => time()]], + ]); + + $this->assertArrayHasKey(7, $protocol->getPendingRequests($sessionId)); + $this->assertSame(["read $sessionId"], $this->log); + } + + #[TestDox('a poll that cannot get the lock returns nothing and leaves the session alone, to try again next turn')] + public function testPollingABusySessionLeavesItAlone(): void + { + $lock = new RecordingSessionLock($this->log, acquirable: false); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession([ + '_mcp.outgoing_queue' => [['message' => '{"jsonrpc":"2.0","method":"notifications/message"}', 'context' => ['type' => 'notification']]], + '_mcp.pending_requests' => [7 => ['request_id' => 7, 'timeout' => 120, 'timestamp' => time()]], + '_mcp.responses' => [7 => ['jsonrpc' => '2.0', 'id' => 7, 'result' => ['ok' => true]]], + ]); + + $this->assertSame([], $protocol->consumeOutgoingMessages($sessionId)); + $this->assertNull($protocol->checkResponse(7, $sessionId)); + $this->assertSame([], $this->log); + + $session = new Session($this->store, $sessionId); + $this->assertCount(1, $session->get('_mcp.outgoing_queue')); + $this->assertArrayHasKey(7, $session->get('_mcp.responses')); + } + + #[TestDox('storing the request a resumed fiber sends locks around its read and save')] + public function testFiberYieldLocks(): void + { + $lock = new RecordingSessionLock($this->log); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession(); + + $requestId = $protocol->handleFiberYield(new RequestSuspension(new PingRequest(), $sessionId->toRfc4122(), 5), $sessionId); + + $this->assertSame(1000, $requestId); + $this->assertSame([ + "acquire $sessionId", + "read $sessionId", + "read $sessionId", + "write $sessionId", + "release $sessionId", + ], $this->log); + $this->assertArrayHasKey(1000, (new Session($this->store, $sessionId))->get('_mcp.pending_requests')); + } + + #[TestDox('a resumed fiber whose session is locked by another request fails loudly: its request must not be lost silently')] + public function testFiberYieldOnABusySessionThrows(): void + { + $lock = new RecordingSessionLock($this->log, acquirable: false); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession(); + + $this->expectException(SessionLockException::class); + + $protocol->handleFiberYield(new RequestSuspension(new PingRequest(), $sessionId->toRfc4122(), 5), $sessionId); + } + + #[TestDox('destroying a session takes its lock, so a request still running cannot write it back')] + public function testDestroyingASessionLocks(): void + { + $lock = new RecordingSessionLock($this->log); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession(); + + $protocol->destroySession($sessionId); + + $this->assertSame(["acquire $sessionId", "destroy $sessionId", "release $sessionId"], $this->log); + $this->assertFalse($this->store->exists($sessionId)); + } + + #[TestDox('a session the client ended is destroyed even when its lock is busy')] + public function testDestroyingABusySessionStillDestroysIt(): void + { + $lock = new RecordingSessionLock($this->log, acquirable: false); + $protocol = $this->protocol($lock); + $sessionId = $this->existingSession(); + + $protocol->destroySession($sessionId); + + $this->assertSame(["destroy $sessionId"], $this->log); + $this->assertFalse($this->store->exists($sessionId)); + } + + #[TestDox('ending an unknown session takes no lock')] + public function testDestroyingAnUnknownSessionTakesNoLock(): void + { + $lock = new RecordingSessionLock($this->log); + $protocol = $this->protocol($lock); + $sessionId = Uuid::v4(); + + $protocol->destroySession($sessionId); + + $this->assertSame(["destroy $sessionId"], $this->log); + } + + #[TestDox('without a lock configured, nothing but the store is touched')] + public function testWithoutALockOnlyTheStoreIsTouched(): void + { + $protocol = new Protocol([new PingHandler()], [], MessageFactory::make(), new SessionManager($this->store, gcProbability: 0)); + $sessionId = $this->existingSession(); + + $protocol->processInput(new RecordingTransport(), '{"jsonrpc":"2.0","id":1,"method":"ping"}', $sessionId); + + $this->assertSame(["read $sessionId", "read $sessionId", "write $sessionId"], $this->log); + } + + private function protocol(RecordingSessionLock $lock, LoggerInterface $logger = new NullLogger()): Protocol + { + return new Protocol( + requestHandlers: [new PingHandler()], + notificationHandlers: [], + messageFactory: MessageFactory::make(), + sessionManager: new SessionManager($this->store, gcProbability: 0), + logger: $logger, + sessionLock: $lock, + ); + } + + /** + * @param array $data + */ + private function existingSession(array $data = []): Uuid + { + $session = new Session($this->store, Uuid::v4()); + foreach ($data as $key => $value) { + $session->set($key, $value); + } + $session->save(); + + $this->log = []; + + return $session->getId(); + } +} diff --git a/tests/Unit/Server/Session/FileSessionLockTest.php b/tests/Unit/Server/Session/FileSessionLockTest.php new file mode 100644 index 00000000..a328473e --- /dev/null +++ b/tests/Unit/Server/Session/FileSessionLockTest.php @@ -0,0 +1,296 @@ +directory = sys_get_temp_dir().'/mcp-file-session-lock-'.bin2hex(random_bytes(6)); + } + + protected function tearDown(): void + { + if (!is_dir($this->directory)) { + return; + } + + @chmod($this->directory, 0775); + + foreach (glob($this->directory.'/*') ?: [] as $file) { + @unlink($file); + } + + @rmdir($this->directory); + } + + #[TestDox('creates the lock directory when it does not exist yet')] + public function testCreatesMissingDirectory(): void + { + new FileSessionLock($this->directory); + + $this->assertDirectoryExists($this->directory); + } + + #[TestDox('rejects an unwritable directory with the SDK\'s own exception')] + public function testUnwritableDirectoryThrows(): void + { + mkdir($this->directory, 0775, true); + chmod($this->directory, 0555); + clearstatcache(true, $this->directory); + + if (is_writable($this->directory)) { + $this->markTestSkipped('The current user can write to a read-only directory (probably root).'); + } + + $this->expectException(RuntimeException::class); + + new FileSessionLock($this->directory); + } + + #[TestDox('rejects a negative timeout')] + public function testNegativeTimeoutThrows(): void + { + $this->expectException(InvalidArgumentException::class); + + new FileSessionLock($this->directory, timeout: -1); + } + + #[TestDox('holds a lock file named after the session, and keeps it after release')] + public function testAcquireCreatesALockFilePerSession(): void + { + $lock = new FileSessionLock($this->directory); + $id = Uuid::v4(); + + $lock->acquire($id); + $this->assertFileExists($this->directory.'/'.$id->toRfc4122().FileSessionLock::FILE_SUFFIX); + + $lock->release($id); + $this->assertFileExists($this->directory.'/'.$id->toRfc4122().FileSessionLock::FILE_SUFFIX); + } + + #[TestDox('another worker cannot acquire a held lock, and can once it is released')] + public function testHeldLockBlocksAnotherWorkerUntilReleased(): void + { + $id = Uuid::v4(); + $first = new FileSessionLock($this->directory, timeout: 0.05); + $second = new FileSessionLock($this->directory, timeout: 0.05); + + $first->acquire($id); + + try { + $second->acquire($id); + $this->fail('The lock was acquired twice.'); + } catch (SessionLockException $e) { + $this->assertInstanceOf(SessionLockTimeoutException::class, $e); + $this->assertStringContainsString('Timed out after 0.050s', $e->getMessage()); + $this->assertStringContainsString($id->toRfc4122(), $e->getMessage()); + } + + $first->release($id); + + $second->acquire($id); + $second->release($id); + } + + #[TestDox('a zero timeout tries once')] + public function testZeroTimeoutDoesNotWait(): void + { + $id = Uuid::v4(); + $first = new FileSessionLock($this->directory); + $second = new FileSessionLock($this->directory, timeout: 0); + + $first->acquire($id); + + $start = hrtime(true); + try { + $second->acquire($id); + $this->fail('The lock was acquired twice.'); + } catch (SessionLockException) { + $this->assertLessThan(1_000_000_000, hrtime(true) - $start); + } + + $first->release($id); + } + + #[TestDox('tryAcquire() takes a free lock')] + public function testTryAcquireTakesAFreeLock(): void + { + $lock = new FileSessionLock($this->directory); + $id = Uuid::v4(); + + $this->assertTrue($lock->tryAcquire($id)); + $this->assertFalse((new FileSessionLock($this->directory))->tryAcquire($id)); + + $lock->release($id); + } + + #[TestDox('tryAcquire() returns false at once for a lock another worker holds, and does not wait for the timeout')] + public function testTryAcquireDoesNotWait(): void + { + $id = Uuid::v4(); + $first = new FileSessionLock($this->directory); + $second = new FileSessionLock($this->directory, timeout: 30); + + $first->acquire($id); + + $start = hrtime(true); + $this->assertFalse($second->tryAcquire($id)); + $this->assertLessThan(1_000_000_000, hrtime(true) - $start); + + $first->release($id); + + $this->assertTrue($second->tryAcquire($id)); + $second->release($id); + } + + #[TestDox('tryAcquire() of a lock this instance already holds is an error')] + public function testTryAcquiringAHeldLockThrows(): void + { + $lock = new FileSessionLock($this->directory); + $id = Uuid::v4(); + $lock->acquire($id); + + $this->expectException(SessionLockException::class); + $this->expectExceptionMessage('already held'); + + $lock->tryAcquire($id); + } + + #[TestDox('a worker that waited on a lock file gc() removed locks the file now at the path, not the removed one')] + public function testWaiterOnARemovedLockFileStartsOver(): void + { + if (!\function_exists('proc_open') || '\\' === \DIRECTORY_SEPARATOR) { + $this->markTestSkipped('Needs proc_open() and POSIX unlink semantics.'); + } + + $id = Uuid::v4(); + $path = $this->directory.'/'.$id->toRfc4122().FileSessionLock::FILE_SUFFIX; + new FileSessionLock($this->directory); + + // Another process does what FileSessionStore::gc() does: it locks the file, unlinks it and releases it. + $script = <<<'PHP' + directory.'/gc.php'; + file_put_contents($file, $script); + + $process = proc_open([\PHP_BINARY, $file, $path], [1 => ['pipe', 'w'], 2 => ['pipe', 'w']], $pipes); + $this->assertIsResource($process); + $this->assertSame("locked\n", fgets($pipes[1])); + + $waiter = new FileSessionLock($this->directory, timeout: 5); + $waiter->acquire($id); + + $this->assertSame('', stream_get_contents($pipes[2])); + fclose($pipes[1]); + fclose($pipes[2]); + $this->assertSame(0, proc_close($process)); + + $this->assertFileExists($path); + $this->assertFalse((new FileSessionLock($this->directory))->tryAcquire($id), 'A second worker locked the session next to the waiter.'); + + $waiter->release($id); + } + + #[TestDox('locks of different sessions do not block each other')] + public function testDifferentSessionsDoNotBlockEachOther(): void + { + $this->expectNotToPerformAssertions(); + + $first = new FileSessionLock($this->directory, timeout: 0.05); + $second = new FileSessionLock($this->directory, timeout: 0.05); + $a = Uuid::v4(); + $b = Uuid::v4(); + + $first->acquire($a); + $second->acquire($b); + + $first->release($a); + $second->release($b); + } + + #[TestDox('acquiring a lock this instance already holds is an error, not a deadlock')] + public function testAcquiringTwiceThrows(): void + { + $lock = new FileSessionLock($this->directory, timeout: 0.05); + $id = Uuid::v4(); + $lock->acquire($id); + + $this->expectException(SessionLockException::class); + $this->expectExceptionMessage('already held'); + + $lock->acquire($id); + } + + #[TestDox('releasing a lock that is not held is an error')] + public function testReleasingAnUnheldLockThrows(): void + { + $lock = new FileSessionLock($this->directory); + + $this->expectException(SessionLockException::class); + $this->expectExceptionMessage('not held'); + + $lock->release(Uuid::v4()); + } + + #[TestDox('a lock can be acquired again by the same instance after release')] + public function testReacquireAfterRelease(): void + { + $this->expectNotToPerformAssertions(); + + $lock = new FileSessionLock($this->directory, timeout: 0.05); + $id = Uuid::v4(); + + $lock->acquire($id); + $lock->release($id); + $lock->acquire($id); + $lock->release($id); + } + + #[TestDox('acquiring refreshes the lock file\'s modification time, so gc() keeps the file of a live session')] + public function testAcquireTouchesTheLockFile(): void + { + $lock = new FileSessionLock($this->directory); + $id = Uuid::v4(); + $path = $this->directory.'/'.$id->toRfc4122().FileSessionLock::FILE_SUFFIX; + + touch($path, time() - 7200); + clearstatcache(true, $path); + $this->assertLessThan(time() - 3600, filemtime($path)); + + $lock->acquire($id); + $lock->release($id); + + clearstatcache(true, $path); + $this->assertGreaterThan(time() - 60, filemtime($path)); + } +} diff --git a/tests/Unit/Server/Session/FileSessionStoreTest.php b/tests/Unit/Server/Session/FileSessionStoreTest.php index b2c04d73..a3a9cf7e 100644 --- a/tests/Unit/Server/Session/FileSessionStoreTest.php +++ b/tests/Unit/Server/Session/FileSessionStoreTest.php @@ -13,6 +13,7 @@ use Mcp\Exception\ExceptionInterface; use Mcp\Exception\RuntimeException; +use Mcp\Server\Session\FileSessionLock; use Mcp\Server\Session\FileSessionStore; use PHPUnit\Framework\Attributes\TestDox; use PHPUnit\Framework\TestCase; @@ -110,6 +111,56 @@ public function testGcCollectsOrphanedTemporaryFiles(): void $this->assertFileExists($this->directory.\DIRECTORY_SEPARATOR.$id->toRfc4122()); } + #[TestDox('gc() collects the lock files of expired sessions, and leaves those of live sessions and foreign .lock files alone')] + public function testGcCollectsExpiredLockFiles(): void + { + $store = new FileSessionStore($this->directory, ttl: 60); + $id = new UuidV4(); + $store->write($id, 'payload'); + + $lock = new FileSessionLock($this->directory); + $lock->acquire($id); + $lock->release($id); + $live = $this->directory.\DIRECTORY_SEPARATOR.$id->toRfc4122().FileSessionLock::FILE_SUFFIX; + $this->assertFileExists($live); + + $expired = $this->directory.\DIRECTORY_SEPARATOR.(new UuidV4())->toRfc4122().FileSessionLock::FILE_SUFFIX; + file_put_contents($expired, ''); + touch($expired, time() - 120); + + $foreign = $this->directory.\DIRECTORY_SEPARATOR.'important.lock'; + file_put_contents($foreign, 'not a session lock'); + touch($foreign, time() - 120); + + $deleted = $store->gc(); + + // A lock file is not a session, so collecting it is not a session deletion. + $this->assertSame([], $deleted); + $this->assertFileDoesNotExist($expired); + $this->assertFileExists($live); + $this->assertFileExists($foreign); + $this->assertFileExists($this->directory.\DIRECTORY_SEPARATOR.$id->toRfc4122()); + } + + #[TestDox('gc() leaves the lock file of a request that holds it alone, however long the request runs')] + public function testGcKeepsAHeldLockFile(): void + { + $store = new FileSessionStore($this->directory, ttl: 60); + $id = new UuidV4(); + $holder = new FileSessionLock($this->directory); + $holder->acquire($id); + + $path = $this->directory.\DIRECTORY_SEPARATOR.$id->toRfc4122().FileSessionLock::FILE_SUFFIX; + touch($path, time() - 7200); + + $store->gc(); + + $this->assertFileExists($path); + $this->assertFalse((new FileSessionLock($this->directory))->tryAcquire($id), 'A second worker locked the session next to its holder.'); + + $holder->release($id); + } + #[TestDox('rejects an unwritable directory with the SDK\'s own exception')] public function testUnwritableDirectoryThrowsPackageException(): void { diff --git a/tests/Unit/Server/Session/Fixture/HeldElsewhereSessionLock.php b/tests/Unit/Server/Session/Fixture/HeldElsewhereSessionLock.php new file mode 100644 index 00000000..3781ef74 --- /dev/null +++ b/tests/Unit/Server/Session/Fixture/HeldElsewhereSessionLock.php @@ -0,0 +1,49 @@ +heldElsewhere) { + throw new SessionLockTimeoutException(\sprintf('Timed out waiting for the lock of session "%s".', $sessionId->toRfc4122())); + } + + $this->lock->acquire($sessionId); + } + + public function tryAcquire(Uuid $sessionId): bool + { + return $this->lock->tryAcquire($sessionId); + } + + public function release(Uuid $sessionId): void + { + $this->lock->release($sessionId); + } +} diff --git a/tests/Unit/Server/Session/Fixture/RecordingLogger.php b/tests/Unit/Server/Session/Fixture/RecordingLogger.php new file mode 100644 index 00000000..0b955cae --- /dev/null +++ b/tests/Unit/Server/Session/Fixture/RecordingLogger.php @@ -0,0 +1,40 @@ +}> */ + public array $records = []; + + /** + * @param string|\Stringable $message + * @param array $context + */ + public function log($level, $message, array $context = []): void + { + $this->records[] = ['level' => (string) $level, 'message' => (string) $message, 'context' => $context]; + } + + /** + * @return list + */ + public function messagesAt(string $level): array + { + return array_values(array_map( + static fn (array $record): string => $record['message'], + array_filter($this->records, static fn (array $record): bool => $level === $record['level']), + )); + } +} diff --git a/tests/Unit/Server/Session/Fixture/RecordingSessionLock.php b/tests/Unit/Server/Session/Fixture/RecordingSessionLock.php new file mode 100644 index 00000000..75a0baea --- /dev/null +++ b/tests/Unit/Server/Session/Fixture/RecordingSessionLock.php @@ -0,0 +1,103 @@ + */ + private array $held = []; + + /** + * @param list $log receives "acquire ", "try-acquire " and "release " entries + * @param bool $acquirable false makes every acquire() fail and every tryAcquire() return false, + * like a lock another worker holds past the timeout + * @param ?string $failure makes every acquire() and tryAcquire() fail with this cause, like a lock backend that is down + */ + public function __construct( + public array &$log, + public bool $acquirable = true, + public ?string $failure = null, + ) { + } + + public function acquire(Uuid $sessionId): void + { + $key = $sessionId->toRfc4122(); + $this->failIfBroken($key); + + if (!$this->acquirable) { + throw new SessionLockTimeoutException(\sprintf('Timed out waiting for the lock of session "%s".', $key)); + } + + $this->hold($key); + $this->log[] = 'acquire '.$key; + } + + public function tryAcquire(Uuid $sessionId): bool + { + $key = $sessionId->toRfc4122(); + $this->failIfBroken($key); + + if (!$this->acquirable) { + return false; + } + + $this->hold($key); + $this->log[] = 'try-acquire '.$key; + + return true; + } + + public function release(Uuid $sessionId): void + { + $key = $sessionId->toRfc4122(); + + if (!isset($this->held[$key])) { + throw new SessionLockException(\sprintf('The lock of session "%s" is not held.', $key)); + } + + unset($this->held[$key]); + $this->log[] = 'release '.$key; + } + + private function failIfBroken(string $key): void + { + if (null !== $this->failure) { + throw new SessionLockException(\sprintf('Failed to acquire the lock of session "%s": %s', $key, $this->failure)); + } + } + + private function hold(string $key): void + { + if (isset($this->held[$key])) { + throw new SessionLockException(\sprintf('The lock of session "%s" is already held.', $key)); + } + + $this->held[$key] = true; + } + + /** + * @return list + */ + public function heldSessions(): array + { + return array_keys($this->held); + } +} diff --git a/tests/Unit/Server/Session/Fixture/RecordingSessionStore.php b/tests/Unit/Server/Session/Fixture/RecordingSessionStore.php new file mode 100644 index 00000000..84b4d4de --- /dev/null +++ b/tests/Unit/Server/Session/Fixture/RecordingSessionStore.php @@ -0,0 +1,50 @@ + $log receives "read ", "write " and "destroy " entries + */ + public function __construct(public array &$log) + { + parent::__construct(); + } + + public function read(Uuid $id): string|false + { + $this->log[] = 'read '.$id->toRfc4122(); + + return parent::read($id); + } + + public function write(Uuid $id, string $data): bool + { + $this->log[] = 'write '.$id->toRfc4122(); + + return parent::write($id, $data); + } + + public function destroy(Uuid $id): bool + { + $this->log[] = 'destroy '.$id->toRfc4122(); + + return parent::destroy($id); + } +} diff --git a/tests/Unit/Server/Session/SymfonyLockSessionLockTest.php b/tests/Unit/Server/Session/SymfonyLockSessionLockTest.php new file mode 100644 index 00000000..6922afea --- /dev/null +++ b/tests/Unit/Server/Session/SymfonyLockSessionLockTest.php @@ -0,0 +1,266 @@ +expectException(InvalidArgumentException::class); + + new SymfonyLockSessionLock(new LockFactory(new InMemoryStore()), timeout: -1); + } + + #[TestDox('rejects a non-positive ttl')] + public function testNonPositiveTtlThrows(): void + { + $this->expectException(InvalidArgumentException::class); + + new SymfonyLockSessionLock(new LockFactory(new InMemoryStore()), ttl: 0); + } + + #[TestDox('another worker cannot acquire a held lock, and can once it is released')] + public function testHeldLockBlocksAnotherWorkerUntilReleased(): void + { + $store = new InMemoryStore(); + $id = Uuid::v4(); + $first = new SymfonyLockSessionLock(new LockFactory($store), timeout: 0.05); + $second = new SymfonyLockSessionLock(new LockFactory($store), timeout: 0.05); + + $first->acquire($id); + + try { + $second->acquire($id); + $this->fail('The lock was acquired twice.'); + } catch (SessionLockException $e) { + $this->assertInstanceOf(SessionLockTimeoutException::class, $e); + $this->assertStringContainsString('Timed out after 0.050s', $e->getMessage()); + $this->assertStringContainsString($id->toRfc4122(), $e->getMessage()); + } + + $first->release($id); + + $second->acquire($id); + $second->release($id); + } + + #[TestDox('tryAcquire() returns false at once for a lock another worker holds, and takes it once it is free')] + public function testTryAcquireDoesNotWait(): void + { + $store = new InMemoryStore(); + $id = Uuid::v4(); + $first = new SymfonyLockSessionLock(new LockFactory($store)); + $second = new SymfonyLockSessionLock(new LockFactory($store), timeout: 30); + + $first->acquire($id); + + $start = hrtime(true); + $this->assertFalse($second->tryAcquire($id)); + $this->assertLessThan(1_000_000_000, hrtime(true) - $start); + + $first->release($id); + + $this->assertTrue($second->tryAcquire($id)); + $second->release($id); + } + + #[TestDox('tryAcquire() of a lock this instance already holds is an error')] + public function testTryAcquiringAHeldLockThrows(): void + { + $lock = new SymfonyLockSessionLock(new LockFactory(new InMemoryStore())); + $id = Uuid::v4(); + $lock->acquire($id); + + $this->expectException(SessionLockException::class); + $this->expectExceptionMessage('already held'); + + $lock->tryAcquire($id); + } + + #[TestDox('tryAcquire() surfaces a failing store as the SDK\'s exception')] + public function testTryAcquireWrapsStoreFailure(): void + { + $store = $this->createMock(PersistingStoreInterface::class); + $store->method('save')->willThrowException(new LockAcquiringException('redis is down')); + $store->method('exists')->willReturn(false); + + $this->expectException(SessionLockException::class); + $this->expectExceptionMessage('Failed to acquire the lock of session'); + + (new SymfonyLockSessionLock(new LockFactory($store)))->tryAcquire(Uuid::v4()); + } + + #[TestDox('acquire() reports a failing store as a failure, not as a timeout')] + public function testAcquireStoreFailureIsNotATimeout(): void + { + $store = $this->createMock(PersistingStoreInterface::class); + $store->method('save')->willThrowException(new LockAcquiringException('Connection refused')); + $store->method('exists')->willReturn(false); + + try { + (new SymfonyLockSessionLock(new LockFactory($store)))->acquire(Uuid::v4()); + $this->fail('The lock was acquired from a failing store.'); + } catch (SessionLockException $e) { + $this->assertNotInstanceOf(SessionLockTimeoutException::class, $e); + $this->assertStringContainsString('Connection refused', $e->getMessage()); + $this->assertInstanceOf(LockAcquiringException::class, $e->getPrevious()); + } + } + + #[TestDox('releaseAll(), registered as a shutdown function, frees every lock still held, e.g. after a fatal error')] + public function testReleaseAllFreesEveryHeldLock(): void + { + $store = new InMemoryStore(); + $crashed = new SymfonyLockSessionLock(new LockFactory($store)); + $other = new SymfonyLockSessionLock(new LockFactory($store)); + $a = Uuid::v4(); + $b = Uuid::v4(); + $crashed->acquire($a); + $crashed->acquire($b); + + $crashed->releaseAll(); + + $this->assertTrue($other->tryAcquire($a)); + $this->assertTrue($other->tryAcquire($b)); + + // Nothing left to release on the next call, e.g. the shutdown after a normal request. + $crashed->releaseAll(); + $this->assertTrue($crashed->tryAcquire(Uuid::v4())); + } + + #[TestDox('releaseAll() logs a lock that fails to release and still releases the others')] + public function testReleaseAllLogsAFailingReleaseAndGoesOn(): void + { + $held = []; + $store = $this->createMock(PersistingStoreInterface::class); + $store->method('save')->willReturnCallback(static function (Key $key) use (&$held): void { + $held[(string) $key] = true; + }); + $store->method('exists')->willReturnCallback(static fn (Key $key): bool => isset($held[(string) $key])); + $store->expects($this->exactly(2))->method('delete')->willThrowException(new LockReleasingException('Connection refused')); + + $logger = new RecordingLogger(); + $lock = new SymfonyLockSessionLock(new LockFactory($store), logger: $logger); + $lock->acquire(Uuid::v4()); + $lock->acquire(Uuid::v4()); + + $lock->releaseAll(); + + $warnings = $logger->messagesAt(LogLevel::WARNING); + $this->assertCount(2, $warnings); + $this->assertStringContainsString('Connection refused', $warnings[0]); + } + + #[TestDox('locks of different sessions do not block each other')] + public function testDifferentSessionsDoNotBlockEachOther(): void + { + $this->expectNotToPerformAssertions(); + + $store = new InMemoryStore(); + $first = new SymfonyLockSessionLock(new LockFactory($store), timeout: 0.05); + $second = new SymfonyLockSessionLock(new LockFactory($store), timeout: 0.05); + $a = Uuid::v4(); + $b = Uuid::v4(); + + $first->acquire($a); + $second->acquire($b); + + $first->release($a); + $second->release($b); + } + + #[TestDox('acquiring a lock this instance already holds is an error, not a deadlock')] + public function testAcquiringTwiceThrows(): void + { + $lock = new SymfonyLockSessionLock(new LockFactory(new InMemoryStore()), timeout: 0.05); + $id = Uuid::v4(); + $lock->acquire($id); + + $this->expectException(SessionLockException::class); + $this->expectExceptionMessage('already held'); + + $lock->acquire($id); + } + + #[TestDox('releasing a lock that is not held is an error')] + public function testReleasingAnUnheldLockThrows(): void + { + $lock = new SymfonyLockSessionLock(new LockFactory(new InMemoryStore())); + + $this->expectException(SessionLockException::class); + $this->expectExceptionMessage('not held'); + + $lock->release(Uuid::v4()); + } + + #[TestDox('names the lock resource after the prefix and the session id, with the configured ttl')] + public function testLockResourceAndTtl(): void + { + $saved = []; + $held = false; + $store = $this->createMock(PersistingStoreInterface::class); + $store->method('save')->willReturnCallback(static function (Key $key) use (&$saved, &$held): void { + $saved[] = (string) $key; + $held = true; + }); + $store->method('delete')->willReturnCallback(static function () use (&$held): void { + $held = false; + }); + $store->method('exists')->willReturnCallback(static function () use (&$held): bool { + return $held; + }); + $store->expects($this->once())->method('putOffExpiration')->with($this->anything(), 42.0); + + $lock = new SymfonyLockSessionLock(new LockFactory($store), ttl: 42.0, prefix: 'app-mcp-'); + $id = Uuid::v4(); + + $lock->acquire($id); + + $this->assertSame(['app-mcp-'.$id->toRfc4122()], $saved); + } + + #[TestDox('a failing store surfaces as the SDK\'s exception')] + public function testStoreFailureIsWrapped(): void + { + $store = $this->createMock(PersistingStoreInterface::class); + $store->method('save')->willThrowException(new LockAcquiringException('redis is down')); + $store->method('exists')->willReturn(false); + + $lock = new SymfonyLockSessionLock(new LockFactory($store), timeout: 0.05); + + try { + $lock->acquire(Uuid::v4()); + $this->fail('The lock was acquired against a failing store.'); + } catch (SessionLockException $e) { + $this->assertStringContainsString('Failed to acquire the lock of session', $e->getMessage()); + $this->assertInstanceOf(LockAcquiringException::class, $e->getPrevious()); + $this->assertSame('redis is down', $e->getPrevious()->getPrevious()?->getMessage()); + } + } +} diff --git a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php b/tests/Unit/Server/Transport/StreamableHttpTransportTest.php index d83095a0..621954ad 100644 --- a/tests/Unit/Server/Transport/StreamableHttpTransportTest.php +++ b/tests/Unit/Server/Transport/StreamableHttpTransportTest.php @@ -11,17 +11,24 @@ namespace Mcp\Tests\Unit\Server\Transport; +use Mcp\Exception\ClientException; use Mcp\Exception\InvalidArgumentException; use Mcp\Schema\JsonRpc\Error; +use Mcp\Schema\Notification\ProgressNotification; +use Mcp\Schema\Request\PingRequest; use Mcp\Server; use Mcp\Server\RequestContext; +use Mcp\Server\Session\FileSessionLock; use Mcp\Server\Session\Session; +use Mcp\Server\Session\SessionLockInterface; +use Mcp\Server\Suspension\RequestSuspension; use Mcp\Server\Transport\Http\Middleware\CorsMiddleware; use Mcp\Server\Transport\Http\Middleware\DnsRebindingProtectionMiddleware; use Mcp\Server\Transport\Http\Middleware\PassthroughMiddleware; use Mcp\Server\Transport\Http\Middleware\ProtocolVersionMiddleware; use Mcp\Server\Transport\StreamableHttpTransport; use Mcp\Server\Transport\TransportInterface; +use Mcp\Tests\Unit\Server\Session\Fixture\HeldElsewhereSessionLock; use Mcp\Tests\Unit\Server\Session\Fixture\InterleavingSessionStore; use Nyholm\Psr7\Factory\Psr17Factory; use PHPUnit\Framework\Attributes\DataProvider; @@ -39,11 +46,26 @@ final class StreamableHttpTransportTest extends TestCase { private Psr17Factory $factory; + private ?string $lockDirectory = null; + protected function setUp(): void { $this->factory = new Psr17Factory(); } + protected function tearDown(): void + { + if (null === $this->lockDirectory || !is_dir($this->lockDirectory)) { + return; + } + + foreach (glob($this->lockDirectory.'/*') ?: [] as $file) { + @unlink($file); + } + + @rmdir($this->lockDirectory); + } + #[TestDox('default middleware is applied when none is passed')] public function testDefaultMiddlewareIsAppliedWhenOmitted(): void { @@ -630,10 +652,77 @@ public function testJsonBatchCarriesQueuedNotificationsAndItsResponses(): void ); } + #[TestDox('with a session lock, a POST arriving while another of the session runs is held off instead of overwriting it')] + public function testSessionLockHoldsOffAConcurrentPostOfTheSameSession(): void + { + $store = new InterleavingSessionStore(); + $sessionId = $this->post($store, '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}', lock: $this->fileLock()) + ->getHeaderLine(StreamableHttpTransport::SESSION_HEADER); + + // B loads the session before A saved it, the lost update: on its own worker, with its own lock instance. + $responseB = null; + $store->interleaveOnNextWrite(function () use ($store, $sessionId, &$responseB): void { + $responseB = $this->post($store, '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"echo","arguments":{"text":"b"}}}', $sessionId, $this->fileLock()); + }, readBeforeWrite: true); + + $responseA = $this->post($store, '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"echo","arguments":{"text":"a"}}}', $sessionId, $this->fileLock()); + + $this->assertSame(200, $responseA->getStatusCode()); + $this->assertSame(2, json_decode((string) $responseA->getBody(), true)['id'] ?? null); + + $this->assertInstanceOf(ResponseInterface::class, $responseB); + $this->assertSame(503, $responseB->getStatusCode()); + $error = json_decode((string) $responseB->getBody(), true); + $this->assertSame(3, $error['id'] ?? null); + $this->assertSame(Error::SERVER_ERROR, $error['error']['code'] ?? null); + } + + #[TestDox('the session lock is released while a handler waits for the client, so the next request of the session proceeds')] + public function testSessionLockIsReleasedWhileAHandlerWaitsForTheClient(): void + { + $store = new InterleavingSessionStore(); + $lock = $this->fileLock(); + $sessionId = $this->post($store, '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}', lock: $lock) + ->getHeaderLine(StreamableHttpTransport::SESSION_HEADER); + + $first = $this->post($store, '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"ask","arguments":{}}}', $sessionId, $lock); + $second = $this->post($store, '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"ask","arguments":{}}}', $sessionId, $lock); + + $this->assertSame('text/event-stream', $first->getHeaderLine('Content-Type')); + $this->assertSame('text/event-stream', $second->getHeaderLine('Content-Type')); + + // Both requests to the client were stored, each under its own id. + $session = new Session($store, Uuid::fromString($sessionId)); + $this->assertSame([1000, 1001], array_keys($session->get('_mcp.pending_requests'))); + } + + #[TestDox('a message a resumed handler cannot store fails in the handler instead of resuming it with nothing')] + public function testMessageThatCannotBeStoredFailsInTheHandler(): void + { + $store = new InterleavingSessionStore(); + $lock = new HeldElsewhereSessionLock($this->fileLock()); + $sessionId = $this->post($store, '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}', lock: $lock) + ->getHeaderLine(StreamableHttpTransport::SESSION_HEADER); + + $stream = $this->post($store, '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"ask-then-notify","arguments":{}}}', $sessionId, $lock); + $this->assertSame('text/event-stream', $stream->getHeaderLine('Content-Type')); + + // The client answers the first request on a POST of its own, then another worker holds the + // session past the timeout, so what the resumed handler sends next cannot be stored. + $answer = $this->post($store, '{"jsonrpc":"2.0","id":1000,"result":{}}', $sessionId, $this->fileLock()); + $this->assertLessThan(300, $answer->getStatusCode()); + $lock->heldElsewhere = true; + + $output = $this->readStream($stream); + + $this->assertMatchesRegularExpression('/"id":2,"result".*first: ok, second: Failed to send the message to the client\., notify: Failed to send the message to the client\./s', $output); + $this->assertStringNotContainsString('"error"', $output); + } + /** * Sends one POST to a fresh server sharing $store, like a PHP worker would. */ - private function post(InterleavingSessionStore $store, string $body, string $sessionId = ''): ResponseInterface + private function post(InterleavingSessionStore $store, string $body, string $sessionId = '', ?SessionLockInterface $lock = null): ResponseInterface { $request = $this->factory ->createServerRequest('POST', 'http://localhost/') @@ -648,7 +737,7 @@ private function post(InterleavingSessionStore $store, string $body, string $ses ->withHeader(StreamableHttpTransport::PROTOCOL_VERSION_HEADER, '2025-06-18'); } - $server = Server::builder() + $builder = Server::builder() ->setServerInfo('test', '1.0') ->setSession($store) ->addTool(static fn (string $text): string => $text, 'echo') @@ -657,9 +746,67 @@ private function post(InterleavingSessionStore $store, string $body, string $ses return 'done'; }, 'progress') - ->build(); + ->addTool(static function (RequestContext $context): string { + // Suspends on a request to the client, as elicitation and sampling do. + \Fiber::suspend(new RequestSuspension(new PingRequest(), $context->getSession()->getId()->toRfc4122(), 5)); + + return 'done'; + }, 'ask') + ->addTool(static function (RequestContext $context): string { + $gateway = $context->getClientGateway(); + $first = $gateway->request(new PingRequest(), 5); + $second = $gateway->request(new PingRequest(), 5); + + try { + $gateway->notify(new ProgressNotification('p', 0.5)); + $notified = 'ok'; + } catch (ClientException $e) { + $notified = $e->getMessage(); + } + + return \sprintf( + 'first: %s, second: %s, notify: %s', + $first instanceof Error ? $first->message : 'ok', + $second instanceof Error ? $second->message : 'ok', + $notified, + ); + }, 'ask-then-notify'); + + if (null !== $lock) { + $builder->setSessionLock($lock); + } + + return $builder->build()->run(new StreamableHttpTransport($request, $this->factory, $this->factory)); + } + + /** + * Reads an SSE response; the stream calls ob_flush() itself, so the output is captured by a handler, not a plain buffer. + */ + private function readStream(ResponseInterface $response): string + { + $output = ''; + ob_start(static function (string $chunk) use (&$output): string { + $output .= $chunk; + + return ''; + }); + try { + $response->getBody()->getContents(); + } finally { + ob_end_flush(); + } + + return $output; + } + + /** + * A lock instance of its own, as each worker has, over one shared directory. + */ + private function fileLock(): FileSessionLock + { + $this->lockDirectory ??= sys_get_temp_dir().'/mcp-session-lock-'.bin2hex(random_bytes(6)); - return $server->run(new StreamableHttpTransport($request, $this->factory, $this->factory)); + return new FileSessionLock($this->lockDirectory, timeout: 0.05); } private function stubAuth401(): MiddlewareInterface