diff --git a/api/admin/controller/collection_settings.py b/api/admin/controller/collection_settings.py
index bbf91a8264..7144c365b6 100644
--- a/api/admin/controller/collection_settings.py
+++ b/api/admin/controller/collection_settings.py
@@ -19,6 +19,7 @@
 )
 from api.circulation import CirculationApiType
 from api.integration.registry.license_providers import LicenseProvidersRegistry
+from core.celery.tasks.collection_delete import collection_delete
 from core.integration.base import HasChildIntegrationConfiguration
 from core.integration.registry import IntegrationRegistry
 from core.model import (
@@ -169,6 +170,7 @@ def process_delete(self, service_id: int) -> Response | ProblemDetail:
 
         # Flag the collection to be deleted by script in the background.
         collection.marked_for_deletion = True
+        collection_delete.delay(collection.id)
         return Response("Deleted", 200)
 
     def process_collection_self_tests(
diff --git a/core/celery/tasks/collection_delete.py b/core/celery/tasks/collection_delete.py
new file mode 100644
index 0000000000..23c88cb23e
--- /dev/null
+++ b/core/celery/tasks/collection_delete.py
@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from celery import shared_task
+from sqlalchemy import select
+from sqlalchemy.orm import Session, sessionmaker
+
+from core.celery.job import Job
+from core.celery.task import Task
+from core.model import Collection
+
+
+class CollectionDeleteJob(Job):
+    def __init__(self, session_maker: sessionmaker[Session], collection_id: int):
+        super().__init__(session_maker)
+        self.collection_id = collection_id
+
+    @staticmethod
+    def collection(session: Session, collection_id: int) -> Collection | None:
+        return (
+            session.execute(select(Collection).where(Collection.id == collection_id))
+            .scalars()
+            .one_or_none()
+        )
+
+    @staticmethod
+    def collection_name(collection: Collection) -> str:
+        return f"{collection.name}/{collection.protocol} ({collection.id})"
+
+    def run(self) -> None:
+        with self.transaction() as session:
+            collection = self.collection(session, self.collection_id)
+            if collection is None:
+                self.log.error(
+                    f"Collection with id {self.collection_id} not found. Unable to delete."
+                )
+                return
+
+            self.log.info(f"Deleting collection {self.collection_name(collection)}")
+            collection.delete()
+
+
+@shared_task(key="high", bind=True)
+def collection_delete(task: Task, collection_id: int) -> None:
+    CollectionDeleteJob(task.session_maker, collection_id).run()
diff --git a/tests/api/admin/controller/test_collections.py b/tests/api/admin/controller/test_collections.py
index db0a62474e..428f5944e4 100644
--- a/tests/api/admin/controller/test_collections.py
+++ b/tests/api/admin/controller/test_collections.py
@@ -1,5 +1,5 @@
 import json
-from unittest.mock import MagicMock, create_autospec
+from unittest.mock import MagicMock, create_autospec, patch
 
 import flask
 import pytest
@@ -706,7 +706,12 @@ def test_collection_delete(
                 collection.integration_configuration.id,
             )
 
-        with flask_app_fixture.test_request_context_system_admin("/", method="DELETE"):
+        with (
+            flask_app_fixture.test_request_context_system_admin("/", method="DELETE"),
+            patch(
+                "api.admin.controller.collection_settings.collection_delete"
+            ) as mock_delete,
+        ):
             assert collection.integration_configuration.id is not None
             response = controller.process_delete(
                 collection.integration_configuration.id
@@ -721,6 +726,10 @@ def test_collection_delete(
         assert fetched_collection == collection
         assert fetched_collection.marked_for_deletion is True
 
+        # The controller called collection_delete with the correct arguments, to
+        # queue up the collection for deletion in the background.
+        mock_delete.delay.assert_called_once_with(collection.id)
+
     def test_collection_delete_cant_delete_parent(
         self,
         controller: CollectionSettingsController,
diff --git a/tests/core/celery/tasks/test_collection_delete.py b/tests/core/celery/tasks/test_collection_delete.py
new file mode 100644
index 0000000000..9c4a02b994
--- /dev/null
+++ b/tests/core/celery/tasks/test_collection_delete.py
@@ -0,0 +1,61 @@
+from logging import INFO
+
+from _pytest.logging import LogCaptureFixture
+from sqlalchemy import select
+from sqlalchemy.orm import sessionmaker
+
+from core.celery.tasks.collection_delete import CollectionDeleteJob, collection_delete
+from core.model import Collection
+from tests.fixtures.celery import CeleryFixture
+from tests.fixtures.database import DatabaseTransactionFixture
+
+
+def test_collection_delete_job_collection(db: DatabaseTransactionFixture):
+    # A non-existent collection should return None
+    assert CollectionDeleteJob.collection(db.session, 1) is None
+
+    collection = db.collection(name="collection1")
+    assert collection.id is not None
+    assert CollectionDeleteJob.collection(db.session, collection.id) == collection
+
+
+def test_collection_delete_job_collection_name(db: DatabaseTransactionFixture):
+    collection = db.collection(name="collection1")
+    assert (
+        CollectionDeleteJob.collection_name(collection)
+        == f"{collection.name}/{collection.protocol} ({collection.id})"
+    )
+
+
+def test_collection_delete_job_run(
+    db: DatabaseTransactionFixture,
+    mock_session_maker: sessionmaker,
+    caplog: LogCaptureFixture,
+):
+    # A non-existent collection should log an error
+    caplog.set_level(INFO)
+    CollectionDeleteJob(mock_session_maker, 1).run()
+    assert "Collection with id 1 not found. Unable to delete." in caplog.text
+
+    collection = db.collection(name="collection1")
+    collection.marked_for_deletion = True
+    query = select(Collection).where(Collection.id == collection.id)
+
+    assert db.session.execute(query).scalar_one_or_none() == collection
+
+    assert collection.id is not None
+    job = CollectionDeleteJob(mock_session_maker, collection.id)
+    job.run()
+    assert db.session.execute(query).scalar_one_or_none() is None
+    assert f"Deleting collection" in caplog.text
+
+
+def test_collection_delete_task(
+    db: DatabaseTransactionFixture, celery_fixture: CeleryFixture
+):
+    collection = db.collection(name="collection1")
+    collection.marked_for_deletion = True
+    query = select(Collection).where(Collection.id == collection.id)
+    assert db.session.execute(query).scalar_one_or_none() == collection
+    collection_delete.delay(collection.id).wait()
+    assert db.session.execute(query).scalar_one_or_none() is None