Transactional outbox trong Laravel: hết cảnh mất event giữa database và broker

Khách thanh toán một đơn hàng. Dòng order đã commit, tiền đã trừ, vậy mà service kho không bao giờ giao. Log cho thấy message orders.placed chưa từng được publish: worker PHP bị deploy giết đúng hai mili giây sau khi commit. Tuần trước thì ngược lại: message publish trước, transaction database rollback vì deadlock, và kho bắt đầu soạn một đơn không hề tồn tại.

Hai sự cố đó là cùng một bug. Code của mình ghi vô hai hệ thống, database và message broker, mà không có transaction nào bao trùm cả hai. Ghi bên nào trước cũng vậy, crash giữa hai lần ghi là hai bên lệch nhau. Transactional outbox pattern dẹp luôn lần ghi thứ hai ra khỏi request.

Bài toán dual-write

Đây là đoạn code gây ra sự cố đầu tiên. Nhìn rất hợp lý, và nó pass mọi test mình từng viết:

use App\Models\Order;
use Illuminate\Support\Facades\DB;

$order = DB::transaction(function () use ($cart) {
    $order = Order::createFromCart($cart);
    $order->items()->createMany($cart->lines());

    return $order;
});

// The process can die right here: OOM, deploy, network blip.
// The order exists, but nobody downstream ever hears about it.
$broker->publish('orders.placed', ['order_id' => $order->id]);

Dời publish vô trong transaction thì lại dính sự cố thứ hai: broker nhận message rồi, sau đó commit fail, mà mình đâu có "unpublish" được. DB::afterCommit() và event ShouldDispatchAfterCommit của Laravel cải thiện thứ tự chứ không cải thiện bảo đảm: callback vẫn chạy trong cùng process PHP sau commit, nên crash, timeout hay broker sập đúng lúc đó vẫn làm mất message.

Không có cách sắp thứ tự khéo léo nào sửa được chuyện này. Cách sửa là chỉ còn một lần ghi quan trọng, nằm trong một transaction.

Bảng outbox

Thay vì nói chuyện với broker, request ghi message vô một bảng trong cùng database, cùng transaction với dữ liệu nghiệp vụ. Hoặc cả order lẫn message đều tồn tại, hoặc không cái nào tồn tại. Một process relay riêng đọc bảng đó rồi publish lên broker, retry tới khi thành công.

use Illuminate\Database\Migrations\Migration;
use Illuminate\Database\Schema\Blueprint;
use Illuminate\Support\Facades\Schema;

return new class extends Migration
{
    public function up(): void
    {
        Schema::create('outbox_messages', function (Blueprint $table) {
            $table->id();
            $table->uuid('message_id')->unique();
            $table->string('topic');
            $table->string('aggregate_type');
            $table->string('aggregate_id');
            $table->json('payload');
            $table->unsignedSmallInteger('attempts')->default(0);
            $table->timestamp('available_at')->useCurrent();
            $table->timestamp('published_at')->nullable();
            $table->timestamp('created_at')->useCurrent();

            $table->index(['published_at', 'available_at', 'id']);
        });
    }

    public function down(): void
    {
        Schema::dropIfExists('outbox_messages');
    }
};

Mấy cột đáng giải thích:

• message_id là UUID đi theo message. Consumer dựa vô nó để phát hiện trùng.
• aggregate_type và aggregate_id cho biết message nói về cái gì. Nhờ vậy giữ được thứ tự theo từng entity và debug dễ hơn nhiều.
• available_at và attempts cho mình backoff mà không cần queue retry riêng.
• Index (published_at, available_at, id) biến câu "các message pending cũ nhất" thành range scan trên index, kể cả khi bảng có hàng triệu dòng đã publish.

Ghi message trong cùng transaction

Một service nhỏ lo ghi dòng outbox. Nó từ chối chạy ngoài transaction, nhờ vậy lỗi phổ biến nhất biến thành exception ngay lúc dev:

namespace App\Outbox;

use Illuminate\Support\Facades\DB;
use Illuminate\Support\Str;
use LogicException;

final class Outbox
{
    public function record(string $topic, string $aggregateType, string|int $aggregateId, array $payload): string
    {
        if (DB::transactionLevel() === 0) {
            throw new LogicException('Outbox messages must be recorded inside a database transaction.');
        }

        $messageId = (string) Str::uuid();

        DB::table('outbox_messages')->insert([
            'message_id' => $messageId,
            'topic' => $topic,
            'aggregate_type' => $aggregateType,
            'aggregate_id' => (string) $aggregateId,
            'payload' => json_encode($payload, JSON_THROW_ON_ERROR),
        ]);

        return $messageId;
    }
}

Use case ghi message ngay cạnh các thao tác nghiệp vụ:

namespace App\Orders;

use App\Models\Cart;
use App\Models\Order;
use App\Outbox\Outbox;
use Illuminate\Support\Facades\DB;

final class PlaceOrder
{
    public function __construct(private Outbox $outbox) {}

    public function __invoke(Cart $cart): Order
    {
        return DB::transaction(function () use ($cart) {
            $order = Order::createFromCart($cart);
            $order->items()->createMany($cart->lines());

            $this->outbox->record('orders.placed', 'order', $order->id, [
                'order_id' => $order->id,
                'customer_id' => $order->customer_id,
                'total' => $order->total,
                'placed_at' => $order->created_at->toIso8601String(),
            ]);

            return $order;
        });
    }
}

Có gì trong closure throw là order, item và dòng outbox biến mất cùng lúc. Commit thành công thì message chắc chắn sẽ được publish, sớm hay muộn, kể cả khi process PHP chết ngay một micro giây sau.

Relay worker

Relay là một vòng lặp: lấy một batch message pending, publish từng cái, đánh dấu đã publish. Đây là nguyên luồng, từ request tới consumer:

Trong Laravel, relay là một Artisan command chạy dưới Supervisor (hoặc gọi --once theo schedule nếu lưu lượng thấp):

namespace App\Console\Commands;

use App\Outbox\MessagePublisher;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\DB;
use Throwable;

final class RelayOutbox extends Command
{
    protected $signature = 'outbox:relay {--batch=100} {--once}';

    protected $description = 'Publish pending outbox messages to the broker';

    public function handle(MessagePublisher $publisher): int
    {
        do {
            $sent = $this->relayBatch($publisher, (int) $this->option('batch'));

            if ($sent === 0 && ! $this->option('once')) {
                usleep(500_000);
            }
        } while (! $this->option('once'));

        return self::SUCCESS;
    }

    private function relayBatch(MessagePublisher $publisher, int $limit): int
    {
        return DB::transaction(function () use ($publisher, $limit) {
            $messages = DB::table('outbox_messages')
                ->whereNull('published_at')
                ->where('available_at', '<=', now())
                ->orderBy('id')
                ->limit($limit)
                ->lock('for update skip locked')
                ->get();

            $sent = 0;

            foreach ($messages as $message) {
                try {
                    $publisher->publish($message->topic, $message->message_id, json_decode($message->payload, true));
                } catch (Throwable $e) {
                    report($e);
                    DB::table('outbox_messages')->where('id', $message->id)->update([
                        'attempts' => $message->attempts + 1,
                        'available_at' => now()->addSeconds(min(300, 2 ** $message->attempts)),
                    ]);

                    break;
                }

                DB::table('outbox_messages')->where('id', $message->id)->update(['published_at' => now()]);
                $sent++;
            }

            return $sent;
        });
    }
}

Mấy quyết định thiết kế trong command đó:

• FOR UPDATE SKIP LOCKED. Kiểu gì mình cũng chạy hơn một relay, ít nhất là lúc deploy. Lock dòng kèm SKIP LOCKED (MySQL 8.0+ và PostgreSQL) cho nhiều worker chia nhau bảng mà không publish trùng một dòng, cũng không chặn nhau.
• Order theo id và dừng ở lỗi đầu tiên. Break khỏi vòng lặp giữ message đúng thứ tự commit, quan trọng khi consumer trông chờ orders.placed tới trước orders.cancelled. Cần song song thì partition theo aggregate_id, đừng bỏ thứ tự toàn cục.
• Backoff lũy thừa qua available_at. Broker sập không biến thành vòng lặp nóng nện vô nó liên tục.
• Publish rồi mới đánh dấu. Worker crash sau khi publish mà trước khi update commit thì lần chạy sau message được publish lại. Đó là cái giá của pattern: giao ít nhất một lần (at-least-once). Giao đúng một lần qua mạng là không có, nên consumer phải tự lo.

Giữ batch nhỏ: lock bị giữ trong lúc nói chuyện với broker, nên 100 message mỗi cái 50 ms là giữ lock 5 giây.

Consumer idempotent

Vì relay có thể giao một message hai lần, mà đa số broker cũng giao lại khi consumer crash, consumer nào cũng phải chạy hai lần với cùng message mà vẫn an toàn. Cách chắc ăn là bảng processed_messages có unique key trên message_id, ghi trong cùng transaction với side effect của consumer:

namespace App\Warehouse;

use Illuminate\Support\Facades\DB;

final class HandleOrderPlaced
{
    public function __construct(private ShipmentPlanner $planner) {}

    public function __invoke(string $messageId, array $payload): void
    {
        DB::transaction(function () use ($messageId, $payload) {
            $inserted = DB::table('processed_messages')->insertOrIgnore([
                'message_id' => $messageId,
                'consumer' => 'warehouse.order-placed',
                'processed_at' => now(),
            ]);

            if ($inserted === 0) {
                return;
            }

            $this->planner->planFor((int) $payload['order_id']);
        });
    }
}

insertOrIgnore trả về số dòng đã insert. Bằng 0 nghĩa là một lần giao khác của cùng message đã tới trước, nên return luôn mà không làm gì. Vì dòng dedupe và kế hoạch giao hàng commit cùng nhau, crash giữa chừng thì không cái nào còn lại, và lần giao lại sẽ chạy đủ handler.

Handler vốn idempotent ("đổi trạng thái thành đã giao", upsert theo order_id) thì khỏi cần bảng. "Cộng điểm thưởng" hay "gửi email" thì không được.

Kiểm thử outbox

Mấy tính chất cần chứng minh chính là mấy cái đã fail trên production: message được ghi nguyên tử cùng order, rollback thì không ghi gì, broker lỗi thì được retry chứ không mất. Bind một publisher giả vô container rồi chạy relay với --once:

use App\Models\Cart;
use App\Orders\PlaceOrder;
use App\Outbox\MessagePublisher;
use Illuminate\Support\Facades\DB;

final class FakePublisher implements MessagePublisher
{
    public array $sent = [];

    public bool $failNext = false;

    public function publish(string $topic, string $messageId, array $payload): void
    {
        if ($this->failNext) {
            $this->failNext = false;

            throw new RuntimeException('broker down');
        }

        $this->sent[] = [$topic, $messageId, $payload];
    }
}

beforeEach(function () {
    $this->publisher = new FakePublisher();
    $this->app->instance(MessagePublisher::class, $this->publisher);
});

it('writes the order and its outbox message in one transaction', function () {
    $order = app(PlaceOrder::class)(Cart::factory()->withLines(2)->create());

    $this->assertDatabaseHas('outbox_messages', [
        'topic' => 'orders.placed',
        'aggregate_id' => (string) $order->id,
        'published_at' => null,
    ]);
});

it('publishes pending messages and retries after a broker failure', function () {
    app(PlaceOrder::class)(Cart::factory()->withLines(1)->create());
    $this->publisher->failNext = true;

    $this->artisan('outbox:relay', ['--once' => true])->assertSuccessful();
    expect($this->publisher->sent)->toBeEmpty();

    $this->travel(5)->seconds();
    $this->artisan('outbox:relay', ['--once' => true])->assertSuccessful();

    expect($this->publisher->sent)->toHaveCount(1)
        ->and(DB::table('outbox_messages')->whereNull('published_at')->count())->toBe(0);
});

$this->travel() đẩy đồng hồ của Laravel qua khỏi khoảng backoff, nên test retry khỏi phải sleep. Nên test thêm case rollback (throw trong một transaction bọc ngoài, assert outbox_messages rỗng) và consumer: gọi nó hai lần với cùng message_id rồi assert side effect chỉ xảy ra một lần.

Những lỗi hay gặp

• Ghi outbox ngoài transaction. Gọi record() sau khi DB::transaction() đã return là kéo bài toán dual-write quay lại nguyên xi. Cái guard transactionLevel() bắt lỗi này sớm.

• Payload bự chà bá. Outbox không phải document store. Chỉ để ID và mấy field consumer cần, đừng nhồi nguyên model kèm relation.

• Bảng phình không giới hạn. Dòng đã publish dồn lên rất nhanh. Đặt schedule chạy mỗi ngày, xóa các dòng đã publish quá bảy ngày, mỗi lần vài ngàn dòng.

• Tưởng là exactly-once. Consumer nào mình không kiểm soát rồi cũng sẽ nhận message trùng. Gửi message_id trong header và ghi rõ đó là key để dedupe.

• Không bao giờ restart relay. Thoát sau N batch giống queue:work --max-jobs, để Supervisor khởi động lại với code và config mới.

• Không monitor gì hết. Đặt alert theo tuổi của message chưa publish lâu nhất. Nó tăng dần là relay đang kẹt, và đơn hàng đang âm thầm chờ.

Khi nào nên và không nên xài outbox

Xài transactional outbox khi:

• thay đổi trong database bắt buộc phải kèm một message tới service hay broker khác, và mất hay bịa ra message đó gây thiệt hại thật: đơn hàng, thanh toán, tồn kho, trạng thái tài khoản;
• mình tích hợp service qua Kafka, RabbitMQ, SQS hay webhook và không thể chung transaction database với tụi nó;
• mình muốn có sẵn lịch sử những gì đã phát ra và lúc nào, coi như miễn phí.

Có lẽ không cần khi:

• side effect kiểu "được thì tốt": ping analytics, làm nóng cache, email chào mừng mà lâu lâu sót một cái cũng chấp nhận được. Dispatch kiểu afterCommit đơn giản hơn;
• mọi thứ nằm trong một app Laravel với queue driver database. Job dispatch bên trong transaction trên queue database vốn đã được ghi nguyên tử cùng dữ liệu, coi như một outbox rồi;
• mình đã chạy change data capture (Debezium đọc binlog). CDC thay được relay polling, dù bảng outbox vẫn là một hợp đồng rõ ràng đáng giữ.

Pattern này tốn một bảng, một worker và một bước check trùng ở mỗi consumer. Đổi lại, chuyện "đơn đã trả tiền mà kho không hay biết gì" không còn là sự cố mình ngồi debug lúc 2 giờ sáng.