From 096f9e919cd9bb0d1810b931c3cb004ce9d7e76c Mon Sep 17 00:00:00 2001 From: DariusIII Date: Wed, 27 May 2026 14:04:06 +0200 Subject: [PATCH] Update PPA handling --- .../AdditionalCandidateQuery.php | 177 ++++++++++++++++++ .../AdditionalProcessingOrchestrator.php | 73 +++++--- .../Config/ProcessingConfiguration.php | 9 +- app/Services/Runners/PostProcessRunner.php | 23 +-- 4 files changed, 240 insertions(+), 42 deletions(-) create mode 100644 app/Services/AdditionalProcessing/AdditionalCandidateQuery.php diff --git a/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php b/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php new file mode 100644 index 000000000..f2eb978a2 --- /dev/null +++ b/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php @@ -0,0 +1,177 @@ + $query + * @return Builder + */ + public static function applyPredicates( + Builder $query, + int|string $groupID = '', + string $guidChar = '', + ?int $minSizeMB = null, + ?int $maxSizeGB = null, + ): Builder { + $min = $minSizeMB ?? self::minSizeMB(); + $max = $maxSizeGB ?? self::maxSizeGB(); + $query + ->where('r.passwordstatus', -1) + ->where('r.haspreview', -1) + ->where('r.nzbstatus', 1) + ->where('c.disablepreview', 0); + if ($min > 0) { + $query->where('r.size', '>', $min * 1048576); + } + if ($max > 0) { + $query->where('r.size', '<', $max * 1073741824); + } + if ($groupID !== '' && $groupID !== 0 && $groupID !== '0') { + $query->where('r.groups_id', $groupID); + } + if ($guidChar !== '') { + $query->where('r.leftguid', $guidChar); + } + + return $query; + } + + /** + * Return a fresh Eloquent builder, joined and predicate-applied, ready for + * the orchestrator to add selects / order / limit. + * + * @return Builder + */ + public static function baseBuilder( + int|string $groupID = '', + string $guidChar = '', + ?int $minSizeMB = null, + ?int $maxSizeGB = null, + ): Builder { + $query = Release::query() + ->from('releases as r') + ->leftJoin('categories as c', 'c.id', '=', 'r.categories_id'); + + return self::applyPredicates($query, $groupID, $guidChar, $minSizeMB, $maxSizeGB); + } + + /** + * Return up to {@see self::BUCKET_LIMIT} distinct GUID first-characters + * that have at least one release matching the candidate predicates. + * + * The fan-out is capped at 16 because `leftguid` is a single hex digit. + * Worker concurrency is then capped further by the `postthreads` setting + * in {@see PostProcessRunner::runPostProcess()}. + * + * @return array + */ + public static function bucketChars(?int $limit = null): array + { + $effectiveLimit = $limit !== null && $limit > 0 + ? min($limit, self::BUCKET_LIMIT) + : self::BUCKET_LIMIT; + $bucketExpr = DB::getDriverName() === 'sqlite' + ? 'substr(r.leftguid, 1, 1)' + : 'LEFT(r.leftguid, 1)'; + $rows = self::baseBuilder() + ->select(DB::raw('DISTINCT '.$bucketExpr.' AS id')) + ->limit($effectiveLimit) + ->get(); + $chars = []; + foreach ($rows as $row) { + $id = (string) ($row->id ?? ''); + if ($id !== '') { + $chars[] = substr($id, 0, 1); + } + } + + return $chars; + } + + /** + * True when there is at least one candidate release anywhere (any char). + * Used by the drain command to know when to stop looping. + */ + public static function hasAnyCandidate(): bool + { + return self::baseBuilder()->limit(1)->exists(); + } +} diff --git a/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php b/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php index 07c96cdcd..48cde71ae 100644 --- a/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php +++ b/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php @@ -49,7 +49,7 @@ class AdditionalProcessingOrchestrator if ($this->totalReleases > 0) { $this->output->echoDescription($this->totalReleases); - $this->processReleases(); + $this->processReleases($guidChar); } } @@ -72,7 +72,7 @@ class AdditionalProcessingOrchestrator $guidChar = $release->leftguid ?? substr($release->guid, 0, 1); $groupID = ''; $this->setupTempPath($guidChar, $groupID); - $this->processReleases(); + $this->processReleases($guidChar); return true; } catch (\Throwable $e) { @@ -99,11 +99,21 @@ class AdditionalProcessingOrchestrator /** * Fetch releases for processing. + * + * The selection predicates (passwordstatus, haspreview, nzbstatus, + * disablepreview, size bounds) are owned by AdditionalCandidateQuery so + * they stay consistent with the bucket-fanout SQL in + * PostProcessRunner::processAdditional(). Do NOT inline new predicates + * here; add them to AdditionalCandidateQuery instead. */ private function fetchReleases(int|string $groupID, string $guidChar): void { - $query = Release::query() - ->from('releases as r') + $query = AdditionalCandidateQuery::baseBuilder( + $groupID, + $guidChar, + $this->config->minSizeMB, + $this->config->maxSizeGB, + ) ->select([ 'r.id', 'r.guid', @@ -118,25 +128,7 @@ class AdditionalProcessingOrchestrator 'r.predb_id', 'r.pp_timeout_count', ]) - ->selectRaw('r.id as releases_id') - ->leftJoin('categories as c', 'c.id', '=', 'r.categories_id') - ->where('r.passwordstatus', -1) - ->where('r.nzbstatus', 1) - ->where('r.haspreview', -1) - ->where('c.disablepreview', 0); - - if ($this->config->maxSizeGB > 0) { - $query->where('r.size', '<', $this->config->maxSizeGB * 1073741824); - } - if ($this->config->minSizeMB > 0) { - $query->where('r.size', '>', $this->config->minSizeMB * 1048576); - } - if ($groupID !== '') { - $query->where('r.groups_id', $groupID); - } - if ($guidChar !== '') { - $query->where('r.leftguid', $guidChar); - } + ->selectRaw('r.id as releases_id'); $this->releases = $query ->orderBy('r.passwordstatus') @@ -149,14 +141,45 @@ class AdditionalProcessingOrchestrator /** * Process all fetched releases. * + * Each release is processed inside its own try/catch so that a single + * poison release cannot stall an entire GUID-character bucket. Without + * this, an exception from ReleaseProcessor::process() would abort the + * foreach for the whole worker, the same release would be re-selected on + * every subsequent cycle, and the "needs additional pp" backlog would + * grow indefinitely without any visible failure. + * * @throws Exception */ - private function processReleases(): void + private function processReleases(string $guidChar = ''): void { + $processed = 0; + $failed = 0; + foreach ($this->releases as $release) { - $this->processor->process(new ReleaseProcessingContext($release), $this->mainTmpPath); + try { + $this->processor->process(new ReleaseProcessingContext($release), $this->mainTmpPath); + $processed++; + } catch (\Throwable $e) { + $failed++; + Log::error('Additional postprocessing failed for release '.($release->id ?? '?').': '.$e->getMessage(), [ + 'release_id' => $release->id ?? null, + 'guid' => $release->guid ?? null, + 'guid_char' => $guidChar, + 'exception' => $e, + ]); + // Don't rethrow: keep draining the bucket. The release will be + // re-selected on the next cycle, and the pp_timeout_count / + // maxpptimeoutcount machinery will eventually drop it. + } } + Log::info('Additional postprocessing run finished', [ + 'guid_char' => $guidChar, + 'picked' => $this->totalReleases, + 'processed' => $processed, + 'failed' => $failed, + ]); + $this->output->endOutput(); } diff --git a/app/Services/AdditionalProcessing/Config/ProcessingConfiguration.php b/app/Services/AdditionalProcessing/Config/ProcessingConfiguration.php index 8e6a1f702..a7dca06bb 100644 --- a/app/Services/AdditionalProcessing/Config/ProcessingConfiguration.php +++ b/app/Services/AdditionalProcessing/Config/ProcessingConfiguration.php @@ -5,6 +5,7 @@ declare(strict_types=1); namespace App\Services\AdditionalProcessing\Config; use App\Models\Settings; +use App\Services\AdditionalProcessing\AdditionalCandidateQuery; /** * Configuration DTO for additional post-processing. @@ -112,8 +113,12 @@ final readonly class ProcessingConfiguration $this->segmentsToDownload = (int) (Settings::settingValue('segmentstodownload') ?: 2); $this->maximumRarSegments = (int) (Settings::settingValue('maxpartsprocessed') ?: 3); $this->maximumRarPasswordChecks = max((int) (Settings::settingValue('passchkattempts') ?: 1), 1); - $this->maxSizeGB = (int) (Settings::settingValue('maxsizetopostprocess') ?: 100); - $this->minSizeMB = (int) (Settings::settingValue('minsizetopostprocess') ?: 100); + // Delegate to AdditionalCandidateQuery so size-filter semantics + // (explicit '0' means disabled, empty/null means default) are owned + // in one place and shared between the bucket query and the per-worker + // fetch. + $this->maxSizeGB = AdditionalCandidateQuery::maxSizeGB(); + $this->minSizeMB = AdditionalCandidateQuery::minSizeMB(); $this->alternateNNTP = (bool) config('nntmux_nntp.use_alternate_nntp_server'); $this->ffmpegDuration = (int) (Settings::settingValue('ffmpeg_duration') ?: 5); $this->addPAR2Files = (bool) config('nntmux_settings.add_par2'); diff --git a/app/Services/Runners/PostProcessRunner.php b/app/Services/Runners/PostProcessRunner.php index 6f1ce3602..445da3291 100644 --- a/app/Services/Runners/PostProcessRunner.php +++ b/app/Services/Runners/PostProcessRunner.php @@ -6,6 +6,7 @@ namespace App\Services\Runners; use App\Models\Category; use App\Models\Settings; +use App\Services\AdditionalProcessing\AdditionalCandidateQuery; use App\Services\NfoService; use Illuminate\Support\Facades\Concurrency; use Illuminate\Support\Facades\DB; @@ -220,22 +221,14 @@ class PostProcessRunner extends BaseRunner public function processAdditional(): void { - $bucketExpr = $this->guidBucketExpression('r.leftguid'); - $ppAddMinSize = Settings::settingValue('minsizetopostprocess') !== '' ? (int) Settings::settingValue('minsizetopostprocess') : 1; - $ppAddMinSize = ($ppAddMinSize > 0 ? ('AND r.size > '.($ppAddMinSize * 1048576)) : ''); - $ppAddMaxSize = (Settings::settingValue('maxsizetopostprocess') !== '') ? (int) Settings::settingValue('maxsizetopostprocess') : 100; - $ppAddMaxSize = ($ppAddMaxSize > 0 ? ('AND r.size < '.($ppAddMaxSize * 1073741824)) : ''); + // Bucket-selection predicates and size filters are owned by + // AdditionalCandidateQuery so they cannot drift away from the + // per-worker fetch in AdditionalProcessingOrchestrator::fetchReleases(). + $chars = AdditionalCandidateQuery::bucketChars(); - $sql = ' - SELECT DISTINCT '.$bucketExpr.' AS id - FROM releases r - LEFT JOIN categories c ON c.id = r.categories_id - WHERE r.passwordstatus = -1 - AND r.haspreview = -1 - AND c.disablepreview = 0 - '.$ppAddMaxSize.' '.$ppAddMinSize.' - LIMIT 16'; - $queue = DB::select($sql); + // Normalize to the shape the rest of runPostProcess() expects: + // an array of objects with an `id` (first GUID char) property. + $queue = array_map(static fn (string $c): object => (object) ['id' => $c], $chars); $maxProcesses = (int) Settings::settingValue('postthreads'); $this->runPostProcess($queue, $maxProcesses, 'additional', 'additional postprocessing');