From f30276252f4b23e7fe5f5fcc96aa06eb5b795256 Mon Sep 17 00:00:00 2001
From: Nikita Ostrenkov <ownik.dev@gmail.com>
Date: Fri, 23 Oct 2026 09:40:55 +0300
Subject: [PATCH] scheduler: deduplicate build and pull of elements with same
 cache key

When one element is included through several junctions, the pipeline
has several instances of it with the same strong cache key. With
multiple builders they are built or pulled in parallel and write the
same artifact concurrently, which can fail the build.

Build and pull queues now process one instance per cache key at a
time. Other instances wait, then link the cached artifact under their
own name, or take over if the first job failed.
---
 src/buildstream/_artifactcache.py             |  22 ++++
 .../_scheduler/queues/buildqueue.py           |   2 +-
 .../_scheduler/queues/pullqueue.py            |   6 +-
 src/buildstream/_scheduler/queues/queue.py    |  85 ++++++++++++++-
 src/buildstream/element.py                    |  17 +++
 tests/scheduler/__init__.py                   |   0
 tests/scheduler/deduplication.py              | 100 ++++++++++++++++++
 .../deduplication/failure/elements/a.bst      |   1 +
 .../deduplication/failure/elements/b.bst      |   1 +
 .../deduplication/failure/plugins/fail.py     |  28 +++++
 .../deduplication/failure/project.conf        |  12 +++
 .../deduplication/multi-stage/elements/a.bst  |   4 +
 .../deduplication/multi-stage/elements/b.bst  |   4 +
 .../multi-stage/elements/subproject-a.bst     |   5 +
 .../multi-stage/elements/subproject-b.bst     |   5 +
 .../deduplication/multi-stage/project.conf    |   8 ++
 .../multi-stage/subproject/elements/hello.bst |   4 +
 .../multi-stage/subproject/files/hello.txt    |   1 +
 .../multi-stage/subproject/project.conf       |   3 +
 .../deduplication/simple/elements/a.bst       |   4 +
 .../deduplication/simple/elements/b.bst       |   4 +
 .../deduplication/simple/files/hello.txt      |   1 +
 .../deduplication/simple/project.conf         |   6 ++
 23 files changed, 318 insertions(+), 5 deletions(-)
 create mode 100644 tests/scheduler/__init__.py
 create mode 100644 tests/scheduler/deduplication.py
 create mode 100644 tests/scheduler/deduplication/failure/elements/a.bst
 create mode 100644 tests/scheduler/deduplication/failure/elements/b.bst
 create mode 100644 tests/scheduler/deduplication/failure/plugins/fail.py
 create mode 100644 tests/scheduler/deduplication/failure/project.conf
 create mode 100644 tests/scheduler/deduplication/multi-stage/elements/a.bst
 create mode 100644 tests/scheduler/deduplication/multi-stage/elements/b.bst
 create mode 100644 tests/scheduler/deduplication/multi-stage/elements/subproject-a.bst
 create mode 100644 tests/scheduler/deduplication/multi-stage/elements/subproject-b.bst
 create mode 100644 tests/scheduler/deduplication/multi-stage/project.conf
 create mode 100644 tests/scheduler/deduplication/multi-stage/subproject/elements/hello.bst
 create mode 100644 tests/scheduler/deduplication/multi-stage/subproject/files/hello.txt
 create mode 100644 tests/scheduler/deduplication/multi-stage/subproject/project.conf
 create mode 100644 tests/scheduler/deduplication/simple/elements/a.bst
 create mode 100644 tests/scheduler/deduplication/simple/elements/b.bst
 create mode 100644 tests/scheduler/deduplication/simple/files/hello.txt
 create mode 100644 tests/scheduler/deduplication/simple/project.conf

diff --git a/src/buildstream/_artifactcache.py b/src/buildstream/_artifactcache.py
index c8328f109..d1e4b392a 100644
--- a/src/buildstream/_artifactcache.py
+++ b/src/buildstream/_artifactcache.py
@@ -238,6 +238,28 @@ def link_key(self, element, oldkey, newkey):
 
         utils.safe_link(os.path.join(self._basedir, oldref), os.path.join(self._basedir, newref))
 
+    # link_element_key():
+    #
+    # Add a reference to an existing artifact for another element
+    # with the same cache key.
+    #
+    # Args:
+    #     element (Element): The Element whose artifact is to be linked
+    #     other (Element): The Element to add the reference for
+    #     key (str): The cache key of the artifact
+    #
+    def link_element_key(self, element, other, key):
+        ref = element.get_artifact_name(key)
+        otherref = other.get_artifact_name(key)
+
+        if ref == otherref:
+            # The two refs are identical, nothing to do
+            return
+
+        dest = os.path.join(self._basedir, otherref)
+        os.makedirs(os.path.dirname(dest), exist_ok=True)
+        utils.safe_link(os.path.join(self._basedir, ref), dest)
+
     # check_remotes_for_element()
     #
     # Check if the element is available in any of the remotes
diff --git a/src/buildstream/_scheduler/queues/buildqueue.py b/src/buildstream/_scheduler/queues/buildqueue.py
index 4b3fadab6..9d499d4f6 100644
--- a/src/buildstream/_scheduler/queues/buildqueue.py
+++ b/src/buildstream/_scheduler/queues/buildqueue.py
@@ -38,7 +38,7 @@ def status(self, element):
         if not element._buildable():
             return QueueStatus.PENDING
 
-        return QueueStatus.READY
+        return self._deduplicate(element)
 
     def done(self, job, element, result, status):
 
diff --git a/src/buildstream/_scheduler/queues/pullqueue.py b/src/buildstream/_scheduler/queues/pullqueue.py
index 1bb3ed55c..4e3dc6e3e 100644
--- a/src/buildstream/_scheduler/queues/pullqueue.py
+++ b/src/buildstream/_scheduler/queues/pullqueue.py
@@ -34,11 +34,11 @@ def get_process_func(self):
         return PullQueue._pull_or_skip
 
     def status(self, element):
-        if element._pull_pending():
-            return QueueStatus.READY
-        else:
+        if not element._pull_pending():
             return QueueStatus.SKIP
 
+        return self._deduplicate(element)
+
     def done(self, _, element, result, status):
 
         if status is JobStatus.FAIL:
diff --git a/src/buildstream/_scheduler/queues/queue.py b/src/buildstream/_scheduler/queues/queue.py
index 108786955..30fc64f82 100644
--- a/src/buildstream/_scheduler/queues/queue.py
+++ b/src/buildstream/_scheduler/queues/queue.py
@@ -32,7 +32,7 @@
 from ...types import FastEnum
 
 if TYPE_CHECKING:
-    from typing import List, Optional
+    from typing import Dict, List, Optional
 
 
 # Queue status for a given element
@@ -48,6 +48,21 @@ class QueueStatus(FastEnum):
     # The element is ready for processing in this queue.
     READY = 3
 
+    # Another element with the same strong cache key is being
+    # processed in this queue, see Queue._deduplicate().
+    WAIT = 4
+
+
+# _KeyGroup()
+#
+# Element instances sharing a strong cache key in a Queue
+#
+class _KeyGroup:
+    def __init__(self):
+        self.owner = None  # The element currently being processed
+        self.cached_element = None  # An element which successfully cached the artifact
+        self.waiters = []  # Elements waiting for the owner to complete
+
 
 # Queue()
 #
@@ -75,6 +90,7 @@ def __init__(self, scheduler, *, imperative=False):
         self._done_queue = deque()  # Processed / Skipped elements
         self._max_retries = 0
         self._queued_elements = 0  # Number of elements queued
+        self._key_groups = {}  # type: Dict[str, _KeyGroup]
 
         self._required_element_check = False  # Whether we should check that elements are required before enqueuing
 
@@ -366,6 +382,70 @@ def _job_done(self, job, element, status, result):
                 self._task_group.add_processed_task()
             else:
                 self._task_group.add_failed_task(element._get_full_name())
+        finally:
+            self._release_key_group(element, status == JobStatus.OK)
+
+    # _deduplicate()
+    #
+    # Ensures that only one of the Element instances sharing a strong
+    # cache key is processed at a time, e.g. the same element loaded
+    # through multiple junctions. Other instances wait for it, then
+    # either adopt its artifact or take over if it failed.
+    #
+    # This is called by Queue implementations from status() for
+    # elements which are otherwise ready to be processed.
+    #
+    # Args:
+    #    element (Element): The element which is ready
+    #
+    # Returns:
+    #    (QueueStatus): The status of the element
+    #
+    def _deduplicate(self, element):
+        key = element._get_cache_key()
+        if key is None:
+            return QueueStatus.READY
+
+        group = self._key_groups.setdefault(key, _KeyGroup())
+        if group.cached_element is not None:
+            element._adopt_artifact(group.cached_element)
+            return QueueStatus.SKIP
+        elif group.owner is not None and group.owner is not element:
+            return QueueStatus.WAIT
+
+        group.owner = element
+        return QueueStatus.READY
+
+    # _release_key_group()
+    #
+    # Re-enqueues elements waiting for the given element, see _deduplicate().
+    #
+    # Args:
+    #    element (Element): The element which completed
+    #    success (bool): Whether the job completed successfully
+    #
+    def _release_key_group(self, element, success):
+        group = self._key_groups.get(element._get_cache_key())
+        if group is None:
+            return
+
+        try:
+            if success and group.cached_element is None and element._cached_success():
+                group.cached_element = element
+
+            if group.owner is element:
+                group.owner = None
+                waiters, group.waiters = group.waiters, []
+                for waiter in waiters:
+                    self._enqueue_element(waiter)
+        except Exception:  # pylint: disable=broad-except
+            # Report unhandled exceptions, the job completion must not be interrupted
+            self._message(
+                element,
+                MessageType.BUG,
+                "Unhandled exception while releasing waiting elements",
+                detail=traceback.format_exc(),
+            )
 
     # Convenience wrapper for Queue implementations to send
     # a message for the element they are processing
@@ -406,6 +486,9 @@ def _enqueue_element(self, element):
             # Push elements which are ready to be processed immediately into the queue
             heapq.heappush(self._ready_queue, (element._depth, self._queued_elements, element))
             self._queued_elements += 1
+        elif status == QueueStatus.WAIT:
+            # Element is re-enqueued by _release_key_group()
+            self._key_groups[element._get_cache_key()].waiters.append(element)
         else:
             # Register a queue specific callback for pending elements
             self.register_pending_element(element)
diff --git a/src/buildstream/element.py b/src/buildstream/element.py
index 626bd386d..e0b0723a4 100644
--- a/src/buildstream/element.py
+++ b/src/buildstream/element.py
@@ -1661,6 +1661,23 @@ def _assemble_done(self, successful):
             workspace.last_build = key
             self._get_context().get_workspaces().save_config()
 
+    # _adopt_artifact():
+    #
+    # Adopt the cached artifact of another element with the same strong
+    # cache key instead of building or pulling it again.
+    #
+    # Args:
+    #    element (Element): The element which cached the artifact
+    #
+    def _adopt_artifact(self, element):
+        self.__artifacts.link_element_key(element, self, self._get_cache_key())
+
+        if self.__assemble_scheduled:
+            self._assemble_done(True)
+        else:
+            self._load_artifact(pull=False)
+            self._load_artifact_done()
+
     # _assemble():
     #
     # Internal method for running the entire build phase.
diff --git a/tests/scheduler/__init__.py b/tests/scheduler/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/scheduler/deduplication.py b/tests/scheduler/deduplication.py
new file mode 100644
index 000000000..1fb9b2a05
--- /dev/null
+++ b/tests/scheduler/deduplication.py
@@ -0,0 +1,100 @@
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+#
+
+# Pylint doesn't play well with fixtures and dependency injection from pytest
+# pylint: disable=redefined-outer-name
+
+import os
+import shutil
+import pytest
+
+from buildstream._testing import cli  # pylint: disable=unused-import
+from buildstream.exceptions import ErrorDomain
+
+from tests.testutils import create_artifact_share
+
+# Project directory
+DATA_DIR = os.path.join(
+    os.path.dirname(os.path.realpath(__file__)),
+    "deduplication",
+)
+
+
+# Elements sharing a strong cache key are built only once
+@pytest.mark.datafiles(DATA_DIR)
+@pytest.mark.parametrize("strict", [True, False], ids=["strict", "non-strict"])
+def test_build(cli, datafiles, strict):
+    project = os.path.join(str(datafiles), "simple")
+    cli.configure({"projects": {"deduplication-simple": {"strict": strict}}})
+
+    assert cli.get_element_key(project, "a.bst") == cli.get_element_key(project, "b.bst")
+
+    result = cli.run(project=project, args=["build", "a.bst", "b.bst"])
+    result.assert_success()
+    assert len(result.get_built_elements()) == 1
+
+    states = cli.get_element_states(project, ["a.bst", "b.bst"])
+    assert states == {"a.bst": "cached", "b.bst": "cached"}
+
+
+# The same element loaded through two junctions is built only once
+@pytest.mark.datafiles(DATA_DIR)
+def test_build_junctions(cli, datafiles):
+    project = os.path.join(str(datafiles), "multi-stage")
+
+    result = cli.run(project=project, args=["build", "a.bst", "b.bst"])
+    result.assert_success()
+
+    built = [element for element in result.get_built_elements() if element.endswith(":hello.bst")]
+    assert len(built) == 1
+
+    states = cli.get_element_states(project, ["subproject-a.bst:hello.bst", "subproject-b.bst:hello.bst"])
+    assert set(states.values()) == {"cached"}
+
+
+# When the build of an element fails, the elements waiting for it are built
+@pytest.mark.datafiles(DATA_DIR)
+def test_build_failure(cli, datafiles):
+    project = os.path.join(str(datafiles), "failure")
+
+    result = cli.run(project=project, args=["--on-error", "continue", "build", "a.bst", "b.bst"])
+    result.assert_main_error(ErrorDomain.STREAM, None)
+
+    assert sorted(result.get_start_order("build")) == ["a.bst", "b.bst"]
+
+    states = cli.get_element_states(project, ["a.bst", "b.bst"])
+    assert states == {"a.bst": "failed", "b.bst": "failed"}
+
+
+# Elements sharing a strong cache key are pulled only once
+@pytest.mark.datafiles(DATA_DIR)
+def test_pull(cli, tmpdir, datafiles):
+    project = os.path.join(str(datafiles), "simple")
+
+    with create_artifact_share(os.path.join(str(tmpdir), "artifactshare")) as share:
+        cli.configure({"artifacts": {"servers": [{"url": share.repo, "push": True}]}})
+
+        result = cli.run(project=project, args=["build", "a.bst", "b.bst"])
+        result.assert_success()
+
+        # Wipe the local cache so that the artifacts must be pulled
+        shutil.rmtree(os.path.join(cli.directory, "cas"))
+        shutil.rmtree(os.path.join(cli.directory, "artifacts"))
+
+        result = cli.run(project=project, args=["artifact", "pull", "a.bst", "b.bst"])
+        result.assert_success()
+        assert len(result.get_pulled_elements()) == 1
+
+        states = cli.get_element_states(project, ["a.bst", "b.bst"])
+        assert states == {"a.bst": "cached", "b.bst": "cached"}
diff --git a/tests/scheduler/deduplication/failure/elements/a.bst b/tests/scheduler/deduplication/failure/elements/a.bst
new file mode 100644
index 000000000..4d9c562ef
--- /dev/null
+++ b/tests/scheduler/deduplication/failure/elements/a.bst
@@ -0,0 +1 @@
+kind: fail
diff --git a/tests/scheduler/deduplication/failure/elements/b.bst b/tests/scheduler/deduplication/failure/elements/b.bst
new file mode 100644
index 000000000..4d9c562ef
--- /dev/null
+++ b/tests/scheduler/deduplication/failure/elements/b.bst
@@ -0,0 +1 @@
+kind: fail
diff --git a/tests/scheduler/deduplication/failure/plugins/fail.py b/tests/scheduler/deduplication/failure/plugins/fail.py
new file mode 100644
index 000000000..90f3b319b
--- /dev/null
+++ b/tests/scheduler/deduplication/failure/plugins/fail.py
@@ -0,0 +1,28 @@
+from buildstream import Element, ElementError
+
+
+class Fail(Element):
+    BST_MIN_VERSION = "2.0"
+
+    def configure(self, node):
+        pass
+
+    def preflight(self):
+        pass
+
+    def get_unique_key(self):
+        return {}
+
+    def configure_sandbox(self, sandbox):
+        pass
+
+    def stage(self, sandbox):
+        pass
+
+    def assemble(self, sandbox):
+        raise ElementError("Failing on purpose", reason="fail")
+
+
+# Plugin entry point
+def setup():
+    return Fail
diff --git a/tests/scheduler/deduplication/failure/project.conf b/tests/scheduler/deduplication/failure/project.conf
new file mode 100644
index 000000000..15100c6cb
--- /dev/null
+++ b/tests/scheduler/deduplication/failure/project.conf
@@ -0,0 +1,12 @@
+name: deduplication-failure
+min-version: 2.0
+element-path: elements
+
+variables:
+  build-root: /buildstream/shared
+
+plugins:
+- origin: local
+  path: plugins
+  elements:
+  - fail
diff --git a/tests/scheduler/deduplication/multi-stage/elements/a.bst b/tests/scheduler/deduplication/multi-stage/elements/a.bst
new file mode 100644
index 000000000..ddabcca93
--- /dev/null
+++ b/tests/scheduler/deduplication/multi-stage/elements/a.bst
@@ -0,0 +1,4 @@
+kind: stack
+
+depends:
+- subproject-a.bst:hello.bst
diff --git a/tests/scheduler/deduplication/multi-stage/elements/b.bst b/tests/scheduler/deduplication/multi-stage/elements/b.bst
new file mode 100644
index 000000000..310454cbc
--- /dev/null
+++ b/tests/scheduler/deduplication/multi-stage/elements/b.bst
@@ -0,0 +1,4 @@
+kind: stack
+
+depends:
+- subproject-b.bst:hello.bst
diff --git a/tests/scheduler/deduplication/multi-stage/elements/subproject-a.bst b/tests/scheduler/deduplication/multi-stage/elements/subproject-a.bst
new file mode 100644
index 000000000..6664eeec6
--- /dev/null
+++ b/tests/scheduler/deduplication/multi-stage/elements/subproject-a.bst
@@ -0,0 +1,5 @@
+kind: junction
+
+sources:
+- kind: local
+  path: subproject
diff --git a/tests/scheduler/deduplication/multi-stage/elements/subproject-b.bst b/tests/scheduler/deduplication/multi-stage/elements/subproject-b.bst
new file mode 100644
index 000000000..6664eeec6
--- /dev/null
+++ b/tests/scheduler/deduplication/multi-stage/elements/subproject-b.bst
@@ -0,0 +1,5 @@
+kind: junction
+
+sources:
+- kind: local
+  path: subproject
diff --git a/tests/scheduler/deduplication/multi-stage/project.conf b/tests/scheduler/deduplication/multi-stage/project.conf
new file mode 100644
index 000000000..6224762b9
--- /dev/null
+++ b/tests/scheduler/deduplication/multi-stage/project.conf
@@ -0,0 +1,8 @@
+name: deduplication-multi-stage
+min-version: 2.0
+element-path: elements
+
+junctions:
+  internal:
+  - subproject-a.bst
+  - subproject-b.bst
diff --git a/tests/scheduler/deduplication/multi-stage/subproject/elements/hello.bst b/tests/scheduler/deduplication/multi-stage/subproject/elements/hello.bst
new file mode 100644
index 000000000..003b5bf39
--- /dev/null
+++ b/tests/scheduler/deduplication/multi-stage/subproject/elements/hello.bst
@@ -0,0 +1,4 @@
+kind: import
+sources:
+- kind: local
+  path: files/hello.txt
diff --git a/tests/scheduler/deduplication/multi-stage/subproject/files/hello.txt b/tests/scheduler/deduplication/multi-stage/subproject/files/hello.txt
new file mode 100644
index 000000000..b6fc4c620
--- /dev/null
+++ b/tests/scheduler/deduplication/multi-stage/subproject/files/hello.txt
@@ -0,0 +1 @@
+hello
\ No newline at end of file
diff --git a/tests/scheduler/deduplication/multi-stage/subproject/project.conf b/tests/scheduler/deduplication/multi-stage/subproject/project.conf
new file mode 100644
index 000000000..c6e13a045
--- /dev/null
+++ b/tests/scheduler/deduplication/multi-stage/subproject/project.conf
@@ -0,0 +1,3 @@
+name: deduplication-multi-stage-subproject
+min-version: 2.0
+element-path: elements
diff --git a/tests/scheduler/deduplication/simple/elements/a.bst b/tests/scheduler/deduplication/simple/elements/a.bst
new file mode 100644
index 000000000..003b5bf39
--- /dev/null
+++ b/tests/scheduler/deduplication/simple/elements/a.bst
@@ -0,0 +1,4 @@
+kind: import
+sources:
+- kind: local
+  path: files/hello.txt
diff --git a/tests/scheduler/deduplication/simple/elements/b.bst b/tests/scheduler/deduplication/simple/elements/b.bst
new file mode 100644
index 000000000..003b5bf39
--- /dev/null
+++ b/tests/scheduler/deduplication/simple/elements/b.bst
@@ -0,0 +1,4 @@
+kind: import
+sources:
+- kind: local
+  path: files/hello.txt
diff --git a/tests/scheduler/deduplication/simple/files/hello.txt b/tests/scheduler/deduplication/simple/files/hello.txt
new file mode 100644
index 000000000..b6fc4c620
--- /dev/null
+++ b/tests/scheduler/deduplication/simple/files/hello.txt
@@ -0,0 +1 @@
+hello
\ No newline at end of file
diff --git a/tests/scheduler/deduplication/simple/project.conf b/tests/scheduler/deduplication/simple/project.conf
new file mode 100644
index 000000000..83365e94e
--- /dev/null
+++ b/tests/scheduler/deduplication/simple/project.conf
@@ -0,0 +1,6 @@
+name: deduplication-simple
+min-version: 2.0
+element-path: elements
+
+variables:
+  build-root: /buildstream/shared