diff --git a/README.md b/README.md index b4a36ae..7ca3667 100644 --- a/README.md +++ b/README.md @@ -85,7 +85,7 @@ func main() { ### Verifying a file The following code demonstrates how to verify an RDB file is not corrupt, and -does not exceed the defined limits of the total data, max entry, and max key sizes. +does not exceed the defined limits of the total data, max entry, max value, and max key sizes. ```go import ( @@ -98,6 +98,7 @@ func main() { opts := rdb.VerifyFileOptions{ MaxDataSize: 256 << 20, // 256 MB MaxEntrySize: 100 << 20, // 100 MB + MaxValueSize: 100 << 20, // 100 MB MaxStreamPELSize: 1000, } err := rdb.VerifyFile("/path/to/dump.rdb", opts) @@ -110,7 +111,7 @@ func main() { ### Verifying a reader The following code demonstrates how to verify an io.Reader to reads an RDB file is not corrupt, and -does not exceed the defined limits of the total data, max entry, and max key sizes. +does not exceed the defined limits of the total data, max entry, max value, and max key sizes. ```go import ( @@ -124,6 +125,7 @@ func main() { opts := rdb.VerifyReaderOptions{ MaxDataSize: 256 << 20, // 256 MB MaxEntrySize: 100 << 20, // 100 MB + MaxValueSize: 100 << 20, // 100 MB MaxStreamPELSize: 1000, } @@ -140,7 +142,7 @@ func main() { ### Verifying a value The following code demonstrates how to verify an RDB value is not corrupt, and -does not exceed the defined limits of the max entry size. +does not exceed the defined limits of the max entry and max value sizes. ```go import ( @@ -152,6 +154,7 @@ import ( func main() { opts := rdb.VerifyValueOptions{ MaxEntrySize: 100 << 20, // 100 MB + MaxValueSize: 100 << 20, // 100 MB MaxStreamPELSize: 1000, } payload := []byte{ /*RDB value payload*/ } diff --git a/stream_reader.go b/stream_reader.go index 1ce16f0..99c773b 100644 --- a/stream_reader.go +++ b/stream_reader.go @@ -362,7 +362,9 @@ func (r *valueReader) readStreamEntries(cb func(StreamEntry) error) error { ID: id, Value: fields, } - cb(entry) + if err := cb(entry); err != nil { + return err + } } } else { numFieldsS, err := lpReader.readListpackEntry() @@ -393,7 +395,9 @@ func (r *valueReader) readStreamEntries(cb func(StreamEntry) error) error { ID: id, Value: fields, } - cb(entry) + if err := cb(entry); err != nil { + return err + } } } diff --git a/verify.go b/verify.go index c04448c..dfe2ae1 100644 --- a/verify.go +++ b/verify.go @@ -10,6 +10,7 @@ import ( var defaultMaxDataSize = 256 << 20 // 256 MB var defaultMaxEntrySize = 100 << 20 // 100 MB +var defaultMaxValueSize = 100 << 20 // 100 MB var defaultMaxKeySize = 32 << 10 // 32 KB var defaultMaxStreamPELSize = 1000 var defaultMaxLibrarySize = 100 << 20 // 100 MB @@ -19,6 +20,7 @@ const maxStreamStrSize = math.MaxUint32 type VerifyFileOptions struct { MaxDataSize int MaxEntrySize int + MaxValueSize int MaxKeySize int MaxStreamPELSize int MaxLibrarySize int @@ -35,6 +37,10 @@ func (o *VerifyFileOptions) maybeSetDefaults() { o.MaxEntrySize = defaultMaxEntrySize } + if o.MaxValueSize <= 0 { + o.MaxValueSize = defaultMaxValueSize + } + if o.MaxKeySize <= 0 { o.MaxKeySize = defaultMaxKeySize } @@ -55,6 +61,7 @@ func VerifyFile(path string, opts VerifyFileOptions) error { v := &verifier{ maxDataSize: opts.MaxDataSize, maxEntrySize: opts.MaxEntrySize, + maxValueSize: opts.MaxValueSize, maxKeySize: opts.MaxKeySize, maxStreamPELSize: opts.MaxStreamPELSize, maxLibrarySize: opts.MaxLibrarySize, @@ -82,6 +89,7 @@ func VerifyFile(path string, opts VerifyFileOptions) error { type VerifyReaderOptions struct { MaxDataSize int MaxEntrySize int + MaxValueSize int MaxKeySize int MaxStreamPELSize int MaxLibrarySize int @@ -98,6 +106,10 @@ func (o *VerifyReaderOptions) maybeSetDefaults() { o.MaxEntrySize = defaultMaxEntrySize } + if o.MaxValueSize <= 0 { + o.MaxValueSize = defaultMaxValueSize + } + if o.MaxKeySize <= 0 { o.MaxKeySize = defaultMaxKeySize } @@ -118,6 +130,7 @@ func VerifyReader(r io.Reader, opts VerifyReaderOptions) error { v := &verifier{ maxDataSize: opts.MaxDataSize, maxEntrySize: opts.MaxEntrySize, + maxValueSize: opts.MaxValueSize, maxKeySize: opts.MaxKeySize, maxStreamPELSize: opts.MaxStreamPELSize, maxLibrarySize: opts.MaxLibrarySize, @@ -132,6 +145,7 @@ func VerifyReader(r io.Reader, opts VerifyReaderOptions) error { type VerifyValueOptions struct { MaxEntrySize int + MaxValueSize int MaxStreamPELSize int } @@ -140,6 +154,10 @@ func (o *VerifyValueOptions) maybeSetDefaults() { o.MaxEntrySize = defaultMaxEntrySize } + if o.MaxValueSize <= 0 { + o.MaxValueSize = defaultMaxValueSize + } + if o.MaxStreamPELSize <= 0 { o.MaxStreamPELSize = defaultMaxStreamPELSize } @@ -151,6 +169,7 @@ func VerifyValue(payload []byte, opts VerifyValueOptions) error { opts.maybeSetDefaults() v := &verifier{ maxEntrySize: opts.MaxEntrySize, + maxValueSize: opts.MaxValueSize, maxStreamPELSize: opts.MaxStreamPELSize, // We don't care about the values below, as they don't // really apply to RDB values. @@ -170,6 +189,10 @@ func errMaxEntrySizeExceeded(current int, limit int) error { return fmt.Errorf("max entry size is exceeded. current: %d, limit: %d", current, limit) } +func errMaxValueSizeExceeded(current int, limit int) error { + return fmt.Errorf("max value size is exceeded. current: %d, limit: %d", current, limit) +} + func errMaxKeySizeExceeded(current int, limit int) error { return fmt.Errorf("max key size is exceeded. current: %d, limit: %d", current, limit) } @@ -189,6 +212,7 @@ func errMaxLibrarySizeExceeded(current int, limit int) error { type verifier struct { maxDataSize int maxEntrySize int + maxValueSize int maxKeySize int maxStreamPELSize int allowPartialVerify bool @@ -203,6 +227,10 @@ func (v *verifier) HandleString(key string, value string) error { return errMaxKeySizeExceeded(len(key), v.maxKeySize) } + if len(value) > v.maxValueSize { + return errMaxValueSizeExceeded(len(value), v.maxValueSize) + } + if len(value) > v.maxEntrySize { return errMaxEntrySizeExceeded(len(value), v.maxEntrySize) } @@ -232,6 +260,10 @@ func (v *verifier) HashEntryHandler(key string) func(field string, value string) var entrySize int return func(field, value string) error { elementSize := len(field) + len(value) + if elementSize > v.maxValueSize { + return errMaxValueSizeExceeded(elementSize, v.maxValueSize) + } + entrySize += elementSize if entrySize > v.maxEntrySize { return errMaxEntrySizeExceeded(entrySize, v.maxEntrySize) @@ -263,6 +295,10 @@ func (v *verifier) HashWithExpEntryHandler(key string) func(field string, value var entrySize int return func(field, value string, exp time.Time) error { elementSize := len(field) + len(value) + 8 + if elementSize > v.maxValueSize { + return errMaxValueSizeExceeded(elementSize, v.maxValueSize) + } + entrySize += elementSize if entrySize > v.maxEntrySize { return errMaxEntrySizeExceeded(entrySize, v.maxEntrySize) @@ -294,6 +330,10 @@ func (v *verifier) ListEntryHandler(key string) func(elem string) error { var entrySize int return func(elem string) error { elementSize := len(elem) + if elementSize > v.maxValueSize { + return errMaxValueSizeExceeded(elementSize, v.maxValueSize) + } + entrySize += elementSize if entrySize > v.maxEntrySize { return errMaxEntrySizeExceeded(entrySize, v.maxEntrySize) @@ -325,6 +365,10 @@ func (v *verifier) SetEntryHandler(key string) func(elem string) error { var entrySize int return func(elem string) error { elementSize := len(elem) + if elementSize > v.maxValueSize { + return errMaxValueSizeExceeded(elementSize, v.maxValueSize) + } + entrySize += elementSize if entrySize > v.maxEntrySize { return errMaxEntrySizeExceeded(entrySize, v.maxEntrySize) @@ -356,6 +400,10 @@ func (v *verifier) ZsetEntryHandler(key string) func(elem string, score float64) var entrySize int return func(elem string, score float64) error { elementSize := len(elem) + 8 + if elementSize > v.maxValueSize { + return errMaxValueSizeExceeded(elementSize, v.maxValueSize) + } + entrySize += elementSize if entrySize > v.maxEntrySize { return errMaxEntrySizeExceeded(entrySize, v.maxEntrySize) @@ -387,6 +435,10 @@ func (v *verifier) ArrayEntryHandler(key string) func(index uint64, value string var entrySize int return func(index uint64, value string) error { elementSize := len(value) + 8 // 8: index + if elementSize > v.maxValueSize { + return errMaxValueSizeExceeded(elementSize, v.maxValueSize) + } + entrySize += elementSize if entrySize > v.maxEntrySize { return errMaxEntrySizeExceeded(entrySize, v.maxEntrySize) @@ -406,6 +458,10 @@ func (v *verifier) HandleModule(key string, value string, marker ModuleMarker) e return errMaxKeySizeExceeded(len(key), v.maxKeySize) } + if len(value) > v.maxValueSize { + return errMaxValueSizeExceeded(len(value), v.maxValueSize) + } + if len(value) > v.maxEntrySize { return errMaxEntrySizeExceeded(len(value), v.maxEntrySize) } @@ -442,11 +498,17 @@ func (v *verifier) StreamEntryHandler(key string) func(entry StreamEntry) error valueSize += len(value) } + entrySize := valueSize + 16 // 8: ID#Seq + 8: ID#Millis + if entrySize > v.maxValueSize { + return errMaxValueSizeExceeded(entrySize, v.maxValueSize) + } + // we don't check for the max entry size here as we store // stream entries on disk. + // the max value size is still checked above, + // as it applies to a single stream entry. - v.dataSize += valueSize - v.dataSize += 16 // 8: ID#Seq + 8: ID#Millis + v.dataSize += entrySize if v.dataSize > v.maxDataSize { return errMaxDataSizeExceeded(v.dataSize, v.maxDataSize) } @@ -477,15 +539,21 @@ func (v *verifier) StreamGroupHandler(key string) func(group StreamConsumerGroup } for _, pe := range consumer.PendingEntries { - groupSize += 32 // 8: ID#Seq + 8: ID#Millis + 8: DeliveryCount + 8: DeliveryTime + pendingEntrySize := 32 // 8: ID#Seq + 8: ID#Millis + 8: DeliveryCount + 8: DeliveryTime for _, val := range pe.Entry.Value { if len(val) > maxStreamStrSize { return errMaxStreamStrSizeExceeded(len(val), maxStreamStrSize) } - groupSize += len(val) + pendingEntrySize += len(val) } + + if pendingEntrySize > v.maxValueSize { + return errMaxValueSizeExceeded(pendingEntrySize, v.maxValueSize) + } + + groupSize += pendingEntrySize } } entrySize += groupSize diff --git a/verify_test.go b/verify_test.go index efd4c07..438103a 100644 --- a/verify_test.go +++ b/verify_test.go @@ -71,6 +71,13 @@ func TestVerifyFile_maxEntrySize(t *testing.T) { require.ErrorContains(t, err, "max entry size") } +func TestVerifyFile_maxValueSize(t *testing.T) { + err := VerifyFile(allTypesRDBPath, VerifyFileOptions{ + MaxValueSize: 5, + }) + require.ErrorContains(t, err, "max value size") +} + func TestVerifyFile_maxKeySize(t *testing.T) { err := VerifyFile(allTypesRDBPath, VerifyFileOptions{ MaxKeySize: 1, @@ -114,6 +121,46 @@ func TestVerifyValue_maxEntrySize(t *testing.T) { require.ErrorContains(t, err, "max entry size") } +func TestVerifyValue_maxValueSize(t *testing.T) { + dump, err := os.ReadFile(stringRDBValuePath) + require.NoError(t, err) + + err = VerifyValue(dump, VerifyValueOptions{ + MaxValueSize: 12, + }) + require.ErrorContains(t, err, "max value size") +} + +func TestVerifyValue_maxValueSize_streamEntry(t *testing.T) { + // The stream has no pending entries, so the limit must be enforced by the + // stream entry callback itself. Its first entry is 47 bytes: 31 bytes of + // fields and values plus the 16-byte stream ID. + dump, err := os.ReadFile(filepath.Join(valueDumpsPath, "stream-listpacks.bin")) + require.NoError(t, err) + + err = VerifyValue(dump, VerifyValueOptions{ + MaxValueSize: 46, + }) + require.ErrorContains(t, err, "max value size") +} + +func TestVerifyValue_maxValueSize_arrayElement(t *testing.T) { + // The array's largest element is "-2305843009213693952", so its element + // size is 28 bytes: 20 bytes of value plus the 8-byte index. + dump, err := os.ReadFile(filepath.Join(valueDumpsPath, "array.bin")) + require.NoError(t, err) + + err = VerifyValue(dump, VerifyValueOptions{ + MaxValueSize: 28, + }) + require.NoError(t, err) + + err = VerifyValue(dump, VerifyValueOptions{ + MaxValueSize: 27, + }) + require.ErrorContains(t, err, "max value size") +} + func TestVerifyValue_maxStreamPELSize(t *testing.T) { dump, err := os.ReadFile(streamWithPELRDBValuePath) require.NoError(t, err) @@ -172,6 +219,19 @@ func TestVerifyReader_maxEntrySize(t *testing.T) { require.ErrorContains(t, err, "max entry size") } +func TestVerifyReader_maxValueSize(t *testing.T) { + file, err := os.Open(allTypesRDBPath) + require.NoError(t, err) + t.Cleanup(func() { + _ = file.Close() + }) + + err = VerifyReader(file, VerifyReaderOptions{ + MaxValueSize: 5, + }) + require.ErrorContains(t, err, "max value size") +} + func TestVerifyReader_maxKeySize(t *testing.T) { file, err := os.Open(allTypesRDBPath) require.NoError(t, err) @@ -255,6 +315,7 @@ func TestVerifier_String_MaxDataSize(t *testing.T) { v := &verifier{ maxDataSize: 3, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 100, } @@ -268,6 +329,7 @@ func TestVerifier_String_MaxKeySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 2, } @@ -278,16 +340,30 @@ func TestVerifier_String_MaxEntrySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 2, + maxValueSize: 10000, maxKeySize: 100, } require.ErrorContains(t, v.HandleString("k", "longvalue"), "max entry size") } +func TestVerifier_String_MaxValueSize(t *testing.T) { + v := &verifier{ + maxDataSize: 100, + maxEntrySize: 100, + maxValueSize: 2, + maxKeySize: 100, + } + + require.NoError(t, v.HandleString("k", "va")) + require.ErrorContains(t, v.HandleString("k", "longvalue"), "max value size") +} + func TestVerifier_HashEntryHandler_MaxDataSize(t *testing.T) { v := &verifier{ maxDataSize: 5, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 100, maxStreamPELSize: 100, } @@ -304,6 +380,7 @@ func TestVerifier_HashEntryHandler_MaxKeySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 2, } @@ -315,6 +392,7 @@ func TestVerifier_HashEntryHandler_MaxEntrySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 3, + maxValueSize: 10000, maxKeySize: 100, } @@ -323,10 +401,26 @@ func TestVerifier_HashEntryHandler_MaxEntrySize(t *testing.T) { require.ErrorContains(t, h("x", "y"), "max entry size") } +func TestVerifier_HashEntryHandler_MaxValueSize(t *testing.T) { + v := &verifier{ + maxDataSize: 100, + maxEntrySize: 100, + maxValueSize: 2, + maxKeySize: 100, + } + + h := v.HashEntryHandler("k") + // the limit is per field and value pair, not cumulative + require.NoError(t, h("f", "v")) + require.NoError(t, h("x", "y")) + require.ErrorContains(t, h("f", "longvalue"), "max value size") +} + func TestVerifier_HashWithExpEntryHandler_MaxDataSize(t *testing.T) { v := &verifier{ maxDataSize: 21, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 100, maxStreamPELSize: 100, } @@ -343,6 +437,7 @@ func TestVerifier_HashWithExpEntryHandler_MaxKeySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 2, } @@ -354,6 +449,7 @@ func TestVerifier_HashWithExpEntryHandler_MaxEntrySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 10, + maxValueSize: 10000, maxKeySize: 100, } @@ -364,10 +460,27 @@ func TestVerifier_HashWithExpEntryHandler_MaxEntrySize(t *testing.T) { require.ErrorContains(t, h("x", "y", time.Now()), "max entry size") } +func TestVerifier_HashWithExpEntryHandler_MaxValueSize(t *testing.T) { + v := &verifier{ + maxDataSize: 100, + maxEntrySize: 100, + maxValueSize: 10, + maxKeySize: 100, + } + + h := v.HashWithExpEntryHandler("k") + // "f" + "v" + 8 = 10, at limit + require.NoError(t, h("f", "v", time.Now())) + require.NoError(t, h("x", "y", time.Now())) + // "f" + "vv" + 8 = 11, exceeds 10 + require.ErrorContains(t, h("f", "vv", time.Now()), "max value size") +} + func TestVerifier_ListEntryHandler_MaxDataSize(t *testing.T) { v := &verifier{ maxDataSize: 3, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 100, maxStreamPELSize: 100, } @@ -384,6 +497,7 @@ func TestVerifier_ListEntryHandler_MaxKeySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 2, } @@ -395,6 +509,7 @@ func TestVerifier_ListEntryHandler_MaxEntrySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 1, + maxValueSize: 10000, maxKeySize: 100, } @@ -403,10 +518,25 @@ func TestVerifier_ListEntryHandler_MaxEntrySize(t *testing.T) { require.ErrorContains(t, h("b"), "max entry size") } +func TestVerifier_ListEntryHandler_MaxValueSize(t *testing.T) { + v := &verifier{ + maxDataSize: 100, + maxEntrySize: 100, + maxValueSize: 1, + maxKeySize: 100, + } + + h := v.ListEntryHandler("k") + require.NoError(t, h("a")) + require.NoError(t, h("b")) + require.ErrorContains(t, h("cc"), "max value size") +} + func TestVerifier_SetEntryHandler_MaxDataSize(t *testing.T) { v := &verifier{ maxDataSize: 3, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 100, maxStreamPELSize: 100, } @@ -423,6 +553,7 @@ func TestVerifier_SetEntryHandler_MaxKeySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 2, } @@ -434,6 +565,7 @@ func TestVerifier_SetEntryHandler_MaxEntrySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 1, + maxValueSize: 10000, maxKeySize: 100, } @@ -442,10 +574,25 @@ func TestVerifier_SetEntryHandler_MaxEntrySize(t *testing.T) { require.ErrorContains(t, h("b"), "max entry size") } +func TestVerifier_SetEntryHandler_MaxValueSize(t *testing.T) { + v := &verifier{ + maxDataSize: 100, + maxEntrySize: 100, + maxValueSize: 1, + maxKeySize: 100, + } + + h := v.SetEntryHandler("k") + require.NoError(t, h("a")) + require.NoError(t, h("b")) + require.ErrorContains(t, h("cc"), "max value size") +} + func TestVerifier_ZsetEntryHandler_MaxDataSize(t *testing.T) { v := &verifier{ maxDataSize: 19, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 100, maxStreamPELSize: 100, } @@ -462,6 +609,7 @@ func TestVerifier_ZsetEntryHandler_MaxKeySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 2, } @@ -473,6 +621,7 @@ func TestVerifier_ZsetEntryHandler_MaxEntrySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 9, + maxValueSize: 10000, maxKeySize: 100, } @@ -483,10 +632,43 @@ func TestVerifier_ZsetEntryHandler_MaxEntrySize(t *testing.T) { require.ErrorContains(t, h("b", 2), "max entry size") } +func TestVerifier_ZsetEntryHandler_MaxValueSize(t *testing.T) { + v := &verifier{ + maxDataSize: 100, + maxEntrySize: 100, + maxValueSize: 9, + maxKeySize: 100, + } + + h := v.ZsetEntryHandler("k") + // "a" + 8 = 9, at limit + require.NoError(t, h("a", 1)) + require.NoError(t, h("b", 2)) + // "cc" + 8 = 10, exceeds 9 + require.ErrorContains(t, h("cc", 3), "max value size") +} + +func TestVerifier_ArrayEntryHandler_MaxValueSize(t *testing.T) { + v := &verifier{ + maxDataSize: 100, + maxEntrySize: 100, + maxValueSize: 9, + maxKeySize: 100, + } + + h := v.ArrayEntryHandler("k") + // "a" + 8 = 9, at limit + require.NoError(t, h(0, "a")) + require.NoError(t, h(1, "b")) + // "cc" + 8 = 10, exceeds 9 + require.ErrorContains(t, h(2, "cc"), "max value size") +} + func TestVerifier_StreamEntryHandler_MaxDataSize(t *testing.T) { v := &verifier{ maxDataSize: 42, maxEntrySize: 1000, + maxValueSize: 10000, maxKeySize: 100, maxStreamPELSize: 10, } @@ -503,6 +685,7 @@ func TestVerifier_StreamEntryHandler_MaxKeySize(t *testing.T) { v := &verifier{ maxDataSize: 100, maxEntrySize: 1000, + maxValueSize: 10000, maxKeySize: 2, } @@ -510,10 +693,28 @@ func TestVerifier_StreamEntryHandler_MaxKeySize(t *testing.T) { require.ErrorContains(t, h(StreamEntry{Value: []string{"a"}}), "max key size") } +func TestVerifier_StreamEntryHandler_MaxValueSize(t *testing.T) { + v := &verifier{ + maxDataSize: 10000, + maxEntrySize: 1000, + maxValueSize: 18, + maxKeySize: 100, + maxStreamPELSize: 10, + } + + h := v.StreamEntryHandler("stream") + // "a" + "b" + 16 = 18, at limit + require.NoError(t, h(StreamEntry{Value: []string{"a", "b"}})) + require.NoError(t, h(StreamEntry{Value: []string{"x", "y"}})) + // "a" + "bb" + 16 = 19, exceeds 18 + require.ErrorContains(t, h(StreamEntry{Value: []string{"a", "bb"}}), "max value size") +} + func TestVerifier_StreamGroupHandler_MaxDataSize(t *testing.T) { v := &verifier{ maxDataSize: 156, maxEntrySize: 1000, + maxValueSize: 10000, maxKeySize: 100, maxStreamPELSize: 10, } @@ -561,6 +762,7 @@ func TestVerifier_StreamGroupHandler_MaxEntrySize(t *testing.T) { v := &verifier{ maxDataSize: 10000, maxEntrySize: 100, + maxValueSize: 10000, maxKeySize: 100, maxStreamPELSize: 10, } @@ -583,10 +785,42 @@ func TestVerifier_StreamGroupHandler_MaxEntrySize(t *testing.T) { require.ErrorContains(t, h(group), "max entry size") } +func TestVerifier_StreamGroupHandler_MaxValueSize(t *testing.T) { + // pending entry size: 32 + value "a" (1) = 33 + v := &verifier{ + maxDataSize: 10000, + maxEntrySize: 10000, + maxValueSize: 33, + maxKeySize: 100, + maxStreamPELSize: 10, + } + + h := v.StreamGroupHandler("stream") + + newGroup := func(value string) StreamConsumerGroup { + return StreamConsumerGroup{ + Name: "g", + Consumers: []StreamConsumer{ + { + Name: "c", + PendingEntries: []*StreamPendingEntry{ + {Entry: StreamEntry{Value: []string{value}}}, + }, + }, + }, + } + } + + require.NoError(t, h(newGroup("a"))) + require.NoError(t, h(newGroup("b"))) + require.ErrorContains(t, h(newGroup("cc")), "max value size") +} + func TestVerifier_StreamGroupHandler_MaxStreamPELSize(t *testing.T) { v := &verifier{ maxDataSize: 10000, maxEntrySize: 10000, + maxValueSize: 10000, maxKeySize: 100, maxStreamPELSize: 0, }