_state.py: Use separate task identifier

`State.add_task()` required the job name to be unique in the session.
However, the tuple `(action_name, full_name)` is not guaranteed to be
unique. E.g., multiple `ArtifactElement` objects with the same element
name may participate in a single session. Use a separate task identifier
to fix this.

2 files changed, 14 insertions, 5 deletions

diff --git a/src/buildstream/_scheduler/jobs/job.py b/src/buildstream/_scheduler/jobs/job.py
index fd4f7720d..c8ff853ed 100644
--- a/src/buildstream/_scheduler/jobs/job.py
+++ b/src/buildstream/_scheduler/jobs/job.py
@@ -23,6 +23,7 @@
 # System imports
 import asyncio
 import datetime
+import itertools
 import multiprocessing
 import os
 import signal
@@ -109,11 +110,17 @@ class _MessageType(FastEnum):
 #    max_retries (int): The maximum number of retries
 #
 class Job:
+    # Unique id generator for jobs
+    #
+    # This is used to identify tasks in the `State` class
+    _id_generator = itertools.count(1)
+
     def __init__(self, scheduler, action_name, logfile, *, max_retries=0):
 
         #
         # Public members
         #
+        self.id = "{}-{}".format(action_name, next(Job._id_generator))
         self.name = None  # The name of the job, set by the job's subclass
         self.action_name = action_name  # The action name for the Queue
         self.child_data = None  # Data to be sent to the main process
diff --git a/src/buildstream/_scheduler/scheduler.py b/src/buildstream/_scheduler/scheduler.py
index 3e6bf1f92..5f0d69a06 100644
--- a/src/buildstream/_scheduler/scheduler.py
+++ b/src/buildstream/_scheduler/scheduler.py
@@ -261,9 +261,9 @@ class Scheduler:
             else:
                 element_info = None
 
-            self._state.fail_task(job.action_name, job.name, element_info)
+            self._state.fail_task(job.id, element_info)
 
-        self._state.remove_task(job.action_name, job.name)
+        self._state.remove_task(job.id)
 
         self._sched()
 
@@ -306,7 +306,7 @@ class Scheduler:
         self._active_jobs.append(job)
         job.start()
 
-        self._state.add_task(job.action_name, job.name, self._state.elapsed_time())
+        self._state.add_task(job.id, job.action_name, job.name, self._state.elapsed_time())
 
     # _sched_queue_jobs()
     #
@@ -497,10 +497,12 @@ class Scheduler:
         self._ticker_callback()
         self.loop.call_later(1, self._tick)
 
-    def _failure_retry(self, action_name, unique_id):
+    def _failure_retry(self, task_id, unique_id):
+        task = self._state.tasks[task_id]
+
         queue = None
         for q in self.queues:
-            if q.action_name == action_name:
+            if q.action_name == task.action_name:
                 queue = q
                 break
         # Assert queue found, we should only be retrying a queued job