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,6 +62,8 @@ 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()
 
         # 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
@@ -58,28 +72,33 @@ def create_tasks(graph_config, taskgraph, label_to_taskid, params, decision_task
         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
-
             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)
 
             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 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
 
+                # 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
+
                 submit(task_id, taskid_to_label[task_id], task_def)
                 to_remove.add(task_id)
 
@@ -90,16 +109,24 @@ 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
+        # Check the result of each future and save the exception for later
+        # printing if it failed.
+        errors = {}
         for f in futures.as_completed(fs.values()):
-            f.result()
+            if exc := f.exception():
+                task_id, label = fs_to_task[f]
+                skipped.add(task_id)
+                errors[label] = exc
+
+        if errors:
+            raise CreateTasksException(errors)
 
 
 def create_task(session, task_id, label, task_def):