From 4e7a236bf341648f3e10067375a0c0a20e926bd0 Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Sat, 10 Oct 2026 03:55:02 +0200 Subject: [PATCH 1/4] [Server] Save only the session keys a request changed Concurrent requests of one session no longer overwrite each other's changes to different keys. --- CHANGELOG.md | 1 + src/Server/Session/Session.php | 88 ++++++++++++++-- tests/Unit/Server/Session/SessionTest.php | 116 ++++++++++++++++++++++ 3 files changed, 196 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e34852a9..ed4adfb3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ All notable changes to `mcp/sdk` will be documented in this file. 0.9.0 ----- +* Fix concurrent requests of one session overwriting each other's session changes: `Session::save()` now writes only the keys it changed, onto what the store holds at that point. * [BC Break] `SchemaValidator` takes an optional `Opis\JsonSchema\Validator` as its first constructor argument, moving `$logger` to second. Pass `logger:` by name. * [BC Break] `SchemaValidator::validateAgainstJsonSchema()` no longer validates an empty array as an object. Pass `new \stdClass()` for an empty object. * Add `Builder::setSchemaValidator()` to configure the validator used for `tools/call` input and output, e.g. with a resolver for external `$ref` schemas. diff --git a/src/Server/Session/Session.php b/src/Server/Session/Session.php index 24798cd4..62e64604 100644 --- a/src/Server/Session/Session.php +++ b/src/Server/Session/Session.php @@ -15,6 +15,9 @@ use Symfony\Component\Uid\UuidV4; /** + * Saving writes only the keys changed since the last save, onto what the store holds by then: + * concurrent requests of one session keep each other's changes as long as they change different keys. + * * @author Kyrian Obikwelu */ class Session implements SessionInterface @@ -31,6 +34,13 @@ class Session implements SessionInterface */ private array $data; + /** + * Keys changed since the last save, or null once clear() or hydrate() replaced all of them. + * + * @var array|null + */ + private ?array $changes = []; + public function __construct( private SessionStoreInterface $store, private Uuid $id = new UuidV4(), @@ -44,7 +54,23 @@ public function getId(): Uuid public function save(): bool { - return $this->store->write($this->id, json_encode($this->readData(), \JSON_THROW_ON_ERROR)); + $data = $this->readData(); + + if (null !== $this->changes) { + $data = $this->load(); + foreach (array_keys($this->changes) as $key) { + $this->apply($data, (string) $key); + } + } + + if (!$this->store->write($this->id, json_encode($data, \JSON_THROW_ON_ERROR))) { + return false; + } + + $this->data = $data; + $this->changes = []; + + return true; } public function get(string $key, mixed $default = null): mixed @@ -79,6 +105,7 @@ public function set(string $key, mixed $value, bool $overwrite = true): void if ($overwrite || !isset($data[$lastKey])) { $data[$lastKey] = $value; + $this->trackChange($key); } } @@ -114,12 +141,16 @@ public function forget(string $key): void $data = &$data[$segment]; } - unset($data[$lastKey]); + if (\array_key_exists($lastKey, $data)) { + unset($data[$lastKey]); + $this->trackChange($key); + } } public function clear(): void { $this->data = []; + $this->changes = null; } public function pull(string $key, mixed $default = null): mixed @@ -138,6 +169,7 @@ public function all(): array public function hydrate(array $attributes): void { $this->data = $attributes; + $this->changes = null; } /** @return array */ @@ -151,23 +183,61 @@ public function jsonSerialize(): array */ private function readData(): array { - if (isset($this->data)) { - return $this->data; - } + return $this->data ??= $this->load(); + } + /** + * @return array + */ + private function load(): array + { $rawData = $this->store->read($this->id); if (false === $rawData) { - return $this->data = []; + return []; } // Empty session should not throw $decoded = json_decode($rawData, true); - if (!\is_array($decoded)) { - return $this->data = []; + return \is_array($decoded) ? $decoded : []; + } + + private function trackChange(string $key): void + { + if (null !== $this->changes) { + $this->changes[$key] = true; } + } - return $this->data = $decoded; + /** + * Carries the current value of a changed key over to $data, or removes it there if it was forgotten. + * + * @param array $data + */ + private function apply(array &$data, string $key): void + { + $segments = explode('.', $key); + $lastKey = array_pop($segments); + $source = $this->data; + $target = &$data; + + foreach ($segments as $segment) { + $source = \is_array($source) ? $source[$segment] ?? null : null; + + if (!isset($target[$segment]) || !\is_array($target[$segment])) { + if (!\is_array($source)) { + return; + } + $target[$segment] = []; + } + $target = &$target[$segment]; + } + + if (\is_array($source) && \array_key_exists($lastKey, $source)) { + $target[$lastKey] = $source[$lastKey]; + } else { + unset($target[$lastKey]); + } } } diff --git a/tests/Unit/Server/Session/SessionTest.php b/tests/Unit/Server/Session/SessionTest.php index be045f23..29dad471 100644 --- a/tests/Unit/Server/Session/SessionTest.php +++ b/tests/Unit/Server/Session/SessionTest.php @@ -365,4 +365,120 @@ public function testAllReturnsEmptyArrayForUndecodablePayload(): void $this->assertSame([], $session->all()); } + + public function testSaveKeepsWhatAConcurrentSessionSaved(): void + { + $id = new UuidV4(); + $this->store->write($id, json_encode(['initialized' => true], \JSON_THROW_ON_ERROR)); + + $first = new Session($this->store, $id); + $second = new Session($this->store, $id); + $first->get('initialized'); + $second->get('initialized'); + + $first->set('_mcp.logging_level', 'debug'); + $first->save(); + $second->set('counter', 1); + $second->save(); + + $this->assertSame( + ['initialized' => true, '_mcp' => ['logging_level' => 'debug'], 'counter' => 1], + (new Session($this->store, $id))->all(), + ); + } + + public function testSaveMergesNestedKeysOfConcurrentSessions(): void + { + $id = new UuidV4(); + $first = new Session($this->store, $id); + $second = new Session($this->store, $id); + $first->all(); + $second->all(); + + $first->set('_mcp.responses.1', ['result' => 'a']); + $first->save(); + $second->set('_mcp.responses.2', ['result' => 'b']); + $second->save(); + + $this->assertSame( + ['1' => ['result' => 'a'], '2' => ['result' => 'b']], + (new Session($this->store, $id))->get('_mcp.responses'), + ); + } + + public function testSaveAppliesForgetOntoConcurrentChanges(): void + { + $id = new UuidV4(); + $this->store->write($id, json_encode(['stale' => true], \JSON_THROW_ON_ERROR)); + + $first = new Session($this->store, $id); + $second = new Session($this->store, $id); + $first->all(); + $second->all(); + + $first->forget('stale'); + $first->save(); + $second->set('fresh', true); + $second->save(); + + $this->assertSame(['fresh' => true], (new Session($this->store, $id))->all()); + } + + public function testSaveDoesNotWriteBackUnchangedKeys(): void + { + $id = new UuidV4(); + $this->store->write($id, json_encode(['level' => 'info'], \JSON_THROW_ON_ERROR)); + + $first = new Session($this->store, $id); + $second = new Session($this->store, $id); + $first->all(); + $second->all(); + + $first->set('level', 'debug'); + $first->save(); + $second->set('level', 'error', overwrite: false); + $second->save(); + + $this->assertSame('debug', (new Session($this->store, $id))->get('level')); + } + + public function testSaveSeesConcurrentChangesAfterwards(): void + { + $id = new UuidV4(); + $first = new Session($this->store, $id); + $second = new Session($this->store, $id); + $first->all(); + + $second->set('other', true); + $second->save(); + $first->set('mine', true); + $first->save(); + + $this->assertSame(['other' => true, 'mine' => true], $first->all()); + } + + public function testSaveAfterClearReplacesStoredData(): void + { + $id = new UuidV4(); + $this->store->write($id, json_encode(['stale' => true], \JSON_THROW_ON_ERROR)); + + $session = new Session($this->store, $id); + $session->clear(); + $session->set('fresh', true); + $session->save(); + + $this->assertSame(['fresh' => true], (new Session($this->store, $id))->all()); + } + + public function testSaveAfterHydrateReplacesStoredData(): void + { + $id = new UuidV4(); + $this->store->write($id, json_encode(['stale' => true], \JSON_THROW_ON_ERROR)); + + $session = new Session($this->store, $id); + $session->hydrate(['fresh' => true]); + $session->save(); + + $this->assertSame(['fresh' => true], (new Session($this->store, $id))->all()); + } } From 7e0e38ad488d534c20371c92b020c78c646ce32a Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Sat, 10 Oct 2026 04:17:53 +0200 Subject: [PATCH 2/4] [Server] Add an opt-in per-session lock Concurrent requests of one session take turns instead of losing each other's changes to the same key. --- CHANGELOG.md | 1 + composer.json | 4 +- docs/run/sessions.md | 35 ++++ src/Server/Builder.php | 15 ++ src/Server/Protocol.php | 129 ++++++++++---- src/Server/Session/SessionLockInterface.php | 39 ++++ src/Server/Session/SymfonySessionLock.php | 87 +++++++++ tests/Unit/Server/ProtocolSessionLockTest.php | 167 ++++++++++++++++++ .../Session/Fixture/FiberSessionLock.php | 67 +++++++ .../Server/Session/SymfonySessionLockTest.php | 68 +++++++ 10 files changed, 581 insertions(+), 31 deletions(-) create mode 100644 src/Server/Session/SessionLockInterface.php create mode 100644 src/Server/Session/SymfonySessionLock.php create mode 100644 tests/Unit/Server/ProtocolSessionLockTest.php create mode 100644 tests/Unit/Server/Session/Fixture/FiberSessionLock.php create mode 100644 tests/Unit/Server/Session/SymfonySessionLockTest.php diff --git a/CHANGELOG.md b/CHANGELOG.md index ed4adfb3..48c93614 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,7 @@ All notable changes to `mcp/sdk` will be documented in this file. ----- * Fix concurrent requests of one session overwriting each other's session changes: `Session::save()` now writes only the keys it changed, onto what the store holds at that point. +* Add `Builder::setSessionLock()` and `SymfonySessionLock`, an opt-in lock that serializes concurrent requests of one session, so they no longer lose each other's changes to the same key. * [BC Break] `SchemaValidator` takes an optional `Opis\JsonSchema\Validator` as its first constructor argument, moving `$logger` to second. Pass `logger:` by name. * [BC Break] `SchemaValidator::validateAgainstJsonSchema()` no longer validates an empty array as an object. Pass `new \stdClass()` for an empty object. * Add `Builder::setSchemaValidator()` to configure the validator used for `tools/call` input and output, e.g. with a resolver for external `$ref` schemas. diff --git a/composer.json b/composer.json index fd3ce3d0..6d1a49b9 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 SymfonySessionLock." }, "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/sessions.md b/docs/run/sessions.md index 93f01c7e..bd4a6fae 100644 --- a/docs/run/sessions.md +++ b/docs/run/sessions.md @@ -127,3 +127,38 @@ class RedisSessionStore implements SessionStoreInterface } } ``` + +## Concurrent Requests + +A client can send several requests of one session at the same time, for example parallel tool calls. Each +request loads the session, changes it, and saves the keys it changed. When two requests change different keys, +both changes are kept. When they change the same key, for example a counter in a tool handler, the request that +saves last wins and the other change is lost. + +To prevent this, configure a session lock. A request then holds the lock from loading its session to saving it, +and the other requests of that session wait for it: + +```php +use Mcp\Server\Session\SymfonySessionLock; +use Symfony\Component\Lock\LockFactory; +use Symfony\Component\Lock\Store\FlockStore; + +// composer require symfony/lock +$server = Server::builder() + ->setSession(new FileSessionStore(__DIR__ . '/sessions')) + ->setSessionLock(new SymfonySessionLock(new LockFactory(new FlockStore()))) + ->build(); +``` + +All workers must share the lock store: `FlockStore` works for workers on one host, use e.g. a `RedisStore` +when the workers run on several hosts. + +- `$timeout` (float): Seconds a request waits for the lock (default: `30.0`). After that, it fails with a + `TimeoutException` and the client gets an internal error. Use a value above the duration of your longest + handler. +- `$ttl` (float): Seconds after which the store expires a lock that was never released, e.g. by a crashed + worker (default: `300.0`). + +A request that waits on the client, for example for an elicitation or a sampling result, releases the lock +while it waits. The lock is off by default, and the stdio transport does not need it, because it serves one +client in one process. Implement `SessionLockInterface` to use another locking mechanism. diff --git a/src/Server/Builder.php b/src/Server/Builder.php index 7b68924b..1c35b929 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,17 @@ public function setSession( return $this; } + /** + * Locks a handshake-era session from loading to saving it, so concurrent requests of one session take turns + * instead of losing each other's changes to the same key. Off by default. + */ + public function setSessionLock(?SessionLockInterface $sessionLock): self + { + $this->sessionLock = $sessionLock; + + return $this; + } + /** * @param string[] $scanDirs * @param string[] $excludeDirs @@ -955,6 +969,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/Protocol.php b/src/Server/Protocol.php index bbd25019..75826917 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -28,6 +28,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; @@ -94,6 +95,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(); } @@ -221,6 +223,16 @@ private function doProcessInput(TransportInterface $transport, string $input, ?U return; } + // A request that suspends saves its session and returns, so the lock is free while it waits on the client. + $this->locked($session->getId(), fn () => $this->handleMessages($transport, $messages, $session, $accessToken)); + } + + /** + * @param TransportInterface $transport + * @param array $messages + */ + private function handleMessages(TransportInterface $transport, array $messages, SessionInterface $session, ?AccessToken $accessToken): void + { foreach ($messages as $message) { // Guarded per message so one faulty message cannot suppress the rest of a batch. try { @@ -578,16 +590,23 @@ 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(); + // The polling loop asks many times a second and mostly finds nothing: it only locks when there is something to take. + if (null !== $this->sessionLock && [] === $this->sessionManager->createWithId($sessionId)->get(self::SESSION_OUTGOING_QUEUE, [])) { + return []; } - return $queue; + return $this->locked($sessionId, function () use ($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(); + } + + return $queue; + }); } /** @@ -601,23 +620,16 @@ 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 (null === $responseData) { + if (null !== $this->sessionLock && null === $this->sessionManager->createWithId($sessionId)->get(self::SESSION_RESPONSES.".{$requestId}")) { return null; } - $this->logger->debug('Found and consuming client response.', [ - 'request_id' => $requestId, - 'session_id' => $sessionId->toRfc4122(), - ]); + // Taken under the lock, from a fresh read, before the transport resumes the fiber with it. + $responseData = $this->locked($sessionId, fn (): mixed => $this->consumeResponse($requestId, $sessionId)); - $session->set(self::SESSION_RESPONSES.".{$requestId}", null); - $pending = $session->get(self::SESSION_PENDING_REQUESTS, []); - unset($pending[$requestId]); - $session->set(self::SESSION_PENDING_REQUESTS, $pending); - $session->save(); + if (null === $responseData) { + return null; + } try { if (isset($responseData['error'])) { @@ -636,6 +648,32 @@ public function checkResponse(int $requestId, Uuid $sessionId): Response|Error|n } } + /** + * Removes the client's response to a request from the session, along with the request pending on it. + */ + private function consumeResponse(int $requestId, Uuid $sessionId): mixed + { + $session = $this->sessionManager->createWithId($sessionId); + $responseData = $session->get(self::SESSION_RESPONSES.".{$requestId}"); + + if (null === $responseData) { + return null; + } + + $this->logger->debug('Found and consuming client response.', [ + 'request_id' => $requestId, + 'session_id' => $sessionId->toRfc4122(), + ]); + + $session->set(self::SESSION_RESPONSES.".{$requestId}", null); + $pending = $session->get(self::SESSION_PENDING_REQUESTS, []); + unset($pending[$requestId]); + $session->set(self::SESSION_PENDING_REQUESTS, $pending); + $session->save(); + + return $responseData; + } + /** * Get pending requests for a session. * @@ -672,8 +710,6 @@ public function handleFiberYield(mixed $yieldedValue, ?Uuid $sessionId): ?int 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, @@ -681,17 +717,50 @@ public function handleFiberYield(mixed $yieldedValue, ?Uuid $sessionId): ?int ]); } - try { - if ($yieldedValue instanceof RequestSuspension) { - return $this->sendRequest($yieldedValue->request, $yieldedValue->timeout, $session); + return $this->locked($sessionId, function () use ($yieldedValue, $sessionId): ?int { + $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(); } - $this->sendNotification($yieldedValue->notification, $session); - } finally { - $session->save(); + return null; + }); + } + + /** + * Runs $callback with the session locked, if a session lock is configured. + * + * @template T + * + * @param callable(): T $callback + * + * @return T + */ + private function locked(Uuid $sessionId, callable $callback): mixed + { + if (null === $this->sessionLock) { + return $callback(); } - return null; + $this->sessionLock->acquire($sessionId); + + try { + return $callback(); + } finally { + try { + $this->sessionLock->release($sessionId); + } catch (\Throwable $e) { + // The lock store expires the lock after its TTL; failing here would only discard what the callback did. + $this->logger->error(\sprintf('Failed to release session lock: %s', $e->getMessage()), ['exception' => $e]); + } + } } /** diff --git a/src/Server/Session/SessionLockInterface.php b/src/Server/Session/SessionLockInterface.php new file mode 100644 index 00000000..642fb41e --- /dev/null +++ b/src/Server/Session/SessionLockInterface.php @@ -0,0 +1,39 @@ + + */ +interface SessionLockInterface +{ + /** + * Blocks until the session is locked for the caller. + * + * @throws ExceptionInterface when the lock cannot be acquired, e.g. in time + */ + public function acquire(Uuid $id): void; + + /** + * Releases the lock acquired for the session, if any. + */ + public function release(Uuid $id): void; +} diff --git a/src/Server/Session/SymfonySessionLock.php b/src/Server/Session/SymfonySessionLock.php new file mode 100644 index 00000000..b8c9aabc --- /dev/null +++ b/src/Server/Session/SymfonySessionLock.php @@ -0,0 +1,87 @@ + + */ +final class SymfonySessionLock implements SessionLockInterface +{ + /** + * @var array + */ + private array $locks = []; + + /** + * @param float $timeout seconds to wait for the lock before giving up + * @param float $ttl seconds after which the lock store expires a lock that was not released, e.g. by a crashed worker + */ + public function __construct( + private readonly LockFactory $factory, + private readonly float $timeout = 30.0, + private readonly float $ttl = 300.0, + private readonly string $prefix = 'mcp-session-', + ) { + if ($timeout < 0) { + throw new InvalidArgumentException('timeout must be greater than or equal to 0.'); + } + } + + public function acquire(Uuid $id): void + { + $key = $id->toRfc4122(); + $lock = $this->factory->createLock($this->prefix.$key, $this->ttl); + $deadline = microtime(true) + $this->timeout; + + try { + while (!$lock->acquire()) { + if (microtime(true) >= $deadline) { + throw new TimeoutException(\sprintf('Could not lock session "%s" within %s seconds.', $key, $this->timeout)); + } + + usleep(20000); + } + } catch (LockException $e) { + throw new RuntimeException(\sprintf('Failed to lock session "%s": %s', $key, $e->getMessage()), 0, $e); + } + + $this->locks[$key] = $lock; + } + + public function release(Uuid $id): void + { + $key = $id->toRfc4122(); + + if (!isset($this->locks[$key])) { + return; + } + + $lock = $this->locks[$key]; + unset($this->locks[$key]); + + try { + $lock->release(); + } catch (LockException $e) { + throw new RuntimeException(\sprintf('Failed to unlock session "%s": %s', $key, $e->getMessage()), 0, $e); + } + } +} diff --git a/tests/Unit/Server/ProtocolSessionLockTest.php b/tests/Unit/Server/ProtocolSessionLockTest.php new file mode 100644 index 00000000..d827a729 --- /dev/null +++ b/tests/Unit/Server/ProtocolSessionLockTest.php @@ -0,0 +1,167 @@ +assertSame(1, $this->incrementConcurrently(null)); + } + + #[TestDox('with a session lock, concurrent requests take turns and keep both increments')] + public function testSessionLockKeepsConcurrentIncrements(): void + { + $this->assertSame(2, $this->incrementConcurrently(new FiberSessionLock())); + } + + #[TestDox('a suspended request releases the session lock while it waits on the client')] + public function testSuspendedRequestReleasesSessionLock(): void + { + [$sessions, $sessionId] = $this->createSession(new InterleavingSessionStore()); + $lock = new FiberSessionLock(); + + $handler = $this->createMock(RequestHandlerInterface::class); + $handler->method('supports')->willReturn(true); + $handler->method('handle')->willReturnCallback(static function (Request $request, SessionInterface $session): Response { + $answer = \Fiber::suspend(new RequestSuspension(new PingRequest(), $session->getId()->toRfc4122(), 5)); + + return new Response($request->getId(), $answer instanceof Response ? $answer->result : null); + }); + + $fiber = null; + $transport = $this->createMock(TransportInterface::class); + $transport->expects($this->once())->method('attachFiberToSession') + ->willReturnCallback(static function (\Fiber $attached) use (&$fiber): void { + $fiber = $attached; + }); + + $waiting = new Protocol([$handler], [], MessageFactory::make(), $sessions, sessionLock: $lock); + $waiting->processInput($transport, \sprintf(self::INCREMENT, 1), $sessionId); + + $this->assertInstanceOf(\Fiber::class, $fiber); + $this->assertTrue($fiber->isSuspended()); + $this->assertFalse($lock->isLocked($sessionId)); + + // The client's answer arrives on another worker, which can lock the session; outside of a fiber it could not wait. + $answering = new Protocol([], [], MessageFactory::make(), $sessions, sessionLock: $lock); + $answering->processInput(new RecordingTransport(), '{"jsonrpc": "2.0", "id": 1000, "result": {"ok": true}}', $sessionId); + + // The polling worker takes the answer under the lock before resuming the fiber with it. + $answer = $waiting->checkResponse(1000, $sessionId); + $this->assertInstanceOf(Response::class, $answer); + $this->assertFalse($lock->isLocked($sessionId)); + $this->assertSame(['acquire', 'release', 'acquire', 'release', 'acquire', 'release'], $lock->log); + + $fiber->resume($answer); + $this->assertTrue($fiber->isTerminated()); + } + + #[TestDox('a request that cannot lock its session in time is answered with an internal error')] + public function testLockTimeoutAnswersWithInternalError(): void + { + [$sessions, $sessionId] = $this->createSession(new InterleavingSessionStore()); + $factory = new LockFactory(new InMemoryStore()); + $held = $factory->createLock('mcp-session-'.$sessionId->toRfc4122()); + $this->assertTrue($held->acquire()); + + $transport = new RecordingTransport(); + $protocol = new Protocol([$this->createIncrementHandler()], [], MessageFactory::make(), $sessions, sessionLock: new SymfonySessionLock($factory, timeout: 0.05)); + $protocol->processInput($transport, \sprintf(self::INCREMENT, 1), $sessionId); + + $this->assertCount(1, $transport->sent); + $this->assertSame(Error::INTERNAL_ERROR, json_decode($transport->sent[0]['message'], true)['error']['code']); + $this->assertNull($sessions->createWithId($sessionId)->get('counter')); + } + + private function incrementConcurrently(?SessionLockInterface $lock): int + { + $store = new InterleavingSessionStore(); + [$sessions, $sessionId] = $this->createSession($store); + + $first = new Protocol([$this->createIncrementHandler()], [], MessageFactory::make(), $sessions, sessionLock: $lock); + $second = new Protocol([$this->createIncrementHandler()], [], MessageFactory::make(), $sessions, sessionLock: $lock); + + // The second request arrives right after the first one read the session. Each runs in its + // own fiber, so a request waiting on the lock suspends where a worker would block. + $store->interleaveAfterNextRead(static function () use ($second, $sessionId): void { + (new \Fiber(static fn () => $second->processInput(new RecordingTransport(), \sprintf(self::INCREMENT, 2), $sessionId)))->start(); + }); + + (new \Fiber(static fn () => $first->processInput(new RecordingTransport(), \sprintf(self::INCREMENT, 1), $sessionId)))->start(); + + return $sessions->createWithId($sessionId)->get('counter'); + } + + /** + * @return array{SessionManager, Uuid} + */ + private function createSession(InterleavingSessionStore $store): array + { + $sessions = new SessionManager($store, gcProbability: 0); + $sessionId = Uuid::v4(); + $session = $sessions->createWithId($sessionId); + $session->set('initialized', true); + $session->save(); + + return [$sessions, $sessionId]; + } + + /** + * @return RequestHandlerInterface + */ + private function createIncrementHandler(): RequestHandlerInterface + { + return new class implements RequestHandlerInterface { + public function supports(Request $request): bool + { + return true; + } + + public function handle(Request $request, SessionInterface $session): Response + { + $session->set('counter', $session->get('counter', 0) + 1); + + return new Response($request->getId(), new EmptyResult()); + } + }; + } +} diff --git a/tests/Unit/Server/Session/Fixture/FiberSessionLock.php b/tests/Unit/Server/Session/Fixture/FiberSessionLock.php new file mode 100644 index 00000000..288d8480 --- /dev/null +++ b/tests/Unit/Server/Session/Fixture/FiberSessionLock.php @@ -0,0 +1,67 @@ + */ + private array $held = []; + + /** @var array>> */ + private array $waiting = []; + + /** @var list */ + public array $log = []; + + public function acquire(Uuid $id): void + { + $key = $id->toRfc4122(); + + while (isset($this->held[$key])) { + if (null === $fiber = \Fiber::getCurrent()) { + throw new TimeoutException('The session is locked and the caller cannot wait.'); + } + + $this->waiting[$key][] = $fiber; + \Fiber::suspend(); + } + + $this->held[$key] = true; + $this->log[] = 'acquire'; + } + + public function release(Uuid $id): void + { + $key = $id->toRfc4122(); + unset($this->held[$key]); + $this->log[] = 'release'; + + if ([] !== ($this->waiting[$key] ?? [])) { + array_shift($this->waiting[$key])->resume(); + } + } + + public function isLocked(Uuid $id): bool + { + return isset($this->held[$id->toRfc4122()]); + } +} diff --git a/tests/Unit/Server/Session/SymfonySessionLockTest.php b/tests/Unit/Server/Session/SymfonySessionLockTest.php new file mode 100644 index 00000000..1981906f --- /dev/null +++ b/tests/Unit/Server/Session/SymfonySessionLockTest.php @@ -0,0 +1,68 @@ +acquire($id); + + $this->expectException(TimeoutException::class); + $this->expectExceptionMessage(\sprintf('Could not lock session "%s" within 0.05 seconds.', $id->toRfc4122())); + + (new SymfonySessionLock($factory, timeout: 0.05))->acquire($id); + } + + public function testReleaseLetsAnotherWorkerAcquire(): void + { + $factory = new LockFactory(new InMemoryStore()); + $id = Uuid::v4(); + $first = new SymfonySessionLock($factory); + $first->acquire($id); + $first->release($id); + + $second = new SymfonySessionLock($factory, timeout: 0.0); + $second->acquire($id); + $second->release($id); + + $this->expectNotToPerformAssertions(); + } + + public function testLocksSessionsIndependently(): void + { + $factory = new LockFactory(new InMemoryStore()); + $holder = new SymfonySessionLock($factory); + $holder->acquire(Uuid::v4()); + + (new SymfonySessionLock($factory, timeout: 0.0))->acquire(Uuid::v4()); + + $this->expectNotToPerformAssertions(); + } + + public function testReleaseWithoutAcquireIsNoop(): void + { + (new SymfonySessionLock(new LockFactory(new InMemoryStore())))->release(Uuid::v4()); + + $this->expectNotToPerformAssertions(); + } +} From b7b211688244f08dfc2b1ba37a9e489d8c984b8c Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Mon, 12 Oct 2026 22:38:19 +0200 Subject: [PATCH 3/4] [Server] Address review: note that overlapping saves can still lose a change --- src/Server/Session/Session.php | 1 + 1 file changed, 1 insertion(+) diff --git a/src/Server/Session/Session.php b/src/Server/Session/Session.php index 62e64604..b44296ed 100644 --- a/src/Server/Session/Session.php +++ b/src/Server/Session/Session.php @@ -17,6 +17,7 @@ /** * Saving writes only the keys changed since the last save, onto what the store holds by then: * concurrent requests of one session keep each other's changes as long as they change different keys. + * The store reads and writes in separate steps, so two saves that overlap can still lose one of them. * * @author Kyrian Obikwelu */ From a4b1337d865508f8deb3dbd1dd3170c02965cf88 Mon Sep 17 00:00:00 2001 From: Christopher Hertel Date: Mon, 12 Oct 2026 22:41:27 +0200 Subject: [PATCH 4/4] [Server] Address review: lock session destruction, document the lock TTL --- docs/run/sessions.md | 7 ++++--- src/Server/Protocol.php | 3 ++- src/Server/Session/SymfonySessionLock.php | 3 ++- tests/Unit/Server/ProtocolSessionLockTest.php | 21 +++++++++++++++++++ 4 files changed, 29 insertions(+), 5 deletions(-) diff --git a/docs/run/sessions.md b/docs/run/sessions.md index bd4a6fae..6e212e59 100644 --- a/docs/run/sessions.md +++ b/docs/run/sessions.md @@ -132,8 +132,8 @@ class RedisSessionStore implements SessionStoreInterface A client can send several requests of one session at the same time, for example parallel tool calls. Each request loads the session, changes it, and saves the keys it changed. When two requests change different keys, -both changes are kept. When they change the same key, for example a counter in a tool handler, the request that -saves last wins and the other change is lost. +both changes are kept, unless both requests save at the same moment. When they change the same key, for example a +counter in a tool handler, the request that saves last wins and the other change is lost. To prevent this, configure a session lock. A request then holds the lock from loading its session to saving it, and the other requests of that session wait for it: @@ -157,7 +157,8 @@ when the workers run on several hosts. `TimeoutException` and the client gets an internal error. Use a value above the duration of your longest handler. - `$ttl` (float): Seconds after which the store expires a lock that was never released, e.g. by a crashed - worker (default: `300.0`). + worker (default: `300.0`). The lock is not refreshed while a request holds it, so use a value above the + duration of your longest handler as well, otherwise another request can take the lock while the first still runs. A request that waits on the client, for example for an elicitation or a sampling result, releases the lock while it waits. The lock is off by default, and the stdio transport does not need it, because it serves one diff --git a/src/Server/Protocol.php b/src/Server/Protocol.php index 75826917..fa450e09 100644 --- a/src/Server/Protocol.php +++ b/src/Server/Protocol.php @@ -871,7 +871,8 @@ private function resolveSession(TransportInterface $transport, ?Uuid $sessionId, */ public function destroySession(Uuid $sessionId): void { - $this->sessionManager->destroy($sessionId); + // Waits for a request working on the session, so its save does not bring the session back. + $this->locked($sessionId, fn (): bool => $this->sessionManager->destroy($sessionId)); $this->logger->info('Session destroyed.', ['session_id' => $sessionId->toRfc4122()]); } } diff --git a/src/Server/Session/SymfonySessionLock.php b/src/Server/Session/SymfonySessionLock.php index b8c9aabc..423a92ac 100644 --- a/src/Server/Session/SymfonySessionLock.php +++ b/src/Server/Session/SymfonySessionLock.php @@ -33,7 +33,8 @@ final class SymfonySessionLock implements SessionLockInterface /** * @param float $timeout seconds to wait for the lock before giving up - * @param float $ttl seconds after which the lock store expires a lock that was not released, e.g. by a crashed worker + * @param float $ttl seconds after which the lock store expires a lock that was not released, e.g. by a crashed worker; + * it is not refreshed, so keep it above the duration of the longest handler */ public function __construct( private readonly LockFactory $factory, diff --git a/tests/Unit/Server/ProtocolSessionLockTest.php b/tests/Unit/Server/ProtocolSessionLockTest.php index d827a729..fa81feed 100644 --- a/tests/Unit/Server/ProtocolSessionLockTest.php +++ b/tests/Unit/Server/ProtocolSessionLockTest.php @@ -112,6 +112,27 @@ public function testLockTimeoutAnswersWithInternalError(): void $this->assertNull($sessions->createWithId($sessionId)->get('counter')); } + #[TestDox('a session ended during a locked request stays destroyed after that request saves')] + public function testSessionEndWaitsForLockedRequest(): void + { + $store = new InterleavingSessionStore(); + [$sessions, $sessionId] = $this->createSession($store); + $lock = new FiberSessionLock(); + + $working = new Protocol([$this->createIncrementHandler()], [], MessageFactory::make(), $sessions, sessionLock: $lock); + $ending = new Protocol([], [], MessageFactory::make(), $sessions, sessionLock: $lock); + + // The client ends the session on another worker while the request is working on it. + $store->interleaveAfterNextRead(static function () use ($ending, $sessionId): void { + (new \Fiber(static fn () => $ending->destroySession($sessionId)))->start(); + }); + + (new \Fiber(static fn () => $working->processInput(new RecordingTransport(), \sprintf(self::INCREMENT, 1), $sessionId)))->start(); + + $this->assertFalse($sessions->exists($sessionId)); + $this->assertFalse($lock->isLocked($sessionId)); + } + private function incrementConcurrently(?SessionLockInterface $lock): int { $store = new InterleavingSessionStore();