feat(alerts): hold a platform delivery until its source releases it

.semgrep/rules/security/idor-team-scoped-models.yaml

@@ -426,6 +426,7 @@ rules:
                               |SubscriptionDelivery
                               |PlatformAlert
                               |PlatformAlertConfiguration
+                              |PlatformAlertDeliveryHold
                               |PlatformAlertDeliveryPath
                               |PlatformAlertThread
                               |AlertDestination
@@ -903,6 +904,7 @@ rules:
                         |SubscriptionDelivery
                         |PlatformAlert
                         |PlatformAlertConfiguration
+                        |PlatformAlertDeliveryHold
                         |PlatformAlertDeliveryPath
                         |PlatformAlertThread
                         |AlertDestination

docs/internal/alerts-platform-temporal.md

@@ -39,7 +39,7 @@ The schedule starts `alerts-platform-orchestrate` with `{}` on the orchestration
 There is no routing flag: the flow is tick → orchestration → evaluation → delivery.
 It uses SKIP overlap, a one-minute catchup window, a 50-second workflow execution timeout,
 and one workflow attempt. Creation does not trigger an immediate run; the next minute starts it.
-Delivery has no schedule: evaluation starts its delivery child.
+Delivery has no schedule of its own: evaluation starts its delivery child, and the [held-delivery sweep](#held-deliveries) starts the ones it releases.
 New schedules start unpaused. Registration updates existing schedules to this policy while retaining their state from Temporal, including manual pauses.
 To stop future ticks, pause the schedule in Temporal; resume it there when ready. Pausing does not stop workflows already running.
 Reverting the code that registers a schedule does not remove one already registered.
@@ -541,6 +541,29 @@ It switches the copies off and keeps their rows, state and history. Checks alrea
 Running the backfill again turns them back on at the production alert's next due time.
 An hourly alert on the platform checks on a UTC grid, while production checks it at the alert's creation minute, so the two stacks check an hourly alert at different minutes.
 
+## Held deliveries
+
+A source can hold an evaluation's delivery instead of starting it, for example while an agent investigates a firing.
+It puts the request on `PlatformAlertOutcome.held_delivery` with a reason and a deadline, and leaves it out of the deliveries it starts.
+`record_outcomes` writes the hold to `PlatformAlertDeliveryHold` in the same transaction as the outcome.
+
+`release_hold` in the facade stamps a hold once, under a row lock.
+It returns the request only for a release with `send`, and only to the releaser that won.
+That releaser starts the delivery with `start_deliveries`, so the delivery keeps its usual id, then calls `record_hold_delivery_started`.
+A retry from the same releaser gets the request back. Another releaser gets nothing.
+`reschedule_hold` moves an unreleased hold's deadline.
+`deliver_evaluation` also refuses an evaluation whose hold is unreleased or released with `suppress`.
+
+`alerts-platform-release-expired-holds-schedule` starts `alerts-platform-release-expired-holds` every minute on the delivery queue, with SKIP overlap and a 5-minute execution timeout.
+The workflow finds up to 200 holds past their deadline, releases each with `send` as the safety net, and starts their deliveries.
+It re-reads each hold's deadline and configuration under the row lock, so a hold rescheduled after the find stays held. It skips a hold another transaction has locked, and a later run finds it again.
+It also starts any released send whose delivery nobody recorded as started 5 minutes after the release, so a releaser that failed between the release and the start still delivers. It starts those with `ALLOW_DUPLICATE_FAILED_ONLY`, so a delivery that completed and lost only its start mark does not run again.
+`alerts_platform_holds_expired_total` counts only holds a run released at their deadline, not recovered starts or retries.
+It stops adding requests once they would outgrow one activity payload, and leaves the rest for the next run.
+It suppresses, rather than sends, the hold of a disabled configuration, so turning the alert back on never sends an old firing. It still sends a hold whose alert went back to its source, because the request records the owner at check time.
+The find step also deletes holds older than 90 days, which is how long the history rows they annotate live.
+A rise in `alerts_platform_holds_expired_total` means a source holds deliveries and does not release them.
+
 ## Postgres connectivity probe
 
 Each evaluation activity issues one explicit `SELECT 1` and checks for `(1,)` through Django's `default` main writer connection.

posthog/temporal/schedule.py

@@ -80,6 +80,7 @@
 from posthog.temporal.weekly_digest.types import WeeklyDigestInput
 
 from products.alerts_platform.backend.facade.temporal import (
+    create_alerts_platform_expired_holds_schedule,
     create_alerts_platform_inventory_schedule,
     create_alerts_platform_tick_schedule,
 )
@@ -995,6 +996,7 @@ async def create_error_tracking_recommendations_refresh_schedule(client: Client)
     create_data_catalog_weekly_digest_schedule,
     create_alerts_platform_tick_schedule,
     create_alerts_platform_inventory_schedule,
+    create_alerts_platform_expired_holds_schedule,
     create_logs_alert_check_schedule,
     disable_logs_volume_tick_schedule,
     create_schedule_due_alert_checks_schedule,

products/alerts_platform/backend/delivery/evaluation.py

@@ -44,6 +44,7 @@
     LegacyEventEmission,
     MessageLink,
 )
+from products.alerts_platform.backend.logic import delivery_holds
 from products.alerts_platform.backend.logic.platform_alert_events import announced_kinds, announcement
 from products.alerts_platform.backend.temporal.metrics import DeliveryOutcomeLabel, increment_deliveries, safe_record
 
@@ -156,11 +157,18 @@
         )
 
 
 def deliver_evaluation(request: AlertDeliveryRequest) -> DeliveryOutcome:
     """Sends what one evaluation decided, to every destination configured for it."""
     team = Team.objects.filter(id=request.team_id).first()
     if team is None or not (request.owned_by_platform or destinations_are_live(team)):
         return DeliveryOutcome(live=False, sent=0, skipped_without_transport=0)
+    if delivery_holds.blocks_delivery(request.team_id, request.configuration_id, request.evaluation_key):
+        logger.info(
+            "alerts_platform_delivery_held",
+            configuration_id=request.configuration_id,
+            evaluation_key=request.evaluation_key,
+        )
+        return DeliveryOutcome(live=True, sent=0, skipped_without_transport=0)
 
     announced = announcement(
         request.team_id,

products/alerts_platform/backend/facade/api.py

@@ -12,8 +12,10 @@
 from uuid import UUID
 
 from products.alerts_platform.backend.facade.contracts import (
+    AlertDeliveryRequest,
     DestinationSend,
     HandbackRefusal,
+    HeldDeliveryAddress,
     NewAlertDestination,
     PlatformAlertCheckInput,
     PlatformAlertConfigurationEdit,
@@ -32,8 +34,10 @@
     SourceKind,
     StoredAlertDestination,
 )
+from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy
 from products.alerts_platform.backend.logic import (
     alert_destinations,
+    delivery_holds,
     destination_sends,
     platform_lifecycle,
     platform_reads,
@@ -60,6 +64,25 @@ def record_outcomes(team_id: int, outcomes: Sequence[PlatformAlertOutcome], now:
     return platform_lifecycle.record_outcomes(team_id, outcomes, now)
 
 
+def release_hold(
+    address: HeldDeliveryAddress, *, action: HoldRelease, released_by: HoldReleasedBy, now: datetime
+) -> AlertDeliveryRequest | None:
+    """Releases a held delivery. Returns the request when the caller should start its delivery,
+    which it does with `start_deliveries` so the delivery keeps its usual id."""
+    return delivery_holds.release_hold(address, action=action, released_by=released_by, now=now)
+
+
+def record_hold_delivery_started(address: HeldDeliveryAddress, *, now: datetime) -> None:
+    """Marks a released send's delivery as started. Call it after `start_deliveries` returns, or the
+    safety net starts the delivery again once a grace period passes, and the delivery's id dedups it."""
+    delivery_holds.record_delivery_started([address], now)
+
+
+def reschedule_hold(address: HeldDeliveryAddress, *, release_after: datetime) -> bool:
+    """Moves an unreleased hold's safety-net deadline. Returns False when nothing was held."""
+    return delivery_holds.reschedule_hold(address, release_after=release_after)
+
+
 def slot_of(next_check_at: datetime | None, cutoff: datetime) -> str:
     """The minute a configuration is due for."""
     return platform_lifecycle.slot_of(next_check_at, cutoff)

products/alerts_platform/backend/facade/contracts.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable, Collection, Mapping, Sequence
-from dataclasses import field
+from dataclasses import asdict, field
 from datetime import UTC, datetime
 from enum import StrEnum
 from typing import TYPE_CHECKING, Any, Final, NotRequired, Protocol, TypedDict
@@ -631,6 +631,9 @@ class PlatformAlertOutcome:
     # groups of a stale check are keyed for a grouping the configuration no longer has, so they are
     # dropped. None when the outcome carries no groups to drop.
     grouping: Grouping | None = None
+    # The delivery this check decided on, held back until its source or the safety net releases it.
+    # The source leaves it out of the deliveries it starts.
+    held_delivery: HeldDelivery | None = None
 
     @classmethod
     def ungrouped(
@@ -704,6 +707,12 @@ def __post_init__(self) -> None:
             raise ValueError("an outcome names each group once")
         if set(self.evicted) & set(keys):
             raise ValueError("an outcome cannot evict a group it reports")
+        held = self.held_delivery
+        if held is not None and (
+            held.request.configuration_id != str(self.configuration_id)
+            or held.request.evaluation_key != self.evaluation_key
+        ):
+            raise ValueError("an outcome can only hold its own delivery")
 
 
 @frozen
@@ -822,6 +831,50 @@ class AlertDeliveryRequest:
     parts: int = 1
     part: int = 0
 
+    def to_stored(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_stored(cls, stored: Mapping[str, Any]) -> AlertDeliveryRequest:
+        # The converter Temporal decodes this type with, so the stored form loads the way the wire form does.
+        import temporalio.converter  # noqa: PLC0415 — keeps temporalio off django.setup(), which loads this via the models
+
+        return temporalio.converter.value_to_type(cls, dict(stored))
+
+
+MAX_HOLD_REASON_LENGTH: Final = 64
+
+
+@frozen
+class HeldDelivery:
+    """A delivery a check decided on and held back until something releases it.
+
+    The source releases it with a verdict, or with the request's send when it cannot run what it
+    held for. The platform's safety net sends it once `release_after_seconds` have passed since the
+    check was recorded, so a release that never comes still delivers.
+    """
+
+    request: AlertDeliveryRequest
+    # Why it is held, which the platform stores and does not read.
+    reason: str
+    release_after_seconds: int
+
+    def __post_init__(self) -> None:
+        # Rejected here rather than by the hold's column, where it would roll back the whole batch's record.
+        if len(self.reason) > MAX_HOLD_REASON_LENGTH:
+            raise ValueError(f"a hold reason is at most {MAX_HOLD_REASON_LENGTH} characters")
+        if self.release_after_seconds < 0:
+            raise ValueError("a hold cannot release before it is recorded")
+
+
+@frozen
+class HeldDeliveryAddress:
+    """The one hold an evaluation of a configuration can have."""
+
+    team_id: int
+    configuration_id: str
+    evaluation_key: str
+
 
 @frozen
 class SourceBatchEvaluation:

products/alerts_platform/backend/facade/enums.py

@@ -58,3 +58,22 @@ class PlatformAlertConfigurationRecurrenceUnit(LabeledStrEnum):
     DAY = "day", "Day"
     WEEK = "week", "Week"
     MONTH = "month", "Month"
+
+
+class HoldRelease(LabeledStrEnum):
+    """What releasing a held delivery does with it."""
+
+    SEND = "send", "Send"
+    SUPPRESS = "suppress", "Suppress"
+
+
+class HoldReleasedBy(LabeledStrEnum):
+    """Who released a held delivery.
+
+    A release returns the request to a retry only from the caller that released it, so the agent
+    and the safety net cannot both send one evaluation.
+    """
+
+    VERDICT = "verdict", "Verdict"
+    CLAIM_REFUSED = "claim_refused", "Claim refused"
+    SAFETY_NET = "safety_net", "Safety net"

products/alerts_platform/backend/facade/temporal.py

@@ -3,6 +3,7 @@
     ALERTS_PLATFORM_LATENCY_HISTOGRAM_METRICS,
 )
 from products.alerts_platform.backend.temporal.schedule import (
+    create_alerts_platform_expired_holds_schedule,
     create_alerts_platform_inventory_schedule,
     create_alerts_platform_tick_schedule,
 )
@@ -30,6 +31,7 @@
     "SHARED_ORCHESTRATION_ACTIVITIES",
     "SHARED_ORCHESTRATION_WORKFLOWS",
     "AlertsPlatformTelemetryInterceptor",
+    "create_alerts_platform_expired_holds_schedule",
     "create_alerts_platform_inventory_schedule",
     "create_alerts_platform_tick_schedule",
     "source_evaluation_timeout",