fix: handle S3 partial delete failures and pass se_name directly

s3_bulk_delete_with_retry now returns failed keys instead of raising,
and _s3_delete_chunk_with_retry retries only the failed keys. The
caller skips DB deletion for sandboxes whose S3 objects were not
removed, preventing dark data.

Also removes the per-batch SELECT DISTINCT SEName query by passing
settings.se_name directly, and updates the stale docstring.

Author: chrisburr

diracx-core/src/diracx/core/s3.py

@@ -89,34 +89,60 @@ def b16_to_b64(hex_string: str) -> str:
 
 async def s3_bulk_delete_with_retry(
     s3_client, bucket: str, objects: list[S3Object]
-) -> None:
-    # S3 delete_objects supports at most 1000 keys per call.
-    # Chunks are sent concurrently for throughput.
+) -> set[str]:
+    """Delete objects from S3 in chunks of 1000, retrying failures.
+
+    Returns:
+        Set of keys that failed to delete after all retries.
+
+    """
     max_chunk_size = 1000
     chunks = [
         objects[i : i + max_chunk_size] for i in range(0, len(objects), max_chunk_size)
     ]
-    async with asyncio.TaskGroup() as tg:
-        for chunk in chunks:
-            tg.create_task(_s3_delete_chunk_with_retry(s3_client, bucket, chunk))
+    tasks = [_s3_delete_chunk_with_retry(s3_client, bucket, chunk) for chunk in chunks]
+    results = await asyncio.gather(*tasks)
+    failed_keys: set[str] = set()
+    for result in results:
+        failed_keys.update(result)
+    return failed_keys
 
 
 async def _s3_delete_chunk_with_retry(
     s3_client, bucket: str, objects: list[S3Object]
-) -> None:
+) -> set[str]:
+    """Try to delete a chunk of S3 objects, retrying partial failures.
+
+    Returns:
+        Set of keys that failed to delete after all retries.
+
+    """
     max_attempts = 5
     delay = 1.0
+    remaining = objects
     for attempt in range(1, max_attempts + 1):
         try:
             response = await s3_client.delete_objects(
                 Bucket=bucket,
-                Delete={"Objects": objects, "Quiet": True},
+                Delete={"Objects": remaining, "Quiet": True},
             )
-            if "Errors" in response and response["Errors"]:
-                raise RuntimeError(f"S3 deletion error: {response['Errors']}")
-            return
-        except (ClientError, RuntimeError) as e:
+        except ClientError:
             if attempt == max_attempts:
-                raise RuntimeError(f"Failed to delete objects in {bucket}") from e
+                return {obj["Key"] for obj in remaining}
             await asyncio.sleep(delay)
             delay *= 2
+            continue
+
+        errors = response.get("Errors", [])
+        if not errors:
+            return set()
+
+        # Retry only the keys that failed
+        failed_keys = {e["Key"] for e in errors}
+        if attempt == max_attempts:
+            return failed_keys
+        remaining = [obj for obj in remaining if obj["Key"] in failed_keys]
+        await asyncio.sleep(delay)
+        delay *= 2
+
+    return set()

diracx-db/src/diracx/db/sql/sandbox_metadata/db.py

@@ -210,19 +210,21 @@ async def unassign_sandboxes_to_jobs(self, jobs_ids: list[int]) -> None:
     async def select_sandboxes_for_deletion(
         self,
         *,
+        se_name: str,
         batch_size: int = 500,
         cursor: int = 0,
     ) -> tuple[list[int], list[str], int]:
-        """Select a batch of sandboxes for deletion.
+        """Select a batch of sandboxes eligible for deletion.
 
-        Uses cursor-based pagination (SBId > cursor) to avoid re-scanning
-        rows that were already checked in previous batches.
+        Uses cursor-based pagination (SBId > cursor) and runs two queries:
+        - Orphaned: assigned but no entity mapping (NOT EXISTS)
+        - Unassigned: not assigned and older than 15 days
 
-        Runs two separate queries so MySQL can use indexes effectively:
-        - Orphaned assigned sandboxes (NOT EXISTS on entity mapping)
-        - Unassigned sandboxes older than 15 days (sargable comparison)
+        No row locking is used — the caller is responsible for crash safety
+        by deleting from S3 before removing DB rows.
 
         Args:
+            se_name: Storage element name to filter on.
             batch_size: Maximum number of sandboxes to select.
             cursor: Only consider sandboxes with SBId > cursor.
 
@@ -231,15 +233,8 @@ async def select_sandboxes_for_deletion(
             new_cursor is the maximum SBId seen, to pass as cursor next time.
 
         """
-        # Fetch distinct SEName values so the queries can use the Location
-        # unique key (SEName, SEPFN) instead of doing a full table scan.
-        se_names_result = await self.conn.execute(select(SandBoxes.SEName).distinct())
-        se_names = [row.SEName for row in se_names_result]
-        if not se_names:
-            return [], [], cursor
-
         s3_condition = and_(
-            SandBoxes.SEName.in_(se_names),
+            SandBoxes.SEName == se_name,
             SandBoxes.SEPFN.like("/S3/%"),
         )
 

diracx-logic/src/diracx/logic/jobs/sandboxes.py

@@ -234,6 +234,7 @@ async def clean_sandboxes(
                 pfns,
                 cursor,
             ) = await sandbox_metadata_db.select_sandboxes_for_deletion(
+                se_name=settings.se_name,
                 batch_size=batch_size,
                 cursor=cursor,
             )
@@ -256,14 +257,39 @@ async def clean_sandboxes(
         # since S3 is cleaned first)
         t0 = time.monotonic()
         objects: list[S3Object] = [{"Key": pfn_to_key(pfn)} for pfn in pfns]
-        await s3_bulk_delete_with_retry(
+        failed_keys = await s3_bulk_delete_with_retry(
             settings.s3_client, settings.bucket_name, objects
         )
         s3_duration = time.monotonic() - t0
+
+        if failed_keys:
+            # Only delete DB rows for sandboxes whose S3 objects were
+            # actually removed — the rest will be retried next run.
+            logger.warning(
+                "Batch %d: %d S3 deletions failed, skipping their DB rows",
+                batch_num,
+                len(failed_keys),
+            )
+            filtered = [
+                (sid, pfn)
+                for sid, pfn in zip(sb_ids, pfns)
+                if pfn_to_key(pfn) not in failed_keys
+            ]
+            if filtered:
+                sb_ids, pfns = map(list, zip(*filtered))
+            else:
+                sb_ids, pfns = [], []
+
         logger.info(
-            "Batch %d: deleted %d from S3 (%.1fs)", batch_num, len(objects), s3_duration
+            "Batch %d: deleted %d from S3 (%.1fs)",
+            batch_num,
+            len(objects) - len(failed_keys),
+            s3_duration,
         )
 
+        if not sb_ids:
+            continue
+
         # Phase 3: Delete from DB in small chunks (each chunk is a short
         # transaction to avoid locking millions of rows in a single DELETE).
         # Up to 10 chunks run concurrently via a semaphore.