add an extra lock for insert and delete on the cache database

ctod/core/factory/factory_cache.py

@@ -48,6 +48,7 @@ def __init__(self, cache_path: str, db_name: str = "factory_cache.db", in_memory
         self.batch_processing = False
         self.batch_rerun = False
         self.lock = asyncio.Lock()
+        self.db_lock = asyncio.Lock()
 
     async def initialize(self):
         await self._create_table()
@@ -108,16 +109,17 @@ async def _add_batched(self) -> None:
 
         if batch_copy:
             try:
-                async with aiosqlite.connect(self.db_name) as db:
-                    for key, value in batch_copy.items():
-                        await db.execute(
-                            """
-                            INSERT OR REPLACE INTO cache (key, value, timestamp) 
-                            VALUES (?, ?, ?)
-                            """,
-                            (key, pickle.dumps(value), time.time()),
-                        )
-                    await db.commit()
+                async with self.db_lock:
+                    async with aiosqlite.connect(self.db_name) as db:
+                        for key, value in batch_copy.items():
+                            await db.execute(
+                                """
+                                INSERT OR REPLACE INTO cache (key, value, timestamp) 
+                                VALUES (?, ?, ?)
+                                """,
+                                (key, pickle.dumps(value), time.time()),
+                            )
+                        await db.commit()
             except aiosqlite.Error as e:
                 logging.error(f"Error adding to factory cache: {e}")
 
@@ -163,11 +165,12 @@ async def get(self, keys: List[str]) -> Dict[str, Any]:
 
     async def clear_expired(self, keys_to_keep: List[str]) -> None:
         try:
-            async with aiosqlite.connect(self.db_name) as db:
-                placeholders = ', '.join('?' for _ in keys_to_keep)
-                query = f"DELETE FROM cache WHERE (strftime('%s', 'now') - timestamp) > ? AND key NOT IN ({placeholders})"
-                await db.execute(query, (self.ttl, *keys_to_keep))
-                await db.commit()
+            async with self.db_lock:
+                async with aiosqlite.connect(self.db_name) as db:
+                    placeholders = ', '.join('?' for _ in keys_to_keep)
+                    query = f"DELETE FROM cache WHERE (strftime('%s', 'now') - timestamp) > ? AND key NOT IN ({placeholders})"
+                    await db.execute(query, (self.ttl, *keys_to_keep))
+                    await db.commit()
 
         except aiosqlite.Error as e:
             logging.error(f"Error clearing expired items from factory cache: {e}")