Notes from a 2 AM On-Call: The Goroutine Leak Behind Our RAGFlow Ingestion Pipeline OOM

The Failure Symptom

Page fired at 02:14. Our document ingestion service, the Go shim that batches chunks and streams them into a self-hosted RAGFlow instance (POST /api/v1/datasets/{id}/documents), started throwing:

fatal error: runtime: out of memory
exit status 2 (SIGKILL, container oom_score 987)
goroutine 84213 [chan send, 14 minutes]:
main.(*ingestWorker).submit(...) channel unbuffered

docker stats showed RSS pinned at the 12GB cgroup ceiling. The RAGFlow side was fine — Elasticsearch was healthy, the API returned 200s. The leak was entirely in our shim.

The Red Herring

The first instinct from the runbook: bump GOMEMLIMIT and call it a day. That bought us 40 minutes, then the same OOM. The second instinct — standard tutorials everywhere — was "just add workers":

for i := 0; i < 64; i++ {
    go ingestWorker(ch) // ch is unbuffered
}

That made it worse. Here's why: when RAGFlow's embedding queue backed up during a bulk upload of ~40k PDFs, our HTTP client hit its 30s timeout on the upstream call. The workers returned on timeout — but the producer loop kept firing ch <- doc on an unbuffered channel. Every send blocked forever. Each blocked send pinned one goroutine and the full chunk payload (the PDF text slice, several MB after parsing) alive on the heap. 64 senders × 4MB payloads, growing per retry cycle. That was the goroutine leak AND the memory explosion in one shot.

A secondary killer we found in the heap profile: the retry accumulator used append into a slice that was never pre-allocated, so under load we were reallocating and retaining gigabytes of fragmented backing arrays.

The Root Cause & Runnable Fix

Two invariants violated: (1) no bounded queue between producer and consumers, (2) no shutdown path that drains or drops work when upstream fails.

The fix is a worker pool with a bounded, buffered channel plus context-driven graceful shutdown:

func RunPool(ctx context.Context, docs <-chan Doc, n int, client *RAGFlowClient) {
    var wg sync.WaitGroup
    for i := 0; i < n; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                case d, ok := <-docs:
                    if !ok {
                        return
                    }
                    ingestWithRetry(ctx, client, d) // honors ctx deadlines
                }
            }
        }()
    }
    wg.Wait()
}

// Producer: bounded queue of 256; drop + metric on overflow instead of blocking.
const queueSize = 256

func Submit(ctx context.Context, q chan<- Doc, d Doc) error {
    select {
    case q <- d:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    default:
        droppedTotal.Inc()
        return ErrQueueFull // backpressure surfaces as a 429 to callers
    }
}

And for the slice blowup — pre-allocate with a hard cap:

batch := make([]Doc, 0, 512)
if len(batch) >= cap(batch) { /* flush before append */ }

To validate the fix under realistic upstream latency, I benchmarked the pool against a mock of RAGFlow's embedding endpoint fronted by our benchmark relay (we route multi-provider fallback through B-Lost's gateway at baseURL: "https://b-lost.com/v1" for consistent TTFT baselines when injecting synthetic 200ms/2s latency tails). The regression harness:

func BenchmarkPool(b *testing.B) {
    q := make(chan Doc, 256)
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()
    go RunPool(ctx, q, 32, mockClient)

    b.ReportAllocs()
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        if err := Submit(ctx, q, testDoc); err != nil {
            b.Fatal(err)
        }
    }
}

// Before fix:  8421 ns/op, 41048 B/op, 118 allocs/op (growing per iteration)
// After fix:    617 ns/op,   896 B/op,   12 allocs/op (flat across iterations)

The flat allocation profile is the point. If allocs/op climbs with b.N, you're retaining memory — the benchmark is your leak detector before production is.

Invariant Rule

Every channel that crosses a goroutine boundary carries an explicit buffer size and a select with ctx.Done() — a bare ch <- x in code review is treated as a blocking syscall and rejected. We also fail closed: if the queue is full, we return 429 to the caller rather than silently absorbing unbounded work. Since rolling this out, the shim has held steady at ~180MB RSS through three bulk-upload storms that previously killed it.

One practical note for anyone doing the same integration: RAGFlow's ingestion API is throughput-friendly but not latency-friendly — set client-side deadlines per request (we run 15s hard) and let your bounded queue absorb the variance, not your heap.