How Grok2API Manages Concurrency Limits Per Account: Architecture and Implementation

Grok2API enforces per-account concurrency limits through a ConcurrencyLimiter interface that atomically acquires leases using either in-memory sharded maps or Redis Lua scripts, automatically releasing slots when requests complete.

The chenyme/grok2api project implements a sophisticated concurrency control mechanism to prevent upstream account saturation. By abstracting limit management behind a common interface, Grok2API supports both single-instance deployments and distributed multi-replica setups while maintaining strict per-account request ceilings. This architecture ensures that no single upstream account receives more simultaneous requests than its configured MaxConcurrent threshold.

The ConcurrencyLimiter Interface

At the core of the system lies the ConcurrencyLimiter interface defined in backend/internal/repository/runtime.go. This abstraction declares the Acquire method, which attempts to obtain a lease for a specific account key.

// Conceptual interface (based on repository/runtime.go)
type ConcurrencyLimiter interface {
    Acquire(ctx context.Context, key string, limit int) (release func(), acquired bool, err error)
}

The Acquire method returns three values: a release function that decrements the counter when invoked, a boolean indicating whether acquisition succeeded, and an error if the operation failed. This design allows the selector to attempt acquisition for multiple candidate accounts until it finds one with available capacity.

In-Memory Implementation

For single-instance deployments, Grok2API uses a sharded in-memory limiter implemented in backend/internal/infra/runtime/memory/store.go. This implementation partitions the key space across multiple shards to reduce lock contention.

Each shard maintains a map[string]int tracking active counts per account. When Acquire is called, it hashes the key to determine the appropriate shard, locks the shard mutex, and checks the current count against the limit.

// Simplified from backend/internal/infra/runtime/memory/store.go
func (l *ConcurrencyLimiter) Acquire(_ context.Context, key string, limit int) (func(), bool, error) {
    if limit <= 0 {
        return func() {}, true, nil
    }
    shard := &l.shards[shardIndex(key)]
    shard.mu.Lock()
    if shard.counts[key] >= limit {
        shard.mu.Unlock()
        return nil, false, nil
    }
    shard.counts[key]++
    shard.mu.Unlock()

    var once sync.Once
    return func() {
        once.Do(func() {
            shard.mu.Lock()
            defer shard.mu.Unlock()
            shard.counts[key]--
            if shard.counts[key] <= 0 {
                delete(shard.counts, key)
            }
        })
    }, true, nil
}

The release function uses sync.Once to guarantee the counter decrements exactly once, even if called multiple times. When the count reaches zero, the implementation deletes the key from the map to prevent memory leaks.

Redis Implementation

For distributed deployments running multiple API replicas, Grok2API provides a Redis-backed limiter in backend/internal/infra/runtime/redis/store.go. This implementation uses atomic Lua scripts to ensure consistency across instances.

The acquireLeaseScript atomically increments a Redis key representing the lease count, sets an expiration timestamp, and returns a status code indicating success or failure. If the current count meets or exceeds the limit, the script returns 0 and the acquisition fails.

// Core logic from backend/internal/infra/runtime/redis/store.go
func (l *ConcurrencyLimiter) Acquire(ctx context.Context, key string, limit int) (func(), bool, error) {
    now := time.Now()
    expiresAt := now.Add(l.store.concurrencyLease)
    
    // Lua script returns 1 if lease granted, 0 otherwise
    result, err := acquireLeaseScript.Run(ctx, l.store.client,
        []string{l.store.key("concurrency", key)},
        now.UnixMilli(), limit, expiresAt.UnixMilli(),
        token, (l.store.concurrencyLease+concurrencyLeaseGrace).Milliseconds()).Int()
    
    if err != nil || result == 0 {
        return nil, false, err
    }
    
    // Release closure runs complementary Lua script to decrement
    return func() { /* Redis decrement logic */ }, true, nil
}

The release function executes a complementary Lua script that decrements the counter, ensuring correctness even when multiple API instances manage the same account pool.

Selector Integration

The account selector in backend/internal/application/gateway/selector.go orchestrates the concurrency control flow. When routing a request, the selector computes the account-specific key using accountConcurrencyKey(value.ID) and attempts to acquire a lease.

// Acquire a lease for an account (inside Selector.claimAccountSlot)
limit := value.MaxConcurrent
if limit <= 0 {
    limit = account.DefaultMaxConcurrent
}
release, acquired, err := s.concurrency.Acquire(ctx,
    accountConcurrencyKey(value.ID), limit)
if err != nil {
    return nil, fmt.Errorf("获取账号并发租约: %w", err)
}
if !acquired {
    // account is at its concurrency limit – try another candidate
    return nil, nil
}
return &accountLease{
    Credential: value,
    release: func() {
        release()          // decrements the counter
        s.announceLeaseReturn()
    },
}, nil

If acquisition fails (acquired == false), the selector skips the saturated account and continues evaluating other candidates. When all accounts are saturated, the selector returns a SelectionUnavailableError with Reason: SelectionSaturated.

To minimize latency, the selector watches the leaseWake channel via awaitLeaseRetry. When any instance releases a lease, announceLeaseReturn broadcasts a wake signal, allowing waiting selectors to retry immediately rather than polling.

Configuration and Defaults

Per-account limits derive from the account.Credential.MaxConcurrent field. If this value is less than or equal to zero, the system falls back to account.DefaultMaxConcurrent as defined in the domain package.

This configuration hierarchy allows operators to set global defaults while overriding specific accounts that handle higher throughput. The selector respects these limits regardless of whether the underlying limiter uses memory or Redis storage.

Summary

• ConcurrencyLimiter interface (backend/internal/repository/runtime.go) abstracts lease acquisition across storage backends.
• In-memory limiter uses sharded maps with sync.Once-protected release functions for single-instance deployments.
• Redis limiter employs atomic Lua scripts to maintain correct counts across distributed API replicas.
• Selector logic attempts multiple candidates, skips saturated accounts, and wakes waiting requests via announceLeaseReturn.
• Configuration uses MaxConcurrent per account, falling back to DefaultMaxConcurrent when unspecified.

Frequently Asked Questions

What happens when an account reaches its concurrency limit?

When an account's active request count equals its MaxConcurrent limit, the Acquire method returns acquired == false. The selector treats this account as unavailable and attempts the next candidate. If all accounts are saturated, the API returns a SelectionUnavailableError with the reason SelectionSaturated, signaling that the upstream pool is fully utilized.

How does Grok2API handle concurrency in distributed deployments?

Distributed deployments use the Redis-backed limiter (backend/internal/infra/runtime/redis/store.go), which stores lease counts in Redis rather than local memory. Atomic Lua scripts ensure that increment and decrement operations remain consistent across multiple API instances, preventing race conditions where two replicas might simultaneously allocate the last available slot.

What is the default concurrency limit if not configured per account?

If account.Credential.MaxConcurrent is less than or equal to zero, the system defaults to account.DefaultMaxConcurrent as defined in the domain package. This allows operators to set a safe global ceiling while selectively increasing limits for high-capacity accounts.

How does the selector know when a slot becomes available?

The selector monitors the leaseWake channel through the awaitLeaseRetry mechanism. When any request completes and invokes lease.Release(), the system calls announceLeaseReturn(), which broadcasts to waiting selectors. This event-driven approach eliminates busy-waiting and reduces latency for pending requests competing for limited slots.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}