From 014077b514c7c9bd38d68fe4d88081a801fca918 Mon Sep 17 00:00:00 2001 From: DariusIII Date: Tue, 11 Aug 2026 10:43:22 +0200 Subject: [PATCH] Refactor additional postprocessing --- .env.example | 1 - .../Commands/AdditionalProcessingDiagnose.php | 66 +++ app/Console/Commands/PostProcessGuid.php | 46 ++- .../Commands/ProcessAdditionalGuid.php | 84 +++- .../AdditionalProcessingServiceProvider.php | 43 +- .../AdditionalCandidateQuery.php | 4 +- .../AdditionalProcessingDiagnostics.php | 250 ++++++++++++ .../AdditionalProcessingOrchestrator.php | 129 ++++-- .../AdditionalWorkPlanner.php | 205 ++++++++++ .../ArchiveExtractionService.php | 59 ++- .../DTO/AdditionalBatchResult.php | 243 +++++++++++ .../DTO/AdditionalWorkPlan.php | 61 +++ .../DTO/ArchiveCandidate.php | 18 + .../DTO/DownloadMetrics.php | 16 + .../DTO/DownloadedArchive.php | 22 + .../DTO/PersistenceMetrics.php | 15 + .../DTO/ReleaseProcessingResult.php | 59 +++ .../Enums/ProcessingOutcome.php | 35 ++ .../Enums/ProcessingStage.php | 20 + .../MediaExtractionService.php | 16 +- .../AdditionalProcessing/NzbContentParser.php | 36 +- .../ReleaseFileManager.php | 85 ++-- .../ReleaseFilesArchiveFallback.php | 142 +++++-- .../AdditionalProcessing/ReleaseProcessor.php | 349 ++++++++++++---- .../ReleaseSearchSyncCoordinator.php | 84 ++++ .../State/PersistenceMetricsCollector.php | 71 ++++ .../State/ProcessingMetrics.php | 59 +++ .../State/ReleaseProcessingContext.php | 58 +++ .../UsenetDownloadService.php | 128 +++++- .../NameFixing/ReleaseUpdateService.php | 13 +- app/Services/Runners/PostProcessRunner.php | 8 +- phpstan-baseline.neon | 36 -- ...AdditionalProcessingNzbSplitRenameTest.php | 19 + ...itionalProcessingOrchestratorClaimTest.php | 48 ++- ...tionalProcessingReleaseFileManagerTest.php | 71 +++- ...PostProcessRunnerAdditionalThreadsTest.php | 161 +++++++- .../AdditionalCandidateQueryMariaDbTest.php | 386 ++++++++++++++++++ .../AdditionalWorkPlannerTest.php | 100 +++++ .../ArchiveExtractionServiceTest.php | 88 ++++ .../MediaExtractionServiceTest.php | 66 +++ .../ProcessingResultTest.php | 142 +++++++ .../ReleaseFilesArchiveFallbackTest.php | 139 ++++++- .../ReleaseProcessorTest.php | 333 ++++++++++++++- .../ReleaseSearchSyncCoordinatorTest.php | 106 +++++ .../UsenetDownloadServiceTest.php | 93 +++++ 45 files changed, 3875 insertions(+), 338 deletions(-) create mode 100644 app/Console/Commands/AdditionalProcessingDiagnose.php create mode 100644 app/Services/AdditionalProcessing/AdditionalProcessingDiagnostics.php create mode 100644 app/Services/AdditionalProcessing/AdditionalWorkPlanner.php create mode 100644 app/Services/AdditionalProcessing/DTO/AdditionalBatchResult.php create mode 100644 app/Services/AdditionalProcessing/DTO/AdditionalWorkPlan.php create mode 100644 app/Services/AdditionalProcessing/DTO/ArchiveCandidate.php create mode 100644 app/Services/AdditionalProcessing/DTO/DownloadMetrics.php create mode 100644 app/Services/AdditionalProcessing/DTO/DownloadedArchive.php create mode 100644 app/Services/AdditionalProcessing/DTO/PersistenceMetrics.php create mode 100644 app/Services/AdditionalProcessing/DTO/ReleaseProcessingResult.php create mode 100644 app/Services/AdditionalProcessing/Enums/ProcessingOutcome.php create mode 100644 app/Services/AdditionalProcessing/Enums/ProcessingStage.php create mode 100644 app/Services/AdditionalProcessing/ReleaseSearchSyncCoordinator.php create mode 100644 app/Services/AdditionalProcessing/State/PersistenceMetricsCollector.php create mode 100644 app/Services/AdditionalProcessing/State/ProcessingMetrics.php create mode 100644 tests/Integration/AdditionalCandidateQueryMariaDbTest.php create mode 100644 tests/Unit/AdditionalProcessing/AdditionalWorkPlannerTest.php create mode 100644 tests/Unit/AdditionalProcessing/MediaExtractionServiceTest.php create mode 100644 tests/Unit/AdditionalProcessing/ProcessingResultTest.php create mode 100644 tests/Unit/AdditionalProcessing/ReleaseSearchSyncCoordinatorTest.php diff --git a/.env.example b/.env.example index 6cb409a82..901036c30 100644 --- a/.env.example +++ b/.env.example @@ -79,7 +79,6 @@ NNTP_SOCKET_TIMEOUT_A=120 NN_MULTIPROCESSING_MAX_CHILD_TIME=1800 NN_CONCURRENCY_TIMEOUT= - ADMIN_USER= ADMIN_PASS= ADMIN_EMAIL= diff --git a/app/Console/Commands/AdditionalProcessingDiagnose.php b/app/Console/Commands/AdditionalProcessingDiagnose.php new file mode 100644 index 000000000..7293dbbd0 --- /dev/null +++ b/app/Console/Commands/AdditionalProcessingDiagnose.php @@ -0,0 +1,66 @@ +diagnostics->inspect(); + + if ($this->option('json')) { + $this->line((string) json_encode($report, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES | JSON_THROW_ON_ERROR)); + + return self::SUCCESS; + } + + $this->table(['Backlog', 'Count'], [ + ['Total', (string) $report['backlog']['total']], + ['Available', (string) $report['backlog']['available']], + ['Active claims', (string) $report['backlog']['active_claims']], + ['Stale claims', (string) $report['backlog']['stale_claims']], + ]); + $this->table(['Capacity', 'Value'], [ + ['Pipeline', (string) $report['settings']['pipeline']], + ['Threads', (string) $report['capacity']['threads']], + ['Releases per batch', (string) $report['capacity']['releases_per_batch']], + ['Maximum in flight', (string) $report['capacity']['max_in_flight']], + ['Claim TTL', $report['claims']['ttl_seconds'].' seconds'], + ['Child timeout', $report['settings']['multiprocessing_child_timeout'].' seconds'], + ]); + $this->table(['Preflight', 'Value'], [ + ['Database driver', (string) $report['database_driver']], + ['Covering claim index', (string) ($report['indexes']['covering_index'] ?? 'missing')], + ['Temp path', (string) $report['temp_path']['path']], + ['Temp path writable', $report['temp_path']['writable'] ? 'yes' : 'no'], + ]); + + if ($report['warnings'] === []) { + $this->info('No additional-processing diagnostic warnings were found.'); + + return self::SUCCESS; + } + + $this->newLine(); + foreach ($report['warnings'] as $warning) { + $this->warn('['.$warning['code'].'] '.$warning['message']); + } + + return self::SUCCESS; + } +} diff --git a/app/Console/Commands/PostProcessGuid.php b/app/Console/Commands/PostProcessGuid.php index 9bb3ebf39..3e854e643 100644 --- a/app/Console/Commands/PostProcessGuid.php +++ b/app/Console/Commands/PostProcessGuid.php @@ -6,6 +6,7 @@ namespace App\Console\Commands; use App\Models\Settings; use App\Services\AdditionalProcessing\AdditionalProcessingOrchestrator; +use App\Services\AdditionalProcessing\DTO\AdditionalBatchResult; use App\Services\NfoService; use App\Services\NNTP\NNTPService; use App\Services\PostProcessService; @@ -24,7 +25,8 @@ class PostProcessGuid extends Command {guid : First character of release guid (a-f, 0-9)} {renamed? : For movie/tv: process renamed only (optional)} {--worker : Drain multiple additional-processing batches} - {--max-batches=4 : Maximum claim batches for worker mode}'; + {--max-batches=4 : Maximum claim batches for worker mode} + {--profile : Emit one machine-readable performance record per additional batch}'; /** * The console command description. @@ -93,13 +95,16 @@ class PostProcessGuid extends Command try { for ($batch = 0; $batch < $maxBatches && microtime(true) < $deadline; $batch++) { - $stats = $this->additionalProcessor->start('', $guid, $workerToken, $excludedReleaseIds); - if ($stats['claimed'] === 0) { + $result = $this->additionalProcessor->start('', $guid, $workerToken, $excludedReleaseIds); + if ((bool) $this->option('profile')) { + $this->writeAdditionalProfile($guid, $batch + 1, $result); + } + if ($result->claimedCount() === 0) { break; } $excludedReleaseIds = array_values(array_unique([ ...$excludedReleaseIds, - ...$stats['claimed_ids'], + ...$result->claimedIds, ])); } } finally { @@ -107,6 +112,39 @@ class PostProcessGuid extends Command } } + private function writeAdditionalProfile(string $guid, int $batch, AdditionalBatchResult $result): void + { + $downloadMetrics = $result->downloadMetrics(); + $persistenceMetrics = $result->persistenceMetrics(); + + $this->line(json_encode([ + 'event' => 'additional-postprocessing-profile', + 'pipeline' => 'v2', + 'guid_char' => $guid, + 'batch' => $batch, + 'claimed' => $result->claimedCount(), + 'attempted' => $result->attemptedCount(), + 'successful' => $result->successfulCount(), + 'failed' => $result->unsuccessfulCount(), + 'outcomes' => $result->outcomeCounts(), + 'artifacts_created' => $result->artifactsCreatedCount(), + 'artifact_yield_percent' => round($result->artifactYieldPercent(), 2), + 'release_files_added' => $result->releaseFilesAdded(), + 'nntp_requests' => $downloadMetrics->networkRequests, + 'nntp_bytes' => $downloadMetrics->bytesDownloaded, + 'database_statements' => $persistenceMetrics->databaseStatements, + 'database_milliseconds' => round($persistenceMetrics->databaseMilliseconds, 3), + 'search_sync_requests' => $persistenceMetrics->searchSyncRequests, + 'search_sync_executions' => $persistenceMetrics->searchSyncExecutions, + 'duplicate_message_ids' => $result->duplicateMessageIdCount(), + 'elapsed_seconds' => round($result->elapsedSeconds, 6), + 'releases_per_hour' => round($result->releasesPerSecond() * 3600, 2), + 'average_release_seconds' => round($result->averageReleaseSeconds(), 6), + 'stage_seconds' => $result->stageDurationTotals(), + 'peak_memory_bytes' => $result->peakMemoryBytes, + ], JSON_THROW_ON_ERROR)); + } + /** * Process NFO files for releases. */ diff --git a/app/Console/Commands/ProcessAdditionalGuid.php b/app/Console/Commands/ProcessAdditionalGuid.php index 0096f4aa1..8f23fc14f 100644 --- a/app/Console/Commands/ProcessAdditionalGuid.php +++ b/app/Console/Commands/ProcessAdditionalGuid.php @@ -7,6 +7,7 @@ namespace App\Console\Commands; use App\Facades\Search; use App\Models\Release; use App\Services\AdditionalProcessing\AdditionalProcessingOrchestrator; +use App\Services\AdditionalProcessing\DTO\ReleaseProcessingResult; use Illuminate\Console\Command; class ProcessAdditionalGuid extends Command @@ -15,7 +16,8 @@ class ProcessAdditionalGuid extends Command {guid? : Release GUID (optional if using --id)} {--id= : Numeric release ID to run additional postprocessing on (alternative to GUID)} {--reset : Reset key additional postprocessing flags before running} - {--no-progress : Suppress progress / status output (forces quiet mode)}'; + {--no-progress : Suppress progress / status output (forces quiet mode)} + {--profile : Display stage, network, and persistence metrics}'; protected $description = 'Run additional postprocessing for a specific release by GUID or ID regardless of prior postprocessing status'; @@ -87,19 +89,89 @@ class ProcessAdditionalGuid extends Command $processor = app(AdditionalProcessingOrchestrator::class); try { - $ok = $processor->processSingleGuid($guid); + $result = $processor->processSingleGuid($guid); } finally { $processor->finish(); } - if (! $ok) { - $this->error('Processing failed or nothing processed for '.($idOpt ? 'ID '.$idOpt : 'GUID '.$guid).'.'); + $identifier = $idOpt ? 'ID '.$idOpt : 'GUID '.$guid; + $this->displayResult($result, $identifier); + + if ($this->option('profile')) { + $this->displayProfile($result); + } + + if (! $result->isSuccessful()) { return 3; } - $this->info('Additional postprocessing completed for '.($idOpt ? 'ID '.$idOpt : 'GUID '.$guid).'.'); - return 0; } + + private function displayResult(ReleaseProcessingResult $result, string $identifier): void + { + $rows = [ + ['Release', $identifier], + ['Pipeline', 'v2'], + ['Outcome', $result->outcome->value], + ['Artifacts created', $result->artifactsCreated ? 'yes' : 'no'], + ['Release files added', (string) $result->releaseFilesAdded], + ['Duration', number_format($result->elapsedSeconds, 3).' seconds'], + ]; + + if ($result->reason !== '') { + $rows[] = ['Reason', $result->reason]; + } + + $this->table(['Result', 'Value'], $rows); + + $message = 'Additional postprocessing ended with outcome '.$result->outcome->value.' for '.$identifier + .' after '.number_format($result->elapsedSeconds, 3).' seconds.'; + + if ($result->isSuccessful()) { + $this->info($message); + + return; + } + + $this->error($message); + } + + private function displayProfile(ReleaseProcessingResult $result): void + { + $stageRows = []; + foreach ($result->stageDurations as $stage => $duration) { + $stageRows[] = [$stage, number_format($duration, 3)]; + } + + if ($stageRows !== []) { + $this->newLine(); + $this->line('Stage profile'); + $this->table(['Stage', 'Seconds'], $stageRows); + } + + if ($result->downloadMetrics !== null) { + $this->newLine(); + $this->line('Download profile'); + $this->table(['Metric', 'Value'], [ + ['Logical requests', (string) $result->downloadMetrics->logicalRequests], + ['Network requests', (string) $result->downloadMetrics->networkRequests], + ['Cache hits', (string) $result->downloadMetrics->cacheHits], + ['Bytes downloaded', (string) $result->downloadMetrics->bytesDownloaded], + ['Bytes reused', (string) $result->downloadMetrics->bytesReused], + ]); + } + + if ($result->persistenceMetrics !== null) { + $this->newLine(); + $this->line('Persistence profile'); + $this->table(['Metric', 'Value'], [ + ['Database statements', (string) $result->persistenceMetrics->databaseStatements], + ['Database milliseconds', number_format($result->persistenceMetrics->databaseMilliseconds, 3)], + ['Search sync requests', (string) $result->persistenceMetrics->searchSyncRequests], + ['Search sync executions', (string) $result->persistenceMetrics->searchSyncExecutions], + ]); + } + } } diff --git a/app/Providers/AdditionalProcessingServiceProvider.php b/app/Providers/AdditionalProcessingServiceProvider.php index 571dfd1d4..83b931017 100644 --- a/app/Providers/AdditionalProcessingServiceProvider.php +++ b/app/Providers/AdditionalProcessingServiceProvider.php @@ -5,6 +5,7 @@ declare(strict_types=1); namespace App\Providers; use App\Services\AdditionalProcessing\AdditionalProcessingOrchestrator; +use App\Services\AdditionalProcessing\AdditionalWorkPlanner; use App\Services\AdditionalProcessing\ArchiveExtractionService; use App\Services\AdditionalProcessing\Config\ProcessingConfiguration; use App\Services\AdditionalProcessing\ConsoleOutputService; @@ -13,15 +14,20 @@ use App\Services\AdditionalProcessing\NzbContentParser; use App\Services\AdditionalProcessing\ReleaseFileManager; use App\Services\AdditionalProcessing\ReleaseFilesArchiveFallback; use App\Services\AdditionalProcessing\ReleaseProcessor; +use App\Services\AdditionalProcessing\ReleaseSearchSyncCoordinator; +use App\Services\AdditionalProcessing\State\PersistenceMetricsCollector; use App\Services\AdditionalProcessing\UsenetDownloadService; use App\Services\Categorization\CategorizationService; use App\Services\NameFixing\NameFixingService; +use App\Services\NameFixing\ReleaseUpdateService; use App\Services\NfoService; use App\Services\Nzb\NzbParserService; use App\Services\Nzb\NzbService; use App\Services\ReleaseExtraService; use App\Services\ReleaseImageService; use App\Services\TempWorkspaceService; +use Illuminate\Database\Events\QueryExecuted; +use Illuminate\Support\Facades\DB; use Illuminate\Support\ServiceProvider; /** @@ -40,6 +46,18 @@ class AdditionalProcessingServiceProvider extends ServiceProvider return new ProcessingConfiguration; }); + $this->app->singleton(PersistenceMetricsCollector::class); + $this->app->singleton(ReleaseSearchSyncCoordinator::class, function ($app) { + return new ReleaseSearchSyncCoordinator( + $app->make(PersistenceMetricsCollector::class), + ); + }); + $this->app->bind(ReleaseUpdateService::class, function ($app) { + return new ReleaseUpdateService( + searchSyncCoordinator: $app->make(ReleaseSearchSyncCoordinator::class), + ); + }); + // Release extra service for video/audio/subtitle data $this->app->singleton(ReleaseExtraService::class, function () { return new ReleaseExtraService; @@ -64,6 +82,10 @@ class AdditionalProcessingServiceProvider extends ServiceProvider ); }); + $this->app->singleton(AdditionalWorkPlanner::class, function ($app) { + return new AdditionalWorkPlanner($app->make(ProcessingConfiguration::class)); + }); + // Archive extraction service $this->app->singleton(ArchiveExtractionService::class, function ($app) { return new ArchiveExtractionService( @@ -85,7 +107,9 @@ class AdditionalProcessingServiceProvider extends ServiceProvider $app->make(ReleaseImageService::class), new NfoService, $app->make(NzbService::class), - new NameFixingService + new NameFixingService($app->make(ReleaseUpdateService::class)), + $app->make(ReleaseUpdateService::class), + searchSyncCoordinator: $app->make(ReleaseSearchSyncCoordinator::class), ); }); @@ -95,18 +119,18 @@ class AdditionalProcessingServiceProvider extends ServiceProvider $app->make(ProcessingConfiguration::class), $app->make(ReleaseImageService::class), $app->make(ReleaseExtraService::class), - new CategorizationService + new CategorizationService, + $app->make(ReleaseSearchSyncCoordinator::class), ); }); $this->app->singleton(ReleaseFilesArchiveFallback::class, function ($app) { return new ReleaseFilesArchiveFallback( - $app->make(ProcessingConfiguration::class), $app->make(ArchiveExtractionService::class), $app->make(MediaExtractionService::class), $app->make(UsenetDownloadService::class), $app->make(ReleaseFileManager::class), - $app->make(ConsoleOutputService::class) + $app->make(ConsoleOutputService::class), ); }); @@ -114,13 +138,16 @@ class AdditionalProcessingServiceProvider extends ServiceProvider return new ReleaseProcessor( $app->make(ProcessingConfiguration::class), $app->make(NzbContentParser::class), + $app->make(AdditionalWorkPlanner::class), $app->make(ArchiveExtractionService::class), $app->make(MediaExtractionService::class), $app->make(UsenetDownloadService::class), $app->make(ReleaseFileManager::class), $app->make(ReleaseFilesArchiveFallback::class), $app->make(TempWorkspaceService::class), - $app->make(ConsoleOutputService::class) + $app->make(ConsoleOutputService::class), + $app->make(ReleaseSearchSyncCoordinator::class), + $app->make(PersistenceMetricsCollector::class), ); }); @@ -143,8 +170,12 @@ class AdditionalProcessingServiceProvider extends ServiceProvider /** * Bootstrap services. */ - public function boot(): void + public function boot(PersistenceMetricsCollector $persistenceMetrics): void { + DB::listen(static function (QueryExecuted $query) use ($persistenceMetrics): void { + $persistenceMetrics->recordDatabaseStatement($query->time); + }); + $this->app->terminating(function (): void { if ($this->app->bound(AdditionalProcessingOrchestrator::class)) { $this->app->make(AdditionalProcessingOrchestrator::class)->finish(); diff --git a/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php b/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php index b2a3791ce..9e865deb5 100644 --- a/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php +++ b/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php @@ -390,14 +390,14 @@ final class AdditionalCandidateQuery }); } - private static function claimTtlSeconds(): int + public static function claimTtlSeconds(): int { $timeout = (int) (Settings::settingValue('releaseprocessingtimeout') ?: 120); return max(300, $timeout * 2); } - private static function claimStaleBefore(): Carbon + public static function claimStaleBefore(): Carbon { return now()->subSeconds(self::claimTtlSeconds()); } diff --git a/app/Services/AdditionalProcessing/AdditionalProcessingDiagnostics.php b/app/Services/AdditionalProcessing/AdditionalProcessingDiagnostics.php new file mode 100644 index 000000000..63253c6de --- /dev/null +++ b/app/Services/AdditionalProcessing/AdditionalProcessingDiagnostics.php @@ -0,0 +1,250 @@ + + */ + private const array REQUIRED_CLAIM_INDEX_COLUMNS = [ + 'passwordstatus', + 'haspreview', + 'nzbstatus', + 'leftguid', + 'additional_pp_claimed_at', + 'postdate', + ]; + + /** + * @return array + */ + public function inspect(): array + { + $warnings = []; + $releaseTimeout = $this->setting('releaseprocessingtimeout', 120); + $threads = max(1, $this->setting('postthreads', 1)); + $batchSize = max(1, $this->setting('maxaddprocessed', 25)); + $childTimeout = (int) (config('nntmux.concurrency_timeout') + ?? config('nntmux.multiprocessing_max_child_time', 1800)); + $claimTtl = AdditionalCandidateQuery::claimTtlSeconds(); + $backlog = $this->backlog($warnings); + $indexes = $this->indexes($warnings); + $tempPath = $this->tempPath($warnings); + $maximumClaimLifetime = min( + max(1, $childTimeout), + max(1, $releaseTimeout) * $batchSize, + ); + + if ($backlog['stale_claims'] > 0) { + $warnings[] = $this->warning( + 'stale-claims', + $backlog['stale_claims'].' additional-processing claim(s) are stale and eligible for recovery.', + ); + } + + if ($claimTtl <= $maximumClaimLifetime) { + $warnings[] = $this->warning( + 'claim-ttl', + 'Claim TTL ('.$claimTtl.'s) does not exceed the estimated claimed-batch lifetime (' + .$maximumClaimLifetime.'s); another worker may reclaim active work.', + ); + } + + $maxInFlight = $threads * $batchSize; + if ($threads > self::OVERSIZED_THREAD_THRESHOLD || $maxInFlight > self::OVERSIZED_IN_FLIGHT_THRESHOLD) { + $warnings[] = $this->warning( + 'oversized-capacity', + 'postthreads='.$threads.' and maxaddprocessed='.$batchSize.' allow up to '.$maxInFlight + .' claimed releases at once; reduce threads or batch size and benchmark end-to-end throughput.', + ); + } + + return [ + 'generated_at' => now()->toIso8601String(), + 'database_driver' => DB::getDriverName(), + 'backlog' => $backlog, + 'settings' => [ + 'pipeline' => 'v2', + 'postthreads' => $threads, + 'maxaddprocessed' => $batchSize, + 'releaseprocessingtimeout' => $releaseTimeout, + 'maxpptimeoutcount' => $this->setting('maxpptimeoutcount', 3), + 'worker_max_batches' => PostProcessRunner::ADDITIONAL_WORKER_MAX_BATCHES, + 'multiprocessing_child_timeout' => $childTimeout, + ], + 'claims' => [ + 'ttl_seconds' => $claimTtl, + 'stale_before' => AdditionalCandidateQuery::claimStaleBefore()->toIso8601String(), + 'estimated_claimed_batch_lifetime_seconds' => $maximumClaimLifetime, + ], + 'capacity' => [ + 'threads' => $threads, + 'releases_per_batch' => $batchSize, + 'max_in_flight' => $maxInFlight, + 'max_per_worker_lifetime' => $batchSize * PostProcessRunner::ADDITIONAL_WORKER_MAX_BATCHES, + ], + 'indexes' => $indexes, + 'temp_path' => $tempPath, + 'warnings' => $warnings, + ]; + } + + /** + * @param list $warnings + * @return array{total: int, available: int, active_claims: int, stale_claims: int, buckets: list} + */ + private function backlog(array &$warnings): array + { + $empty = [ + 'total' => 0, + 'available' => 0, + 'active_claims' => 0, + 'stale_claims' => 0, + 'buckets' => [], + ]; + + if (! Schema::hasTable('releases') || ! Schema::hasTable('categories')) { + $warnings[] = $this->warning( + 'missing-processing-tables', + 'The releases or categories table is missing; backlog health cannot be inspected.', + ); + + return $empty; + } + + $counts = AdditionalCandidateQuery::backlogCounts(); + $activeClaims = 0; + $staleClaims = 0; + + if (AdditionalCandidateQuery::supportsClaims()) { + $claimQuery = AdditionalCandidateQuery::baseBuilder(includeClaimed: true) + ->whereNotNull('r.'.AdditionalCandidateQuery::CLAIMED_AT_COLUMN); + + $activeClaims = (clone $claimQuery) + ->where('r.'.AdditionalCandidateQuery::CLAIMED_AT_COLUMN, '>=', AdditionalCandidateQuery::claimStaleBefore()) + ->count('r.id'); + $staleClaims = (clone $claimQuery) + ->where('r.'.AdditionalCandidateQuery::CLAIMED_AT_COLUMN, '<', AdditionalCandidateQuery::claimStaleBefore()) + ->count('r.id'); + } + + return [ + ...$counts, + 'active_claims' => $activeClaims, + 'stale_claims' => $staleClaims, + 'buckets' => AdditionalCandidateQuery::bucketBacklog(), + ]; + } + + /** + * @param list $warnings + * @return array{required_columns: list, present: list, covering_index: string|null, missing: bool} + */ + private function indexes(array &$warnings): array + { + $present = []; + $coveringIndex = null; + + if (Schema::hasTable('releases')) { + foreach (Schema::getIndexes('releases') as $index) { + $name = (string) ($index['name'] ?? ''); + $columns = array_map( + static fn (mixed $column): string => strtolower((string) $column), + is_array($index['columns'] ?? null) ? $index['columns'] : [], + ); + + if ($name !== '') { + $present[] = $name; + } + if ($columns === self::REQUIRED_CLAIM_INDEX_COLUMNS) { + $coveringIndex = $name; + } + } + } + + sort($present); + $missing = $coveringIndex === null; + if ($missing) { + $warnings[] = $this->warning( + 'missing-index', + 'The releases claim-queue index is missing or has the wrong column order; run pending migrations before processing.', + ); + } + + return [ + 'required_columns' => self::REQUIRED_CLAIM_INDEX_COLUMNS, + 'present' => $present, + 'covering_index' => $coveringIndex, + 'missing' => $missing, + ]; + } + + /** + * @param list $warnings + * @return array{path: string, exists: bool, writable: bool} + */ + private function tempPath(array &$warnings): array + { + $path = rtrim((string) config('nntmux.tmp_unrar_path'), '/\\'); + $exists = $path !== '' && is_dir($path); + $writable = $exists ? is_writable($path) : $this->nearestExistingDirectoryIsWritable($path); + + if (! $writable) { + $warnings[] = $this->warning( + 'unwritable-temp-path', + 'Additional-processing temp path '.$path.' is not writable and cannot be created by the current user.', + ); + } + + return [ + 'path' => $path, + 'exists' => $exists, + 'writable' => $writable, + ]; + } + + private function nearestExistingDirectoryIsWritable(string $path): bool + { + if ($path === '') { + return false; + } + + $candidate = $path; + while (! is_dir($candidate)) { + $parent = dirname($candidate); + if ($parent === $candidate) { + return false; + } + $candidate = $parent; + } + + return is_writable($candidate); + } + + private function setting(string $name, int $default): int + { + $value = Settings::settingValue($name); + + return $value === null || $value === '' ? $default : (int) $value; + } + + /** + * @return array{code: string, message: string} + */ + private function warning(string $code, string $message): array + { + return ['code' => $code, 'message' => $message]; + } +} diff --git a/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php b/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php index c73cc8d52..c23ec7b27 100644 --- a/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php +++ b/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php @@ -6,6 +6,9 @@ namespace App\Services\AdditionalProcessing; use App\Models\Release; use App\Services\AdditionalProcessing\Config\ProcessingConfiguration; +use App\Services\AdditionalProcessing\DTO\AdditionalBatchResult; +use App\Services\AdditionalProcessing\DTO\ReleaseProcessingResult; +use App\Services\AdditionalProcessing\Enums\ProcessingOutcome; use App\Services\AdditionalProcessing\State\ReleaseProcessingContext; use App\Services\TempWorkspaceService; use Exception; @@ -31,6 +34,8 @@ class AdditionalProcessingOrchestrator private string $claimToken = ''; + private string $lastSetupError = ''; + public function __construct( private readonly ProcessingConfiguration $config, private readonly ReleaseProcessor $processor, @@ -42,7 +47,6 @@ class AdditionalProcessingOrchestrator * Start the additional processing. * * @param list $excludedReleaseIds - * @return array{claimed: int, processed: int, failed: int, claimed_ids: list} * * @throws Exception */ @@ -51,10 +55,10 @@ class AdditionalProcessingOrchestrator string $guidChar = '', string $workerToken = '', array $excludedReleaseIds = [] - ): array { + ): AdditionalBatchResult { $this->finish(); if (! $this->setupTempPath($guidChar, $groupID, $workerToken)) { - return ['claimed' => 0, 'processed' => 0, 'failed' => 0, 'claimed_ids' => []]; + return AdditionalBatchResult::setupFailed($this->lastSetupError); } $this->fetchReleases($groupID, $guidChar, $excludedReleaseIds); @@ -65,40 +69,59 @@ class AdditionalProcessingOrchestrator return $this->processReleases($guidChar); } - return ['claimed' => 0, 'processed' => 0, 'failed' => 0, 'claimed_ids' => []]; + return AdditionalBatchResult::empty(); } /** * Process a single release by GUID. */ - public function processSingleGuid(string $guid): bool + public function processSingleGuid(string $guid): ReleaseProcessingResult { + $releaseId = 0; + try { $this->finish(); $release = Release::where('guid', $guid)->first(); if ($release === null) { $this->output->warning('Release not found for GUID: '.$guid); - return false; + return new ReleaseProcessingResult(0, $guid, ProcessingOutcome::NotFound, reason: 'Release not found.'); } + $releaseId = (int) $release->id; + $this->releases = collect([$release]); $this->totalReleases = 1; $guidChar = $release->leftguid ?? substr($release->guid, 0, 1); $groupID = ''; if (! $this->setupTempPath($guidChar, $groupID)) { - return false; + return new ReleaseProcessingResult( + $releaseId, + $guid, + ProcessingOutcome::TemporaryWorkspaceUnavailable, + reason: $this->lastSetupError, + ); } - $this->processReleases($guidChar); + $result = $this->processReleases($guidChar)->firstResult(); - return true; + return $result ?? new ReleaseProcessingResult( + $releaseId, + $guid, + ProcessingOutcome::Failed, + reason: 'No processing result was produced.', + ); } catch (\Throwable $e) { if ($this->config->debugMode) { Log::error('processSingleGuid failed: '.$e->getMessage()); } - return false; + return new ReleaseProcessingResult( + $releaseId, + $guid, + ProcessingOutcome::Failed, + reason: $e->getMessage(), + ); } } @@ -107,6 +130,8 @@ class AdditionalProcessingOrchestrator */ private function setupTempPath(string $guidChar, string $groupID, string $workerToken = ''): bool { + $this->lastSetupError = ''; + try { $this->mainTmpPath = $this->tempWorkspace->ensureMainTempPath( $this->config->tmpUnrarPath, @@ -116,6 +141,7 @@ class AdditionalProcessingOrchestrator ); $this->tempWorkspace->clearDirectory($this->mainTmpPath, true); } catch (\Throwable $e) { + $this->lastSetupError = $e->getMessage(); $this->output->warning('Additional post-processing skipped: '.$e->getMessage()); Log::error('Additional post-processing temp path is unavailable', [ 'tmp_unrar_path' => $this->config->tmpUnrarPath, @@ -183,27 +209,31 @@ 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 = ''): array + private function processReleases(string $guidChar = ''): AdditionalBatchResult { - $startedAt = microtime(true); + $startedAt = hrtime(true); $claimedIds = $this->releases ->pluck('id') ->map(static fn (mixed $id): int => (int) $id) ->values() ->all(); - $processed = 0; - $failed = 0; + $results = []; foreach ($this->releases as $release) { + $releaseStartedAt = hrtime(true); + try { - $this->processor->process(new ReleaseProcessingContext($release), $this->mainTmpPath); - $processed++; + $results[] = $this->processor->process(new ReleaseProcessingContext($release), $this->mainTmpPath); } catch (\Throwable $e) { - $failed++; + $results[] = new ReleaseProcessingResult( + releaseId: (int) ($release->id ?? 0), + guid: (string) ($release->guid ?? ''), + outcome: ProcessingOutcome::Failed, + reason: $e->getMessage(), + elapsedSeconds: $this->elapsedSecondsSince($releaseStartedAt), + ); Log::error('Additional postprocessing failed for release '.($release->id ?? '?').': '.$e->getMessage(), [ 'release_id' => $release->id ?? null, 'guid' => $release->guid ?? null, @@ -220,23 +250,63 @@ class AdditionalProcessingOrchestrator } } + $batchResult = new AdditionalBatchResult( + claimedIds: $claimedIds, + results: $results, + elapsedSeconds: $this->elapsedSecondsSince($startedAt), + peakMemoryBytes: memory_get_peak_usage(true), + ); + $downloadMetrics = $batchResult->downloadMetrics(); + $persistenceMetrics = $batchResult->persistenceMetrics(); + Log::info('Additional postprocessing run finished', [ + 'pipeline' => 'v2', 'guid_char' => $guidChar, 'picked' => $this->totalReleases, - 'processed' => $processed, - 'failed' => $failed, - 'elapsed_seconds' => round(microtime(true) - $startedAt, 3), - 'peak_memory_bytes' => memory_get_peak_usage(true), + 'processed' => $batchResult->successfulCount(), + 'failed' => $batchResult->unsuccessfulCount(), + 'outcomes' => $batchResult->outcomeCounts(), + 'artifacts_created' => $batchResult->artifactsCreatedCount(), + 'artifact_yield_percent' => round($batchResult->artifactYieldPercent(), 2), + 'release_files_added' => $batchResult->releaseFilesAdded(), + 'download_requests' => $downloadMetrics->logicalRequests, + 'nntp_requests' => $downloadMetrics->networkRequests, + 'download_cache_hits' => $downloadMetrics->cacheHits, + 'nntp_bytes' => $downloadMetrics->bytesDownloaded, + 'reused_bytes' => $downloadMetrics->bytesReused, + 'database_statements' => $persistenceMetrics->databaseStatements, + 'database_milliseconds' => round($persistenceMetrics->databaseMilliseconds, 3), + 'search_sync_requests' => $persistenceMetrics->searchSyncRequests, + 'search_sync_executions' => $persistenceMetrics->searchSyncExecutions, + 'duplicate_message_ids' => $batchResult->duplicateMessageIdCount(), + 'unsupported_reasons' => $batchResult->unsupportedReasonCounts(), + 'elapsed_seconds' => round($batchResult->elapsedSeconds, 6), + 'releases_per_second' => round($batchResult->releasesPerSecond(), 4), + 'average_release_seconds' => round($batchResult->averageReleaseSeconds(), 6), + 'stage_seconds' => $this->roundedStageDurations($batchResult->stageDurationTotals()), + 'peak_memory_bytes' => $batchResult->peakMemoryBytes, ]); $this->output->endOutput(); - return [ - 'claimed' => $this->totalReleases, - 'processed' => $processed, - 'failed' => $failed, - 'claimed_ids' => $claimedIds, - ]; + return $batchResult; + } + + private function elapsedSecondsSince(int $startedAtNanoseconds): float + { + return (hrtime(true) - $startedAtNanoseconds) / 1_000_000_000; + } + + /** + * @param array $durations + * @return array + */ + private function roundedStageDurations(array $durations): array + { + return array_map( + static fn (float $duration): float => round($duration, 6), + $durations, + ); } public function finish(): void @@ -249,5 +319,6 @@ class AdditionalProcessingOrchestrator $this->releases = collect(); $this->totalReleases = 0; $this->claimToken = ''; + $this->lastSetupError = ''; } } diff --git a/app/Services/AdditionalProcessing/AdditionalWorkPlanner.php b/app/Services/AdditionalProcessing/AdditionalWorkPlanner.php new file mode 100644 index 000000000..925b01e56 --- /dev/null +++ b/app/Services/AdditionalProcessing/AdditionalWorkPlanner.php @@ -0,0 +1,205 @@ + $nzbContents + */ + public function plan(array $nzbContents, string $groupName): AdditionalWorkPlan + { + $sampleMessageIds = []; + $jpgMessageIds = []; + $mediaInfoMessageId = ''; + $audioInfoMessageId = ''; + $audioInfoExtension = ''; + $archiveCandidates = []; + $bookFileCount = 0; + $duplicateMessageIdCount = 0; + $seenMessageIds = []; + + foreach (array_values($nzbContents) as $sourceIndex => $file) { + if (! is_array($file)) { + continue; + } + + try { + $title = (string) ($file['title'] ?? ''); + $segments = is_array($file['segments'] ?? null) ? $file['segments'] : []; + + if (preg_match($this->config->ignoreBookRegex, $title) === 1) { + $bookFileCount++; + } + + if (preg_match(self::ARCHIVE_PATTERN, $title) === 1) { + $archiveMessageIds = $this->extractSegments( + $segments, + $this->config->maximumRarSegments, + $seenMessageIds, + $duplicateMessageIdCount, + ); + if ($archiveMessageIds !== []) { + $archiveCandidates[] = new ArchiveCandidate( + title: $title, + messageIds: $archiveMessageIds, + likelyFirstVolume: $this->isLikelyFirstVolume($title), + sourceIndex: $sourceIndex, + ); + } + } + + if ($this->isSupportFile($title)) { + continue; + } + + if ($this->config->processThumbnails && $sampleMessageIds === [] && $segments !== [] + && stripos($title, 'sample') !== false + && preg_match('/\.(?:jpe?g|png|webp)$/i', $title) !== 1 + ) { + $sampleMessageIds = $this->extractSegments( + $segments, + $this->config->segmentsToDownload, + $seenMessageIds, + $duplicateMessageIdCount, + ); + } + + if ($this->config->processJPGSample && $jpgMessageIds === [] && $segments !== [] + && preg_match('/flac|lossless|mp3|music|inner-sanctum|sound/i', $groupName) !== 1 + && preg_match('/\.(?:jpe?g|png|webp)[. ")\]]/i', $title) === 1 + ) { + $jpgMessageIds = $this->extractSegments( + $segments, + $this->config->segmentsToDownload, + $seenMessageIds, + $duplicateMessageIdCount, + ); + } + + if ($this->config->processMediaInfo && $mediaInfoMessageId === '' && isset($segments[0]) + && stripos($title, 'sample') !== false + && preg_match('/'.$this->config->videoFileRegex.'[. ")\]]/i', $title) === 1 + ) { + $mediaInfoMessageId = (string) $segments[0]; + $this->recordMessageId($mediaInfoMessageId, $seenMessageIds, $duplicateMessageIdCount); + } + + if ($this->config->processAudioInfo && $audioInfoMessageId === '' && isset($segments[0]) + && preg_match('/'.$this->config->audioFileRegex.'[. ")\]]/i', $title, $type) === 1 + ) { + $audioInfoExtension = (string) ($type[1] ?? ''); + $audioInfoMessageId = (string) $segments[0]; + $this->recordMessageId($audioInfoMessageId, $seenMessageIds, $duplicateMessageIdCount); + } + } catch (\ErrorException $e) { + Log::debug($e->getTraceAsString()); + } + } + + $bookFlood = $bookFileCount > 80 && ($bookFileCount * 2) >= count($nzbContents); + $unsupportedReasons = []; + if ($bookFlood) { + $unsupportedReasons[] = 'book-flood'; + } + if ($archiveCandidates === [] + && $sampleMessageIds === [] + && $jpgMessageIds === [] + && $mediaInfoMessageId === '' + && $audioInfoMessageId === '' + ) { + $unsupportedReasons[] = 'no-supported-candidates'; + } + + return new AdditionalWorkPlan( + sampleMessageIds: $sampleMessageIds, + jpgMessageIds: $jpgMessageIds, + mediaInfoMessageId: $mediaInfoMessageId, + audioInfoMessageId: $audioInfoMessageId, + audioInfoExtension: $audioInfoExtension, + archiveCandidates: $archiveCandidates, + bookFileCount: $bookFileCount, + bookFlood: $bookFlood, + duplicateMessageIdCount: $duplicateMessageIdCount, + unsupportedReasons: $unsupportedReasons, + ); + } + + private function isSupportFile(string $title): bool + { + return preg_match( + '/(?:'.$this->config->supportFileRegex.'|nfo\b|inf\b|ofn\b)($|[ ")]|-])(?!.{20,})/i', + $title, + ) === 1; + } + + private function isLikelyFirstVolume(string $title): bool + { + if (preg_match('/\.part0*(\d+)/i', $title, $part) === 1) { + return (int) $part[1] === 1; + } + + if (preg_match('/"[a-f0-9]{32}\.[1-9]\d{1,2}".*\((\d+)\/\d{2,}\)$/i', $title, $position) === 1) { + return (int) $position[1] === 1; + } + + return preg_match('/\.(rar|zip)($|[ ")]|-])/i', $title) === 1; + } + + /** + * @param array $segments + * @param array $seenMessageIds + * @return list + */ + private function extractSegments( + array $segments, + int $limit, + array &$seenMessageIds, + int &$duplicateMessageIdCount, + ): array { + $messageIds = []; + $requestMessageIds = []; + + foreach (array_slice($segments, 0, max($limit, 0)) as $segment) { + $messageId = (string) $segment; + if ($messageId === '' || isset($requestMessageIds[$messageId])) { + if ($messageId !== '') { + $duplicateMessageIdCount++; + } + + continue; + } + + $requestMessageIds[$messageId] = true; + $messageIds[] = $messageId; + $this->recordMessageId($messageId, $seenMessageIds, $duplicateMessageIdCount); + } + + return $messageIds; + } + + /** + * @param array $seenMessageIds + */ + private function recordMessageId(string $messageId, array &$seenMessageIds, int &$duplicateMessageIdCount): void + { + if (isset($seenMessageIds[$messageId])) { + $duplicateMessageIdCount++; + + return; + } + + $seenMessageIds[$messageId] = true; + } +} diff --git a/app/Services/AdditionalProcessing/ArchiveExtractionService.php b/app/Services/AdditionalProcessing/ArchiveExtractionService.php index 18d257543..a18e94017 100644 --- a/app/Services/AdditionalProcessing/ArchiveExtractionService.php +++ b/app/Services/AdditionalProcessing/ArchiveExtractionService.php @@ -167,8 +167,8 @@ class ArchiveExtractionService /** * Sort files to prioritize NFO files for processing. * - * @param array $files Array of file info arrays. - * @return array Sorted array with NFO files first. + * @param list> $files Array of file info arrays. + * @return list> Sorted array with NFO files first. */ public function sortFilesWithNfoPriority(array $files): array { @@ -325,23 +325,55 @@ class ArchiveExtractionService */ public function extractSpecificFile(string $compressedData, string $filename, string $tmpPath): ?string { - // Try using ArchiveInfo's built-in extraction + return $this->extractSpecificFiles($compressedData, [$filename], $tmpPath)[$filename] ?? null; + } + + /** + * Inspect one archive and extract several exact filenames from it. + * + * @param list $filenames + * @return array + */ + public function extractSpecificFiles(string $compressedData, array $filenames, string $tmpPath): array + { + $filenames = array_values(array_unique(array_filter( + $filenames, + static fn (string $filename): bool => $filename !== '', + ))); + $extractedFiles = []; + $remaining = $filenames; + if ($this->archiveInfo->setData($compressedData, true)) { - try { - $extracted = $this->archiveInfo->getFileData($filename); - if ($extracted !== false && ! empty($extracted)) { - return $extracted; - } - } catch (\Throwable $e) { - if ($this->config->debugMode) { - Log::debug('ArchiveInfo getFileData failed: '.$e->getMessage()); + foreach ($filenames as $filename) { + try { + $extracted = $this->archiveInfo->getFileData($filename); + if ($extracted !== false && $extracted !== '') { + $extractedFiles[$filename] = $extracted; + $remaining = array_values(array_diff($remaining, [$filename])); + } + } catch (\Throwable $e) { + if ($this->config->debugMode) { + Log::debug('ArchiveInfo getFileData failed: '.$e->getMessage()); + } } } } - // Fallback: use external tools to extract to temp directory + foreach ($remaining as $filename) { + $extracted = $this->extractSpecificFileWithExternalTools($compressedData, $filename, $tmpPath); + if ($extracted !== null) { + $extractedFiles[$filename] = $extracted; + } + } - // Try using unrar for RAR files + return $extractedFiles; + } + + private function extractSpecificFileWithExternalTools( + string $compressedData, + string $filename, + string $tmpPath, + ): ?string { if ($this->config->unrarPath) { $extracted = $this->extractFileViaExternalTool( $compressedData, @@ -356,7 +388,6 @@ class ArchiveExtractionService } } - // Try using unzip for ZIP files if ($this->config->unzipPath) { $extracted = $this->extractFileViaExternalTool( $compressedData, diff --git a/app/Services/AdditionalProcessing/DTO/AdditionalBatchResult.php b/app/Services/AdditionalProcessing/DTO/AdditionalBatchResult.php new file mode 100644 index 000000000..e6a9d5a68 --- /dev/null +++ b/app/Services/AdditionalProcessing/DTO/AdditionalBatchResult.php @@ -0,0 +1,243 @@ + $claimedIds + * @param list $results + */ + public function __construct( + public array $claimedIds = [], + public array $results = [], + public string $setupFailure = '', + public float $elapsedSeconds = 0.0, + public int $peakMemoryBytes = 0, + ) {} + + public static function empty(): self + { + return new self; + } + + public static function setupFailed(string $reason): self + { + return new self(setupFailure: $reason); + } + + public function claimedCount(): int + { + return count($this->claimedIds); + } + + public function attemptedCount(): int + { + return count($this->results); + } + + public function successfulCount(): int + { + return count(array_filter( + $this->results, + static fn (ReleaseProcessingResult $result): bool => $result->isSuccessful(), + )); + } + + public function unsuccessfulCount(): int + { + return $this->attemptedCount() - $this->successfulCount(); + } + + /** + * @return array + */ + public function outcomeCounts(): array + { + $counts = []; + + foreach ($this->results as $result) { + $outcome = $result->outcome->value; + $counts[$outcome] = ($counts[$outcome] ?? 0) + 1; + } + + return $counts; + } + + public function firstResult(): ?ReleaseProcessingResult + { + return $this->results[0] ?? null; + } + + public function hasOutcome(ProcessingOutcome $outcome): bool + { + foreach ($this->results as $result) { + if ($result->outcome === $outcome) { + return true; + } + } + + return false; + } + + public function withPerformance(float $elapsedSeconds, int $peakMemoryBytes): self + { + return new self( + claimedIds: $this->claimedIds, + results: $this->results, + setupFailure: $this->setupFailure, + elapsedSeconds: $elapsedSeconds, + peakMemoryBytes: $peakMemoryBytes, + ); + } + + public function releasesPerSecond(): float + { + if ($this->elapsedSeconds <= 0.0) { + return 0.0; + } + + return $this->attemptedCount() / $this->elapsedSeconds; + } + + public function averageReleaseSeconds(): float + { + if ($this->attemptedCount() === 0) { + return 0.0; + } + + return array_sum(array_map( + static fn (ReleaseProcessingResult $result): float => $result->elapsedSeconds, + $this->results, + )) / $this->attemptedCount(); + } + + public function artifactsCreatedCount(): int + { + return count(array_filter( + $this->results, + static fn (ReleaseProcessingResult $result): bool => $result->artifactsCreated, + )); + } + + public function artifactYieldPercent(): float + { + if ($this->successfulCount() === 0) { + return 0.0; + } + + $successfulArtifacts = count(array_filter( + $this->results, + static fn (ReleaseProcessingResult $result): bool => $result->isSuccessful() && $result->artifactsCreated, + )); + + return ($successfulArtifacts / $this->successfulCount()) * 100; + } + + public function releaseFilesAdded(): int + { + return array_sum(array_map( + static fn (ReleaseProcessingResult $result): int => $result->releaseFilesAdded, + $this->results, + )); + } + + public function downloadMetrics(): DownloadMetrics + { + $logicalRequests = 0; + $networkRequests = 0; + $cacheHits = 0; + $bytesDownloaded = 0; + $bytesReused = 0; + + foreach ($this->results as $result) { + if ($result->downloadMetrics === null) { + continue; + } + + $logicalRequests += $result->downloadMetrics->logicalRequests; + $networkRequests += $result->downloadMetrics->networkRequests; + $cacheHits += $result->downloadMetrics->cacheHits; + $bytesDownloaded += $result->downloadMetrics->bytesDownloaded; + $bytesReused += $result->downloadMetrics->bytesReused; + } + + return new DownloadMetrics( + logicalRequests: $logicalRequests, + networkRequests: $networkRequests, + cacheHits: $cacheHits, + bytesDownloaded: $bytesDownloaded, + bytesReused: $bytesReused, + ); + } + + public function persistenceMetrics(): PersistenceMetrics + { + $databaseStatements = 0; + $databaseMilliseconds = 0.0; + $searchSyncRequests = 0; + $searchSyncExecutions = 0; + + foreach ($this->results as $result) { + if ($result->persistenceMetrics === null) { + continue; + } + + $databaseStatements += $result->persistenceMetrics->databaseStatements; + $databaseMilliseconds += $result->persistenceMetrics->databaseMilliseconds; + $searchSyncRequests += $result->persistenceMetrics->searchSyncRequests; + $searchSyncExecutions += $result->persistenceMetrics->searchSyncExecutions; + } + + return new PersistenceMetrics( + databaseStatements: $databaseStatements, + databaseMilliseconds: $databaseMilliseconds, + searchSyncRequests: $searchSyncRequests, + searchSyncExecutions: $searchSyncExecutions, + ); + } + + public function duplicateMessageIdCount(): int + { + return array_sum(array_map( + static fn (ReleaseProcessingResult $result): int => $result->duplicateMessageIdCount, + $this->results, + )); + } + + /** + * @return array + */ + public function unsupportedReasonCounts(): array + { + $counts = []; + + foreach ($this->results as $result) { + foreach ($result->unsupportedReasons as $reason) { + $counts[$reason] = ($counts[$reason] ?? 0) + 1; + } + } + + return $counts; + } + + /** + * @return array + */ + public function stageDurationTotals(): array + { + $totals = []; + + foreach ($this->results as $result) { + foreach ($result->stageDurations as $stage => $duration) { + $totals[$stage] = ($totals[$stage] ?? 0.0) + $duration; + } + } + + return $totals; + } +} diff --git a/app/Services/AdditionalProcessing/DTO/AdditionalWorkPlan.php b/app/Services/AdditionalProcessing/DTO/AdditionalWorkPlan.php new file mode 100644 index 000000000..80c77622f --- /dev/null +++ b/app/Services/AdditionalProcessing/DTO/AdditionalWorkPlan.php @@ -0,0 +1,61 @@ + $sampleMessageIds + * @param list $jpgMessageIds + * @param list $archiveCandidates + * @param list $unsupportedReasons + */ + public function __construct( + public array $sampleMessageIds = [], + public array $jpgMessageIds = [], + public string $mediaInfoMessageId = '', + public string $audioInfoMessageId = '', + public string $audioInfoExtension = '', + public array $archiveCandidates = [], + public int $bookFileCount = 0, + public bool $bookFlood = false, + public int $duplicateMessageIdCount = 0, + public array $unsupportedReasons = [], + ) {} + + public function hasCompressedFile(): bool + { + return $this->archiveCandidates !== []; + } + + /** + * @return list + */ + public function orderedArchiveCandidates(bool $reverse = false): array + { + return $reverse ? array_reverse($this->archiveCandidates) : $this->archiveCandidates; + } + + /** + * Return a stable priority view without changing the legacy processing order. + * + * @return list + */ + public function prioritizedArchiveCandidates(): array + { + $firstVolumes = []; + $remaining = []; + + foreach ($this->archiveCandidates as $candidate) { + if ($candidate->likelyFirstVolume) { + $firstVolumes[] = $candidate; + } else { + $remaining[] = $candidate; + } + } + + return [...$firstVolumes, ...$remaining]; + } +} diff --git a/app/Services/AdditionalProcessing/DTO/ArchiveCandidate.php b/app/Services/AdditionalProcessing/DTO/ArchiveCandidate.php new file mode 100644 index 000000000..12efc15ba --- /dev/null +++ b/app/Services/AdditionalProcessing/DTO/ArchiveCandidate.php @@ -0,0 +1,18 @@ + $messageIds + */ + public function __construct( + public string $title, + public array $messageIds, + public bool $likelyFirstVolume, + public int $sourceIndex, + ) {} +} diff --git a/app/Services/AdditionalProcessing/DTO/DownloadMetrics.php b/app/Services/AdditionalProcessing/DTO/DownloadMetrics.php new file mode 100644 index 000000000..27061424a --- /dev/null +++ b/app/Services/AdditionalProcessing/DTO/DownloadMetrics.php @@ -0,0 +1,16 @@ +> $files + */ + public function __construct( + public string $title, + public string $data, + public array $files, + ) {} + + public function byteSize(): int + { + return strlen($this->data); + } +} diff --git a/app/Services/AdditionalProcessing/DTO/PersistenceMetrics.php b/app/Services/AdditionalProcessing/DTO/PersistenceMetrics.php new file mode 100644 index 000000000..99853dc45 --- /dev/null +++ b/app/Services/AdditionalProcessing/DTO/PersistenceMetrics.php @@ -0,0 +1,15 @@ + $stageDurations + * @param list $unsupportedReasons + */ + public function __construct( + public int $releaseId, + public string $guid, + public ProcessingOutcome $outcome, + public bool $artifactsCreated = false, + public int $releaseFilesAdded = 0, + public string $reason = '', + public float $elapsedSeconds = 0.0, + public array $stageDurations = [], + public ?DownloadMetrics $downloadMetrics = null, + public ?PersistenceMetrics $persistenceMetrics = null, + public int $duplicateMessageIdCount = 0, + public array $unsupportedReasons = [], + ) {} + + public function isSuccessful(): bool + { + return $this->outcome->isSuccessful(); + } + + /** + * @param array $stageDurations + */ + public function withPerformance( + float $elapsedSeconds, + array $stageDurations, + ?DownloadMetrics $downloadMetrics = null, + ?PersistenceMetrics $persistenceMetrics = null, + ): self { + return new self( + releaseId: $this->releaseId, + guid: $this->guid, + outcome: $this->outcome, + artifactsCreated: $this->artifactsCreated, + releaseFilesAdded: $this->releaseFilesAdded, + reason: $this->reason, + elapsedSeconds: $elapsedSeconds, + stageDurations: $stageDurations, + downloadMetrics: $downloadMetrics ?? $this->downloadMetrics, + persistenceMetrics: $persistenceMetrics ?? $this->persistenceMetrics, + duplicateMessageIdCount: $this->duplicateMessageIdCount, + unsupportedReasons: $this->unsupportedReasons, + ); + } +} diff --git a/app/Services/AdditionalProcessing/Enums/ProcessingOutcome.php b/app/Services/AdditionalProcessing/Enums/ProcessingOutcome.php new file mode 100644 index 000000000..ea8657b0b --- /dev/null +++ b/app/Services/AdditionalProcessing/Enums/ProcessingOutcome.php @@ -0,0 +1,35 @@ + true, + default => false, + }; + } + + public function isDeleted(): bool + { + return match ($this) { + self::DeletedAfterTimeout, self::DeletedBrokenNzb => true, + default => false, + }; + } +} diff --git a/app/Services/AdditionalProcessing/Enums/ProcessingStage.php b/app/Services/AdditionalProcessing/Enums/ProcessingStage.php new file mode 100644 index 000000000..9649580f2 --- /dev/null +++ b/app/Services/AdditionalProcessing/Enums/ProcessingStage.php @@ -0,0 +1,20 @@ +searchSyncCoordinator = $searchSyncCoordinator + ?? new ReleaseSearchSyncCoordinator( + new PersistenceMetricsCollector, + ); + } /** * Get video time code for sample extraction. @@ -329,7 +337,7 @@ class MediaExtractionService 'proc_pp' => 1, ]); - Search::updateRelease($context->release->id); + $this->searchSyncCoordinator->request((int) $context->release->id); if ($this->config->echoCLI) { $releaseInfo = (object) [ diff --git a/app/Services/AdditionalProcessing/NzbContentParser.php b/app/Services/AdditionalProcessing/NzbContentParser.php index 753c63025..78c06aae1 100644 --- a/app/Services/AdditionalProcessing/NzbContentParser.php +++ b/app/Services/AdditionalProcessing/NzbContentParser.php @@ -147,16 +147,16 @@ class NzbContentParser /** * Process NZB contents to extract message IDs for different file types. * - * @param array $nzbContents - * @return array - * hasCompressedFile: bool, - * sampleMessageIDs: array, - * jpgMessageIDs: array, - * mediaInfoMessageID: string, - * audioInfoMessageID: string, - * audioInfoExtension: string, - * bookFileCount: int - * } + * @param list> $nzbContents + * @return array{ + * hasCompressedFile: bool, + * sampleMessageIDs: list, + * jpgMessageIDs: list, + * mediaInfoMessageID: string, + * audioInfoMessageID: string, + * audioInfoExtension: string, + * bookFileCount: int + * } */ public function extractMessageIDs( array $nzbContents, @@ -194,7 +194,7 @@ class NzbContentParser // Look for a video sample (not an image) if ($config->processThumbnails && empty($result['sampleMessageIDs']) && ! empty($segments) && stripos($title, 'sample') !== false - && ! preg_match('/\.jpe?g$/i', $title) + && ! preg_match('/\.(?:jpe?g|png|webp)$/i', $title) ) { $result['sampleMessageIDs'] = $this->extractSegments($segments, $config->segmentsToDownload); } @@ -202,7 +202,7 @@ class NzbContentParser // Look for a JPG picture (not a CD cover) if ($config->processJPGSample && empty($result['jpgMessageIDs']) && ! empty($segments) && ! preg_match('/flac|lossless|mp3|music|inner-sanctum|sound/i', $groupName) - && preg_match('/\.jpe?g[. ")\]]/i', $title) + && preg_match('/\.(?:jpe?g|png|webp)[. ")\]]/i', $title) ) { $result['jpgMessageIDs'] = $this->extractSegments($segments, $config->segmentsToDownload); } @@ -238,18 +238,14 @@ class NzbContentParser /** * Extract segment message IDs up to a limit. * - * @param array $segments - * @return array + * @param array $segments + * @return list */ private function extractSegments(array $segments, int $limit): array { $ids = []; - $segCount = count($segments) - 1; - for ($i = 0; $i < $limit; $i++) { - if ($i > $segCount) { - break; - } - $ids[] = (string) $segments[$i]; // @phpstan-ignore offsetAccess.notFound + foreach (array_slice(array_values($segments), 0, max(0, $limit)) as $segment) { + $ids[] = (string) $segment; } return $ids; diff --git a/app/Services/AdditionalProcessing/ReleaseFileManager.php b/app/Services/AdditionalProcessing/ReleaseFileManager.php index 7aa2815f5..157533679 100644 --- a/app/Services/AdditionalProcessing/ReleaseFileManager.php +++ b/app/Services/AdditionalProcessing/ReleaseFileManager.php @@ -12,6 +12,7 @@ use App\Models\Predb; use App\Models\Release; use App\Models\ReleaseFile; use App\Services\AdditionalProcessing\Config\ProcessingConfiguration; +use App\Services\AdditionalProcessing\State\PersistenceMetricsCollector; use App\Services\AdditionalProcessing\State\ReleaseProcessingContext; use App\Services\NameFixing\FileNameCleaner; use App\Services\NameFixing\NameFixingService; @@ -24,6 +25,7 @@ use App\Services\Releases\ReleaseBrowseService; use dariusiii\rarinfo\Par2Info; use Illuminate\Contracts\Filesystem\FileNotFoundException; use Illuminate\Support\Carbon; +use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\File; use Illuminate\Support\Facades\Log; @@ -37,6 +39,8 @@ class ReleaseFileManager private readonly FileNameCleaner $fileNameCleaner; + private readonly ReleaseSearchSyncCoordinator $searchSyncCoordinator; + public function __construct( private readonly ProcessingConfiguration $config, private readonly ReleaseImageService $releaseImage, @@ -44,9 +48,15 @@ class ReleaseFileManager private readonly NzbService $nzb, private readonly NameFixingService $nameFixingService, ?ReleaseUpdateService $releaseUpdateService = null, - ?FileNameCleaner $fileNameCleaner = null + ?FileNameCleaner $fileNameCleaner = null, + ?ReleaseSearchSyncCoordinator $searchSyncCoordinator = null, ) { - $this->releaseUpdateService = $releaseUpdateService ?? new ReleaseUpdateService; + $this->searchSyncCoordinator = $searchSyncCoordinator + ?? new ReleaseSearchSyncCoordinator( + new PersistenceMetricsCollector, + ); + $this->releaseUpdateService = $releaseUpdateService + ?? new ReleaseUpdateService(searchSyncCoordinator: $this->searchSyncCoordinator); $this->fileNameCleaner = $fileNameCleaner ?? new FileNameCleaner; } @@ -156,7 +166,7 @@ class ReleaseFileManager */ public function updateSearchIndex(int $releaseId): void { - Search::updateRelease($releaseId); + $this->searchSyncCoordinator->request($releaseId); } /** @@ -213,8 +223,6 @@ class ReleaseFileManager */ public function finalizeRelease(ReleaseProcessingContext $context, bool $processPasswords): void { - $this->flushQueuedReleaseFiles($context); - $updateRows = ['haspreview' => 0]; // Check for existing samples @@ -230,9 +238,6 @@ class ReleaseFileManager $updateRows['jpgstatus'] = 1; } - // Get file count - $releaseFilesCount = ReleaseFile::whereReleasesId($context->release->id)->count('releases_id') ?? 0; - $passwordStatus = $context->passwordStatus; // Set to no password if processing is off @@ -242,16 +247,43 @@ class ReleaseFileManager $updateRows = array_merge($updateRows, AdditionalCandidateQuery::claimResetValues()); - // Update based on conditions - if (! $context->releaseHasPassword && $context->nzbHasCompressedFile && $releaseFilesCount === 0) { - Release::query()->where('id', $context->release->id)->update($updateRows); - } else { - $updateRows['passwordstatus'] = $processPasswords ? $passwordStatus : ReleaseBrowseService::PASSWD_NONE; - $updateRows['rarinnerfilecount'] = $releaseFilesCount; - Release::query()->where('id', $context->release->id)->update($updateRows); - } + $pendingReleaseFiles = array_values($context->pendingReleaseFiles); + $pendingParHashes = array_values($context->pendingParHashes); - Search::updateRelease((int) $context->release->id); + $insertedReleaseFiles = DB::transaction(function () use ( + $context, + $pendingReleaseFiles, + $pendingParHashes, + $updateRows, + $processPasswords, + $passwordStatus, + ): int { + $inserted = $pendingReleaseFiles === [] + ? 0 + : ReleaseFile::query()->insertOrIgnore($pendingReleaseFiles); + + if ($pendingParHashes !== []) { + ParHash::query()->insertOrIgnore($pendingParHashes); + } + + $releaseFilesCount = ReleaseFile::whereReleasesId($context->release->id)->count('releases_id') ?? 0; + + if (! $context->releaseHasPassword && $context->nzbHasCompressedFile && $releaseFilesCount === 0) { + Release::query()->where('id', $context->release->id)->update($updateRows); + } else { + $updateRows['passwordstatus'] = $processPasswords ? $passwordStatus : ReleaseBrowseService::PASSWD_NONE; + $updateRows['rarinnerfilecount'] = $releaseFilesCount; + Release::query()->where('id', $context->release->id)->update($updateRows); + } + + return $inserted; + }, 3); + + $context->pendingReleaseFiles = []; + $context->pendingParHashes = []; + $context->releaseFilesChanged = $context->releaseFilesChanged || $insertedReleaseFiles > 0; + + $this->searchSyncCoordinator->request((int) $context->release->id); } /** @@ -283,7 +315,7 @@ class ReleaseFileManager 'passwordstatus' => ReleaseBrowseService::PASSWD_NONE, ], AdditionalCandidateQuery::claimResetValues())); - Search::updateRelease((int) $release->id); + $this->searchSyncCoordinator->request((int) $release->id); Log::warning('Release '.$release->id.' skipped after post-processing timeout ('.$newCount.'/'.$maxTimeoutCount.')'); @@ -302,6 +334,7 @@ class ReleaseFileManager $id = (int) $release->id; $guid = $release->guid ?? ''; + $this->searchSyncCoordinator->discard($id); // Delete NZB file try { @@ -352,7 +385,7 @@ class ReleaseFileManager Release::where('id', $id)->delete(); } catch (\Throwable) { } - } catch (\Throwable) { // @phpstan-ignore catch.neverThrown + } catch (\Throwable) { // Last resort: swallow any exception } } @@ -693,20 +726,6 @@ class ReleaseFileManager return true; } - private function flushQueuedReleaseFiles(ReleaseProcessingContext $context): void - { - if ($context->pendingReleaseFiles !== []) { - $inserted = ReleaseFile::query()->insertOrIgnore(array_values($context->pendingReleaseFiles)); - $context->releaseFilesChanged = $context->releaseFilesChanged || $inserted > 0; - $context->pendingReleaseFiles = []; - } - - if ($context->pendingParHashes !== []) { - ParHash::query()->insertOrIgnore(array_values($context->pendingParHashes)); - $context->pendingParHashes = []; - } - } - private function loadExistingReleaseFileNames(ReleaseProcessingContext $context): void { if ($context->existingReleaseFileNames !== null) { diff --git a/app/Services/AdditionalProcessing/ReleaseFilesArchiveFallback.php b/app/Services/AdditionalProcessing/ReleaseFilesArchiveFallback.php index 654328a47..0b83e35c4 100644 --- a/app/Services/AdditionalProcessing/ReleaseFilesArchiveFallback.php +++ b/app/Services/AdditionalProcessing/ReleaseFilesArchiveFallback.php @@ -5,7 +5,7 @@ declare(strict_types=1); namespace App\Services\AdditionalProcessing; use App\Models\ReleaseFile; -use App\Services\AdditionalProcessing\Config\ProcessingConfiguration; +use App\Services\AdditionalProcessing\DTO\ArchiveCandidate; use App\Services\AdditionalProcessing\Enums\DownloadKind; use App\Services\AdditionalProcessing\State\ReleaseProcessingContext; use Illuminate\Database\Eloquent\Collection as EloquentCollection; @@ -17,7 +17,6 @@ use Illuminate\Support\Facades\File; class ReleaseFilesArchiveFallback { public function __construct( - private readonly ProcessingConfiguration $config, private readonly ArchiveExtractionService $archiveService, private readonly MediaExtractionService $mediaService, private readonly UsenetDownloadService $downloadService, @@ -26,7 +25,7 @@ class ReleaseFilesArchiveFallback ) {} /** - * Try JPG/PNG files listed inside an already-downloaded archive summary. + * Try JPG/PNG/WebP files listed inside an already-downloaded archive summary. * * @param array> $files */ @@ -37,7 +36,7 @@ class ReleaseFilesArchiveFallback ): void { $imageFiles = array_values(array_filter( $files, - static fn (array $file): bool => preg_match('/\.(jpe?g|png)$/i', (string) ($file['name'] ?? '')) === 1 + static fn (array $file): bool => preg_match('/\.(jpe?g|png|webp)$/i', (string) ($file['name'] ?? '')) === 1 )); if ($imageFiles === []) { @@ -69,7 +68,7 @@ class ReleaseFilesArchiveFallback } /** - * Re-download archives and extract stored JPG/PNG candidates. + * Re-download archives and extract stored JPG/PNG/WebP candidates. */ public function processJpgFromReleaseFiles(ReleaseProcessingContext $context): void { @@ -78,7 +77,8 @@ class ReleaseFilesArchiveFallback ->where(function ($query) { $query->where('name', 'like', '%.jpg') ->orWhere('name', 'like', '%.jpeg') - ->orWhere('name', 'like', '%.png'); + ->orWhere('name', 'like', '%.png') + ->orWhere('name', 'like', '%.webp'); }) ->orderByDesc('size') ->limit(3) @@ -144,6 +144,45 @@ class ReleaseFilesArchiveFallback ); } + /** + * Extract NFO candidates from archives already downloaded for this release. + */ + public function processNfoFromDownloadedArchives(ReleaseProcessingContext $context): void + { + if ($context->downloadedArchives === [] || ! $context->releaseHasNoNFO) { + return; + } + + $nntp = $this->downloadService->getNNTP(); + + foreach ($context->downloadedArchives as $archive) { + $nfoFiles = array_values(array_filter( + $this->archiveService->sortFilesWithNfoPriority($archive->files), + fn (array $file): bool => $this->archiveService->isNfoFile((string) ($file['name'] ?? '')), + )); + + $this->processExtractedCandidatesFromData( + $archive->data, + array_slice($nfoFiles, 0, 3), + $context, + 'downloaded_nfo_', + function (string $tempPath, ReleaseProcessingContext $context) use ($nntp): bool { + if (! $this->releaseManager->processNfoFile($tempPath, $context, $nntp)) { + return false; + } + + $this->output->echoNfoFound(); + + return true; + }, + ); + + if (! $context->releaseHasNoNFO) { + return; + } + } + } + /** * @param EloquentCollection $candidates * @param callable(ReleaseProcessingContext): bool $shouldStop @@ -156,34 +195,22 @@ class ReleaseFilesArchiveFallback callable $shouldStop, callable $handler ): void { - if ($candidates->isEmpty() || $context->nzbContents === []) { + $archiveCandidates = $this->archiveCandidates($context); + if ($candidates->isEmpty() || $archiveCandidates === []) { return; } - foreach ($context->nzbContents as $nzbFile) { + foreach ($archiveCandidates as $archiveCandidate) { if ($shouldStop($context)) { break; } - $title = (string) ($nzbFile['title'] ?? ''); - if (preg_match( - '/(\.(part0*1|rar|zip))(\s*\.rar)*($|[ ")]|-])|"[a-f0-9]{32}\.[1-9]\d{1,2}".*\(\d+\/\d{2,}\)$/i', - $title - ) !== 1) { - continue; - } - - $messageIDs = $this->extractInitialSegments($nzbFile['segments'] ?? []); - if ($messageIDs === []) { - continue; - } - $result = $this->downloadService->download( DownloadKind::Compressed, - $messageIDs, + $archiveCandidate->messageIds, $context->releaseGroupName, $context->release->id, - $title + $archiveCandidate->title, ); if ($result['groupUnavailable']) { @@ -197,12 +224,19 @@ class ReleaseFilesArchiveFallback continue; } + $filenames = $candidates + ->pluck('name') + ->filter(static fn (mixed $filename): bool => is_string($filename) && $filename !== '') + ->values() + ->all(); + $extractedFiles = $this->extractCandidateFiles( + $result['data'], + $filenames, + $context->tmpPath, + ); + foreach ($candidates as $candidate) { - $fileData = $this->archiveService->extractSpecificFile( - $result['data'], - $candidate->name, - $context->tmpPath - ); + $fileData = $extractedFiles[$candidate->name] ?? null; if ($fileData === null || $fileData === '') { continue; @@ -226,13 +260,23 @@ class ReleaseFilesArchiveFallback string $tempPrefix, callable $handler ): void { + $filenames = array_values(array_filter(array_map( + static fn (array $candidate): string => (string) ($candidate['name'] ?? ''), + $candidates, + ))); + $extractedFiles = $this->extractCandidateFiles( + $compressedData, + $filenames, + $context->tmpPath, + ); + foreach ($candidates as $candidate) { $filename = (string) ($candidate['name'] ?? ''); if ($filename === '') { continue; } - $fileData = $this->archiveService->extractSpecificFile($compressedData, $filename, $context->tmpPath); + $fileData = $extractedFiles[$filename] ?? null; if ($fileData === null || $fileData === '') { continue; } @@ -243,6 +287,30 @@ class ReleaseFilesArchiveFallback } } + /** + * @return list + */ + private function archiveCandidates(ReleaseProcessingContext $context): array + { + if ($context->workPlan === null) { + return []; + } + + return array_values(array_filter( + $context->workPlan->archiveCandidates, + static fn (ArchiveCandidate $candidate): bool => $candidate->likelyFirstVolume, + )); + } + + /** + * @param list $filenames + * @return array + */ + private function extractCandidateFiles(string $archiveData, array $filenames, string $tmpPath): array + { + return $this->archiveService->extractSpecificFiles($archiveData, $filenames, $tmpPath); + } + /** * @param callable(string, ReleaseProcessingContext): bool $handler */ @@ -264,20 +332,4 @@ class ReleaseFilesArchiveFallback File::delete($tempPath); } } - - /** - * @param array $segments - * @return list - */ - private function extractInitialSegments(array $segments): array - { - $messageIDs = []; - $segmentCount = min(count($segments), $this->config->maximumRarSegments); - - for ($index = 0; $index < $segmentCount; $index++) { - $messageIDs[] = (string) $segments[$index]; - } - - return $messageIDs; - } } diff --git a/app/Services/AdditionalProcessing/ReleaseProcessor.php b/app/Services/AdditionalProcessing/ReleaseProcessor.php index 69d334f5c..5a9dd8cd3 100644 --- a/app/Services/AdditionalProcessing/ReleaseProcessor.php +++ b/app/Services/AdditionalProcessing/ReleaseProcessor.php @@ -6,7 +6,12 @@ namespace App\Services\AdditionalProcessing; use App\Models\UsenetGroup; use App\Services\AdditionalProcessing\Config\ProcessingConfiguration; +use App\Services\AdditionalProcessing\DTO\ReleaseProcessingResult; use App\Services\AdditionalProcessing\Enums\DownloadKind; +use App\Services\AdditionalProcessing\Enums\ProcessingOutcome; +use App\Services\AdditionalProcessing\Enums\ProcessingStage; +use App\Services\AdditionalProcessing\State\PersistenceMetricsCollector; +use App\Services\AdditionalProcessing\State\ProcessingMetrics; use App\Services\AdditionalProcessing\State\ReleaseProcessingContext; use App\Services\Releases\ReleaseBrowseService; use App\Services\TempWorkspaceService; @@ -25,99 +30,203 @@ class ReleaseProcessor public function __construct( private readonly ProcessingConfiguration $config, private readonly NzbContentParser $nzbParser, + private readonly AdditionalWorkPlanner $workPlanner, private readonly ArchiveExtractionService $archiveService, private readonly MediaExtractionService $mediaService, private readonly UsenetDownloadService $downloadService, private readonly ReleaseFileManager $releaseManager, private readonly ReleaseFilesArchiveFallback $archiveFallback, private readonly TempWorkspaceService $tempWorkspace, - private readonly ConsoleOutputService $output + private readonly ConsoleOutputService $output, + private readonly ?ReleaseSearchSyncCoordinator $searchSyncCoordinator = null, + private readonly ?PersistenceMetricsCollector $persistenceMetricsCollector = null, ) {} - public function process(ReleaseProcessingContext $context, string $mainTmpPath): void + public function process(ReleaseProcessingContext $context, string $mainTmpPath): ReleaseProcessingResult { + $metrics = new ProcessingMetrics; + $releaseId = (int) $context->release->id; + $persistenceMetricsCollector = $this->persistenceMetricsCollector ?? new PersistenceMetricsCollector; + $searchSyncCoordinator = $this->searchSyncCoordinator + ?? new ReleaseSearchSyncCoordinator( + $persistenceMetricsCollector, + ); + + $persistenceMetricsCollector->beginReleaseScope($releaseId); + $searchSyncCoordinator->beginReleaseScope($releaseId); + $this->downloadService->beginReleaseScope(); + + try { + $result = $this->processRelease($context, $mainTmpPath, $metrics); + } finally { + $context->releaseDownloadedArchives(); + try { + $searchSyncCoordinator->finishReleaseScope(); + } finally { + $persistenceMetrics = $persistenceMetricsCollector->finishReleaseScope(); + $downloadMetrics = $this->downloadService->finishReleaseScope(); + } + } + + return $result->withPerformance( + $metrics->elapsedSeconds(), + $metrics->stageDurations(), + $downloadMetrics, + $persistenceMetrics, + ); + } + + private function processRelease( + ReleaseProcessingContext $context, + string $mainTmpPath, + ProcessingMetrics $metrics, + ): ReleaseProcessingResult { $release = $context->release; $this->output->echoReleaseStart($release->id, $release->size); $this->output->setProcessTitle((int) $release->id); try { - $context->tmpPath = $this->tempWorkspace->createReleaseTempFolder($mainTmpPath, $release->guid); + $context->tmpPath = $metrics->measure( + ProcessingStage::WorkspacePreparation, + fn (): string => $this->tempWorkspace->createReleaseTempFolder($mainTmpPath, $release->guid), + ); } catch (\Throwable $e) { $this->output->warning('Unable to prepare release temp directory: '.$e->getMessage()); - return; + return $this->result( + $context, + ProcessingOutcome::TemporaryWorkspaceUnavailable, + reason: $e->getMessage(), + ); } try { - $nzbResult = $this->nzbParser->parseNzb($release->guid); + $releaseNameChanged = false; + $releaseNeededNfo = false; + $nzbResult = $metrics->measure( + ProcessingStage::NzbParsing, + fn (): array => $this->nzbParser->parseNzb($release->guid), + ); if ($nzbResult['error'] !== null) { $this->output->warning($nzbResult['error']); $this->releaseManager->deleteRelease($release); - return; + return $this->result( + $context, + ProcessingOutcome::DeletedBrokenNzb, + reason: $nzbResult['error'], + ); } $context->nzbContents = array_values($nzbResult['contents']); - if ($this->checkReleaseTimeout($context)) { - return; + if (($timeoutResult = $this->processingTimeoutResult($context, $metrics)) !== null) { + return $timeoutResult; } - $this->initializeContext($context); - $this->releaseManager->processReleaseNameFromNzbContents($context->nzbContents, $context); - $messageIds = $this->nzbParser->extractMessageIDs($context->nzbContents, $context->releaseGroupName, $this->config); + $releaseNameChanged = $metrics->measure( + ProcessingStage::ReleaseInitialization, + function () use ($context): bool { + $this->initializeContext($context); - $context->nzbHasCompressedFile = $messageIds['hasCompressedFile']; - $context->sampleMessageIDs = $messageIds['sampleMessageIDs']; - $context->jpgMessageIDs = $messageIds['jpgMessageIDs']; - $context->mediaInfoMessageIDs = $messageIds['mediaInfoMessageID']; - $context->audioInfoMessageIDs = $messageIds['audioInfoMessageID']; - $context->audioInfoExtension = $messageIds['audioInfoExtension']; - - $bookFlood = $messageIds['bookFileCount'] > 80 - && ($messageIds['bookFileCount'] * 2) >= count($context->nzbContents); + return $this->releaseManager->processReleaseNameFromNzbContents($context->nzbContents, $context); + }, + ); + $releaseNeededNfo = $context->releaseHasNoNFO; + $bookFlood = $metrics->measure( + ProcessingStage::MessageIdSelection, + fn (): bool => $this->prepareMessageIds($context), + ); if ($this->shouldProcessDownloads()) { - $this->processMessageIdDownloads($context); - if ($this->checkReleaseTimeout($context)) { - return; + $metrics->measure( + ProcessingStage::DirectDownloads, + fn () => $this->processMessageIdDownloads($context), + ); + if (($timeoutResult = $this->processingTimeoutResult($context, $metrics)) !== null) { + return $timeoutResult; } if (! $bookFlood && $context->nzbHasCompressedFile) { $triedCompressedMids = []; - $this->processNzbCompressedFiles($context, false, $triedCompressedMids); - if ($this->checkReleaseTimeout($context)) { - return; + $metrics->measure( + ProcessingStage::ArchiveDownloads, + function () use ($context, &$triedCompressedMids): void { + $this->processNzbCompressedFiles($context, false, $triedCompressedMids); + }, + ); + if (($timeoutResult = $this->processingTimeoutResult($context, $metrics)) !== null) { + return $timeoutResult; } if ($this->config->fetchLastFiles) { - $this->processNzbCompressedFiles($context, true, $triedCompressedMids); - if ($this->checkReleaseTimeout($context)) { - return; + $metrics->measure( + ProcessingStage::ArchiveDownloads, + function () use ($context, &$triedCompressedMids): void { + $this->processNzbCompressedFiles($context, true, $triedCompressedMids); + }, + ); + if (($timeoutResult = $this->processingTimeoutResult($context, $metrics)) !== null) { + return $timeoutResult; } } if (! $context->releaseHasPassword) { - $this->processExtractedFiles($context); - if ($this->checkReleaseTimeout($context)) { - return; + $metrics->measure( + ProcessingStage::ExtractedFiles, + fn () => $this->processExtractedFiles($context), + ); + if (($timeoutResult = $this->processingTimeoutResult($context, $metrics)) !== null) { + return $timeoutResult; } } } - if (! $context->foundJPGSample && $this->config->processJPGSample) { - $this->archiveFallback->processJpgFromReleaseFiles($context); - } + $metrics->measure(ProcessingStage::ArchiveFallbacks, function () use ($context): void { + if (! $context->foundJPGSample && $this->config->processJPGSample) { + $this->archiveFallback->processJpgFromReleaseFiles($context); + } - if ($context->releaseHasNoNFO) { - $this->archiveFallback->processNfoFromReleaseFiles($context); - } + if ($context->releaseHasNoNFO) { + $this->archiveFallback->processNfoFromDownloadedArchives($context); + } + + if ($context->releaseHasNoNFO) { + $this->archiveFallback->processNfoFromReleaseFiles($context); + } + }); } - $this->releaseManager->finalizeRelease($context, $this->config->processPasswords); + $metrics->measure( + ProcessingStage::Finalization, + fn () => $this->releaseManager->finalizeRelease($context, $this->config->processPasswords), + ); + + $artifactsCreated = $releaseNameChanged || $this->createdArtifacts($context, $releaseNeededNfo); + $outcome = match (true) { + $context->releaseHasPassword => ProcessingOutcome::Passworded, + $context->groupUnavailable => ProcessingOutcome::GroupUnavailable, + ! $artifactsCreated => ProcessingOutcome::NoUsefulArtifacts, + default => ProcessingOutcome::Completed, + }; + + return $this->result( + $context, + $outcome, + $artifactsCreated, + reason: match ($outcome) { + ProcessingOutcome::Passworded => 'Password protection was detected.', + ProcessingOutcome::GroupUnavailable => 'The release group was unavailable during processing.', + ProcessingOutcome::NoUsefulArtifacts => 'Processing completed without creating useful artifacts.', + default => '', + }, + ); } finally { if ($context->tmpPath !== '') { - $this->tempWorkspace->clearDirectory($context->tmpPath, false); + $metrics->measure( + ProcessingStage::WorkspaceCleanup, + fn () => $this->tempWorkspace->clearDirectory($context->tmpPath, false), + ); } } } @@ -132,6 +241,23 @@ class ReleaseProcessor || $this->config->processJPGSample; } + private function prepareMessageIds(ReleaseProcessingContext $context): bool + { + $workPlan = $this->workPlanner->plan( + $context->nzbContents, + $context->releaseGroupName, + ); + $context->workPlan = $workPlan; + $context->nzbHasCompressedFile = $workPlan->hasCompressedFile(); + $context->sampleMessageIDs = $workPlan->sampleMessageIds; + $context->jpgMessageIDs = $workPlan->jpgMessageIds; + $context->mediaInfoMessageIDs = $workPlan->mediaInfoMessageId; + $context->audioInfoMessageIDs = $workPlan->audioInfoMessageId; + $context->audioInfoExtension = $workPlan->audioInfoExtension; + + return $workPlan->bookFlood; + } + private function initializeContext(ReleaseProcessingContext $context): void { $context->initializeFromConfig( @@ -159,15 +285,20 @@ class ReleaseProcessor $context->resetCounters(); } - private function checkReleaseTimeout(ReleaseProcessingContext $context): bool - { + private function processingTimeoutResult( + ReleaseProcessingContext $context, + ProcessingMetrics $metrics, + ): ?ReleaseProcessingResult { if (! $context->isTimedOut($this->config->releaseProcessingTimeout)) { - return false; + return null; } - $deleted = $this->releaseManager->handleReleaseTimeout( - $context->release, - $this->config->maxPpTimeoutCount + $deleted = $metrics->measure( + ProcessingStage::TimeoutHandling, + fn (): bool => $this->releaseManager->handleReleaseTimeout( + $context->release, + $this->config->maxPpTimeoutCount, + ), ); if ($deleted) { @@ -183,7 +314,44 @@ class ReleaseProcessor $this->tempWorkspace->clearDirectory($context->tmpPath, false); } - return true; + return $this->result( + $context, + $deleted ? ProcessingOutcome::DeletedAfterTimeout : ProcessingOutcome::TimedOut, + reason: $deleted + ? 'The release was deleted after reaching the post-processing timeout limit.' + : 'The release exceeded the post-processing timeout.', + ); + } + + private function createdArtifacts(ReleaseProcessingContext $context, bool $releaseNeededNfo): bool + { + return $context->releaseFilesChanged + || $context->foundPAR2Info + || ($releaseNeededNfo && ! $context->releaseHasNoNFO) + || ($this->config->processVideo && $context->foundVideo) + || ($this->config->processThumbnails && $context->foundSample) + || ($this->config->processJPGSample && $context->foundJPGSample) + || ($this->config->processMediaInfo && $context->foundMediaInfo) + || ($this->config->processAudioInfo && $context->foundAudioInfo) + || ($this->config->processAudioSample && $context->foundAudioSample); + } + + private function result( + ReleaseProcessingContext $context, + ProcessingOutcome $outcome, + bool $artifactsCreated = false, + string $reason = '', + ): ReleaseProcessingResult { + return new ReleaseProcessingResult( + releaseId: (int) $context->release->id, + guid: (string) $context->release->guid, + outcome: $outcome, + artifactsCreated: $artifactsCreated, + releaseFilesAdded: $context->addedFileInfo, + reason: $reason, + duplicateMessageIdCount: $context->duplicateMessageIdCount(), + unsupportedReasons: $context->unsupportedReasons(), + ); } private function processMessageIdDownloads(ReleaseProcessingContext $context): void @@ -318,61 +486,39 @@ class ReleaseProcessor return; } - $nzbContents = $context->nzbContents; - if ($reverse) { - krsort($nzbContents); - } else { + $archiveCandidates = match (true) { + $context->workPlan === null => [], + $reverse => $context->workPlan->orderedArchiveCandidates(true), + default => $context->workPlan->prioritizedArchiveCandidates(), + }; + if (! $reverse) { $triedCompressedMids = []; } $failed = $downloaded = 0; - foreach ($nzbContents as $nzbFile) { + foreach ($archiveCandidates as $archiveCandidate) { if ($downloaded >= $this->config->maximumRarSegments || $failed >= $this->config->maximumRarPasswordChecks || $context->releaseHasPassword - || $context->groupUnavailable ) { break; } - $title = (string) ($nzbFile['title'] ?? ''); - if (preg_match( - '/(\.(part\d+|[rz]\d+|rar|0+|0*10?|zipr\d{2,3}|zipx?)(\s*\.rar)*($|[ ")]|-])|"[a-f0-9]{32}\.[1-9]\d{1,2}".*\(\d+\/\d{2,}\)$)/i', - $title - ) !== 1) { + if ($reverse && array_intersect($archiveCandidate->messageIds, $triedCompressedMids) !== []) { continue; } - $segments = $nzbFile['segments'] ?? []; - $messageIDs = []; - $segmentCount = count($segments) - 1; - - foreach (range(0, $this->config->maximumRarSegments - 1) as $index) { - if ($index > $segmentCount) { - break; - } - - $segment = (string) $segments[$index]; - if (! $reverse) { - $triedCompressedMids[] = $segment; - } elseif (in_array($segment, $triedCompressedMids, false)) { - continue 2; - } - - $messageIDs[] = $segment; - } - - if ($messageIDs === []) { - continue; + if (! $reverse) { + $triedCompressedMids = [...$triedCompressedMids, ...$archiveCandidate->messageIds]; } $result = $this->downloadService->download( DownloadKind::Compressed, - $messageIDs, + $archiveCandidate->messageIds, $context->releaseGroupName, $context->release->id, - $title + $archiveCandidate->title, ); if ($result['groupUnavailable']) { @@ -385,8 +531,13 @@ class ReleaseProcessor $this->output->echoCompressedDownload(); $downloaded++; - $processed = $this->processCompressedData($result['data'], $context, $reverse); - if ($processed || $context->releaseHasPassword) { + $processed = $this->processCompressedData( + $result['data'], + $context, + $reverse, + $archiveCandidate->title, + ); + if ($processed) { break; } } else { @@ -399,7 +550,8 @@ class ReleaseProcessor private function processCompressedData( string $compressedData, ReleaseProcessingContext $context, - bool $reverse + bool $reverse, + string $archiveTitle = '', ): bool { $result = $this->archiveService->processCompressedData( $compressedData, @@ -411,7 +563,7 @@ class ReleaseProcessor $context->releaseHasPassword = true; $context->passwordStatus = $result['passwordStatus']; - return false; + return true; } if (isset($result['standaloneVideoType'])) { @@ -457,6 +609,13 @@ class ReleaseProcessor } } + if ($context->releaseHasNoNFO + && is_array($result['files']) + && $this->containsNfoCandidate($result['files']) + ) { + $context->rememberDownloadedArchive($archiveTitle, $compressedData, $result['files']); + } + if (! $context->foundJPGSample && $this->config->processJPGSample) { $this->archiveFallback->processJpgFromArchiveFileList($compressedData, $result['files'], $context); } @@ -464,6 +623,20 @@ class ReleaseProcessor return $context->totalFileInfo > 0; } + /** + * @param array> $files + */ + private function containsNfoCandidate(array $files): bool + { + foreach ($files as $file) { + if ($this->archiveService->isNfoFile((string) ($file['name'] ?? ''))) { + return true; + } + } + + return false; + } + private function processExtractedFiles(ReleaseProcessingContext $context): void { $nestedLevels = 0; @@ -490,7 +663,7 @@ class ReleaseProcessor $rarData = @File::get($filePath); if (! empty($rarData)) { - $this->processCompressedData($rarData, $context, false); + $this->processCompressedData($rarData, $context, false, basename($filePath)); $foundCompressed = true; } File::delete($filePath); @@ -565,7 +738,7 @@ class ReleaseProcessor continue; } - if (! $context->foundJPGSample && preg_match('/\.(jpe?g|png)$/i', $filePath) === 1) { + if (! $context->foundJPGSample && preg_match('/\.(jpe?g|png|webp)$/i', $filePath) === 1) { if ($this->mediaService->getJPGSample($filePath, $context->release->guid)) { $context->markFound('jpgSample'); $this->output->echoJpgSaved(); @@ -588,7 +761,7 @@ class ReleaseProcessor continue; } - if (! $context->foundJPGSample && preg_match('/^JPE?G|^PNG/i', $output) === 1) { + if (! $context->foundJPGSample && preg_match('/^JPE?G|^PNG|^WebP/i', $output) === 1) { if ($this->mediaService->getJPGSample($filePath, $context->release->guid)) { $context->markFound('jpgSample'); $this->output->echoJpgSaved(); diff --git a/app/Services/AdditionalProcessing/ReleaseSearchSyncCoordinator.php b/app/Services/AdditionalProcessing/ReleaseSearchSyncCoordinator.php new file mode 100644 index 000000000..04fd4cdf4 --- /dev/null +++ b/app/Services/AdditionalProcessing/ReleaseSearchSyncCoordinator.php @@ -0,0 +1,84 @@ +synchronize = $synchronize ?? static function (int $releaseId): void { + Search::updateRelease($releaseId); + }; + } + + public function beginReleaseScope(int $releaseId): void + { + if (! $this->coalesce) { + return; + } + + $this->releaseId = $releaseId; + $this->pending = false; + } + + public function request(int $releaseId): void + { + $this->metrics->recordSearchSyncRequest($releaseId); + + if ($this->coalesce && $this->releaseId === $releaseId) { + $this->pending = true; + + return; + } + + $this->execute($releaseId); + } + + public function discard(int $releaseId): void + { + if ($this->coalesce && $this->releaseId === $releaseId) { + $this->pending = false; + } + } + + public function finishReleaseScope(): void + { + if (! $this->coalesce) { + return; + } + + $releaseId = $this->releaseId; + $pending = $this->pending; + $this->releaseId = null; + $this->pending = false; + + if ($releaseId !== null && $pending) { + $this->execute($releaseId); + } + } + + private function execute(int $releaseId): void + { + $this->metrics->recordSearchSyncExecution($releaseId); + ($this->synchronize)($releaseId); + } +} diff --git a/app/Services/AdditionalProcessing/State/PersistenceMetricsCollector.php b/app/Services/AdditionalProcessing/State/PersistenceMetricsCollector.php new file mode 100644 index 000000000..147a21936 --- /dev/null +++ b/app/Services/AdditionalProcessing/State/PersistenceMetricsCollector.php @@ -0,0 +1,71 @@ +releaseId = $releaseId; + $this->databaseStatements = 0; + $this->databaseMilliseconds = 0.0; + $this->searchSyncRequests = 0; + $this->searchSyncExecutions = 0; + } + + public function recordDatabaseStatement(float $milliseconds): void + { + if ($this->releaseId === null) { + return; + } + + $this->databaseStatements++; + $this->databaseMilliseconds += $milliseconds; + } + + public function recordSearchSyncRequest(int $releaseId): void + { + if ($this->releaseId === $releaseId) { + $this->searchSyncRequests++; + } + } + + public function recordSearchSyncExecution(int $releaseId): void + { + if ($this->releaseId === $releaseId) { + $this->searchSyncExecutions++; + } + } + + public function finishReleaseScope(): PersistenceMetrics + { + $metrics = new PersistenceMetrics( + databaseStatements: $this->databaseStatements, + databaseMilliseconds: $this->databaseMilliseconds, + searchSyncRequests: $this->searchSyncRequests, + searchSyncExecutions: $this->searchSyncExecutions, + ); + + $this->releaseId = null; + $this->databaseStatements = 0; + $this->databaseMilliseconds = 0.0; + $this->searchSyncRequests = 0; + $this->searchSyncExecutions = 0; + + return $metrics; + } +} diff --git a/app/Services/AdditionalProcessing/State/ProcessingMetrics.php b/app/Services/AdditionalProcessing/State/ProcessingMetrics.php new file mode 100644 index 000000000..1d647e2d3 --- /dev/null +++ b/app/Services/AdditionalProcessing/State/ProcessingMetrics.php @@ -0,0 +1,59 @@ + + */ + private array $stageNanoseconds = []; + + public function __construct() + { + $this->startedAtNanoseconds = hrtime(true); + } + + /** + * @template TValue + * + * @param Closure(): TValue $operation + * @return TValue + */ + public function measure(ProcessingStage $stage, Closure $operation): mixed + { + $startedAt = hrtime(true); + + try { + return $operation(); + } finally { + $elapsed = hrtime(true) - $startedAt; + $this->stageNanoseconds[$stage->value] = ($this->stageNanoseconds[$stage->value] ?? 0) + $elapsed; + } + } + + public function elapsedSeconds(): float + { + return $this->nanosecondsToSeconds(hrtime(true) - $this->startedAtNanoseconds); + } + + /** + * @return array + */ + public function stageDurations(): array + { + return array_map($this->nanosecondsToSeconds(...), $this->stageNanoseconds); + } + + private function nanosecondsToSeconds(int $nanoseconds): float + { + return $nanoseconds / 1_000_000_000; + } +} diff --git a/app/Services/AdditionalProcessing/State/ReleaseProcessingContext.php b/app/Services/AdditionalProcessing/State/ReleaseProcessingContext.php index 0cbe0f147..e4d3d525c 100644 --- a/app/Services/AdditionalProcessing/State/ReleaseProcessingContext.php +++ b/app/Services/AdditionalProcessing/State/ReleaseProcessingContext.php @@ -5,6 +5,8 @@ declare(strict_types=1); namespace App\Services\AdditionalProcessing\State; use App\Models\Release; +use App\Services\AdditionalProcessing\DTO\AdditionalWorkPlan; +use App\Services\AdditionalProcessing\DTO\DownloadedArchive; /** * Mutable context object that holds the processing state for a single release. @@ -17,6 +19,10 @@ use App\Models\Release; */ class ReleaseProcessingContext { + private const int MAX_RETAINED_ARCHIVES = 2; + + private const int MAX_RETAINED_ARCHIVE_BYTES = 16_777_216; + // The release being processed public Release $release; @@ -56,6 +62,15 @@ class ReleaseProcessingContext */ public array $nzbContents = []; + public ?AdditionalWorkPlan $workPlan = null; + + /** + * @var list + */ + public array $downloadedArchives = []; + + private int $retainedArchiveBytes = 0; + // Message IDs for downloading /** * @var list @@ -177,10 +192,53 @@ class ReleaseProcessingContext $this->releaseHasPassword = false; $this->nzbHasCompressedFile = false; $this->groupUnavailable = false; + $this->workPlan = null; + $this->releaseDownloadedArchives(); $this->resetMessageIDs(); $this->resetCounters(); } + /** + * Retain only small, immediately reusable archive payloads for this release. + * + * @param array> $files + */ + public function rememberDownloadedArchive(string $title, string $data, array $files): bool + { + $byteSize = strlen($data); + if ($byteSize === 0 + || count($this->downloadedArchives) >= self::MAX_RETAINED_ARCHIVES + || ($this->retainedArchiveBytes + $byteSize) > self::MAX_RETAINED_ARCHIVE_BYTES + ) { + return false; + } + + $archive = new DownloadedArchive($title, $data, $files); + $this->downloadedArchives[] = $archive; + $this->retainedArchiveBytes += $archive->byteSize(); + + return true; + } + + public function releaseDownloadedArchives(): void + { + $this->downloadedArchives = []; + $this->retainedArchiveBytes = 0; + } + + public function duplicateMessageIdCount(): int + { + return $this->workPlan === null ? 0 : $this->workPlan->duplicateMessageIdCount; + } + + /** + * @return list + */ + public function unsupportedReasons(): array + { + return $this->workPlan === null ? [] : $this->workPlan->unsupportedReasons; + } + /** * Mark a processing artifact as found. */ diff --git a/app/Services/AdditionalProcessing/UsenetDownloadService.php b/app/Services/AdditionalProcessing/UsenetDownloadService.php index 2321bb7c1..e621b742a 100644 --- a/app/Services/AdditionalProcessing/UsenetDownloadService.php +++ b/app/Services/AdditionalProcessing/UsenetDownloadService.php @@ -5,6 +5,7 @@ declare(strict_types=1); namespace App\Services\AdditionalProcessing; use App\Services\AdditionalProcessing\Config\ProcessingConfiguration; +use App\Services\AdditionalProcessing\DTO\DownloadMetrics; use App\Services\AdditionalProcessing\Enums\DownloadKind; use App\Services\NNTP\NNTPService; use Exception; @@ -16,8 +17,33 @@ use Illuminate\Support\Facades\Log; */ class UsenetDownloadService { + private const int MAX_CACHE_ENTRIES = 4; + + private const int MAX_CACHE_ENTRY_BYTES = 8_388_608; + + private const int MAX_CACHE_BYTES = 16_777_216; + private NNTPService $nntp; + private bool $releaseScopeActive = false; + + /** + * @var array + */ + private array $releaseCache = []; + + private int $cachedBytes = 0; + + private int $logicalRequests = 0; + + private int $networkRequests = 0; + + private int $cacheHits = 0; + + private int $bytesDownloaded = 0; + + private int $bytesReused = 0; + public function __construct( private readonly ProcessingConfiguration $config, ?NNTPService $nntp = null @@ -25,10 +51,31 @@ class UsenetDownloadService $this->nntp = $nntp ?? new NNTPService; } + public function beginReleaseScope(): void + { + $this->clearReleaseScope(); + $this->releaseScopeActive = true; + } + + public function finishReleaseScope(): DownloadMetrics + { + $metrics = new DownloadMetrics( + logicalRequests: $this->logicalRequests, + networkRequests: $this->networkRequests, + cacheHits: $this->cacheHits, + bytesDownloaded: $this->bytesDownloaded, + bytesReused: $this->bytesReused, + ); + + $this->clearReleaseScope(); + + return $metrics; + } + /** * Download binary content from usenet using message IDs. * - * @param array|string $messageIDs Single or array of message IDs + * @param array|string $messageIDs Single or array of message IDs * @param string $groupName Group name for logging * @param int|null $releaseId Release ID for logging * @return array{success: bool, data: string|null, groupUnavailable: bool, error: string|null} @@ -61,7 +108,8 @@ class UsenetDownloadService if ($this->config->debugMode) { Log::debug('Attempting NNTP fetch', [ 'release_id' => $releaseId, - 'message_ids' => $messageIDs, + 'message_id_count' => count($messageIDs), + 'request_fingerprint' => $this->downloadFingerprint($messageIDs, $groupName), 'group' => $groupName, ]); } @@ -89,7 +137,8 @@ class UsenetDownloadService if ($this->config->debugMode) { Log::debug('NNTP fetch failed', [ 'release_id' => $releaseId, - 'message_ids' => $messageIDs, + 'message_id_count' => count($messageIDs), + 'request_fingerprint' => $this->downloadFingerprint($messageIDs, $groupName), 'group' => $groupName, 'error_object' => is_object($binary) ? get_class($binary) : null, 'error_message' => $errorMessage, @@ -112,7 +161,7 @@ class UsenetDownloadService /** * Download content for a specific processing step. * - * @param array|string $messageIDs + * @param array|string $messageIDs * @return array{success: bool, data: string|null, groupUnavailable: bool, error: string|null} */ public function download( @@ -122,24 +171,51 @@ class UsenetDownloadService ?int $releaseId = null, ?string $fileTitle = null ): array { + $messageIdList = is_array($messageIDs) ? array_values($messageIDs) : [$messageIDs]; + $cacheKey = $this->downloadFingerprint($messageIdList, $groupName); + + if ($this->releaseScopeActive) { + $this->logicalRequests++; + + if (isset($this->releaseCache[$cacheKey])) { + $cached = $this->releaseCache[$cacheKey]; + $this->cacheHits++; + $this->bytesReused += is_string($cached['data']) ? strlen($cached['data']) : 0; + + return $cached; + } + } + if ($this->config->debugMode) { Log::debug('Attempting download', [ 'kind' => $kind->value, 'release_id' => $releaseId, 'file_title' => $fileTitle, - 'message_ids' => $messageIDs, + 'message_id_count' => count($messageIdList), + 'request_fingerprint' => $cacheKey, 'group' => $groupName, ]); } + if ($this->releaseScopeActive && $messageIdList !== [] && $messageIdList !== ['']) { + $this->networkRequests++; + } + $result = $this->downloadByMessageIDs($messageIDs, $groupName, $releaseId); + if ($this->releaseScopeActive && $result['success'] && is_string($result['data'])) { + $byteSize = strlen($result['data']); + $this->bytesDownloaded += $byteSize; + $this->rememberSuccessfulDownload($cacheKey, $result, $byteSize); + } + if (! $result['success'] && $this->config->debugMode) { Log::debug('Download failed', [ 'kind' => $kind->value, 'release_id' => $releaseId, 'file_title' => $fileTitle, - 'message_ids' => $messageIDs, + 'message_id_count' => count($messageIdList), + 'request_fingerprint' => $cacheKey, 'group' => $groupName, 'error' => $result['error'], ]); @@ -148,6 +224,46 @@ class UsenetDownloadService return $result; } + /** + * @param list $messageIds + */ + private function downloadFingerprint(array $messageIds, string $groupName): string + { + return hash('sha256', serialize([ + 'message_ids' => array_map(static fn (mixed $messageId): string => (string) $messageId, $messageIds), + 'group' => $groupName, + 'alternate' => $this->config->alternateNNTP, + ])); + } + + /** + * @param array{success: bool, data: string|null, groupUnavailable: bool, error: string|null} $result + */ + private function rememberSuccessfulDownload(string $cacheKey, array $result, int $byteSize): void + { + if ($byteSize > self::MAX_CACHE_ENTRY_BYTES + || count($this->releaseCache) >= self::MAX_CACHE_ENTRIES + || ($this->cachedBytes + $byteSize) > self::MAX_CACHE_BYTES + ) { + return; + } + + $this->releaseCache[$cacheKey] = $result; + $this->cachedBytes += $byteSize; + } + + private function clearReleaseScope(): void + { + $this->releaseScopeActive = false; + $this->releaseCache = []; + $this->cachedBytes = 0; + $this->logicalRequests = 0; + $this->networkRequests = 0; + $this->cacheHits = 0; + $this->bytesDownloaded = 0; + $this->bytesReused = 0; + } + /** * Get the NNTP client instance. */ diff --git a/app/Services/NameFixing/ReleaseUpdateService.php b/app/Services/NameFixing/ReleaseUpdateService.php index 9c94d51be..8be61c420 100644 --- a/app/Services/NameFixing/ReleaseUpdateService.php +++ b/app/Services/NameFixing/ReleaseUpdateService.php @@ -5,10 +5,11 @@ declare(strict_types=1); namespace App\Services\NameFixing; use App\Events\ReleaseNameFixed; -use App\Facades\Search; use App\Models\Category; use App\Models\Release; use App\Models\UsenetGroup; +use App\Services\AdditionalProcessing\ReleaseSearchSyncCoordinator; +use App\Services\AdditionalProcessing\State\PersistenceMetricsCollector; use App\Services\Categorization\CategorizationService; use App\Services\ReleaseCleaningService; use Illuminate\Support\Arr; @@ -82,6 +83,8 @@ class ReleaseUpdateService protected bool $echoOutput; + private readonly ReleaseSearchSyncCoordinator $searchSyncCoordinator; + /** * The release ID we are trying to rename. */ @@ -109,10 +112,13 @@ class ReleaseUpdateService public function __construct( ?CategorizationService $category = null, - ?FileNameCleaner $fileNameCleaner = null + ?FileNameCleaner $fileNameCleaner = null, + ?ReleaseSearchSyncCoordinator $searchSyncCoordinator = null, ) { $this->category = $category ?? new CategorizationService; $this->fileNameCleaner = $fileNameCleaner ?? new FileNameCleaner; + $this->searchSyncCoordinator = $searchSyncCoordinator + ?? new ReleaseSearchSyncCoordinator(new PersistenceMetricsCollector); $this->echoOutput = config('nntmux.echocli'); } @@ -331,8 +337,7 @@ class ReleaseUpdateService )); }); - // Update search index - Search::updateRelease($release->releases_id); + $this->searchSyncCoordinator->request((int) $release->releases_id); } /** diff --git a/app/Services/Runners/PostProcessRunner.php b/app/Services/Runners/PostProcessRunner.php index 7f11e8d01..5a1c02831 100644 --- a/app/Services/Runners/PostProcessRunner.php +++ b/app/Services/Runners/PostProcessRunner.php @@ -16,7 +16,7 @@ use Illuminate\Support\Facades\Log; class PostProcessRunner extends BaseRunner { - private const int ADDITIONAL_WORKER_MAX_BATCHES = 4; + public const int ADDITIONAL_WORKER_MAX_BATCHES = 4; private function guidBucketExpression(string $column = 'leftguid'): string { @@ -257,13 +257,13 @@ class PostProcessRunner extends BaseRunner $excludedReleaseIds = []; try { for ($batch = 0; $batch < self::ADDITIONAL_WORKER_MAX_BATCHES; $batch++) { - $stats = $orchestrator->start('', $char, $workerToken, $excludedReleaseIds); - if ($stats['claimed'] === 0) { + $result = $orchestrator->start('', $char, $workerToken, $excludedReleaseIds); + if ($result->claimedCount() === 0) { break; } $excludedReleaseIds = array_values(array_unique([ ...$excludedReleaseIds, - ...$stats['claimed_ids'], + ...$result->claimedIds, ])); } } finally { diff --git a/phpstan-baseline.neon b/phpstan-baseline.neon index 478244a27..180e17448 100644 --- a/phpstan-baseline.neon +++ b/phpstan-baseline.neon @@ -480,24 +480,12 @@ parameters: count: 1 path: app/Rules/RecaptchaRule.php - - - message: '#^Method App\\Services\\AdditionalProcessing\\ArchiveExtractionService\:\:sortFilesWithNfoPriority\(\) should return array\ but returns list\\.$#' - identifier: return.type - count: 1 - path: app/Services/AdditionalProcessing/ArchiveExtractionService.php - - message: '#^PHPDoc tag @return with type list\ is incompatible with native type void\.$#' identifier: return.phpDocType count: 1 path: app/Services/AdditionalProcessing/ArchiveExtractionService.php - - - message: '#^Method App\\Services\\AdditionalProcessing\\NzbContentParser\:\:extractSegments\(\) should return array\ but returns list\\.$#' - identifier: return.type - count: 1 - path: app/Services/AdditionalProcessing/NzbContentParser.php - - message: '#^Using nullsafe property access "\?\-\>id" on left side of \?\? is unnecessary\. Use \-\> instead\.$#' identifier: nullsafe.neverNull @@ -510,30 +498,6 @@ parameters: count: 2 path: app/Services/AdditionalProcessing/ReleaseFileManager.php - - - message: '#^Parameter \#2 \$messageIDs of method App\\Services\\AdditionalProcessing\\UsenetDownloadService\:\:download\(\) expects array\\|string, list\ given\.$#' - identifier: argument.type - count: 1 - path: app/Services/AdditionalProcessing/ReleaseFilesArchiveFallback.php - - - - message: '#^Parameter \#1 \$nzbContents of method App\\Services\\AdditionalProcessing\\NzbContentParser\:\:extractMessageIDs\(\) expects array\, list\\> given\.$#' - identifier: argument.type - count: 1 - path: app/Services/AdditionalProcessing/ReleaseProcessor.php - - - - message: '#^Parameter \#2 \$messageIDs of method App\\Services\\AdditionalProcessing\\UsenetDownloadService\:\:download\(\) expects array\\|string, list\ given\.$#' - identifier: argument.type - count: 3 - path: app/Services/AdditionalProcessing/ReleaseProcessor.php - - - - message: '#^Right side of \|\| is always false\.$#' - identifier: booleanOr.rightAlwaysFalse - count: 2 - path: app/Services/AdditionalProcessing/ReleaseProcessor.php - - message: '#^Strict comparison using \=\=\= between non\-empty\-string and null will always evaluate to false\.$#' identifier: identical.alwaysFalse diff --git a/tests/Feature/AdditionalProcessingNzbSplitRenameTest.php b/tests/Feature/AdditionalProcessingNzbSplitRenameTest.php index c34948152..8dc323c28 100644 --- a/tests/Feature/AdditionalProcessingNzbSplitRenameTest.php +++ b/tests/Feature/AdditionalProcessingNzbSplitRenameTest.php @@ -154,6 +154,25 @@ class AdditionalProcessingNzbSplitRenameTest extends TestCase $this->assertSame('MLB.2026.Cincinnati.Reds.vs.Tampa.Bay.Rays.20.04.720pEN60fps', $result); } + public function test_additional_nfo_detection_accepts_alternative_and_scene_names(): void + { + $manager = $this->makeReleaseFileManager(); + + foreach (['file_id.diz', 'readme.txt', 'info.txt', '00-scene.nfo', 'group-release.nfo'] as $filename) { + $this->assertTrue($manager->isNfoFilename($filename), $filename.' should be treated as NFO data'); + } + } + + public function test_additional_nfo_normalization_converts_utf16_and_removes_utf8_bom(): void + { + $manager = $this->makeReleaseFileManager(); + $normalize = new \ReflectionMethod($manager, 'normalizeNfoEncoding'); + $utf16 = "\xFF\xFE".mb_convert_encoding('Scene NFO', 'UTF-16LE', 'UTF-8'); + + $this->assertSame('Scene NFO', $normalize->invoke($manager, $utf16)); + $this->assertSame('Scene NFO', $normalize->invoke($manager, "\xEF\xBB\xBFScene NFO")); + } + private function makeReleaseFileManager(?ReleaseUpdateService $updateService = null): ReleaseFileManager { $updateService ??= $this->createMock(ReleaseUpdateService::class); diff --git a/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php b/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php index f924b64a8..3042cc767 100644 --- a/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php +++ b/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php @@ -7,13 +7,17 @@ namespace Tests\Feature; use App\Models\Release; use App\Services\AdditionalProcessing\AdditionalProcessingOrchestrator; use App\Services\AdditionalProcessing\ConsoleOutputService; +use App\Services\AdditionalProcessing\DTO\ReleaseProcessingResult; +use App\Services\AdditionalProcessing\Enums\ProcessingOutcome; use App\Services\AdditionalProcessing\ReleaseProcessor; use App\Services\AdditionalProcessing\State\ReleaseProcessingContext; use App\Services\TempWorkspaceService; use Illuminate\Contracts\Console\Kernel; use Illuminate\Database\Schema\Blueprint; use Illuminate\Support\Facades\DB; +use Illuminate\Support\Facades\Log; use Illuminate\Support\Facades\Schema; +use Mockery; use PDO; use Tests\TestCase; use Tests\Unit\AdditionalProcessing\CreatesProcessingConfiguration; @@ -87,6 +91,8 @@ class AdditionalProcessingOrchestratorClaimTest extends TestCase public function test_handled_release_exception_clears_claim(): void { + Log::spy(); + DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]); DB::table('releases')->insert([ 'id' => 1, @@ -121,15 +127,49 @@ class AdditionalProcessingOrchestratorClaimTest extends TestCase $output ); - $orchestrator->start('', 'a'); + $result = $orchestrator->start('', 'a'); $this->assertSame(1, $processor->processCalls); + $this->assertSame([1], $result->claimedIds); + $this->assertSame(0, $result->successfulCount()); + $this->assertSame(1, $result->unsuccessfulCount()); + $this->assertTrue($result->hasOutcome(ProcessingOutcome::Failed)); + $this->assertGreaterThan(0.0, $result->elapsedSeconds); + $this->assertGreaterThan(0.0, $result->averageReleaseSeconds()); + $this->assertGreaterThan(0.0, $result->releasesPerSecond()); + $this->assertGreaterThan(0, $result->peakMemoryBytes); $this->assertSame(1, $tempWorkspace->ensureMainTempPathCalls); $this->assertSame(1, $tempWorkspace->clearDirectoryCalls); $this->assertSame(1, $output->echoDescriptionCalls); $this->assertSame(1, $output->endOutputCalls); $this->assertNull(Release::query()->where('id', 1)->value('additional_pp_claimed_at')); $this->assertNull(Release::query()->where('id', 1)->value('additional_pp_claim_token')); + + Log::shouldHaveReceived('info') + ->once() + ->with( + 'Additional postprocessing run finished', + Mockery::on(static fn (array $context): bool => $context['picked'] === 1 + && $context['processed'] === 0 + && $context['failed'] === 1 + && $context['outcomes'] === ['failed' => 1] + && array_key_exists('artifacts_created', $context) + && array_key_exists('artifact_yield_percent', $context) + && array_key_exists('release_files_added', $context) + && array_key_exists('download_requests', $context) + && array_key_exists('nntp_requests', $context) + && array_key_exists('download_cache_hits', $context) + && array_key_exists('database_statements', $context) + && array_key_exists('database_milliseconds', $context) + && array_key_exists('search_sync_requests', $context) + && array_key_exists('search_sync_executions', $context) + && array_key_exists('duplicate_message_ids', $context) + && array_key_exists('unsupported_reasons', $context) + && array_key_exists('releases_per_second', $context) + && array_key_exists('average_release_seconds', $context) + && array_key_exists('stage_seconds', $context) + && array_key_exists('peak_memory_bytes', $context)), + ); } public function test_temp_setup_failure_does_not_claim_releases(): void @@ -168,9 +208,11 @@ class AdditionalProcessingOrchestratorClaimTest extends TestCase $output ); - $orchestrator->start('', 'a'); + $result = $orchestrator->start('', 'a'); $this->assertSame(0, $processor->processCalls); + $this->assertSame(0, $result->claimedCount()); + $this->assertStringContainsString('not writable', $result->setupFailure); $this->assertSame(1, $tempWorkspace->ensureMainTempPathCalls); $this->assertSame(0, $output->echoDescriptionCalls); $this->assertSame(0, $output->endOutputCalls); @@ -246,7 +288,7 @@ class FailingAdditionalReleaseProcessor extends ReleaseProcessor public function __construct() {} - public function process(ReleaseProcessingContext $context, string $mainTmpPath): void + public function process(ReleaseProcessingContext $context, string $mainTmpPath): ReleaseProcessingResult { $this->processCalls++; diff --git a/tests/Feature/AdditionalProcessingReleaseFileManagerTest.php b/tests/Feature/AdditionalProcessingReleaseFileManagerTest.php index a0ab26755..04ad67ae6 100644 --- a/tests/Feature/AdditionalProcessingReleaseFileManagerTest.php +++ b/tests/Feature/AdditionalProcessingReleaseFileManagerTest.php @@ -7,6 +7,7 @@ namespace Tests\Feature; use App\Facades\Search; use App\Models\Release; use App\Services\AdditionalProcessing\ReleaseFileManager; +use App\Services\AdditionalProcessing\State\PersistenceMetricsCollector; use App\Services\AdditionalProcessing\State\ReleaseProcessingContext; use App\Services\CollectionCleanupService; use App\Services\NameFixing\NameFixingService; @@ -14,6 +15,7 @@ use App\Services\NfoService; use App\Services\Nzb\NzbService; use App\Services\ReleaseImageService; use Illuminate\Contracts\Console\Kernel; +use Illuminate\Database\QueryException; use Illuminate\Database\Schema\Blueprint; use Illuminate\Support\Facades\DB; use Illuminate\Support\Facades\File; @@ -129,6 +131,40 @@ class AdditionalProcessingReleaseFileManagerTest extends TestCase $this->assertNull(DB::table('releases')->where('id', 1)->value('additional_pp_claim_token')); } + public function test_finalization_flushes_rows_and_synchronizes_search_immediately(): void + { + DB::table('releases')->insert($this->releaseRow()); + + Search::shouldReceive('updateRelease')->once()->with(1); + + $manager = $this->makeManagerWithImageService(new ReleaseImageService); + $context = new ReleaseProcessingContext(Release::query()->findOrFail(1)); + $this->assertTrue($manager->addFileInfo([ + 'name' => 'Example.Movie.2026.mkv', + 'size' => 2048, + 'date' => 1_788_600_000, + ], $context, '\\.(?:par2|sfv|nzb)')); + + $manager->finalizeRelease($context, false); + + $this->assertSame(1, DB::table('release_files')->count()); + $this->assertSame(1, DB::table('releases')->where('id', 1)->value('rarinnerfilecount')); + $this->assertSame([], $context->pendingReleaseFiles); + } + + public function test_database_statements_are_measured_inside_an_active_release_scope(): void + { + DB::table('releases')->insert($this->releaseRow()); + $collector = app(PersistenceMetricsCollector::class); + + $collector->beginReleaseScope(1); + DB::table('releases')->where('id', 1)->value('guid'); + $metrics = $collector->finishReleaseScope(); + + $this->assertSame(1, $metrics->databaseStatements); + $this->assertGreaterThanOrEqual(0.0, $metrics->databaseMilliseconds); + } + public function test_queued_par_hashes_flush_with_release_files(): void { DB::table('releases')->insert($this->releaseRow()); @@ -156,6 +192,35 @@ class AdditionalProcessingReleaseFileManagerTest extends TestCase $this->assertSame(1, DB::table('par_hashes')->count()); } + public function test_failed_finalization_rolls_back_buffered_rows_and_keeps_them_for_retry(): void + { + DB::table('releases')->insert($this->releaseRow()); + + Search::shouldReceive('updateRelease')->never(); + + $manager = $this->makeManager(); + $context = new ReleaseProcessingContext(Release::query()->findOrFail(1)); + $this->assertTrue($manager->addFileInfo([ + 'name' => 'Example.Movie.2026.mkv', + 'size' => 1024, + 'date' => 1_788_600_000, + ], $context, '\\.(?:par2|sfv|nzb)')); + + DB::unprepared("CREATE TRIGGER fail_release_finalize BEFORE UPDATE ON releases BEGIN SELECT RAISE(ABORT, 'forced finalization failure'); END"); + + try { + $manager->finalizeRelease($context, false); + $this->fail('Finalization should have failed.'); + } catch (QueryException $exception) { + $this->assertStringContainsString('forced finalization failure', $exception->getMessage()); + } finally { + DB::unprepared('DROP TRIGGER IF EXISTS fail_release_finalize'); + } + + $this->assertSame(0, DB::table('release_files')->count()); + $this->assertArrayHasKey('Example.Movie.2026.mkv', $context->pendingReleaseFiles); + } + public function test_float_release_file_size_is_normalized_before_queueing(): void { DB::table('releases')->insert($this->releaseRow()); @@ -230,12 +295,16 @@ class AdditionalProcessingReleaseFileManagerTest extends TestCase return $this->makeManagerWithImageService(new ReleaseImageService, $nameFixing); } + /** + * @param array $configOverrides + */ private function makeManagerWithImageService( ReleaseImageService $imageService, ?NameFixingService $nameFixing = null, + array $configOverrides = [], ): ReleaseFileManager { return new ReleaseFileManager( - $this->makeConfig(), + $this->makeConfig($configOverrides), $imageService, new NfoService, new TestNzbService, diff --git a/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php b/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php index 520e2a160..de585591a 100644 --- a/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php +++ b/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php @@ -5,6 +5,11 @@ declare(strict_types=1); namespace Tests\Feature; use App\Services\AdditionalProcessing\AdditionalProcessingOrchestrator; +use App\Services\AdditionalProcessing\DTO\AdditionalBatchResult; +use App\Services\AdditionalProcessing\DTO\DownloadMetrics; +use App\Services\AdditionalProcessing\DTO\PersistenceMetrics; +use App\Services\AdditionalProcessing\DTO\ReleaseProcessingResult; +use App\Services\AdditionalProcessing\Enums\ProcessingOutcome; use App\Services\Runners\PostProcessRunner; use Illuminate\Contracts\Console\Kernel; use Illuminate\Database\Schema\Blueprint; @@ -186,7 +191,7 @@ class PostProcessRunnerAdditionalThreadsTest extends TestCase $mock->shouldReceive('start') ->once() ->with('', 'a', Mockery::type('string'), []) - ->andReturn(['claimed' => 1, 'processed' => 1, 'failed' => 0, 'claimed_ids' => [1]]); + ->andReturn($this->successfulBatch(1)); $mock->shouldReceive('finish')->once(); }); @@ -205,17 +210,17 @@ class PostProcessRunnerAdditionalThreadsTest extends TestCase ->once() ->ordered() ->with('', 'a', Mockery::type('string'), []) - ->andReturn(['claimed' => 1, 'processed' => 1, 'failed' => 0, 'claimed_ids' => [1]]); + ->andReturn($this->successfulBatch(1)); $mock->shouldReceive('start') ->once() ->ordered() ->with('', 'a', Mockery::type('string'), [1]) - ->andReturn(['claimed' => 1, 'processed' => 1, 'failed' => 0, 'claimed_ids' => [2]]); + ->andReturn($this->successfulBatch(2)); $mock->shouldReceive('start') ->once() ->ordered() ->with('', 'a', Mockery::type('string'), [1, 2]) - ->andReturn(['claimed' => 0, 'processed' => 0, 'failed' => 0, 'claimed_ids' => []]); + ->andReturn(AdditionalBatchResult::empty()); $mock->shouldReceive('finish')->once(); }); @@ -229,6 +234,154 @@ class PostProcessRunnerAdditionalThreadsTest extends TestCase $this->assertSame(0, $status); } + public function test_worker_guid_command_emits_machine_readable_batch_profiles(): void + { + $batchResult = $this->successfulBatch(1)->withPerformance(2.0, 1024); + $this->mock(AdditionalProcessingOrchestrator::class, function (MockInterface $mock) use ($batchResult): void { + $mock->shouldReceive('start')->once()->andReturn($batchResult); + $mock->shouldReceive('finish')->once(); + }); + + $status = Artisan::call('postprocess:guid', [ + 'type' => 'additional', + 'guid' => 'a', + '--profile' => true, + ]); + $profile = json_decode(trim(Artisan::output()), true, flags: JSON_THROW_ON_ERROR); + + $this->assertSame(0, $status); + $this->assertSame('additional-postprocessing-profile', $profile['event']); + $this->assertSame('v2', $profile['pipeline']); + $this->assertSame(1, $profile['attempted']); + $this->assertSame(1800, $profile['releases_per_hour']); + $this->assertSame(1024, $profile['peak_memory_bytes']); + } + + public function test_targeted_additional_command_reports_a_successful_typed_outcome(): void + { + DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]); + DB::table('releases')->insert($this->releaseRow(1, 'a')); + + $this->mock(AdditionalProcessingOrchestrator::class, function (MockInterface $mock): void { + $mock->shouldReceive('processSingleGuid') + ->once() + ->with('a-guid-1') + ->andReturn(new ReleaseProcessingResult( + 1, + 'a-guid-1', + ProcessingOutcome::Completed, + artifactsCreated: true, + releaseFilesAdded: 2, + elapsedSeconds: 1.2345, + stageDurations: ['nzb-parsing' => 0.5], + downloadMetrics: new DownloadMetrics(2, 1, 1, 100, 100), + persistenceMetrics: new PersistenceMetrics(4, 3.5, 2, 1), + )); + $mock->shouldReceive('finish')->once(); + }); + + $status = Artisan::call('releases:additional', ['--id' => 1, '--profile' => true]); + $output = Artisan::output(); + + $this->assertSame(0, $status); + $this->assertStringContainsString('outcome completed', $output); + $this->assertStringContainsString('Artifacts created', $output); + $this->assertStringContainsString('Release files added', $output); + $this->assertStringContainsString('Stage profile', $output); + $this->assertStringContainsString('Download profile', $output); + $this->assertStringContainsString('Persistence profile', $output); + } + + public function test_targeted_additional_command_fails_for_an_unsuccessful_typed_outcome(): void + { + DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]); + DB::table('releases')->insert($this->releaseRow(1, 'a')); + + $this->mock(AdditionalProcessingOrchestrator::class, function (MockInterface $mock): void { + $mock->shouldReceive('processSingleGuid') + ->once() + ->with('a-guid-1') + ->andReturn(new ReleaseProcessingResult( + 1, + 'a-guid-1', + ProcessingOutcome::TimedOut, + reason: 'The release exceeded the post-processing timeout.', + elapsedSeconds: 2.3456, + )); + $mock->shouldReceive('finish')->once(); + }); + + $status = Artisan::call('releases:additional', ['--id' => 1]); + $output = Artisan::output(); + + $this->assertSame(3, $status); + $this->assertStringContainsString('outcome timed-out', $output); + $this->assertStringContainsString('Reason', $output); + $this->assertStringContainsString('The release exceeded the post-processing timeout.', $output); + } + + public function test_additional_diagnostics_reports_stale_claims_and_preflight_warnings_as_json(): void + { + config(['nntmux.tmp_unrar_path' => '']); + DB::table('settings')->upsert([ + ['name' => 'postthreads', 'value' => '33'], + ['name' => 'maxaddprocessed', 'value' => '25'], + ['name' => 'maxpptimeoutcount', 'value' => '3'], + ], ['name'], ['value']); + DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]); + DB::table('releases')->insert([ + ...$this->releaseRow(1, 'a'), + 'additional_pp_claimed_at' => now()->subHour(), + 'additional_pp_claim_token' => 'stale-claim', + ]); + + $status = Artisan::call('nntmux:additional-diagnose', ['--json' => true]); + $report = json_decode(Artisan::output(), true, flags: JSON_THROW_ON_ERROR); + $warningCodes = array_column($report['warnings'], 'code'); + + $this->assertSame(0, $status); + $this->assertSame(1, $report['backlog']['total']); + $this->assertSame(1, $report['backlog']['available']); + $this->assertSame(1, $report['backlog']['stale_claims']); + $this->assertSame(825, $report['capacity']['max_in_flight']); + $this->assertSame('v2', $report['settings']['pipeline']); + $this->assertContains('stale-claims', $warningCodes); + $this->assertContains('claim-ttl', $warningCodes); + $this->assertContains('missing-index', $warningCodes); + $this->assertContains('unwritable-temp-path', $warningCodes); + $this->assertContains('oversized-capacity', $warningCodes); + } + + public function test_additional_diagnostics_recognizes_the_covering_claim_index(): void + { + Schema::table('releases', function (Blueprint $table): void { + $table->index([ + 'passwordstatus', + 'haspreview', + 'nzbstatus', + 'leftguid', + 'additional_pp_claimed_at', + 'postdate', + ], 'ix_releases_add_pp_claim_queue'); + }); + + $status = Artisan::call('nntmux:additional-diagnose', ['--json' => true]); + $report = json_decode(Artisan::output(), true, flags: JSON_THROW_ON_ERROR); + + $this->assertSame(0, $status); + $this->assertFalse($report['indexes']['missing']); + $this->assertSame('ix_releases_add_pp_claim_queue', $report['indexes']['covering_index']); + $this->assertNotContains('missing-index', array_column($report['warnings'], 'code')); + } + + private function successfulBatch(int $releaseId): AdditionalBatchResult + { + return new AdditionalBatchResult( + [$releaseId], + [new ReleaseProcessingResult($releaseId, 'a-guid-'.$releaseId, ProcessingOutcome::Completed)], + ); + } + /** * @return array */ diff --git a/tests/Integration/AdditionalCandidateQueryMariaDbTest.php b/tests/Integration/AdditionalCandidateQueryMariaDbTest.php new file mode 100644 index 000000000..9a37736e2 --- /dev/null +++ b/tests/Integration/AdditionalCandidateQueryMariaDbTest.php @@ -0,0 +1,386 @@ +safeLoad(); + + $this->tablePrefix = 'phase4_'.getmypid().'_'.bin2hex(random_bytes(4)).'_'; + $database = (string) ($_ENV['DB_DATABASE'] ?? 'nntmux'); + + $this->setEnvironmentValue('APP_ENV', 'testing'); + $this->setEnvironmentValue('DB_URL', null); + $this->setEnvironmentValue('DB_CONNECTION', 'mariadb'); + $this->setEnvironmentValue('DB_DATABASE', $database); + + $app = require __DIR__.'/../../bootstrap/app.php'; + $app->make(Kernel::class)->bootstrap(); + $app->make('config')->set('database.connections.mariadb.prefix', $this->tablePrefix); + $app->make('db')->purge('mariadb'); + + return $app; + } + + protected function setUp(): void + { + parent::setUp(); + + if (DB::getDriverName() !== 'mariadb') { + $this->markTestSkipped('MariaDB integration test.'); + } + + $this->createSchema(); + } + + protected function tearDown(): void + { + if (isset($this->tablePrefix) && preg_match('/^phase4_\d+_[a-f0-9]{8}_$/', $this->tablePrefix) === 1) { + DB::statement('DROP TABLE IF EXISTS `'.$this->tableName('releases').'`'); + DB::statement('DROP TABLE IF EXISTS `'.$this->tableName('categories').'`'); + DB::statement('DROP TABLE IF EXISTS `'.$this->tableName('settings').'`'); + } + + DB::disconnect(); + parent::tearDown(); + } + + #[Test] + public function current_indexes_bound_examined_rows_for_realistic_claim_and_backlog_plans(): void + { + $releasesTable = $this->tableName('releases'); + $categoriesTable = $this->tableName('categories'); + + DB::statement(<<analyze(<< 1048576 + AND r.size < 107374182400 + AND r.leftguid = 'a' + AND (r.additional_pp_claimed_at IS NULL OR r.additional_pp_claimed_at < DATE_SUB(NOW(), INTERVAL 300 SECOND)) + ORDER BY r.postdate DESC, r.id ASC + LIMIT 25 + FOR UPDATE + SQL); + $bucketPlan = $this->analyze(<< 1048576 + AND r.size < 107374182400 + GROUP BY r.leftguid + ORDER BY r.leftguid + SQL); + $backlogPlan = $this->analyze(<< 1048576 + AND r.size < 107374182400 + SQL); + + $this->assertStringContainsString('ix_releases_add_pp_claim_queue', $claimPlan); + $this->assertLessThan(5_000, $this->examinedRowsFor($claimPlan, 'r')); + $this->assertLessThanOrEqual(2_000, $this->examinedRowsFor($bucketPlan, 'r')); + $this->assertLessThanOrEqual(2_000, $this->examinedRowsFor($backlogPlan, 'r')); + } + + #[Test] + public function claims_exclude_fresh_rows_recover_stale_rows_and_preserve_ordering(): void + { + DB::table('releases')->insert([ + $this->releaseRow(1, '2026-08-10 12:00:00', now()), + $this->releaseRow(2, '2026-08-10 11:00:00'), + $this->releaseRow(3, '2026-08-10 11:00:00'), + $this->releaseRow(4, '2026-08-10 10:00:00', now()->subSeconds(301)), + ]); + + $firstClaim = AdditionalCandidateQuery::claimBatch('a', 3, 'worker-one', columns: ['id']); + $secondClaim = AdditionalCandidateQuery::claimBatch('a', 3, 'worker-two', columns: ['id']); + + $this->assertSame([2, 3, 4], $firstClaim->pluck('id')->all()); + $this->assertSame([], $secondClaim->pluck('id')->all()); + $this->assertSame('worker-one', DB::table('releases')->where('id', 4)->value('additional_pp_claim_token')); + $this->assertSame('claimed', DB::table('releases')->where('id', 1)->value('additional_pp_claim_token')); + } + + #[Test] + public function concurrent_workers_do_not_claim_the_same_hot_bucket_rows(): void + { + if (! function_exists('pcntl_fork')) { + $this->markTestSkipped('The pcntl extension is required for the MariaDB contention test.'); + } + + DB::table('releases')->insert(array_map( + fn (int $id): array => $this->releaseRow($id, now()->subSeconds($id)->format('Y-m-d H:i:s')), + range(1, 20), + )); + + DB::beginTransaction(); + $firstWorkerIds = AdditionalCandidateQuery::baseBuilder(guidChar: 'a') + ->select('r.id') + ->orderByDesc('r.postdate') + ->orderBy('r.id') + ->limit(5) + ->lockForUpdate() + ->pluck('r.id') + ->map(static fn (mixed $id): int => (int) $id) + ->all(); + + $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); + if ($sockets === false) { + DB::rollBack(); + throw new RuntimeException('Unable to create an IPC socket pair.'); + } + + $pid = pcntl_fork(); + if ($pid === -1) { + DB::rollBack(); + throw new RuntimeException('Unable to fork the competing claim worker.'); + } + + if ($pid === 0) { + fclose($sockets[0]); + + try { + $connection = $this->createMariaDbConnection(); + $connection->beginTransaction(); + $releasesTable = $this->tableName('releases'); + $categoriesTable = $this->tableName('categories'); + $claimedIds = array_map( + static fn (mixed $id): int => (int) $id, + $connection->query(<<fetchAll(PDO::FETCH_COLUMN), + ); + $placeholders = implode(',', array_fill(0, count($claimedIds), '?')); + $update = $connection->prepare("UPDATE `{$releasesTable}` SET additional_pp_claimed_at = ?, additional_pp_claim_token = ? WHERE id IN ({$placeholders})"); + $update->execute([now()->format('Y-m-d H:i:s'), 'worker-two', ...$claimedIds]); + $connection->commit(); + fwrite($sockets[1], json_encode(['ids' => $claimedIds], JSON_THROW_ON_ERROR)); + fclose($sockets[1]); + exit(0); + } catch (\Throwable $exception) { + fwrite($sockets[1], json_encode(['error' => $exception->getMessage()], JSON_THROW_ON_ERROR)); + fclose($sockets[1]); + exit(1); + } + } + + fclose($sockets[1]); + usleep(200_000); + DB::table('releases')->whereIn('id', $firstWorkerIds)->update([ + 'additional_pp_claimed_at' => now(), + 'additional_pp_claim_token' => 'worker-one', + ]); + DB::commit(); + + $childPayload = stream_get_contents($sockets[0]); + fclose($sockets[0]); + pcntl_waitpid($pid, $status); + DB::purge(); + DB::reconnect(); + + /** @var array{ids?: list, error?: string} $childResult */ + $childResult = json_decode($childPayload, true, flags: JSON_THROW_ON_ERROR); + + $this->assertArrayNotHasKey('error', $childResult, $childResult['error'] ?? 'Competing worker failed.'); + $this->assertSame([1, 2, 3, 4, 5], $firstWorkerIds); + $this->assertSame([6, 7, 8, 9, 10], $childResult['ids'] ?? []); + $this->assertSame([], array_intersect($firstWorkerIds, $childResult['ids'] ?? [])); + $this->assertTrue(pcntl_wifexited($status)); + $this->assertSame(0, pcntl_wexitstatus($status)); + } + + private function createSchema(): void + { + $settingsTable = $this->tableName('settings'); + $categoriesTable = $this->tableName('categories'); + $releasesTable = $this->tableName('releases'); + + DB::statement("CREATE TABLE `{$settingsTable}` (`name` VARCHAR(255) PRIMARY KEY, `value` TEXT NULL) ENGINE=InnoDB"); + DB::statement("CREATE TABLE `{$categoriesTable}` (id INT UNSIGNED PRIMARY KEY, disablepreview TINYINT(1) NOT NULL DEFAULT 0) ENGINE=InnoDB"); + DB::statement(<<insert([ + ['name' => 'categorizeforeign', 'value' => '0'], + ['name' => 'catwebdl', 'value' => '0'], + ['name' => 'releaseprocessingtimeout', 'value' => '120'], + ]); + DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]); + } + + /** @return array */ + private function releaseRow(int $id, string $postdate, ?\DateTimeInterface $claimedAt = null): array + { + return [ + 'id' => $id, + 'guid' => 'a-guid-'.$id, + 'leftguid' => 'a', + 'passwordstatus' => -1, + 'haspreview' => -1, + 'nzbstatus' => 1, + 'categories_id' => 1, + 'size' => 2 * 1048576, + 'postdate' => $postdate, + 'additional_pp_claimed_at' => $claimedAt?->format('Y-m-d H:i:s'), + 'additional_pp_claim_token' => $claimedAt === null ? null : 'claimed', + ]; + } + + private function analyze(string $sql): string + { + $row = DB::selectOne('ANALYZE FORMAT=JSON '.$sql); + $values = (array) $row; + + return (string) reset($values); + } + + private function examinedRowsFor(string $plan, string $tableName): float + { + /** @var array $decoded */ + $decoded = json_decode($plan, true, flags: JSON_THROW_ON_ERROR); + $examinedRows = $this->findExaminedRows($decoded, $tableName); + + if ($examinedRows === null) { + throw new RuntimeException("Unable to find runtime row count for table [{$tableName}]."); + } + + return $examinedRows; + } + + /** + * @param array $node + */ + private function findExaminedRows(array $node, string $tableName): ?float + { + if (($node['table_name'] ?? null) === $tableName && isset($node['r_rows'])) { + return (float) $node['r_rows']; + } + + foreach ($node as $value) { + if (! is_array($value)) { + continue; + } + + $rows = $this->findExaminedRows($value, $tableName); + if ($rows !== null) { + return $rows; + } + } + + return null; + } + + private function tableName(string $table): string + { + return $this->tablePrefix.$table; + } + + private function createMariaDbConnection(): PDO + { + $host = (string) ($_ENV['DB_HOST'] ?? 'mariadb'); + $port = (string) ($_ENV['DB_PORT'] ?? '3306'); + $database = (string) ($_ENV['DB_DATABASE'] ?? 'nntmux'); + $username = (string) ($_ENV['DB_USERNAME'] ?? 'nntmux'); + $password = (string) ($_ENV['DB_PASSWORD'] ?? ''); + + return new PDO( + "mysql:host={$host};port={$port};dbname={$database};charset=utf8mb4", + $username, + $password, + [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION], + ); + } + + private function setEnvironmentValue(string $key, ?string $value): void + { + if ($value === null) { + putenv($key); + unset($_ENV[$key], $_SERVER[$key]); + + return; + } + + putenv($key.'='.$value); + $_ENV[$key] = $value; + $_SERVER[$key] = $value; + } +} diff --git a/tests/Unit/AdditionalProcessing/AdditionalWorkPlannerTest.php b/tests/Unit/AdditionalProcessing/AdditionalWorkPlannerTest.php new file mode 100644 index 000000000..c3af5d394 --- /dev/null +++ b/tests/Unit/AdditionalProcessing/AdditionalWorkPlannerTest.php @@ -0,0 +1,100 @@ +makeConfig([ + 'processThumbnails' => true, + 'processJPGSample' => true, + 'processMediaInfo' => true, + 'processAudioInfo' => true, + ])); + + $plan = $planner->plan([ + ['title' => 'release.part02.rar', 'segments' => ['', '']], + ['title' => '"sample.mkv" yEnc', 'segments' => ['', '']], + ['title' => '"cover.jpg" yEnc', 'segments' => ['']], + ['title' => '"track.FLAC" yEnc', 'segments' => ['