Desain pipeline worker untuk indexing file skala besar yang stabil bukan sekadar soal menambah jumlah worker. Masalah utama biasanya muncul dari detail operasional: file yang sama diproses dua kali, status job macet saat worker crash, antrean tertentu menjadi terlalu panas, cache metadata basi, atau retry tanpa kontrol yang justru memperparah beban sistem.

Jika Anda sedang membangun sistem indexing lokal atau terdistribusi untuk ratusan GB file atau video, pendekatan yang aman adalah memecah proses menjadi tahapan yang kecil, idempotent, bisa diulang, dan memiliki sumber kebenaran status yang jelas. Artikel ini membahas arsitektur praktis yang bisa diterapkan pada satu mesin maupun beberapa node backend.

Tujuan desain: stabil lebih dulu, cepat kemudian

Pada indexing skala besar, target pertama bukan throughput maksimum, melainkan correctness under failure. Sistem harus tetap konsisten saat:

  • worker mati di tengah proses,
  • file berubah ketika sedang di-scan,
  • job duplikat masuk ke queue,
  • storage atau extractor sesekali gagal,
  • satu subset file mendominasi antrean.

Prinsip dasarnya:

  • Setiap job harus idempotent: aman dijalankan ulang tanpa menghasilkan state ganda atau rusak.
  • Status harus berbasis storage persisten, bukan hanya memori worker.
  • Lock harus granular, umumnya per file atau per resource turunan.
  • Retry harus dibedakan dari dead-letter: tidak semua error layak dicoba ulang.
  • Backpressure harus eksplisit: sistem harus bisa memperlambat input saat downstream penuh.

Arsitektur pipeline yang direkomendasikan

Untuk indexing file/video, hindari satu job besar yang melakukan semuanya sekaligus. Pecah pipeline menjadi beberapa tahap:

  1. Discovery: menemukan file baru/berubah/hilang.
  2. Fingerprint: membaca identitas file minimum, misalnya path canonical, ukuran, mtime, inode atau hash ringan bila perlu.
  3. Metadata extraction: mengambil metadata dasar, seperti MIME type, durasi video, resolusi, codec, checksum penuh bila memang dibutuhkan.
  4. Content indexing: ekstraksi teks, thumbnail, embedding, atau indeks full-text.
  5. Finalize: menyatukan hasil, menaikkan versi indeks, dan menandai status selesai.

Secara logis, arsitekturnya bisa terlihat seperti ini:

Filesystem / Object Storage
        |
        v
 [Discovery Worker]
        |
        v
   Queue: file_scan
        |
        v
 [Fingerprint Worker] --> Metadata Cache
        |
        v
 Queue: extract_metadata
        |
        v
 [Extractor Worker]
        |
        +--> Queue: generate_thumbnail
        +--> Queue: index_content
        |
        v
   State Store / DB
        |
        v
 [Finalize Worker / Reconciler]

Pada satu mesin, semua komponen ini bisa berjalan sebagai proses terpisah dengan DB dan Redis lokal. Pada beberapa node, queue dan state store dipusatkan, sedangkan worker diskalakan horizontal.

Komponen inti yang perlu dipisahkan

  • Queue broker: untuk distribusi job dan retry.
  • State store: sumber kebenaran status file dan eksekusi.
  • Metadata cache: hasil baca cepat agar extractor tidak mengulang kerja yang sama.
  • Lock manager: mencegah dua worker memproses file yang sama bersamaan.
  • Reconciler: proses periodik untuk mendeteksi job macet dan memperbaiki status.

Kesalahan umum adalah menggabungkan state eksekusi dan cache ke satu tempat tanpa batasan yang jelas. Cache boleh basi; state eksekusi tidak boleh ambigu.

Model data status: jangan hanya mengandalkan queue

Queue memberi tahu bahwa ada pekerjaan yang harus dikerjakan, tetapi queue bukan tempat terbaik untuk melacak kebenaran status file. Simpan state per file pada database atau penyimpanan persisten lain.

Status file yang praktis

Sebuah record file dapat memiliki atribut seperti:

  • file_id: identitas stabil internal.
  • path: path canonical saat ini.
  • version: versi logis file berdasarkan fingerprint.
  • status: discovered, queued, processing, indexed, failed, deleted.
  • stage: discovery, fingerprint, metadata, content, finalize.
  • lock_owner dan lock_until: siapa yang memegang lease.
  • attempt_count: jumlah percobaan untuk stage aktif.
  • last_error_code dan last_error_message.
  • updated_at dan heartbeat_at.

Gunakan versi file agar hasil lama tidak menimpa hasil baru. Jika file berubah setelah job masuk antrean, worker harus bisa mendeteksi bahwa payload mengacu ke versi lama lalu membatalkan atau menjadwalkan ulang pekerjaan.

Contoh payload job

Payload queue sebaiknya cukup kaya untuk validasi, tetapi tidak terlalu gemuk:

{
  "job_id": "01HV...",
  "job_type": "extract_metadata",
  "file_id": "f_12345",
  "path": "/data/videos/demo.mp4",
  "file_version": "mtime:1712345678:size:98234123",
  "attempt": 1,
  "trace_id": "trc_9b2...",
  "enqueued_at": "2026-06-25T10:15:00Z"
}

file_id adalah referensi utama. path dan file_version membantu validasi cepat. trace_id penting untuk observabilitas lintas worker.

Locking per file dan lease yang bisa pulih setelah crash

Tanpa locking per file, duplicate processing hampir pasti terjadi. Namun lock juga tidak boleh permanen, karena worker bisa crash sewaktu-waktu. Solusi yang umum adalah lease-based lock: lock memiliki masa berlaku dan harus diperbarui secara berkala.

Pola lock yang aman

  1. Worker mengambil job dari queue.
  2. Worker mencoba memperoleh lease untuk file_id + stage.
  3. Jika gagal karena lease masih valid, job bisa di-ack lalu diabaikan, atau di-requeue dengan delay kecil tergantung desain.
  4. Jika berhasil, worker mengubah status menjadi processing.
  5. Selama bekerja, worker mengirim heartbeat untuk memperbarui lock_until.
  6. Saat selesai, worker melepas lock dan memajukan stage/status.

Lease ini bisa disimpan di database atau sistem seperti Redis, selama ada cara untuk memastikan pelepasan tidak menimpa lock milik worker lain. Biasanya dilakukan dengan token pemilik lock.

Mengapa lock per file lebih baik daripada lock global

  • Throughput tetap tinggi karena file berbeda bisa diproses paralel.
  • Kontensi hanya terjadi pada item yang sama.
  • Lebih mudah mendiagnosis duplicate processing.

Untuk video besar yang punya subtask turunan, Anda bisa memakai lock per file untuk tahap utama dan lock per artefak untuk turunan seperti thumbnail atau transcript.

Idempotensi job: syarat wajib, bukan tambahan

Pada sistem queue, at-least-once delivery sering kali lebih realistis daripada exactly-once. Artinya job bisa diterima lebih dari sekali. Karena itu, worker harus idempotent.

Cara membuat job idempotent

  • Gunakan kunci unik hasil, misalnya kombinasi file_id + file_version + stage.
  • Upsert, bukan insert buta, saat menulis hasil metadata atau indeks.
  • Cek state saat mulai dan saat commit untuk memastikan versi file belum berubah.
  • Pastikan efek samping eksternal bisa didedup, misalnya nama file thumbnail deterministik berdasarkan file version.

Contoh alur aman:

1. Baca record file dari DB
2. Validasi file_version di payload == file_version saat ini
3. Ambil lease
4. Cek apakah output stage sudah ada untuk file_version ini
5. Jika sudah ada, tandai sukses dan keluar
6. Proses extractor
7. Simpan hasil dengan upsert
8. Update status stage berikutnya dalam transaksi yang konsisten

Kesalahan umum adalah hanya mengandalkan deduplikasi saat enqueue. Itu membantu, tetapi tidak cukup. Job bisa tetap terduplikasi akibat retry, timeout ACK, atau crash setelah write namun sebelum ACK.

Queue design: pisahkan antrean sesuai karakter beban

Satu antrean besar untuk semua jenis file dan semua tahap biasanya cepat menjadi bottleneck. Pisahkan queue berdasarkan sifat pekerjaannya.

Pemisahan antrean yang berguna

  • discovery: I/O ringan, banyak item.
  • metadata: I/O sedang, CPU ringan.
  • video_extract: CPU tinggi, durasi panjang.
  • content_index: bisa berat pada memory atau downstream search engine.
  • finalize: kecil, tetapi penting untuk konsistensi status.

Dengan pemisahan ini, file video besar tidak memblokir pemrosesan file kecil. Anda juga bisa mengalokasikan concurrency berbeda per queue.

Hot queue dan cara mengatasinya

Hot queue terjadi ketika satu jenis job atau subset file mendominasi antrean sehingga job lain menunggu terlalu lama. Contohnya ribuan video besar masuk sekaligus dan menenggelamkan indexing dokumen kecil.

Mitigasi yang umum:

  • Prioritas queue: metadata dasar lebih tinggi daripada ekstraksi berat.
  • Weighted concurrency: batasi jumlah worker untuk queue berat.
  • Sharding: pisahkan antrean berdasarkan tenant, direktori, atau tipe file.
  • Fair scheduling: jangan biarkan satu sumber input menghabiskan semua slot.

Jika Anda hanya punya satu mesin, hot queue sering terlihat sebagai CPU 100% karena satu tahap berat, sementara tahap lain tampak macet. Ini bukan bug queue, tetapi masalah alokasi concurrency.

Backpressure: cegah sistem menerima lebih banyak dari yang bisa diproses

Backpressure adalah mekanisme untuk memperlambat produksi job saat downstream tidak mampu mengejar. Tanpa backpressure, discovery dapat terus memasukkan jutaan job sementara extractor tertinggal berjam-jam atau berhari-hari.

Sinyal backpressure yang layak dipakai

  • panjang queue per stage,
  • usia job tertua,
  • rasio retry,
  • latensi rata-rata per stage,
  • pemakaian CPU, memory, disk I/O,
  • kapasitas downstream seperti database atau search index.

Strategi backpressure yang praktis

  • Throttle discovery ketika queue tahap berikutnya melewati ambang tertentu.
  • Batching adaptif: kurangi ukuran batch saat error atau latency naik.
  • Rate limit per tipe file: video besar diproses lebih hati-hati dibanding file teks kecil.
  • Admission control: tolak atau tunda subtask turunan non-esensial seperti thumbnail ketika sistem tertekan.

Poin pentingnya: backpressure sebaiknya terukur, bukan sekadar sleep acak di worker.

Cache metadata: berguna, tetapi jangan dijadikan sumber kebenaran

Cache metadata membantu menghindari pembacaan file berulang, terutama untuk file besar atau storage lambat. Namun cache mudah menjadi stale jika file berubah, dipindahkan, atau ditimpa.

Apa yang cocok disimpan di cache

  • hasil stat ringan: ukuran, mtime, inode bila tersedia,
  • MIME hasil deteksi,
  • durasi dan resolusi video,
  • checksum yang mahal dihitung,
  • lokasi artefak turunan seperti thumbnail atau transcript.

Cara menghindari cache stale

  • Key cache harus memasukkan versi file, bukan hanya path.
  • TTL jangan jadi satu-satunya invalidasi.
  • Selalu validasi fingerprint minimum sebelum memakai hasil cache.
  • Bedakan cache metadata dan state status.

Kesalahan yang sering terjadi adalah menyimpan metadata dengan key path saja. Jika file diganti dengan isi berbeda tetapi path sama, hasil ekstraksi lama bisa dipakai kembali secara salah.

Retry vs dead-letter queue: bedakan error sementara dan error permanen

Retry yang tidak dibatasi dapat menciptakan loop tak berujung, memperpanjang antrean, dan menutupi masalah yang sebenarnya permanen.

Klasifikasi error yang berguna

  • Transient: timeout jaringan, storage sementara tidak tersedia, proses extractor kehabisan resource sesaat.
  • Permanent: file korup, format tidak didukung, path hilang permanen, payload invalid.
  • Poison job: job yang selalu memicu crash atau bug pada worker.

Strategi retry yang masuk akal

  • Exponential backoff dengan jitter untuk transient error.
  • Batas percobaan per stage, bukan global saja.
  • DLQ untuk permanent error atau retry yang habis.
  • Reason code yang jelas agar item di DLQ bisa ditindaklanjuti.

Contoh reason code:

FILE_NOT_FOUND
UNSUPPORTED_FORMAT
EXTRACTOR_TIMEOUT
LOCK_ACQUISITION_FAILED
VERSION_MISMATCH
OUTPUT_COMMIT_FAILED

VERSION_MISMATCH biasanya bukan kegagalan yang perlu DLQ; cukup drop hasil lama dan enqueue ulang versi terbaru bila perlu.

Konsistensi status saat worker crash

Masalah yang paling sering membingungkan operator adalah progress terlihat macet padahal worker sebenarnya sudah mati atau hasil setengah tersimpan. Solusinya adalah mendesain transisi status yang eksplisit dan mudah dipulihkan.

Pola transisi status yang aman

  1. queued: job telah dijadwalkan.
  2. processing: lease aktif dan worker telah mulai.
  3. committing: hasil sudah siap, sedang ditulis ke DB/index.
  4. done: stage selesai.
  5. failed_retryable atau failed_terminal.

Stage committing berguna untuk membedakan antara komputasi panjang dan penyimpanan hasil. Jika worker crash saat commit, reconciler bisa memeriksa apakah output sebenarnya sudah tersimpan lalu menyelesaikan status tanpa memproses ulang penuh.

Peran reconciler

Reconciler adalah proses periodik yang:

  • mencari record dengan processing tetapi heartbeat kedaluwarsa,
  • melepaskan lease yang orphan,
  • memastikan job tersebut dijadwalkan ulang bila aman,
  • memeriksa item yang stuck di committing,
  • merekonsiliasi state file dengan artefak output yang sudah ada.

Tanpa reconciler, progress macet biasanya hanya bisa diperbaiki secara manual.

Alur job end-to-end yang tahan duplikasi dan crash

Berikut contoh alur untuk stage metadata extraction:

receive(job)
  file = db.getFile(job.file_id)
  if !file: ack_and_drop("missing_file_record")

  if file.version != job.file_version:
    ack_and_drop("stale_job")

  lease = lock.acquire(key=file.id+":metadata", ttl=60s)
  if !lease:
    requeue_with_delay(job)
    return

  try:
    db.markProcessing(file.id, stage="metadata", owner=worker_id, heartbeat=now)

    existing = db.getStageOutput(file.id, file.version, "metadata")
    if existing:
      db.markDone(file.id, stage="metadata")
      ack(job)
      return

    result = extractor.readMetadata(file.path)

    db.transaction:
      db.upsertStageOutput(file.id, file.version, "metadata", result)
      db.advanceStage(file.id, from="metadata", to="content_index")
      queue.enqueue("content_index", payload_for_next_stage)

    ack(job)
  catch transient_error:
    db.markRetryableFailure(...)
    retry(job, backoff)
  catch permanent_error:
    db.markTerminalFailure(...)
    dlq(job)
  finally:
    lock.release(lease)

Poin penting pada alur ini:

  • validasi versi file dilakukan sebelum kerja berat,
  • hasil dicek dulu agar aman terhadap duplikasi,
  • write hasil dan enqueue stage berikutnya sebaiknya dibuat sekonsisten mungkin,
  • release lock dilakukan di finally, tetapi recovery tetap mengandalkan TTL lease.

Pola commit: outbox lebih aman daripada enqueue langsung setelah update

Salah satu race condition klasik adalah: status DB sudah di-update, tetapi enqueue ke stage berikutnya gagal; atau sebaliknya, job tahap berikutnya sudah masuk queue, tetapi commit DB gagal. Ini menyebabkan status tidak sinkron.

Untuk mengurangi masalah ini, pertimbangkan outbox pattern:

  1. Simpan perubahan status dan event untuk stage berikutnya dalam satu transaksi DB.
  2. Proses terpisah membaca tabel outbox dan mengirim event ke queue.
  3. Event di-mark sent setelah sukses terkirim.

Trade-off-nya adalah kompleksitas bertambah, tetapi konsistensi antar DB dan queue menjadi jauh lebih baik.

Observabilitas: tanpa ini, debugging akan mahal

Pipeline indexing yang kelihatannya sederhana bisa sulit di-debug saat skala naik. Pastikan sejak awal ada logging terstruktur, metrics, dan tracing minimum.

Metrics yang wajib ada

  • queue depth per queue,
  • oldest job age,
  • throughput per stage,
  • success/failure rate per stage dan reason code,
  • retry count dan DLQ count,
  • lock contention rate,
  • processing time per tipe file,
  • stuck jobs berdasarkan heartbeat timeout.

Logging yang sebaiknya konsisten

Setiap log penting minimal memuat:

  • trace_id,
  • job_id,
  • file_id,
  • file_version,
  • stage,
  • attempt,
  • worker_id,
  • error_code bila gagal.

Contoh log JSON:

{
  "level": "error",
  "message": "extractor timeout",
  "trace_id": "trc_9b2",
  "job_id": "01HV...",
  "file_id": "f_12345",
  "file_version": "mtime:1712345678:size:98234123",
  "stage": "metadata",
  "attempt": 2,
  "worker_id": "node-3/w-7",
  "error_code": "EXTRACTOR_TIMEOUT"
}

Alert yang benar-benar berguna

  • usia job tertua di queue tertentu melewati ambang,
  • rasio retry melonjak tiba-tiba,
  • DLQ bertambah cepat,
  • banyak lock orphan,
  • jumlah item berstatus processing tanpa heartbeat meningkat.

Masalah operasional nyata dan cara menanganinya

1. Duplicate processing

Gejala: file yang sama diekstrak dua kali, thumbnail ganda, atau indeks dobel.

Penyebab umum: ACK timeout, retry setelah commit, lock tidak efektif, payload stale.

Solusi:

  • lease per file,
  • output stage dengan unique key,
  • idempotent upsert,
  • cek ulang status sebelum commit.

2. Hot queue

Gejala: antrean tertentu terus menumpuk dan job kecil tidak pernah selesai cepat.

Solusi:

  • pisah queue berat dan ringan,
  • batasi concurrency tahap berat,
  • fair scheduling atau partition per sumber.

3. Cache stale

Gejala: metadata lama muncul setelah file diperbarui.

Solusi:

  • gunakan key berbasis versi file,
  • validasi fingerprint minimum sebelum memakai cache,
  • jangan jadikan cache sebagai status utama.

4. Progress macet

Gejala: dashboard menunjukkan ribuan item processing, tetapi tidak ada hasil baru.

Penyebab: worker crash, heartbeat tidak diperbarui, commit menggantung.

Solusi:

  • heartbeat dan lease timeout,
  • reconciler periodik,
  • alert untuk processing tanpa heartbeat.

5. Partial failure

Gejala: metadata berhasil, tetapi thumbnail gagal; atau indeks teks berhasil, tetapi finalize tidak jalan.

Solusi:

  • simpan output per stage secara terpisah,
  • jangan rollback seluruh pipeline jika satu artefak opsional gagal,
  • tetapkan dependency yang jelas antara stage wajib dan opsional.

Satu mesin vs beberapa node: apa yang berubah?

Satu mesin

Cocok jika dataset besar tetapi throughput harian masih moderat dan storage lokal cepat. Fokus utamanya:

  • batasi concurrency agar tidak menghancurkan disk I/O,
  • pisahkan worker CPU-bound dan I/O-bound,
  • gunakan queue lokal atau broker ringan,
  • pastikan state tetap persisten dan recovery otomatis ada.

Beberapa node

Diperlukan jika ekstraksi video berat, banyak direktori, atau SLA indexing lebih ketat. Tambahan perhatian:

  • lock harus benar-benar terdistribusi,
  • path lokal perlu diabstraksikan bila storage tidak identik antar node,
  • observabilitas lintas node wajib,
  • hindari asumsi jam sistem selalu sinkron sempurna untuk lease berbasis waktu.

Pada multi-node, lebih aman bila worker bekerja dengan storage bersama atau URI yang bisa diakses semua node, bukan path lokal yang hanya valid di satu host.

Checklist implementasi

  • Definisikan file_id dan file_version yang stabil.
  • Buat state store terpisah dari cache metadata.
  • Terapkan lease lock per file_id + stage dengan heartbeat.
  • Pastikan setiap stage idempotent dengan unique output key.
  • Pisahkan queue berdasarkan karakter beban.
  • Terapkan retry policy per reason code, bukan satu aturan untuk semua.
  • Sediakan dead-letter queue yang bisa diinspeksi operator.
  • Gunakan reconciler untuk stuck job dan orphan lock.
  • Tambah backpressure berdasarkan queue depth dan oldest job age.
  • Log terstruktur dengan trace_id, job_id, dan file_id.
  • Buat dashboard metrics per stage dan per tipe file.
  • Uji skenario crash: mati sebelum ACK, mati setelah write, mati saat commit, lease expired saat proses lama.
  • Uji file berubah di tengah pipeline.
  • Uji duplicate enqueue dan retry berulang.

Penutup

Desain pipeline worker untuk indexing file skala besar yang stabil bergantung pada disiplin di area yang sering dianggap detail: lock per file, idempotensi, status persisten, backpressure, dan recovery setelah crash. Jika fondasi ini benar, scaling ke ratusan GB pada satu mesin atau beberapa node menjadi jauh lebih dapat diprediksi.

Mulailah dari pipeline yang sederhana tetapi tahan gagal. Setelah itu baru optimalkan throughput. Pada praktiknya, sistem indexing yang stabil hampir selalu lebih bernilai daripada sistem yang kadang cepat tetapi sulit dipercaya saat terjadi gangguan.