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

src/sentry/hybridcloud/tasks/deliver_webhooks.py

@@ -264,6 +264,7 @@ class Dispatcher(enum.StrEnum):
 
     PUSH = "push"
     SCHEDULER = "scheduler"
+    CHAIN = "chain"
 
 
 @dataclasses.dataclass(frozen=True)
@@ -350,10 +351,10 @@ def nearing_deadline(self) -> bool:
         """Whether to stop delivering and release, rather than run into the deadline mid-wave."""
         return timezone.now() >= self.valid_until - RELEASE_MARGIN
 
-    def release_remainder(self, next_id: int, *, extra: Mapping[str, Any]) -> None:
+    def release_remainder(self, next_id: int, *, extra: Mapping[str, Any]) -> int:
         """
         Return the claim's unworked tail to the mailbox so the next dispatcher can
-        claim it now instead of at the deadline.
+        claim it now instead of at the deadline; returns how many rows went back.
 
         Matching the exact `schedule_for` this claim wrote is what makes the
         release safe: a record another claim took carries that claim's timestamp,
@@ -374,6 +375,7 @@ def release_remainder(self, next_id: int, *, extra: Mapping[str, Any]) -> None:
                 tags={**self.delivery_tags, "outcome": "released"},
             )
         logger.info("deliver_webhook.deadline_release", extra={**extra, "released": released})
+        return released
 
     def next_slice(self, start_id: int, size: int) -> list[WebhookPayload] | None:
         """
@@ -404,47 +406,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:
-            # Whoever claimed the mailbox next is delivering the rest. Every
-            # drain resolves through this read until dispatch sends the claim,
-            # so this is where a lost race shows up.
-            _record_lost_head(
-                payload_id,
-                dispatcher=dispatcher,
-                provider=_provider_from_mailbox(mailbox),
-                log_key="deliver_webhook.potential_race",
-            )
-            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
@@ -541,7 +513,7 @@ def _record_dispatch(
 
 
 def _claim_and_dispatch(
-    head_id: int, mailbox_name: str, *, dispatcher: Dispatcher
+    head_id: int, mailbox_name: str, *, dispatcher: Dispatcher, chain_depth: int = 0
 ) -> DispatchOutcome:
     """
     Claim a batch for the mailbox and dispatch its drain.
@@ -564,15 +536,10 @@ def _claim_and_dispatch(
     claim = _claim_mailbox_batch(head_id, mailbox_name, dispatcher=dispatcher)
     if claim is None:
         return DispatchOutcome.NOT_DUE
-    # Only the arguments workers from the previous deploy bind: an unknown kwarg
-    # is a TypeError the taskbroker discards without retry. The drain recovers
-    # the mailbox and deadline from its head row; the full claim shape
-    # (`task_args`) starts crossing the wire one deploy later.
-    drain_mailbox.delay(
-        payload_id=claim.head_id,
-        claimed_count=claim.claimed,
-        dispatcher=claim.dispatcher,
-    )
+    task_args = claim.task_args()
+    if chain_depth:
+        task_args["chain_depth"] = chain_depth
+    drain_mailbox.delay(**task_args)
     outcome = DispatchOutcome.PARALLEL if claim.threaded else DispatchOutcome.SEQUENTIAL
     _record_dispatch(
         dispatcher=dispatcher,
@@ -941,31 +908,78 @@ 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,
     chain_depth: int = 1,
 ) -> 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.
-    `chain_depth` — which link of a chain this drain is, an ordinary dispatch
-    being the first — is accepted ahead of drain chaining for the same reason.
+    The arguments are one claim flattened for the wire (`_MailboxClaim.task_args`).
+    `chain_depth` is which link of a chain this drain is, an ordinary dispatch
+    being the first.
     """
     claim = _begin_drain(payload_id, claimed_count, dispatcher, valid_until, mailbox)
-    if claim is not None:
-        _drain_mailbox(claim)
+    if claim is None:
+        return
+    if _drain_mailbox(claim):
+        _maybe_chain(claim, chain_depth)
+
+
+def _maybe_chain(claim: _MailboxClaim, chain_depth: int) -> None:
+    """
+    Dispatch the mailbox's next claim directly, skipping the scheduler's
+    re-discovery gap, while this lineage is within max_chain_depth links —
+    at the option's default of 1 the ordinary dispatch is the whole chain.
+
+    Strict providers only: their absolute-head gate admits one claim at a time,
+    so a chain stays a single lineage per mailbox — a due-head provider would
+    fork a new one every scheduler cycle. The claim gate still settles
+    ownership; a concurrent dispatcher winning it continues the lineage.
+    """
+    if claim.skip_on_failure:
+        return
+    if chain_depth >= options.get("hybridcloud.webhookpayload.max_chain_depth"):
+        return
+    mailbox_name = claim.mailbox_name
+    guard = _acquire_drain_guard(mailbox_name)
+    if not guard:
+        # Held by another dispatcher, or the cache is unreachable: either way
+        # the scheduler covers the mailbox.
+        return
+    try:
+        head = (
+            WebhookPayload.objects.filter(mailbox_name=mailbox_name)
+            .order_by("id")
+            .values_list("id", "schedule_for")
+            .first()
+        )
+        if head is not None and _is_due(head[1]):
+            _claim_and_dispatch(
+                head[0], mailbox_name, dispatcher=Dispatcher.CHAIN, chain_depth=chain_depth + 1
+            )
+    except Exception:
+        # This drain's work is already delivered; failing the task here would
+        # only retry a finished drain. The scheduler picks the mailbox up.
+        logger.exception("deliver_webhook.chain_failed")
+    finally:
+        if guard:
+            _release_drain_lock(mailbox_name)
 
 
-def _drain_mailbox(claim: _MailboxClaim) -> None:
+def _drain_mailbox(claim: _MailboxClaim) -> bool:
     """
     Deliver the claimed records until a strict provider's record fails, the claim
     nears its deadline, or all of them have been processed. Skip-on-failure claims
     deliver in concurrent waves sized to the work left, falling back to one record
     at a time as the tail shrinks; strict claims always deliver in order.
 
+    Returns whether the drain was healthy and left due work behind — it consumed
+    a full claim, or released a tail it had been delivering toward — the signal
+    `_maybe_chain` acts on.
+
     The drain holds no lock, so it must not deliver past the records its dispatcher
     claimed: beyond them the mailbox head is due again and another dispatcher may
     already be draining it.
@@ -990,17 +1004,20 @@ def _drain_mailbox(claim: _MailboxClaim) -> None:
         while True:
             extra = {**log_context, "delivered": delivered}
             if claim.lapsed(log_key="deliver_webhook.delivery_deadline", extra=extra):
-                break
+                return False
             if claim.nearing_deadline():
-                claim.release_remainder(current_id, extra=extra)
-                break
+                released = claim.release_remainder(current_id, extra=extra)
+                # A drain that delivered nothing before its soft-stop spent its
+                # window in the queue — saturation, exactly when a chain would
+                # add queue load.
+                return released > 0 and failed == 0 and delivered > 0
 
             if index >= len(records):
                 # Slices of 100 keep query duration down and avoid reading records
                 # a failure earlier in the claim means we never get to.
                 fetched = claim.next_slice(current_id, min(100, remaining))
                 if fetched is None:
-                    return
+                    return False
                 if not fetched:
                     if failed > 0:
                         logger.info(
@@ -1009,7 +1026,7 @@ def _drain_mailbox(claim: _MailboxClaim) -> None:
                         )
                     else:
                         logger.debug("deliver_webhook.delivery_complete", extra=extra)
-                    return
+                    return False
                 records = fetched
                 index = 0
 
@@ -1048,7 +1065,7 @@ def _drain_mailbox(claim: _MailboxClaim) -> None:
                         # For providers that require strict ordering, stop on the
                         # first failure so subsequent messages are not delivered
                         # out of order.
-                        return
+                        return False
                     # For allowlisted providers: skip the failed message and
                     # continue. It has already been rescheduled by deliver_message.
 
@@ -1060,7 +1077,9 @@ def _drain_mailbox(claim: _MailboxClaim) -> None:
                     "deliver_webhook.claim_exhausted",
                     extra={**log_context, "delivered": delivered},
                 )
-                return
+                # A claim at the cap saw nothing but due records and stopped at
+                # the boundary, so the prefix likely continues past it.
+                return failed == 0 and claim.claimed == MAX_MAILBOX_DRAIN
     finally:
         deleter.flush()
 
@@ -1275,31 +1294,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,
-    chain_depth: int = 1,
-) -> 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)

src/sentry/options/defaults.py

@@ -2611,6 +2611,15 @@
     ],
     flags=FLAG_ALLOW_EMPTY | FLAG_AUTOMATOR_MODIFIABLE,
 )
+# How many chained drains a strict provider's mailbox may run per lineage,
+# counting the ordinary dispatch as the first link: at 1 a finished drain never
+# chains, and each increment lets a busy mailbox re-dispatch itself once more
+# before falling back to the scheduler.
+register(
+    "hybridcloud.webhookpayload.max_chain_depth",
+    default=1,
+    flags=FLAG_AUTOMATOR_MODIFIABLE,
+)
 # Dispatch skip-on-failure providers' mailboxes from their oldest due record
 # instead of gating on the absolute head, so one record in retry backoff cannot
 # hide every due record behind it. Strict-ordering providers keep the gate.