Belajar Axum - Background Tasks & Concurrency

#Pendahuluan

Ada pekerjaan yang tidak boleh menahan response: kirim email, generate thumbnail, panggil webhook, proses batch. Episode ini membahas background tasks di Axum/tokio: bagaimana melepas kerja berat dari request path, mengantrikannya, dan mengeksekusinya secara konkuren — tanpa membocorkan resource.

Mengapa ini fase "advanced"? Karena background task yang salah ditulis adalah sumber bug halus: task yang di-spawn tapi tidak di-track, worker yang mati diam-diam saat aplikasi shutdown, dan antrian yang memblokir runtime. Episode ini membangun dari pola paling sederhana sampai job queue berbasis Redis.

#Fire-and-Forget: tokio::spawn

Pola paling dasar. Handler melempar kerja ke tokio runtime dan segera mengembalikan response:

Fire-and-forget

use std::sync::Arc;
use tokio::task::JoinHandle;
use crate::state::AppState;
 
pub async fn send_welcome_email(
    State(state): State<AppState>,
    Json(req): Json<WelcomeRequest>,
) -> AppResult<(StatusCode, Json<serde_json::Value>)> {
    let db = state.db.clone();
    let email = req.email.clone();
 
    tokio::spawn(async move {
        // pekerjaan latar: lambat, tidak perlu menunggu
        let user = sqlx::query!(
            "SELECT id, username FROM users WHERE email = $1", email
        )
        .fetch_one(&db)
        .await
        .ok();
 
        if let Some(user) = user {
            email_service::send_welcome(user.username, email).await.ok();
        }
    });
 
    Ok((StatusCode::ACCEPTED, Json(serde_json::json!({ "status": "queued" }))))
}

Poin penting:

tokio::spawn — melempar task ke runtime; handler langsung selesai. Response 202 Accepted (bukan 200) memberi sinyal "diterima, belum selesai".
async move — task mengambil ownership nilai yang di-clone ke dalamnya (db, email); jangan meminjam variabel handler.
Handler harus Send — nilai yang masuk task harus bisa dipindah antar thread.

Warning

Fire-and-forget tanpa tracking berbahaya: task yang error (panic, query gagal) tidak terlihat. Jika task di-spawn untuk hal penting, log error-nya (let _ = tokio::spawn(...).await pada worker, atau instrument). Mulai dari tracing::error! di dalam task — episode 17 — agar kegagalan tidak diam.

#Worker + mpsc: Antrian dalam Proses

Fire-and-forget melempar task per request — kurang terkontrol. Pola yang lebih baik: mpsc channel + worker task yang memproses antrian dengan kecepatan terkendali.

Worker dengan mpsc

use tokio::sync::mpsc;
 
#[derive(Clone)]
pub struct AppState {
    pub db: sqlx::PgPool,
    pub jobs: mpsc::Sender<Job>,
}
 
pub enum Job {
    SendEmail { to: String, subject: String },
    GenerateThumbnail { image_path: String },
}
 
pub fn spawn_worker(mut rx: mpsc::Receiver<Job>, db: sqlx::PgPool) -> tokio::task::JoinHandle<()> {
    tokio::spawn(async move {
        while let Some(job) = rx.recv().await {
            match job {
                Job::SendEmail { to, subject } => {
                    if let Err(err) = email_service::send(to, subject).await {
                        tracing::error!(error = %err, to, "gagal kirim email");
                    }
                }
                Job::GenerateThumbnail { image_path } => {
                    // proses lambat ...
                }
            }
        }
    })
}

Kunci pola ini:

mpsc (multi-producer, single-consumer) — banyak handler mengirim, satu worker memproses. Berbeda dengan broadcast (episode 12) yang menyalin ke semua subscriber.
Buffer channel (mpsc::channel(n)) — jika worker lebih lambat dari produksi, channel menahan maksimal n job; send().await menunggu saat penuh. Pilih n dengan mempertimbangkan memori.
Worker adalah satu task panjang — ia hidup sepanjang aplikasi; while let Some(job) = rx.recv().await berakhir saat semua sender dibuang (shutdown).

Pasang di main:

Buat channel + worker di main

let (tx, rx) = mpsc::channel(1024);
let worker = spawn_worker(rx, db.clone());
 
let state = AppState { db, jobs: tx };

Handler tinggal state.jobs.send(Job::SendEmail { ... }).await.

#Multiple Workers: Konkurensi

Satu worker memproses job sekuensial — job lambat (generate thumbnail 2 detik) memblokir email berikutnya. Solusi: beberapa worker atau spawn per job.

#Worker berganda

N worker sejenis

fn spawn_workers(rx: mpsc::Receiver<Job>, db: sqlx::PgPool, n: usize) {
    for _ in 0..n {
        tokio::spawn(worker_task(rx.clone(), db.clone()));
    }
}

mpsc::Receiver tidak Clone, jadi pola ini butuh channel multi-consumer — gunakan async_channel atau satu receiver yang di-share via Arc<Mutex<...>>. Simpelnya: satu channel tokio::sync::mpsc dengan satu receiver, dan di dalam worker proses job konkuren:

Konkurensi di dalam worker

async fn worker_task(mut rx: mpsc::Receiver<Job>, db: sqlx::PgPool) {
    while let Some(job) = rx.recv().await {
        let db = db.clone();
        tokio::spawn(async move {
            process_job(job, db).await;
        });
    }
}

Tiap job di-spawn — beberapa job berjalan paralel. Batasi jumlah paralel dengan Semaphore jika job berat:

Batas konkurensi dengan Semaphore

use std::sync::Arc;
use tokio::sync::Semaphore;
 
let semaphore = Arc::new(Semaphore::new(8)); // max 8 job paralel
 
async fn worker_task(rx: mpsc::Receiver<Job>, db: sqlx::PgPool, semaphore: Arc<Semaphore>) {
    while let Some(job) = rx.recv().await {
        let permit = semaphore.clone().acquire_owned().await.expect("semaphore ditutup");
        let db = db.clone();
        tokio::spawn(async move {
            process_job(job, db).await;
            drop(permit);
        });
    }
}

Semaphore membatasi penggunaan sumber daya — mencegah 1000 job berat berjalan sekaligus dan menghabiskan CPU/DB.

#Job Queue dengan Redis (deadpool)

Worker dalam proses hilang saat aplikasi restart — job yang belum diproses hilang. Untuk persistensi dan scale-out (banyak instance berbagi antrian), gunakan Redis sebagai queue. deadpool-redis menyediakan pool koneksi:

Tambah deadpool-redis

cargo add deadpool-redis

Job queue Redis

use deadpool_redis::{redis::AsyncCommands, Config as RedisConfig, Runtime};
 
#[derive(Clone)]
pub struct AppState {
    pub db: sqlx::PgPool,
    pub redis: deadpool_redis::Pool,
}
 
pub async fn enqueue_job(state: &AppState, job: &str) -> Result<(), AppError> {
    let mut conn = state.redis.get().await.map_err(AppError::from)?;
    conn.rpush::<_, _, i64>("jobs", job).await.map_err(AppError::from)?;
    Ok(())
}

Worker yang membaca queue Redis — pola list sebagai antrian (LPUSH/RPUSH + BLPOP) atau streams untuk fitur consumer groups:

Worker Redis BLPOP

use deadpool_redis::redis::AsyncCommands;
 
async fn redis_worker(pool: deadpool_redis::Pool) {
    let mut conn = pool.get().await.unwrap();
    loop {
        let job: Option<(String, String)> = conn.blpop("jobs", 0).await.unwrap_or(None);
        if let Some((_queue, payload)) = job {
            process_payload(&payload).await;
        }
    }
}

Keunggulan queue Redis:

Keunggulan	Arti
Persisten	Job tersimpan di Redis — selamat dari restart instance
Multi-instance	Banyak worker/instance berbagi satu antrian
Delay & retry	Redis streams mendukung consumer group + pending list
Hemat	Redis adalah dependency yang umum sudah ada

Trade-off: infrastruktur tambahan (Redis), dan job yang hilang jika Redis belum di-persist. Untuk job yang kritis, kombinasikan dengan status di database (job table dengan state pending/done/failed).

#Shutdown yang Benar untuk Worker

Ingat episode 16: graceful shutdown. Worker yang di-spawn mentah akan dibunuh paksa saat runtime berhenti — job yang sedang diproses hilang. Solusinya: simpan handle dan tunggu saat shutdown.

Worker ikut shutdown

let (tx, rx) = mpsc::channel(1024);
let worker_handle = spawn_worker(rx, db.clone());
 
let state = AppState { db, jobs: tx };
let state_for_shutdown = state.clone();
 
axum::serve(listener, app)
    .with_graceful_shutdown(async move {
        shutdown_signal().await;
 
        // 1. Berhenti menerima job baru
        drop(state_for_shutdown.jobs);
 
        // 2. Tunggu worker selesai memproses antrian
        worker_handle.await.ok();
    })
    .await
    .unwrap();

Alurnya logis: sinyal shutdown → drop(sender) membuat rx.recv() mengembalikan None setelah antrian habis → worker menyelesaikan job terakhir → worker_handle.await selesai. Job yang sudah masuk antrian tetap diproses; yang belum tidak hilang dari Redis queue (kalau pakai Redis).

Note

Untuk Redis queue, worker cukup berhenti mem-BLPOP — job yang belum diproses tetap aman di Redis dan dikerjakan instance lain. Inilah mengapa queue eksternal lebih tahan banting daripada mpsc dalam proses untuk workload kritis.

#Kapan Memakai Pola yang Mana

Pola	Use case	Kelemahan
`tokio::spawn` fire-and-forget	Task sekali, tidak kritis (log, telemetry)	Tidak persisten, tidak ter-track
mpsc + worker dalam proses	Antrian dalam satu instance	Hilang saat restart; tidak scale-out
Redis queue + worker	Job kritis, multi-instance, butuh retry	Butuh infrastruktur Redis

Aturan praktis: mulai dari mpsc dalam proses untuk kesederhanaan; pindah ke Redis saat (a) job harus selamat dari restart, atau (b) aplikasi di-scale ke banyak instance. Jangan pasang Redis sejak hari pertama — kompleksitas tanpa kebutuhan adalah beban.

#Penutup

Pada episode 22 ini aplikasi kalian bisa bekerja di latar belakang:

tokio::spawn untuk fire-and-forget dengan response 202 Accepted.
mpsc channel + worker task untuk antrian dalam proses.
Semaphore untuk membatasi konkurensi job berat.
Redis queue dengan deadpool-redis untuk persistensi & scale-out.
Graceful shutdown yang menunggu worker menyelesaikan antrian.

Di episode 23 selanjutnya kita tangani data besar: streaming & multipart — streaming response/request, pengaturan body limit, dan upload file bertahap dengan multipart. Sampai jumpa di episode 23!