feat(launchpad): ensure artifact assembly task retries in case of kafka error (#94728)

The retry behavior:
- If the Kafka producer raises a KafkaException, the task will
automatically retry
- It will retry up to 3 times with exponential backoff
- (Idempotency) Since the task checks for existing artifacts at the
beginning, re-running won't create duplicates
- The artifact creation happens in a transaction, so partial failures
are rolled back

Author: NicoHinderling

src/sentry/preprod/producer.py

@@ -53,3 +53,4 @@ def produce_preprod_artifact_to_kafka(
             "Failed to send preprod artifact message to Kafka",
             extra={"artifact_id": artifact_id, "project_id": project_id},
         )
+        raise  # Re-raise to trigger task retry

src/sentry/preprod/tasks.py

@@ -23,6 +23,7 @@
 from sentry.tasks.base import instrumented_task
 from sentry.taskworker.config import TaskworkerConfig
 from sentry.taskworker.namespaces import attachments_tasks
+from sentry.taskworker.retry import Retry
 from sentry.utils.sdk import bind_organization_context
 
 logger = logging.getLogger(__name__)
@@ -32,6 +33,7 @@
     name="sentry.preprod.tasks.assemble_preprod_artifact",
     queue="assemble",
     silo_mode=SiloMode.REGION,
+    retry=Retry(times=3),
     taskworker_config=TaskworkerConfig(
         namespace=attachments_tasks,
         processing_deadline_duration=30,
@@ -62,6 +64,7 @@ def assemble_preprod_artifact(
         },
     )
 
+    preprod_artifact = None
     try:
         organization = Organization.objects.get_from_cache(pk=org_id)
         project = Project.objects.get(id=project_id, organization=organization)
@@ -81,67 +84,64 @@ def assemble_preprod_artifact(
 
             if existing_artifact:
                 logger.info(
-                    "PreprodArtifact already exists for this checksum, skipping assembly",
+                    "PreprodArtifact already exists for this checksum, using existing artifact",
                     extra={
                         "preprod_artifact_id": existing_artifact.id,
                         "project_id": project_id,
                         "organization_id": org_id,
                         "checksum": checksum,
                     },
                 )
+                preprod_artifact = existing_artifact
+            else:
                 set_assemble_status(
-                    AssembleTask.PREPROD_ARTIFACT, project_id, checksum, ChunkFileState.OK
-                )
-                return
-
-            set_assemble_status(
-                AssembleTask.PREPROD_ARTIFACT, project_id, checksum, ChunkFileState.ASSEMBLING
-            )
-
-            assemble_result = assemble_file(
-                task=AssembleTask.PREPROD_ARTIFACT,
-                org_or_project=project,
-                name=f"preprod-artifact-{uuid.uuid4().hex}",
-                checksum=checksum,
-                chunks=chunks,
-                file_type="preprod.artifact",
-            )
-
-            if assemble_result is None:
-                return
-
-            build_config = None
-            if build_configuration:
-                build_config, _ = PreprodBuildConfiguration.objects.get_or_create(
-                    project=project,
-                    name=build_configuration,
+                    AssembleTask.PREPROD_ARTIFACT, project_id, checksum, ChunkFileState.ASSEMBLING
                 )
 
-            # Create PreprodArtifact record
-            preprod_artifact = PreprodArtifact.objects.create(
-                project=project,
-                file_id=assemble_result.bundle.id,
-                build_configuration=build_config,
-                state=PreprodArtifact.ArtifactState.UPLOADED,
-            )
+                assemble_result = assemble_file(
+                    task=AssembleTask.PREPROD_ARTIFACT,
+                    org_or_project=project,
+                    name=f"preprod-artifact-{uuid.uuid4().hex}",
+                    checksum=checksum,
+                    chunks=chunks,
+                    file_type="preprod.artifact",
+                )
 
-        logger.info(
-            "Created preprod artifact",
-            extra={
-                "preprod_artifact_id": preprod_artifact.id,
-                "project_id": project_id,
-                "organization_id": org_id,
-                "checksum": checksum,
-            },
-        )
+                if assemble_result is None:
+                    logger.warning(
+                        "Assemble result is None, returning early",
+                        extra={
+                            "project_id": project_id,
+                            "organization_id": org_id,
+                            "checksum": checksum,
+                        },
+                    )
+                    return
+
+                build_config = None
+                if build_configuration:
+                    build_config, _ = PreprodBuildConfiguration.objects.get_or_create(
+                        project=project,
+                        name=build_configuration,
+                    )
+
+                # Create PreprodArtifact record
+                preprod_artifact, _ = PreprodArtifact.objects.get_or_create(
+                    project=project,
+                    file_id=assemble_result.bundle.id,
+                    build_configuration=build_config,
+                    state=PreprodArtifact.ArtifactState.UPLOADED,
+                )
 
-        # where next set of changes will happen
-        # TODO: Trigger artifact processing (size analysis, etc.)
-        # This is where you'd add logic to:
-        # 1. create_or_update a new row in the Commit table as well (once base_sha is added as a column to it)
-        # 2. Detect artifact type (iOS/Android/etc.)
-        # 3. Queue processing tasks
-        # 4. Update state to PROCESSED when done (also update the date_built value to reflect when the artifact was built, among other fields)
+                logger.info(
+                    "Created preprod artifact",
+                    extra={
+                        "preprod_artifact_id": preprod_artifact.id,
+                        "project_id": project_id,
+                        "organization_id": org_id,
+                        "checksum": checksum,
+                    },
+                )
 
     except Exception as e:
         sentry_sdk.capture_exception(e)
@@ -165,27 +165,25 @@ def assemble_preprod_artifact(
     # Mark assembly as successful since the artifact was created successfully
     set_assemble_status(AssembleTask.PREPROD_ARTIFACT, project_id, checksum, ChunkFileState.OK)
 
-    produce_preprod_artifact_to_kafka(
-        project_id=project_id,
-        organization_id=org_id,
-        artifact_id=preprod_artifact.id,
-        checksum=checksum,
-        git_sha=git_sha,
-        build_configuration=build_configuration,
-    )
+    if preprod_artifact:
+        produce_preprod_artifact_to_kafka(
+            project_id=project_id,
+            organization_id=org_id,
+            artifact_id=preprod_artifact.id,
+        )
 
-    logger.info(
-        "Finished preprod artifact assembly and Kafka dispatch",
-        extra={
-            "preprod_artifact_id": preprod_artifact.id,
-            "project_id": project_id,
-            "organization_id": org_id,
-            "checksum": checksum,
-        },
-    )
+        logger.info(
+            "Finished preprod artifact assembly and Kafka dispatch",
+            extra={
+                "preprod_artifact_id": preprod_artifact.id,
+                "project_id": project_id,
+                "organization_id": org_id,
+                "checksum": checksum,
+            },
+        )
 
 
-def _assemble_preprod_artifact(
+def _assemble_preprod_artifact_file(
     assemble_task: str,
     project_id: int,
     org_id: int,
@@ -194,7 +192,7 @@ def _assemble_preprod_artifact(
     callback: Callable[[AssembleResult, Any], None],
 ):
     logger.info(
-        "Starting preprod artifact assembly",
+        "Starting preprod file assembly",
         extra={
             "timestamp": datetime.datetime.now().isoformat(),
             "project_id": project_id,
@@ -229,7 +227,7 @@ def _assemble_preprod_artifact(
         callback(assemble_result, project)
     except Exception as e:
         logger.exception(
-            "Failed to assemble preprod artifact",
+            "Failed to assemble preprod file",
             extra={
                 "project_id": project_id,
                 "organization_id": org_id,
@@ -322,7 +320,7 @@ def assemble_preprod_artifact_size_analysis(
     """
     Creates a size analysis file for a preprod artifact from uploaded chunks.
     """
-    _assemble_preprod_artifact(
+    _assemble_preprod_artifact_file(
         AssembleTask.PREPROD_ARTIFACT_SIZE_ANALYSIS,
         project_id,
         org_id,

tests/sentry/preprod/test_tasks.py

@@ -262,7 +262,7 @@ def __init__(self):
         with (
             patch("sentry.preprod.tasks.assemble_file", return_value=MockAssembleResult()),
             patch.object(
-                PreprodArtifact.objects, "create", side_effect=Exception("Simulated failure")
+                PreprodArtifact.objects, "get_or_create", side_effect=Exception("Simulated failure")
             ),
         ):