Sistem antrean asinkron sering mengalami masalah false-positive timeout: API gateway atau frontend menganggap worker mengalami hang, deadlock, atau crash saat menjalankan komputasi berat. Akibatnya, sistem memicu mekanisme retry otomatis yang memicu duplikasi proses atau thundering herd. Masalah ini diselesaikan dengan menerapkan legibility of effort, yaitu teknik perancangan sistem yang membuat beban kerja, alokasi memori, dan progres internal worker dapat diamati secara deterministik secara realtime tanpa menghambat thread komputasi utama.

Arsitektur Status Worker: Redis Hash vs Redis Streams

Memancarkan telemetri status membutuhkan struktur data Redis yang tepat bergantung pada pola konsumsi metrik.

  • Redis Hash: Cocok untuk titik baca tunggal (single-point of truth) saat client melakukan polling status terkini job (misalnya: status eksekusi, persentase progres, byte terakhir yang diproses). Hash meminimalkan overhead baca karena client hanya mengambil nilai instan melalui HGETALL atau HMGET.
  • Redis Streams: Cocok untuk audit log temporal dan pengiriman update via WebSockets atau Server-Sent Events (SSE). Streams menyimpan urutan peristiwa (append-only log) dengan timestamp Redis (XADD), memungkinkan consumer melacak riwayat perubahan status dari titik kegagalan tertentu.

Pola paling efisien menggabungkan keduanya: Hash menyimpan current snapshot, sedangkan Stream menyimpan time-series checkpoints jika diperlukan reka jejak eksekusi.

Non-Blocking Telemetry and Lock Renewal via Heartbeat

Pekerjaan berat tidak boleh mengorbankan event loop atau thread pemrosesan untuk telemetri I/O jaringan Redis. Solusinya adalah mendelegasikan telemetri dan perpanjangan distributed lock ke goroutine/background worker sekunder yang berkomunikasi via channel atau atomic variable.

Metode ini mengatasi dua tantangan:

  1. Distributed Lock TTL Renewal: Menghindari kondisi di mana kunci rilis sebelum komputasi selesai tanpa harus menyetel nilai TTL awal yang terlalu tinggi (mencegah orphan lock jika node mati mendadak).
  2. Telemetry Batching: Mengumpulkan metrik runtime (seperti runtime.MemStats pada Go atau process.memoryUsage() pada Node.js) dan menyinkronkannya ke Redis secara berkala (misal tiap 1000ms), bukan pada setiap iterasi loop pemrosesan data.

Implementasi Worker Telemetri Deterministik (Go)

Contoh berikut mendemonstrasikan pattern worker di Go dengan background heartbeat runner yang memperbarui progress snapshot dan memperpanjang lock atomik di Redis:

package main

import (
	"context"
	"fmt"
	"runtime"
	"sync/atomic"
	"time"

	"github.com/redis/go-redis/v9"
)

type WorkerTask struct {
	JobID    string
	RDB      *redis.Client
	Progress atomic.Int64
	Stage    atomic.Value
	Done     chan struct{}
}

func (w *WorkerTask) StartHeartbeat(ctx context.Context, lockKey string, lockVal string, ttl time.Duration) {
	ticker := time.NewTicker(ttl / 3)
	defer ticker.Stop()

	// Lua script: perpanjang lock hanya jika val cocok (mencegah overwrite lock milik node lain)
	renewScript := redis.NewScript(`
		if redis.call("get", KEYS[1]) == ARGV[1] then
			return redis.call("pexpire", KEYS[1], ARGV[2])
		else
			return 0
		end
	`)

	for {
		select {
		case <-w.Done:
			return
		case <-ticker.C:
			// 1. Extend lock TTL
			res, err := renewScript.Run(ctx, w.RDB, []string{lockKey}, lockVal, int(ttl.Milliseconds())).Result()
			if err != nil || res.(int64) == 0 {
				// Lock hilang atau terambil alih, batalkan komputasi jika perlu
				return
			}

			// 2. Sample memory footprint
			var m runtime.MemStats
			runtime.ReadMemStats(&m)

			// 3. Update snapshot hash
			hashKey := fmt.Sprintf("job:telemetry:%s", w.JobID)
			pipe := w.RDB.Pipeline()
			pipe.HSet(ctx, hashKey, map[string]interface{}{
				"progress":      w.Progress.Load(),
				"stage":         w.Stage.Load(),
				"alloc_mb":      m.Alloc / 1024 / 1024,
				"last_heartbeat": time.Now().Unix(),
				"status":        "RUNNING",
			})
			pipe.Expire(ctx, hashKey, time.Hour) // Retensi telemetry
			_, _ = pipe.Exec(ctx)
		}
	}
}

func ExecuteJob(ctx context.Context, rdb *redis.Client, jobID string, totalRows int) error {
	lockKey := fmt.Sprintf("lock:job:%s", jobID)
	lockVal := fmt.Sprintf("node-worker-%d", time.Now().UnixNano())
	lockTTL := 6 * time.Second

	// Acquire Lock
	ok, err := rdb.SetNX(ctx, lockKey, lockVal, lockTTL).Result()
	if err != nil || !ok {
		return fmt.Errorf("unable to acquire distributed lock")
	}
	defer func() {
		// Release lock atomik
		delScript := redis.NewScript(`
			if redis.call("get", KEYS[1]) == ARGV[1] then
				return redis.call("del", KEYS[1])
			else
				return 0
			end
		`)
		delScript.Run(context.Background(), rdb, []string{lockKey}, lockVal)
	}()

	task := &WorkerTask{
		JobID: jobID,
		RDB:   rdb,
		Done:  make(chan struct{}),
	}
	task.Stage.Store("INITIALIZING")

	// Jalankan telemetri di thread sekunder
	go task.StartHeartbeat(ctx, lockKey, lockVal, lockTTL)
	defer close(task.Done)

	// Simulasi pemrosesan berat bertahap
	task.Stage.Store("COMPUTING_CHUNKS")
	for i := 1; i <= totalRows; i++ {
		// Heavy CPU/IO computation
		time.Sleep(10 * time.Millisecond)

		// Set status atomik murah tanpa I/O blocking
		task.Progress.Store(int64((i * 100) / totalRows))
	}

	task.Stage.Store("FINALIZING")
	rdb.HSet(ctx, fmt.Sprintf("job:telemetry:%s", jobID), "status", "COMPLETED").Err()
	return nil
}

Mitigasi Race Condition dan Konsistensi Status

Saat banyak proses worker bersaing atau node mati mendadak, dua potensi bahaya utama harus dimitigasi:

1. Out-of-Order Updates

Jaringan lambat dapat menyebabkan write hash telemetri tiba terlambat melebihi pembaruan berikutnya. Gunakan versioning counter pada telemetri data, atau andalkan Redis Streams jika urutan mutlak diperlukan. Pada Redis Hash, periksa timestamp last_heartbeat secara deterministik di sisi client/API handler.

2. Penanganan Worker Crash (Stale State Detection)

Jika server worker mengalami kernel panic atau OOM-kill tanpa sempat mengeksekusi blok defer, nilai status di Redis Hash masih tertulis RUNNING. Atasi hal ini dengan pola heartbeat TTL:

  • Worker secara periodik menyentuh key job:alive:<job_id> dengan TTL pendek (misal 5-10 detik).
  • Saat polling API memeriksa status dari hash, API juga memeriksa keberadaan key job:alive:<job_id>.
  • Jika key liveness hilang namun status hash masih RUNNING, client dapat menyimpulkan bahwa worker mengalami kegagalan (crashed/stale) dan memicu recovery pipeline.

Optimasi Polling Client dengan Minimal Compute Cost

Menghadapi ribuan client polling status bersamaan ke Redis dapat menyebabkan utilisasi bandwidth dan CPU spike. Terapkan strategi optimasi berikut:

  • Short-Lived Caching di API Gateway: Simpan hasil telemetri worker di in-memory cache API gateway (seperti sync.Map atau GoCache) selama 500ms - 1 detik untuk menghindari hit langsung ke cluster Redis saat polling client padat.
  • Subpath Querying: Hindari HGETALL jika UI hanya menampilkan indikator bar progres. Gunakan HMGET job:telemetry:<id> progress status untuk menghemat alokasi buffer deserialisasi jaringan.
  • Exponential Backoff Polling: Frontend harus memperlebar interval polling secara bertahap dari 500ms hingga 5 detik seiring durasi panjang eksekusi job, mengurangi total hit ke backend.