diff --git a/src/buildstream/_frontend/app.py b/src/buildstream/_frontend/app.py
index a6b08cfdd..577d80d4d 100644
--- a/src/buildstream/_frontend/app.py
+++ b/src/buildstream/_frontend/app.py
@@ -282,6 +282,7 @@ def initialized(self, *, session_name=None):
                     self.context,
                     cli_options=self._main_options["option"],
                     default_mirror=self._main_options.get("default_mirror"),
+                    fetch_subprojects=self.stream.fetch_subprojects,
                 )
             except LoadError as e:
 
diff --git a/src/buildstream/_pluginfactory/pluginoriginjunction.py b/src/buildstream/_pluginfactory/pluginoriginjunction.py
index 4f5c10b20..ddbb95c6d 100644
--- a/src/buildstream/_pluginfactory/pluginoriginjunction.py
+++ b/src/buildstream/_pluginfactory/pluginoriginjunction.py
@@ -41,6 +41,8 @@ def get_plugin_paths(self, kind, plugin_type):
             factory = project.source_factory
         elif plugin_type == PluginType.ELEMENT:
             factory = project.element_factory
+        elif plugin_type == PluginType.SOURCE_MIRROR:
+            factory = project.source_mirror_factory
         else:
             assert False, "unreachable"
 
@@ -66,7 +68,7 @@ def get_plugin_paths(self, kind, plugin_type):
             # subproject.
             #
             raise PluginError(
-                "{}: project '{}' referred to by junction '{}' does not declare any {} plugin kind: '{}'".format(
+                "{}: project '{}' referred to by junction '{}' does not declare a {} plugin named: '{}'".format(
                     self.provenance_node.get_provenance(), project.name, self._junction, plugin_type, kind
                 ),
                 reason="junction-plugin-not-found",
diff --git a/src/buildstream/_pluginfactory/pluginoriginpip.py b/src/buildstream/_pluginfactory/pluginoriginpip.py
index 542776e0b..3bded89ab 100644
--- a/src/buildstream/_pluginfactory/pluginoriginpip.py
+++ b/src/buildstream/_pluginfactory/pluginoriginpip.py
@@ -48,6 +48,8 @@ def get_plugin_paths(self, kind, plugin_type):
             entrypoint_group = "buildstream.plugins.sources"
         elif plugin_type == PluginType.ELEMENT:
             entrypoint_group = "buildstream.plugins.elements"
+        elif plugin_type == PluginType.SOURCE_MIRROR:
+            entrypoint_group = "buildstream.plugins.sourcemirrors"
         else:
             assert False, "unreachable"
 
@@ -84,8 +86,8 @@ def get_plugin_paths(self, kind, plugin_type):
 
         if package is None:
             raise PluginError(
-                "{}: Pip package {} does not contain a plugin named '{}'".format(
-                    self.provenance_node.get_provenance(), self._package_name, kind
+                "{}: Pip package {} does not contain a {} plugin named '{}'".format(
+                    self.provenance_node.get_provenance(), self._package_name, plugin_type, kind
                 ),
                 reason="plugin-not-found",
             )
diff --git a/src/buildstream/_project.py b/src/buildstream/_project.py
index 5877619b1..10e410cdf 100644
--- a/src/buildstream/_project.py
+++ b/src/buildstream/_project.py
@@ -90,6 +90,7 @@ def __init__(
         parent_loader: Optional[Loader] = None,
         provenance_node: Optional[ProvenanceInformation] = None,
         search_for_project: bool = True,
+        fetch_subprojects=None
     ):
         #
         # Public members
@@ -161,6 +162,7 @@ def __init__(
             self.load_context = parent_loader.load_context
         else:
             self.load_context = LoadContext(self._context)
+            self.load_context.set_fetch_subprojects(fetch_subprojects)
 
         if search_for_project:
             self.directory, self._invoked_from_workspace_element = self._find_project_dir(directory)
@@ -1100,6 +1102,9 @@ def _load_pass(self, config, output, *, ignore_unknown=False):
         # Override default_mirror if not set by command-line
         output.default_mirror = self._default_mirror or overrides.get_str("default-mirror", default=None)
 
+        # Source url aliases
+        output._aliases = config.get_mapping("aliases", default={})
+
         # First try mirrors specified in user configuration, user configuration
         # is allowed to completely disable mirrors by specifying an empty list,
         # so we check for a None value here too.
@@ -1121,9 +1126,6 @@ def _load_pass(self, config, output, *, ignore_unknown=False):
             if not output.default_mirror:
                 output.default_mirror = mirror.name
 
-        # Source url aliases
-        output._aliases = config.get_mapping("aliases", default={})
-
         # Perform variable substitutions in source aliases
         variables.expand(output._aliases)
 
diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index 973e58e0a..2266b27f0 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -127,8 +127,6 @@ def cleanup(self):
     def set_project(self, project):
         assert self._project is None
         self._project = project
-        if self._project:
-            self._project.load_context.set_fetch_subprojects(self._fetch_subprojects)
 
     # load_selection()
     #
@@ -1323,6 +1321,22 @@ def retry_job(self, action_name: str, unique_id: str) -> None:
         assert queue
         queue.enqueue([element])
 
+    # fetch_subprojects()
+    #
+    # Fetch subprojects as part of the project and element loading process.
+    #
+    # This is passed to the Project in order to handle loading of subprojects
+    #
+    # Args:
+    #    junctions (list of Element): The junctions to fetch
+    #
+    def fetch_subprojects(self, junctions):
+        self._reset()
+        queue = FetchQueue(self._scheduler)
+        queue.enqueue(junctions)
+        self.queues = [queue]
+        self._run()
+
     #############################################################
     #                    Private Methods                        #
     #############################################################
@@ -1343,20 +1357,6 @@ def _assert_project(self, message: str) -> None:
                 message, detail="No project.conf or active workspace was located", reason="project-not-loaded"
             )
 
-    # _fetch_subprojects()
-    #
-    # Fetch subprojects as part of the project and element loading process.
-    #
-    # Args:
-    #    junctions (list of Element): The junctions to fetch
-    #
-    def _fetch_subprojects(self, junctions):
-        self._reset()
-        queue = FetchQueue(self._scheduler)
-        queue.enqueue(junctions)
-        self.queues = [queue]
-        self._run()
-
     # _load_artifacts()
     #
     # Loads artifacts from target artifact refs
diff --git a/tests/frontend/mirror.py b/tests/frontend/mirror.py
index 7701b6616..4c42f23c8 100644
--- a/tests/frontend/mirror.py
+++ b/tests/frontend/mirror.py
@@ -26,6 +26,7 @@
 from buildstream._testing import cli  # pylint: disable=unused-import
 
 from tests.testutils.repo.git import Git
+from tests.testutils.repo.tar import Tar
 from tests.testutils.site import pip_sample_packages  # pylint: disable=unused-import
 from tests.testutils.site import SAMPLE_PACKAGES_SKIP_REASON
 
@@ -817,7 +818,8 @@ def test_mirror_expand_project_and_toplevel_root(cli, tmpdir):
 #
 @pytest.mark.datafiles(DATA_DIR)
 @pytest.mark.usefixtures("datafiles")
-def test_source_mirror_plugin(cli, tmpdir):
+@pytest.mark.parametrize("origin", [("local"), ("junction"), ("pip")])
+def test_source_mirror_plugin(cli, tmpdir, origin):
     output_file = os.path.join(str(tmpdir), "output.txt")
     project_dir = str(tmpdir)
     element_dir = os.path.join(project_dir, "elements")
@@ -827,6 +829,36 @@ def test_source_mirror_plugin(cli, tmpdir):
     element = generate_element(output_file)
     _yaml.roundtrip_dump(element, element_path)
 
+    def source_mirror_plugin_origin():
+        if origin == "local":
+            return {"origin": "local", "path": "sourcemirrors", "source-mirrors": ["mirror"]}
+        elif origin == "pip":
+            return {
+                "origin": "pip",
+                "package-name": "sample-plugins>=1.2",
+                "source-mirrors": ["mirror"],
+            }
+        elif origin == "junction":
+            # For junction loading, just copy in the sample-plugins into a subdir and
+            # create a local junction
+            sample_plugins_dir = os.path.join(TOP_DIR, "..", "plugins", "sample-plugins")
+            sample_plugins_copy_dir = os.path.join(project_dir, "sample-plugins-copy")
+            junction_file = os.path.join(element_dir, "sample-plugins.bst")
+
+            shutil.copytree(sample_plugins_dir, sample_plugins_copy_dir)
+
+            _yaml.roundtrip_dump(
+                {"kind": "junction", "sources": [{"kind": "local", "path": "sample-plugins-copy"}]}, junction_file
+            )
+
+            return {
+                "origin": "junction",
+                "junction": "sample-plugins.bst",
+                "source-mirrors": ["mirror"],
+            }
+        else:
+            assert False
+
     project_file = os.path.join(project_dir, "project.conf")
     project = {
         "name": "test",
@@ -870,7 +902,7 @@ def test_source_mirror_plugin(cli, tmpdir):
         ],
         "plugins": [
             {"origin": "local", "path": "sources", "sources": ["fetch_source"]},
-            {"origin": "local", "path": "sourcemirrors", "source-mirrors": ["mirror"]},
+            source_mirror_plugin_origin(),
         ],
     }
 
@@ -1098,3 +1130,107 @@ def map_alias_valid(x):
                 assert "Fetch RAB/repo2 succeeded from RAB/repo2" in contents
             else:
                 assert "Fetch bar:repo2 succeeded from RAB/repo2" in contents
+
+
+# Test the behavior of loading a SourceMirror plugin across a junction,
+# when the cross junction SourceMirror object has a mirror.
+#
+# Check what happens when the mirror does not need to be exercized (success)
+#
+# Check what happens when the mirror needs to be exercised in order to obtain
+# the mirror plugin itself (failure) and check the failure mode.
+#
+#
+@pytest.mark.parametrize("fetch_source", [("all"), ("mirrors")], ids=["normal", "circular"])
+def test_source_mirror_circular_junction(cli, tmpdir, fetch_source):
+    project_dir = str(tmpdir)
+    element_dir = os.path.join(project_dir, "elements")
+    os.makedirs(element_dir, exist_ok=True)
+
+    cli.configure({"fetch": {"source": fetch_source}})
+
+    # Generate a 2 tar repos with the sample plugins
+    #
+    sample_plugins_dir = os.path.join(TOP_DIR, "..", "plugins", "sample-plugins")
+    base_sample_plugins_repodir = os.path.join(str(tmpdir), "base_sample_plugins")
+    base_sample_plugins_repo = Tar(base_sample_plugins_repodir)
+    base_sample_plugins_ref = base_sample_plugins_repo.create(sample_plugins_dir)
+    mirror_sample_plugins_repodir = os.path.join(str(tmpdir), "mirror_sample_plugins")
+    mirror_sample_plugins_repo = Tar(mirror_sample_plugins_repodir)
+
+    # Don't expect determinism from python tar, just copy over the Tar repo file
+    # and we need to use the same ref for both.
+    shutil.copyfile(
+        os.path.join(base_sample_plugins_repo.repo, "file.tar.gz"),
+        os.path.join(mirror_sample_plugins_repo.repo, "file.tar.gz"),
+    )
+
+    # Generate junction for sample plugins
+    #
+    sample_plugins_junction = {
+        "kind": "junction",
+        "sources": [
+            {
+                "kind": "tar",
+                "url": "samplemirror:file.tar.gz",
+                "ref": base_sample_plugins_ref,
+            }
+        ],
+    }
+    element_path = os.path.join(element_dir, "sample-plugins.bst")
+    _yaml.roundtrip_dump(sample_plugins_junction, element_path)
+
+    # Generate project.conf
+    #
+    project_file = os.path.join(project_dir, "project.conf")
+    project = {
+        "name": "test",
+        "min-version": "2.0",
+        "element-path": "elements",
+        "aliases": {
+            "samplemirror": "file://" + base_sample_plugins_repo.repo + "/",
+        },
+        "mirrors": [
+            {
+                "name": "alternative",
+                "kind": "mirror",
+                "config": {
+                    "aliases": {
+                        "samplemirror": ["file://" + mirror_sample_plugins_repo.repo + "/"],
+                    },
+                },
+            },
+        ],
+        "plugins": [
+            {"origin": "junction", "junction": "sample-plugins.bst", "source-mirrors": ["mirror"]},
+        ],
+    }
+    _yaml.roundtrip_dump(project, project_file)
+
+    # Make a silly element
+    element = {"kind": "import", "sources": [{"kind": "local", "path": "project.conf"}]}
+    element_path = os.path.join(element_dir, "test.bst")
+    _yaml.roundtrip_dump(element, element_path)
+
+    result = cli.run(project=project_dir, args=["show", "test.bst"])
+
+    if fetch_source == "all":
+        result.assert_success()
+    elif fetch_source == "mirrors":
+        #
+        # This error looks like this:
+        #
+        #   Error loading project: tar source at sample-plugins.bst [line 3 column 2]: No fetch URI found for alias 'samplemirror'
+        #
+        #       Check fetch controls in your user configuration
+        #
+        # This is not 100% ideal, as we could theoretically have Source.mark_download_url() detect
+        # the case that we are currently instantiating the specific SourceMirror plugin required
+        # to resolve the URL needed to obtain the same said SourceMirror plugin, and report
+        # something about this being a circular dependency error.
+        #
+        # However, this would be fairly complex to reason about in the code, especially considering
+        # the source alias redirects, and the possibility that a subproject's source mirror is being
+        # redirected to a parent project's aliases and corresponding mirrors.
+        #
+        result.assert_main_error(ErrorDomain.SOURCE, "missing-source-alias-target")
diff --git a/tests/plugins/sample-plugins/project.conf b/tests/plugins/sample-plugins/project.conf
index 60fd37224..d111eae6d 100644
--- a/tests/plugins/sample-plugins/project.conf
+++ b/tests/plugins/sample-plugins/project.conf
@@ -13,3 +13,8 @@ plugins:
   sources:
   - git
   - sample
+
+- origin: local
+  path: src/sample_plugins/sourcemirrors
+  source-mirrors:
+  - mirror
diff --git a/tests/plugins/sample-plugins/setup.py b/tests/plugins/sample-plugins/setup.py
index 14fe1cf5b..04c8361cc 100755
--- a/tests/plugins/sample-plugins/setup.py
+++ b/tests/plugins/sample-plugins/setup.py
@@ -33,6 +33,9 @@
             "sample = sample_plugins.sources.sample",
             "git = sample_plugins.sources.git",
         ],
+        "buildstream.plugins.sourcemirrors": [
+            "mirror = sample_plugins.sourcemirrors.mirror",
+        ],
     },
     zip_safe=False,
 )
diff --git a/tests/plugins/sample-plugins/src/sample_plugins/sourcemirrors/__init__.py b/tests/plugins/sample-plugins/src/sample_plugins/sourcemirrors/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/plugins/sample-plugins/src/sample_plugins/sourcemirrors/mirror.py b/tests/plugins/sample-plugins/src/sample_plugins/sourcemirrors/mirror.py
new file mode 100644
index 000000000..fc737031a
--- /dev/null
+++ b/tests/plugins/sample-plugins/src/sample_plugins/sourcemirrors/mirror.py
@@ -0,0 +1,36 @@
+from typing import Optional, Dict, Any
+
+from buildstream import SourceMirror, MappingNode
+
+
+# This mirror plugin basically implements the default behavior
+# by loading the alias definitions as custom "config" configuration
+# instead, and implementing the translate_url method.
+#
+class Sample(SourceMirror):
+    def configure(self, node):
+        node.validate_keys(["aliases"])
+
+        self.aliases = {}
+
+        aliases = node.get_mapping("aliases")
+        for alias_name, url_list in aliases.items():
+            self.aliases[alias_name] = url_list.as_str_list()
+
+        self.set_supported_aliases(self.aliases.keys())
+
+    def translate_url(
+        self,
+        *,
+        alias: str,
+        alias_url: str,
+        source_url: str,
+        extra_data: Optional[Dict[str, Any]],
+    ) -> str:
+        return self.aliases[alias][0] + source_url
+
+
+# Plugin entry point
+def setup():
+
+    return Sample