diff --git a/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php b/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php index 0de042c2a..14de575dd 100644 --- a/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php +++ b/app/Services/AdditionalProcessing/AdditionalCandidateQuery.php @@ -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 $columns + * @return EloquentCollection + */ + 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 + */ + 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 $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 $columns + * @return list + */ + 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 $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'; + } } diff --git a/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php b/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php index 48cde71ae..bbd53b5a6 100644 --- a/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php +++ b/app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php @@ -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 = ''; } } diff --git a/app/Services/AdditionalProcessing/ReleaseFileManager.php b/app/Services/AdditionalProcessing/ReleaseFileManager.php index af919eb6d..3bac73f9f 100644 --- a/app/Services/AdditionalProcessing/ReleaseFileManager.php +++ b/app/Services/AdditionalProcessing/ReleaseFileManager.php @@ -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 ?? ''); diff --git a/app/Services/AdditionalProcessing/ReleaseProcessor.php b/app/Services/AdditionalProcessing/ReleaseProcessor.php index 912e20f3e..cd66d20b2 100644 --- a/app/Services/AdditionalProcessing/ReleaseProcessor.php +++ b/app/Services/AdditionalProcessing/ReleaseProcessor.php @@ -17,6 +17,11 @@ use Illuminate\Support\Facades\File; */ class ReleaseProcessor { + /** + * @var array + */ + 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); } diff --git a/app/Services/AdditionalProcessing/State/ReleaseProcessingContext.php b/app/Services/AdditionalProcessing/State/ReleaseProcessingContext.php index d35b348d6..0cbe0f147 100644 --- a/app/Services/AdditionalProcessing/State/ReleaseProcessingContext.php +++ b/app/Services/AdditionalProcessing/State/ReleaseProcessingContext.php @@ -91,6 +91,23 @@ class ReleaseProcessingContext public int $compressedFilesChecked = 0; + /** + * @var array> + */ + public array $pendingReleaseFiles = []; + + /** + * @var array> + */ + public array $pendingParHashes = []; + + /** + * @var array|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; } /** diff --git a/app/Services/Runners/PostProcessRunner.php b/app/Services/Runners/PostProcessRunner.php index 445da3291..f565be321 100644 --- a/app/Services/Runners/PostProcessRunner.php +++ b/app/Services/Runners/PostProcessRunner.php @@ -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 $chars + * @return list + */ + 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 $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) { diff --git a/app/Services/TempWorkspaceService.php b/app/Services/TempWorkspaceService.php index f20882060..0f67accd9 100644 --- a/app/Services/TempWorkspaceService.php +++ b/app/Services/TempWorkspaceService.php @@ -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 + )); + } + } } diff --git a/app/Services/Tmux/Tmux.php b/app/Services/Tmux/Tmux.php index d429962cc..9e7bbe231 100644 --- a/app/Services/Tmux/Tmux.php +++ b/app/Services/Tmux/Tmux.php @@ -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']; diff --git a/app/Services/Tmux/TmuxMonitorService.php b/app/Services/Tmux/TmuxMonitorService.php index f74eed055..c227a7244 100644 --- a/app/Services/Tmux/TmuxMonitorService.php +++ b/app/Services/Tmux/TmuxMonitorService.php @@ -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 + */ + 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 + */ + protected function defaultDiffValues(): array + { + return array_fill_keys(array_keys($this->defaultCountValues()), '0'); + } + + /** + * @return array + */ + 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 */ diff --git a/app/Services/Tmux/TmuxTaskRunner.php b/app/Services/Tmux/TmuxTaskRunner.php index 978db39e6..bd49cf86c 100644 --- a/app/Services/Tmux/TmuxTaskRunner.php +++ b/app/Services/Tmux/TmuxTaskRunner.php @@ -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; diff --git a/database/migrations/2026_07_12_000000_add_additional_postprocessing_claim_fields.php b/database/migrations/2026_07_12_000000_add_additional_postprocessing_claim_fields.php new file mode 100644 index 000000000..8d6ce92b7 --- /dev/null +++ b/database/migrations/2026_07_12_000000_add_additional_postprocessing_claim_fields.php @@ -0,0 +1,150 @@ +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 $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; + } +}; diff --git a/tests/Feature/AdditionalCandidateQueryTest.php b/tests/Feature/AdditionalCandidateQueryTest.php index aac3e2cc9..f2bf5001f 100644 --- a/tests/Feature/AdditionalCandidateQueryTest.php +++ b/tests/Feature/AdditionalCandidateQueryTest.php @@ -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 */ - 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(); }); } } diff --git a/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php b/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php new file mode 100644 index 000000000..037c17e83 --- /dev/null +++ b/tests/Feature/AdditionalProcessingOrchestratorClaimTest.php @@ -0,0 +1,311 @@ + + */ + 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 + */ + 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++; + } +} diff --git a/tests/Feature/AdditionalProcessingReleaseFileManagerTest.php b/tests/Feature/AdditionalProcessingReleaseFileManagerTest.php new file mode 100644 index 000000000..38c927f63 --- /dev/null +++ b/tests/Feature/AdditionalProcessingReleaseFileManagerTest.php @@ -0,0 +1,279 @@ + + */ + 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 + */ + 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); + } +} diff --git a/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php b/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php new file mode 100644 index 000000000..0bf3c7bbe --- /dev/null +++ b/tests/Feature/PostProcessRunnerAdditionalThreadsTest.php @@ -0,0 +1,224 @@ + + */ + 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 + */ + 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 + */ + 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 + */ + 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(); + }); + } +} diff --git a/tests/Unit/TempWorkspaceServiceTest.php b/tests/Unit/TempWorkspaceServiceTest.php index 023f8994f..c4e68370a 100644 --- a/tests/Unit/TempWorkspaceServiceTest.php +++ b/tests/Unit/TempWorkspaceServiceTest.php @@ -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 { diff --git a/tests/Unit/TmuxMonitorServiceTest.php b/tests/Unit/TmuxMonitorServiceTest.php new file mode 100644 index 000000000..c20476bef --- /dev/null +++ b/tests/Unit/TmuxMonitorServiceTest.php @@ -0,0 +1,50 @@ +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']); + } +}