⚡️(backend) coalesce and pace the resets of collaboration connections

For big documents, calling yhib for each sub documents can be a big
amount of request at the same time. We decided to group reset by batch
in the path order and call again the reset celery task with a delay to
continue.

Fix #2656

CHANGELOG.md

@@ -83,6 +83,7 @@ and this project adheres to
 - 🐛(docker) pull minio images from pgsty
 - 🐛(backend) retry the duplicate of a document on a tree path collision
 - 🐛(backend) reset the collaboration connections on every access change
+- ⚡️(backend) coalesce and pace the resets of collaboration connections
 - 🐛(frontend) clear callout background on Backspace #2052
 - 🐛(export) keep image aspect ratio in PDF columns #2670
 - 🐛(frontend) fix redirect after deleting a document #2706

documentation/env.md

@@ -171,6 +171,8 @@ These are the environment variables you can set for the `impress-backend` contai
 | YHUB_API_TIMEOUT                                | Timeout (in seconds) of the requests to the yhub API                                                                                                                       | 30                                                                      |
 | YHUB_MIGRATION_TIMEOUT                          | Timeout (in seconds) of the call replaying the legacy history of one document, which reads every one of its S3 versions                                                     | 600                                                                     |
 | YHUB_ORG                                        | yhub organization the documents live in. Must match the YHUB_ORG of the yhub server                                                                                        | docs                                                                    |
+| YHUB_RESET_CONNECTIONS_BATCH_SIZE               | Documents whose connections are re-checked in one run of the walk down a subtree that a change of accesses triggers, before the rest is queued again                       | 50                                                                      |
+| YHUB_RESET_CONNECTIONS_DELAY                    | Seconds between two runs of that walk                                                                                                                                      | 1.0                                                                     |
 | Y_PROVIDER_API_BASE_URL                         | Y Provider url                                                                                                                                                             |                                                                         |
 | Y_PROVIDER_API_KEY                              | Key exempting the calls of the collaboration server from the API throttling, sent as X-Y-Provider-Key. Set the same value on the yhub container                             |                                                                         |
 

src/backend/core/tasks/access.py

@@ -1,9 +1,9 @@
 """Tasks dedicated to document's accesses."""
 
-from functools import partial
 from logging import getLogger
 
 from django.apps import apps
+from django.conf import settings
 from django.db import transaction
 
 from core.services.yhub_services import YHubError, YHubService
@@ -12,9 +12,12 @@ from impress.celery_app import app
 
 logger = getLogger(__name__)
 
+# where the resets queued by the current transaction wait on the connection
+PENDING_RESETS_ATTRIBUTE = "yhub_pending_resets"
+
 
 @app.task
-def reset_service_connections_in_cascade(document_id, user_id=None):
+def reset_service_connections_in_cascade(document_id, user_id=None, after_path=None):
     """
     Reset the connections of a document and all its descendants on the
     collaboration server.
@@ -24,6 +27,12 @@ def reset_service_connections_in_cascade(document_id, user_id=None):
     connection of each document and disconnects the ones that lost their
     access. The endpoint is document scoped, hence the walk down the tree.
 
+    The walk is paced: a run resets at most YHUB_RESET_CONNECTIONS_BATCH_SIZE
+    documents, in the order of their paths, and queues the rest after
+    YHUB_RESET_CONNECTIONS_DELAY seconds, from the last path it reached. A
+    large subtree is thus spread over time rather than fired at the
+    collaboration server at once, and no worker is held for its whole length.
+
     A document failing is logged and does not stop the ones after it, its
     clients keep the rights they connected with until they reconnect.
     """
@@ -41,7 +50,14 @@ def reset_service_connections_in_cascade(document_id, user_id=None):
 
     documents = document_model.objects.filter(
         path__startswith=document.path, depth__gte=document.depth
-    ).order_by("path")
+    )
+    if after_path:
+        documents = documents.filter(path__gt=after_path)
+    batch_size = settings.YHUB_RESET_CONNECTIONS_BATCH_SIZE
+    # one more than the batch: whether anything is left to queue
+    documents = list(documents.order_by("path")[: batch_size + 1])
+    remaining = documents[batch_size:]
+    documents = documents[:batch_size]
 
     service = YHubService()
     for doc in documents:
@@ -50,6 +66,57 @@ def reset_service_connections_in_cascade(document_id, user_id=None):
         except YHubError:
             logger.exception("impossible to reset connections for document %s", doc.id)
 
+    if remaining:
+        reset_service_connections_in_cascade.apply_async(
+            args=[document_id, user_id, documents[-1].path],
+            countdown=settings.YHUB_RESET_CONNECTIONS_DELAY,
+        )
+
+
+class PendingResets:
+    """
+    The resets queued by one transaction, sent coalesced when it commits.
+
+    A change of accesses often queues several resets of the same document: a
+    move deletes every direct access of the document, one signal each, and
+    then asks for the whole subtree. The users asked for are gathered per
+    document, and a document asked for everybody (no user) is reset once, for
+    everybody, whatever else was asked for it.
+
+    Known limit: an ask made in a savepoint that is rolled back stays in the
+    buffer of its transaction and is sent with it. The task reads the accesses
+    back when it runs, so the reset is superfluous, never wrong.
+    """
+
+    def __init__(self, batch):
+        # the callbacks on commit of the transaction these resets belong to
+        self.batch = batch
+        self.users_by_document = {}
+        self.flushed = False
+
+    def add(self, document_id, user_id=None):
+        """Ask for the reset of a document, for a user or for everybody."""
+        users = self.users_by_document.setdefault(str(document_id), set())
+        users.add(str(user_id) if user_id else None)
+
+    def flush(self):
+        """
+        Queue the tasks, one per document and user, or per document.
+
+        Registered on commit by every ask, so that a rollback of the savepoint
+        of one of them still leaves a flush to run; the first to run sends
+        everything, the ones after it find nothing to send.
+        """
+        if self.flushed:
+            return
+        self.flushed = True
+        for document_id, users in self.users_by_document.items():
+            if None in users:
+                reset_service_connections_in_cascade.delay(document_id, None)
+                continue
+            for user_id in sorted(users):
+                reset_service_connections_in_cascade.delay(document_id, user_id)
+
 
 def reset_service_connections_on_commit(document_id, user_id=None):
     """
@@ -65,11 +132,22 @@ def reset_service_connections_on_commit(document_id, user_id=None):
     what the change of a single access needs; an access granted to a team, or
     a change of the whole scope of a document, names nobody and every
     connection is re-checked.
+
+    The resets of one transaction are gathered and sent coalesced on commit
+    (see `PendingResets`). They wait on the connection, bound to the list of
+    callbacks on commit of their transaction: Django replaces that list on
+    every commit and rollback, so what a rolled back transaction left behind
+    is told apart and dropped.
     """
-    transaction.on_commit(
-        partial(
-            reset_service_connections_in_cascade.delay,
-            str(document_id),
-            str(user_id) if user_id else None,
-        )
-    )
+    connection = transaction.get_connection()
+    pending = getattr(connection, PENDING_RESETS_ATTRIBUTE, None)
+    if (
+        pending is None
+        or pending.flushed
+        or pending.batch is not connection.run_on_commit
+    ):
+        pending = PendingResets(connection.run_on_commit)
+        setattr(connection, PENDING_RESETS_ATTRIBUTE, pending)
+    pending.add(document_id, user_id)
+    # outside of a transaction this flushes right away, hence after the add
+    transaction.on_commit(pending.flush)

src/backend/core/tests/commands/test_clean_document.py

@@ -441,11 +441,12 @@ def test_clean_document_reports_the_documents_it_could_not_erase(
 
 
 def test_clean_document_resets_connections(
-    settings, mock_reset_service_connections, django_capture_on_commit_callbacks
+    settings, mock_reset_service_connections, capture_service_resets
 ):
     """
     The link definition of the root changes and its accesses but the owners'
-    are deleted: the collaboration server should re-check its connections.
+    are deleted: the collaboration server should re-check every connection of
+    the root, once, the resets of the deleted accesses being coalesced into it.
     """
     settings.DEBUG = True
     root = factories.DocumentFactory(
@@ -457,16 +458,15 @@ def test_clean_document_resets_connections(
     readers = factories.UserDocumentAccessFactory.create_batch(
         2, document=root, role=choices.RoleChoices.READER
     )
-    mock_reset_service_connections.reset_mock()
 
     with (
         mock.patch("core.management.commands.clean_document.default_storage"),
-        django_capture_on_commit_callbacks(execute=True),
+        capture_service_resets(),
     ):
         call_command("clean_document", str(root.id), "--force")
 
-    calls = mock_reset_service_connections.call_args_list
-    assert mock.call(str(root.id), None) in calls
-    for reader in readers:
-        assert mock.call(str(root.id), str(reader.user_id)) in calls
-    assert mock.call(str(root.id), str(owner.user_id)) not in calls
+    assert not models.DocumentAccess.objects.filter(
+        pk__in=[reader.pk for reader in readers]
+    ).exists()
+    assert models.DocumentAccess.objects.filter(pk=owner.pk).exists()
+    mock_reset_service_connections.assert_called_once_with(str(root.id), None)

src/backend/core/tests/conftest.py

@@ -1,15 +1,18 @@
 """Fixtures for tests in the impress core application"""
 
 import base64
+from contextlib import contextmanager
 from unittest import mock
 
 from django.core.cache import cache
+from django.db import transaction
 
 import pytest
 import responses
 
 from core import factories
 from core.services.yhub_services import YHubService
+from core.tasks.access import PENDING_RESETS_ATTRIBUTE
 from core.tests.utils.urls import reload_urls, restore_urls
 
 USER = "user"
@@ -58,6 +61,29 @@ def mock_reset_service_connections_fixture():
         yield mock_delay
 
 
+@pytest.fixture(name="capture_service_resets")
+def capture_service_resets_fixture(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """
+    Provide a context manager taking the resets queued by what runs in it.
+
+    The resets of a transaction are coalesced and sent on commit, and a test
+    runs whole in one transaction: what its setup queued is forgotten first,
+    then the callbacks queued on commit by the block are run, and the resets
+    they send are on the mock this yields.
+    """
+
+    @contextmanager
+    def _capture_service_resets():
+        setattr(transaction.get_connection(), PENDING_RESETS_ATTRIBUTE, None)
+        mock_reset_service_connections.reset_mock()
+        with django_capture_on_commit_callbacks(execute=True):
+            yield mock_reset_service_connections
+
+    return _capture_service_resets
+
+
 @pytest.fixture(autouse=True, name="mock_delete_service_documents")
 def mock_delete_service_documents_fixture():
     """Take the deletions of documents queued for the collaboration server, as above."""

src/backend/core/tests/documents/test_api_document_accesses.py

@@ -21,18 +21,17 @@ pytestmark = pytest.mark.django_db
 
 @pytest.fixture(name="mock_reset_connections")
 def mock_reset_connections_fixture(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """
-    Provide a context manager that runs the callbacks queued on commit and
+    Provide a context manager that takes the resets queued on commit and
     asserts the ``reset_service_connections_in_cascade`` Celery task is queued
     exactly once for the given document and user when leaving the context.
     """
 
     @contextmanager
     def _mock_reset_connections(document_id, user_id=None):
-        mock_reset_service_connections.reset_mock()
-        with django_capture_on_commit_callbacks(execute=True):
+        with capture_service_resets():
             yield mock_reset_service_connections
         mock_reset_service_connections.assert_called_once_with(
             str(document_id), user_id

src/backend/core/tests/documents/test_api_document_accesses_create.py

@@ -593,7 +593,7 @@ def test_api_document_accesses_create_resets_connections(
     via,
     mock_user_teams,
     mock_reset_service_connections,
-    django_capture_on_commit_callbacks,
+    capture_service_resets,
 ):
     """
     Creating an access should have the collaboration server re-check the
@@ -603,7 +603,6 @@ def test_api_document_accesses_create_resets_connections(
     client = APIClient()
     client.force_login(user)
     document = factories.DocumentFactory(users=[(user, "owner")])
-    mock_reset_service_connections.reset_mock()
 
     if via == USER:
         other_user = factories.UserFactory()
@@ -614,7 +613,7 @@ def test_api_document_accesses_create_resets_connections(
         data = {"team": "lasuite", "role": "editor"}
         expected_user_id = None
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         response = client.post(
             f"/api/v1.0/documents/{document.id!s}/accesses/", data, format="json"
         )

src/backend/core/tests/documents/test_api_documents_ask_for_access.py

@@ -915,7 +915,7 @@ def test_api_document_ask_for_access_throttling(settings):
 
 
 def test_api_documents_ask_for_access_accept_resets_connections(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """Accepting a request creates or changes an access: the user is re-checked."""
     user = UserFactory()
@@ -923,12 +923,11 @@ def test_api_documents_ask_for_access_accept_resets_connections(
     document_ask_for_access = DocumentAskForAccessFactory(
         document=document, role=RoleChoices.READER
     )
-    mock_reset_service_connections.reset_mock()
 
     client = APIClient()
     client.force_login(user)
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         response = client.post(
             f"/api/v1.0/documents/{document.id}/ask-for-access/{document_ask_for_access.id}/accept/"
         )

src/backend/core/tests/documents/test_api_documents_link_configuration.py

@@ -14,18 +14,17 @@ pytestmark = pytest.mark.django_db
 
 @pytest.fixture(name="mock_reset_connections")
 def mock_reset_connections_fixture(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """
-    Provide a context manager that runs the callbacks queued on commit and
+    Provide a context manager that takes the resets queued on commit and
     asserts the ``reset_service_connections_in_cascade`` Celery task is queued
     exactly once for the given document when leaving the context.
     """
 
     @contextmanager
     def _mock_reset_connections(document_id):
-        mock_reset_service_connections.reset_mock()
-        with django_capture_on_commit_callbacks(execute=True):
+        with capture_service_resets():
             yield mock_reset_service_connections
         mock_reset_service_connections.assert_called_once_with(str(document_id), None)
 

src/backend/core/tests/documents/test_api_documents_move.py

@@ -839,12 +839,13 @@ def test_api_documents_move_scope_change_deletion_is_atomic(monkeypatch):
 
 
 def test_api_documents_move_resets_connections_of_the_moved_document(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """
     A moved document inherits the accesses of other ancestors: every
-    connection of its subtree should be re-checked, and the direct accesses
-    it loses with its scope are each reported as well.
+    connection of its subtree should be re-checked. The direct accesses it
+    loses with its scope are each reported as well, and coalesced into that
+    one reset of everybody.
     """
     user = factories.UserFactory()
     client = APIClient()
@@ -853,9 +854,8 @@ def test_api_documents_move_resets_connections_of_the_moved_document(
     document = factories.DocumentFactory(users=[(user, "owner")])
     other_access = factories.UserDocumentAccessFactory(document=document)
     target = factories.DocumentFactory(users=[(user, "owner")])
-    mock_reset_service_connections.reset_mock()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         response = client.post(
             f"/api/v1.0/documents/{document.id!s}/move/",
             data={
@@ -865,18 +865,12 @@ def test_api_documents_move_resets_connections_of_the_moved_document(
         )
 
     assert response.status_code == 200
-    assert sorted(mock_reset_service_connections.call_args_list, key=str) == sorted(
-        [
-            mock.call(str(document.id), None),
-            mock.call(str(document.id), str(user.id)),
-            mock.call(str(document.id), str(other_access.user_id)),
-        ],
-        key=str,
-    )
+    assert not models.DocumentAccess.objects.filter(pk=other_access.pk).exists()
+    mock_reset_service_connections.assert_called_once_with(str(document.id), None)
 
 
 def test_api_documents_move_resets_connections_when_the_scope_is_kept(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """
     Moving within the same tree touches no direct access, the ancestors change
@@ -888,9 +882,8 @@ def test_api_documents_move_resets_connections_when_the_scope_is_kept(
 
     root = factories.DocumentFactory(users=[(user, "owner")])
     document, sibling = factories.DocumentFactory.create_batch(2, parent=root)
-    mock_reset_service_connections.reset_mock()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         response = client.post(
             f"/api/v1.0/documents/{document.id!s}/move/",
             data={
@@ -904,7 +897,7 @@ def test_api_documents_move_resets_connections_when_the_scope_is_kept(
 
 
 def test_api_documents_move_resets_nothing_when_refused(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """A refused move changes nothing, and reports nothing."""
     user = factories.UserFactory()
@@ -913,9 +906,8 @@ def test_api_documents_move_resets_nothing_when_refused(
 
     document = factories.DocumentFactory(users=[(user, "owner")])
     child = factories.DocumentFactory(parent=document)
-    mock_reset_service_connections.reset_mock()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         response = client.post(
             f"/api/v1.0/documents/{document.id!s}/move/",
             data={

src/backend/core/tests/external_api/test_external_api_documents_link_configuration.py

@@ -64,7 +64,7 @@ def test_external_api_documents_link_configuration_can_be_allowed(
     resource_server_backend,
     user_specific_sub,
     mock_reset_service_connections,
-    django_capture_on_commit_callbacks,
+    capture_service_resets,
 ):
     """
     Connected users SHOULD be allowed to update the link configuration of a document
@@ -90,7 +90,7 @@ def test_external_api_documents_link_configuration_can_be_allowed(
         "link_role": models.LinkRoleChoices.EDITOR,
     }
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         response = client.put(
             f"/external_api/v1.0/documents/{document.id!s}/link-configuration/",
             new_data,

src/backend/core/tests/test_models_user_reconciliation.py

@@ -670,7 +670,7 @@ def test_process_reconciliation_updates_favorites(
 def test_process_reconciliation_resets_connections(
     user_reconciliation_users_and_docs,
     mock_reset_service_connections,
-    django_capture_on_commit_callbacks,
+    capture_service_resets,
 ):
     """
     The accesses are updated in bulk, without the signal: every document
@@ -687,9 +687,8 @@ def test_process_reconciliation_resets_connections(
         inactive_email_checked=True,
         status="ready",
     )
-    mock_reset_service_connections.reset_mock()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         process_reconciliation(
             None, None, models.UserReconciliation.objects.filter(id=rec.id)
         )

src/backend/core/tests/test_models_users.py

@@ -758,7 +758,7 @@ def test_tasks_user_delete_error_during_deletion_should_rollback_deletion(monkey
 
 
 def test_models_users_convert_valid_invitations_resets_connections(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """
     The accesses are created in bulk, without the signal: the connections of
@@ -769,9 +769,8 @@ def test_models_users_convert_valid_invitations_resets_connections(
     other_document = factories.DocumentFactory()
     factories.InvitationFactory(email=email, document=document)
     factories.InvitationFactory(email=email, document=other_document)
-    mock_reset_service_connections.reset_mock()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         user = factories.UserFactory(email=email)
 
     assert sorted(mock_reset_service_connections.call_args_list, key=str) == sorted(
@@ -802,7 +801,6 @@ def test_models_users_delete_reports_to_the_collaboration_server(
     shared = factories.DocumentFactory(users=[(user, "owner"), (other_user, "owner")])
     member = factories.DocumentFactory(users=[(user, "editor")])
     user_id = str(user.id)
-    mock_reset_service_connections.reset_mock()
 
     user.delete()
 

src/backend/core/tests/test_settings.py

@@ -33,6 +33,20 @@ def test_invalid_settings_oidc_email_configuration():
     )
 
 
+def test_settings_yhub_reset_connections_batch_size_zero():
+    """A batch of zero documents would reset nothing and is refused."""
+
+    class TestSettings(Base):
+        """Fake test settings."""
+
+        YHUB_RESET_CONNECTIONS_BATCH_SIZE = 0
+
+    with pytest.raises(ValueError) as excinfo:
+        TestSettings().post_setup()
+
+    assert str(excinfo.value) == "YHUB_RESET_CONNECTIONS_BATCH_SIZE must be at least 1."
+
+
 def test_settings_psycopg_pool_not_enabled():
     """
     Test that not changing DB_PSYCOPG_POOL_ENABLED should not configure psycopg in the DATABASES

src/backend/core/tests/test_signals.py

@@ -15,13 +15,13 @@ pytestmark = pytest.mark.django_db
 
 
 def test_signals_document_access_created(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """Creating an access should have the user's connections re-checked."""
     document = factories.DocumentFactory()
     user = factories.UserFactory()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         models.DocumentAccess.objects.create(
             document=document, user=user, role="editor"
         )
@@ -32,12 +32,12 @@ def test_signals_document_access_created(
 
 
 def test_signals_document_access_created_for_a_team(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """An access granted to a team names nobody: every connection is re-checked."""
     document = factories.DocumentFactory()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         models.DocumentAccess.objects.create(
             document=document, team="lasuite", role="reader"
         )
@@ -46,13 +46,12 @@ def test_signals_document_access_created_for_a_team(
 
 
 def test_signals_document_access_updated(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """Saving an access, from wherever, should have the connections re-checked."""
     access = factories.UserDocumentAccessFactory(role="reader")
-    mock_reset_service_connections.reset_mock()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         access.role = "editor"
         access.save()
 
@@ -62,14 +61,13 @@ def test_signals_document_access_updated(
 
 
 def test_signals_document_access_deleted(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """Deleting an access should have the user's connections re-checked."""
     access = factories.UserDocumentAccessFactory()
     document_id, user_id = access.document_id, access.user_id
-    mock_reset_service_connections.reset_mock()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         access.delete()
 
     mock_reset_service_connections.assert_called_once_with(
@@ -78,14 +76,13 @@ def test_signals_document_access_deleted(
 
 
 def test_signals_document_accesses_deleted_in_bulk(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """A queryset deletion goes through the signal for each access."""
     document = factories.DocumentFactory()
     accesses = factories.UserDocumentAccessFactory.create_batch(3, document=document)
-    mock_reset_service_connections.reset_mock()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         models.DocumentAccess.objects.filter(document=document).delete()
 
     assert sorted(mock_reset_service_connections.call_args_list, key=str) == sorted(
@@ -106,13 +103,13 @@ def test_signals_document_reset_is_queued_on_commit(mock_reset_service_connectio
 
 
 def test_signals_document_link_definition_changed(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """Changing the link definition of a document re-checks every connection."""
     document = factories.DocumentFactory(link_reach="restricted", link_role="reader")
     document = models.Document.objects.get(pk=document.pk)
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         document.link_reach = "public"
         document.save()
 
@@ -120,13 +117,13 @@ def test_signals_document_link_definition_changed(
 
 
 def test_signals_document_link_role_changed(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """The link role is part of the link definition."""
     document = factories.DocumentFactory(link_reach="public", link_role="reader")
     document = models.Document.objects.get(pk=document.pk)
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         document.link_role = "editor"
         document.save()
 
@@ -134,13 +131,13 @@ def test_signals_document_link_role_changed(
 
 
 def test_signals_document_link_definition_unchanged(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """Saving a document without touching its link definition queues nothing."""
     document = factories.DocumentFactory(link_reach="public", link_role="reader")
     document = models.Document.objects.get(pk=document.pk)
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         document.title = "renamed"
         document.save()
         document.link_reach = "public"  # the value it already has
@@ -150,13 +147,13 @@ def test_signals_document_link_definition_unchanged(
 
 
 def test_signals_document_link_definition_saved_twice(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """A change is reported once: the resets of a transaction are coalesced."""
     document = factories.DocumentFactory(link_reach="restricted", link_role="reader")
     document = models.Document.objects.get(pk=document.pk)
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         document.link_reach = "authenticated"
         document.save()
         document.save()
@@ -201,14 +198,14 @@ def test_signals_document_link_definition_saved_in_a_rolled_back_savepoint(
 
 
 def test_signals_document_link_definition_refreshed(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """Refreshing from the database takes a new snapshot."""
     document = factories.DocumentFactory(link_reach="restricted", link_role="reader")
     models.Document.objects.filter(pk=document.pk).update(link_reach="public")
     document.refresh_from_db()
 
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         document.save()
 
     mock_reset_service_connections.assert_not_called()
@@ -248,10 +245,10 @@ def test_signals_document_link_definition_deferred_field_read(
 
 
 def test_signals_document_created(
-    mock_reset_service_connections, django_capture_on_commit_callbacks
+    mock_reset_service_connections, capture_service_resets
 ):
     """A new document has no connection to re-check."""
-    with django_capture_on_commit_callbacks(execute=True):
+    with capture_service_resets():
         factories.DocumentFactory(link_reach="public")
 
     mock_reset_service_connections.assert_not_called()

src/backend/core/tests/test_tasks_access.py

@@ -6,6 +6,8 @@ core.tasks.access module.
 from unittest import mock
 from uuid import uuid4
 
+from django.db import transaction
+
 import pytest
 
 from core import factories
@@ -93,11 +95,10 @@ def test_reset_service_connections_on_commit(
     """The reset should be queued once the transaction is committed, ids as strings."""
     document_id, user_id = uuid4(), uuid4()
 
-    with django_capture_on_commit_callbacks(execute=True) as callbacks:
+    with django_capture_on_commit_callbacks(execute=True):
         reset_service_connections_on_commit(document_id, user_id)
         mock_reset_service_connections.assert_not_called()
 
-    assert len(callbacks) == 1
     mock_reset_service_connections.assert_called_once_with(
         str(document_id), str(user_id)
     )
@@ -113,3 +114,191 @@ def test_reset_service_connections_on_commit_without_user(
         reset_service_connections_on_commit(document_id)
 
     mock_reset_service_connections.assert_called_once_with(str(document_id), None)
+
+
+def test_reset_service_connections_on_commit_coalesces_users(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """Asking twice for the same user in one transaction queues one task."""
+    document_id, user_id, other_user_id = uuid4(), uuid4(), uuid4()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        reset_service_connections_on_commit(document_id, user_id)
+        reset_service_connections_on_commit(document_id, user_id)
+        reset_service_connections_on_commit(document_id, other_user_id)
+
+    assert sorted(mock_reset_service_connections.call_args_list, key=str) == sorted(
+        [
+            mock.call(str(document_id), str(user_id)),
+            mock.call(str(document_id), str(other_user_id)),
+        ],
+        key=str,
+    )
+
+
+def test_reset_service_connections_on_commit_everybody_covers_the_users(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """A document asked for everybody is reset once, whatever else was asked."""
+    document_id, user_id = uuid4(), uuid4()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        reset_service_connections_on_commit(document_id, user_id)
+        reset_service_connections_on_commit(document_id)
+        reset_service_connections_on_commit(document_id, uuid4())
+
+    mock_reset_service_connections.assert_called_once_with(str(document_id), None)
+
+
+def test_reset_service_connections_on_commit_keeps_documents_apart(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """Coalescing is per document."""
+    document_id, other_document_id, user_id = uuid4(), uuid4(), uuid4()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        reset_service_connections_on_commit(document_id)
+        reset_service_connections_on_commit(other_document_id, user_id)
+
+    assert sorted(mock_reset_service_connections.call_args_list, key=str) == sorted(
+        [
+            mock.call(str(document_id), None),
+            mock.call(str(other_document_id), str(user_id)),
+        ],
+        key=str,
+    )
+
+
+def test_reset_service_connections_on_commit_new_batch_after_a_commit(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """Once flushed, the next ask starts a new batch of its own."""
+    document_id, user_id = uuid4(), uuid4()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        reset_service_connections_on_commit(document_id, user_id)
+    with django_capture_on_commit_callbacks(execute=True):
+        reset_service_connections_on_commit(document_id, user_id)
+
+    assert mock_reset_service_connections.call_args_list == [
+        mock.call(str(document_id), str(user_id)),
+        mock.call(str(document_id), str(user_id)),
+    ]
+
+
+def test_reset_service_connections_on_commit_drops_what_was_rolled_back(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """The resets of a rolled back transaction are neither sent nor kept."""
+    document_id, other_document_id = uuid4(), uuid4()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        with pytest.raises(RuntimeError), transaction.atomic():
+            reset_service_connections_on_commit(document_id)
+            raise RuntimeError("rolled back")
+        reset_service_connections_on_commit(other_document_id)
+
+    mock_reset_service_connections.assert_called_once_with(str(other_document_id), None)
+
+
+def test_reset_service_connections_on_commit_keeps_a_rolled_back_savepoint(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """
+    Known limit: an ask made in a rolled back savepoint of a transaction that
+    goes on is sent with it, a superfluous reset. Nothing asked is ever lost.
+    """
+    document_id, other_document_id = uuid4(), uuid4()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        reset_service_connections_on_commit(document_id)
+        with pytest.raises(RuntimeError), transaction.atomic():
+            reset_service_connections_on_commit(other_document_id)
+            raise RuntimeError("rolled back")
+
+    assert sorted(mock_reset_service_connections.call_args_list, key=str) == sorted(
+        [mock.call(str(document_id), None), mock.call(str(other_document_id), None)],
+        key=str,
+    )
+
+
+@mock.patch("core.tasks.access.YHubService")
+def test_reset_service_connections_paces_the_walk(mock_service, settings):
+    """
+    A run resets a batch of documents and queues the rest, after a delay and
+    from the last path it reached.
+    """
+    settings.YHUB_RESET_CONNECTIONS_BATCH_SIZE = 2
+    settings.YHUB_RESET_CONNECTIONS_DELAY = 3.5
+    document = factories.DocumentFactory()
+    children = factories.DocumentFactory.create_batch(2, parent=document)
+    grand_child = factories.DocumentFactory(parent=children[1])
+
+    with mock.patch.object(
+        reset_service_connections_in_cascade, "apply_async"
+    ) as mock_apply:
+        reset_service_connections_in_cascade(str(document.id), "user-id")
+
+    assert mock_service.return_value.reset_connections.call_args_list == [
+        mock.call(document, "user-id"),
+        mock.call(children[0], "user-id"),
+    ]
+    mock_apply.assert_called_once_with(
+        args=[str(document.id), "user-id", children[0].path], countdown=3.5
+    )
+
+    # the next run carries on from there
+    mock_service.reset_mock()
+    with mock.patch.object(
+        reset_service_connections_in_cascade, "apply_async"
+    ) as mock_apply:
+        reset_service_connections_in_cascade(
+            str(document.id), "user-id", children[0].path
+        )
+
+    assert mock_service.return_value.reset_connections.call_args_list == [
+        mock.call(children[1], "user-id"),
+        mock.call(grand_child, "user-id"),
+    ]
+    mock_apply.assert_not_called()
+
+
+@mock.patch("core.tasks.access.YHubService")
+def test_reset_service_connections_whole_batch_queues_nothing(mock_service, settings):
+    """A subtree fitting in one batch is done in one run."""
+    settings.YHUB_RESET_CONNECTIONS_BATCH_SIZE = 2
+    document = factories.DocumentFactory()
+    child = factories.DocumentFactory(parent=document)
+
+    with mock.patch.object(
+        reset_service_connections_in_cascade, "apply_async"
+    ) as mock_apply:
+        reset_service_connections_in_cascade(str(document.id))
+
+    assert mock_service.return_value.reset_connections.call_args_list == [
+        mock.call(document, None),
+        mock.call(child, None),
+    ]
+    mock_apply.assert_not_called()
+
+
+@mock.patch("core.tasks.access.YHubService")
+def test_reset_service_connections_paced_walk_covers_the_subtree(
+    mock_service, settings
+):
+    """Run to the end (eagerly here), the walk resets every document once, in order."""
+    settings.YHUB_RESET_CONNECTIONS_BATCH_SIZE = 2
+    document = factories.DocumentFactory()
+    children = factories.DocumentFactory.create_batch(3, parent=document)
+    grand_child = factories.DocumentFactory(parent=children[0])
+    factories.DocumentFactory()  # a document of another tree
+
+    reset_service_connections_in_cascade(str(document.id))
+
+    assert mock_service.return_value.reset_connections.call_args_list == [
+        mock.call(document, None),
+        mock.call(grand_child.get_parent(), None),
+        mock.call(grand_child, None),
+        mock.call(children[1], None),
+        mock.call(children[2], None),
+    ]

src/backend/impress/settings.py

@@ -597,6 +597,20 @@ class Base(Configuration):
         environ_name="YHUB_MIGRATION_TIMEOUT",
         environ_prefix=None,
     )
+    # A change of accesses has the connections of a whole subtree re-checked,
+    # one call to the collaboration server per document. The walk resets this
+    # many documents per run and queues the rest after this many seconds, so
+    # that a large subtree is spread over time instead of fired at once.
+    YHUB_RESET_CONNECTIONS_BATCH_SIZE = values.PositiveIntegerValue(
+        default=50,
+        environ_name="YHUB_RESET_CONNECTIONS_BATCH_SIZE",
+        environ_prefix=None,
+    )
+    YHUB_RESET_CONNECTIONS_DELAY = values.FloatValue(
+        default=1.0,
+        environ_name="YHUB_RESET_CONNECTIONS_DELAY",
+        environ_prefix=None,
+    )
 
     # JWT
     # RSA private key (PEM) used to sign the tokens issued by
@@ -1474,6 +1488,10 @@ class Base(Configuration):
                 "Both OPENAI_SDK and MISTRAL_SDK parameters can not be set simultaneously."
             )
 
+        # a batch of nothing would walk no subtree and never end
+        if cls.YHUB_RESET_CONNECTIONS_BATCH_SIZE < 1:
+            raise ValueError("YHUB_RESET_CONNECTIONS_BATCH_SIZE must be at least 1.")
+
         if cls.POSTHOG_KEY is not None:
             posthog.api_key = cls.POSTHOG_KEY
             posthog.host = cls.POSTHOG_HOST