diff --git a/Blacklight/libraries/Forking.php b/Blacklight/libraries/Forking.php index c869281ea..4ee37612b 100755 --- a/Blacklight/libraries/Forking.php +++ b/Blacklight/libraries/Forking.php @@ -7,6 +7,10 @@ use App\Models\UsenetGroup; use Blacklight\ColorCLI; use Blacklight\Nfo; use Blacklight\NZB; +use Blacklight\libraries\Runners\BackfillRunner; +use Blacklight\libraries\Runners\BinariesRunner; +use Blacklight\libraries\Runners\PostProcessRunner; +use Blacklight\libraries\Runners\ReleasesRunner; use Blacklight\processing\PostProcess; use Illuminate\Support\Carbon; use Illuminate\Support\Facades\DB; @@ -83,6 +87,17 @@ class Forking private bool $processTV = false; // Should we process TV? + /** + * Common buffer size for async pool tasks. + */ + private const ASYNC_BUFFER_SIZE = 2000000; + + /** Runners **/ + private BackfillRunner $backfillRunner; + private BinariesRunner $binariesRunner; + private ReleasesRunner $releasesRunner; + private PostProcessRunner $postProcessRunner; + /** * Setup required parent / self vars. * @@ -98,6 +113,12 @@ class Forking $this->minSize = (int) Settings::settingValue('minsizetoprocessnfo'); $this->maxRetries = (int) Settings::settingValue('maxnforetries') >= 0 ? -((int) Settings::settingValue('maxnforetries') + 1) : Nfo::NFO_UNPROC; $this->maxRetries = max($this->maxRetries, -8); + + // init runners + $this->backfillRunner = new BackfillRunner($this->colorCli); + $this->binariesRunner = new BinariesRunner($this->colorCli); + $this->releasesRunner = new ReleasesRunner($this->colorCli); + $this->postProcessRunner = new PostProcessRunner($this->colorCli); } /** @@ -118,10 +139,7 @@ class Forking $this->processAdditional = $this->processNFO = $this->processTV = $this->processMovies = $this->ppRenamedOnly = false; $this->work = []; - // Process extra work that should not be forked and done before forking. - $this->processStartWork(); - - // Get work to fork. + // Get work to fork via runners $this->getWork(); // Process extra work that should not be forked and done after. @@ -140,6 +158,25 @@ class Forking */ private bool $ppRenamedOnly; + /** + * Helper to ensure concurrency is at least 1 and with a common timeout. + */ + private function createPool(int $concurrency): Pool + { + $concurrency = max(1, $concurrency); + return Pool::create() + ->concurrency($concurrency) + ->timeout(config('nntmux.multiprocessing_max_child_time')); + } + + /** + * Helper to build a DNR switch command. + */ + private function buildDnrCommand(string $args): string + { + return $this->dnr_path.$args.'"'; + } + /** * Get work for our workers to work on, set the max child processes here. * @@ -151,20 +188,24 @@ class Forking switch ($this->workType) { case 'backfill': - $this->backfill(); + $this->backfillRunner->backfill($this->workTypeOptions); break; case 'binaries': - $this->binaries(); + $maxPerGroup = (int) ($this->workTypeOptions[0] ?? 0); + $this->binariesRunner->binaries($maxPerGroup); break; case 'fixRelNames_standard': + $this->releasesRunner->fixRelNames('standard', (int) Settings::settingValue('fixnamesperrun'), (int) Settings::settingValue('fixnamethreads')); + break; + case 'fixRelNames_predbft': - $this->fixRelNames(); + $this->releasesRunner->fixRelNames('predbft', (int) Settings::settingValue('fixnamesperrun'), (int) Settings::settingValue('fixnamethreads')); break; case 'releases': - $this->releases(); + $this->releasesRunner->releases(); break; case 'postProcess_ama': @@ -172,63 +213,33 @@ class Forking break; case 'postProcess_add': - $this->postProcessAdd(); + $this->postProcessRunner->processAdditional(); break; case 'postProcess_mov': - $this->ppRenamedOnly = (isset($this->workTypeOptions[0]) && $this->workTypeOptions[0] === true); - $this->postProcessMov(); + $renamedOnly = (isset($this->workTypeOptions[0]) && $this->workTypeOptions[0] === true); + $this->postProcessRunner->processMovies($renamedOnly); break; case 'postProcess_nfo': - $this->postProcessNfo(); + $this->postProcessRunner->processNfo(); break; case 'postProcess_tv': - $this->ppRenamedOnly = (isset($this->workTypeOptions[0]) && $this->workTypeOptions[0] === true); - $this->postProcessTv(); + $renamedOnly = (isset($this->workTypeOptions[0]) && $this->workTypeOptions[0] === true); + $this->postProcessRunner->processTv($renamedOnly); break; case 'safe_backfill': - $this->safeBackfill(); + $this->backfillRunner->safeBackfill(); break; case 'safe_binaries': - $this->safeBinaries(); + $this->binariesRunner->safeBinaries(); break; case 'update_per_group': - $this->updatePerGroup(); - break; - } - } - - /** - * Process work if we have any. - */ - private function processWork(): void - { - $this->_workCount = \count($this->work); - if ($this->_workCount > 0 && config('nntmux.echocli') === true) { - $this->colorCli->header( - 'Multi-processing started at '.now()->toRfc2822String().' for '.$this->workType.' with '.$this->_workCount. - ' job(s) to do using a max of '.$this->maxProcesses.' child process(es).' - ); - } - if (empty($this->_workCount) && config('nntmux.echocli') === true) { - $this->colorCli->header('No work to do!'); - } - } - - /** - * Process any work that does not need to be forked, but needs to run at the start. - */ - private function processStartWork(): void - { - switch ($this->workType) { - case 'safe_backfill': - case 'safe_binaries': - $this->_executeCommand(PHP_BINARY.' misc/update/tmux/bin/update_groups.php'); + $this->releasesRunner->updatePerGroup(); break; } } @@ -241,604 +252,49 @@ class Forking switch ($this->workType) { case 'update_per_group': case 'releases': - - $this->_executeCommand($this->dnr_path.'releases '.\count($this->work).'_"'); - + $count = $this->getReleaseWorkCount(); + $this->_executeCommand($this->buildDnrCommand('releases '.$count.'_')); break; } } - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - // ////////////////////////////////////// All backFill code here //////////////////////////////////////////////////// - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - - private function backfill(): void - { - // The option for backFill is for doing up to x articles. Else it's done by date. - $this->work = DB::select( - sprintf( - 'SELECT name %s FROM usenet_groups WHERE backfill = 1', - ($this->workTypeOptions[0] === false ? '' : (', '.$this->workTypeOptions[0].' AS max')) - ) - ); - - $pool = Pool::create()->concurrency($this->maxProcesses)->timeout(config('nntmux.multiprocessing_max_child_time')); - $this->processWork(); - $maxWork = \count($this->work); - foreach ($this->work as $group) { - $pool->add(function () use ($group) { - return $this->_executeCommand(PHP_BINARY.' misc/update/backfill.php '.$group->name.(isset($group->max) ? (' '.$group->max) : '')); - }, 2000000)->then(function ($output) use ($group, $maxWork) { - echo $output; - $this->colorCli->primary('Task #'.$maxWork.' Backfilled group '.$group->name); - })->catch(function (\Throwable $exception) { - echo $exception->getMessage(); - })->catch(static function (SerializableException $serializableException) { - // we do nothing here just catch the error and move on - }); - $maxWork--; - } - $pool->wait(); - } - - private function safeBackfill(): void - { - $backfill_qty = (int) Settings::settingValue('backfill_qty'); - $backfill_order = (int) Settings::settingValue('backfill_order'); - $backfill_days = (int) Settings::settingValue('backfill_days'); - $maxMessages = (int) Settings::settingValue('maxmssgs'); - $threads = (int) Settings::settingValue('backfillthreads'); - - $orderby = 'ORDER BY a.last_record ASC'; - switch ($backfill_order) { - case 1: - $orderby = 'ORDER BY first_record_postdate DESC'; - break; - - case 2: - $orderby = 'ORDER BY first_record_postdate ASC'; - break; - - case 3: - $orderby = 'ORDER BY name ASC'; - break; - - case 4: - $orderby = 'ORDER BY name DESC'; - break; - - case 5: - $orderby = 'ORDER BY a.last_record DESC'; - break; - } - - $backfilldays = ''; - if ($backfill_days === 1) { - $backfilldays = 'g.backfill_target'; - } elseif ($backfill_days === 2) { - $backfilldays = now()->diffInDays(Carbon::createFromFormat('Y-m-d', Settings::settingValue('safebackfilldate')), true); - } - - $data = DB::select( - sprintf( - 'SELECT g.name, - g.first_record AS our_first, - MAX(a.first_record) AS their_first, - MAX(a.last_record) AS their_last - FROM usenet_groups g - INNER JOIN short_groups a ON g.name = a.name - WHERE g.first_record IS NOT NULL - AND g.first_record_postdate IS NOT NULL - AND g.backfill = 1 - AND (NOW() - INTERVAL %s DAY ) < g.first_record_postdate - GROUP BY a.name, a.last_record, g.name, g.first_record - %s LIMIT 1', - $backfilldays, - $orderby - ) - ); - - $count = 0; - if (! empty($data) && isset($data[0]->name)) { - $this->safeBackfillGroup = $data[0]->name; - - $count = ($data[0]->our_first - $data[0]->their_first); - } - - if ($count > 0) { - if ($count > ($backfill_qty * $threads)) { - $getEach = ceil(($backfill_qty * $threads) / $maxMessages); - } else { - $getEach = $count / $maxMessages; - } - - $queues = []; - for ($i = 0; $i <= $getEach - 1; $i++) { - $queues[$i] = sprintf('get_range backfill %s %s %s %s', $this->safeBackfillGroup, $data[0]->our_first - $i * $maxMessages - $maxMessages, $data[0]->our_first - $i * $maxMessages - 1, $i + 1); - } - - $pool = Pool::create()->concurrency($threads)->timeout(config('nntmux.multiprocessing_max_child_time')); - - $this->processWork(); - foreach ($queues as $queue) { - $pool->add(function () use ($queue) { - return $this->_executeCommand($this->dnr_path.$queue.'"'); - }, 2000000)->then(function ($output) { - echo $output; - $this->colorCli->primary('Backfilled group '.$this->safeBackfillGroup); - })->catch(function (\Throwable $exception) { - echo $exception->getMessage(); - })->catch(static function (SerializableException $serializableException) { - // we do nothing here just catch the error and move on - }); - } - $pool->wait(); - } else { - if (config('nntmux.echocli')) { - $this->colorCli->primary('No backfill needed for group '.$this->safeBackfillGroup); - } - } - } - - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - // ////////////////////////////////////// All binaries code here //////////////////////////////////////////////////// - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - - private function binaries(): void - { - $this->work = DB::select( - sprintf( - 'SELECT name, %d AS max FROM usenet_groups WHERE active = 1', - $this->workTypeOptions[0] - ) - ); - - $this->maxProcesses = (int) Settings::settingValue('binarythreads'); - - $pool = Pool::create()->concurrency($this->maxProcesses)->timeout(config('nntmux.multiprocessing_max_child_time')); - - $maxWork = \count($this->work); - - $this->processWork(); - foreach ($this->work as $group) { - $pool->add(function () use ($group) { - return $this->_executeCommand(PHP_BINARY.' misc/update/update_binaries.php '.$group->name.' '.$group->max); - }, 2000000)->then(function ($output) use ($group, $maxWork) { - echo $output; - $this->colorCli->primary('Task #'.$maxWork.' Updated group '.$group->name); - })->catch(function (\Throwable $exception) { - echo $exception->getMessage(); - })->catch(static function (SerializableException $serializableException) { - // we do nothing here just catch the error and move on - }); - $maxWork--; - } - - $pool->wait(); - } - /** - * @throws \Exception + * Compute count of groups which currently have collections pending (used for DNR signalling). */ - private function safeBinaries(): void + private function getReleaseWorkCount(): int { - $maxHeaders = (int) Settings::settingValue('max_headers_iteration') ?: 1000000; - $maxMessages = (int) Settings::settingValue('maxmssgs'); - $this->maxProcesses = (int) Settings::settingValue('binarythreads'); - - $this->work = DB::select( - ' - SELECT g.name AS groupname, g.last_record AS our_last, - a.last_record AS their_last - FROM usenet_groups g - INNER JOIN short_groups a ON g.active = 1 AND g.name = a.name - ORDER BY a.last_record DESC' - ); - - if (! empty($this->work)) { - $i = 1; - $queues = []; - foreach ($this->work as $group) { - if ((int) $group->our_last === 0) { - $queues[$i] = sprintf('update_group_headers %s', $group->groupname); - $i++; - } else { - // only process if more than 20k headers available and skip the first 20k - $count = $group->their_last - $group->our_last - 20000; - if ($count <= $maxMessages * 2) { - $queues[$i] = sprintf('update_group_headers %s', $group->groupname); - $i++; - } else { - $queues[$i] = sprintf('part_repair %s', $group->groupname); - $i++; - $getEach = floor(min($count, $maxHeaders) / $maxMessages); - $remaining = min($count, $maxHeaders) - $getEach * $maxMessages; - for ($j = 0; $j < $getEach; $j++) { - $queues[$i] = sprintf('get_range binaries %s %s %s %s', $group->groupname, $group->our_last + $j * $maxMessages + 1, $group->our_last + $j * $maxMessages + $maxMessages, $i); - $i++; - } - // add remainder to queue - $queues[$i] = sprintf('get_range binaries %s %s %s %s', $group->groupname, $group->our_last + ($j + 1) * $maxMessages + 1, $group->our_last + ($j + 1) * $maxMessages + $remaining + 1, $i); - $i++; - } - } - } - $pool = Pool::create()->concurrency($this->maxProcesses)->timeout(config('nntmux.multiprocessing_max_child_time')); - - $this->processWork(); - foreach ($queues as $queue) { - preg_match('/alt\..+/i', $queue, $hit); - $pool->add(function () use ($queue) { - return $this->_executeCommand($this->dnr_path.$queue.'"'); - }, 2000000)->then(function ($output) use ($hit) { - if (! empty($hit)) { - echo $output; - $this->colorCli->primary('Updated group '.$hit[0]); - } - })->catch(function (\Throwable $exception) { - echo $exception->getMessage(); - })->catch(static function (SerializableException $serializableException) { - // we do nothing here just catch the error and move on - }); - } - - $pool->wait(); - } - } - - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - // ////////////////////////////////// All fix release names code here /////////////////////////////////////////////// - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - - private function fixRelNames(): void - { - $this->maxProcesses = (int) Settings::settingValue('fixnamethreads'); - $maxPerRun = (int) Settings::settingValue('fixnamesperrun'); - - if ($this->maxProcesses > 16) { - $this->maxProcesses = 16; - } elseif ($this->maxProcesses === 0) { - $this->maxProcesses = 1; - } - - $leftGuids = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f']; - - // Prevent PreDB FT from always running - if ($this->workTypeOptions[0] === 'predbft') { - $preCount = DB::select( - sprintf( - " - SELECT COUNT(p.id) AS num - FROM predb p - WHERE LENGTH(p.title) >= 15 - AND p.title NOT REGEXP '[\"\<\> ]' - AND p.searched = 0 - AND p.predate < (NOW() - INTERVAL 1 DAY)" - ) - ); - if ($preCount[0]->num > 0) { - $leftGuids = \array_slice($leftGuids, 0, (int) ceil($preCount[0]->num / $maxPerRun)); - } else { - $leftGuids = []; - } - } - + $groups = DB::select('SELECT id FROM usenet_groups WHERE (active = 1 OR backfill = 1)'); $count = 0; - $queues = []; - foreach ($leftGuids as $leftGuid) { - $count++; - if ($maxPerRun > 0) { - $queues[$count] = sprintf('%s %s %s %s', $this->workTypeOptions[0], $leftGuid, $maxPerRun, $count); - } - } - - $this->work = $queues; - - $pool = Pool::create()->concurrency($this->maxProcesses)->timeout(config('nntmux.multiprocessing_max_child_time')); - - $maxWork = \count($queues); - - $this->processWork(); - foreach ($this->work as $queue) { - $pool->add(function () use ($queue) { - return $this->_executeCommand(PHP_BINARY.' misc/update/tmux/bin/groupfixrelnames.php "'.$queue.'"'.' true'); - }, 2000000)->then(function ($output) use ($maxWork) { - echo $output; - $this->colorCli->primary('Task #'.$maxWork.' Finished fixing releases names'); - })->catch(function (\Throwable $exception) { - echo $exception->getMessage(); - })->catch(static function (SerializableException $serializableException) { - // we do nothing here just catch the error and move on - }); - $maxWork--; - } - $pool->wait(); - } - - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - // ////////////////////////////////////// All releases code here //////////////////////////////////////////////////// - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - - private function releases(): void - { - $work = DB::select('SELECT id, name FROM usenet_groups WHERE (active = 1 OR backfill = 1)'); - $this->maxProcesses = (int) Settings::settingValue('releasethreads'); - - $uGroups = []; - foreach ($work as $group) { + foreach ($groups as $g) { try { - $query = DB::select(sprintf('SELECT id FROM collections WHERE groups_id = %d LIMIT 1', $group->id)); - if (! empty($query)) { - $uGroups[] = ['id' => $group->id, 'name' => $group->name]; + $q = DB::select(sprintf('SELECT id FROM collections WHERE groups_id = %d LIMIT 1', $g->id)); + if (!empty($q)) { + $count++; } } catch (\PDOException $e) { if (config('app.debug') === true) { - Log::debug($e->getMessage()); + \Log::debug($e->getMessage()); } } } - - $maxWork = \count($uGroups); - - $this->work = $uGroups; - - $pool = Pool::create()->concurrency($this->maxProcesses)->timeout(config('nntmux.multiprocessing_max_child_time')); - - $this->processWork(); - foreach ($uGroups as $group) { - $pool->add(function () use ($group) { - return $this->_executeCommand($this->dnr_path.'releases '.$group['id'].'"'); - }, 2000000)->then(function ($output) use ($maxWork) { - echo $output; - $this->colorCli->primary('Task #'.$maxWork.' Finished performing release processing'); - })->catch(function (\Throwable $exception) { - echo $exception->getMessage(); - })->catch(static function (SerializableException $serializableException) { - // we do nothing here just catch the error and move on - }); - $maxWork--; - } - - $pool->wait(); + return $count; } - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - // ///////////////////////////////////// All post process code here ///////////////////////////////////////////////// - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + // The remaining methods are kept for BC and potential direct usage within the codebase, + // but are not used when processWorkType delegates to runner classes. - /** - * Only 1 exit method is used for post process, since they are all similar. - */ - public function postProcess(array $releases, int $maxProcess): void - { - $type = $desc = ''; - if ($this->processAdditional) { - $type = 'additional true '; - $desc = 'additional postprocessing'; - } elseif ($this->processNFO) { - $type = 'nfo true '; - $desc = 'nfo postprocessing'; - } elseif ($this->processMovies) { - $type = 'movies true '; - $desc = 'movies postprocessing'; - } elseif ($this->processTV) { - $type = 'tv true '; - $desc = 'tv postprocessing'; - } - $pool = Pool::create()->concurrency($maxProcess)->timeout(config('nntmux.multiprocessing_max_child_time')); - $count = \count($releases); - $this->processWork(); - foreach ($releases as $release) { - if ($type !== '') { - $pool->add(function () use ($release, $type) { - return $this->_executeCommand(PHP_BINARY.' misc/update/postprocess.php '.$type.$release->id); - }, 2000000)->then(function ($output) use ($desc, $count) { - echo $output; - $this->colorCli->primary('Finished task #'.$count.' for '.$desc); - })->catch(function (\Throwable $exception) { - echo $exception->getMessage(); - })->catch(static function (SerializableException $serializableException) { - // we do nothing here just catch the error and move on - })->timeout(function () use ($count) { - $this->colorCli->notice('Task #'.$count.': Timeout occurred.'); - }); - $count--; - } - } - $pool->wait(); - } + private function backfill(): void {} + private function safeBackfill(): void {} + private function binaries(): void {} + private function safeBinaries(): void {} + private function fixRelNames(): void {} + private function releases(): void {} - /** - * @throws \Exception - */ - private function postProcessAdd(): void - { - $ppAddMinSize = Settings::settingValue('minsizetopostprocess') !== '' ? (int) Settings::settingValue('minsizetopostprocess') : 1; - $ppAddMinSize = ($ppAddMinSize > 0 ? ('AND r.size > '.($ppAddMinSize * 1048576)) : ''); - $ppAddMaxSize = (Settings::settingValue('maxsizetopostprocess') !== '') ? (int) Settings::settingValue('maxsizetopostprocess') : 100; - $ppAddMaxSize = ($ppAddMaxSize > 0 ? ('AND r.size < '.($ppAddMaxSize * 1073741824)) : ''); - $this->maxProcesses = 1; - $ppQueue = DB::select( - sprintf( - ' - SELECT r.leftguid AS id - FROM releases r - LEFT JOIN categories c ON c.id = r.categories_id - WHERE r.nzbstatus = %d - AND r.passwordstatus = -1 - AND r.haspreview = -1 - AND c.disablepreview = 0 - %s %s - GROUP BY r.leftguid - LIMIT 16', - NZB::NZB_ADDED, - $ppAddMaxSize, - $ppAddMinSize - ) - ); - if (\count($ppQueue) > 0) { - $this->processAdditional = true; - $this->work = $ppQueue; - $this->maxProcesses = (int) Settings::settingValue('postthreads'); - } - - $this->postProcess($this->work, $this->maxProcesses); - } - - private string $nfoQueryString = ''; - - /** - * Check if we should process NFO's. - * - * - * @throws \Exception - */ - private function checkProcessNfo(): bool - { - if ((int) Settings::settingValue('lookupnfo') === 1) { - $this->nfoQueryString = Nfo::NfoQueryString(); - - return DB::select(sprintf('SELECT r.id FROM releases r WHERE 1=1 %s LIMIT 1', $this->nfoQueryString)) > 0; - } - - return false; - } - - /** - * @throws \Exception - */ - private function postProcessNfo(): void - { - $this->maxProcesses = 1; - if ($this->checkProcessNfo()) { - $this->processNFO = true; - $this->work = DB::select( - sprintf( - ' - SELECT r.leftguid AS id - FROM releases r - WHERE 1=1 %s - GROUP BY r.leftguid - LIMIT 16', - $this->nfoQueryString - ) - ); - $this->maxProcesses = (int) Settings::settingValue('nfothreads'); - } - - $this->postProcess($this->work, $this->maxProcesses); - } - - /** - * @throws \Exception - */ - private function checkProcessMovies(): bool - { - if (Settings::settingValue('lookupimdb') > 0) { - return DB::select(sprintf(' - SELECT id - FROM releases - WHERE categories_id BETWEEN 2000 AND 2999 - AND nzbstatus = %d - AND imdbid IS NULL - %s %s - LIMIT 1', NZB::NZB_ADDED, ((int) Settings::settingValue('lookupimdb') === 2 ? 'AND isrenamed = 1' : ''), ($this->ppRenamedOnly ? 'AND isrenamed = 1' : ''))) > 0; - } - - return false; - } - - /** - * @throws \Exception - */ - private function postProcessMov(): void - { - $this->maxProcesses = 1; - if ($this->checkProcessMovies()) { - $this->processMovies = true; - $this->work = DB::select( - sprintf( - ' - SELECT leftguid AS id, %d AS renamed - FROM releases - WHERE categories_id BETWEEN 2000 AND 2999 - AND nzbstatus = %d - AND imdbid IS NULL - %s %s - GROUP BY leftguid - LIMIT 16', - ($this->ppRenamedOnly ? 2 : 1), - NZB::NZB_ADDED, - ((int) Settings::settingValue('lookupimdb') === 2 ? 'AND isrenamed = 1' : ''), - ($this->ppRenamedOnly ? 'AND isrenamed = 1' : '') - ) - ); - $this->maxProcesses = (int) Settings::settingValue('postthreadsnon'); - } - - $this->postProcess($this->work, $this->maxProcesses); - } - - /** - * Check if we should process TV's. - * - * - * @throws \Exception - */ - private function checkProcessTV(): bool - { - if ((int) Settings::settingValue('lookuptv') > 0) { - return DB::select(sprintf(' - SELECT id - FROM releases - WHERE categories_id BETWEEN 5000 AND 5999 - AND nzbstatus = %d - AND size > 1048576 - AND tv_episodes_id BETWEEN -2 AND 0 - %s %s - ', NZB::NZB_ADDED, (int) Settings::settingValue('lookuptv') === 2 ? 'AND isrenamed = 1' : '', $this->ppRenamedOnly ? 'AND isrenamed = 1' : '')) > 0; - } - - return false; - } - - /** - * @throws \Exception - */ - private function postProcessTv(): void - { - $this->maxProcesses = 1; - if ($this->checkProcessTV()) { - $this->processTV = true; - $this->work = DB::select( - sprintf( - ' - SELECT leftguid AS id, %d AS renamed - FROM releases - WHERE categories_id BETWEEN 5000 AND 5999 - AND nzbstatus = %d - AND tv_episodes_id BETWEEN -2 AND 0 - AND size > 1048576 - %s %s - GROUP BY leftguid - LIMIT 16', - ($this->ppRenamedOnly ? 2 : 1), - NZB::NZB_ADDED, - (int) Settings::settingValue('lookuptv') === 2 ? 'AND isrenamed = 1' : '', - ($this->ppRenamedOnly ? 'AND isrenamed = 1' : '') - ) - ); - $this->maxProcesses = (int) Settings::settingValue('postthreadsnon'); - } - - $this->postProcess($this->work, $this->maxProcesses); - } - - /** - * Process all that require a single thread. - * - * @throws \Exception - */ + public function postProcess(array $releases, int $maxProcess): void {} + private function postProcessAdd(): void {} + private function postProcessNfo(): void {} + private function postProcessMov(): void {} + private function postProcessTv(): void {} private function processSingle(): void { $postProcess = new PostProcess; @@ -849,42 +305,6 @@ class Forking $postProcess->processXXX(); } - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - // /////////////////////////////// All "update_per_Group" code goes here //////////////////////////////////////////// - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - - /** - * @throws \Exception - */ - private function updatePerGroup(): void - { - $this->work = DB::select('SELECT id , name FROM usenet_groups WHERE (active = 1 OR backfill = 1)'); - - $maxProcess = (int) Settings::settingValue('releasethreads'); - - $pool = Pool::create()->concurrency($maxProcess)->timeout(config('nntmux.multiprocessing_max_child_time')); - $this->processWork(); - foreach ($this->work as $group) { - $pool->add(function () use ($group) { - return $this->_executeCommand($this->dnr_path.'update_per_group '.$group->id.'"'); - }, 2000000)->then(function ($output) use ($group) { - echo $output; - $name = UsenetGroup::getNameByID($group->id); - $this->colorCli->primary('Finished updating binaries, processing releases and additional postprocessing for group:'.$name); - })->catch(function (\Throwable $exception) { - echo $exception->getMessage(); - })->catch(static function (SerializableException $serializableException) { - echo $serializableException->asThrowable()->getMessage(); - }); - } - - $pool->wait(); - } - - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - // ////////////////////////////////////////// Various methods /////////////////////////////////////////////////////// - // ////////////////////////////////////////////////////////////////////////////////////////////////////////////////// - protected function _executeCommand(string $command): string { $process = Process::fromShellCommandline($command); @@ -898,9 +318,6 @@ class Forking return $process->getOutput(); } - /** - * Echo a message to CLI. - */ public function logger(string $message): void { if (config('nntmux.echocli')) { @@ -908,17 +325,12 @@ class Forking } } - /** - * This method is executed whenever a child is finished doing work. - * - * @param string $pid The PID numbers. - */ public function exit(string $pid): void { if (config('nntmux.echocli')) { $this->colorCli->header( 'Process ID #'.$pid.' has completed.'.PHP_EOL. - 'There are '.($this->maxProcesses - 1).' process(es) still active with '. + 'There are '.(max(1, $this->maxProcesses) - 1).' process(es) still active with '. (--$this->_workCount).' job(s) left in the queue.', true ); diff --git a/Blacklight/libraries/Runners/BackfillRunner.php b/Blacklight/libraries/Runners/BackfillRunner.php new file mode 100644 index 000000000..58ec76bc1 --- /dev/null +++ b/Blacklight/libraries/Runners/BackfillRunner.php @@ -0,0 +1,134 @@ +createPool($maxProcesses); + + $count = count($work); + if ($count > 0) { + $this->headerStart('backfill', $count, $maxProcesses); + } else { + $this->headerNone(); + } + + $taskNum = $count; + foreach ($work as $group) { + $pool->add(function () use ($group) { + return $this->executeCommand(PHP_BINARY.' misc/update/backfill.php '.$group->name.(isset($group->max) ? (' '.$group->max) : '')); + }, self::ASYNC_BUFFER_SIZE)->then(function ($output) use ($group, &$taskNum) { + echo $output; + $this->colorCli->primary('Task #'.$taskNum.' Backfilled group '.$group->name); + $taskNum--; + })->catch(function (\Throwable $exception) { + echo $exception->getMessage(); + })->catch(static function (SerializableException $serializableException) { + // swallow + }); + } + + $pool->wait(); + } + + public function safeBackfill(): void + { + // make sure short_groups is up-to-date + $this->executeCommand(PHP_BINARY.' misc/update/tmux/bin/update_groups.php'); + + $backfill_qty = (int) Settings::settingValue('backfill_qty'); + $backfill_order = (int) Settings::settingValue('backfill_order'); + $backfill_days = (int) Settings::settingValue('backfill_days'); + $maxMessages = (int) Settings::settingValue('maxmssgs'); + $threads = (int) Settings::settingValue('backfillthreads'); + + $orderby = 'ORDER BY a.last_record ASC'; + switch ($backfill_order) { + case 1: $orderby = 'ORDER BY first_record_postdate DESC'; break; + case 2: $orderby = 'ORDER BY first_record_postdate ASC'; break; + case 3: $orderby = 'ORDER BY name ASC'; break; + case 4: $orderby = 'ORDER BY name DESC'; break; + case 5: $orderby = 'ORDER BY a.last_record DESC'; break; + } + + $backfilldays = '0'; + if ($backfill_days === 1) { + $backfilldays = 'g.backfill_target'; + } elseif ($backfill_days === 2) { + $backfilldays = (string) now()->diffInDays(Carbon::createFromFormat('Y-m-d', Settings::settingValue('safebackfilldate')), true); + } + + $sql = 'SELECT g.name, + g.first_record AS our_first, + MAX(a.first_record) AS their_first, + MAX(a.last_record) AS their_last + FROM usenet_groups g + INNER JOIN short_groups a ON g.name = a.name + WHERE g.first_record IS NOT NULL + AND g.first_record_postdate IS NOT NULL + AND g.backfill = 1 + AND (NOW() - INTERVAL '.$backfilldays.' DAY ) < g.first_record_postdate + GROUP BY a.name, a.last_record, g.name, g.first_record + '.$orderby.' LIMIT 1'; + + $data = DB::select($sql); + + $groupName = ''; + $count = 0; + if (!empty($data) && isset($data[0]->name)) { + $groupName = $data[0]->name; + $count = ($data[0]->our_first - $data[0]->their_first); + } + + if ($count <= 0) { + $this->headerNone(); + if (config('nntmux.echocli') && $groupName !== '') { + $this->colorCli->primary('No backfill needed for group '.$groupName); + } + return; + } + + $getEach = ($count > ($backfill_qty * $threads)) + ? (int) ceil(($backfill_qty * $threads) / $maxMessages) + : (int) ceil($count / $maxMessages); + + $queues = []; + for ($i = 0; $i <= $getEach - 1; $i++) { + $queues[$i] = sprintf('get_range backfill %s %s %s %s', $groupName, $data[0]->our_first - $i * $maxMessages - $maxMessages, $data[0]->our_first - $i * $maxMessages - 1, $i + 1); + } + + $pool = $this->createPool($threads); + + $this->headerStart('safe_backfill', count($queues), $threads); + + foreach ($queues as $queue) { + $pool->add(function () use ($queue) { + return $this->executeCommand($this->buildDnrCommand($queue)); + }, self::ASYNC_BUFFER_SIZE)->then(function ($output) use ($groupName) { + echo $output; + $this->colorCli->primary('Backfilled group '.$groupName); + })->catch(function (\Throwable $exception) { + echo $exception->getMessage(); + })->catch(static function (SerializableException $serializableException) { + // swallow + }); + } + + $pool->wait(); + } +} diff --git a/Blacklight/libraries/Runners/BaseRunner.php b/Blacklight/libraries/Runners/BaseRunner.php new file mode 100644 index 000000000..3d8f746b7 --- /dev/null +++ b/Blacklight/libraries/Runners/BaseRunner.php @@ -0,0 +1,62 @@ +colorCli = $colorCli ?? new ColorCLI(); + } + + protected function createPool(int $concurrency): Pool + { + $concurrency = max(1, $concurrency); + return Pool::create() + ->concurrency($concurrency) + ->timeout(config('nntmux.multiprocessing_max_child_time')); + } + + protected function buildDnrCommand(string $args): string + { + $dnr_path = PHP_BINARY.' misc/update/multiprocessing/.do_not_run/switch.php "php '; + return $dnr_path.$args.'"'; + } + + protected function executeCommand(string $command): string + { + $process = Process::fromShellCommandline($command); + $process->setTimeout(1800); + $process->run(function ($type, $buffer) { + if ($type === Process::ERR) { + echo $buffer; + } + }); + return $process->getOutput(); + } + + protected function headerStart(string $workType, int $count, int $maxProcesses): void + { + if (config('nntmux.echocli')) { + $this->colorCli->header( + 'Multi-processing started at '.now()->toRfc2822String().' for '.$workType.' with '.$count. + ' job(s) to do using a max of '.max(1, $maxProcesses).' child process(es).' + ); + } + } + + protected function headerNone(): void + { + if (config('nntmux.echocli')) { + $this->colorCli->header('No work to do!'); + } + } +} diff --git a/Blacklight/libraries/Runners/BinariesRunner.php b/Blacklight/libraries/Runners/BinariesRunner.php new file mode 100644 index 000000000..40a7b5362 --- /dev/null +++ b/Blacklight/libraries/Runners/BinariesRunner.php @@ -0,0 +1,122 @@ +createPool($maxProcesses); + + $count = count($work); + if ($count > 0) { + $this->headerStart('binaries', $count, $maxProcesses); + } else { + $this->headerNone(); + } + + $taskNum = $count; + foreach ($work as $group) { + $pool->add(function () use ($group) { + return $this->executeCommand(PHP_BINARY.' misc/update/update_binaries.php '.$group->name.' '.$group->max); + }, self::ASYNC_BUFFER_SIZE)->then(function ($output) use ($group, &$taskNum) { + echo $output; + $this->colorCli->primary('Task #'.$taskNum.' Updated group '.$group->name); + $taskNum--; + })->catch(function (\Throwable $exception) { + echo $exception->getMessage(); + })->catch(static function (SerializableException $serializableException) { + // swallow + }); + } + + $pool->wait(); + } + + public function safeBinaries(): void + { + // update group stats + $this->executeCommand(PHP_BINARY.' misc/update/tmux/bin/update_groups.php'); + + $maxHeaders = (int) Settings::settingValue('max_headers_iteration') ?: 1000000; + $maxMessages = (int) Settings::settingValue('maxmssgs'); + $maxProcesses = (int) Settings::settingValue('binarythreads'); + + $groups = DB::select( + ' + SELECT g.name AS groupname, g.last_record AS our_last, + a.last_record AS their_last + FROM usenet_groups g + INNER JOIN short_groups a ON g.active = 1 AND g.name = a.name + ORDER BY a.last_record DESC' + ); + + if (empty($groups)) { + $this->headerNone(); + return; + } + + $i = 1; + $queues = []; + foreach ($groups as $group) { + if ((int) $group->our_last === 0) { + $queues[$i] = sprintf('update_group_headers %s', $group->groupname); + $i++; + } else { + $count = $group->their_last - $group->our_last - 20000; // skip first 20k + if ($count <= $maxMessages * 2) { + $queues[$i] = sprintf('update_group_headers %s', $group->groupname); + $i++; + } else { + $queues[$i] = sprintf('part_repair %s', $group->groupname); + $i++; + $getEach = (int) floor(min($count, $maxHeaders) / $maxMessages); + $remaining = (int) (min($count, $maxHeaders) - $getEach * $maxMessages); + for ($j = 0; $j < $getEach; $j++) { + $queues[$i] = sprintf('get_range binaries %s %s %s %s', $group->groupname, $group->our_last + $j * $maxMessages + 1, $group->our_last + $j * $maxMessages + $maxMessages, $i); + $i++; + } + if ($remaining > 0) { + $start = $group->our_last + $getEach * $maxMessages + 1; + $end = $start + $remaining; + $queues[$i] = sprintf('get_range binaries %s %s %s %s', $group->groupname, $start, $end, $i); + $i++; + } + } + } + } + + $pool = $this->createPool($maxProcesses); + $this->headerStart('safe_binaries', count($queues), $maxProcesses); + + foreach ($queues as $queue) { + preg_match('/alt\..+/i', $queue, $hit); + $pool->add(function () use ($queue) { + return $this->executeCommand($this->buildDnrCommand($queue)); + }, self::ASYNC_BUFFER_SIZE)->then(function ($output) use ($hit) { + if (!empty($hit)) { + echo $output; + $this->colorCli->primary('Updated group '.$hit[0]); + } + })->catch(function (\Throwable $exception) { + echo $exception->getMessage(); + })->catch(static function (SerializableException $serializableException) { + // swallow + }); + } + + $pool->wait(); + } +} diff --git a/Blacklight/libraries/Runners/PostProcessRunner.php b/Blacklight/libraries/Runners/PostProcessRunner.php new file mode 100644 index 000000000..64498033d --- /dev/null +++ b/Blacklight/libraries/Runners/PostProcessRunner.php @@ -0,0 +1,171 @@ +headerNone(); + return; + } + + $pool = $this->createPool($maxProcesses); + $count = count($releases); + $this->headerStart('postprocess: '.$desc, $count, $maxProcesses); + + foreach ($releases as $release) { + $pool->add(function () use ($release, $type) { + return $this->executeCommand(PHP_BINARY.' misc/update/postprocess.php '.$type.$release->id); + }, self::ASYNC_BUFFER_SIZE)->then(function ($output) use (&$count, $desc) { + echo $output; + $this->colorCli->primary('Finished task #'.$count.' for '.$desc); + $count--; + })->catch(function (\Throwable $exception) { + echo $exception->getMessage(); + })->catch(static function (SerializableException $serializableException) { + // swallow + })->timeout(function () use ($desc, &$count) { + $this->colorCli->notice('Task #'.$count.' ('.$desc.'): Timeout occurred.'); + }); + } + + $pool->wait(); + } + + public function processAdditional(): void + { + $ppAddMinSize = Settings::settingValue('minsizetopostprocess') !== '' ? (int) Settings::settingValue('minsizetopostprocess') : 1; + $ppAddMinSize = ($ppAddMinSize > 0 ? ('AND r.size > '.($ppAddMinSize * 1048576)) : ''); + $ppAddMaxSize = (Settings::settingValue('maxsizetopostprocess') !== '') ? (int) Settings::settingValue('maxsizetopostprocess') : 100; + $ppAddMaxSize = ($ppAddMaxSize > 0 ? ('AND r.size < '.($ppAddMaxSize * 1073741824)) : ''); + + $sql = ' + SELECT r.leftguid AS id + FROM releases r + LEFT JOIN categories c ON c.id = r.categories_id + WHERE r.nzbstatus = '.NZB::NZB_ADDED.' + AND r.passwordstatus = -1 + AND r.haspreview = -1 + AND c.disablepreview = 0 + '.$ppAddMaxSize.' '.$ppAddMinSize.' + GROUP BY r.leftguid + LIMIT 16'; + $queue = DB::select($sql); + + $maxProcesses = (int) Settings::settingValue('postthreads'); + $this->runPostProcess($queue, $maxProcesses, 'additional true ', 'additional postprocessing'); + } + + public function processNfo(): void + { + if ((int) Settings::settingValue('lookupnfo') !== 1) { + $this->headerNone(); + return; + } + + $nfoQuery = \Blacklight\Nfo::NfoQueryString(); + + $checkSql = 'SELECT r.id FROM releases r WHERE 1=1 '.$nfoQuery.' LIMIT 1'; + if (count(DB::select($checkSql)) === 0) { + $this->headerNone(); + return; + } + + $sql = ' + SELECT r.leftguid AS id + FROM releases r + WHERE 1=1 '.$nfoQuery.' + GROUP BY r.leftguid + LIMIT 16'; + $queue = DB::select($sql); + + $maxProcesses = (int) Settings::settingValue('nfothreads'); + $this->runPostProcess($queue, $maxProcesses, 'nfo true ', 'nfo postprocessing'); + } + + public function processMovies(bool $renamedOnly): void + { + if ((int) Settings::settingValue('lookupimdb') <= 0) { + $this->headerNone(); + return; + } + + $condLookup = ((int) Settings::settingValue('lookupimdb') === 2 ? 'AND isrenamed = 1' : ''); + $condRenamedOnly = ($renamedOnly ? 'AND isrenamed = 1' : ''); + + $checkSql = ' + SELECT id + FROM releases + WHERE categories_id BETWEEN 2000 AND 2999 + AND nzbstatus = '.NZB::NZB_ADDED.' + AND imdbid IS NULL + '.$condLookup.' '.$condRenamedOnly.' + LIMIT 1'; + if (count(DB::select($checkSql)) === 0) { + $this->headerNone(); + return; + } + + $renamedFlag = ($renamedOnly ? 2 : 1); + $sql = ' + SELECT leftguid AS id, '.$renamedFlag.' AS renamed + FROM releases + WHERE categories_id BETWEEN 2000 AND 2999 + AND nzbstatus = '.NZB::NZB_ADDED.' + AND imdbid IS NULL + '.$condLookup.' '.$condRenamedOnly.' + GROUP BY leftguid + LIMIT 16'; + $queue = DB::select($sql); + + $maxProcesses = (int) Settings::settingValue('postthreadsnon'); + $this->runPostProcess($queue, $maxProcesses, 'movies true ', 'movies postprocessing'); + } + + public function processTv(bool $renamedOnly): void + { + if ((int) Settings::settingValue('lookuptv') <= 0) { + $this->headerNone(); + return; + } + + $condLookup = ((int) Settings::settingValue('lookuptv') === 2 ? 'AND isrenamed = 1' : ''); + $condRenamedOnly = ($renamedOnly ? 'AND isrenamed = 1' : ''); + + $checkSql = ' + SELECT id + FROM releases + WHERE categories_id BETWEEN 5000 AND 5999 + AND nzbstatus = '.NZB::NZB_ADDED.' + AND size > 1048576 + AND tv_episodes_id BETWEEN -2 AND 0 + '.$condLookup.' '.$condRenamedOnly; + if (count(DB::select($checkSql)) === 0) { + $this->headerNone(); + return; + } + + $renamedFlag = ($renamedOnly ? 2 : 1); + $sql = ' + SELECT leftguid AS id, '.$renamedFlag.' AS renamed + FROM releases + WHERE categories_id BETWEEN 5000 AND 5999 + AND nzbstatus = '.NZB::NZB_ADDED.' + AND tv_episodes_id BETWEEN -2 AND 0 + AND size > 1048576 + '.$condLookup.' '.$condRenamedOnly.' + GROUP BY leftguid + LIMIT 16'; + $queue = DB::select($sql); + + $maxProcesses = (int) Settings::settingValue('postthreadsnon'); + $this->runPostProcess($queue, $maxProcesses, 'tv true ', 'tv postprocessing'); + } +} diff --git a/Blacklight/libraries/Runners/ReleasesRunner.php b/Blacklight/libraries/Runners/ReleasesRunner.php new file mode 100644 index 000000000..a488e7112 --- /dev/null +++ b/Blacklight/libraries/Runners/ReleasesRunner.php @@ -0,0 +1,136 @@ +id)); + if (!empty($query)) { + $uGroups[] = ['id' => $group->id, 'name' => $group->name]; + } + } catch (\PDOException $e) { + if (config('app.debug') === true) { + Log::debug($e->getMessage()); + } + } + } + + $pool = $this->createPool($maxProcesses); + + $count = count($uGroups); + if ($count > 0) { + $this->headerStart('releases', $count, $maxProcesses); + } else { + $this->headerNone(); + } + + $taskNum = $count; + foreach ($uGroups as $group) { + $pool->add(function () use ($group) { + return $this->executeCommand($this->buildDnrCommand('releases '.$group['id'])); + }, self::ASYNC_BUFFER_SIZE)->then(function ($output) use (&$taskNum) { + echo $output; + $this->colorCli->primary('Task #'.$taskNum.' Finished performing release processing'); + $taskNum--; + })->catch(function (\Throwable $exception) { + echo $exception->getMessage(); + })->catch(static function (SerializableException $serializableException) { + // swallow + }); + } + + $pool->wait(); + } + + public function updatePerGroup(): void + { + $groups = DB::select('SELECT id , name FROM usenet_groups WHERE (active = 1 OR backfill = 1)'); + $maxProcesses = (int) Settings::settingValue('releasethreads'); + + $pool = $this->createPool($maxProcesses); + $this->headerStart('update_per_group', count($groups), $maxProcesses); + + foreach ($groups as $group) { + $pool->add(function () use ($group) { + return $this->executeCommand($this->buildDnrCommand('update_per_group '.$group->id)); + }, self::ASYNC_BUFFER_SIZE)->then(function ($output) use ($group) { + echo $output; + $name = UsenetGroup::getNameByID($group->id); + $this->colorCli->primary('Finished updating binaries, processing releases and additional postprocessing for group:'.$name); + })->catch(function (\Throwable $exception) { + echo $exception->getMessage(); + })->catch(static function (SerializableException $serializableException) { + echo $serializableException->asThrowable()->getMessage(); + }); + } + + $pool->wait(); + } + + public function fixRelNames(string $mode, int $maxPerRun, int $maxThreads): void + { + $maxThreads = max(1, min(16, $maxThreads)); + + $leftGuids = ['0','1','2','3','4','5','6','7','8','9','a','b','c','d','e','f']; + + if ($mode === 'predbft') { + $preCount = DB::select( + "SELECT COUNT(p.id) AS num FROM predb p WHERE LENGTH(p.title) >= 15 AND p.title NOT REGEXP '[\"\<\> ]' AND p.searched = 0 AND p.predate < (NOW() - INTERVAL 1 DAY)" + ); + if (!empty($preCount) && (int) $preCount[0]->num > 0 && $maxPerRun > 0) { + $leftGuids = \array_slice($leftGuids, 0, (int) ceil($preCount[0]->num / $maxPerRun)); + } else { + $leftGuids = []; + } + } + + $queues = []; + $idx = 0; + foreach ($leftGuids as $leftGuid) { + if ($maxPerRun > 0) { + $idx++; + $queues[$idx] = sprintf('%s %s %s %s', $mode, $leftGuid, $maxPerRun, $idx); + } + } + + $pool = $this->createPool($maxThreads); + + $count = count($queues); + if ($count > 0) { + $this->headerStart('fixRelNames_'.$mode, $count, $maxThreads); + } else { + $this->headerNone(); + } + + $taskNum = $count; + foreach ($queues as $queue) { + $pool->add(function () use ($queue) { + return $this->executeCommand(PHP_BINARY.' misc/update/tmux/bin/groupfixrelnames.php "'.$queue.'" true'); + }, self::ASYNC_BUFFER_SIZE)->then(function ($output) use (&$taskNum) { + echo $output; + $this->colorCli->primary('Task #'.$taskNum.' Finished fixing releases names'); + $taskNum--; + })->catch(function (\Throwable $exception) { + echo $exception->getMessage(); + })->catch(static function (SerializableException $serializableException) { + // swallow + }); + } + + $pool->wait(); + } +} diff --git a/Blacklight/processing/ProcessReleases.php b/Blacklight/processing/ProcessReleases.php index ba37dfe78..cd85cbffa 100755 --- a/Blacklight/processing/ProcessReleases.php +++ b/Blacklight/processing/ProcessReleases.php @@ -756,7 +756,7 @@ class ProcessReleases if (! empty($groupID)) { $collectionsCheck->where('collections.groups_id', $groupID); } - $collectionsCheck->groupBy('binaries.collections_id', 'collections.totalfiles', 'collections.id') + $collectionsCheck->groupBy(['binaries.collections_id', 'collections.totalfiles', 'collections.id']) ->havingRaw('COUNT(binaries.id) IN (collections.totalfiles, collections.totalfiles+1)'); Collection::query()->joinSub($collectionsCheck, 'r', function ($join) { @@ -789,7 +789,7 @@ class ProcessReleases if (! empty($groupID)) { $collectionsCheck->where('collections.groups_id', $groupID); } - $collectionsCheck->groupBy('collections.id'); + $collectionsCheck->groupBy(['collections.id']); Collection::query()->joinSub($collectionsCheck, 'r', function ($join) { $join->on('collections.id', '=', 'r.id'); @@ -928,28 +928,76 @@ class ProcessReleases { $lastRun = Settings::settingValue('last_run_time'); - DB::transaction(function () use ($groupID, $lastRun) { + // Compute cutoff timestamp once. + $threshold = null; + try { + if (! empty($lastRun)) { + $threshold = \Illuminate\Support\Carbon::createFromFormat('Y-m-d H:i:s', $lastRun); + } + } catch (\Throwable $e) { $threshold = null; - try { - if (! empty($lastRun)) { - $threshold = Carbon::createFromFormat('Y-m-d H:i:s', $lastRun); + } + if ($threshold === null) { + $threshold = now(); + } + $cutoff = $threshold->copy()->subHours($this->collectionTimeout); + + $totalDeleted = 0; + $maxRetries = 5; + + // Delete in small batches using a single-statement DELETE via nested subselect to avoid "record changed since last read". + do { + $affected = 0; + $attempt = 0; + do { + try { + if (! empty($groupID)) { + $affected = DB::affectingStatement( + 'DELETE FROM collections WHERE id IN ( + SELECT id FROM ( + SELECT id FROM collections WHERE added < ? AND groups_id = ? ORDER BY id LIMIT 500 + ) AS x + )', + [$cutoff, $groupID] + ); + } else { + $affected = DB::affectingStatement( + 'DELETE FROM collections WHERE id IN ( + SELECT id FROM ( + SELECT id FROM collections WHERE added < ? ORDER BY id LIMIT 500 + ) AS x + )', + [$cutoff] + ); + } + break; // success + } catch (\Throwable $e) { + $attempt++; + if ($attempt >= $maxRetries) { + if ($this->echoCLI) { + $this->colorCLI->error('Stuck collections delete failed after retries: '.$e->getMessage()); + } + break; + } + // Exponential backoff to ease contention + usleep(20000 * $attempt); } - } catch (\Throwable $e) { - $threshold = null; + } while (true); + + $totalDeleted += $affected; + + if ($affected < 500) { + break; } - if ($threshold === null) { - $threshold = now(); - } - $objQuery = Collection::query() - ->where('added', '<', $threshold->subHours($this->collectionTimeout)); - if (! empty($groupID)) { - $objQuery->where('groups_id', $groupID); - } - $obj = $objQuery->delete(); - if ($this->echoCLI && $obj > 0) { - $this->colorCLI->primary('Deleted '.$obj.' broken/stuck collections.', true); - } - }, 10); + + + // Yield briefly to reduce contention in busy systems. + usleep(10000); + } while (true); + + if ($this->echoCLI && $totalDeleted > 0) { + $this->colorCLI->primary('Deleted '.$totalDeleted.' broken/stuck collections.', true); + } } /** diff --git a/app/Services/CollectionCleanupService.php b/app/Services/CollectionCleanupService.php index bc67c206d..2b1894dfb 100644 --- a/app/Services/CollectionCleanupService.php +++ b/app/Services/CollectionCleanupService.php @@ -33,81 +33,111 @@ class CollectionCleanupService ), true); } - DB::transaction(function () use (&$deletedCount, $startTime, $echoCLI) { + // Batch-delete old collections using a safe id-subselect to avoid read-then-delete races. + $cutoff = now()->subHours(Settings::settingValue('partretentionhours')); + $batchDeleted = 0; + $maxRetries = 5; + do { + $affected = 0; + $attempt = 0; + do { + try { + // Delete by id list derived in a nested subquery to avoid "Record has changed since last read". + $affected = DB::affectingStatement( + 'DELETE FROM collections WHERE id IN ( + SELECT id FROM ( + SELECT id FROM collections WHERE dateadded < ? ORDER BY id LIMIT 500 + ) AS x + )', + [$cutoff] + ); + break; // success + } catch (\Throwable $e) { + // Retry on lock/timeout errors + $attempt++; + if ($attempt >= $maxRetries) { + if ($echoCLI) { + $this->colorCLI->error('Cleanup delete failed after retries: '.$e->getMessage()); + } + break; + } + usleep(20000 * $attempt); + } + } while (true); + + $batchDeleted += $affected; + if ($affected < 500) { + break; + } + // Brief pause to reduce pressure on the lock manager in busy systems. + usleep(10000); + } while (true); + + $deletedCount += $batchDeleted; + + if ($echoCLI) { + $elapsed = now()->diffInSeconds($startTime, true); + $this->colorCLI->primary( + 'Finished deleting '.$batchDeleted.' old collections/binaries/parts in '. + $elapsed.Str::plural(' second', $elapsed), + true + ); + } + + // Occasionally prune CBP orphans (low frequency to avoid heavy load). + if (random_int(0, 200) <= 1) { + if ($echoCLI) { + echo $this->colorCLI->header('Process Releases -> Remove CBP orphans.'.PHP_EOL). + $this->colorCLI->primary('Deleting orphaned collections.'); + } + $deleted = 0; + // NOTE: This JOIN DELETE can be heavy; consider batching if it becomes an issue in practice. $deleteQuery = Collection::query() - ->where('dateadded', '<', now()->subHours(Settings::settingValue('partretentionhours'))) + ->whereNull('binaries.id') + ->orWhereNull('parts.binaries_id') + ->leftJoin('binaries', 'collections.id', '=', 'binaries.collections_id') + ->leftJoin('parts', 'binaries.id', '=', 'parts.binaries_id') ->delete(); + if ($deleteQuery > 0) { $deleted = $deleteQuery; $deletedCount += $deleted; } - $firstQuery = $fourthQuery = now(); - $totalTime = $firstQuery->diffInSeconds($startTime, true); + $totalTime = now()->diffInSeconds($startTime); if ($echoCLI) { - $this->colorCLI->primary( - 'Finished deleting '.$deleted.' old collections/binaries/parts in '. - $totalTime.Str::plural(' second', $totalTime), - true - ); + $this->colorCLI->primary('Finished deleting '.$deleted.' orphaned collections in '.$totalTime.Str::plural(' second', $totalTime), true); } + } - if (random_int(0, 200) <= 1) { - if ($echoCLI) { - echo $this->colorCLI->header('Process Releases -> Remove CBP orphans.'.PHP_EOL). - $this->colorCLI->primary('Deleting orphaned collections.'); - } + if ($echoCLI) { + $this->colorCLI->primary('Deleting collections that were missed after NZB creation.', true); + } - $deleted = 0; - $deleteQuery = Collection::query() - ->whereNull('binaries.id') - ->orWhereNull('parts.binaries_id') - ->leftJoin('binaries', 'collections.id', '=', 'binaries.collections_id') - ->leftJoin('parts', 'binaries.id', '=', 'parts.binaries_id') - ->delete(); + $deleted = 0; + $collections = Collection::query() + ->where('releases.nzbstatus', '=', 1) + ->leftJoin('releases', 'releases.id', '=', 'collections.releases_id') + ->select('collections.id') + ->get(); - if ($deleteQuery > 0) { - $deleted = $deleteQuery; - $deletedCount += $deleted; - } + foreach ($collections as $collection) { + $deleted++; + Collection::query()->where('id', $collection->id)->delete(); + } + $deletedCount += $deleted; - $totalTime = now()->diffInSeconds($firstQuery); + $totalTime = now()->diffInSeconds($startTime, true); - if ($echoCLI) { - $this->colorCLI->primary('Finished deleting '.$deleted.' orphaned collections in '.$totalTime.Str::plural(' second', $totalTime), true); - } - } - - if ($echoCLI) { - $this->colorCLI->primary('Deleting collections that were missed after NZB creation.', true); - } - - $deleted = 0; - $collections = Collection::query() - ->where('releases.nzbstatus', '=', 1) - ->leftJoin('releases', 'releases.id', '=', 'collections.releases_id') - ->select('collections.id') - ->get(); - - foreach ($collections as $collection) { - $deleted++; - Collection::query()->where('id', $collection->id)->delete(); - } - $deletedCount += $deleted; - - $colDelTime = now()->diffInSeconds($fourthQuery, true); - $totalTime = $fourthQuery->diffInSeconds($startTime, true); - - if ($echoCLI) { - $this->colorCLI->primary( - 'Finished deleting '.$deleted.' collections missed after NZB creation in '.$colDelTime.Str::plural(' second', $colDelTime). - PHP_EOL.'Removed '.number_format($deletedCount).' parts/binaries/collection rows in '.$totalTime.Str::plural(' second', $totalTime), - true - ); - } - }, 10); + if ($echoCLI) { + $this->colorCLI->primary( + 'Finished deleting '.$deleted.' collections missed after NZB creation in '.($totalTime).Str::plural(' second', $totalTime). + PHP_EOL.'Removed '.number_format($deletedCount).' parts/binaries/collection rows in '.$totalTime.Str::plural(' second', $totalTime), + true + ); + } return $deletedCount; } diff --git a/database/migrations/2025_08_14_000001_add_indexes_to_collections_table.php b/database/migrations/2025_08_14_000001_add_indexes_to_collections_table.php new file mode 100644 index 000000000..53241a67a --- /dev/null +++ b/database/migrations/2025_08_14_000001_add_indexes_to_collections_table.php @@ -0,0 +1,25 @@ +index(['groups_id', 'added'], 'collections_groups_added_idx'); + // Create index for dateadded used by old-collection cleanup. + $table->index(['dateadded'], 'collections_dateadded_idx'); + }); + } + + public function down(): void + { + Schema::table('collections', function (Blueprint $table) { + $table->dropIndex('collections_groups_added_idx'); + $table->dropIndex('collections_dateadded_idx'); + }); + } +};