Profiling Socket Buffer dan Heap Bloat pada Ingestion Webhook Payment Gateway Saat Flash Sale

+-----------------------------------------------------------------------------+
|                        ANATOMI PACKET & MEMORY INGESTION                    |
+-----------------------------------------------------------------------------+
  [ NIC (Network Interface) ]
              |
              v  DMA transfer
    +------------------+
    |  RX Ring Buffer  |
    +------------------+
              |
              v  Hard IRQ -> ksoftirqd (NAPI Poll Loop)
    +--------------------------------------------------------+
    | Linux Kernel Network Subsystem                         |
    |                                                        |
    |  * Alokasi sk_buff (truesize: 2 KB - 4 KB per frame)   |
    |  * TCP State Machine & Congestion Control              |
    |  * Socket Receive Queue (SO_RCVBUF: skmem usage)        |
    |    +--------------+--------------+--------------+      |
    |    | sk_buff #1   | sk_buff #2   | sk_buff #N   | ...  |
    |    +--------------+--------------+--------------+      |
    +---------------------------+----------------------------+
                                |
                                v  epoll_wait() / Go Netpoller
    +--------------------------------------------------------+
    | User-Space Runtime (Go Webhook Ingestion Service)      |
    |                                                        |
    |  * syscall.Read() -> io.Copy() ke Heap                 |
    |  * sync.Pool vs Raw Allocations                        |
    |  * Unbounded / Large Buffered Channel:                 |
    |    ch := make(chan *Payload, 100000)                   |
    |    +-> Heap Bloat + GC Mark/Sweep Latency Spike        |
    +--------------------------------------------------------+

truesize = sizeof(struct sk_buff) + ksize(data_buffer)

ss -ntm '( sport = :8443 or dport = :8443 )'

ESTAB  1412000  0   10.244.3.12:8443  10.244.1.88:45210
     skmem:(r1835008,rb2097152,t0,tb0,f262144,w0,o0,bl0,d0)
ESTAB  1289600  0   10.244.3.12:8443  10.244.2.14:58392
     skmem:(r1677721,rb2097152,t0,tb0,f419431,w0,o0,bl0,d0)

slabtop -o -s c | head -n 15

 Active / Total Objects (% used)    : 3412901 / 3450000 (98.9%)
 Active / Total Slabs (% used)      : 215625 / 215625 (100.0%)
 Active / Total Caches (% used)     : 110 / 142 (77.5%)
 Active / Total Size (% used)       : 3120450K / 3180000K (98.1%)

  OBJS ACTIVE  USE OBJ SIZE  SLABS OBJ/SLAB CACHE SIZE NAME
1245000 1245000 100%    2.00K  77812       16   2490000K kmalloc-2k
 350000  350000 100%    0.77K  35000       10    280000K skbuff_head_cache
 210000  208000  99%    0.21K  11666       18     46664K sock_inode_cache

#include <net/sock.h>
#include <linux/skbuff.h>

kprobe:tcp_data_queue
{
    $sk = (struct sock *)arg0;
    $skb = (struct sk_buff *)arg1;
    
    // Cek apakah buffer soket penuh (rmem_alloc > rcvbuf)
    if (sk->sk_rmem_alloc.counter >sk->sk_rcvbuf) {
        printf("Socket buffer overflow! Pid: %d, Comm: %s, rmem_alloc: %d, rcvbuf: %d
",
            pid, comm, sk->sk_rmem_alloc.counter,sk->sk_rcvbuf);
    }
}

tracepoint:sock:sock_exceed_buf_limit
{
    printf("Buffer limit exceeded on sock: %p, task: %s
", args->sk, curtask->comm);
}

// ANTI-PATTERN: Mengakumulasi memory di heap saat downstream lambat
type IngestionWorker struct {
    payloadQueue chan *WebhookEvent
}

func NewWorker() *IngestionWorker {
    return &IngestionWorker{
        // Unbounded buffer atau queue yang terlalu besar
        payloadQueue: make(chan *WebhookEvent, 200000),
    }
}

func (w *IngestionWorker) HandleWebhook(c *gin.Context) {
    var evt WebhookEvent
    if err := c.ShouldBindJSON(&evt); err != nil {
        c.Status(http.StatusBadRequest)
        return
    }
    
    // Langsung return 200 OK ke payment gateway agar tidak timeout
    // Data ditumpuk di heap tanpa backpressure mekanis
    w.payloadQueue <- &evt
    c.Status(http.StatusOK)
}

+-----------------------------------------------------------------------------+
|                        BACKPRESSURE CONTROL LOOP                            |
+-----------------------------------------------------------------------------+
 [ Payment Gateway / Aggregator ]
               |
               |  1. HTTP POST Webhook (Rate: 150k RPS)
               v
 [ Linux Kernel TCP Subsystem ]
   * tcp_rmem clamped (Max RCVBUF: 256KB - 512KB)
   * Socket buffer mendekati penuh -> TCP Window Clamped ke 0
   * Mengirim TCP Zero-Window Frame (Upstream dipaksa back-off)
               |
               |  2. sys_read() via Netpoller
               v
 [ Go User Space Engine ]
   * Bounded Worker Pool (Fixed concurrency: e.g. 1,024 workers)
   * Fast Rejection (HTTP 429 Too Many Requests) via Sync Pool Token
   * Direct Kafka Handoff (Batch Zero-Copy Serialization)

# Nonaktifkan alokasi buffer kernel yang terlalu agresif
# min: 4KB, default: 64KB, max: 256KB (mencegah single connection memonopoli RAM)
net.ipv4.tcp_rmem = 4096 65536 262144

# Batasi receive buffer default untuk non-TCP socket
net.core.rmem_default = 65536
net.core.rmem_max = 262144

# Optimasi TCP backlog queue untuk koneksi baru yang meledak serentak
net.core.somaxconn = 32768
net.ipv4.tcp_max_syn_backlog = 16384

# Aktifkan TCP syncookies untuk menangkal SYN flood liar saat spike traffic
net.ipv4.tcp_syncookies = 1

# Batasi alokasi total memory TCP (pages: 4KB per page)
# min, pressure, max: jika mencapai batas pressure, kernel otomatis menekan window size
net.ipv4.tcp_mem = 196608 262144 393216

package main

import (
	"context"
	"errors"
	"io"
	"net/http"
	"sync"
	"time"

	"github.com/bytedance/sonic" // High-performance SIMD-accelerated JSON parser
)

const (
	MaxConcurrentRequests = 2000
	MaxPayloadBytes       = 64 * 1024 // 64 KB batas wajar webhook bank
)

// WebhookPayload merepresentasikan schema payload notifikasi payment gateway
type WebhookPayload struct {
	TransactionID string  `json:"transaction_id"`
	ReferenceID   string  `json:"reference_id"`
	Amount        float64 `json:"amount"`
	Status        string  `json:"status"`
	Timestamp     int64   `json:"timestamp"`
}

// Pool untuk mengurangi beban alokasi short-lived objects di Go heap
var payloadPool = sync.Pool{
	New: func() any {
		return new(WebhookPayload)
	},
}

var bufferPool = sync.Pool{
	New: func() any {
		b := make([]byte, MaxPayloadBytes)
		return &b
	},
}

// Bounded Semaphore untuk menerapkan hard-limit concurrent workers
var workerTokens = make(chan struct{}, MaxConcurrentRequests)

func WebhookHandler(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		w.WriteHeader(http.StatusMethodNotAllowed)
		return
	}

	// 1. Non-blocking acquisition: Backpressure sinyal ke upstream payment provider
	select {
	case workerTokens <- struct{}{}:
		defer func() { <-workerTokens }()
	default:
		// Saat pool penuh, langsung kembalikan 429. Payment gateway akan melakukan
		// exponential backoff retry sesuai protokol webhook perbankan standard.
		w.Header().Set("Retry-After", "3")
		w.WriteHeader(http.StatusTooManyRequests)
		return
	}

	// 2. Proteksi ukuran request untuk mencegah malicious OOM payload injection
	r.Body = http.MaxBytesReader(w, r.Body, MaxPayloadBytes)

	bufPtr := bufferPool.Get().(*[]byte)
	defer bufferPool.Put(bufPtr)
	buf := *bufPtr

	n, err := io.ReadFull(r.Body, buf)
	if err != nil && !errors.Is(err, io.ErrUnexpectedEOF) && !errors.Is(err, io.EOF) {
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	payload := payloadPool.Get().(*WebhookPayload)
	defer func() {
		// Reset data sebelum dikembalikan ke pool
		*payload = WebhookPayload{}
		payloadPool.Put(payload)
	}()

	// 3. Fast Parsing menggunakan zero-copy semantics parser
	if err := sonic.Unmarshal(buf[:n], payload); err != nil {
		w.WriteHeader(http.StatusUnprocessableEntity)
		return
	}

	// 4. Delegasi pemrosesan langsung ke broker (misal: sync producer Kafka)
	// Payload langsung dipush dan tidak ditahan dalam in-memory channel Go
	if err := dispatchToPipeline(r.Context(), payload); err != nil {
		w.WriteHeader(http.StatusInternalServerError)
		return
	}

	w.WriteHeader(http.StatusOK)
}

func dispatchToPipeline(ctx context.Context, p *WebhookPayload) error {
	// Simulasi operasi non-blocking I/O direct write ke network layer broker
	select {
	case <-ctx.Done():
		return ctx.Err()
	case <-time.After(5 * time.Millisecond):
		return nil
	}
}

Metrik Pemantauan           Sebelum Tuning              Setelah Tuning
-----------------------------------------------------------------------------
Ingestion RSS Memory        4.0 GB (OOM Killed)         840 MB (Stabil)
Go Runtime In-Use Heap      480 MB                      190 MB
Kernel Slab Mem (kmalloc)   2.8 GB                      140 MB
TCP Recv-Q Peak             1.4 MB / socket             < 32 KB / socket
p99 Ingestion Latency       Timeout (> 15.000 ms)       42 ms
HTTP 500 / Crash Events     Ratusan Pod Restarts        0 Restart
HTTP 429 Backpressure       0% (Silent Failure)         1.8% (Graceful Retry)
-----------------------------------------------------------------------------