fix(batch-exports): Couple of backfill bugfixes (#25854)

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>
PostHog · Oct 28, 2024 · a6abe61 · a6abe61
1 parent 2b389b3
commit a6abe61
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 14 deletions.
diff --git a/posthog/api/test/__snapshots__/test_api_docs.ambr b/posthog/api/test/__snapshots__/test_api_docs.ambr
@@ -97,8 +97,8 @@
     '/home/runner/work/posthog/posthog/posthog/api/survey.py: Warning [SurveyViewSet > SurveySerializer]: unable to resolve type hint for function "get_conditions". Consider using a type hint or @extend_schema_field. Defaulting to string.',
     '/home/runner/work/posthog/posthog/posthog/api/web_experiment.py: Warning [WebExperimentViewSet]: could not derive type of path parameter "project_id" because model "posthog.models.web_experiment.WebExperiment" contained no such field. Consider annotating parameter with @extend_schema. Defaulting to "string".',
     'Warning: encountered multiple names for the same choice set (HrefMatchingEnum). This may be unwanted even though the generated schema is technically correct. Add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
-    'Warning: enum naming encountered a non-optimally resolvable collision for fields named "kind". The same name has been used for multiple choice sets in multiple components. The collision was resolved with "Kind069Enum". add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
     'Warning: enum naming encountered a non-optimally resolvable collision for fields named "kind". The same name has been used for multiple choice sets in multiple components. The collision was resolved with "KindCfaEnum". add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
+    'Warning: enum naming encountered a non-optimally resolvable collision for fields named "kind". The same name has been used for multiple choice sets in multiple components. The collision was resolved with "Kind069Enum". add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
     'Warning: enum naming encountered a non-optimally resolvable collision for fields named "type". The same name has been used for multiple choice sets in multiple components. The collision was resolved with "TypeF73Enum". add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
     'Warning: encountered multiple names for the same choice set (EffectivePrivilegeLevelEnum). This may be unwanted even though the generated schema is technically correct. Add an entry to ENUM_NAME_OVERRIDES to fix the naming.',
     'Warning: encountered multiple names for the same choice set (MembershipLevelEnum). This may be unwanted even though the generated schema is technically correct. Add an entry to ENUM_NAME_OVERRIDES to fix the naming.',

diff --git a/posthog/batch_exports/models.py b/posthog/batch_exports/models.py
@@ -289,7 +289,7 @@ class Status(models.TextChoices):
     @property
     def workflow_id(self) -> str:
         """Return the Workflow id that corresponds to this BatchExportBackfill model."""
-        end_at = self.end_at and self.end_at.isoformat()
-        start_at = self.start_at and self.start_at.isoformat()
+        end_at = self.end_at.astimezone(tz=dt.UTC).isoformat() if self.end_at else "END"
+        start_at = self.start_at.astimezone(tz=dt.UTC).isoformat() if self.start_at else "START"
 
         return f"{self.batch_export.id}-Backfill-{start_at}-{end_at}"
diff --git a/posthog/temporal/batch_exports/backfill_batch_export.py b/posthog/temporal/batch_exports/backfill_batch_export.py
@@ -173,7 +173,7 @@ async def backfill_schedule(inputs: BackfillScheduleInputs) -> None:
         if details:
             # If we receive details from a previous run, it means we were restarted for some reason.
             # Let's not double-backfill and instead wait for any outstanding runs.
-            last_activity_details = HeartbeatDetails(*details[0])
+            last_activity_details = HeartbeatDetails(*details)
 
             workflow_handle = client.get_workflow_handle(last_activity_details.workflow_id)
 
@@ -241,16 +241,20 @@ async def backfill_schedule(inputs: BackfillScheduleInputs) -> None:
 
             await asyncio.sleep(inputs.start_delay)
 
-            workflow_handle = await client.start_workflow(
-                schedule_action.workflow,
-                *args,
-                id=f"{description.id}-{backfill_end_at:%Y-%m-%dT%H:%M:%S}Z",
-                task_queue=schedule_action.task_queue,
-                run_timeout=schedule_action.run_timeout,
-                task_timeout=schedule_action.task_timeout,
-                id_reuse_policy=temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,
-                search_attributes=temporalio.common.TypedSearchAttributes(search_attributes=search_attributes),
-            )
+            try:
+                workflow_handle = await client.start_workflow(
+                    schedule_action.workflow,
+                    *args,
+                    id=f"{description.id}-{backfill_end_at:%Y-%m-%dT%H:%M:%S}Z",
+                    task_queue=schedule_action.task_queue,
+                    run_timeout=schedule_action.run_timeout,
+                    task_timeout=schedule_action.task_timeout,
+                    id_reuse_policy=temporalio.common.WorkflowIDReusePolicy.ALLOW_DUPLICATE,
+                    search_attributes=temporalio.common.TypedSearchAttributes(search_attributes=search_attributes),
+                )
+            except temporalio.exceptions.WorkflowAlreadyStartedError:
+                workflow_handle = client.get_workflow_handle(f"{description.id}-{backfill_end_at:%Y-%m-%dT%H:%M:%S}Z")
+
             details = HeartbeatDetails(
                 schedule_id=inputs.schedule_id,
                 workflow_id=workflow_handle.id,