PLAT-604 [queue] create Stats (to replace PendingCount)

queue/client.go

@@ -24,6 +24,13 @@ type Client struct {
 	ttl time.Duration // ttl for all keys in queue
 }
 
+type Stats struct {
+	// Len is the aggregate length of the queue, as reported by XLEN
+	Len int64
+	// PendingCount is the aggregate count of pending entries, as reported by XPENDING
+	PendingCount int64
+}
+
 func NewClient(rdb redis.Cmdable, ttl time.Duration) *Client {
 	return &Client{
 		rdb: rdb,
@@ -43,11 +50,13 @@ func (c *Client) Len(ctx context.Context, name string) (int64, error) {
 	return lenScript.RunRO(ctx, c.rdb, []string{name}).Int64()
 }
 
-// PendingCount counts the aggregate pending entries (that is, number of
-// messages that have been read but not acknowledged) of the consumer group for
-// the given queue, as reported by XPENDING.
-func (c *Client) PendingCount(ctx context.Context, queue string, group string) (int64, error) {
-	return pendingCountScript.RunRO(ctx, c.rdb, []string{queue}, group).Int64()
+// Stats calculates aggregate statistics about the queue and consumer group.
+func (c *Client) Stats(ctx context.Context, queue string, group string) (Stats, error) {
+	out, err := statsScript.RunRO(ctx, c.rdb, []string{queue}, group).Int64Slice()
+	if err != nil {
+		return Stats{}, err
+	}
+	return Stats{Len: out[0], PendingCount: out[1]}, nil
 }
 
 // Read a single message from the queue. If the Block field of args is

queue/client_test.go

@@ -78,9 +78,10 @@ func TestClientIntegration(t *testing.T) {
 		assert.Contains(t, msg.Values, "id")
 		ids[msg.Values["id"].(string)] = struct{}{}
 
-		pendingCount, err := client.PendingCount(ctx, "test", "mygroup")
+		stats, err := client.Stats(ctx, "test", "mygroup")
 		require.NoError(t, err)
-		assert.EqualValues(t, i+1, pendingCount)
+		assert.EqualValues(t, i+1, stats.PendingCount)
+		assert.EqualValues(t, 15, stats.Len)
 	}
 
 	// We should have read all the messages we enqueued

queue/queue.go

@@ -48,6 +48,10 @@ var (
 	pendingCountCmd    string
 	pendingCountScript = redis.NewScript(pendingCountCmd)
 
+	//go:embed stats.lua
+	statsCmd    string
+	statsScript = redis.NewScript(statsCmd)
+
 	//go:embed read.lua
 	readCmd    string
 	readScript = redis.NewScript(readCmd)
@@ -64,6 +68,9 @@ func prepare(ctx context.Context, rdb redis.Cmdable) error {
 	if err := pendingCountScript.Load(ctx, rdb).Err(); err != nil {
 		return err
 	}
+	if err := statsScript.Load(ctx, rdb).Err(); err != nil {
+		return err
+	}
 	if err := readScript.Load(ctx, rdb).Err(); err != nil {
 		return err
 	}

queue/stats.lua

@@ -0,0 +1,36 @@
+-- stats commands take the form
+--
+--   EVALSHA sha 1 key group
+--
+-- Note: strictly, it is illegal for a script to manipulate keys that are not
+-- explicitly passed to EVAL{,SHA}, but in practice this is fine as long as all
+-- keys are on the same server (e.g. in cluster scenarios). In our case a single
+-- queue, which may be composed of multiple streams and metadata keys, is always
+-- on the same server.
+
+local base = KEYS[1]
+local group = ARGV[1]
+
+local key_meta = base .. ':meta'
+
+local streams = tonumber(redis.call('HGET', key_meta, 'streams') or 1)
+local len = 0
+local pending_count = 0
+
+for idx = 0, streams-1 do
+  local stream = base .. ':s' .. idx
+
+  len = len + redis.call('XLEN', stream)
+  local info = redis.pcall('XPENDING', stream, group)
+  if info['err'] then
+    if string.match(info['err'], '^NOGROUP ') then
+      -- if either the stream or group don't exist, there are zero pending entries
+    else
+      return redis.error_reply(info['err']..' accessing '..stream)
+    end
+  else
+    pending_count = pending_count + info[1]
+  end
+end
+
+return {len, pending_count}