Accept number of digits of server time to include in GC match (#223)

as a way of limiting the keyspace scanned.

Part of RUNT-381

Author: meatballhat

queue/client.go

@@ -21,6 +21,7 @@ var (
 	ErrInvalidWriteArgs          = errors.New("queue: invalid write arguments")
 	ErrNoMatchingMessageInStream = errors.New("queue: no matching message in stream")
 	ErrInvalidMetaCancelation    = errors.New("queue: invalid meta cancelation")
+	ErrInvalidNTimeDigits        = errors.New("queue: invalid number of timestamp digits")
 	ErrStopGC                    = errors.New("queue: stop garbage collection")
 
 	streamSuffixPattern = regexp.MustCompile(`\A:s(\d+)\z`)
@@ -62,10 +63,12 @@ func (c *Client) Prepare(ctx context.Context) error {
 // argument given is the "track values" as extracted from the meta cancelation key.
 type OnGCFunc func(ctx context.Context, trackValues []string) error
 
-// GC performs all garbage collection operations that cannot be automatically
-// performed via key expiry, which is the "meta:cancelation" hash at the time of this
-// writing.
-func (c *Client) GC(ctx context.Context, f OnGCFunc) (uint64, uint64, error) {
+// GC performs all garbage collection operations that cannot be automatically performed
+// via key expiry, which is the "meta:cancelation" hash at the time of this writing. The
+// nTimeDigits argument is used to construct the key match to include that many digits of
+// of the current server time as a way of limiting the keyspace scanned. As a special
+// case, any value <= -1 will result in all keys being scanned.
+func (c *Client) GC(ctx context.Context, nTimeDigits int, f OnGCFunc) (uint64, uint64, error) {
 	now, err := c.rdb.Time(ctx).Result()
 	if err != nil {
 		return 0, 0, err
@@ -78,7 +81,18 @@ func (c *Client) GC(ctx context.Context, f OnGCFunc) (uint64, uint64, error) {
 	idsToDelete := []string{}
 	keysToDelete := []string{}
 
-	iter := c.rdb.HScanNoValues(ctx, MetaCancelationHash, 0, "*:expiry:*", 0).Iterator()
+	match := "*:expiry:*"
+	if nTimeDigits > -1 {
+		nowUnixString := strconv.Itoa(int(nowUnix))
+
+		if nTimeDigits > len(nowUnixString) {
+			return 0, 0, ErrInvalidNTimeDigits
+		}
+
+		match = fmt.Sprintf("*:expiry:%s*", nowUnixString[:nTimeDigits])
+	}
+
+	iter := c.rdb.HScanNoValues(ctx, MetaCancelationHash, 0, match, 0).Iterator()
 	total := uint64(0)
 	twiceDeleted := uint64(0)
 

queue/client_test.go

@@ -458,8 +458,6 @@ func TestClientGCIntegration(t *testing.T) {
 	client := queue.NewTrackingClient(rdb, ttl, "tracketytrack")
 	require.NoError(t, client.Prepare(ctx))
 
-	runClientWriteIntegrationTest(ctx, t, rdb, client, true)
-
 	gcTrackedFields := []string{}
 
 	onGCFunc := func(_ context.Context, trackedFields []string) error {
@@ -468,12 +466,39 @@ func TestClientGCIntegration(t *testing.T) {
 		return nil
 	}
 
-	total, nDeleted, err := client.GC(ctx, onGCFunc)
-	require.NoError(t, err)
-	require.Equal(t, uint64(15), total)
-	require.Equal(t, uint64(10), nDeleted)
+	t.Run("full scan", func(t *testing.T) {
+		gcTrackedFields = []string{}
+
+		require.NoError(t, rdb.FlushAll(t.Context()).Err())
+
+		runClientWriteIntegrationTest(ctx, t, rdb, client, true)
+
+		total, nDeleted, err := client.GC(ctx, -1, onGCFunc)
+		require.NoError(t, err)
+		require.Equal(t, uint64(15), total)
+		require.Equal(t, uint64(10), nDeleted)
+
+		require.Len(t, gcTrackedFields, 10)
+	})
+
+	t.Run("scoped scan", func(t *testing.T) {
+		require.NoError(t, rdb.FlushAll(t.Context()).Err())
 
-	require.Len(t, gcTrackedFields, 10)
+		runClientWriteIntegrationTest(ctx, t, rdb, client, true)
+
+		total, _, err := client.GC(ctx, 6, onGCFunc)
+		require.NoError(t, err)
+		require.Equal(t, uint64(10), total)
+	})
+
+	t.Run("invalid nTimeDigits", func(t *testing.T) {
+		require.NoError(t, rdb.FlushAll(t.Context()).Err())
+
+		runClientWriteIntegrationTest(ctx, t, rdb, client, true)
+
+		_, _, err := client.GC(ctx, 11, onGCFunc)
+		require.Error(t, err)
+	})
 }
 
 // TestPickupLatencyIntegration runs a test with a mostly-empty queue -- by