ref(rules): Add delete methods to Redis Buffer (#68837)

We need a way to delete things we've already processed from the buffer.
This PR adds `hdel` (hash delete) and `zremrangebyscore` methods to the
Redis buffer that will later be used in `apply_delayed`, as well as
replaces `sadd` with `zadd`, replaces `smembers` with `zrange` so we can
remove project ids by the time they were added.

Author: ceorourke

src/sentry/buffer/redis.py

@@ -84,10 +84,12 @@ def callback(self, buffer_hook_event: BufferHookEvent, data: RedisBuffer) -> boo
 
 
 class RedisOperation(Enum):
-    SET_ADD = "sadd"
-    SET_GET = "smembers"
+    SORTED_SET_ADD = "zadd"
+    SORTED_SET_GET_RANGE = "zrange"
+    SORTED_SET_DELETE_RANGE = "zremrangebyscore"
     HASH_ADD = "hset"
     HASH_GET_ALL = "hgetall"
+    HASH_DELETE = "hdel"
 
 
 class PendingBuffer:
@@ -263,19 +265,35 @@ def get_redis_connection(self, key: str, transaction: bool = True) -> Pipeline:
         pipe = conn.pipeline(transaction=transaction)
         return pipe
 
-    def _execute_redis_operation(self, key: str, operation: RedisOperation, *args: Any) -> Any:
+    def _execute_redis_operation(
+        self, key: str, operation: RedisOperation, *args: Any, **kwargs: Any
+    ) -> Any:
         pending_key = self._make_pending_key_from_key(key)
         pipe = self.get_redis_connection(pending_key)
-        getattr(pipe, operation.value)(key, *args)
+        getattr(pipe, operation.value)(key, *args, **kwargs)
         if args:
             pipe.expire(key, self.key_expire)
         return pipe.execute()
 
-    def push_to_set(self, key: str, value: list[int] | int) -> None:
-        self._execute_redis_operation(key, RedisOperation.SET_ADD, value)
+    def push_to_sorted_set(self, key: str, value: list[int] | int) -> None:
+        self._execute_redis_operation(key, RedisOperation.SORTED_SET_ADD, {value: time()})
 
     def get_set(self, key: str) -> list[set[int]]:
-        return self._execute_redis_operation(key, RedisOperation.SET_GET)
+        return self._execute_redis_operation(
+            key, RedisOperation.SORTED_SET_GET_RANGE, start=0, end=-1, withscores=True
+        )
+
+    def delete_key(self, key: str, min: int, max: int) -> None:
+        self._execute_redis_operation(key, RedisOperation.SORTED_SET_DELETE_RANGE, min=min, max=max)
+
+    def delete_hash(
+        self,
+        model: type[models.Model],
+        filters: dict[str, models.Model | str | int],
+        field: str,
+    ) -> None:
+        key = self._make_key(model, filters)
+        self._execute_redis_operation(key, RedisOperation.HASH_DELETE, field)
 
     def push_to_hash(
         self,

tests/sentry/buffer/test_redis.py

@@ -258,14 +258,14 @@ def test_process_pending_partitions_none(self, process_pending, process_incr):
     def group_rule_data_by_project_id(self, buffer, project_ids):
         project_ids_to_rule_data = defaultdict(list)
         for proj_id in project_ids[0]:
-            rule_group_pairs = buffer.get_hash(Project, {"project_id": proj_id})
+            rule_group_pairs = buffer.get_hash(Project, {"project_id": proj_id[0]})
             for pair in rule_group_pairs:
                 for k, v in pair.items():
                     if isinstance(k, bytes):
                         k = k.decode("utf-8")
                     if isinstance(v, bytes):
                         v = v.decode("utf-8")
-                    project_ids_to_rule_data[int(proj_id)].append({k: v})
+                    project_ids_to_rule_data[int(proj_id[0])].append({k: v})
         return project_ids_to_rule_data
 
     def test_enqueue(self):
@@ -283,8 +283,8 @@ def test_enqueue(self):
         event4_id = 11
 
         # store the project ids
-        self.buf.push_to_set(key=PROJECT_ID_BUFFER_LIST_KEY, value=project_id)
-        self.buf.push_to_set(key=PROJECT_ID_BUFFER_LIST_KEY, value=project_id2)
+        self.buf.push_to_sorted_set(key=PROJECT_ID_BUFFER_LIST_KEY, value=project_id)
+        self.buf.push_to_sorted_set(key=PROJECT_ID_BUFFER_LIST_KEY, value=project_id2)
 
         # store the rules and group per project
         self.buf.push_to_hash(
@@ -350,6 +350,67 @@ def test_process_batch(self):
         assert mock.call_count == 1
         assert mock.call_args[0][0] == self.buf
 
+    def test_delete_batch(self):
+        """Test that after we add things to redis we can clean it up"""
+        project_id = 1
+        rule_id = 2
+        group_id = 3
+        event_id = 4
+
+        project2_id = 5
+        rule2_id = 6
+        group2_id = 7
+        event2_id = 8
+
+        now = datetime.datetime(2024, 4, 15, 3, 30, 00, tzinfo=datetime.UTC)
+        one_minute_from_now = (now).replace(minute=31)
+
+        # add a set and a hash to the buffer
+        with freeze_time(now):
+            self.buf.push_to_sorted_set(key=PROJECT_ID_BUFFER_LIST_KEY, value=project_id)
+            self.buf.push_to_hash(
+                model=Project,
+                filters={"project_id": project_id},
+                field=f"{rule_id}:{group_id}",
+                value=event_id,
+            )
+        with freeze_time(one_minute_from_now):
+            self.buf.push_to_sorted_set(key=PROJECT_ID_BUFFER_LIST_KEY, value=project2_id)
+            self.buf.push_to_hash(
+                model=Project,
+                filters={"project_id": project2_id},
+                field=f"{rule2_id}:{group2_id}",
+                value=event2_id,
+            )
+
+        # retrieve them
+        project_ids = self.buf.get_set(PROJECT_ID_BUFFER_LIST_KEY)
+        assert len(project_ids[0]) == 2
+        rule_group_pairs = self.buf.get_hash(Project, {"project_id": project_id})
+        assert len(rule_group_pairs)
+
+        # delete only the first project ID by time
+        self.buf.delete_key(PROJECT_ID_BUFFER_LIST_KEY, min=0, max=now.timestamp())
+
+        # retrieve again to make sure only project_id was removed
+        project_ids = self.buf.get_set(PROJECT_ID_BUFFER_LIST_KEY)
+        if isinstance(project_ids[0][0][0], bytes):
+            assert project_ids == [
+                [(bytes(str(project2_id), "utf-8"), one_minute_from_now.timestamp())]
+            ]
+        else:
+            assert project_ids == [[(str(project2_id), one_minute_from_now.timestamp())]]
+
+        # delete the project_id hash
+        self.buf.delete_hash(
+            model=Project,
+            filters={"project_id": project_id},
+            field=f"{rule_id}:{group_id}",
+        )
+
+        rule_group_pairs = self.buf.get_hash(Project, {"project_id": project_id})
+        assert rule_group_pairs == [{}]
+
     @mock.patch("sentry.buffer.redis.RedisBuffer._make_key", mock.Mock(return_value="foo"))
     @mock.patch("sentry.buffer.base.Buffer.process")
     def test_process_uses_signal_only(self, process):