Laravel Job Batching và Chain: progress, xử lý lỗi, allowFailures và then/catch/finally

Khách hàng upload một file CSV 250.000 dòng. Bản import đầu tiên là một job duy nhất chạy vòng lặp qua hết các dòng: chạy 40 phút, dính worker timeout ở dòng 180.000, bị retry lại từ dòng số 0, và không ai trả lời được khách là import tới đâu rồi. Bản thứ hai chia file ra 250 job, hết timeout nhưng lại sinh ra câu hỏi mới. Làm sao biết tất cả đã xong? Chunk số 117 fail ba lần thì sao? Và làm sao gửi đúng một email "import xong rồi nha" thôi?

Laravel có hai công cụ cho chuyện này, và mỗi cái trả lời một câu hỏi khác nhau. Bus::chain chạy job lần lượt từng cái và dừng ngay ở lỗi đầu tiên. Bus::batch chạy job song song và theo dõi cả nhóm, có progress và các callback vòng đời. Chọn sai là ra mấy đơn hàng bị trừ tiền nửa chừng, hoặc import chạy xong mà hổng bao giờ báo hoàn tất.

Chain với Batch: khác nhau ở đâu

Chain là một chuỗi tuần tự. Job B chỉ được đẩy vô queue sau khi job A thành công. Nếu A fail hẳn (đã xài hết số lần retry), B và C bị bỏ luôn và callback catch của chain chạy. Xài chain khi bước sau phụ thuộc bước trước: giữ hàng trong kho, rồi trừ tiền thẻ, rồi gửi hóa đơn.

Batch là một nhóm. Tất cả job được đẩy vô cùng lúc và bao nhiêu worker cũng có thể nhận. Laravel lưu một dòng trong bảng job_batches với các bộ đếm (total_jobs, pending_jobs, failed_jobs) và cập nhật atomic mỗi khi một job chạy xong. Xài batch khi các job độc lập với nhau và mình quan tâm kết quả của cả nhóm: các chunk import, tạo thumbnail, gửi newsletter theo từng đợt.

Phần dễ nhầm nhất là vòng đời của batch:

Ba quy tắc cần nhớ:

• then chạy chỉ khi mọi job đều thành công. Chỉ cần một job fail hẳn là then không bao giờ chạy, kể cả khi đã bật allowFailures().
• catch chạy đúng một lần, ở job đầu tiên fail hẳn (sau khi xài hết $tries). Mấy lỗi sau không gọi lại nó nữa.
• finally chạy khi mọi job đã chạy xong, bất kể kết quả. Chỗ này mới là nơi để dọn dẹp và gửi thông báo "import đã xong".

Chuẩn bị cho Batch

Batch cần bảng job_batches. Laravel có sẵn lệnh tạo migration:

php artisan make:queue-batches-table
php artisan migrate

Job nào bỏ vô batch cũng phải xài trait Batchable. Trait này cho job hàm $this->batch(), trả về instance Batch để job kiểm tra batch đã bị cancel chưa, hoặc thêm job mới vô batch. Cách gọn nhất để ngưng làm việc sau khi bị cancel là job middleware SkipIfBatchCancelled:

namespace App\Jobs;

use App\Services\CustomerCsv;
use Illuminate\Bus\Batchable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Queue\Queueable;
use Illuminate\Queue\Middleware\SkipIfBatchCancelled;

final class ImportCustomerChunk implements ShouldQueue
{
    use Batchable, Queueable;

    public int $tries = 3;

    public function __construct(
        public readonly int $importId,
        public readonly int $offset,
        public readonly int $limit,
    ) {}

    /** Skip the work entirely once the batch has been cancelled. */
    public function middleware(): array
    {
        return [new SkipIfBatchCancelled()];
    }

    public function handle(CustomerCsv $csv): void
    {
        foreach ($csv->rows($this->importId, $this->offset, $this->limit) as $row) {
            $csv->upsertCustomer($row);
        }
    }

    /** @return list<int> */
    public function backoff(): array
    {
        return [10, 60];
    }
}

Lưu ý là job bị middleware bỏ qua vẫn được tính là đã xử lý. Chuyện này ảnh hưởng tới progress bar: một batch đã cancel vẫn có thể lên 100% "progress" mà gần như không làm gì.

Chain cho các bước phụ thuộc nhau

Checkout là ví dụ kinh điển cho chain. Trừ tiền trước khi giữ hàng là bug, còn gửi hóa đơn cho một lần trừ tiền thất bại thì còn tệ hơn:

use App\Jobs\ChargePayment;
use App\Jobs\ReleaseStock;
use App\Jobs\ReserveStock;
use App\Jobs\SendReceipt;
use Illuminate\Support\Facades\Bus;

$orderId = $order->id;

Bus::chain([
    new ReserveStock($orderId),
    new ChargePayment($orderId),
    new SendReceipt($orderId),
])
    ->onQueue('orders')
    ->catch(function (Throwable $e) use ($orderId): void {
        // Runs once, when a job in the chain fails for good.
        // The jobs after the failed one are never executed.
        ReleaseStock::dispatch($orderId);
        logger()->error('Checkout chain failed', ['order_id' => $orderId, 'error' => $e->getMessage()]);
    })
    ->dispatch();

Closure catch được serialize và lưu chung với chain, nên chỉ capture giá trị đơn giản như $orderId, đừng capture $this hay nguyên một cục model to. Nếu ChargePayment throw ở lần thử cuối, SendReceipt sẽ không bao giờ được đẩy vô queue, và job bù trừ sẽ trả lại hàng vô kho.

Batch cho việc song song có progress

Đây là phần import, chia thành từng chunk, với đủ bốn callback cập nhật vô bản ghi CustomerImport mà UI đang poll:

namespace App\Services;

use App\Jobs\ImportCustomerChunk;
use App\Models\CustomerImport;
use Illuminate\Bus\Batch;
use Illuminate\Support\Facades\Bus;
use Throwable;

final class CustomerImporter
{
    public function start(int $importId, int $rows, int $chunkSize = 1000): Batch
    {
        $jobs = [];

        for ($offset = 0; $offset < $rows; $offset += $chunkSize) {
            $jobs[] = new ImportCustomerChunk($importId, $offset, $chunkSize);
        }

        return Bus::batch($jobs)
            ->name("customer-import-{$importId}")
            ->onQueue('imports')
            ->allowFailures()
            ->progress(function (Batch $batch) use ($importId): void {
                CustomerImport::whereKey($importId)->update(['progress' => $batch->progress()]);
            })
            ->then(function (Batch $batch) use ($importId): void {
                // Only when EVERY job succeeded.
                CustomerImport::whereKey($importId)->update(['status' => 'completed']);
            })
            ->catch(function (Batch $batch, Throwable $e) use ($importId): void {
                // Only for the FIRST job that failed for good.
                CustomerImport::whereKey($importId)->update(['status' => 'has_errors']);
            })
            ->finally(function (Batch $batch) use ($importId): void {
                // When every job has run, success or not.
                CustomerImport::whereKey($importId)->update([
                    'finished_at' => now(),
                    'failed_chunks' => $batch->failedJobs,
                ]);
            })
            ->dispatch();
    }
}

Sao lại bật allowFailures() ở đây? Nếu không bật, chunk đầu tiên fail hẳn sẽ cancel cả batch, và mọi chunk còn trong queue đều bị middleware bỏ qua. Với kiểu "được ăn cả, ngã về không" như checkout thì đúng ý. Còn với import, mất 1 chunk trên 250 không có lý do gì phải vứt luôn 249 chunk kia; mình muốn chạy cho xong, báo lại failed_chunks, rồi để support retry phần lỗi bằng php artisan queue:retry-batch {id}.

Callback progress chạy sau mỗi job thành công. Nó ghi DB một chút, vài trăm job thì không sao. Lên tới hàng chục ngàn job thì nên giới hạn lại (chỉ ghi khi phần trăm nguyên thay đổi) hoặc bỏ luôn và cho UI đọc thẳng batch:

use Illuminate\Support\Facades\Bus;
use Illuminate\Support\Facades\Route;

Route::get('/imports/batches/{batchId}', function (string $batchId) {
    $batch = Bus::findBatch($batchId);

    abort_if($batch === null, 404);

    return [
        'name' => $batch->name,
        'progress' => $batch->progress(),
        'total' => $batch->totalJobs,
        'pending' => $batch->pendingJobs,
        'failed' => $batch->failedJobs,
        'finished' => $batch->finished(),
        'cancelled' => $batch->cancelled(),
    ];
})->middleware('auth');

Bus::findBatch() đọc dòng trong job_batches, nên endpoint này rẻ và luôn khớp với những gì worker đã ghi nhận. Nhớ để sau auth: batch ID không phải là token bí mật.

Kết hợp Chain và Batch

Workflow thực tế hiếm khi thuần một kiểu. Laravel cho lồng hai chiều. Một mảng bên trong batch là một chain; một batch bên trong chain là một bước phải chờ cả batch chạy xong:

use App\Jobs\BuildSearchIndex;
use App\Jobs\ExportInvoices;
use App\Jobs\ExportPayments;
use App\Jobs\NotifyFinanceTeam;
use App\Jobs\ZipExports;
use Illuminate\Support\Facades\Bus;

// A batch whose items are chains: each region runs its two steps in order,
// while the regions run in parallel with each other.
Bus::batch([
    [new ExportInvoices('eu'), new ExportPayments('eu')],
    [new ExportInvoices('us'), new ExportPayments('us')],
    [new ExportInvoices('apac'), new ExportPayments('apac')],
])->name('month-end-exports')->dispatch();

// A chain that contains a batch: the zip step waits for the whole batch.
Bus::chain([
    Bus::batch([
        new ExportInvoices('eu'),
        new ExportInvoices('us'),
    ]),
    new ZipExports(),
    new NotifyFinanceTeam(),
])->dispatch();

// A running job can also grow its own batch.
// Inside a Batchable job's handle():
// $this->batch()->add([new BuildSearchIndex($this->importId)]);

Kiểu thứ hai giúp bỏ được cái hack quen thuộc là dispatch job tiếp theo trong callback then của batch. Khi batch là một bước của chain, sự phụ thuộc hiện rõ ngay chỗ dispatch, không bị giấu trong callback.

Viết test cho Batch và Chain

Feature test hiếm khi cần worker thật. Bus::fake() ghi lại mọi thứ đã dispatch và cho sẵn assertion cho cả hai kiểu:

use App\Jobs\ChargePayment;
use App\Jobs\ImportCustomerChunk;
use App\Jobs\ReserveStock;
use App\Jobs\SendReceipt;
use App\Models\Order;
use App\Services\CheckoutService;
use App\Services\CustomerImporter;
use Illuminate\Bus\PendingBatch;
use Illuminate\Support\Facades\Bus;

it('dispatches one batch job per chunk and tolerates failures', function () {
    Bus::fake();

    app(CustomerImporter::class)->start(importId: 42, rows: 2500, chunkSize: 1000);

    Bus::assertBatched(function (PendingBatch $batch): bool {
        return $batch->name === 'customer-import-42'
            && $batch->jobs->count() === 3
            && $batch->jobs->every(fn ($job) => $job instanceof ImportCustomerChunk)
            && $batch->allowsFailures();
    });
});

it('runs checkout as an ordered chain', function () {
    Bus::fake();

    $order = Order::factory()->create();
    app(CheckoutService::class)->checkout($order);

    Bus::assertChained([
        ReserveStock::class,
        ChargePayment::class,
        SendReceipt::class,
    ]);
});

Mấy test này kiểm tra phần nối dây: đúng số chunk, đúng thứ tự, đã bật allowFailures. Bản thân các job nên có unit test riêng, gọi thẳng handle() với dependency giả. Với callback, tách phần thân ra thành class invokable nhỏ hoặc method trong service để test được mà không cần queue.

Những lỗi hay gặp

• Chờ then chạy dù có job fail. Bật allowFailures() thì batch vẫn chạy tiếp, nhưng then vẫn chỉ chạy khi mọi thứ thành công. Logic "import xong" phải để trong finally và đọc $batch->failedJobs ở đó.
• Capture $this hoặc model trong callback. Callback được serialize vô dòng job_batches. Capture ID rồi query lại bên trong closure.
• Quên là batch đã bị cancel. Không có SkipIfBatchCancelled hoặc một dòng if ($this->batch()?->cancelled()) return;, các job còn trong queue vẫn làm việc thật dù batch đã cancel vì lỗi.
• Retry khác với fail. Job throw mà còn lượt thử thì chưa tính là batch fail. catch chỉ chạy sau lần thử cuối. Đặt $tries và backoff() cho có chủ đích.
• Bảng job_batches phình mãi. Batch nào cũng để lại một dòng. Lên lịch prune:

use Illuminate\Support\Facades\Schedule;

// routes/console.php
Schedule::command('queue:prune-batches --hours=48 --unfinished=72 --cancelled=72')->daily();

• Dispatch bên trong DB transaction. Nếu tạo dòng CustomerImport và dispatch batch trong cùng transaction, worker có thể chạy trước khi commit và không tìm thấy dòng đó. Xài afterCommit() cho job hoặc dispatch sau khi transaction xong.

Khi nào xài cái nào, khi nào không nên

Xài chain khi các bước phải chạy theo thứ tự và bước sau vô nghĩa nếu bước trước fail. Giữ chain ngắn thôi; một chain 15 bước thực chất là workflow engine trá hình và rất khó chạy tiếp từ giữa chừng.

Xài batch khi công việc là tập hợp các phần độc lập và mình cần câu trả lời ở mức cả nhóm: chạy tới đâu rồi, có gì fail không, chạy một việc đúng một lần ở cuối.

Đừng xài cái nào hết nếu một job là đủ. Task chạy vài giây mà đem chia nhỏ thì tốn thêm một round trip DB mỗi job chỉ để ghi sổ. Và đừng biến batch thành state machine sống dài ngày cho quy trình nghiệp vụ; state đó nên nằm trong bảng của mình, còn job chỉ là mấy bước nhỏ, retry được, đẩy state đó đi tới.