✨(backend) use batches in indexing task

Reduce the number of Find API calls by grouping all the latest changes
for indexation : send all the documents updated or deleted since the
triggering of the task.

Signed-off-by: Fabre Florian <ffabre@hybird.org>

Author: joehybird

src/backend/core/models.py

@@ -904,7 +904,8 @@ def soft_delete(self):
 
         # Mark all descendants as soft deleted
         self.get_descendants().filter(ancestors_deleted_at__isnull=True).update(
-            ancestors_deleted_at=self.ancestors_deleted_at
+            ancestors_deleted_at=self.ancestors_deleted_at,
+            updated_at=self.updated_at,
         )
 
     @transaction.atomic

src/backend/core/services/search_indexers.py

@@ -130,16 +130,17 @@ def __init__(self, batch_size=None):
                 "SEARCH_INDEXER_QUERY_URL must be set in Django settings."
             )
 
-    def index(self):
+    def index(self, queryset=None):
         """
         Fetch documents in batches, serialize them, and push to the search backend.
         """
         last_id = 0
         count = 0
+        queryset = queryset or models.Document.objects.all()
 
         while True:
             documents_batch = list(
-                models.Document.objects.filter(
+                queryset.filter(
                     id__gt=last_id,
                 ).order_by("id")[: self.batch_size]
             )

src/backend/core/signals.py

@@ -9,7 +9,7 @@
 from django.dispatch import receiver
 
 from . import models
-from .tasks.search import trigger_document_indexer
+from .tasks.search import trigger_batch_document_indexer
 
 
 @receiver(signals.post_save, sender=models.Document)
@@ -19,7 +19,7 @@ def document_post_save(sender, instance, **kwargs):  # pylint: disable=unused-ar
     Note : Within the transaction we can have an empty content and a serialization
     error.
     """
-    transaction.on_commit(partial(trigger_document_indexer, instance))
+    transaction.on_commit(partial(trigger_batch_document_indexer, instance))
 
 
 @receiver(signals.post_save, sender=models.DocumentAccess)
@@ -28,4 +28,6 @@ def document_access_post_save(sender, instance, created, **kwargs):  # pylint: d
     Asynchronous call to the document indexer at the end of the transaction.
     """
     if not created:
-        transaction.on_commit(partial(trigger_document_indexer, instance.document))
+        transaction.on_commit(
+            partial(trigger_batch_document_indexer, instance.document)
+        )

src/backend/core/tasks/search.py

@@ -4,12 +4,12 @@
 
 from django.conf import settings
 from django.core.cache import cache
+from django.db.models import Q
 
 from django_redis.cache import RedisCache
 
 from core import models
 from core.services.search_indexers import (
-    get_batch_accesses_by_users_and_teams,
     get_document_indexer,
 )
 
@@ -18,16 +18,30 @@
 logger = getLogger(__file__)
 
 
-def indexer_throttle_acquire(document_id, timeout=0, atomic=True):
+@app.task
+def document_indexer_task(document_id):
+    """Celery Task : Sends indexation query for a document."""
+    indexer = get_document_indexer()
+
+    if indexer is None:
+        return
+
+    logger.info("Start document %s indexation", document_id)
+    indexer.index(models.Document.objects.filter(pk=document_id))
+
+
+def batch_indexer_throttle_acquire(timeout: int = 0, atomic: bool = True):
     """
     Enable the task throttle flag for a delay.
     Uses redis locks if available to ensure atomic changes
     """
-    key = f"doc-indexer-throttle-{document_id}"
+    key = "document-batch-indexer-throttle"
 
+    # Redis is used as cache database (not in tests). Use the lock feature here
+    # to ensure atomicity of changes to the throttle flag.
     if isinstance(cache, RedisCache) and atomic:
         with cache.locks(key):
-            return indexer_throttle_acquire(document_id, timeout, atomic=False)
+            return batch_indexer_throttle_acquire(timeout, atomic=False)
 
     # Use add() here :
     #   - set the flag and returns true if not exist
@@ -36,49 +50,48 @@ def indexer_throttle_acquire(document_id, timeout=0, atomic=True):
 
 
 @app.task
-def document_indexer_task(document_id):
-    """Celery Task : Sends indexation query for a document."""
+def batch_document_indexer_task(timestamp):
+    """Celery Task : Sends indexation query for a batch of documents."""
     indexer = get_document_indexer()
 
-    if indexer is None:
-        return
-
-    try:
-        doc = models.Document.objects.get(pk=document_id)
-    except models.Document.DoesNotExist:
-        # Skip the task if the document does not exist.
-        return
-
-    accesses = get_batch_accesses_by_users_and_teams((doc.path,))
-
-    data = indexer.serialize_document(document=doc, accesses=accesses)
+    if indexer:
+        queryset = models.Document.objects.filter(
+            Q(updated_at__gte=timestamp)
+            | Q(deleted_at__gte=timestamp)
+            | Q(ancestors_deleted_at__gte=timestamp)
+        )
 
-    logger.info("Start document %s indexation", document_id)
-    indexer.push(data)
+        count = indexer.index(queryset)
+        logger.info("Indexed %d documents", count)
 
 
-def trigger_document_indexer(document):
+def trigger_batch_document_indexer(item):
     """
     Trigger indexation task with debounce a delay set by the SEARCH_INDEXER_COUNTDOWN setting.
 
     Args:
         document (Document): The document instance.
     """
-    countdown = settings.SEARCH_INDEXER_COUNTDOWN
+    countdown = int(settings.SEARCH_INDEXER_COUNTDOWN)
 
     # DO NOT create a task if indexation if disabled
     if not settings.SEARCH_INDEXER_CLASS:
         return
 
-    # Each time this method is called during a countdown, we increment the
-    # counter and each task decrease it, so the index be run only once.
-    if indexer_throttle_acquire(document.pk, timeout=countdown):
-        logger.info(
-            "Add task for document %s indexation in %.2f seconds",
-            document.pk,
-            countdown,
-        )
-
-        document_indexer_task.apply_async(args=[document.pk])
+    if countdown > 0:
+        # Each time this method is called during a countdown, we increment the
+        # counter and each task decrease it, so the index be run only once.
+        if batch_indexer_throttle_acquire(timeout=countdown):
+            logger.info(
+                "Add task for batch document indexation from updated_at=%s in %d seconds",
+                item.updated_at.isoformat(),
+                countdown,
+            )
+
+            batch_document_indexer_task.apply_async(
+                args=[item.updated_at], countdown=countdown
+            )
+        else:
+            logger.info("Skip task for batch document %s indexation", item.pk)
     else:
-        logger.info("Skip task for document %s indexation", document.pk)
+        document_indexer_task.apply(args=[item.pk])

src/backend/core/tests/conftest.py

@@ -45,6 +45,7 @@ def indexer_settings_fixture(settings):
     settings.SEARCH_INDEXER_QUERY_URL = (
         "http://localhost:8081/api/v1.0/documents/search/"
     )
+    settings.SEARCH_INDEXER_COUNTDOWN = 1
 
     yield settings