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 additional PP

Browse Source
This commit is contained in:
DariusIII
2026-07-12 16:30:39 +02:00
parent 1e6c6f5193
commit 645eba20e2
17 changed files with 1748 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/Services/AdditionalProcessing/AdditionalCandidateQuery.php
+174 -5
View File
@@ -8,7 +8,9 @@ use App\Models\Release;
use App\Models\Settings;
use App\Services\Runners\PostProcessRunner;
use Illuminate\Database\Eloquent\Builder;
use Illuminate\Database\Eloquent\Collection as EloquentCollection;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
/**
* Single source of truth for "needs additional postprocessing" release selection.
@@ -42,6 +44,12 @@ final class AdditionalCandidateQuery
*/
public const int BUCKET_LIMIT = 16;
public const string CLAIMED_AT_COLUMN = 'additional_pp_claimed_at';
public const string CLAIM_TOKEN_COLUMN = 'additional_pp_claim_token';
private static ?bool $supportsClaims = null;
/**
* Resolve the minimum-size filter (megabytes). Returns 0 when disabled.
*
@@ -90,6 +98,7 @@ final class AdditionalCandidateQuery
string $guidChar = '',
?int $minSizeMB = null,
?int $maxSizeGB = null,
bool $includeClaimed = false,
): Builder {
$min = $minSizeMB ?? self::minSizeMB();
$max = $maxSizeGB ?? self::maxSizeGB();
@@ -110,6 +119,9 @@ final class AdditionalCandidateQuery
if ($guidChar !== '') {
$query->where('r.leftguid', $guidChar);
}
if (! $includeClaimed) {
self::applyClaimWindow($query);
}
return $query;
}
@@ -125,12 +137,13 @@ final class AdditionalCandidateQuery
string $guidChar = '',
?int $minSizeMB = null,
?int $maxSizeGB = null,
bool $includeClaimed = false,
): 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 self::applyPredicates($query, $groupID, $guidChar, $minSizeMB, $maxSizeGB, $includeClaimed);
}
/**
@@ -148,11 +161,8 @@ final class AdditionalCandidateQuery
$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 guid_bucket'))
->select(DB::raw('DISTINCT r.leftguid AS guid_bucket'))
->limit($effectiveLimit)
->get();
$chars = [];
@@ -174,4 +184,163 @@ final class AdditionalCandidateQuery
{
return self::baseBuilder()->limit(1)->exists();
}
/**
* Claim a bounded batch of release rows for one worker.
*
* @param list<string> $columns
* @return EloquentCollection<int, Release>
*/
public static function claimBatch(
string $guidChar,
int $limit,
string $token,
int|string $groupID = '',
?int $minSizeMB = null,
?int $maxSizeGB = null,
array $columns = ['*'],
): EloquentCollection {
$effectiveLimit = max(1, $limit);
return DB::transaction(function () use ($guidChar, $effectiveLimit, $token, $groupID, $minSizeMB, $maxSizeGB, $columns): EloquentCollection {
$supportsClaims = self::supportsClaims();
$query = self::baseBuilder($groupID, $guidChar, $minSizeMB, $maxSizeGB)
->select('r.id')
->orderByDesc('r.postdate')
->orderBy('r.id')
->limit($effectiveLimit);
if (DB::getDriverName() !== 'sqlite') {
$query->lockForUpdate();
}
$ids = $query
->pluck('r.id')
->map(static fn (mixed $id): int => (int) $id)
->all();
if ($ids === []) {
return (new Release)->newCollection();
}
if ($supportsClaims) {
Release::query()
->whereIn('id', $ids)
->update([
self::CLAIMED_AT_COLUMN => now(),
self::CLAIM_TOKEN_COLUMN => $token,
]);
}
return Release::query()
->whereIn('id', $ids)
->select(self::selectableColumns($columns, $supportsClaims))
->orderByRaw(self::idOrderExpression($ids))
->get();
}, 3);
}
public static function clearClaim(int $releaseId, ?string $token = null): void
{
if (! self::supportsClaims()) {
return;
}
$query = Release::query()->where('id', $releaseId);
if ($token !== null && $token !== '') {
$query->where(self::CLAIM_TOKEN_COLUMN, $token);
}
$query->update([
self::CLAIMED_AT_COLUMN => null,
self::CLAIM_TOKEN_COLUMN => null,
]);
}
/**
* @return array<string, null>
*/
public static function claimResetValues(): array
{
if (! self::supportsClaims()) {
return [];
}
return [
self::CLAIMED_AT_COLUMN => null,
self::CLAIM_TOKEN_COLUMN => null,
];
}
public static function supportsClaims(): bool
{
if (self::$supportsClaims !== null) {
return self::$supportsClaims;
}
if (! Schema::hasTable('releases')) {
return self::$supportsClaims = false;
}
return self::$supportsClaims = Schema::hasColumn('releases', self::CLAIMED_AT_COLUMN)
&& Schema::hasColumn('releases', self::CLAIM_TOKEN_COLUMN);
}
/**
* @param Builder<Release> $query
*/
private static function applyClaimWindow(Builder $query): void
{
if (! self::supportsClaims()) {
return;
}
$staleBefore = now()->subSeconds(self::claimTtlSeconds());
$query->where(function (Builder $claimQuery) use ($staleBefore): void {
$claimQuery
->whereNull('r.'.self::CLAIMED_AT_COLUMN)
->orWhere('r.'.self::CLAIMED_AT_COLUMN, '<', $staleBefore);
});
}
private static function claimTtlSeconds(): int
{
$timeout = (int) (Settings::settingValue('releaseprocessingtimeout') ?: 120);
return max(300, $timeout * 2);
}
/**
* @param list<string> $columns
* @return list<string>
*/
private static function selectableColumns(array $columns, bool $supportsClaims): array
{
if ($supportsClaims || $columns === ['*']) {
return $columns;
}
return array_values(array_filter(
$columns,
static fn (string $column): bool => ! in_array($column, [self::CLAIMED_AT_COLUMN, self::CLAIM_TOKEN_COLUMN], true),
));
}
/**
* @param list<int> $ids
*/
private static function idOrderExpression(array $ids): string
{
if (DB::getDriverName() !== 'sqlite') {
return 'FIELD(id, '.implode(',', $ids).')';
}
$cases = [];
foreach ($ids as $position => $id) {
$cases[] = 'WHEN '.(int) $id.' THEN '.(int) $position;
}
return 'CASE id '.implode(' ', $cases).' ELSE '.count($ids).' END';
}
}
app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php
+58 -33
View File
@@ -29,6 +29,8 @@ class AdditionalProcessingOrchestrator
private string $mainTmpPath = '';
private string $claimToken = '';
public function __construct(
private readonly ProcessingConfiguration $config,
private readonly ReleaseProcessor $processor,
@@ -44,7 +46,10 @@ class AdditionalProcessingOrchestrator
public function start(string $groupID = '', string $guidChar = ''): void
{
$this->finish();
$this->setupTempPath($guidChar, $groupID);
if (! $this->setupTempPath($guidChar, $groupID)) {
return;
}
$this->fetchReleases($groupID, $guidChar);
if ($this->totalReleases > 0) {
@@ -71,7 +76,10 @@ class AdditionalProcessingOrchestrator
$this->totalReleases = 1;
$guidChar = $release->leftguid ?? substr($release->guid, 0, 1);
$groupID = '';
$this->setupTempPath($guidChar, $groupID);
if (! $this->setupTempPath($guidChar, $groupID)) {
return false;
}
$this->processReleases($guidChar);
return true;
@@ -87,14 +95,29 @@ class AdditionalProcessingOrchestrator
/**
* Set up the main temp path.
*/
private function setupTempPath(string $guidChar, string $groupID): void
private function setupTempPath(string $guidChar, string $groupID): bool
{
$this->mainTmpPath = $this->tempWorkspace->ensureMainTempPath(
$this->config->tmpUnrarPath,
$guidChar,
$groupID
);
$this->tempWorkspace->clearDirectory($this->mainTmpPath, true);
try {
$this->mainTmpPath = $this->tempWorkspace->ensureMainTempPath(
$this->config->tmpUnrarPath,
$guidChar,
$groupID
);
$this->tempWorkspace->clearDirectory($this->mainTmpPath, true);
} catch (\Throwable $e) {
$this->output->warning('Additional post-processing skipped: '.$e->getMessage());
Log::error('Additional post-processing temp path is unavailable', [
'tmp_unrar_path' => $this->config->tmpUnrarPath,
'guid_char' => $guidChar,
'group_id' => $groupID,
'exception' => $e,
]);
$this->mainTmpPath = '';
return false;
}
return true;
}
/**
@@ -108,33 +131,30 @@ class AdditionalProcessingOrchestrator
*/
private function fetchReleases(int|string $groupID, string $guidChar): void
{
$query = AdditionalCandidateQuery::baseBuilder(
$groupID,
$this->claimToken = bin2hex(random_bytes(16));
$this->releases = AdditionalCandidateQuery::claimBatch(
$guidChar,
$this->config->queryLimit > 0 ? $this->config->queryLimit : 25,
$this->claimToken,
$groupID,
$this->config->minSizeMB,
$this->config->maxSizeGB,
)
->select([
'r.id',
'r.guid',
'r.name',
'r.size',
'r.groups_id',
'r.nfostatus',
'r.fromname',
'r.completion',
'r.categories_id',
'r.searchname',
'r.predb_id',
'r.pp_timeout_count',
])
->selectRaw('r.id as releases_id');
$this->releases = $query
->orderBy('r.passwordstatus')
->orderByDesc('r.postdate')
->limit($this->config->queryLimit > 0 ? $this->config->queryLimit : 25)
->get();
[
'id',
'guid',
'name',
'size',
'groups_id',
'nfostatus',
'fromname',
'completion',
'categories_id',
'searchname',
'predb_id',
'pp_timeout_count',
AdditionalCandidateQuery::CLAIM_TOKEN_COLUMN,
],
);
$this->totalReleases = $this->releases->count();
}
@@ -170,6 +190,10 @@ class AdditionalProcessingOrchestrator
// 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.
} finally {
if ($this->claimToken !== '' && ! empty($release->id)) {
AdditionalCandidateQuery::clearClaim((int) $release->id, $this->claimToken);
}
}
}
@@ -192,5 +216,6 @@ class AdditionalProcessingOrchestrator
$this->releases = collect();
$this->totalReleases = 0;
$this->claimToken = '';
}
}
app/Services/AdditionalProcessing/ReleaseFileManager.php
+98 -28
View File
@@ -7,6 +7,7 @@ namespace App\Services\AdditionalProcessing;
use App\Facades\Search;
use App\Models\Category;
use App\Models\MediaInfo as MediaInfoModel;
use App\Models\ParHash;
use App\Models\Predb;
use App\Models\Release;
use App\Models\ReleaseFile;
@@ -107,21 +108,8 @@ class ReleaseFileManager
return false;
}
// Check if a file already exists
$exists = ReleaseFile::query()
->where([
'releases_id' => $context->release->id,
'name' => $file['name'],
'size' => $file['size'] ?? 0,
])
->first();
if ($exists !== null) {
return false;
}
// Add the file
$added = ReleaseFile::addReleaseFiles(
$queued = $this->queueReleaseFile(
$context,
$context->release->id,
$file['name'],
$file['size'] ?? 0,
@@ -131,9 +119,7 @@ class ReleaseFileManager
$file['crc32'] ?? ''
);
if (! empty($added)) {
$context->addedFileInfo++;
if ($queued) {
// Check for codec spam
if (preg_match('#(?:^|[/\\\\])Codec[/\\\\]Setup\.exe$#i', $file['name'])) {
if ($this->config->debugMode) {
@@ -216,6 +202,8 @@ class ReleaseFileManager
*/
public function finalizeRelease(ReleaseProcessingContext $context, bool $processPasswords): void
{
$this->flushQueuedReleaseFiles($context);
$updateRows = ['haspreview' => 0];
// Check for existing samples
@@ -241,6 +229,8 @@ class ReleaseFileManager
$context->releaseHasPassword = false;
}
$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);
@@ -276,11 +266,11 @@ class ReleaseFileManager
}
// Increment the timeout counter and mark as processed to skip re-selection
Release::query()->where('id', $release->id)->update([
Release::query()->where('id', $release->id)->update(array_merge([
'pp_timeout_count' => $newCount,
'haspreview' => 0,
'passwordstatus' => ReleaseBrowseService::PASSWD_NONE,
]);
], AdditionalCandidateQuery::claimResetValues()));
Search::updateRelease((int) $release->id);
@@ -402,12 +392,9 @@ class ReleaseFileManager
// Add to release files
if ($this->config->addPAR2Files) {
if ($filesAdded < 11
&& ReleaseFile::query()
->where(['releases_id' => $context->release->id, 'name' => $file['name']])
->first() === null
) {
if (ReleaseFile::addReleaseFiles(
if ($filesAdded < 11) {
if ($this->queueReleaseFile(
$context,
$context->release->id,
$file['name'],
$file['size'] ?? 0,
@@ -432,8 +419,6 @@ class ReleaseFileManager
}
}
// Update file count
Release::query()->where('id', $context->release->id)->increment('rarinnerfilecount', $filesAdded);
$context->foundPAR2Info = true;
return true;
@@ -656,6 +641,91 @@ class ReleaseFileManager
return $this->fileNameCleaner->normalizeCandidateTitle($title);
}
private function queueReleaseFile(
ReleaseProcessingContext $context,
int|string $releaseId,
string $name,
int|string $size,
mixed $createdTime,
mixed $hasPassword,
string $hash = '',
string $crc = ''
): bool {
if ($name === '') {
return false;
}
$this->loadExistingReleaseFileNames($context);
if (isset($context->existingReleaseFileNames[$name]) || isset($context->pendingReleaseFiles[$name])) {
return false;
}
$context->pendingReleaseFiles[$name] = [
'releases_id' => (int) $releaseId,
'name' => $name,
'size' => (int) $size,
'created_at' => $this->normalizeCreatedTime($createdTime),
'updated_at' => now()->timestamp,
'passworded' => (int) $hasPassword,
'crc32' => $crc,
];
$context->existingReleaseFileNames[$name] = true;
$context->addedFileInfo++;
if (\strlen($hash) === 32) {
$context->pendingParHashes[$hash] = [
'releases_id' => (int) $releaseId,
'hash' => $hash,
];
}
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) {
return;
}
$existingNames = ReleaseFile::query()
->where('releases_id', $context->release->id)
->pluck('name')
->all();
$context->existingReleaseFileNames = [];
foreach ($existingNames as $name) {
$context->existingReleaseFileNames[(string) $name] = true;
}
}
private function normalizeCreatedTime(mixed $createdTime): mixed
{
if (! is_int($createdTime)) {
return $createdTime;
}
if ($createdTime === 0) {
return now()->format('Y-m-d H:i:s');
}
return Carbon::createFromTimestamp($createdTime, date_default_timezone_get())->format('Y-m-d H:i:s');
}
private function releaseHasNzbSplitWrapper(Release $release): bool
{
$name = (string) ($release->name ?? '');
app/Services/AdditionalProcessing/ReleaseProcessor.php
+11 -6
View File
@@ -17,6 +17,11 @@ use Illuminate\Support\Facades\File;
*/
class ReleaseProcessor
{
/**
* @var array<int, string>
*/
private array $groupNameCache = [];
public function __construct(
private readonly ProcessingConfiguration $config,
private readonly NzbContentParser $nzbParser,
@@ -39,7 +44,7 @@ class ReleaseProcessor
try {
$context->tmpPath = $this->tempWorkspace->createReleaseTempFolder($mainTmpPath, $release->guid);
} catch (\Throwable $e) {
$this->output->warning('Unable to create directory: '.$e->getMessage());
$this->output->warning('Unable to prepare release temp directory: '.$e->getMessage());
return;
}
@@ -141,7 +146,11 @@ class ReleaseProcessor
$context->passwordStatus = ReleaseBrowseService::PASSWD_NONE;
$context->releaseHasPassword = false;
try {
$context->releaseGroupName = UsenetGroup::getNameByID($context->release->groups_id);
$groupId = (int) $context->release->groups_id;
if (! array_key_exists($groupId, $this->groupNameCache)) {
$this->groupNameCache[$groupId] = UsenetGroup::getNameByID($groupId);
}
$context->releaseGroupName = $this->groupNameCache[$groupId];
} catch (\Throwable) {
$context->releaseGroupName = '';
}
@@ -448,10 +457,6 @@ class ReleaseProcessor
}
}
if ($context->addedFileInfo > 0) {
$this->releaseManager->updateSearchIndex($context->release->id);
}
if (! $context->foundJPGSample && $this->config->processJPGSample) {
$this->archiveFallback->processJpgFromArchiveFileList($compressedData, $result['files'], $context);
}
app/Services/AdditionalProcessing/State/ReleaseProcessingContext.php
+21
View File
@@ -91,6 +91,23 @@ class ReleaseProcessingContext
public int $compressedFilesChecked = 0;
/**
* @var array<string, array<string, mixed>>
*/
public array $pendingReleaseFiles = [];
/**
* @var array<string, array<string, mixed>>
*/
public array $pendingParHashes = [];
/**
* @var array<string, true>|null
*/
public ?array $existingReleaseFileNames = null;
public bool $releaseFilesChanged = false;
// Temp path for this release
public string $tmpPath = '';
@@ -145,6 +162,10 @@ class ReleaseProcessingContext
$this->addedFileInfo = 0;
$this->totalFileInfo = 0;
$this->compressedFilesChecked = 0;
$this->pendingReleaseFiles = [];
$this->pendingParHashes = [];
$this->existingReleaseFileNames = null;
$this->releaseFilesChanged = false;
}
/**
app/Services/Runners/PostProcessRunner.php
+101 -27
View File
@@ -7,7 +7,9 @@ namespace App\Services\Runners;
use App\Models\Category;
use App\Models\Settings;
use App\Services\AdditionalProcessing\AdditionalCandidateQuery;
use App\Services\AdditionalProcessing\AdditionalProcessingOrchestrator;
use App\Services\NfoService;
use App\Services\TempWorkspaceService;
use Illuminate\Support\Facades\Concurrency;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
@@ -32,8 +34,10 @@ class PostProcessRunner extends BaseRunner
return;
}
// If streaming is enabled, run commands with real-time output
if ((bool) config('nntmux.stream_fork_output', false) === true) {
// Additional post-processing can run for a long time per bucket. Always
// stream it so tmux shows active parallel children instead of buffering
// all output until the entire pool finishes.
if ((bool) config('nntmux.stream_fork_output', false) === true || $type === 'additional') {
$commands = [];
foreach ($releases as $release) {
// id may already be a single GUID bucket char; if not, take first char defensively
@@ -60,29 +64,22 @@ class PostProcessRunner extends BaseRunner
return;
}
// Process in batches using Laravel's native Concurrency facade
$batches = array_chunk($releases, max(1, $maxProcesses));
$commands = [];
foreach ($releases as $idx => $release) {
$char = isset($release->id) ? substr((string) $release->id, 0, 1) : '';
$commands[$idx] = PHP_BINARY.' artisan postprocess:guid '.$type.' '.$char;
}
foreach ($batches as $batchIndex => $batch) {
$tasks = [];
foreach ($batch as $idx => $release) {
$char = isset($release->id) ? substr((string) $release->id, 0, 1) : '';
// Use postprocess:guid command which accepts the GUID character
$command = PHP_BINARY.' artisan postprocess:guid '.$type.' '.$char;
$tasks[$idx] = fn () => $this->executeCommand($command);
}
try {
$results = Concurrency::run($tasks, $this->concurrencyTimeout());
foreach ($results as $taskIdx => $output) {
echo $output;
cli()->primary('Finished task for '.$desc);
}
} catch (\Throwable $e) {
Log::error('Postprocess batch failed: '.$e->getMessage());
cli()->error('Batch '.($batchIndex + 1).' failed: '.$e->getMessage());
try {
$results = $this->runParallelCommands($commands, $maxProcesses, $this->concurrencyTimeout());
foreach ($results as $output) {
echo $output;
cli()->primary('Finished task for '.$desc);
}
} catch (\Throwable $e) {
Log::error('Postprocess batch failed: '.$e->getMessage());
cli()->error('Postprocess batch failed: '.$e->getMessage());
}
}
@@ -226,14 +223,91 @@ class PostProcessRunner extends BaseRunner
// per-worker fetch in AdditionalProcessingOrchestrator::fetchReleases().
$chars = AdditionalCandidateQuery::bucketChars();
// 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');
// Normalize to the shape the rest of runPostProcess() expects:
// an array of objects with an `id` (first GUID char) property. If the
// backlog is concentrated in fewer buckets than configured threads,
// repeat buckets so claimBatch() can split one hot bucket across
// multiple workers.
$queue = $this->additionalQueue($chars, $maxProcesses);
if (! $this->prepareAdditionalTempBuckets($chars)) {
return;
}
if ($maxProcesses <= 1) {
if ($queue === []) {
$this->headerNone();
return;
}
$this->headerStart('postprocess: additional postprocessing', count($queue), 1);
$orchestrator = app(AdditionalProcessingOrchestrator::class);
foreach ($chars as $char) {
try {
$orchestrator->start('', $char);
} finally {
$orchestrator->finish();
}
cli()->primary('Finished task for additional postprocessing');
}
return;
}
$this->runPostProcess($queue, $maxProcesses, 'additional', 'additional postprocessing');
}
/**
* @param list<string> $chars
* @return list<object{id: string}>
*/
private function additionalQueue(array $chars, int $maxProcesses): array
{
if ($chars === []) {
return [];
}
$queue = array_map(static fn (string $c): object => (object) ['id' => $c], $chars);
$target = max(count($queue), $maxProcesses);
for ($index = 0; count($queue) < $target; $index++) {
$queue[] = (object) ['id' => $chars[$index % count($chars)]];
}
return $queue;
}
/**
* @param list<string> $chars
*/
private function prepareAdditionalTempBuckets(array $chars): bool
{
if ($chars === []) {
return true;
}
$tempWorkspace = app(TempWorkspaceService::class);
$basePath = (string) config('nntmux.tmp_unrar_path');
try {
foreach ($chars as $char) {
$tempWorkspace->ensureMainTempPath($basePath, $char);
}
} catch (\Throwable $e) {
cli()->warning('Additional post-processing skipped: '.$e->getMessage());
Log::error('Additional post-processing temp bucket preflight failed', [
'tmp_unrar_path' => $basePath,
'exception' => $e,
]);
return false;
}
return true;
}
public function processNfo(): void
{
if ((int) Settings::settingValue('lookupnfo') !== 1) {
app/Services/TempWorkspaceService.php
+69 -15
View File
@@ -6,6 +6,8 @@ namespace App\Services;
use Illuminate\Support\Facades\File;
use Illuminate\Support\Str;
use RuntimeException;
use Throwable;
class TempWorkspaceService
{
@@ -26,11 +28,7 @@ class TempWorkspaceService
$basePath .= $guidChar.'/';
}
if (! File::isDirectory($basePath)) {
if (! File::makeDirectory($basePath, 0777, true, true) && ! File::isDirectory($basePath)) { // @phpstan-ignore booleanNot.alwaysTrue
throw new \RuntimeException(sprintf('Directory "%s" was not created', $basePath));
}
}
$this->ensureWritableDirectory($basePath, 'Additional post-processing temp path');
return $basePath;
}
@@ -40,15 +38,10 @@ class TempWorkspaceService
*/
public function createReleaseTempFolder(string $mainTmpPath, string $guid): string
{
$this->assertWritableDirectory(rtrim($mainTmpPath, '/\\'), 'Additional post-processing temp path');
$tmpPath = rtrim($mainTmpPath, '/\\').'/'.$guid.'/';
if (! File::isDirectory($tmpPath)) {
if (! File::makeDirectory($tmpPath, 0777, true, false) && ! File::isDirectory($tmpPath)) { // @phpstan-ignore booleanNot.alwaysTrue
// Try again once in case of transient FS issues
if (! File::makeDirectory($tmpPath, 0777, true, false) && ! File::isDirectory($tmpPath)) { // @phpstan-ignore booleanNot.alwaysTrue, booleanNot.alwaysTrue, booleanAnd.alwaysTrue
throw new \RuntimeException('Unable to create directory: '.$tmpPath);
}
}
}
$this->ensureWritableDirectory($tmpPath, 'Release temp path');
return $tmpPath;
}
@@ -91,8 +84,8 @@ class TempWorkspaceService
{
try {
$files = File::allFiles($path);
} catch (\Throwable $e) {
throw new \RuntimeException('ERROR: Could not open temp dir: '.$e->getMessage());
} catch (Throwable $e) {
throw new RuntimeException('ERROR: Could not open temp dir: '.$e->getMessage());
}
if ($pattern !== '') {
@@ -112,4 +105,65 @@ class TempWorkspaceService
return $files;
}
private function ensureWritableDirectory(string $path, string $label): void
{
if (File::isDirectory($path)) {
$this->repairDirectoryIfPossible($path);
} else {
try {
if (! File::makeDirectory($path, 0777, true, true) && ! File::isDirectory($path)) { // @phpstan-ignore booleanNot.alwaysTrue
throw new RuntimeException('mkdir returned false');
}
} catch (Throwable $e) {
throw new RuntimeException(sprintf(
'%s "%s" could not be created: %s',
$label,
$path,
$e->getMessage()
), 0, $e);
}
}
$this->assertWritableDirectory($path, $label);
}
private function repairDirectoryIfPossible(string $path): void
{
if (is_writable($path)) {
return;
}
@chmod($path, 0777);
clearstatcache(true, $path);
if (is_writable($path)) {
return;
}
$normalizedPath = rtrim($path, '/\\');
$parent = dirname($normalizedPath);
if (! File::isDirectory($parent) || ! is_writable($parent)) {
return;
}
File::deleteDirectory($normalizedPath);
if (! File::isDirectory($normalizedPath)) {
File::makeDirectory($normalizedPath, 0777, true, true);
}
}
private function assertWritableDirectory(string $path, string $label): void
{
if (! File::isDirectory($path)) {
throw new RuntimeException(sprintf('%s "%s" is not a directory', $label, $path));
}
if (! is_writable($path)) {
throw new RuntimeException(sprintf(
'%s "%s" is not writable by the PHP/tmux user. Check TEMP_UNRAR_PATH ownership and permissions.',
$label,
$path
));
}
}
}
app/Services/Tmux/Tmux.php
+8 -8
View File
@@ -70,29 +70,29 @@ class Tmux
// 1) Try exact host:port
if (! empty($connections[$ip]) && ! empty($connections[$port])) {
$needle = escapeshellarg($connections[$ip].':'.$connections[$port]);
$runVar['conncounts'][$which]['active'] = str_replace("\n", '', shell_exec("ss -n | grep -- $needle | grep -c -- ESTAB")) ?: '0';
$runVar['conncounts'][$which]['total'] = str_replace("\n", '', shell_exec("ss -n | grep -c -- $needle")) ?: '0';
$runVar['conncounts'][$which]['active'] = str_replace("\n", '', shell_exec("ss -n 2>/dev/null | grep -- $needle | grep -c -- ESTAB")) ?: '0';
$runVar['conncounts'][$which]['total'] = str_replace("\n", '', shell_exec("ss -n 2>/dev/null | grep -c -- $needle")) ?: '0';
}
// 2) Fallback to host:https
if ((int) $runVar['conncounts'][$which]['active'] === 0 && (int) $runVar['conncounts'][$which]['total'] === 0 && ! empty($connections[$ip])) {
$needleHttps = escapeshellarg($connections[$ip].':https');
$runVar['conncounts'][$which]['active'] = str_replace("\n", '', shell_exec("ss -n | grep -- $needleHttps | grep -c -- ESTAB")) ?: '0';
$runVar['conncounts'][$which]['total'] = str_replace("\n", '', shell_exec("ss -n | grep -c -- $needleHttps")) ?: '0';
$runVar['conncounts'][$which]['active'] = str_replace("\n", '', shell_exec("ss -n 2>/dev/null | grep -- $needleHttps | grep -c -- ESTAB")) ?: '0';
$runVar['conncounts'][$which]['total'] = str_replace("\n", '', shell_exec("ss -n 2>/dev/null | grep -c -- $needleHttps")) ?: '0';
}
// 3) Fallback to port only
if ((int) $runVar['conncounts'][$which]['active'] === 0 && (int) $runVar['conncounts'][$which]['total'] === 0 && ! empty($connections[$port])) {
$needlePort = escapeshellarg((string) $connections[$port]);
$runVar['conncounts'][$which]['active'] = str_replace("\n", '', shell_exec("ss -n | grep -- $needlePort | grep -c -- ESTAB")) ?: '0';
$runVar['conncounts'][$which]['total'] = str_replace("\n", '', shell_exec("ss -n | grep -c -- $needlePort")) ?: '0';
$runVar['conncounts'][$which]['active'] = str_replace("\n", '', shell_exec("ss -n 2>/dev/null | grep -- $needlePort | grep -c -- ESTAB")) ?: '0';
$runVar['conncounts'][$which]['total'] = str_replace("\n", '', shell_exec("ss -n 2>/dev/null | grep -c -- $needlePort")) ?: '0';
}
// 4) Fallback to host only
if ((int) $runVar['conncounts'][$which]['active'] === 0 && (int) $runVar['conncounts'][$which]['total'] === 0 && ! empty($connections[$ip])) {
$needleIp = escapeshellarg($connections[$ip]);
$runVar['conncounts'][$which]['active'] = str_replace("\n", '', shell_exec("ss -n | grep -- $needleIp | grep -c -- ESTAB")) ?: '0';
$runVar['conncounts'][$which]['total'] = str_replace("\n", '', shell_exec("ss -n | grep -c -- $needleIp")) ?: '0';
$runVar['conncounts'][$which]['active'] = str_replace("\n", '', shell_exec("ss -n 2>/dev/null | grep -- $needleIp | grep -c -- ESTAB")) ?: '0';
$runVar['conncounts'][$which]['total'] = str_replace("\n", '', shell_exec("ss -n 2>/dev/null | grep -c -- $needleIp")) ?: '0';
}
return $runVar['conncounts'];
app/Services/Tmux/TmuxMonitorService.php
+88 -11
View File
@@ -52,10 +52,10 @@ class TmuxMonitorService
// Initialize counts
$this->runVar['counts'] = [
'iterations' => 1,
'now' => [],
'start' => [],
'diff' => [],
'percent' => [],
'now' => $this->defaultCountValues(),
'start' => $this->defaultCountValues(),
'diff' => $this->defaultDiffValues(),
'percent' => $this->defaultPercentValues(),
];
// Parse fix_crap setting into an array
@@ -214,6 +214,8 @@ class TmuxMonitorService
protected function getProcessCounts(): void
{
$timer = time();
$this->runVar['counts']['now']['work'] = $this->runVar['counts']['now']['work'] ?? 0;
$this->runVar['counts']['now']['work_available'] = $this->runVar['counts']['now']['work_available'] ?? 0;
try {
$dbName = config('nntmux.db_name');
@@ -243,13 +245,12 @@ class TmuxMonitorService
}
}
// "Misc In Process" / `work` is computed via AdditionalCandidateQuery so
// the dashboard counter is guaranteed to match exactly what the
// additional post-processor will pick up. Previously this was an
// inline SQL fragment inside Tmux::proc_query(2) which kept drifting
// (e.g. missing `nzbstatus = 1`) and left releases stuck in the queue
// forever. Do NOT re-introduce a separate predicate here.
$this->runVar['counts']['now']['work'] = AdditionalCandidateQuery::baseBuilder()->count();
// `work` remains the visible additional-processing backlog, including
// rows claimed by active workers. `work_available` is the scheduler
// gate, excluding fresh claims so tmux does not respawn duplicate
// additional workers while a claimed batch is still running.
$this->runVar['counts']['now']['work'] = AdditionalCandidateQuery::baseBuilder(includeClaimed: true)->count();
$this->runVar['counts']['now']['work_available'] = AdditionalCandidateQuery::baseBuilder()->count();
$this->runVar['timers']['query']['proc2_time'] = time() - $timer2;
@@ -341,6 +342,11 @@ class TmuxMonitorService
*/
protected function calculateStatistics(): void
{
$this->runVar['counts']['now'] = array_replace($this->defaultCountValues(), $this->runVar['counts']['now'] ?? []);
$this->runVar['counts']['start'] = array_replace($this->defaultCountValues(), $this->runVar['counts']['start'] ?? []);
$this->runVar['counts']['diff'] = array_replace($this->defaultDiffValues(), $this->runVar['counts']['diff'] ?? []);
$this->runVar['counts']['percent'] = array_replace($this->defaultPercentValues(), $this->runVar['counts']['percent'] ?? []);
// Calculate total work
$this->runVar['counts']['now']['total_work'] = 0;
@@ -402,6 +408,77 @@ class TmuxMonitorService
: 0;
}
/**
* @return array<string, int>
*/
protected function defaultCountValues(): array
{
return [
'active_groups' => 0,
'all_groups' => 0,
'audio' => 0,
'backfill_groups_date' => 0,
'backfill_groups_days' => 0,
'binaries_table' => 0,
'books' => 0,
'collections_table' => 0,
'console' => 0,
'distinct_predb_matched' => 0,
'misc' => 0,
'missed_parts_table' => 0,
'movies' => 0,
'nfo' => 0,
'parts_table' => 0,
'pc' => 0,
'predb' => 0,
'predb_matched' => 0,
'processanime' => 0,
'processbooks' => 0,
'processconsole' => 0,
'processgames' => 0,
'processmovies' => 0,
'processmusic' => 0,
'processnfo' => 0,
'processrenames' => 0,
'processtv' => 0,
'releases' => 0,
'renamed' => 0,
'total_work' => 0,
'tv' => 0,
'work' => 0,
'work_available' => 0,
'xxx' => 0,
];
}
/**
* @return array<string, string>
*/
protected function defaultDiffValues(): array
{
return array_fill_keys(array_keys($this->defaultCountValues()), '0');
}
/**
* @return array<string, int|string>
*/
protected function defaultPercentValues(): array
{
return [
'audio' => 0,
'books' => 0,
'console' => 0,
'misc' => 0,
'movies' => 0,
'nfo' => 0,
'pc' => 0,
'predb_matched' => 0,
'renamed' => 0,
'tv' => 0,
'xxx' => 0,
];
}
/**
* Update connection counts
*/
app/Services/Tmux/TmuxTaskRunner.php
+1 -1
View File
@@ -552,7 +552,7 @@ class TmuxTaskRunner
return $this->disablePane($pane, 'Post-process Additional', 'disabled in settings');
}
$hasWork = (int) ($runVar['counts']['now']['work'] ?? 0) > 0;
$hasWork = (int) ($runVar['counts']['now']['work_available'] ?? $runVar['counts']['now']['work'] ?? 0) > 0;
$hasNfo = (int) ($runVar['counts']['now']['processnfo'] ?? 0) > 0;
$niceness = Settings::settingValue('niceness') ?? 2;
database/migrations/2026_07_12_000000_add_additional_postprocessing_claim_fields.php
+150
View File
@@ -0,0 +1,150 @@
<?php
use Illuminate\Database\Migrations\Migration;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
return new class extends Migration
{
private const string RELEASES_CLAIM_INDEX = 'ix_releases_add_pp_claim_queue';
public function up(): void
{
if (Schema::hasTable('releases')) {
Schema::table('releases', function (Blueprint $table): void {
if (! Schema::hasColumn('releases', 'additional_pp_claimed_at')) {
$table->timestamp('additional_pp_claimed_at')->nullable()->after('pp_timeout_count');
}
if (! Schema::hasColumn('releases', 'additional_pp_claim_token')) {
$table->string('additional_pp_claim_token', 64)->nullable()->after('additional_pp_claimed_at');
}
});
if (! $this->indexExists('releases', self::RELEASES_CLAIM_INDEX)) {
$this->createAdditionalClaimIndex();
}
}
if (Schema::hasTable('release_files')) {
$this->ensureReleaseFilesPrimaryKey();
}
}
public function down(): void
{
if (Schema::hasTable('releases')) {
if ($this->indexExists('releases', self::RELEASES_CLAIM_INDEX)) {
Schema::table('releases', function (Blueprint $table): void {
$table->dropIndex(self::RELEASES_CLAIM_INDEX);
});
}
Schema::table('releases', function (Blueprint $table): void {
if (Schema::hasColumn('releases', 'additional_pp_claim_token')) {
$table->dropColumn('additional_pp_claim_token');
}
if (Schema::hasColumn('releases', 'additional_pp_claimed_at')) {
$table->dropColumn('additional_pp_claimed_at');
}
});
}
}
private function createAdditionalClaimIndex(): void
{
if (DB::getDriverName() === 'sqlite') {
Schema::table('releases', function (Blueprint $table): void {
$table->index([
'passwordstatus',
'haspreview',
'nzbstatus',
'leftguid',
'additional_pp_claimed_at',
'postdate',
], self::RELEASES_CLAIM_INDEX);
});
return;
}
DB::statement(
'CREATE INDEX `'.self::RELEASES_CLAIM_INDEX.'` ON `releases` '.
'(`passwordstatus`, `haspreview`, `nzbstatus`, `leftguid`, `additional_pp_claimed_at`, `postdate` DESC)'
);
}
private function ensureReleaseFilesPrimaryKey(): void
{
if ($this->indexWithColumnsExists('release_files', ['releases_id', 'name'])) {
return;
}
if (DB::getDriverName() === 'sqlite') {
return;
}
DB::statement('ALTER TABLE `release_files` ADD PRIMARY KEY (`releases_id`, `name`)');
}
private function indexExists(string $table, string $indexName): bool
{
if (DB::getDriverName() === 'sqlite') {
$indexes = DB::select("PRAGMA index_list('{$table}')");
foreach ($indexes as $index) {
if (($index->name ?? null) === $indexName) {
return true;
}
}
return false;
}
return DB::select("SHOW INDEX FROM `{$table}` WHERE Key_name = ?", [$indexName]) !== [];
}
/**
* @param list<string> $columns
*/
private function indexWithColumnsExists(string $table, array $columns): bool
{
if (DB::getDriverName() === 'sqlite') {
foreach (DB::select("PRAGMA index_list('{$table}')") as $index) {
$indexName = (string) ($index->name ?? '');
if ($indexName === '') {
continue;
}
$indexColumns = array_map(
static fn (object $row): string => (string) $row->name,
DB::select("PRAGMA index_info('{$indexName}')")
);
if ($indexColumns === $columns) {
return true;
}
}
return false;
}
$rows = DB::select("SHOW INDEX FROM `{$table}`");
$indexes = [];
foreach ($rows as $row) {
$indexes[$row->Key_name][(int) $row->Seq_in_index] = (string) $row->Column_name;
}
foreach ($indexes as $indexColumns) {
ksort($indexColumns);
if (array_values($indexColumns) === $columns) {
return true;
}
}
return false;
}
};
tests/Feature/AdditionalCandidateQueryTest.php
+74 -2
View File
@@ -99,19 +99,86 @@ class AdditionalCandidateQueryTest extends TestCase
$this->assertSame(['0', '9', 'a', 'f'], $chars);
}
public function test_bucket_chars_skip_active_claims_but_include_stale_claims(): void
{
DB::table('categories')->insert([
['id' => 1, 'disablepreview' => 0],
]);
DB::table('releases')->insert([
$this->releaseRow(1, 'a', claimedAt: now()),
$this->releaseRow(2, 'b', claimedAt: now()->subSeconds(301)),
$this->releaseRow(3, 'c'),
]);
$chars = AdditionalCandidateQuery::bucketChars();
sort($chars);
$this->assertSame(['b', 'c'], $chars);
}
public function test_monitor_builder_can_include_claimed_releases_while_available_builder_excludes_them(): void
{
DB::table('categories')->insert([
['id' => 1, 'disablepreview' => 0],
]);
DB::table('releases')->insert([
$this->releaseRow(1, 'a', claimedAt: now()),
$this->releaseRow(2, 'b'),
]);
$this->assertSame(1, AdditionalCandidateQuery::baseBuilder()->count());
$this->assertSame(2, AdditionalCandidateQuery::baseBuilder(includeClaimed: true)->count());
}
public function test_claim_batch_excludes_active_claims_and_recovers_stale_claims(): void
{
DB::table('categories')->insert([
['id' => 1, 'disablepreview' => 0],
]);
DB::table('releases')->insert([
$this->releaseRow(1, 'a', postdate: '2026-07-12 10:00:00'),
$this->releaseRow(2, 'a', postdate: '2026-07-12 09:00:00'),
]);
$first = AdditionalCandidateQuery::claimBatch('a', 1, 'token-one', columns: ['id']);
$this->assertSame([1], $first->pluck('id')->all());
$second = AdditionalCandidateQuery::claimBatch('a', 10, 'token-two', columns: ['id']);
$this->assertSame([2], $second->pluck('id')->all());
DB::table('releases')
->where('id', 1)
->update(['additional_pp_claimed_at' => now()->subSeconds(301)]);
$third = AdditionalCandidateQuery::claimBatch('a', 10, 'token-three', columns: ['id']);
$this->assertSame([1], $third->pluck('id')->all());
}
/**
* @return array<string, mixed>
*/
private function releaseRow(int $id, string $leftguid, int $categoriesId = 1): array
{
private function releaseRow(
int $id,
string $leftguid,
int $categoriesId = 1,
?\DateTimeInterface $claimedAt = null,
string $postdate = '2026-07-12 00:00:00'
): array {
return [
'id' => $id,
'guid' => $leftguid.'-guid-'.$id,
'leftguid' => $leftguid,
'passwordstatus' => -1,
'haspreview' => -1,
'nzbstatus' => 1,
'categories_id' => $categoriesId,
'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',
];
}
@@ -141,6 +208,7 @@ class AdditionalCandidateQueryTest extends TestCase
DB::table('settings')->upsert([
['name' => 'categorizeforeign', 'value' => '0'],
['name' => 'catwebdl', 'value' => '0'],
['name' => 'releaseprocessingtimeout', 'value' => '120'],
], ['name'], ['value']);
Schema::dropIfExists('releases');
@@ -153,12 +221,16 @@ class AdditionalCandidateQueryTest extends TestCase
Schema::create('releases', function (Blueprint $table): void {
$table->unsignedInteger('id')->primary();
$table->string('guid');
$table->char('leftguid', 1);
$table->integer('passwordstatus');
$table->integer('haspreview');
$table->integer('nzbstatus');
$table->unsignedInteger('categories_id');
$table->unsignedBigInteger('size');
$table->dateTime('postdate')->nullable();
$table->timestamp('additional_pp_claimed_at')->nullable();
$table->string('additional_pp_claim_token', 64)->nullable();
});
}
}
tests/Feature/AdditionalProcessingOrchestratorClaimTest.php
+311
View File
@@ -0,0 +1,311 @@
<?php
declare(strict_types=1);
namespace Tests\Feature;
use App\Models\Release;
use App\Services\AdditionalProcessing\AdditionalProcessingOrchestrator;
use App\Services\AdditionalProcessing\ConsoleOutputService;
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\Schema;
use PDO;
use Tests\TestCase;
use Tests\Unit\AdditionalProcessing\CreatesProcessingConfiguration;
class AdditionalProcessingOrchestratorClaimTest extends TestCase
{
use CreatesProcessingConfiguration;
private string $databasePath;
/**
* @var array<string, string|false>
*/
private array $originalEnvironment = [];
public function createApplication()
{
$this->databasePath = sys_get_temp_dir().'/nntmux-orchestrator-claim-test.sqlite';
$this->originalEnvironment = [
'APP_ENV' => getenv('APP_ENV'),
'DB_CONNECTION' => getenv('DB_CONNECTION'),
'DB_DATABASE' => getenv('DB_DATABASE'),
];
if (file_exists($this->databasePath)) {
unlink($this->databasePath);
}
$pdo = new PDO('sqlite:'.$this->databasePath);
$pdo->exec('CREATE TABLE settings (name VARCHAR PRIMARY KEY, value TEXT NULL)');
$pdo->exec("INSERT INTO settings (name, value) VALUES ('categorizeforeign', '0'), ('catwebdl', '0'), ('releaseprocessingtimeout', '120')");
$this->setEnvironmentValue('APP_ENV', 'testing');
$this->setEnvironmentValue('DB_CONNECTION', 'sqlite');
$this->setEnvironmentValue('DB_DATABASE', $this->databasePath);
$app = require __DIR__.'/../../bootstrap/app.php';
$app->make(Kernel::class)->bootstrap();
return $app;
}
protected function setUp(): void
{
parent::setUp();
config([
'database.default' => 'sqlite',
'database.connections.sqlite.database' => $this->databasePath,
]);
DB::purge();
DB::reconnect();
$this->createSchema();
}
protected function tearDown(): void
{
if ($this->databasePath !== '' && file_exists($this->databasePath)) {
unlink($this->databasePath);
}
parent::tearDown();
foreach ($this->originalEnvironment as $key => $value) {
$this->setEnvironmentValue($key, $value === false ? null : $value);
}
}
public function test_handled_release_exception_clears_claim(): void
{
DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]);
DB::table('releases')->insert([
'id' => 1,
'guid' => 'a-guid-1',
'leftguid' => 'a',
'name' => 'Example',
'searchname' => 'Example',
'size' => 2 * 1048576,
'groups_id' => 1,
'nfostatus' => -1,
'fromname' => 'poster@example.test',
'completion' => 100,
'categories_id' => 1,
'predb_id' => 0,
'pp_timeout_count' => 0,
'passwordstatus' => -1,
'haspreview' => -1,
'nzbstatus' => 1,
'postdate' => '2026-07-12 10:00:00',
'additional_pp_claimed_at' => null,
'additional_pp_claim_token' => null,
]);
$processor = new FailingAdditionalReleaseProcessor;
$tempWorkspace = new RecordingTempWorkspaceService;
$output = new RecordingConsoleOutputService;
$orchestrator = new AdditionalProcessingOrchestrator(
$this->makeConfig(['queryLimit' => 25, 'minSizeMB' => 0, 'maxSizeGB' => 100]),
$processor,
$tempWorkspace,
$output
);
$orchestrator->start('', 'a');
$this->assertSame(1, $processor->processCalls);
$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'));
}
public function test_temp_setup_failure_does_not_claim_releases(): void
{
DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]);
DB::table('releases')->insert([
'id' => 1,
'guid' => 'a-guid-1',
'leftguid' => 'a',
'name' => 'Example',
'searchname' => 'Example',
'size' => 2 * 1048576,
'groups_id' => 1,
'nfostatus' => -1,
'fromname' => 'poster@example.test',
'completion' => 100,
'categories_id' => 1,
'predb_id' => 0,
'pp_timeout_count' => 0,
'passwordstatus' => -1,
'haspreview' => -1,
'nzbstatus' => 1,
'postdate' => '2026-07-12 10:00:00',
'additional_pp_claimed_at' => null,
'additional_pp_claim_token' => null,
]);
$processor = new FailingAdditionalReleaseProcessor;
$tempWorkspace = new FailingTempWorkspaceService;
$output = new RecordingConsoleOutputService;
$orchestrator = new AdditionalProcessingOrchestrator(
$this->makeConfig(['queryLimit' => 25, 'minSizeMB' => 0, 'maxSizeGB' => 100]),
$processor,
$tempWorkspace,
$output
);
$orchestrator->start('', 'a');
$this->assertSame(0, $processor->processCalls);
$this->assertSame(1, $tempWorkspace->ensureMainTempPathCalls);
$this->assertSame(0, $output->echoDescriptionCalls);
$this->assertSame(0, $output->endOutputCalls);
$this->assertStringContainsString('Additional post-processing skipped', $output->warnings[0] ?? '');
$this->assertNull(Release::query()->where('id', 1)->value('additional_pp_claimed_at'));
$this->assertNull(Release::query()->where('id', 1)->value('additional_pp_claim_token'));
}
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;
}
private function createSchema(): void
{
if (! Schema::hasTable('settings')) {
Schema::create('settings', function (Blueprint $table): void {
$table->string('name')->primary();
$table->text('value')->nullable();
});
}
DB::table('settings')->upsert([
['name' => 'categorizeforeign', 'value' => '0'],
['name' => 'catwebdl', 'value' => '0'],
['name' => 'releaseprocessingtimeout', 'value' => '120'],
], ['name'], ['value']);
Schema::dropIfExists('releases');
Schema::dropIfExists('categories');
Schema::create('categories', function (Blueprint $table): void {
$table->unsignedInteger('id')->primary();
$table->boolean('disablepreview')->default(false);
});
Schema::create('releases', function (Blueprint $table): void {
$table->unsignedInteger('id')->primary();
$table->string('guid');
$table->char('leftguid', 1);
$table->string('name')->default('');
$table->string('searchname')->default('');
$table->unsignedBigInteger('size')->default(0);
$table->unsignedInteger('groups_id')->default(0);
$table->integer('nfostatus')->default(0);
$table->string('fromname')->nullable();
$table->float('completion')->default(0);
$table->unsignedInteger('categories_id');
$table->unsignedInteger('predb_id')->default(0);
$table->integer('pp_timeout_count')->default(0);
$table->integer('passwordstatus');
$table->integer('haspreview');
$table->integer('nzbstatus');
$table->dateTime('postdate')->nullable();
$table->timestamp('additional_pp_claimed_at')->nullable();
$table->string('additional_pp_claim_token', 64)->nullable();
});
}
}
class FailingAdditionalReleaseProcessor extends ReleaseProcessor
{
public int $processCalls = 0;
public function __construct() {}
public function process(ReleaseProcessingContext $context, string $mainTmpPath): void
{
$this->processCalls++;
throw new \RuntimeException('boom');
}
}
class RecordingTempWorkspaceService extends TempWorkspaceService
{
public int $ensureMainTempPathCalls = 0;
public int $clearDirectoryCalls = 0;
public function ensureMainTempPath(string $basePath, string $guidChar = '', string $groupID = ''): string
{
$this->ensureMainTempPathCalls++;
return '/tmp/additional/';
}
public function clearDirectory(string $path, bool $preserveRoot = true): void
{
$this->clearDirectoryCalls++;
}
}
class FailingTempWorkspaceService extends RecordingTempWorkspaceService
{
public function ensureMainTempPath(string $basePath, string $guidChar = '', string $groupID = ''): string
{
$this->ensureMainTempPathCalls++;
throw new \RuntimeException('Additional post-processing temp path "/root/nope/a/" is not writable');
}
}
class RecordingConsoleOutputService extends ConsoleOutputService
{
public int $echoDescriptionCalls = 0;
public int $endOutputCalls = 0;
/**
* @var list<string>
*/
public array $warnings = [];
public function echoDescription(int $totalReleases): void
{
$this->echoDescriptionCalls++;
}
public function warning(string $message): void
{
$this->warnings[] = $message;
}
public function endOutput(): void
{
$this->endOutputCalls++;
}
}
tests/Feature/AdditionalProcessingReleaseFileManagerTest.php
+279
View File
@@ -0,0 +1,279 @@
<?php
declare(strict_types=1);
namespace Tests\Feature;
use App\Facades\Search;
use App\Models\Release;
use App\Services\AdditionalProcessing\ReleaseFileManager;
use App\Services\AdditionalProcessing\State\ReleaseProcessingContext;
use App\Services\CollectionCleanupService;
use App\Services\NameFixing\NameFixingService;
use App\Services\NfoService;
use App\Services\Nzb\NzbService;
use App\Services\ReleaseImageService;
use Illuminate\Contracts\Console\Kernel;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Mockery;
use PDO;
use ReflectionMethod;
use Tests\TestCase;
use Tests\Unit\AdditionalProcessing\CreatesProcessingConfiguration;
class AdditionalProcessingReleaseFileManagerTest extends TestCase
{
use CreatesProcessingConfiguration;
private string $databasePath;
/**
* @var array<string, string|false>
*/
private array $originalEnvironment = [];
public function createApplication()
{
$this->databasePath = sys_get_temp_dir().'/nntmux-release-file-manager-test.sqlite';
$this->originalEnvironment = [
'APP_ENV' => getenv('APP_ENV'),
'DB_CONNECTION' => getenv('DB_CONNECTION'),
'DB_DATABASE' => getenv('DB_DATABASE'),
];
if (file_exists($this->databasePath)) {
unlink($this->databasePath);
}
$pdo = new PDO('sqlite:'.$this->databasePath);
$pdo->exec('CREATE TABLE settings (name VARCHAR PRIMARY KEY, value TEXT NULL)');
$pdo->exec("INSERT INTO settings (name, value) VALUES ('categorizeforeign', '0'), ('catwebdl', '0')");
$this->setEnvironmentValue('APP_ENV', 'testing');
$this->setEnvironmentValue('DB_CONNECTION', 'sqlite');
$this->setEnvironmentValue('DB_DATABASE', $this->databasePath);
$app = require __DIR__.'/../../bootstrap/app.php';
$app->make(Kernel::class)->bootstrap();
return $app;
}
protected function setUp(): void
{
parent::setUp();
config([
'database.default' => 'sqlite',
'database.connections.sqlite.database' => $this->databasePath,
]);
DB::purge();
DB::reconnect();
$this->createSchema();
}
protected function tearDown(): void
{
Mockery::close();
if ($this->databasePath !== '' && file_exists($this->databasePath)) {
unlink($this->databasePath);
}
parent::tearDown();
foreach ($this->originalEnvironment as $key => $value) {
$this->setEnvironmentValue($key, $value === false ? null : $value);
}
}
public function test_release_file_rows_are_deduped_and_flushed_once_at_finalize(): void
{
DB::table('releases')->insert($this->releaseRow());
Search::shouldReceive('updateRelease')->once()->with(1);
$nameFixing = new CountingNameFixingService;
$manager = $this->makeManager($nameFixing);
$context = new ReleaseProcessingContext(Release::query()->findOrFail(1));
$this->assertTrue($manager->addFileInfo([
'name' => 'Example.Movie.2026.mkv',
'size' => 1024,
'date' => 1_788_600_000,
'pass' => 0,
'crc32' => 'ABC123',
], $context, '\\.(?:par2|sfv|nzb)'));
$this->assertFalse($manager->addFileInfo([
'name' => 'Example.Movie.2026.mkv',
'size' => 1024,
'date' => 1_788_600_000,
'pass' => 0,
'crc32' => 'ABC123',
], $context, '\\.(?:par2|sfv|nzb)'));
$manager->finalizeRelease($context, false);
$this->assertSame(1, $nameFixing->matchPreDbFilesCalls);
$this->assertSame(1, DB::table('release_files')->count());
$this->assertSame(1, DB::table('releases')->where('id', 1)->value('rarinnerfilecount'));
$this->assertNull(DB::table('releases')->where('id', 1)->value('additional_pp_claimed_at'));
$this->assertNull(DB::table('releases')->where('id', 1)->value('additional_pp_claim_token'));
}
public function test_queued_par_hashes_flush_with_release_files(): void
{
DB::table('releases')->insert($this->releaseRow());
Search::shouldReceive('updateRelease')->once()->with(1);
$manager = $this->makeManager();
$context = new ReleaseProcessingContext(Release::query()->findOrFail(1));
$queue = new ReflectionMethod(ReleaseFileManager::class, 'queueReleaseFile');
$queue->invoke(
$manager,
$context,
1,
'Example.Movie.2026.par2',
512,
1_788_600_000,
0,
'1234567890abcdef1234567890abcdef'
);
$manager->finalizeRelease($context, false);
$this->assertSame(1, DB::table('release_files')->count());
$this->assertSame(1, DB::table('par_hashes')->count());
}
private function makeManager(?NameFixingService $nameFixing = null): ReleaseFileManager
{
return new ReleaseFileManager(
$this->makeConfig(),
new ReleaseImageService,
new NfoService,
new TestNzbService,
$nameFixing ?? new CountingNameFixingService
);
}
/**
* @return array<string, mixed>
*/
private function releaseRow(): array
{
return [
'id' => 1,
'guid' => 'guid-1',
'name' => 'Example',
'searchname' => 'Example',
'size' => 1024,
'groups_id' => 1,
'nfostatus' => -1,
'categories_id' => 10,
'passwordstatus' => -1,
'haspreview' => -1,
'nzbstatus' => 1,
'rarinnerfilecount' => 0,
'pp_timeout_count' => 0,
'additional_pp_claimed_at' => now(),
'additional_pp_claim_token' => 'token',
];
}
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;
}
private function createSchema(): void
{
if (! Schema::hasTable('settings')) {
Schema::create('settings', function (Blueprint $table): void {
$table->string('name')->primary();
$table->text('value')->nullable();
});
}
DB::table('settings')->upsert([
['name' => 'categorizeforeign', 'value' => '0'],
['name' => 'catwebdl', 'value' => '0'],
], ['name'], ['value']);
Schema::dropIfExists('par_hashes');
Schema::dropIfExists('release_files');
Schema::dropIfExists('releases');
Schema::create('releases', function (Blueprint $table): void {
$table->unsignedInteger('id')->primary();
$table->string('guid');
$table->string('name')->default('');
$table->string('searchname')->default('');
$table->unsignedBigInteger('size')->default(0);
$table->unsignedInteger('groups_id')->default(0);
$table->integer('nfostatus')->default(0);
$table->integer('categories_id')->default(10);
$table->integer('passwordstatus')->default(-1);
$table->integer('haspreview')->default(-1);
$table->integer('nzbstatus')->default(1);
$table->integer('rarinnerfilecount')->default(0);
$table->integer('pp_timeout_count')->default(0);
$table->timestamp('additional_pp_claimed_at')->nullable();
$table->string('additional_pp_claim_token', 64)->nullable();
});
Schema::create('release_files', function (Blueprint $table): void {
$table->unsignedInteger('releases_id');
$table->string('name');
$table->unsignedBigInteger('size')->default(0);
$table->string('crc32')->default('');
$table->timestamp('created_at')->nullable();
$table->timestamp('updated_at')->nullable();
$table->boolean('passworded')->default(false);
$table->primary(['releases_id', 'name']);
});
Schema::create('par_hashes', function (Blueprint $table): void {
$table->unsignedInteger('releases_id');
$table->string('hash', 32);
$table->primary(['releases_id', 'hash']);
});
}
}
class CountingNameFixingService extends NameFixingService
{
public int $matchPreDbFilesCalls = 0;
public function matchPreDbFiles(object $release, bool $echo, bool $nameStatus, bool $show): int
{
$this->matchPreDbFilesCalls++;
return 0;
}
}
class TestNzbService extends NzbService
{
public function __construct()
{
parent::__construct(new CollectionCleanupService);
}
}
tests/Feature/PostProcessRunnerAdditionalThreadsTest.php
+224
View File
@@ -0,0 +1,224 @@
<?php
declare(strict_types=1);
namespace Tests\Feature;
use App\Services\Runners\PostProcessRunner;
use Illuminate\Contracts\Console\Kernel;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use PDO;
use Tests\TestCase;
class PostProcessRunnerAdditionalThreadsTest extends TestCase
{
private string $databasePath;
private string $tmpUnrarPath;
/**
* @var array<string, string|false>
*/
private array $originalEnvironment = [];
public function createApplication()
{
$this->databasePath = sys_get_temp_dir().'/nntmux-postprocess-additional-threads.sqlite';
$this->tmpUnrarPath = sys_get_temp_dir().'/nntmux-additional-threads-'.uniqid('', true);
$this->originalEnvironment = [
'APP_ENV' => getenv('APP_ENV'),
'DB_CONNECTION' => getenv('DB_CONNECTION'),
'DB_DATABASE' => getenv('DB_DATABASE'),
];
if (file_exists($this->databasePath)) {
unlink($this->databasePath);
}
$pdo = new PDO('sqlite:'.$this->databasePath);
$pdo->exec('CREATE TABLE settings (name VARCHAR PRIMARY KEY, value TEXT NULL)');
$pdo->exec("INSERT INTO settings (name, value) VALUES ('categorizeforeign', '0'), ('catwebdl', '0'), ('postthreads', '5'), ('releaseprocessingtimeout', '120')");
$this->setEnvironmentValue('APP_ENV', 'testing');
$this->setEnvironmentValue('DB_CONNECTION', 'sqlite');
$this->setEnvironmentValue('DB_DATABASE', $this->databasePath);
$app = require __DIR__.'/../../bootstrap/app.php';
$app->make(Kernel::class)->bootstrap();
return $app;
}
protected function setUp(): void
{
parent::setUp();
config([
'database.default' => 'sqlite',
'database.connections.sqlite.database' => $this->databasePath,
'nntmux.echocli' => false,
'nntmux.tmp_unrar_path' => $this->tmpUnrarPath,
]);
DB::purge();
DB::reconnect();
$this->createSchema();
}
protected function tearDown(): void
{
if ($this->databasePath !== '' && file_exists($this->databasePath)) {
unlink($this->databasePath);
}
if ($this->tmpUnrarPath !== '' && is_dir($this->tmpUnrarPath)) {
app('files')->deleteDirectory($this->tmpUnrarPath);
}
parent::tearDown();
foreach ($this->originalEnvironment as $key => $value) {
$this->setEnvironmentValue($key, $value === false ? null : $value);
}
}
public function test_process_additional_uses_configured_postthreads_for_streaming_process_pool(): void
{
DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]);
foreach (['0', '1', '2', '3', '4'] as $index => $leftguid) {
DB::table('releases')->insert($this->releaseRow($index + 1, $leftguid));
}
$runner = new class extends PostProcessRunner
{
public int $capturedMaxProcesses = 0;
/**
* @var array<string|int, string>
*/
public array $capturedCommands = [];
protected function runStreamingCommands(array $commands, int $maxProcesses, string $desc): void
{
$this->capturedCommands = $commands;
$this->capturedMaxProcesses = $maxProcesses;
}
};
$runner->processAdditional();
$this->assertSame(5, $runner->capturedMaxProcesses);
$this->assertCount(5, $runner->capturedCommands);
$this->assertContains(PHP_BINARY.' artisan postprocess:guid additional 0', $runner->capturedCommands);
$this->assertContains(PHP_BINARY.' artisan postprocess:guid additional 4', $runner->capturedCommands);
}
public function test_process_additional_repeats_hot_bucket_to_fill_configured_threads(): void
{
DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]);
foreach (range(1, 5) as $id) {
DB::table('releases')->insert($this->releaseRow($id, 'a'));
}
$runner = new class extends PostProcessRunner
{
public int $capturedMaxProcesses = 0;
/**
* @var array<string|int, string>
*/
public array $capturedCommands = [];
protected function runStreamingCommands(array $commands, int $maxProcesses, string $desc): void
{
$this->capturedCommands = $commands;
$this->capturedMaxProcesses = $maxProcesses;
}
};
$runner->processAdditional();
$this->assertSame(5, $runner->capturedMaxProcesses);
$this->assertCount(5, $runner->capturedCommands);
$this->assertSame(
array_fill(0, 5, PHP_BINARY.' artisan postprocess:guid additional a'),
array_values($runner->capturedCommands)
);
}
/**
* @return array<string, mixed>
*/
private function releaseRow(int $id, string $leftguid): array
{
return [
'id' => $id,
'guid' => $leftguid.'-guid-'.$id,
'leftguid' => $leftguid,
'passwordstatus' => -1,
'haspreview' => -1,
'nzbstatus' => 1,
'categories_id' => 1,
'size' => 2 * 1048576,
'postdate' => '2026-07-12 10:00:00',
'additional_pp_claimed_at' => null,
'additional_pp_claim_token' => null,
];
}
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;
}
private function createSchema(): void
{
if (! Schema::hasTable('settings')) {
Schema::create('settings', function (Blueprint $table): void {
$table->string('name')->primary();
$table->text('value')->nullable();
});
}
DB::table('settings')->upsert([
['name' => 'categorizeforeign', 'value' => '0'],
['name' => 'catwebdl', 'value' => '0'],
['name' => 'postthreads', 'value' => '5'],
['name' => 'releaseprocessingtimeout', 'value' => '120'],
], ['name'], ['value']);
Schema::dropIfExists('releases');
Schema::dropIfExists('categories');
Schema::create('categories', function (Blueprint $table): void {
$table->unsignedInteger('id')->primary();
$table->boolean('disablepreview')->default(false);
});
Schema::create('releases', function (Blueprint $table): void {
$table->unsignedInteger('id')->primary();
$table->string('guid');
$table->char('leftguid', 1);
$table->integer('passwordstatus');
$table->integer('haspreview');
$table->integer('nzbstatus');
$table->unsignedInteger('categories_id');
$table->unsignedBigInteger('size');
$table->dateTime('postdate')->nullable();
$table->timestamp('additional_pp_claimed_at')->nullable();
$table->string('additional_pp_claim_token', 64)->nullable();
});
}
}
tests/Unit/TempWorkspaceServiceTest.php
+31 -4
View File
@@ -3,12 +3,13 @@
namespace Tests\Unit;
use App\Services\TempWorkspaceService;
use Illuminate\Container\Container;
use Illuminate\Filesystem\Filesystem;
use Illuminate\Foundation\Application;
use Illuminate\Support\Facades\Facade;
use Illuminate\Support\Facades\File;
use PHPUnit\Framework\Attributes\WithoutErrorHandler;
use PHPUnit\Framework\TestCase;
use RuntimeException;
class TempWorkspaceServiceTest extends TestCase
{
@@ -20,9 +21,9 @@ class TempWorkspaceServiceTest extends TestCase
{
parent::setUp();
// Minimal Facade container for File facade
$container = new Container;
$container->instance('files', new Filesystem);
Facade::setFacadeApplication($container);
$app = new Application(dirname(__DIR__, 2));
$app->instance('files', new Filesystem);
Facade::setFacadeApplication($app);
$this->svc = new TempWorkspaceService;
// Unique base path under system temp
@@ -56,6 +57,32 @@ class TempWorkspaceServiceTest extends TestCase
$this->assertStringEndsWith('/group42/guid-123/', str_replace('\\', '/', $tmp));
}
#[WithoutErrorHandler]
public function test_ensure_main_temp_path_repairs_existing_unwritable_bucket_directory(): void
{
$bucket = $this->base.DIRECTORY_SEPARATOR.'a';
File::makeDirectory($bucket, 0555, true, true);
chmod($bucket, 0555);
$resolved = $this->svc->ensureMainTempPath($this->base, 'a', '');
$this->assertTrue(File::isDirectory($resolved));
$this->assertTrue(is_writable($resolved));
}
#[WithoutErrorHandler]
public function test_ensure_main_temp_path_reports_path_when_base_cannot_be_created(): void
{
$fileBase = $this->base.DIRECTORY_SEPARATOR.'not-a-directory';
File::put($fileBase, 'x');
$this->expectException(RuntimeException::class);
$this->expectExceptionMessage('Additional post-processing temp path');
$this->expectExceptionMessage('not-a-directory');
$this->svc->ensureMainTempPath($fileBase, 'a', '');
}
#[WithoutErrorHandler]
public function test_list_files_with_and_without_pattern(): void
{
tests/Unit/TmuxMonitorServiceTest.php
+50
View File
@@ -0,0 +1,50 @@
<?php
declare(strict_types=1);
namespace Tests\Unit;
use App\Services\Tmux\TmuxMonitorService;
use PHPUnit\Framework\TestCase;
use ReflectionClass;
use ReflectionMethod;
use ReflectionProperty;
class TmuxMonitorServiceTest extends TestCase
{
public function test_calculate_statistics_backfills_missing_work_count_defaults(): void
{
$reflection = new ReflectionClass(TmuxMonitorService::class);
/** @var TmuxMonitorService $monitor */
$monitor = $reflection->newInstanceWithoutConstructor();
$runVar = [
'counts' => [
'now' => [
'processnfo' => 2,
'tv' => 1,
],
'start' => [],
'diff' => [],
'percent' => [],
],
];
$runVarProperty = new ReflectionProperty(TmuxMonitorService::class, 'runVar');
$runVarProperty->setValue($monitor, $runVar);
$iterationsProperty = new ReflectionProperty(TmuxMonitorService::class, 'iterations');
$iterationsProperty->setValue($monitor, 1);
$calculate = new ReflectionMethod(TmuxMonitorService::class, 'calculateStatistics');
$calculate->invoke($monitor);
$updatedRunVar = $runVarProperty->getValue($monitor);
$this->assertSame(0, $updatedRunVar['counts']['now']['work']);
$this->assertSame(0, $updatedRunVar['counts']['now']['work_available']);
$this->assertSame('0', $updatedRunVar['counts']['diff']['work']);
$this->assertSame('0', $updatedRunVar['counts']['diff']['work_available']);
$this->assertSame(2, $updatedRunVar['counts']['now']['total_work']);
}
}