Fix review issues: Lua ZADD flags, graceful shutdown, expire overflow, key deletion

- Lua ZADD: parse NX/XX/GT/LT flags using shared zaddFlags helpers
- TransactionManager: add Close() with closeCh for graceful shutdown
  of flushRawPending goroutine; replace time.Sleep with select
- Expire: add overflow guard for time.Duration conversion of large TTLs
- stageKeyDeletion/applyDel: delete hash, set, stream, HLL internal
  keys in addition to list, string, TTL, and zset
- removeAll: use clear(sc.chans) instead of nil to avoid map access
  after nil assignment

Author: bootjp

adapter/redis.go

@@ -553,10 +553,15 @@ func parseRedisSetTTL(args [][]byte, index int, opt string, now time.Time) (*tim
 		return nil, index, errors.New("ERR invalid expire time in 'set' command")
 	}
 
-	expireAt := now.Add(time.Duration(n) * time.Millisecond)
+	unit := time.Millisecond
 	if opt == "EX" {
-		expireAt = now.Add(time.Duration(n) * time.Second)
+		unit = time.Second
 	}
+	if n > math.MaxInt64/int64(unit) {
+		return nil, index, errors.New("ERR invalid expire time in 'set' command")
+	}
+
+	expireAt := now.Add(time.Duration(n) * unit)
 	return &expireAt, index + 1, nil
 }
 
@@ -1330,21 +1335,7 @@ func (t *txnContext) applySet(cmd redcon.Command) (redisResult, error) {
 }
 
 func (t *txnContext) applyDel(cmd redcon.Command) (redisResult, error) {
-	// Always stage list deletion so DEL followed by RPUSH in the same MULTI
-	// reliably recreates the list from an empty state.
-	st, err := t.loadListState(cmd.Args[1])
-	if err != nil {
-		return redisResult{}, err
-	}
-	stageListDelete(st)
-
-	tv, err := t.load(cmd.Args[1])
-	if err != nil {
-		return redisResult{}, err
-	}
-	tv.deleted = true
-	tv.dirty = true
-	return redisResult{typ: resultInt, integer: 1}, nil
+	return t.stageKeyDeletion(cmd.Args[1])
 }
 
 func (t *txnContext) applyGet(cmd redcon.Command) (redisResult, error) {
@@ -1514,6 +1505,20 @@ func (t *txnContext) stageKeyDeletion(key []byte) (redisResult, error) {
 	}
 	zs.members = nil
 	zs.dirty = true
+	// Mark hash, set, stream, and HLL internal keys for deletion.
+	for _, internalKey := range [][]byte{
+		redisHashKey(key),
+		redisSetKey(key),
+		redisStreamKey(key),
+		redisHLLKey(key),
+	} {
+		iv, err := t.load(internalKey)
+		if err != nil {
+			return redisResult{}, err
+		}
+		iv.deleted = true
+		iv.dirty = true
+	}
 	return redisResult{typ: resultInt, integer: 1}, nil
 }
 

adapter/redis_lua_context.go

@@ -1654,9 +1654,29 @@ func (c *luaScriptContext) cmdSIsMember(args []string) (luaReply, error) {
 	return luaIntReply(0), nil
 }
 
+func parseLuaZAddArgs(args []string) (zaddFlags, []string, error) {
+	var flags zaddFlags
+	i := 1
+	for i < len(args) {
+		if !flags.applyFlag(strings.ToUpper(args[i])) {
+			break
+		}
+		i++
+	}
+	if err := flags.validate(); err != nil {
+		return zaddFlags{}, nil, err
+	}
+	remaining := args[i:]
+	if len(remaining)%2 != 0 || len(remaining) == 0 {
+		return zaddFlags{}, nil, errors.New("ERR syntax error")
+	}
+	return flags, remaining, nil
+}
+
 func (c *luaScriptContext) cmdZAdd(args []string) (luaReply, error) {
-	if (len(args)-1)%2 != 0 {
-		return luaReply{}, errors.New("ERR syntax error")
+	flags, pairs, err := parseLuaZAddArgs(args)
+	if err != nil {
+		return luaReply{}, err
 	}
 	st, err := c.zsetState([]byte(args[0]))
 	if err != nil {
@@ -1667,15 +1687,20 @@ func (c *luaScriptContext) cmdZAdd(args []string) (luaReply, error) {
 		st.members = map[string]float64{}
 	}
 	added := 0
-	for i := 1; i < len(args); i += 2 {
-		score, err := strconv.ParseFloat(args[i], 64)
+	for j := 0; j < len(pairs); j += 2 {
+		score, err := strconv.ParseFloat(pairs[j], 64)
 		if err != nil {
 			return luaReply{}, errors.WithStack(err)
 		}
-		if _, ok := st.members[args[i+1]]; !ok {
+		member := pairs[j+1]
+		oldScore, exists := st.members[member]
+		if !flags.allows(exists, oldScore, score) {
+			continue
+		}
+		if !exists {
 			added++
 		}
-		st.members[args[i+1]] = score
+		st.members[member] = score
 	}
 	st.loaded = true
 	st.dirty = true

adapter/redis_pubsub.go

@@ -145,7 +145,7 @@ func (ps *redisPubSub) removeAll(sc *pubsubConn) {
 			}
 		}
 	}
-	sc.chans = nil
+	clear(sc.chans)
 }
 
 func (ps *redisPubSub) writeSubscribeReply(sc *pubsubConn, channel string, count int) {

kv/transaction.go

@@ -16,6 +16,7 @@ type TransactionManager struct {
 	mu          sync.Mutex
 	rawPending  []*rawCommitItem
 	rawFlushing bool
+	closeCh     chan struct{}
 }
 
 type rawCommitItem struct {
@@ -34,7 +35,26 @@ var rawBatchWindow = 500 * time.Microsecond
 
 func NewTransaction(raft *raft.Raft) *TransactionManager {
 	return &TransactionManager{
-		raft: raft,
+		raft:    raft,
+		closeCh: make(chan struct{}),
+	}
+}
+
+var errShuttingDown = errors.New("transaction manager is shutting down")
+
+// Close signals the TransactionManager to stop and drains any pending raw
+// commit items, sending each an error so callers are not blocked forever.
+func (t *TransactionManager) Close() {
+	close(t.closeCh)
+
+	t.mu.Lock()
+	pending := t.rawPending
+	t.rawPending = nil
+	t.rawFlushing = false
+	t.mu.Unlock()
+
+	for _, item := range pending {
+		item.done <- rawCommitResult{err: errShuttingDown}
 	}
 }
 
@@ -175,9 +195,21 @@ func (t *TransactionManager) commitRaw(reqs []*pb.Request) (*TransactionResponse
 }
 
 func (t *TransactionManager) flushRawPending() {
-	time.Sleep(rawBatchWindow)
+	timer := time.NewTimer(rawBatchWindow)
+	select {
+	case <-timer.C:
+	case <-t.closeCh:
+		timer.Stop()
+		return
+	}
 
 	for {
+		select {
+		case <-t.closeCh:
+			return
+		default:
+		}
+
 		batch := t.takeRawBatch()
 		if len(batch) == 0 {
 			t.mu.Lock()