Привет! Это третья и заключительная часть истории поиска надёжного способа работы с транзакциями в распределённых системах.

В первой части мы поставили задачу, определили критерии оценки и набросали варианты. Во второй части я подробно разобрал решение на Temporal. В этой статье расскажу чем в итоге всё закончилось, а точнее с чего начнётся. 

Статья может быть полезна разработчикам и архитекторам, которые задумываются о вынесении части бизнес-логики из монолита, написанного на PHP.

Задача и основные требования

• У нас в Профи есть монолит из которого хочется вынести куски бизнес-логики, такие как биллинг, создание заказа, логика чатов и т.п.

• Для этого нужна технология, которая позволит надёжно выполнять мутации и транзакции в распределённой системе.

• При этом крайне желательно, чтобы решение было малоинвазивным — простым для разработчиков и чтобы переиспользовались текущие архитектурные решения.

Подробнее про постановку задачи можно почитать тут. 

Disclaimer: некоторый код в статье может не работать т.к. нужен только для демонстрации алгоритмов.

Для иллюстрации логики решения возьмём транзакцию-прототип одного из сценариев реального продукта. Наша задача — вынести метод purchase в отдельный микросервис.

<?php
class Greeter {
    function greet($name) {
        $valid = $validator->validateName($name);
        if (!$valid) {
            return false;
        }
        $pdo->beginTransaction();
        try {
            $greeting = $repository->createGreeting($name);
            $purchaser->purchase($greeting); 
            $sender->send($greeting);
            $pdo->commit();
            return $greeting;
        } catch (Exception $e) {
            $pdo->rollback();
            return false;
        }    
    }
}

Логика решения

Основные тезисы

• Делаем свой движок для старта, отслеживания, завершения и отката шагов т.к. больше не можем использовать ACID-транзакции.

• Добавляем возможность повторять шаги т.к. теперь у нас распределённая система, со всеми вытекающими.

• При повторах берём результаты из кеша и проверяем детерминированность — последовательность, сигнатуры и аргументы шагов должны совпадать.

• Упрощаем API движка, используя PHP-генераторы.

В результате транзакция выглядит примерно так:

<?php
class Greeter {
    function greet($name) {
        $valid = yield $validatorTasks->validateName($name);
        if (!$valid) {
            return false;
        }
        $greeting = yield $repositoryTasks->createGreeting($name);
        try {
            $task = $purchaserTasks->purchase($greeting);
            yield $task->setRetryPolicy(
                (new RetryPolicy())->setMaxRetries(3)
            );
        } catch (Exception $e) {
            yield $reverterTasks->revert($greeting);
            return false;
        }
        yield $senderTasks->send($greeting);
        return $greeting;
    }    
}

Запускается так:

<?php

// Превращаем существующий класс-врарпер для шагов транзакции
$greeterTasks = new Tasks(new Greeter());
$task = $greeterTasks->greet("John Doe");

// Создаём запуск с политикой повторов
$run = $engine->createRun(
    $task->setRetryPolicy(
        (new RetryPolicy())->setMaxRetries(3)
    )
);

// Запускаем и получаем результат
$result = $engine->performRun($run);
if ($result->isReady()) {
    echo $result->Data();
} 

Повторяем вот так:

<?php
// Получем id запуска из базы или очереди
$runId = ...
$run = $engine->getRun($runId);
// Запускаем повтор и получаем результат
$result = $engine->performRun($run);
if ($result->isReady()) {
    echo $result->Data();
} 

Получаем результаты в любой момент времени:

<?php
$run = $engine->getRun($runId);
$result = $engine->getRunResult($run);

Далее разберу как я пришёл к этому решению и дам ссылку на работающую библиотеку.

Этап 1: Отслеживаем состояние транзакции

Изменение данных в распределённых системах происходят не атомарно и нужно уметь определять состояние транзакции в каждый момент времени. Для этого идентифицируем транзакцию и сохраняем признаки её начала и завершения:

<?php

// Убираем ACID-транзакцию и добавляем ручную обработку отката
// $validator, $repository, $purchaser и т.п. — классы с бизнес-логикой

class Greeter {
    function greet($name) {
        $valid = $validator->validateName($name);
        if (!$valid) {
            return false;
        }
        $greeting = $repository->createGreeting($name);
        try {
            // Внутри purchase происходит GQL-вызов внешнего микросервиса 
            $purchaser->purchase($greeting); 
        } catch (Exception $e) {
            $reverter->revert($greeting);
            return false;
        }
        $sender->send($greeting);
        return $greeting;
    }
}

// Здесь будем держать callable для запуска Greeter::greet
class Transaction {
    function __construct(callable $callable);
}

// Конкретный запуск транзакции
class Run {
    function __construct(Transaction $tran);
}

// Сам движок
class Engine {
    function createRun(Transaction $tran): Run {
        // Тут нужно создать ID по которому можно будет 
        // отслеживать состояние транзакции и получать результат.
    }

    function performRun(Run $run) {
        $this->start($run); // Пишем признак начала транзакции
        $result = $run->task->callable(); // Запускаем Greeter::greet
        $this->finish($run); // Сохраняем признак окончания
        return $result; 
    }
}

// Собираем всё вместе
$greeter = new Greeter();
$tran = new Transaction(fn() => $greeter->greet('John Doe'));
$run = $engine->createRun($tran);
$result = $engine->performRun($run);

В случае, если вызов purchase упадёт, у нас будет запись о начале транзакции, но не будет об окончании и мы сможем решить что делать дальше — повторить, откатить или, например, отправить на ручной разбор.

Какие тут есть проблемы:

• В случае падения шага purchase мы сразу запускаем логику отката, хотя можно попробовать повторить этот шаг.

• Откат $reverter->revert($greeting)тоже может упасть.

• В случае повторного запуска выполняются все шаги, хотя некоторые уже были выполнены в исходном запуске.

• В общем случае повторы выполняются асинхронно — мы можем получить непредсказуемое состояние данных, если между повторами изменился код Greeting::greet. Например, в новом коде мы стали иначе создавать $greeting.

Этап 2: Записываем и проверяем шаги

Чтобы двигаться дальше, нужно завернуть шаги транзакции в прокси-класс, который позволит:

• Кешировать результаты работы бизнес-логики отдельных шагов.

• Проверять детерминированность.

• Ловить исключения, чтобы отправлять конкретный шаг на повтор.

Добавляем прокси-класс:

<?php

// Здесь будем держать callable для запуска бизнес-логики шагов
class Task {
    public function __construct(public callable $callable);
}

// Вместо непосредственного вызова бизнес-логики шагов
// создаём прокси-объекты, которые будут запускаться через движок
class Greeter {
    function greet(Run $run, string $name) {
        $valid = $engine->performTask(
          $run, new Task(fn() => $validator->validateName($name))
        );
        if (!$valid) {
            return false;
        }
        $greeting = $engine->performTask(
          $run, new Task(fn() => $repository->createGreeting($name))
        );
        try {
            $engine->performTask(
              $run, new Task(fn() => $purchaser->purchase($greeting))
            );
        // Ловим специальное исключение 
        // "шаг больше повторять не нужно, откатывай"
        } catch (RollbackTask $e) { 
            $engine->performTask(
              $run, new Task(fn() => $reverter->revert($greeting))
            );
            return false;
        }
    
        $engine->performTask(
          $run, new Task(fn() => $sender->send($greeting))
        );
        return $greeting;
    }    
}

Теперь мы можем уверенней вызывать Greeter::greet повторно, но вот что не нравится:

• В коде самой транзакции слишком много завязок на движок — при изменении API может потребоваться переписывать все транзакции.

• В greet появляется новый аргумент $run, который никак не относится к логике транзакции.

• Для отката нужно ловить специальное исключение.

Этап 3: Из центра наружу

В идеале хочется сделать так, чтобы внутри кода транзакции не было практически никаких знаний про движок. Этого можно добиться, если передать управление функцией изнутри наружу, используя PHP-генераторы. Генераторы — это не только ценный мех про экономию ресурсов, но и возможности:

• Останавливать и возобновлять работу функции с конкретного места, обозначенного ключевым словом yield.

• Передавать в функцию данные через Generator::send— можно использовать конструкцию $foo = yield bar()и устанавливать в $foo нужные нам значения снаружи.

Таким образом мы превращаем нашу Greeter::greet в генератор шагов:

<?php

class Greeter {
    function greet($name): Generator {
        $valid = yield new Task(fn() => $validator->validateName($name));
        if (!$valid) {
            return false;
        }
             
        $greeting = yield new Task(fn() => $repository->createGreeting($name));

        try {
            yield new Task(fn() => $purchaser->purchase($greeting));
        } catch (Exception $e) { 
            yield new Task(fn() => $reverter->revert($greeting));
            return false;
        }
        
        yield new Task(fn() => $sender->send($greeting));
        return $greeting;
    }    
}

Обратите внимание, что теперь для отката нам не нужно перехватывать специальное исключение т.к. вся бизнес-логика шагов выполняется в движке снаружи.

А вот так упрощённо выглядит код движка:

<?php

$tran = new Transaction(fn() => $greeter->greet('John Doe'));
$generator = $tran->callable();

...

while (true) {
    if ($generator->valid()) {
        $task = $generator->current(); // Получаем очередной шаг из генератора
        // @todo тут проверяем кеш для $task и детерминированность шага
        try {
          $result = $task->callable(); // Выполняем callable бизнес-логики
        } catch (Exception $e) {
          // @todo обработка ошибок и отправка на повтор
          break;
        }
        $generator->send($result); // Передаём результат в генератор
    } else { 
        $result = $generator->getReturn();
        break;
    }
}

Что тут происходит:

• Генератор yield-ит прокси-объекты шагов.

• Нам нужно либо выполнить бизнес-логику шага, либо взять из кеша + проверить детерминированность.

• Выполнив шаг, мы возвращаем результат в генератор и двигаем его дальше.

В коде «стало попросторнее», продолжаем.

Этап 4: Очереди и предохранитель

База данных движка — это single source of truth для состояния транзакций и шагов, а очереди это опция, которая позволит быстрее выполнять повторы или запускать транзакции асинхронно.

Добавим такой интерфейс:

<?php

interface RunQueueInterface {
    public function enqueueRun(Run $run);
}

Там, где движок решает, что транзакцию нужно увести на повтор, делаем соответствующий вызов enqueueRun. Саму реализацию RunQueueInterface и воркер для разгребания пока отдаём под ответственность пользователю движка т.к. стек и юзкейсы могут быть разные.

В API движка нужно предусмотреть метод, который селектит запуски, готовые к выполнению, например такой:

<?php

class Engine {
    ...
    function getScheduledRunIds(DateTime $until, int $limit): array {
        ...
        return $runIds;
    }  
}

Этот метод можно использовать как основной механизм получения и запуска повторов или как предохранитель, который поможет очереди, если она упала. Опять же, реализацию этого скрипта отдаём на откуп разработчику — мы предоставляем только API.

Идея в том, что в каждом проекте могут быть разные фреймворки и чтобы эффективно вписаться в неизвестную инфраструктуру лучше использовать низкоуровневый интерфейс, а не opinionated решения.

Этап 5: Доводка

Первое, что напрашивается — добавить политику повторов, чтобы можно было указывать сколько попыток делать, с какими интервалами, факторы для exponential backoff и т.п. Например так:

<?php

$purchaseTask = new Task(fn() => $purchaser->purchase($greeting));

yield $purchaseTask->setRetryPolicy(
    (new RetryPolicy())
        ->setMaxRetries(3)
        ->setMinInterval(1000)
);

Далее избавляемся от бойлерплейта типа new Task(fn() => $repository->createGreeting($name)) внутри транзакций. Добавляем класс-враппер, который на лету будет создавать шаги из обычных методов класса:

<?php

class Tasks {
    public function __construct(private object $object);

    public function __call($method, $args) {
        return new Task(fn() => $this->object->$method($args));
    }
}

Это позволяет писать вот такой код:

<?php

/** @var Validator */
$validatorTasks = new Tasks(new Validator());

... 

class Greeter {
    function greet($name) {
        $valid = yield $validatorTasks->validateName($name);
        if (!$valid) {
            return false;
        }
        ...
        return $greeting;
    }    
}

Обратите внимание, что с помощью phpdoc мы сделали живую ссылку на метод validateName через класс Tasks — можно пользоваться IDE-плюшками поиска использований, переходить внутрь метода и т.п.

Ещё вызывает вопросы класс Transaction. Зачем он нам? Если уже есть Task и движок может сам определять: это транзакция (генератор) или просто бизнес-логика, которую нужно надёжно выполнить — с обработкой ошибок, повторами и т.п. Поэтому избавляемся от Transaction:

<?php
/** @var Greeter */
$greeterTasks = new Tasks(new Greeter());
$task = $greeterTasks->greet("John Doe");
$run = $engine->createRun($task);
$result = $engine->performRun($run);

Что получилось в итоге

Получился простой, надёжный и достаточно аккуратный способ запуска мутаций и транзакций в распределённой системе. Где тут распределённость? — спросите вы. Она внутри бизнес-логики шагов, где можно делать внешние вызовы, писать в другую БД и т.п. с уверенностью, что эта логика будет выполнена.

Решение реализовано в виде библиотеки TSQM c открытым кодом.

Основные сущности:

• Tsqm — сам движок.

• Task — прокси объект для бизнес-логики или генератор таких прокси-объектов.

• Run — определяет запуск Task, отслеживает состояние.

• RunResult — результат запуска транзакции с признаком готовности или ошибки.

Из документации пока только инструкция для разработчиков. Потыкать примеры можно через консольное приложение examples/app.php:

• init:db — обнуляет и создаёт базу.

• example:hello-world — пример транзакции со случайной ошибкой.

• example:hello-world-simple — пример выполнения одного шага со случайной ошибкой.

• list:scheduled — вывести список запланированных запусков.

• run:scheduled — выполнить запланированные и подвисшие запуски, по сути — предохранитель.

• run:one — выполнить конкретный запуск.

Кстати, в основном все тесты библиотеки интеграционные — можно посмотреть примеры проверок для транзакций.

Цена использования

Одной из главных целей создания TSQM было дать инструмент, который легко встраивается в большие легаси-монолиты на PHP, чтобы выносить куски бизнес-логики в другие сервисы и базы. Решение получилось довольно лёгкое, но есть и ограничения:

MUST

• Код транзакции должен быть детерминирован — при повторах должны вызываться те же шаги, в той же последовательности, с теми же сигнатурами и аргументами.

• Аргументы и возвращаемые значения шагов должны быть сериализуемыми.

• Один и тот же шаг (одинаковая сигнатура и аргументы) нельзя использовать в транзакции два раза*. Например, следующий код упадёт с ошибкой:

<?php
yield $sender->send($greeting);
yield $sender->send($greeting);

* Это нужно, чтобы история была чистой — признак запуска и завершения конкретного шага пишется в базу только один раз. Можно, конечно, схлопывать шаги в движке, но надёжней использовать unique constraint в БД.

SHOULD

• Если транзакция уходит на повтор, то нельзя получить её результат синхронно и клиенты должны это учитывать — проверять isReady() у объекта RunResult, сохранять runId и запрашивать результат позже.

• Бизнес-логика вызываемая через Task должна быть идемпотентной без учёта случайных ошибок. Вероятность, что один и тот же код будет вызван дважды — низкая, но не нулевая.

• Если выкатываете новый код транзакции, в идеале его нужно версионировать — например, называть метод транзакции greetV1, greetV2 и т.д.

• При использовании враппера для шагов желательно добавлять инструкции phpdoc типа @var, чтобы IDE корректно понимала типы.

Планы по развитию

Для начала внедрить в Профи, потестить под нагрузкой, отладить и выпустить стабильную версию библиотеки с документацией.

Шорт лист планируемых фич:

• Поддержка неймспейсов, чтобы запуски от разных клиентов можно было держать в одной базе.

• Мониторинг и телеметрия.

• Разные политики для повторов в том числе exponential backoff.

• Конкретные реализации очередей для разных брокеров.

• Встроенный скрипт предохранителя.

• GUI для отслеживания, ручного перезапуска или отмены транзакций.

• Сделать реализацию для других платформ — TypeScript, Go, Java и прочие.

FAQ

Почему не использовали Temporal?

После сборки прототипа на Temporal и написания статьи казалось — вот оно, решение найдено, нужно пробовать. Потом голова немного остыла и вот какие стоперы закрыли тему:

• Комплексити. У нас уже есть отлаженный механизм межсервисного взаимодействия через единое API и GraphQL. Всё хорошо работает — авторизация, подписки, логирование, ревью API-схемы, окружение для разработки и т.п. Есть планы по развитию, народ в целом доволен. Поэтому не хочется влезать в историю с RPC и по-новой внедрять базовые механики.

• Высокий порог входа. “Етить, комбайн этот Temporal, конечно” — вот что сказал наш главный админ. И действительно, чтобы правильно приготовить Temporal в проде, нужно перелопатить доки, всё понять, всё поднять и проверить под нагрузкой. Потом ещё сходить в разработку с мануалами, зарядить devops на стендовую-поддержку ну и пройти прочие радости внедрения новых технологий. Короче, долго и дорого.

Не поймите меня неправильно — Temporal очень крутая штука и я люто рекомендую как минимум изучить её всем, кто работает в распределенных системах, но в нашем случае это конкретный оверкил.

В чём отличие TSQM от Temporal?

TSQM — это не workflow engine, он легче и проще, но менее универсальный — код шагов находится в одной кодовой базе, нет встроенной поддержки workflow, нет GUI и т.п.

TSQM - это оркестратор?

В каком-то смысле да, но в классическом понимании оркестратор всё таки активно использует очереди для запуска и получения результатов, тут же очереди — это опция. По задумке большинство транзакций будут выполняться синхронно, а на повторы уйдут только те, что свалились с ошибкой или по таймауту.

А где тут мутации?

Мутация — это транзакция из одного шага. Под мутацией здесь имеется в виду не только изменение данных, но и изменение состояния бизнес-процесса, например, отправка уведомления пользователю — это мутация состояния процесса покупки.

Что за название такое, TSQM?

TSQM — это библиотека, которая уже несколько лет используется в Профи для запуска и обработки отложенных заданий. Публичный tsqm-php — это её реинкарнация в open source. Расшифровывается как TaSk Queue Manager, смысл библиотеки немного поменялся, но название я решил оставить и использовать как имя собственное.

Спасибо, что дочитали! Пока ????

Статьи из серии

Первая часть про постановку задачи

Вторая часть про Temporal

Третья часть про TSQM: эта статья