From b22fe83f61fae5479e1a2279e454ae1a3f4e2742 Mon Sep 17 00:00:00 2001 From: Sahil Rajput <90515841+SahilRajputdotexe@users.noreply.github.com> Date: Thu, 1 Oct 2026 13:43:17 +0530 Subject: [PATCH] feat: add stream consumer group backlog report Add StreamGroup.PendingOwners, StreamObject.UndeliveredMessages and StreamObject.ConsumerBacklog to relate a group's pending entries to the consumers holding them, and a streambacklog command that writes the result as CSV. --- README.md | 32 ++++- README_CN.md | 4 +- cmd.go | 6 +- helper/streambacklog.go | 71 +++++++++++ helper/streambacklog_test.go | 145 +++++++++++++++++++++++ model/stream.go | 200 ++++++++++++++++++++++++++++++- model/stream_test.go | 220 +++++++++++++++++++++++++++++++++++ 7 files changed, 672 insertions(+), 6 deletions(-) create mode 100644 helper/streambacklog.go create mode 100644 helper/streambacklog_test.go create mode 100644 model/stream_test.go diff --git a/README.md b/README.md index c7fb17f..a5ba19c 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,7 @@ It provides abilities to: - Convert RDB files to Redis Serialization Protocol (or AOF file) - Find the biggest N keys in RDB files - Find the hottest N keys by LFU frequency in RDB files +- Report the pending backlog of stream consumer groups - Draw FlameGraph to analysis which kind of keys occupied most memory - Customize data usage - Generate RDB file @@ -54,7 +55,7 @@ use `rdb` command in terminal, you can see it's manual $ rdb This is a tool to parse Redis' RDB files Options: - -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph + -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph/streambacklog -o output file path -n number of result, using in command: bigkey/hotkey/prefix -port listen port for flame graph web service @@ -96,6 +97,8 @@ parameters between '[' and ']' is optional rdb -c flamegraph [-port 16379] [-sep :] dump.rdb 7. get hottest keys by LFU frequency (requires maxmemory-policy allkeys-lfu/volatile-lfu) rdb -c hotkey [-o hotkey.csv] [-n 50] dump.rdb +8. report the pending backlog of every stream consumer group + rdb -c streambacklog [-o backlog.csv] dump.rdb ``` # Convert to Json @@ -561,6 +564,33 @@ database,key,type,size,size_readable,freq Note: Keys without LFU information are skipped. If the RDB file was not generated under LFU eviction policy, the output will be empty. +# Stream Consumer Group Backlog + +For every consumer group of every stream, RDB can report how far the group is behind: + +``` +rdb -c streambacklog [-o ] +``` + +Example: + +``` +rdb -c streambacklog cases/stream_listpacks_1.rdb +``` + +The CSV has one row for each consumer that holds pending entries, ordered by group name, then by descending `pending`. +Pending entries that no consumer holds are collected into one row with an empty `consumer`, and a group without pending entries still gets a single row. +`pending` is the number of entries the consumer holds, `oldest_delivery_time` and `max_delivery_count` are the smallest delivery time and the largest delivery count among them. +`undelivered` is the number of messages the group has never been delivered, and `dangling` is the number of its pending entries whose message was deleted or is no longer in the stream. +The last two columns describe the whole group, so every row of a group carries the same values. + +```csv +database,key,group,consumer,pending,oldest_delivery_time,max_delivery_count,undelivered,dangling +0,listpack,g1,c1,2,1528516636879,1,144,0 +0,listpack,g1,c2,2,1528516649782,1,144,0 +0,listpack,g4,,0,0,0,0,0 +``` + # Convert to AOF Usage: diff --git a/README_CN.md b/README_CN.md index 8886c4d..b8749ca 100644 --- a/README_CN.md +++ b/README_CN.md @@ -40,7 +40,7 @@ go install github.com/hdt3213/rdb@latest $ rdb This is a tool to parse Redis' RDB files Options: - -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph + -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph/streambacklog -o output file path -n number of result, using in command: bigkey/hotkey/prefix -port listen port for flame graph web service @@ -82,6 +82,8 @@ parameters between '[' and ']' is optional rdb -c flamegraph [-port 16379] [-sep :] dump.rdb 7. get hottest keys by LFU frequency (requires maxmemory-policy allkeys-lfu/volatile-lfu) rdb -c hotkey [-o hotkey.csv] [-n 50] dump.rdb +8. report the pending backlog of every stream consumer group + rdb -c streambacklog [-o backlog.csv] dump.rdb ``` # 转换为 JSON 格式 diff --git a/cmd.go b/cmd.go index 795f131..b42ff40 100644 --- a/cmd.go +++ b/cmd.go @@ -12,7 +12,7 @@ import ( const help = ` This is a tool to parse Redis' RDB files Options: - -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph + -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph/streambacklog -o output file path -n number of result, using in command: bigkey/hotkey/prefix -port listen port for flame graph web service @@ -54,6 +54,8 @@ parameters between '[' and ']' is optional rdb -c flamegraph [-port 16379] [-sep :] dump.rdb 7. get hottest keys by LFU frequency (requires maxmemory-policy allkeys-lfu/volatile-lfu) rdb -c hotkey [-o hotkey.csv] [-n 50] dump.rdb +8. report the pending backlog of every stream consumer group + rdb -c streambacklog [-o backlog.csv] dump.rdb ` type separators []string @@ -158,6 +160,8 @@ func main() { } else { err = helper.PrefixAnalyse(src, n, maxDepth, outputFile, options...) } + case "streambacklog": + err = helper.StreamBacklogReport(src, outputFile, options...) case "flamegraph": _, err = helper.FlameGraph(src, port, seps, options...) if err != nil { diff --git a/helper/streambacklog.go b/helper/streambacklog.go new file mode 100644 index 0000000..ba5be77 --- /dev/null +++ b/helper/streambacklog.go @@ -0,0 +1,71 @@ +package helper + +import ( + "encoding/csv" + "errors" + "fmt" + "os" + "strconv" + + "github.com/hdt3213/rdb/core" + "github.com/hdt3213/rdb/model" +) + +// StreamBacklogReport writes a CSV report of the pending backlog of every consumer group of every stream in the rdb file +func StreamBacklogReport(rdbFilename string, output *os.File, options ...interface{}) error { + if rdbFilename == "" { + return errors.New("src file path is required") + } + rdbFile, err := os.Open(rdbFilename) + if err != nil { + return fmt.Errorf("open rdb %s failed, %v", rdbFilename, err) + } + defer func() { + _ = rdbFile.Close() + }() + var dec decoder = core.NewDecoder(rdbFile) + if dec, err = wrapDecoder(dec, options...); err != nil { + return err + } + _, err = output.WriteString("database,key,group,consumer,pending,oldest_delivery_time,max_delivery_count,undelivered,dangling\n") + if err != nil { + return fmt.Errorf("write header failed: %v", err) + } + csvWriter := csv.NewWriter(output) + defer csvWriter.Flush() + var writeErr error + err = dec.Parse(func(object model.RedisObject) bool { + stream, ok := object.(*model.StreamObject) + if !ok { + return true + } + for _, row := range stream.ConsumerBacklog() { + writeErr = csvWriter.Write([]string{ + strconv.Itoa(object.GetDBIndex()), + object.GetKey(), + row.Group, + row.Consumer, + strconv.Itoa(row.Pending), + strconv.FormatUint(row.OldestDeliveryTime, 10), + strconv.FormatUint(row.MaxDeliveryCount, 10), + strconv.Itoa(row.Undelivered), + strconv.Itoa(row.Dangling), + }) + if writeErr != nil { + return false + } + } + return true + }) + if writeErr != nil { + return fmt.Errorf("csv write failed: %v", writeErr) + } + if err != nil { + return err + } + csvWriter.Flush() + if err = csvWriter.Error(); err != nil { + return fmt.Errorf("csv write failed: %v", err) + } + return nil +} diff --git a/helper/streambacklog_test.go b/helper/streambacklog_test.go new file mode 100644 index 0000000..1dd3915 --- /dev/null +++ b/helper/streambacklog_test.go @@ -0,0 +1,145 @@ +package helper + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "github.com/hdt3213/rdb/core" + "github.com/hdt3213/rdb/model" +) + +const streamBacklogHeader = "database,key,group,consumer,pending,oldest_delivery_time,max_delivery_count,undelivered,dangling" + +func runStreamBacklogReport(t *testing.T, srcRdb string, options ...interface{}) []string { + t.Helper() + outputPath := filepath.Join(t.TempDir(), "backlog.csv") + output, err := os.Create(outputPath) + if err != nil { + t.Fatalf("create output file failed: %v", err) + } + err = StreamBacklogReport(srcRdb, output, options...) + if closeErr := output.Close(); closeErr != nil { + t.Fatalf("close output failed: %v", closeErr) + } + if err != nil { + t.Fatalf("StreamBacklogReport failed: %v", err) + } + content, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read output failed: %v", err) + } + return strings.Split(strings.TrimRight(string(content), "\n"), "\n") +} + +func assertLines(t *testing.T, got, want []string) { + t.Helper() + if strings.Join(got, "\n") != strings.Join(want, "\n") { + t.Errorf("report =\n%s\nwant\n%s", strings.Join(got, "\n"), strings.Join(want, "\n")) + } +} + +func TestStreamBacklogReport(t *testing.T) { + got := runStreamBacklogReport(t, filepath.Join("../cases", "stream_listoacks_3.rdb")) + assertLines(t, got, []string{ + streamBacklogHeader, + "0,mystream,consumer-group-name,consumer-name,1,1704557998397,1,0,0", + }) + + got = runStreamBacklogReport(t, filepath.Join("../cases", "stream_listpacks_1.rdb")) + assertLines(t, got, []string{ + streamBacklogHeader, + "0,listpack,g1,c1,2,1528516636879,1,144,0", + "0,listpack,g1,c2,2,1528516649782,1,144,0", + "0,listpack,g2,c1,1,1528516695691,1,83,0", + "0,listpack,g3,c1,2,1528516699993,1,81,0", + "0,listpack,g4,,0,0,0,0,0", + }) + + got = runStreamBacklogReport(t, filepath.Join("../cases", "stream_listpacks_2.rdb")) + assertLines(t, got, []string{streamBacklogHeader}) +} + +func TestStreamBacklogReportWithRegex(t *testing.T) { + srcRdb := filepath.Join("../cases", "stream_listpacks_1.rdb") + got := runStreamBacklogReport(t, srcRdb, WithRegexOption("^listpack$")) + if len(got) != 6 { + t.Errorf("got %d lines, want the header and five rows", len(got)) + } + got = runStreamBacklogReport(t, srcRdb, WithRegexOption("^no-such-key$")) + assertLines(t, got, []string{streamBacklogHeader}) +} + +func TestStreamBacklogReportCsvQuotingAndOrder(t *testing.T) { + newStream := func(key, group string, consumers ...*model.StreamConsumer) *model.StreamObject { + return &model.StreamObject{ + BaseObject: &model.BaseObject{Key: key}, + Version: 3, + Length: 2, + LastId: &model.StreamId{Ms: 2}, + Entries: []*model.StreamEntry{{ + FirstMsgId: &model.StreamId{Ms: 1}, + Fields: []string{"k"}, + Msgs: []*model.StreamMessage{ + {Id: &model.StreamId{Ms: 1}, Fields: map[string]string{"k": "v"}}, + {Id: &model.StreamId{Ms: 2}, Fields: map[string]string{"k": "v"}}, + }, + }}, + Groups: []*model.StreamGroup{{ + Name: group, + LastId: &model.StreamId{Ms: 1}, + Pending: []*model.StreamNAck{{Id: &model.StreamId{Ms: 1}, DeliveryTime: 111, DeliveryCount: 2}}, + Consumers: consumers, + }}, + } + } + pending := []*model.StreamId{{Ms: 1}} + quoted := newStream("b,stream", "g,1", + &model.StreamConsumer{Name: `c"1`, Pending: pending}, + &model.StreamConsumer{Name: "c2", Pending: pending}, + ) + plain := newStream("zeta", "zg", &model.StreamConsumer{Name: "zc", Pending: pending}) + + srcRdb := filepath.Join(t.TempDir(), "streams.rdb") + file, err := os.Create(srcRdb) + if err != nil { + t.Fatalf("create rdb failed: %v", err) + } + enc := core.NewEncoder(file) + steps := []func() error{ + enc.WriteHeader, + func() error { return enc.WriteDBHeader(0, 2, 0) }, + func() error { return enc.WriteStringObject("plain", []byte("v")) }, + func() error { return enc.WriteStreamObject(quoted.Key, quoted) }, + func() error { return enc.WriteDBHeader(3, 1, 0) }, + func() error { return enc.WriteStreamObject(plain.Key, plain) }, + enc.WriteEnd, + } + for i, step := range steps { + if err := step(); err != nil { + t.Fatalf("write rdb step %d failed: %v", i, err) + } + } + if err := file.Close(); err != nil { + t.Fatalf("close rdb failed: %v", err) + } + + got := runStreamBacklogReport(t, srcRdb) + assertLines(t, got, []string{ + streamBacklogHeader, + `0,"b,stream","g,1","c""1",1,111,2,1,0`, + `0,"b,stream","g,1",c2,1,111,2,1,0`, + `3,zeta,zg,zc,1,111,2,1,0`, + }) +} + +func TestStreamBacklogReportErrors(t *testing.T) { + err := StreamBacklogReport("", os.Stdout) + if err == nil || err.Error() != "src file path is required" { + t.Errorf("empty src: got %v", err) + } + if err = StreamBacklogReport(filepath.Join(t.TempDir(), "missing.rdb"), os.Stdout); err == nil { + t.Error("missing file: want an error") + } +} diff --git a/model/stream.go b/model/stream.go index 22bb3d6..e684436 100644 --- a/model/stream.go +++ b/model/stream.go @@ -1,6 +1,9 @@ package model -import "strconv" +import ( + "sort" + "strconv" +) // StreamObject stores a stream object type StreamObject struct { @@ -49,8 +52,7 @@ type StreamId struct { } func (id *StreamId) MarshalText() (text []byte, err error) { - txt := strconv.FormatUint(id.Ms, 10) + "-" + strconv.FormatUint(id.Sequence, 10) - return []byte(txt), nil + return []byte(streamIdKey(id)), nil } // StreamGroup is a consumer group @@ -80,3 +82,195 @@ type StreamConsumer struct { Pending []*StreamId `json:"pending,omitempty"` ActiveTime uint64 `json:"activeTime"` // since rdb11, see https://github.com/redis/redis/pull/11099 } + +func streamIdKey(id *StreamId) string { + return strconv.FormatUint(id.Ms, 10) + "-" + strconv.FormatUint(id.Sequence, 10) +} + +func (id *StreamId) isAfter(other *StreamId) bool { + if other == nil { + return true + } + if id.Ms != other.Ms { + return id.Ms > other.Ms + } + return id.Sequence > other.Sequence +} + +// PendingOwners maps every distinct id in group.Pending, written as -, to the sorted names of +// the consumers whose own pending list holds it. An id that no consumer holds maps to an empty slice. +// Ids in a consumer's list that are not pending entries of the group are ignored. +func (group *StreamGroup) PendingOwners() map[string][]string { + owners := make(map[string][]string, len(group.Pending)) + for _, nack := range group.Pending { + key := streamIdKey(nack.Id) + if _, ok := owners[key]; !ok { + owners[key] = []string{} + } + } + for _, consumer := range group.Consumers { + for _, id := range consumer.Pending { + key := streamIdKey(id) + names, ok := owners[key] + if !ok { + continue + } + held := false + for _, name := range names { + if name == consumer.Name { + held = true + break + } + } + if !held { + owners[key] = append(names, consumer.Name) + } + } + } + for key := range owners { + sort.Strings(owners[key]) + } + return owners +} + +// UndeliveredMessages returns the live messages that group has never been delivered, ordered by ascending id: +// those with an id greater than group.LastId and no pending entry in the group. +// If the group has no LastId, every message that is not deleted and not pending is returned. +func (obj *StreamObject) UndeliveredMessages(group *StreamGroup) []*StreamMessage { + var lastId *StreamId + held := map[string]bool{} + if group != nil { + lastId = group.LastId + for _, nack := range group.Pending { + held[streamIdKey(nack.Id)] = true + } + } + result := make([]*StreamMessage, 0) + for _, entry := range obj.Entries { + for _, msg := range entry.Msgs { + if msg.Deleted || !msg.Id.isAfter(lastId) || held[streamIdKey(msg.Id)] { + continue + } + result = append(result, msg) + } + } + sort.SliceStable(result, func(i, j int) bool { + return result[j].Id.isAfter(result[i].Id) + }) + return result +} + +// ConsumerBacklog is one row of the backlog report of a consumer group. +// Undelivered and Dangling are numbers of the whole group, so every row of a group carries the same pair. +type ConsumerBacklog struct { + Group string `json:"group"` + // Consumer is empty for the row that collects the pending entries no consumer holds + Consumer string `json:"consumer"` + // Pending is the number of pending entries the consumer holds + Pending int `json:"pending"` + // OldestDeliveryTime is the smallest delivery time among the pending entries + OldestDeliveryTime uint64 `json:"oldestDeliveryTime"` + // MaxDeliveryCount is the largest delivery count among the pending entries + MaxDeliveryCount uint64 `json:"maxDeliveryCount"` + // Undelivered is the number of messages the group has not been delivered yet + Undelivered int `json:"undelivered"` + // Dangling is the number of distinct pending entries of the group whose message is deleted or missing + Dangling int `json:"dangling"` +} + +func (obj *StreamObject) danglingPending(group *StreamGroup) int { + live := map[string]bool{} + for _, entry := range obj.Entries { + for _, msg := range entry.Msgs { + if !msg.Deleted { + live[streamIdKey(msg.Id)] = true + } + } + } + dangling := 0 + counted := map[string]bool{} + for _, nack := range group.Pending { + key := streamIdKey(nack.Id) + if counted[key] { + continue + } + counted[key] = true + if !live[key] { + dangling++ + } + } + return dangling +} + +func (group *StreamGroup) backlogRows(undelivered, dangling int) []*ConsumerBacklog { + owners := group.PendingOwners() + rows := make(map[string]*ConsumerBacklog) + ordered := make([]*ConsumerBacklog, 0, len(owners)) + counted := make(map[string]bool, len(owners)) + for _, nack := range group.Pending { + key := streamIdKey(nack.Id) + if counted[key] { + continue + } + counted[key] = true + names := owners[key] + if len(names) == 0 { + names = []string{""} + } + for _, name := range names { + row := rows[name] + if row == nil { + row = &ConsumerBacklog{ + Group: group.Name, + Consumer: name, + OldestDeliveryTime: nack.DeliveryTime, + Undelivered: undelivered, + Dangling: dangling, + } + rows[name] = row + ordered = append(ordered, row) + } + row.Pending++ + if nack.DeliveryTime < row.OldestDeliveryTime { + row.OldestDeliveryTime = nack.DeliveryTime + } + if nack.DeliveryCount > row.MaxDeliveryCount { + row.MaxDeliveryCount = nack.DeliveryCount + } + } + } + if len(ordered) == 0 { + ordered = append(ordered, &ConsumerBacklog{ + Group: group.Name, + Undelivered: undelivered, + Dangling: dangling, + }) + } + sort.SliceStable(ordered, func(i, j int) bool { + a, b := ordered[i], ordered[j] + if a.Pending != b.Pending { + return a.Pending > b.Pending + } + if a.OldestDeliveryTime != b.OldestDeliveryTime { + return a.OldestDeliveryTime < b.OldestDeliveryTime + } + return a.Consumer < b.Consumer + }) + return ordered +} + +// ConsumerBacklog returns one row for every consumer holding pending entries of a group, plus one row with an +// empty Consumer for the pending entries nobody holds. A group without pending entries yields a single row. +// Rows are ordered by group name, then by descending Pending, ascending OldestDeliveryTime and Consumer. +func (obj *StreamObject) ConsumerBacklog() []*ConsumerBacklog { + groups := make([]*StreamGroup, 0, len(obj.Groups)) + groups = append(groups, obj.Groups...) + sort.SliceStable(groups, func(i, j int) bool { + return groups[i].Name < groups[j].Name + }) + result := make([]*ConsumerBacklog, 0) + for _, group := range groups { + result = append(result, group.backlogRows(len(obj.UndeliveredMessages(group)), obj.danglingPending(group))...) + } + return result +} diff --git a/model/stream_test.go b/model/stream_test.go new file mode 100644 index 0000000..3e8a287 --- /dev/null +++ b/model/stream_test.go @@ -0,0 +1,220 @@ +package model + +import ( + "reflect" + "strconv" + "strings" + "testing" +) + +func mkId(ms, seq uint64) *StreamId { + return &StreamId{Ms: ms, Sequence: seq} +} + +func mkMsg(ms, seq uint64, deleted bool) *StreamMessage { + return &StreamMessage{Id: mkId(ms, seq), Fields: map[string]string{"k": "v"}, Deleted: deleted} +} + +func mkEntry(msgs ...*StreamMessage) *StreamEntry { + return &StreamEntry{FirstMsgId: msgs[0].Id, Fields: []string{"k"}, Msgs: msgs} +} + +func mkNack(ms, seq, deliveryTime, deliveryCount uint64) *StreamNAck { + return &StreamNAck{Id: mkId(ms, seq), DeliveryTime: deliveryTime, DeliveryCount: deliveryCount} +} + +func mkConsumer(name string, pending ...*StreamId) *StreamConsumer { + return &StreamConsumer{Name: name, Pending: pending} +} + +func messageIds(msgs []*StreamMessage) []string { + ids := make([]string, 0, len(msgs)) + for _, m := range msgs { + text, _ := m.Id.MarshalText() + ids = append(ids, string(text)) + } + return ids +} + +func backlogRows(rows []*ConsumerBacklog) []string { + texts := make([]string, 0, len(rows)) + for _, row := range rows { + texts = append(texts, strings.Join([]string{ + row.Group, row.Consumer, + strconv.Itoa(row.Pending), + strconv.FormatUint(row.OldestDeliveryTime, 10), + strconv.FormatUint(row.MaxDeliveryCount, 10), + strconv.Itoa(row.Undelivered), + strconv.Itoa(row.Dangling), + }, "/")) + } + return texts +} + +func TestPendingOwners(t *testing.T) { + group := &StreamGroup{ + Name: "g1", + Pending: []*StreamNAck{ + mkNack(1, 0, 100, 1), mkNack(2, 0, 200, 1), mkNack(3, 0, 300, 1), mkNack(7, 2, 100, 1), mkNack(7, 2, 300, 4), + }, + Consumers: []*StreamConsumer{ + mkConsumer("carol", mkId(1, 0), mkId(9, 9)), + mkConsumer("alice", mkId(1, 0), mkId(1, 0), mkId(2, 0)), + }, + } + want := map[string][]string{ + "1-0": {"alice", "carol"}, + "2-0": {"alice"}, + "3-0": {}, + "7-2": {}, + } + if got := group.PendingOwners(); !reflect.DeepEqual(got, want) { + t.Errorf("PendingOwners() = %v, want %v", got, want) + } +} + +func TestUndeliveredMessages(t *testing.T) { + obj := &StreamObject{Entries: []*StreamEntry{ + mkEntry(mkMsg(9, 0, false)), + mkEntry(mkMsg(1, 0, false), mkMsg(4, 0, false), mkMsg(7, 1, false), mkMsg(7, 2, true), mkMsg(8, 5, false), mkMsg(8, 1, false)), + }} + tests := []struct { + name string + group *StreamGroup + want string + }{ + {"compares every message id, orders by id, skips deleted", &StreamGroup{LastId: mkId(4, 0)}, "7-1,8-1,8-5,9-0"}, + {"compares the sequence within a millisecond", &StreamGroup{LastId: mkId(7, 1)}, "8-1,8-5,9-0"}, + {"no last id returns every live message", &StreamGroup{}, "1-0,4-0,7-1,8-1,8-5,9-0"}, + { + "skips messages the group still holds as pending", + &StreamGroup{LastId: mkId(1, 0), Pending: []*StreamNAck{mkNack(4, 0, 1, 1), mkNack(8, 5, 1, 1)}}, + "7-1,8-1,9-0", + }, + { + "a pending entry above the last id is not undelivered", + &StreamGroup{LastId: mkId(1, 0), Pending: []*StreamNAck{mkNack(9, 0, 1, 1)}}, + "4-0,7-1,8-1,8-5", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := strings.Join(messageIds(obj.UndeliveredMessages(tt.group)), ","); got != tt.want { + t.Errorf("UndeliveredMessages() = %s, want %s", got, tt.want) + } + }) + } +} + +func TestConsumerBacklog(t *testing.T) { + entries := []*StreamEntry{ + mkEntry(mkMsg(1, 0, false), mkMsg(2, 0, false), mkMsg(3, 0, true), mkMsg(4, 0, false)), + } + tests := []struct { + name string + entries []*StreamEntry + groups []*StreamGroup + want string + }{ + { + name: "aggregates the entries of each consumer", + entries: entries, + groups: []*StreamGroup{{ + Name: "g1", + LastId: mkId(2, 0), + Pending: []*StreamNAck{mkNack(1, 0, 500, 2), mkNack(2, 0, 300, 7), mkNack(4, 0, 900, 1)}, + Consumers: []*StreamConsumer{ + mkConsumer("alice", mkId(1, 0), mkId(2, 0)), + mkConsumer("bob", mkId(4, 0)), + }, + }}, + want: "g1/alice/2/300/7/0/0,g1/bob/1/900/1/0/0", + }, + { + name: "an entry held by two consumers counts for both", + entries: entries, + groups: []*StreamGroup{{ + Name: "g1", + LastId: mkId(4, 0), + Pending: []*StreamNAck{mkNack(1, 0, 400, 5)}, + Consumers: []*StreamConsumer{ + mkConsumer("bob", mkId(1, 0)), + mkConsumer("alice", mkId(1, 0), mkId(1, 0)), + }, + }}, + want: "g1/alice/1/400/5/0/0,g1/bob/1/400/5/0/0", + }, + { + name: "a repeated pending id counts once with its first record", + entries: entries, + groups: []*StreamGroup{{ + Name: "g1", + LastId: mkId(4, 0), + Pending: []*StreamNAck{mkNack(1, 0, 400, 2), mkNack(1, 0, 100, 9)}, + Consumers: []*StreamConsumer{mkConsumer("alice", mkId(1, 0))}, + }}, + want: "g1/alice/1/400/2/0/0", + }, + { + name: "entries nobody holds share the row of an empty consumer name", + entries: entries, + groups: []*StreamGroup{{ + Name: "g1", + LastId: mkId(4, 0), + Pending: []*StreamNAck{mkNack(1, 0, 500, 2), mkNack(2, 0, 0, 4), mkNack(4, 0, 800, 9), mkNack(5, 0, 50, 1)}, + Consumers: []*StreamConsumer{mkConsumer("alice", mkId(1, 0), mkId(7, 7)), mkConsumer("", mkId(5, 0))}, + }}, + want: "g1//3/0/9/0/1,g1/alice/1/500/2/0/1", + }, + { + name: "a group without pending entries still has a row", + entries: entries, + groups: []*StreamGroup{ + {Name: "zulu", LastId: mkId(4, 0)}, + {Name: "alpha", LastId: mkId(1, 0), Consumers: []*StreamConsumer{mkConsumer("alice")}}, + }, + want: "alpha//0/0/0/2/0,zulu//0/0/0/0/0", + }, + { + name: "rows are ordered by pending, oldest delivery time, then name", + entries: []*StreamEntry{mkEntry(mkMsg(1, 0, false), mkMsg(2, 0, false), mkMsg(3, 0, false), mkMsg(4, 0, false), mkMsg(5, 0, false), mkMsg(6, 0, false))}, + groups: []*StreamGroup{{ + Name: "g1", + LastId: mkId(6, 0), + Pending: []*StreamNAck{ + mkNack(1, 0, 700, 1), mkNack(2, 0, 200, 1), mkNack(3, 0, 200, 1), mkNack(4, 0, 900, 1), mkNack(5, 0, 100, 1), mkNack(6, 0, 100, 1), + }, + Consumers: []*StreamConsumer{ + mkConsumer("dave", mkId(4, 0)), + mkConsumer("carol", mkId(3, 0)), + mkConsumer("bob", mkId(2, 0)), + mkConsumer("alice", mkId(1, 0)), + mkConsumer("erin", mkId(5, 0), mkId(6, 0)), + }, + }}, + want: "g1/erin/2/100/1/0/0,g1/bob/1/200/1/0/0,g1/carol/1/200/1/0/0,g1/alice/1/700/1/0/0,g1/dave/1/900/1/0/0", + }, + { + name: "undelivered and dangling are per group", + entries: entries, + groups: []*StreamGroup{ + { + Name: "behind", + LastId: mkId(1, 0), + Pending: []*StreamNAck{mkNack(2, 0, 10, 1), mkNack(3, 0, 20, 1), mkNack(9, 0, 30, 1)}, + Consumers: []*StreamConsumer{mkConsumer("alice", mkId(2, 0), mkId(3, 0), mkId(9, 0))}, + }, + {Name: "current", LastId: mkId(4, 0)}, + }, + want: "behind/alice/3/10/1/1/2,current//0/0/0/0/0", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + obj := &StreamObject{Entries: tt.entries, Groups: tt.groups} + if got := strings.Join(backlogRows(obj.ConsumerBacklog()), ","); got != tt.want { + t.Errorf("ConsumerBacklog() = %s, want %s", got, tt.want) + } + }) + } +}