Allow notifications to be retried when an error occurs (#1026)

src/api-service/__app__/onefuzzlib/notifications/ado.py

@@ -31,6 +31,10 @@ from ..secrets import get_secret_string_value
 from .common import Render, fail_task
 
 
+class AdoNotificationException(Exception):
+    pass
+
+
 @cached(ttl=60)
 def get_ado_client(base_url: str, token: str) -> WorkItemTrackingClient:
     connection = Connection(base_url=base_url, creds=BasicAuthentication("PAT", token))
@@ -206,11 +210,26 @@ class ADO:
             self.create_new()
 
 
+def is_transient(err: Exception) -> bool:
+    error_codes = [
+        # "TF401349: An unexpected error has occurred, please verify your request and try again." # noqa: E501
+        "TF401349",
+        # TF26071: This work item has been changed by someone else since you opened it. You will need to refresh it and discard your changes. # noqa: E501
+        "TF26071",
+    ]
+    error_str = str(err)
+    for code in error_codes:
+        if code in error_str:
+            return True
+    return False
+
+
 def notify_ado(
     config: ADOTemplate,
     container: Container,
     filename: str,
     report: Union[Report, RegressionReport],
+    fail_task_on_transient_error: bool,
 ) -> None:
     if isinstance(report, RegressionReport):
         logging.info(
@@ -232,13 +251,17 @@ def notify_ado(
     try:
         ado = ADO(container, filename, config, report)
         ado.process()
-    except AzureDevOpsAuthenticationError as err:
+    except (
-        fail_task(report, err)
+        AzureDevOpsAuthenticationError,
-    except AzureDevOpsClientError as err:
+        AzureDevOpsClientError,
-        fail_task(report, err)
+        AzureDevOpsServiceError,
-    except AzureDevOpsServiceError as err:
+        AzureDevOpsClientRequestError,
-        fail_task(report, err)
+        ValueError,
-    except AzureDevOpsClientRequestError as err:
+    ) as err:
-        fail_task(report, err)
+
-    except ValueError as err:
+        if not fail_task_on_transient_error and is_transient(err):
-        fail_task(report, err)
+            raise AdoNotificationException(
+                "transient ADO notification failure"
+            ) from err
+        else:
+            fail_task(report, err)

src/api-service/__app__/onefuzzlib/notifications/github_issues.py

@@ -128,7 +128,5 @@ def github_issue(
     try:
         handler = GithubIssue(config, container, filename, report)
         handler.process()
-    except GitHubException as err:
+    except (GitHubException, ValueError) as err:
-        fail_task(report, err)
-    except ValueError as err:
         fail_task(report, err)

src/api-service/__app__/onefuzzlib/notifications/main.py

@@ -127,7 +127,9 @@ def get_queue_tasks() -> Sequence[Tuple[Task, Sequence[str]]]:
     return results
 
 
-def new_files(container: Container, filename: str) -> None:
+def new_files(
+    container: Container, filename: str, fail_task_on_transient_error: bool
+) -> None:
     notifications = get_notifications(container)
 
     report = get_report_or_regression(
@@ -149,7 +151,13 @@ def new_files(container: Container, filename: str) -> None:
                 continue
 
             if isinstance(notification.config, ADOTemplate):
-                notify_ado(notification.config, container, filename, report)
+                notify_ado(
+                    notification.config,
+                    container,
+                    filename,
+                    report,
+                    fail_task_on_transient_error,
+                )
 
             if isinstance(notification.config, GithubIssueTemplate):
                 github_issue(notification.config, container, filename, report)

src/api-service/__app__/queue_file_changes/__init__.py

@@ -13,25 +13,29 @@ from ..onefuzzlib.azure.storage import corpus_accounts
 from ..onefuzzlib.events import get_events
 from ..onefuzzlib.notifications.main import new_files
 
+# The number of time the function will be retried if an error occurs
+# https://docs.microsoft.com/en-us/azure/azure-functions/functions-bindings-storage-queue-trigger?tabs=csharp#poison-messages
+MAX_DEQUEUE_COUNT = 5
 
-def file_added(event: Dict) -> None:
+
+def file_added(event: Dict, fail_task_on_transient_error: bool) -> None:
     parts = event["data"]["url"].split("/")[3:]
     container = parts[0]
     path = "/".join(parts[1:])
     logging.info("file added container: %s - path: %s", container, path)
-    new_files(container, path)
+    new_files(container, path, fail_task_on_transient_error)
 
 
 def main(msg: func.QueueMessage, dashboard: func.Out[str]) -> None:
     event = json.loads(msg.get_body())
-
+    last_try = msg.dequeue_count == MAX_DEQUEUE_COUNT
     if event["topic"] not in corpus_accounts():
         return
 
     if event["eventType"] != "Microsoft.Storage.BlobCreated":
         return
 
-    file_added(event)
+    file_added(event, last_try)
 
     events = get_events()
     if events: