adapter/redis: apply valid review suggestions while preserving correct pubsub

- Make hset, hincrby, BZPOPMIN atomic via retryRedisWrite pattern
- Replace reflect/unsafe pubsub introspection with self-contained redisPubSub
- Pass startTS through dispatchElems for MVCC consistency in retry paths
- Guard snapshotTS MaxUint64 sentinel in dispatchElems
- Replace dynamoSelectCount with redisKeywordCount in xread/xrange
- Fix raftstore pebble error wrapping for ErrLogNotFound
- Document known implementation issues in REDIS_KNOWN_ISSUES.md

Author: bootjp

adapter/REDIS_KNOWN_ISSUES.md

@@ -0,0 +1,71 @@
+# Redis Adapter: Known Implementation Issues
+
+This document tracks architectural limitations identified during PR #350 review
+that require larger refactoring to address.
+
+## Performance
+
+### LPUSH / list mutations are O(N)
+- **Location**: `redis_compat_commands.go` (lpush, ltrim, rpush)
+- **Problem**: List modification commands read the entire list, modify in memory,
+  and rewrite the whole list. This is O(N) where N is the number of elements.
+- **Fix**: Push list manipulation down into the storage layer so individual
+  elements can be appended/removed without full rewrite.
+
+### SCAN materializes all keys
+- **Location**: `redis_compat_commands.go` (scan)
+- **Problem**: SCAN builds and sorts the full set of visible keys on every call,
+  using the cursor as an array index. This is O(N) per call and diverges from
+  Redis's incremental cursor semantics.
+- **Fix**: Implement SCAN as an incremental range scan over the underlying store,
+  returning the next cursor based on the last returned key.
+
+### FLUSHDB iterates all keys
+- **Location**: `redis_compat_commands.go` (flushdb)
+- **Problem**: FLUSHDB/FLUSHALL iterates through all visible keys and deletes
+  them one by one. This can be very slow for large datasets.
+- **Fix**: Implement a range deletion capability at the storage layer.
+
+### visibleKeys N x M per-key type checks
+- **Location**: `redis_compat_helpers.go` (visibleKeys)
+- **Problem**: `visibleKeys` calls `logicalExistsAt` per key, which performs
+  multiple `ExistsAt` checks across namespaces (list, hash, set, zset, stream,
+  ttl, etc.). For KEYS, DBSIZE, and SCAN this becomes O(N*M).
+- **Fix**: Filter internal namespaces during scanning and derive logical
+  existence in a single pass.
+
+### localKeysPattern redundant scans
+- **Location**: `redis.go` (localKeysPattern)
+- **Problem**: `localKeysPattern` performs additional scans over list and Redis
+  internal namespaces even when `start/end` are nil (full keyspace). This causes
+  multiple redundant full scans per call.
+- **Fix**: Only do extra namespace scans when the pattern has a bounded user
+  prefix, or skip when `start == nil && end == nil`.
+
+### BZPOPMIN / XREAD busy-polling
+- **Location**: `redis_compat_commands.go` (bzpopmin, xread)
+- **Problem**: Both use `time.Sleep` polling loops to wait for data. This wastes
+  CPU cycles when there are no new elements.
+- **Fix**: Implement a wait/notify pattern where ZADD/XADD signal waiting
+  BZPOPMIN/XREAD goroutines via condition variables or channels.
+
+## Consistency
+
+### EXISTS / GET follower reads
+- **Location**: `redis.go` (exists, get)
+- **Problem**: EXISTS checks local MVCC state without leader proxying, which can
+  return stale results on followers. GET's pre-check via `keyType()` also uses
+  local state, potentially returning false negatives when the follower lags.
+- **Fix**: Proxy EXISTS reads to the leader (or verify leader status), and
+  perform the GET type check via the leader/proxy path as well.
+
+## Design
+
+### PFCOUNT is exact cardinality, not HyperLogLog
+- **Location**: `redis_compat_commands.go` (pfadd, pfcount)
+- **Problem**: PFADD/PFCOUNT stores all unique elements in a set and returns its
+  exact size, rather than using a probabilistic HyperLogLog structure.
+  For large datasets this consumes significantly more memory than a real HLL.
+- **Note**: For Misskey's use case the exact implementation is acceptable. Only
+  relevant if elastickv is used as a general-purpose Redis replacement with
+  large cardinality sets.

adapter/redis.go

@@ -243,9 +243,7 @@ type RedisServer struct {
 	store           store.MVCCStore
 	coordinator     kv.Coordinator
 	redisTranscoder *redisTranscoder
-	pubsub          redcon.PubSub
-	pubsubMu        sync.RWMutex
-	pubsubChannels  map[string]int
+	pubsub          *redisPubSub
 	scriptMu        sync.RWMutex
 	scriptCache     map[string]string
 	traceCommands   bool
@@ -260,9 +258,8 @@ type RedisServer struct {
 }
 
 type connState struct {
-	inTxn         bool
-	queue         []redcon.Command
-	subscriptions map[string]struct{}
+	inTxn bool
+	queue []redcon.Command
 }
 
 type resultType int
@@ -297,7 +294,7 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
 		redisAddr:       redisAddr,
 		relay:           relay,
 		leaderRedis:     leaderRedis,
-		pubsubChannels:  map[string]int{},
+		pubsub:          newRedisPubSub(),
 		scriptCache:     map[string]string{},
 		traceCommands:   os.Getenv("ELASTICKV_REDIS_TRACE") == "1",
 	}
@@ -433,17 +430,18 @@ func (r *RedisServer) Run() error {
 
 			traceID, traceStart := r.traceCommandStart(conn, name, cmd.Args[1:])
 			f(conn, cmd)
-			r.traceCommandFinish(traceID, conn, name, time.Since(traceStart))
+			if r.traceCommands {
+				r.traceCommandFinish(traceID, conn, name, time.Since(traceStart))
+			}
 		},
 		func(conn redcon.Conn) bool {
 			// Use this function to accept or deny the connection.
 			// log.Printf("accept: %s", conn.RemoteAddr())
 			return true
 		},
 		func(conn redcon.Conn, err error) {
-			// This is called when the connection has been closed
-			// log.Printf("closed: %s, err: %v", conn.RemoteAddr(), err)
-			r.cleanupConnSubscriptions(conn)
+			// This is called when the connection has been closed.
+			// PubSub connections clean up their own subscriptions via bgrunner.
 		})
 
 	return errors.WithStack(err)
@@ -609,7 +607,7 @@ func (r *RedisServer) replaceWithString(ctx context.Context, key, value []byte,
 	} else {
 		elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Del, Key: redisTTLKey(key)})
 	}
-	return r.dispatchElems(ctx, true, elems)
+	return r.dispatchElems(ctx, true, 0, elems)
 }
 
 func (r *RedisServer) executeSet(ctx context.Context, key, value []byte, opts redisSetOptions) (redisSetExecution, error) {
@@ -632,40 +630,6 @@ func (r *RedisServer) executeSet(ctx context.Context, key, value []byte, opts re
 	return redisSetExecution{state: state, wroteOldBulk: opts.returnOld}, nil
 }
 
-func (r *RedisServer) trackSubscription(conn redcon.Conn, channel string) {
-	state := getConnState(conn)
-	if state.subscriptions == nil {
-		state.subscriptions = map[string]struct{}{}
-	}
-	if _, ok := state.subscriptions[channel]; ok {
-		return
-	}
-	state.subscriptions[channel] = struct{}{}
-
-	r.pubsubMu.Lock()
-	defer r.pubsubMu.Unlock()
-	r.pubsubChannels[channel]++
-}
-
-func (r *RedisServer) cleanupConnSubscriptions(conn redcon.Conn) {
-	ctx := conn.Context()
-	state, ok := ctx.(*connState)
-	if !ok || len(state.subscriptions) == 0 {
-		return
-	}
-
-	r.pubsubMu.Lock()
-	defer r.pubsubMu.Unlock()
-	for channel := range state.subscriptions {
-		if n := r.pubsubChannels[channel]; n <= 1 {
-			delete(r.pubsubChannels, channel)
-		} else {
-			r.pubsubChannels[channel] = n - 1
-		}
-	}
-	state.subscriptions = nil
-}
-
 func (r *RedisServer) Stop() {
 	_ = r.relayConnCache.Close()
 	_ = r.listen.Close()
@@ -838,7 +802,7 @@ func (r *RedisServer) del(conn redcon.Conn, cmd redcon.Command) {
 			}
 			elems = append(elems, keyElems...)
 		}
-		if err := r.dispatchElems(ctx, true, elems); err != nil {
+		if err := r.dispatchElems(ctx, true, readTS, elems); err != nil {
 			return err
 		}
 		removed = nextRemoved
@@ -1471,12 +1435,31 @@ func (t *txnContext) applyExpire(cmd redcon.Command, unit time.Duration) (redisR
 		return redisResult{typ: resultInt, integer: 0}, nil
 	}
 
+	if ttl <= 0 {
+		return t.stageKeyDeletion(cmd.Args[1])
+	}
+
 	expireAt := time.Now().Add(time.Duration(ttl) * unit)
 	state.value = &expireAt
 	state.dirty = true
 	return redisResult{typ: resultInt, integer: 1}, nil
 }
 
+func (t *txnContext) stageKeyDeletion(key []byte) (redisResult, error) {
+	st, err := t.loadListState(key)
+	if err != nil {
+		return redisResult{}, err
+	}
+	stageListDelete(st)
+	tv, err := t.load(key)
+	if err != nil {
+		return redisResult{}, err
+	}
+	tv.deleted = true
+	tv.dirty = true
+	return redisResult{typ: resultInt, integer: 1}, nil
+}
+
 func parseRangeBounds(startRaw, endRaw []byte, total int) (int, int, error) {
 	start, err := parseInt(startRaw)
 	if err != nil {