Rethinking LLM Gateway Rate Limiting: From an In-Process Leaky Bucket to a Distributed Leaky-Bucket Matrix

Posted on Views: Word count in article: 19k Reading time ≈ 17 mins.
How Alibaba Cloud's Bailian gateway rebuilt LLM rate limiting: fixed window plus a distributed LiteTopic leaky-bucket matrix, and the response channel the article skips.

After reading the Alibaba Cloud Bailian gateway team’s “10× Cheaper Rate Limiting: How the Bailian Gateway Rebuilt LLM Rate Limiting with RocketMQ LiteTopic”, I came away with a new perspective on what a gateway can do. The first half recaps the core of Bailian’s rate-limiting design (fixed window + leaky bucket, the LiteTopic distributed leaky-bucket matrix, Suspend-based pacing); the second half fills in a key link the original article skips: after the consumer calls the model, how does the streamed response get back to the client? That part is my own reasoning, not Bailian’s implementation.

What rate limiting has to solve in the LLM era

Rate limiting used to be about two things: blocking abuse and preventing cascading overload. CPU, memory and bandwidth all scale fast, so handling them was relatively straightforward.

LLMs are different. GPU scale-out cycles are long, unit prices are high, and supply depends on the pace of hardware delivery. Bailian is Alibaba Cloud’s LLM platform service, with millions of users concurrently calling dozens of models including Qwen. Once the user base grew from thousands to millions, the problem rate limiting had to solve changed: how do you divide a finite GPU pool among a million tenants, with each tenant independently isolated and elastic on demand?

One false rate-limit decision means timeouts, retries, or even an outage on the user side. One admission past the overload point wastes scarce GPU compute and drags down other tenants as well.

Algorithm choice: fixed window + leaky bucket

Bailian does not face a single rate-limiting scenario; three stack on top of each other:

  • Baseline SLA rate limiting: stable, meterable, observable
  • Fine-grained control on the User + Model dimension, with cross-account constraints inside a customer’s own organization
  • Absorbing bursts from the largest customers: quota turned way up, but when the burst actually arrives the GPUs cannot keep up — hard limits produce widespread 503s, and letting traffic through causes overload

The final algorithm choice was fixed window + leaky bucket:

Fixed window rather than sliding window, because sliding windows are too strict: bursts at the boundary get counted exactly, while a fixed window is more tolerant of short-term fluctuation.

Leaky bucket rather than token bucket, because a leaky bucket releases at a constant rate, matching GPUs’ preference for steady state and sensitivity to spikes. A token bucket natively allows bursts — exactly the shape GPUs fear most.

The leaky bucket has to leave the process

Algorithm settled, new problem: burst traffic from the largest customers piles up inside the gateway process’s memory, and hundreds of thousands of requests can take down the gateway itself.

The leaky bucket must move from inside the process to outside it, buffered by an external store that is large enough and isolated enough. Physically separating “absorbing the requests” from “releasing them at a paced rate” — this is where LiteTopic enters the picture.

Traditional Topic / Group does not hold up

Bailian first built leaky buckets on traditional RocketMQ topics: a separate topic and consumer group per User + Model, plus a dedicated set of consumer machines. Once the top customers came onboard, three pain points surfaced:

Heavy metadata, slow to take effect. Traditional topics and groups must be pre-created and take tens of seconds to become active; a new user’s burst traffic can arrive before the topic even exists.

Low machine utilization. Shared consumption requires every instance in a consumer group to subscribe to exactly the same set of topics, otherwise messages pile up or get lost. To isolate a customer you need a dedicated group and dedicated machines — even on a day with zero messages, those machines stay resident.

Collateral damage from a single topic’s spike. Forcing multiple customers into one topic means one user’s message surge occupies nearly all consumption threads and blocks everyone else. The only fallback is “one machine set per customer”, trading machine count for isolation.

The traditional approach can serve a few top burst customers, but it cannot let any mid-to-large customer automatically join the leaky bucket. For rate limiting to move from VIP treatment to a basic platform capability, the foundation had to be rebuilt.

LiteTopic’s three differences

LiteTopic is the lightweight queue in RocketMQ 5.x; its three key differences from traditional topics map directly onto the three pain points above.

Lightweight metadata: a single broker sustains LiteTopics in the millions, created on demand at runtime and reclaimed automatically by TTL. Sending a message to a nonexistent LiteTopic creates it implicitly, and the broker cleans up topics after a period of inactivity.

Differentiated subscriptions: instances in the same consumer group can subscribe to different subsets of LiteTopics, without backlog or message loss. All tenants share one pool of consumer pods — no more one machine set per customer.

Suspend-based consumption control: returning Suspend N (milliseconds) from the consume callback makes the broker pause pulling from that LiteTopic for N milliseconds, without affecting other LiteTopics. This is the key switch that lands the leaky bucket at the broker layer.

The reworked architecture

The reworked rate-limiting architecture

One leaky bucket per User + Model: bucket capacity is determined by the LiteTopic backlog limit and TTL, and the outflow rate is determined dynamically by the consumer-side Suspend duration. All buckets share one set of consumer machines; resources scale with total traffic rather than growing linearly with customer count.

What Suspend actually does

The architecture diagram line “rate limit hit → return Suspend(N), broker pauses that LiteTopic” deserves expansion.

After pulling a message, the consumer runs the rate-limit decision first. The result is either admit or pause. If it is pause, the consumer neither calls the model nor ACKs; instead it tells the broker: “give me the next message from this LiteTopic N milliseconds from now.” The consumption thread then returns immediately and goes to pull messages from other LiteTopics.

The key point is that the pause happens on the broker side, not in the consumer’s own sleep. After receiving Suspend(N), the broker will not deliver messages from that LiteTopic to this consumer for the next N milliseconds. But the same consumer’s pulls from other LiteTopics are completely unaffected — messages on other LiteTopics are still pulled and consumed as usual.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
t=0ms:
  consumer pulls a message from liteTopic-A-X
  rate-limit check: must pause
  consumer returns Suspend(500) → consumption thread released immediately

  meanwhile:
  consumer pulls a message from liteTopic-B-Y    ← unaffected
  rate-limit check: admit
  consumer calls the model, ACKs normally

t=500ms:
  broker resumes delivery for liteTopic-A-X
  consumer pulls that message again
  rate-limit check: may admit now
  consumer calls the model, ACKs normally

Contrast that with Thread.sleep(500): the consumption thread is stuck in sleep for 500 milliseconds. During those 500 ms it neither consumes the next messages of liteTopic-A-X nor any other LiteTopic’s messages. Once enough LiteTopics are being rate-limited, the thread pool fills up, and tenants that are not rate-limited start backlogging too.

The original article includes a POC comparison: subscribe to 50 LiteTopics, with the first 5 randomly triggering a 300ms rate limit. With Thread.sleep(300), each throttled topic holds a thread that never returns; when the throttled count grows to 40 the thread pool is nearly exhausted, and the remaining 10 healthy topics all back up. With Suspend(300), the consumption thread returns and is immediately released back to the pool to serve other LiteTopics, and the broker redelivers the suspended one after 300ms — the result is that only the 5 throttled topics back up while the other 45 stay healthy.

So the difference between Suspend and Sleep is not whether you wait — both wait 300ms. The difference is whether the consumption thread can go do something else while waiting. Suspend hands the waiting to the broker’s timer; the consumption thread does not wait and immediately serves other tenants.

This mechanism is also the physical implementation of the leaky bucket’s constant-rate release. Suppose a User + Model’s quota is 10 requests per second: after admitting one message the consumer does return Suspend(100), and the broker delivers the next message only once every 100ms — exactly 10 per second. When the quota changes, change Suspend’s N; no restart, no config change.

Suspend’s current minimum granularity is 30 milliseconds. Finer rate control requires supplementary local sleeps on the consumer side. Bailian’s actual practice: Suspend handles the main leaky-bucket pacing, with Sleep covering the rare sub-30ms precision cases.

Why LiteTopic suits the leaky bucket

Bucket capacity is, in engineering terms, effectively unbounded. An in-process queue has a hard-coded capacity and rejects once full. LiteTopics persist to broker disk: millions of queues per instance, backlog limits far beyond any in-process queue, and physical isolation from each other. Customer spikes get absorbed, accumulated, and released at a paced rate — waiting a few seconds before processing is almost always better than an immediate 429.

One leaky bucket per User + Model. Isolation becomes a naming convention: customer A’s Model X goes into liteTopic-A-X, customer B’s into liteTopic-B-X. Isolation is physical at the broker layer; A getting blocked cannot spread to B. Tenant isolation moves from “maintained by application code” to “naturally provided by the infrastructure”.

Each bucket’s rate can be adjusted independently and dynamically. Suspend N is computed in real time by policy, adjusting any bucket’s release speed at millisecond resolution: important customers keep a higher pace even when model load is high; when model scale-out lands, the whole matrix’s rates go up together; when one model shows signs of congestion, tighten just that column of buckets. No consumer-group restart, no config change, nothing the customer has to notice.

The question the original article leaves open: how the response gets back

The article compresses the core changes into three code snippets; the consume callback is the most important:

1
2
3
4
5
6
7
8
9
10
@Override
public ConsumeResult consume(MessageView msg) {
    String liteTopic = msg.getLiteTopic();
    long suspendMs = rateLimitPolicy.acquireOrSuspend(liteTopic);
    if (suspendMs > 0) {
        return ConsumeResult.Suspend(suspendMs);
    }
    invokeModel(msg);        // call the model directly when not rate-limited
    return ConsumeResult.SUCCESS;
}

The invokeModel(msg) line is skipped in the original. The analysis below is my own reasoning, not Bailian’s implementation — the original gives no details on response delivery, so this is a workable design I derived from the architecture’s constraints.

The problem: the client’s HTTP connection is held by the producer (the gateway entry point), while the model call happens on the consumer (a different set of pods). LLM responses are streamed, token by token. How does the consumer’s streamed response get back in real time to that still-open client connection on the producer?

There is also a hard constraint: consumption threads are precious — only 50 in the POC — and one streamed response cannot occupy a thread for 20 seconds doing nothing.

Why the response channel does not use LiteTopic

Since the request channel already uses LiteTopic, why not use another one for responses? Because the two channels need different things.

The request channel needs rate limiting (Suspend pacing), buffering (backlog to absorb spikes), isolation (millions of tenants), and it can wait tens of seconds. The response channel needs none of these: a chunk should be delivered the moment it arrives, deleted once read, naturally one-to-one (by reqId), and a few milliseconds of extra latency per chunk is perceptible to the user.

LiteTopic’s three-part capability set was tailor-made for the request channel; the response channel cannot use it. An LLM emits tokens one at a time, tens of milliseconds apart; if every chunk in the response channel went through the broker’s full path (routing, persistence, ready-set notification), each token would gain 3-5ms. A response with hundreds of tokens accumulates into visibly degraded fluency. Redis XADD/XREAD are in-memory operations, sub-millisecond, with no extra steps.

Then there is queue count. Bailian already has LiteTopics in the millions; adding a response channel with one stream per request would double the broker’s load to 2 million queues. On the Redis side, a million keys are just a bunch of strings, an order of magnitude smaller in memory.

Using Redis Streams as the response channel

My approach is to use Redis Streams as the response pipe between producer and consumer, one independent stream per request, keyed response:{reqId}. The whole path has two channels: requests go through LiteTopic, responses go through Redis, and reqId ties them together.

Full data flow of the request and response channels

Follow the numbered steps:

  1. The client sends an HTTP request to the producer; the producer generates a reqId.
  2. The producer writes the request message along with the reqId into LiteTopic (routed by User+Model).
  3. The consumer pulls the message from LiteTopic; if it hits the rate limit it returns Suspend(N) and waits, otherwise it continues.
  4. Using reqId as the correlation identifier, the consumer initiates a streaming call to the GPU inference service.
  5. The GPU returns the response token by token.
  6. For each chunk it receives, the consumer XADDs it into the Redis Stream response:{reqId}.
  7. The producer is continuously XREADing this stream in blocking mode and pushes each chunk to the client immediately.
  8. The producer streams the chunks back to the client over SSE, looping until it receives the done marker.

The two channels each do their own job: LiteTopic governs “at what pace requests are released to the GPU”; the Redis Stream governs “at what latency responses get back to the client”.

reqId threads the whole path

The producer generates the reqId and writes it into the LiteTopic message header. The consumer reads it from the message and builds the Redis Stream key by the same rule to write the response. The producer reads with the same key. Neither side needs extra notification: the key rule is the convention, and matching reqId is enough.

reqId threading the whole path

Producer core: the blocking XREAD loop

After writing to LiteTopic, the producer reads responses from the Redis Stream in blocking mode. XREAD’s BLOCK parameter keeps the wait from consuming CPU — it wakes up as soon as data arrives. An overall timeout is added as well, so the producer does not wait forever if the consumer crashes:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
func (g *GatewayProducer) HandleRequest(w http.ResponseWriter, r *http.Request) {
    reqID := uuid.New().String()
    streamKey := "response:" + reqID

    // build the message, stuff reqId into the header, route by user+model to a LiteTopic
    msg := &primitive.Message{Topic: g.parentTopic, Body: payload}
    msg.SetLiteTopic(fmt.Sprintf("litetopic-%s-%s", user, model))
    msg.WithProperty("reqId", reqID)
    g.mqProducer.SendSync(r.Context(), msg)

    // start SSE, stream the response from the Redis Stream and push it to the client
    w.Header().Set("Content-Type", "text/event-stream")
    flusher, _ := w.(http.Flusher)
    lastID := "0" // read from the start so the first chunk is not missed

    // overall timeout: if the consumer crashes or Redis dies, keep the producer from waiting forever
    // LLM streamed responses usually finish within 60 seconds; 120 seconds as a backstop
    ctx, cancel := context.WithTimeout(r.Context(), 120*time.Second)
    defer cancel()

    for {
        // XREAD BLOCK: block when there is no data, return immediately when there is
        result, err := g.rdb.XRead(ctx, &redis.XReadArgs{
            Streams: []string{streamKey, lastID},
            Count:   10,
            Block:   30 * time.Second,
        }).Result()

        if err == redis.Nil {
            continue // timed out with no data, keep waiting (the message may still be queued in Suspend)
        }
        if err != nil {
            // context deadline exceeded, or the client disconnected, or the Redis connection failed
            if ctx.Err() == context.DeadlineExceeded {
                fmt.Fprintf(w, "event: timeout\ndata: response incomplete\n\n")
                flusher.Flush()
            }
            return
        }

        for _, stream := range result {
            for _, message := range stream.Messages {
                lastID = message.ID // advance the cursor; the next read starts after this

                switch message.Values["type"] {
                case "chunk":
                    fmt.Fprintf(w, "data: %s\n\n", message.Values["chunk"])
                    flusher.Flush() // push to the client immediately
                case "done":
                    fmt.Fprintf(w, "event: done\ndata: [DONE]\n\n")
                    flusher.Flush()
                    return
                case "error":
                    fmt.Fprintf(w, "event: error\ndata: %s\n\n", message.Values["error"])
                    return
                }
            }
        }
    }
}

XREAD pulls by an ID cursor: you tell Redis the last ID you read, and it returns all new messages after that ID, blocking if there are none. lastID lives in the producer’s own memory and advances with every message read — no duplicates, no gaps, consistent order. Count: 10 caps each read at 10 messages, so a sudden burst of chunks from the consumer is not returned all at once, which would stall the loop. This value does not affect correctness — the lastID cursor guarantees no message loss; 5 or 20 work just as well.

This for loop neither runs forever nor busy-loops on CPU. It cannot run forever because there are four exits: a done event returns normally; an error event returns abnormally; when the client disconnects, r.Context() is canceled automatically, XREAD returns an error, and the if err != nil branch returns; and when the 120-second overall timeout hits the context deadline, it returns as well. It does not burn CPU because XREAD BLOCK 30000 is server-side blocking: the goroutine parks on IO wait and consumes no CPU, and the Redis server also parks the connection when there is no data, waiting to be woken by new messages. After 30 seconds with no data it returns Nil, continues, issues another XREAD, and parks again. A million concurrent requests means a million goroutines each parked waiting on their own key, with CPU consumption close to zero.

The overall timeout is a backstop. Consumer crash, Redis outage, a message landing in the dead-letter queue — all of these ultimately look like “the producer never receives done”. Without an overall timeout the producer would continue forever and the client would hang indefinitely. If no done arrives within 120 seconds, assume the consumer has a problem and return a timeout event to the client. One caveat: if the consumer stays Suspended for a long time (say rate-limited for 60 seconds), the HTTP connection stays open the whole time, so the timeout settings of the LB and Nginx in front of the gateway must exceed the maximum Suspend duration — otherwise the LB will cut the connection first.

Consumer core: call the model in the callback, write chunks to Redis

Besides the rate-limit decision and idempotency check, the consumer adds a processing:{reqId} marker to prevent a duplicate model call when a message is redelivered, and aborts the model call when WriteChunk fails:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
func (c *GatewayConsumer) Consume(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
    for _, msg := range msgs {
        reqID := msg.GetProperty("reqId")
        liteTopic := msg.GetLiteTopic()

        // rate-limit check: > 0 means Suspend is required
        suspendMs := c.rateLimitPolicy(liteTopic)
        if suspendMs > 0 {
            return consumer.Suspend(suspendMs), nil // consumption thread released immediately
        }

        // idempotency check: avoid a duplicate model call on redelivery
        // done marker: already fully processed last time
        if exists, _ := c.rdb.Exists(ctx, "done:"+reqID).Result(); exists > 0 {
            continue
        }
        // processing marker: the last attempt crashed halfway, done was never written
        // do not call the model again on redelivery, or the client would receive two stitched-together segments
        if exists, _ := c.rdb.Exists(ctx, "processing:"+reqID).Result(); exists > 0 {
            // write an error telling the client to re-issue the request
            WriteError(ctx, c.rdb, reqID, "previous attempt incomplete, please retry")
            continue
        }

        // mark as in-progress, auto-expires after 120 seconds (backstop in case of a crash before cleanup)
        c.rdb.Set(ctx, "processing:"+reqID, "1", 120*time.Second)

        // streaming model call, writing each chunk to the Redis Stream
        err := c.modelClient.Invoke(ctx, model, prompt, func(chunk string) error {
            // check whether the client has disconnected
            if n, _ := c.rdb.Exists(ctx, "cancel:"+reqID).Result(); n > 0 {
                return fmt.Errorf("client disconnected") // abort the model call, saving GPU
            }
            // a WriteChunk failure means Redis is down: the response cannot get back,
            // and continuing to call the model only burns GPU for nothing
            if err := WriteChunk(ctx, c.rdb, reqID, chunk); err != nil {
                return fmt.Errorf("redis unavailable, abort model call: %w", err)
            }
            return nil
        })

        if err != nil {
            // model call aborted (client disconnected or Redis is down)
            // write an error to notify the producer (if Redis is still alive), then ACK the current message
            // the current message is finished (including failure); move on to the next one
            WriteError(ctx, c.rdb, reqID, err.Error())
            c.rdb.Del(ctx, "processing:"+reqID)
            continue
        }

        // write the done marker, clear the processing marker, ACK
        WriteDone(ctx, c.rdb, reqID)
        c.rdb.Del(ctx, "processing:"+reqID)
    }
    return consumer.ConsumeSuccess, nil
}

The consumer adds two defenses. The processing:{reqId} marker addresses “the process crashed mid-call and the message got redelivered, triggering a second model call” — the first attempt already wrote several chunks to the Redis Stream and the client received part of the response; calling the model again would hand the client two stitched-together segments. With the marker in place, a redelivery that finds a processing marker knows the last attempt never finished: write an error asking the client to re-issue the request instead of calling again.

Aborting the model call when WriteChunk fails addresses “Redis is down but we keep burning GPU”. After Redis dies, no one can read the chunks the consumer writes; continuing to call the model is pure waste. The callback returns an error, Invoke aborts, and no ACK happens (a Redis outage counts as an infrastructure failure — let the MQ redeliver and wait for Redis to recover).

Variant: producer and consumer in one process — use a channel instead of Redis

The earlier design assumed producer and consumer are separate pods, so responses had to cross processes and Redis Streams was used. But if producer and consumer live in the same process — the gateway process both accepts HTTP requests and writes LiteTopic, and also acts as the consumer pulling messages and calling the model — the response channel does not need Redis at all; a Go channel inside the process is enough.

Requests still go through LiteTopic (rate limiting, buffering and isolation still need the broker to provide them); only the response changes from “cross-process over Redis” to “in-process over a channel”:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
Same gateway process
┌───────────────────────────────────────────────────────────────┐
│                                                               │
│  HTTP Handler (goroutine A)             Consumer (goroutine B)│
│  ┌────────────────┐                ┌─────────────────────┐    │
│  │ generate reqId │                │ pull message        │    │
│  │ write LiteTopic│                │ rate-limit check    │    │
│  │                │                │ call model          │    │
│  │ wait on chan   │ ◄───────────── │ chunk → chan        │    │
│  │ push to client │ Go channel     │ done → chan         │    │
│  └────────────────┘                └─────────────────────┘    │
│                                                               │
│         ┌──────────────────────┐                              │
│         │  sync.Map            │                              │
│         │  reqId → chan chunk  │                              │
│         └──────────────────────┘                              │
└───────────────────────────────────────────────────────────────┘

The producer registers a channel keyed by reqId, and the consumer finds that channel through reqId and writes to it directly:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
// in-process response registry
var responseChans sync.Map // reqId -> chan StreamEvent

type StreamEvent struct {
    Type  string // "chunk" / "done" / "error"
    Chunk string
    Error string
}

// producer: register the channel, write LiteTopic, then block and wait
func (g *Gateway) HandleRequest(w http.ResponseWriter, r *http.Request) {
    reqID := uuid.New().String()
    ch := make(chan StreamEvent, 100) // buffered, so the consumer's writes do not block
    responseChans.Store(reqID, ch)
    defer responseChans.Delete(reqID)

    // write LiteTopic (same as before)
    msg := &primitive.Message{Topic: g.parentTopic, Body: payload}
    msg.SetLiteTopic(fmt.Sprintf("litetopic-%s-%s", user, model))
    msg.WithProperty("reqId", reqID)
    g.mqProducer.SendSync(r.Context(), msg)

    // SSE: read responses from the channel and push them to the client
    w.Header().Set("Content-Type", "text/event-stream")
    flusher, _ := w.(http.Flusher)

    ctx, cancel := context.WithTimeout(r.Context(), 120*time.Second)
    defer cancel()

    for {
        select {
        case <-ctx.Done():
            fmt.Fprintf(w, "event: timeout\ndata: response incomplete\n\n")
            flusher.Flush()
            return
        case event := <-ch:
            switch event.Type {
            case "chunk":
                fmt.Fprintf(w, "data: %s\n\n", event.Chunk)
                flusher.Flush()
            case "done":
                fmt.Fprintf(w, "event: done\ndata: [DONE]\n\n")
                flusher.Flush()
                return
            case "error":
                fmt.Fprintf(w, "event: error\ndata: %s\n\n", event.Error)
                return
            }
        }
    }
}

// consumer: after pulling a message, find the channel by reqId and write to it directly
func (g *Gateway) Consume(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
    for _, msg := range msgs {
        reqID := msg.GetProperty("reqId")
        liteTopic := msg.GetLiteTopic()

        suspendMs := g.rateLimitPolicy(liteTopic)
        if suspendMs > 0 {
            return consumer.Suspend(suspendMs), nil
        }

        // find the corresponding channel in the registry
        val, ok := responseChans.Load(reqID)
        if !ok {
            continue // the producer is already gone (client disconnected), skip
        }
        ch := val.(chan StreamEvent)

        // call the model, writing each chunk straight to the channel
        err := g.modelClient.Invoke(ctx, model, prompt, func(chunk string) error {
            // check the channel is still there (after a client disconnect the producer runs defer Delete)
            _, ok := responseChans.Load(reqID)
            if !ok {
                return fmt.Errorf("client disconnected")
            }
            ch <- StreamEvent{Type: "chunk", Chunk: chunk}
            return nil
        })

        if err != nil {
            ch <- StreamEvent{Type: "error", Error: err.Error()}
        } else {
            ch <- StreamEvent{Type: "done"}
        }
    }
    return consumer.ConsumeSuccess, nil
}

The channel design needs no Redis: no XADD/XREAD, no TTL cleanup of stream keys, no done/processing/cancel markers in Redis. A client disconnect is just the producer goroutine exiting; defer responseChans.Delete(reqID) cleans up automatically, and when the consumer’s Load finds no channel it knows the client is gone and aborts the model call. An entire Redis dependency disappears.

But the cost is obvious. Producer and consumer are bound to the same process: if the process dies, both sides die with it. In the Redis design, a consumer crash just means the MQ redelivers to another consumer instance; with the in-process design, a process restart loses every in-flight request, and clients can only detect it through the overall timeout and retry. Also, if the consumer’s model call involves CPU-intensive preprocessing (image encoding, for example), it may compete with the HTTP handler for CPU — a non-issue for the split-process design.

At small to medium scale — thousands to tens of thousands of tenants — one process plus a channel is sufficient, with one less Redis dependency. At Bailian’s scale of a million tenants, producer and consumer must be deployed separately and scaled independently; Redis Streams is the more realistic choice.

A few engineering details

When to ACK

ACK comes right after invokeModel, but it depends on the case:

1
2
3
model call succeeds        → write done  → ACK
model business failure     → write error → ACK (no redelivery; a retry would only burn GPU again)
Redis/infrastructure fault → no ACK      → broker redelivers

Model business errors are not redelivered: a retry would very likely produce the same result and waste GPU. Only infrastructure failures cause redelivery.

Idempotency

ACKs between broker and consumer can be lost, so the same message may be delivered twice. The consumer code has two checks: the done:{reqId} marker guards against redelivery of a request that was fully processed — skip it, do not call the model again; the processing:{reqId} marker guards against redelivery of a request that crashed halfway — write an error so the client can re-issue, again without a duplicate model call. A single LLM inference burns seconds of GPU time; a duplicate call is pure waste.

Client disconnect

When the client closes the browser, the producer detects the broken connection (in Go, r.Context().Done() fires automatically) and writes a cancel marker to cancel:{reqID}. The consumer checks between chunks and aborts the model call as soon as it sees the cancel. Without this handling, the model keeps running inference after the user is gone, wasting seconds to tens of seconds of GPU time per call.

Takeaways

Bailian’s move from an in-process leaky bucket to a distributed leaky-bucket matrix is not really about algorithm novelty — fixed windows and leaky buckets are decades old. The real point is that at a million tenants the bottleneck is not the algorithm; it is what physical carrier you use to hold the structure you split out. Beyond algorithm selection, it is the leaky bucket’s physical carrier, the cost model of tenant isolation, and the latency characteristics of the response channel that determine whether the design scales.

The article hides the entire response-delivery mechanism inside the single line invokeModel(msg), yet that is the key to whether the whole thing can actually run. The response channel depends on the deployment shape: a Go channel when producer and consumer share a process — the simplest possible in-process delivery; Redis Streams when they are split — cross-process delivery with fault tolerance. Requests go through LiteTopic for rate limiting and buffering; responses use a channel or Redis depending on the deployment. Two channels, each doing its own job — do not force them into one.

Suspend vs. Sleep, idempotent handling of message redelivery, the producer’s overall-timeout backstop, aborting the model call on client disconnect — these look like details, but at million-tenant scale any one of them failing is a production incident. The original Bailian article focuses on the rate-limiting architecture itself; these engineering details have to be filled in yourself.

References

This article is the English edition of a post originally published in Chinese.