fix(crons): Fix bug where processing error data growth is not capped. (#72867)

This fixes a bug where we don't delete the body of a processing error
after evicting it from the set of data for that monitor. This occurred
when we started storing the bodies of the errors in a separate key and
we forgot to delete it.

Author: wedamija

src/sentry/monitors/processing_errors/__init__.py

@@ -131,10 +131,17 @@ def store_error(error: CheckinProcessingError, monitor: Monitor | None):
     pipeline = redis_client.pipeline(transaction=False)
     pipeline.zadd(error_set_key, {error.id.hex: error.checkin.ts.timestamp()})
     pipeline.set(error_key, serialized_error, ex=MONITOR_ERRORS_LIFETIME)
-    # Cap the error list to the `MAX_ERRORS_PER_SET` most recent errors
-    pipeline.zremrangebyrank(error_set_key, 0, -(MAX_ERRORS_PER_SET + 1))
     pipeline.expire(error_set_key, MONITOR_ERRORS_LIFETIME)
-    pipeline.execute()
+    pipeline.zrange(error_set_key, 0, -(MAX_ERRORS_PER_SET + 1))
+    processing_errors_to_remove = pipeline.execute()[-1]
+    # Cap the error list to the `MAX_ERRORS_PER_SET` most recent errors
+    if processing_errors_to_remove:
+        pipeline = redis_client.pipeline(transaction=False)
+        pipeline.delete(
+            *[build_error_identifier(uuid.UUID(result)) for result in processing_errors_to_remove]
+        )
+        pipeline.zrem(error_set_key, *processing_errors_to_remove)
+        pipeline.execute()
 
 
 def delete_error(project: Project, uuid: uuid.UUID):

src/sentry/monitors/processing_errors/manager.py

@@ -86,6 +86,10 @@ def test_store_max(self):
         assert len(retrieved_errors) == 2
         assert_processing_errors_equal(processing_errors[2], retrieved_errors[0])
         assert_processing_errors_equal(processing_errors[1], retrieved_errors[1])
+        redis_client = _get_cluster()
+        assert not redis_client.exists(build_error_identifier(processing_errors[0].id))
+        assert redis_client.exists(build_error_identifier(processing_errors[1].id))
+        assert redis_client.exists(build_error_identifier(processing_errors[2].id))
 
     def test_get_for_monitor_empty(self):
         monitor = self.create_monitor()