feat: queue task to cache GCS results in redis

when we miss the redis cache but find the test rollup results in GCS
we want to queue up a task to cache it in redis so subsequent calls
to the GQL endpoint for the same repo get to hit redis instead of GCS

Author: joseph-sentry

graphql_api/tests/test_test_analytics.py

@@ -151,16 +151,18 @@ def test_get_test_results(
         assert results.equals(test_results_table)
 
     def test_get_test_results_no_storage(self, transactional_db, repository):
-        with pytest.raises(FileNotFoundError):
-            get_results(repository.repoid, repository.branch, 30)
+        assert get_results(repository.repoid, repository.branch, 30) is None
 
     def test_get_test_results_no_redis(
-        self, transactional_db, repository, store_in_storage
+        self, mocker, transactional_db, repository, store_in_storage
     ):
+        m = mocker.patch("services.task.TaskService.cache_test_results_redis")
         results = get_results(repository.repoid, repository.branch, 30)
         assert results is not None
         assert results.equals(test_results_table)
 
+        m.assert_called_once_with(repository.repoid, repository.branch)
+
     def test_test_results(self, transactional_db, repository, store_in_redis):
         test_results = generate_test_results(
             repoid=repository.repoid,

services/task/task.py

@@ -418,3 +418,9 @@ def delete_component_measurements(self, repoid: int, component_id: str) -> None:
                 measurement_id=component_id,
             ),
         ).apply_async()
+
+    def cache_test_results_redis(self, repoid: int, branch: str) -> None:
+        self._create_signature(
+            celery_config.cache_test_rollups_redis_task_name,
+            kwargs=dict(repoid=repoid, branch=branch),
+        ).apply_async()

utils/test_results.py

@@ -3,27 +3,32 @@
 
 from services.redis_configuration import get_redis_connection
 from services.storage import StorageService
+from services.task import TaskService
 
 
-def get_redis_key(
+def redis_key(
     repoid: int,
     branch: str,
     interval_start: int,
     interval_end: int | None = None,
 ) -> str:
-    if interval_end is None:
-        return f"test_results:{repoid}:{branch}:{interval_start}"
-    else:
-        return f"test_results:{repoid}:{branch}:{interval_start}:{interval_end}"
+    key = f"test_results:{repoid}:{branch}:{interval_start}"
 
+    if interval_end is not None:
+        key = f"{key}:{interval_end}"
 
-def get_storage_key(
+    return key
+
+
+def storage_key(
     repoid: int, branch: str, interval_start: int, interval_end: int | None = None
 ) -> str:
-    if interval_end is None:
-        return f"test_results/rollups/{repoid}/{branch}/{interval_start}"
-    else:
-        return f"test_results/rollups/{repoid}/{branch}/{interval_start}_{interval_end}"
+    key = f"test_results/rollups/{repoid}/{branch}/{interval_start}"
+
+    if interval_end is not None:
+        key = f"{key}_{interval_end}"
+
+    return key
 
 
 def get_results(
@@ -32,33 +37,34 @@ def get_results(
     interval_start: int,
     interval_end: int | None = None,
 ) -> pl.DataFrame | None:
-    serialized_table = None
-
+    """
+    try redis
+    if redis is empty
+        try storage
+        if storage is empty
+            return None
+        else
+            cache to redis
+    deserialize
+    """
+    # try redis
     redis_conn = get_redis_connection()
-    redis_key = get_redis_key(repoid, branch, interval_start, interval_end)
+    key = redis_key(repoid, branch, interval_start, interval_end)
+    result: bytes | None = redis_conn.get(key)
 
-    redis_result = redis_conn.get(redis_key)
-
-    if redis_result is not None:
-        serialized_table = redis_result
-
-    if serialized_table is None:
+    if result is None:
+        # try storage
         storage_service = StorageService()
-        storage_key = get_storage_key(repoid, branch, interval_start, interval_end)
-
+        key = storage_key(repoid, branch, interval_start, interval_end)
         try:
-            serialized_table = storage_service.read_file(
-                bucket_name="codecov", path=storage_key
-            )
-        except FileNotInStorageError as e:
-            if interval_end is not None:
-                return None
-            else:
-                raise FileNotFoundError(f"File not found in archive: {e}")
-
-    if serialized_table is None:
-        return None
+            result = storage_service.read_file(bucket_name="codecov", path=key)
+            # cache to redis
+            TaskService().cache_test_results_redis(repoid, branch)
+        except FileNotInStorageError:
+            # give up
+            return None
 
-    table = pl.read_ipc(serialized_table)
+    # deserialize
+    table = pl.read_ipc(result)
 
     return table