Updater: Clearing old cache items.

Author: xsedla1o

docs/configuration/updater.md

@@ -26,4 +26,16 @@ update_batch_period: "5m"
 ```
 
 Try to find a balance between having batches run too often (some batches may execute empty,
-leading to unnecessary overhead) and too rarely.
+leading to unnecessary overhead) and too rarely.
+
+Additional options are available for the updater module, but they are not required.
+They are used to configure when cache management runs and the maximum number of entries in the cache.
+These are the options with their default values:
+
+```yaml
+cache_management_cron: { hour: "2", minute: "0", second: "0" }  # (1)!
+cache_max_entries: 32  # (2)!
+```
+
+1. `cache_management_cron` - a cron schedule for the cache management. See [CronExpression docs][dp3.common.config.CronExpression] for details.
+2. `cache_max_entries` - the maximum number of entries in the cache. If the number of cache items (counted for each individual period) exceeds this number, the oldest entries are removed.

dp3/core/updater.py

@@ -30,11 +30,16 @@ class UpdaterConfig(BaseModel, extra="forbid"):
         update_batch_cron: A CRON expression for the periodic update.
         update_batch_period: The period of the periodic update.
             Should equal to the period of `update_batch_cron`.
+        cache_management_cron: A CRON expression for the cache management.
+        cache_max_entries: The maximum number of finished cache entries per thread_id.
     """
 
     update_batch_cron: CronExpression
     update_batch_period: ParsedTimedelta
 
+    cache_management_cron: CronExpression = CronExpression(hour=2, minute=0, second=0)
+    cache_max_entries: int = 32
+
 
 class UpdateThreadState(BaseModel, validate_assignment=True):
     """
@@ -131,6 +136,27 @@ def upsert(self, state: UpdateThreadState) -> UpdateResult:
         }
         return self._cache.update_one(filter_dict, update=update_dict, upsert=True)
 
+    def register_management(
+        self, scheduler: Scheduler, trigger: CronExpression, max_entries: int
+    ) -> int:
+        """Registers the cache management task with the scheduler."""
+        return scheduler.register(
+            self._manage_cache, func_args=[max_entries], **trigger.model_dump()
+        )
+
+    def _manage_cache(self, max_entries: int):
+        """Removes the oldest finished cache entries if their count exceeds the limit."""
+        counts_per_thread_id = defaultdict(int)
+        to_delete = []
+
+        for state in self._cache.find({"type": "state", "finished": True}).sort("t_created", -1):
+            state_obj = UpdateThreadState.model_validate(state)
+            counts_per_thread_id[state_obj.thread_id] += 1
+            if counts_per_thread_id[state_obj.thread_id] > max_entries:
+                to_delete.append(state["_id"])
+
+        self._cache.delete_many({"_id": {"$in": to_delete}})
+
     def _setup_cache_indexes(self):
         """Sets up the indexes of the state cache."""
         self._cache.create_index("t_created", background=True)
@@ -163,6 +189,9 @@ def __init__(
 
         # Get state cache
         self.cache = UpdaterCache(self.db.get_module_cache("Updater"))
+        self.cache.register_management(
+            scheduler, self.config.cache_management_cron, self.config.cache_max_entries
+        )
 
         self.update_thread_hooks = defaultdict(dict)