fix(workflows): reload a saved workflow only after its commit

products/messaging/backend/services/transactional_email.py

@@ -16,10 +16,7 @@
 
 from posthog.integrations import IntegrationManager, IntegrationNotFound
 from posthog.models.team import Team
-from posthog.plugins.plugin_server_api import (
-    reload_hog_flows_on_workers,
-    send_transactional_email as run_transactional_workflow,
-)
+from posthog.plugins.plugin_server_api import send_transactional_email as run_transactional_workflow
 
 from products.messaging.backend.facade.contracts import (
     DeliveryUnavailable,
@@ -308,15 +305,6 @@ def check_transactional_update(template_id: UUID, team_id: int, fields: dict[str
     )
 
 
-def _reload_workers(team_id: int, hog_flow_id: UUID) -> None:
-    # The edit is already committed, so a failed reload must not fail the request. Workers still pick up the
-    # new version when their cache expires.
-    try:
-        reload_hog_flows_on_workers(team_id=team_id, hog_flow_ids=[str(hog_flow_id)])
-    except Exception:
-        logger.exception("transactional_email_worker_reload_failed", team_id=team_id, hog_flow_id=str(hog_flow_id))
-
-
 def sync_transactional_workflow(team_id: int, template_id: UUID, *, user_id: int | None) -> None:
     """Write the template's current content as a new version of its workflow, if it has one."""
     template = MessageTemplate.objects.get(team_id=team_id, id=template_id)
@@ -334,9 +322,6 @@ def sync_transactional_workflow(team_id: int, template_id: UUID, *, user_id: int
             hog_flow_id=hog_flow_id,
             content=_content(template),
         )
-        # The save already told the workers to reload, but before the caller's transaction committed, so a
-        # worker can cache the old version. This reload runs once the new version is visible.
-        transaction.on_commit(lambda: _reload_workers(team_id, hog_flow_id))
     except WorkflowNotFound:
         # Deleted between the check above and the update.
         MessageTemplate.objects.filter(team_id=team_id, id=template_id).update(transactional_hog_flow_id=None)

products/messaging/backend/tests/api/test_transactional_email.py

@@ -449,7 +449,7 @@ def test_rejects_making_a_template_transactional_it_cannot_send(
         template.refresh_from_db()
         assert template.transactional_hog_flow_id is None
 
-    @patch("products.messaging.backend.services.transactional_email.reload_hog_flows_on_workers")
+    @patch("products.workflows.backend.models.hog_flow.hog_flow.reload_hog_flows_on_workers")
     def test_saving_a_transactional_template_writes_a_new_workflow_version(self, reload, mock_post, _flag, _capture):
         MessageTemplate.objects.filter(id=self.template.id).update(draft_content={"email": {"subject": "Draft"}})
         security = Integration.objects.create(
@@ -468,6 +468,7 @@ def test_saving_a_transactional_template_writes_a_new_workflow_version(self, rel
                 {"content": {"email": email_content}},
                 format="json",
             )
+            reload.assert_not_called()
 
         assert response.status_code == status.HTTP_200_OK, response.json()
         # Workers reload once the new version is committed, not only from inside the transaction.

products/workflows/backend/management/commands/rewrite_email_asset_url.py

@@ -9,7 +9,7 @@
 from django.db.models.functions import Cast
 from django.utils import timezone
 
-from posthog.plugins.plugin_server_api import reload_hog_flows_on_workers, reload_hog_functions_on_workers
+from posthog.plugins.plugin_server_api import reload_hog_functions_on_workers
 
 from products.cdp.backend.models.hog_functions.hog_function import HogFunction
 from products.messaging.backend.facade.api import find_template_ids_containing, rewrite_template_content
@@ -22,9 +22,8 @@
 RowRewrite = Callable[[Any], int]
 
 # Only executed config has a worker-side cache to invalidate; templates are copied at creation time
-# and never read by a worker, so they need no reload.
+# and never read by a worker, so they need no reload. The workflow model reloads its own rows on commit.
 WORKER_RELOADS: Final[dict[str, WorkerReload]] = {
-    "hog_flows": lambda team_id, row_id: reload_hog_flows_on_workers(team_id=team_id, hog_flow_ids=[row_id]),
     "hog_functions": lambda team_id, row_id: reload_hog_functions_on_workers(
         team_id=team_id, hog_function_ids=[row_id]
     ),
@@ -288,7 +287,7 @@ def _rewrite_row(
             # is published for the row, and bulk_update fires no signal at all.
             row.save(update_fields=[*changed_fields, *timestamp_fields])
 
-            # The post_save receiver publishes from inside this transaction, so a worker can re-read
+            # The hog function post_save receiver publishes from inside this transaction, so a worker can re-read
             # the row before the UPDATE commits and cache the old blob for another refresh window.
             # Publishing again on commit is what workers actually act on; the duplicate is harmless.
             # Draft-only rewrites publish nothing, matching the receiver that skips them.

products/workflows/backend/management/commands/strip_placeholder_sender_overrides.py

@@ -1,14 +1,11 @@
 import logging
-from functools import partial
 from typing import Any
 
 from django.core.management.base import BaseCommand, CommandParser
 from django.db import transaction
 from django.db.models import Q
 from django.utils import timezone
 
-from posthog.plugins.plugin_server_api import reload_hog_flows_on_workers
-
 from products.workflows.backend.models.hog_flow.hog_flow import HogFlow
 
 logger = logging.getLogger(__name__)
@@ -106,20 +103,6 @@ def handle(self, *args: Any, **options: Any) -> None:
                         locked.draft_updated_at = timezone.now()
                         update_fields += ["draft", "draft_updated_at"]
                     locked.save(update_fields=update_fields)
-
-                    # The post_save receiver publishes the worker reload from inside this
-                    # transaction, so a worker can re-read the row before the UPDATE commits and
-                    # cache the old blob for another refresh window. Publishing again on commit is
-                    # what workers actually act on; the duplicate is harmless. Draft-only strips
-                    # publish nothing, matching the receiver that skips them.
-                    if live_stripped:
-                        transaction.on_commit(
-                            partial(
-                                reload_hog_flows_on_workers,
-                                team_id=locked.team_id,
-                                hog_flow_ids=[str(locked.pk)],
-                            )
-                        )
             except Exception:
                 logger.exception("Failed to strip placeholder sender", extra={"hog_flow_id": str(flow.pk)})
                 self.stderr.write(f"  FAILED on flow id={flow.id} team_id={flow.team_id}; continuing")

products/workflows/backend/models/hog_flow/hog_flow.py

@@ -334,10 +334,14 @@ def defer_hog_flow_reloads() -> Iterator[None]:
         yield
     finally:
         _deferred_reloads.reset(token)
-        for team_id, hog_flow_ids in pending.items():
-            ordered = sorted(hog_flow_ids)
-            for start in range(0, len(ordered), RELOAD_BATCH_SIZE):
-                _reload_batch_on_workers(team_id, ordered[start : start + RELOAD_BATCH_SIZE])
+        transaction.on_commit(lambda: _reload_deferred_on_workers(pending))
+
+
+def _reload_deferred_on_workers(pending: dict[int, set[str]]) -> None:
+    for team_id, hog_flow_ids in pending.items():
+        ordered = sorted(hog_flow_ids)
+        for start in range(0, len(ordered), RELOAD_BATCH_SIZE):
+            _reload_batch_on_workers(team_id, ordered[start : start + RELOAD_BATCH_SIZE])
 
 
 def _reload_batch_on_workers(team_id: int, hog_flow_ids: list[str]) -> None:
@@ -362,7 +366,9 @@ def hog_flow_saved(sender, instance: HogFlow, created, update_fields=None, **kwa
     if pending is not None:
         pending.setdefault(instance.team_id, set()).add(str(instance.id))
         return
-    reload_hog_flows_on_workers(team_id=instance.team_id, hog_flow_ids=[str(instance.id)])
+    team_id = instance.team_id
+    hog_flow_id = str(instance.id)
+    transaction.on_commit(lambda: _reload_batch_on_workers(team_id, [hog_flow_id]))
 
 
 @receiver(post_delete, sender=HogFlow)
@@ -371,7 +377,7 @@ def hog_flow_deleted(sender, instance: HogFlow, **kwargs):
     hog_flow_id = str(instance.id)
     # post_delete fires inside the delete transaction, so publish only after commit; otherwise a
     # worker could re-read the still-live row and cache it as active for another TTL.
-    transaction.on_commit(lambda: reload_hog_flows_on_workers(team_id=team_id, hog_flow_ids=[hog_flow_id]))
+    transaction.on_commit(lambda: _reload_batch_on_workers(team_id, [hog_flow_id]))
 
 
 @receiver(post_save, sender=Action)

products/workflows/backend/services/workflow_email_health.py

@@ -420,10 +420,9 @@ def pause_workflow_email_sending(
         # A pause landing inside the repeat window of the previous resume is staff-resumable only,
         # and its email must not point the customer at a resume button that will refuse them.
         staff_only = pause_requires_staff(paused_at=now, paused_by=paused_by, resumed_at=flow.email_sending_resumed_at)
-        # A queryset update rather than save: the post_save signal publishes the worker reload
-        # immediately, and inside this transaction that is before the pause commits, so a worker
-        # could reload, read the still-unpaused row, and cache it for minutes. The explicit publish
-        # below waits for the commit; it is what stops in-flight runs and queued batch sends.
+        # A queryset update rather than save: it writes only the pause columns and skips the
+        # post_save reload, so the explicit publish below is the one reload. It waits for the
+        # commit; it is what stops in-flight runs and queued batch sends.
         HogFlow.objects.filter(id=hog_flow_id, team_id=team_id).update(
             email_sending_paused_at=now,
             email_sending_paused_reason=reason,
@@ -614,8 +613,7 @@ def resume_email_sending(
             resumed_at=locked.email_sending_resumed_at,
         ):
             raise StaffPausedError("Only PostHog staff can resume this pause.")
-        # Queryset update for the same reason as the pause writer: the reload must publish only
-        # once the resume is committed.
+        # Queryset update for the same reason as the pause writer.
         HogFlow.objects.filter(id=hog_flow_id, team_id=team_id).update(
             email_sending_paused_at=None,
             email_sending_paused_reason="",

products/workflows/backend/test/test_hog_flow.py

@@ -1,13 +1,17 @@
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
 
 from django.core.management import call_command
+from django.db import transaction
 from django.test import TestCase
 from django.utils import timezone
 
+from parameterized import parameterized
+from redis.exceptions import RedisError
+
 from posthog.models.user import User
 
 from products.actions.backend.models.action import Action
-from products.workflows.backend.models.hog_flow.hog_flow import HogFlow
+from products.workflows.backend.models.hog_flow.hog_flow import HogFlow, defer_hog_flow_reloads
 
 
 class TestHogFlow(TestCase):
@@ -20,19 +24,32 @@ def setUp(self):
 
     @patch("products.workflows.backend.models.hog_flow.hog_flow.reload_hog_flows_on_workers")
     def test_hog_flow_saved_receiver(self, mock_reload):
-        hog_flow = HogFlow.objects.create(name="Test Flow", team=self.team)
+        with self.captureOnCommitCallbacks(execute=True):
+            hog_flow = HogFlow.objects.create(name="Test Flow", team=self.team)
         mock_reload.assert_called_once_with(team_id=self.team.id, hog_flow_ids=[str(hog_flow.id)])
 
+    @patch("products.workflows.backend.models.hog_flow.hog_flow.reload_hog_flows_on_workers")
+    def test_deferred_saves_reload_once_after_the_commit(self, mock_reload):
+        with self.captureOnCommitCallbacks(execute=True):
+            with defer_hog_flow_reloads():
+                first = HogFlow.objects.create(name="First", team=self.team)
+                second = HogFlow.objects.create(name="Second", team=self.team)
+            mock_reload.assert_not_called()
+
+        mock_reload.assert_called_once_with(team_id=self.team.id, hog_flow_ids=sorted([str(first.id), str(second.id)]))
+
     @patch("products.workflows.backend.models.hog_flow.hog_flow.reload_hog_flows_on_workers")
     def test_draft_only_save_skips_worker_reload(self, mock_reload):
         hog_flow = HogFlow.objects.create(name="Test Flow", team=self.team)
         mock_reload.reset_mock()
 
         hog_flow.draft = {"name": "Draft name"}
-        hog_flow.save(update_fields=["draft", "draft_updated_at"])
+        with self.captureOnCommitCallbacks(execute=True):
+            hog_flow.save(update_fields=["draft", "draft_updated_at"])
         mock_reload.assert_not_called()
 
-        hog_flow.save(update_fields=["draft", "name"])
+        with self.captureOnCommitCallbacks(execute=True):
+            hog_flow.save(update_fields=["draft", "name"])
         mock_reload.assert_called_once_with(team_id=self.team.id, hog_flow_ids=[str(hog_flow.id)])
 
     @patch("products.workflows.backend.models.hog_flow.hog_flow.reload_hog_flows_on_workers")
@@ -44,10 +61,36 @@ def test_email_pause_save_reloads_the_workers(self, mock_reload):
 
         hog_flow.email_sending_paused_at = timezone.now()
         hog_flow.email_sending_paused_reason = "Spam complaints reached 2%."
-        hog_flow.save(update_fields=["email_sending_paused_at", "email_sending_paused_reason"])
+        with self.captureOnCommitCallbacks(execute=True):
+            hog_flow.save(update_fields=["email_sending_paused_at", "email_sending_paused_reason"])
+            mock_reload.assert_not_called()
 
         mock_reload.assert_called_once_with(team_id=self.team.id, hog_flow_ids=[str(hog_flow.id)])
 
+    @parameterized.expand([("save",), ("delete",)])
+    @patch("products.workflows.backend.models.hog_flow.hog_flow.capture_exception")
+    @patch("products.workflows.backend.models.hog_flow.hog_flow.reload_hog_flows_on_workers")
+    def test_a_failed_reload_publish_is_retried_reported_and_runs_the_callbacks_after_it(
+        self, change, mock_reload, mock_capture_exception
+    ):
+        hog_flow = HogFlow.objects.create(name="Test Flow", team=self.team)
+        hog_flow_id = str(hog_flow.id)
+        mock_reload.side_effect = RedisError("Redis is down")
+        callback_after_the_reload = MagicMock()
+
+        with self.captureOnCommitCallbacks(execute=True):
+            if change == "save":
+                hog_flow.name = "Renamed"
+                hog_flow.save(update_fields=["name"])
+            else:
+                hog_flow.delete()
+            transaction.on_commit(callback_after_the_reload)
+
+        assert mock_reload.call_count == 2
+        mock_reload.assert_called_with(team_id=self.team.id, hog_flow_ids=[hog_flow_id])
+        mock_capture_exception.assert_called_once()
+        callback_after_the_reload.assert_called_once()
+
     @patch("products.workflows.backend.tasks.hog_flows.refresh_affected_hog_flows.delay")
     def test_action_saved_receiver(self, mock_refresh):
         action = Action.objects.create(team=self.team, name="Test Action")

products/workflows/backend/test/test_refresh_hog_flows.py

@@ -106,7 +106,8 @@ def test_refresh_all_hog_flows(self, mock_reload):
         """Test refreshing all HogFlows across all teams."""
 
         out = StringIO()
-        call_command("refresh_hog_flows", stdout=out)
+        with self.captureOnCommitCallbacks(execute=True):
+            call_command("refresh_hog_flows", stdout=out)
 
         # Should have refreshed all 4 flows (including archived)
         assert mock_reload.call_count == 4
@@ -122,7 +123,8 @@ def test_refresh_by_team_id(self, mock_reload):
         """Test refreshing HogFlows for a specific team."""
 
         out = StringIO()
-        call_command("refresh_hog_flows", team_id=self.team.id, stdout=out)
+        with self.captureOnCommitCallbacks(execute=True):
+            call_command("refresh_hog_flows", team_id=self.team.id, stdout=out)
 
         # Should have refreshed flows from team1 (hog_flow1, hog_flow2, archived_flow)
         assert mock_reload.call_count == 3
@@ -137,7 +139,8 @@ def test_refresh_by_hog_flow_id(self, mock_reload):
         """Test refreshing a specific HogFlow by ID."""
 
         out = StringIO()
-        call_command("refresh_hog_flows", hog_flow_id=str(self.hog_flow1.id), stdout=out)
+        with self.captureOnCommitCallbacks(execute=True):
+            call_command("refresh_hog_flows", hog_flow_id=str(self.hog_flow1.id), stdout=out)
 
         # Should have refreshed only the specific flow
         assert mock_reload.call_count == 1
@@ -152,7 +155,8 @@ def test_nonexistent_team_id(self, mock_reload):
         """Test handling of nonexistent team ID."""
 
         out = StringIO()
-        call_command("refresh_hog_flows", team_id=99999, stdout=out)
+        with self.captureOnCommitCallbacks(execute=True):
+            call_command("refresh_hog_flows", team_id=99999, stdout=out)
 
         assert mock_reload.call_count == 0
 
@@ -167,7 +171,8 @@ def test_nonexistent_hog_flow_id(self, mock_reload):
         out = StringIO()
         # Use a valid UUID format that doesn't exist
         nonexistent_uuid = "00000000-0000-0000-0000-000000000000"
-        call_command("refresh_hog_flows", hog_flow_id=nonexistent_uuid, stdout=out)
+        with self.captureOnCommitCallbacks(execute=True):
+            call_command("refresh_hog_flows", hog_flow_id=nonexistent_uuid, stdout=out)
 
         assert mock_reload.call_count == 0
 
@@ -181,7 +186,8 @@ def test_page_size_option(self, mock_reload):
 
         out = StringIO()
         # Set page size to 2, should process all flows but in multiple pages
-        call_command("refresh_hog_flows", page_size=2, stdout=out)
+        with self.captureOnCommitCallbacks(execute=True):
+            call_command("refresh_hog_flows", page_size=2, stdout=out)
 
         # Should still refresh all 4 flows
         assert mock_reload.call_count == 4
@@ -224,17 +230,18 @@ def mock_save_method(self):
             self.assertIn("Errors: 1", output)  # 1 failure
             self.assertIn("Check logs for details on 1 errors encountered", output)
 
-    @patch("products.workflows.backend.models.hog_flow.hog_flow.reload_hog_flows_on_workers")
-    def test_keeps_an_edit_made_to_a_workflow_while_the_run_is_in_progress(self, mock_reload):
+    def test_keeps_an_edit_made_to_a_workflow_while_the_run_is_in_progress(self):
         first_saved: list[str] = []
+        save = HogFlow.save
 
-        def turn_off_the_rest_once(team_id: int, hog_flow_ids: list[str]) -> None:
+        def turn_off_the_rest_once(flow: HogFlow, *args: Any, **kwargs: Any) -> None:
+            save(flow, *args, **kwargs)
             if not first_saved:
-                first_saved.extend(hog_flow_ids)
-                HogFlow.objects.exclude(id__in=hog_flow_ids).update(status=HogFlow.State.DRAFT)
+                first_saved.append(str(flow.id))
+                HogFlow.objects.exclude(id=flow.id).update(status=HogFlow.State.DRAFT)
 
-        mock_reload.side_effect = turn_off_the_rest_once
-        call_command("refresh_hog_flows", stdout=StringIO())
+        with patch.object(HogFlow, "save", turn_off_the_rest_once):
+            call_command("refresh_hog_flows", stdout=StringIO())
 
         statuses = set(HogFlow.objects.exclude(id__in=first_saved).values_list("status", flat=True))
         assert statuses == {HogFlow.State.DRAFT}