starred/newznab-tmux
1
0
Fork 0
mirror of https://github.com/NNTmux/newznab-tmux.git synced 2026-08-28 17:01:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update PPA handling

Browse Source
This commit is contained in:
DariusIII
2026-05-27 14:04:06 +02:00
parent 77f873ed9b
commit 096f9e919c
4 changed files with 240 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/Services/AdditionalProcessing/AdditionalCandidateQuery.php
+177
View File
@@ -0,0 +1,177 @@
<?php
declare(strict_types=1);
namespace App\Services\AdditionalProcessing;
use App\Models\Release;
use App\Models\Settings;
use App\Services\Runners\PostProcessRunner;
use Illuminate\Database\Eloquent\Builder;
use Illuminate\Support\Facades\DB;
/**
* Single source of truth for "needs additional postprocessing" release selection.
*
* Both the bucket-fanout query in
* {@see PostProcessRunner::processAdditional()} and the
* per-worker release fetch in
* {@see AdditionalProcessingOrchestrator::fetchReleases()}
* MUST go through this class so their predicates can never drift apart.
*
* History: the two queries were maintained independently, and a mismatch on
* size filters / nzbstatus caused releases to be advertised by the bucket
* query but rejected by the orchestrator, accumulating forever in the
* "passwordstatus=-1, haspreview=-1" backlog.
*/
final class AdditionalCandidateQuery
{
/** Default size lower bound when the setting is empty/unset (megabytes). */
public const int DEFAULT_MIN_SIZE_MB = 1;
/** Default size upper bound when the setting is empty/unset (gigabytes). */
public const int DEFAULT_MAX_SIZE_GB = 100;
/**
* Hard cap on the bucket fan-out. `leftguid` is the first character of a
* hex GUID, so there are at most 16 distinct values (0-9, a-f). There is
* no reason to dispatch more buckets than that per scheduler cycle.
* Per-cycle concurrency is governed by the existing `postthreads` setting
* inside {@see PostProcessRunner::runPostProcess()},
* so a separate setting is unnecessary.
*/
public const int BUCKET_LIMIT = 16;
/**
* Resolve the minimum-size filter (megabytes). Returns 0 when disabled.
*
* An explicit '0' setting means "no minimum size filter". An empty/null
* setting falls back to {@see self::DEFAULT_MIN_SIZE_MB}.
*/
public static function minSizeMB(): int
{
$value = Settings::settingValue('minsizetopostprocess');
if ($value === '' || $value === null) {
return self::DEFAULT_MIN_SIZE_MB;
}
return max(0, (int) $value);
}
/**
* Resolve the maximum-size filter (gigabytes). Returns 0 when disabled.
*
* An explicit '0' setting means "no maximum size filter". An empty/null
* setting falls back to {@see self::DEFAULT_MAX_SIZE_GB}.
*/
public static function maxSizeGB(): int
{
$value = Settings::settingValue('maxsizetopostprocess');
if ($value === '' || $value === null) {
return self::DEFAULT_MAX_SIZE_GB;
}
return max(0, (int) $value);
}
/**
* Apply the candidate-selection predicates to an Eloquent builder.
*
* The builder MUST already be aliased as `r` for releases and joined to
* `categories as c`. Optional group / GUID-character constraints can be
* applied on top.
*
* @param Builder<Release> $query
* @return Builder<Release>
*/
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<Release>
*/
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<int, string>
*/
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();
}
}
app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php
+48 -25
View File
@@ -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();
}
app/Services/AdditionalProcessing/Config/ProcessingConfiguration.php
+7 -2
View File
@@ -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');
app/Services/Runners/PostProcessRunner.php
+8 -15
View File
@@ -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');