feat(nimbus): Celery tasks to run holdback analysis weekly

docs/experimenter/openapi-schema.json

@@ -1633,11 +1633,11 @@
           },
           "enrollmentEndDate": {
             "type": "string",
-            "format": "date"
+            "readOnly": true
           },
           "endDate": {
             "type": "string",
-            "format": "date"
+            "readOnly": true
           },
           "proposedDuration": {
             "type": "string",
@@ -1697,8 +1697,6 @@
           "channel",
           "bucketConfig",
           "startDate",
-          "enrollmentEndDate",
-          "endDate",
           "publishedDate"
         ]
       },

docs/experimenter/swagger-ui.html

@@ -1645,11 +1645,11 @@
           },
           "enrollmentEndDate": {
             "type": "string",
-            "format": "date"
+            "readOnly": true
           },
           "endDate": {
             "type": "string",
-            "format": "date"
+            "readOnly": true
           },
           "proposedDuration": {
             "type": "string",
@@ -1709,8 +1709,6 @@
           "channel",
           "bucketConfig",
           "startDate",
-          "enrollmentEndDate",
-          "endDate",
           "publishedDate"
         ]
       },

experimenter/experimenter/experiments/api/v8/serializers.py

@@ -1,4 +1,5 @@
 import contextlib
+import datetime
 import json
 
 from django.conf import settings
@@ -102,10 +103,10 @@ class NimbusExperimentSerializer(serializers.ModelSerializer):
     outcomes = serializers.SerializerMethodField()
     segments = serializers.SerializerMethodField()
     startDate = serializers.DateField(source="start_date")
-    enrollmentEndDate = serializers.DateField(source="actual_enrollment_end_date")
-    endDate = serializers.DateField(source="end_date")
+    enrollmentEndDate = serializers.SerializerMethodField()
+    endDate = serializers.SerializerMethodField()
     proposedDuration = serializers.ReadOnlyField(source="proposed_duration")
-    proposedEnrollment = serializers.ReadOnlyField(source="proposed_enrollment")
+    proposedEnrollment = serializers.SerializerMethodField()
     referenceBranch = serializers.SerializerMethodField()
     featureValidationOptOut = serializers.ReadOnlyField(
         source="is_client_schema_disabled"
@@ -161,6 +162,34 @@ class Meta:
             "requiresRestart",
         )
 
+    def _holdback_enrollment_end(self, obj):
+        if obj.is_holdback and not obj.end_date and not obj.actual_enrollment_end_date:
+            return datetime.date.today() - datetime.timedelta(
+                days=settings.HOLDBACK_OBSERVATION_DAYS
+            )
+        return None
+
+    def get_enrollmentEndDate(self, obj):
+        holdback_end = self._holdback_enrollment_end(obj)
+        if holdback_end:
+            return holdback_end.isoformat()
+        enrollment_end = obj.actual_enrollment_end_date
+        return enrollment_end.isoformat() if enrollment_end else None
+
+    def get_endDate(self, obj):
+        holdback_end = self._holdback_enrollment_end(obj)
+        if holdback_end:
+            return (
+                holdback_end + datetime.timedelta(days=settings.HOLDBACK_OBSERVATION_DAYS)
+            ).isoformat()
+        return obj.end_date.isoformat() if obj.end_date else None
+
+    def get_proposedEnrollment(self, obj):
+        holdback_end = self._holdback_enrollment_end(obj)
+        if holdback_end and obj.start_date:
+            return (holdback_end - obj.start_date).days
+        return obj.proposed_enrollment
+
     def get_appName(self, obj):
         return obj.application_config.app_name
 

experimenter/experimenter/experiments/models.py

@@ -3307,6 +3307,7 @@ class Messages:
         COMPLETED = "Experiment is complete"
         RESULTS_UPDATED = "Experiment results updated"
         MONITORING_DATA_UPDATED = "Experiment monitoring data updated"
+        HOLDBACK_ENROLLMENT_UPDATED = "Holdback enrollment period updated"
         EXPIRED_FROM_PREVIEW = "Expired from preview collection after 30 days"
         REMOVED_FROM_PREVIEW = "Removed from preview collection"
         PUSHED_TO_PREVIEW = "Pushed to preview collection"

experimenter/experimenter/experiments/tests/api/v8/test_serializers.py

@@ -416,6 +416,34 @@ def test_localized_localizations_json(self, l10n_json, expected):
         else:
             self.assertEqual(serializer.data["localizations"], expected)
 
+    def test_holdback_serializer_overrides(self):
+        today = datetime.date.today()
+        start = today - datetime.timedelta(days=50)
+        expected_enrollment_end = today - datetime.timedelta(days=21)
+        experiment = NimbusExperimentFactory.create_with_lifecycle(
+            NimbusExperimentFactory.Lifecycles.LIVE_ENROLLING,
+            is_holdback=True,
+            _start_date=start,
+        )
+        serializer = NimbusExperimentSerializer(experiment)
+        data = serializer.data
+
+        self.assertEqual(data["enrollmentEndDate"], expected_enrollment_end.isoformat())
+        self.assertEqual(data["endDate"], today.isoformat())
+        self.assertEqual(
+            data["proposedEnrollment"],
+            (expected_enrollment_end - start).days,
+        )
+
+    def test_non_holdback_proposed_enrollment_uses_model_value(self):
+        experiment = NimbusExperimentFactory.create_with_lifecycle(
+            NimbusExperimentFactory.Lifecycles.LIVE_ENROLLING,
+            is_holdback=False,
+            proposed_enrollment=14,
+        )
+        serializer = NimbusExperimentSerializer(experiment)
+        self.assertEqual(serializer.data["proposedEnrollment"], 14)
+
     def _experiment_data_without_branches_and_featureIds(
         self, experiment_data, min_required_version
     ) -> dict[str, Any]:

experimenter/experimenter/jetstream/tasks.py

@@ -196,3 +196,46 @@ def fetch_monitoring_data():
         metrics.incr("fetch_monitoring_data.failed")
         logger.exception(f"Fatal error in fetch_monitoring_data task: {e}")
         raise
+
+
+@app.task
+@metrics.timer_decorator("update_holdback_enrollment_period")
+def update_holdback_enrollment_period():
+    metrics.incr("update_holdback_enrollment_period.started")
+    try:
+        today = timezone.now().date()
+        now = timezone.now()
+        enrollment_end = today - dt.timedelta(days=settings.HOLDBACK_OBSERVATION_DAYS)
+
+        experiments = NimbusExperiment.objects.filter(
+            is_holdback=True,
+            status=NimbusExperiment.Status.LIVE,
+            _end_date=None,
+        ).exclude(_start_date=None)
+
+        updated_count = 0
+        for experiment in experiments:
+            if enrollment_end <= experiment.start_date:
+                continue
+
+            NimbusExperiment.objects.filter(pk=experiment.pk).update(
+                do_rerun=True,
+                do_rerun_timestamp=now,
+            )
+            experiment.refresh_from_db()
+            generate_nimbus_changelog(
+                experiment,
+                get_kinto_user(),
+                message=NimbusChangeLog.Messages.HOLDBACK_ENROLLMENT_UPDATED,
+            )
+            updated_count += 1
+
+        logger.info(
+            f"update_holdback_enrollment_period: updated {updated_count} experiments"
+        )
+        metrics.incr("update_holdback_enrollment_period.completed")
+
+    except Exception as e:
+        metrics.incr("update_holdback_enrollment_period.failed")
+        logger.exception(f"Fatal error in update_holdback_enrollment_period: {e}")
+        raise

experimenter/experimenter/jetstream/tests/test_tasks.py

@@ -3949,3 +3949,76 @@ def test_result_is_cached(self):
             get_featmon_slugs()
             get_featmon_slugs()
             mock_read.assert_called_once()
+
+
+class TestUpdateHoldbackEnrollmentPeriod(TestCase):
+    def test_sets_do_rerun(self):
+        today = datetime.date.today()
+        start = today - datetime.timedelta(days=50)
+        experiment = NimbusExperimentFactory.create_with_lifecycle(
+            NimbusExperimentFactory.Lifecycles.LIVE_ENROLLING,
+            is_holdback=True,
+            _start_date=start,
+            proposed_enrollment=14,
+            proposed_duration=84,
+        )
+        tasks.update_holdback_enrollment_period()
+        experiment.refresh_from_db()
+
+        self.assertTrue(experiment.do_rerun)
+        self.assertIsNotNone(experiment.do_rerun_timestamp)
+        self.assertIsNone(experiment._enrollment_end_date)
+
+    def test_skips_experiment_started_within_observation_period(self):
+        today = datetime.date.today()
+        experiment = NimbusExperimentFactory.create_with_lifecycle(
+            NimbusExperimentFactory.Lifecycles.LIVE_ENROLLING,
+            is_holdback=True,
+            _start_date=today - datetime.timedelta(days=10),
+            proposed_enrollment=14,
+            proposed_duration=84,
+        )
+        tasks.update_holdback_enrollment_period()
+        experiment.refresh_from_db()
+
+        self.assertFalse(experiment.do_rerun)
+
+    def test_skips_ended_holdback(self):
+        today = datetime.date.today()
+        experiment = NimbusExperimentFactory.create_with_lifecycle(
+            NimbusExperimentFactory.Lifecycles.LIVE_ENROLLING,
+            is_holdback=True,
+            _start_date=today - datetime.timedelta(days=50),
+            proposed_enrollment=14,
+            proposed_duration=84,
+        )
+        experiment._end_date = today - datetime.timedelta(days=1)
+        experiment.save(update_fields=["_end_date"])
+        tasks.update_holdback_enrollment_period()
+        experiment.refresh_from_db()
+
+        self.assertFalse(experiment.do_rerun)
+
+    def test_skips_non_holdback_experiments(self):
+        today = datetime.date.today()
+        experiment = NimbusExperimentFactory.create_with_lifecycle(
+            NimbusExperimentFactory.Lifecycles.LIVE_ENROLLING,
+            is_holdback=False,
+            _start_date=today - datetime.timedelta(days=50),
+            proposed_enrollment=14,
+            proposed_duration=84,
+        )
+        tasks.update_holdback_enrollment_period()
+        experiment.refresh_from_db()
+
+        self.assertFalse(experiment.do_rerun)
+
+    def test_raises_on_unexpected_error(self):
+        with (
+            patch(
+                "experimenter.jetstream.tasks.NimbusExperiment.objects.filter",
+                side_effect=Exception("db error"),
+            ),
+            self.assertRaises(Exception, msg="db error"),
+        ):
+            tasks.update_holdback_enrollment_period()

experimenter/experimenter/settings.py

@@ -380,6 +380,8 @@
 API_CACHE_WARMING_TTL = 60 * 60 * 24
 SIZING_DATA_KEY = "population_sizing"
 
+HOLDBACK_OBSERVATION_DAYS = 21
+
 # Celery
 CELERY_BROKER_URL = f"redis://{REDIS_HOST}:{REDIS_PORT}/{REDIS_DB}"
 CELERY_BEAT_SCHEDULE = {
@@ -421,6 +423,10 @@
         "task": "experimenter.experiments.tasks.warm_api_caches",
         "schedule": config("API_CACHE_WARMING_INTERVAL", default=3600, cast=int),
     },
+    "update_holdback_enrollment_period": {
+        "task": "experimenter.jetstream.tasks.update_holdback_enrollment_period",
+        "schedule": crontab(minute=0, hour=7),
+    },
 }
 CELERY_TASK_ROUTES = {
     "experimenter.kinto.tasks.*": {"queue": "remote_settings"},