Compare commits
3
Commits
00daa6b903
...
6a2b9079e9
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6a2b9079e9 | ||
|
|
211c4370d0 | ||
|
|
99f19e1052 |
No files matched your search
@@ -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.
|
||||
@@ -0,0 +1 @@
|
||||
docs: update project-status with integration status (2026-10-14)
|
||||
@@ -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),
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
use Phinx\Migration\AbstractMigration;
|
||||
|
||||
final class ImportGroupingDecisions extends AbstractMigration
|
||||
{
|
||||
public function up(): void
|
||||
{
|
||||
$this->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();
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
@@ -1,9 +1,40 @@
|
||||
## 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
|
||||
source, tests, and recorded runtime evidence. Implementation is not milestone
|
||||
acceptance. The earlier assessment understated remaining M2/M3 work.
|
||||
|
||||
## Explicit grouping sibling implementation — 2026-10-18
|
||||
|
||||
`MeshVaultSharedContracts` now contains confirmed opt-in grouping, stable identity reuse,
|
||||
immutable/idempotent decisions, legacy-default wiring, durable collision plans before
|
||||
filesystem publication, exact-path no-overwrite publication, recovery and conservative
|
||||
compensation. Not merged or committed; main implementation and issue acceptance unchanged.
|
||||
|
||||
Remote disposable PostgreSQL RED then GREEN evidence is recorded in sibling
|
||||
`.superpowers/shared-contracts/explicit-grouping-report.md`: final **483 tests,
|
||||
2776 assertions, 11 skips**, foundation and lint exit 0. Linux SQLite additionally passed
|
||||
483 tests / 2407 assertions / 44 skips. Fresh review, mutation controls and dedicated
|
||||
concurrent decision-writer tests remain outstanding. Simulated crash-state coverage is
|
||||
not process-kill/power-loss proof; directory fsync is absent. Uncertain compensation retains
|
||||
files. Multipart new grouping must select a single created model ID (`create_new` is per-file).
|
||||
No live DB changes, push, merge, issue closure or M2/M3 completion is claimed.
|
||||
|
||||
|
||||
## Catalog-core Task 4 evidence — 2026-10-13
|
||||
|
||||
Tasks 1–3 were approved at handoff (371 tests / 2032 assertions / 22 skips,
|
||||
|
||||
@@ -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
|
||||
<?php
|
||||
declare(strict_types=1);
|
||||
|
||||
use Phinx\Migration\AbstractMigration;
|
||||
|
||||
final class ImportGroupingDecisions extends AbstractMigration
|
||||
{
|
||||
public function up(): void
|
||||
{
|
||||
$this->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
|
||||
<?php
|
||||
declare(strict_types=1);
|
||||
|
||||
use Phinx\Migration\AbstractMigration;
|
||||
|
||||
final class ModelIdentityStability extends AbstractMigration
|
||||
{
|
||||
public function up(): void
|
||||
{
|
||||
// No schema changes — this migration documents the stability contract
|
||||
// Models are now immutable in identity (folder_id + name unique)
|
||||
// Files reference models via nullable FK (from Task 4)
|
||||
// Explicit grouping decisions recorded in import_jobs_decisions
|
||||
}
|
||||
|
||||
public function down(): void
|
||||
{
|
||||
// No-op: identity stability is a contract, not a schema change
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 5. Import Service Integration
|
||||
|
||||
```php
|
||||
// src/Indexer/ImportService.php (modified)
|
||||
namespace MeshVault\Indexer;
|
||||
|
||||
use MeshVault\Indexer\Contracts\ExplicitImportGroupingAdapterInterface;
|
||||
use MeshVault\Indexer\Contracts\FileEnumeratorInterface;
|
||||
use MeshVault\Repository\ImportJobRepositoryInterface;
|
||||
use Throwable;
|
||||
|
||||
final readonly class ImportService
|
||||
{
|
||||
public function __construct(
|
||||
private FileEnumeratorInterface $enumerator,
|
||||
private ImportJobRepositoryInterface $jobs,
|
||||
private ExplicitImportGroupingAdapterInterface $groupingAdapter, // Changed from LegacyGuessed
|
||||
private SpaceChecker $space,
|
||||
private string $libraryRoot,
|
||||
) {}
|
||||
|
||||
/** @param array<string, mixed> $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<array<string, mixed>> */
|
||||
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.
|
||||
@@ -0,0 +1,49 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace MeshVault\Indexer\Contracts;
|
||||
|
||||
use MeshVault\Indexer\ScannedFile;
|
||||
|
||||
/**
|
||||
* Explicit, user-confirmed import grouping with decision persistence for replay/idempotency.
|
||||
* Contrasts with LegacyGuessedImportGroupingAdapter which performs fresh stateless lookups.
|
||||
*/
|
||||
interface ExplicitImportGroupingAdapterInterface
|
||||
{
|
||||
/**
|
||||
* Resolve or create a model for the given file under explicit user confirmation.
|
||||
*
|
||||
* @param string $sourcePath Absolute source root
|
||||
* @param ScannedFile $file File with relativePath, size, hash
|
||||
* @param array<string, mixed> $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<array{source_path: string, model_id: int, context: array}> $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<string, int> Map of source_path => model_id
|
||||
*/
|
||||
public function recoverDecisions(int $jobId): array;
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace MeshVault\Indexer;
|
||||
|
||||
use DomainException;
|
||||
use InvalidArgumentException;
|
||||
use MeshVault\Indexer\Contracts\ExplicitImportGroupingAdapterInterface;
|
||||
use MeshVault\Repository\CatalogRepositoryInterface;
|
||||
use MeshVault\Repository\ImportJobRepositoryInterface;
|
||||
|
||||
/** Opt-in grouping: never infer identity from source filenames. */
|
||||
final readonly class ExplicitImportGroupingAdapter implements ExplicitImportGroupingAdapterInterface
|
||||
{
|
||||
public function __construct(
|
||||
private CatalogRepositoryInterface $catalog,
|
||||
private ImportJobRepositoryInterface $jobs,
|
||||
) {
|
||||
}
|
||||
|
||||
public function modelFor(string $sourcePath, ScannedFile $file, array $context = []): int
|
||||
{
|
||||
if (($context['confirmed'] ?? null) !== true) {
|
||||
throw new InvalidArgumentException('Explicit grouping requires confirmed=true.');
|
||||
}
|
||||
if (isset($context['create_new']) && !is_bool($context['create_new'])) {
|
||||
throw new InvalidArgumentException('create_new must be boolean.');
|
||||
}
|
||||
foreach (['folder_path', 'model_name'] as $key) {
|
||||
if (array_key_exists($key, $context) && (!is_string($context[$key]) || trim($context[$key]) === '')) {
|
||||
throw new InvalidArgumentException('Explicit folder and title must be nonempty strings.');
|
||||
}
|
||||
}
|
||||
if (array_key_exists('model_id', $context)) {
|
||||
if (!is_int($context['model_id']) || $context['model_id'] <= 0 || ($context['create_new'] ?? false)) {
|
||||
throw new InvalidArgumentException('Select a positive model ID or create a model, not both.');
|
||||
}
|
||||
$aggregate = $this->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);
|
||||
}
|
||||
}
|
||||
@@ -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<string> $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<array<string, mixed>> */
|
||||
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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<string, int> 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. */
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace MeshVault\Tests\Integration;
|
||||
|
||||
use DomainException;
|
||||
use InvalidArgumentException;
|
||||
use MeshVault\Indexer\ExplicitImportGroupingAdapter;
|
||||
use MeshVault\Indexer\ScannedFile;
|
||||
use MeshVault\Repository\PdoCatalogRepository;
|
||||
use MeshVault\Repository\PdoImportJobRepository;
|
||||
use PHPUnit\Framework\TestCase;
|
||||
|
||||
final class ExplicitGroupingTest extends TestCase
|
||||
{
|
||||
private string $database;
|
||||
private \PDO $pdo;
|
||||
private PdoCatalogRepository $catalog;
|
||||
private PdoImportJobRepository $jobs;
|
||||
|
||||
protected function setUp(): void
|
||||
{
|
||||
$this->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));
|
||||
}
|
||||
}
|
||||
@@ -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');
|
||||
|
||||
Reference in new issue
Block a user