diff --git a/CHANGELOG.md b/CHANGELOG.md index e34852a9..48c93614 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,8 @@ 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. +* 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/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/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/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()); + } } 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(); + } +}