[Server] Stop concurrent writes from corrupting session files

src/Server/Session/FileSessionStore.php

@@ -85,15 +85,21 @@ public function read(Uuid $id): string|false
             return false;
         }
 
+        if ('' === $data) {
+            $this->logger->warning('Ignored an empty session file.', ['path' => $path]);
+
+            return false;
+        }
+
         return $data;
     }
 
     public function write(Uuid $id, string $data): bool
     {
         $path = $this->pathFor($id);
 
-        $tmp = $path.'.tmp';
-        if (false === @file_put_contents($tmp, $data, \LOCK_EX)) {
+        $tmp = $path.'.'.bin2hex(random_bytes(6)).'.tmp';
+        if (false === @file_put_contents($tmp, $data)) {
             $this->logger->warning('Failed to write session file.', [
                 'path' => $tmp,
                 'error' => error_get_last()['message'] ?? 'unknown',
@@ -137,7 +143,8 @@ public function destroy(Uuid $id): bool
     }
 
     /**
-     * Remove sessions older than the configured TTL.
+     * Remove sessions older than the configured TTL, along with the temporary files
+     * of writes that never made it into place.
      * Returns an array of deleted session IDs (UUID instances).
      */
     public function gc(): array
@@ -161,8 +168,8 @@ public function gc(): array
                 continue;
             }
 
-            // Only delete files this store owns: sessions are named by their RFC 4122 UUID
-            if (!Uuid::isValid($entry)) {
+            $isSession = Uuid::isValid($entry);
+            if (!$isSession && !$this->isTemporaryFile($entry)) {
                 continue;
             }
 
@@ -182,7 +189,9 @@ public function gc(): array
                     continue;
                 }
 
-                $deleted[] = Uuid::fromString($entry);
+                if ($isSession) {
+                    $deleted[] = Uuid::fromString($entry);
+                }
             }
         }
 
@@ -191,6 +200,15 @@ public function gc(): array
         return $deleted;
     }
 
+    private function isTemporaryFile(string $entry): bool
+    {
+        if (1 !== preg_match('/^(?<id>[^.]+)\\.[0-9a-f]{12}\\.tmp$/', $entry, $matches)) {
+            return false;
+        }
+
+        return Uuid::isValid($matches['id']);
+    }
+
     private function pathFor(Uuid $id): string
     {
         return $this->directory.\DIRECTORY_SEPARATOR.$id->toRfc4122();

src/Server/Session/Session.php

@@ -162,7 +162,8 @@ private function readData(): array
             return $this->data = [];
         }
 
-        $decoded = json_decode($rawData, true, flags: \JSON_THROW_ON_ERROR);
+        // Empty session should not throw
+        $decoded = json_decode($rawData, true);
 
         if (!\is_array($decoded)) {
             return $this->data = [];

tests/Unit/Server/Session/FileSessionStoreTest.php

@@ -86,6 +86,30 @@ public function testGcLeavesForeignFilesAlone(): void
         $this->assertFileExists($foreign);
     }
 
+    #[TestDox('gc() collects the temporary files of writes that never made it into place')]
+    public function testGcCollectsOrphanedTemporaryFiles(): void
+    {
+        $store = new FileSessionStore($this->directory, ttl: 60);
+        $id = new UuidV4();
+        $store->write($id, 'payload');
+
+        $orphan = $this->directory.\DIRECTORY_SEPARATOR.$id->toRfc4122().'.0123456789ab.tmp';
+        file_put_contents($orphan, 'half a payload');
+        touch($orphan, time() - 120);
+
+        $fresh = $this->directory.\DIRECTORY_SEPARATOR.(new UuidV4())->toRfc4122().'.ba9876543210.tmp';
+        file_put_contents($fresh, 'a write still in flight');
+
+        $deleted = $store->gc();
+
+        // The orphan belongs to a session that is still alive, so collecting it is not a session
+        // deletion, and a write that is still running must be left alone.
+        $this->assertSame([], $deleted);
+        $this->assertFileDoesNotExist($orphan);
+        $this->assertFileExists($fresh);
+        $this->assertFileExists($this->directory.\DIRECTORY_SEPARATOR.$id->toRfc4122());
+    }
+
     #[TestDox('rejects an unwritable directory with the SDK\'s own exception')]
     public function testUnwritableDirectoryThrowsPackageException(): void
     {
@@ -207,6 +231,117 @@ public function testHappyPathLogsNothing(): void
 
         $this->assertSame([], $logger->warnings);
     }
+
+    #[TestDox('reports a zero-byte session file as nothing read, and says so')]
+    public function testReadReportsZeroByteFileAsNothingRead(): void
+    {
+        $logger = new WarningCollectingLogger();
+        $store = new FileSessionStore($this->directory, logger: $logger);
+        $id = new UuidV4();
+        $store->write($id, '{"initialized":true}');
+
+        $path = $this->directory.\DIRECTORY_SEPARATOR.$id->toRfc4122();
+        file_put_contents($path, '');
+
+        $this->assertFalse($store->read($id));
+        $this->assertCount(1, $logger->warnings);
+        $this->assertSame('Ignored an empty session file.', $logger->warnings[0]['message']);
+        $this->assertSame($path, $logger->warnings[0]['context']['path']);
+    }
+
+    #[TestDox('never serves a partial payload while another process writes the same session')]
+    public function testConcurrentWritesNeverPublishAPartialPayload(): void
+    {
+        if (!\function_exists('proc_open')) {
+            $this->markTestSkipped('proc_open() is needed to run truly concurrent writers.');
+        }
+
+        $store = new FileSessionStore($this->directory);
+        $id = new UuidV4();
+        $store->write($id, '{"initialized":true}');
+
+        $writers = [];
+        for ($i = 0; $i < 3; ++$i) {
+            $writers[] = $this->spawnWriter($id, $i);
+        }
+
+        // Read the session while the other processes write it. Every payload the store hands out
+        // has to be complete: a writer must never be able to publish a file another one is still
+        // filling, and never one it has just emptied.
+        $reads = 0;
+        $partial = 0;
+        $exitCodes = [];
+        $deadline = microtime(true) + 20.0;
+        do {
+            $raw = $store->read($id);
+            ++$reads;
+
+            if (false !== $raw && !\is_array(json_decode($raw, true))) {
+                ++$partial;
+            }
+
+            foreach ($writers as $i => $writer) {
+                // Take the exit code from the first status that reports the writer as gone: before
+                // PHP 8.3, that call reaps the process, and every later one reports -1 instead.
+                if (!isset($exitCodes[$i]) && !($status = proc_get_status($writer))['running']) {
+                    $exitCodes[$i] = $status['exitcode'];
+                }
+            }
+        } while (\count($exitCodes) < \count($writers) && microtime(true) < $deadline);
+
+        foreach ($writers as $writer) {
+            proc_close($writer);
+        }
+
+        // The writers finish in whatever order they like.
+        ksort($exitCodes);
+
+        $this->assertSame([0, 0, 0], $exitCodes, 'The concurrent writers did not all run to completion.');
+
+        foreach (array_keys($writers) as $i) {
+            $this->assertSame('', file_get_contents($this->directory.\DIRECTORY_SEPARATOR.'writer-'.$i.'.err'));
+        }
+
+        $this->assertSame(0, $partial, \sprintf('%d of %d reads returned a partial session payload.', $partial, $reads));
+    }
+
+    /**
+     * Writes the same session from another process, so the writes really do overlap.
+     *
+     * @return resource
+     */
+    private function spawnWriter(UuidV4 $id, int $index)
+    {
+        $script = <<<'PHP'
+            <?php
+
+            require $argv[1];
+
+            $store = new Mcp\Server\Session\FileSessionStore($argv[2]);
+            $id = Symfony\Component\Uid\Uuid::fromString($argv[3]);
+            $payload = json_encode(['queue' => array_fill(0, 200, str_repeat('x', 200))]);
+
+            for ($i = 0; $i < 200; ++$i) {
+                $store->write($id, $payload);
+            }
+            PHP;
+
+        $file = $this->directory.\DIRECTORY_SEPARATOR.'writer.php';
+        file_put_contents($file, $script);
+
+        $process = proc_open(
+            [\PHP_BINARY, $file, \dirname(__DIR__, 4).'/vendor/autoload.php', $this->directory, $id->toRfc4122()],
+            [
+                1 => ['file', $this->directory.\DIRECTORY_SEPARATOR.'writer-'.$index.'.out', 'w'],
+                2 => ['file', $this->directory.\DIRECTORY_SEPARATOR.'writer-'.$index.'.err', 'w'],
+            ],
+            $pipes,
+        );
+
+        $this->assertIsResource($process);
+
+        return $process;
+    }
 }
 
 /**

tests/Unit/Server/Session/SessionTest.php

@@ -326,4 +326,43 @@ public function testAllReturnsEmptyArrayForNullPayload(): void
 
         $this->assertSame([], $result);
     }
+
+    public function testGetReturnsDefaultForEmptyPayload(): void
+    {
+        $store = $this->getMockBuilder(InMemorySessionStore::class)
+            ->disableOriginalConstructor()
+            ->onlyMethods(['read'])
+            ->getMock();
+        $store->expects($this->once())->method('read')->willReturn('');
+
+        $session = new Session($store);
+
+        $this->assertSame([], $session->get('outgoing_queue', []));
+    }
+
+    public function testAllReturnsEmptyArrayForEmptyPayload(): void
+    {
+        $store = $this->getMockBuilder(InMemorySessionStore::class)
+            ->disableOriginalConstructor()
+            ->onlyMethods(['read'])
+            ->getMock();
+        $store->expects($this->once())->method('read')->willReturn('');
+
+        $session = new Session($store);
+
+        $this->assertSame([], $session->all());
+    }
+
+    public function testAllReturnsEmptyArrayForUndecodablePayload(): void
+    {
+        $store = $this->getMockBuilder(InMemorySessionStore::class)
+            ->disableOriginalConstructor()
+            ->onlyMethods(['read'])
+            ->getMock();
+        $store->expects($this->once())->method('read')->willReturn('{"initialized":');
+
+        $session = new Session($store);
+
+        $this->assertSame([], $session->all());
+    }
 }