ref(hybridcloud): Require the claim's deadline and mailbox on drains

The transitional fallbacks from the drain-claim work come out: valid_until and mailbox are now required, so _begin_drain reduces to building the claim and asking whether it lapsed — no head-row read and no nullable mailbox downstream. The drain_mailbox_parallel alias goes with them; dispatch stopped enqueueing it when the drains merged.

Must deploy at least one deploy after the drain merge: a drain enqueued before it carries neither argument and now raises TypeError. That failure is noisy rather than lossy — the rows keep their claim's schedule_for and the next dispatcher re-claims them once it passes.

Author: vaind

src/sentry/hybridcloud/tasks/deliver_webhooks.py

@@ -391,40 +391,17 @@ def _begin_drain(
     payload_id: int,
     claimed_count: int,
     dispatcher: str | None,
-    valid_until: float | None,
-    mailbox: str | None,
+    valid_until: float,
+    mailbox: str,
 ) -> _MailboxClaim | None:
-    """
-    The claim a drain runs under, or None when it must stand down first.
-
-    A drain enqueued before dispatch sent the mailbox and deadline reads both off
-    its head row: its claim already wrote its deadline as the rows' schedule_for.
-    That one-query fallback goes away once no such drains are left in flight.
-    """
-    deadline = (
-        datetime.datetime.fromtimestamp(valid_until, tz=datetime.UTC)
-        if valid_until is not None
-        else None
-    )
-    if mailbox is None or deadline is None:
-        head = (
-            WebhookPayload.objects.filter(id=payload_id)
-            .values_list("mailbox_name", "schedule_for")
-            .first()
-        )
-        if head is None:
-            # Head already delivered or discarded; whoever claimed the mailbox
-            # next is delivering the rest.
-            return None
-        mailbox = mailbox if mailbox is not None else head[0]
-        deadline = deadline if deadline is not None else head[1]
+    """The claim a drain runs under, or None when it has already lapsed."""
     _set_webhook_delivery_sentry_context(mailbox, _provider_from_mailbox(mailbox))
     claim = _MailboxClaim(
         claimed=claimed_count,
         head_id=payload_id,
         mailbox_name=mailbox,
         dispatcher=dispatcher,
-        valid_until=deadline,
+        valid_until=datetime.datetime.fromtimestamp(valid_until, tz=datetime.UTC),
     )
     if claim.lapsed(log_key="deliver_webhook.stale_claim", extra={"id": payload_id}):
         return None
@@ -913,15 +890,15 @@ def drain_mailbox(
     payload_id: int,
     claimed_count: int,
     dispatcher: str | None = None,
-    valid_until: float | None = None,
-    mailbox: str | None = None,
+    *,
+    valid_until: float,
+    mailbox: str,
 ) -> None:
     """
     Deliver webhooks from the mailbox that `payload_id` is the head of — in order,
     or in concurrent waves when the claim qualifies (`_MailboxClaim.threaded`).
 
-    The arguments are one claim flattened for the wire (`_MailboxClaim.task_args`);
-    each defaults so a rolling deploy can bind drains the previous version sent.
+    The arguments are one claim flattened for the wire (`_MailboxClaim.task_args`).
     """
     claim = _begin_drain(payload_id, claimed_count, dispatcher, valid_until, mailbox)
     if claim is not None:
@@ -1244,30 +1221,6 @@ def _run_parallel_delivery_batch(
     return delivered
 
 
-@instrumented_task(
-    name="sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox_parallel",
-    namespace=hybridcloud_control_tasks,
-    # The pre-merge task's deadline, kept for the in-flight drains this shim serves.
-    processing_deadline_duration=int(BATCH_SCHEDULE_OFFSET.total_seconds() + 10),
-    silo_mode=SiloMode.CONTROL,
-)
-def drain_mailbox_parallel(
-    payload_id: int,
-    claimed_count: int,
-    dispatcher: str | None = None,
-    valid_until: float | None = None,
-    mailbox: str | None = None,
-) -> None:
-    """
-    Transitional alias from when sequential and parallel delivery were separate
-    tasks; `drain_mailbox` now runs both modes. Dispatch no longer enqueues this,
-    so it is deletable once no drains from the previous deploy are left in flight.
-    """
-    claim = _begin_drain(payload_id, claimed_count, dispatcher, valid_until, mailbox)
-    if claim is not None:
-        _drain_mailbox(claim)
-
-
 def deliver_message_parallel(payload: WebhookPayload) -> tuple[WebhookPayload, Exception | None]:
     try:
         perform_request(payload)