Implementasi tooling AI berbasis CLI seperti Claude Code atau GitHub Copilot CLI di level korporat umumnya dialihkan melalui centralized reverse proxy gateway untuk kebutuhan audit, filtering, dan manajemen kuota token. Masalah kritis muncul ketika deployment proxy mengalami lonjakan error HTTP 429 (Rate Limit Exceeded), HTTP 502 (Bad Gateway), memory leak, dan token exhaustion tak terkendali meskipun traffic CLI tampak normal.
Akar masalah utama berasal dari kegagalan propagasi pembatalan koneksi (client abort/SIGINT) pada transmisi chunked Server-Sent Events (SSE). Proxy mempertahankan koneksi upstream ke model provider dan terus menampung response buffer di memori sampai proses selesai.
Gejala dan Dampak Sistem
Anomali terdeteksi melalui metrik gateway dan laporan pengguna CLI dengan karakteristik berikut:
- Lonjakan Kuota dan HTTP 429: Konsumsi token upstream (seperti Anthropic atau OpenAI) membengkak di luar volume output riil yang diterima engineer. Upstream mengembalikan HTTP 429 akibat throughput token per minute (TPM) habis.
- Memory Spike pada Proxy Pods: Node.js event loop lag melonjak tajam atau Go runtime garbage collector (GC) stress akibat heap memory menahan buffer chunked data yang tidak pernah di-flush ke client.
- HTTP 502 Bad Gateway: Proxy kehabisan socket atau memory threshold (OOMKilled) terlewati, memicu restart container dan memutus koneksi aktif lainnya.
- Dangling Upstream Connection: Developer menekan
Ctrl+C(SIGINT) pada terminal untuk membatalkan prompt, namun proxy upstream tetap aktif melakukan inferensi hingga token konteks maksimal tercapai.
Analisis Root Cause: Chunked Buffer Leak
Saat CLI memutus koneksi TCP downstream, HTTP client standar di sisi proxy tidak otomatis menghentikan request upstream jika signal event listener tidak dipasang secara eksplisit.
Pada gateway berbasis Node.js/Go tanpa penanganan lifecycle request yang tepat, pipeline stream downstream terputus (broken pipe), namun upstream reader tetap mengonsumsi chunk secara background. Buffer internal terus menimbun data yang tidak dapat di-drain, menyebabkan buffer leak di memori dan billing token provider terus berjalan hingga stream upstream mengirimkan token [DONE].
Langkah Isolasi Log Trace
Identifikasi request yatim (orphaned requests) menggunakan korelasi request_id pada log gateway. Filter kondisi di mana durasi upstream jauh melampaui downstream disconnect event:
// Contoh trace log JSON yang terisolasi
{
"timestamp": "2026-03-30T10:14:02.110Z",
"level": "WARN",
"trace_id": "trace-ai-cli-8fbc92",
"event": "client_disconnected_early",
"client_ip": "10.200.4.12",
"bytes_sent_downstream": 4128
}
{
"timestamp": "2026-03-30T10:14:48.330Z",
"level": "ERROR",
"trace_id": "trace-ai-cli-8fbc92",
"event": "upstream_stream_finished_orphaned",
"tokens_consumed": 3840,
"upstream_duration_ms": 46220,
"wasted_cost_usd": 0.0576
}Log menunjukkan downstream terputus pada detik ke-02, tetapi upstream request tetap aktif mengonsumsi 3.840 token selama 46 detik berikutnya.
Solusi Perbaikan: Propagasi Context Cancellation
Mitigasi dilakukan dengan mengikat lifecycle koneksi downstream ke request lifecycle upstream menggunakan context.WithCancel (Go) atau AbortController (Node.js).
Implementasi Go Reverse Proxy
Gunakan r.Context() yang secara native mendeteksi penutupan koneksi client:
package main
import (
"context"
"io"
"net/http"
)
func streamProxyHandler(w http.ResponseWriter, r *http.Request) {
// Buat context turunan dari downstream request
ctx, cancel := context.WithCancel(r.Context())
defer cancel()
reqUpstream, err := http.NewRequestWithContext(ctx, http.MethodPost, "https://api.anthropic.com/v1/messages", r.Body)
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
reqUpstream.Header = r.Header.Clone()
respUpstream, err := http.DefaultClient.Do(reqUpstream)
if err != nil {
return
}
defer respUpstream.Body.Close()
// Replikasi header SSE
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache")
w.Header().Set("Connection", "keep-alive")
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "Streaming unsupported", http.StatusInternalServerError)
return
}
buf := make([]byte, 4096)
for {
select {
case <-ctx.Done():
// Client menutup koneksi (SIGINT/abort), putus upstream
return
default:
n, err := respUpstream.Body.Read(buf)
if n > 0 {
_, writeErr := w.Write(buf[:n])
if writeErr != nil {
// Client write gagal, batalkan upstream segera
cancel()
return
}
flusher.Flush()
}
if err != nil {
if err == io.EOF {
return
}
return
}
}
}
}Implementasi Node.js
Pada Node.js gateway (Express/Fastify/Native HTTP), sinkronkan event close client ke AbortController upstream fetch:
import http from "node:http";
export function handleProxy(clientReq, clientRes) {
const controller = new AbortController();
const { signal } = controller;
// Tangkap pembatalan downstream
clientReq.on("close", () => {
if (!clientRes.writableEnded) {
controller.abort();
}
});
fetch("https://api.openai.com/v1/chat/completions", {
method: "POST",
headers: clientReq.headers,
body: clientReq,
duplex: "half",
signal,
})
.then((upstreamRes) => {
clientRes.writeHead(upstreamRes.status, Object.fromEntries(upstreamRes.headers));
const reader = upstreamRes.body.getReader();
function pump() {
return reader.read().then(({ done, value }) => {
if (done) {
clientRes.end();
return;
}
// Backpressure handling: tunggu drain jika buffer penuh
const canContinue = clientRes.write(value);
if (canContinue) {
return pump();
} else {
clientRes.once("drain", pump);
}
});
}
return pump();
})
.catch((err) => {
if (err.name === "AbortError") {
// Stream downstream dihentikan secara normal oleh client
return;
}
if (!clientRes.headersSent) {
clientRes.writeHead(502);
}
clientRes.end();
});
}Verifikasi Perbaikan dengan Integration Test
Jalankan integration test tanpa framework tambahan untuk memastikan connection abort downstream mematikan upstream reader secara deterministik:
package main
import (
"context"
"net/http"
"net/http/httptest"
"sync/atomic"
"testing"
"time"
)
func TestClientAbortStopsUpstreamConsumption(t *testing.T) {
var upstreamTokensConsumed int64
// Mock Upstream Server: menghasilkan stream tanpa batas
upstreamMock := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
flusher := w.(http.Flusher)
for {
select {
case <-r.Context().Done():
return
default:
atomic.AddInt64(&upstreamTokensConsumed, 1)
w.Write([]byte("data: chunk\n\n"))
flusher.Flush()
time.Sleep(10 * time.Millisecond)
}
}
}))
defer upstreamMock.Close()
// Jalankan proxy handler mengarah ke mock upstream
proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
reqUpstream, _ := http.NewRequestWithContext(r.Context(), http.MethodGet, upstreamMock.URL, nil)
resp, err := http.DefaultClient.Do(reqUpstream)
if err != nil {
return
}
defer resp.Body.Close()
w.Header().Set("Content-Type", "text/event-stream")
io.Copy(w, resp.Body)
}))
defer proxy.Close()
// Client simulasi: membaca sebentar lalu memutus koneksi (SIGINT)
ctx, clientCancel := context.WithCancel(context.Background())
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, proxy.URL, nil)
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("Request failed: %v", err)
}
buf := make([]byte, 64)
resp.Body.Read(buf) // baca 1 chunk
clientCancel() // trigger client abort
resp.Body.Close()
time.Sleep(50 * time.Millisecond)
consumedAfterCancel := atomic.LoadInt64(&upstreamTokensConsumed)
time.Sleep(100 * time.Millisecond)
consumedFinal := atomic.LoadInt64(&upstreamTokensConsumed)
if consumedFinal > consumedAfterCancel+1 {
t.Fatalf("Leak terdeteksi! Upstream terus mengonsumsi data setelah cancel. Awal: %d, Akhir: %d",
consumedAfterCancel, consumedFinal)
}
}Kesimpulan
Mencegah token exhaustion pada implementasi LLM CLI tidak cukup hanya dengan menerapkan rate-limiting di layer API. Propagasi sinyal pembatalan TCP downstream ke upstream context, disertai penanganan backpressure event pada chunked buffer, adalah persyaratan wajib arsitektur gateway LLM untuk mencegah kebocoran kuota token dan kehabisan memori runtime.
Komentar
0 komentar
Masuk ke akun kamu untuk ikut berkomentar.
Belum ada komentar
Jadilah yang pertama ikut berdiskusi!