Belajar NATS - Stream Manager & Work Queues

#Pendahuluan

Episode 10 memperkenalkan penyimpanan data; episode 11 ini merangkainya menjadi pola yang paling sering dipakai di produksi: work queue. Antrean pekerjaan yang andal, di mana setiap job dikerjakan tepat oleh satu worker, dan tidak ada job yang hilang meski worker crash.

Kita akan membangun work queue dengan stream WorkQueue dan pull consumer, lalu menutup celah terakhir: exactly-once dan idempotency. Inilah pola tulang punggung sistem job processing.

#Pola Work Queue

#Konsep Dasar

Work queue berbeda dari broadcast: setiap pesan (job) hanya dikerjakan satu worker. Di Core NATS, queue groups sudah melakukan load balancing, tapi tanpa jaminan penyimpanan. JetStream menambahkannya dengan stream WorkQueue.

publisher --> stream JOB (retention WorkQueue)
                  └── pull consumer
                        ├── worker-1 mengerjakan job
                        ├── worker-2 mengerjakan job
                        └── worker-3 mengerjakan job

Alur publisher -> stream JOB (retention WorkQueue) menunjukkan job disimpan dulu di stream, lalu didistribusikan ke worker. Karena stream menyimpan pesan, job tidak hilang meski semua worker sedang sibuk.

#Membuat Stream WorkQueue

Buat stream dengan retention WorkQueue:

nats stream add JOB --subjects "job.>" --retention workqueue --storage file

Perintah nats stream add JOB --subjects "job.>" --retention workqueue membuat stream JOB yang menyimpan setiap pesan hanya untuk satu consumer. Begitu job diack, ia dihapus dari stream — tidak ada replay, tidak ada duplikasi antar worker.

#Pull Consumer untuk Job Queue

#Mengapa Pull

Untuk banyak worker, pull consumer adalah pilihan tepat: setiap worker menarik job sesuai kapasitasnya, menerapkan backpressure alami, dan tidak menerima lebih dari yang bisa dikerjakan.

nats consumer add JOB WORKER --pull --ack explicit --max-deliver 3 --ack-wait 60s

--pull --ack explicit membuat consumer yang dipakai worker untuk menarik job. --max-deliver 3 membatasi job dikirim ulang maksimal 3 kali, --ack-wait 60s memberi worker satu menit untuk mengack.

#Mengambil Job dari CLI

Dari CLI, worker bisa menarik job satu per satu:

nats consumer next JOB WORKER

nats consumer next JOB WORKER mengambil satu job berikutnya dari stream. Karena consumer bersifat pull, job tidak datang tanpa diminta — worker mengontrol ritme kerjanya sendiri.

#Client Go untuk Worker

Di kode, pola worker memakai fetch:

nc, _ := nats.Connect("nats://localhost:4222")
js, _ := nc.JetStream()
 
for {
    msgs, _ := js.Fetch("JOB", "WORKER", 1, nats.MaxWait(30*time.Second))
    if len(msgs) == 0 {
        continue
    }
    for _, msg := range msgs {
        processJob(msg.Data)
        msg.Ack()
    }
}

js.Fetch("JOB", "WORKER", 1, nats.MaxWait(30*time.Second)) menarik satu job dengan waktu tunggu 30 detik, dan msg.Ack() menandai selesai. Loop ini adalah inti dari hampir semua worker NATS di produksi.

#Exactly-Once dan Idempotency

#Jaminan Publish vs Jaminan Proses

Ada dua tingkat jaminan yang harus dibedakan:

• Exactly-once publish: pesan masuk stream tepat satu kali.
• Exactly-once processing: efek proses terjadi tepat satu kali.

exactly-once publish  -> dijamin server (dedupe window)
exactly-once process  -> dijamin aplikasi (idempotency)

Pemisahan exactly-once publish -> dijamin server penting: dedupe window mencegah pesan ganda masuk stream, tapi worker yang crash setelah memproses sebelum ack akan menerima pesan lagi. Menangani bagian kedua adalah tugas aplikasi.

#Dedupe Window Publisher

#Mencegah Publish Ganda

Seperti yang dibahas di episode 8, publisher bisa menyertakan ID untuk dedupe:

Publish job dengan dedupe

import asyncio
import nats
 
async def main():
    nc = await nats.connect("nats://localhost:4222")
    js = nc.jetstream()
    await js.publish("job.payroll", b"payroll-2026-08", headers={"Nats-Msg-Id": "job-2026-08-10-001"})
 
asyncio.run(main())

headers={"Nats-Msg-Id": "job-2026-08-10-001"} memberi identitas unik pada publish. Jika client kehilangan ack dan mengirim ulang dengan ID sama, JetStream menolak duplikatnya. Ini memastikan setiap job masuk stream tepat satu kali.

#Jendela Dedupe yang Cukup

nats stream edit JOB --dedupe-window 10m

--dedupe-window 10m memperpanjang jendela dedupe menjadi 10 menit. Nilai harus lebih lama dari waktu maksimal client menunggu ack, sehingga retry setelah timeout masih dalam jendela.

#Idempotency Key dan Sinkronisasi

#Menangani Proses Ulang

Meski publish sudah dedupe, worker bisa menerima job yang sama dua kali — misalnya setelah crash sebelum ack. Solusinya adalah idempotency key yang dipakai untuk menolak efek ganda:

Idempotency dengan KV store

from datetime import timedelta
 
entry = await kv.get("processed", key="job-2026-08-10-001")
if entry is None:
    entry = await kv.create("processed", key="job-2026-08-10-001", value=b"done", ttl=timedelta(days=7))
    if entry is not None:
        await process_job(payload)

Logika kv.create(...) berhasil hanya jika key belum ada. Jika dua worker menerima job yang sama, hanya satu yang berhasil membuat key — yang lain tahu job sudah diproses. Ini exactly-once processing tanpa perlu database eksternal.

#Sinkronisasi dengan Database

Untuk transaksi yang lebih kompleks, sinkronkan dengan database aplikasi:

worker menerima job --> mulai transaksi DB
                        ├── update status job = PROCESSING
                        ├── lakukan proses
                        ├── update status job = DONE
                        └── commit + ack JetStream

Alur update status job = PROCESSING di database menjadi guard: jika job tiba lagi, worker melihat status DONE dan menolak memproses ulang. Kombinasi idempotency key di KV dan transaksi database memberi lapisan perlindungan ganda.

#Penutup

Episode 11 merakit pola work queue produksi: stream WorkQueue yang menyimpan setiap job untuk satu worker, pull consumer yang memberi worker kontrol ritme, dedupe window untuk exactly-once publish, dan idempotency key — via KV store atau database — untuk exactly-once processing.

Inti yang harus dibawa pulang:

• Stream WorkQueue menyimpan setiap pesan untuk satu consumer saja.
• Pull consumer memberi worker backpressure alami dan kontrol ritme.
• nats consumer next menarik job; client memakai Fetch atau fetch di worker.
• Dedupe window dengan Nats-Msg-Id memastikan publish tepat satu kali.
• Exactly-once process memerlukan idempotency key di sisi aplikasi.
• KV store atau database dipakai sebagai guard proses ulang.

Di episode 12 selanjutnya kita akan membahas accounts & multi-tenancy — isolasi logika antar tim dan service lewat account, users beserta permission subscribe, publish, dan response, lalu nsc untuk membuat accounts dan users berbasis JWT, serta ekspor dan impor subject antar account. Satu server NATS mulai melayani banyak tenant.