diff --git a/CHANGELOG.md b/CHANGELOG.md index b2685277..eeba9ce7 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 c81ef103..024aa4e3 100644 --- a/src/Server/Session/Session.php +++ b/src/Server/Session/Session.php @@ -15,6 +15,10 @@ 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. + * The store reads and writes in separate steps, so two saves that overlap can still lose one of them. + * * @author Kyrian Obikwelu */ class Session implements SessionInterface @@ -36,6 +40,13 @@ class Session implements SessionInterface */ private bool $stored = false; + /** + * 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(), @@ -49,17 +60,27 @@ public function getId(): Uuid public function save(): bool { - $data = json_encode($this->readData(), \JSON_THROW_ON_ERROR); + $data = $this->readData(); // A session destroyed since it was read, e.g. by a DELETE while a request ran, must stay gone. if ($this->stored && !$this->store->exists($this->id)) { return false; } - if (!$this->store->write($this->id, $data)) { + 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 $this->stored = true; } @@ -95,6 +116,7 @@ public function set(string $key, mixed $value, bool $overwrite = true): void if ($overwrite || !isset($data[$lastKey])) { $data[$lastKey] = $value; + $this->trackChange($key); } } @@ -130,12 +152,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 @@ -154,6 +180,7 @@ public function all(): array public function hydrate(array $attributes): void { $this->data = $attributes; + $this->changes = null; } /** @return array */ @@ -167,14 +194,18 @@ 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 []; } $this->stored = true; @@ -182,10 +213,44 @@ private function readData(): array // 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 0eb29e58..70323ca3 100644 --- a/tests/Unit/Server/Session/SessionTest.php +++ b/tests/Unit/Server/Session/SessionTest.php @@ -402,6 +402,122 @@ 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()); + } + private static function arrayCache(): CacheInterface { return new class implements CacheInterface {