Add task for catching missing sha256 checksums

Author: jjnesbitt

dandiapi/api/tasks/__init__.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
 from celery import shared_task
 from celery.utils.log import get_task_logger
 
@@ -13,6 +15,9 @@
 )
 from dandiapi.api.models import Asset, AssetBlob, Version
 
+if TYPE_CHECKING:
+    from uuid import UUID
+
 logger = get_task_logger(__name__)
 
 
@@ -25,9 +30,9 @@ def remove_asset_blob_embargoed_tag_task(blob_id: str) -> None:
 
 
 @shared_task(queue='calculate_sha256', soft_time_limit=86_400)
-def calculate_sha256(blob_id: str) -> None:
+def calculate_sha256(blob_id: str | UUID) -> None:
     asset_blob = AssetBlob.objects.get(blob_id=blob_id)
-    logger.info('Found AssetBlob %s', blob_id)
+    logger.info('Calculating sha256 checksum for asset blob %s', blob_id)
     sha256 = asset_blob.blob.storage.sha256_checksum(asset_blob.blob.name)
 
     # TODO: Run dandi-cli validation

dandiapi/api/tasks/scheduled.py

@@ -22,11 +22,12 @@
 from dandiapi.analytics.tasks import collect_s3_log_records_task
 from dandiapi.api.mail import send_dandisets_to_unembargo_message, send_pending_users_message
 from dandiapi.api.models import UserMetadata, Version
-from dandiapi.api.models.asset import Asset
+from dandiapi.api.models.asset import Asset, AssetBlob
 from dandiapi.api.models.dandiset import Dandiset
 from dandiapi.api.services.metadata import version_aggregate_assets_summary
 from dandiapi.api.services.metadata.exceptions import VersionMetadataConcurrentlyModifiedError
 from dandiapi.api.tasks import (
+    calculate_sha256,
     validate_asset_metadata_task,
     validate_version_metadata_task,
     write_manifest_files,
@@ -63,6 +64,19 @@ def aggregate_assets_summary_task(version_id: int):
     version_aggregate_assets_summary(version)
 
 
+@shared_task(soft_time_limit=30)
+def recalculate_missing_sha256_checksums():
+    blobs_missing = AssetBlob.objects.filter(sha256__isnull=True).values_list('blob_id', flat=True)
+    num_missing = blobs_missing.count()
+    if num_missing == 0:
+        logger.info('No blobs with missing checksums found')
+        return
+
+    logger.info('Found %s blobs with missing checksums', num_missing)
+    for blob_id in blobs_missing:
+        calculate_sha256.delay(blob_id)
+
+
 @shared_task(soft_time_limit=30)
 def validate_pending_asset_metadata():
     validatable_assets = (
@@ -156,3 +170,6 @@ def register_scheduled_tasks(sender: Celery, **kwargs):
 
     # Process new S3 logs every hour
     sender.add_periodic_task(timedelta(hours=1), collect_s3_log_records_task.s())
+
+    # Check for asset blobs with missing sha256 checksums once a day
+    sender.add_periodic_task(crontab(hour=3, minute=0), recalculate_missing_sha256_checksums.s())