diff --git a/beets/autotag/hooks.py b/beets/autotag/hooks.py
index e0e2a5a795..d1f9e17bcf 100644
--- a/beets/autotag/hooks.py
+++ b/beets/autotag/hooks.py
@@ -23,6 +23,7 @@
 
 from typing_extensions import Self
 
+from beets import plugins
 from beets.util import cached_classproperty
 from beets.util.deprecation import deprecate_for_maintainers
 
@@ -59,6 +60,18 @@ def __hash__(self) -> int:  # type: ignore[override]
 class Info(AttrDict[Any]):
     """Container for metadata about a musical entity."""
 
+    Identifier = tuple[str | None, str | None]
+
+    @property
+    def id(self) -> str | None:
+        """Return the provider-specific identifier for this metadata object."""
+        raise NotImplementedError
+
+    @property
+    def identifier(self) -> Identifier:
+        """Return a cross-provider key in ``(data_source, id)`` form."""
+        return (self.data_source, self.id)
+
     @cached_property
     def name(self) -> str:
         raise NotImplementedError
@@ -118,6 +131,10 @@ class AlbumInfo(Info):
     user items, and later to drive tagging decisions once selected.
     """
 
+    @property
+    def id(self) -> str | None:
+        return self.album_id
+
     @cached_property
     def name(self) -> str:
         return self.album or ""
@@ -194,6 +211,10 @@ class TrackInfo(Info):
     stand alone for singleton matching.
     """
 
+    @property
+    def id(self) -> str | None:
+        return self.track_id
+
     @cached_property
     def name(self) -> str:
         return self.title or ""
@@ -262,6 +283,10 @@ class AlbumMatch(Match):
     extra_items: list[Item]
     extra_tracks: list[TrackInfo]
 
+    def __post_init__(self) -> None:
+        """Notify listeners when an album candidate has been matched."""
+        plugins.send("album_matched", match=self)
+
     @property
     def item_info_pairs(self) -> list[tuple[Item, TrackInfo]]:
         return list(self.mapping.items())
diff --git a/beets/autotag/match.py b/beets/autotag/match.py
index 374ea3c13c..d99369c015 100644
--- a/beets/autotag/match.py
+++ b/beets/autotag/match.py
@@ -19,16 +19,17 @@
 from __future__ import annotations
 
 from enum import IntEnum
-from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar
+from typing import TYPE_CHECKING, NamedTuple, TypeVar
 
 import lap
 import numpy as np
 
-from beets import config, logging, metadata_plugins, plugins
+from beets import config, logging, metadata_plugins
 from beets.autotag import AlbumMatch, TrackMatch, hooks
 from beets.util import get_most_common_tags
 
 from .distance import VA_ARTISTS, distance, track_distance
+from .hooks import Info
 
 if TYPE_CHECKING:
     from collections.abc import Iterable, Sequence
@@ -36,6 +37,10 @@
     from beets.autotag import AlbumInfo, TrackInfo
     from beets.library import Item
 
+
+AnyMatch = TypeVar("AnyMatch", TrackMatch, AlbumMatch)
+Candidates = dict[Info.Identifier, AnyMatch]
+
 # Global logger.
 log = logging.getLogger("beets")
 
@@ -99,28 +104,21 @@ def assign_items(
     return list(mapping.items()), extra_items, extra_tracks
 
 
-def match_by_id(items: Iterable[Item]) -> AlbumInfo | None:
-    """If the items are tagged with an external source ID, return an
-    AlbumInfo object for the corresponding album. Otherwise, returns
-    None.
-    """
-    albumids = (item.mb_albumid for item in items if item.mb_albumid)
+def match_by_id(album_id: str | None, consensus: bool) -> Iterable[AlbumInfo]:
+    """Return album candidates for the given album id.
 
-    # Did any of the items have an MB album ID?
-    try:
-        first = next(albumids)
-    except StopIteration:
+    Make sure that the ID is present and that there is consensus on it among
+    the items being tagged.
+    """
+    if not album_id:
         log.debug("No album ID found.")
-        return None
+    elif not consensus:
+        log.debug("No album ID consensus.")
+    else:
+        log.debug("Searching for discovered album ID: {}", album_id)
+        return metadata_plugins.albums_for_ids([album_id])
 
-    # Is there a consensus on the MB album ID?
-    for other in albumids:
-        if other != first:
-            log.debug("No album ID consensus.")
-            return None
-    # If all album IDs are equal, look up the album.
-    log.debug("Searching for discovered album ID: {}", first)
-    return metadata_plugins.album_for_id(first)
+    return ()
 
 
 def _recommendation(
@@ -180,9 +178,6 @@ def _recommendation(
     return rec
 
 
-AnyMatch = TypeVar("AnyMatch", TrackMatch, AlbumMatch)
-
-
 def _sort_candidates(candidates: Iterable[AnyMatch]) -> Sequence[AnyMatch]:
     """Sort candidates by distance."""
     return sorted(candidates, key=lambda match: match.distance)
@@ -190,7 +185,7 @@ def _sort_candidates(candidates: Iterable[AnyMatch]) -> Sequence[AnyMatch]:
 
 def _add_candidate(
     items: Sequence[Item],
-    results: dict[Any, AlbumMatch],
+    results: Candidates[AlbumMatch],
     info: AlbumInfo,
 ):
     """Given a candidate AlbumInfo object, attempt to add the candidate
@@ -198,7 +193,10 @@ def _add_candidate(
     checking the track count, ordering the items, checking for
     duplicates, and calculating the distance.
     """
-    log.debug("Candidate: {0.artist} - {0.album} ({0.album_id})", info)
+    log.debug(
+        "Candidate: {0.artist} - {0.album} ({0.album_id}) from {0.data_source}",
+        info,
+    )
 
     # Discard albums with zero tracks.
     if not info.tracks:
@@ -206,7 +204,7 @@ def _add_candidate(
         return
 
     # Prevent duplicates.
-    if info.album_id and info.album_id in results:
+    if info.album_id and info.identifier in results:
         log.debug("Duplicate.")
         return
 
@@ -234,7 +232,7 @@ def _add_candidate(
             return
 
     log.debug("Success. Distance: {}", dist)
-    results[info.album_id] = hooks.AlbumMatch(
+    results[info.identifier] = hooks.AlbumMatch(
         dist, info, dict(item_info_pairs), extra_items, extra_tracks
     )
 
@@ -268,39 +266,37 @@ def tag_album(
     cur_album: str = likelies["album"]
     log.debug("Tagging {} - {}", cur_artist, cur_album)
 
-    # The output result, keys are the MB album ID.
-    candidates: dict[Any, AlbumMatch] = {}
+    # The output result, keys are (data_source, album_id) pairs, values are
+    # AlbumMatch objects.
+    candidates: Candidates[AlbumMatch] = {}
 
     # Search by explicit ID.
     if search_ids:
-        for search_id in search_ids:
-            log.debug("Searching for album ID: {}", search_id)
-            if info := metadata_plugins.album_for_id(search_id):
-                _add_candidate(items, candidates, info)
-                if opt_candidate := candidates.get(info.album_id):
-                    plugins.send("album_matched", match=opt_candidate)
+        log.debug("Searching for album IDs: {}", ", ".join(search_ids))
+        for _info in metadata_plugins.albums_for_ids(search_ids):
+            _add_candidate(items, candidates, _info)
 
     # Use existing metadata or text search.
     else:
         # Try search based on current ID.
-        if info := match_by_id(items):
+        for info in match_by_id(
+            likelies["mb_albumid"], consensus["mb_albumid"]
+        ):
             _add_candidate(items, candidates, info)
-            for candidate in candidates.values():
-                plugins.send("album_matched", match=candidate)
-
-            rec = _recommendation(list(candidates.values()))
-            log.debug("Album ID match recommendation is {}", rec)
-            if candidates and not config["import"]["timid"]:
-                # If we have a very good MBID match, return immediately.
-                # Otherwise, this match will compete against metadata-based
-                # matches.
-                if rec == Recommendation.strong:
-                    log.debug("ID match.")
-                    return (
-                        cur_artist,
-                        cur_album,
-                        Proposal(list(candidates.values()), rec),
-                    )
+
+        rec = _recommendation(list(candidates.values()))
+        log.debug("Album ID match recommendation is {}", rec)
+        if candidates and not config["import"]["timid"]:
+            # If we have a very good MBID match, return immediately.
+            # Otherwise, this match will compete against metadata-based
+            # matches.
+            if rec == Recommendation.strong:
+                log.debug("ID match.")
+                return (
+                    cur_artist,
+                    cur_album,
+                    Proposal(list(candidates.values()), rec),
+                )
 
         # Search terms.
         if not (search_artist and search_name):
@@ -321,8 +317,6 @@ def tag_album(
             items, search_artist, search_name, va_likely
         ):
             _add_candidate(items, candidates, matched_candidate)
-            if opt_candidate := candidates.get(matched_candidate.album_id):
-                plugins.send("album_matched", match=opt_candidate)
 
     log.debug("Evaluating {} candidates.", len(candidates))
     # Sort and get the recommendation.
@@ -344,27 +338,24 @@ def tag_item(
     metadata in the search query. `search_ids` may be used for restricting the
     search to a list of metadata backend IDs.
     """
-    # Holds candidates found so far: keys are MBIDs; values are
-    # (distance, TrackInfo) pairs.
-    candidates = {}
+    # Holds candidates found so far: keys are (data_source, track_id) pairs,
+    # values TrackMatch objects
+    candidates: Candidates[TrackMatch] = {}
     rec: Recommendation | None = None
 
     # First, try matching by the external source ID.
     trackids = search_ids or [t for t in [item.mb_trackid] if t]
     if trackids:
-        for trackid in trackids:
-            log.debug("Searching for track ID: {}", trackid)
-            if info := metadata_plugins.track_for_id(trackid):
-                dist = track_distance(item, info, incl_artist=True)
-                candidates[info.track_id] = hooks.TrackMatch(dist, info)
-                # If this is a good match, then don't keep searching.
-                rec = _recommendation(_sort_candidates(candidates.values()))
-                if (
-                    rec == Recommendation.strong
-                    and not config["import"]["timid"]
-                ):
-                    log.debug("Track ID match.")
-                    return Proposal(_sort_candidates(candidates.values()), rec)
+        log.debug("Searching for track IDs: {}", ", ".join(trackids))
+        for info in metadata_plugins.tracks_for_ids(trackids):
+            dist = track_distance(item, info, incl_artist=True)
+            candidates[info.identifier] = hooks.TrackMatch(dist, info)
+
+        # If this is a good match, then don't keep searching.
+        rec = _recommendation(_sort_candidates(candidates.values()))
+        if rec == Recommendation.strong and not config["import"]["timid"]:
+            log.debug("Track ID match.")
+            return Proposal(_sort_candidates(candidates.values()), rec)
 
     # If we're searching by ID, don't proceed.
     if search_ids:
@@ -384,7 +375,7 @@ def tag_item(
         item, search_artist, search_name
     ):
         dist = track_distance(item, track_info, incl_artist=True)
-        candidates[track_info.track_id] = hooks.TrackMatch(dist, track_info)
+        candidates[track_info.identifier] = hooks.TrackMatch(dist, track_info)
 
     # Sort by distance and return with recommendation.
     log.debug("Found {} candidates.", len(candidates))
diff --git a/beets/metadata_plugins.py b/beets/metadata_plugins.py
index 441415b344..9e5e2081e2 100644
--- a/beets/metadata_plugins.py
+++ b/beets/metadata_plugins.py
@@ -9,7 +9,7 @@
 
 import abc
 import re
-from contextlib import contextmanager, nullcontext
+from contextlib import contextmanager
 from functools import cache, cached_property, wraps
 from typing import (
     TYPE_CHECKING,
@@ -27,7 +27,7 @@
 from beets.util import cached_classproperty
 from beets.util.id_extractors import extract_release_id
 
-from .plugins import BeetsPlugin, find_plugins, notify_info_yielded
+from .plugins import BeetsPlugin, find_plugins, notify_info_yielded, send
 
 Ret = TypeVar("Ret")
 QueryType = Literal["album", "track"]
@@ -49,14 +49,27 @@ def find_metadata_source_plugins() -> list[MetadataSourcePlugin]:
     return [p for p in find_plugins() if hasattr(p, "data_source")]  # type: ignore[misc]
 
 
+@cache
+def get_metadata_source(name: str) -> MetadataSourcePlugin | None:
+    """Get metadata source plugin by name."""
+    name = name.lower()
+    plugins = find_metadata_source_plugins()
+    return next((p for p in plugins if p.data_source.lower() == name), None)
+
+
 @contextmanager
-def handle_plugin_error(plugin: MetadataSourcePlugin, method_name: str):
+def maybe_handle_plugin_error(plugin: MetadataSourcePlugin, method_name: str):
     """Safely call a plugin method, catching and logging exceptions."""
-    try:
+    if config["raise_on_error"]:
         yield
-    except Exception as e:
-        log.error("Error in '{}.{}': {}", plugin.data_source, method_name, e)
-        log.debug("Exception details:", exc_info=True)
+    else:
+        try:
+            yield
+        except Exception as e:
+            log.error(
+                "Error in '{}.{}': {}", plugin.data_source, method_name, e
+            )
+            log.debug("Exception details:", exc_info=True)
 
 
 def _yield_from_plugins(
@@ -68,11 +81,7 @@ def _yield_from_plugins(
     def wrapper(*args, **kwargs) -> Iterator[Ret]:
         for plugin in find_metadata_source_plugins():
             method = getattr(plugin, method_name)
-            with (
-                nullcontext()
-                if config["raise_on_error"]
-                else handle_plugin_error(plugin, method_name)
-            ):
+            with maybe_handle_plugin_error(plugin, method_name):
                 yield from filter(None, method(*args, **kwargs))
 
     return wrapper
@@ -102,12 +111,26 @@ def tracks_for_ids(*args, **kwargs) -> Iterator[TrackInfo]:
     yield from ()
 
 
-def album_for_id(_id: str) -> AlbumInfo | None:
-    return next(albums_for_ids([_id]), None)
+def album_for_id(_id: str, data_source: str) -> AlbumInfo | None:
+    """Get AlbumInfo object for the given ID and data source."""
+    if plugin := get_metadata_source(data_source):
+        with maybe_handle_plugin_error(plugin, "album_for_id"):
+            if info := plugin.album_for_id(_id):
+                send("albuminfo_received", info=info)
+                return info
+
+    return None
+
 
+def track_for_id(_id: str, data_source: str) -> TrackInfo | None:
+    """Get TrackInfo object for the given ID and data source."""
+    if plugin := get_metadata_source(data_source):
+        with maybe_handle_plugin_error(plugin, "track_for_id"):
+            if info := plugin.track_for_id(_id):
+                send("trackinfo_received", info=info)
+                return info
 
-def track_for_id(_id: str) -> TrackInfo | None:
-    return next(tracks_for_ids([_id]), None)
+    return None
 
 
 @cache
diff --git a/beetsplug/mbsync.py b/beetsplug/mbsync.py
index 45f34e865e..70f774295f 100644
--- a/beetsplug/mbsync.py
+++ b/beetsplug/mbsync.py
@@ -72,17 +72,19 @@ def singletons(self, lib, query, move, pretend, write):
         query.
         """
         for item in lib.items([*query, "singleton:true"]):
-            if not item.mb_trackid:
+            if not (track_id := item.mb_trackid):
                 self._log.info(
                     "Skipping singleton with no mb_trackid: {}", item
                 )
                 continue
 
             if not (
-                track_info := metadata_plugins.track_for_id(item.mb_trackid)
+                track_info := metadata_plugins.track_for_id(
+                    track_id, item.get("data_source", "MusicBrainz")
+                )
             ):
                 self._log.info(
-                    "Recording ID not found: {0.mb_trackid} for track {0}", item
+                    "Recording ID not found: {} for track {}", track_id, item
                 )
                 continue
 
@@ -97,15 +99,20 @@ def albums(self, lib, query, move, pretend, write):
         """
         # Process matching albums.
         for album in lib.albums(query):
-            if not album.mb_albumid:
+            if not (album_id := album.mb_albumid):
                 self._log.info("Skipping album with no mb_albumid: {}", album)
                 continue
 
+            data_source = album.get("data_source") or album.items()[0].get(
+                "data_source", "MusicBrainz"
+            )
             if not (
-                album_info := metadata_plugins.album_for_id(album.mb_albumid)
+                album_info := metadata_plugins.album_for_id(
+                    album_id, data_source
+                )
             ):
                 self._log.info(
-                    "Release ID {0.mb_albumid} not found for album {0}", album
+                    "Release ID {} not found for album {}", album_id, album
                 )
                 continue
 
diff --git a/beetsplug/missing.py b/beetsplug/missing.py
index d2aae14e98..271e90b06a 100644
--- a/beetsplug/missing.py
+++ b/beetsplug/missing.py
@@ -227,10 +227,15 @@ def _missing(self, album: Album) -> Iterator[Item]:
         if len(album.items()) == album.albumtotal:
             return
 
-        item_mbids = {x.mb_trackid for x in album.items()}
         # fetch missing items
         # TODO: Implement caching that without breaking other stuff
-        if album_info := metadata_plugins.album_for_id(album.mb_albumid):
+        data_source = album.get("data_source") or album.items()[0].get(
+            "data_source", "MusicBrainz"
+        )
+        if album_info := metadata_plugins.album_for_id(
+            album.mb_albumid, data_source
+        ):
+            item_mbids = {x.mb_trackid for x in album.items()}
             for track_info in album_info.tracks:
                 if track_info.track_id not in item_mbids:
                     self._log.debug(
diff --git a/docs/changelog.rst b/docs/changelog.rst
index 4a944484f7..d4cbe1ba9a 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -20,10 +20,18 @@ Bug fixes
   edge-case mismatches (e.g., a song titled "1:00 AM" would incorrectly be
   considered a Windows drive path).
 - :doc:`plugins/fish`: Fix AttributeError. :bug:`6340`
+- :ref:`import-cmd` Autotagging by explicit release or recording IDs now keeps
+  candidates from all enabled metadata sources instead of dropping matches when
+  different providers share the same ID. :bug:`6178` :bug:`6181`
+- :doc:`plugins/mbsync` and :doc:`plugins/missing` now use each item's stored
+  ``data_source`` for ID lookups, with a fallback to ``MusicBrainz``.
 
-..
-    For plugin developers
-    ~~~~~~~~~~~~~~~~~~~~~
+For plugin developers
+~~~~~~~~~~~~~~~~~~~~~
+
+- :py:func:`beets.metadata_plugins.album_for_id` and
+  :py:func:`beets.metadata_plugins.track_for_id` now require a ``data_source``
+  argument and query only that provider.
 
 Other changes
 ~~~~~~~~~~~~~
diff --git a/docs/dev/plugins/autotagger.rst b/docs/dev/plugins/autotagger.rst
index fd08c81389..dc0e41be34 100644
--- a/docs/dev/plugins/autotagger.rst
+++ b/docs/dev/plugins/autotagger.rst
@@ -113,6 +113,18 @@ IDs are expected to be strings. If your source uses specific formats, consider
 contributing an extractor regex to the core module:
 :py:mod:`beets.util.id_extractors`.
 
+When beets matches by explicit IDs (for example via ``--search-id`` or existing
+``mb_*id`` fields), it asks every enabled metadata source plugin for candidates
+using :py:meth:`~MetadataSourcePlugin.albums_for_ids` and
+:py:meth:`~MetadataSourcePlugin.tracks_for_ids`. Candidate identity is tracked
+by ``(data_source, id)``, so identical IDs from different providers remain
+separate options.
+
+If you need to query one specific provider, use the module helpers
+:py:func:`beets.metadata_plugins.album_for_id` and
+:py:func:`beets.metadata_plugins.track_for_id` and pass both the ID and the
+provider ``data_source`` name.
+
 Best practices
 --------------
 
diff --git a/docs/dev/plugins/events.rst b/docs/dev/plugins/events.rst
index aaab9ccd77..ace7b93874 100644
--- a/docs/dev/plugins/events.rst
+++ b/docs/dev/plugins/events.rst
@@ -180,10 +180,9 @@ registration process in this case:
 
 ``album_matched``
     :Parameters: ``match`` (``AlbumMatch``)
-    :Description: Called after ``Item`` objects from a folder that's being
-        imported have been matched to an ``AlbumInfo`` and the corresponding
-        distance has been calculated. Missing and extra tracks, if any, are
-        included in the match.
+    :Description: Called each time an ``AlbumMatch`` candidate is created while
+        importing. This applies to both ID-driven and text-search matching.
+        Missing and extra tracks, if any, are included in the match.
 
 ``before_choose_candidate``
     :Parameters: ``task`` (|ImportTask|), ``session`` (|ImportSession|)
diff --git a/docs/plugins/mbsync.rst b/docs/plugins/mbsync.rst
index d2f80d1f88..9d0b793fb0 100644
--- a/docs/plugins/mbsync.rst
+++ b/docs/plugins/mbsync.rst
@@ -17,6 +17,9 @@ Enable the ``mbsync`` plugin in your configuration (see :ref:`using-plugins`)
 and then run ``beet mbsync QUERY`` to fetch updated metadata for a part of your
 collection (or omit the query to run over your whole library).
 
+ID lookups use each item's stored ``data_source``. If a row has no
+``data_source``, ``mbsync`` falls back to ``MusicBrainz``.
+
 This plugin treats albums and singletons (non-album tracks) separately. It first
 processes all matching singletons and then proceeds on to full albums. The same
 query is used to search for both kinds of entities.
diff --git a/docs/plugins/missing.rst b/docs/plugins/missing.rst
index d286e43ccb..5c1cd54553 100644
--- a/docs/plugins/missing.rst
+++ b/docs/plugins/missing.rst
@@ -10,6 +10,8 @@ Usage
 
 The ``beet missing`` command fetches album information from the origin data
 source and lists names of the **tracks** that are missing from your library.
+Track-level checks use the album's stored ``data_source`` and fall back to
+``MusicBrainz`` when no source is stored.
 
 It can also list the names of missing **albums** for each artist, although this
 is limited to albums from the MusicBrainz data source only.
diff --git a/test/autotag/__init__.py b/test/autotag/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/test/autotag/test_autotag.py b/test/autotag/test_autotag.py
index 903f9095c0..cfbae06913 100644
--- a/test/autotag/test_autotag.py
+++ b/test/autotag/test_autotag.py
@@ -19,101 +19,11 @@
 import pytest
 
 from beets import autotag
-from beets.autotag import AlbumInfo, TrackInfo, correct_list_fields, match
+from beets.autotag import AlbumInfo, TrackInfo, correct_list_fields
 from beets.library import Item
 from beets.test.helper import BeetsTestCase
 
 
-class TestAssignment:
-    A = "one"
-    B = "two"
-    C = "three"
-
-    @pytest.fixture(autouse=True)
-    def config(self, config):
-        config["match"]["track_length_grace"] = 10
-        config["match"]["track_length_max"] = 30
-
-    @pytest.mark.parametrize(
-        # 'expected' is a tuple of expected (mapping, extra_items, extra_tracks)
-        "item_titles, track_titles, expected",
-        [
-            # items ordering gets corrected
-            ([A, C, B], [A, B, C], ({A: A, B: B, C: C}, [], [])),
-            # unmatched tracks are returned as 'extra_tracks'
-            # the first track is unmatched
-            ([B, C], [A, B, C], ({B: B, C: C}, [], [A])),
-            # the middle track is unmatched
-            ([A, C], [A, B, C], ({A: A, C: C}, [], [B])),
-            # the last track is unmatched
-            ([A, B], [A, B, C], ({A: A, B: B}, [], [C])),
-            # unmatched items are returned as 'extra_items'
-            ([A, C, B], [A, C], ({A: A, C: C}, [B], [])),
-        ],
-    )
-    def test_assign_tracks(self, item_titles, track_titles, expected):
-        expected_mapping, expected_extra_items, expected_extra_tracks = expected
-
-        items = [Item(title=title) for title in item_titles]
-        tracks = [TrackInfo(title=title) for title in track_titles]
-
-        item_info_pairs, extra_items, extra_tracks = match.assign_items(
-            items, tracks
-        )
-
-        assert (
-            {i.title: t.title for i, t in item_info_pairs},
-            [i.title for i in extra_items],
-            [t.title for t in extra_tracks],
-        ) == (expected_mapping, expected_extra_items, expected_extra_tracks)
-
-    def test_order_works_when_track_names_are_entirely_wrong(self):
-        # A real-world test case contributed by a user.
-        def item(i, length):
-            return Item(
-                artist="ben harper",
-                album="burn to shine",
-                title=f"ben harper - Burn to Shine {i}",
-                track=i,
-                length=length,
-            )
-
-        items = []
-        items.append(item(1, 241.37243007106997))
-        items.append(item(2, 342.27781704375036))
-        items.append(item(3, 245.95070222338137))
-        items.append(item(4, 472.87662515485437))
-        items.append(item(5, 279.1759535763187))
-        items.append(item(6, 270.33333768012))
-        items.append(item(7, 247.83435613222923))
-        items.append(item(8, 216.54504531525072))
-        items.append(item(9, 225.72775379800484))
-        items.append(item(10, 317.7643606963552))
-        items.append(item(11, 243.57001238834192))
-        items.append(item(12, 186.45916150485752))
-
-        def info(index, title, length):
-            return TrackInfo(title=title, length=length, index=index)
-
-        trackinfo = []
-        trackinfo.append(info(1, "Alone", 238.893))
-        trackinfo.append(info(2, "The Woman in You", 341.44))
-        trackinfo.append(info(3, "Less", 245.59999999999999))
-        trackinfo.append(info(4, "Two Hands of a Prayer", 470.49299999999999))
-        trackinfo.append(info(5, "Please Bleed", 277.86599999999999))
-        trackinfo.append(info(6, "Suzie Blue", 269.30599999999998))
-        trackinfo.append(info(7, "Steal My Kisses", 245.36000000000001))
-        trackinfo.append(info(8, "Burn to Shine", 214.90600000000001))
-        trackinfo.append(info(9, "Show Me a Little Shame", 224.0929999999999))
-        trackinfo.append(info(10, "Forgiven", 317.19999999999999))
-        trackinfo.append(info(11, "Beloved One", 243.733))
-        trackinfo.append(info(12, "In the Lord's Arms", 186.13300000000001))
-
-        expected = list(zip(items, trackinfo)), [], []
-
-        assert match.assign_items(items, trackinfo) == expected
-
-
 class ApplyTest(BeetsTestCase):
     def _apply(self, per_disc_numbering=False, artist_credit=False):
         info = self.info
diff --git a/test/autotag/test_match.py b/test/autotag/test_match.py
new file mode 100644
index 0000000000..97c7dd8f83
--- /dev/null
+++ b/test/autotag/test_match.py
@@ -0,0 +1,183 @@
+from typing import ClassVar
+
+import pytest
+
+from beets import metadata_plugins
+from beets.autotag import AlbumInfo, TrackInfo, match
+from beets.library import Item
+
+
+class TestAssignment:
+    A = "one"
+    B = "two"
+    C = "three"
+
+    @pytest.fixture(autouse=True)
+    def config(self, config):
+        config["match"]["track_length_grace"] = 10
+        config["match"]["track_length_max"] = 30
+
+    @pytest.mark.parametrize(
+        # 'expected' is a tuple of expected (mapping, extra_items, extra_tracks)
+        "item_titles, track_titles, expected",
+        [
+            # items ordering gets corrected
+            ([A, C, B], [A, B, C], ({A: A, B: B, C: C}, [], [])),
+            # unmatched tracks are returned as 'extra_tracks'
+            # the first track is unmatched
+            ([B, C], [A, B, C], ({B: B, C: C}, [], [A])),
+            # the middle track is unmatched
+            ([A, C], [A, B, C], ({A: A, C: C}, [], [B])),
+            # the last track is unmatched
+            ([A, B], [A, B, C], ({A: A, B: B}, [], [C])),
+            # unmatched items are returned as 'extra_items'
+            ([A, C, B], [A, C], ({A: A, C: C}, [B], [])),
+        ],
+    )
+    def test_assign_tracks(self, item_titles, track_titles, expected):
+        expected_mapping, expected_extra_items, expected_extra_tracks = expected
+
+        items = [Item(title=title) for title in item_titles]
+        tracks = [TrackInfo(title=title) for title in track_titles]
+
+        item_info_pairs, extra_items, extra_tracks = match.assign_items(
+            items, tracks
+        )
+
+        assert (
+            {i.title: t.title for i, t in item_info_pairs},
+            [i.title for i in extra_items],
+            [t.title for t in extra_tracks],
+        ) == (expected_mapping, expected_extra_items, expected_extra_tracks)
+
+    def test_order_works_when_track_names_are_entirely_wrong(self):
+        # A real-world test case contributed by a user.
+        def item(i, length):
+            return Item(
+                artist="ben harper",
+                album="burn to shine",
+                title=f"ben harper - Burn to Shine {i}",
+                track=i,
+                length=length,
+            )
+
+        items = []
+        items.append(item(1, 241.37243007106997))
+        items.append(item(2, 342.27781704375036))
+        items.append(item(3, 245.95070222338137))
+        items.append(item(4, 472.87662515485437))
+        items.append(item(5, 279.1759535763187))
+        items.append(item(6, 270.33333768012))
+        items.append(item(7, 247.83435613222923))
+        items.append(item(8, 216.54504531525072))
+        items.append(item(9, 225.72775379800484))
+        items.append(item(10, 317.7643606963552))
+        items.append(item(11, 243.57001238834192))
+        items.append(item(12, 186.45916150485752))
+
+        def info(index, title, length):
+            return TrackInfo(title=title, length=length, index=index)
+
+        trackinfo = []
+        trackinfo.append(info(1, "Alone", 238.893))
+        trackinfo.append(info(2, "The Woman in You", 341.44))
+        trackinfo.append(info(3, "Less", 245.59999999999999))
+        trackinfo.append(info(4, "Two Hands of a Prayer", 470.49299999999999))
+        trackinfo.append(info(5, "Please Bleed", 277.86599999999999))
+        trackinfo.append(info(6, "Suzie Blue", 269.30599999999998))
+        trackinfo.append(info(7, "Steal My Kisses", 245.36000000000001))
+        trackinfo.append(info(8, "Burn to Shine", 214.90600000000001))
+        trackinfo.append(info(9, "Show Me a Little Shame", 224.0929999999999))
+        trackinfo.append(info(10, "Forgiven", 317.19999999999999))
+        trackinfo.append(info(11, "Beloved One", 243.733))
+        trackinfo.append(info(12, "In the Lord's Arms", 186.13300000000001))
+
+        expected = list(zip(items, trackinfo)), [], []
+
+        assert match.assign_items(items, trackinfo) == expected
+
+
+class TestTagMultipleDataSources:
+    @pytest.fixture
+    def shared_track_id(self):
+        return "track-12345"
+
+    @pytest.fixture
+    def shared_album_id(self):
+        return "album-12345"
+
+    @pytest.fixture(autouse=True)
+    def _setup_plugins(self, monkeypatch, shared_album_id, shared_track_id):
+        class StubPlugin:
+            data_source: ClassVar[str]
+            data_source_mismatch_penalty = 0
+
+            @property
+            def track(self):
+                return TrackInfo(
+                    artist="Artist",
+                    title="Title",
+                    track_id=shared_track_id,
+                    data_source=self.data_source,
+                )
+
+            @property
+            def album(self):
+                return AlbumInfo(
+                    [self.track],
+                    artist="Albumartist",
+                    album="Album",
+                    album_id=shared_album_id,
+                    data_source=self.data_source,
+                )
+
+            def albums_for_ids(self, *_):
+                yield self.album
+
+            def tracks_for_ids(self, *_):
+                yield self.track
+
+            def candidates(self, *_, **__):
+                yield self.album
+
+            def item_candidates(self, *_, **__):
+                yield self.track
+
+        class DeezerPlugin(StubPlugin):
+            data_source = "Deezer"
+
+        class DiscogsPlugin(StubPlugin):
+            data_source = "Discogs"
+
+        monkeypatch.setattr(
+            metadata_plugins,
+            "find_metadata_source_plugins",
+            lambda: [DeezerPlugin(), DiscogsPlugin()],
+        )
+
+    def check_proposal(self, proposal):
+        sources = [
+            candidate.info.data_source for candidate in proposal.candidates
+        ]
+        assert len(sources) == 2
+        assert set(sources) == {"Discogs", "Deezer"}
+
+    def test_search_album_ids(self, shared_album_id):
+        _, _, proposal = match.tag_album([Item()], search_ids=[shared_album_id])
+
+        self.check_proposal(proposal)
+
+    def test_search_album_current_id(self, shared_album_id):
+        _, _, proposal = match.tag_album([Item(mb_albumid=shared_album_id)])
+
+        self.check_proposal(proposal)
+
+    def test_search_track_ids(self, shared_track_id):
+        proposal = match.tag_item(Item(), search_ids=[shared_track_id])
+
+        self.check_proposal(proposal)
+
+    def test_search_track_current_id(self, shared_track_id):
+        proposal = match.tag_item(Item(mb_trackid=shared_track_id))
+
+        self.check_proposal(proposal)
diff --git a/test/plugins/test_mbsync.py b/test/plugins/test_mbsync.py
index bb88e5e631..714b374e32 100644
--- a/test/plugins/test_mbsync.py
+++ b/test/plugins/test_mbsync.py
@@ -45,10 +45,15 @@ def test_update_library(self):
             album="old album",
             mb_albumid="album id",
             mb_trackid="track id",
+            data_source="data_source",
         )
         self.lib.add_album([album_item])
 
-        singleton = Item(title="old title", mb_trackid="singleton id")
+        singleton = Item(
+            title="old title",
+            mb_trackid="singleton id",
+            data_source="data_source",
+        )
         self.lib.add(singleton)
 
         self.run_command("mbsync")
diff --git a/test/test_importer.py b/test/test_importer.py
index fe37072fe3..1a8983c11d 100644
--- a/test/test_importer.py
+++ b/test/test_importer.py
@@ -1557,7 +1557,7 @@ def test_import_pretend_empty(self):
         assert self.__run(importer) == [f"No files imported from {empty_path}"]
 
 
-def mocked_get_album_by_id(id_):
+def mocked_get_albums_by_ids(ids):
     """Return album candidate for the given id.
 
     The two albums differ only in the release title and artist name, so that
@@ -1565,32 +1565,34 @@ def mocked_get_album_by_id(id_):
     ImportHelper.prepare_album_for_import().
     """
     # Map IDs to (release title, artist), so the distances are different.
-    album, artist = {
+    album_artist_map = {
         ImportIdTest.ID_RELEASE_0: ("VALID_RELEASE_0", "TAG ARTIST"),
         ImportIdTest.ID_RELEASE_1: ("VALID_RELEASE_1", "DISTANT_MATCH"),
-    }[id_]
-
-    return AlbumInfo(
-        album_id=id_,
-        album=album,
-        artist_id="some-id",
-        artist=artist,
-        albumstatus="Official",
-        tracks=[
-            TrackInfo(
-                track_id="bar",
-                title="foo",
-                artist_id="some-id",
-                artist=artist,
-                length=59,
-                index=9,
-                track_allt="A2",
-            )
-        ],
-    )
+    }
+
+    for id_ in ids:
+        album, artist = album_artist_map[id_]
+        yield AlbumInfo(
+            album_id=id_,
+            album=album,
+            artist_id="some-id",
+            artist=artist,
+            albumstatus="Official",
+            tracks=[
+                TrackInfo(
+                    track_id="bar",
+                    title="foo",
+                    artist_id="some-id",
+                    artist=artist,
+                    length=59,
+                    index=9,
+                    track_allt="A2",
+                )
+            ],
+        )
 
 
-def mocked_get_track_by_id(id_):
+def mocked_get_tracks_by_ids(ids):
     """Return track candidate for the given id.
 
     The two tracks differ only in the release title and artist name, so that
@@ -1598,27 +1600,29 @@ def mocked_get_track_by_id(id_):
     ImportHelper.prepare_album_for_import().
     """
     # Map IDs to (recording title, artist), so the distances are different.
-    title, artist = {
+    title_artist_map = {
         ImportIdTest.ID_RECORDING_0: ("VALID_RECORDING_0", "TAG ARTIST"),
         ImportIdTest.ID_RECORDING_1: ("VALID_RECORDING_1", "DISTANT_MATCH"),
-    }[id_]
-
-    return TrackInfo(
-        track_id=id_,
-        title=title,
-        artist_id="some-id",
-        artist=artist,
-        length=59,
-    )
+    }
+
+    for id_ in ids:
+        title, artist = title_artist_map[id_]
+        yield TrackInfo(
+            track_id=id_,
+            title=title,
+            artist_id="some-id",
+            artist=artist,
+            length=59,
+        )
 
 
 @patch(
-    "beets.metadata_plugins.track_for_id",
-    Mock(side_effect=mocked_get_track_by_id),
+    "beets.metadata_plugins.tracks_for_ids",
+    Mock(side_effect=mocked_get_tracks_by_ids),
 )
 @patch(
-    "beets.metadata_plugins.album_for_id",
-    Mock(side_effect=mocked_get_album_by_id),
+    "beets.metadata_plugins.albums_for_ids",
+    Mock(side_effect=mocked_get_albums_by_ids),
 )
 class ImportIdTest(ImportTestCase):
     ID_RELEASE_0 = "00000000-0000-0000-0000-000000000000"
diff --git a/test/test_metadata_plugins.py b/test/test_metadata_plugins.py
index da1dc5c992..ca82367ede 100644
--- a/test/test_metadata_plugins.py
+++ b/test/test_metadata_plugins.py
@@ -32,6 +32,7 @@ class TestMetadataPluginsException(PluginMixin):
     @pytest.fixture(autouse=True)
     def setup(self):
         metadata_plugins.find_metadata_source_plugins.cache_clear()
+        metadata_plugins.get_metadata_source.cache_clear()
         self.register_plugin(ErrorMetadataMockPlugin)
         yield
         self.unload_plugins()
@@ -45,25 +46,23 @@ def _call():
         return _call
 
     @pytest.mark.parametrize(
-        "method_name,error_method_name,args",
+        "method_name,args",
         [
-            ("candidates", "candidates", ()),
-            ("item_candidates", "item_candidates", ()),
-            ("albums_for_ids", "albums_for_ids", (["some_id"],)),
-            ("tracks_for_ids", "tracks_for_ids", (["some_id"],)),
-            # Currently, singular methods call plural ones internally and log
-            # errors from there
-            ("album_for_id", "albums_for_ids", ("some_id",)),
-            ("track_for_id", "tracks_for_ids", ("some_id",)),
+            ("candidates", ()),
+            ("item_candidates", ()),
+            ("albums_for_ids", (["some_id"],)),
+            ("tracks_for_ids", (["some_id"],)),
+            ("album_for_id", ("some_id", "ErrorMetadataMock")),
+            ("track_for_id", ("some_id", "ErrorMetadataMock")),
         ],
     )
-    def test_logging(self, caplog, call_method, error_method_name):
+    def test_logging(self, caplog, call_method, method_name):
         self.config["raise_on_error"] = False
 
         call_method()
 
         assert (
-            f"Error in 'ErrorMetadataMock.{error_method_name}': Mocked error"
+            f"Error in 'ErrorMetadataMock.{method_name}': Mocked error"
             in caplog.text
         )
 
@@ -72,8 +71,10 @@ def test_logging(self, caplog, call_method, error_method_name):
         [
             ("candidates", ()),
             ("item_candidates", ()),
-            ("album_for_id", ("some_id",)),
-            ("track_for_id", ("some_id",)),
+            ("albums_for_ids", (["some_id"],)),
+            ("tracks_for_ids", (["some_id"],)),
+            ("album_for_id", ("some_id", "ErrorMetadataMock")),
+            ("track_for_id", ("some_id", "ErrorMetadataMock")),
         ],
     )
     def test_raising(self, call_method):