Track all meta cancelation records in one hash (#220)

so as not to dramatically increase the number of keys used.

Part of RUNT-369 which broadly addresses cancelation of work that is
considered "in the queue".

Author: meatballhat

queue/client.go

@@ -19,6 +19,7 @@ var (
 	ErrInvalidReadArgs           = errors.New("queue: invalid read arguments")
 	ErrInvalidWriteArgs          = errors.New("queue: invalid write arguments")
 	ErrNoMatchingMessageInStream = errors.New("queue: no matching message in stream")
+	ErrInvalidMetaCancelation    = errors.New("queue: invalid meta cancelation")
 
 	streamSuffixPattern = regexp.MustCompile(`\A:s(\d+)\z`)
 )
@@ -51,6 +52,16 @@ func (c *Client) Prepare(ctx context.Context) error {
 	return prepare(ctx, c.rdb)
 }
 
+// GC performs all garbage collection operations that cannot be automatically
+// performed via key expiry.
+func (c *Client) GC(ctx context.Context) error {
+	if _, err := gcMetaCancelation(ctx, c.rdb); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // Len calculates the aggregate length (XLEN) of the queue. It adds up the
 // lengths of all the streams in the queue.
 func (c *Client) Len(ctx context.Context, name string) (int64, error) {
@@ -286,9 +297,9 @@ type metaCancelation struct {
 // Del supports removal of a message when the given `fieldValue` matches a "meta
 // cancelation" key as written when using a client with tracking support.
 func (c *Client) Del(ctx context.Context, fieldValue string) error {
-	metaCancelationKey := fmt.Sprintf("_meta:cancelation:%x", sha1.Sum([]byte(fieldValue)))
+	metaCancelationKey := fmt.Sprintf("%x", sha1.Sum([]byte(fieldValue)))
 
-	msgBytes, err := c.rdb.Get(ctx, metaCancelationKey).Bytes()
+	msgBytes, err := c.rdb.HGet(ctx, MetaCancelationHash, metaCancelationKey).Bytes()
 	if err != nil {
 		return err
 	}
@@ -298,6 +309,14 @@ func (c *Client) Del(ctx context.Context, fieldValue string) error {
 		return err
 	}
 
+	if msg.StreamID == "" {
+		return fmt.Errorf("empty stream_id: %w", ErrInvalidMetaCancelation)
+	}
+
+	if msg.MsgID == "" {
+		return fmt.Errorf("empty msg_id: %w", ErrInvalidMetaCancelation)
+	}
+
 	n, err := c.rdb.XDel(ctx, msg.StreamID, msg.MsgID).Result()
 	if err != nil {
 		return err

queue/client_test.go

@@ -439,19 +439,32 @@ func TestClientDelIntegration(t *testing.T) {
 	require.Error(t, client.Del(ctx, trackIDs[0]+"oops"))
 	require.Error(t, client.Del(ctx, "bogustown"))
 
-	metaCancelationKey := "_meta:cancelation:" + fmt.Sprintf("%x", sha1.Sum([]byte(trackIDs[1])))
+	metaCancelationKey := fmt.Sprintf("%x", sha1.Sum([]byte(trackIDs[1])))
 
-	metaCancel, err := rdb.Get(ctx, metaCancelationKey).Result()
+	metaCancel, err := rdb.HGet(ctx, queue.MetaCancelationHash, metaCancelationKey).Result()
 	require.NoError(t, err)
 
-	rdb.SetEx(ctx, metaCancelationKey, "{{[,"+metaCancel, 5*time.Second)
+	rdb.HSet(ctx, queue.MetaCancelationHash, metaCancelationKey, "{{[,bogus"+metaCancel)
 
 	require.Error(t, client.Del(ctx, trackIDs[1]))
 
 	require.NoError(t, client.Del(ctx, trackIDs[2]))
 	require.ErrorIs(t, client.Del(ctx, trackIDs[2]), queue.ErrNoMatchingMessageInStream)
 }
 
+func TestClientGCIntegration(t *testing.T) {
+	ctx := test.Context(t)
+	rdb := test.Redis(ctx, t)
+
+	ttl := 24 * time.Hour
+	client := queue.NewTrackingClient(rdb, ttl, "tracketytrack")
+	require.NoError(t, client.Prepare(ctx))
+
+	runClientWriteIntegrationTest(ctx, t, rdb, client, true)
+
+	require.NoError(t, client.GC(ctx))
+}
+
 // TestPickupLatencyIntegration runs a test with a mostly-empty queue -- by
 // running artificially slow producers and full-speed consumers -- to ensure
 // that the blocking read operation has low latency.

queue/queue.go

@@ -35,6 +35,9 @@ package queue
 import (
 	"context"
 	_ "embed" // to provide go:embed support
+	"strconv"
+	"strings"
+	"time"
 
 	"github.com/redis/go-redis/v9"
 )
@@ -62,7 +65,19 @@ var (
 
 	//go:embed writetracking.lua
 	writeTrackingCmd    string
-	writeTrackingScript = redis.NewScript(writeTrackingCmd)
+	writeTrackingScript = redis.NewScript(
+		strings.ReplaceAll(
+			writeTrackingCmd,
+			"__META_CANCELATION_HASH__",
+			MetaCancelationHash,
+		),
+	)
+)
+
+const (
+	MetaCancelationHash = "meta:cancelation"
+
+	metaCancelationGCBatchSize = 100
 )
 
 func prepare(ctx context.Context, rdb redis.Cmdable) error {
@@ -86,3 +101,48 @@ func prepare(ctx context.Context, rdb redis.Cmdable) error {
 	}
 	return nil
 }
+
+func gcMetaCancelation(ctx context.Context, rdb redis.Cmdable) (int, error) {
+	now := time.Now().UTC().Unix()
+	keysToDelete := []string{}
+	iter := rdb.HScan(ctx, MetaCancelationHash, 0, "*:expiry:*", 0).Iterator()
+
+	for iter.Next(ctx) {
+		key := iter.Val()
+
+		keyParts := strings.Split(key, ":")
+		if len(keyParts) != 3 {
+			continue
+		}
+
+		keyTime, err := strconv.ParseInt(keyParts[2], 0, 64)
+		if err != nil {
+			continue
+		}
+
+		if keyTime > now {
+			keysToDelete = append(keysToDelete, key, keyParts[0])
+		}
+	}
+
+	if err := iter.Err(); err != nil {
+		return 0, err
+	}
+
+	for i := 0; i < len(keysToDelete); i += metaCancelationGCBatchSize {
+		sliceEnd := i + metaCancelationGCBatchSize
+		if sliceEnd > len(keysToDelete) {
+			sliceEnd = len(keysToDelete)
+		}
+
+		if err := rdb.HDel(
+			ctx,
+			MetaCancelationHash,
+			keysToDelete[i:sliceEnd]...,
+		).Err(); err != nil {
+			return 0, err
+		}
+	}
+
+	return len(keysToDelete), nil
+}

queue/writetracking.lua

@@ -116,15 +116,22 @@ local id = redis.call('XADD', key_stream, '*', unpack(fields))
 redis.call('XADD', key_notifications, 'MAXLEN', '1', '*', 's', selected_sid)
 
 if track_value ~= '' then
+  local cancelation_key = redis.sha1hex(track_value)
+  local server_time = redis.call('TIME')
+  local expiry_unixtime = tonumber(server_time[1]) + 90000 -- 25 hours
+  local cancelation_expiry_key = cancelation_key .. ':expiry:' .. tostring(expiry_unixtime)
+
   redis.call(
-    'SETEX',
-    '_meta:cancelation:' .. redis.sha1hex(track_value),
-    90000, -- 25 hours
+    'HSET',
+    '__META_CANCELATION_HASH__',
+    cancelation_key,
     cjson.encode({
       ['stream_id'] = key_stream,
       ['track_value'] = track_value,
       ['msg_id'] = id,
-    })
+    }),
+    cancelation_expiry_key,
+    '1'
   )
 end