24606763703

Committed 18 Apr 2026 02:27PM UTC coverage: 61.646% (-2.7%) from 64.328%

Build # 24606763703

Build Type

push

github

Committed by

HDT3213

Commit Message

fix compatibility for redis and valkey

Coverage Stats

5 of 105 new or added lines in 2 files covered. (4.76%)

258 existing lines in 3 files now uncovered.

1858 of 3014 relevant lines covered (61.65%)

0.68 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

60.22

/core/decoder.go

// Package core is RDB core core
package core

import (
        "bufio"
        "bytes"
        "encoding/binary"
        "errors"
        "fmt"
        "io"
        "strconv"
        "time"

        "github.com/hdt3213/rdb/memprofiler"
        "github.com/hdt3213/rdb/model"
)

// Decoder is an instance of rdb parsing process
type Decoder struct {
        input     *bufio.Reader
        readCount int
        buffer    []byte

        withSpecialOpCode bool
        withSpecialTypes  map[string]ModuleTypeHandleFunc

        valkey     bool
        rdbVersion int
}

// NewDecoder creates a new RDB decoder
func NewDecoder(reader io.Reader) *Decoder {
        parser := new(Decoder)
        parser.input = bufio.NewReader(reader)
        parser.buffer = make([]byte, 8)
        parser.withSpecialTypes = make(map[string]ModuleTypeHandleFunc)
        return parser
}

// WithSpecialOpCode enables returning model.AuxObject to callback
func (dec *Decoder) WithSpecialOpCode() *Decoder {
        dec.withSpecialOpCode = true
        return dec
}

// WithSpecialType enables returning redis module data structure to callback
func (dec *Decoder) WithSpecialType(moduleType string, f ModuleTypeHandleFunc) *Decoder {
        dec.withSpecialTypes[moduleType] = f
        return dec
}

var magicNumberRedis = []byte("REDIS")
var magicNumberValkey = []byte("VALKEY")

const (
        minVersion       = 1
        maxVersion       = 12
        minVersionValkey = 80
        maxVersionValkey = 80
)

const (
        opCodeSlotImport   = 243 /* Slot import state. */
        opCodeSlotInfo     = 244 /* Foreign slot info, safe to ignore. */
        opCodeFunction     = 245 /* function library data */
        opCodeModuleAux    = 247 /* Module auxiliary data. */
        opCodeIdle         = 248 /* LRU idle time. */
        opCodeFreq         = 249 /* LFU frequency. */
        opCodeAux          = 250 /* RDB aux field. */
        opCodeResizeDB     = 251 /* Hash table resize hint. */
        opCodeExpireTimeMs = 252 /* Expire time in milliseconds. */
        opCodeExpireTime   = 253 /* Old expire time in seconds. */
        opCodeSelectDB     = 254 /* DB number of the following keys. */
        opCodeEOF          = 255
)

const (
        typeString = iota
        typeList
        typeSet
        typeZset
        typeHash
        typeZset2 /* ZSET version 2 with doubles stored in binary. */
        typeModule
        typeModule2 // Module value parser should be registered with Decoder.WithSpecialType
        _
        typeHashZipMap
        typeListZipList
        typeSetIntSet
        typeZsetZipList
        typeHashZipList
        typeListQuickList
        typeStreamListPacks
        typeHashListPack
        typeZsetListPack
        typeListQuickList2
        typeStreamListPacks2
        typeSetListPack
        typeStreamListPacks3
        typeHashWithHfeRc         // rdb 12 (only redis 7.4 rc)
        typeHashListPackWithHfeRc // rdb 12 (only redis 7.4 rc)
        typeHashWithHfe           // since rdb 12 (redis 7.4)
        typeHashListPackWithHfe   // since rdb 12 (redis 7.4)

        typeHash2 = typeHashWithHfeRc // Hash with field-level expiration (Valkey 9+)
)

const (
        EB_EXPIRE_TIME_MAX     int64 = 0x0000FFFFFFFFFFFF
        EB_EXPIRE_TIME_INVALID int64 = EB_EXPIRE_TIME_MAX + 1
        HFE_MAX_ABS_TIME_MSEC  int64 = EB_EXPIRE_TIME_MAX >> 2
)

var encodingMap = map[int]string{
        typeString:                model.StringEncoding,
        typeList:                  model.ListEncoding,
        typeSet:                   model.SetEncoding,
        typeZset:                  model.ZSetEncoding,
        typeHash:                  model.HashEncoding,
        typeZset2:                 model.ZSet2Encoding,
        typeHashZipMap:            model.ZipMapEncoding,
        typeListZipList:           model.ZipListEncoding,
        typeSetIntSet:             model.IntSetEncoding,
        typeZsetZipList:           model.ZipListEncoding,
        typeHashZipList:           model.ZipListEncoding,
        typeListQuickList:         model.QuickListEncoding,
        typeStreamListPacks:       model.ListPackEncoding,
        typeStreamListPacks2:      model.ListPackEncoding,
        typeHashListPack:          model.ListPackEncoding,
        typeZsetListPack:          model.ListPackEncoding,
        typeListQuickList2:        model.QuickList2Encoding,
        typeSetListPack:           model.ListPackEncoding,
        typeHashWithHfeRc:         model.HashExEncoding,
        typeHashListPackWithHfeRc: model.ListPackExEncoding,
        typeHashWithHfe:           model.HashExEncoding,
        typeHashListPackWithHfe:   model.ListPackExEncoding,
        // typeHash2: model.HashExEncoding, // same 22 as typeHashWithHfeRc
}

// checkHeader checks whether input has valid RDB file header
func (dec *Decoder) checkHeader() error {
        header := make([]byte, 9)
        err := dec.readFull(header)
        if err == io.EOF {
                return errors.New("empty file")
        }
        if err != nil {
                return fmt.Errorf("io error: %v", err)
        }
        var versionString string
        if bytes.HasPrefix(header, magicNumberRedis) {
                dec.valkey = false
                versionString = string(bytes.TrimPrefix(header, magicNumberRedis))
        } else if bytes.HasPrefix(header, magicNumberValkey) {
                dec.valkey = true
                versionString = string(bytes.TrimPrefix(header, magicNumberValkey))
        } else {
                return errors.New("file is not a RDB file")
        }
        version, err := strconv.Atoi(versionString)
        if err != nil {
                return fmt.Errorf("%s is not valid version number", versionString)
        }
        if !dec.valkey && (version < minVersion || version > maxVersion) {
                return fmt.Errorf("cannot parse version: %d", version)
        }
        if dec.valkey && (version < minVersionValkey || version > maxVersionValkey) {
                return fmt.Errorf("cannot parse version: %d", version)
        }
        dec.rdbVersion = version
        return nil
}

func (dec *Decoder) readObject(flag byte, base *model.BaseObject) (model.RedisObject, error) {
        base.Encoding = encodingMap[int(flag)]
        switch flag {
        case typeString:
                bs, err := dec.readString()
                if err != nil {
                        return nil, err
                }
                return &model.StringObject{
                        BaseObject: base,
                        Value:      bs,
                }, nil
        case typeList:
                list, err := dec.readList()
                if err != nil {
                        return nil, err
                }
                return &model.ListObject{
                        BaseObject: base,
                        Values:     list,
                }, nil
        case typeSet:
                set, err := dec.readSet()
                if err != nil {
                        return nil, err
                }
                return &model.SetObject{
                        BaseObject: base,
                        Members:    set,
                }, nil
        case typeSetIntSet:
                set, extra, err := dec.readIntSet()
                if err != nil {
                        return nil, err
                }
                base.Extra = extra
                return &model.SetObject{
                        BaseObject: base,
                        Members:    set,
                }, nil
        case typeHash:
                hash, err := dec.readHashMap()
                if err != nil {
                        return nil, err
                }
                return &model.HashObject{
                        BaseObject: base,
                        Hash:       hash,
                }, nil
        case typeListZipList:
                list, err := dec.readZipList()
                if err != nil {
                        return nil, err
                }
                return &model.ListObject{
                        BaseObject: base,
                        Values:     list,
                }, nil
        case typeListQuickList:
                list, extra, err := dec.readQuickList()
                if err != nil {
                        return nil, err
                }
                base.Extra = extra
                return &model.ListObject{
                        BaseObject: base,
                        Values:     list,
                }, nil
        case typeListQuickList2:
                list, extra, err := dec.readQuickList2()
                if err != nil {
                        return nil, err
                }
                base.Extra = extra
                return &model.ListObject{
                        BaseObject: base,
                        Values:     list,
                }, nil
        case typeHashZipMap:
                m, err := dec.readZipMapHash()
                if err != nil {
                        return nil, err
                }
                return &model.HashObject{
                        BaseObject: base,
                        Hash:       m,
                }, nil
        case typeHashZipList:
                m, extra, err := dec.readZipListHash()
                if err != nil {
                        return nil, err
                }
                base.Extra = extra
                return &model.HashObject{
                        BaseObject: base,
                        Hash:       m,
                }, nil
        case typeHashListPack:
                m, extra, err := dec.readListPackHash()
                if err != nil {
                        return nil, err
                }
                base.Extra = extra
                return &model.HashObject{
                        BaseObject: base,
                        Hash:       m,
                }, nil
        case typeZset:
                entries, err := dec.readZSet(false)
                if err != nil {
                        return nil, err
                }
                return &model.ZSetObject{
                        BaseObject: base,
                        Entries:    entries,
                }, nil
        case typeZset2:
                entries, err := dec.readZSet(true)
                if err != nil {
                        return nil, err
                }
                return &model.ZSetObject{
                        BaseObject: base,
                        Entries:    entries,
                }, nil
        case typeZsetZipList:
                entries, extra, err := dec.readZipListZSet()
                if err != nil {
                        return nil, err
                }
                base.Extra = extra
                return &model.ZSetObject{
                        BaseObject: base,
                        Entries:    entries,
                }, nil
        case typeZsetListPack:
                entries, extra, err := dec.readListPackZSet()
                if err != nil {
                        return nil, err
                }
                base.Extra = extra
                return &model.ZSetObject{
                        BaseObject: base,
                        Entries:    entries,
                }, nil
        case typeStreamListPacks, typeStreamListPacks2, typeStreamListPacks3:
                var version uint = 1
                if flag == typeStreamListPacks2 {
                        version = 2
                } else if flag == typeStreamListPacks3 {
                        version = 3
                }
                stream, err := dec.readStreamListPacks(version)
                if err != nil {
                        return nil, err
                }
                stream.BaseObject = base
                return stream, nil
        case typeModule2:
                moduleType, val, err := dec.readModuleType()
                if err != nil {
                        return nil, err
                }
                return &model.ModuleTypeObject{
                        BaseObject: base,
                        ModuleType: moduleType,
                        Value:      val,
                }, nil
        case typeSetListPack:
                set, extra, err := dec.readListPackSet()
                if err != nil {
                        return nil, err
                }
                base.Extra = extra
                return &model.SetObject{
                        BaseObject: base,
                        Members:    set,
                }, nil
        case typeHashWithHfe, typeHashWithHfeRc: // typeHash2 == typeHashWithHfeRc, same value 22
                var hash map[string][]byte
                var expire map[string]int64
                var err error
                if dec.valkey {
                        // Valkey 9+ Hash2: absolute timestamps after each field-value pair
                        hash, expire, err = dec.readHashMapExValkey()
                } else {
                        // Redis 7.4: typeHashWithHfeRc (rc=true) or typeHashWithHfe (rc=false)
                        hash, expire, err = dec.readHashMapEx(flag == typeHashWithHfeRc)
                }
                if err != nil {
                        return nil, err
                }
                return &model.HashObject{
                        BaseObject:       base,
                        Hash:             hash,
                        FieldExpirations: expire,
                }, nil
        case typeHashListPackWithHfe, typeHashListPackWithHfeRc:
                m, e, extra, err := dec.readListPackHashEx(func() bool { return flag == typeHashListPackWithHfeRc }())
                if err != nil {
                        return nil, err
                }
                base.Extra = extra
                return &model.HashObject{
                        BaseObject:       base,
                        Hash:             m,
                        FieldExpirations: e,
                }, nil
        }
        return nil, fmt.Errorf("unknown type flag: %b", flag)
}

func (dec *Decoder) parse(cb func(object model.RedisObject) bool) error {
        var dbIndex int
        var expireMs int64
        var lru *int64
        var lfu *int64
        for {
                b, err := dec.readByte()
                if err != nil {
                        return err
                }
                if b == opCodeEOF {
                        break
                } else if b == opCodeSelectDB {
                        dbIndex64, _, err := dec.readLength()
                        if err != nil {
                                return err
                        }
                        dbIndex = int(dbIndex64)
                        continue
                } else if b == opCodeExpireTime {
                        err = dec.readFull(dec.buffer[:4])
                        if err != nil {
                                return err
                        }
                        expireMs = int64(binary.LittleEndian.Uint32(dec.buffer)) * 1000
                        continue
                } else if b == opCodeExpireTimeMs {
                        err = dec.readFull(dec.buffer)
                        if err != nil {
                                return err
                        }
                        expireMs = int64(binary.LittleEndian.Uint64(dec.buffer))
                        continue
                } else if b == opCodeResizeDB {
                        keyCount, _, err := dec.readLength()
                        if err != nil {
                                return err
                        }
                        ttlCount, _, err := dec.readLength()
                        if err != nil {
                                err = errors.New("Parse Aux value failed: " + err.Error())
                                break
                        }
                        if dec.withSpecialOpCode {
                                obj := &model.DBSizeObject{
                                        BaseObject: &model.BaseObject{},
                                }
                                obj.DB = dbIndex
                                obj.KeyCount = keyCount
                                obj.TTLCount = ttlCount
                                tbc := cb(obj)
                                if !tbc {
                                        break
                                }
                        }
                        continue
                } else if b == opCodeAux {
                        key, err := dec.readString()
                        if err != nil {
                                return err
                        }
                        value, err := dec.readString()
                        if err != nil {
                                err = errors.New("Parse Aux value failed: " + err.Error())
                                break
                        }
                        if dec.withSpecialOpCode {
                                obj := &model.AuxObject{
                                        BaseObject: &model.BaseObject{},
                                }
                                obj.Type = model.AuxType
                                obj.Key = unsafeBytes2Str(key)
                                obj.Value = unsafeBytes2Str(value)
                                tbc := cb(obj)
                                if !tbc {
                                        break
                                }
                        }
                        continue
                } else if b == opCodeFreq {
                        freq, err := dec.readByte()
                        if err != nil {
                                return err
                        }
                        v := int64(freq)
                        lfu = &v
                        continue
                } else if b == opCodeIdle {
                        idle, _, err := dec.readLength()
                        if err != nil {
                                return err
                        }
                        v := int64(idle)
                        lru = &v
                        continue
                } else if b == opCodeModuleAux {
                        _, _, err = dec.readModuleType()
                        if err != nil {
                                return err
                        }
                        continue
                } else if b == opCodeFunction {
                        functionsLua, err := dec.readString()
                        if err != nil {
                                return err
                        }
                        if dec.withSpecialOpCode {
                                obj := &model.FunctionsObject{
                                        BaseObject: &model.BaseObject{},
                                }
                                obj.Key = "functions"
                                obj.Type = model.FunctionsType
                                obj.Encoding = "functions"
                                obj.FunctionsLua = unsafeBytes2Str(functionsLua)
                                tbc := cb(obj)
                                if !tbc {
                                        break
                                }
                        }
                        continue
                } else if b == opCodeSlotInfo {
                        if !dec.valkey {
                                return fmt.Errorf("unsupported opcode 244 in Redis RDB version %d", dec.rdbVersion)
                        }
                        // Valkey 9+: slot info metadata, safe to skip
                        var err error
                        var slot_id, slot_size, expires_slot_size uint64
                        slot_id, _, err = dec.readLength()
                        if err == nil {
                                slot_size, _, err = dec.readLength()
                        }
                        if err == nil {
                                expires_slot_size, _, err = dec.readLength()
                        }
                        if err != nil {
                                return err
                        }
                        _, _, _ = slot_id, slot_size, expires_slot_size // safe to skip
                        continue
                } else if b == opCodeSlotImport { // opcode 243: Valkey=SlotImport, Redis 8.0+=KeyMeta
                        if dec.valkey {
                                // Valkey 9+: slot import state
                                job, err := dec.readString()
                                if err != nil {
                                        return err
                                }
                                num_slot_ranges, _, err := dec.readLength()
                                if err != nil {
                                        return err
                                }
                                var slot_from, slot_to uint64
                                ranges := make([]string, num_slot_ranges)
                                for i := uint64(0); i < num_slot_ranges; i++ {
                                        slot_from, _, err = dec.readLength()
                                        if err == nil {
                                                slot_to, _, err = dec.readLength()
                                        }
                                        if err != nil {
                                                return err
                                        }
                                        ranges[i] = fmt.Sprintf("%d-%d", slot_from, slot_to)
                                }
                                _, _ = job, ranges // safe to skip
                        } else {
                                // Redis 8.0+: RDB_OPCODE_KEY_META (same opcode value 243)
                                // Not yet supported; return error to avoid silent data corruption
                                return fmt.Errorf("unsupported opcode: RDB_OPCODE_KEY_META (243) in Redis RDB version %d", dec.rdbVersion)
                        }
                        continue
                }
                key, err := dec.readString()
                if err != nil {
                        return err
                }
                base := &model.BaseObject{
                        DB:  dbIndex,
                        Key: unsafeBytes2Str(key),
                }
                if expireMs > 0 {
                        expiration := time.Unix(0, expireMs*int64(time.Millisecond))
                        base.Expiration = &expiration
                        expireMs = 0 // reset expire ms
                }
                base.IdleTime = lru
                lru = nil // reset lru
                base.Freq = lfu
                lfu = nil // reset lfu
                obj, err := dec.readObject(b, base)
                if err != nil {
                        return err
                }
                base.Size = memprofiler.SizeOfObject(obj)
                base.Type = obj.GetType()
                tbc := cb(obj)
                if !tbc {
                        break
                }
        }
        // read crc64 at the end
        _ = dec.readFull(dec.buffer)
        return nil
}

// Parse parses rdb and callback
// cb returns true to continue, returns false to stop the iteration
func (dec *Decoder) Parse(cb func(object model.RedisObject) bool) (err error) {
        defer func() {
                if err2 := recover(); err2 != nil {
                        err = fmt.Errorf("panic: %v", err2)
                }
        }()
        err = dec.checkHeader()
        if err != nil {
                return err
        }
        return dec.parse(cb)
}

func (dec *Decoder) GetReadCount() int {
        return dec.readCount
}

1	// Package core is RDB core core
2	package core
3
4	import (
5	"bufio"
6	"bytes"
7	"encoding/binary"
8	"errors"
9	"fmt"
10	"io"
11	"strconv"
12	"time"
13
14	"github.com/hdt3213/rdb/memprofiler"
15	"github.com/hdt3213/rdb/model"
16	)
17
18	// Decoder is an instance of rdb parsing process
19	type Decoder struct {
20	input *bufio.Reader
21	readCount int
22	buffer []byte
23
24	withSpecialOpCode bool
25	withSpecialTypes map[string]ModuleTypeHandleFunc
26
27	valkey bool
28	rdbVersion int
29	}
30
31	// NewDecoder creates a new RDB decoder
32	func NewDecoder(reader io.Reader) *Decoder {	1✔
33	parser := new(Decoder)	1✔
34	parser.input = bufio.NewReader(reader)	1✔
35	parser.buffer = make([]byte, 8)	1✔
36	parser.withSpecialTypes = make(map[string]ModuleTypeHandleFunc)	1✔
37	return parser	1✔
38	}	1✔
39
40	// WithSpecialOpCode enables returning model.AuxObject to callback
41	func (dec Decoder) WithSpecialOpCode() Decoder {	1✔
42	dec.withSpecialOpCode = true	1✔
43	return dec	1✔
44	}	1✔
45
46	// WithSpecialType enables returning redis module data structure to callback
47	func (dec Decoder) WithSpecialType(moduleType string, f ModuleTypeHandleFunc) Decoder {	1✔
48	dec.withSpecialTypes[moduleType] = f	1✔
49	return dec	1✔
50	}	1✔
51
52	var magicNumberRedis = []byte("REDIS")
53	var magicNumberValkey = []byte("VALKEY")
54
55	const (
56	minVersion = 1
57	maxVersion = 12
58	minVersionValkey = 80
59	maxVersionValkey = 80
60	)
61
62	const (
63	opCodeSlotImport = 243 /* Slot import state. */
64	opCodeSlotInfo = 244 /* Foreign slot info, safe to ignore. */
65	opCodeFunction = 245 /* function library data */
66	opCodeModuleAux = 247 /* Module auxiliary data. */
67	opCodeIdle = 248 /* LRU idle time. */
68	opCodeFreq = 249 /* LFU frequency. */
69	opCodeAux = 250 /* RDB aux field. */
70	opCodeResizeDB = 251 /* Hash table resize hint. */
71	opCodeExpireTimeMs = 252 /* Expire time in milliseconds. */
72	opCodeExpireTime = 253 /* Old expire time in seconds. */
73	opCodeSelectDB = 254 /* DB number of the following keys. */
74	opCodeEOF = 255
75	)
76
77	const (
78	typeString = iota
79	typeList
80	typeSet
81	typeZset
82	typeHash
83	typeZset2 /* ZSET version 2 with doubles stored in binary. */
84	typeModule
85	typeModule2 // Module value parser should be registered with Decoder.WithSpecialType
86	_
87	typeHashZipMap
88	typeListZipList
89	typeSetIntSet
90	typeZsetZipList
91	typeHashZipList
92	typeListQuickList
93	typeStreamListPacks
94	typeHashListPack
95	typeZsetListPack
96	typeListQuickList2
97	typeStreamListPacks2
98	typeSetListPack
99	typeStreamListPacks3
100	typeHashWithHfeRc // rdb 12 (only redis 7.4 rc)
101	typeHashListPackWithHfeRc // rdb 12 (only redis 7.4 rc)
102	typeHashWithHfe // since rdb 12 (redis 7.4)
103	typeHashListPackWithHfe // since rdb 12 (redis 7.4)
104
105	typeHash2 = typeHashWithHfeRc // Hash with field-level expiration (Valkey 9+)
106	)
107
108	const (
109	EB_EXPIRE_TIME_MAX int64 = 0x0000FFFFFFFFFFFF
110	EB_EXPIRE_TIME_INVALID int64 = EB_EXPIRE_TIME_MAX + 1
111	HFE_MAX_ABS_TIME_MSEC int64 = EB_EXPIRE_TIME_MAX >> 2
112	)
113
114	var encodingMap = map[int]string{
115	typeString: model.StringEncoding,
116	typeList: model.ListEncoding,
117	typeSet: model.SetEncoding,
118	typeZset: model.ZSetEncoding,
119	typeHash: model.HashEncoding,
120	typeZset2: model.ZSet2Encoding,
121	typeHashZipMap: model.ZipMapEncoding,
122	typeListZipList: model.ZipListEncoding,
123	typeSetIntSet: model.IntSetEncoding,
124	typeZsetZipList: model.ZipListEncoding,
125	typeHashZipList: model.ZipListEncoding,
126	typeListQuickList: model.QuickListEncoding,
127	typeStreamListPacks: model.ListPackEncoding,
128	typeStreamListPacks2: model.ListPackEncoding,
129	typeHashListPack: model.ListPackEncoding,
130	typeZsetListPack: model.ListPackEncoding,
131	typeListQuickList2: model.QuickList2Encoding,
132	typeSetListPack: model.ListPackEncoding,
133	typeHashWithHfeRc: model.HashExEncoding,
134	typeHashListPackWithHfeRc: model.ListPackExEncoding,
135	typeHashWithHfe: model.HashExEncoding,
136	typeHashListPackWithHfe: model.ListPackExEncoding,
137	// typeHash2: model.HashExEncoding, // same 22 as typeHashWithHfeRc
138	}
139
140	// checkHeader checks whether input has valid RDB file header
141	func (dec *Decoder) checkHeader() error {	1✔
142	header := make([]byte, 9)	1✔
143	err := dec.readFull(header)	1✔
144	if err == io.EOF {	1✔
145	return errors.New("empty file")	×
146	}	×
147	if err != nil {	1✔
148	return fmt.Errorf("io error: %v", err)	×
149	}	×
150	var versionString string	1✔
151	if bytes.HasPrefix(header, magicNumberRedis) {	2✔
152	dec.valkey = false	1✔
153	versionString = string(bytes.TrimPrefix(header, magicNumberRedis))	1✔
154	} else if bytes.HasPrefix(header, magicNumberValkey) {	1✔
UNCOV 155	dec.valkey = true	×
UNCOV 156	versionString = string(bytes.TrimPrefix(header, magicNumberValkey))	×
UNCOV 157	} else {	×
UNCOV 158	return errors.New("file is not a RDB file")	×
159	}	×
160	version, err := strconv.Atoi(versionString)	1✔
161	if err != nil {	1✔
UNCOV 162	return fmt.Errorf("%s is not valid version number", versionString)	×
UNCOV 163	}	×
164	if !dec.valkey && (version < minVersion \|\| version > maxVersion) {	1✔
UNCOV 165	return fmt.Errorf("cannot parse version: %d", version)	×
UNCOV 166	}	×
167	if dec.valkey && (version < minVersionValkey \|\| version > maxVersionValkey) {	1✔
168	return fmt.Errorf("cannot parse version: %d", version)	×
169	}	×
170	dec.rdbVersion = version	1✔
171	return nil	1✔
172	}
173
174	func (dec Decoder) readObject(flag byte, base model.BaseObject) (model.RedisObject, error) {	1✔
175	base.Encoding = encodingMap[int(flag)]	1✔
176	switch flag {	1✔
177	case typeString:	1✔
178	bs, err := dec.readString()	1✔
179	if err != nil {	1✔
UNCOV 180	return nil, err	×
UNCOV 181	}	×
182	return &model.StringObject{	1✔
183	BaseObject: base,	1✔
184	Value: bs,	1✔
185	}, nil	1✔
186	case typeList:	1✔
187	list, err := dec.readList()	1✔
188	if err != nil {	1✔
UNCOV 189	return nil, err	×
UNCOV 190	}	×
191	return &model.ListObject{	1✔
192	BaseObject: base,	1✔
193	Values: list,	1✔
194	}, nil	1✔
195	case typeSet:	1✔
196	set, err := dec.readSet()	1✔
197	if err != nil {	1✔
UNCOV 198	return nil, err	×
UNCOV 199	}	×
200	return &model.SetObject{	1✔
201	BaseObject: base,	1✔
202	Members: set,	1✔
203	}, nil	1✔
204	case typeSetIntSet:	1✔
205	set, extra, err := dec.readIntSet()	1✔
206	if err != nil {	1✔
UNCOV 207	return nil, err	×
UNCOV 208	}	×
209	base.Extra = extra	1✔
210	return &model.SetObject{	1✔
211	BaseObject: base,	1✔
212	Members: set,	1✔
213	}, nil	1✔
214	case typeHash:	1✔
215	hash, err := dec.readHashMap()	1✔
216	if err != nil {	1✔
UNCOV 217	return nil, err	×
UNCOV 218	}	×
219	return &model.HashObject{	1✔
220	BaseObject: base,	1✔
221	Hash: hash,	1✔
222	}, nil	1✔
223	case typeListZipList:	1✔
224	list, err := dec.readZipList()	1✔
225	if err != nil {	1✔
UNCOV 226	return nil, err	×
UNCOV 227	}	×
228	return &model.ListObject{	1✔
229	BaseObject: base,	1✔
230	Values: list,	1✔
231	}, nil	1✔
232	case typeListQuickList:	1✔
233	list, extra, err := dec.readQuickList()	1✔
234	if err != nil {	1✔
235	return nil, err	×
UNCOV 236	}	×
237	base.Extra = extra	1✔
238	return &model.ListObject{	1✔
239	BaseObject: base,	1✔
240	Values: list,	1✔
241	}, nil	1✔
242	case typeListQuickList2:	1✔
243	list, extra, err := dec.readQuickList2()	1✔
244	if err != nil {	1✔
UNCOV 245	return nil, err	×
UNCOV 246	}	×
247	base.Extra = extra	1✔
248	return &model.ListObject{	1✔
249	BaseObject: base,	1✔
250	Values: list,	1✔
251	}, nil	1✔
252	case typeHashZipMap:	1✔
253	m, err := dec.readZipMapHash()	1✔
254	if err != nil {	1✔
UNCOV 255	return nil, err	×
UNCOV 256	}	×
257	return &model.HashObject{	1✔
258	BaseObject: base,	1✔
259	Hash: m,	1✔
260	}, nil	1✔
261	case typeHashZipList:	1✔
262	m, extra, err := dec.readZipListHash()	1✔
263	if err != nil {	1✔
264	return nil, err	×
UNCOV 265	}	×
266	base.Extra = extra	1✔
267	return &model.HashObject{	1✔
268	BaseObject: base,	1✔
269	Hash: m,	1✔
270	}, nil	1✔
271	case typeHashListPack:	1✔
272	m, extra, err := dec.readListPackHash()	1✔
273	if err != nil {	1✔
UNCOV 274	return nil, err	×
UNCOV 275	}	×
276	base.Extra = extra	1✔
277	return &model.HashObject{	1✔
278	BaseObject: base,	1✔
279	Hash: m,	1✔
280	}, nil	1✔
281	case typeZset:	1✔
282	entries, err := dec.readZSet(false)	1✔
283	if err != nil {	1✔
UNCOV 284	return nil, err	×
UNCOV 285	}	×
286	return &model.ZSetObject{	1✔
287	BaseObject: base,	1✔
288	Entries: entries,	1✔
289	}, nil	1✔
290	case typeZset2:	1✔
291	entries, err := dec.readZSet(true)	1✔
292	if err != nil {	1✔
UNCOV 293	return nil, err	×
UNCOV 294	}	×
295	return &model.ZSetObject{	1✔
296	BaseObject: base,	1✔
297	Entries: entries,	1✔
298	}, nil	1✔
299	case typeZsetZipList:	1✔
300	entries, extra, err := dec.readZipListZSet()	1✔
301	if err != nil {	1✔
UNCOV 302	return nil, err	×
UNCOV 303	}	×
304	base.Extra = extra	1✔
305	return &model.ZSetObject{	1✔
306	BaseObject: base,	1✔
307	Entries: entries,	1✔
308	}, nil	1✔
309	case typeZsetListPack:	1✔
310	entries, extra, err := dec.readListPackZSet()	1✔
311	if err != nil {	1✔
UNCOV 312	return nil, err	×
UNCOV 313	}	×
314	base.Extra = extra	1✔
315	return &model.ZSetObject{	1✔
316	BaseObject: base,	1✔
317	Entries: entries,	1✔
318	}, nil	1✔
319	case typeStreamListPacks, typeStreamListPacks2, typeStreamListPacks3:	1✔
320	var version uint = 1	1✔
321	if flag == typeStreamListPacks2 {	2✔
322	version = 2	1✔
323	} else if flag == typeStreamListPacks3 {	3✔
324	version = 3	1✔
325	}	1✔
326	stream, err := dec.readStreamListPacks(version)	1✔
327	if err != nil {	1✔
UNCOV 328	return nil, err	×
UNCOV 329	}	×
330	stream.BaseObject = base	1✔
331	return stream, nil	1✔
332	case typeModule2:	1✔
333	moduleType, val, err := dec.readModuleType()	1✔
334	if err != nil {	1✔
335	return nil, err	×
336	}	×
337	return &model.ModuleTypeObject{	1✔
338	BaseObject: base,	1✔
339	ModuleType: moduleType,	1✔
340	Value: val,	1✔
341	}, nil	1✔
342	case typeSetListPack:	1✔
343	set, extra, err := dec.readListPackSet()	1✔
344	if err != nil {	1✔
345	return nil, err	×
346	}	×
347	base.Extra = extra	1✔
348	return &model.SetObject{	1✔
349	BaseObject: base,	1✔
350	Members: set,	1✔
351	}, nil	1✔
NEW 352	case typeHashWithHfe, typeHashWithHfeRc: // typeHash2 == typeHashWithHfeRc, same value 22	×
NEW 353	var hash map[string][]byte	×
NEW 354	var expire map[string]int64	×
NEW 355	var err error	×
NEW 356	if dec.valkey {	×
NEW 357	// Valkey 9+ Hash2: absolute timestamps after each field-value pair	×
NEW 358	hash, expire, err = dec.readHashMapExValkey()	×
NEW 359	} else {	×
NEW 360	// Redis 7.4: typeHashWithHfeRc (rc=true) or typeHashWithHfe (rc=false)	×
NEW 361	hash, expire, err = dec.readHashMapEx(flag == typeHashWithHfeRc)	×
NEW 362	}	×
UNCOV 363	if err != nil {	×
UNCOV 364	return nil, err	×
UNCOV 365	}	×
UNCOV 366	return &model.HashObject{	×
UNCOV 367	BaseObject: base,	×
UNCOV 368	Hash: hash,	×
UNCOV 369	FieldExpirations: expire,	×
UNCOV 370	}, nil	×
371	case typeHashListPackWithHfe, typeHashListPackWithHfeRc:	×
372	m, e, extra, err := dec.readListPackHashEx(func() bool { return flag == typeHashListPackWithHfeRc }())	×
UNCOV 373	if err != nil {	×
UNCOV 374	return nil, err	×
UNCOV 375	}	×
UNCOV 376	base.Extra = extra	×
UNCOV 377	return &model.HashObject{	×
378	BaseObject: base,	×
379	Hash: m,	×
UNCOV 380	FieldExpirations: e,	×
UNCOV 381	}, nil	×
382	}
383	return nil, fmt.Errorf("unknown type flag: %b", flag)	×
384	}
385
386	func (dec *Decoder) parse(cb func(object model.RedisObject) bool) error {	1✔
387	var dbIndex int	1✔
388	var expireMs int64	1✔
389	var lru *int64	1✔
390	var lfu *int64	1✔
391	for {	2✔
392	b, err := dec.readByte()	1✔
393	if err != nil {	1✔
UNCOV 394	return err	×
UNCOV 395	}	×
396	if b == opCodeEOF {	2✔
397	break	1✔
398	} else if b == opCodeSelectDB {	2✔
399	dbIndex64, _, err := dec.readLength()	1✔
400	if err != nil {	1✔
UNCOV 401	return err	×
UNCOV 402	}	×
403	dbIndex = int(dbIndex64)	1✔
404	continue	1✔
405	} else if b == opCodeExpireTime {	1✔
UNCOV 406	err = dec.readFull(dec.buffer[:4])	×
UNCOV 407	if err != nil {	×
UNCOV 408	return err	×
UNCOV 409	}	×
UNCOV 410	expireMs = int64(binary.LittleEndian.Uint32(dec.buffer)) * 1000	×
UNCOV 411	continue	×
412	} else if b == opCodeExpireTimeMs {	2✔
413	err = dec.readFull(dec.buffer)	1✔
414	if err != nil {	1✔
415	return err	×
UNCOV 416	}	×
417	expireMs = int64(binary.LittleEndian.Uint64(dec.buffer))	1✔
418	continue	1✔
419	} else if b == opCodeResizeDB {	2✔
420	keyCount, _, err := dec.readLength()	1✔
421	if err != nil {	1✔
422	return err	×
423	}	×
424	ttlCount, _, err := dec.readLength()	1✔
425	if err != nil {	1✔
426	err = errors.New("Parse Aux value failed: " + err.Error())	×
427	break	×
428	}
429	if dec.withSpecialOpCode {	2✔
430	obj := &model.DBSizeObject{	1✔
431	BaseObject: &model.BaseObject{},	1✔
432	}	1✔
433	obj.DB = dbIndex	1✔
434	obj.KeyCount = keyCount	1✔
435	obj.TTLCount = ttlCount	1✔
436	tbc := cb(obj)	1✔
437	if !tbc {	1✔
438	break	×
439	}
440	}
441	continue	1✔
442	} else if b == opCodeAux {	2✔
443	key, err := dec.readString()	1✔
444	if err != nil {	1✔
445	return err	×
446	}	×
447	value, err := dec.readString()	1✔
448	if err != nil {	1✔
449	err = errors.New("Parse Aux value failed: " + err.Error())	×
450	break	×
451	}
452	if dec.withSpecialOpCode {	2✔
453	obj := &model.AuxObject{	1✔
454	BaseObject: &model.BaseObject{},	1✔
455	}	1✔
456	obj.Type = model.AuxType	1✔
457	obj.Key = unsafeBytes2Str(key)	1✔
458	obj.Value = unsafeBytes2Str(value)	1✔
459	tbc := cb(obj)	1✔
460	if !tbc {	1✔
UNCOV 461	break	×
462	}
463	}
464	continue	1✔
465	} else if b == opCodeFreq {	1✔
UNCOV 466	freq, err := dec.readByte()	×
UNCOV 467	if err != nil {	×
UNCOV 468	return err	×
UNCOV 469	}	×
NEW 470	v := int64(freq)	×
NEW 471	lfu = &v	×
UNCOV 472	continue	×
473	} else if b == opCodeIdle {	1✔
UNCOV 474	idle, _, err := dec.readLength()	×
UNCOV 475	if err != nil {	×
UNCOV 476	return err	×
UNCOV 477	}	×
NEW 478	v := int64(idle)	×
NEW 479	lru = &v	×
UNCOV 480	continue	×
481	} else if b == opCodeModuleAux {	1✔
482	_, _, err = dec.readModuleType()	×
483	if err != nil {	×
484	return err	×
485	}	×
486	continue	×
487	} else if b == opCodeFunction {	2✔
488	functionsLua, err := dec.readString()	1✔
489	if err != nil {	1✔
490	return err	×
491	}	×
492	if dec.withSpecialOpCode {	2✔
493	obj := &model.FunctionsObject{	1✔
494	BaseObject: &model.BaseObject{},	1✔
495	}	1✔
496	obj.Key = "functions"	1✔
497	obj.Type = model.FunctionsType	1✔
498	obj.Encoding = "functions"	1✔
499	obj.FunctionsLua = unsafeBytes2Str(functionsLua)	1✔
500	tbc := cb(obj)	1✔
501	if !tbc {	2✔
502	break	1✔
503	}
504	}
505	continue	1✔
506	} else if b == opCodeSlotInfo {	1✔
NEW 507	if !dec.valkey {	×
NEW 508	return fmt.Errorf("unsupported opcode 244 in Redis RDB version %d", dec.rdbVersion)	×
NEW 509	}	×
510	// Valkey 9+: slot info metadata, safe to skip
UNCOV 511	var err error	×
UNCOV 512	var slot_id, slot_size, expires_slot_size uint64	×
UNCOV 513	slot_id, _, err = dec.readLength()	×
UNCOV 514	if err == nil {	×
UNCOV 515	slot_size, _, err = dec.readLength()	×
UNCOV 516	}	×
UNCOV 517	if err == nil {	×
518	expires_slot_size, _, err = dec.readLength()	×
519	}	×
UNCOV 520	if err != nil {	×
UNCOV 521	return err	×
UNCOV 522	}	×
UNCOV 523	_, _, _ = slot_id, slot_size, expires_slot_size // safe to skip	×
UNCOV 524	continue	×
525	} else if b == opCodeSlotImport { // opcode 243: Valkey=SlotImport, Redis 8.0+=KeyMeta	1✔
NEW 526	if dec.valkey {	×
NEW 527	// Valkey 9+: slot import state	×
NEW 528	job, err := dec.readString()	×
NEW 529	if err != nil {	×
NEW 530	return err	×
UNCOV 531	}	×
NEW 532	num_slot_ranges, _, err := dec.readLength()	×
533	if err != nil {	×
534	return err	×
535	}	×
NEW 536	var slot_from, slot_to uint64	×
NEW 537	ranges := make([]string, num_slot_ranges)	×
NEW 538	for i := uint64(0); i < num_slot_ranges; i++ {	×
NEW 539	slot_from, _, err = dec.readLength()	×
NEW 540	if err == nil {	×
NEW 541	slot_to, _, err = dec.readLength()	×
NEW 542	}	×
NEW 543	if err != nil {	×
NEW 544	return err	×
NEW 545	}	×
NEW 546	ranges[i] = fmt.Sprintf("%d-%d", slot_from, slot_to)	×
547	}
NEW 548	_, _ = job, ranges // safe to skip	×
NEW 549	} else {	×
NEW 550	// Redis 8.0+: RDB_OPCODE_KEY_META (same opcode value 243)	×
NEW 551	// Not yet supported; return error to avoid silent data corruption	×
NEW 552	return fmt.Errorf("unsupported opcode: RDB_OPCODE_KEY_META (243) in Redis RDB version %d", dec.rdbVersion)	×
UNCOV 553	}	×
UNCOV 554	continue	×
555	}
556	key, err := dec.readString()	1✔
557	if err != nil {	1✔
UNCOV 558	return err	×
UNCOV 559	}	×
560	base := &model.BaseObject{	1✔
561	DB: dbIndex,	1✔
562	Key: unsafeBytes2Str(key),	1✔
563	}	1✔
564	if expireMs > 0 {	2✔
565	expiration := time.Unix(0, expireMs*int64(time.Millisecond))	1✔
566	base.Expiration = &expiration	1✔
567	expireMs = 0 // reset expire ms	1✔
568	}	1✔
569	base.IdleTime = lru	1✔
570	lru = nil // reset lru	1✔
571	base.Freq = lfu	1✔
572	lfu = nil // reset lfu	1✔
573	obj, err := dec.readObject(b, base)	1✔
574	if err != nil {	1✔
UNCOV 575	return err	×
UNCOV 576	}	×
577	base.Size = memprofiler.SizeOfObject(obj)	1✔
578	base.Type = obj.GetType()	1✔
579	tbc := cb(obj)	1✔
580	if !tbc {	2✔
581	break	1✔
582	}
583	}
584	// read crc64 at the end
585	_ = dec.readFull(dec.buffer)	1✔
586	return nil	1✔
587	}
588
589	// Parse parses rdb and callback
590	// cb returns true to continue, returns false to stop the iteration
591	func (dec *Decoder) Parse(cb func(object model.RedisObject) bool) (err error) {	1✔
592	defer func() {	2✔
593	if err2 := recover(); err2 != nil {	1✔
UNCOV 594	err = fmt.Errorf("panic: %v", err2)	×
UNCOV 595	}	×
596	}()
597	err = dec.checkHeader()	1✔
598	if err != nil {	1✔
UNCOV 599	return err	×
UNCOV 600	}	×
601	return dec.parse(cb)	1✔
602	}
603
UNCOV 604	func (dec *Decoder) GetReadCount() int {	×
UNCOV 605	return dec.readCount	×
UNCOV 606	}	×

HDT3213 / rdb / 24606763703

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous