diff --git a/src/sentry/hybridcloud/tasks/deliver_webhooks.py b/src/sentry/hybridcloud/tasks/deliver_webhooks.py index 7e9cf2aec13c..f2cbe731634f 100644 --- a/src/sentry/hybridcloud/tasks/deliver_webhooks.py +++ b/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) diff --git a/src/sentry/options/defaults.py b/src/sentry/options/defaults.py index ddf72f12bf81..03c109eac804 100644 --- a/src/sentry/options/defaults.py +++ b/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. diff --git a/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py b/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py index f776fad27655..04f889c61919 100644 --- a/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py +++ b/tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py @@ -1,6 +1,6 @@ -from datetime import timedelta +from datetime import UTC, datetime, timedelta from typing import Any -from unittest.mock import MagicMock, PropertyMock, patch +from unittest.mock import ANY, MagicMock, PropertyMock, patch import pytest import responses @@ -23,7 +23,6 @@ _claim_and_dispatch, _due_mailbox_heads, drain_mailbox, - drain_mailbox_parallel, maybe_trigger_drain, schedule_webhook_delivery, ) @@ -187,6 +186,8 @@ def test_schedule_due_head_dispatches_past_backoff_head(self, mock_deliver: Magi payload_id=due.id, claimed_count=1, dispatcher=Dispatcher.SCHEDULER, + valid_until=ANY, + mailbox=ANY, ) # The backing-off record keeps its retry schedule. backoff_schedule = backoff.schedule_for @@ -229,6 +230,8 @@ def test_schedule_due_head_claim_stops_at_backoff_record(self, mock_deliver: Mag payload_id=due_one.id, claimed_count=1, dispatcher=Dispatcher.SCHEDULER, + valid_until=ANY, + mailbox=ANY, ) backoff_schedule = backoff.schedule_for backoff.refresh_from_db() @@ -254,6 +257,8 @@ def test_schedule_due_head_does_not_reclaim_active_drain(self, mock_deliver: Mag payload_id=expired_backoff.id, claimed_count=1, dispatcher=Dispatcher.SCHEDULER, + valid_until=ANY, + mailbox=ANY, ) @override_options(DUE_HEAD_OPTIONS) @@ -405,6 +410,8 @@ def test_claim_and_dispatch_claims_in_a_single_query(self, mock_drain: MagicMock payload_id=webhook.id, claimed_count=1, dispatcher=Dispatcher.SCHEDULER, + valid_until=ANY, + mailbox=ANY, ) # Option reads are served from the option store's cache in production, # so only payload-table statements count toward the round-trip budget. @@ -950,7 +957,12 @@ def assert_drain_skips_failed_message( responses.add(responses.POST, url, status=200, body="") records = create_payloads(5, f"{provider}:123", provider=provider) - drain_mailbox(records[0].id, claimed_count=claimed_count, valid_until=fresh_deadline()) + drain_mailbox( + records[0].id, + claimed_count=claimed_count, + valid_until=fresh_deadline(), + mailbox=f"{provider}:123", + ) assert len(responses.calls) == 5 assert WebhookPayload.objects.count() == 1 @@ -965,7 +977,7 @@ def assert_drain_skips_failed_message( class DrainMailboxTest(TestCase): @responses.activate def test_drain_missing_payload(self) -> None: - drain_mailbox(99, claimed_count=MAX_MAILBOX_DRAIN) + drain_mailbox(99, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123") assert len(responses.calls) == 0 @responses.activate @@ -975,7 +987,9 @@ def test_drain_unknown_region(self) -> None: cell_name="lolnope", ) with pytest.raises(CellResolutionError): - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) assert len(responses.calls) == 0 @responses.activate @@ -988,7 +1002,12 @@ def test_drain_success_partial(self) -> None: responses.add(responses.POST, url, status=200, body="") responses.add(responses.POST, url, status=200, body="") records = create_payloads(5, "github:123", provider="github") - drain_mailbox(records[0].id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + records[0].id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) # github is in the skip-on-failure allowlist: failed messages are skipped # and processing continues. All 5 messages are attempted. @@ -1013,7 +1032,9 @@ def test_drain_stops_at_claimed_count(self) -> None: responses.add(responses.POST, url, status=200, body="") records = create_payloads(8, "github:123", provider="github") - drain_mailbox(records[0].id, claimed_count=5, valid_until=fresh_deadline()) + drain_mailbox( + records[0].id, claimed_count=5, valid_until=fresh_deadline(), mailbox="github:123" + ) assert len(responses.calls) == 5 remaining = set(WebhookPayload.objects.values_list("id", flat=True)) @@ -1028,7 +1049,10 @@ def test_drain_stops_at_claimed_count_threaded(self) -> None: records = create_payloads(8, "github:123", provider="github") drain_mailbox( - records[0].id, claimed_count=MIN_RECORDS_PER_THREAD + 1, valid_until=fresh_deadline() + records[0].id, + claimed_count=MIN_RECORDS_PER_THREAD + 1, + valid_until=fresh_deadline(), + mailbox="github:123", ) assert len(responses.calls) == 6 @@ -1055,7 +1079,9 @@ def test_drain_stale_discards_consume_claim_budget_in_waves(self) -> None: fresh = create_payloads(5, "github:123", provider="github") # The claim covered the 3 stale rows plus 3 fresh ones. - drain_mailbox(stale[0].id, claimed_count=6, valid_until=fresh_deadline()) + drain_mailbox( + stale[0].id, claimed_count=6, valid_until=fresh_deadline(), mailbox="github:123" + ) # The 3 stale rows are discarded without requests and only the 3 claimed # fresh rows are delivered; the rest stay for the next claim. @@ -1081,7 +1107,7 @@ def test_drain_discards_stale_rows_instead_of_delivering(self) -> None: head = WebhookPayload.objects.order_by("id").first() assert head - drain_mailbox(head.id, claimed_count=4, valid_until=fresh_deadline()) + drain_mailbox(head.id, claimed_count=4, valid_until=fresh_deadline(), mailbox="github:123") # Only the two fresh rows produce requests; everything is drained. assert len(responses.calls) == 2 @@ -1095,7 +1121,12 @@ def test_drain_stops_on_failure_for_non_allowlisted_provider(self) -> None: responses.add(responses.POST, url, status=200, body="") responses.add(responses.POST, url, status=500, body="") records = create_payloads(5, "jira:123", provider="jira") - drain_mailbox(records[0].id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + records[0].id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="jira:123", + ) # jira is not in the allowlist: processing stops on the first failure # to preserve strict mailbox ordering. @@ -1145,7 +1176,9 @@ def test_drain_batch_deletes_delivered_rows(self) -> None: records = create_payloads(4, "github:123", provider="github") with CaptureQueriesContext(connections["control"]) as ctx: - drain_mailbox(records[0].id, claimed_count=4, valid_until=fresh_deadline()) + drain_mailbox( + records[0].id, claimed_count=4, valid_until=fresh_deadline(), mailbox="github:123" + ) assert len(responses.calls) == 4 assert WebhookPayload.objects.count() == 0 @@ -1164,7 +1197,9 @@ def test_drain_batch_deletes_flush_when_drain_stops_on_failure(self) -> None: responses.add(responses.POST, url, status=500, body="") records = create_payloads(5, "jira:123", provider="jira") - drain_mailbox(records[0].id, claimed_count=5, valid_until=fresh_deadline()) + drain_mailbox( + records[0].id, claimed_count=5, valid_until=fresh_deadline(), mailbox="jira:123" + ) # jira requires strict ordering: the drain stops at the failure, but the # two messages delivered before it must still have their rows removed. @@ -1198,7 +1233,9 @@ def test_drain_batch_deletes_discarded_rows(self) -> None: create_payloads(2, "github:123", provider="github") with CaptureQueriesContext(connections["control"]) as ctx: - drain_mailbox(stale.id, claimed_count=4, valid_until=fresh_deadline()) + drain_mailbox( + stale.id, claimed_count=4, valid_until=fresh_deadline(), mailbox="github:123" + ) # Only the two fresh rows are delivered; the stale and attempts-exhausted # rows are discarded without a request. @@ -1220,7 +1257,9 @@ def test_drain_batch_deletes_are_bounded(self) -> None: records = create_payloads(5, "github:123", provider="github") with CaptureQueriesContext(connections["control"]) as ctx: - drain_mailbox(records[0].id, claimed_count=5, valid_until=fresh_deadline()) + drain_mailbox( + records[0].id, claimed_count=5, valid_until=fresh_deadline(), mailbox="github:123" + ) assert WebhookPayload.objects.count() == 0 # Two full batches during the walk plus the remainder at the end. @@ -1238,7 +1277,9 @@ def test_drain_batch_deletes_span_waves(self) -> None: records = create_payloads(8, "github:123", provider="github") with CaptureQueriesContext(connections["control"]) as ctx: - drain_mailbox(records[0].id, claimed_count=8, valid_until=fresh_deadline()) + drain_mailbox( + records[0].id, claimed_count=8, valid_until=fresh_deadline(), mailbox="github:123" + ) assert len(responses.calls) == 8 assert WebhookPayload.objects.count() == 0 @@ -1251,7 +1292,12 @@ def test_drain_mailbox_multiple_consecutive_failures(self) -> None: url = "http://us.testserver/extensions/github/webhook/" responses.add(responses.POST, url, status=500, body="") records = create_payloads(5, "github:123", provider="github") - drain_mailbox(records[0].id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + records[0].id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) # All 5 messages are attempted even though all fail. assert len(responses.calls) == 5 @@ -1272,7 +1318,12 @@ def test_drain_success(self) -> None: body="", ) records = create_payloads(3, "github:123") - drain_mailbox(records[0].id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + records[0].id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) # Mailbox should be empty assert not WebhookPayload.objects.filter().exists() @@ -1285,7 +1336,9 @@ def test_drain_too_many_attempts(self) -> None: cell_name="us", attempts=MAX_ATTEMPTS, ) - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) assert not WebhookPayload.objects.filter(id=webhook_one.id).exists() assert len(responses.calls) == 0 @@ -1297,7 +1350,9 @@ def test_drain_more_than_max_attempts(self) -> None: cell_name="us", attempts=MAX_ATTEMPTS + 1, ) - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) assert not WebhookPayload.objects.filter(id=webhook_one.id).exists() assert len(responses.calls) == 0 @@ -1315,7 +1370,9 @@ def test_drain_fatality(self) -> None: cell_name="us", ) with pytest.raises(ValueError): - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) hook = WebhookPayload.objects.filter(id=webhook_one.id).first() assert hook assert hook.attempts == 1 @@ -1334,7 +1391,9 @@ def test_drain_host_error(self) -> None: mailbox_name="github:123", cell_name="us", ) - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) hook = WebhookPayload.objects.filter(id=webhook_one.id).first() assert hook assert len(responses.calls) == 1 @@ -1354,7 +1413,9 @@ def test_drain_conflict(self) -> None: mailbox_name="github:123", cell_name="us", ) - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) assert not WebhookPayload.objects.filter(id=webhook_one.id).exists() assert len(responses.calls) == 1 @@ -1371,7 +1432,9 @@ def test_drain_api_error_unauthorized(self) -> None: mailbox_name="github:123", cell_name="us", ) - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) hook = WebhookPayload.objects.filter(id=webhook_one.id).first() # We don't retry 401 assert hook is None @@ -1390,7 +1453,9 @@ def test_drain_api_error_bad_request(self) -> None: mailbox_name="github:123", cell_name="us", ) - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) hook = WebhookPayload.objects.filter(id=webhook_one.id).first() # We don't retry 400 assert hook is None @@ -1409,7 +1474,9 @@ def test_drain_api_error_forbidden(self) -> None: mailbox_name="github:123", cell_name="us", ) - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) hook = WebhookPayload.objects.filter(id=webhook_one.id).first() # We don't retry 403 assert hook is None @@ -1429,7 +1496,9 @@ def test_drain_not_found(self) -> None: cell_name="us", request_path="/plugins/github/organizations/123/webhook/", ) - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="plugins:123" + ) # We don't retry if the region 404s hook = WebhookPayload.objects.filter(id=webhook_one.id).first() @@ -1446,7 +1515,9 @@ def test_drain_timeout(self) -> None: mailbox_name="github:123", cell_name="us", ) - drain_mailbox(webhook_one.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) hook = WebhookPayload.objects.filter(id=webhook_one.id).first() assert hook assert hook.schedule_for > timezone.now() @@ -1464,36 +1535,15 @@ def test_drain_success_api_gateway_address(self) -> None: body="", ) records = create_payloads(3, "github:123") - drain_mailbox(records[0].id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) - - # Mailbox should be empty - assert not WebhookPayload.objects.filter().exists() - - -@control_silo_test -class DrainMailboxParallelShimTest(TestCase): - @responses.activate - @override_cells(cell_config) - def test_in_flight_drain_still_delivers(self) -> None: - # Dispatch no longer enqueues this task; drains enqueued by the deploy - # before the drains merged must still bind and deliver. Dies with the shim. - responses.add( - responses.POST, - "http://us.testserver/extensions/github/webhook/", - status=200, - body="", - ) - records = create_payloads(3, "github:123") - - drain_mailbox_parallel( - payload_id=records[0].id, - claimed_count=3, - dispatcher=Dispatcher.PUSH, - valid_until=(timezone.now() + BATCH_SCHEDULE_OFFSET).timestamp(), + drain_mailbox( + records[0].id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), mailbox="github:123", ) - assert not WebhookPayload.objects.exists() + # Mailbox should be empty + assert not WebhookPayload.objects.filter().exists() @control_silo_test @@ -1522,7 +1572,12 @@ def test_slow_delivery_logged(self) -> None: expected_date_added = webhook.date_added.isoformat() with self.assertLogs("sentry.hybridcloud.tasks.deliver_webhooks", level="WARNING") as cm: - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) slow_log = next(r for r in cm.records if "deliver_webhook.slow_delivery" in r.msg) # extra dict from logger becomes attributes on LogRecord at runtime @@ -1552,7 +1607,12 @@ def test_delivery_time_metrics_cell_sent_to(self, mock_metrics: MagicMock) -> No mailbox_name="github:123", cell_name="us", ) - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) delivery_time_tags = self.distribution_tags(mock_metrics, DELIVERY_TIME_METRIC) assert len(delivery_time_tags) == 1 @@ -1580,7 +1640,12 @@ def test_delivery_time_metrics_github_event_type(self, mock_metrics: MagicMock) cell_name="us", provider="github", ) - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123:0:pull_request", + ) delivery_time_tags = self.distribution_tags(mock_metrics, DELIVERY_TIME_METRIC) assert len(delivery_time_tags) == 1 @@ -1604,7 +1669,12 @@ def test_delivery_time_metrics_non_github_event_type(self, mock_metrics: MagicMo cell_name="us", provider="stripe", ) - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="stripe:123", + ) delivery_time_tags = self.distribution_tags(mock_metrics, DELIVERY_TIME_METRIC) assert len(delivery_time_tags) == 1 @@ -1631,7 +1701,12 @@ def test_delivery_time_metrics_github_mailbox_without_event_suffix( cell_name="us", provider="github", ) - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) delivery_time_tags = self.distribution_tags(mock_metrics, DELIVERY_TIME_METRIC) assert len(delivery_time_tags) == 1 @@ -1662,7 +1737,12 @@ def test_drain_conflict_not_counted_as_delivered(self, mock_metrics: MagicMock) ) webhook = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) assert not WebhookPayload.objects.filter(id=webhook.id).exists() assert self.delivery_outcomes(mock_metrics) == ["conflict"] @@ -1680,7 +1760,12 @@ def test_drain_unauthorized_not_counted_as_delivered(self, mock_metrics: MagicMo ) webhook = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) assert not WebhookPayload.objects.filter(id=webhook.id).exists() assert self.delivery_outcomes(mock_metrics) == ["dropped_4xx"] @@ -1698,7 +1783,12 @@ def test_drain_success_still_counted_as_delivered(self, mock_metrics: MagicMock) ) webhook = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) assert self.delivery_outcomes(mock_metrics) == ["ok"] assert len(self.distribution_calls(mock_metrics, DELIVERY_TIME_METRIC)) == 1 @@ -1715,7 +1805,9 @@ def test_serial_conflict_not_counted_as_delivered(self, mock_metrics: MagicMock) ) webhook = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") - drain_mailbox(webhook.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) assert not WebhookPayload.objects.filter(id=webhook.id).exists() assert self.delivery_outcomes(mock_metrics) == ["conflict"] @@ -1733,7 +1825,9 @@ def test_serial_unauthorized_not_counted_as_delivered(self, mock_metrics: MagicM ) webhook = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") - drain_mailbox(webhook.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) assert not WebhookPayload.objects.filter(id=webhook.id).exists() assert self.delivery_outcomes(mock_metrics) == ["dropped_4xx"] @@ -1760,7 +1854,12 @@ def test_dropped_payload_does_not_stall_ordered_mailbox(self, mock_metrics: Magi first = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") second = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") - drain_mailbox(first.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + first.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) assert not WebhookPayload.objects.filter(id=second.id).exists() assert self.delivery_outcomes(mock_metrics) == ["dropped_4xx", "ok"] @@ -1787,7 +1886,12 @@ def test_delivery_tagged_with_provider(self, mock_metrics: MagicMock) -> None: mailbox_name="github:123", cell_name="us", provider="github" ) - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) assert self.tags_for(mock_metrics, DELIVERY_METRIC) == [ {**UNATTRIBUTED, "outcome": "ok", "provider": "github"} @@ -1807,7 +1911,12 @@ def test_failure_tagged_with_provider(self, mock_metrics: MagicMock) -> None: mailbox_name="github:123", cell_name="us", provider="github" ) - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) assert self.tags_for(mock_metrics, "hybridcloud.deliver_webhooks.failure") == [ {"reason": "unauthorized", "destination_region": "us", "provider": "github"} @@ -1829,7 +1938,12 @@ def test_dropped_outcomes_tagged_with_provider(self, mock_metrics: MagicMock) -> mailbox_name="github:123", cell_name="us", provider="github" ) - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) assert self.tags_for(mock_metrics, DELIVERY_METRIC) == [ {**UNATTRIBUTED, "outcome": "conflict", "provider": "github"} @@ -1849,7 +1963,9 @@ def test_serial_dropped_outcome_tagged_with_provider(self, mock_metrics: MagicMo mailbox_name="github:123", cell_name="us", provider="github" ) - drain_mailbox(webhook.id, claimed_count=1, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, claimed_count=1, valid_until=fresh_deadline(), mailbox="github:123" + ) assert self.tags_for(mock_metrics, DELIVERY_METRIC) == [ {**UNATTRIBUTED, "outcome": "dropped_4xx", "provider": "github"} @@ -1870,7 +1986,12 @@ def test_provider_comes_from_the_mailbox_not_the_row(self, mock_metrics: MagicMo webhook = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") webhook.update(provider=None) - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) assert self.tags_for(mock_metrics, DELIVERY_METRIC) == [ {**UNATTRIBUTED, "outcome": "ok", "provider": "github"} @@ -1886,7 +2007,12 @@ def test_provider_falls_back_to_unknown(self, mock_metrics: MagicMock) -> None: ) webhook = self.create_webhook_payload(mailbox_name="legacy", cell_name="us") - drain_mailbox(webhook.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="legacy", + ) assert self.tags_for(mock_metrics, DELIVERY_METRIC) == [ {**UNATTRIBUTED, "outcome": "ok", "provider": "unknown"} @@ -1983,6 +2109,7 @@ def test_delivery_carries_dispatch_attribution(self, mock_metrics: MagicMock) -> claimed_count=1, dispatcher=Dispatcher.SCHEDULER, valid_until=fresh_deadline(), + mailbox="github:123", ) assert self.tags_for(mock_metrics, DELIVERY_METRIC) == [ @@ -2010,6 +2137,7 @@ def test_delivery_time_carries_dispatch_attribution(self, mock_metrics: MagicMoc claimed_count=1, dispatcher=Dispatcher.SCHEDULER, valid_until=fresh_deadline(), + mailbox="github:123", ) assert self.distribution_tags(mock_metrics, DELIVERY_TIME_METRIC) == [ @@ -2040,6 +2168,7 @@ def test_wave_delivery_time_carries_dispatch_attribution(self, mock_metrics: Mag claimed_count=MIN_RECORDS_PER_THREAD + 1, dispatcher=Dispatcher.PUSH, valid_until=fresh_deadline(), + mailbox="github:123", ) expected = { @@ -2179,47 +2308,25 @@ def test_drain_stops_at_the_claim_deadline_not_a_fresh_one(self) -> None: records[0].id, claimed_count=1, valid_until=(timezone.now() + timedelta(minutes=5)).timestamp(), + mailbox="github:123", ) assert len(responses.calls) == 1 assert WebhookPayload.objects.count() == 0 - @responses.activate - @override_cells(cell_config) - def test_undated_claim_reads_deadline_from_its_head_row(self) -> None: - # Drains enqueued before dispatch sent a deadline find it on their rows: - # their claim wrote it there as schedule_for. - responses.add( - responses.POST, "http://us.testserver/extensions/github/webhook/", status=200, body="" - ) - webhook = self.create_webhook_payload( - mailbox_name="github:123", - cell_name="us", - provider="github", - schedule_for=timezone.now() + BATCH_SCHEDULE_OFFSET, - ) - - drain_mailbox(webhook.id, claimed_count=1) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") + def test_dispatch_passes_the_rows_own_deadline(self, mock_drain: MagicMock) -> None: + # The drain must be handed the schedule_for its claim wrote, not a value + # recomputed from an offset that may differ between dispatcher and worker. + webhook = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") - assert len(responses.calls) == 1 - assert WebhookPayload.objects.count() == 0 + schedule_webhook_delivery() - @responses.activate - @override_cells(cell_config) - def test_undated_claim_with_lapsed_rows_stands_down(self) -> None: - # An undated drain whose rows are already claimable belongs to whoever - # claims them next, exactly like a dated drain past its deadline. - responses.add( - responses.POST, "http://us.testserver/extensions/github/webhook/", status=200, body="" + valid_until = datetime.fromtimestamp( + mock_drain.delay.call_args.kwargs["valid_until"], tz=UTC ) - webhook = self.create_webhook_payload( - mailbox_name="github:123", cell_name="us", provider="github" - ) - - drain_mailbox(webhook.id, claimed_count=1) - - assert len(responses.calls) == 0 - assert WebhookPayload.objects.count() == 1 + webhook.refresh_from_db() + assert webhook.schedule_for == valid_until @responses.activate @override_cells(cell_config) @@ -2233,7 +2340,7 @@ def test_fresh_claim_delivers(self, mock_metrics: MagicMock) -> None: ) valid_until = (timezone.now() + BATCH_SCHEDULE_OFFSET).timestamp() - drain_mailbox(webhook.id, claimed_count=1, valid_until=valid_until) + drain_mailbox(webhook.id, claimed_count=1, valid_until=valid_until, mailbox="github:123") assert len(responses.calls) == 1 assert WebhookPayload.objects.count() == 0 @@ -2263,26 +2370,6 @@ def test_claim_is_stale_at_its_deadline_exactly(self, mock_metrics: MagicMock) - {**UNATTRIBUTED, "outcome": "delivery_deadline", "provider": "github"} ] - @responses.activate - @override_cells(cell_config) - def test_drain_enqueued_before_deploy_delivers(self) -> None: - # Drains queued before this deploys carry no deadline and must keep - # running on the one their claim wrote to the rows. - responses.add( - responses.POST, "http://us.testserver/extensions/github/webhook/", status=200, body="" - ) - webhook = self.create_webhook_payload( - mailbox_name="github:123", - cell_name="us", - provider="github", - schedule_for=timezone.now() + BATCH_SCHEDULE_OFFSET, - ) - - drain_mailbox(webhook.id, claimed_count=1, dispatcher=Dispatcher.SCHEDULER) - - assert len(responses.calls) == 1 - assert WebhookPayload.objects.count() == 0 - @responses.activate @patch("sentry.hybridcloud.tasks.deliver_webhooks.metrics") def test_expired_claim_on_deleted_head_still_names_the_provider( @@ -2315,27 +2402,17 @@ def test_expired_claim_without_a_mailbox_reads_it_off_the_head( ) drain_mailbox( - webhook.id, claimed_count=1, dispatcher=Dispatcher.PUSH, valid_until=self.expired() + webhook.id, + claimed_count=1, + dispatcher=Dispatcher.PUSH, + valid_until=self.expired(), + mailbox="github:123", ) assert self.tags_for(mock_metrics, DELIVERY_METRIC) == [ {"dispatcher": "push", "outcome": "delivery_deadline", "provider": "github"} ] - @responses.activate - @override_cells(cell_config) - @patch("sentry.hybridcloud.tasks.deliver_webhooks.metrics") - def test_expired_claim_with_neither_mailbox_nor_head_reports_a_race( - self, mock_metrics: MagicMock - ) -> None: - # Nothing left to name the provider with; the row read is still the race - # site for every drain until dispatch sends the claim. - drain_mailbox(99, claimed_count=1, dispatcher=Dispatcher.PUSH, valid_until=self.expired()) - - assert self.tags_for(mock_metrics, DELIVERY_METRIC) == [ - {"dispatcher": "push", "outcome": "race", "provider": "unknown"} - ] - @responses.activate @override_cells(cell_config) @patch("sentry.hybridcloud.tasks.deliver_webhooks._run_parallel_delivery_batch") @@ -2447,6 +2524,176 @@ def test_release_covers_only_the_tail_behind_delivered_rows(self) -> None: assert record.schedule_for <= timezone.now() +@control_silo_test +class ChainDispatchTest(TestCase): + """ + A strict provider's drain that ends healthy with due work behind it + dispatches the mailbox's next claim itself, while its lineage is within + max_chain_depth links. + """ + + def _respond_ok(self, provider: str = "jira") -> None: + responses.add( + responses.POST, + f"http://us.testserver/extensions/{provider}/webhook/", + status=200, + body="", + ) + + @responses.activate + @override_cells(cell_config) + @override_options({"hybridcloud.webhookpayload.max_chain_depth": 3}) + @patch.object(deliver_webhooks, "MAX_MAILBOX_DRAIN", 3) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") + def test_chains_after_draining_a_full_claim(self, mock_drain: MagicMock) -> None: + self._respond_ok() + records = create_payloads(4, "jira:123", provider="jira") + + drain_mailbox( + records[0].id, claimed_count=3, valid_until=fresh_deadline(), mailbox="jira:123" + ) + + assert len(responses.calls) == 3 + kwargs = mock_drain.delay.call_args.kwargs + assert kwargs["payload_id"] == records[3].id + assert kwargs["dispatcher"] == Dispatcher.CHAIN + assert kwargs["chain_depth"] == 2 + + @responses.activate + @override_cells(cell_config) + @override_options({"hybridcloud.webhookpayload.max_chain_depth": 3}) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") + def test_release_chains_the_tail(self, mock_drain: MagicMock) -> None: + self._respond_ok() + valid_until = timezone.now() + BATCH_SCHEDULE_OFFSET + records = create_payloads(3, "jira:123", provider="jira") + WebhookPayload.objects.filter(id__in=[r.id for r in records]).update( + schedule_for=valid_until + ) + + with patch.object( + deliver_webhooks._MailboxClaim, "nearing_deadline", side_effect=[False, True] + ): + drain_mailbox( + records[0].id, + claimed_count=3, + valid_until=valid_until.timestamp(), + mailbox="jira:123", + ) + + # One delivered, two released — the chain claims the released tail. + assert len(responses.calls) == 1 + kwargs = mock_drain.delay.call_args.kwargs + assert kwargs["payload_id"] == records[1].id + assert kwargs["dispatcher"] == Dispatcher.CHAIN + + @responses.activate + @override_cells(cell_config) + @patch.object(deliver_webhooks, "MAX_MAILBOX_DRAIN", 3) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") + def test_no_chain_at_the_default_depth(self, mock_drain: MagicMock) -> None: + # The ordinary dispatch is the first link, so the default of 1 means a + # finished drain never chains. + self._respond_ok() + records = create_payloads(4, "jira:123", provider="jira") + + drain_mailbox( + records[0].id, claimed_count=3, valid_until=fresh_deadline(), mailbox="jira:123" + ) + + assert len(responses.calls) == 3 + mock_drain.delay.assert_not_called() + + @responses.activate + @override_cells(cell_config) + @override_options({"hybridcloud.webhookpayload.max_chain_depth": 3}) + @patch.object(deliver_webhooks, "MAX_MAILBOX_DRAIN", 3) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") + def test_no_chain_past_the_depth_ceiling(self, mock_drain: MagicMock) -> None: + self._respond_ok() + records = create_payloads(4, "jira:123", provider="jira") + + drain_mailbox( + records[0].id, + claimed_count=3, + valid_until=fresh_deadline(), + mailbox="jira:123", + chain_depth=3, + ) + + mock_drain.delay.assert_not_called() + + @responses.activate + @override_cells(cell_config) + @override_options({"hybridcloud.webhookpayload.max_chain_depth": 3}) + @patch.object(deliver_webhooks, "MAX_MAILBOX_DRAIN", 3) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") + def test_no_chain_for_skip_on_failure_provider(self, mock_drain: MagicMock) -> None: + # Due-head providers would fork a new pipeline every scheduler cycle. + self._respond_ok("github") + records = create_payloads(4, "github:123", provider="github") + + drain_mailbox( + records[0].id, claimed_count=3, valid_until=fresh_deadline(), mailbox="github:123" + ) + + assert len(responses.calls) == 3 + mock_drain.delay.assert_not_called() + + @responses.activate + @override_cells(cell_config) + @override_options({"hybridcloud.webhookpayload.max_chain_depth": 3}) + @patch.object(deliver_webhooks, "MAX_MAILBOX_DRAIN", 3) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") + def test_no_chain_after_a_failure_stop(self, mock_drain: MagicMock) -> None: + url = "http://us.testserver/extensions/jira/webhook/" + responses.add(responses.POST, url, status=200, body="") + responses.add(responses.POST, url, status=500, body="") + records = create_payloads(4, "jira:123", provider="jira") + + drain_mailbox( + records[0].id, claimed_count=3, valid_until=fresh_deadline(), mailbox="jira:123" + ) + + assert len(responses.calls) == 2 + mock_drain.delay.assert_not_called() + + @responses.activate + @override_cells(cell_config) + @override_options({"hybridcloud.webhookpayload.max_chain_depth": 3}) + @patch.object(deliver_webhooks, "MAX_MAILBOX_DRAIN", 3) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") + def test_no_chain_on_a_short_claim(self, mock_drain: MagicMock) -> None: + # A claim under the cap means the due prefix ended; nothing to chain to. + self._respond_ok() + records = create_payloads(2, "jira:123", provider="jira") + + drain_mailbox( + records[0].id, claimed_count=2, valid_until=fresh_deadline(), mailbox="jira:123" + ) + + assert len(responses.calls) == 2 + mock_drain.delay.assert_not_called() + + @responses.activate + @override_cells(cell_config) + @override_options({"hybridcloud.webhookpayload.max_chain_depth": 3}) + @patch.object(deliver_webhooks, "MAX_MAILBOX_DRAIN", 3) + @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") + def test_no_chain_while_another_dispatcher_holds_the_lock(self, mock_drain: MagicMock) -> None: + self._respond_ok() + records = create_payloads(4, "jira:123", provider="jira") + cache.add("wh:drain_active:jira:123", 1, timeout=15) + + drain_mailbox( + records[0].id, claimed_count=3, valid_until=fresh_deadline(), mailbox="jira:123" + ) + + mock_drain.delay.assert_not_called() + # The other dispatcher's guard must survive the skipped chain. + assert cache.get("wh:drain_active:jira:123") is not None + + @control_silo_test class PushTriggerTest(MetricCallsMixin, TestCase): @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") @@ -2457,6 +2704,8 @@ def test_push_trigger_enqueues_drain_for_idle_mailbox(self, mock_drain: MagicMoc payload_id=webhook.id, claimed_count=1, dispatcher=Dispatcher.PUSH, + valid_until=ANY, + mailbox=ANY, ) # The batch is claimed before dispatch; the claim is what keeps other # dispatchers off the mailbox while the drain runs. @@ -2498,6 +2747,8 @@ def test_push_trigger_drains_from_mailbox_head_not_new_payload( payload_id=older_webhook.id, claimed_count=2, dispatcher=Dispatcher.PUSH, + valid_until=ANY, + mailbox=ANY, ) @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") @@ -2534,6 +2785,8 @@ def test_push_trigger_due_head_dispatches_past_backoff_head( payload_id=due.id, claimed_count=1, dispatcher=Dispatcher.PUSH, + valid_until=ANY, + mailbox=ANY, ) @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") @@ -2604,6 +2857,8 @@ def test_scheduler_skips_locked_mailboxes(self, mock_drain: MagicMock) -> None: payload_id=webhook_b.id, claimed_count=1, dispatcher=Dispatcher.SCHEDULER, + valid_until=ANY, + mailbox=ANY, ) @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") @@ -2646,7 +2901,12 @@ def test_push_trigger_fires_immediately_after_drain_completes( body="", ) webhook_one = self.create_webhook_payload(mailbox_name="github:123", cell_name="us") - drain_mailbox(webhook_one.id, claimed_count=MAX_MAILBOX_DRAIN, valid_until=fresh_deadline()) + drain_mailbox( + webhook_one.id, + claimed_count=MAX_MAILBOX_DRAIN, + valid_until=fresh_deadline(), + mailbox="github:123", + ) # The drain emptied the mailbox; a new webhook arriving now must be able to # trigger a fresh drain right away. @@ -2656,6 +2916,8 @@ def test_push_trigger_fires_immediately_after_drain_completes( payload_id=webhook_two.id, claimed_count=1, dispatcher=Dispatcher.PUSH, + valid_until=ANY, + mailbox=ANY, ) @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") @@ -2689,6 +2951,8 @@ def test_push_trigger_reports_parallel_drain_for_deep_mailbox( payload_id=records[0].id, claimed_count=MIN_RECORDS_PER_THREAD + 1, dispatcher=Dispatcher.PUSH, + valid_until=ANY, + mailbox=ANY, ) assert self.tags_for(mock_metrics, DISPATCH_METRIC) == [ {"dispatcher": "push", "drain": "parallel", "provider": "github"} @@ -2708,6 +2972,8 @@ def test_push_trigger_reports_sequential_drain_for_shallow_mailbox( payload_id=records[0].id, claimed_count=MIN_RECORDS_PER_THREAD, dispatcher=Dispatcher.PUSH, + valid_until=ANY, + mailbox=ANY, ) assert self.tags_for(mock_metrics, DISPATCH_METRIC) == [ {"dispatcher": "push", "drain": "sequential", "provider": "github"} @@ -2727,11 +2993,8 @@ def test_push_trigger_reports_sequential_drain_for_deep_strict_mailbox( {"dispatcher": "push", "drain": "sequential", "provider": "jira"} ] - @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox_parallel") @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") - def test_push_trigger_claim_keeps_scheduler_off( - self, mock_drain: MagicMock, mock_drain_parallel: MagicMock - ) -> None: + def test_push_trigger_claim_keeps_scheduler_off(self, mock_drain: MagicMock) -> None: create_payloads(3, "github:123") maybe_trigger_drain("github:123") @@ -2742,8 +3005,6 @@ def test_push_trigger_claim_keeps_scheduler_off( # The push trigger's claim moved the head past the drain deadline, so the # scheduler must not double-dispatch a drain for this mailbox. assert mock_drain.delay.call_count == 1 - # Tripwire: dispatch must never reach the transitional shim task. - assert mock_drain_parallel.delay.call_count == 0 @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox") def test_scheduler_claim_blocks_push_trigger(self, mock_drain: MagicMock) -> None: