From de6eb6f46f747a0caef82af0b45d5c0defc0c0e9 Mon Sep 17 00:00:00 2001 From: Manuel Raynaud Date: Thu, 24 Sep 2026 16:05:00 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9A=A1=EF=B8=8F(backend)=20coalesce=20and=20?= =?UTF-8?q?pace=20the=20resets=20of=20collaboration=20connections?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 1 + documentation/env.md | 2 + src/backend/core/tasks/access.py | 98 ++++++++- .../tests/commands/test_clean_document.py | 18 +- src/backend/core/tests/conftest.py | 26 +++ .../documents/test_api_document_accesses.py | 7 +- .../test_api_document_accesses_create.py | 5 +- .../test_api_documents_ask_for_access.py | 5 +- .../test_api_documents_link_configuration.py | 7 +- .../documents/test_api_documents_move.py | 30 +-- ...ternal_api_documents_link_configuration.py | 4 +- .../tests/test_models_user_reconciliation.py | 5 +- src/backend/core/tests/test_models_users.py | 6 +- src/backend/core/tests/test_settings.py | 14 ++ src/backend/core/tests/test_signals.py | 47 ++--- src/backend/core/tests/test_tasks_access.py | 193 +++++++++++++++++- src/backend/impress/settings.py | 18 ++ 17 files changed, 398 insertions(+), 88 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 406780d7c..b115ee5c1 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/documentation/env.md b/documentation/env.md index c66e84a7b..961d55415 100644 --- a/documentation/env.md +++ b/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 | | diff --git a/src/backend/core/tasks/access.py b/src/backend/core/tasks/access.py index 50318f518..64af031f2 100644 --- a/src/backend/core/tasks/access.py +++ b/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) diff --git a/src/backend/core/tests/commands/test_clean_document.py b/src/backend/core/tests/commands/test_clean_document.py index af5175bd7..1bde7fc0a 100644 --- a/src/backend/core/tests/commands/test_clean_document.py +++ b/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) diff --git a/src/backend/core/tests/conftest.py b/src/backend/core/tests/conftest.py index 5764ec8b3..feee0abe2 100644 --- a/src/backend/core/tests/conftest.py +++ b/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.""" diff --git a/src/backend/core/tests/documents/test_api_document_accesses.py b/src/backend/core/tests/documents/test_api_document_accesses.py index 5ad01f705..72c763f2c 100644 --- a/src/backend/core/tests/documents/test_api_document_accesses.py +++ b/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 diff --git a/src/backend/core/tests/documents/test_api_document_accesses_create.py b/src/backend/core/tests/documents/test_api_document_accesses_create.py index 95612b9f3..faea01ca9 100644 --- a/src/backend/core/tests/documents/test_api_document_accesses_create.py +++ b/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" ) diff --git a/src/backend/core/tests/documents/test_api_documents_ask_for_access.py b/src/backend/core/tests/documents/test_api_documents_ask_for_access.py index 8e3c39cac..fb0daf948 100644 --- a/src/backend/core/tests/documents/test_api_documents_ask_for_access.py +++ b/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/" ) diff --git a/src/backend/core/tests/documents/test_api_documents_link_configuration.py b/src/backend/core/tests/documents/test_api_documents_link_configuration.py index 2638ade41..7a516afea 100644 --- a/src/backend/core/tests/documents/test_api_documents_link_configuration.py +++ b/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) diff --git a/src/backend/core/tests/documents/test_api_documents_move.py b/src/backend/core/tests/documents/test_api_documents_move.py index 949b2bc8b..6654245a1 100644 --- a/src/backend/core/tests/documents/test_api_documents_move.py +++ b/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={ diff --git a/src/backend/core/tests/external_api/test_external_api_documents_link_configuration.py b/src/backend/core/tests/external_api/test_external_api_documents_link_configuration.py index 27dcca25d..a71f73929 100644 --- a/src/backend/core/tests/external_api/test_external_api_documents_link_configuration.py +++ b/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, diff --git a/src/backend/core/tests/test_models_user_reconciliation.py b/src/backend/core/tests/test_models_user_reconciliation.py index 4b42161a6..5182144bb 100644 --- a/src/backend/core/tests/test_models_user_reconciliation.py +++ b/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) ) diff --git a/src/backend/core/tests/test_models_users.py b/src/backend/core/tests/test_models_users.py index 29f2f70d7..dd7a852fe 100644 --- a/src/backend/core/tests/test_models_users.py +++ b/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() diff --git a/src/backend/core/tests/test_settings.py b/src/backend/core/tests/test_settings.py index 37ac9f905..bb4df3799 100644 --- a/src/backend/core/tests/test_settings.py +++ b/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 diff --git a/src/backend/core/tests/test_signals.py b/src/backend/core/tests/test_signals.py index 0e1f1d822..dedeed4be 100644 --- a/src/backend/core/tests/test_signals.py +++ b/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() diff --git a/src/backend/core/tests/test_tasks_access.py b/src/backend/core/tests/test_tasks_access.py index dcf088f96..80dd98f26 100644 --- a/src/backend/core/tests/test_tasks_access.py +++ b/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), + ] diff --git a/src/backend/impress/settings.py b/src/backend/impress/settings.py index 7b2c9cf6a..7add35573 100755 --- a/src/backend/impress/settings.py +++ b/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