test: validate M3 imports and locking on PostgreSQL

Add isolated PostgreSQL fixtures, two-session locking coverage, and a disposable remote Docker test runner. Correct the canonical-root fixture and document verified SQLite and PostgreSQL results.
2026-10-12 13:19:10 -05:00
1 parent 9d277d9e72
commit c1ef2ee97a
@@ -42,9 +42,13 @@ final class ImportPipelineTest extends TestCase
    protected function tearDown(): void
    {
        unset($this->pdo, $this->jobs, $this->catalog);
-        @unlink($this->database);
-        RepositoryDatabase::remove($this->library);
-        RepositoryDatabase::remove($this->source);
+        RepositoryDatabase::cleanup($this->database);
+        if (isset($this->library)) {
+            RepositoryDatabase::remove($this->library);
+        }
+        if (isset($this->source)) {
+            RepositoryDatabase::remove($this->source);
+        }
    }

    private function service(?SpaceChecker $space = null): ImportService
@@ -374,22 +378,15 @@ final class ImportPipelineTest extends TestCase

    public function testWorkerUsesCanonicalLibraryRootForRelativePaths(): void
    {
-        // Test that the worker uses canonical (realpath) library root for
-        // computing relative paths, not the raw string which might have
-        // symlinks or case differences.
-        //
-        // Create library with a symlink in the path
-        $realLibrary = $this->library;
-        $linkLibrary = $realLibrary . '-link';
-        if (!@symlink($realLibrary, $linkLibrary)) {
-            self::markTestSkipped('Host cannot create symlinks.');
-        }
+        // A trailing separator is safe but differs from the canonical root.
+        // Symlink roots are deliberately rejected by PathSafety.
+        $rawLibrary = $this->library . DIRECTORY_SEPARATOR;

        $this->writeStl('planes/wing.stl');
        $job = $this->service()->import($this->source);

-        // Create worker with symlinked library root
-        $safety = new PathSafety($linkLibrary);
+        $safety = new PathSafety($rawLibrary);
+        self::assertNotSame($rawLibrary, $safety->libraryRoot());
        $sourceSafety = new PathSafety($this->source);
        $worker = new ImportWorker(
            $this->jobs,
@@ -399,20 +396,17 @@ final class ImportPipelineTest extends TestCase
            new StlValidator($sourceSafety),
            $safety,
            new SpaceChecker(),
-            $linkLibrary,  // Symlinked path
+            $rawLibrary,
            new NullLogger(),
        );

        $worker->runOnce();
        $item = $this->jobs->items($job)[0];

-        // The managed_path should be relative to the CANONICAL root, not the symlink
-        self::assertStringNotContainsString('link', $item['managed_path']);
+        self::assertSame('Completed', $item['status']);
+        self::assertStringStartsWith('planes/', $item['managed_path']);
        $published = $this->library . DIRECTORY_SEPARATOR
            . str_replace('/', DIRECTORY_SEPARATOR, $item['managed_path']);
        self::assertFileExists($published);
-
-        // Cleanup
-        unlink($linkLibrary);
    }
 }
@@ -0,0 +1,162 @@
+<?php
+
+declare(strict_types=1);
+
+namespace MeshVault\Tests\Integration;
+
+use MeshVault\Repository\PdoImportJobRepository;
+use PDO;
+use PDOException;
+use PHPUnit\Framework\TestCase;
+
+/** Opt in with MESHVAULT_TEST_PG_DSN (and optional PG_USER / PG_PASSWORD with the same prefix). */
+final class PostgresRepositoryTest extends TestCase
+{
+    private string $database;
+    private PDO $pdo;
+    private PDO $other;
+
+    protected function setUp(): void
+    {
+        if (!RepositoryDatabase::postgresConfigured()) {
+            self::markTestSkipped('Set MESHVAULT_TEST_PG_DSN to run PostgreSQL concurrency tests.');
+        }
+        // Configured connection/migration failures must be errors, never skips.
+        $this->database = tempnam(sys_get_temp_dir(), 'meshvault-pg-');
+        $this->pdo = RepositoryDatabase::migrate($this->database);
+        $this->other = RepositoryDatabase::connect($this->database);
+        foreach ([$this->pdo, $this->other] as $pdo) {
+            $pdo->exec("SET lock_timeout = '250ms'");
+            $pdo->exec("SET statement_timeout = '5s'");
+        }
+    }
+
+    protected function tearDown(): void
+    {
+        foreach (['pdo', 'other'] as $property) {
+            if (isset($this->$property) && $this->$property->inTransaction()) {
+                $this->$property->rollBack();
+            }
+            unset($this->$property);
+        }
+        if (isset($this->database)) {
+            RepositoryDatabase::cleanup($this->database);
+        }
+    }
+
+    public function testMigrationsAndCleanupAreIsolatedToTheGeneratedSchema(): void
+    {
+        $path = tempnam(sys_get_temp_dir(), 'meshvault-pg-isolated-');
+        try {
+            $isolated = RepositoryDatabase::migrate($path);
+            $schema = $isolated->query('SELECT current_schema()')->fetchColumn();
+            $original = $this->pdo->query('SELECT current_schema()')->fetchColumn();
+            self::assertMatchesRegularExpression('/^meshvault_test_[a-f0-9]{32}$/', $schema);
+            self::assertNotSame($original, $schema);
+            self::assertSame($original, $this->other->query('SELECT current_schema()')->fetchColumn());
+            self::assertNotSame(
+                $this->pdo->query('SELECT pg_backend_pid()')->fetchColumn(),
+                $this->other->query('SELECT pg_backend_pid()')->fetchColumn(),
+            );
+            $jobs = new PdoImportJobRepository($this->pdo);
+            $job = $jobs->create('/survives-cleanup', []);
+            self::assertSame(0, (int) $isolated->query('SELECT COUNT(*) FROM import_jobs')->fetchColumn());
+            unset($isolated);
+            RepositoryDatabase::cleanup($path);
+            $exists = $this->pdo->prepare('SELECT COUNT(*) FROM pg_namespace WHERE nspname = ?');
+            $exists->execute([$schema]);
+            self::assertSame(0, (int) $exists->fetchColumn());
+            self::assertSame('/survives-cleanup', $jobs->get($job)['source_path']);
+            // An unregistered path is never interpreted as a schema identifier.
+            RepositoryDatabase::cleanup($schema);
+        } finally {
+            unset($isolated);
+            RepositoryDatabase::cleanup($path);
+        }
+    }
+
+    public function testOverlappingClaimsSkipLockedJobsAndReturnNullWhenAllAreLocked(): void
+    {
+        $jobs = new PdoImportJobRepository($this->pdo);
+        $other = new PdoImportJobRepository($this->other);
+        $first = $jobs->create('/first', []);
+        $second = $jobs->create('/second', []);
+
+        // The outer transaction must retain claimNext's row lock after it returns.
+        $this->pdo->beginTransaction();
+        self::assertSame($first, (int) $jobs->claimNext()['id']);
+        self::assertTrue($this->pdo->inTransaction());
+        self::assertSame('Pending', $other->get($first)['status']);
+        self::assertSame($second, (int) $other->claimNext()['id']);
+        self::assertNull($other->claimNext());
+        self::assertTrue($this->pdo->inTransaction());
+        $this->pdo->commit();
+        self::assertSame('Copying', $other->get($first)['status']);
+        self::assertSame('Copying', $other->get($second)['status']);
+    }
+
+    public function testRolledBackClaimCanBeReclaimedByTheOtherConnection(): void
+    {
+        $jobs = new PdoImportJobRepository($this->pdo);
+        $other = new PdoImportJobRepository($this->other);
+        $job = $jobs->create('/rollback', []);
+        $this->pdo->beginTransaction();
+        self::assertSame($job, (int) $jobs->claimNext()['id']);
+        self::assertNull($other->claimNext());
+        $this->pdo->rollBack();
+        self::assertSame('Pending', $other->get($job)['status']);
+        self::assertNull($other->get($job)['started_at']);
+        self::assertSame($job, (int) $other->claimNext()['id']);
+        self::assertNull($jobs->claimNext());
+    }
+
+    public function testTerminalNoOpStillAcquiresExplicitItemReadLock(): void
+    {
+        $jobs = new PdoImportJobRepository($this->pdo);
+        $other = new PdoImportJobRepository($this->other);
+        $job = $jobs->create('/terminal', [['source_path' => '/terminal/a', 'size' => 7]]);
+        $item = (int) $jobs->items($job)[0]['id'];
+        $jobs->finishItem($item, 'Completed');
+        $this->pdo->beginTransaction();
+        $statement = $this->pdo->prepare('SELECT id FROM import_items WHERE id = ? FOR SHARE');
+        $statement->execute([$item]);
+        $statement->closeCursor();
+        try {
+            // Already terminal: no UPDATE occurs, so only SELECT FOR UPDATE can block.
+            $other->finishItem($item, 'Completed');
+            self::fail('The explicit item read lock must conflict with FOR SHARE.');
+        } catch (PDOException $error) {
+            self::assertSame('55P03', $error->getCode());
+        }
+        self::assertFalse($this->other->inTransaction());
+        $this->pdo->commit();
+        $other->finishItem($item, 'Completed');
+        self::assertSame(1, (int) $jobs->get($job)['processed_files']);
+    }
+
+    public function testItemLockTimesOutWithoutPublishingProgressAndCanBeRetried(): void
+    {
+        $jobs = new PdoImportJobRepository($this->pdo);
+        $other = new PdoImportJobRepository($this->other);
+        $job = $jobs->create('/item', [['source_path' => '/item/a', 'size' => 7]]);
+        $jobs->claimNext();
+        $item = (int) $jobs->items($job)[0]['id'];
+        $this->pdo->beginTransaction();
+        $jobs->planItem($item, 'a.stl', str_repeat('a', 64), 7);
+        self::assertTrue($this->pdo->inTransaction());
+        try {
+            $other->finishItem($item, 'Completed');
+            self::fail('A concurrent item transition must wait for the item lock.');
+        } catch (PDOException $error) {
+            self::assertSame('55P03', $error->getCode(), 'Expected PostgreSQL lock_not_available.');
+        }
+        self::assertFalse($this->other->inTransaction());
+        self::assertSame('Pending', $other->items($job)[0]['status']);
+        self::assertSame(0, (int) $other->get($job)['processed_files']);
+        $this->pdo->commit();
+        $other->finishItem($item, 'Completed');
+        self::assertSame('Completed', $jobs->items($job)[0]['status']);
+        self::assertSame(1, (int) $jobs->get($job)['processed_files']);
+        self::assertSame(7, (int) $jobs->get($job)['copied_bytes']);
+    }
+}
@@ -10,27 +10,102 @@ use Phinx\Migration\Manager;
 use Symfony\Component\Console\Input\ArrayInput;
 use Symfony\Component\Console\Output\NullOutput;

-/** Runs the production migrations against SQLite and manages scratch directories. */
+/** Runs production migrations against SQLite, or an explicitly configured PostgreSQL database. */
 final class RepositoryDatabase
 {
+    /** @var array<string, array{schema: string, connection: PDO}> */
+    private static array $schemas = [];
+
+    public static function postgresConfigured(): bool
+    {
+        $dsn = getenv('MESHVAULT_TEST_PG_DSN');
+        return $dsn !== false && $dsn !== '';
+    }
+
    /** Never a hand-maintained test schema: drift must fail the suite. */
    public static function migrate(string $path): PDO
    {
+        if (isset(self::$schemas[$path])) {
+            throw new \LogicException('Clean up the previous database before migrating this path again.');
+        }
+        $environment = ['adapter' => 'sqlite'];
+        if (self::postgresConfigured()) {
+            $pdo = self::postgresConnection();
+            $schema = 'meshvault_test_' . bin2hex(random_bytes(16));
+            $pdo->exec('CREATE SCHEMA "' . $schema . '"');
+            self::$schemas[$path] = ['schema' => $schema, 'connection' => $pdo];
+            $pdo->exec('SET search_path TO "' . $schema . '"');
+            $environment = ['adapter' => 'pgsql', 'schema' => $schema];
+        } else {
+            $pdo = self::connect($path);
+        }
+        try {
+            $config = new Config([
+                'paths' => ['migrations' => dirname(__DIR__, 2) . '/db/migrations'],
+                'environments' => [
+                    'default_migration_table' => 'phinxlog',
+                    'default_database' => 'test',
+                    'test' => $environment + ['connection' => $pdo],
+                ],
+            ]);
+            (new Manager($config, new ArrayInput([]), new NullOutput()))->migrate('test');
+            if (isset(self::$schemas[$path])) {
+                // Phinx may add public to search_path; repository queries must remain isolated.
+                $pdo->exec('SET search_path TO "' . self::$schemas[$path]['schema'] . '"');
+            }
+            return $pdo;
+        } catch (\Throwable $error) {
+            self::cleanup($path);
+            throw $error;
+        }
+    }
+
+    public static function connect(string $path): PDO
+    {
+        if (isset(self::$schemas[$path])) {
+            $pdo = self::postgresConnection();
+            $pdo->exec('SET search_path TO "' . self::$schemas[$path]['schema'] . '"');
+            return $pdo;
+        }
+        if (self::postgresConfigured()) {
+            throw new \LogicException('Migrate this path before connecting to PostgreSQL.');
+        }
        $pdo = new PDO('sqlite:' . $path, null, null, [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]);
-        $config = new Config([
-            'paths' => ['migrations' => dirname(__DIR__, 2) . '/db/migrations'],
-            'environments' => [
-                'default_migration_table' => 'phinxlog',
-                'default_database' => 'test',
-                'test' => ['adapter' => 'sqlite', 'connection' => $pdo],
-            ],
-        ]);
-        (new Manager($config, new ArrayInput([]), new NullOutput()))->migrate('test');
        $pdo->exec('PRAGMA foreign_keys = ON');
        $pdo->exec('PRAGMA busy_timeout = 5000');
        return $pdo;
    }

+    public static function cleanup(string $path): void
+    {
+        if (isset(self::$schemas[$path])) {
+            ['schema' => $schema, 'connection' => $pdo] = self::$schemas[$path];
+            if ($pdo->inTransaction()) {
+                $pdo->rollBack();
+            }
+            // Only identifiers generated and registered by migrate() can reach this statement.
+            $pdo->exec('DROP SCHEMA "' . $schema . '" CASCADE');
+            unset(self::$schemas[$path]);
+        }
+        if (is_file($path)) {
+            unlink($path);
+        }
+    }
+
+    private static function postgresConnection(): PDO
+    {
+        $dsn = (string) getenv('MESHVAULT_TEST_PG_DSN');
+        if (!str_starts_with($dsn, 'pgsql:')) {
+            throw new \InvalidArgumentException('MESHVAULT_TEST_PG_DSN must be a pgsql: DSN.');
+        }
+        return new PDO(
+            $dsn,
+            ($user = getenv('MESHVAULT_TEST_PG_USER')) === false ? null : $user,
+            ($password = getenv('MESHVAULT_TEST_PG_PASSWORD')) === false ? null : $password,
+            [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION],
+        );
+    }
+
    public static function directory(string $prefix): string
    {
        $path = sys_get_temp_dir() . DIRECTORY_SEPARATOR . 'meshvault-' . $prefix . '-' . bin2hex(random_bytes(6));
@@ -27,7 +27,15 @@ final class RepositoryTest extends TestCase
    protected function tearDown(): void
    {
        unset($this->pdo);
-        unlink($this->database);
+        RepositoryDatabase::cleanup($this->database);
+    }
+
+    public function testSecondConnectionUsesTheMigratedDatabase(): void
+    {
+        $jobs = new PdoImportJobRepository($this->pdo);
+        $job = $jobs->create('/shared', []);
+        $other = new PdoImportJobRepository(RepositoryDatabase::connect($this->database));
+        self::assertSame('/shared', $other->get($job)['source_path']);
    }

    public function testCatalogGroupingAndFileLifecycle(): void
@@ -126,7 +134,7 @@ final class RepositoryTest extends TestCase
        $catalog = new PdoCatalogRepository($this->pdo);
        $job = $jobs->create('/', [['source_path' => '/a', 'size' => 5, 'model_id' => null]]);
        $second = $jobs->create('/', []);
-        $other = new PdoImportJobRepository(new PDO('sqlite:' . $this->database));
+        $other = new PdoImportJobRepository(RepositoryDatabase::connect($this->database));
        self::assertSame($job, (int) $jobs->claimNext()['id']);
        self::assertSame($second, (int) $other->claimNext()['id']);
        self::assertNull($other->claimNext());