Emitovanie po commite do databázy

Request vloží riadok v transakcii a emitne o ňom správu. Worker si správu môže vyzdvihnúť skôr, ako sa transakcia commitne, hľadať riadok a nenájsť nič. Keď sa transakcia vráti späť, správa je už odoslaná k riadku, ktorý nikdy neexistoval.

Tomaj\Hermes\BufferedEmitter (od 6.0) obalí ľubovoľný EmitterInterface a drží správy v pamäti, kým nezavoláte flush() z hooku po commite. Hook po rollbacku volá discard().

use Tomaj\Hermes\BufferedEmitter;
use Tomaj\Hermes\Emitter;

$emitter = new BufferedEmitter(new Emitter($driver, $logger), [$database, 'isInTransaction']);
$database->registerOnCommit([$emitter, 'flush']);
$database->registerOnRollback([$emitter, 'discard']);

Producenti ďalej uvádzajú typ EmitterInterface; emit() vracia BufferedEmitter.

Čo robí ktorá metóda

Metóda	Správanie
__construct(EmitterInterface $emitter, ?callable $shouldBuffer = null)	$shouldBuffer sa volá pri každom emit() so správou a prioritou. true správu podrží, čokoľvek iné ju hneď odošle. null drží vždy. Fungujú closures, [$object, 'method'] aj iné callables. Callable bez parametrov, napríklad $pdo->inTransaction(...), sa volá bez argumentov, takže vstavané funkcie PHP nevyhodia ArgumentCountError.
emit($message, $priority)	Podrží alebo odošle, vráti $this.
flush()	Odošle buffer vnútornému emitteru v poradí volaní emit(), každú správu s jej prioritou. Pri prázdnom bufferi vnútorný emitter nevolá.
discard(): int	Zahodí buffer bez odoslania a vráti počet zahodených správ.
count(): int	Počet podržaných správ; BufferedEmitter je Countable.

Správu emitnutú počas flush(), napríklad listenerom vnútorného emittera, odošle ten istý flush(). flush() zavolaný počas iného flush() hneď skončí, takže sa nič neodošle dvakrát.

Keď flush() vyhodí výnimku

flush() sa zastaví na prvej správe, ktorú vnútorný emitter neodošle, zvyčajne preto, že broker nebeží. Správy pred ňou sú odoslané a zostanú odoslané. Neodoslaná správa a zvyšok zostanú v bufferi, count() ich ukáže a flush() vyhodí tú istú výnimku.

Transakcia je v tej chvíli už commitnutá, takže riadky existujú a správy nie. Nenechajte výnimku zmiznúť:

1. Zalogujte ju, pokojne aj s count().
2. Zavolajte flush() znova. Odošle len to, čo zostalo.
3. Keď zlyhá aj to, zavolajte discard() a stratené správy riešte tak, ako aplikácia rieši iný zlyhaný vedľajší efekt: cron, ktorý nájde riadky bez príznaku spracovania, alebo alert.

Buffer, ktorý nikto neodošle ani nezahodí, odošle až ďalší úspešný flush() v tom istom procese po ďalšom commite. V dlho bežiacom procese (worker, consumer fronty) to môže byť oveľa neskôr, preto vždy skončite jednou z týchto dvoch metód.

Nie je to transactional outbox

Buffer žije v PHP procese. Keď proces medzi commitom a flush() zomrie (fatal error, OOM kill, deploy), správy sa stratia, hoci riadky sú commitnuté. Transactional outbox zapíše správy do tabuľky v tej istej transakcii a odosiela ich samostatný relay, ktorý toto prežije. BufferedEmitter to nerobí. Rieši častejšiu chybu, worker rýchlejší ako commit, jednou triedou a bez ďalšej tabuľky.

Príklad

examples/emit-after-commit registruje používateľov do SQLite, uvítaciu správu zaradí cez FileSystemDriver a nepotrebuje žiadne služby. make e2e spustí workera počas registrácií a overí, že pred commitom nie je nič vo fronte, že vrátená registrácia nič neodošle a že worker nájde každý riadok.

Malý obal nad PDO s callbackmi po commite a rollbacku:

<?php

declare(strict_types=1);

namespace App;

use Closure;
use PDO;
use Throwable;

/**
 * A small PDO wrapper with commit and rollback callbacks, like Nette Database or Laravel offer.
 */
final class Database
{
    public readonly PDO $pdo;

    /** @var list<Closure(): mixed> */
    private array $onCommit = [];

    /** @var list<Closure(): mixed> */
    private array $onRollback = [];

    public function __construct(string $file)
    {
        $this->pdo = new PDO('sqlite:' . $file, options: [PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION]);
        $this->pdo->exec('PRAGMA busy_timeout = 5000');
        $this->pdo->exec('CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY, email TEXT NOT NULL UNIQUE)');
    }

    public function isInTransaction(): bool
    {
        return $this->pdo->inTransaction();
    }

    public function registerOnCommit(callable $callback): void
    {
        $this->onCommit[] = $callback(...);
    }

    public function registerOnRollback(callable $callback): void
    {
        $this->onRollback[] = $callback(...);
    }

    /**
     * @template T
     * @param callable(PDO): T $work
     * @return T
     */
    public function transaction(callable $work): mixed
    {
        $this->pdo->beginTransaction();
        try {
            $result = $work($this->pdo);
            $this->pdo->commit();
        } catch (Throwable $e) {
            $this->pdo->rollBack();
            foreach ($this->onRollback as $callback) {
                $callback();
            }
            throw $e;
        }
        foreach ($this->onCommit as $callback) {
            $callback();
        }

        return $result;
    }
}

Emitter s hookom po commite, ktorý zlyhaný flush() zaloguje, raz zopakuje a potom zahodí:

<?php

declare(strict_types=1);

namespace App;

use Throwable;
use Tomaj\Hermes\BufferedEmitter;
use Tomaj\Hermes\Driver\FileSystemDriver;
use Tomaj\Hermes\Emitter;

final class Hermes
{
    public static function directory(): string
    {
        return getenv('HERMES_DIR') ?: __DIR__ . '/../var';
    }

    public static function database(): Database
    {
        @mkdir(self::directory(), 0o775, true);

        return new Database(self::directory() . '/app.sqlite');
    }

    public static function driver(): FileSystemDriver
    {
        return new FileSystemDriver(self::directory() . '/hermes', 'users', refreshInterval: 1);
    }

    /**
     * Emits right away outside a transaction; inside one, holds the messages until the commit.
     */
    public static function emitter(Database $database): BufferedEmitter
    {
        $emitter = new BufferedEmitter(new Emitter(self::driver()), [$database, 'isInTransaction']);

        $database->registerOnCommit(static function () use ($emitter): void {
            try {
                $emitter->flush();
            } catch (Throwable $e) {
                // the rows are committed, the unsent messages are still in the buffer
                error_log("hermes: flush after commit failed, retrying: {$e->getMessage()}");
                try {
                    $emitter->flush();
                } catch (Throwable $e) {
                    error_log(sprintf('hermes: dropping %d messages: %s', $emitter->discard(), $e->getMessage()));
                }
            }
        });
        $database->registerOnRollback([$emitter, 'discard']);

        return $emitter;
    }
}

Registrácia; s fail ako druhým argumentom sa vráti späť:

<?php

declare(strict_types=1);

use App\Hermes;
use Tomaj\Hermes\Message;

require __DIR__ . '/../vendor/autoload.php';

// php bin/register.php <email> [fail]
$email = $argv[1] ?? 'alice@example.com';
$fail = ($argv[2] ?? '') === 'fail';

$database = Hermes::database();
$emitter = Hermes::emitter($database);

try {
    $database->transaction(static function (PDO $pdo) use ($emitter, $email, $fail): void {
        $pdo->prepare('INSERT INTO users (email) VALUES (?)')->execute([$email]);
        $emitter->emit(new Message('user.registered', ['user_id' => (int) $pdo->lastInsertId()]));

        echo "in transaction: {$emitter->count()} buffered, " . count(glob(Hermes::directory() . '/hermes/users/new/*') ?: []) . " queued\n";
        if ($fail) {
            throw new RuntimeException('payment failed');
        }
    });
    echo "registered {$email}\n";
} catch (RuntimeException $e) {
    echo "rolled back {$email}: {$e->getMessage()}, {$emitter->count()} buffered\n";
}

Handler prečíta riadok, na ktorý správa ukazuje:

<?php

declare(strict_types=1);

namespace App;

use Tomaj\Hermes\Handler\HandlerInterface;
use Tomaj\Hermes\MessageInterface;

final readonly class WelcomeHandler implements HandlerInterface
{
    public function __construct(private Database $database, private string $outputFile) {}

    public function handle(MessageInterface $message): bool
    {
        $id = $message->getPayload()['user_id'] ?? null;
        $select = $this->database->pdo->prepare('SELECT email FROM users WHERE id = ?');
        $select->execute([$id]);
        $email = $select->fetchColumn();

        // without BufferedEmitter the worker could get here before the commit and find nothing
        $line = $email === false ? "missing user {$id}\n" : "welcome {$email}\n";
        file_put_contents($this->outputFile, $line, FILE_APPEND | LOCK_EX);

        return $email !== false;
    }
}

Frameworky

• Nette: registerOnCommit() a registerOnRollback() v setup služby.
• Laravel: listenery TransactionCommitted a TransactionRolledBack, ktoré kontrolujú transactionLevel().
• Symfony: Doctrine DBAL nemá callbacky po commite, preto volajte flush() po návrate z transactional().

Testovanie

BufferedEmitterTest pokrýva poradie a priority, predikát, callables [$object, 'method'], čiastočné zlyhanie s druhým flush(), správy emitnuté počas flush() aj vnorený flush(). BufferedEmitterIntegrationTest ho spúšťa pred Emitter a RedisSetDriver. Vo vlastných testoch skontrolujte count() po testovanom kóde a flush() alebo discard() zavolajte sami.

Časté otázky

Potrebujem BufferedEmitter, keď nepoužívam transakcie?

Nie. Bez transakcie je riadok commitnutý skôr, ako sa emit() spustí. S predikátom ako [$database, 'isInTransaction'] ten istý emitter mimo transakcie odosiela hneď, takže ho môžete používať všade.

Čo pri vnorených transakciách?

Závisí to od hookov. Odosielajte, až keď sa commitne najvonkajšia transakcia, a zahadzujte, až keď sa vráti späť najvonkajšia. Overte, ako vaša služba transakcií volá callbacky; v Laraveli kontrolujte v listeneri transactionLevel() === 0. V Laraveli správa emitnutá vo vnorenej transakcii, ktorá sa vrátila späť, zostane v bufferi a odošle sa po commite vonkajšej transakcie.

Funguje s každým driverom?

Áno. Obaľuje EmitterInterface, nie driver, takže funguje s každým driverom aj s vlastným emitterom.

Zdieľa sa buffer medzi requestami alebo workermi?

Nie. Každá inštancia BufferedEmitter má vlastný buffer v pamäti. Použite jednu inštanciu na kontajner, rovnako ako pri Emitter.

Prečo nerozšíriť Emitter?

Emitter je od 6.0 final. Dekorovanie EmitterInterface oddeľuje buffer od odosielania a funguje s ľubovoľným emitterom.