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

Split concerns into services

Browse Source
This commit is contained in:
DariusIII
2025-08-15 11:45:27 +02:00
parent b2d632cb24
commit 607229a1b1
9 changed files with 891 additions and 751 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Blacklight/libraries/Forking.php
+83 -671
View File
@@ -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
);
Blacklight/libraries/Runners/BackfillRunner.php
+134
View File
@@ -0,0 +1,134 @@
<?php
namespace Blacklight\libraries\Runners;
use App\Models\Settings;
use Illuminate\Support\Carbon;
use Illuminate\Support\Facades\DB;
use Spatie\Async\Output\SerializableException;
class BackfillRunner extends BaseRunner
{
public function backfill(array $options = []): void
{
$select = 'SELECT name';
if (($options[0] ?? false) !== false) {
$select .= ', '.$options[0].' AS max';
}
$select .= ' FROM usenet_groups WHERE backfill = 1';
$work = DB::select($select);
$maxProcesses = (int) Settings::settingValue('backfillthreads');
$pool = $this->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();
}
}
Blacklight/libraries/Runners/BaseRunner.php
+62
View File
@@ -0,0 +1,62 @@
<?php
namespace Blacklight\libraries\Runners;
use Blacklight\ColorCLI;
use Spatie\Async\Pool;
use Symfony\Component\Process\Process;
abstract class BaseRunner
{
protected ColorCLI $colorCli;
protected const ASYNC_BUFFER_SIZE = 2000000;
public function __construct(?ColorCLI $colorCli = null)
{
$this->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!');
}
}
}
Blacklight/libraries/Runners/BinariesRunner.php
+122
View File
@@ -0,0 +1,122 @@
<?php
namespace Blacklight\libraries\Runners;
use App\Models\Settings;
use Illuminate\Support\Facades\DB;
use Spatie\Async\Output\SerializableException;
class BinariesRunner extends BaseRunner
{
public function binaries(int $maxPerGroup): void
{
$work = DB::select(
sprintf(
'SELECT name, %d AS max FROM usenet_groups WHERE active = 1',
$maxPerGroup
)
);
$maxProcesses = (int) Settings::settingValue('binarythreads');
$pool = $this->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();
}
}
Blacklight/libraries/Runners/PostProcessRunner.php
+171
View File
@@ -0,0 +1,171 @@
<?php
namespace Blacklight\libraries\Runners;
use App\Models\Settings;
use Blacklight\NZB;
use Spatie\Async\Output\SerializableException;
use Illuminate\Support\Facades\DB;
class PostProcessRunner extends BaseRunner
{
private function runPostProcess(array $releases, int $maxProcesses, string $type, string $desc): void
{
if (empty($releases)) {
$this->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');
}
}
Blacklight/libraries/Runners/ReleasesRunner.php
+136
View File
@@ -0,0 +1,136 @@
<?php
namespace Blacklight\libraries\Runners;
use App\Models\Settings;
use App\Models\UsenetGroup;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use Spatie\Async\Output\SerializableException;
class ReleasesRunner extends BaseRunner
{
public function releases(): void
{
$groups = DB::select('SELECT id, name FROM usenet_groups WHERE (active = 1 OR backfill = 1)');
$maxProcesses = (int) Settings::settingValue('releasethreads');
$uGroups = [];
foreach ($groups as $group) {
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];
}
} 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();
}
}
Blacklight/processing/ProcessReleases.php
+69 -21
View File
@@ -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);
}
}
/**
app/Services/CollectionCleanupService.php
+89 -59
View File
@@ -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;
}
database/migrations/2025_08_14_000001_add_indexes_to_collections_table.php
+25
View File
@@ -0,0 +1,25 @@
<?php
use Illuminate\Database\Migrations\Migration;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\Schema;
return new class extends Migration {
public function up(): void
{
Schema::table('collections', function (Blueprint $table) {
// Create composite index for (groups_id, added) used by stuck-collection cleanup.
$table->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');
});
}
};