diff --git a/src/palace/manager/api/circulation.py b/src/palace/manager/api/circulation.py
index 11d020d49..b7e423da0 100644
--- a/src/palace/manager/api/circulation.py
+++ b/src/palace/manager/api/circulation.py
@@ -1143,6 +1143,14 @@ def borrow(
             )
             if existing_hold:
                 # The book was on hold, and now we have a loan.
+                # collect cm event to commemorate the conversion:
+                self._collect_event(
+                    patron=patron,
+                    licensepool=licensepool,
+                    name=CirculationEvent.CM_HOLD_CONVERTED_TO_LOAN,
+                    include_neighborhood=True,
+                )
+
                 # Delete the record of the hold.
                 self._db.delete(existing_hold)
             __transaction.commit()
@@ -1197,6 +1205,11 @@ def borrow(
             self._collect_event(patron, licensepool, CirculationEvent.CM_HOLD_PLACE)
 
         if existing_loan:
+            # Send out analytics event capturing the unusual circumstance  that a loan was converted to a hold
+            # TODO: Do we know what the conditions under which this situation can occur?
+            self._collect_event(
+                patron, licensepool, CirculationEvent.CM_LOAN_CONVERTED_TO_HOLD
+            )
             self._db.delete(existing_loan)
         __transaction.commit()
         return None, hold, is_new
diff --git a/src/palace/manager/api/monitor.py b/src/palace/manager/api/monitor.py
index 073e2cf49..3f9eed858 100644
--- a/src/palace/manager/api/monitor.py
+++ b/src/palace/manager/api/monitor.py
@@ -2,6 +2,7 @@
 
 from palace.manager.api.opds_for_distributors import OPDSForDistributorsAPI
 from palace.manager.core.monitor import ReaperMonitor
+from palace.manager.sqlalchemy.model.circulationevent import CirculationEvent
 from palace.manager.sqlalchemy.model.collection import Collection
 from palace.manager.sqlalchemy.model.integration import IntegrationConfiguration
 from palace.manager.sqlalchemy.model.licensing import LicensePool
@@ -14,6 +15,10 @@ class LoanlikeReaperMonitor(ReaperMonitor):
         OPDSForDistributorsAPI.label(),
     ]
 
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._events_to_be_logged = []
+
     @property
     def where_clause(self):
         """We never want to automatically reap loans or holds for situations
@@ -44,6 +49,25 @@ def where_clause(self):
         )
         return ~self.MODEL_CLASS.id.in_(source_of_truth_subquery)
 
+    def delete(self, row) -> None:
+        ce = CirculationEvent
+        event_type = ce.CM_LOAN_EXPIRED if isinstance(row, Loan) else ce.CM_HOLD_EXPIRED
+        event = dict(
+            library=row.library,
+            license_pool=row.license_pool,
+            event_type=event_type,
+            patron=row.patron,
+        )
+        super().delete(row)
+        self._events_to_be_logged.append(event)
+
+    def after_commit(self) -> None:
+        super().after_commit()
+        copy_of_list = list(self._events_to_be_logged)
+        for event in copy_of_list:
+            self.services.analytics.collect_event(**event)
+            self._events_to_be_logged.remove(event)
+
 
 class LoanReaper(LoanlikeReaperMonitor):
     """Remove expired and abandoned loans from the database."""
diff --git a/src/palace/manager/celery/tasks/opds_odl.py b/src/palace/manager/celery/tasks/opds_odl.py
index f51931b8f..03d8a1b45 100644
--- a/src/palace/manager/celery/tasks/opds_odl.py
+++ b/src/palace/manager/celery/tasks/opds_odl.py
@@ -1,4 +1,5 @@
 import datetime
+from dataclasses import dataclass
 
 from celery import shared_task
 from sqlalchemy import delete, select
@@ -7,35 +8,63 @@
 
 from palace.manager.api.odl.api import OPDS2WithODLApi
 from palace.manager.celery.task import Task
+from palace.manager.service.analytics.analytics import Analytics
 from palace.manager.service.celery.celery import QueueNames
 from palace.manager.service.redis.models.lock import RedisLock
 from palace.manager.service.redis.redis import Redis
+from palace.manager.sqlalchemy.model.circulationevent import CirculationEvent
 from palace.manager.sqlalchemy.model.collection import Collection
+from palace.manager.sqlalchemy.model.library import Library
 from palace.manager.sqlalchemy.model.licensing import License, LicensePool
-from palace.manager.sqlalchemy.model.patron import Hold
+from palace.manager.sqlalchemy.model.patron import Hold, Patron
 from palace.manager.util.datetime_helpers import utc_now
 
 
-def remove_expired_holds_for_collection(db: Session, collection_id: int) -> int:
+@dataclass
+class CirculationEventData:
+    library: Library
+    license_pool: LicensePool
+    event_type: str
+    patron: Patron
+
+
+def remove_expired_holds_for_collection(
+    db: Session,
+    collection_id: int,
+) -> list[CirculationEventData]:
     """
     Remove expired holds from the database for this collection.
     """
+
+    select_query = select(Hold).where(
+        Hold.position == 0,
+        Hold.end < utc_now(),
+        Hold.license_pool_id == LicensePool.id,
+        LicensePool.collection_id == collection_id,
+    )
+
+    expired_holds = db.scalars(select_query).all()
+    expired_hold_events: list[CirculationEventData] = []
+    for hold in expired_holds:
+        expired_hold_events.append(
+            CirculationEventData(
+                library=hold.library,
+                license_pool=hold.license_pool,
+                event_type=CirculationEvent.CM_HOLD_EXPIRED,
+                patron=hold.patron,
+            )
+        )
+
+    # delete the holds
     query = (
         delete(Hold)
-        .where(
-            Hold.position == 0,
-            Hold.end < utc_now(),
-            Hold.license_pool_id == LicensePool.id,
-            LicensePool.collection_id == collection_id,
-        )
+        .where(Hold.id.in_(h.id for h in expired_holds))
         .execution_options(synchronize_session="fetch")
     )
-    result = db.execute(query)
-    # We need the type ignores here because result doesn't always have
-    # a rowcount, but the sqlalchemy docs swear it will in the case of
-    # a delete statement.
-    # https://docs.sqlalchemy.org/en/20/tutorial/data_update.html#getting-affected-row-count-from-update-delete
-    return result.rowcount  # type: ignore[attr-defined,no-any-return]
+
+    db.execute(query)
+
+    return expired_hold_events
 
 
 def licensepool_ids_with_holds(
@@ -74,7 +103,7 @@ def lock_licenses(license_pool: LicensePool) -> None:
 def recalculate_holds_for_licensepool(
     license_pool: LicensePool,
     reservation_period: datetime.timedelta,
-) -> int:
+) -> tuple[int, list[CirculationEventData]]:
     # We take out row level locks on all the licenses and holds for this license pool, so that
     # everything is in a consistent state while we update the hold queue. This means we should be
     # quickly committing the transaction, to avoid contention or deadlocks.
@@ -88,6 +117,8 @@ def recalculate_holds_for_licensepool(
     waiting = holds[reserved:]
     updated = 0
 
+    events: list[CirculationEventData] = []
+
     # These holds have a copy reserved for them.
     for hold in ready:
         # If this hold isn't already in position 0, the hold just became available.
@@ -96,6 +127,14 @@ def recalculate_holds_for_licensepool(
             hold.position = 0
             hold.end = utc_now() + reservation_period
             updated += 1
+            events.append(
+                CirculationEventData(
+                    library=hold.library,
+                    license_pool=hold.license_pool,
+                    event_type=CirculationEvent.CM_HOLD_READY_FOR_CHECKOUT,
+                    patron=hold.patron,
+                )
+            )
 
     # Update the position for the remaining holds.
     for idx, hold in enumerate(waiting):
@@ -105,13 +144,35 @@ def recalculate_holds_for_licensepool(
             hold.end = None
             updated += 1
 
-    return updated
+    return updated, events
+
+
+@shared_task(queue=QueueNames.default, bind=True)
+def remove_expired_holds_for_collection_task(task: Task, collection_id: int) -> None:
+    """
+    A shared task for removing expired holds from the database for a collection
+    """
+    analytics = task.services.analytics.analytics()
+
+    with task.transaction() as session:
+        collection = Collection.by_id(session, collection_id)
+        events = remove_expired_holds_for_collection(
+            session,
+            collection_id,
+        )
+
+        collection_name = None if not collection else collection.name
+        task.log.info(
+            f"Removed {len(events)} expired holds for collection {collection_name} ({collection_id})."
+        )
+
+    collect_events(task, events, analytics)
 
 
 @shared_task(queue=QueueNames.default, bind=True)
 def remove_expired_holds(task: Task) -> None:
     """
-    Remove expired holds from the database.
+    Issue remove expired hold tasks for eligible collections
     """
     registry = task.services.integration_registry.license_providers()
     protocols = registry.get_protocols(OPDS2WithODLApi, default=False)
@@ -122,11 +183,7 @@ def remove_expired_holds(task: Task) -> None:
             if collection.id is not None
         ]
     for collection_id, collection_name in collections:
-        with task.transaction() as session:
-            removed = remove_expired_holds_for_collection(session, collection_id)
-            task.log.info(
-                f"Removed {removed} expired holds for collection {collection_name} ({collection_id})."
-            )
+        remove_expired_holds_for_collection_task.delay(collection_id)
 
 
 @shared_task(queue=QueueNames.default, bind=True)
@@ -151,6 +208,27 @@ def _redis_lock_recalculate_holds(client: Redis, collection_id: int) -> RedisLoc
     )
 
 
+def collect_events(
+    task: Task, events: list[CirculationEventData], analytics: Analytics
+) -> None:
+    """
+    Collect events after successful database is commit and any row locks are removed.
+    We perform this operation outside after completed the transaction to ensure that any row locks
+    are held for the shortest possible duration in case writing to the s3 analytics provider is slow.
+    """
+    with task.session() as session:
+        for e in events:
+            session.refresh(e.library)
+            session.refresh(e.license_pool)
+            session.refresh(e.patron)
+            analytics.collect_event(
+                event_type=e.event_type,
+                library=e.library,
+                license_pool=e.license_pool,
+                patron=e.patron,
+            )
+
+
 @shared_task(queue=QueueNames.default, bind=True)
 def recalculate_hold_queue_collection(
     task: Task, collection_id: int, batch_size: int = 100, after_id: int | None = None
@@ -159,6 +237,7 @@ def recalculate_hold_queue_collection(
     Recalculate the hold queue for a collection.
     """
     lock = _redis_lock_recalculate_holds(task.services.redis.client(), collection_id)
+    analytics = task.services.analytics.analytics()
     with lock.lock() as locked:
         if not locked:
             task.log.info(
@@ -200,8 +279,10 @@ def recalculate_hold_queue_collection(
                         f"Skipping license pool {license_pool_id} because it no longer exists."
                     )
                     continue
-                updated = recalculate_holds_for_licensepool(
-                    license_pool, reservation_period
+
+                updated, events = recalculate_holds_for_licensepool(
+                    license_pool,
+                    reservation_period,
                 )
                 edition = license_pool.presentation_edition
                 title = edition.title if edition else None
@@ -211,6 +292,8 @@ def recalculate_hold_queue_collection(
                     f"{updated} holds out of date."
                 )
 
+            collect_events(task, events, analytics)
+
     if len(license_pool_ids) == batch_size:
         # We are done this batch, but there is probably more work to do, we queue up the next batch.
         raise task.replace(
diff --git a/src/palace/manager/core/monitor.py b/src/palace/manager/core/monitor.py
index a9022239b..444f04579 100644
--- a/src/palace/manager/core/monitor.py
+++ b/src/palace/manager/core/monitor.py
@@ -885,10 +885,16 @@ def run_once(self, *args, **kwargs):
                 self.delete(i)
                 rows_deleted += 1
             self._db.commit()
+
+            self.after_commit()
+
             count = qu.count()
         return TimestampData(achievements="Items deleted: %d" % rows_deleted)
 
-    def delete(self, row):
+    def after_commit(self) -> None:
+        return None
+
+    def delete(self, row) -> None:
         """Delete a row from the database.
 
         CAUTION: If you override this method such that it doesn't
diff --git a/src/palace/manager/sqlalchemy/model/circulationevent.py b/src/palace/manager/sqlalchemy/model/circulationevent.py
index 91b9d5eed..c2c1c33f9 100644
--- a/src/palace/manager/sqlalchemy/model/circulationevent.py
+++ b/src/palace/manager/sqlalchemy/model/circulationevent.py
@@ -99,6 +99,11 @@ class CirculationEvent(Base):
     CM_CHECKIN = "circulation_manager_check_in"
     CM_HOLD_PLACE = "circulation_manager_hold_place"
     CM_HOLD_RELEASE = "circulation_manager_hold_release"
+    CM_HOLD_EXPIRED = "circulation_manager_hold_expired"
+    CM_HOLD_READY_FOR_CHECKOUT = "circulation_manager_hold_ready"
+    CM_LOAN_EXPIRED = "circulation_manager_loan_expired"
+    CM_HOLD_CONVERTED_TO_LOAN = "circulation_manager_hold_converted_to_loan"
+    CM_LOAN_CONVERTED_TO_HOLD = "circulation_manager_loan_converted_to_hold"
     CM_FULFILL = "circulation_manager_fulfill"
 
     # Events that we hear about from a distributor.
diff --git a/tests/manager/api/test_authenticator.py b/tests/manager/api/test_authenticator.py
index 69b2ea3b2..2f892d83a 100644
--- a/tests/manager/api/test_authenticator.py
+++ b/tests/manager/api/test_authenticator.py
@@ -404,7 +404,7 @@ def test_apply_on_incomplete_information(self, db: DatabaseTransactionFixture):
         )
         authenticated_by_weird_identifier.apply(patron)
         assert "1234" == patron.authorization_identifier
-        assert None == patron.last_external_sync
+        assert patron.last_external_sync is None
 
     def test_get_or_create_patron(
         self, patron_data: PatronData, db: DatabaseTransactionFixture
@@ -417,15 +417,15 @@ def test_get_or_create_patron(
         patron, is_new = patron_data.get_or_create_patron(
             db.session, default_library.id, analytics
         )
-        assert "2" == patron.authorization_identifier
+        assert patron.authorization_identifier == "2"
         assert default_library == patron.library
-        assert True == is_new
-        assert CirculationEvent.NEW_PATRON == analytics.event_type
-        assert 1 == analytics.count
+        assert is_new is True
+        assert analytics.last_event_type == CirculationEvent.NEW_PATRON
+        assert analytics.count == 1
 
         # Patron.neighborhood was set, even though there is no
         # value and that's not a database field.
-        assert None == patron.neighborhood
+        assert patron.neighborhood is None
 
         # Set a neighborhood and try again.
         patron_data.neighborhood = "Achewood"
@@ -435,15 +435,14 @@ def test_get_or_create_patron(
         patron, is_new = patron_data.get_or_create_patron(
             db.session, default_library.id, analytics
         )
-        assert "2" == patron.authorization_identifier
-        assert False == is_new
-        assert "Achewood" == patron.neighborhood
-        assert 1 == analytics.count
+        assert patron.authorization_identifier == "2"
+        assert is_new is False
+        assert patron.neighborhood == "Achewood"
+        assert analytics.count == 1
 
     def test_to_response_parameters(self, patron_data: PatronData):
         params = patron_data.to_response_parameters
         assert dict(name="4") == params
-
         patron_data.personal_name = None
         params = patron_data.to_response_parameters
         assert dict() == params
diff --git a/tests/manager/api/test_circulationapi.py b/tests/manager/api/test_circulationapi.py
index 27042a399..5dd707a36 100644
--- a/tests/manager/api/test_circulationapi.py
+++ b/tests/manager/api/test_circulationapi.py
@@ -129,18 +129,18 @@ def test_borrow_sends_analytics_event(self, circulation_api: CirculationAPIFixtu
         # The Loan looks good.
         assert loaninfo.identifier == loan.license_pool.identifier.identifier
         assert circulation_api.patron == loan.patron
-        assert None == hold
-        assert True == is_new
+        assert hold is None
+        assert is_new == True
         assert loaninfo.external_identifier == loan.external_identifier
 
         # An analytics event was created.
-        assert 1 == circulation_api.analytics.count
-        assert CirculationEvent.CM_CHECKOUT == circulation_api.analytics.event_type
+        assert circulation_api.analytics.count == 1
+        assert circulation_api.analytics.last_event_type == CirculationEvent.CM_CHECKOUT
 
         # Try to 'borrow' the same book again.
         circulation_api.remote.queue_checkout(AlreadyCheckedOut())
         loan, hold, is_new = self.borrow(circulation_api)
-        assert False == is_new
+        assert is_new == False
         assert loaninfo.external_identifier == loan.external_identifier
 
         # Since the loan already existed, no new analytics event was
@@ -150,18 +150,18 @@ def test_borrow_sends_analytics_event(self, circulation_api: CirculationAPIFixtu
         # Now try to renew the book.
         circulation_api.remote.queue_checkout(loaninfo)
         loan, hold, is_new = self.borrow(circulation_api)
-        assert False == is_new
+        assert is_new == False
 
         # Renewals are counted as loans, since from an accounting
         # perspective they _are_ loans.
-        assert 2 == circulation_api.analytics.count
+        assert circulation_api.analytics.count == 2
 
         # Loans of open-access books go through a different code
         # path, but they count as loans nonetheless.
         circulation_api.pool.open_access = True
         circulation_api.remote.queue_checkout(loaninfo)
         loan, hold, is_new = self.borrow(circulation_api)
-        assert 3 == circulation_api.analytics.count
+        assert circulation_api.analytics.count == 3
 
     @freeze_time()
     def test_attempt_borrow_with_existing_remote_loan(
@@ -357,21 +357,51 @@ def test_hold_sends_analytics_event(self, circulation_api: CirculationAPIFixture
         # The Hold looks good.
         assert holdinfo.identifier == hold.license_pool.identifier.identifier
         assert circulation_api.patron == hold.patron
-        assert None == loan
-        assert True == is_new
+        assert loan is None
+        assert is_new == True
 
         # An analytics event was created.
         assert 1 == circulation_api.analytics.count
-        assert CirculationEvent.CM_HOLD_PLACE == circulation_api.analytics.event_type
+        assert (
+            circulation_api.analytics.last_event_type == CirculationEvent.CM_HOLD_PLACE
+        )
 
         # Try to 'borrow' the same book again.
         circulation_api.remote.queue_checkout(AlreadyOnHold())
         loan, hold, is_new = self.borrow(circulation_api)
-        assert False == is_new
+        assert is_new == False
 
         # Since the hold already existed, no new analytics event was
         # sent.
-        assert 1 == circulation_api.analytics.count
+        assert circulation_api.analytics.count == 1
+
+    def test_hold_is_ready_converts_to_loan_on_borrow(
+        self, circulation_api: CirculationAPIFixture
+    ):
+        now = utc_now()
+        loaninfo = LoanInfo.from_license_pool(
+            circulation_api.pool,
+            start_date=now,
+            end_date=now + timedelta(seconds=3600),
+            external_identifier=circulation_api.db.fresh_str(),
+        )
+        circulation_api.remote.queue_checkout(loaninfo)
+        circulation_api.pool.on_hold_to(patron=circulation_api.patron, position=0)
+        loan, hold, is_new = self.borrow(circulation_api)
+
+        # The Hold is gone and there is a new loan.
+        assert loan is not None
+        assert hold is None
+        assert is_new is True
+
+        assert circulation_api.analytics.count == 2
+        # A hold converted analytics event was recorded
+        assert (
+            circulation_api.analytics.event_types[0]
+            == CirculationEvent.CM_HOLD_CONVERTED_TO_LOAN
+        )
+        # A check event was recorded
+        assert circulation_api.analytics.event_types[1] == CirculationEvent.CM_CHECKOUT
 
     def test_borrow_with_expired_card_fails(
         self, circulation_api: CirculationAPIFixture
@@ -808,11 +838,11 @@ def test_fulfill(self, circulation_api: CirculationAPIFixture):
         )
 
         # The fulfillment looks good.
-        assert fulfillment == result
+        assert result == fulfillment
 
         # An analytics event was created.
-        assert 1 == circulation_api.analytics.count
-        assert CirculationEvent.CM_FULFILL == circulation_api.analytics.event_type
+        assert circulation_api.analytics.count == 1
+        assert circulation_api.analytics.last_event_type == CirculationEvent.CM_FULFILL
 
     def test_fulfill_without_loan(self, circulation_api: CirculationAPIFixture):
         # By default, a title cannot be fulfilled unless there is an active
@@ -852,11 +882,11 @@ def test_revoke_loan(self, circulation_api: CirculationAPIFixture, open_access):
         result = circulation_api.circulation.revoke_loan(
             circulation_api.patron, "1234", circulation_api.pool
         )
-        assert True == result
+        assert result == True
 
         # An analytics event was created.
-        assert 1 == circulation_api.analytics.count
-        assert CirculationEvent.CM_CHECKIN == circulation_api.analytics.event_type
+        assert circulation_api.analytics.count == 1
+        assert circulation_api.analytics.last_event_type == CirculationEvent.CM_CHECKIN
 
     @pytest.mark.parametrize("open_access", [True, False])
     def test_release_hold(self, circulation_api: CirculationAPIFixture, open_access):
@@ -868,11 +898,14 @@ def test_release_hold(self, circulation_api: CirculationAPIFixture, open_access)
         result = circulation_api.circulation.release_hold(
             circulation_api.patron, "1234", circulation_api.pool
         )
-        assert True == result
+        assert result == True
 
         # An analytics event was created.
-        assert 1 == circulation_api.analytics.count
-        assert CirculationEvent.CM_HOLD_RELEASE == circulation_api.analytics.event_type
+        assert circulation_api.analytics.count == 1
+        assert (
+            circulation_api.analytics.last_event_type
+            == CirculationEvent.CM_HOLD_RELEASE
+        )
 
     def test__collect_event(self, circulation_api: CirculationAPIFixture):
         # Test the _collect_event method, which gathers information
diff --git a/tests/manager/api/test_monitor.py b/tests/manager/api/test_monitor.py
index ba4086759..3ecbec9a8 100644
--- a/tests/manager/api/test_monitor.py
+++ b/tests/manager/api/test_monitor.py
@@ -8,11 +8,13 @@
     LoanReaper,
 )
 from palace.manager.api.opds_for_distributors import OPDSForDistributorsAPI
+from palace.manager.sqlalchemy.model.circulationevent import CirculationEvent
 from palace.manager.sqlalchemy.model.datasource import DataSource
 from palace.manager.sqlalchemy.model.patron import Annotation
 from palace.manager.sqlalchemy.util import get_one_or_create
 from palace.manager.util.datetime_helpers import utc_now
 from tests.fixtures.database import DatabaseTransactionFixture
+from tests.fixtures.services import ServicesFixture
 
 
 class TestLoanlikeReaperMonitor:
@@ -26,7 +28,9 @@ def test_source_of_truth_protocols(self):
             OPDSForDistributorsAPI.label()
         ]
 
-    def test_reaping(self, db: DatabaseTransactionFixture):
+    def test_reaping(
+        self, db: DatabaseTransactionFixture, services_fixture: ServicesFixture
+    ):
         # This patron stopped using the circulation manager a long time
         # ago.
         inactive_patron = db.patron()
@@ -152,6 +156,7 @@ def test_reaping(self, db: DatabaseTransactionFixture):
 
         # Now we fire up the loan reaper.
         monitor = LoanReaper(db.session)
+        monitor.services.analytics = services_fixture.analytics_fixture.analytics_mock
         monitor.run()
 
         # All of the inactive patron's loans have been reaped,
@@ -163,16 +168,19 @@ def test_reaping(self, db: DatabaseTransactionFixture):
         assert {open_access_loan, sot_loan, unlimited_access_loan} == set(
             inactive_patron.loans
         )
-        assert 3 == len(inactive_patron.holds)
+        assert len(inactive_patron.holds) == 3
 
         # The active patron's loans and holds are unaffected, either
         # because they have not expired or because they have no known
         # expiration date and were created relatively recently.
-        assert 2 == len(current_patron.loans)
-        assert 2 == len(current_patron.holds)
+        assert len(current_patron.loans) == 2
+        assert len(current_patron.holds) == 2
 
         # Now fire up the hold reaper.
         hold_monitor = HoldReaper(db.session)
+        hold_monitor.services.analytics = (
+            services_fixture.analytics_fixture.analytics_mock
+        )
         hold_monitor.run()
 
         # All of the inactive patron's holds have been reaped,
@@ -181,6 +189,19 @@ def test_reaping(self, db: DatabaseTransactionFixture):
         assert [sot_hold] == inactive_patron.holds
         assert 2 == len(current_patron.holds)
 
+        # verify expected circ event count and order for two monitor operations.
+        call_args_list = (
+            services_fixture.analytics_fixture.analytics_mock.collect_event.call_args_list
+        )
+        assert len(call_args_list) == 4
+        event_types = [call_args.kwargs["event_type"] for call_args in call_args_list]
+        assert event_types == [
+            CirculationEvent.CM_LOAN_EXPIRED,
+            CirculationEvent.CM_LOAN_EXPIRED,
+            CirculationEvent.CM_HOLD_EXPIRED,
+            CirculationEvent.CM_HOLD_EXPIRED,
+        ]
+
 
 class TestIdlingAnnotationReaper:
     def test_where_clause(self, db: DatabaseTransactionFixture):
diff --git a/tests/manager/celery/tasks/test_opds_odl.py b/tests/manager/celery/tasks/test_opds_odl.py
index 213802ed1..829a1677d 100644
--- a/tests/manager/celery/tasks/test_opds_odl.py
+++ b/tests/manager/celery/tasks/test_opds_odl.py
@@ -16,8 +16,10 @@
     recalculate_holds_for_licensepool,
     remove_expired_holds,
     remove_expired_holds_for_collection,
+    remove_expired_holds_for_collection_task,
 )
 from palace.manager.service.logging.configuration import LogLevel
+from palace.manager.sqlalchemy.model.circulationevent import CirculationEvent
 from palace.manager.sqlalchemy.model.collection import Collection
 from palace.manager.sqlalchemy.model.licensing import License, LicensePool
 from palace.manager.sqlalchemy.model.patron import Hold, Patron
@@ -26,11 +28,13 @@
 from tests.fixtures.celery import CeleryFixture
 from tests.fixtures.database import DatabaseTransactionFixture
 from tests.fixtures.redis import RedisFixture
+from tests.fixtures.services import ServicesFixture
 
 
 class OpdsTaskFixture:
-    def __init__(self, db: DatabaseTransactionFixture):
+    def __init__(self, db: DatabaseTransactionFixture, services: ServicesFixture):
         self.db = db
+        self.services = services
 
         self.two_weeks_ago = utc_now() - timedelta(weeks=2)
         self.yesterday = utc_now() - timedelta(days=1)
@@ -117,8 +121,10 @@ def pool_with_licenses(
 
 
 @pytest.fixture
-def opds_task_fixture(db: DatabaseTransactionFixture) -> OpdsTaskFixture:
-    return OpdsTaskFixture(db)
+def opds_task_fixture(
+    db: DatabaseTransactionFixture, services_fixture: ServicesFixture
+) -> OpdsTaskFixture:
+    return OpdsTaskFixture(db, services_fixture)
 
 
 def _hold_sort_key(hold: Hold) -> int:
@@ -128,7 +134,9 @@ def _hold_sort_key(hold: Hold) -> int:
 
 
 def test_remove_expired_holds_for_collection(
-    db: DatabaseTransactionFixture, opds_task_fixture: OpdsTaskFixture
+    db: DatabaseTransactionFixture,
+    opds_task_fixture: OpdsTaskFixture,
+    celery_fixture: CeleryFixture,
 ):
     collection = db.collection(protocol=OPDS2WithODLApi)
     decoy_collection = db.collection(protocol=OverdriveAPI)
@@ -144,7 +152,10 @@ def test_remove_expired_holds_for_collection(
 
     # Remove the expired holds
     assert collection.id is not None
-    removed = remove_expired_holds_for_collection(db.session, collection.id)
+    events = remove_expired_holds_for_collection(
+        db.session,
+        collection.id,
+    )
 
     # Assert that the correct holds were removed
     current_holds = {h.id for h in db.session.scalars(select(Hold))}
@@ -154,8 +165,6 @@ def test_remove_expired_holds_for_collection(
     assert decoy_non_expired_holds.issubset(current_holds)
     assert decoy_expired_holds.issubset(current_holds)
 
-    assert removed == 10
-
     pools_after = db.session.scalars(
         select(func.count()).select_from(LicensePool)
     ).one()
@@ -163,6 +172,12 @@ def test_remove_expired_holds_for_collection(
     # Make sure the license pools for those holds were not deleted
     assert pools_before == pools_after
 
+    # verify that the correct analytics calls were made
+    assert len(events) == 10
+    for event in events:
+        assert event.event_type == CirculationEvent.CM_HOLD_EXPIRED
+        assert event.library == db.default_library()
+
 
 def test_licensepools_with_holds(
     db: DatabaseTransactionFixture, opds_task_fixture: OpdsTaskFixture
@@ -209,6 +224,7 @@ def test_recalculate_holds_for_licensepool(
     collection = db.collection(protocol=OPDS2WithODLApi)
     pool, [license1, license2] = opds_task_fixture.pool_with_licenses(collection)
 
+    analytics = opds_task_fixture.services.analytics_fixture.analytics_mock
     # Recalculate the hold queue
     recalculate_holds_for_licensepool(pool, timedelta(days=5))
 
@@ -221,7 +237,7 @@ def test_recalculate_holds_for_licensepool(
     license1.checkouts_available = 1
     license2.checkouts_available = 2
     reservation_time = timedelta(days=5)
-    recalculate_holds_for_licensepool(pool, reservation_time)
+    _, events = recalculate_holds_for_licensepool(pool, reservation_time)
 
     assert pool.licenses_reserved == 3
     assert pool.licenses_available == 0
@@ -250,33 +266,53 @@ def test_recalculate_holds_for_licensepool(
         )
         assert hold.start and expected_start and hold.start >= expected_start
 
+    # verify that the correct analytics events were returned
+    assert len(events) == 3
+    for event in events:
+        assert event.event_type == CirculationEvent.CM_HOLD_READY_FOR_CHECKOUT
 
-def test_remove_expired_holds(
+
+def test_remove_expired_holds_for_collection_task(
     celery_fixture: CeleryFixture,
     db: DatabaseTransactionFixture,
     opds_task_fixture: OpdsTaskFixture,
 ):
     collection1 = db.collection(protocol=OPDS2WithODLApi)
-    collection2 = db.collection(protocol=OPDS2WithODLApi)
-    decoy_collection = db.collection(protocol=OverdriveAPI)
 
     expired_holds1, non_expired_holds1 = opds_task_fixture.holds(collection1)
-    expired_holds2, non_expired_holds2 = opds_task_fixture.holds(collection2)
-    decoy_expired_holds, decoy_non_expired_holds = opds_task_fixture.holds(
-        decoy_collection
-    )
 
     # Remove the expired holds
-    remove_expired_holds.delay().wait()
+    remove_expired_holds_for_collection_task.delay(collection1.id).wait()
+
+    assert len(
+        opds_task_fixture.services.analytics_fixture.analytics_mock.method_calls
+    ) == len(expired_holds1)
 
     current_holds = {h.id for h in db.session.scalars(select(Hold))}
     assert expired_holds1.isdisjoint(current_holds)
-    assert expired_holds2.isdisjoint(current_holds)
 
-    assert decoy_non_expired_holds.issubset(current_holds)
-    assert decoy_expired_holds.issubset(current_holds)
     assert non_expired_holds1.issubset(current_holds)
-    assert non_expired_holds2.issubset(current_holds)
+
+
+def test_remove_expired_holds(
+    celery_fixture: CeleryFixture,
+    redis_fixture: RedisFixture,
+    db: DatabaseTransactionFixture,
+    opds_task_fixture: OpdsTaskFixture,
+):
+    collection1 = db.collection(protocol=OPDS2WithODLApi)
+    collection2 = db.collection(protocol=OPDS2WithODLApi)
+    decoy_collection = db.collection(protocol=OverdriveAPI)
+
+    with patch.object(
+        opds_odl, "remove_expired_holds_for_collection_task"
+    ) as mock_remove:
+        remove_expired_holds.delay().wait()
+
+    assert mock_remove.delay.call_count == 2
+    mock_remove.delay.assert_has_calls(
+        [call(collection1.id), call(collection2.id)], any_order=True
+    )
 
 
 def test_recalculate_hold_queue(
diff --git a/tests/mocks/analytics_provider.py b/tests/mocks/analytics_provider.py
index e2805565e..c9b821379 100644
--- a/tests/mocks/analytics_provider.py
+++ b/tests/mocks/analytics_provider.py
@@ -13,10 +13,13 @@ def __init__(self, integration=None, services=None, library=None):
         """
         self.count = 0
         self.event = None
+        self.event_types = []
+        self.last_event_type = None
         if integration:
             self.url = integration.url
 
     def collect_event(self, library, lp, event_type, time=None, **kwargs):
         self.count = self.count + 1
-        self.event_type = event_type
+        self.last_event_type = event_type
+        self.event_types.append(event_type)
         self.time = time