Add cache locks

src/app/services/cache_updater.py

@@ -6,6 +6,8 @@ from typing import Optional, cast
 
 from fastapi import UploadFile
 
+import aioredis
+from aioredis.exceptions import LockError
 from arq.connections import ArqRedis
 from arq.worker import Retry
 import httpx
@@ -105,6 +107,8 @@ async def cache_file_by_book_id(
     file_type: str,
     by_request: bool = True,
 ) -> Optional[CachedFile]:
+    r_client: aioredis.Redis = ctx["redis"]
+
     try:
         book = await get_book(book_id)
     except httpx.ConnectError:
@@ -115,8 +119,14 @@ async def cache_file_by_book_id(
     if file_type not in book.available_types:
         raise FileTypeNotAllowed(f"{file_type} not in {book.available_types}!")
 
+    lock = r_client.lock(f"{book_id}_{file_type}", blocking_timeout=5)
+
     try:
-        return await cache_file(book, file_type)
+        try:
+            async with lock:
+                return await cache_file(book, file_type)
+        except LockError:
+            raise Retry(defer=timedelta(minutes=15))
     except Retry as e:
         if by_request:
             return None

src/core/setup_arq.py

@@ -1,9 +1,12 @@
+import aioredis
+
 from app.services.cache_updater import (
     check_books,
     cache_file_by_book_id,
     check_books_page,
 )
 from core.arq_pool import get_redis_settings, get_arq_pool
+from core.config import env_config
 from core.db import database
 import core.sentry  # noqa: F401
 
@@ -13,6 +16,9 @@ async def startup(ctx):
         await database.connect()
 
     ctx["arc_pool"] = await get_arq_pool()
+    ctx["redis"] = aioredis.Redis(
+        host=env_config.REDIS_HOST, port=env_config.REDIS_PORT, db=env_config.REDIS_DB
+    )
 
 
 async def shutdown(ctx):