feat: media deduplication

synapse/media/_base.py

@@ -534,6 +534,9 @@ class FileInfo:
     url_cache: bool = False
     # Whether the file is a thumbnail or not.
     thumbnail: ThumbnailInfo | None = None
+    # The SHA-256 of the original media, if known. Files are stored under this
+    # instead of `file_id`, with the `file_id` path kept as a fallback for older files.
+    sha256: str | None = None
 
     # The below properties exist to maintain compatibility with third-party modules.
     @property

synapse/media/storage_provider.py

@@ -20,10 +20,11 @@
 #
 
 import abc
+import errno
 import logging
 import os
 import shutil
-from typing import TYPE_CHECKING, Callable
+from typing import TYPE_CHECKING
 
 from synapse.config._base import Config
 from synapse.logging.context import defer_to_thread, run_in_background
@@ -66,6 +67,28 @@ async def fetch(self, path: str, file_info: FileInfo) -> Responder | None:
             Returns a Responder if the provider has the file, otherwise returns None.
         """
 
+    @abc.abstractmethod
+    async def delete_file(self, path: str, file_info: FileInfo) -> None:
+        """Remove a previously stored file. Missing files are ignored.
+
+        Content-addressed paths are shared, so callers only invoke this when the
+        path has no remaining references. The default implementation does nothing
+        so existing third-party providers keep working.
+
+        Args:
+            path: Relative path of file in local cache
+            file_info: The metadata of the file.
+        """
+
+    @abc.abstractmethod
+    async def delete_directory(self, path: str, file_info: FileInfo) -> None:
+        """Remove a previously stored directory, such as a thumbnail directory.
+
+        Args:
+            path: Relative path of the directory in the local cache.
+            file_info: The metadata of the media the directory belongs to.
+        """
+
 
 class StorageProviderWrapper(StorageProvider):
     """Wraps a storage provider and provides various config options
@@ -133,6 +156,27 @@ async def fetch(self, path: str, file_info: FileInfo) -> Responder | None:
         # against improper implementations.
         return await maybe_awaitable(self.backend.fetch(path, file_info))
 
+    def _stores_file(self, file_info: FileInfo) -> bool:
+        if file_info.url_cache:
+            return False
+        if not file_info.server_name and not self.store_local:
+            return False
+        if file_info.server_name and not self.store_remote:
+            return False
+        return True
+
+    @trace_with_opname("StorageProviderWrapper.delete_file")
+    async def delete_file(self, path: str, file_info: FileInfo) -> None:
+        if not self._stores_file(file_info):
+            return
+        await maybe_awaitable(self.backend.delete_file(path, file_info))  # type: ignore[arg-type]
+
+    @trace_with_opname("StorageProviderWrapper.delete_directory")
+    async def delete_directory(self, path: str, file_info: FileInfo) -> None:
+        if not self._stores_file(file_info):
+            return
+        await maybe_awaitable(self.backend.delete_directory(path, file_info))  # type: ignore[arg-type]
+
 
 class FileStorageProviderBackend(StorageProvider):
     """A storage provider that stores files in a directory on a filesystem.
@@ -158,19 +202,25 @@ async def store_file(self, path: str, file_info: FileInfo) -> None:
         primary_fname = os.path.join(self.cache_directory, path)
         backup_fname = os.path.join(self.base_directory, path)
 
-        dirname = os.path.dirname(backup_fname)
-        os.makedirs(dirname, exist_ok=True)
-
-        # mypy needs help inferring the type of the second parameter, which is generic
-        shutil_copyfile: Callable[[str, str], str] = shutil.copyfile
-        with start_active_span("shutil_copyfile"):
+        with start_active_span("copy if absent"):
             await defer_to_thread(
-                self.reactor,
-                shutil_copyfile,
-                primary_fname,
-                backup_fname,
+                self.reactor, _copy_if_absent, primary_fname, backup_fname
             )
 
+    @trace_with_opname("FileStorageProviderBackend.delete_file")
+    async def delete_file(self, path: str, file_info: FileInfo) -> None:
+        """See StorageProvider.delete_file"""
+        backup_fname = os.path.join(self.base_directory, path)
+        await defer_to_thread(self.reactor, _remove_file, backup_fname)
+
+    @trace_with_opname("FileStorageProviderBackend.delete_directory")
+    async def delete_directory(self, path: str, file_info: FileInfo) -> None:
+        """See StorageProvider.delete_directory"""
+        backup_directory = os.path.join(self.base_directory, path)
+        await defer_to_thread(
+            self.reactor, shutil.rmtree, backup_directory, ignore_errors=True
+        )
+
     @trace_with_opname("FileStorageProviderBackend.fetch")
     async def fetch(self, path: str, file_info: FileInfo) -> Responder | None:
         """See StorageProvider.fetch"""
@@ -195,3 +245,32 @@ def parse_config(config: dict) -> str:
         just pull that out.
         """
         return Config.ensure_directory(config["directory"])
+
+
+def _copy_if_absent(src: str, dest: str) -> None:
+    """Copy `src` to `dest` unless `dest` already exists.
+
+    A hard link is used when both paths are on the same filesystem, so the
+    destination appears with its final contents. An existing destination is
+    left unchanged.
+    """
+    os.makedirs(os.path.dirname(dest), exist_ok=True)
+    try:
+        os.link(src, dest)
+        return
+    except FileExistsError:
+        return
+    except OSError as e:
+        if e.errno != errno.EXDEV:
+            raise
+    if os.path.exists(dest):
+        return
+    shutil.copyfile(src, dest)
+
+
+def _remove_file(path: str) -> None:
+    """Remove `path`, ignoring a missing file."""
+    try:
+        os.remove(path)
+    except FileNotFoundError:
+        pass

synapse/media/thumbnailer.py

@@ -451,6 +451,7 @@ async def respond_local_thumbnail(
             server_name=None,
             for_federation=for_federation,
             media_info=media_info,
+            sha256=media_info.sha256,
         )
 
     async def select_or_generate_local_thumbnail(
@@ -496,6 +497,7 @@ async def select_or_generate_local_thumbnail(
                     file_id=media_id,
                     url_cache=bool(media_info.url_cache),
                     thumbnail=info,
+                    sha256=media_info.sha256,
                 )
 
                 responder = await self.media_storage.fetch_media(file_info)
@@ -527,6 +529,7 @@ async def select_or_generate_local_thumbnail(
             desired_type,
             url_cache=bool(media_info.url_cache),
             animated=animated,
+            sha256=media_info.sha256,
         )
 
         if thumbnail_result:
@@ -604,6 +607,7 @@ async def select_or_generate_remote_thumbnail(
                     server_name=server_name,
                     file_id=file_id,
                     thumbnail=info,
+                    sha256=media_info.sha256,
                 )
 
                 responder = await self.media_storage.fetch_media(file_info)
@@ -625,6 +629,7 @@ async def select_or_generate_remote_thumbnail(
             desired_method,
             desired_type,
             animated=animated,
+            sha256=media_info.sha256,
         )
 
         if file_path:
@@ -686,6 +691,7 @@ async def respond_remote_thumbnail(
             url_cache=False,
             server_name=server_name,
             for_federation=False,
+            sha256=media_info.sha256,
         )
 
     async def _select_and_respond_with_thumbnail(
@@ -702,6 +708,7 @@ async def _select_and_respond_with_thumbnail(
         for_federation: bool,
         media_info: LocalMedia | None = None,
         server_name: str | None = None,
+        sha256: str | None = None,
     ) -> None:
         """
         Respond to a request with an appropriate thumbnail from the previously generated thumbnails.
@@ -718,6 +725,7 @@ async def _select_and_respond_with_thumbnail(
             server_name: The server name, if this is a remote thumbnail.
             for_federation: whether the request is from the federation /thumbnail request
             media_info: metadata about the media being requested.
+            sha256: The SHA-256 of the media, if known.
         """
         logger.debug(
             "_select_and_respond_with_thumbnail: media_id=%s desired=%sx%s (%s) thumbnail_infos=%s",
@@ -742,6 +750,7 @@ async def _select_and_respond_with_thumbnail(
                 file_id,
                 url_cache,
                 server_name,
+                sha256,
             )
             if not file_info:
                 logger.info("Couldn't find a thumbnail matching the desired inputs")
@@ -797,6 +806,7 @@ async def _select_and_respond_with_thumbnail(
                     t_method=file_info.thumbnail.method,
                     t_type=file_info.thumbnail.type,
                     animated=regen_animated,
+                    sha256=sha256,
                 )
             else:
                 await self.media_repo.generate_local_exact_thumbnail(
@@ -807,6 +817,7 @@ async def _select_and_respond_with_thumbnail(
                     t_type=file_info.thumbnail.type,
                     url_cache=url_cache,
                     animated=regen_animated,
+                    sha256=sha256,
                 )
 
             responder = await self.media_storage.fetch_media(file_info)
@@ -863,6 +874,7 @@ def _select_thumbnail(
         file_id: str,
         url_cache: bool,
         server_name: str | None,
+        sha256: str | None = None,
     ) -> FileInfo | None:
         """
         Choose an appropriate thumbnail from the previously generated thumbnails.
@@ -876,6 +888,7 @@ def _select_thumbnail(
             file_id: The ID of the media that a thumbnail is being requested for.
             url_cache: True if this is from a URL cache.
             server_name: The server name, if this is a remote thumbnail.
+            sha256: The SHA-256 of the media, if known.
 
         Returns:
              The thumbnail which best matches the desired parameters.
@@ -976,6 +989,7 @@ def _select_thumbnail(
                 url_cache=url_cache,
                 server_name=server_name,
                 thumbnail=thumbnail_info,
+                sha256=sha256,
             )
 
         # No matching thumbnail was found.

synapse/storage/databases/main/media_repository.py

@@ -988,6 +988,34 @@ def _delete_url_cache_media_txn(txn: LoggingTransaction) -> None:
             "delete_url_cache_media", _delete_url_cache_media_txn
         )
 
+    async def is_media_sha256_in_use(
+        self, sha256: str, origin: str | None = None
+    ) -> bool:
+        """Whether any media is still stored under the given SHA-256 path.
+
+        Args:
+            sha256: The SHA-256 of the media.
+            origin: The origin to check remote media from, or None to check local
+                media. URL cache media is ignored, as it is not stored by SHA-256.
+        """
+        if origin is None:
+            sql = """
+                SELECT 1 FROM local_media_repository
+                WHERE sha256 = ? AND url_cache IS NULL
+                LIMIT 1
+            """
+            rows = await self.db_pool.execute("is_media_sha256_in_use", sql, sha256)
+        else:
+            sql = """
+                SELECT 1 FROM remote_media_cache
+                WHERE media_origin = ? AND sha256 = ?
+                LIMIT 1
+            """
+            rows = await self.db_pool.execute(
+                "is_media_sha256_in_use", sql, origin, sha256
+            )
+        return bool(rows)
+
     async def get_is_hash_quarantined(self, sha256: str) -> bool:
         """Get whether a specific sha256 hash digest matches any quarantined media.