feat(experiments): replace the 24 schedules with one hourly schedule

Author: rodrigoi

posthog/temporal/experiments/README.md

@@ -92,13 +92,19 @@ The `ExperimentSavedMetricsWorkflow` follows the same structure but:
 
 The timeseries sync row is assembled from that day's points, so its window is approximate and a
 metric the daily run could not compute has no row at all. A separate workflow therefore starts a
-real recalculation for each eligible experiment, on its own 24 schedules
+real recalculation for each eligible experiment, on one hourly schedule
 (`products/experiments/backend/temporal/schedule.py`).
 
-Those schedules fire at `:30`, after the timeseries schedules at `:00`. The coordinator selects
-experiments with the same rules the daily discovery uses, plus a 12-hour minimum age, an
-organization feature flag, and a 50-exposure floor. It then starts an ordinary
-`ExperimentMetricsRecalculationWorkflow` per experiment, through the same function the API uses.
+That schedule fires at `:30`, after the timeseries schedules at `:00`, and carries no input.
+Discovery reads the hour and selects the teams configured for it, so one schedule serves all 24
+hours. The coordinator selects experiments with the same rules the daily discovery uses, plus a
+12-hour minimum age, an organization feature flag, and a 50-exposure floor. It then starts an
+ordinary `ExperimentMetricsRecalculationWorkflow` per experiment, through the same function the
+API uses.
+
+The timeseries workflows keep one schedule per hour, because each runs its own metric queries and
+can outlive its hour. This one starts other workflows and returns, so a single schedule with a
+`SKIP` overlap policy covers it.
 
 It skips an experiment whose recalculation is already running, or whose last one finished within
 the hour. That freshness check ignores `timeseries_sync` rows: one is published 30 minutes before

products/experiments/backend/temporal/models.py

@@ -219,13 +219,6 @@ class ExperimentPrecomputeEnrollmentCensusInputs:
 SCHEDULED_RECALCULATION_WORKFLOW_NAME = "experiment-scheduled-recalculation-workflow"
 
 
-@dataclasses.dataclass(frozen=False)
-class ScheduledRecalculationWorkflowInputs:
-    """Input to the scheduled recalculation coordinator."""
-
-    hour: int  # 0-23, which hour's teams to process
-
-
 @frozen
 class ScheduledRecalculationStartResult:
     """Outcome of one experiment's start attempt, for the coordinator's summary counts."""

products/experiments/backend/temporal/schedule.py

@@ -19,7 +19,6 @@
     SCHEDULED_RECALCULATION_WORKFLOW_NAME,
     ExperimentPrecomputeCanaryInputs,
     ExperimentPrecomputeEnrollmentCensusInputs,
-    ScheduledRecalculationWorkflowInputs,
 )
 
 CANARY_SCHEDULE_ID = "experiment-precompute-canary-schedule"
@@ -79,42 +78,57 @@ async def create_experiment_precompute_enrollment_census_schedule(client: Client
         await a_create_schedule(client, ENROLLMENT_CENSUS_SCHEDULE_ID, schedule, trigger_immediately=False)
 
 
-SCHEDULED_RECALCULATION_SCHEDULE_ID_PREFIX = "experiment-scheduled-recalculation-hour"
+SCHEDULED_RECALCULATION_SCHEDULE_ID = "experiment-scheduled-recalculation"
+
+# Each hour had its own schedule before the workflow resolved the hour itself. Temporal keeps a
+# schedule until it is deleted, so the replaced ones are removed when the single one is created.
+LEGACY_HOURLY_SCHEDULE_ID_PREFIX = "experiment-scheduled-recalculation-hour"
 
 
 async def create_experiment_scheduled_recalculation_schedules(client: Client) -> None:
-    """Create or update 24 schedules, one per hour, that start a real metrics recalculation for
-    every eligible experiment belonging to teams configured for that hour.
+    """Create or update the hourly schedule that starts a real metrics recalculation for every
+    experiment eligible in the hour it fires.
+
+    Fires at :30, after the daily timeseries schedule at :00, so the timeseries sync publish
+    usually lands before a real run supersedes it. The workflow takes no input: discovery reads
+    the hour and selects the teams configured for it, so one schedule serves all 24 hours.
 
-    Each fires at :30, after the daily timeseries schedule at :00, so the timeseries sync publish
-    usually lands before a real run supersedes it. 24 separate schedules rather than one hourly
-    schedule, matching the timeseries workflows: a run longer than an hour cannot overlap itself,
-    so no overlap policy is needed.
+    SKIP overlap: a run that outlives its hour must not stack on the next one.
     """
-    for hour in range(24):
-        schedule_id = f"{SCHEDULED_RECALCULATION_SCHEDULE_ID_PREFIX}-{hour:02d}"
+    schedule = Schedule(
+        action=ScheduleActionStartWorkflow(
+            SCHEDULED_RECALCULATION_WORKFLOW_NAME,
+            id=f'{SCHEDULED_RECALCULATION_SCHEDULE_ID}-{{{{.ScheduledTime.Format "2006-01-02T15"}}}}',
+            task_queue=settings.GENERAL_PURPOSE_TASK_QUEUE,
+        ),
+        spec=ScheduleSpec(cron_expressions=["30 * * * *"]),
+        policy=SchedulePolicy(overlap=ScheduleOverlapPolicy.SKIP),
+    )
+
+    if await a_schedule_exists(client, SCHEDULED_RECALCULATION_SCHEDULE_ID):
+        await a_update_schedule(client, SCHEDULED_RECALCULATION_SCHEDULE_ID, schedule)
+    else:
+        await a_create_schedule(client, SCHEDULED_RECALCULATION_SCHEDULE_ID, schedule, trigger_immediately=False)
 
-        schedule = Schedule(
-            action=ScheduleActionStartWorkflow(
-                SCHEDULED_RECALCULATION_WORKFLOW_NAME,
-                ScheduledRecalculationWorkflowInputs(hour=hour),
-                id=f'{SCHEDULED_RECALCULATION_SCHEDULE_ID_PREFIX}-{hour:02d}-{{{{.ScheduledTime.Format "2006-01-02"}}}}',
-                task_queue=settings.GENERAL_PURPOSE_TASK_QUEUE,
-            ),
-            spec=ScheduleSpec(cron_expressions=[f"30 {hour} * * *"]),
-            policy=SchedulePolicy(overlap=ScheduleOverlapPolicy.SKIP),
-        )
+    await _delete_legacy_hourly_schedules(client)
 
-        if await a_schedule_exists(client, schedule_id):
-            await a_update_schedule(client, schedule_id, schedule)
-        else:
-            await a_create_schedule(client, schedule_id, schedule, trigger_immediately=False)
 
+async def _delete_legacy_hourly_schedules(client: Client) -> None:
+    """Remove the 24 per-hour schedules the single schedule replaces.
 
-async def delete_experiment_scheduled_recalculation_schedules(client: Client) -> None:
-    """Delete all 24 scheduled recalculation schedules."""
+    Without this they keep firing alongside it, starting the workflow 25 times an hour.
+    """
     for hour in range(24):
         try:
-            await a_delete_schedule(client, f"{SCHEDULED_RECALCULATION_SCHEDULE_ID_PREFIX}-{hour:02d}")
+            await a_delete_schedule(client, f"{LEGACY_HOURLY_SCHEDULE_ID_PREFIX}-{hour:02d}")
         except Exception:
-            pass  # Schedule might not exist
+            pass  # Already gone, which is the steady state after the first run.
+
+
+async def delete_experiment_scheduled_recalculation_schedules(client: Client) -> None:
+    """Delete the scheduled recalculation schedule, and any legacy per-hour ones left behind."""
+    try:
+        await a_delete_schedule(client, SCHEDULED_RECALCULATION_SCHEDULE_ID)
+    except Exception:
+        pass  # Schedule might not exist
+    await _delete_legacy_hourly_schedules(client)

products/experiments/backend/temporal/test_scheduled_recalculation_schedule.py

@@ -5,7 +5,8 @@
 
 from products.experiments.backend.temporal.models import SCHEDULED_RECALCULATION_WORKFLOW_NAME
 from products.experiments.backend.temporal.schedule import (
-    SCHEDULED_RECALCULATION_SCHEDULE_ID_PREFIX,
+    LEGACY_HOURLY_SCHEDULE_ID_PREFIX,
+    SCHEDULED_RECALCULATION_SCHEDULE_ID,
     create_experiment_scheduled_recalculation_schedules,
 )
 
@@ -14,41 +15,53 @@
 MODULE = "products.experiments.backend.temporal.schedule"
 
 
-async def test_creates_one_schedule_per_hour_offset_from_the_timeseries_run():
+async def test_creates_one_hourly_schedule_offset_from_the_timeseries_run():
     with (
         patch(f"{MODULE}.a_schedule_exists", AsyncMock(return_value=False)),
         patch(f"{MODULE}.a_create_schedule", AsyncMock()) as create,
         patch(f"{MODULE}.a_update_schedule", AsyncMock()) as update,
+        patch(f"{MODULE}.a_delete_schedule", AsyncMock()),
     ):
         await create_experiment_scheduled_recalculation_schedules(AsyncMock())
 
-    assert create.await_count == 24
+    assert create.await_count == 1
     update.assert_not_awaited()
 
-    schedule_ids = [call.args[1] for call in create.await_args_list]
-    assert schedule_ids[0] == f"{SCHEDULED_RECALCULATION_SCHEDULE_ID_PREFIX}-00"
-    assert schedule_ids[23] == f"{SCHEDULED_RECALCULATION_SCHEDULE_ID_PREFIX}-23"
+    # Every field a typo could break: a wrong workflow name or task queue fails only in
+    # production, where nothing reports it.
+    call = create.await_args_list[0]
+    schedule_id, schedule = call.args[1], call.args[2]
+    assert schedule_id == SCHEDULED_RECALCULATION_SCHEDULE_ID
+    # :30, so the daily timeseries run at :00 publishes before a real run supersedes it.
+    assert schedule.spec.cron_expressions == ["30 * * * *"]
+    assert schedule.action.workflow == SCHEDULED_RECALCULATION_WORKFLOW_NAME
+    assert schedule.action.task_queue == settings.GENERAL_PURPOSE_TASK_QUEUE
+    # No input: discovery resolves the hour, so one schedule serves every hour.
+    assert schedule.action.args == []
 
-    # Every field a typo could break, checked per schedule rather than at the ends: a wrong
-    # workflow name or task queue fails only in production, where nothing reports it.
-    for hour, call in enumerate(create.await_args_list):
-        schedule_id, schedule = call.args[1], call.args[2]
-        assert schedule_id == f"{SCHEDULED_RECALCULATION_SCHEDULE_ID_PREFIX}-{hour:02d}"
-        # :30, so the daily timeseries run at :00 publishes before a real run supersedes it.
-        assert schedule.spec.cron_expressions == [f"30 {hour} * * *"]
-        assert schedule.action.workflow == SCHEDULED_RECALCULATION_WORKFLOW_NAME
-        assert schedule.action.task_queue == settings.GENERAL_PURPOSE_TASK_QUEUE
-        assert schedule.action.args[0].hour == hour
-        assert schedule.action.id.startswith(f"{SCHEDULED_RECALCULATION_SCHEDULE_ID_PREFIX}-{hour:02d}-")
+
+async def test_the_replaced_per_hour_schedules_are_deleted():
+    # Temporal keeps a schedule until it is deleted, so leaving the 24 behind would start the
+    # workflow 25 times an hour.
+    with (
+        patch(f"{MODULE}.a_schedule_exists", AsyncMock(return_value=False)),
+        patch(f"{MODULE}.a_create_schedule", AsyncMock()),
+        patch(f"{MODULE}.a_delete_schedule", AsyncMock()) as delete,
+    ):
+        await create_experiment_scheduled_recalculation_schedules(AsyncMock())
+
+    deleted = [call.args[1] for call in delete.await_args_list]
+    assert deleted == [f"{LEGACY_HOURLY_SCHEDULE_ID_PREFIX}-{hour:02d}" for hour in range(24)]
 
 
-async def test_existing_schedules_are_updated_not_recreated():
+async def test_an_existing_schedule_is_updated_not_recreated():
     with (
         patch(f"{MODULE}.a_schedule_exists", AsyncMock(return_value=True)),
         patch(f"{MODULE}.a_create_schedule", AsyncMock()) as create,
         patch(f"{MODULE}.a_update_schedule", AsyncMock()) as update,
+        patch(f"{MODULE}.a_delete_schedule", AsyncMock()),
     ):
         await create_experiment_scheduled_recalculation_schedules(AsyncMock())
 
-    assert update.await_count == 24
+    assert update.await_count == 1
     create.assert_not_awaited()