Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a4f668929c | |||
| 209622c523 | |||
| 0076df5782 | |||
| e2344dcc17 | |||
| e183ee9ed7 | |||
| 5ce3ab71d3 | |||
| 0d09f761dc | |||
| ce5f9a1417 | |||
| 83a24c3796 |
@@ -6,10 +6,15 @@ and this project adheres to
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Changed
|
||||
|
||||
- 💄(frontend) improve comments highlights #1961
|
||||
|
||||
## [v4.8.3] - 2026-03-23
|
||||
|
||||
### Changed
|
||||
|
||||
- ✨(backend) improve indexing command with checkpoint recovery and asynchronicity
|
||||
- ♿️(frontend) improve version history list accessibility #2033
|
||||
- ♿(frontend) focus skip link on headings and skip grid dropzone #1983
|
||||
- ♿️(frontend) add sr-only format to export download button #2088
|
||||
@@ -17,6 +22,7 @@ and this project adheres to
|
||||
- ✨(frontend) add markdown copy icon for Copy as Markdown option #2096
|
||||
- ♻️(backend) skip saving in database a document when payload is empty #2062
|
||||
- ♻️(frontend) refacto Version modal to fit with the design system #2091
|
||||
- ⚡️(frontend) add debounce WebSocket reconnect #2104
|
||||
|
||||
### Fixed
|
||||
|
||||
|
||||
@@ -0,0 +1,52 @@
|
||||
# Index Command
|
||||
|
||||
The `index` management command is used to index documents to the remote search indexer.
|
||||
|
||||
It sends an asynchronous task to the Celery worker.
|
||||
|
||||
## Usage
|
||||
|
||||
### Command Line
|
||||
|
||||
```bash
|
||||
python manage.py index \
|
||||
--lower-time-bound "2024-01-01T00:00:00" \
|
||||
--upper-time-bound "2024-01-31T23:59:59" \
|
||||
--batch-size 200 \
|
||||
--crash-safe-mode false
|
||||
```
|
||||
|
||||
### Django Admin
|
||||
|
||||
The command is available in the Django admin interface:
|
||||
|
||||
1. Go to `/admin/`
|
||||
2. Click on **"Run Indexing"** in the CORE section
|
||||
3. Fill in the form with the desired parameters
|
||||
4. Click **"Run Indexing Command"**
|
||||
|
||||
## Parameters
|
||||
|
||||
### `--batch-size`
|
||||
- **type:** Integer
|
||||
- **default:** `50`
|
||||
- **description:** Number of documents to process per batch. Higher values may improve performance but use more memory.
|
||||
|
||||
### `--lower-time-bound`
|
||||
- **optional**: true
|
||||
- **type:** ISO 8601 datetime string
|
||||
- **default:** `None`
|
||||
- **description:** Only documents updated after this date will be indexed.
|
||||
|
||||
### `--upper-time-bound` (optional)
|
||||
- **optional**: true
|
||||
- **type:** ISO 8601 datetime string
|
||||
- **default:** `None`
|
||||
- **description:** Only documents updated before this date will be indexed.
|
||||
|
||||
### `--crash-safe-mode`
|
||||
- **type:** Boolean flag
|
||||
- **default:** `True`
|
||||
- **description:** When enabled, the command orders documents by `updated_at` and stores the last indexed document's timestamp in cache. This allows resuming indexing from the last successful batch in case of a crash. However, it is more computationally expensive due to sorting.b
|
||||
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
"""Admin classes and registrations for core app."""
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib import admin, messages
|
||||
from django.contrib.auth import admin as auth_admin
|
||||
from django.shortcuts import redirect
|
||||
from django.core.management import call_command
|
||||
from django.shortcuts import redirect, render
|
||||
from django.utils.translation import gettext_lazy as _
|
||||
|
||||
from treebeard.admin import TreeAdmin
|
||||
@@ -227,3 +229,36 @@ class InvitationAdmin(admin.ModelAdmin):
|
||||
def save_model(self, request, obj, form, change):
|
||||
obj.issuer = request.user
|
||||
obj.save()
|
||||
|
||||
|
||||
@admin.register(models.RunIndexing)
|
||||
class RunIndexingAdmin(admin.ModelAdmin):
|
||||
"""Admin for running indexing commands."""
|
||||
|
||||
def changelist_view(self, request, extra_context=None):
|
||||
"""Override to avoid querying the database and handle form submission."""
|
||||
if request.method == "POST":
|
||||
try:
|
||||
call_command(
|
||||
"index",
|
||||
batch_size=int(request.POST.get("batch_size")),
|
||||
lower_time_bound=request.POST.get("lower_time_bound"),
|
||||
upper_time_bound=request.POST.get("upper_time_bound"),
|
||||
crash_safe_mode=request.POST.get("crash_safe_mode"),
|
||||
)
|
||||
messages.success(request, _("Indexing triggered!"))
|
||||
except Exception as e:
|
||||
messages.error(
|
||||
request,
|
||||
_("Error running indexing command: %(error)s") % {"error": str(e)},
|
||||
)
|
||||
|
||||
return render(
|
||||
request=request,
|
||||
template_name="runindexing.html",
|
||||
context={
|
||||
**self.admin_site.each_context(request),
|
||||
"title": "Run Indexing Command",
|
||||
"default_batch_size": settings.SEARCH_INDEXER_BATCH_SIZE,
|
||||
},
|
||||
)
|
||||
|
||||
@@ -3,11 +3,12 @@ Handle search setup that needs to be done at bootstrap time.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
from django.core.management.base import BaseCommand, CommandError
|
||||
|
||||
from core.services.search_indexers import get_document_indexer
|
||||
from core.tasks.indexing import batch_document_indexer_task
|
||||
|
||||
logger = logging.getLogger("docs.search.bootstrap_search")
|
||||
|
||||
@@ -27,6 +28,31 @@ class Command(BaseCommand):
|
||||
default=50,
|
||||
help="Indexation query batch size",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--lower-time-bound",
|
||||
action="store",
|
||||
dest="lower_time_bound",
|
||||
type=datetime.fromisoformat,
|
||||
default=None,
|
||||
help="DateTime in ISO format. Only documents updated after this date will be indexed",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--upper-time-bound",
|
||||
action="store",
|
||||
dest="upper_time_bound",
|
||||
type=datetime.fromisoformat,
|
||||
default=None,
|
||||
help="DateTime in ISO format. Only documents updated before this date will be indexed",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--crash-safe-mode",
|
||||
action="store_true",
|
||||
dest="crash_safe_mode",
|
||||
default=True,
|
||||
help="If True, order documents by updated_at and store the last indexed document.updated_at in cache. "
|
||||
"This allows resuming indexing from the last successful batch in case of a crash "
|
||||
"but is more computationally expensive due to sorting.",
|
||||
)
|
||||
|
||||
def handle(self, *args, **options):
|
||||
"""Launch and log search index generation."""
|
||||
@@ -35,18 +61,18 @@ class Command(BaseCommand):
|
||||
if not indexer:
|
||||
raise CommandError("The indexer is not enabled or properly configured.")
|
||||
|
||||
logger.info("Starting to regenerate Find index...")
|
||||
start = time.perf_counter()
|
||||
batch_size = options["batch_size"]
|
||||
|
||||
try:
|
||||
count = indexer.index(batch_size=batch_size)
|
||||
batch_document_indexer_task.apply_async(
|
||||
kwargs={
|
||||
"lower_time_bound": options["lower_time_bound"],
|
||||
"upper_time_bound": options["upper_time_bound"],
|
||||
"batch_size": options["batch_size"],
|
||||
"crash_safe_mode": options["crash_safe_mode"],
|
||||
},
|
||||
)
|
||||
except Exception as err:
|
||||
raise CommandError("Unable to regenerate index") from err
|
||||
raise CommandError("Unable send indexing task to worker") from err
|
||||
|
||||
duration = time.perf_counter() - start
|
||||
logger.info(
|
||||
"Search index regenerated from %d document(s) in %.2f seconds.",
|
||||
count,
|
||||
duration,
|
||||
"Document indexing task sent to worker",
|
||||
)
|
||||
|
||||
@@ -0,0 +1,23 @@
|
||||
# Generated by Django 5.2.12 on 2026-03-26 16:34
|
||||
|
||||
from django.db import migrations, models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
dependencies = [
|
||||
('core', '0031_clean_onboarding_accesses'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='RunIndexing',
|
||||
fields=[
|
||||
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'Run Indexing',
|
||||
'managed': False,
|
||||
},
|
||||
),
|
||||
]
|
||||
@@ -2029,3 +2029,13 @@ class Invitation(BaseModel):
|
||||
"partial_update": is_admin_or_owner,
|
||||
"retrieve": is_admin_or_owner,
|
||||
}
|
||||
|
||||
|
||||
class RunIndexing(models.Model):
|
||||
"""Proxy model for indexing management in admin."""
|
||||
|
||||
class Meta:
|
||||
"""Meta options."""
|
||||
|
||||
managed = False
|
||||
verbose_name = _("Run Indexing")
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Document search index management utilities and indexers"""
|
||||
|
||||
import itertools
|
||||
import logging
|
||||
from abc import ABC, abstractmethod
|
||||
from collections import defaultdict
|
||||
@@ -7,6 +8,7 @@ from functools import cache
|
||||
|
||||
from django.conf import settings
|
||||
from django.contrib.auth.models import AnonymousUser
|
||||
from django.core.cache import cache as django_cache
|
||||
from django.core.exceptions import ImproperlyConfigured
|
||||
from django.utils.module_loading import import_string
|
||||
|
||||
@@ -18,6 +20,9 @@ from core.enums import SearchType
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
BULK_INDEXER_CHECKPOINT = "bulk-indexer-checkpoint"
|
||||
|
||||
|
||||
@cache
|
||||
def get_document_indexer():
|
||||
"""Returns an instance of indexer service if enabled and properly configured."""
|
||||
@@ -125,7 +130,7 @@ class BaseDocumentIndexer(ABC):
|
||||
if not self.search_url:
|
||||
raise ImproperlyConfigured("SEARCH_URL must be set in Django settings.")
|
||||
|
||||
def index(self, queryset=None, batch_size=None):
|
||||
def index(self, queryset=None, batch_size=None, crash_safe_mode=False):
|
||||
"""
|
||||
Fetch documents in batches, serialize them, and push to the search backend.
|
||||
|
||||
@@ -134,35 +139,37 @@ class BaseDocumentIndexer(ABC):
|
||||
Defaults to all documents without filter.
|
||||
batch_size (int, optional): Number of documents per batch.
|
||||
Defaults to settings.SEARCH_INDEXER_BATCH_SIZE.
|
||||
crash_safe_mode (bool, optional): If True, order documents by updated_at
|
||||
and store the last indexed document.updated_at in cache.
|
||||
This allows resuming indexing from the last successful batch in case of a crash
|
||||
but is more computationally expensive due to sorting.
|
||||
"""
|
||||
last_id = 0
|
||||
count = 0
|
||||
queryset = queryset or models.Document.objects.all()
|
||||
batch_size = batch_size or self.batch_size
|
||||
|
||||
while True:
|
||||
documents_batch = list(
|
||||
queryset.filter(
|
||||
id__gt=last_id,
|
||||
).order_by("id")[:batch_size]
|
||||
)
|
||||
|
||||
if not documents_batch:
|
||||
break
|
||||
if crash_safe_mode:
|
||||
queryset = queryset.order_by("updated_at")
|
||||
|
||||
for documents_batch in itertools.batched(queryset.iterator(), batch_size):
|
||||
doc_paths = [doc.path for doc in documents_batch]
|
||||
last_id = documents_batch[-1].id
|
||||
accesses_by_document_path = get_batch_accesses_by_users_and_teams(doc_paths)
|
||||
|
||||
serialized_batch = [
|
||||
self.serialize_document(document, accesses_by_document_path)
|
||||
for document in documents_batch
|
||||
if document.content or document.title
|
||||
]
|
||||
|
||||
if serialized_batch:
|
||||
self.push(serialized_batch)
|
||||
count += len(serialized_batch)
|
||||
if not serialized_batch:
|
||||
continue
|
||||
|
||||
self.push(serialized_batch)
|
||||
count += len(serialized_batch)
|
||||
|
||||
if crash_safe_mode:
|
||||
django_cache.set(
|
||||
BULK_INDEXER_CHECKPOINT, serialized_batch[-1]["updated_at"]
|
||||
)
|
||||
|
||||
return count
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ from django.db.models import signals
|
||||
from django.dispatch import receiver
|
||||
|
||||
from core import models
|
||||
from core.tasks.search import trigger_batch_document_indexer
|
||||
from core.tasks.indexing import trigger_batch_document_indexer
|
||||
from core.utils import get_users_sharing_documents_with_cache_key
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,130 @@
|
||||
"""Trigger document indexation using celery task."""
|
||||
|
||||
from logging import getLogger
|
||||
|
||||
from django.conf import settings
|
||||
from django.core.cache import cache
|
||||
|
||||
from django_redis.cache import RedisCache
|
||||
|
||||
from core import models
|
||||
from core.services.search_indexers import (
|
||||
get_document_indexer,
|
||||
)
|
||||
from core.utils import build_indexable_documents_queryset
|
||||
|
||||
from impress.celery_app import app
|
||||
|
||||
logger = getLogger(__file__)
|
||||
|
||||
|
||||
@app.task
|
||||
def document_indexer_task(document_id):
|
||||
"""
|
||||
Celery Task: Indexes a single document by its ID.
|
||||
|
||||
Args:
|
||||
document_id: Primary key of the document to index.
|
||||
"""
|
||||
indexer = get_document_indexer()
|
||||
|
||||
if indexer:
|
||||
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):
|
||||
"""
|
||||
Acquire a throttle lock to prevent multiple batch indexation tasks during countdown.
|
||||
|
||||
implements a debouncing pattern: only the first call during the timeout period
|
||||
will succeed, subsequent calls are skipped until the timeout expires.
|
||||
|
||||
Args:
|
||||
timeout (int): Lock duration in seconds (countdown period).
|
||||
atomic (bool): Use Redis locks for atomic operations if available.
|
||||
|
||||
Returns:
|
||||
bool: True if lock acquired (first call), False if already held (subsequent calls).
|
||||
"""
|
||||
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 batch_indexer_throttle_acquire(timeout, atomic=False)
|
||||
|
||||
# cache.add() is atomic test-and-set operation:
|
||||
# - If key doesn't exist: creates it with timeout and returns True
|
||||
# - If key already exists: does nothing and returns False
|
||||
# The key expires after timeout seconds, releasing the lock.
|
||||
# The value 1 is irrelevant, only the key presence/absence matters.
|
||||
return cache.add(key, 1, timeout=timeout)
|
||||
|
||||
|
||||
@app.task
|
||||
def batch_document_indexer_task(lower_time_bound=None, upper_time_bound=None, **kwargs):
|
||||
"""
|
||||
Celery Task: Batch indexes all documents modified since timestamp.
|
||||
|
||||
Args:
|
||||
lower_time_bound (datetime|isoformat, optional):
|
||||
indexes documents updated or deleted after this timestamp.
|
||||
upper_time_bound (datetime|isoformat, optional):
|
||||
indexes documents updated or deleted before this timestamp.
|
||||
"""
|
||||
indexer = get_document_indexer()
|
||||
|
||||
if not indexer:
|
||||
logger.warning("Indexing task triggered but no indexer configured: skipping")
|
||||
return
|
||||
|
||||
count = indexer.index(
|
||||
queryset=build_indexable_documents_queryset(
|
||||
lower_time_bound=lower_time_bound, upper_time_bound=upper_time_bound
|
||||
),
|
||||
**kwargs,
|
||||
)
|
||||
logger.info("Indexed %d documents", count)
|
||||
|
||||
|
||||
def trigger_batch_document_indexer(document):
|
||||
"""
|
||||
Trigger document indexation with optional debounce mechanism.
|
||||
|
||||
behavior depends on SEARCH_INDEXER_COUNTDOWN setting:
|
||||
- if countdown > 0 sec (async batch mode):
|
||||
* schedules a batch indexation task after countdown in seconds
|
||||
* uses throttle mechanism to ensure only ONE batch task runs per countdown period
|
||||
* all documents modified since first trigger are indexed together
|
||||
- if countdown == 0 sec (sync mode):
|
||||
* executes indexation synchronously in the current thread
|
||||
* no batching, no throttling, no Celery task queuing
|
||||
|
||||
Args:
|
||||
document (Document): the document instance that triggered the indexation.
|
||||
"""
|
||||
countdown = int(settings.SEARCH_INDEXER_COUNTDOWN)
|
||||
|
||||
# DO NOT create a task if indexation is disabled
|
||||
if not settings.SEARCH_INDEXER_CLASS:
|
||||
return
|
||||
|
||||
if countdown > 0:
|
||||
# use throttle to ensure only one task is scheduled per countdown period.
|
||||
# if throttle acquired, schedule batch task; otherwise skip.
|
||||
if batch_indexer_throttle_acquire(timeout=countdown):
|
||||
logger.info(
|
||||
"Add task for batch document indexation from updated_at=%s in %d seconds",
|
||||
document.updated_at.isoformat(),
|
||||
countdown,
|
||||
)
|
||||
|
||||
batch_document_indexer_task.apply_async(
|
||||
kwargs={"lower_time_bound": document.updated_at}, countdown=countdown
|
||||
)
|
||||
else:
|
||||
logger.info("Skip task for batch document %s indexation", document.pk)
|
||||
else:
|
||||
document_indexer_task.apply(args=[document.pk])
|
||||
@@ -1,95 +0,0 @@
|
||||
"""Trigger document indexation using celery task."""
|
||||
|
||||
from logging import getLogger
|
||||
|
||||
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_document_indexer,
|
||||
)
|
||||
|
||||
from impress.celery_app import app
|
||||
|
||||
logger = getLogger(__file__)
|
||||
|
||||
|
||||
@app.task
|
||||
def document_indexer_task(document_id):
|
||||
"""Celery Task : Sends indexation query for a document."""
|
||||
indexer = get_document_indexer()
|
||||
|
||||
if indexer:
|
||||
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 = "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 batch_indexer_throttle_acquire(timeout, atomic=False)
|
||||
|
||||
# Use add() here :
|
||||
# - set the flag and returns true if not exist
|
||||
# - do nothing and return false if exist
|
||||
return cache.add(key, 1, timeout=timeout)
|
||||
|
||||
|
||||
@app.task
|
||||
def batch_document_indexer_task(timestamp):
|
||||
"""Celery Task : Sends indexation query for a batch of documents."""
|
||||
indexer = get_document_indexer()
|
||||
|
||||
if indexer:
|
||||
queryset = models.Document.objects.filter(
|
||||
Q(updated_at__gte=timestamp)
|
||||
| Q(deleted_at__gte=timestamp)
|
||||
| Q(ancestors_deleted_at__gte=timestamp)
|
||||
)
|
||||
|
||||
count = indexer.index(queryset)
|
||||
logger.info("Indexed %d documents", count)
|
||||
|
||||
|
||||
def trigger_batch_document_indexer(document):
|
||||
"""
|
||||
Trigger indexation task with debounce a delay set by the SEARCH_INDEXER_COUNTDOWN setting.
|
||||
|
||||
Args:
|
||||
document (Document): The document instance.
|
||||
"""
|
||||
countdown = int(settings.SEARCH_INDEXER_COUNTDOWN)
|
||||
|
||||
# DO NOT create a task if indexation if disabled
|
||||
if not settings.SEARCH_INDEXER_CLASS:
|
||||
return
|
||||
|
||||
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",
|
||||
document.updated_at.isoformat(),
|
||||
countdown,
|
||||
)
|
||||
|
||||
batch_document_indexer_task.apply_async(
|
||||
args=[document.updated_at], countdown=countdown
|
||||
)
|
||||
else:
|
||||
logger.info("Skip task for batch document %s indexation", document.pk)
|
||||
else:
|
||||
document_indexer_task.apply(args=[document.pk])
|
||||
@@ -0,0 +1,43 @@
|
||||
{% extends "admin/base_site.html" %}
|
||||
{% load i18n %}
|
||||
|
||||
{% block content %}
|
||||
|
||||
<form method="POST" >
|
||||
{% csrf_token %}
|
||||
|
||||
<hr style="margin-bottom: 10px;">
|
||||
|
||||
<p>
|
||||
This command triggers the indexing of all documents within the specified time bound.
|
||||
</p>
|
||||
<p>
|
||||
See <b>docs/commands/index.md</b> for more details.
|
||||
</p>
|
||||
|
||||
<hr style="margin-bottom: 20px;">
|
||||
|
||||
<div>
|
||||
<label for="batch_size" style="margin-right: 10px">Batch size</label>
|
||||
<input id="batch_size" type="number" name="batch_size" value={{ default_batch_size }} style="width: 80px;" >
|
||||
</div>
|
||||
|
||||
<div>
|
||||
<label for="lower_time_bound" style="margin-right: 10px">Lower time bound (optional)</label>
|
||||
<input id="lower_time_bound" type="datetime-local" name="lower_time_bound" >
|
||||
</div>
|
||||
|
||||
<div>
|
||||
<label for="upper_time_bound" style="margin-right: 10px">Upper time bound (optional)</label>
|
||||
<input id="upper_time_bound" type="datetime-local" name="upper_time_bound" >
|
||||
</div>
|
||||
|
||||
<div>
|
||||
<label for="crash_safe_mode" style="margin-right: 10px">Crash save mode</label>
|
||||
<input id="crash_safe_mode" type="checkbox" name="crash_safe_mode" value="true" checked>
|
||||
</div>
|
||||
|
||||
<input type="submit" value="{% translate 'Run Indexing' %}" style="margin-top: 20px;">
|
||||
</form>
|
||||
|
||||
{% endblock %}
|
||||
@@ -2,21 +2,24 @@
|
||||
Unit test for `index` command.
|
||||
"""
|
||||
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from operator import itemgetter
|
||||
from unittest import mock
|
||||
|
||||
from django.core.cache import cache
|
||||
from django.core.management import CommandError, call_command
|
||||
from django.db import transaction
|
||||
|
||||
import pytest
|
||||
|
||||
from core import factories
|
||||
from core.services.search_indexers import FindDocumentIndexer
|
||||
from core.services.search_indexers import BULK_INDEXER_CHECKPOINT, FindDocumentIndexer
|
||||
from core.tests.conftest import create_document_with_updated_at
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.usefixtures("indexer_settings")
|
||||
def test_index():
|
||||
def test_index_without_bound_success():
|
||||
"""Test the command `index` that run the Find app indexer for all the available documents."""
|
||||
user = factories.UserFactory()
|
||||
indexer = FindDocumentIndexer()
|
||||
@@ -37,20 +40,152 @@ def test_index():
|
||||
}
|
||||
|
||||
with mock.patch.object(FindDocumentIndexer, "push") as mock_push:
|
||||
call_command("index")
|
||||
call_command("index", crash_safe_mode=False)
|
||||
|
||||
push_call_args = [call.args[0] for call in mock_push.call_args_list]
|
||||
push_call_args = [call.args[0] for call in mock_push.call_args_list]
|
||||
|
||||
# called once but with a batch of docs
|
||||
mock_push.assert_called_once()
|
||||
# called once but with a batch of docs
|
||||
mock_push.assert_called_once()
|
||||
|
||||
assert sorted(push_call_args[0], key=itemgetter("id")) == sorted(
|
||||
[
|
||||
indexer.serialize_document(doc, accesses),
|
||||
indexer.serialize_document(no_title_doc, accesses),
|
||||
],
|
||||
key=itemgetter("id"),
|
||||
assert sorted(push_call_args[0], key=itemgetter("id")) == sorted(
|
||||
[
|
||||
indexer.serialize_document(doc, accesses),
|
||||
indexer.serialize_document(no_title_doc, accesses),
|
||||
],
|
||||
key=itemgetter("id"),
|
||||
)
|
||||
|
||||
# crash_safe_mode deactivated -> no checkpoint stored
|
||||
assert cache.get(BULK_INDEXER_CHECKPOINT) is None
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.usefixtures("indexer_settings")
|
||||
def test_index_with_both_bounds_success():
|
||||
"""Test the command `index` for all documents within time bound."""
|
||||
cache.clear()
|
||||
lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
|
||||
upper_time_bound = lower_time_bound + timedelta(days=30)
|
||||
|
||||
document_too_early = create_document_with_updated_at(
|
||||
lower_time_bound - timedelta(days=10)
|
||||
)
|
||||
document_in_window_1 = create_document_with_updated_at(
|
||||
lower_time_bound + timedelta(days=5)
|
||||
)
|
||||
document_in_window_2 = create_document_with_updated_at(
|
||||
lower_time_bound + timedelta(days=15)
|
||||
)
|
||||
document_too_late = create_document_with_updated_at(
|
||||
upper_time_bound + timedelta(days=10)
|
||||
)
|
||||
|
||||
with mock.patch.object(FindDocumentIndexer, "push") as mock_push:
|
||||
call_command(
|
||||
"index",
|
||||
lower_time_bound=lower_time_bound.isoformat(),
|
||||
upper_time_bound=upper_time_bound.isoformat(),
|
||||
)
|
||||
all_push_call_args = [
|
||||
document["id"]
|
||||
for call_arg_list in mock_push.call_args_list
|
||||
for document in call_arg_list.args[0]
|
||||
]
|
||||
|
||||
# Only documents in window should be indexed
|
||||
assert str(document_too_early.id) not in all_push_call_args
|
||||
assert str(document_in_window_1.id) in all_push_call_args
|
||||
assert str(document_in_window_2.id) in all_push_call_args
|
||||
assert str(document_too_late.id) not in all_push_call_args
|
||||
|
||||
# Checkpoint should be set to last indexed document's updated_at
|
||||
assert (
|
||||
cache.get(BULK_INDEXER_CHECKPOINT)
|
||||
== document_in_window_2.updated_at.isoformat()
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
@pytest.mark.usefixtures("indexer_settings")
|
||||
def test_index_with_crash_recovery():
|
||||
"""Test resuming indexing from checkpoint after a crash."""
|
||||
cache.clear()
|
||||
lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
|
||||
upper_time_bound = lower_time_bound + timedelta(days=60)
|
||||
|
||||
batch_size = 2
|
||||
documents = [
|
||||
# batch 0
|
||||
create_document_with_updated_at(lower_time_bound + timedelta(days=5)),
|
||||
create_document_with_updated_at(lower_time_bound + timedelta(days=10)),
|
||||
# batch 1
|
||||
create_document_with_updated_at(lower_time_bound + timedelta(days=20)),
|
||||
create_document_with_updated_at(lower_time_bound + timedelta(days=25)),
|
||||
# batch 2 - will crash here
|
||||
create_document_with_updated_at(lower_time_bound + timedelta(days=30)),
|
||||
create_document_with_updated_at(lower_time_bound + timedelta(days=35)),
|
||||
# batch 3
|
||||
create_document_with_updated_at(lower_time_bound + timedelta(days=40)),
|
||||
create_document_with_updated_at(lower_time_bound + timedelta(days=45)),
|
||||
]
|
||||
|
||||
# First run: simulate crash on batch 3
|
||||
with mock.patch.object(FindDocumentIndexer, "push") as mock_push:
|
||||
|
||||
def push_with_failure(data):
|
||||
# Crash when we encounter document at index 4 (batch 2 with batch_size=2)
|
||||
if str(documents[4].id) in [document["id"] for document in data]:
|
||||
raise ConnectionError("Simulated indexing error")
|
||||
|
||||
mock_push.side_effect = push_with_failure
|
||||
|
||||
call_command(
|
||||
"index",
|
||||
batch_size=batch_size,
|
||||
lower_time_bound=lower_time_bound.isoformat(),
|
||||
upper_time_bound=upper_time_bound.isoformat(),
|
||||
)
|
||||
all_push_call_args = [
|
||||
document["id"]
|
||||
for call_arg_list in mock_push.call_args_list
|
||||
for document in call_arg_list.args[0]
|
||||
]
|
||||
|
||||
# First 2 batches should have been indexed and checkpoint saved
|
||||
checkpoint = cache.get(BULK_INDEXER_CHECKPOINT)
|
||||
assert checkpoint == documents[3].updated_at.isoformat()
|
||||
# first 2 batches should be indexed successfully
|
||||
for i in range(0, 4):
|
||||
assert str(documents[i].id) in all_push_call_args
|
||||
# next batch should have been attempted but failed
|
||||
for i in range(4, 6):
|
||||
assert str(documents[i].id) in all_push_call_args
|
||||
# last batches indexing should not have been attempted
|
||||
for i in range(6, 8):
|
||||
assert str(documents[i].id) not in all_push_call_args
|
||||
|
||||
# Second run: resume from checkpoint
|
||||
with mock.patch.object(FindDocumentIndexer, "push") as mock_push:
|
||||
call_command(
|
||||
"index",
|
||||
batch_size=batch_size,
|
||||
lower_time_bound=checkpoint,
|
||||
upper_time_bound=upper_time_bound.isoformat(),
|
||||
)
|
||||
all_push_call_args = [
|
||||
document["id"]
|
||||
for call_arg_list in mock_push.call_args_list
|
||||
for document in call_arg_list.args[0]
|
||||
]
|
||||
|
||||
# first 2 batches should NOT be re-indexed
|
||||
# except the last document of the last batch which is on the checkpoint boundary
|
||||
for i in range(0, 3):
|
||||
assert str(documents[i].id) not in all_push_call_args
|
||||
# next batches should be indexed including the document at the checkpoint boundary
|
||||
# which has already been indexed and is re-indexed
|
||||
for i in range(4, 8):
|
||||
assert str(documents[i].id) in all_push_call_args
|
||||
|
||||
|
||||
@pytest.mark.django_db
|
||||
|
||||
@@ -8,7 +8,7 @@ from django.core.cache import cache
|
||||
import pytest
|
||||
import responses
|
||||
|
||||
from core import factories
|
||||
from core import factories, models
|
||||
from core.tests.utils.urls import reload_urls
|
||||
|
||||
USER = "user"
|
||||
@@ -143,3 +143,10 @@ def user_token():
|
||||
A fixture to create a user token for testing.
|
||||
"""
|
||||
return build_authorization_bearer("some_token")
|
||||
|
||||
|
||||
def create_document_with_updated_at(updated_at, **kwargs):
|
||||
"""Helper to create a document with a specific updated_at, bypassing auto_now=True."""
|
||||
document = factories.DocumentFactory(**kwargs)
|
||||
models.Document.objects.filter(pk=document.pk).update(updated_at=updated_at)
|
||||
return models.Document.objects.get(pk=document.pk)
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
import base64
|
||||
import uuid
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
from django.core.cache import cache
|
||||
|
||||
@@ -9,6 +10,7 @@ import pycrdt
|
||||
import pytest
|
||||
|
||||
from core import factories, utils
|
||||
from core.tests.conftest import create_document_with_updated_at
|
||||
|
||||
pytestmark = pytest.mark.django_db
|
||||
|
||||
@@ -240,3 +242,127 @@ def test_utils_get_value_by_pattern_no_match():
|
||||
result = utils.get_value_by_pattern(data, r"^title\.")
|
||||
|
||||
assert result == []
|
||||
|
||||
|
||||
def test_utils_build_indexable_documents_queryset_no_filters():
|
||||
"""Test build_indexable_documents_queryset with no filters returns all documents."""
|
||||
factories.DocumentFactory.create_batch(3)
|
||||
|
||||
queryset = utils.build_indexable_documents_queryset()
|
||||
|
||||
assert queryset.count() == 3
|
||||
|
||||
|
||||
def test_utils_build_indexable_documents_queryset_oldest_updated_at():
|
||||
"""Test filtering by oldest_updated_at includes documents updated after timestamp."""
|
||||
lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
|
||||
|
||||
document_too_old = create_document_with_updated_at(
|
||||
updated_at=lower_time_bound - timedelta(days=30)
|
||||
)
|
||||
document_at_boundary = create_document_with_updated_at(lower_time_bound)
|
||||
document_recent = create_document_with_updated_at(
|
||||
updated_at=lower_time_bound + timedelta(days=15)
|
||||
)
|
||||
|
||||
queryset = utils.build_indexable_documents_queryset(
|
||||
lower_time_bound=lower_time_bound
|
||||
)
|
||||
|
||||
assert queryset.count() == 2
|
||||
assert document_too_old not in queryset
|
||||
assert document_at_boundary in queryset
|
||||
assert document_recent in queryset
|
||||
|
||||
|
||||
def test_utils_build_indexable_documents_queryset_newest_updated_at():
|
||||
"""Test filtering by newest_updated_at includes documents updated before timestamp."""
|
||||
upper_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
|
||||
|
||||
document_old = create_document_with_updated_at(
|
||||
updated_at=upper_time_bound - timedelta(days=30)
|
||||
)
|
||||
document_at_boundary = create_document_with_updated_at(upper_time_bound)
|
||||
document_too_recent = create_document_with_updated_at(
|
||||
updated_at=upper_time_bound + timedelta(days=15)
|
||||
)
|
||||
|
||||
queryset = utils.build_indexable_documents_queryset(
|
||||
upper_time_bound=upper_time_bound
|
||||
)
|
||||
|
||||
assert queryset.count() == 2
|
||||
assert document_old in queryset
|
||||
assert document_at_boundary in queryset
|
||||
assert document_too_recent not in queryset
|
||||
|
||||
|
||||
def test_utils_build_indexable_documents_queryset_both_bounds():
|
||||
"""Test filtering with both oldest and newest bounds."""
|
||||
lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
|
||||
upper_time_bound = lower_time_bound + timedelta(days=30)
|
||||
|
||||
document_too_early = create_document_with_updated_at(
|
||||
updated_at=lower_time_bound - timedelta(days=10)
|
||||
)
|
||||
document_in_window = create_document_with_updated_at(
|
||||
updated_at=lower_time_bound + timedelta(days=5)
|
||||
)
|
||||
other_document_in_window = create_document_with_updated_at(
|
||||
updated_at=lower_time_bound + timedelta(days=15)
|
||||
)
|
||||
document_too_late = create_document_with_updated_at(
|
||||
updated_at=upper_time_bound + timedelta(days=10)
|
||||
)
|
||||
|
||||
queryset = utils.build_indexable_documents_queryset(
|
||||
lower_time_bound=lower_time_bound, upper_time_bound=upper_time_bound
|
||||
)
|
||||
|
||||
assert queryset.count() == 2
|
||||
assert document_too_early not in queryset
|
||||
assert document_in_window in queryset
|
||||
assert other_document_in_window in queryset
|
||||
assert document_too_late not in queryset
|
||||
|
||||
|
||||
def test_utils_build_indexable_documents_queryset_deleted_at():
|
||||
"""Test filtering includes documents with deleted_at in range."""
|
||||
lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
|
||||
|
||||
document_deleted_in_range = create_document_with_updated_at(
|
||||
updated_at=lower_time_bound - timedelta(days=30),
|
||||
deleted_at=lower_time_bound + timedelta(days=15),
|
||||
)
|
||||
document_updated_before_range = create_document_with_updated_at(
|
||||
updated_at=lower_time_bound - timedelta(days=30)
|
||||
)
|
||||
|
||||
queryset = utils.build_indexable_documents_queryset(
|
||||
lower_time_bound=lower_time_bound
|
||||
)
|
||||
|
||||
assert queryset.count() == 1
|
||||
assert document_deleted_in_range in queryset
|
||||
assert document_updated_before_range not in queryset
|
||||
|
||||
|
||||
def test_utils_build_indexable_documents_queryset_ancestors_deleted_at():
|
||||
"""Test filtering includes documents with ancestors_deleted_at in range."""
|
||||
lower_time_bound = datetime(2024, 2, 1, tzinfo=timezone.utc)
|
||||
|
||||
document_ancestor_deleted_in_range = create_document_with_updated_at(
|
||||
updated_at=lower_time_bound - timedelta(days=30),
|
||||
ancestors_deleted_at=lower_time_bound + timedelta(days=15),
|
||||
)
|
||||
document_updated_before_range = create_document_with_updated_at(
|
||||
updated_at=lower_time_bound - timedelta(days=30)
|
||||
)
|
||||
|
||||
queryset = utils.build_indexable_documents_queryset(
|
||||
lower_time_bound=lower_time_bound
|
||||
)
|
||||
|
||||
assert queryset.count() == 1
|
||||
assert document_ancestor_deleted_in_range in queryset
|
||||
assert document_updated_before_range not in queryset
|
||||
|
||||
@@ -8,7 +8,7 @@ from collections import defaultdict
|
||||
|
||||
from django.core.cache import cache
|
||||
from django.db import models as db
|
||||
from django.db.models import Subquery
|
||||
from django.db.models import Q, Subquery
|
||||
|
||||
import pycrdt
|
||||
from bs4 import BeautifulSoup
|
||||
@@ -168,3 +168,47 @@ def users_sharing_documents_with(user):
|
||||
elapsed,
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
def build_indexable_documents_queryset(lower_time_bound=None, upper_time_bound=None):
|
||||
"""
|
||||
Build a queryset of documents to index filtered by timestamp range.
|
||||
|
||||
filters out documents that have been updated, or deleted outside the time range.
|
||||
|
||||
Args:
|
||||
lower_time_bound (datetime, optional):
|
||||
keeps documents updated or deleted after this timestamp.
|
||||
upper_time_bound (datetime, optional):
|
||||
keeps documents updated or deleted before this timestamp.
|
||||
|
||||
Returns:
|
||||
QuerySet: Filtered Document queryset ready for indexation.
|
||||
"""
|
||||
|
||||
queryset = models.Document.objects.all()
|
||||
|
||||
conditions = Q()
|
||||
if lower_time_bound and upper_time_bound:
|
||||
conditions = (
|
||||
Q(updated_at__gte=lower_time_bound, updated_at__lte=upper_time_bound)
|
||||
| Q(deleted_at__gte=lower_time_bound, deleted_at__lte=upper_time_bound)
|
||||
| Q(
|
||||
ancestors_deleted_at__gte=lower_time_bound,
|
||||
ancestors_deleted_at__lte=upper_time_bound,
|
||||
)
|
||||
)
|
||||
elif lower_time_bound:
|
||||
conditions = (
|
||||
Q(updated_at__gte=lower_time_bound)
|
||||
| Q(deleted_at__gte=lower_time_bound)
|
||||
| Q(ancestors_deleted_at__gte=lower_time_bound)
|
||||
)
|
||||
elif upper_time_bound:
|
||||
conditions = (
|
||||
Q(updated_at__lte=upper_time_bound)
|
||||
| Q(deleted_at__lte=upper_time_bound)
|
||||
| Q(ancestors_deleted_at__lte=upper_time_bound)
|
||||
)
|
||||
|
||||
return queryset.filter(conditions)
|
||||
|
||||
@@ -131,12 +131,13 @@ test.describe('Doc Comments', () => {
|
||||
await thread.getByRole('paragraph').first().fill('This is a comment');
|
||||
await thread.locator('[data-test="save"]').click();
|
||||
await expect(thread.getByText('This is a comment').first()).toBeHidden();
|
||||
await expect(editor.getByText('Hello')).toHaveClass('bn-thread-mark');
|
||||
|
||||
// Check background color changed
|
||||
await expect(editor.getByText('Hello')).toHaveCSS(
|
||||
'background-color',
|
||||
'rgba(237, 180, 0, 0.4)',
|
||||
'color(srgb 0.882353 0.831373 0.717647 / 0.4)',
|
||||
);
|
||||
|
||||
await editor.first().click();
|
||||
await editor.getByText('Hello').click();
|
||||
|
||||
@@ -185,6 +186,7 @@ test.describe('Doc Comments', () => {
|
||||
await thread.getByText('This is an edited comment').first().hover();
|
||||
await thread.locator('[data-test="resolve"]').click();
|
||||
await expect(thread).toBeHidden();
|
||||
|
||||
await expect(editor.getByText('Hello')).toHaveCSS(
|
||||
'background-color',
|
||||
'rgba(0, 0, 0, 0)',
|
||||
@@ -196,11 +198,13 @@ test.describe('Doc Comments', () => {
|
||||
|
||||
await thread.getByRole('paragraph').first().fill('This is a new comment');
|
||||
await thread.locator('[data-test="save"]').click();
|
||||
await expect(editor.getByText('Hello')).toHaveClass('bn-thread-mark');
|
||||
|
||||
await expect(editor.getByText('Hello')).toHaveCSS(
|
||||
'background-color',
|
||||
'rgba(237, 180, 0, 0.4)',
|
||||
'color(srgb 0.882353 0.831373 0.717647 / 0.4)',
|
||||
);
|
||||
|
||||
await editor.first().click();
|
||||
await editor.getByText('Hello').click();
|
||||
|
||||
@@ -208,6 +212,7 @@ test.describe('Doc Comments', () => {
|
||||
await thread.locator('[data-test="moreactions"]').first().click();
|
||||
await getMenuItem(thread, 'Delete comment').click();
|
||||
|
||||
await expect(editor.getByText('Hello')).not.toHaveClass('bn-thread-mark');
|
||||
await expect(editor.getByText('Hello')).toHaveCSS(
|
||||
'background-color',
|
||||
'rgba(0, 0, 0, 0)',
|
||||
@@ -263,7 +268,7 @@ test.describe('Doc Comments', () => {
|
||||
|
||||
await expect(otherEditor.getByText('Hello')).toHaveCSS(
|
||||
'background-color',
|
||||
'rgba(237, 180, 0, 0.4)',
|
||||
'color(srgb 0.882353 0.831373 0.717647 / 0.4)',
|
||||
);
|
||||
|
||||
// We change the role of the second user to reader
|
||||
@@ -298,7 +303,7 @@ test.describe('Doc Comments', () => {
|
||||
|
||||
await expect(otherEditor.getByText('Hello')).toHaveCSS(
|
||||
'background-color',
|
||||
'rgba(237, 180, 0, 0.4)',
|
||||
'color(srgb 0.882353 0.831373 0.717647 / 0.4)',
|
||||
);
|
||||
await otherEditor.getByText('Hello').click();
|
||||
await expect(
|
||||
@@ -344,7 +349,7 @@ test.describe('Doc Comments', () => {
|
||||
|
||||
await expect(editor1.getByText('Document One')).toHaveCSS(
|
||||
'background-color',
|
||||
'rgba(237, 180, 0, 0.4)',
|
||||
'color(srgb 0.882353 0.831373 0.717647 / 0.4)',
|
||||
);
|
||||
|
||||
await editor1.getByText('Document One').click();
|
||||
|
||||
+29
-4
@@ -8,12 +8,37 @@ export const cssComments = (
|
||||
& .--docs--main-editor .ProseMirror {
|
||||
// Comments marks in the editor
|
||||
.bn-editor {
|
||||
.bn-thread-mark:not([data-orphan='true']),
|
||||
.bn-thread-mark-selected:not([data-orphan='true']) {
|
||||
background: ${canSeeComment ? '#EDB40066' : 'transparent'};
|
||||
color: var(--c--globals--colors--gray-700);
|
||||
// Resets blocknote comments styles
|
||||
.bn-thread-mark,
|
||||
.bn-thread-mark-selected {
|
||||
background-color: transparent;
|
||||
}
|
||||
|
||||
${canSeeComment &&
|
||||
css`
|
||||
.bn-thread-mark:not([data-orphan='true']) {
|
||||
background-color: color-mix(
|
||||
in srgb,
|
||||
var(--c--contextuals--background--palette--yellow--tertiary) 40%,
|
||||
transparent
|
||||
);
|
||||
border-bottom: 2px solid
|
||||
var(--c--contextuals--background--palette--yellow--secondary);
|
||||
|
||||
mix-blend-mode: multiply;
|
||||
|
||||
transition:
|
||||
background-color var(--c--globals--transitions--duration),
|
||||
border-bottom-color var(--c--globals--transitions--duration);
|
||||
|
||||
&:has(.bn-thread-mark-selected) {
|
||||
background-color: var(
|
||||
--c--contextuals--background--palette--yellow--tertiary
|
||||
);
|
||||
}
|
||||
}
|
||||
`}
|
||||
|
||||
[data-show-selection] {
|
||||
color: HighlightText;
|
||||
}
|
||||
|
||||
+5
-1
@@ -30,6 +30,8 @@ const defaultValues = {
|
||||
|
||||
type ExtendedCloseEvent = CloseEvent & { wasClean: boolean };
|
||||
|
||||
let reconnectTimeout: ReturnType<typeof setTimeout> | undefined;
|
||||
|
||||
export const useProviderStore = create<UseCollaborationStore>((set, get) => ({
|
||||
...defaultValues,
|
||||
createProvider: (wsUrl, storeId, initialDoc) => {
|
||||
@@ -60,7 +62,8 @@ export const useProviderStore = create<UseCollaborationStore>((set, get) => ({
|
||||
return;
|
||||
}
|
||||
|
||||
void provider.connect();
|
||||
clearTimeout(reconnectTimeout);
|
||||
reconnectTimeout = setTimeout(() => void provider.connect(), 1000);
|
||||
}
|
||||
},
|
||||
onAuthenticationFailed() {
|
||||
@@ -119,6 +122,7 @@ export const useProviderStore = create<UseCollaborationStore>((set, get) => ({
|
||||
return provider;
|
||||
},
|
||||
destroyProvider: () => {
|
||||
clearTimeout(reconnectTimeout);
|
||||
const provider = get().provider;
|
||||
if (provider) {
|
||||
provider.destroy();
|
||||
|
||||
Reference in New Issue
Block a user