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

Update additional postprocessing

Browse Source
This commit is contained in:
DariusIII
2026-08-04 01:00:24 +02:00
parent de9007c9a5
commit d384f3d60d
12 changed files with 403 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/Console/Commands/PostProcessGuid.php
+21 -4
View File
@@ -22,7 +22,9 @@ class PostProcessGuid extends Command
protected $signature = 'postprocess:guid
{type : Type: additional, nfo, movie, tv, anime, books, music, console, or games}
{guid : First character of release guid (a-f, 0-9)}
{renamed? : For movie/tv: process renamed only (optional)}';
{renamed? : For movie/tv: process renamed only (optional)}
{--worker : Drain multiple additional-processing batches}
{--max-batches=4 : Maximum claim batches for worker mode}';
/**
* The console command description.
@@ -55,7 +57,7 @@ class PostProcessGuid extends Command
try {
match ($type) {
'additional' => $this->processAdditional($guid),
'additional' => $this->processAdditional($guid, (bool) $this->option('worker')),
'nfo' => $this->processNfo($guid),
'movie' => $this->postProcessService->processMovies('', $guid, $renamed),
'tv' => $this->postProcessService->processTv('', $guid, $renamed),
@@ -81,10 +83,25 @@ class PostProcessGuid extends Command
/**
* Process additional data for releases.
*/
private function processAdditional(string $guid): void
private function processAdditional(string $guid, bool $workerMode): void
{
$maxBatches = $workerMode ? max(1, (int) $this->option('max-batches')) : 1;
$workerToken = bin2hex(random_bytes(16));
$childTimeout = (int) config('nntmux.multiprocessing_max_child_time', 1800);
$deadline = microtime(true) + max(1, $childTimeout - 30);
$excludedReleaseIds = [];
try {
$this->additionalProcessor->start('', $guid);
for ($batch = 0; $batch < $maxBatches && microtime(true) < $deadline; $batch++) {
$stats = $this->additionalProcessor->start('', $guid, $workerToken, $excludedReleaseIds);
if ($stats['claimed'] === 0) {
break;
}
$excludedReleaseIds = array_values(array_unique([
...$excludedReleaseIds,
...$stats['claimed_ids'],
]));
}
} finally {
$this->additionalProcessor->finish();
}
app/Services/AdditionalProcessing/AdditionalCandidateQuery.php
+91 -12
View File
@@ -161,19 +161,87 @@ final class AdditionalCandidateQuery
$effectiveLimit = $limit !== null && $limit > 0
? min($limit, self::BUCKET_LIMIT)
: self::BUCKET_LIMIT;
$rows = self::baseBuilder()
->select(DB::raw('DISTINCT r.leftguid AS guid_bucket'))
->limit($effectiveLimit)
->get();
$chars = [];
foreach ($rows as $row) {
$id = strtolower((string) ($row->guid_bucket ?? ''));
if ($id !== '') {
$chars[] = substr($id, 0, 1);
return array_slice(
array_column(self::availableBucketCounts(), 'bucket'),
0,
$effectiveLimit,
);
}
/**
* Return available candidate counts keyed by GUID bucket.
*
* @return list<array{bucket: string, count: int}>
*/
public static function availableBucketCounts(): array
{
$counts = [];
foreach (self::bucketBacklog() as $backlog) {
if ($backlog['available'] > 0) {
$counts[] = [
'bucket' => $backlog['bucket'],
'count' => $backlog['available'],
];
}
}
return array_values(array_unique($chars));
return $counts;
}
/**
* Return total and currently claimable candidates for every GUID bucket.
*
* @return list<array{bucket: string, total: int, available: int}>
*/
public static function bucketBacklog(): array
{
$query = self::baseBuilder(includeClaimed: true)
->select('r.leftguid')
->selectRaw('COUNT(*) AS total_count')
->groupBy('r.leftguid')
->orderBy('r.leftguid');
if (self::supportsClaims()) {
$query->selectRaw(
'SUM(CASE WHEN r.'.self::CLAIMED_AT_COLUMN.' IS NULL OR r.'.self::CLAIMED_AT_COLUMN.' < ? THEN 1 ELSE 0 END) AS available_count',
[self::claimStaleBefore()],
);
} else {
$query->selectRaw('COUNT(*) AS available_count');
}
$backlog = [];
foreach ($query->get() as $row) {
$bucket = strtolower(substr((string) ($row->leftguid ?? ''), 0, 1));
if ($bucket === '') {
continue;
}
$backlog[] = [
'bucket' => $bucket,
'total' => (int) ($row->total_count ?? 0),
'available' => (int) ($row->available_count ?? 0),
];
}
return $backlog;
}
/**
* @return array{total: int, available: int}
*/
public static function backlogCounts(): array
{
$counts = ['total' => 0, 'available' => 0];
foreach (self::bucketBacklog() as $backlog) {
$counts['total'] += $backlog['total'];
$counts['available'] += $backlog['available'];
}
return $counts;
}
/**
@@ -189,6 +257,7 @@ final class AdditionalCandidateQuery
* Claim a bounded batch of release rows for one worker.
*
* @param list<string> $columns
* @param list<int> $excludedReleaseIds
* @return EloquentCollection<int, Release>
*/
public static function claimBatch(
@@ -199,10 +268,11 @@ final class AdditionalCandidateQuery
?int $minSizeMB = null,
?int $maxSizeGB = null,
array $columns = ['*'],
array $excludedReleaseIds = [],
): EloquentCollection {
$effectiveLimit = max(1, $limit);
return DB::transaction(function () use ($guidChar, $effectiveLimit, $token, $groupID, $minSizeMB, $maxSizeGB, $columns): EloquentCollection {
return DB::transaction(function () use ($guidChar, $effectiveLimit, $token, $groupID, $minSizeMB, $maxSizeGB, $columns, $excludedReleaseIds): EloquentCollection {
$supportsClaims = self::supportsClaims();
$query = self::baseBuilder($groupID, $guidChar, $minSizeMB, $maxSizeGB)
->select('r.id')
@@ -210,6 +280,10 @@ final class AdditionalCandidateQuery
->orderBy('r.id')
->limit($effectiveLimit);
if ($excludedReleaseIds !== []) {
$query->whereNotIn('r.id', $excludedReleaseIds);
}
if (DB::getDriverName() !== 'sqlite') {
$query->lockForUpdate();
}
@@ -295,7 +369,7 @@ final class AdditionalCandidateQuery
return;
}
$staleBefore = now()->subSeconds(self::claimTtlSeconds());
$staleBefore = self::claimStaleBefore();
$query->where(function (Builder $claimQuery) use ($staleBefore): void {
$claimQuery
@@ -311,6 +385,11 @@ final class AdditionalCandidateQuery
return max(300, $timeout * 2);
}
private static function claimStaleBefore(): \Illuminate\Support\Carbon
{
return now()->subSeconds(self::claimTtlSeconds());
}
/**
* @param list<string> $columns
* @return list<string>
app/Services/AdditionalProcessing/AdditionalProcessingOrchestrator.php
+43 -11
View File
@@ -41,21 +41,31 @@ class AdditionalProcessingOrchestrator
/**
* Start the additional processing.
*
* @param list<int> $excludedReleaseIds
* @return array{claimed: int, processed: int, failed: int, claimed_ids: list<int>}
*
* @throws Exception
*/
public function start(string $groupID = '', string $guidChar = ''): void
{
public function start(
string $groupID = '',
string $guidChar = '',
string $workerToken = '',
array $excludedReleaseIds = []
): array {
$this->finish();
if (! $this->setupTempPath($guidChar, $groupID)) {
return;
if (! $this->setupTempPath($guidChar, $groupID, $workerToken)) {
return ['claimed' => 0, 'processed' => 0, 'failed' => 0, 'claimed_ids' => []];
}
$this->fetchReleases($groupID, $guidChar);
$this->fetchReleases($groupID, $guidChar, $excludedReleaseIds);
if ($this->totalReleases > 0) {
$this->output->echoDescription($this->totalReleases);
$this->processReleases($guidChar);
return $this->processReleases($guidChar);
}
return ['claimed' => 0, 'processed' => 0, 'failed' => 0, 'claimed_ids' => []];
}
/**
@@ -95,13 +105,14 @@ class AdditionalProcessingOrchestrator
/**
* Set up the main temp path.
*/
private function setupTempPath(string $guidChar, string $groupID): bool
private function setupTempPath(string $guidChar, string $groupID, string $workerToken = ''): bool
{
try {
$this->mainTmpPath = $this->tempWorkspace->ensureMainTempPath(
$this->config->tmpUnrarPath,
$guidChar,
$groupID
$groupID,
$workerToken !== '' ? $workerToken : bin2hex(random_bytes(16)),
);
$this->tempWorkspace->clearDirectory($this->mainTmpPath, true);
} catch (\Throwable $e) {
@@ -129,7 +140,10 @@ class AdditionalProcessingOrchestrator
* PostProcessRunner::processAdditional(). Do NOT inline new predicates
* here; add them to AdditionalCandidateQuery instead.
*/
private function fetchReleases(int|string $groupID, string $guidChar): void
/**
* @param list<int> $excludedReleaseIds
*/
private function fetchReleases(int|string $groupID, string $guidChar, array $excludedReleaseIds = []): void
{
$this->claimToken = bin2hex(random_bytes(16));
$this->releases = AdditionalCandidateQuery::claimBatch(
@@ -154,6 +168,7 @@ class AdditionalProcessingOrchestrator
'pp_timeout_count',
AdditionalCandidateQuery::CLAIM_TOKEN_COLUMN,
],
$excludedReleaseIds,
);
$this->totalReleases = $this->releases->count();
}
@@ -168,10 +183,18 @@ class AdditionalProcessingOrchestrator
* every subsequent cycle, and the "needs additional pp" backlog would
* grow indefinitely without any visible failure.
*
* @return array{claimed: int, processed: int, failed: int, claimed_ids: list<int>}
*
* @throws Exception
*/
private function processReleases(string $guidChar = ''): void
private function processReleases(string $guidChar = ''): array
{
$startedAt = microtime(true);
$claimedIds = $this->releases
->pluck('id')
->map(static fn (mixed $id): int => (int) $id)
->values()
->all();
$processed = 0;
$failed = 0;
@@ -202,15 +225,24 @@ class AdditionalProcessingOrchestrator
'picked' => $this->totalReleases,
'processed' => $processed,
'failed' => $failed,
'elapsed_seconds' => round(microtime(true) - $startedAt, 3),
'peak_memory_bytes' => memory_get_peak_usage(true),
]);
$this->output->endOutput();
return [
'claimed' => $this->totalReleases,
'processed' => $processed,
'failed' => $failed,
'claimed_ids' => $claimedIds,
];
}
public function finish(): void
{
if ($this->mainTmpPath !== '') {
$this->tempWorkspace->clearDirectory($this->mainTmpPath, true);
$this->tempWorkspace->clearDirectory($this->mainTmpPath, false);
$this->mainTmpPath = '';
}
app/Services/Runners/PostProcessRunner.php
+63 -11
View File
@@ -16,6 +16,8 @@ use Illuminate\Support\Facades\Log;
class PostProcessRunner extends BaseRunner
{
private const int ADDITIONAL_WORKER_MAX_BATCHES = 4;
private function guidBucketExpression(string $column = 'leftguid'): string
{
return DB::getDriverName() === 'sqlite'
@@ -43,7 +45,11 @@ class PostProcessRunner extends BaseRunner
// id may already be a single GUID bucket char; if not, take first char defensively
$char = isset($release->id) ? substr((string) $release->id, 0, 1) : '';
// Use postprocess:guid command which accepts the GUID character
$commands[] = PHP_BINARY.' artisan postprocess:guid '.$type.' '.$char;
$command = PHP_BINARY.' artisan postprocess:guid '.$type.' '.$char;
if ($type === 'additional') {
$command .= ' --worker --max-batches='.self::ADDITIONAL_WORKER_MAX_BATCHES;
}
$commands[] = $command;
}
$this->runStreamingCommands($commands, $maxProcesses, $desc); // @phpstan-ignore argument.type
@@ -221,16 +227,18 @@ class PostProcessRunner extends BaseRunner
// Bucket-selection predicates and size filters are owned by
// AdditionalCandidateQuery so they cannot drift away from the
// per-worker fetch in AdditionalProcessingOrchestrator::fetchReleases().
$chars = AdditionalCandidateQuery::bucketChars();
$bucketCounts = AdditionalCandidateQuery::availableBucketCounts();
$chars = array_column($bucketCounts, 'bucket');
$maxProcesses = (int) Settings::settingValue('postthreads');
$queryLimit = (int) (Settings::settingValue('maxaddprocessed') ?: 25);
// Normalize to the shape the rest of runPostProcess() expects:
// an array of objects with an `id` (first GUID char) property. If the
// backlog is concentrated in fewer buckets than configured threads,
// repeat buckets so claimBatch() can split one hot bucket across
// multiple workers.
$queue = $this->additionalQueue($chars, $maxProcesses);
$queue = $this->additionalQueue($bucketCounts, $maxProcesses, $queryLimit);
if (! $this->prepareAdditionalTempBuckets($chars)) {
return;
}
@@ -245,8 +253,19 @@ class PostProcessRunner extends BaseRunner
$this->headerStart('postprocess: additional postprocessing', count($queue), 1);
$orchestrator = app(AdditionalProcessingOrchestrator::class);
foreach ($chars as $char) {
$workerToken = bin2hex(random_bytes(16));
$excludedReleaseIds = [];
try {
$orchestrator->start('', $char);
for ($batch = 0; $batch < self::ADDITIONAL_WORKER_MAX_BATCHES; $batch++) {
$stats = $orchestrator->start('', $char, $workerToken, $excludedReleaseIds);
if ($stats['claimed'] === 0) {
break;
}
$excludedReleaseIds = array_values(array_unique([
...$excludedReleaseIds,
...$stats['claimed_ids'],
]));
}
} finally {
$orchestrator->finish();
}
@@ -260,20 +279,46 @@ class PostProcessRunner extends BaseRunner
}
/**
* @param list<string> $chars
* @param list<array{bucket: string, count: int}> $bucketCounts
* @return list<object{id: string}>
*/
private function additionalQueue(array $chars, int $maxProcesses): array
private function additionalQueue(array $bucketCounts, int $maxProcesses, int $queryLimit): array
{
if ($chars === []) {
if ($bucketCounts === []) {
return [];
}
$chars = array_column($bucketCounts, 'bucket');
$queue = array_map(static fn (string $c): object => (object) ['id' => $c], $chars);
$target = max(count($queue), $maxProcesses);
$allocations = array_fill(0, count($bucketCounts), 1);
$demand = [];
foreach ($bucketCounts as $index => $bucketCount) {
// Demand is intentionally measured in immediately claimable batches:
// postthreads remains the hard connection cap while hot buckets use
// available parallelism instead of draining all batches serially.
$demand[$index] = max(1, (int) ceil($bucketCount['count'] / max(1, $queryLimit)));
}
for ($index = 0; count($queue) < $target; $index++) {
$queue[] = (object) ['id' => $chars[$index % count($chars)]];
$target = max(count($queue), min(max(1, $maxProcesses), array_sum($demand)));
while (count($queue) < $target) {
$nextIndex = null;
$largestRemainingDemand = 0;
foreach ($chars as $index => $char) {
$remainingDemand = $demand[$index] - $allocations[$index];
if ($remainingDemand > $largestRemainingDemand) {
$largestRemainingDemand = $remainingDemand;
$nextIndex = $index;
}
}
if ($nextIndex === null) {
break;
}
$queue[] = (object) ['id' => $chars[$nextIndex]];
$allocations[$nextIndex]++;
}
return $queue;
@@ -293,7 +338,14 @@ class PostProcessRunner extends BaseRunner
try {
foreach ($chars as $char) {
$tempWorkspace->ensureMainTempPath($basePath, $char);
$bucketPath = $tempWorkspace->ensureMainTempPath($basePath, $char);
$tempWorkspace->pruneStaleWorkerDirectories(
$bucketPath,
max(
3600,
(int) config('nntmux.multiprocessing_max_child_time', 1800) * 2,
),
);
}
} catch (\Throwable $e) {
cli()->warning('Additional post-processing skipped: '.$e->getMessage());
app/Services/TempWorkspaceService.php
+32 -2
View File
@@ -15,8 +15,12 @@ class TempWorkspaceService
* Ensure the main temp path exists and is namespaced by groupID or guidChar.
* Returns the resolved main temp path.
*/
public function ensureMainTempPath(string $basePath, string $guidChar = '', string $groupID = ''): string
{
public function ensureMainTempPath(
string $basePath,
string $guidChar = '',
string $groupID = '',
string $workerToken = ''
): string {
// Normalize separator at end
if (! Str::endsWith($basePath, ['/', '\\'])) {
$basePath .= '/';
@@ -28,6 +32,10 @@ class TempWorkspaceService
$basePath .= $guidChar.'/';
}
if ($workerToken !== '') {
$basePath .= hash('sha256', $workerToken).'/';
}
$this->ensureWritableDirectory($basePath, 'Additional post-processing temp path');
return $basePath;
@@ -72,6 +80,28 @@ class TempWorkspaceService
}
}
public function pruneStaleWorkerDirectories(string $bucketPath, int $staleAfterSeconds): int
{
if ($bucketPath === '' || ! File::isDirectory($bucketPath)) {
return 0;
}
$removed = 0;
$staleBefore = time() - max(1, $staleAfterSeconds);
foreach (File::directories($bucketPath) as $directory) {
if (File::lastModified($directory) >= $staleBefore) {
continue;
}
if (File::deleteDirectory($directory)) {
$removed++;
}
}
return $removed;
}
/**
* List files under a directory.
* If $pattern is provided, return an array of preg_match($pattern, $relativePath, $matches) arrays,
app/Services/Tmux/TmuxMonitorService.php
+3 -2
View File
@@ -249,8 +249,9 @@ class TmuxMonitorService
// rows claimed by active workers. `work_available` is the scheduler
// gate, excluding fresh claims so tmux does not respawn duplicate
// additional workers while a claimed batch is still running.
$this->runVar['counts']['now']['work'] = AdditionalCandidateQuery::baseBuilder(includeClaimed: true)->count();
$this->runVar['counts']['now']['work_available'] = AdditionalCandidateQuery::baseBuilder()->count();
$additionalBacklog = AdditionalCandidateQuery::backlogCounts();
$this->runVar['counts']['now']['work'] = $additionalBacklog['total'];
$this->runVar['counts']['now']['work_available'] = $additionalBacklog['available'];
$this->runVar['timers']['query']['proc2_time'] = time() - $timer2;
resources/views/admin/site/edit.blade.php
+3 -4
View File
@@ -771,7 +771,7 @@
class="flex-1 px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md focus:ring-blue-500 focus:border-blue-500 bg-white dark:bg-gray-700 text-gray-900 dark:text-gray-100">
<span class="px-3 py-2 bg-gray-100 dark:bg-gray-800 border border-gray-300 dark:border-gray-600 rounded-md">seconds</span>
</div>
<p class="mt-1 text-sm text-gray-500">Maximum wall-clock seconds to spend processing a single release before skipping it. Set to 0 to disable. Default: 120.</p>
<p class="mt-1 text-sm text-gray-500">Maximum wall-clock seconds to spend processing a single release before skipping it. Keep this below the multiprocessing child timeout. Set to 0 to disable. Default: 120.</p>
</div>
<div>
@@ -789,7 +789,7 @@
</label>
<input type="text" id="maxaddprocessed" name="maxaddprocessed" value="{{ $site['maxaddprocessed'] ?? '' }}"
class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md focus:ring-blue-500 focus:border-blue-500 bg-white dark:bg-gray-700 text-gray-900 dark:text-gray-100">
<p class="mt-1 text-sm text-gray-500">The maximum amount of releases to process for passwords/previews/mediainfo per run. Every release gets processed here. This uses NNTP an connection, 1 per thread. This does not query external metadata providers.</p>
<p class="mt-1 text-sm text-gray-500">Releases claimed per worker batch for passwords, previews, and media info. Larger batches reduce query overhead but increase each worker's memory and runtime. Each thread can hold one NNTP connection.</p>
</div>
<div>
@@ -1046,7 +1046,7 @@
</label>
<input type="text" id="postthreads" name="postthreads" value="{{ $site['postthreads'] ?? '' }}"
class="w-full px-3 py-2 border border-gray-300 dark:border-gray-600 rounded-md focus:ring-blue-500 focus:border-blue-500 bg-white dark:bg-gray-700 text-gray-900 dark:text-gray-100">
<p class="mt-1 text-sm text-gray-500">The number of threads for additional postprocessing. This includes deep rar inspection, preview and sample creation and nfo processing.</p>
<p class="mt-1 text-sm text-gray-500">Maximum simultaneous additional-processing workers and NNTP sessions. Workers reuse their process and connection across bounded batches; raise this only within CPU, memory, disk I/O, and provider connection limits.</p>
</div>
<div>
@@ -1100,4 +1100,3 @@
</div>
</div>
@endsection
resources/views/admin/site/tmux-edit-content.blade.php
+1 -1
View File
@@ -225,7 +225,7 @@
</x-select>
</x-form.group>
<x-form.group label="Postprocess Additional Sleep Timer" for="post_timer">
<x-form.group label="Postprocess Additional Sleep Timer" for="post_timer" help="Delay after a bounded worker cycle. Persistent workers drain several batches before this delay, so very low polling intervals are usually unnecessary.">
<div class="flex gap-2">
<x-input id="post_timer" name="post_timer" type="number" value="{{ $site['post_timer'] ?? 300 }}" class="flex-1" />
<span class="px-3 py-2 bg-gray-100 dark:bg-gray-800 border border-gray-300 dark:border-gray-600 rounded-md text-sm">seconds</span>
tests/Feature/AdditionalCandidateQueryTest.php
+20
View File
@@ -132,6 +132,26 @@ class AdditionalCandidateQueryTest extends TestCase
$this->assertSame(2, AdditionalCandidateQuery::baseBuilder(includeClaimed: true)->count());
}
public function test_backlog_counts_are_aggregated_by_bucket_in_one_shape(): void
{
DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]);
DB::table('releases')->insert([
$this->releaseRow(1, 'a'),
$this->releaseRow(2, 'a', claimedAt: now()),
$this->releaseRow(3, 'b', claimedAt: now()->subSeconds(301)),
]);
$this->assertSame([
['bucket' => 'a', 'total' => 2, 'available' => 1],
['bucket' => 'b', 'total' => 1, 'available' => 1],
], AdditionalCandidateQuery::bucketBacklog());
$this->assertSame(['total' => 3, 'available' => 2], AdditionalCandidateQuery::backlogCounts());
$this->assertSame([
['bucket' => 'a', 'count' => 1],
['bucket' => 'b', 'count' => 1],
], AdditionalCandidateQuery::availableBucketCounts());
}
public function test_claim_batch_excludes_active_claims_and_recovers_stale_claims(): void
{
DB::table('categories')->insert([
tests/Feature/AdditionalProcessingOrchestratorClaimTest.php
+12 -4
View File
@@ -260,8 +260,12 @@ class RecordingTempWorkspaceService extends TempWorkspaceService
public int $clearDirectoryCalls = 0;
public function ensureMainTempPath(string $basePath, string $guidChar = '', string $groupID = ''): string
{
public function ensureMainTempPath(
string $basePath,
string $guidChar = '',
string $groupID = '',
string $workerToken = ''
): string {
$this->ensureMainTempPathCalls++;
return '/tmp/additional/';
@@ -275,8 +279,12 @@ class RecordingTempWorkspaceService extends TempWorkspaceService
class FailingTempWorkspaceService extends RecordingTempWorkspaceService
{
public function ensureMainTempPath(string $basePath, string $guidChar = '', string $groupID = ''): string
{
public function ensureMainTempPath(
string $basePath,
string $guidChar = '',
string $groupID = '',
string $workerToken = ''
): string {
$this->ensureMainTempPathCalls++;
throw new \RuntimeException('Additional post-processing temp path "/root/nope/a/" is not writable');
tests/Feature/PostProcessRunnerAdditionalThreadsTest.php
+84 -4
View File
@@ -4,11 +4,15 @@ declare(strict_types=1);
namespace Tests\Feature;
use App\Services\AdditionalProcessing\AdditionalProcessingOrchestrator;
use App\Services\Runners\PostProcessRunner;
use Illuminate\Contracts\Console\Kernel;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\Artisan;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Schema;
use Mockery;
use Mockery\MockInterface;
use PDO;
use Tests\TestCase;
@@ -112,14 +116,14 @@ class PostProcessRunnerAdditionalThreadsTest extends TestCase
$this->assertSame(5, $runner->capturedMaxProcesses);
$this->assertCount(5, $runner->capturedCommands);
$this->assertContains(PHP_BINARY.' artisan postprocess:guid additional 0', $runner->capturedCommands);
$this->assertContains(PHP_BINARY.' artisan postprocess:guid additional 4', $runner->capturedCommands);
$this->assertContains(PHP_BINARY.' artisan postprocess:guid additional 0 --worker --max-batches=4', $runner->capturedCommands);
$this->assertContains(PHP_BINARY.' artisan postprocess:guid additional 4 --worker --max-batches=4', $runner->capturedCommands);
}
public function test_process_additional_repeats_hot_bucket_to_fill_configured_threads(): void
{
DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]);
foreach (range(1, 5) as $id) {
foreach (range(1, 125) as $id) {
DB::table('releases')->insert($this->releaseRow($id, 'a'));
}
@@ -144,11 +148,87 @@ class PostProcessRunnerAdditionalThreadsTest extends TestCase
$this->assertSame(5, $runner->capturedMaxProcesses);
$this->assertCount(5, $runner->capturedCommands);
$this->assertSame(
array_fill(0, 5, PHP_BINARY.' artisan postprocess:guid additional a'),
array_fill(0, 5, PHP_BINARY.' artisan postprocess:guid additional a --worker --max-batches=4'),
array_values($runner->capturedCommands)
);
}
public function test_process_additional_does_not_start_idle_workers_for_a_small_hot_bucket(): void
{
DB::table('categories')->insert(['id' => 1, 'disablepreview' => 0]);
foreach (range(1, 5) as $id) {
DB::table('releases')->insert($this->releaseRow($id, 'a'));
}
$runner = new class extends PostProcessRunner
{
/**
* @var array<string|int, string>
*/
public array $capturedCommands = [];
protected function runStreamingCommands(array $commands, int $maxProcesses, string $desc): void
{
$this->capturedCommands = $commands;
}
};
$runner->processAdditional();
$this->assertSame([
PHP_BINARY.' artisan postprocess:guid additional a --worker --max-batches=4',
], array_values($runner->capturedCommands));
}
public function test_direct_guid_command_processes_one_batch(): void
{
$this->mock(AdditionalProcessingOrchestrator::class, function (MockInterface $mock): void {
$mock->shouldReceive('start')
->once()
->with('', 'a', Mockery::type('string'), [])
->andReturn(['claimed' => 1, 'processed' => 1, 'failed' => 0, 'claimed_ids' => [1]]);
$mock->shouldReceive('finish')->once();
});
$status = Artisan::call('postprocess:guid', [
'type' => 'additional',
'guid' => 'a',
]);
$this->assertSame(0, $status);
}
public function test_worker_guid_command_processes_bounded_distinct_batches(): void
{
$this->mock(AdditionalProcessingOrchestrator::class, function (MockInterface $mock): void {
$mock->shouldReceive('start')
->once()
->ordered()
->with('', 'a', Mockery::type('string'), [])
->andReturn(['claimed' => 1, 'processed' => 1, 'failed' => 0, 'claimed_ids' => [1]]);
$mock->shouldReceive('start')
->once()
->ordered()
->with('', 'a', Mockery::type('string'), [1])
->andReturn(['claimed' => 1, 'processed' => 1, 'failed' => 0, 'claimed_ids' => [2]]);
$mock->shouldReceive('start')
->once()
->ordered()
->with('', 'a', Mockery::type('string'), [1, 2])
->andReturn(['claimed' => 0, 'processed' => 0, 'failed' => 0, 'claimed_ids' => []]);
$mock->shouldReceive('finish')->once();
});
$status = Artisan::call('postprocess:guid', [
'type' => 'additional',
'guid' => 'a',
'--worker' => true,
'--max-batches' => 4,
]);
$this->assertSame(0, $status);
}
/**
* @return array<string, mixed>
*/
tests/Unit/TempWorkspaceServiceTest.php
+30
View File
@@ -57,6 +57,36 @@ class TempWorkspaceServiceTest extends TestCase
$this->assertStringEndsWith('/group42/guid-123/', str_replace('\\', '/', $tmp));
}
#[WithoutErrorHandler]
public function test_worker_workspaces_are_isolated_within_a_guid_bucket(): void
{
$first = $this->svc->ensureMainTempPath($this->base, 'a', '', 'worker-one');
$second = $this->svc->ensureMainTempPath($this->base, 'a', '', 'worker-two');
File::put($first.'first.bin', 'first');
File::put($second.'second.bin', 'second');
$this->svc->clearDirectory($first, false);
$this->assertFalse(File::exists($first));
$this->assertTrue(File::isFile($second.'second.bin'));
$this->assertNotSame($first, $second);
}
#[WithoutErrorHandler]
public function test_prune_stale_worker_directories_keeps_active_siblings(): void
{
$bucket = $this->svc->ensureMainTempPath($this->base, 'a');
$stale = $this->svc->ensureMainTempPath($this->base, 'a', '', 'stale-worker');
$active = $this->svc->ensureMainTempPath($this->base, 'a', '', 'active-worker');
touch(rtrim($stale, '/\\'), time() - 7200);
$removed = $this->svc->pruneStaleWorkerDirectories($bucket, 3600);
$this->assertSame(1, $removed);
$this->assertFalse(File::exists($stale));
$this->assertTrue(File::isDirectory($active));
}
#[WithoutErrorHandler]
public function test_ensure_main_temp_path_repairs_existing_unwritable_bucket_directory(): void
{