♻️(backend) save item to mirror in the database

In order to be more fault tolerant, to not lose data if the redis is
reset, etc. We decided to put the item to mirror in a new table and then
the task will manage the status in the database.
This is the first step to then add tasks in the admin to retry or check
why the task has failed and also implement a retry strategy.

The Dockerfile must also be modified, we need the libmagic library to
build the image at the collect stage because the magic module is
imported in the storage module and all tasks are imported when the
application starts.

Dockerfile

@@ -9,19 +9,19 @@ RUN python -m pip install --upgrade pip
 
 # Upgrade system packages to install security updates
 RUN apk update && \
-  apk upgrade && \
-  apk add git
+    apk upgrade && \
+    apk add git
 
 # ---- Back-end builder image ----
 FROM base AS back-builder
 
 
-ENV UV_COMPILE_BYTECODE=1 
+ENV UV_COMPILE_BYTECODE=1
 ENV UV_LINK_MODE=copy
 
 # Disable Python downloads, because we want to use the system interpreter
 # across both images. If using a managed Python version, it needs to be
-# copied from the build image into the final image; 
+# copied from the build image into the final image;
 ENV UV_PYTHON_DOWNLOADS=0
 
 # install uv
@@ -52,10 +52,11 @@ RUN yarn install --frozen-lockfile && \
 FROM base AS link-collector
 ARG DRIVE_STATIC_ROOT=/data/static
 
-# Install pango & rdfind
+# Install libmagic, pango & rdfind
 RUN apk add \
-  pango \
-  rdfind
+    libmagic \
+    pango \
+    rdfind
 
 WORKDIR /app
 
@@ -80,16 +81,16 @@ ENV PYTHONUNBUFFERED=1
 
 # Install required system libs
 RUN apk add \
-  cairo \
-  file \
-  font-noto \
-  font-noto-emoji \
-  gettext \
-  gdk-pixbuf \
-  libffi-dev \
-  pandoc \
-  pango \
-  shared-mime-info
+    cairo \
+    file \
+    font-noto \
+    font-noto-emoji \
+    gettext \
+    gdk-pixbuf \
+    libffi-dev \
+    pandoc \
+    pango \
+    shared-mime-info
 
 RUN wget https://svn.apache.org/repos/asf/httpd/httpd/trunk/docs/conf/mime.types -O /etc/mime.types
 

src/backend/core/api/viewsets.py

@@ -43,13 +43,13 @@ from rest_framework_api_key.permissions import HasAPIKey
 
 from core import enums, models
 from core.entitlements import get_entitlements_backend
+from core.services.mirror import mirror_item
 from core.services.sdk_relay import SDKRelayManager
 from core.services.search_indexers import (
     get_file_indexer,
     get_visited_items_ids_of,
 )
 from core.tasks.item import process_item_deletion, rename_file
-from core.tasks.storage import mirror_file
 from wopi.services import access as access_service
 from wopi.utils import compute_wopi_launch_url, get_wopi_client_config
 
@@ -734,7 +734,7 @@ class ItemViewSet(
                 )
 
         malware_detection.analyse_file(item.file_key, item_id=item.id)
-        mirror_file.delay(item.file_key)
+        mirror_item(item)
 
         serializer = self.get_serializer(item)
 

src/backend/core/factories.py

@@ -174,3 +174,12 @@ class InvitationFactory(factory.django.DjangoModelFactory):
     item = factory.SubFactory(ItemFactory)
     role = factory.fuzzy.FuzzyChoice([role[0] for role in RoleChoices.choices])
     issuer = factory.SubFactory(UserFactory)
+
+
+class MirrorItemTaskFactory(factory.django.DjangoModelFactory):
+    """A factory to create mirror item tasks for testing."""
+
+    class Meta:
+        model = models.MirrorItemTask
+
+    item = factory.SubFactory(ItemFactory)

src/backend/core/migrations/0018_mirroritemtask.py

@@ -0,0 +1,32 @@
+# Generated by Django 5.2.11 on 2026-02-10 07:08
+
+import django.db.models.deletion
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0017_alter_user_short_name'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='MirrorItemTask',
+            fields=[
+                ('id', models.UUIDField(default=uuid.uuid4, editable=False, help_text='primary key for the record as UUID', primary_key=True, serialize=False, verbose_name='id')),
+                ('created_at', models.DateTimeField(auto_now_add=True, help_text='date and time at which a record was created', verbose_name='created on')),
+                ('updated_at', models.DateTimeField(auto_now=True, help_text='date and time at which a record was last updated', verbose_name='updated on')),
+                ('status', models.CharField(choices=[('pending', 'Pending'), ('processing', 'Processing'), ('completed', 'Completed'), ('failed', 'Failed')], default='pending', max_length=25)),
+                ('error_details', models.TextField(blank=True, null=True)),
+                ('retries', models.IntegerField(default=0)),
+                ('item', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='mirror_tasks', to='core.item')),
+            ],
+            options={
+                'verbose_name': 'Mirror item task',
+                'verbose_name_plural': 'Mirror item tasks',
+                'db_table': 'drive_mirror_item_task',
+            },
+        ),
+    ]

src/backend/core/models.py

@@ -77,6 +77,15 @@ class ItemUploadStateChoices(models.TextChoices):
     READY = "ready", _("Ready")
 
 
+class MirrorItemTaskStatusChoices(models.TextChoices):
+    """Defines the possible statuses for a mirroring task."""
+
+    PENDING = "pending", _("Pending")
+    PROCESSING = "processing", _("Processing")
+    COMPLETED = "completed", _("Completed")
+    FAILED = "failed", _("Failed")
+
+
 class DuplicateEmailError(Exception):
     """Raised when an email is already associated with a pre-existing user."""
 
@@ -1036,6 +1045,31 @@ class Item(TreeModel, BaseModel):
             self._meta.model.objects.filter(pk=old_parent_id).update(**update)
 
 
+class MirrorItemTask(BaseModel):
+    """Model managing a status for a mirroring task."""
+
+    item = models.ForeignKey(
+        Item,
+        on_delete=models.CASCADE,
+        related_name="mirror_tasks",
+    )
+    status = models.CharField(
+        max_length=25,
+        choices=MirrorItemTaskStatusChoices.choices,
+        default=MirrorItemTaskStatusChoices.PENDING,
+    )
+    error_details = models.TextField(null=True, blank=True)
+    retries = models.IntegerField(default=0)
+
+    class Meta:
+        db_table = "drive_mirror_item_task"
+        verbose_name = _("Mirror item task")
+        verbose_name_plural = _("Mirror item tasks")
+
+    def __str__(self):
+        return f"Mirror task for item {self.item!s} with status {self.status!s}"
+
+
 class LinkTrace(BaseModel):
     """
     Relation model to trace accesses to an item via a link by a logged-in user.

src/backend/core/services/mirror.py

@@ -0,0 +1,47 @@
+"""Services related to mirroring."""
+
+import logging
+from uuid import UUID
+
+from core.models import Item, MirrorItemTask, MirrorItemTaskStatusChoices
+from core.tasks.storage import get_mirror_s3_client, mirror_file
+
+logger = logging.getLogger(__name__)
+
+
+def mirror_item_from_file_key(file_key: str):
+    """Mirror an item from a file key."""
+
+    if not file_key.startswith("item/"):
+        logger.error("File key %s is not a valid item file key", file_key)
+        return
+
+    # key is in the form item/<item_id>/<filename>
+    parts = file_key.split("/")
+
+    try:
+        item_id = UUID(parts[1])
+    except ValueError:
+        logger.error("Item ID %s is not a valid UUID", parts[1])
+        return
+
+    try:
+        item = Item.objects.get(id=item_id)
+    except Item.DoesNotExist:
+        logger.error("Item %s does not exist", item_id)
+        return
+    mirror_item(item)
+
+
+def mirror_item(item: Item):
+    """Mirror an item to the mirroring S3 bucket."""
+    mirror_s3_client = get_mirror_s3_client()
+    if not mirror_s3_client:
+        logger.info("Mirroring S3 bucket is not configured, skipping mirroring")
+        return
+
+    mirror_task = MirrorItemTask.objects.create(
+        item=item, status=MirrorItemTaskStatusChoices.PENDING
+    )
+
+    mirror_file.delay(mirror_task.id)

src/backend/core/storage/s3_mirroring_storage.py

@@ -2,7 +2,7 @@
 
 from storages.backends.s3 import S3Storage
 
-from core.tasks.storage import mirror_file
+from core.services.mirror import mirror_item_from_file_key
 
 # pylint: disable=abstract-method
 
@@ -14,5 +14,5 @@ class S3MirroringStorage(S3Storage):
         """Dispatch a celery task to mirror the file on the other S3 buckets."""
         name = super().save(name, content, max_length)
 
-        mirror_file.delay(name)
+        mirror_item_from_file_key(name)
         return name

src/backend/core/tasks/storage.py

@@ -1,7 +1,7 @@
 """Tasks related to storage."""
 
-from functools import cache
 import logging
+from functools import cache
 
 from django.conf import settings
 from django.core.files.storage import default_storage
@@ -9,14 +9,14 @@ from django.core.files.storage import default_storage
 import boto3
 import botocore
 
+from core.api.utils import get_item_file_head_object
+from core.models import MirrorItemTask, MirrorItemTaskStatusChoices
+
 from drive.celery_app import app
 
 logger = logging.getLogger(__name__)
 
 
-class MissingMirroringConfigurationError(ValueError):
-    """Exception raised when the mirroring S3 bucket configuration is missing."""
-
 @cache
 def get_mirror_s3_client():
     """Get the S3 client for the mirroring S3 bucket."""
@@ -25,9 +25,7 @@ def get_mirror_s3_client():
         or not settings.AWS_S3_MIRRORING_SECRET_ACCESS_KEY
         or not settings.AWS_S3_MIRRORING_ENDPOINT_URL
     ):
-        raise MissingMirroringConfigurationError(
-            "Missing required configuration for mirroring S3 bucket"
-        )
+        return None
 
     return boto3.client(
         "s3",
@@ -43,24 +41,45 @@ def get_mirror_s3_client():
     )
 
 
-@app.task
-def mirror_file(name):
+@app.task()
+def mirror_file(mirror_task_id):
     """Copy the file to the mirroring S3 bucket."""
-    try:
-        mirror_s3_client = get_mirror_s3_client()
-    except MissingMirroringConfigurationError:
+    mirror_s3_client = get_mirror_s3_client()
+    if not mirror_s3_client:
         logger.info("Mirroring S3 bucket is not configured, skipping mirroring")
         return
 
+    try:
+        mirror_task = MirrorItemTask.objects.select_related("item").get(
+            id=mirror_task_id
+        )
+    except MirrorItemTask.DoesNotExist:
+        logger.error("Mirror task %s does not exist", mirror_task_id)
+        return
+
+    if mirror_task.status != MirrorItemTaskStatusChoices.PENDING:
+        logger.info("Mirror task %s is not pending, skipping mirroring", mirror_task_id)
+        return
+
+    mirror_task.status = MirrorItemTaskStatusChoices.PROCESSING
+    mirror_task.save(update_fields=["status", "updated_at"])
+    item_key = mirror_task.item.file_key
+
     mirror_bucket = settings.AWS_S3_MIRRORING_STORAGE_BUCKET_NAME
 
-    logger.info("Starting mirror of file %s to bucket %s", name, mirror_bucket)
+    logger.info("Starting mirror of file %s to bucket %s", item_key, mirror_bucket)
 
-    with default_storage.open(name, mode="rb") as source_file:
+    head_object = get_item_file_head_object(mirror_task.item)
+
+    with default_storage.open(item_key, mode="rb") as source_file:
         mirror_s3_client.put_object(
             Bucket=mirror_bucket,
-            Key=name,
+            Key=item_key,
             Body=source_file,
+            ContentType=mirror_task.item.mimetype,
+            Metadata=head_object["Metadata"],
         )
 
-    logger.info("Successfully mirrored file %s to bucket %s", name, mirror_bucket)
+    logger.info("Successfully mirrored file %s to bucket %s", item_key, mirror_bucket)
+    mirror_task.status = MirrorItemTaskStatusChoices.COMPLETED
+    mirror_task.save(update_fields=["status", "updated_at"])

src/backend/core/tests/items/test_api_item_upload_ended.py

@@ -10,7 +10,7 @@ import pytest
 from rest_framework.test import APIClient
 
 from core import factories, models
-from core.api.viewsets import malware_detection, mirror_file
+from core.api.viewsets import malware_detection
 from core.models import ItemTypeChoices, ItemUploadStateChoices, LinkRoleChoices
 
 pytestmark = pytest.mark.django_db
@@ -115,12 +115,12 @@ def test_api_item_upload_ended_success():
 
     with (
         mock.patch.object(malware_detection, "analyse_file") as mock_analyse_file,
-        mock.patch.object(mirror_file, "delay") as mock_mirror_file,
+        mock.patch("core.api.viewsets.mirror_item") as mock_mirror_item,
     ):
         response = client.post(f"/api/v1.0/items/{item.id!s}/upload-ended/")
 
     mock_analyse_file.assert_called_once_with(item.file_key, item_id=item.id)
-    mock_mirror_file.assert_called_once_with(item.file_key)
+    mock_mirror_item.assert_called_once_with(item)
     assert response.status_code == 200
 
     item.refresh_from_db()
@@ -144,12 +144,12 @@ def test_api_item_upload_ended_empty_file():
 
     with (
         mock.patch.object(malware_detection, "analyse_file") as mock_analyse_file,
-        mock.patch.object(mirror_file, "delay") as mock_mirror_file,
+        mock.patch("core.api.viewsets.mirror_item") as mock_mirror_item,
     ):
         response = client.post(f"/api/v1.0/items/{item.id!s}/upload-ended/")
 
     mock_analyse_file.assert_called_once_with(item.file_key, item_id=item.id)
-    mock_mirror_file.assert_called_once_with(item.file_key)
+    mock_mirror_item.assert_called_once_with(item)
     assert response.status_code == 200
 
     item.refresh_from_db()

src/backend/core/tests/services/test_mirror.py

@@ -0,0 +1,85 @@
+"""Test the mirror services."""
+
+from unittest import mock
+from uuid import uuid4
+
+import pytest
+
+from core.factories import ItemFactory
+from core.models import ItemTypeChoices, MirrorItemTask, MirrorItemTaskStatusChoices
+from core.services.mirror import (
+    mirror_file,
+    mirror_item,
+    mirror_item_from_file_key,
+)
+
+pytestmark = pytest.mark.django_db
+
+
+def test_mirror_item_from_file_key():
+    """Test the mirror item from file key service."""
+
+    item = ItemFactory(
+        filename="test.txt",
+        type=ItemTypeChoices.FILE,
+    )
+
+    with mock.patch("core.services.mirror.mirror_item") as mock_mirror_item:
+        mirror_item_from_file_key(item.file_key)
+
+    mock_mirror_item.assert_called_once_with(item)
+
+
+def test_mirror_item_from_file_key_invalid_file_key(caplog):
+    """Test the mirror item from file key service with an invalid file key."""
+    with caplog.at_level("ERROR", logger="core.services.mirror"):
+        mirror_item_from_file_key("invalid/file/key")
+
+    assert "File key invalid/file/key is not a valid item file key" in caplog.text
+
+
+def test_mirror_item_from_file_key_item_not_found(caplog):
+    """Test the mirror item from file key service with an item not found."""
+    item_id = uuid4()
+    with caplog.at_level("ERROR", logger="core.services.mirror"):
+        mirror_item_from_file_key(f"item/{item_id}/test.txt")
+
+    assert f"Item {item_id} does not exist" in caplog.text
+
+
+def test_mirror_item_from_file_key_invalid_item_id(caplog):
+    """Test the mirror item from file key service with an invalid item ID."""
+    with caplog.at_level("ERROR", logger="core.services.mirror"):
+        mirror_item_from_file_key("item/invalid/test.txt")
+
+    assert "Item ID invalid is not a valid UUID" in caplog.text
+
+
+def test_mirror_item():
+    """Test the mirror item service."""
+
+    item = ItemFactory(
+        filename="test.txt",
+        type=ItemTypeChoices.FILE,
+    )
+
+    with (
+        mock.patch("core.services.mirror.get_mirror_s3_client", return_value=True),
+        mock.patch.object(mirror_file, "delay") as mock_mirror_file,
+    ):
+        mirror_item(item)
+
+    mirror_item_task = MirrorItemTask.objects.get(item=item)
+    assert mirror_item_task.status == MirrorItemTaskStatusChoices.PENDING
+    mock_mirror_file.assert_called_once_with(mirror_item_task.id)
+
+
+def test_mirror_item_no_mirror_s3_client(caplog):
+    """Test the mirror item service with no mirror S3 client."""
+
+    item = ItemFactory()
+
+    with caplog.at_level("INFO", logger="core.services.mirror"):
+        mirror_item(item)
+
+    assert "Mirroring S3 bucket is not configured, skipping mirroring" in caplog.text

src/backend/core/tests/storage/test_s3_mirroring_storage.py

@@ -1,18 +1,20 @@
 """Test the S3MirroringStorage class."""
 
 from io import BytesIO
-from unittest.mock import patch
+from unittest import mock
 
-from core.storage.s3_mirroring_storage import S3MirroringStorage, mirror_file
+from core.storage.s3_mirroring_storage import S3MirroringStorage
 
 
 def test_s3_mirroring_storage():
     """Test the S3MirroringStorage class."""
     storage = S3MirroringStorage()
 
-    with patch.object(mirror_file, "delay") as mock_mirror_file:
+    with mock.patch(
+        "core.storage.s3_mirroring_storage.mirror_item_from_file_key"
+    ) as mock_mirror_item_from_file_key:
         name = storage.save("test.txt", BytesIO(b"test"))
 
-    mock_mirror_file.assert_called_once_with("test.txt")
+    mock_mirror_item_from_file_key.assert_called_once_with("test.txt")
     assert storage.exists("test.txt")
     assert name == "test.txt"

src/backend/core/tests/tasks/storage/test_mirror_file.py

@@ -2,26 +2,35 @@
 
 from io import BytesIO
 from unittest import mock
+from uuid import uuid4
 
 from django.core.files.storage import default_storage
 
 import pytest
 
+from core.factories import MirrorItemTaskFactory
+from core.models import ItemTypeChoices, MirrorItemTaskStatusChoices
 from core.tasks.storage import (
-    MissingMirroringConfigurationError,
     get_mirror_s3_client,
     mirror_file,
 )
 
+pytestmark = pytest.mark.django_db
 
-def test_get_mirror_s3_client_no_config_should_raise_an_exception(settings):
-    """Test get_mirror_s3_client without config should raise an exception."""
+
+@pytest.fixture(autouse=True)
+def clear_get_mirror_s3_client_cache():
+    """Clear the cache for get_mirror_s3_client."""
+    get_mirror_s3_client.cache_clear()
+
+
+def test_get_mirror_s3_client_no_config_should_return_none(settings):
+    """Test get_mirror_s3_client without config should return None."""
     settings.AWS_S3_MIRRORING_ACCESS_KEY_ID = None
     settings.AWS_S3_MIRRORING_SECRET_ACCESS_KEY = None
     settings.AWS_S3_MIRRORING_ENDPOINT_URL = None
 
-    with pytest.raises(MissingMirroringConfigurationError):
-        get_mirror_s3_client()
+    assert get_mirror_s3_client() is None
 
 
 def test_mirror_no_file_s3_config_should_abort(settings, caplog):
@@ -39,16 +48,30 @@ def test_mirror_no_file_s3_config_should_abort(settings, caplog):
         )
 
 
-def test_mirror_file(settings, caplog):
-    """Test mirror file correctly configured."""
+def test_mirror_file_existing_record_not_existing(caplog):
+    """Test mirror file mirror_item_task not existing."""
 
-    settings.AWS_S3_MIRRORING_ACCESS_KEY_ID = "access_key_id"
-    settings.AWS_S3_MIRRORING_SECRET_ACCESS_KEY = "secret_access_key"
-    settings.AWS_S3_MIRRORING_ENDPOINT_URL = "endpoint_url"
-    settings.AWS_S3_MIRRORING_STORAGE_BUCKET_NAME = "test_mirror"
+    mirror_item_task_id = uuid4()
 
-    file_content = b"content to mirror"
-    default_storage.save("test_mirror.txt", BytesIO(file_content))
+    with (
+        mock.patch(
+            "core.tasks.storage.get_mirror_s3_client"
+        ) as mock_get_mirror_s3_client,
+        caplog.at_level("ERROR", logger="core.tasks.storage"),
+    ):
+        mock_mirror_s3_client = mock.MagicMock()
+        mock_get_mirror_s3_client.return_value = mock_mirror_s3_client
+        mirror_file(mirror_item_task_id)
+
+    assert f"Mirror task {mirror_item_task_id} does not exist" in caplog.text
+
+
+def test_mirror_file_existing_record_not_in_pending_status(caplog):
+    """Test mirror file mirror_item_task not in pending status."""
+
+    mirror_item_task = MirrorItemTaskFactory(
+        status=MirrorItemTaskStatusChoices.PROCESSING
+    )
 
     with (
         mock.patch(
@@ -58,23 +81,60 @@ def test_mirror_file(settings, caplog):
     ):
         mock_mirror_s3_client = mock.MagicMock()
         mock_get_mirror_s3_client.return_value = mock_mirror_s3_client
-        mirror_file("test_mirror.txt")
+        mirror_file(mirror_item_task.id)
+
+    assert (
+        f"Mirror task {mirror_item_task.id} is not pending, skipping mirroring"
+        in caplog.text
+    )
+
+    mirror_item_task.refresh_from_db()
+    assert mirror_item_task.status == MirrorItemTaskStatusChoices.PROCESSING
+
+
+def test_mirror_file(settings, caplog):
+    """Test mirror file correctly configured."""
+    settings.AWS_S3_MIRRORING_STORAGE_BUCKET_NAME = "test_mirror"
+
+    mirror_item_task = MirrorItemTaskFactory(
+        item__type=ItemTypeChoices.FILE,
+        item__filename="test_mirror.txt",
+        item__mimetype="text/plain",
+        status=MirrorItemTaskStatusChoices.PENDING,
+    )
+    item = mirror_item_task.item
+
+    file_content = b"content to mirror"
+    default_storage.save(item.file_key, BytesIO(file_content))
+
+    with (
+        mock.patch(
+            "core.tasks.storage.get_mirror_s3_client"
+        ) as mock_get_mirror_s3_client,
+        caplog.at_level("INFO", logger="core.tasks.storage"),
+    ):
+        mock_mirror_s3_client = mock.MagicMock()
+        mock_get_mirror_s3_client.return_value = mock_mirror_s3_client
+        mirror_file(mirror_item_task.id)
 
         mock_get_mirror_s3_client.assert_called_once()
         mock_mirror_s3_client.put_object.assert_called_once_with(
             Bucket="test_mirror",
-            Key="test_mirror.txt",
+            Key=item.file_key,
             Body=mock.ANY,
+            ContentType="text/plain",
+            Metadata={},
         )
-
+        mirror_item_task.refresh_from_db()
+        assert mirror_item_task.status == MirrorItemTaskStatusChoices.COMPLETED
         assert (
             mock_mirror_s3_client.put_object.call_args[1]["Body"].read() == file_content
         )
         assert (
-            "Starting mirror of file test_mirror.txt to bucket test_mirror"
+            f"Starting mirror of file {item.file_key} to bucket test_mirror"
             in caplog.text
         )
         assert (
-            "Successfully mirrored file test_mirror.txt to bucket test_mirror"
+            f"Successfully mirrored file {item.file_key} to bucket test_mirror"
             in caplog.text
         )