perf(redis): bounded Lua VM pool with configurable max-idle cap

Replace the sync.Pool-backed luaStatePool with a buffered-channel
pool sized at a configurable cap. The previous implementation had
two operational issues visible from the production 5-node cluster:

  - No operator-tunable upper bound. sync.Pool's only governor was
    the Go GC; under steady high-rate EVAL load the pool retained
    however many states the GC happened not to clear that cycle,
    so the memory floor moved with allocation rate and GOGC.
  - No saturation signal. There was no way to observe whether the
    pool was being right-sized for the workload — Hits()/Misses()
    showed reuse rate but never said "this many puts were rejected
    because the pool was at capacity."

New pool:

  - Backing: buffered chan *pooledLuaState of capacity maxIdle.
  - Default cap: 64. Empirically each pooled state holds ~200 KiB
    of long-lived heap (base stdlib + redis/cjson/cmsgpack closures
    + per-state snapshot tables); 64 covers redcon's typical
    per-connection EVAL concurrency without retaining a long warm
    tail after bursts.
  - get(): non-blocking recv; on empty channel falls through to a
    fresh allocation (records a Miss). Throughput is therefore not
    capped — only the steady-state idle count is.
  - put(): non-blocking send; on full channel drops the state,
    closes it eagerly, and increments Drops. Drops > 0 is the
    "raise maxIdle" diagnostic.
  - Idle() and MaxIdle() exposed for metrics / admin diagnostics.

Configuration:

  - main.go: new --redisLuaMaxIdleStates int flag (default 64).
  - adapter.WithLuaPoolMaxIdle(n int) RedisServerOption records the
    cap; NewRedisServer materializes the pool AFTER the option
    loop so the cap takes effect. Test fixtures that bypass
    NewRedisServer get the default via getLuaPool's lazy init.
  - Non-positive values clamp to defaultLuaPoolMaxIdle so a
    misconfigured --redisLuaMaxIdleStates=0 yields the default
    rather than a permanently-disabled pool (which would silently
    drop every put).

Caller audit (semantic change — sync.Pool → bounded channel):

  - All in-tree callers of newLuaStatePool / *luaStatePool are in
    adapter/redis*.go and adapter/redis_lua*_test.go. The pool's
    Get/Put surface is unchanged; new accessors (Drops, Idle,
    MaxIdle) are additive.
  - newLuaStatePool() keeps its no-arg signature for test fixtures;
    its behaviour is unchanged (defaults to defaultLuaPoolMaxIdle,
    which is large enough to retain anything the existing test
    suite puts back).
  - The production hot path (adapter/redis_lua.go runLuaScript)
    only calls Get/Put — no behavioural change.
  - Behavioural difference vs sync.Pool that callers MAY observe:
    the bounded pool will not retain states beyond maxIdle. The
    existing TestLua_PoolRecordsReuseVsAllocation tolerates GC-
    driven drops, so it stays green. No test asserted "the pool
    retains states past GC," which would have been incorrect
    under sync.Pool's documented contract anyway.

Tests:

  - All existing TestLua_VMReuse* / TestLua_PoolSerial* /
    TestLua_PoolRecordsReuseVsAllocation continue to pass.
  - New TestLua_PoolBoundedOverflow locks in the cap invariant:
    2*maxIdle puts retain exactly maxIdle, surplus increments
    Drops, draining produces exactly maxIdle hits.
  - New TestLua_PoolBoundedClampsZeroAndNegative locks in the
    non-positive→default clamp so a misconfigured CLI doesn't
    silently disable pooling.

Self-review (CLAUDE.md 5 lenses):

  1. Data loss — N/A. Pool drops only reset/closed *lua.LStates;
     no committed state lives there.
  2. Concurrency — channel ops are O(1) atomic CAS; no new locks
     introduced. Race tests pass.
  3. Performance — channel ops are comparable to sync.Pool's per-P
     slabs; the change does not alter the hot path's allocation
     pattern. Pool saturation now produces eager state.Close() on
     overflow, freeing per-state resources sooner than GC would.
  4. Data consistency — N/A.
  5. Test coverage — two new table-driven tests cover the new
     bounded invariants; all existing pool tests retained.

Test command:
  go test -race -count=1 -short ./adapter   -- 564s, all green.
  go vet ./adapter/                          -- clean.
  go build ./...                              -- clean.

Author: bootjp

adapter/redis.go

@@ -181,6 +181,12 @@ type RedisServer struct {
 	scriptCache         map[string]string
 	luaPool             *luaStatePool
 	luaPoolOnce         sync.Once
+	// luaPoolMaxIdle is the configured cap on idle pooled *lua.LStates.
+	// Set via WithLuaPoolMaxIdle before NewRedisServer materializes the
+	// pool; getLuaPool falls back to defaultLuaPoolMaxIdle when the
+	// value is non-positive (covers test fixtures that bypass
+	// NewRedisServer).
+	luaPoolMaxIdle int
 	traceCommands       bool
 	traceSeq            atomic.Uint64
 	redisAddr           string
@@ -278,6 +284,23 @@ func WithLuaFastPathObserver(observer monitoring.LuaFastPathObserver) RedisServe
 	}
 }
 
+// WithLuaPoolMaxIdle caps the number of idle *lua.LState instances
+// the Lua VM pool retains between EVALs. The cap controls the steady-
+// state memory floor of the pool (maxIdle * per-state footprint —
+// empirically ~200 KiB) without bounding throughput: get() falls
+// through to a fresh allocation when the pool is empty, and put()
+// drops a state to the GC when the pool is full. n <= 0 is clamped
+// to defaultLuaPoolMaxIdle, matching newLuaStatePoolWithMaxIdle.
+//
+// Passing this option overrides the default. The option records the
+// requested cap on the RedisServer; the pool itself is constructed
+// after all options are applied so the recorded cap takes effect.
+func WithLuaPoolMaxIdle(n int) RedisServerOption {
+	return func(r *RedisServer) {
+		r.luaPoolMaxIdle = n
+	}
+}
+
 // luaFastPathCmdZRangeByScore is the shared label for ZRANGEBYSCORE
 // and ZREVRANGEBYSCORE fast-path outcomes. Both directions take the
 // same branch through zsetRangeByScoreFast so sharing one label
@@ -395,8 +418,12 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
 		leaderClients:   make(map[string]*redis.Client),
 		pubsub:          newRedisPubSub(),
 		scriptCache:     map[string]string{},
-		luaPool:         newLuaStatePool(),
-		traceCommands:   os.Getenv("ELASTICKV_REDIS_TRACE") == "1",
+		// luaPool is materialized after the option loop so
+		// WithLuaPoolMaxIdle can influence its sizing. Test fixtures
+		// that bypass NewRedisServer construct the pool lazily via
+		// getLuaPool, which honors luaPoolMaxIdle the same way.
+		luaPool:       nil,
+		traceCommands: os.Getenv("ELASTICKV_REDIS_TRACE") == "1",
 		baseCtx:         baseCtx,
 		baseCancel:      baseCancel,
 		streamWaiters:   newKeyWaiterRegistry(),
@@ -414,6 +441,14 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
 			opt(r)
 		}
 	}
+	// Materialize the Lua VM pool after option processing so
+	// WithLuaPoolMaxIdle can choose the cap. newLuaStatePoolWithMaxIdle
+	// clamps non-positive values to defaultLuaPoolMaxIdle, so callers
+	// that omit the option still get a sensible default. The
+	// luaPoolOnce barrier in getLuaPool keeps test fixtures that build
+	// a bare &RedisServer{} literal (and never call NewRedisServer)
+	// from racing on the same field.
+	r.luaPool = newLuaStatePoolWithMaxIdle(r.luaPoolMaxIdle)
 
 	return r
 }

adapter/redis_lua_pool.go

@@ -1,7 +1,6 @@
 package adapter
 
 import (
-	"sync"
 	"sync/atomic"
 
 	lua "github.com/yuin/gopher-lua"
@@ -89,12 +88,46 @@ const luaWhitelistedTableHint = 8
 // The registry-backed binding is also the reason redis.call is
 // lock-free in the hot path, unlike the first iteration which used
 // a package-level map guarded by sync.RWMutex.
+// defaultLuaPoolMaxIdle is the default upper bound on idle pooled
+// *lua.LState instances retained for reuse. Each pooled state holds
+// the base stdlib + redis/cjson/cmsgpack closures + per-state
+// snapshot tables (globals / tables / metatables); empirically ~200
+// KiB of long-lived heap per state. 64 is sized to comfortably cover
+// typical Redis-side EVAL/EVALSHA concurrency (one in-flight script
+// per connection up to redcon's default worker pool) without
+// retaining a long tail of warm states after a burst subsides.
+//
+// Operators expecting sustained higher concurrency can raise the cap
+// with --redisLuaMaxIdleStates; concurrency that exceeds the cap
+// still works correctly — excess get() calls fall through to a fresh
+// allocation (miss) and excess put() calls drop the state for the GC
+// (drop). The cap therefore controls memory floor, not throughput
+// ceiling.
+const defaultLuaPoolMaxIdle = 64
+
+// luaStatePool pools *lua.LState instances to cut heap/GC pressure on
+// high-rate EVAL / EVALSHA workloads (e.g. BullMQ ~10 scripts/s, where
+// each fresh state allocs ~34% of in-use heap via newFuncContext,
+// newRegistry, newFunctionProto).
+//
+// Internal storage is a buffered channel of capacity maxIdle. We
+// previously used sync.Pool, which GC-clears on every cycle and has
+// no operator-tunable capacity; the bounded channel gives a
+// predictable memory floor (maxIdle * per-state footprint) and an
+// observable "states dropped because the pool was full" counter that
+// makes mis-sizing visible. The channel ops are O(1) atomic CAS on
+// the buffered chan; under high concurrency the contention is
+// comparable to sync.Pool's per-P slabs and well below the cost of
+// a single Lua eval. See TestLua_PoolBoundedOverflow for the
+// invariants.
 type luaStatePool struct {
-	pool sync.Pool
+	idle    chan *pooledLuaState
+	maxIdle int
 
-	// hits / misses are exposed for tests and metrics.
+	// hits / misses / drops are exposed for tests and metrics.
 	hits   atomic.Uint64
 	misses atomic.Uint64
+	drops  atomic.Uint64
 }
 
 // pooledLuaState wraps a *lua.LState plus the immutable snapshot of
@@ -188,23 +221,33 @@ func luaLookupContext(state *lua.LState) (*luaScriptContext, bool) {
 func (r *RedisServer) getLuaPool() *luaStatePool {
 	r.luaPoolOnce.Do(func() {
 		if r.luaPool == nil {
-			r.luaPool = newLuaStatePool()
+			r.luaPool = newLuaStatePoolWithMaxIdle(r.luaPoolMaxIdle)
 		}
 	})
 	return r.luaPool
 }
 
-// newLuaStatePool returns a pool that lazily allocates
-// *pooledLuaState instances on demand. The pool deliberately does NOT
-// set sync.Pool.New: if it did, p.pool.Get() would auto-invoke the
-// constructor on an empty pool and we could not distinguish a fresh
-// allocation from a reused instance. Instead, get() inspects the
-// result of p.pool.Get() -- a nil return signals an empty pool and
-// drives the miss counter plus an explicit newPooledLuaState() call.
-// This keeps the hit/miss metrics honest, which is what the serial
-// reuse tests and the observability counters rely on.
+// newLuaStatePool returns a bounded pool sized at defaultLuaPoolMaxIdle.
+// Used by test fixtures and any caller that does not thread a
+// configured cap through. Production wires the explicit cap via
+// newLuaStatePoolWithMaxIdle from NewRedisServer.
 func newLuaStatePool() *luaStatePool {
-	return &luaStatePool{}
+	return newLuaStatePoolWithMaxIdle(defaultLuaPoolMaxIdle)
+}
+
+// newLuaStatePoolWithMaxIdle returns a pool whose idle backing
+// channel is sized at maxIdle. Non-positive values clamp to
+// defaultLuaPoolMaxIdle to keep get/put semantics well-defined
+// (cap=0 would make every put() drop, which is never what we want
+// — callers asking for "no pool" should bypass the pool entirely).
+func newLuaStatePoolWithMaxIdle(maxIdle int) *luaStatePool {
+	if maxIdle < 1 {
+		maxIdle = defaultLuaPoolMaxIdle
+	}
+	return &luaStatePool{
+		idle:    make(chan *pooledLuaState, maxIdle),
+		maxIdle: maxIdle,
+	}
 }
 
 // newPooledLuaState builds a fresh pooled state: base libs, dangerous
@@ -484,36 +527,37 @@ func resetTableContents(tbl *lua.LTable, originalFields map[lua.LValue]lua.LValu
 // pointer write to the state-local ctxBinding userdata -- no lock,
 // no global map.
 //
-// Because newLuaStatePool does NOT set sync.Pool.New, p.pool.Get()
-// returns nil when the pool is empty; that is the signal for a miss
-// (fresh allocation). A non-nil return is a genuine reuse and counts
-// as a hit. The defensive type-assertion guard preserves behaviour if
-// a future refactor ever puts something unexpected into the pool.
+// A non-blocking channel recv on an empty idle pool returns the
+// zero value (no element), which we treat as a miss. The defensive
+// nil guard preserves behaviour if a future refactor ever sends an
+// unexpected value through the channel.
 func (p *luaStatePool) get(ctx *luaScriptContext) *pooledLuaState {
-	v := p.pool.Get()
-	if v == nil {
-		p.misses.Add(1)
-		pls := newPooledLuaState()
+	select {
+	case pls := <-p.idle:
+		if pls == nil {
+			// Defence in depth: a nil element on the channel is
+			// treated as an allocation miss rather than a panic.
+			p.misses.Add(1)
+			pls = newPooledLuaState()
+			pls.ctxBinding.Value = ctx
+			return pls
+		}
+		p.hits.Add(1)
 		pls.ctxBinding.Value = ctx
 		return pls
-	}
-	pls, ok := v.(*pooledLuaState)
-	if !ok || pls == nil {
-		// Defence in depth: anything other than a *pooledLuaState is
-		// treated as an allocation miss rather than a silent hit.
+	default:
 		p.misses.Add(1)
-		pls = newPooledLuaState()
+		pls := newPooledLuaState()
 		pls.ctxBinding.Value = ctx
 		return pls
 	}
-	p.hits.Add(1)
-	pls.ctxBinding.Value = ctx
-	return pls
 }
 
-// put resets the state and returns it to the pool. If the state is
-// somehow closed (shouldn't happen on the happy path), it is dropped
-// so a dead VM is never handed out again.
+// put resets the state and tries to return it to the pool. If the
+// state is closed (shouldn't happen on the happy path) or the idle
+// channel is full it is dropped so the GC can reclaim it; the drop
+// counter makes pool saturation observable so operators can tune
+// maxIdle.
 func (p *luaStatePool) put(pls *pooledLuaState) {
 	if pls == nil || pls.state == nil {
 		return
@@ -528,14 +572,33 @@ func (p *luaStatePool) put(pls *pooledLuaState) {
 		return
 	}
 	pls.reset()
-	p.pool.Put(pls)
+	select {
+	case p.idle <- pls:
+	default:
+		// Idle channel is full — drop the state. Close it so any
+		// per-state resources are released eagerly; the channel
+		// holds room only for steady-state reuse, not for warm
+		// states left over from a burst peak.
+		p.drops.Add(1)
+		pls.state.Close()
+	}
 }
 
-// Hits / Misses are test hooks. They count Get() outcomes, not
-// allocations proper, but in practice they track allocation avoidance
-// well enough for the "is the pool actually being used?" test.
+// Hits / Misses / Drops are test hooks. They count Get/Put outcomes,
+// not allocations proper, but in practice they track allocation
+// avoidance well enough for the "is the pool actually being used?"
+// test and the "is maxIdle too low for the workload?" diagnostic.
 func (p *luaStatePool) Hits() uint64   { return p.hits.Load() }
 func (p *luaStatePool) Misses() uint64 { return p.misses.Load() }
+func (p *luaStatePool) Drops() uint64  { return p.drops.Load() }
+
+// Idle reports the number of states currently sitting in the pool.
+// Useful for metrics gauges and for tests asserting bounded retention.
+func (p *luaStatePool) Idle() int { return len(p.idle) }
+
+// MaxIdle reports the configured cap. Exposed for diagnostics so
+// /admin can surface "(idle / maxIdle)" to operators.
+func (p *luaStatePool) MaxIdle() int { return p.maxIdle }
 
 // registerPooledRedisModule installs redis.call / redis.pcall /
 // redis.sha1hex / redis.status_reply / redis.error_reply where the

adapter/redis_lua_pool_test.go

@@ -627,3 +627,92 @@ func TestLua_VMReuseClearsContext(t *testing.T) {
 			"pooled LState leaked the original request ctx across put")
 	}
 }
+
+// TestLua_PoolBoundedOverflow is the load-bearing invariant for the
+// channel-backed bounded pool. With sync.Pool the upper bound on
+// retained states was implicit (GC-driven, varies per cycle); the new
+// implementation must:
+//
+//  1. Retain exactly maxIdle put() arrivals; the (maxIdle+1)th arrival
+//     drops cleanly and is reflected in Drops().
+//  2. Continue to serve subsequent get() calls from the retained pool
+//     until the pool is drained.
+//  3. Never panic or leak a half-bound state when a put() overflows.
+func TestLua_PoolBoundedOverflow(t *testing.T) {
+	t.Parallel()
+
+	const maxIdle = 4
+	pool := newLuaStatePoolWithMaxIdle(maxIdle)
+	require.Equal(t, maxIdle, pool.MaxIdle(),
+		"MaxIdle must echo the configured cap so operators can verify it via metrics")
+
+	// Acquire 2*maxIdle distinct states. Because the empty pool
+	// allocates a fresh state on every get(), these are all misses
+	// and all references are independent.
+	const acquired = 2 * maxIdle
+	plsList := make([]*pooledLuaState, 0, acquired)
+	for i := 0; i < acquired; i++ {
+		plsList = append(plsList, pool.get(nil))
+	}
+	require.Equal(t, uint64(0), pool.Hits(), "no put() yet, so no get() can be a hit")
+	require.Equal(t, uint64(acquired), pool.Misses(),
+		"every empty-pool get() must record a miss")
+	require.Equal(t, 0, pool.Idle(), "idle count must be zero before any put()")
+
+	// Release all 2*maxIdle back. Only maxIdle should survive in the
+	// pool; the rest drop and increment the drop counter.
+	for _, pls := range plsList {
+		pool.put(pls)
+	}
+	require.Equal(t, maxIdle, pool.Idle(),
+		"pool must retain exactly maxIdle states after %d puts on a cap=%d pool", acquired, maxIdle)
+	require.Equal(t, uint64(acquired-maxIdle), pool.Drops(),
+		"overflow puts must increment the drop counter so saturation is observable")
+
+	// Draining the pool must produce maxIdle hits and then fall through
+	// to misses, proving the surviving states are usable, not corrupted
+	// by the overflow path.
+	hitsBefore := pool.Hits()
+	for i := 0; i < maxIdle; i++ {
+		drained := pool.get(nil)
+		require.NotNil(t, drained, "drain get must return a usable state")
+	}
+	require.Equal(t, uint64(maxIdle), pool.Hits()-hitsBefore,
+		"draining the pool must produce exactly maxIdle hits")
+	require.Equal(t, 0, pool.Idle(), "pool must be empty after draining maxIdle states")
+
+	// One more get must miss (allocate fresh), confirming the bounded
+	// pool does not silently grow above maxIdle.
+	pool.get(nil)
+	require.Equal(t, uint64(acquired+1), pool.Misses(),
+		"first get() after full drain must record a miss")
+}
+
+// TestLua_PoolBoundedClampsZeroAndNegative locks in the contract that
+// non-positive maxIdle values clamp to defaultLuaPoolMaxIdle rather
+// than silently producing a cap=0 pool (which would make every put()
+// drop). Operators that misconfigure --redisLuaMaxIdleStates=0 get the
+// default behaviour, not a permanently disabled pool.
+func TestLua_PoolBoundedClampsZeroAndNegative(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name    string
+		maxIdle int
+	}{
+		{"zero clamps to default", 0},
+		{"negative clamps to default", -7},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			pool := newLuaStatePoolWithMaxIdle(tc.maxIdle)
+			require.Equal(t, defaultLuaPoolMaxIdle, pool.MaxIdle(),
+				"non-positive maxIdle must clamp to defaultLuaPoolMaxIdle")
+			// And the pool must actually retain on put (proving cap > 0).
+			pool.put(pool.get(nil))
+			require.Equal(t, 1, pool.Idle(),
+				"clamped pool must retain put() arrivals, not silently drop them")
+		})
+	}
+}

main.go

@@ -97,6 +97,7 @@ var (
 	raftId               = flag.String("raftId", "", "Node id used by Raft")
 	raftEngineName       = flag.String("raftEngine", string(raftEngineEtcd), "Raft engine implementation (etcd|hashicorp)")
 	raftDir              = flag.String("raftDataDir", "data/", "Raft data dir")
+	redisLuaMaxIdleStates = flag.Int("redisLuaMaxIdleStates", 64, "Maximum number of idle *lua.LState instances retained by the Redis Lua VM pool. Each state holds ~200 KiB; lower values reduce steady-state memory at the cost of more allocations under burst, higher values absorb bursts at the cost of memory floor. Non-positive values clamp to the default.")
 	raftBootstrap        = flag.Bool("raftBootstrap", false, "Whether to bootstrap the Raft cluster")
 	raftBootstrapMembers = flag.String("raftBootstrapMembers", "", "Comma-separated bootstrap raft members (raftID=host:port,...)")
 	raftJoinAsLearner    = flag.Bool("raftJoinAsLearner", false, "Local node expects to join an existing cluster as a learner; if a post-apply ConfState lists this node as a voter instead, an ERROR-level alarm fires (the node keeps running -- the flag is an operator alarm, not a consensus veto). See docs/design/2026_04_26_proposed_raft_learner.md §4.5.")
@@ -1524,6 +1525,7 @@ func startRedisServer(ctx context.Context, lc *net.ListenConfig, eg *errgroup.Gr
 		adapter.WithLuaObserver(metricsRegistry.LuaObserver()),
 		adapter.WithLuaFastPathObserver(metricsRegistry.LuaFastPathObserver()),
 		adapter.WithRedisCompactor(deltaCompactor),
+		adapter.WithLuaPoolMaxIdle(*redisLuaMaxIdleStates),
 	)
 	eg.Go(func() error {
 		defer redisServer.Stop()