From 100M Stuck Messages to Zero OOM: A 7-Month War with a Go Event Pipeline

Introduction

At TrueProfit, our TruePixel service processes real-time e-commerce events for Shopify merchants — ad tracking, attribution, visitor analytics, product stats. At peak, we handle 100K+ data points per hour across hundreds of thousands of unique users.

Over 7 months — August 2025 to March 2026 — I fought three distinct battles with this pipeline. Each fix revealed the next problem. Phase 1 cleared a 100-million-message backlog. Phase 2 slowed but didn't stop an OOM cycle. Phase 3 eliminated it entirely. This is the full story.

The Architecture

The Event Processor sits at the center: it consumes from Kafka, routes each event through 5 stages (index → ad_lead → attribution → customer → product_stats), and writes to Elasticsearch, MongoDB, and Redis. 30 pods in production.

Simple enough in theory. The pain came from scale: sustained high throughput, 50K+ unique shopID:clientID keys in flight, and incremental features shipped without considering memory budgets.

Phase 1: The 100M Message Meltdown

The Crisis

Monday morning. The Kafka consumer lag dashboard turns red. 100 million+ messages are queued and growing. Every 7 days, the backlog accumulates to 100M+, pods restart under memory pressure, and the cycle repeats. I spent 3 days in a war room fixing this.

Root Cause: Unbounded Worker Concurrency

The event processor was configured with 150 concurrent workers and unlimited per-user goroutines:

// Production config — this will hurt you
workerPool: pond.NewPool(150, pond.WithQueueSize(300))
// No limit on concurrent users
// Cleanup: every 5 minutes (too slow)
// Result: 150 workers × goroutines-per-event × 50K+ users = CPU overload

150 workers times multiple goroutines per event times 50K+ unique users: the Go scheduler couldn't cope. GC pauses cascaded. Kafka consumer lag grew exponentially.

Worse: there was a race condition in event ordering. Events for the same user were processed out of order, corrupting attribution data.

The Fix: 48 Hours, 12 Hotfix Commits

August 27: Fix the race condition first — route events through per-user queues keyed by shopID:clientID. Each user's events are now processed sequentially.

August 28, 09:51: Emergency CPU reduction — slash the worker pool:

// AFTER: Bounded and explicit
workerPool: pond.NewPool(50, pond.WithQueueSize(100))  // 150 → 50
maxUsers:   300                                         // was unlimited
cleanupTicker: 30 * time.Second                        // was 5 minutes

August 28, 13:06: Final hardening — a bounded semaphore on the Kafka message handler itself:

// Bounded concurrency: at most 10 messages in flight
workerCount := 10
sem := make(chan struct{}, workerCount)

for msg := range partition.Messages() {
    sem <- struct{}{}  // Block if 10 in-flight — natural backpressure
    go func(m *sarama.ConsumerMessage) {
        defer func() { <-sem }()
        processEvent(m)
    }(msg)
}

Results

The backlog cleared. Pipeline stabilized. I thought the war was over. It wasn't.

Phase 2: The OOM That Wouldn't Die

New Feature, New Problem

With the pipeline stable, a team member (Dat) shipped a new feature: visitor tracking. Track unique visitors per week per shop — straightforward business requirement. The implementation added three storage layers per event:

1. Ristretto in-memory cache — check if visitor seen recently
2. Elasticsearch — check weekly visitor index
3. MongoDB — increment weekly unique visitor counter

Sounds reasonable. Except the cache was configured like this:

// 1 GB cache budget in a pod with 512 MiB memory limit.
cache, _ := ristretto.NewCache(&ristretto.Config[string, string]{
    NumCounters: 1e7,      // 10 MILLION key slots
    MaxCost:     1 << 30,  // 1 GB memory budget  ← the bomb
    BufferItems: 64,
})

1 GB cache budget in a pod with a 512 MiB memory limit. The math was always going to end one way.

The Slow Bleed

The OOM didn't happen immediately. Memory crept up over 3 days, then the pod was killed and the cycle restarted:

30 pods, all on the same death cycle. Grafana showed a perfect sawtooth pattern — memory climbs linearly, hits 512 MiB, drops to zero, climbs again. Two pod families (595d8d6476-* and 78cdb9f9dd-*) showed the same slope even after a deploy, confirming this was a code-level leak, not infrastructure.

The visitor tracking code also had compounding issues:

// One timeout wrapping everything — if ES is slow, MongoDB waits too
ctx, cancel := context.WithTimeout(baseCtx, 60*time.Second)
defer cancel()

checkCache(ctx)          // sub-millisecond, but context held for 60s
queryES(ctx)             // could wait up to 60s — holds ctx open
writeMongoDB(ctx)        // could wait up to 60s — holds ctx open

// Info-level logs in the hot path
// 1,000+ visitors/sec = 1,000+ log allocations/sec
logger.Info("visitor cache hit", "shopID", shopID)

The Fix (December 4, 2025)

Hung (with my review) landed the OOM triage in commit 0aae4ed:

// FIX 1: Halve the cache budget (stops the immediate bleeding)
cache, _ := ristretto.NewCache(&ristretto.Config[string, string]{
    NumCounters: 5e6,       // 5M (was 10M)
    MaxCost:     500 << 20, // 500 MB (was 1 GB)
})

// FIX 2: Auto-calculate cost — prevents over-admission
cache.Set(key, "exists", -1)  // -1 = auto-calculate, not manual "1"

// FIX 3: Per-operation timeouts — no more shared deadline
esCtx, esCancel := context.WithTimeout(baseCtx, 30*time.Second)
existsInES, _ := repo.VisitorExistsInES(esCtx, shopID, clientID)
esCancel()  // Release ES context immediately after use

mongoCtx, mongoCancel := context.WithTimeout(baseCtx, 15*time.Second)
repo.IncrementWeeklyVisitors(mongoCtx, shopID, clientID)
mongoCancel()

// FIX 4: Debug logs — no allocations on production hot path
logger.Debug("visitor cache hit", "shopID", shopID)  // was Info

December 12: Hung migrated visitor checks from MongoDB to Redis — eliminating the 190ms MongoDB upsert per event. January 6, 2026: I removed the MongoDB dependency from the product stats path entirely (b5935ba).

OOM frequency dropped. But the sawtooth pattern persisted at a slower rate. The real root cause was still lurking.

Phase 3: The Final Fix

Still Leaking

Even after the Phase 2 fixes, Grafana showed the same sawtooth: 32 MiB → slow climb → 480 MiB → OOM kill. Just took longer. I spent a week profiling with pprof. The architecture comparison tells the story:

Six root causes I finally identified through profiling:

The Nuclear Option: Rewrite

I deleted the entire OrderedEventProcessor (1,291 lines) and kafka_clean.go (294 lines) and replaced them with a bounded keyed dispatcher. The new processing pipeline:

The key innovation — the userDispatcher with dual-budget backpressure:

// The dispatcher: bounded workers + backpressure + per-user ordering
type userDispatcher struct {
    queues          map[string][]*sarama.ConsumerMessage
    pendingMessages int
    pendingBytes    int64
    readyKeys       chan string
    budgetAvailable chan struct{}
    mu              sync.Mutex
}

func (d *userDispatcher) dispatch(msg *sarama.ConsumerMessage) {
    userKey := string(msg.Key)
    for {
        d.mu.Lock()
        // BACKPRESSURE: Block the Kafka consume loop when budget exhausted.
        // This tells Kafka "slow down" — backpressure, not buffering.
        if d.pendingMessages >= 512 || d.pendingBytes > 32<<20 {
            d.mu.Unlock()
            <-d.budgetAvailable  // Wait for a worker to drain
            continue
        }
        d.queues[userKey] = append(d.queues[userKey], msg)
        d.pendingMessages++
        d.pendingBytes += int64(len(msg.Value))
        // Only signal readyKeys if this user isn't already in-flight.
        // Preserves per-user ordering without goroutine-per-user.
        if len(d.queues[userKey]) == 1 {
            d.readyKeys <- userKey
        }
        d.mu.Unlock()
        break
    }
}

Product stats: batch flush instead of load-all:

// BEFORE: Load ALL 50K Redis keys at once → 480 MiB spike
keys := getAllKeysFromRedis()  // 50,000 keys in one []string
bulkWriteToES(ctx, keys)      // one enormous allocation

// AFTER: Scan in batches of 100 → ~10 MiB peak
for batch := range scanKeysInBatches(ctx, 100) {
    bulkWriteToES(ctx, batch)  // process and release; GC can reclaim
}

Lua script SHA caching — eliminate 500 SCRIPT LOAD calls per second:

// BEFORE: SCRIPT LOAD on every flush — 500 extra roundtrips/sec
redis.Call("EVAL", luaScript, keys, args...)

// AFTER: Compute SHA once at startup, reuse forever
var atomicGetDeleteSHA = func() string {
    sha, err := rdb.ScriptLoad(ctx, redisScript).Result()
    if err != nil {
        panic(fmt.Errorf("failed to load redis script: %w", err))
    }
    return sha
}()

redis.Call("EVALSHA", atomicGetDeleteSHA, keys, args...)  // 0 extra roundtrips

The full Kafka claim scheduler flow — showing how backpressure propagates from dispatcher back to the consumer loop:

The Final Results

What I Learned: 7 Key Takeaways

The War Timeline