From d384f3d60d7d346296c458e50615f056dac782f5 Mon Sep 17 00:00:00 2001 From: DariusIII Date: Tue, 4 Aug 2026 01:00:24 +0200 Subject: [PATCH] Update additional postprocessing --- app/Console/Commands/PostProcessGuid.php | 25 ++++- .../AdditionalCandidateQuery.php | 103 ++++++++++++++++-- .../AdditionalProcessingOrchestrator.php | 54 +++++++-- app/Services/Runners/PostProcessRunner.php | 74 +++++++++++-- app/Services/TempWorkspaceService.php | 34 +++++- app/Services/Tmux/TmuxMonitorService.php | 5 +- resources/views/admin/site/edit.blade.php | 7 +- .../admin/site/tmux-edit-content.blade.php | 2 +- .../Feature/AdditionalCandidateQueryTest.php | 20 ++++ ...itionalProcessingOrchestratorClaimTest.php | 16 ++- ...PostProcessRunnerAdditionalThreadsTest.php | 88 ++++++++++++++- tests/Unit/TempWorkspaceServiceTest.php | 30 +++++ 12 files changed, 403 insertions(+), 55 deletions(-) diff --git a/app/Console/Commands/PostProcessGuid.php b/app/Console/Commands/PostProcessGuid.php index a5fb9ac6a..9bb3ebf39 100644 --- a/app/Console/Commands/PostProcessGuid.php +++ b/app/Console/Commands/PostProcessGuid.php @@ -22,7 +22,9 @@ class PostProcessGuid extends Command protected $signature = 'postprocess:guid {type : Type: additional, nfo, movie, tv, anime, books, music, console, or games} {guid : First character of release guid (a-f, 0-9)} - {renamed? : For movie/tv: process renamed only (optional)}'; + {renamed? : For movie/tv: process renamed only (optional)} + {--worker : Drain multiple additional-processing batches} + {--max-batches=4 : Maximum claim batches for worker mode}'; /** * The console command description. @@ -55,7 +57,7 @@ class PostProcessGuid extends Command try { match ($type) { - 'additional' => $this->processAdditional($guid), + 'additional' => $this->processAdditional($guid, (bool) $this->option('worker')), 'nfo' => $this->processNfo($guid), 'movie' => $this->postProcessService->processMovies('', $guid, $renamed), 'tv' => $this->postProcessService->processTv('', $guid, $renamed), @@ -81,10 +83,25 @@ class PostProcessGuid extends Command /** * Process additional data for releases. */ - private function processAdditional(string $guid): void + private function processAdditional(string $guid, bool $workerMode): void { + $maxBatches = $workerMode ? max(1, (int) $this->option('max-batches')) : 1; + $workerToken = bin2hex(random_bytes(16)); + $childTimeout = (int) config('nntmux.multiprocessing_max_child_time', 1800); + $deadline = microtime(true) + max(1, $childTimeout - 30); + $excludedReleaseIds = []; + try { - $this->additionalProcessor->start('', $guid); + for ($batch = 0; $batch < $maxBatches && microtime(true) < $deadline; $batch++) { + $stats = $this->additionalProcessor->start('', $guid, $workerToken, $excludedReleaseIds); + if ($stats['claimed'] === 0) { + break; + } + $excludedReleaseIds = array_values(array_unique([ + ...$excludedReleaseIds, + ...$stats['claimed_ids'], + ])); + } } finally { $this->additionalProcessor->finish(); } diff --git a/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php b/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php index 14de575dd..fe6a4c234 100644 --- a/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php +++ b/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php @@ -161,19 +161,87 @@ final class AdditionalCandidateQuery $effectiveLimit = $limit !== null && $limit > 0 ? min($limit, self::BUCKET_LIMIT) : self::BUCKET_LIMIT; - $rows = self::baseBuilder() - ->select(DB::raw('DISTINCT r.leftguid AS guid_bucket')) - ->limit($effectiveLimit) - ->get(); - $chars = []; - foreach ($rows as $row) { - $id = strtolower((string) ($row->guid_bucket ?? '')); - if ($id !== '') { - $chars[] = substr($id, 0, 1); + + return array_slice( + array_column(self::availableBucketCounts(), 'bucket'), + 0, + $effectiveLimit, + ); + } + + /** + * Return available candidate counts keyed by GUID bucket. + * + * @return list + */ + public static function availableBucketCounts(): array + { + $counts = []; + + foreach (self::bucketBacklog() as $backlog) { + if ($backlog['available'] > 0) { + $counts[] = [ + 'bucket' => $backlog['bucket'], + 'count' => $backlog['available'], + ]; } } - return array_values(array_unique($chars)); + return $counts; + } + + /** + * Return total and currently claimable candidates for every GUID bucket. + * + * @return list + */ + public static function bucketBacklog(): array + { + $query = self::baseBuilder(includeClaimed: true) + ->select('r.leftguid') + ->selectRaw('COUNT(*) AS total_count') + ->groupBy('r.leftguid') + ->orderBy('r.leftguid'); + + if (self::supportsClaims()) { + $query->selectRaw( + 'SUM(CASE WHEN r.'.self::CLAIMED_AT_COLUMN.' IS NULL OR r.'.self::CLAIMED_AT_COLUMN.' < ? THEN 1 ELSE 0 END) AS available_count', + [self::claimStaleBefore()], + ); + } else { + $query->selectRaw('COUNT(*) AS available_count'); + } + + $backlog = []; + foreach ($query->get() as $row) { + $bucket = strtolower(substr((string) ($row->leftguid ?? ''), 0, 1)); + if ($bucket === '') { + continue; + } + + $backlog[] = [ + 'bucket' => $bucket, + 'total' => (int) ($row->total_count ?? 0), + 'available' => (int) ($row->available_count ?? 0), + ]; + } + + return $backlog; + } + + /** + * @return array{total: int, available: int} + */ + public static function backlogCounts(): array + { + $counts = ['total' => 0, 'available' => 0]; + + foreach (self::bucketBacklog() as $backlog) { + $counts['total'] += $backlog['total']; + $counts['available'] += $backlog['available']; + } + + return $counts; } /** @@ -189,6 +257,7 @@ final class AdditionalCandidateQuery * Claim a bounded batch of release rows for one worker. * * @param list $columns + * @param list $excludedReleaseIds * @return EloquentCollection */ public static function claimBatch( @@ -199,10 +268,11 @@ final class AdditionalCandidateQuery ?int $minSizeMB = null, ?int $maxSizeGB = null, array $columns = ['*'], + array $excludedReleaseIds = [], ): EloquentCollection { $effectiveLimit = max(1, $limit); - return DB::transaction(function () use ($guidChar, $effectiveLimit, $token, $groupID, $minSizeMB, $maxSizeGB, $columns): EloquentCollection { + return DB::transaction(function () use ($guidChar, $effectiveLimit, $token, $groupID, $minSizeMB, $maxSizeGB, $columns, $excludedReleaseIds): EloquentCollection { $supportsClaims = self::supportsClaims(); $query = self::baseBuilder($groupID, $guidChar, $minSizeMB, $maxSizeGB) ->select('r.id') @@ -210,6 +280,10 @@ final class AdditionalCandidateQuery ->orderBy('r.id') ->limit($effectiveLimit); + if ($excludedReleaseIds !== []) { + $query->whereNotIn('r.id', $excludedReleaseIds); + } + if (DB::getDriverName() !== 'sqlite') { $query->lockForUpdate(); } @@ -295,7 +369,7 @@ final class AdditionalCandidateQuery return; } - $staleBefore = now()->subSeconds(self::claimTtlSeconds()); + $staleBefore = self::claimStaleBefore(); $query->where(function (Builder $claimQuery) use ($staleBefore): void { $claimQuery @@ -311,6 +385,11 @@ final class AdditionalCandidateQuery return max(300, $timeout * 2); } + private static function claimStaleBefore(): \Illuminate\Support\Carbon + { + return now()->subSeconds(self::claimTtlSeconds()); + } + /** * @param list $columns * @return list diff --git a/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php b/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php index bbd53b5a6..c73cc8d52 100644 --- a/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php +++ b/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php @@ -41,21 +41,31 @@ class AdditionalProcessingOrchestrator /** * Start the additional processing. * + * @param list $excludedReleaseIds + * @return array{claimed: int, processed: int, failed: int, claimed_ids: list} + * * @throws Exception */ - public function start(string $groupID = '', string $guidChar = ''): void - { + public function start( + string $groupID = '', + string $guidChar = '', + string $workerToken = '', + array $excludedReleaseIds = [] + ): array { $this->finish(); - if (! $this->setupTempPath($guidChar, $groupID)) { - return; + if (! $this->setupTempPath($guidChar, $groupID, $workerToken)) { + return ['claimed' => 0, 'processed' => 0, 'failed' => 0, 'claimed_ids' => []]; } - $this->fetchReleases($groupID, $guidChar); + $this->fetchReleases($groupID, $guidChar, $excludedReleaseIds); if ($this->totalReleases > 0) { $this->output->echoDescription($this->totalReleases); - $this->processReleases($guidChar); + + return $this->processReleases($guidChar); } + + return ['claimed' => 0, 'processed' => 0, 'failed' => 0, 'claimed_ids' => []]; } /** @@ -95,13 +105,14 @@ class AdditionalProcessingOrchestrator /** * Set up the main temp path. */ - private function setupTempPath(string $guidChar, string $groupID): bool + private function setupTempPath(string $guidChar, string $groupID, string $workerToken = ''): bool { try { $this->mainTmpPath = $this->tempWorkspace->ensureMainTempPath( $this->config->tmpUnrarPath, $guidChar, - $groupID + $groupID, + $workerToken !== '' ? $workerToken : bin2hex(random_bytes(16)), ); $this->tempWorkspace->clearDirectory($this->mainTmpPath, true); } catch (\Throwable $e) { @@ -129,7 +140,10 @@ class AdditionalProcessingOrchestrator * PostProcessRunner::processAdditional(). Do NOT inline new predicates * here; add them to AdditionalCandidateQuery instead. */ - private function fetchReleases(int|string $groupID, string $guidChar): void + /** + * @param list $excludedReleaseIds + */ + private function fetchReleases(int|string $groupID, string $guidChar, array $excludedReleaseIds = []): void { $this->claimToken = bin2hex(random_bytes(16)); $this->releases = AdditionalCandidateQuery::claimBatch( @@ -154,6 +168,7 @@ class AdditionalProcessingOrchestrator 'pp_timeout_count', AdditionalCandidateQuery::CLAIM_TOKEN_COLUMN, ], + $excludedReleaseIds, ); $this->totalReleases = $this->releases->count(); } @@ -168,10 +183,18 @@ class AdditionalProcessingOrchestrator * every subsequent cycle, and the "needs additional pp" backlog would * grow indefinitely without any visible failure. * + * @return array{claimed: int, processed: int, failed: int, claimed_ids: list} + * * @throws Exception */ - private function processReleases(string $guidChar = ''): void + private function processReleases(string $guidChar = ''): array { + $startedAt = microtime(true); + $claimedIds = $this->releases + ->pluck('id') + ->map(static fn (mixed $id): int => (int) $id) + ->values() + ->all(); $processed = 0; $failed = 0; @@ -202,15 +225,24 @@ class AdditionalProcessingOrchestrator 'picked' => $this->totalReleases, 'processed' => $processed, 'failed' => $failed, + 'elapsed_seconds' => round(microtime(true) - $startedAt, 3), + 'peak_memory_bytes' => memory_get_peak_usage(true), ]); $this->output->endOutput(); + + return [ + 'claimed' => $this->totalReleases, + 'processed' => $processed, + 'failed' => $failed, + 'claimed_ids' => $claimedIds, + ]; } public function finish(): void { if ($this->mainTmpPath !== '') { - $this->tempWorkspace->clearDirectory($this->mainTmpPath, true); + $this->tempWorkspace->clearDirectory($this->mainTmpPath, false); $this->mainTmpPath = ''; } diff --git a/app/Services/Runners/PostProcessRunner.php b/app/Services/Runners/PostProcessRunner.php index f565be321..7f11e8d01 100644 --- a/app/Services/Runners/PostProcessRunner.php +++ b/app/Services/Runners/PostProcessRunner.php @@ -16,6 +16,8 @@ use Illuminate\Support\Facades\Log; class PostProcessRunner extends BaseRunner { + private const int ADDITIONAL_WORKER_MAX_BATCHES = 4; + private function guidBucketExpression(string $column = 'leftguid'): string { return DB::getDriverName() === 'sqlite' @@ -43,7 +45,11 @@ class PostProcessRunner extends BaseRunner // id may already be a single GUID bucket char; if not, take first char defensively $char = isset($release->id) ? substr((string) $release->id, 0, 1) : ''; // Use postprocess:guid command which accepts the GUID character - $commands[] = PHP_BINARY.' artisan postprocess:guid '.$type.' '.$char; + $command = PHP_BINARY.' artisan postprocess:guid '.$type.' '.$char; + if ($type === 'additional') { + $command .= ' --worker --max-batches='.self::ADDITIONAL_WORKER_MAX_BATCHES; + } + $commands[] = $command; } $this->runStreamingCommands($commands, $maxProcesses, $desc); // @phpstan-ignore argument.type @@ -221,16 +227,18 @@ class PostProcessRunner extends BaseRunner // Bucket-selection predicates and size filters are owned by // AdditionalCandidateQuery so they cannot drift away from the // per-worker fetch in AdditionalProcessingOrchestrator::fetchReleases(). - $chars = AdditionalCandidateQuery::bucketChars(); + $bucketCounts = AdditionalCandidateQuery::availableBucketCounts(); + $chars = array_column($bucketCounts, 'bucket'); $maxProcesses = (int) Settings::settingValue('postthreads'); + $queryLimit = (int) (Settings::settingValue('maxaddprocessed') ?: 25); // Normalize to the shape the rest of runPostProcess() expects: // an array of objects with an `id` (first GUID char) property. If the // backlog is concentrated in fewer buckets than configured threads, // repeat buckets so claimBatch() can split one hot bucket across // multiple workers. - $queue = $this->additionalQueue($chars, $maxProcesses); + $queue = $this->additionalQueue($bucketCounts, $maxProcesses, $queryLimit); if (! $this->prepareAdditionalTempBuckets($chars)) { return; } @@ -245,8 +253,19 @@ class PostProcessRunner extends BaseRunner $this->headerStart('postprocess: additional postprocessing', count($queue), 1); $orchestrator = app(AdditionalProcessingOrchestrator::class); foreach ($chars as $char) { + $workerToken = bin2hex(random_bytes(16)); + $excludedReleaseIds = []; try { - $orchestrator->start('', $char); + for ($batch = 0; $batch < self::ADDITIONAL_WORKER_MAX_BATCHES; $batch++) { + $stats = $orchestrator->start('', $char, $workerToken, $excludedReleaseIds); + if ($stats['claimed'] === 0) { + break; + } + $excludedReleaseIds = array_values(array_unique([ + ...$excludedReleaseIds, + ...$stats['claimed_ids'], + ])); + } } finally { $orchestrator->finish(); } @@ -260,20 +279,46 @@ class PostProcessRunner extends BaseRunner } /** - * @param list $chars + * @param list $bucketCounts * @return list */ - private function additionalQueue(array $chars, int $maxProcesses): array + private function additionalQueue(array $bucketCounts, int $maxProcesses, int $queryLimit): array { - if ($chars === []) { + if ($bucketCounts === []) { return []; } + $chars = array_column($bucketCounts, 'bucket'); $queue = array_map(static fn (string $c): object => (object) ['id' => $c], $chars); - $target = max(count($queue), $maxProcesses); + $allocations = array_fill(0, count($bucketCounts), 1); + $demand = []; + foreach ($bucketCounts as $index => $bucketCount) { + // Demand is intentionally measured in immediately claimable batches: + // postthreads remains the hard connection cap while hot buckets use + // available parallelism instead of draining all batches serially. + $demand[$index] = max(1, (int) ceil($bucketCount['count'] / max(1, $queryLimit))); + } - for ($index = 0; count($queue) < $target; $index++) { - $queue[] = (object) ['id' => $chars[$index % count($chars)]]; + $target = max(count($queue), min(max(1, $maxProcesses), array_sum($demand))); + + while (count($queue) < $target) { + $nextIndex = null; + $largestRemainingDemand = 0; + + foreach ($chars as $index => $char) { + $remainingDemand = $demand[$index] - $allocations[$index]; + if ($remainingDemand > $largestRemainingDemand) { + $largestRemainingDemand = $remainingDemand; + $nextIndex = $index; + } + } + + if ($nextIndex === null) { + break; + } + + $queue[] = (object) ['id' => $chars[$nextIndex]]; + $allocations[$nextIndex]++; } return $queue; @@ -293,7 +338,14 @@ class PostProcessRunner extends BaseRunner try { foreach ($chars as $char) { - $tempWorkspace->ensureMainTempPath($basePath, $char); + $bucketPath = $tempWorkspace->ensureMainTempPath($basePath, $char); + $tempWorkspace->pruneStaleWorkerDirectories( + $bucketPath, + max( + 3600, + (int) config('nntmux.multiprocessing_max_child_time', 1800) * 2, + ), + ); } } catch (\Throwable $e) { cli()->warning('Additional post-processing skipped: '.$e->getMessage()); diff --git a/app/Services/TempWorkspaceService.php b/app/Services/TempWorkspaceService.php index 0f67accd9..9b2558e35 100644 --- a/app/Services/TempWorkspaceService.php +++ b/app/Services/TempWorkspaceService.php @@ -15,8 +15,12 @@ class TempWorkspaceService * Ensure the main temp path exists and is namespaced by groupID or guidChar. * Returns the resolved main temp path. */ - public function ensureMainTempPath(string $basePath, string $guidChar = '', string $groupID = ''): string - { + public function ensureMainTempPath( + string $basePath, + string $guidChar = '', + string $groupID = '', + string $workerToken = '' + ): string { // Normalize separator at end if (! Str::endsWith($basePath, ['/', '\\'])) { $basePath .= '/'; @@ -28,6 +32,10 @@ class TempWorkspaceService $basePath .= $guidChar.'/'; } + if ($workerToken !== '') { + $basePath .= hash('sha256', $workerToken).'/'; + } + $this->ensureWritableDirectory($basePath, 'Additional post-processing temp path'); return $basePath; @@ -72,6 +80,28 @@ class TempWorkspaceService } } + public function pruneStaleWorkerDirectories(string $bucketPath, int $staleAfterSeconds): int + { + if ($bucketPath === '' || ! File::isDirectory($bucketPath)) { + return 0; + } + + $removed = 0; + $staleBefore = time() - max(1, $staleAfterSeconds); + + foreach (File::directories($bucketPath) as $directory) { + if (File::lastModified($directory) >= $staleBefore) { + continue; + } + + if (File::deleteDirectory($directory)) { + $removed++; + } + } + + return $removed; + } + /** * List files under a directory. * If $pattern is provided, return an array of preg_match($pattern, $relativePath, $matches) arrays, diff --git a/app/Services/Tmux/TmuxMonitorService.php b/app/Services/Tmux/TmuxMonitorService.php index c227a7244..bea3cd6d5 100644 --- a/app/Services/Tmux/TmuxMonitorService.php +++ b/app/Services/Tmux/TmuxMonitorService.php @@ -249,8 +249,9 @@ class TmuxMonitorService // rows claimed by active workers. `work_available` is the scheduler // gate, excluding fresh claims so tmux does not respawn duplicate // additional workers while a claimed batch is still running. - $this->runVar['counts']['now']['work'] = AdditionalCandidateQuery::baseBuilder(includeClaimed: true)->count(); - $this->runVar['counts']['now']['work_available'] = AdditionalCandidateQuery::baseBuilder()->count(); + $additionalBacklog = AdditionalCandidateQuery::backlogCounts(); + $this->runVar['counts']['now']['work'] = $additionalBacklog['total']; + $this->runVar['counts']['now']['work_available'] = $additionalBacklog['available']; $this->runVar['timers']['query']['proc2_time'] = time() - $timer2; diff --git a/resources/views/admin/site/edit.blade.php b/resources/views/admin/site/edit.blade.php index 36dd8caeb..7c7cbcc3e 100644 --- a/resources/views/admin/site/edit.blade.php +++ b/resources/views/admin/site/edit.blade.php @@ -771,7 +771,7 @@ class="flex-1 px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md focus:ring-blue-500 focus:border-blue-500 bg-white dark:bg-gray-700 text-gray-900 dark:text-gray-100"> seconds -

Maximum wall-clock seconds to spend processing a single release before skipping it. Set to 0 to disable. Default: 120.

+

Maximum wall-clock seconds to spend processing a single release before skipping it. Keep this below the multiprocessing child timeout. Set to 0 to disable. Default: 120.

@@ -789,7 +789,7 @@ -

The maximum amount of releases to process for passwords/previews/mediainfo per run. Every release gets processed here. This uses NNTP an connection, 1 per thread. This does not query external metadata providers.

+

Releases claimed per worker batch for passwords, previews, and media info. Larger batches reduce query overhead but increase each worker's memory and runtime. Each thread can hold one NNTP connection.

@@ -1046,7 +1046,7 @@ -

The number of threads for additional postprocessing. This includes deep rar inspection, preview and sample creation and nfo processing.

+

Maximum simultaneous additional-processing workers and NNTP sessions. Workers reuse their process and connection across bounded batches; raise this only within CPU, memory, disk I/O, and provider connection limits.

@@ -1100,4 +1100,3 @@
@endsection - diff --git a/resources/views/admin/site/tmux-edit-content.blade.php b/resources/views/admin/site/tmux-edit-content.blade.php index db6db4a0a..569cd1248 100644 --- a/resources/views/admin/site/tmux-edit-content.blade.php +++ b/resources/views/admin/site/tmux-edit-content.blade.php @@ -225,7 +225,7 @@ - +
seconds diff --git a/tests/Feature/AdditionalCandidateQueryTest.php b/tests/Feature/AdditionalCandidateQueryTest.php index f2bf5001f..5fa7c12fd 100644 --- a/tests/Feature/AdditionalCandidateQueryTest.php +++ b/tests/Feature/AdditionalCandidateQueryTest.php @@ -132,6 +132,26 @@ class AdditionalCandidateQueryTest extends TestCase $this->assertSame(2, AdditionalCandidateQuery::baseBuilder(includeClaimed: true)->count()); } + public function test_backlog_counts_are_aggregated_by_bucket_in_one_shape(): void + { + DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]); + DB::table('releases')->insert([ + $this->releaseRow(1, 'a'), + $this->releaseRow(2, 'a', claimedAt: now()), + $this->releaseRow(3, 'b', claimedAt: now()->subSeconds(301)), + ]); + + $this->assertSame([ + ['bucket' => 'a', 'total' => 2, 'available' => 1], + ['bucket' => 'b', 'total' => 1, 'available' => 1], + ], AdditionalCandidateQuery::bucketBacklog()); + $this->assertSame(['total' => 3, 'available' => 2], AdditionalCandidateQuery::backlogCounts()); + $this->assertSame([ + ['bucket' => 'a', 'count' => 1], + ['bucket' => 'b', 'count' => 1], + ], AdditionalCandidateQuery::availableBucketCounts()); + } + public function test_claim_batch_excludes_active_claims_and_recovers_stale_claims(): void { DB::table('categories')->insert([ diff --git a/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php b/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php index 037c17e83..f924b64a8 100644 --- a/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php +++ b/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php @@ -260,8 +260,12 @@ class RecordingTempWorkspaceService extends TempWorkspaceService public int $clearDirectoryCalls = 0; - public function ensureMainTempPath(string $basePath, string $guidChar = '', string $groupID = ''): string - { + public function ensureMainTempPath( + string $basePath, + string $guidChar = '', + string $groupID = '', + string $workerToken = '' + ): string { $this->ensureMainTempPathCalls++; return '/tmp/additional/'; @@ -275,8 +279,12 @@ class RecordingTempWorkspaceService extends TempWorkspaceService class FailingTempWorkspaceService extends RecordingTempWorkspaceService { - public function ensureMainTempPath(string $basePath, string $guidChar = '', string $groupID = ''): string - { + public function ensureMainTempPath( + string $basePath, + string $guidChar = '', + string $groupID = '', + string $workerToken = '' + ): string { $this->ensureMainTempPathCalls++; throw new \RuntimeException('Additional post-processing temp path "/root/nope/a/" is not writable'); diff --git a/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php b/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php index 0bf3c7bbe..520e2a160 100644 --- a/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php +++ b/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php @@ -4,11 +4,15 @@ declare(strict_types=1); namespace Tests\Feature; +use App\Services\AdditionalProcessing\AdditionalProcessingOrchestrator; use App\Services\Runners\PostProcessRunner; use Illuminate\Contracts\Console\Kernel; use Illuminate\Database\Schema\Blueprint; +use Illuminate\Support\Facades\Artisan; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\Schema; +use Mockery; +use Mockery\MockInterface; use PDO; use Tests\TestCase; @@ -112,14 +116,14 @@ class PostProcessRunnerAdditionalThreadsTest extends TestCase $this->assertSame(5, $runner->capturedMaxProcesses); $this->assertCount(5, $runner->capturedCommands); - $this->assertContains(PHP_BINARY.' artisan postprocess:guid additional 0', $runner->capturedCommands); - $this->assertContains(PHP_BINARY.' artisan postprocess:guid additional 4', $runner->capturedCommands); + $this->assertContains(PHP_BINARY.' artisan postprocess:guid additional 0 --worker --max-batches=4', $runner->capturedCommands); + $this->assertContains(PHP_BINARY.' artisan postprocess:guid additional 4 --worker --max-batches=4', $runner->capturedCommands); } public function test_process_additional_repeats_hot_bucket_to_fill_configured_threads(): void { DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]); - foreach (range(1, 5) as $id) { + foreach (range(1, 125) as $id) { DB::table('releases')->insert($this->releaseRow($id, 'a')); } @@ -144,11 +148,87 @@ class PostProcessRunnerAdditionalThreadsTest extends TestCase $this->assertSame(5, $runner->capturedMaxProcesses); $this->assertCount(5, $runner->capturedCommands); $this->assertSame( - array_fill(0, 5, PHP_BINARY.' artisan postprocess:guid additional a'), + array_fill(0, 5, PHP_BINARY.' artisan postprocess:guid additional a --worker --max-batches=4'), array_values($runner->capturedCommands) ); } + public function test_process_additional_does_not_start_idle_workers_for_a_small_hot_bucket(): void + { + DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]); + foreach (range(1, 5) as $id) { + DB::table('releases')->insert($this->releaseRow($id, 'a')); + } + + $runner = new class extends PostProcessRunner + { + /** + * @var array + */ + public array $capturedCommands = []; + + protected function runStreamingCommands(array $commands, int $maxProcesses, string $desc): void + { + $this->capturedCommands = $commands; + } + }; + + $runner->processAdditional(); + + $this->assertSame([ + PHP_BINARY.' artisan postprocess:guid additional a --worker --max-batches=4', + ], array_values($runner->capturedCommands)); + } + + public function test_direct_guid_command_processes_one_batch(): void + { + $this->mock(AdditionalProcessingOrchestrator::class, function (MockInterface $mock): void { + $mock->shouldReceive('start') + ->once() + ->with('', 'a', Mockery::type('string'), []) + ->andReturn(['claimed' => 1, 'processed' => 1, 'failed' => 0, 'claimed_ids' => [1]]); + $mock->shouldReceive('finish')->once(); + }); + + $status = Artisan::call('postprocess:guid', [ + 'type' => 'additional', + 'guid' => 'a', + ]); + + $this->assertSame(0, $status); + } + + public function test_worker_guid_command_processes_bounded_distinct_batches(): void + { + $this->mock(AdditionalProcessingOrchestrator::class, function (MockInterface $mock): void { + $mock->shouldReceive('start') + ->once() + ->ordered() + ->with('', 'a', Mockery::type('string'), []) + ->andReturn(['claimed' => 1, 'processed' => 1, 'failed' => 0, 'claimed_ids' => [1]]); + $mock->shouldReceive('start') + ->once() + ->ordered() + ->with('', 'a', Mockery::type('string'), [1]) + ->andReturn(['claimed' => 1, 'processed' => 1, 'failed' => 0, 'claimed_ids' => [2]]); + $mock->shouldReceive('start') + ->once() + ->ordered() + ->with('', 'a', Mockery::type('string'), [1, 2]) + ->andReturn(['claimed' => 0, 'processed' => 0, 'failed' => 0, 'claimed_ids' => []]); + $mock->shouldReceive('finish')->once(); + }); + + $status = Artisan::call('postprocess:guid', [ + 'type' => 'additional', + 'guid' => 'a', + '--worker' => true, + '--max-batches' => 4, + ]); + + $this->assertSame(0, $status); + } + /** * @return array */ diff --git a/tests/Unit/TempWorkspaceServiceTest.php b/tests/Unit/TempWorkspaceServiceTest.php index c4e68370a..76b2bec44 100644 --- a/tests/Unit/TempWorkspaceServiceTest.php +++ b/tests/Unit/TempWorkspaceServiceTest.php @@ -57,6 +57,36 @@ class TempWorkspaceServiceTest extends TestCase $this->assertStringEndsWith('/group42/guid-123/', str_replace('\\', '/', $tmp)); } + #[WithoutErrorHandler] + public function test_worker_workspaces_are_isolated_within_a_guid_bucket(): void + { + $first = $this->svc->ensureMainTempPath($this->base, 'a', '', 'worker-one'); + $second = $this->svc->ensureMainTempPath($this->base, 'a', '', 'worker-two'); + File::put($first.'first.bin', 'first'); + File::put($second.'second.bin', 'second'); + + $this->svc->clearDirectory($first, false); + + $this->assertFalse(File::exists($first)); + $this->assertTrue(File::isFile($second.'second.bin')); + $this->assertNotSame($first, $second); + } + + #[WithoutErrorHandler] + public function test_prune_stale_worker_directories_keeps_active_siblings(): void + { + $bucket = $this->svc->ensureMainTempPath($this->base, 'a'); + $stale = $this->svc->ensureMainTempPath($this->base, 'a', '', 'stale-worker'); + $active = $this->svc->ensureMainTempPath($this->base, 'a', '', 'active-worker'); + touch(rtrim($stale, '/\\'), time() - 7200); + + $removed = $this->svc->pruneStaleWorkerDirectories($bucket, 3600); + + $this->assertSame(1, $removed); + $this->assertFalse(File::exists($stale)); + $this->assertTrue(File::isDirectory($active)); + } + #[WithoutErrorHandler] public function test_ensure_main_temp_path_repairs_existing_unwritable_bucket_directory(): void {