feat: M3 import & indexing engine delivery layer with durable collision handling, space checking and console
This commit is contained in:
Cline committed
1 parent
93d991703c
commit
9d277d9e72
48 files changed
+5147
-4
No files matched your search
@@ -0,0 +1,311 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace MeshVault\Repository;
|
||||
|
||||
use DomainException;
|
||||
use InvalidArgumentException;
|
||||
use PDO;
|
||||
|
||||
final class PdoImportJobRepository extends PdoRepository implements ImportJobRepositoryInterface
|
||||
{
|
||||
public function create(string $sourcePath, array $items): int
|
||||
{
|
||||
if ($sourcePath === '') {
|
||||
throw new InvalidArgumentException('Source path must not be empty.');
|
||||
}
|
||||
foreach ($items as $index => $item) {
|
||||
if (!isset($item['source_path']) || !isset($item['size'])) {
|
||||
throw new InvalidArgumentException("Item $index requires source_path and size.");
|
||||
}
|
||||
if ($item['size'] < 0) {
|
||||
throw new InvalidArgumentException("Item $index size must be non-negative.");
|
||||
}
|
||||
$allowed = ['Pending', 'Copying', 'Completed', 'Failed', 'Skipped'];
|
||||
if (isset($item['status']) && !in_array($item['status'], $allowed, true)) {
|
||||
throw new InvalidArgumentException("Item $index has invalid status.");
|
||||
}
|
||||
}
|
||||
return $this->transaction(function () use ($sourcePath, $items): int {
|
||||
$job = $this->row(
|
||||
'INSERT INTO import_jobs (source_path, total_files, total_bytes) VALUES (?, ?, ?) RETURNING id',
|
||||
[$sourcePath, count($items), array_sum(array_column($items, 'size'))]
|
||||
);
|
||||
$jobId = (int) $job['id'];
|
||||
if ($items === []) {
|
||||
// Empty manifests stay claimable; completion is an explicit caller decision.
|
||||
return $jobId;
|
||||
}
|
||||
$statement = $this->pdo->prepare(
|
||||
'INSERT INTO import_items (job_id, source_path, size, model_id, status, failure_reason)'
|
||||
. ' VALUES (?, ?, ?, ?, ?, ?)'
|
||||
);
|
||||
foreach ($items as $item) {
|
||||
$status = $item['status'] ?? 'Pending';
|
||||
$failureReason = $item['failure_reason'] ?? null;
|
||||
$statement->execute([
|
||||
$jobId,
|
||||
$item['source_path'],
|
||||
$item['size'],
|
||||
$item['model_id'] ?? null,
|
||||
$status,
|
||||
$failureReason,
|
||||
]);
|
||||
}
|
||||
// Scan-time failures arrive terminal, so publish progress immediately.
|
||||
$this->refresh($jobId);
|
||||
return $jobId;
|
||||
});
|
||||
}
|
||||
|
||||
public function get(int $jobId): ?array
|
||||
{
|
||||
return $this->row('SELECT * FROM import_jobs WHERE id = ?', [$jobId]);
|
||||
}
|
||||
|
||||
public function items(int $jobId): array
|
||||
{
|
||||
return $this->execute('SELECT * FROM import_items WHERE job_id = ? ORDER BY id', [$jobId])
|
||||
->fetchAll(PDO::FETCH_ASSOC);
|
||||
}
|
||||
|
||||
public function claimNext(): ?array
|
||||
{
|
||||
return $this->transaction(function (): ?array {
|
||||
if ($this->pdo->getAttribute(PDO::ATTR_DRIVER_NAME) === 'pgsql') {
|
||||
// SKIP LOCKED lets concurrent workers claim distinct jobs without blocking.
|
||||
$candidate = $this->row(
|
||||
"SELECT id FROM import_jobs WHERE status = 'Pending'"
|
||||
. ' ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 1'
|
||||
);
|
||||
if ($candidate === null) {
|
||||
return null;
|
||||
}
|
||||
return $this->row(
|
||||
"UPDATE import_jobs SET status = 'Copying', started_at = CURRENT_TIMESTAMP"
|
||||
. ' WHERE id = ? RETURNING *',
|
||||
[$candidate['id']]
|
||||
);
|
||||
}
|
||||
// SQLite serializes writers, so a single conditional UPDATE is already atomic.
|
||||
return $this->row(
|
||||
"UPDATE import_jobs SET status = 'Copying', started_at = CURRENT_TIMESTAMP"
|
||||
. " WHERE id = (SELECT id FROM import_jobs WHERE status = 'Pending' ORDER BY id LIMIT 1)"
|
||||
. ' RETURNING *'
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
public function resumeInterrupted(): void
|
||||
{
|
||||
$this->transaction(function (): void {
|
||||
// Plans (managed_path/hash) are deliberately preserved for idempotent recovery.
|
||||
$this->execute(
|
||||
"UPDATE import_items SET status = 'Pending', started_at = NULL WHERE status = 'Copying'"
|
||||
. " AND job_id IN (SELECT id FROM import_jobs WHERE status IN ('Scanning', 'Copying'))"
|
||||
);
|
||||
$this->execute(
|
||||
"UPDATE import_jobs SET status = 'Pending' WHERE status IN ('Scanning', 'Copying')"
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
public function planItem(int $itemId, string $managedPath, string $hash, int $size): void
|
||||
{
|
||||
if ($managedPath === '' || $hash === '' || $size < 0) {
|
||||
throw new InvalidArgumentException('A plan requires a path, hash and non-negative size.');
|
||||
}
|
||||
$this->transaction(function () use ($itemId, $managedPath, $hash, $size): void {
|
||||
$item = $this->lockedItem($itemId);
|
||||
if (in_array($item['status'], ['Completed', 'Failed', 'Skipped'], true)) {
|
||||
throw new DomainException('Cannot plan a terminal import item.');
|
||||
}
|
||||
if ($item['managed_path'] !== null) {
|
||||
if (
|
||||
$item['managed_path'] !== $managedPath || $item['hash'] !== $hash
|
||||
|| (int) $item['size'] !== $size
|
||||
) {
|
||||
throw new DomainException('An import plan is immutable once recorded.');
|
||||
}
|
||||
return;
|
||||
}
|
||||
$this->execute(
|
||||
"UPDATE import_items SET status = 'Copying', managed_path = ?, hash = ?, size = ?,"
|
||||
. ' started_at = CURRENT_TIMESTAMP WHERE id = ?',
|
||||
[$managedPath, $hash, $size, $itemId]
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
public function updateManagedPath(int $itemId, string $managedPath): void
|
||||
{
|
||||
if ($managedPath === '') {
|
||||
throw new InvalidArgumentException('A managed path must not be empty.');
|
||||
}
|
||||
$this->transaction(function () use ($itemId, $managedPath): void {
|
||||
$item = $this->lockedItem($itemId);
|
||||
if (in_array($item['status'], ['Completed', 'Failed', 'Skipped'], true)) {
|
||||
throw new DomainException('Cannot update a terminal import item.');
|
||||
}
|
||||
$this->execute(
|
||||
'UPDATE import_items SET managed_path = ? WHERE id = ?',
|
||||
[$managedPath, $itemId]
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
public function finishItem(int $itemId, string $status, ?string $failureReason = null): void
|
||||
{
|
||||
if (!in_array($status, ['Completed', 'Failed', 'Skipped'], true)) {
|
||||
throw new InvalidArgumentException('Only terminal item statuses may finish an item.');
|
||||
}
|
||||
$this->transaction(function () use ($itemId, $status, $failureReason): void {
|
||||
$item = $this->lockedItem($itemId);
|
||||
if (in_array($item['status'], ['Completed', 'Failed', 'Skipped'], true)) {
|
||||
if ($item['status'] !== $status) {
|
||||
throw new DomainException('Import item already reached a different terminal status.');
|
||||
}
|
||||
return;
|
||||
}
|
||||
$this->execute(
|
||||
'UPDATE import_items SET status = ?, failure_reason = ?, completed_at = CURRENT_TIMESTAMP'
|
||||
. ' WHERE id = ?',
|
||||
[$status, $failureReason, $itemId]
|
||||
);
|
||||
$this->refresh((int) $item['job_id']);
|
||||
});
|
||||
}
|
||||
|
||||
public function release(int $jobId): void
|
||||
{
|
||||
$this->transaction(function () use ($jobId): void {
|
||||
$this->execute(
|
||||
"UPDATE import_items SET status = 'Pending', started_at = NULL"
|
||||
. " WHERE job_id = ? AND status = 'Copying'",
|
||||
[$jobId]
|
||||
);
|
||||
$this->execute(
|
||||
"UPDATE import_jobs SET status = 'Pending' WHERE id = ? AND status IN ('Scanning', 'Copying')",
|
||||
[$jobId]
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
public function cancel(int $jobId): void
|
||||
{
|
||||
$this->terminate($jobId, 'Cancelled', 'Skipped', null);
|
||||
}
|
||||
|
||||
public function fail(int $jobId, string $failureReason): void
|
||||
{
|
||||
if ($failureReason === '') {
|
||||
throw new InvalidArgumentException('A failure reason is required.');
|
||||
}
|
||||
$this->terminate($jobId, 'Failed', 'Failed', $failureReason);
|
||||
}
|
||||
|
||||
public function completeEmpty(int $jobId): void
|
||||
{
|
||||
$this->transaction(function () use ($jobId): void {
|
||||
$job = $this->row('SELECT * FROM import_jobs WHERE id = ?', [$jobId]);
|
||||
if ($job === null) {
|
||||
throw new DomainException('Unknown import job.');
|
||||
}
|
||||
if ($this->items($jobId) !== []) {
|
||||
throw new DomainException('Import job still has items.');
|
||||
}
|
||||
if (in_array($job['status'], ['Completed', 'Failed', 'Cancelled'], true)) {
|
||||
return;
|
||||
}
|
||||
$this->execute(
|
||||
"UPDATE import_jobs SET status = 'Completed', processed_files = 0,"
|
||||
. ' completed_at = CURRENT_TIMESTAMP WHERE id = ?',
|
||||
[$jobId]
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
public function atomic(callable $callback): mixed
|
||||
{
|
||||
return $this->transaction($callback);
|
||||
}
|
||||
|
||||
/** Terminal job transitions share one shape: finish open items, then close the job. */
|
||||
private function terminate(int $jobId, string $jobStatus, string $itemStatus, ?string $reason): void
|
||||
{
|
||||
$this->transaction(function () use ($jobId, $jobStatus, $itemStatus, $reason): void {
|
||||
$job = $this->row('SELECT * FROM import_jobs WHERE id = ?', [$jobId]);
|
||||
if ($job === null) {
|
||||
throw new DomainException('Unknown import job.');
|
||||
}
|
||||
if (in_array($job['status'], ['Completed', 'Failed', 'Cancelled'], true)) {
|
||||
return;
|
||||
}
|
||||
$this->execute(
|
||||
'UPDATE import_items SET status = ?, failure_reason = ?, completed_at = CURRENT_TIMESTAMP'
|
||||
. " WHERE job_id = ? AND status IN ('Pending', 'Copying')",
|
||||
[$itemStatus, $reason, $jobId]
|
||||
);
|
||||
$this->execute(
|
||||
'UPDATE import_jobs SET status = ?, processed_files = ?, copied_bytes = ?,'
|
||||
. ' completed_at = CURRENT_TIMESTAMP WHERE id = ?',
|
||||
[$jobStatus, $this->processedFiles($jobId), $this->copiedBytes($jobId), $jobId]
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/** Progress is recomputed from items so repeated or crashed updates cannot drift. */
|
||||
private function refresh(int $jobId): void
|
||||
{
|
||||
$job = $this->row('SELECT * FROM import_jobs WHERE id = ?', [$jobId]);
|
||||
if ($job === null || in_array($job['status'], ['Completed', 'Failed', 'Cancelled'], true)) {
|
||||
return;
|
||||
}
|
||||
$total = (int) $this->row('SELECT COUNT(*) AS total FROM import_items WHERE job_id = ?', [$jobId])['total'];
|
||||
$processed = $this->processedFiles($jobId);
|
||||
$failed = (int) $this->row(
|
||||
"SELECT COUNT(*) AS failed FROM import_items WHERE job_id = ? AND status = 'Failed'",
|
||||
[$jobId]
|
||||
)['failed'];
|
||||
if ($total === 0 || $processed < $total) {
|
||||
$this->execute(
|
||||
'UPDATE import_jobs SET processed_files = ?, copied_bytes = ? WHERE id = ?',
|
||||
[$processed, $this->copiedBytes($jobId), $jobId]
|
||||
);
|
||||
return;
|
||||
}
|
||||
$this->execute(
|
||||
'UPDATE import_jobs SET status = ?, processed_files = ?, copied_bytes = ?,'
|
||||
. ' completed_at = CURRENT_TIMESTAMP WHERE id = ?',
|
||||
[$failed > 0 ? 'Failed' : 'Completed', $processed, $this->copiedBytes($jobId), $jobId]
|
||||
);
|
||||
}
|
||||
|
||||
private function processedFiles(int $jobId): int
|
||||
{
|
||||
return (int) $this->row(
|
||||
'SELECT COUNT(*) AS processed FROM import_items WHERE job_id = ?'
|
||||
. " AND status IN ('Completed', 'Failed', 'Skipped')",
|
||||
[$jobId]
|
||||
)['processed'];
|
||||
}
|
||||
|
||||
private function copiedBytes(int $jobId): int
|
||||
{
|
||||
return (int) $this->row(
|
||||
"SELECT COALESCE(SUM(size), 0) AS copied FROM import_items WHERE job_id = ? AND status = 'Completed'",
|
||||
[$jobId]
|
||||
)['copied'];
|
||||
}
|
||||
|
||||
private function lockedItem(int $itemId): array
|
||||
{
|
||||
$suffix = $this->pdo->getAttribute(PDO::ATTR_DRIVER_NAME) === 'pgsql' ? ' FOR UPDATE' : '';
|
||||
$item = $this->row('SELECT * FROM import_items WHERE id = ?' . $suffix, [$itemId]);
|
||||
if ($item === null) {
|
||||
throw new DomainException('Unknown import item.');
|
||||
}
|
||||
return $item;
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user