Correctly track session elements in dynamic build plan

src/buildstream/_scheduler/queues/queue.py

@@ -62,6 +62,7 @@ class Queue:
     # Resources this queues' jobs want
     resources = []  # type: List[int]
     log_to_file = True
+    session_elements = None
 
     def __init__(self, scheduler, *, imperative=False):
 
@@ -256,6 +257,17 @@ def harvest_jobs(self):
     def set_required_element_check(self):
         self._required_element_check = True
 
+    # set_session_elements()
+    #
+    # This passes a reference to a set used to keep track of the
+    # elements enqued. This is used in the first queue to determine
+    # which elements are actually considered in the current session.
+    #
+    # Args:
+    #    session_elements (set): a set to put session elements
+    def set_session_elements(self, session_elements):
+        self.session_elements = session_elements
+
     # any_failed_elements()
     #
     # Returns whether any elements in this queue have failed their jobs
@@ -381,6 +393,9 @@ def _element_log_path(self, element):
     #    element (Element): The Element to enqueue
     #
     def _enqueue_element(self, element):
+        if self.session_elements is not None:
+            self.session_elements.add(element)
+
         status = self.status(element)
 
         if status == QueueStatus.SKIP:

src/buildstream/_stream.py

@@ -72,7 +72,7 @@ def __init__(
         # Public members
         #
         self.targets = []  # Resolved target elements
-        self.session_elements = []  # List of elements being processed this session
+        self.session_elements = set()  # Set of elements being processed this session
         self.total_elements = []  # Total list of elements based on targets
         self.queues = []  # Queue objects
 
@@ -1802,7 +1802,7 @@ def _load(
     #
     def _reset(self):
         self._scheduler.clear_queues()
-        self.session_elements = []
+        self.session_elements = set()
         self.total_elements = []
 
     # _add_queue()
@@ -1817,6 +1817,9 @@ def _add_queue(self, queue, *, track=False):
             # First non-track queue
             queue.set_required_element_check()
 
+        if not self.queues:
+            queue.set_session_elements(self.session_elements)
+
         self.queues.append(queue)
 
     # _enqueue_plan()
@@ -1830,7 +1833,6 @@ def _add_queue(self, queue, *, track=False):
     def _enqueue_plan(self, plan, *, queue=None):
         queue = queue or self.queues[0]
         queue.enqueue(plan)
-        self.session_elements += plan
 
     # _run()
     #