bugfix: sequence delta of stream item may be negative(#27)

HDT3213 · Jan 4, 2024 · 0532e58 · 0532e58
1 parent 4c2b9a3
commit 0532e58
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 12 deletions.
diff --git a/cases/issue27.json b/cases/issue27.json
diff --git a/cases/issue27.rdb b/cases/issue27.rdb
diff --git a/core/listpack.go b/core/listpack.go
@@ -158,12 +158,12 @@ func (dec *Decoder) readListPackEntry(buf []byte, cursor *int) ([]byte, uint32,
 	return nil, 0, fmt.Errorf("unknown entry header")
 }
 
-func (dec *Decoder) readListPackEntryAsUint(buf []byte, cursor *int) (uint64, error) {
+func (dec *Decoder) readListPackEntryAsInt(buf []byte, cursor *int) (int64, error) {
 	bin, _, err := dec.readListPackEntry(buf, cursor)
 	if err != nil {
 		return 0, fmt.Errorf("read from failed: %v", err)
 	}
-	i, err := strconv.ParseUint(string(bin), 10, 64)
+	i, err := strconv.ParseInt(string(bin), 10, 64)
 	if err != nil {
 		return 0, fmt.Errorf("%s is not a uint", string(bin))
 	}

diff --git a/core/stream.go b/core/stream.go
@@ -3,6 +3,7 @@ package core
 import (
 	"encoding/binary"
 	"fmt"
+
 	"github.com/hdt3213/rdb/model"
 )
 
@@ -123,17 +124,17 @@ func (dec *Decoder) readStreamEntries() ([]*model.StreamEntry, error) {
 // readStreamEntryContent read messages in a stream entry
 func (dec *Decoder) readStreamEntryContent(buf []byte, cursor *int, firstId *model.StreamId) (*model.StreamEntry, error) {
 	// read count
-	count, err := dec.readListPackEntryAsUint(buf, cursor)
+	count, err := dec.readListPackEntryAsInt(buf, cursor)
 	if err != nil {
 		return nil, fmt.Errorf("read stream entry count failed: %v", err)
 	}
-	deleted, err := dec.readListPackEntryAsUint(buf, cursor)
+	deleted, err := dec.readListPackEntryAsInt(buf, cursor)
 	if err != nil {
 		return nil, fmt.Errorf("read stream entry deleted count failed: %v", err)
 	}
 
 	// read field names of master entry
-	fieldNum0, err := dec.readListPackEntryAsUint(buf, cursor)
+	fieldNum0, err := dec.readListPackEntryAsInt(buf, cursor)
 	if err != nil {
 		return nil, fmt.Errorf("read stream field number failed: %v", err)
 	}
@@ -153,26 +154,27 @@ func (dec *Decoder) readStreamEntryContent(buf []byte, cursor *int, firstId *mod
 
 	total := count + deleted
 	msgs := make([]*model.StreamMessage, 0, total)
-	for i := uint64(0); i < total; i++ {
-		flag, err := dec.readListPackEntryAsUint(buf, cursor)
+	for i := int64(0); i < total; i++ {
+		flag, err := dec.readListPackEntryAsInt(buf, cursor)
 		if err != nil {
 			return nil, fmt.Errorf("read stream item flag failed: %v", err)
 		}
-		ms, err := dec.readListPackEntryAsUint(buf, cursor)
+		ms, err := dec.readListPackEntryAsInt(buf, cursor)
 		if err != nil {
 			return nil, fmt.Errorf("read stream item id ms failed: %v", err)
 		}
-		seq, err := dec.readListPackEntryAsUint(buf, cursor)
+		seq, err := dec.readListPackEntryAsInt(buf, cursor)
 		if err != nil {
 			return nil, fmt.Errorf("read stream item id seq failed: %v", err)
 		}
+		// ms and seq may be negative
 		msgId := &model.StreamId{
-			Ms:       ms + firstId.Ms,
-			Sequence: seq + firstId.Sequence,
+			Ms:       uint64(ms + int64(firstId.Ms)),
+			Sequence: uint64(seq + int64(firstId.Sequence)),
 		}
 		fieldNum := masterFieldNum
 		if flag&StreamItemFlagSameFields == 0 {
-			fieldNum0, err := dec.readListPackEntryAsUint(buf, cursor)
+			fieldNum0, err := dec.readListPackEntryAsInt(buf, cursor)
 			if err != nil {
 				return nil, fmt.Errorf("read stream item field number failed: %v", err)
 			}

diff --git a/helper/converter_test.go b/helper/converter_test.go
@@ -66,6 +66,7 @@ func TestToJson(t *testing.T) {
 		}
 	}()
 	testCases := []string{
+		"issue27",
 		"set_listpack",
 		"stream_listpacks_2",
 		"stream_listpacks_1",