feat: Log per-object warehouse delivery outcomes

api/experimentation/constants.py

@@ -15,6 +15,7 @@
 CONTROL_VARIANT_KEY = "control"
 
 DELIVERY_INTERVAL = timedelta(minutes=10)
+DELIVERY_LOG_RETENTION = timedelta(days=60)
 
 # Below these per-variant floors a metric shows "collecting data" rather than
 # inference; sample-ratio is only checked once there is enough traffic to judge.

api/experimentation/migrations/0012_warehouse_delivery_log.py

@@ -0,0 +1,58 @@
+# Generated by Django 5.2.16 on 2026-07-31 08:17
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("experimentation", "0011_organisation_ingestion_infrastructure"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="WarehouseDeliveryLog",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("s3_key", models.CharField(max_length=1024)),
+                (
+                    "outcome",
+                    models.CharField(
+                        choices=[("delivered", "Delivered"), ("rejected", "Rejected")],
+                        max_length=50,
+                    ),
+                ),
+                ("rows_count", models.PositiveIntegerField(blank=True, null=True)),
+                ("error", models.TextField(blank=True, null=True)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                (
+                    "connection",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="delivery_logs",
+                        to="experimentation.warehouseconnection",
+                    ),
+                ),
+            ],
+            options={
+                "indexes": [
+                    models.Index(
+                        fields=["connection", "created_at"],
+                        name="experimenta_connect_11e6cf_idx",
+                    ),
+                    models.Index(
+                        fields=["created_at"], name="experimenta_created_4fef63_idx"
+                    ),
+                ],
+            },
+        ),
+    ]

api/experimentation/models.py

@@ -125,6 +125,35 @@ class OrganisationIngestionInfrastructure(models.Model):
     updated_at = models.DateTimeField(auto_now=True)
 
 
+class WarehouseDeliveryOutcome(models.TextChoices):
+    DELIVERED = "delivered", "Delivered"
+    REJECTED = "rejected", "Rejected"
+
+
+class WarehouseDeliveryLog(models.Model):
+    connection = models.ForeignKey(
+        WarehouseConnection,
+        on_delete=models.CASCADE,
+        related_name="delivery_logs",
+    )
+    # S3's own key length limit.
+    s3_key = models.CharField(max_length=1024)
+    outcome = models.CharField(
+        max_length=50,
+        choices=WarehouseDeliveryOutcome.choices,
+    )
+    rows_count = models.PositiveIntegerField(null=True, blank=True)
+    error = models.TextField(null=True, blank=True)
+    created_at = models.DateTimeField(auto_now_add=True)
+
+    class Meta:
+        indexes = [
+            models.Index(fields=["connection", "created_at"]),
+            # Serves the retention cleanup, which filters on created_at alone.
+            models.Index(fields=["created_at"]),
+        ]
+
+
 class ExperimentStatus(models.TextChoices):
     CREATED = "created", "Created"
     RUNNING = "running", "Running"

api/experimentation/services.py

@@ -56,6 +56,8 @@
     MetricAggregation,
     MetricDirection,
     WarehouseConnectionStatus,
+    WarehouseDeliveryLog,
+    WarehouseDeliveryOutcome,
     WarehouseType,
 )
 from experimentation.results_query import _EXPOSURES_CTE, ResultsQueryBuilder
@@ -861,19 +863,25 @@ def _deliver_pending_objects(
             )
             break
         try:
-            rows_count += warehouse_delivery_service.deliver_object(
+            object_rows_count = warehouse_delivery_service.deliver_object(
                 client,
                 bucket_name,
                 s3_key,
             )
-        except warehouse_delivery_service.ObjectRejectedError:
+        except warehouse_delivery_service.ObjectRejectedError as exc:
             # This object's contents are the problem; the ones behind it are
             # still deliverable.
             warehouse_delivery_service.move_object(
                 bucket_name,
                 s3_key,
                 to_prefix=warehouse_delivery_service.FAILED_PREFIX,
             )
+            WarehouseDeliveryLog.objects.create(
+                connection=connection,
+                s3_key=s3_key,
+                outcome=WarehouseDeliveryOutcome.REJECTED,
+                error=str(exc),
+            )
             rejected_count += 1
             flagsmith_experimentation_warehouse_delivery_objects_total.labels(
                 result="rejected"
@@ -889,6 +897,13 @@ def _deliver_pending_objects(
             s3_key,
             to_prefix=warehouse_delivery_service.ARCHIVE_PREFIX,
         )
+        WarehouseDeliveryLog.objects.create(
+            connection=connection,
+            s3_key=s3_key,
+            outcome=WarehouseDeliveryOutcome.DELIVERED,
+            rows_count=object_rows_count,
+        )
+        rows_count += object_rows_count
         delivered_count += 1
         flagsmith_experimentation_warehouse_delivery_objects_total.labels(
             result="delivered"

api/experimentation/tasks.py

@@ -9,12 +9,13 @@
 
 from environments.models import Environment, EnvironmentAPIKey
 from experimentation import ingestion_sync_service
-from experimentation.constants import DELIVERY_INTERVAL
+from experimentation.constants import DELIVERY_INTERVAL, DELIVERY_LOG_RETENTION
 from experimentation.models import (
     Experiment,
     ExperimentExposures,
     ExperimentResults,
     WarehouseConnection,
+    WarehouseDeliveryLog,
     WarehouseType,
 )
 from experimentation.organisation_ingestion_service import (
@@ -154,6 +155,13 @@ def deliver_events_for_connection(connection_id: int) -> None:
     deliver_warehouse_events(connection, bucket_name=infrastructure.bucket_name)
 
 
+@register_recurring_task(run_every=timedelta(days=1))
+def clean_up_old_warehouse_delivery_logs() -> None:
+    WarehouseDeliveryLog.objects.filter(
+        created_at__lt=timezone.now() - DELIVERY_LOG_RETENTION,
+    ).delete()
+
+
 @register_task_handler()
 def compute_experiment_exposures(experiment_id: int) -> None:
     experiment = (

api/tests/unit/experimentation/test_tasks.py

@@ -17,6 +17,7 @@
 
 from environments.models import Environment, EnvironmentAPIKey
 from experimentation import warehouse_delivery_service
+from experimentation.constants import DELIVERY_LOG_RETENTION
 from experimentation.dataclasses import (
     ExposuresSummary,
     ExposuresTimeseries,
@@ -34,10 +35,13 @@
     OrganisationIngestionInfrastructure,
     WarehouseConnection,
     WarehouseConnectionStatus,
+    WarehouseDeliveryLog,
+    WarehouseDeliveryOutcome,
     WarehouseType,
 )
 from experimentation.stats import VariantStats
 from experimentation.tasks import (
+    clean_up_old_warehouse_delivery_logs,
     compute_experiment_exposures,
     compute_experiment_results,
     deliver_events_for_connection,
@@ -908,6 +912,26 @@ def test_deliver_events_for_connection__pending_objects__delivers_archives_and_r
         == []
     )
 
+    # Then each delivery is recorded in the audit ledger
+    assert list(
+        WarehouseDeliveryLog.objects.filter(connection=clickhouse_connection)
+        .order_by("s3_key")
+        .values_list("s3_key", "outcome", "rows_count", "error")
+    ) == [
+        (
+            _pending_key(environment.api_key, hour="13"),
+            WarehouseDeliveryOutcome.DELIVERED,
+            100,
+            None,
+        ),
+        (
+            _pending_key(environment.api_key, hour="14"),
+            WarehouseDeliveryOutcome.DELIVERED,
+            100,
+            None,
+        ),
+    ]
+
     # Then the delivery success resolves the earlier breakage
     clickhouse_connection.refresh_from_db()
     assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED
@@ -980,6 +1004,27 @@ def test_deliver_events_for_connection__rejected_object__moves_to_failed_and_con
     assert clickhouse_connection.status == WarehouseConnectionStatus.CONNECTED
     assert _delivery_objects_count("rejected") == rejected_objects_before + 1
     assert log.has("delivery.object_rejected", level="error")
+
+    # Then both outcomes are recorded in the audit ledger, the rejection with
+    # its warehouse error
+    assert list(
+        WarehouseDeliveryLog.objects.filter(connection=clickhouse_connection)
+        .order_by("s3_key")
+        .values_list("s3_key", "outcome", "rows_count", "error")
+    ) == [
+        (
+            _pending_key(environment.api_key, hour="13"),
+            WarehouseDeliveryOutcome.REJECTED,
+            None,
+            "Constraint `event_not_empty` violated",
+        ),
+        (
+            _pending_key(environment.api_key, hour="14"),
+            WarehouseDeliveryOutcome.DELIVERED,
+            100,
+            None,
+        ),
+    ]
     assert {
         "level": "info",
         "event": "delivery.completed",
@@ -1114,3 +1159,31 @@ def test_deliver_events_for_connection__time_budget_exhausted__defers_remaining(
         "organisation__id": environment.project.organisation_id,
         "objects__remaining_count": 2,
     } in log.events
+
+
+def test_clean_up_old_warehouse_delivery_logs__old_and_recent_logs__deletes_only_expired(
+    clickhouse_connection: WarehouseConnection,
+    environment: Environment,
+) -> None:
+    # Given a log older than the retention window and a recent one
+    expired_log = WarehouseDeliveryLog.objects.create(
+        connection=clickhouse_connection,
+        s3_key=_pending_key(environment.api_key, hour="13"),
+        outcome=WarehouseDeliveryOutcome.DELIVERED,
+        rows_count=100,
+    )
+    WarehouseDeliveryLog.objects.filter(id=expired_log.id).update(
+        created_at=timezone.now() - DELIVERY_LOG_RETENTION - timedelta(days=1),
+    )
+    recent_log = WarehouseDeliveryLog.objects.create(
+        connection=clickhouse_connection,
+        s3_key=_pending_key(environment.api_key, hour="14"),
+        outcome=WarehouseDeliveryOutcome.REJECTED,
+        error="Constraint `event_not_empty` violated",
+    )
+
+    # When
+    clean_up_old_warehouse_delivery_logs()
+
+    # Then
+    assert list(WarehouseDeliveryLog.objects.all()) == [recent_log]