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']); } }