Arsitektur antrean berbasis orchestrator sentral seperti Temporal, Airflow, atau Kubernetes CronJob sering kali memperkenalkan single point of failure dan overhead operasional yang tinggi pada skala tertentu. Desain worker otonom memindahkan tanggung jawab orkestrasi ke masing-masing node komputasi. Setiap worker bertindak independen, memvalidasi status tugas, mengklaim beban kerja melalui konsensus data store, serta memulihkan kegagalan eksekusi secara mandiri.

1. Pola Akuisisi Tugas: SKIP LOCKED vs Lease Lock

Untuk menghindari tabrakan antar-worker independen, sistem membutuhkan mekanisme konsensus status tugas tanpa koordinator. Dua pendekatan paling andal adalah RDBMS row-level locking dan distributed lease lock.

PostgreSQL: FOR UPDATE SKIP LOCKED

Pendekatan native RDBMS memanfaatkan transaksi ACID dengan klausul FOR UPDATE SKIP LOCKED. Pola ini memungkinkan worker mengunci dan mengambil baris pertama yang belum terkunci oleh worker lain tanpa menimbulkan antrean blocking.

-- Skema antrean tugas
CREATE TABLE job_queue (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    payload JSONB NOT NULL,
    state VARCHAR(20) NOT NULL DEFAULT 'pending',
    attempts INT NOT NULL DEFAULT 0,
    max_attempts INT NOT NULL DEFAULT 5,
    locked_until TIMESTAMPTZ,
    idempotency_key VARCHAR(64) UNIQUE NOT NULL,
    last_error TEXT,
    created_at TIMESTAMPTZ DEFAULT clock_timestamp(),
    updated_at TIMESTAMPTZ DEFAULT clock_timestamp()
);

-- Pola konsumsi tugas otonom (Atomic Fetch and Lock)
UPDATE job_queue
SET state = 'processing',
    locked_until = clock_timestamp() + INTERVAL '2 minutes',
    attempts = attempts + 1,
    updated_at = clock_timestamp()
WHERE id = (
    SELECT id
    FROM job_queue
    WHERE (state = 'pending')
       OR (state = 'processing' AND locked_until < clock_timestamp())
    ORDER BY created_at ASC
    FOR UPDATE SKIP LOCKED
    LIMIT 1
)
RETURNING id, payload, attempts, max_attempts;

Klausa OR (state = 'processing' AND locked_until < clock_timestamp()) mengimplementasikan self-healing timeout secara otomatis. Jika sebuah worker mati mendadak saat memproses tugas, batas waktu locked_until akan terlewati, dan worker lain dapat langsung merebut tugas tersebut tanpa perlu intervensi supervisor.

Redis Redlock vs Single-Instance Lease

Jika menggunakan Redis, gunakan lease lock dengan TTL menggunakan perintah SET key worker_id NX PX 30000. Hindari algoritma multi-node Redlock jika konsistensi mutlak diperlukan saat partisi jaringan terjadi; Postgres ACID tetap lebih superior untuk reliabilitas state transaksional.

2. Idempotency Key dan State Transition

Jaringan terdistribusi menganut prinsip at-least-once delivery. Oleh karena itu, worker harus menjamin bahwa eksekusi ulang pada tugas yang sama tidak menduplikasi mutasi data di sisi domain.

// Implementasi Idempotent Execution di Go
package main

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
)

type Job struct {
	ID             string
	IdempotencyKey string
	Payload        []byte
}

func ProcessJob(ctx context.Context, db *sql.DB, job Job) error {
	tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return err
	}
	defer tx.Rollback()

	// 1. Cek riwayat idempotency
	var exists bool
	checkQuery := `SELECT EXISTS(SELECT 1 FROM processed_keys WHERE key = $1)`
	if err := tx.QueryRowContext(ctx, checkQuery, job.IdempotencyKey).Scan(&exists); err != nil {
		return err
	}
	if exists {
		// Tugas sudah pernah sukses, tandai selesai pada job queue
		_, _ = tx.ExecContext(ctx, `UPDATE job_queue SET state = 'completed' WHERE id = $1`, job.ID)
		return tx.Commit()
	}

	// 2. Eksekusi logika bisnis inti
	if err := executeBusinessLogic(ctx, tx, job.Payload); err != nil {
		return err
	}

	// 3. Catat idempotency key dan mutasi status
	_, err = tx.ExecContext(ctx, `INSERT INTO processed_keys (key, created_at) VALUES ($1, NOW())`, job.IdempotencyKey)
	if err != nil {
		return err
	}

	_, err = tx.ExecContext(ctx, `UPDATE job_queue SET state = 'completed', updated_at = NOW() WHERE id = $1`, job.ID)
	if err != nil {
		return err
	}

	return tx.Commit()
}

func executeBusinessLogic(ctx context.Context, tx *sql.Tx, payload []byte) error {
	// Mutasi domain data dalam konteks transaksi yang sama
	return nil
}

3. Mitigasi Poison Pill dan Penanganan DLQ Mandiri

Poison pill adalah pesan payload corrupt yang menyebabkan worker crash seketika (misal: unhandled panic, out-of-memory). Tanpa penanganan, tugas ini akan diambil ulang berulang kali dan mematikan seluruh instance worker (cascading failure).

Worker otonom harus memitigasi ini dengan metode defensive execution dan isolasi internal:

  • Crash Isolation (Panic Recovery): Tangkap unhandled exception di level goroutine/thread loop agar proses utama worker tidak mati.
  • Batas Percobaan Dinamis: Bandingkan nilai attempts dengan max_attempts saat query akuisisi.
  • Dead Letter Isolation: Jika attempts >= max_attempts, alihkan status langsung ke dead_letter tanpa mengeksekusi payload.
// Recover dari panic dan isolasi poison pill
func RunSafe(ctx context.Context, db *sql.DB, job Job, attempts, maxAttempts int) {
	if attempts > maxAttempts {
		routeToDLQ(db, job.ID, "Max attempts exceeded / Poison pill detected")
		return
	}

	defer func() {
		if r := recover(); r != nil {
			errMsg := fmt.Sprintf("panic recovered: %v", r)
			recordFailure(db, job.ID, errMsg)
		}
	}()

	if err := ProcessJob(ctx, db, job); err != nil {
		recordFailure(db, job.ID, err.Error())
	}
}

func routeToDLQ(db *sql.DB, id, reason string) {
	query := `UPDATE job_queue SET state = 'dead_letter', last_error = $2, updated_at = NOW() WHERE id = $1`
	_, _ = db.Exec(query, id, reason)
}

func recordFailure(db *sql.DB, id, reason string) {
	query := `UPDATE job_queue SET state = 'pending', locked_until = NULL, last_error = $2, updated_at = NOW() WHERE id = $1`
	_, _ = db.Exec(query, id, reason)
}

4. Observability dan Metrik Esensial

Meniadakan orchestrator sentral berarti visibilitas status sistem bergantung sepenuhnya pada telemetri metrik. Pantau metrik berikut:

  • Queue Lag / Oldest Unprocessed Job: Selisih waktu NOW() - created_at untuk baris dengan state = 'pending'. Indikator utama kapasitas worker kurang.
  • Dead Letter Queue Rate: Jumlah transisi status ke dead_letter per menit. Kenaikan mendadak menandakan deployment bug atau malformed payload.
  • Lease Expiration Rate: Frekuensi tugas yang diambil worker lain karena locked_until < NOW(). Nilai tinggi menandakan eksekusi worker terlalu lambat melebihi estimasi lock TTL atau terjadi proses freeze (e.g. Stop-The-World GC).

Gunakan indeks parsial PostgreSQL seperti CREATE INDEX idx_pending_jobs ON job_queue (created_at) WHERE state = 'pending'; untuk menjaga latensi fetch tetap di bawah 1 milidetik meskipun ukuran tabel telah membesar.