Skip to content

Commit 5c3b2a4

Browse files
bootjpbootjp
committed
adapter/redis: add SETEX, GETDEL, SETNX commands and fix Lua cmsgpack nil handling
Add three Redis commands required by Misskey's WebAuthn, distributed lock, and cache subsystems: - SETEX: SET with EX in a single command - GETDEL: atomic GET + DELETE - SETNX: SET if not exists (integer reply) All three are registered in the command router, argsLen table, and Lua script context. Fix a critical bug in cmsgpack.unpack where Go nil (from msgpack nil) was converted to Lua false instead of Lua nil. This caused BullMQ's addJob Lua script to misinterpret absent parentKey as a present-but-missing key, returning error code -5. Real Redis cmsgpack maps msgpack nil to Lua nil. Verified: all 30 Misskey e2e test suites (1300+ tests) pass with the same results as real Redis.
1 parent 12b1606 commit 5c3b2a4

5 files changed

Lines changed: 235 additions & 3 deletions

File tree

‎adapter/redis.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@ const (
4040
cmdFlushAll = "FLUSHALL"
4141
cmdFlushDB = "FLUSHDB"
4242
cmdGet = "GET"
43+
cmdGetDel = "GETDEL"
4344
cmdHDel = "HDEL"
4445
cmdHExists = "HEXISTS"
4546
cmdHGet = "HGET"
@@ -79,6 +80,8 @@ const (
7980
cmdScan = "SCAN"
8081
cmdSelect = "SELECT"
8182
cmdSet = "SET"
83+
cmdSetEx = "SETEX"
84+
cmdSetNX = "SETNX"
8285
cmdSIsMember = "SISMEMBER"
8386
cmdSMembers = "SMEMBERS"
8487
cmdSRem = "SREM"
@@ -166,6 +169,7 @@ var argsLen = map[string]int{
166169
cmdFlushAll: 1,
167170
cmdFlushDB: 1,
168171
cmdGet: 2,
172+
cmdGetDel: 2,
169173
cmdHDel: -3,
170174
cmdHExists: 3,
171175
cmdHGet: 3,
@@ -205,6 +209,8 @@ var argsLen = map[string]int{
205209
cmdScan: -2,
206210
cmdSelect: 2,
207211
cmdSet: -3,
212+
cmdSetEx: 4,
213+
cmdSetNX: 3,
208214
cmdSIsMember: 3,
209215
cmdSMembers: 2,
210216
cmdSRem: -3,
@@ -311,6 +317,7 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
311317
cmdFlushAll: r.flushall,
312318
cmdFlushDB: r.flushdb,
313319
cmdGet: r.get,
320+
cmdGetDel: r.getdel,
314321
cmdHDel: r.hdel,
315322
cmdHExists: r.hexists,
316323
cmdHGet: r.hget,
@@ -350,6 +357,8 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
350357
cmdScan: r.scan,
351358
cmdSelect: r.selectDB,
352359
cmdSet: r.set,
360+
cmdSetEx: r.setex,
361+
cmdSetNX: r.setnx,
353362
cmdSIsMember: r.sismember,
354363
cmdSMembers: r.smembers,
355364
cmdSRem: r.srem,

‎adapter/redis_compat_commands.go‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,82 @@ func (r *RedisServer) info(conn redcon.Conn, _ redcon.Command) {
7171
}, "\r\n"))
7272
}
7373

74+
// SETEX key seconds value — equivalent to SET key value EX seconds
75+
func (r *RedisServer) setex(conn redcon.Conn, cmd redcon.Command) {
76+
seconds, err := strconv.ParseInt(string(cmd.Args[2]), 10, 64)
77+
if err != nil || seconds <= 0 {
78+
conn.WriteError("ERR invalid expire time in 'setex' command")
79+
return
80+
}
81+
ttl := time.Now().Add(time.Duration(seconds) * time.Second)
82+
83+
ctx, cancel := context.WithTimeout(context.Background(), redisDispatchTimeout)
84+
defer cancel()
85+
if err := r.saveString(ctx, cmd.Args[1], cmd.Args[3], &ttl); err != nil {
86+
conn.WriteError(err.Error())
87+
return
88+
}
89+
conn.WriteString("OK")
90+
}
91+
92+
// GETDEL key — get the value and delete the key atomically
93+
func (r *RedisServer) getdel(conn redcon.Conn, cmd redcon.Command) {
94+
key := cmd.Args[1]
95+
readTS := r.readTS()
96+
97+
typ, err := r.keyTypeAt(context.Background(), key, readTS)
98+
if err != nil {
99+
conn.WriteError(err.Error())
100+
return
101+
}
102+
if typ == redisTypeNone {
103+
conn.WriteNull()
104+
return
105+
}
106+
if typ != redisTypeString {
107+
conn.WriteError(wrongTypeMessage)
108+
return
109+
}
110+
111+
v, err := r.readValueAt(key, readTS)
112+
if err != nil {
113+
conn.WriteNull()
114+
return
115+
}
116+
117+
ctx, cancel := context.WithTimeout(context.Background(), redisDispatchTimeout)
118+
defer cancel()
119+
if err := r.retryRedisWrite(ctx, func() error {
120+
elems, _, err := r.deleteLogicalKeyElems(ctx, key, r.readTS())
121+
if err != nil {
122+
return err
123+
}
124+
return r.dispatchElems(ctx, true, elems)
125+
}); err != nil {
126+
conn.WriteError(err.Error())
127+
return
128+
}
129+
conn.WriteBulk(v)
130+
}
131+
132+
// SETNX key value — set if not exists, returns 1 on success, 0 on failure
133+
func (r *RedisServer) setnx(conn redcon.Conn, cmd redcon.Command) {
134+
ctx, cancel := context.WithTimeout(context.Background(), redisDispatchTimeout)
135+
defer cancel()
136+
137+
opts := redisSetOptions{missingCond: true}
138+
result, err := r.executeSet(ctx, cmd.Args[1], cmd.Args[2], opts)
139+
if err != nil {
140+
conn.WriteError(err.Error())
141+
return
142+
}
143+
if result.wroteNull {
144+
conn.WriteInt(0)
145+
return
146+
}
147+
conn.WriteInt(1)
148+
}
149+
74150
func (r *RedisServer) client(conn redcon.Conn, cmd redcon.Command) {
75151
sub := strings.ToUpper(string(cmd.Args[1]))
76152
switch sub {

‎adapter/redis_lua.go‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -512,7 +512,7 @@ func goToLuaValue(state *lua.LState, value any) lua.LValue {
512512
func goToLuaScalar(value any) (lua.LValue, bool) {
513513
switch x := value.(type) {
514514
case nil:
515-
return lua.LFalse, true
515+
return lua.LNil, true
516516
case bool:
517517
if x {
518518
return lua.LTrue, true
@@ -585,7 +585,11 @@ func goToLuaFloatNumber(value any) (lua.LValue, bool) {
585585
func goSliceToLuaValue(state *lua.LState, values []any) lua.LValue {
586586
tbl := state.NewTable()
587587
for i, item := range values {
588-
tbl.RawSetInt(i+luaTypeArrayBase, goToLuaValue(state, item))
588+
v := goToLuaValue(state, item)
589+
if v == lua.LNil {
590+
continue // nil entries are not stored in Lua tables (matching Redis cmsgpack)
591+
}
592+
tbl.RawSetInt(i+luaTypeArrayBase, v)
589593
}
590594
return tbl
591595
}
@@ -598,7 +602,11 @@ func goMapToLuaValue(state *lua.LState, value map[string]any) lua.LValue {
598602
}
599603
sort.Strings(keys)
600604
for _, key := range keys {
601-
tbl.RawSetString(key, goToLuaValue(state, value[key]))
605+
v := goToLuaValue(state, value[key])
606+
if v == lua.LNil {
607+
continue
608+
}
609+
tbl.RawSetString(key, v)
602610
}
603611
return tbl
604612
}

‎adapter/redis_lua_context.go‎

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,9 @@ var luaCommandHandlers = map[string]luaCommandHandler{
146146
cmdZRemRangeByScore: (*luaScriptContext).cmdZRemRangeByScore,
147147
cmdXAdd: (*luaScriptContext).cmdXAdd,
148148
cmdXTrim: (*luaScriptContext).cmdXTrim,
149+
cmdSetEx: (*luaScriptContext).cmdSetEx,
150+
cmdGetDel: (*luaScriptContext).cmdGetDel,
151+
cmdSetNX: (*luaScriptContext).cmdSetNX,
149152
}
150153

151154
var luaRenameHandlers = map[redisValueType]luaRenameHandler{
@@ -737,6 +740,58 @@ func (c *luaScriptContext) cmdSet(args []string) (luaReply, error) {
737740
return luaStatusReply("OK"), nil
738741
}
739742

743+
// SETEX key seconds value
744+
func (c *luaScriptContext) cmdSetEx(args []string) (luaReply, error) {
745+
key := []byte(args[0])
746+
seconds, err := strconv.ParseInt(args[1], 10, 64)
747+
if err != nil || seconds <= 0 {
748+
return luaReply{}, errors.New("ERR invalid expire time in 'setex' command")
749+
}
750+
value := []byte(args[2])
751+
ttl := time.Now().Add(time.Duration(seconds) * time.Second)
752+
753+
c.deleteLogical(key)
754+
c.markStringValue(key, value)
755+
c.setTTLValue(key, &ttl)
756+
return luaStatusReply("OK"), nil
757+
}
758+
759+
// GETDEL key
760+
func (c *luaScriptContext) cmdGetDel(args []string) (luaReply, error) {
761+
key := []byte(args[0])
762+
st, err := c.stringState(key)
763+
if err != nil {
764+
if errors.Is(err, store.ErrKeyNotFound) {
765+
return luaNilReply(), nil
766+
}
767+
return luaReply{}, err
768+
}
769+
if !st.exists {
770+
return luaNilReply(), nil
771+
}
772+
val := string(st.value)
773+
c.deleteLogical(key)
774+
return luaStringReply(val), nil
775+
}
776+
777+
// SETNX key value — returns 1 if set, 0 if already exists
778+
func (c *luaScriptContext) cmdSetNX(args []string) (luaReply, error) {
779+
key := []byte(args[0])
780+
value := []byte(args[1])
781+
782+
exists, _, err := c.loadLuaSetContext(key)
783+
if err != nil {
784+
return luaReply{}, err
785+
}
786+
if exists {
787+
return luaIntReply(0), nil
788+
}
789+
c.deleteLogical(key)
790+
c.markStringValue(key, value)
791+
c.clearTTL(key)
792+
return luaIntReply(1), nil
793+
}
794+
740795
func (c *luaScriptContext) loadLuaSetContext(key []byte) (bool, *time.Time, error) {
741796
exists, err := c.logicalExists(key)
742797
if err != nil {

‎adapter/redis_misskey_compat_test.go‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -312,3 +312,87 @@ func TestRedis_MisskeyFeaturedRankingTransaction(t *testing.T) {
312312
require.NoError(t, err)
313313
require.Greater(t, ttl, time.Duration(0))
314314
}
315+
316+
func TestRedis_MisskeySETEX(t *testing.T) {
317+
t.Parallel()
318+
nodes, _, _ := createNode(t, 3)
319+
defer shutdown(nodes)
320+
321+
rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
322+
ctx := context.Background()
323+
324+
// SETEX key seconds value
325+
res := rdb.Do(ctx, "SETEX", "webauthn:challenge:u1", "120", "challenge-data")
326+
require.NoError(t, res.Err())
327+
require.Equal(t, "OK", res.Val())
328+
329+
// Verify value was stored
330+
val, err := rdb.Get(ctx, "webauthn:challenge:u1").Result()
331+
require.NoError(t, err)
332+
require.Equal(t, "challenge-data", val)
333+
334+
// Verify TTL was set
335+
ttl, err := rdb.TTL(ctx, "webauthn:challenge:u1").Result()
336+
require.NoError(t, err)
337+
require.Greater(t, ttl, time.Duration(0))
338+
require.LessOrEqual(t, ttl, 120*time.Second)
339+
340+
// SETEX overwrites existing value
341+
res = rdb.Do(ctx, "SETEX", "webauthn:challenge:u1", "60", "new-challenge")
342+
require.NoError(t, res.Err())
343+
val, err = rdb.Get(ctx, "webauthn:challenge:u1").Result()
344+
require.NoError(t, err)
345+
require.Equal(t, "new-challenge", val)
346+
}
347+
348+
func TestRedis_MisskeyGETDEL(t *testing.T) {
349+
t.Parallel()
350+
nodes, _, _ := createNode(t, 3)
351+
defer shutdown(nodes)
352+
353+
rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
354+
ctx := context.Background()
355+
356+
// Set a value, then GETDEL
357+
require.NoError(t, rdb.Set(ctx, "webauthn:reg:u1", "reg-data", 0).Err())
358+
359+
val := rdb.Do(ctx, "GETDEL", "webauthn:reg:u1")
360+
require.NoError(t, val.Err())
361+
require.Equal(t, "reg-data", val.Val())
362+
363+
// Key should be deleted now
364+
require.ErrorIs(t, rdb.Get(ctx, "webauthn:reg:u1").Err(), redis.Nil)
365+
366+
// GETDEL on non-existent key returns nil
367+
val = rdb.Do(ctx, "GETDEL", "nonexistent")
368+
require.ErrorIs(t, val.Err(), redis.Nil)
369+
}
370+
371+
func TestRedis_MisskeySETNX(t *testing.T) {
372+
t.Parallel()
373+
nodes, _, _ := createNode(t, 3)
374+
defer shutdown(nodes)
375+
376+
rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
377+
ctx := context.Background()
378+
379+
// SETNX on non-existent key succeeds (returns 1)
380+
res := rdb.Do(ctx, "SETNX", "lock:distributed:1", "owner1")
381+
require.NoError(t, res.Err())
382+
require.Equal(t, int64(1), res.Val())
383+
384+
// Verify value was stored
385+
val, err := rdb.Get(ctx, "lock:distributed:1").Result()
386+
require.NoError(t, err)
387+
require.Equal(t, "owner1", val)
388+
389+
// SETNX on existing key fails (returns 0)
390+
res = rdb.Do(ctx, "SETNX", "lock:distributed:1", "owner2")
391+
require.NoError(t, res.Err())
392+
require.Equal(t, int64(0), res.Val())
393+
394+
// Value should not have changed
395+
val, err = rdb.Get(ctx, "lock:distributed:1").Result()
396+
require.NoError(t, err)
397+
require.Equal(t, "owner1", val)
398+
}

0 commit comments

Comments
 (0)