diff --git a/CHANGELOG.md b/CHANGELOG.md
index cb5f0a818..406780d7c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -82,6 +82,7 @@ and this project adheres to
 - 🐛(frontend) open search results in a new tab with ctrl/cmd+click #2719
 - 🐛(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
 - 🐛(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/src/backend/core/api/viewsets.py b/src/backend/core/api/viewsets.py
index ca8456ac8..2e37e96c3 100644
--- a/src/backend/core/api/viewsets.py
+++ b/src/backend/core/api/viewsets.py
@@ -69,7 +69,7 @@ from core.services.search_indexers import (
     get_visited_document_ids_of,
 )
 from core.services.yhub_services import YHubError, YHubService
-from core.tasks.access import reset_service_connections_in_cascade
+from core.tasks.access import reset_service_connections_on_commit
 from core.tasks.documents import sync_service_deletions_in_cascade
 from core.tasks.mail import send_ask_for_access_mail
 from core.tasks.search import trigger_batch_document_indexer
@@ -1089,6 +1089,11 @@ class DocumentViewSet(
         # Invalidate the nb_accesses cache, the value has probably changed after the move.
         document.invalidate_nb_accesses_cache()
 
+        # The document and its descendants now inherit the accesses and the link
+        # definition of other ancestors, whether or not a direct access was
+        # touched: every connection of the subtree is re-checked.
+        reset_service_connections_on_commit(document.id)
+
         posthog_capture(
             PosthogEventName.DOC_MOVED,
             user,
@@ -1823,11 +1828,10 @@ class DocumentViewSet(
         )
         serializer.is_valid(raise_exception=True)
 
+        # saving a changed link definition is what tells the collaboration
+        # server to re-check the connections of the document and its descendants
         serializer.save()
 
-        # Notify collaboration server about the link updated
-        reset_service_connections_in_cascade.delay(str(document.id))
-
         return drf.response.Response(serializer.data, status=drf.status.HTTP_200_OK)
 
     @drf.decorators.action(detail=True, methods=["post", "delete"], url_path="favorite")
@@ -2701,28 +2705,17 @@ class DocumentAccessViewSet(
                 or settings.LANGUAGE_CODE,
             )
 
-    def perform_update(self, serializer):
-        """Update an access to the document and notify the collaboration server."""
-        access = serializer.save()
-
-        access_user_id = None
-        if access.user:
-            access_user_id = str(access.user.id)
-
-        # Notify collaboration server about the access change
-        reset_service_connections_in_cascade.delay(
-            str(access.document.id), access_user_id
-        )
-
     def perform_destroy(self, instance):
-        """Delete an access to the document and notify the collaboration server."""
+        """
+        Delete an access to the document.
+
+        Saving or deleting an access is what notifies the collaboration server,
+        through the signals of the model: nothing to do here beyond deleting.
+        """
         # Snapshot the identifiers before deletion as Django resets the primary key
         # on the instance once it is deleted.
         access_id = str(instance.id)
         document_id = str(instance.document_id)
-        # an access is granted either to a user or to a team, only a user has
-        # connections of their own to reset
-        user_id = str(instance.user.id) if instance.user else None
 
         instance.delete()
 
@@ -2732,9 +2725,6 @@ class DocumentAccessViewSet(
             {"access_id": access_id, "document_id": document_id},
         )
 
-        # Notify collaboration server about the access removed
-        reset_service_connections_in_cascade.delay(document_id, user_id)
-
     @drf.decorators.action(
         detail=False,
         methods=["get"],
diff --git a/src/backend/core/models.py b/src/backend/core/models.py
index 0c23abab4..3c5f88547 100644
--- a/src/backend/core/models.py
+++ b/src/backend/core/models.py
@@ -4,9 +4,11 @@ Declare and configure the models for the impress core application
 
 # pylint: disable=too-many-lines
 
+import operator
 import smtplib
 import uuid
 from datetime import timedelta
+from functools import partial, reduce
 from logging import getLogger
 
 from django.conf import settings
@@ -256,15 +258,42 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
         )
 
     def _delete_documents_single_owner(self):
-        """Delete the documents where the user is the single owner."""
-        Document.objects.filter(
+        """
+        Delete the documents where the user is the single owner.
+
+        Deleted for good, so the collaboration server, which holds their
+        content and serves them to whoever is editing them, is told by id
+        once the deletion is committed.
+        """
+        from core.tasks.documents import (  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
+            delete_service_documents,
+        )
+
+        documents = Document.objects.filter(
             accesses__user=self, accesses__role=RoleChoices.OWNER
-        ).delete()
+        )
+        # the descendants go with their ancestor, whoever they are shared with
+        paths = list(documents.values_list("path", flat=True))
+        document_ids = []
+        if paths:
+            subtrees = reduce(
+                operator.or_, (models.Q(path__startswith=path) for path in paths)
+            )
+            document_ids = [
+                str(document_id)
+                for document_id in Document.objects.filter(subtrees).values_list(
+                    "id", flat=True
+                )
+            ]
+        documents.delete()
         logger.info(
             "user_delete: documents where the user %s is the sole owner deleted",
             self.id,
         )
 
+        if document_ids:
+            transaction.on_commit(partial(delete_service_documents.delay, document_ids))
+
     def _clear_user_created_documents(self):
         """Set creator to Null for documents where the user is the creator."""
         Document.objects.filter(creator=self).update(creator=None)
@@ -386,6 +415,16 @@ class User(AbstractBaseUser, BaseModel, auth_models.PermissionsMixin):
 
         # Set creator of documents if not yet set (e.g. documents created via server-to-server API)
         document_ids = [invitation.document_id for invitation in valid_invitations]
+
+        # created in bulk, so without the signal that reports a new access to the
+        # collaboration server
+        from core.tasks.access import (  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
+            reset_service_connections_on_commit,
+        )
+
+        for document_id in document_ids:
+            reset_service_connections_on_commit(document_id, self.id)
+
         Document.objects.filter(id__in=document_ids, creator__isnull=True).update(
             creator=self
         )
@@ -550,6 +589,17 @@ class UserReconciliation(BaseModel):
             ids_to_delete = [entry.id for entry in removed_accesses]
             DocumentAccess.objects.filter(id__in=ids_to_delete).delete()
 
+        # Updated in bulk, so without the signal that reports a changed access to
+        # the collaboration server. Both users are concerned, the one gaining
+        # the accesses and the one being deactivated: every connection of the
+        # document is re-checked.
+        from core.tasks.access import (  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
+            reset_service_connections_on_commit,
+        )
+
+        for document_id in {access.document_id for access in updated_accesses}:
+            reset_service_connections_on_commit(document_id)
+
         DocumentFavorite.objects.bulk_update(update_favorites, ["user"])
         if removed_favorites:
             ids_to_delete = [entry.id for entry in removed_favorites]
@@ -1106,6 +1156,56 @@ class Document(MP_Node, BaseModel):
         """Returns the number of accesses related to the document or one of its ancestors."""
         return self.get_nb_accesses()[1]
 
+    # the link definition as it was loaded from, or saved to, the database
+    _saved_link_definition = None
+
+    @classmethod
+    def from_db(cls, db, field_names, values):
+        """Load a document, remembering its link definition to spot its changes."""
+        instance = super().from_db(db, field_names, values)
+        instance.remember_link_definition()  # pylint: disable=no-member
+        return instance
+
+    def refresh_from_db(self, *args, **kwargs):
+        """
+        Reload a document, taking a new snapshot of the link fields reloaded.
+
+        Django reloads a single field this way when a deferred one is read:
+        the other one keeps its snapshot, its value in memory may be unsaved.
+        """
+        super().refresh_from_db(*args, **kwargs)
+        # `fields` is the second positional argument of Django's signature
+        fields = kwargs.get("fields", args[1] if len(args) > 1 else None)
+        if fields is None or {"link_reach", "link_role"} & set(fields):
+            self.remember_link_definition(fields)
+
+    def remember_link_definition(self, fields=None):
+        """
+        Snapshot the link definition as it is in the database.
+
+        Read from the instance's own state so that a deferred field is not
+        loaded for it; a field not loaded is remembered as unknown. Naming
+        `fields` restricts the snapshot to the link fields among them, the
+        others keep what was remembered of them.
+        """
+        reach, role = self._saved_link_definition or (None, None)
+        if fields is None or "link_reach" in fields:
+            reach = self.__dict__.get("link_reach")
+        if fields is None or "link_role" in fields:
+            role = self.__dict__.get("link_role")
+        self._saved_link_definition = (reach, role)
+
+    def link_definition_changed(self):
+        """
+        Tell whether `link_reach` or `link_role` differ from the last snapshot.
+
+        A document that was never loaded from the database, nor saved, is
+        reported as changed: nothing is known of what its link definition was.
+        """
+        if self._saved_link_definition is None:
+            return True
+        return self._saved_link_definition != (self.link_reach, self.link_role)
+
     def invalidate_nb_accesses_cache(self):
         """
         Invalidate the cache for number of accesses, including on affected descendants.
diff --git a/src/backend/core/services/yhub_services.py b/src/backend/core/services/yhub_services.py
index 944cd59e0..5d7b5727b 100644
--- a/src/backend/core/services/yhub_services.py
+++ b/src/backend/core/services/yhub_services.py
@@ -169,10 +169,16 @@ class YHubService:
         return JWKSClient(self.jwks_url)
 
     def build_url(self, endpoint, document):
-        """Build the url of a document scoped endpoint of the yhub API."""
+        """
+        Build the url of a document scoped endpoint of the yhub API.
+
+        `document` is a document, or the id of one: what is deleted for good
+        from the database can only be named by its id.
+        """
+        document_id = getattr(document, "id", document)
         return (
             f"{self.base_url}/{self.api_prefix}/{endpoint}/{self.api_version}"
-            f"/{self.org}/{document.id}"
+            f"/{self.org}/{document_id}"
         )
 
     @staticmethod
diff --git a/src/backend/core/signals.py b/src/backend/core/signals.py
index 5acf5f050..d19c56aba 100644
--- a/src/backend/core/signals.py
+++ b/src/backend/core/signals.py
@@ -10,27 +10,43 @@ from django.db.models import signals
 from django.dispatch import receiver
 
 from core import models
+from core.tasks.access import reset_service_connections_on_commit
 from core.tasks.search import trigger_batch_document_indexer
 from core.utils.users import get_users_sharing_documents_with_cache_key
 
 
 @receiver(signals.post_save, sender=models.Document)
-def document_post_save(sender, instance, **kwargs):  # pylint: disable=unused-argument
+def document_post_save(sender, instance, created, **kwargs):  # pylint: disable=unused-argument
     """
     Asynchronous call to the document indexer at the end of the transaction.
     Note : Within the transaction we can have an empty content and a serialization
     error.
+
+    A change of the link definition changes who may open the document and its
+    descendants, from wherever it was saved: the collaboration server is told
+    to re-check their connections.
     """
     transaction.on_commit(
         partial(trigger_batch_document_indexer, instance.pk, instance.updated_at)
     )
 
+    if not created and instance.link_definition_changed():
+        reset_service_connections_on_commit(instance.pk)
+    # once committed: taken now, a save rolled back with its savepoint would
+    # leave the snapshot of a value the database never got, and saving that
+    # value again would report no change
+    transaction.on_commit(instance.remember_link_definition)
+
 
 @receiver(signals.post_save, sender=models.DocumentAccess)
 def document_access_post_save(sender, instance, created, **kwargs):  # pylint: disable=unused-argument
     """
     Asynchronous call to the document indexer at the end of the transaction.
     Clear cache for the affected user.
+
+    Every change of an access, wherever it comes from (API, admin, an accepted
+    access request...), is reported to the collaboration server so that the
+    connections of the user, or of everybody for a team, are re-checked.
     """
     if not created:
         document = instance.document
@@ -42,11 +58,17 @@ def document_access_post_save(sender, instance, created, **kwargs):  # pylint: d
     cache_key = get_users_sharing_documents_with_cache_key(instance.user_id)
     cache.delete(cache_key)
 
+    reset_service_connections_on_commit(instance.document_id, instance.user_id)
+
 
 @receiver(signals.post_delete, sender=models.DocumentAccess)
 def document_access_post_delete(sender, instance, **kwargs):  # pylint: disable=unused-argument
     """
-    Clear cache for the affected user when document access is deleted.
+    Clear cache for the affected user when document access is deleted, and
+    have the collaboration server re-check their connections: they may have
+    lost their access, or an inherited one.
     """
     cache_key = get_users_sharing_documents_with_cache_key(instance.user_id)
     cache.delete(cache_key)
+
+    reset_service_connections_on_commit(instance.document_id, instance.user_id)
diff --git a/src/backend/core/tasks/access.py b/src/backend/core/tasks/access.py
index 821fdb809..50318f518 100644
--- a/src/backend/core/tasks/access.py
+++ b/src/backend/core/tasks/access.py
@@ -1,8 +1,11 @@
 """Tasks dedicated to document's accesses."""
 
+from functools import partial
 from logging import getLogger
 
-from core import models
+from django.apps import apps
+from django.db import transaction
+
 from core.services.yhub_services import YHubError, YHubService
 
 from impress.celery_app import app
@@ -24,13 +27,19 @@ def reset_service_connections_in_cascade(document_id, user_id=None):
     A document failing is logged and does not stop the ones after it, its
     clients keep the rights they connected with until they reconnect.
     """
+    # resolved at run time: the models queue these tasks, importing them here
+    # would import the models back
+    document_model = apps.get_model("core", "Document")
     try:
-        document = models.Document.objects.get(pk=document_id)
-    except models.Document.DoesNotExist:
-        logger.error("Document %s does not exists anymore", document_id)
+        document = document_model.objects.get(pk=document_id)
+    except document_model.DoesNotExist:
+        # deleted for good in the meantime, its accesses with it: there is no
+        # connection left to re-check, the deletion is reported to the
+        # collaboration server by the code that deleted the document
+        logger.info("Document %s does not exist anymore, nothing to reset", document_id)
         return
 
-    documents = models.Document.objects.filter(
+    documents = document_model.objects.filter(
         path__startswith=document.path, depth__gte=document.depth
     ).order_by("path")
 
@@ -40,3 +49,27 @@ def reset_service_connections_in_cascade(document_id, user_id=None):
             service.reset_connections(doc, user_id)
         except YHubError:
             logger.exception("impossible to reset connections for document %s", doc.id)
+
+
+def reset_service_connections_on_commit(document_id, user_id=None):
+    """
+    Queue the reset of the connections of a document, and of its descendants,
+    for when the current transaction is committed.
+
+    The task reads the accesses back from the database to know what the
+    collaboration server should re-check them against: queued before the
+    commit, it could run against the accesses as they were. Outside of a
+    transaction the task is queued right away.
+
+    Naming a user restricts the re-check to their own connections, which is
+    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.
+    """
+    transaction.on_commit(
+        partial(
+            reset_service_connections_in_cascade.delay,
+            str(document_id),
+            str(user_id) if user_id else None,
+        )
+    )
diff --git a/src/backend/core/tasks/documents.py b/src/backend/core/tasks/documents.py
index 0b4297671..451bcc6e8 100644
--- a/src/backend/core/tasks/documents.py
+++ b/src/backend/core/tasks/documents.py
@@ -2,7 +2,10 @@
 
 from logging import getLogger
 
-from core import models
+from django.apps import apps
+
+from celery.exceptions import MaxRetriesExceededError
+
 from core.services.yhub_services import YHubError, YHubService
 
 from impress.celery_app import app
@@ -31,13 +34,16 @@ def sync_service_deletions_in_cascade(document_id):
     A document failing is logged and does not stop the ones after it; the
     collaboration server keeps serving it until something says so again.
     """
+    # resolved at run time: the models queue these tasks, importing them here
+    # would import the models back
+    document_model = apps.get_model("core", "Document")
     try:
-        document = models.Document.objects.get(pk=document_id)
-    except models.Document.DoesNotExist:
+        document = document_model.objects.get(pk=document_id)
+    except document_model.DoesNotExist:
         logger.error("Document %s does not exists anymore", document_id)
         return
 
-    documents = models.Document.objects.filter(
+    documents = document_model.objects.filter(
         path__startswith=document.path, depth__gte=document.depth
     ).order_by("path")
 
@@ -57,3 +63,57 @@ def sync_service_deletions_in_cascade(document_id):
                 "delete" if deleted else "restore",
                 doc.id,
             )
+
+
+# how long a retry of `delete_service_documents` waits, doubled at each attempt
+DELETE_RETRY_COUNTDOWN = 30
+DELETE_RETRY_MAX_COUNTDOWN = 600
+
+
+@app.task(bind=True, max_retries=5)
+def delete_service_documents(self, document_ids):
+    """
+    Report to the collaboration server the deletion of documents that are gone
+    for good from the database.
+
+    `sync_service_deletions_in_cascade` reads the documents back to know what
+    to report, which a hard deletion leaves nothing of: the ids are all that
+    is left, and the walk down the tree is up to the caller. Each document is
+    deleted on its own, one failing does not stop the others.
+
+    Nothing else knows of these deletions anymore, so a failure is retried,
+    for the documents that failed only, with a countdown doubling at each
+    attempt. Once the retries are spent, the ids are logged as an error: it
+    is all that is left to delete them by hand.
+    """
+    service = YHubService()
+    failed = []
+    for document_id in document_ids:
+        try:
+            service.delete_ydoc(document_id)
+        except YHubError:
+            logger.warning(
+                "impossible to delete document %s on the collaboration server, "
+                "will retry",
+                document_id,
+            )
+            failed.append(document_id)
+
+    if not failed:
+        return
+
+    try:
+        self.retry(
+            args=[failed],
+            countdown=min(
+                DELETE_RETRY_MAX_COUNTDOWN,
+                DELETE_RETRY_COUNTDOWN * 2**self.request.retries,
+            ),
+        )
+    except MaxRetriesExceededError:
+        logger.error(
+            "giving up on deleting documents %s on the collaboration server "
+            "after %d attempts",
+            ", ".join(failed),
+            self.request.retries + 1,
+        )
diff --git a/src/backend/core/tests/commands/test_clean_document.py b/src/backend/core/tests/commands/test_clean_document.py
index bcc554b4d..af5175bd7 100644
--- a/src/backend/core/tests/commands/test_clean_document.py
+++ b/src/backend/core/tests/commands/test_clean_document.py
@@ -438,3 +438,35 @@ def test_clean_document_reports_the_documents_it_could_not_erase(
     assert "Erased collaboration content for 1 document(s)." in captured.out
     assert str(root.id) in captured.err
     assert str(child.id) not in captured.err
+
+
+def test_clean_document_resets_connections(
+    settings, mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """
+    The link definition of the root changes and its accesses but the owners'
+    are deleted: the collaboration server should re-check its connections.
+    """
+    settings.DEBUG = True
+    root = factories.DocumentFactory(
+        link_reach=LinkReachChoices.PUBLIC, link_role=LinkRoleChoices.EDITOR
+    )
+    owner = factories.UserDocumentAccessFactory(
+        document=root, role=choices.RoleChoices.OWNER
+    )
+    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),
+    ):
+        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
diff --git a/src/backend/core/tests/conftest.py b/src/backend/core/tests/conftest.py
index 45e0a4dbb..5764ec8b3 100644
--- a/src/backend/core/tests/conftest.py
+++ b/src/backend/core/tests/conftest.py
@@ -42,6 +42,31 @@ def restore_urlconf():
     restore_urls()
 
 
+@pytest.fixture(autouse=True, name="mock_reset_service_connections")
+def mock_reset_service_connections_fixture():
+    """
+    Take the resets of connections queued for the collaboration server.
+
+    Every change of an access queues one, at the commit of the transaction: in
+    a transactional test the Celery task would then run inline and reach for
+    the collaboration server. What was queued is checked on this mock, once
+    the callbacks on commit have run (`django_capture_on_commit_callbacks`).
+    """
+    with mock.patch(
+        "core.tasks.access.reset_service_connections_in_cascade.delay"
+    ) as mock_delay:
+        yield mock_delay
+
+
+@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."""
+    with mock.patch(
+        "core.tasks.documents.delete_service_documents.delay"
+    ) as mock_delay:
+        yield mock_delay
+
+
 @pytest.fixture
 def mock_user_teams():
     """Mock for the "teams" property on the User model."""
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 2f4393881..5ad01f705 100644
--- a/src/backend/core/tests/documents/test_api_document_accesses.py
+++ b/src/backend/core/tests/documents/test_api_document_accesses.py
@@ -20,20 +20,23 @@ pytestmark = pytest.mark.django_db
 
 
 @pytest.fixture(name="mock_reset_connections")
-def mock_reset_connections_fixture():
+def mock_reset_connections_fixture(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
     """
-    Provide a context manager that patches the ``reset_service_connections_in_cascade``
-    Celery task and asserts its ``delay`` method is called exactly once for the given
-    document and user when leaving the context.
+    Provide a context manager that runs the callbacks 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):
-        with mock.patch(
-            "core.api.viewsets.reset_service_connections_in_cascade.delay"
-        ) as mock_delay:
-            yield mock_delay
-            mock_delay.assert_called_once_with(str(document_id), user_id)
+        mock_reset_service_connections.reset_mock()
+        with django_capture_on_commit_callbacks(execute=True):
+            yield mock_reset_service_connections
+        mock_reset_service_connections.assert_called_once_with(
+            str(document_id), user_id
+        )
 
     return _mock_reset_connections
 
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 6ed96129c..95612b9f3 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
@@ -586,3 +586,40 @@ def test_api_document_accesses_create_email_in_receivers_language(via, mock_user
                 in email_subject.lower()
             )
         assert "docs/" + str(document.id) + "/" in email_content.lower()
+
+
+@pytest.mark.parametrize("via", VIA)
+def test_api_document_accesses_create_resets_connections(
+    via,
+    mock_user_teams,
+    mock_reset_service_connections,
+    django_capture_on_commit_callbacks,
+):
+    """
+    Creating an access should have the collaboration server re-check the
+    connections: the user's own for a user, everybody's for a team.
+    """
+    user = factories.UserFactory()
+    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()
+        data = {"user_id": str(other_user.id), "role": "editor"}
+        expected_user_id = str(other_user.id)
+    else:
+        mock_user_teams.return_value = ["lasuite"]
+        data = {"team": "lasuite", "role": "editor"}
+        expected_user_id = None
+
+    with django_capture_on_commit_callbacks(execute=True):
+        response = client.post(
+            f"/api/v1.0/documents/{document.id!s}/accesses/", data, format="json"
+        )
+
+    assert response.status_code == 201
+    mock_reset_service_connections.assert_called_once_with(
+        str(document.id), expected_user_id
+    )
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 9c260f402..8e3c39cac 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
@@ -912,3 +912,28 @@ def test_api_document_ask_for_access_throttling(settings):
     settings.REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"]["document_ask_for_access"] = (
         current_rate
     )
+
+
+def test_api_documents_ask_for_access_accept_resets_connections(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """Accepting a request creates or changes an access: the user is re-checked."""
+    user = UserFactory()
+    document = DocumentFactory(users=[(user, RoleChoices.OWNER)])
+    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):
+        response = client.post(
+            f"/api/v1.0/documents/{document.id}/ask-for-access/{document_ask_for_access.id}/accept/"
+        )
+
+    assert response.status_code == 204
+    mock_reset_service_connections.assert_called_once_with(
+        str(document.id), str(document_ask_for_access.user_id)
+    )
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 9a3a3c8fd..2638ade41 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
@@ -1,7 +1,6 @@
 """Tests for link configuration of documents on API endpoint"""
 
 from contextlib import contextmanager
-from unittest import mock
 
 import pytest
 from rest_framework.test import APIClient
@@ -14,20 +13,21 @@ pytestmark = pytest.mark.django_db
 
 
 @pytest.fixture(name="mock_reset_connections")
-def mock_reset_connections_fixture():
+def mock_reset_connections_fixture(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
     """
-    Provide a context manager that patches the ``reset_service_connections_in_cascade``
-    Celery task and asserts its ``delay`` method is called exactly once for the given
-    document when leaving the context.
+    Provide a context manager that runs the callbacks 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):
-        with mock.patch(
-            "core.api.viewsets.reset_service_connections_in_cascade.delay"
-        ) as mock_delay:
-            yield mock_delay
-            mock_delay.assert_called_once_with(str(document_id))
+        mock_reset_service_connections.reset_mock()
+        with django_capture_on_commit_callbacks(execute=True):
+            yield mock_reset_service_connections
+        mock_reset_service_connections.assert_called_once_with(str(document_id), None)
 
     return _mock_reset_connections
 
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 aef95d78a..949b2bc8b 100644
--- a/src/backend/core/tests/documents/test_api_documents_move.py
+++ b/src/backend/core/tests/documents/test_api_documents_move.py
@@ -836,3 +836,93 @@ def test_api_documents_move_scope_change_deletion_is_atomic(monkeypatch):
     document.refresh_from_db()
     assert document.accesses.count() == 2
     assert document.invitations.count() == 1
+
+
+def test_api_documents_move_resets_connections_of_the_moved_document(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """
+    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.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    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):
+        response = client.post(
+            f"/api/v1.0/documents/{document.id!s}/move/",
+            data={
+                "target_document_id": str(target.id),
+                "position": enums.MoveNodePositionChoices.LAST_CHILD,
+            },
+        )
+
+    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,
+    )
+
+
+def test_api_documents_move_resets_connections_when_the_scope_is_kept(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """
+    Moving within the same tree touches no direct access, the ancestors change
+    all the same: the subtree is re-checked.
+    """
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    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):
+        response = client.post(
+            f"/api/v1.0/documents/{document.id!s}/move/",
+            data={
+                "target_document_id": str(sibling.id),
+                "position": enums.MoveNodePositionChoices.LAST_CHILD,
+            },
+        )
+
+    assert response.status_code == 200
+    mock_reset_service_connections.assert_called_once_with(str(document.id), None)
+
+
+def test_api_documents_move_resets_nothing_when_refused(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """A refused move changes nothing, and reports nothing."""
+    user = factories.UserFactory()
+    client = APIClient()
+    client.force_login(user)
+
+    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):
+        response = client.post(
+            f"/api/v1.0/documents/{document.id!s}/move/",
+            data={
+                "target_document_id": str(child.id),
+                "position": enums.MoveNodePositionChoices.LAST_CHILD,
+            },
+        )
+
+    assert response.status_code == 400
+    mock_reset_service_connections.assert_not_called()
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 7c1e6a308..27dcca25d 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
@@ -6,8 +6,6 @@ because the resource server viewsets inherit from the api viewsets.
 
 """
 
-from unittest.mock import patch
-
 from django.test import override_settings
 
 import pytest
@@ -61,9 +59,12 @@ def test_external_api_documents_link_configuration_not_allowed(
         },
     },
 )
-@patch("core.api.viewsets.reset_service_connections_in_cascade.delay")
 def test_external_api_documents_link_configuration_can_be_allowed(
-    mock_reset, user_token, resource_server_backend, user_specific_sub
+    user_token,
+    resource_server_backend,
+    user_specific_sub,
+    mock_reset_service_connections,
+    django_capture_on_commit_callbacks,
 ):
     """
     Connected users SHOULD be allowed to update the link configuration of a document
@@ -89,11 +90,12 @@ def test_external_api_documents_link_configuration_can_be_allowed(
         "link_role": models.LinkRoleChoices.EDITOR,
     }
 
-    response = client.put(
-        f"/external_api/v1.0/documents/{document.id!s}/link-configuration/",
-        new_data,
-        format="json",
-    )
+    with django_capture_on_commit_callbacks(execute=True):
+        response = client.put(
+            f"/external_api/v1.0/documents/{document.id!s}/link-configuration/",
+            new_data,
+            format="json",
+        )
 
     assert response.status_code == 200
 
@@ -103,4 +105,4 @@ def test_external_api_documents_link_configuration_can_be_allowed(
     assert document.link_role == models.LinkRoleChoices.EDITOR
 
     # the collaboration server should be notified through the Celery task
-    mock_reset.assert_called_once_with(str(document.id))
+    mock_reset_service_connections.assert_called_once_with(str(document.id), None)
diff --git a/src/backend/core/tests/test_models_user_reconciliation.py b/src/backend/core/tests/test_models_user_reconciliation.py
index aefe2a155..4b42161a6 100644
--- a/src/backend/core/tests/test_models_user_reconciliation.py
+++ b/src/backend/core/tests/test_models_user_reconciliation.py
@@ -4,6 +4,7 @@ Unit tests for the UserReconciliationCsvImport model
 
 import uuid
 from pathlib import Path
+from unittest import mock
 
 from django.core import mail
 from django.core.files.base import ContentFile
@@ -664,3 +665,44 @@ def test_process_reconciliation_updates_favorites(
     assert models.DocumentFavorite.objects.filter(
         user=user_1, document=doc_active_only
     ).exists()
+
+
+def test_process_reconciliation_resets_connections(
+    user_reconciliation_users_and_docs,
+    mock_reset_service_connections,
+    django_capture_on_commit_callbacks,
+):
+    """
+    The accesses are updated in bulk, without the signal: every document
+    getting an access moved or raised should have its connections re-checked,
+    both users being concerned. The accesses removed go through the signal.
+    """
+    user_1, user_2, userdocs_u1, userdocs_u2 = user_reconciliation_users_and_docs
+    rec = models.UserReconciliation.objects.create(
+        active_email=user_1.email,
+        inactive_email=user_2.email,
+        active_user=user_1,
+        inactive_user=user_2,
+        active_email_checked=True,
+        inactive_email_checked=True,
+        status="ready",
+    )
+    mock_reset_service_connections.reset_mock()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        process_reconciliation(
+            None, None, models.UserReconciliation.objects.filter(id=rec.id)
+        )
+
+    calls = mock_reset_service_connections.call_args_list
+    # moved to the active user: the documents of the inactive user alone
+    for access in userdocs_u2[6:]:
+        assert mock.call(str(access.document_id), None) in calls
+    # raised on the active user: the documents where the inactive user had more
+    for access in userdocs_u2[0:3] + userdocs_u1[3:6]:
+        assert mock.call(str(access.document_id), None) in calls
+    # the documents where the active user already had as much are untouched,
+    # only the removed access of the inactive user is reported
+    for access in userdocs_u1[0:3] + userdocs_u2[3:6]:
+        assert mock.call(str(access.document_id), None) not in calls
+        assert mock.call(str(access.document_id), str(user_2.id)) in calls
diff --git a/src/backend/core/tests/test_models_users.py b/src/backend/core/tests/test_models_users.py
index 196aa8681..29f2f70d7 100644
--- a/src/backend/core/tests/test_models_users.py
+++ b/src/backend/core/tests/test_models_users.py
@@ -4,6 +4,7 @@ Unit tests for the User model
 
 import uuid
 from concurrent.futures import ThreadPoolExecutor
+from unittest import mock
 from unittest.mock import patch
 
 from django.core.exceptions import ValidationError
@@ -754,3 +755,77 @@ def test_tasks_user_delete_error_during_deletion_should_rollback_deletion(monkey
     assert len(documents_to_delete) == 9
 
     assert models.User.objects.filter(id=user_to_delete.id).exists() is True
+
+
+def test_models_users_convert_valid_invitations_resets_connections(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """
+    The accesses are created in bulk, without the signal: the connections of
+    the new user should be re-checked on each document all the same.
+    """
+    email = "test@example.com"
+    document = factories.DocumentFactory()
+    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):
+        user = factories.UserFactory(email=email)
+
+    assert sorted(mock_reset_service_connections.call_args_list, key=str) == sorted(
+        [
+            mock.call(str(document.id), str(user.id)),
+            mock.call(str(other_document.id), str(user.id)),
+        ],
+        key=str,
+    )
+
+
+@pytest.mark.django_db(transaction=True)
+def test_models_users_delete_reports_to_the_collaboration_server(
+    mock_reset_service_connections, mock_delete_service_documents
+):
+    """
+    Deleting a user removes their accesses, which are re-checked one by one,
+    and deletes the documents they solely own for good, descendants included:
+    the collaboration server is told by id.
+    """
+    user = factories.UserFactory()
+    other_user = factories.UserFactory()
+    owned = factories.DocumentFactory(users=[(user, "owner")])
+    owned_child = factories.DocumentFactory(parent=owned)
+    owned_grand_child = factories.DocumentFactory(
+        parent=owned_child, users=[(other_user, "editor")]
+    )
+    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()
+
+    # the shared documents lose an access, the accesses of the deleted
+    # documents go with them (the task finds no document and does nothing)
+    assert mock.call(str(shared.id), user_id) in (
+        mock_reset_service_connections.call_args_list
+    )
+    assert mock.call(str(member.id), user_id) in (
+        mock_reset_service_connections.call_args_list
+    )
+    mock_delete_service_documents.assert_called_once()
+    assert sorted(mock_delete_service_documents.call_args.args[0]) == sorted(
+        [str(owned.id), str(owned_child.id), str(owned_grand_child.id)]
+    )
+
+
+@pytest.mark.django_db(transaction=True)
+def test_models_users_delete_nothing_owned(mock_delete_service_documents):
+    """A user owning no document has nothing to delete on the collaboration server."""
+    user = factories.UserFactory()
+    factories.DocumentFactory(users=[(user, "editor")])
+
+    user.delete()
+
+    mock_delete_service_documents.assert_not_called()
diff --git a/src/backend/core/tests/test_services_yhub_services.py b/src/backend/core/tests/test_services_yhub_services.py
index 8d780677e..01ae800f0 100644
--- a/src/backend/core/tests/test_services_yhub_services.py
+++ b/src/backend/core/tests/test_services_yhub_services.py
@@ -61,6 +61,13 @@ def test_build_url():
     assert url == f"http://yhub:3002/collaboration/ydoc/v1/docs/{DOCUMENT.id!s}"
 
 
+def test_build_url_with_an_id():
+    """A document deleted for good can only be named by its id."""
+    url = YHubService().build_url("ydoc", str(DOCUMENT.id))
+
+    assert url == f"http://yhub:3002/collaboration/ydoc/v1/docs/{DOCUMENT.id!s}"
+
+
 def test_jwks_url():
     """The keys validating what yhub signs should be read from yhub itself."""
     service = YHubService()
diff --git a/src/backend/core/tests/test_signals.py b/src/backend/core/tests/test_signals.py
new file mode 100644
index 000000000..0e1f1d822
--- /dev/null
+++ b/src/backend/core/tests/test_signals.py
@@ -0,0 +1,257 @@
+"""
+Tests of the signals reporting the changes of accesses to the collaboration
+server, whichever code changed them.
+"""
+
+from unittest import mock
+
+from django.db import transaction
+
+import pytest
+
+from core import factories, models
+
+pytestmark = pytest.mark.django_db
+
+
+def test_signals_document_access_created(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """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):
+        models.DocumentAccess.objects.create(
+            document=document, user=user, role="editor"
+        )
+
+    mock_reset_service_connections.assert_called_once_with(
+        str(document.id), str(user.id)
+    )
+
+
+def test_signals_document_access_created_for_a_team(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """An access granted to a team names nobody: every connection is re-checked."""
+    document = factories.DocumentFactory()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        models.DocumentAccess.objects.create(
+            document=document, team="lasuite", role="reader"
+        )
+
+    mock_reset_service_connections.assert_called_once_with(str(document.id), None)
+
+
+def test_signals_document_access_updated(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """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):
+        access.role = "editor"
+        access.save()
+
+    mock_reset_service_connections.assert_called_once_with(
+        str(access.document_id), str(access.user_id)
+    )
+
+
+def test_signals_document_access_deleted(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """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):
+        access.delete()
+
+    mock_reset_service_connections.assert_called_once_with(
+        str(document_id), str(user_id)
+    )
+
+
+def test_signals_document_accesses_deleted_in_bulk(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """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):
+        models.DocumentAccess.objects.filter(document=document).delete()
+
+    assert sorted(mock_reset_service_connections.call_args_list, key=str) == sorted(
+        [mock.call(str(document.id), str(access.user_id)) for access in accesses],
+        key=str,
+    )
+
+
+def test_signals_document_reset_is_queued_on_commit(mock_reset_service_connections):
+    """Nothing is queued before the transaction is committed."""
+    document = factories.DocumentFactory()
+
+    models.DocumentAccess.objects.create(
+        document=document, user=factories.UserFactory(), role="editor"
+    )
+
+    mock_reset_service_connections.assert_not_called()
+
+
+def test_signals_document_link_definition_changed(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """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):
+        document.link_reach = "public"
+        document.save()
+
+    mock_reset_service_connections.assert_called_once_with(str(document.id), None)
+
+
+def test_signals_document_link_role_changed(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """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):
+        document.link_role = "editor"
+        document.save()
+
+    mock_reset_service_connections.assert_called_once_with(str(document.id), None)
+
+
+def test_signals_document_link_definition_unchanged(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """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):
+        document.title = "renamed"
+        document.save()
+        document.link_reach = "public"  # the value it already has
+        document.save()
+
+    mock_reset_service_connections.assert_not_called()
+
+
+def test_signals_document_link_definition_saved_twice(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """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):
+        document.link_reach = "authenticated"
+        document.save()
+        document.save()
+
+    mock_reset_service_connections.assert_called_once_with(str(document.id), None)
+
+
+def test_signals_document_link_definition_snapshot_follows_the_commit(
+    mock_reset_service_connections, capture_service_resets
+):
+    """Once committed, saving the same value again reports no change."""
+    document = factories.DocumentFactory(link_reach="restricted", link_role="reader")
+    document = models.Document.objects.get(pk=document.pk)
+
+    with capture_service_resets():
+        document.link_reach = "authenticated"
+        document.save()
+    with capture_service_resets():
+        document.save()
+
+    mock_reset_service_connections.assert_not_called()
+
+
+def test_signals_document_link_definition_saved_in_a_rolled_back_savepoint(
+    mock_reset_service_connections, capture_service_resets
+):
+    """
+    A save rolled back with its savepoint leaves no snapshot behind: saving
+    the same value again is a change, and is reported.
+    """
+    document = factories.DocumentFactory(link_reach="restricted", link_role="reader")
+    document = models.Document.objects.get(pk=document.pk)
+
+    with capture_service_resets():
+        with pytest.raises(RuntimeError), transaction.atomic():
+            document.link_role = "editor"
+            document.save()
+            raise RuntimeError("rolled back")
+        document.save()
+
+    mock_reset_service_connections.assert_called_once_with(str(document.id), None)
+
+
+def test_signals_document_link_definition_refreshed(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """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):
+        document.save()
+
+    mock_reset_service_connections.assert_not_called()
+
+
+def test_signals_document_link_definition_partly_refreshed(
+    mock_reset_service_connections, capture_service_resets
+):
+    """
+    Reloading one link field, as reading a deferred one does, must not take
+    the unsaved value of the other one for the saved one.
+    """
+    document = factories.DocumentFactory(link_reach="public", link_role="reader")
+    document = models.Document.objects.get(pk=document.pk)
+    document.link_role = "editor"
+    document.refresh_from_db(fields=["link_reach"])
+
+    with capture_service_resets():
+        document.save()
+
+    mock_reset_service_connections.assert_called_once_with(str(document.id), None)
+
+
+def test_signals_document_link_definition_deferred_field_read(
+    mock_reset_service_connections, capture_service_resets
+):
+    """The same, through Django loading a deferred field on read."""
+    document = factories.DocumentFactory(link_reach="public", link_role="reader")
+    document = models.Document.objects.defer("link_reach").get(pk=document.pk)
+    document.link_role = "editor"
+    assert document.link_reach == "public"  # loads the deferred field
+
+    with capture_service_resets():
+        document.save()
+
+    mock_reset_service_connections.assert_called_once_with(str(document.id), None)
+
+
+def test_signals_document_created(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """A new document has no connection to re-check."""
+    with django_capture_on_commit_callbacks(execute=True):
+        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 05a7add72..dcf088f96 100644
--- a/src/backend/core/tests/test_tasks_access.py
+++ b/src/backend/core/tests/test_tasks_access.py
@@ -4,12 +4,16 @@ core.tasks.access module.
 """
 
 from unittest import mock
+from uuid import uuid4
 
 import pytest
 
 from core import factories
 from core.services.yhub_services import ServiceUnavailableError
-from core.tasks.access import reset_service_connections_in_cascade
+from core.tasks.access import (
+    reset_service_connections_in_cascade,
+    reset_service_connections_on_commit,
+)
 
 pytestmark = pytest.mark.django_db
 
@@ -81,3 +85,31 @@ def test_reset_service_connections_keeps_going_on_failure(mock_service):
         mock.call(document, None),
         mock.call(child, None),
     ]
+
+
+def test_reset_service_connections_on_commit(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """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:
+        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)
+    )
+
+
+def test_reset_service_connections_on_commit_without_user(
+    mock_reset_service_connections, django_capture_on_commit_callbacks
+):
+    """Naming nobody re-checks every connection."""
+    document_id = uuid4()
+
+    with django_capture_on_commit_callbacks(execute=True):
+        reset_service_connections_on_commit(document_id)
+
+    mock_reset_service_connections.assert_called_once_with(str(document_id), None)
diff --git a/src/backend/core/tests/test_tasks_documents.py b/src/backend/core/tests/test_tasks_documents.py
index 47840b212..38f53b29d 100644
--- a/src/backend/core/tests/test_tasks_documents.py
+++ b/src/backend/core/tests/test_tasks_documents.py
@@ -6,10 +6,14 @@ core.tasks.documents module.
 from unittest import mock
 
 import pytest
+from celery.exceptions import MaxRetriesExceededError
 
 from core import factories
 from core.services.yhub_services import ServiceUnavailableError
-from core.tasks.documents import sync_service_deletions_in_cascade
+from core.tasks.documents import (
+    delete_service_documents,
+    sync_service_deletions_in_cascade,
+)
 
 pytestmark = pytest.mark.django_db
 
@@ -116,3 +120,80 @@ def test_sync_service_deletions_keeps_going_on_failure(mock_service):
         mock.call(document),
         mock.call(child),
     ]
+
+
+@mock.patch("core.tasks.documents.YHubService")
+def test_delete_service_documents(mock_service):
+    """Documents deleted for good are deleted on the collaboration server by id."""
+    result = delete_service_documents.apply(args=[["first-id", "second-id"]])
+
+    assert result.successful()
+
+    assert mock_service.return_value.delete_ydoc.call_args_list == [
+        mock.call("first-id"),
+        mock.call("second-id"),
+    ]
+
+
+@mock.patch("core.tasks.documents.YHubService")
+def test_delete_service_documents_retries_the_failed_ones(mock_service):
+    """A document failing should not stop the others, and be retried, alone."""
+    mock_service.return_value.delete_ydoc.side_effect = [
+        ServiceUnavailableError("yhub is down"),
+        None,
+        ServiceUnavailableError("yhub is down"),
+    ]
+
+    with mock.patch.object(delete_service_documents, "retry") as mock_retry:
+        result = delete_service_documents.apply(
+            args=[["first-id", "second-id", "third-id"]]
+        )
+
+    assert result.successful()
+
+    assert mock_service.return_value.delete_ydoc.call_args_list == [
+        mock.call("first-id"),
+        mock.call("second-id"),
+        mock.call("third-id"),
+    ]
+    mock_retry.assert_called_once_with(args=[["first-id", "third-id"]], countdown=30)
+
+
+@mock.patch("core.tasks.documents.YHubService")
+def test_delete_service_documents_retry_countdown_doubles(mock_service):
+    """Each attempt waits twice as long as the one before, up to a ceiling."""
+    mock_service.return_value.delete_ydoc.side_effect = ServiceUnavailableError("down")
+
+    with mock.patch.object(delete_service_documents, "retry") as mock_retry:
+        result = delete_service_documents.apply(args=[["first-id"]], retries=3)
+
+    assert result.successful()
+    mock_retry.assert_called_once_with(args=[["first-id"]], countdown=240)
+
+
+@mock.patch("core.tasks.documents.YHubService")
+def test_delete_service_documents_gives_up_after_the_retries(mock_service, caplog):
+    """Once the retries are spent, the ids are logged: nothing else holds them."""
+    mock_service.return_value.delete_ydoc.side_effect = ServiceUnavailableError("down")
+
+    with mock.patch.object(
+        delete_service_documents, "retry", side_effect=MaxRetriesExceededError()
+    ):
+        result = delete_service_documents.apply(args=[["first-id", "second-id"]])
+
+    assert result.successful()
+    assert (
+        "giving up on deleting documents first-id, second-id on the collaboration "
+        "server"
+    ) in caplog.text
+
+
+@mock.patch("core.tasks.documents.YHubService")
+def test_delete_service_documents_no_retry_when_all_done(mock_service):
+    """Nothing to retry when every deletion went through."""
+    with mock.patch.object(delete_service_documents, "retry") as mock_retry:
+        result = delete_service_documents.apply(args=[["first-id"]])
+
+    assert result.successful()
+    mock_service.return_value.delete_ydoc.assert_called_once_with("first-id")
+    mock_retry.assert_not_called()