fix: only emit CleanupStaleItemsCompletedEvent if queue item was deleted (#454)

src/deriver/consumer.py

@@ -356,16 +356,16 @@ async def process_reconciler(payload: ReconcilerPayload) -> None:
 
     elif reconciler_type == ReconcilerType.CLEANUP_QUEUE:
         logger.debug("Processing cleanup_queue task")
-        await cleanup_queue_items()
+        deleted_count = await cleanup_queue_items()
 
         duration_ms = (time.perf_counter() - start_time) * 1000
 
-        # Emit telemetry event for cleanup stale items
-        emit(
-            CleanupStaleItemsCompletedEvent(
-                total_duration_ms=duration_ms,
+        if deleted_count > 0:
+            # Emit telemetry event for cleanup stale items
+            emit(
+                CleanupStaleItemsCompletedEvent(
+                    total_duration_ms=duration_ms,
+                )
             )
-        )
-
     else:
         raise ValueError(f"Unsupported reconciler type: {reconciler_type}")

src/reconciler/queue_cleanup.py

@@ -6,8 +6,9 @@ This module provides a periodic cleanup job that removes old processed queue ite
 
 import logging
 from datetime import datetime, timedelta, timezone
+from typing import Any, cast
 
-from sqlalchemy import delete
+from sqlalchemy import CursorResult, delete
 
 from src import models
 from src.config import settings
@@ -16,12 +17,15 @@ from src.dependencies import tracked_db
 logger = logging.getLogger(__name__)
 
 
-async def cleanup_queue_items() -> None:
+async def cleanup_queue_items() -> int:
     """
     Delete processed queue items.
 
     Successfully processed queue items are deleted immediately,
     while errored queue items are deleted after retention window.
+
+    Returns:
+        The number of queue items deleted.
     """
     async with tracked_db("cleanup_queue_items") as db:
         now = datetime.now(timezone.utc)
@@ -29,17 +33,22 @@ async def cleanup_queue_items() -> None:
             seconds=settings.DERIVER.QUEUE_ERROR_RETENTION_SECONDS
         )
 
-        await db.execute(
-            delete(models.QueueItem).where(
-                models.QueueItem.processed
-                & (
-                    models.QueueItem.error.is_(None)
-                    | (
-                        models.QueueItem.error.is_not(None)
-                        & (models.QueueItem.created_at < error_cutoff)
+        result = cast(
+            CursorResult[Any],
+            await db.execute(
+                delete(models.QueueItem).where(
+                    models.QueueItem.processed
+                    & (
+                        models.QueueItem.error.is_(None)
+                        | (
+                            models.QueueItem.error.is_not(None)
+                            & (models.QueueItem.created_at < error_cutoff)
+                        )
                     )
                 )
-            )
+            ),
         )
         await db.commit()
-        logger.info("Queue cleanup completed")
+        deleted_count = result.rowcount
+        logger.info("Queue cleanup completed, deleted %d items", deleted_count)
+        return deleted_count