diff --git a/commit-msg.txt b/commit-msg.txt new file mode 100644 index 0000000..45a6343 --- /dev/null +++ b/commit-msg.txt @@ -0,0 +1,29 @@ +feat: catalog shared contracts (Tasks 1-4) + +Task 1: Establish catalog identity and atomic file ownership +- CatalogService, CatalogServiceInterface, ModelMetadata DTO +- PdoCatalogRepository with atomic operations, savepoint support +- FOR UPDATE file locks in ascending order, FOR NO KEY UPDATE for metadata +- FK-compatible lock composition documented (spec §6) + +Task 2: M3 import & indexing engine delivery layer +- ImportService, ImportWorker with durable collision handling +- SpaceChecker, StreamingHasher, VerifiedCopier, StlValidator +- FileEnumerator, LocalFilesystem, PathSafety +- Console command wiring via ContainerWiringTest + +Task 3: Importer integration and repair +- LegacyGuessedImportGroupingAdapter isolates first-directory guessing +- ImportService uses adapter; ImportWorker uses CatalogServiceInterface +- Null owners remain null; DomainException mapped to CATALOG_CONFLICT +- SQL failure after registration/path update with caller sentinel rollback + +Task 4: PostgreSQL contention evidence and bounded acceptance +- 6 new PostgresCatalogContractsTest cases with independent sessions +- Generated schemas, 250ms lock timeouts, 5s statement timeouts +- Ascending file locks, FOR SHARE contention, stale-owner rejection +- Rollback/retry, nested transaction survival, root creation contention +- Negative controls: replay FOR UPDATE removal (11 failures), metadata FOR UPDATE (4 FK KEY SHARE timeouts) +- Fresh Docker PostgreSQL 17 / PHP 8.4.26: 473 tests, 2718 assertions, 10 skips, exit 0 + +All local: 474 tests, 2347 assertions, 54 skips, foundation/lint clean. \ No newline at end of file diff --git a/commit-msg2.txt b/commit-msg2.txt new file mode 100644 index 0000000..ce1ae9d --- /dev/null +++ b/commit-msg2.txt @@ -0,0 +1 @@ +docs: update project-status with integration status (2026-10-14) \ No newline at end of file diff --git a/config/container.php b/config/container.php index 67fd981..11779be 100644 --- a/config/container.php +++ b/config/container.php @@ -58,6 +58,14 @@ return static fn (AppConfig $config): array => [ CatalogServiceInterface::class => DI\get(CatalogService::class), LegacyGuessedImportGroupingAdapter::class => static fn (ContainerInterface $c): LegacyGuessedImportGroupingAdapter => new LegacyGuessedImportGroupingAdapter($c->get(CatalogRepositoryInterface::class)), + // Explicit grouping is opt-in; the default ImportService below remains legacy-compatible. + \MeshVault\Indexer\ExplicitImportGroupingAdapter::class => static fn (ContainerInterface $c) + => new \MeshVault\Indexer\ExplicitImportGroupingAdapter( + $c->get(CatalogRepositoryInterface::class), + $c->get(ImportJobRepositoryInterface::class), + ), + \MeshVault\Indexer\Contracts\ExplicitImportGroupingAdapterInterface::class + => DI\get(\MeshVault\Indexer\ExplicitImportGroupingAdapter::class), ImportService::class => static fn (ContainerInterface $c): ImportService => new ImportService( $c->get(FileEnumeratorInterface::class), $c->get(ImportJobRepositoryInterface::class), diff --git a/db/migrations/20261014120001_import_grouping_decisions.php b/db/migrations/20261014120001_import_grouping_decisions.php new file mode 100644 index 0000000..66e98f0 --- /dev/null +++ b/db/migrations/20261014120001_import_grouping_decisions.php @@ -0,0 +1,26 @@ +table('import_jobs_decisions', ['id' => false, 'primary_key' => ['job_id', 'source_path']]) + ->addColumn('job_id', 'integer', ['signed' => false]) + ->addColumn('source_path', 'text') + ->addColumn('model_id', 'integer', ['signed' => false]) + ->addColumn('context', 'jsonb', ['default' => '{}']) + ->addColumn('created_at', 'timestamp', ['default' => 'CURRENT_TIMESTAMP']) + ->addForeignKey('job_id', 'import_jobs', 'id', ['delete' => 'CASCADE', 'update' => 'NO_ACTION']) + ->addIndex(['model_id'], ['name' => 'idx_import_decisions_model']) + ->create(); + } + + public function down(): void + { + $this->table('import_jobs_decisions')->drop()->save(); + } +} diff --git a/docs/explicit-grouping.md b/docs/explicit-grouping.md new file mode 100644 index 0000000..b15f456 --- /dev/null +++ b/docs/explicit-grouping.md @@ -0,0 +1,61 @@ +# Explicit import grouping + +Explicit grouping is opt-in. The container still injects `LegacyGuessedImportGroupingAdapter` +into the default `ImportService`. To opt in, construct/override `ImportService` with the +container's `ExplicitImportGroupingAdapterInterface` binding instead. All repositories must +share the same PDO connection. + +Call `import($source, $extensions, $contexts)`. Context keys are scan-relative paths; +`'*'` supplies a fallback. Every successfully scanned file must resolve to a context with +literal boolean `confirmed => true` and either: + +- a positive integer `model_id` (optional folder/title assertions must match), or +- nonempty `folder_path` and `model_name` strings. + +No source-path guessing occurs in the explicit adapter. Missing confirmation/context is +rejected. `create_new => true` creates a distinct model; otherwise existing catalog +`ensureModel()` supplies stable identity and rejects ambiguous folder/title matches. +For a multipart *new* model, create it once and use its ID for all parts; `create_new` +is a per-file request, not a named group shared across files. + +The service persists model assignments, job/items, and confirmed contexts in one transaction. +The existing catalog APIs are sufficient: no additional catalog interface methods or +identity migration are needed. Existing root-lock and nested-savepoint behavior is reused. + +`recoverDecisions($jobId)` returns a durable `source_path => model_id` map, including +from a new repository/connection. `persistDecisions()` is all-or-nothing and identical +retries are idempotent. Conflicting model IDs or contexts raise `DomainException` rather +than overwrite an existing decision. Job deletion cascades to decisions. The supplied +migration intentionally remains unchanged; decision model IDs are historical IDs without +a model foreign key. Recovery does not recreate deleted models or guess replacement IDs. + +Worker publication/replay is a separate implementation concern; this adapter does not copy +files, mutate source files, or automatically re-plan a previously persisted job. + +## Verification + +Integration tests in `ExplicitGroupingTest` exercise strict confirmation, missing contexts, +stable/recreated/ambiguous identities, ID assertions, per-file/wildcard service contexts, +durable recovery, immutable retry decisions, and caller-owned rollback. Linux PostgreSQL +runs use `sh bin/test-postgres.sh` in an isolated `/tmp` directory and disposable Docker +resources. The final combined run includes the actual foundation/deployment metadata +(the earlier grouping-only run used placeholders and is superseded). + +## Publication and recovery + +The worker chooses a collision-free destination before committing the immutable item +plan. It then publishes to that exact name using a no-overwrite hard link. A competing +writer occupying the planned name causes failure, never an unrecorded suffix. Restart +verifies size and SHA-256 at the recorded destination and completes catalog registration +without requiring the source to remain available. + +Filesystem changes cannot roll back with SQL. If completion fails, the worker removes +only a file published by that invocation, after verifying its identity/content and that +it is not catalogued. Adopted files are never removed. Uncertain ownership or cleanup +failure conservatively retains the file and logs the cleanup error. Caller-owned +transactions are rejected before claiming a job: publication requires committed intent. + +This is process-crash recovery, not power-loss durability (no directory fsync). Library +mutation must remain controlled; PHP path checks cannot prevent adversarial path swaps. +Crash recovery tests construct persisted crash states rather than kill a worker process. +See `.superpowers/shared-contracts/explicit-grouping-report.md` for RED/GREEN evidence. diff --git a/docs/project-status.md b/docs/project-status.md index b152d5b..2b4ad2f 100644 --- a/docs/project-status.md +++ b/docs/project-status.md @@ -1,3 +1,17 @@ +## Explicit grouping extension — implementation worktree status + +Implemented confirmed opt-in grouping, stable existing catalog identity, immutable persisted +decisions, and durable collision planning before no-overwrite publication with recovery and +conservative compensation. Default legacy wiring is retained. Final isolated Linux PostgreSQL +run: 483 tests / 2776 assertions / 11 skips; foundation and lint passed (exit 0). +Linux SQLite: 483 tests / 2407 assertions / 44 skips. Remote RED preceded implementation. +Fresh review/integration pending; no commits, pushes, merges or issue closures in this work. +Mutation controls and dedicated concurrent decision tests remain outstanding. See +`.superpowers/shared-contracts/explicit-grouping-report.md` and `docs/explicit-grouping.md` +for exact evidence and filesystem durability limitations. Main-checkout status update is a +parent handoff because this implementer is restricted to MeshVaultSharedContracts writes. + + # Acceptance reconciliation — 2026-10-12 Baseline: `c1ef2ee`. This review compares the live Gitea acceptance criteria with diff --git a/docs/superpowers/specs/2026-10-14-catalog-explicit-grouping-contract.md b/docs/superpowers/specs/2026-10-14-catalog-explicit-grouping-contract.md new file mode 100644 index 0000000..a082959 --- /dev/null +++ b/docs/superpowers/specs/2026-10-14-catalog-explicit-grouping-contract.md @@ -0,0 +1,488 @@ +# Bounded #28 Catalog Contract — Explicit Grouping & Stable Identity + +**Status:** Design phase — not implemented. Extends Tasks 1-4 contracts without breaking them. +**Baseline:** `main` at `9dd8c07` (Tasks 1-4 integrated, #28 catalog core closed). + +## 1. Problem Statement + +The current `LegacyGuessedImportGroupingAdapter` implements **guessed** grouping: +- First directory component under source root → model name +- Files in source root → model named after source directory +- Stateless, fresh lookup on every call via `ensureModel()` +- **No user confirmation**, no persistence of grouping decisions + +Gaps from reconciliation table: +- **#18**: Confirmed explicit grouping (user confirmation) missing +- **#20**: Explicit user-confirmed grouping (not guessed); stable model identity across imports; grouping persistence +- **#24**: Crash between publish and suffix save → duplicate suffixed copies; orphaned published files on catalog failure; replay idempotency + +## 2. Design Principles + +1. **Retain PHP/PDO/Phinx and existing repositories** — no schema rewrites, new migrations only +2. **Extend, don't replace** — `CatalogServiceInterface`/`CatalogRepositoryInterface` unchanged +3. **Explicit grouping is opt-in** — legacy guessed adapter remains for backward compatibility +4. **Bounded scope** — no UI, no scheduler, no async workers; pure service/contract layer +5. **Lock order preserved** — root → folder → model → file (same as Task 4) +6. **Savepoint semantics** — caller transaction owns commit/rollback; catalog operations use savepoints + +## 3. New Contracts (Interfaces) + +### 3.1 Explicit Grouping Adapter Interface + +```php +// src/Indexer/Contracts/ExplicitImportGroupingAdapterInterface.php +namespace MeshVault\Indexer\Contracts; + +use MeshVault\Indexer\ScannedFile; + +interface ExplicitImportGroupingAdapterInterface +{ + public function modelFor(string $sourcePath, ScannedFile $file, array $context = []): int; + public function persistDecisions(int $jobId, array $decisions): void; + public function recoverDecisions(int $jobId): array; +} +``` + +### 3.2 Explicit Grouping Adapter Implementation + +```php +// src/Indexer/ExplicitImportGroupingAdapter.php +namespace MeshVault\Indexer; + +use MeshVault\Indexer\Contracts\ExplicitImportGroupingAdapterInterface; +use MeshVault\Indexer\ScannedFile; +use MeshVault\Repository\CatalogRepositoryInterface; +use MeshVault\Repository\ImportJobRepositoryInterface; +use DomainException; +use InvalidArgumentException; + +final class ExplicitImportGroupingAdapter implements ExplicitImportGroupingAdapterInterface +{ + public function __construct( + private readonly CatalogRepositoryInterface $catalog, + private readonly ImportJobRepositoryInterface $jobs, + ) {} + + public function modelFor(string $sourcePath, ScannedFile $file, array $context = []): int + { + if (isset($context['model_id'])) { + $modelId = (int) $context['model_id']; + $aggregate = $this->catalog->aggregate($modelId); + if ($aggregate === null) { + throw new DomainException('Explicit model not found.'); + } + if (isset($context['model_name']) && $aggregate['model']['name'] !== $context['model_name']) { + throw new DomainException('Model title mismatch.'); + } + if (isset($context['folder_path'])) { + $expectedFolder = '/' . ltrim($context['folder_path'], '/'); + if ($aggregate['folder']['path'] !== $expectedFolder) { + throw new DomainException('Model folder mismatch.'); + } + } + return $modelId; + } + + if (isset($context['folder_path']) && isset($context['model_name'])) { + $folderPath = '/' . ltrim($context['folder_path'], '/'); + $modelName = $context['model_name']; + if ($context['create_new'] ?? false) { + return $this->catalog->atomic(fn () => $this->catalog->createModel($folderPath, $modelName)); + } + return $this->catalog->atomic(fn () => $this->catalog->ensureModel($folderPath, $modelName)); + } + + $parent = trim(dirname(str_replace('\\', '/', $file->relativePath)), '/'); + $group = $parent === '' || $parent === '.' + ? basename(rtrim(str_replace('\\', '/', $sourcePath), '/')) + : explode('/', $parent)[0]; + return $this->catalog->atomic(fn () => $this->catalog->ensureModel('/' . $group, $group)); + } + + public function persistDecisions(int $jobId, array $decisions): void + { + $this->jobs->atomic(function () use ($jobId, $decisions) { + foreach ($decisions as $decision) { + $this->jobs->recordGroupingDecision( + $jobId, + $decision['source_path'], + $decision['model_id'], + $decision['context'] ?? [] + ); + } + }); + } + + public function recoverDecisions(int $jobId): array + { + return $this->jobs->getGroupingDecisions($jobId); + } +} +``` + +### 3.3 Import Job Repository Extensions + +```php +// src/Repository/ImportJobRepositoryInterface.php (additions) +namespace MeshVault\Repository; + +interface ImportJobRepositoryInterface +{ + // ... existing methods ... + public function recordGroupingDecision(int $jobId, string $sourcePath, int $modelId, array $context): void; + public function getGroupingDecisions(int $jobId): array; +} +``` + +### 3.4 Catalog Service Extensions (for stable identity) + +```php +// src/Catalog/CatalogServiceInterface.php (additions) +namespace MeshVault\Catalog; + +interface CatalogServiceInterface +{ + // ... existing methods ... + public function findModel(string $folderPath, string $modelName): ?array; + public function listModelsInFolder(string $folderPath): array; + public function getOrCreateModel(string $folderPath, string $modelName): array; +} +``` + +## 4. Database Migrations + +### 4.1 Migration: `20261014120001_import_grouping_decisions.php` + +```php +table('import_jobs_decisions', ['id' => false, 'primary_key' => ['job_id', 'source_path']]) + ->addColumn('job_id', 'integer', ['signed' => false]) + ->addColumn('source_path', 'text') + ->addColumn('model_id', 'integer', ['signed' => false]) + ->addColumn('context', 'jsonb', ['default' => '{}']) + ->addColumn('created_at', 'timestamp', ['default' => 'CURRENT_TIMESTAMP']) + ->addForeignKey('job_id', 'import_jobs', 'id', ['delete' => 'CASCADE', 'update' => 'NO_ACTION']) + ->addIndex(['model_id'], ['name' => 'idx_import_decisions_model']) + ->create(); + } + + public function down(): void + { + $this->table('import_jobs_decisions')->drop()->save(); + } +} +``` + +### 4.2 Migration: `20261014120002_model_identity_stability.php` + +```php + $groupingContext Optional explicit grouping context per file */ + public function import(string $sourcePath, array $extensions = [], array $groupingContext = []): int + { + $scan = $this->enumerator->enumerate($sourcePath, $extensions); + if ($scan->files === []) { + $detail = $scan->errors === [] ? 'No importable files were found.' : $scan->errors[0]->message; + throw new ImportException('SCAN_EMPTY', $detail); + } + $this->space->check($this->libraryRoot, $scan->totalBytes); + + try { + return $this->jobs->atomic( + fn (): int => $this->createJobWithGrouping($sourcePath, $scan, $groupingContext) + ); + } catch (ImportException $error) { + throw $error; + } catch (Throwable $error) { + throw new ImportException('DB_ERROR', 'Could not persist the import job.', $error); + } + } + + /** @return list> */ + private function createJobWithGrouping(string $sourcePath, ScanResult $scan, array $groupingContext): int + { + $items = []; + $decisions = []; + + foreach ($scan->files as $file) { + $fileContext = $groupingContext[$file->relativePath] ?? $groupingContext['*'] ?? []; + $modelId = $this->groupingAdapter->modelFor($sourcePath, $file, $fileContext); + + $items[] = [ + 'source_path' => $file->sourcePath, + 'size' => $file->size, + 'model_id' => $modelId, + ]; + + $decisions[] = [ + 'source_path' => $file->sourcePath, + 'model_id' => $modelId, + 'context' => $fileContext, + ]; + } + + foreach ($scan->errors as $error) { + $items[] = [ + 'source_path' => $error->sourcePath, + 'size' => 0, + 'model_id' => null, + 'status' => 'Failed', + 'failure_reason' => $error->code . ': ' . $error->message, + ]; + } + + $jobId = $this->jobs->create($sourcePath, $items); + + if ($decisions !== []) { + $this->groupingAdapter->persistDecisions($jobId, $decisions); + } + + return $jobId; + } +} +``` + +## 6. Worker Recovery & Idempotency (Addresses #24) + +### 6.1 Current Gap +- Worker persists planned name **before** copy +- Saves collision suffix **after** publication +- Crash between → duplicate suffixed copies on replay +- Catalog failure after publish → orphaned file on disk + +### 6.2 Solution: Atomic Publish-Then-Register + +```php +// src/Indexer/ImportWorker.php (modified publish method) +private function publish(array $plan, string $destination): string +{ + $directory = dirname($destination); + if (!is_dir($directory) && !@mkdir($directory, 0775, true) && !is_dir($directory)) { + throw new ImportException('COPY_FAILED', 'Cannot create the managed destination directory.'); + } + + $destination = $this->pathSafety->ensureDestSafe($destination); + clearstatcache(true, $destination); + + if (is_file($destination)) { + if (hash_file('sha256', $destination) === $plan['hash']) { + return $destination; + } + } else { + $this->spaceChecker->check($directory, $plan['size']); + } + + // CRITICAL CHANGE: Copy to TEMP, then atomic rename + catalog register in ONE transaction + $tempPath = $destination . '.tmp.' . bin2hex(random_bytes(8)); + try { + $this->copier->copy($plan['source'], $tempPath, $plan['size'], $plan['hash']); + + $managedPath = $this->relative($tempPath); + $this->catalog->atomic(function () use ($tempPath, $destination, $plan, &$managedPath) { + if (!@rename($tempPath, $destination)) { + throw new ImportException('COPY_FAILED', 'Cannot finalize the published file.'); + } + + $managedPath = $this->relative($destination); + + $fileId = $this->catalog->registerFile( + $plan['model_id'], + $managedPath, + $plan['hash'], + $plan['size'], + $plan['mime'] ?? null, + 'Present' + ); + + $this->jobs->updateItemManagedPath($plan['item_id'], $managedPath, $fileId); + }); + + return $destination; + } catch (Throwable $error) { + if (is_file($tempPath)) { + @unlink($tempPath); + } + throw $error; + } +} +``` + +### 6.3 New Job Repository Method + +```php +// src/Repository/ImportJobRepositoryInterface.php (addition) +public function updateItemManagedPath(int $itemId, string $managedPath, int $fileId): void; +``` + +## 7. Lock Order & Transaction Boundaries + +| Operation | Locks Acquired (Order) | Savepoint | +|-----------|------------------------|-----------| +| `createModel` | Root → Folder → Model | Yes | +| `ensureModel` | Root → Folder → Model (SELECT FOR SHARE) | Yes | +| `registerFile` | Root → Folder → File | Yes | +| `changeFileOwner` | Root → File (model FK key-share) | Yes | +| `updateModelMetadata` | Model (FOR NO KEY UPDATE) | Yes | +| `changeFileStatus` | File (FOR UPDATE) | Yes | +| **Explicit grouping decide** | Root → Folder → Model (same as ensureModel) | Yes | +| **Worker publish+register** | Root → Folder → File (same as registerFile) | Yes | + +**Invariant:** Never hold job/item locks while entering catalog mutations. The worker's file lock (`.meshvault-import.lock`) is **outside** database transactions. + +## 8. Test Plan (RED → GREEN) + +### 8.1 New Test Cases + +| Test | Scenario | Expected | +|------|----------|----------| +| `testExplicitGroupingWithModelId` | Provide existing model_id in context | Reuses model, validates folder/title match | +| `testExplicitGroupingNewModel` | Provide folder_path + model_name + create_new | Creates new model at exact path | +| `testExplicitGroupingFindOrCreate` | Provide folder_path + model_name (no create_new) | Finds existing or creates | +| `testExplicitGroupingAmbiguityRejected` | Two models match folder/title (legacy data) | DomainException | +| `testGroupingDecisionsPersisted` | Import with explicit context | Decisions recoverable after job creation | +| `testGroupingDecisionsRecovered` | Resume interrupted job | Reuses recorded model_ids | +| `testWorkerPublishAtomic` | Crash simulation between copy/register | No orphan temp, no duplicate suffix | +| `testWorkerReplayIdempotent` | Re-run worker on completed item | Adopts existing file, no re-copy | +| `testWorkerCatalogFailureRollback` | Catalog error after file copy | Temp cleaned, no orphan on disk | +| `testStableModelIdentity` | Multiple imports of same source tree | Same model IDs reused (not recreated) | + +### 8.2 Mutation Tests (Negative Controls) + +| Mutation | Expected Failure | +|----------|------------------| +| Remove FOR SHARE in `ensureModel` | Ambiguous title not detected | +| Remove temp-file atomic rename | Orphan temp files on crash | +| Skip decision persistence | Recovered job uses guessed grouping | + +## 9. Backward Compatibility + +- `LegacyGuessedImportGroupingAdapter` **unchanged** — remains default in container +- New `ExplicitImportGroupingAdapter` is opt-in via container configuration +- No schema changes to existing tables (new table only) +- Existing `ImportService::import()` signature extended with optional `$groupingContext` + +## 10. Container Wiring (config/container.php) + +```php +// Legacy (default) +$container->set(ImportService::class, fn ($c) => new ImportService( + $c->get(FileEnumeratorInterface::class), + $c->get(ImportJobRepositoryInterface::class), + $c->get(LegacyGuessedImportGroupingAdapter::class), // Existing + $c->get(SpaceChecker::class), + $c->get(string::class, 'library_root'), +)); + +// Explicit (new, opt-in) +$container->set(ExplicitImportGroupingAdapter::class, fn ($c) => new ExplicitImportGroupingAdapter( + $c->get(CatalogRepositoryInterface::class), + $c->get(ImportJobRepositoryInterface::class), +)); + +$container->set(ImportService::class, fn ($c) => new ImportService( + $c->get(FileEnumeratorInterface::class), + $c->get(ImportJobRepositoryInterface::class), + $c->get(ExplicitImportGroupingAdapter::class), // New + $c->get(SpaceChecker::class), + $c->get(string::class, 'library_root'), +)); +``` + +## 11. Acceptance Criteria (Per #28 Bullets) + +| Bullet | Evidence | +|--------|----------| +| **1. Transactional multipart aggregate** | Existing `CatalogService::atomic` + `aggregate()` — Tasks 1-4 | +| **2. Shared service contracts consumed by import/metadata/status** | `ImportService` uses `CatalogServiceInterface`; `ImportWorker` uses same; metadata/status via `CatalogService` — Tasks 1-4 | +| **3. Explicit reassignment/rollback/FK evidence** | `reassignFiles`, `detachFiles`, `changeFileOwner` with FK KEY SHARE — Task 4 | +| **NEW: Explicit confirmed grouping** | `ExplicitImportGroupingAdapter` with persisted decisions — This design | +| **NEW: Stable model identity** | `findModel`, `getOrCreateModel`, decision recovery — This design | +| **NEW: Crash-safe publish** | Atomic temp-rename + catalog register in savepoint — This design | + +## 12. Out of Scope (Future Milestones) + +- UI for grouping confirmation (#14) +- Scanner incremental/resumable (#15) +- Missing-file replace/remove UI (#25) +- Progress/ETA/pause/resume contracts (#26) +- Retry/backoff/dead-letter contracts (#27) +- Full-text search/facets (#29) +- Duplicate detection/review queue (#30) + +## 13. Implementation Sequence + +1. **Migration 1**: `import_jobs_decisions` table +2. **Interface**: `ExplicitImportGroupingAdapterInterface` +3. **Implementation**: `ExplicitImportGroupingAdapter` +4. **Repository extension**: `recordGroupingDecision`, `getGroupingDecisions`, `updateItemManagedPath` +5. **Catalog extension**: `findModel`, `listModelsInFolder`, `getOrCreateModel` +6. **ImportService integration**: Optional `$groupingContext` parameter +7. **Worker atomic publish**: Temp copy → atomic rename + register in savepoint +8. **Tests**: RED → GREEN for all cases in §8 +9. **Mutation verification**: Two negative controls +10. **Documentation**: Update ledger and project-status + +## 14. Risk Assessment + +| Risk | Mitigation | +|------|------------| +| Lock order violation | All new paths follow root→folder→model→file; code review + mutation tests | +| Savepoint leakage | All catalog ops use existing `atomic()` wrapper; no raw transactions | +| Decision table growth | One row per file per job; bounded by job size; cleaned with job | +| Temp file cleanup on crash | `finally` block in worker; temp files in library root (same FS) | +| Backward compatibility break | Legacy adapter unchanged; explicit adapter opt-in only | +| Migration failure on existing data | Preflight checks (Task 1 pattern); reject if unassigned/ambiguous | + +--- + +**Next Steps:** Implement RED tests first, then GREEN implementation per sequence in §13. No commits to `main` until fresh whole-branch review. \ No newline at end of file diff --git a/src/Indexer/Contracts/ExplicitImportGroupingAdapterInterface.php b/src/Indexer/Contracts/ExplicitImportGroupingAdapterInterface.php new file mode 100644 index 0000000..beecb5b --- /dev/null +++ b/src/Indexer/Contracts/ExplicitImportGroupingAdapterInterface.php @@ -0,0 +1,49 @@ + $context Optional confirmation context: + * - 'model_id' => int: user-selected existing model (must match folder/title) + * - 'model_name' => string: user-provided model title + * - 'folder_path' => string: user-provided folder path (must exist or be creatable) + * - 'create_new' => bool: force new model even if match exists + * @return int Model ID (existing or newly created) + * @throws \DomainException If context conflicts with existing catalog state + * @throws \InvalidArgumentException If context is malformed + */ + public function modelFor(string $sourcePath, ScannedFile $file, array $context = []): int; + + /** + * Persist the user's grouping decision for replay/idempotency. + * Called after successful import job creation with confirmed context. + * + * @param int $jobId Import job ID + * @param list $decisions + * @return void + */ + public function persistDecisions(int $jobId, array $decisions): void; + + /** + * Recover grouping decisions for a resumed/interrupted job. + * Returns model_id per source_path, or empty array if no decision was recorded. + * + * @param int $jobId Import job ID + * @return array Map of source_path => model_id + */ + public function recoverDecisions(int $jobId): array; +} diff --git a/src/Indexer/ExplicitImportGroupingAdapter.php b/src/Indexer/ExplicitImportGroupingAdapter.php new file mode 100644 index 0000000..415ff32 --- /dev/null +++ b/src/Indexer/ExplicitImportGroupingAdapter.php @@ -0,0 +1,98 @@ +catalog->aggregate($context['model_id']); + if ($aggregate === null) { + throw new DomainException('Explicit model not found.'); + } + if (isset($context['model_name']) && $aggregate['model']['name'] !== $context['model_name']) { + throw new DomainException('Model title mismatch.'); + } + if ( + isset($context['folder_path']) + && $aggregate['folder']['path'] !== $this->folder($context['folder_path']) + ) { + throw new DomainException('Model folder mismatch.'); + } + return $context['model_id']; + } + if (!isset($context['folder_path'], $context['model_name'])) { + throw new InvalidArgumentException('Select a model ID or supply both folder_path and model_name.'); + } + $folder = $this->folder($context['folder_path']); + return ($context['create_new'] ?? false) + ? $this->catalog->createModel($folder, $context['model_name']) + : $this->catalog->ensureModel($folder, $context['model_name']); + } + + private function folder(string $path): string + { + $path = '/' . trim($path, '/'); + foreach (explode('/', $path) as $component) { + if ( + $component === '.' || $component === '..' + || str_contains($component, '\\') || str_contains($component, "\0") + ) { + throw new InvalidArgumentException('Folder must be a canonical catalog path.'); + } + } + if (str_contains($path, '//')) { + throw new InvalidArgumentException('Folder must be a canonical catalog path.'); + } + return $path; + } + + public function persistDecisions(int $jobId, array $decisions): void + { + $this->jobs->atomic(function () use ($jobId, $decisions): void { + foreach ($decisions as $decision) { + $this->jobs->recordGroupingDecision( + $jobId, + $decision['source_path'], + $decision['model_id'], + $decision['context'] ?? [] + ); + } + }); + } + + public function recoverDecisions(int $jobId): array + { + return $this->jobs->getGroupingDecisions($jobId); + } +} diff --git a/src/Indexer/ImportService.php b/src/Indexer/ImportService.php index 5393286..08971cb 100644 --- a/src/Indexer/ImportService.php +++ b/src/Indexer/ImportService.php @@ -23,7 +23,7 @@ final readonly class ImportService public function __construct( private FileEnumeratorInterface $enumerator, private ImportJobRepositoryInterface $jobs, - private LegacyGuessedImportGroupingAdapter $groupingAdapter, + private LegacyGuessedImportGroupingAdapter|Contracts\ExplicitImportGroupingAdapterInterface $groupingAdapter, private SpaceChecker $space, private string $libraryRoot, ) { @@ -35,8 +35,11 @@ final readonly class ImportService * @param list $extensions Case-insensitive extensions without dots. * @throws ImportException SCAN_EMPTY, DISK_FULL, SPACE_CHECK_FAILED, DB_ERROR. */ - public function import(string $sourcePath, array $extensions = FileEnumeratorInterface::DEFAULT_EXTENSIONS): int - { + public function import( + string $sourcePath, + array $extensions = FileEnumeratorInterface::DEFAULT_EXTENSIONS, + array $groupingContext = [], + ): int { $scan = $this->enumerator->enumerate($sourcePath, $extensions); if ($scan->files === []) { $detail = $scan->errors === [] ? 'No importable files were found.' : $scan->errors[0]->message; @@ -46,7 +49,15 @@ final readonly class ImportService $this->space->check($this->libraryRoot, $scan->totalBytes); try { return $this->jobs->atomic( - fn (): int => $this->jobs->create($sourcePath, $this->manifest($sourcePath, $scan)) + function () use ($sourcePath, $scan, $groupingContext): int { + $decisions = []; + $items = $this->manifest($sourcePath, $scan, $groupingContext, $decisions); + $jobId = $this->jobs->create($sourcePath, $items); + if ($this->groupingAdapter instanceof Contracts\ExplicitImportGroupingAdapterInterface) { + $this->groupingAdapter->persistDecisions($jobId, $decisions); + } + return $jobId; + } ); } catch (ImportException $error) { throw $error; @@ -56,14 +67,19 @@ final readonly class ImportService } /** @return list> */ - private function manifest(string $sourcePath, ScanResult $scan): array + private function manifest(string $sourcePath, ScanResult $scan, array $contexts, array &$decisions): array { $items = []; foreach ($scan->files as $file) { + $context = $contexts[$file->relativePath] ?? $contexts['*'] ?? []; + $modelId = $this->groupingAdapter instanceof Contracts\ExplicitImportGroupingAdapterInterface + ? $this->groupingAdapter->modelFor($sourcePath, $file, $context) + : $this->groupingAdapter->modelFor($sourcePath, $file); + $decisions[] = ['source_path' => $file->sourcePath, 'model_id' => $modelId, 'context' => $context]; $items[] = [ 'source_path' => $file->sourcePath, 'size' => $file->size, - 'model_id' => $this->groupingAdapter->modelFor($sourcePath, $file), + 'model_id' => $modelId, ]; } // Scan failures become terminal items so the job reports them durably. diff --git a/src/Indexer/ImportWorker.php b/src/Indexer/ImportWorker.php index d3c8542..fab6232 100644 --- a/src/Indexer/ImportWorker.php +++ b/src/Indexer/ImportWorker.php @@ -52,6 +52,7 @@ final readonly class ImportWorker */ public function runOnce(): ?int { + $this->jobs->assertPublicationAutocommit(); $lock = $this->lock(); try { // Recovery runs under the lock, so no live worker can be reset. @@ -104,13 +105,28 @@ final readonly class ImportWorker private function process(int $jobId, array $item): void { $itemId = (int) $item['id']; + $owned = null; try { + // The manifest FK can become NULL on model deletion; durable decisions cannot. + // Validate before planning AND recovery so explicit intent never becomes unassigned. + $decisions = $this->jobs->getGroupingDecisions($jobId); + $source = (string) $item['source_path']; + if (array_key_exists($source, $decisions)) { + $modelId = $decisions[$source]; + if ( + $item['model_id'] === null || (int) $item['model_id'] !== $modelId + || $this->catalog->aggregate($modelId) === null + ) { + throw new ImportException('CATALOG_CONFLICT', 'Explicit grouping model is missing or mismatched.'); + } + } $plan = $this->plan($item); $destination = $this->pathSafety->ensureDestSafe( $this->libraryRoot . DIRECTORY_SEPARATOR . str_replace('/', DIRECTORY_SEPARATOR, $plan['managed_path']) ); - $published = $this->publish($plan, $destination); + $this->jobs->assertPublicationAutocommit(); + $published = $this->publish($plan, $destination, $owned); $relative = $this->relative($published); // Catalog and progress commit together, so a crash here simply replays. $this->catalog->atomic(function () use ($item, $itemId, $plan, $relative): void { @@ -131,6 +147,9 @@ final readonly class ImportWorker $this->jobs->finishItem($itemId, 'Completed'); }); } catch (Throwable $error) { + if ($owned !== null) { + $this->compensate($destination, $owned, $plan); + } $this->failItem($jobId, $itemId, $error); } } @@ -144,9 +163,6 @@ final readonly class ImportWorker private function plan(array $item): array { $source = (string) $item['source_path']; - if (strtolower(pathinfo($source, PATHINFO_EXTENSION)) === 'stl') { - $this->validator->validate($source); - } if ($item['managed_path'] !== null && $item['hash'] !== null) { // Recovered plan: trust the committed intent, never choose a new name. return [ @@ -156,12 +172,18 @@ final readonly class ImportWorker 'size' => (int) $item['size'], ]; } + if (strtolower(pathinfo($source, PATHINFO_EXTENSION)) === 'stl') { + $this->validator->validate($source); + } $hash = $this->hasher->hash($source); $size = filesize($source); if ($size === false) { throw new ImportException('SCAN_READ_ERROR', 'Cannot determine the source size.'); } - $managed = $this->managedPath($item, $source); + $managed = $this->relative($this->copier->selectDestination( + $this->libraryRoot . DIRECTORY_SEPARATOR . $this->managedPath($item, $source) + )); + $this->jobs->assertPublicationAutocommit(); $this->jobs->planItem((int) $item['id'], $managed, $hash, $size); return ['source' => $source, 'managed_path' => $managed, 'hash' => $hash, 'size' => $size]; } @@ -183,9 +205,9 @@ final readonly class ImportWorker * Publish, or adopt an already-published destination left by a crashed run. * * @param array{source: string, managed_path: string, hash: string, size: int} $plan - * @return string The actual published path (may have collision suffix) + * @return string The exact persisted path; collisions never change the plan. */ - private function publish(array $plan, string $destination): string + private function publish(array $plan, string $destination, ?array &$owned): string { $directory = dirname($destination); if (!is_dir($directory) && !@mkdir($directory, 0775, true) && !is_dir($directory)) { @@ -196,17 +218,48 @@ final readonly class ImportWorker clearstatcache(true, $destination); if (is_file($destination)) { // Idempotent restart: the planned destination already holds our bytes. - if (hash_file('sha256', $destination) === $plan['hash']) { + if (filesize($destination) === $plan['size'] && hash_file('sha256', $destination) === $plan['hash']) { return $destination; } - // Different content at the planned destination: let VerifiedCopier resolve the collision - // by appending a _001 suffix. This keeps restart idempotent (we don't re-copy the same - // file) while still allowing two same-basename sources with different content to coexist. + throw new ImportException('COLLISION', 'Persisted destination contains different bytes.'); } else { // New file being copied: check free space before copying $this->spaceChecker->check($directory, $plan['size']); } - return $this->copier->copy($plan['source'], $destination, $plan['size'], $plan['hash']); + if ($this->catalog->fileByManagedPath($plan['managed_path']) !== null) { + throw new ImportException('CATALOG_CONFLICT', 'Persisted destination is already cataloged.'); + } + $published = $this->copier->copy($plan['source'], $destination, $plan['size'], $plan['hash'], true); + clearstatcache(true, $published); + $owned = lstat($published) ?: null; + return $published; + } + + /** Only compensate this invocation's publication, never adopted bytes or a replacement. */ + private function compensate(string $destination, array $owned, array $plan): void + { + try { + $this->pathSafety->ensureDestSafe($destination); + clearstatcache(true, $destination); + $current = @lstat($destination); + // A committed catalog row (including an ambiguous commit result) wins over cleanup. + if ($this->catalog->fileByManagedPath($plan['managed_path']) !== null) { + return; + } + if ( + $current !== false && $current['dev'] === $owned['dev'] && $current['ino'] === $owned['ino'] + && is_file($destination) && $current['size'] === $plan['size'] + && hash_file('sha256', $destination) === $plan['hash'] + ) { + if (!@unlink($destination)) { + throw new ImportException('COPY_FAILED', 'Cannot compensate publication.'); + } + } + } catch (Throwable $cleanupError) { + $this->logger->error('Publication compensation could not be verified.', [ + 'managed_path' => $plan['managed_path'], 'error' => $cleanupError->getMessage(), + ]); + } } private function relative(string $published): string diff --git a/src/Indexer/VerifiedCopier.php b/src/Indexer/VerifiedCopier.php index b559233..b22fe36 100644 --- a/src/Indexer/VerifiedCopier.php +++ b/src/Indexer/VerifiedCopier.php @@ -37,12 +37,18 @@ final readonly class VerifiedCopier /** * Copy an existing regular source to a filename under the managed library. * Parent must exist. Returns absolute published path; collisions try _001 - * through _999 before the last extension. Never deletes/overwrites a winner. + * through _999 before the last extension. With exact=true, collision fails + * without suffixing (for durable worker plans). Never deletes/overwrites a winner. * Throws PathSafetyException unchanged; ImportException reasons are * COPY_FAILED, HASH_MISMATCH (size or digest), COLLISION, INVALID_ARGUMENT. */ - public function copy(string $source, string $destination, int $expectedSize, string $expectedSha256): string - { + public function copy( + string $source, + string $destination, + int $expectedSize, + string $expectedSha256, + bool $exact = false, + ): string { if ($expectedSize < 0 || preg_match('/\A[0-9a-fA-F]{64}\z/', $expectedSha256) !== 1) { throw new ImportException('INVALID_ARGUMENT', 'Expected a nonnegative size and SHA-256 hex digest.'); } @@ -99,7 +105,7 @@ final readonly class VerifiedCopier throw new ImportException('COPY_FAILED', 'Temporary close failed.'); } $output = null; - return $this->publish($temporary, $destination); + return $this->publish($temporary, $destination, $exact); } catch (ImportException | PathSafetyException $error) { throw $error; } catch (Throwable $error) { @@ -121,6 +127,23 @@ final readonly class VerifiedCopier } + /** Select only; caller must durably persist this name and use exact publication. */ + public function selectDestination(string $destination): string + { + $extension = pathinfo($destination, PATHINFO_EXTENSION); + $stem = $extension === '' ? $destination : substr($destination, 0, -strlen($extension) - 1); + for ($suffix = 0; $suffix <= 999; ++$suffix) { + $candidate = $suffix === 0 ? $destination + : $stem . sprintf('_%03d', $suffix) . ($extension === '' ? '' : '.' . $extension); + $candidate = $this->pathSafety->ensureDestSafe($candidate); + clearstatcache(true, $candidate); + if (!file_exists($candidate)) { + return $candidate; + } + } + throw new ImportException('COLLISION', 'Destination and all 999 suffixes are occupied.'); + } + /** @param resource $stream */ private function verify($stream, int $size, string $digest): void { @@ -142,11 +165,11 @@ final readonly class VerifiedCopier } } - private function publish(string $temporary, string $destination): string + private function publish(string $temporary, string $destination, bool $exact): string { $extension = pathinfo($destination, PATHINFO_EXTENSION); $stem = $extension === '' ? $destination : substr($destination, 0, -strlen($extension) - 1); - for ($suffix = 0; $suffix <= 999; ++$suffix) { + for ($suffix = 0; $suffix <= ($exact ? 0 : 999); ++$suffix) { $candidate = $suffix === 0 ? $destination : $stem . sprintf('_%03d', $suffix) . ($extension === '' ? '' : '.' . $extension); $candidate = $this->pathSafety->ensureDestSafe($candidate); diff --git a/src/Repository/ImportJobRepositoryInterface.php b/src/Repository/ImportJobRepositoryInterface.php index 8c25cc8..387fa53 100644 --- a/src/Repository/ImportJobRepositoryInterface.php +++ b/src/Repository/ImportJobRepositoryInterface.php @@ -6,12 +6,21 @@ namespace MeshVault\Repository; interface ImportJobRepositoryInterface { + /** Confirmed decisions are immutable; identical retries are idempotent. */ + public function recordGroupingDecision(int $jobId, string $sourcePath, int $modelId, array $context): void; + + /** @return array Durable source path to model ID map. */ + public function getGroupingDecisions(int $jobId): array; + public function create(string $sourcePath, array $items): int; public function get(int $jobId): ?array; public function items(int $jobId): array; + /** Fail closed if worker intent would be nested in an uncommitted caller transaction. */ + public function assertPublicationAutocommit(): void; + public function claimNext(): ?array; /** Caller MUST hold the global library flock; never use age-based recovery. */ diff --git a/src/Repository/PdoImportJobRepository.php b/src/Repository/PdoImportJobRepository.php index 4efd2ec..0797a09 100644 --- a/src/Repository/PdoImportJobRepository.php +++ b/src/Repository/PdoImportJobRepository.php @@ -10,6 +10,46 @@ use PDO; final class PdoImportJobRepository extends PdoRepository implements ImportJobRepositoryInterface { + public function recordGroupingDecision(int $jobId, string $sourcePath, int $modelId, array $context): void + { + if ($jobId <= 0 || $modelId <= 0 || $sourcePath === '' || ($context['confirmed'] ?? null) !== true) { + throw new InvalidArgumentException('A confirmed decision requires positive IDs and a source path.'); + } + $json = json_encode($context, JSON_THROW_ON_ERROR); + $this->transaction(function () use ($jobId, $sourcePath, $modelId, $json): void { + $this->execute( + 'INSERT INTO import_jobs_decisions (job_id, source_path, model_id, context) VALUES (?, ?, ?, ?)' + . ' ON CONFLICT (job_id, source_path) DO NOTHING', + [$jobId, $sourcePath, $modelId, $json] + ); + $stored = $this->row( + 'SELECT model_id, context FROM import_jobs_decisions WHERE job_id = ? AND source_path = ?', + [$jobId, $sourcePath] + ); + if ( + (int) $stored['model_id'] !== $modelId + || json_decode($stored['context'], true, 512, JSON_THROW_ON_ERROR) + != json_decode($json, true, 512, JSON_THROW_ON_ERROR) + ) { + throw new DomainException('Grouping decisions are immutable.'); + } + }); + } + + public function getGroupingDecisions(int $jobId): array + { + $decisions = []; + foreach ( + $this->execute( + 'SELECT source_path, model_id FROM import_jobs_decisions WHERE job_id = ? ORDER BY source_path', + [$jobId] + )->fetchAll(PDO::FETCH_ASSOC) as $row + ) { + $decisions[$row['source_path']] = (int) $row['model_id']; + } + return $decisions; + } + public function create(string $sourcePath, array $items): int { if ($sourcePath === '') { @@ -70,6 +110,13 @@ final class PdoImportJobRepository extends PdoRepository implements ImportJobRep ->fetchAll(PDO::FETCH_ASSOC); } + public function assertPublicationAutocommit(): void + { + if ($this->pdo->inTransaction()) { + throw new \LogicException('Import publication cannot run inside a caller transaction.'); + } + } + public function claimNext(): ?array { return $this->transaction(function (): ?array { diff --git a/tests/Integration/ExplicitGroupingTest.php b/tests/Integration/ExplicitGroupingTest.php new file mode 100644 index 0000000..8a5e155 --- /dev/null +++ b/tests/Integration/ExplicitGroupingTest.php @@ -0,0 +1,150 @@ +database = tempnam(sys_get_temp_dir(), 'grouping-'); + $this->pdo = RepositoryDatabase::migrate($this->database); + $this->catalog = new PdoCatalogRepository($this->pdo); + $this->jobs = new PdoImportJobRepository($this->pdo); + self::assertTrue(class_exists(ExplicitImportGroupingAdapter::class), 'Explicit adapter must exist'); + } + + protected function tearDown(): void + { + RepositoryDatabase::cleanup($this->database); + } + + public function testServicePersistsPerFileAndWildcardDecisionsAtomically(): void + { + $source = RepositoryDatabase::directory('explicit-source'); + $library = RepositoryDatabase::directory('explicit-library'); + try { + file_put_contents($source . '/a.stl', 'solid a'); + file_put_contents($source . '/b.stl', 'solid b'); + $adapter = new ExplicitImportGroupingAdapter($this->catalog, $this->jobs); + $service = new \MeshVault\Indexer\ImportService( + new \MeshVault\Indexer\FileEnumerator( + new \MeshVault\Indexer\Filesystem\LocalFilesystem(), + new \MeshVault\Indexer\Filesystem\PathSafety($source) + ), + $this->jobs, + $adapter, + new \MeshVault\Indexer\SpaceChecker(), + $library + ); + $context = ['*' => ['confirmed' => true, 'folder_path' => '/parts', 'model_name' => 'Part']]; + $job = $service->import($source, ['stl'], $context); + $map = $adapter->recoverDecisions($job); + self::assertCount(2, $map); + self::assertSame($map, $adapter->recoverDecisions($service->import($source, ['stl'], $context))); + $before = (int) $this->pdo->query('SELECT count(*) FROM import_jobs')->fetchColumn(); + try { + $service->import($source, ['stl'], ['a.stl' => $context['*']]); + self::fail('Incomplete grouping accepted'); + } catch (\MeshVault\Indexer\ImportException) { + self::assertSame($before, (int) $this->pdo->query('SELECT count(*) FROM import_jobs')->fetchColumn()); + } + } finally { + RepositoryDatabase::remove($source); + RepositoryDatabase::remove($library); + } + } + + private function resolve(array $context): int + { + return (new ExplicitImportGroupingAdapter($this->catalog, $this->jobs))->modelFor( + '/source', + new ScannedFile('/source/a.stl', 'a.stl', 1, 'model/stl'), + $context + ); + } + + public function testConfirmationAndCompleteContextAreRequired(): void + { + foreach ( + [[], ['confirmed' => true], ['confirmed' => 1, 'model_id' => 1], + ['confirmed' => true, 'folder_path' => '/x'], ['confirmed' => true, 'model_id' => '1']] as $context + ) { + try { + $this->resolve($context); + self::fail('Malformed or unconfirmed context accepted'); + } catch (InvalidArgumentException) { + self::assertSame(0, (int) $this->pdo->query('SELECT count(*) FROM models')->fetchColumn()); + } + } + } + + public function testStableIdentityNewIdentityAndAmbiguity(): void + { + $context = ['confirmed' => true, 'folder_path' => '/parts', 'model_name' => 'Part']; + $id = $this->resolve($context); + self::assertSame($id, $this->resolve($context)); + self::assertSame($id, $this->resolve(['model_id' => $id] + $context)); + self::assertNotSame($id, $this->resolve(['create_new' => true] + $context)); + $this->expectException(DomainException::class); + $this->resolve($context); + } + + public function testIdentityMismatchAndMissingModelAreRejected(): void + { + $id = $this->catalog->createModel('/parts', 'Part'); + foreach ( + [['model_id' => $id, 'model_name' => 'Other'], + ['model_id' => $id, 'folder_path' => '/other'], ['model_id' => 999999]] as $context + ) { + try { + $this->resolve(['confirmed' => true] + $context); + self::fail('Conflicting identity accepted'); + } catch (DomainException) { + self::assertTrue(true); + } + } + } + + public function testDurableImmutableIdempotentDecisionsAndCallerRollback(): void + { + $id = $this->resolve(['confirmed' => true, 'folder_path' => '/parts', 'model_name' => 'Part']); + $job = $this->jobs->create('/source', [['source_path' => '/source/a.stl', 'size' => 1, 'model_id' => $id]]); + $adapter = new ExplicitImportGroupingAdapter($this->catalog, $this->jobs); + $decision = ['source_path' => '/source/a.stl', 'model_id' => $id, 'context' => ['confirmed' => true]]; + $adapter->persistDecisions($job, [$decision]); + $adapter->persistDecisions($job, [$decision]); + $fresh = new ExplicitImportGroupingAdapter( + $this->catalog, + new PdoImportJobRepository(RepositoryDatabase::connect($this->database)) + ); + self::assertSame(['/source/a.stl' => $id], $fresh->recoverDecisions($job)); + try { + $changed = array_replace($decision, ['context' => ['confirmed' => true, 'extra' => 1]]); + $adapter->persistDecisions($job, [$changed]); + self::fail('Decision changed'); + } catch (DomainException) { + self::assertSame(['/source/a.stl' => $id], $adapter->recoverDecisions($job)); + } + $this->pdo->beginTransaction(); + $other = $this->jobs->create('/other', []); + $adapter->persistDecisions($other, [$decision]); + self::assertTrue($this->pdo->inTransaction()); + $this->pdo->rollBack(); + self::assertSame([], $adapter->recoverDecisions($other)); + } +} diff --git a/tests/Integration/ImportPipelineTest.php b/tests/Integration/ImportPipelineTest.php index 2272888..57d604d 100644 --- a/tests/Integration/ImportPipelineTest.php +++ b/tests/Integration/ImportPipelineTest.php @@ -63,7 +63,7 @@ final class ImportPipelineTest extends TestCase ); } - private function worker(?callable $shouldStop = null): ImportWorker + private function worker(?callable $shouldStop = null, array $operations = []): ImportWorker { $safety = new PathSafety($this->library); $sourceSafety = new PathSafety($this->source); @@ -71,7 +71,7 @@ final class ImportPipelineTest extends TestCase $this->jobs, new \MeshVault\Catalog\CatalogService($this->catalog), new StreamingHasher($sourceSafety), - new VerifiedCopier($safety), + new VerifiedCopier($safety, $operations), new StlValidator($sourceSafety), $safety, new SpaceChecker(), @@ -121,10 +121,7 @@ final class ImportPipelineTest extends TestCase self::assertStringStartsWith('CATALOG_CONFLICT:', $item['failure_reason']); self::assertSame($other, (int) $this->catalog->file($id)['model_id']); self::assertSame($sourceBefore, PhysicalFileSnapshot::capture($this->source)); - self::assertSame( - $sourceBefore['planes/wing.stl']['bytes'], - file_get_contents($this->library . '/planes/wing.stl') - ); + self::assertFileDoesNotExist($this->library . '/planes/wing.stl'); } public function testCompletionRegistersBeforeUpdatingItemPath(): void @@ -159,10 +156,97 @@ final class ImportPipelineTest extends TestCase self::assertStringStartsWith('CATALOG_CONFLICT:', $item['failure_reason']); self::assertSame(0, (int) $this->pdo->query('SELECT COUNT(*) FROM files')->fetchColumn()); self::assertSame($sourceBefore, PhysicalFileSnapshot::capture($this->source)); - self::assertSame($sourceBefore['loose.stl']['bytes'], file_get_contents($this->library . '/unfiled/loose.stl')); + self::assertFileDoesNotExist($this->library . '/unfiled/loose.stl'); } - public function testCompletionSqlFailureRollsBackCatalogPathAndProgressInsideCallerTransaction(): void + public static function invalidExplicitOwners(): array + { + return [ + 'deleted before execution' => [true, false, false], + 'deleted before recovery' => [true, true, false], + 'different manifest owner' => [false, false, true], + 'missing historical owner with replacement' => [true, false, true], + ]; + } + + #[\PHPUnit\Framework\Attributes\DataProvider('invalidExplicitOwners')] + public function testExplicitOwnershipFailsClosed(bool $delete, bool $recover, bool $replace): void + { + $source = $this->writeStl('part.stl'); + $adapter = new \MeshVault\Indexer\ExplicitImportGroupingAdapter($this->catalog, $this->jobs); + $service = new ImportService( + new FileEnumerator(new LocalFilesystem(), new PathSafety($this->source)), + $this->jobs, + $adapter, + new SpaceChecker(), + $this->library, + ); + $model = $this->catalog->createModel('/confirmed', 'Part'); + $job = $service->import($this->source, ['stl'], [ + '*' => ['confirmed' => true, 'model_id' => $model], + ]); + $item = $this->jobs->items($job)[0]; + $destination = $this->library . '/confirmed/part.stl'; + if ($recover) { + $this->jobs->claimNext(); + $this->jobs->planItem( + (int) $item['id'], + 'confirmed/part.stl', + hash_file('sha256', $source), + filesize($source) + ); + mkdir(dirname($destination)); + copy($source, $destination); + unlink($source); // Recovery must validate ownership even without the source. + } + if ($delete) { + $this->pdo->exec('DELETE FROM models WHERE id = ' . $model); + self::assertNull($this->jobs->items($job)[0]['model_id']); + } + if ($replace) { + $other = $this->catalog->createModel('/replacement', 'Other'); + $this->pdo->exec('UPDATE import_items SET model_id = ' . $other . ' WHERE id = ' . (int) $item['id']); + } + self::assertSame([$source => $model], $adapter->recoverDecisions($job)); + $sourceBefore = PhysicalFileSnapshot::capture($this->source); + $this->worker()->runOnce(); + $item = $this->jobs->items($job)[0]; + self::assertSame('Failed', $item['status']); + self::assertStringStartsWith('CATALOG_CONFLICT:', $item['failure_reason']); + self::assertSame('Failed', $this->jobs->get($job)['status']); + self::assertSame(0, (int) $this->pdo->query('SELECT COUNT(*) FROM files')->fetchColumn()); + self::assertSame([$source => $model], $adapter->recoverDecisions($job)); + self::assertSame($sourceBefore, PhysicalFileSnapshot::capture($this->source)); + self::assertFileDoesNotExist($this->library . '/unfiled/part.stl'); + self::assertFileDoesNotExist($this->library . '/replacement/part.stl'); + if ($recover) { + self::assertSame($item['hash'], hash_file('sha256', $destination)); + self::assertSame('confirmed/part.stl', $item['managed_path']); + } else { + self::assertNull($item['managed_path']); + self::assertFileDoesNotExist($destination); + } + } + + public function testCallerTransactionIsRejectedBeforePublication(): void + { + $this->writeStl('planes/wing.stl'); + $job = $this->service()->import($this->source); + $this->pdo->beginTransaction(); + try { + $this->worker()->runOnce(); + self::fail('Uncommitted publication intent accepted'); + } catch (\LogicException $error) { + self::assertStringContainsString('transaction', $error->getMessage()); + } finally { + self::assertTrue($this->pdo->inTransaction()); + $this->pdo->rollBack(); + } + self::assertSame('Pending', $this->jobs->items($job)[0]['status']); + self::assertFileDoesNotExist($this->library . '/planes/wing.stl'); + } + + public function testCompletionSqlFailureCompensatesOwnedSuffix(): void { if ($this->pdo->getAttribute(PDO::ATTR_DRIVER_NAME) !== 'sqlite') { self::markTestSkipped('SQLite failure-injection trigger.'); @@ -176,27 +260,17 @@ final class ImportPipelineTest extends TestCase // Fail after registration and item-path update, during progress refresh. $this->pdo->exec("CREATE TRIGGER reject_completion BEFORE UPDATE ON import_jobs" . " WHEN NEW.status = 'Completed' BEGIN SELECT RAISE(ABORT, 'injected completion failure'); END"); - $this->pdo->beginTransaction(); - $sentinel = $this->catalog->createModel('/sentinel', 'sentinel'); $this->worker()->runOnce(); - self::assertTrue($this->pdo->inTransaction()); - self::assertNotNull($this->catalog->aggregate($sentinel)); self::assertSame(0, (int) $this->pdo->query('SELECT COUNT(*) FROM files')->fetchColumn()); $item = $this->jobs->items($job)[0]; - self::assertSame('planes/wing.stl', $item['managed_path']); + self::assertSame('planes/wing_001.stl', $item['managed_path']); self::assertSame('Failed', $item['status']); self::assertStringStartsWith('COPY_FAILED:', $item['failure_reason']); self::assertSame('Failed', $this->jobs->get($job)['status']); self::assertSame($sourceBefore, PhysicalFileSnapshot::capture($this->source)); $published = PhysicalFileSnapshot::capture($this->library); self::assertSame($libraryBefore['planes/wing.stl'], $published['planes/wing.stl']); - self::assertSame($sourceBefore['planes/wing.stl']['bytes'], $published['planes/wing_001.stl']['bytes']); - self::assertSame($sourceBefore['planes/wing.stl']['size'], $published['planes/wing_001.stl']['size']); - $this->pdo->rollBack(); - self::assertSame($sourceBefore, PhysicalFileSnapshot::capture($this->source)); - self::assertSame($published, PhysicalFileSnapshot::capture($this->library)); - self::assertNull($this->catalog->aggregate($sentinel)); - self::assertSame('Pending', $this->jobs->items($job)[0]['status']); + self::assertArrayNotHasKey('planes/wing_001.stl', $published); } public function testImportCreatesJobWithLegacyGuessedSubfolderGrouping(): void @@ -300,6 +374,60 @@ final class ImportPipelineTest extends TestCase self::assertCount(1, $this->catalog->aggregate((int) $item['model_id'])['files']); } + public function testSuffixIntentIsCommittedBeforeLinkAndRecoveryNeedsNoSource(): void + { + $source = $this->writeStl('planes/wing.stl'); + $job = $this->service()->import($this->source); + mkdir($this->library . '/planes'); + file_put_contents($this->library . '/planes/wing.stl', 'occupied'); + $observed = null; + $this->worker(null, ['link' => function (string $temp, string $dest) use ($job, &$observed): bool { + $reader = RepositoryDatabase::connect($this->database); + $observed = $reader->query('SELECT managed_path FROM import_items WHERE job_id = ' . $job)->fetchColumn(); + return link($temp, $dest); + }])->runOnce(); + self::assertSame('planes/wing_001.stl', $observed); + $this->pdo->exec("UPDATE import_items SET status = 'Copying', completed_at = NULL"); + $this->pdo->exec("UPDATE import_jobs SET status = 'Copying', processed_files = 0"); + $this->pdo->exec('DELETE FROM files'); + unlink($source); + $this->worker()->runOnce(); + self::assertSame('Completed', $this->jobs->items($job)[0]['status']); + self::assertFileDoesNotExist($this->library . '/planes/wing_002.stl'); + } + + public function testPublicationRaceFailsWithoutSilentSuffix(): void + { + $this->writeStl('planes/wing.stl'); + $job = $this->service()->import($this->source); + $this->worker(null, ['link' => static function (string $temp, string $dest): bool { + file_put_contents($dest, 'race winner'); + return false; + }])->runOnce(); + self::assertSame('Failed', $this->jobs->items($job)[0]['status']); + self::assertSame('race winner', file_get_contents($this->library . '/planes/wing.stl')); + self::assertFileDoesNotExist($this->library . '/planes/wing_001.stl'); + } + + public function testAdoptedBytesSurviveCompletionFailure(): void + { + if ($this->pdo->getAttribute(PDO::ATTR_DRIVER_NAME) !== 'sqlite') { + self::markTestSkipped('SQLite failure-injection trigger.'); + } + $source = $this->writeStl('planes/wing.stl'); + $job = $this->service()->import($this->source); + $item = $this->jobs->items($job)[0]; + $this->jobs->planItem((int) $item['id'], 'planes/wing.stl', hash_file('sha256', $source), filesize($source)); + mkdir($this->library . '/planes'); + copy($source, $this->library . '/planes/wing.stl'); + $this->pdo->exec("CREATE TRIGGER reject_adoption BEFORE UPDATE ON import_jobs" + . " WHEN NEW.status = 'Completed' BEGIN SELECT RAISE(ABORT, 'completion failure'); END"); + $this->worker()->runOnce(); + self::assertSame('Failed', $this->jobs->items($job)[0]['status']); + self::assertSame(file_get_contents($source), file_get_contents($this->library . '/planes/wing.stl')); + self::assertSame(0, (int) $this->pdo->query('SELECT COUNT(*) FROM files')->fetchColumn()); + } + public function testGracefulStopLeavesJobResumable(): void { $this->writeStl('planes/wing.stl');