perf(redis): bounded Lua VM pool — r1 review fixes

Round-1 review fixes on commit 972e1da:

1. gemini medium @ redis_lua_pool.go get() — centralize binding/return

   Refactor the three-branch select (hit / nil-element fallback /
   empty pool) so the ctxBinding assignment and return happen
   exactly once after the select. Pure restructuring; behaviour
   and metrics are unchanged.

2. gemini medium @ redis_lua_pool.go put() — fast-path drop before reset

   Skip pls.reset() when len(p.idle) already equals maxIdle. The
   reset walks every globals / tables / metatables snapshot and is
   the most expensive part of put(); doing it for a state that
   will immediately be Close()'d in the channel-full branch is
   wasted work the EVAL hot path would feel under sustained
   saturation.

   Race-safety: len() on a buffered channel is an atomically-loaded
   snapshot, so the fast-path can lose a race with a concurrent
   recv (empty channel that we mistakenly skip-and-drop is fine —
   that's equivalent to picking up a tiny drop budget) or win a
   race with a concurrent send (full channel that we correctly
   drop — saves one reset). The existing select-based drop branch
   stays in place as the correctness fallback when the snapshot
   was stale.

   Caller audit (semantic change in put — drop happens before
   reset): only production caller is adapter/redis_lua.go:135
   (`defer luaPool.put(pls)`). The caller does not touch pls
   after put; the externally visible behaviour (put returns, state
   may be reused later) is unchanged. No callers grep'd that
   depend on the reset-then-drop ordering.

3. golangci magic-number @ main.go — derive flag default from
   adapter.DefaultLuaPoolMaxIdle

   Export defaultLuaPoolMaxIdle → DefaultLuaPoolMaxIdle and use
   adapter.DefaultLuaPoolMaxIdle as the --redisLuaMaxIdleStates
   flag default. Single source of truth, removes the mnd lint.

4. golangci gci @ adapter/redis.go, main.go — auto-format

   `golangci-lint fmt` cleanup, no semantic change.

Test:
  go test -race -count=1 -run "TestLua_Pool|TestLua_VMReuse" ./adapter
                                                            -- green.
  golangci-lint run --config=.golangci.yaml ./adapter/ .    -- 0 issues.

Author: bootjp

adapter/redis.go

@@ -171,22 +171,22 @@ var txnApplyHandlers = map[string]txnCommandHandler{
 // single source of truth.
 
 type RedisServer struct {
-	listen              net.Listener
-	store               store.MVCCStore
-	coordinator         kv.Coordinator
-	readTracker         *kv.ActiveTimestampTracker
-	redisTranscoder     *redisTranscoder
-	pubsub              *redisPubSub
-	scriptMu            sync.RWMutex
-	scriptCache         map[string]string
-	luaPool             *luaStatePool
-	luaPoolOnce         sync.Once
+	listen          net.Listener
+	store           store.MVCCStore
+	coordinator     kv.Coordinator
+	readTracker     *kv.ActiveTimestampTracker
+	redisTranscoder *redisTranscoder
+	pubsub          *redisPubSub
+	scriptMu        sync.RWMutex
+	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
+	// pool; getLuaPool falls back to DefaultLuaPoolMaxIdle when the
 	// value is non-positive (covers test fixtures that bypass
 	// NewRedisServer).
-	luaPoolMaxIdle int
+	luaPoolMaxIdle      int
 	traceCommands       bool
 	traceSeq            atomic.Uint64
 	redisAddr           string
@@ -290,7 +290,7 @@ func WithLuaFastPathObserver(observer monitoring.LuaFastPathObserver) RedisServe
 // 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.
+// to DefaultLuaPoolMaxIdle, matching newLuaStatePoolWithMaxIdle.
 //
 // Passing this option overrides the default. The option records the
 // requested cap on the RedisServer; the pool itself is constructed
@@ -424,10 +424,10 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
 		// getLuaPool, which honors luaPoolMaxIdle the same way.
 		luaPool:       nil,
 		traceCommands: os.Getenv("ELASTICKV_REDIS_TRACE") == "1",
-		baseCtx:         baseCtx,
-		baseCancel:      baseCancel,
-		streamWaiters:   newKeyWaiterRegistry(),
-		zsetWaiters:     newKeyWaiterRegistry(),
+		baseCtx:       baseCtx,
+		baseCancel:    baseCancel,
+		streamWaiters: newKeyWaiterRegistry(),
+		zsetWaiters:   newKeyWaiterRegistry(),
 	}
 	r.relay.Bind(r.publishLocal)
 
@@ -443,7 +443,7 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
 	}
 	// Materialize the Lua VM pool after option processing so
 	// WithLuaPoolMaxIdle can choose the cap. newLuaStatePoolWithMaxIdle
-	// clamps non-positive values to defaultLuaPoolMaxIdle, so callers
+	// 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)

adapter/redis_lua_pool.go

@@ -88,7 +88,7 @@ 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
+// 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
@@ -103,7 +103,12 @@ const luaWhitelistedTableHint = 8
 // 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
+//
+// Exported so main.go can use it as the default for the
+// --redisLuaMaxIdleStates flag instead of duplicating the literal
+// (which trips the mnd lint and creates a drift source if the
+// adapter-side default changes).
+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
@@ -227,22 +232,22 @@ func (r *RedisServer) getLuaPool() *luaStatePool {
 	return r.luaPool
 }
 
-// newLuaStatePool returns a bounded pool sized at defaultLuaPoolMaxIdle.
+// 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 newLuaStatePoolWithMaxIdle(defaultLuaPoolMaxIdle)
+	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
+// 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
+		maxIdle = DefaultLuaPoolMaxIdle
 	}
 	return &luaStatePool{
 		idle:    make(chan *pooledLuaState, maxIdle),
@@ -527,37 +532,51 @@ func resetTableContents(tbl *lua.LTable, originalFields map[lua.LValue]lua.LValu
 // pointer write to the state-local ctxBinding userdata -- no lock,
 // no global map.
 //
-// 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.
+// A non-blocking recv that does not match either case fires the
+// default branch (allocation miss). The defensive nil guard
+// preserves behaviour if a future refactor ever sends an unexpected
+// value through the idle channel.
+//
+// The ctxBinding assignment and return are centralized after the
+// select so every path goes through one binding step. This was a
+// gemini r1 review nit: refactor for clarity rather than duplicate
+// the binding in each branch.
 func (p *luaStatePool) get(ctx *luaScriptContext) *pooledLuaState {
+	var pls *pooledLuaState
 	select {
-	case pls := <-p.idle:
-		if pls == nil {
+	case pls = <-p.idle:
+		if pls != nil {
+			p.hits.Add(1)
+		} else {
 			// 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
 	default:
 		p.misses.Add(1)
-		pls := newPooledLuaState()
-		pls.ctxBinding.Value = ctx
-		return pls
+		pls = newPooledLuaState()
 	}
+	pls.ctxBinding.Value = ctx
+	return pls
 }
 
 // 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.
+//
+// Fast-path: when len(p.idle) already equals maxIdle the put is
+// guaranteed to overflow, so skip the (non-trivial) pls.reset()
+// — restoring the globals / tables / metatables snapshots is
+// pointless work for a state that is about to be Close()'d. This is
+// a gemini r1 review optimisation; under saturation the EVAL hot
+// path spends much less CPU on doomed resets. The check is a racy
+// snapshot of len(); concurrent puts can still observe stale
+// "len < max" and race into the select below, where the same
+// channel-full guard catches them — so the fast-path is a strict
+// improvement, not a new correctness requirement.
 func (p *luaStatePool) put(pls *pooledLuaState) {
 	if pls == nil || pls.state == nil {
 		return
@@ -571,14 +590,17 @@ func (p *luaStatePool) put(pls *pooledLuaState) {
 	if pls.state.IsClosed() {
 		return
 	}
+	if len(p.idle) >= p.maxIdle {
+		p.drops.Add(1)
+		pls.state.Close()
+		return
+	}
 	pls.reset()
 	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.
+		// Idle channel filled between the fast-path check and the
+		// select (concurrent puts winning the race). Drop the state.
 		p.drops.Add(1)
 		pls.state.Close()
 	}

adapter/redis_lua_pool_test.go

@@ -689,7 +689,7 @@ func TestLua_PoolBoundedOverflow(t *testing.T) {
 }
 
 // TestLua_PoolBoundedClampsZeroAndNegative locks in the contract that
-// non-positive maxIdle values clamp to defaultLuaPoolMaxIdle rather
+// 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.
@@ -707,8 +707,8 @@ func TestLua_PoolBoundedClampsZeroAndNegative(t *testing.T) {
 		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")
+			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(),

main.go

@@ -80,33 +80,33 @@ func durationToTicks(timeout time.Duration, tick time.Duration, min int) int {
 }
 
 var (
-	myAddr               = flag.String("address", "localhost:50051", "TCP host+port for this node")
-	redisAddr            = flag.String("redisAddress", "localhost:6379", "TCP host+port for redis")
-	dynamoAddr           = flag.String("dynamoAddress", "localhost:8000", "TCP host+port for DynamoDB-compatible API")
-	s3Addr               = flag.String("s3Address", "", "TCP host+port for S3-compatible API; empty to disable")
-	s3Region             = flag.String("s3Region", "us-east-1", "S3 signing region")
-	s3CredsFile          = flag.String("s3CredentialsFile", "", "Path to a JSON file containing static S3 credentials")
-	s3PathStyleOnly      = flag.Bool("s3PathStyleOnly", true, "Only accept path-style S3 requests")
-	sqsAddr              = flag.String("sqsAddress", "", "TCP host+port for SQS-compatible API; empty to disable")
-	sqsRegion            = flag.String("sqsRegion", "us-east-1", "SQS signing region")
-	sqsCredsFile         = flag.String("sqsCredentialsFile", "", "Path to a JSON file containing static SQS credentials")
-	metricsAddr          = flag.String("metricsAddress", "localhost:9090", "TCP host+port for Prometheus metrics")
-	metricsToken         = flag.String("metricsToken", "", "Bearer token for Prometheus metrics; required for non-loopback metricsAddress")
-	pprofAddr            = flag.String("pprofAddress", "localhost:6060", "TCP host+port for pprof debug endpoints; empty to disable")
-	pprofToken           = flag.String("pprofToken", "", "Bearer token for pprof; required for non-loopback pprofAddress")
-	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.")
-	raftGroups           = flag.String("raftGroups", "", "Comma-separated raft groups (groupID=host:port,...)")
-	shardRanges          = flag.String("shardRanges", "", "Comma-separated shard ranges (start:end=groupID,...)")
-	raftRedisMap         = flag.String("raftRedisMap", "", "Map of Raft address to Redis address (raftAddr=redisAddr,...)")
-	raftS3Map            = flag.String("raftS3Map", "", "Map of Raft address to S3 address (raftAddr=s3Addr,...)")
-	raftDynamoMap        = flag.String("raftDynamoMap", "", "Map of Raft address to DynamoDB address (raftAddr=dynamoAddr,...)")
-	raftSqsMap           = flag.String("raftSqsMap", "", "Map of Raft address to SQS address (raftAddr=sqsAddr,...)")
+	myAddr                = flag.String("address", "localhost:50051", "TCP host+port for this node")
+	redisAddr             = flag.String("redisAddress", "localhost:6379", "TCP host+port for redis")
+	dynamoAddr            = flag.String("dynamoAddress", "localhost:8000", "TCP host+port for DynamoDB-compatible API")
+	s3Addr                = flag.String("s3Address", "", "TCP host+port for S3-compatible API; empty to disable")
+	s3Region              = flag.String("s3Region", "us-east-1", "S3 signing region")
+	s3CredsFile           = flag.String("s3CredentialsFile", "", "Path to a JSON file containing static S3 credentials")
+	s3PathStyleOnly       = flag.Bool("s3PathStyleOnly", true, "Only accept path-style S3 requests")
+	sqsAddr               = flag.String("sqsAddress", "", "TCP host+port for SQS-compatible API; empty to disable")
+	sqsRegion             = flag.String("sqsRegion", "us-east-1", "SQS signing region")
+	sqsCredsFile          = flag.String("sqsCredentialsFile", "", "Path to a JSON file containing static SQS credentials")
+	metricsAddr           = flag.String("metricsAddress", "localhost:9090", "TCP host+port for Prometheus metrics")
+	metricsToken          = flag.String("metricsToken", "", "Bearer token for Prometheus metrics; required for non-loopback metricsAddress")
+	pprofAddr             = flag.String("pprofAddress", "localhost:6060", "TCP host+port for pprof debug endpoints; empty to disable")
+	pprofToken            = flag.String("pprofToken", "", "Bearer token for pprof; required for non-loopback pprofAddress")
+	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", adapter.DefaultLuaPoolMaxIdle, "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.")
+	raftGroups            = flag.String("raftGroups", "", "Comma-separated raft groups (groupID=host:port,...)")
+	shardRanges           = flag.String("shardRanges", "", "Comma-separated shard ranges (start:end=groupID,...)")
+	raftRedisMap          = flag.String("raftRedisMap", "", "Map of Raft address to Redis address (raftAddr=redisAddr,...)")
+	raftS3Map             = flag.String("raftS3Map", "", "Map of Raft address to S3 address (raftAddr=s3Addr,...)")
+	raftDynamoMap         = flag.String("raftDynamoMap", "", "Map of Raft address to DynamoDB address (raftAddr=dynamoAddr,...)")
+	raftSqsMap            = flag.String("raftSqsMap", "", "Map of Raft address to SQS address (raftAddr=sqsAddr,...)")
 	// HT-FIFO partition assignment (Phase 3.D §5). Distinct from
 	// --raftSqsMap (which maps raftAddr=sqsAddr for the
 	// proxyToLeader endpoint resolution). The grammar is