From bd2bb217e04e04428862d9bd80841676433f0a71 Mon Sep 17 00:00:00 2001 From: William Durand Date: Wed, 2 Sep 2026 15:32:54 +0200 Subject: [PATCH] Retry scanner webhook deliveries instead of a fixed grace period --- docs/topics/development/scanner_pipeline.md | 42 +++ src/olympia/constants/config_keys.py | 5 + src/olympia/constants/reviewers.py | 2 - src/olympia/constants/scanners.py | 3 + src/olympia/lib/settings_base.py | 1 + .../management/commands/auto_approve.py | 30 +- src/olympia/reviewers/tests/test_commands.py | 55 ++- src/olympia/scanners/admin.py | 1 + .../0090_scannerresult_delivery_attempts.py | 22 ++ src/olympia/scanners/models.py | 10 + src/olympia/scanners/tasks.py | 209 +++++++++-- src/olympia/scanners/tests/test_tasks.py | 348 ++++++++++++++++++ 12 files changed, 672 insertions(+), 56 deletions(-) create mode 100644 src/olympia/scanners/migrations/0090_scannerresult_delivery_attempts.py diff --git a/docs/topics/development/scanner_pipeline.md b/docs/topics/development/scanner_pipeline.md index 42354aee58ee..4dc619309302 100644 --- a/docs/topics/development/scanner_pipeline.md +++ b/docs/topics/development/scanner_pipeline.md @@ -108,6 +108,7 @@ field: The `results` field should contain the same data structure as a synchronous response would return. +(skipping-an-event)= #### Skipping an event Scanners can use the `204 No Content` HTTP status code to indicate that they @@ -115,6 +116,47 @@ intentionally skipped the event (e.g., the event is not relevant for this scanner). No results will be stored for the scanner result associated with this event. +(scanner-delivery-retries)= +### Delivery retries + +Some events block auto-approval: a version submitted while a scanner is +subscribed to `during_validation` or `on_version_created` is not auto-approved +until that scanner is done with it. A [scanner result](#scanner-results) counts +as done when the scanner [skipped the event](#skipping-an-event) or sent its +`matchedRules`, [synchronously](#synchronous-response) or +[asynchronously](#asynchronous-scanning). + +A scanner can fail to send its results back, e.g., because of an outage. Rather +than waiting forever, AMO delivers the webhook again, with a delay that doubles +after each attempt: + +| Delivery | Sent | +| -------- | --------------------------------------- | +| 1 | when the event occurs | +| 2 | 1 hour after the previous delivery | +| 3 | 2 hours after the previous delivery | +| 4 | 4 hours after the previous delivery | +| 5 | 8 hours after the previous delivery | + +That is `WEBHOOK_MAX_DELIVERY_ATTEMPTS` deliveries over 15 hours, the initial +call included. The first delay is configurable in seconds through the +`scanner-webhook-retry-initial-delay` config key. Retries are scheduled by the +`auto_approve` command (except in `--dry-run` mode) and performed by the +`retry_webhook_deliveries_on_version` task. + +A retry sends the same payload as the initial call, to the same +`scanner_result_url`, so a scanner that eventually answers a retry does not need +to do anything differently. Every delivery increments `delivery_attempts` on the +scanner result, which is visible in the admin; the counter is incremented before +the call, so failed calls count too. + +AMO gives up on a scanner for a given version when the attempts are exhausted, +when the payload can no longer be rebuilt (e.g., the uploaded file a +`during_validation` payload points to is gone), or when no scanner result was +ever created for the event. The version is then flagged for human review with +the `WAITING_ON_SCANNERS` reason, so it leaves auto-approval and reaches the +reviewer queue instead of remaining stuck. + (scanner-annotations)= ### Annotations diff --git a/src/olympia/constants/config_keys.py b/src/olympia/constants/config_keys.py index fa28fd425538..098c676c8f6f 100644 --- a/src/olympia/constants/config_keys.py +++ b/src/olympia/constants/config_keys.py @@ -60,6 +60,11 @@ def dump(self, value): REVIEWERS_MOTD = ConfigKey('reviewers_review_motd') +# Delay in seconds before the first webhook retry, doubled at each attempt. +SCANNER_WEBHOOK_RETRY_INITIAL_DELAY = IntConfigKey( + 'scanner-webhook-retry-initial-delay', 60 * 60 +) + SITE_NOTICE = ConfigKey('site_notice') SUBMIT_NOTIFICATION_WARNING = ConfigKey('submit_notification_warning') diff --git a/src/olympia/constants/reviewers.py b/src/olympia/constants/reviewers.py index d62cbc443cbe..4325576556c4 100644 --- a/src/olympia/constants/reviewers.py +++ b/src/olympia/constants/reviewers.py @@ -80,5 +80,3 @@ class HELD_DECISION_CHOICES(StrEnumChoices): HELD_DECISION_CHOICES.add_subset('ADDON', ('YES', 'CANCEL')) HELD_DECISION_CHOICES.add_subset('OTHER', ('YES', 'NO')) - -WAIT_ON_SCANNERS_TIMEOUT = 7200 # seconds (2 hours) diff --git a/src/olympia/constants/scanners.py b/src/olympia/constants/scanners.py index 82458cbbc2c5..85d149514be8 100644 --- a/src/olympia/constants/scanners.py +++ b/src/olympia/constants/scanners.py @@ -94,6 +94,9 @@ WEBHOOK_ON_VERSION_CREATED, ] +# Max number of webhook deliveries for a single scanner result. +WEBHOOK_MAX_DELIVERY_ATTEMPTS = 5 + # Special rule name used as fallback when a scanner has no better rule to # associate with an annotation. ANNOTATIONS_RULE_NAME = 'ANNOTATIONS' diff --git a/src/olympia/lib/settings_base.py b/src/olympia/lib/settings_base.py index 24a4d2f6a41d..7ffd01bfa499 100644 --- a/src/olympia/lib/settings_base.py +++ b/src/olympia/lib/settings_base.py @@ -831,6 +831,7 @@ def get_language_url_map(): 'olympia.devhub.tasks.validate_upload': {'queue': 'devhub'}, 'olympia.files.tasks.repack_fileupload': {'queue': 'devhub'}, 'olympia.scanners.tasks.call_webhooks_during_validation': {'queue': 'devhub'}, + 'olympia.scanners.tasks.retry_webhook_deliveries_on_version': {'queue': 'devhub'}, 'olympia.scanners.tasks.run_narc_on_version': {'queue': 'devhub'}, 'olympia.scanners.tasks.run_yara': {'queue': 'devhub'}, 'olympia.versions.tasks.call_webhooks_on_source_code_uploaded': {'queue': 'devhub'}, diff --git a/src/olympia/reviewers/management/commands/auto_approve.py b/src/olympia/reviewers/management/commands/auto_approve.py index d1ebc237a4bf..5af534e33cb4 100644 --- a/src/olympia/reviewers/management/commands/auto_approve.py +++ b/src/olympia/reviewers/management/commands/auto_approve.py @@ -1,5 +1,4 @@ from collections import Counter -from datetime import datetime, timedelta from django.conf import settings from django.core.management.base import BaseCommand @@ -15,7 +14,6 @@ from olympia.abuse.tasks import report_decision_to_cinder_and_notify from olympia.amo.decorators import use_primary_db from olympia.constants.abuse import DECISION_ACTIONS -from olympia.constants.reviewers import WAIT_ON_SCANNERS_TIMEOUT from olympia.files.utils import lock from olympia.lib.crypto.signing import SigningError from olympia.reviewers.models import ( @@ -27,7 +25,10 @@ ) from olympia.reviewers.utils import ReviewHelper from olympia.scanners.models import ScannerResult -from olympia.scanners.tasks import run_narc_on_version +from olympia.scanners.tasks import ( + retry_webhook_deliveries_on_version, + run_narc_on_version, +) from olympia.versions.models import Version @@ -142,13 +143,20 @@ def process(self, version_id): summary is not None and summary.scanner_actions_executed is not False ) + is_waiting_on_scanners = ( + AutoApprovalSummary.check_is_waiting_on_scanners(version) + ) + if is_waiting_on_scanners and not self.dry_run: + # In case a scanner never sent its results back. + retry_webhook_deliveries_on_version.delay(version.pk) + if already_executed: log.info( 'Not running run_actions() on version %s because it ' 'has already been executed', version.pk, ) - elif AutoApprovalSummary.check_is_waiting_on_scanners(version): + elif is_waiting_on_scanners: log.info( 'Not running run_actions() on version %s because it ' 'is still waiting on scanners', @@ -281,7 +289,6 @@ def disapprove(self, version): 'has_auto_approval_disabled': ( NeedsHumanReview.REASONS.AUTO_APPROVAL_DISABLED ), - 'is_waiting_on_scanners': NeedsHumanReview.REASONS.WAITING_ON_SCANNERS, } # For the specific reasons that cause an add-on to be added to the # (human) review queue, we add the corresponding NeedsHumanReview flag @@ -296,18 +303,6 @@ def disapprove(self, version): has_decision_waiting_for_2nd_level_approval = ( version.contentdecision_set.awaiting_action().exists() ) - # AutoApprovalSummary is created on the very first `auto_approve` - # run. When we're still waiting on scanners after a long time, it - # might mean a scanner has had an issue and will likely never send - # its results. That's why we should NHR the version. - should_still_wait_on_scanners = ( - version.autoapprovalsummary.is_waiting_on_scanners - and datetime.now() - <= ( - version.autoapprovalsummary.created - + timedelta(seconds=WAIT_ON_SCANNERS_TIMEOUT) - ) - ) already_has_same_active_nhr = version.needshumanreview_set.filter( reason=reason, is_active=True ).exists() @@ -316,7 +311,6 @@ def disapprove(self, version): and not version.pending_rejection and not has_decision_waiting_for_2nd_level_approval and not already_has_same_active_nhr - and not should_still_wait_on_scanners ): NeedsHumanReview.objects.create(version=version, reason=reason) diff --git a/src/olympia/reviewers/tests/test_commands.py b/src/olympia/reviewers/tests/test_commands.py index 8fc5a4bafd13..9965c53442cb 100644 --- a/src/olympia/reviewers/tests/test_commands.py +++ b/src/olympia/reviewers/tests/test_commands.py @@ -615,23 +615,14 @@ def test_disapproves_has_auto_approval_disabled(self): assert nhr.reason == NeedsHumanReview.REASONS.AUTO_APPROVAL_DISABLED assert nhr.is_active - def test_disapproves_is_waiting_on_scanners_after_grace_period(self): - summary = AutoApprovalSummary(is_waiting_on_scanners=True) - summary.created = datetime.now() - timedelta(hours=3) - self.version.autoapprovalsummary = summary - command = auto_approve.Command() - command.disapprove(self.version) - nhr = self.version.needshumanreview_set.get() - assert nhr.reason == NeedsHumanReview.REASONS.WAITING_ON_SCANNERS - assert nhr.is_active - - def test_disapproves_is_waiting_on_scanners_within_grace_period(self): - summary = AutoApprovalSummary(is_waiting_on_scanners=True) - summary.created = datetime.now() - timedelta(hours=1) - self.version.autoapprovalsummary = summary + def test_does_not_disapprove_is_waiting_on_scanners(self): + # We only flag the version once we have given up, in the scanners task. + self.version.autoapprovalsummary = AutoApprovalSummary( + is_waiting_on_scanners=True + ) command = auto_approve.Command() command.disapprove(self.version) - assert not self.version.needshumanreview_set.filter(is_active=True).exists() + assert not self.version.needshumanreview_set.exists() def test_disapproves_is_promoted_but_decision_waiting_for_2nd_level_exists(self): self.version.autoapprovalsummary = AutoApprovalSummary( @@ -1095,6 +1086,40 @@ def test_only_executes_run_actions_once_after_waiting_on_scanners( assert not run_actions_mock.called + @mock.patch( + 'olympia.reviewers.management.commands.auto_approve.' + 'retry_webhook_deliveries_on_version' + ) + def test_retries_webhook_deliveries(self, retry_mock): + # Keep the version out of auto-approval so that it remains a candidate. + AddonReviewerFlags.objects.create(addon=self.addon, auto_approval_disabled=True) + scanner_result = self.create_pending_webhook_scanner_result() + + call_command('auto_approve') + + retry_mock.delay.assert_called_with(self.version.pk) + + # Nothing to retry once the scanner has sent its results. + retry_mock.reset_mock() + scanner_result.update(results={'matchedRules': []}) + + call_command('auto_approve') + + assert not retry_mock.delay.called + + @mock.patch( + 'olympia.reviewers.management.commands.auto_approve.' + 'retry_webhook_deliveries_on_version' + ) + def test_does_not_retry_webhook_deliveries_in_dry_run(self, retry_mock): + # Keep the version out of auto-approval so that it remains a candidate. + AddonReviewerFlags.objects.create(addon=self.addon, auto_approval_disabled=True) + self.create_pending_webhook_scanner_result() + + call_command('auto_approve', '--dry-run') + + assert not retry_mock.delay.called + @mock.patch.object(ScannerResult, 'run_actions') def test_langpack_does_not_wait_on_scanners(self, run_actions_mock): self.addon.update(type=amo.ADDON_LPAPP) diff --git a/src/olympia/scanners/admin.py b/src/olympia/scanners/admin.py index e30fe1b65fdf..d85238bfed7f 100644 --- a/src/olympia/scanners/admin.py +++ b/src/olympia/scanners/admin.py @@ -605,6 +605,7 @@ class ScannerResultAdmin(AbstractScannerResultAdminMixin, AMOModelAdmin): 'guid', 'formatted_scanner', 'created', + 'delivery_attempts', formatted_matched_rules_with_files_and_data, 'formatted_results', 'activity_log', diff --git a/src/olympia/scanners/migrations/0090_scannerresult_delivery_attempts.py b/src/olympia/scanners/migrations/0090_scannerresult_delivery_attempts.py new file mode 100644 index 000000000000..d2b94bc8a93e --- /dev/null +++ b/src/olympia/scanners/migrations/0090_scannerresult_delivery_attempts.py @@ -0,0 +1,22 @@ +# Generated by Django 5.2.16 on 2026-09-02 09:55 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ('scanners', '0089_alter_scannerwebhook_name'), + ] + + operations = [ + migrations.AddField( + model_name='scannerresult', + name='delivery_attempts', + field=models.PositiveSmallIntegerField( + default=0, + help_text=( + 'Number of times the webhook has been called for this result' + ), + ), + ), + ] diff --git a/src/olympia/scanners/models.py b/src/olympia/scanners/models.py index f4ffb1f9c303..007090dab4a6 100644 --- a/src/olympia/scanners/models.py +++ b/src/olympia/scanners/models.py @@ -440,6 +440,10 @@ class ScannerResult(AbstractScannerResult): 'ScannerRule', through='ScannerMatch', related_name='results' ) has_matches = models.BooleanField(null=True) + delivery_attempts = models.PositiveSmallIntegerField( + default=0, + help_text='Number of times the webhook has been called for this result', + ) class Meta(AbstractScannerResult.Meta): db_table = 'scanners_results' @@ -461,6 +465,12 @@ def rule_model(self): def webhook(self): return self.webhook_event.webhook if self.webhook_event else None + @property + def is_complete(self): + """Whether the scanner is done with this result: `results` is None + (event skipped) or has `matchedRules`.""" + return self.results is None or 'matchedRules' in self.results + def get_rules_queryset(self): # See: https://github.com/mozilla/addons-server/issues/13143 return super().get_rules_queryset().filter(is_active=True) diff --git a/src/olympia/scanners/tasks.py b/src/olympia/scanners/tasks.py index 7306d0adfb02..b5072b4bb0c3 100644 --- a/src/olympia/scanners/tasks.py +++ b/src/olympia/scanners/tasks.py @@ -5,6 +5,7 @@ import os import uuid from collections import defaultdict, namedtuple +from datetime import datetime, timedelta from django.conf import settings from django.db.models import F @@ -42,6 +43,7 @@ WEBHOOK, WEBHOOK_DURING_VALIDATION, WEBHOOK_EVENTS, + WEBHOOK_MAX_DELIVERY_ATTEMPTS, WEBHOOK_ON_VERSION_CREATED, YARA, ) @@ -49,6 +51,7 @@ from olympia.files.models import FileManifest, FileUpload from olympia.files.utils import ManifestJSONExtractor, SafeZip from olympia.versions.models import Version +from olympia.zadmin.models import get_config from .models import ( ImproperScannerQueryRuleStateError, @@ -105,8 +108,7 @@ def call_webhooks(event_id, payload, upload=None, version=None, activity_log=Non webhook__is_active=True, ).all(): log.info('Calling webhook "%s".', event.webhook.name) - event_name = WEBHOOK_EVENTS.get(event_id, event_id) - statsd_name = f'devhub.webhook.{slugify(event.webhook.name)}.{event_name}' + statsd_name = _get_webhook_statsd_name(event) try: scanner_result = ScannerResult.objects.create( @@ -117,25 +119,7 @@ def call_webhooks(event_id, payload, upload=None, version=None, activity_log=Non activity_log=activity_log, ) - with statsd.timer(statsd_name): - data = _call_webhook( - webhook=event.webhook, - payload={ - **payload, - 'event': event_name, - 'scanner_result_url': absolutify( - reverse( - 'v5:scanner-result-patch', - args=[scanner_result.pk], - ) - ), - }, - ) - - scanner_result.results = data - # We don't pass `update_fields` because the `save()` method - # also updates other fields (e.g. has_matches, matched_rules). - scanner_result.save() + _deliver_webhook(scanner_result, payload) statsd.incr(f'{statsd_name}.success') except Exception as exc: @@ -144,6 +128,11 @@ def call_webhooks(event_id, payload, upload=None, version=None, activity_log=Non raise exc +def _get_webhook_statsd_name(event): + event_name = WEBHOOK_EVENTS.get(event.event, event.event) + return f'devhub.webhook.{slugify(event.webhook.name)}.{event_name}' + + def build_webhook_payload(event_id, *, upload=None, version=None): """Return the payload for the given event. @@ -165,6 +154,184 @@ def build_webhook_payload(event_id, *, upload=None, version=None): raise ValueError(f'No payload for webhook event {event_id}') +def _deliver_webhook(scanner_result, payload): + """Call the webhook for an existing ScannerResult and store what it + returned. Exceptions are left to the caller.""" + event = scanner_result.webhook_event + + # Count the attempt before the call, so that failures count too. + previous_results = scanner_result.results + scanner_result.update( + delivery_attempts=scanner_result.delivery_attempts + 1, + modified=datetime.now(), + ) + + with statsd.timer(_get_webhook_statsd_name(event)): + data = _call_webhook( + webhook=event.webhook, + payload={ + **payload, + 'event': WEBHOOK_EVENTS.get(event.event, event.event), + 'scanner_result_url': absolutify( + reverse( + 'v5:scanner-result-patch', + args=[scanner_result.pk], + ) + ), + }, + ) + + scanner_result.reload() + if scanner_result.results != previous_results: + # The scanner sent its results while we were calling it. + return + + scanner_result.results = data + # We don't pass `update_fields` because the `save()` method + # also updates other fields (e.g. has_matches, matched_rules). + scanner_result.save() + + +def _flag_version_as_waiting_on_scanners(version): + """Flag the version for human review because we gave up on a scanner. + + Same conditions as the `auto_approve` command's disapprove(), which used to + do this. Return whether the flag was added.""" + from olympia.reviewers.models import NeedsHumanReview + + reason = NeedsHumanReview.REASONS.WAITING_ON_SCANNERS + if ( + version.pending_rejection + or version.contentdecision_set.awaiting_action().exists() + # Only active flags matter: a version can leave the queue and re-enter it. + or version.needshumanreview_set.filter(reason=reason, is_active=True).exists() + ): + return False + + version.needshumanreview_set.create(reason=reason) + return True + + +def _give_up_on_scanner(version, event, scanner_result=None): + """Stop expecting results from a scanner for this version.""" + if ( + scanner_result is not None + and scanner_result.delivery_attempts < WEBHOOK_MAX_DELIVERY_ATTEMPTS + ): + # Make sure we don't ask this scanner again. + scanner_result.update(delivery_attempts=WEBHOOK_MAX_DELIVERY_ATTEMPTS) + + if not _flag_version_as_waiting_on_scanners(version): + # We have given up already, no need to say it again on every run. + return + + statsd.incr(f'{_get_webhook_statsd_name(event)}.gave_up') + log.error( + 'Giving up on scanner "%s" for version %s.', + event.webhook.name, + version.pk, + ) + + +def _build_retry_payload(scanner_result): + """Return the payload to send again, or None if it can't be rebuilt.""" + event_id = scanner_result.webhook_event.event + + if event_id == WEBHOOK_DURING_VALIDATION: + upload = scanner_result.upload + if not upload or not os.path.exists(upload.file_path): + log.error( + 'Cannot retry the webhook for scanner result %s because its ' + 'file upload is gone.', + scanner_result.pk, + ) + return None + return build_webhook_payload(event_id, upload=upload) + + return build_webhook_payload(event_id, version=scanner_result.version) + + +@task +@use_primary_db +def retry_webhook_deliveries_on_version(version_pk): + """Call the webhooks again for the events this version is still waiting on. + + Scanners can fail to send their results back, e.g. because of an outage on our + side, so we ask them again with a backoff, up to WEBHOOK_MAX_DELIVERY_ATTEMPTS + times. + """ + version = Version.unfiltered.get(pk=version_pk) + initial_delay = get_config(amo.config_keys.SCANNER_WEBHOOK_RETRY_INITIAL_DELAY) + events = ScannerWebhookEvent.blocking_auto_approval_for(version) + results_by_event_id = { + result.webhook_event_id: result + for result in ScannerResult.objects.filter( + version=version, webhook_event__in=events + ) + } + + for event in events: + scanner_result = results_by_event_id.get(event.pk) + + if scanner_result is None: + # The scanner was never called for this event: the task errored + # out before creating the result, or the event was added after + # the file had been validated. The result being our only reference + # to what we should send, there is nothing to retry. + if datetime.now() >= version.created + timedelta(seconds=initial_delay): + _give_up_on_scanner(version, event) + continue + + if scanner_result.is_complete: + continue + + attempts = scanner_result.delivery_attempts + if attempts >= WEBHOOK_MAX_DELIVERY_ATTEMPTS: + # We have exhausted all our attempts, this version has to be looked + # at by reviewers. + _flag_version_as_waiting_on_scanners(version) + continue + + # Each attempt doubles the delay since the previous one: 1h, 2h, 4h + # then 8h by default, i.e. 15h in total before we give up. + due_date = scanner_result.modified + timedelta( + seconds=initial_delay * 2 ** max(attempts - 1, 0) + ) + if datetime.now() < due_date: + continue + + payload = _build_retry_payload(scanner_result) + if payload is None: + # We cannot rebuild what we should send, there is nothing we can do. + _give_up_on_scanner(version, event, scanner_result) + continue + + log.info( + 'Retrying scanner "%s" for version %s (attempt %s/%s).', + event.webhook.name, + version.pk, + attempts + 1, + WEBHOOK_MAX_DELIVERY_ATTEMPTS, + ) + statsd_name = _get_webhook_statsd_name(event) + try: + _deliver_webhook(scanner_result, payload) + statsd.incr(f'{statsd_name}.success') + except Exception: + statsd.incr(f'{statsd_name}.failure') + log.exception( + 'Error while retrying scanner "%s" for version %s.', + event.webhook.name, + version.pk, + ) + + if ( + scanner_result.delivery_attempts >= WEBHOOK_MAX_DELIVERY_ATTEMPTS + and not scanner_result.is_complete + ): + _give_up_on_scanner(version, event, scanner_result) + + def _call_webhook(webhook, payload): with requests.Session() as http: adapter = make_adapter_with_retry() diff --git a/src/olympia/scanners/tests/test_tasks.py b/src/olympia/scanners/tests/test_tasks.py index f1602a9529ed..ec6a29820c39 100644 --- a/src/olympia/scanners/tests/test_tasks.py +++ b/src/olympia/scanners/tests/test_tasks.py @@ -1,5 +1,6 @@ import json import os +from datetime import datetime, timedelta from unittest import mock from django.conf import settings @@ -17,6 +18,7 @@ block_factory, user_factory, version_factory, + version_review_flags_factory, ) from olympia.constants.promoted import NOTABLE_API_NAME from olympia.constants.scanners import ( @@ -29,11 +31,14 @@ SCHEDULED, WEBHOOK, WEBHOOK_DURING_VALIDATION, + WEBHOOK_MAX_DELIVERY_ATTEMPTS, + WEBHOOK_ON_VERSION_CREATED, YARA, ) from olympia.files.models import File from olympia.files.tests.test_models import UploadMixin from olympia.files.utils import parse_addon +from olympia.reviewers.models import NeedsHumanReview from olympia.scanners.models import ( ScannerQueryResult, ScannerQueryRule, @@ -48,6 +53,7 @@ call_webhooks, call_webhooks_during_validation, mark_scanner_query_rule_as_completed_or_aborted, + retry_webhook_deliveries_on_version, run_narc_on_version, run_scanner, run_scanner_query_rule, @@ -55,6 +61,7 @@ run_yara, ) from olympia.versions.models import Version +from olympia.zadmin.models import set_config class TestRunScanner(UploadMixin, TestCase): @@ -2559,6 +2566,7 @@ def test_call_webhooks(self, _call_webhook_mock): for result in results: assert result.scanner == WEBHOOK assert result.results == returned_data + assert result.delivery_attempts == 1 assert results[0].webhook_event == event_1 assert results[1].webhook_event == event_3 @@ -2908,3 +2916,343 @@ def test_call_webhooks_during_validation_without_file_path_ignore_exceptions(sel results = call_webhooks_during_validation(self.results, self.upload.pk) assert self.results == results + + +@mock.patch('olympia.scanners.tasks._call_webhook') +class TestRetryWebhookDeliveriesOnVersion(UploadMixin, TestCase): + def setUp(self): + super().setUp() + + user_factory(id=settings.TASK_USER_ID) + self.addon = addon_factory(file_kw={'status': amo.STATUS_AWAITING_REVIEW}) + self.version = self.addon.versions.get() + self.webhook = ScannerWebhook.objects.create( + name='some-scanner', + url='https://example.org/webhook', + api_key='some-api-key', + ) + self.webhook.update(modified=self.days_ago(42)) + self.event = ScannerWebhookEvent.objects.create( + event=WEBHOOK_ON_VERSION_CREATED, webhook=self.webhook + ) + + def create_result(self, **kwargs): + kwargs.setdefault('results', {'ok': True}) + kwargs.setdefault('delivery_attempts', 1) + return ScannerResult.objects.create( + scanner=WEBHOOK, + webhook_event=self.event, + version=self.version, + **kwargs, + ) + + def age_last_attempt(self, scanner_result, **kwargs): + """Pretend the last delivery attempt was made a while ago.""" + scanner_result.update(modified=datetime.now() - timedelta(**kwargs)) + + def age_version(self, **kwargs): + Version.objects.filter(pk=self.version.pk).update( + created=datetime.now() - timedelta(**kwargs) + ) + self.version = self.version.reload() + + def test_before_initial_delay(self, _call_webhook_mock): + result = self.create_result() + self.age_last_attempt(result, minutes=30) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert not _call_webhook_mock.called + assert result.reload().delivery_attempts == 1 + + def test_retries_after_initial_delay(self, _call_webhook_mock): + _call_webhook_mock.return_value = {'ack': True} + result = self.create_result() + self.age_last_attempt(result, hours=2) + + retry_webhook_deliveries_on_version(self.version.pk) + + _call_webhook_mock.assert_called_with( + webhook=self.webhook, + payload={ + 'addon': mock.ANY, + 'version': mock.ANY, + 'event': 'on_version_created', + 'scanner_result_url': ( + f'http://testserver/api/v5/scanner/results/{result.pk}/' + ), + }, + ) + result = result.reload() + assert result.delivery_attempts == 2 + assert result.results == {'ack': True} + + def test_backoff(self, _call_webhook_mock): + _call_webhook_mock.return_value = {'ack': True} + # Each attempt doubles the delay since the previous one. + for attempts, hours in ((1, 1), (2, 2), (3, 4), (4, 8)): + result = self.create_result(delivery_attempts=attempts) + + self.age_last_attempt(result, minutes=hours * 60 - 1) + retry_webhook_deliveries_on_version(self.version.pk) + assert not _call_webhook_mock.called, attempts + + self.age_last_attempt(result, hours=hours) + retry_webhook_deliveries_on_version(self.version.pk) + assert _call_webhook_mock.call_count == 1, attempts + assert result.reload().delivery_attempts == attempts + 1 + + _call_webhook_mock.reset_mock() + result.delete() + + def test_records_each_attempt(self, _call_webhook_mock): + _call_webhook_mock.return_value = {'ack': True} + result = self.create_result(delivery_attempts=0, results={}) + + # No more than WEBHOOK_MAX_DELIVERY_ATTEMPTS deliveries, however + # many times the task runs. + for _ in range(WEBHOOK_MAX_DELIVERY_ATTEMPTS * 2): + self.age_last_attempt(result, days=1) + retry_webhook_deliveries_on_version(self.version.pk) + + assert _call_webhook_mock.call_count == WEBHOOK_MAX_DELIVERY_ATTEMPTS + assert result.reload().delivery_attempts == WEBHOOK_MAX_DELIVERY_ATTEMPTS + + def test_honors_initial_delay_config(self, _call_webhook_mock): + _call_webhook_mock.return_value = {'ack': True} + set_config(amo.config_keys.SCANNER_WEBHOOK_RETRY_INITIAL_DELAY, 60) + result = self.create_result() + self.age_last_attempt(result, minutes=30) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert _call_webhook_mock.called + + def test_no_more_attempt_left(self, _call_webhook_mock): + result = self.create_result(delivery_attempts=WEBHOOK_MAX_DELIVERY_ATTEMPTS) + self.age_last_attempt(result, days=1) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert not _call_webhook_mock.called + assert result.reload().delivery_attempts == WEBHOOK_MAX_DELIVERY_ATTEMPTS + + @mock.patch('olympia.scanners.tasks.statsd.incr') + def test_last_attempt_is_exhausted(self, statsd_incr_mock, _call_webhook_mock): + _call_webhook_mock.return_value = {'ack': True} + result = self.create_result(delivery_attempts=WEBHOOK_MAX_DELIVERY_ATTEMPTS - 1) + self.age_last_attempt(result, days=1) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert _call_webhook_mock.called + statsd_incr_mock.assert_has_calls( + [ + mock.call('devhub.webhook.some-scanner.on_version_created.success'), + mock.call('devhub.webhook.some-scanner.on_version_created.gave_up'), + ] + ) + nhr = self.version.needshumanreview_set.get() + assert nhr.reason == NeedsHumanReview.REASONS.WAITING_ON_SCANNERS + assert nhr.is_active + + def test_no_needs_human_review_before_the_last_attempt(self, _call_webhook_mock): + _call_webhook_mock.return_value = {'ack': True} + result = self.create_result() + self.age_last_attempt(result, hours=2) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert _call_webhook_mock.called + assert not self.version.needshumanreview_set.exists() + + @mock.patch('olympia.scanners.tasks.statsd.incr') + def test_flags_version_again_after_flag_was_cleared( + self, statsd_incr_mock, _call_webhook_mock + ): + _call_webhook_mock.return_value = {'ack': True} + result = self.create_result(delivery_attempts=WEBHOOK_MAX_DELIVERY_ATTEMPTS - 1) + self.age_last_attempt(result, days=1) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert self.version.needshumanreview_set.count() == 1 + # The version should not be flagged twice while it is in the queue... + retry_webhook_deliveries_on_version(self.version.pk) + assert self.version.needshumanreview_set.count() == 1 + + # ...but it should be flagged again once a reviewer has cleared it, + # since we are still waiting on the scanner. + self.version.needshumanreview_set.update(is_active=False) + statsd_incr_mock.reset_mock() + + retry_webhook_deliveries_on_version(self.version.pk) + + assert self.version.needshumanreview_set.filter(is_active=True).count() == 1 + # We only give up (and say so) once. + assert _call_webhook_mock.call_count == 1 + assert not statsd_incr_mock.called + + def test_does_not_flag_version_pending_rejection(self, _call_webhook_mock): + _call_webhook_mock.return_value = {'ack': True} + version_review_flags_factory( + version=self.version, + pending_rejection=datetime.now() + timedelta(days=1), + pending_rejection_by=user_factory(), + pending_content_rejection=False, + ) + result = self.create_result(delivery_attempts=WEBHOOK_MAX_DELIVERY_ATTEMPTS - 1) + self.age_last_attempt(result, days=1) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert not self.version.needshumanreview_set.exists() + + @mock.patch('olympia.scanners.tasks.statsd.incr') + def test_delivery_failure_is_counted(self, statsd_incr_mock, _call_webhook_mock): + _call_webhook_mock.side_effect = ValueError('oops') + result = self.create_result() + self.age_last_attempt(result, hours=2) + + retry_webhook_deliveries_on_version(self.version.pk) + + result = result.reload() + assert result.delivery_attempts == 2 + # The next attempt should not be due immediately. + assert result.modified > datetime.now() - timedelta(minutes=1) + statsd_incr_mock.assert_called_with( + 'devhub.webhook.some-scanner.on_version_created.failure' + ) + + @mock.patch('olympia.scanners.tasks.statsd.timer') + def test_calls_statsd_timer(self, timer_mock, _call_webhook_mock): + # A retry is timed under the same name as the initial delivery, so that + # `TestCallWebhooks.test_statsd_success` and this test agree. + _call_webhook_mock.return_value = {'ack': True} + result = self.create_result() + self.age_last_attempt(result, hours=2) + + retry_webhook_deliveries_on_version(self.version.pk) + + timer_mock.assert_called_once_with( + 'devhub.webhook.some-scanner.on_version_created' + ) + + def test_does_not_overwrite_results_sent_while_calling(self, _call_webhook_mock): + result = self.create_result() + self.age_last_attempt(result, hours=2) + + def send_results(*args, **kwargs): + # The scanner patches the result while we are calling it. + ScannerResult.objects.get(pk=result.pk).update( + results={'matchedRules': ['some-rule']} + ) + return {'ack': True} + + _call_webhook_mock.side_effect = send_results + + retry_webhook_deliveries_on_version(self.version.pk) + + assert result.reload().results == {'matchedRules': ['some-rule']} + + def test_skips_completed_results(self, _call_webhook_mock): + for results in (None, {'matchedRules': []}, {'matchedRules': ['some-rule']}): + result = self.create_result(results=results) + self.age_last_attempt(result, days=1) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert not _call_webhook_mock.called, results + result.delete() + + def test_skips_inactive_webhook(self, _call_webhook_mock): + self.webhook.update(is_active=False) + result = self.create_result() + self.age_last_attempt(result, days=1) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert not _call_webhook_mock.called + + def test_gives_up_on_missing_result(self, _call_webhook_mock): + # The result is our only reference to what we should send, so there is + # nothing to retry when it is missing, whatever the event. + for event_id in (WEBHOOK_ON_VERSION_CREATED, WEBHOOK_DURING_VALIDATION): + self.event.update(event=event_id) + self.age_version(hours=2) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert not _call_webhook_mock.called, event_id + assert not ScannerResult.objects.exists(), event_id + nhr = self.version.needshumanreview_set.get() + assert nhr.reason == NeedsHumanReview.REASONS.WAITING_ON_SCANNERS + nhr.delete() + + def test_does_not_give_up_on_missing_result_too_early(self, _call_webhook_mock): + self.age_version(minutes=30) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert not _call_webhook_mock.called + assert not ScannerResult.objects.exists() + assert not self.version.needshumanreview_set.exists() + + @mock.patch('olympia.scanners.tasks.statsd.incr') + def test_only_gives_up_on_missing_result_once( + self, statsd_incr_mock, _call_webhook_mock + ): + self.age_version(hours=2) + + retry_webhook_deliveries_on_version(self.version.pk) + retry_webhook_deliveries_on_version(self.version.pk) + + assert self.version.needshumanreview_set.count() == 1 + statsd_incr_mock.assert_called_once_with( + 'devhub.webhook.some-scanner.on_version_created.gave_up' + ) + + def test_retries_during_validation(self, _call_webhook_mock): + _call_webhook_mock.return_value = {'ack': True} + self.event.update(event=WEBHOOK_DURING_VALIDATION) + upload = self.get_upload('webextension.xpi') + result = self.create_result(upload=upload) + self.age_last_attempt(result, hours=2) + + retry_webhook_deliveries_on_version(self.version.pk) + + _call_webhook_mock.assert_called_with( + webhook=self.webhook, + payload={ + 'download_url': upload.get_authenticated_download_url(), + 'event': 'during_validation', + 'scanner_result_url': ( + f'http://testserver/api/v5/scanner/results/{result.pk}/' + ), + }, + ) + + def test_flags_version_when_delivery_cannot_be_retried(self, _call_webhook_mock): + # There is nothing left to try when we cannot rebuild the payload. + self.event.update(event=WEBHOOK_DURING_VALIDATION) + result = self.create_result() + self.age_last_attempt(result, hours=2) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert not _call_webhook_mock.called + assert result.reload().delivery_attempts == WEBHOOK_MAX_DELIVERY_ATTEMPTS + nhr = self.version.needshumanreview_set.get() + assert nhr.reason == NeedsHumanReview.REASONS.WAITING_ON_SCANNERS + + def test_does_not_retry_during_validation_without_file(self, _call_webhook_mock): + self.event.update(event=WEBHOOK_DURING_VALIDATION) + upload = self.get_upload('webextension.xpi') + upload.update(path='/not-a-file') + result = self.create_result(upload=upload) + self.age_last_attempt(result, hours=2) + + retry_webhook_deliveries_on_version(self.version.pk) + + assert not _call_webhook_mock.called + assert result.reload().delivery_attempts == WEBHOOK_MAX_DELIVERY_ATTEMPTS