Saga Lara Flow: тривалі воркфлоу та компенсуючі транзакції на основі черг Laravel

Saga Lara Flow - Laravel-пакет від Андрія Карпишина, який дозволяє описувати довготривалі бізнес-процеси у вигляді окремих PHP-методів поверх черг Laravel. Списати гроші з картки, зарезервувати товар на складі, оформити доставку - усе це пишеться послідовно в методі handle(), без ланцюжків джобів і без машин станів.

Ключова особливість пакета - він записує кожен крок до бази даних після завершення. Коли воркфлоу відновлюється або підхоплюється воркером, механізм повторно виконує метод handle(), але $this->action() перехоплює кожен виклик: якщо крок уже завершено, повертається збережений результат без повторного запуску класу екшену. Коли повтор досягає невиконаного коду, нормальне виконання продовжується. Якщо крок викидає виняток, механізм запускає компенсаційну логіку для попередніх кроків у зворотному порядку.

Ключові можливості пакета

• Воркфлоу як звичайні методи: метод handle() викликає екшени послідовно, а призупинення між кроками реалізовано через винятки замість ланцюжків джобів
• Компенсації: реєстрація екшену відміни або closure для кожного кроку через compensateWith(), що відкочує завершені кроки у разі збою наступних
• Сигнали: $this->signal() призупиняє виконання до передачі даних ззовні, з опціональною підтримкою таймаутів
• Паралельні блоки: одночасне виконання кількох екшенів зі збором результатів у масив
• Вкладені воркфлоу: запуск вкладеного воркфлоу з політикою закриття, що контролює його поведінку після завершення батьківського
• Запис побічних ефектів: обгортка недетермінованих значень (UUID, timestamp) для використання оригінального результату при повторах
• Запити на основі тегів: додавання ключ-значення тегів при створенні або всередині виконання для пошуку за класом воркфлоу, тегом і статусом
• Artisan-команди: перегляд запусків, інспекція стану, доставка сигналів, скасування виконань, моніторинг закінчення термінів та очищення старих записів

Воркфлоу та екшени

Воркфлоу успадковується від класу Workflow пакета й викликає екшени через $this->action(). Екшени - окремі класи, що отримуються з контейнера, тому їхні залежності інжектяться разом з аргументами, які ви передаєте:

use DiscoveryUkraine\SagaLaraFlow\Workflow;

class ProvisionAccountWorkflow extends Workflow
{
    public function handle(string $email): array
    {
        $tenantId = $this->action(CreateTenant::class, $email)->run();
        $this->action(SendWelcomeEmail::class, $email)->run();
        return ['tenant' => $tenantId];
    }
}

use DiscoveryUkraine\SagaLaraFlow\Action;

class CreateTenant extends Action
{
    public function handle(TenantRepository $tenants, string $email): string
    {
        return $tenants->provision($email)->id;
    }
}

Екшени мають власні налаштування черг. Властивість $tries контролює кількість спроб повтору, $timeout обмежує час кожної спроби, а expiresAt() встановлює дедлайн для окремого кроку. Коли спроби вичерпано, метод воркфлоу отримує ActionFailedException, який можна перехопити та обробити, а прострочений дедлайн виявляється як FlowExpiredException.

Запуск відбувається через фасад SagaFlow. Метод run() додає воркфлоу до черги й одразу повертає очікуваний запуск, тоді як runSync() виконує кожен крок синхронно та повертає завершений запуск, що зручно для перевірок у тестах:

use DiscoveryUkraine\SagaLaraFlow\Facades\SagaFlow;

$run = SagaFlow::create(ProvisionAccountWorkflow::class)
    ->withArguments('jane@example.com')
    ->runSync();
    
$this->assertTrue($run->isCompleted());
$this->assertEquals('tenant-123', $run->result()['tenant']);

Повтор працює лише якщо кожен крок повертає те саме значення, що й під час першого виконання. Будь-що недетерміноване потрібно записати при першому виконанні. Обгорніть це в sideEffect(), і збережене значення повертатиметься при кожному наступному проході:

$reference = $this->sideEffect('reference', fn () => (string) Str::uuid());

Компенсація невдалих транзакцій

Сага-половина пакета - це поведінка відкату. Кожен крок може зареєструвати екшен, що його скасовує, і ці скасування спрацьовують у зворотному порядку, коли наступний крок зазнає невдачі:

public function handle(string $orderId): void
{
    $this->action(ChargeCard::class, $orderId)
        ->compensateWith(RefundCard::class, $orderId)
        ->run();
        
    $this->action(ReserveStock::class, $orderId)
        ->compensateWith(ReleaseStock::class, $orderId)
        ->run();
        
    // Якщо це викине виняток, спочатку запуститься ReleaseStock, потім RefundCard
    $this->action(ShipOrder::class, $orderId)->run();
}

Для невеликих компенсацій можна передати closure замість класу екшену. Коли група кроків має відкочуватися як одне ціле, $this->saga() створює явний блок з двома додатковими контролями: onCompensationFailure() визначає, чи невдала компенсація перериває відкат чи дозволяє йому продовжитися, а compensateInParallel() запускає компенсації групи паралельно замість послідовно.

use DiscoveryUkraine\SagaLaraFlow\Enums\CompensationFailurePolicy;

$this->saga()
    ->onCompensationFailure(CompensationFailurePolicy::Continue)
    ->compensateInParallel()
    ->step(ChargeCard::class, $orderId)->compensateWith(RefundCard::class, $orderId)
    ->step(ReserveStock::class, $orderId)->compensateWith(ReleaseStock::class, $orderId)
    ->run();

Очікування зовнішніх даних

Сигнали обробляють випадки, коли процес призупиняється в очікуванні схвалення від людини або зворотного виклику від сторонньої системи. $this->signal() призупиняє запуск і звільняє воркер до доставки іменованого сигналу. Виклик wait() відновлює виконання після отримання, а timeoutAfter() додає дедлайн:

use DiscoveryUkraine\SagaLaraFlow\Exceptions\AwaitSignalTimeoutException;

try {
    $decision = $this->signal('approval')
        ->timeoutAfter(now()->addDay())
        ->wait();
} catch (AwaitSignalTimeoutException $e) {
    $this->action(AutoReject::class)->run();
}

Доставка відбувається з будь-якого місця застосунку через handle запуску. Також є варіант signalIfRunning(), який повертає false замість викидання винятку, коли запуск уже завершено або скасовано:

SagaFlow::loadFlow($runId)->signal('approval', ['approved' => true]);

Теги дозволяють знайти потрібний запуск без збереження його ID. Додавайте їх при створенні через withTags() або всередині handle() через $this->tag(), потім фільтруйте за класом воркфлоу, тегом і статусом:

SagaFlow::query()
    ->whereWorkflow(ProvisionCompanyWorkflow::class)
    ->whereTag('company', $companyId)
    ->signalable()
    ->handles()
    ->first()
    ?->signal('owner-synced');

Той самий білдер корисний для операційних перевірок. Скоупи на кшталт running(), waiting(), failed() і before() повертають або моделі FlowRun, або handles для маніпуляцій, тож знайти всі запуски, що застрягли в стані очікування понад годину, можна одним запитом.

Паралельність, опціональні кроки та вкладеність

Незалежна робота виконується в паралельному блоці, який повертає результати екшенів у вигляді списку для деструктуризації. Стандартна політика failFast() скасовує блок при першій невдачі:

[$pricing, $inventory, $reviews] = $this->parallel()
    ->action(FetchPricing::class, $sku)
    ->action(FetchInventory::class, $sku)
    ->action(FetchReviews::class, $sku)
    ->run();

Крок, який не повинен зупиняти весь запуск, отримує continueOnFailure() разом із запасним значенням, а optionalAction() є скороченням:

$score = $this->optionalAction(FetchRiskScore::class, $orderId)
    ->fallbackValueOnFail(0)
    ->run();

Воркфлоу можуть також викликати дочірні воркфлоу через $this->child(), передаючи ChildClosePolicy, що визначає, чи дочірній воркфлоу має бути скасовано або залишено працювати після завершення батьківського:

use DiscoveryUkraine\SagaLaraFlow\Enums\ChildClosePolicy;

$result = $this->child(ConfigureSubdomainWorkflow::class, $domain)
    ->onParentClose(ChildClosePolicy::Terminate)
    ->run();

Версіонування та моніторинг

Воркфлоу, запущений сьогодні, може працювати й наступного місяця, для чого й потрібне версіонування. Передача version('v2') при створенні запуску прив'язує його до визначення воркфлоу, тож запуски в процесі продовжують повторюватися з кодом, з яким стартували, тоді як нові запуски підхоплюють оновлений клас.

Дедлайни потребують чогось, що стежить за годинником і помічає їх спливання, тому пакет постачається з командою saga-flow:monitor для реєстрації в планувальнику Laravel:

Schedule::command('saga-flow:monitor')->everyMinute();

Встановлення та команди

Пакет вимагає PHP 8.5 та Laravel 13, випущений під ліцензією MIT:

composer require discovery-ukraine/saga-lara-flow
php artisan migrate
php artisan vendor:publish --tag="saga-lara-flow-config"

Опублікований конфігураційний файл охоплює підключення до бази даних і чергу для стану воркфлоу, поведінку блокувань та хуки мультитенантності. Для мультитенантних застосунків надайте closure capture і restore, щоб запуск відновлювався на воркері в тому самому контексті орендаря, в якому стартував.

Пакет включає Artisan-команди для створення заготовок і керування виконаннями воркфлоу:

• make:workflow і make:action: створення заготовок нових класів воркфлоу та екшенів
• saga-flow:list: інспекція активних, очікуваних і невдалих виконань
• saga-flow:signal: доставка сигналів безпосередньо до воркфлоу з CLI
• saga-flow:prune: очищення записів завершених і скасованих запусків

Повна документація, включно з розділами про моніторинг закінчення термінів, тестування та мультитенантність, доступна на sagalaraflow.dev. Більше про пакет, повні інструкції зі встановлення та вихідний код можна знайти в GitHub-репозиторії saga-lara-flow.