feat(autopodbor): AutopodborStudyScheduler — одна дорожка + round-robin
This commit is contained in:
Дмитрий
@@ -0,0 +1,154 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace App\Services\Autopodbor;
|
||||
|
||||
use App\Jobs\Autopodbor\RunAutopodborStudyJob;
|
||||
use Illuminate\Support\Facades\Cache;
|
||||
use Illuminate\Support\Facades\DB;
|
||||
|
||||
/**
|
||||
* Планировщик шага 2 «изучение» — одна дорожка на весь портал + честный
|
||||
* round-robin между тенантами.
|
||||
*
|
||||
* Изучение конкурентов — тяжёлый сетевой шаг. Чтобы не забивать очередь одним
|
||||
* тенантом и не гонять два изучения разом, планировщик держит РОВНО ОДНУ активную
|
||||
* дорожку (не более одного study-прогона в статусе `running` на весь портал) и,
|
||||
* когда дорожка свободна, отдаёт её «наименее недавно обслуженному» тенанту.
|
||||
*
|
||||
* Все чтения идут через соединение `pgsql_supplier` (роль crm_supplier_worker,
|
||||
* BYPASSRLS на проде) — распорядителю нужна межтенантная видимость всей очереди.
|
||||
*/
|
||||
class AutopodborStudyScheduler
|
||||
{
|
||||
/** Redis-соединение под межтенантную видимость (BYPASSRLS на проде). */
|
||||
private const CONNECTION = 'pgsql_supplier';
|
||||
|
||||
private const LOCK_KEY = 'autopodbor:study:tick';
|
||||
|
||||
/**
|
||||
* Выбрать id следующего study-прогона для запуска, либо null.
|
||||
*
|
||||
* Правила:
|
||||
* 1. Если хоть один study-прогон уже `running` (по ВСЕМ тенантам) — дорожка
|
||||
* занята, вернуть null.
|
||||
* 2. Иначе среди тенантов с `queued` study-прогонами выбрать «наименее недавно
|
||||
* обслуженного»: ключ тенанта = MAX(finished_at) его study-прогонов; NULL
|
||||
* (никогда не обслуживался) — наивысший приоритет; тай-брейк — самый старый
|
||||
* created_at среди его queued-прогонов.
|
||||
* 3. Внутри выбранного тенанта — его самый старый queued-прогон. Вернуть id.
|
||||
*/
|
||||
public function pickNextRunId(): ?int
|
||||
{
|
||||
$conn = DB::connection(self::CONNECTION);
|
||||
|
||||
// (1) одна дорожка: есть running study → занято.
|
||||
$busy = $conn->table('autopodbor_runs')
|
||||
->where('kind', 'study')
|
||||
->where('status', 'running')
|
||||
->exists();
|
||||
if ($busy) {
|
||||
return null;
|
||||
}
|
||||
|
||||
// Все queued study-прогоны (по всем тенантам).
|
||||
$queued = $conn->table('autopodbor_runs')
|
||||
->where('kind', 'study')
|
||||
->where('status', 'queued')
|
||||
->get(['id', 'tenant_id', 'created_at']);
|
||||
if ($queued->isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
// last-served по тенанту = MAX(finished_at) среди ВСЕХ его study-прогонов.
|
||||
// NULL означает «никогда не обслуживался».
|
||||
$lastServed = $conn->table('autopodbor_runs')
|
||||
->where('kind', 'study')
|
||||
->groupBy('tenant_id')
|
||||
->select('tenant_id', DB::raw('MAX(finished_at) AS last_finished'))
|
||||
->pluck('last_finished', 'tenant_id');
|
||||
|
||||
// (2) выбираем тенанта.
|
||||
$best = null; // ['tenant' => int, 'last' => ?string, 'oldestCreated' => string]
|
||||
foreach ($queued->groupBy('tenant_id') as $tenantId => $runs) {
|
||||
$candidate = [
|
||||
'tenant' => (int) $tenantId,
|
||||
'last' => $lastServed[$tenantId] ?? null,
|
||||
'oldestCreated' => (string) $runs->min('created_at'),
|
||||
];
|
||||
if ($best === null || $this->isBetter($candidate, $best)) {
|
||||
$best = $candidate;
|
||||
}
|
||||
}
|
||||
|
||||
// (3) внутри выбранного тенанта — самый старый queued.
|
||||
$winner = $queued
|
||||
->where('tenant_id', $best['tenant'])
|
||||
->sortBy('created_at')
|
||||
->first();
|
||||
|
||||
return (int) $winner->id;
|
||||
}
|
||||
|
||||
/**
|
||||
* Запустить одну итерацию планировщика под Redis-локом: если дорожка свободна и
|
||||
* есть кандидат — задиспатчить его изучение. Лок защищает от гонки нескольких
|
||||
* тиков (два воркера не отправят два изучения разом).
|
||||
*/
|
||||
public function tick(): void
|
||||
{
|
||||
$lock = Cache::lock(self::LOCK_KEY, 10);
|
||||
|
||||
if (! $lock->get()) {
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
$runId = $this->pickNextRunId();
|
||||
if ($runId === null) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Перечитать прогон под тем же межтенантным соединением: статус мог
|
||||
// измениться между выбором и диспатчем.
|
||||
$run = DB::connection(self::CONNECTION)
|
||||
->table('autopodbor_runs')
|
||||
->where('id', $runId)
|
||||
->first();
|
||||
|
||||
if ($run === null || $run->status !== 'queued') {
|
||||
return;
|
||||
}
|
||||
|
||||
RunAutopodborStudyJob::dispatch((int) $run->id, (int) $run->tenant_id);
|
||||
} finally {
|
||||
$lock->release();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* $a приоритетнее $b? Приоритет: никогда не обслуженный (last === null) вперёд;
|
||||
* затем обслуженный раньше вперёд; тай-брейк — более старый created_at.
|
||||
*
|
||||
* @param array{tenant:int,last:?string,oldestCreated:string} $a
|
||||
* @param array{tenant:int,last:?string,oldestCreated:string} $b
|
||||
*/
|
||||
private function isBetter(array $a, array $b): bool
|
||||
{
|
||||
$aNever = $a['last'] === null;
|
||||
$bNever = $b['last'] === null;
|
||||
|
||||
if ($aNever !== $bNever) {
|
||||
return $aNever; // никогда не обслуженный — выше
|
||||
}
|
||||
|
||||
if (! $aNever && $a['last'] !== $b['last']) {
|
||||
// оба обслужены: обслуженный раньше — выше
|
||||
return strcmp((string) $a['last'], (string) $b['last']) < 0;
|
||||
}
|
||||
|
||||
// тай-брейк: самый старый queued created_at — выше
|
||||
return strcmp($a['oldestCreated'], $b['oldestCreated']) < 0;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,65 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
use App\Models\AutopodborRun;
|
||||
use App\Models\Tenant;
|
||||
use App\Services\Autopodbor\AutopodborStudyScheduler;
|
||||
use Illuminate\Foundation\Testing\DatabaseTransactions;
|
||||
use Tests\Concerns\SharesSupplierPdo;
|
||||
|
||||
uses(DatabaseTransactions::class, SharesSupplierPdo::class);
|
||||
|
||||
// Логический номер тенанта (1,2,…) → реальный tenants.id (FK autopodbor_runs.tenant_id).
|
||||
// Сбрасываем перед каждым тестом: DatabaseTransactions откатывает строки, поэтому
|
||||
// карта не должна протекать между тестами.
|
||||
beforeEach(function () {
|
||||
$GLOBALS['__sched_tenant_map'] = [];
|
||||
});
|
||||
|
||||
function studyRun(int $tenant, string $status, ?string $finishedAt = null, ?string $createdAt = null): AutopodborRun {
|
||||
if (! isset($GLOBALS['__sched_tenant_map'][$tenant])) {
|
||||
$GLOBALS['__sched_tenant_map'][$tenant] = Tenant::factory()->create()->id;
|
||||
}
|
||||
$tenantId = $GLOBALS['__sched_tenant_map'][$tenant];
|
||||
|
||||
$run = AutopodborRun::create([
|
||||
'tenant_id' => $tenantId, 'kind' => 'study', 'status' => $status,
|
||||
'competitor_id' => null, 'region_code' => 1, 'params' => [],
|
||||
'finished_at' => $finishedAt,
|
||||
]);
|
||||
if ($createdAt !== null) {
|
||||
// created_at не в timestamps — задаём явно через query builder (мимо fillable),
|
||||
// чтобы порядок был детерминирован
|
||||
AutopodborRun::whereKey($run->id)->update(['created_at' => $createdAt]);
|
||||
$run->refresh();
|
||||
}
|
||||
return $run;
|
||||
}
|
||||
|
||||
it('дорожка занята: есть running → pickNext возвращает null', function () {
|
||||
studyRun(1, 'running');
|
||||
studyRun(2, 'queued');
|
||||
expect(app(AutopodborStudyScheduler::class)->pickNextRunId())->toBeNull();
|
||||
});
|
||||
|
||||
it('чередует тенантов: наименее недавно обслуженный вперёд', function () {
|
||||
studyRun(1, 'done', finishedAt: now()->toDateTimeString());
|
||||
studyRun(1, 'queued');
|
||||
studyRun(2, 'done', finishedAt: now()->subHour()->toDateTimeString());
|
||||
$t2q = studyRun(2, 'queued');
|
||||
expect(app(AutopodborStudyScheduler::class)->pickNextRunId())->toBe($t2q->id);
|
||||
});
|
||||
|
||||
it('внутри тенанта берёт самый старый queued', function () {
|
||||
$old = studyRun(1, 'queued', createdAt: now()->subMinutes(10)->toDateTimeString());
|
||||
studyRun(1, 'queued', createdAt: now()->toDateTimeString());
|
||||
expect(app(AutopodborStudyScheduler::class)->pickNextRunId())->toBe($old->id);
|
||||
});
|
||||
|
||||
it('никогда не обслуженный тенант имеет приоритет', function () {
|
||||
studyRun(1, 'done', finishedAt: now()->subDay()->toDateTimeString());
|
||||
studyRun(1, 'queued');
|
||||
$t2q = studyRun(2, 'queued');
|
||||
expect(app(AutopodborStudyScheduler::class)->pickNextRunId())->toBe($t2q->id);
|
||||
});
|
||||
Reference in New Issue
Block a user