From 806a1450aa18e113ca1fa7e09567de215f4aa40a Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 26 Oct 2026 11:48:13 -0700 Subject: [PATCH 1/6] feat(alerts): hold a platform delivery until its source releases it A source can now hold an evaluation's delivery instead of starting it. record_outcomes writes the hold in the outcome's transaction, a release hands the request back to one winner, and a per-minute sweep sends any hold past its deadline. Co-Authored-By: Claude Opus 5.5 --- docs/internal/alerts-platform-temporal.md | 21 ++- posthog/temporal/schedule.py | 2 + .../backend/delivery/evaluation.py | 8 + .../alerts_platform/backend/facade/api.py | 17 ++ .../backend/facade/contracts.py | 55 +++++- .../alerts_platform/backend/facade/enums.py | 19 ++ .../backend/facade/temporal.py | 2 + .../backend/logic/delivery_holds.py | 144 +++++++++++++++ .../backend/logic/platform_lifecycle.py | 17 +- .../backend/migrations/0020_delivery_hold.py | 92 ++++++++++ .../backend/migrations/max_migration.txt | 2 +- .../backend/models/__init__.py | 2 + .../backend/models/platform_alert.py | 48 ++++- .../backend/temporal/metrics.py | 11 ++ .../backend/temporal/schedule.py | 18 ++ .../backend/temporal/workflows.py | 75 +++++++- .../backend/tests/test_delivery_evaluation.py | 39 +++- .../backend/tests/test_delivery_holds.py | 166 ++++++++++++++++++ 18 files changed, 728 insertions(+), 10 deletions(-) create mode 100644 products/alerts_platform/backend/logic/delivery_holds.py create mode 100644 products/alerts_platform/backend/migrations/0020_delivery_hold.py create mode 100644 products/alerts_platform/backend/tests/test_delivery_holds.py diff --git a/docs/internal/alerts-platform-temporal.md b/docs/internal/alerts-platform-temporal.md index 3d66f5fd84d6..e062a1d03f59 100644 --- a/docs/internal/alerts-platform-temporal.md +++ b/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,25 @@ 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. +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 skips the holds of a disabled configuration. 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. +`alerts_platform_holds_expired_total` counts the holds the safety net sent. A rise 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. diff --git a/posthog/temporal/schedule.py b/posthog/temporal/schedule.py index 93e870daa291..f37ecfd8acb3 100644 --- a/posthog/temporal/schedule.py +++ b/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, diff --git a/products/alerts_platform/backend/delivery/evaluation.py b/products/alerts_platform/backend/delivery/evaluation.py index 31fa353332e0..5b03f967d083 100644 --- a/products/alerts_platform/backend/delivery/evaluation.py +++ b/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 @@ -161,6 +162,13 @@ def deliver_evaluation(request: AlertDeliveryRequest) -> DeliveryOutcome: 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, diff --git a/products/alerts_platform/backend/facade/api.py b/products/alerts_platform/backend/facade/api.py index 0935bc76db70..fd8ea6e8c0e7 100644 --- a/products/alerts_platform/backend/facade/api.py +++ b/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,19 @@ 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 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) diff --git a/products/alerts_platform/backend/facade/contracts.py b/products/alerts_platform/backend/facade/contracts.py index dcc95448bac2..e95ee578d8a0 100644 --- a/products/alerts_platform/backend/facade/contracts.py +++ b/products/alerts_platform/backend/facade/contracts.py @@ -8,12 +8,14 @@ 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 from uuid import UUID +import temporalio.converter + from posthog.dataclasses import frozen from posthog.enums import LabeledStrEnum @@ -631,6 +633,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 +709,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 +833,48 @@ 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. + 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: diff --git a/products/alerts_platform/backend/facade/enums.py b/products/alerts_platform/backend/facade/enums.py index 6507a5cb74d2..0ede6bb3fd6d 100644 --- a/products/alerts_platform/backend/facade/enums.py +++ b/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" diff --git a/products/alerts_platform/backend/facade/temporal.py b/products/alerts_platform/backend/facade/temporal.py index ad9891d7cc0b..b0aab84288c4 100644 --- a/products/alerts_platform/backend/facade/temporal.py +++ b/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", diff --git a/products/alerts_platform/backend/logic/delivery_holds.py b/products/alerts_platform/backend/logic/delivery_holds.py new file mode 100644 index 000000000000..a0ab12763d5d --- /dev/null +++ b/products/alerts_platform/backend/logic/delivery_holds.py @@ -0,0 +1,144 @@ +"""Deliveries a source held back, and the releases that decide whether they go out. + +A source holds an evaluation's delivery while it works out whether the delivery should be sent, +for example while an agent investigates a firing. The platform knows nothing of why. It keeps the +request, hands it back to whoever releases it with a send, and sends it itself once the hold's +deadline passes, so a source that never answers still delivers. +""" + +from collections.abc import Sequence +from datetime import datetime, timedelta +from typing import Final + +from django.db import transaction + +from products.alerts_platform.backend.facade.contracts import ( + AlertDeliveryRequest, + HeldDeliveryAddress, + PlatformAlertOutcome, +) +from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy +from products.alerts_platform.backend.models import PlatformAlertConfiguration, PlatformAlertDeliveryHold +from products.alerts_platform.backend.models.platform_alert_events_sql import PLATFORM_ALERT_EVENTS_TTL_DAYS + +# As long as the history rows a hold annotates live, so a reader never finds one without the other. +HOLD_RETENTION: Final = timedelta(days=PLATFORM_ALERT_EVENTS_TTL_DAYS) + + +def hold_for( + configuration: PlatformAlertConfiguration, outcome: PlatformAlertOutcome, now: datetime +) -> PlatformAlertDeliveryHold | None: + """The hold an outcome asks for, unsaved, so `record_outcomes` writes it in its own transaction.""" + held = outcome.held_delivery + if held is None: + return None + return PlatformAlertDeliveryHold( + team_id=configuration.team_id, + configuration_id=configuration.id, + evaluation_key=outcome.evaluation_key, + reason=held.reason, + request=held.request.to_stored(), + held_at=now, + release_after=now + timedelta(seconds=held.release_after_seconds), + ) + + +def release_hold( + address: HeldDeliveryAddress, *, action: HoldRelease, released_by: HoldReleasedBy, now: datetime +) -> AlertDeliveryRequest | None: + """Releases one hold. Returns the request when the caller should start its delivery. + + The first release wins. A later one changes nothing and gets the request back only when it + repeats the winning send from the same releaser, which is a retry of a caller that committed + the release and failed before it started the delivery. The delivery's id then dedups it. + """ + with transaction.atomic(): + hold = ( + PlatformAlertDeliveryHold.objects.for_team(address.team_id) + .select_for_update() + .filter(configuration_id=address.configuration_id, evaluation_key=address.evaluation_key) + .first() + ) + if hold is None: + return None + if hold.released_at is not None: + repeated = hold.release_action == action == HoldRelease.SEND and hold.released_by == released_by + return AlertDeliveryRequest.from_stored(hold.request) if repeated else None + hold.released_at = now + hold.release_action = action.value + hold.released_by = released_by.value + hold.save(update_fields=["released_at", "release_action", "released_by"]) + return AlertDeliveryRequest.from_stored(hold.request) if action == HoldRelease.SEND else None + + +def reschedule_hold(address: HeldDeliveryAddress, *, release_after: datetime) -> bool: + """Moves an unreleased hold's deadline. Returns False when the hold is gone or already released. + + A source calls this in the transaction that records why it has not released yet, so the + deadline cannot pass between the two. + """ + updated = ( + PlatformAlertDeliveryHold.objects.for_team(address.team_id) + .filter( + configuration_id=address.configuration_id, + evaluation_key=address.evaluation_key, + released_at__isnull=True, + ) + .update(release_after=release_after) + ) + return updated > 0 + + +def blocks_delivery(team_id: int, configuration_id: str, evaluation_key: str) -> bool: + """Whether an evaluation has a hold that is not released with a send. + + A second guard behind the source, which leaves a held request out of the deliveries it starts. + """ + return ( + PlatformAlertDeliveryHold.objects.for_team(team_id) + .filter(configuration_id=configuration_id, evaluation_key=evaluation_key) + # An unreleased hold has no action, and the exclusion keeps it. + .exclude(release_action=HoldRelease.SEND) + .exists() + ) + + +def prune_holds(now: datetime, *, limit: int) -> int: + """Deletes up to `limit` holds older than the history rows they annotate. Returns how many.""" + stale = list( + PlatformAlertDeliveryHold.objects.unscoped() + .filter(held_at__lt=now - HOLD_RETENTION) + .values_list("id", flat=True)[:limit] + ) + return PlatformAlertDeliveryHold.objects.unscoped().filter(id__in=stale).delete()[0] if stale else 0 + + +def expired_holds(now: datetime, *, limit: int) -> tuple[HeldDeliveryAddress, ...]: + """The holds past their deadline, oldest first. + + Every team's, so one sweep covers the fleet. A disabled configuration's holds are left held, + the way the source's own safety net skips a disabled alert. An alert rolled back to its source + is still swept, because its held request was decided while the platform owned it. + """ + rows = ( + PlatformAlertDeliveryHold.objects.unscoped() + .filter(released_at__isnull=True, release_after__lte=now, configuration__enabled=True) + .order_by("release_after") + .values_list("team_id", "configuration_id", "evaluation_key")[:limit] + ) + return tuple( + HeldDeliveryAddress(team_id=team_id, configuration_id=str(configuration_id), evaluation_key=key) + for team_id, configuration_id, key in rows + ) + + +def release_expired(addresses: Sequence[HeldDeliveryAddress], now: datetime) -> tuple[AlertDeliveryRequest, ...]: + """Releases each hold with a send from the safety net, and returns the requests it won. + + One transaction per hold, so a hold another releaser locked delays only itself. + """ + released = ( + release_hold(address, action=HoldRelease.SEND, released_by=HoldReleasedBy.SAFETY_NET, now=now) + for address in addresses + ) + return tuple(request for request in released if request is not None) diff --git a/products/alerts_platform/backend/logic/platform_lifecycle.py b/products/alerts_platform/backend/logic/platform_lifecycle.py index 5e9b6f502121..627bbb3cdf53 100644 --- a/products/alerts_platform/backend/logic/platform_lifecycle.py +++ b/products/alerts_platform/backend/logic/platform_lifecycle.py @@ -49,8 +49,14 @@ to_recurrence_interval, validate_and_normalize_schedule_start_time, ) +from products.alerts_platform.backend.logic import delivery_holds from products.alerts_platform.backend.logic.platform_alert_events import PlatformAlertEventRow, insert_events -from products.alerts_platform.backend.models import PlatformAlert, PlatformAlertConfiguration, PlatformAlertThread +from products.alerts_platform.backend.models import ( + PlatformAlert, + PlatformAlertConfiguration, + PlatformAlertDeliveryHold, + PlatformAlertThread, +) def due_q(moment: datetime) -> Q: @@ -479,7 +485,7 @@ def _without_stale_groups( return outcome if outcome.grouping.splits_like(Grouping.from_stored(configuration.grouping)): return outcome - return replace(outcome, groups=(), overflowed=0, overflow_is_lower_bound=False, evicted=()) + return replace(outcome, groups=(), overflowed=0, overflow_is_lower_bound=False, evicted=(), held_delivery=None) _HELD_BY_MUTE: Final = { @@ -562,8 +568,12 @@ def record_outcomes(team_id: int, outcomes: Sequence[PlatformAlertOutcome], now: team_timezone = Team.objects.filter(id=team_id).values_list("timezone", flat=True).first() or "UTC" rows: list[PlatformAlertEventRow] = [] + holds: list[PlatformAlertDeliveryHold] = [] for configuration in configurations: outcome = by_id[str(configuration.id)] + hold = delivery_holds.hold_for(configuration, outcome, now) + if hold is not None: + holds.append(hold) turned_away = 0 for reported in outcome.groups: alert = alerts.get( @@ -649,6 +659,9 @@ def record_outcomes(team_id: int, outcomes: Sequence[PlatformAlertOutcome], now: "overflow_is_lower_bound", ], ) + # In this transaction, because a recorded firing whose hold was lost would never be sent. A + # conflict keeps the first hold, so a repeated evaluation key cannot fail every later record. + PlatformAlertDeliveryHold.objects.for_team(team_id).bulk_create(holds, ignore_conflicts=True) # `on_commit` rather than a statement after the block, so a caller that wraps this in its # own `atomic()` cannot leave history for state its rollback removed. transaction.on_commit(lambda: record_history(team_id, rows)) diff --git a/products/alerts_platform/backend/migrations/0020_delivery_hold.py b/products/alerts_platform/backend/migrations/0020_delivery_hold.py new file mode 100644 index 000000000000..d18b51c64e22 --- /dev/null +++ b/products/alerts_platform/backend/migrations/0020_delivery_hold.py @@ -0,0 +1,92 @@ +# Generated by Django 5.2.18 on 2026-10-26 18:31 + +import django.db.models.deletion +from django.db import migrations, models + +import posthog.uuidt + + +class Migration(migrations.Migration): + dependencies = [ + ("alerts_platform", "0019_overflow_and_instance_labels"), + ("posthog", "1413_event_filter_pipeline_override"), + ] + + operations = [ + migrations.CreateModel( + name="PlatformAlertDeliveryHold", + fields=[ + ( + "id", + models.UUIDField( + default=posthog.uuidt.uuid7, + editable=False, + primary_key=True, + serialize=False, + ), + ), + ("evaluation_key", models.CharField(max_length=255)), + ("reason", models.CharField(max_length=64)), + ("request", models.JSONField()), + ("held_at", models.DateTimeField()), + ("release_after", models.DateTimeField()), + ("released_at", models.DateTimeField(blank=True, null=True)), + ( + "release_action", + models.CharField( + blank=True, + choices=[("send", "Send"), ("suppress", "Suppress")], + max_length=16, + null=True, + ), + ), + ( + "released_by", + models.CharField( + blank=True, + choices=[ + ("verdict", "Verdict"), + ("claim_refused", "Claim refused"), + ("safety_net", "Safety net"), + ], + max_length=16, + null=True, + ), + ), + ( + "configuration", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="delivery_holds", + to="alerts_platform.platformalertconfiguration", + ), + ), + ( + "team", + models.ForeignKey( + db_constraint=False, + on_delete=django.db.models.deletion.CASCADE, + related_name="+", + to="posthog.team", + ), + ), + ], + options={ + "db_table": "alerts_platformalertdeliveryhold", + "indexes": [ + models.Index( + condition=models.Q(("released_at__isnull", True)), + fields=["release_after"], + name="platform_alert_hold_due_idx", + ), + models.Index(fields=["held_at"], name="platform_alert_hold_age_idx"), + ], + "constraints": [ + models.UniqueConstraint( + fields=("configuration", "evaluation_key"), + name="platform_alert_delivery_hold_one_per_evaluation", + ) + ], + }, + ), + ] diff --git a/products/alerts_platform/backend/migrations/max_migration.txt b/products/alerts_platform/backend/migrations/max_migration.txt index fe91e7572cbf..cb00bee8cb2f 100644 --- a/products/alerts_platform/backend/migrations/max_migration.txt +++ b/products/alerts_platform/backend/migrations/max_migration.txt @@ -1 +1 @@ -0019_overflow_and_instance_labels +0020_delivery_hold diff --git a/products/alerts_platform/backend/models/__init__.py b/products/alerts_platform/backend/models/__init__.py index 5ca17dc57461..04c4493390a9 100644 --- a/products/alerts_platform/backend/models/__init__.py +++ b/products/alerts_platform/backend/models/__init__.py @@ -2,6 +2,7 @@ AlertDestination, PlatformAlert, PlatformAlertConfiguration, + PlatformAlertDeliveryHold, PlatformAlertDeliveryPath, PlatformAlertThread, ) @@ -10,6 +11,7 @@ "AlertDestination", "PlatformAlert", "PlatformAlertConfiguration", + "PlatformAlertDeliveryHold", "PlatformAlertDeliveryPath", "PlatformAlertThread", ] diff --git a/products/alerts_platform/backend/models/platform_alert.py b/products/alerts_platform/backend/models/platform_alert.py index 83c05b8031b5..d917a1fda8f4 100644 --- a/products/alerts_platform/backend/models/platform_alert.py +++ b/products/alerts_platform/backend/models/platform_alert.py @@ -18,9 +18,11 @@ from posthog.models.scoping.root_mixin import TeamScopedRootMixin from posthog.models.utils import UUIDModel -from products.alerts_platform.backend.facade.contracts import DestinationType, Grouping +from products.alerts_platform.backend.facade.contracts import MAX_HOLD_REASON_LENGTH, DestinationType, Grouping from products.alerts_platform.backend.facade.enums import ( DeliveryPath, + HoldRelease, + HoldReleasedBy, PlatformAlertCheckStatus, PlatformAlertConfigurationRecurrenceUnit, PlatformAlertConfigurationSourceKind, @@ -274,6 +276,50 @@ class Meta: ] +class PlatformAlertDeliveryHold(TeamScopedRootMixin, UUIDModel): + """One evaluation's delivery, held back until a release says whether to send it. + + Written in the same transaction as the outcome that held it, because a recorded firing whose + hold was lost would be neither sent nor swept. Released once: the first release wins, and only + a send hands the request back, so two releasers cannot both deliver it. + """ + + team = models.ForeignKey("posthog.Team", on_delete=models.CASCADE, db_constraint=False, related_name="+") + configuration = models.ForeignKey( + PlatformAlertConfiguration, on_delete=models.CASCADE, related_name="delivery_holds" + ) + + evaluation_key = models.CharField(max_length=255) + reason = models.CharField(max_length=MAX_HOLD_REASON_LENGTH) + # The stored form of the facade's `AlertDeliveryRequest`. + request = models.JSONField() + + held_at = models.DateTimeField() + # When the safety net sends it if nothing released it first. + release_after = models.DateTimeField() + released_at = models.DateTimeField(null=True, blank=True) + release_action = models.CharField(max_length=16, choices=HoldRelease.choices, null=True, blank=True) + released_by = models.CharField(max_length=16, choices=HoldReleasedBy.choices, null=True, blank=True) + + class Meta: + db_table = "alerts_platformalertdeliveryhold" + constraints = [ + models.UniqueConstraint( + fields=["configuration", "evaluation_key"], name="platform_alert_delivery_hold_one_per_evaluation" + ) + ] + indexes = [ + # The safety net's sweep. Partial, because a released hold is never swept again. + models.Index( + fields=["release_after"], + name="platform_alert_hold_due_idx", + condition=models.Q(released_at__isnull=True), + ), + # Pruning, which deletes holds once the history rows they annotate have expired. + models.Index(fields=["held_at"], name="platform_alert_hold_age_idx"), + ] + + class AlertDestination(TeamScopedRootMixin, UUIDModel): """Where an alert sends, for an alert whose `destinations_owner` is the platform. diff --git a/products/alerts_platform/backend/temporal/metrics.py b/products/alerts_platform/backend/temporal/metrics.py index ecc850bede78..db70b513de88 100644 --- a/products/alerts_platform/backend/temporal/metrics.py +++ b/products/alerts_platform/backend/temporal/metrics.py @@ -184,6 +184,17 @@ def increment_deliveries_previewed(source: str) -> None: ).add(1) +def increment_holds_expired(source: str) -> None: + """Held deliveries the safety net sent because nothing released them before their deadline. + + Each one is a source that held a delivery and never answered, so a rise means a stalled source. + """ + get_metric_meter({"source": source}).create_counter( + "alerts_platform_holds_expired_total", + "Held deliveries the safety net sent after their deadline passed", + ).add(1) + + def increment_outcomes_recorded(count: int) -> None: get_metric_meter().create_counter( "alerts_platform_outcomes_recorded_total", diff --git a/products/alerts_platform/backend/temporal/schedule.py b/products/alerts_platform/backend/temporal/schedule.py index 1f58f047bdc8..cab628a54701 100644 --- a/products/alerts_platform/backend/temporal/schedule.py +++ b/products/alerts_platform/backend/temporal/schedule.py @@ -13,6 +13,7 @@ SCHEDULE_ID = "alerts-platform-check-due-schedule" INVENTORY_SCHEDULE_ID = "alerts-platform-record-inventory-schedule" +EXPIRED_HOLDS_SCHEDULE_ID = "alerts-platform-release-expired-holds-schedule" async def create_alerts_platform_tick_schedule(client: "Client") -> None: @@ -50,6 +51,23 @@ async def create_alerts_platform_inventory_schedule(client: "Client") -> None: await _create_or_update(client, INVENTORY_SCHEDULE_ID, schedule) +async def create_alerts_platform_expired_holds_schedule(client: "Client") -> None: + """Sends held deliveries nothing released. A skipped run is harmless, because the next finds the same holds.""" + schedule = Schedule( + action=ScheduleActionStartWorkflow( + "alerts-platform-release-expired-holds", + {}, + id=EXPIRED_HOLDS_SCHEDULE_ID, + task_queue=settings.ALERTS_PLATFORM_DELIVERY_TASK_QUEUE, + execution_timeout=dt.timedelta(minutes=5), + retry_policy=RetryPolicy(maximum_attempts=1), + ), + spec=ScheduleSpec(cron_expressions=["* * * * *"]), + policy=SchedulePolicy(overlap=ScheduleOverlapPolicy.SKIP, catchup_window=dt.timedelta(minutes=1)), + ) + await _create_or_update(client, EXPIRED_HOLDS_SCHEDULE_ID, schedule) + + async def _create_or_update(client: "Client", schedule_id: str, schedule: Schedule) -> None: if await a_schedule_exists(client, schedule_id): description = await client.get_schedule_handle(schedule_id).describe() diff --git a/products/alerts_platform/backend/temporal/workflows.py b/products/alerts_platform/backend/temporal/workflows.py index 608b20b16ad6..7d16ed67881c 100644 --- a/products/alerts_platform/backend/temporal/workflows.py +++ b/products/alerts_platform/backend/temporal/workflows.py @@ -22,6 +22,7 @@ from products.alerts_platform.backend.temporal.metrics import ( increment_deliveries_previewed, increment_discovery_omitted, + increment_holds_expired, safe_record, ) from products.alerts_platform.backend.temporal.outcomes import alerts_platform_record_outcomes_activity @@ -42,6 +43,7 @@ AlertDeliveryRequest, AlertDemand, DemandDiscoveryInputs, + HeldDeliveryAddress, OrchestrateInputs, OrchestrateResult, SourceDispatchInputs, @@ -49,6 +51,7 @@ SourceEvaluationInputs, TickPage, ) + from products.alerts_platform.backend.logic import delivery_holds from products.alerts_platform.backend.logic.demand import discover_demand from products.alerts_platform.backend.logic.inventory import count_inventory from products.alerts_platform.backend.temporal.inventory_metrics import push_inventory @@ -294,6 +297,70 @@ async def start_deliveries(deliveries: Sequence[AlertDeliveryRequest]) -> int: return started +# Holds one sweep releases. A sweep that finds more leaves the rest to the next minute's run. +MAX_HOLDS_PER_SWEEP = 200 + + +@activity.defn +async def alerts_platform_find_expired_holds_activity() -> list[HeldDeliveryAddress]: + now = dt.datetime.now(dt.UTC) + # A failed prune is logged, so cleanup never keeps a due hold from being sent. + try: + pruned = await database_sync_to_async_pool(delivery_holds.prune_holds)(now, limit=MAX_HOLDS_PER_SWEEP) + except Exception as error: + await LOGGER.awarning("alerts_platform_holds_prune_failed", error=str(error)) + else: + if pruned: + await LOGGER.ainfo("alerts_platform_holds_pruned", count=pruned) + return list(await database_sync_to_async_pool(delivery_holds.expired_holds)(now, limit=MAX_HOLDS_PER_SWEEP)) + + +@activity.defn +async def alerts_platform_release_expired_holds_activity( + addresses: list[HeldDeliveryAddress], +) -> list[AlertDeliveryRequest]: + """Separate from the find, so a retry after a committed release gets the same requests back.""" + released = await database_sync_to_async_pool(delivery_holds.release_expired)(addresses, dt.datetime.now(dt.UTC)) + for request in released: + await LOGGER.awarning( + "alerts_platform_hold_expired", + source=request.source.value, + configuration_id=request.configuration_id, + evaluation_key=request.evaluation_key, + ) + safe_record(increment_holds_expired, request.source.value) + return list(released) + + +@workflow.defn(name="alerts-platform-release-expired-holds") +class AlertsPlatformReleaseExpiredHoldsWorkflow(PostHogWorkflow): + """The safety net for held deliveries: sends each one whose deadline passed with no release. + + One release of a hold wins, so a source that releases it at the same moment cannot also send it. + The delivery keeps its usual id, so a retried run that starts it again starts nothing new. + """ + + inputs_cls = AlertsPlatformInputs + inputs_optional = True + + @workflow.run + async def run(self, inputs: AlertsPlatformInputs) -> int: + addresses = await workflow.execute_activity( + alerts_platform_find_expired_holds_activity, + start_to_close_timeout=dt.timedelta(seconds=30), + retry_policy=RetryPolicy(maximum_attempts=3), + ) + if not addresses: + return 0 + released = await workflow.execute_activity( + alerts_platform_release_expired_holds_activity, + addresses, + start_to_close_timeout=dt.timedelta(seconds=60), + retry_policy=RetryPolicy(maximum_attempts=3), + ) + return await start_deliveries(released) + + @workflow.defn(name="alerts-platform-deliver") class AlertsPlatformDeliverWorkflow(PostHogWorkflow): inputs_cls = AlertsPlatformInputs @@ -565,9 +632,15 @@ async def run(self, inputs: OrchestrateInputs) -> OrchestrateResult: alerts_platform_probe_postgres_activity, alerts_platform_record_outcomes_activity, ] -DELIVERY_WORKFLOWS: list[type[PostHogWorkflow]] = [AlertsPlatformDeliverWorkflow, AlertsPlatformDeliverPreviewWorkflow] +DELIVERY_WORKFLOWS: list[type[PostHogWorkflow]] = [ + AlertsPlatformDeliverWorkflow, + AlertsPlatformDeliverPreviewWorkflow, + AlertsPlatformReleaseExpiredHoldsWorkflow, +] DELIVERY_ACTIVITIES: list[Callable[..., object]] = [ alerts_platform_deliver_activity, alerts_platform_deliver_preview_activity, alerts_platform_emit_transitions_activity, + alerts_platform_find_expired_holds_activity, + alerts_platform_release_expired_holds_activity, ] diff --git a/products/alerts_platform/backend/tests/test_delivery_evaluation.py b/products/alerts_platform/backend/tests/test_delivery_evaluation.py index 0fd3438d66b7..5e17002b778d 100644 --- a/products/alerts_platform/backend/tests/test_delivery_evaluation.py +++ b/products/alerts_platform/backend/tests/test_delivery_evaluation.py @@ -33,13 +33,15 @@ MessageLink, SourceKind, ) -from products.alerts_platform.backend.models import PlatformAlertConfiguration +from products.alerts_platform.backend.facade.enums import HoldRelease +from products.alerts_platform.backend.models import PlatformAlertConfiguration, PlatformAlertDeliveryHold from products.alerts_platform.backend.tests.recorded_metrics import recorded_metrics _MODULE = "products.alerts_platform.backend.delivery.evaluation" _LEGACY_MODULE = "products.alerts_platform.backend.delivery.legacy_webhook" _WEBHOOK_URL_MODULE = "products.alerts_platform.backend.delivery.webhook_url" FIRING = datetime(2026, 9, 30, 9, tzinfo=UTC) +CONFIGURATION_ID = "01900000-0000-0000-0000-0000000000c1" FIRING_EVENT = "$logs_alert_firing" RESOLVED_EVENT = "$logs_alert_resolved" @@ -106,7 +108,7 @@ def deliver(self, **kwargs: Any) -> MessageHandle | None: def _request( team_id: int, *, - configuration_id: str = "cfg-1", + configuration_id: str = CONFIGURATION_ID, sends_messages: bool = True, incident_actions: dict[str, IncidentAction] | None = None, owned_by_platform: bool = False, @@ -152,7 +154,7 @@ def _transition( def _announcement(*transitions: AnnouncedTransition) -> EvaluationAnnouncement: return EvaluationAnnouncement( source=SourceKind.LOGS, - configuration_id="cfg-1", + configuration_id=CONFIGURATION_ID, alert_name="API errors", consecutive_failures=0, transitions=transitions, @@ -443,6 +445,37 @@ def test_an_evaluation_that_announced_nothing_sends_nothing(self) -> None: assert RecordingTransport.sends == [] assert outcome.sent == 0 + @parameterized.expand( + [("unreleased", None, 0), ("suppressed", HoldRelease.SUPPRESS, 0), ("sent", HoldRelease.SEND, 1)] + ) + def test_a_held_evaluation_sends_only_once_released_with_a_send( + self, _name: str, release: HoldRelease | None, sent: int + ) -> None: + configuration = PlatformAlertConfiguration.objects.for_team(self.team.id).create( + id=CONFIGURATION_ID, + team=self.team, + name="API errors", + source_kind=PlatformAlertConfiguration.SourceKind.LOGS, + check_interval_minutes=10, + ) + PlatformAlertDeliveryHold.objects.for_team(self.team.id).create( + team=self.team, + configuration=configuration, + evaluation_key="eval-1", + reason="investigation", + request={}, + held_at=FIRING, + release_after=FIRING, + released_at=FIRING if release else None, + release_action=release, + ) + + outcome = self._run( + _announcement(_transition(AlertEventKind.FIRING)), {FIRING_EVENT: [_group(SLACK)]}, owned_by_platform=True + ) + + assert outcome.sent == sent + def test_a_team_without_the_flag_is_not_contacted(self) -> None: outcome = self._run( _announcement(_transition(AlertEventKind.FIRING)), {FIRING_EVENT: [_group(SLACK)]}, live=False diff --git a/products/alerts_platform/backend/tests/test_delivery_holds.py b/products/alerts_platform/backend/tests/test_delivery_holds.py new file mode 100644 index 000000000000..67d9dc797557 --- /dev/null +++ b/products/alerts_platform/backend/tests/test_delivery_holds.py @@ -0,0 +1,166 @@ +from datetime import UTC, datetime, timedelta + +from posthog.test.base import BaseTest + +from parameterized import parameterized + +from products.alerts_platform.backend.facade.api import record_outcomes, release_hold, reschedule_hold +from products.alerts_platform.backend.facade.contracts import ( + AlertDeliveryRequest, + AlertEventKind, + FiringEpisode, + GroupOutcome, + HeldDelivery, + HeldDeliveryAddress, + IncidentAction, + PlatformAlertOutcome, + SourceKind, +) +from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy +from products.alerts_platform.backend.logic.delivery_holds import ( + HOLD_RETENTION, + expired_holds, + prune_holds, + release_expired, +) +from products.alerts_platform.backend.models import PlatformAlertConfiguration, PlatformAlertDeliveryHold + +NOW = datetime(2026, 10, 20, 9, tzinfo=UTC) + + +class TestDeliveryHolds(BaseTest): + def setUp(self) -> None: + super().setUp() + self.configuration = self._configuration() + + def _configuration(self, *, enabled: bool = True) -> PlatformAlertConfiguration: + return PlatformAlertConfiguration.objects.for_team(self.team.id).create( + team=self.team, + name="Signups", + enabled=enabled, + source_kind=PlatformAlertConfiguration.SourceKind.INSIGHT, + check_interval_minutes=60, + next_check_at=NOW - timedelta(minutes=1), + ) + + def _request(self, configuration: PlatformAlertConfiguration, key: str = "slot:1") -> AlertDeliveryRequest: + return AlertDeliveryRequest( + source=SourceKind.INSIGHT, + team_id=self.team.id, + configuration_id=str(configuration.id), + evaluation_key=key, + destination_alert_id="legacy-1", + event_ids_by_kind={"firing": "$insight_alert_firing"}, + incident_actions={"": IncidentAction.TRIGGER}, + owned_by_platform=True, + ) + + def _address(self, configuration: PlatformAlertConfiguration, key: str = "slot:1") -> HeldDeliveryAddress: + return HeldDeliveryAddress(team_id=self.team.id, configuration_id=str(configuration.id), evaluation_key=key) + + def _hold( + self, + configuration: PlatformAlertConfiguration, + *, + key: str = "slot:1", + held_at: datetime = NOW, + release_after: datetime = NOW, + released: bool = False, + ) -> None: + PlatformAlertDeliveryHold.objects.for_team(self.team.id).create( + team=self.team, + configuration=configuration, + evaluation_key=key, + reason="investigation", + request=self._request(configuration, key).to_stored(), + held_at=held_at, + release_after=release_after, + released_at=held_at if released else None, + release_action=HoldRelease.SEND if released else None, + ) + + def test_a_held_check_records_its_hold_with_the_outcome_and_a_replay_adds_none(self) -> None: + request = self._request(self.configuration) + + def outcome(reason: str) -> PlatformAlertOutcome: + return PlatformAlertOutcome( + configuration_id=self.configuration.id, + evaluation_key="slot:1", + consecutive_failures=0, + groups=( + GroupOutcome( + grouping_key="", + kind=AlertEventKind.FIRING, + new_state="firing", + notified=True, + firing_episode=FiringEpisode(started_at=NOW, ended=False), + ), + ), + held_delivery=HeldDelivery(request=request, reason=reason, release_after_seconds=5400), + ) + + assert record_outcomes(self.team.id, [outcome("investigation")], NOW) == 1 + assert record_outcomes(self.team.id, [outcome("replayed")], NOW) == 0 + + hold = PlatformAlertDeliveryHold.objects.for_team(self.team.id).get() + assert (hold.reason, hold.release_after, hold.released_at) == ( + "investigation", + NOW + timedelta(minutes=90), + None, + ) + assert AlertDeliveryRequest.from_stored(hold.request) == request + + @parameterized.expand( + [ + ("the first send wins", [(HoldRelease.SEND, HoldReleasedBy.VERDICT, True)]), + ( + "a retry of the winning send gets the request again", + [(HoldRelease.SEND, HoldReleasedBy.VERDICT, True), (HoldRelease.SEND, HoldReleasedBy.VERDICT, True)], + ), + ( + "another releaser after the winner gets nothing", + [ + (HoldRelease.SEND, HoldReleasedBy.VERDICT, True), + (HoldRelease.SEND, HoldReleasedBy.SAFETY_NET, False), + ], + ), + ( + "a suppress keeps a later send from going out", + [ + (HoldRelease.SUPPRESS, HoldReleasedBy.VERDICT, False), + (HoldRelease.SEND, HoldReleasedBy.SAFETY_NET, False), + ], + ), + ] + ) + def test_one_release_wins_and_only_its_send_returns_the_request( + self, _name: str, releases: list[tuple[HoldRelease, HoldReleasedBy, bool]] + ) -> None: + self._hold(self.configuration) + address = self._address(self.configuration) + + for action, released_by, returned in releases: + request = release_hold(address, action=action, released_by=released_by, now=NOW) + assert request == (self._request(self.configuration) if returned else None) + + hold = PlatformAlertDeliveryHold.objects.for_team(self.team.id).get() + assert (hold.release_action, hold.released_by) == (releases[0][0], releases[0][1]) + assert reschedule_hold(address, release_after=NOW + timedelta(hours=1)) is False + + def test_the_safety_net_sends_only_holds_past_their_deadline_on_enabled_alerts(self) -> None: + self._hold(self.configuration, key="due", release_after=NOW - timedelta(minutes=1)) + self._hold(self.configuration, key="rescheduled", release_after=NOW - timedelta(minutes=1)) + self._hold(self.configuration, key="waiting", release_after=NOW + timedelta(minutes=1)) + self._hold(self.configuration, key="released", release_after=NOW - timedelta(minutes=1), released=True) + self._hold(self._configuration(enabled=False), key="disabled", release_after=NOW - timedelta(minutes=1)) + self._hold(self.configuration, key="expired", held_at=NOW - HOLD_RETENTION - timedelta(days=1)) + assert reschedule_hold( + self._address(self.configuration, "rescheduled"), release_after=NOW + timedelta(minutes=5) + ) + + assert prune_holds(NOW, limit=10) == 1 + found = expired_holds(NOW, limit=10) + + assert found == (self._address(self.configuration, "due"),) + assert release_expired(found, NOW) == (self._request(self.configuration, "due"),) + assert expired_holds(NOW, limit=10) == () From 1375f479264907a9f0fddcba1dd390fe9560e58d Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 26 Oct 2026 12:45:04 -0700 Subject: [PATCH 2/6] fix(alerts): harden the held-delivery sweep and suppress disabled holds Register the hold model in the IDOR semgrep rules. The sweep now re-reads the deadline under the row lock, suppresses the hold of a disabled alert instead of sending it on re-enable, starts released sends whose start was lost, and keeps its result within one payload. Co-Authored-By: Claude Opus 5.5 --- .../security/idor-team-scoped-models.yaml | 2 + docs/internal/alerts-platform-temporal.md | 7 +- .../alerts_platform/backend/facade/api.py | 6 + .../backend/logic/delivery_holds.py | 117 +++++++++++++++--- .../backend/migrations/0020_delivery_hold.py | 11 +- .../backend/models/platform_alert.py | 8 ++ .../backend/temporal/workflows.py | 49 +++++--- .../backend/tests/test_delivery_holds.py | 54 +++++--- 8 files changed, 201 insertions(+), 53 deletions(-) diff --git a/.semgrep/rules/security/idor-team-scoped-models.yaml b/.semgrep/rules/security/idor-team-scoped-models.yaml index 2ada6507f784..964e86c6037e 100644 --- a/.semgrep/rules/security/idor-team-scoped-models.yaml +++ b/.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 diff --git a/docs/internal/alerts-platform-temporal.md b/docs/internal/alerts-platform-temporal.md index e062a1d03f59..eb26afb48276 100644 --- a/docs/internal/alerts-platform-temporal.md +++ b/docs/internal/alerts-platform-temporal.md @@ -549,14 +549,17 @@ It puts the request on `PlatformAlertOutcome.held_delivery` with a reason and a `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. +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 skips the holds of a disabled configuration. It still sends a hold whose alert went back to its source, because the request records the owner at check time. +It re-reads each hold's deadline and configuration under the row lock, so a hold rescheduled after the find stays held. +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 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. `alerts_platform_holds_expired_total` counts the holds the safety net sent. A rise means a source holds deliveries and does not release them. diff --git a/products/alerts_platform/backend/facade/api.py b/products/alerts_platform/backend/facade/api.py index fd8ea6e8c0e7..d2032ce41ea0 100644 --- a/products/alerts_platform/backend/facade/api.py +++ b/products/alerts_platform/backend/facade/api.py @@ -72,6 +72,12 @@ def release_hold( 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) diff --git a/products/alerts_platform/backend/logic/delivery_holds.py b/products/alerts_platform/backend/logic/delivery_holds.py index a0ab12763d5d..332bb0033deb 100644 --- a/products/alerts_platform/backend/logic/delivery_holds.py +++ b/products/alerts_platform/backend/logic/delivery_holds.py @@ -12,17 +12,35 @@ from django.db import transaction +from posthog.dataclasses import frozen + from products.alerts_platform.backend.facade.contracts import ( AlertDeliveryRequest, HeldDeliveryAddress, PlatformAlertOutcome, ) from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy +from products.alerts_platform.backend.facade.payload import payload_bytes from products.alerts_platform.backend.models import PlatformAlertConfiguration, PlatformAlertDeliveryHold from products.alerts_platform.backend.models.platform_alert_events_sql import PLATFORM_ALERT_EVENTS_TTL_DAYS # As long as the history rows a hold annotates live, so a reader never finds one without the other. HOLD_RETENTION: Final = timedelta(days=PLATFORM_ALERT_EVENTS_TTL_DAYS) +# A releaser starts its delivery seconds after the release. One that has not started by then failed +# between the two, and the sweep starts it instead. +START_GRACE: Final = timedelta(minutes=5) + + +@frozen +class HoldSweep: + """What one sweep found to send: holds past their deadline, and released sends nobody started.""" + + expired: tuple[HeldDeliveryAddress, ...] + unstarted: tuple[HeldDeliveryAddress, ...] + + +def _address(team_id: int, configuration_id: object, evaluation_key: str) -> HeldDeliveryAddress: + return HeldDeliveryAddress(team_id=team_id, configuration_id=str(configuration_id), evaluation_key=evaluation_key) def hold_for( @@ -44,18 +62,28 @@ def hold_for( def release_hold( - address: HeldDeliveryAddress, *, action: HoldRelease, released_by: HoldReleasedBy, now: datetime + address: HeldDeliveryAddress, + *, + action: HoldRelease, + released_by: HoldReleasedBy, + now: datetime, + only_if_due: bool = False, ) -> AlertDeliveryRequest | None: """Releases one hold. Returns the request when the caller should start its delivery. The first release wins. A later one changes nothing and gets the request back only when it repeats the winning send from the same releaser, which is a retry of a caller that committed the release and failed before it started the delivery. The delivery's id then dedups it. + + `only_if_due` re-reads the deadline under the lock, so a hold rescheduled after the sweep found + it stays held. It also suppresses, rather than sends, the hold of a configuration that is + disabled by then, so turning the alert back on never sends a firing that old. """ with transaction.atomic(): hold = ( PlatformAlertDeliveryHold.objects.for_team(address.team_id) - .select_for_update() + .select_related("configuration") + .select_for_update(of=("self",)) .filter(configuration_id=address.configuration_id, evaluation_key=address.evaluation_key) .first() ) @@ -64,6 +92,10 @@ def release_hold( if hold.released_at is not None: repeated = hold.release_action == action == HoldRelease.SEND and hold.released_by == released_by return AlertDeliveryRequest.from_stored(hold.request) if repeated else None + if only_if_due and hold.release_after > now: + return None + if only_if_due and not hold.configuration.enabled: + action = HoldRelease.SUPPRESS hold.released_at = now hold.release_action = action.value hold.released_by = released_by.value @@ -71,6 +103,17 @@ def release_hold( return AlertDeliveryRequest.from_stored(hold.request) if action == HoldRelease.SEND else None +def record_delivery_started(addresses: Sequence[HeldDeliveryAddress], now: datetime) -> None: + """Marks released sends whose delivery a caller started, so the sweep does not start them again.""" + for address in addresses: + PlatformAlertDeliveryHold.objects.for_team(address.team_id).filter( + configuration_id=address.configuration_id, + evaluation_key=address.evaluation_key, + release_action=HoldRelease.SEND, + delivery_started_at__isnull=True, + ).update(delivery_started_at=now) + + def reschedule_hold(address: HeldDeliveryAddress, *, release_after: datetime) -> bool: """Moves an unreleased hold's deadline. Returns False when the hold is gone or already released. @@ -113,32 +156,66 @@ def prune_holds(now: datetime, *, limit: int) -> int: return PlatformAlertDeliveryHold.objects.unscoped().filter(id__in=stale).delete()[0] if stale else 0 -def expired_holds(now: datetime, *, limit: int) -> tuple[HeldDeliveryAddress, ...]: - """The holds past their deadline, oldest first. +def holds_to_send(now: datetime, *, limit: int) -> HoldSweep: + """The holds the safety net sends now, oldest first, up to `limit` of each kind. - Every team's, so one sweep covers the fleet. A disabled configuration's holds are left held, - the way the source's own safety net skips a disabled alert. An alert rolled back to its source - is still swept, because its held request was decided while the platform owned it. + Every team's, so one sweep covers the fleet. A disabled configuration's holds are found too, and + suppressed when released. An alert rolled back to its source is still swept, because its held + request was decided while the platform owned it. """ - rows = ( + expired = ( PlatformAlertDeliveryHold.objects.unscoped() - .filter(released_at__isnull=True, release_after__lte=now, configuration__enabled=True) + .filter(released_at__isnull=True, release_after__lte=now) .order_by("release_after") .values_list("team_id", "configuration_id", "evaluation_key")[:limit] ) - return tuple( - HeldDeliveryAddress(team_id=team_id, configuration_id=str(configuration_id), evaluation_key=key) - for team_id, configuration_id, key in rows + unstarted = ( + PlatformAlertDeliveryHold.objects.unscoped() + .filter( + release_action=HoldRelease.SEND, + delivery_started_at__isnull=True, + released_at__lte=now - START_GRACE, + ) + .order_by("released_at") + .values_list("team_id", "configuration_id", "evaluation_key")[:limit] + ) + return HoldSweep( + expired=tuple(_address(*row) for row in expired), unstarted=tuple(_address(*row) for row in unstarted) ) -def release_expired(addresses: Sequence[HeldDeliveryAddress], now: datetime) -> tuple[AlertDeliveryRequest, ...]: - """Releases each hold with a send from the safety net, and returns the requests it won. +def _stored_request(address: HeldDeliveryAddress) -> AlertDeliveryRequest | None: + stored = ( + PlatformAlertDeliveryHold.objects.for_team(address.team_id) + .filter(configuration_id=address.configuration_id, evaluation_key=address.evaluation_key) + .values_list("request", flat=True) + .first() + ) + return None if stored is None else AlertDeliveryRequest.from_stored(stored) + - One transaction per hold, so a hold another releaser locked delays only itself. +def send_swept(sweep: HoldSweep, now: datetime, *, max_bytes: int) -> tuple[AlertDeliveryRequest, ...]: + """The requests the sweep starts: released sends nobody started, then expired holds it releases. + + Stops before the requests outgrow `max_bytes`, because they return through one activity result. + A hold left out stays as it was, so the next sweep finds it again. Each expired hold is released + in a transaction of its own, so a hold another releaser locked delays only itself. """ - released = ( - release_hold(address, action=HoldRelease.SEND, released_by=HoldReleasedBy.SAFETY_NET, now=now) - for address in addresses - ) - return tuple(request for request in released if request is not None) + requests: list[AlertDeliveryRequest] = [] + spent = 0 + for address in (*sweep.unstarted, *sweep.expired): + request = _stored_request(address) + if request is None: + continue + size = payload_bytes(request) + if requests and spent + size > max_bytes: + break + if address in sweep.expired: + request = release_hold( + address, action=HoldRelease.SEND, released_by=HoldReleasedBy.SAFETY_NET, now=now, only_if_due=True + ) + if request is None: + continue + requests.append(request) + spent += size + return tuple(requests) diff --git a/products/alerts_platform/backend/migrations/0020_delivery_hold.py b/products/alerts_platform/backend/migrations/0020_delivery_hold.py index d18b51c64e22..c64e32d7eb4b 100644 --- a/products/alerts_platform/backend/migrations/0020_delivery_hold.py +++ b/products/alerts_platform/backend/migrations/0020_delivery_hold.py @@ -1,4 +1,4 @@ -# Generated by Django 5.2.18 on 2026-10-26 18:31 +# Generated by Django 5.2.18 on 2026-10-26 19:32 import django.db.models.deletion from django.db import migrations, models @@ -53,6 +53,7 @@ class Migration(migrations.Migration): null=True, ), ), + ("delivery_started_at", models.DateTimeField(blank=True, null=True)), ( "configuration", models.ForeignKey( @@ -79,6 +80,14 @@ class Migration(migrations.Migration): fields=["release_after"], name="platform_alert_hold_due_idx", ), + models.Index( + condition=models.Q( + ("delivery_started_at__isnull", True), + ("release_action", "send"), + ), + fields=["released_at"], + name="platform_alert_hold_start_idx", + ), models.Index(fields=["held_at"], name="platform_alert_hold_age_idx"), ], "constraints": [ diff --git a/products/alerts_platform/backend/models/platform_alert.py b/products/alerts_platform/backend/models/platform_alert.py index d917a1fda8f4..51464807385e 100644 --- a/products/alerts_platform/backend/models/platform_alert.py +++ b/products/alerts_platform/backend/models/platform_alert.py @@ -300,6 +300,8 @@ class PlatformAlertDeliveryHold(TeamScopedRootMixin, UUIDModel): released_at = models.DateTimeField(null=True, blank=True) release_action = models.CharField(max_length=16, choices=HoldRelease.choices, null=True, blank=True) released_by = models.CharField(max_length=16, choices=HoldReleasedBy.choices, null=True, blank=True) + # When the releaser of a send started its delivery. The sweep starts any send left unstarted. + delivery_started_at = models.DateTimeField(null=True, blank=True) class Meta: db_table = "alerts_platformalertdeliveryhold" @@ -315,6 +317,12 @@ class Meta: name="platform_alert_hold_due_idx", condition=models.Q(released_at__isnull=True), ), + # The sweep's recovery of released sends whose releaser never started them. + models.Index( + fields=["released_at"], + name="platform_alert_hold_start_idx", + condition=models.Q(release_action="send", delivery_started_at__isnull=True), + ), # Pruning, which deletes holds once the history rows they annotate have expired. models.Index(fields=["held_at"], name="platform_alert_hold_age_idx"), ] diff --git a/products/alerts_platform/backend/temporal/workflows.py b/products/alerts_platform/backend/temporal/workflows.py index 7d16ed67881c..e67ef2c6c0a1 100644 --- a/products/alerts_platform/backend/temporal/workflows.py +++ b/products/alerts_platform/backend/temporal/workflows.py @@ -51,6 +51,7 @@ SourceEvaluationInputs, TickPage, ) + from products.alerts_platform.backend.facade.payload import MAX_BATCH_PAYLOAD_BYTES from products.alerts_platform.backend.logic import delivery_holds from products.alerts_platform.backend.logic.demand import discover_demand from products.alerts_platform.backend.logic.inventory import count_inventory @@ -302,7 +303,7 @@ async def start_deliveries(deliveries: Sequence[AlertDeliveryRequest]) -> int: @activity.defn -async def alerts_platform_find_expired_holds_activity() -> list[HeldDeliveryAddress]: +async def alerts_platform_find_expired_holds_activity() -> delivery_holds.HoldSweep: now = dt.datetime.now(dt.UTC) # A failed prune is logged, so cleanup never keeps a due hold from being sent. try: @@ -312,16 +313,16 @@ async def alerts_platform_find_expired_holds_activity() -> list[HeldDeliveryAddr else: if pruned: await LOGGER.ainfo("alerts_platform_holds_pruned", count=pruned) - return list(await database_sync_to_async_pool(delivery_holds.expired_holds)(now, limit=MAX_HOLDS_PER_SWEEP)) + return await database_sync_to_async_pool(delivery_holds.holds_to_send)(now, limit=MAX_HOLDS_PER_SWEEP) @activity.defn -async def alerts_platform_release_expired_holds_activity( - addresses: list[HeldDeliveryAddress], -) -> list[AlertDeliveryRequest]: +async def alerts_platform_release_expired_holds_activity(sweep: delivery_holds.HoldSweep) -> list[AlertDeliveryRequest]: """Separate from the find, so a retry after a committed release gets the same requests back.""" - released = await database_sync_to_async_pool(delivery_holds.release_expired)(addresses, dt.datetime.now(dt.UTC)) - for request in released: + requests = await database_sync_to_async_pool(delivery_holds.send_swept)( + sweep, dt.datetime.now(dt.UTC), max_bytes=MAX_BATCH_PAYLOAD_BYTES + ) + for request in requests: await LOGGER.awarning( "alerts_platform_hold_expired", source=request.source.value, @@ -329,7 +330,18 @@ async def alerts_platform_release_expired_holds_activity( evaluation_key=request.evaluation_key, ) safe_record(increment_holds_expired, request.source.value) - return list(released) + return list(requests) + + +@activity.defn +async def alerts_platform_record_hold_deliveries_started_activity(requests: list[AlertDeliveryRequest]) -> None: + addresses = [ + HeldDeliveryAddress( + team_id=request.team_id, configuration_id=request.configuration_id, evaluation_key=request.evaluation_key + ) + for request in requests + ] + await database_sync_to_async_pool(delivery_holds.record_delivery_started)(addresses, dt.datetime.now(dt.UTC)) @workflow.defn(name="alerts-platform-release-expired-holds") @@ -337,7 +349,8 @@ class AlertsPlatformReleaseExpiredHoldsWorkflow(PostHogWorkflow): """The safety net for held deliveries: sends each one whose deadline passed with no release. One release of a hold wins, so a source that releases it at the same moment cannot also send it. - The delivery keeps its usual id, so a retried run that starts it again starts nothing new. + A send is marked started only once its delivery started, so a run that fails between the two + leaves it for a later run. The delivery keeps its usual id, so starting it twice starts nothing new. """ inputs_cls = AlertsPlatformInputs @@ -345,20 +358,27 @@ class AlertsPlatformReleaseExpiredHoldsWorkflow(PostHogWorkflow): @workflow.run async def run(self, inputs: AlertsPlatformInputs) -> int: - addresses = await workflow.execute_activity( + sweep = await workflow.execute_activity( alerts_platform_find_expired_holds_activity, start_to_close_timeout=dt.timedelta(seconds=30), retry_policy=RetryPolicy(maximum_attempts=3), ) - if not addresses: + if not sweep.expired and not sweep.unstarted: return 0 - released = await workflow.execute_activity( + requests = await workflow.execute_activity( alerts_platform_release_expired_holds_activity, - addresses, + sweep, start_to_close_timeout=dt.timedelta(seconds=60), retry_policy=RetryPolicy(maximum_attempts=3), ) - return await start_deliveries(released) + started = await start_deliveries(requests) + await workflow.execute_activity( + alerts_platform_record_hold_deliveries_started_activity, + requests, + start_to_close_timeout=dt.timedelta(seconds=30), + retry_policy=RetryPolicy(maximum_attempts=3), + ) + return started @workflow.defn(name="alerts-platform-deliver") @@ -643,4 +663,5 @@ async def run(self, inputs: OrchestrateInputs) -> OrchestrateResult: alerts_platform_emit_transitions_activity, alerts_platform_find_expired_holds_activity, alerts_platform_release_expired_holds_activity, + alerts_platform_record_hold_deliveries_started_activity, ] diff --git a/products/alerts_platform/backend/tests/test_delivery_holds.py b/products/alerts_platform/backend/tests/test_delivery_holds.py index 67d9dc797557..90a21253a4bb 100644 --- a/products/alerts_platform/backend/tests/test_delivery_holds.py +++ b/products/alerts_platform/backend/tests/test_delivery_holds.py @@ -17,11 +17,15 @@ SourceKind, ) from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy +from products.alerts_platform.backend.facade.payload import MAX_BATCH_PAYLOAD_BYTES from products.alerts_platform.backend.logic.delivery_holds import ( HOLD_RETENTION, - expired_holds, + START_GRACE, + HoldSweep, + holds_to_send, prune_holds, - release_expired, + record_delivery_started, + send_swept, ) from products.alerts_platform.backend.models import PlatformAlertConfiguration, PlatformAlertDeliveryHold @@ -65,7 +69,7 @@ def _hold( key: str = "slot:1", held_at: datetime = NOW, release_after: datetime = NOW, - released: bool = False, + released_at: datetime | None = None, ) -> None: PlatformAlertDeliveryHold.objects.for_team(self.team.id).create( team=self.team, @@ -75,8 +79,8 @@ def _hold( request=self._request(configuration, key).to_stored(), held_at=held_at, release_after=release_after, - released_at=held_at if released else None, - release_action=HoldRelease.SEND if released else None, + released_at=released_at, + release_action=HoldRelease.SEND if released_at else None, ) def test_a_held_check_records_its_hold_with_the_outcome_and_a_replay_adds_none(self) -> None: @@ -147,20 +151,38 @@ def test_one_release_wins_and_only_its_send_returns_the_request( assert (hold.release_action, hold.released_by) == (releases[0][0], releases[0][1]) assert reschedule_hold(address, release_after=NOW + timedelta(hours=1)) is False - def test_the_safety_net_sends_only_holds_past_their_deadline_on_enabled_alerts(self) -> None: - self._hold(self.configuration, key="due", release_after=NOW - timedelta(minutes=1)) - self._hold(self.configuration, key="rescheduled", release_after=NOW - timedelta(minutes=1)) + def test_the_safety_net_sends_expired_holds_and_released_sends_nobody_started(self) -> None: + late = NOW - timedelta(minutes=1) + self._hold(self.configuration, key="due", release_after=late) + self._hold(self.configuration, key="rescheduled", release_after=late) self._hold(self.configuration, key="waiting", release_after=NOW + timedelta(minutes=1)) - self._hold(self.configuration, key="released", release_after=NOW - timedelta(minutes=1), released=True) - self._hold(self._configuration(enabled=False), key="disabled", release_after=NOW - timedelta(minutes=1)) - self._hold(self.configuration, key="expired", held_at=NOW - HOLD_RETENTION - timedelta(days=1)) + self._hold(self.configuration, key="unstarted", release_after=late, released_at=NOW - START_GRACE) + self._hold(self.configuration, key="starting", release_after=late, released_at=NOW - timedelta(minutes=1)) + self._hold(self.configuration, key="started", release_after=late, released_at=NOW - START_GRACE) + record_delivery_started([self._address(self.configuration, "started")], NOW) + disabled = self._configuration(enabled=False) + self._hold(disabled, key="disabled", release_after=late) + self._hold(self.configuration, key="old", held_at=NOW - HOLD_RETENTION - timedelta(days=1)) + assert prune_holds(NOW, limit=10) == 1 + + sweep = holds_to_send(NOW, limit=10) + # A source can move the deadline after the sweep found the hold, and must keep it held. assert reschedule_hold( self._address(self.configuration, "rescheduled"), release_after=NOW + timedelta(minutes=5) ) + requests = send_swept(sweep, NOW, max_bytes=MAX_BATCH_PAYLOAD_BYTES) - assert prune_holds(NOW, limit=10) == 1 - found = expired_holds(NOW, limit=10) + assert requests == (self._request(self.configuration, "unstarted"), self._request(self.configuration, "due")) + suppressed = PlatformAlertDeliveryHold.objects.for_team(self.team.id).get(configuration=disabled) + assert (suppressed.release_action, suppressed.released_by) == (HoldRelease.SUPPRESS, HoldReleasedBy.SAFETY_NET) + record_delivery_started([self._address(self.configuration, r.evaluation_key) for r in requests], NOW) + assert holds_to_send(NOW, limit=10) == HoldSweep(expired=(), unstarted=()) + + def test_one_sweep_sends_no_more_than_fits_one_payload(self) -> None: + for key in ("first", "second"): + self._hold(self.configuration, key=key, release_after=NOW - timedelta(minutes=1)) + + requests = send_swept(holds_to_send(NOW, limit=10), NOW, max_bytes=1) - assert found == (self._address(self.configuration, "due"),) - assert release_expired(found, NOW) == (self._request(self.configuration, "due"),) - assert expired_holds(NOW, limit=10) == () + assert len(requests) == 1 + assert len(holds_to_send(NOW, limit=10).expired) == 1 From 9b09f7809674252c6e2057582bd1de371c0f328b Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 26 Oct 2026 13:00:32 -0700 Subject: [PATCH 3/6] fix(alerts): keep temporalio off the django startup path Defer the Temporal converter and payload imports used by held deliveries, because the alerts_platform models load the contracts module during django.setup(). Co-Authored-By: Claude Opus 5.5 --- products/alerts_platform/backend/facade/contracts.py | 4 ++-- products/alerts_platform/backend/logic/delivery_holds.py | 4 +++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/products/alerts_platform/backend/facade/contracts.py b/products/alerts_platform/backend/facade/contracts.py index e95ee578d8a0..337245eeacd8 100644 --- a/products/alerts_platform/backend/facade/contracts.py +++ b/products/alerts_platform/backend/facade/contracts.py @@ -14,8 +14,6 @@ from typing import TYPE_CHECKING, Any, Final, NotRequired, Protocol, TypedDict from uuid import UUID -import temporalio.converter - from posthog.dataclasses import frozen from posthog.enums import LabeledStrEnum @@ -839,6 +837,8 @@ def to_stored(self) -> dict[str, Any]: @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)) diff --git a/products/alerts_platform/backend/logic/delivery_holds.py b/products/alerts_platform/backend/logic/delivery_holds.py index 332bb0033deb..3317ae17bfb0 100644 --- a/products/alerts_platform/backend/logic/delivery_holds.py +++ b/products/alerts_platform/backend/logic/delivery_holds.py @@ -20,7 +20,6 @@ PlatformAlertOutcome, ) from products.alerts_platform.backend.facade.enums import HoldRelease, HoldReleasedBy -from products.alerts_platform.backend.facade.payload import payload_bytes from products.alerts_platform.backend.models import PlatformAlertConfiguration, PlatformAlertDeliveryHold from products.alerts_platform.backend.models.platform_alert_events_sql import PLATFORM_ALERT_EVENTS_TTL_DAYS @@ -201,6 +200,9 @@ def send_swept(sweep: HoldSweep, now: datetime, *, max_bytes: int) -> tuple[Aler A hold left out stays as it was, so the next sweep finds it again. Each expired hold is released in a transaction of its own, so a hold another releaser locked delays only itself. """ + # Deferred: the payload helper imports temporalio, which must stay off django.setup(). + from products.alerts_platform.backend.facade.payload import payload_bytes # noqa: PLC0415 + requests: list[AlertDeliveryRequest] = [] spent = 0 for address in (*sweep.unstarted, *sweep.expired): From 02385c6ecceea3eb41b39521c18f9cc13db41ed9 Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 26 Oct 2026 13:40:35 -0700 Subject: [PATCH 4/6] fix(alerts): keep the hold sweep from re-running or blocking deliveries The sweep restarts lost delivery starts with ALLOW_DUPLICATE_FAILED_ONLY, so a completed delivery does not run again. It skips holds another transaction has locked, and counts only holds it released at their deadline as expired. Co-Authored-By: Claude Opus 5.5 --- docs/internal/alerts-platform-temporal.md | 7 +- .../backend/logic/delivery_holds.py | 64 +++++++++++++------ .../backend/temporal/workflows.py | 21 ++++-- .../backend/tests/test_delivery_holds.py | 10 ++- 4 files changed, 69 insertions(+), 33 deletions(-) diff --git a/docs/internal/alerts-platform-temporal.md b/docs/internal/alerts-platform-temporal.md index eb26afb48276..c7f1cbb220b6 100644 --- a/docs/internal/alerts-platform-temporal.md +++ b/docs/internal/alerts-platform-temporal.md @@ -556,12 +556,13 @@ A retry from the same releaser gets the request back. Another releaser gets noth `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 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 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. -`alerts_platform_holds_expired_total` counts the holds the safety net sent. A rise means a source holds deliveries and does not release them. +A rise in `alerts_platform_holds_expired_total` means a source holds deliveries and does not release them. ## Postgres connectivity probe diff --git a/products/alerts_platform/backend/logic/delivery_holds.py b/products/alerts_platform/backend/logic/delivery_holds.py index 3317ae17bfb0..c3135de79ca0 100644 --- a/products/alerts_platform/backend/logic/delivery_holds.py +++ b/products/alerts_platform/backend/logic/delivery_holds.py @@ -30,6 +30,14 @@ START_GRACE: Final = timedelta(minutes=5) +@frozen +class SweptHolds: + """The requests a sweep starts, and which of them it released at their deadline just now.""" + + requests: tuple[AlertDeliveryRequest, ...] + expired: tuple[AlertDeliveryRequest, ...] + + @frozen class HoldSweep: """What one sweep found to send: holds past their deadline, and released sends nobody started.""" @@ -61,45 +69,55 @@ def hold_for( def release_hold( - address: HeldDeliveryAddress, - *, - action: HoldRelease, - released_by: HoldReleasedBy, - now: datetime, - only_if_due: bool = False, + address: HeldDeliveryAddress, *, action: HoldRelease, released_by: HoldReleasedBy, now: datetime ) -> AlertDeliveryRequest | None: """Releases one hold. Returns the request when the caller should start its delivery. The first release wins. A later one changes nothing and gets the request back only when it repeats the winning send from the same releaser, which is a retry of a caller that committed the release and failed before it started the delivery. The delivery's id then dedups it. + """ + request, _ = _release(address, action=action, released_by=released_by, now=now, by_safety_net=False) + return request - `only_if_due` re-reads the deadline under the lock, so a hold rescheduled after the sweep found - it stays held. It also suppresses, rather than sends, the hold of a configuration that is - disabled by then, so turning the alert back on never sends a firing that old. + +def _release( + address: HeldDeliveryAddress, + *, + action: HoldRelease, + released_by: HoldReleasedBy, + now: datetime, + by_safety_net: bool, +) -> tuple[AlertDeliveryRequest | None, bool]: + """`release_hold`, plus whether this call made the release. + + The safety net skips a hold another transaction has locked, so a slow source delays only its + own hold, and a later sweep finds it again. It re-reads the deadline under the lock, so a hold + rescheduled after the sweep found it stays held. It suppresses, rather than sends, the hold of a + configuration that is disabled by then, so turning the alert back on never sends a firing that old. """ with transaction.atomic(): hold = ( PlatformAlertDeliveryHold.objects.for_team(address.team_id) .select_related("configuration") - .select_for_update(of=("self",)) + .select_for_update(of=("self",), skip_locked=by_safety_net) .filter(configuration_id=address.configuration_id, evaluation_key=address.evaluation_key) .first() ) if hold is None: - return None + return None, False if hold.released_at is not None: repeated = hold.release_action == action == HoldRelease.SEND and hold.released_by == released_by - return AlertDeliveryRequest.from_stored(hold.request) if repeated else None - if only_if_due and hold.release_after > now: - return None - if only_if_due and not hold.configuration.enabled: + return (AlertDeliveryRequest.from_stored(hold.request) if repeated else None), False + if by_safety_net and hold.release_after > now: + return None, False + if by_safety_net and not hold.configuration.enabled: action = HoldRelease.SUPPRESS hold.released_at = now hold.release_action = action.value hold.released_by = released_by.value hold.save(update_fields=["released_at", "release_action", "released_by"]) - return AlertDeliveryRequest.from_stored(hold.request) if action == HoldRelease.SEND else None + return (AlertDeliveryRequest.from_stored(hold.request) if action == HoldRelease.SEND else None), True def record_delivery_started(addresses: Sequence[HeldDeliveryAddress], now: datetime) -> None: @@ -193,7 +211,7 @@ def _stored_request(address: HeldDeliveryAddress) -> AlertDeliveryRequest | None return None if stored is None else AlertDeliveryRequest.from_stored(stored) -def send_swept(sweep: HoldSweep, now: datetime, *, max_bytes: int) -> tuple[AlertDeliveryRequest, ...]: +def send_swept(sweep: HoldSweep, now: datetime, *, max_bytes: int) -> SweptHolds: """The requests the sweep starts: released sends nobody started, then expired holds it releases. Stops before the requests outgrow `max_bytes`, because they return through one activity result. @@ -204,6 +222,7 @@ def send_swept(sweep: HoldSweep, now: datetime, *, max_bytes: int) -> tuple[Aler from products.alerts_platform.backend.facade.payload import payload_bytes # noqa: PLC0415 requests: list[AlertDeliveryRequest] = [] + expired: list[AlertDeliveryRequest] = [] spent = 0 for address in (*sweep.unstarted, *sweep.expired): request = _stored_request(address) @@ -213,11 +232,14 @@ def send_swept(sweep: HoldSweep, now: datetime, *, max_bytes: int) -> tuple[Aler if requests and spent + size > max_bytes: break if address in sweep.expired: - request = release_hold( - address, action=HoldRelease.SEND, released_by=HoldReleasedBy.SAFETY_NET, now=now, only_if_due=True + released, fresh = _release( + address, action=HoldRelease.SEND, released_by=HoldReleasedBy.SAFETY_NET, now=now, by_safety_net=True ) - if request is None: + if released is None: continue + if fresh: + expired.append(released) + request = released requests.append(request) spent += size - return tuple(requests) + return SweptHolds(requests=tuple(requests), expired=tuple(expired)) diff --git a/products/alerts_platform/backend/temporal/workflows.py b/products/alerts_platform/backend/temporal/workflows.py index e67ef2c6c0a1..fe0be5d202ab 100644 --- a/products/alerts_platform/backend/temporal/workflows.py +++ b/products/alerts_platform/backend/temporal/workflows.py @@ -4,7 +4,7 @@ from dataclasses import replace from temporalio import activity, workflow -from temporalio.common import RetryPolicy, SearchAttributeKey +from temporalio.common import RetryPolicy, SearchAttributeKey, WorkflowIDReusePolicy from temporalio.exceptions import ( ActivityError, ApplicationError, @@ -263,7 +263,11 @@ def delivery_execution_timeout(delivery: AlertDeliveryRequest) -> dt.timedelta: return DELIVERY_EXECUTION_TIMEOUT * max(1, delivery.parts) -async def start_deliveries(deliveries: Sequence[AlertDeliveryRequest]) -> int: +async def start_deliveries( + deliveries: Sequence[AlertDeliveryRequest], + *, + id_reuse_policy: WorkflowIDReusePolicy = WorkflowIDReusePolicy.ALLOW_DUPLICATE, +) -> int: """Starts one delivery workflow per request, from inside a source's evaluation workflow. Returns how many started. The id names the configuration and the evaluation key, because the @@ -283,6 +287,7 @@ async def start_deliveries(deliveries: Sequence[AlertDeliveryRequest]) -> int: task_queue=settings.ALERTS_PLATFORM_DELIVERY_TASK_QUEUE, parent_close_policy=workflow.ParentClosePolicy.ABANDON, execution_timeout=delivery_execution_timeout(delivery), + id_reuse_policy=id_reuse_policy, ) for delivery in deliveries ), @@ -319,10 +324,12 @@ async def alerts_platform_find_expired_holds_activity() -> delivery_holds.HoldSw @activity.defn async def alerts_platform_release_expired_holds_activity(sweep: delivery_holds.HoldSweep) -> list[AlertDeliveryRequest]: """Separate from the find, so a retry after a committed release gets the same requests back.""" - requests = await database_sync_to_async_pool(delivery_holds.send_swept)( + swept = await database_sync_to_async_pool(delivery_holds.send_swept)( sweep, dt.datetime.now(dt.UTC), max_bytes=MAX_BATCH_PAYLOAD_BYTES ) - for request in requests: + # Only the holds this attempt released at their deadline, so a recovered send or a retry is not + # counted as a source that stalled. + for request in swept.expired: await LOGGER.awarning( "alerts_platform_hold_expired", source=request.source.value, @@ -330,7 +337,7 @@ async def alerts_platform_release_expired_holds_activity(sweep: delivery_holds.H evaluation_key=request.evaluation_key, ) safe_record(increment_holds_expired, request.source.value) - return list(requests) + return list(swept.requests) @activity.defn @@ -371,7 +378,9 @@ async def run(self, inputs: AlertsPlatformInputs) -> int: start_to_close_timeout=dt.timedelta(seconds=60), retry_policy=RetryPolicy(maximum_attempts=3), ) - started = await start_deliveries(requests) + # A recovered send may have delivered already and lost only its start mark. A completed + # delivery is not run again; a failed one is. + started = await start_deliveries(requests, id_reuse_policy=WorkflowIDReusePolicy.ALLOW_DUPLICATE_FAILED_ONLY) await workflow.execute_activity( alerts_platform_record_hold_deliveries_started_activity, requests, diff --git a/products/alerts_platform/backend/tests/test_delivery_holds.py b/products/alerts_platform/backend/tests/test_delivery_holds.py index 90a21253a4bb..8f094680e0b1 100644 --- a/products/alerts_platform/backend/tests/test_delivery_holds.py +++ b/products/alerts_platform/backend/tests/test_delivery_holds.py @@ -170,9 +170,13 @@ def test_the_safety_net_sends_expired_holds_and_released_sends_nobody_started(se assert reschedule_hold( self._address(self.configuration, "rescheduled"), release_after=NOW + timedelta(minutes=5) ) - requests = send_swept(sweep, NOW, max_bytes=MAX_BATCH_PAYLOAD_BYTES) + swept = send_swept(sweep, NOW, max_bytes=MAX_BATCH_PAYLOAD_BYTES) + requests = swept.requests assert requests == (self._request(self.configuration, "unstarted"), self._request(self.configuration, "due")) + # Only the deadline release counts as a stalled source, not the recovered start. + assert swept.expired == (self._request(self.configuration, "due"),) + assert send_swept(sweep, NOW, max_bytes=MAX_BATCH_PAYLOAD_BYTES).expired == () suppressed = PlatformAlertDeliveryHold.objects.for_team(self.team.id).get(configuration=disabled) assert (suppressed.release_action, suppressed.released_by) == (HoldRelease.SUPPRESS, HoldReleasedBy.SAFETY_NET) record_delivery_started([self._address(self.configuration, r.evaluation_key) for r in requests], NOW) @@ -182,7 +186,7 @@ def test_one_sweep_sends_no_more_than_fits_one_payload(self) -> None: for key in ("first", "second"): self._hold(self.configuration, key=key, release_after=NOW - timedelta(minutes=1)) - requests = send_swept(holds_to_send(NOW, limit=10), NOW, max_bytes=1) + swept = send_swept(holds_to_send(NOW, limit=10), NOW, max_bytes=1) - assert len(requests) == 1 + assert len(swept.requests) == 1 assert len(holds_to_send(NOW, limit=10).expired) == 1 From 1f2b1628f29084ac82f722917c098c01423e2f11 Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 26 Oct 2026 13:51:58 -0700 Subject: [PATCH 5/6] fix(alerts): justify the fleet-wide hold prune to the IDOR rule The prune deletes holds older than their history by id, across teams, with no request or team input. Mark it for idor-lookup-without-team. Co-Authored-By: Claude Opus 5.5 --- products/alerts_platform/backend/logic/delivery_holds.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/products/alerts_platform/backend/logic/delivery_holds.py b/products/alerts_platform/backend/logic/delivery_holds.py index c3135de79ca0..c8d1f1e34727 100644 --- a/products/alerts_platform/backend/logic/delivery_holds.py +++ b/products/alerts_platform/backend/logic/delivery_holds.py @@ -170,7 +170,10 @@ def prune_holds(now: datetime, *, limit: int) -> int: .filter(held_at__lt=now - HOLD_RETENTION) .values_list("id", flat=True)[:limit] ) - return PlatformAlertDeliveryHold.objects.unscoped().filter(id__in=stale).delete()[0] if stale else 0 + if not stale: + return 0 + # nosemgrep: idor-lookup-without-team — fleet-wide retention prune by age; no request or team input reaches it + return PlatformAlertDeliveryHold.objects.unscoped().filter(id__in=stale).delete()[0] def holds_to_send(now: datetime, *, limit: int) -> HoldSweep: From f7919aa60b116a5ae34ad22f7a68e36937c8c37b Mon Sep 17 00:00:00 2001 From: Jimmy Zhu Date: Mon, 26 Oct 2026 14:13:15 -0700 Subject: [PATCH 6/6] chore: retrigger ci