Job Chains

Tidak semua rangkaian pekerjaan boleh berjalan paralel. Sebelum invoice dibuat, pembayaran harus ditagih; dan sebelum ditagih, pesanan harus tervalidasi. Job chains menjamin urutan itu di level antrean: job kedua baru didorong ke antrean setelah job pertama sukses, job ketiga menunggu job kedua, dan seterusnya. Ini berbeda dari job batching yang mengeksekusi semuanya sekaligus demi satu laporan progres. Pada chapter ini kita buktikan urutan tersebut lewat log worker, lalu kita lihat apa yang terjadi ketika job di tengah chain gagal.

Mengirim Chain #

Pipeline pesanan kita terdiri dari tiga job: validasi, penagihan, dan pembuatan invoice. Ketiganya digabung dengan Bus::chain():

use App\Jobs\ChargePaymentJob;
use App\Jobs\GenerateInvoiceJob;
use App\Jobs\ValidateOrderJob;
use Illuminate\Support\Facades\Bus;
use Throwable;

Bus::chain([
    new ValidateOrderJob('INV-100'),
    new ChargePaymentJob('INV-100'),
    new GenerateInvoiceJob('INV-100'),
])->catch(function (Throwable $e) {
    logger()->info('catch(): '.$e->getMessage());
})->dispatch();

Method catch() menerima Throwable penyebab kegagalan — satu-satunya tempat bereaksi bila salah satu job chain melempar exception. Yang menarik adalah isi tabel jobs sesaat setelah dispatch:

use Illuminate\Support\Facades\DB;

$job = DB::table('jobs')->first();
echo 'jobs_table: '.DB::table('jobs')->count()
    .' baris, payload displayName: '.json_decode($job->payload)->displayName."\n";
$chain = json_decode($job->payload)->data->command ?? '';
echo 'payload_memuat_sisa_chain: '
    .var_export(str_contains($chain, 'GenerateInvoiceJob'), true)."\n";

Output yang Dihasilkan:

jobs_table: 1 baris, payload displayName: App\Jobs\ValidateOrderJob
payload_memuat_sisa_chain: true

Hanya ValidateOrderJob yang berada di antrean — dua job sisanya tersimpan terserialisasi di dalam properti chained payload job pertama. Setiap kali sebuah job chain sukses, worker mendorong job berikutnya dari payload ke antrean, membawa sisa chain di payload barunya. Satu putaran queue:work --stop-when-empty memproses ketiganya, dan log membuktikan urutannya:

[INV-100] Validasi pesanan selesai.
[INV-100] Pembayaran ditagih.
[INV-100] Invoice dibuat.

Closure dan Penambahan Job dari Dalam Chain #

Di antara job, kita bisa menyisipkan Closure untuk logika singkat tanpa membuat class baru — chain berlanjut begitu closure selesai:

Bus::chain([
    function (): void {
        logger()->info('Menyiapkan invoice (closure)...');
    },
    new GenerateInvoiceJob('INV-400'),
])->dispatch();

Output yang Dihasilkan (setelah worker berjalan):

Menyiapkan invoice (closure)...
[INV-400] Invoice dibuat.

Ketika sebuah job perlu menambahkan langkah lanjutan berdasarkan kondisi saat runtime, method appendToChain() dari dalam handle() menyelipkan job baru ke ujung chain, sedangkan prependToChain() memasukkannya tepat setelah job yang sedang berjalan:

public function handle(): void
{
    parent::handle();

    $this->appendToChain(new SendReceiptJob($this->orderId));
}

Pada sandbox, class turunan GenerateInvoiceJob dengan appendToChain() di atas menghasilkan urutan log: Invoice dibuat. lalu Receipt dikirim ke pelanggan. — bukti job baru benar-benar menempel di ekor chain.

Chain vs Batch #

Job Tengah Gagal, Chain Berhenti #

Sekarang kita sabotase pipeline-nya: ChargePaymentJob menerima flag simulasiGagal yang membuatnya melempar exception. Chain yang sama didispatch ulang dengan simulasiGagal: true pada job kedua:

Bus::chain([
    new ValidateOrderJob('INV-200'),
    new ChargePaymentJob('INV-200', simulasiGagal: true),
    new GenerateInvoiceJob('INV-200'),
])->catch(function (Throwable $e) {
    logger()->info('catch(): '.$e->getMessage());
})->dispatch();

Setelah queue:work --stop-when-empty selesai, log dan tabel menceritakan akhirnya:

[INV-200] Validasi pesanan selesai.
catch(): Payment gateway menolak pesanan INV-200

jobs_table: 0
failed_jobs: 1 baris

Job pertama sukses, job kedua gagal dan berpindah ke failed_jobs (dengan $tries = 1 job langsung dianggap gagal permanen), dan GenerateInvoiceJob tidak pernah didispatch — chain berhenti total. Callback catch() menerima exception penyebabnya, persiapan yang tepat untuk mengirim notifikasi atau mengembalikan stok.

Koneksi, Queue, dan Delay #

Seluruh anggota chain bisa diarahkan ke antrean tertentu dengan onQueue(), dan delay() menunda eksekusi job pertama:

Bus::chain([
    new ValidateOrderJob('INV-300'),
    new ChargePaymentJob('INV-300'),
    new GenerateInvoiceJob('INV-300'),
])->onQueue('order')->delay(60)->dispatch();

$job = DB::table('jobs')->first();
echo 'queue: '.$job->queue
    .', delay_dari_sekarang: '.($job->available_at - time())." detik\n";

Output yang Dihasilkan:

queue: order, delay_dari_sekarang: 60 detik

Kolom queue job pertama terisi order — nilai ini terbawa sebagai chainQueue sehingga job berikutnya juga masuk antrean order. Worker harus mengawasi antrean itu secara eksplisit: php artisan queue:work --queue=order,default. Delay berlaku untuk job pertama saja, karena job berikutnya memang baru lahir ketika pendahulunya selesai.

Urutan sudah kita kunci. Selanjutnya kita hadapi musuh berbeda: job yang sah untuk dijalankan, tetapi tidak boleh terlalu sering — misalnya panggilan ke API pihak ketiga yang membatasi laju request. Pembatasan itulah topik chapter Rate Limited Jobs , dan mekanisme chain lengkapnya terdokumentasi di dokumentasi Queues .