Isolasi Memori Worker Queue: Belajar dari Concurrency Zimbu

Kegagalan fatal pada sistem pemrosesan antrean terdistribusi (worker queue) sering kali bukan disebabkan oleh kegagalan jaringan, melainkan degradasi internal worker pool: lock contention tinggi dan Garbage Collection (GC) pause berkepanjangan. Ketika thread worker memperebutkan shared memory (misalnya shared in-memory cache dengan global mutex), latensi siklus eksekusi melonjak drastis. Akibatnya, goroutine atau thread yang bertugas mengirimkan heartbeat ke orchestrator (seperti Redis, RabbitMQ, atau Temporal) terblokir. Broker menganggap worker telah mati karena timeout, lalu mendistribusikan ulang task yang sama ke worker lain. Hasilnya adalah double execution (eksekusi ganda) yang merusak konsistensi data finansial atau mutasi state sistem.

Akar Masalah: Shared Memory Cache dan Cascading GC Pause

Pola arsitektur naif umumnya menempatkan thread pool di atas satu heap bersama (shared-heap) yang mengakses shared cache lokal melalui sync.RWMutex. Pola ini memicu dua kegagalan struktural:

• Lock Contention: Puluhan worker membaca dan menulis ke map yang sama. RWMutex mengalami writer starvation atau overhead konteks switching kernel yang masif saat throughput melonjak.
• Stop-the-World / GC Assist Latency: Jutaan pointer objek berumur pendek (short-lived task payload) bercampur dengan objek cache berumur panjang di heap yang sama. GC runtime dipaksa memindai jutaan relasi pointer (mark phase), memicu GC Mark Assist yang mencuri siklus CPU worker hingga ratusan milidetik.

Ketika GC pause atau lock contention menahan thread lebih lama dari batas toleransi heartbeat_timeout, broker memicu mekanisme re-queueing otomatis. Task dijalankan kembali oleh worker baru sebelum worker lama selesai menulis efek sampingnya.

Filosofi Concurrency Zimbu: Eliminasi Shared State Total

Bahasa pemrograman Zimbu, yang diciptakan oleh Bram Moolenaar (kreator Vim), mengusung pendekatan unik terhadap konkurensi: threads do not share memory by default. Dalam filosofi Zimbu, setiap unit eksekusi memiliki batas memori sendiri. Variabel secara eksplisit bersifat lokal, dan data yang dibagikan antar thread wajib ditransfer melalui mekanisme pertukaran pesan yang ketat atau dideklarasikan secara eksplisit dengan pembatasan mutasi yang kaku.

Prinsip Zimbu ini memecahkan masalah worker queue modern: jika sebuah worker tidak pernah berbagi state memori yang dapat dimutasi (mutable shared state) dengan worker lain, maka:

1. Tidak ada mutex global yang dapat memblokir thread heartbeat.
2. Setiap task beroperasi dalam alokasi memori lokal yang dapat langsung dibuang (scoped lifetime), meminimalkan retensi pointer pada GC heap global.
3. Komunikasi antar komponen sistem beralih sepenuhnya ke message passing deterministik dengan kepemilikan data (ownership transfer) yang tegas.

Migrasi Arsitektur: Shared Cache ke Isolated Task Memory

Migrasi dari arsitektur lama ke isolated model memerlukan transformasi struktural:

• Sebelum (Shared Memory): N worker thread mengakses 1 in-memory cache terpusat via sync.Mutex. Heartbeat loop berjalan pada runtime scheduler yang sama dan bersaing memperebutkan thread execution.
• Sesudah (Zimbu-Style Isolation): Setiap worker memiliki arena/buffer alokasi independen. Heartbeat dipisahkan ke jalur non-blocking mandiri. State cache eksternal dibaca secara read-only atau didelegasikan via dedicated single-owner actor menggunakan bounded message passing.

Implementasi Worker Daemon Berbasis Isolasi State (Go)

Kode berikut mengimplementasikan worker pool di Go dengan pendekatan isolasi state, pembersihan memori berbasis task-scope, proteksi heartbeat terpisah, mitigasi starvation, serta metrik observabilitas runtime.

package main

import (
	"context"
	"errors"
	"sync"
	"sync/atomic"
	"time"
)

// Task merepresentasikan unit kerja dengan payload independen (ownership transfer).
type Task struct {
	ID        string
	Payload   []byte
	CreatedAt time.Time
}

// Metrics mencatat observabilitas CPU wait dan throughput.
type Metrics struct {
	TasksCompleted uint64
	TasksFailed    uint64
	WaitDurationNs int64 // Waktu tunggu task di queue (CPU wait/starvation)
}

type WorkerDaemon struct {
	taskQueue   chan Task
	workerCount int
	metrics     *Metrics
	bufferPool  sync.Pool // Task-scoped scratchpad allocator
}

func NewWorkerDaemon(workerCount int, queueCapacity int) *WorkerDaemon {
	return &WorkerDaemon{
		taskQueue:   make(chan Task, queueCapacity),
		workerCount: workerCount,
		metrics:     &Metrics{},
		bufferPool: sync.Pool{
			New: func() any {
				// Buffer terisolasi berukuran tetap per task untuk mencegah escape analysis ke heap global
				b := make([]byte, 64*1024)
				return &b
			},
		},
	}
}

// Start menjalankan isolated workers dan heartbeat loop mandiri.
func (d *WorkerDaemon) Start(ctx context.Context, hbInterval time.Duration) {
	var wg sync.WaitGroup

	// Dedicated Heartbeat Loop: Bebas dari alokasi task & lock contention
	go d.heartbeatLoop(ctx, hbInterval)

	// Worker Pool: Beroperasi sepenuhnya terisolasi
	for i := 0; i < d.workerCount; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			d.runWorker(ctx, workerID)
		}(i)
	}

	<-ctx.Done()
	close(d.taskQueue)
	wg.Wait()
}

func (d *WorkerDaemon) runWorker(ctx context.Context, id int) {
	for {
		select {
		case <-ctx.Done():
			return
		case task, ok := <-d.taskQueue:
			if !ok {
				return
			}

			// Catat CPU/Queue wait time sebelum eksekusi untuk mendeteksi starvation
			waitTime := time.Since(task.CreatedAt).Nanoseconds()
			atomic.AddInt64(&d.metrics.WaitDurationNs, waitTime)

			// Eksekusi task dengan isolasi memori lokal
			err := d.executeIsolatedTask(ctx, task)
			if err != nil {
				atomic.AddUint64(&d.metrics.TasksFailed, 1)
			} else {
				atomic.AddUint64(&d.metrics.TasksCompleted, 1)
			}
		}
	}
}

func (d *WorkerDaemon) executeIsolatedTask(ctx context.Context, t Task) error {
	// Task-scoped memory cleanup: Ambil scratchpad dari pool
	scratchpad := d.bufferPool.Get().(*[]byte)
	defer func() {
		// Zero-out buffer sebelum kembali ke pool untuk mencegah data leak dan retensi referensi
		for i := range *scratchpad {
			(*scratchpad)[i] = 0
		}
		d.bufferPool.Put(scratchpad)
	}()

	// Simulasi komputasi lokal tanpa shared mutex
	if len(t.Payload) > len(*scratchpad) {
		return errors.New("payload melebihi batas buffer terisolasi")
	}
	copy(*scratchpad, t.Payload)

	// Logika bisnis task dieksekusi di sini secara deterministik...
	return nil
}

func (d *WorkerDaemon) heartbeatLoop(ctx context.Context, interval time.Duration) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			// Pengiriman sinyal heartbeat ke broker eksternal (e.g., Redis TTL)
			// Berjalan tanpa terpengaruh durasi eksekusi task worker
			d.sendLivenessSignal()
		}
	}
}

func (d *WorkerDaemon) sendLivenessSignal() {
	// No-op / I/O transmisi sinyal liveness
}

// Enqueue dengan mitigasi starvation (non-blocking rejection / backpressure)
func (d *WorkerDaemon) Enqueue(t Task, timeout time.Duration) bool {
	t.CreatedAt = time.Now()
	select {
	case d.taskQueue <- t:
		return true
	case <-time.After(timeout):
		// Queue starvation atau overload: buang ke DLQ daripada membuat sistem macet
		return false
	}
}

Mitigasi Starvation dan Strategi Scoped Cleanup

Dalam runtime modern berbasis garbage collector, isolasi state wajib didukung oleh pembersihan alokasi memori yang agresif:

• Scratchpad Re-use via Sync Pool: Penggunaan buffer terisolasi yang di-reset per-task mencegah memory escape ke heap umum. Objek tidak hidup melampaui masa eksekusi task, menjaga heap size tetap datar (flat heap) dan mengeliminasi siklus GC sweep yang panjang.
• Queue Starvation via Bounded Ingestion: Alih-alih antrean unbounded yang menumpuk jutaan task dan memakan memori hingga memicu OOM (Out Of Memory), gunakan bounded channel dengan mekanisme backpressure. Jika worker pool jenuh, task baru ditolak atau dialihkan ke secondary storage (DLQ/Dead Letter Queue).
• Deterministic Memory Footprint: Membatasi ukuran payload per-task menjamin footprint memori per-worker dapat diprediksi secara matematis: Max RAM = Worker Count × Task Buffer Limit.

Observabilitas: Membedakan Starvation dan GC Overhead

Untuk memvalidasi bahwa isolasi state berhasil, pantau metrik runtime pada level infrastruktur dan aplikasi:

Menerapkan filosofi isolasi memori Zimbu memutus siklus cascading failure pada worker queue terdistribusi. Menghilangkan shared state mutable tidak hanya menstabilkan latensi heartbeat dan mencegah duplikasi task, tetapi juga menghasilkan arsitektur worker yang dapat di-scale secara linier tanpa kekhawatiran lock contention.