Merge pull request #486 from bootjp/feat/redis-str-prefix

feat: add !redis|str| prefix to Redis string keys and scope FLUSHALL

Author: bootjp

adapter/redis.go

@@ -39,6 +39,7 @@ const (
 	cmdExpire           = "EXPIRE"
 	cmdFlushAll         = "FLUSHALL"
 	cmdFlushDB          = "FLUSHDB"
+	cmdFlushLegacy      = "FLUSHLEGACY"
 	cmdGet              = "GET"
 	cmdGetDel           = "GETDEL"
 	cmdHDel             = "HDEL"
@@ -113,6 +114,7 @@ const (
 const (
 	redisLatestCommitTimeout = 5 * time.Second
 	redisDispatchTimeout     = 10 * time.Second
+	redisFlushLegacyTimeout  = 10 * time.Minute
 	redisRelayPublishTimeout = 2 * time.Second
 	redisTraceArgLimit       = 6
 	redisTraceArgMaxLen      = 96
@@ -170,6 +172,7 @@ var argsLen = map[string]int{
 	cmdExpire:           -3,
 	cmdFlushAll:         1,
 	cmdFlushDB:          1,
+	cmdFlushLegacy:      1,
 	cmdGet:              2,
 	cmdGetDel:           2,
 	cmdHDel:             -3,
@@ -358,6 +361,7 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
 		cmdExpire:           r.expire,
 		cmdFlushAll:         r.flushall,
 		cmdFlushDB:          r.flushdb,
+		cmdFlushLegacy:      r.flushlegacy,
 		cmdGet:              r.get,
 		cmdGetDel:           r.getdel,
 		cmdHDel:             r.hdel,
@@ -692,7 +696,7 @@ func (r *RedisServer) loadRedisSetState(key []byte, readTS uint64, returnOld boo
 		return state, nil
 	}
 
-	oldValue, err := r.readValueAt(key, readTS)
+	oldValue, err := r.readRedisStringAt(key, readTS)
 	if err != nil && !errors.Is(err, store.ErrKeyNotFound) {
 		return redisSetState{}, err
 	}
@@ -709,7 +713,7 @@ func (r *RedisServer) replaceWithStringTxn(ctx context.Context, key, value []byt
 		}
 		elems = append(elems, delElems...)
 	}
-	elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Put, Key: key, Value: bytes.Clone(value)})
+	elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Put, Key: redisStrKey(key), Value: bytes.Clone(value)})
 	if ttl != nil {
 		elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Put, Key: redisTTLKey(key), Value: encodeRedisTTL(*ttl)})
 	} else {
@@ -955,7 +959,7 @@ func (r *RedisServer) get(conn redcon.Conn, cmd redcon.Command) {
 		return
 	}
 
-	v, err := r.readValueAt(key, readTS)
+	v, err := r.readRedisStringAt(key, readTS)
 	if err != nil {
 		if errors.Is(err, store.ErrKeyNotFound) {
 			conn.WriteNull()
@@ -1423,28 +1427,51 @@ func (t *txnContext) trackTypeReadKeys(key []byte) {
 		redisZSetKey(key),
 		redisStreamKey(key),
 		redisHLLKey(key),
-		key,
+		redisStrKey(key),
+		key, // legacy bare key for fallback reads
 		redisTTLKey(key),
 	} {
 		t.trackReadKey(readKey)
 	}
 }
 
 func (t *txnContext) load(key []byte) (*txnValue, error) {
-	k := string(key)
+	// If the key is already an internal key (e.g., !redis|hash|...,
+	// !lst|..., !txn|..., !ddb|..., !s3|..., !dist|...), use it as-is.
+	// Otherwise, it's a bare user key for a string value — prefix it.
+	storageKey := key
+	userKey := extractRedisInternalUserKey(key)
+	if !isKnownInternalKey(key) {
+		storageKey = redisStrKey(key)
+		userKey = key
+	}
+	k := string(storageKey)
 	if tv, ok := t.working[k]; ok {
 		return tv, nil
 	}
-	t.trackReadKey(key)
-	if userKey := extractRedisInternalUserKey(key); userKey != nil {
+	t.trackReadKey(storageKey)
+	if !isKnownInternalKey(key) {
+		// Track the bare key too for conflict detection on legacy fallback reads.
+		t.trackReadKey(key)
+	}
+	if userKey != nil {
 		t.trackReadKey(redisTTLKey(userKey))
-	} else if !bytes.HasPrefix(key, []byte(redisTTLPrefix)) {
-		t.trackReadKey(redisTTLKey(key))
 	}
 	tv := &txnValue{}
-	val, err := t.server.readValueAt(key, t.startTS)
-	if err != nil && !errors.Is(err, store.ErrKeyNotFound) {
-		return nil, errors.WithStack(err)
+	var val []byte
+	if !isKnownInternalKey(key) {
+		// For bare user string keys, use the fallback-aware reader.
+		var err error
+		val, err = t.server.readRedisStringAt(key, t.startTS)
+		if err != nil && !errors.Is(err, store.ErrKeyNotFound) {
+			return nil, errors.WithStack(err)
+		}
+	} else {
+		var err error
+		val, err = t.server.readValueAt(storageKey, t.startTS)
+		if err != nil && !errors.Is(err, store.ErrKeyNotFound) {
+			return nil, errors.WithStack(err)
+		}
 	}
 	tv.raw = val
 	tv.loaded = true
@@ -1565,7 +1592,7 @@ func (t *txnContext) stagedListType(key string) (redisValueType, bool) {
 }
 
 func (t *txnContext) stagedStringType(key string) (redisValueType, bool) {
-	tv, ok := t.working[key]
+	tv, ok := t.working[string(redisStrKey([]byte(key)))]
 	if !ok {
 		return redisTypeNone, false
 	}
@@ -1811,6 +1838,16 @@ func (t *txnContext) stageKeyDeletion(key []byte) (redisResult, error) {
 		iv.deleted = true
 		iv.dirty = true
 	}
+	// Mark legacy bare string key for deletion. We bypass load() here
+	// because load() auto-prefixes bare keys to !redis|str|.
+	// Track the bare key in the read set for conflict detection.
+	t.trackReadKey(key)
+	bareK := string(key)
+	if _, ok := t.working[bareK]; !ok {
+		t.working[bareK] = &txnValue{}
+	}
+	t.working[bareK].deleted = true
+	t.working[bareK].dirty = true
 	return redisResult{typ: resultInt, integer: 1}, nil
 }
 
@@ -1908,12 +1945,12 @@ func (t *txnContext) buildKeyElems() []*kv.Elem[kv.OP] {
 		if !tv.dirty {
 			continue
 		}
-		key := []byte(k)
+		storageKey := []byte(k)
 		if tv.deleted {
-			elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Del, Key: key})
+			elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Del, Key: storageKey})
 			continue
 		}
-		elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Put, Key: key, Value: tv.raw})
+		elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Put, Key: storageKey, Value: tv.raw})
 	}
 	return elems
 }

adapter/redis_compat_commands.go

@@ -112,7 +112,7 @@ func (r *RedisServer) getdel(conn redcon.Conn, cmd redcon.Command) {
 		if typ != redisTypeString {
 			return wrongTypeError()
 		}
-		raw, err := r.readValueAt(key, readTS)
+		raw, err := r.readRedisStringAt(key, readTS)
 		if err != nil {
 			// Key may have expired or been deleted between type check and read.
 			v = nil
@@ -458,6 +458,76 @@ func (r *RedisServer) flushall(conn redcon.Conn, _ redcon.Command) {
 	r.flushDatabase(conn, true)
 }
 
+// deleteLegacyKeys scans the full keyspace and deletes keys that do not belong
+// to any known internal prefix. Returns the number of user-visible legacy keys
+// deleted. TTL keys are intentionally NOT deleted because the !redis|ttl|
+// namespace is shared across all Redis types — deleting them could strip
+// expiration from already-migrated or newly-created keys.
+func (r *RedisServer) deleteLegacyKeys(ctx context.Context, readTS uint64) (int, error) {
+	const batchSize = 1000
+	var totalDeleted int
+	cursor := make([]byte, 0, batchSize)
+	for {
+		kvs, err := r.store.ScanAt(ctx, cursor, nil, batchSize, readTS)
+		if err != nil {
+			return totalDeleted, fmt.Errorf("scan: %w", err)
+		}
+		if len(kvs) == 0 {
+			break
+		}
+
+		elems := make([]*kv.Elem[kv.OP], 0, len(kvs))
+		legacyCount := 0
+		for _, pair := range kvs {
+			if !isKnownInternalKey(pair.Key) {
+				legacyCount++
+				elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Del, Key: pair.Key})
+			}
+		}
+
+		if len(elems) > 0 {
+			if err := r.dispatchElems(ctx, false, readTS, elems); err != nil {
+				return totalDeleted, err
+			}
+			totalDeleted += legacyCount
+		}
+
+		// Advance cursor past the last key in this batch.
+		lastKey := kvs[len(kvs)-1].Key
+		cursor = make([]byte, len(lastKey)+1)
+		copy(cursor, lastKey)
+
+		// Yield briefly between batches to avoid saturating the Raft log.
+		time.Sleep(time.Millisecond)
+	}
+	return totalDeleted, nil
+}
+
+// flushlegacy deletes old unprefixed Redis string keys that were written before
+// the !redis|str| prefix migration. It scans all keys and deletes those that
+// do not match any known internal prefix. This is a one-time migration operation.
+func (r *RedisServer) flushlegacy(conn redcon.Conn, _ redcon.Command) {
+	if !r.coordinator.IsLeader() {
+		n, err := r.proxyFlushLegacy()
+		if err != nil {
+			conn.WriteError(err.Error())
+			return
+		}
+		conn.WriteInt(n)
+		return
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), redisFlushLegacyTimeout)
+	defer cancel()
+
+	totalDeleted, err := r.deleteLegacyKeys(ctx, r.readTS())
+	if err != nil {
+		conn.WriteError(err.Error())
+		return
+	}
+	conn.WriteInt(totalDeleted)
+}
+
 func (r *RedisServer) flushDatabase(conn redcon.Conn, all bool) {
 	if !r.coordinator.IsLeader() {
 		if err := r.proxyFlushDatabase(all); err != nil {
@@ -476,12 +546,17 @@ func (r *RedisServer) flushDatabase(conn redcon.Conn, all bool) {
 			return fmt.Errorf("verify leader: %w", err)
 		}
 
-		// Dispatch a single DEL_PREFIX operation. The FSM on each node will
-		// scan locally and write tombstones, avoiding the need to enumerate
-		// all keys here and send them through the Raft log.
+		// Delete only Redis-related keys. Three DEL_PREFIX operations cover
+		// all Redis namespaces: "!redis|" (str, hash, set, zset, hll,
+		// stream, ttl), "!lst|" (list meta + items), and "!zs|" (zset
+		// wide-column meta/member/score).
+		// Legacy bare keys are NOT deleted here to avoid a full keyspace
+		// scan. Run FLUSHLEGACY first to clean up legacy data.
 		_, err := r.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{
 			Elems: []*kv.Elem[kv.OP]{
-				{Op: kv.DelPrefix, Key: nil},
+				{Op: kv.DelPrefix, Key: []byte("!redis|")},
+				{Op: kv.DelPrefix, Key: []byte("!lst|")},
+				{Op: kv.DelPrefix, Key: []byte("!zs|")},
 			},
 		})
 		if err != nil {
@@ -1169,7 +1244,7 @@ func (r *RedisServer) incr(conn redcon.Conn, cmd redcon.Command) {
 
 		current = 0
 		if typ == redisTypeString {
-			raw, err := r.readValueAt(cmd.Args[1], readTS)
+			raw, err := r.readRedisStringAt(cmd.Args[1], readTS)
 			if err != nil {
 				return err
 			}
@@ -1181,7 +1256,7 @@ func (r *RedisServer) incr(conn redcon.Conn, cmd redcon.Command) {
 		current++
 
 		return r.dispatchElems(ctx, true, readTS, []*kv.Elem[kv.OP]{
-			{Op: kv.Put, Key: cmd.Args[1], Value: []byte(strconv.FormatInt(current, 10))},
+			{Op: kv.Put, Key: redisStrKey(cmd.Args[1]), Value: []byte(strconv.FormatInt(current, 10))},
 			{Op: kv.Del, Key: redisTTLKey(cmd.Args[1])},
 		})
 	}); err != nil {

adapter/redis_compat_helpers.go

@@ -29,6 +29,9 @@ func (r *RedisServer) rawKeyTypeAt(ctx context.Context, key []byte, readTS uint6
 		{typ: redisTypeStream, key: redisStreamKey(key)},
 		// HyperLogLog is a Redis string subtype. Treat it as "string" for TYPE.
 		{typ: redisTypeString, key: redisHLLKey(key)},
+		{typ: redisTypeString, key: redisStrKey(key)},
+		// Fallback: check bare key for legacy data written before the
+		// !redis|str| prefix migration.
 		{typ: redisTypeString, key: key},
 	}
 	for _, check := range checks {
@@ -137,9 +140,23 @@ func (r *RedisServer) dispatchElems(ctx context.Context, isTxn bool, startTS uin
 	return errors.WithStack(err)
 }
 
+// readRedisStringAt reads a Redis string value, trying the prefixed key first
+// and falling back to the bare key for legacy data written before the
+// !redis|str| prefix migration.
+func (r *RedisServer) readRedisStringAt(key []byte, readTS uint64) ([]byte, error) {
+	v, err := r.readValueAt(redisStrKey(key), readTS)
+	if err == nil {
+		return v, nil
+	}
+	if !errors.Is(err, store.ErrKeyNotFound) {
+		return nil, err
+	}
+	return r.readValueAt(key, readTS)
+}
+
 func (r *RedisServer) saveString(ctx context.Context, key []byte, value []byte, ttl *time.Time) error {
 	elems := []*kv.Elem[kv.OP]{
-		{Op: kv.Put, Key: key, Value: bytes.Clone(value)},
+		{Op: kv.Put, Key: redisStrKey(key), Value: bytes.Clone(value)},
 	}
 	if ttl == nil {
 		elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Del, Key: redisTTLKey(key)})
@@ -157,15 +174,9 @@ func (r *RedisServer) deleteLogicalKeyElems(ctx context.Context, key []byte, rea
 
 	elems := []*kv.Elem[kv.OP]{}
 
-	stringExists, err := r.store.ExistsAt(ctx, key, readTS)
-	if err != nil {
-		return nil, false, errors.WithStack(err)
-	}
-	if stringExists {
-		elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Del, Key: key})
-	}
-
 	for _, internalKey := range [][]byte{
+		redisStrKey(key),
+		key, // legacy bare string key
 		redisHashKey(key),
 		redisSetKey(key),
 		redisHLLKey(key),