fix: improve error handling on task creation failure

This prints the label of the task we failed to create for convenience,
as well as defers failing the Decision task until after we've attempted
all tasks. This way we'll see all the scope errors at once rather than
needing to fix them one at a time.

Author: ahal

src/taskgraph/create.py

@@ -20,6 +20,18 @@
 testing = False
 
 
+class CreateTasksException(Exception):
+    """Exception raised when one or more tasks could not be created."""
+
+    def __init__(self, errors: dict[str, Exception]):
+        message = ""
+        for label, exc in errors.items():
+            message += f"\nERROR: Could not create '{label}':\n\n"
+            message += "\n".join(f"    {line}" for line in str(exc).splitlines()) + "\n"
+
+        super().__init__(message)
+
+
 def create_tasks(graph_config, taskgraph, label_to_taskid, params, decision_task_id):
     taskid_to_label = {t: l for l, t in label_to_taskid.items()}
 
@@ -50,33 +62,48 @@ def create_tasks(graph_config, taskgraph, label_to_taskid, params, decision_task
     session = get_session()
     with futures.ThreadPoolExecutor(concurrency) as e:
         fs = {}
+        fs_to_task = {}
+        skipped = set()
+        errors = {}
 
         # We can't submit a task until its dependencies have been submitted.
         # So our strategy is to walk the graph and submit tasks once all
         # their dependencies have been submitted.
         tasklist = set(taskgraph.graph.visit_postorder())
         alltasks = tasklist.copy()
 
-        def schedule_tasks():
-            # bail out early if any futures have failed
-            if any(f.done() and f.exception() for f in fs.values()):
-                return
+        def handle_exception(fut):
+            if exc := fut.exception():
+                task_id, label = fs_to_task[fut]
+                skipped.add(task_id)
+                errors[label] = exc
 
+        def schedule_tasks():
             to_remove = set()
             new = set()
 
             def submit(task_id, label, task_def):
                 fut = e.submit(create_task, session, task_id, label, task_def)
                 new.add(fut)
                 fs[task_id] = fut
+                fs_to_task[fut] = (task_id, label)
+                fut.add_done_callback(handle_exception)
 
             for task_id in tasklist:
                 task_def = taskgraph.tasks[task_id].task
-                # If we haven't finished submitting all our dependencies yet,
-                # come back to this later.
                 # Some dependencies aren't in our graph, so make sure to filter
                 # those out
                 deps = set(task_def.get("dependencies", [])) & alltasks
+
+                # If one of the dependencies didn't get created, then
+                # don't attempt to submit as it would fail.
+                if any(d in skipped for d in deps):
+                    skipped.add(task_id)
+                    to_remove.add(task_id)
+                    continue
+
+                # If we haven't finished submitting all our dependencies yet,
+                # come back to this later.
                 if any((d not in fs or not fs[d].done()) for d in deps):
                     continue
 
@@ -90,16 +117,18 @@ def submit(task_id, label, task_def):
                     submit(slugid(), taskid_to_label[task_id], task_def)
             tasklist.difference_update(to_remove)
 
-            # as each of those futures complete, try to schedule more tasks
+            # As each of those futures complete, try to schedule more tasks.
             for f in futures.as_completed(new):
                 schedule_tasks()
 
-        # start scheduling tasks and run until everything is scheduled
+        # Start scheduling tasks and run until everything is scheduled.
         schedule_tasks()
 
-        # check the result of each future, raising an exception if it failed
-        for f in futures.as_completed(fs.values()):
-            f.result()
+        # Wait for all futures to complete.
+        futures.wait(fs.values())
+
+        if errors:
+            raise CreateTasksException(errors)
 
 
 def create_task(session, task_id, label, task_def):

test/test_create.py

@@ -8,10 +8,10 @@
 from unittest import mock
 
 import responses
-from taskcluster.exceptions import TaskclusterRestFailure
 
 from taskgraph import create
 from taskgraph.config import GraphConfig
+from taskgraph.create import CreateTasksException
 from taskgraph.graph import Graph
 from taskgraph.task import Task
 from taskgraph.taskgraph import TaskGraph
@@ -151,11 +151,54 @@ def test_create_tasks_fails_if_create_fails(self):
         graph = Graph(nodes={"tid-a"}, edges=set())
         taskgraph = TaskGraph(tasks, graph)
 
-        with self.assertRaises(TaskclusterRestFailure):
+        with self.assertRaises(CreateTasksException):
             create.create_tasks(
                 GRAPH_CONFIG,
                 taskgraph,
                 label_to_taskid,
                 {"level": "4"},
                 decision_task_id="decisiontask",
             )
+
+    @responses.activate
+    @mock.patch.dict(
+        "os.environ",
+        {"TASKCLUSTER_ROOT_URL": "https://tc.example.com"},
+        clear=True,
+    )
+    def test_create_tasks_collects_multiple_errors(self):
+        "create_tasks collects all errors from multiple failing tasks"
+        mock_taskcluster_api(
+            error_status=409,
+            error_message={
+                "tid-a": "scope error for task a",
+                "tid-b": "scope error for task b",
+            },
+            error_task_ids={"tid-a", "tid-b"},
+        )
+
+        tasks = {
+            "tid-a": Task(
+                kind="test", label="a", attributes={}, task={"payload": "hello world"}
+            ),
+            "tid-b": Task(
+                kind="test", label="b", attributes={}, task={"payload": "hello world"}
+            ),
+        }
+        label_to_taskid = {"a": "tid-a", "b": "tid-b"}
+        graph = Graph(nodes={"tid-a", "tid-b"}, edges=set())
+        taskgraph = TaskGraph(tasks, graph)
+
+        with self.assertRaises(CreateTasksException) as cm:
+            create.create_tasks(
+                GRAPH_CONFIG,
+                taskgraph,
+                label_to_taskid,
+                {"level": "4"},
+                decision_task_id="decisiontask",
+            )
+
+        # Verify both errors are in the exception message
+        exception_message = str(cm.exception)
+        self.assertIn("Could not create 'a'", exception_message)
+        self.assertIn("Could not create 'b'", exception_message)